This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new b2e1a8174b Fixes #7316: Resynchronize the gateway after failed
websocket config updates (#7344)
b2e1a8174b is described below
commit b2e1a8174b25d500768811909fc0976d673b5139
Author: BobSong <[email protected]>
AuthorDate: Tue Oct 6 10:21:58 2026 +0800
Fixes #7316: Resynchronize the gateway after failed websocket config
updates (#7344)
* fix(websocket): resynchronize the gateway after failed config updates
* chore(ci): re-trigger e2e workflow after infra flake
* fix(websocket): keep the recovery loop alive after bounded sync failures
* test(websocket): verify delayed full resync and unknown events
* fix(websocket): reset failure budget after complete full sync
* fix(websocket): preserve failure budget across unframed sync payloads
---------
Co-authored-by: BobSong-dev <[email protected]>
---
.../data/websocket/client/InitialSyncState.java | 21 ++
.../websocket/client/ShenyuWebsocketClient.java | 121 +++++++--
.../websocket/client/InitialSyncStateTest.java | 31 +++
.../client/ShenyuWebsocketClientTest.java | 273 +++++++++++++++++++++
4 files changed, 420 insertions(+), 26 deletions(-)
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncState.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncState.java
index a504214786..fd3c024d33 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncState.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncState.java
@@ -33,6 +33,8 @@ public final class InitialSyncState {
private final AtomicBoolean ready;
+ private final Runnable onComplete;
+
private String requestId;
private int sequence;
@@ -50,7 +52,17 @@ public final class InitialSyncState {
* @param ready shared startup latch
*/
public InitialSyncState(final AtomicBoolean ready) {
+ this(ready, () -> { });
+ }
+
+ /**
+ * Create connection state with a full-attempt completion callback.
+ * @param ready shared startup latch
+ * @param onComplete callback after the end frame and all application work
succeed
+ */
+ public InitialSyncState(final AtomicBoolean ready, final Runnable
onComplete) {
this.ready = ready;
+ this.onComplete = Objects.requireNonNull(onComplete);
}
/**
@@ -140,9 +152,18 @@ public final class InitialSyncState {
if (ended && pending == 0 && !failed && System.nanoTime() - startedAt
< TimeUnit.SECONDS.toNanos(60)) {
ready.set(true);
requestId = null;
+ onComplete.run();
}
}
+ /**
+ * Whether the latest attempt completed successfully, independent of the
startup latch.
+ * @return complete attempt
+ */
+ public synchronized boolean isComplete() {
+ return ended && !failed && Objects.isNull(requestId);
+ }
+
/**
* Invalidate the current attempt without revoking an earlier success.
*/
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
index 1d4fe3bb9a..64522afd22 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java
@@ -88,10 +88,15 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
private static final long MAX_RECONNECT_BACKOFF_MS =
TimeUnit.SECONDS.toMillis(60);
+ private static final int MAX_CONSECUTIVE_SYNC_FAILURES = 3;
+
private volatile boolean alreadySync = Boolean.FALSE;
private InitialSyncState initialSyncState;
+ private volatile long nextSyncRetryAt;
+
+
private final WebsocketDataHandler websocketDataHandler;
private final Timer timer;
@@ -110,6 +115,8 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
private final AtomicBoolean reconnecting = new AtomicBoolean(false);
+ private final AtomicInteger consecutiveSyncFailures = new AtomicInteger(0);
+
private volatile long lastReconnectAttemptTime;
private final AtomicInteger reconnectBackoff = new AtomicInteger(0);
@@ -205,7 +212,7 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
final AtomicBoolean initialSyncReady) {
super(serverUri, headers);
if (Objects.nonNull(initialSyncReady)) {
- this.initialSyncState = new InitialSyncState(initialSyncReady);
+ this.initialSyncState = new InitialSyncState(initialSyncReady,
this::resetSyncFailures);
}
this.namespaceId = namespaceId;
LOG.info("shenyu bootstrap websocket namespaceId: {}", namespaceId);
@@ -270,25 +277,26 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
@Override
public void onMessage(final String result) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("onMessage server[{}] result({})",
this.getURI().toString(), result);
+ final Map<String, Object> jsonToMap;
+ try {
+ jsonToMap = JsonUtils.jsonToMap(result);
+ if (Objects.isNull(jsonToMap)) {
+ return;
+ }
+ } catch (RuntimeException ex) {
+ LOG.warn("Ignoring unparseable websocket frame from server[{}]",
this.getURI());
+ return;
}
-
try {
- Map<String, Object> jsonToMap = JsonUtils.jsonToMap(result);
Object eventType = jsonToMap.get(RunningModeConstants.EVENT_TYPE);
if (Objects.equals(WebsocketSyncFrame.EVENT_TYPE, eventType) &&
Objects.nonNull(initialSyncState)) {
initialSyncState.accept(GsonUtils.getInstance().fromJson(result,
WebsocketSyncFrame.class), this::handleResult);
- return;
- }
- if (Objects.equals(DataEventTypeEnum.RUNNING_MODE.name(),
eventType)) {
- LOG.info("server[{}] handle running mode result({})",
this.getURI().toString(), result);
+ } else if (Objects.equals(DataEventTypeEnum.RUNNING_MODE.name(),
eventType)) {
this.runningMode =
String.valueOf(jsonToMap.get(RunningModeConstants.RUNNING_MODE));
- if (Objects.equals(RunningModeEnum.STANDALONE.name(),
runningMode)) {
- return;
+ if (!Objects.equals(RunningModeEnum.STANDALONE.name(),
runningMode)) {
+ this.masterUrl =
String.valueOf(jsonToMap.get(RunningModeConstants.MASTER_URL));
+ this.isConnectedToMaster =
Boolean.TRUE.equals(jsonToMap.get(RunningModeConstants.IS_MASTER));
}
- this.masterUrl =
String.valueOf(jsonToMap.get(RunningModeConstants.MASTER_URL));
- this.isConnectedToMaster =
Boolean.TRUE.equals(jsonToMap.get(RunningModeConstants.IS_MASTER));
} else if (Objects.nonNull(initialSyncState)) {
initialSyncState.applyIncremental(() -> handleResult(result));
} else {
@@ -298,10 +306,10 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
if (Objects.nonNull(initialSyncState)) {
initialSyncState.invalidate();
}
- LOG.warn("Failed to handle websocket message from server[{}], the
message will be ignored", this.getURI(), ex);
+ LOG.warn("Failed to handle websocket frame from server[{}]",
this.getURI(), ex);
}
}
-
+
@Override
public void onClose(final int i, final String s, final boolean b) {
this.close();
@@ -344,6 +352,14 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
if (this.manuallyClosed.get()) {
return;
}
+ if (nextSyncRetryAt != 0) {
+ if (System.nanoTime() - nextSyncRetryAt < 0) {
+ return;
+ }
+ nextSyncRetryAt = 0;
+ close();
+ return;
+ }
if (Objects.nonNull(initialSyncState) && this.isOpen() &&
initialSyncState.needsReconnect()) {
close();
return;
@@ -421,20 +437,73 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
*/
private void handleResult(final String result) {
LOG.info("server [{}] handleResult({})", this.getURI().toString(),
result);
- WebsocketData<?> websocketData =
GsonUtils.getInstance().fromJson(result, WebsocketData.class);
- ConfigGroupEnum groupEnum =
ConfigGroupEnum.acquireByName(websocketData.getGroupType());
- String eventType = websocketData.getEventType();
- String json = GsonUtils.getInstance().toJson(websocketData.getData());
- if (websocketData.isFullSnapshot()) {
- if (!DataEventTypeEnum.REFRESH.name().equals(eventType) &&
!DataEventTypeEnum.MYSELF.name().equals(eventType)) {
- throw new IllegalArgumentException("Snapshot requires a
refresh event");
+ WebsocketData<?> websocketData;
+ try {
+ websocketData = GsonUtils.getInstance().fromJson(result,
WebsocketData.class);
+ } catch (RuntimeException ex) {
+ // a frame that cannot be interpreted as a config message is a
protocol-shape
+ // mismatch, not a recoverable config change: ignore it like the
previous
+ // behavior instead of dropping the connection
+ LOG.warn("Failed to parse websocket message from server[{}], the
message will be ignored", this.getURI(), ex);
+ return;
+ }
+ ConfigGroupEnum groupEnum;
+ String eventType;
+ String json;
+ try {
+ groupEnum =
ConfigGroupEnum.acquireByName(websocketData.getGroupType());
+ eventType = websocketData.getEventType();
+ DataEventTypeEnum.acquireByName(eventType);
+ json = GsonUtils.getInstance().toJson(websocketData.getData());
+ } catch (RuntimeException ex) {
+ LOG.warn("Failed to resolve websocket message group from
server[{}], the message will be ignored", this.getURI(), ex);
+ return;
+ }
+ try {
+ if (websocketData.isFullSnapshot()) {
+ if (!DataEventTypeEnum.REFRESH.name().equals(eventType) &&
!DataEventTypeEnum.MYSELF.name().equals(eventType)) {
+ throw new IllegalArgumentException("Snapshot requires a
refresh event");
+ }
+ websocketDataHandler.snapshot(groupEnum, json,
websocketData.getNamespaceId(), namespaceId);
+ } else {
+ websocketDataHandler.executor(groupEnum, json, eventType);
+ }
+ // Individual payloads, including legacy MYSELF and standalone
snapshots,
+ // cannot prove that the complete sync cycle succeeded. Only the
framed
+ // completion callback may reset the failure budget and cooldown.
+ } catch (RuntimeException ex) {
+ handleSyncFailure(ex, groupEnum.name(), eventType);
+ if
(org.apache.shenyu.common.utils.InitialSyncApplication.isActive()) {
+ throw ex;
}
- websocketDataHandler.snapshot(groupEnum, json,
websocketData.getNamespaceId(), namespaceId);
- } else {
- websocketDataHandler.executor(groupEnum, json, eventType);
}
}
-
+
+ private void resetSyncFailures() {
+ consecutiveSyncFailures.set(0);
+ nextSyncRetryAt = 0;
+ }
+
+ /**
+ * Recover failed configuration application without abandoning the timer.
+ * @param ex application failure
+ * @param groupType configuration group
+ * @param eventType event type
+ */
+ private void handleSyncFailure(final RuntimeException ex, final String
groupType, final String eventType) {
+ int failures = consecutiveSyncFailures.updateAndGet(value ->
Math.min(value + 1, MAX_CONSECUTIVE_SYNC_FAILURES));
+ if (failures >= MAX_CONSECUTIVE_SYNC_FAILURES) {
+ if (nextSyncRetryAt == 0) {
+ nextSyncRetryAt = System.nanoTime() + MAX_RECONNECT_BACKOFF_MS
* 1_000_000L;
+ }
+ LOG.warn("websocket sync failed, group={}, eventType={}; full
resync scheduled after backoff", groupType, eventType, ex);
+ return;
+ }
+ LOG.warn("websocket sync failed, group={}, eventType={},
consecutiveFailures={}; reconnecting for full resync",
+ groupType, eventType, failures, ex);
+ this.close();
+ }
+
/**
* Gets the master url.
*
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncStateTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncStateTest.java
index 7d98aab7bc..07a84327f9 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncStateTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/InitialSyncStateTest.java
@@ -29,7 +29,9 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -39,6 +41,35 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
class InitialSyncStateTest {
+ @Test
+ void testCompletionCallbackRejectsFailedAndAbandonedAttempts() {
+ AtomicBoolean ready = new AtomicBoolean(true);
+ AtomicInteger completed = new AtomicInteger();
+ InitialSyncState state = new InitialSyncState(ready,
completed::incrementAndGet);
+ String failed = state.begin();
+ CompletableFuture<Void> failure = new CompletableFuture<>();
+ state.accept(new WebsocketSyncFrame(failed, 0, "data"), value ->
InitialSyncApplication.register(failure));
+ state.accept(new WebsocketSyncFrame(failed, 1, null), value -> { });
+ failure.completeExceptionally(new IllegalStateException("poison"));
+ assertEquals(0, completed.get());
+ assertFalse(state.isComplete());
+ String abandoned = state.begin();
+ CompletableFuture<Void> pending = new CompletableFuture<>();
+ state.accept(new WebsocketSyncFrame(abandoned, 0, "data"), value ->
InitialSyncApplication.register(pending));
+ state.accept(new WebsocketSyncFrame(abandoned, 1, null), value -> { });
+ state.invalidate();
+ final String retry = state.begin();
+ pending.complete(null);
+ assertEquals(0, completed.get());
+ assertFalse(state.isComplete());
+ state.accept(new WebsocketSyncFrame(retry, 0, null), value -> { });
+ assertEquals(1, completed.get());
+ assertTrue(state.isComplete());
+ state.accept(new WebsocketSyncFrame(retry, 0, null), value -> { });
+ assertEquals(1, completed.get());
+ }
+
+
@Test
void testInterleavedIncrementalMustFinishBeforeReadiness() {
AtomicBoolean ready = new AtomicBoolean();
diff --git
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
index e2e345687f..deb8eeee9c 100644
---
a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java
@@ -25,6 +25,8 @@ import org.apache.shenyu.common.enums.DataEventTypeEnum;
import org.apache.shenyu.common.exception.ShenyuException;
import org.apache.shenyu.common.timer.TimerTask;
import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.common.utils.InitialSyncApplication;
+import
org.apache.shenyu.plugin.sync.data.websocket.handler.WebsocketDataHandler;
import org.apache.shenyu.sync.data.api.AuthDataSubscriber;
import org.apache.shenyu.sync.data.api.MetaDataSubscriber;
import org.apache.shenyu.sync.data.api.PluginDataSubscriber;
@@ -44,8 +46,10 @@ import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -61,6 +65,7 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.withSettings;
@@ -202,6 +207,14 @@ public class ShenyuWebsocketClientTest {
setField(client, "reconnecting", new AtomicBoolean(false));
setField(client, "reconnectBackoff", new AtomicInteger(0));
setField(client, "lastReconnectAttemptTime", 0L);
+ setField(client, "consecutiveSyncFailures", new AtomicInteger(0));
+ setField(client, "websocketDataHandler",
mock(WebsocketDataHandler.class));
+ return client;
+ }
+
+ private ShenyuWebsocketClient createSyncTestClient() {
+ ShenyuWebsocketClient client = createMockClient();
+ doReturn(URI.create("ws://localhost:9090")).when(client).getURI();
return client;
}
@@ -448,4 +461,264 @@ public class ShenyuWebsocketClientTest {
assertTrue(Thread.interrupted(), "Interrupt status should be preserved
after reconnect");
assertFalse(((AtomicBoolean) getField(client, "reconnecting")).get());
}
+
+ // ---------- sync failure recovery tests ----------
+
+ private WebsocketDataHandler getHandler(final ShenyuWebsocketClient
client) {
+ return (WebsocketDataHandler) getField(client, "websocketDataHandler");
+ }
+
+ @Test
+ void testSyncFailureClosesConnectionToTriggerResync() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doNothing().when(client).close();
+ doThrow(new RuntimeException("transient handler
failure")).when(getHandler(client))
+ .executor(any(), anyString(), anyString());
+ String json = GsonUtils.getInstance().toJson(websocketData);
+
+ client.onMessage(json);
+
+ verify(client, times(1)).close();
+ }
+
+ @Test
+ void testPoisonDataKeepsConnectionAfterBoundedFailures() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doNothing().when(client).close();
+ WebsocketDataHandler handler = getHandler(client);
+ doThrow(new RuntimeException("poison
data")).when(handler).executor(any(), anyString(), anyString());
+ String json = GsonUtils.getInstance().toJson(websocketData);
+
+ // failures 1 and 2 close the connection so the reconnect pulls a full
snapshot
+ client.onMessage(json);
+ client.onMessage(json);
+ verify(client, times(2)).close();
+ verify(client, never()).nowClose();
+
+ // at the cap: stop closing, keep the connection and the health check
alive
+ client.onMessage(json);
+ verify(client, times(2)).close();
+ verify(client, never()).nowClose();
+
+ // A single successful legacy payload cannot prove a full-cycle
recovery.
+ doNothing().when(handler).executor(any(), anyString(), anyString());
+ client.onMessage(json);
+ assertEquals(3, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ doThrow(new RuntimeException("poison
again")).when(handler).executor(any(), anyString(), anyString());
+ client.onMessage(json);
+ verify(client, times(2)).close();
+ verify(client, never()).nowClose();
+ }
+
+ @Test
+ void testNonJsonFrameIsIgnoredWithoutClosingConnection() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doNothing().when(client).close();
+ WebsocketDataHandler handler = getHandler(client);
+
+ // a frame the client cannot parse is ignored without dropping the
connection
+ // and without consuming the bounded recovery budget
+ client.onMessage("{invalid json");
+ verify(client, never()).close();
+ verify(handler, never()).executor(any(), anyString(), anyString());
+
+ String json = GsonUtils.getInstance().toJson(websocketData);
+ doThrow(new
RuntimeException("transient")).when(handler).executor(any(), anyString(),
anyString());
+ client.onMessage(json);
+ verify(client, times(1)).close();
+ }
+
+ @Test
+ void testUnknownEventDoesNotConsumeFailureBudget() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+
client.onMessage("{\"groupType\":\"RULE\",\"eventType\":\"FUTURE_EVENT\",\"data\":[{}]}");
+ verify(getHandler(client), never()).executor(any(), anyString(),
anyString());
+ verify(client, never()).close();
+ assertEquals(0, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ assertEquals(0L, getField(client, "nextSyncRetryAt"));
+ }
+
+ @Test
+ void testHealthCheckRetriesFullSyncAfterFailureCap() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doNothing().when(client).close();
+ doNothing().when(client).send(anyString());
+ WebsocketDataHandler handler = getHandler(client);
+ doThrow(new RuntimeException("poison
data")).when(handler).executor(any(), anyString(), anyString());
+ String json = GsonUtils.getInstance().toJson(websocketData);
+ client.onMessage(json);
+ client.onMessage(json);
+ client.onMessage(json);
+ long deadline = (long) getField(client, "nextSyncRetryAt");
+ assertTrue(deadline - System.nanoTime() > 0);
+ client.onMessage(json);
+ assertEquals(deadline, getField(client, "nextSyncRetryAt"));
+ invokePrivate(client, "healthCheck");
+ verify(client, times(2)).close();
+ setField(client, "nextSyncRetryAt", System.nanoTime() - 1);
+ invokePrivate(client, "healthCheck");
+ verify(client, times(3)).close();
+ client.onOpen(mock(ServerHandshake.class));
+ verify(client).send(DataEventTypeEnum.MYSELF.name());
+ doNothing().when(handler).executor(any(), anyString(), anyString());
+ client.onMessage(json);
+ assertEquals(3, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ verify(client, never()).nowClose();
+ }
+
+ @Test
+ void testGoodThenPoisonSnapshotReachesCooldownAndRecovers() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doReturn(false).when(client).isOpen();
+ AtomicBoolean ready = new AtomicBoolean();
+ InitialSyncState state = new InitialSyncState(ready, () ->
invokePrivate(client, "resetSyncFailures"));
+ setField(client, "initialSyncState", state);
+ AtomicReference<String> request = new AtomicReference<>();
+ doAnswer(invocation -> {
+ String message = invocation.getArgument(0);
+ if (message.startsWith(WebsocketSyncFrame.REQUEST_PREFIX)) {
+
request.set(message.substring(WebsocketSyncFrame.REQUEST_PREFIX.length()));
+ }
+ return null;
+ }).when(client).send(anyString());
+ String good =
"{\"groupType\":\"PLUGIN\",\"eventType\":\"REFRESH\",\"data\":[]}";
+ String poison =
"{\"groupType\":\"PLUGIN\",\"eventType\":\"REFRESH\",\"data\":[{\"id\":\"poison\"}]}";
+ doAnswer(invocation -> {
+ if (!"[]".equals(invocation.getArgument(1))) {
+ throw new IllegalStateException("same poison item on every
replay");
+ }
+ return null;
+ }).when(getHandler(client)).executor(any(), anyString(), anyString());
+ for (int attempt = 1; attempt <= 3; attempt++) {
+ client.onOpen(mock(ServerHandshake.class));
+ sendSyncFrame(client, request.get(), 0, good);
+ client.onMessage(good);
+ assertEquals(attempt - 1, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ sendSyncFrame(client, request.get(), 1, poison);
+ sendSyncFrame(client, request.get(), 2, null);
+ assertEquals(attempt, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ assertFalse(ready.get());
+ }
+ verify(client, times(2)).close();
+ long deadline = (long) getField(client, "nextSyncRetryAt");
+ assertTrue(deadline - System.nanoTime() > 0);
+ invokePrivate(client, "healthCheck");
+ verify(client, times(2)).close();
+ setField(client, "nextSyncRetryAt", System.nanoTime() - 1);
+ invokePrivate(client, "healthCheck");
+ verify(client, times(3)).close();
+ doNothing().when(getHandler(client)).executor(any(), anyString(),
anyString());
+ client.onOpen(mock(ServerHandshake.class));
+ sendSyncFrame(client, request.get(), 0, good);
+ sendSyncFrame(client, request.get(), 1, poison);
+ assertEquals(3, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ sendSyncFrame(client, request.get(), 2, null);
+ assertTrue(ready.get());
+ assertEquals(0, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ assertEquals(0L, getField(client, "nextSyncRetryAt"));
+ verify(client, never()).nowClose();
+ }
+
+ @Test
+ void testFullSyncBudgetResetsOnlyAfterDeferredApplicationCompletes() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ AtomicBoolean ready = new AtomicBoolean();
+ InitialSyncState state = new InitialSyncState(ready, () ->
invokePrivate(client, "resetSyncFailures"));
+ setField(client, "initialSyncState", state);
+ ((AtomicInteger) getField(client, "consecutiveSyncFailures")).set(2);
+ setField(client, "nextSyncRetryAt", System.nanoTime() +
60_000_000_000L);
+ CompletableFuture<Void> pending = new CompletableFuture<>();
+ doAnswer(invocation -> {
+ InitialSyncApplication.register(pending);
+ return null;
+ }).when(getHandler(client)).executor(any(), anyString(), anyString());
+ String request = state.begin();
+ sendSyncFrame(client, request, 0,
GsonUtils.getInstance().toJson(websocketData));
+ sendSyncFrame(client, request, 1, null);
+ assertFalse(ready.get());
+ assertEquals(2, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ pending.complete(null);
+ assertTrue(ready.get());
+ assertEquals(0, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ assertEquals(0L, getField(client, "nextSyncRetryAt"));
+ }
+
+ private void sendSyncFrame(final ShenyuWebsocketClient client, final
String request, final int sequence, final String payload) {
+ client.onMessage(GsonUtils.getInstance().toJson(new
WebsocketSyncFrame(request, sequence, payload)));
+ }
+
+ @Test
+ void testLegacyGoodThenPoisonSnapshotReachesCooldown() {
+ assertUnframedReplayIsBounded(false);
+ }
+
+ @Test
+ void testStandaloneGoodThenPoisonSnapshotReachesCooldown() {
+ assertUnframedReplayIsBounded(true);
+ }
+
+ private void assertUnframedReplayIsBounded(final boolean fullSnapshot) {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doReturn(false).when(client).isOpen();
+ doNothing().when(client).send(anyString());
+ setField(client, "namespaceId", "namespace-a");
+ WebsocketDataHandler handler = getHandler(client);
+ String envelope = "{\"groupType\":\"PLUGIN\",\"eventType\":\"MYSELF\","
+ + "\"namespaceId\":\"namespace-a\",\"fullSnapshot\":" +
fullSnapshot + ",\"data\":";
+ String good = envelope + "[]}";
+ String poison = envelope + "[{\"id\":\"poison\"}]}";
+ org.mockito.stubbing.Answer<Void> apply = invocation -> {
+ if (!"[]".equals(invocation.getArgument(1))) {
+ throw new IllegalStateException("same poison item on every
replay");
+ }
+ return null;
+ };
+ if (fullSnapshot) {
+ doAnswer(apply).when(handler).snapshot(any(), anyString(),
anyString(), anyString());
+ } else {
+ doAnswer(apply).when(handler).executor(any(), anyString(),
anyString());
+ }
+ for (int attempt = 1; attempt <= 3; attempt++) {
+ client.onOpen(mock(ServerHandshake.class));
+ client.onMessage(good);
+ assertEquals(attempt - 1, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ client.onMessage(poison);
+ assertEquals(attempt, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ }
+ verify(client, times(2)).close();
+ long deadline = (long) getField(client, "nextSyncRetryAt");
+ assertTrue(deadline - System.nanoTime() > 0);
+ client.onMessage(good);
+ assertEquals(3, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ assertEquals(deadline, getField(client, "nextSyncRetryAt"));
+ client.onMessage(poison);
+ assertEquals(deadline, getField(client, "nextSyncRetryAt"));
+ invokePrivate(client, "healthCheck");
+ verify(client, times(2)).close();
+ setField(client, "nextSyncRetryAt", System.nanoTime() - 1);
+ invokePrivate(client, "healthCheck");
+ verify(client, times(3)).close();
+ client.onOpen(mock(ServerHandshake.class));
+ verify(client, times(4)).send(DataEventTypeEnum.MYSELF.name());
+ verify(client, never()).nowClose();
+ }
+
+ @Test
+ void testSuccessfulLegacyPayloadPreservesFailureCounter() {
+ ShenyuWebsocketClient client = createSyncTestClient();
+ doNothing().when(client).close();
+ WebsocketDataHandler handler = getHandler(client);
+ String json = GsonUtils.getInstance().toJson(websocketData);
+
+ doThrow(new
RuntimeException("transient")).when(handler).executor(any(), anyString(),
anyString());
+ client.onMessage(json);
+ doNothing().when(handler).executor(any(), anyString(), anyString());
+ client.onMessage(json);
+ doThrow(new RuntimeException("transient
again")).when(handler).executor(any(), anyString(), anyString());
+ client.onMessage(json);
+
+ assertEquals(2, ((AtomicInteger) getField(client,
"consecutiveSyncFailures")).get());
+ verify(client, times(2)).close();
+ verify(client, never()).nowClose();
+ }
}
