diff --git a/server/src/services/roverSnapshotService/poller.js b/server/src/services/roverSnapshotService/poller.js index 28cbbeec..c592ef08 100644 --- a/server/src/services/roverSnapshotService/poller.js +++ b/server/src/services/roverSnapshotService/poller.js @@ -10,6 +10,7 @@ const SNAPSHOT_DIR = process.env.ROVER_SNAPSHOT_DIR || '/var/lib/rover-snapshots const POLL_INTERVAL_MS = 300; const roverState = new Map(); const events = new EventEmitter(); +const readCounts = new Map(); let pollTimer = null; function markState(id, updates = {}) { @@ -24,24 +25,32 @@ function getSnapshotPath(id) { } function createRoverSnapshotPoller({ roverManager }) { - async function fetchSnapshot(id) { + async function fetchSnapshot(id, options = {}) { + const force = Boolean(options?.force); const state = roverState.get(id); if (state?.fetching) return; markState(id, { fetching: true }); try { const filePath = getSnapshotPath(id); const stats = await fs.stat(filePath); - if (state?.mtimeMs && stats.mtimeMs <= state.mtimeMs) return; + if (!force && state?.mtimeMs && stats.mtimeMs <= state.mtimeMs) return; const buffer = await fs.readFile(filePath); const ts = stats.mtimeMs || Date.now(); markState(id, { frame: buffer, ts, error: null, failures: 0, mtimeMs: stats.mtimeMs }); + const count = (readCounts.get(id) || 0) + 1; + readCounts.set(id, count); + if (count === 1 || count % 30 === 0) { + logger.warn('Snapshot read ok', { id, count, bytes: buffer.length, ts, mtimeMs: stats.mtimeMs }); + } events.emit('frame', { id, buffer, ts }); + return { frame: buffer, ts }; } catch (err) { const failures = (state?.failures || 0) + 1; const message = err.code === 'ENOENT' ? 'Snapshot missing' : err.message; markState(id, { error: message, failures }); events.emit('status', { id, error: message }); if (failures % 20 === 1) logger.warn('Snapshot read failed', { id, err: message }); + return null; } finally { markState(id, { fetching: false }); } diff --git a/server/src/services/roverSnapshotService/socketGateway.js b/server/src/services/roverSnapshotService/socketGateway.js index 7e3702eb..e892c443 100644 --- a/server/src/services/roverSnapshotService/socketGateway.js +++ b/server/src/services/roverSnapshotService/socketGateway.js @@ -30,6 +30,7 @@ function registerRoverSnapshotSocketGateway({ const socketSubscriptions = new Map(); const subscribeBuckets = new Map(); const lastSentBySocket = new Map(); + const sentCounts = new Map(); function addSubscription(socket, roverId) { if (!roverSubscribers.has(roverId)) roverSubscribers.set(roverId, new Set()); @@ -89,6 +90,12 @@ function registerRoverSnapshotSocketGateway({ const now = ts || Date.now(); if (now - lastSent < STREAM_INTERVAL_MS) return; lastMap.set(id, now); + const key = `${socketId}:${id}`; + const sentCount = (sentCounts.get(key) || 0) + 1; + sentCounts.set(key, sentCount); + if (sentCount === 1 || sentCount % 30 === 0) { + logger.warn('Snapshot frame sent', { socketId, roverId: id, sentCount, ts, bytes: buffer.length }); + } sendFrame(socket, id, { ts }, buffer); }); }); @@ -126,11 +133,13 @@ function registerRoverSnapshotSocketGateway({ subscribedIds: validIds, }); validIds.forEach((roverId) => addSubscription(socket, roverId)); - validIds.forEach((roverId) => { + validIds.forEach(async (roverId) => { if (typeof fetchSnapshotNow === 'function') { - fetchSnapshotNow(String(roverId)).catch((err) => { + try { + await fetchSnapshotNow(String(roverId), { force: true }); + } catch (err) { logger.warn('Immediate snapshot fetch failed', { roverId, err: err.message }); - }); + } } const state = getRoverSnapshotState(roverId); if (state?.frame) sendFrame(socket, roverId, { ts: state.ts }, state.frame);