fix(torghut): stabilize scheduler dependency health

argocd/applications/torghut/knative-service.yaml

@@ -44,7 +44,7 @@ spec:
             - name: TRADING_SCHEDULER_SHUTDOWN_DRAIN_SECONDS
               value: "45"
             - name: TRADING_SCHEDULER_SUCCESS_MAX_AGE_SECONDS
-              value: "30"
+              value: "300"
             - name: TRADING_BROKER_MUTATION_RECOVERY_ENABLED
               value: "true"
             - name: TRADING_BROKER_MUTATION_HTTP_TIMEOUT_SECONDS

argocd/applications/torghut/scheduler-deployment.yaml

@@ -59,7 +59,7 @@ spec:
             - name: TRADING_SCHEDULER_SHUTDOWN_DRAIN_SECONDS
               value: "45"
             - name: TRADING_SCHEDULER_SUCCESS_MAX_AGE_SECONDS
-              value: "30"
+              value: "300"
             - name: TRADING_SCHEDULER_LEADERSHIP_REQUIRED
               value: "true"
             - name: TRADING_SCHEDULER_LEADERSHIP_CHECK_SECONDS

services/torghut/app/api/health_checks/tigerbeetle_health.py

@@ -3,16 +3,21 @@
 from __future__ import annotations
 
 import logging
+import threading
 from collections.abc import Mapping, Sequence
-from concurrent.futures import ThreadPoolExecutor, TimeoutError
 from typing import cast
 
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.db import ping
-from app.trading.tigerbeetle_client import check_tigerbeetle_health
+from app.trading.tigerbeetle_client import (
+    RealTigerBeetleClient,
+    check_tigerbeetle_health,
+    create_tigerbeetle_client,
+    parse_replica_addresses,
+)
 from app.trading.tigerbeetle_reconcile import (
     BLOCKER_RECONCILIATION_STALE,
     latest_tigerbeetle_reconciliation_payload,
@@ -22,6 +27,39 @@
 
 logger = logging.getLogger(__name__)
 
+_protocol_health_lock = threading.Lock()
+_protocol_health_client: RealTigerBeetleClient | None = None
+
+
+def _close_tigerbeetle_protocol_health_client_locked() -> None:
+    global _protocol_health_client
+
+    client = _protocol_health_client
+    _protocol_health_client = None
+    if client is None:
+        return
+    client.close()
+
+
+def close_tigerbeetle_protocol_health_client() -> None:
+    """Close the process-owned status probe client."""
+
+    with _protocol_health_lock:
+        _close_tigerbeetle_protocol_health_client_locked()
+
+
+def _protocol_health_client_for_settings(
+    timeout_seconds: float,
+) -> RealTigerBeetleClient:
+    global _protocol_health_client
+
+    if _protocol_health_client is None:
+        _protocol_health_client = create_tigerbeetle_client(
+            settings,
+            rpc_timeout_seconds=timeout_seconds,
+        )
+    return _protocol_health_client
+
 
 def apply_status_read_statement_timeout(
     session: Session,
@@ -54,18 +92,16 @@ def check_postgres(session: Session) -> dict[str, object]:
 
 def check_tigerbeetle_protocol_health() -> dict[str, object]:
     if not settings.tigerbeetle_enabled:
+        close_tigerbeetle_protocol_health_client()
         health = check_tigerbeetle_health(settings)
         payload = health.as_dict()
         payload["protocol_ok"] = True
         payload["protocol_probe_skipped"] = False
         return payload
 
-    replica_addresses = [
-        item.strip()
-        for item in settings.tigerbeetle_replica_addresses.split(",")
-        if item.strip()
-    ]
+    replica_addresses = parse_replica_addresses(settings.tigerbeetle_replica_addresses)
     if not (settings.tigerbeetle_required or settings.tigerbeetle_reconcile_required):
+        close_tigerbeetle_protocol_health_client()
         return {
             "enabled": True,
             "required": settings.tigerbeetle_required,
@@ -78,26 +114,11 @@ def check_tigerbeetle_protocol_health() -> dict[str, object]:
         }
 
     timeout_seconds = max(0.1, float(settings.tigerbeetle_health_timeout_seconds))
-    executor = ThreadPoolExecutor(max_workers=1)
-    future = executor.submit(check_tigerbeetle_health, settings)
-    try:
-        health = future.result(timeout=timeout_seconds)
-    except TimeoutError:
-        return {
-            "enabled": True,
-            "required": settings.tigerbeetle_required,
-            "ok": not settings.tigerbeetle_required,
-            "protocol_ok": False,
-            "protocol_probe_skipped": False,
-            "cluster_id": settings.tigerbeetle_cluster_id,
-            "replica_addresses": replica_addresses,
-            "last_error": (
-                f"TimeoutError: tigerbeetle protocol health timed out after "
-                f"{timeout_seconds:.2f}s"
-            ),
-        }
-    finally:
-        executor.shutdown(wait=False, cancel_futures=True)
+    with _protocol_health_lock:
+        client = _protocol_health_client_for_settings(timeout_seconds)
+        health = check_tigerbeetle_health(settings, client=client)
+        if not health.ok:
+            _close_tigerbeetle_protocol_health_client_locked()
 
     payload = health.as_dict()
     protocol_ok = bool(payload.get("ok"))
@@ -432,6 +453,7 @@ def build_tigerbeetle_ledger_status(session: Session) -> dict[str, object]:
     "build_tigerbeetle_ledger_status",
     "check_postgres",
     "check_tigerbeetle_protocol_health",
+    "close_tigerbeetle_protocol_health_client",
     "empty_tigerbeetle_ref_counts",
     "latest_reconciliation_ref_counts",
     "sqlalchemy_error_indicates_statement_timeout",

services/torghut/app/bootstrap.py

@@ -14,6 +14,9 @@
 from sqlalchemy.exc import SQLAlchemyError
 
 from .api.build_metadata import BUILD_COMMIT, BUILD_VERSION
+from .api.health_checks.tigerbeetle_health import (
+    close_tigerbeetle_protocol_health_client,
+)
 from .config import settings
 from .db import SessionLocal, ensure_schema
 from .trading.autonomy import assert_runtime_gate_policy_contract
@@ -244,6 +247,7 @@ async def lifespan(app: FastAPI):
         if whitepaper_worker is not None:
             await whitepaper_worker.stop()
         await scheduler.stop()
+        close_tigerbeetle_protocol_health_client()
         logger.info("Torghut shutdown complete")
 
 

services/torghut/app/scheduler_main.py

@@ -12,6 +12,9 @@
 from starlette.types import ExceptionHandler
 
 from .api.application import build_registered_app
+from .api.health_checks.tigerbeetle_health import (
+    close_tigerbeetle_protocol_health_client,
+)
 from .bootstrap import assert_dspy_cutover_migration_guard, sqlalchemy_exception_handler
 from .config import settings
 from .db import SessionLocal, ensure_schema
@@ -85,6 +88,7 @@ async def scheduler_lifespan(app: FastAPI):
         logger.info("Torghut scheduler shutdown initiated")
         await whitepaper_worker.stop()
         await scheduler.stop()
+        close_tigerbeetle_protocol_health_client()
         logger.info("Torghut scheduler shutdown complete")
 
 

services/torghut/app/trading/tigerbeetle_client.py

@@ -322,13 +322,21 @@ def lookup_transfers(self, ids: Sequence[int]) -> Sequence[object]:
         return [self.transfers[item] for item in ids if item in self.transfers]
 
 
-def create_tigerbeetle_client(settings: Settings) -> RealTigerBeetleClient:
+def create_tigerbeetle_client(
+    settings: Settings,
+    *,
+    rpc_timeout_seconds: float | None = None,
+) -> RealTigerBeetleClient:
     return RealTigerBeetleClient(
         cluster_id=settings.tigerbeetle_cluster_id,
         replica_addresses=parse_replica_addresses(
             settings.tigerbeetle_replica_addresses
         ),
-        rpc_timeout_seconds=settings.tigerbeetle_rpc_timeout_seconds,
+        rpc_timeout_seconds=(
+            settings.tigerbeetle_rpc_timeout_seconds
+            if rpc_timeout_seconds is None
+            else rpc_timeout_seconds
+        ),
     )
 
 

services/torghut/tests/live_config_manifest_contract/test_single_writer_scheduler_manifest.py

@@ -106,7 +106,7 @@ def test_scheduler_is_singleton_recreate_and_has_dedicated_probes(self) -> None:
             env["TRADING_SCHEDULER_SHUTDOWN_DRAIN_SECONDS"].get("value"), "45"
         )
         self.assertEqual(
-            env["TRADING_SCHEDULER_SUCCESS_MAX_AGE_SECONDS"].get("value"), "30"
+            env["TRADING_SCHEDULER_SUCCESS_MAX_AGE_SECONDS"].get("value"), "300"
         )
         self.assertEqual(
             env["TRADING_BROKER_MUTATION_RECOVERY_ENABLED"].get("value"), "true"

services/torghut/tests/test_tigerbeetle_client.py

@@ -168,6 +168,14 @@ def test_create_tigerbeetle_client_uses_normalized_settings(self) -> None:
             rpc_timeout_seconds=settings.tigerbeetle_rpc_timeout_seconds,
         )
 
+    def test_create_tigerbeetle_client_accepts_operation_timeout(self) -> None:
+        settings = Settings(TORGHUT_TIGERBEETLE_CLUSTER_ID=77)
+
+        with patch("app.trading.tigerbeetle_client.RealTigerBeetleClient") as cls:
+            create_tigerbeetle_client(settings, rpc_timeout_seconds=0.25)
+
+        self.assertEqual(cls.call_args.kwargs["rpc_timeout_seconds"], 0.25)
+
     def test_timeout_helper_normalizes_operation_errors(self) -> None:
         with self.assertRaisesRegex(
             TigerBeetleClientError,

services/torghut/tests/test_tigerbeetle_status.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
-import time
 from datetime import datetime, timezone
 from decimal import Decimal
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from sqlalchemy import create_engine
 from sqlalchemy.exc import SQLAlchemyError
@@ -50,8 +49,18 @@ def setUp(self) -> None:
         settings.tigerbeetle_reconcile_required = False
         settings.tigerbeetle_reconcile_max_age_seconds = 3600
         settings.tigerbeetle_health_timeout_seconds = 1.0
+        health_checks_context.close_tigerbeetle_protocol_health_client()
+        self.protocol_client = MagicMock()
+        self._create_protocol_client_patch = patch.object(
+            health_checks_context,
+            "create_tigerbeetle_client",
+            return_value=self.protocol_client,
+        )
+        self.create_protocol_client = self._create_protocol_client_patch.start()
 
     def tearDown(self) -> None:
+        health_checks_context.close_tigerbeetle_protocol_health_client()
+        self._create_protocol_client_patch.stop()
         settings.tigerbeetle_enabled = self._orig_enabled
         settings.tigerbeetle_required = self._orig_required
         settings.tigerbeetle_journal_enabled = self._orig_journal_enabled
@@ -591,31 +600,72 @@ def test_optional_protocol_probe_is_skipped_until_required(self) -> None:
         self.assertIsNone(payload["last_error"])
         health_mock.assert_not_called()
 
-    def test_required_protocol_timeout_blocks_readiness_dependency(self) -> None:
+    def test_required_protocol_probe_reuses_one_bounded_client(self) -> None:
         settings.tigerbeetle_enabled = True
         settings.tigerbeetle_required = True
-        settings.tigerbeetle_health_timeout_seconds = 0.01
+        settings.tigerbeetle_health_timeout_seconds = 0.25
+        health = TigerBeetleHealth(
+            enabled=True,
+            required=True,
+            ok=True,
+            cluster_id=2001,
+            replica_addresses=["tb:3000"],
+            last_error=None,
+        )
 
-        def slow_health(_settings: object) -> TigerBeetleHealth:
-            time.sleep(0.2)
-            return TigerBeetleHealth(
-                enabled=True,
-                required=True,
-                ok=True,
-                cluster_id=2001,
-                replica_addresses=["tb:3000"],
-                last_error=None,
-            )
+        with patch.object(
+            health_checks_context,
+            "check_tigerbeetle_health",
+            return_value=health,
+        ) as health_mock:
+            first = check_tigerbeetle_protocol_health()
+            second = check_tigerbeetle_protocol_health()
+
+        self.assertTrue(first["ok"])
+        self.assertTrue(second["ok"])
+        self.create_protocol_client.assert_called_once_with(
+            settings,
+            rpc_timeout_seconds=0.25,
+        )
+        self.assertEqual(health_mock.call_count, 2)
+        for call in health_mock.call_args_list:
+            self.assertIs(call.kwargs["client"], self.protocol_client)
+
+    def test_failed_protocol_probe_discards_client_before_retry(self) -> None:
+        settings.tigerbeetle_enabled = True
+        settings.tigerbeetle_required = True
+        first_client = MagicMock()
+        second_client = MagicMock()
+        self.create_protocol_client.side_effect = [first_client, second_client]
+        failed = TigerBeetleHealth(
+            enabled=True,
+            required=True,
+            ok=False,
+            cluster_id=2001,
+            replica_addresses=["tb:3000"],
+            last_error="TigerBeetleClientTimeoutError: timed out",
+        )
+        healthy = TigerBeetleHealth(
+            enabled=True,
+            required=True,
+            ok=True,
+            cluster_id=2001,
+            replica_addresses=["tb:3000"],
+            last_error=None,
+        )
 
         with patch.object(
-            health_checks_context, "check_tigerbeetle_health", side_effect=slow_health
+            health_checks_context,
+            "check_tigerbeetle_health",
+            side_effect=[failed, healthy],
         ):
-            payload = check_tigerbeetle_protocol_health()
+            first = check_tigerbeetle_protocol_health()
+            second = check_tigerbeetle_protocol_health()
 
-        self.assertFalse(payload["ok"])
-        self.assertFalse(payload["protocol_ok"])
-        self.assertFalse(payload["protocol_probe_skipped"])
-        self.assertIn("TimeoutError", str(payload["last_error"]))
+        self.assertFalse(first["ok"])
+        self.assertTrue(second["ok"])
+        first_client.close.assert_called_once_with()
+        self.assertEqual(self.create_protocol_client.call_count, 2)
 
     def test_latest_reconciliation_blockers_are_reported(self) -> None:
         settings.tigerbeetle_enabled = True