Skip to content

Commit 5199bfa

Browse files
steipeteNianJiuZst
steipete
and
NianJiuZst
authored
fix(line): preserve webhook events through dispatch failures (#109655)
* test(line): remove obsolete replay-cache coverage * fix(line): durably spool webhook events Co-authored-by: NianJiuZst <[email protected]> * test(line): tighten webhook spool lifecycle proof * fix(line): use canonical turn adoption lifecycle Co-authored-by: NianJiuZst <[email protected]> * refactor(line): keep spool types private Co-authored-by: NianJiuZst <[email protected]> --------- Co-authored-by: NianJiuZst <[email protected]>
1 parent cc0b847 commit 5199bfa

11 files changed

Lines changed: 1341 additions & 462 deletions

File tree

‎docs/channels/line.md‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,9 +42,12 @@ openclaw plugins install ./path/to/local/line-plugin
4242
https://gateway-host/line/webhook
4343
```
4444

45-
The Gateway answers LINE's webhook verification (GET) and acknowledges signed
46-
inbound events (POST) immediately after signature and payload validation; agent
47-
processing continues asynchronously.
45+
The Gateway answers LINE's webhook verification (GET). For signed inbound events
46+
(POST), it writes each event to the durable ingress queue before returning `200`;
47+
agent processing continues asynchronously. Failed delivery is retried from the
48+
queue, including after a Gateway restart, and poison events become failed queue
49+
records after bounded retries. If durable persistence fails, the request returns
50+
`500` instead of acknowledging an event that could be lost.
4851
If you need a custom path, set `channels.line.webhookPath` or
4952
`channels.line.accounts.<id>.webhookPath` and update the URL accordingly.
5053

‎extensions/line/src/bot-handlers.test.ts‎

Lines changed: 1 addition & 222 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vites
55
import type { LineAccountConfig } from "./types.js";
66

77
type MessageEvent = webhook.MessageEvent;
8-
type PostbackEvent = webhook.PostbackEvent;
98

109
// Avoid pulling in globals/pairing/media dependencies; this suite only asserts
1110
// allowlist/groupPolicy gating and message-context wiring.
@@ -176,7 +175,6 @@ vi.mock("./bot-message-context.js", () => ({
176175
}));
177176

178177
let handleLineWebhookEvents: typeof import("./bot-handlers.js").handleLineWebhookEvents;
179-
let createLineWebhookReplayCache: typeof import("./bot-handlers.js").createLineWebhookReplayCache;
180178
type LineWebhookContext = Parameters<typeof import("./bot-handlers.js").handleLineWebhookEvents>[1];
181179

182180
const createRuntime = () => ({ log: vi.fn(), error: vi.fn(), exit: vi.fn() });
@@ -228,7 +226,6 @@ function createLineWebhookTestContext(params: {
228226
groupAllowFrom?: LineAccountConfig["groupAllowFrom"];
229227
requireMention?: boolean;
230228
groupHistories?: Map<string, HistoryEntry[]>;
231-
replayCache?: ReturnType<typeof createLineWebhookReplayCache>;
232229
accessGroups?: Record<string, { type: "message.senders"; members: Record<string, string[]> }>;
233230
}): Parameters<typeof handleLineWebhookEvents>[1] {
234231
const allowFrom = params.allowFrom ?? (params.dmPolicy === "open" ? ["*"] : undefined);
@@ -260,22 +257,9 @@ function createLineWebhookTestContext(params: {
260257
mediaMaxBytes: 1,
261258
processMessage: params.processMessage,
262259
...(params.groupHistories ? { groupHistories: params.groupHistories } : {}),
263-
...(params.replayCache ? { replayCache: params.replayCache } : {}),
264260
};
265261
}
266262

267-
function createOpenGroupReplayContext(
268-
processMessage: LineWebhookContext["processMessage"],
269-
replayCache: ReturnType<typeof createLineWebhookReplayCache>,
270-
): Parameters<typeof handleLineWebhookEvents>[1] {
271-
return createLineWebhookTestContext({
272-
processMessage,
273-
groupPolicy: "open",
274-
requireMention: false,
275-
replayCache,
276-
});
277-
}
278-
279263
async function expectGroupMessageBlocked(params: {
280264
processMessage: LineWebhookContext["processMessage"];
281265
event: MessageEvent;
@@ -300,23 +284,9 @@ async function expectRequireMentionGroupMessageProcessed(event: MessageEvent) {
300284
expect(processMessage).toHaveBeenCalledTimes(1);
301285
}
302286

303-
async function startInflightReplayDuplicate(params: {
304-
event: MessageEvent;
305-
processMessage: LineWebhookContext["processMessage"];
306-
}) {
307-
const context = createOpenGroupReplayContext(
308-
params.processMessage,
309-
createLineWebhookReplayCache(),
310-
);
311-
const firstRun = handleLineWebhookEvents([params.event], context);
312-
await Promise.resolve();
313-
const secondRun = handleLineWebhookEvents([params.event], context);
314-
return { firstRun, secondRun };
315-
}
316-
317287
describe("handleLineWebhookEvents", () => {
318288
beforeAll(async () => {
319-
({ handleLineWebhookEvents, createLineWebhookReplayCache } = await import("./bot-handlers.js"));
289+
({ handleLineWebhookEvents } = await import("./bot-handlers.js"));
320290
});
321291

322292
afterAll(() => {
@@ -803,173 +773,6 @@ describe("handleLineWebhookEvents", () => {
803773
expect(pairingRequest?.accountId).toBe("work");
804774
});
805775

806-
it("deduplicates replayed webhook events by webhookEventId before processing", async () => {
807-
const processMessage = vi.fn();
808-
const event = createReplayMessageEvent({
809-
messageId: "m-replay",
810-
groupId: "group-replay",
811-
userId: "user-replay",
812-
webhookEventId: "evt-replay-1",
813-
isRedelivery: true,
814-
});
815-
const context = createOpenGroupReplayContext(processMessage, createLineWebhookReplayCache());
816-
817-
await handleLineWebhookEvents([event], context);
818-
await handleLineWebhookEvents([event], context);
819-
820-
expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1);
821-
expect(processMessage).toHaveBeenCalledTimes(1);
822-
});
823-
824-
it("skips concurrent redeliveries while the first event is still processing", async () => {
825-
let resolveFirst: (() => void) | undefined;
826-
const firstDone = new Promise<void>((resolve) => {
827-
resolveFirst = resolve;
828-
});
829-
const processMessage = vi.fn(async () => {
830-
await firstDone;
831-
});
832-
const event = createReplayMessageEvent({
833-
messageId: "m-inflight",
834-
groupId: "group-inflight",
835-
userId: "user-inflight",
836-
webhookEventId: "evt-inflight-1",
837-
isRedelivery: true,
838-
});
839-
const { firstRun, secondRun } = await startInflightReplayDuplicate({ event, processMessage });
840-
resolveFirst?.();
841-
await Promise.all([firstRun, secondRun]);
842-
843-
expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1);
844-
expect(processMessage).toHaveBeenCalledTimes(1);
845-
});
846-
847-
it("commits in-flight failures so concurrent duplicates do not retry", async () => {
848-
let rejectFirst: ((err: Error) => void) | undefined;
849-
const firstDone = new Promise<void>((_, reject) => {
850-
rejectFirst = reject;
851-
});
852-
const processMessage = vi.fn(async () => {
853-
await firstDone;
854-
});
855-
const event = createReplayMessageEvent({
856-
messageId: "m-inflight-fail",
857-
groupId: "group-inflight",
858-
userId: "user-inflight",
859-
webhookEventId: "evt-inflight-fail-1",
860-
isRedelivery: true,
861-
});
862-
const { firstRun, secondRun } = await startInflightReplayDuplicate({ event, processMessage });
863-
const firstFailure = expect(firstRun).rejects.toThrow("transient inflight failure");
864-
rejectFirst?.(new Error("transient inflight failure"));
865-
866-
await firstFailure;
867-
await expect(secondRun).resolves.toBeUndefined();
868-
expect(processMessage).toHaveBeenCalledTimes(1);
869-
});
870-
871-
it("deduplicates redeliveries by LINE message id when webhookEventId changes", async () => {
872-
const processMessage = vi.fn();
873-
const event = {
874-
type: "message",
875-
message: { id: "m-dup-1", type: "text", text: "hello" },
876-
replyToken: "reply-token",
877-
timestamp: Date.now(),
878-
source: { type: "group", groupId: "group-dup", userId: "user-dup" },
879-
mode: "active",
880-
webhookEventId: "evt-dup-1",
881-
deliveryContext: { isRedelivery: false },
882-
} as MessageEvent;
883-
884-
const context: Parameters<typeof handleLineWebhookEvents>[1] = {
885-
cfg: {
886-
channels: { line: { groupPolicy: "allowlist", groupAllowFrom: ["user-dup"] } },
887-
},
888-
account: {
889-
accountId: "default",
890-
enabled: true,
891-
channelAccessToken: "token",
892-
channelSecret: "secret",
893-
tokenSource: "config",
894-
config: {
895-
groupPolicy: "allowlist",
896-
groupAllowFrom: ["user-dup"],
897-
groups: { "*": { requireMention: false } },
898-
},
899-
},
900-
runtime: createRuntime(),
901-
mediaMaxBytes: 1,
902-
processMessage,
903-
replayCache: createLineWebhookReplayCache(),
904-
};
905-
906-
await handleLineWebhookEvents([event], context);
907-
await handleLineWebhookEvents(
908-
[
909-
{
910-
...event,
911-
webhookEventId: "evt-dup-redelivery",
912-
deliveryContext: { isRedelivery: true },
913-
} as MessageEvent,
914-
],
915-
context,
916-
);
917-
918-
expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1);
919-
expect(processMessage).toHaveBeenCalledTimes(1);
920-
});
921-
922-
it("deduplicates postback redeliveries by webhookEventId when replyToken changes", async () => {
923-
const processMessage = vi.fn();
924-
buildLinePostbackContextMock.mockResolvedValue({
925-
ctxPayload: { From: "line:user:user-postback" },
926-
route: { agentId: "default" },
927-
isGroup: false,
928-
accountId: "default",
929-
});
930-
const event = {
931-
type: "postback",
932-
postback: { data: "action=confirm" },
933-
replyToken: "reply-token-1",
934-
timestamp: Date.now(),
935-
source: { type: "user", userId: "user-postback" },
936-
mode: "active",
937-
webhookEventId: "evt-postback-1",
938-
deliveryContext: { isRedelivery: false },
939-
} as PostbackEvent;
940-
941-
const context: Parameters<typeof handleLineWebhookEvents>[1] = {
942-
cfg: { channels: { line: { dmPolicy: "open", allowFrom: ["*"] } } },
943-
account: {
944-
accountId: "default",
945-
enabled: true,
946-
channelAccessToken: "token",
947-
channelSecret: "secret",
948-
tokenSource: "config",
949-
config: { dmPolicy: "open", allowFrom: ["*"] },
950-
},
951-
runtime: createRuntime(),
952-
mediaMaxBytes: 1,
953-
processMessage,
954-
replayCache: createLineWebhookReplayCache(),
955-
};
956-
957-
await handleLineWebhookEvents([event], context);
958-
await handleLineWebhookEvents(
959-
[
960-
{
961-
...event,
962-
replyToken: "reply-token-2",
963-
deliveryContext: { isRedelivery: true },
964-
} as PostbackEvent,
965-
],
966-
context,
967-
);
968-
969-
expect(buildLinePostbackContextMock).toHaveBeenCalledTimes(1);
970-
expect(processMessage).toHaveBeenCalledTimes(1);
971-
});
972-
973776
it("skips group messages by default when requireMention is not configured", async () => {
974777
const processMessage = vi.fn();
975778
const event = createTestMessageEvent({
@@ -1508,29 +1311,5 @@ describe("handleLineWebhookEvents", () => {
15081311
// Should be skipped because there is a non-bot mention and the bot was not mentioned.
15091312
expect(processMessage).not.toHaveBeenCalled();
15101313
});
1511-
1512-
it("keeps replay cache committed after a non-retryable event failure", async () => {
1513-
const processMessage = vi
1514-
.fn()
1515-
.mockRejectedValueOnce(new Error("transient failure"))
1516-
.mockResolvedValueOnce(undefined);
1517-
const event = createReplayMessageEvent({
1518-
messageId: "m-fail-then-retry",
1519-
groupId: "group-retry",
1520-
userId: "user-retry",
1521-
webhookEventId: "evt-fail-then-retry",
1522-
isRedelivery: false,
1523-
});
1524-
const context = createOpenGroupReplayContext(processMessage, createLineWebhookReplayCache());
1525-
1526-
await expect(handleLineWebhookEvents([event], context)).rejects.toThrow("transient failure");
1527-
await handleLineWebhookEvents([event], context);
1528-
1529-
expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1);
1530-
expect(processMessage).toHaveBeenCalledTimes(1);
1531-
expect(context.runtime.error).toHaveBeenCalledWith(
1532-
"line: event handler failed: Error: transient failure",
1533-
);
1534-
});
15351314
});
15361315
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */

0 commit comments

Comments
 (0)