Multi-replica: move all cross-request state into Postgres

CHANGELOG.md

@@ -48,6 +48,29 @@ navigation and console layout, and the internal seams (event dispatch, service
 ownership, projection) that the multi-replica work depends on.
 
 ### Added
+- **The app runs on more than one replica** (#213) — every piece of cross-request state
+  moves into PostgreSQL, so there is no longer a single-replica constraint and no Redis
+  or broker to operate. WebSocket frames and config invalidation travel over
+  `LISTEN`/`NOTIFY` as compact id descriptors that each replica reads back and renders
+  for its own sockets; scheduled inject releases, triggered communications, LLM
+  pipelines and the nightly audit purge become durable jobs (procrastinate); rate-limit
+  counters become rows, so the login, registration, and reset limits no longer multiply
+  by the replica count; and startup migrations serialise on a Postgres advisory lock.
+
+  Everything is published or enqueued **inside** the transaction that makes it true.
+  Postgres holds a `NOTIFY` until COMMIT and discards it on rollback, and a job row
+  commits with the state change that warranted it — so "committed but never announced"
+  and "committed but never enqueued" stop being possible. That closes the bug class
+  behind #211 (triggered communications lost on restart) and #218 (a scheduled release
+  stranded when a worker stood down mid-response) structurally rather than case by case,
+  and retires the hand-built coordination those fixes needed.
+
+  Kubernetes manifests keep `replicas: 1` in the base for a storage reason rather than a
+  state one — the uploads volume is `ReadWriteOnce` and a PVC's access modes cannot be
+  changed in place. Clusters with an RWX StorageClass apply the new
+  `k8s/overlays/multi-replica` for two replicas and rolling deploys. Do not front the app
+  with a transaction-mode connection pooler: `LISTEN` needs a session that outlives a
+  transaction.
 - **Runtime configuration** — non-secret settings move out of env-only config and into
   the admin UI, following the singleton-row + cached-config pattern already used by
   `/admin/audit` and `/admin/proxy`. Email/SMTP, general settings (registration, token

CLAUDE.md

@@ -36,8 +36,11 @@ structure. Subsystem deep-dives are indexed there and live in [PLAN.md](PLAN.md)
   transaction; **whoever commits dispatches** (`await dispatch(session)`, session open).
 - Services own queries; routers authorize/call/serialize; `app/schemas/` holds only
   boundary-crossing models (placement rule #214).
-- **Single-replica app**: all cross-request state (ws_manager, timers, rate limits,
-  config caches) is in-process. Read that section before adding any.
+- **Multi-replica**: cross-request state goes through Postgres — `pg_bus` (LISTEN/NOTIFY)
+  for anything other replicas must hear, `task_queue` for anything that must happen later.
+  Never add in-process cross-request state; `ws_manager` (this process's own sockets) is
+  the one legitimate exception. Publish and enqueue **inside** the transaction that makes
+  the thing true, never after the commit.
 - New UI control classes must re-assert the 40px min-height floor (class selectors beat
   the global floor; a Playwright touch-target test enforces ≥40/44px).
 - Branching wording trap: **participants choose the path; the facilitator controls the

README.md

@@ -207,7 +207,7 @@ Prefer plain `kubectl apply -f` over Kustomize? Apply the individual files under
 >
 > **Pod hardening**: every workload runs non-root under a PSS-`restricted`-style `securityContext` (no privilege escalation, all capabilities dropped, `RuntimeDefault` seccomp), and every container uses a read-only root filesystem — app, init, Caddy, Postgres, and the backup CronJob alike. The Postgres StatefulSet runs as uid 999 with `fsGroup: 999`, which needs a StorageClass that honours `fsGroup`.
 
-> **Note**: The app must run as a single replica (`replicas: 1`); the manifests enforce it with `strategy: Recreate`. Every piece of cross-request state lives in the **process**, so a second replica would not share any of it: the **WebSocket manager** (needs a shared bus, e.g. Redis pub/sub), **scheduled inject release** and **delayed triggered comms** (in-process `asyncio` tasks — need a task queue such as Celery or ARQ), the **login/registration/reset rate limiters** (per-process counters, so the effective limits would multiply by the replica count), the **SIEM and proxy config caches** (an admin's change would not reach the other replica), and **in-flight LLM assessments**. See the [deployment docs](https://icebergai.github.io/IcebergTTX/deployment/) for the full table.
+> **Scaling out**: the app shares all of its cross-request state through PostgreSQL, so it runs on more than one replica: WebSocket frames and config invalidation travel over `LISTEN`/`NOTIFY`, scheduled inject releases and triggered communications are durable jobs, and the rate limiters count in a table. There is no Redis and no broker to operate. The default manifests still ship `replicas: 1` because the uploads volume is `ReadWriteOnce` and a PVC's access modes cannot be changed in place; on a cluster with an RWX StorageClass, apply `k8s/overlays/multi-replica` for two replicas and rolling deploys. See the [deployment docs](https://icebergai.github.io/IcebergTTX/deployment/).
 
 ## Attachment reconciliation
 

SECURITY.md

@@ -46,5 +46,6 @@ participant reading another team's data, or an unauthenticated caller reaching a
 them) remain in scope.
 
 Deployment hardening (secret management, TLS termination, network policy, and the
-single-replica WebSocket constraint) is the operator's responsibility; see the
-deployment notes in [README.md](README.md) and [CLAUDE.md](CLAUDE.md).
+storage prerequisites for running more than one replica) is the operator's
+responsibility; see the deployment notes in [README.md](README.md) and
+[CLAUDE.md](CLAUDE.md).

alembic/env.py

@@ -30,6 +30,7 @@
     llm_settings,
     oidc_settings,
     proxy_settings,
+    rate_limit,
     report_summary,
     response,
     scenario,
@@ -38,7 +39,11 @@
 )
 
 config = context.config
-config.set_main_option("sqlalchemy.url", make_async_url(settings.database_url))
+# Settings are the source of truth, but an explicitly configured URL wins — that is how
+# a caller migrates a database other than the app's own (the migration tests point this
+# at a throwaway one; ``settings`` is bound at import and cannot be redirected).
+if not config.get_main_option("sqlalchemy.url", None):
+    config.set_main_option("sqlalchemy.url", make_async_url(settings.database_url))
 
 if config.config_file_name is not None:
     # run_migrations() runs Alembic in-process during the app lifespan, so the
@@ -50,25 +55,56 @@
 target_metadata = SQLModel.metadata
 
 
+def include_name(name, type_, parent_names) -> bool:
+    """Keep procrastinate's own tables out of autogenerate and ``alembic check``.
+
+    The task queue's schema is owned by the library and installed verbatim by its own
+    revision (#213), so it is deliberately absent from ``SQLModel.metadata``. Without
+    this filter every autogenerate would propose dropping it, and the ``alembic check``
+    that gates CI would fail on a database that is in fact correct.
+    """
+    if type_ == "table" and name is not None:
+        return not name.startswith("procrastinate_")
+    return True
+
+
 def run_migrations_offline() -> None:
     context.configure(
         url=config.get_main_option("sqlalchemy.url"),
         target_metadata=target_metadata,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
         compare_type=True,
+        include_name=include_name,
     )
     with context.begin_transaction():
         context.run_migrations()
 
 
+# Any 64-bit constant works; this one is arbitrary and only has to stay stable, since
+# two processes agree on a lock by using the same number.
+_MIGRATION_LOCK_ID = 8_213_100_213
+
+
 def _do_run_migrations(connection) -> None:
     context.configure(
         connection=connection,
         target_metadata=target_metadata,
         compare_type=True,
+        include_name=include_name,
     )
     with context.begin_transaction():
+        # Serialise migrations across replicas (#213). Every replica migrates on startup,
+        # so a rollout would otherwise run several `alembic upgrade head` concurrently
+        # against one database — two processes creating the same table is a crash loop,
+        # not a race you can retry past. Whoever loses the lock waits, then finds the
+        # schema already at head and applies nothing.
+        #
+        # Taken *inside* Alembic's transaction, and an xact lock rather than a session
+        # one, for two reasons: it is released automatically however the migration ends,
+        # and issuing any statement before this block would autobegin a second
+        # transaction that Alembic never commits — silently discarding every migration.
+        connection.exec_driver_sql(f"SELECT pg_advisory_xact_lock({_MIGRATION_LOCK_ID})")
         context.run_migrations()
 
 

alembic/versions/c7d8e9f0a1b2_add_procrastinate_schema.py

@@ -0,0 +1,58 @@
+"""Install the procrastinate task-queue schema (#213).
+
+The DDL is procrastinate's own, read from the installed package rather than copied here,
+so upgrading the library is a new revision wrapping its migration script instead of a
+hand-merged diff of ours.
+
+Revision ID: c7d8e9f0a1b2
+Revises: b3c4d5e6f7a8
+"""
+
+import inspect
+from collections.abc import Sequence
+
+from sqlalchemy.util import await_only
+
+from alembic import op
+
+revision: str = "c7d8e9f0a1b2"
+down_revision: str | None = "b3c4d5e6f7a8"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _run_script(sql: str) -> None:
+    """Execute a multi-statement script on the migration's own connection.
+
+    SQLAlchemy's asyncpg adapter routes every execute through a prepared statement, and
+    Postgres refuses to prepare more than one command at a time — so ``exec_driver_sql``
+    cannot run this. asyncpg's own ``execute`` uses the simple query protocol, which
+    can; reaching it directly also keeps the DDL inside the migration's transaction, so
+    a failure rolls back the schema and the version stamp together.
+    """
+    connection = op.get_bind()
+    raw = connection.connection.driver_connection
+    if inspect.iscoroutinefunction(getattr(raw, "execute", None)):
+        await_only(raw.execute(sql))
+    else:  # a synchronous driver (psycopg) needs none of the above
+        connection.exec_driver_sql(sql)
+
+
+def upgrade() -> None:
+    from procrastinate.schema import SchemaManager
+
+    _run_script(SchemaManager.get_schema())
+
+
+def downgrade() -> None:
+    _run_script(
+        """
+        DROP TABLE IF EXISTS procrastinate_events CASCADE;
+        DROP TABLE IF EXISTS procrastinate_periodic_defers CASCADE;
+        DROP TABLE IF EXISTS procrastinate_jobs CASCADE;
+        DROP TABLE IF EXISTS procrastinate_workers CASCADE;
+        DROP TYPE IF EXISTS procrastinate_job_status CASCADE;
+        DROP TYPE IF EXISTS procrastinate_job_event_type CASCADE;
+        DROP TYPE IF EXISTS procrastinate_job_to_defer_v1 CASCADE;
+        """
+    )

alembic/versions/d8e9f0a1b2c3_add_rate_limit_hits.py

@@ -0,0 +1,44 @@
+"""Move rate-limit counters into an unlogged table (#213).
+
+UNLOGGED is the point of writing this by hand rather than autogenerating it: these rows
+are expendable throttle state, so keeping them out of the WAL is what makes a write per
+login attempt acceptable. Postgres truncates the table after an unclean shutdown, which
+at worst forgives some in-progress lockouts.
+
+Revision ID: d8e9f0a1b2c3
+Revises: c7d8e9f0a1b2
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "d8e9f0a1b2c3"
+down_revision: str | None = "c7d8e9f0a1b2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute(
+        sa.text(
+            """
+            CREATE UNLOGGED TABLE rate_limit_hits (
+                id serial PRIMARY KEY,
+                scope varchar(32) NOT NULL,
+                key varchar(320) NOT NULL,
+                at timestamptz NOT NULL
+            )
+            """
+        )
+    )
+    op.create_index(
+        "ix_rate_limit_hits_scope_key_at", "rate_limit_hits", ["scope", "key", "at"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_rate_limit_hits_scope_key_at", table_name="rate_limit_hits")
+    op.drop_table("rate_limit_hits")

app/database.py

@@ -29,6 +29,20 @@ def make_async_url(url: str) -> str:
     return url
 
 
+def make_asyncpg_dsn(url: str) -> str:
+    """Strip the SQLAlchemy driver marker so raw asyncpg can dial the same database.
+
+    ``asyncpg.connect`` speaks libpq DSNs and rejects SQLAlchemy's ``+driver`` suffix.
+    The LISTEN connection (#213) is raw asyncpg rather than a pooled checkout, so it
+    needs the plain form of whatever ``DATABASE_URL`` was configured with.
+    """
+    async_url = make_async_url(url)
+    for prefix in ("postgresql+asyncpg://", "postgres+asyncpg://"):
+        if async_url.startswith(prefix):
+            return "postgresql://" + async_url[len(prefix) :]
+    return async_url
+
+
 engine = create_async_engine(make_async_url(settings.database_url), pool_pre_ping=True)
 
 
@@ -59,8 +73,12 @@ async def run_migrations() -> None:
 
     Run in a worker thread because Alembic's async ``env.py`` calls
     ``asyncio.run``, which cannot be invoked from the already-running lifespan
-    event loop. Safe for the single-replica deployment; multi-replica rollouts
-    should instead run ``alembic upgrade head`` as a dedicated deploy step.
+    event loop.
+
+    Safe with several replicas starting at once: ``env.py`` takes a Postgres advisory
+    lock first, so they migrate one at a time and the losers find the schema already at
+    head (#213). Migrations must still be forward-compatible across one release — during
+    a rolling update the previous version is serving against the new schema.
     """
     await asyncio.to_thread(_upgrade_to_head)
 

app/main.py

@@ -29,6 +29,7 @@
     inject_comment,
     llm_settings,
     proxy_settings,
+    rate_limit,
     report_summary,
     response,
     scenario,
@@ -70,7 +71,6 @@
 )
 from app.routers.ui import UIRedirect
 from app.services import audit_service, background
-from app.services.retention_service import retention_task
 from app.services.ws_manager import heartbeat_task
 
 logger = logging.getLogger("iceberg_ttx")
@@ -184,47 +184,55 @@ async def lifespan(app: FastAPI):
     from app.services.oidc import service as oidc_service
 
     oidc_service.register_providers()
-    # Re-arm persisted inject and communication schedules for exercises that were active
-    # before a restart (#116, #194). Timers remain single-process only.
-    from app.services.schedule_service import rehydrate_schedules
-
-    await rehydrate_schedules()
+    # Subscribe to the cross-replica bus, so a config saved on another replica lands in
+    # this one's caches (#213). Started after the loaders above: the first connect has
+    # nothing to recover, and connecting is deliberately not awaited to completion —
+    # a database blip must not stop a replica from serving what it already can.
+    from app.services import config_sync, ws_relay
+    from app.services.pg_listener import listener
+
+    config_sync.install()
+    ws_relay.install()
+    await listener.start()
+    # Run a queue worker in this replica, so a single container is still the whole
+    # deployment (#213). Schedules are durable jobs now, so nothing has to be rehydrated
+    # from memory; reconcile_release_jobs is a safety net that also carries over
+    # exercises which were running under a build that kept its timers in a dict.
+    from app.services import task_queue
+    from app.services.schedule_service import reconcile_release_jobs
+
+    await task_queue.start_worker()
+    await reconcile_release_jobs()
     audit_service.emit("app.startup", severity="info")
     task = asyncio.create_task(heartbeat_task())
-    # Prune audit history past the configured window and dead auth tokens: once now,
-    # then daily (#251). Not awaited inline like rehydrate_schedules above — a first
-    # pass over a legacy table is unbounded and would delay readiness. Single-process,
-    # like the timers: a second live replica would run a second concurrent sweep.
-    retention = asyncio.create_task(retention_task())
     yield
     # Graceful teardown (#250), strictly ordered:
     # 1. Record app.shutdown FIRST, while the loop is still live, so its audit-persist and
     #    SIEM-forward tasks are spawned in time to join the drain below instead of being
     #    abandoned into a dying loop (which reliably lost the shutdown record to stdout).
     audit_service.emit("app.shutdown", severity="info")
-    # 2. Stop the recurring loops and await their cancellation. Both are bare
-    #    create_task, so background.drain below never sees them — dropping this leaves a
-    #    task holding a pooled connection when engine.dispose() runs. It must also come
-    #    *before* the drain: a sweep's own audit event spawns persist/forward children
-    #    that are drainable, so cancelling first means they are drained, not abandoned
-    #    (same reasoning as step 1). A sweep cancelled mid-batch rolls back and redoes
-    #    that batch next boot.
-    for loop_task in (task, retention):
-        loop_task.cancel()
-    for loop_task in (task, retention):
-        with suppress(asyncio.CancelledError):
-            await loop_task
-    # 3. Drain in-flight background work (mail, audit persist, SIEM forward, LLM runs) and
-    #    armed timers in one bounded, converging pass. cancel_all_schedules, re-invoked each
-    #    round, cancels still-sleeping timers (rehydration re-arms them next boot) and hands
-    #    back any worker already mid-release so it finishes its commit and dispatch atomically
-    #    (#218); the drain re-collects the task sets after each wait so audit/SIEM writes and
-    #    triggered-comm timers spawned *by* those releases are drained too, not disposed out
-    #    from under (see background.drain).
-    from app.services.schedule_service import cancel_all_schedules
-
-    await background.drain(collect_extra=cancel_all_schedules)
-    # 4. Close the asyncpg pool cleanly so Postgres doesn't log unexpected EOFs on deploy.
+    # 2. Stop the socket heartbeat and await its cancellation. It is a bare create_task,
+    #    so background.drain below never sees it — dropping this leaves a task holding a
+    #    pooled connection when engine.dispose() runs. It must also come *before* the
+    #    drain: its own audit events spawn persist/forward children that are drainable,
+    #    so cancelling first means they are drained, not abandoned (as in step 1).
+    task.cancel()
+    with suppress(asyncio.CancelledError):
+        await task
+    # 3. Stop the queue worker, letting a running job finish within a bounded grace.
+    #    A job killed here is not lost: it stays *doing* until retry_stalled_jobs
+    #    returns it to the queue, and every task is safe to run twice (#213).
+    await task_queue.stop_worker()
+    # 4. Unsubscribe from the bus (#213). Same reasoning: a handler mid-refresh is
+    #    holding a pooled connection, and nothing arriving now can be acted on by a
+    #    replica that is going away.
+    await listener.stop()
+    # 5. Drain in-flight background work (mail, audit persist, SIEM forward) in one
+    #    bounded, converging pass — the drain re-collects the task set after each wait,
+    #    so audit and SIEM writes spawned *by* the work above are drained too rather
+    #    than disposed out from under (see background.drain).
+    await background.drain()
+    # 6. Close the asyncpg pool cleanly so Postgres doesn't log unexpected EOFs on deploy.
     from app.database import engine
 
     await engine.dispose()