"""Redis XINFO GROUPS helpers, per phase-0a-test-suite-spec.md ยง7.""" import time def group_lag(redis_client, stream: str, group: str) -> dict: """Return the XINFO GROUPS entry for `group` on `stream` as a dict.""" groups = redis_client.xinfo_groups(stream) for g in groups: if g.get("name") == group: return g raise ValueError(f"consumer group {group!r} not found on stream {stream!r}: {groups}") def wait_for_drain(redis_client, stream: str, group: str, timeout: float = 15.0) -> None: """Block until `group`'s lag and pending count both reach 0, or raise TimeoutError.""" deadline = time.monotonic() + timeout last = None while time.monotonic() < deadline: g = group_lag(redis_client, stream, group) lag = g.get("lag") or 0 pending = g.get("pending") or 0 last = (lag, pending) if lag == 0 and pending == 0: return time.sleep(0.25) raise TimeoutError( f"group {group!r} on {stream!r} did not drain within {timeout}s (last lag/pending={last})" )