29 lines
1.1 KiB
Python
29 lines
1.1 KiB
Python
|
"""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})"
|
||
|
|
)
|