From 67dc1459d11fabfaac964e615a80fee8e1ab8fbe Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 25 Sep 2026 15:41:41 -0500 Subject: [PATCH 1/2] fix: Stop the worker thread promptly when the client is closed during a reconnect wait --- lib/ld-eventsource/client.rb | 12 +++++- spec/client_spec.rb | 73 ++++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index 936a6f9..5111115 100644 --- a/lib/ld-eventsource/client.rb +++ b/lib/ld-eventsource/client.rb @@ -124,6 +124,7 @@ def initialize(uri, http_client_options: nil) @uri = URI(uri) @stopped = Concurrent::AtomicBoolean.new(false) + @stop_event = Concurrent::Event.new @retry_enabled = retry_enabled @headers = headers.clone @@ -279,6 +280,7 @@ def query_params(&action) # def close if @stopped.make_true + @stop_event.set reset_http end end @@ -323,7 +325,11 @@ def run_stream end # There's a potential race if close was called in the middle of the previous line, i.e. after we # connected but before @cxn was set. Checking the variable again is a bit clunky but avoids that. - return if @stopped.value + if @stopped.value + # close did not see this connection, so close it here. + reset_http + return + end read_stream(resp) unless resp.nil? rescue => e # When we deliberately close the connection, it will usually trigger an exception. The exact type @@ -352,7 +358,9 @@ def connect @first_attempt = false if interval > 0 @logger.info { "Will retry connection after #{'%.3f' % interval} seconds" } - sleep(interval) + # This wait ends early when close sets the event. + @stop_event.wait(interval) + return if @stopped.value end cxn = nil begin diff --git a/spec/client_spec.rb b/spec/client_spec.rb index e961950..acc900a 100644 --- a/spec/client_spec.rb +++ b/spec/client_spec.rb @@ -454,6 +454,79 @@ def send_stream_content(res, content, keep_open:) end end + describe "close during a reconnect wait" do + let(:long_reconnect_time) { 5 } + + def start_failing_client(server, attempts) + server.setup_response("/") do |req,res| + attempts << Time.now + res.status = 500 + res.body = "sorry" + res.keep_alive = false + end + + error_sink = Queue.new + threads_before = Thread.list + client = subject.new(server.base_uri, reconnect_time: long_reconnect_time) do |c| + c.on_error { |error| error_sink << error } + end + worker = (Thread.list - threads_before).detect { |t| t.name == 'LD/SSEClient' } + error_sink.pop # the client now waits before it tries again + [client, worker] + end + + it "stops the worker thread promptly" do + with_server do |server| + client, worker = start_failing_client(server, Queue.new) + with_client(client) do |c| + c.close + expect(worker.join(1)).not_to be_nil + end + end + end + + it "does not send another request" do + with_server do |server| + attempts = Queue.new + client, worker = start_failing_client(server, attempts) + with_client(client) do |c| + c.close + worker.join(1) + expect(attempts.size).to eq 1 + end + end + end + end + + it "closes a connection that opens while the client is closed" do + server = TCPServer.new("127.0.0.1", 0) + server_result = Queue.new + server_thread = Thread.new do + socket = server.accept + while (line = socket.gets) && line != "\r\n"; end + socket.write("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n") + readable = IO.select([socket], nil, nil, 2) + eof = readable && socket.read_nonblock(1, exception: false).nil? + server_result << (eof ? :closed : :open) + socket.close + end + + # query_params runs after the last stop check and before the request, so close cannot close this connection. + client = subject.new("http://127.0.0.1:#{server.addr[1]}") do |c| + c.query_params do + c.close + {} + end + end + + with_client(client) do + expect(server_result.pop).to eq :closed + end + ensure + server_thread&.join(1) + server&.close + end + describe "HTTP method parameter" do it "defaults to GET method" do with_server do |server| From 2433eef1f948f5de692b51162198adcbcd1f77a2 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 25 Sep 2026 15:50:30 -0500 Subject: [PATCH 2/2] fix: Use one event for the client's stop state --- lib/ld-eventsource/client.rb | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index 5111115..ace552e 100644 --- a/lib/ld-eventsource/client.rb +++ b/lib/ld-eventsource/client.rb @@ -123,7 +123,6 @@ def initialize(uri, retry_enabled: true, http_client_options: nil) @uri = URI(uri) - @stopped = Concurrent::AtomicBoolean.new(false) @stop_event = Concurrent::Event.new @retry_enabled = retry_enabled @@ -279,8 +278,7 @@ def query_params(&action) # has no effect if called a second time. # def close - if @stopped.make_true - @stop_event.set + if @stop_event.try? reset_http end end @@ -291,7 +289,7 @@ def close # @return [Boolean] true if the client has been shut down # def closed? - @stopped.value + @stop_event.set? end private @@ -316,7 +314,7 @@ def default_logger end def run_stream - until @stopped.value + until @stop_event.set? close_connection begin resp = connect @@ -325,7 +323,7 @@ def run_stream end # There's a potential race if close was called in the middle of the previous line, i.e. after we # connected but before @cxn was set. Checking the variable again is a bit clunky but avoids that. - if @stopped.value + if @stop_event.set? # close did not see this connection, so close it here. reset_http return @@ -333,8 +331,8 @@ def run_stream read_stream(resp) unless resp.nil? rescue => e # When we deliberately close the connection, it will usually trigger an exception. The exact type - # of exception depends on the specific Ruby runtime. But @stopped will always be set in this case. - if @stopped.value + # 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") @@ -353,14 +351,14 @@ def run_stream # Try to establish a streaming connection. Returns the StreamingHTTPConnection object if successful. def connect loop do - return if @stopped.value + return if @stop_event.set? interval = @first_attempt ? 0 : @backoff.next_interval @first_attempt = false if interval > 0 @logger.info { "Will retry connection after #{'%.3f' % interval} seconds" } # This wait ends early when close sets the event. @stop_event.wait(interval) - return if @stopped.value + return if @stop_event.set? end cxn = nil begin @@ -403,7 +401,7 @@ def read_stream(cxn) chunks = Enumerator.new do |gen| loop do - if @stopped.value + if @stop_event.set? break else begin @@ -425,7 +423,7 @@ def read_stream(cxn) event_parser = Impl::EventParser.new(Impl::BufferedLineReader.lines_from(chunks), @last_id) event_parser.items.each do |item| - return if @stopped.value + return if @stop_event.set? case item when StreamEvent dispatch_event(item)