feat(scheduler): dead-letter quarantine for poisoned stream messages

[larryluozhang]

Problem

Messages whose handlers deterministically fail (e.g. an oversized batch that
the LLM provider always rejects with HTTP 400) are redelivered forever via
xautoclaim. In production we observed one 40k-token batch failing every cycle
for days — burning LLM quota, spamming logs, and churning memory.

Fix

On the claim path, check times_delivered via xpending. Messages exceeding
SCHEDULER_MAX_DELIVERIES (default 5) are:

1. copied to scheduler:messages:stream:deadletter (capped, with payload +
   delivery count + source stream for forensics),
2. acked and deleted from the source stream.

Lookup failures fail open (message is processed rather than dropped by
mistake). No change for healthy messages.

[Memtensor-AI]

🤖 Open Code Review

Target: PR #2403
Task: dd6f3f0951d41218
Base: main
Head: upstream-pr/poison-deadletter

🔍 OpenCodeReview found 3 issue(s) in this PR.

---

1. src/memos/mem_scheduler/task_schedule_modules/redis_queue.py (L940-L944)

Dead code and incorrect control flow due to wrong indentation in the retry branch.

At this second call site (inside the except block's retry path), claimed = self._filter_poison_messages(...) and the return statement are dedented to function-body level instead of staying inside the elif len(claimed_result) == 3 / else branches. The result is:

1. The return [(stream_key, claimed)] if claimed else [] on line 943 runs unconditionally after the raise ValueError(...) branch — it can never be reached when the raise executes, and when it does execute, claimed may not be bound (if the unpacking paths were not entered).
2. The return [] on line 944 is unreachable dead code — it is placed after an unconditional return statement at the same indentation level.

The filter call and the return must be re-indented to sit inside the if/elif block, matching the pattern used in the first call site (lines 912–913):

                    # Quarantine poison messages (deliveries > limit) to a
                    # dead-letter stream instead of infinite redelivery.
                    claimed = self._filter_poison_messages(stream_key, claimed)
                    return [(stream_key, claimed)] if claimed else []
                return []

💡 Suggested Change

Before:

                    # Quarantine poison messages (deliveries > limit) to a
                # dead-letter stream instead of infinite redelivery.
                claimed = self._filter_poison_messages(stream_key, claimed)
                return [(stream_key, claimed)] if claimed else []
                return []

After:

                    # Quarantine poison messages (deliveries > limit) to a
                    # dead-letter stream instead of infinite redelivery.
                    claimed = self._filter_poison_messages(stream_key, claimed)
                    return [(stream_key, claimed)] if claimed else []
                return []

---

2. src/memos/mem_scheduler/task_schedule_modules/redis_queue.py (L800)

int(os.getenv("SCHEDULER_MAX_DELIVERIES", "5")) will raise an unhandled ValueError if the environment variable is set to a non-integer string (e.g., SCHEDULER_MAX_DELIVERIES=five). Unlike the xpending_range failure path — which has an explicit except that returns claimed safely — this conversion sits outside any try/except, so a bad env-var value will propagate as an uncaught exception from a method designed to fail safe. Wrap the conversion or provide a fallback:

try:
    max_deliveries = int(os.getenv("SCHEDULER_MAX_DELIVERIES", "5"))
except (ValueError, TypeError):
    max_deliveries = 5

💡 Suggested Change

Before:

        max_deliveries = int(os.getenv("SCHEDULER_MAX_DELIVERIES", "5"))

After:

        try:
            max_deliveries = int(os.getenv("SCHEDULER_MAX_DELIVERIES", "5"))
        except (ValueError, TypeError):
            max_deliveries = 5

---

3. src/memos/mem_scheduler/task_schedule_modules/redis_queue.py (L803-L805)

xpending_range is called with a hardcoded count=1000. If the stream has more than 1000 pending messages — exactly the high-backlog scenario this feature targets (the docstring cites a 400-message batch that loops infinitely) — the pend dict will be incomplete. Poison messages beyond position 1000 in the pending list will have a looked-up delivery count of 0, be passed through as good, and never be quarantined. Consider paginating, or at minimum setting the cap to a much larger sentinel (e.g., count=claimed_ids_set restricted to only the claimed IDs) to avoid silently skipping poison detection.

💡 Suggested Change

Before:

            for e in self._redis_conn.xpending_range(
                stream_key, self.consumer_group, "-", "+", 1000
            ):

After:

            claimed_ids = {mid for mid, _ in claimed}
            for e in self._redis_conn.xpending_range(
                stream_key, self.consumer_group, "-", "+", max(1000, len(claimed_ids))
            ):

Generated by cloud-assistant via Open Code Review.

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (2/2 executed). memos_python_core/changed-python-source: 2/2. Duration: 7s [advisory, non-gating] AI-generated tests on branch test/auto-gen-dd6f3f0951d41218-20260922091932: 112/112 passed — these do NOT affect the PR verdict; review the branch manually.

Branch: upstream-pr/poison-deadletter