Skip to content

Commit 1fc1230

Browse files
lee-xydtclaude
lee-xydt
and
claude
committed
fix(channels): add idempotency guard to nack() (#104903)
nack() lacked the idempotency guard that ack() already has. In receive pipelines that revisit completed stages, duplicate nack() calls would re-fire onNack callbacks and overwrite nackErrorMessage. Add a guard matching ack(): skip when ackState is not "pending". Fixes #104903. Co-Authored-By: Claude <[email protected]>
1 parent 9c95abd commit 1fc1230

3 files changed

Lines changed: 174 additions & 0 deletions

File tree

‎scripts/proof-issue-104903.ts‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
/**
2+
* Proof script for issue #104903 fix.
3+
*
4+
* Demonstrates that nack() is now idempotent — duplicate calls
5+
* are no-ops and onNack fires only once.
6+
*
7+
* Usage: npx tsx scripts/proof-issue-104903.ts
8+
*/
9+
import { createMessageReceiveContext } from "../src/channels/message/receive.js";
10+
11+
const divider = "=".repeat(64);
12+
let exitCode = 0;
13+
14+
function check(cond: boolean, label: string): void {
15+
console.log(` ${cond ? "✓" : "✗"} ${label}`);
16+
if (!cond) {
17+
exitCode = 1;
18+
}
19+
}
20+
21+
console.log(divider);
22+
console.log("PROOF: nack() idempotency guard — issue #104903");
23+
console.log(divider);
24+
25+
// ── Test: duplicate nack calls ──────────────────────────────────────
26+
console.log("\n--- Duplicate nack calls ---\n");
27+
28+
let nackCount = 0;
29+
const ctx = createMessageReceiveContext({
30+
id: "test",
31+
channel: "test",
32+
message: {},
33+
onNack: () => {
34+
nackCount++;
35+
},
36+
});
37+
38+
await ctx.nack(new Error("first error"));
39+
console.log(" First nack: onNack called");
40+
await ctx.nack(new Error("second error - should be ignored"));
41+
console.log(" Second nack: no-op (guard prevented onNack)");
42+
43+
console.log();
44+
check(nackCount === 1, `onNack called exactly once (${nackCount})`);
45+
check(ctx.ackState === "nacked", `ackState is "nacked"`);
46+
check(ctx.nackErrorMessage?.includes("first error") ?? false, "first error message preserved");
47+
48+
// ── BEFORE/AFTER ───────────────────────────────────────────────────
49+
console.log("\n" + divider);
50+
console.log("BEFORE/AFTER");
51+
console.log(divider);
52+
console.log(`
53+
BEFORE FIX:
54+
nack: async (error) => {
55+
await params.onNack?.(error); // Fires every call
56+
ctx.ackState = "nacked";
57+
ctx.nackErrorMessage = normalizeAckErrorMessage(error);
58+
},
59+
→ Duplicate calls re-fire onNack and overwrite nackErrorMessage
60+
61+
AFTER FIX:
62+
nack: async (error) => {
63+
if (ctx.ackState !== "pending") { return; } // Idempotent guard
64+
await params.onNack?.(error);
65+
ctx.ackState = "nacked";
66+
ctx.nackErrorMessage = normalizeAckErrorMessage(error);
67+
},
68+
→ Duplicate calls are no-ops, matching ack() behavior
69+
`);
70+
71+
console.log(divider);
72+
console.log("RESULT");
73+
console.log(divider);
74+
if (exitCode === 0) {
75+
console.log(" ALL CHECKS PASSED ✓");
76+
} else {
77+
console.log(" SOME CHECKS FAILED ✗");
78+
}
79+
console.log();
80+
console.log("Fix: src/channels/message/receive.ts (+4 lines)");
81+
console.log("Test: src/channels/message/receive.test.ts (5 tests)");
82+
console.log("Verified on: " + new Date().toISOString());
83+
console.log(divider);
84+
process.exit(exitCode);

‎src/channels/message/receive.test.ts‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
/**
2+
* Tests for createMessageReceiveContext with ack/nack idempotency.
3+
*/
4+
import { describe, expect, it, vi } from "vitest";
5+
import { createMessageReceiveContext } from "./receive.js";
6+
7+
describe("createMessageReceiveContext ack/nack", () => {
8+
it("nack ignores duplicate calls (idempotent guard) (#104903)", async () => {
9+
const onNack = vi.fn();
10+
const ctx = createMessageReceiveContext({
11+
id: "test",
12+
channel: "test",
13+
message: {},
14+
onNack,
15+
});
16+
17+
await ctx.nack(new Error("first"));
18+
await ctx.nack(new Error("second"));
19+
20+
// onNack must only fire once — duplicate calls are no-ops.
21+
expect(onNack).toHaveBeenCalledTimes(1);
22+
});
23+
24+
it("nack preserves the first error message on duplicate calls", async () => {
25+
const ctx = createMessageReceiveContext({
26+
id: "test",
27+
channel: "test",
28+
message: {},
29+
});
30+
31+
await ctx.nack(new Error("first error"));
32+
33+
// Second call must not overwrite the first nack error.
34+
await ctx.nack(new Error("duplicate error"));
35+
36+
expect(ctx.nackErrorMessage).toContain("first error");
37+
expect(ctx.nackErrorMessage).not.toContain("duplicate error");
38+
});
39+
40+
it("ack remains idempotent (regression guard)", async () => {
41+
const onAck = vi.fn();
42+
const ctx = createMessageReceiveContext({
43+
id: "test",
44+
channel: "test",
45+
message: {},
46+
onAck,
47+
});
48+
49+
await ctx.ack();
50+
await ctx.ack();
51+
52+
expect(onAck).toHaveBeenCalledTimes(1);
53+
});
54+
55+
it("nack after ack is a no-op", async () => {
56+
const onAck = vi.fn();
57+
const onNack = vi.fn();
58+
const ctx = createMessageReceiveContext({
59+
id: "test",
60+
channel: "test",
61+
message: {},
62+
onAck,
63+
onNack,
64+
});
65+
66+
await ctx.ack();
67+
await ctx.nack(new Error("after ack"));
68+
69+
expect(onAck).toHaveBeenCalledTimes(1);
70+
expect(onNack).not.toHaveBeenCalled();
71+
});
72+
73+
it("nack sets ackState correctly", async () => {
74+
const ctx = createMessageReceiveContext({
75+
id: "test",
76+
channel: "test",
77+
message: {},
78+
});
79+
expect(ctx.ackState).toBe("pending");
80+
81+
await ctx.nack(new Error("test"));
82+
expect(ctx.ackState).toBe("nacked");
83+
});
84+
});

‎src/channels/message/receive.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,12 @@ export function createMessageReceiveContext<TMessage>(params: {
8888
delete ctx.nackErrorMessage;
8989
},
9090
nack: async (error) => {
91+
// Nack callbacks must be idempotent because receive pipelines may
92+
// revisit completed stages — same reason as ack() above.
93+
// See: https://github.com/openclaw/openclaw/issues/104903
94+
if (ctx.ackState !== "pending") {
95+
return;
96+
}
9197
await params.onNack?.(error);
9298
ctx.ackState = "nacked";
9399
ctx.nackErrorMessage = normalizeAckErrorMessage(error);

0 commit comments

Comments
 (0)