From fde36aebbd866606ac1cf0f9397677792e4a9c6d Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 2 Oct 2026 09:54:04 -0400 Subject: [PATCH 1/4] test: Prove the stream ignores connect_timeout Neither streaming data source passes a connect timeout to SSE::Client. The FDv1 stream ignores Config#connect_timeout, and the FDv2 stream ignores the builder's connect_timeout. Stream connects therefore use the ld-eventsource default of 10 seconds instead of the configured value (default 2 seconds). Polling and events already pass it. A connect that hangs then costs a new client 10 seconds before its first retry, which exceeds a 10-second initialization wait and makes the application serve fallback values. These specs fail until the streams pass the option. Later commits fix both streams and add specs that use a socket factory that never connects, so the HTTP client has to enforce the timeout. Those specs also cover polling and events, and they run on JRuby. --- spec/impl/data_source/stream_spec.rb | 25 ++++++++++++++++ .../streaming_synchronizer_spec.rb | 29 +++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/spec/impl/data_source/stream_spec.rb b/spec/impl/data_source/stream_spec.rb index 12080697..1ca5018e 100644 --- a/spec/impl/data_source/stream_spec.rb +++ b/spec/impl/data_source/stream_spec.rb @@ -101,6 +101,31 @@ def with_connect_handler(processor) end end + describe 'SSE client options' do + def captured_sse_options(processor) + captured = nil + allow(SSE::Client).to receive(:new) do |_uri, **opts| + captured = opts + double("SSE::Client", close: nil) + end + + processor.start + processor.stop + captured + end + + it 'passes the default connect timeout' do + opts = captured_sse_options(processor) + expect(opts[:connect_timeout]).to eq(Config.default_connect_timeout) + end + + it 'passes a configured connect timeout' do + config = Config.new(connect_timeout: 0.5) + opts = captured_sse_options(subject.new("sdk_key", config)) + expect(opts[:connect_timeout]).to eq(0.5) + end + end + describe '#log_connection_result' do it "logs successful connection when diagnostic_accumulator is provided" do diagnostic_accumulator = double("DiagnosticAccumulator") diff --git a/spec/impl/data_system/streaming_synchronizer_spec.rb b/spec/impl/data_system/streaming_synchronizer_spec.rb index b1ff865a..9b0951fe 100644 --- a/spec/impl/data_system/streaming_synchronizer_spec.rb +++ b/spec/impl/data_system/streaming_synchronizer_spec.rb @@ -375,6 +375,35 @@ def initialize(type, data = nil) end end + describe 'SSE client options' do + let(:config) { LaunchDarkly::Config.new(logger: $null_log) } + + def captured_sse_options(synchronizer) + captured = nil + allow(SSE::Client).to receive(:new) do |_uri, **opts| + captured = opts + double("SSE::Client", close: nil) + end + + # Stop first so that sync returns as soon as it creates the client. + synchronizer.stop + synchronizer.sync(double("SelectorStore")) + captured + end + + it "passes the default connect timeout" do + synchronizer = LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new.build(sdk_key, config) + opts = captured_sse_options(synchronizer) + expect(opts[:connect_timeout]).to eq(HttpConfigOptions::DEFAULT_CONNECT_TIMEOUT) + end + + it "passes a configured connect timeout" do + synchronizer = LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new.connect_timeout(0.5).build(sdk_key, config) + opts = captured_sse_options(synchronizer) + expect(opts[:connect_timeout]).to eq(0.5) + end + end + describe 'diagnostic event recording' do let(:synchronizer) { LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new.build(sdk_key, config) } From 626b9e333042f46d01a776ae38dfa22f73c49433 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 2 Oct 2026 09:54:47 -0400 Subject: [PATCH 2/4] fix: Pass connect_timeout to the streaming SSE client --- lib/ldclient-rb/impl/data_source/stream.rb | 1 + lib/ldclient-rb/impl/data_system/streaming.rb | 1 + 2 files changed, 2 insertions(+) diff --git a/lib/ldclient-rb/impl/data_source/stream.rb b/lib/ldclient-rb/impl/data_source/stream.rb index c7b4cb2d..3f194871 100644 --- a/lib/ldclient-rb/impl/data_source/stream.rb +++ b/lib/ldclient-rb/impl/data_source/stream.rb @@ -47,6 +47,7 @@ def start opts = { headers: headers, read_timeout: READ_TIMEOUT_SECONDS, + connect_timeout: @config.connect_timeout, logger: @config.logger, socket_factory: @config.socket_factory, reconnect_time: @config.initial_reconnect_delay, diff --git a/lib/ldclient-rb/impl/data_system/streaming.rb b/lib/ldclient-rb/impl/data_system/streaming.rb index 78c74d63..abf88617 100644 --- a/lib/ldclient-rb/impl/data_system/streaming.rb +++ b/lib/ldclient-rb/impl/data_system/streaming.rb @@ -97,6 +97,7 @@ def sync(ss) opts = { headers: headers, read_timeout: STREAM_READ_TIMEOUT, + connect_timeout: @http_config.connect_timeout, logger: @logger, socket_factory: @http_config.socket_factory, reconnect_time: @initial_reconnect_delay, From f5f432e668f01053be1995f2cc7bae0c5a9fed3e Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 2 Oct 2026 10:03:09 -0400 Subject: [PATCH 3/4] test: Prove the stream abandons a hung connect after the connect timeout --- spec/http_util.rb | 20 +++++++++++++++++ spec/impl/data_source/stream_spec.rb | 22 +++++++++++++++++++ .../streaming_synchronizer_spec.rb | 22 +++++++++++++++++++ 3 files changed, 64 insertions(+) diff --git a/spec/http_util.rb b/spec/http_util.rb index 1bd34073..76dec6b5 100644 --- a/spec/http_util.rb +++ b/spec/http_util.rb @@ -103,3 +103,23 @@ def open(uri, timeout) TCPSocket.new '127.0.0.1', @ports[uri] end end + +# A socket factory whose connect attempts never complete. Each attempt pushes +# the time it blocked onto a queue when the HTTP client's connect timeout +# interrupts it. +class HangingSocketFactory + attr_reader :blocked_durations + + def initialize + @blocked_durations = Queue.new + end + + def open(_host, _port) + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + begin + sleep + ensure + @blocked_durations << Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + end + end +end diff --git a/spec/impl/data_source/stream_spec.rb b/spec/impl/data_source/stream_spec.rb index 1ca5018e..64a68565 100644 --- a/spec/impl/data_source/stream_spec.rb +++ b/spec/impl/data_source/stream_spec.rb @@ -2,6 +2,7 @@ require "ld-eventsource" require "model_builders" require "spec_helper" +require "http_util" module LaunchDarkly describe Impl::DataSource::StreamProcessor do @@ -124,6 +125,27 @@ def captured_sse_options(processor) opts = captured_sse_options(subject.new("sdk_key", config)) expect(opts[:connect_timeout]).to eq(0.5) end + + it 'abandons a connect attempt after the connect timeout' do + socket_factory = HangingSocketFactory.new + config = Config.new( + stream_uri: "http://stream.example.com", + connect_timeout: 0.2, + socket_factory: socket_factory, + logger: $null_log + ) + processor = subject.new("sdk_key", config) + + processor.start + begin + # Without the connect timeout, the attempt blocks for the SSE client default of 10 seconds. + blocked = socket_factory.blocked_durations.pop(timeout: 3) + expect(blocked).not_to be_nil + expect(blocked).to be < 1 + ensure + processor.stop + end + end end describe '#log_connection_result' do diff --git a/spec/impl/data_system/streaming_synchronizer_spec.rb b/spec/impl/data_system/streaming_synchronizer_spec.rb index 9b0951fe..9953599a 100644 --- a/spec/impl/data_system/streaming_synchronizer_spec.rb +++ b/spec/impl/data_system/streaming_synchronizer_spec.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require "spec_helper" +require "http_util" require "ldclient-rb/impl/data_system/streaming" require "ldclient-rb/interfaces" require "json" @@ -402,6 +403,27 @@ def captured_sse_options(synchronizer) opts = captured_sse_options(synchronizer) expect(opts[:connect_timeout]).to eq(0.5) end + + it "abandons a connect attempt after the connect timeout" do + socket_factory = HangingSocketFactory.new + synchronizer = LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new + .base_uri("http://stream.example.com") + .connect_timeout(0.2) + .socket_factory(socket_factory) + .build(sdk_key, config) + selector_store = double("SelectorStore", selector: nil) + + sync_thread = Thread.new { synchronizer.sync(selector_store) { |_update| } } + begin + # Without the connect timeout, the attempt blocks for the SSE client default of 10 seconds. + blocked = socket_factory.blocked_durations.pop(timeout: 3) + expect(blocked).not_to be_nil + expect(blocked).to be < 1 + ensure + synchronizer.stop + sync_thread.join(5) + end + end end describe 'diagnostic event recording' do From cdb3fbb2919d5876e95eca15966e4c095759df14 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 2 Oct 2026 10:13:43 -0400 Subject: [PATCH 4/4] test: Cover the stream client options and the polling and events connect timeouts --- spec/http_util.rb | 32 +++++++++++++---- spec/impl/data_source/requestor_spec.rb | 25 +++++++++++++ spec/impl/data_source/stream_spec.rb | 16 ++++++++- .../data_system/polling_requester_spec.rb | 35 +++++++++++++++++++ .../streaming_synchronizer_spec.rb | 21 +++++++++-- spec/impl/event_sender_spec.rb | 23 ++++++++++++ 6 files changed, 142 insertions(+), 10 deletions(-) create mode 100644 spec/impl/data_system/polling_requester_spec.rb diff --git a/spec/http_util.rb b/spec/http_util.rb index 76dec6b5..3198cdcb 100644 --- a/spec/http_util.rb +++ b/spec/http_util.rb @@ -104,14 +104,14 @@ def open(uri, timeout) end end -# A socket factory whose connect attempts never complete. Each attempt pushes -# the time it blocked onto a queue when the HTTP client's connect timeout -# interrupts it. +# A socket factory whose connect attempts never complete. The HTTP client's +# connect timeout interrupts each attempt, and the factory records how long the +# attempt blocked. class HangingSocketFactory - attr_reader :blocked_durations - def initialize - @blocked_durations = Queue.new + @lock = Mutex.new + @interrupted = ConditionVariable.new + @blocked_durations = [] end def open(_host, _port) @@ -119,7 +119,25 @@ def open(_host, _port) begin sleep ensure - @blocked_durations << Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + blocked = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + @lock.synchronize do + @blocked_durations << blocked + @interrupted.broadcast + end + end + end + + # Waits for the first connect attempt to end. Returns how long that attempt + # blocked, or nil if no attempt ends within the timeout. + def first_blocked_duration(timeout) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + @lock.synchronize do + while @blocked_durations.empty? + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + return nil if remaining <= 0 + @interrupted.wait(@lock, remaining) + end + @blocked_durations.first end end end diff --git a/spec/impl/data_source/requestor_spec.rb b/spec/impl/data_source/requestor_spec.rb index 0ed7676a..4068501a 100644 --- a/spec/impl/data_source/requestor_spec.rb +++ b/spec/impl/data_source/requestor_spec.rb @@ -16,6 +16,31 @@ def with_requestor(base_uri, opts = {}) end end + it "abandons a connect attempt after the connect timeout" do + socket_factory = HangingSocketFactory.new + requestor = Impl::DataSource::Requestor.new(sdk_key, Config.new( + base_uri: "http://sdk.example.com", + connect_timeout: 0.2, + socket_factory: socket_factory, + logger: $null_log + )) + + request_thread = Thread.new do + requestor.request_all_data + rescue StandardError + # A timed-out connect raises; only the blocked duration matters here. + end + begin + # Without the connect timeout, the attempt blocks until the thread is killed. + blocked = socket_factory.first_blocked_duration(3) + expect(blocked).not_to be_nil + expect(blocked).to be < 1 + ensure + request_thread.kill unless request_thread.join(5) + requestor.stop + end + end + describe "request_all_flags", flaky: true do it "uses expected URI and headers" do with_server do |server| diff --git a/spec/impl/data_source/stream_spec.rb b/spec/impl/data_source/stream_spec.rb index 64a68565..2b7b02db 100644 --- a/spec/impl/data_source/stream_spec.rb +++ b/spec/impl/data_source/stream_spec.rb @@ -126,6 +126,20 @@ def captured_sse_options(processor) expect(opts[:connect_timeout]).to eq(0.5) end + it 'passes the configured client options' do + socket_factory = Object.new + config = Config.new(socket_factory: socket_factory, initial_reconnect_delay: 3, logger: $null_log) + opts = captured_sse_options(subject.new("sdk_key", config)) + expect(opts).to eq( + headers: Impl::Util.default_http_headers("sdk_key", config), + read_timeout: Impl::DataSource::READ_TIMEOUT_SECONDS, + connect_timeout: Config.default_connect_timeout, + logger: $null_log, + socket_factory: socket_factory, + reconnect_time: 3 + ) + end + it 'abandons a connect attempt after the connect timeout' do socket_factory = HangingSocketFactory.new config = Config.new( @@ -139,7 +153,7 @@ def captured_sse_options(processor) processor.start begin # Without the connect timeout, the attempt blocks for the SSE client default of 10 seconds. - blocked = socket_factory.blocked_durations.pop(timeout: 3) + blocked = socket_factory.first_blocked_duration(3) expect(blocked).not_to be_nil expect(blocked).to be < 1 ensure diff --git a/spec/impl/data_system/polling_requester_spec.rb b/spec/impl/data_system/polling_requester_spec.rb new file mode 100644 index 00000000..572eead1 --- /dev/null +++ b/spec/impl/data_system/polling_requester_spec.rb @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +require "spec_helper" +require "http_util" +require "ldclient-rb/impl/data_system/polling" + +module LaunchDarkly + module Impl + module DataSystem + [HTTPPollingRequester, HTTPFDv1PollingRequester].each do |requester_class| + RSpec.describe requester_class do + it "abandons a connect attempt after the connect timeout" do + socket_factory = HangingSocketFactory.new + http_config = HttpConfigOptions.new( + base_uri: "http://sdk.example.com", + socket_factory: socket_factory, + connect_timeout: 0.2 + ) + requester = requester_class.new("sdk_key", http_config, Config.new(logger: $null_log)) + + fetch_thread = Thread.new { requester.fetch(nil) } + begin + # Without the connect timeout, the attempt blocks until the thread is killed. + blocked = socket_factory.first_blocked_duration(3) + expect(blocked).not_to be_nil + expect(blocked).to be < 1 + ensure + fetch_thread.kill unless fetch_thread.join(5) + end + end + end + end + end + end +end diff --git a/spec/impl/data_system/streaming_synchronizer_spec.rb b/spec/impl/data_system/streaming_synchronizer_spec.rb index 9953599a..8ad2c02d 100644 --- a/spec/impl/data_system/streaming_synchronizer_spec.rb +++ b/spec/impl/data_system/streaming_synchronizer_spec.rb @@ -404,6 +404,23 @@ def captured_sse_options(synchronizer) expect(opts[:connect_timeout]).to eq(0.5) end + it "passes the configured client options" do + socket_factory = Object.new + synchronizer = LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new + .socket_factory(socket_factory) + .initial_reconnect_delay(3) + .build(sdk_key, config) + opts = captured_sse_options(synchronizer) + expect(opts).to eq( + headers: Impl::Util.default_http_headers(sdk_key, config), + read_timeout: STREAM_READ_TIMEOUT, + connect_timeout: HttpConfigOptions::DEFAULT_CONNECT_TIMEOUT, + logger: $null_log, + socket_factory: socket_factory, + reconnect_time: 3 + ) + end + it "abandons a connect attempt after the connect timeout" do socket_factory = HangingSocketFactory.new synchronizer = LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new @@ -411,12 +428,12 @@ def captured_sse_options(synchronizer) .connect_timeout(0.2) .socket_factory(socket_factory) .build(sdk_key, config) - selector_store = double("SelectorStore", selector: nil) + selector_store = Struct.new(:selector).new(nil) sync_thread = Thread.new { synchronizer.sync(selector_store) { |_update| } } begin # Without the connect timeout, the attempt blocks for the SSE client default of 10 seconds. - blocked = socket_factory.blocked_durations.pop(timeout: 3) + blocked = socket_factory.first_blocked_duration(3) expect(blocked).not_to be_nil expect(blocked).to be < 1 ensure diff --git a/spec/impl/event_sender_spec.rb b/spec/impl/event_sender_spec.rb index fce600b2..0e82eb21 100644 --- a/spec/impl/event_sender_spec.rb +++ b/spec/impl/event_sender_spec.rb @@ -7,6 +7,29 @@ module LaunchDarkly module Impl + describe EventSender do + it "abandons a connect attempt after the connect timeout" do + socket_factory = HangingSocketFactory.new + sender = EventSender.new("sdk_key", Config.new( + events_uri: "http://events.example.com", + connect_timeout: 0.2, + socket_factory: socket_factory, + logger: $null_log + ), 0.1) + + send_thread = Thread.new { sender.send_event_data("[]", "events", false) } + begin + # Without the connect timeout, the attempt blocks until the thread is killed. + blocked = socket_factory.first_blocked_duration(3) + expect(blocked).not_to be_nil + expect(blocked).to be < 1 + ensure + send_thread.kill unless send_thread.join(5) + sender.stop + end + end + end + describe EventSender, flaky: true do subject { EventSender }