From 660875c7522f17c9aa5ee2eede648cb0a95c2076 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:24:31 -0700 Subject: [PATCH] fix: Reconnect the FDv2 stream after a restart that follows an earlier disconnect The FDv2 streaming synchronizer restarts its stream with interrupt() on a goodbye, invalid data, or a protocol error. After any earlier disconnect, okhttp-eventsource 5.0.0 reports that interrupt() as a StreamClosedByCallerException fault before it reconnects. The synchronizer treated every such fault as a shutdown and stopped reading, so the stream stopped with no result for the data system. Now the synchronizer stops only when close() was called. For a restart it continues to read, and the event source reconnects. The restart already queued its result and recorded the stream init, so the fault adds neither. --- .../sdk/server/StreamingSynchronizerImpl.java | 19 +- .../server/StreamingSynchronizerImplTest.java | 265 ++++++++++++++++++ 2 files changed, 278 insertions(+), 6 deletions(-) diff --git a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java index b5d80286..f6b56f22 100644 --- a/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java +++ b/lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java @@ -367,15 +367,22 @@ private void interruptedWithException(Exception e, DataSourceStatusProvider.Erro } private boolean handleError(StreamException e) { - // Check if this was a deliberate shutdown/restart rather than an actual error - boolean streamFailed = !(e instanceof StreamClosedByCallerException); - recordStreamInit(streamFailed); - if (e instanceof StreamClosedByCallerException) { - // We closed it ourselves (shutdown was called or stream was deliberately restarted) - return false; + // The event source can send this fault after close() or after restartStream(). + // Only close() stops the stream. + synchronized (closeLock) { + if (closed) { + recordStreamInit(false); + return false; + } + } + // restartStream() already queued a result and recorded the stream init. + // Continue to read events so that the event source connects again. + return true; } + recordStreamInit(true); + if (e instanceof StreamHttpErrorException) { int status = ((StreamHttpErrorException) e).getCode(); DataSourceStatusProvider.ErrorInfo errorInfo = DataSourceStatusProvider.ErrorInfo.fromHttpError(status); diff --git a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java index c5b7ca12..97a88053 100644 --- a/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java +++ b/lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/StreamingSynchronizerImplTest.java @@ -1640,4 +1640,269 @@ public void errorAfterSuccessfulChangesetRecordsNewDiagnostic() throws Exception } } + + @Test + public void goodbyeAfterEarlierDisconnectRestartsStream() throws Exception { + DiagnosticStore diagnosticStore = basicDiagnosticStore(); + + String serverIntent = makeEvent("server-intent", "{\"payloads\":[{\"id\":\"payload-1\",\"target\":100,\"intentCode\":\"xfer-full\",\"reason\":\"payload-missing\"}]}"); + String payloadTransferred = makeEvent("payload-transferred", "{\"state\":\"(p:payload-1:100)\",\"version\":100}"); + String goodbyeEvent = makeEvent("goodbye", "{\"reason\":\"service-unavailable\"}"); + + // First connection: changeset, then the server ends the response. + // Second connection: changeset + goodbye. + // Third connection: changeset. + try (HttpServer server = HttpServer.start(Handlers.sequential( + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred)), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.event(goodbyeEvent), + Handlers.SSE.leaveOpen()), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.leaveOpen())))) { + + HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp()); + SelectorSource selectorSource = mockSelectorSource(); + + StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl( + httpProperties, + server.getUri(), + "/stream", + testLogger, + selectorSource, + null, + Duration.ofMillis(100), + Thread.NORM_PRIORITY, + diagnosticStore + ); + + FDv2SourceResult result1 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.CHANGE_SET, result1.getResultType()); + + // The server ended the first response + FDv2SourceResult result2 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.STATUS, result2.getResultType()); + assertEquals(SourceSignal.INTERRUPTED, result2.getStatus().getState()); + assertEquals(DataSourceStatusProvider.ErrorKind.NETWORK_ERROR, result2.getStatus().getErrorInfo().getKind()); + + FDv2SourceResult result3 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.CHANGE_SET, result3.getResultType()); + + FDv2SourceResult result4 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.STATUS, result4.getResultType()); + assertEquals(SourceSignal.GOODBYE, result4.getStatus().getState()); + + // The synchronizer restarts the stream and gets the next changeset + FDv2SourceResult result5 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.CHANGE_SET, result5.getResultType()); + assertEquals(3, server.getRecorder().count()); + + // One successful init for each connection that delivered a changeset + LDValue event = diagnosticStore.createEventAndReset(0, 0).getJsonValue(); + LDValue streamInits = event.get("streamInits"); + assertEquals(3, streamInits.size()); + assertFalse(streamInits.get(0).get("failed").booleanValue()); + assertFalse(streamInits.get(1).get("failed").booleanValue()); + assertFalse(streamInits.get(2).get("failed").booleanValue()); + + synchronizer.close(); + } + } + + @Test + public void invalidDataAfterEarlierDisconnectRestartsStream() throws Exception { + DiagnosticStore diagnosticStore = basicDiagnosticStore(); + + String serverIntent = makeEvent("server-intent", "{\"payloads\":[{\"id\":\"payload-1\",\"target\":100,\"intentCode\":\"xfer-full\",\"reason\":\"payload-missing\"}]}"); + String payloadTransferred = makeEvent("payload-transferred", "{\"state\":\"(p:payload-1:100)\",\"version\":100}"); + String invalidPayload = makeEvent("payload-transferred", "{malformed json}"); + + // First connection: changeset, then the server ends the response. + // Second connection: invalid data. + // Third connection: changeset. + try (HttpServer server = HttpServer.start(Handlers.sequential( + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred)), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(invalidPayload), + Handlers.SSE.leaveOpen()), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.leaveOpen())))) { + + HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp()); + SelectorSource selectorSource = mockSelectorSource(); + + StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl( + httpProperties, + server.getUri(), + "/stream", + testLogger, + selectorSource, + null, + Duration.ofMillis(100), + Thread.NORM_PRIORITY, + diagnosticStore + ); + + FDv2SourceResult result1 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.CHANGE_SET, result1.getResultType()); + + // The server ended the first response + FDv2SourceResult result2 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.STATUS, result2.getResultType()); + assertEquals(SourceSignal.INTERRUPTED, result2.getStatus().getState()); + assertEquals(DataSourceStatusProvider.ErrorKind.NETWORK_ERROR, result2.getStatus().getErrorInfo().getKind()); + + FDv2SourceResult result3 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.STATUS, result3.getResultType()); + assertEquals(SourceSignal.INTERRUPTED, result3.getStatus().getState()); + assertEquals(DataSourceStatusProvider.ErrorKind.INVALID_DATA, result3.getStatus().getErrorInfo().getKind()); + + // The synchronizer restarts the stream and gets the next changeset + FDv2SourceResult result4 = synchronizer.next().get(5, TimeUnit.SECONDS); + assertEquals(SourceResultType.CHANGE_SET, result4.getResultType()); + assertEquals(3, server.getRecorder().count()); + + // The restart records the failed init. The reconnect does not record a second one. + LDValue event = diagnosticStore.createEventAndReset(0, 0).getJsonValue(); + LDValue streamInits = event.get("streamInits"); + assertEquals(3, streamInits.size()); + assertFalse(streamInits.get(0).get("failed").booleanValue()); + assertTrue(streamInits.get(1).get("failed").booleanValue()); + assertFalse(streamInits.get(2).get("failed").booleanValue()); + + synchronizer.close(); + } + } + + @Test + public void closeDuringRestartAfterEarlierDisconnectDoesNotReconnect() throws Exception { + String serverIntent = makeEvent("server-intent", "{\"payloads\":[{\"id\":\"payload-1\",\"target\":100,\"intentCode\":\"xfer-full\",\"reason\":\"payload-missing\"}]}"); + String payloadTransferred = makeEvent("payload-transferred", "{\"state\":\"(p:payload-1:100)\",\"version\":100}"); + String goodbyeEvent = makeEvent("goodbye", "{\"reason\":\"service-unavailable\"}"); + + // First connection: changeset, then the server ends the response. + // Second connection: changeset + goodbye. + // Third connection: the synchronizer must not make this request. + try (HttpServer server = HttpServer.start(Handlers.sequential( + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred)), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.event(goodbyeEvent), + Handlers.SSE.leaveOpen()), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.leaveOpen())))) { + + HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp()); + SelectorSource selectorSource = mockSelectorSource(); + + // With this initial delay, the reconnect after the goodbye waits at least 200 milliseconds. + StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl( + httpProperties, + server.getUri(), + "/stream", + testLogger, + selectorSource, + null, + Duration.ofMillis(200), + Thread.NORM_PRIORITY, + null + ); + + assertEquals(SourceResultType.CHANGE_SET, synchronizer.next().get(5, TimeUnit.SECONDS).getResultType()); + assertEquals(SourceSignal.INTERRUPTED, synchronizer.next().get(5, TimeUnit.SECONDS).getStatus().getState()); + assertEquals(SourceResultType.CHANGE_SET, synchronizer.next().get(5, TimeUnit.SECONDS).getResultType()); + assertEquals(SourceSignal.GOODBYE, synchronizer.next().get(5, TimeUnit.SECONDS).getStatus().getState()); + + // Close the synchronizer while the restarted stream waits to reconnect + Thread.sleep(100); + synchronizer.close(); + + FDv2SourceResult result = synchronizer.next().get(1, TimeUnit.SECONDS); + assertEquals(SourceResultType.STATUS, result.getResultType()); + assertEquals(SourceSignal.SHUTDOWN, result.getStatus().getState()); + + // Wait longer than the longest possible reconnect delay + Thread.sleep(1200); + assertEquals(2, server.getRecorder().count()); + } + } + + @Test + public void closeAfterEarlierDisconnectDoesNotReconnect() throws Exception { + String serverIntent = makeEvent("server-intent", "{\"payloads\":[{\"id\":\"payload-1\",\"target\":100,\"intentCode\":\"xfer-full\",\"reason\":\"payload-missing\"}]}"); + String payloadTransferred = makeEvent("payload-transferred", "{\"state\":\"(p:payload-1:100)\",\"version\":100}"); + + // First connection: changeset, then the server ends the response. + // Second connection: changeset. + // Third connection: the synchronizer must not make this request. + try (HttpServer server = HttpServer.start(Handlers.sequential( + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred)), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.leaveOpen()), + Handlers.all( + Handlers.SSE.start(), + Handlers.SSE.event(serverIntent), + Handlers.SSE.event(payloadTransferred), + Handlers.SSE.leaveOpen())))) { + + HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp()); + SelectorSource selectorSource = mockSelectorSource(); + + StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl( + httpProperties, + server.getUri(), + "/stream", + testLogger, + selectorSource, + null, + Duration.ofMillis(100), + Thread.NORM_PRIORITY, + null + ); + + assertEquals(SourceResultType.CHANGE_SET, synchronizer.next().get(5, TimeUnit.SECONDS).getResultType()); + assertEquals(SourceSignal.INTERRUPTED, synchronizer.next().get(5, TimeUnit.SECONDS).getStatus().getState()); + assertEquals(SourceResultType.CHANGE_SET, synchronizer.next().get(5, TimeUnit.SECONDS).getResultType()); + + synchronizer.close(); + + FDv2SourceResult result = synchronizer.next().get(1, TimeUnit.SECONDS); + assertEquals(SourceResultType.STATUS, result.getResultType()); + assertEquals(SourceSignal.SHUTDOWN, result.getStatus().getState()); + + // Wait longer than the longest possible reconnect delay + Thread.sleep(1000); + assertEquals(2, server.getRecorder().count()); + } + } }