feat(postgres): enforce distributed connection capacity

CHANGELOG.md

@@ -12,6 +12,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma
 - Internal PostgreSQL schema v3 and a complete `ChatStorage` implementation for rooms, bounded messages, Unicode-normalized search, moderation, audit history, monotonic read state, stable spooled exports, explicit retention, and cross-instance idempotency/capacity enforcement.
 - Transactional PostgreSQL webhook outbox with stable delivery IDs, database-time scheduling, expiring worker-owned claims, `SKIP LOCKED` work selection, crash recovery, bounded terminal-history pruning, and operator replay.
 - Internal per-process PostgreSQL realtime relay with durable cursors, ordered polling/replay, post-dispatch acknowledgement, archive/ban socket teardown, and lease-loss fencing. Polling is the correctness path; a future `LISTEN`/`NOTIFY` listener may only reduce latency.
+- PostgreSQL schema v4 and an internal connection registry with database-time socket leases, atomic deployment-wide and per-room capacity, owner-bound renewal/release, archived-room rejection, and crashed-process reclamation.
 
 ### Security and operations
 
@@ -22,6 +23,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma
 - PostgreSQL event and webhook envelopes remain bounded at 512 KiB so every valid 100,000-character message, including four-byte Unicode, fits without turning a valid domain write into a coordination failure.
 - PostgreSQL webhook claims can be acknowledged only by the live lease owner. Expired claims are safely redelivered with the same ID; receivers must still deduplicate because delivery is at least once.
 - A PostgreSQL relay that loses its database lease closes every local socket before renewing the same durable cursor. Unsupported internal event types are not forwarded onto the public WebSocket protocol.
+- Expired process IDs discard their stale socket rows before re-registration, preventing a restarted replica from reviving phantom occupancy. Archived and expired reservations stop consuming capacity even before physical cleanup.
 
 ## 0.12.0 — 2026-08-02
 

ROADMAP.md

@@ -85,8 +85,9 @@ This makes the supported topology repeatable without implying that a container m
 - [x] define a storage-neutral application boundary without changing the single-instance default;
 - [x] implement the internal PostgreSQL authoritative store, ordered event log, read state, stable exports, retention, and leased webhook outbox;
 - [x] implement a cursor-backed per-process realtime relay with ordered replay and lease-loss socket fencing;
+- [x] implement PostgreSQL-owned expiring connection leases with atomic global/per-room caps and crash reclamation;
 - [ ] expose guarded PostgreSQL configuration only after the remaining topology gates are proven;
-- [ ] implement and test cross-worker fan-out, presence, distributed rate limits, and reconnect recovery;
+- [ ] implement and test presence, distributed rate limits, and reconnect recovery, then wire the proven cross-worker fan-out path into the application;
 - [ ] run sustained load/soak and reconnect-storm tests and publish measured limits;
 - [ ] add OpenTelemetry hooks only when an operator needs them, with telemetry disabled by default.
 

docs/MULTI_INSTANCE_ARCHITECTURE.md

@@ -16,6 +16,8 @@ The initial event-log implementation serializes sequence allocation with a trans
 
 Schema v3 uses PostgreSQL database time for room/message/moderation ordering, read cursors, webhook due times and leases, and retention boundaries. The internal store now implements the full storage protocol, including monotonic subject-scoped read state, transactionally stable bounded-memory exports, explicit retention, and a leased transactional webhook outbox. Transaction-scoped capacity locks currently serialize room creation, message mutation/retention, bounded audit insertion, and webhook capacity changes across replicas. Deletion and retention cancel unsent sensitive webhook payloads and scrub message bodies from older durable event and terminal-webhook envelopes before commit. These conservative global locks make correctness inspectable first; the load gate must measure their throughput before v0.13 receives a scale claim.
 
+Schema v4 adds one database-owned lease per admitted WebSocket. A transaction locks the live owning instance and usable room, serializes the capacity decision, removes stale reservations, and atomically checks deployment-wide and per-room caps before inserting. Renewal and release require the owning instance ID. Counts exclude expired socket leases, expired owners, and archived rooms; cleanup returns those stale rows for future presence convergence. Re-registering an expired stable instance ID deletes its former socket rows before renewing the process lease, so a crash/restart cannot revive phantom occupancy. All replicas must use identical capacity settings; public application admission and presence wiring remain release gates.
+
 No Redis dependency is planned for the first supported topology. Redis Pub/Sub is at-most-once, while Streams introduce a second durable system whose commit cannot be atomic with the authoritative database without an additional outbox relay. PostgreSQL already supplies transactions, row locks, advisory locks, `SKIP LOCKED`, and commit-coupled notifications needed by this product's current scale boundary.
 
 ## Why multi-process SQLite is rejected
@@ -43,7 +45,7 @@ An instance that falls behind the retained event window must close its local soc
 PostgreSQL-backed deployments require all of the following:
 
 - **Migrations:** one transaction-scoped advisory lock serializes schema inspection and migration. A newer unsupported schema fails closed.
-- **Connection capacity:** expiring per-socket leases enforce deployment-wide and per-room caps. Heartbeats renew leases; crashed-instance leases expire.
+- **Connection capacity:** implemented schema-v4 per-socket leases enforce deployment-wide and per-room caps with database-time expiry and owner-bound renewal/release. Heartbeats will renew leases when application wiring lands; crashed-instance and archived-room rows are already excluded and reclaimable.
 - **Presence:** join/leave transitions derive from connection leases. A leader-elected sweeper emits bounded expiry transitions for crashed instances. Presence remains best effort and carries no authorization meaning.
 - **Rate limits:** atomic time-bucket counters enforce deployment-wide subject/client limits. Database time defines bucket boundaries so host clock skew cannot multiply quotas.
 - **Typing:** transition events use the durable event path but expire automatically and are not retained as chat history or audit content.
@@ -71,7 +73,7 @@ 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;
 - 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;
+- global and per-room connection caps plus rate limits hold across processes (the storage-level concurrent connection-cap case is implemented; real-process admission and rate-limit gates remain);
 - 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;

samsarix_chat_engine/postgres.py

@@ -23,7 +23,7 @@
 from psycopg.types.json import Jsonb
 from psycopg_pool import AsyncConnectionPool, PoolTimeout
 
-POSTGRES_SCHEMA_VERSION = 3
+POSTGRES_SCHEMA_VERSION = 4
 POSTGRES_MIGRATION_LOCK_ID = 7_495_346_927_831_819_041
 POSTGRES_EVENT_SEQUENCE_LOCK_ID = 7_495_346_927_831_819_042
 REALTIME_CHANNEL = "samsarix_realtime_v1"
@@ -198,6 +198,24 @@ async def register_instance(self, instance_id: str, *, lease_seconds: int) -> in
 
         _validate_instance(instance_id, lease_seconds)
         async with self.transaction() as connection:
+            cursor = await connection.execute(
+                """
+                SELECT lease_expires_at <= clock_timestamp()
+                FROM public.samsarix_instance_cursors
+                WHERE instance_id = %s
+                FOR UPDATE
+                """,
+                (instance_id,),
+            )
+            existing = await cursor.fetchone()
+            if existing is not None and bool(existing[0]):
+                # A stable instance ID may be reused after a crash. Its old
+                # sockets no longer exist and must not become live again when
+                # the owner lease is renewed.
+                await connection.execute(
+                    "DELETE FROM public.samsarix_connection_leases WHERE instance_id = %s",
+                    (instance_id,),
+                )
             cursor = await connection.execute(
                 """
                 INSERT INTO public.samsarix_instance_cursors (
@@ -411,6 +429,49 @@ async def _initialize_schema(self) -> None:
                     )
                     """
                 )
+                await connection.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS public.samsarix_connection_leases (
+                        connection_id TEXT PRIMARY KEY CHECK (
+                            char_length(connection_id) BETWEEN 1 AND 128
+                            AND connection_id ~ '^[A-Za-z0-9][A-Za-z0-9._:-]*$'
+                        ),
+                        instance_id TEXT NOT NULL REFERENCES public.samsarix_instance_cursors(instance_id)
+                            ON DELETE CASCADE,
+                        room_id TEXT NOT NULL REFERENCES public.samsarix_rooms(id) ON DELETE CASCADE,
+                        username TEXT NOT NULL CHECK (char_length(username) BETWEEN 1 AND 64),
+                        subject TEXT CHECK (subject IS NULL OR char_length(subject) BETWEEN 1 AND 64),
+                        lease_expires_at TIMESTAMPTZ NOT NULL,
+                        created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+                        updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+                        CHECK (lease_expires_at > created_at)
+                    )
+                    """
+                )
+                await connection.execute(
+                    """
+                    CREATE INDEX IF NOT EXISTS samsarix_connection_leases_expiry
+                    ON public.samsarix_connection_leases (lease_expires_at, connection_id)
+                    """
+                )
+                await connection.execute(
+                    """
+                    CREATE INDEX IF NOT EXISTS samsarix_connection_leases_room
+                    ON public.samsarix_connection_leases (room_id, lease_expires_at, connection_id)
+                    """
+                )
+                await connection.execute(
+                    """
+                    CREATE INDEX IF NOT EXISTS samsarix_connection_leases_instance
+                    ON public.samsarix_connection_leases (instance_id, lease_expires_at, connection_id)
+                    """
+                )
+                await connection.execute(
+                    """
+                    CREATE INDEX IF NOT EXISTS samsarix_connection_leases_member
+                    ON public.samsarix_connection_leases (room_id, subject, lease_expires_at)
+                    """
+                )
                 await connection.execute(
                     """
                     CREATE TABLE IF NOT EXISTS public.samsarix_messages (