diff --git a/.changeset/watch-says-heads.md b/.changeset/watch-says-heads.md new file mode 100644 index 000000000..d0254f07b --- /dev/null +++ b/.changeset/watch-says-heads.md @@ -0,0 +1,12 @@ +--- +"@inkandswitch/patchwork-bootloader": patch +"@inkandswitch/patchwork-filesystem": patch +--- + +A restart of the automerge worker no longer makes every tab look up every package at once. Those lookups reached the worker while it was shutting down, and answering them pushed the replacement over its memory threshold, so it recycled again. That kept the worker restarting about every 30s, with package lookups failing with HTTP 556 until the page was reloaded. + +The sync node's `changed` now carries its heads for the sedimentree. They're attached only once the tree holds every record that caused the change. Otherwise `changed` comes without them. The node also answers each `watch` with the heads it has, so a replacement node tells watchers what it holds, including anything the node before it stored. `PortSyncRemote.reconnect()` re-sends its watches and no longer calls every watcher. + +These are Subduction's heads, not the document's: once a document has a long history they include ancestors of its automerge heads. So they only say whether anything arrived, never which version to load. `watchDocument(url, onChange)` calls `onChange(seen)` with a summary of what the node holds, or with no argument when the node doesn't say. `PackageTracker.watch`'s `onChange` takes the same optional `seen`. `ModuleWatcher` looks a package up when `seen` changes or is absent, and does nothing when a restarted node says the same as before. + +A reload also waits for its import to finish before the next one starts, so the changes that land meanwhile collapse into one more import at the latest heads. diff --git a/core/bootloader/src/automerge-protocol-handler.ts b/core/bootloader/src/automerge-protocol-handler.ts index bab400d08..a2157b868 100644 --- a/core/bootloader/src/automerge-protocol-handler.ts +++ b/core/bootloader/src/automerge-protocol-handler.ts @@ -185,6 +185,16 @@ function getRepo(): Promise { return repoPromise; } +let builtRepo: Repo | undefined; + +// For the console. The Repo is only built once something has to be resolved, +// so before then reading `repo` starts building it and gives its promise: +// `await repo` works either way. +Object.defineProperty(self, "repo", { + configurable: true, + get: () => builtRepo ?? getRepo(), +}); + async function buildRepo(): Promise { await control.holding; await initWasm(); @@ -193,7 +203,7 @@ async function buildRepo(): Promise { ? await buildKeyhiveRepo(syncServer.keyhive) : { repo: await buildPlainRepo() }; - (self as any).repo = repo; + builtRepo = repo; if (hive) (self as any).hive = hive; return repo; } diff --git a/core/bootloader/src/setup.ts b/core/bootloader/src/setup.ts index 102f3cfb8..c354a5ba5 100644 --- a/core/bootloader/src/setup.ts +++ b/core/bootloader/src/setup.ts @@ -166,14 +166,22 @@ export function getSyncRemote(): PortSyncRemote { * returned function is called. Nothing is loaded for it: the sync node keeps * the document subscribed with the server and says when something arrives for * it, whether from the server or from another tab. That's the way to follow a - * document whose contents you only need now and then; find it again when - * `onChange` fires. + * document whose contents you only need now and then. + * + * `seen` sums up what the sync node holds of the document: the same `seen` + * twice means nothing arrived in between. It comes straight after watching, + * and after the automerge worker restarts, whether or not anything moved. It + * isn't a version of the document: the node's heads are Subduction's, which + * can include ancestors of the Automerge heads, so find the document to know + * where it is. It's absent when the node doesn't say. */ export function watchDocument( url: AutomergeUrl, - onChange: () => void + onChange: (seen?: string) => void ): () => void { - return getSyncRemote().watch(getSedimentreeId(url), onChange); + return getSyncRemote().watch(getSedimentreeId(url), (heads) => + onChange(heads && [...heads].sort().join("|")) + ); } /** Dev escape hatch: make the shared worker close so the next use respawns it. */ diff --git a/core/bootloader/src/sync-node.ts b/core/bootloader/src/sync-node.ts index 10067f34b..4998c5122 100644 --- a/core/bootloader/src/sync-node.ts +++ b/core/bootloader/src/sync-node.ts @@ -118,6 +118,15 @@ export class SyncNode { readonly #echoing = new Map(); /** Peer deliveries waiting for this microtask's flush, per sedimentree. */ readonly #outbound = new Map(); + /** + * Sedimentrees to tell watchers about shortly, each with whether to stay + * quiet when this node doesn't know its heads, and the records that moved + * it, which its heads have to include before they're worth sending. + */ + readonly #changed = new Map< + string, + { ifKnown: boolean; records: Array<{ kind: RecordKind; id: string }> } + >(); readonly #ephemeralSubscribers = new Map(); /** Per sedimentree, the round with the server in progress. */ readonly #rounds = new Map(); @@ -278,11 +287,94 @@ export class SyncNode { } } - /** Tell every Repo watching `sid` that it may have new heads. */ - #sendChanged(sid: string) { - const message: NodeMessage = { type: "changed", sid }; - for (const client of this.#clients) { - if (client.watching.has(sid)) this.#send(client, message); + /** + * Tell every Repo watching `sid` that it may have new heads, and what they + * are, so it can tell for itself whether they've moved. With `ifKnown` + * (answering a watch), say nothing for a sedimentree this node has no + * heads for: there's nothing to report until a round brings some. + * `records` are what arrived, if anything did. + */ + #sendChanged( + sid: string, + { + ifKnown = false, + records = [], + }: { + ifKnown?: boolean; + records?: Array<{ kind: RecordKind; id: string }>; + } = {} + ) { + if (![...this.#clients].some((client) => client.watching.has(sid))) return; + // Not a microtask: a record is stored before subduction adds it to the + // tree, and the heads are read from the tree. + if (this.#changed.size === 0) { + setTimeout(() => + this.#announceChanged().catch((error) => + this.#log("couldn't announce changes", error) + ) + ); + } + const queued = this.#changed.get(sid); + this.#changed.set(sid, { + ifKnown: ifKnown && (queued?.ifKnown ?? true), + records: [...(queued?.records ?? []), ...records], + }); + } + + /** + * Whether the tree has every one of `records` in it yet, so that its heads + * account for them. Heads read before then could be the ones from before + * they arrived, which would tell a watcher nothing moved. + */ + async #treeHas( + sid: string, + records: Array<{ kind: RecordKind; id: string }> + ): Promise { + if (records.length === 0) return true; + const id = sedimentreeId(sid); + const [commits, fragments] = await Promise.all([ + this.subduction.getCommits(id), + this.subduction.getFragments(id), + ]); + const known = new Set([ + ...(commits ?? []).map((commit) => commit.commitId.toHexString()), + ...(fragments ?? []).map((fragment) => fragment.head.toHexString()), + ]); + return records.every((record) => known.has(record.id)); + } + + // One getAllHeads for everything queued: a restarted worker's tabs watch + // every package again at once. + async #announceChanged() { + const changed = new Map(this.#changed); + this.#changed.clear(); + const heads = new Map(); + // Without heads a change still goes out, and the Repo looks for itself. + const trees = await this.subduction.getAllHeads().catch((error) => { + this.#log("couldn't read heads", error); + return []; + }); + for (const tree of trees) { + const sid = tree.id.toString(); + // An empty list is a sedimentree with no heads yet. + if (changed.has(sid) && tree.heads.length > 0) { + heads.set( + sid, + tree.heads.map((head) => head.toHexString()) + ); + } + } + for (const [sid, { ifKnown, records }] of changed) { + let known = heads.get(sid); + if (!known && ifKnown) continue; + // Without heads the watcher looks for itself, which is right either way. + if (known && !(await this.#treeHas(sid, records).catch(() => false))) { + known = undefined; + } + const message: NodeMessage = { type: "changed", sid, heads: known }; + for (const client of this.#clients) { + if (client.watching.has(sid)) this.#send(client, message); + } } } @@ -380,6 +472,7 @@ export class SyncNode { return; case "watch": client.watching.add(request.sid); + this.#sendChanged(request.sid, { ifKnown: true }); this.#subscribe(request.sid); return; case "unwatch": @@ -562,7 +655,7 @@ export class SyncNode { // heads announcement lands. this.#sendToInterested(sid, { type: "data", sid, records: forward }, from); // The writer too: what watches a document in a Repo isn't what holds it. - this.#sendChanged(sid); + this.#sendChanged(sid, { records: forward }); const keys = forward.map((record) => `${sid}:${record.id}`); for (const key of keys) this.#echoes.add(key); @@ -629,7 +722,7 @@ export class SyncNode { if (!records) return; this.#outbound.delete(sid); this.#sendToInterested(sid, { type: "data", sid, records }); - this.#sendChanged(sid); + this.#sendChanged(sid, { records }); } async #subscribeEphemeral(sid: string) { diff --git a/core/bootloader/src/sync-protocol.ts b/core/bootloader/src/sync-protocol.ts index aa920d9f4..d44a27f64 100644 --- a/core/bootloader/src/sync-protocol.ts +++ b/core/bootloader/src/sync-protocol.ts @@ -25,6 +25,7 @@ export type RemoteRequest = /** * Hear when a sedimentree may have new heads, without holding it: the node * says `changed` and sends no data, so the Repo needn't load the document. + * It says so straight away too, with the heads it has, if it has any. */ | { type: "watch"; sid: string } | { type: "unwatch"; sid: string } @@ -50,7 +51,13 @@ export type NodeMessage = records: Array<{ kind: RecordKind; id: string; blob: Uint8Array }>; } | { type: "heads"; sid: string; peerId: string; heads: string[] } - | { type: "changed"; sid: string } + /** + * `heads` are the node's own heads for the sedimentree, as hex. They're + * Subduction's heads, not the document's: they can include ancestors of + * its automerge heads. The same heads twice means nothing arrived. Absent + * from a node that doesn't know them (or predates them). + */ + | { type: "changed"; sid: string; heads?: string[] } | { type: "ephemeral"; sid: string; senderId: string; payload: Uint8Array } | { type: "keyhive-frame"; peer: Uint8Array; payload: Uint8Array } | { type: "keyhive-peer-disconnected"; peer: Uint8Array }; diff --git a/core/bootloader/src/sync-remote.ts b/core/bootloader/src/sync-remote.ts index cfe423e65..da5334455 100644 --- a/core/bootloader/src/sync-remote.ts +++ b/core/bootloader/src/sync-remote.ts @@ -53,7 +53,7 @@ export class PortSyncRemote implements SubductionRemote { readonly #pending = new Map(); readonly #ephemeral = new Set(); /** Per sedimentree, who wants to hear it may have changed. */ - readonly #watchers = new Map void>>(); + readonly #watchers = new Map void>>(); #keyhiveHandler: KeyhiveFrameHandler | null = null; /** * Writes the node hasn't heard about because there was no port to tell it @@ -83,12 +83,11 @@ export class PortSyncRemote implements SubductionRemote { for (const sid of this.#ephemeral) { this.#post({ type: "ephemeral-subscribe", sid }); } - // A replacement node only says `changed` for what reaches it after this, - // and a watched document may have moved while there was no node to say - // so. Everyone looks again. - for (const [sid, watchers] of this.#watchers) { + // A replacement node answers each watch with the heads it has, which + // include whatever the node before it stored, so a watcher whose + // document moved while there was no node to say so hears it then. + for (const sid of this.#watchers.keys()) { this.#post({ type: "watch", sid }); - for (const onChange of watchers) onChange(); } if (this.#keyhiveHandler) this.#post({ type: "keyhive-listen" }); const unsent = this.#unsent; @@ -158,7 +157,7 @@ export class PortSyncRemote implements SubductionRemote { return; case "changed": for (const onChange of this.#watchers.get(message.sid) ?? []) { - onChange(); + onChange(message.heads); } return; case "ephemeral": @@ -248,17 +247,21 @@ export class PortSyncRemote implements SubductionRemote { /** * Call `onChange` whenever the sedimentree `sid` may have new heads, whether * or not this Repo holds it, until the returned function is called. The - * node keeps it subscribed with the server meanwhile. + * node keeps it subscribed with the server meanwhile. It's called with the + * node's heads for it (hex sedimentree heads), which may not have moved, + * straight after watching and after a worker restart too; without them + * when the node doesn't say. */ - watch(sid: string, onChange: () => void): () => void { + watch(sid: string, onChange: (heads?: string[]) => void): () => void { let watchers = this.#watchers.get(sid); if (!watchers) { watchers = new Set(); this.#watchers.set(sid, watchers); - this.#post({ type: "watch", sid }); } + // Again for a second watcher, so it hears the heads too. + this.#post({ type: "watch", sid }); // Wrapped, so that watching twice with one function takes two unwatches. - const watcher = () => onChange(); + const watcher = (heads?: string[]) => onChange(heads); watchers.add(watcher); return () => { if (!watchers.delete(watcher) || watchers.size > 0) return; diff --git a/core/bootloader/test/sync-node.test.ts b/core/bootloader/test/sync-node.test.ts index b0a49c2ac..e763243d3 100644 --- a/core/bootloader/test/sync-node.test.ts +++ b/core/bootloader/test/sync-node.test.ts @@ -16,7 +16,7 @@ import { SedimentreeId, Subduction, } from "@automerge/automerge-subduction"; -import { parseAutomergeUrl } from "@automerge/automerge-repo"; +import { decodeHeads, parseAutomergeUrl } from "@automerge/automerge-repo"; import { SyncNode } from "../src/sync-node.js"; import { PortSyncRemote, getSedimentreeId } from "../src/sync-remote.js"; @@ -477,6 +477,52 @@ describe("watching a document", () => { await eventually(() => counter.rounds > before); }); + it("says the document's heads, and the replacement answers a watch with them", async () => { + const server = startServer(); + const other = elsewhere(server); + const theirs = other.create({ count: 0 }); + await other.flush(); + + const site = origin(server); + const remote = site.remote(); + cleanup.push(async () => remote.shutdown()); + const heard: Array = []; + remote.watch(getSedimentreeId(theirs.url), (heads) => heard.push(heads)); + await new Promise((resolve) => setTimeout(resolve, 500)); + theirs.change((doc) => { + doc.count = 1; + }); + const moved = decodeHeads(theirs.heads()); + await eventually(() => + heard.some((heads) => heads?.join() === moved.join()) + ); + + const before = heard.length; + await site.restartNode(); + await eventually(() => heard.length > before); + // Nothing moved, and every call since says so. + await new Promise((resolve) => setTimeout(resolve, 500)); + expect(heard.slice(before).map((heads) => heads?.join())).toEqual( + heard.slice(before).map(() => moved.join()) + ); + + // A document the node already has: heads read before the edit is in its + // tree would be the old ones, and say nothing moved. + const since = heard.length; + theirs.change((doc) => { + doc.count = 2; + }); + const movedAgain = decodeHeads(theirs.heads()); + await eventually(() => heard.length > since); + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(heard.slice(since).map((heads) => heads?.join())).not.toContain( + moved.join() + ); + await eventually(() => + heard.some((heads) => heads?.join() === movedAgain.join()) + ); + }); + it("keeps hearing the server across a worker restart", async () => { const server = startServer(); const other = elsewhere(server); @@ -487,7 +533,7 @@ describe("watching a document", () => { const heard = startWatching(site, theirs.url); await new Promise((resolve) => setTimeout(resolve, 500)); await site.restartNode(); - // Losing the node makes everyone look again, whether or not it moved. + // The replacement answers the watch with the heads it has. await eventually(() => heard.changes > 0); await new Promise((resolve) => setTimeout(resolve, 500)); const before = heard.changes; diff --git a/core/filesystem/src/module-watcher.ts b/core/filesystem/src/module-watcher.ts index 31799c08e..845ff4b6a 100644 --- a/core/filesystem/src/module-watcher.ts +++ b/core/filesystem/src/module-watcher.ts @@ -80,10 +80,12 @@ type WatchedModule = { /** The package doc being watched: a branches doc's is its default branch. */ packageUrl: AutomergeUrl; unwatch: () => void; - onChange: () => void; + onChange: (seen?: string) => void; timer?: ReturnType; /** The url at heads this watch takes to be current. */ version?: string; + /** What the tracker last said it holds of the package doc. */ + seen?: string; /** The first lookup only sets `version`, rather than reloading. */ baseline: boolean; /** Bumped per lookup, so only the newest one's answer is acted on. */ @@ -362,7 +364,16 @@ export class ModuleWatcher { if (this.#disposed) return undefined; const watched = this.#watchedModules.get(importName); if (watched) return watched; - const onChange = () => { + const onChange = (seen?: string) => { + if (seen !== undefined) { + // The tracker holds what it did: nothing arrived. + if (seen === entry.seen) return; + const first = entry.seen === undefined; + entry.seen = seen; + // The lookup that settles the version is still to come, and finds + // whatever this is. + if (first && entry.version === undefined) return; + } if (entry.timer) clearTimeout(entry.timer); entry.timer = setTimeout(() => { entry.timer = undefined; @@ -404,17 +415,19 @@ export class ModuleWatcher { warnBranchesUnsupported(importName as AutomergeUrl); entry.unwatch(); entry.packageUrl = packageUrl; + // What the tracker said it held was of the other doc. + entry.seen = undefined; entry.unwatch = this.tracker.watch(packageUrl, entry.onChange); } return latest; } /** - * One lookup at a time per package. A package doc that's still syncing in - * says it changed with every batch, and a lookup can take as long as the - * automerge worker's resolve timeout; starting one per change piles them up - * on a worker that's slow because it's busy. Changes that land meanwhile - * collapse into one more lookup once this one's back. + * One lookup and import at a time per package. A package doc that's still + * syncing in says it changed with every batch, and a lookup or an import + * can take as long as the automerge worker's resolve timeout; starting one + * per change piles them up on a worker that's slow because it's busy. + * Changes that land meanwhile collapse into one more once this one's back. */ private async reloadIfMoved(importName: string, entry: WatchedModule) { if (entry.looking) { @@ -450,7 +463,7 @@ export class ModuleWatcher { log(`change in ${importName}, reloading at ${latest}`); // Keyed by the stable importName so this reload supersedes any // still-retrying announce of an older version of the same module. - void this.announce(latest, importName, latest); + await this.announce(latest, importName, latest); } private detachDocWatcher(importName: string) { diff --git a/core/filesystem/src/package-tracker.ts b/core/filesystem/src/package-tracker.ts index e0a403e8d..769a147a5 100644 --- a/core/filesystem/src/package-tracker.ts +++ b/core/filesystem/src/package-tracker.ts @@ -37,9 +37,13 @@ export type PackageTracker = { findLatest(url: AutomergeUrl): Promise; /** * Call `onChange` whenever the doc at `url` may have new heads, until the - * returned function is called. A spurious call costs a `findLatest`. + * returned function is called. A tracker that can tell passes `seen`, a + * summary of what it holds of the doc: the same `seen` twice means nothing + * arrived in between, so there's nothing to look up. It's not a version to + * import, and it may come when nothing moved. Without it, every call costs + * a `findLatest`. */ - watch(url: AutomergeUrl, onChange: () => void): () => void; + watch(url: AutomergeUrl, onChange: (seen?: string) => void): () => void; }; /** @@ -57,30 +61,33 @@ export function trackPackagesInRepo(repo: Repo): PackageTracker { watch(url, onChange) { let handle: DocHandle | undefined; let stopped = false; + // The change event's payload isn't what the tracker says. + const changed = () => onChange(); Promise.resolve() .then(() => repo.find(url)) .then( (found) => { if (stopped) return; handle = found; - handle.on("change", onChange); + handle.on("change", changed); }, (error) => console.warn(`couldn't watch ${url}`, error) ); return () => { stopped = true; - handle?.off("change", onChange); + handle?.off("change", changed); }; }, }; } /** - * Package docs resolved by the service worker, so they never load on this - * thread: the automerge worker redirects a package url without heads to the - * one at the doc's latest heads, and lets go of the doc soon after. `watch` - * is the sync node's (`watchDocument` from - * `@inkandswitch/patchwork-bootloader`). + * Package docs that never load on this thread. `watch` is the sync node's + * (`watchDocument` from `@inkandswitch/patchwork-bootloader`), which says what + * it holds, so a restarted node saying the same needs no lookup. The service + * worker finds the latest version: the automerge worker redirects a package + * url without heads to the one at the doc's latest heads, and lets go of the + * doc soon after. */ export function trackPackagesViaServiceWorker( watch: PackageTracker["watch"] diff --git a/core/filesystem/test/module-watcher.test.ts b/core/filesystem/test/module-watcher.test.ts index 9d44487df..7f97a832c 100644 --- a/core/filesystem/test/module-watcher.test.ts +++ b/core/filesystem/test/module-watcher.test.ts @@ -402,7 +402,7 @@ describe("ModuleWatcher package trackers", () => { */ function trackElsewhere(repo: InstanceType) { const lookups: string[] = []; - const watching = new Map void>(); + const watching = new Map[1]>(); const real = trackPackagesInRepo(repo); const tracker: PackageTracker = { async findLatest(url) { @@ -466,6 +466,58 @@ describe("ModuleWatcher package trackers", () => { } }); + it("looks the package up only when what the tracker holds changes", async () => { + const folders = new AutomergeRepo({ peerId: "folders" as PeerId }); + const tab = new AutomergeRepo({ peerId: "tab" as PeerId }); + try { + const folder = folders.create({ "@patchwork": { type: "folder" } }); + const settings = tab.create({ + "@patchwork": { type: "patchwork:module-settings" }, + modules: [folder.url], + }); + const { tracker, lookups, watching } = trackElsewhere(folders); + const imported: string[] = []; + const watcher = new ModuleWatcher( + tab, + { system: settings.url }, + () => {}, + undefined, + async (url) => { + imported.push(url); + return { plugins: [] }; + }, + tracker + ); + await watcher.doneLoading; + const said = watching.get(folder.url)!; + // What it holds, once it's watching. + said("a"); + await pause(400); + lookups.length = 0; + // One with heads is answered without going anywhere. + const headless = () => lookups.filter((url) => !url.includes("#")); + + // A restarted tracker saying the same: nothing arrived. + said("a"); + await pause(400); + expect(headless()).toEqual([]); + expect(imported).toHaveLength(1); + + // Something arrived. What it says isn't the version to import: the + // lookup finds that. + folder.change((d: any) => { + d.rev = 1; + }); + said("b"); + await pause(400); + expect(headless()).toEqual([folder.url]); + expect(imported).toEqual([imported[0], folder.view(folder.heads()).url]); + watcher.dispose(); + } finally { + await Promise.all([folders.shutdown(), tab.shutdown()]).catch(() => {}); + } + }); + it("reloads for a change that lands while the first lookup is in flight", async () => { const folders = new AutomergeRepo({ peerId: "folders" as PeerId }); const tab = new AutomergeRepo({ peerId: "tab" as PeerId });