From f358b4e95476468b592f696fd8c639b605ebfc31 Mon Sep 17 00:00:00 2001 From: Le Ha Quang Date: Wed, 26 Aug 2026 15:11:06 +0200 Subject: [PATCH 1/2] refactor(sync): extract submitLocalBatch as the single local intake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Local batches enter the sync coordinator through two paths today — the store-commit path (attachSync.sendBatch → rebase set + pump) and a direct oplog write (getBoardPersistenceRef().record) that skips the rebase set and send trigger, desyncing synced boards. Extract `submitLocalBatch(batch)` as the one sync-correct intake (track for rebase + trigger a pump), route the store producer through it, and publish it via `getBoardSyncRef()` so a headless/off-scene producer can enter the same path instead of writing the oplog directly. Behavior-neutral: the store path is unchanged; the ref + seam are additive groundwork for the headless producer (S7b). --- .../board/harness/canvas/use-board-sync-v2.ts | 5 +++ .../board/harness/sync/board-sync-ref.ts | 23 +++++++++++++ .../board/harness/sync/board-sync.test.ts | 34 ++++++++++++++++++- .../features/board/harness/sync/board-sync.ts | 29 ++++++++++++---- 4 files changed, 84 insertions(+), 7 deletions(-) create mode 100644 webui/src/features/board/harness/sync/board-sync-ref.ts diff --git a/webui/src/features/board/harness/canvas/use-board-sync-v2.ts b/webui/src/features/board/harness/canvas/use-board-sync-v2.ts index a033085d..eeb9560c 100644 --- a/webui/src/features/board/harness/canvas/use-board-sync-v2.ts +++ b/webui/src/features/board/harness/canvas/use-board-sync-v2.ts @@ -34,6 +34,7 @@ import { dedupeRepeatUpdates } from "../sync/outbound-dedupe" import { ReconnectSupervisor } from "../sync/reconnect-supervisor" import { attachBoardSync } from "../sync/board-sync" import type { BoardSyncHandle } from "../sync/board-sync" +import { setBoardSyncRef } from "../sync/board-sync-ref" import { createWebSocketRelay } from "../sync/ws-relay" import { applyGraphToStore } from "../persist/snapshot-load" import { applyContentToStore } from "@/features/board/persist/local/apply-content" @@ -152,6 +153,9 @@ export const useBoardSyncV2 = ( enrichOutbound: (batch) => enrichEdgeMidpoints(dedupeRepeatUpdates(batch), store), coalesceMs: 75, // merge a burst (e.g. rotate's per-tick ops) into one send }) + // Publish the sync-correct intake so headless/cross-layer writers enter + // the same rebase + send path instead of writing the oplog directly. + setBoardSyncRef(handle) }) }) .catch((err) => { @@ -167,6 +171,7 @@ export const useBoardSyncV2 = ( cancelled = true document.removeEventListener("visibilitychange", onVisibility) supervisor?.stop() + setBoardSyncRef(null) handle?.detach() detachPersist?.() setBoardPersistenceRef(null) diff --git a/webui/src/features/board/harness/sync/board-sync-ref.ts b/webui/src/features/board/harness/sync/board-sync-ref.ts new file mode 100644 index 00000000..1fca2feb --- /dev/null +++ b/webui/src/features/board/harness/sync/board-sync-ref.ts @@ -0,0 +1,23 @@ +import type { BoardSyncHandle } from "./board-sync" + + +/** + * Module ref to the active synced board's sync coordinator. Mirrors + * `board-persistence-ref` so code outside the harness (e.g. a headless, + * cross-layer writer) can enter the ONE sync-correct local-batch intake + * (`submitLocalBatch`) instead of writing the oplog directly — which would + * skip the rebase set + send trigger and desync a synced board. + * + * Set by the v2 sync mount, cleared on unmount / scope change. `null` on a + * purely local board (no relay) — callers fall back to the persistence ref, + * which is sync-correct there because a local board has no outbox to desync. + */ +let _sync: BoardSyncHandle | null = null + + +export const setBoardSyncRef = (s: BoardSyncHandle | null): void => { + _sync = s +} + + +export const getBoardSyncRef = (): BoardSyncHandle | null => _sync diff --git a/webui/src/features/board/harness/sync/board-sync.test.ts b/webui/src/features/board/harness/sync/board-sync.test.ts index b1ab2334..84acc6c3 100644 --- a/webui/src/features/board/harness/sync/board-sync.test.ts +++ b/webui/src/features/board/harness/sync/board-sync.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest" -import { asBatchId, asClientId, asNodeId, type OpBatch } from "@canvas-harness/core" +import { asBatchId, asClientId, asNodeId, type Node, type OpBatch } from "@canvas-harness/core" import type { CanvasStore } from "@canvas-harness/core" +import { makeBatch } from "@/features/board/harness/make-batch" import { addEdge, addNode, freshStore } from "@/test/canvas" import type { DimNodeData } from "@/features/board/model" import { labelText } from "@/features/board/model" @@ -107,6 +108,37 @@ const setLabel = (store: CanvasStore, id: string, label: string): void => { describe("board sync coordinator", () => { + it("submitLocalBatch enters the same send + rebase path as a store commit", async () => { + const relay = new MemoryRelay() + const a = makeClient(relay, "A") + const b = makeClient(relay, "B") + + // A headless producer: a batch NOT applied to A's live store. It records to + // the oplog itself (as a real off-scene caller does), then enters the shared + // intake — no direct store write. + const node = { + id: asNodeId("h1"), + type: "rect", + x: 0, + y: 0, + w: 100, + h: 50, + angle: 0, + groups: [], + data: { label: { markdown: "headless" }, meta: { v: 1, createdAt: 0, updatedAt: 0 } }, + } as unknown as Node + const batch = makeBatch(a.store, "local", [{ type: "node.add", node }]) + a.persistence.record(batch) + a.sync.submitLocalBatch(batch) + await a.sync.settle() + await b.sync.settle() + + // The peer converges from the headless write, and it shipped as one message — + // proving the batch went through the outbox/rebase path, not a bypass. + expect(relay.log).toHaveLength(1) + expect(ids(b.store)).toEqual(["h1"]) + }) + it("propagates a local edit to the other client (no self-echo, no dupes)", async () => { const relay = new MemoryRelay() const a = makeClient(relay, "A") diff --git a/webui/src/features/board/harness/sync/board-sync.ts b/webui/src/features/board/harness/sync/board-sync.ts index 38459bbe..a3ee904a 100644 --- a/webui/src/features/board/harness/sync/board-sync.ts +++ b/webui/src/features/board/harness/sync/board-sync.ts @@ -89,6 +89,15 @@ export type BoardSyncOptions = { export type BoardSyncHandle = { + /** + * The single sync-correct intake for a locally-produced batch: track it as an + * unacked rebase entry and trigger a pump. The store producer (via + * `attachSync.sendBatch`) is one caller; a headless / off-scene producer is + * another — both share the same rebase set + send path, so nothing desyncs. + * The batch must already be recorded to the oplog (the store path is recorded + * by `persistence`; a headless caller records before submitting). + */ + submitLocalBatch: (batch: OpBatch) => void /** Detach sync + persistence wiring and close the connection. */ detach: () => void /** Simulate going offline (close connection; keep editing locally). */ @@ -305,14 +314,21 @@ export const attachBoardSync = (opts: BoardSyncOptions): BoardSyncHandle => { enqueue(pump) } + // The single intake for a locally-produced batch's SYNC side: track it as an + // unacked rebase entry (applied on top of every remote op so local edits stay + // "latest") and trigger a pump. The send source is the outbox, so the batch + // object itself is only used for rebase. Both the store producer (via + // `attachSync.sendBatch`) and future headless producers route through here. + const submitLocalBatch = (batch: OpBatch): void => { + pending.set(batch.id, batch) + schedulePump() + } + const adapter: SyncAdapter = { capabilities: { causalOrdering: true }, - // A local commit: track it as unacked (rebase set) and trigger a pump. The - // send source is the outbox, so the batch itself is only used for rebase. - sendBatch: (batch: OpBatch) => { - pending.set(batch.id, batch) - schedulePump() - }, + // A local (store) commit enters the shared intake — same rebase + send path + // a headless producer uses. + sendBatch: submitLocalBatch, sendPresence: (patch: PresencePatch) => { const state = { ...patch, clientId: opts.clientId } as PresenceState connection?.send({ kind: "presence", clientId: opts.clientId, state }) @@ -337,6 +353,7 @@ export const attachBoardSync = (opts: BoardSyncOptions): BoardSyncHandle => { } return { + submitLocalBatch, detach: () => { clearTimer() detachSync() From a7fb33eb00f6c9ad2fef905edb0743177824d3ee Mon Sep 17 00:00:00 2001 From: Le Ha Quang Date: Wed, 26 Aug 2026 15:17:13 +0200 Subject: [PATCH 2/2] fix(sync): split submitLocalBatch scene vs headless to prevent rebase injection A headless / off-scene batch's ops are not in the loaded store, so adding it to the rebase set would make applyRemote replay those ops into the current scene on the next remote op. Gate rebase tracking behind `scene` (default true for store commits); `scene: false` batches are pumped but not rebased. Strengthen the seam test to drive a remote op after a headless submit and assert the off-scene node is never injected into the submitter's store. --- .../board/harness/sync/board-sync-ref.ts | 7 +-- .../board/harness/sync/board-sync.test.ts | 29 ++++++++---- .../features/board/harness/sync/board-sync.ts | 47 ++++++++++++------- 3 files changed, 54 insertions(+), 29 deletions(-) diff --git a/webui/src/features/board/harness/sync/board-sync-ref.ts b/webui/src/features/board/harness/sync/board-sync-ref.ts index 1fca2feb..dcbab93b 100644 --- a/webui/src/features/board/harness/sync/board-sync-ref.ts +++ b/webui/src/features/board/harness/sync/board-sync-ref.ts @@ -4,9 +4,10 @@ import type { BoardSyncHandle } from "./board-sync" /** * Module ref to the active synced board's sync coordinator. Mirrors * `board-persistence-ref` so code outside the harness (e.g. a headless, - * cross-layer writer) can enter the ONE sync-correct local-batch intake - * (`submitLocalBatch`) instead of writing the oplog directly — which would - * skip the rebase set + send trigger and desync a synced board. + * cross-layer writer) can enter the sync-correct local-batch intake + * (`submitLocalBatch(batch, { scene: false })`) instead of writing the oplog + * directly — a direct write skips the send trigger, so the batch ships only + * opportunistically on the next unrelated pump and desyncs a synced board. * * Set by the v2 sync mount, cleared on unmount / scope change. `null` on a * purely local board (no relay) — callers fall back to the persistence ref, diff --git a/webui/src/features/board/harness/sync/board-sync.test.ts b/webui/src/features/board/harness/sync/board-sync.test.ts index 84acc6c3..efdb81a5 100644 --- a/webui/src/features/board/harness/sync/board-sync.test.ts +++ b/webui/src/features/board/harness/sync/board-sync.test.ts @@ -108,14 +108,18 @@ const setLabel = (store: CanvasStore, id: string, label: string): void => { describe("board sync coordinator", () => { - it("submitLocalBatch enters the same send + rebase path as a store commit", async () => { + it("submitLocalBatch(scene:false) ships a headless batch without injecting it into the local scene", async () => { const relay = new MemoryRelay() const a = makeClient(relay, "A") const b = makeClient(relay, "B") - // A headless producer: a batch NOT applied to A's live store. It records to - // the oplog itself (as a real off-scene caller does), then enters the shared - // intake — no direct store write. + // Seed an in-scene edit on A so there IS a rebasable local entry, to prove + // the headless batch is treated differently from a store commit. + addNode(a.store, "local1", "in-scene") + + // A headless producer: a batch NOT applied to A's live store (e.g. a + // cross-layer write to an unloaded folder). It records to the oplog itself, + // then enters the intake as an off-scene batch — no direct store write. const node = { id: asNodeId("h1"), type: "rect", @@ -129,14 +133,21 @@ describe("board sync coordinator", () => { } as unknown as Node const batch = makeBatch(a.store, "local", [{ type: "node.add", node }]) a.persistence.record(batch) - a.sync.submitLocalBatch(batch) + a.sync.submitLocalBatch(batch, { scene: false }) await a.sync.settle() await b.sync.settle() - // The peer converges from the headless write, and it shipped as one message — - // proving the batch went through the outbox/rebase path, not a bypass. - expect(relay.log).toHaveLength(1) - expect(ids(b.store)).toEqual(["h1"]) + // The peer converges from the headless write (it shipped through the outbox), + // but it never lands in the submitter's own loaded scene. + expect(ids(b.store)).toEqual(["h1", "local1"]) + expect(ids(a.store)).toEqual(["local1"]) // h1 was never applied to A's store + + // A concurrent remote op triggers a rebase. Because the headless batch is NOT + // in the rebase set, its off-scene ops must NOT be injected into A's scene. + addNode(b.store, "b1", "peer") + await b.sync.settle() + await a.sync.settle() + expect(ids(a.store)).toEqual(["b1", "local1"]) // still no h1 }) it("propagates a local edit to the other client (no self-echo, no dupes)", async () => { diff --git a/webui/src/features/board/harness/sync/board-sync.ts b/webui/src/features/board/harness/sync/board-sync.ts index a3ee904a..3dd46968 100644 --- a/webui/src/features/board/harness/sync/board-sync.ts +++ b/webui/src/features/board/harness/sync/board-sync.ts @@ -90,14 +90,18 @@ export type BoardSyncOptions = { export type BoardSyncHandle = { /** - * The single sync-correct intake for a locally-produced batch: track it as an - * unacked rebase entry and trigger a pump. The store producer (via - * `attachSync.sendBatch`) is one caller; a headless / off-scene producer is - * another — both share the same rebase set + send path, so nothing desyncs. - * The batch must already be recorded to the oplog (the store path is recorded - * by `persistence`; a headless caller records before submitting). + * The single sync-correct intake for a locally-produced batch: (maybe) track + * it for rebase and trigger a pump, so it ships through the outbox rather than + * a direct oplog write. The store producer (via `attachSync.sendBatch`) is one + * caller; a headless / off-scene producer is another. + * + * `scene: false` marks a batch whose ops are NOT in the loaded store (a + * cross-layer / headless write): it is pumped but kept OUT of the rebase set, + * since replaying its ops would inject off-layer nodes into the current scene. + * A `scene: false` caller must record the batch to the oplog itself (store + * commits are already recorded by `persistence`). */ - submitLocalBatch: (batch: OpBatch) => void + submitLocalBatch: (batch: OpBatch, opts?: { scene?: boolean }) => void /** Detach sync + persistence wiring and close the connection. */ detach: () => void /** Simulate going offline (close connection; keep editing locally). */ @@ -314,21 +318,30 @@ export const attachBoardSync = (opts: BoardSyncOptions): BoardSyncHandle => { enqueue(pump) } - // The single intake for a locally-produced batch's SYNC side: track it as an - // unacked rebase entry (applied on top of every remote op so local edits stay - // "latest") and trigger a pump. The send source is the outbox, so the batch - // object itself is only used for rebase. Both the store producer (via - // `attachSync.sendBatch`) and future headless producers route through here. - const submitLocalBatch = (batch: OpBatch): void => { - pending.set(batch.id, batch) + // The single intake for a locally-produced batch's SYNC side: (maybe) track it + // for rebase and trigger a pump. The send source is the outbox, so the batch + // object is only used for the in-scene rebase. Both the store producer (via + // `attachSync.sendBatch`) and headless producers route through here. + // + // `scene` splits the two kinds of local batch — they need different handling: + // - `true` (default, store commits): the ops WERE applied to the live store + // optimistically, so track the batch as an unacked rebase entry — it's + // undone + replayed on top of every remote op so the local edit stays + // "latest". This is today's `sendBatch` behavior, unchanged. + // - `false` (headless / off-scene, e.g. a cross-layer write): the ops are NOT + // in the loaded store, so it must NOT enter the rebase set — `applyRemote` + // would otherwise inject the off-layer ops into the current scene on the + // next remote op. It only needs to be recorded (by the caller) + pumped; + // `serverSeq` is stamped on ack via the outbox path like any sent record. + const submitLocalBatch = (batch: OpBatch, { scene = true }: { scene?: boolean } = {}): void => { + if (scene) pending.set(batch.id, batch) schedulePump() } const adapter: SyncAdapter = { capabilities: { causalOrdering: true }, - // A local (store) commit enters the shared intake — same rebase + send path - // a headless producer uses. - sendBatch: submitLocalBatch, + // A local store commit: its ops are in the live scene → rebase-tracked. + sendBatch: (batch: OpBatch) => submitLocalBatch(batch), sendPresence: (patch: PresencePatch) => { const state = { ...patch, clientId: opts.clientId } as PresenceState connection?.send({ kind: "presence", clientId: opts.clientId, state })