From 0a9d9980aaf8e57889931d3bb7425c6c0c4940c7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Henrik=20Hauge=20Bj=C3=B8rnskov?=
 <19725+henrikbjorn@users.noreply.github.com>
Date: Sun, 19 Jul 2026 12:48:42 +0200
Subject: [PATCH] Fix event-task leak with a counter + drain promise
 (WaitGroup)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Alternative to the Set-based fix: instead of tracking in-flight event-hook
tasks in a collection, keep only a count (@event_tasks) plus a one-shot drain
Promise. Each event increments before spawning and decrements in its ensure;
connection_closed sets @closing, and the last hook to finish (or connection_closed
itself, if already idle) resolves the promise, which the drain awaits.

Retention is a single integer for the whole life of the connection — nothing
per-event is stored — vs the old Async::Barrier, which retained one task node
per event until #wait (called only on disconnect), leaking ~GBs over days on a
long-lived inbound connection subscribed to many events.

vs the Set approach: no collection to snapshot/iterate on drain, and no
first-error truncation of the drain loop. Same regression test (in-flight count
returns to zero). Full suite green.
---
 lib/librevox/listener/base.rb                 | 27 ++++++++++++++++---
 .../functional/librevox/listener/base_test.rb | 17 ++++++++++++
 2 files changed, 40 insertions(+), 4 deletions(-)

diff --git a/lib/librevox/listener/base.rb b/lib/librevox/listener/base.rb
index ab06335..6722fb7 100644
--- a/lib/librevox/listener/base.rb
+++ b/lib/librevox/listener/base.rb
@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 
-require 'async/barrier'
+require 'async/promise'
 require 'async/semaphore'
 require 'securerandom'
 
@@ -21,7 +21,15 @@ def initialize(connection)
         @connection = connection
         @reply_promises = []
         @app_promises = {}
-        @event_barrier = Async::Barrier.new
+        # Event hooks run in fire-and-forget tasks. We only need to *count* the
+        # in-flight ones (so a disconnect can drain them) — not hold references to
+        # them. A count is one integer regardless of how many events pass through;
+        # an Async::Barrier, by contrast, retains a node per task until #wait, so
+        # it grew without bound on a long-lived connection (one node per event for
+        # the life of the socket — a slow memory leak).
+        @event_tasks = 0
+        @drained = Async::Promise.new
+        @closing = false
         @write_lock = Async::Semaphore.new(1)
       end
 
@@ -92,9 +100,14 @@ def receive_message(response)
             @app_promises.delete(app_uuid)&.resolve(response)
           end
 
-          @event_barrier.async do
+          @event_tasks += 1
+          Async do
             on_event(response)
             invoke_event_hooks(response)
+          ensure
+            @event_tasks -= 1
+            # The last hook to finish during a drain rings the bell.
+            @drained.resolve(true) if @closing && @event_tasks.zero?
           end
         end
       end
@@ -108,7 +121,13 @@ def connection_closed
         @reply_promises.clear
         @app_promises.clear
 
-        @event_barrier.wait
+        # Drain in-flight event hooks: wait until the count hits zero. If nothing
+        # is in flight, resolve immediately so #wait returns at once; otherwise the
+        # last hook's ensure resolves it. Promise (not Condition) so a resolve that
+        # races ahead of the wait is still delivered.
+        @closing = true
+        @drained.resolve(true) if @event_tasks.zero?
+        @drained.wait
       rescue ConnectionError
         # Expected — event hooks may have been mid-command when disconnected
       end
diff --git a/test/functional/librevox/listener/base_test.rb b/test/functional/librevox/listener/base_test.rb
index 7892439..bb23568 100644
--- a/test/functional/librevox/listener/base_test.rb
+++ b/test/functional/librevox/listener/base_test.rb
@@ -26,4 +26,21 @@ def test_on_event_with_api_does_not_block_handle_response
 
     assert completed, "Deadlock: on_event with api command blocked handle_response"
   end
+
+  # Regression: a long-lived inbound connection dispatches one event-handler task
+  # per ESL event. These are fire-and-forget, so once finished nothing about them
+  # must be retained. Previously they were spawned on a per-connection
+  # Async::Barrier, whose task list only shrinks on #wait (called just once, on
+  # disconnect) — so every event leaked a task node for the life of the socket
+  # (~GBs over days on a busy switch). The in-flight count must return to zero.
+  def test_completed_event_tasks_are_not_retained
+    Sync do
+      100.times { event "SOME_EVENT" }
+
+      in_flight = @listener.instance_variable_get(:@event_tasks)
+      assert_equal 0, in_flight,
+        "in-flight event-task count must return to zero after events complete " \
+        "(got #{in_flight} still counted after 100 events)"
+    end
+  end
 end