PERF-95: coalescing, concurrency-bounded node-position saves

packages/web/src/components/InteractiveGraphVisualization.tsx

@@ -53,6 +53,7 @@ import { edgeBorderEndpoints, minEdgeLength, clampToMinNeighbors } from '../lib/
 import { directionStrategy, arrowVisibility, labelVisibility, perpendicularOffset } from '../lib/edgeLOD';
 import { assignParallelEdgeIndices } from '../lib/parallelEdges';
 import { isTap, touchHitSize, isCoarsePointer } from '../lib/touchInteraction';
+import { createPositionSaveQueue } from '../lib/positionSaveQueue';
 import { spawnCelebration } from '../lib/celebration';
 import { buildNeighborhood } from '../lib/graphAdjacency';
 import { UndoStack } from '../lib/undoStack';
@@ -721,6 +722,18 @@ export function InteractiveGraphVisualization({ onResetLayout, onNodeSelected, i
     }
   }, [updateWorkItemMutation]);
 
+  // #95: coalescing, concurrency-bounded position writes. A large-graph reflow
+  // moves hundreds of nodes at once; without this each fired its own mutation in
+  // parallel (a D1 write-amplification spike). The queue keeps only the latest
+  // position per node and drains with a hard cap on in-flight writes. Bound to
+  // the latest saveNodePosition via a ref so the queue itself is created once.
+  const saveNodePositionRef = useRef(saveNodePosition);
+  saveNodePositionRef.current = saveNodePosition;
+  const positionSaveQueueRef = useRef(createPositionSaveQueue({
+    save: (s) => saveNodePositionRef.current(s.id, s.x, s.y),
+    concurrency: 4,
+  }));
+
   // Durable layout: persist the current position of every node whose live
   // position has moved away from its last-saved position by more than a pixel.
   // This makes a physics-laid-out arrangement (and grown/new nodes) survive a
@@ -733,12 +746,16 @@ export function InteractiveGraphVisualization({ onResetLayout, onNodeSelected, i
       typeof n.x === 'number' && typeof n.y === 'number' &&
       (Math.abs((n.positionX ?? 0) - n.x) > 1 || Math.abs((n.positionY ?? 0) - n.y) > 1)
     );
+    // Coalesce all moved nodes into the queue, then drain with bounded
+    // concurrency — one reflow of N nodes does at most `concurrency` writes at a
+    // time instead of N in parallel (#95).
     moved.forEach((n: any) => {
       n.positionX = n.x;
       n.positionY = n.y;
-      saveNodePosition(n.id, n.x, n.y);
+      positionSaveQueueRef.current.queue(n.id, n.x, n.y);
     });
-  }, [saveNodePosition]);
+    positionSaveQueueRef.current.flush();
+  }, []);
 
   // Save the settled layout on page hide so a tidy arrangement is never lost
   // even if the user never dragged a node.
@@ -2340,8 +2357,10 @@ export function InteractiveGraphVisualization({ onResetLayout, onNodeSelected, i
           // Don't release fx/fy - let the node stay where the user put it
           // The physics will adapt around the fixed position
 
-          // Save the new position to the database
-          saveNodePosition(d.id, d.fx, d.fy);
+          // Save the new position via the coalescing queue (#95) so a drag during
+          // a reflow doesn't add another parallel write.
+          positionSaveQueueRef.current.queue(d.id, d.fx, d.fy);
+          positionSaveQueueRef.current.flush();
 
           // Let the simulation cool to a full stop after the neighbors settle
           setTimeout(() => {

packages/web/src/lib/__tests__/positionSaveQueue.test.ts

@@ -0,0 +1,69 @@
+import { describe, it, expect } from 'vitest';
+import { drainWithLimit, createPositionSaveQueue } from '../positionSaveQueue';
+
+const defer = () => { let resolve!: () => void; const p = new Promise<void>((r) => { resolve = r; }); return { p, resolve }; };
+
+describe('drainWithLimit', () => {
+  it('never exceeds the concurrency limit and runs every item', async () => {
+    let inFlight = 0; let maxInFlight = 0; const done: number[] = [];
+    const items = Array.from({ length: 50 }, (_, i) => i);
+    await drainWithLimit(items, 4, async (i) => {
+      inFlight++; maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((r) => setTimeout(r, 1));
+      inFlight--; done.push(i);
+    });
+    expect(maxInFlight).toBeLessThanOrEqual(4);
+    expect(done.sort((a, b) => a - b)).toEqual(items);
+  });
+
+  it('handles empty + tiny batches without exceeding item count', async () => {
+    let max = 0; let n = 0;
+    await drainWithLimit([1, 2], 10, async () => { n++; max = Math.max(max, n); await Promise.resolve(); n--; });
+    expect(max).toBeLessThanOrEqual(2);
+    await expect(drainWithLimit([], 4, async () => { throw new Error('should not run'); })).resolves.toBeUndefined();
+  });
+});
+
+describe('createPositionSaveQueue', () => {
+  it('coalesces repeated writes for the same node — last position wins', async () => {
+    const saved: Array<{ id: string; x: number; y: number }> = [];
+    const q = createPositionSaveQueue({ save: async (s) => { saved.push(s); } });
+    q.queue('a', 1, 1); q.queue('a', 2, 2); q.queue('a', 9, 9); q.queue('b', 5, 5);
+    expect(q.size).toBe(2); // only 2 distinct nodes pending
+    await q.flush();
+    expect(saved).toHaveLength(2); // one write per node, not 4
+    expect(saved.find((s) => s.id === 'a')).toEqual({ id: 'a', x: 9, y: 9 });
+  });
+
+  it('caps in-flight writes at the configured concurrency', async () => {
+    let inFlight = 0; let max = 0;
+    const q = createPositionSaveQueue({
+      concurrency: 3,
+      save: async () => { inFlight++; max = Math.max(max, inFlight); await new Promise((r) => setTimeout(r, 1)); inFlight--; },
+    });
+    for (let i = 0; i < 30; i++) q.queue(`n${i}`, i, i);
+    await q.flush();
+    expect(max).toBeLessThanOrEqual(3);
+    expect(q.size).toBe(0);
+  });
+
+  it('a save queued mid-flush is not lost (coalesced into a follow-up round)', async () => {
+    const saved: string[] = [];
+    const gate = defer();
+    let first = true;
+    const q = createPositionSaveQueue({
+      concurrency: 1,
+      save: async (s) => {
+        saved.push(s.id);
+        if (first) { first = false; q.queue('late', 0, 0); await gate.p; } // queue during the flush
+      },
+    });
+    q.queue('early', 0, 0);
+    const flushing = q.flush();
+    gate.resolve();
+    await flushing;
+    expect(saved).toContain('early');
+    expect(saved).toContain('late');
+    expect(q.size).toBe(0);
+  });
+});

packages/web/src/lib/positionSaveQueue.ts

@@ -0,0 +1,63 @@
+/**
+ * Coalescing, concurrency-bounded queue for node-position writes (#95).
+ *
+ * A large-graph reflow moves hundreds of nodes at once; persisting each with its
+ * own mutation fired N parallel writes — a D1 write-amplification spike. This
+ * queue (a) coalesces repeated writes for the same node (last position wins) and
+ * (b) drains them with a hard cap on in-flight writes, so a 500-node reflow does
+ * at most `concurrency` writes at a time instead of 500 in parallel. The pure
+ * pieces are unit-tested with an injected `save`.
+ */
+export interface PositionSave { id: string; x: number; y: number; }
+
+/** Run `worker` over `items` with at most `limit` in flight; resolve when all done. */
+export async function drainWithLimit<T>(
+  items: T[],
+  limit: number,
+  worker: (item: T) => Promise<void>,
+): Promise<void> {
+  const queue = [...items];
+  const runNext = async (): Promise<void> => {
+    const item = queue.shift();
+    if (item === undefined) return;
+    await worker(item);
+    return runNext();
+  };
+  const n = Math.max(1, Math.min(limit, items.length));
+  await Promise.all(Array.from({ length: n }, () => runNext()));
+}
+
+export function createPositionSaveQueue(opts: {
+  save: (s: PositionSave) => Promise<void>;
+  concurrency?: number;
+}) {
+  const pending = new Map<string, PositionSave>();
+  const concurrency = opts.concurrency ?? 4;
+  let flushing = false;
+
+  /** Queue/overwrite a node's pending position (last write wins — coalesced). */
+  const queue = (id: string, x: number, y: number): void => {
+    pending.set(id, { id, x, y });
+  };
+
+  /** Drain all pending writes with bounded concurrency. Saves queued mid-flush
+   *  coalesce into a follow-up round so nothing is lost. */
+  const flush = async (): Promise<void> => {
+    if (flushing || pending.size === 0) return;
+    flushing = true;
+    try {
+      const batch = [...pending.values()];
+      pending.clear();
+      await drainWithLimit(batch, concurrency, opts.save);
+    } finally {
+      flushing = false;
+    }
+    if (pending.size > 0) await flush();
+  };
+
+  return {
+    queue,
+    flush,
+    get size() { return pending.size; },
+  };
+}