From 8e4891d60820d08e92f43ee359aad7dac6d5a748 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 29 Jul 2026 19:19:35 +0300 Subject: [PATCH] fix(system): keep worker telemetry polling live --- .../src/core/system/sequentialPolling.ts | 45 ++++++++++++++ .../src/core/system/useWorkerTelemetry.ts | 36 ++++++----- .../test/systemTelemetry.test.mjs | 61 +++++++++++++++++++ 3 files changed, 126 insertions(+), 16 deletions(-) create mode 100644 apps/control-station/src/core/system/sequentialPolling.ts diff --git a/apps/control-station/src/core/system/sequentialPolling.ts b/apps/control-station/src/core/system/sequentialPolling.ts new file mode 100644 index 0000000..21dc93d --- /dev/null +++ b/apps/control-station/src/core/system/sequentialPolling.ts @@ -0,0 +1,45 @@ +export interface SequentialPollingScheduler { + setTimeout(callback: () => void, delayMilliseconds: number): number; + clearTimeout(handle: number): void; +} + +function createBrowserScheduler(): SequentialPollingScheduler { + return { + setTimeout: (callback, delayMilliseconds) => + window.setTimeout(callback, delayMilliseconds), + clearTimeout: (handle) => window.clearTimeout(handle), + }; +} + +export function startSequentialPolling( + task: () => Promise, + delayMilliseconds: number, + scheduler = createBrowserScheduler(), +): () => void { + let stopped = false; + let timer: number | null = null; + + const run = async () => { + if (stopped) return; + try { + await task(); + } catch { + // The caller owns error reporting. A transient failure must not stop polling. + } + if (stopped) return; + timer = scheduler.setTimeout(() => { + timer = null; + void run(); + }, delayMilliseconds); + }; + + void run(); + + return () => { + stopped = true; + if (timer !== null) { + scheduler.clearTimeout(timer); + timer = null; + } + }; +} diff --git a/apps/control-station/src/core/system/useWorkerTelemetry.ts b/apps/control-station/src/core/system/useWorkerTelemetry.ts index 6882a59..2f32f55 100644 --- a/apps/control-station/src/core/system/useWorkerTelemetry.ts +++ b/apps/control-station/src/core/system/useWorkerTelemetry.ts @@ -4,6 +4,7 @@ import { fetchWorkerTelemetry, type WorkerTelemetry, } from "./workerTelemetry"; +import { startSequentialPolling } from "./sequentialPolling"; import { normalizeWorkerTelemetryPollMilliseconds } from "./telemetryPollInterval"; export const DEFAULT_WORKER_TELEMETRY_POLL_MILLISECONDS = 3_000; @@ -38,28 +39,31 @@ export function useWorkerTelemetry( return; } const controller = new AbortController(); - setLoading(true); - void fetchWorkerTelemetry(contourId, controller.signal) - .then((document) => { + const stopPolling = startSequentialPolling(async () => { + setLoading(true); + try { + const document = await fetchWorkerTelemetry(contourId, controller.signal); if (controller.signal.aborted) return; setTelemetry(document); setError(null); - }) - .catch((reason: unknown) => { + } catch (reason: unknown) { if (controller.signal.aborted) return; setError(reason instanceof Error ? reason.message : "Worker 006 не ответил."); - }) - .finally(() => { + } finally { if (!controller.signal.aborted) setLoading(false); - }); - return () => controller.abort(); - }, [contourId, enabled, externalRefreshGeneration, generation]); - - useEffect(() => { - if (!enabled || loading) return; - const timer = window.setTimeout(refresh, normalizedPollMilliseconds); - return () => window.clearTimeout(timer); - }, [enabled, loading, normalizedPollMilliseconds, refresh]); + } + }, normalizedPollMilliseconds); + return () => { + stopPolling(); + controller.abort(); + }; + }, [ + contourId, + enabled, + externalRefreshGeneration, + generation, + normalizedPollMilliseconds, + ]); return { telemetry, loading, error, refresh }; } diff --git a/apps/control-station/test/systemTelemetry.test.mjs b/apps/control-station/test/systemTelemetry.test.mjs index fcfc977..58c1ccc 100644 --- a/apps/control-station/test/systemTelemetry.test.mjs +++ b/apps/control-station/test/systemTelemetry.test.mjs @@ -24,6 +24,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => contourSettings, contourContract, pollIntervalContract, + pollingScheduler, telemetryStyles, styles, ] = await Promise.all([ @@ -40,6 +41,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => read("components/system/ComputeContourSettingsWindow.tsx"), read("core/system/computeContours.ts"), read("core/system/telemetryPollInterval.ts"), + read("core/system/sequentialPolling.ts"), read("styles/system-telemetry.css"), read("styles.css"), ]); @@ -109,6 +111,9 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => ); assert.match(pollIntervalContract, /MIN_TELEMETRY_POLL_INTERVAL_SECONDS\s*=\s*1/); assert.match(telemetryPolling, /normalizeWorkerTelemetryPollMilliseconds/); + assert.match(telemetryPolling, /startSequentialPolling/); + assert.doesNotMatch(telemetryPolling, /if\s*\(\s*!enabled\s*\|\|\s*loading\s*\)/); + assert.match(pollingScheduler, /A transient failure must not stop polling/); assert.match(styles, /system-telemetry\.css/); }); @@ -145,3 +150,59 @@ test("MQTT polling interval keeps an unrestricted draft until explicit commit", assert.equal(contract.normalizeWorkerTelemetryPollMilliseconds(100), 1_000); assert.equal(contract.normalizeWorkerTelemetryPollMilliseconds(3_000), 3_000); }); + +test("Worker telemetry polling repeats after fast responses and transient errors", async () => { + const source = await read("core/system/sequentialPolling.ts"); + const javascript = ts.transpileModule(source, { + compilerOptions: { + module: ts.ModuleKind.ES2022, + target: ts.ScriptTarget.ES2022, + }, + }).outputText; + const polling = await import( + `data:text/javascript;base64,${Buffer.from(javascript).toString("base64")}` + ); + + let nextHandle = 1; + const timers = new Map(); + const cleared = []; + const scheduler = { + setTimeout(callback, delayMilliseconds) { + const handle = nextHandle++; + timers.set(handle, { callback, delayMilliseconds }); + return handle; + }, + clearTimeout(handle) { + cleared.push(handle); + timers.delete(handle); + }, + }; + let attempts = 0; + const stop = polling.startSequentialPolling(async () => { + attempts += 1; + if (attempts === 2) throw new Error("transient"); + }, 3_000, scheduler); + + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(attempts, 1); + assert.equal(timers.size, 1); + assert.equal([...timers.values()][0].delayMilliseconds, 3_000); + + let [handle, scheduled] = [...timers.entries()][0]; + timers.delete(handle); + scheduled.callback(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(attempts, 2); + assert.equal(timers.size, 1, "a transient error must schedule the next poll"); + + [handle, scheduled] = [...timers.entries()][0]; + timers.delete(handle); + scheduled.callback(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(attempts, 3); + assert.equal(timers.size, 1); + + stop(); + assert.equal(timers.size, 0); + assert.deepEqual(cleared, [nextHandle - 1]); +});