From d18d1d4f033f9dae20fabbbd8f1545fa8d36491d Mon Sep 17 00:00:00 2001 From: Deathcharge Date: Sat, 8 Aug 2026 04:48:10 -0400 Subject: [PATCH 1/3] feat(postgres): derive presence from connection leases --- CHANGELOG.md | 4 +- ROADMAP.md | 3 +- docs/MULTI_INSTANCE_ARCHITECTURE.md | 6 +- samsarix_chat_engine/postgres.py | 77 ++++-- samsarix_chat_engine/postgres_connections.py | 253 ++++++++++++++++--- samsarix_chat_engine/postgres_typing.py | 1 + tests/test_postgres_presence.py | 183 ++++++++++++++ tests/test_postgres_store.py | 95 ++++++- tests/test_postgres_typing.py | 5 +- 9 files changed, 561 insertions(+), 66 deletions(-) create mode 100644 tests/test_postgres_presence.py diff --git a/CHANGELOG.md b/CHANGELOG.md index efdd885..1ca20b2 100644 --- a/CHANGELOG.md +++ b/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,10 @@ 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. ## 0.12.0 — 2026-08-02 diff --git a/ROADMAP.md b/ROADMAP.md index 233d8f7..b4e8b4c 100644 --- a/ROADMAP.md +++ b/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. diff --git a/docs/MULTI_INSTANCE_ARCHITECTURE.md b/docs/MULTI_INSTANCE_ARCHITECTURE.md index 35593f5..5b0c605 100644 --- a/docs/MULTI_INSTANCE_ARCHITECTURE.md +++ b/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. diff --git a/samsarix_chat_engine/postgres.py b/samsarix_chat_engine/postgres.py index 228e9ff..ed40019 100644 --- a/samsarix_chat_engine/postgres.py +++ b/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 @@ -462,8 +493,10 @@ async def _initialize_schema(self) -> None: ) 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( diff --git a/samsarix_chat_engine/postgres_connections.py b/samsarix_chat_engine/postgres_connections.py index dda5ed8..1df17b5 100644 --- a/samsarix_chat_engine/postgres_connections.py +++ b/samsarix_chat_engine/postgres_connections.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from datetime import datetime from typing import Any, cast +from uuid import UUID from psycopg import AsyncConnection from psycopg.errors import UniqueViolation @@ -38,6 +39,7 @@ class ConnectionLease: connection_id: str instance_id: str + instance_generation: UUID room_id: str username: str subject: str | None @@ -53,6 +55,18 @@ class ConnectionCounts: room: int +@dataclass(frozen=True, slots=True) +class PresenceTransition: + """One committed join/leave transition derived from a connection lease.""" + + active: bool + connection_id: str + room_id: str + username: str + active_connections: int + sequence: int + + class PostgresConnectionRegistry: """Atomically reserve global and per-room connection capacity across processes.""" @@ -92,15 +106,17 @@ async def try_acquire( async with self.foundation.transaction() as connection: cursor = await connection.execute( """ - SELECT 1 + SELECT generation FROM public.samsarix_instance_cursors WHERE instance_id = %s AND lease_expires_at > clock_timestamp() FOR SHARE """, (instance_id,), ) - if await cursor.fetchone() is None: + owner = await cursor.fetchone() + if owner is None: raise InstanceLeaseError("instance lease is missing or expired") + instance_generation = cast(UUID, owner[0]) cursor = await connection.execute( """ @@ -116,7 +132,8 @@ async def try_acquire( raise ConnectionRoomUnavailableError("connection room is missing or archived") await connection.execute("SELECT pg_advisory_xact_lock(%s)", (POSTGRES_CONNECTION_CAP_LOCK_ID,)) - await _delete_expired(connection) + stale = await _delete_expired(connection, limit=1_000, prioritize_connection_id=connection_id) + await self._append_departures(connection, stale) cursor = await connection.execute( """ @@ -126,6 +143,7 @@ async def try_acquire( FROM public.samsarix_connection_leases AS lease JOIN public.samsarix_instance_cursors AS owner ON owner.instance_id = lease.instance_id + AND owner.generation = lease.instance_generation JOIN public.samsarix_rooms AS room ON room.id = lease.room_id WHERE lease.lease_expires_at > clock_timestamp() @@ -144,34 +162,48 @@ async def try_acquire( cursor = await connection.execute( """ INSERT INTO public.samsarix_connection_leases ( - connection_id, instance_id, room_id, username, subject, lease_expires_at + connection_id, instance_id, instance_generation, room_id, + username, subject, lease_expires_at ) SELECT - %s, %s, %s, %s, %s, + %s, %s, %s, %s, %s, %s, clock_timestamp() + make_interval(secs => %s) WHERE EXISTS ( SELECT 1 FROM public.samsarix_instance_cursors WHERE instance_id = %s + AND generation = %s AND lease_expires_at > clock_timestamp() ) RETURNING - connection_id, instance_id, room_id, username, subject, + connection_id, instance_id, instance_generation, room_id, username, subject, lease_expires_at, created_at """, ( connection_id, instance_id, + instance_generation, room_id, username, subject, self.lease_seconds, instance_id, + instance_generation, ), ) except UniqueViolation: raise ConnectionLeaseError("connection ID is already leased") from None row = await cursor.fetchone() + if row is not None: + active_connections = await _room_active_count(connection, room_id) + await self._append_presence( + connection, + active=True, + connection_id=connection_id, + room_id=room_id, + username=username, + active_connections=active_connections, + ) if row is None: raise InstanceLeaseError("instance lease expired before connection reservation") return _lease_from_row(row) @@ -193,6 +225,7 @@ async def renew(self, *, connection_id: str, instance_id: str) -> datetime: AND lease.instance_id = %s AND lease.lease_expires_at > clock_timestamp() AND owner.instance_id = lease.instance_id + AND owner.generation = lease.instance_generation AND owner.lease_expires_at > clock_timestamp() AND room.id = lease.room_id AND room.archived_at IS NULL @@ -211,14 +244,40 @@ async def release(self, *, connection_id: str, instance_id: str) -> bool: _validate_connection_id(connection_id) _validate_instance_id(instance_id) async with self.foundation.transaction() as connection: + await connection.execute("SELECT pg_advisory_xact_lock(%s)", (POSTGRES_CONNECTION_CAP_LOCK_ID,)) cursor = await connection.execute( """ - DELETE FROM public.samsarix_connection_leases - WHERE connection_id = %s AND instance_id = %s + SELECT lease.room_id, lease.username, + EXISTS ( + SELECT 1 FROM public.samsarix_typing_states AS typing + WHERE typing.connection_id = lease.connection_id + ) + FROM public.samsarix_connection_leases AS lease + WHERE lease.connection_id = %s AND lease.instance_id = %s + FOR UPDATE """, (connection_id, instance_id), ) - return cursor.rowcount == 1 + row = await cursor.fetchone() + if row is None: + return False + room_id, username, was_typing = str(row[0]), str(row[1]), bool(row[2]) + await connection.execute( + "DELETE FROM public.samsarix_connection_leases WHERE connection_id = %s", + (connection_id,), + ) + if was_typing: + await self._append_typing_stopped(connection, connection_id, room_id, username) + active_connections = await _room_active_count(connection, room_id) + await self._append_presence( + connection, + active=False, + connection_id=connection_id, + room_id=room_id, + username=username, + active_connections=active_connections, + ) + return True async def counts(self, *, room_id: str) -> ConnectionCounts: """Return live occupancy without allowing stale rows to inflate it.""" @@ -234,6 +293,7 @@ async def counts(self, *, room_id: str) -> ConnectionCounts: FROM public.samsarix_connection_leases AS lease JOIN public.samsarix_instance_cursors AS owner ON owner.instance_id = lease.instance_id + AND owner.generation = lease.instance_generation JOIN public.samsarix_rooms AS room ON room.id = lease.room_id WHERE lease.lease_expires_at > clock_timestamp() @@ -245,49 +305,168 @@ async def counts(self, *, room_id: str) -> ConnectionCounts: row = await cursor.fetchone() return ConnectionCounts(total=int(row[0]), room=int(row[1])) if row is not None else ConnectionCounts(0, 0) - async def reap_expired(self) -> list[ConnectionLease]: - """Delete and return rows whose socket/owner expired or whose room was archived.""" + async def reap_expired(self, *, limit: int = 100) -> list[PresenceTransition]: + """Delete a bounded stale batch and emit best-effort leave transitions.""" + if not 1 <= limit <= 1_000: + raise ValueError("connection expiry batch must be between 1 and 1000") async with self.foundation.transaction() as connection: await connection.execute("SELECT pg_advisory_xact_lock(%s)", (POSTGRES_CONNECTION_CAP_LOCK_ID,)) - rows = await _delete_expired(connection) - return [_lease_from_row(row) for row in rows] + rows = await _delete_expired(connection, limit=limit) + transitions = await self._append_departures(connection, rows) + return transitions + async def _append_departures( + self, + connection: AsyncConnection[tuple[Any, ...]], + rows: list[tuple[Any, ...]], + ) -> list[PresenceTransition]: + transitions: list[PresenceTransition] = [] + for row in sorted(rows, key=lambda item: str(item[0])): + connection_id, room_id, username = str(row[0]), str(row[3]), str(row[4]) + if bool(row[8]): + await self._append_typing_stopped(connection, connection_id, room_id, username) + active_connections = await _room_active_count(connection, room_id) + sequence = await self._append_presence( + connection, + active=False, + connection_id=connection_id, + room_id=room_id, + username=username, + active_connections=active_connections, + ) + transitions.append( + PresenceTransition(False, connection_id, room_id, username, active_connections, sequence) + ) + return transitions -async def _delete_expired(connection: AsyncConnection[tuple[Any, ...]]) -> list[tuple[Any, ...]]: + async def _append_presence( + self, + connection: AsyncConnection[tuple[Any, ...]], + *, + active: bool, + connection_id: str, + room_id: str, + username: str, + active_connections: int, + ) -> int: + event_type = "presence.joined" if active else "presence.left" + return await self.foundation.append_event( + connection, + room_id=room_id, + event_type=event_type, + payload={ + "type": event_type, + "username": username, + "active_connections": active_connections, + "origin_connection_id": connection_id, + }, + ) + + async def _append_typing_stopped( + self, + connection: AsyncConnection[tuple[Any, ...]], + connection_id: str, + room_id: str, + username: str, + ) -> int: + return await self.foundation.append_event( + connection, + room_id=room_id, + event_type="typing.stopped", + payload={ + "type": "typing.stopped", + "username": username, + "origin_connection_id": connection_id, + }, + ) + + +async def _delete_expired( + connection: AsyncConnection[tuple[Any, ...]], + *, + limit: int, + prioritize_connection_id: str | None = None, +) -> list[tuple[Any, ...]]: cursor = await connection.execute( """ - DELETE FROM public.samsarix_connection_leases AS lease - WHERE lease.lease_expires_at <= clock_timestamp() - OR EXISTS ( - SELECT 1 - FROM public.samsarix_instance_cursors AS owner - WHERE owner.instance_id = lease.instance_id - AND owner.lease_expires_at <= clock_timestamp() - ) - OR EXISTS ( - SELECT 1 - FROM public.samsarix_rooms AS room - WHERE room.id = lease.room_id - AND room.archived_at IS NOT NULL - ) - RETURNING - lease.connection_id, lease.instance_id, lease.room_id, lease.username, - lease.subject, lease.lease_expires_at, lease.created_at - """ + WITH stale AS MATERIALIZED ( + SELECT + lease.connection_id, lease.instance_id, lease.instance_generation, + lease.room_id, lease.username, lease.subject, + lease.lease_expires_at, lease.created_at, + EXISTS ( + SELECT 1 FROM public.samsarix_typing_states AS typing + WHERE typing.connection_id = lease.connection_id + ) AS was_typing + FROM public.samsarix_connection_leases AS lease + WHERE lease.lease_expires_at <= clock_timestamp() + OR EXISTS ( + SELECT 1 + FROM public.samsarix_instance_cursors AS owner + WHERE owner.instance_id = lease.instance_id + AND ( + owner.lease_expires_at <= clock_timestamp() + OR owner.generation <> lease.instance_generation + ) + ) + OR EXISTS ( + SELECT 1 + FROM public.samsarix_rooms AS room + WHERE room.id = lease.room_id + AND room.archived_at IS NOT NULL + ) + ORDER BY (lease.connection_id = %s) DESC, lease.lease_expires_at, lease.connection_id + LIMIT %s + FOR UPDATE OF lease SKIP LOCKED + ), deleted AS ( + DELETE FROM public.samsarix_connection_leases AS lease + USING stale + WHERE lease.connection_id = stale.connection_id + RETURNING lease.connection_id + ) + SELECT + stale.connection_id, stale.instance_id, stale.instance_generation, + stale.room_id, stale.username, stale.subject, + stale.lease_expires_at, stale.created_at, stale.was_typing + FROM stale + JOIN deleted USING (connection_id) + """, + (prioritize_connection_id, limit), ) return list(await cursor.fetchall()) +async def _room_active_count(connection: AsyncConnection[tuple[Any, ...]], room_id: str) -> int: + cursor = await connection.execute( + """ + SELECT COUNT(*) + FROM public.samsarix_connection_leases AS lease + JOIN public.samsarix_instance_cursors AS owner + ON owner.instance_id = lease.instance_id + AND owner.generation = lease.instance_generation + JOIN public.samsarix_rooms AS room ON room.id = lease.room_id + WHERE lease.room_id = %s + AND lease.lease_expires_at > clock_timestamp() + AND owner.lease_expires_at > clock_timestamp() + AND room.archived_at IS NULL + """, + (room_id,), + ) + row = await cursor.fetchone() + return int(row[0]) if row is not None else 0 + + def _lease_from_row(row: tuple[Any, ...]) -> ConnectionLease: return ConnectionLease( connection_id=str(row[0]), instance_id=str(row[1]), - room_id=str(row[2]), - username=str(row[3]), - subject=None if row[4] is None else str(row[4]), - lease_expires_at=cast(datetime, row[5]), - created_at=cast(datetime, row[6]), + instance_generation=cast(UUID, row[2]), + room_id=str(row[3]), + username=str(row[4]), + subject=None if row[5] is None else str(row[5]), + lease_expires_at=cast(datetime, row[6]), + created_at=cast(datetime, row[7]), ) diff --git a/samsarix_chat_engine/postgres_typing.py b/samsarix_chat_engine/postgres_typing.py index 2c626e5..197500b 100644 --- a/samsarix_chat_engine/postgres_typing.py +++ b/samsarix_chat_engine/postgres_typing.py @@ -49,6 +49,7 @@ async def start(self, *, connection_id: str, instance_id: str) -> TypingTransiti FROM public.samsarix_connection_leases AS lease JOIN public.samsarix_instance_cursors AS owner ON owner.instance_id = lease.instance_id + AND owner.generation = lease.instance_generation JOIN public.samsarix_rooms AS room ON room.id = lease.room_id WHERE lease.connection_id = %s diff --git a/tests/test_postgres_presence.py b/tests/test_postgres_presence.py new file mode 100644 index 0000000..0034982 --- /dev/null +++ b/tests/test_postgres_presence.py @@ -0,0 +1,183 @@ +# Copyright (c) 2026 Samsarix LLC +# SPDX-License-Identifier: MPL-2.0 +"""Live tests for lease-derived PostgreSQL presence convergence.""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("psycopg") + +from samsarix_chat_engine.models import RoomCreate # noqa: E402 +from samsarix_chat_engine.postgres_connections import ConnectionCounts, PostgresConnectionRegistry # noqa: E402 +from samsarix_chat_engine.postgres_store import PostgresChatStore # noqa: E402 +from samsarix_chat_engine.postgres_typing import PostgresTypingRegistry, TypingStateError # noqa: E402 + +pytestmark = pytest.mark.postgres + + +def _store(conninfo: str) -> PostgresChatStore: + return PostgresChatStore( + conninfo, + max_rooms=5, + max_stored_messages=10, + max_stored_messages_per_room=10, + ) + + +@pytest.mark.asyncio +async def test_join_and_owned_release_emit_exact_room_counts(clean_postgres_database: str) -> None: + store = _store(clean_postgres_database) + await store.initialize() + try: + await store.create_room(RoomCreate(id="general", name="General")) + await store.foundation.register_instance("node-a", lease_seconds=30) + await store.foundation.register_instance("node-b", lease_seconds=30) + await store.foundation.register_instance("observer", lease_seconds=30) + registry = PostgresConnectionRegistry( + store.foundation, + max_connections=5, + max_connections_per_room=5, + ) + assert await registry.try_acquire( + connection_id="socket-a", + instance_id="node-a", + room_id="general", + username="alice", + subject="alice", + ) + assert await registry.try_acquire( + connection_id="socket-b", + instance_id="node-b", + room_id="general", + username="bob", + subject="bob", + ) + assert not await registry.release(connection_id="socket-a", instance_id="node-b") + assert await registry.release(connection_id="socket-a", instance_id="node-a") + + events = await store.foundation.read_events("observer") + assert [event.event_type for event in events] == [ + "presence.joined", + "presence.joined", + "presence.left", + ] + assert [event.payload["active_connections"] for event in events] == [1, 2, 1] + assert [event.payload["username"] for event in events] == ["alice", "bob", "alice"] + assert events[0].payload["origin_connection_id"] == "socket-a" + finally: + await store.close() + + +@pytest.mark.asyncio +async def test_same_id_restart_rotates_generation_and_cannot_revive_stale_socket( + clean_postgres_database: str, +) -> None: + store = _store(clean_postgres_database) + await store.initialize() + try: + await store.create_room(RoomCreate(id="general", name="General")) + await store.foundation.register_instance("stable-node", lease_seconds=30) + registry = PostgresConnectionRegistry( + store.foundation, + max_connections=1, + max_connections_per_room=1, + ) + old = await registry.try_acquire( + connection_id="socket-old", + instance_id="stable-node", + room_id="general", + username="alice", + subject="alice", + ) + assert old is not None + await store.foundation.register_instance("observer", lease_seconds=30) + async with store.foundation.transaction() as connection: + await connection.execute( + """ + UPDATE public.samsarix_instance_cursors + SET lease_expires_at = clock_timestamp() - interval '1 second' + WHERE instance_id = 'stable-node' + """ + ) + await store.foundation.register_instance("stable-node", lease_seconds=30) + async with store.foundation.transaction() as connection: + cursor = await connection.execute( + "SELECT generation FROM public.samsarix_instance_cursors WHERE instance_id = 'stable-node'" + ) + row = await cursor.fetchone() + assert row is not None and row[0] != old.instance_generation + assert await registry.counts(room_id="general") == ConnectionCounts(0, 0) + with pytest.raises(TypingStateError, match="unavailable"): + await PostgresTypingRegistry(store.foundation).start( + connection_id="socket-old", + instance_id="stable-node", + ) + + assert await registry.try_acquire( + connection_id="socket-new", + instance_id="stable-node", + room_id="general", + username="alice", + subject="alice", + ) + events = await store.foundation.read_events("observer") + assert [event.event_type for event in events] == ["presence.left", "presence.joined"] + assert [event.payload["active_connections"] for event in events] == [0, 1] + finally: + await store.close() + + +@pytest.mark.asyncio +async def test_crash_sweep_is_bounded_and_stops_typing_before_presence(clean_postgres_database: str) -> None: + store = _store(clean_postgres_database) + await store.initialize() + try: + await store.create_room(RoomCreate(id="general", name="General")) + await store.foundation.register_instance("node-a", lease_seconds=30) + registry = PostgresConnectionRegistry( + store.foundation, + max_connections=5, + max_connections_per_room=5, + ) + typing = PostgresTypingRegistry(store.foundation) + for index in range(2): + assert await registry.try_acquire( + connection_id=f"socket-{index}", + instance_id="node-a", + room_id="general", + username=f"user-{index}", + subject=None, + ) + await store.foundation.register_instance("observer", lease_seconds=30) + for index in range(2): + assert await typing.start(connection_id=f"socket-{index}", instance_id="node-a") + async with store.foundation.transaction() as connection: + await connection.execute( + """ + UPDATE public.samsarix_instance_cursors + SET lease_expires_at = clock_timestamp() - interval '1 second' + WHERE instance_id = 'node-a' + """ + ) + + assert len(await registry.reap_expired(limit=1)) == 1 + assert len(await registry.reap_expired(limit=1)) == 1 + assert await registry.reap_expired() == [] + events = await store.foundation.read_events("observer") + assert [event.event_type for event in events] == [ + "typing.started", + "typing.started", + "typing.stopped", + "presence.left", + "typing.stopped", + "presence.left", + ] + left_counts = [ + event.payload.get("active_connections") for event in events if event.event_type == "presence.left" + ] + assert left_counts == [0, 0] + with pytest.raises(ValueError, match="between 1 and 1000"): + await registry.reap_expired(limit=0) + finally: + await store.close() diff --git a/tests/test_postgres_store.py b/tests/test_postgres_store.py index 957f99b..3cea6c2 100644 --- a/tests/test_postgres_store.py +++ b/tests/test_postgres_store.py @@ -96,7 +96,7 @@ async def test_schema_v2_migrates_transactionally_and_widens_event_payloads( service = _store(clean_postgres_database) await service.initialize() try: - assert await service.foundation.schema_version() == POSTGRES_SCHEMA_VERSION == 6 + assert await service.foundation.schema_version() == POSTGRES_SCHEMA_VERSION == 7 assert await service.check_ready() assert await service.list_rooms() == [] async with service.foundation.transaction() as connection: @@ -110,6 +110,99 @@ async def test_schema_v2_migrates_transactionally_and_widens_event_payloads( await service.close() +@pytest.mark.asyncio +async def test_schema_v6_backfills_matching_instance_generations( + clean_postgres_database: str, +) -> None: + async with await psycopg.AsyncConnection.connect(clean_postgres_database, autocommit=True) as connection: + await connection.execute( + """ + CREATE TABLE public.samsarix_schema_metadata ( + singleton BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (singleton), + version INTEGER NOT NULL CHECK (version > 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() + ) + """ + ) + await connection.execute("INSERT INTO public.samsarix_schema_metadata (singleton, version) VALUES (TRUE, 6)") + await connection.execute( + """ + CREATE TABLE public.samsarix_instance_cursors ( + instance_id TEXT PRIMARY KEY, + last_sequence BIGINT NOT NULL CHECK (last_sequence >= 0), + lease_expires_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() + ) + """ + ) + await connection.execute( + """ + CREATE TABLE public.samsarix_rooms ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), + archived_at TIMESTAMPTZ, + frozen_at TIMESTAMPTZ + ) + """ + ) + await connection.execute( + """ + CREATE TABLE public.samsarix_connection_leases ( + connection_id TEXT PRIMARY KEY, + 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, + subject TEXT, + 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( + """ + INSERT INTO public.samsarix_instance_cursors ( + instance_id, last_sequence, lease_expires_at + ) VALUES ('legacy-node', 0, clock_timestamp() + interval '30 seconds') + """ + ) + await connection.execute("INSERT INTO public.samsarix_rooms (id, name) VALUES ('general', 'General')") + await connection.execute( + """ + INSERT INTO public.samsarix_connection_leases ( + connection_id, instance_id, room_id, username, subject, lease_expires_at + ) VALUES ( + 'legacy-socket', 'legacy-node', 'general', 'alice', 'alice', + clock_timestamp() + interval '30 seconds' + ) + """ + ) + + service = _store(clean_postgres_database) + await service.initialize() + try: + assert await service.foundation.schema_version() == POSTGRES_SCHEMA_VERSION == 7 + async with service.foundation.transaction() as connection: + cursor = await connection.execute( + """ + SELECT owner.generation, lease.instance_generation + FROM public.samsarix_connection_leases AS lease + JOIN public.samsarix_instance_cursors AS owner + ON owner.instance_id = lease.instance_id + WHERE lease.connection_id = 'legacy-socket' + """ + ) + generations = await cursor.fetchone() + assert generations is not None + assert generations[0] is not None and generations[0] == generations[1] + finally: + await service.close() + + @pytest.mark.asyncio async def test_room_lifecycle_audit_and_events_are_atomic(store: PostgresChatStore) -> None: await store.foundation.register_instance("room-observer", lease_seconds=30) diff --git a/tests/test_postgres_typing.py b/tests/test_postgres_typing.py index 52a3cd5..9dec19f 100644 --- a/tests/test_postgres_typing.py +++ b/tests/test_postgres_typing.py @@ -32,7 +32,6 @@ async def test_typing_refresh_is_transition_only_and_stop_is_owner_bound(clean_p try: await store.create_room(RoomCreate(id="general", name="General")) await store.foundation.register_instance("node-a", lease_seconds=30) - await store.foundation.register_instance("observer", lease_seconds=30) connections = PostgresConnectionRegistry( store.foundation, max_connections=5, @@ -45,6 +44,7 @@ async def test_typing_refresh_is_transition_only_and_stop_is_owner_bound(clean_p username="alice", subject="alice-subject", ) + await store.foundation.register_instance("observer", lease_seconds=30) typing = PostgresTypingRegistry(store.foundation, timeout_seconds=8) started = await typing.start(connection_id="socket-one", instance_id="node-a") @@ -74,7 +74,6 @@ async def test_expiry_sweep_is_bounded_and_emits_stops(clean_postgres_database: try: await store.create_room(RoomCreate(id="general", name="General")) await store.foundation.register_instance("node-a", lease_seconds=30) - await store.foundation.register_instance("observer", lease_seconds=30) connections = PostgresConnectionRegistry( store.foundation, max_connections=5, @@ -89,6 +88,8 @@ async def test_expiry_sweep_is_bounded_and_emits_stops(clean_postgres_database: username=f"user-{index}", subject=None, ) + await store.foundation.register_instance("observer", lease_seconds=30) + for index in range(2): assert await typing.start(connection_id=f"socket-{index}", instance_id="node-a") async with store.foundation.transaction() as connection: await connection.execute( From 3f6e28633fa41123164381247f68454a8724cac3 Mon Sep 17 00:00:00 2001 From: Deathcharge Date: Mon, 10 Aug 2026 22:49:50 -0400 Subject: [PATCH 2/3] chore(deps): require fixed cryptography release --- CHANGELOG.md | 1 + pyproject.toml | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ca20b2..35f9c5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma - 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.x, excluding the vulnerable 49.0.0 release identified by `PYSEC-2026-3552`. ## 0.12.0 — 2026-08-02 diff --git a/pyproject.toml b/pyproject.toml index 2ccff06..9a65c8d 100644 --- a/pyproject.toml +++ b/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", From 4cb945401ab2dd66c608552902c1db6f1eefc87c Mon Sep 17 00:00:00 2001 From: Deathcharge Date: Tue, 11 Aug 2026 00:09:40 -0400 Subject: [PATCH 3/3] fix(postgres): bound presence cleanup work --- CHANGELOG.md | 2 +- samsarix_chat_engine/postgres.py | 2 ++ samsarix_chat_engine/postgres_connections.py | 7 ++++++- tests/test_postgres_presence.py | 4 ++-- tests/test_postgres_store.py | 20 ++++++++++++++++++++ 5 files changed, 31 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35f9c5c..d1e6efc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,7 +30,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma - 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.x, excluding the vulnerable 49.0.0 release identified by `PYSEC-2026-3552`. +- 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 diff --git a/samsarix_chat_engine/postgres.py b/samsarix_chat_engine/postgres.py index ed40019..43be8fc 100644 --- a/samsarix_chat_engine/postgres.py +++ b/samsarix_chat_engine/postgres.py @@ -491,6 +491,8 @@ 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_generation diff --git a/samsarix_chat_engine/postgres_connections.py b/samsarix_chat_engine/postgres_connections.py index 1df17b5..0830ce2 100644 --- a/samsarix_chat_engine/postgres_connections.py +++ b/samsarix_chat_engine/postgres_connections.py @@ -20,6 +20,7 @@ ) POSTGRES_CONNECTION_CAP_LOCK_ID = 7_495_346_927_831_819_047 +_ADMISSION_SWEEP_LIMIT = 16 _CONNECTION_ID_MAX_CHARS = 128 _USERNAME_MAX_CHARS = 64 _SUBJECT_MAX_CHARS = 64 @@ -132,7 +133,11 @@ async def try_acquire( raise ConnectionRoomUnavailableError("connection room is missing or archived") await connection.execute("SELECT pg_advisory_xact_lock(%s)", (POSTGRES_CONNECTION_CAP_LOCK_ID,)) - stale = await _delete_expired(connection, limit=1_000, prioritize_connection_id=connection_id) + stale = await _delete_expired( + connection, + limit=_ADMISSION_SWEEP_LIMIT, + prioritize_connection_id=connection_id, + ) await self._append_departures(connection, stale) cursor = await connection.execute( diff --git a/tests/test_postgres_presence.py b/tests/test_postgres_presence.py index 0034982..e40304a 100644 --- a/tests/test_postgres_presence.py +++ b/tests/test_postgres_presence.py @@ -161,8 +161,8 @@ async def test_crash_sweep_is_bounded_and_stops_typing_before_presence(clean_pos """ ) - assert len(await registry.reap_expired(limit=1)) == 1 - assert len(await registry.reap_expired(limit=1)) == 1 + transitions = await registry.reap_expired(limit=10) + assert [transition.connection_id for transition in transitions] == ["socket-0", "socket-1"] assert await registry.reap_expired() == [] events = await store.foundation.read_events("observer") assert [event.event_type for event in events] == [ diff --git a/tests/test_postgres_store.py b/tests/test_postgres_store.py index 3cea6c2..1f55234 100644 --- a/tests/test_postgres_store.py +++ b/tests/test_postgres_store.py @@ -163,6 +163,12 @@ async def test_schema_v6_backfills_matching_instance_generations( ) """ ) + await connection.execute( + """ + CREATE INDEX samsarix_connection_leases_instance + ON public.samsarix_connection_leases (instance_id, lease_expires_at, connection_id) + """ + ) await connection.execute( """ INSERT INTO public.samsarix_instance_cursors ( @@ -197,8 +203,22 @@ async def test_schema_v6_backfills_matching_instance_generations( """ ) generations = await cursor.fetchone() + cursor = await connection.execute( + """ + SELECT indexname + FROM pg_indexes + WHERE schemaname = 'public' + AND indexname IN ( + 'samsarix_connection_leases_instance', + 'samsarix_connection_leases_instance_generation' + ) + ORDER BY indexname + """ + ) + lease_indexes = [str(row[0]) for row in await cursor.fetchall()] assert generations is not None assert generations[0] is not None and generations[0] == generations[1] + assert lease_indexes == ["samsarix_connection_leases_instance_generation"] finally: await service.close()