From af040bdeafeaae1255c3e8d189ae3b8ede7b1b8e Mon Sep 17 00:00:00 2001 From: chee Date: Thu, 22 Oct 2026 21:36:42 +0100 Subject: [PATCH 1/7] lock first, then wait --- .changeset/automerge-worker-nudge-grace.md | 11 + core/bootloader/package.json | 4 + .../src/automerge-protocol-handler-worker.ts | 1035 +---------------- .../src/automerge-protocol-handler.ts | 1020 ++++++++++++++++ core/bootloader/src/service-worker.ts | 27 +- core/bootloader/src/worker-control.ts | 51 +- core/bootloader/src/worker-early.ts | 94 ++ .../src/vite/service-worker-plugin.ts | 6 + packages/e2e/tests/automerge-worker.spec.ts | 108 ++ 9 files changed, 1312 insertions(+), 1044 deletions(-) create mode 100644 .changeset/automerge-worker-nudge-grace.md create mode 100644 core/bootloader/src/automerge-protocol-handler.ts create mode 100644 core/bootloader/src/worker-early.ts create mode 100644 packages/e2e/tests/automerge-worker.spec.ts diff --git a/.changeset/automerge-worker-nudge-grace.md b/.changeset/automerge-worker-nudge-grace.md new file mode 100644 index 00000000..2079b642 --- /dev/null +++ b/.changeset/automerge-worker-nudge-grace.md @@ -0,0 +1,11 @@ +--- +"@inkandswitch/patchwork-bootloader": patch +"@inkandswitch/patchwork": patch +"@inkandswitch/patchwork-e2e": patch +--- + +The service worker no longer mistakes a booting automerge worker for a missing one. The worker only asked for its lock once its whole import graph (most of a megabyte) had loaded, so asset requests made in that window found no lock and logged "no automerge worker; asking tabs to start one". Each tab then dropped its connection to the booting worker and reconnected sync. In a single tab that happened on every reload. + +The worker's script, `automerge-protocol-handler-worker.js`, is now a small module that takes the lock and holds on to connecting tabs before it imports the worker itself, now `@inkandswitch/patchwork-bootloader/automerge-protocol-handler`. The vite plugin emits that as a second file, `automerge-protocol-handler.js`, beside the first. As a backstop, the service worker waits three seconds before nudging, and only nudges if the request is still unanswered and there's still no lock. + +The e2e suite gains `automerge-worker.spec.ts`, which checks that the worker holds its lock once a tab is up and that a request racing a cold worker boot isn't reported as a missing worker. diff --git a/core/bootloader/package.json b/core/bootloader/package.json index e73b7fab..caa29afb 100644 --- a/core/bootloader/package.json +++ b/core/bootloader/package.json @@ -43,6 +43,10 @@ "import": "./dist/automerge-protocol-handler-worker.js", "types": "./dist/automerge-protocol-handler-worker.d.ts" }, + "./automerge-protocol-handler": { + "import": "./dist/automerge-protocol-handler.js", + "types": "./dist/automerge-protocol-handler.d.ts" + }, "./module-loader": { "import": "./dist/module-loader.js", "types": "./dist/module-loader.d.ts" diff --git a/core/bootloader/src/automerge-protocol-handler-worker.ts b/core/bootloader/src/automerge-protocol-handler-worker.ts index 53fdb618..4890f374 100644 --- a/core/bootloader/src/automerge-protocol-handler-worker.ts +++ b/core/bootloader/src/automerge-protocol-handler-worker.ts @@ -1,1019 +1,18 @@ -// The Repo that resolves `automerge:` URLs for the service worker, in a -// SharedWorker: one instance serves every tab and lives as long as any tab -// does. -// -// It also hosts the origin's sync node (sync-node.ts), the one Subduction node -// with a connection to the sync server. Every tab's Repo syncs through it, and -// so does the resolver Repo here, which is otherwise a Repo like any tab's: the -// same IndexedDB and the storage channel to the tabs. -// When the service worker misses the cache for a request that looks like a URL -// encoded URL, it broadcasts a HandoffRequestMessage on HANDOFF_CHANNEL; we -// resolve the automerge URL, write the response into the service worker's -// cache (keyed by a Request reconstructed to match the one it's holding), and -// reply on the same channel. -import { - initializeWasm, - getChangesMetaSince, - hasHeads, - stats, -} from "@automerge/automerge/slim"; -// eslint-disable-next-line -// @ts-ignore — the default init is a wasm-bindgen runtime helper not in the .d.ts -import initSubduction from "@automerge/automerge-subduction/slim"; - -import { - INTERCEPTOR_PREFIX, - Repo, - isImmutableString, - isValidAutomergeUrl, - parseAutomergeUrl, - stringifyAutomergeUrl, - type AnyDocumentId, - type AutomergeUrl, - type DocHandle, - type DocumentId, - type PeerId, - type StorageId, -} from "@automerge/automerge-repo/slim"; -import { - getDefaultBranch, - packageDocuments, - resolvePath, -} from "@inkandswitch/patchwork-filesystem"; - -import { IndexedDBStorageAdapter } from "@automerge/automerge-repo-storage-indexeddb"; -import { WebSocketWorkerClientAdapter } from "@automerge/automerge-repo-network-websocket"; -import type { - AutomergeRepoKeyhive, - SyncServerSelection, -} from "@automerge/automerge-repo-keyhive"; - -import { DEFAULT_CLASSIC_SYNC_SERVER } from "./sync-config.js"; -import { loadKeyhiveSigner, loadOrCreateSigner } from "./signer.js"; -import { keyhiveStorageName, storagePrefix } from "./storage.js"; -import { startWorkerControl } from "./worker-control.js"; -import { SyncNode } from "./sync-node.js"; -import { PortSyncRemote, relayKeyhive } from "./sync-remote.js"; -import { - AUTOMERGE_WORKER_LOCK, - HANDOFF_CHANNEL, - PACKAGE_MEMORY_CHANNEL, - type HandoffCachedMessage, - type HandoffOnlineMessage, - type HandoffAbortMessage, - type HandoffRequestMessage, - type HandoffResponseMessage, - type PackageMemoryAcceptedMessage, - type PackageMemoryDocument, - type PackageMemoryReplyMessage, - type PackageMemoryRequestMessage, -} from "./types.js"; - -declare const __SYNC_SERVER__: { - url: string; - keyhive?: SyncServerSelection; - useIdFactory?: boolean; -}; - -const syncServer = - typeof __SYNC_SERVER__ !== "undefined" - ? __SYNC_SERVER__ - : { url: "wss://subduction.sync.inkandswitch.com" }; - -const RESOLVE_TIMEOUT_MS = 30_000; - -const CACHEABLE_STATUSES = [200, 203, 204]; - -const control = startWorkerControl("automerge-protocol-handler-worker", { - onMessage: handleControlMessage, - onRestart: drainForRestart, - lock: AUTOMERGE_WORKER_LOCK, -}); -// The sync node and the Repo wait for the lock: until the previous instance -// is gone, it's still syncing and resolving, against the same IndexedDB. -let holdingLock = false; -void control.holding.then(() => (holdingLock = true)); -const log = control.log; - -// ── The sync node ────────────────────────────────────────────────────── - -let wasmPromise: Promise | null = null; - -function initWasm(): Promise { - if (!wasmPromise) { - log("fetching wasm"); - watchWasmMemories(); - wasmPromise = Promise.all([ - initializeWasm(new Request("/automerge.wasm")), - initSubduction(new Request("/subduction.wasm")), - ]).then(() => log("wasm initialized")); - wasmPromise.catch(() => { - wasmPromise = null; - }); - } - return wasmPromise; -} - -let syncNodePromise: Promise | null = null; - -function getSyncNode(): Promise { - if (!syncNodePromise) { - syncNodePromise = (async () => { - await control.holding; - await initWasm(); - const storage = new IndexedDBStorageAdapter(); - // On keyhive sites the Repos store encrypted documents under the - // interceptor's prefix and sign as the keyhive identity. - const node = new SyncNode({ - signer: syncServer.keyhive - ? await loadKeyhiveSigner( - new IndexedDBStorageAdapter(keyhiveStorageName) - ) - : await loadOrCreateSigner(storage), - storage, - storagePrefix: syncServer.keyhive ? INTERCEPTOR_PREFIX : undefined, - endpoints: [syncServer.url], - log, - }); - (self as any).syncNode = node; - return node; - })(); - // Nothing syncs without it: start again from scratch, as a new instance. - syncNodePromise.catch((error) => { - syncNodePromise = null; - control.fail("the sync node failed to start", error); - }); - } - return syncNodePromise; -} - -/** Serve a Repo on `port` once the node is up; messages queue until then. */ -function connectToSyncNode(port: MessagePort): void { - getSyncNode().then( - (node) => node.connect(port), - (error) => { - console.error("sync node failed to start", error); - port.close(); - } - ); -} - -/** The resolver Repo's connection to the node, which lives right here. */ -function resolverRemote(): PortSyncRemote { - return new PortSyncRemote(() => { - const { port1, port2 } = new MessageChannel(); - connectToSyncNode(port2); - return port1; - }); -} - -// ── The repo ─────────────────────────────────────────────────────────── - -let repoPromise: Promise | null = null; - -function getRepo(): Promise { - if (!repoPromise) { - repoPromise = buildRepo(); - // Nothing resolves without it (the wasm fetch failed, say): start again - // from scratch, as a new instance. - repoPromise.catch((error) => { - repoPromise = null; - control.fail("the resolver's Repo failed to start", error); - }); - } - return repoPromise; -} - -async function buildRepo(): Promise { - await control.holding; - await initWasm(); - - const { repo, hive } = syncServer.keyhive - ? await buildKeyhiveRepo(syncServer.keyhive) - : { repo: await buildPlainRepo() }; - - (self as any).repo = repo; - if (hive) (self as any).hive = hive; - return repo; -} - -async function buildPlainRepo(): Promise { - const storage = new IndexedDBStorageAdapter(); - return new Repo({ - signer: await loadOrCreateSigner(storage), - storage, - peerId: - `${storagePrefix}-resolver-${Math.random().toString(36).slice(2)}` as PeerId, - subductionRemote: resolverRemote(), - headsChannel: `${storagePrefix}-heads`, - enableRemoteHeadsGossiping: true, - }); -} - -async function buildKeyhiveRepo( - keyhiveSyncServer: SyncServerSelection -): Promise<{ repo: Repo; hive: AutomergeRepoKeyhive }> { - const { initKeyhiveWasm, initializeAutomergeRepoKeyhive } = - await import("@automerge/automerge-repo-keyhive"); - initKeyhiveWasm(); - const remote = resolverRemote(); - const { hive, repo } = await initializeAutomergeRepoKeyhive({ - // ARK injects an `idFactory` deriving document ids from keyhive. A site - // can opt out of it with `keyhive: { useIdFactory: false }`. - createRepo: ({ idFactory, ...config }) => - relayKeyhive( - new Repo( - syncServer.useIdFactory === false ? config : { ...config, idFactory } - ), - remote - ), - storage: new IndexedDBStorageAdapter(keyhiveStorageName), - peerIdSuffix: - `${storagePrefix}-resolver` + Math.random().toString(36).slice(2), - automaticArchiveIngestion: true, - cachingMode: "periodic", - // ARK selects the relay via `syncServer`, which pairs the contact card with - // the matching peer id. Omitting it defaults to "subduction". - syncServer: keyhiveSyncServer, - repo: { - storage: new IndexedDBStorageAdapter(), - subductionRemote: remote, - headsChannel: `${storagePrefix}-heads`, - enableRemoteHeadsGossiping: true, - }, - }); - - hive.networkAdapter.whenReady().then(() => { - (hive.networkAdapter as any).syncKeyhive(); - }); - - return { repo, hive }; -} - -// ── Classic sync ─────────────────────────────────────────────────────── - -let classicSyncServer = DEFAULT_CLASSIC_SYNC_SERVER; -let classicSyncAdapter: WebSocketWorkerClientAdapter | null = null; -let classicSyncConnect: Promise | null = null; - -function connectClassicSyncNetwork(server: string): Promise { - const url = server.trim() || DEFAULT_CLASSIC_SYNC_SERVER; - if (classicSyncConnect && classicSyncServer === url) - return classicSyncConnect; - - if (classicSyncAdapter && classicSyncServer !== url) { - classicSyncAdapter.disconnect(); - classicSyncAdapter = null; - } - - classicSyncServer = url; - const connecting = (async () => { - const repo = await getRepo(); - if (!classicSyncAdapter) { - classicSyncAdapter = new WebSocketWorkerClientAdapter(url); - repo.networkSubsystem.addNetworkAdapter(classicSyncAdapter); - } - await classicSyncAdapter.whenReady(); - log("classic sync connected", url); - })(); - - // Clear the memo on failure so a later attempt can retry, and swallow the - // rejection on this copy so it isn't reported as unhandled — callers get it - // from the promise we return. - classicSyncConnect = connecting; - connecting.catch(() => { - if (classicSyncConnect === connecting) classicSyncConnect = null; - }); - return connecting; -} - -// ── Control protocol ─────────────────────────────────────────────────── - -function handleControlMessage( - data: any, - controlPort: MessagePort, - event: MessageEvent -): void { - if (data?.type === "sync-connect") { - const [port] = event.ports; - if (port) connectToSyncNode(port); - return; - } - if (data?.type !== "connect-classic-sync") return; - const [replyPort] = event.ports; - const server = - typeof data.server === "string" ? data.server : DEFAULT_CLASSIC_SYNC_SERVER; - connectClassicSyncNetwork(server).then( - () => { - replyPort?.postMessage({ type: "connect-classic-sync-ready" }); - replyPort?.close(); - }, - (err) => { - console.error("connectClassicSyncNetwork failed", err); - replyPort?.postMessage({ - type: "connect-classic-sync-failed", - error: String(err), - }); - replyPort?.close(); - } - ); -} - -// ── Resolving ────────────────────────────────────────────────────────── - -function waitForHeads( - handle: DocHandle, - hexHeads: string[], - signal: AbortSignal -): Promise { - if (hasHeads(handle.doc(), hexHeads)) return Promise.resolve(true); - if (signal.aborted) return Promise.resolve(false); - return new Promise((resolve) => { - const cleanup = () => { - handle.off("heads-changed", check); - signal.removeEventListener("abort", onAbort); - }; - const check = () => { - if (!hasHeads(handle.doc(), hexHeads)) return; - cleanup(); - resolve(true); - }; - const onAbort = () => { - cleanup(); - resolve(false); - }; - handle.on("heads-changed", check); - signal.addEventListener("abort", onAbort); - // The heads may have landed between the check above and subscribing. - check(); - }); -} - -const CATCH_UP_MS = 3_000; - -async function caughtUpWithPeers( - repo: Repo, - handle: DocHandle, - signal: AbortSignal -): Promise { - if (!repo.isSubductionConnected()) return; - const peers = (await repo.connectedSubductionPeerIds()) as StorageId[]; - const caughtUp = () => { - const states = peers.map((peer) => handle.isCaughtUpWith(peer)); - return ( - states.some((state) => state !== undefined) && !states.includes(false) - ); - }; - if (caughtUp() || signal.aborted) return; - await new Promise((resolve) => { - const done = () => { - clearTimeout(timer); - handle.off("remote-heads", check); - handle.off("heads-changed", check); - signal.removeEventListener("abort", done); - resolve(); - }; - const check = () => { - if (caughtUp()) done(); - }; - const timer = setTimeout(done, CATCH_UP_MS); - handle.on("remote-heads", check); - handle.on("heads-changed", check); - signal.addEventListener("abort", done); - if (signal.aborted) done(); - else check(); - }); -} - -/** - * Thrown instead of returning a Response when the request should fail as a - * network error rather than resolve to something the caller can memoize. - * See {@link HandoffAbortMessage}. - */ -class AbortHandoff extends Error {} - -async function resolveAutomergeUrl( - automergeURL: URL, - signal: AbortSignal -): Promise { - const repo = await getRepo(); - // What it finds, to record against the package. - const found: DocHandle[] = []; - const find = async (id: AnyDocumentId) => { - const handle = await repo.find(id, { signal }); - found.push(handle as DocHandle); - return handle; - }; - // `resolvePath` only finds documents. - const tracked: Repo = Object.create(repo, { find: { value: find } }); - const [maybeAutomergeUrl, ...path] = automergeURL.href.split("/"); - - if (!isValidAutomergeUrl(maybeAutomergeUrl)) { - return new Response("invalid automerge url", { status: 400 }); - } - - if (path.length && !path[path.length - 1]) path.pop(); - - const { heads, hexHeads, documentId } = parseAutomergeUrl(maybeAutomergeUrl); - - // todo, maybe a bad idea? maybe we should throw instead of es-module-caching - // the headless req - if (!heads) { - let folder = await find(maybeAutomergeUrl); - await caughtUpWithPeers(repo, folder, signal); - // Packages used to be published as a branches doc, which stands for its - // default branch. The tab's ModuleWatcher looks packages up through here, - // so this is where that's followed. - const branch = getDefaultBranch(folder.doc()); - if (branch) { - folder = await find(branch); - await caughtUpWithPeers(repo, folder, signal); - branchOf.set(documentId, folder.documentId); - } - const url = stringifyAutomergeUrl({ - documentId: folder.documentId, - heads: folder.heads(), - }); - const location = `/${encodeURIComponent(url)}${path.length ? `/${path.join("/")}` : ""}`; - return Response.redirect(location, 307); - } - - const baseHandle = await find(stringifyAutomergeUrl({ documentId })); - if (!(await waitForHeads(baseHandle, hexHeads ?? [], signal))) { - throw new AbortHandoff( - `heads not found for ${maybeAutomergeUrl} within ${RESOLVE_TIMEOUT_MS}ms` - ); - } - - const resolved = await resolvePath( - tracked, - baseHandle.view(heads), - path.map(decodeURIComponent) - ); - if (!resolved) { - throw new Error( - `couldn't resolve ${path.join("/")} in folder at ${maybeAutomergeUrl}` - ); - } - recordLoaded(documentId, found, path.map(decodeURIComponent).join("/")); - - const body: BodyInit = - resolved.content instanceof Uint8Array - ? (new Uint8Array(resolved.content) as BlobPart) - : resolved.content; - - return new Response(body, { - status: 200, - headers: { "content-type": resolved.type }, - }); -} - -const handoffChannel = new BroadcastChannel(HANDOFF_CHANNEL); - -function replyToHandoff(id: string, status: number, body: string): void { - handoffChannel.postMessage({ - id, - type: "response", - response: { status, body, headers: { "content-type": "text/plain" } }, - } satisfies HandoffResponseMessage); -} - -function impatience(limit: number) { - return new Promise((_, reject) => - setTimeout( - () => reject(new Error(`resolve timeout after ${limit}ms`)), - limit - ) - ); -} - -const EVICT_IDLE_MS = 5_000; -let handoffsInFlight = 0; -let evictTimer: ReturnType | undefined; - -function handoffStarted() { - handoffsInFlight++; - clearTimeout(evictTimer); - clearTimeout(recycleTimer); -} - -function handoffFinished() { - if (--handoffsInFlight > 0) return; - clearTimeout(evictTimer); - evictTimer = setTimeout(() => { - evictLoaded().catch((error) => console.error("eviction failed", error)); - }, EVICT_IDLE_MS); - clearTimeout(recycleTimer); - recycleTimer = setTimeout(recycleIfHolding, RECYCLE_IDLE_MS); -} - -// ── Recycling ────────────────────────────────────────────────────────── -// A wasm memory grows and never shrinks: a module holds the most it ever -// needed at once until its global goes. Resolving packages at boot takes the -// resolver Repo's automerge to 90–130MB, and letting the documents go gives -// none of it back. So once handoffs stop for RECYCLE_IDLE_MS, an instance whose -// automerge holds more than RECYCLE_AT_MB restarts itself, as it does for a -// new build: tabs spawn the replacement and reconnect their sync, the service -// worker re-sends any handoff still unanswered, and what was resolved is in -// its cache, so the replacement doesn't load it all again. - -const RECYCLE_AT_MB = 64; -const RECYCLE_IDLE_MS = 15_000; -let recycleTimer: ReturnType | undefined; - -/** Every wasm memory instantiated here, with the file it was compiled from. */ -const wasmMemories: Array<{ url: string; memory: WebAssembly.Memory }> = []; -let watchingWasm = false; - -/** - * Note the memory of every wasm instance instantiated from a response from - * here on: automerge and subduction instantiate from their `Request`s that - * way, and don't hand back the instance. - */ -function watchWasmMemories() { - if (watchingWasm) return; - watchingWasm = true; - const wasm = WebAssembly as { - instantiateStreaming: typeof WebAssembly.instantiateStreaming; - }; - const instantiateStreaming = wasm.instantiateStreaming; - wasm.instantiateStreaming = async (source, imports) => { - const response = await source; - const result = await instantiateStreaming(response, imports); - for (const value of Object.values(result.instance.exports)) { - if (value instanceof WebAssembly.Memory) { - wasmMemories.push({ url: response.url, memory: value }); - } - } - return result; - }; -} - -function automergeMemoryMb(): number { - let bytes = 0; - for (const { url, memory } of wasmMemories) { - if (new URL(url, self.location.href).pathname.endsWith("/automerge.wasm")) { - bytes += memory.buffer.byteLength; - } - } - return Math.round(bytes / 1048576); -} - -function recycleIfHolding() { - if (handoffsInFlight > 0 || draining) return; - // A replacement doesn't connect to it again; whoever asked for it would - // have to. - if (classicSyncConnect) return; - const mb = automergeMemoryMb(); - if (mb <= RECYCLE_AT_MB) return; - control.restart( - `recycling: automerge holds ${mb}MB of wasm memory with nothing to resolve` - ); -} - -async function evictLoaded() { - const repo = await repoPromise?.catch(() => null); - if (!repo) return; - const ids = Object.keys(repo.handles) as DocumentId[]; - log( - `evicting ${ids.length} document(s) after ${EVICT_IDLE_MS}ms without a handoff` - ); - for (const id of ids) { - if (handoffsInFlight > 0) return; - await repo.removeFromCache(id); - } -} - -async function handleHandoffRequest(message: HandoffRequestMessage) { - const { id, request } = message; - - let handoff: URL; - try { - handoff = new URL(request.handoffURL); - } catch { - console.error("couldn't parse handoff url", request); - replyToHandoff( - id, - 400, - `couldn't parse a special url out of ${request.url}` - ); - return; - } - - // Other handlers may be listening on the channel for other schemes, so stay - // quiet rather than clobbering their reply with an error. - if (handoff.protocol !== "automerge:") { - log( - `ignoring handoff ${id} for non-automerge url ${handoff}. not my circus, not my monkeys` - ); - return; - } - - handoffStarted(); - try { - await respondToHandoff(message, handoff); - } finally { - handoffFinished(); - } -} - -async function respondToHandoff(message: HandoffRequestMessage, handoff: URL) { - const { id, cachename, request } = message; - - let response: Response; - try { - log(`resolving handoff ${id} for ${handoff}`); - const signal = AbortSignal.timeout(RESOLVE_TIMEOUT_MS); - response = await Promise.race([ - resolveAutomergeUrl(handoff, signal), - impatience(RESOLVE_TIMEOUT_MS), - ]); - } catch (error) { - if (error instanceof AbortHandoff) { - handoffChannel.postMessage({ - id, - type: "abort", - reason: error.message, - } satisfies HandoffAbortMessage); - return; - } - console.error(`error resolving ${request.url}`, error); - replyToHandoff( - id, - 557, - error instanceof Error - ? `${error.message}\n\n${error.stack}` - : String(error) - ); - return; - } - - try { - if (!CACHEABLE_STATUSES.includes(response.status)) { - // Errors, redirects and the like go back inline for the service worker to - // serve directly, so they aren't cached forever (still in esmodulecache, - // cleared after a refresh) - log(`responding inline to ${request.url} with ${response.status}`); - handoffChannel.postMessage({ - id, - type: "response", - response: { - status: response.status, - headers: Object.fromEntries(response.headers.entries()), - body: response.body ? await response.text() : undefined, - }, - } satisfies HandoffResponseMessage); - return; - } - - // Reconstruct the request the service worker is holding so the entry matches - // its cache.match. `destination` isn't constructible but doesn't participate - // in cache matching. - const cacheKey = new Request(request.url, { - method: request.method, - headers: request.headers, - referrer: request.referrer, - }); - const cache = await caches.open(cachename); - await cache.put(cacheKey, response); - log(`cached ${cacheKey.url} in ${cachename}`); - handoffChannel.postMessage({ - id, - type: "cached", - } satisfies HandoffCachedMessage); - } catch (error) { - console.error(`failed to reply for ${request.url}`, error); - replyToHandoff(id, 558, String(error)); - } -} - -// How long a restart waits for in-flight handoffs before closing anyway. -const DRAIN_HANDOFFS_MS = 3_000; -let draining = false; - -// Stop taking new handoffs and let the ones in flight finish. The service -// worker keeps every unanswered handoff pending and re-broadcasts them when -// our replacement announces itself online, so nothing dropped here is lost — -// waiting is a courtesy to requests that are nearly done, not a requirement. -async function drainForRestart(): Promise { - draining = true; - const deadline = Date.now() + DRAIN_HANDOFFS_MS; - while (handoffsInFlight > 0 && Date.now() < deadline) { - await new Promise((resolve) => setTimeout(resolve, 50)); - } - if (handoffsInFlight > 0) { - console.warn( - `[lifecycle] closing with ${handoffsInFlight} handoff(s) in flight; the service worker will re-send them` - ); - } - const repo = await repoPromise?.catch(() => null); - await repo?.flush(); - await saveLoaded().catch((error) => - console.warn("couldn't keep the package records for the next instance", error) - ); - const node = await syncNodePromise?.catch(() => null); - await node?.shutdown(); -} - -handoffChannel.addEventListener("message", (event) => { - if (event.data?.type !== "request") return; - if (draining) { - log(`ignoring handoff ${event.data.id} while restarting`); - return; - } - // The previous instance is still answering; once it's gone, `online` gets - // the service worker to send us whatever it left unanswered. - if (!holdingLock) return; - void handleHandoffRequest(event.data as HandoffRequestMessage); -}); - -// Announce ourselves so the service worker can re-broadcast handoff requests -// sent while we were booting, or left by the instance before us. -void control.holding.then(() => - handoffChannel.postMessage({ type: "online" } satisfies HandoffOnlineMessage) +// The automerge worker's script. It takes the worker's lock before anything +// else loads, so the service worker can see there's a worker from the moment +// it starts, then imports the worker itself (automerge-protocol-handler.ts). +// Keep its imports small: none of this runs until they've all loaded. +import { AUTOMERGE_WORKER_LOCK } from "./types.js"; +import { importRest, startEarly } from "./worker-early.js"; + +// Built as a file of its own next to this one, so a variable rather than a +// literal: the bundler leaves the import alone. +const REST = "./automerge-protocol-handler.js"; + +// Not awaited: this script should finish evaluating now, so nothing about the +// worker waits on the import but the code that's in it. +void importRest( + startEarly(AUTOMERGE_WORKER_LOCK), + "automerge-protocol-handler-worker", + new URL(REST, import.meta.url).href ); - -// ── Measuring packages ───────────────────────────────────────────────── - -/** - * What resolves have loaded of each package, by its root doc: each document - * measured when it was loaded (again only if its heads have moved), and the - * paths asked for through it. - */ -const loadedByPackage = new Map< - DocumentId, - Map< - DocumentId, - { - heads: string; - measured: PackageMemoryDocument; - through: Set; - /** It's the file a path ended on, not a folder on the way. */ - file: boolean; - } - > ->(); - -/** Branches docs followed to their default branch, which holds the package. */ -const branchOf = new Map(); - -// What's been loaded outlives the instance: a restart (a recycle, or a new -// build) saves it for the replacement, which resolves from the service -// worker's cache rather than loading the packages again. -const LOADED_CACHE = "patchwork-package-memory"; -const LOADED_KEY = "/loaded-by-package"; - -type SavedLoaded = { - packages: Array< - [ - DocumentId, - Array< - [ - DocumentId, - { - heads: string; - measured: PackageMemoryDocument; - through: string[]; - file: boolean; - }, - ] - >, - ] - >; - branches: Array<[DocumentId, DocumentId]>; -}; - -async function saveLoaded() { - await restored; - const saved: SavedLoaded = { - packages: [...loadedByPackage].map(([packageId, documents]) => [ - packageId, - [...documents].map(([id, entry]) => [ - id, - { ...entry, through: [...entry.through] }, - ]), - ]), - branches: [...branchOf], - }; - const cache = await caches.open(LOADED_CACHE); - await cache.put(LOADED_KEY, new Response(JSON.stringify(saved))); -} - -/** Settles once what the instance before saved has been taken back. */ -const restored: Promise = (async () => { - try { - const cache = await caches.open(LOADED_CACHE); - const response = await cache.match(LOADED_KEY); - if (!response) return; - const saved = (await response.json()) as SavedLoaded; - for (const [packageId, documents] of saved.packages) { - const into = loadedByPackage.get(packageId) ?? new Map(); - for (const [id, entry] of documents) { - // Loaded since by this instance: that's newer. - if (into.has(id)) continue; - into.set(id, { ...entry, through: new Set(entry.through) }); - } - loadedByPackage.set(packageId, into); - } - for (const [from, to] of saved.branches) { - if (!branchOf.has(from)) branchOf.set(from, to); - } - } catch (error) { - console.warn("couldn't take back the package records", error); - } -})(); - -/** `handles` were found in turn resolving `path`: the root first, the file last. */ -function recordLoaded( - packageId: DocumentId, - handles: DocHandle[], - path: string -) { - let documents = loadedByPackage.get(packageId); - if (!documents) loadedByPackage.set(packageId, (documents = new Map())); - handles.forEach((handle, i) => { - const heads = handle.heads().join(","); - let entry = documents.get(handle.documentId); - if (!entry || entry.heads !== heads) { - entry = { - heads, - measured: measureDocument("", handle), - through: entry?.through ?? new Set(), - file: entry?.file ?? false, - }; - documents.set(handle.documentId, entry); - } - entry.through.add(path); - if (i > 0 && i === handles.length - 1) entry.file = true; - }); -} - -/** What's been loaded of the package at `url`, root first. */ -function loadedOf(url: string): PackageMemoryDocument[] { - if (!isValidAutomergeUrl(url)) return []; - const { documentId } = parseAutomergeUrl(url); - const packageId = branchOf.get(documentId) ?? documentId; - return [...(loadedByPackage.get(packageId) ?? [])].map(([id, entry]) => { - const paths = [...entry.through]; - const path = - id === packageId - ? "" - : entry.file - ? paths[0] - : sharedDirectory(paths); - return { ...entry.measured, path }; - }); -} - -/** The directory every one of `paths` is in. */ -function sharedDirectory(paths: string[]): string { - const dirs = paths.map((path) => path.split("/").slice(0, -1)); - const shared = dirs[0] ?? []; - let n = shared.length; - for (const dir of dirs) { - while (n > 0 && dir.slice(0, n).join("/") !== shared.slice(0, n).join("/")) n--; - } - return shared.slice(0, n).join("/"); -} - -const MEASURE_TIMEOUT_MS = 300_000; -// Most of a package's documents (READMEs, images) aren't stored here until -// it's measured, since modules never import them; one that doesn't turn up -// in this long is reported as unavailable. -const MEASURE_FIND_TIMEOUT_MS = 15_000; -const packageMemoryChannel = new BroadcastChannel(PACKAGE_MEMORY_CHANNEL); - -// One package at a time, so measuring them all doesn't load them all at once. -let measuring: Promise = Promise.resolve(); - -packageMemoryChannel.addEventListener("message", (event) => { - const message = event.data as PackageMemoryRequestMessage; - // Only one instance answers: the one holding the lock. - if (message?.type !== "request" || draining || !holdingLock) return; - packageMemoryChannel.postMessage({ - id: message.id, - type: "accepted", - } satisfies PackageMemoryAcceptedMessage); - if (!message.walk) { - void restored.then(() => - packageMemoryChannel.postMessage({ - id: message.id, - type: "reply", - documents: loadedOf(message.url), - } satisfies PackageMemoryReplyMessage) - ); - return; - } - measuring = measuring.then(() => replyWithPackageMemory(message)); -}); - -async function replyWithPackageMemory({ id, url }: PackageMemoryRequestMessage) { - const documents: PackageMemoryDocument[] = []; - let error: string | undefined; - // Holds off the eviction of everything loaded, as a handoff does; the - // documents measured are let go one at a time instead. - handoffStarted(); - try { - await measurePackage(url, (document) => documents.push(document)); - } catch (e) { - error = e instanceof Error ? e.message : String(e); - } finally { - handoffFinished(); - } - packageMemoryChannel.postMessage({ - id, - type: "reply", - documents, - error, - } satisfies PackageMemoryReplyMessage); -} - -/** - * Find the package at `url` the way a resolve does, then each of its - * documents in turn, letting each go once it's measured unless a resolve is - * under way, which may be using it too. What's left goes with the rest once - * handoffs stop. - */ -async function measurePackage( - url: string, - found: (document: PackageMemoryDocument) => void -) { - if (!isValidAutomergeUrl(url)) throw new Error(`not an automerge url: ${url}`); - const repo = await getRepo(); - const signal = AbortSignal.timeout(MEASURE_TIMEOUT_MS); - const release = async (id: DocumentId) => { - // This measurement is the only handoff in flight. - if (handoffsInFlight === 1 && repo.handles[id]) { - await repo.removeFromCache(id); - } - }; - const find = (id: AnyDocumentId) => { - return repo.find(id, { - signal: AbortSignal.any([ - signal, - AbortSignal.timeout(MEASURE_FIND_TIMEOUT_MS), - ]), - }); - }; - const tracked: Repo = Object.create(repo, { find: { value: find } }); - const { documentId, heads, hexHeads } = parseAutomergeUrl(url); - let root = await find(stringifyAutomergeUrl({ documentId })); - if (heads) { - if (!(await waitForHeads(root, hexHeads ?? [], signal))) { - throw new Error(`heads not found for ${url}`); - } - root = root.view(heads); - } else { - await caughtUpWithPeers(repo, root, signal); - const branch = getDefaultBranch(root.doc()); - if (branch) { - await release(root.documentId); - root = await find(branch); - await caughtUpWithPeers(repo, root, signal); - } - } - for await (const document of packageDocuments(tracked, root)) { - if ("error" in document) { - const { path, url, error } = document; - found({ - url, - path, - ops: 0, - changes: 0, - largestChange: 0, - error: error instanceof Error ? error.message : String(error), - }); - continue; - } - found(measureDocument(document.path, document.handle)); - await release(document.handle.documentId); - } -} - -function measureDocument( - path: string, - handle: DocHandle -): PackageMemoryDocument { - const doc = handle.fullDoc(); - const { numOps, numChanges } = stats(doc); - let largestChange = 0; - for (const { startOp, maxOp } of getChangesMetaSince(doc, [])) { - largestChange = Math.max(largestChange, maxOp - startOp + 1); - } - const content = (doc as { content?: unknown } | undefined)?.content; - return { - url: handle.url, - path, - ops: numOps, - changes: numChanges, - largestChange, - content: - typeof content === "string" - ? { kind: "text", length: content.length } - : isImmutableString(content) - ? { kind: "immutable", length: String(content).length } - : content instanceof Uint8Array - ? { kind: "bytes", length: content.byteLength } - : undefined, - }; -} diff --git a/core/bootloader/src/automerge-protocol-handler.ts b/core/bootloader/src/automerge-protocol-handler.ts new file mode 100644 index 00000000..39a63e5d --- /dev/null +++ b/core/bootloader/src/automerge-protocol-handler.ts @@ -0,0 +1,1020 @@ +// The Repo that resolves `automerge:` URLs for the service worker, in a +// SharedWorker: one instance serves every tab and lives as long as any tab +// does. The worker's script is automerge-protocol-handler-worker.ts, which +// takes the lock and then imports this. +// +// It also hosts the origin's sync node (sync-node.ts), the one Subduction node +// with a connection to the sync server. Every tab's Repo syncs through it, and +// so does the resolver Repo here, which is otherwise a Repo like any tab's: the +// same IndexedDB and the storage channel to the tabs. +// When the service worker misses the cache for a request that looks like a URL +// encoded URL, it broadcasts a HandoffRequestMessage on HANDOFF_CHANNEL; we +// resolve the automerge URL, write the response into the service worker's +// cache (keyed by a Request reconstructed to match the one it's holding), and +// reply on the same channel. +import { + initializeWasm, + getChangesMetaSince, + hasHeads, + stats, +} from "@automerge/automerge/slim"; +// eslint-disable-next-line +// @ts-ignore — the default init is a wasm-bindgen runtime helper not in the .d.ts +import initSubduction from "@automerge/automerge-subduction/slim"; + +import { + INTERCEPTOR_PREFIX, + Repo, + isImmutableString, + isValidAutomergeUrl, + parseAutomergeUrl, + stringifyAutomergeUrl, + type AnyDocumentId, + type AutomergeUrl, + type DocHandle, + type DocumentId, + type PeerId, + type StorageId, +} from "@automerge/automerge-repo/slim"; +import { + getDefaultBranch, + packageDocuments, + resolvePath, +} from "@inkandswitch/patchwork-filesystem"; + +import { IndexedDBStorageAdapter } from "@automerge/automerge-repo-storage-indexeddb"; +import { WebSocketWorkerClientAdapter } from "@automerge/automerge-repo-network-websocket"; +import type { + AutomergeRepoKeyhive, + SyncServerSelection, +} from "@automerge/automerge-repo-keyhive"; + +import { DEFAULT_CLASSIC_SYNC_SERVER } from "./sync-config.js"; +import { loadKeyhiveSigner, loadOrCreateSigner } from "./signer.js"; +import { keyhiveStorageName, storagePrefix } from "./storage.js"; +import { startWorkerControl } from "./worker-control.js"; +import { SyncNode } from "./sync-node.js"; +import { PortSyncRemote, relayKeyhive } from "./sync-remote.js"; +import { + AUTOMERGE_WORKER_LOCK, + HANDOFF_CHANNEL, + PACKAGE_MEMORY_CHANNEL, + type HandoffCachedMessage, + type HandoffOnlineMessage, + type HandoffAbortMessage, + type HandoffRequestMessage, + type HandoffResponseMessage, + type PackageMemoryAcceptedMessage, + type PackageMemoryDocument, + type PackageMemoryReplyMessage, + type PackageMemoryRequestMessage, +} from "./types.js"; + +declare const __SYNC_SERVER__: { + url: string; + keyhive?: SyncServerSelection; + useIdFactory?: boolean; +}; + +const syncServer = + typeof __SYNC_SERVER__ !== "undefined" + ? __SYNC_SERVER__ + : { url: "wss://subduction.sync.inkandswitch.com" }; + +const RESOLVE_TIMEOUT_MS = 30_000; + +const CACHEABLE_STATUSES = [200, 203, 204]; + +const control = startWorkerControl("automerge-protocol-handler-worker", { + onMessage: handleControlMessage, + onRestart: drainForRestart, + lock: AUTOMERGE_WORKER_LOCK, +}); +// The sync node and the Repo wait for the lock: until the previous instance +// is gone, it's still syncing and resolving, against the same IndexedDB. +let holdingLock = false; +void control.holding.then(() => (holdingLock = true)); +const log = control.log; + +// ── The sync node ────────────────────────────────────────────────────── + +let wasmPromise: Promise | null = null; + +function initWasm(): Promise { + if (!wasmPromise) { + log("fetching wasm"); + watchWasmMemories(); + wasmPromise = Promise.all([ + initializeWasm(new Request("/automerge.wasm")), + initSubduction(new Request("/subduction.wasm")), + ]).then(() => log("wasm initialized")); + wasmPromise.catch(() => { + wasmPromise = null; + }); + } + return wasmPromise; +} + +let syncNodePromise: Promise | null = null; + +function getSyncNode(): Promise { + if (!syncNodePromise) { + syncNodePromise = (async () => { + await control.holding; + await initWasm(); + const storage = new IndexedDBStorageAdapter(); + // On keyhive sites the Repos store encrypted documents under the + // interceptor's prefix and sign as the keyhive identity. + const node = new SyncNode({ + signer: syncServer.keyhive + ? await loadKeyhiveSigner( + new IndexedDBStorageAdapter(keyhiveStorageName) + ) + : await loadOrCreateSigner(storage), + storage, + storagePrefix: syncServer.keyhive ? INTERCEPTOR_PREFIX : undefined, + endpoints: [syncServer.url], + log, + }); + (self as any).syncNode = node; + return node; + })(); + // Nothing syncs without it: start again from scratch, as a new instance. + syncNodePromise.catch((error) => { + syncNodePromise = null; + control.fail("the sync node failed to start", error); + }); + } + return syncNodePromise; +} + +/** Serve a Repo on `port` once the node is up; messages queue until then. */ +function connectToSyncNode(port: MessagePort): void { + getSyncNode().then( + (node) => node.connect(port), + (error) => { + console.error("sync node failed to start", error); + port.close(); + } + ); +} + +/** The resolver Repo's connection to the node, which lives right here. */ +function resolverRemote(): PortSyncRemote { + return new PortSyncRemote(() => { + const { port1, port2 } = new MessageChannel(); + connectToSyncNode(port2); + return port1; + }); +} + +// ── The repo ─────────────────────────────────────────────────────────── + +let repoPromise: Promise | null = null; + +function getRepo(): Promise { + if (!repoPromise) { + repoPromise = buildRepo(); + // Nothing resolves without it (the wasm fetch failed, say): start again + // from scratch, as a new instance. + repoPromise.catch((error) => { + repoPromise = null; + control.fail("the resolver's Repo failed to start", error); + }); + } + return repoPromise; +} + +async function buildRepo(): Promise { + await control.holding; + await initWasm(); + + const { repo, hive } = syncServer.keyhive + ? await buildKeyhiveRepo(syncServer.keyhive) + : { repo: await buildPlainRepo() }; + + (self as any).repo = repo; + if (hive) (self as any).hive = hive; + return repo; +} + +async function buildPlainRepo(): Promise { + const storage = new IndexedDBStorageAdapter(); + return new Repo({ + signer: await loadOrCreateSigner(storage), + storage, + peerId: + `${storagePrefix}-resolver-${Math.random().toString(36).slice(2)}` as PeerId, + subductionRemote: resolverRemote(), + headsChannel: `${storagePrefix}-heads`, + enableRemoteHeadsGossiping: true, + }); +} + +async function buildKeyhiveRepo( + keyhiveSyncServer: SyncServerSelection +): Promise<{ repo: Repo; hive: AutomergeRepoKeyhive }> { + const { initKeyhiveWasm, initializeAutomergeRepoKeyhive } = + await import("@automerge/automerge-repo-keyhive"); + initKeyhiveWasm(); + const remote = resolverRemote(); + const { hive, repo } = await initializeAutomergeRepoKeyhive({ + // ARK injects an `idFactory` deriving document ids from keyhive. A site + // can opt out of it with `keyhive: { useIdFactory: false }`. + createRepo: ({ idFactory, ...config }) => + relayKeyhive( + new Repo( + syncServer.useIdFactory === false ? config : { ...config, idFactory } + ), + remote + ), + storage: new IndexedDBStorageAdapter(keyhiveStorageName), + peerIdSuffix: + `${storagePrefix}-resolver` + Math.random().toString(36).slice(2), + automaticArchiveIngestion: true, + cachingMode: "periodic", + // ARK selects the relay via `syncServer`, which pairs the contact card with + // the matching peer id. Omitting it defaults to "subduction". + syncServer: keyhiveSyncServer, + repo: { + storage: new IndexedDBStorageAdapter(), + subductionRemote: remote, + headsChannel: `${storagePrefix}-heads`, + enableRemoteHeadsGossiping: true, + }, + }); + + hive.networkAdapter.whenReady().then(() => { + (hive.networkAdapter as any).syncKeyhive(); + }); + + return { repo, hive }; +} + +// ── Classic sync ─────────────────────────────────────────────────────── + +let classicSyncServer = DEFAULT_CLASSIC_SYNC_SERVER; +let classicSyncAdapter: WebSocketWorkerClientAdapter | null = null; +let classicSyncConnect: Promise | null = null; + +function connectClassicSyncNetwork(server: string): Promise { + const url = server.trim() || DEFAULT_CLASSIC_SYNC_SERVER; + if (classicSyncConnect && classicSyncServer === url) + return classicSyncConnect; + + if (classicSyncAdapter && classicSyncServer !== url) { + classicSyncAdapter.disconnect(); + classicSyncAdapter = null; + } + + classicSyncServer = url; + const connecting = (async () => { + const repo = await getRepo(); + if (!classicSyncAdapter) { + classicSyncAdapter = new WebSocketWorkerClientAdapter(url); + repo.networkSubsystem.addNetworkAdapter(classicSyncAdapter); + } + await classicSyncAdapter.whenReady(); + log("classic sync connected", url); + })(); + + // Clear the memo on failure so a later attempt can retry, and swallow the + // rejection on this copy so it isn't reported as unhandled — callers get it + // from the promise we return. + classicSyncConnect = connecting; + connecting.catch(() => { + if (classicSyncConnect === connecting) classicSyncConnect = null; + }); + return connecting; +} + +// ── Control protocol ─────────────────────────────────────────────────── + +function handleControlMessage( + data: any, + controlPort: MessagePort, + event: MessageEvent +): void { + if (data?.type === "sync-connect") { + const [port] = event.ports; + if (port) connectToSyncNode(port); + return; + } + if (data?.type !== "connect-classic-sync") return; + const [replyPort] = event.ports; + const server = + typeof data.server === "string" ? data.server : DEFAULT_CLASSIC_SYNC_SERVER; + connectClassicSyncNetwork(server).then( + () => { + replyPort?.postMessage({ type: "connect-classic-sync-ready" }); + replyPort?.close(); + }, + (err) => { + console.error("connectClassicSyncNetwork failed", err); + replyPort?.postMessage({ + type: "connect-classic-sync-failed", + error: String(err), + }); + replyPort?.close(); + } + ); +} + +// ── Resolving ────────────────────────────────────────────────────────── + +function waitForHeads( + handle: DocHandle, + hexHeads: string[], + signal: AbortSignal +): Promise { + if (hasHeads(handle.doc(), hexHeads)) return Promise.resolve(true); + if (signal.aborted) return Promise.resolve(false); + return new Promise((resolve) => { + const cleanup = () => { + handle.off("heads-changed", check); + signal.removeEventListener("abort", onAbort); + }; + const check = () => { + if (!hasHeads(handle.doc(), hexHeads)) return; + cleanup(); + resolve(true); + }; + const onAbort = () => { + cleanup(); + resolve(false); + }; + handle.on("heads-changed", check); + signal.addEventListener("abort", onAbort); + // The heads may have landed between the check above and subscribing. + check(); + }); +} + +const CATCH_UP_MS = 3_000; + +async function caughtUpWithPeers( + repo: Repo, + handle: DocHandle, + signal: AbortSignal +): Promise { + if (!repo.isSubductionConnected()) return; + const peers = (await repo.connectedSubductionPeerIds()) as StorageId[]; + const caughtUp = () => { + const states = peers.map((peer) => handle.isCaughtUpWith(peer)); + return ( + states.some((state) => state !== undefined) && !states.includes(false) + ); + }; + if (caughtUp() || signal.aborted) return; + await new Promise((resolve) => { + const done = () => { + clearTimeout(timer); + handle.off("remote-heads", check); + handle.off("heads-changed", check); + signal.removeEventListener("abort", done); + resolve(); + }; + const check = () => { + if (caughtUp()) done(); + }; + const timer = setTimeout(done, CATCH_UP_MS); + handle.on("remote-heads", check); + handle.on("heads-changed", check); + signal.addEventListener("abort", done); + if (signal.aborted) done(); + else check(); + }); +} + +/** + * Thrown instead of returning a Response when the request should fail as a + * network error rather than resolve to something the caller can memoize. + * See {@link HandoffAbortMessage}. + */ +class AbortHandoff extends Error {} + +async function resolveAutomergeUrl( + automergeURL: URL, + signal: AbortSignal +): Promise { + const repo = await getRepo(); + // What it finds, to record against the package. + const found: DocHandle[] = []; + const find = async (id: AnyDocumentId) => { + const handle = await repo.find(id, { signal }); + found.push(handle as DocHandle); + return handle; + }; + // `resolvePath` only finds documents. + const tracked: Repo = Object.create(repo, { find: { value: find } }); + const [maybeAutomergeUrl, ...path] = automergeURL.href.split("/"); + + if (!isValidAutomergeUrl(maybeAutomergeUrl)) { + return new Response("invalid automerge url", { status: 400 }); + } + + if (path.length && !path[path.length - 1]) path.pop(); + + const { heads, hexHeads, documentId } = parseAutomergeUrl(maybeAutomergeUrl); + + // todo, maybe a bad idea? maybe we should throw instead of es-module-caching + // the headless req + if (!heads) { + let folder = await find(maybeAutomergeUrl); + await caughtUpWithPeers(repo, folder, signal); + // Packages used to be published as a branches doc, which stands for its + // default branch. The tab's ModuleWatcher looks packages up through here, + // so this is where that's followed. + const branch = getDefaultBranch(folder.doc()); + if (branch) { + folder = await find(branch); + await caughtUpWithPeers(repo, folder, signal); + branchOf.set(documentId, folder.documentId); + } + const url = stringifyAutomergeUrl({ + documentId: folder.documentId, + heads: folder.heads(), + }); + const location = `/${encodeURIComponent(url)}${path.length ? `/${path.join("/")}` : ""}`; + return Response.redirect(location, 307); + } + + const baseHandle = await find(stringifyAutomergeUrl({ documentId })); + if (!(await waitForHeads(baseHandle, hexHeads ?? [], signal))) { + throw new AbortHandoff( + `heads not found for ${maybeAutomergeUrl} within ${RESOLVE_TIMEOUT_MS}ms` + ); + } + + const resolved = await resolvePath( + tracked, + baseHandle.view(heads), + path.map(decodeURIComponent) + ); + if (!resolved) { + throw new Error( + `couldn't resolve ${path.join("/")} in folder at ${maybeAutomergeUrl}` + ); + } + recordLoaded(documentId, found, path.map(decodeURIComponent).join("/")); + + const body: BodyInit = + resolved.content instanceof Uint8Array + ? (new Uint8Array(resolved.content) as BlobPart) + : resolved.content; + + return new Response(body, { + status: 200, + headers: { "content-type": resolved.type }, + }); +} + +const handoffChannel = new BroadcastChannel(HANDOFF_CHANNEL); + +function replyToHandoff(id: string, status: number, body: string): void { + handoffChannel.postMessage({ + id, + type: "response", + response: { status, body, headers: { "content-type": "text/plain" } }, + } satisfies HandoffResponseMessage); +} + +function impatience(limit: number) { + return new Promise((_, reject) => + setTimeout( + () => reject(new Error(`resolve timeout after ${limit}ms`)), + limit + ) + ); +} + +const EVICT_IDLE_MS = 5_000; +let handoffsInFlight = 0; +let evictTimer: ReturnType | undefined; + +function handoffStarted() { + handoffsInFlight++; + clearTimeout(evictTimer); + clearTimeout(recycleTimer); +} + +function handoffFinished() { + if (--handoffsInFlight > 0) return; + clearTimeout(evictTimer); + evictTimer = setTimeout(() => { + evictLoaded().catch((error) => console.error("eviction failed", error)); + }, EVICT_IDLE_MS); + clearTimeout(recycleTimer); + recycleTimer = setTimeout(recycleIfHolding, RECYCLE_IDLE_MS); +} + +// ── Recycling ────────────────────────────────────────────────────────── +// A wasm memory grows and never shrinks: a module holds the most it ever +// needed at once until its global goes. Resolving packages at boot takes the +// resolver Repo's automerge to 90–130MB, and letting the documents go gives +// none of it back. So once handoffs stop for RECYCLE_IDLE_MS, an instance whose +// automerge holds more than RECYCLE_AT_MB restarts itself, as it does for a +// new build: tabs spawn the replacement and reconnect their sync, the service +// worker re-sends any handoff still unanswered, and what was resolved is in +// its cache, so the replacement doesn't load it all again. + +const RECYCLE_AT_MB = 64; +const RECYCLE_IDLE_MS = 15_000; +let recycleTimer: ReturnType | undefined; + +/** Every wasm memory instantiated here, with the file it was compiled from. */ +const wasmMemories: Array<{ url: string; memory: WebAssembly.Memory }> = []; +let watchingWasm = false; + +/** + * Note the memory of every wasm instance instantiated from a response from + * here on: automerge and subduction instantiate from their `Request`s that + * way, and don't hand back the instance. + */ +function watchWasmMemories() { + if (watchingWasm) return; + watchingWasm = true; + const wasm = WebAssembly as { + instantiateStreaming: typeof WebAssembly.instantiateStreaming; + }; + const instantiateStreaming = wasm.instantiateStreaming; + wasm.instantiateStreaming = async (source, imports) => { + const response = await source; + const result = await instantiateStreaming(response, imports); + for (const value of Object.values(result.instance.exports)) { + if (value instanceof WebAssembly.Memory) { + wasmMemories.push({ url: response.url, memory: value }); + } + } + return result; + }; +} + +function automergeMemoryMb(): number { + let bytes = 0; + for (const { url, memory } of wasmMemories) { + if (new URL(url, self.location.href).pathname.endsWith("/automerge.wasm")) { + bytes += memory.buffer.byteLength; + } + } + return Math.round(bytes / 1048576); +} + +function recycleIfHolding() { + if (handoffsInFlight > 0 || draining) return; + // A replacement doesn't connect to it again; whoever asked for it would + // have to. + if (classicSyncConnect) return; + const mb = automergeMemoryMb(); + if (mb <= RECYCLE_AT_MB) return; + control.restart( + `recycling: automerge holds ${mb}MB of wasm memory with nothing to resolve` + ); +} + +async function evictLoaded() { + const repo = await repoPromise?.catch(() => null); + if (!repo) return; + const ids = Object.keys(repo.handles) as DocumentId[]; + log( + `evicting ${ids.length} document(s) after ${EVICT_IDLE_MS}ms without a handoff` + ); + for (const id of ids) { + if (handoffsInFlight > 0) return; + await repo.removeFromCache(id); + } +} + +async function handleHandoffRequest(message: HandoffRequestMessage) { + const { id, request } = message; + + let handoff: URL; + try { + handoff = new URL(request.handoffURL); + } catch { + console.error("couldn't parse handoff url", request); + replyToHandoff( + id, + 400, + `couldn't parse a special url out of ${request.url}` + ); + return; + } + + // Other handlers may be listening on the channel for other schemes, so stay + // quiet rather than clobbering their reply with an error. + if (handoff.protocol !== "automerge:") { + log( + `ignoring handoff ${id} for non-automerge url ${handoff}. not my circus, not my monkeys` + ); + return; + } + + handoffStarted(); + try { + await respondToHandoff(message, handoff); + } finally { + handoffFinished(); + } +} + +async function respondToHandoff(message: HandoffRequestMessage, handoff: URL) { + const { id, cachename, request } = message; + + let response: Response; + try { + log(`resolving handoff ${id} for ${handoff}`); + const signal = AbortSignal.timeout(RESOLVE_TIMEOUT_MS); + response = await Promise.race([ + resolveAutomergeUrl(handoff, signal), + impatience(RESOLVE_TIMEOUT_MS), + ]); + } catch (error) { + if (error instanceof AbortHandoff) { + handoffChannel.postMessage({ + id, + type: "abort", + reason: error.message, + } satisfies HandoffAbortMessage); + return; + } + console.error(`error resolving ${request.url}`, error); + replyToHandoff( + id, + 557, + error instanceof Error + ? `${error.message}\n\n${error.stack}` + : String(error) + ); + return; + } + + try { + if (!CACHEABLE_STATUSES.includes(response.status)) { + // Errors, redirects and the like go back inline for the service worker to + // serve directly, so they aren't cached forever (still in esmodulecache, + // cleared after a refresh) + log(`responding inline to ${request.url} with ${response.status}`); + handoffChannel.postMessage({ + id, + type: "response", + response: { + status: response.status, + headers: Object.fromEntries(response.headers.entries()), + body: response.body ? await response.text() : undefined, + }, + } satisfies HandoffResponseMessage); + return; + } + + // Reconstruct the request the service worker is holding so the entry matches + // its cache.match. `destination` isn't constructible but doesn't participate + // in cache matching. + const cacheKey = new Request(request.url, { + method: request.method, + headers: request.headers, + referrer: request.referrer, + }); + const cache = await caches.open(cachename); + await cache.put(cacheKey, response); + log(`cached ${cacheKey.url} in ${cachename}`); + handoffChannel.postMessage({ + id, + type: "cached", + } satisfies HandoffCachedMessage); + } catch (error) { + console.error(`failed to reply for ${request.url}`, error); + replyToHandoff(id, 558, String(error)); + } +} + +// How long a restart waits for in-flight handoffs before closing anyway. +const DRAIN_HANDOFFS_MS = 3_000; +let draining = false; + +// Stop taking new handoffs and let the ones in flight finish. The service +// worker keeps every unanswered handoff pending and re-broadcasts them when +// our replacement announces itself online, so nothing dropped here is lost — +// waiting is a courtesy to requests that are nearly done, not a requirement. +async function drainForRestart(): Promise { + draining = true; + const deadline = Date.now() + DRAIN_HANDOFFS_MS; + while (handoffsInFlight > 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 50)); + } + if (handoffsInFlight > 0) { + console.warn( + `[lifecycle] closing with ${handoffsInFlight} handoff(s) in flight; the service worker will re-send them` + ); + } + const repo = await repoPromise?.catch(() => null); + await repo?.flush(); + await saveLoaded().catch((error) => + console.warn("couldn't keep the package records for the next instance", error) + ); + const node = await syncNodePromise?.catch(() => null); + await node?.shutdown(); +} + +handoffChannel.addEventListener("message", (event) => { + if (event.data?.type !== "request") return; + if (draining) { + log(`ignoring handoff ${event.data.id} while restarting`); + return; + } + // The previous instance is still answering; once it's gone, `online` gets + // the service worker to send us whatever it left unanswered. + if (!holdingLock) return; + void handleHandoffRequest(event.data as HandoffRequestMessage); +}); + +// Announce ourselves so the service worker can re-broadcast handoff requests +// sent while we were booting, or left by the instance before us. +void control.holding.then(() => + handoffChannel.postMessage({ type: "online" } satisfies HandoffOnlineMessage) +); + +// ── Measuring packages ───────────────────────────────────────────────── + +/** + * What resolves have loaded of each package, by its root doc: each document + * measured when it was loaded (again only if its heads have moved), and the + * paths asked for through it. + */ +const loadedByPackage = new Map< + DocumentId, + Map< + DocumentId, + { + heads: string; + measured: PackageMemoryDocument; + through: Set; + /** It's the file a path ended on, not a folder on the way. */ + file: boolean; + } + > +>(); + +/** Branches docs followed to their default branch, which holds the package. */ +const branchOf = new Map(); + +// What's been loaded outlives the instance: a restart (a recycle, or a new +// build) saves it for the replacement, which resolves from the service +// worker's cache rather than loading the packages again. +const LOADED_CACHE = "patchwork-package-memory"; +const LOADED_KEY = "/loaded-by-package"; + +type SavedLoaded = { + packages: Array< + [ + DocumentId, + Array< + [ + DocumentId, + { + heads: string; + measured: PackageMemoryDocument; + through: string[]; + file: boolean; + }, + ] + >, + ] + >; + branches: Array<[DocumentId, DocumentId]>; +}; + +async function saveLoaded() { + await restored; + const saved: SavedLoaded = { + packages: [...loadedByPackage].map(([packageId, documents]) => [ + packageId, + [...documents].map(([id, entry]) => [ + id, + { ...entry, through: [...entry.through] }, + ]), + ]), + branches: [...branchOf], + }; + const cache = await caches.open(LOADED_CACHE); + await cache.put(LOADED_KEY, new Response(JSON.stringify(saved))); +} + +/** Settles once what the instance before saved has been taken back. */ +const restored: Promise = (async () => { + try { + const cache = await caches.open(LOADED_CACHE); + const response = await cache.match(LOADED_KEY); + if (!response) return; + const saved = (await response.json()) as SavedLoaded; + for (const [packageId, documents] of saved.packages) { + const into = loadedByPackage.get(packageId) ?? new Map(); + for (const [id, entry] of documents) { + // Loaded since by this instance: that's newer. + if (into.has(id)) continue; + into.set(id, { ...entry, through: new Set(entry.through) }); + } + loadedByPackage.set(packageId, into); + } + for (const [from, to] of saved.branches) { + if (!branchOf.has(from)) branchOf.set(from, to); + } + } catch (error) { + console.warn("couldn't take back the package records", error); + } +})(); + +/** `handles` were found in turn resolving `path`: the root first, the file last. */ +function recordLoaded( + packageId: DocumentId, + handles: DocHandle[], + path: string +) { + let documents = loadedByPackage.get(packageId); + if (!documents) loadedByPackage.set(packageId, (documents = new Map())); + handles.forEach((handle, i) => { + const heads = handle.heads().join(","); + let entry = documents.get(handle.documentId); + if (!entry || entry.heads !== heads) { + entry = { + heads, + measured: measureDocument("", handle), + through: entry?.through ?? new Set(), + file: entry?.file ?? false, + }; + documents.set(handle.documentId, entry); + } + entry.through.add(path); + if (i > 0 && i === handles.length - 1) entry.file = true; + }); +} + +/** What's been loaded of the package at `url`, root first. */ +function loadedOf(url: string): PackageMemoryDocument[] { + if (!isValidAutomergeUrl(url)) return []; + const { documentId } = parseAutomergeUrl(url); + const packageId = branchOf.get(documentId) ?? documentId; + return [...(loadedByPackage.get(packageId) ?? [])].map(([id, entry]) => { + const paths = [...entry.through]; + const path = + id === packageId + ? "" + : entry.file + ? paths[0] + : sharedDirectory(paths); + return { ...entry.measured, path }; + }); +} + +/** The directory every one of `paths` is in. */ +function sharedDirectory(paths: string[]): string { + const dirs = paths.map((path) => path.split("/").slice(0, -1)); + const shared = dirs[0] ?? []; + let n = shared.length; + for (const dir of dirs) { + while (n > 0 && dir.slice(0, n).join("/") !== shared.slice(0, n).join("/")) n--; + } + return shared.slice(0, n).join("/"); +} + +const MEASURE_TIMEOUT_MS = 300_000; +// Most of a package's documents (READMEs, images) aren't stored here until +// it's measured, since modules never import them; one that doesn't turn up +// in this long is reported as unavailable. +const MEASURE_FIND_TIMEOUT_MS = 15_000; +const packageMemoryChannel = new BroadcastChannel(PACKAGE_MEMORY_CHANNEL); + +// One package at a time, so measuring them all doesn't load them all at once. +let measuring: Promise = Promise.resolve(); + +packageMemoryChannel.addEventListener("message", (event) => { + const message = event.data as PackageMemoryRequestMessage; + // Only one instance answers: the one holding the lock. + if (message?.type !== "request" || draining || !holdingLock) return; + packageMemoryChannel.postMessage({ + id: message.id, + type: "accepted", + } satisfies PackageMemoryAcceptedMessage); + if (!message.walk) { + void restored.then(() => + packageMemoryChannel.postMessage({ + id: message.id, + type: "reply", + documents: loadedOf(message.url), + } satisfies PackageMemoryReplyMessage) + ); + return; + } + measuring = measuring.then(() => replyWithPackageMemory(message)); +}); + +async function replyWithPackageMemory({ id, url }: PackageMemoryRequestMessage) { + const documents: PackageMemoryDocument[] = []; + let error: string | undefined; + // Holds off the eviction of everything loaded, as a handoff does; the + // documents measured are let go one at a time instead. + handoffStarted(); + try { + await measurePackage(url, (document) => documents.push(document)); + } catch (e) { + error = e instanceof Error ? e.message : String(e); + } finally { + handoffFinished(); + } + packageMemoryChannel.postMessage({ + id, + type: "reply", + documents, + error, + } satisfies PackageMemoryReplyMessage); +} + +/** + * Find the package at `url` the way a resolve does, then each of its + * documents in turn, letting each go once it's measured unless a resolve is + * under way, which may be using it too. What's left goes with the rest once + * handoffs stop. + */ +async function measurePackage( + url: string, + found: (document: PackageMemoryDocument) => void +) { + if (!isValidAutomergeUrl(url)) throw new Error(`not an automerge url: ${url}`); + const repo = await getRepo(); + const signal = AbortSignal.timeout(MEASURE_TIMEOUT_MS); + const release = async (id: DocumentId) => { + // This measurement is the only handoff in flight. + if (handoffsInFlight === 1 && repo.handles[id]) { + await repo.removeFromCache(id); + } + }; + const find = (id: AnyDocumentId) => { + return repo.find(id, { + signal: AbortSignal.any([ + signal, + AbortSignal.timeout(MEASURE_FIND_TIMEOUT_MS), + ]), + }); + }; + const tracked: Repo = Object.create(repo, { find: { value: find } }); + const { documentId, heads, hexHeads } = parseAutomergeUrl(url); + let root = await find(stringifyAutomergeUrl({ documentId })); + if (heads) { + if (!(await waitForHeads(root, hexHeads ?? [], signal))) { + throw new Error(`heads not found for ${url}`); + } + root = root.view(heads); + } else { + await caughtUpWithPeers(repo, root, signal); + const branch = getDefaultBranch(root.doc()); + if (branch) { + await release(root.documentId); + root = await find(branch); + await caughtUpWithPeers(repo, root, signal); + } + } + for await (const document of packageDocuments(tracked, root)) { + if ("error" in document) { + const { path, url, error } = document; + found({ + url, + path, + ops: 0, + changes: 0, + largestChange: 0, + error: error instanceof Error ? error.message : String(error), + }); + continue; + } + found(measureDocument(document.path, document.handle)); + await release(document.handle.documentId); + } +} + +function measureDocument( + path: string, + handle: DocHandle +): PackageMemoryDocument { + const doc = handle.fullDoc(); + const { numOps, numChanges } = stats(doc); + let largestChange = 0; + for (const { startOp, maxOp } of getChangesMetaSince(doc, [])) { + largestChange = Math.max(largestChange, maxOp - startOp + 1); + } + const content = (doc as { content?: unknown } | undefined)?.content; + return { + url: handle.url, + path, + ops: numOps, + changes: numChanges, + largestChange, + content: + typeof content === "string" + ? { kind: "text", length: content.length } + : isImmutableString(content) + ? { kind: "immutable", length: String(content).length } + : content instanceof Uint8Array + ? { kind: "bytes", length: content.byteLength } + : undefined, + }; +} diff --git a/core/bootloader/src/service-worker.ts b/core/bootloader/src/service-worker.ts index 93747aba..0f9ef08c 100644 --- a/core/bootloader/src/service-worker.ts +++ b/core/bootloader/src/service-worker.ts @@ -183,23 +183,32 @@ handoffChannel.addEventListener("message", (event) => { }); const MISSING_WORKER_NUDGE_MS = 5_000; +// A worker that's been spawned only asks for the lock once its imports have +// evaluated, so on a cold start it looks missing for a while. +const MISSING_WORKER_GRACE_MS = 3_000; let lastMissingWorkerNudge = 0; +async function automergeWorkerExists(): Promise { + const { held = [], pending = [] } = await navigator.locks.query(); + return [...held, ...pending].some( + (lock) => lock.name === AUTOMERGE_WORKER_LOCK && lock.mode === "exclusive" + ); +} + /** * If nothing holds the automerge worker's lock, or is waiting for it (a * replacement that hasn't started yet), there's no worker to answer: ask the * tabs to start one, since only a page can. A worker that's been spawned but - * hasn't asked for the lock yet looks the same, which is why this only - * nudges, and spawning it again just reattaches to it. + * hasn't asked for the lock yet looks the same, so this waits out its boot + * first, and only nudges if handoff `id` is still unanswered by then. */ -async function startWorkerIfMissing() { +async function startWorkerIfMissing(id: string) { if (!navigator.locks) return; + if (await automergeWorkerExists()) return; + await new Promise((resolve) => setTimeout(resolve, MISSING_WORKER_GRACE_MS)); + if (!pendingHandoffs.has(id)) return; if (Date.now() - lastMissingWorkerNudge < MISSING_WORKER_NUDGE_MS) return; - const { held = [], pending = [] } = await navigator.locks.query(); - const worker = [...held, ...pending].some( - (lock) => lock.name === AUTOMERGE_WORKER_LOCK && lock.mode === "exclusive" - ); - if (worker) return; + if (await automergeWorkerExists()) return; lastMissingWorkerNudge = Date.now(); lifecycle("warn", "no automerge worker; asking tabs to start one"); await postToClients({ @@ -233,7 +242,7 @@ async function handoff( pendingHandoffs.set(id, { message, resolvers }); log(`broadcasting handoff request for cache ${cachename}`, message); handoffChannel.postMessage(message); - void startWorkerIfMissing(); + void startWorkerIfMissing(id); const timeout = setTimeout(() => { lifecycle( diff --git a/core/bootloader/src/worker-control.ts b/core/bootloader/src/worker-control.ts index 64e27141..58f4937b 100644 --- a/core/bootloader/src/worker-control.ts +++ b/core/bootloader/src/worker-control.ts @@ -4,6 +4,7 @@ // worker's own business and arrives through `onMessage`. import { buildId, isNewerBuild } from "./build-id.js"; +import { takeEarly } from "./worker-early.js"; const MAX_BUFFER = 200; // How long a restart waits on `onRestart` before closing regardless. @@ -190,7 +191,11 @@ export function startWorkerControl( if (failed) return; failed = true; const detail = - error instanceof Error ? error.message : error === undefined ? "" : String(error); + error instanceof Error + ? error.message + : error === undefined + ? "" + : String(error); console.error( `${name} SharedWorker failed, restarting: ${reason}`, error instanceof Error ? error.stack : detail @@ -202,8 +207,8 @@ export function startWorkerControl( }); }; - self.addEventListener("connect", (event) => { - const port = (event as MessageEvent).ports[0]; + const connect = (event: MessageEvent) => { + const port = event.ports[0]; handlers.onConnect?.(port); port.addEventListener("message", (messageEvent) => { @@ -248,34 +253,46 @@ export function startWorkerControl( // A tab that connected mid-drain still needs to hear it, or it would wait // on this instance until the port closes. if (restarting) postToPort(port, restarting); - }); + }; + + // The worker's script may have asked for the lock already, and held on to + // tabs that connected while the rest of it loaded (see worker-early.ts). + const early = takeEarly(); + self.addEventListener("connect", (event) => connect(event as MessageEvent)); + for (const event of early?.take() ?? []) connect(event); console.warn( `[lifecycle] ${name} SharedWorker started` + (buildId === undefined ? "" : ` (build ${buildId})`) ); - const { lock } = handlers; - const holding = - lock && self.navigator?.locks + // Asking again for a lock the script already asked for would wait on itself. + const lock = early ? early.lock : handlers.lock; + const holding = early + ? early.holding + : lock && self.navigator?.locks ? new Promise((resolve) => { - const waiting = setTimeout( - () => - console.warn( - `[lifecycle] ${name} SharedWorker waiting for the previous instance to go` - ), - 1_000 - ); void navigator.locks.request(lock, () => { - clearTimeout(waiting); - alive = { type: "alive", lock }; - for (const port of ports) postToPort(port, alive); resolve(); // Held until this global goes. return new Promise(() => {}); }); }) : Promise.resolve(); + if (lock && self.navigator?.locks) { + const waiting = setTimeout( + () => + console.warn( + `[lifecycle] ${name} SharedWorker waiting for the previous instance to go` + ), + 1_000 + ); + void holding.then(() => { + clearTimeout(waiting); + alive = { type: "alive", lock }; + for (const port of ports) postToPort(port, alive); + }); + } return { log: (...args: unknown[]) => { diff --git a/core/bootloader/src/worker-early.ts b/core/bootloader/src/worker-early.ts new file mode 100644 index 00000000..11a7f561 --- /dev/null +++ b/core/bootloader/src/worker-early.ts @@ -0,0 +1,94 @@ +// What a SharedWorker's script does before it imports the rest of itself. +// +// A module's body only runs once its whole static import graph has been +// fetched and evaluated, which for the automerge worker is most of a megabyte +// and takes from half a second to several. Until it asks for its lock, nothing +// can tell it exists: the service worker, finding no lock, asks the tabs to +// start a worker. So a worker's script is a small module that does this, then +// imports the rest; startWorkerControl picks it up from there. +// +// Handed over through the global rather than a module, so it survives the two +// halves being bundled separately (the dev server bundles each worker on its +// own). + +const EARLY = Symbol.for("@patchwork/worker-early"); + +export type EarlyStart = { + lock: string | undefined; + /** Resolves once this instance holds `lock` (straight away without one). */ + holding: Promise; + /** Stop buffering connections, and return the ones made so far. */ + take(): MessageEvent[]; +}; + +/** + * Ask for `lock` and hold on to every tab that connects. A SharedWorker's + * `connect` events go nowhere without a listener, and tabs connect while the + * rest of the worker is still loading. + */ +export function startEarly(lock?: string): EarlyStart { + const connections: MessageEvent[] = []; + const buffer = (event: Event) => connections.push(event as MessageEvent); + self.addEventListener("connect", buffer); + + const holding = + lock && self.navigator?.locks + ? new Promise((resolve) => { + void navigator.locks.request(lock, () => { + resolve(); + // Held until this global goes. + return new Promise(() => {}); + }); + }) + : Promise.resolve(); + + const early: EarlyStart = { + lock, + holding, + take() { + self.removeEventListener("connect", buffer); + return connections.splice(0); + }, + }; + (globalThis as any)[EARLY] = early; + return early; +} + +/** What {@link startEarly} started, if this worker's script called it. */ +export function takeEarly(): EarlyStart | undefined { + const early = (globalThis as any)[EARLY] as EarlyStart | undefined; + delete (globalThis as any)[EARLY]; + return early; +} + +/** + * Import the rest of the worker. If it doesn't load there's nothing here to + * run it, but this instance holds the lock, so it closes and the tabs start + * another; after a pause, so one that can't load doesn't spin. + */ +export async function importRest( + early: EarlyStart, + name: string, + url: string +): Promise { + try { + await import(/* @vite-ignore */ url); + } catch (error) { + const message = `${name} SharedWorker failed to load ${url}, closing`; + console.error(message, error); + const args = [ + message, + error instanceof Error ? error.stack || error.message : String(error), + ]; + setTimeout(() => { + // The tabs show forwarded console messages, and `closing` is their cue + // to start a replacement (see ClosingMessage); nothing else here can. + for (const event of early.take()) { + const port = event.ports[0]; + port?.postMessage({ type: "console", level: "error", args }); + port?.postMessage({ type: "closing" }); + } + self.close(); + }, 1_000); + } +} diff --git a/core/patchwork/src/vite/service-worker-plugin.ts b/core/patchwork/src/vite/service-worker-plugin.ts index 3b1bb80b..9ea9094a 100644 --- a/core/patchwork/src/vite/service-worker-plugin.ts +++ b/core/patchwork/src/vite/service-worker-plugin.ts @@ -23,6 +23,12 @@ export const workers = [ "@inkandswitch/patchwork-bootloader/automerge-protocol-handler-worker", fileName: "automerge-protocol-handler-worker.js", }, + // What that worker's script imports once it has the lock: a separate file, + // so the script itself has next to nothing to load first. + { + specifier: "@inkandswitch/patchwork-bootloader/automerge-protocol-handler", + fileName: "automerge-protocol-handler.js", + }, { specifier: "@inkandswitch/patchwork-bootloader/module-loader-worker", fileName: "module-loader-worker.js", diff --git a/packages/e2e/tests/automerge-worker.spec.ts b/packages/e2e/tests/automerge-worker.spec.ts new file mode 100644 index 00000000..e4c22b7a --- /dev/null +++ b/packages/e2e/tests/automerge-worker.spec.ts @@ -0,0 +1,108 @@ +import { expect, test, type Page } from "@playwright/test"; +import { + createDoc, + waitForRepoReady, + waitForServiceWorkerActive, +} from "./helpers.js"; + +/** + * The automerge worker (a SharedWorker) holds a Web Lock for its lifetime, and + * the service worker checks for it before handing an `automerge:` request + * off: no lock, no worker, so it asks the tabs to start one. + * + * A worker that's been spawned but is still evaluating its imports hasn't + * asked for the lock yet. Requests made in that window mustn't be mistaken + * for a missing worker: each tab answers the nudge by dropping its connection + * to the booting worker and reconnecting sync. + */ + +// AUTOMERGE_WORKER_LOCK in core/bootloader/src/types.ts. +const LOCK = "@patchwork/automerge-worker"; + +function workerLockHeld(page: Page): Promise { + return page.evaluate(async (lock) => { + const { held = [] } = await navigator.locks.query(); + return held.some((l) => l.name === lock && l.mode === "exclusive"); + }, LOCK); +} + +declare global { + interface Window { + missingWorkerNudges: number; + lockedAtSpawn: Promise; + } +} + +test("the automerge worker holds its lock once the tab is up", async ({ + page, +}) => { + await page.goto("/"); + await waitForServiceWorkerActive(page); + await waitForRepoReady(page); + // Not holding it means it failed before taking it, or it's a build from + // before the lock: either way every handoff would nudge the tabs. + await expect.poll(() => workerLockHeld(page)).toBe(true); +}); + +test("an automerge worker that's still booting isn't reported missing", async ({ + context, +}) => { + const first = await context.newPage(); + await first.goto("/"); + await waitForServiceWorkerActive(first); + await waitForRepoReady(first); + const url = await createDoc(first, { title: "booting" }); + await first.close(); + + // The worker goes with its last tab. Wait for that from a same-origin page + // that doesn't boot the app, so the next tab has to start a fresh one. + const probe = await context.newPage(); + await probe.goto("/automerge-protocol-handler-worker.js"); + await expect.poll(() => workerLockHeld(probe)).toBe(false); + await probe.close(); + + const tab = await context.newPage(); + await tab.addInitScript( + ([lock, docUrl]) => { + if (location.pathname !== "/") return; + window.missingWorkerNudges = 0; + navigator.serviceWorker.addEventListener("message", (event) => { + if (event.data?.type === "automerge-worker-missing") { + window.missingWorkerNudges++; + } + }); + // Right as the tab spawns the worker, so the request reaches the + // service worker while the worker is booting. Any earlier and there + // really is no worker yet, which a page slow to get this far (Firefox, + // often) would be right to be told about. + const SharedWorker = window.SharedWorker; + let raced = false; + class Racing extends SharedWorker { + constructor(url: string | URL, options?: string | WorkerOptions) { + if (!raced && String(url).includes("automerge-protocol-handler")) { + raced = true; + window.lockedAtSpawn = navigator.locks + .query() + .then(({ held = [] }) => held.some((l) => l.name === lock)); + void fetch(`/${encodeURIComponent(docUrl)}`, { + redirect: "manual", + }).catch(() => {}); + } + super(url, options); + } + } + window.SharedWorker = Racing; + }, + [LOCK, url] as const + ); + await tab.goto("/"); + await waitForRepoReady(tab); + + // Otherwise this passes without having raced anything. + expect(await tab.evaluate(() => window.lockedAtSpawn)).toBe(false); + + await expect.poll(() => workerLockHeld(tab)).toBe(true); + // Past the service worker's grace period, so a nudge would have landed. + await tab.waitForTimeout(5_000); + expect(await tab.evaluate(() => window.missingWorkerNudges)).toBe(0); +}); From 020fe2b2254474c8afaabaf3d3b5fa814a6ed594 Mon Sep 17 00:00:00 2001 From: chee Date: Thu, 22 Oct 2026 22:09:39 +0100 Subject: [PATCH 2/7] include reason for failure in failure --- .changeset/package-lookup-reason.md | 5 +++++ core/filesystem/src/package-tracker.ts | 6 +++++- core/filesystem/test/module-watcher.test.ts | 15 +++++++++++++++ 3 files changed, 25 insertions(+), 1 deletion(-) create mode 100644 .changeset/package-lookup-reason.md diff --git a/.changeset/package-lookup-reason.md b/.changeset/package-lookup-reason.md new file mode 100644 index 00000000..f4296692 --- /dev/null +++ b/.changeset/package-lookup-reason.md @@ -0,0 +1,5 @@ +--- +"@inkandswitch/patchwork-filesystem": patch +--- + +When the service worker can't redirect a package url to its heads, the error now includes the automerge worker's reason (the first line of its 557 body — e.g. the doc is unavailable, or the resolve timed out) instead of only the status. diff --git a/core/filesystem/src/package-tracker.ts b/core/filesystem/src/package-tracker.ts index 292fcb6d..e0a403e8 100644 --- a/core/filesystem/src/package-tracker.ts +++ b/core/filesystem/src/package-tracker.ts @@ -104,8 +104,12 @@ export async function fetchLatestViaServiceWorker( new URL(response.url).pathname.split("/")[1] ?? "" ); if (!isValidAutomergeUrl(latest) || !parseAutomergeUrl(latest).heads) { + // The automerge worker's 557 carries why it couldn't resolve the doc. + const reason = (await response.text?.().catch(() => "")) + ?.split("\n")[0] + ?.trim(); throw new Error( - `the service worker didn't redirect ${url} to its heads (HTTP ${response.status} from ${response.url})` + `the service worker didn't redirect ${url} to its heads (HTTP ${response.status} from ${response.url})${reason ? `: ${reason}` : ""}` ); } return latest; diff --git a/core/filesystem/test/module-watcher.test.ts b/core/filesystem/test/module-watcher.test.ts index 9407ec6e..ee3e8bd7 100644 --- a/core/filesystem/test/module-watcher.test.ts +++ b/core/filesystem/test/module-watcher.test.ts @@ -584,6 +584,21 @@ describe("ModuleWatcher package trackers", () => { expect(fetch).toHaveBeenCalledTimes(1); }); + it("passes on the automerge worker's reason when it couldn't resolve", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async (input: URL) => ({ + status: 557, + url: String(input), + text: async () => "Document 2Jts is unavailable\n\nError: ...stack", + })) + ); + vi.stubGlobal("location", { origin: "http://example.test" }); + await expect(fetchLatestViaServiceWorker(AUTOMERGE_URL)).rejects.toThrow( + /HTTP 557 .*: Document 2Jts is unavailable$/ + ); + }); + it("says so when the service worker didn't redirect", async () => { vi.stubGlobal( "fetch", From 06a024afd6271e189062e2a984e862ed70d12694 Mon Sep 17 00:00:00 2001 From: chee Date: Thu, 22 Oct 2026 22:32:28 +0100 Subject: [PATCH 3/7] run e2e on branches --- .github/workflows/e2e.yml | 71 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 .github/workflows/e2e.yml diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml new file mode 100644 index 00000000..afaaf97d --- /dev/null +++ b/.github/workflows/e2e.yml @@ -0,0 +1,71 @@ +name: e2e + +on: + pull_request: + push: + branches: + - main + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +env: + SHELL_REPO: inkandswitch/patchwork.inkandswitch.com + SITE_DIR: sites/patchwork.inkandswitch.com + +jobs: + e2e: + name: e2e (${{ matrix.browser }}) + runs-on: ubuntu-latest + # Firefox's page sometimes never finishes its own wasm init and the boot + # times out, on main too. Its failures show here without failing the run + # until that's fixed. + continue-on-error: ${{ matrix.browser == 'firefox' }} + strategy: + fail-fast: false + matrix: + browser: [chromium, firefox, webkit] + steps: + - uses: actions/checkout@v4 + + - uses: actions/checkout@v4 + with: + repository: ${{ env.SHELL_REPO }} + path: ${{ env.SITE_DIR }} + + - uses: pnpm/action-setup@v4 + with: + run_install: false + + - uses: actions/setup-node@v4 + with: + node-version: 24 + cache: pnpm + + # The same build the preview deploys: the shell, against this branch's + # packages. + - name: build the shell against this branch's packages + run: | + node .github/scripts/use-workspace-packages.mjs "$GITHUB_WORKSPACE/$SITE_DIR" + pnpm install --no-frozen-lockfile + pnpm build + + - name: install ${{ matrix.browser }} + working-directory: packages/e2e + run: pnpm exec playwright install --with-deps ${{ matrix.browser }} + + - name: run the suite + working-directory: packages/e2e + env: + PATCHWORK_E2E_SITE_DIR: ${{ github.workspace }}/${{ env.SITE_DIR }} + run: pnpm exec playwright test --project=${{ matrix.browser }} + + - uses: actions/upload-artifact@v4 + if: failure() + with: + name: e2e-${{ matrix.browser }} + path: | + packages/e2e/playwright-report/ + packages/e2e/test-results/ + retention-days: 7 From 32cb087d40ff0a0724cf25ecb344d3358f30b866 Mon Sep 17 00:00:00 2001 From: chee Date: Thu, 22 Oct 2026 22:33:37 +0100 Subject: [PATCH 4/7] run e2e on branches --- .github/workflows/e2e.yml | 4 ---- 1 file changed, 4 deletions(-) diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index afaaf97d..dba95677 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -18,10 +18,6 @@ jobs: e2e: name: e2e (${{ matrix.browser }}) runs-on: ubuntu-latest - # Firefox's page sometimes never finishes its own wasm init and the boot - # times out, on main too. Its failures show here without failing the run - # until that's fixed. - continue-on-error: ${{ matrix.browser == 'firefox' }} strategy: fail-fast: false matrix: From 72d757e844aa73f30e11d62ec9cb79d344190589 Mon Sep 17 00:00:00 2001 From: chee Date: Thu, 22 Oct 2026 23:31:52 +0100 Subject: [PATCH 5/7] Potential fix for pull request finding 'CodeQL / Workflow does not contain permissions' Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- .github/workflows/e2e.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index dba95677..834ae515 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -1,5 +1,8 @@ name: e2e +permissions: + contents: read + on: pull_request: push: From 32ea25d9000435178f6d963293410268bb381d1a Mon Sep 17 00:00:00 2001 From: chee Date: Fri, 23 Oct 2026 00:53:24 +0100 Subject: [PATCH 6/7] check time after, no race --- .changeset/automerge-worker-nudge-grace.md | 2 +- core/bootloader/src/service-worker.ts | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/.changeset/automerge-worker-nudge-grace.md b/.changeset/automerge-worker-nudge-grace.md index 2079b642..45b3bab4 100644 --- a/.changeset/automerge-worker-nudge-grace.md +++ b/.changeset/automerge-worker-nudge-grace.md @@ -6,6 +6,6 @@ The service worker no longer mistakes a booting automerge worker for a missing one. The worker only asked for its lock once its whole import graph (most of a megabyte) had loaded, so asset requests made in that window found no lock and logged "no automerge worker; asking tabs to start one". Each tab then dropped its connection to the booting worker and reconnected sync. In a single tab that happened on every reload. -The worker's script, `automerge-protocol-handler-worker.js`, is now a small module that takes the lock and holds on to connecting tabs before it imports the worker itself, now `@inkandswitch/patchwork-bootloader/automerge-protocol-handler`. The vite plugin emits that as a second file, `automerge-protocol-handler.js`, beside the first. As a backstop, the service worker waits three seconds before nudging, and only nudges if the request is still unanswered and there's still no lock. +The worker's script, `automerge-protocol-handler-worker.js`, is now a small module that takes the lock and holds on to connecting tabs before it imports the worker itself, now `@inkandswitch/patchwork-bootloader/automerge-protocol-handler`. The vite plugin emits that as a second file, `automerge-protocol-handler.js`, beside the first. As a backstop, the service worker waits three seconds before nudging, and only nudges if the request is still unanswered and there's still no lock. The nudge is throttled to once every five seconds, however many requests are waiting. Before, every request that raced the same cold start got past the throttle and logged its own nudge. The e2e suite gains `automerge-worker.spec.ts`, which checks that the worker holds its lock once a tab is up and that a request racing a cold worker boot isn't reported as a missing worker. diff --git a/core/bootloader/src/service-worker.ts b/core/bootloader/src/service-worker.ts index 0f9ef08c..261c7799 100644 --- a/core/bootloader/src/service-worker.ts +++ b/core/bootloader/src/service-worker.ts @@ -207,8 +207,11 @@ async function startWorkerIfMissing(id: string) { if (await automergeWorkerExists()) return; await new Promise((resolve) => setTimeout(resolve, MISSING_WORKER_GRACE_MS)); if (!pendingHandoffs.has(id)) return; - if (Date.now() - lastMissingWorkerNudge < MISSING_WORKER_NUDGE_MS) return; if (await automergeWorkerExists()) return; + // Check and claim the throttle with no await in between: every handoff that + // raced the same cold start comes through here at once, and would all pass a + // check made before the lock query. + if (Date.now() - lastMissingWorkerNudge < MISSING_WORKER_NUDGE_MS) return; lastMissingWorkerNudge = Date.now(); lifecycle("warn", "no automerge worker; asking tabs to start one"); await postToClients({ From a386e6ef76e130406061f467c78fe7ae6580839c Mon Sep 17 00:00:00 2001 From: chee Date: Fri, 23 Oct 2026 01:48:53 +0100 Subject: [PATCH 7/7] dedup handoffs, let them all ride the same promise --- .changeset/coalesce-package-lookups.md | 8 +++ .../src/automerge-protocol-handler.ts | 44 +++++++++++-- core/filesystem/src/module-watcher.ts | 29 +++++++++ core/filesystem/test/module-watcher.test.ts | 62 +++++++++++++++++++ 4 files changed, 138 insertions(+), 5 deletions(-) create mode 100644 .changeset/coalesce-package-lookups.md diff --git a/.changeset/coalesce-package-lookups.md b/.changeset/coalesce-package-lookups.md new file mode 100644 index 00000000..763dc06e --- /dev/null +++ b/.changeset/coalesce-package-lookups.md @@ -0,0 +1,8 @@ +--- +"@inkandswitch/patchwork-filesystem": patch +"@inkandswitch/patchwork-bootloader": patch +--- + +Package heads lookups no longer pile up on a busy automerge worker. A package doc that's still syncing, as everything is on a fresh login, reports a change with every batch that arrives. The `ModuleWatcher` started a lookup for each one, and the automerge worker ran each as a full resolve with its own 30-second timeout. A worker slowed by the login sync then timed them all out, and each logged its own 557. + +The `ModuleWatcher` now runs one change-triggered lookup per package at a time. Changes that land while one is in flight collapse into a single lookup after it. The automerge worker shares one resolve between concurrent requests for the same package url without heads, so tabs asking at once wait on the same lookup. diff --git a/core/bootloader/src/automerge-protocol-handler.ts b/core/bootloader/src/automerge-protocol-handler.ts index cddf6e6d..bab400d0 100644 --- a/core/bootloader/src/automerge-protocol-handler.ts +++ b/core/bootloader/src/automerge-protocol-handler.ts @@ -623,17 +623,51 @@ async function handleHandoffRequest(message: HandoffRequestMessage) { } } +function resolveWithin(handoff: URL): Promise { + const signal = AbortSignal.timeout(RESOLVE_TIMEOUT_MS); + return Promise.race([ + resolveAutomergeUrl(handoff, signal), + impatience(RESOLVE_TIMEOUT_MS), + ]); +} + +/** Lookups of a url without heads in flight, by that url. */ +const latestLookups = new Map>(); + +/** + * Resolve `handoff`, sharing the lookup with any already in flight for the + * same url without heads. Every tab's ModuleWatcher asks for each package's + * latest heads, again each time it changes, and none of it is cached, since + * the answer moves; on a busy worker those pile up into one full resolve + * each. One that joins late gets the shared lookup's answer, or its timeout, + * a little sooner than it would have its own. + */ +function resolveHandoff(handoff: URL): Promise { + const [maybeAutomergeUrl] = handoff.href.split("/"); + if ( + !isValidAutomergeUrl(maybeAutomergeUrl) || + parseAutomergeUrl(maybeAutomergeUrl).heads + ) { + return resolveWithin(handoff); + } + let lookup = latestLookups.get(handoff.href); + if (!lookup) { + lookup = resolveWithin(handoff).finally(() => { + latestLookups.delete(handoff.href); + }); + latestLookups.set(handoff.href, lookup); + } + // A redirect, so there's no body to share; each reply gets its own anyway. + return lookup.then((response) => response.clone()); +} + async function respondToHandoff(message: HandoffRequestMessage, handoff: URL) { const { id, cachename, request } = message; let response: Response; try { log(`resolving handoff ${id} for ${handoff}`); - const signal = AbortSignal.timeout(RESOLVE_TIMEOUT_MS); - response = await Promise.race([ - resolveAutomergeUrl(handoff, signal), - impatience(RESOLVE_TIMEOUT_MS), - ]); + response = await resolveHandoff(handoff); } catch (error) { if (error instanceof AbortHandoff) { handoffChannel.postMessage({ diff --git a/core/filesystem/src/module-watcher.ts b/core/filesystem/src/module-watcher.ts index 84981f2e..31799c08 100644 --- a/core/filesystem/src/module-watcher.ts +++ b/core/filesystem/src/module-watcher.ts @@ -88,6 +88,10 @@ type WatchedModule = { baseline: boolean; /** Bumped per lookup, so only the newest one's answer is acted on. */ lookups: number; + /** A change-triggered lookup is in flight. */ + looking: boolean; + /** A change landed while `looking`, so look again once it's back. */ + stale: boolean; }; /** The url of the document `url` names, without heads. */ @@ -371,6 +375,8 @@ export class ModuleWatcher { unwatch: this.tracker.watch(packageUrl, onChange), baseline, lookups: 0, + looking: false, + stale: false, }; this.#watchedModules.set(importName, entry); if (baseline) void this.reloadIfMoved(importName, entry); @@ -403,7 +409,30 @@ export class ModuleWatcher { 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. + */ private async reloadIfMoved(importName: string, entry: WatchedModule) { + if (entry.looking) { + entry.stale = true; + return; + } + entry.looking = true; + try { + do { + entry.stale = false; + await this.lookUpAndReload(importName, entry); + } while (entry.stale && this.#watchedModules.get(importName) === entry); + } finally { + entry.looking = false; + } + } + + private async lookUpAndReload(importName: string, entry: WatchedModule) { let latest: AutomergeUrl | undefined; try { latest = await this.findLatestVersion(importName, entry); diff --git a/core/filesystem/test/module-watcher.test.ts b/core/filesystem/test/module-watcher.test.ts index ee3e8bd7..9d44487d 100644 --- a/core/filesystem/test/module-watcher.test.ts +++ b/core/filesystem/test/module-watcher.test.ts @@ -511,6 +511,68 @@ describe("ModuleWatcher package trackers", () => { } }); + it("looks up once more, not once per change, for changes during a slow lookup", 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 realFindLatest = tracker.findLatest; + // Holds the next headless lookup until released. + let hold: Promise | undefined; + tracker.findLatest = async (url) => { + if (hold && !url.includes("#")) await hold; + return realFindLatest(url); + }; + const imported: string[] = []; + const watcher = new ModuleWatcher( + tab, + { system: settings.url }, + () => {}, + undefined, + async (url) => { + imported.push(url); + return { plugins: [] }; + }, + tracker + ); + await watcher.doneLoading; + lookups.length = 0; + + const { promise, resolve: release } = Promise.withResolvers(); + hold = promise; + folder.change((d: any) => { + d.rev = 1; + }); + watching.get(folder.url)!(); + // Past the debounce, so the first lookup is in flight. + await pause(400); + // Each past the debounce too, so each would start a lookup of its own. + for (let rev = 2; rev <= 6; rev++) { + folder.change((d: any) => { + d.rev = rev; + }); + watching.get(folder.url)!(); + await pause(300); + } + hold = undefined; + release(); + await pause(400); + + // Imports look up their pinned url too; only headless lookups reach the + // automerge worker. + expect(lookups.filter((url) => !url.includes("#"))).toHaveLength(2); + expect(imported.at(-1)).toBe(folder.view(folder.heads()).url); + watcher.dispose(); + } finally { + await Promise.all([folders.shutdown(), tab.shutdown()]).catch(() => {}); + } + }); + it("watches a branches doc's default branch, and looks up through the branches doc", async () => { const folders = new AutomergeRepo({ peerId: "folders" as PeerId }); const tab = new AutomergeRepo({ peerId: "tab" as PeerId });