perf(collab): scope board-load connections to what's actually open
Presence-scoped sibling restage: every tab announces the document it is actively editing in the project presence room (PresenceState.sheetPath, re-published on eeschema sheet navigation), and a pcbnew session connects a sibling schematic's board-room only while a peer announces it open — instead of eagerly holding one socket per sheet for the whole session. Works because /files/ materializes from the ydoc, so the boot MEMFS snapshot is room-fresh; a sheet can only drift while someone is editing it, and that someone is in the roster. On leave the watch lingers 30s, flushes the pending restage, and closes; without a presence room the eager mode remains as fallback. A solo board session now holds 3 sockets (doc, presence, lib-mirror mux) — down from 92 on an 8-project repo before this series. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W4EzyhjhDzLdNZsFAYjz7X
This commit is contained in:
parent
7ccedbf973
commit
b5ed68ef87
6 changed files with 257 additions and 45 deletions
|
|
@ -1 +1 @@
|
||||||
Subproject commit 0f839cda767b428fa5677c62591ad289c7260e19
|
Subproject commit 9e269711deebbb7ed5fbb3679082c2b90224a188
|
||||||
|
|
@ -1539,6 +1539,10 @@ export function WasmTool({
|
||||||
provider: yjsProviderConfig(),
|
provider: yjsProviderConfig(),
|
||||||
user: presenceUser(),
|
user: presenceUser(),
|
||||||
tool,
|
tool,
|
||||||
|
// Announce the open document (the active sheet for
|
||||||
|
// eeschema, re-published on navigation below) — peers'
|
||||||
|
// sibling-restage scopes its sockets to announced files.
|
||||||
|
docPath: targetPath,
|
||||||
});
|
});
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
append(`[collab] cross-app presence connect failed: ${String(err)}`);
|
append(`[collab] cross-app presence connect failed: ${String(err)}`);
|
||||||
|
|
@ -1706,6 +1710,10 @@ export function WasmTool({
|
||||||
onActiveChange: (activeRoom) => {
|
onActiveChange: (activeRoom) => {
|
||||||
driftRef.current?.stop();
|
driftRef.current?.stop();
|
||||||
driftRef.current = null;
|
driftRef.current = null;
|
||||||
|
// Re-announce the actively-edited sheet in the project room —
|
||||||
|
// peers' sibling-restage tracks it (a pcbnew tab only mirrors
|
||||||
|
// sheets someone actually has open).
|
||||||
|
crossAppRef.current?.setDocPath(activeRoom?.sheetPath);
|
||||||
startPresence(activeRoom?.provider, activeRoom?.sheetPath, activeRoom?.doc);
|
startPresence(activeRoom?.provider, activeRoom?.sheetPath, activeRoom?.doc);
|
||||||
startComments(activeRoom?.doc);
|
startComments(activeRoom?.doc);
|
||||||
if (activeRoom && !readOnly) {
|
if (activeRoom && !readOnly) {
|
||||||
|
|
@ -1752,6 +1760,10 @@ export function WasmTool({
|
||||||
projectId,
|
projectId,
|
||||||
files,
|
files,
|
||||||
targetPath,
|
targetPath,
|
||||||
|
// Presence-scoped: connect a sheet's room only while a peer
|
||||||
|
// announces it open (zero sibling sockets when alone). Absent
|
||||||
|
// (provider "none" / connect failed) ⇒ eager fallback.
|
||||||
|
presence: crossAppRef.current ?? undefined,
|
||||||
provider: yjsProviderConfig(),
|
provider: yjsProviderConfig(),
|
||||||
log: append,
|
log: append,
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -32,6 +32,13 @@ export interface CrossAppPeer {
|
||||||
export interface CrossAppHandle {
|
export interface CrossAppHandle {
|
||||||
/** Publish this tab's selection (uuids + pcbnew footprint paths). */
|
/** Publish this tab's selection (uuids + pcbnew footprint paths). */
|
||||||
setSelection(uuids: string[], paths?: string[]): void;
|
setSelection(uuids: string[], paths?: string[]): void;
|
||||||
|
/**
|
||||||
|
* Publish which document this tab is actively editing (project-relative
|
||||||
|
* path; the active sheet for eeschema, re-announced on sheet navigation).
|
||||||
|
* Sibling-restage peers use it to connect a sheet's room only while someone
|
||||||
|
* actually has it open.
|
||||||
|
*/
|
||||||
|
setDocPath(path: string | undefined): void;
|
||||||
/**
|
/**
|
||||||
* Peers in a DIFFERENT tool, one entry per awareness client. Unlike the
|
* Peers in a DIFFERENT tool, one entry per awareness client. Unlike the
|
||||||
* room roster this INCLUDES the own user's other tabs — one person with
|
* room roster this INCLUDES the own user's other tabs — one person with
|
||||||
|
|
@ -49,6 +56,8 @@ export async function startCrossAppPresence(opts: {
|
||||||
provider: ProviderConfig;
|
provider: ProviderConfig;
|
||||||
user: PresenceUser;
|
user: PresenceUser;
|
||||||
tool: string;
|
tool: string;
|
||||||
|
/** Initial doc this tab edits (see CrossAppHandle.setDocPath). */
|
||||||
|
docPath?: string;
|
||||||
}): Promise<CrossAppHandle | undefined> {
|
}): Promise<CrossAppHandle | undefined> {
|
||||||
if (opts.provider.kind === "none") return undefined;
|
if (opts.provider.kind === "none") return undefined;
|
||||||
|
|
||||||
|
|
@ -72,6 +81,7 @@ export async function startCrossAppPresence(opts: {
|
||||||
|
|
||||||
let selection: string[] = [];
|
let selection: string[] = [];
|
||||||
let selectionPaths: string[] | undefined;
|
let selectionPaths: string[] | undefined;
|
||||||
|
let docPath = opts.docPath;
|
||||||
|
|
||||||
const publish = () => {
|
const publish = () => {
|
||||||
const state: PresenceState = {
|
const state: PresenceState = {
|
||||||
|
|
@ -82,6 +92,9 @@ export async function startCrossAppPresence(opts: {
|
||||||
cursor: null,
|
cursor: null,
|
||||||
selection,
|
selection,
|
||||||
...(selectionPaths?.length ? { selectionPaths } : {}),
|
...(selectionPaths?.length ? { selectionPaths } : {}),
|
||||||
|
// The actively-edited document, so peers can scope work (e.g. sibling
|
||||||
|
// restage sockets) to sheets that are ACTUALLY open somewhere.
|
||||||
|
...(docPath ? { sheetPath: docPath } : {}),
|
||||||
updatedAt: Date.now(),
|
updatedAt: Date.now(),
|
||||||
};
|
};
|
||||||
awareness.setLocalState(state);
|
awareness.setLocalState(state);
|
||||||
|
|
@ -109,6 +122,11 @@ export async function startCrossAppPresence(opts: {
|
||||||
selectionPaths = paths;
|
selectionPaths = paths;
|
||||||
publish();
|
publish();
|
||||||
},
|
},
|
||||||
|
setDocPath(path) {
|
||||||
|
if (path === docPath) return;
|
||||||
|
docPath = path;
|
||||||
|
publish();
|
||||||
|
},
|
||||||
peers() {
|
peers() {
|
||||||
const out: CrossAppPeer[] = [];
|
const out: CrossAppPeer[] = [];
|
||||||
for (const [clientId, raw] of awareness.getStates()) {
|
for (const [clientId, raw] of awareness.getStates()) {
|
||||||
|
|
|
||||||
|
|
@ -161,6 +161,7 @@ function stubCrossApp(initial: CrossAppPeer[] = []) {
|
||||||
const subscribers = new Set<() => void>();
|
const subscribers = new Set<() => void>();
|
||||||
const handle: CrossAppHandle & { firePeers(p: CrossAppPeer[]): void } = {
|
const handle: CrossAppHandle & { firePeers(p: CrossAppPeer[]): void } = {
|
||||||
setSelection: vi.fn(),
|
setSelection: vi.fn(),
|
||||||
|
setDocPath: vi.fn(),
|
||||||
peers: () => peers,
|
peers: () => peers,
|
||||||
subscribe(cb) {
|
subscribe(cb) {
|
||||||
subscribers.add(cb);
|
subscribers.add(cb);
|
||||||
|
|
|
||||||
|
|
@ -17,7 +17,7 @@ vi.mock("@pcbjam/shared", () => ({
|
||||||
docToFile: () => "(kicad_sch materialized)",
|
docToFile: () => "(kicad_sch materialized)",
|
||||||
}));
|
}));
|
||||||
|
|
||||||
import { startSiblingRestage } from "./sibling-restage";
|
import { startSiblingRestage, type SiblingPresence } from "./sibling-restage";
|
||||||
|
|
||||||
interface FakeSession {
|
interface FakeSession {
|
||||||
room: string;
|
room: string;
|
||||||
|
|
@ -52,18 +52,36 @@ function makeSession(room: string): FakeSession {
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
function start(files: string[]) {
|
function start(files: string[], presence?: SiblingPresence) {
|
||||||
return startSiblingRestage({
|
return startSiblingRestage({
|
||||||
win: {} as never,
|
win: {} as never,
|
||||||
slug: "proj",
|
slug: "proj",
|
||||||
scopeId: "S",
|
scopeId: "S",
|
||||||
projectId: "P",
|
projectId: "P",
|
||||||
files: files.map((path) => ({ path })),
|
files: files.map((path) => ({ path })),
|
||||||
|
presence,
|
||||||
provider: { kind: "none" } as never,
|
provider: { kind: "none" } as never,
|
||||||
log: () => {},
|
log: () => {},
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Roster fake: `announce` swaps the open-path set and fires subscribers. */
|
||||||
|
function fakePresence(initial: string[] = []) {
|
||||||
|
let paths = initial;
|
||||||
|
const subs = new Set<() => void>();
|
||||||
|
return {
|
||||||
|
peers: () => paths.map((sheetPath) => ({ state: { sheetPath } })),
|
||||||
|
subscribe(cb: () => void) {
|
||||||
|
subs.add(cb);
|
||||||
|
return () => subs.delete(cb);
|
||||||
|
},
|
||||||
|
announce(next: string[]) {
|
||||||
|
paths = next;
|
||||||
|
subs.forEach((cb) => cb());
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
vi.useFakeTimers();
|
vi.useFakeTimers();
|
||||||
sessions = [];
|
sessions = [];
|
||||||
|
|
@ -127,4 +145,53 @@ describe("startSiblingRestage", () => {
|
||||||
expect(s.doc.destroy).toHaveBeenCalled();
|
expect(s.doc.destroy).toHaveBeenCalled();
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("presence-scoped mode", () => {
|
||||||
|
it("holds zero sockets while no peer announces a sheet", async () => {
|
||||||
|
await start(["main.kicad_sch", "sub.kicad_sch"], fakePresence());
|
||||||
|
expect(connectKicadDoc).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("connects only announced in-scope sheets, once", async () => {
|
||||||
|
const presence = fakePresence();
|
||||||
|
await start(["main.kicad_sch", "sub.kicad_sch"], presence);
|
||||||
|
presence.announce(["main.kicad_sch", "other-dir.kicad_pcb"]);
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
expect(sessions.map((s) => s.room)).toEqual(["S:P:main.kicad_sch"]);
|
||||||
|
// A second announce of the same sheet (another peer joining) is a no-op.
|
||||||
|
presence.announce(["main.kicad_sch"]);
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
expect(connectKicadDoc).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("lingers after the peer leaves, flushes the pending restage, closes", async () => {
|
||||||
|
const presence = fakePresence(["main.kicad_sch"]);
|
||||||
|
await start(["main.kicad_sch"], presence);
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
expect(sessions).toHaveLength(1);
|
||||||
|
restageFile.mockClear();
|
||||||
|
sessions[0]!.doc.emitRemote(); // debounced restage now pending
|
||||||
|
presence.announce([]); // peer closes their tab
|
||||||
|
await vi.advanceTimersByTimeAsync(60_000); // past linger + debounce
|
||||||
|
expect(sessions[0]!.provider.destroy).toHaveBeenCalled();
|
||||||
|
expect(sessions[0]!.doc.destroy).toHaveBeenCalled();
|
||||||
|
expect(restageFile).toHaveBeenCalled(); // last edits reached MEMFS
|
||||||
|
// Peer comes back: a fresh session dials again.
|
||||||
|
presence.announce(["main.kicad_sch"]);
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
expect(connectKicadDoc).toHaveBeenCalledTimes(2);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("a rejoin during the linger keeps the existing session", async () => {
|
||||||
|
const presence = fakePresence(["main.kicad_sch"]);
|
||||||
|
await start(["main.kicad_sch"], presence);
|
||||||
|
await vi.runAllTimersAsync();
|
||||||
|
presence.announce([]); // leave…
|
||||||
|
await vi.advanceTimersByTimeAsync(5_000); // …but rejoin within linger
|
||||||
|
presence.announce(["main.kicad_sch"]);
|
||||||
|
await vi.advanceTimersByTimeAsync(120_000);
|
||||||
|
expect(connectKicadDoc).toHaveBeenCalledTimes(1);
|
||||||
|
expect(sessions[0]!.provider.destroy).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -8,27 +8,54 @@ import type { ProviderConfig } from "./provider";
|
||||||
* Live sibling-document mirror for pcbnew sessions (project-sync 0001 bug 3).
|
* Live sibling-document mirror for pcbnew sessions (project-sync 0001 bug 3).
|
||||||
*
|
*
|
||||||
* Boot stages every project file into MEMFS exactly once (kicad-runner
|
* Boot stages every project file into MEMFS exactly once (kicad-runner
|
||||||
* syncProjectToMemfs), so the `.kicad_sch` a PCB session would sync from is a
|
* syncProjectToMemfs) — and since the files route materializes from the ydoc,
|
||||||
* page-load snapshot while collaborators keep editing it in its own room. This
|
* that snapshot is room-fresh at fetch time. A sheet can therefore only drift
|
||||||
* subscribes — as an invisible, data-only observer — to every sibling sheet's
|
* from its MEMFS copy while SOMEONE IS EDITING IT — and whoever does sits in
|
||||||
* collab room and re-materializes the file into MEMFS on updates, debounced
|
* the project presence room announcing which document their tab has open
|
||||||
* like the lib refresh (synced-source scheduleEditorReload). It also restages
|
* (PresenceState.sheetPath). So instead of eagerly holding one board-room
|
||||||
* once right after connect: the room is the source of truth since ysync, so it
|
* WebSocket per sibling schematic for the whole session, this watches the
|
||||||
* can already be ahead of the API snapshot boot fetched (unsaved collab edits
|
* presence roster and connects a sheet's room only while a peer (including the
|
||||||
* live only in the room).
|
* same user's other tab) actually has it open — a solo session holds ZERO
|
||||||
|
* sibling sockets. On connect it restages once (the peer may have edited
|
||||||
|
* between our boot fetch and now), then re-materializes into MEMFS on updates,
|
||||||
|
* debounced like the lib refresh. When the peer leaves, the socket lingers
|
||||||
|
* briefly (reload flaps), flushes a final restage, and closes.
|
||||||
|
*
|
||||||
|
* 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.
|
||||||
*
|
*
|
||||||
* 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. A sheet created by a peer
|
* the schematic from MEMFS when the sync runs. Accepted v1 gaps: a sheet
|
||||||
* mid-session is not picked up (its path isn't in the boot file list); v1
|
* created by a peer mid-session (path not in the boot file list), and a peer
|
||||||
* accepts that gap.
|
* whose edit lands between our boot fetch and their presence entry reaching us
|
||||||
|
* (a seconds-wide window during boot; the next load heals it).
|
||||||
*/
|
*/
|
||||||
|
|
||||||
const RESTAGE_DEBOUNCE_MS = 400;
|
const RESTAGE_DEBOUNCE_MS = 400;
|
||||||
|
/** How long a watch outlives its last announcing peer (reload/nav flaps). */
|
||||||
|
const CLOSE_LINGER_MS = 30_000;
|
||||||
|
|
||||||
export interface SiblingRestageHandle {
|
export interface SiblingRestageHandle {
|
||||||
destroy(): void;
|
destroy(): void;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The slice of `CrossAppHandle` this consumes (structural, so tests fake it):
|
||||||
|
* the project-presence roster + its change feed.
|
||||||
|
*/
|
||||||
|
export interface SiblingPresence {
|
||||||
|
peers(): Array<{ state: { sheetPath?: string } }>;
|
||||||
|
subscribe(cb: () => void): () => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface Watch {
|
||||||
|
session?: KicadDocSession;
|
||||||
|
connecting?: Promise<void>;
|
||||||
|
linger?: ReturnType<typeof setTimeout>;
|
||||||
|
/** Connect failed — don't re-dial on every roster change. */
|
||||||
|
failed?: boolean;
|
||||||
|
}
|
||||||
|
|
||||||
export async function startSiblingRestage(opts: {
|
export async function startSiblingRestage(opts: {
|
||||||
win: ToolWindow;
|
win: ToolWindow;
|
||||||
slug: string;
|
slug: string;
|
||||||
|
|
@ -37,6 +64,8 @@ export async function startSiblingRestage(opts: {
|
||||||
files: { path: string }[];
|
files: { path: string }[];
|
||||||
/** The opened `.kicad_pcb` — scopes the watch to ITS KiCad project. */
|
/** The opened `.kicad_pcb` — scopes the watch to ITS KiCad project. */
|
||||||
targetPath?: string;
|
targetPath?: string;
|
||||||
|
/** Project presence roster; omitted ⇒ eager mode (watch every sheet). */
|
||||||
|
presence?: SiblingPresence;
|
||||||
provider: ProviderConfig;
|
provider: ProviderConfig;
|
||||||
log: (m: string) => void;
|
log: (m: string) => void;
|
||||||
}): Promise<SiblingRestageHandle> {
|
}): Promise<SiblingRestageHandle> {
|
||||||
|
|
@ -44,18 +73,18 @@ export async function startSiblingRestage(opts: {
|
||||||
// Only the opened board's own KiCad project can be synced from: pcbnew's
|
// Only the opened board's own KiCad project can be synced from: pcbnew's
|
||||||
// "update from schematic" reads the sheets next to the .kicad_pcb (same
|
// "update from schematic" reads the sheets next to the .kicad_pcb (same
|
||||||
// directory tree). A backend project holding SEVERAL KiCad projects (a
|
// directory tree). A backend project holding SEVERAL KiCad projects (a
|
||||||
// repo of boards) must not fan out one room per schematic repo-wide —
|
// repo of boards) must not fan out repo-wide — that held ~27 idle
|
||||||
// that held ~27 idle board-room sockets for an 8-board repo. Sheets a
|
// board-room sockets for an 8-board repo. Sheets a project references
|
||||||
// project references OUTSIDE its directory (rare ../ sheet paths) fall
|
// OUTSIDE its directory (rare ../ sheet paths) fall back to the boot
|
||||||
// back to the boot snapshot — same gap v1 already accepts for new sheets.
|
// snapshot — same gap v1 already accepts for new sheets.
|
||||||
const dir = opts.targetPath
|
const dir = opts.targetPath
|
||||||
? opts.targetPath.slice(0, opts.targetPath.lastIndexOf("/") + 1)
|
? opts.targetPath.slice(0, opts.targetPath.lastIndexOf("/") + 1)
|
||||||
: "";
|
: "";
|
||||||
const sheetPaths = opts.files
|
const sheetPaths = opts.files
|
||||||
.map((f) => f.path)
|
.map((f) => f.path)
|
||||||
.filter((p) => p.endsWith(".kicad_sch") && p.startsWith(dir));
|
.filter((p) => p.endsWith(".kicad_sch") && p.startsWith(dir));
|
||||||
|
const sheetSet = new Set(sheetPaths);
|
||||||
|
|
||||||
const sessions: KicadDocSession[] = [];
|
|
||||||
const timers = new Map<string, ReturnType<typeof setTimeout>>();
|
const timers = new Map<string, ReturnType<typeof setTimeout>>();
|
||||||
let destroyed = false;
|
let destroyed = false;
|
||||||
|
|
||||||
|
|
@ -83,41 +112,126 @@ export async function startSiblingRestage(opts: {
|
||||||
);
|
);
|
||||||
};
|
};
|
||||||
|
|
||||||
await Promise.all(
|
const openSession = async (sheetPath: string): Promise<KicadDocSession | null> => {
|
||||||
sheetPaths.map(async (sheetPath) => {
|
const session = await connectKicadDoc({
|
||||||
try {
|
provider: opts.provider,
|
||||||
const session = await connectKicadDoc({
|
room: collabRoomId(opts.scopeId, opts.projectId, sheetPath),
|
||||||
provider: opts.provider,
|
});
|
||||||
room: collabRoomId(opts.scopeId, opts.projectId, sheetPath),
|
if (destroyed) {
|
||||||
});
|
session.provider.destroy();
|
||||||
if (destroyed) {
|
session.doc.destroy();
|
||||||
session.provider.destroy();
|
return null;
|
||||||
session.doc.destroy();
|
}
|
||||||
return;
|
// Data-only observer: never appear in the sheet's presence roster
|
||||||
|
// (mirrors the sheet-manager's read-only invisible-observer handling).
|
||||||
|
session.provider.awareness?.setLocalState(null);
|
||||||
|
session.doc.on("update", () => schedule(sheetPath, session.doc));
|
||||||
|
log(`[sibling] watching ${sheetPath}`);
|
||||||
|
restageFromDoc(sheetPath, session.doc);
|
||||||
|
return session;
|
||||||
|
};
|
||||||
|
|
||||||
|
/* ------------------------- eager fallback (no presence room) ------------ */
|
||||||
|
|
||||||
|
if (!opts.presence) {
|
||||||
|
const sessions: KicadDocSession[] = [];
|
||||||
|
await Promise.all(
|
||||||
|
sheetPaths.map(async (sheetPath) => {
|
||||||
|
try {
|
||||||
|
const session = await openSession(sheetPath);
|
||||||
|
if (session) sessions.push(session);
|
||||||
|
} catch (err) {
|
||||||
|
log(`[sibling] room connect failed for ${sheetPath}: ${String(err)}`);
|
||||||
}
|
}
|
||||||
// Data-only observer: never appear in the sheet's presence roster
|
}),
|
||||||
// (mirrors the sheet-manager's read-only invisible-observer handling).
|
);
|
||||||
session.provider.awareness?.setLocalState(null);
|
return {
|
||||||
sessions.push(session);
|
destroy() {
|
||||||
session.doc.on("update", () => schedule(sheetPath, session.doc));
|
destroyed = true;
|
||||||
log(`[sibling] watching ${sheetPath}`);
|
for (const t of timers.values()) clearTimeout(t);
|
||||||
restageFromDoc(sheetPath, session.doc);
|
timers.clear();
|
||||||
} catch (err) {
|
for (const s of sessions) {
|
||||||
log(`[sibling] room connect failed for ${sheetPath}: ${String(err)}`);
|
s.provider.destroy();
|
||||||
|
s.doc.destroy();
|
||||||
|
}
|
||||||
|
sessions.length = 0;
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/* ------------------------- presence-driven mode ------------------------- */
|
||||||
|
|
||||||
|
const presence = opts.presence;
|
||||||
|
const watches = new Map<string, Watch>();
|
||||||
|
|
||||||
|
const closeNow = (sheetPath: string, w: Watch, flushPending: boolean) => {
|
||||||
|
const t = timers.get(sheetPath);
|
||||||
|
if (t) {
|
||||||
|
clearTimeout(t);
|
||||||
|
timers.delete(sheetPath);
|
||||||
|
// A debounced restage was pending — run it before dropping the doc, or
|
||||||
|
// the peer's last burst of edits never reaches MEMFS.
|
||||||
|
if (flushPending && w.session) restageFromDoc(sheetPath, w.session.doc);
|
||||||
|
}
|
||||||
|
if (w.linger) clearTimeout(w.linger);
|
||||||
|
w.session?.provider.destroy();
|
||||||
|
w.session?.doc.destroy();
|
||||||
|
watches.delete(sheetPath);
|
||||||
|
};
|
||||||
|
|
||||||
|
const reconcile = () => {
|
||||||
|
if (destroyed) return;
|
||||||
|
const open = new Set<string>();
|
||||||
|
for (const p of presence.peers()) {
|
||||||
|
const sp = p.state.sheetPath;
|
||||||
|
if (sp && sheetSet.has(sp)) open.add(sp);
|
||||||
|
}
|
||||||
|
for (const sheetPath of open) {
|
||||||
|
let w = watches.get(sheetPath);
|
||||||
|
if (w?.linger) {
|
||||||
|
clearTimeout(w.linger);
|
||||||
|
w.linger = undefined;
|
||||||
}
|
}
|
||||||
}),
|
if (!w) {
|
||||||
|
const watch: Watch = {};
|
||||||
|
watches.set(sheetPath, watch);
|
||||||
|
watch.connecting = (async () => {
|
||||||
|
try {
|
||||||
|
const session = await openSession(sheetPath);
|
||||||
|
if (session) watch.session = session;
|
||||||
|
} catch (err) {
|
||||||
|
watch.failed = true;
|
||||||
|
log(`[sibling] room connect failed for ${sheetPath}: ${String(err)}`);
|
||||||
|
} finally {
|
||||||
|
watch.connecting = undefined;
|
||||||
|
}
|
||||||
|
})();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (const [sheetPath, w] of watches) {
|
||||||
|
if (open.has(sheetPath) || w.linger) continue;
|
||||||
|
w.linger = setTimeout(() => {
|
||||||
|
if (destroyed) return;
|
||||||
|
log(`[sibling] releasing ${sheetPath} (no peer has it open)`);
|
||||||
|
closeNow(sheetPath, w, true);
|
||||||
|
}, CLOSE_LINGER_MS);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
const unsubscribe = presence.subscribe(reconcile);
|
||||||
|
reconcile();
|
||||||
|
log(
|
||||||
|
`[sibling] presence-scoped watch over ${sheetPaths.length} sheet(s) — ` +
|
||||||
|
`connecting only while a peer has one open`,
|
||||||
);
|
);
|
||||||
|
|
||||||
return {
|
return {
|
||||||
destroy() {
|
destroy() {
|
||||||
destroyed = true;
|
destroyed = true;
|
||||||
|
unsubscribe();
|
||||||
for (const t of timers.values()) clearTimeout(t);
|
for (const t of timers.values()) clearTimeout(t);
|
||||||
timers.clear();
|
timers.clear();
|
||||||
for (const s of sessions) {
|
for (const [sheetPath, w] of [...watches]) closeNow(sheetPath, w, false);
|
||||||
s.provider.destroy();
|
|
||||||
s.doc.destroy();
|
|
||||||
}
|
|
||||||
sessions.length = 0;
|
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue