fix(replay): address review — stopReplay leak, stale session-end job, throttle, a11y

- web SDK: stopReplay() now detaches the session-rotation listener so a later
  rotation can't restart the recorder after a manual stop. Extract a shared
  startForSession() helper (was duplicated between initial start + restart).
- worker: createSessionEndJob replaces any existing delayed job for the device
  before enqueueing. Previously a client session_id rotation left the stale
  (deviceId-keyed) job with the old payload, so every later event saw
  isNewSession=true and spawned duplicate session_start events.
- recorder: throttle onUserActivity to <=1/sec — MouseMove/Scroll fired a
  localStorage write on every event. lastActivityMs still updates on all.
- dashboard: aria-pressed on the Skip-idle toggle for screen readers.

Author: ayushjhanwar-png

apps/start/src/components/sessions/replay/index.tsx

@@ -324,6 +324,7 @@ function ReplayContent({
                 {hasReplay && (
                   <button
                     type="button"
+                    aria-pressed={skipInactive}
                     onClick={() => setSkipInactive((v) => !v)}
                     title={
                       skipInactive

apps/worker/src/utils/session-handler.ts

@@ -29,6 +29,20 @@ export async function createSessionEndJob({
 }: {
   payload: IServiceCreateEventPayload;
 }) {
+  const jobId = getSessionEndJobId(payload.projectId, payload.deviceId);
+  // The job is keyed by deviceId, and BullMQ's add() is a no-op when a job
+  // with the same id already exists. When a client rotates its session_id
+  // (Phase 5) a stale delayed job still carries the OLD session's payload —
+  // without replacing it, later events keep matching the old session and
+  // spawn duplicate session_start events. Remove any existing delayed job so
+  // the new one carries the current session's payload.
+  const existing = await sessionsQueue.getJob(jobId);
+  if (existing) {
+    await existing.remove().catch(() => {
+      // Job may be active/locked — remove can throw. Fall back to the add
+      // below (a no-op if the id is still present); non-fatal.
+    });
+  }
   return sessionsQueue.add(
     'session',
     {
@@ -37,7 +51,7 @@ export async function createSessionEndJob({
     },
     {
       delay: SESSION_TIMEOUT,
-      jobId: getSessionEndJobId(payload.projectId, payload.deviceId),
+      jobId,
       attempts: 3,
       backoff: {
         type: 'exponential',

packages/sdks/web/src/index.ts

@@ -69,6 +69,13 @@ export class OpenPanel extends OpenPanelBase {
    */
   private windowId?: string;
 
+  /**
+   * Unsubscribe for the session-rotation listener registered in
+   * maybeStartReplay(). Captured so stopReplay() can detach it — otherwise a
+   * later rotation would restart the recorder after a manual stop.
+   */
+  private replaySessionUnsub?: () => void;
+
   constructor(public options: OpenPanelOptions) {
     super({
       sdk: 'web',
@@ -218,21 +225,28 @@ export class OpenPanel extends OpenPanelBase {
     // start a new one with the fresh id (chunk_index=0 + FullSnapshot).
     let activeSessionId = this.sessionId;
     const bumpActivity = () => this.sessionManager?.bumpActivity();
-    startReplayRecorder(
-      opts,
-      (chunk) => {
-        this.send({
-          type: 'replay',
-          payload: {
-            ...chunk,
-            session_id: activeSessionId,
-          },
-        });
-      },
-      bumpActivity,
-    );
+    // Shared recorder-start so the initial start and the post-rotation restart
+    // stay in sync (single source of truth for the chunk-send closure).
+    const startForSession = (sessionId: string) => {
+      activeSessionId = sessionId;
+      startReplayRecorder(
+        opts,
+        (chunk) => {
+          this.send({
+            type: 'replay',
+            payload: {
+              ...chunk,
+              session_id: activeSessionId,
+            },
+          });
+        },
+        bumpActivity,
+      );
+    };
 
-    this.sessionManager.onSessionIdChanged((newId) => {
+    startForSession(this.sessionId);
+
+    this.replaySessionUnsub = this.sessionManager.onSessionIdChanged((newId) => {
       // Match PostHog: a session rotation also mints a fresh window_id. The
       // recorder restarts with a new FullSnapshot at chunk_index=0, so the
       // post-rotation recording is fully self-describing — the dashboard
@@ -246,24 +260,15 @@ export class OpenPanel extends OpenPanelBase {
       });
       this.sessionId = newId;
       stopReplayRecorder();
-      activeSessionId = newId;
-      startReplayRecorder(
-        opts,
-        (chunk) => {
-          this.send({
-            type: 'replay',
-            payload: {
-              ...chunk,
-              session_id: activeSessionId,
-            },
-          });
-        },
-        bumpActivity,
-      );
+      startForSession(newId);
     });
   }
 
   public stopReplay() {
+    // Detach the rotation listener first, so a later session rotation can't
+    // restart the recorder after a manual stop.
+    this.replaySessionUnsub?.();
+    this.replaySessionUnsub = undefined;
     stopReplayRecorder();
   }
 

packages/sdks/web/src/replay/recorder.ts

@@ -83,6 +83,10 @@ export function startReplayRecorder(
   // always captured. lastActivityMs tracks the last real user interaction.
   let isIdle = false;
   let lastActivityMs = Date.now();
+  // MouseMove/Scroll fire many times a second and each onUserActivity() bump
+  // writes localStorage. Throttle the notify to at most once/sec — lastActivityMs
+  // still updates on every interactive event (in-memory, cheap).
+  let lastActivityNotifyMs = 0;
 
   function flush(isFullSnapshot: boolean): void {
     if (buffer.length === 0) return;
@@ -148,7 +152,10 @@ export function startReplayRecorder(
 
       if (interactive) {
         lastActivityMs = event.timestamp;
-        onUserActivity?.();
+        if (event.timestamp - lastActivityNotifyMs >= 1000) {
+          lastActivityNotifyMs = event.timestamp;
+          onUserActivity?.();
+        }
         if (isIdle) {
           // Returning from idle. The player's DOM mirror is stale (we dropped
           // the mutations that happened while idle), so force a fresh