collab: passive pull facade, sibling mirror without BoardRoom wake (load-path-rework 0004)
- GatewayDocFacade: passiveSync (Step1 on subscribe + on touched, answered by the gateway from at-rest state), onReset; activate() after a passive fill still sends act + a participant Step1. - sibling-restage subscribes passive+passiveSync; reset drops the watch (no flush) and re-dials while a peer still has the sheet open. - files-watch: upload/job hints on room-backed paths restage + announce onRoomBackedChanged; sheet-manager.invalidate() drops a parked room. - kicad-binding: normalize server-serialized bodies on the editorMatchesDoc path (runner-seeded ydocs carry kicad-cli's serialization). - pcbjam-shared -> 0f4d3a1 (reset control, kicadDocToYdocUpdate). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UQsdqaX57xzcCWUjqP7ctV
This commit is contained in:
parent
7a9aeb5181
commit
602f5c6fed
13 changed files with 332 additions and 13 deletions
|
|
@ -1 +1 @@
|
||||||
Subproject commit 4573a7c2707d891fc2107db24b23023c0dae849f
|
Subproject commit 0f4d3a1e336fd5398fc071bd3a4181983b7eb3fb
|
||||||
|
|
@ -1324,6 +1324,9 @@ export function WasmTool({
|
||||||
onNewPath: (p) => {
|
onNewPath: (p) => {
|
||||||
if (p.endsWith(".kicad_sch")) void sheetManagerRef.current?.onboard(p);
|
if (p.endsWith(".kicad_sch")) void sheetManagerRef.current?.onboard(p);
|
||||||
},
|
},
|
||||||
|
// A cold room replaced at rest (re-upload / resave install, 0004
|
||||||
|
// §2.5): a parked sheet doc must not carry the old epoch.
|
||||||
|
onRoomBackedChanged: (p) => sheetManagerRef.current?.invalidate(p),
|
||||||
onTargetChanged: (c) => {
|
onTargetChanged: (c) => {
|
||||||
const who = c.by ?? "a collaborator";
|
const who = c.by ?? "a collaborator";
|
||||||
setStatus(
|
setStatus(
|
||||||
|
|
|
||||||
|
|
@ -65,7 +65,7 @@ describe("files hint router (project-sync 0002 §3)", () => {
|
||||||
expect(restaged).toEqual(["x.kicad_pro"]);
|
expect(restaged).toEqual(["x.kicad_pro"]);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("room-backed paths are never restaged from the row", async () => {
|
it("room-backed paths are never restaged from an EDITOR-origin row", async () => {
|
||||||
const { router, restaged, events } = makeRouter();
|
const { router, restaged, events } = makeRouter();
|
||||||
router.handle(1, [ch({ path: "root.kicad_sch", revision: 9, by: "peer" })]);
|
router.handle(1, [ch({ path: "root.kicad_sch", revision: 9, by: "peer" })]);
|
||||||
await tick();
|
await tick();
|
||||||
|
|
@ -73,6 +73,29 @@ describe("files hint router (project-sync 0002 §3)", () => {
|
||||||
expect(events).toEqual([]);
|
expect(events).toEqual([]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("an upload/job over a room-backed path IS restaged and announced (0004 §2.5)", async () => {
|
||||||
|
const events: string[] = [];
|
||||||
|
const { router, restaged } = makeRouter({
|
||||||
|
onRoomBackedChanged: (p) => events.push(`replaced:${p}`),
|
||||||
|
});
|
||||||
|
router.handle(1, [ch({ path: "root.kicad_sch", revision: 9, origin: "upload", by: "peer" })]);
|
||||||
|
router.handle(2, [ch({ path: "root.kicad_sch", revision: 10, origin: "job" })]);
|
||||||
|
await tick();
|
||||||
|
expect(restaged).toEqual(["root.kicad_sch"]);
|
||||||
|
expect(events).toEqual(["replaced:root.kicad_sch", "replaced:root.kicad_sch"]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("an upload over the room-backed open target only notifies", async () => {
|
||||||
|
const { router, restaged, events } = makeRouter({
|
||||||
|
targetPath: "root.kicad_sch",
|
||||||
|
isRoomBacked: () => true,
|
||||||
|
});
|
||||||
|
router.handle(1, [ch({ path: "root.kicad_sch", revision: 9, origin: "upload", by: "peer" })]);
|
||||||
|
await tick();
|
||||||
|
expect(restaged).toEqual([]);
|
||||||
|
expect(events).toEqual(["target:root.kicad_sch@9"]);
|
||||||
|
});
|
||||||
|
|
||||||
it("Tier 2: the open target on the PUT channel only notifies", async () => {
|
it("Tier 2: the open target on the PUT channel only notifies", async () => {
|
||||||
const { router, restaged, events } = makeRouter({ isRoomBacked: () => false });
|
const { router, restaged, events } = makeRouter({ isRoomBacked: () => false });
|
||||||
router.handle(1, [ch({ path: "board.kicad_pcb", revision: 4, by: "peer" })]);
|
router.handle(1, [ch({ path: "board.kicad_pcb", revision: 4, by: "peer" })]);
|
||||||
|
|
|
||||||
|
|
@ -17,8 +17,13 @@ import { connectProvider, type ProviderConfig, type YjsProvider } from "./provid
|
||||||
* no room): `onTargetChanged` — the host shows a reload/conflict notice;
|
* no room): `onTargetChanged` — the host shows a reload/conflict notice;
|
||||||
* the CAS lane keeps guarding the next save.
|
* the CAS lane keeps guarding the next save.
|
||||||
*
|
*
|
||||||
* Room-backed paths (listing hasYdoc/isLive) are ignored: the room is the
|
* Room-backed paths (listing hasYdoc/isLive) are ignored for EDITOR-origin
|
||||||
* truth there and already carries its own `touched`/frames.
|
* hints: the room is the truth there and already carries its own
|
||||||
|
* `touched`/frames. An `upload`/`job` hint on a room-backed path is an
|
||||||
|
* at-rest replacement of a cold doc (load-path-rework 0004 §2.5 — a
|
||||||
|
* re-upload, or the resave job installing its ydoc): it IS restaged, and
|
||||||
|
* `onRoomBackedChanged` lets the eeschema sheet pool drop a parked doc that
|
||||||
|
* would otherwise carry the old epoch into its next activation.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
export const FILES_RESTAGE_DEBOUNCE_MS = 400;
|
export const FILES_RESTAGE_DEBOUNCE_MS = 400;
|
||||||
|
|
@ -53,6 +58,8 @@ export interface FilesWatchOptions {
|
||||||
restage: (relPath: string, bytes: Uint8Array) => void;
|
restage: (relPath: string, bytes: Uint8Array) => void;
|
||||||
/** A path not in the boot listing appeared (a peer's "Add Sheet"). */
|
/** A path not in the boot listing appeared (a peer's "Add Sheet"). */
|
||||||
onNewPath?: (relPath: string) => void;
|
onNewPath?: (relPath: string) => void;
|
||||||
|
/** A room-backed path was replaced at rest by an upload/job (0004 §2.5). */
|
||||||
|
onRoomBackedChanged?: (relPath: string) => void;
|
||||||
onTargetChanged?: (change: GatewayFileChange) => void;
|
onTargetChanged?: (change: GatewayFileChange) => void;
|
||||||
onListingStale?: () => void;
|
onListingStale?: () => void;
|
||||||
log: (m: string) => void;
|
log: (m: string) => void;
|
||||||
|
|
@ -107,7 +114,15 @@ export function createFilesHintRouter(opts: FilesWatchOptions) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
opts.rememberObserved(change.path, change.revision);
|
opts.rememberObserved(change.path, change.revision);
|
||||||
if (opts.isRoomBacked(change.path)) continue; // the room owns it
|
if (opts.isRoomBacked(change.path)) {
|
||||||
|
// The room owns editor writes; an at-rest replacement is different.
|
||||||
|
if (change.origin === "editor" || change.deleted) continue;
|
||||||
|
if (change.path === opts.targetPath) {
|
||||||
|
opts.onTargetChanged?.(change);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
opts.onRoomBackedChanged?.(change.path);
|
||||||
|
}
|
||||||
if (change.path === opts.targetPath) {
|
if (change.path === opts.targetPath) {
|
||||||
opts.onTargetChanged?.(change);
|
opts.onTargetChanged?.(change);
|
||||||
continue;
|
continue;
|
||||||
|
|
|
||||||
|
|
@ -236,6 +236,118 @@ describe("gateway facade — passive warm pool (the laziness contract)", () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("gateway facade — passive pull (load-path-rework 0004 §2.2)", () => {
|
||||||
|
const passiveSyncOpts = (docPath: string) => ({ ...facadeOpts(docPath, true), passiveSync: true });
|
||||||
|
|
||||||
|
it("sends Step1 on subscribe while staying passive; Step2 fills the doc, whenSynced resolves", async () => {
|
||||||
|
newProject();
|
||||||
|
const doc = new Y.Doc();
|
||||||
|
const facade = new GatewayDocFacade(doc, passiveSyncOpts("c.kicad_sch"));
|
||||||
|
track(facade, doc);
|
||||||
|
const ws = FakeWebSocket.instances.at(-1)!;
|
||||||
|
ws.open();
|
||||||
|
const [sub] = ws.controls();
|
||||||
|
expect(sub).toMatchObject({ t: "sub", mode: "passive" });
|
||||||
|
expect(ws.controls().some((m) => m.t === "act")).toBe(false);
|
||||||
|
const step1 = ws.frames().find((f) => f.type === 0);
|
||||||
|
expect(step1).toBeTruthy();
|
||||||
|
|
||||||
|
const serverDoc = new Y.Doc();
|
||||||
|
serverDoc.getMap("m").set("k", "at-rest");
|
||||||
|
cleanups.push(() => serverDoc.destroy());
|
||||||
|
ws.receiveFrame(sub!.ch, step2From(serverDoc, step1!.frame));
|
||||||
|
await facade.whenSynced();
|
||||||
|
expect(doc.getMap("m").get("k")).toBe("at-rest");
|
||||||
|
// Applying the Step2 must not have pushed anything back (no participant writes).
|
||||||
|
expect(ws.frames().filter((f) => f.type === 0).length).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("re-pulls with its own state vector on every touched", async () => {
|
||||||
|
newProject();
|
||||||
|
const doc = new Y.Doc();
|
||||||
|
const facade = new GatewayDocFacade(doc, passiveSyncOpts("c.kicad_sch"));
|
||||||
|
track(facade, doc);
|
||||||
|
const ws = FakeWebSocket.instances.at(-1)!;
|
||||||
|
ws.open();
|
||||||
|
const ch = ws.controls()[0]!.ch;
|
||||||
|
const serverDoc = new Y.Doc();
|
||||||
|
serverDoc.getMap("m").set("k", 1);
|
||||||
|
cleanups.push(() => serverDoc.destroy());
|
||||||
|
ws.receiveFrame(ch, step2From(serverDoc, ws.frames().find((f) => f.type === 0)!.frame));
|
||||||
|
await facade.whenSynced();
|
||||||
|
|
||||||
|
let touched = 0;
|
||||||
|
facade.onTouched(() => touched++);
|
||||||
|
ws.sent = [];
|
||||||
|
ws.receiveText(JSON.stringify({ t: "touched", ch }));
|
||||||
|
expect(touched).toBe(1);
|
||||||
|
const step1 = ws.frames().find((f) => f.type === 0);
|
||||||
|
expect(step1).toBeTruthy();
|
||||||
|
// The pull carries a NON-empty SV: the server answers with only the delta.
|
||||||
|
const dec = decoding.createDecoder(step1!.frame.slice());
|
||||||
|
decoding.readVarUint(dec);
|
||||||
|
decoding.readVarUint(dec);
|
||||||
|
const sv = decoding.readVarUint8Array(dec);
|
||||||
|
expect(Y.decodeStateVector(sv).size).toBe(1);
|
||||||
|
serverDoc.getMap("m").set("k", 2);
|
||||||
|
ws.receiveFrame(ch, step2From(serverDoc, step1!.frame));
|
||||||
|
expect(doc.getMap("m").get("k")).toBe(2);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("activate() after a passive fill still sends act + a participant Step1", async () => {
|
||||||
|
newProject();
|
||||||
|
const doc = new Y.Doc();
|
||||||
|
const facade = new GatewayDocFacade(doc, passiveSyncOpts("c.kicad_sch"));
|
||||||
|
track(facade, doc);
|
||||||
|
const ws = FakeWebSocket.instances.at(-1)!;
|
||||||
|
ws.open();
|
||||||
|
const ch = ws.controls()[0]!.ch;
|
||||||
|
const serverDoc = new Y.Doc();
|
||||||
|
serverDoc.getMap("m").set("k", 1);
|
||||||
|
cleanups.push(() => serverDoc.destroy());
|
||||||
|
ws.receiveFrame(ch, step2From(serverDoc, ws.frames().find((f) => f.type === 0)!.frame));
|
||||||
|
await facade.whenSynced();
|
||||||
|
|
||||||
|
ws.sent = [];
|
||||||
|
const syncing = facade.activate();
|
||||||
|
expect(ws.controls()).toEqual([{ t: "act", ch }]);
|
||||||
|
const step1 = ws.frames().find((f) => f.type === 0);
|
||||||
|
expect(step1).toBeTruthy();
|
||||||
|
ws.receiveFrame(ch, step2From(serverDoc, step1!.frame));
|
||||||
|
await syncing;
|
||||||
|
// Now a participant: a second activate is a no-op.
|
||||||
|
ws.sent = [];
|
||||||
|
await facade.activate();
|
||||||
|
expect(ws.controls()).toEqual([]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("reset fires the owner's callback (0004 §2.3)", () => {
|
||||||
|
newProject();
|
||||||
|
const doc = new Y.Doc();
|
||||||
|
const facade = new GatewayDocFacade(doc, passiveSyncOpts("c.kicad_sch"));
|
||||||
|
track(facade, doc);
|
||||||
|
const ws = FakeWebSocket.instances.at(-1)!;
|
||||||
|
ws.open();
|
||||||
|
const ch = ws.controls()[0]!.ch;
|
||||||
|
let resets = 0;
|
||||||
|
facade.onReset(() => resets++);
|
||||||
|
ws.receiveText(JSON.stringify({ t: "reset", ch }));
|
||||||
|
expect(resets).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("a plain passive facade (no passiveSync) still never pulls", () => {
|
||||||
|
newProject();
|
||||||
|
const doc = new Y.Doc();
|
||||||
|
const facade = new GatewayDocFacade(doc, facadeOpts("b.kicad_sch", true));
|
||||||
|
track(facade, doc);
|
||||||
|
const ws = FakeWebSocket.instances.at(-1)!;
|
||||||
|
ws.open();
|
||||||
|
const ch = ws.controls()[0]!.ch;
|
||||||
|
ws.receiveText(JSON.stringify({ t: "touched", ch }));
|
||||||
|
expect(ws.frames().filter((f) => f.type === 0)).toEqual([]);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("gateway facade — suberr is terminal", () => {
|
describe("gateway facade — suberr is terminal", () => {
|
||||||
it("rejects pending and future syncs with CollabSubRejectedError", async () => {
|
it("rejects pending and future syncs with CollabSubRejectedError", async () => {
|
||||||
newProject();
|
newProject();
|
||||||
|
|
|
||||||
|
|
@ -252,6 +252,14 @@ export interface GatewayFacadeOpts {
|
||||||
/** Passive = register interest only (parked warm-pool sheet): no SyncStep1,
|
/** Passive = register interest only (parked warm-pool sheet): no SyncStep1,
|
||||||
* no BoardRoom wake; `touched` hints + awareness still flow. */
|
* no BoardRoom wake; `touched` hints + awareness still flow. */
|
||||||
passive?: boolean;
|
passive?: boolean;
|
||||||
|
/**
|
||||||
|
* Passive PULL (load-path-rework 0004 §2.2): while passive, send SyncStep1
|
||||||
|
* on subscribe and on every `touched` — the gateway answers from the doc's
|
||||||
|
* at-rest state (R2, or the live room's RPC), never by dialing its
|
||||||
|
* BoardRoom. The doc then tracks the server without ever being a
|
||||||
|
* participant. Ignored for active/presence/hint channels.
|
||||||
|
*/
|
||||||
|
passiveSync?: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -264,11 +272,16 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
private readonly ch: number;
|
private readonly ch: number;
|
||||||
private readonly isPresence: boolean;
|
private readonly isPresence: boolean;
|
||||||
private mode: GatewaySubMode;
|
private mode: GatewaySubMode;
|
||||||
|
private readonly passiveSync: boolean;
|
||||||
private dead: CollabSubRejectedError | null = null;
|
private dead: CollabSubRejectedError | null = null;
|
||||||
private destroyed = false;
|
private destroyed = false;
|
||||||
|
/** The doc holds server state (a Step2 arrived) — in EITHER mode. */
|
||||||
private synced = false;
|
private synced = false;
|
||||||
|
/** Sync state was reached as an ACTIVE participant (relay-backed). */
|
||||||
|
private activeSynced = false;
|
||||||
private subEverSent = false;
|
private subEverSent = false;
|
||||||
private readonly touchedCbs: Array<() => void> = [];
|
private readonly touchedCbs: Array<() => void> = [];
|
||||||
|
private readonly resetCbs: Array<() => void> = [];
|
||||||
/** `files` hints (project-sync 0002) — only ever delivered on `~files`. */
|
/** `files` hints (project-sync 0002) — only ever delivered on `~files`. */
|
||||||
private readonly filesCbs: Array<(seq: number, changes: GatewayFileChange[]) => void> = [];
|
private readonly filesCbs: Array<(seq: number, changes: GatewayFileChange[]) => void> = [];
|
||||||
/** The hint-only channel: no doc, no awareness — control frames only. */
|
/** The hint-only channel: no doc, no awareness — control frames only. */
|
||||||
|
|
@ -291,6 +304,8 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
this.docPath = opts.docPath;
|
this.docPath = opts.docPath;
|
||||||
this.mode =
|
this.mode =
|
||||||
(opts.passive && !this.isPresence) || this.isHintOnly ? "passive" : "active";
|
(opts.passive && !this.isPresence) || this.isHintOnly ? "passive" : "active";
|
||||||
|
this.passiveSync =
|
||||||
|
this.mode === "passive" && !this.isHintOnly && opts.passiveSync === true;
|
||||||
this.awareness = new Awareness(doc);
|
this.awareness = new Awareness(doc);
|
||||||
this.conn = acquireConnection(
|
this.conn = acquireConnection(
|
||||||
opts.endpoint,
|
opts.endpoint,
|
||||||
|
|
@ -313,6 +328,13 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
whenSynced(): Promise<void> {
|
whenSynced(): Promise<void> {
|
||||||
if (this.dead) return Promise.reject(this.dead);
|
if (this.dead) return Promise.reject(this.dead);
|
||||||
if (this.mode === "active" && !this.isPresence) return this.activate();
|
if (this.mode === "active" && !this.isPresence) return this.activate();
|
||||||
|
// Passive pull: "synced" = the first at-rest Step2 landed (0004 §2.2).
|
||||||
|
if (this.passiveSync && this.mode === "passive") {
|
||||||
|
if (this.synced) return Promise.resolve();
|
||||||
|
return new Promise((resolve, reject) => {
|
||||||
|
this.syncWaiters.push({ resolve, reject });
|
||||||
|
});
|
||||||
|
}
|
||||||
// Passive/presence: "synced" = the subscription reached an open socket.
|
// Passive/presence: "synced" = the subscription reached an open socket.
|
||||||
if (this.subEverSent) return Promise.resolve();
|
if (this.subEverSent) return Promise.resolve();
|
||||||
return new Promise((resolve, reject) => {
|
return new Promise((resolve, reject) => {
|
||||||
|
|
@ -328,7 +350,10 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
activate(): Promise<void> {
|
activate(): Promise<void> {
|
||||||
if (this.dead) return Promise.reject(this.dead);
|
if (this.dead) return Promise.reject(this.dead);
|
||||||
if (this.isPresence || this.isHintOnly) return this.whenSynced();
|
if (this.isPresence || this.isHintOnly) return this.whenSynced();
|
||||||
if (this.synced) return Promise.resolve();
|
// A passive pull may already have filled the doc — that is NOT active
|
||||||
|
// sync: the gateway must still see `act` (relay demand) and a fresh
|
||||||
|
// Step1 as a participant before writes may flow.
|
||||||
|
if (this.activeSynced) return Promise.resolve();
|
||||||
const wasPassive = this.mode === "passive";
|
const wasPassive = this.mode === "passive";
|
||||||
this.mode = "active";
|
this.mode = "active";
|
||||||
if (this.conn.isOpen()) {
|
if (this.conn.isOpen()) {
|
||||||
|
|
@ -366,6 +391,13 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
this.touchedCbs.push(cb);
|
this.touchedCbs.push(cb);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** `reset` (0004 §2.3): this doc's history was replaced server-side and
|
||||||
|
* our copy cannot be merged — the owner must drop doc + facade and
|
||||||
|
* subscribe afresh. Never fires for active channels. */
|
||||||
|
onReset(cb: () => void): void {
|
||||||
|
this.resetCbs.push(cb);
|
||||||
|
}
|
||||||
|
|
||||||
/** `files` hints — project rows changed on the files route (0002 §1). */
|
/** `files` hints — project rows changed on the files route (0002 §1). */
|
||||||
onFiles(cb: (seq: number, changes: GatewayFileChange[]) => void): void {
|
onFiles(cb: (seq: number, changes: GatewayFileChange[]) => void): void {
|
||||||
this.filesCbs.push(cb);
|
this.filesCbs.push(cb);
|
||||||
|
|
@ -405,6 +437,8 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
this.sendQueryAwareness();
|
this.sendQueryAwareness();
|
||||||
if (this.awareness.getLocalState() !== null) this.publishLocalAwareness();
|
if (this.awareness.getLocalState() !== null) this.publishLocalAwareness();
|
||||||
if (this.mode === "active" && !this.isPresence) this.beginSync();
|
if (this.mode === "active" && !this.isPresence) this.beginSync();
|
||||||
|
// Passive pull: fill (or refresh, after a reconnect) from the at-rest state.
|
||||||
|
else if (this.passiveSync && this.mode === "passive") this.sendSyncStep1();
|
||||||
}
|
}
|
||||||
|
|
||||||
handleSocketDown(): void {
|
handleSocketDown(): void {
|
||||||
|
|
@ -417,6 +451,7 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
removeAwarenessStates(this.awareness, remote, "connection closed");
|
removeAwarenessStates(this.awareness, remote, "connection closed");
|
||||||
}
|
}
|
||||||
this.synced = false;
|
this.synced = false;
|
||||||
|
this.activeSynced = false;
|
||||||
}
|
}
|
||||||
|
|
||||||
handleControl(msg: GatewayServerMsg): void {
|
handleControl(msg: GatewayServerMsg): void {
|
||||||
|
|
@ -454,7 +489,17 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
}
|
}
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
// touched
|
if (msg.t === "reset") {
|
||||||
|
// Only a passive puller can be ahead of a replaced epoch; an active
|
||||||
|
// channel never receives this (its Step1 goes to the BoardRoom).
|
||||||
|
for (const cb of this.resetCbs) cb();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// touched: a passive puller re-pulls (diff against its own SV); every
|
||||||
|
// passive channel still gets the dirty-flag callback.
|
||||||
|
if (this.passiveSync && this.mode === "passive" && this.conn.isOpen()) {
|
||||||
|
this.sendSyncStep1();
|
||||||
|
}
|
||||||
for (const cb of this.touchedCbs) cb();
|
for (const cb of this.touchedCbs) cb();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -484,8 +529,9 @@ export class GatewayDocFacade implements YjsProvider {
|
||||||
if (encoding.length(encoder) > 1) {
|
if (encoding.length(encoder) > 1) {
|
||||||
this.send(encoding.toUint8Array(encoder));
|
this.send(encoding.toUint8Array(encoder));
|
||||||
}
|
}
|
||||||
if (messageType === syncProtocol.messageYjsSyncStep2 && !this.synced) {
|
if (messageType === syncProtocol.messageYjsSyncStep2) {
|
||||||
this.synced = true;
|
this.synced = true;
|
||||||
|
if (this.mode === "active") this.activeSynced = true;
|
||||||
for (const w of this.syncWaiters.splice(0)) w.resolve();
|
for (const w of this.syncWaiters.splice(0)) w.resolve();
|
||||||
}
|
}
|
||||||
return;
|
return;
|
||||||
|
|
|
||||||
|
|
@ -160,6 +160,9 @@ export async function connectKicadDoc(opts: {
|
||||||
* resolves on subscription, and `provider.activate?.()` is the real sync
|
* resolves on subscription, and `provider.activate?.()` is the real sync
|
||||||
* barrier before any bind/read/write of the doc. */
|
* barrier before any bind/read/write of the doc. */
|
||||||
passive?: boolean;
|
passive?: boolean;
|
||||||
|
/** Gateway only (0004 §2.2): pull at-rest state while passive — a
|
||||||
|
* data-only mirror that never wakes the doc's BoardRoom. */
|
||||||
|
passiveSync?: boolean;
|
||||||
}): Promise<KicadDocSession> {
|
}): Promise<KicadDocSession> {
|
||||||
const timeoutMs = opts.timeoutMs ?? CONNECT_TIMEOUT_MS;
|
const timeoutMs = opts.timeoutMs ?? CONNECT_TIMEOUT_MS;
|
||||||
// An already-aborted owner never gets a session — even one that could
|
// An already-aborted owner never gets a session — even one that could
|
||||||
|
|
@ -192,6 +195,7 @@ export async function connectKicadDoc(opts: {
|
||||||
const providerPromise = connectProvider(doc, opts.provider, {
|
const providerPromise = connectProvider(doc, opts.provider, {
|
||||||
room: opts.room,
|
room: opts.room,
|
||||||
passive: opts.passive,
|
passive: opts.passive,
|
||||||
|
passiveSync: opts.passiveSync,
|
||||||
});
|
});
|
||||||
providerPromise.catch(() => {}); // may lose the race and reject later
|
providerPromise.catch(() => {}); // may lose the race and reject later
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -308,7 +308,23 @@ export function bindKicadCollab(
|
||||||
// differ — otherwise the first local edit would re-emit the full model.
|
// differ — otherwise the first local edit would re-emit the full model.
|
||||||
clog(`seed: editor matches doc (${items.size} item(s)) → baseline only, no apply`);
|
clog(`seed: editor matches doc (${items.size} item(s)) → baseline only, no apply`);
|
||||||
try {
|
try {
|
||||||
bridge.snapshotItems();
|
const snapshot = bridge.snapshotItems();
|
||||||
|
// A doc seeded server-side (load-path-rework 0004 §2.4: the runner
|
||||||
|
// installs the resaved upload as the ydoc) carries kicad-cli's
|
||||||
|
// serialization of each body, not this writer's. Same normalization
|
||||||
|
// as the file-seed branch below: re-upsert in the editor's form so
|
||||||
|
// drift-compare and upsertYItem's no-op skip see identical bodies.
|
||||||
|
// Identical bodies cost nothing; a viewer never writes.
|
||||||
|
if (!readOnly) {
|
||||||
|
const wire = parseItemsWireDelta(snapshot);
|
||||||
|
const local = itemsWireToDelta(wire, itemsView(), warnSkip);
|
||||||
|
if (!isEmptyKicadDelta(local)) {
|
||||||
|
clog(
|
||||||
|
`seed: normalizing ${local.updated.length} server-serialized body(ies) to the editor's form`,
|
||||||
|
);
|
||||||
|
applyDeltaToY(doc, local, ORIGIN);
|
||||||
|
}
|
||||||
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
cwarn("seed: snapshotItems baseline failed", err);
|
cwarn("seed: snapshotItems baseline failed", err);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -40,6 +40,12 @@ export interface YjsProvider {
|
||||||
* subscription was passive (the sheet manager's parked-dirty flag).
|
* subscription was passive (the sheet manager's parked-dirty flag).
|
||||||
*/
|
*/
|
||||||
onTouched?(cb: () => void): void;
|
onTouched?(cb: () => void): void;
|
||||||
|
/**
|
||||||
|
* Gateway transport only (load-path-rework 0004 §2.3): the doc's server-side
|
||||||
|
* history was replaced and this passive copy cannot be merged — the owner
|
||||||
|
* must destroy doc + provider and connect afresh.
|
||||||
|
*/
|
||||||
|
onReset?(cb: () => void): void;
|
||||||
}
|
}
|
||||||
|
|
||||||
export type ProviderKind =
|
export type ProviderKind =
|
||||||
|
|
@ -175,7 +181,7 @@ async function hocuspocusProvider(
|
||||||
export async function connectProvider(
|
export async function connectProvider(
|
||||||
doc: Y.Doc,
|
doc: Y.Doc,
|
||||||
config: ProviderConfig,
|
config: ProviderConfig,
|
||||||
opts: { room: string; passive?: boolean },
|
opts: { room: string; passive?: boolean; passiveSync?: boolean },
|
||||||
): Promise<YjsProvider> {
|
): Promise<YjsProvider> {
|
||||||
switch (config.kind) {
|
switch (config.kind) {
|
||||||
case "broadcastchannel":
|
case "broadcastchannel":
|
||||||
|
|
@ -195,6 +201,7 @@ export async function connectProvider(
|
||||||
docPath: parsed.docPath,
|
docPath: parsed.docPath,
|
||||||
token: config.params?.token,
|
token: config.params?.token,
|
||||||
passive: opts.passive,
|
passive: opts.passive,
|
||||||
|
passiveSync: opts.passiveSync,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
return partyKitProvider(doc, requireEndpoint(config), opts.room, config.params);
|
return partyKitProvider(doc, requireEndpoint(config), opts.room, config.params);
|
||||||
|
|
|
||||||
|
|
@ -155,6 +155,22 @@ describe("sheet-manager warm pool", () => {
|
||||||
expect(bindings.at(-1)!.lastSeedOpts).toEqual({ editorMatchesDoc: false });
|
expect(bindings.at(-1)!.lastSeedOpts).toEqual({ editorMatchesDoc: false });
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("invalidate drops a PARKED room (reconnects fresh on the next switch) but never the bound one", async () => {
|
||||||
|
const m = makeManager();
|
||||||
|
await m.connectAll(["a.kicad_sch", "b.kicad_sch"]);
|
||||||
|
await m.switchTo("a.kicad_sch");
|
||||||
|
expect(sessions.length).toBe(2);
|
||||||
|
m.invalidate("a.kicad_sch"); // bound → untouched
|
||||||
|
expect(sessions[0]!.provider.destroy).not.toHaveBeenCalled();
|
||||||
|
m.invalidate("b.kicad_sch"); // parked → dropped
|
||||||
|
expect(sessions[1]!.provider.destroy).toHaveBeenCalled();
|
||||||
|
expect(sessions[1]!.doc.destroy).toHaveBeenCalled();
|
||||||
|
await m.switchTo("b.kicad_sch");
|
||||||
|
expect(sessions.length).toBe(3);
|
||||||
|
expect(sessions[2]!.room).toBe("S:P:b.kicad_sch");
|
||||||
|
m.invalidate("unknown.kicad_sch"); // no-op
|
||||||
|
});
|
||||||
|
|
||||||
it("onboard connects a mid-session sheet exactly once", async () => {
|
it("onboard connects a mid-session sheet exactly once", async () => {
|
||||||
const m = makeManager();
|
const m = makeManager();
|
||||||
await m.onboard("new.kicad_sch");
|
await m.onboard("new.kicad_sch");
|
||||||
|
|
|
||||||
|
|
@ -48,6 +48,14 @@ export interface SheetCollabManager {
|
||||||
switchTo(sheetPath: string): Promise<void>;
|
switchTo(sheetPath: string): Promise<void>;
|
||||||
/** Warm a sheet created mid-session (driven by the save hook on an unknown path). */
|
/** Warm a sheet created mid-session (driven by the save hook on an unknown path). */
|
||||||
onboard(sheetPath: string): Promise<void>;
|
onboard(sheetPath: string): Promise<void>;
|
||||||
|
/**
|
||||||
|
* The sheet's content was replaced at rest (a re-upload / job resave —
|
||||||
|
* load-path-rework 0004 §2.5): a PARKED room's doc may hold the previous
|
||||||
|
* epoch, so drop it; the next switch reconnects and seeds from the
|
||||||
|
* (freshly restaged) file. The bound sheet is left alone — the host
|
||||||
|
* already shows a reload notice for the open target.
|
||||||
|
*/
|
||||||
|
invalidate(sheetPath: string): void;
|
||||||
/**
|
/**
|
||||||
* Coarse non-item layout sync from a just-saved sheet file (miss 08B): title
|
* Coarse non-item layout sync from a just-saved sheet file (miss 08B): title
|
||||||
* block / paper / settings edits reconcile into the sheet's room doc, which
|
* block / paper / settings edits reconcile into the sheet's room doc, which
|
||||||
|
|
@ -359,6 +367,20 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab
|
||||||
return attempt;
|
return attempt;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function invalidate(sheetPath: string): void {
|
||||||
|
const room = rooms.get(sheetPath);
|
||||||
|
if (!room || sheetPath === activePath || room.binding) return;
|
||||||
|
log(`[sheet] ${sheetPath} replaced at rest — dropping the parked room`);
|
||||||
|
try {
|
||||||
|
room.detachWatch?.();
|
||||||
|
room.session.provider.destroy();
|
||||||
|
room.doc.destroy();
|
||||||
|
} catch (err) {
|
||||||
|
cwarn(`[sheet] invalidate ${sheetPath} failed`, err);
|
||||||
|
}
|
||||||
|
rooms.delete(sheetPath);
|
||||||
|
}
|
||||||
|
|
||||||
async function onboard(sheetPath: string): Promise<void> {
|
async function onboard(sheetPath: string): Promise<void> {
|
||||||
if (rooms.has(sheetPath)) return;
|
if (rooms.has(sheetPath)) return;
|
||||||
log(`[sheet] onboarding new sheet ${sheetPath}`);
|
log(`[sheet] onboarding new sheet ${sheetPath}`);
|
||||||
|
|
@ -447,7 +469,7 @@ export function createSheetCollabManager(opts: SheetManagerOptions): SheetCollab
|
||||||
opts.onActiveChange?.(null);
|
opts.onActiveChange?.(null);
|
||||||
}
|
}
|
||||||
|
|
||||||
return { connectAll, switchTo, onboard, syncLayoutFromSave, active, destroy };
|
return { connectAll, switchTo, onboard, invalidate, syncLayoutFromSave, active, destroy };
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface SheetChangedWindow {
|
export interface SheetChangedWindow {
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,9 @@ interface FakeSession {
|
||||||
provider: {
|
provider: {
|
||||||
destroy: ReturnType<typeof vi.fn>;
|
destroy: ReturnType<typeof vi.fn>;
|
||||||
awareness: { setLocalState: ReturnType<typeof vi.fn> };
|
awareness: { setLocalState: ReturnType<typeof vi.fn> };
|
||||||
|
onReset: ReturnType<typeof vi.fn>;
|
||||||
|
/** Fire the gateway's `reset` (0004 §2.3). */
|
||||||
|
emitReset: () => void;
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -37,6 +40,7 @@ let sessions: FakeSession[];
|
||||||
|
|
||||||
function makeSession(room: string): FakeSession {
|
function makeSession(room: string): FakeSession {
|
||||||
const handlers = new Set<() => void>();
|
const handlers = new Set<() => void>();
|
||||||
|
const resets = new Set<() => void>();
|
||||||
return {
|
return {
|
||||||
room,
|
room,
|
||||||
doc: {
|
doc: {
|
||||||
|
|
@ -49,6 +53,8 @@ function makeSession(room: string): FakeSession {
|
||||||
provider: {
|
provider: {
|
||||||
destroy: vi.fn(),
|
destroy: vi.fn(),
|
||||||
awareness: { setLocalState: vi.fn() },
|
awareness: { setLocalState: vi.fn() },
|
||||||
|
onReset: vi.fn((cb: () => void) => resets.add(cb)),
|
||||||
|
emitReset: () => resets.forEach((h) => h()),
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
@ -111,6 +117,13 @@ describe("startSiblingRestage", () => {
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("watches as a PASSIVE PULL — never a participant, never a relay (0004 §2.2)", async () => {
|
||||||
|
await start(["main.kicad_sch"]);
|
||||||
|
expect(connectKicadDoc).toHaveBeenCalledWith(
|
||||||
|
expect.objectContaining({ passive: true, passiveSync: true }),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
it("restages once on connect when the room already holds state", async () => {
|
it("restages once on connect when the room already holds state", async () => {
|
||||||
await start(["main.kicad_sch"]);
|
await start(["main.kicad_sch"]);
|
||||||
expect(restageFile).toHaveBeenCalledTimes(1);
|
expect(restageFile).toHaveBeenCalledTimes(1);
|
||||||
|
|
@ -227,6 +240,23 @@ describe("startSiblingRestage", () => {
|
||||||
expect(connectKicadDoc).toHaveBeenCalledTimes(3); // no further dials
|
expect(connectKicadDoc).toHaveBeenCalledTimes(3); // no further dials
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("a reset drops the watch without flushing and re-dials while still wanted (0004 §2.3)", async () => {
|
||||||
|
const presence = fakePresence(["main.kicad_sch"]);
|
||||||
|
await start(["main.kicad_sch"], presence);
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
expect(sessions.length).toBe(1);
|
||||||
|
restageFile.mockClear();
|
||||||
|
sessions[0]!.doc.emitRemote(); // a pending restage from the OLD epoch…
|
||||||
|
sessions[0]!.provider.emitReset();
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
// …is discarded, the old session torn down, a fresh one dialed.
|
||||||
|
expect(sessions[0]!.provider.destroy).toHaveBeenCalled();
|
||||||
|
expect(sessions[0]!.doc.destroy).toHaveBeenCalled();
|
||||||
|
expect(sessions.length).toBe(2);
|
||||||
|
// Only the fresh session's connect-time restage ran.
|
||||||
|
expect(restageFile).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
|
||||||
it("a rejoin during the linger keeps the existing session", async () => {
|
it("a rejoin during the linger keeps the existing session", async () => {
|
||||||
const presence = fakePresence(["main.kicad_sch"]);
|
const presence = fakePresence(["main.kicad_sch"]);
|
||||||
await start(["main.kicad_sch"], presence);
|
await start(["main.kicad_sch"], presence);
|
||||||
|
|
|
||||||
|
|
@ -24,6 +24,13 @@ import type { ProviderConfig } from "./provider";
|
||||||
* Without a presence room (provider "none", connect failure) it falls back to
|
* Without a presence room (provider "none", connect failure) it falls back to
|
||||||
* the eager mode: every in-scope sheet is watched for the whole session.
|
* the eager mode: every in-scope sheet is watched for the whole session.
|
||||||
*
|
*
|
||||||
|
* Transport (load-path-rework 0004 §2.2): the watch is a PASSIVE PULL — the
|
||||||
|
* gateway answers its SyncStep1 from the sheet's at-rest state and every
|
||||||
|
* `touched` re-pulls a diff. The sheet's BoardRoom is never woken by this
|
||||||
|
* mirror (an active subscription used to hold a relay for as long as the
|
||||||
|
* peer had the sheet open). A `reset` (the sheet's history was replaced
|
||||||
|
* server-side) drops the watch and re-dials with a fresh doc.
|
||||||
|
*
|
||||||
* MEMFS-only: nothing is uploaded and no editor poke is needed — pcbnew reads
|
* MEMFS-only: nothing is uploaded and no editor poke is needed — pcbnew reads
|
||||||
* the schematic from MEMFS when the sync runs. Accepted v1 gaps: a sheet
|
* the schematic from MEMFS when the sync runs. Accepted v1 gaps: a sheet
|
||||||
* created by a peer mid-session (path not in the boot file list), and a peer
|
* created by a peer mid-session (path not in the boot file list), and a peer
|
||||||
|
|
@ -124,10 +131,15 @@ export async function startSiblingRestage(opts: {
|
||||||
);
|
);
|
||||||
};
|
};
|
||||||
|
|
||||||
const openSession = async (sheetPath: string): Promise<KicadDocSession | null> => {
|
const openSession = async (
|
||||||
|
sheetPath: string,
|
||||||
|
onReset?: () => void,
|
||||||
|
): Promise<KicadDocSession | null> => {
|
||||||
const session = await connectKicadDoc({
|
const session = await connectKicadDoc({
|
||||||
provider: opts.provider,
|
provider: opts.provider,
|
||||||
room: collabRoomId(opts.scopeId, opts.projectId, sheetPath),
|
room: collabRoomId(opts.scopeId, opts.projectId, sheetPath),
|
||||||
|
passive: true,
|
||||||
|
passiveSync: true,
|
||||||
});
|
});
|
||||||
if (destroyed) {
|
if (destroyed) {
|
||||||
session.provider.destroy();
|
session.provider.destroy();
|
||||||
|
|
@ -138,6 +150,7 @@ export async function startSiblingRestage(opts: {
|
||||||
// (mirrors the sheet-manager's read-only invisible-observer handling).
|
// (mirrors the sheet-manager's read-only invisible-observer handling).
|
||||||
session.provider.awareness?.setLocalState(null);
|
session.provider.awareness?.setLocalState(null);
|
||||||
session.doc.on("update", () => schedule(sheetPath, session.doc));
|
session.doc.on("update", () => schedule(sheetPath, session.doc));
|
||||||
|
if (onReset) session.provider.onReset?.(onReset);
|
||||||
log(`[sibling] watching ${sheetPath}`);
|
log(`[sibling] watching ${sheetPath}`);
|
||||||
restageFromDoc(sheetPath, session.doc);
|
restageFromDoc(sheetPath, session.doc);
|
||||||
return session;
|
return session;
|
||||||
|
|
@ -199,7 +212,19 @@ export async function startSiblingRestage(opts: {
|
||||||
const dial = (sheetPath: string, watch: Watch): void => {
|
const dial = (sheetPath: string, watch: Watch): void => {
|
||||||
watch.connecting = (async () => {
|
watch.connecting = (async () => {
|
||||||
try {
|
try {
|
||||||
const session = await openSession(sheetPath);
|
const session = await openSession(sheetPath, () => {
|
||||||
|
// The sheet's history was replaced under us (0004 §2.3): this doc
|
||||||
|
// can't be merged. Drop it (no flush — its content is the OLD
|
||||||
|
// epoch) and re-dial while a peer still has the sheet open.
|
||||||
|
if (destroyed || watches.get(sheetPath) !== watch) return;
|
||||||
|
log(`[sibling] ${sheetPath} was replaced server-side — re-syncing`);
|
||||||
|
closeNow(sheetPath, watch, false);
|
||||||
|
if (wanted(sheetPath)) {
|
||||||
|
const fresh: Watch = {};
|
||||||
|
watches.set(sheetPath, fresh);
|
||||||
|
dial(sheetPath, fresh);
|
||||||
|
}
|
||||||
|
});
|
||||||
if (session) {
|
if (session) {
|
||||||
watch.session = session;
|
watch.session = session;
|
||||||
watch.retryDelayMs = undefined; // success resets the backoff
|
watch.retryDelayMs = undefined; // success resets the backoff
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue