fix: invert TTL refresh guard in DefaultDictWithTimeout (memory leak)

pr_agent/servers/github_app.py

@@ -200,10 +200,30 @@ async def handle_push_trigger_for_new_commits(body: Dict[str, Any],
             await _perform_auto_commands_github("push_commands", agent, body, api_url, log_context)
 
     finally:
-        # release the waiting task block
-        async with _pending_task_duplicate_push_conditions[api_url]:
-            _pending_task_duplicate_push_conditions[api_url].notify(1)
-            _duplicate_push_triggers[api_url] -= 1
+        # Release the next waiting task, then remove the shared per-PR state once
+        # no task remains. The decrement, the "is anyone left?" decision, and the
+        # removal all run under the same condition lock so a newly admitted task
+        # cannot interleave between the decision and the removal.
+        condition = _pending_task_duplicate_push_conditions[api_url]
+        async with condition:
+            condition.notify(1)
+            try:
+                _duplicate_push_triggers[api_url] -= 1
+                no_tasks_left = _duplicate_push_triggers[api_url] <= 0
+            except KeyError:
+                # The counter was already evicted (e.g. by the TTL sweep while
+                # this task was processing). Leave any stray condition entry for
+                # the TTL sweep rather than removing state a concurrent task may
+                # still rely on.
+                get_logger().debug(f"Push trigger counter already removed for {api_url=}")
+                no_tasks_left = False
+            if no_tasks_left:
+                # pop() (not del) tolerates an already-missing key and keeps
+                # DefaultDictWithTimeout's internal key-time map in sync.
+                _duplicate_push_triggers.pop(api_url, None)
+                _pending_task_duplicate_push_conditions.pop(api_url, None)
+
+    return {}
 
 
 def handle_closed_pr(body, event, action, log_context):

pr_agent/servers/utils.py

@@ -84,3 +84,10 @@ def __setitem__(self, __key, __value):
     def __delitem__(self, __key):
         del self.__key_times[__key]
         return super().__delitem__(__key)
+
+    def pop(self, __key, *args):
+        # Keep the internal key-time map in sync. Unlike __delitem__, pop must
+        # tolerate a missing key (callers rely on the ``default`` argument), so
+        # the timestamp is discarded with pop(..., None) rather than del.
+        self.__key_times.pop(__key, None)
+        return super().pop(__key, *args)

tests/unittest/test_github_app_timeout_core.py

@@ -137,6 +137,36 @@
         assert "a" not in d
         assert "a" not in _key_times(d)
 
+    def test_pop_removes_key_time(self, fake_clock):
+        d = DefaultDictWithTimeout(lambda: 0, ttl=10, refresh_interval=1000)
+        d["a"] = 1
+        assert d.pop("a") == 1
+        assert "a" not in d
+        assert "a" not in _key_times(d)
+
+    def test_pop_missing_key_with_default_does_not_raise(self, fake_clock):
+        d = DefaultDictWithTimeout(lambda: 0, ttl=10, refresh_interval=1000)
+        assert d.pop("missing", "fallback") == "fallback"
+        assert "missing" not in _key_times(d)
+
+    def test_pop_then_refresh_does_not_raise(self, fake_clock):
+        # Regression: pop() must drop the key-time too. Otherwise a later
+        # __refresh() builds `to_delete` from a stale key-time and runs
+        # `del self[key]` for a key already gone, raising KeyError.
+        d = DefaultDictWithTimeout(
+            lambda: 0, ttl=2, refresh_interval=5, update_key_time_on_get=False
+        )
+        d["a"] = 1
+        d.pop("a")
+
+        # Advance past both the TTL and the refresh interval so __refresh runs
+        # its deletion pass on the next access.
+        fake_clock["t"] += 10
+
+        # Must not raise while refreshing.
+        _ = d["fresh"]
+        assert "fresh" in d
+
     def test_refresh_runs_after_long_idle_period(self, fake_clock):
         d = DefaultDictWithTimeout(
             lambda: 0, ttl=2, refresh_interval=5, update_key_time_on_get=False
@@ -417,7 +447,7 @@
 
 
 class TestPushTriggerDedupe:
-    def test_first_event_runs_perform_and_decrements_counter(self, push_trigger_env):
+    def test_first_event_runs_perform_and_cleans_up_entries(self, push_trigger_env):
         body = _push_body()
         api_url = body["pull_request"]["url"]
 
@@ -428,8 +458,10 @@
         )
 
         assert push_trigger_env["count"] == 1
-        # Counter incremented then decremented back to 0.
-        assert github_app._duplicate_push_triggers[api_url] == 0
+        # Counter incremented to 1, decremented back to 0, then both dedupe
+        # entries are removed so the caches don't grow without bound.
+        assert api_url not in github_app._duplicate_push_triggers
+        assert api_url not in github_app._pending_task_duplicate_push_conditions
 
     def test_skips_when_before_equals_after(self, push_trigger_env):
         body = _push_body()
@@ -495,3 +527,59 @@
         )
 
         assert push_trigger_env["count"] == 0
+
+    def test_does_not_clean_up_while_another_task_active(
+        self, push_trigger_env, monkeypatch
+    ):
+        # A second push arriving mid-processing bumps the counter to 2. When the
+        # first task finishes, the counter decrements to 1 (still active), so the
+        # cleanup must be skipped and the dedupe entries preserved for the waiter.
+        body = _push_body()
+        api_url = body["pull_request"]["url"]
+
+        async def perform_then_simulate_concurrent(*args, **kwargs):
+            push_trigger_env["count"] += 1
+            github_app._duplicate_push_triggers[api_url] += 1
+
+        monkeypatch.setattr(
+            github_app, "_perform_auto_commands_github", perform_then_simulate_concurrent
+        )
+
+        asyncio.run(
+            github_app.handle_push_trigger_for_new_commits(
+                body, "push", "alice", "1", "synchronize", {}, agent=None
+            )
+        )
+
+        assert push_trigger_env["count"] == 1
+        # Counter left at 1 (the still-active concurrent task); nothing evicted.
+        assert github_app._duplicate_push_triggers[api_url] == 1
+        assert api_url in github_app._pending_task_duplicate_push_conditions
+
+    def test_cleanup_tolerates_ttl_evicted_counter(
+        self, push_trigger_env, monkeypatch
+    ):
+        # If TTL eviction removes the counter entry while the task is processing,
+        # the finally block's `-= 1` hits a missing key. The guarded KeyError must
+        # be swallowed so webhook handling does not blow up.
+        body = _push_body()
+        api_url = body["pull_request"]["url"]
+
+        async def perform_then_evict(*args, **kwargs):
+            push_trigger_env["count"] += 1
+            # Simulate TTL eviction of the counter entry mid-processing.
+            github_app._duplicate_push_triggers.pop(api_url, None)
+
+        monkeypatch.setattr(
+            github_app, "_perform_auto_commands_github", perform_then_evict
+        )
+
+        # Must complete without raising despite the missing-key decrement.
+        asyncio.run(
+            github_app.handle_push_trigger_for_new_commits(
+                body, "push", "alice", "1", "synchronize", {}, agent=None
+            )
+        )
+
+        assert push_trigger_env["count"] == 1
+        assert api_url not in github_app._duplicate_push_triggers