fix: Close the connection before reporting a stream error, and keep the reconnect loop alive

lib/ld-eventsource/client.rb

@@ -315,7 +315,8 @@ def default_logger
 
     def run_stream
       until @stop_event.set?
-        close_connection
+        log_close_errors { close_connection }
+        error = nil
         begin
           resp = connect
           @lock.synchronize do
@@ -330,25 +331,25 @@ def run_stream
           end
           read_stream(resp) unless resp.nil?
         rescue => e
+          error = e
+        end
+        # Close the connection before the error handler runs, because the handler can block for a long time.
+        log_close_errors { reset_http }
+        unless error.nil?
           # When we deliberately close the connection, it will usually trigger an exception. The exact type
           # of exception depends on the specific Ruby runtime. But the client will always be closed in this case.
           if @stop_event.set?
             @logger.info { "Stream connection closed" }
           else
-            log_and_dispatch_error(e, "Unexpected error from event source")
+            log_and_dispatch_error(error, "Unexpected error from event source")
           end
         end
-        begin
-          reset_http
-        rescue StandardError => e
-          log_and_dispatch_error(e, "Unexpected error while closing stream")
-        end
 
         return close unless @retry_enabled
       end
     end
 
-    # Try to establish a streaming connection. Returns the StreamingHTTPConnection object if successful.
+    # Try to establish a streaming connection. Returns the HTTP::Response object if successful.
     def connect
       loop do
         return if @stop_event.set?
@@ -392,8 +393,7 @@ def connect
       end
     end
 
-    # Pipe the output of the StreamingHTTPConnection into the EventParser, and dispatch events as
-    # they arrive.
+    # Pipe the output of the HTTP::Response into the EventParser, and dispatch events as they arrive.
     def read_stream(cxn)
       # Tell the Backoff object that the connection is now in a valid state. It uses that information so
       # it can automatically reset itself if enough time passes between failures.
@@ -453,6 +453,15 @@ def log_and_dispatch_error(e, message)
       end
     end
 
+    # A failure to close is logged but not dispatched, so each connection attempt gives the error
+    # handler at most one error, and the worker thread continues.
+    def log_close_errors
+      yield
+    rescue StandardError => e
+      @logger.warn { "Unexpected error while closing stream: #{e.inspect}" }
+      @logger.debug { "Exception trace: #{e.backtrace}" }
+    end
+
     def build_headers
       h = {
         'Accept' => 'text/event-stream',

lib/ld-eventsource/impl/backoff.rb

@@ -39,13 +39,13 @@ def next_interval
           @attempts = 0 if good_duration >= @reconnect_reset_interval
         end
         @last_good_time = nil
+        # A zero base must not be multiplied: 0.0 * Infinity is NaN. Also, rand(0) is not permitted.
+        return 0 if @base_interval <= 0 || @max_interval <= 0
+
         target = [@base_interval * (2 ** @attempts), @max_interval].min.to_f
-        @attempts += 1
-        if target == 0
-          0  # in some Ruby versions it's illegal to call rand(0)
-        else
-          (target / 2) + @jitter_rand.rand(target / 2)
-        end
+        # Stop at the ceiling, so that the power of 2 cannot grow without limit.
+        @attempts += 1 if target < @max_interval
+        (target / 2) + @jitter_rand.rand(target / 2)
       end
 
       #

spec/backoff_spec.rb

@@ -58,6 +58,50 @@ module Impl
           expect(interval).to eq(0)
         end
       end
+
+      it "always returns zero if the initial delay is 0.0" do
+        b = Backoff.new(0.0, 60)
+
+        6.times do
+          expect(b.next_interval).to eq(0)
+        end
+      end
+
+      it "doubles the delay for each attempt until it reaches the maximum" do
+        max = 30
+        b = Backoff.new(1, max)
+
+        [1, 2, 4, 8, 16, 30, 30, 30].each do |target|
+          interval = b.next_interval
+          expect(interval).to be >= target / 2.0
+          expect(interval).to be <= target
+        end
+      end
+
+      it "stops doubling the delay after it reaches the maximum" do
+        b = Backoff.new(1, 4)
+        100.times { b.next_interval }
+
+        # The delay doubled twice to reach the maximum of 4. If the doubling had continued, a
+        # smaller initial delay would still give the maximum.
+        b.base_interval = 0.5
+        interval = b.next_interval
+        expect(interval).to be >= 1
+        expect(interval).to be < 2
+      end
+
+      [0.0, 1.0].each do |initial|
+        it "returns a finite interval after many attempts with a Float initial delay of #{initial}" do
+          max = 30
+          b = Backoff.new(initial, max)
+
+          1100.times do
+            interval = b.next_interval
+            expect(interval).to be_finite
+            expect(interval).to be_between(0, max)
+          end
+        end
+      end
     end
   end
 end

spec/client_spec.rb

@@ -1622,4 +1622,91 @@ def test_object.to_s
       end
     end
   end
+
+  describe "connection cleanup after an error" do
+    def pop_with_timeout(queue)
+      Timeout.timeout(5) { queue.pop }
+    end
+
+    it "closes the connection before it calls the error handler" do
+      steps = Queue.new
+      allow_any_instance_of(HTTP::Connection).to receive(:close).and_wrap_original do |m, *args|
+        steps << :connection_closed
+        m.call(*args)
+      end
+
+      with_server do |server|
+        server.setup_response("/") do |req,res|
+          send_stream_content(res, simple_event_1_text, keep_open: true)
+        end
+
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap) do |c|
+          c.on_event { |event| raise "event handler failure" }
+          c.on_error { |error| steps << :error_dispatched }
+        end
+
+        with_client(client) do |c|
+          expect(pop_with_timeout(steps)).to eq(:connection_closed)
+          step = pop_with_timeout(steps) until step == :error_dispatched
+        end
+      end
+    end
+
+    it "does not pass a second error to the error handler when closing the connection fails" do
+      close_calls = 0
+      allow_any_instance_of(SSE::Client).to receive(:reset_http).and_wrap_original do |m, *args|
+        close_calls += 1
+        raise IOError, "close failure" if close_calls == 1
+        m.call(*args)
+      end
+
+      with_server do |server|
+        attempt = 0
+        server.setup_response("/") do |req,res|
+          attempt += 1
+          send_stream_content(res, attempt == 1 ? simple_event_1_text : simple_event_2_text,
+            keep_open: attempt == 2)
+        end
+
+        event_sink = Queue.new
+        error_sink = Queue.new
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap) do |c|
+          c.on_event { |event| event_sink << event }
+          c.on_error { |error| error_sink << error }
+        end
+
+        with_client(client) do |c|
+          expect(pop_with_timeout(event_sink)).to eq(simple_event_1)
+          expect(pop_with_timeout(event_sink)).to eq(simple_event_2)
+          expect(close_calls).to be >= 1
+          expect(error_sink.size).to eq 1
+          expect(error_sink.pop).to be_a(SSE::Errors::StreamClosedByServerError)
+        end
+      end
+    end
+
+    it "keeps the worker thread alive when closing the previous connection fails" do
+      close_calls = 0
+      allow_any_instance_of(SSE::Client).to receive(:close_connection).and_wrap_original do |m, *args|
+        close_calls += 1
+        raise IOError, "close failure" if close_calls == 1
+        m.call(*args)
+      end
+
+      with_server do |server|
+        server.setup_response("/") do |req,res|
+          send_stream_content(res, simple_event_1_text, keep_open: true)
+        end
+
+        event_sink = Queue.new
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap) do |c|
+          c.on_event { |event| event_sink << event }
+        end
+
+        with_client(client) do |c|
+          expect(pop_with_timeout(event_sink)).to eq(simple_event_1)
+        end
+      end
+    end
+  end
 end