[players] Guard the review room against event feedback storms

src/composables/previewRoom.js

@@ -19,10 +19,29 @@
  * Socket handlers are auto-registered on mount and removed on unmount.
  * Bound handler references are kept locally so `socket.off()` always
  * targets the correct function instance.
+ *
+ * Room-updated events carry a sequence number (see lib/players/roomSync)
+ * so stale in-flight updates are dropped, broadcasts are muted while a
+ * remote state is applied, and outgoing bursts are throttled — guards
+ * against the event feedback storm of issue #1574.
  */
 import { computed, onBeforeUnmount, onMounted, ref, unref } from 'vue'
 
 import { isValidRoomId, PREVIEW_ROOM_EVENTS } from '@/lib/players/events'
+import {
+  nextRoomSeq,
+  roomSeqOf,
+  shouldApplyRoomUpdate,
+  withRoomSeq
+} from '@/lib/players/roomSync'
+
+// Collapse bursts of status updates (rapid shot changes) into a leading
+// and a trailing emit instead of one event per change (#1574).
+const ROOM_UPDATE_THROTTLE_MS = 200
+// How long broadcasts stay muted after applying a remote update: the
+// component watchers on the applied refs fire on later ticks (some via
+// nextTick or the next video time-update) and would echo the state back.
+const REMOTE_APPLY_MUTE_MS = 500
 
 const noop = () => {}
 
@@ -172,6 +191,17 @@ export const usePreviewRoom = options => {
     setComparisonPanZoom = noop
   } = options
 
+  // ---- Event-storm guards (#1574) ----
+  // See lib/players/roomSync.js for the sequence scheme.
+
+  let localSeq = 0
+  let lastAppliedSeq = 0
+  let isApplyingRemoteUpdate = false
+  let applyMuteTimer = null
+  let pendingEmitTimer = null
+  let lastEmitTime = 0
+  let pendingPreviousPreviewFileId = null
+
   // ---- Computed ----
 
   const joinedRoom = computed(() => {
@@ -242,14 +272,19 @@ export const usePreviewRoom = options => {
 
   // ---- Broadcasts ----
 
-  const updateRoomStatus = (previousPreviewFileId = null) => {
+  const emitRoomStatus = () => {
     const r = unref(room)
     if (!isValidRoomId(r) || !joinedRoom.value) return
     const entity = unref(currentEntity)
     const preview = unref(currentPreview)
+    const previousPreviewFileId = pendingPreviousPreviewFileId
+    pendingPreviousPreviewFileId = null
     socket.emit(PREVIEW_ROOM_EVENTS.roomUpdated, {
       user_id: unref(userId),
-      local_id: r.localId,
+      // The sequence rides inside local_id because the Zou relay only
+      // passes whitelisted fields through (see lib/players/roomSync.js).
+      local_id: withRoomSeq(r.localId, localSeq),
+      room_seq: localSeq,
       playlist_id: r.id,
       is_playing: unref(isPlaying),
       current_entity_id: entity?.id,
@@ -276,6 +311,32 @@ export const usePreviewRoom = options => {
     })
   }
 
+  const updateRoomStatus = (previousPreviewFileId = null) => {
+    // Never re-broadcast a state just applied from the network: the
+    // watchers on the applied refs would echo it back and two clients
+    // would replay each other's updates forever (#1574).
+    if (isApplyingRemoteUpdate) return
+    const r = unref(room)
+    if (!isValidRoomId(r) || !joinedRoom.value) return
+    // Stamp the action now so queued remote echoes older than the user's
+    // latest action are dropped even while the emit is throttled.
+    localSeq = nextRoomSeq(localSeq, lastAppliedSeq)
+    if (previousPreviewFileId !== null) {
+      pendingPreviousPreviewFileId = previousPreviewFileId
+    }
+    const elapsed = Date.now() - lastEmitTime
+    if (elapsed >= ROOM_UPDATE_THROTTLE_MS) {
+      lastEmitTime = Date.now()
+      emitRoomStatus()
+    } else if (!pendingEmitTimer) {
+      pendingEmitTimer = setTimeout(() => {
+        pendingEmitTimer = null
+        lastEmitTime = Date.now()
+        emitRoomStatus()
+      }, ROOM_UPDATE_THROTTLE_MS - elapsed)
+    }
+  }
+
   const postPanZoomChanged = (x, y, zoom) => {
     const r = unref(room)
     if (!isValidRoomId(r) || !joinedRoom.value) return
@@ -499,10 +560,31 @@ export const usePreviewRoom = options => {
   const onRoomUpdated = eventData => {
     const r = unref(room)
     if (!isValidRoomId(r)) return
-    if (r.localId === eventData.local_id) return
     if (!joinedRoom.value) return
     if (eventData.only_newcomer && !r.newComer) return
-    loadRoomCurrentState(eventData)
+    const accepted = shouldApplyRoomUpdate({
+      localId: r.localId,
+      localSeq,
+      lastAppliedSeq,
+      eventData
+    })
+    if (!accepted) return
+    const seq = roomSeqOf(eventData)
+    if (seq !== null) lastAppliedSeq = seq
+    // A newer remote state supersedes any throttled outgoing broadcast.
+    if (pendingEmitTimer) {
+      clearTimeout(pendingEmitTimer)
+      pendingEmitTimer = null
+    }
+    isApplyingRemoteUpdate = true
+    clearTimeout(applyMuteTimer)
+    try {
+      loadRoomCurrentState(eventData)
+    } finally {
+      applyMuteTimer = setTimeout(() => {
+        isApplyingRemoteUpdate = false
+      }, REMOTE_APPLY_MUTE_MS)
+    }
   }
 
   const onPanzoomChanged = eventData => {
@@ -574,6 +656,8 @@ export const usePreviewRoom = options => {
 
   onBeforeUnmount(() => {
     handlerPairs.forEach(([event, handler]) => socket.off(event, handler))
+    clearTimeout(pendingEmitTimer)
+    clearTimeout(applyMuteTimer)
   })
 
   return {

src/lib/players/roomSync.js

@@ -0,0 +1,56 @@
+/*
+ * Ordering and echo guards for the preview-room "room-updated" events
+ * (issue #1574): without them, changing shots quickly floods the room and
+ * each client re-applies stale in-flight states and re-broadcasts them,
+ * leaving two clients oscillating between the last two videos.
+ *
+ * The sequence number is a hybrid logical clock: wall-clock milliseconds
+ * merged with the highest sequence seen in the room, so values are
+ * comparable across clients and strictly monotonic locally.
+ *
+ * The Zou event server rebuilds the relayed payload from a field
+ * whitelist, so a standalone `room_seq` field is stripped in transit on
+ * current servers. The sequence therefore also rides inside `local_id`
+ * ("<uuid>#<seq>"), which is relayed verbatim. Old clients only compare
+ * `local_id` with their own plain uuid, so the suffix is invisible to
+ * them, and events without any sequence are always applied.
+ */
+
+const SEQ_SEPARATOR = '#'
+
+export const nextRoomSeq = (localSeq = 0, lastAppliedSeq = 0) =>
+  Math.max(Date.now(), localSeq + 1, lastAppliedSeq + 1)
+
+export const withRoomSeq = (localId, seq) => `${localId}${SEQ_SEPARATOR}${seq}`
+
+export const senderIdOf = localId => {
+  if (typeof localId !== 'string') return localId
+  const index = localId.indexOf(SEQ_SEPARATOR)
+  return index === -1 ? localId : localId.slice(0, index)
+}
+
+export const roomSeqOf = eventData => {
+  if (Number.isFinite(eventData?.room_seq)) return eventData.room_seq
+  if (typeof eventData?.local_id !== 'string') return null
+  const index = eventData.local_id.indexOf(SEQ_SEPARATOR)
+  if (index === -1) return null
+  const seq = Number(eventData.local_id.slice(index + 1))
+  return Number.isFinite(seq) ? seq : null
+}
+
+export const shouldApplyRoomUpdate = ({
+  localId,
+  localSeq = 0,
+  lastAppliedSeq = 0,
+  eventData
+}) => {
+  if (!eventData) return false
+  // Self-echo: the server sends our own updates back to us.
+  if (localId && senderIdOf(eventData.local_id) === localId) return false
+  const seq = roomSeqOf(eventData)
+  // No sequence: event from an older client, apply as before.
+  if (seq === null) return true
+  // Stale or out-of-order: older than the last applied remote update or
+  // than the user's own last local action.
+  return seq > localSeq && seq > lastAppliedSeq
+}

tests/unit/composables/previewRoom.spec.js

@@ -230,7 +230,9 @@ describe('composables/previewRoom', () => {
         'preview-room:room-updated',
         expect.objectContaining({
           user_id: 'user-1',
-          local_id: 'local-abc',
+          // The room sequence rides inside local_id (see lib/players/roomSync)
+          local_id: expect.stringMatching(/^local-abc#\d+$/),
+          room_seq: expect.any(Number),
           playlist_id: 'playlist-1',
           previous_preview_file_id: 'prev-preview-id',
           is_playing: false,
@@ -627,4 +629,111 @@ describe('composables/previewRoom', () => {
       wrapper.unmount()
     })
   })
+
+  describe('event storm guards (#1574)', () => {
+    const handlerFor = (socket, event) => {
+      const call = socket.on.mock.calls.find(([name]) => name === event)
+      return call?.[1]
+    }
+
+    const roomUpdateCount = socket =>
+      socket.emit.mock.calls.filter(
+        ([event]) => event === 'preview-room:room-updated'
+      ).length
+
+    const mountPlayer = (extra = {}) => {
+      const socket = makeSocket()
+      const playEntity = vi.fn()
+      const mounted = mountWithRoom({
+        room: makeRoom({ people: ['user-1'] }),
+        userId: 'user-1',
+        socket,
+        playEntity,
+        ...extra
+      })
+      const handler = handlerFor(socket, 'preview-room:room-updated')
+      return { ...mounted, socket, playEntity, handler }
+    }
+
+    beforeEach(() => {
+      vi.useFakeTimers()
+    })
+
+    afterEach(() => {
+      vi.useRealTimers()
+    })
+
+    it('drops stale incoming updates (sequence behind the last applied)', () => {
+      const { wrapper, playEntity, handler } = mountPlayer()
+      handler({
+        local_id: 'other#2000',
+        is_playing: false,
+        current_entity_index: 2
+      })
+      expect(playEntity).toHaveBeenCalledTimes(1)
+      handler({
+        local_id: 'other#1000',
+        is_playing: false,
+        current_entity_index: 5
+      })
+      expect(playEntity).toHaveBeenCalledTimes(1)
+      wrapper.unmount()
+    })
+
+    it('applies updates without a sequence (older clients)', () => {
+      const { wrapper, playEntity, handler } = mountPlayer()
+      handler({ local_id: 'other', is_playing: false, current_entity_index: 2 })
+      handler({ local_id: 'other', is_playing: false, current_entity_index: 5 })
+      expect(playEntity).toHaveBeenCalledTimes(2)
+      wrapper.unmount()
+    })
+
+    it("drops incoming updates older than the user's own last action", () => {
+      const { wrapper, api, playEntity, handler } = mountPlayer()
+      api.updateRoomStatus()
+      handler({
+        local_id: `other#${Date.now() - 1000}`,
+        is_playing: false,
+        current_entity_index: 2
+      })
+      expect(playEntity).not.toHaveBeenCalled()
+      handler({
+        local_id: `other#${Date.now() + 1000}`,
+        is_playing: false,
+        current_entity_index: 3
+      })
+      expect(playEntity).toHaveBeenCalledTimes(1)
+      wrapper.unmount()
+    })
+
+    it('never re-broadcasts a state just applied from the network', () => {
+      const { wrapper, api, socket, handler } = mountPlayer()
+      handler({
+        local_id: `other#${Date.now()}`,
+        is_playing: false,
+        current_entity_index: 2
+      })
+      socket.emit.mockClear()
+      // A watcher on an applied ref echoing the state back
+      api.updateRoomStatus()
+      expect(roomUpdateCount(socket)).toBe(0)
+      vi.advanceTimersByTime(600)
+      api.updateRoomStatus()
+      expect(roomUpdateCount(socket)).toBe(1)
+      wrapper.unmount()
+    })
+
+    it('collapses a burst of updates into a leading and a trailing emit', () => {
+      const { wrapper, api, socket } = mountPlayer()
+      api.updateRoomStatus()
+      api.updateRoomStatus()
+      api.updateRoomStatus()
+      expect(roomUpdateCount(socket)).toBe(1)
+      vi.advanceTimersByTime(250)
+      expect(roomUpdateCount(socket)).toBe(2)
+      vi.advanceTimersByTime(1000)
+      expect(roomUpdateCount(socket)).toBe(2)
+      wrapper.unmount()
+    })
+  })
 })