diff --git a/src/index.ts b/src/index.ts index f7f374e0..2d0af722 100644 --- a/src/index.ts +++ b/src/index.ts @@ -81,7 +81,12 @@ import type { import type { IPCMessage } from './typings/shared.types.ts' import type { SourceInstance } from './typings/sources/source.types.ts' import type { VoiceRelay } from './typings/voice/voice.types.ts' -import { getGitInfo, getVersion, logger } from './utils.ts' +import { + getGitInfo, + getVersion, + logger, + queueSessionEvent +} from './utils.ts' import { parseVoiceFrameHeader } from './voice/voiceFrames.ts' import { createVoiceRelay } from './voice/voiceRelay.ts' @@ -402,12 +407,10 @@ class NodelinkServer extends EventEmitter { const session = this.sessions.get(sessionId) if (!session) return - if (session.isPaused && session.resuming) { - session.eventQueue.push(data) - return + const queued = queueSessionEvent(session, data) + if (!queued && !session.socket?.destroyed) { + session.socket?.send(data) } - - session.socket?.send(data) } private _handleWorkerStats( diff --git a/src/managers/playerManager.ts b/src/managers/playerManager.ts index e94bb329..a9cce5fd 100644 --- a/src/managers/playerManager.ts +++ b/src/managers/playerManager.ts @@ -490,7 +490,8 @@ export default class PlayerManager { userId, socket: session.socket, eventQueue: session.eventQueue, - isPaused: session.isPaused + isPaused: session.isPaused, + resuming: session.resuming } const { Player } = await import('../playback/player.ts') diff --git a/src/managers/sessionManager.ts b/src/managers/sessionManager.ts index 92ad8715..17eae0d6 100644 --- a/src/managers/sessionManager.ts +++ b/src/managers/sessionManager.ts @@ -7,7 +7,11 @@ import type { SessionSocket } from '../typings/index.types.ts' import type { ClientInfo } from '../typings/shared.types.ts' -import { generateRandomLetters, logger } from '../utils.ts' +import { + clearSessionEventQueue, + generateRandomLetters, + logger +} from '../utils.ts' import GroupManager from './groupManager.ts' /** @@ -154,6 +158,13 @@ export default class SessionManager { session.socket = null this.resumableSessions.set(sessionId, session) + // [feat] session-resuming: clear any stale resume timer before scheduling + // a new one, so a leftover timeout from a previous pause cannot destroy + // the session early or leak the handle. + if (session.timeoutFuture) { + clearTimeout(session.timeoutFuture) + } + session.timeoutFuture = setTimeout(() => { logger( 'info', @@ -241,6 +252,7 @@ export default class SessionManager { } session.groups.destroy() + clearSessionEventQueue(session) session.socket?.destroy?.() } diff --git a/src/playback/player.ts b/src/playback/player.ts index bdaf93db..a35c4866 100644 --- a/src/playback/player.ts +++ b/src/playback/player.ts @@ -40,7 +40,7 @@ import type { TrackInfoExtended } from '../typings/playback/player.types.ts' import type { TrackUrlResult } from '../typings/sources/source.types.ts' -import { logger } from '../utils.ts' +import { logger, queueSessionEvent } from '../utils.ts' import { AutoMixRegistry } from './processing/AutoMixRegistry.ts' import { DuckingController } from './processing/DuckingController.ts' @@ -249,12 +249,20 @@ export class Player { }) if (this.session.isPaused) { - this.session.eventQueue.push(eventData) - logger( - 'debug', - 'Player', - `Queued event ${type} for paused session ${this.session.id}` - ) + const queued = queueSessionEvent(this.session, eventData) + if (queued) { + logger( + 'debug', + 'Player', + `Queued event ${type} for paused session ${this.session.id}` + ) + } else { + logger( + 'debug', + 'Player', + `Dropped event ${type} for paused session ${this.session.id} (not resumable)` + ) + } return } diff --git a/src/server/wsSession.ts b/src/server/wsSession.ts index 18d26b78..44c4b16a 100644 --- a/src/server/wsSession.ts +++ b/src/server/wsSession.ts @@ -8,7 +8,7 @@ import type { SessionSocket } from '../typings/index.types.ts' import type { ClientInfo } from '../typings/shared.types.ts' -import { logger } from '../utils.ts' +import { clearSessionEventQueue, logger } from '../utils.ts' /* INFO: Formats human-readable client name and version tag for logs */ function _formatClientTag(clientInfo: ClientInfo): string { @@ -165,7 +165,7 @@ function _resumeSession( for (const event of session.eventQueue) { socket.send(event) } - session.eventQueue.length = 0 + clearSessionEventQueue(session) _syncResumedPlayers(context, session) return true diff --git a/src/typings/playback/player.types.ts b/src/typings/playback/player.types.ts index cb6c0f0f..af98bfe5 100644 --- a/src/typings/playback/player.types.ts +++ b/src/typings/playback/player.types.ts @@ -501,6 +501,12 @@ export interface Session { userId: string socket: { send: (data: string) => void } isPaused: boolean + /** + * Whether the session is resumable while paused. When true, events are + * buffered into `eventQueue` for replay on resume; when false or + * undefined, they are dropped while paused. + */ + resuming?: boolean eventQueue: string[] } diff --git a/src/utils.ts b/src/utils.ts index 5fe08c3e..e5669aaf 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -24,6 +24,7 @@ import type { ApiResponse } from './typings/api/api.types.ts' import type { ClientInfo } from './typings/shared.types.ts' +import type { Session } from './typings/index.types.ts' import type { BestMatchCandidate, BestMatchOptions, @@ -1216,6 +1217,78 @@ const generateRandomLetters = (length: number): string => String.fromCharCode((b % 52) + (b % 52 < 26 ? 65 : 71)) ).join('') +/** + * Maximum number of events kept in a resumable session's event queue. + * + * While a session is paused, every player event is buffered so it can be + * replayed on resume. Without a cap, a client that never returns while its + * players keep playing would grow this queue without bound (memory leak). + */ +const MAX_SESSION_EVENT_QUEUE_SIZE = 500 + +/** + * Buffers a player event for a paused, resumable session with a hard cap. + * + * Events are only queued when the session is actually paused for resumption. + * When the queue is full, the oldest event is dropped to make room + * (drop-oldest keeps the newest state, which matters most on resume). + * + * @param session - Session that should buffer the event. + * @param data - Serialized event payload (JSON string). + * @returns True when the event was queued, false when it was sent live. + * @public + */ +function queueSessionEvent( + session: Pick, + data: string +): boolean { + if (!session.isPaused || !session.resuming) return false + + const queue = session.eventQueue + if (queue.length >= MAX_SESSION_EVENT_QUEUE_SIZE) { + queue.shift() + logger( + 'warn', + 'SessionManager', + `Event queue full for paused session ${session.id} (cap ${MAX_SESSION_EVENT_QUEUE_SIZE}), dropping oldest events` + ) + } else if ( + queue.length === Math.floor(MAX_SESSION_EVENT_QUEUE_SIZE / 2) || + queue.length === MAX_SESSION_EVENT_QUEUE_SIZE - 1 + ) { + logger( + 'warn', + 'SessionManager', + `Event queue for paused session ${session.id} at ${queue.length + 1}/${MAX_SESSION_EVENT_QUEUE_SIZE} events` + ) + } + + queue.push(data) + return true +} + +/** + * Clears a session's event queue, logging how many events were discarded. + * + * @param session - Session whose queue should be cleared. + * @public + */ +function clearSessionEventQueue( + session: Pick +): void { + const drained = session.eventQueue.length + if (!drained) return + + session.eventQueue.length = 0 + logger( + 'debug', + 'SessionManager', + `Cleared ${drained} queued event(s) for session ${session.id}` + ) +} + +/** + * Parses the `Client-Name` header into a structured object. /** * Parses the `Client-Name` header into a structured object. * @@ -2825,6 +2898,8 @@ export { encodeTrack, fetchSponsorBlockSegments, generateRandomLetters, + queueSessionEvent, + clearSessionEventQueue, getGitInfo, getStats, getVersion,