fix: log clean one-liner when PostgreSQL becomes unavailable (#288)

* fix: log clean one-liner when PostgreSQL becomes unavailable

Wrap all 174 connection sites in PostgreSQLBackend through a _conn()
context manager that catches OperationalError, emits a single
database.unavailable log line (with connection URL), and suppresses
repeats until the connection is restored (database.connection_restored).

* fix: add StorageUnavailableError and cover all heartbeat loops

Address review feedback:
- Separate connect-phase from execution-phase in _conn() so that
  OperationalError during caller code (e.g. BEGIN IMMEDIATE lock
  contention) is not misclassified as a connectivity failure.
- Add StorageUnavailableError exception class so callers can
  distinguish transient DB outages without redundant tracebacks.
- Apply the same _conn() wrapper to SQLiteBackend for consistency.
- Catch StorageUnavailableError in all 7 periodic loops: watch
  runner, server heartbeat, channel heartbeat, console heartbeat,
  collector discovery, rebalancer, and scheduler.
- Guard dedup flag with threading.Lock.
- Add tests for dedup logging and PostgreSQL path.

tests/test_storage_registry.py

@@ -1,8 +1,17 @@
 """Tests for the storage backend registry."""
 
-import pytest
+from unittest.mock import patch
 
-from turnstone.core.storage import get_storage, init_storage, reset_storage
+import pytest
+import sqlalchemy as sa
+
+from turnstone.core.storage import (
+    StorageUnavailableError,
+    get_storage,
+    init_storage,
+    reset_storage,
+)
+from turnstone.core.storage._postgresql import PostgreSQLBackend
 from turnstone.core.storage._sqlite import SQLiteBackend
 
 
@@ -52,3 +61,69 @@ class TestResetStorage:
         init_storage("sqlite", path=str(tmp_path / "test2.db"), run_migrations=False)
         s2 = get_storage()
         assert s1 is not s2
+
+
+class TestConnUnavailableLogging:
+    """Test that _conn() deduplicates DB unavailable/restored logging."""
+
+    def _make_backend(self, tmp_path):
+        """Create a minimal SQLite backend for testing _conn()."""
+        from turnstone.core.storage._sqlite import SQLiteBackend
+
+        return SQLiteBackend(str(tmp_path / "test.db"), create_tables=True)
+
+    def test_logs_unavailable_once(self, tmp_path, caplog: pytest.LogCaptureFixture) -> None:
+        backend = self._make_backend(tmp_path)
+        with patch.object(backend, "_engine") as mock_engine:
+            mock_engine.connect.side_effect = sa.exc.OperationalError(
+                "conn", {}, Exception("refused")
+            )
+            for _ in range(3):
+                with pytest.raises(StorageUnavailableError), backend._conn():
+                    pass  # pragma: no cover
+        unavailable_msgs = [r for r in caplog.records if "database.unavailable" in r.message]
+        assert len(unavailable_msgs) == 1
+
+    def test_logs_restored_on_recovery(self, tmp_path, caplog: pytest.LogCaptureFixture) -> None:
+        import logging
+
+        caplog.set_level(logging.INFO)
+        backend = self._make_backend(tmp_path)
+        # Simulate outage
+        with patch.object(backend, "_engine") as mock_engine:
+            mock_engine.connect.side_effect = sa.exc.OperationalError(
+                "conn", {}, Exception("refused")
+            )
+            with pytest.raises(StorageUnavailableError), backend._conn():
+                pass  # pragma: no cover
+        assert backend._db_unavailable is True
+        # Real connection — should log restored
+        caplog.clear()
+        with backend._conn():
+            pass
+        restored_msgs = [r for r in caplog.records if "database.connection_restored" in r.message]
+        assert len(restored_msgs) == 1
+        assert backend._db_unavailable is False
+
+    def test_postgresql_conn_raises_storage_unavailable(self) -> None:
+        import threading
+
+        backend = PostgreSQLBackend.__new__(PostgreSQLBackend)
+        backend._db_unavailable = False
+        backend._db_unavailable_lock = threading.Lock()
+
+        def _raise_op_error():
+            raise sa.exc.OperationalError("conn", {}, Exception("refused"))
+
+        mock_engine = type(
+            "E",
+            (),
+            {
+                "connect": staticmethod(_raise_op_error),
+                "url": sa.engine.make_url("postgresql://user:pass@localhost/db"),
+            },
+        )()
+        backend._engine = mock_engine
+        with pytest.raises(StorageUnavailableError), backend._conn():
+            pass  # pragma: no cover
+        assert backend._db_unavailable is True

turnstone/channels/cli.py

@@ -282,10 +282,14 @@ def main() -> None:
 
             async def _heartbeat_loop() -> None:
                 """Periodically update service heartbeat."""
+                from turnstone.core.storage._registry import StorageUnavailableError
+
                 while True:
                     await asyncio.sleep(30)
                     try:
                         await asyncio.to_thread(storage.heartbeat_service, "channel", service_id)
+                    except StorageUnavailableError:
+                        pass  # already logged by storage layer
                     except Exception:
                         log.exception("channel.heartbeat_failed")
 

turnstone/console/collector.py

@@ -289,9 +289,13 @@ class ClusterCollector:
 
     def _discovery_loop(self) -> None:
         """Periodically scan the service registry for active nodes."""
+        from turnstone.core.storage._registry import StorageUnavailableError
+
         while self._running:
             try:
                 self._discover_nodes()
+            except StorageUnavailableError:
+                pass  # already logged by storage layer
             except Exception:
                 log.exception("Node discovery error")
             time.sleep(self._discovery_interval)

turnstone/console/rebalancer.py

@@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, Any
 import structlog
 
 from turnstone.core.hash_ring import RING_SIZE, RingNode, bucket_of
+from turnstone.core.storage._registry import StorageUnavailableError
 
 if TYPE_CHECKING:
     from turnstone.console.collector import ClusterCollector
@@ -149,6 +150,8 @@ class Rebalancer:
                 result = self.rebalance_once(trigger=trigger)
                 self._last_result = result
                 self._record_result_metrics(result)
+            except StorageUnavailableError:
+                pass  # already logged by storage layer
             except Exception:
                 log.exception("rebalancer.error")
             finally:

turnstone/console/scheduler.py

@@ -90,9 +90,13 @@ class TaskScheduler:
 
     def _loop(self) -> None:
         """Main scheduler loop — tick then sleep."""
+        from turnstone.core.storage._registry import StorageUnavailableError
+
         while not self._stop_event.is_set():
             try:
                 self._tick()
+            except StorageUnavailableError:
+                pass  # already logged by storage layer
             except Exception:
                 log.exception("scheduler.tick_error")
             self._stop_event.wait(self._check_interval)

turnstone/console/server.py

@@ -1167,10 +1167,14 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
             import asyncio
 
             async def _console_heartbeat() -> None:
+                from turnstone.core.storage._registry import StorageUnavailableError
+
                 while True:
                     await asyncio.sleep(30)
                     try:
                         storage.heartbeat_service("console", "console")
+                    except StorageUnavailableError:
+                        pass  # already logged by storage layer
                     except Exception:
                         log.warning("console.heartbeat_failed", exc_info=True)
 

turnstone/core/storage/__init__.py

@@ -4,10 +4,16 @@ Supports SQLite (default, zero-config) and PostgreSQL (multi-node, production).
 """
 
 from turnstone.core.storage._protocol import StorageBackend
-from turnstone.core.storage._registry import get_storage, init_storage, reset_storage
+from turnstone.core.storage._registry import (
+    StorageUnavailableError,
+    get_storage,
+    init_storage,
+    reset_storage,
+)
 
 __all__ = [
     "StorageBackend",
+    "StorageUnavailableError",
     "get_storage",
     "init_storage",
     "reset_storage",

turnstone/core/storage/_registry.py

@@ -15,6 +15,14 @@ log = get_logger(__name__)
 _storage: StorageBackend | None = None
 
 
+class StorageUnavailableError(Exception):
+    """Raised when the database is unreachable.
+
+    The storage layer has already logged a clean one-liner — callers
+    should catch this to avoid duplicate tracebacks.
+    """
+
+
 def init_storage(
     backend: str = "sqlite",
     *,

turnstone/core/watch.py

@@ -271,9 +271,13 @@ class WatchRunner:
     # -- Main loop -----------------------------------------------------------
 
     def _run(self) -> None:
+        from turnstone.core.storage._registry import StorageUnavailableError
+
         while not self._stop_event.is_set():
             try:
                 self._tick()
+            except StorageUnavailableError:
+                pass  # already logged by storage layer
             except Exception:
                 log.exception("watch_runner.tick_error")
             self._stop_event.wait(self._check_interval)

turnstone/server.py

@@ -2389,10 +2389,14 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
 
         async def _heartbeat_loop() -> None:
             """Periodically update service heartbeat."""
+            from turnstone.core.storage._registry import StorageUnavailableError
+
             while True:
                 await asyncio.sleep(30)
                 try:
                     await asyncio.to_thread(_svc_storage.heartbeat_service, "server", _svc_node_id)
+                except StorageUnavailableError:
+                    pass  # already logged by storage layer
                 except Exception:
                     log.exception("server.heartbeat_failed")