From 2aabde72623d71e19a16f11a0b080eb13f8a8960 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 30 Jul 2026 23:15:19 -0700 Subject: [PATCH] fix: deep health checks no longer reuse passive snapshots (#116681) * fix(gateway): preserve explicit health probes * chore: defer release note to release process --- src/gateway/server/health-state.test.ts | 254 ++++++++++++++++++++---- src/gateway/server/health-state.ts | 123 ++++++++---- 2 files changed, 296 insertions(+), 81 deletions(-) diff --git a/src/gateway/server/health-state.test.ts b/src/gateway/server/health-state.test.ts index 1ca487285c97..a5d95234b0e3 100644 --- a/src/gateway/server/health-state.test.ts +++ b/src/gateway/server/health-state.test.ts @@ -42,6 +42,16 @@ function createHealthSummary(): HealthSummary { }; } +function createDeferredHealthSummary() { + let resolve!: (summary: HealthSummary) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + async function loadHealthState() { vi.resetModules(); collectGatewayHealthSnapshotMock.mockReset(); @@ -54,28 +64,158 @@ describe("refreshGatewayHealthSnapshot", () => { vi.restoreAllMocks(); }); - it("keeps refreshes coalesced while preserving the first probe intent", async () => { + it("does not let a post-connect passive refresh absorb an explicit probe", async () => { const healthState = await loadHealthState(); - let resolveSnapshot: ((summary: HealthSummary) => void) | undefined; - collectGatewayHealthSnapshotMock.mockImplementation( - () => - new Promise((resolve) => { - resolveSnapshot = resolve; - }), - ); + const passiveDeferred = createDeferredHealthSummary(); + const probeDeferred = createDeferredHealthSummary(); + const passiveSummary = createHealthSummary(); + const probeSummary = createHealthSummary(); + const broadcast = vi.fn(); + collectGatewayHealthSnapshotMock + .mockImplementationOnce(() => passiveDeferred.promise) + .mockImplementationOnce(() => probeDeferred.promise); + healthState.setBroadcastHealthUpdate(broadcast); + const version = healthState.getHealthVersion(); - const first = healthState.refreshGatewayHealthSnapshot({ probe: false }); - const second = healthState.refreshGatewayHealthSnapshot({ probe: true }); + const postConnectRefresh = healthState.refreshGatewayHealthSnapshot({ probe: false }); + const explicitProbe = healthState.refreshGatewayHealthSnapshot({ probe: true }); - expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(1); - expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledWith({ + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(healthSnapshotCallArg()).toMatchObject({ audience: "public", probe: false, runtimeSnapshot: undefined, }); - expect(Object.hasOwn(healthSnapshotCallArg() ?? {}, "eventLoop")).toBe(false); - resolveSnapshot?.(createHealthSummary()); - await expect(Promise.all([first, second])).resolves.toHaveLength(2); + expect(healthSnapshotCallArg(1)).toMatchObject({ + audience: "public", + probe: true, + runtimeSnapshot: undefined, + }); + + probeDeferred.resolve(probeSummary); + await expect(explicitProbe).resolves.toBe(probeSummary); + expect(healthState.getHealthCache()).toBe(probeSummary); + expect(healthState.getHealthVersion()).toBe(version + 1); + expect(broadcast).toHaveBeenCalledTimes(1); + expect(broadcast).toHaveBeenLastCalledWith(probeSummary); + + passiveDeferred.resolve(passiveSummary); + await expect(postConnectRefresh).resolves.toBe(passiveSummary); + expect(healthState.getHealthCache()).toBe(probeSummary); + expect(healthState.getHealthVersion()).toBe(version + 1); + expect(broadcast).toHaveBeenCalledTimes(1); + }); + + it("publishes both generations in order when the passive refresh finishes first", async () => { + const healthState = await loadHealthState(); + const passiveDeferred = createDeferredHealthSummary(); + const probeDeferred = createDeferredHealthSummary(); + const passiveSummary = createHealthSummary(); + const probeSummary = createHealthSummary(); + const broadcast = vi.fn(); + collectGatewayHealthSnapshotMock + .mockImplementationOnce(() => passiveDeferred.promise) + .mockImplementationOnce(() => probeDeferred.promise); + healthState.setBroadcastHealthUpdate(broadcast); + const version = healthState.getHealthVersion(); + + const passive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + const probe = healthState.refreshGatewayHealthSnapshot({ probe: true }); + + passiveDeferred.resolve(passiveSummary); + await expect(passive).resolves.toBe(passiveSummary); + expect(healthState.getHealthCache()).toBe(passiveSummary); + expect(healthState.getHealthVersion()).toBe(version + 1); + + probeDeferred.resolve(probeSummary); + await expect(probe).resolves.toBe(probeSummary); + expect(healthState.getHealthCache()).toBe(probeSummary); + expect(healthState.getHealthVersion()).toBe(version + 2); + expect(broadcast.mock.calls.map(([summary]) => summary)).toEqual([ + passiveSummary, + probeSummary, + ]); + }); + + it("lets a passive refresh join an in-flight explicit probe", async () => { + const healthState = await loadHealthState(); + const probeDeferred = createDeferredHealthSummary(); + const probeSummary = createHealthSummary(); + collectGatewayHealthSnapshotMock.mockImplementationOnce(() => probeDeferred.promise); + + const probe = healthState.refreshGatewayHealthSnapshot({ probe: true }); + const passive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(1); + expect(healthSnapshotCallArg()?.probe).toBe(true); + probeDeferred.resolve(probeSummary); + await expect(Promise.all([probe, passive])).resolves.toEqual([probeSummary, probeSummary]); + }); + + it("coalesces concurrent explicit probe waiters", async () => { + const healthState = await loadHealthState(); + const probeDeferred = createDeferredHealthSummary(); + const probeSummary = createHealthSummary(); + collectGatewayHealthSnapshotMock.mockImplementationOnce(() => probeDeferred.promise); + + const first = healthState.refreshGatewayHealthSnapshot({ probe: true }); + const second = healthState.refreshGatewayHealthSnapshot({ probe: true }); + + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(1); + probeDeferred.resolve(probeSummary); + await expect(Promise.all([first, second])).resolves.toEqual([probeSummary, probeSummary]); + }); + + it("retains a displaced passive refresh after a faster probe settles", async () => { + const healthState = await loadHealthState(); + const passiveDeferred = createDeferredHealthSummary(); + const probeDeferred = createDeferredHealthSummary(); + const passiveSummary = createHealthSummary(); + collectGatewayHealthSnapshotMock + .mockImplementationOnce(() => passiveDeferred.promise) + .mockImplementationOnce(() => probeDeferred.promise); + + const firstPassive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + const probe = healthState.refreshGatewayHealthSnapshot({ probe: true }); + probeDeferred.reject(new Error("probe failed")); + await expect(probe).rejects.toThrow("probe failed"); + + const secondPassive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); + passiveDeferred.resolve(passiveSummary); + await expect(Promise.all([firstPassive, secondPassive])).resolves.toEqual([ + passiveSummary, + passiveSummary, + ]); + }); + + it("detaches an older passive refresh after a newer probe succeeds", async () => { + const healthState = await loadHealthState(); + const firstPassiveDeferred = createDeferredHealthSummary(); + const probeDeferred = createDeferredHealthSummary(); + const secondPassiveDeferred = createDeferredHealthSummary(); + const firstPassiveSummary = createHealthSummary(); + const probeSummary = createHealthSummary(); + const secondPassiveSummary = createHealthSummary(); + collectGatewayHealthSnapshotMock + .mockImplementationOnce(() => firstPassiveDeferred.promise) + .mockImplementationOnce(() => probeDeferred.promise) + .mockImplementationOnce(() => secondPassiveDeferred.promise); + + const firstPassive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + const probe = healthState.refreshGatewayHealthSnapshot({ probe: true }); + probeDeferred.resolve(probeSummary); + await expect(probe).resolves.toBe(probeSummary); + + const secondPassive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(3); + secondPassiveDeferred.resolve(secondPassiveSummary); + await expect(secondPassive).resolves.toBe(secondPassiveSummary); + expect(healthState.getHealthCache()).toBe(secondPassiveSummary); + + firstPassiveDeferred.resolve(firstPassiveSummary); + await expect(firstPassive).resolves.toBe(firstPassiveSummary); + expect(healthState.getHealthCache()).toBe(secondPassiveSummary); }); it("passes event-loop health only when the hook returns a snapshot", async () => { @@ -180,42 +320,76 @@ describe("refreshGatewayHealthSnapshot", () => { expect(broadcast).toHaveBeenCalledWith(safeSummary); }); - it("keeps sensitive and public refreshes on separate in-flight promises", async () => { + it("keeps strength-aware admin and public refresh lanes isolated", async () => { const healthState = await loadHealthState(); - const sensitiveSummary = createHealthSummary(); - const safeSummary = createHealthSummary(); - let resolveSensitive: (() => void) | undefined; - let resolveSafe: (() => void) | undefined; + const adminPassiveDeferred = createDeferredHealthSummary(); + const publicProbeDeferred = createDeferredHealthSummary(); + const adminProbeDeferred = createDeferredHealthSummary(); + const adminPassiveSummary = createHealthSummary(); + const publicProbeSummary = createHealthSummary(); + const adminProbeSummary = createHealthSummary(); collectGatewayHealthSnapshotMock - .mockImplementationOnce( - () => - new Promise((resolve) => { - resolveSensitive = () => resolve(sensitiveSummary); - }), - ) - .mockImplementationOnce( - () => - new Promise((resolve) => { - resolveSafe = () => resolve(safeSummary); - }), - ); + .mockImplementationOnce(() => adminPassiveDeferred.promise) + .mockImplementationOnce(() => publicProbeDeferred.promise) + .mockImplementationOnce(() => adminProbeDeferred.promise); - const sensitive = healthState.refreshGatewayHealthSnapshot({ + const adminPassive = healthState.refreshGatewayHealthSnapshot({ + probe: false, + includeSensitive: true, + }); + const publicProbe = healthState.refreshGatewayHealthSnapshot({ probe: true }); + const publicPassive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + const adminProbe = healthState.refreshGatewayHealthSnapshot({ probe: true, includeSensitive: true, }); - const safe = healthState.refreshGatewayHealthSnapshot({ probe: false }); - expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(3); expect(healthSnapshotCallArg()?.audience).toBe("admin"); expect(healthSnapshotCallArg(1)?.audience).toBe("public"); + expect(healthSnapshotCallArg(2)?.audience).toBe("admin"); + expect(healthSnapshotCallArg()?.probe).toBe(false); + expect(healthSnapshotCallArg(1)?.probe).toBe(true); + expect(healthSnapshotCallArg(2)?.probe).toBe(true); - resolveSensitive?.(); - resolveSafe?.(); + adminProbeDeferred.resolve(adminProbeSummary); + publicProbeDeferred.resolve(publicProbeSummary); + adminPassiveDeferred.resolve(adminPassiveSummary); - await expect(sensitive).resolves.toBe(sensitiveSummary); - await expect(safe).resolves.toBe(safeSummary); - expect(healthState.getHealthCache()).toBe(safeSummary); + await expect(adminProbe).resolves.toBe(adminProbeSummary); + await expect(Promise.all([publicProbe, publicPassive])).resolves.toEqual([ + publicProbeSummary, + publicProbeSummary, + ]); + await expect(adminPassive).resolves.toBe(adminPassiveSummary); + expect(healthState.getHealthCache()).toBe(publicProbeSummary); + }); + + it("recovers each strength lane after rejection without discarding an older success", async () => { + const healthState = await loadHealthState(); + const passiveDeferred = createDeferredHealthSummary(); + const probeDeferred = createDeferredHealthSummary(); + const passiveSummary = createHealthSummary(); + const recoveredProbeSummary = createHealthSummary(); + collectGatewayHealthSnapshotMock + .mockImplementationOnce(() => passiveDeferred.promise) + .mockImplementationOnce(() => probeDeferred.promise) + .mockResolvedValueOnce(recoveredProbeSummary); + + const passive = healthState.refreshGatewayHealthSnapshot({ probe: false }); + const probe = healthState.refreshGatewayHealthSnapshot({ probe: true }); + probeDeferred.reject(new Error("probe failed")); + await expect(probe).rejects.toThrow("probe failed"); + + passiveDeferred.resolve(passiveSummary); + await expect(passive).resolves.toBe(passiveSummary); + expect(healthState.getHealthCache()).toBe(passiveSummary); + + await expect(healthState.refreshGatewayHealthSnapshot({ probe: true })).resolves.toBe( + recoveredProbeSummary, + ); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(3); + expect(healthState.getHealthCache()).toBe(recoveredProbeSummary); }); it.each([ diff --git a/src/gateway/server/health-state.ts b/src/gateway/server/health-state.ts index 8a34fac1ab4b..32e263643bd3 100644 --- a/src/gateway/server/health-state.ts +++ b/src/gateway/server/health-state.ts @@ -18,10 +18,33 @@ import type { GatewayEventLoopHealth } from "./event-loop-health.js"; let presenceVersion = 1; let healthVersion = 1; let healthCache: HealthSummary | null = null; -let healthRefresh: Promise | null = null; -let sensitiveHealthRefresh: Promise | null = null; let broadcastHealthUpdate: ((snap: HealthSummary) => void) | null = null; +type HealthAudience = "public" | "admin"; +type HealthRefreshStrength = "passive" | "probe"; +type HealthRefreshOperation = { + generation: number; + promise: Promise; +}; +type HealthRefreshState = { + nextGeneration: number; + committedGeneration: number; + inFlight: Record; +}; + +const healthRefreshStates: Record = { + public: { + nextGeneration: 0, + committedGeneration: 0, + inFlight: { passive: null, probe: null }, + }, + admin: { + nextGeneration: 0, + committedGeneration: 0, + inFlight: { passive: null, probe: null }, + }, +}; + export function buildGatewaySnapshot(opts?: { includeSensitive?: boolean }): Snapshot { const cfg = getRuntimeConfig(); const defaultAgentId = resolveDefaultAgentId(cfg); @@ -86,44 +109,62 @@ export async function refreshGatewayHealthSnapshot(opts?: { getConfigReloaderHotReloadStatus?: () => GatewayHotReloadStatus | undefined; }) { const includeSensitive = opts?.includeSensitive === true; - let refresh = includeSensitive ? sensitiveHealthRefresh : healthRefresh; - if (!refresh) { - refresh = (async () => { - let runtimeSnapshot: ChannelRuntimeSnapshot | undefined; - try { - runtimeSnapshot = opts?.getRuntimeSnapshot?.(); - } catch { - runtimeSnapshot = undefined; - } - const eventLoop = opts?.getEventLoopHealth?.(); - const configReloadHotReloadStatus = opts?.getConfigReloaderHotReloadStatus?.(); - const snap = await collectGatewayHealthSnapshot({ - audience: includeSensitive ? "admin" : "public", - probe: opts?.probe !== false, - runtimeSnapshot, - ...(eventLoop ? { eventLoop } : {}), - ...(configReloadHotReloadStatus ? { configReloadHotReloadStatus } : {}), - }); - if (!includeSensitive) { - healthCache = snap; - healthVersion += 1; - if (broadcastHealthUpdate) { - broadcastHealthUpdate(snap); - } - } - return snap; - })().finally(() => { - if (includeSensitive) { - sensitiveHealthRefresh = null; - } else { - healthRefresh = null; - } - }); - if (includeSensitive) { - sensitiveHealthRefresh = refresh; - } else { - healthRefresh = refresh; - } + const audience: HealthAudience = includeSensitive ? "admin" : "public"; + const state = healthRefreshStates[audience]; + const strength: HealthRefreshStrength = opts?.probe === false ? "passive" : "probe"; + // Passive callers can reuse a stronger probe, but an explicit probe must not + // inherit a passive refresh that deliberately skipped live channel checks. + const existing = + strength === "passive" + ? (state.inFlight.probe ?? state.inFlight.passive) + : state.inFlight.probe; + if (existing) { + return existing.promise; } - return refresh; + + const generation = state.nextGeneration + 1; + state.nextGeneration = generation; + const promise = (async () => { + let runtimeSnapshot: ChannelRuntimeSnapshot | undefined; + try { + runtimeSnapshot = opts?.getRuntimeSnapshot?.(); + } catch { + runtimeSnapshot = undefined; + } + const eventLoop = opts?.getEventLoopHealth?.(); + const configReloadHotReloadStatus = opts?.getConfigReloaderHotReloadStatus?.(); + const snap = await collectGatewayHealthSnapshot({ + audience, + probe: strength === "probe", + runtimeSnapshot, + ...(eventLoop ? { eventLoop } : {}), + ...(configReloadHotReloadStatus ? { configReloadHotReloadStatus } : {}), + }); + if ( + strength === "probe" && + state.inFlight.passive && + state.inFlight.passive.generation < generation + ) { + // Existing passive waiters may finish, but new callers must not join + // weaker work after this newer live probe has succeeded. + state.inFlight.passive = null; + } + // Concurrent passive/probe refreshes can finish out of order. Only a + // generation newer than the published cache may advance version/broadcast. + if (!includeSensitive && generation > state.committedGeneration) { + state.committedGeneration = generation; + healthCache = snap; + healthVersion += 1; + if (broadcastHealthUpdate) { + broadcastHealthUpdate(snap); + } + } + return snap; + })().finally(() => { + if (state.inFlight[strength]?.generation === generation) { + state.inFlight[strength] = null; + } + }); + state.inFlight[strength] = { generation, promise }; + return promise; }