diff --git a/backend/app/activity_stream.py b/backend/app/activity_stream.py index 56311382b..a30e589e9 100644 --- a/backend/app/activity_stream.py +++ b/backend/app/activity_stream.py @@ -9,118 +9,661 @@ from __future__ import annotations +import asyncio from typing import Any +from uuid import UUID import redis.asyncio as redis from fastapi import Request +from redis.exceptions import WatchError from lineageweave.observability import traced +_SYNC_ACTIVITY_WATCH_RETRY_LIMIT = 8 +_MAX_ACTIVITY_READ_COUNT = 1000 +_ACTIVITY_ALIAS_INDEX_WRITE_BATCH_SIZE = 128 +_ACTIVITY_STREAM_PREFIX = "activity:" +_ACTIVITY_ALIAS_INDEX_PREFIX = "activity-aliases:" -def create_valkey_client(url: str) -> redis.Redis: - """One shared async client for the process, mirroring db.create_pool.""" - return redis.from_url(url, decode_responses=True) + +def create_valkey_client(valkey_url: str) -> redis.Redis: + """Create the process-wide async client for the configured Valkey endpoint. + + ``valkey_url`` is deployment configuration, not a per-request destination. + Connection establishment stays lazy in redis-py; startup therefore owns the + client lifecycle while request handlers only consume the stored client. + """ + return redis.from_url(valkey_url, decode_responses=True) def get_valkey(request: Request) -> redis.Redis: - """FastAPI dependency: the client stored on ``app.state`` at startup.""" + """Return the Valkey client installed on FastAPI application state. + + Startup is responsible for creating ``app.state.valkey``. This dependency + deliberately does not construct a fallback client from request data or + environment state because that would create a second connection authority. + """ return request.app.state.valkey def _stream_key(post_id: str) -> str: - return f"activity:{post_id}" + """Map one source-post identity to its stable activity-stream wire key. + + ``source_post.post_id`` is PostgreSQL ``uuid``. PostgreSQL accepts more than + one textual spelling for the same UUID, so using raw request text as the + Valkey suffix can fork one database identity into case- or format-variant + streams. Syntactically valid UUID strings therefore converge on Python's + canonical lowercase hyphenated representation before key construction. + Non-UUID fixture/legacy strings retain exact spelling, and non-string values + are still rejected rather than being stringified onto another identity. + """ + if type(post_id) is not str: + raise TypeError("post_id must be a string") + try: + canonical_post_id = str(UUID(post_id)) + except ValueError: + canonical_post_id = post_id + return f"{_ACTIVITY_STREAM_PREFIX}{canonical_post_id}" + + +def _activity_alias_index_key(post_id: str) -> str: + """Return the durable legacy-alias index for one canonical post UUID.""" + return f"{_ACTIVITY_ALIAS_INDEX_PREFIX}{str(UUID(post_id))}" + + +async def _flush_activity_alias_index_writes( + valkey_client: redis.Redis, + writes: list[tuple[str, str]], +) -> int: + """Persist one bounded batch of independent alias-index set insertions.""" + if not writes: + return 0 + + pipeline_factory = getattr(valkey_client, "pipeline", None) + if pipeline_factory is None: + indexed = 0 + for index_key, stream_key in writes: + indexed += int(await valkey_client.sadd(index_key, stream_key)) + return indexed + + async with pipeline_factory(transaction=False) as write_pipeline: + for index_key, stream_key in writes: + write_pipeline.sadd(index_key, stream_key) + results = await write_pipeline.execute() + return sum(int(result) for result in results) + + +async def index_legacy_activity_stream_aliases(valkey_client: redis.Redis) -> int: + """Index every existing UUID stream alias before canonical-only reads begin. + + PostgreSQL accepts UUID input spellings with braces, upper-case digits, and + non-standard hyphen placement. Older routes used that raw spelling in the + Valkey key, so enumerating a few common variants cannot preserve all valid + history. Startup performs one cursor scan and records the finite aliases + that actually exist. New writes remain canonical-only. Independent alias-set + writes are flushed through bounded non-transactional pipeline batches so + rollout readiness does not add one serial network wait per historical alias. + """ + indexed = 0 + pending_writes: list[tuple[str, str]] = [] + async for raw_key in valkey_client.scan_iter(match=f"{_ACTIVITY_STREAM_PREFIX}*"): + stream_key = raw_key.decode() if isinstance(raw_key, bytes) else raw_key + if not isinstance(stream_key, str) or stream_key.startswith(_ACTIVITY_ALIAS_INDEX_PREFIX): + continue + raw_post_id = stream_key.removeprefix(_ACTIVITY_STREAM_PREFIX) + try: + canonical_post_id = str(UUID(raw_post_id)) + except ValueError: + continue + canonical_key = f"{_ACTIVITY_STREAM_PREFIX}{canonical_post_id}" + if stream_key == canonical_key: + continue + pending_writes.append( + ( + f"{_ACTIVITY_ALIAS_INDEX_PREFIX}{canonical_post_id}", + stream_key, + ) + ) + if len(pending_writes) >= _ACTIVITY_ALIAS_INDEX_WRITE_BATCH_SIZE: + indexed += await _flush_activity_alias_index_writes( + valkey_client, + pending_writes, + ) + pending_writes.clear() + indexed += await _flush_activity_alias_index_writes(valkey_client, pending_writes) + return indexed + + +async def _activity_read_stream_keys( + valkey_client: redis.Redis, + post_id: str, +) -> tuple[str, ...]: + """Return the canonical stream plus bounded pre-canonical compatibility aliases. + + New writes converge on the canonical UUID key. Before that invariant existed, + an uppercase UUID route spelling could create an independent raw stream. The + startup alias index therefore supplies the finite UUID-equivalent stream keys + that actually exist. The request never scans unrelated keys or guesses a + subset of PostgreSQL's accepted UUID spellings. This is a read-only bridge: + it does not create new alias streams. + """ + canonical_key = _stream_key(post_id) + if type(post_id) is not str: + raise TypeError("post_id must be a string") + try: + UUID(post_id) + except ValueError: + return (canonical_key,) + + indexed_aliases: list[str] = [] + async for alias in valkey_client.sscan_iter(_activity_alias_index_key(post_id)): + indexed_aliases.append(alias) + if len(indexed_aliases) > _MAX_ACTIVITY_READ_COUNT - 1: + raise RuntimeError("Activity history has too many retained compatibility streams") + candidate_keys = (canonical_key, *sorted(indexed_aliases)) + return tuple(dict.fromkeys(candidate_keys)) + + +async def _activity_initial_uuid_read( + valkey_client: redis.Redis, + post_id: str, + event_count: int, +) -> tuple[list[tuple[str, dict[str, str]]], tuple[str, ...]] | None: + """Read complete alias admission and canonical events in one UUID exchange. + + The retained-alias set exists only to bridge pre-canonical UUID spellings. + Redis pipelines preserve the independent SSCAN and canonical XREVRANGE while + sending them together. When the bounded SSCAN page is complete, its aliases + are authoritative for this request and are returned with the already-fetched + canonical page instead of being discarded and scanned again. A nonzero scan + cursor falls back to the established bounded iterator so history is never + hidden merely to reduce latency. + + Minimal Redis-compatible test adapters without pipeline support return + ``None`` and exercise the established fallback; production clients created + by :func:`create_valkey_client` always provide the async pipeline contract. + """ + canonical_key = _stream_key(post_id) + try: + UUID(post_id) + except ValueError: + return None + + pipeline_factory = getattr(valkey_client, "pipeline", None) + if pipeline_factory is None: + return None + + async with pipeline_factory(transaction=False) as read_pipeline: + read_pipeline.sscan( + _activity_alias_index_key(post_id), + cursor=0, + count=_MAX_ACTIVITY_READ_COUNT, + ) + read_pipeline.xrevrange(canonical_key, count=event_count) + with traced( + "lineageweave.valkey.activity_read_pipeline", + { + "db.system": "redis", + "db.operation.name": "pipeline", + "lineageweave.stream.kind": "activity", + }, + ): + alias_page, canonical_entries = await read_pipeline.execute() + + alias_cursor, aliases = alias_page + if int(alias_cursor) != 0: + return None + candidate_keys = (canonical_key, *sorted(aliases)) + stream_keys = tuple(dict.fromkeys(candidate_keys)) + if len(stream_keys) > _MAX_ACTIVITY_READ_COUNT: + raise RuntimeError("Activity history has too many retained compatibility streams") + return canonical_entries, stream_keys[1:] + + +def _activity_stream_entry_order(entry_id: str) -> tuple[int, int]: + """Parse one Valkey stream entry id into its stream-local numeric order.""" + milliseconds, separator, sequence = entry_id.partition("-") + if separator != "-": + raise ValueError("Valkey activity event id is malformed") + try: + return int(milliseconds), int(sequence) + except ValueError as exc: + raise ValueError("Valkey activity event id is malformed") from exc + + +def _activity_compatibility_merge_order( + entry_id: str, + stream_index: int, +) -> tuple[int, int, int]: + """Order compatibility reads without inventing cross-stream sequence chronology. + + Redis stream sequence numbers are ordered only inside one stream. Historical + alias streams can therefore contain the same millisecond with unrelated + sequence counters. The millisecond remains comparable; an equal-millisecond + tie uses the declared stream precedence (canonical first, then bounded legacy + aliases), and the sequence number only orders entries that came from that + same stream. This fallback is deterministic but deliberately does not claim + to reconstruct unknowable sub-millisecond chronology across old streams. + """ + milliseconds, sequence = _activity_stream_entry_order(entry_id) + return milliseconds, -stream_index, sequence + + +def _activity_public_event_id(entry_id: str, stream_index: int) -> str: + """Return an opaque event identity unique across bounded compatibility streams. + + Valkey stream IDs are unique only inside one stream, so a canonical stream and + a historical alias can legitimately both contain ``600-0``. Canonical events + keep their established public ID. Alias events receive a bounded stream-index + namespace that contains no post identifier or raw stream key, preventing + duplicate React/API identities without inventing event chronology or mutating + persisted records. + """ + if stream_index == 0: + return entry_id + return f"legacy-{stream_index}:{entry_id}" def ticket_created_summary(ticket_title: str) -> str: - """The ``summary`` field ``ticket_created`` producers must share.""" + """Build the stable human-readable summary for a ticket-created event. + + The summary is display text and only one field of reseed identity; callers + must not use it alone to decide whether two activity facts are the same. + """ return f"Ticket created: {ticket_title}" def ticket_status_changed_summary(status_label: str) -> str: - """The ``summary`` field ``ticket_status_changed`` producers must share. + """Build the stable summary for a ticket-status transition. - ``status_label`` is the ``common_lookup_value`` label (Open, In - progress, Closed), never the raw code. A missing lookup already - fell back to the code in ``_attach_status_labels``; this helper - does not invent a name. + ``status_label`` is the ``common_lookup_value`` label (Open, In progress, + Closed), never the raw code. A missing lookup already fell back to the code + in ``_attach_status_labels``; this helper does not invent a replacement. """ return f"Ticket status changed to {status_label}" -def _activity_fields(event_type: str, actor_account_id: str, summary: str) -> dict[str, str]: +def _activity_text(value: Any, field_name: str) -> str: + """Require one exact string before it participates in activity identity. + + Valkey accepts several scalar types, so implicit coercion can collapse a + malformed numeric identity such as ``7`` onto the distinct canonical string + identity ``"7"``. Reject that alias before any stream read or mutation. + """ + if type(value) is not str: + raise TypeError(f"{field_name} must be a string") + return value + + +def _activity_event_count(value: Any) -> int: + """Validate one bounded buyer-facing activity-stream read count. + + Redis accepts integer-like values at a lower protocol layer, but the product + read contract must not coerce booleans, floats, or strings into a request + budget. The upper bound matches the retained stream window so callers cannot + request work beyond the product's own retention contract. + """ + if type(value) is not int: + raise TypeError("event_count must be an integer") + if not 1 <= value <= _MAX_ACTIVITY_READ_COUNT: + raise ValueError( + f"event_count must be between 1 and {_MAX_ACTIVITY_READ_COUNT}" + ) + return value + + +def _activity_fields( + event_type: str, + actor_account_id: str, + activity_summary: str, +) -> dict[str, str]: + """Translate semantic activity values to the established Valkey wire shape. + + Internal names may become more specific, but the persisted ``summary`` key + is compatibility-sensitive. This adapter is the only intentional mapping + between the bounded-context name and that historical field name. Identity + and display fields cross the Valkey boundary as exact strings rather than + being coerced from other scalar types. + """ return { - "event_type": event_type, - "actor_account_id": actor_account_id, - "summary": summary, + "event_type": _activity_text(event_type, "event_type"), + "actor_account_id": _activity_text(actor_account_id, "actor_account_id"), + "summary": _activity_text(activity_summary, "activity_summary"), } +def _matches_activity_fields( + activity_fields: dict[str, str], + expected_fields: dict[str, str], +) -> bool: + """Return whether a retained wire record has the legacy reseed identity.""" + return all( + activity_fields.get(field_name) == expected_value + for field_name, expected_value in expected_fields.items() + ) + + async def publish_activity_event( - client: redis.Redis, + valkey_client: redis.Redis, post_id: str, event_type: str, actor_account_id: str, - summary: str, + activity_summary: str, ) -> str: - """``XADD`` one event onto the post's stream. Returns the entry id. + """Append one activity fact to the post stream and return its entry id. - Approximately trimmed to the most recent 1000 entries (``maxlen``, - ``approximate=True``) so one very active post's stream can't grow - without bound -- the panel only ever shows the most recent 50 anyway. + The ordinary runtime path never performs reseed deduplication: each accepted + application event is appended once by its caller. Approximate trimming keeps + one very active post from growing without bound while preserving the recent + activity window consumed by the UI. """ with traced( "lineageweave.valkey.activity_xadd", {"db.system": "redis", "db.operation.name": "xadd", "lineageweave.stream.kind": "activity"}, ): - return await client.xadd( + return await valkey_client.xadd( _stream_key(post_id), - _activity_fields(event_type, actor_account_id, summary), + _activity_fields(event_type, actor_account_id, activity_summary), maxlen=1000, approximate=True, ) def publish_activity_event_sync( - client: Any, + valkey_client: Any, post_id: str, event_type: str, actor_account_id: str, - summary: str, + activity_summary: str, ) -> str | None: - """Sync ``XADD`` for ``make seed``. Returns None if ``summary`` is already on the stream.""" - key = _stream_key(post_id) - with traced( - "lineageweave.valkey.activity_xrevrange", - {"db.system": "redis", "db.operation.name": "xrevrange", "lineageweave.stream.kind": "activity"}, - ): - existing = client.xrevrange(key, count=50) - if any(fields.get("summary") == summary for _entry_id, fields in existing): - return None - with traced( - "lineageweave.valkey.activity_xadd", - {"db.system": "redis", "db.operation.name": "xadd", "lineageweave.stream.kind": "activity"}, - ): - return client.xadd( - key, - _activity_fields(event_type, str(actor_account_id), summary), - maxlen=1000, - approximate=True, + """Append one replay-safe seed/admin activity with optimistic concurrency. + + ``make seed`` can run in more than one process. A plain ``XREVRANGE`` then + ``XADD`` check is racy because two callers can read the same old snapshot + and both append. The synchronous seed path therefore WATCHes the post stream, + reads the retained identity tuple, and commits the append with MULTI/EXEC. + A concurrent mutation invalidates the watched snapshot and retries against + fresh stream state, but persistent contention fails after a bounded number + of attempts rather than leaving an operator command spinning indefinitely. + A recovered WATCH conflict is expected concurrency control, not a failed + Valkey operation, so it does not leave the bounded XADD span in an error + state. Exhausting the retry budget marks only the terminal XADD attempt as a + safe RuntimeError failure span. Ordinary async event publication remains + append-only. Contention failures identify the operation and retry limit + without embedding the post-scoped Valkey key in exception text or an + exception cause that may be exported by logging or telemetry. + """ + stream_key = _stream_key(post_id) + expected_fields = _activity_fields( + event_type, + actor_account_id, + activity_summary, + ) + watch_retry_exhausted = False + + for watch_attempt in range(1, _SYNC_ACTIVITY_WATCH_RETRY_LIMIT + 1): + with valkey_client.pipeline() as transaction: + try: + transaction.watch(stream_key) + with traced( + "lineageweave.valkey.activity_xrevrange", + { + "db.system": "redis", + "db.operation.name": "xrevrange", + "lineageweave.stream.kind": "activity", + }, + ): + existing_entries = transaction.xrevrange(stream_key) + + if any( + _matches_activity_fields(activity_fields, expected_fields) + for _entry_id, activity_fields in existing_entries + ): + transaction.unwatch() + return None + + transaction.multi() + transaction.xadd( + stream_key, + expected_fields, + maxlen=1000, + approximate=True, + ) + watch_conflicted = False + committed_entries: list[str] = [] + with traced( + "lineageweave.valkey.activity_xadd", + { + "db.system": "redis", + "db.operation.name": "xadd", + "lineageweave.stream.kind": "activity", + }, + ): + try: + committed_entries = transaction.execute() + except WatchError: + watch_conflicted = True + if ( + watch_conflicted + and watch_attempt == _SYNC_ACTIVITY_WATCH_RETRY_LIMIT + ): + raise RuntimeError( + "Activity reseed exceeded " + f"{_SYNC_ACTIVITY_WATCH_RETRY_LIMIT} WATCH retries" + ) + + if watch_conflicted: + continue + return committed_entries[0] + except WatchError: + if watch_attempt == _SYNC_ACTIVITY_WATCH_RETRY_LIMIT: + watch_retry_exhausted = True + break + + if watch_retry_exhausted: + raise RuntimeError( + "Activity reseed exceeded " + f"{_SYNC_ACTIVITY_WATCH_RETRY_LIMIT} WATCH retries" ) + raise RuntimeError( + "Activity reseed exhausted its bounded WATCH retry loop" + ) + -async def read_activity_events(client: redis.Redis, post_id: str, count: int = 50) -> list[dict[str, Any]]: - """The post's most recent events, newest first.""" +async def read_activity_events( + valkey_client: redis.Redis, + post_id: str, + event_count: int = 50, +) -> list[dict[str, Any]]: + """Read the newest retained activity events for one post. + + ``event_count`` is an exact 1..1000 buyer-facing read budget matching the + retained stream window. Invalid values fail before Valkey access; this + function does not broaden the query to other posts or reconstruct missing + facts. UUID reads include a bounded compatibility bridge for historical + UUID-equivalent alias streams recorded by startup while all current writers + remain canonical-only. Cross-stream chronology is comparable at millisecond + precision only; equal-millisecond historical ties use deterministic + canonical-first stream precedence rather than pretending stream-local + sequence counters form a global clock. The final buyer limit is applied + only after this bounded merge. Canonical entries retain their historical + public ``event_id``; alias entries are namespaced by bounded stream ordinal + because Valkey stream IDs are not globally unique across independent keys. + """ + bounded_event_count = _activity_event_count(event_count) + initial_uuid_read = await _activity_initial_uuid_read( + valkey_client, + post_id, + bounded_event_count, + ) + prefetched_canonical_entries: list[tuple[str, dict[str, str]]] | None = None + if initial_uuid_read is not None: + canonical_entries, indexed_aliases = initial_uuid_read + if not indexed_aliases: + return [ + { + "event_id": entry_id, + "event_type": activity_fields["event_type"], + "actor_account_id": activity_fields["actor_account_id"], + "summary": activity_fields["summary"], + } + for entry_id, activity_fields in canonical_entries + ] + canonical_key = _stream_key(post_id) + stream_keys = (canonical_key, *indexed_aliases) + prefetched_canonical_entries = canonical_entries + else: + stream_keys = await _activity_read_stream_keys(valkey_client, post_id) + + if len(stream_keys) == 1: + with traced( + "lineageweave.valkey.activity_xrevrange", + { + "db.system": "redis", + "db.operation.name": "xrevrange", + "lineageweave.stream.kind": "activity", + }, + ): + canonical_entries = await valkey_client.xrevrange( + stream_keys[0], count=bounded_event_count + ) + return [ + { + "event_id": entry_id, + "event_type": activity_fields["event_type"], + "actor_account_id": activity_fields["actor_account_id"], + "summary": activity_fields["summary"], + } + for entry_id, activity_fields in canonical_entries + ] + + pipeline_factory = getattr(valkey_client, "pipeline", None) + if pipeline_factory is not None: + prefetch_count = max( + 1, + min( + bounded_event_count, + _MAX_ACTIVITY_READ_COUNT // len(stream_keys), + ), + ) + if prefetched_canonical_entries is not None: + stream_pages = [prefetched_canonical_entries[:prefetch_count]] + async with pipeline_factory(transaction=False) as read_pipeline: + for stream_key in stream_keys[1:]: + read_pipeline.xrevrange(stream_key, count=prefetch_count) + with traced( + "lineageweave.valkey.activity_read_pipeline", + { + "db.system": "redis", + "db.operation.name": "pipeline", + "lineageweave.stream.kind": "activity", + }, + ): + stream_pages.extend(await read_pipeline.execute()) + else: + async with pipeline_factory(transaction=False) as read_pipeline: + for stream_key in stream_keys: + read_pipeline.xrevrange(stream_key, count=prefetch_count) + with traced( + "lineageweave.valkey.activity_read_pipeline", + { + "db.system": "redis", + "db.operation.name": "pipeline", + "lineageweave.stream.kind": "activity", + }, + ): + stream_pages = await read_pipeline.execute() + + next_entries: list[tuple[tuple[str, dict[str, str]], int]] = [] + stream_positions = [0] * len(stream_keys) + with traced( + "lineageweave.valkey.activity_xrevrange", + { + "db.system": "redis", + "db.operation.name": "xrevrange", + "lineageweave.stream.kind": "activity", + }, + ): + while len(next_entries) < bounded_event_count: + available = [ + (stream_pages[stream_index][stream_position], stream_index) + for stream_index, stream_position in enumerate(stream_positions) + if stream_position < len(stream_pages[stream_index]) + ] + if not available: + break + newest_entry, newest_stream_index = max( + available, + key=lambda item: _activity_compatibility_merge_order( + item[0][0], item[1] + ), + ) + next_entries.append((newest_entry, newest_stream_index)) + stream_positions[newest_stream_index] += 1 + + if ( + len(next_entries) < bounded_event_count + and stream_positions[newest_stream_index] + >= len(stream_pages[newest_stream_index]) + and len(stream_pages[newest_stream_index]) == prefetch_count + ): + next_page = await valkey_client.xrevrange( + stream_keys[newest_stream_index], + max=f"({newest_entry[0]}", + count=prefetch_count, + ) + stream_pages[newest_stream_index] = next_page + stream_positions[newest_stream_index] = 0 + + return [ + { + "event_id": _activity_public_event_id(entry_id, stream_index), + "event_type": activity_fields["event_type"], + "actor_account_id": activity_fields["actor_account_id"], + "summary": activity_fields["summary"], + } + for (entry_id, activity_fields), stream_index in next_entries + ] + + next_entries: list[tuple[tuple[str, dict[str, str]], int]] = [] + stream_results: list[tuple[str, dict[str, str]] | None] = [None] * len(stream_keys) with traced( "lineageweave.valkey.activity_xrevrange", {"db.system": "redis", "db.operation.name": "xrevrange", "lineageweave.stream.kind": "activity"}, ): - entries = await client.xrevrange(_stream_key(post_id), count=count) + first_pages = await asyncio.gather( + *(valkey_client.xrevrange(stream_key, count=1) for stream_key in stream_keys) + ) + for stream_index, page in enumerate(first_pages): + if page: + stream_results[stream_index] = page[0] + + while len(next_entries) < bounded_event_count: + available = [ + (entry, stream_index) + for stream_index, entry in enumerate(stream_results) + if entry is not None + ] + if not available: + break + newest_entry, newest_stream_index = max( + available, + key=lambda item: _activity_compatibility_merge_order(item[0][0], item[1]), + ) + next_entries.append((newest_entry, newest_stream_index)) + next_page = await valkey_client.xrevrange( + stream_keys[newest_stream_index], + max=f"({newest_entry[0]}", + count=1, + ) + stream_results[newest_stream_index] = next_page[0] if next_page else None return [ { - "event_id": entry_id, - "event_type": fields["event_type"], - "actor_account_id": fields["actor_account_id"], - "summary": fields["summary"], + "event_id": _activity_public_event_id(entry_id, stream_index), + "event_type": activity_fields["event_type"], + "actor_account_id": activity_fields["actor_account_id"], + "summary": activity_fields["summary"], } - for entry_id, fields in entries + for (entry_id, activity_fields), stream_index in next_entries ] diff --git a/backend/app/main.py b/backend/app/main.py index 122165990..af8db62ff 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -42,6 +42,7 @@ from backend.app.activity_stream import ( create_valkey_client, get_valkey, + index_legacy_activity_stream_aliases, publish_activity_event, read_activity_events, ticket_created_summary, @@ -297,6 +298,7 @@ async def lifespan(app: FastAPI): pool = await create_pool(settings.database_url) app.state.pool = pool valkey = create_valkey_client(settings.valkey_url) + await index_legacy_activity_stream_aliases(valkey) app.state.valkey = valkey analysis_worker = asyncio.create_task( run_analysis_run_worker( diff --git a/docs/adr/0363-canonical-activity-stream-identity.md b/docs/adr/0363-canonical-activity-stream-identity.md new file mode 100644 index 000000000..cb229092a --- /dev/null +++ b/docs/adr/0363-canonical-activity-stream-identity.md @@ -0,0 +1,178 @@ +# ADR 0363: Canonical activity-stream identity + +- Status: Proposed +- Date: 2026-09-04 + +## Context + +`source_post.post_id` is a PostgreSQL `uuid`, while historical HTTP requests +could spell the same UUID with upper-case digits, braces, omitted hyphens, or +other PostgreSQL-accepted hyphen placement. Earlier activity writes embedded +the request spelling in a Valkey stream key. Canonicalizing only new writes can +therefore strand authorized history, and enumerating selected spellings cannot +cover PostgreSQL's input grammar. + +The activity feed is a retained operational projection. It is not a source of +record truth, and its compatibility work must not broaden a request into a +keyspace scan. + +## Decision + +New activity writes use PostgreSQL's canonical lower-case, hyphenated UUID +form. Before the application begins serving reads, startup scans the existing +activity-key namespace once and records every actually present, UUID-equivalent +legacy key in a durable canonical-post alias set. A request reads the canonical +stream and only the aliases in that set. Non-UUID synthetic and legacy keys +retain exact spelling. + +Cross-stream ordering uses the persisted millisecond component and declared +canonical-first precedence for equal milliseconds. Stream-local sequence +numbers never become a global chronology. Alias event identifiers remain +namespaced so two streams cannot emit the same public identity. + +WATCH retries, retained-window read limits, and alias-index startup are fixed +product contracts. Failure to build the alias index makes the application +unready; the product does not silently hide historical activity. + +Alias discovery may find many independent legacy keys. Startup therefore keeps +at most 128 pending alias-set writes in memory and flushes each bounded group +through a non-transactional redis-py pipeline. `SADD` result integers are summed +exactly, preserving the existing count of newly indexed aliases. Minimal test +adapters without pipeline support retain the sequential compatibility fallback; +the production client created by `create_valkey_client` uses the batched path. +The batching changes transport cost only: discovery, durability, and readiness +semantics remain unchanged. + +The supported deployment is the repository's single `lineageweave` Compose +backend service. Deployment stops the preceding backend before starting the +replacement, so no pre-canonical writer may overlap the alias scan. A rolling +multi-version replica deployment is unavailable until it has a separate +writer-fencing contract. + +Reads enumerate at most the retained-window number of streams. For a UUID read, +the common path queues the first bounded alias-index `SSCAN` page and canonical +`XREVRANGE count=N` in one non-transactional redis-py pipeline. When that scan +returns cursor zero, its alias members and paired canonical page are both reused +by the same request. An empty member set completes the canonical-only response +in that one exchange. A complete non-empty member set enters compatibility +merge without scanning the alias index again or re-fetching the canonical page; +only retained alias pages are fetched in the next pipeline exchange. Reusing a +complete page does not bypass the existing fan-out ceiling: canonical stream plus +retained aliases may contain at most 1,000 distinct stream keys. A nonzero cursor +falls back to the bounded compatibility iterator and never treats an incomplete +alias scan as complete history. + +When aliases exist, production redis-py clients prefetch a bounded slice from +every admitted stream before merging locally. The per-stream page size is +`min(event_count, 1000 // stream_count)`, with a minimum of one. When the first +UUID pipeline already fetched the canonical page, that page is trimmed to this +per-stream budget before retained aliases are fetched, and the alias-only second +pipeline fills the remaining stream pages. Thus resident compatibility pages +still stay within the same 1,000-entry product ceiling. If one stream supplies +more than its prefetched share, only that exhausted stream is refilled with +another bounded page. Minimal adapters without pipeline support retain the older +one-entry incremental merger. Both paths preserve complete retained history, +canonical-first equal-millisecond ordering, and the final 1..1000 buyer-facing +output budget. + +## Consequences + +- Canonical reads retain every historical UUID spelling that actually exists. +- Startup alias writes use at most 128 pending `(index key, legacy stream key)` + tuples per production pipeline exchange instead of one awaited network write + per alias. +- The alias-free UUID buyer path admits compatibility metadata and fetches the + bounded canonical window in one redis-py pipeline network exchange. +- A complete retained-alias admission page is reused rather than discarded: the + one-alias compatibility fixture now needs two network exchanges, not three. +- Reused admission pages fail closed above 1,000 distinct canonical-plus-alias + stream keys before any retained-alias fan-out is issued. +- The retained-alias production path batches bounded history pages instead of + issuing one network command per stream and one additional command per returned + event. +- Compatibility prefetch retains at most 1,000 stream entries at a time before + output construction; skewed history refills only the stream that exhausted + its bounded page. +- Request-time records and calls remain bounded by the retained stream and output + limits; excessive alias cardinality is explicitly unavailable. +- Startup performs a cursor scan and must finish before readiness. +- The alias index is additional derived Valkey state and can be rebuilt from + retained activity keys. + +## Alternatives considered + +- Enumerate common UUID spellings: rejected because PostgreSQL accepts more + forms than a finite hand-picked list would honestly cover. +- Await one `SADD` for every discovered alias: rejected because rollout/readiness + latency then adds one serial Valkey write round trip per historical alias even + though those writes are independent. +- Buffer every discovered alias and issue one unbounded pipeline at the end: + rejected because startup memory and command-buffer size would scale with the + entire retained keyspace rather than a fixed batch. +- Perform an alias-index lookup and only afterward issue the canonical + `XREVRANGE`: rejected because the normal UUID path then has at least two + sequential Valkey network waits even when the alias set is empty. +- Discard a complete non-empty first alias page and rescan it before compatibility + merge: rejected because it adds a deterministic network exchange and repeats + admission work without improving history correctness. +- Reuse a complete alias page without reapplying the total-stream ceiling: + rejected because 1,000 aliases plus the canonical stream would silently expand + request fan-out beyond the bounded contract formerly enforced by the iterator. +- Fetch one compatibility entry per stream and then await one `XREVRANGE` per + returned event: rejected because an authorized 50-event legacy-history panel + still pays roughly one sequential Valkey wait per result. +- Fetch `event_count` entries from every compatibility stream in one pipeline: + rejected because worst-case buffering becomes `event_count * stream_count`, + up to one million retained records under the existing limits. +- Scan the activity keyspace on each request: rejected because latency and work + would scale with unrelated posts. +- Drop legacy aliases: rejected because canonicalization would hide retained + authorized history. + +## Implementation evidence + +- Review `5121693984` identified that the earlier read-budget test counted only + `XREVRANGE` and omitted the preceding alias-index `SSCAN` network wait. +- RED `7dcc5c16bc61393fc6a533e216baee8c43363894` counts both direct Valkey waits + and therefore rejects the earlier two-exchange canonical UUID path. +- Production repair `02417fcf92969ffe7924f2722b192f40ff953e0d` + pipelines bounded alias admission with the canonical page while preserving the + existing compatibility fallback. +- Test convergence `2b594942c12c37fe68459c02fbf47a2d11a727e4` models the pipeline as one + network exchange and retains the exact canonical stream/count assertion. +- Review `5121832019` identified the independent startup N+1 where alias + discovery awaited one `SADD` network write for every retained legacy key. +- RED `4cfa402acadb43c3f60e4dba7bad8917230b9460` rejects direct per-alias `SADD` + on a two-alias fixture while preserving exact durable members and added count. +- Production repair `c4a25ce1f03ee225874a56c3f102ffa8b48a0621` introduces a bounded 128-write + non-transactional pipeline batch with a compatibility fallback for minimal + adapters that do not expose redis-py pipeline support. +- Review `5121915874` identified the retained-alias buyer-path N+1 left after the + canonical fast-path repairs: initial probes were concurrent, but each emitted + event still triggered another awaited `XREVRANGE count=1`. +- RED `7d44315c81bd942db58af75ea9af9e9885c46dc5` requires a four-event, + two-stream compatibility read to preserve exact merge order without paying one + network exchange per event. +- Production repair `2ee36cf7844526a5f27803100f2b469e31573a98` batches bounded per-stream + history pages through one non-transactional pipeline and refills only an + exhausted stream, with total initial buffering capped at 1,000 entries. +- Review `5122107164` found that the retained-alias path still discarded the + complete first SSCAN page and canonical XREVRANGE, then repeated alias + admission and canonical retrieval before merging. +- RED `15746e2be27effca16c71f445f8d77f5d55fe6b8` requires the one-alias fixture + to preserve exact four-event ordering in exactly two network exchanges, with + pipeline batches `[2, 1]`. +- Production repair `07fa552f9fecbe1cda79a9fccc7eeacd82c4fba5` carries a complete first + alias page plus the canonical page into compatibility merge and fetches only + retained alias pages in the second pipeline exchange. +- Follow-up review `5122123572` found that this reuse path initially skipped the + iterator's existing total-stream fan-out guard. +- RED `a9c080c1f2fa030bd2b0cec0e2fb6b43e4328bbc` supplies a complete first page + with 1,000 aliases and requires fail-closed behavior after the first two-command + exchange, before a retained-alias pipeline is issued. +- Production repair `8ef3a42608739e5a32b5841e4a14494df5326a3f` reapplies the 1,000 distinct + canonical-plus-alias stream ceiling to reused admission pages. +- redis-py asyncio pipeline documentation specifies that pipeline commands are + buffered and executed together when awaited through `execute()`; transactions + remain optional and are not required for these independent read or set-write + commands. diff --git a/docs/adr/README.md b/docs/adr/README.md index 8979111c1..6f1203b7c 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -27,6 +27,7 @@ decision from them. | [`operability/http-concurrency-evidence.md`](../operability/http-concurrency-evidence.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0212](0212-single-query-authorized-post-filter-options.md), [0213](0213-global-ask-embedding-pool-release.md) | | [`operability/mcp-concurrency-evidence.md`](../operability/mcp-concurrency-evidence.md) | [0218](0218-current-contract-mcp-global-ask.md) | | Evidence operations Dashboard (`/`) | [0206](0206-evidence-operations-dashboard.md) | +| Activity stream (`backend/app/activity_stream.py`) | [0363](0363-canonical-activity-stream-identity.md) | | [`temporal-topic-context-influence-research.md`](../temporal-topic-context-influence-research.md) | [0210](0210-temporal-topic-context-influence-dashboard.md) | | [`python-mathematical-compute-boundary-audit.md`](../doctoring/python-mathematical-compute-boundary-audit.md) | [0208](0208-externalize-local-mathematical-compute.md) | | [`WORKER_FUNCTION_TAXONOMY_REFERENCES.md`](../doctoring/WORKER_FUNCTION_TAXONOMY_REFERENCES.md) | [0232](0232-worker-function-taxonomy-in-the-published-ontology.md) | diff --git a/tests/test_activity_stream.py b/tests/test_activity_stream.py index 40ea78f4f..ac0a06c02 100644 --- a/tests/test_activity_stream.py +++ b/tests/test_activity_stream.py @@ -7,8 +7,15 @@ from __future__ import annotations +from inspect import signature + +from redis.exceptions import WatchError + from backend.app.activity_stream import ( + create_valkey_client, + publish_activity_event, publish_activity_event_sync, + read_activity_events, ticket_created_summary, ticket_status_changed_summary, ) @@ -17,34 +24,139 @@ class _FakeStream: + """Small in-memory stand-in for the Valkey stream methods under contract.""" + def __init__(self) -> None: + """Start with no retained activity entries or key mutations.""" self.entries: list[tuple[str, dict[str, str]]] = [] + self.version = 0 - def xrevrange(self, key: str, count: int = 50): + def xrevrange(self, key: str, count: int | None = None): + """Return newest-first entries with the same optional count boundary.""" del key - return list(reversed(self.entries[-count:])) + entries = list(reversed(self.entries)) + return entries if count is None else entries[:count] def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None): + """Append a copied wire record and advance the watched key version.""" del key, maxlen, approximate entry_id = f"1-{len(self.entries)}" self.entries.append((entry_id, dict(fields))) + self.version += 1 return entry_id + def pipeline(self): + """Create the WATCH/MULTI pipeline used by synchronous reseeding.""" + return _FakePipeline(self) + + +class _FakePipeline: + """Model the redis-py WATCH/MULTI behavior needed by the reseed contract.""" + + def __init__(self, stream: _FakeStream) -> None: + """Bind the transaction to one fake stream client.""" + self.stream = stream + self.watched_version: int | None = None + self.pending_xadd: tuple[str, dict[str, str], object, object] | None = None + self.in_multi = False + + def __enter__(self): + """Return this transaction context.""" + return self + + def __exit__(self, exc_type, exc, traceback) -> bool: + """Never suppress transaction exceptions.""" + del exc_type, exc, traceback + return False + + def watch(self, key: str) -> None: + """Remember the stream version at the optimistic-lock boundary.""" + del key + self.watched_version = self.stream.version + + def unwatch(self) -> None: + """Release the optimistic lock after finding an existing fact.""" + self.watched_version = None + + def xrevrange(self, key: str, count: int | None = None): + """Read immediately while the key is watched, as redis-py does.""" + return self.stream.xrevrange(key, count=count) + + def multi(self) -> None: + """Begin queuing the append for optimistic execution.""" + self.in_multi = True + + def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None): + """Queue one append after MULTI rather than mutating immediately.""" + if not self.in_multi: + return self.stream.xadd(key, fields, maxlen=maxlen, approximate=approximate) + self.pending_xadd = (key, dict(fields), maxlen, approximate) + return self + + def execute(self): + """Reject a stale watched snapshot or atomically apply the queued append.""" + if self.watched_version != self.stream.version: + raise WatchError("synthetic concurrent stream mutation") + if self.pending_xadd is None: + return [] + key, fields, maxlen, approximate = self.pending_xadd + return [self.stream.xadd(key, fields, maxlen=maxlen, approximate=approximate)] + + +class _RaceInjectingStream(_FakeStream): + """Inject one concurrent seed after the caller reads its stale snapshot.""" + + def __init__(self, concurrent_fields: dict[str, str]) -> None: + """Remember the fact another seed execution will publish once.""" + super().__init__() + self.concurrent_fields = dict(concurrent_fields) + self.injected = False + + def xrevrange(self, key: str, count: int | None = None): + """Return the old snapshot, then simulate another process appending it.""" + entries = super().xrevrange(key, count=count) + if not self.injected: + self.injected = True + super().xadd(key, self.concurrent_fields) + return entries + + +def test_activity_stream_owned_parameters_use_semantic_names() -> None: + """Organization-owned activity helpers expose bounded-context vocabulary.""" + assert list(signature(create_valkey_client).parameters) == ["valkey_url"] + expected_event_parameters = [ + "valkey_client", + "post_id", + "event_type", + "actor_account_id", + "activity_summary", + ] + assert list(signature(publish_activity_event).parameters) == expected_event_parameters + assert list(signature(publish_activity_event_sync).parameters) == expected_event_parameters + assert list(signature(read_activity_events).parameters) == [ + "valkey_client", + "post_id", + "event_count", + ] + def test_ticket_created_summary_matches_the_live_api_wording() -> None: + """Ticket creation retains the wording consumed by the live Activity UI.""" assert ticket_created_summary("Send Northridge Grid the revised quote") == ( "Ticket created: Send Northridge Grid the revised quote" ) def test_ticket_status_changed_summary_uses_the_lookup_label() -> None: + """Status activity uses the resolved label rather than leaking its raw code.""" assert ticket_status_changed_summary("In progress") == ( "Ticket status changed to In progress" ) assert "in_progress" not in ticket_status_changed_summary("In progress") -def test_publish_activity_event_sync_skips_a_matching_summary() -> None: +def test_publish_activity_event_sync_skips_a_matching_activity_fact() -> None: + """An exact retained event-type, actor, and summary tuple is replay-idempotent.""" client = _FakeStream() first = publish_activity_event_sync( client, @@ -67,6 +179,140 @@ def test_publish_activity_event_sync_skips_a_matching_summary() -> None: assert "Send Northridge Grid the revised quote" in client.entries[0][1]["summary"] +def test_publish_activity_event_sync_keeps_distinct_events_with_same_summary() -> None: + """Reseeding must not collapse different activity facts onto summary text.""" + client = _FakeStream() + shared_summary = "Assignment updated" + client.xadd( + "activity:post-1", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-2", + "summary": shared_summary, + }, + ) + + created = publish_activity_event_sync( + client, + "post-1", + "ticket_created", + "acct-1", + shared_summary, + ) + + assert created == "1-1" + assert len(client.entries) == 2 + assert client.entries[-1][1] == { + "event_type": "ticket_created", + "actor_account_id": "acct-1", + "summary": shared_summary, + } + + +def test_publish_activity_event_sync_keeps_same_event_type_for_different_actor() -> None: + """Actor identity independently distinguishes otherwise equal reseed facts.""" + client = _FakeStream() + shared_summary = "Assignment updated" + client.xadd( + "activity:post-1", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-2", + "summary": shared_summary, + }, + ) + + created = publish_activity_event_sync( + client, + "post-1", + "ticket_status_changed", + "acct-1", + shared_summary, + ) + + assert created == "1-1" + assert client.entries[-1][1]["actor_account_id"] == "acct-1" + + +def test_publish_activity_event_sync_keeps_same_actor_for_different_event_type() -> None: + """Event type independently distinguishes otherwise equal reseed facts.""" + client = _FakeStream() + shared_summary = "Assignment updated" + client.xadd( + "activity:post-1", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-1", + "summary": shared_summary, + }, + ) + + created = publish_activity_event_sync( + client, + "post-1", + "ticket_created", + "acct-1", + shared_summary, + ) + + assert created == "1-1" + assert client.entries[-1][1]["event_type"] == "ticket_created" + + +def test_publish_activity_event_sync_scans_the_retained_stream_for_reseed_idempotency() -> None: + """A retained seed event stays idempotent after more than 50 newer events.""" + client = _FakeStream() + seed_summary = ticket_created_summary("Send Northridge Grid the revised quote") + assert publish_activity_event_sync( + client, + "post-1", + "ticket_created", + "acct-1", + seed_summary, + ) == "1-0" + + for index in range(51): + client.xadd( + "activity:post-1", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-1", + "summary": f"newer activity {index}", + }, + ) + + entry_count = len(client.entries) + assert publish_activity_event_sync( + client, + "post-1", + "ticket_created", + "acct-1", + seed_summary, + ) is None + assert len(client.entries) == entry_count + + +def test_publish_activity_event_sync_retries_when_another_seed_wins_the_race() -> None: + """Concurrent make-seed executions must not append the same activity twice.""" + expected_fields = { + "event_type": "ticket_created", + "actor_account_id": "acct-1", + "summary": ticket_created_summary("Send Northridge Grid the revised quote"), + } + client = _RaceInjectingStream(expected_fields) + + created = publish_activity_event_sync( + client, + "post-1", + expected_fields["event_type"], + expected_fields["actor_account_id"], + expected_fields["summary"], + ) + + assert created is None + assert client.entries == [("1-0", expected_fields)] + + def test_valkey_child_span_shares_parent_trace_id(monkeypatch) -> None: """Same-process Valkey work inherits the parent TraceId.""" from opentelemetry import trace @@ -75,7 +321,10 @@ def test_valkey_child_span_shares_parent_trace_id(monkeypatch) -> None: captured: dict[str, str] = {} class _Client(_FakeStream): + """Capture tracing identity at the fake Valkey append boundary.""" + def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None): + """Record the current child span before delegating to the fake stream.""" span = trace.get_current_span() captured["trace_id"] = format(span.get_span_context().trace_id, "032x") captured["span_id"] = format(span.get_span_context().span_id, "016x") diff --git a/tests/test_activity_stream_alias_index_budget.py b/tests/test_activity_stream_alias_index_budget.py new file mode 100644 index 000000000..144dd0078 --- /dev/null +++ b/tests/test_activity_stream_alias_index_budget.py @@ -0,0 +1,103 @@ +"""Startup legacy-activity alias indexing stays bounded at the Valkey wire.""" + +from __future__ import annotations + +import asyncio +from uuid import UUID + +from backend.app.activity_stream import index_legacy_activity_stream_aliases + + +class _AliasIndexPipeline: + """Model independent alias-set writes completed by one pipeline exchange.""" + + def __init__(self, client: _AliasIndexClient) -> None: + self.client = client + self.pending: list[tuple[str, str]] = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback) -> bool: + del exc_type, exc, traceback + return False + + def sadd(self, key: str, member: str): + self.pending.append((key, member)) + return self + + async def execute(self) -> list[int]: + self.client.pipeline_round_trips += 1 + self.client.pipeline_batch_sizes.append(len(self.pending)) + added: list[int] = [] + for key, member in self.pending: + members = self.client.members.setdefault(key, set()) + before = len(members) + members.add(member) + added.append(int(len(members) != before)) + return added + + +class _AliasIndexClient: + """Expose historical aliases while rejecting per-alias network writes.""" + + def __init__(self, keys: tuple[str, ...] | None = None) -> None: + self.keys = keys or ( + "activity:550e8400-e29b-41d4-a716-446655440000", + "activity:550e8400e29b41d4a716446655440000", + "activity:{550E8400-E29B-41D4-A716-446655440000}", + ) + self.members: dict[str, set[str]] = {} + self.direct_sadd_calls = 0 + self.pipeline_round_trips = 0 + self.pipeline_batch_sizes: list[int] = [] + + async def scan_iter(self, *, match: str): + assert match == "activity:*" + for key in self.keys: + yield key + + async def sadd(self, key: str, member: str) -> int: + del key, member + self.direct_sadd_calls += 1 + raise AssertionError("startup alias indexing must not await one SADD per alias") + + def pipeline(self, *, transaction: bool): + assert transaction is False + return _AliasIndexPipeline(self) + + +def test_startup_alias_index_batches_independent_sadd_writes() -> None: + """Two retained aliases must not add two serial Valkey write round trips.""" + client = _AliasIndexClient() + + indexed = asyncio.run(index_legacy_activity_stream_aliases(client)) # type: ignore[arg-type] + + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + assert indexed == 2 + assert client.direct_sadd_calls == 0 + assert client.pipeline_round_trips == 1 + assert client.pipeline_batch_sizes == [2] + assert client.members == { + f"activity-aliases:{canonical_post_id}": { + "activity:550e8400e29b41d4a716446655440000", + "activity:{550E8400-E29B-41D4-A716-446655440000}", + } + } + + +def test_startup_alias_index_flushes_before_exceeding_the_batch_ceiling() -> None: + """The 129th independent alias starts a second bounded pipeline exchange.""" + aliases = tuple( + f"activity:{{{str(UUID(int=index + 1)).upper()}}}" + for index in range(129) + ) + client = _AliasIndexClient(aliases) + + indexed = asyncio.run(index_legacy_activity_stream_aliases(client)) # type: ignore[arg-type] + + assert indexed == 129 + assert client.direct_sadd_calls == 0 + assert client.pipeline_round_trips == 2 + assert client.pipeline_batch_sizes == [128, 1] + assert sum(len(members) for members in client.members.values()) == 129 diff --git a/tests/test_activity_stream_compatibility_budget.py b/tests/test_activity_stream_compatibility_budget.py new file mode 100644 index 000000000..783c3c5a6 --- /dev/null +++ b/tests/test_activity_stream_compatibility_budget.py @@ -0,0 +1,132 @@ +"""Retained UUID-alias reads stay bounded at the Valkey wire.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from backend.app.activity_stream import read_activity_events + + +class _CompatibilityPipeline: + """Model queued SSCAN/XREVRANGE commands as one network exchange.""" + + def __init__(self, client: _CompatibilityReadClient) -> None: + self.client = client + self.commands: list[tuple[str, tuple[object, ...], dict[str, object]]] = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback) -> bool: + del exc_type, exc, traceback + return False + + def sscan(self, key: str, *, cursor: int, count: int): + self.commands.append(("sscan", (key,), {"cursor": cursor, "count": count})) + return self + + def xrevrange(self, key: str, *, count: int, max: str = "+"): + self.commands.append(("xrevrange", (key,), {"count": count, "max": max})) + return self + + async def execute(self): + self.client.round_trips += 1 + self.client.pipeline_batch_sizes.append(len(self.commands)) + results: list[object] = [] + for command, args, kwargs in self.commands: + key = str(args[0]) + if command == "sscan": + results.append((0, self.client.aliases)) + continue + entries = self.client.entries_by_key[key] + max_value = str(kwargs["max"]) + if max_value.startswith("("): + boundary = max_value[1:] + entries = [entry for entry in entries if entry[0] != boundary] + results.append(entries[: int(kwargs["count"])]) + return results + + +class _CompatibilityReadClient: + """Expose one canonical stream and one retained UUID-equivalent alias.""" + + def __init__(self) -> None: + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + self.canonical_key = f"activity:{canonical_post_id}" + self.legacy_key = "activity:{550E8400-E29B-41D4-A716-446655440000}" + self.aliases = [self.legacy_key] + self.round_trips = 0 + self.pipeline_batch_sizes: list[int] = [] + self.entries_by_key = { + self.canonical_key: [ + ("400-0", {"event_type": "status", "actor_account_id": "acct", "summary": "canonical 400"}), + ("200-0", {"event_type": "status", "actor_account_id": "acct", "summary": "canonical 200"}), + ], + self.legacy_key: [ + ("300-0", {"event_type": "created", "actor_account_id": "acct", "summary": "legacy 300"}), + ("100-0", {"event_type": "created", "actor_account_id": "acct", "summary": "legacy 100"}), + ], + } + + def pipeline(self, *, transaction: bool): + assert transaction is False + return _CompatibilityPipeline(self) + + async def sscan_iter(self, key: str): + assert key == "activity-aliases:550e8400-e29b-41d4-a716-446655440000" + self.round_trips += 1 + for alias in self.aliases: + yield alias + + async def xrevrange(self, key: str, *, count: int, max: str = "+"): + self.round_trips += 1 + entries = self.entries_by_key[key] + if max.startswith("("): + boundary = max[1:] + entries = [entry for entry in entries if entry[0] != boundary] + return entries[:count] + + +def test_retained_alias_read_reuses_complete_admission_page_before_local_merge() -> None: + """A complete alias admission page must not be discarded and scanned again.""" + client = _CompatibilityReadClient() + + events = asyncio.run( + read_activity_events( + client, # type: ignore[arg-type] + "550E8400-E29B-41D4-A716-446655440000", + event_count=4, + ) + ) + + assert [event["summary"] for event in events] == [ + "canonical 400", + "legacy 300", + "canonical 200", + "legacy 100", + ] + assert client.round_trips == 2 + assert client.pipeline_batch_sizes == [2, 1] + + +def test_complete_admission_page_preserves_total_stream_fanout_ceiling() -> None: + """Reusing SSCAN results must not bypass the canonical-plus-alias stream cap.""" + client = _CompatibilityReadClient() + client.aliases = [f"activity:legacy-{index}" for index in range(1000)] + + with pytest.raises( + RuntimeError, + match="Activity history has too many retained compatibility streams", + ): + asyncio.run( + read_activity_events( + client, # type: ignore[arg-type] + "550E8400-E29B-41D4-A716-446655440000", + event_count=4, + ) + ) + + assert client.round_trips == 1 + assert client.pipeline_batch_sizes == [2] diff --git a/tests/test_activity_stream_identity_types.py b/tests/test_activity_stream_identity_types.py new file mode 100644 index 000000000..21ceb9b73 --- /dev/null +++ b/tests/test_activity_stream_identity_types.py @@ -0,0 +1,323 @@ +"""Fail-closed type and canonical-key boundaries for Valkey activity identity fields.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from backend.app.activity_stream import ( + index_legacy_activity_stream_aliases, + publish_activity_event, + publish_activity_event_sync, + read_activity_events, + ticket_created_summary, +) + + +class _UnexpectedValkeyAccess: + """Fail if malformed identity reaches the synchronous Valkey boundary.""" + + def pipeline(self): + """Prove validation happens before any stream read or mutation.""" + raise AssertionError("malformed activity identity reached Valkey") + + +class _UnexpectedAsyncValkeyAccess: + """Fail if malformed identity reaches the ordinary async append boundary.""" + + async def xadd(self, *args, **kwargs): + """Prove runtime publication validates identity before issuing XADD.""" + del args, kwargs + raise AssertionError("malformed activity identity reached async Valkey") + + +class _RecordingAsyncValkey: + """Record stream keys so equivalent source-post UUID spellings cannot split identity.""" + + def __init__(self) -> None: + """Start without observed Valkey keys.""" + self.keys: list[str] = [] + + async def xadd(self, key: str, *args, **kwargs) -> str: + """Capture the selected stream key and return a stable fake entry id.""" + del args, kwargs + self.keys.append(key) + return f"1-{len(self.keys) - 1}" + + +class _LegacyAliasReadValkey: + """Expose canonical and pre-canonical UUID streams to the current reader.""" + + def __init__(self, entries_by_key: dict[str, list[tuple[str, dict[str, str]]]]) -> None: + """Keep immutable-looking fixtures and record each bounded read.""" + self.entries_by_key = entries_by_key + self.reads: list[tuple[str, int]] = [] + + async def xrevrange(self, key: str, *, count: int, max: str = "+"): + """Return the newest fake entries for one exact requested stream key.""" + self.reads.append((key, count)) + entries = list(self.entries_by_key.get(key, ())) + if max.startswith("("): + boundary = max[1:] + entries = [entry for entry in entries if entry[0] != boundary] + return entries[:count] + + async def sscan_iter(self, key: str): + """Return the durable aliases indexed for the canonical UUID.""" + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + assert key == f"activity-aliases:{canonical_post_id}" + for stream_key in { + stream_key + for stream_key in self.entries_by_key + if stream_key != f"activity:{canonical_post_id}" + }: + yield stream_key + + +class _LegacyAliasIndexValkey: + """Model cursor discovery and durable alias-set writes at startup.""" + + def __init__(self) -> None: + """Expose canonical, compact, braced, and non-UUID retained streams.""" + self.keys = ( + "activity:550e8400-e29b-41d4-a716-446655440000", + "activity:550e8400e29b41d4a716446655440000", + "activity:{550E8400-E29B-41D4-A716-446655440000}", + "activity:synthetic-post", + "activity-aliases:550e8400-e29b-41d4-a716-446655440000", + ) + self.members: dict[str, set[str]] = {} + + async def scan_iter(self, *, match: str): + """Yield the finite retained keyspace selected by the namespace pattern.""" + assert match == "activity:*" + for key in self.keys: + yield key + + async def sadd(self, key: str, member: str) -> int: + """Record one discovered alias with Redis-compatible added count.""" + members = self.members.setdefault(key, set()) + before = len(members) + members.add(member) + return int(len(members) != before) + + +def test_sync_activity_rejects_numeric_actor_identity_before_valkey_access() -> None: + """A numeric actor id must not alias the canonical string identity ``\"7\"``.""" + with pytest.raises(TypeError, match="actor_account_id must be a string"): + publish_activity_event_sync( + _UnexpectedValkeyAccess(), + "post-1", + "ticket_created", + 7, # type: ignore[arg-type] + ticket_created_summary("Send Northridge Grid the revised quote"), + ) + + +def test_sync_activity_rejects_numeric_post_identity_before_valkey_access() -> None: + """A numeric post id must not alias the canonical string stream identity ``\"7\"``.""" + with pytest.raises(TypeError, match="post_id must be a string"): + publish_activity_event_sync( + _UnexpectedValkeyAccess(), + 7, # type: ignore[arg-type] + "ticket_created", + "acct-1", + ticket_created_summary("Send Northridge Grid the revised quote"), + ) + + +def test_async_activity_rejects_numeric_actor_identity_before_xadd() -> None: + """Ordinary runtime publication shares the exact actor-identity admission rule.""" + with pytest.raises(TypeError, match="actor_account_id must be a string"): + asyncio.run( + publish_activity_event( + _UnexpectedAsyncValkeyAccess(), # type: ignore[arg-type] + "post-1", + "ticket_created", + 7, # type: ignore[arg-type] + ticket_created_summary("Send Northridge Grid the revised quote"), + ) + ) + + +def test_activity_stream_key_collapses_equivalent_source_post_uuid_spellings() -> None: + """One PostgreSQL UUID identity must never fork into case-variant Valkey streams.""" + client = _RecordingAsyncValkey() + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + uppercase_post_id = canonical_post_id.upper() + + async def publish_both_spellings() -> None: + await publish_activity_event( + client, # type: ignore[arg-type] + canonical_post_id, + "ticket_created", + "acct-1", + "Ticket created: canonical UUID", + ) + await publish_activity_event( + client, # type: ignore[arg-type] + uppercase_post_id, + "ticket_created", + "acct-1", + "Ticket created: equivalent UUID spelling", + ) + + asyncio.run(publish_both_spellings()) + + assert client.keys == [ + f"activity:{canonical_post_id}", + f"activity:{canonical_post_id}", + ] + + +def test_startup_indexes_every_existing_uuid_alias_without_guessing_forms() -> None: + """Compact and braced historical keys become durable canonical-read aliases.""" + client = _LegacyAliasIndexValkey() + + indexed = asyncio.run(index_legacy_activity_stream_aliases(client)) # type: ignore[arg-type] + + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + assert indexed == 2 + assert client.members == { + f"activity-aliases:{canonical_post_id}": { + "activity:550e8400e29b41d4a716446655440000", + "activity:{550E8400-E29B-41D4-A716-446655440000}", + } + } + + +def test_activity_read_preserves_precanonical_uppercase_uuid_alias_events() -> None: + """Canonical reads retain events written to the historical uppercase UUID key.""" + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + uppercase_post_id = canonical_post_id.upper() + canonical_key = f"activity:{canonical_post_id}" + legacy_key = f"activity:{uppercase_post_id}" + client = _LegacyAliasReadValkey( + { + canonical_key: [ + ( + "200-0", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-1", + "summary": "Ticket status changed to Closed", + }, + ) + ], + legacy_key: [ + ( + "100-0", + { + "event_type": "ticket_created", + "actor_account_id": "acct-1", + "summary": "Ticket created: legacy uppercase route", + }, + ) + ], + } + ) + + events = asyncio.run( + read_activity_events( + client, # type: ignore[arg-type] + canonical_post_id, + event_count=10, + ) + ) + + assert [event["event_id"] for event in events] == ["200-0", "legacy-1:100-0"] + assert client.reads == [ + (canonical_key, 1), + (legacy_key, 1), + (canonical_key, 1), + (legacy_key, 1), + ] + + +def test_activity_read_does_not_compare_cross_stream_sequence_numbers() -> None: + """Same-millisecond alias ties use deterministic stream precedence, not local sequence.""" + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + uppercase_post_id = canonical_post_id.upper() + canonical_key = f"activity:{canonical_post_id}" + legacy_key = f"activity:{uppercase_post_id}" + client = _LegacyAliasReadValkey( + { + canonical_key: [ + ( + "500-0", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-1", + "summary": "Canonical current stream", + }, + ) + ], + legacy_key: [ + ( + "500-99", + { + "event_type": "ticket_created", + "actor_account_id": "acct-1", + "summary": "Historical alias stream", + }, + ) + ], + } + ) + + events = asyncio.run( + read_activity_events( + client, # type: ignore[arg-type] + canonical_post_id, + event_count=1, + ) + ) + + assert [event["summary"] for event in events] == ["Canonical current stream"] + + +def test_activity_read_namespaces_colliding_legacy_stream_entry_ids() -> None: + """Independent Valkey streams must not emit duplicate buyer event identities.""" + canonical_post_id = "550e8400-e29b-41d4-a716-446655440000" + uppercase_post_id = canonical_post_id.upper() + canonical_key = f"activity:{canonical_post_id}" + legacy_key = f"activity:{uppercase_post_id}" + client = _LegacyAliasReadValkey( + { + canonical_key: [ + ( + "600-0", + { + "event_type": "ticket_status_changed", + "actor_account_id": "acct-1", + "summary": "Canonical current stream", + }, + ) + ], + legacy_key: [ + ( + "600-0", + { + "event_type": "ticket_created", + "actor_account_id": "acct-1", + "summary": "Historical alias stream", + }, + ) + ], + } + ) + + events = asyncio.run( + read_activity_events( + client, # type: ignore[arg-type] + canonical_post_id, + event_count=2, + ) + ) + + assert [event["event_id"] for event in events] == [ + "600-0", + "legacy-1:600-0", + ] + assert len({event["event_id"] for event in events}) == 2 diff --git a/tests/test_activity_stream_read_budget.py b/tests/test_activity_stream_read_budget.py new file mode 100644 index 000000000..a33af887b --- /dev/null +++ b/tests/test_activity_stream_read_budget.py @@ -0,0 +1,157 @@ +"""Buyer-facing activity reads stay bounded before reaching Valkey.""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest + +from backend.app.activity_stream import read_activity_events + + +class _ReadClient: + """Record whether an invalid read request reaches the Valkey boundary.""" + + def __init__(self) -> None: + self.calls = 0 + + async def xrevrange( + self, + key: str, + max: str = "+", + min: str = "-", + count: int | None = None, + ) -> list[tuple[str, dict[str, str]]]: + del key, max, min, count + self.calls += 1 + return [] + + +class _PopulatedReadClient: + """Expose one canonical UUID stream and no retained compatibility aliases.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, str, int | None]] = [] + self.round_trips = 0 + self.entries = [ + ("300-0", {"event_type": "third", "actor_account_id": "actor", "summary": "third"}), + ("200-0", {"event_type": "second", "actor_account_id": "actor", "summary": "second"}), + ("100-0", {"event_type": "first", "actor_account_id": "actor", "summary": "first"}), + ] + + def pipeline(self, *, transaction: bool): + """Queue alias admission and canonical data read onto one network exchange.""" + assert transaction is False + return _ReadPipeline(self) + + async def sscan_iter(self, key: str): + """Count any compatibility fallback alias scan as another exchange.""" + del key + self.round_trips += 1 + if False: + yield "" + + async def xrevrange( + self, + key: str, + max: str = "+", + min: str = "-", + count: int | None = None, + ) -> list[tuple[str, dict[str, str]]]: + del min + self.round_trips += 1 + self.calls.append((key, max, count)) + if max == "+": + eligible = self.entries + else: + upper = max.removeprefix("(") + upper_ms = int(upper.partition("-")[0]) + eligible = [entry for entry in self.entries if int(entry[0].partition("-")[0]) < upper_ms] + return eligible[:count] + + +class _ReadPipeline: + """Model two queued Redis commands completed by one pipeline exchange.""" + + def __init__(self, client: _PopulatedReadClient) -> None: + self.client = client + self.alias_key: str | None = None + self.read_request: tuple[str, str, int | None] | None = None + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback) -> bool: + del exc_type, exc, traceback + return False + + def sscan(self, key: str, *, cursor: int, count: int): + assert cursor == 0 + assert count == 1000 + self.alias_key = key + return self + + def xrevrange( + self, + key: str, + max: str = "+", + min: str = "-", + count: int | None = None, + ): + del min + self.read_request = (key, max, count) + return self + + async def execute(self): + assert self.alias_key == "activity-aliases:aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" + assert self.read_request is not None + key, max_value, count = self.read_request + self.client.round_trips += 1 + self.client.calls.append((key, max_value, count)) + eligible = self.client.entries + return [(0, []), eligible[:count]] + + +@pytest.mark.parametrize("event_count", (True, 1.0, "50")) +def test_activity_read_count_requires_an_exact_integer(event_count: Any) -> None: + """Scalar coercion must not create an implicit buyer-facing read budget.""" + client = _ReadClient() + + with pytest.raises(TypeError, match="event_count must be an integer"): + asyncio.run(read_activity_events(client, "post-1", event_count=event_count)) + + assert client.calls == 0 + + +@pytest.mark.parametrize("event_count", (0, -1, 1001)) +def test_activity_read_count_is_bounded_to_the_retained_window( + event_count: int, +) -> None: + """Invalid counts fail before an unbounded or nonsensical Valkey read.""" + client = _ReadClient() + + with pytest.raises(ValueError, match="event_count must be between 1 and 1000"): + asyncio.run(read_activity_events(client, "post-1", event_count=event_count)) + + assert client.calls == 0 + + +def test_activity_read_count_accepts_the_retained_window_ceiling() -> None: + """The largest supported request remains an explicit bounded Valkey read.""" + client = _ReadClient() + + assert asyncio.run(read_activity_events(client, "post-1", event_count=1000)) == [] + assert client.calls == 1 + + +def test_canonical_activity_read_uses_one_bounded_valkey_round_trip() -> None: + """The normal UUID path counts alias admission and data read as real I/O.""" + client = _PopulatedReadClient() + post_id = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + + events = asyncio.run(read_activity_events(client, post_id, event_count=3)) + + assert [event["event_id"] for event in events] == ["300-0", "200-0", "100-0"] + assert client.calls == [("activity:aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", "+", 3)] + assert client.round_trips == 1 diff --git a/tests/test_activity_stream_retry_limit.py b/tests/test_activity_stream_retry_limit.py new file mode 100644 index 000000000..ae4aaff9a --- /dev/null +++ b/tests/test_activity_stream_retry_limit.py @@ -0,0 +1,87 @@ +"""Bound the optimistic retry loop used by synchronous activity reseeding.""" + +from __future__ import annotations + +import pytest +from redis.exceptions import WatchError + +from backend.app.activity_stream import publish_activity_event_sync + + +class _ConflictStream: + """Force repeated WATCH conflicts without requiring a live Valkey process.""" + + def __init__(self, conflict_attempts: int) -> None: + """Allow exactly ``conflict_attempts`` synthetic optimistic conflicts.""" + self.conflict_attempts = conflict_attempts + self.execute_attempts = 0 + + def pipeline(self): + """Return a transaction facade for one optimistic reseed attempt.""" + return _ConflictPipeline(self) + + +class _ConflictPipeline: + """Model the transaction surface while forcing bounded WATCH conflicts.""" + + def __init__(self, stream: _ConflictStream) -> None: + """Bind the transaction to the shared conflict counter.""" + self.stream = stream + + def __enter__(self): + """Return the transaction facade.""" + return self + + def __exit__(self, exc_type, exc, traceback) -> bool: + """Never suppress transaction exceptions.""" + del exc_type, exc, traceback + return False + + def watch(self, key: str) -> None: + """Accept the production WATCH call without external state.""" + del key + + def xrevrange(self, key: str): + """Expose an empty retained stream so the append path is exercised.""" + del key + return [] + + def multi(self) -> None: + """Accept the production MULTI transition.""" + + def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None): + """Accept the queued append without mutating external state.""" + del key, fields, maxlen, approximate + return self + + def execute(self): + """Raise WATCH conflicts, then fail if production retries past the bound.""" + self.stream.execute_attempts += 1 + if self.stream.execute_attempts <= self.stream.conflict_attempts: + raise WatchError( + "synthetic conflict while watching activity:post-1 for post-1" + ) + raise AssertionError("activity reseed retried beyond the expected bound") + + +def test_publish_activity_event_sync_fails_after_bounded_watch_conflicts() -> None: + """Contention failure must not retain the raw key through exception chaining.""" + client = _ConflictStream(conflict_attempts=8) + + with pytest.raises(RuntimeError) as error_info: + publish_activity_event_sync( + client, + "post-1", + "ticket_created", + "acct-1", + "Ticket created: bounded retry", + ) + + error_message = str(error_info.value) + assert "activity reseed" in error_message.lower() + assert "8" in error_message + assert "post-1" not in error_message + assert "activity:post-1" not in error_message + assert error_info.value.__cause__ is None + assert error_info.value.__context__ is None + assert client.execute_attempts == 8 diff --git a/tests/test_activity_stream_watch_observability.py b/tests/test_activity_stream_watch_observability.py new file mode 100644 index 000000000..ed6702c45 --- /dev/null +++ b/tests/test_activity_stream_watch_observability.py @@ -0,0 +1,91 @@ +"""Observability regressions for replay-safe Valkey activity reseeding.""" + +from __future__ import annotations + +from opentelemetry.trace import StatusCode + +from backend.app.activity_stream import ( + publish_activity_event_sync, + ticket_created_summary, +) +from tests.test_activity_stream import _RaceInjectingStream +from tests.test_activity_stream_retry_limit import _ConflictStream +from tests.test_observability import attach_inmemory_tracer + + +def test_recoverable_watch_conflict_is_not_exported_as_failed_xadd(monkeypatch) -> None: + """A successfully retried optimistic conflict must not become failure telemetry. + + ``WATCH`` conflicts are expected concurrency control for synchronous seed/admin + replay. If another seed wins with the same activity fact, the loser retries, + observes that fact, and returns successfully. The transient conflict therefore + must not mark the Valkey XADD span as an operation failure or emit an exception + event that can be mistaken for a buyer-visible/infrastructure incident. + """ + exporter = attach_inmemory_tracer(monkeypatch) + expected_fields = { + "event_type": "ticket_created", + "actor_account_id": "acct-1", + "summary": ticket_created_summary("Send Northridge Grid the revised quote"), + } + client = _RaceInjectingStream(expected_fields) + + assert publish_activity_event_sync( + client, + "post-1", + expected_fields["event_type"], + expected_fields["actor_account_id"], + expected_fields["summary"], + ) is None + + xadd_spans = [ + span + for span in exporter.get_finished_spans() + if span.name == "lineageweave.valkey.activity_xadd" + ] + assert len(xadd_spans) == 1 + assert xadd_spans[0].status.status_code is StatusCode.UNSET + assert all(event.name != "exception" for event in xadd_spans[0].events) + + +def test_exhausted_watch_conflict_marks_only_terminal_xadd_as_failure(monkeypatch) -> None: + """Persistent contention must stay visible without flagging recoverable retries. + + Seven optimistic conflicts are normal retry attempts. The eighth exhausts the + bounded operator contract and is therefore the one XADD attempt that should be + exported as an error. Its exported exception data must stay limited to the safe + RuntimeError type; the raw post-scoped stream key remains absent. + """ + exporter = attach_inmemory_tracer(monkeypatch) + client = _ConflictStream(conflict_attempts=8) + + try: + publish_activity_event_sync( + client, + "post-1", + "ticket_created", + "acct-1", + "Ticket created: bounded retry", + ) + except RuntimeError as error: + assert error.__cause__ is None + assert error.__context__ is None + else: # pragma: no cover - persistent contention is required to fail closed + raise AssertionError("persistent WATCH contention must fail closed") + + xadd_spans = [ + span + for span in exporter.get_finished_spans() + if span.name == "lineageweave.valkey.activity_xadd" + ] + assert len(xadd_spans) == 8 + assert all( + span.status.status_code is StatusCode.UNSET for span in xadd_spans[:-1] + ) + terminal_span = xadd_spans[-1] + assert terminal_span.status.status_code is StatusCode.ERROR + exception_events = [event for event in terminal_span.events if event.name == "exception"] + assert len(exception_events) == 1 + assert exception_events[0].attributes["exception.type"] == "RuntimeError" + assert "post-1" not in str(exception_events[0].attributes) + assert "activity:post-1" not in str(exception_events[0].attributes)