fix: notify Jira after queue retries exhaust#155
Conversation
| queued_for_retry = await self._retry_queue.enqueue_for_retry( | ||
| entry.message, str(e) | ||
| ) | ||
| if not queued_for_retry: |
There was a problem hiding this comment.
Nice job keeping notification failures from interrupting DLQ handling. Could we also acknowledge the original stream entry when queued_for_retry is false? This branch currently notifies and then continues, so terminal failures reached through the retry poller remain indefinitely in the Redis pending-entry list. A best-effort xack, regardless of whether the callback succeeds, plus a test for this path would keep the PEL clean.
There was a problem hiding this comment.
Addressed in 3f3036b. The retry-poller terminal path now performs a best-effort xack after the callback attempt, including when the callback fails. Added focused tests for both callback success and failure; the full unit suite and PR CI pass.
| redis_client = await self._get_redis() | ||
| await redis_client.xack(stream, CONSUMER_GROUP, message_id) | ||
|
|
||
| async def _notify_terminal_failure(self, message: QueueMessage, error: str) -> None: |
There was a problem hiding this comment.
Thanks for adding the cross-worker notification claim here. One concern: this claims only the callback; the retry entry itself is still read non-atomically by every worker via get_due_messages(), so two pollers can process the same event concurrently. If one succeeds while another exhausts the counter, we could move a successfully processed event to the DLQ and post a misleading terminal-failure comment; concurrent failures can also advance the retry counter more quickly than intended. Could we atomically claim/lease the retry entry before dispatching it (for example, with a Redis transaction or Lua operation), and add a multi-worker test?
There was a problem hiding this comment.
Addressed in 3f3036b. Due retry entries are now atomically claimed with a Redis Lua operation that advances their score to a 15-minute visibility lease before dispatch. A crashed worker’s entry becomes claimable again after the lease. Added both a concurrent-poller unit test and a real-Redis integration test; the full unit suite and PR CI pass.
Summary
QueueConsumerand invoke it only after a message reaches the dead-letter queueCloses #140
Test plan
uv run pytest tests/unit/queue/test_consumer_retry.py tests/unit/orchestrator/test_worker.py -q(55 passed)uv run pytest tests/unit/ -q(1462 passed, 2 skipped)uv run ruff check src/forge/queue/consumer.py src/forge/orchestrator/worker.py tests/unit/queue/test_consumer_retry.py tests/unit/orchestrator/test_worker.pyuv run ruff format --check src/forge/queue/consumer.py src/forge/orchestrator/worker.py tests/unit/queue/test_consumer_retry.py tests/unit/orchestrator/test_worker.pyType checking
uv run mypy src/forge/queue/consumer.py src/forge/orchestrator/worker.pystill reports the existing strict-mode errors in these modules (15 total); the new callback and Jira handler add no new reported mypy errors.