From f26735b9e560889bd1eaabb2106868f251d391c8 Mon Sep 17 00:00:00 2001 From: legop3 Date: Wed, 29 Apr 2026 14:25:28 -0400 Subject: [PATCH] flhj --- .../roverSnapshotService/socketGateway.js | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/server/src/services/roverSnapshotService/socketGateway.js b/server/src/services/roverSnapshotService/socketGateway.js index e892c443..31ae18ce 100644 --- a/server/src/services/roverSnapshotService/socketGateway.js +++ b/server/src/services/roverSnapshotService/socketGateway.js @@ -9,6 +9,7 @@ const { isAdmin, isLockdownAdmin, getRole } = require('../roleService'); const SUBSCRIBE_LIMIT = 50; const SUBSCRIBE_WINDOW_MS = 10000; const STREAM_INTERVAL_MS = 333; +const FALLBACK_REFRESH_MS = 500; function passesMode(socket) { const mode = getMode(); @@ -31,6 +32,8 @@ function registerRoverSnapshotSocketGateway({ const subscribeBuckets = new Map(); const lastSentBySocket = new Map(); const sentCounts = new Map(); + const lastForcedTsBySocket = new Map(); + let fallbackTimer = null; function addSubscription(socket, roverId) { if (!roverSubscribers.has(roverId)) roverSubscribers.set(roverId, new Set()); @@ -75,6 +78,29 @@ function registerRoverSnapshotSocketGateway({ socket.emit('roverSnapshot:status', { id: roverId, ...status }); } + function ensureFallbackTimer() { + if (fallbackTimer || typeof fetchSnapshotNow !== 'function') return; + fallbackTimer = setInterval(() => { + socketSubscriptions.forEach((roverIds, socketId) => { + const socket = io.sockets.sockets.get(socketId); + if (!socket) return; + roverIds.forEach(async (roverId) => { + try { + const result = await fetchSnapshotNow(String(roverId), { force: true }); + if (!result?.frame) return; + const key = `${socketId}:${roverId}`; + const prevTs = lastForcedTsBySocket.get(key) || 0; + if ((result.ts || 0) <= prevTs) return; + lastForcedTsBySocket.set(key, result.ts || Date.now()); + sendFrame(socket, roverId, { ts: result.ts || Date.now() }, result.frame); + } catch (err) { + logger.warn('Fallback refresh failed', { socketId, roverId, err: err.message }); + } + }); + }); + }, FALLBACK_REFRESH_MS); + } + roverSnapshotEvents.on('frame', ({ id, buffer, ts }) => { const bucket = roverSubscribers.get(id); if (!bucket || !buffer) return; @@ -133,6 +159,7 @@ function registerRoverSnapshotSocketGateway({ subscribedIds: validIds, }); validIds.forEach((roverId) => addSubscription(socket, roverId)); + ensureFallbackTimer(); validIds.forEach(async (roverId) => { if (typeof fetchSnapshotNow === 'function') { try { @@ -165,6 +192,9 @@ function registerRoverSnapshotSocketGateway({ removeAllSubscriptions(socket.id); subscribeBuckets.delete(socket.id); lastSentBySocket.delete(socket.id); + Array.from(lastForcedTsBySocket.keys()).forEach((key) => { + if (key.startsWith(`${socket.id}:`)) lastForcedTsBySocket.delete(key); + }); }); }); }