Skip to content

fix: serialize shared edge writes #70

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
creatornader wants to merge 1 commit into from
Closed

fix: serialize shared edge writes #70

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions docs/troubleshooting.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
95 changes: 63 additions & 32 deletions src/sqlite-edge-store.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -342,6 +343,7 @@ export class SQLiteEdgeStore {
private readonly db: Database;
private readonly databasePath: string;
private readonly preexistingFiles: ReadonlySet<string>;
private readonly writeGate: SQLiteWriteGate | undefined;
private readonly key: string;
private initialized = false;
private initialization?: Promise<void>;
Expand All @@ -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<T>(work: () => T | Promise<T>): Promise<T> {
return this.writeGate ? this.writeGate.run(work) : work();
}

private restrictFiles(includeSidecars = true): void {
Expand Down Expand Up @@ -449,11 +456,12 @@ export class SQLiteEdgeStore {
}

private async initializeOnce(): Promise<void> {
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");
Expand All @@ -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<void> {
Expand All @@ -510,6 +519,7 @@ export class SQLiteEdgeStore {
/** Refuse new normal-client writes before any remote attempt begins. */
async assertScopeActive(): Promise<void> {
await this.ready();
await this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
this.assertScopeActiveInTransaction();
Expand All @@ -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<void> {
await this.ready();
if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError();
await this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
this.assertDrainLeaseInTransaction(lease, now);
Expand All @@ -532,6 +544,7 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

private async claimDrainLease(
Expand All @@ -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();
Expand All @@ -565,6 +579,7 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

/** Start a new durable drain only from an active edge scope. */
Expand All @@ -588,6 +603,7 @@ export class SQLiteEdgeStore {
async assertDrainComplete(lease: EdgeDrainLease, now = new Date()): Promise<void> {
await this.ready();
if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError();
await this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
this.assertDrainLeaseInTransaction(lease, now);
Expand All @@ -610,12 +626,14 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async renewDrainLease(lease: EdgeDrainLease, now = new Date(), leaseMs = 30_000): Promise<EdgeDrainLease> {
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);
Expand All @@ -632,11 +650,13 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async retireScope(lease: EdgeDrainLease, now = new Date()): Promise<void> {
await this.ready();
if (lease.scopeKey !== this.key) throw new EdgeMigrationLeaseError();
await this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
this.assertDrainLeaseInTransaction(lease, now);
Expand All @@ -654,6 +674,7 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async enqueue(input: PendingMessage, now = new Date()): Promise<EdgeEnqueueResult> {
Expand All @@ -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();
Expand All @@ -685,11 +707,13 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async claimNext(now = new Date(), leaseMs = 30_000, migrationLease?: EdgeDrainLease): Promise<EdgeOutboxRecord | undefined> {
await this.ready();
const nowText = now.toISOString();
return this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
this.assertClaimAuthorityInTransaction(now, migrationLease);
Expand Down Expand Up @@ -724,23 +748,24 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async retry(record: EdgeOutboxRecord, error: string, availableAt: Date): Promise<void> {
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<void> {
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);
}

Expand All @@ -750,6 +775,7 @@ export class SQLiteEdgeStore {
now = new Date(),
): Promise<void> {
await this.ready();
await this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
this.cacheOne(message);
Expand All @@ -766,6 +792,7 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

private cacheOne(message: BridgeMessage): void {
Expand Down Expand Up @@ -797,6 +824,7 @@ export class SQLiteEdgeStore {
async cachePage(messages: BridgeMessage[], cursor: string | undefined, now = new Date()): Promise<void> {
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);
Expand All @@ -817,10 +845,12 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async cacheLatest(messages: BridgeMessage[], now = new Date()): Promise<void> {
await this.ready();
await this.write(() => {
this.db.exec("BEGIN IMMEDIATE");
try {
for (const message of messages) this.cacheOne(message);
Expand All @@ -833,6 +863,7 @@ export class SQLiteEdgeStore {
this.db.exec("ROLLBACK");
throw error;
}
});
}

async list(query: MessageQuery = {}): Promise<MessagePage> {
Expand Down Expand Up @@ -927,14 +958,14 @@ export class SQLiteEdgeStore {

async noteError(error: string): Promise<void> {
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<void> {
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<EdgeStats> {
Expand Down
Loading
Loading