fix: restart FDv2 stream on parse or protocol error

Author: beekld

libs/server-sdk/src/data_systems/fdv2/streaming_synchronizer.cpp

@@ -192,6 +192,10 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) {
         LD_LOG(logger_, LogLevel::kError) << kIdentity << ": " << msg;
         Notify(FDv2SourceResult{FDv2SourceResult::Interrupted{
             MakeError(ErrorKind::kInvalidData, 0, std::move(msg))}});
+        std::lock_guard lock(mutex_);
+        if (sse_client_) {
+            sse_client_->async_restart("FDv2 parse error");
+        }
         return;
     }
 
@@ -251,6 +255,10 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) {
                     << kIdentity << ": " << r.message;
                 Notify(FDv2SourceResult{FDv2SourceResult::Interrupted{
                     MakeError(ErrorKind::kInvalidData, 0, r.message)}});
+                std::lock_guard lock(mutex_);
+                if (sse_client_) {
+                    sse_client_->async_restart("FDv2 protocol error");
+                }
             } else {
                 static_assert(always_false_v<T>, "non-exhaustive visitor");
             }

libs/server-sdk/tests/fdv2_streaming_synchronizer_test.cpp

@@ -584,6 +584,9 @@ TEST(FDv2StreamingSynchronizerTest, MalformedJsonEventReturnsInterrupted) {
         1s);
     FDv2StreamingSynchronizerTestPeer::MarkStarted(synchronizer);
 
+    auto mock_client = std::make_shared<MockSseClient>();
+    FDv2StreamingSynchronizerTestPeer::SetSseClient(synchronizer, mock_client);
+
     sse::Event bad_event("server-intent", "this is not json");
 
     // Act: deliver an event whose data field cannot be parsed as JSON.
@@ -592,13 +595,43 @@ TEST(FDv2StreamingSynchronizerTest, MalformedJsonEventReturnsInterrupted) {
     auto result = future.WaitForResult(2s);
 
     // Assert: the synchronizer reports Interrupted{kInvalidData} so the
-    // orchestrator knows the stream produced unparseable bytes.
+    // orchestrator knows the stream produced unparseable bytes, and drives
+    // the SSE client to restart so the next connection starts clean.
     ASSERT_TRUE(result.has_value());
     auto* interrupted =
         std::get_if<FDv2SourceResult::Interrupted>(&result->value);
     ASSERT_NE(interrupted, nullptr);
     EXPECT_EQ(interrupted->error.Kind(),
               FDv2SourceResult::ErrorInfo::ErrorKind::kInvalidData);
+    EXPECT_EQ(mock_client->restart_count_, 1);
+}
+
+TEST(FDv2StreamingSynchronizerTest,
+     SchemaViolationServerIntentTriggersRestart) {
+    auto logger = MakeNullLogger();
+    IoContextRunner runner;
+
+    FDv2StreamingSynchronizer synchronizer(
+        runner.context().get_executor(), logger,
+        MakeEndpoints("http://localhost"), MakeHttpProperties(), std::nullopt,
+        1s);
+    FDv2StreamingSynchronizerTestPeer::MarkStarted(synchronizer);
+
+    auto mock_client = std::make_shared<MockSseClient>();
+    FDv2StreamingSynchronizerTestPeer::SetSseClient(synchronizer, mock_client);
+
+    // Well-formed JSON, but the shape doesn't match a server-intent payload.
+    sse::Event bad_event("server-intent",
+                         R"({"data":{"flags":true,"segments":{}}})");
+
+    FDv2StreamingSynchronizerTestPeer::OnEvent(synchronizer, bad_event);
+    auto future = synchronizer.Next(data_model::Selector{});
+    auto result = future.WaitForResult(2s);
+
+    ASSERT_TRUE(result.has_value());
+    ASSERT_NE(std::get_if<FDv2SourceResult::Interrupted>(&result->value),
+              nullptr);
+    EXPECT_EQ(mock_client->restart_count_, 1);
 }
 
 TEST(FDv2StreamingSynchronizerTest, TranslationFailureReturnsInterrupted) {