fix: Pass connect_timeout to the streaming SSE client

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,

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,

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

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|

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")

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

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) }
 

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 }