diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index dc1a0da9..3af4c15b 100644 --- a/src/forge/orchestrator/worker.py +++ b/src/forge/orchestrator/worker.py @@ -102,12 +102,46 @@ def __init__( """ self.settings = get_settings() self.consumer_name = consumer_name or f"worker-{uuid.uuid4().hex[:8]}" - self.consumer = QueueConsumer(self.consumer_name) + self.consumer = QueueConsumer( + self.consumer_name, + terminal_failure_handler=self._handle_terminal_failure, + ) self.router = router or create_default_router() self._shutdown_event = asyncio.Event() self._checkpointer = None self._compiled_workflows: dict[str, Any] = {} # Cache compiled workflows by name + async def _handle_terminal_failure(self, message: QueueMessage, error: str) -> None: + """Post one Jira comment after queue retries are exhausted.""" + jira = JiraClient() + event_marker = f"Event/correlation ID: {message.event_id}" + try: + comments = await jira.get_comments(message.ticket_key) + if any(event_marker in comment.body for comment in comments): + logger.info( + f"Terminal failure notification already exists for event {message.event_id}" + ) + return + + safe_error = redact_secrets(error) + if len(safe_error) > 500: + safe_error = f"{safe_error[:500]}..." + details = ( + f"{safe_error}\n\n" + f"Ticket: {message.ticket_key}\n" + f"{event_marker}\n" + "Recovery: inspect the dead-letter entry, resolve the root cause, " + "then requeue the event." + ) + await jira.add_error_comment( + issue_key=message.ticket_key, + error_message=details, + node_name="queue execution (retries exhausted)", + ) + logger.info(f"Posted terminal queue failure notification to {message.ticket_key}") + finally: + await jira.close() + async def _handle_jira_event(self, message: QueueMessage) -> None: """Handle a Jira webhook event. diff --git a/src/forge/queue/consumer.py b/src/forge/queue/consumer.py index f86e0cd1..518c8ea5 100644 --- a/src/forge/queue/consumer.py +++ b/src/forge/queue/consumer.py @@ -2,6 +2,7 @@ import asyncio import logging +import uuid from collections import defaultdict from collections.abc import Callable, Coroutine from typing import Any @@ -24,8 +25,15 @@ # How often (seconds) the retry-queue poller wakes up POLL_INTERVAL_SECONDS = 10 +# Terminal-failure callbacks are claimed across workers because every worker +# polls the shared retry sorted set. The short lease recovers if a worker dies +# while posting the notification; completed claims are retained for idempotency. +TERMINAL_FAILURE_CLAIM_PREFIX = "forge:queue:terminal-failure:" +TERMINAL_FAILURE_CLAIM_TTL_SECONDS = 300 + # Handler type for message processing MessageHandler = Callable[[QueueMessage], Coroutine[Any, Any, None]] +TerminalFailureHandler = Callable[[QueueMessage, str], Coroutine[Any, Any, None]] class QueueConsumer: @@ -41,6 +49,7 @@ def __init__( redis_client: redis.Redis | None = None, jira_client: JiraClient | None = None, max_concurrent_tasks: int | None = None, + terminal_failure_handler: TerminalFailureHandler | None = None, ): """Initialize the queue consumer. @@ -50,6 +59,8 @@ def __init__( jira_client: Optional Jira client for freshness checks. max_concurrent_tasks: Maximum concurrent in-flight tasks. Defaults to ``settings.queue_max_concurrent_tasks`` when not provided. + terminal_failure_handler: Optional callback invoked after a message + is moved to the dead-letter queue. """ self.consumer_name = consumer_name self._redis = redis_client @@ -65,6 +76,7 @@ def __init__( self._semaphore = asyncio.Semaphore(concurrency) self._active_tasks: set[asyncio.Task[None]] = set() self._retry_queue = RetryQueue() + self._terminal_failure_handler = terminal_failure_handler async def _get_redis(self) -> redis.Redis: """Get or create Redis client.""" @@ -134,6 +146,59 @@ async def _ack(self, stream: str, message_id: str) -> None: 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: + """Invoke the terminal callback once per event without blocking DLQ handling.""" + if self._terminal_failure_handler is None: + return + + claim_key = f"{TERMINAL_FAILURE_CLAIM_PREFIX}{message.event_id}" + claim_token = uuid.uuid4().hex + try: + redis_client = await self._get_redis() + claimed = await redis_client.set( + claim_key, + claim_token, + ex=TERMINAL_FAILURE_CLAIM_TTL_SECONDS, + nx=True, + ) + except Exception as claim_error: + logger.error( + f"Could not claim terminal failure callback for {message.event_id}: {claim_error}" + ) + return + + if not claimed: + logger.info(f"Terminal failure already handled for event {message.event_id}") + return + + try: + await self._terminal_failure_handler(message, error) + except Exception as callback_error: + logger.error( + f"Terminal failure callback failed for {message.event_id}: {callback_error}" + ) + return + + try: + # A completed claim intentionally has no expiry: DLQ entries are + # retained too, and replaying one must not create another comment. + await redis_client.set(claim_key, "completed") + except Exception as completion_error: + logger.error( + f"Could not complete terminal failure claim for " + f"{message.event_id}: {completion_error}" + ) + + async def _ack_terminal_message(self, message: QueueMessage, stream: str) -> None: + """Best-effort acknowledge a message after it has reached the DLQ.""" + try: + await self._ack(stream, message.message_id) + except Exception as ack_error: + logger.warning( + f"xack failed for terminal event {message.event_id}; " + f"PEL entry may linger: {ack_error}" + ) + async def _process_message( self, message: QueueMessage, @@ -193,7 +258,8 @@ async def _process_message( try: moved_to_dlq = not await self._retry_queue.enqueue_for_retry(message, str(e)) if moved_to_dlq: - await self._ack(stream, message.message_id) + await self._notify_terminal_failure(message, str(e)) + await self._ack_terminal_message(message, stream) except Exception as retry_err: logger.error( f"Failed to enqueue {message.event_id} for retry: {retry_err}. " @@ -249,7 +315,7 @@ async def _process_retry_queue(self) -> None: """ while self._running: try: - entries = await self._retry_queue.get_due_messages() + entries = await self._retry_queue.claim_due_messages() for entry in entries: retry_stream = ( JIRA_STREAM if entry.message.source == EventSource.JIRA else GITHUB_STREAM @@ -268,7 +334,12 @@ async def _process_retry_queue(self) -> None: # the counter keeps accumulating and the message can eventually # reach the dead-letter queue. await self._retry_queue.remove_from_retry_without_counter_reset(entry) - await self._retry_queue.enqueue_for_retry(entry.message, str(e)) + queued_for_retry = await self._retry_queue.enqueue_for_retry( + entry.message, str(e) + ) + if not queued_for_retry: + await self._notify_terminal_failure(entry.message, str(e)) + await self._ack_terminal_message(entry.message, retry_stream) continue # Message processing succeeded — clean up retry state and diff --git a/src/forge/queue/retry.py b/src/forge/queue/retry.py index 6733f443..c76cac14 100644 --- a/src/forge/queue/retry.py +++ b/src/forge/queue/retry.py @@ -22,6 +22,20 @@ RETRY_BACKOFF_MULTIPLIER = 2 INITIAL_RETRY_DELAY_SECONDS = 30 MAX_RETRY_DELAY_SECONDS = 3600 # 1 hour +RETRY_CLAIM_LEASE_SECONDS = 900 # 15 minutes + +# Atomically move due entries beyond the visibility window before returning +# them. Other pollers cannot claim the same entries while this lease is active; +# if a worker exits before cleanup, the entries become visible again. +_CLAIM_DUE_MESSAGES_SCRIPT = """ +local entries = redis.call( + "ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1], "LIMIT", 0, ARGV[3] +) +for _, entry in ipairs(entries) do + redis.call("ZADD", KEYS[1], ARGV[2], entry) +end +return entries +""" @dataclass @@ -153,25 +167,27 @@ async def _move_to_dead_letter( f"{attempt} attempts. Error: {error}" ) - async def get_due_messages(self, limit: int = 10) -> list[RetryEntry]: - """Get messages that are due for retry. + async def claim_due_messages(self, limit: int = 10) -> list[RetryEntry]: + """Atomically lease and return messages that are due for retry. Args: limit: Maximum number of messages to return. Returns: - List of retry entries ready to be processed. + List of retry entries exclusively leased to this poller for the + visibility window. """ redis = await self._get_redis() now = datetime.utcnow().timestamp() + lease_until = now + RETRY_CLAIM_LEASE_SECONDS - # Get messages with score <= now - entries = await redis.zrangebyscore( + entries = await redis.eval( + _CLAIM_DUE_MESSAGES_SCRIPT, + 1, RETRY_QUEUE_KEY, - "-inf", now, - start=0, - num=limit, + lease_until, + limit, ) results = [] diff --git a/tests/integration/redis/test_queue_integration.py b/tests/integration/redis/test_queue_integration.py index d6ccdd2a..826aa650 100644 --- a/tests/integration/redis/test_queue_integration.py +++ b/tests/integration/redis/test_queue_integration.py @@ -5,6 +5,8 @@ """ import asyncio +import json +from datetime import datetime, timedelta import pytest @@ -12,6 +14,7 @@ from forge.queue.consumer import CONSUMER_GROUP, QueueConsumer from forge.queue.models import QueueMessage from forge.queue.producer import GITHUB_STREAM, JIRA_STREAM, QueueProducer +from forge.queue.retry import RETRY_QUEUE_KEY, RetryEntry, RetryQueue @pytest.mark.integration @@ -262,6 +265,48 @@ async def handler(message: QueueMessage): assert len(messages) == 0 +@pytest.mark.integration +class TestRetryQueue: + """Test retry leasing against a real shared Redis instance.""" + + async def test_due_entry_is_claimed_by_only_one_worker(self, redis_client): + """Concurrent pollers must not receive the same due retry entry.""" + message = QueueMessage( + message_id="1-0", + event_id="lease-test-1", + source=EventSource.JIRA, + event_type="issue_updated", + ticket_key="TEST-LEASE", + ) + entry = RetryEntry( + message=message, + attempt=1, + next_retry=datetime.utcnow() - timedelta(seconds=1), + last_error="temporary failure", + ) + serialized_entry = json.dumps(entry.to_dict()) + await redis_client.zadd( + RETRY_QUEUE_KEY, + {serialized_entry: entry.next_retry.timestamp()}, + ) + + first = RetryQueue() + second = RetryQueue() + first._redis = redis_client + second._redis = redis_client + + claims = await asyncio.gather( + first.claim_due_messages(), + second.claim_due_messages(), + ) + + assert sorted(len(claim) for claim in claims) == [0, 1] + assert sum(claim[0].message.event_id == message.event_id for claim in claims if claim) == 1 + lease_score = await redis_client.zscore(RETRY_QUEUE_KEY, serialized_entry) + assert lease_score is not None + assert lease_score > datetime.utcnow().timestamp() + + @pytest.mark.integration class TestQueueMessageSerialization: """Test message serialization through the queue.""" diff --git a/tests/unit/orchestrator/test_worker.py b/tests/unit/orchestrator/test_worker.py index 30a7fe62..690c1e58 100644 --- a/tests/unit/orchestrator/test_worker.py +++ b/tests/unit/orchestrator/test_worker.py @@ -74,6 +74,57 @@ async def test_report_new_workflow_error_skips_non_reportable_errors( notify.assert_not_awaited() +@pytest.mark.asyncio +async def test_terminal_failure_posts_sanitized_recovery_comment(): + worker = OrchestratorWorker(consumer_name="test-worker") + message = QueueMessage( + message_id="1-0", + event_id="evt-terminal-1", + source=EventSource.JIRA, + event_type="issue_updated", + ticket_key="TEST-123", + ) + jira = AsyncMock() + jira.get_comments = AsyncMock(return_value=[]) + + with patch("forge.orchestrator.worker.JiraClient", return_value=jira): + await worker._handle_terminal_failure( + message, + "clone https://ghp_abcdefghijklmnopqrstuvwxyz123456@github.com/acme/repo failed", + ) + + jira.add_error_comment.assert_awaited_once() + kwargs = jira.add_error_comment.await_args.kwargs + assert kwargs["issue_key"] == "TEST-123" + assert "[REDACTED]" in kwargs["error_message"] + assert "ghp_" not in kwargs["error_message"] + assert "Event/correlation ID: evt-terminal-1" in kwargs["error_message"] + assert "Recovery:" in kwargs["error_message"] + jira.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_terminal_failure_skips_existing_event_comment(): + worker = OrchestratorWorker(consumer_name="test-worker") + message = QueueMessage( + message_id="1-0", + event_id="evt-terminal-1", + source=EventSource.JIRA, + event_type="issue_updated", + ticket_key="TEST-123", + ) + jira = AsyncMock() + jira.get_comments = AsyncMock( + return_value=[MagicMock(body="Event/correlation ID: evt-terminal-1")] + ) + + with patch("forge.orchestrator.worker.JiraClient", return_value=jira): + await worker._handle_terminal_failure(message, "failed") + + jira.add_error_comment.assert_not_awaited() + jira.close.assert_awaited_once() + + class TestQuestionDetection: """Tests for Q&A mode question detection.""" diff --git a/tests/unit/queue/test_consumer.py b/tests/unit/queue/test_consumer.py index 1ffc99b1..e88c7e8c 100644 --- a/tests/unit/queue/test_consumer.py +++ b/tests/unit/queue/test_consumer.py @@ -51,7 +51,7 @@ def _make_consumer(redis_mock: MagicMock, max_tasks: int = 20) -> QueueConsumer: # the message for retry. retry_mock = MagicMock(spec=RetryQueue) retry_mock.enqueue_for_retry = AsyncMock(return_value=True) # queued, not DLQ - retry_mock.get_due_messages = AsyncMock(return_value=[]) + retry_mock.claim_due_messages = AsyncMock(return_value=[]) retry_mock.remove_from_retry = AsyncMock() retry_mock.remove_from_retry_without_counter_reset = AsyncMock() consumer._retry_queue = retry_mock diff --git a/tests/unit/queue/test_consumer_retry.py b/tests/unit/queue/test_consumer_retry.py index f854a921..8d115d66 100644 --- a/tests/unit/queue/test_consumer_retry.py +++ b/tests/unit/queue/test_consumer_retry.py @@ -36,7 +36,7 @@ def make_consumer() -> QueueConsumer: # Replace the real RetryQueue with a mock consumer._retry_queue = MagicMock(spec=RetryQueue) consumer._retry_queue.enqueue_for_retry = AsyncMock(return_value=True) - consumer._retry_queue.get_due_messages = AsyncMock(return_value=[]) + consumer._retry_queue.claim_due_messages = AsyncMock(return_value=[]) consumer._retry_queue.remove_from_retry = AsyncMock() consumer._retry_queue.remove_from_retry_without_counter_reset = AsyncMock() return consumer @@ -160,6 +160,43 @@ async def xreadgroup_side_effect(*_args, **_kwargs): # xack must be called to clear PEL after DLQ move redis_mock.xack.assert_called_once_with("stream:jira", CONSUMER_GROUP, "1-0") + @pytest.mark.asyncio + async def test_terminal_callback_failure_does_not_block_dlq_ack(self): + """A notification failure must not prevent acknowledging a DLQ message.""" + callback = AsyncMock(side_effect=RuntimeError("Jira unavailable")) + consumer = QueueConsumer("test-worker", terminal_failure_handler=callback) + consumer._retry_queue = MagicMock(spec=RetryQueue) + consumer._retry_queue.enqueue_for_retry = AsyncMock(return_value=False) + + redis_mock = AsyncMock() + redis_mock.set = AsyncMock(return_value=True) + redis_mock.xack = AsyncMock() + consumer._redis = redis_mock + consumer.register_handler( + EventSource.JIRA, + AsyncMock(side_effect=RuntimeError("permanent failure")), + ) + + await consumer._process_message(make_message(), "stream:jira") + + callback.assert_awaited_once() + redis_mock.xack.assert_awaited_once_with("stream:jira", CONSUMER_GROUP, "1-0") + + @pytest.mark.asyncio + async def test_terminal_callback_is_claimed_once_per_event(self): + """Duplicate terminal handling must not invoke the callback twice.""" + callback = AsyncMock() + consumer = QueueConsumer("test-worker", terminal_failure_handler=callback) + redis_mock = AsyncMock() + redis_mock.set = AsyncMock(side_effect=[True, None, False]) + consumer._redis = redis_mock + message = make_message() + + await consumer._notify_terminal_failure(message, "failed") + await consumer._notify_terminal_failure(message, "failed again") + + callback.assert_awaited_once_with(message, "failed") + # --------------------------------------------------------------------------- # _process_retry_queue — background poller @@ -181,7 +218,7 @@ async def test_successful_retry_removes_entry(self): call_count = 0 - async def get_due_side_effect(*_args, **_kwargs): + async def claim_due_side_effect(*_args, **_kwargs): nonlocal call_count call_count += 1 if call_count == 1: @@ -189,7 +226,7 @@ async def get_due_side_effect(*_args, **_kwargs): consumer._running = False return [] - consumer._retry_queue.get_due_messages = AsyncMock(side_effect=get_due_side_effect) + consumer._retry_queue.claim_due_messages = AsyncMock(side_effect=claim_due_side_effect) handler = AsyncMock() consumer.register_handler(EventSource.JIRA, handler) @@ -213,7 +250,7 @@ async def test_failed_retry_reenqueues(self): call_count = 0 - async def get_due_side_effect(*_args, **_kwargs): + async def claim_due_side_effect(*_args, **_kwargs): nonlocal call_count call_count += 1 if call_count == 1: @@ -221,7 +258,7 @@ async def get_due_side_effect(*_args, **_kwargs): consumer._running = False return [] - consumer._retry_queue.get_due_messages = AsyncMock(side_effect=get_due_side_effect) + consumer._retry_queue.claim_due_messages = AsyncMock(side_effect=claim_due_side_effect) failing_handler = AsyncMock(side_effect=RuntimeError("still broken")) consumer.register_handler(EventSource.JIRA, failing_handler) @@ -239,6 +276,88 @@ async def get_due_side_effect(*_args, **_kwargs): assert enqueue_args[0].event_id == "evt-001" assert "still broken" in enqueue_args[1] + @pytest.mark.asyncio + async def test_exhausted_retry_invokes_terminal_callback(self): + """The retry poller notifies only after re-enqueue moves the event to DLQ.""" + callback = AsyncMock() + consumer = make_consumer() + consumer._terminal_failure_handler = callback + message = make_message() + entry = RetryEntry( + message=message, + attempt=3, + next_retry=datetime.utcnow(), + last_error="still broken", + ) + + call_count = 0 + + async def claim_due_side_effect(*_args, **_kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + return [entry] + consumer._running = False + return [] + + consumer._retry_queue.claim_due_messages = AsyncMock(side_effect=claim_due_side_effect) + consumer._retry_queue.enqueue_for_retry = AsyncMock(return_value=False) + consumer._redis = AsyncMock() + consumer._redis.set = AsyncMock(side_effect=[True, None]) + consumer.register_handler( + EventSource.JIRA, + AsyncMock(side_effect=RuntimeError("retries exhausted")), + ) + + with patch("forge.queue.consumer.asyncio.sleep", new_callable=AsyncMock): + await consumer._process_retry_queue() + + callback.assert_awaited_once_with(message, "retries exhausted") + consumer._redis.xack.assert_awaited_once_with( + "forge:events:jira", CONSUMER_GROUP, message.message_id + ) + + @pytest.mark.asyncio + async def test_terminal_callback_failure_still_acknowledges_retry(self): + """A callback failure cannot leave a terminal retry in the PEL.""" + callback = AsyncMock(side_effect=RuntimeError("Jira unavailable")) + consumer = make_consumer() + consumer._terminal_failure_handler = callback + message = make_message() + entry = RetryEntry( + message=message, + attempt=3, + next_retry=datetime.utcnow(), + last_error="still broken", + ) + + call_count = 0 + + async def claim_due_side_effect(*_args, **_kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + return [entry] + consumer._running = False + return [] + + consumer._retry_queue.claim_due_messages = AsyncMock(side_effect=claim_due_side_effect) + consumer._retry_queue.enqueue_for_retry = AsyncMock(return_value=False) + consumer._redis = AsyncMock() + consumer._redis.set = AsyncMock(return_value=True) + consumer.register_handler( + EventSource.JIRA, + AsyncMock(side_effect=RuntimeError("retries exhausted")), + ) + + with patch("forge.queue.consumer.asyncio.sleep", new_callable=AsyncMock): + await consumer._process_retry_queue() + + callback.assert_awaited_once_with(message, "retries exhausted") + consumer._redis.xack.assert_awaited_once_with( + "forge:events:jira", CONSUMER_GROUP, message.message_id + ) + @pytest.mark.asyncio async def test_empty_retry_queue_no_action(self): """Poller with no due messages does nothing.""" @@ -246,14 +365,14 @@ async def test_empty_retry_queue_no_action(self): call_count = 0 - async def get_due_side_effect(*_args, **_kwargs): + async def claim_due_side_effect(*_args, **_kwargs): nonlocal call_count call_count += 1 if call_count >= 2: consumer._running = False return [] - consumer._retry_queue.get_due_messages = AsyncMock(side_effect=get_due_side_effect) + consumer._retry_queue.claim_due_messages = AsyncMock(side_effect=claim_due_side_effect) with patch("forge.queue.consumer.asyncio.sleep", new_callable=AsyncMock): await consumer._process_retry_queue() diff --git a/tests/unit/queue/test_retry_queue.py b/tests/unit/queue/test_retry_queue.py index 1ee18088..eddf0b81 100644 --- a/tests/unit/queue/test_retry_queue.py +++ b/tests/unit/queue/test_retry_queue.py @@ -1,5 +1,6 @@ """Unit tests for RetryQueue class.""" +import asyncio import json from datetime import datetime from unittest.mock import AsyncMock, patch @@ -11,6 +12,7 @@ from forge.queue.retry import ( DEAD_LETTER_KEY, MAX_RETRY_ATTEMPTS, + RETRY_CLAIM_LEASE_SECONDS, RETRY_QUEUE_KEY, RetryEntry, RetryQueue, @@ -38,6 +40,7 @@ def make_redis_mock() -> AsyncMock: mock.expire = AsyncMock() mock.zadd = AsyncMock() mock.zrangebyscore = AsyncMock(return_value=[]) + mock.eval = AsyncMock(return_value=[]) mock.zrem = AsyncMock() mock.delete = AsyncMock() mock.rpush = AsyncMock() @@ -111,11 +114,11 @@ async def capture_zadd(_key, mapping): # --------------------------------------------------------------------------- -# get_due_messages +# claim_due_messages # --------------------------------------------------------------------------- -class TestGetDueMessages: +class TestClaimDueMessages: @pytest.mark.asyncio async def test_returns_parsed_entries(self): message = make_message() @@ -130,25 +133,66 @@ async def test_returns_parsed_entries(self): rq = RetryQueue() redis = make_redis_mock() - redis.zrangebyscore = AsyncMock(return_value=[json.dumps(raw_entry).encode()]) + redis.eval = AsyncMock(return_value=[json.dumps(raw_entry).encode()]) rq._redis = redis - results = await rq.get_due_messages() + results = await rq.claim_due_messages() assert len(results) == 1 assert results[0].message.event_id == "evt-001" assert results[0].attempt == 1 + eval_args = redis.eval.await_args.args + assert eval_args[1] == 1 + assert eval_args[2] == RETRY_QUEUE_KEY + assert eval_args[4] - eval_args[3] == RETRY_CLAIM_LEASE_SECONDS + assert eval_args[5] == 10 @pytest.mark.asyncio async def test_empty_queue_returns_empty_list(self): rq = RetryQueue() redis = make_redis_mock() - redis.zrangebyscore = AsyncMock(return_value=[]) + redis.eval = AsyncMock(return_value=[]) rq._redis = redis - results = await rq.get_due_messages() + results = await rq.claim_due_messages() assert results == [] + @pytest.mark.asyncio + async def test_concurrent_workers_receive_entry_once(self): + """Two workers sharing Redis cannot claim the same due entry.""" + entry = RetryEntry( + message=make_message(), + attempt=1, + next_retry=datetime(2024, 1, 1), + last_error="oops", + ) + serialized_entry = json.dumps(entry.to_dict()).encode() + claim_lock = asyncio.Lock() + entry_is_due = True + + async def atomic_eval(*_args): + nonlocal entry_is_due + async with claim_lock: + if not entry_is_due: + return [] + entry_is_due = False + return [serialized_entry] + + redis = make_redis_mock() + redis.eval = AsyncMock(side_effect=atomic_eval) + first = RetryQueue() + second = RetryQueue() + first._redis = redis + second._redis = redis + + claims = await asyncio.gather( + first.claim_due_messages(), + second.claim_due_messages(), + ) + + assert sorted(len(claim) for claim in claims) == [0, 1] + assert redis.eval.await_count == 2 + # --------------------------------------------------------------------------- # remove_from_retry