fix(discord): Even more reconnect handling improvements · foxxmd.dev/multi-scrobbler@85e4555

+118 -103

2 changed files

Expand all

src

backend

common

vendor

discord

DiscordWSClient.ts

utils

NetworkUtils.ts

+103 -103

src/backend/common/vendor/discord/DiscordWSClient.ts

··· 19 19 import { urlToMusicService } from "../ListenbrainzApiClient.js"; 20 20 import { urlContainsKnownMediaDomain } from "../../../utils/RequestUtils.js"; 21 21 import { CoverArtApiClient } from "../musicbrainz/CoverArtApiClient.js"; 22 - import { formatWebsocketClose, isCloseEvent, isErrorEvent } from "../../../utils/NetworkUtils.js"; 22 + import { formatWebsocketClose, isCloseEvent, isErrorEvent, wsReadyStateToStr } from "../../../utils/NetworkUtils.js"; 23 23 24 24 const ARTWORK_PLACEHOLDER = 'https://raw.githubusercontent.com/FoxxMD/multi-scrobbler/master/assets/default-artwork.png'; 25 25 const MS_ART = 'https://raw.githubusercontent.com/FoxxMD/multi-scrobbler/master/assets/icon.png'; ··· 40 40 41 41 declare config: DiscordStrongData; 42 42 43 - heartbeatInterval: NodeJS.Timeout 43 + heartbeatInterval?: NodeJS.Timeout 44 44 acknowledged: boolean = true; 45 45 46 46 // https://docs.discord.com/developers/events/gateway#ready-event ··· 63 63 lastActiveStatus?: PresenceUpdateStatus = PresenceUpdateStatus.Offline; 64 64 lastActivities: GatewayActivity[] = []; 65 65 66 - activityTimeout: NodeJS.Timeout; 66 + activityTimeout?: NodeJS.Timeout; 67 + clearLastActivitiesTimeout?: NodeJS.Timeout; 68 + 69 + get friendlySocketState() { return `Socket state: ${wsReadyStateToStr(this.client.readyState)}`} 67 70 68 71 emitter: EventEmitter; 69 72 ··· 81 84 } 82 85 83 86 initClient = async () => { 84 - // let baseUrl: string; 85 - // if (this.resume_gateway_url !== undefined) { 86 - // baseUrl = this.resume_gateway_url; 87 - // } else { 88 - // if (this.initialGatewayUrl === undefined) { 89 - // try { 90 - // await this.fetchGatewayUrl(); 91 - // baseUrl = this.initialGatewayUrl; 92 - // } catch (e) { 93 - // throw new Error('Could not get initial gateway url', { cause: e }); 94 - // } 95 - // } else { 96 - // baseUrl = this.initialGatewayUrl; 97 - // } 98 - // } 99 - 100 - // const gatewayUrl = `${baseUrl}?encoding=json&v=10`; 101 - // this.logger.debug(`Using Gateway URL ${gatewayUrl}`); 102 87 103 88 const url = () => { 104 89 let baseUrl: string; ··· 219 204 protected authenticate = () => { 220 205 if (this.canResume && this.session_id !== undefined) { 221 206 // using resume 222 - this.handleResume(); 207 + this.sendResume(); 223 208 } else { 224 209 // initial identify 225 - this.handleIdentify(); 210 + this.sendIdentify(); 226 211 } 227 212 } 228 213 ··· 295 280 } 296 281 } 297 282 298 - handleIdentify() { 283 + sendIdentify() { 299 284 const data: GatewayIdentify = { 300 285 op: GatewayOpcodes.Identify, 301 286 d: { ··· 321 306 // jitter 322 307 // https://docs.discord.com/developers/events/gateway#heartbeat-interval 323 308 const sleepTime = randomInt(data.heartbeat_interval - 1); 324 - this.logger.debug(`Heartbeat Interval: ${data.heartbeat_interval}ms (${Math.floor(data.heartbeat_interval / 1000)}s), waiting ${Math.floor(sleepTime / 1000)}s before sending first heartbeat.`); 309 + this.logger.debug(`Heartbeat Interval: ${data.heartbeat_interval}ms (${Math.floor(data.heartbeat_interval / 1000)}s), waiting ${Math.floor(sleepTime / 1000)}s before sending first heartbeat. (Seq ${this.sequence})`); 325 310 await sleep(sleepTime); 326 311 if (this.client.readyState !== this.client.OPEN) { 327 312 this.logger.warn(`Not continuing with heartbeat because connection is not open`); ··· 336 321 return; 337 322 } 338 323 if (!this.acknowledged) { 339 - // zombied! 324 + this.logger.warn('Did not recieve heartbeat acknowledgment! May be a zombie so trying to reconnect.'); 340 325 return this.handleReconnect().then(() => null).catch((e) => this.logger.error(e)); 341 326 } 342 327 this.sendHeartbeat(); ··· 356 341 this.logger.warn(`Cannot send heartbeat because connection is not open`); 357 342 return; 358 343 } else if(isDebugMode()) { 359 - this.logger.debug('Sending heartbeat'); 344 + this.logger.debug(`Sending heartbeat (Seq ${this.sequence})`); 360 345 } 361 346 this.client.send(JSON.stringify(heartbeatRequest)); 362 347 } ··· 373 358 this.ready = true; 374 359 this.authOK = true; 375 360 this.reconnecting = false; 376 - this.logger.verbose(`Gateway Connection READY for ${this.user.username}`); 361 + this.logger.verbose(`Gateway Connection READY for ${this.user.username} | Session ${this.session_id} (Seq ${this.sequence})`); 362 + this.cancelClearLastActivities(); 363 + this.emitter.emit('ready', {ready: true}); 364 + } 365 + 366 + handleResume() { 367 + this.logger.verbose(`Recieved Resumed | Session ${this.session_id} (Seq ${this.sequence})`); 368 + this.canResume = true; 369 + this.ready = true; 370 + this.authOK = true; 371 + this.reconnecting = false; 372 + this.cancelClearLastActivities(); 377 373 this.emitter.emit('ready', {ready: true}); 378 374 } 379 375 ··· 384 380 } 385 381 386 382 async handleReconnect() { 387 - 383 + this.logger.verbose('Starting reconnect attempt'); 388 384 // on a manual close ws-isosocket does not retry 389 385 // so we need to do it manually 390 386 this.client.close(); 391 - const result = await Promise.race([ 392 - pEvent(this.client, 'close'), 393 - sleep(3000), 394 - ]); 395 - if(result === undefined) { 396 - throw new Error('Waited too long for client to close'); 387 + 388 + // closing manually also does not trigger a 'close' event so we need to check readyState 389 + const now = dayjs(); 390 + let elapsed = 0; 391 + if(this.client.readyState !== this.client.CLOSED) { 392 + this.logger.debug(`${this.friendlySocketState}, waiting for it to be closed...`); 393 + while(this.client.readyState !== this.client.CLOSED && elapsed < 10000) { 394 + elapsed = Math.abs(dayjs().diff(now, 'ms')); 395 + this.logger.debug(`Elapsed ${elapsed}ms | ${this.friendlySocketState}`); 396 + await sleep(1000); 397 + } 397 398 } 399 + 400 + if(this.client.readyState !== this.client.CLOSED && this.client.readyState !== this.client.CONNECTING) { 401 + throw new Error('Waited too long for socket to close'); 402 + } 403 + this.logger.debug('Socket closed, reconnecting'); 404 + this.cleanupConnectionSync(); 398 405 try { 399 406 await this.tryAuthenticate(); 400 407 } catch (e) { 401 408 throw new Error('Could not manually reconnect', {cause: e}); 402 409 } 403 - //await this.cleanupConnection(); 404 - // maybe don't do this if we've failed N times 405 - //this.initClient(); 406 - //this.connect(); 407 410 } 408 411 409 - handleResume() { 412 + sendResume() { 410 413 const data: GatewayResumeData = { 411 414 token: this.config.token, 412 415 session_id: this.session_id, ··· 417 420 this.logger.warn(`Cannot send resume because connection is not open`); 418 421 return; 419 422 } else if(isDebugMode()) { 420 - this.logger.debug('Sending resume'); 423 + this.logger.verbose(`Sending resume | Session ${this.session_id} | Seq ${this.sequence}`); 421 424 } 422 425 this.client.send(JSON.stringify({ op: GatewayOpcodes.Resume, d: data })); 423 426 } 424 427 425 - async cleanupConnection() { 426 - clearInterval(this.heartbeatInterval); 427 - this.heartbeatInterval = undefined; 428 - clearTimeout(this.activityTimeout); 429 - this.activityTimeout = undefined; 430 - 431 - if (this.client.CLOSED !== this.client.readyState) { 432 - this.client.close(); 433 - // wait for close or just give it a few seconds 434 - const result = await Promise.race([ 435 - pEvent(this.client, 'close'), 436 - sleep(3000), 437 - ]); 438 - } else if(this.client.CLOSING === this.client.readyState) { 439 - this.logger.debug('Giving the client time to close...'); 440 - await sleep(3000); 441 - } 442 - this.ready = false; 443 - if (!this.canResume) { 444 - this.logger.debug('Cannot resume session, clearing session data for clean reconnect'); 445 - this.session_id = undefined; 446 - this.sequence = undefined; 447 - this.resume_gateway_url = undefined; 448 - this.user = undefined; 428 + clearLastActivities(now: boolean = false) { 429 + this.cancelClearLastActivities(); 430 + if(!now) { 431 + // clear activities if takes longer than 5 seconds to get READY 432 + // 433 + // allows us to reasonably assume activities have not changed during a clean reconnect that did not take very long 434 + // so that we don't need to wait for another SESSION_REPLACE and can immediately update now playing 435 + // 436 + // but if user has to restart later, or something else goes wrong, we are sure we have reset activities when next READY 437 + this.logger.debug('Delaying last activities clear for 5 seconds...'); 438 + this.clearLastActivitiesTimeout = setTimeout(() => { 439 + this.logger.debug('Clearing last activities'); 440 + this.lastActiveStatus = PresenceUpdateStatus.Offline; 441 + this.lastActivities = []; 442 + this.clearLastActivitiesTimeout = undefined; 443 + }, 5000); 444 + } else { 445 + this.logger.debug('Clearing last activities'); 449 446 this.lastActiveStatus = PresenceUpdateStatus.Offline; 450 447 this.lastActivities = []; 451 448 } 452 449 } 453 450 451 + cancelClearLastActivities() { 452 + if(this.clearLastActivitiesTimeout !== undefined) { 453 + this.logger.debug('Cancelling clear activities timeout'); 454 + clearTimeout(this.clearLastActivitiesTimeout); 455 + this.clearLastActivitiesTimeout = undefined; 456 + } 457 + } 458 + 454 459 cleanupConnectionSync() { 455 460 if(this.heartbeatInterval !== undefined) { 456 461 clearInterval(this.heartbeatInterval); ··· 462 467 } 463 468 this.ready = false; 464 469 if (!this.canResume) { 465 - this.logger.debug('Cannot resume session, clearing session data for clean reconnect'); 470 + this.logger.verbose('Cannot resume session, clearing session data for clean reconnect'); 466 471 this.session_id = undefined; 467 472 this.sequence = undefined; 468 473 this.resume_gateway_url = undefined; 469 474 this.user = undefined; 470 - this.lastActiveStatus = PresenceUpdateStatus.Offline; 471 - this.lastActivities = []; 475 + this.clearLastActivities(); 472 476 } 473 477 } 474 478 475 479 handleUserSessionUpdates = (data: UserSession[]) => { 476 - this.logger.debug('Recieved updated user sessions'); 477 480 if (data.filter(x => x.session_id !== this.session_id && x.session_id !== 'all').length === 0) { 478 - this.logger.debug('No other user sessions exist, marking our session presence as inactive'); 481 + this.logger.debug(`Recieved updated user sessions (Seq ${this.sequence}) => No other user sessions exist, marking our session presence as inactive`); 479 482 this.lastActiveStatus = PresenceUpdateStatus.Offline; 480 483 this.lastActivities = []; 481 484 return; ··· 491 494 }; 492 495 return sessionId; 493 496 }); 494 - this.logger.debug(sessionSummaries.join('\n')); 497 + this.logger.debug(`Recieved updated user sessions (Seq ${this.sequence})\n${sessionSummaries.join('\n')}`); 495 498 496 499 const last = this.lastActiveStatus; 497 500 ··· 518 521 const ourSession = data.find(x => x.session_id === this.session_id); 519 522 if(ourSession !== undefined && ourSession.activities.length > 0) { 520 523 this.logger.debug(`Clearing our session presence, MS presence no longer allowed because ${reason}`); 521 - this.clearActivity(); 524 + this.sendClearActivity(); 522 525 } 523 526 } 524 527 } 525 528 526 529 async handleMessage(message: _DataPayload<GatewayDispatchEvents> | _NonDispatchPayload) { 530 + const { op, s } = message; 531 + if (s !== null && s !== undefined) { 532 + this.sequence = s; 533 + } 534 + const seqHint = `Seq ${this.sequence}`; 527 535 try { 528 - const { op, s } = message; 529 - if (s !== null && s !== undefined) { 530 - this.sequence = s; 531 - } 532 - if (isDebugMode()) { 533 - const friendlyOp = opcodeToFriendly(op); 534 - let handleHint = `Got opcode ${op}${friendlyOp !== op ? ` (${friendlyOp})` : ''}`; 535 - if (op === GatewayOpcodes.Dispatch) { 536 - handleHint += ` w/ Dispatch Event ${message.t}`; 537 - } 538 - this.logger.debug(handleHint); 539 - } 540 536 541 537 switch (op) { 542 538 case GatewayOpcodes.Hello: ··· 544 540 break; 545 541 case GatewayOpcodes.HeartbeatAck: 546 542 if (isDebugMode()) { 547 - this.logger.debug("Heartbeat acknowledged"); 543 + this.logger.debug(`Received Heartbeat acknowledgement (${seqHint})`); 548 544 } 549 545 this.acknowledged = true; 550 546 break; 551 547 case GatewayOpcodes.Heartbeat: 552 548 if (isDebugMode()) { 553 - this.logger.debug("Received Heartbeat"); 549 + this.logger.debug(`Received Heartbeat request (${seqHint})`); 554 550 } 555 551 this.sendHeartbeat(); 556 552 break; ··· 569 565 // @ts-expect-error 570 566 this.handleUserSessionUpdates(message.d as UserSession[]); 571 567 break; 568 + case GatewayDispatchEvents.Resumed: 569 + this.handleResume(); 570 + break; 571 + default: 572 + if (isDebugMode()) { 573 + this.logger.debug(`Recieved Dispatch Event ${message.t} (${seqHint})`); 574 + } 572 575 }; 573 576 break; 574 577 case GatewayOpcodes.InvalidSession: 575 - this.logger.debug('Recieved invalid session opcode'); 578 + this.logger.verbose(`Recieved invalid session opcode (${seqHint})`); 576 579 this.handleInvalidSession(message.d as GatewayInvalidSessionData); 577 580 break; 578 581 case GatewayOpcodes.Reconnect: 579 - this.logger.debug('Recieved reconnect opcode'); 582 + this.logger.verbose(`Recieved reconnect opcode (${seqHint})`); 580 583 await this.handleReconnect(); 581 584 break; 582 - case GatewayOpcodes.Resume: 583 - this.logger.debug({ data: message.d }, 'Recieved Resumed session'); 584 - this.canResume = true; 585 - this.ready = true; 586 - this.authOK = true; 587 - this.reconnecting = false; 588 - this.emitter.emit('ready', {ready: true}); 589 - break; 590 585 case GatewayOpcodes.Identify: 591 - this.logger.debug({ data: message.d }, 'Recieved Identifiy opcode'); 586 + this.logger.debug({ data: message.d }, `Recieved Identify opcode (${seqHint})`); 592 587 break; 593 588 case GatewayOpcodes.PresenceUpdate: 594 - this.logger.debug({ data: message.d }, 'Recieved Presence Update opcode'); 589 + this.logger.debug({ data: message.d }, `Recieved Presence Update opcode (${seqHint})`); 595 590 break; 596 591 default: 597 - this.logger.debug(`Recieved unhandled opcode: ${op}`); 592 + this.logger.debug(`Recieved opcode: ${op} (${seqHint})`); 598 593 break; 599 594 } 600 595 } catch (error) { 601 - throw new Error('Error handling gateway message', { cause: error }); 596 + const friendlyOp = opcodeToFriendly(op); 597 + let handleHint = `opcode ${op}${friendlyOp !== op ? ` (${friendlyOp})` : ''}`; 598 + if (op === GatewayOpcodes.Dispatch) { 599 + handleHint += ` w/ Dispatch Event ${message.t}`; 600 + } 601 + throw new Error(`Error handling gateway message for ${handleHint} (${seqHint})`, { cause: error }); 602 602 } 603 603 } 604 604 ··· 666 666 667 667 sendActivity = async (data: SourceData | undefined) => { 668 668 if(data === undefined) { 669 - this.clearActivity(); 669 + this.sendClearActivity(); 670 670 return; 671 671 } 672 672 const [sendOk, reasons] = this.checkOkToSend(); ··· 702 702 clearTimeout(this.activityTimeout); 703 703 } 704 704 this.activityTimeout = setTimeout(() => { 705 - this.clearActivity(); 705 + this.sendClearActivity(); 706 706 }, Math.abs(clearTime.diff(dayjs(), 'ms'))); 707 707 } 708 708 709 - clearActivity = () => { 709 + sendClearActivity = () => { 710 710 if (this.activityTimeout !== undefined) { 711 711 clearTimeout(this.activityTimeout); 712 712 this.activityTimeout = undefined;

+15

src/backend/utils/NetworkUtils.ts

··· 250 250 } 251 251 export const isRetryEvent = (e: Event): e is RetryEvent => { 252 252 return e.type === 'retry'; 253 + } 254 + 255 + export const wsReadyStateToStr = (state: number): string => { 256 + switch(state) { 257 + case 0: 258 + return 'connecting'; 259 + case 1: 260 + return 'open'; 261 + case 2: 262 + return 'closing'; 263 + case 3: 264 + return 'closed'; 265 + default: 266 + return state.toString(); 267 + } 253 268 }

Configure Feed

Configure Feed