only treat new heads as new heads

.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.

core/bootloader/src/automerge-protocol-handler.ts

@@ -185,6 +185,16 @@ function getRepo(): Promise<Repo> {
   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<Repo> {
   await control.holding;
   await initWasm();
@@ -193,7 +203,7 @@ async function buildRepo(): Promise<Repo> {
     ? await buildKeyhiveRepo(syncServer.keyhive)
     : { repo: await buildPlainRepo() };
 
-  (self as any).repo = repo;
+  builtRepo = repo;
   if (hive) (self as any).hive = hive;
   return repo;
 }

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. */

core/bootloader/src/sync-node.ts

@@ -118,6 +118,15 @@ export class SyncNode {
   readonly #echoing = new Map<string, number>();
   /** Peer deliveries waiting for this microtask's flush, per sedimentree. */
   readonly #outbound = new Map<string, DataRecord[]>();
+  /**
+   * 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<string, number>();
   /** Per sedimentree, the round with the server in progress. */
   readonly #rounds = new Map<string, Round>();
@@ -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<boolean> {
+    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<string, string[]>();
+    // 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) {

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 };

core/bootloader/src/sync-remote.ts

@@ -53,7 +53,7 @@ export class PortSyncRemote implements SubductionRemote {
   readonly #pending = new Map<number, Pending>();
   readonly #ephemeral = new Set<string>();
   /** Per sedimentree, who wants to hear it may have changed. */
-  readonly #watchers = new Map<string, Set<() => void>>();
+  readonly #watchers = new Map<string, Set<(heads?: string[]) => 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;

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<Doc>({ count: 0 });
+    await other.flush();
+
+    const site = origin(server);
+    const remote = site.remote();
+    cleanup.push(async () => remote.shutdown());
+    const heard: Array<string[] | undefined> = [];
+    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;