From 34857e5f91151d1938eaeb07bff2863fc46ebfc0 Mon Sep 17 00:00:00 2001 From: Kentaro Wakayama Date: Sun, 23 Aug 2026 21:37:03 +0200 Subject: [PATCH 1/5] fix(agent): floor heartbeat attempt timeouts --- docs/api-reference/veryfront/agent.md | 2 +- src/agent/service/registration.test.ts | 96 +++++++++++++------------- src/agent/service/registration.ts | 18 ++--- 3 files changed, 57 insertions(+), 59 deletions(-) diff --git a/docs/api-reference/veryfront/agent.md b/docs/api-reference/veryfront/agent.md index c5a3bf48f4..9288d67a5e 100644 --- a/docs/api-reference/veryfront/agent.md +++ b/docs/api-reference/veryfront/agent.md @@ -930,7 +930,7 @@ Input delivered to a hosted agent-service detached execution callback. | `recoverConversationRunAppendFailure` | Recover conversation run append failure helper. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable.ts#L346) | | `recoverConversationRunCursorMismatch` | Recover conversation run cursor mismatch helper. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable.ts#L247) | | `registerAgent` | Registers agent. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/composition/composition.ts#L233) | -| `resolveAgentServiceRegistrationInput` | Input payload for resolve agent service registration. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/service/registration.ts#L311) | +| `resolveAgentServiceRegistrationInput` | Input payload for resolve agent service registration. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/service/registration.ts#L313) | | `resolveConversationHostedStreamErrorState` | State for resolve conversation hosted stream error. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/hosted-terminal.ts#L69) | | `resolveConversationHostedTerminalState` | State for resolve conversation hosted terminal. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/hosted-terminal.ts#L44) | | `resolveConversationRunTargets` | Resolves conversation run targets. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable-contracts.ts#L34) | diff --git a/src/agent/service/registration.test.ts b/src/agent/service/registration.test.ts index bdc7705c8a..cf13de7e38 100644 --- a/src/agent/service/registration.test.ts +++ b/src/agent/service/registration.test.ts @@ -2,6 +2,7 @@ import "#veryfront/schemas/_test-setup.ts"; import { assert, assertEquals, assertRejects } from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; import { waitFor } from "#veryfront/testing"; +import { FakeTime } from "#std/testing/time"; import { NETWORK_ERROR } from "#veryfront/errors"; import { type AgentServiceRegistrationLogger, @@ -461,10 +462,12 @@ describe("agent/agent-service-registration heartbeat retry", () => { }); it("times out a permanently hung heartbeat and escalates in bounded time", async () => { + using time = new FakeTime(); const intervalMs = 20; let heartbeatRequests = 0; let inFlight = 0; let maxConcurrent = 0; + let lastHeartbeatSignal: AbortSignal | null | undefined; const log = recordingLogger(); const fetch: typeof globalThis.fetch = (input, init) => { @@ -475,6 +478,7 @@ describe("agent/agent-service-registration heartbeat retry", () => { inFlight++; maxConcurrent = Math.max(maxConcurrent, inFlight); const signal = init?.signal; + lastHeartbeatSignal = signal; return new Promise((_resolve, reject) => { const abort = () => { inFlight--; @@ -489,16 +493,16 @@ describe("agent/agent-service-registration heartbeat retry", () => { lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), ); - const escalationBudgetMs = 1_500; - const startedAt = Date.now(); + const escalationBudgetMs = 100_000; + const startedAt = time.now; try { - await waitFor(() => log.errors.length > 0, { - timeout: escalationBudgetMs, - interval: 10, - message: "hung heartbeat attempts never reached persistent-failure escalation", - }); + for (let step = 0; step < 30 && log.errors.length === 0; step++) { + await time.tickAsync(5_001); + await time.runMicrotasks(); + } + assert(log.errors.length > 0, "hung heartbeat attempts never escalated"); assert( - Date.now() - startedAt < escalationBudgetMs, + time.now - startedAt < escalationBudgetMs, "hung heartbeat escalation exceeded its bounded test budget", ); assertEquals( @@ -511,26 +515,26 @@ describe("agent/agent-service-registration heartbeat retry", () => { "three failed ticks must each exhaust the three-attempt retry policy", ); assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard"); + await time.tickAsync(intervalMs); + await time.runMicrotasks(); } finally { + assertEquals(lastHeartbeatSignal?.aborted, false, "teardown needs an active hung request"); lifecycle.stop(); - await waitFor(() => inFlight === 0, { - timeout: 1_000, - interval: 10, - message: "stop() did not abort the hung heartbeat request", - }); + for (let turn = 0; turn < 5 && inFlight > 0; turn++) { + await time.runMicrotasks(); + } + assertEquals(lastHeartbeatSignal?.aborted, true, "stop() did not abort the hung request"); } }); - it("leaves healthy and intermittently slow heartbeats alone", async () => { - // The deadline is one interval, so a heartbeat that answers inside its own - // interval must never be cut off. This is the false-positive guard: an - // eager timeout would escalate a merely slow control plane. - // The slow answer sits at 40% of the deadline, far enough inside it that a - // loaded runner cannot push it over, and far enough outside a quarter of - // the interval that an over-eager deadline would still be caught. - const intervalMs = 500; - const slowLatencyMs = 200; + it("lets a known-valid slow heartbeat finish with a short configured interval", async () => { + // The control plane has been seen answering in about 2.7s while degraded. + // A short heartbeat interval must not turn that known-valid latency into a + // persistent-failure escalation. + const intervalMs = 100; + const slowLatencyMs = 3_000; let heartbeatRequests = 0; + let completedHeartbeats = 0; let inFlight = 0; const log = recordingLogger(); @@ -539,8 +543,6 @@ describe("agent/agent-service-registration heartbeat retry", () => { return Promise.resolve(jsonResponse(serviceResponse)); } heartbeatRequests++; - // Alternate a fast answer with one that eats most of the interval. - const latencyMs = heartbeatRequests % 2 === 0 ? slowLatencyMs : 5; const signal = init?.signal; inFlight++; return new Promise((resolve, reject) => { @@ -548,9 +550,10 @@ describe("agent/agent-service-registration heartbeat retry", () => { // early actually shows up here instead of being answered late anyway. const timer = setTimeout(() => { signal?.removeEventListener("abort", onAbort); + completedHeartbeats++; inFlight--; resolve(jsonResponse(serviceResponse)); - }, latencyMs); + }, slowLatencyMs); function onAbort() { clearTimeout(timer); inFlight--; @@ -565,29 +568,23 @@ describe("agent/agent-service-registration heartbeat retry", () => { lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), ); - await waitFor(() => heartbeatRequests >= 4, { - timeout: 10_000, - interval: 10, - message: "the heartbeat never issued enough requests to observe the deadline", - }); - lifecycle.stop(); - await waitFor(() => inFlight === 0, { - timeout: 10_000, - interval: 10, - message: "in-flight heartbeat requests never settled after stop()", - }); - - assertEquals( - log.errors.length, - 0, - "a heartbeat answering inside its interval must never escalate", - ); - assertEquals( - log.warnings.length, - 0, - `a heartbeat answering inside its interval must not retry or skip, saw ` + - `${log.warnings.map((entry) => entry.message).join(", ")}`, - ); + try { + await waitFor(() => completedHeartbeats > 0 || log.errors.length > 0, { + timeout: 10_000, + interval: 10, + message: "the slow heartbeat neither completed nor escalated", + }); + assertEquals(completedHeartbeats, 1, "a known-valid slow heartbeat must complete"); + assertEquals(heartbeatRequests, 1, "a slow success must not be retried"); + assertEquals(log.errors.length, 0, "a slow success must not escalate"); + } finally { + lifecycle.stop(); + await waitFor(() => inFlight === 0, { + timeout: 1_000, + interval: 10, + message: "stop() did not abort the slow heartbeat request", + }); + } }); it("lets a slow-but-successful heartbeat finish at the production interval", async () => { @@ -632,7 +629,8 @@ describe("agent/agent-service-registration heartbeat retry", () => { // Driven directly rather than through the interval: this is about the // deadline on one attempt, not about scheduling. Nothing here asserts on // escalation, because escalation is counted in the interval tick and a 30s - // interval never fires inside this test. The 500ms test above covers that. + // interval never fires inside this test. The short-interval test above + // covers that. await lifecycle.heartbeat(); lifecycle.stop(); diff --git a/src/agent/service/registration.ts b/src/agent/service/registration.ts index d2ea2f5dd0..4f67ec41b9 100644 --- a/src/agent/service/registration.ts +++ b/src/agent/service/registration.ts @@ -236,12 +236,14 @@ function getErrorMessage(error: unknown): string { const HEARTBEAT_MAX_ATTEMPTS = 3; /** Backoff before the first retry when the heartbeat interval leaves room for it. */ const HEARTBEAT_RETRY_BASE_DELAY_MS = 250; +/** Minimum time allowed for a control-plane heartbeat attempt. */ +const HEARTBEAT_MIN_ATTEMPT_TIMEOUT_MS = 5_000; /** * Share of one heartbeat interval the retry backoff may occupy. * * This bounds only the waits between attempts. Each request has its own - * interval-sized deadline, so a complete retry sequence can still outlive one - * interval. The in-flight guard below prevents overlap in that case. + * interval-derived deadline, so a complete retry sequence can still outlive + * one interval. The in-flight guard below prevents overlap in that case. */ const HEARTBEAT_RETRY_BUDGET_RATIO = 0.25; @@ -496,13 +498,11 @@ async function heartbeatAgentPushRuntimeService( return await retryWithBackoff((signal) => sendHeartbeatRequest(input, fetchImpl, signal), { maxAttempts: schedule.maxAttempts, abortSignal: options.abortSignal, - // One interval is the deadline for a single attempt. A heartbeat that has - // not answered by the time the next one is due is not slow, it is gone, and - // without a deadline a hung request never fails, so the failure counter - // never advances and the service looks alive to itself while doing nothing. - // Deriving the deadline from the configured interval keeps the two in step: - // raise the interval for a slow link and the deadline moves with it. - timeoutMs: input.heartbeatIntervalMs, + // Derive the deadline from the configured interval, but keep enough room + // for a known-valid slow control-plane response. Raising the interval for a + // slower link still raises the deadline, while the floor prevents a short + // interval from falsely escalating a healthy service. + timeoutMs: Math.max(input.heartbeatIntervalMs, HEARTBEAT_MIN_ATTEMPT_TIMEOUT_MS), computeDelay: (attempt) => schedule.delaysMs[attempt] ?? 0, shouldRetry: isRetryableHeartbeatFailure, onRetry: ({ error, attempt, delay }) => { From 0c79f37c06f25ed4557cd70742c9e20c60675b6d Mon Sep 17 00:00:00 2001 From: Kentaro Wakayama Date: Sun, 23 Aug 2026 22:18:47 +0200 Subject: [PATCH 2/5] test(agent): make heartbeat timing checks deterministic --- src/agent/service/registration.test.ts | 49 ++++++++++++-------------- 1 file changed, 23 insertions(+), 26 deletions(-) diff --git a/src/agent/service/registration.test.ts b/src/agent/service/registration.test.ts index cf13de7e38..aa9162cd07 100644 --- a/src/agent/service/registration.test.ts +++ b/src/agent/service/registration.test.ts @@ -2,7 +2,6 @@ import "#veryfront/schemas/_test-setup.ts"; import { assert, assertEquals, assertRejects } from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; import { waitFor } from "#veryfront/testing"; -import { FakeTime } from "#std/testing/time"; import { NETWORK_ERROR } from "#veryfront/errors"; import { type AgentServiceRegistrationLogger, @@ -462,12 +461,10 @@ describe("agent/agent-service-registration heartbeat retry", () => { }); it("times out a permanently hung heartbeat and escalates in bounded time", async () => { - using time = new FakeTime(); const intervalMs = 20; let heartbeatRequests = 0; - let inFlight = 0; + const activeSignals = new Set(); let maxConcurrent = 0; - let lastHeartbeatSignal: AbortSignal | null | undefined; const log = recordingLogger(); const fetch: typeof globalThis.fetch = (input, init) => { @@ -475,13 +472,13 @@ describe("agent/agent-service-registration heartbeat retry", () => { return Promise.resolve(jsonResponse(serviceResponse)); } heartbeatRequests++; - inFlight++; - maxConcurrent = Math.max(maxConcurrent, inFlight); const signal = init?.signal; - lastHeartbeatSignal = signal; + assert(signal, "heartbeat requests must carry an abort signal"); + activeSignals.add(signal); + maxConcurrent = Math.max(maxConcurrent, activeSignals.size); return new Promise((_resolve, reject) => { const abort = () => { - inFlight--; + activeSignals.delete(signal); reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); }; if (signal?.aborted) abort(); @@ -493,16 +490,16 @@ describe("agent/agent-service-registration heartbeat retry", () => { lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), ); - const escalationBudgetMs = 100_000; - const startedAt = time.now; + const escalationBudgetMs = 50_000; + const startedAt = Date.now(); try { - for (let step = 0; step < 30 && log.errors.length === 0; step++) { - await time.tickAsync(5_001); - await time.runMicrotasks(); - } - assert(log.errors.length > 0, "hung heartbeat attempts never escalated"); + await waitFor(() => log.errors.length > 0, { + timeout: escalationBudgetMs, + interval: 10, + message: "hung heartbeat attempts never reached persistent-failure escalation", + }); assert( - time.now - startedAt < escalationBudgetMs, + Date.now() - startedAt < escalationBudgetMs, "hung heartbeat escalation exceeded its bounded test budget", ); assertEquals( @@ -515,15 +512,13 @@ describe("agent/agent-service-registration heartbeat retry", () => { "three failed ticks must each exhaust the three-attempt retry policy", ); assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard"); - await time.tickAsync(intervalMs); - await time.runMicrotasks(); } finally { - assertEquals(lastHeartbeatSignal?.aborted, false, "teardown needs an active hung request"); lifecycle.stop(); - for (let turn = 0; turn < 5 && inFlight > 0; turn++) { - await time.runMicrotasks(); - } - assertEquals(lastHeartbeatSignal?.aborted, true, "stop() did not abort the hung request"); + await waitFor(() => activeSignals.size === 0, { + timeout: 1_000, + interval: 10, + message: "stop() did not abort the hung heartbeat request", + }); } }); @@ -533,7 +528,6 @@ describe("agent/agent-service-registration heartbeat retry", () => { // persistent-failure escalation. const intervalMs = 100; const slowLatencyMs = 3_000; - let heartbeatRequests = 0; let completedHeartbeats = 0; let inFlight = 0; const log = recordingLogger(); @@ -542,7 +536,6 @@ describe("agent/agent-service-registration heartbeat retry", () => { if (!input.toString().endsWith("/heartbeat")) { return Promise.resolve(jsonResponse(serviceResponse)); } - heartbeatRequests++; const signal = init?.signal; inFlight++; return new Promise((resolve, reject) => { @@ -575,7 +568,11 @@ describe("agent/agent-service-registration heartbeat retry", () => { message: "the slow heartbeat neither completed nor escalated", }); assertEquals(completedHeartbeats, 1, "a known-valid slow heartbeat must complete"); - assertEquals(heartbeatRequests, 1, "a slow success must not be retried"); + assertEquals( + log.warnings.filter((entry) => entry.message.includes("retrying")).length, + 0, + "a slow success must not be retried", + ); assertEquals(log.errors.length, 0, "a slow success must not escalate"); } finally { lifecycle.stop(); From d258d506fede01c84b61bc40488059447d755a5a Mon Sep 17 00:00:00 2001 From: Kentaro Wakayama Date: Sun, 23 Aug 2026 22:34:07 +0200 Subject: [PATCH 3/5] test(agent): allow bounded heartbeat escalation --- src/agent/service/registration.test.ts | 120 +++++++++++++------------ 1 file changed, 62 insertions(+), 58 deletions(-) diff --git a/src/agent/service/registration.test.ts b/src/agent/service/registration.test.ts index aa9162cd07..601bfc6dfd 100644 --- a/src/agent/service/registration.test.ts +++ b/src/agent/service/registration.test.ts @@ -460,67 +460,71 @@ describe("agent/agent-service-registration heartbeat retry", () => { ); }); - it("times out a permanently hung heartbeat and escalates in bounded time", async () => { - const intervalMs = 20; - let heartbeatRequests = 0; - const activeSignals = new Set(); - let maxConcurrent = 0; - const log = recordingLogger(); - - const fetch: typeof globalThis.fetch = (input, init) => { - if (!input.toString().endsWith("/heartbeat")) { - return Promise.resolve(jsonResponse(serviceResponse)); - } - heartbeatRequests++; - const signal = init?.signal; - assert(signal, "heartbeat requests must carry an abort signal"); - activeSignals.add(signal); - maxConcurrent = Math.max(maxConcurrent, activeSignals.size); - return new Promise((_resolve, reject) => { - const abort = () => { - activeSignals.delete(signal); - reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); - }; - if (signal?.aborted) abort(); - else signal?.addEventListener("abort", abort, { once: true }); - }); - }; + it( + "times out a permanently hung heartbeat and escalates in bounded time", + { timeout: 60_000 }, + async () => { + const intervalMs = 20; + let heartbeatRequests = 0; + const activeSignals = new Set(); + let maxConcurrent = 0; + const log = recordingLogger(); - const lifecycle = await createAgentServiceRegistrationLifecycle( - lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), - ); + const fetch: typeof globalThis.fetch = (input, init) => { + if (!input.toString().endsWith("/heartbeat")) { + return Promise.resolve(jsonResponse(serviceResponse)); + } + heartbeatRequests++; + const signal = init?.signal; + assert(signal, "heartbeat requests must carry an abort signal"); + activeSignals.add(signal); + maxConcurrent = Math.max(maxConcurrent, activeSignals.size); + return new Promise((_resolve, reject) => { + const abort = () => { + activeSignals.delete(signal); + reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); + }; + if (signal?.aborted) abort(); + else signal?.addEventListener("abort", abort, { once: true }); + }); + }; - const escalationBudgetMs = 50_000; - const startedAt = Date.now(); - try { - await waitFor(() => log.errors.length > 0, { - timeout: escalationBudgetMs, - interval: 10, - message: "hung heartbeat attempts never reached persistent-failure escalation", - }); - assert( - Date.now() - startedAt < escalationBudgetMs, - "hung heartbeat escalation exceeded its bounded test budget", - ); - assertEquals( - log.errors[0]?.metadata?.consecutiveFailures, - 3, - "a timeout must count as one failed tick after its retries are exhausted", - ); - assert( - heartbeatRequests >= 9, - "three failed ticks must each exhaust the three-attempt retry policy", + const lifecycle = await createAgentServiceRegistrationLifecycle( + lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), ); - assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard"); - } finally { - lifecycle.stop(); - await waitFor(() => activeSignals.size === 0, { - timeout: 1_000, - interval: 10, - message: "stop() did not abort the hung heartbeat request", - }); - } - }); + + const escalationBudgetMs = 50_000; + const startedAt = Date.now(); + try { + await waitFor(() => log.errors.length > 0, { + timeout: escalationBudgetMs, + interval: 10, + message: "hung heartbeat attempts never reached persistent-failure escalation", + }); + assert( + Date.now() - startedAt < escalationBudgetMs, + "hung heartbeat escalation exceeded its bounded test budget", + ); + assertEquals( + log.errors[0]?.metadata?.consecutiveFailures, + 3, + "a timeout must count as one failed tick after its retries are exhausted", + ); + assert( + heartbeatRequests >= 9, + "three failed ticks must each exhaust the three-attempt retry policy", + ); + assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard"); + } finally { + lifecycle.stop(); + await waitFor(() => activeSignals.size === 0, { + timeout: 1_000, + interval: 10, + message: "stop() did not abort the hung heartbeat request", + }); + } + }, + ); it("lets a known-valid slow heartbeat finish with a short configured interval", async () => { // The control plane has been seen answering in about 2.7s while degraded. From 90e0ed9e754077e57736b150986a82be73166489 Mon Sep 17 00:00:00 2001 From: Kentaro Wakayama Date: Mon, 24 Aug 2026 07:36:08 +0200 Subject: [PATCH 4/5] test(agent): harden heartbeat timeout regressions --- src/agent/service/registration.test.ts | 235 +++++++++++++++---------- src/agent/service/registration.ts | 12 +- 2 files changed, 155 insertions(+), 92 deletions(-) diff --git a/src/agent/service/registration.test.ts b/src/agent/service/registration.test.ts index 601bfc6dfd..fa6b937f45 100644 --- a/src/agent/service/registration.test.ts +++ b/src/agent/service/registration.test.ts @@ -2,6 +2,7 @@ import "#veryfront/schemas/_test-setup.ts"; import { assert, assertEquals, assertRejects } from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; import { waitFor } from "#veryfront/testing"; +import { FakeTime } from "#std/testing/time"; import { NETWORK_ERROR } from "#veryfront/errors"; import { type AgentServiceRegistrationLogger, @@ -290,6 +291,25 @@ function lifecycleOptions( }; } +function abortAwareDelayedJsonResponse( + body: unknown, + delayMs: number, + signal?: AbortSignal | null, +): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(jsonResponse(body)); + }, delayMs); + function onAbort() { + clearTimeout(timer); + reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); + } + if (signal?.aborted) onAbort(); + else signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + describe("agent/agent-service-registration heartbeat retry", () => { it("retries a transient 500 so a one-second blip never counts as a failure", async () => { const script = scriptedHeartbeatFetch([500, 200]); @@ -460,71 +480,120 @@ describe("agent/agent-service-registration heartbeat retry", () => { ); }); - it( - "times out a permanently hung heartbeat and escalates in bounded time", - { timeout: 60_000 }, - async () => { - const intervalMs = 20; - let heartbeatRequests = 0; - const activeSignals = new Set(); - let maxConcurrent = 0; - const log = recordingLogger(); + it("times out a permanently hung heartbeat and escalates with fake time", async () => { + using time = new FakeTime(); + const intervalMs = 20; + let heartbeatRequests = 0; + const activeSignals = new Set(); + let maxConcurrent = 0; + const log = recordingLogger(); + + const fetch: typeof globalThis.fetch = (input, init) => { + if (!input.toString().endsWith("/heartbeat")) { + return Promise.resolve(jsonResponse(serviceResponse)); + } + heartbeatRequests++; + const signal = init?.signal; + assert(signal, "heartbeat requests must carry an abort signal"); + activeSignals.add(signal); + maxConcurrent = Math.max(maxConcurrent, activeSignals.size); + return new Promise((_resolve, reject) => { + const abort = () => { + activeSignals.delete(signal); + reject(signal.reason ?? new DOMException("Aborted", "AbortError")); + }; + if (signal.aborted) abort(); + else signal.addEventListener("abort", abort, { once: true }); + }); + }; - const fetch: typeof globalThis.fetch = (input, init) => { - if (!input.toString().endsWith("/heartbeat")) { - return Promise.resolve(jsonResponse(serviceResponse)); - } - heartbeatRequests++; - const signal = init?.signal; - assert(signal, "heartbeat requests must carry an abort signal"); - activeSignals.add(signal); - maxConcurrent = Math.max(maxConcurrent, activeSignals.size); - return new Promise((_resolve, reject) => { - const abort = () => { - activeSignals.delete(signal); - reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); - }; - if (signal?.aborted) abort(); - else signal?.addEventListener("abort", abort, { once: true }); - }); - }; + const lifecycle = await createAgentServiceRegistrationLifecycle( + lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), + ); - const lifecycle = await createAgentServiceRegistrationLifecycle( - lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }), - ); + await time.tickAsync(intervalMs); + for (let step = 0; step < 20 && log.errors.length === 0; step++) { + await time.tickAsync(5_010); + await time.tickAsync(0); + } + assertEquals(heartbeatRequests, 9, "three failed ticks must exhaust three attempts each"); + assertEquals( + log.errors[0]?.metadata?.consecutiveFailures, + 3, + "a timeout must count as one failed tick after its retries are exhausted", + ); + assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard"); + + await time.tickAsync(intervalMs); + assertEquals(activeSignals.size, 1, "the next heartbeat must be in flight before teardown"); + lifecycle.stop(); + await time.tickAsync(0); + assertEquals(activeSignals.size, 0, "stop() must abort the hung heartbeat request"); + }); - const escalationBudgetMs = 50_000; - const startedAt = Date.now(); - try { - await waitFor(() => log.errors.length > 0, { - timeout: escalationBudgetMs, - interval: 10, - message: "hung heartbeat attempts never reached persistent-failure escalation", - }); - assert( - Date.now() - startedAt < escalationBudgetMs, - "hung heartbeat escalation exceeded its bounded test budget", - ); - assertEquals( - log.errors[0]?.metadata?.consecutiveFailures, - 3, - "a timeout must count as one failed tick after its retries are exhausted", - ); - assert( - heartbeatRequests >= 9, - "three failed ticks must each exhaust the three-attempt retry policy", - ); - assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard"); - } finally { - lifecycle.stop(); - await waitFor(() => activeSignals.size === 0, { - timeout: 1_000, - interval: 10, - message: "stop() did not abort the hung heartbeat request", - }); + it("floors a short heartbeat attempt timeout at five seconds", async () => { + using time = new FakeTime(); + let aborts = 0; + const fetch: typeof globalThis.fetch = (input, init) => { + if (!input.toString().endsWith("/heartbeat")) { + return Promise.resolve(jsonResponse(serviceResponse)); } - }, - ); + const signal = init?.signal; + assert(signal, "heartbeat requests must carry an abort signal"); + return new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => { + aborts++; + reject(signal.reason); + }, { once: true }); + }); + }; + const lifecycle = await createAgentServiceRegistrationLifecycle( + lifecycleOptions(fetch, { heartbeatIntervalMs: 100 }), + ); + const pending = lifecycle.heartbeat().catch(() => undefined); + await time.tickAsync(0); + + await time.tickAsync(4_999); + assertEquals(aborts, 0, "a short interval must not abort before the five-second floor"); + await time.tickAsync(1); + assertEquals(aborts, 1, "the first attempt must abort at the five-second floor"); + + lifecycle.stop(); + await time.tickAsync(0); + await pending; + }); + + it("keeps a configured heartbeat attempt timeout above five seconds", async () => { + using time = new FakeTime(); + let aborts = 0; + const fetch: typeof globalThis.fetch = (input, init) => { + if (!input.toString().endsWith("/heartbeat")) { + return Promise.resolve(jsonResponse(serviceResponse)); + } + const signal = init?.signal; + assert(signal, "heartbeat requests must carry an abort signal"); + return new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => { + aborts++; + reject(signal.reason); + }, { once: true }); + }); + }; + const lifecycle = await createAgentServiceRegistrationLifecycle( + lifecycleOptions(fetch, { heartbeatIntervalMs: 30_000 }), + ); + const pending = lifecycle.heartbeat().catch(() => undefined); + await time.tickAsync(0); + + await time.tickAsync(29_999); + assertEquals(aborts, 0, "a higher configured interval must remain the attempt timeout"); + await time.tickAsync(1); + assertEquals(aborts, 1, "the first attempt must abort at the configured higher interval"); + + lifecycle.stop(); + await time.tickAsync(0); + await pending; + }); it("lets a known-valid slow heartbeat finish with a short configured interval", async () => { // The control plane has been seen answering in about 2.7s while degraded. @@ -542,23 +611,17 @@ describe("agent/agent-service-registration heartbeat retry", () => { } const signal = init?.signal; inFlight++; - return new Promise((resolve, reject) => { - // Honour the abort the way a real fetch does, so a deadline that fires - // early actually shows up here instead of being answered late anyway. - const timer = setTimeout(() => { - signal?.removeEventListener("abort", onAbort); + return abortAwareDelayedJsonResponse(serviceResponse, slowLatencyMs, signal).then( + (response) => { completedHeartbeats++; inFlight--; - resolve(jsonResponse(serviceResponse)); - }, slowLatencyMs); - function onAbort() { - clearTimeout(timer); + return response; + }, + (error) => { inFlight--; - reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); - } - if (signal?.aborted) onAbort(); - else signal?.addEventListener("abort", onAbort, { once: true }); - }); + throw error; + }, + ); }; const lifecycle = await createAgentServiceRegistrationLifecycle( @@ -572,6 +635,13 @@ describe("agent/agent-service-registration heartbeat retry", () => { message: "the slow heartbeat neither completed nor escalated", }); assertEquals(completedHeartbeats, 1, "a known-valid slow heartbeat must complete"); + assertEquals( + log.warnings.filter((entry) => + entry.message === "Agent service heartbeat tick skipped, previous tick still running" + ).length, + 1, + "one slow successful heartbeat must log at most one skip warning", + ); assertEquals( log.warnings.filter((entry) => entry.message.includes("retrying")).length, 0, @@ -607,20 +677,7 @@ describe("agent/agent-service-registration heartbeat retry", () => { } heartbeatRequests++; const signal = init?.signal; - return new Promise((resolve, reject) => { - // Raw timer on purpose: this is the double's simulated latency, on the - // same unscaled clock as the deadline it is being measured against. - const timer = setTimeout(() => { - signal?.removeEventListener("abort", onAbort); - resolve(jsonResponse(serviceResponse)); - }, slowLatencyMs); - function onAbort() { - clearTimeout(timer); - reject(signal?.reason ?? new DOMException("Aborted", "AbortError")); - } - if (signal?.aborted) onAbort(); - else signal?.addEventListener("abort", onAbort, { once: true }); - }); + return abortAwareDelayedJsonResponse(serviceResponse, slowLatencyMs, signal); }; const lifecycle = await createAgentServiceRegistrationLifecycle( diff --git a/src/agent/service/registration.ts b/src/agent/service/registration.ts index 4f67ec41b9..23c3e6841f 100644 --- a/src/agent/service/registration.ts +++ b/src/agent/service/registration.ts @@ -553,6 +553,7 @@ export async function createAgentServiceRegistrationLifecycle( let consecutiveHeartbeatFailures = 0; let heartbeatInFlight = false; + let heartbeatSkipLogged = false; const interval = setInterval(() => { // Retries make a tick outlive its interval whenever the control plane is @@ -560,12 +561,17 @@ export async function createAgentServiceRegistrationLifecycle( // that is already struggling, and would let the failure counter advance // out of order, so the beat is skipped instead. if (heartbeatInFlight) { - options.logger?.warn?.("Agent service heartbeat tick skipped, previous tick still running", { - serviceId: service.id, - }); + if (!heartbeatSkipLogged) { + heartbeatSkipLogged = true; + options.logger?.warn?.( + "Agent service heartbeat tick skipped, previous tick still running", + { serviceId: service.id }, + ); + } return; } heartbeatInFlight = true; + heartbeatSkipLogged = false; void heartbeat().then(() => { consecutiveHeartbeatFailures = 0; }).catch((error: unknown) => { From 5ad37624d9cc3c9bd13c5be14dc1ab03a71ca665 Mon Sep 17 00:00:00 2001 From: Kentaro Wakayama Date: Mon, 24 Aug 2026 07:54:23 +0200 Subject: [PATCH 5/5] fix(agent): share in-flight lifecycle heartbeat --- src/agent/service/registration.test.ts | 59 +++++++++++++++++++++++++- src/agent/service/registration.ts | 56 ++++++++++++------------ 2 files changed, 88 insertions(+), 27 deletions(-) diff --git a/src/agent/service/registration.test.ts b/src/agent/service/registration.test.ts index fa6b937f45..b65a824891 100644 --- a/src/agent/service/registration.test.ts +++ b/src/agent/service/registration.test.ts @@ -1,5 +1,10 @@ import "#veryfront/schemas/_test-setup.ts"; -import { assert, assertEquals, assertRejects } from "#veryfront/testing/assert.ts"; +import { + assert, + assertEquals, + assertRejects, + assertStrictEquals, +} from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; import { waitFor } from "#veryfront/testing"; import { FakeTime } from "#std/testing/time"; @@ -428,6 +433,58 @@ describe("agent/agent-service-registration heartbeat retry", () => { ); }); + it("shares a scheduled in-flight heartbeat with direct callers", async () => { + using time = new FakeTime(); + const intervalMs = 100; + let heartbeatRequests = 0; + const activeSignals = new Set(); + + const fetch: typeof globalThis.fetch = (input, init) => { + if (!input.toString().endsWith("/heartbeat")) { + return Promise.resolve(jsonResponse(serviceResponse)); + } + heartbeatRequests++; + const signal = init?.signal; + assert(signal, "heartbeat requests must carry an abort signal"); + activeSignals.add(signal); + return new Promise((_resolve, reject) => { + const abort = () => { + activeSignals.delete(signal); + reject(signal.reason ?? new DOMException("Aborted", "AbortError")); + }; + if (signal.aborted) abort(); + else signal.addEventListener("abort", abort, { once: true }); + }); + }; + + const lifecycle = await createAgentServiceRegistrationLifecycle( + lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs }), + ); + + await time.tickAsync(intervalMs); + assertEquals(heartbeatRequests, 1, "the scheduled heartbeat must be in flight"); + + const firstDirectCaller = lifecycle.heartbeat(); + const secondDirectCaller = lifecycle.heartbeat(); + assertStrictEquals( + firstDirectCaller, + secondDirectCaller, + "overlapping direct callers must receive the shared in-flight promise", + ); + await time.tickAsync(0); + assertEquals( + heartbeatRequests, + 1, + "direct callers must await the scheduled heartbeat instead of starting another request", + ); + assertEquals(activeSignals.size, 1, "only one heartbeat request may be active"); + + lifecycle.stop(); + await time.tickAsync(0); + await Promise.all([firstDirectCaller, secondDirectCaller]); + assertEquals(activeSignals.size, 0, "stop() must abort the shared heartbeat request"); + }); + it("still escalates when slow failures force ticks to be skipped", async () => { // Each tick needs 3 attempts x 80ms, so it outlives the 100ms interval and // the beats in between are skipped. A skipped beat is neither a success nor diff --git a/src/agent/service/registration.ts b/src/agent/service/registration.ts index 23c3e6841f..bc019a57f0 100644 --- a/src/agent/service/registration.ts +++ b/src/agent/service/registration.ts @@ -525,35 +525,43 @@ export async function createAgentServiceRegistrationLifecycle( const service = await registerAgentPushRuntimeService(input, fetchImpl); let stopped = false; const teardown = new AbortController(); + let heartbeatInFlight: Promise | undefined; + let heartbeatSkipLogged = false; - const heartbeat = async () => { - if (stopped) { - return; - } - try { - await heartbeatAgentPushRuntimeService( - { - apiUrl: input.apiUrl, - authToken: input.authToken, - serviceId: service.id, - heartbeatIntervalMs: input.heartbeatIntervalMs, - }, - fetchImpl, - { logger: options.logger, abortSignal: teardown.signal }, - ); - } catch (error) { - // stop() aborts the in-flight request and any pending backoff. That is a - // teardown, not a heartbeat failure, so it must not reach the counter. + const heartbeat = () => { + if (heartbeatInFlight) return heartbeatInFlight; + + heartbeatSkipLogged = false; + heartbeatInFlight = (async () => { if (stopped) { return; } - throw error; - } + try { + await heartbeatAgentPushRuntimeService( + { + apiUrl: input.apiUrl, + authToken: input.authToken, + serviceId: service.id, + heartbeatIntervalMs: input.heartbeatIntervalMs, + }, + fetchImpl, + { logger: options.logger, abortSignal: teardown.signal }, + ); + } catch (error) { + // stop() aborts the in-flight request and any pending backoff. That is a + // teardown, not a heartbeat failure, so it must not reach the counter. + if (stopped) { + return; + } + throw error; + } + })().finally(() => { + heartbeatInFlight = undefined; + }); + return heartbeatInFlight; }; let consecutiveHeartbeatFailures = 0; - let heartbeatInFlight = false; - let heartbeatSkipLogged = false; const interval = setInterval(() => { // Retries make a tick outlive its interval whenever the control plane is @@ -570,8 +578,6 @@ export async function createAgentServiceRegistrationLifecycle( } return; } - heartbeatInFlight = true; - heartbeatSkipLogged = false; void heartbeat().then(() => { consecutiveHeartbeatFailures = 0; }).catch((error: unknown) => { @@ -590,8 +596,6 @@ export async function createAgentServiceRegistrationLifecycle( error: getErrorMessage(error), }); } - }).finally(() => { - heartbeatInFlight = false; }); }, input.heartbeatIntervalMs);