fix(observability): wait for concurrent schema setup

Signed-off-by: Alex Fournier <afournier@nvidia.com>

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)

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"