Send server heartbeats so idle realtime connections stay alive

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;
 

src/main/java/gg/modl/backend/realtime/schedule/RealtimeServerHeartbeatEmitter.java

@@ -0,0 +1,87 @@
+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;
+import org.springframework.web.socket.WebSocketSession;
+
+/**
+ * Emits unsolicited heartbeats to authenticated connections so idle sessions keep seeing inbound
+ * traffic.
+ *
+ * <p>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 <em>client</em> heartbeats have stopped.</p>
+ *
+ * <p>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.</p>
+ */
+@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;
+
+    @Scheduled(fixedDelayString = "${modl.realtime.ws.server-heartbeat-interval-ms:25000}")
+    public void emitHeartbeats() {
+        if (!properties.isEnabled()) {
+            return;
+        }
+
+        Map<String, List<RealtimeConnectionRegistry.RealtimeConnectionSnapshot>> byServer = new LinkedHashMap<>();
+        for (RealtimeConnectionRegistry.RealtimeConnectionSnapshot snapshot : connectionRegistry.snapshot()) {
+            String serverId = snapshot.state().getServerId();
+            if (serverId == null || !isEligible(snapshot)) {
+                continue;
+            }
+            byServer.computeIfAbsent(serverId, key -> new ArrayList<>()).add(snapshot);
+        }
+
+        byServer.forEach((serverId, snapshots) -> dispatchExecutor.execute(serverId, () -> sendAll(snapshots)));
+    }
+
+    private void sendAll(List<RealtimeConnectionRegistry.RealtimeConnectionSnapshot> 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(
+                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();
+    }
+}

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;
     }

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<Topic> 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.
+     *
+     * <p>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.</p>
+     */
+    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)

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}