diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index ace552e..cda745f 100644 --- a/lib/ld-eventsource/client.rb +++ b/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', diff --git a/lib/ld-eventsource/impl/backoff.rb b/lib/ld-eventsource/impl/backoff.rb index a540bf3..2cd7e1e 100644 --- a/lib/ld-eventsource/impl/backoff.rb +++ b/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 # diff --git a/spec/backoff_spec.rb b/spec/backoff_spec.rb index e3b81da..03e9066 100644 --- a/spec/backoff_spec.rb +++ b/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 diff --git a/spec/client_spec.rb b/spec/client_spec.rb index acc900a..be9f11f 100644 --- a/spec/client_spec.rb +++ b/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