diff --git a/explorer/src/workspaces/GraphWorkspace/GraphWorkspace.tsx b/explorer/src/workspaces/GraphWorkspace/GraphWorkspace.tsx index 1a9cface..f18a2fe6 100644 --- a/explorer/src/workspaces/GraphWorkspace/GraphWorkspace.tsx +++ b/explorer/src/workspaces/GraphWorkspace/GraphWorkspace.tsx @@ -41,6 +41,7 @@ import { } from "./plugins"; import { explorationEffectsShouldLoad, neighborhoodPanelShouldLoad, temporalOverlayShouldLoad } from "./pluginRegistryPredicates"; import { shouldFetchTemporalBounds, shouldFetchTemporalSnapshot } from "./temporalLifecyclePredicates"; +import { createTemporalSnapshotGuards, type TemporalSnapshotResponse } from "./temporalSnapshotGuards"; import type { LinkPrediction, PathResponse } from "./GraphInspectorPanel"; import type { GraphSceneHandle, GraphSceneRuntime } from "./scene"; import type { @@ -1479,6 +1480,23 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap summary?.edgeCount, ]); + // Guards the snapshot lifecycle: at most one in-flight request per scrubber + // position (identical-`at` polls are deduplicated, breaking the idle/play + // polling loop), applied snapshots are cached and re-applied on revisit, and + // a response applies only while the scrubber is still on its position + // (out-of-order responses cannot clobber the active-node count). + const temporalSnapshotGuardsRef = useRef | null>(null); + if (temporalSnapshotGuardsRef.current === null) { + temporalSnapshotGuardsRef.current = createTemporalSnapshotGuards(); + } + const temporalSnapshotGuards = temporalSnapshotGuardsRef.current; + + // A new graph summary means the graph data was replaced (reload/retry); + // snapshots cached against the previous graph are stale, so reset all state. + useEffect(() => { + temporalSnapshotGuards.reset(); + }, [summary]); + useEffect(() => { if (!canFetchTemporalSnapshot) { return; @@ -1488,37 +1506,67 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap return; } + const atMs = debouncedTime.getTime(); + const { seq, cached } = temporalSnapshotGuards.begin(atMs); + if (seq === null) { + // An identical request is already in flight: one request per position. + return; + } + let cancelled = false; + const applyData = (data: TemporalSnapshotResponse) => { + const nextActiveIds = new Set(data.active_node_ids); + requestAnimationFrame(() => { + if (cancelled) return; + if (!temporalSnapshotGuards.shouldApply(atMs, seq)) { + // The scrubber moved on (or this request was superseded): release the + // position so a return to it refetches instead of stalling. + temporalSnapshotGuards.finish(atMs, seq); + return; + } + const previous = prevActiveIdsRef.current; + previous.forEach((id) => { + if (!nextActiveIds.has(id) && graph.hasNode(id)) { + graph.setNodeAttribute(id, "hidden", true); + } + }); + nextActiveIds.forEach((id) => { + if (graph.hasNode(id)) { + graph.setNodeAttribute(id, "hidden", false); + } + }); + prevActiveIdsRef.current = nextActiveIds; + setActiveNodeCount(data.active_node_count); + setGraphVersion((current) => current + 1); + sceneRef.current?.getRuntime()?.requestRender(); + temporalSnapshotGuards.apply(atMs, seq, data); + }); + }; + + if (cached) { + // Returning to a position whose snapshot was already applied: re-apply + // the cached result without a network request. + applyData(cached); + return; + } + const applySnapshot = async () => { try { const at = debouncedTime.toISOString(); const response = await fetch(`/api/temporal/snapshot?at=${encodeURIComponent(at)}`); - if (!response.ok || cancelled) return; - - const data: { active_node_ids: string[]; active_node_count: number } = await response.json(); + if (!response.ok) { + // A failed request must be retryable if the scrubber returns. + if (!cancelled) temporalSnapshotGuards.finish(atMs, seq); + return; + } if (cancelled) return; - const nextActiveIds = new Set(data.active_node_ids); - requestAnimationFrame(() => { - if (cancelled) return; - const previous = prevActiveIdsRef.current; - previous.forEach((id) => { - if (!nextActiveIds.has(id) && graph.hasNode(id)) { - graph.setNodeAttribute(id, "hidden", true); - } - }); - nextActiveIds.forEach((id) => { - if (graph.hasNode(id)) { - graph.setNodeAttribute(id, "hidden", false); - } - }); - prevActiveIdsRef.current = nextActiveIds; - setActiveNodeCount(data.active_node_count); - setGraphVersion((current) => current + 1); - sceneRef.current?.getRuntime()?.requestRender(); - }); + const data: TemporalSnapshotResponse = await response.json(); + if (cancelled) return; + applyData(data); } catch (fetchError) { + temporalSnapshotGuards.finish(atMs, seq); if (!cancelled) { console.error("[Temporal] Snapshot fetch failed", fetchError); } @@ -1528,6 +1576,8 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap applySnapshot(); return () => { cancelled = true; + // A cancelled request must be retryable when its position is revisited. + temporalSnapshotGuards.finish(atMs, seq); }; }, [ canFetchTemporalSnapshot, diff --git a/explorer/src/workspaces/GraphWorkspace/temporalSnapshotGuards.ts b/explorer/src/workspaces/GraphWorkspace/temporalSnapshotGuards.ts new file mode 100644 index 00000000..6aa3fde2 --- /dev/null +++ b/explorer/src/workspaces/GraphWorkspace/temporalSnapshotGuards.ts @@ -0,0 +1,113 @@ +/** + * Guards for the temporal snapshot fetch/apply lifecycle. + * + * The snapshot effect previously fetched /api/temporal/snapshot with no + * idempotency or ordering protection. Upstream churn (timeline recreation + * while bounds settle, play ticks resetting the playhead, drag events) could + * re-request the same `at` repeatedly, and responses could arrive after the + * scrubber had moved on. + * + * The guards enforce: + * - at most one in-flight request per scrubber position (identical `at` + * values are deduplicated while a request is pending, breaking the + * idle/play polling loop); + * - successful snapshots are cached per position and re-applied when the + * scrubber returns (play wrap-around, back-scrubbing) without a refetch; + * - a response is applied only while the scrubber is still on its position, + * so out-of-order responses cannot clobber a newer position's count; + * - failed, cancelled, or superseded requests release their position so it + * can be fetched again on the next visit; + * - `reset()` drops all state when the underlying graph data is replaced + * (reload/retry), because cached snapshots describe the previous graph. + * + * `createTemporalSnapshotGuards()` is stateful by design. + */ + +export interface TemporalSnapshotResponse { + active_node_ids: string[]; + active_node_count: number; +} + +export interface TemporalSnapshotRequest { + /** null when the request was deduplicated because one is already in flight. */ + seq: number | null; + /** The snapshot previously applied for this position, when revisiting it. */ + cached: TemporalSnapshotResponse | null; +} + +export interface TemporalSnapshotGuards { + /** Begin (or dedupe) a request for `atMs`; marks it as the current position. */ + begin(atMs: number): TemporalSnapshotRequest; + /** True when the response for `atMs`/`seq` may be applied (scrubber still on `atMs`). */ + shouldApply(atMs: number, seq: number): boolean; + /** Record a successful application and cache its snapshot for revisits. */ + apply(atMs: number, seq: number, data: TemporalSnapshotResponse): void; + /** Release a position whose request failed, was cancelled, or was superseded. */ + finish(atMs: number, seq: number): void; + /** Drop all state; call when the underlying graph data is replaced (reload). */ + reset(): void; +} + +interface SnapshotEntry { + seq: number; + /** null while the request is in flight (or before the first success). */ + data: TemporalSnapshotResponse | null; +} + +/** Upper bound on cached positions so long scrubbing sessions stay bounded. */ +const MAX_CACHED_POSITIONS = 256; + +export function createTemporalSnapshotGuards(): TemporalSnapshotGuards { + const entries = new Map(); + let latestRequestSeq = 0; + let currentAtMs: number | null = null; + + const evictOldest = () => { + while (entries.size > MAX_CACHED_POSITIONS) { + const oldestAtMs = entries.keys().next().value; + if (oldestAtMs === undefined) return; + entries.delete(oldestAtMs); + } + }; + + return { + begin(atMs) { + const existing = entries.get(atMs); + if (existing && existing.data === null) { + // Identical request already in flight: dedupe, but the scrubber is here now. + currentAtMs = atMs; + return { seq: null, cached: null }; + } + latestRequestSeq += 1; + const seq = latestRequestSeq; + entries.set(atMs, { seq, data: existing?.data ?? null }); + currentAtMs = atMs; + evictOldest(); + return { seq, cached: existing?.data ?? null }; + }, + + shouldApply(atMs, seq) { + return atMs === currentAtMs && entries.get(atMs)?.seq === seq; + }, + + apply(atMs, seq, data) { + const entry = entries.get(atMs); + if (entry && entry.seq === seq) { + entry.data = data; + } + }, + + finish(atMs, seq) { + const entry = entries.get(atMs); + if (entry && entry.seq === seq && entry.data === null) { + entries.delete(atMs); + } + }, + + reset() { + entries.clear(); + latestRequestSeq = 0; + currentAtMs = null; + }, + }; +} diff --git a/explorer/tests/temporalSnapshotGuards.test.ts b/explorer/tests/temporalSnapshotGuards.test.ts new file mode 100644 index 00000000..33879d79 --- /dev/null +++ b/explorer/tests/temporalSnapshotGuards.test.ts @@ -0,0 +1,150 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { createTemporalSnapshotGuards } from "../src/workspaces/GraphWorkspace/temporalSnapshotGuards.ts"; + +const POSITION_1 = new Date("2023-07-02T00:00:00Z").getTime(); +const POSITION_2 = new Date("2024-01-02T00:00:00Z").getTime(); +const POSITION_3 = new Date("2024-07-02T00:00:00Z").getTime(); + +const SNAPSHOT = { active_node_ids: ["n1", "n2"], active_node_count: 2 }; + +// ── begin: one request per scrubber position ───────────────────────────────── + +test("begin: a new position returns a fresh request sequence", () => { + const guards = createTemporalSnapshotGuards(); + assert.deepEqual(guards.begin(POSITION_1), { seq: 1, cached: null }); +}); + +test("begin: an identical in-flight request is deduplicated (no duplicate fetch)", () => { + const guards = createTemporalSnapshotGuards(); + guards.begin(POSITION_1); + assert.deepEqual(guards.begin(POSITION_1), { seq: null, cached: null }); +}); + +test("begin: distinct positions request independently", () => { + const guards = createTemporalSnapshotGuards(); + assert.equal(guards.begin(POSITION_1).seq, 1); + assert.equal(guards.begin(POSITION_2).seq, 2); +}); + +test("begin: revisiting an applied position returns its cached snapshot", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.apply(POSITION_1, seq, SNAPSHOT); + const revisit = guards.begin(POSITION_1); + assert.equal(revisit.seq, 2); + assert.deepEqual(revisit.cached, SNAPSHOT); +}); + +test("begin: a failed position (finished) can be requested again", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.finish(POSITION_1, seq); + const retry = guards.begin(POSITION_1); + assert.equal(retry.seq, 2); + assert.equal(retry.cached, null); +}); + +test("finish: does not clear a position whose snapshot was already applied", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.apply(POSITION_1, seq, SNAPSHOT); + guards.finish(POSITION_1, seq); + assert.deepEqual(guards.begin(POSITION_1).cached, SNAPSHOT); +}); + +test("finish: a stale sequence cannot release a newer request's position", () => { + const guards = createTemporalSnapshotGuards(); + const first = guards.begin(POSITION_1); + guards.finish(POSITION_1, first.seq); + guards.begin(POSITION_1); // seq 2, in flight again + guards.finish(POSITION_1, first.seq); // stale seq: must not release seq 2 + assert.deepEqual(guards.begin(POSITION_1), { seq: null, cached: null }); +}); + +// ── shouldApply: applied only while the scrubber is on that position ───────── + +test("shouldApply: the current position's response is applied", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + assert.equal(guards.shouldApply(POSITION_1, seq), true); +}); + +test("shouldApply: a response for a position the scrubber left is discarded", () => { + const guards = createTemporalSnapshotGuards(); + const { seq: seq1 } = guards.begin(POSITION_1); + guards.begin(POSITION_2); + assert.equal(guards.shouldApply(POSITION_1, seq1), false); + assert.equal(guards.shouldApply(POSITION_2, 2), true); +}); + +test("shouldApply: a late response for the position the scrubber returned to is applied", () => { + const guards = createTemporalSnapshotGuards(); + const { seq: seq1 } = guards.begin(POSITION_1); + const { seq: seq2 } = guards.begin(POSITION_2); + guards.begin(POSITION_1); // back to 1: deduplicated, no new request + assert.equal(guards.shouldApply(POSITION_1, seq1), true); + assert.equal(guards.shouldApply(POSITION_2, seq2), false); +}); + +test("shouldApply: an unknown sequence is discarded", () => { + const guards = createTemporalSnapshotGuards(); + guards.begin(POSITION_1); + assert.equal(guards.shouldApply(POSITION_1, 99), false); +}); + +test("shouldApply: after a reset no pre-reset response applies", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.reset(); + assert.equal(guards.shouldApply(POSITION_1, seq), false); +}); + +// ── apply: caching for revisits ───────────────────────────────────────────── + +test("apply: stores the snapshot so a revisit re-applies it without a request", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.apply(POSITION_1, seq, SNAPSHOT); + guards.begin(POSITION_2); + assert.deepEqual(guards.begin(POSITION_1).cached, SNAPSHOT); +}); + +test("apply: play wrap-around re-applies the wrapped-to position's snapshot", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.apply(POSITION_1, seq, SNAPSHOT); + guards.begin(POSITION_2); + guards.begin(POSITION_3); + const wrap = guards.begin(POSITION_1); + assert.deepEqual(wrap.cached, SNAPSHOT); + assert.equal(guards.shouldApply(POSITION_1, wrap.seq), true); +}); + +// ── reset: graph reload ───────────────────────────────────────────────────── + +test("reset: clears requested and cached state so positions refetch", () => { + const guards = createTemporalSnapshotGuards(); + const { seq } = guards.begin(POSITION_1); + guards.apply(POSITION_1, seq, SNAPSHOT); + guards.reset(); + const fresh = guards.begin(POSITION_1); + assert.equal(fresh.seq, 1); + assert.equal(fresh.cached, null); +}); + +// ── cache bound ───────────────────────────────────────────────────────────── + +test("cache: oldest positions are evicted when the cache is full", () => { + const guards = createTemporalSnapshotGuards(); + const count = 300; + for (let i = 0; i < count; i++) { + const { seq } = guards.begin(POSITION_1 + i * 1000); + guards.apply(POSITION_1 + i * 1000, seq, SNAPSHOT); + } + const oldest = guards.begin(POSITION_1); + assert.equal(oldest.cached, null); // evicted: must refetch on revisit + const newest = guards.begin(POSITION_1 + (count - 1) * 1000); + assert.deepEqual(newest.cached, SNAPSHOT); // still cached +});