4.55.0

app/events/event_source.py

@@ -85,24 +85,28 @@ class DeadLetterEventSource(EventSource):
     def __init__(self, max_retries: int):
         self.__max_retries = max_retries
 
+    def execute_loop(
+        self, on_event: Callable[[SyncEvent], NoReturn]
+    ) -> list[SyncEvent]:
+        threshold = arrow.utcnow().shift(minutes=-_DEAD_LETTER_THRESHOLD_MINUTES)
+        events = SyncEvent.get_dead_letter(
+            older_than=threshold, max_retries=self.__max_retries
+        )
+        if events:
+            LOG.info(f"Got {len(events)} dead letter events")
+            newrelic.agent.record_custom_metric(
+                "Custom/dead_letter_events_to_process", len(events)
+            )
+            for event in events:
+                if event.mark_as_taken(allow_taken_older_than=threshold):
+                    on_event(event)
+        return events
+
     @newrelic.agent.background_task()
     def run(self, on_event: Callable[[SyncEvent], NoReturn]):
         while True:
             try:
-                threshold = arrow.utcnow().shift(
-                    minutes=-_DEAD_LETTER_THRESHOLD_MINUTES
-                )
-                events = SyncEvent.get_dead_letter(
-                    older_than=threshold, max_retries=self.__max_retries
-                )
-                if events:
-                    LOG.info(f"Got {len(events)} dead letter events")
-                    if events:
-                        newrelic.agent.record_custom_metric(
-                            "Custom/dead_letter_events_to_process", len(events)
-                        )
-                    for event in events:
-                        on_event(event)
+                events = self.execute_loop(on_event)
                 Session.close()  # Ensure that we have a new connection and we don't have a dangling tx with a lock
                 if not events:
                     LOG.debug("No dead letter events")