From f07f51c1cec5570a401ba51e5fc401fe988c604a Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 5 Nov 2026 12:48:07 -0500 Subject: [PATCH] fix: Report one data source status for an invalid stream event An FDv2 stream event with invalid JSON produced two INTERRUPTED updates. The event handler yielded one with INVALID_DATA and re-raised, and the SSE client gave the re-raised error to the error handler, which yielded a second one with UNKNOWN. The data source status then reported UNKNOWN as the last error. Any other error raised while handling an event was also reported twice. The event handler now only logs and re-raises. The error handler reports the error once, and maps JSON::ParserError to INVALID_DATA, as the FDv1 stream did. --- lib/ldclient-rb/impl/data_system/streaming.rb | 43 ++++++--------- .../streaming_synchronizer_spec.rb | 53 +++++++++++++++++++ 2 files changed, 69 insertions(+), 27 deletions(-) diff --git a/lib/ldclient-rb/impl/data_system/streaming.rb b/lib/ldclient-rb/impl/data_system/streaming.rb index ad00531b..0ff596c0 100644 --- a/lib/ldclient-rb/impl/data_system/streaming.rb +++ b/lib/ldclient-rb/impl/data_system/streaming.rb @@ -133,35 +133,10 @@ def sync(ss) fdv1_fallback_pending = false stop end - rescue JSON::ParserError => e - @logger.info { "[LDClient] Error parsing stream event; will restart stream: #{e}" } - yield LaunchDarkly::Interfaces::DataSystem::Update.new( - state: LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, - error: LaunchDarkly::Interfaces::DataSource::ErrorInfo.new( - LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA, - 0, - e.to_s, - Time.now - ), - environment_id: envid - ) - - # Re-raise the exception so the SSE implementation can catch it and restart the stream. - raise rescue => e @logger.info { "[LDClient] Error while handling stream event; will restart stream: #{e}" } - yield LaunchDarkly::Interfaces::DataSystem::Update.new( - state: LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, - error: LaunchDarkly::Interfaces::DataSource::ErrorInfo.new( - LaunchDarkly::Interfaces::DataSource::ErrorInfo::UNKNOWN, - 0, - e.to_s, - Time.now - ), - environment_id: envid - ) - - # Re-raise the exception so the SSE implementation can catch it and restart the stream. + + # The SSE client gives the exception to on_error, which reports it, and then restarts the stream. raise end end @@ -345,6 +320,20 @@ def stop @logger.warn { "[LDClient] #{error_info.message}" } + when JSON::ParserError + @logger.error { "[LDClient] Invalid data on stream connection: #{error}, will retry" } + + update = LaunchDarkly::Interfaces::DataSystem::Update.new( + state: LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, + error: LaunchDarkly::Interfaces::DataSource::ErrorInfo.new( + LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA, + 0, + error.to_s, + Time.now + ), + environment_id: envid + ) + when SSE::Errors::HTTPContentTypeError, SSE::Errors::HTTPProxyError, SSE::Errors::ReadTimeoutError, SSE::Errors::StreamClosedByServerError @logger.warn { "[LDClient] Network error on stream connection: #{error}, will retry" } diff --git a/spec/impl/data_system/streaming_synchronizer_spec.rb b/spec/impl/data_system/streaming_synchronizer_spec.rb index 261b7368..d1934188 100644 --- a/spec/impl/data_system/streaming_synchronizer_spec.rb +++ b/spec/impl/data_system/streaming_synchronizer_spec.rb @@ -481,6 +481,59 @@ def initialize(type, data = nil) expect(update.error.kind).to eq(LaunchDarkly::Interfaces::DataSource::ErrorInfo::NETWORK_ERROR) expect(update.environment_id).to eq("env-abc") end + + it "reports invalid JSON as invalid data" do + expect(logger).to receive(:error) + + update = synchronizer.send(:handle_error, JSON::ParserError.new("bad"), "env-abc", false) + + expect(update.state).to eq(LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED) + expect(update.error.kind).to eq(LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA) + expect(update.environment_id).to eq("env-abc") + end + + it "reports one invalid data status for an invalid stream event" do + transfer_none = "event: server-intent\ndata: " + JSON.generate( + LaunchDarkly::Interfaces::DataSystem::ServerIntent.new( + payload: LaunchDarkly::Interfaces::DataSystem::Payload.new( + id: "id", target: 1, code: LaunchDarkly::Interfaces::DataSystem::IntentCode::TRANSFER_NONE, reason: "up-to-date" + ) + ).to_h + ) + "\n\n" + request_count = 0 + with_server do |server| + server.setup_response(FDV2_STREAMING_ENDPOINT) do |_req, res| + request_count += 1 + res.status = 200 + res.content_type = "text/event-stream" + res.body = request_count == 1 ? "event: server-intent\ndata: {not json\n\n" : transfer_none + end + + stream = LaunchDarkly::DataSystem::StreamingDataSourceBuilder.new + .base_uri(server.base_uri.to_s) + .initial_reconnect_delay(0.01) + .build(sdk_key, LaunchDarkly::Config.new(logger: $null_log)) + updates = Queue.new + thread = Thread.new { stream.sync(Struct.new(:selector).new(nil)) { |update| updates << update } } + begin + received = [] + while (update = updates.pop(timeout: 5)) + received << update + break if update.state == LaunchDarkly::Interfaces::DataSource::Status::VALID + end + ensure + stream.stop + thread.join(5) + end + + expected_states = [ + LaunchDarkly::Interfaces::DataSource::Status::INTERRUPTED, + LaunchDarkly::Interfaces::DataSource::Status::VALID, + ] + expect(received.map(&:state)).to eq(expected_states) + expect(received.first.error.kind).to eq(LaunchDarkly::Interfaces::DataSource::ErrorInfo::INVALID_DATA) + end + end end describe 'SSE client options' do