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.
+ * + *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 +public class RealtimeServerHeartbeatEmitter { + private final RealtimeProperties properties; + private final RealtimeConnectionRegistry connectionRegistry; + private final RealtimeCodec codec; + private final RealtimeSessionOperations sessionOperations; + 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() { + if (!properties.isEnabled()) { + return; + } + + for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : connectionRegistry.snapshot()) { + if (!isEligible(snapshot)) { + continue; + } + deliveryExecutor.execute(() -> send(snapshot)); + } + } + + 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); + } + } + + 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(); + } + + @PreDestroy + public void shutdown() { + if (deliveryExecutor instanceof ExecutorService service) { + service.shutdownNow(); + } + } +} 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, CollectionBuilt 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..6cdc837 --- /dev/null +++ b/src/test/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitterTest.java @@ -0,0 +1,291 @@ +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.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.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.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; +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("server-id")), 1); + + fixture.emitter.emitHeartbeats(); + + ArgumentCaptorBoth 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 wedgedConnectionDoesNotDelayHeartbeatForAnother() throws Exception { + RealtimeProperties properties = new RealtimeProperties(); + properties.setEnabled(true); + + 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("shared-server")), 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("shared-server")), 1); + + // 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 -> 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); + ExecutorService deliveryExecutor = Executors.newVirtualThreadPerTaskExecutor(); + RealtimeServerHeartbeatEmitter emitter = new RealtimeServerHeartbeatEmitter( + properties, registry, new RealtimeCodec(properties), + new RealtimeSessionOperations(registry, cleanup, metrics), deliveryExecutor); + + Thread sweep = new Thread(emitter::emitHeartbeats, "heartbeat-sweep"); + sweep.setDaemon(true); + try { + sweep.start(); + + assertTrue( + healthyReceived.await(5, TimeUnit.SECONDS), + "healthy connection must receive its heartbeat while another write is wedged"); + } finally { + releaseWedged.countDown(); + sweep.join(TimeUnit.SECONDS.toMillis(15)); + deliveryExecutor.shutdownNow(); + } + } + + private static final class Fixture { + private final RealtimeProperties properties = new RealtimeProperties(); + private final RealtimeConnectionRegistry registry; + private final List