Derive PostgreSQL presence from connection leases

CHANGELOG.md

@@ -15,6 +15,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma
 - 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.
 - PostgreSQL schema v5 and internal deployment-wide message, search, and typing rate buckets with atomic per-identity consumption, database-time boundaries, bounded active cardinality, and raw-key minimization.
 - PostgreSQL schema v6 and internal connection-bound typing state with transition-only starts, refresh without event storms, explicit stops, database-time expiry, and bounded concurrent sweeping into durable coordination events.
+- PostgreSQL schema v7 and lease-derived presence transitions with exact join/explicit-leave counts, bounded crashed-process sweeping, typing-before-leave ordering, and process-generation fencing for safe stable-ID restarts.
 
 ### Security and operations
 
@@ -25,9 +26,11 @@ 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.
+- Expired process IDs rotate their generation token on re-registration, preventing a restarted replica from reviving phantom occupancy while leaving stale socket rows available for bounded presence convergence. Archived, expired, and generation-mismatched reservations stop consuming capacity even before physical cleanup.
 - Distributed rate buckets persist only a scope-separated SHA-256 digest of the caller key. The digest reduces routine identity exposure but is not anonymization; database access and retention still require normal privacy controls.
 - Internal typing coordination events retain an opaque origin connection ID so future application wiring can exclude the sender. The public realtime relay does not forward these events until that exclusion contract is implemented; clients must continue treating advertised expiry as the stop-event backstop.
+- A restarted process rotates its database generation token after lease expiry. Connection, typing, count, renewal, and cleanup queries require the matching generation so stale sockets cannot regain capacity, activity, or presence merely because an operator reused a stable instance name.
+- The asymmetric-authentication, test, and development dependency ranges now require `cryptography>=50,<51`, excluding the `cryptography>=44.0.0,<50.0.0` range affected by `PYSEC-2026-3552`.
 
 ## 0.12.0 — 2026-08-02
 

ROADMAP.md

@@ -88,8 +88,9 @@ This makes the supported topology repeatable without implying that a container m
 - [x] implement PostgreSQL-owned expiring connection leases with atomic global/per-room caps and crash reclamation;
 - [x] implement bounded PostgreSQL-owned message, search, and typing rate buckets with database-time windows;
 - [x] implement connection-bound PostgreSQL typing transitions, refresh, and bounded expiry sweeping;
+- [x] derive join/leave presence from connection leases with generation-fenced restart and crash convergence;
 - [ ] expose guarded PostgreSQL configuration only after the remaining topology gates are proven;
-- [ ] implement and test lease-derived presence and reconnect recovery, then wire the proven cross-worker fan-out, typing, and rate-control paths into the application;
+- [ ] implement reconnect recovery, then wire the proven cross-worker fan-out, presence, typing, and rate-control paths 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,12 +16,14 @@ 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.
+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 presence convergence. Schema v7 supersedes the original stable-ID restart behavior with generation fencing so expired rows remain available for leave-event recovery without becoming live again. All replicas must use identical capacity settings; public application admission and presence wiring remain release gates.
 
 Schema v5 adds fixed-window counters for message, search, and typing controls. PostgreSQL time chooses each boundary, and an atomic row update admits no more than the configured count for a scope/key across all replicas. Existing identities contend only on their own row; creation of new identity buckets uses a separate advisory lock to prune expiry and enforce a hard cardinality bound. Raw subjects and client addresses are not stored: a scope-separated SHA-256 digest is persisted instead. That digest is data minimization, not anonymization, because predictable identities may still be guessed. All replicas must use identical limits, window lengths, and bucket capacity. Fixed windows can admit traffic on both sides of a boundary; load tests must validate whether that declared behavior is sufficient before public wiring.
 
 Schema v6 binds ephemeral typing rows to live connection leases. An inactive start writes one durable `typing.started` coordination event; repeated starts refresh the database-time deadline without producing an event storm. Explicit stop and a bounded `SKIP LOCKED` expiry sweep delete state before writing `typing.stopped` in the same transaction. Events retain an opaque origin connection ID so later relay wiring can exclude the sender, but the public relay intentionally ignores typing events until that exclusion is implemented. Connection deletion cascades typing state; a missed stop after a crash remains safe because clients already expire typing locally from `expires_in`. Typing rows and events are operational coordination, never chat history or audit records.
 
+Schema v7 derives presence transitions from committed connection changes. Admission serializes capacity, stale cleanup, insertion, the resulting live room count, and `presence.joined`; explicit owned release serializes deletion, optional `typing.stopped`, the decremented count, and `presence.left`. A bounded sweeper uses `SKIP LOCKED` to reclaim crashed sockets and emits typing stops before leaves. Each process registration has a UUID generation: renewing a live process preserves it, while re-registering an expired stable ID rotates it. Every connection stores the admitting generation, and counts, renewal, typing, and stale detection require a match. Old sockets therefore remain non-live and sweepable instead of being silently revived or deleted before convergence. Presence is still best effort and non-authoritative; durable coordination makes cross-process delivery recoverable but reconnecting clients must accept a fresh room count.
+
 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
@@ -50,7 +52,7 @@ 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:** 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.
+- **Presence:** implemented schema-v7 joins/leaves derive from connection transactions, and a bounded generation-aware sweeper emits expiry transitions for crashed instances. Presence remains best effort, carries no authorization meaning, and stays off the public relay until sender exclusion is wired.
 - **Rate limits:** implemented schema-v5 atomic time-bucket counters enforce deployment-wide subject/client limits for message, search, and typing scopes. Database time defines boundaries so host clock skew cannot multiply quotas; public request-path wiring remains gated.
 - **Typing:** implemented schema-v6 state uses transition-only durable events, database-time refresh, and bounded expiry sweeping. It is not chat history or audit content; public relay/origin-exclusion wiring remains gated.
 - **Moderation teardown:** room archive and member-ban events reach every instance, which closes matching local sockets deterministically.

pyproject.toml

@@ -36,13 +36,13 @@ dependencies = [
 
 [project.optional-dependencies]
 asymmetric-auth = [
-    "cryptography>=46,<50",
+    "cryptography>=50,<51",
 ]
 postgres = [
     "psycopg[binary,pool]>=3.3,<4",
 ]
 test = [
-    "cryptography>=46,<50",
+    "cryptography>=50,<51",
     "httpx2>=2,<3",
     "pytest>=8,<10",
     "pytest-asyncio>=0.24,<2.0",
@@ -51,7 +51,7 @@ test = [
 ]
 dev = [
     "build>=1.2,<2.0",
-    "cryptography>=46,<50",
+    "cryptography>=50,<51",
     "httpx2>=2,<3",
     "mypy>=1.11,<2.0",
     "pip-audit>=2.10,<3.0",

samsarix_chat_engine/postgres.py

@@ -23,7 +23,7 @@
 from psycopg.types.json import Jsonb
 from psycopg_pool import AsyncConnectionPool, PoolTimeout
 
-POSTGRES_SCHEMA_VERSION = 6
+POSTGRES_SCHEMA_VERSION = 7
 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,36 +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 (
-                    instance_id, last_sequence, lease_expires_at, updated_at
+                    instance_id, generation, last_sequence, lease_expires_at, updated_at
                 )
                 VALUES (
                     %s,
+                    gen_random_uuid(),
                     (SELECT COALESCE(MAX(sequence), 0) FROM public.samsarix_realtime_events),
                     clock_timestamp() + make_interval(secs => %s),
                     clock_timestamp()
                 )
                 ON CONFLICT (instance_id) DO UPDATE SET
+                    generation = CASE
+                        WHEN samsarix_instance_cursors.lease_expires_at <= clock_timestamp()
+                        THEN EXCLUDED.generation
+                        ELSE samsarix_instance_cursors.generation
+                    END,
                     lease_expires_at = EXCLUDED.lease_expires_at,
                     updated_at = clock_timestamp()
                 RETURNING last_sequence
@@ -394,6 +382,7 @@ async def _initialize_schema(self) -> None:
                             char_length(instance_id) BETWEEN 1 AND 128
                             AND instance_id ~ '^[A-Za-z0-9][A-Za-z0-9._:-]*$'
                         ),
+                        generation UUID NOT NULL DEFAULT gen_random_uuid(),
                         last_sequence BIGINT NOT NULL CHECK (last_sequence >= 0),
                         lease_expires_at TIMESTAMPTZ NOT NULL,
                         updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
@@ -438,6 +427,7 @@ async def _initialize_schema(self) -> None:
                         ),
                         instance_id TEXT NOT NULL REFERENCES public.samsarix_instance_cursors(instance_id)
                             ON DELETE CASCADE,
+                        instance_generation UUID NOT NULL,
                         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),
@@ -448,6 +438,47 @@ async def _initialize_schema(self) -> None:
                     )
                     """
                 )
+                await connection.execute(
+                    """
+                    ALTER TABLE public.samsarix_instance_cursors
+                    ADD COLUMN IF NOT EXISTS generation UUID
+                    """
+                )
+                await connection.execute(
+                    """
+                    UPDATE public.samsarix_instance_cursors
+                    SET generation = gen_random_uuid()
+                    WHERE generation IS NULL
+                    """
+                )
+                await connection.execute(
+                    """
+                    ALTER TABLE public.samsarix_instance_cursors
+                    ALTER COLUMN generation SET DEFAULT gen_random_uuid(),
+                    ALTER COLUMN generation SET NOT NULL
+                    """
+                )
+                await connection.execute(
+                    """
+                    ALTER TABLE public.samsarix_connection_leases
+                    ADD COLUMN IF NOT EXISTS instance_generation UUID
+                    """
+                )
+                await connection.execute(
+                    """
+                    UPDATE public.samsarix_connection_leases AS lease
+                    SET instance_generation = owner.generation
+                    FROM public.samsarix_instance_cursors AS owner
+                    WHERE owner.instance_id = lease.instance_id
+                      AND lease.instance_generation IS NULL
+                    """
+                )
+                await connection.execute(
+                    """
+                    ALTER TABLE public.samsarix_connection_leases
+                    ALTER COLUMN instance_generation SET NOT NULL
+                    """
+                )
                 await connection.execute(
                     """
                     CREATE INDEX IF NOT EXISTS samsarix_connection_leases_expiry
@@ -460,10 +491,14 @@ async def _initialize_schema(self) -> None:
                     ON public.samsarix_connection_leases (room_id, lease_expires_at, connection_id)
                     """
                 )
+                if current_version < 7:
+                    await connection.execute("DROP INDEX IF EXISTS public.samsarix_connection_leases_instance")
                 await connection.execute(
                     """
-                    CREATE INDEX IF NOT EXISTS samsarix_connection_leases_instance
-                    ON public.samsarix_connection_leases (instance_id, lease_expires_at, connection_id)
+                    CREATE INDEX IF NOT EXISTS samsarix_connection_leases_instance_generation
+                    ON public.samsarix_connection_leases (
+                        instance_id, instance_generation, lease_expires_at, connection_id
+                    )
                     """
                 )
                 await connection.execute(