Make DB writes rollback-safe and self-heal poisoned connections

nerve/db/audit.py

@@ -19,14 +19,12 @@ async def log_audit(
     ) -> int:
         """Write an entry to the memU audit log."""
         now = datetime.now(timezone.utc).isoformat()
-        async with self.db.execute(
+        result = await self._write(
             "INSERT INTO memu_audit_log (timestamp, action, target_type, target_id, source, details) "
             "VALUES (?, ?, ?, ?, ?, ?)",
             (now, action, target_type, target_id, source, json.dumps(details) if details else None),
-        ) as cursor:
-            log_id = cursor.lastrowid
-        await self.db.commit()
-        return log_id
+        )
+        return result.lastrowid
 
     async def get_audit_logs(
         self,

nerve/db/base.py

@@ -11,7 +11,7 @@
 import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
-from typing import AsyncIterator
+from typing import AsyncIterator, NamedTuple
 
 import aiosqlite
 
@@ -39,6 +39,18 @@
 SCHEMA_VERSION = max(v for v, _ in discover_migrations()) if discover_migrations() else 0
 
 
+class WriteResult(NamedTuple):
+    """Outcome of a single-statement write executed via :meth:`Database._write`.
+
+    Mirrors the two cursor attributes writer call sites actually use. The
+    live cursor is closed before the commit, so callers can never fetch from
+    a cursor whose transaction has already ended.
+    """
+
+    lastrowid: int | None
+    rowcount: int
+
+
 # Connection pragmas applied on every ``connect()``. These mirror the tuning
 # memU already uses for its own SQLite connections (see
 # ``nerve/memory/memu_bridge.py``) — the primary operational DB is where the
@@ -141,16 +153,98 @@ def db(self) -> aiosqlite.Connection:
             raise RuntimeError("Database not connected. Call connect() first.")
         return self._db
 
+    # -- Write path ---------------------------------------------------------
+    #
+    # All writes on the single shared connection MUST go through ``_atomic()``
+    # (multi-statement transactions) or ``_write()`` (single statements).
+    # Both serialize under ``_write_lock`` and guarantee the connection is
+    # never left inside an open transaction — on success they COMMIT, on any
+    # error (including ``asyncio.CancelledError``) they ROLLBACK.
+    #
+    # Why this is load-bearing (production outage, 2026-07-06): a write task
+    # was abandoned mid-transaction, leaving the shared connection inside an
+    # open transaction pinned to a WAL read snapshot. A second process then
+    # committed to the same DB file, making that snapshot stale. From that
+    # moment every write on the shared connection failed *instantly* with
+    # "database is locked" (SQLITE_BUSY_SNAPSHOT — the busy handler is
+    # deliberately not invoked for snapshot conflicts, so ``busy_timeout``
+    # does not apply), reads silently served the frozen snapshot, and nothing
+    # ever called ROLLBACK — wedging the daemon for 10 hours until a restart.
+    # ``_heal_leaked_txn()`` is the belt-and-suspenders guard that recovers
+    # from that state even if some future code path leaks a transaction.
+
     @asynccontextmanager
     async def _atomic(self) -> AsyncIterator[None]:
-        """Acquire write lock for multi-statement transactions.
+        """Serialize a multi-statement write and make it a real transaction.
+
+        Once a coroutine begins a multi-statement write, no other coroutine
+        can interleave writes before the commit. The body's statements are
+        committed on success and rolled back on any exception — including
+        task cancellation — so a failed body neither half-commits nor leaves
+        the shared connection inside an open (poisoned) transaction.
+
+        Statements inside the body must use ``self.db.execute(...)`` directly
+        (never ``_write()``, which would deadlock on the non-reentrant lock).
+        """
+        async with self._write_lock:
+            await self._heal_leaked_txn()
+            try:
+                yield
+                # Shield so a cancellation arriving mid-commit cannot abandon
+                # a half-finished transaction: the inner task runs to
+                # completion on aiosqlite's worker thread regardless.
+                await asyncio.shield(self.db.commit())
+            except BaseException:
+                await self._rollback_quietly()
+                raise
 
-        Ensures that once a coroutine begins a multi-statement write,
-        no other coroutine can interleave writes before the commit.
+    async def _write(self, sql: str, params: tuple | list = ()) -> WriteResult:
+        """Execute one write statement and commit, under the write lock.
+
+        The single-statement counterpart of :meth:`_atomic` with the same
+        guarantees: serialized against all other writers (so its commit can
+        never flush someone else's in-flight transaction) and commit-or-
+        rollback semantics (so an error or cancellation can never leave the
+        connection mid-transaction).
         """
         async with self._write_lock:
-            yield
-            await self.db.commit()
+            await self._heal_leaked_txn()
+            try:
+                cursor = await self.db.execute(sql, params)
+                result = WriteResult(cursor.lastrowid, cursor.rowcount)
+                await cursor.close()
+                await asyncio.shield(self.db.commit())
+                return result
+            except BaseException:
+                await self._rollback_quietly()
+                raise
+
+    async def _heal_leaked_txn(self) -> None:
+        """Roll back a leaked open transaction on the shared connection.
+
+        Called under ``_write_lock``. Every legitimate transaction commits or
+        rolls back before releasing the lock, so ``in_transaction`` being true
+        here means some code path abandoned a transaction (see the write-path
+        comment above for the outage this causes). Recover loudly.
+        """
+        if self.db.in_transaction:
+            logger.error(
+                "Leaked open transaction detected on the shared connection — "
+                "rolling back to prevent a wedged write path "
+                "(SQLITE_BUSY_SNAPSHOT poisoning)",
+            )
+            await self._rollback_quietly()
+
+    async def _rollback_quietly(self) -> None:
+        """Best-effort ROLLBACK that never raises and survives cancellation."""
+        try:
+            await asyncio.shield(self.db.rollback())
+        except asyncio.CancelledError:
+            # The shielded rollback still runs to completion on the worker
+            # thread; re-raise so the caller's cancellation proceeds.
+            raise
+        except Exception:
+            logger.exception("Rollback failed on the shared connection")
 
     async def _check_fts_integrity(self) -> None:
         """FTS integrity check — runs every startup.
@@ -167,27 +261,29 @@ async def _check_fts_integrity(self) -> None:
                 "FTS index mismatch: %d tasks vs %d FTS entries — reseeding",
                 task_count, fts_count,
             )
-            await self.db.execute("DELETE FROM tasks_fts")
             # Read content from disk files (source of truth) instead of seeding
             # empty. Task file_path values are relative to the workspace root,
             # which is NOT the DB directory (~/.nerve) — fall back to it only
             # when no workspace was provided.
             workspace = (self.workspace or self.db_path.parent).expanduser()
-            async with self.db.execute("SELECT id, title, file_path FROM tasks") as cur:
-                rows = await cur.fetchall()
-            for row in rows:
-                content = ""
-                try:
-                    fp = workspace / row["file_path"]
-                    if fp.exists():
-                        content = await asyncio.to_thread(
-                            fp.read_text, encoding="utf-8",
-                        )
-                except Exception as e:
-                    logger.warning("Failed to read %s for FTS reseed: %s", row["file_path"], e)
-                await self.db.execute(
-                    "INSERT INTO tasks_fts (task_id, title, content) VALUES (?, ?, ?)",
-                    (row["id"], row["title"], content),
-                )
-            await self.db.commit()
+            async with self._atomic():
+                await self.db.execute("DELETE FROM tasks_fts")
+                async with self.db.execute(
+                    "SELECT id, title, file_path FROM tasks",
+                ) as cur:
+                    rows = await cur.fetchall()
+                for row in rows:
+                    content = ""
+                    try:
+                        fp = workspace / row["file_path"]
+                        if fp.exists():
+                            content = await asyncio.to_thread(
+                                fp.read_text, encoding="utf-8",
+                            )
+                    except Exception as e:
+                        logger.warning("Failed to read %s for FTS reseed: %s", row["file_path"], e)
+                    await self.db.execute(
+                        "INSERT INTO tasks_fts (task_id, title, content) VALUES (?, ?, ?)",
+                        (row["id"], row["title"], content),
+                    )
             logger.info("FTS reseeded with %d tasks (content from disk)", task_count)

nerve/db/cron.py

@@ -9,24 +9,21 @@ class CronStore:
     """Mixin providing cron job logging operations."""
 
     async def log_cron_start(self, job_id: str) -> int:
-        async with self.db.execute(
+        result = await self._write(
             "INSERT INTO cron_logs (job_id) VALUES (?)", (job_id,)
-        ) as cursor:
-            log_id = cursor.lastrowid
-        await self.db.commit()
-        return log_id
+        )
+        return result.lastrowid
 
     async def set_cron_log_session(self, log_id: int, session_id: str) -> None:
         """Link a run log to its session while the run is still in flight.
 
         Written at run start so the UI can open the chat of a running cron;
         log_cron_finish keeps it as a fallback for older code paths.
         """
-        await self.db.execute(
+        await self._write(
             "UPDATE cron_logs SET session_id = ? WHERE id = ?",
             (session_id, log_id),
         )
-        await self.db.commit()
 
     async def log_cron_finish(
         self,
@@ -37,12 +34,11 @@ async def log_cron_finish(
         session_id: str | None = None,
     ) -> None:
         now = utc_now_iso()
-        await self.db.execute(
+        await self._write(
             "UPDATE cron_logs SET finished_at = ?, status = ?, output = ?, error = ?, "
             "session_id = COALESCE(?, session_id) WHERE id = ?",
             (now, status, output, error, session_id, log_id),
         )
-        await self.db.commit()
 
     async def get_cron_logs(
         self, job_id: str | None = None, limit: int = 50, offset: int = 0,
@@ -122,11 +118,8 @@ async def cleanup_old_cron_logs(self, days: int = 14) -> int:
         thousands of rows and the unfiltered "latest N" query (used by the
         diagnostics endpoint) degrades to a full-table scan + memory sort.
         """
-        cursor = await self.db.execute(
+        result = await self._write(
             "DELETE FROM cron_logs WHERE started_at < datetime('now', ? || ' days')",
             (f"-{days}",),
         )
-        deleted = cursor.rowcount or 0
-        await cursor.close()
-        await self.db.commit()
-        return deleted
+        return result.rowcount or 0

nerve/db/files.py

@@ -20,12 +20,11 @@ async def save_uploaded_file(
         disk_path: str,
     ) -> None:
         now = datetime.now(timezone.utc).isoformat()
-        await self.db.execute(
+        await self._write(
             """INSERT INTO uploaded_files (id, session_id, filename, media_type, file_type, file_size, disk_path, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
             (file_id, session_id, filename, media_type, file_type, file_size, disk_path, now),
         )
-        await self.db.commit()
 
     async def get_uploaded_file(self, file_id: str) -> dict | None:
         async with self.db.execute(
@@ -51,9 +50,8 @@ async def delete_uploaded_files(self, session_id: str) -> list[str]:
             (session_id,),
         ) as cursor:
             paths = [row[0] async for row in cursor]
-        await self.db.execute(
+        await self._write(
             "DELETE FROM uploaded_files WHERE session_id = ?",
             (session_id,),
         )
-        await self.db.commit()
         return paths

nerve/db/maintenance.py

@@ -165,20 +165,24 @@ async def checkpoint(self) -> None:
         """Truncate the WAL after a prune pass (best-effort).
 
         Frees the WAL file; does not shrink the main DB (see :meth:`vacuum`).
+        Runs under the write lock: an unlocked bare ``commit()`` here could
+        prematurely flush another coroutine's in-flight transaction.
         """
-        await self.db.commit()
-        await self.db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+        async with self._write_lock:
+            await self._heal_leaked_txn()
+            await self.db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
 
     async def vacuum(self) -> None:
         """Rewrite the DB file to reclaim freelist pages (shrinks the file).
 
-        Takes a write lock and cannot run inside a transaction, so it is an
-        explicit operator step, never on the background loop. Run with the
-        daemon stopped to avoid lock contention.
+        Cannot run inside a transaction (``_heal_leaked_txn`` guarantees the
+        connection is clean) and autocommits on completion. Serialized under
+        the write lock; still an explicit operator step, never on the
+        background loop. Run with the daemon stopped to avoid lock contention.
         """
-        await self.db.commit()
-        await self.db.execute("VACUUM")
-        await self.db.commit()
+        async with self._write_lock:
+            await self._heal_leaked_txn()
+            await self.db.execute("VACUUM")
 
     async def run_retention(
         self,

nerve/db/mcp.py

@@ -17,7 +17,7 @@ async def upsert_mcp_server(
     ) -> None:
         """Register or update an MCP server entry."""
         now = datetime.now(timezone.utc).isoformat()
-        await self.db.execute(
+        await self._write(
             """INSERT INTO mcp_servers (name, type, enabled, tool_count, first_seen_at, last_seen_at)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(name) DO UPDATE SET
@@ -29,7 +29,6 @@ async def upsert_mcp_server(
                    last_seen_at = excluded.last_seen_at""",
             (name, server_type, enabled, tool_count, now, now),
         )
-        await self.db.commit()
 
     async def record_mcp_tool_usage(
         self,
@@ -41,13 +40,12 @@ async def record_mcp_tool_usage(
         error: str | None = None,
     ) -> None:
         """Log an MCP tool invocation."""
-        await self.db.execute(
+        await self._write(
             """INSERT INTO mcp_tool_usage
                (server_name, tool_name, session_id, duration_ms, success, error)
                VALUES (?, ?, ?, ?, ?, ?)""",
             (server_name, tool_name, session_id, duration_ms, success, error),
         )
-        await self.db.commit()
 
     async def get_mcp_server_stats(self) -> list[dict]:
         """List all MCP servers with aggregated usage stats."""

nerve/db/messages.py

@@ -160,11 +160,10 @@ async def merge_tool_result_into_call(
                 break
         if not updated:
             return None
-        await self.db.execute(
+        await self._write(
             "UPDATE messages SET blocks = ? WHERE id = ?",
             (json.dumps(blocks), msg_id),
         )
-        await self.db.commit()
         return msg_id
 
     async def merge_workflow_into_call(
@@ -211,11 +210,10 @@ async def merge_workflow_into_call(
                     updated = True
                     break
             if updated:
-                await self.db.execute(
+                await self._write(
                     "UPDATE messages SET blocks = ? WHERE id = ?",
                     (json.dumps(blocks), row["id"]),
                 )
-                await self.db.commit()
                 return row["id"]
         return None
 
@@ -242,13 +240,12 @@ async def save_file_snapshot(
         Uses INSERT OR IGNORE so only the first touch per session+file is stored.
         """
         now = datetime.now(timezone.utc).isoformat()
-        await self.db.execute(
+        await self._write(
             """INSERT OR IGNORE INTO session_file_snapshots
                (session_id, file_path, original_content, created_at)
                VALUES (?, ?, ?, ?)""",
             (session_id, file_path, content, now),
         )
-        await self.db.commit()
 
     async def get_file_snapshot(
         self, session_id: str, file_path: str,
@@ -272,11 +269,10 @@ async def get_session_snapshots(self, session_id: str) -> list[dict]:
 
     async def delete_session_snapshots(self, session_id: str) -> None:
         """Delete all file snapshots for a session."""
-        await self.db.execute(
+        await self._write(
             "DELETE FROM session_file_snapshots WHERE session_id = ?",
             (session_id,),
         )
-        await self.db.commit()
 
     async def count_messages(self, session_id: str) -> int:
         async with self.db.execute(