diff --git a/CHANGES.txt b/CHANGES.txt index e12623f5..ad9ff732 100644 --- a/CHANGES.txt +++ b/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 diff --git a/lib/splitclient-rb/engine/push_manager.rb b/lib/splitclient-rb/engine/push_manager.rb index e36a9588..742082d4 100644 --- a/lib/splitclient-rb/engine/push_manager.rb +++ b/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 diff --git a/lib/splitclient-rb/engine/synchronizer.rb b/lib/splitclient-rb/engine/synchronizer.rb index 841aff6d..2d9442a4 100644 --- a/lib/splitclient-rb/engine/synchronizer.rb +++ b/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 diff --git a/lib/splitclient-rb/sse/event_source/client.rb b/lib/splitclient-rb/sse/event_source/client.rb index 74c0cfc3..cad1b4b6 100644 --- a/lib/splitclient-rb/sse/event_source/client.rb +++ b/lib/splitclient-rb/sse/event_source/client.rb @@ -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 + @socket = nil rescue StandardError => e @config.logger.error("SSEClient close Error: #{e.inspect}") end @@ -80,7 +78,7 @@ 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 @@ -92,12 +90,9 @@ def connect_stream(latch) if IO.select([@socket], nil, nil, @read_timeout) begin partial_data = @socket.readpartial(10_000) - read_first_event(partial_data, latch) - if partial_data == :eof - @config.logger.error("SSE recived EOF unexpectedly") - return Constants::PUSH_RETRYABLE_ERROR - end + 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) @@ -110,7 +105,7 @@ 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}") + @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}") @@ -126,12 +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 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) @@ -158,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 diff --git a/lib/splitclient-rb/version.rb b/lib/splitclient-rb/version.rb index 24237148..efc0fb8e 100644 --- a/lib/splitclient-rb/version.rb +++ b/lib/splitclient-rb/version.rb @@ -1,3 +1,3 @@ module SplitIoClient - VERSION = '8.11.1-beta1' + VERSION = '8.11.1' end diff --git a/spec/engine/push_manager_spec.rb b/spec/engine/push_manager_spec.rb index bd3a3540..b7cb88db 100644 --- a/spec/engine/push_manager_spec.rb +++ b/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) diff --git a/spec/engine/sync_manager_spec.rb b/spec/engine/sync_manager_spec.rb index 9bbcc13c..c3771f3b 100644 --- a/spec/engine/sync_manager_spec.rb +++ b/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) diff --git a/spec/sse/event_source/client_spec.rb b/spec/sse/event_source/client_spec.rb index ca14cc47..39827dc4 100644 --- a/spec/sse/event_source/client_spec.rb +++ b/spec/sse/event_source/client_spec.rb @@ -208,16 +208,16 @@ it 'receive error event' do mock_server do |server| server.setup_response('/') do |_, res| - send_stream_content(res, event_error, 400) + send_stream_content(res, event_error, 200) end start_workers sse_client = subject.new(config, api_token, telemetry_runtime_producer, event_parser, notification_manager_keeper, notification_processor, push_status_queue) connected = sse_client.start(server.base_uri) - - expect(connected).to eq(false) - expect(sse_client.connected?).to eq(false) - expect { push_status_queue.pop(true) }.to raise_error(ThreadError) + latch = Concurrent::CountDownLatch.new(1) + res = sse_client.send(:connect_stream, latch) + push_status_queue.pop(true) + expect(push_status_queue.pop(true)).to eq(SplitIoClient::Constants::PUSH_RETRYABLE_ERROR) stop_workers end @@ -261,9 +261,10 @@ start_workers sse_client = subject.new(config, api_token, telemetry_runtime_producer, event_parser, notification_manager_keeper, notification_processor, push_status_queue) - connected = sse_client.start(server.base_uri) - expect(connected).to eq(false) - expect { push_status_queue.pop(true) }.to raise_error(ThreadError) + sse_client.instance_variable_set(:@uri, URI(server.base_uri)) + latch = Concurrent::CountDownLatch.new(1) + res = sse_client.send(:connect_stream, latch) + expect(res).to eq(SplitIoClient::Constants::PUSH_RETRYABLE_ERROR) stop_workers end @@ -315,12 +316,12 @@ allow(sse_client).to receive(:read_first_event).and_raise(EOFError) sleep(1) - thr1 = Thread.new do - sse_client.send(:connect_stream, latch) - end + + res = sse_client.send(:connect_stream, latch) + expect(res).to eq(SplitIoClient::Constants::PUSH_RETRYABLE_ERROR) sleep(1) - allow(sse_client).to receive(:read_first_event).and_return(true) - expect(log.string).to include 'SSE read operation EOF Exception' + + expect(log.string).to include 'SSE read operation EOF, server closed the connection, will reconnect' stop_workers end @@ -340,9 +341,7 @@ allow(sse_client).to receive(:read_first_event).and_raise(Errno::EAGAIN) sleep(1) - thr1 = Thread.new do - sse_client.send(:connect_stream, latch) - end + sse_client.send(:connect_stream, latch) sleep(1) allow(sse_client).to receive(:read_first_event).and_return(true) expect(log.string).to include 'SSE client transient error' @@ -368,10 +367,10 @@ allow(sse_client2).to receive(:read_first_event).and_raise(IO::EWOULDBLOCKWaitReadable) sleep(1) - thr2 = Thread.new do - sse_client2.send(:connect_stream, latch) - end + + sse_client2.send(:connect_stream, latch) sleep(1) + allow(sse_client2).to receive(:read_first_event).and_return(true) expect(log2.string).to include 'SSE client IO::WaitReadable transient error' @@ -394,10 +393,8 @@ sse_client2.instance_variable_set(:@uri, URI(server.base_uri)) latch = Concurrent::CountDownLatch.new(1) - thr2 = Thread.new do - res = sse_client2.send(:connect_stream, latch) - expect(res).to eq(SplitIoClient::Constants::PUSH_RETRYABLE_ERROR) - end + res = sse_client2.send(:connect_stream, latch) + expect(res).to eq(SplitIoClient::Constants::PUSH_RETRYABLE_ERROR) stop_workers end diff --git a/spec/sse/sse_handler_spec.rb b/spec/sse/sse_handler_spec.rb index 1c81abcf..582afb07 100644 --- a/spec/sse/sse_handler_spec.rb +++ b/spec/sse/sse_handler_spec.rb @@ -71,13 +71,11 @@ sse_handler = subject.new(config, splits_worker, segments_worker, sse_client) connected = sse_handler.start('token-test', 'channel-test') - expect(connected).to eq(false) + sleep 1 expect(sse_handler.connected?).to eq(false) expect(sse_handler.sse_client.connected?).to eq(false) - expect { push_status_queue.pop(true) }.to raise_error(ThreadError) sse_handler.stop - expect { push_status_queue.pop(true) }.to raise_error(ThreadError) end end