fix(cron): close sqlite connections deterministically in execution ledger

cron/executions.py

@@ -12,8 +12,9 @@ import os
 import sqlite3
 import threading
 import uuid
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterator, List, Optional
 
 from hermes_constants import get_hermes_home
 from hermes_time import now as _hermes_now
@@ -26,10 +27,13 @@ _PROCESS_ID = uuid.uuid4().hex
 
 
 def _connect() -> sqlite3.Connection:
+    EXECUTIONS_FILE.parent.mkdir(parents=True, exist_ok=True)
+    return sqlite3.connect(EXECUTIONS_FILE, timeout=5)
+
+
+def _initialize_schema(conn: sqlite3.Connection) -> None:
     from hermes_state import apply_wal_with_fallback
 
-    EXECUTIONS_FILE.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(EXECUTIONS_FILE, timeout=5)
     conn.row_factory = sqlite3.Row
     conn.execute("PRAGMA busy_timeout=5000")
     apply_wal_with_fallback(conn, db_label="cron/executions.db")
@@ -58,7 +62,27 @@ def _connect() -> sqlite3.Connection:
         "CREATE INDEX IF NOT EXISTS idx_executions_status_claimed "
         "ON executions(status, claimed_at DESC, id DESC)"
     )
-    return conn
+
+
+@contextmanager
+def _transaction() -> Iterator[sqlite3.Connection]:
+    """Open a connection, commit/rollback on exit, always close.
+
+    ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back
+    the transaction; it does not close the connection. Relying on that alone
+    leaks a connection (and its WAL/SHM file descriptors) on every call,
+    since closing then depends on the garbage collector. Schema init runs
+    inside the ``try`` too, so a PRAGMA/DDL failure after a successful
+    ``connect()`` still closes the connection instead of leaking it.
+    """
+    with _lock:
+        conn = _connect()
+        try:
+            _initialize_schema(conn)
+            with conn:
+                yield conn
+        finally:
+            conn.close()
 
 
 def _record(row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]:
@@ -103,7 +127,7 @@ def create_execution(job_id: str, *, source: str) -> Dict[str, Any]:
     now = _hermes_now().isoformat()
     execution_id = uuid.uuid4().hex
     pid = os.getpid()
-    with _lock, _connect() as conn:
+    with _transaction() as conn:
         conn.execute(
             """INSERT INTO executions
                (id, job_id, source, process_id, pid, process_started_at,
@@ -121,7 +145,7 @@ def create_execution(job_id: str, *, source: str) -> Dict[str, Any]:
 def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]:
     """Transition one claimed attempt to running exactly once."""
     now = _hermes_now().isoformat()
-    with _lock, _connect() as conn:
+    with _transaction() as conn:
         cur = conn.execute(
             """UPDATE executions SET status='running', started_at=?
                WHERE id=? AND status='claimed'""",
@@ -141,7 +165,7 @@ def finish_execution(
     now = _hermes_now().isoformat()
     status = "completed" if success else "failed"
     detail = None if success else (str(error) if error else "unknown failure")
-    with _lock, _connect() as conn:
+    with _transaction() as conn:
         cur = conn.execute(
             """UPDATE executions SET status=?, finished_at=?, error=?
                WHERE id=? AND status IN ('claimed','running')""",
@@ -159,7 +183,7 @@ def recover_interrupted_executions() -> int:
     """Mark provably abandoned attempts unknown without scheduling retries."""
     now = _hermes_now().isoformat()
     changed = 0
-    with _lock, _connect() as conn:
+    with _transaction() as conn:
         rows = conn.execute(
             """SELECT id, process_id, pid, process_started_at FROM executions
                WHERE status IN ('claimed','running')"""
@@ -198,7 +222,7 @@ def list_executions(
         params.append(str(before_claimed_at))
     where = " WHERE " + " AND ".join(clauses) if clauses else ""
     params.append(max(1, min(int(limit), 500)))
-    with _lock, _connect() as conn:
+    with _transaction() as conn:
         rows = conn.execute(
             "SELECT * FROM executions" + where
             + " ORDER BY claimed_at DESC, id DESC LIMIT ?",
@@ -218,7 +242,7 @@ def latest_executions(job_ids: List[str]) -> Dict[str, Dict[str, Any]]:
     if not clean:
         return {}
     placeholders = ",".join("?" for _ in clean)
-    with _lock, _connect() as conn:
+    with _transaction() as conn:
         rows = conn.execute(
             f"""SELECT e.* FROM executions e
                 WHERE e.job_id IN ({placeholders})

tests/cron/test_execution_ledger.py

@@ -304,6 +304,126 @@ def test_external_provider_start_recovers_interrupted_records(monkeypatch):
     assert events == ["recover", "reconcile"]
 
 
+class _TrackingConnection:
+    """Delegates to a real sqlite3.Connection while recording close() calls.
+
+    sqlite3.Connection is a static C type: it has no per-instance __dict__
+    and its class methods can't be monkeypatched, so open/close tracking is
+    done via a delegating wrapper returned in place of the real connection.
+    """
+
+    def __init__(self, real, closed_ids):
+        object.__setattr__(self, "_real", real)
+        object.__setattr__(self, "_closed_ids", closed_ids)
+
+    def close(self):
+        self._closed_ids.append(id(self._real))
+        self._real.close()
+
+    def __enter__(self):
+        self._real.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return self._real.__exit__(exc_type, exc, tb)
+
+    def __getattr__(self, name):
+        return getattr(self._real, name)
+
+    def __setattr__(self, name, value):
+        setattr(self._real, name, value)
+
+
+def _count_open_connections(executions, monkeypatch):
+    """Wrap sqlite3.connect to track open/close balance for the ledger module."""
+    opened_ids = []
+    closed_ids = []
+    real_connect = sqlite3.connect
+
+    def tracking_connect(*args, **kwargs):
+        conn = real_connect(*args, **kwargs)
+        opened_ids.append(id(conn))
+        return _TrackingConnection(conn, closed_ids)
+
+    monkeypatch.setattr(executions.sqlite3, "connect", tracking_connect)
+    return opened_ids, closed_ids
+
+
+def test_ledger_operations_close_every_connection(monkeypatch, tmp_path):
+    """Regression for #69567: every ledger call must close its connection
+    deterministically instead of relying on garbage collection."""
+    executions = _point_ledger(monkeypatch, tmp_path)
+    opened, closed = _count_open_connections(executions, monkeypatch)
+
+    record = executions.create_execution("leak-check", source="builtin")
+    executions.mark_execution_running(record["id"])
+    executions.finish_execution(record["id"], success=True)
+    executions.list_executions(job_id="leak-check")
+    executions.latest_executions(["leak-check"])
+    executions.recover_interrupted_executions()
+
+    assert len(opened) == 6
+    assert len(closed) == 6
+    assert set(opened) == set(closed)
+
+
+def test_early_return_still_closes_connection(monkeypatch, tmp_path):
+    """mark_execution_running returns None mid-block on a bad transition;
+    the connection must still be closed rather than leaked."""
+    executions = _point_ledger(monkeypatch, tmp_path)
+    opened, closed = _count_open_connections(executions, monkeypatch)
+
+    assert executions.mark_execution_running("does-not-exist") is None
+
+    assert len(opened) == 1
+    assert len(closed) == 1
+
+
+def test_exception_during_operation_still_closes_connection(monkeypatch, tmp_path):
+    """A failing statement inside the transaction must roll back and close,
+    not leak the connection."""
+    executions = _point_ledger(monkeypatch, tmp_path)
+    opened, closed = _count_open_connections(executions, monkeypatch)
+
+    with __import__("pytest").raises(sqlite3.IntegrityError):
+        with executions._transaction() as conn:
+            conn.execute(
+                "INSERT INTO executions (id, job_id, source, process_id, pid, "
+                "status, claimed_at) VALUES ('x', 'x', 'x', 'x', 1, 'bogus-status', 'now')"
+            )
+
+    assert len(opened) == 1
+    assert len(closed) == 1
+
+
+def test_schema_init_failure_still_closes_connection(monkeypatch, tmp_path):
+    """If PRAGMA/DDL setup in _connect() fails after sqlite3.connect()
+    succeeds, the partially-initialized connection must still be closed."""
+    executions = _point_ledger(monkeypatch, tmp_path)
+    opened_ids = []
+    closed_ids = []
+    real_connect = sqlite3.connect
+
+    class _FailingSchemaConnection(_TrackingConnection):
+        def execute(self, sql, *args, **kwargs):
+            if "CREATE TABLE" in sql:
+                raise sqlite3.OperationalError("simulated schema init failure")
+            return self._real.execute(sql, *args, **kwargs)
+
+    def tracking_connect(*args, **kwargs):
+        conn = real_connect(*args, **kwargs)
+        opened_ids.append(id(conn))
+        return _FailingSchemaConnection(conn, closed_ids)
+
+    monkeypatch.setattr(executions.sqlite3, "connect", tracking_connect)
+
+    with __import__("pytest").raises(sqlite3.OperationalError):
+        executions.create_execution("init-fail", source="builtin")
+
+    assert len(opened_ids) == 1
+    assert len(closed_ids) == 1
+
+
 def test_job_listing_exposes_latest_execution(monkeypatch, tmp_path):
     import cron.jobs as jobs