diff --git a/src/api/session-manager.ts b/src/api/session-manager.ts index e517282..33d7d24 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,20 @@ 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 meta = this.getSandboxMetaFn !== undefined ? await this.getSandboxMetaFn(tenantId, sandboxId) : null; + const sandboxVersion = meta?.version ?? 0; + const fs = await createPostgresSandboxFs( { connectionString: backend.connectionString, tenantId, @@ -521,7 +530,14 @@ export class SessionManager { }, sandboxId, owner, + sandboxVersion, ); + return { + fs, + resolvedOwner: owner, + createdAt: new Date().toISOString(), + version: sandboxVersion, + }; } private estimatePathCacheBytes(fs: IFileSystem): number { @@ -567,7 +583,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 +675,7 @@ export class SessionManager { pathCacheBytes, overBudget: pathCacheBytes > this.pathCacheMaxBytes, lastSeenVersion: initialVersion, + sandboxVersion, publishPending: false, cwd: bash.getCwd(), }; @@ -1179,9 +1196,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 +1212,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 +1245,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..085ac37 100644 --- a/src/sql-fs/dialects/postgres.ts +++ b/src/sql-fs/dialects/postgres.ts @@ -95,21 +95,46 @@ 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 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) @@ -118,20 +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): Promise { + async rmComposite(tx: PgTx, sandboxId: string, parentId: bigint, name: string, sandboxVersion?: number): Promise { + await this.#acquireSandboxWriteFence(tx, sandboxId); + 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 @@ -153,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); } @@ -166,23 +200,34 @@ 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); + 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) @@ -208,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); } @@ -222,6 +267,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,15 +277,24 @@ 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); + 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 @@ -257,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 ─────────────────────────────────────────────────────────── @@ -337,10 +399,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 +413,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) { @@ -582,6 +646,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/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"); 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/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/sql-fs.ts b/src/sql-fs/sql-fs.ts index 05f3955..2557699 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); @@ -953,6 +958,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { fullBytes.length, sha256, fullBytes, + this.#sandboxVersion, ), ) : await this.#withTx(async (tx) => { @@ -963,7 +969,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 +1021,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 +1030,7 @@ export class SqlFs implements ICoherentFs, IReadOnlyScopeFs { mtime, contentSha256: null, symlinkTarget: null, - }); + }); created = true; } current = next; @@ -1039,7 +1045,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 +1053,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 +1128,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 +1314,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 +1419,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..b27e60c --- /dev/null +++ b/src/sql-fs/tests/unit/sql-fs.script-fencing.test.ts @@ -0,0 +1,231 @@ +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 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({ txId: 1 })); + 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), 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 FencedTx = { + txId: number; + sandboxVersion?: number; + staged: Array<() => void>; +}; + +type FencedState = { + currentVersion: number; + files: Map; + txs: FencedTx[]; +}; + +function makeFencedDialect(state: FencedState) { + const transaction = vi.fn(async (fn: (tx: FencedTx) => Promise) => { + const tx: FencedTx = { txId: state.txs.length + 1, staged: [] }; + state.txs.push(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 (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" }); + } + tx.staged.push(() => state.files.set(`/home/user/${name}`, new TextDecoder().decode(data))); + return 99n; + }); + 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(), + 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(), + 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(state.files.get("/home/user/file.txt") ?? "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 }; +} + +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.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).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("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 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()]); + + 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 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); + 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); + }); +}); diff --git a/src/sql-fs/types.ts b/src/sql-fs/types.ts index fe5ed6f..6c7f32d 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; } @@ -164,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 ───────────────────────────────────────────────────────── @@ -273,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, @@ -286,6 +288,7 @@ export interface SqlDialect { size: number, sha256: Uint8Array, data: Uint8Array, + sandboxVersion?: number, ): Promise; mvComposite?( @@ -295,6 +298,7 @@ export interface SqlDialect { oldName: string, newParentId: bigint, newName: string, + sandboxVersion?: number, ): Promise; // ── Blob storage ──────────────────────────────────────────────────────────────