From becad8de6aa83e00825e2a54dddef080a542f33b Mon Sep 17 00:00:00 2001 From: Matthew Valancy Date: Thu, 18 Jun 2026 19:39:35 -0700 Subject: [PATCH] PERF-95: coalescing, concurrency-bounded node-position saves (avoid N parallel D1 writes) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit persistAllPositions (fired when the sim settles after a reflow) looped every moved node and called saveNodePosition in parallel — a large-graph reflow produced N simultaneous mutations, i.e. N parallel D1 writes (write-amplification spike). Now those writes go through a queue that (a) coalesces repeated writes per node (last position wins) and (b) drains with a hard cap of 4 in-flight writes, so a 500-node reflow does at most 4 writes at a time. Drag-end saves route through the same queue so a drag during a reflow doesn't add another parallel write. - New pure lib/positionSaveQueue.ts (5 unit tests): drainWithLimit (never exceeds the concurrency limit, runs every item) + createPositionSaveQueue (coalesce last-write-wins; cap in-flight; saves queued mid-flush aren't lost). - Queue bound to the latest saveNodePosition via a ref (created once). Verified: positionSaveQueue 5/5; web typecheck clean; THE GATE 5/5 incl. the layout-persistence test (an arranged node still survives a reload through the queue). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../InteractiveGraphVisualization.tsx | 27 ++++++-- .../lib/__tests__/positionSaveQueue.test.ts | 69 +++++++++++++++++++ packages/web/src/lib/positionSaveQueue.ts | 63 +++++++++++++++++ 3 files changed, 155 insertions(+), 4 deletions(-) create mode 100644 packages/web/src/lib/__tests__/positionSaveQueue.test.ts create mode 100644 packages/web/src/lib/positionSaveQueue.ts diff --git a/packages/web/src/components/InteractiveGraphVisualization.tsx b/packages/web/src/components/InteractiveGraphVisualization.tsx index 512cd728..2780e8f0 100644 --- a/packages/web/src/components/InteractiveGraphVisualization.tsx +++ b/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(() => { diff --git a/packages/web/src/lib/__tests__/positionSaveQueue.test.ts b/packages/web/src/lib/__tests__/positionSaveQueue.test.ts new file mode 100644 index 00000000..2cff37f3 --- /dev/null +++ b/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((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); + }); +}); diff --git a/packages/web/src/lib/positionSaveQueue.ts b/packages/web/src/lib/positionSaveQueue.ts new file mode 100644 index 00000000..73a62c42 --- /dev/null +++ b/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( + items: T[], + limit: number, + worker: (item: T) => Promise, +): Promise { + const queue = [...items]; + const runNext = async (): Promise => { + 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; + concurrency?: number; +}) { + const pending = new Map(); + 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 => { + 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; }, + }; +}