Skip to content

Commit 68120e7

Browse files
luluizCopilot
andcommitted
fix(telemetry): address review feedback on accounting and lifecycle
- PanelExtensionAdapter: do not count a tick without messages as a skipped render while a render is pending. - Document that range-delivered bytes are approximate when data is merged into the middle of allFrames (count stays exact). - IterablePlayer: end the preload phase with reason "cancelled" when the last full-preload subscription is removed; emit the datasource close start/end pair at most once per player. - HydratedSourcePool: settle the over-budget state right after victims are evicted instead of after their close() settles. - WorkerSerializedIterableSource: bound the final telemetry drain with one shared deadline so the worker is always disposed; stop the bridge timer when enabling worker telemetry fails. - WorkerSerializedIterableSourceWorker: drop the redundant publish in terminate(); the final drain publishes on dispose. - CurrentLayoutProvider: emit layout.applied only when the selection was accepted (not for unsupported working layout versions). - BufferedTelemetrySink: reject counter totals that overflow to Infinity (record and ingest paths); keep the sink's self-metric series separate from caller series and reject caller metrics using the reserved self-metric names. - Tests for each change; assert initialize() is called once in the stale-init worker source test. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 7b5651a commit 68120e7

19 files changed

Lines changed: 558 additions & 88 deletions

‎packages/suite-base/src/components/PanelExtensionAdapter/PanelExtensionAdapter.telemetry.test.tsx‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -284,6 +284,44 @@ describe("PanelExtensionAdapter telemetry", () => {
284284
expect(counterCalls(telemetry, PANEL_TELEMETRY_METRICS.renderCount)).toHaveLength(0);
285285
});
286286

287+
it("should not count a tick without messages as skipped while a render is pending", async () => {
288+
// GIVEN - an enabled sink, a pipeline that does not wait for the panel, and a panel watching
289+
// frames that never calls done()
290+
const telemetry = TelemetryBuilder.mockTelemetry();
291+
setTelemetrySink(telemetry);
292+
const initPanel = (context: PanelExtensionContext) => {
293+
context.watch("currentFrame");
294+
context.subscribe([{ topic: "/x" }]);
295+
context.onRender = jest.fn();
296+
};
297+
const handle = render(
298+
<Wrapper
299+
currentTime={{ sec: 1, nsec: 0 }}
300+
frame={{ "/x": [buildMessage(10)] }}
301+
initPanel={initPanel}
302+
pauseFrame={nonBlockingPauseFrame}
303+
/>,
304+
);
305+
await flush();
306+
const skippedBefore = counterCalls(telemetry, PANEL_TELEMETRY_METRICS.renderSkipped).length;
307+
308+
// WHEN - a player-state update carrying no messages for the panel arrives while the render is
309+
// still pending
310+
handle.rerender(
311+
<Wrapper
312+
currentTime={{ sec: 2, nsec: 0 }}
313+
initPanel={initPanel}
314+
pauseFrame={nonBlockingPauseFrame}
315+
/>,
316+
);
317+
await flush();
318+
319+
// THEN - no skipped frame is recorded
320+
expect(counterCalls(telemetry, PANEL_TELEMETRY_METRICS.renderSkipped)).toHaveLength(
321+
skippedBefore,
322+
);
323+
});
324+
287325
it("should not record telemetry from a stale done() called after the panel re-initialised", async () => {
288326
// GIVEN - a first panel instance whose render is still pending (done() captured, not called)
289327
const telemetry = TelemetryBuilder.mockTelemetry();

‎packages/suite-base/src/components/PanelExtensionAdapter/PanelExtensionAdapter.tsx‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -352,8 +352,13 @@ function PanelExtensionAdapter(
352352

353353
if (renderingRef.current) {
354354
setSlowRender(true);
355-
// Only a new frame input counts as skipped, not a re-run caused by another input.
356-
if (telemetryEnabled && messageEvents !== lastAccountedFrameRef.current) {
355+
// Only a new, non-empty frame input counts as skipped, not a re-run caused by another input
356+
// nor a player-state update that carried no messages for this panel.
357+
if (
358+
telemetryEnabled &&
359+
messageEvents != undefined &&
360+
messageEvents !== lastAccountedFrameRef.current
361+
) {
357362
lastAccountedFrameRef.current = messageEvents;
358363
recordRenderSkipped(telemetry, telemetryLabels);
359364
}

‎packages/suite-base/src/components/PanelExtensionAdapter/constants.ts‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,10 @@ export const PANEL_TELEMETRY_METRICS = {
4141
* `useSubscribeMessageRange` batches or through new entries in `renderState.allFrames`
4242
*/
4343
rangeMessagesDelivered: "panel.rangeMessages.delivered",
44-
/** counter: sum of `sizeInBytes` of those preloaded message events */
44+
/**
45+
* counter: sum of `sizeInBytes` of those preloaded message events. Approximate for `allFrames`:
46+
* when new data is merged into the middle of `allFrames` only the appended tail is summed
47+
* (re-summing the whole array on every block update is too expensive). The count is exact.
48+
*/
4549
rangeMessagesDeliveredBytes: "panel.rangeMessages.deliveredBytes",
4650
} as const;

‎packages/suite-base/src/players/IterablePlayer/IterablePlayer.telemetry.test.ts‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,19 @@ class FailingTerminateSource extends TelemetryTestSource {
108108
}
109109
}
110110

111+
// Counts terminate() calls and signals the second one (a close() after the close completed).
112+
class CountingTerminateSource extends TelemetryTestSource {
113+
public readonly secondTerminate = signal();
114+
#terminateCalls = 0;
115+
116+
public async terminate(): Promise<void> {
117+
this.#terminateCalls++;
118+
if (this.#terminateCalls === 2) {
119+
this.secondTerminate.resolve();
120+
}
121+
}
122+
}
123+
111124
function setupSink(onEvent?: (event: TelemetryEvent) => void): {
112125
sink: BufferedTelemetrySink;
113126
events: TelemetryEvent[];
@@ -321,6 +334,26 @@ describe("IterablePlayer telemetry", () => {
321334
]);
322335
});
323336

337+
it("should end preload with reason 'cancelled' when the last full-preload subscription is removed while loading", async () => {
338+
// Given: a source whose full (block loader) read is held, so the preload phase stays active
339+
const { events } = setupSink();
340+
const source = new GatedPreloadSource();
341+
const player = new IterablePlayer({ source, enablePreload: true, sourceId: SOURCE_ID });
342+
player.setSubscriptions([{ topic: TOPIC, preloadType: "full" }]);
343+
player.setListener(async () => {});
344+
await source.fullReadStarted;
345+
346+
// When
347+
player.setSubscriptions([{ topic: TOPIC }]);
348+
const preloadAfterRemoval = summarize(events).filter((entry) => entry.startsWith("preload"));
349+
player.close();
350+
source.gate.resolve();
351+
await player.isClosed;
352+
353+
// Then
354+
expect(preloadAfterRemoval).toEqual(["preload:start", "preload:end:cancelled"]);
355+
});
356+
324357
it("should end preload with reason 'error' when the block loader fails", async () => {
325358
// Given: a source whose full (block loader) read rejects
326359
const preloadEnded = signal();
@@ -458,6 +491,25 @@ describe("IterablePlayer telemetry", () => {
458491
expect(close).toEqual(["datasource.close:start", "datasource.close:end"]);
459492
});
460493

494+
it("close() after close completed emits no second close pair", async () => {
495+
// Given: a player whose close phase already completed
496+
const { events } = setupSink();
497+
const source = new CountingTerminateSource();
498+
const player = new IterablePlayer({ source, enablePreload: false, sourceId: SOURCE_ID });
499+
await waitForInitialFrame(player);
500+
player.close();
501+
await player.isClosed;
502+
503+
// When: close() is called again and the close state runs again
504+
player.close();
505+
await source.secondTerminate;
506+
await new Promise((resolve) => setTimeout(resolve, 0));
507+
508+
// Then
509+
const close = summarize(events).filter((entry) => entry.startsWith("datasource.close"));
510+
expect(close).toEqual(["datasource.close:start", "datasource.close:end"]);
511+
});
512+
461513
it("should emit datasource.firstMessage only once per player", async () => {
462514
// Given
463515
const { events } = setupSink();

‎packages/suite-base/src/players/IterablePlayer/IterablePlayer.ts‎

Lines changed: 18 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -233,12 +233,13 @@ export class IterablePlayer implements Player {
233233
// --- Phase telemetry. Every emit is guarded by getTelemetry().isEnabled().
234234
// #firstMessageSeen and #blockLoadingStarted are cheap booleans set unconditionally (no
235235
// allocation); only the event emission that reads them is gated. The #telemetryPlaybackActive /
236-
// #telemetryCloseActive / #telemetryPreload markers pair start/end events and only leave their
237-
// initial value once a start event has been emitted, i.e. while telemetry is enabled. ---
236+
// #telemetryClose / #telemetryPreload markers pair start/end events and only leave their
237+
// initial value once a start event has been emitted, i.e. while telemetry is enabled. Close and
238+
// preload are emitted at most once per player. ---
238239
#telemetryLabels?: TelemetryLabels;
239240
#firstMessageSeen = false;
240241
#telemetryPlaybackActive = false;
241-
#telemetryCloseActive = false;
242+
#telemetryClose: "pending" | "active" | "done" = "pending";
242243
#telemetryPreload: "pending" | "active" | "done" = "pending";
243244
#blockLoadingStarted = false;
244245

@@ -428,7 +429,12 @@ export class IterablePlayer implements Player {
428429
this.#preloadTopics = preloadTopics;
429430

430431
this.#blockLoader?.setTopics(this.#preloadTopics);
431-
this.#maybeStartPreloadTelemetry();
432+
if (this.#preloadTopics.size === 0) {
433+
// The last full-preload subscription is gone: nothing is left to preload.
434+
this.#onPreloadEnded("cancelled");
435+
} else {
436+
this.#maybeStartPreloadTelemetry();
437+
}
432438

433439
// If the player is playing, the playing state will detect any subscription changes and adjust
434440
// iterators accordingly. However if we are idle or already seeking then we need to manually
@@ -489,8 +495,10 @@ export class IterablePlayer implements Player {
489495
}
490496

491497
public close(): void {
492-
if (!this.#telemetryCloseActive && getTelemetry().isEnabled()) {
493-
this.#telemetryCloseActive = true;
498+
// Telemetry only: a close() after the close phase completed re-runs the close state but does
499+
// not emit a second start/end pair.
500+
if (this.#telemetryClose === "pending" && getTelemetry().isEnabled()) {
501+
this.#telemetryClose = "active";
494502
this.#emitPhase("datasource.close", "start");
495503
}
496504
this.#setState("close");
@@ -1449,19 +1457,19 @@ export class IterablePlayer implements Player {
14491457
}
14501458

14511459
#onCloseEnded(reason: string | undefined): void {
1452-
if (!this.#telemetryCloseActive) {
1460+
if (this.#telemetryClose !== "active") {
14531461
return;
14541462
}
1455-
this.#telemetryCloseActive = false;
1463+
this.#telemetryClose = "done";
14561464
this.#emitPhase("datasource.close", "end", reason);
14571465
}
14581466

14591467
/**
14601468
* `preload` start: once per player, when block loading is running AND at least one
14611469
* `preloadType: "full"` subscription exists (otherwise the BlockLoader has nothing to load).
14621470
* `preload` end: when the BlockLoader progress first covers the full [0, 1] range; if loading
1463-
* stops first, end is emitted with reason "cancelled" (close) or "error". Later topic changes
1464-
* do not restart the phase.
1471+
* stops first, end is emitted with reason "cancelled" (close, or the last `preloadType: "full"`
1472+
* subscription removed while loading) or "error". Later topic changes do not restart the phase.
14651473
*/
14661474
#maybeStartPreloadTelemetry(): void {
14671475
if (

‎packages/suite-base/src/players/IterablePlayer/WorkerSerializedIterableSource.test.ts‎

Lines changed: 70 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import {
1010
import { TelemetryBuilder } from "@lichtblick/suite-base/util/telemetry/builders/TelemetryBuilder";
1111

1212
import { WorkerSerializedIterableSource } from "./WorkerSerializedIterableSource";
13+
import { TELEMETRY_FINAL_DRAIN_TIMEOUT_MS } from "./constants";
1314

1415
jest.mock("@lichtblick/den/worker", () => ({
1516
ComlinkWrap: jest.fn(),
@@ -326,6 +327,72 @@ describe("WorkerSerializedIterableSource", () => {
326327
expect(jest.getTimerCount()).toBe(0);
327328
});
328329

330+
it("enable failure stops the timer", async () => {
331+
// Given
332+
setTelemetrySink(mainSink);
333+
const remote = makeMockRemote({
334+
enableTelemetry: jest.fn().mockRejectedValue(new Error("no telemetry")),
335+
});
336+
const dispose = mockWorker(remote);
337+
const source = makeSource();
338+
339+
// When
340+
await source.initialize();
341+
const timersAfterInitialize = jest.getTimerCount();
342+
await jest.advanceTimersByTimeAsync(5000);
343+
await source.terminate();
344+
345+
// Then: no periodic drain ran, only terminate()'s final drain, and the worker is disposed.
346+
expect(timersAfterInitialize).toBe(0);
347+
expect(remote.drainTelemetry).toHaveBeenCalledTimes(1);
348+
expect(dispose).toHaveBeenCalledTimes(1);
349+
});
350+
351+
it("should still dispose the worker when a periodic drain in flight never settles", async () => {
352+
// Given: a periodic drain that never settles (dead worker)
353+
setTelemetrySink(mainSink);
354+
const remote = makeMockRemote({
355+
drainTelemetry: jest.fn().mockReturnValue(new Promise<never>(() => {})),
356+
});
357+
const dispose = mockWorker(remote);
358+
const source = makeSource();
359+
await source.initialize();
360+
await jest.advanceTimersByTimeAsync(1000);
361+
expect(remote.drainTelemetry).toHaveBeenCalledTimes(1);
362+
363+
// When
364+
const terminating = source.terminate();
365+
await jest.advanceTimersByTimeAsync(TELEMETRY_FINAL_DRAIN_TIMEOUT_MS);
366+
await terminating;
367+
368+
// Then: the worker is disposed, no final drain was started after the deadline and no timer
369+
// is left running.
370+
expect(dispose).toHaveBeenCalledTimes(1);
371+
expect(remote.drainTelemetry).toHaveBeenCalledTimes(1);
372+
expect(jest.getTimerCount()).toBe(0);
373+
});
374+
375+
it("should still dispose the worker when the final drain never settles", async () => {
376+
// Given: a worker whose drain never settles (dead worker), and no periodic drain in flight
377+
setTelemetrySink(mainSink);
378+
const remote = makeMockRemote({
379+
drainTelemetry: jest.fn().mockReturnValue(new Promise<never>(() => {})),
380+
});
381+
const dispose = mockWorker(remote);
382+
const source = makeSource();
383+
await source.initialize();
384+
385+
// When
386+
const terminating = source.terminate();
387+
await jest.advanceTimersByTimeAsync(TELEMETRY_FINAL_DRAIN_TIMEOUT_MS);
388+
await terminating;
389+
390+
// Then
391+
expect(dispose).toHaveBeenCalledTimes(1);
392+
expect(remote.drainTelemetry).toHaveBeenCalledTimes(1);
393+
expect(jest.getTimerCount()).toBe(0);
394+
});
395+
329396
it("should ingest an in-flight periodic batch before the final batch on terminate", async () => {
330397
// Given
331398
setTelemetrySink(mainSink);
@@ -388,8 +455,10 @@ describe("WorkerSerializedIterableSource", () => {
388455
await initializing;
389456
await jest.advanceTimersByTimeAsync(5000);
390457

391-
// Then: the stale worker is not bridged nor published as the active remote.
458+
// Then: the stale worker is not bridged nor published as the active remote; its initialize()
459+
// still runs once so the pending initialize() call settles.
392460
expect(dispose).toHaveBeenCalledTimes(1);
461+
expect(remote.initialize).toHaveBeenCalledTimes(1);
393462
expect(remote.enableTelemetry).not.toHaveBeenCalled();
394463
expect(remote.drainTelemetry).not.toHaveBeenCalled();
395464
expect(jest.getTimerCount()).toBe(0);

‎packages/suite-base/src/players/IterablePlayer/WorkerSerializedIterableSource.ts‎

Lines changed: 34 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ import type {
2323
} from "./IIterableSource";
2424
import type { SourceWorkerRemote, TelemetryBridge } from "./WorkerSerializedIterableSource.types";
2525
import type { WorkerSerializedIterableSourceWorker } from "./WorkerSerializedIterableSourceWorker";
26-
import { TELEMETRY_DRAIN_INTERVAL_MS } from "./constants";
26+
import { TELEMETRY_DRAIN_INTERVAL_MS, TELEMETRY_FINAL_DRAIN_TIMEOUT_MS } from "./constants";
2727

2828
Comlink.transferHandlers.set("abortsignal", abortSignalTransferHandler);
2929

@@ -65,6 +65,34 @@ async function drainBridge(bridge: TelemetryBridge): Promise<void> {
6565
await bridge.inFlight;
6666
}
6767

68+
// Final drain on terminate(): wait for any periodic drain in flight (so batches are ingested in
69+
// order), then drain once more. Both waits share one deadline because a dead worker never answers
70+
// and the caller must still dispose it; no drain is started once the deadline has expired.
71+
async function finalDrainBridge(bridge: TelemetryBridge): Promise<void> {
72+
let timer: ReturnType<typeof setTimeout> | undefined;
73+
const deadline = new Promise<"expired">((resolve) => {
74+
timer = setTimeout(() => {
75+
resolve("expired");
76+
}, TELEMETRY_FINAL_DRAIN_TIMEOUT_MS);
77+
});
78+
let outcome: "drained" | "expired" = "drained";
79+
try {
80+
if (bridge.inFlight != undefined) {
81+
outcome = await Promise.race([bridge.inFlight.then(() => "drained" as const), deadline]);
82+
}
83+
if (outcome === "drained") {
84+
outcome = await Promise.race([drainBridge(bridge).then(() => "drained" as const), deadline]);
85+
}
86+
} finally {
87+
clearTimeout(timer);
88+
}
89+
if (outcome === "expired") {
90+
log.debug(
91+
`Worker telemetry final drain timed out after ${TELEMETRY_FINAL_DRAIN_TIMEOUT_MS} ms`,
92+
);
93+
}
94+
}
95+
6896
function stopBridgeTimer(bridge: TelemetryBridge | undefined): void {
6997
if (bridge?.timer != undefined) {
7098
clearInterval(bridge.timer);
@@ -128,6 +156,9 @@ export class WorkerSerializedIterableSource implements ISerializedIterableSource
128156
await remote.enableTelemetry(`worker:${++nextWorkerRealmOrdinal}`);
129157
} catch (err: unknown) {
130158
log.debug("Failed to enable worker telemetry", err);
159+
// Stop polling a worker without telemetry; the bridge stays registered so terminate() still
160+
// performs its (bounded) final drain.
161+
stopBridgeTimer(bridge);
131162
}
132163
if (this.#telemetryBridge !== bridge) {
133164
// terminate()/re-initialize() replaced this bridge during the await: make sure it stays
@@ -236,10 +267,9 @@ export class WorkerSerializedIterableSource implements ISerializedIterableSource
236267
try {
237268
if (telemetryBridge != undefined) {
238269
// Final drain (after the worker flushed its final pool counters, before the worker is
239-
// killed). Wait for any periodic drain in flight so batches are ingested in order.
270+
// killed), bounded so a dead worker cannot block the disposal below.
240271
stopBridgeTimer(telemetryBridge);
241-
await telemetryBridge.inFlight;
242-
await drainBridge(telemetryBridge);
272+
await finalDrainBridge(telemetryBridge);
243273
}
244274
} finally {
245275
// Always dispose the captured worker, even if its graceful terminate() rejected.

‎packages/suite-base/src/players/IterablePlayer/WorkerSerializedIterableSourceWorker.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ describe("WorkerSerializedIterableSourceWorker telemetry", () => {
8989
]);
9090
});
9191

92-
it("lets a pool created during initialize() report through the worker sink, flushes on terminate and disposes on the final drain", async () => {
92+
it("lets a pool created during initialize() report through the worker sink and publishes and disposes it on the final drain", async () => {
9393
// Given: telemetry enabled before initialize(), as done by WorkerSerializedIterableSource.
9494
const worker = new WorkerSerializedIterableSourceWorker(new PoolBackedSource());
9595
worker.enableTelemetry("worker:1");

0 commit comments

Comments
 (0)