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, diff --git a/spec/http_util.rb b/spec/http_util.rb index 1bd34073..3198cdcb 100644 --- a/spec/http_util.rb +++ b/spec/http_util.rb @@ -103,3 +103,41 @@ def open(uri, timeout) TCPSocket.new '127.0.0.1', @ports[uri] end end + +# 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 + def initialize + @lock = Mutex.new + @interrupted = ConditionVariable.new + @blocked_durations = [] + end + + def open(_host, _port) + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + begin + sleep + ensure + 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 12080697..2b7b02db 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 @@ -101,6 +102,66 @@ 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 + + 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( + 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.first_blocked_duration(3) + expect(blocked).not_to be_nil + expect(blocked).to be < 1 + ensure + processor.stop + end + 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/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 b1ff865a..8ad2c02d 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" @@ -375,6 +376,73 @@ 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 + + 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 + .base_uri("http://stream.example.com") + .connect_timeout(0.2) + .socket_factory(socket_factory) + .build(sdk_key, config) + 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.first_blocked_duration(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 let(:synchronizer) { LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new.build(sdk_key, config) } 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 }