Exercise PostgreSQL real-process crash recovery

.github/workflows/ci.yml

@@ -46,6 +46,7 @@ jobs:
 
   postgres-foundation:
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     services:
       postgres:
         image: postgres:18.4-bookworm
@@ -72,10 +73,11 @@ jobs:
           cache: pip
       - run: python -m pip install --upgrade pip setuptools wheel
       - run: python -m pip install -e ".[postgres,test]"
-      - run: python -m pytest -m postgres
+      - run: python -m pytest -m postgres --timeout=90
 
   quality:
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     services:
       postgres:
         image: postgres:18.4-bookworm
@@ -106,7 +108,7 @@ jobs:
       - run: python -m ruff format --check .
       - run: python -m mypy samsarix_chat_engine
       - run: python -m pip_audit
-      - run: python -m pytest --cov=samsarix_chat_engine --cov-report=term-missing
+      - run: python -m pytest --cov=samsarix_chat_engine --cov-report=term-missing --timeout=90
       - run: python -m build
       - run: python -m twine check dist/*
       - run: python -m venv .wheel-smoke

CHANGELOG.md

@@ -19,6 +19,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma
 - PostgreSQL schema v8 realtime-retention metadata, bounded count/age pruning behind every live cursor, explicit retained-gap detection, and relay recovery that fences sockets, rotates the stale process generation, and resumes from the authoritative event head.
 - Guarded PostgreSQL application configuration and lifecycle orchestration for shared HTTP storage, cross-instance WebSocket messages and room state, global socket capacity/rate controls/stats, leased connection renewal, sender-excluded presence/typing, bounded maintenance, and readiness.
 - A PostgreSQL preview deployment guide covering protected URL files, mandatory remote `verify-full` TLS, unique replica identities, pool/lease/retention bounds, migration, backup/rollback ownership, and remaining release gates.
+- A real-network acceptance test that launches two independent Uvicorn processes, verifies cross-process HTTP/WebSocket delivery, kills one replica, observes lease-derived presence/capacity convergence, restarts its stable identity after expiry, and reloads durable history.
 
 ### Security and operations
 

ROADMAP.md

@@ -92,7 +92,8 @@ This makes the supported topology repeatable without implying that a container m
 - [x] bound the retained event log behind live cursors and fence/recover stale workers that return after a gap;
 - [x] expose explicitly guarded PostgreSQL preview configuration without changing the SQLite default;
 - [x] wire cross-instance fan-out, presence, typing, rate controls, connection leases, maintenance, and readiness into the application;
-- [ ] prove subprocess kill/restart, listener interruption, archive/ban teardown, and reconnect recovery against real network processes;
+- [x] prove two real Uvicorn processes share HTTP/WebSocket state, reclaim a killed replica's socket lease, restart its stable identity, and recover durable history;
+- [ ] prove listener interruption, archive/ban teardown, and explicit client reconnect recovery against real network processes;
 - [ ] run sustained load/soak and reconnect-storm tests and publish measured limits;
 - [ ] verify deployment manifests assign a unique stable instance ID to every replica and reject duplicate live ownership;
 - [ ] validate live-lag and `NOTIFY` interruption behavior under measured traffic;

docs/MULTI_INSTANCE_ARCHITECTURE.md

@@ -10,7 +10,7 @@ The supported multi-instance topology will use PostgreSQL as both the authoritat
 
 PostgreSQL `LISTEN`/`NOTIFY` will be a low-latency wake-up hint only. Every client-visible realtime event will first be inserted into a bounded, ordered database event log in the same transaction as the state change. Each service instance reads committed rows from its own durable cursor. A listener reconnect, notification-queue loss, or process pause therefore causes polling/replay rather than silent event loss.
 
-The relay implements that polling/replay correctness path and is now wired into the guarded PostgreSQL application runtime. Each process exclusively claims an expiring generation-owned cursor, dispatches an ordered batch to its local socket manager, and checkpoints the last successfully dispatched event at the batch or failure boundary. A duplicate live instance ID fails closed; after replacement, a stale generation cannot heartbeat, read, or acknowledge. A dispatch exception keeps the relay alive and retries from the failed event without replaying acknowledged predecessors. A database acknowledgement failure can make the relay-to-process handoff at least once across a crash boundary; the process fences its old local sockets before replay, preserving the existing at-most-once boundary for any individual socket connection. An expired lease or database interruption likewise fences all local sockets before the process renews the same cursor. Schema v8 records a durable pruning watermark: a returning cursor behind that watermark closes every local socket, rotates its process generation so old connection leases remain stale, and advances to the current authoritative head before serving later events. Public message, room-state, presence, and typing envelopes are broadcast; sender exclusion strips the internal origin connection ID before delivery. Archive and active-ban events invoke deterministic local teardown, while other internal lifecycle records are not leaked onto the public protocol. Real subprocess recovery and notification-assisted latency remain release gates.
+The relay implements that polling/replay correctness path and is now wired into the guarded PostgreSQL application runtime. Each process exclusively claims an expiring generation-owned cursor, dispatches an ordered batch to its local socket manager, and checkpoints the last successfully dispatched event at the batch or failure boundary. A duplicate live instance ID fails closed; after replacement, a stale generation cannot heartbeat, read, or acknowledge. A dispatch exception keeps the relay alive and retries from the failed event without replaying acknowledged predecessors. A database acknowledgement failure can make the relay-to-process handoff at least once across a crash boundary; the process fences its old local sockets before replay, preserving the existing at-most-once boundary for any individual socket connection. An expired lease or database interruption likewise fences all local sockets before the process renews the same cursor. Schema v8 records a durable pruning watermark: a returning cursor behind that watermark closes every local socket, rotates its process generation so old connection leases remain stale, and advances to the current authoritative head before serving later events. Public message, room-state, presence, and typing envelopes are broadcast; sender exclusion strips the internal origin connection ID before delivery. Archive and active-ban events invoke deterministic local teardown, while other internal lifecycle records are not leaked onto the public protocol. Real subprocess kill/lease-expiry/restart recovery now runs in CI; forced database/listener interruption and notification-assisted latency remain release gates.
 
 The initial event-log implementation serializes sequence allocation with a transaction-scoped advisory lock. PostgreSQL identity values alone are not commit ordered: without this lock, a later sequence could commit and be acknowledged before an earlier transaction becomes visible. Event append must remain the final lock-taking phase of a domain mutation. Sustained-load acceptance tests will determine whether this intentionally simple global sequencer is sufficient or must be partitioned without weakening cursor correctness.
 
@@ -78,14 +78,14 @@ Application replicas must run the exact same Samsarix version and security confi
 
 v0.13 cannot claim multi-instance support until CI proves:
 
-- two or more real app processes share one PostgreSQL database and deliver create/update/delete events exactly once to each connected test socket under normal operation (two independently lifespanned application instances now pass; separate OS subprocesses remain);
+- two or more real app processes share one PostgreSQL database and deliver create/update/delete events exactly once to each connected test socket under normal operation (two independent OS processes now pass for message creation; update/delete process coverage remains);
 - a listener disconnect/reconnect replays committed event rows without relying on `NOTIFY` delivery (the internal polling relay and cursor replay case are implemented; the real-process listener gate remains);
 - concurrent idempotent message creation returns one authoritative message;
 - global and per-room connection caps plus rate limits hold across processes (storage-level concurrency and real application request paths are implemented; separate OS subprocess contention remains);
 - archive and ban actions close matching sockets on every process;
 - two webhook workers never hold the same live claim, and a killed worker's claim is recovered (the storage-level lease/recovery case is implemented; the killed-process gate remains);
 - migration concurrency is serialized and newer schemas fail closed;
-- crashed connection leases expire and presence converges;
+- crashed connection leases expire and presence converges (a killed real Uvicorn process now emits the expected lease-derived leave/count on its surviving peer, then restarts under the same stable ID after expiry);
 - an event-log gap fences the lagging instance and clients recover through history (the storage/relay contract is implemented; the real-client recovery gate remains);
 - sustained load and forced database/network interruptions have measured, published outcomes;
 - SQLite single-instance behavior, package installation, Windows support, and rollback documentation remain green.

docs/POSTGRES_PREVIEW.md

@@ -2,7 +2,7 @@
 
 Status: **guarded, unreleased v0.13 preview**.
 
-The repository's development branch can run the complete HTTP and WebSocket application against PostgreSQL. SQLite remains the default and the v0.12 supported topology remains one process. Do not describe the preview as production-supported until the remaining interruption, subprocess, load/soak, backup, and rollback gates in the [multi-instance architecture](MULTI_INSTANCE_ARCHITECTURE.md) are published.
+The repository's development branch can run the complete HTTP and WebSocket application against PostgreSQL. SQLite remains the default and the v0.12 supported topology remains one process. CI now exercises two real Uvicorn processes and crash-lease recovery, but do not describe the preview as production-supported until the remaining interruption, reconnect, load/soak, backup, and rollback gates in the [multi-instance architecture](MULTI_INSTANCE_ARCHITECTURE.md) are published.
 
 ## What the preview wires
 
@@ -79,7 +79,7 @@ Rolling back to a binary that supports an older schema requires restoring its ma
 
 ## Known preview boundaries
 
-- Real subprocess kill/restart, network interruption, reconnect-storm, and sustained load/soak evidence is still pending.
+- Two-process normal delivery and kill/lease-expiry/restart recovery run in CI; forced database/listener interruption, explicit reconnect recovery, reconnect storms, and sustained load/soak evidence remain pending.
 - A live but extremely slow replica can currently hold the event-retention floor; the configurable live-lag fence is not implemented yet.
 - Polling, rather than `LISTEN`/`NOTIFY`, currently determines normal fan-out latency.
 - The bundled Compose profile is still the supported SQLite single-replica example and does not provision PostgreSQL.

samsarix_chat_engine/app.py

@@ -1194,6 +1194,7 @@ async def room_websocket(
             username,
             principal.subject,
             connection_id=connection_id,
+            broadcast_ready=False,
         )
         if not registered:
             if admitted and postgres_runtime is not None:
@@ -1237,7 +1238,7 @@ async def room_websocket(
             active_connections = (await postgres_runtime.connection_counts(room_id)).room
         else:
             active_connections = manager.room_connections(room_id)
-        await manager.send(
+        ready_sent = await manager.send(
             websocket,
             _event(
                 "ready",
@@ -1247,14 +1248,22 @@ async def room_websocket(
                 max_message_chars=resolved.max_message_chars,
             ),
         )
-        await manager.send(
+        if not ready_sent:
+            if postgres_runtime is not None:
+                await postgres_runtime.release_connection(connection_id)
+            return
+        history_sent = await manager.send(
             websocket,
             _event(
                 "history",
                 items=[message.model_dump(mode="json") for message in history],
                 next_before=next_before,
             ),
         )
+        if not history_sent or not await manager.activate(websocket):
+            if postgres_runtime is not None:
+                await postgres_runtime.release_connection(connection_id)
+            return
         if postgres_runtime is None:
             await manager.broadcast(
                 room_id,

samsarix_chat_engine/websocket_manager.py

@@ -20,6 +20,7 @@ class ConnectionMetadata(NamedTuple):
     subject: str | None
     connection_id: str | None
     operation_lock: asyncio.Lock
+    broadcast_ready: bool
 
 
 class ConnectionManager:
@@ -41,6 +42,7 @@ async def register(
         subject: str | None = None,
         *,
         connection_id: str | None = None,
+        broadcast_ready: bool = True,
     ) -> bool:
         """Register an already-accepted socket, returning false at capacity."""
 
@@ -56,9 +58,20 @@ async def register(
                 subject,
                 connection_id,
                 asyncio.Lock(),
+                broadcast_ready,
             )
             return True
 
+    async def activate(self, websocket: WebSocket) -> bool:
+        """Allow broadcasts only after the socket's initial handshake is sent."""
+
+        async with self._lock:
+            metadata = self._metadata.get(websocket)
+            if metadata is None:
+                return False
+            self._metadata[websocket] = metadata._replace(broadcast_ready=True)
+            return True
+
     async def unregister(self, websocket: WebSocket) -> tuple[str, str] | None:
         """Remove a socket and return its room/user metadata once."""
 
@@ -111,6 +124,7 @@ async def broadcast(
                 (connection, self._metadata[connection].operation_lock)
                 for connection in self._rooms.get(room_id, ())
                 if connection is not exclude
+                and self._metadata[connection].broadcast_ready
                 and (exclude_connection_id is None or self._metadata[connection].connection_id != exclude_connection_id)
             )
         if recipients:

tests/test_connection_manager.py

@@ -99,6 +99,27 @@ async def test_broadcast_can_exclude_an_origin_connection_id() -> None:
     assert peer.sent == [{"type": "typing.started"}]
 
 
+@pytest.mark.asyncio
+async def test_pending_connection_receives_initial_frames_before_broadcasts() -> None:
+    manager = ConnectionManager(max_connections=2, max_per_room=2, send_timeout=0.1)
+    pending = FakeWebSocket()
+    websocket = as_websocket(pending)
+    await manager.register(websocket, "room", "Pending", broadcast_ready=False)
+
+    await manager.broadcast("room", {"type": "presence.joined"})
+    assert pending.sent == []
+    assert await manager.send(websocket, {"type": "ready"}) is True
+    assert await manager.send(websocket, {"type": "history"}) is True
+    assert await manager.activate(websocket) is True
+    await manager.broadcast("room", {"type": "message.created"})
+
+    assert pending.sent == [
+        {"type": "ready"},
+        {"type": "history"},
+        {"type": "message.created"},
+    ]
+
+
 @pytest.mark.asyncio
 async def test_send_and_close_are_serialized_per_connection() -> None:
     manager = ConnectionManager(max_connections=1, max_per_room=1, send_timeout=0.1)