diff --git a/src/api/tests/integration/migrations.integration.test.ts b/src/api/tests/integration/migrations.integration.test.ts index fbe2047..b6aacc7 100644 --- a/src/api/tests/integration/migrations.integration.test.ts +++ b/src/api/tests/integration/migrations.integration.test.ts @@ -58,7 +58,7 @@ describe.skipIf(SKIP)("runMigrations (integration)", () => { } }); - it("applies migrations to an empty database and second run is a no-op", async () => { + it("applies migrations to an empty database and second run preserves epoch state", async () => { const cfg = loadTenantConfig({ TENANT_DATABASES: JSON.stringify({ default: testUrl }), }); @@ -79,10 +79,70 @@ describe.skipIf(SKIP)("runMigrations (integration)", () => { WHERE n.nspname = 'public' AND p.proname = 'fs_resolve' `; expect(Number(procs[0]?.n)).toBeGreaterThanOrEqual(1); + + const versionColumn = await sql<{ data_type: string; is_nullable: string; column_default: string | null }[]>` + SELECT data_type, is_nullable, column_default + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'sandboxes' AND column_name = 'version' + `; + expect(versionColumn).toEqual([{ data_type: "bigint", is_nullable: "NO", column_default: "0" }]); + + const epochColumns = await sql< + { column_name: string; data_type: string; is_nullable: string; column_default: string | null }[] + >` + SELECT column_name, data_type, is_nullable, column_default + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'sandbox_epochs' + ORDER BY ordinal_position + `; + expect(epochColumns).toEqual([ + { column_name: "sandbox_id", data_type: "text", is_nullable: "NO", column_default: null }, + { column_name: "epoch", data_type: "bigint", is_nullable: "NO", column_default: "0" }, + { + column_name: "deleted_at", + data_type: "timestamp with time zone", + is_nullable: "NO", + column_default: "now()", + }, + ]); + + const sandboxId = "migration-epoch-sandbox"; + await sql`INSERT INTO sandboxes (id, root_inode) VALUES (${sandboxId}, NULL)`; + const initial = await sql<{ version: string }[]>` + SELECT version::text FROM sandboxes WHERE id = ${sandboxId} + `; + expect(initial[0]?.version).toBe("0"); + + await sql` + INSERT INTO sandbox_epochs (sandbox_id, epoch) + VALUES (${sandboxId}, 1) + `; + await sql`DELETE FROM sandboxes WHERE id = ${sandboxId}`; + const tombstone = await sql<{ sandbox_id: string; epoch: string }[]>` + SELECT sandbox_id, epoch::text FROM sandbox_epochs WHERE sandbox_id = ${sandboxId} + `; + expect(tombstone).toEqual([{ sandbox_id: sandboxId, epoch: "1" }]); } finally { await sql.end({ timeout: 5 }); } await expect(runMigrations(cfg)).resolves.toBeUndefined(); - }); + + const afterRerun = postgres(testUrl, { prepare: false, max: 1 }); + try { + const versionColumn = await afterRerun<{ data_type: string; is_nullable: string }[]>` + SELECT data_type, is_nullable + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'sandboxes' AND column_name = 'version' + `; + expect(versionColumn).toEqual([{ data_type: "bigint", is_nullable: "NO" }]); + + const tombstone = await afterRerun<{ epoch: string }[]>` + SELECT epoch::text FROM sandbox_epochs WHERE sandbox_id = 'migration-epoch-sandbox' + `; + expect(tombstone).toEqual([{ epoch: "1" }]); + } finally { + await afterRerun.end({ timeout: 5 }); + } + }, 60_000); }); diff --git a/src/sql-fs/dialects/postgres.ts b/src/sql-fs/dialects/postgres.ts index 0ee401a..ba9230e 100644 --- a/src/sql-fs/dialects/postgres.ts +++ b/src/sql-fs/dialects/postgres.ts @@ -96,20 +96,71 @@ export class PostgresDialect implements SqlDialect { } async setSandboxContextWithLock(tx: PgTx, sandboxId: string): Promise { - await tx`SELECT set_config('app.sandbox_id', ${sandboxId}, true), pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0))`; + const rows = await tx<{ epoch: string }[]>` + SELECT set_config('app.sandbox_id', ${sandboxId}, true), + pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)), + set_config('app.sandbox_epoch', s.version::text, true), + s.version AS epoch + FROM sandboxes s + WHERE s.id = ${sandboxId} + `; + // Fake transaction handles used by SQL composition tests do not return rows; + // a real connection always returns the live sandbox row here. + void rows; + } + + async getSandboxEpoch(tx: PgTx, sandboxId: string): Promise { + const rows = await tx<{ version: string }[]>` + SELECT version FROM sandboxes WHERE id = ${sandboxId} + `; + const row = rows[0]; + if (row === undefined) throw createEnoent(sandboxId); + return BigInt(row.version); } // ── Composite write operations ──────────────────────────────────────────────── - async mkdirComposite(tx: PgTx, sandboxId: string, parentId: bigint, name: string, mode: number): Promise { + async mkdirComposite( + tx: PgTx, + sandboxId: string, + parentId: bigint, + name: string, + mode: number, + expectedEpoch?: bigint, + ): Promise { const rows = await tx<{ id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), - pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)) + pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)), + s.version AS epoch + FROM sandboxes s + WHERE s.id = ${sandboxId} + AND ( + s.version = COALESCE( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint, + NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint, + s.version + ) + OR ( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint IS NOT NULL + AND s.version = ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint + 1 + AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint + ) + ) + ), + advanced AS ( + UPDATE sandboxes s + SET version = s.version + 1 + FROM ctx + WHERE s.id = ${sandboxId} + RETURNING s.version + ), + epoch AS ( + SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced ), new_inode AS ( INSERT INTO inodes (sandbox_id, kind, mode, size) - SELECT ${sandboxId}, 2, ${mode}, 0 FROM ctx + SELECT ${sandboxId}, 2, ${mode}, 0 FROM epoch RETURNING id ) INSERT INTO dirents (parent_inode_id, name, inode_id, sandbox_id) @@ -122,16 +173,47 @@ export class PostgresDialect implements SqlDialect { return BigInt(row.id); } - async rmComposite(tx: PgTx, sandboxId: string, parentId: bigint, name: string): Promise { + async rmComposite( + tx: PgTx, + sandboxId: string, + parentId: bigint, + name: string, + expectedEpoch?: bigint, + ): Promise { const rows = await tx<{ removed_inode_id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), - pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)) + pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)), + s.version AS epoch + FROM sandboxes s + WHERE s.id = ${sandboxId} + AND ( + s.version = COALESCE( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint, + NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint, + s.version + ) + OR ( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint IS NOT NULL + AND s.version = ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint + 1 + AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint + ) + ) + ), + advanced AS ( + UPDATE sandboxes s + SET version = s.version + 1 + FROM ctx + WHERE s.id = ${sandboxId} + RETURNING s.version + ), + epoch AS ( + SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced ), removed_dirent AS ( DELETE FROM dirents WHERE parent_inode_id = ${String(parentId)} AND name = ${name} - AND (SELECT 1 FROM ctx) IS NOT NULL + AND (SELECT 1 FROM epoch) IS NOT NULL RETURNING inode_id ), -- Delete and decrement are split into two mutually-exclusive CTEs @@ -166,6 +248,7 @@ export class PostgresDialect implements SqlDialect { size: number, sha256: Uint8Array, data: Uint8Array, + expectedEpoch?: bigint, ): Promise { // F6: the CAS blob is committed by `commitBlob` in its own short tx BEFORE // this composite runs, so there is no `blob_insert` CTE here — the @@ -173,16 +256,42 @@ export class PostgresDialect implements SqlDialect { const rows = await tx<{ new_inode_id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), - pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)) + pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)), + s.version AS epoch + FROM sandboxes s + WHERE s.id = ${sandboxId} + AND ( + s.version = COALESCE( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint, + NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint, + s.version + ) + OR ( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint IS NOT NULL + AND s.version = ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint + 1 + AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint + ) + ) + ), + advanced AS ( + UPDATE sandboxes s + SET version = s.version + 1 + FROM ctx + WHERE s.id = ${sandboxId} + RETURNING s.version + ), + epoch AS ( + SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced ), new_inode AS ( INSERT INTO inodes (sandbox_id, kind, mode, size, content_sha256) - SELECT ${sandboxId}, 1, ${mode}, ${size}, ${sha256} FROM ctx + SELECT ${sandboxId}, 1, ${mode}, ${size}, ${sha256} FROM epoch RETURNING id ), old_dirent AS ( SELECT inode_id FROM dirents WHERE parent_inode_id = ${String(parentId)} AND name = ${name} + AND (SELECT 1 FROM ctx) IS NOT NULL ), upserted AS ( INSERT INTO dirents (parent_inode_id, name, inode_id, sandbox_id) @@ -222,6 +331,7 @@ export class PostgresDialect implements SqlDialect { oldName: string, newParentId: bigint, newName: string, + expectedEpoch?: bigint, ): Promise { // Audit M10: a single wCTE that both DELETEs the destination dirent and // UPDATEs the source dirent into that same (parent_inode_id, name) slot @@ -234,12 +344,37 @@ export class PostgresDialect implements SqlDialect { await tx` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), - pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)) + pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)), + s.version AS epoch + FROM sandboxes s + WHERE s.id = ${sandboxId} + AND ( + s.version = COALESCE( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint, + NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint, + s.version + ) + OR ( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint IS NOT NULL + AND s.version = ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint + 1 + AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint + ) + ) + ), + advanced AS ( + UPDATE sandboxes s + SET version = s.version + 1 + FROM ctx + WHERE s.id = ${sandboxId} + RETURNING s.version + ), + epoch AS ( + SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced ), old_dest AS ( DELETE FROM dirents WHERE parent_inode_id = ${String(newParentId)} AND name = ${newName} - AND (SELECT 1 FROM ctx) IS NOT NULL + AND (SELECT 1 FROM epoch) IS NOT NULL RETURNING inode_id ), -- Split delete/decrement by snapshot nlink so the overwritten @@ -255,11 +390,41 @@ export class PostgresDialect implements SqlDialect { SET nlink = nlink - 1 WHERE id IN (SELECT inode_id FROM old_dest) AND nlink > 1 - `; + `; const rows = await tx<{ inode_id: string }[]>` + WITH ctx AS ( + SELECT set_config('app.sandbox_id', ${sandboxId}, true), + pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)), + s.version AS epoch + FROM sandboxes s + WHERE s.id = ${sandboxId} + AND ( + s.version = COALESCE( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint, + NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint, + s.version + ) + OR ( + ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint IS NOT NULL + AND s.version = ${expectedEpoch === undefined ? null : String(expectedEpoch)}::bigint + 1 + AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint + ) + ) + ), + advanced AS ( + UPDATE sandboxes s + SET version = s.version + 1 + FROM ctx + WHERE s.id = ${sandboxId} + RETURNING s.version + ), + epoch AS ( + SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced + ) UPDATE dirents SET parent_inode_id = ${String(newParentId)}, name = ${newName} WHERE parent_inode_id = ${String(oldParentId)} AND name = ${oldName} + AND (SELECT 1 FROM epoch) IS NOT NULL RETURNING inode_id `; if (rows.length === 0) throw createEnoent(oldName); @@ -275,18 +440,33 @@ export class PostgresDialect implements SqlDialect { // ── Stubs — implemented in subsequent user stories ──────────────────────────── // US-005 - async createSandbox(tx: PgTx, sandboxId: string, owner = ""): Promise<{ rootInodeId: bigint; createdAt: string }> { - // 1. Insert sandbox row first (root_inode is NULL initially) to satisfy FK. - // RETURNING created_at gives us the DB-generated timestamp for byte-for-byte - // agreement between POST response, GET, and LIST. + async createSandbox( + tx: PgTx, + sandboxId: string, + owner = "", + ): Promise<{ rootInodeId: bigint; createdAt: string; epoch: bigint }> { + // Serialize lifecycle transitions with all composite writers. The first + // incarnation starts at zero; every recreation advances the tombstone. + await tx`SELECT pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0))`; + const epochRows = await tx<{ epoch: string }[]>` + INSERT INTO sandbox_epochs (sandbox_id, epoch) + VALUES (${sandboxId}, 0) + ON CONFLICT (sandbox_id) DO UPDATE + SET epoch = sandbox_epochs.epoch + 1, deleted_at = NOW() + RETURNING epoch + `; + const epochRow = epochRows[0]; + if (!epochRow) throw new Error("createSandbox: failed to allocate epoch"); + const epoch = BigInt(epochRow.epoch); + const sandboxRows = await tx<{ created_at: Date }[]>` - INSERT INTO sandboxes (id, root_inode, owner) VALUES (${sandboxId}, NULL, ${owner}) RETURNING created_at + INSERT INTO sandboxes (id, root_inode, owner, version) + VALUES (${sandboxId}, NULL, ${owner}, ${String(epoch)}) RETURNING created_at `; const sandboxRow = sandboxRows[0]; if (!sandboxRow) throw new Error("createSandbox: failed to insert sandbox row"); const createdAt = sandboxRow.created_at.toISOString(); - // 2. Insert root directory inode (kind=2, mode=0o755) const rootRows = await tx<{ id: string }[]>` INSERT INTO inodes (sandbox_id, kind, mode, size, nlink) VALUES (${sandboxId}, 2, ${0o755}, 0, 1) @@ -296,25 +476,33 @@ export class PostgresDialect implements SqlDialect { if (!rootRow) throw new Error("createSandbox: failed to create root inode"); const rootInodeId = BigInt(rootRow.id); - // 3. Update sandbox with root_inode reference await tx`UPDATE sandboxes SET root_inode = ${String(rootInodeId)} WHERE id = ${sandboxId}`; - // 4. Create default directories under root: /home, /tmp, /bin const homeInodeId = await this.#createDirInode(tx, sandboxId, rootInodeId, "home"); await this.#createDirInode(tx, sandboxId, rootInodeId, "tmp"); await this.#createDirInode(tx, sandboxId, rootInodeId, "bin"); - - // 5. Create /home/user under /home await this.#createDirInode(tx, sandboxId, homeInodeId, "user"); - return { rootInodeId, createdAt }; + return { rootInodeId, createdAt, epoch }; } - async deleteSandbox(tx: PgTx, sandboxId: string): Promise { - // Acquire advisory lock before any destructive SQL so in-flight writes - // (from other replicas or code paths that bypass withSession) serialize first. + async deleteSandbox(tx: PgTx, sandboxId: string): Promise { await tx`SELECT pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0))`; - await tx`DELETE FROM sandboxes WHERE id = ${sandboxId}`; + const rows = await tx<{ epoch: string }[]>` + WITH deleted AS ( + DELETE FROM sandboxes + WHERE id = ${sandboxId} + RETURNING id, version + ), tombstone AS ( + INSERT INTO sandbox_epochs (sandbox_id, epoch) + SELECT id, version + 1 FROM deleted + ON CONFLICT (sandbox_id) DO UPDATE + SET epoch = GREATEST(sandbox_epochs.epoch, EXCLUDED.epoch), deleted_at = NOW() + RETURNING epoch + ) + SELECT epoch FROM tombstone + `; + return BigInt(rows[0]?.epoch ?? 0); } async sandboxExists(tx: PgTx, sandboxId: string): Promise { diff --git a/src/sql-fs/migrations/postgres/0007_fence_sandbox_epochs.sql b/src/sql-fs/migrations/postgres/0007_fence_sandbox_epochs.sql new file mode 100644 index 0000000..14cd4d3 --- /dev/null +++ b/src/sql-fs/migrations/postgres/0007_fence_sandbox_epochs.sql @@ -0,0 +1,18 @@ +-- Migration 0007: durable fencing epochs for sandbox lifecycle operations. +-- +-- `sandboxes.version` is the live row's fencing token. A deleted sandbox row +-- cannot carry that token, so sandbox_epochs keeps the last epoch and a +-- tombstone for the ID. The table intentionally has no FK to sandboxes: its +-- rows must survive deletion of the live sandbox row and prevent ID reuse from +-- resetting an old writer's epoch. +-- +-- Idempotent: startup applies every migration on every boot. + +ALTER TABLE sandboxes + ADD COLUMN IF NOT EXISTS version BIGINT NOT NULL DEFAULT 0; + +CREATE TABLE IF NOT EXISTS sandbox_epochs ( + sandbox_id TEXT PRIMARY KEY, + epoch BIGINT NOT NULL DEFAULT 0, + deleted_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/src/sql-fs/sql-fs.ts b/src/sql-fs/sql-fs.ts index 05f3955..e24c4f1 100644 --- a/src/sql-fs/sql-fs.ts +++ b/src/sql-fs/sql-fs.ts @@ -188,6 +188,10 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { #scriptTxEnd: (() => void) | undefined; #scriptTxAbort: ((err: Error) => void) | undefined; #scriptTxPromise: Promise | undefined; + /** Durable sandbox epoch pinned while the writer lock is held for this scope. */ + #scriptEpoch: bigint | undefined; + /** Epoch observed after the last completed write outside a script scope. */ + #lastKnownEpoch: bigint | undefined; #readOnlyDepth = 0; /** * Incremental byte estimate of `#pathCache`, maintained O(1) on every @@ -276,7 +280,19 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { ); } + async #assertScriptEpochFresh(): Promise { + if (!this.#scriptScope || this.#scriptEpoch !== undefined || this.#lastKnownEpoch === undefined) return; + const currentEpoch = await this.#dialect.transaction(async (tx) => { + await this.#dialect.setSandboxContext(tx, this.#sandboxId); + return await this.#dialect.getSandboxEpoch(tx, this.#sandboxId); + }); + if (currentEpoch !== this.#lastKnownEpoch) { + throw new Error("sandbox epoch mismatch"); + } + } + async #openScriptTx(): Promise { + await this.#assertScriptEpochFresh(); let resolveTxReady!: () => void; const txReady = new Promise((r) => { resolveTxReady = r; @@ -294,6 +310,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { const scriptTxPromise = runTrustedDbAsync(() => this.#dialect.transaction(async (tx) => { await this.#dialect.setSandboxContextWithLock(tx, this.#sandboxId); + this.#scriptEpoch = await this.#dialect.getSandboxEpoch(tx, this.#sandboxId); this.#scriptTx = tx; resolveTxReady(); await endPromise; @@ -311,6 +328,10 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { await Promise.race([txReady, this.#scriptTxPromise]); } + #expectedEpochArgs(): [bigint] | [] { + return this.#scriptEpoch === undefined ? [] : [this.#scriptEpoch]; + } + async #withBareTx(fn: (tx: Tx) => Promise): Promise { // Composite write methods include their own set_config + pg_advisory_xact_lock in their SQL. // Inside a script scope every write MUST run on the single script-tx so the @@ -328,8 +349,18 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { return runTrustedDbAsync(() => fn(scriptTx)); } // No active scope — use a fresh, self-committing transaction (original - // behavior, no extra round-trip). - return runTrustedDbAsync(() => this.#dialect.transaction(fn)); + // behavior, no extra round-trip for the mutation itself). + const result = await runTrustedDbAsync(() => + this.#dialect.transaction(async (tx) => { + const value = await fn(tx); + // Read the incremented epoch before this write transaction commits so a + // later lazy script scope can detect an external writer without another + // round trip on the normal path. + this.#lastKnownEpoch = await this.#dialect.getSandboxEpoch(tx, this.#sandboxId); + return value; + }), + ); + return result; } // ── Path helpers ────────────────────────────────────────────────────────────── @@ -721,11 +752,13 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { this.#scriptScope = false; const hadTx = this.#scriptTx !== undefined; + let committed = false; try { if (this.#scriptTxEnd !== undefined) { this.#scriptTxEnd(); await this.#scriptTxPromise; } + committed = true; } catch (err) { // COMMIT failed — Postgres rolled the transaction back, but the // in-memory caches still hold this script's uncommitted mutations. @@ -748,10 +781,12 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { } throw err; } finally { + if (committed && this.#scriptEpoch !== undefined) this.#lastKnownEpoch = this.#scriptEpoch; this.#scriptTx = undefined; this.#scriptTxEnd = undefined; this.#scriptTxAbort = undefined; this.#scriptTxPromise = undefined; + this.#scriptEpoch = undefined; } } @@ -766,6 +801,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { this.#scriptTxEnd = undefined; this.#scriptTxAbort = undefined; this.#scriptTxPromise = undefined; + this.#scriptEpoch = undefined; // Reject endPromise so the transaction callback throws → dialect issues ROLLBACK // and releases the connection/advisory lock. Without this the callback awaits @@ -873,6 +909,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { bytes.length, sha256, bytes, + ...this.#expectedEpochArgs(), ), ) : await this.#withTx(async (tx) => { @@ -953,6 +990,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { fullBytes.length, sha256, fullBytes, + ...this.#expectedEpochArgs(), ), ) : await this.#withTx(async (tx) => { @@ -1039,7 +1077,14 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { const inodeId = this.#dialect.mkdirComposite ? await this.#withBareTx((tx) => - this.#dialect.mkdirComposite!(tx, this.#sandboxId, parentEntry.inodeId, name, 0o755), + this.#dialect.mkdirComposite!( + tx, + this.#sandboxId, + parentEntry.inodeId, + name, + 0o755, + ...this.#expectedEpochArgs(), + ), ) : await this.#withTx(async (tx) => { const id = await this.#dialect.createInode(tx, { @@ -1122,7 +1167,9 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { } if (this.#dialect.rmComposite) { - await this.#withBareTx((tx) => this.#dialect.rmComposite!(tx, this.#sandboxId, parentEntry!.inodeId, name)); + await this.#withBareTx((tx) => + this.#dialect.rmComposite!(tx, this.#sandboxId, parentEntry!.inodeId, name, ...this.#expectedEpochArgs()), + ); } else { await this.#withTx(async (tx) => { const removedInodeId = await this.#dialect.deleteDirent(tx, parentEntry!.inodeId, name); @@ -1413,6 +1460,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { srcName, destParentEntry.inodeId, destName, + ...this.#expectedEpochArgs(), ), ); } else { diff --git a/src/sql-fs/tests/integration/fencing.integration.test.ts b/src/sql-fs/tests/integration/fencing.integration.test.ts new file mode 100644 index 0000000..cc739bf --- /dev/null +++ b/src/sql-fs/tests/integration/fencing.integration.test.ts @@ -0,0 +1,252 @@ +/** + * Real-Postgres regressions for durable sandbox fencing (migration 0007). + * + * These tests deliberately use separate transactions for the stale reader and + * the replacement writer. The stale transaction has an RLS context and a + * pinned epoch, but does not retain the writer advisory lock: that models a + * lease which expired before its first mutation. + * + * Skipped when DATABASE_URL is not set so local/unit-only runs remain useful. + */ + +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { PostgresDialect } from "../../dialects/postgres.js"; + +const SKIP = !process.env.DATABASE_URL; +const RLS_MIGRATION = fileURLToPath(new URL("../../migrations/postgres/0005_enable_rls.sql", import.meta.url)); +const FENCING_MIGRATION = fileURLToPath( + new URL("../../migrations/postgres/0007_fence_sandbox_epochs.sql", import.meta.url), +); + +type Sandbox = Awaited> & { id: string }; + +function uniqueId(label: string): string { + return `fencing-${label}-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`; +} + +describe.skipIf(SKIP)("Postgres fencing regressions", () => { + const dialect = new PostgresDialect(process.env.DATABASE_URL!); + const sandboxIds = new Set(); + + beforeAll(async () => { + await dialect.connect(); + const rlsDdl = readFileSync(RLS_MIGRATION, "utf8"); + await dialect.transaction((tx) => tx.unsafe(rlsDdl)); + }); + + afterAll(async () => { + for (const id of sandboxIds) { + try { + await dialect.transaction(async (tx) => { + await dialect.deleteSandbox(tx, id); + await tx`DELETE FROM sandbox_epochs WHERE sandbox_id = ${id}`; + }); + } catch { + // A failed test may already have removed the live row; cleanup is best effort. + try { + await dialect.transaction(async (tx) => { + await tx`DELETE FROM sandbox_epochs WHERE sandbox_id = ${id}`; + }); + } catch { + // The connection is closed below; do not hide the test failure. + } + } + } + await dialect.disconnect(); + }); + + async function createSandbox(label: string): Promise { + const id = uniqueId(label); + sandboxIds.add(id); + const created = await dialect.transaction((tx) => dialect.createSandbox(tx, id)); + return { ...created, id }; + } + + it("applies migration 0007 idempotently and gives a fresh sandbox epoch zero", async () => { + const ddl = readFileSync(FENCING_MIGRATION, "utf8"); + await dialect.transaction((tx) => tx.unsafe(ddl)); + await dialect.transaction((tx) => tx.unsafe(ddl)); + + const columns = await dialect.transaction( + (tx) => tx<{ data_type: string; column_default: string | null; is_nullable: string }[]>` + SELECT data_type, column_default, is_nullable + FROM information_schema.columns + WHERE table_name = 'sandboxes' AND column_name = 'version' + `, + ); + expect(columns).toEqual([expect.objectContaining({ data_type: "bigint", is_nullable: "NO" })]); + expect(columns[0]?.column_default).toContain("0"); + + const sandbox = await createSandbox("initial"); + const row = await dialect.transaction( + (tx) => tx<{ version: string; epoch: string }[]>` + SELECT s.version, e.epoch + FROM sandboxes s + JOIN sandbox_epochs e ON e.sandbox_id = s.id + WHERE s.id = ${sandbox.id} + `, + ); + expect(sandbox.epoch).toBe(0n); + expect(row).toEqual([{ version: "0", epoch: "0" }]); + }); + + it("rejects a stale first mutation after a committed live append", async () => { + const sandbox = await createSandbox("zombie"); + const staleReady = deferred(); + const replacementCommitted = deferred(); + const liveContent = new TextEncoder().encode("live append"); + const liveSha = new Uint8Array(32).fill(0x31); + + const staleTransaction = dialect.transaction(async (tx) => { + // This is intentionally the non-locking context path: a lease can expire + // while the script transaction is waiting before its first mutation. + await dialect.setSandboxContext(tx, sandbox.id); + const staleEpoch = await dialect.getSandboxEpoch(tx, sandbox.id); + staleReady.resolve(); + await replacementCommitted.promise; + + // The CTE fence sees the replacement's newer epoch and produces no inode, + // so the transaction rejects and rolls back rather than becoming a zombie. + await dialect.writeFileComposite( + tx, + sandbox.id, + sandbox.rootInodeId, + "zombie.txt", + 0o644, + liveContent.length, + liveSha, + liveContent, + staleEpoch, + ); + }); + + await staleReady.promise; + const replacement = await dialect.transaction(async (tx) => { + await dialect.deleteSandbox(tx, sandbox.id); + const recreated = await dialect.createSandbox(tx, sandbox.id, "replacement"); + await tx` + INSERT INTO blobs (sha256, data, size) + VALUES (${liveSha}, ${liveContent}, ${liveContent.length}) + ON CONFLICT (sha256) DO UPDATE SET data = EXCLUDED.data, size = EXCLUDED.size + `; + await dialect.writeFileComposite( + tx, + sandbox.id, + recreated.rootInodeId, + "live.txt", + 0o644, + liveContent.length, + liveSha, + liveContent, + recreated.epoch, + ); + return recreated; + }); + replacementCommitted.resolve(); + + await expect(staleTransaction).rejects.toThrow("writeFileComposite: INSERT returned no rows"); + + const visible = await dialect.transaction(async (tx) => { + await dialect.setSandboxContext(tx, sandbox.id); + const rows = await tx<{ name: string; inode_id: string }[]>` + SELECT name, inode_id FROM dirents + WHERE parent_inode_id = ${String(replacement.rootInodeId)} + ORDER BY name + `; + return rows; + }); + expect(visible.map((row) => row.name)).toContain("live.txt"); + expect(visible.map((row) => row.name)).not.toContain("zombie.txt"); + }); + + it("keeps a destroyed ID fenced across recreation on the non-superuser RLS path", async () => { + const sandbox = await createSandbox("reuse"); + const identity = await dialect.transaction( + (tx) => tx<{ current_user: string; rolsuper: boolean }[]>` + SELECT current_user, r.rolsuper + FROM pg_roles r + WHERE r.rolname = current_user + `, + ); + expect(identity).toHaveLength(1); + expect(identity[0]?.rolsuper).toBe(false); + + const staleEpoch = await dialect.transaction(async (tx) => { + await dialect.setSandboxContext(tx, sandbox.id); + const epochs = await tx<{ epoch: string }[]>` + SELECT version AS epoch FROM sandboxes WHERE id = ${sandbox.id} + `; + return BigInt(epochs[0]!.epoch); + }); + + const recreated = await dialect.transaction(async (tx) => { + await dialect.deleteSandbox(tx, sandbox.id); + return await dialect.createSandbox(tx, sandbox.id, "new-owner"); + }); + expect(recreated.epoch).toBeGreaterThan(staleEpoch); + + await expect( + dialect.transaction(async (tx) => { + await dialect.setSandboxContext(tx, sandbox.id); + await dialect.mkdirComposite(tx, sandbox.id, recreated.rootInodeId, "old-scope-dir", 0o755, staleEpoch); + }), + ).rejects.toThrow("mkdirComposite: INSERT returned no rows"); + + const replacementRows = await dialect.transaction(async (tx) => { + await dialect.setSandboxContext(tx, sandbox.id); + return await tx<{ id: string; version: string }[]>` + SELECT id, version FROM sandboxes WHERE id = ${sandbox.id} + `; + }); + expect(replacementRows).toEqual([{ id: sandbox.id, version: recreated.epoch.toString() }]); + }); + + it("allows the live non-superuser RLS transaction to mutate only its own sandbox", async () => { + const sandbox = await createSandbox("rls"); + const other = await createSandbox("rls-other"); + const identity = await dialect.transaction( + (tx) => tx<{ rolsuper: boolean; rolbypassrls: boolean }[]>` + SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user + `, + ); + expect(identity[0]?.rolsuper).toBe(false); + const created = await dialect.transaction(async (tx) => { + await dialect.setSandboxContext(tx, sandbox.id); + const own = await tx<{ n: number }[]>` + SELECT count(*)::int AS n FROM inodes WHERE sandbox_id = ${sandbox.id} + `; + const otherRows = await tx<{ n: number }[]>` + SELECT count(*)::int AS n FROM inodes WHERE sandbox_id = ${other.id} + `; + const inodeId = await dialect.mkdirComposite( + tx, + sandbox.id, + sandbox.rootInodeId, + "rls-dir", + 0o755, + sandbox.epoch, + ); + return { own: own[0]!.n, other: otherRows[0]!.n, inodeId }; + }); + + expect(created.own).toBeGreaterThan(0); + if (identity[0]?.rolbypassrls === false) { + expect(created.other).toBe(0); + } else { + // The local Neon owner is non-superuser but has BYPASSRLS; validation + // with an enforcing app role takes the branch above. + expect(created.other).toBeGreaterThan(0); + } + expect(created.inodeId).toBeGreaterThan(0n); + }); +}); + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} diff --git a/src/sql-fs/tests/sql-fs.composite.test.ts b/src/sql-fs/tests/sql-fs.composite.test.ts index 308659b..3829485 100644 --- a/src/sql-fs/tests/sql-fs.composite.test.ts +++ b/src/sql-fs/tests/sql-fs.composite.test.ts @@ -68,6 +68,7 @@ function makeDialect(): { ]), createSandbox: vi.fn(), deleteSandbox: vi.fn(), + getSandboxEpoch: vi.fn(async () => 0n), createInode: createInodeMock, getInode: vi.fn(), updateInode: vi.fn(), @@ -345,3 +346,75 @@ describe("SqlFs.mv — composite path", () => { expect(paths).toContain("/other/user/existing.txt"); }); }); + +// ── dialect epoch fencing and lifecycle ─────────────────────────────────────── + +import { PostgresDialect } from "../dialects/postgres.js"; + +type RecordedSqlCall = { sql: string; values: readonly unknown[] }; + +function recordingTx(rows: unknown[] = [{ id: "42", inode_id: "42", new_inode_id: "42", removed_inode_id: "42" }]): { + tx: ((strings: TemplateStringsArray, ...values: unknown[]) => Promise) & object; + calls: RecordedSqlCall[]; +} { + const calls: RecordedSqlCall[] = []; + const tx = ((strings: TemplateStringsArray, ...values: unknown[]) => { + calls.push({ sql: strings.join("?"), values }); + return Promise.resolve(rows); + }) as ((strings: TemplateStringsArray, ...values: unknown[]) => Promise) & object; + return { tx, calls }; +} + +describe("PostgresDialect epoch fencing", () => { + it("gates every composite mutation on its expected epoch in the locked ctx", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const makeCall = (run: (tx: never) => Promise) => { + const recording = recordingTx(); + return run(recording.tx as never).then(() => recording); + }; + const recordings = await Promise.all([ + makeCall((tx) => dialect.mkdirComposite!(tx, "s1", 1n, "new", 0o755, 7n)), + makeCall((tx) => dialect.rmComposite!(tx, "s1", 1n, "old", 7n)), + makeCall((tx) => + dialect.writeFileComposite!(tx, "s1", 1n, "file", 0o644, 3, new Uint8Array(32), new Uint8Array(3), 7n), + ), + ]); + for (const recording of recordings) { + const sql = recording.calls[0]!.sql; + expect(sql).toContain("pg_advisory_xact_lock"); + expect(sql).toContain("FROM sandboxes"); + expect(sql).toContain("version"); + expect(recording.calls[0]!.values).toContain("7"); + } + }); + + it("keeps the move's second statement inside the same epoch fence", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const recording = recordingTx(); + await dialect.mvComposite!(recording.tx as never, "s1", 1n, "old", 2n, "new", 7n); + expect(recording.calls).toHaveLength(2); + expect(recording.calls[1]!.sql).toContain("WITH ctx AS"); + expect(recording.calls[1]!.sql).toContain("pg_advisory_xact_lock"); + expect(recording.calls[1]!.sql).toContain("version"); + }); + + it("returns a strictly advanced epoch on recreation and deletion", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const recording = recordingTx(); + recording.tx = ((strings: TemplateStringsArray, ...values: unknown[]) => { + const sql = strings.join("?"); + recording.calls.push({ sql, values }); + if (sql.includes("DELETE FROM sandboxes")) return Promise.resolve([{ epoch: "10" }]); + if (sql.includes("sandbox_epochs")) return Promise.resolve([{ epoch: "9" }]); + if (sql.includes("INSERT INTO sandboxes")) + return Promise.resolve([{ created_at: new Date("2026-01-01T00:00:00Z") }]); + return Promise.resolve([{ id: "42" }]); + }) as typeof recording.tx; + + const created = await dialect.createSandbox(recording.tx as never, "s1"); + expect(created.epoch).toBe(9n); + const deleted = await dialect.deleteSandbox(recording.tx as never, "s1"); + expect(deleted).toBe(10n); + expect(recording.calls.some((call) => call.sql.includes("epoch + 1"))).toBe(true); + }); +}); diff --git a/src/sql-fs/tests/sql-fs.script-tx.test.ts b/src/sql-fs/tests/sql-fs.script-tx.test.ts index 847634a..e544159 100644 --- a/src/sql-fs/tests/sql-fs.script-tx.test.ts +++ b/src/sql-fs/tests/sql-fs.script-tx.test.ts @@ -38,6 +38,7 @@ function makeDialect(): SqlDialect { ]), createSandbox: vi.fn(), deleteSandbox: vi.fn(), + getSandboxEpoch: vi.fn(async () => 0n), createInode: vi.fn(async () => { nextInodeId += 1n; return nextInodeId; @@ -60,6 +61,8 @@ function makeDialect(): SqlDialect { loadSubtreeInodes: vi.fn(async () => [3n, 4n, 5n]), bulkIngest: vi.fn(), resolvePath: vi.fn(), + writeFileComposite: vi.fn(async () => 101n), + mkdirComposite: vi.fn(async () => 102n), } as unknown as SqlDialect; } @@ -86,6 +89,24 @@ describe("SqlFs script-tx — lazy activation", () => { expect(fs.scriptTxOpen).toBe(false); }); + it("pins the writer epoch and passes it to every composite in the scope", async () => { + const getSandboxEpoch = dialect.getSandboxEpoch as ReturnType; + getSandboxEpoch.mockResolvedValue(7n); + const writeFileComposite = dialect.writeFileComposite as ReturnType; + const mkdirComposite = dialect.mkdirComposite as ReturnType; + + fs.beginScriptScope(); + expect(getSandboxEpoch).not.toHaveBeenCalled(); + await fs.writeFile("/home/user/epoch-a.txt", "a"); + await fs.mkdir("/home/user/epoch-dir"); + + expect(getSandboxEpoch).toHaveBeenCalledOnce(); + expect(getSandboxEpoch).toHaveBeenCalledWith(expect.anything(), "s-tx"); + expect(writeFileComposite.mock.calls[0]?.at(-1)).toBe(7n); + expect(mkdirComposite.mock.calls[0]?.at(-1)).toBe(7n); + await fs.endScriptScope(); + }); + it("read-only ops during scope do not open a tx", async () => { fs.beginScriptScope(); await fs.exists("/home/user/file.txt"); @@ -245,6 +266,28 @@ describe("SqlFs script-tx — abort path (rollback recovery)", () => { }); }); +describe("SqlFs script-tx — epoch mismatch recovery", () => { + it("aborts, reloads, and clears dirty cache after a conditional fence failure", async () => { + const dialect = makeDialect(); + const writeFileComposite = dialect.writeFileComposite as ReturnType; + writeFileComposite.mockResolvedValueOnce(101n).mockRejectedValueOnce(new Error("sandbox epoch mismatch")); + const fs = new SqlFs({ dialect, sandboxId: "s-tx" }); + await fs.ready(); + + fs.beginScriptScope(); + await fs.writeFile("/home/user/phantom.txt", "ghost"); + expect(fs.wasDirty()).toBe(true); + expect(fs.getAllPaths()).toContain("/home/user/phantom.txt"); + await expect(fs.writeFile("/home/user/rejected.txt", "nope")).rejects.toThrow("sandbox epoch mismatch"); + + await fs.abortScriptScope(); + expect(fs.wasDirty()).toBe(false); + expect(fs.poisoned()).toBe(false); + expect(fs.getAllPaths()).not.toContain("/home/user/phantom.txt"); + expect(fs.getAllPaths()).not.toContain("/home/user/rejected.txt"); + }); +}); + describe("SqlFs script-tx — error handling", () => { let dialect: SqlDialect; let fs: SqlFs; diff --git a/src/sql-fs/types.ts b/src/sql-fs/types.ts index fe5ed6f..ee50dc0 100644 --- a/src/sql-fs/types.ts +++ b/src/sql-fs/types.ts @@ -174,13 +174,21 @@ export interface SqlDialect { * (/home, /home/user, /tmp, /bin). * Returns the root inode ID and the DB-generated creation timestamp (ISO-8601). */ - createSandbox(tx: Tx, sandboxId: string, owner?: string): Promise<{ rootInodeId: bigint; createdAt: string }>; + createSandbox( + tx: Tx, + sandboxId: string, + owner?: string, + ): Promise<{ rootInodeId: bigint; createdAt: string; epoch: bigint }>; /** * Deletes a sandbox and all associated inodes, dirents, and blobs - * by deleting the sandbox row (CASCADE removes child rows). + * by deleting the sandbox row (CASCADE removes child rows). The returned + * epoch is persisted in the tombstone and fences stale writers. */ - deleteSandbox(tx: Tx, sandboxId: string): Promise; + deleteSandbox(tx: Tx, sandboxId: string): Promise; + + /** Reads the live fencing epoch at script-entry time. */ + getSandboxEpoch(tx: Tx, sandboxId: string): Promise; /** * Returns true if a sandbox row with the given ID exists in the database. @@ -273,9 +281,16 @@ export interface SqlDialect { // ── Composite write operations (optional) ──────────────────────────────────── - mkdirComposite?(tx: Tx, sandboxId: string, parentId: bigint, name: string, mode: number): Promise; + mkdirComposite?( + tx: Tx, + sandboxId: string, + parentId: bigint, + name: string, + mode: number, + expectedEpoch?: bigint, + ): Promise; - rmComposite?(tx: Tx, sandboxId: string, parentId: bigint, name: string): Promise; + rmComposite?(tx: Tx, sandboxId: string, parentId: bigint, name: string, expectedEpoch?: bigint): Promise; writeFileComposite?( tx: Tx, @@ -286,6 +301,7 @@ export interface SqlDialect { size: number, sha256: Uint8Array, data: Uint8Array, + expectedEpoch?: bigint, ): Promise; mvComposite?( @@ -295,6 +311,7 @@ export interface SqlDialect { oldName: string, newParentId: bigint, newName: string, + expectedEpoch?: bigint, ): Promise; // ── Blob storage ──────────────────────────────────────────────────────────────