77ack (a terminal state), so retries (re-kicked as new messages) and crash
88redelivery (unacked, reclaimed via XAUTOCLAIM from the PEL) are unaffected.
99
10- `ProducerOnlyRedisStreamBroker` skips consumer-group declaration on startup, for
11- a broker used only to register tasks and `.kiq()` — never to `.listen()`. The
12- base `startup()` unconditionally XGROUP-CREATEs a group that then sits at 0-0
13- forever (0 consumers), reporting lag == XLEN permanently.
10+ `ProducerOnlyMixin` skips consumer-group declaration on startup, for a broker
11+ used only to register tasks and `.kiq()` — never to `.listen()`. The base
12+ `startup()` unconditionally XGROUP-CREATEs a group that then sits at 0-0 forever
13+ (0 consumers), reporting lag == XLEN permanently. Every producer broker (one per
14+ queue) must mix this in; only the consumer side declares the group.
15+
16+ `stable_consumer_name()` derives the consumer name from the container id
17+ (`HOSTNAME`), so it is unique per concurrent process — N replicas get N distinct
18+ names and never collide on one PEL — while being far more legible than the base
19+ broker's per-construction `uuid4()`. The name still rotates per redeploy (new
20+ container id), so pair it with `prune_idle_consumers()` at boot to bound registry
21+ growth: without pruning, each redeploy leaves a dead consumer behind and the
22+ group registry climbs to tens of thousands of idle (pending=0) entries.
1423"""
1524
25+ import os
26+ import socket
1627from typing import Awaitable , Callable
1728
1829from redis .asyncio import Redis
1930from taskiq_redis import RedisStreamBroker
2031from taskiq_redis .redis_broker import BaseRedisBroker
2132
2233
34+ def stable_consumer_name (consumer_group_name : str ) -> str :
35+ """Per-container consumer name: unique per concurrent process (N-replica safe)."""
36+ host = os .environ .get ("HOSTNAME" ) or socket .gethostname ()
37+ return f"{ consumer_group_name } { host }
38+
39+
40+ # 24h: a rolling deploy runs old+new replicas concurrently, so a live peer's
41+ # idle time never approaches this — only long-dead containers get pruned.
42+ PRUNE_MIN_IDLE_MS = 24 * 60 * 60 * 1000
43+
44+
45+ async def prune_idle_consumers (
46+ * ,
47+ url : str ,
48+ queue_name : str ,
49+ consumer_group_name : str ,
50+ min_idle_ms : int = PRUNE_MIN_IDLE_MS ,
51+ keep : str | None = None ,
52+ ) -> int :
53+ """DELCONSUMER every idle, empty (pending=0) consumer in the group.
54+
55+ Bounds registry growth from per-redeploy consumer-name rotation. Only removes
56+ consumers with pending=0 (DELCONSUMER on a consumer holding pending entries
57+ would silently orphan them) and idle >= `min_idle_ms`, and never `keep` (this
58+ process's own name). Best-effort: swallows errors so a prune failure never
59+ blocks worker startup.
60+ """
61+ removed = 0
62+ try :
63+ async with Redis .from_url (url ) as redis_conn :
64+ try :
65+ consumers = await redis_conn .xinfo_consumers (
66+ queue_name , consumer_group_name
67+ )
68+ except Exception :
69+ return 0
70+ for c in consumers :
71+ name = c ["name" ]
72+ name = name .decode () if isinstance (name , bytes ) else name
73+ if name == keep :
74+ continue
75+ if int (c ["pending" ]) != 0 or int (c ["idle" ]) < min_idle_ms :
76+ continue
77+ try :
78+ await redis_conn .xgroup_delconsumer (
79+ queue_name , consumer_group_name , name
80+ )
81+ removed += 1
82+ except Exception :
83+ continue
84+ except Exception :
85+ return removed
86+ return removed
87+
88+
2389class TrimOnAckRedisStreamBroker (RedisStreamBroker ):
2490 def _ack_generator (self , id : str , queue_name : str ) -> Callable [[], Awaitable [None ]]:
2591 async def _ack () -> None :
@@ -30,6 +96,12 @@ async def _ack() -> None:
3096 return _ack
3197
3298
33- class ProducerOnlyRedisStreamBroker (RedisStreamBroker ):
99+ class ProducerOnlyMixin :
100+ """Skip consumer-group declaration on startup — for `.kiq()`-only brokers."""
101+
34102 async def startup (self ) -> None :
35103 await BaseRedisBroker .startup (self )
104+
105+
106+ class ProducerOnlyRedisStreamBroker (ProducerOnlyMixin , RedisStreamBroker ):
107+ pass
0 commit comments