diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index af618f44e09..0c4912a0a80 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -25,6 +25,7 @@ from .shared_metrics_contract import ( _PACKAGE_SCHEMA_VERSION = "hermes.shared_metrics.v1" _STORE_SCHEMA_VERSION = "1" _BUSY_TIMEOUT_MS = 250 +_SCHEMA_BUSY_TIMEOUT_MS = 5_000 def _utc_now() -> datetime: @@ -140,14 +141,18 @@ class SharedMetricsStore: ] @contextmanager - def _connection(self) -> Iterator[sqlite3.Connection]: + def _connection( + self, + *, + busy_timeout_ms: int = _BUSY_TIMEOUT_MS, + ) -> Iterator[sqlite3.Connection]: connection = sqlite3.connect( self.database_path, - timeout=_BUSY_TIMEOUT_MS / 1000, + timeout=busy_timeout_ms / 1000, ) try: connection.row_factory = sqlite3.Row - connection.execute(f"PRAGMA busy_timeout={_BUSY_TIMEOUT_MS}") + connection.execute(f"PRAGMA busy_timeout={busy_timeout_ms}") with connection: yield connection finally: @@ -170,7 +175,7 @@ class SharedMetricsStore: pass def _ensure_schema(self) -> None: - with self._connection() as connection: + with self._connection(busy_timeout_ms=_SCHEMA_BUSY_TIMEOUT_MS) as connection: # Serialize first-run creation and upgrades across Hermes processes. with write_txn(connection): self._ensure_schema_in_transaction(connection) diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index 8aa746cf683..6dfb11eec9c 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -8,6 +8,7 @@ import os import sqlite3 import stat import threading +import time import uuid from concurrent.futures import ThreadPoolExecutor from copy import deepcopy @@ -722,6 +723,30 @@ def test_cross_process_model_call_updates_are_transactional(tmp_path): assert restarted.counter_snapshot()[0]["value"] == 20 +def test_schema_initialization_waits_for_an_existing_writer(tmp_path): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + database_path.touch() + blocker = sqlite3.connect(database_path) + blocker.execute("BEGIN IMMEDIATE") + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit( + SharedMetricsStore, + database_path, + outbox_directory, + ) + try: + time.sleep(0.4) + assert not future.done() + finally: + blocker.rollback() + blocker.close() + store = future.result(timeout=2) + + assert store.counter_snapshot() == [] + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission modes are unavailable") def test_store_and_export_are_owner_only(tmp_path): database_path = tmp_path / "private-store" / "metrics.sqlite3"