From fc6b48ea952b1d5dafca62e0b6de769a391525ca Mon Sep 17 00:00:00 2001 From: Nader Helmy Date: Sun, 26 Jul 2026 22:35:45 -0500 Subject: [PATCH] fix: preserve SQLite transaction errors --- CHANGELOG.md | 11 ++++++++++- ROADMAP.md | 5 +++++ package-lock.json | 4 ++-- package.json | 2 +- src/sqlite-bridge-store.ts | 24 ++++++++++++------------ src/sqlite-edge-store.ts | 34 +++++++++++++++++----------------- src/sqlite-retry.ts | 6 ++++++ test/sqlite-write-gate.test.ts | 10 ++++++++++ 8 files changed, 63 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d6b099c..e51f4c2 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). +## [0.6.4] - 2026-07-27 + +### Fixed + +- Preserve the original SQLite failure when SQLite has already ended a failed + transaction. Local and edge stores no longer replace it with a second + "cannot rollback" error. + ## [0.6.3] - 2026-07-26 ### Fixed @@ -453,7 +461,8 @@ First tagged release. Marks the point where agent-bridge has shipped its initial - Narrative-leak detection in CI + on commit via `creatornader/textleaks@v0.2.0` (renamed from leakguard). [0.1.0]: https://github.com/creatornader/agent-bridge/releases/tag/v0.1.0 -[Unreleased]: https://github.com/creatornader/agent-bridge/compare/v0.6.3...HEAD +[Unreleased]: https://github.com/creatornader/agent-bridge/compare/v0.6.4...HEAD +[0.6.4]: https://github.com/creatornader/agent-bridge/compare/v0.6.3...v0.6.4 [0.6.3]: https://github.com/creatornader/agent-bridge/compare/v0.6.2...v0.6.3 [0.6.2]: https://github.com/creatornader/agent-bridge/compare/v0.6.1...v0.6.2 [0.6.1]: https://github.com/creatornader/agent-bridge/compare/v0.6.0...v0.6.1 diff --git a/ROADMAP.md b/ROADMAP.md index 8510539..7977814 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -165,6 +165,11 @@ The published npm version is the authority for whether this package line has shi expired crashed-writer lease safely, and keep gateway calls outside the local write boundary. +### 0.6.4 package contents + +- Preserve primary SQLite errors when the engine already ended a failed + transaction, rather than raising a secondary rollback error. + ### Production validation and adoption Completed through the published and deployed 0.6.1 package: diff --git a/package-lock.json b/package-lock.json index b2a217f..7748144 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@creatornader/agent-bridge", - "version": "0.6.3", + "version": "0.6.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@creatornader/agent-bridge", - "version": "0.6.3", + "version": "0.6.4", "license": "Apache-2.0", "dependencies": { "@modelcontextprotocol/sdk": "^1.12.1", diff --git a/package.json b/package.json index 0b1490b..df80422 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@creatornader/agent-bridge", - "version": "0.6.3", + "version": "0.6.4", "description": "Let AI agents message each other and hand off work across tools, sessions, and machines.", "type": "module", "license": "Apache-2.0", diff --git a/src/sqlite-bridge-store.ts b/src/sqlite-bridge-store.ts index 05c93e1..e2a9220 100644 --- a/src/sqlite-bridge-store.ts +++ b/src/sqlite-bridge-store.ts @@ -5,7 +5,7 @@ import type { DatabaseSync as Database, SQLInputValue } from "node:sqlite"; import { DeliveryStateConflictError, cursorScope, decodeCursor, decodeScopedCursor, encodeCursor, encodeScopedCursor, scopedCursorScope, validateDeliveryCursorPosition, validateEventCursorPosition, type AgentPresence, type BridgeDelivery, type BridgeDeliveryEvent, type BridgeMessage, type BridgePrincipal } from "./bridge-domain.js"; import type { BridgeDiagnostics, BridgeStore, ClaimOptions, DeliveryQuery, InsertMessageResult, MessagePage, MessageQuery } from "./bridge-store.js"; import { assertIdempotentReplay } from "./idempotency.js"; -import { retrySqliteBusy, SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS } from "./sqlite-retry.js"; +import { retrySqliteBusy, rollbackSqliteTransaction, SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS } from "./sqlite-retry.js"; import { preparePrivateSqliteLocation, securePrivatePath, securePrivateSqliteSidecar, verifyPrivatePathAccess } from "./private-path.js"; import { assertLocalUpgradeCandidate, installLocalAuthorityMarkers } from "./sqlite-database-contract.js"; @@ -127,7 +127,7 @@ export class SQLiteBridgeStore implements BridgeStore { .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; } + } catch (error) { rollbackSqliteTransaction(this.db); throw error; } }, Math.max(1, deadline - Date.now())); if (claimed) break; if (Date.now() >= deadline) throw new Error("local write coordinator remained busy"); @@ -147,7 +147,7 @@ export class SQLiteBridgeStore implements BridgeStore { 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; } + } catch (error) { rollbackSqliteTransaction(this.db); throw error; } }, waitMs); } catch (error) { if (!completed) throw error; @@ -164,7 +164,7 @@ export class SQLiteBridgeStore implements BridgeStore { 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; } + catch (error) { rollbackSqliteTransaction(this.db); throw error; } }, timeoutMs); } private restrictFiles(includeSidecars = true): void { @@ -359,7 +359,7 @@ export class SQLiteBridgeStore implements BridgeStore { installLocalAuthorityMarkers(this.db); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } this.restrictFiles(); @@ -381,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) { rollbackSqliteTransaction(this.db); 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(); @@ -408,7 +408,7 @@ 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(); 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 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) { rollbackSqliteTransaction(this.db); 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; return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); try { @@ -420,7 +420,7 @@ 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) { rollbackSqliteTransaction(this.db); throw e; } }); } 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 { @@ -439,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) { rollbackSqliteTransaction(this.db); throw caught; } }); } async diagnostics(principal: BridgePrincipal): Promise { await this.ready(); @@ -488,7 +488,7 @@ export class SQLiteBridgeStore implements BridgeStore { this.db.exec("COMMIT"); return result; } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); } @@ -537,7 +537,7 @@ 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) { rollbackSqliteTransaction(this.db); throw error; } }); } async requeueDelivery(principal: BridgePrincipal,id:string) { await this.ready(); return this.write(() => { this.db.exec("BEGIN IMMEDIATE"); @@ -549,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) { rollbackSqliteTransaction(this.db); throw error; } }); } async close(): Promise { this.db.close(); } } diff --git a/src/sqlite-edge-store.ts b/src/sqlite-edge-store.ts index d9ef8ab..10a1056 100644 --- a/src/sqlite-edge-store.ts +++ b/src/sqlite-edge-store.ts @@ -12,7 +12,7 @@ import { type JsonValue, } from "./bridge-domain.js"; import type { MessagePage, MessageQuery } from "./bridge-store.js"; -import { retrySqliteBusy, SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS } from "./sqlite-retry.js"; +import { retrySqliteBusy, rollbackSqliteTransaction, 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"; @@ -388,7 +388,7 @@ export class SQLiteEdgeStore { ); this.db.exec("COMMIT"); return update.changes === 1; - } catch (error) { this.db.exec("ROLLBACK"); throw error; } + } catch (error) { rollbackSqliteTransaction(this.db); throw error; } }, Math.max(1, deadline - Date.now())); if (claimed) break; if (Date.now() >= deadline) throw new EdgeConflictError("edge write coordinator remained busy"); @@ -408,7 +408,7 @@ export class SQLiteEdgeStore { 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; } + } catch (error) { rollbackSqliteTransaction(this.db); throw error; } }, Math.max(1, Math.trunc(this.writeGateWaitMs))); } catch (error) { if (!completed) throw error; @@ -425,7 +425,7 @@ export class SQLiteEdgeStore { 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; } + catch (error) { rollbackSqliteTransaction(this.db); throw error; } }, timeoutMs); } @@ -544,7 +544,7 @@ export class SQLiteEdgeStore { installEdgeMarkers(this.db); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } this.restrictFiles(); @@ -563,7 +563,7 @@ export class SQLiteEdgeStore { 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; } + } catch (error) { rollbackSqliteTransaction(this.db); throw error; } } }); } @@ -586,7 +586,7 @@ export class SQLiteEdgeStore { this.assertScopeActiveInTransaction(); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -602,7 +602,7 @@ export class SQLiteEdgeStore { this.assertDrainLeaseInTransaction(lease, now); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -637,7 +637,7 @@ export class SQLiteEdgeStore { this.db.exec("COMMIT"); return { scopeKey: this.key, operationId: normalizedOperationId, leaseToken, leaseExpiresAt }; } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -684,7 +684,7 @@ export class SQLiteEdgeStore { } this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -708,7 +708,7 @@ export class SQLiteEdgeStore { this.db.exec("COMMIT"); return { ...lease, leaseExpiresAt }; } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -732,7 +732,7 @@ export class SQLiteEdgeStore { if (updated.changes !== 1) throw new EdgeMigrationLeaseError(); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -765,7 +765,7 @@ export class SQLiteEdgeStore { this.db.exec("COMMIT"); return { draft, created: true }; } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -806,7 +806,7 @@ export class SQLiteEdgeStore { createdAt: String(head.created_at), }; } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -850,7 +850,7 @@ export class SQLiteEdgeStore { .run(now.toISOString(), this.key); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -903,7 +903,7 @@ export class SQLiteEdgeStore { .run(now.toISOString(), this.key); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); @@ -921,7 +921,7 @@ export class SQLiteEdgeStore { .run(now.toISOString(), this.key); this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + rollbackSqliteTransaction(this.db); throw error; } }); diff --git a/src/sqlite-retry.ts b/src/sqlite-retry.ts index c106155..a1897e9 100644 --- a/src/sqlite-retry.ts +++ b/src/sqlite-retry.ts @@ -22,3 +22,9 @@ export async function retrySqliteBusy( } } } + +/** SQLite may already have rolled a transaction back after a statement error. */ +export function rollbackSqliteTransaction(database: DatabaseSync): void { + if (database.isTransaction) database.exec("ROLLBACK"); +} +import type { DatabaseSync } from "node:sqlite"; diff --git a/test/sqlite-write-gate.test.ts b/test/sqlite-write-gate.test.ts index 0a16396..1a53359 100644 --- a/test/sqlite-write-gate.test.ts +++ b/test/sqlite-write-gate.test.ts @@ -5,6 +5,7 @@ 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 { rollbackSqliteTransaction } from "../src/sqlite-retry.js"; import { privateTestDirectory } from "./private-test-path.js"; const execFileAsync = promisify(execFile); @@ -19,6 +20,15 @@ function edgePath(): string { } describe("SQLite edge write coordinator", () => { + it("does not mask an error after SQLite already ended a transaction", () => { + const { DatabaseSync } = require("node:sqlite") as typeof import("node:sqlite"); + const database = new DatabaseSync(":memory:"); + database.exec("BEGIN IMMEDIATE"); + database.exec("ROLLBACK"); + expect(() => rollbackSqliteTransaction(database)).not.toThrow(); + database.close(); + }); + 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)};