Skip to content

Commit abbd5ae

Browse files
steipetesteipete
authored
fix(discord): recover from failed gateway resumes (#103596)
* fix(discord): recover from failed gateway resumes * chore: leave release notes to release workflow
1 parent f35b6e5 commit abbd5ae

2 files changed

Lines changed: 190 additions & 22 deletions

File tree

‎extensions/discord/src/internal/gateway.test.ts‎

Lines changed: 122 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -70,14 +70,16 @@ class FakeSocket extends EventEmitter {
7070
class TestGatewayPlugin extends GatewayPlugin {
7171
sockets: FakeSocket[] = [];
7272
connectCalls: boolean[] = [];
73+
urls: string[] = [];
7374

7475
override connect(resume = false): void {
7576
this.connectCalls.push(resume);
7677
super.connect(resume);
7778
}
7879

79-
protected override createWebSocket(): never {
80+
protected override createWebSocket(url: string): never {
8081
const socket = new FakeSocket();
82+
this.urls.push(url);
8183
this.sockets.push(socket);
8284
return socket as never;
8385
}
@@ -211,7 +213,7 @@ describe("GatewayPlugin", () => {
211213
originalSocket?.emit("close", 1006);
212214

213215
await vi.advanceTimersByTimeAsync(2_000);
214-
expect(gateway.connectCalls).toEqual([false, true]);
216+
expect(gateway.connectCalls).toEqual([false, false]);
215217
const replacementSocket = gateway.sockets[1];
216218
replacementSocket?.emit("open");
217219

@@ -347,6 +349,7 @@ describe("GatewayPlugin", () => {
347349
};
348350
gateway.isConnected = false;
349351
(gateway as unknown as { reconnectAttempts: number }).reconnectAttempts = 7;
352+
(gateway as unknown as { consecutiveResumeFailures: number }).consecutiveResumeFailures = 2;
350353

351354
await (
352355
gateway as unknown as {
@@ -359,6 +362,9 @@ describe("GatewayPlugin", () => {
359362

360363
expect(gateway.isConnected).toBe(true);
361364
expect((gateway as unknown as { reconnectAttempts: number }).reconnectAttempts).toBe(0);
365+
expect(
366+
(gateway as unknown as { consecutiveResumeFailures: number }).consecutiveResumeFailures,
367+
).toBe(0);
362368
});
363369

364370
it("queues outbound gateway events when the connection window is exhausted", () => {
@@ -462,22 +468,133 @@ describe("GatewayPlugin", () => {
462468
clearInterval(heartbeat);
463469
});
464470

465-
it("reconnects after active remote normal closes", async () => {
471+
it("logs and re-identifies after a resumable close without session state", async () => {
466472
vi.useFakeTimers();
467473
const gateway = new TestGatewayPlugin({
468474
autoInteractions: false,
469475
url: "wss://gateway.example.test",
470476
});
477+
const debugSpy = vi.fn();
478+
gateway.emitter.on("debug", debugSpy);
471479

472480
gateway.connect(false);
473481
gateway.sockets[0]?.emit("open");
474482
gateway.sockets[0]?.emit("close", 1000);
475483

476484
expect(gateway.sockets).toHaveLength(1);
485+
expect(debugSpy).toHaveBeenCalledWith(
486+
"Gateway reconnect scheduled in 2000ms (close, resume=false)",
487+
);
477488
await vi.advanceTimersByTimeAsync(2_000);
478489

479-
expect(gateway.connectCalls).toEqual([false, true]);
490+
expect(gateway.connectCalls).toEqual([false, false]);
480491
expect(gateway.sockets).toHaveLength(2);
492+
const reconnectSocket = gateway.sockets[1];
493+
reconnectSocket?.emit("open");
494+
reconnectSocket?.emit(
495+
"message",
496+
JSON.stringify({
497+
op: GatewayOpcodes.Hello,
498+
d: { heartbeat_interval: 45_000 },
499+
s: null,
500+
}),
501+
);
502+
await vi.advanceTimersByTimeAsync(0);
503+
expect(sentGatewayOpcodes(reconnectSocket?.send ?? vi.fn())).toContain(GatewayOpcodes.Identify);
504+
expect(sentGatewayOpcodes(reconnectSocket?.send ?? vi.fn())).not.toContain(
505+
GatewayOpcodes.Resume,
506+
);
507+
});
508+
509+
it("falls back to a fresh IDENTIFY after three failed resume attempts", async () => {
510+
vi.useFakeTimers();
511+
vi.setSystemTime(0);
512+
const gateway = new TestGatewayPlugin({
513+
autoInteractions: false,
514+
url: "wss://gateway.example.test",
515+
});
516+
const debugSpy = vi.fn();
517+
gateway.emitter.on("debug", debugSpy);
518+
(gateway as unknown as { client: unknown }).client = {
519+
options: { token: "token" },
520+
dispatchGatewayEvent: vi.fn(async () => {}),
521+
};
522+
523+
gateway.connect(false);
524+
const initialSocket = gateway.sockets[0];
525+
initialSocket?.emit("open");
526+
initialSocket?.emit(
527+
"message",
528+
JSON.stringify({
529+
op: GatewayOpcodes.Hello,
530+
d: { heartbeat_interval: 45_000 },
531+
s: null,
532+
}),
533+
);
534+
await vi.advanceTimersByTimeAsync(0);
535+
expect(sentGatewayOpcodes(initialSocket?.send ?? vi.fn())).toContain(GatewayOpcodes.Identify);
536+
initialSocket?.emit(
537+
"message",
538+
JSON.stringify({
539+
op: GatewayOpcodes.Dispatch,
540+
t: GatewayDispatchEvents.Ready,
541+
s: 42,
542+
d: {
543+
session_id: "session-1",
544+
resume_gateway_url: "wss://resume.example.test",
545+
},
546+
}),
547+
);
548+
await vi.advanceTimersByTimeAsync(0);
549+
550+
for (const delayMs of [2_000, 4_000, 8_000]) {
551+
gateway.sockets.at(-1)?.emit("close", 1006);
552+
await vi.advanceTimersByTimeAsync(delayMs);
553+
const resumeSocket = gateway.sockets.at(-1);
554+
expect(gateway.urls.at(-1)).toMatch(/^wss:\/\/resume\.example\.test\//);
555+
resumeSocket?.emit("open");
556+
resumeSocket?.emit(
557+
"message",
558+
JSON.stringify({
559+
op: GatewayOpcodes.Hello,
560+
d: { heartbeat_interval: 45_000 },
561+
s: null,
562+
}),
563+
);
564+
expect(sentGatewayOpcodes(resumeSocket?.send ?? vi.fn())).toContain(GatewayOpcodes.Resume);
565+
expect(debugSpy).toHaveBeenCalledWith(
566+
`Gateway reconnect scheduled in ${delayMs}ms (close, resume=true)`,
567+
);
568+
}
569+
570+
gateway.sockets.at(-1)?.emit("close", 1006);
571+
expect(debugSpy).toHaveBeenCalledWith(
572+
"Gateway forcing fresh IDENTIFY after 3 failed resume attempts",
573+
);
574+
expect(debugSpy).toHaveBeenCalledWith(
575+
"Gateway reconnect scheduled in 16000ms (close, resume=false)",
576+
);
577+
578+
await vi.advanceTimersByTimeAsync(16_000);
579+
const freshSocket = gateway.sockets.at(-1);
580+
expect(gateway.urls.at(-1)).toMatch(/^wss:\/\/gateway\.example\.test\//);
581+
freshSocket?.emit("open");
582+
freshSocket?.emit(
583+
"message",
584+
JSON.stringify({
585+
op: GatewayOpcodes.Hello,
586+
d: { heartbeat_interval: 45_000 },
587+
s: null,
588+
}),
589+
);
590+
await vi.advanceTimersByTimeAsync(0);
591+
592+
expect(sentGatewayOpcodes(freshSocket?.send ?? vi.fn())).toContain(GatewayOpcodes.Identify);
593+
expect(sentGatewayOpcodes(freshSocket?.send ?? vi.fn())).not.toContain(GatewayOpcodes.Resume);
594+
const sessionState = gatewaySessionState(gateway);
595+
expect(sessionState.sessionId).toBeNull();
596+
expect(sessionState.resumeGatewayUrl).toBeNull();
597+
expect(sessionState.sequence).toBeNull();
481598
});
482599

483600
it.each([GatewayCloseCodes.InvalidSeq, GatewayCloseCodes.AlreadyAuthenticated])(
@@ -546,7 +663,7 @@ describe("GatewayPlugin", () => {
546663
expect(gateway.connectCalls).toEqual([false]);
547664

548665
await vi.advanceTimersByTimeAsync(1);
549-
expect(gateway.connectCalls).toEqual([false, true]);
666+
expect(gateway.connectCalls).toEqual([false, false]);
550667
});
551668

552669
it("includes close code details when reconnect attempts are exhausted", async () => {

‎extensions/discord/src/internal/gateway.ts‎

Lines changed: 68 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,18 @@ type GatewayPluginOptions = {
4343
shard?: [number, number];
4444
url?: string;
4545
};
46+
type GatewayReconnectReason =
47+
| "close"
48+
| "identify"
49+
| "invalid-session"
50+
| "reconnect-opcode"
51+
| "zombie";
52+
type GatewayReconnectOptions = {
53+
reason: GatewayReconnectReason;
54+
preferResume: boolean;
55+
closeCode?: number;
56+
minDelayMs?: number;
57+
};
4658

4759
const READY_STATE_OPEN = 1;
4860
const DEFAULT_GATEWAY_URL = "wss://gateway.discord.gg/";
@@ -54,6 +66,7 @@ export const DISCORD_GATEWAY_WS_CLIENT_OPTIONS = Object.freeze({
5466
}) satisfies ws.ClientOptions;
5567
const INVALID_SESSION_MIN_DELAY_MS = 1_000;
5668
const INVALID_SESSION_JITTER_MS = 4_000;
69+
const RESUME_FAILURE_THRESHOLD = 3;
5770

5871
function ensureGatewayParams(url: string): string {
5972
const parsed = new URL(url);
@@ -92,6 +105,7 @@ export class GatewayPlugin extends Plugin {
92105
private sessionId: string | null = null;
93106
private resumeGatewayUrl: string | null = null;
94107
private reconnectAttempts = 0;
108+
private consecutiveResumeFailures = 0;
95109
private shouldReconnect = false;
96110
private isConnecting = false;
97111
private readonly heartbeatTimers = new GatewayHeartbeatTimers();
@@ -173,6 +187,7 @@ export class GatewayPlugin extends Plugin {
173187
this.isConnecting = false;
174188
this.isConnected = false;
175189
this.reconnectAttempts = 0;
190+
this.consecutiveResumeFailures = 0;
176191
}
177192

178193
protected createWebSocket(url: string): ws.WebSocket {
@@ -224,7 +239,11 @@ export class GatewayPlugin extends Plugin {
224239
if (!canResume) {
225240
this.resetSessionState();
226241
}
227-
this.scheduleReconnect(canResume, closeCode);
242+
this.scheduleReconnect({
243+
reason: "close",
244+
preferResume: canResume,
245+
closeCode,
246+
});
228247
});
229248
socket.on("error", (error) => {
230249
if (socket !== this.ws) {
@@ -243,18 +262,19 @@ export class GatewayPlugin extends Plugin {
243262
this.sequence = payload.s;
244263
}
245264
switch (payload.op) {
246-
case GatewayOpcodes.Hello:
265+
case GatewayOpcodes.Hello: {
247266
this.startHeartbeat(
248267
(payload.d as { heartbeat_interval?: number }).heartbeat_interval ?? 45_000,
249268
);
250-
if (resume && this.sessionId) {
269+
const resumeState = resume ? this.getResumeState() : null;
270+
if (resumeState) {
251271
this.send(
252272
{
253273
op: GatewayOpcodes.Resume,
254274
d: {
255275
token: this.client?.options.token ?? "",
256-
session_id: this.sessionId,
257-
seq: this.sequence ?? 0,
276+
session_id: resumeState.sessionId,
277+
seq: resumeState.sequence,
258278
},
259279
} as GatewaySendPayload,
260280
true,
@@ -268,6 +288,7 @@ export class GatewayPlugin extends Plugin {
268288
});
269289
}
270290
break;
291+
}
271292
case GatewayOpcodes.HeartbeatAck:
272293
this.lastHeartbeatAck = true;
273294
break;
@@ -286,14 +307,15 @@ export class GatewayPlugin extends Plugin {
286307
if (!payload.d) {
287308
this.resetSessionState();
288309
}
289-
this.scheduleReconnect(
290-
payload.d,
291-
undefined,
292-
INVALID_SESSION_MIN_DELAY_MS + Math.floor(Math.random() * INVALID_SESSION_JITTER_MS),
293-
);
310+
this.scheduleReconnect({
311+
reason: "invalid-session",
312+
preferResume: payload.d,
313+
minDelayMs:
314+
INVALID_SESSION_MIN_DELAY_MS + Math.floor(Math.random() * INVALID_SESSION_JITTER_MS),
315+
});
294316
break;
295317
case GatewayOpcodes.Reconnect:
296-
this.scheduleReconnect(true);
318+
this.scheduleReconnect({ reason: "reconnect-opcode", preferResume: true });
297319
break;
298320
}
299321
}
@@ -305,7 +327,7 @@ export class GatewayPlugin extends Plugin {
305327
onHeartbeat: () => this.sendHeartbeat(),
306328
onAckTimeout: () => {
307329
this.emitter.emit("error", new Error("Gateway heartbeat ACK timeout"));
308-
this.scheduleReconnect(true);
330+
this.scheduleReconnect({ reason: "zombie", preferResume: true });
309331
},
310332
});
311333
}
@@ -351,7 +373,7 @@ export class GatewayPlugin extends Plugin {
351373
return;
352374
}
353375
if (socket.readyState !== READY_STATE_OPEN) {
354-
this.scheduleReconnect(false);
376+
this.scheduleReconnect({ reason: "identify", preferResume: false });
355377
return;
356378
}
357379
this.identify();
@@ -390,10 +412,12 @@ export class GatewayPlugin extends Plugin {
390412
this.sessionId = ready.session_id ?? null;
391413
this.resumeGatewayUrl = ready.resume_gateway_url ?? null;
392414
this.reconnectAttempts = 0;
415+
this.consecutiveResumeFailures = 0;
393416
this.isConnected = true;
394417
}
395418
if (payload.t === GatewayDispatchEvents.Resumed) {
396419
this.reconnectAttempts = 0;
420+
this.consecutiveResumeFailures = 0;
397421
this.isConnected = true;
398422
}
399423
dispatchVoiceGatewayEvent(this.client, payload.t, payload.d);
@@ -408,9 +432,16 @@ export class GatewayPlugin extends Plugin {
408432
this.sessionId = null;
409433
this.resumeGatewayUrl = null;
410434
this.sequence = null;
435+
this.consecutiveResumeFailures = 0;
436+
}
437+
438+
private getResumeState(): { sessionId: string; sequence: number } | null {
439+
return this.sessionId && this.sequence !== null
440+
? { sessionId: this.sessionId, sequence: this.sequence }
441+
: null;
411442
}
412443

413-
private scheduleReconnect(resume: boolean, closeCode?: number, minDelayMs = 0): void {
444+
private scheduleReconnect(options: GatewayReconnectOptions): void {
414445
if (!this.shouldReconnect) {
415446
return;
416447
}
@@ -427,17 +458,37 @@ export class GatewayPlugin extends Plugin {
427458
this.emitter.emit(
428459
"error",
429460
new Error(
430-
`Max reconnect attempts (${maxAttempts}) reached${closeCode !== undefined ? ` after close code ${closeCode}` : ""}`,
461+
`Max reconnect attempts (${maxAttempts}) reached${options.closeCode !== undefined ? ` after close code ${options.closeCode}` : ""}`,
431462
),
432463
);
433464
return;
434465
}
466+
let shouldResume = options.preferResume && this.getResumeState() !== null;
467+
// Abnormal closes can leave a cached session permanently rejected. READY or RESUMED
468+
// resets this streak; after the threshold, discard the poisoned session and IDENTIFY.
469+
if (shouldResume && this.consecutiveResumeFailures >= RESUME_FAILURE_THRESHOLD) {
470+
this.resetSessionState();
471+
shouldResume = false;
472+
this.emitter.emit(
473+
"debug",
474+
`Gateway forcing fresh IDENTIFY after ${RESUME_FAILURE_THRESHOLD} failed resume attempts`,
475+
);
476+
}
477+
if (shouldResume) {
478+
this.consecutiveResumeFailures += 1;
479+
} else {
480+
this.consecutiveResumeFailures = 0;
481+
}
435482
const delay = Math.max(
436-
minDelayMs,
483+
options.minDelayMs ?? 0,
437484
Math.min(30_000, 1_000 * 2 ** Math.min(this.reconnectAttempts, 5)),
438485
);
486+
this.emitter.emit(
487+
"debug",
488+
`Gateway reconnect scheduled in ${delay}ms (${options.reason}, resume=${String(shouldResume)})`,
489+
);
439490
this.reconnectTimer.schedule(delay, () => {
440-
this.connect(resume);
491+
this.connect(shouldResume);
441492
});
442493
}
443494

0 commit comments

Comments
 (0)