feat: Adjust SR Timestamps to NR Server Time (#939)

Author: metal-messiah

src/features/session_replay/aggregate/index.js

@@ -50,6 +50,8 @@ export class Aggregate extends AggregateBase {
 
     /** set by BCS response */
     this.entitled = false
+    /** set at BCS response, stored in runtime */
+    this.timeKeeper = undefined
 
     this.recorder = args?.recorder
     if (this.recorder) this.recorder.parent = this
@@ -179,7 +181,8 @@ export class Aggregate extends AggregateBase {
     // we are not actively recording SR... DO NOT import or run the recording library
     // session replay samples can only be decided on the first load of a session
     // session replays can continue if already in progress
-    const { session } = getRuntime(this.agentIdentifier)
+    const { session, timeKeeper } = getRuntime(this.agentIdentifier)
+    this.timeKeeper = timeKeeper
     if (!session.isNew && !ignoreSession) { // inherit the mode of the existing session
       this.mode = session.state.sessionReplayMode
     } else {
@@ -241,7 +244,7 @@ export class Aggregate extends AggregateBase {
   }
 
   prepareHarvest ({ opts } = {}) {
-    if (!this.recorder) return
+    if (!this.recorder || !this.timeKeeper?.ready) return
     const recorderEvents = this.recorder.getEvents()
     // get the event type and use that to trigger another harvest if needed
     if (!recorderEvents.events.length || (this.mode !== MODE.FULL) || this.blocked) return
@@ -254,7 +257,10 @@ export class Aggregate extends AggregateBase {
 
     let len = 0
     if (!!this.gzipper && !!this.u8) {
-      payload.body = this.gzipper(this.u8(`[${payload.body.map(e => e.__serialized).join(',')}]`))
+      payload.body = this.gzipper(this.u8(`[${payload.body.map(e => {
+        if (e.__serialized) return e.__serialized
+        return stringify(e)
+      }).join(',')}]`))
       len = payload.body.length
       this.scheduler.opts.gzip = true
     } else {
@@ -300,13 +306,12 @@ export class Aggregate extends AggregateBase {
       recorderEvents.hasMeta = !!events.find(x => x.type === RRWEB_EVENT_TYPES.Meta)
     }
 
-    const agentOffset = getRuntime(this.agentIdentifier).offset
     const relativeNow = now()
 
     const firstEventTimestamp = events[0]?.timestamp // from rrweb node
     const lastEventTimestamp = events[events.length - 1]?.timestamp // from rrweb node
-    const firstTimestamp = firstEventTimestamp || recorderEvents.cycleTimestamp // from rrweb node || from when the harvest cycle started
-    const lastTimestamp = lastEventTimestamp || agentOffset + relativeNow
+    const firstTimestamp = firstEventTimestamp || this.timeKeeper.correctAbsoluteTimestamp(recorderEvents.cycleTimestamp) // from rrweb node || from when the harvest cycle started
+    const lastTimestamp = lastEventTimestamp || this.timeKeeper.convertRelativeTimestamp(relativeNow)
 
     const agentMetadata = agentRuntime.appMetadata?.agents?.[0] || {}
     return {
@@ -315,15 +320,14 @@ export class Aggregate extends AggregateBase {
         type: 'SessionReplay',
         app_id: info.applicationID,
         protocol_version: '0',
+        timestamp: firstTimestamp,
         attributes: encodeObj({
           // this section of attributes must be controllable and stay below the query param padding limit -- see QUERY_PARAM_PADDING
           // if not, data could be lost to truncation at time of sending, potentially breaking parsing / API behavior in NR1
           ...(!!this.gzipper && !!this.u8 && { content_encoding: 'gzip' }),
           ...(agentMetadata.entityGuid && { entityGuid: agentMetadata.entityGuid }),
           'replay.firstTimestamp': firstTimestamp,
-          'replay.firstTimestampOffset': firstTimestamp - agentOffset,
           'replay.lastTimestamp': lastTimestamp,
-          'replay.durationMs': lastTimestamp - firstTimestamp,
           'replay.nodes': events.length,
           'session.durationMs': agentRuntime.session.getDuration(),
           agentVersion: agentRuntime.version,

src/features/session_replay/shared/recorder-events.js

@@ -1,11 +1,16 @@
 export class RecorderEvents {
-  constructor () {
+  constructor ({ canCorrectTimestamps }) {
     /** The buffer to hold recorder event nodes */
     this.events = []
     /** Payload metadata -- Should indicate when a replay blob started recording.  Resets each time a harvest occurs.
      * cycle timestamps are used as fallbacks if event timestamps cannot be used
      */
     this.cycleTimestamp = Date.now()
+    /** Payload metadata -- Whether timestamps can be corrected, defaults as false, can be set to true if timekeeper is present at init time.  Used to determine
+     * if harvest needs to re-loop through nodes and correct them before sending.  Ideal behavior is to correct them as they flow into the recorder
+     * to prevent re-looping, but is not always possible since the timekeeper is not set until after page load and the recorder can be preloaded.
+    */
+    this.canCorrectTimestamps = !!canCorrectTimestamps
     /** A value which increments with every new mutation node reported. Resets after a harvest is sent */
     this.payloadBytesEstimation = 0
     /** Payload metadata -- Should indicate that the payload being sent has a full DOM snapshot. This can happen

src/features/session_replay/shared/recorder.js

@@ -11,15 +11,18 @@ import { FEATURE_NAMES } from '../../../loaders/features/features'
 
 export class Recorder {
   /** Each page mutation or event will be stored (raw) in this array. This array will be cleared on each harvest */
-  #events = new RecorderEvents()
+  #events
   /** Backlog used for a 2-part sliding window to guarantee a 15-30s buffer window */
-  #backloggedEvents = new RecorderEvents()
+  #backloggedEvents
   /** array of recorder events -- Will be filled only if forced harvest was triggered and harvester does not exist */
-  #preloaded = [new RecorderEvents()]
+  #preloaded
   /** flag that if true, blocks events from being "stored".  Only set to true when a full snapshot has incomplete nodes (only stylesheets ATM) */
   #fixing = false
 
   constructor (parent) {
+    this.#events = new RecorderEvents({ canCorrectTimestamps: !!parent.timeKeeper?.ready })
+    this.#backloggedEvents = new RecorderEvents({ canCorrectTimestamps: !!parent.timeKeeper?.ready })
+    this.#preloaded = [new RecorderEvents({ canCorrectTimestamps: !!parent.timeKeeper?.ready })]
     /** True when actively recording, false when paused or stopped */
     this.recording = false
     /** The pointer to the current bucket holding rrweb events */
@@ -37,9 +40,14 @@ export class Recorder {
   }
 
   getEvents () {
-    if (this.#preloaded[0]?.events.length) return { ...this.#preloaded[0], type: 'preloaded' }
+    if (this.#preloaded[0]?.events.length) {
+      const preloadedEvents = this.returnCorrectTimestamps(this.#preloaded[0])
+      return { ...this.#preloaded[0], events: preloadedEvents, type: 'preloaded' }
+    }
+    const backloggedEvents = this.returnCorrectTimestamps(this.#backloggedEvents)
+    const events = this.returnCorrectTimestamps(this.#events)
     return {
-      events: [...this.#backloggedEvents.events, ...this.#events.events].filter(x => x),
+      events: [...backloggedEvents, ...events].filter(x => x),
       type: 'standard',
       cycleTimestamp: Math.min(this.#backloggedEvents.cycleTimestamp, this.#events.cycleTimestamp),
       payloadBytesEstimation: this.#backloggedEvents.payloadBytesEstimation + this.#events.payloadBytesEstimation,
@@ -50,12 +58,24 @@ export class Recorder {
     }
   }
 
+  /**
+   * Returns time-corrected events. If the events were correctable from the beginning, this correction will have already been applied.
+   * @param {SessionReplayEvent[]} events The array of buffered SR nodes
+   * @returns {CorrectedSessionReplayEvent[]}
+   */
+  returnCorrectTimestamps (events) {
+    if (!this.parent.timeKeeper?.ready) return events.events
+    return events.canCorrectTimestamps
+      ? events.events
+      : events.events.map(({ __serialized, timestamp, ...e }) => ({ timestamp: this.parent.timeKeeper.correctAbsoluteTimestamp(timestamp), ...e }))
+  }
+
   /** Clears the buffer (this.#events), and resets all payload metadata properties */
   clearBuffer () {
     if (this.#preloaded[0]?.events.length) this.#preloaded.shift()
     else if (this.parent.mode === MODE.ERROR) this.#backloggedEvents = this.#events
-    else this.#backloggedEvents = new RecorderEvents()
-    this.#events = new RecorderEvents()
+    else this.#backloggedEvents = new RecorderEvents({ canCorrectTimestamps: !!this.parent.timeKeeper?.ready })
+    this.#events = new RecorderEvents({ canCorrectTimestamps: !!this.parent.timeKeeper?.ready })
   }
 
   /** Begin recording using configured recording lib */
@@ -122,12 +142,16 @@ export class Recorder {
   /** Store a payload in the buffer (this.#events).  This should be the callback to the recording lib noticing a mutation */
   store (event, isCheckout) {
     if (!event) return
-    event.__serialized = stringify(event)
 
     if (!this.parent.scheduler && this.#preloaded.length) this.currentBufferTarget = this.#preloaded[this.#preloaded.length - 1]
     else this.currentBufferTarget = this.#events
 
     if (this.parent.blocked) return
+
+    if (this.currentBufferTarget.canCorrectTimestamps) {
+      event.timestamp = this.parent.timeKeeper.correctAbsoluteTimestamp(event.timestamp)
+    }
+    event.__serialized = stringify(event)
     const eventBytes = event.__serialized.length
     /** The estimated size of the payload after compression */
     const payloadSize = this.getPayloadSize(eventBytes)
@@ -160,7 +184,7 @@ export class Recorder {
         this.parent.scheduler.runHarvest()
       } else {
         // we are still in "preload" and it triggered a "stop point".  Make a new set, which will get pointed at on next cycle
-        this.#preloaded.push(new RecorderEvents())
+        this.#preloaded.push(new RecorderEvents({ canCorrectTimestamps: !!this.parent.timeKeeper?.ready }))
       }
     }
   }

tests/assets/rrweb-instrumented.html

@@ -24,10 +24,9 @@
       window.wasPreloaded = false;
       window.addEventListener("load", () => {
         try {
-          const replayEvents = Object.values(newrelic.initializedAgents)[0].features.session_replay.recorder.getEvents();
-          window.wasPreloaded = !!replayEvents.events.length && replayEvents.type === "preloaded";
+          window.wasPreloaded = !!Object.values(newrelic.initializedAgents)[0].features.session_replay.recorder
         } catch (err) {
-          // do nothing because it failed to get the recorder events -- which means they dont exist
+          // do nothing because it failed to get the recorder -- which inherently also means it was not preloaded
         }
       });
     </script>

tests/components/session_replay/aggregate.test.js

@@ -98,12 +98,13 @@ describe('Session Replay', () => {
 
       setConfiguration(agentIdentifier, { ...init })
       sr = new SessionReplayAgg(agentIdentifier, new Aggregator({}))
+      sr.scheduler.runHarvest = jest.fn()
       sr.ee.emit('rumresp', [{ sr: 1 }])
       await wait(1)
       expect(sr.initialized).toBeTruthy()
       expect(sr.recorder.recording).toBeTruthy()
       sr.ee.emit(SESSION_EVENTS.RESET)
-      expect(global.XMLHttpRequest).toHaveBeenCalled()
+      expect(sr.scheduler.runHarvest).toHaveBeenCalled()
       expect(sr.recorder.recording).toBeFalsy()
       expect(sr.blocked).toBeTruthy()
     })
@@ -359,8 +360,11 @@ function wait (ms = 0) {
 
 function primeSessionAndReplay (sess = new SessionEntity({ agentIdentifier, key: 'SESSION', storage: new LocalMemory() })) {
   const timeKeeper = new TimeKeeper(Date.now())
-  const agent = { agentIdentifier, timeKeeper }
+  timeKeeper.processRumRequest({
+    getResponseHeader: jest.fn(() => (new Date()).toUTCString())
+  }, 450, 600)
+  const agent = { agentIdentifier }
   setNREUMInitializedAgent(agentIdentifier, agent)
   session = sess
-  configure(agent, { info, runtime: { session, isolatedBacklog: false }, init: {} }, 'test', true)
+  configure(agent, { info, runtime: { session, isolatedBacklog: false, timeKeeper }, init: {} }, 'test', true)
 }