fix XDEL after XACK

Author: jp-agenta

api/entrypoints/worker_queues.py

@@ -31,7 +31,8 @@
 from taskiq import AsyncBroker, TaskiqEvents
 from taskiq.receiver import Receiver
 from taskiq.cli.worker.run import shutdown_broker
-from taskiq_redis import RedisStreamBroker
+
+from oss.src.tasks.taskiq.shared.broker import TrimOnAckRedisStreamBroker
 
 from oss.src.core.embeds.service import EmbedsService
 from oss.src.core.environments.service import EnvironmentsService
@@ -123,7 +124,7 @@ def _selected_queues() -> List[str]:
 
 
 def _build_webhooks_broker() -> tuple[AsyncBroker, int]:
-    broker = RedisStreamBroker(
+    broker = TrimOnAckRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:webhooks",
         consumer_group_name="worker-webhooks",
@@ -135,7 +136,7 @@ def _build_webhooks_broker() -> tuple[AsyncBroker, int]:
 
 
 def _build_triggers_broker() -> tuple[AsyncBroker, int]:
-    broker = RedisStreamBroker(
+    broker = TrimOnAckRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:triggers",
         consumer_group_name="worker-triggers",
@@ -175,7 +176,7 @@ def _build_triggers_broker() -> tuple[AsyncBroker, int]:
 
 
 def _build_interactions_broker() -> tuple[AsyncBroker, int]:
-    broker = RedisStreamBroker(
+    broker = TrimOnAckRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:interactions",
         consumer_group_name="worker-interactions",

api/entrypoints/worker_streams.py

@@ -17,7 +17,8 @@
 from typing import List
 
 from redis.asyncio import Redis
-from taskiq_redis import RedisStreamBroker
+
+from oss.src.tasks.taskiq.shared.broker import ProducerOnlyRedisStreamBroker
 
 from oss.src.core.events.service import EventsService
 from oss.src.core.secrets.services import VaultService
@@ -85,9 +86,11 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer:
 async def _build_events_worker(redis_client: Redis) -> StreamConsumer:
     events_service = EventsService(events_dao=EventsDAO())
 
-    # Webhook dispatch runs inside the events loop as its post-hook.
+    # Webhook dispatch runs inside the events loop as its post-hook: this broker
+    # only produces (.kiq), so it must not declare a consumer group it never
+    # reads — that group sits at 0-0 and reports lag == XLEN forever.
     webhooks_dao = WebhooksDAO()
-    broker = RedisStreamBroker(
+    broker = ProducerOnlyRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:webhooks",
         consumer_group_name="worker-events-webhooks-dispatcher",

api/oss/src/core/evaluations/runtime/broker.py

@@ -6,8 +6,8 @@
 """
 
 from taskiq import AsyncBroker
-from taskiq_redis import RedisStreamBroker
 
+from oss.src.tasks.taskiq.shared.broker import TrimOnAckRedisStreamBroker
 from oss.src.core.evaluations.runtime.runner import TaskiqEvaluationTaskRunner
 from oss.src.core.evaluations.service import EvaluationsService
 from oss.src.core.evaluators.service import SimpleEvaluatorsService
@@ -22,12 +22,13 @@
 MAXLEN_QUEUES_EVALUATIONS = 100_000
 
 
-class NoRedeliveryRedisStreamBroker(RedisStreamBroker):
+class NoRedeliveryRedisStreamBroker(TrimOnAckRedisStreamBroker):
     """Stream broker that never redelivers. `listen()` reads only NEW messages
     (`>`) and skips the XAUTOCLAIM pending-replay block, so a task that crashed
     mid-run is not re-served to later workers. Evaluation tasks are not safely
     re-runnable (`retry_on_error=False`); a stuck unacked entry replaying on every
-    worker restart is worse than dropping it.
+    worker restart is worse than dropping it. Inherits XDEL-on-ack so completed
+    entries leave the stream (XLEN = backlog).
     """
 
     async def listen(self):

api/oss/src/tasks/taskiq/shared/__init__.py

@@ -0,0 +1,35 @@
+"""Shared TaskIQ RedisStreamBroker mixins.
+
+`TrimOnAckRedisStreamBroker` XDELs each entry right after XACK. Without it, a
+TaskIQ queue retains every acked entry until maxlen-trims it, so XLEN reflects
+retained history (a rolling maxlen window) rather than backlog — the stream
+workers already XACK+XDEL, so this makes the queues match. XDEL fires only on
+ack (a terminal state), so retries (re-kicked as new messages) and crash
+redelivery (unacked, reclaimed via XAUTOCLAIM from the PEL) are unaffected.
+
+`ProducerOnlyRedisStreamBroker` skips consumer-group declaration on startup, for
+a broker used only to register tasks and `.kiq()` — never to `.listen()`. The
+base `startup()` unconditionally XGROUP-CREATEs a group that then sits at 0-0
+forever (0 consumers), reporting lag == XLEN permanently.
+"""
+
+from typing import Awaitable, Callable
+
+from redis.asyncio import Redis
+from taskiq_redis import RedisStreamBroker
+from taskiq_redis.redis_broker import BaseRedisBroker
+
+
+class TrimOnAckRedisStreamBroker(RedisStreamBroker):
+    def _ack_generator(self, id: str, queue_name: str) -> Callable[[], Awaitable[None]]:
+        async def _ack() -> None:
+            async with Redis(connection_pool=self.connection_pool) as redis_conn:
+                await redis_conn.xack(queue_name, self.consumer_group_name, id)
+                await redis_conn.xdel(queue_name, id)
+
+        return _ack
+
+
+class ProducerOnlyRedisStreamBroker(RedisStreamBroker):
+    async def startup(self) -> None:
+        await BaseRedisBroker.startup(self)