Plug tag_map leak in job_not_running that flaked test_mem_leak_in_event_listener

changelog/70152.fixed.md

@@ -0,0 +1 @@
+Plug the ``event_listener.tag_map`` / ``timeout_map`` leak in ``saltnado.job_not_running`` when the outer job finishes with an in-flight ``saltutil.find_job`` ping.

salt/netapi/rest_tornado/saltnado.py

@@ -1172,6 +1172,25 @@ def job_not_running(self, jid, tgt, tgt_type, minions, is_finished):
                 if f is is_finished:
                     if not event.done():
                         event.set_result(None)
+                    # ``set_result(None)`` resolves the Future but leaves
+                    # it in ``event_listener.tag_map`` / ``timeout_map``.
+                    # ``job_not_running`` runs as a ``spawn_callback``
+                    # coroutine independent of the handler; if it has
+                    # already registered a ping wait when the handler's
+                    # ``on_finish`` -> ``clean_by_request`` runs, that
+                    # entry escapes cleanup and lingers up to
+                    # ``gather_job_timeout`` seconds until the timeout
+                    # callback fires. Reap it inline so the maps drain
+                    # promptly (see ``test_mem_leak_in_event_listener``).
+                    listener = self.application.event_listener
+                    timeout_handle = listener.timeout_map.pop(event, None)
+                    if timeout_handle is not None:
+                        tornado.ioloop.IOLoop.current().remove_timeout(timeout_handle)
+                    listener._timeout_future(
+                        ping_tag,
+                        EventListener.prefix_matcher,
+                        event,
+                    )
                     raise tornado.gen.Return(True)
                 event = f.result()
             except TimeoutException: