diff --git a/extensions/crabbox/src/crabbox-worker-heartbeat.ts b/extensions/crabbox/src/crabbox-worker-heartbeat.ts index ceee77f9218d..655b7c321607 100644 --- a/extensions/crabbox/src/crabbox-worker-heartbeat.ts +++ b/extensions/crabbox/src/crabbox-worker-heartbeat.ts @@ -1,12 +1,12 @@ import type { SpawnResult } from "openclaw/plugin-sdk/process-runtime"; import { crabboxCommandError } from "./crabbox-worker-command-error.js"; -const CRABBOX_HEARTBEAT_UPGRADE = - "upgrade Crabbox to a release that includes `crabbox heartbeat` (added after v0.43.0)"; +const CRABBOX_HEARTBEAT_UPGRADE = "upgrade Crabbox to v0.44.0 or newer for `crabbox heartbeat`"; type HeartbeatContext = { binary: string; heartbeatIntervalMs: number; + heartbeatTimeoutMs: number; id: string; idleTimeout: string; provider: string; diff --git a/extensions/crabbox/src/crabbox-worker-profile.ts b/extensions/crabbox/src/crabbox-worker-profile.ts index 14f3abd35c9f..93d7622e17ad 100644 --- a/extensions/crabbox/src/crabbox-worker-profile.ts +++ b/extensions/crabbox/src/crabbox-worker-profile.ts @@ -7,6 +7,7 @@ import { type WorkerProfile, } from "openclaw/plugin-sdk/plugin-entry"; import { normalizeOptionalString as nonEmptyString } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { CRABBOX_HEARTBEAT_TIMEOUT_MS } from "./crabbox-worker-timeouts.js"; export { nonEmptyString }; @@ -39,6 +40,7 @@ type CrabboxProfile = { class: string; desktop?: boolean; heartbeatIntervalMs: number; + heartbeatTimeoutMs: number; idleTimeout: string; provider: string; ttl: string; @@ -60,14 +62,18 @@ type IsExecutable = (candidate: string) => boolean; export const CRABBOX_WORKER_PROVIDER_ID = "crabbox"; -function requirePositiveDuration(value: unknown, key: string): string { +function requirePositiveDuration( + value: unknown, + key: string, +): { duration: string; milliseconds: number } { const duration = nonEmptyString(value); - if (!duration || parsePositiveGoDurationNanoseconds(duration) === undefined) { + const nanoseconds = duration ? parsePositiveGoDurationNanoseconds(duration) : undefined; + if (!duration || nanoseconds === undefined) { throw new WorkerProviderError( `Crabbox profile ${key} must be a positive Go duration such as 60m`, ); } - return duration; + return { duration, milliseconds: Number(nanoseconds) / 1_000_000 }; } function parsePositiveGoDurationNanoseconds(duration: string): bigint | undefined { @@ -98,12 +104,7 @@ function parsePositiveGoDurationNanoseconds(duration: string): bigint | undefine return total > 0n ? total : undefined; } -function heartbeatIntervalMs(idleTimeout: string): number { - const idleNanoseconds = parsePositiveGoDurationNanoseconds(idleTimeout); - if (idleNanoseconds === undefined) { - throw new Error("Crabbox heartbeat requires a positive idle timeout"); - } - const idleTimeoutMs = Number(idleNanoseconds) / 1_000_000; +function heartbeatIntervalMs(idleTimeoutMs: number): number { const referenceIntervalMs = Math.max(5_000, Math.min(60_000, idleTimeoutMs / 3)); // Crabbox's floor can exceed short accepted timeouts. Keep renewal ahead of // coordinator idle expiry without changing the profile contract. @@ -125,8 +126,11 @@ export function parseCrabboxProfile(profile: WorkerProfile): CrabboxProfile { if (!machineClass) { throw new WorkerProviderError("Crabbox profile class must be a non-empty string"); } - const ttl = requirePositiveDuration(profile.ttl, "ttl"); - const idleTimeout = requirePositiveDuration(profile.idleTimeout, "idleTimeout"); + const { duration: ttl } = requirePositiveDuration(profile.ttl, "ttl"); + const { duration: idleTimeout, milliseconds: idleTimeoutMs } = requirePositiveDuration( + profile.idleTimeout, + "idleTimeout", + ); const binaryValue = profile.binary; const binary = binaryValue === undefined ? undefined : nonEmptyString(binaryValue); if (binaryValue !== undefined && !binary) { @@ -153,7 +157,11 @@ export function parseCrabboxProfile(profile: WorkerProfile): CrabboxProfile { binary, class: machineClass, desktop, - heartbeatIntervalMs: heartbeatIntervalMs(idleTimeout), + heartbeatIntervalMs: heartbeatIntervalMs(idleTimeoutMs), + heartbeatTimeoutMs: Math.min( + CRABBOX_HEARTBEAT_TIMEOUT_MS, + Math.max(1, Math.floor(idleTimeoutMs / 2)), + ), idleTimeout, provider, setup, diff --git a/extensions/crabbox/src/crabbox-worker-provider.test.ts b/extensions/crabbox/src/crabbox-worker-provider.test.ts index 3da9f640ae7c..9d2764888d6c 100644 --- a/extensions/crabbox/src/crabbox-worker-provider.test.ts +++ b/extensions/crabbox/src/crabbox-worker-provider.test.ts @@ -1956,20 +1956,25 @@ describe("Crabbox worker provider", () => { }); it.each([ - { idleTimeout: "1s", idleTimeoutMs: 1_000, intervalMs: 500 }, - { idleTimeout: "2s", idleTimeoutMs: 2_000, intervalMs: 1_000 }, - { idleTimeout: "5s", idleTimeoutMs: 5_000, intervalMs: 2_500 }, - { idleTimeout: "12s", idleTimeoutMs: 12_000, intervalMs: 5_000 }, - { idleTimeout: "30s", idleTimeoutMs: 30_000, intervalMs: 10_000 }, - { idleTimeout: "6m", idleTimeoutMs: 360_000, intervalMs: 60_000 }, + { idleTimeout: "1s", idleTimeoutMs: 1_000, intervalMs: 500, timeoutMs: 500 }, + { idleTimeout: "2s", idleTimeoutMs: 2_000, intervalMs: 1_000, timeoutMs: 1_000 }, + { idleTimeout: "5s", idleTimeoutMs: 5_000, intervalMs: 2_500, timeoutMs: 2_500 }, + { idleTimeout: "12s", idleTimeoutMs: 12_000, intervalMs: 5_000, timeoutMs: 6_000 }, + { idleTimeout: "30s", idleTimeoutMs: 30_000, intervalMs: 10_000, timeoutMs: 15_000 }, + { idleTimeout: "6m", idleTimeoutMs: 360_000, intervalMs: 60_000, timeoutMs: 150_000 }, + { idleTimeout: "45m", idleTimeoutMs: 2_700_000, intervalMs: 60_000, timeoutMs: 150_000 }, ])( "heartbeats an active lease every $intervalMs ms for idleTimeout=$idleTimeout", - async ({ idleTimeout, idleTimeoutMs, intervalMs }) => { + async ({ idleTimeout, idleTimeoutMs, intervalMs, timeoutMs }) => { vi.useFakeTimers(); const calls: string[][] = []; + const heartbeatTimeouts: number[] = []; const profile = { ...PROFILE, idleTimeout }; - const provider = providerWithRunner(async (argv) => { + const provider = providerWithRunner(async (argv, options) => { calls.push(argv); + if (argv[1] === "heartbeat") { + heartbeatTimeouts.push(options.timeoutMs); + } return argv[1] === "inspect" ? commandResult({ stdout: inspectJson({ sshHostKey: HOST_KEY }) }) : commandResult(); @@ -1994,6 +1999,7 @@ describe("Crabbox worker provider", () => { "--json", ], ]); + expect(heartbeatTimeouts).toEqual([timeoutMs]); await vi.advanceTimersByTimeAsync(intervalMs - 1); expect(heartbeatCalls()).toHaveLength(1); @@ -2083,7 +2089,7 @@ describe("Crabbox worker provider", () => { expect(calls.filter((argv) => argv[1] === "heartbeat")).toHaveLength(1); expect(warnings).toEqual([ - `Crabbox heartbeat is unavailable for worker lease ${LEASE_ID}; upgrade Crabbox to a release that includes \`crabbox heartbeat\` (added after v0.43.0); cloud worker machines may be reaped after 60m of coordinator-idle time`, + `Crabbox heartbeat is unavailable for worker lease ${LEASE_ID}; upgrade Crabbox to v0.44.0 or newer for \`crabbox heartbeat\`; cloud worker machines may be reaped after 60m of coordinator-idle time`, ]); } finally { await provider.destroy(lease); diff --git a/extensions/crabbox/src/crabbox-worker-provider.ts b/extensions/crabbox/src/crabbox-worker-provider.ts index da32883f8c42..606a46a261de 100644 --- a/extensions/crabbox/src/crabbox-worker-provider.ts +++ b/extensions/crabbox/src/crabbox-worker-provider.ts @@ -77,7 +77,7 @@ type CrabboxProfile = ReturnType; type LeaseCommandContext = { binary: string; id: string; provider: string }; type LeaseHeartbeatContext = LeaseCommandContext & - Pick; + Pick; type ProvisionInspectContext = Omit & { deadline: number; inspect: ParsedInspect; @@ -433,7 +433,7 @@ export function createCrabboxWorkerProvider( binary: context.binary, runCommand, signal, - timeoutMs: Math.min(CRABBOX_LIFECYCLE_TIMEOUT_MS, context.heartbeatIntervalMs), + timeoutMs: context.heartbeatTimeoutMs, }), warn, }); @@ -466,6 +466,7 @@ export function createCrabboxWorkerProvider( return { binary: resolveBinary(parsed.binary), heartbeatIntervalMs: parsed.heartbeatIntervalMs, + heartbeatTimeoutMs: parsed.heartbeatTimeoutMs, id: lease.leaseId, idleTimeout: parsed.idleTimeout, provider: parsed.provider, @@ -637,6 +638,7 @@ export function createCrabboxWorkerProvider( heartbeats.start({ binary, heartbeatIntervalMs: parsed.heartbeatIntervalMs, + heartbeatTimeoutMs: parsed.heartbeatTimeoutMs, id: leaseId, idleTimeout: parsed.idleTimeout, provider: parsed.provider, diff --git a/extensions/crabbox/src/crabbox-worker-timeouts.ts b/extensions/crabbox/src/crabbox-worker-timeouts.ts index 7ab19458240a..ecfaa3ebd6fb 100644 --- a/extensions/crabbox/src/crabbox-worker-timeouts.ts +++ b/extensions/crabbox/src/crabbox-worker-timeouts.ts @@ -5,6 +5,8 @@ type CrabboxProvisionTimeoutProfile = { export const CRABBOX_WARMUP_TIMEOUT_MS = 240_000; export const CRABBOX_LIFECYCLE_TIMEOUT_MS = 60_000; +// AWS coordinator heartbeat latency reached 107.6 seconds in production measurements. +export const CRABBOX_HEARTBEAT_TIMEOUT_MS = 150_000; // `providers --json` is a static compiled report: no network, no credentials, // measured well under a second. The picker awaits it, so cap it far below the