foxxmd.dev / multi-scrobbler
[READ-ONLY] Mirror of https://github.com/FoxxMD/multi-scrobbler. Scrobble plays from multiple sources to multiple clients docs.multi-scrobbler.app
deezer docker jellyfin koito lastfm listenbrainz maloja mopidy mpris music music-assistant plex scrobble self-hosted spotify subsonic tautulli youtube-music
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

refactor: Implement abortable source polling

Use abort-controller-x with signals to interrupt polling process

#483

FoxxMD (Apr 2, 2026, 6:36 PM UTC) 03f8b207 c49cdba7

+208 -72
+25 -6
package-lock.json
··· 40 40 "@tailwindcss/vite": "^4.1.16", 41 41 "@types/express-serve-static-core": "^4.19.6", 42 42 "@xhayper/discord-rpc": "^1.3.0", 43 + "abort-controller-x": "^0.5.0", 43 44 "address": "^1.2.2", 44 45 "ajv": "^8.18.0", 45 46 "ajv-formats": "^3.0.1", ··· 155 156 "@types/xml2js": "^0.4.11", 156 157 "@vitejs/plugin-react": "^4.2.1", 157 158 "chai": "^4.3.6", 158 - "chai-as-promised": "^7.1.1", 159 + "chai-as-promised": "^8.0.2", 159 160 "eslint": "^8.56.0", 160 161 "eslint-plugin-prefer-arrow-functions": "^3.2.4", 161 162 "eslint-plugin-storybook": "10.1.11", ··· 175 176 "with-local-tmp-dir": "^6.0.0" 176 177 }, 177 178 "engines": { 178 - "node": ">=20.19.2", 179 + "node": ">=24.14.0", 179 180 "npm": ">=11.12.1" 180 181 } 181 182 }, ··· 5675 5676 "node": ">=6.5" 5676 5677 } 5677 5678 }, 5679 + "node_modules/abort-controller-x": { 5680 + "version": "0.5.0", 5681 + "resolved": "https://registry.npmjs.org/abort-controller-x/-/abort-controller-x-0.5.0.tgz", 5682 + "integrity": "sha512-yTt9CI0x+nRfX6BFMenEGP8ooPvErGH6AbFz20C2IeOLIlDsrw/VHpgne3GsCEuTA410IiFiaLVFKmgM4bKEPQ==", 5683 + "license": "MIT" 5684 + }, 5678 5685 "node_modules/accepts": { 5679 5686 "version": "1.3.8", 5680 5687 "license": "MIT", ··· 6375 6382 } 6376 6383 }, 6377 6384 "node_modules/chai-as-promised": { 6378 - "version": "7.1.2", 6385 + "version": "8.0.2", 6386 + "resolved": "https://registry.npmjs.org/chai-as-promised/-/chai-as-promised-8.0.2.tgz", 6387 + "integrity": "sha512-1GadL+sEJVLzDjcawPM4kjfnL+p/9vrxiEUonowKOAzvVg0PixJUdtuDzdkDeQhK3zfOE76GqGkZIQ7/Adcrqw==", 6379 6388 "dev": true, 6380 - "license": "WTFPL", 6389 + "license": "MIT", 6381 6390 "dependencies": { 6382 - "check-error": "^1.0.2" 6391 + "check-error": "^2.1.1" 6383 6392 }, 6384 6393 "peerDependencies": { 6385 - "chai": ">= 2.1.2 < 6" 6394 + "chai": ">= 2.1.2 < 7" 6395 + } 6396 + }, 6397 + "node_modules/chai-as-promised/node_modules/check-error": { 6398 + "version": "2.1.3", 6399 + "resolved": "https://registry.npmjs.org/check-error/-/check-error-2.1.3.tgz", 6400 + "integrity": "sha512-PAJdDJusoxnwm1VwW07VWwUN1sl7smmC3OKggvndJFadxxDRyFJBX/ggnu/KE4kQAB7a3Dp8f/YXC1FlUprWmA==", 6401 + "dev": true, 6402 + "license": "MIT", 6403 + "engines": { 6404 + "node": ">= 16" 6386 6405 } 6387 6406 }, 6388 6407 "node_modules/chalk": {
+2 -1
package.json
··· 78 78 "@tailwindcss/vite": "^4.1.16", 79 79 "@types/express-serve-static-core": "^4.19.6", 80 80 "@xhayper/discord-rpc": "^1.3.0", 81 + "abort-controller-x": "^0.5.0", 81 82 "address": "^1.2.2", 82 83 "ajv": "^8.18.0", 83 84 "ajv-formats": "^3.0.1", ··· 193 194 "@types/xml2js": "^0.4.11", 194 195 "@vitejs/plugin-react": "^4.2.1", 195 196 "chai": "^4.3.6", 196 - "chai-as-promised": "^7.1.1", 197 + "chai-as-promised": "^8.0.2", 197 198 "eslint": "^8.56.0", 198 199 "eslint-plugin-prefer-arrow-functions": "^3.2.4", 199 200 "eslint-plugin-storybook": "10.1.11",
+32 -8
src/backend/common/errors/MSErrors.ts
··· 1 1 import { parseRegexSingle } from "@foxxmd/regex-buddy-core"; 2 2 import mergeErrorCause from 'merge-error-cause'; 3 - import { findCauseByFunc, findCauseByReference } from "../../utils/ErrorUtils.js"; 3 + import { findCauseByFunc, findCauseByReference, isAbortReasonErrorLike } from "../../utils/ErrorUtils.js"; 4 4 import { UpstreamError, UpstreamErrorOptions } from "./UpstreamError.js"; 5 + import { isAbortError } from "abort-controller-x"; 5 6 6 7 export abstract class NamedError extends Error { 7 8 public abstract name: string; ··· 39 40 simple: boolean; 40 41 name = 'Error'; 41 42 43 + stackShortened: boolean = false; 44 + 45 + shortenStack() { 46 + const atIndex = parseRegexSingle(STACK_AT_REGEX,this.stack); 47 + if(atIndex !== undefined) { 48 + const firstn = this.stack.indexOf('\n', atIndex.index + atIndex.match.length); 49 + if(firstn !== -1) { 50 + this.stack = this.stack.slice(0, firstn); 51 + this.stackShortened = true; 52 + } 53 + } 54 + } 55 + 42 56 public constructor(msg: string, options?: ErrorOptions & { simple?: boolean, shortStack?: boolean }) { 43 57 super(msg, options); 44 58 const { ··· 46 60 shortStack = false 47 61 } = options || {}; 48 62 this.simple = simple; 63 + Error.captureStackTrace(this, this.constructor); 49 64 if(shortStack) { 50 - const atIndex = parseRegexSingle(STACK_AT_REGEX,this.stack); 51 - if(atIndex !== undefined) { 52 - const firstn = this.stack.indexOf('\n', atIndex.index + atIndex.match.length); 53 - if(firstn !== -1) { 54 - this.stack = this.stack.slice(0, firstn); 55 - } 56 - } 65 + this.shortenStack(); 57 66 } 58 67 } 59 68 } ··· 104 113 super(message, options); 105 114 this.payload = options?.payload; 106 115 } 116 + } 117 + 118 + export class AbortedError extends SimpleError { 119 + name = 'Aborted Operation'; 120 + } 121 + export const generateLoggableAbortReason = (msg: string, signal: AbortSignal): AbortedError => { 122 + const reason = signal.reason; 123 + let err: AbortedError; 124 + if(isAbortReasonErrorLike(signal)) { 125 + err = new AbortedError(msg, {cause: reason}); 126 + } else { 127 + err = new AbortedError(`${msg} => ${reason ?? 'No Reason Given'}`, {simple: true, shortStack: true}); 128 + } 129 + Error.captureStackTrace(err, generateLoggableAbortReason); 130 + return err; 107 131 }
+2 -1
src/backend/server/api.ts
··· 33 33 import ScrobbleSources from "../sources/ScrobbleSources.js"; 34 34 import ScrobbleClients from "../scrobblers/ScrobbleClients.js"; 35 35 import prom from 'prom-client'; 36 + import { SimpleError } from "../common/errors/MSErrors.js"; 36 37 37 38 const maxBufferSize = 300; 38 39 const output: Record<number, FixedSizeList<LogDataPretty>> = {}; ··· 435 436 436 437 if(source.polling) { 437 438 source.logger.info('Source is already polling! Restarting polling...'); 438 - const stopRes = await source.tryStopPolling(); 439 + const stopRes = await source.tryStopPolling(new SimpleError('user initiated', {simple: true, shortStack: true})); 439 440 if(stopRes === true) { 440 441 source.poll({force, notify: false}).catch(e => source.logger.error(e)); 441 442 }
+89 -55
src/backend/sources/AbstractSource.ts
··· 38 38 import { getRoot } from '../ioc.js'; 39 39 import { componentFileLogger } from '../common/logging.js'; 40 40 import { WebhookPayload } from '../common/infrastructure/config/health/webhooks.js'; 41 - import { messageWithCauses, messageWithCausesTruncatedDefault } from '../utils/ErrorUtils.js'; 41 + import { isAbortReasonErrorLike, messageWithCauses, messageWithCausesTruncatedDefault } from '../utils/ErrorUtils.js'; 42 42 import { genericSourcePlayMatch } from '../utils/PlayComparisonUtils.js'; 43 43 import { findAsync, staggerMapper, StaggerOptions } from '../utils/AsyncUtils.js'; 44 44 import pMap, {pMapIterable} from 'p-map'; 45 45 import prom, { Counter, Gauge } from 'prom-client'; 46 46 import { normalizeStr } from '../utils/StringUtils.js'; 47 + import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever, AbortError, throwIfAborted } from 'abort-controller-x'; 48 + import { AbortedError, generateLoggableAbortReason } from '../common/errors/MSErrors.js'; 47 49 48 50 export interface RecentlyPlayedOptions { 49 51 limit?: number ··· 71 73 canPoll: boolean = false; 72 74 polling: boolean = false; 73 75 canBacklog: boolean = false; 74 - userPollingStopSignal: undefined | any; 76 + protected abortController: AbortController | undefined; 77 + protected pollingPromise: Promise<void> | undefined; 78 + stopPollingWaitInterval: number = 200; 75 79 pollRetries: number = 0; 76 80 tracksDiscovered: number = 0; 77 81 ··· 119 123 120 124 async [Symbol.asyncDispose]() { 121 125 if(this.canPoll) { 122 - await this.tryStopPolling(); 126 + await this.tryStopPolling('Source is being disposed'); 123 127 } 124 128 } 125 129 ··· 307 311 } 308 312 } 309 313 310 - protected processBacklog = async () => { 314 + protected processBacklog = async (signal: AbortSignal) => { 311 315 if (this.canBacklog) { 312 316 313 317 const { ··· 334 338 try { 335 339 this.logger.verbose(`Fetching the last ${backlogLimit}${backlogLimit === this.SCROBBLE_BACKLOG_COUNT ? ' (max) ' : ''} listens to check for backlogging...`); 336 340 backlogPlays = await this.getBackloggedPlays({limit: backlogLimit}); 341 + signal.throwIfAborted(); 337 342 } catch (e) { 338 343 throw new Error('Error occurred while fetching backlogged plays', {cause: e}); 339 344 } ··· 342 347 if (scrobbleBacklog) { 343 348 if (discovered.length > 0) { 344 349 this.logger.info('Scrobbling backlogged tracks...'); 350 + signal.throwIfAborted(); 345 351 await this.scrobble(discovered); 346 352 this.logger.info('Backlog scrobbling complete.'); 347 353 } else { ··· 351 357 this.logger.info('Backlog scrobbling is disabled by config, skipping...'); 352 358 } 353 359 } 360 + return; 354 361 } 355 362 356 363 protected getBackloggedPlays = async (options: RecentlyPlayedOptions): Promise<PlayObject[]> => { ··· 369 376 poll = async (options: {force?: boolean, notify?: boolean} = {}) => { 370 377 const {force = false, notify = false} = options; 371 378 379 + if(this.polling) { 380 + this.logger.error('Already polling!'); 381 + return; 382 + } 383 + 372 384 // TODO refactor to only use tryInitialize 373 385 if(!this.isReady() || force) { 374 386 try { ··· 387 399 if(!(await this.onPollPostAuthCheck())) { 388 400 return; 389 401 } 390 - try { 391 - await this.processBacklog(); 392 - } catch (e) { 393 - this.logger.error(new Error('Cannot start polling because error occurred while processing backlog', {cause: e})); 394 - await this.notify({ 395 - title: `${this.getIdentifier()} - Polling Error`, 396 - message: 'Cannot start polling because error occurred while processing backlog.', 397 - priority: 'error' 402 + 403 + this.abortController = new AbortController(); 404 + this.pollingPromise = spawn(this.abortController.signal, async (signal, { defer, fork }) => { 405 + defer(async () => { 406 + this.polling = false; 407 + this.isSleeping = false; 408 + this.emitEvent('statusChange', {status: 'Idle'}); 398 409 }); 399 - return; 400 - } 401 410 402 - await this.startPolling(); 411 + fork(async (fSignal) => { 412 + try { 413 + await this.processBacklog(fSignal); 414 + } catch (e) { 415 + throwIfAborted(fSignal); 416 + await this.notify({ 417 + title: `${this.getIdentifier()} - Polling Error`, 418 + message: 'Polling interrupted because error occurred while processing backlog.', 419 + priority: 'error' 420 + }); 421 + throw new Error('Polling interrupted because error occurred while processing backlog', { cause: e }); 422 + } 423 + }); 424 + await this.startPolling(signal); 425 + }).catch((e) => { 426 + if (isAbortError(e)) { 427 + const err = generateLoggableAbortReason('Polling stopped', this.abortController.signal); 428 + this.logger.info(err); 429 + this.logger.trace(e) 430 + } else { 431 + this.logger.warn(new Error('Polling stopped with error', { cause: e })); 432 + } 433 + }).finally(() => { 434 + this.abortController = undefined; 435 + this.pollingPromise = undefined; 436 + }); 403 437 } 404 438 405 - startPolling = async () => { 439 + startPolling = async (signal: AbortSignal) => { 440 + signal.throwIfAborted(); 406 441 // reset poll attempts if already previously run 407 442 this.pollRetries = 0; 408 443 ··· 421 456 return; 422 457 } 423 458 424 - let pollRes: boolean | undefined = undefined; 425 - while (pollRes === undefined && this.pollRetries <= maxRetries) { 459 + while (this.pollRetries <= maxRetries) { 426 460 try { 427 461 if(!this.isReady() && this.buildOK) { 428 462 this.logger.verbose(`Source is no longer ready! Will attempt to reinitialize => Connection OK: ${this.connectionOK} | Auth OK: ${this.authed}`); ··· 430 464 if(init === false) { 431 465 throw new Error('Source failed reinitialization'); 432 466 } 433 - } 434 - pollRes = await this.doPolling(); 435 - if(pollRes === true) { 436 - break; 467 + signal.throwIfAborted(); 437 468 } 469 + await this.doPolling(signal); 438 470 } catch (e) { 471 + if(isAbortError(e)) { 472 + throw e; 473 + } 439 474 if (this.pollRetries < maxRetries) { 440 475 const delayFor = pollingBackoff(this.pollRetries + 1, retryMultiplier); 441 476 this.logger.info(`Poll retries (${this.pollRetries}) less than max poll retries (${maxRetries}), restarting polling after ${delayFor} second delay...`); 442 477 await this.notify({title: `${this.getIdentifier()} - Polling Retry`, message: `Encountered error while polling but retries (${this.pollRetries}) are less than max poll retries (${maxRetries}), restarting polling after ${delayFor} second delay. | Error: ${e.message}`, priority: 'warn'}); 443 478 await sleep((delayFor) * 1000); 479 + this.pollRetries++; 444 480 } else { 445 481 this.logger.warn(`Poll retries (${this.pollRetries}) equal to max poll retries (${maxRetries}), stopping polling!`); 446 482 await this.notify({title: `${this.getIdentifier()} - Polling Error`, message: `Encountered error while polling and retries (${this.pollRetries}) are equal to max poll retries (${maxRetries}), stopping polling!. | Error: ${e.message}`, priority: 'error'}); 483 + throw e; 447 484 } 448 - this.pollRetries++; 449 485 } 450 486 } 451 487 } 452 488 453 - tryStopPolling = async () => { 489 + tryStopPolling = async (reason?: string | Error) => { 454 490 if(this.polling === false) { 455 491 this.logger.warn(`Polling is already stopped!`); 456 - return; 492 + return true; 457 493 } 458 - this.userPollingStopSignal = true; 459 - let secsPassed = 0; 460 - while(this.userPollingStopSignal !== undefined && secsPassed < 10) { 461 - await sleep(2000); 462 - secsPassed += 2; 463 - this.logger.verbose(`Waiting for polling stop signal to be acknowledged (waited ${secsPassed}s)`); 494 + if(this.abortController === undefined) { 495 + this.logger.error('No abort controller found! Nothing to stop.'); 496 + return false; 464 497 } 465 - if(this.userPollingStopSignal !== undefined) { 498 + this.abortController.abort(reason); 499 + let elapsed = 0; 500 + let lastlog: Dayjs; 501 + while(this.polling && elapsed < 10) { 502 + if(lastlog === undefined || dayjs().diff(lastlog, 's') >= 2) { 503 + this.logger.verbose(`Waiting for polling stop signal to be acknowledged (waited ${formatNumber(elapsed/1000)}s)`); 504 + } 505 + await sleep(this.stopPollingWaitInterval); 506 + elapsed += this.stopPollingWaitInterval; 507 + } 508 + if(this.polling) { 466 509 this.logger.warn('Could not stop polling! Or polling signal was lost :('); 467 510 return false; 468 511 } 469 512 return true; 470 513 } 471 514 472 - protected doStopPolling = (reason: string = 'system') => { 473 - this.polling = false; 474 - this.userPollingStopSignal = undefined; 475 - this.emitEvent('statusChange', {status: 'Idle'}); 476 - this.logger.info(`Stopped polling due to: ${reason}`); 477 - } 478 - 479 - protected shouldStopPolling = () => this.polling === false || this.userPollingStopSignal !== undefined; 515 + protected doPolling = async (signal: AbortSignal): Promise<true | undefined> => { 516 + signal.throwIfAborted(); 480 517 481 - protected doPolling = async (): Promise<true | undefined> => { 482 - if (this.polling === true) { 483 - return true; 484 - } 485 518 this.logger.info('Polling started'); 486 519 this.emitEvent('statusChange', {status: 'Running'}); 487 520 await this.notify({title: `${this.getIdentifier()} - Polling Started`, message: 'Polling Started', priority: 'info'}); ··· 494 527 495 528 try { 496 529 this.polling = true; 497 - while (!this.shouldStopPolling()) { 530 + while (true) { 531 + signal.throwIfAborted(); 498 532 const pollFrom = dayjs(); 499 533 let lastActivityLogLevel: LogLevel = 'trace'; 500 534 ··· 503 537 playObjs = await this.getRecentlyPlayed({formatted: true}); 504 538 } catch (e) { 505 539 throw new Error('Error occurred while refreshing recently played', {cause: e}); 540 + } finally { 541 + signal.throwIfAborted(); 506 542 } 507 543 508 544 ··· 526 562 await sleep(maxDelay * 1000); 527 563 } 528 564 newDiscovered = await this.discover(playObjs); 565 + signal.throwIfAborted(); 529 566 this.scrobble(newDiscovered, 530 567 { 531 568 forceRefresh: closeToInterval ··· 571 608 this.logger[lastActivityLogLevel](activityMsgs.join(' | ')); 572 609 this.setWakeAt(pollFrom.add(sleepTime, 'seconds')); 573 610 this.setIsSleeping(true); 574 - while(!this.shouldStopPolling() && dayjs().isBefore(this.getWakeAt())) { 611 + while(dayjs().isBefore(this.getWakeAt())) { 575 612 // check for polling status every half second and wait till wake up time 576 - await sleep(500); 613 + await delay(signal, 500); 577 614 } 578 615 this.setIsSleeping(false); 579 - 616 + // if we have made it this far in the loop we can reset poll retries 617 + this.pollRetries = 0; 580 618 } 581 - if(this.shouldStopPolling()) { 582 - this.doStopPolling(this.userPollingStopSignal !== undefined ? 'user input' : undefined); 583 - return true; 619 + } catch (e) { 620 + if(!isAbortError(e)) { 621 + this.logger.error(new Error('Error occurred while polling', {cause: e})); 584 622 } 585 - } catch (e) { 586 - this.logger.error(new Error('Error occurred while polling', {cause: e})); 587 623 if(e.message.includes('Status code: 401')) { 588 624 this.authed = false; 589 625 this.authFailure = true; 590 626 } 591 - this.emitEvent('statusChange', {status: 'Idle'}); 592 - this.polling = false; 593 627 throw e; 594 628 } finally { 595 629 this.setIsSleeping(false);
+54
src/backend/tests/promises/abortable.test.ts
··· 1 + import chai from 'chai'; 2 + import asPromised from 'chai-as-promised'; 3 + import { describe, it } from 'mocha'; 4 + import { sleep } from "../../utils.js"; 5 + import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever } from 'abort-controller-x'; 6 + 7 + chai.should(); 8 + chai.use(asPromised); 9 + 10 + const expect = chai.expect; 11 + 12 + describe('#Abortable', function () { 13 + 14 + it('Executes defer on non-abort error', async function () { 15 + 16 + const controller = new AbortController(); 17 + let didUseDefer = false; 18 + 19 + try { 20 + await spawn(controller.signal, async (signal, { defer }) => { 21 + defer(async () => { 22 + didUseDefer = true; 23 + }); 24 + 25 + await delay(signal, 100); 26 + throw new Error('Not an abort signal'); 27 + }) 28 + } catch (e) { 29 + rethrowAbortError(e); 30 + } 31 + expect(didUseDefer).is.true; 32 + }); 33 + 34 + it('Executes defer on abort error', async function () { 35 + 36 + const controller = new AbortController(); 37 + let didUseDefer = false; 38 + 39 + const spawnPromise = spawn(controller.signal, async (signal, { defer }) => { 40 + defer(async () => { 41 + didUseDefer = true; 42 + }); 43 + 44 + await forever(signal); 45 + }).catch(catchAbortError); 46 + 47 + await sleep(50); 48 + controller.abort(); 49 + 50 + await spawnPromise.should.be.fulfilled; 51 + expect(didUseDefer).is.true; 52 + }); 53 + 54 + });
+4 -1
src/backend/utils/ErrorUtils.ts
··· 1 + import { isAbortError } from "abort-controller-x"; 1 2 import { truncateStringToLength } from "../../core/StringUtils.js"; 2 3 3 4 /** ··· 116 117 return (err: Error) => messageWithCauses(err, t); 117 118 } 118 119 119 - export const messageWithCausesTruncatedDefault = messageWithCausesTruncated(100); 120 + export const messageWithCausesTruncatedDefault = messageWithCausesTruncated(100); 121 + 122 + export const isAbortReasonErrorLike = (signal: AbortSignal) => signal.aborted && signal.reason !== undefined && (isAbortError(signal.reason) || signal.reason instanceof Error);