fix: force streaming to reconnect if a critical error is encountered

Author: sighphyre

examples/streaming-example/src/main/java/io/getunleash/example/StreamingExample.java

@@ -34,11 +34,11 @@ public static void main(String[] args) {
                 UnleashConfig.builder()
                         .appName("streaming-example")
                         .instanceId("streaming-example-instance")
-                        .unleashAPI(getOrElse("UNLEASH_API_URL", "https://app.unleash-hosted.com/demo/api"))
+                        .unleashAPI(getOrElse("UNLEASH_API_URL", "http://localhost:4242/api"))
                         .customHttpHeader(
                                 "Authorization",
                                 getOrElse("UNLEASH_API_TOKEN",
-                                        "*:development.25a06b75248528f8ca93ce179dcdd141aedfb632231e0d21fd8ff349"))
+                                        "*:development.0e244a346fec7bca236182d59ffa232baa16c14c68acb2ec21b2c1d3"))
                         .experimentalStreamingMode()
                         .subscriber(subscriber)
                         .build();

src/main/java/io/getunleash/repository/PollingFeatureFetcher.java

@@ -8,7 +8,6 @@
 import io.getunleash.util.Throttler;
 import io.getunleash.util.UnleashConfig;
 import io.getunleash.util.UnleashScheduledExecutor;
-
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
@@ -58,10 +57,10 @@ public void start() {
                 runInitialFetch(this.unleashConfig.getStartupExceptionHandler()).run();
             } else {
                 runInitialFetch(
-                        // just throw exception handler
-                        e -> {
-                            throw e;
-                        })
+                                // just throw exception handler
+                                e -> {
+                                    throw e;
+                                })
                         .run();
             }
         } else if (!unleashConfig.isDisablePolling()) {

src/main/java/io/getunleash/repository/StreamingFeatureFetcherImpl.java

@@ -64,7 +64,7 @@ class StreamingFeatureFetcherImpl implements FetchWorker {
         this.modeController = modeController;
     }
 
-    public void start() {
+    public synchronized void start() {
         try {
             URI streamingUri = config.getUnleashURLs().getStreamingURL().toURI();
 
@@ -104,7 +104,7 @@ public void start() {
         }
     }
 
-    public void stop() {
+    public synchronized void stop() {
         try {
             BackgroundEventSource currentEventSource = eventSource;
             if (currentEventSource != null) {
@@ -116,7 +116,7 @@ public void stop() {
         }
     }
 
-    private void reconnect() {
+    void reconnect() {
         stop();
         start();
     }

src/test/java/io/getunleash/repository/StreamingFeatureFetchingTest.java

@@ -3,14 +3,17 @@
 import static com.github.tomakehurst.wiremock.client.WireMock.*;
 import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 
 import com.github.tomakehurst.wiremock.junit5.WireMockExtension;
+import com.github.tomakehurst.wiremock.verification.LoggedRequest;
 import io.getunleash.DefaultUnleash;
 import io.getunleash.SynchronousTestExecutor;
 import io.getunleash.Unleash;
 import io.getunleash.engine.UnleashEngine;
+import io.getunleash.engine.YggdrasilInvalidInputException;
 import io.getunleash.event.ClientFeaturesResponse;
 import io.getunleash.event.EventDispatcher;
 import io.getunleash.event.GatedEventEmitter;
@@ -22,6 +25,7 @@
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -178,6 +182,69 @@ public void should_write_backup_file_when_streaming_update_received() throws Exc
         assertThat(savedBackupContent).doesNotContain("\"events\""); // store state
     }
 
+    @Test
+    void should_reconnect_without_last_event_id_header() throws Exception {
+        String hydration =
+                "{\"events\":[{\"type\":\"hydration\",\"eventId\":1,\"features\":[{\"name\":\"deltaFeature\",\"enabled\":true,\"strategies\":[],\"variants\":[]}],\"segments\":[]}]}";
+
+        String badUpdate = "{not-json";
+
+        stubFor(
+                get(urlEqualTo("/api/client/streaming"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withHeader("Content-Type", "text/event-stream")
+                                        .withBody(
+                                                "event: unleash-connected\n"
+                                                        + "data: "
+                                                        + hydration
+                                                        + "\n\n"
+                                                        + "event: unleash-updated\n"
+                                                        + "data: "
+                                                        + badUpdate
+                                                        + "\n\n")));
+
+        FeatureBackupHandlerFile backupHandler = mock(FeatureBackupHandlerFile.class);
+        UnleashEngine mockEngine = mock(UnleashEngine.class);
+
+        doThrow(new YggdrasilInvalidInputException("Invalid input"))
+                .when(mockEngine)
+                .takeState(badUpdate);
+
+        StreamingFeatureFetcherImpl streamingFetcher =
+                new StreamingFeatureFetcherImpl(
+                        config,
+                        new GatedEventEmitter(new EventDispatcher(config)),
+                        mockEngine,
+                        backupHandler,
+                        null);
+        streamingFetcher.start();
+
+        // Wait for reconnect to happen by polling WireMock's request log
+        long deadline = System.currentTimeMillis() + 5_000;
+        while (System.currentTimeMillis() < deadline) {
+            int count = serverMock.getAllServeEvents().size();
+            if (count >= 2) break;
+            Thread.sleep(50);
+        }
+
+        var serveEvents = serverMock.getAllServeEvents();
+        assertThat(serveEvents.size()).isGreaterThanOrEqualTo(2);
+
+        List<LoggedRequest> requests =
+                serveEvents.stream()
+                        .map(se -> se.getRequest())
+                        .filter(req -> req.getMethod().getName().equals("GET"))
+                        .filter(req -> req.getUrl().equals("/api/client/streaming"))
+                        .collect(Collectors.toList());
+
+        LoggedRequest reconnectionRequest = requests.get(1);
+        // The important part - a reconnection should never include the last-event-id header so that
+        // we get a fresh hydration
+        assertThat(reconnectionRequest.getHeader("Last-Event-ID")).isNull();
+    }
+
     private static class TestSubscriber implements UnleashSubscriber {
         private final CountDownLatch togglesFetchedLatch = new CountDownLatch(1);
         private final CountDownLatch multipleTogglesFetchedLatch = new CountDownLatch(2);