standalone: file-change hints — ~files watch, sibling restage, target notice (project-sync 0002)

GatewayDocFacade.onFiles + hint-only channel; files-watch router (Tier 0
echo/observed bookkeeping, Tier 1 debounced sibling restage, Tier 2 open-
target notice); WasmTool/ToolPage wiring with observed-revision seams.
Bumps pcbjam-shared (4fd6af2).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012Wd1r3ewftpV1DBSEArpRa
This commit is contained in:
Gergő Törcsvári 2026-08-25 13:55:01 +02:00
commit 9a19b96b6c
No known key found for this signature in database
GPG key ID: 8E75F2CDE64E5322
8 changed files with 450 additions and 3 deletions

@ -1 +1 @@
Subproject commit 905063094caf183420f09d09e973132132481d2b
Subproject commit 4fd6af24a2a8d818970931bbf13d15c9818e6169

View file

@ -72,9 +72,11 @@ import { memfsFilePath, memfsProjectDir, TOOL_BUNDLE, TOOL_FRAME } from "@/wasm/
import {
driveProjectIntoTool,
readStagedFile,
restageFile,
usedLibNicknames,
type ToolFile,
} from "@/wasm/kicad-runner";
import { startFilesWatch, type FilesWatchHandle } from "@/wasm/collab/files-watch";
import { resolveSheetHierarchy } from "@/wasm/collab/sheet-hierarchy";
import { dump as dumpTrace, mark } from "@/wasm/load-trace";
import { errorMessage, isTerminalError } from "@/wasm/terminal-error";
@ -974,6 +976,8 @@ export function WasmTool({
targetPath,
fetchBytes,
onStagedRevision,
observedRevision,
rememberObservedRevision,
saveBytes,
createFile,
docSource,
@ -1015,6 +1019,15 @@ export function WasmTool({
* the CAS ancestry `saveBytes` must publish against (see DriveOptions).
*/
onStagedRevision?: (relPath: string, revision: number) => void;
/**
* Files-route change hints (project-sync 0002): the latest server revision
* this client observed for a path (its own PUT ack the echo check) and
* the recorder for revisions learned from a peer's hint. Absent hints
* still restage siblings but every hint stamped with our user is treated
* as a peer's (no echo suppression).
*/
observedRevision?: (relPath: string) => number | undefined;
rememberObservedRevision?: (relPath: string, revision: number) => void;
/**
* Persist one file the user saved in the editor (FileSave writes MEMFS, then
* the wasm fires window.kicadCollab.onSave this). API upload for backend
@ -1072,6 +1085,7 @@ export function WasmTool({
// eeschema sheet rebinds — the bridge re-reads it on every startPresence.
const crossAppRef = React.useRef<CrossAppHandle | null>(null);
const siblingRestageRef = React.useRef<SiblingRestageHandle | null>(null);
const filesWatchRef = React.useRef<FilesWatchHandle | null>(null);
// Set at the boot effect's cleanup; deferred starters bail on it (the
// sibling-restage idle stagger can fire after unmount).
const disposedRef = React.useRef(false);
@ -1252,6 +1266,8 @@ export function WasmTool({
crossAppRef.current = null;
siblingRestageRef.current?.destroy();
siblingRestageRef.current = null;
filesWatchRef.current?.destroy();
filesWatchRef.current = null;
driftRef.current?.stop();
driftRef.current = null;
// Tears down every warm room's provider/doc (the only place providers are
@ -2287,6 +2303,40 @@ export function WasmTool({
});
}
}
// Files-route change hints (project-sync 0002 §3): peers' PUT-channel
// writes (.kicad_pro after assign-footprints, uploads, job resaves)
// restage into MEMFS; a hint for the open non-room target becomes a
// reload/conflict notice. Rides the same gateway socket as presence.
if ((tool === "pcbnew" || tool === "eeschema") && !readOnly && !collabOptOut) {
void startFilesWatch({
scopeId,
projectId,
provider: yjsProviderConfig(),
targetPath,
selfUser: presenceUser().id,
knownPaths: files.map((f) => f.path),
isRoomBacked: (p) => roomBacked.has(p),
observedRevision: (p) => observedRevision?.(p),
rememberObserved: (p, r) => rememberObservedRevision?.(p, r),
fetchBytes,
restage: (p, bytes) => restageFile(win, slug, p, bytes, append),
onNewPath: (p) => {
if (p.endsWith(".kicad_sch")) void sheetManagerRef.current?.onboard(p);
},
onTargetChanged: (c) => {
const who = c.by ?? "a collaborator";
setStatus(
`${c.path} was updated by ${who} (rev ${c.revision}) — reload to see it; your next save will report a conflict`,
);
},
onListingStale: () => append("[files] hint gap — project listing is stale until reload"),
log: append,
}).then((handle) => {
if (!handle) return;
if (disposedRef.current) handle.destroy();
else filesWatchRef.current = handle;
});
}
// Viewer selection feed (viewer-panels): read-only sessions never
// bind presence (no room, no awareness), so the SelectionInspector's
// store is fed by a minimal onSelection handler + the C++ canvas

View file

@ -165,6 +165,20 @@ export function rememberFileBaseRevision(
projectSource().rememberBaseRevision?.(slug, relPath, revision);
}
/** Latest server revision seen for a path (project-sync 0002 echo check). */
export function observedFileRevision(slug: string, relPath: string): number | undefined {
return projectSource().observedRevision?.(slug, relPath);
}
/** Record a revision learned from a `files` hint; observed only, never base. */
export function rememberFileObservedRevision(
slug: string,
relPath: string,
revision: number,
): void {
projectSource().rememberObservedRevision?.(slug, relPath, revision);
}
/** Observe the server revision after an ambiguous save; never rebases this model. */
export async function refreshFileRevision(
slug: string,

View file

@ -91,6 +91,14 @@ export interface ProjectSource {
* was vouched against, so it IS the model's ancestry, not merely observed.
*/
rememberBaseRevision?(slug: string, relPath: string, revision: number): void;
/**
* Latest server revision this source has SEEN for a path (listing, GET or
* PUT response, or a file-change hint) diagnostic/observed only, never a
* write precondition. Undefined when the path was never observed.
*/
observedRevision?(slug: string, relPath: string): number | undefined;
/** Record an observed revision (e.g. from a `files` hint). Never rebases. */
rememberObservedRevision?(slug: string, relPath: string, revision: number): void;
}
// --- remote (REST backend over the shared contract) ---------------------------
@ -270,6 +278,12 @@ function remoteProjectSource(): ProjectSource {
rememberObservedRevision(slug, relPath, revision);
baseRevisions.set(revisionKey(slug, relPath), revision);
},
observedRevision(slug, relPath) {
return observedRevisions.get(revisionKey(slug, relPath));
},
rememberObservedRevision(slug, relPath, revision) {
rememberObservedRevision(slug, relPath, revision);
},
// Editor saves publish through the CAS PUT (findings D-3 client half): the
// expected revision is the model's ANCESTRY (baseRevisions), never the
// latest observed metadata — so a save issued after a conflict cannot
@ -509,6 +523,13 @@ function compositeProjectSource(
rememberBaseRevision: (slug, p, revision) => {
void route(slug).then((s) => s.rememberBaseRevision?.(slug, p, revision));
},
// Composite reads route asynchronously; the observed getter is sync, so
// it answers from whichever source already routed this slug (primary
// first — local projects never carry server revisions).
observedRevision: (slug, p) => primary.observedRevision?.(slug, p),
rememberObservedRevision: (slug, p, revision) => {
void route(slug).then((s) => s.rememberObservedRevision?.(slug, p, revision));
},
};
}

View file

@ -5,7 +5,9 @@ import { Loader2 } from "lucide-react";
import {
createProjectFileIfMissing,
fetchFileBytes,
observedFileRevision,
rememberFileBaseRevision,
rememberFileObservedRevision,
uploadFileBytes,
useProjectBoot,
useSourceDescriptor,
@ -104,6 +106,10 @@ export function ToolPage() {
onStagedRevision={(relPath, revision) =>
rememberFileBaseRevision(slug, relPath, revision)
}
observedRevision={(relPath) => observedFileRevision(slug, relPath)}
rememberObservedRevision={(relPath, revision) =>
rememberFileObservedRevision(slug, relPath, revision)
}
saveBytes={
readOnly
? undefined

View file

@ -0,0 +1,153 @@
import { describe, expect, it, vi } from "vitest";
import type { GatewayFileChange } from "@pcbjam/shared";
import { createFilesHintRouter, startFilesWatch, type FilesWatchOptions } from "./files-watch";
const ch = (over: Partial<GatewayFileChange>): GatewayFileChange => ({
path: "x.kicad_pro",
revision: 2,
origin: "editor",
...over,
});
function makeRouter(over: Partial<FilesWatchOptions> = {}) {
const observed = new Map<string, number>();
const restaged: string[] = [];
const fetched: string[] = [];
const events: string[] = [];
const opts: FilesWatchOptions = {
scopeId: "s",
projectId: "p",
provider: { kind: "none" },
targetPath: "board.kicad_pcb",
selfUser: "me",
isRoomBacked: (p) => p.endsWith(".kicad_sch"),
observedRevision: (p) => observed.get(p),
rememberObserved: (p, r) => void observed.set(p, r),
fetchBytes: async (p) => {
fetched.push(p);
return new Uint8Array([1]);
},
restage: (p) => void restaged.push(p),
onNewPath: (p) => events.push(`new:${p}`),
onTargetChanged: (c) => events.push(`target:${c.path}@${c.revision}`),
onListingStale: () => events.push("stale"),
log: () => {},
debounceMs: 0,
...over,
};
const router = createFilesHintRouter(opts);
router.seedKnown(["board.kicad_pcb", "x.kicad_pro", "root.kicad_sch"]);
return { router, observed, restaged, fetched, events };
}
const tick = () => new Promise((r) => setTimeout(r, 5));
describe("files hint router (project-sync 0002 §3)", () => {
it("Tier 1: a plain sibling is refetched + restaged once (debounced)", async () => {
const { router, restaged, fetched, observed } = makeRouter();
router.handle(1, [ch({ revision: 2, by: "peer" })]);
router.handle(2, [ch({ revision: 3, by: "peer" })]);
await tick();
expect(fetched).toEqual(["x.kicad_pro"]);
expect(restaged).toEqual(["x.kicad_pro"]);
expect(observed.get("x.kicad_pro")).toBe(3);
});
it("own echo (by me, revision already observed from the PUT ack) is ignored", async () => {
const { router, restaged, observed } = makeRouter();
observed.set("x.kicad_pro", 2);
router.handle(1, [ch({ revision: 2, by: "me" })]);
await tick();
expect(restaged).toEqual([]);
// Same user, a DIFFERENT revision (another tab) is not an echo.
router.handle(2, [ch({ revision: 3, by: "me" })]);
await tick();
expect(restaged).toEqual(["x.kicad_pro"]);
});
it("room-backed paths are never restaged from the row", async () => {
const { router, restaged, events } = makeRouter();
router.handle(1, [ch({ path: "root.kicad_sch", revision: 9, by: "peer" })]);
await tick();
expect(restaged).toEqual([]);
expect(events).toEqual([]);
});
it("Tier 2: the open target on the PUT channel only notifies", async () => {
const { router, restaged, events } = makeRouter({ isRoomBacked: () => false });
router.handle(1, [ch({ path: "board.kicad_pcb", revision: 4, by: "peer" })]);
await tick();
expect(restaged).toEqual([]);
expect(events).toEqual(["target:board.kicad_pcb@4"]);
});
it("a seq gap flags the listing stale; a new path is announced", async () => {
const { router, events, restaged } = makeRouter();
router.handle(5, [ch({ path: "sub.kicad_pro", revision: 1, by: "peer" })]);
router.handle(8, []); // gap + oversized-batch shape
await tick();
expect(events).toEqual(["new:sub.kicad_pro", "stale"]);
expect(restaged).toEqual(["sub.kicad_pro"]);
});
it("deleted rows are logged, not restaged", async () => {
const log = vi.fn();
const { router, restaged } = makeRouter({ log });
router.handle(1, [ch({ revision: 0, deleted: true, origin: "job" })]);
await tick();
expect(restaged).toEqual([]);
expect(log.mock.calls.some((c) => String(c[0]).includes("deleted"))).toBe(true);
});
it("destroy cancels pending restages", async () => {
const { router, restaged } = makeRouter({ debounceMs: 20 });
router.handle(1, [ch({ by: "peer" })]);
router.destroy();
await new Promise((r) => setTimeout(r, 40));
expect(restaged).toEqual([]);
});
});
describe("startFilesWatch", () => {
it("wires the hint source to the router and tears both down", async () => {
let cb: ((seq: number, c: GatewayFileChange[]) => void) | undefined;
const destroy = vi.fn();
const restaged: string[] = [];
const handle = await startFilesWatch({
scopeId: "s",
projectId: "p",
provider: { kind: "none" },
knownPaths: [],
isRoomBacked: () => false,
observedRevision: () => undefined,
rememberObserved: () => {},
fetchBytes: async () => new Uint8Array(),
restage: (p) => void restaged.push(p),
log: () => {},
debounceMs: 0,
connect: async () => ({ onFiles: (f) => void (cb = f), destroy }),
});
expect(handle).toBeDefined();
cb!(1, [ch({ path: "n.kicad_pro", by: "peer" })]);
await tick();
expect(restaged).toEqual(["n.kicad_pro"]);
handle!.destroy();
expect(destroy).toHaveBeenCalledTimes(1);
});
it("is a no-op without the gateway transport", async () => {
const handle = await startFilesWatch({
scopeId: "s",
projectId: "p",
provider: { kind: "broadcastchannel" },
knownPaths: [],
isRoomBacked: () => false,
observedRevision: () => undefined,
rememberObserved: () => {},
fetchBytes: async () => new Uint8Array(),
restage: () => {},
log: () => {},
});
expect(handle).toBeUndefined();
});
});

View file

@ -0,0 +1,184 @@
import { collabRoomId, FILES_DOC_PATH, type GatewayFileChange } from "@pcbjam/shared";
import * as Y from "yjs";
import { connectProvider, type ProviderConfig, type YjsProvider } from "./provider";
/**
* Files-route change watch (project-sync 0002 §3): subscribes the project's
* `~files` gateway channel and turns each `files` hint into INVALIDATION
* never a write into an open document:
*
* - Tier 0 (always): observed-revision bookkeeping + own-echo suppression,
* `onListingStale` on a seq gap (reconnect / oversized batch).
* - Tier 1: a changed path that is neither the open target nor room-backed
* (`.kicad_pro`, netlists, a peer's new sheet) is re-fetched and restaged
* into MEMFS after a short debounce the next "Update PCB from
* Schematic" / project-settings read sees the peer's version.
* - Tier 2: the open target itself changed on the PUT channel (a file with
* no room): `onTargetChanged` the host shows a reload/conflict notice;
* the CAS lane keeps guarding the next save.
*
* Room-backed paths (listing hasYdoc/isLive) are ignored: the room is the
* truth there and already carries its own `touched`/frames.
*/
export const FILES_RESTAGE_DEBOUNCE_MS = 400;
export interface FilesWatchHandle {
destroy(): void;
}
/** The slice of the gateway facade this consumes (structural; tests fake it). */
export interface FilesHintSource {
onFiles(cb: (seq: number, changes: GatewayFileChange[]) => void): void;
destroy(): void;
}
export interface FilesWatchOptions {
scopeId: string;
projectId: string;
provider: ProviderConfig;
/** The open document (Tier 2 routing). */
targetPath?: string;
/** This session's user slug — hints stamped `by` us are echo candidates. */
selfUser?: string;
/** Listing-derived: room-owned paths are never restaged from the row. */
isRoomBacked: (relPath: string) => boolean;
/** Latest revision this client observed for a path (its own PUT ack). */
observedRevision: (relPath: string) => number | undefined;
rememberObserved: (relPath: string, revision: number) => void;
/** Fresh bytes for a sibling (goes through the project source: the cache
* validator no longer matches, so this is a real GET + base-revision record). */
fetchBytes: (relPath: string) => Promise<Uint8Array>;
/** MEMFS write of a sibling (kicad-runner restageFile). */
restage: (relPath: string, bytes: Uint8Array) => void;
/** A path not in the boot listing appeared (a peer's "Add Sheet"). */
onNewPath?: (relPath: string) => void;
onTargetChanged?: (change: GatewayFileChange) => void;
onListingStale?: () => void;
log: (m: string) => void;
/** Test seam: replace the gateway connect. */
connect?: () => Promise<FilesHintSource>;
/** Test seam. */
debounceMs?: number;
}
/** Pure hint router — exported for tests; `startFilesWatch` wires it to the gateway. */
export function createFilesHintRouter(opts: FilesWatchOptions) {
const knownPaths = new Set<string>();
let lastSeq: number | null = null;
let destroyed = false;
const timers = new Map<string, ReturnType<typeof setTimeout>>();
const debounceMs = opts.debounceMs ?? FILES_RESTAGE_DEBOUNCE_MS;
const restageLater = (relPath: string): void => {
const prev = timers.get(relPath);
if (prev) clearTimeout(prev);
timers.set(
relPath,
setTimeout(() => {
timers.delete(relPath);
if (destroyed) return;
void opts
.fetchBytes(relPath)
.then((bytes) => {
if (destroyed) return;
opts.restage(relPath, bytes);
opts.log(`[files] restaged ${relPath} from a peer's write`);
})
.catch((err) => opts.log(`[files] restage failed for ${relPath}: ${String(err)}`));
}, debounceMs),
);
};
const handle = (seq: number, changes: GatewayFileChange[]): void => {
if (destroyed) return;
if (lastSeq !== null && seq !== lastSeq + 1) {
opts.log(`[files] hint seq gap (${lastSeq}${seq}) — listing is stale`);
opts.onListingStale?.();
}
lastSeq = seq;
for (const change of changes) {
// Own echo: our PUT ack already recorded exactly this revision.
if (
opts.selfUser &&
change.by === opts.selfUser &&
opts.observedRevision(change.path) === change.revision
) {
continue;
}
opts.rememberObserved(change.path, change.revision);
if (opts.isRoomBacked(change.path)) continue; // the room owns it
if (change.path === opts.targetPath) {
opts.onTargetChanged?.(change);
continue;
}
if (change.deleted) {
// v1: MEMFS keeps the last copy; the next boot drops it.
opts.log(`[files] ${change.path} deleted by ${change.by ?? "a job"} — kept in MEMFS until reload`);
continue;
}
if (!knownPaths.has(change.path)) {
knownPaths.add(change.path);
opts.onNewPath?.(change.path);
}
restageLater(change.path);
}
};
return {
handle,
seedKnown(paths: Iterable<string>) {
for (const p of paths) knownPaths.add(p);
},
destroy() {
destroyed = true;
for (const t of timers.values()) clearTimeout(t);
timers.clear();
},
};
}
export async function startFilesWatch(
opts: FilesWatchOptions & { knownPaths: Iterable<string> },
): Promise<FilesWatchHandle | undefined> {
if (opts.provider.kind !== "partykit" && !opts.connect) return undefined; // gateway only
const router = createFilesHintRouter(opts);
router.seedKnown(opts.knownPaths);
let source: FilesHintSource;
try {
source = opts.connect ? await opts.connect() : await connectGateway(opts);
} catch (err) {
opts.log(`[files] watch connect failed: ${String(err)}`);
router.destroy();
return undefined;
}
source.onFiles(router.handle);
opts.log("[files] watching project file changes");
return {
destroy() {
router.destroy();
source.destroy();
},
};
}
async function connectGateway(opts: FilesWatchOptions): Promise<FilesHintSource> {
const doc = new Y.Doc();
const provider: YjsProvider = await connectProvider(doc, opts.provider, {
room: collabRoomId(opts.scopeId, opts.projectId, FILES_DOC_PATH),
passive: true,
});
const facade = provider as YjsProvider & Partial<FilesHintSource>;
if (typeof facade.onFiles !== "function") {
provider.destroy();
doc.destroy();
throw new Error("provider has no files channel");
}
return {
onFiles: (cb) => facade.onFiles!(cb),
destroy: () => {
provider.destroy();
doc.destroy();
},
};
}

View file

@ -25,9 +25,11 @@ import * as syncProtocol from "y-protocols/sync";
import type * as Y from "yjs";
import {
type GatewayClientMsg,
type GatewayFileChange,
type GatewayServerMsg,
type GatewaySubMode,
parseGatewayServerMsg,
FILES_DOC_PATH,
PRESENCE_DOC_PATH,
projectRoomName,
tagGatewayFrame,
@ -267,6 +269,10 @@ export class GatewayDocFacade implements YjsProvider {
private synced = false;
private subEverSent = false;
private readonly touchedCbs: Array<() => void> = [];
/** `files` hints (project-sync 0002) — only ever delivered on `~files`. */
private readonly filesCbs: Array<(seq: number, changes: GatewayFileChange[]) => void> = [];
/** The hint-only channel: no doc, no awareness — control frames only. */
private readonly isHintOnly: boolean;
private readonly syncWaiters: Array<{
resolve: () => void;
reject: (e: unknown) => void;
@ -281,8 +287,10 @@ export class GatewayDocFacade implements YjsProvider {
opts: GatewayFacadeOpts,
) {
this.isPresence = opts.docPath === PRESENCE_DOC_PATH;
this.isHintOnly = opts.docPath === FILES_DOC_PATH;
this.docPath = opts.docPath;
this.mode = opts.passive && !this.isPresence ? "passive" : "active";
this.mode =
(opts.passive && !this.isPresence) || this.isHintOnly ? "passive" : "active";
this.awareness = new Awareness(doc);
this.conn = acquireConnection(
opts.endpoint,
@ -319,7 +327,7 @@ export class GatewayDocFacade implements YjsProvider {
*/
activate(): Promise<void> {
if (this.dead) return Promise.reject(this.dead);
if (this.isPresence) return this.whenSynced();
if (this.isPresence || this.isHintOnly) return this.whenSynced();
if (this.synced) return Promise.resolve();
const wasPassive = this.mode === "passive";
this.mode = "active";
@ -358,6 +366,11 @@ export class GatewayDocFacade implements YjsProvider {
this.touchedCbs.push(cb);
}
/** `files` hints — project rows changed on the files route (0002 §1). */
onFiles(cb: (seq: number, changes: GatewayFileChange[]) => void): void {
this.filesCbs.push(cb);
}
destroy(): void {
if (this.destroyed) return;
this.destroyed = true;
@ -387,6 +400,7 @@ export class GatewayDocFacade implements YjsProvider {
if (this.destroyed || this.dead) return;
this.subEverSent = true;
for (const w of this.subWaiters.splice(0)) w.resolve();
if (this.isHintOnly) return; // nothing to announce, nothing to sync
// (Re)announce presence: a query for peers' states + our own, if any.
this.sendQueryAwareness();
if (this.awareness.getLocalState() !== null) this.publishLocalAwareness();
@ -423,6 +437,10 @@ export class GatewayDocFacade implements YjsProvider {
if (this.mode === "active" && !this.isPresence) this.sendSyncStep1();
return;
}
if (msg.t === "files") {
for (const cb of this.filesCbs) cb(msg.seq, msg.changes);
return;
}
// touched
for (const cb of this.touchedCbs) cb();
}
@ -497,6 +515,7 @@ export class GatewayDocFacade implements YjsProvider {
}
private publishLocalAwareness(): void {
if (this.isHintOnly) return;
const encoder = encoding.createEncoder();
encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
encoding.writeVarUint8Array(