pkgevents: drop s3:TestEvent silently, report other failures per message (#5162)

sir-sigurd · claude · web-flow · commit c2596cb08d42 · 2026-08-03T19:24:23.000+05:00
Co-authored-by: Claude Opus 5 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/lambdas/pkgevents/CHANGELOG.md b/lambdas/pkgevents/CHANGELOG.md
@@ -17,6 +17,7 @@ where verb is one of
 
 ## Changes
 
+- [Fixed] Drop `s3:TestEvent` messages and report other failures per message via `ReportBatchItemFailures`, instead of failing the whole SQS batch and dead-lettering the real package events in it ([#5162](https://github.com/quiltdata/quilt/pull/5162))
 - [Fixed] Process package pointers from year 2026+ ([#4683](https://github.com/quiltdata/quilt/pull/4683))
 - [Changed] Migrate to proper package structure ([#4647](https://github.com/quiltdata/quilt/pull/4647))
 - [Changed] Switch to uv ([#4647](https://github.com/quiltdata/quilt/pull/4647))
diff --git a/lambdas/pkgevents/src/t4_lambda_pkgevents/__init__.py b/lambdas/pkgevents/src/t4_lambda_pkgevents/__init__.py
@@ -9,44 +9,17 @@
 
 EXPECTED_POINTER_SIZE = 64
 
+# S3 sends this to the notification queue when a bucket's notification
+# configuration is created; it carries no 'Records'.
+TEST_EVENT = 's3:TestEvent'
+
+PUT_EVENTS_MAX_ENTRIES = 10  # PutEvents API limit
+
 event_bridge = boto3.client('events')
 s3 = boto3.client('s3')
 logger = get_quilt_logger()
 
 
-class PutEventsException(Exception):
-    pass
-
-
-class EventsQueue:
-    MAX_SIZE = 10
-
-    def __init__(self):
-        self._events = []
-
-    def append(self, event):
-        self._events.append(event)
-        if len(self) >= self.MAX_SIZE:
-            self._flush()
-
-    def _flush(self):
-        events = self._events
-        self._events = []
-        resp = event_bridge.put_events(Entries=events)
-        if resp['FailedEntryCount']:
-            raise PutEventsException(resp)
-
-    def flush(self):
-        if self:
-            self._flush()
-
-    def __len__(self):
-        return len(self._events)
-
-    def __bool__(self):
-        return bool(self._events)
-
-
 PKG_POINTER_REGEX = re.compile(r'\.quilt/named_packages/([\w-]+/[\w-]+)/([0-9]{10})')
 
 
@@ -94,13 +67,56 @@ def pkg_created_event(s3_event):
     }
 
 
+def get_s3_events(body):
+    """Return the S3 events from an SQS message body ([] for S3 test events);
+    raise for any other unexpected shape."""
+    body = json.loads(body)
+    if body.get('Event') == TEST_EVENT:
+        # arrives on every bucket add, not worth logging
+        return []
+    return body['Records']
+
+
+def publish(entries):
+    """Publish (message id, event) pairs to EventBridge and return the ids of
+    the messages whose events failed to publish."""
+    failed_message_ids = set()
+    for chunk in itertools.batched(entries, PUT_EVENTS_MAX_ENTRIES):
+        resp = event_bridge.put_events(Entries=[event for _, event in chunk])
+        if resp['FailedEntryCount']:
+            # response entries are in the same order as the request entries
+            for (message_id, _), resp_entry in zip(chunk, resp['Entries'], strict=True):
+                if 'ErrorCode' in resp_entry:
+                    logger.warning('failed to publish event from message %s: %s', message_id, resp_entry)
+                    failed_message_ids.add(message_id)
+    return failed_message_ids
+
+
 def handler(event, context):
-    s3_events = itertools.chain.from_iterable(
-        json.loads(record['body'])['Records']
-        for record in event['Records']
-    )
-    queue = EventsQueue()
-    for event in filter(None, map(pkg_created_event, s3_events)):
-        queue.append(event)
-
-    queue.flush()
+    failed_message_ids = set()
+    entries = []
+    for record in event['Records']:
+        message_id = record['messageId']
+        try:
+            entries += [
+                (message_id, pkg_event)
+                for pkg_event in map(pkg_created_event, get_s3_events(record['body']))
+                if pkg_event is not None
+            ]
+        except Exception:
+            # unexpected message shape (S3 is the expected producer, not the only
+            # possible one) or a failure while processing one of its events: fail
+            # the message alone so it's retried and dead-lettered by itself,
+            # ending up in the DLQ instead of being deleted without a trace
+            logger.exception('failed to process message %s', message_id)
+            failed_message_ids.add(message_id)
+
+    failed_message_ids |= publish(entries)
+    # Messages not listed here are considered processed and get deleted from the
+    # queue, so an error that can't be attributed to specific messages (a failed
+    # put_events call) must raise, failing the whole batch.
+    return {
+        'batchItemFailures': [
+            {'itemIdentifier': message_id} for message_id in failed_message_ids
+        ],
+    }
diff --git a/lambdas/pkgevents/tests/test_index.py b/lambdas/pkgevents/tests/test_index.py
@@ -7,10 +7,10 @@
 from botocore.stub import Stubber
 
 from t4_lambda_pkgevents import (
-    EventsQueue,
-    PutEventsException,
+    PUT_EVENTS_MAX_ENTRIES,
     handler,
     pkg_created_event,
+    publish,
     s3,
 )
 
@@ -146,44 +146,161 @@ def test_pkg_created_event(pointer_file):
         stubber.assert_no_pending_responses()
 
 
-@mock.patch("t4_lambda_pkgevents.EventsQueue.flush")
-@mock.patch("t4_lambda_pkgevents.EventsQueue.append")
+@mock.patch("t4_lambda_pkgevents.publish", return_value=frozenset())
 @mock.patch("t4_lambda_pkgevents.pkg_created_event", wraps=str)
-def test_handler(pkg_created_event_mock, queue_append_mock, queue_flush_mock):
+def test_handler(pkg_created_event_mock, publish_mock):
     event = {
         'Records': [
             {
+                'messageId': f'message-{idx}',
                 'body': json.dumps(
                     {
                         'Records': records
                     }
                 )
             }
-            for records in (
-                (0, 1),
-                (2, 3, 4),
-                (5,)
+            for idx, records in enumerate(
+                (
+                    (0, 1),
+                    (2, 3, 4),
+                    (5,)
+                )
             )
         ]
     }
-    handler(event, None)
+    assert handler(event, None) == {'batchItemFailures': []}
     assert pkg_created_event_mock.call_args_list == [((x,),) for x in range(6)]
-    assert queue_append_mock.call_args_list == [((str(x),),) for x in range(6)]
-    queue_flush_mock.assert_called_once_with()
+    publish_mock.assert_called_once_with(
+        [
+            (f'message-{idx}', str(x))
+            for x, idx in zip(range(6), (0, 0, 1, 1, 1, 2), strict=True)
+        ]
+    )
+
+
+@mock.patch("t4_lambda_pkgevents.logger")
+@mock.patch("t4_lambda_pkgevents.publish", return_value=frozenset())
+@mock.patch("t4_lambda_pkgevents.pkg_created_event", wraps=str)
+def test_handler_drops_test_events_silently(
+    pkg_created_event_mock, publish_mock, logger_mock
+):
+    test_event_body = json.dumps(
+        {
+            'Service': 'Amazon S3',
+            'Event': 's3:TestEvent',
+            'Time': '2026-07-30T00:38:18.000Z',
+            'Bucket': 'test-bucket',
+            'RequestId': 'AAAAAAAAAAAAAAAA',
+            'HostId': 'aaaaaaaaaaaaaaaa',
+        }
+    )
+    event = {
+        'Records': [
+            {'messageId': 'message-0', 'body': json.dumps({'Records': (0, 1)})},
+            {'messageId': 'message-1', 'body': test_event_body},
+            {'messageId': 'message-2', 'body': json.dumps({'Records': (2,)})},
+        ]
+    }
+    assert handler(event, None) == {'batchItemFailures': []}
+    assert pkg_created_event_mock.call_args_list == [((x,),) for x in range(3)]
+    publish_mock.assert_called_once_with(
+        [
+            (message_id, str(x))
+            for x, message_id in zip(range(3), ('message-0', 'message-0', 'message-2'), strict=True)
+        ]
+    )
+    # a test event arrives on every bucket add, it must not be logged
+    logger_mock.warning.assert_not_called()
+    logger_mock.exception.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    'bad_body',
+    (
+        json.dumps({'Event': 'unexpected'}),
+        'not JSON',
+        'null',
+        '[]',
+        '"s3:TestEvent"',
+    ),
+)
+@mock.patch("t4_lambda_pkgevents.publish", return_value=frozenset())
+@mock.patch("t4_lambda_pkgevents.pkg_created_event", wraps=str)
+def test_handler_reports_messages_without_records_as_failed(
+    pkg_created_event_mock, publish_mock, bad_body
+):
+    event = {
+        'Records': [
+            {'messageId': 'message-0', 'body': json.dumps({'Records': (0,)})},
+            {'messageId': 'message-1', 'body': bad_body},
+        ]
+    }
+    assert handler(event, None) == {'batchItemFailures': [{'itemIdentifier': 'message-1'}]}
+    assert pkg_created_event_mock.call_args_list == [((0,),)]
+    publish_mock.assert_called_once_with([('message-0', '0')])
 
 
-@pytest.mark.parametrize('failed_count', (0, 1))
-def test_queue(failed_count):
+@mock.patch("t4_lambda_pkgevents.publish", return_value=frozenset())
+@mock.patch("t4_lambda_pkgevents.pkg_created_event")
+def test_handler_reports_message_with_failing_event(
+    pkg_created_event_mock, publish_mock
+):
+    pkg_created_event_mock.side_effect = (None, Exception('boom'), None)
+    event = {
+        'Records': [
+            # the event after the failing one must not be processed
+            {'messageId': 'message-0', 'body': json.dumps({'Records': (0, 1, 2)})},
+            {'messageId': 'message-1', 'body': json.dumps({'Records': (3,)})},
+        ]
+    }
+    assert handler(event, None) == {'batchItemFailures': [{'itemIdentifier': 'message-0'}]}
+    assert pkg_created_event_mock.call_args_list == [((x,),) for x in (0, 1, 3)]
+    publish_mock.assert_called_once_with([])
+
+
+@mock.patch("t4_lambda_pkgevents.publish", return_value=frozenset({'message-0'}))
+@mock.patch("t4_lambda_pkgevents.pkg_created_event", wraps=str)
+def test_handler_reports_publish_failures(
+    pkg_created_event_mock, publish_mock
+):
+    event = {
+        'Records': [
+            {'messageId': 'message-0', 'body': json.dumps({'Records': (0,)})},
+        ]
+    }
+    assert handler(event, None) == {'batchItemFailures': [{'itemIdentifier': 'message-0'}]}
+
+
+def test_publish_success():
     with mock.patch("t4_lambda_pkgevents.event_bridge.put_events") as put_events_mock:
-        put_events_mock.return_value = {'FailedEntryCount': failed_count}
-        q = EventsQueue()
-        for x in range(EventsQueue.MAX_SIZE - 1):
-            q.append(x)
-            put_events_mock.assert_not_called()
-
-        if failed_count:
-            with pytest.raises(PutEventsException):
-                q.append(EventsQueue.MAX_SIZE - 1)
-        else:
-            q.append(EventsQueue.MAX_SIZE - 1)
-            put_events_mock.assert_called_once_with(Entries=list(range(EventsQueue.MAX_SIZE)))
+        put_events_mock.return_value = {'FailedEntryCount': 0}
+        entries = [(f'message-{x}', x) for x in range(PUT_EVENTS_MAX_ENTRIES + 1)]
+
+        assert publish(entries) == set()
+        assert put_events_mock.call_args_list == [
+            mock.call(Entries=list(range(PUT_EVENTS_MAX_ENTRIES))),
+            mock.call(Entries=[PUT_EVENTS_MAX_ENTRIES]),
+        ]
+
+
+def test_publish_nothing():
+    with mock.patch("t4_lambda_pkgevents.event_bridge.put_events") as put_events_mock:
+        assert publish([]) == set()
+        put_events_mock.assert_not_called()
+
+
+def test_publish_failed_entries():
+    with mock.patch("t4_lambda_pkgevents.event_bridge.put_events") as put_events_mock:
+        first_entries = [{'EventId': str(x)} for x in range(PUT_EVENTS_MAX_ENTRIES)]
+        first_entries[3] = {'ErrorCode': 'ThrottlingException', 'ErrorMessage': 'try later'}
+        put_events_mock.side_effect = (
+            {'FailedEntryCount': 1, 'Entries': first_entries},
+            {
+                'FailedEntryCount': 1,
+                'Entries': [{'ErrorCode': 'InternalFailure', 'ErrorMessage': 'oops'}],
+            },
+        )
+        entries = [(f'message-{x}', x) for x in range(PUT_EVENTS_MAX_ENTRIES + 1)]
+
+        assert publish(entries) == {'message-3', f'message-{PUT_EVENTS_MAX_ENTRIES}'}
+        assert put_events_mock.call_count == 2
