test(cli): exercise adaptive S3 pack reads

package.json

@@ -44,6 +44,7 @@
         "test:contract:context:obsidian": "npm run build && npm run test:e2e:obsidian:smoke",
         "test:e2e:cli": "npm run test:e2e:ci --workspace self-hosted-livesync-cli",
         "test:e2e:cli:p2p": "npm run test:e2e:p2p --workspace self-hosted-livesync-cli",
+        "test:e2e:cli:adaptive-s3": "npm run test:e2e:adaptive-s3 --workspace self-hosted-livesync-cli",
         "test:e2e:cli:all": "npm run test:e2e:all --workspace self-hosted-livesync-cli",
         "test:integration": "npx dotenv-cli -e .env -e .test.env -- vitest run --config vitest.config.integration.ts",
         "test:unit:coverage": "vitest run --config vitest.config.unit.ts --coverage",

src/apps/cli/commands/runCommand.ts

@@ -25,6 +25,8 @@ import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFu
 import { fsPromises as fs, path } from "@vrtmrz/livesync-commonlib/node";
 import type { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
 import type { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
+import type { JournalSyncCore } from "@vrtmrz/livesync-commonlib/compat/replication/journal/JournalSyncCore";
+import { journalProtocolConfigurationForSettings } from "@vrtmrz/livesync-commonlib/journal-storage";
 import { writeStderrLine, writeStdoutLine } from "@/apps/cli/cliOutput";
 
 function redactConnectionString(uri: string): string {
@@ -60,7 +62,19 @@ async function verifyRemoteState(
             }
             milestone = await dbRet.db.get(MILESTONE_DOCID);
         } else if (settings.remoteType === REMOTE_MINIO) {
-            milestone = await (replicator as LiveSyncJournalReplicator).client.downloadJson("_00000000-milestone.json");
+            const journalReplicator = replicator as LiveSyncJournalReplicator;
+            if (journalProtocolConfigurationForSettings(settings).journalFormat === "adaptive-v1") {
+                try {
+                    await journalReplicator.client.ensureCheckpointCachesAreFresh();
+                    standardIo.writeStderr("[Verification] Adaptive Journal repository is available.\n");
+                    return true;
+                } catch (error) {
+                    const message = error instanceof Error ? error.message : String(error);
+                    standardIo.writeStderr(`[Verification] Failed to verify Adaptive Journal repository: ${message}\n`);
+                    return false;
+                }
+            }
+            milestone = await (journalReplicator.client as JournalSyncCore).downloadJson("_00000000-milestone.json");
         }
 
         if (milestone) {

src/apps/cli/commands/runCommand.unit.spec.ts

@@ -2,7 +2,12 @@ import { fsPromises as fs, os, path } from "@vrtmrz/livesync-commonlib/node";
 import * as processSetting from "@vrtmrz/livesync-commonlib/compat/API/processSetting";
 import { ConnectionStringParser } from "@vrtmrz/livesync-commonlib/compat/common/ConnectionString";
 import { configURIBase } from "@vrtmrz/livesync-commonlib/compat/common/models/shared.const";
-import { DEFAULT_SETTINGS, REMOTE_COUCHDB, REMOTE_MINIO, REMOTE_P2P } from "@vrtmrz/livesync-commonlib/compat/common/types";
+import {
+    DEFAULT_SETTINGS,
+    REMOTE_COUCHDB,
+    REMOTE_MINIO,
+    REMOTE_P2P,
+} from "@vrtmrz/livesync-commonlib/compat/common/types";
 import { describe, expect, it, vi, beforeEach, afterEach } from "vitest";
 import { runCommand } from "./runCommand";
 import type { CLIOptions } from "./types";
@@ -717,6 +722,34 @@ describe("runCommand abnormal cases", () => {
             expect(core.services.control.applySettings).not.toHaveBeenCalled();
         });
 
+        it("verifies an Adaptive Journal repository without reading the legacy milestone", async () => {
+            const core = createCoreMock();
+            const settings = core.services.setting.currentSettings();
+            settings.remoteType = REMOTE_MINIO;
+            settings.journalFormat = "adaptive-v1";
+            settings.packReadPolicy = "whole-pack";
+
+            const ensureCheckpointCachesAreFresh = vi.fn(async () => {});
+            core.services.replicator.getActiveReplicator.mockReturnValue({
+                nodeid: "test-node-id",
+                initializeDatabaseForReplication: vi.fn(async () => {}),
+                client: {
+                    ensureCheckpointCachesAreFresh,
+                },
+            });
+
+            const result = await runCommand(makeOptions("mark-resolved", []), {
+                ...context,
+                core,
+            });
+
+            expect(result).toBe(true);
+            expect(ensureCheckpointCachesAreFresh).toHaveBeenCalledTimes(1);
+            expect(core.services.context.standardIo.writeStderr).toHaveBeenCalledWith(
+                "[Verification] Adaptive Journal repository is available.\n"
+            );
+        });
+
         it("mark-resolved with remote-id temporarily activates it and runs markResolved", async () => {
             const core = createCoreMock();
             const settings = core.services.setting.currentSettings();

src/apps/cli/package.json

@@ -22,6 +22,8 @@
         "pretest:e2e:ci": "npm run build",
         "test:e2e:ci": "deno task --cwd testdeno test:ci",
         "test:e2e:p2p": "deno task --cwd testdeno test:p2p:compose",
+        "pretest:e2e:adaptive-s3": "npm run build",
+        "test:e2e:adaptive-s3": "deno task --cwd testdeno test:adaptive-journal-s3",
         "test:e2e:mirror": "bash test/test-mirror-linux.sh",
         "test:e2e:remote-commands": "bash test/test-remote-commands-linux.sh",
         "pretest:e2e:all": "npm run build",

src/apps/cli/services/NodeKeyValueDBService.ts

@@ -25,6 +25,10 @@ type SerializableContainer =
     | {
           [NODE_KV_TYPED_KEY]: "ArrayBuffer";
           [NODE_KV_VALUES_KEY]: number[];
+      }
+    | {
+          [NODE_KV_TYPED_KEY]: "BigInt";
+          [NODE_KV_VALUES_KEY]: string;
       };
 
 function isRecord(value: unknown): value is Record<string, unknown> {
@@ -32,6 +36,12 @@ function isRecord(value: unknown): value is Record<string, unknown> {
 }
 
 function serializeForNodeKV(value: unknown): unknown {
+    if (typeof value === "bigint") {
+        return {
+            [NODE_KV_TYPED_KEY]: "BigInt",
+            [NODE_KV_VALUES_KEY]: value.toString(10),
+        } satisfies SerializableContainer;
+    }
     if (value instanceof Set) {
         return {
             [NODE_KV_TYPED_KEY]: "Set",
@@ -78,6 +88,9 @@ function deserializeFromNodeKV(value: unknown): unknown {
     if (taggedType === "ArrayBuffer" && Array.isArray(taggedValues)) {
         return Uint8Array.from(taggedValues).buffer;
     }
+    if (taggedType === "BigInt" && typeof taggedValues === "string" && /^-?(?:0|[1-9]\d*)$/u.test(taggedValues)) {
+        return BigInt(taggedValues);
+    }
 
     return Object.fromEntries(Object.entries(value).map(([k, v]) => [k, deserializeFromNodeKV(v)]));
 }

src/apps/cli/services/NodeKeyValueDBService.unit.spec.ts

@@ -1,8 +1,39 @@
 import { describe, expect, it, vi } from "vitest";
 import { createServiceContext } from "@vrtmrz/livesync-commonlib/compat/services/base/ServiceBase";
+import { fsPromises as fs, os, path } from "@vrtmrz/livesync-commonlib/node";
 import type { NodeKeyValueDBDependencies } from "./NodeKeyValueDBService";
 import { NodeKeyValueDBService } from "./NodeKeyValueDBService";
 
+function createInitialisableDependencies(): {
+    dependencies: NodeKeyValueDBDependencies;
+    initialise: () => Promise<boolean>;
+} {
+    let initialise: (() => Promise<boolean>) | undefined;
+    const dependencies = {
+        appLifecycle: {
+            onSettingLoaded: {
+                addHandler: vi.fn((handler: () => Promise<boolean>) => {
+                    initialise = handler;
+                }),
+            },
+        },
+        databaseEvents: {
+            onResetDatabase: { addHandler: vi.fn() },
+            onDatabaseInitialisation: { addHandler: vi.fn() },
+            onUnloadDatabase: { addHandler: vi.fn() },
+            onCloseDatabase: { addHandler: vi.fn() },
+        },
+        vault: {},
+    } as unknown as NodeKeyValueDBDependencies;
+    return {
+        dependencies,
+        initialise: async () => {
+            if (!initialise) throw new Error("Initialisation handler was not registered");
+            return await initialise();
+        },
+    };
+}
+
 describe("NodeKeyValueDBService.openSimpleStore", () => {
     it("creates a namespaced store handle before the backing database is initialised", () => {
         const dependencies = {
@@ -44,4 +75,29 @@ describe("NodeKeyValueDBService.openSimpleStore", () => {
 
         await expect(store.get("key")).rejects.toThrow("KeyValueDB is not initialized yet");
     });
+
+    it("preserves bigint values used by Adaptive Journal state", async () => {
+        const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "livesync-node-kv-bigint-"));
+        const filePath = path.join(tempDir, "keyvalue-db.json");
+        const writerState = {
+            lastCommittedSequence: 9007199254740993n,
+            pendingCommit: { sequence: 18446744073709551615n },
+            writerEpoch: "test-writer-epoch",
+        };
+
+        try {
+            const firstLifecycle = createInitialisableDependencies();
+            const first = new NodeKeyValueDBService(createServiceContext(), firstLifecycle.dependencies, filePath);
+            await expect(firstLifecycle.initialise()).resolves.toBe(true);
+            await first.openSimpleStore("adaptive").set("writer-state", writerState);
+
+            const secondLifecycle = createInitialisableDependencies();
+            const second = new NodeKeyValueDBService(createServiceContext(), secondLifecycle.dependencies, filePath);
+            await expect(secondLifecycle.initialise()).resolves.toBe(true);
+
+            await expect(second.openSimpleStore("adaptive").get("writer-state")).resolves.toEqual(writerState);
+        } finally {
+            await fs.rm(tempDir, { recursive: true, force: true });
+        }
+    });
 });

src/apps/cli/testdeno/deno.json

@@ -34,7 +34,8 @@
         "test:e2e-matrix:couchdb-enc0": "deno test --env-file=.test.env -A --no-check --filter='e2e matrix: COUCHDB-enc0' test-e2e-two-vaults-matrix.ts",
         "test:e2e-matrix:couchdb-enc1": "deno test --env-file=.test.env -A --no-check --filter='e2e matrix: COUCHDB-enc1' test-e2e-two-vaults-matrix.ts",
         "test:e2e-matrix:minio-enc0": "deno test --env-file=.test.env -A --no-check --filter='e2e matrix: MINIO-enc0' test-e2e-two-vaults-matrix.ts",
-        "test:e2e-matrix:minio-enc1": "deno test --env-file=.test.env -A --no-check --filter='e2e matrix: MINIO-enc1' test-e2e-two-vaults-matrix.ts"
+        "test:e2e-matrix:minio-enc1": "deno test --env-file=.test.env -A --no-check --filter='e2e matrix: MINIO-enc1' test-e2e-two-vaults-matrix.ts",
+        "test:adaptive-journal-s3": "deno test --env-file=.test.env -A --no-check test-adaptive-journal-s3.ts"
     },
     "imports": {
         "@std/assert": "jsr:@std/assert@^1.0.13",

src/apps/cli/testdeno/helpers/docker.ts

@@ -466,6 +466,70 @@ export async function stopMinio(): Promise<void> {
     untrackContainer(MINIO_CONTAINER);
 }
 
+export async function listMinioObjectKeys(
+    minioEndpoint: string,
+    accessKey: string,
+    secretKey: string,
+    bucket: string
+): Promise<string[]> {
+    const cmd =
+        `mc alias set myminio ${shQuote(minioEndpoint)} ${shQuote(accessKey)} ${shQuote(secretKey)} >/dev/null 2>&1 && ` +
+        `mc ls --recursive --json myminio/${shQuote(bucket)}`;
+    const result = await docker(
+        "run",
+        "--rm",
+        "--network",
+        "host",
+        "--entrypoint",
+        "/bin/sh",
+        MINIO_MC_IMAGE,
+        "-c",
+        cmd
+    );
+    if (result.code !== 0) {
+        throw new Error(`Could not list MinIO objects: ${result.stderr.trim()}`);
+    }
+
+    return result.stdout
+        .split(/\r?\n/u)
+        .filter((line) => line.trim().length > 0)
+        .map((line) => JSON.parse(line) as { key?: unknown })
+        .map(({ key }) => {
+            if (typeof key !== "string") {
+                throw new Error("MinIO returned an object without a string key");
+            }
+            return key;
+        })
+        .sort();
+}
+
+export async function readMinioObjectText(
+    minioEndpoint: string,
+    accessKey: string,
+    secretKey: string,
+    bucket: string,
+    key: string
+): Promise<string> {
+    const cmd =
+        `mc alias set myminio ${shQuote(minioEndpoint)} ${shQuote(accessKey)} ${shQuote(secretKey)} >/dev/null 2>&1 && ` +
+        `mc cat myminio/${shQuote(bucket)}/${shQuote(key)}`;
+    const result = await docker(
+        "run",
+        "--rm",
+        "--network",
+        "host",
+        "--entrypoint",
+        "/bin/sh",
+        MINIO_MC_IMAGE,
+        "-c",
+        cmd
+    );
+    if (result.code !== 0) {
+        throw new Error(`Could not read MinIO object ${key}: ${result.stderr.trim()}`);
+    }
+    return result.stdout;
+}
+
 async function initMinioBucket(
     minioEndpoint: string,
     accessKey: string,

src/apps/cli/testdeno/helpers/settings.ts

@@ -125,6 +125,9 @@ export async function applyRemoteSyncSettings(
         passphrase?: string;
         enableCompression?: boolean;
         usePathObfuscation?: boolean;
+        journalFormat?: "adaptive-v1" | "opaque-v1";
+        expectedRepositoryId?: string;
+        packReadPolicy?: "range" | "whole-pack";
     }
 ): Promise<void> {
     const data = JSON.parse(await Deno.readTextFile(settingsFile));
@@ -143,6 +146,15 @@ export async function applyRemoteSyncSettings(
         data.secretKey = options.minioSecretKey;
         data.region = "auto";
         data.forcePathStyle = true;
+        if (options.journalFormat !== undefined) {
+            data.journalFormat = options.journalFormat;
+        }
+        if (options.expectedRepositoryId !== undefined) {
+            data.expectedRepositoryId = options.expectedRepositoryId;
+        }
+        if (options.packReadPolicy !== undefined) {
+            data.packReadPolicy = options.packReadPolicy;
+        }
     }
 
     data.liveSync = true;

src/apps/cli/testdeno/run-ci-suite.ts

@@ -11,6 +11,7 @@ const TASKS = [
     "test:e2e-matrix:couchdb-enc1",
     "test:e2e-matrix:minio-enc0",
     "test:e2e-matrix:minio-enc1",
+    "test:adaptive-journal-s3",
 ] as const;
 
 for (const [index, task] of TASKS.entries()) {

src/apps/cli/testdeno/test-adaptive-journal-s3.ts

@@ -0,0 +1,146 @@
+import { assert, assertEquals } from "@std/assert";
+import { TempDir } from "./helpers/temp.ts";
+import { assertFilesEqual, runCli, runCliOrFail, runCliWithInputOrFail, sanitiseCatStdout } from "./helpers/cli.ts";
+import { applyRemoteSyncSettings, initSettingsFile } from "./helpers/settings.ts";
+import { listMinioObjectKeys, readMinioObjectText, startMinio, stopMinio } from "./helpers/docker.ts";
+
+const EXTERNAL_PACK_TEST_BYTES = 9 * 1024 * 1024;
+
+function deterministicBytes(length: number, seed: number): Uint8Array {
+    const bytes = new Uint8Array(length);
+    let state = seed;
+    for (let index = 0; index < bytes.byteLength; index += 1) {
+        state ^= state << 13;
+        state ^= state >>> 17;
+        state ^= state << 5;
+        bytes[index] = state & 0xff;
+    }
+    return bytes;
+}
+
+function requireEnv(...keys: string[]): string {
+    for (const key of keys) {
+        const value = Deno.env.get(key)?.trim();
+        if (value) return value;
+    }
+    throw new Error(`Required environment variable is missing: ${keys.join(" or ")}`);
+}
+
+Deno.test("e2e: two CLI vaults synchronise through Adaptive Journal S3", async () => {
+    const suffix = `${Date.now()}-${Math.floor(Math.random() * 100000)}`;
+    const endpoint = requireEnv("MINIO_ENDPOINT", "minioEndpoint").replace(/\/$/u, "");
+    const accessKey = requireEnv("MINIO_ACCESS_KEY", "accessKey");
+    const secretKey = requireEnv("MINIO_SECRET_KEY", "secretKey");
+    const bucket = `${requireEnv("MINIO_BUCKET_NAME", "bucketName")}-${suffix}`;
+    const passphrase = "adaptive-journal-cli-e2e-passphrase";
+
+    await using workDir = await TempDir.create("livesync-cli-adaptive-journal-s3");
+    const vaultA = workDir.join("vault-a");
+    const vaultB = workDir.join("vault-b");
+    const settingsA = workDir.join("settings-a.json");
+    const settingsB = workDir.join("settings-b.json");
+    const binarySourceA = workDir.join("source-a.bin");
+    const binarySourceB = workDir.join("source-b.bin");
+    const binaryDestinationA = workDir.join("destination-a.bin");
+    const binaryDestinationB = workDir.join("destination-b.bin");
+    await Deno.mkdir(vaultA, { recursive: true });
+    await Deno.mkdir(vaultB, { recursive: true });
+
+    const keepDocker = Deno.env.get("LIVESYNC_DEBUG_KEEP_DOCKER") === "1";
+    await startMinio(endpoint, accessKey, secretKey, bucket);
+
+    try {
+        await initSettingsFile(settingsA);
+        await initSettingsFile(settingsB);
+        await applyRemoteSyncSettings(settingsA, {
+            remoteType: "MINIO",
+            minioBucket: bucket,
+            minioEndpoint: endpoint,
+            minioAccessKey: accessKey,
+            minioSecretKey: secretKey,
+            encrypt: true,
+            passphrase,
+            enableCompression: false,
+            journalFormat: "adaptive-v1",
+            packReadPolicy: "whole-pack",
+        });
+        await applyRemoteSyncSettings(settingsB, {
+            remoteType: "MINIO",
+            minioBucket: bucket,
+            minioEndpoint: endpoint,
+            minioAccessKey: accessKey,
+            minioSecretKey: secretKey,
+            encrypt: true,
+            passphrase,
+            enableCompression: false,
+            journalFormat: "adaptive-v1",
+            packReadPolicy: "range",
+        });
+
+        const textPath = "adaptive/text.md";
+        const binaryPath = "adaptive/data.bin";
+        await runCliWithInputOrFail(`created-by-a-${suffix}\n`, vaultA, "--settings", settingsA, "put", textPath);
+        await Deno.writeFile(binarySourceA, deterministicBytes(EXTERNAL_PACK_TEST_BYTES, 0x1a2b3c4d));
+        await runCliOrFail(vaultA, "--settings", settingsA, "push", binarySourceA, binaryPath);
+
+        await runCliOrFail(vaultA, "--settings", settingsA, "sync");
+        await runCliOrFail(vaultB, "--settings", settingsB, "sync");
+        assertEquals(
+            sanitiseCatStdout(await runCliOrFail(vaultB, "--settings", settingsB, "cat", textPath)).trimEnd(),
+            `created-by-a-${suffix}`
+        );
+        await runCliOrFail(vaultB, "--settings", settingsB, "pull", binaryPath, binaryDestinationB);
+        await assertFilesEqual(binarySourceA, binaryDestinationB, "Adaptive Journal Range transfer differs");
+
+        await runCliWithInputOrFail(`updated-by-b-${suffix}\n`, vaultB, "--settings", settingsB, "put", textPath);
+        await Deno.writeFile(binarySourceB, deterministicBytes(EXTERNAL_PACK_TEST_BYTES, 0x5e6f7788));
+        await runCliOrFail(vaultB, "--settings", settingsB, "push", binarySourceB, binaryPath);
+        await runCliOrFail(vaultB, "--settings", settingsB, "sync");
+        await runCliOrFail(vaultA, "--settings", settingsA, "sync");
+        assertEquals(
+            sanitiseCatStdout(await runCliOrFail(vaultA, "--settings", settingsA, "cat", textPath)).trimEnd(),
+            `updated-by-b-${suffix}`
+        );
+        await runCliOrFail(vaultA, "--settings", settingsA, "pull", binaryPath, binaryDestinationA);
+        await assertFilesEqual(binarySourceB, binaryDestinationA, "Adaptive Journal whole-Pack transfer differs");
+
+        await runCliOrFail(vaultA, "--settings", settingsA, "rm", binaryPath);
+        await runCliOrFail(vaultA, "--settings", settingsA, "sync");
+        await runCliOrFail(vaultB, "--settings", settingsB, "sync");
+        const deleted = await runCli(vaultB, "--settings", settingsB, "cat", binaryPath);
+        assert(deleted.code !== 0, `Deleted binary remained readable:\n${deleted.combined}`);
+
+        const objectKeys = await listMinioObjectKeys(endpoint, accessKey, secretKey, bucket);
+        assert(objectKeys.includes("a1~manifest.json"), `Adaptive manifest is missing:\n${objectKeys.join("\n")}`);
+        for (const prefix of ["a1~writer~", "a1~pack~", "a1~commit~"]) {
+            assert(
+                objectKeys.some((key) => key.startsWith(prefix)),
+                `Adaptive object with prefix ${prefix} is missing:\n${objectKeys.join("\n")}`
+            );
+        }
+        const packKeys = objectKeys.filter((key) => key.startsWith("a1~pack~"));
+        assert(packKeys.length >= 2, `Expected external Packs from both CLI writers:\n${objectKeys.join("\n")}`);
+        for (const legacyPrefix of ["a1~index~", "a1~delta~", "a1~metadata~"]) {
+            assert(
+                !objectKeys.some((key) => key.startsWith(legacyPrefix)),
+                `Legacy Adaptive object with prefix ${legacyPrefix} was written:\n${objectKeys.join("\n")}`
+            );
+        }
+        const manifest = JSON.parse(
+            await readMinioObjectText(endpoint, accessKey, secretKey, bucket, "a1~manifest.json")
+        ) as { objectLayout?: unknown };
+        assertEquals(manifest.objectLayout, "commit-bundle-v1");
+        assert(
+            !objectKeys.some((key) => key.startsWith("a1~probe~")),
+            `Adaptive capability probe objects were not removed:\n${objectKeys.join("\n")}`
+        );
+        assert(
+            !objectKeys.includes("_00000000-milestone.json"),
+            `Legacy Journal milestone was written into the Adaptive repository:\n${objectKeys.join("\n")}`
+        );
+    } finally {
+        if (!keepDocker) {
+            await stopMinio().catch(() => {});
+        }
+    }
+});