Skip to content

Commit 1c55e28

Browse files
pash-openaipash-openai
committed
fix(slack): wait for relay dispatch before ack
1 parent db5d750 commit 1c55e28

4 files changed

Lines changed: 248 additions & 102 deletions

File tree

‎extensions/slack/src/monitor/message-handler.test.ts‎

Lines changed: 76 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,9 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
33

44
const enqueueMock = vi.fn(async (_entry: unknown) => {});
55
const flushKeyMock = vi.fn(async (_key: string) => {});
6+
const onFlushCallbacks: Array<(entries: Array<Record<string, unknown>>) => Promise<void>> = [];
7+
const prepareSlackMessageMock = vi.fn(async () => ({ ctxPayload: {} }));
8+
const dispatchPreparedSlackMessageMock = vi.fn(async () => {});
69
const resolveThreadTsMock = vi.fn(async ({ message }: { message: Record<string, unknown> }) => ({
710
...message,
811
}));
@@ -14,13 +17,18 @@ vi.mock("openclaw/plugin-sdk/channel-inbound", async () => {
1417
);
1518
return {
1619
...actual,
17-
createChannelInboundDebouncer: () => ({
18-
debounceMs: 10,
19-
debouncer: {
20-
enqueue: (entry: unknown) => enqueueMock(entry),
21-
flushKey: (key: string) => flushKeyMock(key),
22-
},
23-
}),
20+
createChannelInboundDebouncer: (params: {
21+
onFlush: (entries: Array<Record<string, unknown>>) => Promise<void>;
22+
}) => {
23+
onFlushCallbacks.push(params.onFlush);
24+
return {
25+
debounceMs: 10,
26+
debouncer: {
27+
enqueue: (entry: unknown) => enqueueMock(entry),
28+
flushKey: (key: string) => flushKeyMock(key),
29+
},
30+
};
31+
},
2432
shouldDebounceTextInbound: ({ hasMedia }: { hasMedia?: boolean }) => !hasMedia,
2533
};
2634
});
@@ -31,6 +39,16 @@ vi.mock("./thread-resolution.js", () => ({
3139
}),
3240
}));
3341

42+
vi.mock("./message-handler/pipeline.runtime.js", () => ({
43+
prepareSlackMessage: prepareSlackMessageMock,
44+
dispatchPreparedSlackMessage: dispatchPreparedSlackMessageMock,
45+
}));
46+
47+
vi.mock("./inbound-delivery-state.js", () => ({
48+
hasSlackInboundMessageDelivery: vi.fn(async () => false),
49+
recordSlackInboundMessageDeliveries: vi.fn(async () => {}),
50+
}));
51+
3452
function createContext(overrides?: {
3553
markMessageSeen?: (channel: string | undefined, ts: string | undefined) => boolean;
3654
releaseSeenMessage?: (channel: string | undefined, ts: string | undefined) => void;
@@ -80,6 +98,9 @@ describe("createSlackMessageHandler", () => {
8098
beforeEach(() => {
8199
enqueueMock.mockClear();
82100
flushKeyMock.mockClear();
101+
onFlushCallbacks.length = 0;
102+
prepareSlackMessageMock.mockClear();
103+
dispatchPreparedSlackMessageMock.mockClear();
83104
resolveThreadTsMock.mockClear();
84105
});
85106

@@ -201,4 +222,52 @@ describe("createSlackMessageHandler", () => {
201222

202223
expect(flushKeyMock).toHaveBeenCalledWith("slack:default:C111:1709000000.000100:U111");
203224
});
225+
226+
it("waits for debounced dispatch completion when requested by relay delivery", async () => {
227+
const { handler } = createHandlerWithTracker();
228+
const handled = handler(
229+
{
230+
type: "message",
231+
channel: "C111",
232+
user: "U111",
233+
ts: "1709000000.000500",
234+
text: "relay message",
235+
} as never,
236+
{ source: "message", awaitDispatch: true },
237+
);
238+
239+
await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(1));
240+
const entry = enqueueMock.mock.calls[0]?.[0] as Record<string, unknown>;
241+
let settled = false;
242+
void handled.then(() => {
243+
settled = true;
244+
});
245+
await Promise.resolve();
246+
expect(settled).toBe(false);
247+
248+
await onFlushCallbacks[0]?.([entry]);
249+
await expect(handled).resolves.toBeUndefined();
250+
expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1);
251+
});
252+
253+
it("propagates debounced dispatch failures to relay delivery", async () => {
254+
dispatchPreparedSlackMessageMock.mockRejectedValueOnce(new Error("dispatch failed"));
255+
const { handler } = createHandlerWithTracker();
256+
const handled = handler(
257+
{
258+
type: "message",
259+
channel: "C111",
260+
user: "U111",
261+
ts: "1709000000.000600",
262+
text: "relay message",
263+
} as never,
264+
{ source: "message", awaitDispatch: true },
265+
);
266+
267+
await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(1));
268+
const entry = enqueueMock.mock.calls[0]?.[0] as Record<string, unknown>;
269+
const handledFailure = expect(handled).rejects.toThrow("dispatch failed");
270+
const flushFailure = expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow("dispatch failed");
271+
await Promise.all([handledFailure, flushFailure]);
272+
});
204273
});

‎extensions/slack/src/monitor/message-handler.ts‎

Lines changed: 141 additions & 88 deletions
Original file line numberDiff line numberDiff line change
@@ -38,9 +38,31 @@ export type SlackMessageHandler = (
3838
source: "message" | "app_mention";
3939
wasMentioned?: boolean;
4040
relayIdentity?: SlackSendIdentity;
41+
/** Wait until any inbound debounce flush and dispatch has completed. */
42+
awaitDispatch?: boolean;
4143
},
4244
) => Promise<void>;
4345

46+
type SlackDispatchCompletion = {
47+
promise: Promise<void>;
48+
resolve: () => void;
49+
reject: (error: unknown) => void;
50+
};
51+
52+
type QueuedSlackMessageOptions = Parameters<SlackMessageHandler>[1] & {
53+
dispatchCompletion?: Omit<SlackDispatchCompletion, "promise">;
54+
};
55+
56+
function createSlackDispatchCompletion(): SlackDispatchCompletion {
57+
let resolve!: () => void;
58+
let reject!: (error: unknown) => void;
59+
const promise = new Promise<void>((nextResolve, nextReject) => {
60+
resolve = nextResolve;
61+
reject = nextReject;
62+
});
63+
return { promise, resolve, reject };
64+
}
65+
4466
const APP_MENTION_RETRY_TTL_MS = 60_000;
4567

4668
export class SlackRetryableInboundError extends Error {
@@ -76,107 +98,123 @@ export function createSlackMessageHandler(params: {
7698
const { ctx, account, trackEvent } = params;
7799
const { debounceMs, debouncer } = createChannelInboundDebouncer<{
78100
message: SlackMessageEvent;
79-
opts: {
80-
source: "message" | "app_mention";
81-
wasMentioned?: boolean;
82-
relayIdentity?: SlackSendIdentity;
83-
};
101+
opts: QueuedSlackMessageOptions;
84102
}>({
85103
cfg: ctx.cfg,
86104
channel: "slack",
87105
buildKey: (entry) => buildSlackDebounceKey(entry.message, ctx.accountId),
88106
shouldDebounce: (entry) => shouldDebounceSlackMessage(entry.message, ctx.cfg),
89107
onFlush: async (entries) => {
90-
const last = entries.at(-1);
91-
if (!last) {
92-
return;
93-
}
94-
const flushedKey = buildSlackDebounceKey(last.message, ctx.accountId);
95-
const topLevelConversationKey = buildTopLevelSlackConversationKey(
96-
last.message,
97-
ctx.accountId,
98-
);
99-
if (flushedKey && topLevelConversationKey) {
100-
const pendingKeys = pendingTopLevelDebounceKeys.get(topLevelConversationKey);
101-
if (pendingKeys) {
102-
pendingKeys.delete(flushedKey);
103-
if (pendingKeys.size === 0) {
104-
pendingTopLevelDebounceKeys.delete(topLevelConversationKey);
105-
}
106-
}
107-
}
108-
const combinedText =
109-
entries.length === 1
110-
? (last.message.text ?? "")
111-
: entries
112-
.map((entry) => entry.message.text ?? "")
113-
.filter(Boolean)
114-
.join("\n");
115-
const combinedMentioned = entries.some((entry) => Boolean(entry.opts.wasMentioned));
116-
const syntheticMessage: SlackMessageEvent = {
117-
...last.message,
118-
text: combinedText,
119-
};
120-
const seenMessageKey = buildSeenMessageKey(last.message.channel, last.message.ts);
108+
const completions = entries
109+
.map((entry) => entry.opts.dispatchCompletion)
110+
.filter((completion) => completion !== undefined);
121111
try {
122-
const { prepareSlackMessage, dispatchPreparedSlackMessage } =
123-
await loadSlackMessagePipeline();
124-
const prepared = await prepareSlackMessage({
125-
ctx,
126-
account,
127-
message: syntheticMessage,
128-
opts: {
129-
...last.opts,
130-
wasMentioned: combinedMentioned || last.opts.wasMentioned,
131-
},
132-
});
133-
if (!prepared) {
134-
return;
135-
}
136-
if (seenMessageKey) {
137-
pruneAppMentionRetryKeys(Date.now());
138-
if (last.opts.source === "app_mention") {
139-
// If app_mention wins the race and dispatches first, drop the later message dispatch.
140-
rememberExpiringAppMentionKey(appMentionDispatchedKeys, seenMessageKey);
141-
} else if (
142-
last.opts.source === "message" &&
143-
appMentionDispatchedKeys.has(seenMessageKey)
144-
) {
145-
appMentionDispatchedKeys.delete(seenMessageKey);
146-
appMentionRetryKeys.delete(seenMessageKey);
112+
await (async () => {
113+
const last = entries.at(-1);
114+
if (!last) {
147115
return;
148116
}
149-
appMentionRetryKeys.delete(seenMessageKey);
150-
}
151-
if (entries.length > 1) {
152-
const ids = entries.map((entry) => entry.message.ts).filter(Boolean) as string[];
153-
if (ids.length > 0) {
154-
prepared.ctxPayload.MessageSids = ids;
155-
prepared.ctxPayload.MessageSidFirst = ids[0];
156-
prepared.ctxPayload.MessageSidLast = ids[ids.length - 1];
117+
const flushedKey = buildSlackDebounceKey(last.message, ctx.accountId);
118+
const topLevelConversationKey = buildTopLevelSlackConversationKey(
119+
last.message,
120+
ctx.accountId,
121+
);
122+
if (flushedKey && topLevelConversationKey) {
123+
const pendingKeys = pendingTopLevelDebounceKeys.get(topLevelConversationKey);
124+
if (pendingKeys) {
125+
pendingKeys.delete(flushedKey);
126+
if (pendingKeys.size === 0) {
127+
pendingTopLevelDebounceKeys.delete(topLevelConversationKey);
128+
}
129+
}
157130
}
158-
}
159-
try {
160-
await dispatchPreparedSlackMessage(prepared);
161-
await recordSlackInboundMessageDeliveries({
162-
accountId: ctx.accountId,
163-
messages: entries.map((entry) => entry.message),
164-
});
165-
} catch (error) {
166-
if (!(error instanceof SlackRetryableInboundError)) {
167-
await recordSlackInboundMessageDeliveries({
168-
accountId: ctx.accountId,
169-
messages: entries.map((entry) => entry.message),
131+
const combinedText =
132+
entries.length === 1
133+
? (last.message.text ?? "")
134+
: entries
135+
.map((entry) => entry.message.text ?? "")
136+
.filter(Boolean)
137+
.join("\n");
138+
const combinedMentioned = entries.some((entry) => Boolean(entry.opts.wasMentioned));
139+
const syntheticMessage: SlackMessageEvent = {
140+
...last.message,
141+
text: combinedText,
142+
};
143+
const seenMessageKey = buildSeenMessageKey(last.message.channel, last.message.ts);
144+
try {
145+
const { prepareSlackMessage, dispatchPreparedSlackMessage } =
146+
await loadSlackMessagePipeline();
147+
const {
148+
dispatchCompletion: _completion,
149+
awaitDispatch: _awaitDispatch,
150+
...lastOpts
151+
} = last.opts;
152+
const prepared = await prepareSlackMessage({
153+
ctx,
154+
account,
155+
message: syntheticMessage,
156+
opts: {
157+
...lastOpts,
158+
wasMentioned: combinedMentioned || last.opts.wasMentioned,
159+
},
170160
});
161+
if (!prepared) {
162+
return;
163+
}
164+
if (seenMessageKey) {
165+
pruneAppMentionRetryKeys(Date.now());
166+
if (last.opts.source === "app_mention") {
167+
// If app_mention wins the race and dispatches first, drop the later message dispatch.
168+
rememberExpiringAppMentionKey(appMentionDispatchedKeys, seenMessageKey);
169+
} else if (
170+
last.opts.source === "message" &&
171+
appMentionDispatchedKeys.has(seenMessageKey)
172+
) {
173+
appMentionDispatchedKeys.delete(seenMessageKey);
174+
appMentionRetryKeys.delete(seenMessageKey);
175+
return;
176+
}
177+
appMentionRetryKeys.delete(seenMessageKey);
178+
}
179+
if (entries.length > 1) {
180+
const ids = entries.map((entry) => entry.message.ts).filter(Boolean) as string[];
181+
if (ids.length > 0) {
182+
prepared.ctxPayload.MessageSids = ids;
183+
prepared.ctxPayload.MessageSidFirst = ids[0];
184+
prepared.ctxPayload.MessageSidLast = ids[ids.length - 1];
185+
}
186+
}
187+
try {
188+
await dispatchPreparedSlackMessage(prepared);
189+
await recordSlackInboundMessageDeliveries({
190+
accountId: ctx.accountId,
191+
messages: entries.map((entry) => entry.message),
192+
});
193+
} catch (error) {
194+
if (!(error instanceof SlackRetryableInboundError)) {
195+
await recordSlackInboundMessageDeliveries({
196+
accountId: ctx.accountId,
197+
messages: entries.map((entry) => entry.message),
198+
});
199+
}
200+
throw error;
201+
}
202+
} catch (error) {
203+
if (error instanceof SlackRetryableInboundError) {
204+
if (seenMessageKey) {
205+
appMentionDispatchedKeys.delete(seenMessageKey);
206+
}
207+
ctx.releaseSeenMessage(last.message.channel, last.message.ts);
208+
}
209+
throw error;
171210
}
172-
throw error;
211+
})();
212+
for (const completion of completions) {
213+
completion.resolve();
173214
}
174215
} catch (error) {
175-
if (error instanceof SlackRetryableInboundError) {
176-
if (seenMessageKey) {
177-
appMentionDispatchedKeys.delete(seenMessageKey);
178-
}
179-
ctx.releaseSeenMessage(last.message.channel, last.message.ts);
216+
for (const completion of completions) {
217+
completion.reject(error);
180218
}
181219
throw error;
182220
}
@@ -293,6 +331,21 @@ export function createSlackMessageHandler(params: {
293331
pendingKeys.add(debounceKey);
294332
pendingTopLevelDebounceKeys.set(conversationKey, pendingKeys);
295333
}
296-
await debouncer.enqueue({ message: resolvedMessage, opts });
334+
const dispatchCompletion = opts.awaitDispatch ? createSlackDispatchCompletion() : undefined;
335+
await debouncer.enqueue({
336+
message: resolvedMessage,
337+
opts: {
338+
...opts,
339+
...(dispatchCompletion
340+
? {
341+
dispatchCompletion: {
342+
resolve: dispatchCompletion.resolve,
343+
reject: dispatchCompletion.reject,
344+
},
345+
}
346+
: {}),
347+
},
348+
});
349+
await dispatchCompletion?.promise;
297350
};
298351
}

0 commit comments

Comments
 (0)