fix: Report one data source status for an invalid stream event

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

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