fix: cap session event queue and clear stale resume timers

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(

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')

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?.()
   }
 

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
       }
 

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

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[]
 }
 

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<Session, 'id' | 'isPaused' | 'resuming' | 'eventQueue'>,
+  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<Session, 'id' | 'eventQueue'>
+): 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,