From 163c48d43593e85ee5d076a1fe78d77701b2c4ea Mon Sep 17 00:00:00 2001 From: Nader Helmy Date: Sun, 26 Jul 2026 15:58:51 -0500 Subject: [PATCH] fix: coordinate concurrent edge writes --- CHANGELOG.md | 8 ++ docs/troubleshooting.md | 17 ++++ src/sqlite-database-contract.ts | 4 + src/sqlite-edge-store.ts | 139 ++++++++++++++++++++++++-------- test/offline-sync.test.ts | 4 +- test/sqlite-write-gate.test.ts | 77 ++++++++++++++++++ 6 files changed, 215 insertions(+), 34 deletions(-) create mode 100644 test/sqlite-write-gate.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 79fb1fa..43f8a70 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ All notable changes to agent-bridge are documented in this file. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and the project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## Unreleased + +### Fixed + +- Serialize concurrent gateway edge mutations through a token-checked lease stored in + SQLite. The coordinator recovers an expired crashed-writer lease without releasing a + newer owner, and keeps gateway calls outside the local write boundary. + ## [0.6.2] - 2026-07-21 ### Fixed diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 877d371..bcd5946 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -81,6 +81,23 @@ Queued sends retain their idempotency keys. Long-lived MCP clients retry them, a `agent-bridge sync` triggers the same bounded replay manually. Claims, lease changes, delivery settlement, presence, and read-receipt writes still require the gateway. +## Local edge database is locked + +Gateway clients share an owner-private local edge database for their outbox and inbox +cache. Several active MCP processes may use that database at once. Agent Bridge stores +one token-checked, short-lived write lease in that database. The lease serializes local +mutations, survives a crashed process, and is never held while a client calls the +gateway. + +If an older installed executable reports `fatal local edge failure: database is locked`, +upgrade the package and restart only the affected MCP host. Do not delete +`edge.sqlite3` or its `-wal` or `-shm` sidecars while clients are running. Those files +carry durable queued work and may be open in another session. + +Use `lsof` to identify the host before restarting it. Multiple Codex task processes are +normal. A stale child is one whose parent host has exited, not simply one that has been +running for a long time. + ## Legacy Supabase provider was removed Agent Bridge 0.6.0 rejects `legacy`, `supabase`, `legacy-supabase`, and key-only diff --git a/src/sqlite-database-contract.ts b/src/sqlite-database-contract.ts index f29958c..f2f0417 100644 --- a/src/sqlite-database-contract.ts +++ b/src/sqlite-database-contract.ts @@ -14,6 +14,8 @@ export const LOCAL_SQLITE_SCHEMA_CONTRACTS = Object.freeze([ Object.freeze({ id: "current-upgraded-delivery-events", sha256: "35bdab0cbdf4ce619ba1c7d36032379ae4d9692bd7509e49bb90b409bbbcf632" }), ] as const); export const EDGE_SQLITE_SCHEMA_CONTRACTS = Object.freeze([ + Object.freeze({ id: "current-upgraded-write-coordinator", sha256: "973d010ff81a726d6520dc482a4450697ef2295f58f42a45dee30b6174b7b9f9" }), + Object.freeze({ id: "current-upgraded-project-column-migration-gate-write-coordinator", sha256: "06c159fe10fade817b211197dea14f178e75f54c2f55b69d63f8fafb629e731c" }), Object.freeze({ id: "current-created-schema", sha256: "27f22b2f4024585c87e8d6f76f8999a8df92bb1f585d46594a7a1e852fd53c4c" }), Object.freeze({ id: "current-upgraded-project-column", sha256: "171ae11f520b4963f517d3102de64cb8a5f45c080078ff02e20aeb17891b585c" }), Object.freeze({ id: "current-upgraded-project-column-migration-gate", sha256: "790b4bfed373ff776ba6700065154f7a3cbbecd94f37ea09a654768ac2a8455c" }), @@ -89,6 +91,7 @@ const edgeColumns: Record = { edge_migration_gates: ["scope_key", "state", "operation_id", "lease_token", "lease_expires_at", "updated_at"], edge_outbox: ["position", "scope_key", "message_id", "idempotency_key", "payload_hash", "draft_json", "state", "attempts", "available_at", "lease_token", "lease_expires_at", "last_error", "blocked_at", "created_at"], edge_inbox: ["scope_key", "message_id", "remote_sequence", "sequence_key", "workspace", "project", "source", "type", "thread_id", "created_at", "expires_at", "message_json"], + edge_write_gates: ["gate_key", "lease_token", "lease_expires_at"], }; const edgeObjects = new Map([ @@ -97,6 +100,7 @@ const edgeObjects = new Map([ ["edge_migration_gates", ["table", "edge_migration_gates"]], ["edge_outbox", ["table", "edge_outbox"]], ["edge_inbox", ["table", "edge_inbox"]], + ["edge_write_gates", ["table", "edge_write_gates"]], ["edge_inbox_created", ["index", "edge_inbox"]], ["edge_inbox_cursor", ["index", "edge_inbox"]], ["edge_inbox_project", ["index", "edge_inbox"]], diff --git a/src/sqlite-edge-store.ts b/src/sqlite-edge-store.ts index 3e1757e..ec201f6 100644 --- a/src/sqlite-edge-store.ts +++ b/src/sqlite-edge-store.ts @@ -115,6 +115,11 @@ CREATE INDEX IF NOT EXISTS edge_inbox_thread ON edge_inbox(scope_key, thread_id, sequence_key); CREATE INDEX IF NOT EXISTS edge_inbox_created ON edge_inbox(scope_key, created_at, sequence_key); +CREATE TABLE IF NOT EXISTS edge_write_gates ( + gate_key TEXT PRIMARY KEY, + lease_token TEXT, + lease_expires_at TEXT +); `; type Row = Record; @@ -153,14 +158,17 @@ export function inspectEdgeScopeReadOnly(path: string, scope: EdgeScope, now = n } const db = openReadOnlyDatabase(path); try { - const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('edge_scopes','edge_migration_gates','edge_outbox','edge_inbox') ORDER BY name") + const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('edge_scopes','edge_migration_gates','edge_outbox','edge_inbox','edge_write_gates') ORDER BY name") .all() as Row[]; if (tables.length === 0) { const userTables = db.prepare("SELECT count(*) AS count FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").get() as Row; if (Number(userTables.count) !== 0) throw new EdgeConflictError("edge database is not an Agent Bridge edge store"); return { exists: true, gate: { scopeKey: edgeScopeKey(scope), state: "active", operationId: undefined, leaseExpiresAt: undefined, updatedAt: new Date(0).toISOString() }, pending: 0, due: 0, scheduled: 0, leased: 0, blocked: 0, cached: 0 }; } - if (tables.length !== 4) throw new EdgeConflictError("edge database schema is incomplete"); + const names = new Set(tables.map((table) => String(table.name))); + for (const name of ["edge_scopes", "edge_migration_gates", "edge_outbox", "edge_inbox"]) { + if (!names.has(name)) throw new EdgeConflictError("edge database schema is incomplete"); + } const key = edgeScopeKey(scope); const scopeRow = db.prepare("SELECT 1 AS present FROM edge_scopes WHERE scope_key=?").get(key) as Row | undefined; if (!scopeRow) { @@ -346,7 +354,14 @@ export class SQLiteEdgeStore { private initialized = false; private initialization?: Promise; - constructor(private readonly path: string, private readonly scope: EdgeScope, private readonly busyTimeoutMs = 2_000) { + constructor( + private readonly path: string, + private readonly scope: EdgeScope, + private readonly busyTimeoutMs = 2_000, + private readonly writeGateLeaseMs = 30_000, + // Windows first-open ACL work is materially slower with many concurrent clients. + private readonly writeGateWaitMs = process.platform === "win32" ? 60_000 : SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS, + ) { const selected = preparePrivateSqliteLocation(path, true, "defer"); this.databasePath = selected; this.preexistingFiles = new Set(selected === ":memory:" ? [] : [selected, `${selected}-wal`, `${selected}-shm`].filter(existsSync)); @@ -363,6 +378,42 @@ export class SQLiteEdgeStore { this.key = edgeScopeKey(scope); } + private async write(work: () => T | Promise): Promise { + if (this.databasePath === ":memory:") return work(); + const token = randomUUID(); + const deadline = Date.now() + Math.max(1, Math.trunc(this.writeGateWaitMs)); + let pause = 2; + while (true) { + const now = new Date(); + const claimed = await retrySqliteBusy(() => { + this.db.exec("BEGIN IMMEDIATE"); + try { + this.db.prepare("INSERT INTO edge_write_gates (gate_key,lease_token,lease_expires_at) VALUES ('edge',NULL,NULL) ON CONFLICT(gate_key) DO NOTHING").run(); + const update = this.db.prepare(`UPDATE edge_write_gates SET lease_token=?,lease_expires_at=? + WHERE gate_key='edge' AND (lease_token IS NULL OR lease_expires_at<=?)`).run( + token, new Date(now.getTime() + Math.max(1, Math.trunc(this.writeGateLeaseMs))).toISOString(), now.toISOString(), + ); + this.db.exec("COMMIT"); + return update.changes === 1; + } catch (error) { this.db.exec("ROLLBACK"); throw error; } + }, Math.max(1, deadline - Date.now())); + if (claimed) break; + if (Date.now() >= deadline) throw new EdgeConflictError("edge write coordinator remained busy"); + await new Promise((resolve) => setTimeout(resolve, pause)); + pause = Math.min(Math.ceil(pause * 1.7), 25); + } + try { return await work(); } + finally { + await retrySqliteBusy(() => { + this.db.exec("BEGIN IMMEDIATE"); + try { + this.db.prepare("UPDATE edge_write_gates SET lease_token=NULL,lease_expires_at=NULL WHERE gate_key='edge' AND lease_token=?").run(token); + this.db.exec("COMMIT"); + } catch (error) { this.db.exec("ROLLBACK"); throw error; } + }, Math.max(1, Math.trunc(this.writeGateWaitMs))); + } + } + private restrictFiles(includeSidecars = true): void { if (this.databasePath === ":memory:") return; const paths = includeSidecars @@ -452,8 +503,9 @@ export class SQLiteEdgeStore { assertEdgeUpgradeCandidate(this.db); const initializationTimeoutMs = Math.max(this.busyTimeoutMs, SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS); await retrySqliteBusy(() => this.db.exec(schema), initializationTimeoutMs); - await retrySqliteBusy(() => this.db.exec("BEGIN IMMEDIATE"), initializationTimeoutMs); - try { + await this.write(async () => { + await retrySqliteBusy(() => this.db.exec("BEGIN IMMEDIATE"), initializationTimeoutMs); + try { const scopeColumns = this.db.prepare("PRAGMA table_info(edge_scopes)").all() as Row[]; if (!scopeColumns.some((column) => column.name === "cache_contract")) { this.db.exec("ALTER TABLE edge_scopes ADD COLUMN cache_contract INTEGER NOT NULL DEFAULT 0"); @@ -473,29 +525,30 @@ export class SQLiteEdgeStore { this.db.exec("ALTER TABLE edge_outbox ADD COLUMN blocked_at TEXT"); } installEdgeMarkers(this.db); - this.db.exec("COMMIT"); - } catch (error) { - this.db.exec("ROLLBACK"); - throw error; - } - this.restrictFiles(); - const endpointHash = createHash("sha256").update(this.scope.endpoint.replace(/\/$/, "")).digest("hex"); - this.db.prepare(`INSERT INTO edge_scopes (scope_key, endpoint_hash, workspace, agent) + this.db.exec("COMMIT"); + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + this.restrictFiles(); + const endpointHash = createHash("sha256").update(this.scope.endpoint.replace(/\/$/, "")).digest("hex"); + this.db.prepare(`INSERT INTO edge_scopes (scope_key, endpoint_hash, workspace, agent) VALUES (?, ?, ?, ?) ON CONFLICT(scope_key) DO NOTHING`) - .run(this.key, endpointHash, this.scope.principal.workspace, this.scope.principal.agent); - this.db.prepare(`INSERT INTO edge_migration_gates + .run(this.key, endpointHash, this.scope.principal.workspace, this.scope.principal.agent); + this.db.prepare(`INSERT INTO edge_migration_gates (scope_key,state,operation_id,lease_token,lease_expires_at,updated_at) SELECT scope_key,'active',NULL,NULL,NULL,? FROM edge_scopes WHERE scope_key=? ON CONFLICT(scope_key) DO NOTHING`) - .run(new Date().toISOString(), this.key); - const contract = this.db.prepare("SELECT cache_contract FROM edge_scopes WHERE scope_key=?").get(this.key) as Row; - if (Number(contract.cache_contract) < 1) { - this.db.exec("BEGIN IMMEDIATE"); - try { - this.db.prepare("UPDATE edge_scopes SET pull_cursor=NULL, cache_contract=1 WHERE scope_key=?").run(this.key); - this.db.exec("COMMIT"); - } catch (error) { this.db.exec("ROLLBACK"); throw error; } - } + .run(new Date().toISOString(), this.key); + const contract = this.db.prepare("SELECT cache_contract FROM edge_scopes WHERE scope_key=?").get(this.key) as Row; + if (Number(contract.cache_contract) < 1) { + this.db.exec("BEGIN IMMEDIATE"); + try { + this.db.prepare("UPDATE edge_scopes SET pull_cursor=NULL, cache_contract=1 WHERE scope_key=?").run(this.key); + this.db.exec("COMMIT"); + } catch (error) { this.db.exec("ROLLBACK"); throw error; } + } + }); } private async ready(): Promise { @@ -510,6 +563,7 @@ export class SQLiteEdgeStore { /** Refuse new normal-client writes before any remote attempt begins. */ async assertScopeActive(): Promise { await this.ready(); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertScopeActiveInTransaction(); @@ -518,12 +572,14 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } /** Refuse remote publication unless this worker still owns the durable drain lease. */ async assertDrainLease(lease: EdgeDrainLease, now = new Date()): Promise { await this.ready(); if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError(); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertDrainLeaseInTransaction(lease, now); @@ -532,6 +588,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } private async claimDrainLease( @@ -540,6 +597,7 @@ export class SQLiteEdgeStore { await this.ready(); const normalizedOperationId = safeOperationId(operationId); const duration = Math.max(1, Math.trunc(leaseMs)); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { const gate = this.gateInTransaction(); @@ -565,6 +623,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } /** Start a new durable drain only from an active edge scope. */ @@ -588,6 +647,7 @@ export class SQLiteEdgeStore { async assertDrainComplete(lease: EdgeDrainLease, now = new Date()): Promise { await this.ready(); if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError(); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertDrainLeaseInTransaction(lease, now); @@ -610,12 +670,14 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async renewDrainLease(lease: EdgeDrainLease, now = new Date(), leaseMs = 30_000): Promise { await this.ready(); if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError(); const duration = Math.max(1, Math.trunc(leaseMs)); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertDrainLeaseInTransaction(lease, now); @@ -632,11 +694,13 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async retireScope(lease: EdgeDrainLease, now = new Date()): Promise { await this.ready(); if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError(); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertDrainLeaseInTransaction(lease, now); @@ -654,6 +718,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async enqueue(input: PendingMessage, now = new Date()): Promise { @@ -662,6 +727,7 @@ export class SQLiteEdgeStore { const payloadHash = edgeMessageFingerprint(draft); const serialized = canonical(draft as unknown as JsonValue); const nowText = now.toISOString(); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertScopeActiveInTransaction(); @@ -685,11 +751,13 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async claimNext(now = new Date(), leaseMs = 30_000, migrationLease?: EdgeDrainLease): Promise { await this.ready(); const nowText = now.toISOString(); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.assertClaimAuthorityInTransaction(now, migrationLease); @@ -724,23 +792,24 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async retry(record: EdgeOutboxRecord, error: string, availableAt: Date): Promise { await this.ready(); - this.db.prepare(`UPDATE edge_outbox SET attempts=attempts+1, available_at=?, + await this.write(() => this.db.prepare(`UPDATE edge_outbox SET attempts=attempts+1, available_at=?, lease_token=NULL, lease_expires_at=NULL, last_error=? WHERE scope_key=? AND position=? AND lease_token=?`) - .run(availableAt.toISOString(), error.slice(0, 256), this.key, record.position, record.leaseToken); + .run(availableAt.toISOString(), error.slice(0, 256), this.key, record.position, record.leaseToken)); await this.noteError(error); } async block(record: EdgeOutboxRecord, error: string, now = new Date()): Promise { await this.ready(); - this.db.prepare(`UPDATE edge_outbox SET state='blocked', attempts=attempts+1, + await this.write(() => this.db.prepare(`UPDATE edge_outbox SET state='blocked', attempts=attempts+1, lease_token=NULL, lease_expires_at=NULL, last_error=?, blocked_at=? WHERE scope_key=? AND position=? AND lease_token=?`) - .run(error.slice(0, 256), now.toISOString(), this.key, record.position, record.leaseToken); + .run(error.slice(0, 256), now.toISOString(), this.key, record.position, record.leaseToken)); await this.noteError(error); } @@ -750,6 +819,7 @@ export class SQLiteEdgeStore { now = new Date(), ): Promise { await this.ready(); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.cacheOne(message); @@ -766,6 +836,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } private cacheOne(message: BridgeMessage): void { @@ -797,6 +868,7 @@ export class SQLiteEdgeStore { async cachePage(messages: BridgeMessage[], cursor: string | undefined, now = new Date()): Promise { await this.ready(); const scope = cursorScope(this.scope.principal, { mailbox: "all", includeExpired: true }); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { for (const message of messages) this.cacheOne(message); @@ -817,10 +889,12 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async cacheLatest(messages: BridgeMessage[], now = new Date()): Promise { await this.ready(); + await this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { for (const message of messages) this.cacheOne(message); @@ -833,6 +907,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async list(query: MessageQuery = {}): Promise { @@ -927,14 +1002,14 @@ export class SQLiteEdgeStore { async noteError(error: string): Promise { await this.ready(); - this.db.prepare("UPDATE edge_scopes SET last_error=? WHERE scope_key=?") - .run(error.slice(0, 256), this.key); + await this.write(() => this.db.prepare("UPDATE edge_scopes SET last_error=? WHERE scope_key=?") + .run(error.slice(0, 256), this.key)); } async noteAttempt(now = new Date()): Promise { await this.ready(); - this.db.prepare("UPDATE edge_scopes SET last_attempt_at=? WHERE scope_key=?") - .run(now.toISOString(), this.key); + await this.write(() => this.db.prepare("UPDATE edge_scopes SET last_attempt_at=? WHERE scope_key=?") + .run(now.toISOString(), this.key)); } async stats(now = new Date()): Promise { diff --git a/test/offline-sync.test.ts b/test/offline-sync.test.ts index e810095..d50dec6 100644 --- a/test/offline-sync.test.ts +++ b/test/offline-sync.test.ts @@ -1014,7 +1014,7 @@ describe("offline SQLite synchronization", () => { await upgraded.close(); const verified = new DatabaseSync(path, { readOnly: true }); expect(sqliteSchemaContractHash(verified)).toBe(EDGE_SQLITE_SCHEMA_CONTRACTS.find( - (contract) => contract.id === "current-upgraded-project-column-migration-gate", + (contract) => contract.id === "current-upgraded-project-column-migration-gate-write-coordinator", )!.sha256); verified.close(); }); @@ -1106,7 +1106,7 @@ describe("offline SQLite synchronization", () => { }>; expect(columns.filter((column) => column.name === "project")).toHaveLength(1); expect(sqliteSchemaContractHash(upgraded)).toBe(EDGE_SQLITE_SCHEMA_CONTRACTS.find( - (contract) => contract.id === "current-upgraded-project-column-migration-gate", + (contract) => contract.id === "current-upgraded-project-column-migration-gate-write-coordinator", )!.sha256); upgraded.close(); }, process.platform === "win32" ? 45_000 : 15_000); diff --git a/test/sqlite-write-gate.test.ts b/test/sqlite-write-gate.test.ts new file mode 100644 index 0000000..43c42bc --- /dev/null +++ b/test/sqlite-write-gate.test.ts @@ -0,0 +1,77 @@ +import { execFile } from "node:child_process"; +import { rmSync } from "node:fs"; +import { createRequire } from "node:module"; +import { join } from "node:path"; +import { promisify } from "node:util"; +import { afterEach, describe, expect, it } from "vitest"; +import { SQLiteEdgeStore } from "../src/sqlite-edge-store.js"; +import { privateTestDirectory } from "./private-test-path.js"; + +const execFileAsync = promisify(execFile); +const require = createRequire(import.meta.url); +const roots: string[] = []; +afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); + +function edgePath(): string { + const root = privateTestDirectory("agent-bridge-edge-write-"); + roots.push(root); + return join(root, "edge.sqlite3"); +} + +describe("SQLite edge write coordinator", () => { + it("serializes simultaneous post, get, and ack cache mutations", async () => { + const path = edgePath(); + const worker = `import { SQLiteEdgeStore } from ${JSON.stringify(new URL("../dist/sqlite.js", import.meta.url).pathname)}; +const path=process.argv[1];const worker=Number(process.argv[2]);const edge=new SQLiteEdgeStore(path,{endpoint:'https://bridge.test',principal:{workspace:'w',agent:'codex',instance:'shared'}});await edge.initialize();for(let index=0;index<4;index+=1){const draft={id:'worker-'+worker+'-'+index,source:'codex',targets:[],type:'context',content:'x',contentType:'text/plain',priority:'info',deliveryPolicy:{mode:'mailbox'},idempotencyKey:'worker-'+worker+'-'+index};await edge.enqueue(draft);await edge.list({mailbox:'all'});let claimed;for(let attempt=0;attempt<100&&!claimed;attempt+=1){claimed=await edge.claimNext();if(!claimed)await new Promise(resolve=>setTimeout(resolve,2));}if(!claimed)throw new Error('outbox claim timed out');await edge.commit(claimed,{...claimed.draft,workspace:'w',sequence:String(worker*4+index+1),createdAt:'2026-07-26T00:00:00.000Z'});}await edge.close();`; + await execFileAsync(process.execPath, ["--input-type=module", "--eval", worker, path, "0"]); + await Promise.all(Array.from({ length: 20 }, (_, workerId) => execFileAsync( + process.execPath, ["--input-type=module", "--eval", worker, path, String(workerId + 1)], + ))); + const edge = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex", instance: "shared" } }); + await edge.initialize(); + expect(await edge.stats()).toMatchObject({ pending: 0, cached: 84 }); + await edge.close(); + }, 60_000); + + it("recovers a crashed writer lease and survives a WAL checkpoint and restart", async () => { + const path = edgePath(); + const crashedWriter = `import { DatabaseSync } from 'node:sqlite'; +const db=new DatabaseSync(process.argv[1]);db.prepare(\"UPDATE edge_write_gates SET lease_token='crashed',lease_expires_at=? WHERE gate_key='edge'\").run('1970-01-01T00:00:00.000Z');db.close();`; + const initial = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex" } }); + await initial.initialize(); + await initial.close(); + await execFileAsync(process.execPath, ["--input-type=module", "--eval", crashedWriter, path]); + + const recovered = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex" } }); + await recovered.enqueue({ id: "recovered", source: "codex", targets: [], type: "context", content: "x", contentType: "text/plain", priority: "info", deliveryPolicy: { mode: "mailbox" }, idempotencyKey: "recovered" }); + await recovered.close(); + + const { DatabaseSync } = require("node:sqlite") as typeof import("node:sqlite"); + const checkpoint = new DatabaseSync(path); + checkpoint.exec("PRAGMA wal_checkpoint(TRUNCATE)"); + checkpoint.close(); + const restarted = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex" } }); + await restarted.initialize(); + expect((await restarted.stats()).pending).toBe(1); + await restarted.close(); + }); + + it("waits behind a live writer lease without returning SQLITE_BUSY", async () => { + const path = edgePath(); + const initial = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex" } }); + await initial.initialize(); + await initial.close(); + const { DatabaseSync } = require("node:sqlite") as typeof import("node:sqlite"); + const blocker = new DatabaseSync(path); + blocker.prepare("UPDATE edge_write_gates SET lease_token='live',lease_expires_at=? WHERE gate_key='edge'") + .run(new Date(Date.now() + 100).toISOString()); + blocker.close(); + + const edge = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex" } }, 2_000, 100, 1_000); + const startedAt = Date.now(); + await edge.enqueue({ id: "after-live-lease", source: "codex", targets: [], type: "context", content: "x", contentType: "text/plain", priority: "info", deliveryPolicy: { mode: "mailbox" }, idempotencyKey: "after-live-lease" }); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(50); + await edge.close(); + }); + +});