fix(system): keep worker telemetry polling live
This commit is contained in:
@@ -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<void>,
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -4,6 +4,7 @@ import {
|
|||||||
fetchWorkerTelemetry,
|
fetchWorkerTelemetry,
|
||||||
type WorkerTelemetry,
|
type WorkerTelemetry,
|
||||||
} from "./workerTelemetry";
|
} from "./workerTelemetry";
|
||||||
|
import { startSequentialPolling } from "./sequentialPolling";
|
||||||
import { normalizeWorkerTelemetryPollMilliseconds } from "./telemetryPollInterval";
|
import { normalizeWorkerTelemetryPollMilliseconds } from "./telemetryPollInterval";
|
||||||
|
|
||||||
export const DEFAULT_WORKER_TELEMETRY_POLL_MILLISECONDS = 3_000;
|
export const DEFAULT_WORKER_TELEMETRY_POLL_MILLISECONDS = 3_000;
|
||||||
@@ -38,28 +39,31 @@ export function useWorkerTelemetry(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
|
const stopPolling = startSequentialPolling(async () => {
|
||||||
setLoading(true);
|
setLoading(true);
|
||||||
void fetchWorkerTelemetry(contourId, controller.signal)
|
try {
|
||||||
.then((document) => {
|
const document = await fetchWorkerTelemetry(contourId, controller.signal);
|
||||||
if (controller.signal.aborted) return;
|
if (controller.signal.aborted) return;
|
||||||
setTelemetry(document);
|
setTelemetry(document);
|
||||||
setError(null);
|
setError(null);
|
||||||
})
|
} catch (reason: unknown) {
|
||||||
.catch((reason: unknown) => {
|
|
||||||
if (controller.signal.aborted) return;
|
if (controller.signal.aborted) return;
|
||||||
setError(reason instanceof Error ? reason.message : "Worker 006 не ответил.");
|
setError(reason instanceof Error ? reason.message : "Worker 006 не ответил.");
|
||||||
})
|
} finally {
|
||||||
.finally(() => {
|
|
||||||
if (!controller.signal.aborted) setLoading(false);
|
if (!controller.signal.aborted) setLoading(false);
|
||||||
});
|
}
|
||||||
return () => controller.abort();
|
}, normalizedPollMilliseconds);
|
||||||
}, [contourId, enabled, externalRefreshGeneration, generation]);
|
return () => {
|
||||||
|
stopPolling();
|
||||||
useEffect(() => {
|
controller.abort();
|
||||||
if (!enabled || loading) return;
|
};
|
||||||
const timer = window.setTimeout(refresh, normalizedPollMilliseconds);
|
}, [
|
||||||
return () => window.clearTimeout(timer);
|
contourId,
|
||||||
}, [enabled, loading, normalizedPollMilliseconds, refresh]);
|
enabled,
|
||||||
|
externalRefreshGeneration,
|
||||||
|
generation,
|
||||||
|
normalizedPollMilliseconds,
|
||||||
|
]);
|
||||||
|
|
||||||
return { telemetry, loading, error, refresh };
|
return { telemetry, loading, error, refresh };
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
|
|||||||
contourSettings,
|
contourSettings,
|
||||||
contourContract,
|
contourContract,
|
||||||
pollIntervalContract,
|
pollIntervalContract,
|
||||||
|
pollingScheduler,
|
||||||
telemetryStyles,
|
telemetryStyles,
|
||||||
styles,
|
styles,
|
||||||
] = await Promise.all([
|
] = await Promise.all([
|
||||||
@@ -40,6 +41,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
|
|||||||
read("components/system/ComputeContourSettingsWindow.tsx"),
|
read("components/system/ComputeContourSettingsWindow.tsx"),
|
||||||
read("core/system/computeContours.ts"),
|
read("core/system/computeContours.ts"),
|
||||||
read("core/system/telemetryPollInterval.ts"),
|
read("core/system/telemetryPollInterval.ts"),
|
||||||
|
read("core/system/sequentialPolling.ts"),
|
||||||
read("styles/system-telemetry.css"),
|
read("styles/system-telemetry.css"),
|
||||||
read("styles.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(pollIntervalContract, /MIN_TELEMETRY_POLL_INTERVAL_SECONDS\s*=\s*1/);
|
||||||
assert.match(telemetryPolling, /normalizeWorkerTelemetryPollMilliseconds/);
|
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/);
|
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(100), 1_000);
|
||||||
assert.equal(contract.normalizeWorkerTelemetryPollMilliseconds(3_000), 3_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]);
|
||||||
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user