Account for ignorable payloads in retry script, try to prevent them in runner (#1000)

* Move guard to check for valid action before refreshing bug data

* We never log the string "skipping events", so remove assertions that those logs don't exist

* Slightly change mocking in test_retry

- use `mocker`fixture for pytests fixtures
- change import to import mock module
- use more descriptive name for function that returns a mock with an aiter error

* Account for invalid requests in queue processing

Author: grahamalama

jbi/retry.py

@@ -9,6 +9,7 @@
 
 import jbi.runner as runner
 from jbi.configuration import ACTIONS
+from jbi.errors import IgnoreInvalidRequestError
 from jbi.queue import get_dl_queue
 
 CONSTANT_RETRY = getenv("DL_QUEUE_CONSTANT_RETRY", "false") == "true"
@@ -62,6 +63,10 @@ async def retry_failed(item_executor=runner.execute_action, queue=get_dl_queue()
                     item_executor(item.payload, ACTIONS)
                     await queue.done(item)
                     metrics["events_processed"] += 1
+                except IgnoreInvalidRequestError:
+                    logger.warning("removing invalid event %s", item.identifier)
+                    await queue.done(item)
+                    metrics["events_processed"] += 1
                 except Exception:
                     metrics["events_failed"] += 1
                     logger.exception(

jbi/runner.py

@@ -217,6 +217,13 @@ def execute_action(
         if bug.is_private:
             raise IgnoreInvalidRequestError("private bugs are not supported")
 
+        try:
+            action = lookup_action(bug, actions)
+        except ActionNotFoundError as err:
+            raise IgnoreInvalidRequestError(
+                f"no bug whiteboard matching action tags: {err}"
+            ) from err
+
         logger.info(
             "Handling incoming request",
             extra=runner_context.model_dump(),
@@ -229,12 +236,7 @@ def execute_action(
             raise IgnoreInvalidRequestError(str(err)) from err
 
         runner_context = runner_context.update(bug=bug)
-        try:
-            action = lookup_action(bug, actions)
-        except ActionNotFoundError as err:
-            raise IgnoreInvalidRequestError(
-                f"no bug whiteboard matching action tags: {err}"
-            ) from err
+
         runner_context = runner_context.update(action=action)
 
         linked_issue_key: Optional[str] = bug.extract_from_see_also(

tests/unit/test_retry.py

@@ -1,17 +1,18 @@
 from datetime import UTC, datetime, timedelta
-from unittest.mock import MagicMock
+from unittest import mock
 
 import pytest
 
+from jbi.errors import IgnoreInvalidRequestError
 from jbi.retry import RETRY_TIMEOUT_DAYS, retry_failed
 from jbi.runner import execute_action
 
 
-def iter_error():
-    mock = MagicMock()
-    mock.__aiter__.return_value = None
-    mock.__aiter__.side_effect = Exception("Throwing an exception")
-    return mock
+def mock_aiter_error():
+    _mock = mock.MagicMock()
+    _mock.__aiter__.return_value = None
+    _mock.__aiter__.side_effect = Exception("Throwing an exception")
+    return _mock
 
 
 async def aiter_sync(iterable):
@@ -20,8 +21,8 @@ async def aiter_sync(iterable):
 
 
 @pytest.fixture
-def mock_executor():
-    return MagicMock(spec=execute_action)
+def mock_executor(mocker):
+    return mocker.MagicMock(spec=execute_action)
 
 
 @pytest.mark.asyncio
@@ -116,7 +117,6 @@ async def test_retry_remove_expired(
         len(mock_queue.done.call_args_list) == 2
     ), "both items should have been marked as done"
     assert caplog.text.count("failed to reprocess event") == 0
-    assert caplog.text.count("skipping events") == 0
     assert caplog.text.count("removing expired event") == 1
     mock_executor.assert_called_once()  # only one item should have been attempted to be processed
     assert metrics == {
@@ -128,18 +128,42 @@ async def test_retry_remove_expired(
     }
 
 
+@pytest.mark.asyncio
+async def test_retry_remove_invalid(
+    caplog, mock_queue, mock_executor, queue_item_factory
+):
+    mock_queue.retrieve.return_value = {
+        1: aiter_sync(queue_item_factory.create_batch(2))
+    }
+    mock_executor.side_effect = [
+        IgnoreInvalidRequestError("How did this get in here"),
+        mock.DEFAULT,
+    ]
+    metrics = await retry_failed(item_executor=mock_executor, queue=mock_queue)
+    assert (
+        len(mock_queue.done.call_args_list) == 2
+    ), "both items should have been marked as done"
+    assert caplog.text.count("removing invalid event") == 1
+    assert metrics == {
+        "bug_count": 1,
+        "events_processed": 2,
+        "events_skipped": 0,
+        "events_failed": 0,
+        "bugs_failed": 0,
+    }
+
+
 @pytest.mark.asyncio
 async def test_retry_bug_failed(caplog, mock_queue, mock_executor, queue_item_factory):
     mock_queue.retrieve.return_value = {
         1: aiter_sync([queue_item_factory(payload__bug__id=1)]),
-        2: iter_error(),
+        2: mock_aiter_error(),
     }
 
     metrics = await retry_failed(item_executor=mock_executor, queue=mock_queue)
     mock_queue.retrieve.assert_called_once()
     mock_queue.done.assert_called_once()  # one item should have been marked as done
     assert caplog.text.count("failed to reprocess event") == 0
-    assert caplog.text.count("skipping events") == 0
     assert caplog.text.count("removing expired event") == 0
     assert caplog.text.count("failed to parse events for bug") == 1
     mock_executor.assert_called_once()  # only one item should have been attempted to be processed

tests/unit/test_runner.py

@@ -160,7 +160,6 @@ def test_execution_logging_for_ignored_requests(
             execute_action(request=webhook, actions=actions)
 
     assert capturelogs.messages == [
-        "Handling incoming request",
         "Ignore incoming request: no bug whiteboard matching action tags: devtest",
     ]