From 3e3f0fb7e035c68b420db8fa4838a20f91b6ad9a Mon Sep 17 00:00:00 2001 From: Theo Date: Fri, 31 Jul 2026 14:43:26 -0700 Subject: [PATCH 1/3] Send server heartbeats so idle realtime connections stay alive The Minecraft plugin force-reconnects when it sees no inbound frame for 75s (MinecraftRealtimeClient#heartbeatTick). The backend never emitted an unsolicited frame: RealtimeCodec only built serverHello/error/ deployDrainAdvice, and client heartbeats were consumed without a reply (`case HEARTBEAT -> state.recordHeartbeat()`). So on a server with no punishments or notifications flowing, the plugin received exactly one frame ever -- the ServerHello -- and then silence. Its watchdog tripped on the first 25s tick past 75s, tore down a perfectly healthy connection, and reconnected, logging: [Realtime] No frames received within 75s; forcing reconnect roughly every 100 seconds, forever. Each cycle also dropped the plugin to fallback polling and re-ran a full baseline fetch. RealtimeCodec.heartbeat(long) previously existed but was never wired to a caller and was removed as dead code in 19217af, two minutes after the plugin-side watchdog landed in minecraft@326275e. Add RealtimeServerHeartbeatEmitter, a scheduled counterpart to RealtimeHeartbeatSweeper: the sweeper closes connections whose client heartbeats stopped, the emitter keeps client watchdogs fed. Default interval is 25s, giving ~3 heartbeats per 75s client window. The heartbeat envelope deliberately carries no event_id -- both the plugin and the panel transport-ACK any frame that has one, which would turn every keepalive into a request/response pair. Fixing this backend-side repairs every already-deployed plugin without requiring server operators to update their jar. Co-Authored-By: Claude Opus 5 (1M context) --- .../realtime/config/RealtimeProperties.java | 8 + .../RealtimeServerHeartbeatEmitter.java | 53 ++++++ .../state/RealtimeConnectionState.java | 6 + .../realtime/transport/RealtimeCodec.java | 21 +++ src/main/resources/application.properties | 2 + .../RealtimeServerHeartbeatEmitterTest.java | 167 ++++++++++++++++++ 6 files changed, 257 insertions(+) create mode 100644 src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java create mode 100644 src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java diff --git a/src/main/java/gg/modl/backend/realtime/config/RealtimeProperties.java b/src/main/java/gg/modl/backend/realtime/config/RealtimeProperties.java index 4c2d888..d4f962e 100644 --- a/src/main/java/gg/modl/backend/realtime/config/RealtimeProperties.java +++ b/src/main/java/gg/modl/backend/realtime/config/RealtimeProperties.java @@ -26,6 +26,14 @@ public class RealtimeProperties { @Min(5) private long heartbeatTimeoutSeconds = 60; + /** + * Interval between unsolicited server -> client heartbeats. Must stay comfortably below the + * client-side inbound liveness timeout (the Minecraft plugin force-reconnects after 75s of + * silence), so the default gives roughly three heartbeats per client window. + */ + @Min(1000) + private long serverHeartbeatIntervalMs = 25_000; + @Min(1) private long handshakeTimeoutSeconds = 10; diff --git a/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java b/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java new file mode 100644 index 0000000..16df4dd --- /dev/null +++ b/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java @@ -0,0 +1,53 @@ +package gg.modl.backend.realtime.schedule; + +import gg.modl.backend.realtime.config.RealtimeProperties; +import gg.modl.backend.realtime.state.RealtimeConnectionRegistry; +import gg.modl.backend.realtime.state.RealtimeConnectionState; +import gg.modl.backend.realtime.transport.RealtimeCodec; +import gg.modl.backend.realtime.transport.RealtimeSessionOperations; +import lombok.RequiredArgsConstructor; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.WebSocketSession; + +/** + * Emits unsolicited heartbeats to authenticated connections so idle sessions keep seeing inbound + * traffic. + * + *

Clients treat prolonged inbound silence as a dead connection. Without this, a connection that + * is healthy and fully subscribed but simply has no domain events to carry looks dead to the + * client: the Minecraft plugin tears it down and reconnects roughly every 100 seconds, re-running a + * full baseline fetch each cycle. {@link RealtimeHeartbeatSweeper} is the inbound counterpart — it + * closes connections whose client heartbeats have stopped.

+ */ +@Component +@RequiredArgsConstructor +public class RealtimeServerHeartbeatEmitter { + private final RealtimeProperties properties; + private final RealtimeConnectionRegistry connectionRegistry; + private final RealtimeCodec codec; + private final RealtimeSessionOperations sessionOperations; + + @Scheduled(fixedDelayString = "${modl.realtime.ws.server-heartbeat-interval-ms:25000}") + public void emitHeartbeats() { + if (!properties.isEnabled()) { + return; + } + + for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : connectionRegistry.snapshot()) { + RealtimeConnectionState state = snapshot.state(); + // Unauthenticated sessions are the handshake sweeper's business; sending to a closing or + // terminal session would only race its close frame. + if (!state.isAuthenticated() || state.isClosing() || state.getTerminalSince() != null) { + continue; + } + WebSocketSession session = snapshot.session(); + if (!session.isOpen()) { + continue; + } + // Best effort: a failed keepalive is not itself grounds for tearing down the connection. + // A genuinely dead peer stops sending client heartbeats and the sweeper closes it. + sessionOperations.trySend(session, state, codec.heartbeat(state.nextOutboundHeartbeatSequence())); + } + } +} diff --git a/src/main/java/gg/modl/backend/realtime/state/RealtimeConnectionState.java b/src/main/java/gg/modl/backend/realtime/state/RealtimeConnectionState.java index dc1d397..1386879 100644 --- a/src/main/java/gg/modl/backend/realtime/state/RealtimeConnectionState.java +++ b/src/main/java/gg/modl/backend/realtime/state/RealtimeConnectionState.java @@ -17,6 +17,7 @@ public class RealtimeConnectionState { private final Object sendLock = new Object(); private final AtomicLong deliveryAttempts = new AtomicLong(); private final AtomicLong deliveryFailures = new AtomicLong(); + private final AtomicLong outboundHeartbeatSequence = new AtomicLong(); private volatile Instant lastHeartbeat = Instant.now(); private volatile RealtimePrincipal principal; private volatile int protocolVersion; @@ -110,6 +111,11 @@ public void recordHeartbeat() { lastHeartbeat = Instant.now(); } + /** Sequence for the next unsolicited server -> client heartbeat on this connection. */ + public long nextOutboundHeartbeatSequence() { + return outboundHeartbeatSequence.incrementAndGet(); + } + public void setLastAcknowledgedEventId(@Nullable String lastAcknowledgedEventId) { this.lastAcknowledgedEventId = lastAcknowledgedEventId; } diff --git a/src/main/java/gg/modl/backend/realtime/transport/RealtimeCodec.java b/src/main/java/gg/modl/backend/realtime/transport/RealtimeCodec.java index 8ed1982..e25edd2 100644 --- a/src/main/java/gg/modl/backend/realtime/transport/RealtimeCodec.java +++ b/src/main/java/gg/modl/backend/realtime/transport/RealtimeCodec.java @@ -4,6 +4,7 @@ import com.google.protobuf.Timestamp; import gg.modl.backend.realtime.config.RealtimeProperties; import gg.modl.proto.modl.v1.ErrorCode; +import gg.modl.proto.modl.v1.Heartbeat; import gg.modl.proto.modl.v1.RealtimeEnvelope; import gg.modl.proto.modl.v1.ReconnectAction; import gg.modl.proto.modl.v1.ReconnectAdvice; @@ -44,6 +45,26 @@ public BinaryMessage serverHello(String connectionId, Collection accepted return toMessage(baseEnvelope().setServerHello(hello).build()); } + /** + * Unsolicited server -> client keepalive. Clients treat prolonged inbound silence as a dead + * connection (the Minecraft plugin force-reconnects after 75s without a frame), so idle + * connections need this even when no domain events are flowing. + * + *

Built without an {@code event_id} on purpose: both the plugin and the panel transport-ACK + * any frame carrying one, which would turn every keepalive into a request/response pair.

+ */ + public BinaryMessage heartbeat(long sequence) { + Instant now = Instant.now(); + return toMessage(RealtimeEnvelope.newBuilder() + .setProtocolVersion(properties.getProtocolVersion()) + .setTimestamp(Timestamp.newBuilder() + .setSeconds(now.getEpochSecond()) + .setNanos(now.getNano()) + .build()) + .setHeartbeat(Heartbeat.newBuilder().setSequence(sequence)) + .build()); + } + public BinaryMessage error(ErrorCode code, String message) { gg.modl.proto.modl.v1.Error error = gg.modl.proto.modl.v1.Error.newBuilder() .setCode(code) diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 3e49e6b..3d0e49d 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -49,6 +49,8 @@ modl.realtime.ws.protocol-version=${MODL_REALTIME_WS_PROTOCOL_VERSION:1} modl.realtime.ws.heartbeat-timeout-seconds=${MODL_REALTIME_WS_HEARTBEAT_TIMEOUT_SECONDS:60} modl.realtime.ws.handshake-timeout-seconds=${MODL_REALTIME_WS_HANDSHAKE_TIMEOUT_SECONDS:10} modl.realtime.ws.heartbeat-sweep-interval-ms=${MODL_REALTIME_WS_HEARTBEAT_SWEEP_INTERVAL_MS:15000} +# Must stay well under the client inbound-liveness timeout (plugin force-reconnects after 75s of silence) +modl.realtime.ws.server-heartbeat-interval-ms=${MODL_REALTIME_WS_SERVER_HEARTBEAT_INTERVAL_MS:25000} modl.realtime.ws.inbound-rate-limit-messages=${MODL_REALTIME_WS_INBOUND_RATE_LIMIT_MESSAGES:120} modl.realtime.ws.inbound-rate-limit-window-seconds=${MODL_REALTIME_WS_INBOUND_RATE_LIMIT_WINDOW_SECONDS:10} modl.realtime.ws.deploy-drain-close-code=${MODL_REALTIME_WS_DEPLOY_DRAIN_CLOSE_CODE:1012} diff --git a/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java new file mode 100644 index 0000000..487bdd9 --- /dev/null +++ b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java @@ -0,0 +1,167 @@ +package gg.modl.backend.realtime.schedule; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import gg.modl.backend.realtime.auth.RealtimePrincipal; +import gg.modl.backend.realtime.config.RealtimeProperties; +import gg.modl.backend.realtime.lifecycle.RealtimeConnectionCleanup; +import gg.modl.backend.realtime.metrics.RealtimeMetrics; +import gg.modl.backend.realtime.rate.RealtimeMessageRateLimiter; +import gg.modl.backend.realtime.state.RealtimeConnectionRegistry; +import gg.modl.backend.realtime.state.RealtimeConnectionState; +import gg.modl.backend.realtime.transport.RealtimeCodec; +import gg.modl.backend.realtime.transport.RealtimeSessionOperations; +import gg.modl.backend.server.data.Server; +import gg.modl.backend.server.data.ServerPlan; +import gg.modl.proto.modl.v1.RealtimeEnvelope; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.concurrent.ConcurrentHashMap; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.web.socket.BinaryMessage; +import org.springframework.web.socket.WebSocketSession; + +/** + * Guards the server -> client liveness contract. The Minecraft plugin force-reconnects when it sees + * no inbound frame for 75s, so the backend must emit unsolicited heartbeats on idle connections; + * without them a healthy, fully subscribed connection is torn down roughly every 100 seconds. + */ +class RealtimeServerHeartbeatEmitterTest { + + @Test + void sendsHeartbeatToAuthenticatedSession() throws Exception { + Fixture fixture = new Fixture(); + WebSocketSession session = fixture.openSession("authenticated"); + RealtimeConnectionState state = fixture.registry.register(session); + state.authenticate(RealtimePrincipal.minecraft(server()), 1); + + fixture.emitter.emitHeartbeats(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(BinaryMessage.class); + verify(session).sendMessage(captor.capture()); + RealtimeEnvelope envelope = RealtimeEnvelope.parseFrom(payload(captor.getValue())); + assertEquals(RealtimeEnvelope.PayloadCase.HEARTBEAT, envelope.getPayloadCase()); + } + + /** + * Both the plugin and the panel transport-ACK any frame carrying a non-empty event_id. An + * event_id on a heartbeat would therefore double every keepalive into a request/response pair. + */ + @Test + void heartbeatCarriesNoEventIdSoClientsDoNotAckIt() throws Exception { + Fixture fixture = new Fixture(); + WebSocketSession session = fixture.openSession("no-ack"); + RealtimeConnectionState state = fixture.registry.register(session); + state.authenticate(RealtimePrincipal.minecraft(server()), 1); + + fixture.emitter.emitHeartbeats(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(BinaryMessage.class); + verify(session).sendMessage(captor.capture()); + RealtimeEnvelope envelope = RealtimeEnvelope.parseFrom(payload(captor.getValue())); + assertTrue(envelope.getEventId().isEmpty(), "heartbeat must not carry an event_id"); + assertEquals(1, envelope.getProtocolVersion()); + } + + @Test + void heartbeatSequenceAdvancesPerConnection() throws Exception { + Fixture fixture = new Fixture(); + WebSocketSession session = fixture.openSession("sequenced"); + RealtimeConnectionState state = fixture.registry.register(session); + state.authenticate(RealtimePrincipal.minecraft(server()), 1); + + fixture.emitter.emitHeartbeats(); + fixture.emitter.emitHeartbeats(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(BinaryMessage.class); + verify(session, org.mockito.Mockito.times(2)).sendMessage(captor.capture()); + long first = RealtimeEnvelope.parseFrom(payload(captor.getAllValues().get(0))).getHeartbeat().getSequence(); + long second = RealtimeEnvelope.parseFrom(payload(captor.getAllValues().get(1))).getHeartbeat().getSequence(); + assertEquals(1L, first); + assertEquals(2L, second); + } + + @Test + void doesNotHeartbeatUnauthenticatedSession() throws Exception { + Fixture fixture = new Fixture(); + WebSocketSession session = fixture.openSession("unauthenticated"); + fixture.registry.register(session); + + fixture.emitter.emitHeartbeats(); + + verify(session, never()).sendMessage(any(BinaryMessage.class)); + } + + @Test + void doesNotHeartbeatTerminalSession() throws Exception { + Fixture fixture = new Fixture(); + WebSocketSession session = fixture.openSession("terminal"); + RealtimeConnectionState state = fixture.registry.register(session); + state.authenticate(RealtimePrincipal.minecraft(server()), 1); + state.markClosing(); + state.markTerminal(); + + fixture.emitter.emitHeartbeats(); + + verify(session, never()).sendMessage(any(BinaryMessage.class)); + } + + @Test + void doesNothingWhenRealtimeDisabled() throws Exception { + Fixture fixture = new Fixture(); + fixture.properties.setEnabled(false); + WebSocketSession session = fixture.openSession("disabled"); + RealtimeConnectionState state = fixture.registry.register(session); + state.authenticate(RealtimePrincipal.minecraft(server()), 1); + + fixture.emitter.emitHeartbeats(); + + verify(session, never()).sendMessage(any(BinaryMessage.class)); + } + + private static final class Fixture { + private final RealtimeProperties properties = new RealtimeProperties(); + private final RealtimeConnectionRegistry registry; + private final RealtimeServerHeartbeatEmitter emitter; + + private Fixture() { + properties.setEnabled(true); + registry = new RealtimeConnectionRegistry(properties); + RealtimeMetrics metrics = new RealtimeMetrics(new SimpleMeterRegistry()); + RealtimeConnectionCleanup cleanup = + new RealtimeConnectionCleanup(registry, new RealtimeMessageRateLimiter(properties), metrics); + emitter = new RealtimeServerHeartbeatEmitter( + properties, + registry, + new RealtimeCodec(properties), + new RealtimeSessionOperations(registry, cleanup, metrics) + ); + } + + private WebSocketSession openSession(String id) { + WebSocketSession session = mock(WebSocketSession.class); + when(session.getId()).thenReturn(id); + when(session.isOpen()).thenReturn(true); + when(session.getAttributes()).thenReturn(new ConcurrentHashMap<>()); + return session; + } + } + + private static byte[] payload(BinaryMessage message) { + byte[] payload = new byte[message.getPayloadLength()]; + message.getPayload().get(payload); + return payload; + } + + private static Server server() { + Server server = new Server("server", "server", "server_db", "admin@example.com", true, ServerPlan.FREE); + server.setId("server-id"); + return server; + } +} From 42eeea12a0a08d172b7b3cba4354d9195ee3baf2 Mon Sep 17 00:00:00 2001 From: Theo Date: Fri, 31 Jul 2026 15:09:20 -0700 Subject: [PATCH 2/3] Dispatch realtime heartbeats per server instead of sending inline The sweep wrote to every connection in the process on the scheduler thread. A WebSocket write to a peer that has stopped reading blocks: ConcurrentWebSocketSessionDecorator only buffers for *concurrent* senders (tryFlushMessageBuffer uses a non-blocking tryLock), so the sole sender always takes the blocking path through StandardWebSocketSession -> getBasicRemote().sendBinary(), bounded only by Tomcat's ~20s BLOCKING_SEND_TIMEOUT. One wedged client therefore stalled every connection behind it in the sweep -- across all tenants -- starving healthy clients of the very heartbeat that keeps their 75s watchdog quiet. That reintroduces the reconnect storm this component exists to prevent. Hand delivery to RealtimeDispatchExecutor keyed by server, matching how InProcessRealtimeEventDispatcher already fans out domain events: the stall stays inside the offending tenant's shard and off the scheduler thread. Eligibility is re-checked inside the task since a connection can close between the sweep and the task reaching its worker. Tests: doesNotWriteOnTheSweepThread and partitionsDeliveryPerServerSoOneTenantCannotStallAnother pin the contract structurally; wedgedTenantDoesNotDelayHeartbeatForAnotherTenant exercises the real sharded executor with a genuinely blocked write. All three were confirmed to fail against the previous inline implementation. Co-Authored-By: Claude Opus 5 (1M context) --- .../RealtimeServerHeartbeatEmitter.java | 48 ++++- .../RealtimeServerHeartbeatEmitterTest.java | 178 ++++++++++++++++-- 2 files changed, 205 insertions(+), 21 deletions(-) diff --git a/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java b/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java index 16df4dd..c5a93f2 100644 --- a/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java +++ b/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java @@ -1,10 +1,15 @@ package gg.modl.backend.realtime.schedule; import gg.modl.backend.realtime.config.RealtimeProperties; +import gg.modl.backend.realtime.dispatch.RealtimeDispatchExecutor; import gg.modl.backend.realtime.state.RealtimeConnectionRegistry; import gg.modl.backend.realtime.state.RealtimeConnectionState; import gg.modl.backend.realtime.transport.RealtimeCodec; import gg.modl.backend.realtime.transport.RealtimeSessionOperations; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import lombok.RequiredArgsConstructor; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @@ -19,6 +24,13 @@ * client: the Minecraft plugin tears it down and reconnects roughly every 100 seconds, re-running a * full baseline fetch each cycle. {@link RealtimeHeartbeatSweeper} is the inbound counterpart — it * closes connections whose client heartbeats have stopped.

+ * + *

Delivery is handed to {@link RealtimeDispatchExecutor} per server rather than sent inline. A + * WebSocket write to a peer that has stopped reading blocks until the container's blocking-send + * timeout (~20s), so sending inline would let one wedged client delay every connection behind it in + * the sweep — starving healthy clients of the very heartbeat that keeps their watchdog quiet. The + * sharded executor keeps that stall inside the offending tenant and off the scheduler thread, which + * is how outbound domain events are already delivered.

*/ @Component @RequiredArgsConstructor @@ -27,6 +39,7 @@ public class RealtimeServerHeartbeatEmitter { private final RealtimeConnectionRegistry connectionRegistry; private final RealtimeCodec codec; private final RealtimeSessionOperations sessionOperations; + private final RealtimeDispatchExecutor dispatchExecutor; @Scheduled(fixedDelayString = "${modl.realtime.ws.server-heartbeat-interval-ms:25000}") public void emitHeartbeats() { @@ -34,20 +47,41 @@ public void emitHeartbeats() { return; } + Map> byServer = new LinkedHashMap<>(); for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : connectionRegistry.snapshot()) { - RealtimeConnectionState state = snapshot.state(); - // Unauthenticated sessions are the handshake sweeper's business; sending to a closing or - // terminal session would only race its close frame. - if (!state.isAuthenticated() || state.isClosing() || state.getTerminalSince() != null) { + String serverId = snapshot.state().getServerId(); + if (serverId == null || !isEligible(snapshot)) { continue; } - WebSocketSession session = snapshot.session(); - if (!session.isOpen()) { + byServer.computeIfAbsent(serverId, key -> new ArrayList<>()).add(snapshot); + } + + byServer.forEach((serverId, snapshots) -> dispatchExecutor.execute(serverId, () -> sendAll(snapshots))); + } + + private void sendAll(List snapshots) { + for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : snapshots) { + // Re-checked here because the connection may have closed between the sweep and this task + // reaching the front of its worker queue. + if (!isEligible(snapshot)) { continue; } + RealtimeConnectionState state = snapshot.state(); // Best effort: a failed keepalive is not itself grounds for tearing down the connection. // A genuinely dead peer stops sending client heartbeats and the sweeper closes it. - sessionOperations.trySend(session, state, codec.heartbeat(state.nextOutboundHeartbeatSequence())); + sessionOperations.trySend( + snapshot.session(), state, codec.heartbeat(state.nextOutboundHeartbeatSequence())); } } + + private boolean isEligible(RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot) { + RealtimeConnectionState state = snapshot.state(); + WebSocketSession session = snapshot.session(); + // Unauthenticated sessions are the handshake sweeper's business; sending to a closing or + // terminal session would only race its close frame. + return state.isAuthenticated() + && !state.isClosing() + && state.getTerminalSince() == null + && session.isOpen(); + } } diff --git a/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java index 487bdd9..1592585 100644 --- a/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java +++ b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java @@ -1,15 +1,20 @@ package gg.modl.backend.realtime.schedule; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import gg.modl.backend.realtime.auth.RealtimePrincipal; import gg.modl.backend.realtime.config.RealtimeProperties; +import gg.modl.backend.realtime.dispatch.RealtimeDispatchExecutor; import gg.modl.backend.realtime.lifecycle.RealtimeConnectionCleanup; import gg.modl.backend.realtime.metrics.RealtimeMetrics; import gg.modl.backend.realtime.rate.RealtimeMessageRateLimiter; @@ -21,7 +26,11 @@ import gg.modl.backend.server.data.ServerPlan; import gg.modl.proto.modl.v1.RealtimeEnvelope; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.springframework.web.socket.BinaryMessage; @@ -39,7 +48,7 @@ void sendsHeartbeatToAuthenticatedSession() throws Exception { Fixture fixture = new Fixture(); WebSocketSession session = fixture.openSession("authenticated"); RealtimeConnectionState state = fixture.registry.register(session); - state.authenticate(RealtimePrincipal.minecraft(server()), 1); + state.authenticate(RealtimePrincipal.minecraft(server("server-id")), 1); fixture.emitter.emitHeartbeats(); @@ -58,7 +67,7 @@ void heartbeatCarriesNoEventIdSoClientsDoNotAckIt() throws Exception { Fixture fixture = new Fixture(); WebSocketSession session = fixture.openSession("no-ack"); RealtimeConnectionState state = fixture.registry.register(session); - state.authenticate(RealtimePrincipal.minecraft(server()), 1); + state.authenticate(RealtimePrincipal.minecraft(server("server-id")), 1); fixture.emitter.emitHeartbeats(); @@ -74,13 +83,13 @@ void heartbeatSequenceAdvancesPerConnection() throws Exception { Fixture fixture = new Fixture(); WebSocketSession session = fixture.openSession("sequenced"); RealtimeConnectionState state = fixture.registry.register(session); - state.authenticate(RealtimePrincipal.minecraft(server()), 1); + state.authenticate(RealtimePrincipal.minecraft(server("server-id")), 1); fixture.emitter.emitHeartbeats(); fixture.emitter.emitHeartbeats(); ArgumentCaptor captor = ArgumentCaptor.forClass(BinaryMessage.class); - verify(session, org.mockito.Mockito.times(2)).sendMessage(captor.capture()); + verify(session, times(2)).sendMessage(captor.capture()); long first = RealtimeEnvelope.parseFrom(payload(captor.getAllValues().get(0))).getHeartbeat().getSequence(); long second = RealtimeEnvelope.parseFrom(payload(captor.getAllValues().get(1))).getHeartbeat().getSequence(); assertEquals(1L, first); @@ -103,7 +112,7 @@ void doesNotHeartbeatTerminalSession() throws Exception { Fixture fixture = new Fixture(); WebSocketSession session = fixture.openSession("terminal"); RealtimeConnectionState state = fixture.registry.register(session); - state.authenticate(RealtimePrincipal.minecraft(server()), 1); + state.authenticate(RealtimePrincipal.minecraft(server("server-id")), 1); state.markClosing(); state.markTerminal(); @@ -118,50 +127,191 @@ void doesNothingWhenRealtimeDisabled() throws Exception { fixture.properties.setEnabled(false); WebSocketSession session = fixture.openSession("disabled"); RealtimeConnectionState state = fixture.registry.register(session); - state.authenticate(RealtimePrincipal.minecraft(server()), 1); + state.authenticate(RealtimePrincipal.minecraft(server("server-id")), 1); fixture.emitter.emitHeartbeats(); verify(session, never()).sendMessage(any(BinaryMessage.class)); } + /** + * A WebSocket write to a peer that stopped reading blocks for the container's blocking-send + * timeout, so the sweep must never write on its own thread — one wedged client would otherwise + * delay every connection behind it past the very 75s watchdog this component exists to satisfy. + */ + @Test + void doesNotWriteOnTheSweepThread() throws Exception { + Fixture fixture = new Fixture(false); + WebSocketSession session = fixture.openSession("deferred"); + RealtimeConnectionState state = fixture.registry.register(session); + state.authenticate(RealtimePrincipal.minecraft(server("server-id")), 1); + + fixture.emitter.emitHeartbeats(); + + verify(session, never()).sendMessage(any(BinaryMessage.class)); + assertEquals(1, fixture.deferred.size(), "heartbeat delivery must be handed to the dispatch executor"); + + fixture.deferred.forEach(Runnable::run); + verify(session).sendMessage(any(BinaryMessage.class)); + } + + @Test + void partitionsDeliveryPerServerSoOneTenantCannotStallAnother() { + Fixture fixture = new Fixture(false); + WebSocketSession first = fixture.openSession("tenant-a"); + fixture.registry.register(first).authenticate(RealtimePrincipal.minecraft(server("server-a")), 1); + WebSocketSession second = fixture.openSession("tenant-b"); + fixture.registry.register(second).authenticate(RealtimePrincipal.minecraft(server("server-b")), 1); + + fixture.emitter.emitHeartbeats(); + + ArgumentCaptor serverIds = ArgumentCaptor.forClass(String.class); + verify(fixture.dispatchExecutor, times(2)).execute(serverIds.capture(), any(Runnable.class)); + assertEquals(List.of("server-a", "server-b"), serverIds.getAllValues().stream().sorted().toList()); + } + + /** + * End-to-end guard on the real sharded executor: a tenant whose socket write is wedged must not + * hold up a tenant on another shard. + * + *

The wedged connection is deliberately placed first in the sweep order. Registry + * iteration is hash-ordered, so relying on natural order would let this pass against a purely + * sequential implementation whenever the healthy tenant happened to sort first.

+ */ + @Test + void wedgedTenantDoesNotDelayHeartbeatForAnotherTenant() throws Exception { + RealtimeProperties properties = new RealtimeProperties(); + properties.setEnabled(true); + properties.setDispatchWorkers(4); + properties.setDispatchQueueCapacity(64); + String wedgedServer = "server-0"; + String healthyServer = distinctShardServerId(wedgedServer, properties.getDispatchWorkers()); + + CountDownLatch releaseWedged = new CountDownLatch(1); + CountDownLatch healthyReceived = new CountDownLatch(1); + + WebSocketSession wedged = openMockSession("wedged"); + doAnswer(invocation -> { + releaseWedged.await(10, TimeUnit.SECONDS); + return null; + }).when(wedged).sendMessage(any(BinaryMessage.class)); + RealtimeConnectionState wedgedState = new RealtimeConnectionState(); + wedgedState.authenticate(RealtimePrincipal.minecraft(server(wedgedServer)), 1); + + WebSocketSession healthy = openMockSession("healthy"); + doAnswer(invocation -> { + healthyReceived.countDown(); + return null; + }).when(healthy).sendMessage(any(BinaryMessage.class)); + RealtimeConnectionState healthyState = new RealtimeConnectionState(); + healthyState.authenticate(RealtimePrincipal.minecraft(server(healthyServer)), 1); + + // Fixed sweep order: wedged tenant first, so a sequential implementation cannot pass. + RealtimeConnectionRegistry registry = mock(RealtimeConnectionRegistry.class); + when(registry.snapshot()).thenReturn(List.of( + new RealtimeConnectionRegistry.RealtimeConnectionSnapshot("wedged", wedged, wedgedState), + new RealtimeConnectionRegistry.RealtimeConnectionSnapshot("healthy", healthy, healthyState))); + when(registry.getSession(any(WebSocketSession.class))) + .thenAnswer(invocation -> java.util.Optional.of(invocation.getArgument(0))); + when(registry.isTerminal(any(WebSocketSession.class))).thenReturn(false); + + RealtimeMetrics metrics = new RealtimeMetrics(new SimpleMeterRegistry()); + RealtimeConnectionCleanup cleanup = + new RealtimeConnectionCleanup(registry, new RealtimeMessageRateLimiter(properties), metrics); + RealtimeDispatchExecutor dispatchExecutor = + new RealtimeDispatchExecutor(properties, new SimpleMeterRegistry()); + RealtimeServerHeartbeatEmitter emitter = new RealtimeServerHeartbeatEmitter( + properties, registry, new RealtimeCodec(properties), + new RealtimeSessionOperations(registry, cleanup, metrics), dispatchExecutor); + + // Driven from a separate thread so the assertion runs *while* the wedged write is still + // blocked. Asserting after emitHeartbeats() returns would pass against a sequential + // implementation too, since by then the delayed send has already completed. + Thread sweep = new Thread(emitter::emitHeartbeats, "heartbeat-sweep"); + sweep.setDaemon(true); + try { + sweep.start(); + + assertTrue( + healthyReceived.await(5, TimeUnit.SECONDS), + "healthy tenant must receive its heartbeat while another tenant's write is wedged"); + } finally { + releaseWedged.countDown(); + sweep.join(TimeUnit.SECONDS.toMillis(15)); + dispatchExecutor.shutdown(); + } + } + + /** Finds a server id that lands on a different dispatch shard than {@code other}. */ + private static String distinctShardServerId(String other, int workers) { + int otherShard = Math.floorMod(other.hashCode(), workers); + for (int index = 1; index < 1000; index++) { + String candidate = "server-" + index; + if (Math.floorMod(candidate.hashCode(), workers) != otherShard) { + assertNotEquals(other, candidate); + return candidate; + } + } + throw new IllegalStateException("no server id found on a different shard"); + } + private static final class Fixture { private final RealtimeProperties properties = new RealtimeProperties(); private final RealtimeConnectionRegistry registry; + private final RealtimeDispatchExecutor dispatchExecutor = mock(RealtimeDispatchExecutor.class); + private final List deferred = new ArrayList<>(); private final RealtimeServerHeartbeatEmitter emitter; private Fixture() { + this(true); + } + + private Fixture(boolean runInline) { properties.setEnabled(true); registry = new RealtimeConnectionRegistry(properties); RealtimeMetrics metrics = new RealtimeMetrics(new SimpleMeterRegistry()); RealtimeConnectionCleanup cleanup = new RealtimeConnectionCleanup(registry, new RealtimeMessageRateLimiter(properties), metrics); + doAnswer(invocation -> { + Runnable task = invocation.getArgument(1); + if (runInline) { + task.run(); + } else { + deferred.add(task); + } + return null; + }).when(dispatchExecutor).execute(anyString(), any(Runnable.class)); emitter = new RealtimeServerHeartbeatEmitter( properties, registry, new RealtimeCodec(properties), - new RealtimeSessionOperations(registry, cleanup, metrics) + new RealtimeSessionOperations(registry, cleanup, metrics), + dispatchExecutor ); } private WebSocketSession openSession(String id) { - WebSocketSession session = mock(WebSocketSession.class); - when(session.getId()).thenReturn(id); - when(session.isOpen()).thenReturn(true); - when(session.getAttributes()).thenReturn(new ConcurrentHashMap<>()); - return session; + return openMockSession(id); } } + private static WebSocketSession openMockSession(String id) { + WebSocketSession session = mock(WebSocketSession.class); + when(session.getId()).thenReturn(id); + when(session.isOpen()).thenReturn(true); + when(session.getAttributes()).thenReturn(new ConcurrentHashMap<>()); + return session; + } + private static byte[] payload(BinaryMessage message) { byte[] payload = new byte[message.getPayloadLength()]; message.getPayload().get(payload); return payload; } - private static Server server() { + private static Server server(String id) { Server server = new Server("server", "server", "server_db", "admin@example.com", true, ServerPlan.FREE); - server.setId("server-id"); + server.setId(id); return server; } } From 434dc90e809215519f65d27c831a94237f967af0 Mon Sep 17 00:00:00 2001 From: Theo Date: Fri, 31 Jul 2026 15:32:10 -0700 Subject: [PATCH 3/3] Deliver realtime heartbeats per connection on virtual threads Sharding heartbeat delivery by server narrowed the stall but did not remove it. dispatchWorkers caps at min(cores, 8), so with more than eight tenants unrelated servers collide on a worker by pigeonhole: one wedged peer's blocking write (~20s, Tomcat BLOCKING_SEND_TIMEOUT) stalls every other server batched behind it on that single-threaded worker. Enough collisions and a healthy client crosses its 75s inbound-liveness window and force-reconnects -- the exact failure this component prevents. Sharing the dispatch executor also coupled keepalives to domain-event delivery: a wedged heartbeat could hold up real punishment and notification fan-out for every tenant on the shard, and heartbeats have no ordering requirement that justified using an order-preserving per-tenant queue. Any delivery granularity coarser than one connection leaves head-of-line blocking, so dispatch each connection independently onto its own virtual thread. That matches the granularity of the per-connection send lock in RealtimeSessionOperations, so a wedged peer can only ever stall itself. Java 21 virtual threads make it cheap: the blocking write parks and unmounts its carrier instead of occupying a pooled thread. The sweep is also simpler -- no grouping, no shard keying. The package-private constructor that injects the executor for tests makes Spring's constructor set ambiguous, which fails at context startup with "No default constructor found" rather than at compile time. The injection point is annotated explicitly and pinned by RealtimeServerHeartbeatEmitterWiringTest, since no other test boots a context. submitsOneTaskPerConnection and wedgedConnectionDoesNotDelayHeartbeatFor- Another were both confirmed to fail against the previous per-server batching, and the wiring test against an unannotated constructor. Co-Authored-By: Claude Opus 5 (1M context) --- .../RealtimeServerHeartbeatEmitter.java | 90 +++++++++++++------ .../RealtimeServerHeartbeatEmitterTest.java | 88 +++++++----------- ...ltimeServerHeartbeatEmitterWiringTest.java | 34 +++++++ 3 files changed, 127 insertions(+), 85 deletions(-) create mode 100644 src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterWiringTest.java diff --git a/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java b/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java index c5a93f2..fd48043 100644 --- a/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java +++ b/src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java @@ -1,16 +1,16 @@ package gg.modl.backend.realtime.schedule; import gg.modl.backend.realtime.config.RealtimeProperties; -import gg.modl.backend.realtime.dispatch.RealtimeDispatchExecutor; import gg.modl.backend.realtime.state.RealtimeConnectionRegistry; import gg.modl.backend.realtime.state.RealtimeConnectionState; import gg.modl.backend.realtime.transport.RealtimeCodec; import gg.modl.backend.realtime.transport.RealtimeSessionOperations; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import lombok.RequiredArgsConstructor; +import jakarta.annotation.PreDestroy; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.springframework.web.socket.WebSocketSession; @@ -25,21 +25,50 @@ * full baseline fetch each cycle. {@link RealtimeHeartbeatSweeper} is the inbound counterpart — it * closes connections whose client heartbeats have stopped.

* - *

Delivery is handed to {@link RealtimeDispatchExecutor} per server rather than sent inline. A - * WebSocket write to a peer that has stopped reading blocks until the container's blocking-send - * timeout (~20s), so sending inline would let one wedged client delay every connection behind it in - * the sweep — starving healthy clients of the very heartbeat that keeps their watchdog quiet. The - * sharded executor keeps that stall inside the offending tenant and off the scheduler thread, which - * is how outbound domain events are already delivered.

+ *

Each connection is delivered on its own virtual thread. A WebSocket write to a peer that has + * stopped reading blocks until the container's blocking-send timeout (~20s), so any shared + * delivery thread lets one wedged peer delay the connections queued behind it past the very 75s + * watchdog this component exists to satisfy — including a per-tenant sharded pool, where unrelated + * servers collide on the same worker. Per-connection dispatch matches the granularity of the + * per-connection send lock in {@link RealtimeSessionOperations}, so a wedged peer can only stall + * itself. Virtual threads make that isolation cheap: a blocked socket write parks and unmounts its + * carrier rather than occupying a pooled thread.

*/ +@Slf4j @Component -@RequiredArgsConstructor public class RealtimeServerHeartbeatEmitter { private final RealtimeProperties properties; private final RealtimeConnectionRegistry connectionRegistry; private final RealtimeCodec codec; private final RealtimeSessionOperations sessionOperations; - private final RealtimeDispatchExecutor dispatchExecutor; + private final Executor deliveryExecutor; + + // Explicit: the package-private test constructor below makes the candidate set ambiguous, and + // Spring falls back to a (non-existent) no-arg constructor unless one is annotated. + @Autowired + public RealtimeServerHeartbeatEmitter( + RealtimeProperties properties, + RealtimeConnectionRegistry connectionRegistry, + RealtimeCodec codec, + RealtimeSessionOperations sessionOperations + ) { + this(properties, connectionRegistry, codec, sessionOperations, + Executors.newVirtualThreadPerTaskExecutor()); + } + + RealtimeServerHeartbeatEmitter( + RealtimeProperties properties, + RealtimeConnectionRegistry connectionRegistry, + RealtimeCodec codec, + RealtimeSessionOperations sessionOperations, + Executor deliveryExecutor + ) { + this.properties = properties; + this.connectionRegistry = connectionRegistry; + this.codec = codec; + this.sessionOperations = sessionOperations; + this.deliveryExecutor = deliveryExecutor; + } @Scheduled(fixedDelayString = "${modl.realtime.ws.server-heartbeat-interval-ms:25000}") public void emitHeartbeats() { @@ -47,30 +76,28 @@ public void emitHeartbeats() { return; } - Map> byServer = new LinkedHashMap<>(); for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : connectionRegistry.snapshot()) { - String serverId = snapshot.state().getServerId(); - if (serverId == null || !isEligible(snapshot)) { + if (!isEligible(snapshot)) { continue; } - byServer.computeIfAbsent(serverId, key -> new ArrayList<>()).add(snapshot); + deliveryExecutor.execute(() -> send(snapshot)); } - - byServer.forEach((serverId, snapshots) -> dispatchExecutor.execute(serverId, () -> sendAll(snapshots))); } - private void sendAll(List snapshots) { - for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : snapshots) { - // Re-checked here because the connection may have closed between the sweep and this task - // reaching the front of its worker queue. - if (!isEligible(snapshot)) { - continue; - } - RealtimeConnectionState state = snapshot.state(); + private void send(RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot) { + // Re-checked here because the connection may have closed between the sweep and this task + // being scheduled. + if (!isEligible(snapshot)) { + return; + } + RealtimeConnectionState state = snapshot.state(); + try { // Best effort: a failed keepalive is not itself grounds for tearing down the connection. // A genuinely dead peer stops sending client heartbeats and the sweeper closes it. sessionOperations.trySend( snapshot.session(), state, codec.heartbeat(state.nextOutboundHeartbeatSequence())); + } catch (RuntimeException exception) { + log.warn("Failed to send realtime heartbeat connection={}", state.getConnectionId(), exception); } } @@ -84,4 +111,11 @@ private boolean isEligible(RealtimeConnectionRegistry.RealtimeConnectionSnapshot && state.getTerminalSince() == null && session.isOpen(); } + + @PreDestroy + public void shutdown() { + if (deliveryExecutor instanceof ExecutorService service) { + service.shutdownNow(); + } + } } diff --git a/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java index 1592585..6cdc837 100644 --- a/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java +++ b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java @@ -1,10 +1,8 @@ package gg.modl.backend.realtime.schedule; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -14,7 +12,6 @@ import gg.modl.backend.realtime.auth.RealtimePrincipal; import gg.modl.backend.realtime.config.RealtimeProperties; -import gg.modl.backend.realtime.dispatch.RealtimeDispatchExecutor; import gg.modl.backend.realtime.lifecycle.RealtimeConnectionCleanup; import gg.modl.backend.realtime.metrics.RealtimeMetrics; import gg.modl.backend.realtime.rate.RealtimeMessageRateLimiter; @@ -28,8 +25,11 @@ import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import java.util.ArrayList; import java.util.List; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -149,43 +149,44 @@ void doesNotWriteOnTheSweepThread() throws Exception { fixture.emitter.emitHeartbeats(); verify(session, never()).sendMessage(any(BinaryMessage.class)); - assertEquals(1, fixture.deferred.size(), "heartbeat delivery must be handed to the dispatch executor"); + assertEquals(1, fixture.deferred.size(), "heartbeat delivery must be handed to the executor"); fixture.deferred.forEach(Runnable::run); verify(session).sendMessage(any(BinaryMessage.class)); } + /** + * Delivery must be submitted per connection, not batched per server or per shard. Any coarser + * grouping puts unrelated connections behind a single blocking write. + */ @Test - void partitionsDeliveryPerServerSoOneTenantCannotStallAnother() { + void submitsOneTaskPerConnection() { Fixture fixture = new Fixture(false); - WebSocketSession first = fixture.openSession("tenant-a"); + WebSocketSession first = fixture.openSession("conn-a"); fixture.registry.register(first).authenticate(RealtimePrincipal.minecraft(server("server-a")), 1); - WebSocketSession second = fixture.openSession("tenant-b"); - fixture.registry.register(second).authenticate(RealtimePrincipal.minecraft(server("server-b")), 1); + WebSocketSession second = fixture.openSession("conn-b"); + fixture.registry.register(second).authenticate(RealtimePrincipal.minecraft(server("server-a")), 1); + WebSocketSession third = fixture.openSession("conn-c"); + fixture.registry.register(third).authenticate(RealtimePrincipal.minecraft(server("server-b")), 1); fixture.emitter.emitHeartbeats(); - ArgumentCaptor serverIds = ArgumentCaptor.forClass(String.class); - verify(fixture.dispatchExecutor, times(2)).execute(serverIds.capture(), any(Runnable.class)); - assertEquals(List.of("server-a", "server-b"), serverIds.getAllValues().stream().sorted().toList()); + assertEquals(3, fixture.deferred.size(), "each connection must be dispatched independently"); } /** - * End-to-end guard on the real sharded executor: a tenant whose socket write is wedged must not - * hold up a tenant on another shard. + * End-to-end guard with a genuinely blocked socket write. * - *

The wedged connection is deliberately placed first in the sweep order. Registry - * iteration is hash-ordered, so relying on natural order would let this pass against a purely - * sequential implementation whenever the healthy tenant happened to sort first.

+ *

Both connections share one server, which is the worst case: it is what a per-tenant sharded + * pool groups into a single sequential batch, and it subsumes the case of two different servers + * colliding on the same shard. The wedged connection is placed first in the sweep order, and the + * assertion runs while its write is still blocked — asserting after the sweep returns would pass + * against a sequential implementation too, since by then the delayed send has completed.

*/ @Test - void wedgedTenantDoesNotDelayHeartbeatForAnotherTenant() throws Exception { + void wedgedConnectionDoesNotDelayHeartbeatForAnother() throws Exception { RealtimeProperties properties = new RealtimeProperties(); properties.setEnabled(true); - properties.setDispatchWorkers(4); - properties.setDispatchQueueCapacity(64); - String wedgedServer = "server-0"; - String healthyServer = distinctShardServerId(wedgedServer, properties.getDispatchWorkers()); CountDownLatch releaseWedged = new CountDownLatch(1); CountDownLatch healthyReceived = new CountDownLatch(1); @@ -196,7 +197,7 @@ void wedgedTenantDoesNotDelayHeartbeatForAnotherTenant() throws Exception { return null; }).when(wedged).sendMessage(any(BinaryMessage.class)); RealtimeConnectionState wedgedState = new RealtimeConnectionState(); - wedgedState.authenticate(RealtimePrincipal.minecraft(server(wedgedServer)), 1); + wedgedState.authenticate(RealtimePrincipal.minecraft(server("shared-server")), 1); WebSocketSession healthy = openMockSession("healthy"); doAnswer(invocation -> { @@ -204,29 +205,25 @@ void wedgedTenantDoesNotDelayHeartbeatForAnotherTenant() throws Exception { return null; }).when(healthy).sendMessage(any(BinaryMessage.class)); RealtimeConnectionState healthyState = new RealtimeConnectionState(); - healthyState.authenticate(RealtimePrincipal.minecraft(server(healthyServer)), 1); + healthyState.authenticate(RealtimePrincipal.minecraft(server("shared-server")), 1); - // Fixed sweep order: wedged tenant first, so a sequential implementation cannot pass. + // Fixed sweep order: wedged connection first, so a sequential implementation cannot pass. RealtimeConnectionRegistry registry = mock(RealtimeConnectionRegistry.class); when(registry.snapshot()).thenReturn(List.of( new RealtimeConnectionRegistry.RealtimeConnectionSnapshot("wedged", wedged, wedgedState), new RealtimeConnectionRegistry.RealtimeConnectionSnapshot("healthy", healthy, healthyState))); when(registry.getSession(any(WebSocketSession.class))) - .thenAnswer(invocation -> java.util.Optional.of(invocation.getArgument(0))); + .thenAnswer(invocation -> Optional.of(invocation.getArgument(0))); when(registry.isTerminal(any(WebSocketSession.class))).thenReturn(false); RealtimeMetrics metrics = new RealtimeMetrics(new SimpleMeterRegistry()); RealtimeConnectionCleanup cleanup = new RealtimeConnectionCleanup(registry, new RealtimeMessageRateLimiter(properties), metrics); - RealtimeDispatchExecutor dispatchExecutor = - new RealtimeDispatchExecutor(properties, new SimpleMeterRegistry()); + ExecutorService deliveryExecutor = Executors.newVirtualThreadPerTaskExecutor(); RealtimeServerHeartbeatEmitter emitter = new RealtimeServerHeartbeatEmitter( properties, registry, new RealtimeCodec(properties), - new RealtimeSessionOperations(registry, cleanup, metrics), dispatchExecutor); + new RealtimeSessionOperations(registry, cleanup, metrics), deliveryExecutor); - // Driven from a separate thread so the assertion runs *while* the wedged write is still - // blocked. Asserting after emitHeartbeats() returns would pass against a sequential - // implementation too, since by then the delayed send has already completed. Thread sweep = new Thread(emitter::emitHeartbeats, "heartbeat-sweep"); sweep.setDaemon(true); try { @@ -234,31 +231,17 @@ properties, registry, new RealtimeCodec(properties), assertTrue( healthyReceived.await(5, TimeUnit.SECONDS), - "healthy tenant must receive its heartbeat while another tenant's write is wedged"); + "healthy connection must receive its heartbeat while another write is wedged"); } finally { releaseWedged.countDown(); sweep.join(TimeUnit.SECONDS.toMillis(15)); - dispatchExecutor.shutdown(); - } - } - - /** Finds a server id that lands on a different dispatch shard than {@code other}. */ - private static String distinctShardServerId(String other, int workers) { - int otherShard = Math.floorMod(other.hashCode(), workers); - for (int index = 1; index < 1000; index++) { - String candidate = "server-" + index; - if (Math.floorMod(candidate.hashCode(), workers) != otherShard) { - assertNotEquals(other, candidate); - return candidate; - } + deliveryExecutor.shutdownNow(); } - throw new IllegalStateException("no server id found on a different shard"); } private static final class Fixture { private final RealtimeProperties properties = new RealtimeProperties(); private final RealtimeConnectionRegistry registry; - private final RealtimeDispatchExecutor dispatchExecutor = mock(RealtimeDispatchExecutor.class); private final List deferred = new ArrayList<>(); private final RealtimeServerHeartbeatEmitter emitter; @@ -272,21 +255,12 @@ private Fixture(boolean runInline) { RealtimeMetrics metrics = new RealtimeMetrics(new SimpleMeterRegistry()); RealtimeConnectionCleanup cleanup = new RealtimeConnectionCleanup(registry, new RealtimeMessageRateLimiter(properties), metrics); - doAnswer(invocation -> { - Runnable task = invocation.getArgument(1); - if (runInline) { - task.run(); - } else { - deferred.add(task); - } - return null; - }).when(dispatchExecutor).execute(anyString(), any(Runnable.class)); emitter = new RealtimeServerHeartbeatEmitter( properties, registry, new RealtimeCodec(properties), new RealtimeSessionOperations(registry, cleanup, metrics), - dispatchExecutor + runInline ? Runnable::run : deferred::add ); } diff --git a/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterWiringTest.java b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterWiringTest.java new file mode 100644 index 0000000..856a8ca --- /dev/null +++ b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterWiringTest.java @@ -0,0 +1,34 @@ +package gg.modl.backend.realtime.schedule; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.Mockito.mock; + +import gg.modl.backend.realtime.config.RealtimeProperties; +import gg.modl.backend.realtime.state.RealtimeConnectionRegistry; +import gg.modl.backend.realtime.transport.RealtimeCodec; +import gg.modl.backend.realtime.transport.RealtimeSessionOperations; +import org.junit.jupiter.api.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; + +/** + * The emitter carries a second, package-private constructor for injecting a delivery executor in + * tests. Two candidate constructors make Spring's autowiring ambiguous, and it then falls back to a + * no-arg constructor that does not exist — failing at context startup rather than at compile time, + * where no other test would catch it. This pins the annotated constructor as the injection point. + */ +class RealtimeServerHeartbeatEmitterWiringTest { + + @Test + void springResolvesTheAnnotatedConstructor() { + try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) { + context.registerBean(RealtimeProperties.class, RealtimeProperties::new); + context.registerBean(RealtimeConnectionRegistry.class, () -> mock(RealtimeConnectionRegistry.class)); + context.registerBean(RealtimeCodec.class, () -> mock(RealtimeCodec.class)); + context.registerBean(RealtimeSessionOperations.class, () -> mock(RealtimeSessionOperations.class)); + context.registerBean(RealtimeServerHeartbeatEmitter.class); + context.refresh(); + + assertNotNull(context.getBean(RealtimeServerHeartbeatEmitter.class)); + } + } +}