From d548cfbec2be18ed9a0bfbb9147b74743168d1da Mon Sep 17 00:00:00 2001 From: Nader Helmy Date: Sun, 26 Jul 2026 17:24:38 -0500 Subject: [PATCH] fix: coordinate concurrent local writes --- CHANGELOG.md | 7 +- CLAUDE.md | 2 +- README.md | 5 +- docs/architecture-v2.md | 6 ++ docs/troubleshooting.md | 20 +++--- src/sqlite-bridge-store.ts | 98 ++++++++++++++++++++++------ src/sqlite-database-contract.ts | 6 ++ src/sqlite-edge-store.ts | 51 ++++++++++----- test/bridge_store.test.ts | 6 +- test/sqlite-local-write-gate.test.ts | 82 +++++++++++++++++++++++ test/sqlite-write-gate.test.ts | 21 ++++-- 11 files changed, 241 insertions(+), 63 deletions(-) create mode 100644 test/sqlite-local-write-gate.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 43f8a70..ee2425e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,9 +8,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and ### 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. +- Serialize concurrent gateway edge and local-authority 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 diff --git a/CLAUDE.md b/CLAUDE.md index 8a5a897..aba3051 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -109,7 +109,7 @@ Sync triggers: - Immutable publisher delivery policy owns delivery mode, retry limits, and backoff. Cancel and requeue are publisher-only. Requeue resets cycle attempt but not lifetime attempt. Consumer `maxAttempts` and `retryPolicy` inputs are validated and ignored for one compatibility release. - Exact idempotent replay deduplicates. Changed content under an existing idempotency key fails. - Historical direct-database rows are handled only by migration and reconciliation commands. The normal remote path uses the authenticated gateway. -- Local and edge SQLite files use WAL, owner-only modes where supported, and bounded busy waits. Initialization and schema upgrade use a 15-second minimum retry window; normal operations retain the configured timeout. Windows verifies the private parent and main database before open, then applies explicit sidecar ACLs after WAL setup and the serialized schema transaction. Replacement during one ACL check still fails. +- Local and edge SQLite files use WAL, owner-only modes where supported, and bounded busy waits. Each file-backed store has a durable token-checked write lease that serializes mutations across independent processes, expires after a crash, and never covers gateway I/O. Initialization and schema upgrade use a 15-second minimum retry window; normal operations retain the configured timeout. Windows verifies the private parent and main database before open, then applies explicit sidecar ACLs after WAL setup and the serialized schema transaction. Replacement during one ACL check still fails. - Host-adapter manifests and installers inject identity per installed client. The v1 manifest field named `runtime` is a compatibility key for the installation target. It does not mean a live process. Installers do not write one identity into shared config. - The `codex` adapter configures the profile shared by the Codex CLI and the Codex surface in the ChatGPT desktop app. Claude Code and Claude Desktop use separate adapters and registrations. - `AGENT_BRIDGE_INSTANCE` is an optional caller-supplied stable consumer key. Supported installers generate and persist it; direct clients may manage it themselves. The gateway does not bind it to an installer registration. Unless `AGENT_BRIDGE_CURSOR` is explicit, the key selects cursor storage. It also selects leases and instance presence. Without a key or explicit cursor path, cursor storage uses `default`. Lease ownership falls back to the principal, and presence is unavailable. It is not a PID or session. Per-process presence must be additive. diff --git a/README.md b/README.md index 4f1223e..b3ede96 100644 --- a/README.md +++ b/README.md @@ -246,8 +246,9 @@ for its relationship to MCP, A2A, agmsg, brokers, and agent runtimes. - Node.js 22.23.1 or newer. - SQLite 3.51.3 or newer for local and edge storage. The supported Node version includes it. -- Concurrent local and edge initialization may wait up to 15 seconds for schema work; - normal database operations retain their configured busy timeout. +- Concurrent local and edge initialization may wait up to 15 seconds for schema work. + File-backed local and edge stores serialize mutations with a durable token-checked + lease, so independently launched MCP clients do not contend as SQLite writers. - PostgreSQL 15, 16, 17, or 18 for gateway mode. New PostgreSQL majors fail the migration prerequisite and live readiness checks until their catalog digest is certified. diff --git a/docs/architecture-v2.md b/docs/architecture-v2.md index 3847dbc..5a9c022 100644 --- a/docs/architecture-v2.md +++ b/docs/architecture-v2.md @@ -82,6 +82,12 @@ SQLite WAL provides local-only operation, a durable outbox, an inbox cache, curs The gateway outbox queues immutable message publication. Receipt and lease mutations still require the remote authority because replaying them after ownership or identity changes can settle the wrong work. +Every file-backed local authority and gateway edge database has a durable, token-checked +write lease. It serializes mutations across independently launched MCP processes, +expires after a crash, and is released only by its current token holder. The lease +covers local transactions, never gateway calls. This keeps one shared local database +correct without a daemon or per-client replicas. + Long-lived gateway MCP clients own a cancellable transport loop that replays publications and refreshes the inbox cache with bounded exponential backoff. It is transport maintenance, not agent monitoring. Manual MCP and CLI sync use the same replay path. A publication timeout after the gateway may have committed remains queued under its stable idempotency key. A retry resolves to the original immutable message instead of publishing a duplicate. Local initialization must be idempotent. Concurrent initialization and schema upgrade diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index bcd5946..ead92b7 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -81,18 +81,18 @@ 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 +## Shared SQLite 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. +cache. Local-mode clients can also share a local-authority database. Several active MCP +processes may use either database at once. Agent Bridge stores one token-checked, +short-lived write lease in each 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 `database is locked`, upgrade the package and +restart only the affected MCP host. Do not delete `bridge.sqlite3`, `edge.sqlite3`, or +their `-wal` or `-shm` sidecars while clients are running. Those files carry durable +state 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 diff --git a/src/sqlite-bridge-store.ts b/src/sqlite-bridge-store.ts index c0cd1c2..05c93e1 100644 --- a/src/sqlite-bridge-store.ts +++ b/src/sqlite-bridge-store.ts @@ -16,9 +16,6 @@ function openDatabase(path: string): Database { return new DatabaseSync(path); } const schema = ` -PRAGMA journal_mode = WAL; -PRAGMA synchronous = FULL; -PRAGMA foreign_keys = ON; CREATE TABLE IF NOT EXISTS bridge_messages ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL UNIQUE, workspace TEXT NOT NULL, project TEXT, source TEXT NOT NULL, type TEXT NOT NULL, content TEXT NOT NULL, content_type TEXT NOT NULL, data TEXT, targets TEXT NOT NULL DEFAULT '[]', @@ -83,6 +80,7 @@ CREATE INDEX IF NOT EXISTS bridge_presence_active ON bridge_presence(workspace, CREATE TRIGGER IF NOT EXISTS bridge_messages_no_update BEFORE UPDATE ON bridge_messages BEGIN SELECT RAISE(ABORT, 'bridge messages are immutable'); END; CREATE TRIGGER IF NOT EXISTS bridge_messages_no_delete BEFORE DELETE ON bridge_messages BEGIN SELECT RAISE(ABORT, 'bridge messages are immutable'); END; `; +const writeGateSchema = "CREATE TABLE IF NOT EXISTS bridge_write_gates (gate_key TEXT PRIMARY KEY, lease_token TEXT, lease_expires_at TEXT)"; function stringify(value: unknown): string | null { return value === undefined ? null : JSON.stringify(value); } function parse(value: unknown): any { return typeof value === "string" ? JSON.parse(value) : undefined; } @@ -113,6 +111,62 @@ export class SQLiteBridgeStore implements BridgeStore { } } catch (error) { this.db.close(); throw error; } } + private async write(work: () => T | Promise): Promise { + if (this.databasePath === ":memory:") return work(); + const token = randomUUID(); + const waitMs = process.platform === "win32" ? 60_000 : SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS; + const deadline = Date.now() + waitMs; + let pause = 2; + while (true) { + const now = new Date(); + const claimed = await retrySqliteBusy(() => { + this.db.exec("BEGIN IMMEDIATE"); + try { + this.db.prepare("INSERT INTO bridge_write_gates (gate_key,lease_token,lease_expires_at) VALUES ('local',NULL,NULL) ON CONFLICT(gate_key) DO NOTHING").run(); + const result = this.db.prepare("UPDATE bridge_write_gates SET lease_token=?,lease_expires_at=? WHERE gate_key='local' AND (lease_token IS NULL OR lease_expires_at<=?)") + .run(token, new Date(now.getTime() + 30_000).toISOString(), now.toISOString()); + this.db.exec("COMMIT"); + return result.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 Error("local write coordinator remained busy"); + await new Promise((resolve) => setTimeout(resolve, pause)); + pause = Math.min(Math.ceil(pause * 1.7), 25); + } + let completed = false; + let result: T; + try { + result = await work(); + completed = true; + } + finally { + try { + await retrySqliteBusy(() => { + this.db.exec("BEGIN IMMEDIATE"); + try { + this.db.prepare("UPDATE bridge_write_gates SET lease_token=NULL,lease_expires_at=NULL WHERE gate_key='local' AND lease_token=?").run(token); + this.db.exec("COMMIT"); + } catch (error) { this.db.exec("ROLLBACK"); throw error; } + }, waitMs); + } catch (error) { + if (!completed) throw error; + } + } + return result!; + } + + private async bootstrapWriteGate(timeoutMs: number): Promise { + await retrySqliteBusy(() => this.db.exec("PRAGMA journal_mode = WAL"), timeoutMs); + const exists = () => Boolean(this.db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='bridge_write_gates'").get()); + if (exists()) return; + await retrySqliteBusy(() => { + if (exists()) return; + this.db.exec("BEGIN IMMEDIATE"); + try { if (!exists()) this.db.exec(writeGateSchema); this.db.exec("COMMIT"); } + catch (error) { this.db.exec("ROLLBACK"); throw error; } + }, timeoutMs); + } private restrictFiles(includeSidecars = true): void { if (this.databasePath === ":memory:") return; const paths = includeSidecars @@ -145,9 +199,12 @@ export class SQLiteBridgeStore implements BridgeStore { if (encoded < 3_051_003) throw new Error("SQLite 3.51.3 or newer is required"); assertLocalUpgradeCandidate(this.db); const initializationTimeoutMs = Math.max(this.busyTimeoutMs, SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS); - await retrySqliteBusy(() => this.db.exec(schema), initializationTimeoutMs); + await this.bootstrapWriteGate(initializationTimeoutMs); + this.db.exec("PRAGMA synchronous = FULL; PRAGMA foreign_keys = ON"); + await this.write(async () => { await retrySqliteBusy(() => this.db.exec("BEGIN IMMEDIATE"), initializationTimeoutMs); try { + this.db.exec(schema); const columns = this.db.prepare("PRAGMA table_info(bridge_messages)").all() as Row[]; if (!columns.some((column) => column.name === "project")) { this.db.exec("ALTER TABLE bridge_messages ADD COLUMN project TEXT"); @@ -306,11 +363,12 @@ export class SQLiteBridgeStore implements BridgeStore { throw error; } this.restrictFiles(); + }); } private async ready() { await this.initialize(); } async insertMessage(input: Omit): Promise { await this.ready(); const createdAt = new Date().toISOString(); const storedPolicy = input.deliveryPolicy ?? (input.targets.length ? { mode: "leased" as const, maxAttempts: 5, retryBaseDelayMs: 1_000, retryMaxDelayMs: 60_000, retryJitterRatio: 0.2 } : { mode: "mailbox" as const }); - this.db.exec("BEGIN IMMEDIATE"); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { if (input.idempotencyKey) { const existing = this.db.prepare("SELECT * FROM bridge_messages WHERE workspace = ? AND source = ? AND idempotency_key = ?").get(input.workspace, input.source, input.idempotencyKey) as Row | undefined; if (existing) { const replay = message(existing); assertIdempotentReplay(replay, input); this.db.exec("COMMIT"); return { message: replay, created: false }; } } this.db.prepare(`INSERT INTO bridge_messages (id,workspace,project,source,type,content,content_type,data,targets,thread_id,reply_to_id,correlation_id,causation_id,priority,expires_at,idempotency_key,atrib_receipt_id,informed_by,metadata,delivery_policy,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`).run(input.id,input.workspace,input.project ?? null,input.source,input.type,input.content,input.contentType,stringify(input.data),JSON.stringify(input.targets),input.threadId ?? null,input.replyToId ?? null,input.correlationId ?? null,input.causationId ?? null,input.priority,input.expiresAt ?? null,input.idempotencyKey ?? null,input.atribReceiptId ?? null,stringify(input.informedBy),stringify(input.metadata),JSON.stringify(storedPolicy),createdAt); @@ -323,7 +381,7 @@ export class SQLiteBridgeStore implements BridgeStore { } } const row = this.db.prepare("SELECT * FROM bridge_messages WHERE id = ?").get(input.id) as Row; this.db.exec("COMMIT"); return { message: message(row), created: true }; - } catch (error) { this.db.exec("ROLLBACK"); throw error; } + } catch (error) { this.db.exec("ROLLBACK"); throw error; } }); } async listMessages(principal: BridgePrincipal, query: MessageQuery = {}): Promise { await this.ready(); const scope = cursorScope(principal, query); const cursor = decodeCursor(query.cursor, scope); const limit = Math.min(Math.max(Math.trunc(query.limit ?? 50), 1), 200); const now = new Date().toISOString(); @@ -350,10 +408,10 @@ export class SQLiteBridgeStore implements BridgeStore { if (messages.length === limit) return { messages, cursor: encodeCursor(last!.sequence, scope) }; return { messages, cursor: highWater ? encodeCursor(highWater, scope) : query.cursor }; } - async recordReceipt(principal: BridgePrincipal, messageIds: string[], readAt = new Date()): Promise { await this.ready(); const stmt = this.db.prepare("INSERT OR IGNORE INTO bridge_receipts (workspace,message_id,principal,read_at) SELECT workspace,id,?,? FROM bridge_messages WHERE workspace=? AND id=? AND (targets='[]' OR EXISTS (SELECT 1 FROM json_each(targets) WHERE value=?))"); let changed = 0; this.db.exec("BEGIN IMMEDIATE"); try { for (const id of messageIds) changed += Number(stmt.run(principal.agent,readAt.toISOString(),principal.workspace,id,principal.agent).changes); this.db.exec("COMMIT"); return changed; } catch (e) { this.db.exec("ROLLBACK"); throw e; } } + async recordReceipt(principal: BridgePrincipal, messageIds: string[], readAt = new Date()): Promise { await this.ready(); return this.write(() => { const stmt = this.db.prepare("INSERT OR IGNORE INTO bridge_receipts (workspace,message_id,principal,read_at) SELECT workspace,id,?,? FROM bridge_messages WHERE workspace=? AND id=? AND (targets='[]' OR EXISTS (SELECT 1 FROM json_each(targets) WHERE value=?))"); let changed = 0; this.db.exec("BEGIN IMMEDIATE"); try { for (const id of messageIds) changed += Number(stmt.run(principal.agent,readAt.toISOString(),principal.workspace,id,principal.agent).changes); this.db.exec("COMMIT"); return changed; } catch (e) { this.db.exec("ROLLBACK"); throw e; } }); } async claimDelivery(principal: BridgePrincipal, options: ClaimOptions): Promise { await this.ready(); const now = options.now ?? new Date(); const nowText = now.toISOString(); const expires = new Date(now.getTime() + options.leaseMs).toISOString(); const owner = principal.instance ?? principal.agent; - this.db.exec("BEGIN IMMEDIATE"); try { + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.db.prepare("UPDATE bridge_deliveries SET state='dead',last_error='message expired',lease_token=NULL,lease_owner=NULL,lease_expires_at=NULL,last_actor='agent-bridge',last_action='message_expired' WHERE workspace=? AND recipient=? AND (? IS NULL OR message_id=?) AND state IN ('pending','retrying','claimed') AND EXISTS (SELECT 1 FROM bridge_messages message WHERE message.workspace=bridge_deliveries.workspace AND message.id=bridge_deliveries.message_id AND message.expires_at IS NOT NULL AND message.expires_at<=?)").run(principal.workspace,principal.agent,options.messageId ?? null,options.messageId ?? null,nowText); this.db.prepare("UPDATE bridge_deliveries SET state='dead',last_error='maximum attempts reached',lease_token=NULL,lease_owner=NULL,lease_expires_at=NULL,last_actor='agent-bridge',last_action='attempts_exhausted' WHERE workspace=? AND recipient=? AND (? IS NULL OR message_id=?) AND cycle_attempt>=(SELECT json_extract(delivery_policy,'$.maxAttempts') FROM bridge_messages WHERE workspace=bridge_deliveries.workspace AND id=bridge_deliveries.message_id) AND (state IN ('pending','retrying') OR (state='claimed' AND lease_expires_at<=?))").run(principal.workspace,principal.agent,options.messageId ?? null,options.messageId ?? null,nowText); this.db.prepare("UPDATE bridge_deliveries SET state='retrying',available_at=?,last_error='lease expired',lease_token=NULL,lease_owner=NULL,lease_expires_at=NULL,last_actor='agent-bridge',last_action='lease_expired' WHERE workspace=? AND recipient=? AND (? IS NULL OR message_id=?) AND state='claimed' AND lease_expires_at<=? AND cycle_attempt<(SELECT json_extract(delivery_policy,'$.maxAttempts') FROM bridge_messages WHERE workspace=bridge_deliveries.workspace AND id=bridge_deliveries.message_id)").run(nowText,principal.workspace,principal.agent,options.messageId ?? null,options.messageId ?? null,nowText); @@ -362,12 +420,12 @@ export class SQLiteBridgeStore implements BridgeStore { const candidateId = String(candidate.id); const token = randomUUID(); this.db.prepare("UPDATE bridge_deliveries SET state='claimed',attempt=attempt+1,cycle_attempt=cycle_attempt+1,lease_token=?,lease_owner=?,lease_expires_at=?,last_error=NULL,last_actor=?,last_action='claim' WHERE id=?").run(token,owner,expires,principal.agent,candidateId); const claimed = this.db.prepare("SELECT * FROM bridge_deliveries WHERE id=?").get(candidateId) as Row; this.db.exec("COMMIT"); return delivery(claimed); - } catch (e) { this.db.exec("ROLLBACK"); throw e; } + } catch (e) { this.db.exec("ROLLBACK"); throw e; } }); } - async renewDelivery(principal: BridgePrincipal, id: string, token: string, leaseMs: number): Promise { await this.ready(); const now = new Date(); const expires = new Date(now.getTime() + leaseMs).toISOString(); const owner = principal.instance ?? principal.agent; const result = this.db.prepare("UPDATE bridge_deliveries SET lease_expires_at=? WHERE workspace=? AND recipient=? AND lease_owner=? AND id=? AND lease_token=? AND state='claimed' AND lease_expires_at > ?").run(expires,principal.workspace,principal.agent,owner,id,token,now.toISOString()); if (!result.changes) return null; return delivery(this.db.prepare("SELECT * FROM bridge_deliveries WHERE id=?").get(id) as Row); } + async renewDelivery(principal: BridgePrincipal, id: string, token: string, leaseMs: number): Promise { await this.ready(); return this.write(() => { const now = new Date(); const expires = new Date(now.getTime() + leaseMs).toISOString(); const owner = principal.instance ?? principal.agent; const result = this.db.prepare("UPDATE bridge_deliveries SET lease_expires_at=? WHERE workspace=? AND recipient=? AND lease_owner=? AND id=? AND lease_token=? AND state='claimed' AND lease_expires_at > ?").run(expires,principal.workspace,principal.agent,owner,id,token,now.toISOString()); if (!result.changes) return null; return delivery(this.db.prepare("SELECT * FROM bridge_deliveries WHERE id=?").get(id) as Row); }); } async settleDelivery(principal: BridgePrincipal, id: string, token: string, state: "acked" | "retrying" | "dead", error?: string, _retryPolicy?: import("./bridge-domain.js").RetryPolicy): Promise { await this.ready(); const now = new Date(); const owner = principal.instance ?? principal.agent; - this.db.exec("BEGIN IMMEDIATE"); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { const current = this.db.prepare("SELECT delivery.*,message.delivery_policy FROM bridge_deliveries delivery JOIN bridge_messages message ON message.workspace=delivery.workspace AND message.id=delivery.message_id WHERE delivery.workspace=? AND delivery.recipient=? AND delivery.lease_owner=? AND delivery.id=? AND delivery.lease_token=? AND delivery.state='claimed' AND delivery.lease_expires_at>?").get(principal.workspace,principal.agent,owner,id,token,now.toISOString()) as Row | undefined; if (!current) { this.db.exec("COMMIT"); return null; } @@ -381,7 +439,7 @@ export class SQLiteBridgeStore implements BridgeStore { const action = state === "acked" ? "ack" : state === "dead" ? "nack_dead" : exhausted ? "attempts_exhausted" : "nack_retry"; this.db.prepare("UPDATE bridge_deliveries SET state=?,available_at=?,last_error=?,lease_token=NULL,lease_owner=NULL,lease_expires_at=NULL,last_actor=?,last_action=? WHERE id=?").run(nextState,available,error?.slice(0,1024) ?? null,principal.agent,action,id); const result = delivery(this.db.prepare("SELECT * FROM bridge_deliveries WHERE id=?").get(id) as Row); this.db.exec("COMMIT"); return result; - } catch (caught) { this.db.exec("ROLLBACK"); throw caught; } + } catch (caught) { this.db.exec("ROLLBACK"); throw caught; } }); } async diagnostics(principal: BridgePrincipal): Promise { await this.ready(); @@ -407,7 +465,7 @@ export class SQLiteBridgeStore implements BridgeStore { const now = new Date(); const nowText = now.toISOString(); const expires = new Date(now.getTime() + leaseMs).toISOString(); - this.db.exec("BEGIN IMMEDIATE"); + return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { this.db.prepare("DELETE FROM bridge_presence WHERE workspace=? AND lease_expires_at<=?") .run(principal.workspace, nowText); @@ -432,13 +490,13 @@ export class SQLiteBridgeStore implements BridgeStore { } catch (error) { this.db.exec("ROLLBACK"); throw error; - } + } }); } async listPresence(principal: BridgePrincipal): Promise { await this.ready(); const now = new Date().toISOString(); - this.db.prepare("DELETE FROM bridge_presence WHERE workspace=? AND lease_expires_at<=?") - .run(principal.workspace, now); + await this.write(() => this.db.prepare("DELETE FROM bridge_presence WHERE workspace=? AND lease_expires_at<=?") + .run(principal.workspace, now)); const rows = this.db.prepare("SELECT * FROM bridge_presence WHERE workspace=? ORDER BY agent,instance") .all(principal.workspace) as Row[]; return rows.map(presence); @@ -471,7 +529,7 @@ export class SQLiteBridgeStore implements BridgeStore { return { events, cursor: events.length === limit && last ? encodeScopedCursor(scope, last.sequence) : undefined }; } async cancelDelivery(principal: BridgePrincipal,id:string) { - await this.ready(); this.db.exec("BEGIN IMMEDIATE"); + await this.ready(); return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { const current = this.db.prepare("SELECT delivery.* FROM bridge_deliveries delivery JOIN bridge_messages message ON message.workspace=delivery.workspace AND message.id=delivery.message_id WHERE delivery.id=? AND delivery.workspace=? AND message.source=?").get(id,principal.workspace,principal.agent) as Row | undefined; if (!current) { this.db.exec("COMMIT"); return null; } @@ -479,10 +537,10 @@ export class SQLiteBridgeStore implements BridgeStore { if (!["pending","retrying","claimed"].includes(String(current.state))) throw new DeliveryStateConflictError(`cannot cancel a ${current.state} delivery`); this.db.prepare("UPDATE bridge_deliveries SET state='cancelled',lease_token=NULL,lease_owner=NULL,lease_expires_at=NULL,last_error=NULL,last_actor=?,last_action='cancel' WHERE id=?").run(principal.agent,id); const result = delivery(this.db.prepare("SELECT * FROM bridge_deliveries WHERE id=?").get(id) as Row); this.db.exec("COMMIT"); return result; - } catch (error) { this.db.exec("ROLLBACK"); throw error; } + } catch (error) { this.db.exec("ROLLBACK"); throw error; } }); } async requeueDelivery(principal: BridgePrincipal,id:string) { - await this.ready(); this.db.exec("BEGIN IMMEDIATE"); + await this.ready(); return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { const current = this.db.prepare("SELECT delivery.*,message.expires_at,message.delivery_policy FROM bridge_deliveries delivery JOIN bridge_messages message ON message.workspace=delivery.workspace AND message.id=delivery.message_id WHERE delivery.id=? AND delivery.workspace=? AND message.source=?").get(id,principal.workspace,principal.agent) as Row | undefined; if (!current) { this.db.exec("COMMIT"); return null; } @@ -491,7 +549,7 @@ export class SQLiteBridgeStore implements BridgeStore { const policy = parse(current.delivery_policy); const availableAt = policy.notBefore && policy.notBefore > now ? policy.notBefore : now; this.db.prepare("UPDATE bridge_deliveries SET state='pending',available_at=?,cycle_attempt=0,requeue_count=requeue_count+1,lease_token=NULL,lease_owner=NULL,lease_expires_at=NULL,last_error=NULL,last_actor=?,last_action='requeue' WHERE id=?").run(availableAt,principal.agent,id); const result = delivery(this.db.prepare("SELECT * FROM bridge_deliveries WHERE id=?").get(id) as Row); this.db.exec("COMMIT"); return result; - } catch (error) { this.db.exec("ROLLBACK"); throw error; } + } catch (error) { this.db.exec("ROLLBACK"); throw error; } }); } async close(): Promise { this.db.close(); } } diff --git a/src/sqlite-database-contract.ts b/src/sqlite-database-contract.ts index f2f0417..22ef024 100644 --- a/src/sqlite-database-contract.ts +++ b/src/sqlite-database-contract.ts @@ -8,6 +8,10 @@ export const EDGE_SQLITE_APPLICATION_ID = 0x41424745; export const SQLITE_DATABASE_SCHEMA_VERSION = 1; export const SQLITE_METADATA_TABLE = "agent_bridge_metadata"; export const LOCAL_SQLITE_SCHEMA_CONTRACTS = Object.freeze([ + Object.freeze({ id: "current-upgraded-write-coordinator", sha256: "d8504fc1092045ac5dc50aa64b0291a68ccb57af1cfebe5d1997b79f433b2eab" }), + Object.freeze({ id: "current-upgraded-project-column-write-coordinator", sha256: "37845f7e49b4c56dffe40295ffb31320d4b33c2acc2873e48514a1b8e36a490b" }), + Object.freeze({ id: "current-upgraded-delivery-policy-write-coordinator", sha256: "a0079dfe925a529b613425f779b5d978464a4c6b23e6cb7439fbf74f70da89bb" }), + Object.freeze({ id: "current-upgraded-delivery-events-write-coordinator", sha256: "a7de84c001b5f467a10c3aefdb6a92c86c7ce621cb5889cde90bc3c9f0b2f2ab" }), Object.freeze({ id: "current-created-schema", sha256: "e2f978c27ea5d151a4b4b3ec349166833441e42401fedfe60dc3ba20f3d713aa" }), Object.freeze({ id: "current-upgraded-project-column", sha256: "cb0975bbc1ccc7a2a66d8f4d76df619804449cf81119972337afad6cdab64451" }), Object.freeze({ id: "current-upgraded-delivery-policy", sha256: "d4fd8905ea73057b27662454c16c038834f5a1bf017db9cfc041792104aa2e6d" }), @@ -55,6 +59,7 @@ const localColumns: Record = { "requeue_count", "lease_owner", "error", "actor", "action", "created_at", ], bridge_presence: ["workspace", "agent", "instance", "runtime_type", "capabilities", "lease_expires_at", "last_seen_at"], + bridge_write_gates: ["gate_key", "lease_token", "lease_expires_at"], }; const localObjects = new Map([ @@ -64,6 +69,7 @@ const localObjects = new Map([ ["bridge_messages", ["table", "bridge_messages"]], ["bridge_presence", ["table", "bridge_presence"]], ["bridge_receipts", ["table", "bridge_receipts"]], + ["bridge_write_gates", ["table", "bridge_write_gates"]], ["bridge_deliveries_claim", ["index", "bridge_deliveries"]], ["bridge_deliveries_publisher", ["index", "bridge_deliveries"]], ["bridge_deliveries_terminal", ["index", "bridge_deliveries"]], diff --git a/src/sqlite-edge-store.ts b/src/sqlite-edge-store.ts index ec201f6..d9ef8ab 100644 --- a/src/sqlite-edge-store.ts +++ b/src/sqlite-edge-store.ts @@ -31,9 +31,6 @@ function openReadOnlyDatabase(path: string): Database { const MAX_SEQUENCE = 9_223_372_036_854_775_807n; const schema = ` -PRAGMA journal_mode = WAL; -PRAGMA synchronous = FULL; -PRAGMA foreign_keys = ON; CREATE TABLE IF NOT EXISTS edge_scopes ( scope_key TEXT PRIMARY KEY, endpoint_hash TEXT NOT NULL, @@ -115,12 +112,8 @@ 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 -); `; +const writeGateSchema = "CREATE TABLE IF NOT EXISTS edge_write_gates (gate_key TEXT PRIMARY KEY, lease_token TEXT, lease_expires_at TEXT)"; type Row = Record; export type PendingMessage = Omit; @@ -402,16 +395,38 @@ export class SQLiteEdgeStore { await new Promise((resolve) => setTimeout(resolve, pause)); pause = Math.min(Math.ceil(pause * 1.7), 25); } - try { return await work(); } + let completed = false; + let result: T; + try { + result = await work(); + completed = true; + } 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))); + try { + 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))); + } catch (error) { + if (!completed) throw error; + } } + return result!; + } + + private async bootstrapWriteGate(timeoutMs: number): Promise { + await retrySqliteBusy(() => this.db.exec("PRAGMA journal_mode = WAL"), timeoutMs); + const exists = () => Boolean(this.db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='edge_write_gates'").get()); + if (exists()) return; + await retrySqliteBusy(() => { + if (exists()) return; + this.db.exec("BEGIN IMMEDIATE"); + try { if (!exists()) this.db.exec(writeGateSchema); this.db.exec("COMMIT"); } + catch (error) { this.db.exec("ROLLBACK"); throw error; } + }, timeoutMs); } private restrictFiles(includeSidecars = true): void { @@ -502,10 +517,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 this.bootstrapWriteGate(initializationTimeoutMs); + this.db.exec("PRAGMA synchronous = FULL; PRAGMA foreign_keys = ON"); await this.write(async () => { await retrySqliteBusy(() => this.db.exec("BEGIN IMMEDIATE"), initializationTimeoutMs); try { + this.db.exec(schema); 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"); diff --git a/test/bridge_store.test.ts b/test/bridge_store.test.ts index 32014c8..ff925b4 100644 --- a/test/bridge_store.test.ts +++ b/test/bridge_store.test.ts @@ -709,7 +709,7 @@ describe("SQLite project schema upgrade", () => { }); expect(page.messages[0]?.project).toBeUndefined(); const upgraded = new DatabaseSync(path); - expectUpgradedMessageInsertContract(upgraded, "current-upgraded-project-column"); + expectUpgradedMessageInsertContract(upgraded, "current-upgraded-project-column-write-coordinator"); upgraded.close(); }); @@ -818,7 +818,7 @@ describe("SQLite project schema upgrade", () => { mode: "leased", maxAttempts: 4, retryBaseDelayMs: 2_000, retryMaxDelayMs: 70_000, retryJitterRatio: 0.3, futureLeased: "keep", }); - expectUpgradedMessageInsertContract(upgraded, "current-upgraded-delivery-policy"); + expectUpgradedMessageInsertContract(upgraded, "current-upgraded-delivery-policy-write-coordinator"); upgraded.close(); }); @@ -904,7 +904,7 @@ describe("SQLite project schema upgrade", () => { { action: "nack_dead", actor: "worker-instance" }, { action: "attempts_exhausted", actor: "agent-bridge" }, ]); - expectUpgradedMessageInsertContract(upgraded, "current-upgraded-delivery-events"); + expectUpgradedMessageInsertContract(upgraded, "current-upgraded-delivery-events-write-coordinator"); upgraded.close(); }, process.platform === "win32" ? 45_000 : 15_000); }); diff --git a/test/sqlite-local-write-gate.test.ts b/test/sqlite-local-write-gate.test.ts new file mode 100644 index 0000000..2fe4938 --- /dev/null +++ b/test/sqlite-local-write-gate.test.ts @@ -0,0 +1,82 @@ +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 { SQLiteBridgeStore } from "../src/sqlite-bridge-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 databasePath(): string { + const root = privateTestDirectory("agent-bridge-local-write-"); + roots.push(root); + return join(root, "bridge.sqlite3"); +} + +describe("SQLite local-authority write coordinator", () => { + it("serializes simultaneous post, get, and receipt mutations", async () => { + const path = databasePath(); + const worker = `import { SQLiteBridgeStore } from ${JSON.stringify(new URL("../dist/sqlite.js", import.meta.url).pathname)}; +const path=process.argv[1];const worker=Number(process.argv[2]);const store=new SQLiteBridgeStore(path,20);await store.initialize();for(let index=0;index<4;index+=1){const id='worker-'+worker+'-'+index;await store.insertMessage({id,workspace:'w',source:'codex',targets:[],type:'context',content:'x',contentType:'text/plain',priority:'info',deliveryPolicy:{mode:'mailbox'}});await store.listMessages({workspace:'w',agent:'codex'},{mailbox:'all'});await store.recordReceipt({workspace:'w',agent:'codex'},[id]);}await store.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 store = new SQLiteBridgeStore(path); + await store.initialize(); + expect((await store.listMessages({ workspace: "w", agent: "codex" }, { mailbox: "all", limit: 200 })).messages).toHaveLength(84); + expect(await store.recordReceipt({ workspace: "w", agent: "codex" }, ["worker-0-0"])).toBe(0); + await store.close(); + }, 60_000); + + it("recovers a crashed writer lease and survives a WAL checkpoint and restart", async () => { + const path = databasePath(); + const initial = new SQLiteBridgeStore(path); + await initial.initialize(); + await initial.close(); + const crashedWriter = `import { DatabaseSync } from 'node:sqlite';const db=new DatabaseSync(process.argv[1]);db.prepare("UPDATE bridge_write_gates SET lease_token='crashed',lease_expires_at=? WHERE gate_key='local'").run(new Date(Date.now()+100).toISOString());process.kill(process.pid,'SIGKILL');`; + await execFileAsync(process.execPath, ["--input-type=module", "--eval", crashedWriter, path]).catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 120)); + + const recovered = new SQLiteBridgeStore(path); + await recovered.insertMessage({ id: "recovered", workspace: "w", source: "codex", targets: [], type: "context", content: "x", contentType: "text/plain", priority: "info", deliveryPolicy: { mode: "mailbox" } }); + 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 SQLiteBridgeStore(path); + await restarted.initialize(); + expect((await restarted.listMessages({ workspace: "w", agent: "codex" })).messages.map((message) => message.id)).toEqual(["recovered"]); + await restarted.close(); + }); + + it("waits for a long transaction after its writer lease expires", async () => { + const path = databasePath(); + const initial = new SQLiteBridgeStore(path); + await initial.initialize(); + await initial.close(); + const { DatabaseSync } = require("node:sqlite") as typeof import("node:sqlite"); + const blocker = new DatabaseSync(path); + blocker.exec("BEGIN IMMEDIATE"); + blocker.prepare("UPDATE bridge_write_gates SET lease_token='live',lease_expires_at=? WHERE gate_key='local'") + .run(new Date(Date.now() + 100).toISOString()); + + const store = new SQLiteBridgeStore(path); + const startedAt = Date.now(); + const pending = store.insertMessage({ id: "after-live-lease", workspace: "w", source: "codex", targets: [], type: "context", content: "x", contentType: "text/plain", priority: "info", deliveryPolicy: { mode: "mailbox" } }); + await new Promise((resolve) => setTimeout(resolve, 250)); + blocker.exec("COMMIT"); + blocker.close(); + await pending; + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(200); + await store.close(); + }); +}); diff --git a/test/sqlite-write-gate.test.ts b/test/sqlite-write-gate.test.ts index 43c42bc..0a16396 100644 --- a/test/sqlite-write-gate.test.ts +++ b/test/sqlite-write-gate.test.ts @@ -36,11 +36,12 @@ const path=process.argv[1];const worker=Number(process.argv[2]);const edge=new S 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 db=new DatabaseSync(process.argv[1]);db.prepare(\"UPDATE edge_write_gates SET lease_token='crashed',lease_expires_at=? WHERE gate_key='edge'\").run(new Date(Date.now()+100).toISOString());process.kill(process.pid,'SIGKILL');`; 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]); + await execFileAsync(process.execPath, ["--input-type=module", "--eval", crashedWriter, path]).catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 120)); 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" }); @@ -56,21 +57,27 @@ const db=new DatabaseSync(process.argv[1]);db.prepare(\"UPDATE edge_write_gates await restarted.close(); }); - it("waits behind a live writer lease without returning SQLITE_BUSY", async () => { + it("waits for a long transaction after its writer lease expires", 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.exec("BEGIN IMMEDIATE"); 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 edge = new SQLiteEdgeStore(path, { endpoint: "https://bridge.test", principal: { workspace: "w", agent: "codex" } }, 2_000, 100, 5_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); + const pending = 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" }) + .then(() => undefined, (error: unknown) => error); + await new Promise((resolve) => setTimeout(resolve, 250)); + blocker.exec("COMMIT"); + blocker.close(); + const error = await pending; + if (error) throw error; + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(200); await edge.close(); });