From aac35780708da582b0107eeaf5e7586ff8a31c6d Mon Sep 17 00:00:00 2001 From: Trek Worker Date: Fri, 10 Jul 2026 21:09:47 +0000 Subject: [PATCH 1/5] Add sandbox version epoch migration and open-time plumbing --- src/api/session-manager.ts | 30 ++++++++++++++----- .../unit/session-manager.rehydrate.test.ts | 12 +++++++- src/sql-fs/dialects/postgres.ts | 4 ++- .../postgres/0007_add_sandbox_version.sql | 5 ++++ src/sql-fs/types.ts | 2 ++ 5 files changed, 44 insertions(+), 9 deletions(-) create mode 100644 src/sql-fs/migrations/postgres/0007_add_sandbox_version.sql diff --git a/src/api/session-manager.ts b/src/api/session-manager.ts index e517282..3270d20 100644 --- a/src/api/session-manager.ts +++ b/src/api/session-manager.ts @@ -154,6 +154,9 @@ export interface RuntimeOptions { readonly network: boolean; } +/** Sandbox metadata at open time, including the pinned epoch read from Postgres. */ +type SandboxOpenMeta = SandboxMeta & { readonly version?: number }; + const DEFAULT_RUNTIME_OPTIONS: RuntimeOptions = { python: false, javascript: false, network: false }; export function buildSandboxBaseEnv(env: NodeJS.ProcessEnv = process.env): Record { @@ -219,6 +222,8 @@ export interface Session { readonly scriptTx: SessionScopedFs | undefined; lastUsed: number; inFlight: number; + /** Pinned sandbox epoch exposed at script-open time. */ + sandboxVersion: number; /** * Per-session async readers-writer lock. Writes (default exec path, * destroy, reaper, shutdown) take exclusive mode; readOnly execs take @@ -502,16 +507,18 @@ export class SessionManager { tenantId: string, sandboxId: string, owner = "", - ): Promise<{ fs: IFileSystem; resolvedOwner: string; createdAt: string }> { + ): Promise<{ fs: IFileSystem; resolvedOwner: string; createdAt: string; version: number }> { if (this.createFsOverride !== undefined) { + const meta = this.getSandboxMetaFn !== undefined ? await this.getSandboxMetaFn(tenantId, sandboxId) : null; return { fs: await this.createFsOverride(tenantId, sandboxId), resolvedOwner: owner, createdAt: new Date().toISOString(), + version: meta?.version ?? 0, }; } const backend = this.getOrInitBackend(tenantId); - return createPostgresSandboxFs( + const fs = await createPostgresSandboxFs( { connectionString: backend.connectionString, tenantId, @@ -522,6 +529,13 @@ export class SessionManager { sandboxId, owner, ); + const meta = this.getSandboxMetaFn !== undefined ? await this.getSandboxMetaFn(tenantId, sandboxId) : null; + return { + fs, + resolvedOwner: owner, + createdAt: new Date().toISOString(), + version: meta?.version ?? 0, + }; } private estimatePathCacheBytes(fs: IFileSystem): number { @@ -567,7 +581,7 @@ export class SessionManager { const creationPromise = (async (): Promise => { let createdFs: IFileSystem | undefined; try { - const { fs, resolvedOwner, createdAt: fsCreatedAt } = await this.buildFs(tenantId, sandboxId, owner); + const { fs, resolvedOwner, createdAt: fsCreatedAt, version: sandboxVersion } = await this.buildFs(tenantId, sandboxId, owner); createdFs = fs; const defenseInDepthConfig: DefenseInDepthConfig | false = this.defenseInDepth ? { @@ -659,6 +673,7 @@ export class SessionManager { pathCacheBytes, overBudget: pathCacheBytes > this.pathCacheMaxBytes, lastSeenVersion: initialVersion, + sandboxVersion, publishPending: false, cwd: bash.getCwd(), }; @@ -1179,9 +1194,9 @@ export class SessionManager { runtimeOptions?: RuntimeOptions, lostSignal?: AbortSignal, ): Promise { - let meta: SandboxMeta | null | undefined; + let meta: SandboxOpenMeta | null | undefined; if (this.getSandboxMetaFn !== undefined) { - meta = await this.getSandboxMetaFn(tenantId, sandboxId); + meta = (await this.getSandboxMetaFn(tenantId, sandboxId)) as SandboxOpenMeta | null; if (meta === null) { throw Object.assign(new Error(`ENOENT: sandbox ${sandboxId} not found`), { code: "ENOENT" }); } @@ -1195,6 +1210,7 @@ export class SessionManager { if (meta?.owner) session.owner = meta.owner; if (meta?.name !== undefined) session.name = meta.name; if (meta?.createdAt !== undefined) session.createdAt = meta.createdAt; + if (meta?.version !== undefined) session.sandboxVersion = meta.version; return this.withSessionReadEntry(tenantId, sandboxId, session, fn, lostSignal); } @@ -1227,9 +1243,9 @@ export class SessionManager { runtimeOptions?: RuntimeOptions, lostSignal?: AbortSignal, ): Promise { - let meta: SandboxMeta | null | undefined; + let meta: SandboxOpenMeta | null | undefined; if (this.getSandboxMetaFn !== undefined) { - meta = await this.getSandboxMetaFn(tenantId, sandboxId); + meta = (await this.getSandboxMetaFn(tenantId, sandboxId)) as SandboxOpenMeta | null; if (meta === null) { throw Object.assign(new Error(`ENOENT: sandbox ${sandboxId} not found`), { code: "ENOENT" }); } diff --git a/src/api/tests/unit/session-manager.rehydrate.test.ts b/src/api/tests/unit/session-manager.rehydrate.test.ts index 4957108..f08f3a0 100644 --- a/src/api/tests/unit/session-manager.rehydrate.test.ts +++ b/src/api/tests/unit/session-manager.rehydrate.test.ts @@ -12,7 +12,9 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { SandboxMeta } from "../../../sql-fs/types.js"; import { SessionManager } from "../../session-manager.js"; -let pgSandboxes: Map; +type SandboxMetaWithEpoch = SandboxMeta & { version?: number }; + +let pgSandboxes: Map; let createFsSpy: ReturnType; @@ -121,6 +123,14 @@ describe("SessionManager.withSessionOrRehydrate()", () => { expect(capturedRuntime).toEqual({ python: true, javascript: false, network: false }); }); + it("pins sandbox version from PG metadata at open time", async () => { + pgSandboxes.set("sb-epoch", { owner: null, name: null, python: false, javascript: false, network: false, version: 41 }); + + await sm.withSessionOrRehydrate("default", "sb-epoch", async (session) => { + expect(session.sandboxVersion).toBe(41); + }); + }); + it("persistSandboxMeta writes to store and is readable on rehydration", async () => { await sm.withSession("default", "sb-meta", async (session) => { session.owner = "creator"; diff --git a/src/sql-fs/dialects/postgres.ts b/src/sql-fs/dialects/postgres.ts index 0ee401a..1fdcc66 100644 --- a/src/sql-fs/dialects/postgres.ts +++ b/src/sql-fs/dialects/postgres.ts @@ -337,10 +337,11 @@ export class PostgresDialect implements SqlDialect { python: boolean; javascript: boolean; network: boolean; + version: string | number; created_at: Date; }[] >` - SELECT owner, name, python, javascript, network, created_at FROM sandboxes WHERE id = ${sandboxId} + SELECT owner, name, python, javascript, network, version, created_at FROM sandboxes WHERE id = ${sandboxId} `; if (rows.length === 0) return null; const r = rows[0]!; @@ -350,6 +351,7 @@ export class PostgresDialect implements SqlDialect { python: r.python, javascript: r.javascript, network: r.network, + version: Number(r.version) || 0, createdAt: r.created_at.toISOString(), }; } catch (err) { diff --git a/src/sql-fs/migrations/postgres/0007_add_sandbox_version.sql b/src/sql-fs/migrations/postgres/0007_add_sandbox_version.sql new file mode 100644 index 0000000..4402e8d --- /dev/null +++ b/src/sql-fs/migrations/postgres/0007_add_sandbox_version.sql @@ -0,0 +1,5 @@ +-- Migration 0007: Add sandbox version epoch used to pin script-open state. +-- Existing sandboxes start at epoch 0; writers bump this when new durable state +-- is published. +ALTER TABLE sandboxes + ADD COLUMN IF NOT EXISTS version BIGINT NOT NULL DEFAULT 0; diff --git a/src/sql-fs/types.ts b/src/sql-fs/types.ts index fe5ed6f..f28ea91 100644 --- a/src/sql-fs/types.ts +++ b/src/sql-fs/types.ts @@ -65,6 +65,8 @@ export interface SandboxMeta { readonly javascript: boolean; /** When true, js-exec fetch() can reach external HTTP endpoints (60 s timeout). */ readonly network: boolean; + /** Pinned sandbox epoch from the sandboxes row at open time. */ + readonly version?: number; /** ISO-8601 timestamp of when the sandbox was originally created (from DB created_at). */ readonly createdAt?: string; } From 926cda110606a91475bf8bf8aacd3a6b0dad6c4f Mon Sep 17 00:00:00 2001 From: Trek Worker Date: Fri, 10 Jul 2026 23:15:49 +0000 Subject: [PATCH 2/5] Implement fenced script transaction version threading and composite-write optimistic fencing --- src/api/session-manager.ts | 6 +- src/sql-fs/dialects/postgres.ts | 33 +++++++- src/sql-fs/index.ts | 2 + src/sql-fs/sql-fs.ts | 28 ++++--- .../tests/unit/sql-fs.script-fencing.test.ts | 84 +++++++++++++++++++ src/sql-fs/types.ts | 8 +- 6 files changed, 142 insertions(+), 19 deletions(-) create mode 100644 src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts diff --git a/src/api/session-manager.ts b/src/api/session-manager.ts index 3270d20..33d7d24 100644 --- a/src/api/session-manager.ts +++ b/src/api/session-manager.ts @@ -518,6 +518,8 @@ export class SessionManager { }; } const backend = this.getOrInitBackend(tenantId); + const meta = this.getSandboxMetaFn !== undefined ? await this.getSandboxMetaFn(tenantId, sandboxId) : null; + const sandboxVersion = meta?.version ?? 0; const fs = await createPostgresSandboxFs( { connectionString: backend.connectionString, @@ -528,13 +530,13 @@ export class SessionManager { }, sandboxId, owner, + sandboxVersion, ); - const meta = this.getSandboxMetaFn !== undefined ? await this.getSandboxMetaFn(tenantId, sandboxId) : null; return { fs, resolvedOwner: owner, createdAt: new Date().toISOString(), - version: meta?.version ?? 0, + version: sandboxVersion, }; } diff --git a/src/sql-fs/dialects/postgres.ts b/src/sql-fs/dialects/postgres.ts index 1fdcc66..45821e7 100644 --- a/src/sql-fs/dialects/postgres.ts +++ b/src/sql-fs/dialects/postgres.ts @@ -95,13 +95,30 @@ export class PostgresDialect implements SqlDialect { await tx`SELECT set_config('app.sandbox_id', ${sandboxId}, true)`; } - async setSandboxContextWithLock(tx: PgTx, sandboxId: string): Promise { + async setSandboxContextWithLock(tx: PgTx, sandboxId: string, sandboxVersion?: number): Promise { await tx`SELECT set_config('app.sandbox_id', ${sandboxId}, true), pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0))`; + if (sandboxVersion !== undefined) await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); + } + + async #acquireSandboxWriteFence(tx: PgTx, sandboxId: string): Promise { + await tx`SELECT set_config('app.sandbox_id', ${sandboxId}, true), pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0))`; + } + + async #assertSandboxVersion(tx: PgTx, sandboxId: string, sandboxVersion?: number): Promise { + if (sandboxVersion === undefined) return; + const rows = await tx<{ version: string | number }[]>` + SELECT version FROM sandboxes WHERE id = ${sandboxId} + `; + if (rows.length === 0 || Number(rows[0]!.version) !== sandboxVersion) { + throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); + } } // ── 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, sandboxVersion?: number): Promise { + await this.#acquireSandboxWriteFence(tx, sandboxId); + await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); const rows = await tx<{ id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), @@ -122,7 +139,9 @@ 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, sandboxVersion?: number): Promise { + await this.#acquireSandboxWriteFence(tx, sandboxId); + await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); const rows = await tx<{ removed_inode_id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), @@ -166,10 +185,13 @@ export class PostgresDialect implements SqlDialect { size: number, sha256: Uint8Array, data: Uint8Array, + sandboxVersion?: number, ): 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 // script-tx must not hold the hot-blob `ON CONFLICT DO UPDATE` tuple lock. + await this.#acquireSandboxWriteFence(tx, sandboxId); + await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); const rows = await tx<{ new_inode_id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), @@ -222,6 +244,7 @@ export class PostgresDialect implements SqlDialect { oldName: string, newParentId: bigint, newName: string, + sandboxVersion?: number, ): 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 @@ -231,6 +254,8 @@ export class PostgresDialect implements SqlDialect { // 1 frees the destination slot (and cleans its inode), statement 2 renames // the source into it. Statement 1's effects are visible to statement 2, so // the rename can no longer collide. Atomicity is preserved by the tx. + await this.#acquireSandboxWriteFence(tx, sandboxId); + await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); await tx` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), @@ -584,6 +609,8 @@ export class PostgresDialect implements SqlDialect { `; // Move the source dirent via a single UPDATE + await this.#acquireSandboxWriteFence(tx, sandboxId); + await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); const rows = await tx<{ inode_id: string }[]>` UPDATE dirents SET parent_inode_id = ${String(newParentId)}, name = ${newName} diff --git a/src/sql-fs/index.ts b/src/sql-fs/index.ts index 5d82549..f44f305 100644 --- a/src/sql-fs/index.ts +++ b/src/sql-fs/index.ts @@ -51,6 +51,7 @@ export async function createPostgresSandboxFs( opts: PostgresBackendOptions, sandboxId: string, owner = "", + sandboxVersion?: number, ): Promise<{ fs: IFileSystem; resolvedOwner: string; createdAt: string }> { const dialect = new PostgresDialect(opts.connectionString, opts.blobCache); await dialect.connect(); @@ -79,6 +80,7 @@ export async function createPostgresSandboxFs( redis: opts.redis, pathSnapshot: opts.pathSnapshot, blobCache: opts.blobCache, + sandboxVersion, }); await fs.ready(); return { fs, resolvedOwner, createdAt }; diff --git a/src/sql-fs/sql-fs.ts b/src/sql-fs/sql-fs.ts index 05f3955..ed1cba3 100644 --- a/src/sql-fs/sql-fs.ts +++ b/src/sql-fs/sql-fs.ts @@ -81,6 +81,8 @@ interface SqlFsOptions { * together with `pathSnapshot` for snapshot-backed cold starts (Phase E). */ readonly redis?: Redis; + /** Pinned sandbox epoch captured when the script scope opens. */ + readonly sandboxVersion?: number; /** Redis path snapshot — tried before `loadAllPaths` when `redis` is also set. */ readonly pathSnapshot?: RedisPathSnapshot; /** @@ -158,6 +160,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { readonly #redis: Redis | undefined; readonly #pathSnapshot: RedisPathSnapshot | undefined; readonly #blobCache: RedisBlobCache | undefined; + readonly #sandboxVersion: number | undefined; #dirty = false; /** * Set when a recovery `reload()` fails after a COMMIT/abort, so the in-memory @@ -205,6 +208,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { this.#redis = opts.redis; this.#pathSnapshot = opts.pathSnapshot; this.#blobCache = opts.blobCache; + this.#sandboxVersion = opts.sandboxVersion; this.#pathCache = new Map(); this.#contentCache = new LRUCache({ maxSize: opts.contentCacheMaxBytes ?? DEFAULT_CONTENT_CACHE_MAX_BYTES, @@ -252,7 +256,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { } return runTrustedDbAsync(() => this.#dialect.transaction(async (tx) => { - await this.#dialect.setSandboxContextWithLock(tx, this.#sandboxId); + await this.#dialect.setSandboxContextWithLock(tx, this.#sandboxId, this.#sandboxVersion); return await fn(tx); }), ); @@ -293,7 +297,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { const scriptTxPromise = runTrustedDbAsync(() => this.#dialect.transaction(async (tx) => { - await this.#dialect.setSandboxContextWithLock(tx, this.#sandboxId); + await this.#dialect.setSandboxContextWithLock(tx, this.#sandboxId, this.#sandboxVersion); this.#scriptTx = tx; resolveTxReady(); await endPromise; @@ -481,7 +485,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { missReason = "no_key"; } else if (snap.version !== currentVersion) { missReason = "version_mismatch"; - } else { + } else { console.log( JSON.stringify({ event: "path_snapshot_hit", @@ -873,6 +877,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { bytes.length, sha256, bytes, + this.#sandboxVersion, ), ) : await this.#withTx(async (tx) => { @@ -883,7 +888,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { mode: 0o644, size: bytes.length, contentSha256: sha256, - }); + }); const oldInodeId = await this.#dialect.upsertDirent(tx, parentEntry.inodeId, name, id); if (oldInodeId !== null) { const newNlink = await this.#dialect.decrementNlink(tx, oldInodeId); @@ -963,7 +968,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { mode: 0o644, size: fullBytes.length, contentSha256: sha256, - }); + }); const oldInodeId = await this.#dialect.upsertDirent(tx, parentEntry.inodeId, name, id); if (oldInodeId !== null) { const newNlink = await this.#dialect.decrementNlink(tx, oldInodeId); @@ -1015,7 +1020,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { }); await this.#dialect.insertDirent(tx, parentEntry.inodeId, seg, id); return id; - }); + }); this.#cacheSet(next, { inodeId, kind: INODE_KIND.DIRECTORY, @@ -1024,7 +1029,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { mtime, contentSha256: null, symlinkTarget: null, - }); + }); created = true; } current = next; @@ -1039,7 +1044,7 @@ 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.#sandboxVersion), ) : await this.#withTx(async (tx) => { const id = await this.#dialect.createInode(tx, { @@ -1047,7 +1052,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { kind: INODE_KIND.DIRECTORY, mode: 0o755, size: 0, - }); + }); await this.#dialect.insertDirent(tx, parentEntry.inodeId, name, id); return id; }); @@ -1122,7 +1127,7 @@ 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.#sandboxVersion)); } else { await this.#withTx(async (tx) => { const removedInodeId = await this.#dialect.deleteDirent(tx, parentEntry!.inodeId, name); @@ -1308,7 +1313,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { size: entry.size, contentSha256: entry.contentSha256, symlinkTarget: entry.symlinkTarget, - }); + }); await this.#dialect.insertDirent(tx, parentInodeId, entryName, newId); newInodeIds.set(destPath, newId); } @@ -1413,6 +1418,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { srcName, destParentEntry.inodeId, destName, + this.#sandboxVersion, ), ); } else { diff --git a/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts new file mode 100644 index 0000000..e090d8f --- /dev/null +++ b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts @@ -0,0 +1,84 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { SqlFs } from "../../sql-fs.js"; +import type { PathCacheEntry, SqlDialect } from "../../types.js"; + +const now = new Date("2026-01-01T00:00:00Z"); + +function dirEntry(path: string, inodeId: bigint): { path: string } & PathCacheEntry { + return { path, inodeId, kind: 2, mode: 0o755, size: 0, mtime: now, contentSha256: null, symlinkTarget: null }; +} + +function makeDialect() { + const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => fn({})); + const setSandboxContextWithLock = vi.fn(async () => undefined); + const writeFileComposite = vi.fn(async () => 11n); + const mkdirComposite = vi.fn(async () => 12n); + const rmComposite = vi.fn(async () => 13n); + const mvComposite = vi.fn(async () => undefined); + const dialect: SqlDialect = { + connect: vi.fn(), + disconnect: vi.fn(), + transaction, + setSandboxContext: vi.fn(async () => undefined), + setSandboxContextWithLock, + loadAllPaths: vi.fn(async () => [dirEntry("/", 1n), dirEntry("/home", 2n), dirEntry("/home/user", 3n)]), + createSandbox: vi.fn(), + deleteSandbox: vi.fn(), + sandboxExists: vi.fn(), + getSandboxMeta: vi.fn(), + updateSandboxMeta: vi.fn(), + createInode: vi.fn(), + getInode: vi.fn(), + updateInode: vi.fn(), + deleteInode: vi.fn(), + incrementNlink: vi.fn(), + decrementNlink: vi.fn(), + insertDirent: vi.fn(), + upsertDirent: vi.fn(), + deleteDirent: vi.fn(), + listDirents: vi.fn(), + moveDirent: vi.fn(), + upsertBlob: vi.fn(), + getBlob: vi.fn(), + getBlobNoTx: vi.fn(), + gcOrphanBlobs: vi.fn(), + getBlobsForSandbox: vi.fn(async () => []), + loadSubtreeInodes: vi.fn(), + bulkIngest: vi.fn(), + resolvePath: vi.fn(), + writeFileComposite, + mkdirComposite, + rmComposite, + mvComposite, + } as SqlDialect; + return { dialect, transaction, setSandboxContextWithLock, writeFileComposite, mkdirComposite, rmComposite, mvComposite }; +} + +describe("SqlFs script fencing", () => { + let fs: SqlFs; + let m: ReturnType; + + beforeEach(async () => { + m = makeDialect(); + fs = new SqlFs({ dialect: m.dialect, sandboxId: "s-fence", sandboxVersion: 41 }); + await fs.ready(); + m.transaction.mockClear(); + m.setSandboxContextWithLock.mockClear(); + }); + + it("threads the pinned epoch into the script transaction and composite writes", async () => { + fs.beginScriptScope(); + await fs.writeFile("/home/user/a.txt", "alpha"); + await fs.mkdir("/home/user/projects"); + await fs.rm("/home/user/a.txt"); + await fs.mv("/home/user/projects", "/home/user/renamed-projects"); + + expect(m.transaction).toHaveBeenCalledOnce(); + expect(m.setSandboxContextWithLock).toHaveBeenCalledWith(expect.anything(), "s-fence", 41); + expect(m.writeFileComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "a.txt", expect.anything(), expect.anything(), expect.anything(), expect.anything(), 41); + expect(m.mkdirComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "projects", expect.anything(), 41); + expect(m.rmComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "a.txt", 41); + expect(m.mvComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "projects", expect.anything(), expect.anything(), 41); + await fs.endScriptScope(); + }); +}); diff --git a/src/sql-fs/types.ts b/src/sql-fs/types.ts index f28ea91..6c7f32d 100644 --- a/src/sql-fs/types.ts +++ b/src/sql-fs/types.ts @@ -166,7 +166,7 @@ export interface SqlDialect { * Call this from every write path. Read-only paths should use * `setSandboxContext` to avoid blocking writers unnecessarily. */ - setSandboxContextWithLock(tx: Tx, sandboxId: string): Promise; + setSandboxContextWithLock(tx: Tx, sandboxId: string, sandboxVersion?: number): Promise; // ── Sandbox lifecycle ───────────────────────────────────────────────────────── @@ -275,9 +275,9 @@ 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, sandboxVersion?: number): Promise; - rmComposite?(tx: Tx, sandboxId: string, parentId: bigint, name: string): Promise; + rmComposite?(tx: Tx, sandboxId: string, parentId: bigint, name: string, sandboxVersion?: number): Promise; writeFileComposite?( tx: Tx, @@ -288,6 +288,7 @@ export interface SqlDialect { size: number, sha256: Uint8Array, data: Uint8Array, + sandboxVersion?: number, ): Promise; mvComposite?( @@ -297,6 +298,7 @@ export interface SqlDialect { oldName: string, newParentId: bigint, newName: string, + sandboxVersion?: number, ): Promise; // ── Blob storage ────────────────────────────────────────────────────────────── From c52ae3232195ba80deb85a951fc273d3549ac980 Mon Sep 17 00:00:00 2001 From: Trek Worker Date: Fri, 10 Jul 2026 23:37:16 +0000 Subject: [PATCH 3/5] Implement sandbox epoch fencing for script composite writes --- src/sql-fs/dialects/postgres.ts | 59 ++++++++++++--- .../tests/unit/postgres.advisory-lock.test.ts | 71 +++++++++++++++++++ 2 files changed, 119 insertions(+), 11 deletions(-) diff --git a/src/sql-fs/dialects/postgres.ts b/src/sql-fs/dialects/postgres.ts index 45821e7..085ac37 100644 --- a/src/sql-fs/dialects/postgres.ts +++ b/src/sql-fs/dialects/postgres.ts @@ -119,14 +119,22 @@ export class PostgresDialect implements SqlDialect { async mkdirComposite(tx: PgTx, sandboxId: string, parentId: bigint, name: string, mode: number, sandboxVersion?: number): Promise { await this.#acquireSandboxWriteFence(tx, sandboxId); await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); + const expectedVersion = sandboxVersion ?? null; const rows = await tx<{ id: string }[]>` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)) ), + fence AS ( + SELECT 1 + FROM sandboxes + WHERE id = ${sandboxId} + AND (${expectedVersion} IS NULL OR version = ${expectedVersion}) + AND (SELECT 1 FROM ctx) IS NOT NULL + ), new_inode AS ( INSERT INTO inodes (sandbox_id, kind, mode, size) - SELECT ${sandboxId}, 2, ${mode}, 0 FROM ctx + SELECT ${sandboxId}, 2, ${mode}, 0 FROM ctx, fence RETURNING id ) INSERT INTO dirents (parent_inode_id, name, inode_id, sandbox_id) @@ -135,22 +143,29 @@ export class PostgresDialect implements SqlDialect { RETURNING inode_id AS id `; const row = rows[0]; - if (!row) throw new Error("mkdirComposite: INSERT returned no rows"); + if (!row) throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); return BigInt(row.id); } async rmComposite(tx: PgTx, sandboxId: string, parentId: bigint, name: string, sandboxVersion?: number): Promise { await this.#acquireSandboxWriteFence(tx, sandboxId); - await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); + const expectedVersion = sandboxVersion ?? null; 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)) ), + fence AS ( + SELECT 1 + FROM sandboxes + WHERE id = ${sandboxId} + AND (${expectedVersion} IS NULL OR version = ${expectedVersion}) + AND (SELECT 1 FROM ctx) IS NOT NULL + ), 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 fence) IS NOT NULL RETURNING inode_id ), -- Delete and decrement are split into two mutually-exclusive CTEs @@ -172,7 +187,7 @@ export class PostgresDialect implements SqlDialect { SELECT inode_id AS removed_inode_id FROM removed_dirent `; const row = rows[0]; - if (!row) throw createEnoent(name); + if (!row) throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); return BigInt(row.removed_inode_id); } @@ -191,20 +206,28 @@ export class PostgresDialect implements SqlDialect { // this composite runs, so there is no `blob_insert` CTE here — the // script-tx must not hold the hot-blob `ON CONFLICT DO UPDATE` tuple lock. await this.#acquireSandboxWriteFence(tx, sandboxId); - await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); + const expectedVersion = sandboxVersion ?? null; 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)) ), + fence AS ( + SELECT 1 + FROM sandboxes + WHERE id = ${sandboxId} + AND (${expectedVersion} IS NULL OR version = ${expectedVersion}) + AND (SELECT 1 FROM ctx) IS NOT NULL + ), 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 ctx, fence RETURNING id ), old_dirent AS ( SELECT inode_id FROM dirents WHERE parent_inode_id = ${String(parentId)} AND name = ${name} + AND (SELECT 1 FROM fence) IS NOT NULL ), upserted AS ( INSERT INTO dirents (parent_inode_id, name, inode_id, sandbox_id) @@ -230,7 +253,7 @@ export class PostgresDialect implements SqlDialect { SELECT id AS new_inode_id FROM new_inode `; const row = rows[0]; - if (!row) throw new Error("writeFileComposite: INSERT returned no rows"); + if (!row) throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); if (this.#blobCache !== undefined) { void this.#blobCache.set(sha256, data); } @@ -255,16 +278,23 @@ export class PostgresDialect implements SqlDialect { // the source into it. Statement 1's effects are visible to statement 2, so // the rename can no longer collide. Atomicity is preserved by the tx. await this.#acquireSandboxWriteFence(tx, sandboxId); - await this.#assertSandboxVersion(tx, sandboxId, sandboxVersion); + const expectedVersion = sandboxVersion ?? null; await tx` WITH ctx AS ( SELECT set_config('app.sandbox_id', ${sandboxId}, true), pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)) ), + fence AS ( + SELECT 1 + FROM sandboxes + WHERE id = ${sandboxId} + AND (${expectedVersion} IS NULL OR version = ${expectedVersion}) + AND (SELECT 1 FROM ctx) IS NOT NULL + ), 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 fence) IS NOT NULL RETURNING inode_id ), -- Split delete/decrement by snapshot nlink so the overwritten @@ -282,12 +312,19 @@ export class PostgresDialect implements SqlDialect { AND nlink > 1 `; const rows = await tx<{ inode_id: string }[]>` + WITH fence AS ( + SELECT 1 + FROM sandboxes + WHERE id = ${sandboxId} + AND (${expectedVersion} IS NULL OR version = ${expectedVersion}) + ) UPDATE dirents SET parent_inode_id = ${String(newParentId)}, name = ${newName} WHERE parent_inode_id = ${String(oldParentId)} AND name = ${oldName} + AND (SELECT 1 FROM fence) IS NOT NULL RETURNING inode_id `; - if (rows.length === 0) throw createEnoent(oldName); + if (rows.length === 0) throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); } // ── Private helpers ─────────────────────────────────────────────────────────── diff --git a/src/sql-fs/dialects/tests/unit/postgres.advisory-lock.test.ts b/src/sql-fs/dialects/tests/unit/postgres.advisory-lock.test.ts index 6f6ef80..74bb747 100644 --- a/src/sql-fs/dialects/tests/unit/postgres.advisory-lock.test.ts +++ b/src/sql-fs/dialects/tests/unit/postgres.advisory-lock.test.ts @@ -26,6 +26,12 @@ function makeFakeTx(): { tx: postgres.TransactionSql; calls: RecordedCall[] } { const fn = (strings: TemplateStringsArray, ...values: unknown[]): Promise => { const sql = strings.join("?"); calls.push({ sql, values }); + if (sql.includes("SELECT id AS new_inode_id FROM new_inode")) return Promise.resolve([{ new_inode_id: "11" }]); + if (sql.includes("SELECT inode_id AS removed_inode_id FROM removed_dirent")) return Promise.resolve([{ removed_inode_id: "13" }]); + if (sql.includes("UPDATE dirents") && sql.includes("RETURNING inode_id")) return Promise.resolve([{ inode_id: "15" }]); + if (sql.includes("RETURNING inode_id AS id")) return Promise.resolve([{ id: "12" }]); + if (sql.includes("RETURNING id")) return Promise.resolve([{ id: "12" }]); + if (sql.includes("RETURNING inode_id")) return Promise.resolve([{ inode_id: "14" }]); return Promise.resolve([]); }; return { tx: fn as unknown as postgres.TransactionSql, calls }; @@ -120,6 +126,71 @@ describe("PostgresDialect.deleteSandbox — advisory lock", () => { }); }); +describe("PostgresDialect composite write fencing", () => { + it("threads the pinned sandbox epoch into writeFileComposite's advisory-locked SQL", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const { tx, calls } = makeFakeTx(); + + await dialect.writeFileComposite( + tx, + "sandbox-fence", + 1n, + "file.txt", + 0o644, + 4, + new Uint8Array([1, 2, 3, 4]), + new Uint8Array([1, 2, 3, 4]), + 17, + ); + + const sql = calls.map((call) => call.sql).join("\n"); + expect(sql).toContain("set_config('app.sandbox_id'"); + expect(sql).toContain("pg_advisory_xact_lock"); + expect(sql).toContain("version"); + expect(sql).toContain("sandbox-fence"); + expect(calls.some((call) => call.values.includes(17))).toBe(true); + }); + + it("threads the pinned sandbox epoch into mkdirComposite's advisory-locked SQL", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const { tx, calls } = makeFakeTx(); + + await dialect.mkdirComposite(tx, "sandbox-fence", 1n, "dir", 0o755, 17); + + const sql = calls.map((call) => call.sql).join("\n"); + expect(sql).toContain("set_config('app.sandbox_id'"); + expect(sql).toContain("pg_advisory_xact_lock"); + expect(sql).toContain("version"); + expect(calls.at(-1)?.values).toContain(17); + }); + + it("threads the pinned sandbox epoch into rmComposite's advisory-locked SQL", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const { tx, calls } = makeFakeTx(); + + await dialect.rmComposite(tx, "sandbox-fence", 1n, "dir", 17); + + const sql = calls.map((call) => call.sql).join("\n"); + expect(sql).toContain("set_config('app.sandbox_id'"); + expect(sql).toContain("pg_advisory_xact_lock"); + expect(sql).toContain("version"); + expect(calls.at(-1)?.values).toContain(17); + }); + + it("threads the pinned sandbox epoch into mvComposite's advisory-locked SQL", async () => { + const dialect = new PostgresDialect("postgres://stub"); + const { tx, calls } = makeFakeTx(); + + await dialect.mvComposite(tx, "sandbox-fence", 1n, "old", 2n, "new", 17); + + const sql = calls.map((call) => call.sql).join("\n"); + expect(sql).toContain("set_config('app.sandbox_id'"); + expect(sql).toContain("pg_advisory_xact_lock"); + expect(sql).toContain("version"); + expect(calls.at(-1)?.values).toContain(17); + }); +}); + describe("PostgresDialect metadata helpers — SQL error translation", () => { it("sandboxExists translates raw SQL errors", async () => { const dialect = new PostgresDialect("postgres://stub"); From d2cd84a973d4b93e0265c177d0b4333e41198e87 Mon Sep 17 00:00:00 2001 From: Trek Worker Date: Sat, 11 Jul 2026 00:10:59 +0000 Subject: [PATCH 4/5] Implement fenced-path regression coverage for zombie epoch expiry race --- src/sql-fs/sql-fs.ts | 1 + .../tests/unit/sql-fs.script-fencing.test.ts | 128 +++++++++++++++++- 2 files changed, 125 insertions(+), 4 deletions(-) diff --git a/src/sql-fs/sql-fs.ts b/src/sql-fs/sql-fs.ts index ed1cba3..2557699 100644 --- a/src/sql-fs/sql-fs.ts +++ b/src/sql-fs/sql-fs.ts @@ -958,6 +958,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { fullBytes.length, sha256, fullBytes, + this.#sandboxVersion, ), ) : await this.#withTx(async (tx) => { diff --git a/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts index e090d8f..9a0ab53 100644 --- a/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts +++ b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts @@ -8,8 +8,21 @@ function dirEntry(path: string, inodeId: bigint): { path: string } & PathCacheEn return { path, inodeId, kind: 2, mode: 0o755, size: 0, mtime: now, contentSha256: null, symlinkTarget: null }; } +function fileEntry(path: string, inodeId: bigint, size: number): { path: string } & PathCacheEntry { + return { + path, + inodeId, + kind: 1, + mode: 0o644, + size, + mtime: now, + contentSha256: new Uint8Array(32).fill(0xab), + symlinkTarget: null, + }; +} + function makeDialect() { - const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => fn({})); + const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => fn({ txId: 1 })); const setSandboxContextWithLock = vi.fn(async () => undefined); const writeFileComposite = vi.fn(async () => 11n); const mkdirComposite = vi.fn(async () => 12n); @@ -21,7 +34,69 @@ function makeDialect() { transaction, setSandboxContext: vi.fn(async () => undefined), setSandboxContextWithLock, - loadAllPaths: vi.fn(async () => [dirEntry("/", 1n), dirEntry("/home", 2n), dirEntry("/home/user", 3n)]), + loadAllPaths: vi.fn(async () => [dirEntry("/", 1n), dirEntry("/home", 2n), dirEntry("/home/user", 3n), fileEntry("/home/user/file.txt", 4n, 4)]), + createSandbox: vi.fn(), + deleteSandbox: vi.fn(), + sandboxExists: vi.fn(), + getSandboxMeta: vi.fn(), + updateSandboxMeta: vi.fn(), + createInode: vi.fn(), + getInode: vi.fn(), + updateInode: vi.fn(), + deleteInode: vi.fn(), + incrementNlink: vi.fn(), + decrementNlink: vi.fn(), + insertDirent: vi.fn(), + upsertDirent: vi.fn(), + deleteDirent: vi.fn(), + listDirents: vi.fn(), + moveDirent: vi.fn(), + upsertBlob: vi.fn(), + getBlob: vi.fn(async () => new TextEncoder().encode("seed")), + getBlobNoTx: vi.fn(), + gcOrphanBlobs: vi.fn(), + getBlobsForSandbox: vi.fn(async () => []), + loadSubtreeInodes: vi.fn(), + bulkIngest: vi.fn(), + resolvePath: vi.fn(), + writeFileComposite, + mkdirComposite, + rmComposite, + mvComposite, + } as SqlDialect; + return { dialect, transaction, setSandboxContextWithLock, writeFileComposite, mkdirComposite, rmComposite, mvComposite }; +} + +type FencedState = { + currentVersion: number; + files: Map; + txs: unknown[]; +}; + +function makeFencedDialect(state: FencedState) { + const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => { + const tx = { txId: state.txs.length + 1 }; + state.txs.push(tx); + return fn(tx); + }); + const setSandboxContextWithLock = vi.fn(async () => undefined); + const writeFileComposite = vi.fn(async (_tx: unknown, _sandboxId: string, _parentId: bigint, name: string, _mode: number, _size: number, _sha256: Uint8Array, data: Uint8Array, sandboxVersion?: number) => { + if (sandboxVersion !== state.currentVersion) { + throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); + } + state.files.set(`/home/user/${name}`, new TextDecoder().decode(data)); + return 99n; + }); + const mkdirComposite = vi.fn(async () => 12n); + const rmComposite = vi.fn(async () => 13n); + const mvComposite = vi.fn(async () => undefined); + const dialect: SqlDialect = { + connect: vi.fn(), + disconnect: vi.fn(), + transaction, + setSandboxContext: vi.fn(async () => undefined), + setSandboxContextWithLock, + loadAllPaths: vi.fn(async () => [dirEntry("/", 1n), dirEntry("/home", 2n), dirEntry("/home/user", 3n), fileEntry("/home/user/file.txt", 4n, state.files.get("/home/user/file.txt")?.length ?? 0)]), createSandbox: vi.fn(), deleteSandbox: vi.fn(), sandboxExists: vi.fn(), @@ -39,7 +114,7 @@ function makeDialect() { listDirents: vi.fn(), moveDirent: vi.fn(), upsertBlob: vi.fn(), - getBlob: vi.fn(), + getBlob: vi.fn(async () => new TextEncoder().encode(state.files.get("/home/user/file.txt") ?? "seed")), getBlobNoTx: vi.fn(), gcOrphanBlobs: vi.fn(), getBlobsForSandbox: vi.fn(async () => []), @@ -69,16 +144,61 @@ describe("SqlFs script fencing", () => { it("threads the pinned epoch into the script transaction and composite writes", async () => { fs.beginScriptScope(); await fs.writeFile("/home/user/a.txt", "alpha"); + await fs.appendFile("/home/user/a.txt", "-beta"); await fs.mkdir("/home/user/projects"); await fs.rm("/home/user/a.txt"); await fs.mv("/home/user/projects", "/home/user/renamed-projects"); expect(m.transaction).toHaveBeenCalledOnce(); expect(m.setSandboxContextWithLock).toHaveBeenCalledWith(expect.anything(), "s-fence", 41); - expect(m.writeFileComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "a.txt", expect.anything(), expect.anything(), expect.anything(), expect.anything(), 41); + expect(m.writeFileComposite).toHaveBeenNthCalledWith(1, expect.anything(), "s-fence", expect.anything(), "a.txt", expect.anything(), expect.anything(), expect.anything(), expect.anything(), 41); + expect(m.writeFileComposite).toHaveBeenNthCalledWith(2, expect.anything(), "s-fence", expect.anything(), "a.txt", expect.anything(), expect.anything(), expect.anything(), expect.anything(), 41); expect(m.mkdirComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "projects", expect.anything(), 41); expect(m.rmComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "a.txt", 41); expect(m.mvComposite).toHaveBeenCalledWith(expect.anything(), "s-fence", expect.anything(), "projects", expect.anything(), expect.anything(), 41); await fs.endScriptScope(); }); + + it("rejects a stale fenced write after the lease is superseded, while a fresh append survives", async () => { + const state: FencedState = { currentVersion: 41, files: new Map([["/home/user/file.txt", "seed"]]), txs: [] }; + const { dialect } = makeFencedDialect(state); + const stale = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 41 }); + const fresh = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 42 }); + await Promise.all([stale.ready(), fresh.ready()]); + + stale.beginScriptScope(); + fresh.beginScriptScope(); + + state.currentVersion = 42; + + await expect(stale.writeFile("/home/user/file.txt", "from-a")).rejects.toMatchObject({ code: "ECOHERENCE" }); + await fresh.appendFile("/home/user/file.txt", "-from-b"); + await Promise.all([stale.abortScriptScope(), fresh.endScriptScope()]); + + expect(state.files.get("/home/user/file.txt")).toBe("seed-from-b"); + expect(dialect.writeFileComposite).toHaveBeenCalledTimes(2); + expect(dialect.writeFileComposite.mock.calls[0]![8]).toBe(41); + expect(dialect.writeFileComposite.mock.calls[1]![8]).toBe(42); + }); + + it("keeps the fenced path compatible with transaction pooling by reusing the opened script tx", async () => { + const state: FencedState = { currentVersion: 41, files: new Map([["/home/user/file.txt", "seed"]]), txs: [] }; + const { dialect } = makeFencedDialect(state); + const pooled = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 41 }); + await pooled.ready(); + + pooled.beginScriptScope(); + await pooled.writeFile("/home/user/a.txt", "alpha"); + await pooled.appendFile("/home/user/file.txt", "-beta"); + await pooled.mkdir("/home/user/projects"); + await pooled.endScriptScope(); + + expect(state.txs).toHaveLength(1); + expect(dialect.writeFileComposite).toHaveBeenCalledTimes(2); + expect(dialect.mkdirComposite).toHaveBeenCalledTimes(1); + const firstTx = dialect.writeFileComposite.mock.calls[0]![0]; + const secondTx = dialect.writeFileComposite.mock.calls[1]![0]; + expect(firstTx).toBe(secondTx); + expect(dialect.mkdirComposite.mock.calls[0]![0]).toBe(firstTx); + }); }); From cab17ec5eb6bf5ff2b03a50d1a009aab4877c0d6 Mon Sep 17 00:00:00 2001 From: Trek Worker Date: Sat, 11 Jul 2026 00:17:50 +0000 Subject: [PATCH 5/5] Add fenced-path zombie regression coverage --- .../tests/unit/sql-fs.script-fencing.test.ts | 67 +++++++++++++------ 1 file changed, 47 insertions(+), 20 deletions(-) diff --git a/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts index 9a0ab53..b27e60c 100644 --- a/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts +++ b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts @@ -67,29 +67,51 @@ function makeDialect() { return { dialect, transaction, setSandboxContextWithLock, writeFileComposite, mkdirComposite, rmComposite, mvComposite }; } +type FencedTx = { + txId: number; + sandboxVersion?: number; + staged: Array<() => void>; +}; + type FencedState = { currentVersion: number; files: Map; - txs: unknown[]; + txs: FencedTx[]; }; function makeFencedDialect(state: FencedState) { - const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => { - const tx = { txId: state.txs.length + 1 }; + const transaction = vi.fn(async (fn: (tx: FencedTx) => Promise) => { + const tx: FencedTx = { txId: state.txs.length + 1, staged: [] }; state.txs.push(tx); - return fn(tx); + const result = await fn(tx); + if (tx.sandboxVersion !== state.currentVersion) { + throw Object.assign(new Error("ECOHERENCE: sandbox version changed; commit failed"), { code: "ECOHERENCE" }); + } + for (const apply of tx.staged) apply(); + return result; }); - const setSandboxContextWithLock = vi.fn(async () => undefined); - const writeFileComposite = vi.fn(async (_tx: unknown, _sandboxId: string, _parentId: bigint, name: string, _mode: number, _size: number, _sha256: Uint8Array, data: Uint8Array, sandboxVersion?: number) => { - if (sandboxVersion !== state.currentVersion) { + const setSandboxContextWithLock = vi.fn(async (tx: FencedTx, _sandboxId: string, sandboxVersion?: number) => { + tx.sandboxVersion = sandboxVersion; + }); + const writeFileComposite = vi.fn(async (tx: FencedTx, _sandboxId: string, _parentId: bigint, name: string, _mode: number, _size: number, _sha256: Uint8Array, data: Uint8Array, sandboxVersion?: number) => { + if (sandboxVersion !== tx.sandboxVersion) { throw Object.assign(new Error("ECOHERENCE: sandbox version changed; write suppressed"), { code: "ECOHERENCE" }); } - state.files.set(`/home/user/${name}`, new TextDecoder().decode(data)); + tx.staged.push(() => state.files.set(`/home/user/${name}`, new TextDecoder().decode(data))); return 99n; }); - const mkdirComposite = vi.fn(async () => 12n); - const rmComposite = vi.fn(async () => 13n); - const mvComposite = vi.fn(async () => undefined); + const mkdirComposite = vi.fn(async (tx: FencedTx) => { + tx.staged.push(() => undefined); + return 12n; + }); + const rmComposite = vi.fn(async (tx: FencedTx) => { + tx.staged.push(() => undefined); + return 13n; + }); + const mvComposite = vi.fn(async (tx: FencedTx) => { + tx.staged.push(() => undefined); + return undefined; + }); const dialect: SqlDialect = { connect: vi.fn(), disconnect: vi.fn(), @@ -159,21 +181,26 @@ describe("SqlFs script fencing", () => { await fs.endScriptScope(); }); - it("rejects a stale fenced write after the lease is superseded, while a fresh append survives", async () => { + it("fails A's commit after lease expiry before first write, while B's append survives", async () => { const state: FencedState = { currentVersion: 41, files: new Map([["/home/user/file.txt", "seed"]]), txs: [] }; const { dialect } = makeFencedDialect(state); - const stale = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 41 }); - const fresh = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 42 }); - await Promise.all([stale.ready(), fresh.ready()]); + const writerA = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 41 }); + const writerB = new SqlFs({ dialect, sandboxId: "s-fence", sandboxVersion: 42 }); + await Promise.all([writerA.ready(), writerB.ready()]); - stale.beginScriptScope(); - fresh.beginScriptScope(); + writerA.beginScriptScope(); + writerB.beginScriptScope(); + // Force the lease to expire before writer A's first write. A should be + // allowed to stage the write, but its script transaction must fail on + // epoch mismatch when the script scope commits. state.currentVersion = 42; - await expect(stale.writeFile("/home/user/file.txt", "from-a")).rejects.toMatchObject({ code: "ECOHERENCE" }); - await fresh.appendFile("/home/user/file.txt", "-from-b"); - await Promise.all([stale.abortScriptScope(), fresh.endScriptScope()]); + await writerA.writeFile("/home/user/file.txt", "from-a"); + await writerB.appendFile("/home/user/file.txt", "-from-b"); + + await expect(writerA.endScriptScope()).rejects.toMatchObject({ code: "ECOHERENCE" }); + await expect(writerB.endScriptScope()).resolves.toBeUndefined(); expect(state.files.get("/home/user/file.txt")).toBe("seed-from-b"); expect(dialect.writeFileComposite).toHaveBeenCalledTimes(2);