Close SQLite connections so PHC acquisition cannot stall silently

CHANGELOG.md

@@ -4,6 +4,18 @@ All notable changes will be documented in this file.
 
 ## Unreleased
 
+- Fixed a SQLite descriptor leak that silently stopped raw PHC acquisition a
+  few minutes after the collector started. A `sqlite3.Connection` used directly
+  as a context manager only manages the transaction, so every recorded sample
+  leaked a descriptor pair across both the PHC ring store and the experiment
+  recorder. Once the limit was reached, `/dev/ptp*` could no longer be opened,
+  `take_phc_sample()` returned nothing, and the loop no-opped indefinitely while
+  systemd still reported the service as running: the Observatory showed
+  `0.0 Hz acquired` and "Waiting for direct PHC samples". Connections are now
+  owned by a context manager that commits or rolls back and always closes, and
+  the collector exits for a supervisor restart after a sustained run of cycles
+  that deliver nothing, so an unrecoverable acquisition stall self-heals instead
+  of persisting silently.
 - Upgraded PI autotuning to a replay-only safe Bayesian optimizer with a
   global plus log-local candidate set, settling/overshoot constraints,
   identified ARX stability penalties, optional H-infinity sensitivity

agent/ptpbox_phc_collector.py

@@ -14,17 +14,50 @@
 
 def collect(stop: threading.Event) -> None:
     deadline = time.monotonic()
+    consecutive_failures = 0
+    delivered_any = False
+    reported_idle = False
     while not stop.is_set():
         requested_rate = agent.configured_phc_sample_rate_hz()
         period = 1.0 / requested_rate
+        delivered = False
         sample = agent.take_phc_sample()
         if sample is not None:
             temperatures = agent.clock_temperatures()
             try:
                 append_sample(agent.PHC_STORE_FILE, sample, temperatures)
                 agent.experiment_store().record_phc(sample, temperatures)
+                delivered = True
             except (OSError, sqlite3.Error) as exc:
                 print(f"PTPBox PHC store write failed: {exc}", flush=True)
+
+        if delivered:
+            if reported_idle:
+                print("PTPBox PHC acquisition resumed", flush=True)
+            delivered_any = True
+            consecutive_failures = 0
+            reported_idle = False
+        else:
+            consecutive_failures += 1
+            if not reported_idle:
+                # Report the transition once rather than every cycle: before the
+                # cascade is started there is legitimately nothing to sample.
+                print("PTPBox PHC samples unavailable; waiting", flush=True)
+                reported_idle = True
+
+        # Exiting is only justified once acquisition has actually worked and then
+        # stopped, which indicates a condition this loop cannot repair (exhausted
+        # descriptors, a revoked store path, a PHC that disappeared).  A process
+        # that has never acquired is simply waiting for the cascade to start, so
+        # it must keep waiting instead of forcing a supervisor restart loop.
+        if delivered_any and consecutive_failures >= max(10, int(round(requested_rate * 10))):
+            print(
+                f"PTPBox PHC collector stopped delivering for {consecutive_failures} "
+                "consecutive cycles after working; exiting for supervisor restart",
+                flush=True,
+            )
+            raise SystemExit(1)
+
         deadline += period
         now = time.monotonic()
         if deadline < now - period:

agent/ptpbox_phc_store.py

@@ -3,9 +3,11 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
 import sqlite3
 import time
+from collections.abc import Iterator
 from pathlib import Path
 from typing import Any
 
@@ -34,6 +36,23 @@ def _connect(path: Path) -> sqlite3.Connection:
     return connection
 
 
+@contextlib.contextmanager
+def _session(path: Path) -> Iterator[sqlite3.Connection]:
+    """Own the connection: commit or roll back, then always close it.
+
+    A ``sqlite3.Connection`` used directly as a context manager only manages the
+    transaction; it does not close the handle. Depending on that leaked one file
+    descriptor pair per call, which exhausted the collector's descriptor limit
+    after a few minutes at the Sync cadence and silently stopped acquisition.
+    """
+    connection = _connect(path)
+    try:
+        with connection:
+            yield connection
+    finally:
+        connection.close()
+
+
 def append_sample(
     path: Path,
     sample: dict[str, Any],
@@ -48,7 +67,7 @@ def append_sample(
         {"sample": sample, "temperatures": temperatures or {}},
         separators=(",", ":"),
     )
-    with _connect(path) as connection:
+    with _session(path) as connection:
         connection.execute(
             "INSERT OR REPLACE INTO phc_samples(observed_at, sample_id, payload) VALUES(?, ?, ?)",
             (observed_at, str(sample["sample_id"]), payload),
@@ -79,7 +98,7 @@ def read_records(
     cutoff = time.time() - max(5.0, min(float(history_seconds), MAX_AGE_SECONDS))
     lower_bound = max(cutoff, float(since)) if since is not None else cutoff
     try:
-        with _connect(path) as connection:
+        with _session(path) as connection:
             rows = connection.execute(
                 "SELECT payload FROM phc_samples WHERE observed_at > ? ORDER BY observed_at",
                 (lower_bound,),

agent/ptpbox_research.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import csv
 import io
 import json
@@ -19,6 +20,7 @@
 import threading
 import time
 from collections import deque
+from collections.abc import Iterator
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Iterable, Sequence
@@ -3802,9 +3804,25 @@ def _connect(self) -> sqlite3.Connection:
         connection.execute("PRAGMA synchronous=NORMAL")
         return connection
 
+    @contextlib.contextmanager
+    def _session(self) -> Iterator[sqlite3.Connection]:
+        """Own the connection: commit or roll back, then always close it.
+
+        A ``sqlite3.Connection`` used directly as a context manager only manages
+        the transaction; it does not close the handle. Relying on that leaked a
+        descriptor pair per recorded sample, so a long-lived writer such as the
+        PHC collector eventually hit its descriptor limit.
+        """
+        connection = self._connect()
+        try:
+            with connection:
+                yield connection
+        finally:
+            connection.close()
+
     def _initialize(self) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        with self._connect() as connection:
+        with self._session() as connection:
             connection.executescript(
                 """
                 CREATE TABLE IF NOT EXISTS experiments (
@@ -3846,7 +3864,7 @@ def _initialize(self) -> None:
             )
 
     def active(self) -> dict[str, Any] | None:
-        with self._connect() as connection:
+        with self._session() as connection:
             row = connection.execute(
                 "SELECT * FROM experiments WHERE state='running' ORDER BY started_at DESC LIMIT 1"
             ).fetchone()
@@ -3858,7 +3876,7 @@ def start(self, payload: dict[str, Any], config: dict[str, Any]) -> dict[str, An
         name = str(payload.get("name") or f"{str(payload.get('kind') or payload.get('type') or 'capture').title()} {identifier[-6:]}")
         kind = str(payload.get("kind") or payload.get("type") or "capture")
         metadata = {key: value for key, value in payload.items() if key not in {"name", "kind", "type"}}
-        with self._lock, self._connect() as connection:
+        with self._lock, self._session() as connection:
             connection.execute(
                 "UPDATE experiments SET state='completed', stopped_at=? WHERE state='running'",
                 (now,),
@@ -3874,15 +3892,15 @@ def stop(self, identifier: str | None = None) -> dict[str, Any] | None:
         if not active:
             return None
         now = time.time()
-        with self._lock, self._connect() as connection:
+        with self._lock, self._session() as connection:
             connection.execute(
                 "UPDATE experiments SET state='completed', stopped_at=? WHERE id=? AND state='running'",
                 (now, active["id"]),
             )
         return self.get(str(active["id"]))
 
     def get(self, identifier: str) -> dict[str, Any] | None:
-        with self._connect() as connection:
+        with self._session() as connection:
             row = connection.execute(
                 """
                 SELECT e.*,
@@ -3900,7 +3918,7 @@ def get(self, identifier: str) -> dict[str, Any] | None:
         return item
 
     def list(self, limit: int = 30) -> list[dict[str, Any]]:
-        with self._connect() as connection:
+        with self._session() as connection:
             rows = connection.execute(
                 "SELECT id FROM experiments ORDER BY started_at DESC LIMIT ?",
                 (max(1, min(200, limit)),),
@@ -3930,7 +3948,7 @@ def record_phc(self, sample: dict[str, Any], temperatures: dict[str, float] | No
             )
             for clock in sample.get("clocks", [])
         ]
-        with self._lock, self._connect() as connection:
+        with self._lock, self._session() as connection:
             connection.executemany(
                 "INSERT INTO samples VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                 rows,
@@ -3943,7 +3961,7 @@ def phc_samples(self, identifier: str, since: float | None = None) -> list[dict[
         if since is not None:
             predicate = " AND observed_at>=?"
             parameters.append(float(since))
-        with self._connect() as connection:
+        with self._session() as connection:
             rows = connection.execute(
                 f"""
                 SELECT observed_at, cycle_id, clock_id, offset_ns, hop_offset_ns,
@@ -3966,7 +3984,7 @@ def phc_holdover_summary(
             return []
         placeholders = ",".join("?" for _clock in clock_ids)
         parameters: list[Any] = [identifier, float(since), *clock_ids]
-        with self._connect() as connection:
+        with self._session() as connection:
             rows = connection.execute(
                 f"""
                 SELECT clock_id,
@@ -4028,7 +4046,7 @@ def phc_holdover_series(
         """Return uniformly decimated raw cycles while retaining the final cycle."""
         if not clock_ids:
             return [], 0, 1
-        with self._connect() as connection:
+        with self._session() as connection:
             cycle_count = int(
                 connection.execute(
                     """
@@ -4071,7 +4089,7 @@ def phc_holdover_series(
 
     def event(self, category: str, severity: str, message: str, payload: dict[str, Any] | None = None) -> None:
         active = self.active()
-        with self._lock, self._connect() as connection:
+        with self._lock, self._session() as connection:
             connection.execute(
                 "INSERT INTO events VALUES (?, ?, ?, ?, ?, ?)",
                 (
@@ -4103,7 +4121,7 @@ def export_csv(self, identifier: str) -> str:
                 "valid",
             ]
         )
-        with self._connect() as connection:
+        with self._session() as connection:
             rows = connection.execute(
                 """
                 SELECT experiment_id, observed_at, cycle_id, clock_id, source,

tests/test_phc_store.py

@@ -1,4 +1,7 @@
+import gc
 import importlib.util
+import os
+import sys
 import tempfile
 import time
 import unittest
@@ -11,6 +14,24 @@
 STORE = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(STORE)
 
+RESEARCH_PATH = Path(__file__).parents[1] / "agent" / "ptpbox_research.py"
+RESEARCH_SPEC = importlib.util.spec_from_file_location("ptpbox_research_store_test", RESEARCH_PATH)
+assert RESEARCH_SPEC and RESEARCH_SPEC.loader
+RESEARCH = importlib.util.module_from_spec(RESEARCH_SPEC)
+sys.modules[RESEARCH_SPEC.name] = RESEARCH
+RESEARCH_SPEC.loader.exec_module(RESEARCH)
+
+
+def _open_descriptors() -> int:
+    # Collect first: a descriptor still reachable after a full collection is a
+    # genuine leak, whereas one pending finalisation is only GC timing and
+    # differs between platforms.
+    gc.collect()
+    for candidate in (f"/proc/{os.getpid()}/fd", "/dev/fd"):
+        if os.path.isdir(candidate):
+            return len(os.listdir(candidate))
+    raise unittest.SkipTest("no descriptor listing on this platform")
+
 
 class PhcStoreTests(unittest.TestCase):
     def test_round_trip_preserves_raw_samples_and_temperatures(self) -> None:
@@ -62,5 +83,79 @@ def test_quality_reports_achieved_rate_and_gaps(self) -> None:
             self.assertGreater(quality["achieved_rate_hz"], 2.0)
 
 
+class DescriptorLifetimeTests(unittest.TestCase):
+    """A long-lived writer must not accumulate SQLite descriptors.
+
+    A sqlite3.Connection used directly as a context manager only manages the
+    transaction, so the handle stayed open. At the Sync cadence that exhausted
+    the collector's descriptor limit within minutes, after which opening
+    /dev/ptp* also failed and acquisition stopped silently.
+    """
+
+    ITERATIONS = 400
+
+    @staticmethod
+    def _sample(index: int) -> dict:
+        observed_at = time.time() + index
+        return {
+            "observed_at": observed_at,
+            "sample_id": f"sample-{index}",
+            "clocks": [
+                {
+                    "id": "BC1",
+                    "observed_at": observed_at,
+                    "offset_ns": 0.0,
+                    "valid": True,
+                    "comparison_uncertainty_ns": 1.0,
+                },
+                {
+                    "id": "BC2",
+                    "observed_at": observed_at,
+                    "offset_ns": 4.0,
+                    "previous_hop_offset_ns": 4.0,
+                    "valid": True,
+                    "comparison_uncertainty_ns": 2.0,
+                },
+            ],
+        }
+
+    def test_append_and_read_do_not_leak_descriptors(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "samples.sqlite3"
+            STORE.append_sample(path, self._sample(0), {})
+            before = _open_descriptors()
+            for index in range(1, self.ITERATIONS):
+                STORE.append_sample(path, self._sample(index), {"BC1": 80.0})
+                if index % 100 == 0:
+                    STORE.read_records(path, 600.0)
+            after = _open_descriptors()
+
+            self.assertLessEqual(after - before, 2, "descriptors grew across writes")
+            self.assertEqual(self.ITERATIONS, len(STORE.read_records(path, 3600.0)))
+
+    def test_experiment_recorder_does_not_leak_descriptors(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = RESEARCH.ExperimentStore(Path(directory) / "experiments.sqlite3")
+            # record_phc returns immediately without an active run, so the run
+            # must exist or this test silently exercises nothing.
+            store.start({"kind": "capture", "name": "descriptor probe"}, {})
+            self.assertIsNotNone(store.active())
+
+            store.record_phc(self._sample(0), {})
+            before = _open_descriptors()
+            for index in range(1, self.ITERATIONS):
+                store.record_phc(self._sample(index), {"BC1": 80.0})
+            after = _open_descriptors()
+
+            run = store.active()
+            assert run is not None
+            self.assertEqual(
+                self.ITERATIONS * 2,  # two clocks per sample
+                len(store.phc_samples(str(run["id"]))),
+                "recorded rows missing: the write path was not exercised",
+            )
+            self.assertLessEqual(after - before, 2, "descriptors grew across recorded samples")
+
+
 if __name__ == "__main__":
     unittest.main()