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..640da83 100644 --- a/agent/ptpbox_phc_collector.py +++ b/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: 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..f9d1ce3 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() @@ -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,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"]), @@ -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)),), @@ -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, diff --git a/tests/test_phc_store.py b/tests/test_phc_store.py index c730e5a..babc5ed 100644 --- a/tests/test_phc_store.py +++ b/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()