Fix Video Stream Memory Leak

[Shubhrakanti]

Issue

A memory leak was identified in the VideoStream (and AudioStream) class where the Promise resolver for the async iterator was not being properly reset after use. This occurred in the onEvent method when processing incoming video frames from the LiveKit server. When a frame is received and a queueResolve function exists (indicating a waiting iterator consumer), the resolver was being called but the reference was not cleared afterward. If a new Frame arrives before we call next(), we call resolver on the stale reference learning to:

1. A memory leak.
2. A bug where frames coming in faster than calls to next() are not actually being stored in the queue. There being tossed into a stale resolver.

We can solve this by just simply clearing the reference to the resolver after it is called.

Testing

Wrote a test script that consume video from a LK server.

• Before fix: 2210MB usage after 2 mins
• After fix: 818MB usage after 2 mins

[changeset-bot]

🦋 Changeset detected

Latest commit: 61763fe

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 6 packages

Name	Type
@livekit/rtc-node	Minor
@livekit/rtc-node-darwin-arm64	Minor
@livekit/rtc-node-darwin-x64	Minor
@livekit/rtc-node-linux-arm64-gnu	Minor
@livekit/rtc-node-linux-x64-gnu	Minor
@livekit/rtc-node-win32-x64-msvc	Minor

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[davidzhao]

nice find! this makes sense.

could you do me a favor and:

• indicate in the PR description how this leaks, and share any test results
• also fix the same for AudioStream

[Shubhrakanti]

I added some log statements that helped me debug the issue. For future reference.

VideoStream: Created new instance
⚠️ Running in MEMORY LEAK demonstration mode ⚠️
Starting to process video frames...
VideoStream.next() called at 1743294654037, queue size: 0
VideoStream.next() acquired mutex at 1743294654037
VideoStream.next() no items in queue, creating promise and setting queueResolve
VideoStream.next() queueResolve set at 1743294654038
VideoStream.next() returning promise, waiting for resolution
[Frame #1] Received frame at 1743294654118, queueResolve exists: true, queue size: 0
[Frame #1] Resolving directly via queueResolve (#1)
VideoStream.next() called at 1743294654169, queue size: 0
VideoStream.next() acquired mutex at 1743294654170
VideoStream.next() no items in queue, creating promise and setting queueResolve
VideoStream.next() queueResolve set at 1743294654170
VideoStream.next() returning promise, waiting for resolution
[Frame #2] Received frame at 1743294654174, queueResolve exists: true, queue size: 0
[Frame #2] Resolving directly via queueResolve (#2)
VideoStream.next() called at 1743294654226, queue size: 0
VideoStream.next() acquired mutex at 1743294654226
VideoStream.next() no items in queue, creating promise and setting queueResolve
VideoStream.next() queueResolve set at 1743294654226
VideoStream.next() returning promise, waiting for resolution
[Frame #3] Received frame at 1743294654230, queueResolve exists: true, queue size: 0
[Frame #3] Resolving directly via queueResolve (#3)
VideoStream.next() called at 1743294654282, queue size: 0
VideoStream.next() acquired mutex at 1743294654282
VideoStream.next() no items in queue, creating promise and setting queueResolve
VideoStream.next() queueResolve set at 1743294654282
VideoStream.next() returning promise, waiting for resolution
[Frame #4] Received frame at 1743294654315, queueResolve exists: true, queue size: 0
[Frame #4] Resolving directly via queueResolve (#4)
VideoStream.next() called at 1743294654366, queue size: 0
VideoStream.next() acquired mutex at 1743294654366
VideoStream.next() no items in queue, creating promise and setting queueResolve
VideoStream.next() queueResolve set at 1743294654366
VideoStream.next() returning promise, waiting for resolution
[Frame #5] Received frame at 1743294654393, queueResolve exists: true, queue size: 0
[Frame #5] Resolving directly via queueResolve (#5)
[Frame #6] Received frame at 1743294654417, queueResolve exists: true, queue size: 0
[Frame #6] Resolving directly via queueResolve (#6) *INCORRECT BEHAVIOR*

This is wrong we should be adding Frame 6 to the queue instead of passing it to the resolver that we created for Frame 5

[Shubhrakanti]

@davidzhao I wonder if it's cleaner to try and abstract away the Async Queue part of the logic into a separate class. Something like

export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
  private asyncQueue = new AsyncQueue<VideoFrameEvent | null>();
  
  constructor(track: Track) {
    // Setup code...
    
    FfiClient.instance.on(FfiClientEvent.FfiEvent, (ev) => {
      if (ev.message.case === 'videoStreamEvent' && 
          ev.message.value.streamHandle === this.ffiHandle.handle) {
        
        if (ev.message.value.message.case === 'frameReceived') {
          // Process frame and add to queue
          const frameEvent = this.processFrame(ev.message.value.message.value);
          this.asyncQueue.push(frameEvent);
        } else if (ev.message.value.message.case === 'eos') {
          this.asyncQueue.push(null); // Signal end of stream
          FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
        }
      }
    });
  }

  async next(): Promise<IteratorResult<VideoFrameEvent>> {
    const item = await this.asyncQueue.next();
    return item === null ? 
      { done: true, value: undefined } : 
      { done: false, value: item };
  }
}

[davidzhao]

@davidzhao I wonder if it's cleaner to try and abstract away the Async Queue part of the logic into a separate class.

yeah! I think that's a good idea.. this pattern will come up elsewhere too

[Shubhrakanti]

@davidzhao I wonder if it's cleaner to try and abstract away the Async Queue part of the logic into a separate class.

yeah! I think that's a good idea.. this pattern will come up elsewhere too

cool I'll do in a follow up PR.