ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-28 05:16:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor: consolidate channel and auto-reply tests (#114330)

Browse Source
This commit is contained in:
Peter Steinberger
2026-07-27 01:26:47 -04:00
committed by GitHub
parent b65dae1511
commit 71a472bf54
4 changed files with 582 additions and 1627 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auto-reply/reply/get-reply-run.media-only.test.ts
+135 -365
View File
File diff suppressed because it is too large Load Diff
src/auto-reply/reply/queue.collect.test.ts
+364 -986
View File
File diff suppressed because it is too large Load Diff
src/channels/message/ingress-drain.test.ts
+37 -117
View File
@@ -24,6 +24,21 @@ import {
type Payload = { text: string };
function createTestIngressQueue(
stateDir: string,
options: Omit<
Parameters<typeof createChannelIngressQueue>[0],
"channelId" | "accountId" | "stateDir"
> = {},
) {
return createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
...options,
});
}
async function withTempState<T>(fn: (stateDir: string) => Promise<T>): Promise<T> {
const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ingress-drain-"));
try {
@@ -46,12 +61,7 @@ describe("channel ingress drain", () => {
it("crash-window: lost claim is recovered and dispatched exactly once", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => 1_000,
});
const queue = createTestIngressQueue(stateDir, { now: () => 1_000 });
await queue.enqueue("evt-1", { text: "hello" }, { laneKey: "lane-a" });
const orphanClaim = await queue.claim("evt-1", { ownerId: "999:1:dead-owner" });
expect(orphanClaim).not.toBeNull();
@@ -84,11 +94,7 @@ describe("channel ingress drain", () => {
it("dispatches a resubmitted dead letter exactly once", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-replay", { text: "recover" }, { laneKey: "lane-a" });
const originalClaim = await queue.claim("evt-replay", { ownerId: "worker" });
if (!originalClaim) {
@@ -125,11 +131,7 @@ describe("channel ingress drain", () => {
it("complete-at-adoption: adoption tombstones; settle is not required", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-adopt", { text: "x" }, { laneKey: "l1" });
let settleResolve!: () => void;
@@ -162,11 +164,7 @@ describe("channel ingress drain", () => {
it("deferred holds claim without complete until adopted or abandoned", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-def", { text: "x" }, { laneKey: "l1" });
const capturedLifecycles: ChannelIngressDispatchLifecycle[] = [];
@@ -200,11 +198,7 @@ describe("channel ingress drain", () => {
it("lets callers await an abandoned claim release", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-await-abandon", { text: "x" }, { laneKey: "l1" });
let finishRelease!: () => void;
@@ -248,11 +242,7 @@ describe("channel ingress drain", () => {
it("abandoned via turnAdoptionLifecycle releases claim with attempt increment", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-q", { text: "x" }, { laneKey: "l1" });
const drain = createChannelIngressDrain<Payload>({
@@ -279,11 +269,7 @@ describe("channel ingress drain", () => {
it("queued deferral→admission completes the claim exactly once via turnAdoptionLifecycle", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-admit", { text: "x" }, { laneKey: "l1" });
let adoptCount = 0;
@@ -319,12 +305,7 @@ describe("channel ingress drain", () => {
it("watchdog only guillotines pre-adoption stalls with handler-timeout", async () => {
await withTempState(async (stateDir) => {
let clock = 10_000;
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => clock,
});
const queue = createTestIngressQueue(stateDir, { now: () => clock });
await queue.enqueue("evt-stall", { text: "x" }, { laneKey: "l1" });
const drain = createChannelIngressDrain<Payload>({
@@ -355,12 +336,7 @@ describe("channel ingress drain", () => {
it("watchdog guillotines deferred phase (timer not cleared by deferral)", async () => {
await withTempState(async (stateDir) => {
let clock = 30_000;
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => clock,
});
const queue = createTestIngressQueue(stateDir, { now: () => clock });
await queue.enqueue("evt-def-stall", { text: "x" }, { laneKey: "l1" });
const drain = createChannelIngressDrain<Payload>({
@@ -392,12 +368,7 @@ describe("channel ingress drain", () => {
it("watchdog does not kill healthy long turns after adoption", async () => {
await withTempState(async (stateDir) => {
let clock = 20_000;
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => clock,
});
const queue = createTestIngressQueue(stateDir, { now: () => clock });
await queue.enqueue("evt-long", { text: "x" }, { laneKey: "l1" });
let settleResolve!: () => void;
@@ -432,11 +403,7 @@ describe("channel ingress drain", () => {
it("supersede tombstones the superseded claim (never re-dispatches)", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("old", { text: "old" }, { laneKey: "shared" });
const firstLifecycles: ChannelIngressDispatchLifecycle[] = [];
@@ -497,11 +464,7 @@ describe("channel ingress drain", () => {
it("does not supersede without predicate", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("a1", { text: "a" }, { laneKey: "lane" });
let hold!: () => void;
@@ -542,12 +505,7 @@ describe("channel ingress drain", () => {
await withTempState(async (stateDir) => {
const receivedAt = 100;
let clock = receivedAt;
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => clock,
});
const queue = createTestIngressQueue(stateDir, { now: () => clock });
await queue.enqueue("poison", { text: "x" }, { laneKey: "l", receivedAt });
// Burn attempts without aging past the gate.
@@ -635,12 +593,7 @@ describe("channel ingress drain", () => {
it("refreshes active claims on claimLeaseMs/3 while deferred", async () => {
await withTempState(async (stateDir) => {
let clock = 1_000;
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => clock,
});
const queue = createTestIngressQueue(stateDir, { now: () => clock });
await queue.enqueue("evt-refresh", { text: "x" }, { laneKey: "l1" });
const refreshClaim = vi.fn(async () => true);
@@ -687,11 +640,7 @@ describe("channel ingress drain", () => {
it("throws IngressAdoptionLostError when onAdopted races supersede", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("old", { text: "old" }, { laneKey: "shared" });
const lifecycles: ChannelIngressDispatchLifecycle[] = [];
@@ -734,11 +683,7 @@ describe("channel ingress drain", () => {
it("retries tombstone complete failures then commits", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-tombstone", { text: "x" }, { laneKey: "l1" });
let completeAttempts = 0;
@@ -775,11 +720,7 @@ describe("channel ingress drain", () => {
it("holds claim ownership when tombstone complete keeps failing", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-wedge", { text: "x" }, { laneKey: "l1" });
queue.complete = async () => {
@@ -815,11 +756,7 @@ describe("channel ingress drain", () => {
it("does not steal live peer-drain claims; recovers after owner abort", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-peer", { text: "x" }, { laneKey: "l1" });
let releaseFirst!: () => void;
@@ -873,11 +810,7 @@ describe("channel ingress drain", () => {
it("throws IngressAdoptionLostError when complete returns false (lease reclaimed)", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-reclaim", { text: "x" }, { laneKey: "l1" });
queue.complete = async () => false;
@@ -913,11 +846,7 @@ describe("channel ingress drain", () => {
// Failure window: dispatch returns completed (side effects ran) but complete()
// write fails while phase was still dispatching — must not release for replay.
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("evt-completed-tombstone-fail", { text: "ran" }, { laneKey: "l1" });
queue.complete = async () => {
@@ -959,12 +888,7 @@ describe("channel ingress drain", () => {
it("refreshClaim false aborts the handler mid-dispatch (lease reclaimed)", async () => {
await withTempState(async (stateDir) => {
let clock = 1_000;
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
now: () => clock,
});
const queue = createTestIngressQueue(stateDir, { now: () => clock });
await queue.enqueue("evt-refresh-false", { text: "x" }, { laneKey: "l1" });
const refreshClaim = vi.fn(async () => false);
@@ -1018,11 +942,7 @@ describe("channel ingress drain", () => {
// Failure window: async shouldSupersedePending resolves after the pending
// handler has already adopted — must revalidate and no-op.
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<Payload>({
channelId: "test",
accountId: "a",
stateDir,
});
const queue = createTestIngressQueue(stateDir);
await queue.enqueue("old", { text: "old" }, { laneKey: "shared" });
let releaseOld!: () => void;
src/channels/message/ingress-queue.test.ts
+46 -159
View File
@@ -19,6 +19,21 @@ import { createChannelIngressQueue } from "./ingress-queue.js";
type ChannelIngressTestDatabase = Pick<OpenClawStateKyselyDatabase, "channel_ingress_events">;
function createTestIngressQueue<TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
stateDir: string,
options: Omit<
Parameters<typeof createChannelIngressQueue>[0],
"channelId" | "accountId" | "stateDir"
> = {},
) {
return createChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>({
channelId: "test",
accountId: "account",
stateDir,
...options,
});
}
async function withTempState<T>(fn: (stateDir: string) => Promise<T>): Promise<T> {
const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ingress-queue-"));
try {
@@ -36,16 +51,11 @@ describe("channel ingress queue", () => {
it("deduplicates pending and completed ingress events", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<
const queue = createTestIngressQueue<
{ text: string },
{ source: string },
{ handledBy: string }
>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 100,
});
>(stateDir, { now: () => 100 });
const accepted = await queue.enqueue(
"event-1",
@@ -137,12 +147,7 @@ describe("channel ingress queue", () => {
it("can bound pending scans and prune stale pending rows", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ index: number }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ index: number }>(stateDir, { now: () => clock++ });
await queue.enqueue("0002", { index: 2 });
await queue.enqueue("0001", { index: 1 });
@@ -160,12 +165,7 @@ describe("channel ingress queue", () => {
it("does not prune protected rows while enforcing max-entry limits", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ index: number }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 10,
});
const queue = createTestIngressQueue<{ index: number }>(stateDir, { now: () => 10 });
await queue.enqueue("z", { index: 1 });
await queue.enqueue("a", { index: 2 });
@@ -180,12 +180,7 @@ describe("channel ingress queue", () => {
it("prunes max-entry overflow across bounded batches", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ index: number }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ index: number }>(stateDir, { now: () => clock++ });
for (let index = 0; index < 520; index += 1) {
await queue.enqueue(String(index).padStart(4, "0"), { index });
@@ -202,12 +197,7 @@ describe("channel ingress queue", () => {
it("claims, releases, and skips blocked lanes", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ });
await queue.enqueue("a", { text: "blocked" }, { laneKey: "chat-1", receivedAt: 1 });
await queue.enqueue("b", { text: "open" }, { laneKey: "chat-2", receivedAt: 2 });
@@ -245,12 +235,7 @@ describe("channel ingress queue", () => {
it("claims next pending row by id when requested", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ });
await queue.enqueue("0002", { text: "second" }, { receivedAt: 1 });
await queue.enqueue("0001", { text: "first" }, { receivedAt: 2 });
@@ -267,12 +252,7 @@ describe("channel ingress queue", () => {
it("claims next only from candidate ids when provided", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ });
await queue.enqueue("a", { text: "outside snapshot" }, { receivedAt: 1 });
await queue.enqueue("b", { text: "inside snapshot" }, { receivedAt: 2 });
@@ -290,12 +270,7 @@ describe("channel ingress queue", () => {
it("derives missing lane keys before claiming next", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ lane: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ });
await queue.enqueue("a", { lane: "blocked" }, { receivedAt: 1 });
await queue.enqueue("b", { lane: "open" }, { receivedAt: 2 });
@@ -317,12 +292,7 @@ describe("channel ingress queue", () => {
it("blocks lanes claimed by candidate rows before claiming later candidates", async () => {
await withTempState(async (stateDir) => {
let clock = 1;
const queue = createChannelIngressQueue<{ lane: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => clock++,
});
const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ });
await queue.enqueue("a", { lane: "chat-1" }, { receivedAt: 1 });
await queue.enqueue("b", { lane: "chat-1" }, { receivedAt: 2 });
@@ -345,12 +315,7 @@ describe("channel ingress queue", () => {
it("requires claim tokens before mutating claimed rows", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 10,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 });
await queue.enqueue("event-1", { text: "claimed" });
const claimed = await queue.claim("event-1", { ownerId: "worker" });
@@ -371,12 +336,7 @@ describe("channel ingress queue", () => {
it("refreshes claimed rows only with the active claim token", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 10,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 });
await queue.enqueue("event-1", { text: "claimed" });
const claimed = await queue.claim("event-1", { ownerId: "worker" });
@@ -407,12 +367,7 @@ describe("channel ingress queue", () => {
it("does not let old claim tokens refresh recovered and reclaimed rows", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 10,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 });
await queue.enqueue("event-1", { text: "claimed" });
const oldClaim = await queue.claim("event-1", { ownerId: "worker-1" });
@@ -436,12 +391,7 @@ describe("channel ingress queue", () => {
it("does not recover a claim refreshed after stale recovery snapshots it", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 10,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 });
await queue.enqueue("event-1", { text: "claimed" });
const claimed = await queue.claim("event-1", { ownerId: "worker" });
@@ -470,12 +420,7 @@ describe("channel ingress queue", () => {
it("recovers stale claims and prunes completed or failed rows", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
now: () => 10,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 });
await queue.enqueue("old", { text: "old" });
await queue.enqueue("keep", { text: "keep" });
@@ -589,11 +534,7 @@ describe("channel ingress queue", () => {
it("skips a pending row with corrupt payload_json in listPending", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
await queue.enqueue("good-1", { text: "hello" });
insertCorruptRow(stateDir, '["test","account"]', "bad-1", {
@@ -609,11 +550,7 @@ describe("channel ingress queue", () => {
it("applies listPending limits after excluding corrupt payloads", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
for (let index = 0; index < 100; index += 1) {
insertCorruptRow(
stateDir,
@@ -632,11 +569,7 @@ describe("channel ingress queue", () => {
it("uses the queue JSON contract when listing deeply nested payloads", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<unknown>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<unknown>(stateDir);
const nestedJson = `${"[".repeat(1001)}0${"]".repeat(1001)}`;
const payload = JSON.parse(nestedJson);
@@ -648,11 +581,7 @@ describe("channel ingress queue", () => {
it("skips corrupt metadata_json in listPending", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }, { source: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }, { source: string }>(stateDir);
await queue.enqueue("ev-1", { text: "ok" }, { metadata: { source: "good" } });
insertCorruptRow(stateDir, '["test","account"]', "ev-bad-meta", {
@@ -669,11 +598,7 @@ describe("channel ingress queue", () => {
it("skips a claimed row with corrupt payload_json in listClaims", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
await queue.enqueue("claim-ok", { text: "ok" });
insertCorruptRow(stateDir, '["test","account"]', "claim-bad", {
@@ -699,11 +624,9 @@ describe("channel ingress queue", () => {
it("skips corrupt completed_metadata_json during duplicate detection", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }, unknown, { handler: string }>({
channelId: "test",
accountId: "account",
const queue = createTestIngressQueue<{ text: string }, unknown, { handler: string }>(
stateDir,
});
);
await queue.enqueue("comp-1", { text: "first" });
await queue.complete("comp-1", { metadata: { handler: "worker" }, completedAt: 150 });
@@ -729,11 +652,7 @@ describe("channel ingress queue", () => {
it("claimNext skips a corrupt first pending row without lane derivation", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
// Insert the bad row first so it sorts before the good row.
const earlyTime = 10;
@@ -774,11 +693,7 @@ describe("channel ingress queue", () => {
it("makes durable progress when a corrupt prefix fills the claim scan limit", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
insertCorruptRow(stateDir, '["test","account"]', "bad-first", {
payload_json: "{corrupt",
});
@@ -794,11 +709,7 @@ describe("channel ingress queue", () => {
it("bounds corrupt reconciliation work per claimNext call", async () => {
await withTempState(async (stateDir) => {
const queueName = '["test","account"]';
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
for (let index = 0; index < 101; index += 1) {
insertCorruptRow(stateDir, queueName, `bad-${index.toString().padStart(3, "0")}`, {
payload_json: "{corrupt",
@@ -825,11 +736,7 @@ describe("channel ingress queue", () => {
it("claim returns null for a corrupt pending row", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
await queue.enqueue("good-1", { text: "hello" });
insertCorruptRow(stateDir, '["test","account"]', "bad-direct", {
@@ -860,11 +767,7 @@ describe("channel ingress queue", () => {
it("handles valid JSON null payload correctly", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<null>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<null>(stateDir);
// Valid JSON null should parse as null, not be treated as corrupt.
await queue.enqueue("null-ok", null);
@@ -876,11 +779,7 @@ describe("channel ingress queue", () => {
it("tombstones a corrupt pending row on duplicate enqueue", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
insertCorruptRow(stateDir, '["test","account"]', "dup-bad", {
payload_json: "{corrupt",
@@ -915,11 +814,7 @@ describe("channel ingress queue", () => {
it("does not tombstone a corrupt actively claimed row on duplicate enqueue", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
insertCorruptRow(stateDir, '["test","account"]', "dup-claimed-bad", {
payload_json: "{corrupt",
status: "claimed",
@@ -955,11 +850,7 @@ describe("channel ingress queue", () => {
it("tombstones corrupt claimed rows during stale recovery", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
const oldTime = 10;
insertCorruptRow(stateDir, '["test","account"]', "stale-bad", {
@@ -998,11 +889,7 @@ describe("channel ingress queue", () => {
it("does not bypass recovery policy for a corrupt stale claim", async () => {
await withTempState(async (stateDir) => {
const queue = createChannelIngressQueue<{ text: string }>({
channelId: "test",
accountId: "account",
stateDir,
});
const queue = createTestIngressQueue<{ text: string }>(stateDir);
insertCorruptRow(stateDir, '["test","account"]', "stale-policy-bad", {
payload_json: "{corrupt",
status: "claimed",