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