diff --git a/package-lock.json b/package-lock.json index aa40c50f..491d54da 100644 --- a/package-lock.json +++ b/package-lock.json @@ -40,6 +40,7 @@ "@tailwindcss/vite": "^4.1.16", "@types/express-serve-static-core": "^4.19.6", "@xhayper/discord-rpc": "^1.3.0", + "abort-controller-x": "^0.5.0", "address": "^1.2.2", "ajv": "^8.18.0", "ajv-formats": "^3.0.1", @@ -155,7 +156,7 @@ "@types/xml2js": "^0.4.11", "@vitejs/plugin-react": "^4.2.1", "chai": "^4.3.6", - "chai-as-promised": "^7.1.1", + "chai-as-promised": "^8.0.2", "eslint": "^8.56.0", "eslint-plugin-prefer-arrow-functions": "^3.2.4", "eslint-plugin-storybook": "10.1.11", @@ -175,7 +176,7 @@ "with-local-tmp-dir": "^6.0.0" }, "engines": { - "node": ">=20.19.2", + "node": ">=24.14.0", "npm": ">=11.12.1" } }, @@ -5675,6 +5676,12 @@ "node": ">=6.5" } }, + "node_modules/abort-controller-x": { + "version": "0.5.0", + "resolved": "https://registry.npmjs.org/abort-controller-x/-/abort-controller-x-0.5.0.tgz", + "integrity": "sha512-yTt9CI0x+nRfX6BFMenEGP8ooPvErGH6AbFz20C2IeOLIlDsrw/VHpgne3GsCEuTA410IiFiaLVFKmgM4bKEPQ==", + "license": "MIT" + }, "node_modules/accepts": { "version": "1.3.8", "license": "MIT", @@ -6375,14 +6382,26 @@ } }, "node_modules/chai-as-promised": { - "version": "7.1.2", + "version": "8.0.2", + "resolved": "https://registry.npmjs.org/chai-as-promised/-/chai-as-promised-8.0.2.tgz", + "integrity": "sha512-1GadL+sEJVLzDjcawPM4kjfnL+p/9vrxiEUonowKOAzvVg0PixJUdtuDzdkDeQhK3zfOE76GqGkZIQ7/Adcrqw==", "dev": true, - "license": "WTFPL", + "license": "MIT", "dependencies": { - "check-error": "^1.0.2" + "check-error": "^2.1.1" }, "peerDependencies": { - "chai": ">= 2.1.2 < 6" + "chai": ">= 2.1.2 < 7" + } + }, + "node_modules/chai-as-promised/node_modules/check-error": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/check-error/-/check-error-2.1.3.tgz", + "integrity": "sha512-PAJdDJusoxnwm1VwW07VWwUN1sl7smmC3OKggvndJFadxxDRyFJBX/ggnu/KE4kQAB7a3Dp8f/YXC1FlUprWmA==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 16" } }, "node_modules/chalk": { diff --git a/package.json b/package.json index 475e35f2..70bb8178 100644 --- a/package.json +++ b/package.json @@ -78,6 +78,7 @@ "@tailwindcss/vite": "^4.1.16", "@types/express-serve-static-core": "^4.19.6", "@xhayper/discord-rpc": "^1.3.0", + "abort-controller-x": "^0.5.0", "address": "^1.2.2", "ajv": "^8.18.0", "ajv-formats": "^3.0.1", @@ -193,7 +194,7 @@ "@types/xml2js": "^0.4.11", "@vitejs/plugin-react": "^4.2.1", "chai": "^4.3.6", - "chai-as-promised": "^7.1.1", + "chai-as-promised": "^8.0.2", "eslint": "^8.56.0", "eslint-plugin-prefer-arrow-functions": "^3.2.4", "eslint-plugin-storybook": "10.1.11", diff --git a/src/backend/common/errors/MSErrors.ts b/src/backend/common/errors/MSErrors.ts index 5eb8d065..dcfdc382 100644 --- a/src/backend/common/errors/MSErrors.ts +++ b/src/backend/common/errors/MSErrors.ts @@ -1,7 +1,8 @@ import { parseRegexSingle } from "@foxxmd/regex-buddy-core"; import mergeErrorCause from 'merge-error-cause'; -import { findCauseByFunc, findCauseByReference } from "../../utils/ErrorUtils.js"; +import { findCauseByFunc, findCauseByReference, isAbortReasonErrorLike } from "../../utils/ErrorUtils.js"; import { UpstreamError, UpstreamErrorOptions } from "./UpstreamError.js"; +import { isAbortError } from "abort-controller-x"; export abstract class NamedError extends Error { public abstract name: string; @@ -39,6 +40,19 @@ export class SimpleError extends Error implements HasSimpleError { simple: boolean; name = 'Error'; + stackShortened: boolean = false; + + shortenStack() { + const atIndex = parseRegexSingle(STACK_AT_REGEX,this.stack); + if(atIndex !== undefined) { + const firstn = this.stack.indexOf('\n', atIndex.index + atIndex.match.length); + if(firstn !== -1) { + this.stack = this.stack.slice(0, firstn); + this.stackShortened = true; + } + } + } + public constructor(msg: string, options?: ErrorOptions & { simple?: boolean, shortStack?: boolean }) { super(msg, options); const { @@ -46,14 +60,9 @@ export class SimpleError extends Error implements HasSimpleError { shortStack = false } = options || {}; this.simple = simple; + Error.captureStackTrace(this, this.constructor); if(shortStack) { - const atIndex = parseRegexSingle(STACK_AT_REGEX,this.stack); - if(atIndex !== undefined) { - const firstn = this.stack.indexOf('\n', atIndex.index + atIndex.match.length); - if(firstn !== -1) { - this.stack = this.stack.slice(0, firstn); - } - } + this.shortenStack(); } } } @@ -104,4 +113,19 @@ export class ScrobbleSubmitError extends U super(message, options); this.payload = options?.payload; } +} + +export class AbortedError extends SimpleError { + name = 'Aborted Operation'; +} +export const generateLoggableAbortReason = (msg: string, signal: AbortSignal): AbortedError => { + const reason = signal.reason; + let err: AbortedError; + if(isAbortReasonErrorLike(signal)) { + err = new AbortedError(msg, {cause: reason}); + } else { + err = new AbortedError(`${msg} => ${reason ?? 'No Reason Given'}`, {simple: true, shortStack: true}); + } + Error.captureStackTrace(err, generateLoggableAbortReason); + return err; } \ No newline at end of file diff --git a/src/backend/server/api.ts b/src/backend/server/api.ts index b9e244de..5f14c80b 100644 --- a/src/backend/server/api.ts +++ b/src/backend/server/api.ts @@ -33,6 +33,7 @@ import { setupWebscrobblerRoutes } from "./webscrobblerRoutes.js"; import ScrobbleSources from "../sources/ScrobbleSources.js"; import ScrobbleClients from "../scrobblers/ScrobbleClients.js"; import prom from 'prom-client'; +import { SimpleError } from "../common/errors/MSErrors.js"; const maxBufferSize = 300; const output: Record> = {}; @@ -435,7 +436,7 @@ export const setupApi = (app: ExpressWithAsync, logger: Logger, appLoggerStream: if(source.polling) { source.logger.info('Source is already polling! Restarting polling...'); - const stopRes = await source.tryStopPolling(); + const stopRes = await source.tryStopPolling(new SimpleError('user initiated', {simple: true, shortStack: true})); if(stopRes === true) { source.poll({force, notify: false}).catch(e => source.logger.error(e)); } diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index c5861ac1..57157fb0 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -38,12 +38,14 @@ import { todayAwareFormat } from "../../core/TimeUtils.js"; import { getRoot } from '../ioc.js'; import { componentFileLogger } from '../common/logging.js'; import { WebhookPayload } from '../common/infrastructure/config/health/webhooks.js'; -import { messageWithCauses, messageWithCausesTruncatedDefault } from '../utils/ErrorUtils.js'; +import { isAbortReasonErrorLike, messageWithCauses, messageWithCausesTruncatedDefault } from '../utils/ErrorUtils.js'; import { genericSourcePlayMatch } from '../utils/PlayComparisonUtils.js'; import { findAsync, staggerMapper, StaggerOptions } from '../utils/AsyncUtils.js'; import pMap, {pMapIterable} from 'p-map'; import prom, { Counter, Gauge } from 'prom-client'; import { normalizeStr } from '../utils/StringUtils.js'; +import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever, AbortError, throwIfAborted } from 'abort-controller-x'; +import { AbortedError, generateLoggableAbortReason } from '../common/errors/MSErrors.js'; export interface RecentlyPlayedOptions { limit?: number @@ -71,7 +73,9 @@ export default abstract class AbstractSource extends AbstractComponent implement canPoll: boolean = false; polling: boolean = false; canBacklog: boolean = false; - userPollingStopSignal: undefined | any; + protected abortController: AbortController | undefined; + protected pollingPromise: Promise | undefined; + stopPollingWaitInterval: number = 200; pollRetries: number = 0; tracksDiscovered: number = 0; @@ -119,7 +123,7 @@ export default abstract class AbstractSource extends AbstractComponent implement async [Symbol.asyncDispose]() { if(this.canPoll) { - await this.tryStopPolling(); + await this.tryStopPolling('Source is being disposed'); } } @@ -307,7 +311,7 @@ export default abstract class AbstractSource extends AbstractComponent implement } } - protected processBacklog = async () => { + protected processBacklog = async (signal: AbortSignal) => { if (this.canBacklog) { const { @@ -334,6 +338,7 @@ export default abstract class AbstractSource extends AbstractComponent implement try { this.logger.verbose(`Fetching the last ${backlogLimit}${backlogLimit === this.SCROBBLE_BACKLOG_COUNT ? ' (max) ' : ''} listens to check for backlogging...`); backlogPlays = await this.getBackloggedPlays({limit: backlogLimit}); + signal.throwIfAborted(); } catch (e) { throw new Error('Error occurred while fetching backlogged plays', {cause: e}); } @@ -342,6 +347,7 @@ export default abstract class AbstractSource extends AbstractComponent implement if (scrobbleBacklog) { if (discovered.length > 0) { this.logger.info('Scrobbling backlogged tracks...'); + signal.throwIfAborted(); await this.scrobble(discovered); this.logger.info('Backlog scrobbling complete.'); } else { @@ -351,6 +357,7 @@ export default abstract class AbstractSource extends AbstractComponent implement this.logger.info('Backlog scrobbling is disabled by config, skipping...'); } } + return; } protected getBackloggedPlays = async (options: RecentlyPlayedOptions): Promise => { @@ -369,6 +376,11 @@ export default abstract class AbstractSource extends AbstractComponent implement poll = async (options: {force?: boolean, notify?: boolean} = {}) => { const {force = false, notify = false} = options; + if(this.polling) { + this.logger.error('Already polling!'); + return; + } + // TODO refactor to only use tryInitialize if(!this.isReady() || force) { try { @@ -387,22 +399,45 @@ export default abstract class AbstractSource extends AbstractComponent implement if(!(await this.onPollPostAuthCheck())) { return; } - try { - await this.processBacklog(); - } catch (e) { - this.logger.error(new Error('Cannot start polling because error occurred while processing backlog', {cause: e})); - await this.notify({ - title: `${this.getIdentifier()} - Polling Error`, - message: 'Cannot start polling because error occurred while processing backlog.', - priority: 'error' + + this.abortController = new AbortController(); + this.pollingPromise = spawn(this.abortController.signal, async (signal, { defer, fork }) => { + defer(async () => { + this.polling = false; + this.isSleeping = false; + this.emitEvent('statusChange', {status: 'Idle'}); }); - return; - } - await this.startPolling(); + fork(async (fSignal) => { + try { + await this.processBacklog(fSignal); + } catch (e) { + throwIfAborted(fSignal); + await this.notify({ + title: `${this.getIdentifier()} - Polling Error`, + message: 'Polling interrupted because error occurred while processing backlog.', + priority: 'error' + }); + throw new Error('Polling interrupted because error occurred while processing backlog', { cause: e }); + } + }); + await this.startPolling(signal); + }).catch((e) => { + if (isAbortError(e)) { + const err = generateLoggableAbortReason('Polling stopped', this.abortController.signal); + this.logger.info(err); + this.logger.trace(e) + } else { + this.logger.warn(new Error('Polling stopped with error', { cause: e })); + } + }).finally(() => { + this.abortController = undefined; + this.pollingPromise = undefined; + }); } - startPolling = async () => { + startPolling = async (signal: AbortSignal) => { + signal.throwIfAborted(); // reset poll attempts if already previously run this.pollRetries = 0; @@ -421,8 +456,7 @@ export default abstract class AbstractSource extends AbstractComponent implement return; } - let pollRes: boolean | undefined = undefined; - while (pollRes === undefined && this.pollRetries <= maxRetries) { + while (this.pollRetries <= maxRetries) { try { if(!this.isReady() && this.buildOK) { this.logger.verbose(`Source is no longer ready! Will attempt to reinitialize => Connection OK: ${this.connectionOK} | Auth OK: ${this.authed}`); @@ -430,58 +464,57 @@ export default abstract class AbstractSource extends AbstractComponent implement if(init === false) { throw new Error('Source failed reinitialization'); } + signal.throwIfAborted(); } - pollRes = await this.doPolling(); - if(pollRes === true) { - break; - } + await this.doPolling(signal); } catch (e) { + if(isAbortError(e)) { + throw e; + } if (this.pollRetries < maxRetries) { const delayFor = pollingBackoff(this.pollRetries + 1, retryMultiplier); this.logger.info(`Poll retries (${this.pollRetries}) less than max poll retries (${maxRetries}), restarting polling after ${delayFor} second delay...`); 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'}); await sleep((delayFor) * 1000); + this.pollRetries++; } else { this.logger.warn(`Poll retries (${this.pollRetries}) equal to max poll retries (${maxRetries}), stopping polling!`); 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'}); + throw e; } - this.pollRetries++; } } } - tryStopPolling = async () => { + tryStopPolling = async (reason?: string | Error) => { if(this.polling === false) { this.logger.warn(`Polling is already stopped!`); - return; + return true; } - this.userPollingStopSignal = true; - let secsPassed = 0; - while(this.userPollingStopSignal !== undefined && secsPassed < 10) { - await sleep(2000); - secsPassed += 2; - this.logger.verbose(`Waiting for polling stop signal to be acknowledged (waited ${secsPassed}s)`); + if(this.abortController === undefined) { + this.logger.error('No abort controller found! Nothing to stop.'); + return false; + } + this.abortController.abort(reason); + let elapsed = 0; + let lastlog: Dayjs; + while(this.polling && elapsed < 10) { + if(lastlog === undefined || dayjs().diff(lastlog, 's') >= 2) { + this.logger.verbose(`Waiting for polling stop signal to be acknowledged (waited ${formatNumber(elapsed/1000)}s)`); + } + await sleep(this.stopPollingWaitInterval); + elapsed += this.stopPollingWaitInterval; } - if(this.userPollingStopSignal !== undefined) { + if(this.polling) { this.logger.warn('Could not stop polling! Or polling signal was lost :('); return false; } return true; } - protected doStopPolling = (reason: string = 'system') => { - this.polling = false; - this.userPollingStopSignal = undefined; - this.emitEvent('statusChange', {status: 'Idle'}); - this.logger.info(`Stopped polling due to: ${reason}`); - } - - protected shouldStopPolling = () => this.polling === false || this.userPollingStopSignal !== undefined; + protected doPolling = async (signal: AbortSignal): Promise => { + signal.throwIfAborted(); - protected doPolling = async (): Promise => { - if (this.polling === true) { - return true; - } this.logger.info('Polling started'); this.emitEvent('statusChange', {status: 'Running'}); await this.notify({title: `${this.getIdentifier()} - Polling Started`, message: 'Polling Started', priority: 'info'}); @@ -494,7 +527,8 @@ export default abstract class AbstractSource extends AbstractComponent implement try { this.polling = true; - while (!this.shouldStopPolling()) { + while (true) { + signal.throwIfAborted(); const pollFrom = dayjs(); let lastActivityLogLevel: LogLevel = 'trace'; @@ -503,6 +537,8 @@ export default abstract class AbstractSource extends AbstractComponent implement playObjs = await this.getRecentlyPlayed({formatted: true}); } catch (e) { throw new Error('Error occurred while refreshing recently played', {cause: e}); + } finally { + signal.throwIfAborted(); } @@ -526,6 +562,7 @@ export default abstract class AbstractSource extends AbstractComponent implement await sleep(maxDelay * 1000); } newDiscovered = await this.discover(playObjs); + signal.throwIfAborted(); this.scrobble(newDiscovered, { forceRefresh: closeToInterval @@ -571,25 +608,22 @@ export default abstract class AbstractSource extends AbstractComponent implement this.logger[lastActivityLogLevel](activityMsgs.join(' | ')); this.setWakeAt(pollFrom.add(sleepTime, 'seconds')); this.setIsSleeping(true); - while(!this.shouldStopPolling() && dayjs().isBefore(this.getWakeAt())) { + while(dayjs().isBefore(this.getWakeAt())) { // check for polling status every half second and wait till wake up time - await sleep(500); + await delay(signal, 500); } this.setIsSleeping(false); - - } - if(this.shouldStopPolling()) { - this.doStopPolling(this.userPollingStopSignal !== undefined ? 'user input' : undefined); - return true; + // if we have made it this far in the loop we can reset poll retries + this.pollRetries = 0; } } catch (e) { - this.logger.error(new Error('Error occurred while polling', {cause: e})); + if(!isAbortError(e)) { + this.logger.error(new Error('Error occurred while polling', {cause: e})); + } if(e.message.includes('Status code: 401')) { this.authed = false; this.authFailure = true; } - this.emitEvent('statusChange', {status: 'Idle'}); - this.polling = false; throw e; } finally { this.setIsSleeping(false); diff --git a/src/backend/tests/promises/abortable.test.ts b/src/backend/tests/promises/abortable.test.ts new file mode 100644 index 00000000..63a908a6 --- /dev/null +++ b/src/backend/tests/promises/abortable.test.ts @@ -0,0 +1,54 @@ +import chai from 'chai'; +import asPromised from 'chai-as-promised'; +import { describe, it } from 'mocha'; +import { sleep } from "../../utils.js"; +import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever } from 'abort-controller-x'; + +chai.should(); +chai.use(asPromised); + +const expect = chai.expect; + +describe('#Abortable', function () { + + it('Executes defer on non-abort error', async function () { + + const controller = new AbortController(); + let didUseDefer = false; + + try { + await spawn(controller.signal, async (signal, { defer }) => { + defer(async () => { + didUseDefer = true; + }); + + await delay(signal, 100); + throw new Error('Not an abort signal'); + }) + } catch (e) { + rethrowAbortError(e); + } + expect(didUseDefer).is.true; + }); + + it('Executes defer on abort error', async function () { + + const controller = new AbortController(); + let didUseDefer = false; + + const spawnPromise = spawn(controller.signal, async (signal, { defer }) => { + defer(async () => { + didUseDefer = true; + }); + + await forever(signal); + }).catch(catchAbortError); + + await sleep(50); + controller.abort(); + + await spawnPromise.should.be.fulfilled; + expect(didUseDefer).is.true; + }); + +}); \ No newline at end of file diff --git a/src/backend/utils/ErrorUtils.ts b/src/backend/utils/ErrorUtils.ts index 9fe14552..2e7bee38 100644 --- a/src/backend/utils/ErrorUtils.ts +++ b/src/backend/utils/ErrorUtils.ts @@ -1,3 +1,4 @@ +import { isAbortError } from "abort-controller-x"; import { truncateStringToLength } from "../../core/StringUtils.js"; /** @@ -116,4 +117,6 @@ export const messageWithCausesTruncated = (length: number) => { return (err: Error) => messageWithCauses(err, t); } -export const messageWithCausesTruncatedDefault = messageWithCausesTruncated(100); \ No newline at end of file +export const messageWithCausesTruncatedDefault = messageWithCausesTruncated(100); + +export const isAbortReasonErrorLike = (signal: AbortSignal) => signal.aborted && signal.reason !== undefined && (isAbortError(signal.reason) || signal.reason instanceof Error); \ No newline at end of file -- 2.51.2 From fb6f6a18c8842b01346ebc80be44ae557e4fc413 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Thu, 2 Apr 2026 18:41:45 +0000 Subject: [PATCH 2/4] feat(source): Make discover abortable --- src/backend/sources/AbstractSource.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index 57157fb0..4ce07631 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -265,11 +265,13 @@ export default abstract class AbstractSource extends AbstractComponent implement return existing !== undefined; } - discover = async (plays: PlayObject[], options: { checkAll?: boolean, [key: string]: any } = {}): Promise => { + discover = async (plays: PlayObject[], options: { checkAll?: boolean, signal?: AbortSignal, [key: string]: any } = {}): Promise => { const newDiscoveredPlays: PlayObject[] = []; for await(const play of pMapIterable(plays, this.staggerMappers.preCompare(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { + options.signal?.throwIfAborted(); if(!(await this.alreadyDiscovered(play, options))) { + options.signal?.throwIfAborted() this.addPlayToDiscovered(play); newDiscoveredPlays.push(play); } @@ -342,7 +344,7 @@ export default abstract class AbstractSource extends AbstractComponent implement } catch (e) { throw new Error('Error occurred while fetching backlogged plays', {cause: e}); } - const discovered = await this.discover(backlogPlays, {discoverLocation: 'backlog'}); + const discovered = await this.discover(backlogPlays, {discoverLocation: 'backlog', signal}); if (scrobbleBacklog) { if (discovered.length > 0) { @@ -561,7 +563,7 @@ export default abstract class AbstractSource extends AbstractComponent implement this.logger.info(`Potential plays were discovered close to polling interval! Delaying scrobble clients refresh by ${maxDelay} seconds so other clients have time to scrobble first`); await sleep(maxDelay * 1000); } - newDiscovered = await this.discover(playObjs); + newDiscovered = await this.discover(playObjs, {signal}); signal.throwIfAborted(); this.scrobble(newDiscovered, { -- 2.51.2 From 1f165096b3f526bbd55c2ef14debb1be39eca363 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Thu, 2 Apr 2026 19:42:59 +0000 Subject: [PATCH 3/4] feat(client): make scrobble processing abortable --- .../scrobblers/AbstractScrobbleClient.ts | 283 ++++++++++-------- src/backend/server/api.ts | 2 +- .../tests/scrobbler/scrobblers.test.ts | 8 +- 3 files changed, 170 insertions(+), 123 deletions(-) diff --git a/src/backend/scrobblers/AbstractScrobbleClient.ts b/src/backend/scrobblers/AbstractScrobbleClient.ts index 9da59847..a9f818f0 100644 --- a/src/backend/scrobblers/AbstractScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractScrobbleClient.ts @@ -64,9 +64,10 @@ import { comparePlayArtistsNormalized, comparePlayTracksNormalized, existingScro import { lifecyclelessInvariantTransform } from "../../core/PlayUtils.js"; import { normalizeStr } from "../utils/StringUtils.js"; import prom, { Counter, Gauge } from 'prom-client'; -import { ScrobbleSubmitError, SimpleError } from "../common/errors/MSErrors.js"; +import { generateLoggableAbortReason, ScrobbleSubmitError, SimpleError } from "../common/errors/MSErrors.js"; import {serializeError} from 'serialize-error'; import { DEFAULT_NEW_PADDING, groupPlaysToTimeRanges } from "../utils/ListenFetchUtils.js"; +import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever, AbortError, throwIfAborted } from 'abort-controller-x'; type PlatformMappedPlays = Map; type NowPlayingQueue = Map; @@ -95,9 +96,10 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i scrobbleDelay: number = 1000; scrobbleSleep: number = 2000; scrobbleWaitStopInterval: number = 2000; + protected scrobbleQueueAbortController: AbortController | undefined; + protected scrobbleQueuePromise: Promise | undefined; scrobbleRetries: number = 0; scrobbling: boolean = false; - userScrobblingStopSignal: undefined | any; queuedScrobbles: QueuedScrobble[] = []; deadLetterScrobbles: DeadLetterScrobble[] = []; @@ -518,15 +520,20 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i return [matchPlayDate, dtInvariantMatches]; } - public scrobble = async (playObj: PlayObject, opts?: { delay?: number | false }): Promise => { - const {delay} = opts || {}; - const scrobbleDelay = delay === undefined ? this.scrobbleDelay : (delay === false ? 0 : delay); + public scrobble = async (playObj: PlayObject, opts?: { delay?: number | false, signal?: AbortSignal }): Promise => { + const {delay: delayDuration, signal} = opts || {}; + const scrobbleDelay = delayDuration === undefined ? this.scrobbleDelay : (delayDuration === false ? 0 : delayDuration); if (scrobbleDelay !== 0) { const lastScrobbleDiff = dayjs().diff(this.lastScrobbleAttempt, 'ms'); const remainingDelay = scrobbleDelay - lastScrobbleDiff; if (remainingDelay > 0) { this.logger.debug(`Waiting ${remainingDelay}ms to scrobble so time passed since previous scrobble is at least ${scrobbleDelay}ms`); - await sleep(scrobbleDelay); + if(signal !== undefined) { + await delay(signal, scrobbleDelay); + } else { + await sleep(scrobbleDelay); + } + } } try { @@ -566,13 +573,32 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } } - this.startScrobbling().catch((e) => { - // do nothing, should have already been caught and logged + this.scrobbleQueueAbortController = new AbortController(); + this.scrobbleQueuePromise = spawn(this.scrobbleQueueAbortController.signal, async (signal, { defer, fork }) => { + + defer(async () => { + this.scrobbling = false; + this.emitEvent('statusChange', {status: 'Idle'}); + }); + + await this.startScrobbling(signal); + }).catch((e) => { + if (isAbortError(e)) { + const err = generateLoggableAbortReason('Scrobble processing stopped', this.scrobbleQueueAbortController.signal); + this.logger.info(err); + this.logger.trace(e) + } else { + this.logger.warn(new Error('Scrobble processing stopped with error', { cause: e })); + } + }).finally(() => { + this.scrobbleQueueAbortController = undefined; + this.scrobbleQueuePromise = undefined; }); - return; } - startScrobbling = async () => { + startScrobbling = async (signal: AbortSignal) => { + signal.throwIfAborted(); + // reset poll attempts if already previously run this.scrobbleRetries = 0; @@ -591,14 +617,13 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i return; } - let pollRes: boolean | undefined = undefined; - while (pollRes === undefined && this.scrobbleRetries <= maxRetries) { + while (this.scrobbleRetries <= maxRetries) { try { - pollRes = await this.doProcessing(); - if(pollRes === true) { - break; - } + await this.doProcessing(signal); } catch (e) { + if(isAbortError(e)) { + throw e; + } if(!this.isUsable()) { this.logger.warn('Stopping scrobble processing due to client no longer usable.'); await this.notify({title: `${this.getIdentifier()} - Processing Error`, message: `Encountered error while scrobble processing and client is no longer usable, stopping processing!. | Error: ${e.message}`, priority: 'error'}); @@ -621,38 +646,31 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } } - tryStopScrobbling = async () => { + tryStopScrobbling = async (reason?: string | Error) => { if(this.scrobbling === false) { this.logger.warn(`Polling is already stopped!`); return; } - this.userScrobblingStopSignal = true; + if(this.scrobbleQueueAbortController === undefined) { + this.logger.error('No abort controller found! Nothing to stop.'); + return false; + } + this.scrobbleQueueAbortController.abort(reason) let timePasssed = 0; - while(this.userScrobblingStopSignal !== undefined && timePasssed < (this.scrobbleWaitStopInterval * 10)) { + while(this.scrobbling === true && timePasssed < (this.scrobbleWaitStopInterval * 10)) { await sleep(this.scrobbleWaitStopInterval); timePasssed += this.scrobbleWaitStopInterval; this.logger.verbose(`Waiting for scrobble processing stop signal to be acknowledged (waited ${timePasssed}ms)`); } - if(this.userScrobblingStopSignal !== undefined) { + if(this.scrobbling === true) { this.logger.warn('Could not stop scrobble processing! Or signal was lost :('); return false; } return true; } - protected doStopScrobbling = (reason: string = 'system') => { - this.scrobbling = false; - this.userScrobblingStopSignal = undefined; - this.emitEvent('statusChange', {status: 'Idle'}); - this.logger.info(`Stopped scrobble processing due to: ${reason}`); - } - - protected shouldStopScrobbleProcessing = () => this.scrobbling === false || this.userScrobblingStopSignal !== undefined; - - protected doProcessing = async (): Promise => { - if (this.scrobbling === true) { - return true; - } + protected doProcessing = async (signal: AbortSignal): Promise => { + signal.throwIfAborted(); this.logger.info('Scrobble processing started'); this.emitEvent('statusChange', {status: 'Running'}); @@ -661,105 +679,134 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i if(!this.upstreamRefresh.refreshEnabled) { this.logger.verbose('Scrobble refresh is DISABLED. All queued scrobbles will likely always be scrobbled (nothing to check duplicates against).'); } - while (!this.shouldStopScrobbleProcessing()) { + while (true) { + signal.throwIfAborted(); let queueEmpty = this.queuedScrobbles.length === 0; while (this.queuedScrobbles.length > 0) { - this.handleQueuedScrobbleRanges(); - if(!this.upstreamRefresh.refreshEnabled) { - this.logger.trace('Scrobble refresh is DISABLED.'); - } - - const currQueuedPlay = this.queuedScrobbles.shift(); - - let historicalPlays: PlayObject[] = []; - let historicalError: Error | undefined; - - if(this.upstreamRefresh.refreshEnabled) { - try { - historicalPlays = await this.getSOTScrobblesForPlay(currQueuedPlay.play); - } catch (e) { - historicalError = e; - if(e.message === 'Cannot get historical plays due to cached error') { - this.logger.warn(`${buildTrackString(currQueuedPlay.play)} from Source '${currQueuedPlay.source}' => Previous error while getting historical scrobbles means this scrobble cannot be compared, will queue as dead for now.`); - this.logger.trace(e); - } else { - this.logger.warn(new SimpleError(`${buildTrackString(currQueuedPlay.play)} from Source '${currQueuedPlay.source}' => cannot get historical scrobbles, will queue as dead for now.`, {cause: e, shortStack: true})); - } - this.addDeadLetterScrobble(currQueuedPlay, e); - } - } - if(historicalError === undefined) { - const {summary, ...matchResult} = await this.existingScrobble(currQueuedPlay.play, historicalPlays); - const { - scrobble = {}, - ...lifeRest - } = currQueuedPlay.play.meta.lifecycle ?? {steps: [], original: currQueuedPlay.play}; - currQueuedPlay.play.meta.lifecycle = { - ...lifeRest, - scrobble: { - ...scrobble, - match: matchResult - } - } - if(!matchResult.match) { - const transformedScrobble = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.postCompare); - if(transformedScrobble.meta.lifecycle === undefined) { - transformedScrobble.meta.lifecycle = { - original: transformedScrobble, - steps: [] - }; - } - try { - const scrobbledPlay = await this.scrobble(transformedScrobble); - this.emitEvent('scrobble', {play: transformedScrobble}); - this.addScrobbledTrack(scrobbledPlay, scrobbledPlay.meta.lifecycle.scrobble.mergedScrobble ?? scrobbledPlay); - } catch (e) { - currQueuedPlay.play.meta.lifecycle.scrobble = { - }; - - const submitError = findCauseByReference(e, ScrobbleSubmitError); - if(submitError !== undefined) { - currQueuedPlay.play.meta.lifecycle.scrobble.payload = submitError.payload; - currQueuedPlay.play.meta.lifecycle.scrobble.response = submitError.responseBody; - currQueuedPlay.play.meta.lifecycle.scrobble.error = serializeError(submitError); - } else { - currQueuedPlay.play.meta.lifecycle.scrobble.payload = this.playToClientPayload(transformedScrobble); - currQueuedPlay.play.meta.lifecycle.scrobble.error = serializeError(e); - } - - if (hasUpstreamError(e, false)) { - this.addDeadLetterScrobble(currQueuedPlay, e); - this.logger.warn(new Error(`Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${currQueuedPlay.source}' but error was not show stopping. Adding scrobble to Dead Letter Queue and will retry on next heartbeat.`, {cause: e})); - } else { - this.queuedScrobbles.unshift(currQueuedPlay); - this.updateQueuedScrobblesCache(); - throw new Error('Error occurred while trying to scrobble', {cause: e}); - } - } - } - } - this.updateQueuedScrobblesCache(); - this.queuedGauge.labels(this.getPrometheusLabels()).set(this.queuedScrobbles.length); - this.emitEvent('scrobbleDequeued', {queuedScrobble: currQueuedPlay}) + await this.processQueueCurrentScrobble(signal); } if(!queueEmpty) { this.emitEvent('queueEmptied', {}); } - await sleep(this.scrobbleSleep); - } - if (this.shouldStopScrobbleProcessing()) { - this.doStopScrobbling(this.userScrobblingStopSignal !== undefined ? 'user input' : undefined); - return true; + await delay(signal, this.scrobbleSleep); } } catch (e) { - this.logger.error('Scrobble processing interrupted'); - this.logger.error(e); + if(!isAbortError(e)) { + this.logger.error('Scrobble processing interrupted'); + this.logger.error(e); + } this.emitEvent('statusChange', {status: 'Idle'}); this.scrobbling = false; throw e; } } + protected processQueueCurrentScrobble = async (signal: AbortSignal) => { + signal.throwIfAborted(); + if (this.queuedScrobbles.length === 0) { + return; + } + + this.handleQueuedScrobbleRanges(); + if (!this.upstreamRefresh.refreshEnabled) { + // TODO add signal for this to scrobble match + this.logger.trace('Scrobble refresh is DISABLED.'); + } + + let handledShiftedPlay = false; + const currQueuedPlay = this.queuedScrobbles.shift(); + + let historicalPlays: PlayObject[] = []; + let historicalError: Error | undefined; + + try { + + if (this.upstreamRefresh.refreshEnabled) { + try { + historicalPlays = await this.getSOTScrobblesForPlay(currQueuedPlay.play); + } catch (e) { + historicalError = e; + if (e.message === 'Cannot get historical plays due to cached error') { + this.logger.warn(`${buildTrackString(currQueuedPlay.play)} from Source '${currQueuedPlay.source}' => Previous error while getting historical scrobbles means this scrobble cannot be compared, will queue as dead for now.`); + this.logger.trace(e); + } else { + this.logger.warn(new SimpleError(`${buildTrackString(currQueuedPlay.play)} from Source '${currQueuedPlay.source}' => cannot get historical scrobbles, will queue as dead for now.`, { cause: e, shortStack: true })); + } + this.addDeadLetterScrobble(currQueuedPlay, e); + handledShiftedPlay = true; + } + signal.throwIfAborted(); + } + if (historicalError === undefined) { + const { summary, ...matchResult } = await this.existingScrobble(currQueuedPlay.play, historicalPlays); + signal.throwIfAborted(); + const { + scrobble = {}, + ...lifeRest + } = currQueuedPlay.play.meta.lifecycle ?? { steps: [], original: currQueuedPlay.play }; + currQueuedPlay.play.meta.lifecycle = { + ...lifeRest, + scrobble: { + ...scrobble, + match: matchResult + } + } + if (!matchResult.match) { + const transformedScrobble = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.postCompare); + signal.throwIfAborted(); + if (transformedScrobble.meta.lifecycle === undefined) { + transformedScrobble.meta.lifecycle = { + original: transformedScrobble, + steps: [] + }; + } + try { + const scrobbledPlay = await this.scrobble(transformedScrobble, {signal}); + this.emitEvent('scrobble', { play: transformedScrobble }); + this.addScrobbledTrack(scrobbledPlay, scrobbledPlay.meta.lifecycle.scrobble.mergedScrobble ?? scrobbledPlay); + handledShiftedPlay = true; + signal.throwIfAborted(); + } catch (e) { + currQueuedPlay.play.meta.lifecycle.scrobble = { + }; + + const submitError = findCauseByReference(e, ScrobbleSubmitError); + if (submitError !== undefined) { + currQueuedPlay.play.meta.lifecycle.scrobble.payload = submitError.payload; + currQueuedPlay.play.meta.lifecycle.scrobble.response = submitError.responseBody; + currQueuedPlay.play.meta.lifecycle.scrobble.error = serializeError(submitError); + } else { + currQueuedPlay.play.meta.lifecycle.scrobble.payload = this.playToClientPayload(transformedScrobble); + currQueuedPlay.play.meta.lifecycle.scrobble.error = serializeError(e); + } + + if (hasUpstreamError(e, false)) { + this.addDeadLetterScrobble(currQueuedPlay, e); + handledShiftedPlay = true; + this.logger.warn(new Error(`Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${currQueuedPlay.source}' but error was not show stopping. Adding scrobble to Dead Letter Queue and will retry on next heartbeat.`, { cause: e })); + } else { + this.queuedScrobbles.unshift(currQueuedPlay); + handledShiftedPlay = true; + this.updateQueuedScrobblesCache(); + throw new Error('Error occurred while trying to scrobble', { cause: e }); + } + } + } + } + this.updateQueuedScrobblesCache(); + this.queuedGauge.labels(this.getPrometheusLabels()).set(this.queuedScrobbles.length); + this.emitEvent('scrobbleDequeued', { queuedScrobble: currQueuedPlay }) + signal.throwIfAborted(); + // reset retries if we've made this far + this.scrobbleRetries = 0; + } catch (e) { + if(!handledShiftedPlay) { + this.queuedScrobbles.unshift(currQueuedPlay); + } + throw e; + } + } + processDeadLetterQueue = async (attemptWithRetries?: number) => { if (this.deadLetterScrobbles.length === 0) { diff --git a/src/backend/server/api.ts b/src/backend/server/api.ts index 5f14c80b..5247295e 100644 --- a/src/backend/server/api.ts +++ b/src/backend/server/api.ts @@ -489,7 +489,7 @@ export const setupApi = (app: ExpressWithAsync, logger: Logger, appLoggerStream: client.logger.verbose(`User requested${force ? ' a FORCED' :''} (re)init via API call`); client.logger.info('Checking (and trying) to stop scrobbler if already running...'); - if(false === (await client.tryStopScrobbling())) { + if(false === (await client.tryStopScrobbling(new SimpleError('user initiated', {simple: true, shortStack: true})))) { return res.status(500).send(); } diff --git a/src/backend/tests/scrobbler/scrobblers.test.ts b/src/backend/tests/scrobbler/scrobblers.test.ts index 1dc8e0cf..a49d17a0 100644 --- a/src/backend/tests/scrobbler/scrobblers.test.ts +++ b/src/backend/tests/scrobbler/scrobblers.test.ts @@ -525,7 +525,7 @@ describe('Upstream Scrobbles', function() { const play = generatePlay({playDate: dayjs().subtract(60, 's')}); await scrobbler.queueScrobble(play, 'test'); const emptied = pEvent(scrobbler.emitter, 'queueEmptied'); - scrobbler.startScrobbling().then(() => null); + scrobbler.startScrobbling(new AbortController().signal).then(() => null); await emptied; scrobbler.tryStopScrobbling().then(() => null); expect(sp.called).is.true; @@ -545,7 +545,7 @@ describe('Upstream Scrobbles', function() { const play2 = generatePlay({playDate: dayjs().subtract(1, 'm')}); await scrobbler.queueScrobble([play1, play2], 'test'); const emptied = pEvent(scrobbler.emitter, 'queueEmptied'); - scrobbler.startScrobbling().then(() => null); + scrobbler.startScrobbling(new AbortController().signal).then(() => null); await emptied; scrobbler.tryStopScrobbling().then(() => null); expect(sp.callCount).to.eq(1); @@ -566,7 +566,7 @@ describe('Upstream Scrobbles', function() { const play3 = generatePlay({playDate: dayjs().subtract(DEFAULT_CONSOLIDATE_DURATION.add(4, 'm'))}); await scrobbler.queueScrobble([play1, play2, play3], 'test'); const emptied = pEvent(scrobbler.emitter, 'queueEmptied'); - scrobbler.startScrobbling().then(() => null); + scrobbler.startScrobbling(new AbortController().signal).then(() => null); await emptied; scrobbler.tryStopScrobbling().then(() => null); expect(sp.callCount).to.eq(2); @@ -586,7 +586,7 @@ describe('Upstream Scrobbles', function() { const play2 = generatePlay({playDate: dayjs().subtract(1, 'm')}); await scrobbler.queueScrobble([play1], 'test'); const emptied = pEvent(scrobbler.emitter, 'queueEmptied'); - scrobbler.startScrobbling().then(() => null); + scrobbler.startScrobbling(new AbortController().signal).then(() => null); await emptied; expect(sp.calledOnce).is.true; -- 2.51.2 From 0607aedf04ed9f639189144c7e5eaa77a34e01be Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Thu, 2 Apr 2026 21:23:26 +0000 Subject: [PATCH 4/4] feat(musicbrainz): Handle rate limiting within api calls Instead of using staggerMap with MB transformer settings, handle at api call level for each host --- src/backend/common/infrastructure/Atomic.ts | 1 - .../transforms/MusicbrainzTransformer.ts | 4 +-- .../musicbrainz/MusicbrainzApiClient.ts | 32 ++++++++++++++++--- 3 files changed, 30 insertions(+), 7 deletions(-) diff --git a/src/backend/common/infrastructure/Atomic.ts b/src/backend/common/infrastructure/Atomic.ts index d8ae64a7..a0f1dee9 100644 --- a/src/backend/common/infrastructure/Atomic.ts +++ b/src/backend/common/infrastructure/Atomic.ts @@ -291,7 +291,6 @@ export interface CacheConfigOptions { export interface MusicbrainzApiConfigData { url?: string - rateLimit?: [number, number] contact: string, apiKey?: string requestTimeout?: number diff --git a/src/backend/common/transforms/MusicbrainzTransformer.ts b/src/backend/common/transforms/MusicbrainzTransformer.ts index a288054f..e9749cb2 100644 --- a/src/backend/common/transforms/MusicbrainzTransformer.ts +++ b/src/backend/common/transforms/MusicbrainzTransformer.ts @@ -356,8 +356,8 @@ export default class MusicbrainzTransformer extends AtomicPartsTransformer = { + ...mbConfig, + hostname: u.url.hostname, + minRequestIntervalDuration: 1000, + lastRequest: dayjs().subtract(1000 + 1000, 'ms') + } if(mb === undefined) { const api = new MusicBrainzApi({ appName: 'multi-scrobbler', appVersion: version, appContactInfo: mbConfig.contact, baseUrl: u.url.toString(), - rateLimit: mbConfig.rateLimit ?? [1,1], preRequest: options.logUrl === true || isDebugMode() ? (method, url, headers) => { const cacheKey = this.asyncStore.getStore() ?? nanoid(); this.cache.set(`${cacheKey}-url`, `${method} - ${url}`); @@ -84,11 +91,17 @@ export class MusicbrainzApiClient extends AbstractApiClient { requestTimeout: mbConfig.requestTimeout ?? 6000, retryLimit: 2 }); - mbApis[u.url.hostname] = {api, ...mbConfig, hostname: u.url.hostname}; + mbApis[u.url.hostname] = { + ...mbApiConfig, + api, + }; mbMap.set(u.url.hostname, api); mb = api; } else if(mbApis[u.url.hostname] === undefined) { - mbApis[u.url.hostname] = {api: mb, ...mbConfig, hostname: u.url.hostname}; + mbApis[u.url.hostname] = { + ...mbApiConfig, + api: mb, + }; } } @@ -127,6 +140,17 @@ export class MusicbrainzApiClient extends AbstractApiClient { const triedHosts: string[] = []; while(!triedHosts.includes(apiConfig.hostname)) { + // keep track of last request init at and wait until at least 1 second since that + // to help prevent rate limiting + let waitTime = 0; + const sinceLast = dayjs().diff(apiConfig.lastRequest, 'ms'); + waitTime = Math.max(0, apiConfig.minRequestIntervalDuration - sinceLast); + apiConfig.lastRequest = dayjs().add(waitTime, 'ms'); + //this.logger.trace(`Waiting ${waitTime}ms to call ${apiConfig.hostname} request at ${apiConfig.lastRequest.toISOString()}`) + if(waitTime > 0) { + await sleep(waitTime); + } + try { const res = await this.callApiEndpoint(apiConfig.api, func, options); if(cacheKey !== undefined) {