diff --git a/src/services/logger.ts b/src/services/logger.ts index d7259d6d..255fa12e 100644 --- a/src/services/logger.ts +++ b/src/services/logger.ts @@ -141,6 +141,19 @@ export function serializeError(error: unknown): { return { error: String(error) }; } +/** + * Log without letting a logger failure escape. `log()` initialises the log + * directory lazily and can throw on ENOSPC/EACCES; callers on a teardown path + * must never let a diagnostic write skip the work that follows it. + */ +export function safeLog(message: string, data?: unknown): void { + try { + log(message, data); + } catch { + // swallow — the caller's cleanup matters more than this line + } +} + export function log(message: string, data?: unknown) { ensureLoggerInitialized(); diff --git a/src/services/turso/connection-manager.ts b/src/services/turso/connection-manager.ts index 21909bb4..cd634c35 100644 --- a/src/services/turso/connection-manager.ts +++ b/src/services/turso/connection-manager.ts @@ -3,7 +3,7 @@ import type { DatabaseOpts, EncryptionOpts } from "@tursodatabase/database-commo import { existsSync, mkdirSync } from "node:fs"; import { dirname, resolve, relative, isAbsolute, sep } from "node:path"; import { CONFIG } from "../../config.js"; -import { log } from "../logger.js"; +import { safeLog } from "../logger.js"; import { collectReleasedSqliteHandles } from "./sqlite-handle-release.js"; import { TursoDb } from "./turso-db.js"; import { resolveOrCreateDatabaseEncryptionKey } from "./encryption-key.js"; @@ -188,7 +188,7 @@ export class TursoConnectionManager { try { await db.close(); } catch (error) { - log("Error closing Turso database", { path: dbPath, error: String(error) }); + safeLog("Error closing Turso database", { path: dbPath, error: String(error) }); } this.connections.delete(dbPath); @@ -209,13 +209,17 @@ export class TursoConnectionManager { this.closingPromise = (async () => { await Promise.allSettled([...this.pending.values(), ...this.closingConnections.values()]); - for (const [path, db] of this.connections) { - try { - await db.close(); - } catch (error) { - log("Error closing Turso database", { path, error: String(error) }); - } - } + // Close in parallel: each close runs a bounded WAL checkpoint, and the + // host allows only ~5s between SIGTERM and SIGKILL. + await Promise.allSettled( + [...this.connections].map(async ([path, db]) => { + try { + await db.close(); + } catch (error) { + safeLog("Error closing Turso database", { path, error: String(error) }); + } + }) + ); this.connections.clear(); this.pending.clear(); await collectReleasedSqliteHandles(); @@ -231,9 +235,9 @@ export class TursoConnectionManager { closeAllSync(): void { for (const [path, db] of this.connections) { try { - void db.close(); + db.closeSync(); } catch (error) { - log("Error closing Turso database (sync)", { path, error: String(error) }); + safeLog("Error closing Turso database (sync)", { path, error: String(error) }); } } this.connections.clear(); diff --git a/src/services/turso/turso-db.ts b/src/services/turso/turso-db.ts index b84cba02..f7f3105b 100644 --- a/src/services/turso/turso-db.ts +++ b/src/services/turso/turso-db.ts @@ -1,4 +1,5 @@ import type { Database } from "@tursodatabase/database"; +import { safeLog } from "../logger.js"; export type SqlValue = null | string | number | bigint | boolean | Uint8Array | ArrayBuffer; export type SqlArgs = SqlValue[] | Record; @@ -112,8 +113,18 @@ function bindArgs(args?: SqlArgs): SqlValue[] | [Record] { return [normalized]; } +/** + * Upper bound for the pre-close WAL checkpoint. The host SIGTERMs and SIGKILLs + * the service after ~5s, so a checkpoint that cannot finish (another process + * holds a read lock) must not eat the whole shutdown budget. + */ +const CHECKPOINT_TIMEOUT_MS = 1500; + export class TursoDb { - constructor(private readonly database: Database) {} + constructor( + private readonly database: Database, + private readonly checkpointTimeoutMs: number = CHECKPOINT_TIMEOUT_MS + ) {} getClient(): Database { return this.database; @@ -203,7 +214,66 @@ export class TursoDb { return wrapped[txnMode](); } + /** + * Checkpoint the WAL into the main DB before closing so a SIGKILL during + * teardown cannot leave a 0-byte `-wal` beside a stale `-tshm` (every later + * open then fails with "short read on WAL frame"). Never throws: a busy or + * slow checkpoint is logged and skipped. + * + * The timeout only bounds how long we *prefer* to wait before logging a + * failure — turso's `close()` does not take `execLock`, so we must still + * await settlement of the pragma before returning. Closing under an + * in-flight statement would recreate the tear this helper exists to avoid. + * A truncate stuck on a foreign reader can therefore still hold shutdown + * past the bound until the reader leaves (or the host SIGKILLs). + */ + private async checkpointBeforeClose(): Promise { + let timer: ReturnType | undefined; + const checkpoint = this.database.all("PRAGMA wal_checkpoint(TRUNCATE)"); + try { + const rows = await Promise.race([ + checkpoint, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`timed out after ${this.checkpointTimeoutMs}ms`)), + this.checkpointTimeoutMs + ); + }), + ]); + // `PRAGMA wal_checkpoint` returns {busy, log, checkpointed}; busy>0 means a + // reader blocked the truncate and the WAL was only partially checkpointed. + const result = Array.isArray(rows) ? rows[0] : undefined; + safeLog("Turso WAL checkpoint before close", { + busy: result?.busy, + log: result?.log, + checkpointed: result?.checkpointed, + }); + } catch (error) { + safeLog("Turso WAL checkpoint before close failed", { error: String(error) }); + } finally { + if (timer !== undefined) clearTimeout(timer); + // Always drain the pragma before close(), including after a timeout win: + // a late rejection must not escape unhandled, and the native close must + // not race an in-flight statement on the same handle. + await checkpoint.catch(() => {}); + } + } + async close(): Promise { + await this.checkpointBeforeClose(); await this.database.close(); } + + /** + * Close without the WAL checkpoint, for the synchronous `exit` handler where + * async continuations never run. The turso async `close()` calls the native + * close in its first statement with nothing awaited before it, so an + * un-awaited call really does release the handle — but it leaves the `-wal` + * in place, which is unavoidable without an event loop. + */ + closeSync(): void { + // Errors surface as a rejected promise (never a throw) and are ignored: + // the process is exiting and there is no budget left to act on them. + void this.database.close().catch(() => {}); + } } diff --git a/tests/turso-close-checkpoint.test.ts b/tests/turso-close-checkpoint.test.ts new file mode 100644 index 00000000..f58fa5b8 --- /dev/null +++ b/tests/turso-close-checkpoint.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { Database } from "@tursodatabase/database"; +import { CONFIG } from "../src/config.js"; +import { TursoConnectionManager } from "../src/services/turso/connection-manager.js"; +import { TursoDb } from "../src/services/turso/turso-db.js"; + +function fakeDatabase(all: (sql: string) => Promise) { + const calls: string[] = []; + const database = { + calls, + async all(sql: string) { + calls.push(sql); + return all(sql); + }, + async run(sql: string) { + calls.push(sql); + return { changes: 0, lastInsertRowid: 0 }; + }, + async get() { + return undefined; + }, + async batch() { + return []; + }, + async close() { + calls.push("close"); + }, + }; + return database as unknown as Database & { calls: string[] }; +} + +describe("turso close checkpoint", () => { + it("checkpoints the WAL before closing", async () => { + const database = fakeDatabase(async () => []); + await new TursoDb(database).close(); + + expect(database.calls).toEqual(["PRAGMA wal_checkpoint(TRUNCATE)", "close"]); + }); + + it("still closes when the checkpoint fails", async () => { + const database = fakeDatabase(async () => { + throw new Error("database is locked"); + }); + await new TursoDb(database).close(); + + expect(database.calls).toEqual(["PRAGMA wal_checkpoint(TRUNCATE)", "close"]); + }); + + it("does not leak the checkpoint rejection that lands after the timeout", async () => { + // Bun's test runner fails any test that leaks an unhandled rejection, so a + // regression here surfaces as a suite failure rather than a silent hang. + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandled); + try { + const calls: string[] = []; + const database = { + async all() { + calls.push("all"); + await new Promise((resolve) => setTimeout(resolve, 80)); + throw new Error("database is locked"); + }, + async close() { + calls.push("close"); + }, + } as unknown as Database; + + const db = new TursoDb(database, 10); + await db.close(); + + expect(calls).toEqual(["all", "close"]); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } + }); + + it("waits for a timed-out checkpoint to settle before closing", async () => { + // turso's close() does not take execLock — closing under an in-flight + // pragma would race the native handle. After the timeout fires we must + // still drain the checkpoint before close runs. + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const calls: string[] = []; + const database = { + async all() { + calls.push("all"); + await gate; + return [{ busy: 0, log: 0, checkpointed: 0 }]; + }, + async close() { + calls.push("close"); + }, + } as unknown as Database; + + const closing = new TursoDb(database, 10).close(); + await new Promise((resolve) => setTimeout(resolve, 40)); + expect(calls).toEqual(["all"]); + + release(); + await closing; + expect(calls).toEqual(["all", "close"]); + }); + + it("still closes when logging fails", async () => { + // Point the log file inside a regular file so the logger's mkdir/write + // throws (ENOTDIR) on both the success and failure log calls. + const sandbox = mkdtempSync(join(tmpdir(), "turso-log-fail-")); + const previousLogFile = process.env.OPENCODE_MEM_LOG_FILE; + writeFileSync(join(sandbox, "blocker"), "x"); + process.env.OPENCODE_MEM_LOG_FILE = join(sandbox, "blocker", "opencode-mem.log"); + try { + const database = fakeDatabase(async () => [{ busy: 0, log: 0, checkpointed: 0 }]); + await new TursoDb(database).close(); + + expect(database.calls).toEqual(["PRAGMA wal_checkpoint(TRUNCATE)", "close"]); + } finally { + if (previousLogFile === undefined) { + delete process.env.OPENCODE_MEM_LOG_FILE; + } else { + process.env.OPENCODE_MEM_LOG_FILE = previousLogFile; + } + rmSync(sandbox, { recursive: true, force: true }); + } + }); + + it("closeSync closes without running the checkpoint", () => { + const database = fakeDatabase(async () => []); + new TursoDb(database).closeSync(); + + // No await: the native close still happened on this tick. + expect(database.calls).toEqual(["close"]); + }); + + it("closeAllSync releases the native handle on the exit path", async () => { + const previousStoragePath = CONFIG.storagePath; + const sandbox = mkdtempSync(join(tmpdir(), "turso-sync-close-")); + CONFIG.storagePath = sandbox; + try { + const manager = new TursoConnectionManager(); + const db = await manager.getConnection(join(sandbox, "sync.db")); + await db.run("CREATE TABLE probe (id INTEGER PRIMARY KEY)"); + const client = db.getClient(); + expect(client.open).toBeTrue(); + + manager.closeAllSync(); + + // Same tick, before any await: the async turso close() runs the native + // close before its first (and only) statement can yield. + expect(client.open).toBeFalse(); + } finally { + CONFIG.storagePath = previousStoragePath; + rmSync(sandbox, { recursive: true, force: true }); + } + }); +});