fix: delete taskiq stream entries on ack to bound Redis memory

[Felipe Alvarado (falvaradorodriguez)]

Redis memory grew without bound. RedisStreamBroker acknowledges messages with XACK, which only clears the consumer group's pending entries list — the entry itself stays in the append-only stream forever. Every task ever enqueued was still there.

Measured on staging:

• 761,008 entries / 199.7 MB in the taskiq stream, ~262 bytes per entry
• pending: 0, lag: 0 — workers fully caught up, so 100% of those entries were already acknowledged, pure garbage
• db0:keys=3213,expires=3212 — the stream was the only key in the whole instance without a TTL
• The stream was only 3.78 days old (recorded-first-entry-id → last-generated-id), which means ~201,000 entries/day ≈ 53 MB/day, with no ceiling

Production has not been measured yet, but it is likely worse rather than better: both cron tasks enqueue one task per row of the contract table, and production holds far more contracts than staging.

The leak dates back to the dramatiq → taskiq migration (#307, June 22nd). The dramatiq broker deleted the message on ack (HDEL in its dispatch.lua), so memory was proportional to messages in flight; streams do not do that, and retention silently became our responsibility.

How was it fixed? 🎯

DeleteOnAckRedisStreamBroker overrides the ack callback to delete the entry alongside acknowledging it, keeping the stream size proportional to the messages in flight. Both commands run in a single transaction so an acknowledged entry is never left behind.

Why not maxlen, the parameter the library offers for this (added in taskiq-redis#81, and requested for the cluster broker in #92): it trims by length without checking whether an entry was processed. Sizing it means choosing how many enqueued tasks you are willing to lose in the worst case, and losing tasks is not acceptable here.

Why not XACKDEL, which does this atomically in one command: it requires Redis 8.2 or newer and production runs 7.4.10. Same trade-off malvex/sheppy made, whose PR #20 moved from XACKDEL back to XACK+XDEL purely to lower its minimum Redis version from 8.2 to 6.2.

References

This is a well-known problem, not something specific to us:

• redis#7059 — "Redis Streams With One Consumer Group Do Not Delete Messages When Acknowledged"
• redis#6941 — "How to prevent Redis stream memory increases infinitely?"
• redis#13441 — Managing Streams memory on high volatility scenarios
• taskiq-redis#118 — a user on Kubernetes setting maxlen=1_500 by hand
• Redis considered delete-on-ack common enough to add XACKDEL in 8.2, and their own team treats it as the standard path: redis-benchmarks-specification#467 adds coverage for "stream delete strategies (XDELEX / XACKDEL, DELREF and ACKED)"

Compatibility

• XACK and XDEL are available since Redis 5.0, so this works on both production (7.4.10) and staging (8.10). Verified by running the full test suite against a 7.4.10 server. - Works on cluster: both are single-key commands on the same stream key, so the transact
  #### Verification
  - Full test suite green against Redis 7.4.10: 87 passed
• The new test fails without the fix (AssertionError: 1 != 0), confirming it catches the regression - Ack on the success path: xlen 1 → 0, pending 0, lag 0
• Ack on the failure path: confirmed the ack happens after execution (AcknowledgeType.WHEN_SAVED is the default in receiver.py:77 and cli/worker/args.py:53, and we do not pass --ack-type in docker/web/taskiq/worker/run.sh), and that SmartRetryMiddleware re-enqueues **befores inside run_task, receiver.py:349-359). A task that always fails produces its 5attempts and leaves the stream empty: no task is lost.
• If a worker dies before acking, the entry stays in the stream and the PEL and is recla idle_timeout, re-executed and only then deleted. At-least-once delivery is preserved.