From 4f2f71f9dc27bb186d18612abeda6d1a0d129a8c Mon Sep 17 00:00:00 2001 From: Doga Gursoy Date: Mon, 6 Jul 2026 23:51:32 +0300 Subject: [PATCH] Seed subscriber bookmarks at startup so Reactions actually fire Reactions (side-effecting subscribers: run_debriefer, caution_drafter, caution_promoter, authority_revocation_holder, and any future ones) own no proj_* table, so unlike projections they have no migration to seed their projection_bookmarks row. Nothing else seeded it either. The worker's first advance therefore called read_bookmark, which raises MissingBookmarkError; the advance loop caught it, wrote a no-op failure UPDATE (no row to update), backed off, and retried forever. Every Reaction was silently inert in a real deployment, including the just-shipped kill-switch authority_revocation_holder. It went unnoticed because every subscriber test either calls apply() directly or hand-seeds the bookmark, so the production worker path was never exercised. Fix: add ensure_bookmarks(pool, names) and call it once from projection_worker_lifespan before the worker starts, for every registered subscriber. For a projection the row already exists so the INSERT is a no-op via ON CONFLICT DO NOTHING; for a reaction it creates the missing row so its first advance reads a cursor instead of raising. The projection "fail loud if the migration never landed" guard is preserved by test_projection_table_match, which still enforces every projection has its CREATE TABLE + bookmark insert. Tests (real Postgres): a bookmark-less reaction's advance raises MissingBookmarkError (reproduces the bug); after ensure_bookmarks the same worker-path advance delivers the event; the ensure is idempotent across a restart (does not rewind an advanced cursor); and entering the real projection_worker_lifespan leaves every registered reaction with a readable bookmark (pins the fix at the production seam so a refactor that drops the step regresses here). Also drops the now-stale bookmark.py em-dash allowlist entry (the edit scrubbed its last em-dash). Verified: pyright + ruff + tach clean, 5400 projection/worker/subscriber integration + architecture tests green. Co-Authored-By: Claude --- .../infrastructure/projection/bookmark.py | 40 ++++- .../infrastructure/projection/lifespan.py | 9 ++ .../tests/architecture/test_no_em_dashes.py | 1 - .../test_ensure_bookmarks_postgres.py | 140 ++++++++++++++++++ 4 files changed, 184 insertions(+), 6 deletions(-) create mode 100644 apps/api/tests/integration/test_ensure_bookmarks_postgres.py diff --git a/apps/api/src/cora/infrastructure/projection/bookmark.py b/apps/api/src/cora/infrastructure/projection/bookmark.py index c5fe175b18d..5dd72252b40 100644 --- a/apps/api/src/cora/infrastructure/projection/bookmark.py +++ b/apps/api/src/cora/infrastructure/projection/bookmark.py @@ -23,11 +23,19 @@ surface (admin endpoint + lag sampler + OTel) which stays deferred until the trigger fires. -Bookmark rows are created by per-projection migrations (`INSERT INTO -projection_bookmarks (name) VALUES (...) ON CONFLICT DO NOTHING`), -not by the worker — registering a projection without its migration -having landed will fail loudly at first advance, which is the -behavior we want. +Bookmark rows for PROJECTIONS are created by their per-projection +migration (`INSERT INTO projection_bookmarks (name) VALUES (...) ON +CONFLICT DO NOTHING`). Registering a projection whose migration never +landed fails loudly at first advance (`test_projection_table_match` +also enforces the migration exists), which is the behavior we want. + +REACTIONS (side-effecting subscribers) own no `proj_*` table and thus +no migration, so nothing would seed their bookmark. `ensure_bookmarks` +closes that gap: the lifespan calls it once for every registered +subscriber before the worker starts, creating any missing row +idempotently. Without it a reaction's first advance raises +`MissingBookmarkError` forever inside the worker's backoff loop and the +reaction never fires. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -73,6 +81,12 @@ WHERE name = $1 """ +_ENSURE_BOOKMARK_SQL = """ +INSERT INTO projection_bookmarks (name) +VALUES ($1) +ON CONFLICT (name) DO NOTHING +""" + class MissingBookmarkError(Exception): """No bookmark row exists for the given projection name. Indicates @@ -152,8 +166,24 @@ async def write_bookmark_failure( await conn.execute(_WRITE_BOOKMARK_FAILURE_SQL, name, truncated) +async def ensure_bookmarks(pool: asyncpg.Pool, names: frozenset[str]) -> None: + """Idempotently create a bookmark row for each given subscriber name. + + Called once at worker startup for every registered subscriber. For a + PROJECTION the row already exists (its migration seeded it), so the INSERT is + a no-op via `ON CONFLICT DO NOTHING`. For a REACTION (no table, no migration) + this creates the otherwise-missing row so its first advance can read a cursor + instead of raising `MissingBookmarkError` forever. Ordered by name for a + deterministic write sequence. + """ + async with pool.acquire() as conn: + for name in sorted(names): + await conn.execute(_ENSURE_BOOKMARK_SQL, name) + + __all__ = [ "MissingBookmarkError", + "ensure_bookmarks", "read_bookmark", "write_bookmark", "write_bookmark_failure", diff --git a/apps/api/src/cora/infrastructure/projection/lifespan.py b/apps/api/src/cora/infrastructure/projection/lifespan.py index 32701d25359..c2d9ac59b5e 100644 --- a/apps/api/src/cora/infrastructure/projection/lifespan.py +++ b/apps/api/src/cora/infrastructure/projection/lifespan.py @@ -15,6 +15,7 @@ from cora.infrastructure.config import Settings from cora.infrastructure.kernel import Kernel from cora.infrastructure.logging import get_logger +from cora.infrastructure.projection.bookmark import ensure_bookmarks from cora.infrastructure.projection.registry import ProjectionRegistry from cora.infrastructure.projection.wakeup import ( ListenNotifyWakeup, @@ -50,6 +51,14 @@ async def projection_worker_lifespan( yield return + # Ensure a bookmark row exists for every registered subscriber before the + # worker starts. Projections already have theirs (seeded by their migration); + # this creates the otherwise-missing rows for Reactions (side-effecting + # subscribers own no proj_* table, hence no migration, hence no seed). Without + # it a Reaction's first advance raises MissingBookmarkError forever inside the + # worker's backoff loop and it never fires. Idempotent (ON CONFLICT DO NOTHING). + await ensure_bookmarks(deps.pool, registry.names()) + wakeup: WakeupSource = ( ListenNotifyWakeup(deps.pool) if settings.projection_use_listen_notify else PollOnlyWakeup() ) diff --git a/apps/api/tests/architecture/test_no_em_dashes.py b/apps/api/tests/architecture/test_no_em_dashes.py index 27f511f2b1b..31f66ee85a5 100644 --- a/apps/api/tests/architecture/test_no_em_dashes.py +++ b/apps/api/tests/architecture/test_no_em_dashes.py @@ -193,7 +193,6 @@ "src/cora/infrastructure/ports/profile_store.py", "src/cora/infrastructure/ports/signer.py", "src/cora/infrastructure/ports/token_verifier.py", - "src/cora/infrastructure/projection/bookmark.py", "src/cora/infrastructure/projection/drain.py", "src/cora/infrastructure/projection/handler.py", "src/cora/infrastructure/projection/wakeup.py", diff --git a/apps/api/tests/integration/test_ensure_bookmarks_postgres.py b/apps/api/tests/integration/test_ensure_bookmarks_postgres.py new file mode 100644 index 00000000000..5f4d556d00d --- /dev/null +++ b/apps/api/tests/integration/test_ensure_bookmarks_postgres.py @@ -0,0 +1,140 @@ +"""Integration: ensure_bookmarks seeds a Reaction's bookmark so the worker path +can advance it (the subscriber-bookmark gap fix). + +Reactions (side-effecting subscribers) own no proj_* table and thus no migration, +so nothing seeds their projection_bookmarks row. Before the fix, a Reaction's +first advance through the worker raised MissingBookmarkError forever and it never +fired. This test reproduces that (advance raises without a bookmark) and pins the +fix (after ensure_bookmarks the same advance succeeds and the Reaction sees its +event). +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false + +from datetime import UTC, datetime +from uuid import uuid4 + +import asyncpg +import pytest + +from cora.infrastructure.adapters.postgres_event_store import PostgresEventStore +from cora.infrastructure.ports.event_store import NewEvent, StoredEvent +from cora.infrastructure.projection.bookmark import ( + MissingBookmarkError, + ensure_bookmarks, +) +from cora.infrastructure.projection.handler import ConnectionLike +from cora.infrastructure.projection.worker import advance_subscriber_once + +_NOW = datetime(2026, 7, 6, 12, 0, 0, tzinfo=UTC) + + +class _RecordingReaction: + """Minimal Reaction: no proj_* table, no migration (the bug's shape). + + Captures the events it is handed so the test can assert the worker path + actually delivered them once a bookmark exists. + """ + + name = "test_ensure_bookmarks_reaction" + subscribed_event_types = frozenset({"EnsureBookmarksProbeEvent"}) + batch_size = 1 + + def __init__(self) -> None: + self.seen: list[StoredEvent] = [] + + async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: + _ = conn + self.seen.append(event) + + +async def _append_probe(store: PostgresEventStore) -> None: + await store.append( + "TestStream", + uuid4(), + 0, + [ + NewEvent( + event_id=uuid4(), + event_type="EnsureBookmarksProbeEvent", + schema_version=1, + payload={"k": "v"}, + occurred_at=_NOW, + correlation_id=uuid4(), + causation_id=None, + metadata={}, + principal_id=uuid4(), + ) + ], + ) + + +@pytest.mark.integration +async def test_advance_without_bookmark_raises_missing_bookmark(db_pool: asyncpg.Pool) -> None: + """Reproduce the bug: a Reaction with no seeded bookmark cannot advance.""" + reaction = _RecordingReaction() + with pytest.raises(MissingBookmarkError): + await advance_subscriber_once(db_pool, reaction) + + +@pytest.mark.integration +async def test_ensure_bookmarks_lets_the_reaction_advance(db_pool: asyncpg.Pool) -> None: + """After ensure_bookmarks seeds the row, the same worker-path advance delivers + the event to the Reaction.""" + store = PostgresEventStore(db_pool) + reaction = _RecordingReaction() + await _append_probe(store) + + await ensure_bookmarks(db_pool, frozenset({reaction.name})) + processed = await advance_subscriber_once(db_pool, reaction) + + assert processed == 1 + assert len(reaction.seen) == 1 + assert reaction.seen[0].event_type == "EnsureBookmarksProbeEvent" + + +@pytest.mark.integration +async def test_ensure_bookmarks_is_idempotent(db_pool: asyncpg.Pool) -> None: + """Calling ensure_bookmarks twice (restart) does not reset an advanced cursor: + the second call is a no-op via ON CONFLICT DO NOTHING.""" + store = PostgresEventStore(db_pool) + reaction = _RecordingReaction() + await ensure_bookmarks(db_pool, frozenset({reaction.name})) + await _append_probe(store) + assert await advance_subscriber_once(db_pool, reaction) == 1 + + # Second ensure (as on a restart) must not rewind the bookmark to zero. + await ensure_bookmarks(db_pool, frozenset({reaction.name})) + # No new events -> nothing to process; if the cursor had reset, the probe + # would be re-delivered and this would be 1. + assert await advance_subscriber_once(db_pool, reaction) == 0 + + +@pytest.mark.integration +async def test_lifespan_seeds_registered_reaction_bookmark(db_pool: asyncpg.Pool) -> None: + """The production wiring: projection_worker_lifespan seeds a bookmark for every + registered subscriber before the worker starts, so a bookmark-less Reaction is + no longer stranded. Pins the fix at the real lifespan seam (not just the helper), + so a future refactor that drops the ensure step regresses here.""" + from cora.infrastructure.config import Settings + from cora.infrastructure.projection import ProjectionRegistry + from cora.infrastructure.projection.bookmark import read_bookmark + from cora.infrastructure.projection.lifespan import projection_worker_lifespan + from tests.integration._helpers import build_postgres_deps + + reaction = _RecordingReaction() + registry = ProjectionRegistry() + registry.register(reaction) + deps = build_postgres_deps(db_pool, now=_NOW) + settings = Settings() # type: ignore[call-arg] + + async def _read_cursor() -> tuple[int, int]: + async with db_pool.acquire() as conn, conn.transaction(): + return await read_bookmark(conn, reaction.name) + + async with projection_worker_lifespan(deps, registry, settings): + # Inside the context the worker is running; the bookmark MUST exist (the + # ensure step ran before the worker spawned). read_bookmark raises + # MissingBookmarkError if the row is absent. + cursor = await _read_cursor() + assert cursor == (0, 0)