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,121 @@
+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 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;
+
+/**
+ * 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>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 <em>any</em> 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.</p>
+ */
+@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();
+        }
+    }
+}

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}