From 1d721a66f7fc63e5e05a9cd8412dce3fc366db5a Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Wed, 22 Jul 2026 17:21:03 -0300 Subject: [PATCH] fix(cron): close sqlite connections deterministically in execution ledger --- cron/executions.py | 44 +++++++--- tests/cron/test_execution_ledger.py | 120 ++++++++++++++++++++++++++++ 2 files changed, 154 insertions(+), 10 deletions(-) diff --git a/cron/executions.py b/cron/executions.py index 6ffe1eef1fc8..fac24357335c 100644 --- a/cron/executions.py +++ b/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}) diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 7d032b06a857..0be6aa63ec8f 100644 --- a/tests/cron/test_execution_ledger.py +++ b/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