vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-02 00:51:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(diagnostics): prevent heartbeat stalls from aborting healthy runs (#108455)

Browse Source 
* fix(diagnostics): scale heartbeat recovery gate

* fix(diagnostics): harden stalled-heartbeat recovery

Co-authored-by: Harjoth Khara <harjoth.khara@gmail.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
Harjoth Khara
2026-07-16 00:09:01 -07:00
committed by GitHub
parent 6f13a5802f
commit 96e5ca3599
2 changed files with 174 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
src/logging/diagnostic.test.ts
View File

@@ -456,6 +456,167 @@ describe("stuck session diagnostics threshold", () => {
);
});
it.each([
{ stuckSessionAbortMs: 1_000, heartbeatOverdueMs: 1_000 },
{ stuckSessionAbortMs: 30_000, heartbeatOverdueMs: 1_000 },
{ stuckSessionAbortMs: 31_000, heartbeatOverdueMs: 1_000 },
{ stuckSessionAbortMs: 45_000, heartbeatOverdueMs: 15_000 },
{ stuckSessionAbortMs: 89_000, heartbeatOverdueMs: 59_000 },
])(
"defers a liveness-delayed heartbeat with a $stuckSessionAbortMs ms abort threshold",
({ stuckSessionAbortMs, heartbeatOverdueMs }) => {
const recoverStuckSession = vi.fn();
const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
vi.setSystemTime(0);
startDiagnosticHeartbeat(
{
diagnostics: {
enabled: true,
stuckSessionWarnMs: 1_000,
stuckSessionAbortMs,
},
},
{ recoverStuckSession },
);
logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" });
vi.setSystemTime(heartbeatOverdueMs);
vi.advanceTimersByTime(30_000);
expectLoggerMessageContaining(warnSpy, "liveness heartbeat delayed");
expect(recoverStuckSession).not.toHaveBeenCalled();
vi.advanceTimersByTime(30_000);
expectRecoveryCall(
recoverStuckSession,
{ sessionId: "s1", sessionKey: "main", queueDepth: 0, allowActiveAbort: true },
["ageMs", "stateGeneration"],
);
},
);
it.each([
{ stuckSessionAbortMs: 1_000, schedulingJitterMs: 1 },
{ stuckSessionAbortMs: 30_000, schedulingJitterMs: 999 },
])(
"recovers at a $stuckSessionAbortMs ms abort threshold with ordinary scheduling jitter",
({ stuckSessionAbortMs, schedulingJitterMs }) => {
const recoverStuckSession = vi.fn();
const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
vi.setSystemTime(0);
startDiagnosticHeartbeat(
{
diagnostics: {
enabled: true,
stuckSessionWarnMs: 1_000,
stuckSessionAbortMs,
},
},
{ recoverStuckSession },
);
logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" });
vi.setSystemTime(schedulingJitterMs);
vi.advanceTimersByTime(30_000);
expectNoLoggerMessageContaining(warnSpy, "liveness heartbeat delayed");
expectRecoveryCall(
recoverStuckSession,
{ sessionId: "s1", sessionKey: "main", queueDepth: 0, allowActiveAbort: true },
["ageMs", "stateGeneration"],
);
},
);
it("defers a material heartbeat stall even when elapsed time is below the abort threshold", () => {
const recoverStuckSession = vi.fn();
const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
vi.setSystemTime(0);
startDiagnosticHeartbeat(
{
diagnostics: {
enabled: true,
stuckSessionWarnMs: 1_000,
stuckSessionAbortMs: 45_000,
},
},
{ recoverStuckSession },
);
logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" });
vi.advanceTimersByTime(20_000);
markDiagnosticSessionProgress({ sessionId: "s1", sessionKey: "main" });
markDiagnosticRunProgressForTest({
sessionId: "s1",
sessionKey: "main",
reason: "embedded_run:progress",
});
vi.advanceTimersByTime(10_000);
vi.setSystemTime(35_000);
vi.advanceTimersByTime(30_000);
expectLoggerMessageContaining(warnSpy, "liveness heartbeat delayed");
expect(recoverStuckSession).not.toHaveBeenCalled();
vi.advanceTimersByTime(30_000);
expectRecoveryCall(
recoverStuckSession,
{ sessionId: "s1", sessionKey: "main", queueDepth: 0, allowActiveAbort: true },
["ageMs", "stateGeneration"],
);
});
it("does not let ordinary heartbeat jitter consume the remaining abort budget", () => {
const recoverStuckSession = vi.fn();
const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
vi.setSystemTime(0);
startDiagnosticHeartbeat(
{
diagnostics: {
enabled: true,
stuckSessionWarnMs: 1_000,
stuckSessionAbortMs: 45_000,
},
},
{ recoverStuckSession },
);
logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" });
vi.advanceTimersByTime(15_500);
markDiagnosticSessionProgress({ sessionId: "s1", sessionKey: "main" });
markDiagnosticRunProgressForTest({
sessionId: "s1",
sessionKey: "main",
reason: "embedded_run:progress",
});
vi.advanceTimersByTime(14_500);
vi.setSystemTime(30_999);
vi.advanceTimersByTime(30_000);
expectNoLoggerMessageContaining(warnSpy, "liveness heartbeat delayed");
expect(recoverStuckSession).not.toHaveBeenCalled();
vi.advanceTimersByTime(30_000);
expectRecoveryCall(
recoverStuckSession,
{ sessionId: "s1", sessionKey: "main", queueDepth: 0, allowActiveAbort: true },
["ageMs", "stateGeneration"],
);
});
it("does not warn while a processing session continues reporting progress", () => {
const events: DiagnosticEventPayload[] = [];
const unsubscribe = onDiagnosticEvent((event) => {

24
src/logging/diagnostic.ts
View File

@@ -89,7 +89,6 @@ const DEFAULT_LIVENESS_EVENT_LOOP_UTILIZATION_WARN = 0.95;
const DEFAULT_LIVENESS_CPU_CORE_RATIO_WARN = 0.9;
const DEFAULT_LIVENESS_WARN_COOLDOWN_MS = 120_000;
const DIAGNOSTIC_HEARTBEAT_INTERVAL_MS = 30_000;
const DIAGNOSTIC_HEARTBEAT_DELAY_RECOVERY_SKIP_MS = 3 * DIAGNOSTIC_HEARTBEAT_INTERVAL_MS;
const loadStuckSessionRecoveryRuntime = createLazyRuntimeModule(
() => import("./diagnostic-stuck-session-recovery.runtime.js"),
);
@@ -1237,15 +1236,18 @@ export function startDiagnosticHeartbeat(
const stuckSessionAbortMs = resolveStuckSessionAbortMs(heartbeatConfig, stuckSessionWarnMs);
const compactionSafetyTimeoutMs = resolveCompactionTimeoutMs(heartbeatConfig);
const now = Date.now();
const tickDelayMs =
const heartbeatElapsedMs =
lastDiagnosticHeartbeatTickAt === undefined ? 0 : now - lastDiagnosticHeartbeatTickAt;
lastDiagnosticHeartbeatTickAt = now;
// A late interval tick means the process was stalled, not the sessions.
// Acting on inflated ages here can abort healthy runs; the next on-time tick decides.
const skipRecoveryThisTick = tickDelayMs > DIAGNOSTIC_HEARTBEAT_DELAY_RECOVERY_SKIP_MS;
if (skipRecoveryThisTick) {
const heartbeatOverdueMs = Math.max(0, heartbeatElapsedMs - DIAGNOSTIC_HEARTBEAT_INTERVAL_MS);
// Observe ordinary timer jitter at the scheduled tick so it cannot consume
// a run's remaining recovery budget. Material lateness can also hide queued
// progress events, so the next healthy heartbeat owns recovery instead.
const recoveryObservationNow = now - heartbeatOverdueMs;
const shouldDeferRecovery = heartbeatOverdueMs >= DEFAULT_LIVENESS_EVENT_LOOP_DELAY_WARN_MS;
if (shouldDeferRecovery) {
diag.warn(
`liveness heartbeat delayed ${Math.round(tickDelayMs)}ms; deferring recovery decisions`,
`liveness heartbeat delayed ${Math.round(heartbeatElapsedMs)}ms; deferring recovery decisions`,
);
}
pruneDiagnosticSessionStates(now, true);
@@ -1305,10 +1307,10 @@ export function startDiagnosticHeartbeat(
});
for (const [, state] of diagnosticSessionStates) {
const ageMs = now - state.lastActivity;
const ageMs = recoveryObservationNow - state.lastActivity;
const activity = getDiagnosticSessionActivitySnapshot(
{ sessionId: state.sessionId, sessionKey: state.sessionKey },
now,
recoveryObservationNow,
);
const idleQueuedRecoverableStall = isIdleQueuedRecoverableSessionStall({
state,
@@ -1330,7 +1332,7 @@ export function startDiagnosticHeartbeat(
thresholdMs: stuckSessionWarnMs,
abortThresholdMs: stuckSessionAbortMs,
});
if (classification?.recoveryEligible && !skipRecoveryThisTick) {
if (classification?.recoveryEligible && !shouldDeferRecovery) {
requestStuckSessionRecovery({
recover: opts?.recoverStuckSession ?? recoverStuckSession,
classification,
@@ -1348,7 +1350,7 @@ export function startDiagnosticHeartbeat(
});
} else if (
classification &&
!skipRecoveryThisTick &&
!shouldDeferRecovery &&
isActiveAbortRecoveryEligible({
classification,
activity,