fix(turso): checkpoint the WAL before closing a database

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();
 

src/services/turso/connection-manager.ts

@@ -3,7 +3,7 @@
 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";
@@ -140,7 +140,7 @@
             // ignore close errors during cleanup
           }
         }
         if (encryption) {
           const message = error instanceof Error ? error.message : String(error);
           if (isTursoMultiProcessLockError(error)) {
             throw wrapTursoOpenError(dbPath, error);
@@ -148,7 +148,7 @@
           throw new Error(
             `Failed to open encrypted database ${dbPath}: ${message}. ` +
               `Check databaseEncryptionKey / cipher, or remove encryption config for plaintext shards.`,
             { cause: error }
           );
         }
         throw wrapTursoOpenError(dbPath, error);
@@ -156,7 +156,7 @@
     })();
 
     this.pending.set(dbPath, openPromise);
 
     try {
       return await openPromise;
     } catch (error) {
@@ -166,14 +166,14 @@
       this.pending.delete(dbPath);
     }
   }
 
   async closeConnection(dbPath: string): Promise<void> {
     if (this.closingPromise) {
       await this.closingPromise;
     }
     const existingClose = this.closingConnections.get(dbPath);
     if (existingClose) {
       return existingClose;
     }
 
     const closePromise = Promise.resolve()
@@ -187,8 +187,8 @@
         if (db) {
           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);
@@ -204,18 +204,22 @@
     return closePromise;
   }
 
   async closeAll(): Promise<void> {
     if (this.closingPromise) return this.closingPromise;
 
     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();
@@ -227,13 +231,13 @@
       this.closingPromise = null;
     }
   }
 
   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();
@@ -242,3 +246,3 @@
 }
 
 export const tursoConnectionManager = new TursoConnectionManager();

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<string, SqlValue>;
@@ -112,8 +113,18 @@ function bindArgs(args?: SqlArgs): SqlValue[] | [Record<string, SqlValue>] {
   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<void> {
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const checkpoint = this.database.all("PRAGMA wal_checkpoint(TRUNCATE)");
+    try {
+      const rows = await Promise.race([
+        checkpoint,
+        new Promise<never>((_, 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<void> {
+    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(() => {});
+  }
 }

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<unknown[]>) {
+  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<void>((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 });
+    }
+  });
+});