From cdba9a038e41c6c807a2b58bd43e83dcf98d74b7 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 04:55:28 -0700 Subject: [PATCH] perf(server): idle servers no longer scan every process every 5 seconds (#19399) Co-authored-by: Claude Opus 5.5 --- .../NativeTelemetryClient.test.ts | 117 ++++++++++++++++-- .../NativeTelemetryClient.ts | 16 +-- .../settings/DiagnosticsSettings.tsx | 10 +- docs/internals/resource-telemetry.md | 8 +- 4 files changed, 130 insertions(+), 21 deletions(-) diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts index 7d365d0c0bacc585a472551da90b6387f249f997..a58bfe985bbdbc80abb3e4486dbf82ec202226ad 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts @@ -1,12 +1,30 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import type { HostPowerSnapshot } from "@t3tools/contracts"; +import { + RESOURCE_MONITOR_PROTOCOL_VERSION, + ResourceMonitorCommand, + type ResourceMonitorHelloEvent, +} from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; +import * as Sink from "effect/Sink"; +import * as Stream from "effect/Stream"; +import { ChildProcessSpawner } from "effect/process"; +import * as ServerConfig from "../config.ts"; +import * as NativeTelemetryClient from "./NativeTelemetryClient.ts"; +import * as ResourceMonitorBinary from "./ResourceMonitorBinary.ts"; import { canCommandNativeTelemetrySidecar, canRequestNativeTelemetryRetry, @@ -40,25 +58,106 @@ describe("resolveNativeSampleIntervalMs", () => { expect(resolveNativeSampleIntervalMs({ ...basePower, onBattery: "true" }, 1)).toBe(5_000); }); - it("slows background telemetry and serves live diagnostics at 1Hz", () => { + it("stops timed sampling without a live subscriber and serves live diagnostics at 1Hz", () => { const unknown: HostPowerSnapshot = { ...basePower, source: "unknown", stale: true, }; - expect(resolveNativeSampleIntervalMs(unknown, 0)).toBe(5_000); + expect(resolveNativeSampleIntervalMs(unknown, 0)).toBe(0); expect(resolveNativeSampleIntervalMs(unknown, 1)).toBe(1_000); - expect( - resolveNativeSampleIntervalMs( - { ...basePower, stale: true, locked: "true", suspended: true }, - 0, - ), - ).toBe(5_000); - expect(resolveNativeSampleIntervalMs(basePower, 0)).toBe(5_000); + expect(resolveNativeSampleIntervalMs({ ...basePower, locked: "true" }, 0)).toBe(0); + expect(resolveNativeSampleIntervalMs({ ...basePower, onBattery: "true" }, 0)).toBe(0); + expect(resolveNativeSampleIntervalMs(basePower, 0)).toBe(0); expect(resolveNativeSampleIntervalMs(basePower, 1)).toBe(1_000); }); }); +const decodeCommandLine = Schema.decodeUnknownSync(Schema.fromJsonString(ResourceMonitorCommand)); + +const hello: ResourceMonitorHelloEvent = { + version: RESOURCE_MONITOR_PROTOCOL_VERSION, + type: "hello", + sidecarVersion: "test", + sidecarPid: 4242, + platform: "linux", + arch: "x64", + capabilities: { + cumulativeCpuTime: true, + currentCpuPercent: true, + residentMemory: true, + virtualMemory: true, + ioBytes: true, + processStartTime: true, + processTree: true, + }, +}; + +describe("NativeTelemetryClient", () => { + it.effect("asks the sidecar for timed samples only while a live subscriber exists", () => + Effect.gen(function* () { + const commands = yield* Queue.unbounded(); + // A sidecar that says hello, stays alive, and records every command it receives. + const spawner = ChildProcessSpawner.make(() => + Effect.succeed( + ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(hello.sidecarPid), + exitCode: Effect.never, + isRunning: Effect.succeed(true), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.forEach((chunk: Uint8Array) => + Effect.forEach(new TextDecoder().decode(chunk).split("\n").filter(Boolean), (line) => + Queue.offer(commands, decodeCommandLine(line)), + ), + ), + stdout: Stream.encodeText(Stream.make(`${JSON.stringify(hello)}\n`)).pipe( + Stream.concat(Stream.never), + ), + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }), + ), + ); + const client = yield* NativeTelemetryClient.make().pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.provideService(ResourceMonitorBinary.ResourceMonitorBinary, { + resolve: Effect.succeed("/test/t3-resource-monitor"), + }), + ); + + const configure = yield* Queue.take(commands); + expect(configure).toMatchObject({ type: "configure", sampleIntervalMs: 0 }); + expect(yield* Queue.take(commands)).toMatchObject({ type: "setExternalProcesses" }); + + const live = yield* Scope.make(); + yield* client.snapshots.pipe(Stream.runDrain, Effect.forkIn(live)); + expect(yield* Queue.take(commands)).toMatchObject({ + type: "setSampleInterval", + sampleIntervalMs: 1_000, + }); + expect(yield* Queue.take(commands)).toMatchObject({ type: "setStreaming", enabled: true }); + + yield* Scope.close(live, Exit.void); + expect(yield* Queue.take(commands)).toMatchObject({ + type: "setSampleInterval", + sampleIntervalMs: 0, + }); + expect(yield* Queue.take(commands)).toMatchObject({ type: "setStreaming", enabled: false }); + expect((yield* client.health).sampleIntervalMs).toBe(0); + }).pipe( + Effect.provide( + Layer.mergeAll( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-native-telemetry-test-" }), + NodeCrypto.layer, + ).pipe(Layer.provideMerge(NodeServices.layer)), + ), + ), + ); +}); + describe("canRequestNativeTelemetryRetry", () => { it("only accepts retry while the supervisor is waiting without a live sidecar", () => { expect(canRequestNativeTelemetryRetry("degraded", false)).toBe(true); diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts index ec8168a69cd4f474a8ee36c77f4385342c3fa476..9d252fadb4f7ab3b34f2020e3d443ce371fd00ad 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts @@ -40,7 +40,8 @@ import * as ServerConfig from "../config.ts"; import { subscribeBeforeSnapshotWithoutMutex } from "../utils/subscribeBeforeSnapshot.ts"; const SAMPLE_INTERVAL_MS = 1_000; -const UNKNOWN_BACKGROUND_SAMPLE_INTERVAL_MS = 5_000; +// The sidecar treats 0 as "sample only on request". +const IDLE_SAMPLE_INTERVAL_MS = 0; const BATTERY_SAMPLE_INTERVAL_MS = 5_000; const CONSTRAINED_SAMPLE_INTERVAL_MS = 15_000; const HANDSHAKE_TIMEOUT = Duration.seconds(5); @@ -258,13 +259,14 @@ function isThermallyConstrained(snapshot: HostPowerSnapshot): boolean { return snapshot.thermalState === "serious" || snapshot.thermalState === "critical"; } +// Without a live subscriber the sidecar does not sample on a timer, so an idle server +// never walks the process table. History then covers only time with a subscriber. export function resolveNativeSampleIntervalMs( snapshot: HostPowerSnapshot, liveSubscriberCount: number, ): number { - if (snapshot.stale || snapshot.source === "unknown") { - return liveSubscriberCount > 0 ? SAMPLE_INTERVAL_MS : UNKNOWN_BACKGROUND_SAMPLE_INTERVAL_MS; - } + if (liveSubscriberCount === 0) return IDLE_SAMPLE_INTERVAL_MS; + if (snapshot.stale || snapshot.source === "unknown") return SAMPLE_INTERVAL_MS; if ( snapshot.suspended || snapshot.locked === "true" || @@ -274,7 +276,7 @@ export function resolveNativeSampleIntervalMs( return CONSTRAINED_SAMPLE_INTERVAL_MS; } if (snapshot.onBattery === "true") return BATTERY_SAMPLE_INTERVAL_MS; - return liveSubscriberCount > 0 ? SAMPLE_INTERVAL_MS : UNKNOWN_BACKGROUND_SAMPLE_INTERVAL_MS; + return SAMPLE_INTERVAL_MS; } export function commitCollectionControlUpdate( @@ -382,7 +384,7 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu updatedAt: initializedAt, }, liveSubscriberCount: 0, - sampleIntervalMs: UNKNOWN_BACKGROUND_SAMPLE_INTERVAL_MS, + sampleIntervalMs: IDLE_SAMPLE_INTERVAL_MS, }); const appliedCollectionControl = yield* Ref.make(yield* Ref.get(collectionControl)); const externalProcesses = yield* Ref.make>([]); @@ -1071,7 +1073,7 @@ export const layerTest = ( lastSampleAt: Option.none(), lastError: Option.some("Resource monitor test implementation is unavailable."), restartCount: 0, - sampleIntervalMs: UNKNOWN_BACKGROUND_SAMPLE_INTERVAL_MS, + sampleIntervalMs: IDLE_SAMPLE_INTERVAL_MS, }); return Layer.succeed( NativeTelemetryClient, diff --git a/apps/web/src/components/settings/DiagnosticsSettings.tsx b/apps/web/src/components/settings/DiagnosticsSettings.tsx index 645c043dee7a317dc7820758ec792b054f6dc481..94e82775c9d58128af134f5a135824eb4dfe93c3 100644 --- a/apps/web/src/components/settings/DiagnosticsSettings.tsx +++ b/apps/web/src/components/settings/DiagnosticsSettings.tsx @@ -1038,11 +1038,17 @@ export function DiagnosticsSettingsPanel() {