mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-02 07:31:33 +00:00
fix: deep health checks no longer reuse passive snapshots (#116681)
* fix(gateway): preserve explicit health probes * chore: defer release note to release process
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
53e9e21365
commit
2aabde7262
@@ -42,6 +42,16 @@ function createHealthSummary(): HealthSummary {
|
||||
};
|
||||
}
|
||||
|
||||
function createDeferredHealthSummary() {
|
||||
let resolve!: (summary: HealthSummary) => void;
|
||||
let reject!: (error: unknown) => void;
|
||||
const promise = new Promise<HealthSummary>((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<HealthSummary>((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<HealthSummary>((resolve) => {
|
||||
resolveSensitive = () => resolve(sensitiveSummary);
|
||||
}),
|
||||
)
|
||||
.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise<HealthSummary>((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([
|
||||
|
||||
@@ -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<HealthSummary> | null = null;
|
||||
let sensitiveHealthRefresh: Promise<HealthSummary> | null = null;
|
||||
let broadcastHealthUpdate: ((snap: HealthSummary) => void) | null = null;
|
||||
|
||||
type HealthAudience = "public" | "admin";
|
||||
type HealthRefreshStrength = "passive" | "probe";
|
||||
type HealthRefreshOperation = {
|
||||
generation: number;
|
||||
promise: Promise<HealthSummary>;
|
||||
};
|
||||
type HealthRefreshState = {
|
||||
nextGeneration: number;
|
||||
committedGeneration: number;
|
||||
inFlight: Record<HealthRefreshStrength, HealthRefreshOperation | null>;
|
||||
};
|
||||
|
||||
const healthRefreshStates: Record<HealthAudience, HealthRefreshState> = {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user