From d56002cb98c7c4a9073b334fdf8c0efe770e4c64 Mon Sep 17 00:00:00 2001 From: Ahmad Byagowi Date: Mon, 27 Jul 2026 08:57:38 -0700 Subject: [PATCH 1/3] Close SQLite connections so PHC acquisition cannot stall A sqlite3.Connection used directly as a context manager commits or rolls back the transaction but never closes the handle. Both the PHC ring store and the experiment recorder relied on that, so every recorded sample leaked a descriptor pair. At the Sync cadence the collector reached its 1024-descriptor limit in under three minutes; after that /dev/ptp* could not be opened either, take_phc_sample() returned None, and the loop spun silently forever. systemd still reported the unit active, so the only visible symptom was the Observatory reporting 0.0 Hz acquired and waiting for direct PHC samples. - add an owning _session() context manager to both stores that commits or rolls back and then always closes (10 call sites) - log unavailable samples instead of skipping silently, and exit for a supervisor restart after a sustained run of cycles delivering nothing - add regression tests asserting descriptors stay flat across sustained writes and recorded samples Verified: 1800 writes at a 256-descriptor limit produce zero descriptor growth; the same load leaks on the previous code. --- CHANGELOG.md | 12 ++++++ agent/ptpbox_phc_collector.py | 19 ++++++++++ agent/ptpbox_phc_store.py | 23 +++++++++++- agent/ptpbox_research.py | 34 +++++++++++++---- tests/test_phc_store.py | 70 +++++++++++++++++++++++++++++++++++ 5 files changed, 148 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d17471..73c96b2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/agent/ptpbox_phc_collector.py b/agent/ptpbox_phc_collector.py index c2523e6..38f21a5 100644 --- a/agent/ptpbox_phc_collector.py +++ b/agent/ptpbox_phc_collector.py @@ -14,17 +14,36 @@ def collect(stop: threading.Event) -> None: deadline = time.monotonic() + consecutive_failures = 0 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) + else: + print("PTPBox PHC sample unavailable", flush=True) + + # A sustained inability to acquire or store is not something this loop can + # repair: exhausted descriptors, a revoked store path, or a disappearing + # PHC all persist. Exit so the supervisor restarts a clean process + # instead of leaving a live service that silently delivers nothing. + consecutive_failures = 0 if delivered else consecutive_failures + 1 + if consecutive_failures >= max(10, int(round(requested_rate * 10))): + print( + f"PTPBox PHC collector delivered nothing for {consecutive_failures} " + "consecutive cycles; exiting for supervisor restart", + flush=True, + ) + raise SystemExit(1) + deadline += period now = time.monotonic() if deadline < now - period: diff --git a/agent/ptpbox_phc_store.py b/agent/ptpbox_phc_store.py index b2e0929..5149b8b 100644 --- a/agent/ptpbox_phc_store.py +++ b/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,), diff --git a/agent/ptpbox_research.py b/agent/ptpbox_research.py index 0fa8aa8..4e1ccff 100644 --- a/agent/ptpbox_research.py +++ b/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() @@ -3882,7 +3900,7 @@ def stop(self, identifier: str | None = None) -> dict[str, Any] | None: 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)),), @@ -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( """ @@ -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, diff --git a/tests/test_phc_store.py b/tests/test_phc_store.py index c730e5a..dde1b94 100644 --- a/tests/test_phc_store.py +++ b/tests/test_phc_store.py @@ -1,4 +1,6 @@ import importlib.util +import os +import sys import tempfile import time import unittest @@ -11,6 +13,20 @@ 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: + 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 +78,59 @@ 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: + return { + "observed_at": time.time() + index, + "sample_id": f"sample-{index}", + "clocks": [ + {"id": "BC1", "offset_ns": 0.0, "valid": True, "comparison_uncertainty_ns": 1.0}, + { + "id": "BC2", + "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") + 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() + + self.assertLessEqual(after - before, 2, "descriptors grew across recorded samples") + + if __name__ == "__main__": unittest.main() From cd9fd49fed2b366fa44d604631752efef6e00b1b Mon Sep 17 00:00:00 2001 From: Ahmad Byagowi Date: Mon, 27 Jul 2026 09:07:54 -0700 Subject: [PATCH 2/3] Only restart the collector after acquisition has actually worked The watchdog added in the previous commit could not tell a broken collector from one that had not started yet. On a cold boot the cascade is not running, so there are no mapped PHCs, every cycle legitimately yields nothing, and the collector exited every ten seconds into a supervisor restart loop. Exit only once a sample has been delivered and delivery then stops, which is the condition that actually indicates an unrepairable state. A process that has never acquired now waits quietly for the cascade instead. Also report the idle/resumed transitions once each rather than logging every cycle, which was filling the journal at the Sync cadence. --- agent/ptpbox_phc_collector.py | 32 +++++++++++++++++++++++--------- 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/agent/ptpbox_phc_collector.py b/agent/ptpbox_phc_collector.py index 38f21a5..640da83 100644 --- a/agent/ptpbox_phc_collector.py +++ b/agent/ptpbox_phc_collector.py @@ -15,6 +15,8 @@ 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 @@ -28,18 +30,30 @@ def collect(stop: threading.Event) -> None: 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: - print("PTPBox PHC sample unavailable", flush=True) + 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 - # A sustained inability to acquire or store is not something this loop can - # repair: exhausted descriptors, a revoked store path, or a disappearing - # PHC all persist. Exit so the supervisor restarts a clean process - # instead of leaving a live service that silently delivers nothing. - consecutive_failures = 0 if delivered else consecutive_failures + 1 - if consecutive_failures >= max(10, int(round(requested_rate * 10))): + # 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 delivered nothing for {consecutive_failures} " - "consecutive cycles; exiting for supervisor restart", + f"PTPBox PHC collector stopped delivering for {consecutive_failures} " + "consecutive cycles after working; exiting for supervisor restart", flush=True, ) raise SystemExit(1) From d1594db8c9d731fe50b37c8b10236bce923d9a4a Mon Sep 17 00:00:00 2001 From: Ahmad Byagowi Date: Mon, 27 Jul 2026 09:20:32 -0700 Subject: [PATCH 3/3] Close the remaining experiment-store connections The first pass migrated only the plain "with self._connect()" form and missed four sites that acquire the lock in the same statement, including record_phc, which runs once per acquired sample. On hardware the collector therefore still leaked two descriptors per sample through experiments.sqlite3 while the PHC ring store stayed flat. The regression test also passed for the wrong reason: record_phc returns early without an active run, so it exercised nothing. It now starts a run, asserts the rows were actually written, and collects garbage before counting descriptors so the measurement is not sensitive to platform GC timing. --- agent/ptpbox_research.py | 8 ++++---- tests/test_phc_store.py | 29 +++++++++++++++++++++++++++-- 2 files changed, 31 insertions(+), 6 deletions(-) diff --git a/agent/ptpbox_research.py b/agent/ptpbox_research.py index 4e1ccff..f9d1ce3 100644 --- a/agent/ptpbox_research.py +++ b/agent/ptpbox_research.py @@ -3876,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,), @@ -3892,7 +3892,7 @@ 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"]), @@ -3948,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, @@ -4089,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 (?, ?, ?, ?, ?, ?)", ( diff --git a/tests/test_phc_store.py b/tests/test_phc_store.py index dde1b94..babc5ed 100644 --- a/tests/test_phc_store.py +++ b/tests/test_phc_store.py @@ -1,3 +1,4 @@ +import gc import importlib.util import os import sys @@ -22,6 +23,10 @@ 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)) @@ -91,13 +96,21 @@ class DescriptorLifetimeTests(unittest.TestCase): @staticmethod def _sample(index: int) -> dict: + observed_at = time.time() + index return { - "observed_at": time.time() + index, + "observed_at": observed_at, "sample_id": f"sample-{index}", "clocks": [ - {"id": "BC1", "offset_ns": 0.0, "valid": True, "comparison_uncertainty_ns": 1.0}, + { + "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, @@ -123,12 +136,24 @@ def test_append_and_read_do_not_leak_descriptors(self) -> None: 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")