Fix Critical Memory Leak in Session Replay and Refactor for Better Error Handling (#1328)

Author: matux

src/browser/replay/recorder.js

@@ -68,39 +68,36 @@ export default class Recorder {
   checkoutEveryNms() {
     // Recording may be up to two checkout intervals, therefore the checkout
     // interval is set to half of the maxSeconds.
-    return (this.options.maxSeconds || 10) * 1000 / 2;
+    return ((this.options.maxSeconds || 10) * 1000) / 2;
   }
 
   /**
-   * Converts recorded events into a formatted payload ready for transport.
+   * Exports the recording span with all recorded events.
    *
    * This method takes the recorder's stored events, creates a new span with the
    * provided tracing context, attaches all events with their timestamps as span
-   * events, and then returns a payload ready for transport to the server.
+   * events, and exports the span to the tracing exporter. This is a side-effect
+   * function that doesn't return anything - the span is exported internally.
    *
    * @param {Object} tracing - The tracing system instance to create spans
-   * @param {string} replayId - Unique identifier to associate with this replay recording
-   * @returns {Object|null} A formatted payload containing spans data in OTLP format, or null if no events exist
+   * @param {Object} attributes - Attributes to add to the span
+   *  (e.g., rollbar.replay.id, rollbar.occurrence.uuid)
    */
-  dump(tracing, replayId, occurrenceUuid) {
-    const events = this._events.previous.concat(this._events.current);
+  exportRecordingSpan(tracing, attributes = {}) {
+    const events = this._collectEvents();
 
-    if (events.length < 2) {
-      logger.error('Replay recording cannot have less than 2 events');
-      return null;
+    if (events.length < 3) {
+      // TODO(matux): improve how we consider a recording valid
+      throw new Error('Replay recording cannot have less than 3 events');
     }
 
     const recordingSpan = tracing.startSpan('rrweb-replay-recording', {});
 
     recordingSpan.setAttributes({
       ...(tracing.session?.attributes ?? {}),
-      'rollbar.replay.id': replayId
+      ...attributes,
     });
 
-    if (occurrenceUuid) {
-      recordingSpan.setAttribute('rollbar.occurrence.uuid', occurrenceUuid);
-    }
-
     const earliestEvent = events.reduce((earliestEvent, event) =>
       event.timestamp < earliestEvent.timestamp ? event : earliestEvent,
     );
@@ -118,11 +115,7 @@ export default class Recorder {
       );
     }
 
-    this._addEndEvent(recordingSpan, replayId);
-
     recordingSpan.end();
-
-    return tracing.exporter.toPayload();
   }
 
   start() {
@@ -176,6 +169,20 @@ export default class Recorder {
     };
   }
 
+  _collectEvents() {
+    const events = this._events.previous.concat(this._events.current);
+
+    // Helps the application correctly align playback by adding a noop event
+    // to the end of the recording.
+    events.push({
+      timestamp: Date.now(),
+      type: EventType.Custom,
+      data: { tag: 'replay.end', payload: {} },
+    });
+
+    return events;
+  }
+
   _logEvent(event, isCheckout) {
     logger.log(
       `Recorder: ${isCheckout ? 'checkout' : ''} event\n`,
@@ -195,19 +202,4 @@ export default class Recorder {
       })(event),
     );
   }
-
-  /**
-   * Helps the application correctly align playback by adding a noop event
-   * to the end of the recording.
-   **/
-  _addEndEvent(recordingSpan, replayId) {
-    recordingSpan.addEvent(
-      'rrweb-replay-events',
-      {
-        eventType: 5,
-        json: JSON.stringify({tag: "replay.end", payload: {}}),
-      },
-      hrtime.fromMillis(Date.now()),
-    );
-  }
 }

src/browser/replay/replayManager.js

@@ -42,33 +42,31 @@ export default class ReplayManager {
   }
 
   /**
-   * Processes a replay by converting recorder events into a transport-ready payload.
+   * Exports recording and telemetry spans, then stores the tracing payload in the map.
    *
-   * Calls recorder.dump() to capture events as spans, formats them into a proper payload,
-   * and stores the result in the map using replayId as the key.
+   * Exports both telemetry and recording spans, then generates the complete payload
+   * using the tracing exporter and stores it in the map using replayId as the key.
+   * This is an async operation that runs in the background.
    *
    * @param {string} replayId - The unique ID for this replay
-   * @returns {Promise<string>} A promise resolving to the processed replayId
+   * @param {string} occurrenceUuid - The UUID of the associated error occurrence
    * @private
    */
-  async _processReplay(replayId, occurrenceUuid) {
+  async _exportSpansAndAddTracingPayload(replayId, occurrenceUuid) {
     try {
-      this._telemeter?.exportTelemetrySpan({ 'rollbar.replay.id': replayId });
-
-      const payload = this._recorder.dump(
-        this._tracing,
-        replayId,
-        occurrenceUuid,
-      );
-
-      this._map.set(replayId, payload);
-    } catch (transformError) {
-      logger.error('Error transforming spans:', transformError);
-
-      this._map.set(replayId, null); // TODO(matux): Error span?
+      this._recorder.exportRecordingSpan(this._tracing, {
+        'rollbar.replay.id': replayId,
+        'rollbar.occurrence.uuid': occurrenceUuid,
+      });
+    } catch (error) {
+      logger.error('Error exporting recording span:', error);
+      return;
     }
 
-    return replayId;
+    this._telemeter?.exportTelemetrySpan({ 'rollbar.replay.id': replayId });
+
+    const payload = this._tracing.exporter.toPayload();
+    this._map.set(replayId, payload);
   }
 
   /**
@@ -83,9 +81,8 @@ export default class ReplayManager {
   add(replayId, occurrenceUuid) {
     replayId = replayId || id.gen(8);
 
-    this._processReplay(replayId, occurrenceUuid).catch((error) => {
-      logger.error('Failed to process replay:', error);
-    });
+    // Start processing the replay in the background
+    this._exportSpansAndAddTracingPayload(replayId, occurrenceUuid);
 
     return replayId;
   }
@@ -103,15 +100,11 @@ export default class ReplayManager {
    */
   async send(replayId) {
     if (!replayId) {
-      logger.error('ReplayManager.send: No replayId provided');
-      return false;
+      throw Error('ReplayManager.send: No replayId provided');
     }
 
     if (!this._map.has(replayId)) {
-      logger.error(
-        `ReplayManager.send: No replay found for replayId: ${replayId}`,
-      );
-      return false;
+      throw Error(`ReplayManager.send: No replay found for id: ${replayId}`);
     }
 
     const payload = this._map.get(replayId);
@@ -124,19 +117,10 @@ export default class ReplayManager {
       (payload.resourceSpans && payload.resourceSpans.length === 0);
 
     if (isEmpty) {
-      logger.error(
-        `ReplayManager.send: No payload found for replayId: ${replayId}`,
-      );
-      return false;
+      throw Error(`ReplayManager.send: No payload found for id: ${replayId}`);
     }
 
-    try {
-      await this._api.postSpans(payload, { 'X-Rollbar-Replay-Id': replayId });
-      return true;
-    } catch (error) {
-      logger.error('Error sending replay:', error);
-      return false;
-    }
+    await this._api.postSpans(payload, { 'X-Rollbar-Replay-Id': replayId });
   }
 
   /**

src/queue.js

@@ -102,12 +102,10 @@ class Queue {
     }
 
     if (this.replayManager && data.body) {
-      const replayId = data?.attributes?.find(
-        (a) => a.key === 'replay_id',
-      )?.value;
+      const rid = data.attributes?.find((a) => a.key === 'replay_id')?.value;
 
-      if (replayId) {
-        item.replayId = this.replayManager.add(replayId, data.uuid);
+      if (rid) {
+        item.replayId = this.replayManager.add(rid, data.uuid);
       }
     }
 
@@ -116,15 +114,20 @@ class Queue {
       this._makeApiRequest(data, (err, resp, headers) => {
         this._dequeuePendingRequest(data);
 
-        if (!err && resp && item.replayId) {
-          this._handleReplayResponse(item.replayId, resp, headers);
+        if (item.replayId) {
+          this._sendOrDiscardReplay(item.replayId, err, resp, headers);
         }
 
         callback(err, resp);
       });
-    } catch (e) {
+    } catch (err) {
       this._dequeuePendingRequest(data);
-      callback(e);
+
+      if (item.replayId) {
+        this.replayManager?.discard(item.replayId);
+      }
+
+      callback(err);
     }
   }
 
@@ -305,51 +308,54 @@ class Queue {
   }
 
   /**
-   * Handles the API response for an item with a replay ID.
-   * Based on the success or failure status of the response,
-   * it either sends or discards the associated session replay.
+   * Determines if a replay can be sent based on the API response.
+   *
+   * Returns true only when all conditions are met:
+   * - No error occurred during the API request
+   * - The response indicates success (err === 0)
+   * - Replay is enabled on the server
+   * - Rate limit quota is not exhausted
+   *
+   * @param {Error|null} err - Error from the API request, if any
+   * @param {Object} response - The API response object
+   * @param {Object} headers - Response headers from the API
+   * @returns {boolean} True if the replay can be sent, false otherwise
+   */
+  _canSendReplay(err, response, headers) {
+    return (
+      !err &&
+      response?.err === 0 &&
+      headers?.['Rollbar-Replay-Enabled'] === 'true' &&
+      headers?.['Rollbar-Replay-RateLimit-Remaining'] !== '0'
+    );
+  }
+
+  /**
+   * Sends or discards a replay based on the API response.
+   *
+   * Sends the replay if the response indicates success and replay is enabled.
+   *
+   * Discards the replay if there's an error, replay is disabled, or rate limit
+   * is exceeded.
    *
-   * @param {string} replayId - The ID of the replay to handle
+   * @param {string} replayId - The ID of the replay to send or discard
+   * @param {Object} err - The error object from the API request, if any
    * @param {Object} response - The API response
    * @param {Object} headers - The response headers
-   * @returns {Promise<boolean>} A promise that resolves to true if replay was sent successfully,
-   *                             false if replay was discarded or an error occurred
    */
-  async _handleReplayResponse(replayId, response, headers) {
-    if (!this.replayManager) {
-      console.warn('Queue._handleReplayResponse: ReplayManager not available');
-      return false;
-    }
-
-    if (!replayId) {
-      console.warn('Queue._handleReplayResponse: No replayId provided');
-      return false;
-    }
+  async _sendOrDiscardReplay(replayId, err, response, headers) {
+    const canSendReplay = this._canSendReplay(err, response, headers);
 
-    try {
-      if (this._shouldSendReplay(response, headers)) {
-        return await this.replayManager.send(replayId);
-      } else {
+    if (canSendReplay) {
+      try {
+        await this.replayManager.send(replayId);
+      } catch (error) {
+        console.error('Failed to send replay:', error);
         this.replayManager.discard(replayId);
-        return false;
       }
-    } catch (error) {
-      console.error('Error handling replay response:', error);
-      return false;
-    }
-  }
-
-  _shouldSendReplay(response, headers) {
-    if (
-      response?.err !== 0 ||
-      !headers ||
-      headers['Rollbar-Replay-Enabled'] !== 'true' ||
-      headers['Rollbar-Replay-RateLimit-Remaining'] === '0'
-    ) {
-      return false;
+    } else {
+      this.replayManager.discard(replayId);
     }
-
-    return true;
   }
 }