Skip to content

Commit 47839d3

Browse files
Qinsammukhtharcm
Qinsam
and
mukhtharcm
authored
fix(mattermost): detect stale websocket after bot disable/enable cycle (#53604)
Merged via squash. Prepared head SHA: 818d437 Co-authored-by: Qinsam <[email protected]> Co-authored-by: mukhtharcm <[email protected]> Reviewed-by: @mukhtharcm
1 parent c6ded0f commit 47839d3

14 files changed

Lines changed: 339 additions & 27 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,6 +230,7 @@ Docs: https://docs.openclaw.ai
230230
- Control UI/gateway: reconnect the browser client when gateway event sequence gaps are detected, so stale non-chat state recovers automatically instead of only telling the user to refresh. (#23912) thanks @Olshansk.
231231
- ClawDock/docs: move the helper scripts to `scripts/clawdock`, publish ClawDock as a first-class docs page on the docs site, and document reinstalling local helper copies from the new raw GitHub path. (#23912) thanks @Olshansk.
232232
- Control UI/gateway: clear queued browser connect timeouts on client stop so aborted or replaced gateway clients do not send delayed connect requests after shutdown. (#57338) thanks @gumadeiras.
233+
- Mattermost: detect stale websocket sessions after bot disable/enable cycles by polling the bot account `update_at` and forcing a reconnect when it changes. (#53604) Thanks @Qinsam.
233234

234235
## 2026.3.24
235236

‎extensions/mattermost/src/mattermost/client.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ export type MattermostUser = {
1818
nickname?: string | null;
1919
first_name?: string | null;
2020
last_name?: string | null;
21+
update_at?: number;
2122
};
2223

2324
export type MattermostChannel = {

‎extensions/mattermost/src/mattermost/monitor-websocket.test.ts‎

Lines changed: 186 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,191 @@ describe("mattermost websocket monitor", () => {
227227
emoji_name: "thumbsup",
228228
}),
229229
);
230-
expect(payload.data?.reaction).toBeDefined();
230+
});
231+
232+
it("terminates when bot update_at changes (disable/enable cycle)", async () => {
233+
vi.useFakeTimers();
234+
const socket = new FakeWebSocket();
235+
const runtime = testRuntime();
236+
let updateAt = 1000;
237+
const connectOnce = createMattermostConnectOnce({
238+
wsUrl: "wss://example.invalid/api/v4/websocket",
239+
botToken: "token",
240+
runtime,
241+
nextSeq: () => 1,
242+
onPosted: async () => {},
243+
webSocketFactory: () => socket,
244+
getBotUpdateAt: async () => updateAt,
245+
healthCheckIntervalMs: 100,
246+
});
247+
248+
const connected = connectOnce();
249+
socket.emitOpen();
250+
251+
// Let initial getBotUpdateAt resolve
252+
await vi.advanceTimersByTimeAsync(0);
253+
254+
// update_at unchanged — no terminate
255+
await vi.advanceTimersByTimeAsync(100);
256+
expect(socket.terminateCalls).toBe(0);
257+
258+
// Simulate disable/enable — update_at changes
259+
updateAt = 2000;
260+
await vi.advanceTimersByTimeAsync(100);
261+
expect(socket.terminateCalls).toBe(1);
262+
expect(runtime.log).toHaveBeenCalledWith(
263+
"mattermost: bot account updated (update_at changed: 1000 → 2000) — reconnecting",
264+
);
265+
266+
socket.emitClose(1006);
267+
await connected;
268+
vi.useRealTimers();
269+
});
270+
271+
it("keeps connection alive when update_at stays the same", async () => {
272+
vi.useFakeTimers();
273+
const socket = new FakeWebSocket();
274+
const connectOnce = createMattermostConnectOnce({
275+
wsUrl: "wss://example.invalid/api/v4/websocket",
276+
botToken: "token",
277+
runtime: testRuntime(),
278+
nextSeq: () => 1,
279+
onPosted: async () => {},
280+
webSocketFactory: () => socket,
281+
getBotUpdateAt: async () => 1000,
282+
healthCheckIntervalMs: 100,
283+
});
284+
285+
const connected = connectOnce();
286+
socket.emitOpen();
287+
288+
await vi.advanceTimersByTimeAsync(0);
289+
await vi.advanceTimersByTimeAsync(300);
290+
expect(socket.terminateCalls).toBe(0);
291+
292+
socket.emitClose(1000);
293+
await connected;
294+
vi.useRealTimers();
295+
});
296+
297+
it("does not terminate when getBotUpdateAt throws", async () => {
298+
vi.useFakeTimers();
299+
const socket = new FakeWebSocket();
300+
const runtime = testRuntime();
301+
let shouldThrow = false;
302+
const connectOnce = createMattermostConnectOnce({
303+
wsUrl: "wss://example.invalid/api/v4/websocket",
304+
botToken: "token",
305+
runtime,
306+
nextSeq: () => 1,
307+
onPosted: async () => {},
308+
webSocketFactory: () => socket,
309+
getBotUpdateAt: async () => {
310+
if (shouldThrow) throw new Error("network error");
311+
return 1000;
312+
},
313+
healthCheckIntervalMs: 100,
314+
});
315+
316+
const connected = connectOnce();
317+
socket.emitOpen();
318+
319+
await vi.advanceTimersByTimeAsync(0);
320+
321+
// API error — should log but not terminate
322+
shouldThrow = true;
323+
await vi.advanceTimersByTimeAsync(100);
324+
expect(socket.terminateCalls).toBe(0);
325+
expect(runtime.error).toHaveBeenCalledWith(
326+
"mattermost: health check error: Error: network error",
327+
);
328+
329+
socket.emitClose(1000);
330+
await connected;
331+
vi.useRealTimers();
332+
});
333+
334+
it("keeps polling when the initial getBotUpdateAt call fails", async () => {
335+
vi.useFakeTimers();
336+
const socket = new FakeWebSocket();
337+
const runtime = testRuntime();
338+
const responses: Array<number | Error> = [new Error("network error"), 1000, 2000];
339+
const connectOnce = createMattermostConnectOnce({
340+
wsUrl: "wss://example.invalid/api/v4/websocket",
341+
botToken: "token",
342+
runtime,
343+
nextSeq: () => 1,
344+
onPosted: async () => {},
345+
webSocketFactory: () => socket,
346+
getBotUpdateAt: async () => {
347+
const next = responses.shift();
348+
if (next instanceof Error) {
349+
throw next;
350+
}
351+
return next ?? 2000;
352+
},
353+
healthCheckIntervalMs: 100,
354+
});
355+
356+
const connected = connectOnce();
357+
socket.emitOpen();
358+
359+
await vi.advanceTimersByTimeAsync(0);
360+
expect(runtime.error).toHaveBeenCalledWith(
361+
"mattermost: failed to get initial update_at: Error: network error",
362+
);
363+
364+
await vi.advanceTimersByTimeAsync(100);
365+
expect(socket.terminateCalls).toBe(0);
366+
367+
await vi.advanceTimersByTimeAsync(100);
368+
expect(socket.terminateCalls).toBe(1);
369+
expect(runtime.log).toHaveBeenCalledWith(
370+
"mattermost: bot account updated (update_at changed: 1000 → 2000) — reconnecting",
371+
);
372+
373+
socket.emitClose(1006);
374+
await connected;
375+
vi.useRealTimers();
376+
});
377+
378+
it("does not overlap health checks when a prior poll is still running", async () => {
379+
vi.useFakeTimers();
380+
const socket = new FakeWebSocket();
381+
const resolvers: Array<(value: number) => void> = [];
382+
let pollCount = 0;
383+
const connectOnce = createMattermostConnectOnce({
384+
wsUrl: "wss://example.invalid/api/v4/websocket",
385+
botToken: "token",
386+
runtime: testRuntime(),
387+
nextSeq: () => 1,
388+
onPosted: async () => {},
389+
webSocketFactory: () => socket,
390+
getBotUpdateAt: async () => {
391+
pollCount++;
392+
return await new Promise<number>((resolve) => {
393+
resolvers.push(resolve);
394+
});
395+
},
396+
healthCheckIntervalMs: 100,
397+
});
398+
399+
const connected = connectOnce();
400+
socket.emitOpen();
401+
402+
await vi.advanceTimersByTimeAsync(0);
403+
expect(pollCount).toBe(1);
404+
405+
await vi.advanceTimersByTimeAsync(300);
406+
expect(pollCount).toBe(1);
407+
408+
resolvers[0]?.(1000);
409+
await vi.advanceTimersByTimeAsync(0);
410+
await vi.advanceTimersByTimeAsync(100);
411+
expect(pollCount).toBe(2);
412+
413+
socket.emitClose(1000);
414+
await connected;
415+
vi.useRealTimers();
231416
});
232417
});

‎extensions/mattermost/src/mattermost/monitor-websocket.ts‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,15 @@ type CreateMattermostConnectOnceOpts = {
8989
onPosted: (post: MattermostPost, payload: MattermostEventPayload) => Promise<void>;
9090
onReaction?: (payload: MattermostEventPayload) => Promise<void>;
9191
webSocketFactory?: MattermostWebSocketFactory;
92+
/**
93+
* Called periodically to check whether the bot account has been modified
94+
* (e.g. disabled then re-enabled) since the WebSocket was opened.
95+
* Returns the bot's current `update_at` timestamp. When it differs from
96+
* the value recorded at connect time, the connection is terminated so the
97+
* reconnect loop can establish a fresh one.
98+
*/
99+
getBotUpdateAt?: () => Promise<number>;
100+
healthCheckIntervalMs?: number;
92101
};
93102

94103
export const defaultMattermostWebSocketFactory: MattermostWebSocketFactory = (url) =>
@@ -126,27 +135,94 @@ export function createMattermostConnectOnce(
126135
opts: CreateMattermostConnectOnceOpts,
127136
): () => Promise<void> {
128137
const webSocketFactory = opts.webSocketFactory ?? defaultMattermostWebSocketFactory;
138+
const healthCheckIntervalMs = opts.healthCheckIntervalMs ?? 30_000;
129139
return async () => {
130140
const ws = webSocketFactory(opts.wsUrl);
131141
const onAbort = () => ws.terminate();
132142
opts.abortSignal?.addEventListener("abort", onAbort, { once: true });
143+
const getBotUpdateAt = opts.getBotUpdateAt;
133144

134145
try {
135146
return await new Promise<void>((resolve, reject) => {
136147
let opened = false;
137148
let settled = false;
149+
let healthCheckEnabled = getBotUpdateAt != null;
150+
let healthCheckInFlight = false;
151+
let healthCheckTimer: ReturnType<typeof setTimeout> | undefined;
152+
let initialUpdateAt: number | undefined;
153+
154+
const clearTimers = () => {
155+
if (healthCheckTimer !== undefined) {
156+
clearTimeout(healthCheckTimer);
157+
healthCheckTimer = undefined;
158+
}
159+
};
160+
161+
const stopHealthChecks = () => {
162+
healthCheckEnabled = false;
163+
clearTimers();
164+
};
165+
166+
const scheduleHealthCheck = () => {
167+
if (!getBotUpdateAt || !healthCheckEnabled || settled || healthCheckInFlight) {
168+
return;
169+
}
170+
healthCheckTimer = setTimeout(() => {
171+
healthCheckTimer = undefined;
172+
void runHealthCheck();
173+
}, healthCheckIntervalMs);
174+
};
175+
176+
const runHealthCheck = async () => {
177+
if (!getBotUpdateAt || !healthCheckEnabled || settled || healthCheckInFlight) {
178+
return;
179+
}
180+
healthCheckInFlight = true;
181+
try {
182+
const current = await getBotUpdateAt();
183+
if (!healthCheckEnabled || settled) {
184+
return;
185+
}
186+
if (initialUpdateAt === undefined) {
187+
initialUpdateAt = current;
188+
return;
189+
}
190+
if (current !== initialUpdateAt) {
191+
opts.runtime.log?.(
192+
`mattermost: bot account updated (update_at changed: ${initialUpdateAt} → ${current}) — reconnecting`,
193+
);
194+
stopHealthChecks();
195+
ws.terminate();
196+
}
197+
} catch (err) {
198+
if (!healthCheckEnabled || settled) {
199+
return;
200+
}
201+
const label =
202+
initialUpdateAt === undefined
203+
? "mattermost: failed to get initial update_at"
204+
: "mattermost: health check error";
205+
opts.runtime.error?.(`${label}: ${String(err)}`);
206+
} finally {
207+
healthCheckInFlight = false;
208+
scheduleHealthCheck();
209+
}
210+
};
211+
138212
const resolveOnce = () => {
139213
if (settled) {
140214
return;
141215
}
142216
settled = true;
217+
stopHealthChecks();
143218
resolve();
144219
};
145220
const rejectOnce = (error: Error) => {
146221
if (settled) {
147222
return;
148223
}
149224
settled = true;
225+
stopHealthChecks();
150226
reject(error);
151227
};
152228

@@ -164,6 +240,15 @@ export function createMattermostConnectOnce(
164240
data: { token: opts.botToken },
165241
}),
166242
);
243+
244+
// Periodically check if the bot account was modified (e.g. disable/enable).
245+
// After such a cycle the WebSocket silently stops delivering events even
246+
// though the connection itself stays alive. Comparing update_at detects
247+
// this reliably regardless of how quickly the cycle happens.
248+
if (getBotUpdateAt) {
249+
// Use a recursive timeout so only one REST poll can be in flight at a time.
250+
void runHealthCheck();
251+
}
167252
});
168253

169254
ws.on("message", async (data) => {
@@ -200,6 +285,7 @@ export function createMattermostConnectOnce(
200285
});
201286

202287
ws.on("close", (code, reason) => {
288+
stopHealthChecks();
203289
const message = reasonToString(reason);
204290
opts.statusSink?.({
205291
connected: false,

‎extensions/mattermost/src/mattermost/monitor.ts‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -275,7 +275,33 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {}
275275
botToken,
276276
allowPrivateNetwork: account.config?.allowPrivateNetwork === true,
277277
});
278-
const botUser = await fetchMattermostMe(client);
278+
279+
// Wait for the Mattermost API to accept our bot token before proceeding.
280+
// When a bot account is disabled and re-enabled, the session is invalidated
281+
// and API calls return 401 until the account is fully active again. Retrying
282+
// here (with exponential backoff) keeps the monitor alive and prevents the
283+
// framework's auto-restart budget from being exhausted.
284+
let botUser!: MattermostUser;
285+
await runWithReconnect(
286+
async () => {
287+
botUser = await fetchMattermostMe(client);
288+
},
289+
{
290+
abortSignal: opts.abortSignal,
291+
jitterRatio: 0.2,
292+
shouldReconnect: ({ outcome }) => outcome === "rejected",
293+
onError: (err) => {
294+
runtime.error?.(`mattermost: API auth failed: ${String(err)}`);
295+
opts.statusSink?.({ lastError: String(err), connected: false });
296+
},
297+
onReconnect: (delayMs) => {
298+
runtime.log?.(`mattermost: API not accessible, retrying in ${Math.round(delayMs / 1000)}s`);
299+
},
300+
},
301+
);
302+
if (opts.abortSignal?.aborted) {
303+
return;
304+
}
279305
const botUserId = botUser.id;
280306
const botUsername = botUser.username?.trim() || undefined;
281307
runtime.log?.(`mattermost connected as ${botUsername ? `@${botUsername}` : botUserId}`);
@@ -1645,6 +1671,10 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {}
16451671
runtime,
16461672
webSocketFactory: opts.webSocketFactory,
16471673
nextSeq: () => seq++,
1674+
getBotUpdateAt: async () => {
1675+
const me = await fetchMattermostMe(client);
1676+
return me.update_at ?? 0;
1677+
},
16481678
onPosted: async (post, payload) => {
16491679
await debouncer.enqueue({ post, payload });
16501680
},

0 commit comments

Comments
 (0)