feat(postgres): enforce distributed rate buckets

CHANGELOG.md

@@ -13,6 +13,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma
 - 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.
+- 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.
 
 ### Security and operations
 
@@ -24,6 +25,7 @@ This project follows semantic versioning while it is in alpha: minor versions ma
 - 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.
+- 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.
 
 ## 0.12.0 — 2026-08-02
 

ROADMAP.md

@@ -86,8 +86,9 @@ This makes the supported topology repeatable without implying that a container m
 - [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;
+- [x] implement bounded PostgreSQL-owned message, search, and typing rate buckets with database-time windows;
 - [ ] expose guarded PostgreSQL configuration only after the remaining topology gates are proven;
-- [ ] implement and test presence, distributed rate limits, and reconnect recovery, then wire the proven cross-worker fan-out path into the application;
+- [ ] implement and test presence/typing expiry and reconnect recovery, then wire the proven cross-worker fan-out 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

@@ -18,6 +18,8 @@ Schema v3 uses PostgreSQL database time for room/message/moderation ordering, re
 
 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 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.
+
 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
@@ -47,7 +49,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.
-- **Rate limits:** atomic time-bucket counters enforce deployment-wide subject/client limits. Database time defines bucket boundaries so host clock skew cannot multiply quotas.
+- **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:** transition events use the durable event path but expire automatically and are not retained as chat history or audit content.
 - **Moderation teardown:** room archive and member-ban events reach every instance, which closes matching local sockets deterministically.
 - **Webhook work:** implemented outbox workers claim due rows with an expiring owner lease using row locks and `SKIP LOCKED`. Only the current unexpired owner can acknowledge a claim. A crashed claim becomes eligible for redelivery with its stable ID; receivers still deduplicate that ID.
@@ -73,7 +75,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 (the storage-level concurrent connection-cap case is implemented; real-process admission and rate-limit gates remain);
+- global and per-room connection caps plus rate limits hold across processes (storage-level concurrent connection and rate-bucket cases are implemented; real-process admission/request-path 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 = 4
+POSTGRES_SCHEMA_VERSION = 5
 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"
@@ -472,6 +472,27 @@ async def _initialize_schema(self) -> None:
                     ON public.samsarix_connection_leases (room_id, subject, lease_expires_at)
                     """
                 )
+                await connection.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS public.samsarix_rate_buckets (
+                        scope TEXT NOT NULL CHECK (scope IN ('message', 'search', 'typing')),
+                        key_digest BYTEA NOT NULL CHECK (octet_length(key_digest) = 32),
+                        window_started_at TIMESTAMPTZ NOT NULL,
+                        event_count INTEGER NOT NULL CHECK (event_count > 0),
+                        expires_at TIMESTAMPTZ NOT NULL,
+                        created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+                        updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+                        PRIMARY KEY (scope, key_digest, window_started_at),
+                        CHECK (expires_at > window_started_at)
+                    )
+                    """
+                )
+                await connection.execute(
+                    """
+                    CREATE INDEX IF NOT EXISTS samsarix_rate_buckets_expiry
+                    ON public.samsarix_rate_buckets (expires_at, scope, key_digest)
+                    """
+                )
                 await connection.execute(
                     """
                     CREATE TABLE IF NOT EXISTS public.samsarix_messages (

samsarix_chat_engine/postgres_rate_limits.py

@@ -0,0 +1,241 @@
+# Copyright (c) 2026 Samsarix LLC
+# SPDX-License-Identifier: MPL-2.0
+"""Atomic PostgreSQL rate buckets for multi-instance request controls."""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Any, Literal, cast
+
+from psycopg import AsyncConnection
+
+from .postgres import PostgresFoundation
+
+POSTGRES_RATE_BUCKET_CAP_LOCK_ID = 7_495_346_927_831_819_048
+RateLimitScope = Literal["message", "search", "typing"]
+_RATE_LIMIT_SCOPES = frozenset({"message", "search", "typing"})
+_MAX_RATE_KEY_BYTES = 1_024
+
+
+class RateBucketCapacityError(RuntimeError):
+    """Raised when active rate-bucket cardinality reaches its configured bound."""
+
+
+@dataclass(frozen=True, slots=True)
+class RateLimitDecision:
+    """Result of one atomic rate consumption attempt."""
+
+    allowed: bool
+    remaining: int
+    retry_after_seconds: int
+    reset_at: datetime
+
+
+class PostgresRateLimiter:
+    """Enforce one fixed database-time window across all service replicas."""
+
+    def __init__(
+        self,
+        foundation: PostgresFoundation,
+        *,
+        scope: RateLimitScope,
+        limit: int,
+        window_seconds: int = 60,
+        max_buckets: int = 100_000,
+    ) -> None:
+        if scope not in _RATE_LIMIT_SCOPES:
+            raise ValueError("invalid PostgreSQL rate-limit scope")
+        if not 1 <= limit <= 100_000:
+            raise ValueError("PostgreSQL rate limit must be between 1 and 100000")
+        if not 1 <= window_seconds <= 3_600:
+            raise ValueError("PostgreSQL rate window must be between 1 and 3600 seconds")
+        if not 1 <= max_buckets <= 10_000_000:
+            raise ValueError("PostgreSQL rate-bucket capacity must be between 1 and 10000000")
+        self.foundation = foundation
+        self.scope = scope
+        self.limit = limit
+        self.window_seconds = window_seconds
+        self.max_buckets = max_buckets
+
+    async def allow(self, key: str) -> bool:
+        """Consume one allowance and return whether it was accepted."""
+
+        return (await self.consume(key)).allowed
+
+    async def consume(self, key: str) -> RateLimitDecision:
+        """Atomically consume one allowance without persisting the raw identity key."""
+
+        digest = _digest_key(self.scope, key)
+        async with self.foundation.transaction() as connection:
+            now, window_started_at = await self._database_window(connection)
+
+            decision = await self._consume_existing(connection, digest, now, window_started_at)
+            if decision is not None:
+                return decision
+
+            # Only new-cardinality work takes the global capacity lock. Hot
+            # identities contend on their own bucket row rather than a global
+            # request-path mutex.
+            await connection.execute("SELECT pg_advisory_xact_lock(%s)", (POSTGRES_RATE_BUCKET_CAP_LOCK_ID,))
+            await connection.execute("DELETE FROM public.samsarix_rate_buckets WHERE expires_at <= clock_timestamp()")
+            # Lock acquisition may span a boundary under a new-key storm.
+            # Recompute from database time so an already-expired bucket is
+            # never inserted for the prior window.
+            now, window_started_at = await self._database_window(connection)
+
+            decision = await self._consume_existing(connection, digest, now, window_started_at)
+            if decision is not None:
+                return decision
+
+            cursor = await connection.execute("SELECT COUNT(*) FROM public.samsarix_rate_buckets")
+            row = await cursor.fetchone()
+            if row is not None and int(row[0]) >= self.max_buckets:
+                raise RateBucketCapacityError("PostgreSQL rate-bucket capacity reached")
+            reset_at = window_started_at + timedelta(seconds=self.window_seconds)
+            cursor = await connection.execute(
+                """
+                INSERT INTO public.samsarix_rate_buckets (
+                    scope, key_digest, window_started_at, event_count, expires_at
+                )
+                VALUES (%s, %s, %s, 1, %s)
+                RETURNING event_count
+                """,
+                (self.scope, digest, window_started_at, reset_at),
+            )
+            inserted = await cursor.fetchone()
+            if inserted is None:  # pragma: no cover - PostgreSQL guarantees RETURNING
+                raise RuntimeError("PostgreSQL did not return a rate bucket")
+            return self._decision(True, int(inserted[0]), now, window_started_at)
+
+    async def active_bucket_count(self) -> int:
+        """Return active bucket cardinality for bounded operational visibility."""
+
+        async with self.foundation.transaction() as connection:
+            cursor = await connection.execute(
+                "SELECT COUNT(*) FROM public.samsarix_rate_buckets WHERE expires_at > clock_timestamp()"
+            )
+            row = await cursor.fetchone()
+        return int(row[0]) if row is not None else 0
+
+    async def prune_expired(self) -> int:
+        """Delete expired buckets under the same lock used for capacity decisions."""
+
+        async with self.foundation.transaction() as connection:
+            await connection.execute("SELECT pg_advisory_xact_lock(%s)", (POSTGRES_RATE_BUCKET_CAP_LOCK_ID,))
+            cursor = await connection.execute(
+                "DELETE FROM public.samsarix_rate_buckets WHERE expires_at <= clock_timestamp()"
+            )
+        return cursor.rowcount
+
+    async def _increment_existing(
+        self,
+        connection: AsyncConnection[tuple[Any, ...]],
+        digest: bytes,
+        window_started_at: datetime,
+    ) -> int | None:
+        cursor = await connection.execute(
+            """
+            UPDATE public.samsarix_rate_buckets
+            SET event_count = event_count + 1,
+                updated_at = clock_timestamp()
+            WHERE scope = %s
+              AND key_digest = %s
+              AND window_started_at = %s
+              AND event_count < %s
+              AND expires_at > clock_timestamp()
+            RETURNING event_count
+            """,
+            (self.scope, digest, window_started_at, self.limit),
+        )
+        row = await cursor.fetchone()
+        return int(row[0]) if row is not None else None
+
+    async def _consume_existing(
+        self,
+        connection: AsyncConnection[tuple[Any, ...]],
+        digest: bytes,
+        now: datetime,
+        window_started_at: datetime,
+    ) -> RateLimitDecision | None:
+        while True:
+            count = await self._increment_existing(connection, digest, window_started_at)
+            if count is not None:
+                return self._decision(True, count, now, window_started_at)
+            existing = await self._read_existing(connection, digest, window_started_at)
+            if existing is None:
+                return None
+            count, observed_at = existing
+            if count >= self.limit:
+                return self._decision(False, count, observed_at, window_started_at)
+            # The bucket committed between our UPDATE snapshot and SELECT.
+            # It still has allowance, so retry consumption rather than falsely
+            # rejecting the request merely because the row now exists.
+
+    async def _database_window(
+        self,
+        connection: AsyncConnection[tuple[Any, ...]],
+    ) -> tuple[datetime, datetime]:
+        cursor = await connection.execute(
+            """
+            SELECT
+                statement_timestamp(),
+                date_bin(
+                    make_interval(secs => %s),
+                    statement_timestamp(),
+                    TIMESTAMPTZ '2000-01-01 00:00:00+00'
+                )
+            """,
+            (self.window_seconds,),
+        )
+        timing = await cursor.fetchone()
+        if timing is None:  # pragma: no cover - PostgreSQL always returns this scalar row
+            raise RuntimeError("PostgreSQL did not return rate-bucket time")
+        return cast(datetime, timing[0]), cast(datetime, timing[1])
+
+    async def _read_existing(
+        self,
+        connection: AsyncConnection[tuple[Any, ...]],
+        digest: bytes,
+        window_started_at: datetime,
+    ) -> tuple[int, datetime] | None:
+        cursor = await connection.execute(
+            """
+            SELECT event_count, clock_timestamp()
+            FROM public.samsarix_rate_buckets
+            WHERE scope = %s
+              AND key_digest = %s
+              AND window_started_at = %s
+              AND expires_at > clock_timestamp()
+            """,
+            (self.scope, digest, window_started_at),
+        )
+        row = await cursor.fetchone()
+        return (int(row[0]), cast(datetime, row[1])) if row is not None else None
+
+    def _decision(
+        self,
+        allowed: bool,
+        count: int,
+        now: datetime,
+        window_started_at: datetime,
+    ) -> RateLimitDecision:
+        reset_at = window_started_at + timedelta(seconds=self.window_seconds)
+        retry_after = max(1, math.ceil((reset_at - now).total_seconds())) if not allowed else 0
+        return RateLimitDecision(
+            allowed=allowed,
+            remaining=max(0, self.limit - count),
+            retry_after_seconds=retry_after,
+            reset_at=reset_at,
+        )
+
+
+def _digest_key(scope: RateLimitScope, key: str) -> bytes:
+    if not key:
+        raise ValueError("PostgreSQL rate-limit key is required")
+    encoded = key.encode("utf-8")
+    if len(encoded) > _MAX_RATE_KEY_BYTES:
+        raise ValueError("PostgreSQL rate-limit key exceeds 1024 bytes")
+    return hashlib.sha256(scope.encode("ascii") + b"\x00" + encoded).digest()