Skip to content

Commit fbdfd19

Browse files
steipete-oaisteipete
steipete-oai
and
steipete
authored
Fix Slack retry for session init conflicts (#99647)
* Fix Slack retry for session init conflicts * fix(slack): separate native and relay retries * style(slack): make completion exits explicit --------- Co-authored-by: Peter Steinberger <[email protected]>
1 parent 0cd936c commit fbdfd19

2 files changed

Lines changed: 223 additions & 14 deletions

File tree

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

Lines changed: 141 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ const flushKeyMock = vi.fn(async (_key: string) => {});
66
const onFlushCallbacks: Array<(entries: Array<Record<string, unknown>>) => Promise<void>> = [];
77
const prepareSlackMessageMock = vi.fn(async () => ({ ctxPayload: {} }));
88
const dispatchPreparedSlackMessageMock = vi.fn(async () => {});
9+
const hasSlackInboundMessageDeliveryMock = vi.fn(async () => false);
10+
const recordSlackInboundMessageDeliveriesMock = vi.fn(async () => {});
911
const resolveThreadTsMock = vi.fn(async ({ message }: { message: Record<string, unknown> }) => ({
1012
...message,
1113
}));
@@ -45,8 +47,8 @@ vi.mock("./message-handler/pipeline.runtime.js", () => ({
4547
}));
4648

4749
vi.mock("./inbound-delivery-state.js", () => ({
48-
hasSlackInboundMessageDelivery: vi.fn(async () => false),
49-
recordSlackInboundMessageDeliveries: vi.fn(async () => {}),
50+
hasSlackInboundMessageDelivery: hasSlackInboundMessageDeliveryMock,
51+
recordSlackInboundMessageDeliveries: recordSlackInboundMessageDeliveriesMock,
5052
}));
5153

5254
function createContext(overrides?: {
@@ -101,6 +103,9 @@ describe("createSlackMessageHandler", () => {
101103
onFlushCallbacks.length = 0;
102104
prepareSlackMessageMock.mockClear();
103105
dispatchPreparedSlackMessageMock.mockClear();
106+
hasSlackInboundMessageDeliveryMock.mockReset();
107+
hasSlackInboundMessageDeliveryMock.mockResolvedValue(false);
108+
recordSlackInboundMessageDeliveriesMock.mockClear();
104109
resolveThreadTsMock.mockClear();
105110
});
106111

@@ -270,4 +275,138 @@ describe("createSlackMessageHandler", () => {
270275
const flushFailure = expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow("dispatch failed");
271276
await Promise.all([handledFailure, flushFailure]);
272277
});
278+
279+
it("retries native session initialization conflicts through the delivery gates", async () => {
280+
const releaseSeenMessage = vi.fn();
281+
dispatchPreparedSlackMessageMock.mockRejectedValueOnce(
282+
new Error("Slack dispatch failed", {
283+
cause: new Error(
284+
"reply session initialization conflicted for agent:main:main:thread:123.456",
285+
),
286+
}),
287+
);
288+
const { handler } = createHandlerWithTracker({ releaseSeenMessage });
289+
await handler(
290+
{
291+
type: "message",
292+
channel: "C111",
293+
user: "U111",
294+
ts: "1709000000.000700",
295+
text: "native message",
296+
} as never,
297+
{ source: "message" },
298+
);
299+
300+
const entry = enqueueMock.mock.calls[0]?.[0] as Record<string, unknown>;
301+
vi.useFakeTimers();
302+
try {
303+
await expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow("Slack dispatch failed");
304+
await vi.advanceTimersByTimeAsync(1000);
305+
306+
expect(releaseSeenMessage).toHaveBeenCalledWith("C111", "1709000000.000700");
307+
expect(recordSlackInboundMessageDeliveriesMock).not.toHaveBeenCalled();
308+
expect(hasSlackInboundMessageDeliveryMock).toHaveBeenCalledTimes(2);
309+
expect(enqueueMock).toHaveBeenCalledTimes(2);
310+
expect(enqueueMock.mock.calls[1]?.[0]).toMatchObject({
311+
opts: {
312+
retryAttempt: 1,
313+
},
314+
});
315+
expect(enqueueMock.mock.calls[1]?.[0]).not.toHaveProperty("opts.dispatchCompletion");
316+
} finally {
317+
vi.useRealTimers();
318+
}
319+
});
320+
321+
it("leaves relay session conflict retries to unacknowledged redelivery", async () => {
322+
const releaseSeenMessage = vi.fn();
323+
dispatchPreparedSlackMessageMock.mockRejectedValueOnce(
324+
new Error("Slack dispatch failed", {
325+
cause: new Error(
326+
"reply session initialization conflicted for agent:main:main:thread:123.456",
327+
),
328+
}),
329+
);
330+
const { handler } = createHandlerWithTracker({ releaseSeenMessage });
331+
const handled = handler(
332+
{
333+
type: "message",
334+
channel: "C111",
335+
user: "U111",
336+
ts: "1709000000.000800",
337+
text: "relay message",
338+
} as never,
339+
{ source: "message", awaitDispatch: true },
340+
);
341+
342+
await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(1));
343+
const entry = enqueueMock.mock.calls[0]?.[0] as Record<string, unknown>;
344+
vi.useFakeTimers();
345+
try {
346+
const handledFailure = expect(handled).rejects.toThrow("Slack dispatch failed");
347+
const flushFailure = expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow(
348+
"Slack dispatch failed",
349+
);
350+
await Promise.all([handledFailure, flushFailure]);
351+
await vi.advanceTimersByTimeAsync(1000);
352+
353+
expect(releaseSeenMessage).toHaveBeenCalledWith("C111", "1709000000.000800");
354+
expect(recordSlackInboundMessageDeliveriesMock).not.toHaveBeenCalled();
355+
expect(enqueueMock).toHaveBeenCalledTimes(1);
356+
} finally {
357+
vi.useRealTimers();
358+
}
359+
});
360+
361+
it("settles an already-delivered relay event without enqueueing", async () => {
362+
hasSlackInboundMessageDeliveryMock.mockResolvedValueOnce(true);
363+
const { handler } = createHandlerWithTracker();
364+
365+
await expect(
366+
handler(
367+
{
368+
type: "message",
369+
channel: "C111",
370+
user: "U111",
371+
ts: "1709000000.000850",
372+
text: "relay replay",
373+
} as never,
374+
{ source: "message", awaitDispatch: true },
375+
),
376+
).resolves.toBeUndefined();
377+
378+
expect(enqueueMock).not.toHaveBeenCalled();
379+
});
380+
381+
it("skips a native retry when another delivery already succeeded", async () => {
382+
dispatchPreparedSlackMessageMock.mockRejectedValueOnce(
383+
new Error("reply session initialization conflicted for agent:main:main:thread:123.456"),
384+
);
385+
hasSlackInboundMessageDeliveryMock.mockResolvedValueOnce(false).mockResolvedValueOnce(true);
386+
const { handler } = createHandlerWithTracker();
387+
await handler(
388+
{
389+
type: "message",
390+
channel: "C111",
391+
user: "U111",
392+
ts: "1709000000.000900",
393+
text: "native message",
394+
} as never,
395+
{ source: "message" },
396+
);
397+
398+
const entry = enqueueMock.mock.calls[0]?.[0] as Record<string, unknown>;
399+
vi.useFakeTimers();
400+
try {
401+
await expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow(
402+
"reply session initialization conflicted",
403+
);
404+
await vi.advanceTimersByTimeAsync(1000);
405+
406+
expect(hasSlackInboundMessageDeliveryMock).toHaveBeenCalledTimes(2);
407+
expect(enqueueMock).toHaveBeenCalledTimes(1);
408+
} finally {
409+
vi.useRealTimers();
410+
}
411+
});
273412
});

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

Lines changed: 82 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import {
33
createChannelInboundDebouncer,
44
shouldDebounceTextInbound,
55
} from "openclaw/plugin-sdk/channel-inbound";
6-
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
6+
import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
77
import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime";
88
import {
99
asDateTimestampMs,
@@ -45,7 +45,11 @@ type SlackDispatchCompletion = {
4545
reject: (error: unknown) => void;
4646
};
4747

48-
type QueuedSlackMessageOptions = Parameters<SlackMessageHandler>[1] & {
48+
type IngressSlackMessageOptions = Parameters<SlackMessageHandler>[1] & {
49+
retryAttempt?: number;
50+
};
51+
52+
type QueuedSlackMessageOptions = IngressSlackMessageOptions & {
4953
dispatchCompletion?: Omit<SlackDispatchCompletion, "promise">;
5054
};
5155

@@ -60,6 +64,9 @@ function createSlackDispatchCompletion(): SlackDispatchCompletion {
6064
}
6165

6266
const APP_MENTION_RETRY_TTL_MS = 60_000;
67+
const RETRYABLE_FLUSH_MAX_ATTEMPTS = 3;
68+
const RETRYABLE_FLUSH_RETRY_DELAY_MS = 1_000;
69+
const REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE = /reply session initialization conflicted for \S+/u;
6370

6471
export class SlackRetryableInboundError extends Error {
6572
constructor(message: string, options?: ErrorOptions) {
@@ -68,6 +75,15 @@ export class SlackRetryableInboundError extends Error {
6875
}
6976
}
7077

78+
function isRetryableSlackInboundError(error: unknown): boolean {
79+
if (error instanceof SlackRetryableInboundError) {
80+
return true;
81+
}
82+
return collectErrorGraphCandidates(error, (current) => [current.cause, current.error]).some(
83+
(candidate) => REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE.test(formatErrorMessage(candidate)),
84+
);
85+
}
86+
7187
function shouldDebounceSlackMessage(message: SlackMessageEvent, cfg: SlackMonitorContext["cfg"]) {
7288
const text = message.text ?? "";
7389
const textForCommandDetection = stripSlackMentionsForCommandDetection(text);
@@ -101,6 +117,46 @@ export function createSlackMessageHandler(params: {
101117
buildKey: (entry) => buildSlackDebounceKey(entry.message, ctx.accountId),
102118
shouldDebounce: (entry) => shouldDebounceSlackMessage(entry.message, ctx.cfg),
103119
onFlush: async (entries) => {
120+
const retryEntries = (sourceError: unknown): boolean => {
121+
if (!isRetryableSlackInboundError(sourceError)) {
122+
return false;
123+
}
124+
const nextEntries = entries
125+
.map((entry) => {
126+
// Relay delivery owns retry until its dispatch completion is acknowledged.
127+
// Scheduling here as well can race the router redelivery and duplicate a reply.
128+
if (entry.opts.dispatchCompletion) {
129+
return null;
130+
}
131+
const retryAttempt = entry.opts.retryAttempt ?? 0;
132+
if (retryAttempt >= RETRYABLE_FLUSH_MAX_ATTEMPTS) {
133+
return null;
134+
}
135+
const { dispatchCompletion: _dispatchCompletion, ...retryOpts } = entry.opts;
136+
return {
137+
...entry,
138+
opts: {
139+
...retryOpts,
140+
retryAttempt: retryAttempt + 1,
141+
},
142+
};
143+
})
144+
.filter((entry) => entry !== null);
145+
if (nextEntries.length === 0) {
146+
return false;
147+
}
148+
const retryTimer = setTimeout(() => {
149+
for (const entry of nextEntries) {
150+
// Re-enter ingress so a relay replay or another successful attempt wins
151+
// through the normal delivery and seen-message gates before dispatch.
152+
void enqueueSlackMessage(entry.message, entry.opts).catch((err: unknown) => {
153+
ctx.runtime.error?.(`slack inbound retry enqueue failed: ${formatErrorMessage(err)}`);
154+
});
155+
}
156+
}, RETRYABLE_FLUSH_RETRY_DELAY_MS);
157+
retryTimer.unref?.();
158+
return true;
159+
};
104160
const completions = entries
105161
.map((entry) => entry.opts.dispatchCompletion)
106162
.filter((completion) => completion !== undefined);
@@ -187,7 +243,7 @@ export function createSlackMessageHandler(params: {
187243
messages: entries.map((entry) => entry.message),
188244
});
189245
} catch (error) {
190-
if (!(error instanceof SlackRetryableInboundError)) {
246+
if (!isRetryableSlackInboundError(error)) {
191247
await recordSlackInboundMessageDeliveries({
192248
accountId: ctx.accountId,
193249
messages: entries.map((entry) => entry.message),
@@ -196,11 +252,16 @@ export function createSlackMessageHandler(params: {
196252
throw error;
197253
}
198254
} catch (error) {
199-
if (error instanceof SlackRetryableInboundError) {
200-
if (seenMessageKey) {
201-
appMentionDispatchedKeys.delete(seenMessageKey);
255+
if (isRetryableSlackInboundError(error)) {
256+
// Every buffered event passed the seen gate before this combined dispatch.
257+
// Release all of them so the retry can rebuild the same batch.
258+
for (const entry of entries) {
259+
const entrySeenKey = buildSeenMessageKey(entry.message.channel, entry.message.ts);
260+
if (entrySeenKey) {
261+
appMentionDispatchedKeys.delete(entrySeenKey);
262+
}
263+
ctx.releaseSeenMessage(entry.message.channel, entry.message.ts);
202264
}
203-
ctx.releaseSeenMessage(last.message.channel, last.message.ts);
204265
}
205266
throw error;
206267
}
@@ -209,6 +270,7 @@ export function createSlackMessageHandler(params: {
209270
completion.resolve();
210271
}
211272
} catch (error) {
273+
retryEntries(error);
212274
for (const completion of completions) {
213275
completion.reject(error);
214276
}
@@ -271,9 +333,12 @@ export function createSlackMessageHandler(params: {
271333
return true;
272334
};
273335

274-
return async (message, opts) => {
336+
async function enqueueSlackMessage(
337+
message: SlackMessageEvent,
338+
opts: IngressSlackMessageOptions,
339+
): Promise<SlackDispatchCompletion | undefined> {
275340
if (opts.source === "message" && message.type !== "message") {
276-
return;
341+
return undefined;
277342
}
278343
if (
279344
opts.source === "message" &&
@@ -282,7 +347,7 @@ export function createSlackMessageHandler(params: {
282347
message.subtype !== "bot_message" &&
283348
message.subtype !== "thread_broadcast"
284349
) {
285-
return;
350+
return undefined;
286351
}
287352
const seenMessageKey = buildSeenMessageKey(message.channel, message.ts);
288353
if (
@@ -293,7 +358,7 @@ export function createSlackMessageHandler(params: {
293358
ts: message.ts,
294359
}))
295360
) {
296-
return;
361+
return undefined;
297362
}
298363
const wasSeen = seenMessageKey ? ctx.markMessageSeen(message.channel, message.ts) : false;
299364
if (seenMessageKey && opts.source === "message" && !wasSeen) {
@@ -305,7 +370,7 @@ export function createSlackMessageHandler(params: {
305370
// Allow exactly one app_mention retry if the same ts was previously dropped
306371
// from the message stream before it reached dispatch.
307372
if (opts.source !== "app_mention" || !consumeAppMentionRetryKey(seenMessageKey)) {
308-
return;
373+
return undefined;
309374
}
310375
}
311376
trackEvent?.();
@@ -342,6 +407,11 @@ export function createSlackMessageHandler(params: {
342407
: {}),
343408
},
344409
});
410+
return dispatchCompletion;
411+
}
412+
413+
return async (message, opts) => {
414+
const dispatchCompletion = await enqueueSlackMessage(message, opts);
345415
await dispatchCompletion?.promise;
346416
};
347417
}

0 commit comments

Comments
 (0)