ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

test(infra): skip real recovery pacing waits (#109338)

Browse Source
This commit is contained in:
Peter Steinberger
2026-07-16 14:12:07 -07:00
committed by GitHub
parent 7d71d7cf6b
commit 411c09f9be
4 changed files with 70 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infra/outbound/delivery-queue.reconnect-drain.test.ts
+9 -2
View File
@@ -1,6 +1,7 @@
// Covers reconnect-triggered queue drain selection, active claims, backoff
// bypass, and concurrent drain suppression.
import { beforeEach, describe, expect, it, vi } from "vitest";
import { controlNextRecoverySleep } from "../../../test/helpers/infra/delivery-recovery.js";
import type { OpenClawConfig } from "../../config/config.js";
import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js";
import { loadPendingDeliveries } from "./delivery-queue-storage.js";
@@ -25,6 +26,9 @@ const RECOVERY_REPLAY_SPACING_MS = 250;
const MAX_RETRIES = 5;
const stubCfg = {} as OpenClawConfig;
const NO_LISTENER_ERROR = "No active DirectChat listener";
const sleepMock = vi.hoisted(() => vi.fn<(ms: number) => Promise<void>>());
vi.mock("../../utils/sleep.js", () => ({ sleep: sleepMock }));
function normalizeReconnectAccountIdForTest(accountId?: string | null): string {
return (accountId ?? "").trim() || "default";
@@ -140,6 +144,8 @@ describe("drainPendingDeliveries for reconnect", () => {
beforeEach(() => {
tmpDir = fixtures.tmpDir();
sleepMock.mockReset();
sleepMock.mockResolvedValue(undefined);
});
it("drains entries that failed with 'no listener' error", async () => {
@@ -308,6 +314,7 @@ describe("drainPendingDeliveries for reconnect", () => {
const startedAt = new Date("2026-04-23T00:00:00.000Z");
vi.setSystemTime(startedAt);
try {
const controlledSleep = controlNextRecoverySleep(sleepMock);
const log = createRecoveryLog();
const startupLog = createRecoveryLog();
let firstStarted!: () => void;
@@ -344,9 +351,9 @@ describe("drainPendingDeliveries for reconnect", () => {
});
releaseFirst();
await vi.advanceTimersByTimeAsync(RECOVERY_REPLAY_SPACING_MS - 1);
await expect(controlledSleep.started).resolves.toBe(RECOVERY_REPLAY_SPACING_MS);
expect(deliver).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
controlledSleep.release();
await Promise.all([reconnectDrain, startupRecovery]);
expect(deliver).toHaveBeenCalledTimes(2);
src/infra/outbound/delivery-queue.recovery.test.ts
+12 -21
View File
@@ -4,6 +4,7 @@ import fs from "node:fs/promises";
import path from "node:path";
import { MAX_DATE_TIMESTAMP_MS } from "@openclaw/normalization-core/number-coercion";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { controlNextRecoverySleep } from "../../../test/helpers/infra/delivery-recovery.js";
import type { TrustedMessageAuditEvent } from "../../audit/message-audit-events.js";
import { onTrustedMessageAuditEventForTest as onTrustedMessageAuditEvent } from "../../audit/message-audit-events.test-support.js";
import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js";
@@ -34,10 +35,12 @@ import {
const RECOVERY_REPLAY_SPACING_MS = 250;
const MAX_RETRIES = 5;
const resolveOutboundChannelMessageAdapterMock = vi.hoisted(() => vi.fn());
const sleepMock = vi.hoisted(() => vi.fn<(ms: number) => Promise<void>>());
vi.mock("./channel-resolution.js", () => ({
resolveOutboundChannelMessageAdapter: resolveOutboundChannelMessageAdapterMock,
}));
vi.mock("../../utils/sleep.js", () => ({ sleep: sleepMock }));
function mockCallArg(mock: { mock: { calls: unknown[][] } }, index = 0): unknown {
const call = mock.mock.calls[index];
@@ -68,6 +71,8 @@ describe("delivery-queue recovery", () => {
beforeEach(() => {
resolveOutboundChannelMessageAdapterMock.mockReset();
sleepMock.mockReset();
sleepMock.mockResolvedValue(undefined);
});
const enqueueCrashRecoveryEntries = async () => {
@@ -199,28 +204,19 @@ describe("delivery-queue recovery", () => {
const startedAt = new Date("2026-04-23T00:00:00.000Z");
vi.setSystemTime(startedAt);
try {
const controlledSleep = controlNextRecoverySleep(sleepMock);
await enqueueCrashRecoveryEntries();
let firstDelivered!: () => void;
const firstDeliveredPromise = new Promise<void>((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);
await expect(controlledSleep.started).resolves.toBe(RECOVERY_REPLAY_SPACING_MS);
expect(deliver).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
controlledSleep.release();
const { result } = await recovery;
expect(deliver).toHaveBeenCalledTimes(2);
@@ -236,28 +232,23 @@ describe("delivery-queue recovery", () => {
const startedAt = new Date("2026-04-23T00:00:00.000Z");
vi.setSystemTime(startedAt);
try {
const controlledSleep = controlNextRecoverySleep(sleepMock);
await enqueueCrashRecoveryEntries();
await enqueueDelivery(
{ channel: "demo-channel-c", to: "#c", payloads: [{ text: "c" }] },
tmpDir(),
);
let firstDelivered!: () => void;
const firstDeliveredPromise = new Promise<void>((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);
await expect(controlledSleep.started).resolves.toBe(1);
expect(deliver).toHaveBeenCalledTimes(1);
controlledSleep.release();
const { result } = await recovery;
expect(deliver).toHaveBeenCalledTimes(1);
src/infra/session-delivery-queue.recovery.test.ts
+18 -22
View File
@@ -1,9 +1,14 @@
// Covers session delivery queue recovery behavior.
import { MAX_DATE_TIMESTAMP_MS } from "@openclaw/normalization-core/number-coercion";
import { describe, expect, it, vi } from "vitest";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { controlNextRecoverySleep } from "../../test/helpers/infra/delivery-recovery.js";
import { withTempDir } from "../test-helpers/temp-dir.js";
import { upsertDeliveryQueueEntry } from "./delivery-queue-sqlite.js";
const RECOVERY_REPLAY_SPACING_MS = 250;
const sleepMock = vi.hoisted(() => vi.fn<(ms: number) => Promise<void>>());
vi.mock("../utils/sleep.js", () => ({ sleep: sleepMock }));
import {
deferSessionDelivery,
failSessionDelivery,
@@ -22,6 +27,11 @@ import {
} from "./session-delivery-queue.js";
describe("session-delivery queue recovery", () => {
beforeEach(() => {
sleepMock.mockReset();
sleepMock.mockResolvedValue(undefined);
});
it("replays and acks pending entries on recovery", async () => {
await withTempDir({ prefix: "openclaw-session-delivery-" }, async (tempDir) => {
await enqueueSessionDelivery(
@@ -267,6 +277,7 @@ describe("session-delivery queue recovery", () => {
const startedAt = new Date("2026-04-23T00:00:00.000Z");
vi.setSystemTime(startedAt);
try {
const controlledSleep = controlNextRecoverySleep(sleepMock);
await withTempDir({ prefix: "openclaw-session-delivery-" }, async (tempDir) => {
await enqueueSessionDelivery(
{
@@ -285,16 +296,9 @@ describe("session-delivery queue recovery", () => {
tempDir,
);
let firstDelivered!: () => void;
const firstDeliveredPromise = new Promise<void>((resolve) => {
firstDelivered = resolve;
});
const deliveryTimes: number[] = [];
const deliver = vi.fn(async () => {
deliveryTimes.push(Date.now());
if (deliveryTimes.length === 1) {
firstDelivered();
}
});
const recovery = recoverPendingSessionDeliveries({
@@ -306,13 +310,10 @@ describe("session-delivery queue recovery", () => {
error: vi.fn(),
},
});
await firstDeliveredPromise;
expect(deliver).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(RECOVERY_REPLAY_SPACING_MS - 1);
await expect(controlledSleep.started).resolves.toBe(RECOVERY_REPLAY_SPACING_MS);
expect(deliver).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
controlledSleep.release();
const summary = await recovery;
expect(deliver).toHaveBeenCalledTimes(2);
@@ -329,6 +330,7 @@ describe("session-delivery queue recovery", () => {
const startedAt = new Date("2026-04-23T00:00:00.000Z");
vi.setSystemTime(startedAt);
try {
const controlledSleep = controlNextRecoverySleep(sleepMock);
await withTempDir({ prefix: "openclaw-session-delivery-" }, async (tempDir) => {
for (const text of ["first", "second", "third"]) {
await enqueueSessionDelivery(
@@ -341,16 +343,9 @@ describe("session-delivery queue recovery", () => {
);
}
let firstDelivered!: () => void;
const firstDeliveredPromise = new Promise<void>((resolve) => {
firstDelivered = resolve;
});
const deliveryTimes: number[] = [];
const deliver = vi.fn(async () => {
deliveryTimes.push(Date.now());
if (deliveryTimes.length === 1) {
firstDelivered();
}
});
const recovery = recoverPendingSessionDeliveries({
@@ -363,9 +358,10 @@ describe("session-delivery queue recovery", () => {
error: vi.fn(),
},
});
await firstDeliveredPromise;
await vi.advanceTimersByTimeAsync(1);
await expect(controlledSleep.started).resolves.toBe(1);
expect(deliver).toHaveBeenCalledTimes(1);
controlledSleep.release();
const summary = await recovery;
expect(deliver).toHaveBeenCalledTimes(1);
test/helpers/infra/delivery-recovery.ts
+31
View File
@@ -0,0 +1,31 @@
// Controls mocked recovery sleeps so tests prove callers stay blocked until release.
import { vi } from "vitest";
type SleepMock = {
mockImplementation(implementation: (ms: number) => Promise<void>): unknown;
};
export function controlNextRecoverySleep(sleepMock: SleepMock) {
let releaseSleep: (() => void) | undefined;
const started = new Promise<number>((resolveStarted) => {
sleepMock.mockImplementation(
(ms) =>
new Promise<void>((resolve) => {
releaseSleep = () => {
vi.setSystemTime(Date.now() + ms);
resolve();
};
resolveStarted(ms);
}),
);
});
return {
started,
release() {
if (!releaseSleep) {
throw new Error("Expected recovery sleep to start before release");
}
releaseSleep();
},
};
}