Merge pull request #629 from splitio/development

Release 8.11.1

Author: chillaq

CHANGES.txt

@@ -1,5 +1,10 @@
 CHANGES
 
+8.11.1 (Jul, 23, 2026)
+- Added a retry mechanism in case the initial SSE socket connection failed, and extended the list of retryable errors when socket exceptions are caught during streaming.
+- Fixed debug logging in rule-based segment matcher class when debug_enabled parameter is not set in configuration .
+- Fixed a debug log that caused exception when streaming socket closes.
+
 8.11.0 (Mar, 12, 2026)
 - Added the ability to listen to different events triggered by the SDK. Read more in our docs.
    - SDK_UPDATE notify when a flag or user segment has changed

lib/splitclient-rb/engine/matchers/rule_based_segment_matcher.rb

@@ -40,7 +40,7 @@ def match?(args)
         break if matched
       end
 
-      @logger.debug("[InRuleSegmentMatcher] #{@segment_name} is in rule based segment -> #{matched}")
+      @config.logger.debug("[InRuleSegmentMatcher] #{@segment_name} is in rule based segment -> #{matched}") if @config.debug_enabled
       matched
     end
 

lib/splitclient-rb/engine/push_manager.rb

@@ -16,17 +16,21 @@ def start_sse
         response = @auth_api_client.authenticate(@api_key)
         @config.logger.debug("Auth service response push_enabled: #{response[:push_enabled]}") if @config.debug_enabled
 
-        if response[:push_enabled] && @sse_handler.start(response[:token], response[:channels])
-          schedule_next_token_refresh(response[:exp])
-          @back_off.reset
-          record_telemetry(response[:exp])
+        unless response[:push_enabled]
+          schedule_next_token_refresh(@back_off.interval) if response[:retry]
+          return false
+        end
 
-          return true
+        unless @sse_handler.start(response[:token], response[:channels])
+          @config.logger.debug('Streaming server returned error') if @config.debug_enabled
+          stop_sse
+          return false
         end
 
-        stop_sse
-        schedule_next_token_refresh(@back_off.interval) if response[:retry]
-        false
+        schedule_next_token_refresh(response[:exp])
+        @back_off.reset
+        record_telemetry(response[:exp])
+        true
       rescue StandardError => e
         @config.logger.error("start_sse: #{e.inspect}")
       end

lib/splitclient-rb/engine/sync_manager.rb

@@ -6,14 +6,8 @@ class SyncManager
       SYNC_MODE_STREAMING = 0
       SYNC_MODE_POLLING = 1
 
-      def initialize(config,
-                     synchronizer,
-                     telemetry_runtime_producer,
-                     telemetry_synchronizer,
-                     status_manager,
-                     sse_handler,
-                     push_manager,
-                     status_queue)
+      def initialize(config, synchronizer, telemetry_runtime_producer, telemetry_synchronizer,
+                     status_manager, sse_handler, push_manager, status_queue)
         @config = config
         @synchronizer = synchronizer
         @telemetry_runtime_producer = telemetry_runtime_producer
@@ -23,6 +17,7 @@ def initialize(config,
         @push_manager = push_manager
         @status_queue = status_queue
         @sse_connected = Concurrent::AtomicBoolean.new(false)
+        @back_off = Engine::BackOff.new(1, 3)
       end
 
       def start
@@ -119,7 +114,7 @@ def process_forced_stop
       end
 
       def process_disconnect(reconnect)
-        unless @sse_connected.value
+        unless @sse_connected.value || reconnect
           @config.logger.debug('Streaming already disconnected.') if @config.debug_enabled
           return
         end
@@ -130,6 +125,9 @@ def process_disconnect(reconnect)
         record_telemetry(Telemetry::Domain::Constants::SYNC_MODE, SYNC_MODE_POLLING)
 
         if reconnect
+          wait_interval = @back_off.interval
+          @config.logger.debug("Retrying streaming connection in: #{wait_interval} seconds")
+          sleep(wait_interval)
           @push_manager.stop_sse
           @synchronizer.sync_all
           @push_manager.start_sse
@@ -155,6 +153,7 @@ def incoming_push_status_handler
 
           case status
           when Constants::PUSH_CONNECTED
+            @back_off.reset
             process_connected
           when Constants::PUSH_RETRYABLE_ERROR
             process_disconnect(true)

lib/splitclient-rb/engine/synchronizer.rb

@@ -55,8 +55,8 @@ def start_periodic_data_recording
       end
 
       def start_periodic_fetch
-        @split_fetcher.call
-        @segment_fetcher.call
+        @split_fetcher.call unless Helpers::ThreadHelper.alive?(:split_fetcher, @config)
+        @segment_fetcher.call unless Helpers::ThreadHelper.alive?(:segment_fetcher, @config)
       end
 
       def stop_periodic_fetch

lib/splitclient-rb/sse/event_source/client.rb

@@ -10,7 +10,7 @@ module SSE
     module EventSource
       class Client
         DEFAULT_READ_TIMEOUT = 70
-        CONNECT_TIMEOUT = 30_000
+        CONNECT_TIMEOUT = 30
         OK_CODE = 200
         KEEP_ALIVE_RESPONSE = "c\r\n:keepalive\n\n\r\n".freeze
         ERROR_EVENT_TYPE = 'error'.freeze
@@ -37,17 +37,15 @@ def initialize(config,
         end
 
         def close(status = nil)
-          unless connected?
-            @config.logger.debug('SSEClient already disconected.') if @config.debug_enabled
-            return
-          end
-          @config.logger.debug("Closing SSEClient socket") if @config.debug_enabled
+          return if @socket.nil?
 
+          @config.logger.debug("Closing SSEClient socket") if @config.debug_enabled
           push_status(status)
           @connected.make_false
           @socket.sync_close = true if @socket.is_a? OpenSSL::SSL::SSLSocket
           @socket.close
-          @config.logger.debug("SSEClient socket state #{@socket.state}") if @socket.is_a? OpenSSL::SSL::SSLSocket && @config.debug_enabled
+          @config.logger.debug("SSEClient socket state #{@socket.state}") if @socket.is_a?(OpenSSL::SSL::SSLSocket) && @config.debug_enabled
+          @socket = nil
         rescue StandardError => e
           @config.logger.error("SSEClient close Error: #{e.inspect}")
         end
@@ -80,21 +78,21 @@ def connect_thread(latch)
           @config.threads[:connect_stream] = Thread.new do
             @config.logger.info('Starting connect_stream thread ...')
             new_status = connect_stream(latch)
-            push_status(new_status)
+            push_status(new_status) unless new_status.nil?
             @config.logger.info('connect_stream thread finished.')
           end
         end
 
         def connect_stream(latch)
-          return Constants::PUSH_NONRETRYABLE_ERROR unless socket_write(latch)
+          return Constants::PUSH_RETRYABLE_ERROR unless socket_write(latch)
           while connected? || @first_event.value
             begin 
               if IO.select([@socket], nil, nil, @read_timeout)
                 begin
                   partial_data = @socket.readpartial(10_000)
-                  read_first_event(partial_data, latch)
 
-                  raise 'eof exception' if partial_data == :eof
+                  first_event_status = read_first_event(partial_data, latch)
+                  return first_event_status unless first_event_status.nil?
                 rescue IO::WaitReadable => e
                   @config.logger.debug("SSE client IO::WaitReadable transient error: #{e.inspect}") if @config.debug_enabled
                   IO.select([@socket], nil, nil, @read_timeout)
@@ -107,8 +105,8 @@ def connect_stream(latch)
                   @config.logger.error("SSE read operation timed out!: #{e.inspect}")
                   return Constants::PUSH_RETRYABLE_ERROR
                 rescue EOFError => e
-                  @config.logger.error("SSE read operation EOF Exception!: #{e.inspect}")
-                  raise 'eof exception'
+                  @config.logger.error("SSE read operation EOF, server closed the connection, will reconnect: #{e.inspect}")
+                  return Constants::PUSH_RETRYABLE_ERROR
                 rescue Errno::EBADF, IOError => e
                   @config.logger.error("SSE read operation EBADF or IOError: #{e.inspect}")
                   return Constants::PUSH_RETRYABLE_ERROR
@@ -123,14 +121,9 @@ def connect_stream(latch)
                 @config.logger.error("SSE read operation timed out, no data available.")
                 return Constants::PUSH_RETRYABLE_ERROR
               end
-            rescue Errno::EBADF
-              @config.logger.debug("SSE socket is not connected (Errno::EBADF)") if @config.debug_enabled
-              break
-            rescue RuntimeError
-              raise 'eof exception'
             rescue Exception => e
               @config.logger.debug("SSE socket is not connected: #{e.inspect}") if @config.debug_enabled
-              break
+              return Constants::PUSH_RETRYABLE_ERROR
             end
 
             process_data(partial_data)
@@ -157,19 +150,20 @@ def read_first_event(data, latch)
           response_code = @event_parser.first_event(data)
           @config.logger.debug("SSE client first event code: #{response_code}") if @config.debug_enabled
 
-          error_event = false
-          events = @event_parser.parse(data)
-          events.each { |e| error_event = true if e.event_type == ERROR_EVENT_TYPE }
           @first_event.make_false
 
-          if response_code == OK_CODE && !error_event
-            @connected.make_true
-            @config.logger.debug("SSE client first event Connected is true") if @config.debug_enabled
-            @telemetry_runtime_producer.record_streaming_event(Telemetry::Domain::Constants::SSE_CONNECTION_ESTABLISHED, nil)
-            push_status(Constants::PUSH_CONNECTED)
+          if response_code != OK_CODE
+            @config.logger.error("SSE first event failed, code: #{response_code}")
+            latch.count_down
+            return Constants::PUSH_RETRYABLE_ERROR
           end
 
+          @connected.make_true
+          @config.logger.debug("SSE client first event Connected is true") if @config.debug_enabled
+          @telemetry_runtime_producer.record_streaming_event(Telemetry::Domain::Constants::SSE_CONNECTION_ESTABLISHED, nil)
+          push_status(Constants::PUSH_CONNECTED)
           latch.count_down
+          return nil
         end
 
         def socket_connect

lib/splitclient-rb/version.rb

@@ -1,3 +1,3 @@
 module SplitIoClient
-  VERSION = '8.11.0'
+  VERSION = '8.11.1'
 end

spec/engine/push_manager_spec.rb

@@ -56,6 +56,7 @@
         expect(a_request(:get, config.auth_service_url + "?s=1.3")).to have_been_made.times(1)
 
         sleep(1.5)
+        expect(config.threads.has_key?(:schedule_next_token_refresh)).to eq(true)
         expect(connected).to eq(true)
         expect(sse_handler.connected?).to eq(true)
         expect(push_status_queue.pop(true)).to eq(SplitIoClient::Constants::PUSH_CONNECTED)
@@ -93,6 +94,29 @@
       expect(connected).to eq(false)
       expect(sse_handler.connected?).to eq(false)
     end
+
+    it 'must not retry if server return 400' do
+      mock_server do |server|
+        server.setup_response('/') do |_, res|
+          send_mock_content(res, 'content', 400)
+        end
+
+        stub_request(:get, config.auth_service_url + "?s=1.3").to_return(status: 200, body: body_response)
+        config.streaming_service_url = server.base_uri
+
+        sse_handler = SplitIoClient::SSE::SSEHandler.new(config, splits_worker, segments_worker, sse_client)
+        push_manager = subject.new(config, sse_handler, api_key, runtime_producer)
+        connected = push_manager.start_sse
+
+        expect(a_request(:get, config.auth_service_url + "?s=1.3")).to have_been_made.times(1)
+
+        sleep(1.5)
+        expect(config.threads.has_key?(:schedule_next_token_refresh)).to eq(false)
+        expect(connected).to eq(false)
+        expect(sse_handler.connected?).to eq(false)
+
+      end
+    end
   end
 
   context 'stop_sse' do
@@ -125,9 +149,9 @@
   end
 end
 
-def send_mock_content(res, content)
+def send_mock_content(res, content, status = 200)
   res.content_type = 'text/event-stream'
-  res.status = 200
+  res.status = status
   res.chunked = true
   rd, wr = IO.pipe
   wr.write(content)

spec/engine/sync_manager_spec.rb

@@ -94,6 +94,7 @@
   end
 
   it 'start sync manager with wrong sse host url and non connect to server, must start polling.' do
+    ENV['SPLITCLIENT_ENV'] = "prod"
     mock_server do |server|
       server.setup_response('/') do |_, res|
         send_content(res, 'content')
@@ -108,9 +109,10 @@
       sleep(2)
       expect(a_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=-1&rbSince=-1')).to have_been_made.once
 
-      expect(config.threads.size).to eq(8)
+      expect(config.threads.size).to eq(9)
       config.threads.values.each { |thread| Thread.kill(thread) }
     end
+    ENV['SPLITCLIENT_ENV'] = "test"
   end
 
   it 'start sync manager receiving control message, must switch to polling' do
@@ -135,6 +137,28 @@
     end
   end
 
+  it 'start sync manager receiving 400 error, must switch to polling' do
+    mock_server do |server|
+      server.setup_response('/') do |_, res|
+        send_content(res, event_control, 400)
+      end
+
+      config.streaming_service_url = server.base_uri
+
+      sync_manager = subject.new(config, synchronizer, telemetry_runtime_producer, telemetry_synchronizer, status_manager, sse_handler, push_manager, push_status_queue)
+      sync_manager.start
+
+      sleep(2)
+      config.threads.select { |name, _| name.to_s.end_with? 'worker' }.values.each do |thread|
+        expect(thread.status).to eq(false) # Status fasle: when this thread is terminated normally as expected
+      end
+
+      sse_handler = sync_manager.instance_variable_get(:@sse_handler)
+      expect(sse_handler.connected?).to eq(false)
+      config.threads.values.each { |thread| Thread.kill(thread) }
+    end
+  end
+
   private
 
   def mock_split_changes_with_since(splits_json, since)
@@ -147,9 +171,9 @@ def mock_segment_changes(segment_name, segment_json, since)
       .to_return(status: 200, body: segment_json)
   end
 
-  def send_content(res, content)
+  def send_content(res, content, status = 200)
     res.content_type = 'text/event-stream'
-    res.status = 200
+    res.status = status
     res.chunked = true
     rd, wr = IO.pipe
     wr.write(content)

spec/integrations/push_client_spec.rb

@@ -84,6 +84,38 @@
       end
     end
 
+    it 'Retry streaming connection' do
+      mock_splits_request(splits, -1)
+      mock_splits_request(splits2, 1_585_948_850_109)
+      mock_splits_request(splits3, 1_585_948_850_110)
+      mock_segment_changes('segment3', segment3, '-1')
+
+      mock_server do |server|
+        server.setup_response('/') do |_, res|
+          send_content(res, event_split_kill_must_not_fetch, 500)
+        end
+
+        stub_request(:get, auth_service_url + "?s=1.3").to_return(status: 200, body: auth_body_response)
+
+        streaming_service_url = server.base_uri
+        factory = SplitIoClient::SplitFactory.new(
+          'test_api_key',
+          streaming_enabled: true,
+          streaming_service_url: streaming_service_url,
+          auth_service_url: auth_service_url
+        )
+
+        client = factory.client
+        client.block_until_ready(1)
+        server.setup_response('/') do |_, res|
+          send_content(res, event_split_update_must_fetch)
+        end
+        sleep(2)
+        expect(client.get_treatment('admin', 'push_test')).to eq('after_fetch')
+        client.destroy
+      end
+    end
+
     it 'processing split update event without fetch' do
       mock_splits_request(splits, -1)
       mock_splits_request(splits2, 1_585_948_850_109)
@@ -587,14 +619,14 @@
         expect(client.get_treatment('admin', 'push_test')).to eq('after_fetch')
         client.destroy
       end
-    end
+    end    
   end
 
   private
 
-  def send_content(res, content)
+  def send_content(res, content, status=200)
     res.content_type = 'text/event-stream'
-    res.status = 200
+    res.status = status
     res.chunked = true
     rd, wr = IO.pipe
     wr.write(content)