fix: detect dead WebSocket connections to eliminate ghost voice users
Add server-side heartbeat using RFC 6455 protocol-level ping/pong frames to detect abruptly disconnected clients (e.g. PC shutdown without TCP FIN). Dead connections are terminated within ~65s, triggering the existing cleanup chain to remove ghost users from voice channels and presence.
This commit is contained in:
@@ -18,6 +18,10 @@ import type {
|
|||||||
} from '@backspace/shared';
|
} from '@backspace/shared';
|
||||||
import { sanitizeUser } from '../utils/sanitize.js';
|
import { sanitizeUser } from '../utils/sanitize.js';
|
||||||
|
|
||||||
|
// ─── Heartbeat State ──────────────────────────────────────────────────────────
|
||||||
|
const wsIsAlive: WeakMap<WebSocket, boolean> = new WeakMap();
|
||||||
|
let heartbeatInterval: ReturnType<typeof setInterval> | null = null;
|
||||||
|
|
||||||
// SQLite's SQLITE_MAX_VARIABLE_NUMBER default is 999.
|
// SQLite's SQLITE_MAX_VARIABLE_NUMBER default is 999.
|
||||||
// Chunk inArray() calls to stay safely under this limit.
|
// Chunk inArray() calls to stay safely under this limit.
|
||||||
const BATCH_CHUNK_SIZE = 500;
|
const BATCH_CHUNK_SIZE = 500;
|
||||||
@@ -193,6 +197,9 @@ class ConnectionManager {
|
|||||||
status: 'offline',
|
status: 'offline',
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Clean up userSpaces (re-populated on next connect via setUserSpaces)
|
||||||
|
this.userSpaces.delete(userId);
|
||||||
}
|
}
|
||||||
|
|
||||||
getUserConnections(userId: string): Set<WebSocket> {
|
getUserConnections(userId: string): Set<WebSocket> {
|
||||||
@@ -529,6 +536,10 @@ class ConnectionManager {
|
|||||||
return Array.from(this.connections.keys());
|
return Array.from(this.connections.keys());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
getAllConnections(): Map<string, Set<WebSocket>> {
|
||||||
|
return this.connections;
|
||||||
|
}
|
||||||
|
|
||||||
/** Push a fresh ready payload to a specific user, forcing full store re-sync. */
|
/** Push a fresh ready payload to a specific user, forcing full store re-sync. */
|
||||||
pushReadyPayload(userId: string): void {
|
pushReadyPayload(userId: string): void {
|
||||||
const connections = this.getUserConnections(userId);
|
const connections = this.getUserConnections(userId);
|
||||||
@@ -965,6 +976,9 @@ export async function registerWebSocket(app: FastifyInstance): Promise<void> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Any received message proves liveness
|
||||||
|
wsIsAlive.set(ws, true);
|
||||||
|
|
||||||
if (!authenticated) {
|
if (!authenticated) {
|
||||||
// First message must be auth
|
// First message must be auth
|
||||||
if (parsed.type !== 'auth' || typeof parsed.token !== 'string') {
|
if (parsed.type !== 'auth' || typeof parsed.token !== 'string') {
|
||||||
@@ -987,6 +1001,10 @@ export async function registerWebSocket(app: FastifyInstance): Promise<void> {
|
|||||||
// Add connection
|
// Add connection
|
||||||
connectionManager.addConnection(userId, ws);
|
connectionManager.addConnection(userId, ws);
|
||||||
|
|
||||||
|
// Mark alive for heartbeat detection; browsers auto-respond to ping frames (RFC 6455)
|
||||||
|
wsIsAlive.set(ws, true);
|
||||||
|
ws.on('pong', () => { wsIsAlive.set(ws, true); });
|
||||||
|
|
||||||
// Build and send ready payload
|
// Build and send ready payload
|
||||||
const readyData = buildReadyPayload(userId);
|
const readyData = buildReadyPayload(userId);
|
||||||
ws.send(JSON.stringify({
|
ws.send(JSON.stringify({
|
||||||
@@ -1039,4 +1057,30 @@ export async function registerWebSocket(app: FastifyInstance): Promise<void> {
|
|||||||
clearTimeout(authTimeout);
|
clearTimeout(authTimeout);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// ─── Heartbeat Sweep ──────────────────────────────────────────────────────
|
||||||
|
// Detect dead connections (e.g. PC shut off without TCP FIN).
|
||||||
|
// Sends protocol-level ping frames; browsers auto-respond with pong (RFC 6455).
|
||||||
|
// Worst-case detection: 30s + 30s + 5s grace = ~65s.
|
||||||
|
const HEARTBEAT_INTERVAL_MS = 30_000;
|
||||||
|
|
||||||
|
heartbeatInterval = setInterval(() => {
|
||||||
|
for (const [, userConnections] of connectionManager.getAllConnections()) {
|
||||||
|
for (const ws of userConnections) {
|
||||||
|
if (wsIsAlive.get(ws) === false) {
|
||||||
|
ws.terminate(); // Emits 'close' → removeConnection → scheduleDisconnect → finalizeDisconnect
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
wsIsAlive.set(ws, false);
|
||||||
|
if (ws.readyState === 1) ws.ping();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}, HEARTBEAT_INTERVAL_MS);
|
||||||
|
|
||||||
|
app.addHook('onClose', async () => {
|
||||||
|
if (heartbeatInterval) {
|
||||||
|
clearInterval(heartbeatInterval);
|
||||||
|
heartbeatInterval = null;
|
||||||
|
}
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user