From fbc9f248ff0cec1196cc43cc4ac32ccbe06d0f28 Mon Sep 17 00:00:00 2001 From: Nader Helmy Date: Sun, 26 Jul 2026 15:31:42 -0500 Subject: [PATCH] fix: serialize shared edge writes --- docs/troubleshooting.md | 16 ++++++ src/sqlite-edge-store.ts | 95 ++++++++++++++++++++++------------ src/sqlite-write-gate.ts | 76 +++++++++++++++++++++++++++ test/sqlite-write-gate.test.ts | 46 ++++++++++++++++ 4 files changed, 201 insertions(+), 32 deletions(-) create mode 100644 src/sqlite-write-gate.ts create mode 100644 test/sqlite-write-gate.test.ts diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 877d371..ce595f1 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -81,6 +81,22 @@ 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 +serializes local mutations with a short-lived crash-recoverable write gate. It does not +hold that gate while it 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`, its `-wal` or `-shm` sidecars, or a write-lock file 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-edge-store.ts b/src/sqlite-edge-store.ts index 3e1757e..2fbbb73 100644 --- a/src/sqlite-edge-store.ts +++ b/src/sqlite-edge-store.ts @@ -15,6 +15,7 @@ import type { MessagePage, MessageQuery } from "./bridge-store.js"; import { retrySqliteBusy, SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS } from "./sqlite-retry.js"; import { preparePrivateSqliteLocation, securePrivatePath, securePrivateSqliteSidecar, verifyPrivatePathAccess } from "./private-path.js"; import { assertEdgeUpgradeCandidate, installEdgeMarkers } from "./sqlite-database-contract.js"; +import { SQLiteWriteGate } from "./sqlite-write-gate.js"; const require = createRequire(import.meta.url); function openDatabase(path: string): Database { @@ -342,6 +343,7 @@ export class SQLiteEdgeStore { private readonly db: Database; private readonly databasePath: string; private readonly preexistingFiles: ReadonlySet; + private readonly writeGate: SQLiteWriteGate | undefined; private readonly key: string; private initialized = false; private initialization?: Promise; @@ -361,6 +363,11 @@ export class SQLiteEdgeStore { } } catch (error) { this.db.close(); throw error; } this.key = edgeScopeKey(scope); + this.writeGate = selected === ":memory:" ? undefined : new SQLiteWriteGate(selected); + } + + private async write(work: () => T | Promise): Promise { + return this.writeGate ? this.writeGate.run(work) : work(); } private restrictFiles(includeSidecars = true): void { @@ -449,11 +456,12 @@ export class SQLiteEdgeStore { } private async initializeOnce(): Promise { - 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 () => { + 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 { 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 +481,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 +519,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 +528,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 +544,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } private async claimDrainLease( @@ -540,6 +553,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 +579,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } /** Start a new durable drain only from an active edge scope. */ @@ -588,6 +603,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 +626,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 +650,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 +674,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async enqueue(input: PendingMessage, now = new Date()): Promise { @@ -662,6 +683,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 +707,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 +748,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 +775,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 +792,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } private cacheOne(message: BridgeMessage): void { @@ -797,6 +824,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 +845,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 +863,7 @@ export class SQLiteEdgeStore { this.db.exec("ROLLBACK"); throw error; } + }); } async list(query: MessageQuery = {}): Promise { @@ -927,14 +958,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/src/sqlite-write-gate.ts b/src/sqlite-write-gate.ts new file mode 100644 index 0000000..3788471 --- /dev/null +++ b/src/sqlite-write-gate.ts @@ -0,0 +1,76 @@ +import { closeSync, existsSync, openSync, readFileSync, rmSync, unlinkSync, writeFileSync } from "node:fs"; +import { hostname } from "node:os"; + +const DEFAULT_WAIT_MS = 5_000; + +export class SQLiteWriteGateError extends Error { + constructor() { + super("edge write coordinator remained busy"); + this.name = "SQLiteWriteGateError"; + this.code = "edge_write_coordinator_busy"; + } + + readonly code: string; +} + +interface LockRecord { + schema: "agent-bridge.edge-write-lock"; + version: 1; + pid: number; + host: string; +} + +function delay(milliseconds: number): Promise { + return new Promise((resolve) => setTimeout(resolve, milliseconds)); +} + +function liveLocalHolder(path: string): boolean { + try { + const record = JSON.parse(readFileSync(path, "utf8")) as Partial; + if (record.schema !== "agent-bridge.edge-write-lock" || record.version !== 1 + || record.host !== hostname() || !Number.isInteger(record.pid) || record.pid! <= 0) return false; + process.kill(record.pid!, 0); + return true; + } catch (error) { + return !(error && typeof error === "object" && "code" in error && (error.code === "ESRCH" || error.code === "ENOENT")); + } +} + +export class SQLiteWriteGate { + private readonly path: string; + + constructor(databasePath: string, private readonly waitMs = DEFAULT_WAIT_MS) { + this.path = `${databasePath}.write.lock`; + } + + async run(work: () => T | Promise): Promise { + const deadline = Date.now() + Math.max(1, Math.trunc(this.waitMs)); + let descriptor: number | undefined; + let pause = 2; + while (descriptor === undefined) { + try { + descriptor = openSync(this.path, "wx", 0o600); + writeFileSync(descriptor, `${JSON.stringify({ schema: "agent-bridge.edge-write-lock", version: 1, pid: process.pid, host: hostname() })}\n`); + } catch (error) { + if (descriptor !== undefined) throw error; + if (!(error && typeof error === "object" && "code" in error && error.code === "EEXIST")) throw error; + if (!liveLocalHolder(this.path)) { + try { unlinkSync(this.path); } catch (removeError) { + if (!(removeError && typeof removeError === "object" && "code" in removeError && removeError.code === "ENOENT")) throw removeError; + } + continue; + } + if (Date.now() >= deadline) throw new SQLiteWriteGateError(); + await delay(pause); + pause = Math.min(Math.ceil(pause * 1.7), 25); + } + } + try { + return await work(); + } finally { + try { closeSync(descriptor); } finally { + if (existsSync(this.path)) rmSync(this.path, { force: true }); + } + } + } +} diff --git a/test/sqlite-write-gate.test.ts b/test/sqlite-write-gate.test.ts new file mode 100644 index 0000000..634af58 --- /dev/null +++ b/test/sqlite-write-gate.test.ts @@ -0,0 +1,46 @@ +import { execFile } from "node:child_process"; +import { existsSync, rmSync } from "node:fs"; +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 { SQLiteWriteGate } from "../src/sqlite-write-gate.js"; +import { privateTestDirectory } from "./private-test-path.js"; + +const execFileAsync = promisify(execFile); +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 multi-process outbox writes", 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=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<32;index+=1){await edge.enqueue({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.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()).pending).toBe(672); + await edge.close(); + }, 60_000); + + it("reclaims a crashed writer and releases after a long local critical section", async () => { + const path = edgePath(); + const lock = `${path}.write.lock`; + const crashedWriter = `const fs=require('node:fs');const os=require('node:os');fs.writeFileSync(process.argv[1],JSON.stringify({schema:'agent-bridge.edge-write-lock',version:1,pid:process.pid,host:os.hostname()})+'\\n',{mode:0o600});`; + await execFileAsync(process.execPath, ["--eval", crashedWriter, lock]); + const gate = new SQLiteWriteGate(path, 1_000); + const started = Date.now(); + await gate.run(async () => { await new Promise((resolve) => setTimeout(resolve, 80)); }); + expect(Date.now() - started).toBeGreaterThanOrEqual(75); + expect(existsSync(lock)).toBe(false); + }); +});