fix(web): guard run live-sync snapshot sync against unstable references (#2797)

Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
2026-08-19 12:07:36 +02:00
parent 3949f6f5e8
commit 59728fe14d
2 changed files with 168 additions and 2 deletions
@@ -0,0 +1,122 @@
+import { act, renderHook } from "@testing-library/react";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { RunEvent } from "@/lib/api/domains/office-extended-api";
+import type { RunEventAppendedPayload } from "@/lib/types/backend";
+import { useRunLiveSync } from "./use-run-live-sync";
+
+const clients = vi.hoisted(() => ({
+  active: { subscribeRun: vi.fn() },
+}));
+
+vi.mock("@/lib/ws/connection", () => ({
+  getWebSocketClient: () => clients.active,
+  useWebSocketClient: () => clients.active,
+}));
+
+const handlers = vi.hoisted(() => ({
+  listener: undefined as ((payload: RunEventAppendedPayload) => void) | undefined,
+}));
+
+vi.mock("@/lib/ws/handlers/run", () => ({
+  subscribeRunEvents: (_runId: string, listener: (payload: RunEventAppendedPayload) => void) => {
+    handlers.listener = listener;
+    return () => {
+      handlers.listener = undefined;
+    };
+  },
+}));
+
+function runEvent(seq: number, eventType: string): RunEvent {
+  return {
+    seq,
+    event_type: eventType,
+    level: "info",
+    payload: "{}",
+    created_at: "2026-01-01T00:00:00Z",
+  };
+}
+
+type Props = {
+  runId: string;
+  initialEvents: RunEvent[];
+  initialStatus: "claimed";
+};
+
+function renderLiveSync(initialProps: Props) {
+  return renderHook(
+    (props: Props) => useRunLiveSync(props.runId, props.initialEvents, props.initialStatus),
+    { initialProps },
+  );
+}
+
+const initialProps: Props = { runId: "run-1", initialEvents: [], initialStatus: "claimed" };
+
+describe("useRunLiveSync", () => {
+  beforeEach(() => {
+    clients.active = { subscribeRun: vi.fn(() => vi.fn()) };
+    handlers.listener = undefined;
+  });
+
+  it("does not loop when the snapshot reference changes but its content does not", () => {
+    // A fresh array literal on every render changes `initialEvents` identity
+    // without changing its content. Without the content guard the snapshot
+    // sync effect re-writes state on every render, feeding an endless
+    // render->effect cycle that exhausts the worker (this test times out
+    // rather than completing).
+    const { result, rerender } = renderHook(() => useRunLiveSync("run-1", [], "claimed"));
+
+    rerender();
+    rerender();
+
+    expect(clients.active.subscribeRun).toHaveBeenCalledOnce();
+    expect(result.current.events).toEqual([]);
+    expect(result.current.status).toBe("claimed");
+  });
+
+  it("does not clobber a live event when a fresh but unchanged snapshot rerenders", () => {
+    // A caller that supplies a fresh-but-unchanged `initialEvents` array (the
+    // exact scenario the snapshot guard exists for) must not erase events that
+    // arrived over the WS. The snapshot sync decision has to compare against
+    // the last incoming snapshot, not against the live-event dedup set: once
+    // the first WS event adds its seq to the dedup set, comparing the stale
+    // snapshot against that set looks like a content change and resets events
+    // (and the dedup set) to the snapshot, making the live event vanish.
+    const { result, rerender } = renderLiveSync(initialProps);
+
+    const liveEvent = runEvent(1, "started");
+    act(() => {
+      handlers.listener!({ run_id: "run-1", event: liveEvent });
+    });
+    expect(result.current.events).toEqual([liveEvent]);
+
+    rerender({ ...initialProps, initialEvents: [] });
+
+    expect(result.current.events).toEqual([liveEvent]);
+  });
+
+  it("merges a changed snapshot without dropping a live event", () => {
+    const initialEvent = runEvent(1, "init");
+    const liveEvent = runEvent(2, "started");
+    const snapshotEvent = runEvent(3, "step");
+    const { result, rerender } = renderLiveSync({
+      ...initialProps,
+      initialEvents: [initialEvent],
+    });
+
+    act(() => {
+      handlers.listener!({ run_id: "run-1", event: liveEvent });
+    });
+    rerender({ ...initialProps, initialEvents: [initialEvent, snapshotEvent] });
+
+    expect(result.current.events).toEqual([initialEvent, liveEvent, snapshotEvent]);
+  });
+
+  it("re-syncs events when the snapshot content actually changes", () => {
+    const started = runEvent(1, "started");
+    const { result, rerender } = renderLiveSync(initialProps);
+
+    rerender({ ...initialProps, initialEvents: [started] });
+
+    expect(result.current.events).toEqual([started]);
+  });
+});
@@ -40,12 +40,46 @@ export function useRunLiveSync(
 ): { events: RunEvent[]; status: Status } {
  const [events, setEvents] = useState<RunEvent[]>(initialEvents);
  const [status, setStatus] = useState<Status>(initialStatus);
+  // Live-event dedup only: seqs of events already appended over the WS (plus
+  // the seqs of the mount-time snapshot, so a reconnect replay of a
+  // snapshot-covered event cannot double-insert). Never reset to a snapshot:
+  // doing so erases live events on a fresh-but-unchanged rerender.
  const seenSeqsRef = useRef<Set<number>>(new Set(initialEvents.map((e) => e.seq)));
+  // Last incoming snapshot, tracked separately from the live dedup set. The
+  // sync decision compares against this, so live-added seqs can never make
+  // an unchanged snapshot look like a content change.
+  const lastSnapshotRef = useRef<{ runId: string; seqs: Set<number> }>({
+    runId,
+    seqs: new Set(initialEvents.map((e) => e.seq)),
+  });

  useEffect(() => {
-    seenSeqsRef.current = new Set(initialEvents.map((e) => e.seq));
+    const nextSeqs = new Set(initialEvents.map((e) => e.seq));
+    const prev = lastSnapshotRef.current;
+    if (
+      prev.runId === runId &&
+      prev.seqs.size === nextSeqs.size &&
+      [...nextSeqs].every((seq) => prev.seqs.has(seq))
+    ) {
+      // Same run, same event set as the last incoming snapshot. Skip the state
+      // write so an unstable `initialEvents` reference (a fresh array literal
+      // on every render) cannot feed an endless render->effect cycle, and so a
+      // stale snapshot cannot clobber events appended live over the WS.
+      return;
+    }
+    lastSnapshotRef.current = { runId, seqs: nextSeqs };
+    if (prev.runId === runId) {
+      // Fresh snapshot for the same run: merge it with live-added events and
+      // fold its seqs into the dedup set so reconnect replays cannot duplicate.
+      for (const seq of nextSeqs) seenSeqsRef.current.add(seq);
+      setEvents((current) => mergeRunEvents(current, initialEvents));
+      return;
+    } else {
+      // Different run: restart dedup from the new snapshot; seqs are per-run.
+      seenSeqsRef.current = new Set(nextSeqs);
+    }
    setEvents(initialEvents);
-  }, [initialEvents]);
+  }, [initialEvents, runId]);

  useEffect(() => {
    setStatus(initialStatus);
@@ -97,3 +131,13 @@ function mergeRunEvent(prev: RunEvent[], evt: RunEvent): RunEvent[] {
  next.splice(lo, 0, evt);
  return next;
 }
+
+function mergeRunEvents(current: RunEvent[], snapshot: RunEvent[]): RunEvent[] {
+  if (current.length === 0) return snapshot;
+  if (snapshot.length === 0) return current;
+
+  const bySeq = new Map<number, RunEvent>();
+  for (const event of current) bySeq.set(event.seq, event);
+  for (const event of snapshot) bySeq.set(event.seq, event);
+  return [...bySeq.values()].sort((a, b) => a.seq - b.seq);
+}