diff --git a/decnet_web/src/components/MazeNET/MazeNET.tsx b/decnet_web/src/components/MazeNET/MazeNET.tsx index d7a148ef..acab7f50 100644 --- a/decnet_web/src/components/MazeNET/MazeNET.tsx +++ b/decnet_web/src/components/MazeNET/MazeNET.tsx @@ -124,10 +124,14 @@ async function _dropArchetype( { name, services: dServices, x: nx, y: ny, decky_config: { archetype: drag.slug } }, overNetId, net.name, ); - if (dRes.kind !== 'applied') return; - const decky = dRes.data; + // On a live topology the add is STAGED (kind 'enqueued') — no server + // uuid yet, so use a temp id and render the node optimistically. + // Without this the decky is invisible until UPDATE + refetch. Mirrors + // the pending-net placeholder in _dropNetwork. + const id = dRes.kind === 'applied' ? dRes.data.uuid : `pending-decky-${name}`; + const nodeName = dRes.kind === 'applied' ? dRes.data.name : name; setNodes((p) => [...p, { - kind: 'decky', id: decky.uuid, netId: overNetId, name: decky.name, + kind: 'decky', id, netId: overNetId, name: nodeName, archetype: drag.slug, services: dServices, status: 'idle', x: nx, y: ny, } as DeckyNode]); } catch (err) { @@ -181,7 +185,7 @@ const MazeNET: React.FC = () => { const { nets, setNets, nodes, setNodes, edges, setEdges, topoMeta, services, archetypes, - loadErr, actionErr, commitErr, clearCommitErr, flashErr, + loadErr, actionErr, commitErr, clearCommitErr, flashErr, setRefetchPaused, deploying, onDeploy, streamLive, lastEventAt, streamEnabled, refetch, @@ -531,9 +535,37 @@ const MazeNET: React.FC = () => { }, []); const canDeploy = topoStatus === 'pending' && nets.length > 0; + const liveTopo = topoStatus === 'active' || topoStatus === 'degraded'; + const pendingCount = editor.pendingCount; + const [committing, setCommitting] = useState(false); const deckyNodes = nodes.filter((n) => n.kind === 'decky'); const runningDeckies = deckyNodes.filter((n) => n.status === 'active').length; + /* UPDATE button: flush the staged changeset as one sequential mutation + batch. SSE refetch is paused so per-mutation applied events don't wipe + still-staged placeholders mid-batch; one refetch reconciles at the end + (success or failure). A failed op throws MutationFailedError, which + flashErr pins as a persistent banner. */ + const handleCommit = useCallback(async () => { + if (!topologyId || pendingCount === 0) return; + const n = pendingCount; + setCommitting(true); + setRefetchPaused(true); + try { + const applied = await editor.commitStaged(); + pushToast({ + text: `UPDATED · ${applied} CHANGE${applied === 1 ? '' : 'S'}`, + tone: 'matrix', icon: 'check-circle', + }); + } catch (err) { + flashErr(err, `update failed after ${n - editor.pendingCount}/${n} changes`); + } finally { + setRefetchPaused(false); + await refetch(); + setCommitting(false); + } + }, [editor, topologyId, pendingCount, refetch, pushToast, flashErr, setRefetchPaused]); + return (
@@ -592,8 +624,13 @@ const MazeNET: React.FC = () => { {fullscreen ? : } {fullscreen ? ' EXIT FULL' : ' FULLSCREEN'} - - + {liveTopo ? ( + + ) : ( + + )}
diff --git a/decnet_web/src/components/MazeNET/useTopologyData.ts b/decnet_web/src/components/MazeNET/useTopologyData.ts index 60431d90..a4031032 100644 --- a/decnet_web/src/components/MazeNET/useTopologyData.ts +++ b/decnet_web/src/components/MazeNET/useTopologyData.ts @@ -1,5 +1,5 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -import { useCallback, useEffect, useState } from 'react'; +import { useCallback, useEffect, useRef, useState } from 'react'; import type { ApiError } from '../../utils/api'; import type { Net, MazeNode, Edge } from './types'; import { DEFAULT_SERVICES, ARCHETYPES as DEFAULT_ARCHETYPES } from './data'; @@ -48,6 +48,10 @@ export interface UseTopologyDataResult { commitErr: string | null; clearCommitErr: () => void; flashErr: (err: unknown, fallback: string) => void; + /** Pause SSE-driven refetch while a commit batch is in flight, so the + * per-mutation ``applied`` events don't wipe the still-staged + * placeholders mid-batch. The committer does one refetch at the end. */ + setRefetchPaused: (paused: boolean) => void; // Deploy deploying: boolean; @@ -87,6 +91,11 @@ export function useTopologyData( const clearCommitErr = useCallback(() => setCommitErr(null), []); + const refetchPausedRef = useRef(false); + const setRefetchPaused = useCallback((paused: boolean) => { + refetchPausedRef.current = paused; + }, []); + const flashErr = useCallback((err: unknown, fallback: string) => { // A failed live mutation is loud + persistent: the queue halted and // the topology probably degraded — don't let it vanish in 4s. @@ -153,7 +162,8 @@ export function useTopologyData( if (event.name === 'mutation.applied' || event.name === 'mutation.failed' || event.name === 'status') { - void refetch(); + // Suppressed mid-commit — the committer drives one refetch at the end. + if (!refetchPausedRef.current) void refetch(); } // Live service mutations from another tab / admin: optimistically // patch local state so the chip set reflects shape without a full @@ -203,7 +213,7 @@ export function useTopologyData( edges, setEdges, topoMeta, services, archetypes, - loadErr, actionErr, commitErr, clearCommitErr, flashErr, + loadErr, actionErr, commitErr, clearCommitErr, flashErr, setRefetchPaused, deploying, onDeploy, streamLive, lastEventAt, streamEnabled, refetch, diff --git a/decnet_web/src/components/MazeNET/useTopologyEditor.test.ts b/decnet_web/src/components/MazeNET/useTopologyEditor.test.ts index 508df954..241cd2c5 100644 --- a/decnet_web/src/components/MazeNET/useTopologyEditor.test.ts +++ b/decnet_web/src/components/MazeNET/useTopologyEditor.test.ts @@ -19,27 +19,32 @@ const editorFor = (api: MazeApi, topoVersion = 5) => useTopologyEditor({ api, topoStatus: 'active', topoVersion }), ); -describe('useTopologyEditor live mutation queue', () => { - it('serialises concurrent submits and advances expected_version per enqueue', async () => { +describe('useTopologyEditor live staging', () => { + it('stages live edits without sending; commit flushes them in order with a version cursor', async () => { const enqueue = vi.fn().mockResolvedValue({ mutation_id: 'm', state: 'pending' }); const api = buildApi({ enqueueMutation: enqueue }); const { result } = editorFor(api, 5); - // Fire two structural ops in the SAME tick — the pre-fix bug was both - // sending expected_version=5 and the loser 409ing. await act(async () => { - await Promise.all([ - result.current.createLan('t', { name: 'a', is_dmz: false, x: 0, y: 0 }), - result.current.deleteLan('t', 'lid', 'b'), - ]); + await result.current.createLan('t', { name: 'a', is_dmz: false, x: 0, y: 0 }); + await result.current.deleteLan('t', 'lid', 'b'); + }); + + // Staged, not sent. + expect(result.current.pendingCount).toBe(2); + expect(enqueue).not.toHaveBeenCalled(); + + await act(async () => { + await result.current.commitStaged(); }); expect(enqueue).toHaveBeenCalledTimes(2); expect(enqueue.mock.calls[0][3]).toBe(5); // first uses server version expect(enqueue.mock.calls[1][3]).toBe(6); // second advanced by the cursor + expect(result.current.pendingCount).toBe(0); }); - it('throws MutationFailedError on a failed mutation but keeps the queue alive', async () => { + it('commit stops loudly on a failed op, keeps the remainder, and retries cleanly', async () => { const wait = vi .fn() .mockResolvedValueOnce({ state: 'failed', reason: 'post-apply validation failed: IP_COLLISION' }) @@ -48,16 +53,36 @@ describe('useTopologyEditor live mutation queue', () => { const { result } = editorFor(api, 1); await act(async () => { - await expect( - result.current.createLan('t', { name: 'a', is_dmz: false, x: 0, y: 0 }), - ).rejects.toBeInstanceOf(MutationFailedError); + await result.current.createLan('t', { name: 'a', is_dmz: false, x: 0, y: 0 }); + await result.current.deleteLan('t', 'lid', 'b'); }); + expect(result.current.pendingCount).toBe(2); - // A failed op must not wedge the chain — the next submit still resolves. await act(async () => { - await expect( - result.current.deleteLan('t', 'lid', 'b'), - ).resolves.toEqual({ kind: 'enqueued', mutationId: 'm' }); + await expect(result.current.commitStaged()).rejects.toBeInstanceOf(MutationFailedError); }); + // First op failed → nothing applied → both stay staged for retry. + expect(result.current.pendingCount).toBe(2); + + // Retry: waitForMutation now resolves 'applied' for both. + await act(async () => { + await result.current.commitStaged(); + }); + expect(result.current.pendingCount).toBe(0); + }); + + it('discardStaged drops the batch without sending', async () => { + const enqueue = vi.fn().mockResolvedValue({ mutation_id: 'm', state: 'pending' }); + const api = buildApi({ enqueueMutation: enqueue }); + const { result } = editorFor(api, 1); + + await act(async () => { + await result.current.createLan('t', { name: 'a', is_dmz: false, x: 0, y: 0 }); + }); + expect(result.current.pendingCount).toBe(1); + + act(() => result.current.discardStaged()); + expect(result.current.pendingCount).toBe(0); + expect(enqueue).not.toHaveBeenCalled(); }); }); diff --git a/decnet_web/src/components/MazeNET/useTopologyEditor.ts b/decnet_web/src/components/MazeNET/useTopologyEditor.ts index 1a94a935..70e11707 100644 --- a/decnet_web/src/components/MazeNET/useTopologyEditor.ts +++ b/decnet_web/src/components/MazeNET/useTopologyEditor.ts @@ -16,7 +16,7 @@ * primitive because mutation ops are name-keyed while direct CRUD is * uuid-keyed. Callers plumb both. */ -import { useCallback, useEffect, useMemo, useRef } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import type { CreateDeckyBody, CreateLanBody, @@ -53,6 +53,12 @@ export type PrimitiveResult = | { kind: 'applied'; data: T } | { kind: 'enqueued'; mutationId: string }; +interface StagedOp { + topologyId: string; + op: MutationOp; + payload: Record; +} + export interface UseTopologyEditor { createLan(topologyId: string, body: CreateLanBody): Promise>; updateLan( @@ -108,6 +114,18 @@ export interface UseTopologyEditor { deckyName: string, lanName: string, ): Promise>; + + // ── Staging (live topologies only) ─────────────────────────────────── + /** Count of staged-but-unsent live edits. 0 on a pending topology. */ + pendingCount: number; + /** Flush staged edits as one sequential mutation batch (version-cursored, + * each awaited to a terminal state). Stops on the first failure, keeping + * the failing op + remainder staged, and rethrows MutationFailedError so + * the caller can surface it. Resolves with the count applied. */ + commitStaged(): Promise; + /** Drop all staged edits without sending (paired with a refetch to wipe + * their optimistic placeholders). */ + discardStaged(): void; } export function useTopologyEditor( @@ -116,19 +134,19 @@ export function useTopologyEditor( const { api, topoStatus, topoVersion } = opts; const live = topoStatus === 'active' || topoStatus === 'degraded'; - // Serialised mutation submission. Two problems this solves, both - // proven against the live backend: - // 1. expected_version is bumped at ENQUEUE (not at apply), so two - // ops fired back-to-back race: whichever HTTP request the server - // sees second carries a stale version and 409s. We chain submits - // so only one enqueue is ever in flight, in submission order. - // 2. A failed mutation silently degrades the topology. We await each - // mutation to a terminal state and throw MutationFailedError on - // 'failed' so the caller can surface it loudly. - const chainRef = useRef>(Promise.resolve()); + // Live edits STAGE rather than send. Each live primitive records its + // op here and returns immediately; nothing hits the backend until + // commitStaged() flushes the batch (the UPDATE button). Staging is what + // lets the user assemble a coherent changeset (e.g. add a net AND bridge + // it) before any of it lands — and it kills the per-action SSE-refetch + // race that duplicated optimistic placeholders. + const [staged, setStaged] = useState([]); + // Optimistic expected_version cursor. enqueue bumps the server version - // by exactly 1, so we advance locally rather than waiting for a refetch - // between queued ops (onReparent fires detach + attach in one handler). + // by exactly 1, so within a commit batch we advance locally rather than + // waiting for a refetch between ops. NB: a *failed* mutation still bumps + // the version (the check happens at enqueue), so we advance after enqueue + // regardless of the apply outcome. const cursorRef = useRef(topoVersion); useEffect(() => { // Adopt a higher server version (a refetch landed, or another editor @@ -139,25 +157,45 @@ export function useTopologyEditor( const submit = useCallback( (topologyId: string, op: MutationOp, payload: Record): Promise => { - const task = chainRef.current.then(async () => { - const expected = cursorRef.current; - const res = await api.enqueueMutation(topologyId, op, payload, expected); - cursorRef.current = expected + 1; - const row = await api.waitForMutation(topologyId, res.mutation_id); - if (row.state === 'failed') { - throw new MutationFailedError(op, row.reason ?? 'unknown reason'); - } - return res.mutation_id; - }); - // Keep the chain alive after a rejection so one failed op doesn't - // wedge every subsequent submit. - chainRef.current = task.then(() => undefined, () => undefined); - return task; + setStaged((prev) => [...prev, { topologyId, op, payload }]); + // Sentinel id — callers thread this into optimistic state but it + // never reaches the backend; the post-commit refetch reconciles to + // real ids. + return Promise.resolve('staged'); }, - [api], + [], ); - return useMemo(() => ({ + const discardStaged = useCallback(() => setStaged([]), []); + + const commitStaged = useCallback(async (): Promise => { + const ops = staged; + if (ops.length === 0) return 0; + let applied = 0; + try { + for (const o of ops) { + const expected = cursorRef.current; + const res = await api.enqueueMutation(o.topologyId, o.op, o.payload, expected); + // Advance even if the apply fails below — enqueue already bumped + // the server version. + cursorRef.current = expected + 1; + const row = await api.waitForMutation(o.topologyId, res.mutation_id); + if (row.state === 'failed') { + throw new MutationFailedError(o.op, row.reason ?? 'unknown reason'); + } + applied += 1; + } + setStaged([]); + return applied; + } catch (err) { + // Drop the applied prefix; keep the failing op + the rest so the user + // can fix and retry without re-staging everything. + setStaged(ops.slice(applied)); + throw err; + } + }, [staged, api]); + + const primitives = useMemo>(() => ({ // ── LAN ──────────────────────────────────────────────────────────── async createLan(topologyId, body) { if (!live) { @@ -274,4 +312,9 @@ export function useTopologyEditor( return { kind: 'enqueued', mutationId }; }, }), [api, live, submit]); + + return useMemo( + () => ({ ...primitives, pendingCount: staged.length, commitStaged, discardStaged }), + [primitives, staged.length, commitStaged, discardStaged], + ); }