diff --git a/airflow/jobs/triggerer_job.py b/airflow/jobs/triggerer_job.py
index 1c8642986cb81..4277b5f2ee4ee 100644
--- a/airflow/jobs/triggerer_job.py
+++ b/airflow/jobs/triggerer_job.py
@@ -125,6 +125,7 @@ def _run_trigger_loop(self) -> None:
         while not self.runner.stop:
             # Clean out unused triggers
             Trigger.clean_unused()
+            self.purge_completed_triggers_list()
             # Load/delete triggers
             self.load_triggers()
             # Handle events
@@ -148,6 +149,25 @@ def load_triggers(self):
         ids = Trigger.ids_for_triggerer(self.id)
         self.runner.update_triggers(set(ids))
 
+    @provide_session
+    def purge_completed_triggers_list(self, session):
+        """
+        The runner will wait to remove a trigger from its ``triggers`` dictionary until it is
+        purged from the ``completed_triggers`` list.
+        Once a trigger is deleted from the database, there's no chance it will be created again
+        and we can remove it.
+        """
+        if self.runner.completed_triggers:
+            ids = {
+                trigger_id
+                for (trigger_id,) in session.query(Trigger.id)
+                .filter(Trigger.id.in_(self.runner.completed_triggers))
+                .all()
+            }
+            purge_ids = self.runner.completed_triggers.difference(ids)
+            if purge_ids:
+                self.runner.completed_triggers.difference_update(purge_ids)
+
     def handle_events(self):
         """
         Handles outbound events from triggers - dispatching them into the Trigger
@@ -213,12 +233,16 @@ class TriggerRunner(threading.Thread, LoggingMixin):
     # Outbound queue of failed triggers
     failed_triggers: Deque[int]
 
+    # Waiting area after triggers have completed but before they've been deleted from the database
+    completed_triggers: Set[int]
+
     # Should-we-stop flag
     stop: bool = False
 
     def __init__(self):
         super().__init__()
         self.triggers = {}
+        self.completed_triggers = set()
         self.trigger_cache = {}
         self.to_create = deque()
         self.to_delete = deque()
@@ -315,6 +339,7 @@ async def cleanup_finished_triggers(self):
                         details["name"],
                     )
                     self.failed_triggers.append(trigger_id)
+                self.completed_triggers.add(trigger_id)
                 del self.triggers[trigger_id]
             await asyncio.sleep(0)
 
@@ -399,6 +424,9 @@ def update_triggers(self, requested_trigger_ids: Set[int]):
                 # Either the trigger code or the path to it is bad. Fail the trigger.
                 self.failed_triggers.append(new_id)
                 continue
+            # the trigger recently completed and we should not create it again
+            if new_id in self.completed_triggers:
+                continue
             self.to_create.append((new_id, trigger_class(**new_triggers[new_id].kwargs)))
         # Remove old triggers
         for old_id in old_trigger_ids: