From 1b0c7052fbe0fb8048a0229d4b339ab8386a5596 Mon Sep 17 00:00:00 2001 From: Rocky Jaiswal Date: Sat, 8 Aug 2026 11:34:50 +0200 Subject: [PATCH] Feature parity with TS version and load testing setup --- README.md | 2 + loadtest/README.md | 262 +++++++++++++++++++ loadtest/harness/ledger.py | 264 ++++++++++++++++++++ loadtest/harness/supervisor.py | 318 ++++++++++++++++++++++++ loadtest/harness/worker.py | 177 +++++++++++++ skills/xstreamq/reference/cli.md | 10 + skills/xstreamq/reference/operations.md | 15 ++ src/xstreamq/__init__.py | 8 +- src/xstreamq/cli.py | 106 +++++++- src/xstreamq/client.py | 184 +++++++++++--- src/xstreamq/consumer.py | 10 + src/xstreamq/keys.py | 67 +++++ tests/test_cli.py | 123 +++++++++ tests/test_health.py | 63 +++++ tests/test_integration.py | 113 ++++++++- tests/test_keys.py | 53 ++++ 16 files changed, 1735 insertions(+), 40 deletions(-) create mode 100644 loadtest/README.md create mode 100644 loadtest/harness/ledger.py create mode 100644 loadtest/harness/supervisor.py create mode 100644 loadtest/harness/worker.py diff --git a/README.md b/README.md index 3f6d469..e996125 100644 --- a/README.md +++ b/README.md @@ -297,6 +297,8 @@ await dq.stream_length("orders") # entries retained in the log (XLEN await dq.pending("orders", "order-service") # delivered-but-not-acked (PEL) — in flight right now await dq.lag("orders", "order-service") # published-but-not-yet-delivered (Redis ≥ 7) await dq.peek("orders", count=10) # read without consuming +await dq.registered() # every (stream, group) registered against this Redis +await dq.health() # itemised liveness; healthy() is this, as a bool await dq.dlq_messages("orders", "order-service") # triage the DLQ await dq.redrive_dlq("orders", "order-service") # put a batch back diff --git a/loadtest/README.md b/loadtest/README.md new file mode 100644 index 0000000..a0b6167 --- /dev/null +++ b/loadtest/README.md @@ -0,0 +1,262 @@ +# xstreamq load test — Python consumers, Kotlin producer + +A load and chaos harness for the Python [xstreamq](../README.md): a JVM producer +publishing straight to Redis while the real Python library consumes — and, in chaos +mode, while its consumer processes are being SIGKILLed underneath it. + +This is a port of the [TypeScript twin's harness][ts-harness]. The supervisor and +worker take the **same flags**, the ledger uses the **same Redis keys**, and the +producer is literally the same JVM program, so a Python run and a TypeScript run +are directly comparable. + +| | what it is | what it proves | +|---|---|---| +| **producer** (Kotlin) | N virtual threads doing blocking `XADD` | how fast xstreamq's stream can be fed, and where the latency actually goes | +| **supervisor** (Python) | spawns consumer processes, optionally kills them, owns the tally | no loss under hard kills, and what at-least-once actually costs | +| **worker** (Python) | one real `XStreamQ` subscriber process | the system under test | + +## The producer lives in the TypeScript repo + +There is only one producer and it is **shared**, at +`../../01-node/xstreamq-ts/loadtest/`. It is a JVM process that imports neither +library — it reimplements xstreamq's envelope and key naming from the outside +(`Wire.kt`) and writes entries directly. + +That is the point, and it is why one copy serves both twins. xstreamq documents +its stream envelope as an **open contract**: a foreign producer may write a bare +`{b: }` entry and either consumer will decode it, defaulting `rc` to 0. A +load test that drove xstreamq's own `publish()` would be measuring the library +against itself; this one measures it against an independent implementation of its +own published contract. + +`Wire.kt` mirrors `envelope.py` / `keys.py` and their TypeScript equivalents. **If +any of the three drift, this stops testing xstreamq and starts testing a fiction.** + +## Running + +Redis first: + +```sh +make redis-up # port 6399 +``` + +Then the consumer fleet, which stays up until every message is accounted for: + +```sh +uv run python loadtest/harness/supervisor.py --messages=100000 --failure-rate=0.25 +``` + +And the producer, in another shell — note the absolute path into the other repo: + +```sh +gradle -p ../../01-node/xstreamq-ts/loadtest run \ + --args="--messages=100000 --concurrency=1024 --connections=64" +``` + +`--help` on either lists every knob. The two `--messages` values should match: the +producer uses it as how many to send, the supervisor as when to stop waiting. + +### The knobs that matter + +| producer | | +|---|---| +| `--rate=` | target msgs/sec; `0` (default) means unbounded — find the ceiling | +| `--concurrency=` | virtual threads. Cheap. Raise it until `pool wait` dominates | +| `--connections=` | pooled sockets. The actually-scarce resource | + +| supervisor | | +|---|---| +| `--workers=` | consumer **processes** | +| `--kill-interval=` | SIGKILL a random worker this often. `0` (default) disables chaos | +| `--respawn-delay=` | how long a killed worker stays down | +| `--failure-rate=<0..1>` | chance each *delivery* throws — so a message that fails once can still succeed on retry | +| `--max-retries=` | attempts before the DLQ | +| `--retry-on-idle=` | the visibility timeout. **Without it there are no retries at all** | +| `--dedupe=` | turn on idempotency markers and watch the duplicate count | + +Every flag is denominated in **milliseconds**, matching the TypeScript harness. +The Python API takes seconds for two of them, and `worker.py` does that conversion +at the boundary so the two harnesses stay invocation-compatible: + +``` +--reclaim-interval= -> retry_reclaim_interval_s +--dedupe= -> dedupe_ttl_s +``` + +## Chaos mode + +`--kill-interval=` SIGKILLs a random worker on a timer and respawns it under +the same consumer name, the way a restarted pod would come back. SIGKILL +specifically — no handler, no drain, no chance to ack — because that is the failure +the pending-entries list exists for. + +Three outcomes are worth separating, because they are not the same thing: + +- **loss** — `resolved` must reach the published count. A message orphaned by a + kill and never reclaimed shows up here and nowhere else. +- **duplicates** — a worker killed *between* a successful handler and its ack gets + that message redelivered. That is at-least-once working as documented, not a bug. +- **retry budget** — a reclaim bumps `rc` whether the handler failed or the worker + died. Enough kills and a message is dead-lettered having never once failed. + +### The accounting has to live in Redis + +Chaos mode kills processes, so anything a worker held in local variables dies with +it — and dies *in proportion to how much chaos happened*, which is the worst +possible bias. [`ledger.py`](harness/ledger.py) therefore keeps everything in +Redis: bitmaps indexed by sequence number for `seen` / `dead` / `failed` / `dup`, +plus counter and latency hashes. + +Bitmaps because the interesting questions are all set operations, and Redis does +those server-side. `|seen ∪ dead|` is how many messages reached a terminal state. +`dead ∖ failed` is the one that matters most here: messages dead-lettered *without +the handler ever failing*. + +Where the TypeScript ledger leans on ioredis auto-pipelining to collapse a +delivery's writes into one round trip, redis-py has no equivalent, so each +recording method builds an explicit `pipeline(transaction=False)` and executes it +once. Same round-trip count, stated outright rather than inferred from a client +setting. That round trip is a real cost and it is the price of accounting that +survives a kill. + +The `on_dlq` callback fires in whichever worker's reclaimer did the move, and that +worker can be killed a moment later. So the final reconciliation reads the DLQ +**stream** and sets the `dead` bit from there — the stream is the authoritative +record, the callbacks are just the live signal. + +## Runs + +All on the repo's Redis container, 12-core box, 128-byte payloads, `--work-ms=0`. + +> **Comparing against the TypeScript numbers:** the runs below use **100,000** +> messages where [the TS README][ts-runs] used 200,000, because the point here was +> to validate the port, not to re-derive throughput. Rates and *ratios* compare +> directly; absolute counts do not. + +### 1. Baseline — no chaos, 25% failure rate + +100,000 messages, unbounded producer, `max_retries=3`, 4 consumers. + +``` + PRODUCER (Kotlin, 1024 virtual threads) + published 100,000 in 2.9s → 34,495 msg/s + redis rtt p50 1.65ms p90 2.45ms p99 4.11ms + pool wait p50 0.01ms p90 95ms p99 300ms + + CONSUMER succeeded 99,618 dead-lettered 382 duplicates 0 + ✓ every published message accounted for (100,000/100,000) +``` + +**382 dead letters against a theoretical 391.** With `max_retries=3` a message must +lose the coin flip four times, so the expected count is `0.25⁴ × 100,000 = 391`. +Observed 382, about 0.4σ off — the same agreement the TypeScript run showed, which +is the useful signal: both twins spend their retry budget identically. + +**Zero duplicates without chaos**, as it should be. + +Producer-side numbers are within noise of the TypeScript run (34.5k vs 36.8k +msg/s), which is expected — the producer is the same program and never touches +either library. + +### 2. Chaos — 85 workers SIGKILLed mid-flight + +100,000 messages, 6 consumer processes, a kill every 400ms, 2% failure rate, +`max_retries=3`. + +``` + workers killed 85 + delivered 102,080 succeeded 100,000 + failed attempts 2,056 duplicates 24 dead-lettered 0 + ✓ every published message accounted for (100,000/100,000) +``` + +**Zero loss across 85 hard kills.** The visibility timeout did its job every time. + +The retry stream held **3,268** reclaims against only **2,056** injected failures — +so **1,212 messages were orphaned by kills** and brought back by the reclaimer. Of +those, **24 had already been handled**: killed in the window between the handler +returning and `XACK` landing, redelivered, and handled again. + +The accounting closes exactly: `100,000 + 2,056 + 24 = 102,080` deliveries. + +That is **~2.0% of kill-orphaned messages processed twice**, against the +TypeScript twin's 1.44% on the same knobs. Both are the same phenomenon at the +same order of magnitude; the gap is not a meaningful difference between the twins +at n=24 duplicates (Poisson noise alone is ±20% here), and the two runs also +differ in message count and therefore in how long each message sat in flight. + +### 3. Retry budget — kills alone can dead-letter a healthy message + +50,000 messages, 6 consumers, a kill every 250ms, **`--failure-rate=0`**, +**`--max-retries=1`**. With no injected failures at all, anything reaching the DLQ +got there purely by being orphaned. + +``` + workers killed 116 + failed attempts 0 + dead-lettered 3 + duplicates 21 + dlq, never failed 3 ← retry budget spent on kills, not on failures + ✓ no loss: 50,000/50,000 accounted for +``` + +All three dead letters carry `rc=2` — orphaned exactly twice, `max_retries=1` +exhausted, handler never once completed for them: + +``` +1786177189387-0 rc=2 oid=1786177183257-13 {"seq":1258,...} +1786177197423-0 rc=2 oid=1786177184040-9 {"seq":25505,...} +1786177198181-0 rc=2 oid=1786177184190-20 {"seq":22139,...} +``` + +**This reproduces the TypeScript finding exactly**, down to the `rc=2`. xstreamq +cannot distinguish "the handler failed" from "the worker died" — both are just an +un-acked entry going idle, and both cost one retry. A flappy deployment with a low +`max_retries` will therefore dead-letter perfectly healthy messages. Nothing is +lost (they're in the DLQ, and redrive puts them back), but restart churn and +handler failures draw on the same budget. + +Note `succeeded (49,998) + dead-lettered (3) = 50,001`, one more than published. +That is not an accounting error: one message was handled, killed before its ack, +redelivered, orphaned again, and dead-lettered — so it is in *both* sets. +`resolved` is `|seen ∪ dead|`, which is exactly 50,000. + +## What this does not cover + +- **Consumer throughput as a headline number.** The Redis ledger adds a round trip + per delivery specifically to survive kills. Consume rate here is not xstreamq's + ceiling. +- **The drain/sustained-rate ceilings** measured for the TypeScript twin. Those + runs have not been repeated here, so there are no Python numbers to quote and + none should be assumed from the TS ones — the consumer is the half that differs. +- **Dedupe under chaos.** `--dedupe` is wired through and works, but the + duplicate-halving run has not been repeated on this side. +- **Correlated failure.** Kills are uniform-random and independent. A real bad + deploy takes every worker down at once, which is a different shape. +- **Redis itself failing.** No failover, no partition, no `appendfsync` stall. +- **Redis under memory pressure.** Nothing trims the main stream, so a long run + grows it without bound. Realistic for a burst test, wrong for a soak test. +- **Realistic handlers.** Every run above uses a no-op handler. That is the right + choice for finding the queue's ceiling and the wrong one for predicting yours. +- **Bigger payloads.** Everything is 128 bytes; `--payload` is right there. + +## Layout + +``` +loadtest/ + harness/ + supervisor.py spawns workers, runs chaos, owns the tally + worker.py one real xstreamq consumer process — the system under test + ledger.py kill-safe run accounting in Redis bitmaps + + (producer: ../../01-node/xstreamq-ts/loadtest/, shared, JVM, library-agnostic) +``` + +The producer and the supervisor talk through one Redis key +(`xstreamq:loadtest:progress`), rewritten about once a second, so the producer's +console can show both sides at once. That key's schema is shared with the +TypeScript harness, which is why the producer can report on either twin without +knowing which one is running. + +[ts-harness]: ../../01-node/xstreamq-ts/loadtest/README.md +[ts-runs]: ../../01-node/xstreamq-ts/loadtest/README.md#runs diff --git a/loadtest/harness/ledger.py b/loadtest/harness/ledger.py new file mode 100644 index 0000000..c683b7f --- /dev/null +++ b/loadtest/harness/ledger.py @@ -0,0 +1,264 @@ +"""Run accounting, kept in Redis rather than in a process's memory. + +Chaos mode SIGKILLs consumers. Anything a worker was holding in local variables +dies with it, so a ledger in memory would lose exactly the numbers the chaos run +exists to produce — and would lose them *in proportion to how much chaos +happened*, which is the worst possible bias. Everything here is therefore written +to Redis as it happens, and no worker holds unflushed state. + +The shape is chosen so recording costs **one round trip per delivery**:: + + seen bitmap, seq -> handled at least once + dead bitmap, seq -> landed in the DLQ + failed bitmap, seq -> the handler threw for it at least once + dup bitmap, seq -> was delivered again after already being handled + counters hash of raw volume counters + latency hash of 10ms buckets -> end-to-end latency histogram + +Bitmaps because the interesting questions are all set operations over sequence +numbers, and Redis does those server-side: ``|seen u dead|`` is how many messages +reached a terminal state, and ``dead \\ failed`` is the one that matters most in +chaos mode — messages dead-lettered *without the handler ever failing*, i.e. +killed off by retry-budget exhaustion alone. + +This is a port of the TypeScript twin's ``loadtest/harness/ledger.ts``; the key +names and bucket sizes are deliberately identical so a Python run and a +TypeScript run produce directly comparable numbers. + +One deliberate difference: ioredis collapses a delivery's writes into one round +trip with auto-pipelining, which redis-py has no equivalent of. Each recording +method here therefore builds an **explicit** ``pipeline(transaction=False)`` and +executes it once — same round-trip count, stated outright rather than inferred +from a client setting. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any, NamedTuple + +if TYPE_CHECKING: + import redis.asyncio as aioredis + +PREFIX = "xstreamq:loadtest:" + + +class Keys: + """Ledger key names. Shared verbatim with the TypeScript harness.""" + + seen = f"{PREFIX}seen" + dead = f"{PREFIX}dead" + failed = f"{PREFIX}failed" + dup = f"{PREFIX}dup" + counters = f"{PREFIX}counters" + latency = f"{PREFIX}latency" + progress = f"{PREFIX}progress" + # Scratch space for BITOP results; rewritten on every aggregation. + scratch_resolved = f"{PREFIX}scratch:resolved" + scratch_dead_failed = f"{PREFIX}scratch:deadfailed" + + @classmethod + def all(cls) -> list[str]: + return [ + cls.seen, + cls.dead, + cls.failed, + cls.dup, + cls.counters, + cls.latency, + cls.progress, + cls.scratch_resolved, + cls.scratch_dead_failed, + ] + + +#: 10ms buckets up to a minute. Finer than that is noise at these latencies. +BUCKET_MS = 10 +MAX_BUCKET = 6_000 + + +def bucket_of(ms: float) -> int: + return min(MAX_BUCKET, max(0, round(ms / BUCKET_MS))) + + +class WorkerLedger: + """The write side, used by each consumer process.""" + + def __init__(self, redis: aioredis.Redis) -> None: + self._redis = redis + + async def success(self, seq: int, latency_ms: float) -> None: + """A delivery whose handler succeeded. + + Called from *inside* the handler, so it lands before xstreamq acks. That + ordering is what makes duplicate detection meaningful: a worker killed + between this write and the ack will have the message redelivered, and the + ``setbit`` will then come back 1. + """ + pipe = self._redis.pipeline(transaction=False) + pipe.setbit(Keys.seen, seq, 1) + pipe.hincrby(Keys.counters, "delivered", 1) + pipe.hincrby(Keys.latency, str(bucket_of(latency_ms)), 1) + previous, *_ = await pipe.execute() + + if previous == 1: + dup = self._redis.pipeline(transaction=False) + dup.setbit(Keys.dup, seq, 1) + dup.hincrby(Keys.counters, "duplicateDeliveries", 1) + await dup.execute() + + async def failure(self, seq: int, latency_ms: float) -> None: + """A delivery whose handler threw — an injected failure, not a kill.""" + pipe = self._redis.pipeline(transaction=False) + pipe.setbit(Keys.failed, seq, 1) + pipe.hincrby(Keys.counters, "delivered", 1) + pipe.hincrby(Keys.counters, "failedAttempts", 1) + pipe.hincrby(Keys.latency, str(bucket_of(latency_ms)), 1) + await pipe.execute() + + async def dead_lettered(self, seq: int) -> None: + pipe = self._redis.pipeline(transaction=False) + pipe.setbit(Keys.dead, seq, 1) + pipe.hincrby(Keys.counters, "dlqEvents", 1) + await pipe.execute() + + +class Tally(NamedTuple): + """One aggregated view of the whole run.""" + + #: Every handler invocation, including redeliveries. + delivered: int + #: Distinct messages handled at least once. + succeeded: int + failed_attempts: int + #: DLQ callback firings. + dlq_events: int + #: Distinct messages in the DLQ. + dead_lettered: int + #: Distinct messages delivered again *after* already being handled. + duplicated: int + #: Distinct messages dead-lettered whose handler never once threw. + #: + #: In chaos mode this is the headline: every one of these exhausted its retry + #: budget purely by being in flight on workers that kept getting killed. + dead_lettered_never_failed: int + #: Distinct messages that reached a terminal state — handled or dead-lettered. + resolved: int + e2e_p50_ms: int + e2e_p99_ms: int + + +class LedgerReader: + """The read side, used by the supervisor.""" + + def __init__(self, redis: aioredis.Redis) -> None: + self._redis = redis + + async def reset(self) -> None: + """Clear a previous run. Stale bitmaps would silently corrupt the next one.""" + await self._redis.delete(*Keys.all()) + + async def read(self) -> Tally: + # `resolved` is |seen u dead| and `dead_failed` is |dead n failed|, both + # computed server-side — the alternative is shipping megabytes of bitmap + # to the client once a second. + pipe = self._redis.pipeline(transaction=False) + pipe.bitop("OR", Keys.scratch_resolved, Keys.seen, Keys.dead) + pipe.bitop("AND", Keys.scratch_dead_failed, Keys.dead, Keys.failed) + pipe.bitcount(Keys.scratch_resolved) + pipe.bitcount(Keys.seen) + pipe.bitcount(Keys.dead) + pipe.bitcount(Keys.dup) + pipe.bitcount(Keys.scratch_dead_failed) + pipe.hgetall(Keys.counters) + pipe.hgetall(Keys.latency) + ( + _, + _, + resolved, + succeeded, + dead_lettered, + duplicated, + dead_and_failed, + counters, + latency, + ) = await pipe.execute() + + def count(field: str) -> int: + return int(counters.get(field, 0)) + + return Tally( + delivered=count("delivered"), + succeeded=succeeded, + failed_attempts=count("failedAttempts"), + dlq_events=count("dlqEvents"), + dead_lettered=dead_lettered, + duplicated=duplicated, + dead_lettered_never_failed=dead_lettered - dead_and_failed, + resolved=resolved, + e2e_p50_ms=percentile(latency, 50), + e2e_p99_ms=percentile(latency, 99), + ) + + async def reconcile_dlq(self, dlq_key: str) -> int: + """Set the ``dead`` bit for every message actually sitting in the DLQ stream. + + The ``on_dlq`` callback is the live signal, but it fires in whichever + worker's reclaimer did the move — and that worker can be killed a moment + later. The DLQ stream itself is the authoritative record, so the final + reconciliation reads it directly rather than trusting the callbacks to + have all landed. + """ + cursor = "-" + recovered = 0 + page_size = 500 + + while True: + page = await self._redis.xrange(dlq_key, cursor, "+", count=page_size) + if not page: + break + + pipe = self._redis.pipeline(transaction=False) + queued = 0 + for _entry_id, fields in page: + seq = seq_of(fields) + if seq is not None: + pipe.setbit(Keys.dead, seq, 1) + queued += 1 + if queued: + recovered += sum(1 for previous in await pipe.execute() if previous == 0) + + if len(page) < page_size: + break + cursor = f"({page[-1][0]}" + + return recovered + + +def seq_of(fields: dict[str, Any]) -> int | None: + """Pull the load-test sequence number out of a raw stream entry's fields.""" + raw = fields.get("b") + if raw is None: + return None + try: + seq = json.loads(raw).get("seq") + except (ValueError, AttributeError): + # A body that will not parse is a poison message, which has no seq to + # recover — the DLQ is exactly where it belongs. + return None + return seq if isinstance(seq, int) else None + + +def percentile(buckets: dict[str, str], p: int) -> int: + entries = sorted((int(bucket), int(count)) for bucket, count in buckets.items()) + total = sum(count for _, count in entries) + if total == 0: + return 0 + + target = -(-p * total // 100) # ceil(p/100 * total) + cumulative = 0 + for bucket, count in entries: + cumulative += count + if cumulative >= target: + return bucket * BUCKET_MS + return MAX_BUCKET * BUCKET_MS diff --git a/loadtest/harness/supervisor.py b/loadtest/harness/supervisor.py new file mode 100644 index 0000000..a7093da --- /dev/null +++ b/loadtest/harness/supervisor.py @@ -0,0 +1,318 @@ +"""Runs the consumer fleet, optionally killing it at random, and owns the tally. + +Chaos mode is the sharpest test xstreamq has: SIGKILL a worker holding +delivered-but-unacked messages and see whether the visibility timeout actually +brings them back. SIGKILL specifically — no handler, no drain, no chance to ack — +because that is the failure the pending-entries list exists for. + +Three outcomes are worth watching, and they are not the same thing: + + loss `resolved` must reach the published count. A message orphaned + by a kill and never reclaimed shows up here and nowhere else. + + duplicates a worker killed *between* a successful handler and its ack + gets that message redelivered. That is at-least-once working + as documented, not a bug — but it is the number people + actually want to know, and it should be ~0 without chaos. + + retry budget a reclaim bumps `rc` whether the handler failed or the worker + died. Enough kills and a message is dead-lettered having never + once failed. `dead_lettered_never_failed` counts exactly those. + + uv run python loadtest/harness/supervisor.py --messages=200000 --kill-interval=2000 + +Ported from the TypeScript twin's ``loadtest/harness/supervisor.ts``, flag for +flag, so a Python run and a TypeScript run are invoked identically. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +import random +import signal +import sys +import time +from dataclasses import dataclass, field + +import redis.asyncio as aioredis + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from loadtest.harness.ledger import Keys, LedgerReader, Tally + +from xstreamq import keys + +USAGE = """ +xstreamq load harness — consumer fleet with optional chaos + + uv run python loadtest/harness/supervisor.py [options] + + supervisor + --messages= how many the producer will send (default 100000) + --workers= consumer processes (default 4) + --kill-interval= SIGKILL a random worker this often; 0 = no chaos (default 0) + --respawn-delay= how long a killed worker stays down (default 1000) + --timeout= give up waiting for a drain (default 600) + --no-reset keep the previous run's ledger + --stay do not exit once every message is accounted for + + forwarded to each worker + --redis= default redis://localhost:6399 + --stream= default loadtest + --group= default loadtest-service + --failure-rate=<0..1> chance each delivery throws (default 0.05) + --work-ms= simulated handler work (default 0) + --retry-on-idle= visibility timeout (default 2000) + --max-retries= attempts before the DLQ (default 3) + --reclaim-interval= PEL scan period (default 500) + --count= entries per read (default 64) + --dedupe= enable idempotency markers (default off) +""" + +#: Flags the supervisor consumes itself; everything else is forwarded to workers. +SUPERVISOR_FLAGS = ["messages", "workers", "kill-interval", "respawn-delay", "timeout"] +SUPERVISOR_SWITCHES = ["--no-reset", "--stay", "--help", "-h"] + +WORKER_SCRIPT = os.path.join(os.path.dirname(__file__), "worker.py") + + +@dataclass +class Slot: + name: str + process: asyncio.subprocess.Process | None = None + restarts: int = 0 + #: The respawn timer, held so shutdown can cancel a pending relaunch. + relaunch: asyncio.Task | None = field(default=None, repr=False) + #: The exit watcher. Held for the same reason as `relaunch`: a task nobody + #: references can be garbage-collected mid-run, and losing this one would + #: silently stop respawning that slot for the rest of the run. + watcher: asyncio.Task | None = field(default=None, repr=False) + + +class Supervisor: + def __init__(self, argv: list[str]) -> None: + def flag(name: str) -> str | None: + prefix = f"--{name}=" + for arg in argv: + if arg.startswith(prefix): + return arg[len(prefix) :] + return None + + def num(name: str, fallback: float) -> float: + raw = flag(name) + return fallback if raw is None else float(raw) + + self.messages = int(num("messages", 100_000)) + self.workers = int(num("workers", 4)) + self.kill_interval_ms = num("kill-interval", 0) + self.respawn_delay_ms = num("respawn-delay", 1_000) + self.timeout_seconds = num("timeout", 600) + self.reset = "--no-reset" not in argv + self.stay = "--stay" in argv + self.redis_url = flag("redis") or "redis://localhost:6399" + self.stream = flag("stream") or "loadtest" + self.group = flag("group") or "loadtest-service" + + # Everything the supervisor did not claim belongs to the workers. + self.forwarded = [ + arg + for arg in argv + if arg not in SUPERVISOR_SWITCHES + and not any(arg.startswith(f"--{name}=") for name in SUPERVISOR_FLAGS) + ] + + self.shutting_down = False + self.kills = 0 + self.slots = [Slot(name=f"worker-{i}") for i in range(self.workers)] + self.exit_code = 1 + + async def launch(self, slot: Slot) -> None: + if self.shutting_down: + return + slot.process = await asyncio.create_subprocess_exec( + sys.executable, + WORKER_SCRIPT, + *self.forwarded, + f"--consumer={slot.name}", + stdin=asyncio.subprocess.DEVNULL, + stdout=None, + stderr=asyncio.subprocess.DEVNULL, + ) + slot.watcher = asyncio.create_task(self._watch(slot)) + + async def _watch(self, slot: Slot) -> None: + process = slot.process + if process is None: + return + await process.wait() + slot.process = None + if self.shutting_down: + return + # A killed worker comes back with the *same* consumer name, the way a + # restarted pod would. Its orphaned PEL entries are the reclaimer's + # problem either way — XPENDING scans the group, not the consumer. + slot.restarts += 1 + slot.relaunch = asyncio.create_task(self._relaunch_after_delay(slot)) + + async def _relaunch_after_delay(self, slot: Slot) -> None: + await asyncio.sleep(self.respawn_delay_ms / 1000) + await self.launch(slot) + + async def chaos_loop(self) -> None: + while not self.shutting_down: + await asyncio.sleep(self.kill_interval_ms / 1000) + alive = [slot for slot in self.slots if slot.process is not None] + if not alive: + continue + victim = random.choice(alive) + # Exiting between the liveness check and the signal is normal, not an + # error — the victim may have already been killed a moment ago. + with contextlib.suppress(ProcessLookupError): + # SIGKILL, not SIGTERM: no drain, no ack, nothing cleaned up. + victim.process.kill() # type: ignore[union-attr] + self.kills += 1 + + async def run(self) -> int: + redis = aioredis.from_url(self.redis_url, decode_responses=True) + reader = LedgerReader(redis) + + if self.reset: + await reader.reset() + + chaos = self.kill_interval_ms > 0 + detail = ( + f"SIGKILL every {self.kill_interval_ms:g}ms, respawn after {self.respawn_delay_ms:g}ms" + if chaos + else "no chaos" + ) + print(f"harness: {self.workers} worker processes, {detail}") + print(f"watching {self.stream}/{self.group} for {self.messages:,} messages\n") + + for slot in self.slots: + await self.launch(slot) + + chaos_task = asyncio.create_task(self.chaos_loop()) if chaos else None + + started_at = time.monotonic() + deadline = started_at + self.timeout_seconds + reason = "complete" + + while True: + await asyncio.sleep(1) + tally = await reader.read() + await self._publish(redis, tally) + + seconds = round(time.monotonic() - started_at) + up = sum(1 for slot in self.slots if slot.process is not None) + print( + f"[{seconds:4d}s] workers {up}/{self.workers} kills {self.kills} │ " + f"resolved {tally.resolved:,}/{self.messages:,} " + f"ok {tally.succeeded:,} dlq {tally.dead_lettered:,}" + f"({tally.dead_lettered_never_failed} never-failed) dup {tally.duplicated:,}", + flush=True, + ) + + if self.stay: + continue + if tally.resolved >= self.messages: + break + if time.monotonic() > deadline: + reason = "timed out" + break + + if chaos_task is not None: + chaos_task.cancel() + await self._shutdown(redis, reader, reason) + await redis.aclose() + return self.exit_code + + async def _publish(self, redis: aioredis.Redis, tally: Tally) -> None: + await redis.set( + Keys.progress, + json.dumps( + { + "delivered": tally.delivered, + "succeeded": tally.succeeded, + "failedAttempts": tally.failed_attempts, + "dlq": tally.dead_lettered, + "duplicates": tally.duplicated, + "resolved": tally.resolved, + "e2eP50Ms": tally.e2e_p50_ms, + "e2eP99Ms": tally.e2e_p99_ms, + "kills": self.kills, + "dlqNeverFailed": tally.dead_lettered_never_failed, + } + ), + ) + + async def _shutdown(self, redis: aioredis.Redis, reader: LedgerReader, reason: str) -> None: + self.shutting_down = True + + print(f"\nstopping workers ({reason})…") + for slot in self.slots: + if slot.relaunch is not None: + slot.relaunch.cancel() + if slot.process is not None: + with contextlib.suppress(ProcessLookupError): + slot.process.send_signal(signal.SIGTERM) + await asyncio.sleep(2) + for slot in self.slots: + if slot.process is not None: + with contextlib.suppress(ProcessLookupError): + slot.process.kill() + + # The DLQ stream is the authoritative record; on_dlq callbacks can be lost + # with the worker that fired them. + recovered = await reader.reconcile_dlq(keys.dlq_key(self.stream, self.group)) + tally = await reader.read() + await self._publish(redis, tally) + + self._report(tally, recovered) + self.exit_code = 0 if tally.resolved >= self.messages else 1 + + def _report(self, tally: Tally, recovered_from_dlq_stream: int) -> None: + restarts = sum(slot.restarts for slot in self.slots) + shortfall = self.messages - tally.resolved + + rows = [ + "", + f" delivered {tally.delivered:,} (includes redeliveries)", + f" succeeded {tally.succeeded:,}", + f" failed attempts {tally.failed_attempts:,}", + f" dead-lettered {tally.dead_lettered:,}", + f" duplicates {tally.duplicated:,}", + f" end-to-end p50 {tally.e2e_p50_ms}ms p99 {tally.e2e_p99_ms}ms", + "", + f" workers killed {self.kills} ({restarts} restarts)", + f" dlq, never failed {tally.dead_lettered_never_failed}" + " ← retry budget spent on kills, not on failures", + ( + f" dlq reconciled {recovered_from_dlq_stream} dead letters" + " the callbacks missed" + if recovered_from_dlq_stream > 0 + else "" + ), + "", + ( + f" ✓ no loss: {tally.resolved:,}/{self.messages:,} accounted for" + if shortfall == 0 + else f" ✗ LOSS: {shortfall:,} of {self.messages:,} never reached a terminal state" + ), + "", + ] + print("\n".join(row for row in rows if row != "")) + + +def main() -> None: + argv = sys.argv[1:] + if "--help" in argv or "-h" in argv: + print(USAGE) + return + raise SystemExit(asyncio.run(Supervisor(argv).run())) + + +if __name__ == "__main__": + main() diff --git a/loadtest/harness/worker.py b/loadtest/harness/worker.py new file mode 100644 index 0000000..d673892 --- /dev/null +++ b/loadtest/harness/worker.py @@ -0,0 +1,177 @@ +"""One consumer process — the system under test. + +Deliberately a *process*, not a task, because chaos mode kills it with SIGKILL +and there has to be nothing left running. It holds no accounting state of its +own: every outcome goes straight to the Redis ledger, so being killed loses +nothing but the work in flight (which is the point — xstreamq's visibility +timeout is supposed to notice and redeliver it). + +Normally launched by ``supervisor.py``, which owns the run. Running it directly +is fine for poking at a single consumer:: + + uv run python loadtest/harness/worker.py --consumer=solo --failure-rate=0.1 + +Ported from the TypeScript twin's ``loadtest/harness/worker.ts``. Two option +names differ because the Python API's units differ, and the translation happens +here so both harnesses take the same millisecond-denominated flags: + + --reclaim-interval= -> retry_reclaim_interval_s (seconds) + --dedupe= -> dedupe_ttl_s (seconds) +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import random +import signal +import sys +import time +from dataclasses import dataclass +from typing import Any + +import redis.asyncio as aioredis + +# Run from a source checkout without installing the harness as a package. +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from loadtest.harness.ledger import WorkerLedger + +from xstreamq import Message, XStreamQ + + +@dataclass(frozen=True) +class WorkerConfig: + redis: str = "redis://localhost:6399" + stream: str = "loadtest" + group: str = "loadtest-service" + consumer: str = "worker-0" + failure_rate: float = 0.05 + work_ms: float = 0 + retry_on_idle_ms: int = 2_000 + max_retries: int = 3 + reclaim_interval_ms: float = 500 + backoff_multiplier: float = 1.5 + count: int = 64 + dedupe_ttl_ms: float | None = None + + +def worker_config_from_argv(argv: list[str]) -> WorkerConfig: + def flag(name: str) -> str | None: + prefix = f"--{name}=" + for arg in argv: + if arg.startswith(prefix): + return arg[len(prefix) :] + return None + + def num(name: str, fallback: float) -> float: + raw = flag(name) + if raw is None: + return fallback + try: + return float(raw) + except ValueError: + raise SystemExit(f"--{name} must be a number, got {raw}") from None + + dedupe = flag("dedupe") + + return WorkerConfig( + redis=flag("redis") or "redis://localhost:6399", + stream=flag("stream") or "loadtest", + group=flag("group") or "loadtest-service", + consumer=flag("consumer") or "worker-0", + failure_rate=num("failure-rate", 0.05), + work_ms=num("work-ms", 0), + retry_on_idle_ms=int(num("retry-on-idle", 2_000)), + max_retries=int(num("max-retries", 3)), + reclaim_interval_ms=num("reclaim-interval", 500), + backoff_multiplier=num("backoff", 1.5), + count=int(num("count", 64)), + dedupe_ttl_ms=None if dedupe is None else float(dedupe), + ) + + +async def run_worker(config: WorkerConfig) -> None: + # A second connection, separate from xstreamq's own, so ledger writes never + # queue behind a blocking XREADGROUP. + ledger_redis = aioredis.from_url(config.redis, decode_responses=True) + ledger = WorkerLedger(ledger_redis) + + # Quiet, because a run injects tens of thousands of *expected* handler + # failures; logging each would cost more than the work being measured. The + # TypeScript twin passes `silentLogger`; the Python twin uses stdlib logging, + # so the equivalent is raising the level on its logger. + logging.getLogger("xstreamq").setLevel(logging.CRITICAL) + + client = XStreamQ(config.redis) + + async def handle(message: Message) -> None: + body = message.body + if not isinstance(body, dict): + return + seq = body.get("seq") + if not isinstance(seq, int): + return + + published_at = body.get("t") + latency_ms = ( + (time.time() * 1000 - published_at) if isinstance(published_at, int | float) else 0 + ) + + if config.work_ms > 0: + await asyncio.sleep(config.work_ms / 1000) + + if random.random() < config.failure_rate: + await ledger.failure(seq, latency_ms) + # Not acking is the contract: the entry stays in the PEL and the + # reclaimer takes it from there after the visibility timeout. + raise RuntimeError("injected failure") + + # Recorded before returning, and therefore before xstreamq acks. A kill + # landing in the gap between the two is precisely what produces a + # duplicate, and this ordering is what lets the ledger see it as one. + await ledger.success(seq, latency_ms) + + async def on_dlq(body: Any, _original_id: str, _retry_count: int) -> None: + # Async rather than sync-plus-create_task: xstreamq awaits an awaitable + # callback, whereas a bare create_task() returns a task nobody holds a + # reference to, which the loop is free to garbage-collect mid-write. The + # end-of-run reconcile_dlq() would paper over the loss, so it would show + # up only as a live counter that quietly undercounts. + if isinstance(body, dict) and isinstance(body.get("seq"), int): + await ledger.dead_lettered(body["seq"]) + + await client.subscribe( + config.stream, + handle, + group=config.group, + consumer=config.consumer, + count=config.count, + retry_on_idle_ms=config.retry_on_idle_ms, + max_retries=config.max_retries, + retry_reclaim_interval_s=config.reclaim_interval_ms / 1000, + backoff_multiplier=config.backoff_multiplier, + start_from="0", + on_dlq=on_dlq, + **({} if config.dedupe_ttl_ms is None else {"dedupe_ttl_s": config.dedupe_ttl_ms / 1000}), + ) + + await client.start() + sys.stderr.write(f"worker {config.consumer} up (pid {os.getpid()})\n") + sys.stderr.flush() + + # SIGTERM is the orderly stop the supervisor sends at the end of a run. + # SIGKILL, obviously, does not get a handler — that is the whole experiment. + stopping = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, stopping.set) + + await stopping.wait() + await client.stop(drain_timeout=5.0) + await ledger_redis.aclose() + + +if __name__ == "__main__": + asyncio.run(run_worker(worker_config_from_argv(sys.argv[1:]))) diff --git a/skills/xstreamq/reference/cli.md b/skills/xstreamq/reference/cli.md index 0280b01..f18a72c 100644 --- a/skills/xstreamq/reference/cli.md +++ b/skills/xstreamq/reference/cli.md @@ -24,13 +24,23 @@ Add `--json` / `-j` to any subcommand for machine-readable output to pipe into ` ## Commands ```bash +# ls — every (stream, group) xstreamq has registered against this Redis +xstreamq ls +xstreamq ls --registry custom:registry # only if the app set a non-default key + # depth — XLEN; add --group/-g to also show that group's pending (PEL) and lag xstreamq depth orders xstreamq depth orders --group order-service +# pending — what is delivered but not yet acked, and to whom +xstreamq pending orders order-service # exits 1 if the group does not exist + # peek — read the head of a stream without consuming (no ack) xstreamq peek orders --count 5 # -n is the short form of --count +# health — probe the connection; exits 1 when unhealthy (for container healthchecks) +xstreamq health + # compact — safe, no-loss trim (see operations.md) xstreamq compact orders diff --git a/skills/xstreamq/reference/operations.md b/skills/xstreamq/reference/operations.md index 79e0d9b..42e38fd 100644 --- a/skills/xstreamq/reference/operations.md +++ b/skills/xstreamq/reference/operations.md @@ -23,6 +23,21 @@ await dq.lag("orders", "order-service") # published-but-not-yet-delivered - **`pending`** is what is currently being worked on (or stuck) — the in-flight tray. - **`lag`** is "how far behind is this consumer?" — returns `None` if Redis can't compute it after a trim, and always `None` on **Redis < 7.0**. +`pending()` returns a **`PendingSummary`**, not a bare count: + +```python +summary = await dq.pending("orders", "order-service") +summary.count # total in flight +summary.min_id, summary.max_id # oldest / newest pending ids (None when empty) +[(c.name, c.count) for c in summary.consumers] # per-consumer breakdown +``` + +The breakdown is the point: a **stuck worker** shows up as one consumer holding a +growing share of the PEL while the others drain. `pending()` **raises +`XStreamQError` if the group does not exist** — "never created" and "nothing in +flight" are different answers, and returning `0` for both is how a missing +subscriber reads as a healthy one. + ## `healthy()` liveness probe A bounded, end-to-end liveness check for a k8s/ECS probe or your own `/health` endpoint: diff --git a/src/xstreamq/__init__.py b/src/xstreamq/__init__.py index 967c948..635f0ee 100644 --- a/src/xstreamq/__init__.py +++ b/src/xstreamq/__init__.py @@ -17,15 +17,21 @@ async def handle(msg): ... """ -from .client import XStreamQ +from .client import ConsumerHealth, HealthReport, PendingConsumer, PendingSummary, XStreamQ from .envelope import Message from .exceptions import DecodeError, UnredrivableError, XStreamQError +from .keys import RegistryEntry from .reclaimer import ReclaimerConfig __all__ = [ + "ConsumerHealth", "DecodeError", + "HealthReport", "Message", + "PendingConsumer", + "PendingSummary", "ReclaimerConfig", + "RegistryEntry", "UnredrivableError", "XStreamQ", "XStreamQError", diff --git a/src/xstreamq/cli.py b/src/xstreamq/cli.py index 34fe71b..6f37908 100644 --- a/src/xstreamq/cli.py +++ b/src/xstreamq/cli.py @@ -36,21 +36,31 @@ ) -def _exec(url: str, op: Callable[[XStreamQ], Awaitable[T]]) -> T: +def _exec( + url: str, op: Callable[[XStreamQ], Awaitable[T]], *, registry_key: str | None = None +) -> T: """Connect, run one async op against a live XStreamQ, disconnect. Connection/Redis errors are reported cleanly (no traceback) and exit non-zero, so the CLI behaves like a CLI rather than dumping a Python stack. + + ``registry_key`` only matters to ``ls``, and only when the application + configured a non-default one — every other command addresses streams directly. """ from redis.exceptions import RedisError + from .exceptions import XStreamQError + async def _go() -> T: - async with XStreamQ(url) as dq: + dq = XStreamQ(url) if registry_key is None else XStreamQ(url, registry_key=registry_key) + async with dq: return await op(dq) try: return asyncio.run(_go()) - except (RedisError, OSError) as exc: + except (RedisError, OSError, XStreamQError) as exc: + # XStreamQError covers the deliberate ops refusals — "no consumer group + # 'g' on {orders}" is a message for the user, not a stack trace. print(f"error: {exc}", file=sys.stderr) raise SystemExit(1) from exc @@ -96,6 +106,23 @@ def build_app() -> "typer.Typer": def _root(ctx: typer.Context, url: str = url_opt) -> None: ctx.obj = url + @app.command("ls") + def ls_( + ctx: typer.Context, + registry: str | None = typer.Option( + None, "--registry", help="Registry set key, if the app configured a non-default one." + ), + json_out: bool = json_opt, + ) -> None: + """Every (stream, group) xstreamq has registered against this Redis.""" + url: str = ctx.obj + entries = _exec(url, lambda dq: dq.registered(), registry_key=registry) + _emit( + json_out, + "\n".join(f"{e.stream} {e.group}" for e in entries) or "nothing registered", + [{"stream": e.stream, "group": e.group} for e in entries], + ) + @app.command() def depth( ctx: typer.Context, @@ -112,7 +139,8 @@ async def op(dq: XStreamQ) -> dict[str, Any]: out: dict[str, Any] = {"stream": stream, "length": await dq.stream_length(stream)} if group is not None: out["group"] = group - out["pending"] = await dq.pending(stream, group) + # The count only; the per-consumer breakdown belongs to `pending`. + out["pending"] = (await dq.pending(stream, group)).count out["lag"] = await dq.lag(stream, group) return out @@ -122,6 +150,41 @@ async def op(dq: XStreamQ) -> dict[str, Any]: human += f"\n group {group}: pending={res['pending']} lag={res['lag']}" _emit(json_out, human, res) + @app.command() + def pending( + ctx: typer.Context, + stream: str = typer.Argument(..., help="Stream name."), + group: str = typer.Argument(..., help="Consumer group."), + json_out: bool = json_opt, + ) -> None: + """What is delivered but not yet acked, and to whom.""" + url: str = ctx.obj + summary = _exec(url, lambda dq: dq.pending(stream, group)) + + if summary.count == 0: + human = "nothing pending" + else: + rows = [ + f" pending {summary.count}", + f" oldest {summary.min_id or '—'}", + f" newest {summary.max_id or '—'}", + ] + if summary.consumers: + rows += ["", "by consumer"] + rows += [f" {c.name} {c.count}" for c in summary.consumers] + human = "\n".join(rows) + + _emit( + json_out, + human, + { + "count": summary.count, + "min_id": summary.min_id, + "max_id": summary.max_id, + "consumers": [{"name": c.name, "count": c.count} for c in summary.consumers], + }, + ) + @app.command() def peek( ctx: typer.Context, @@ -149,6 +212,41 @@ def compact( removed = _exec(url, lambda dq: dq.compact(stream)) _emit(json_out, f"compacted {stream}: removed {removed} entries", {"removed": removed}) + @app.command() + def health(ctx: typer.Context, json_out: bool = json_opt) -> None: + """Probe the connection; exits 1 when unhealthy. + + This CLI has no subscriptions of its own, so `consumers` is always 0 and + `reclaimer` always "none" — from here it is a Redis reachability check. + The same call inside a running application reports that app's loops too. + """ + url: str = ctx.obj + report = _exec(url, lambda dq: dq.health()) + _emit( + json_out, + "\n".join( + [ + "healthy" if report.healthy else "UNHEALTHY", + f" redis {report.redis}", + f" reclaimer {report.reclaimer}", + f" consumers {len(report.consumers)}", + ] + ), + { + "healthy": report.healthy, + "running": report.running, + "redis": report.redis, + "reclaimer": report.reclaimer, + "consumers": [ + {"stream": c.stream, "group": c.group, "alive": c.alive} + for c in report.consumers + ], + }, + ) + # An exit code a container healthcheck or a deploy script can branch on. + if not report.healthy: + raise SystemExit(1) + @dlq_app.command("ls") def dlq_ls( ctx: typer.Context, diff --git a/src/xstreamq/client.py b/src/xstreamq/client.py index e15fa19..07b9683 100644 --- a/src/xstreamq/client.py +++ b/src/xstreamq/client.py @@ -17,13 +17,12 @@ async def handle(msg): import asyncio import itertools -import json import logging import os import socket from collections.abc import Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Literal, cast import redis.asyncio as aioredis from redis.exceptions import RedisError, ResponseError @@ -32,7 +31,7 @@ async def handle(msg): from .connection import make_client from .consumer import Consumer, Handler from .envelope import Message, decode_lenient, encode -from .exceptions import UnredrivableError +from .exceptions import UnredrivableError, XStreamQError from .reclaimer import OnDLQ, Reclaimer, ReclaimerConfig from .scripts import register_redrive from .serde import Schema, resolve_schema, to_jsonable @@ -62,6 +61,53 @@ async def handle(msg): DEFAULT_REGISTRY_KEY = "xstreamq:registry" +@dataclass(frozen=True) +class PendingConsumer: + """One consumer's share of a group's pending-entries list.""" + + name: str + count: int + + +@dataclass(frozen=True) +class PendingSummary: + """The ``XPENDING`` summary: how much is in flight, and with whom.""" + + #: Entries delivered but not yet acked. + count: int + #: Oldest and newest pending ids; ``None`` when nothing is pending. + min_id: str | None + max_id: str | None + #: Per-consumer breakdown — where a stuck worker shows up, as one consumer + #: holding a growing share of the PEL while the others drain. + consumers: list[PendingConsumer] + + +@dataclass(frozen=True) +class ConsumerHealth: + """One consume loop's liveness, main and retry alike.""" + + stream: str + group: str + alive: bool + + +@dataclass(frozen=True) +class HealthReport: + """An itemised :meth:`XStreamQ.health` result.""" + + #: Every check passed. This is what :meth:`XStreamQ.healthy` returns. + healthy: bool + #: Whether ``start()`` has been called and ``stop()`` has not. + running: bool + #: Whether the main connection answered a ``PING``. + redis: Literal["ok", "unreachable"] + #: One entry per consume loop in *this process*. + consumers: list[ConsumerHealth] + #: ``"none"`` when no subscription enabled retries, so there is nothing to run. + reclaimer: Literal["ok", "degraded", "none"] + + @dataclass class _Subscription: """A resolved registration: the keys/consumers/reclaimers for one (stream, group).""" @@ -400,17 +446,13 @@ def _reclaimer_config(self, sub: _Subscription, *, source: str, group: str) -> R ) async def _register_in_registry(self, sub: _Subscription) -> None: - """Record this subscription's keys so a v2 dashboard can enumerate them.""" - record = json.dumps( - { - "stream": sub.stream_key, - "group": sub.group, - "retry": sub.retry_stream, - "dlq": sub.dlq_stream, - }, - separators=(",", ":"), - sort_keys=True, - ) + """Record this (stream, group) so a v2 dashboard can enumerate it. + + The member encoding is shared byte-for-byte with the TypeScript twin and + lives in :func:`keys.registry_member` — read the invariants documented + there before changing it. + """ + record = keys.registry_member(sub.stream, sub.group) try: await self._require_client().sadd(self._registry_key, record) except RedisError: @@ -482,17 +524,73 @@ async def healthy(self, *, timeout: float = 2.0) -> bool: Bounded by design: the ``PING`` is wrapped in ``asyncio.wait_for(timeout)`` so a wedged connection can't make the probe itself hang. """ - if not self._running or self._client is None: - return False - try: - await asyncio.wait_for(self._client.ping(), timeout=timeout) - except Exception: - # Any failure (timeout, connection error, …) means unhealthy. We don't - # catch CancelledError (BaseException) — caller cancellation propagates. - return False - if any(not consumer.alive for consumer in self._consumers): - return False - return self._reclaimer.all_alive + return (await self.health(timeout=timeout)).healthy + + async def health(self, *, timeout: float = 2.0) -> HealthReport: + """The same checks as :meth:`healthy`, but itemised. + + Use this when "unhealthy" is not a useful enough answer — a readiness + probe wants the boolean, a human or a dashboard wants to know *which* + check failed. Like :meth:`healthy` it never raises. + + Note what this can and cannot see: ``consumers`` and ``reclaimer`` + describe **this process's** loops, not the deployment's. A client with no + subscriptions — which is what the ``xstreamq`` CLI constructs — always + reports no consumers and a reclaimer of ``"none"``, so from the CLI this + is effectively a Redis reachability probe. + """ + consumers = [ + ConsumerHealth(stream=consumer.stream, group=consumer.group, alive=consumer.alive) + for consumer in self._consumers + ] + + # "none" and "ok" are different answers: nothing registered is vacuously + # fine, whereas tasks that were started and have since exited are not. + if self._reclaimer.empty: + reclaimer: Literal["ok", "degraded", "none"] = "none" + else: + reclaimer = "ok" if self._reclaimer.all_alive else "degraded" + + redis: Literal["ok", "unreachable"] = "unreachable" + if self._client is not None: + try: + await asyncio.wait_for(self._client.ping(), timeout=timeout) + redis = "ok" + except Exception: + # Any failure (timeout, connection error, …) means unreachable. We + # don't catch CancelledError (BaseException) — caller cancellation + # propagates. + redis = "unreachable" + + return HealthReport( + healthy=( + self._running + and redis == "ok" + and all(c.alive for c in consumers) + and reclaimer != "degraded" + ), + running=self._running, + redis=redis, + consumers=consumers, + reclaimer=reclaimer, + ) + + async def registered(self) -> list[keys.RegistryEntry]: + """Every (stream, group) any xstreamq has registered against this Redis. + + Reads the registry set rather than scanning the keyspace, which is the + whole reason the set is written. Unparseable members are skipped — the set + is shared and long-lived, so one row from some other version must not make + the listing unreadable. + + Sorted, because a Redis set has no order and an ops listing that shuffles + between invocations is hard to read and impossible to diff. + """ + # cast: redis-py types SMEMBERS as bytes|str for clients in general, but + # xstreamq pins decode_responses=True on every connection it makes. + members = cast("set[str]", await self._require_client().smembers(self._registry_key)) + entries = (keys.parse_registry_member(member) for member in members) + return sorted(entry for entry in entries if entry is not None) async def stream_length(self, stream: str) -> int: """Total entries retained in a stream (``XLEN``). @@ -518,11 +616,37 @@ async def peek(self, stream: str, *, count: int = 10) -> list[Message]: entries = cast("list[tuple[str, dict[str, str]]]", raw) return [decode_lenient(eid, fields, stream=skey) for eid, fields in entries] - async def pending(self, stream: str, group: str) -> int: - """Messages delivered to ``group`` but not yet acked — the PEL size, i.e. - what's in-flight / being worked on right now (``XPENDING``).""" - summary = await self._require_client().xpending(keys.stream_key(stream), group) - return summary["pending"] + async def pending(self, stream: str, group: str) -> PendingSummary: + """What ``group`` has been delivered but not yet acked — in-flight work. + + The per-consumer breakdown is where a stuck worker shows up: one consumer + holding a growing share of the PEL while the others drain. For just the + total, read ``.count``. + + Raises :class:`XStreamQError` if the group does not exist, because "never + created" and "nothing in flight" are different answers and only one of + them is good news — returning 0 for both is how a missing subscriber gets + read as a healthy one. + """ + key = keys.stream_key(stream) + try: + reply = await self._require_client().xpending(key, group) + except ResponseError as exc: + if "NOGROUP" in str(exc): + raise XStreamQError(f"no consumer group {group!r} on {key}") from exc + raise + + return PendingSummary( + count=reply["pending"], + min_id=reply["min"], + max_id=reply["max"], + # redis-py names the per-consumer count "pending"; ours is "count", to + # match the summary's own field and the TypeScript twin. + consumers=[ + PendingConsumer(name=c["name"], count=c["pending"]) + for c in reply.get("consumers") or [] + ], + ) async def lag(self, stream: str, group: str) -> int | None: """Messages published to ``stream`` but not yet delivered to ``group`` — diff --git a/src/xstreamq/consumer.py b/src/xstreamq/consumer.py index 13f8446..863e855 100644 --- a/src/xstreamq/consumer.py +++ b/src/xstreamq/consumer.py @@ -90,6 +90,16 @@ async def ensure_group(self, *, at: str | None = None) -> None: if "BUSYGROUP" not in str(exc): raise + @property + def stream(self) -> str: + """The stream *key* this loop reads (hash-tagged, not the user-facing name).""" + return self._stream + + @property + def group(self) -> str: + """The consumer group this loop reads under (``-retry`` for a retry loop).""" + return self._group + @property def alive(self) -> bool: """True while the consume loop is supposed to be running and its task diff --git a/src/xstreamq/keys.py b/src/xstreamq/keys.py index 526a1ec..bb2c4b1 100644 --- a/src/xstreamq/keys.py +++ b/src/xstreamq/keys.py @@ -14,6 +14,9 @@ from __future__ import annotations +import json +from typing import NamedTuple + def stream_key(name: str) -> str: """Return the internal stream key for a user-facing stream name. @@ -50,3 +53,67 @@ def dedupe_prefix(stream: str, group: str) -> str: def dedupe_key(stream: str, group: str, dedup_id: str) -> str: """The idempotency marker key for one message's stable :attr:`Message.dedup_id`.""" return f"{dedupe_prefix(stream, group)}{dedup_id}" + + +#: The set every known (stream, group) is registered in, so ops can enumerate +#: without ``SCAN``. +REGISTRY_KEY = "xstreamq:registry" + + +class RegistryEntry(NamedTuple): + """One (stream, group) pair, as recorded in the registry set. + + A NamedTuple rather than a dataclass so it still compares equal to a plain + ``(stream, group)`` tuple — the codec below predates this type and callers + (and tests) may unpack it either way. + """ + + stream: str + group: str + + +def registry_member(stream: str, group: str) -> str: + """How a (stream, group) pair is encoded as a member of :data:`REGISTRY_KEY`. + + JSON rather than a delimiter, because both halves are user-chosen strings and + a group named ``a.b`` against a stream named ``a`` would otherwise be + indistinguishable from a group named ``b`` against a stream named ``a.a``. + + Three details here are load-bearing, because set membership is exact-string + and this member must be **byte-identical** to the one the TypeScript twin + writes (``keys.ts``, ``registryMember``) — otherwise the same subscription + registered by both twins shows up as two rows and the shared dashboard the + set exists for counts everything twice: + + 1. The *user-facing* stream name, not :func:`stream_key`. ``orders``, never + ``{orders}``. + 2. Field order ``stream`` then ``group`` — so no ``sort_keys``, which would + emit ``group`` first and silently fork the encoding. + 3. No whitespace, matching ``JSON.stringify``'s default. + + The derived retry/DLQ keys are deliberately *not* stored: both are pure + functions of (stream, group), so writing them would be a second source of + truth that can go stale. + """ + return json.dumps({"stream": stream, "group": group}, separators=(",", ":")) + + +def parse_registry_member(member: str) -> RegistryEntry | None: + """Decode a registry member, or ``None`` if it isn't one. + + Unrecognised members are skipped rather than fatal: the set is shared, + long-lived, and written by whatever xstreamq version happened to be deployed + — one stale row must not make the whole listing unreadable. + """ + try: + parsed = json.loads(member) + except ValueError: + return None + + if not isinstance(parsed, dict): + return None + + stream, group = parsed.get("stream"), parsed.get("group") + if isinstance(stream, str) and isinstance(group, str): + return RegistryEntry(stream, group) + return None diff --git a/tests/test_cli.py b/tests/test_cli.py index 1b5c69e..33ba1f0 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -136,3 +136,126 @@ def test_purge_all_aborts_without_confirmation(cli_redis, stream_name): res = _invoke("dlq", "purge", stream_name, "--group", "g", "--all") # no --yes; prompt assert res.exit_code != 0 # aborted at the confirmation prompt assert cli_redis.xlen(dkey) == 1 # nothing purged + + +# -- ls / health (ported from the TypeScript CLI) ------------------------ + + +def test_ls_lists_registered_pairs(cli_redis, stream_name): + cli_redis.sadd(keys.REGISTRY_KEY, keys.registry_member(stream_name, "g")) + + res = _invoke("ls") + assert res.exit_code == 0 + assert f"{stream_name} g" in res.stdout + + +def test_ls_json_and_empty(cli_redis): + cli_redis.delete(keys.REGISTRY_KEY) + assert "nothingregistered" in _normalize(_invoke("ls").stdout) + + cli_redis.sadd(keys.REGISTRY_KEY, keys.registry_member("orders", "svc")) + data = json.loads(_invoke("ls", "--json").stdout) + assert {"stream": "orders", "group": "svc"} in data + + +def test_ls_skips_unreadable_members(cli_redis): + """A shared, long-lived set may hold rows from any version — one bad row must + not make the listing unreadable.""" + cli_redis.delete(keys.REGISTRY_KEY) + cli_redis.sadd(keys.REGISTRY_KEY, keys.registry_member("orders", "svc"), "junk", "{}") + + data = json.loads(_invoke("ls", "--json").stdout) + assert data == [{"stream": "orders", "group": "svc"}] + + +def test_ls_honours_custom_registry_key(cli_redis): + cli_redis.delete("custom:registry") + cli_redis.sadd("custom:registry", keys.registry_member("orders", "svc")) + + res = _invoke("ls", "--registry", "custom:registry") + assert res.exit_code == 0 + assert "orders svc" in res.stdout + + +def test_health_ok_exits_zero(cli_redis): + res = _invoke("health") + assert res.exit_code == 0 + out = _normalize(res.stdout) + assert "healthy" in out + assert "unhealthy" not in out + # The CLI holds no subscriptions of its own, so these are structurally fixed. + assert "reclaimernone" in out + assert "consumers0" in out + + +def test_health_unreachable_exits_one(): + """The exit code a container healthcheck branches on.""" + res = runner.invoke(build_app(), ["--url", "redis://localhost:6401", "health"]) + assert res.exit_code == 1 + assert "unhealthy" in _normalize(res.stdout) + + +def test_health_json_reports_each_check(cli_redis): + data = json.loads(_invoke("health", "--json").stdout) + assert data["healthy"] is True + assert data["running"] is True + assert data["redis"] == "ok" + assert data["reclaimer"] == "none" + assert data["consumers"] == [] + + +def test_pending_shows_breakdown(cli_redis, stream_name): + skey = keys.stream_key(stream_name) + cli_redis.xgroup_create(skey, "g", id="0", mkstream=True) + for i in range(3): + cli_redis.xadd(skey, encode({"i": i})) + cli_redis.xreadgroup("g", "worker-a", {skey: ">"}, count=2) + cli_redis.xreadgroup("g", "worker-b", {skey: ">"}, count=1) + + res = _invoke("pending", stream_name, "g") + assert res.exit_code == 0 + out = _normalize(res.stdout) + assert "pending3" in out + assert "byconsumer" in out + assert "workera2" in out + assert "workerb1" in out + + +def test_pending_json(cli_redis, stream_name): + skey = keys.stream_key(stream_name) + cli_redis.xgroup_create(skey, "g", id="0", mkstream=True) + cli_redis.xadd(skey, encode({"i": 1})) + cli_redis.xreadgroup("g", "solo", {skey: ">"}, count=1) + + data = json.loads(_invoke("pending", stream_name, "g", "--json").stdout) + assert data["count"] == 1 + assert data["min_id"] == data["max_id"] is not None + assert data["consumers"] == [{"name": "solo", "count": 1}] + + +def test_pending_empty_group(cli_redis, stream_name): + cli_redis.xgroup_create(keys.stream_key(stream_name), "g", id="0", mkstream=True) + + res = _invoke("pending", stream_name, "g") + assert res.exit_code == 0 + assert "nothingpending" in _normalize(res.stdout) + + +def test_pending_missing_group_errors_cleanly(cli_redis, stream_name): + """A deliberate refusal must read as a message, not a Python traceback.""" + cli_redis.xadd(keys.stream_key(stream_name), encode({"i": 1})) + + res = _invoke("pending", stream_name, "nope") + assert res.exit_code == 1 + assert "Traceback" not in res.stdout + str(res.stderr) + + +def test_depth_group_still_reports_a_plain_count(cli_redis, stream_name): + """depth shows the total only; the breakdown belongs to `pending`.""" + skey = keys.stream_key(stream_name) + cli_redis.xgroup_create(skey, "g", id="0", mkstream=True) + cli_redis.xadd(skey, encode({"i": 1})) + cli_redis.xreadgroup("g", "solo", {skey: ">"}, count=1) + + data = json.loads(_invoke("depth", stream_name, "--group", "g", "--json").stdout) + assert data["pending"] == 1 diff --git a/tests/test_health.py b/tests/test_health.py index 82d7088..9f0358f 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -101,3 +101,66 @@ async def handle(msg): # Redis still PINGs fine, but a loop is dead → unhealthy. assert await dq.healthy() is False + + +# -- health(): the itemised form healthy() delegates to ------------------ + + +async def test_health_itemises_a_serving_queue(redis_url, stream_name): + dq = XStreamQ(redis_url) + + @dq.subscriber(stream_name, group="g", retry_on_idle_ms=100) + async def handle(msg): + pass + + await dq.start() + try: + report = await dq.health() + assert report.healthy is True + assert report.running is True + assert report.redis == "ok" + assert report.reclaimer == "ok" # retry_on_idle_ms registered one + # Main loop + retry loop. + assert len(report.consumers) == 2 + assert all(c.alive for c in report.consumers) + finally: + await dq.stop() + + +async def test_health_reclaimer_is_none_without_retries(redis_url, stream_name): + """Nothing registered is vacuously fine — distinct from 'started and died'.""" + dq = XStreamQ(redis_url) + + @dq.subscriber(stream_name, group="g") + async def handle(msg): + pass + + await dq.start() + try: + report = await dq.health() + assert report.reclaimer == "none" + assert report.healthy is True + finally: + await dq.stop() + + +async def test_health_before_start_is_not_running(redis_url): + report = await XStreamQ(redis_url).health() + assert report.running is False + assert report.healthy is False + + +async def test_healthy_agrees_with_health(redis_url, stream_name): + """healthy() is defined as health().healthy — they must never disagree.""" + dq = XStreamQ(redis_url) + assert await dq.healthy() == (await dq.health()).healthy + + @dq.subscriber(stream_name, group="g", retry_on_idle_ms=100) + async def handle(msg): + pass + + await dq.start() + try: + assert await dq.healthy() == (await dq.health()).healthy is True + finally: + await dq.stop() diff --git a/tests/test_integration.py b/tests/test_integration.py index 2cc8c14..80fc179 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -37,6 +37,11 @@ async def _eq(coro, n): return await coro == n +async def _pending_count(dq, stream, group): + """pending() returns a summary; most assertions here only want the total.""" + return (await dq.pending(stream, group)).count + + async def test_publish_consume_happy_path(dq, stream_name): received = [] @@ -202,8 +207,12 @@ async def handle(msg): await dq.start() members = await redis_client.smembers("xstreamq:registry") - skey = keys.stream_key(stream_name) - assert any(skey in m and '"group":"g"' in m for m in members) + + # Pinned exactly, not by substring: set membership is exact-string, and this + # member must match the TypeScript twin byte-for-byte or a shared registry + # lists the same subscription twice. Note the *untagged* stream name. + assert keys.registry_member(stream_name, "g") in members + assert f'{{"stream":"{stream_name}","group":"g"}}' in members async def test_stream_length(dq, stream_name): @@ -232,12 +241,16 @@ async def handle(msg): # One message is delivered and parked in the handler → pending == 1, # the remaining four are published-but-undelivered → lag == 4. assert await _wait_until(lambda: blocked.is_set(), timeout=5) - assert await _wait_until_async(lambda: _eq(dq.pending(stream_name, "g"), 1), timeout=5) + assert await _wait_until_async( + lambda: _eq(_pending_count(dq, stream_name, "g"), 1), timeout=5 + ) assert await dq.lag(stream_name, "g") == 4 release.set() # Everything drains: nothing in-flight, nothing undelivered. - assert await _wait_until_async(lambda: _eq(dq.pending(stream_name, "g"), 0), timeout=5) + assert await _wait_until_async( + lambda: _eq(_pending_count(dq, stream_name, "g"), 0), timeout=5 + ) assert await dq.lag(stream_name, "g") == 0 finally: release.set() @@ -337,7 +350,7 @@ async def handle(msg): # 0..4 acked, 5 delivered-and-pending, 6..9 not yet delivered. assert await _wait_until(lambda: 5 in seen) - assert await _wait_until_async(lambda: _eq(dq.pending(stream_name, "g"), 1)) + assert await _wait_until_async(lambda: _eq(_pending_count(dq, stream_name, "g"), 1)) try: removed = await dq.compact(stream_name) assert removed == 5 # only 0..4 (below the pending entry 5) are trimmed @@ -459,3 +472,93 @@ async def handle(msg): # it runs again — if we'd wrongly marked the failure, this retry would be skipped. assert await _wait_until(lambda: 1 in attempts, timeout=8) assert attempts[0] == 0 + + +async def test_registered_enumerates_the_registry(dq, redis_client, stream_name): + @dq.subscriber(stream_name, group="g", retry_on_idle_ms=100) + async def handle(msg): + pass + + await dq.start() + entries = await dq.registered() + assert (stream_name, "g") in entries + # The user-facing name, not the hash-tagged key. + assert all(not e.stream.startswith("{") for e in entries) + + +async def test_registered_skips_unreadable_members(dq, redis_client, stream_name): + """One stale row from another version must not break the whole listing.""" + await dq.start() + await redis_client.sadd("xstreamq:registry", "not-json", '["a"]', '{"stream":"s"}') + await redis_client.sadd("xstreamq:registry", keys.registry_member(stream_name, "g")) + + entries = await dq.registered() + assert (stream_name, "g") in entries + + +async def test_registered_is_sorted(dq, redis_client): + """A Redis set has no order; an ops listing that shuffles is impossible to diff.""" + await dq.start() + await redis_client.delete("xstreamq:registry") + for stream, group in [("b", "2"), ("a", "1"), ("a", "2")]: + await redis_client.sadd("xstreamq:registry", keys.registry_member(stream, group)) + + assert await dq.registered() == [("a", "1"), ("a", "2"), ("b", "2")] + + +async def test_pending_reports_per_consumer_breakdown(redis_url, redis_client, stream_name): + """The breakdown is the point of the summary: a stuck worker shows up as one + consumer holding a share of the PEL while the others drain.""" + skey = keys.stream_key(stream_name) + await redis_client.xgroup_create(skey, "g", id="0", mkstream=True) + for i in range(3): + await redis_client.xadd(skey, encode({"i": i})) + # Deliver without acking, from two distinct consumer names. + await redis_client.xreadgroup("g", "worker-a", {skey: ">"}, count=2) + await redis_client.xreadgroup("g", "worker-b", {skey: ">"}, count=1) + + dq = XStreamQ(redis_url) + await dq.start() + try: + summary = await dq.pending(stream_name, "g") + assert summary.count == 3 + assert summary.min_id is not None and summary.max_id is not None + assert summary.min_id < summary.max_id + assert {(c.name, c.count) for c in summary.consumers} == {("worker-a", 2), ("worker-b", 1)} + # The parts must add up to the whole. + assert sum(c.count for c in summary.consumers) == summary.count + finally: + await dq.stop() + + +async def test_pending_empty_group_has_no_ids(redis_url, redis_client, stream_name): + skey = keys.stream_key(stream_name) + await redis_client.xgroup_create(skey, "g", id="0", mkstream=True) + + dq = XStreamQ(redis_url) + await dq.start() + try: + summary = await dq.pending(stream_name, "g") + assert summary.count == 0 + assert summary.min_id is None + assert summary.max_id is None + assert summary.consumers == [] + finally: + await dq.stop() + + +async def test_pending_raises_when_group_does_not_exist(redis_url, redis_client, stream_name): + """'never created' and 'nothing in flight' are different answers — returning 0 + for both is how a missing subscriber gets read as a healthy one.""" + import pytest + + from xstreamq import XStreamQError + + await redis_client.xadd(keys.stream_key(stream_name), encode({"i": 1})) + dq = XStreamQ(redis_url) + await dq.start() + try: + with pytest.raises(XStreamQError, match="no consumer group"): + await dq.pending(stream_name, "nope") + finally: + await dq.stop() diff --git a/tests/test_keys.py b/tests/test_keys.py index 7539a84..06a0c90 100644 --- a/tests/test_keys.py +++ b/tests/test_keys.py @@ -55,3 +55,56 @@ def test_dedupe_key_is_independent_of_retry_suffix(): # Main group "g" and its retry group "g-retry" must NOT collide — but the retry # consumer is wired with the BASE group, so both resolve to the same marker. assert keys.dedupe_key("s", "g", "x") != keys.dedupe_key("s", "g-retry", "x") + + +# -- registry membership ------------------------------------------------- +# +# These pin the cross-twin wire contract. The TypeScript implementation is +# `registryMember` in xstreamq-ts/src/keys.ts; if either side changes, the same +# subscription registered by both twins becomes two rows in one shared set and +# the v2 dashboard (the only reason the set exists) double-counts it. + + +def test_registry_member_matches_typescript_byte_for_byte(): + # Golden string, produced by `JSON.stringify({stream, group})` in the twin. + assert keys.registry_member("orders", "g") == '{"stream":"orders","group":"g"}' + + +def test_registry_member_uses_untagged_stream_name(): + # The name the user typed, not stream_key() — no braces. + assert "{orders}" not in keys.registry_member("orders", "g") + + +def test_registry_member_field_order_is_stream_then_group(): + # sort_keys would put "group" first and silently fork the encoding. + m = keys.registry_member("orders", "g") + assert m.index('"stream"') < m.index('"group"') + + +def test_registry_member_is_stable_across_calls(): + # Exact-string set membership: an unstable encoding would double up on restart. + assert keys.registry_member("s", "g") == keys.registry_member("s", "g") + + +def test_registry_member_round_trips(): + assert keys.parse_registry_member(keys.registry_member("orders", "g")) == ("orders", "g") + + +def test_registry_member_distinguishes_ambiguous_pairs(): + # The reason this is JSON and not "stream.group": both halves are user-chosen. + assert keys.registry_member("a", "b.c") != keys.registry_member("a.b", "c") + + +def test_parse_registry_member_rejects_non_members(): + # A shared, long-lived set may hold rows from any deployed version; one + # unreadable row must not break the whole listing. + assert keys.parse_registry_member("not json") is None + assert keys.parse_registry_member('"a string"') is None + assert keys.parse_registry_member("[1,2]") is None + assert keys.parse_registry_member('{"stream":"s"}') is None + assert keys.parse_registry_member('{"stream":1,"group":"g"}') is None + + +def test_parse_registry_member_tolerates_extra_fields(): + # Forward compatibility with a future twin that adds fields. + assert keys.parse_registry_member('{"stream":"s","group":"g","v":2}') == ("s", "g")