From 78e4672d651159700987d5be224fbb16eb0819e5 Mon Sep 17 00:00:00 2001 From: Wynne668 Date: Tue, 7 Jul 2026 12:58:12 +0800 Subject: [PATCH] fix: pace delivery recovery after startup outages (#101118) * fix: pace delivery recovery replays * fix: preserve paced delivery recovery budget * fix(delivery): preserve recovery deadline while pacing * fix(delivery): coordinate recovery replay pacing --------- Co-authored-by: Peter Steinberger --- CHANGELOG.md | 1 + src/infra/delivery-recovery.shared.ts | 44 +++++++ src/infra/outbound/delivery-queue-recovery.ts | 10 ++ .../delivery-queue.reconnect-drain.test.ts | 57 +++++++++ .../outbound/delivery-queue.recovery.test.ts | 76 ++++++++++++ src/infra/session-delivery-queue-recovery.ts | 10 ++ .../session-delivery-queue.recovery.test.ts | 117 ++++++++++++++++++ 7 files changed, 315 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ccf478c01a..92865559aef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,7 @@ Docs: https://docs.openclaw.ai - **Android hardware keyboard chat:** send with unmodified Enter on physical keyboards while preserving Shift+Enter and other modified Enter combinations for multiline input. (#101239) Thanks @3ninyt3nin-creator. - **CJK Markdown emphasis:** render adjacent Chinese, Japanese, and Korean emphasis punctuation through the shared Markdown pipeline instead of leaking literal markers across channels. (#101230, #101120) Thanks @nicknmorty. - **Codex yielded native subagents:** keep the parent app-server subscription and shared client alive until yielded native subagent completion delivery settles, preventing lost wakeups and leaked one-shot cleanup. +- **Delivery recovery pacing:** pace eligible outbound and restart-continuation replays after gateway startup so outage backlogs do not burst into channel rate limits, while preserving the wall-clock recovery budget. (#101118, #101058) Thanks @ZengWen-DT. - **Discord streamed finals:** send completion replies as fresh messages so inactive channels become unread, while preserving targeted mentions without escalating `@everyone` or `@here`. (#99711, #99662) Thanks @davelutztx. - **OpenAI-compatible SSE parsing:** recognize event streams mislabeled as JSON without prepending a second `data:` prefix, preserving valid streamed responses from non-conforming providers. (#96503) Thanks @ZengWen-DT. - **LM Studio embedding preload:** honor model- and provider-level context-window limits when preloading embedding models, preventing avoidable GPU out-of-memory failures. (#100750) Thanks @zak-li, @ZOOWH, and @hxz398. diff --git a/src/infra/delivery-recovery.shared.ts b/src/infra/delivery-recovery.shared.ts index bf090e69cf6..8c4e4c3927b 100644 --- a/src/infra/delivery-recovery.shared.ts +++ b/src/infra/delivery-recovery.shared.ts @@ -1,4 +1,7 @@ +import { sleep } from "../utils/sleep.js"; + const RECOVERY_BACKOFF_MS: readonly number[] = [5_000, 25_000, 120_000, 600_000]; +export const RECOVERY_REPLAY_SPACING_MS = 250; export function computeBackoffMs(retryCount: number): number { if (retryCount <= 0) { @@ -28,3 +31,44 @@ export function claimRecoveryEntry(entriesInProgress: Set, entryId: stri export function releaseRecoveryEntry(entriesInProgress: Set, entryId: string): void { entriesInProgress.delete(entryId); } + +export function createRecoveryReplayPacer(): { + wait(deadlineMs?: number): Promise<"ready" | "deadline-exceeded">; +} { + let lastReplayStartedAt = 0; + let waitQueue = Promise.resolve(); + + return { + async wait(deadlineMs) { + let releaseWaiter: () => void = () => {}; + const previousWaiter = waitQueue; + waitQueue = new Promise((resolve) => { + releaseWaiter = resolve; + }); + await previousWaiter; + + try { + const now = Date.now(); + if (deadlineMs !== undefined && now >= deadlineMs) { + return "deadline-exceeded"; + } + // Clock rollback starts a fresh pacing epoch. Otherwise concurrent startup + // and reconnect drains serialize here so neither can bypass the spacing floor. + const elapsedMs = now - lastReplayStartedAt; + const waitMs = elapsedMs < 0 ? 0 : Math.max(0, RECOVERY_REPLAY_SPACING_MS - elapsedMs); + if (waitMs > 0) { + const remainingBudgetMs = + deadlineMs === undefined ? waitMs : Math.max(0, deadlineMs - now); + await sleep(Math.min(waitMs, remainingBudgetMs)); + } + if (deadlineMs !== undefined && Date.now() >= deadlineMs) { + return "deadline-exceeded"; + } + lastReplayStartedAt = Date.now(); + return "ready"; + } finally { + releaseWaiter(); + } + }, + }; +} diff --git a/src/infra/outbound/delivery-queue-recovery.ts b/src/infra/outbound/delivery-queue-recovery.ts index f89574e3594..d574fd40a0e 100644 --- a/src/infra/outbound/delivery-queue-recovery.ts +++ b/src/infra/outbound/delivery-queue-recovery.ts @@ -12,6 +12,7 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { claimRecoveryEntry as claimSharedRecoveryEntry, computeBackoffMs, + createRecoveryReplayPacer, getErrnoCode, releaseRecoveryEntry as releaseSharedRecoveryEntry, } from "../delivery-recovery.shared.js"; @@ -93,6 +94,7 @@ const PERMANENT_ERROR_PATTERNS: readonly RegExp[] = [ const drainInProgress = new Map(); const entriesInProgress = new Set(); +const recoveryReplayPacer = createRecoveryReplayPacer(); function resolveRecoveryDeadlineMs(maxRecoveryMs: number | undefined): number { const durationMs = @@ -686,6 +688,8 @@ export async function drainPendingDeliveries(opts: { } } + await recoveryReplayPacer.wait(); + const result = await drainQueuedEntry({ entry: currentEntry, cfg: opts.cfg, @@ -778,6 +782,12 @@ export async function recoverPendingDeliveries(opts: { continue; } + const paceResult = await recoveryReplayPacer.wait(deadline); + if (paceResult === "deadline-exceeded") { + opts.log.warn(`Recovery time budget exceeded — remaining entries deferred to next startup`); + break; + } + const result = await drainQueuedEntry({ entry: currentEntry, cfg: opts.cfg, diff --git a/src/infra/outbound/delivery-queue.reconnect-drain.test.ts b/src/infra/outbound/delivery-queue.reconnect-drain.test.ts index 14b278ed756..d696295ffe9 100644 --- a/src/infra/outbound/delivery-queue.reconnect-drain.test.ts +++ b/src/infra/outbound/delivery-queue.reconnect-drain.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../../config/config.js"; import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js"; +import { RECOVERY_REPLAY_SPACING_MS } from "../delivery-recovery.shared.js"; import { type DeliverFn, drainPendingDeliveries, @@ -302,6 +303,62 @@ describe("drainPendingDeliveries for reconnect", () => { await startupRecovery; }); + it("shares replay pacing between reconnect and startup drains", async () => { + vi.useFakeTimers(); + const startedAt = new Date("2026-04-23T00:00:00.000Z"); + vi.setSystemTime(startedAt); + try { + const log = createRecoveryLog(); + const startupLog = createRecoveryLog(); + let firstStarted!: () => void; + const firstStartedPromise = new Promise((resolve) => { + firstStarted = resolve; + }); + let releaseFirst!: () => void; + const firstBlocked = new Promise((resolve) => { + releaseFirst = resolve; + }); + const deliveryTimes: number[] = []; + const deliver = vi.fn(async () => { + deliveryTimes.push(Date.now()); + if (deliveryTimes.length === 1) { + firstStarted(); + await firstBlocked; + } + }); + + for (const to of ["+1000", "+2000"]) { + await enqueueDelivery( + { channel: "directchat", to, payloads: [{ text: "hi" }], accountId: "acct1" }, + tmpDir, + ); + } + + const reconnectDrain = drainAcct1DirectChatReconnect({ deliver, log, stateDir: tmpDir }); + await firstStartedPromise; + const startupRecovery = recoverPendingDeliveries({ + cfg: stubCfg, + deliver, + log: startupLog, + stateDir: tmpDir, + }); + releaseFirst(); + + await vi.advanceTimersByTimeAsync(RECOVERY_REPLAY_SPACING_MS - 1); + expect(deliver).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + await Promise.all([reconnectDrain, startupRecovery]); + + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliveryTimes).toEqual([ + startedAt.getTime(), + startedAt.getTime() + RECOVERY_REPLAY_SPACING_MS, + ]); + } finally { + vi.useRealTimers(); + } + }); + it("does not re-deliver a stale startup snapshot after reconnect already acked it", async () => { const log = createRecoveryLog(); const startupLog = createRecoveryLog(); diff --git a/src/infra/outbound/delivery-queue.recovery.test.ts b/src/infra/outbound/delivery-queue.recovery.test.ts index b6ae91bf005..d81ad6a4993 100644 --- a/src/infra/outbound/delivery-queue.recovery.test.ts +++ b/src/infra/outbound/delivery-queue.recovery.test.ts @@ -3,6 +3,7 @@ import { MAX_DATE_TIMESTAMP_MS } from "@openclaw/normalization-core/number-coercion"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js"; +import { RECOVERY_REPLAY_SPACING_MS } from "../delivery-recovery.shared.js"; import { OutboundDeliveryError, type OutboundPayloadDeliveryOutcome } from "./deliver-types.js"; import { attachOutboundDeliveryCommitHook } from "./delivery-commit-hooks.js"; import { @@ -117,6 +118,81 @@ describe("delivery-queue recovery", () => { expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0); }); + it("paces startup replay instead of draining eligible entries back-to-back", async () => { + vi.useFakeTimers(); + const startedAt = new Date("2026-04-23T00:00:00.000Z"); + vi.setSystemTime(startedAt); + try { + await enqueueCrashRecoveryEntries(); + let firstDelivered!: () => void; + const firstDeliveredPromise = new Promise((resolve) => { + firstDelivered = resolve; + }); + const deliveryTimes: number[] = []; + const deliver = vi.fn(async () => { + deliveryTimes.push(Date.now()); + if (deliveryTimes.length === 1) { + firstDelivered(); + } + return []; + }); + + const recovery = runRecovery({ deliver, maxRecoveryMs: 60_000 }); + await firstDeliveredPromise; + expect(deliver).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(RECOVERY_REPLAY_SPACING_MS - 1); + expect(deliver).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(1); + const { result } = await recovery; + + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliveryTimes[1]).toBe(startedAt.getTime() + RECOVERY_REPLAY_SPACING_MS); + expect(result).toMatchObject({ recovered: 2, deferredBackoff: 0 }); + } finally { + vi.useRealTimers(); + } + }); + + it("counts replay pacing against the recovery budget and defers the backlog tail", async () => { + vi.useFakeTimers(); + const startedAt = new Date("2026-04-23T00:00:00.000Z"); + vi.setSystemTime(startedAt); + try { + await enqueueCrashRecoveryEntries(); + await enqueueDelivery( + { channel: "demo-channel-c", to: "#c", payloads: [{ text: "c" }] }, + tmpDir(), + ); + let firstDelivered!: () => void; + const firstDeliveredPromise = new Promise((resolve) => { + firstDelivered = resolve; + }); + const deliveryTimes: number[] = []; + const deliver = vi.fn(async () => { + deliveryTimes.push(Date.now()); + if (deliveryTimes.length === 1) { + firstDelivered(); + } + return []; + }); + + const recovery = runRecovery({ deliver, maxRecoveryMs: 1 }); + await firstDeliveredPromise; + + await vi.advanceTimersByTimeAsync(1); + const { result } = await recovery; + + expect(deliver).toHaveBeenCalledTimes(1); + expect(deliveryTimes).toEqual([startedAt.getTime()]); + expect(result).toMatchObject({ recovered: 1, deferredBackoff: 0 }); + expect(await loadPendingDeliveries(tmpDir())).toHaveLength(2); + } finally { + vi.useRealTimers(); + } + }); + it("moves entries that exceeded max retries to failed/", async () => { const id = await enqueueDelivery( { channel: "demo-channel-a", to: "+1", payloads: [{ text: "a" }] }, diff --git a/src/infra/session-delivery-queue-recovery.ts b/src/infra/session-delivery-queue-recovery.ts index 5a70128cf40..9efae9a44db 100644 --- a/src/infra/session-delivery-queue-recovery.ts +++ b/src/infra/session-delivery-queue-recovery.ts @@ -7,6 +7,7 @@ import { import { claimRecoveryEntry as claimSharedRecoveryEntry, computeBackoffMs, + createRecoveryReplayPacer, getErrnoCode, releaseRecoveryEntry as releaseSharedRecoveryEntry, } from "./delivery-recovery.shared.js"; @@ -46,6 +47,7 @@ const MAX_SESSION_DELIVERY_RETRIES = 5; const drainInProgress = new Map(); const entriesInProgress = new Set(); +const recoveryReplayPacer = createRecoveryReplayPacer(); function createEmptyRecoverySummary(): SessionDeliveryRecoverySummary { return { @@ -249,6 +251,14 @@ export async function recoverPendingSessionDeliveries(opts: { continue; } + const paceResult = await recoveryReplayPacer.wait(deadline); + if (paceResult === "deadline-exceeded") { + opts.log.warn( + "Session delivery recovery time budget exceeded — remaining entries deferred", + ); + break; + } + const result = await drainQueuedEntry({ entry: currentEntry, deliver: opts.deliver, diff --git a/src/infra/session-delivery-queue.recovery.test.ts b/src/infra/session-delivery-queue.recovery.test.ts index 4713da63ea8..eb8c8ddd950 100644 --- a/src/infra/session-delivery-queue.recovery.test.ts +++ b/src/infra/session-delivery-queue.recovery.test.ts @@ -2,6 +2,7 @@ import { MAX_DATE_TIMESTAMP_MS } from "@openclaw/normalization-core/number-coercion"; import { describe, expect, it, vi } from "vitest"; import { withTempDir } from "../test-helpers/temp-dir.js"; +import { RECOVERY_REPLAY_SPACING_MS } from "./delivery-recovery.shared.js"; import { drainPendingSessionDeliveries, enqueueSessionDelivery, @@ -40,6 +41,122 @@ describe("session-delivery queue recovery", () => { }); }); + it("paces startup replay for multiple eligible session deliveries", async () => { + vi.useFakeTimers(); + const startedAt = new Date("2026-04-23T00:00:00.000Z"); + vi.setSystemTime(startedAt); + try { + await withTempDir({ prefix: "openclaw-session-delivery-" }, async (tempDir) => { + await enqueueSessionDelivery( + { + kind: "systemEvent", + sessionKey: "agent:main:main", + text: "first", + }, + tempDir, + ); + await enqueueSessionDelivery( + { + kind: "systemEvent", + sessionKey: "agent:main:main", + text: "second", + }, + tempDir, + ); + + let firstDelivered!: () => void; + const firstDeliveredPromise = new Promise((resolve) => { + firstDelivered = resolve; + }); + const deliveryTimes: number[] = []; + const deliver = vi.fn(async () => { + deliveryTimes.push(Date.now()); + if (deliveryTimes.length === 1) { + firstDelivered(); + } + }); + + const recovery = recoverPendingSessionDeliveries({ + deliver, + stateDir: tempDir, + log: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, + }); + await firstDeliveredPromise; + expect(deliver).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(RECOVERY_REPLAY_SPACING_MS - 1); + expect(deliver).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(1); + const summary = await recovery; + + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliveryTimes[1]).toBe(startedAt.getTime() + RECOVERY_REPLAY_SPACING_MS); + expect(summary.recovered).toBe(2); + }); + } finally { + vi.useRealTimers(); + } + }); + + it("counts replay pacing against the session recovery budget", async () => { + vi.useFakeTimers(); + const startedAt = new Date("2026-04-23T00:00:00.000Z"); + vi.setSystemTime(startedAt); + try { + await withTempDir({ prefix: "openclaw-session-delivery-" }, async (tempDir) => { + for (const text of ["first", "second", "third"]) { + await enqueueSessionDelivery( + { + kind: "systemEvent", + sessionKey: "agent:main:main", + text, + }, + tempDir, + ); + } + + let firstDelivered!: () => void; + const firstDeliveredPromise = new Promise((resolve) => { + firstDelivered = resolve; + }); + const deliveryTimes: number[] = []; + const deliver = vi.fn(async () => { + deliveryTimes.push(Date.now()); + if (deliveryTimes.length === 1) { + firstDelivered(); + } + }); + + const recovery = recoverPendingSessionDeliveries({ + deliver, + stateDir: tempDir, + maxRecoveryMs: 1, + log: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, + }); + await firstDeliveredPromise; + + await vi.advanceTimersByTimeAsync(1); + const summary = await recovery; + + expect(deliver).toHaveBeenCalledTimes(1); + expect(deliveryTimes).toEqual([startedAt.getTime()]); + expect(summary.recovered).toBe(1); + expect(await loadPendingSessionDeliveries(tempDir)).toHaveLength(2); + }); + } finally { + vi.useRealTimers(); + } + }); + it("defers recovery when the recovery budget would exceed the date range", async () => { vi.useFakeTimers(); vi.setSystemTime(new Date(MAX_DATE_TIMESTAMP_MS));