diff --git a/.changeset/subduction-forty-eight.md b/.changeset/subduction-forty-eight.md new file mode 100644 index 00000000..c5f45603 --- /dev/null +++ b/.changeset/subduction-forty-eight.md @@ -0,0 +1,8 @@ +--- +"@inkandswitch/patchwork-bootloader": patch +"@inkandswitch/patchwork-filesystem": patch +"@inkandswitch/patchwork": patch +"@inkandswitch/patchwork-plugins": patch +--- + +`@automerge/automerge` goes to `3.4.1`, and `@automerge/automerge-repo-network-broadcastchannel` joins the automerge-repo family at `2.6.0-subduction.48`. diff --git a/.changeset/tab-worker-subduction.md b/.changeset/tab-worker-subduction.md new file mode 100644 index 00000000..2960c15c --- /dev/null +++ b/.changeset/tab-worker-subduction.md @@ -0,0 +1,20 @@ +--- +"@inkandswitch/patchwork-bootloader": minor +"@inkandswitch/patchwork": minor +--- + +Every Repo on the origin is its own Subduction node. A tab holds this origin's IndexedDB, keeps its own WebSocket to the sync server, and meets the other tabs over a BroadcastChannel carrying Subduction (`siblingAdapters()` in `@inkandswitch/patchwork-bootloader/siblings`, passed to `new Repo({ subductionAdapters })`). The automerge SharedWorker no longer sits between tabs and storage — it is one more such node, kept only to resolve `automerge:` URLs for the service worker, which can't own a Repo itself. The websocket proxy worker is gone with it. + +Benchmarked against the shared-worker arrangement (`sites/bench`): boot and memory are a wash or better, cross-tab propagation matches, and two shared-worker failures go away — a `find()` racing a sibling's `create()` settled as unavailable, and edits made just before a tab closed were lost, since a storageless tab had nothing to flush to. Each tab flushing its own IndexedDB closes both. + +Keyhive sites use the subduction-backed hive in both the tab and the worker, each talking to the sync server directly. + +`patchwork.sw.subscribeSyncState(documentId, listener)` stays, now a filter over the tab's own Repo's `subduction-remote-heads` event, replaying the server's current heads from `handle.getSyncInfo()` on subscribe; `SyncStateDocMessage` is exported from `@inkandswitch/patchwork`. Removed from `setupServiceWorker()`'s result: `subscribeToRepoChannel`, `getRepoChannel`, `subscribeSyncState`. The `@patchwork/syncstate` BroadcastChannel and the other `SyncState*` message types are gone too; a tab's own Repo has what they carried — `repo.isSubductionConnected()` and the `subduction-connection` event for the link, `repo.connectedSubductionPeerIds()` for which peers are the server, and `patchwork.signerIdentity` for this tab's peer id. `createRepo` in `@inkandswitch/patchwork` takes no arguments. + +`@inkandswitch/patchwork-bootloader` depends on `@automerge/automerge-repo-network-broadcastchannel`, which is also on the importmap. + +That worker is renamed for what it now does: `@inkandswitch/patchwork-bootloader/automerge-worker` is now `@inkandswitch/patchwork-bootloader/automerge-protocol-handler-worker`, emitted as `automerge-protocol-handler-worker.js` (the `workerPath` option on `setupServiceWorker` still overrides it), and `getAutomergeWorker()` is now `getAutomergeProtocolHandlerWorker()`. + +Inter-tab sync is Subduction, not classic automerge sync. `connectSiblings(repo, hive)` is replaced by `siblingAdapters()`, which returns the `subductionAdapters` entries for a Repo rather than mutating one after the fact, so it is passed to the `Repo` constructor. The frames on the siblings BroadcastChannel are Subduction transport frames authenticated by each node's own signer; the keyhive network adapter no longer wraps that channel, since keyhive material reaches siblings the same way it reaches the sync server. The one remaining classic-sync path is the opt-in classic sync server the protocol handler worker connects to on request. + +Subduction's handshake has an initiator and a responder, and a BroadcastChannel is a mesh, so the siblings adapter is passed with `role: "mesh"`, added to the automerge-repo fork's `subductionAdapters` by this repo's pnpm patch: for each pair of peers on the adapter, the one whose peer id sorts lower initiates the handshake and the other accepts. diff --git a/core/bootloader/package.json b/core/bootloader/package.json index 341a89df..6e80524e 100644 --- a/core/bootloader/package.json +++ b/core/bootloader/package.json @@ -26,6 +26,10 @@ "import": "./dist/externals-list.js", "types": "./dist/externals-list.d.ts" }, + "./siblings": { + "import": "./dist/siblings.js", + "types": "./dist/siblings.d.ts" + }, "./storage": { "import": "./dist/storage.js", "types": "./dist/storage.d.ts" @@ -34,9 +38,9 @@ "import": "./dist/service-worker.js", "types": "./dist/service-worker.d.ts" }, - "./automerge-worker": { - "import": "./dist/automerge-worker.js", - "types": "./dist/automerge-worker.d.ts" + "./automerge-protocol-handler-worker": { + "import": "./dist/automerge-protocol-handler-worker.js", + "types": "./dist/automerge-protocol-handler-worker.d.ts" }, "./module-loader": { "import": "./dist/module-loader.js", @@ -56,6 +60,7 @@ "@automerge/automerge": "catalog:", "@automerge/automerge-repo": "catalog:", "@automerge/automerge-repo-keyhive": "catalog:", + "@automerge/automerge-repo-network-broadcastchannel": "catalog:", "@automerge/automerge-repo-network-messagechannel": "catalog:", "@automerge/automerge-repo-network-websocket": "catalog:", "@automerge/automerge-repo-storage-indexeddb": "catalog:", diff --git a/core/bootloader/src/automerge-protocol-handler-worker.ts b/core/bootloader/src/automerge-protocol-handler-worker.ts new file mode 100644 index 00000000..690b01f3 --- /dev/null +++ b/core/bootloader/src/automerge-protocol-handler-worker.ts @@ -0,0 +1,421 @@ +// 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 is a node like any tab's: the same IndexedDB, its own sync-server socket, +// and the siblings channel to the tabs. Resolving requests is its whole job. +// 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, hasHeads } from "@automerge/automerge/slim"; +// eslint-disable-next-line +// @ts-ignore — initSync is a wasm-bindgen runtime helper not in the .d.ts +import { initSync as initSubductionSync } from "@automerge/automerge-subduction/slim"; +import { MemorySigner } from "@automerge/automerge-subduction/slim"; + +import { + Repo, + isValidAutomergeUrl, + parseAutomergeUrl, + stringifyAutomergeUrl, + type AutomergeUrl, + type DocHandle, + type PeerId, +} from "@automerge/automerge-repo/slim"; +import { resolvePath } from "@inkandswitch/patchwork-filesystem"; + +import { IndexedDBWorkerStorageAdapter } from "@automerge/automerge-repo-storage-indexeddb/IndexedDBWorkerStorageAdapter"; +import { WebSocketWorkerClientAdapter } from "@automerge/automerge-repo-network-websocket"; +import { + initializeAutomergeRepoKeyhive, + initKeyhiveWasm, + type AutomergeRepoKeyhive, + type SyncServerSelection, +} from "@automerge/automerge-repo-keyhive"; + +import { DEFAULT_CLASSIC_SYNC_SERVER } from "./sync-config.js"; +import { siblingAdapters } from "./siblings.js"; +import { keyhiveStorageName, storagePrefix } from "./storage.js"; +import { startWorkerControl } from "./worker-control.js"; +import { + HANDOFF_CHANNEL, + type HandoffCachedMessage, + type HandoffOnlineMessage, + type HandoffAbortMessage, + type HandoffRequestMessage, + type HandoffResponseMessage, +} from "./types.js"; + +declare const __SYNC_SERVER__: { + url: string; + keyhive?: SyncServerSelection; +}; + +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, +}); +const log = control.log; + +// ── The repo ─────────────────────────────────────────────────────────── + +let repoPromise: Promise | null = null; + +function getRepo(): Promise { + if (!repoPromise) { + repoPromise = buildRepo(); + // Don't cache a rejection (e.g. the wasm fetch failed): clear the slot so + // the next caller retries from scratch. + repoPromise.catch(() => { + repoPromise = null; + }); + } + return repoPromise; +} + +async function buildRepo(): Promise { + log("fetching wasm"); + const [automergeWasm, subductionWasm] = await Promise.all([ + fetch("/automerge.wasm").then((r) => r.arrayBuffer()), + fetch("/subduction.wasm").then((r) => r.arrayBuffer()), + ]); + initSubductionSync(new Uint8Array(subductionWasm)); + await initializeWasm(new Uint8Array(automergeWasm)); + log("wasm initialized"); + + const { repo, hive } = syncServer.keyhive + ? await buildKeyhiveRepo(syncServer.keyhive) + : { repo: buildPlainRepo() }; + + (self as any).repo = repo; + if (hive) (self as any).hive = hive; + return repo; +} + +function buildPlainRepo(): Repo { + return new Repo({ + signer: new MemorySigner(), + storage: new IndexedDBWorkerStorageAdapter(), + peerId: + `${storagePrefix}-resolver-${Math.random().toString(36).slice(2)}` as PeerId, + subductionWebsocketEndpoints: [syncServer.url], + subductionAdapters: siblingAdapters(), + enableRemoteHeadsGossiping: true, + }); +} + +async function buildKeyhiveRepo( + keyhiveSyncServer: SyncServerSelection +): Promise<{ repo: Repo; hive: AutomergeRepoKeyhive }> { + initKeyhiveWasm(); + const { hive, repo } = await initializeAutomergeRepoKeyhive({ + createRepo: (config) => new Repo(config), + storage: new IndexedDBWorkerStorageAdapter(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 IndexedDBWorkerStorageAdapter(), + subductionWebsocketEndpoints: [syncServer.url], + subductionAdapters: siblingAdapters(), + 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 !== "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(); + }); +} + +/** + * 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(); + 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) { + const folder = await repo.find(maybeAutomergeUrl, { signal }); + const url = stringifyAutomergeUrl({ documentId, heads: folder.heads() }); + const location = `/${encodeURIComponent(url)}${path.length ? `/${path.join("/")}` : ""}`; + return Response.redirect(location, 307); + } + + const baseHandle = await repo.find(stringifyAutomergeUrl({ documentId }), { + signal, + }); + if (!(await waitForHeads(baseHandle, hexHeads ?? [], signal))) { + throw new AbortHandoff( + `heads not found for ${maybeAutomergeUrl} within ${RESOLVE_TIMEOUT_MS}ms` + ); + } + + const resolved = await resolvePath( + repo, + baseHandle.view(heads), + path.map(decodeURIComponent) + ); + if (!resolved) { + throw new Error( + `couldn't resolve ${path.join("/")} in folder at ${maybeAutomergeUrl}` + ); + } + + 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 + ) + ); +} + +async function handleHandoffRequest(message: HandoffRequestMessage) { + const { id, cachename, 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; + } + + 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)); + } +} + +handoffChannel.addEventListener("message", (event) => { + if (event.data?.type === "request") { + void handleHandoffRequest(event.data as HandoffRequestMessage); + } +}); + +// Announce ourselves so the service worker can re-broadcast handoff requests +// sent while we were booting. +handoffChannel.postMessage({ type: "online" } satisfies HandoffOnlineMessage); diff --git a/core/bootloader/src/automerge-worker.ts b/core/bootloader/src/automerge-worker.ts deleted file mode 100644 index 01aa2013..00000000 --- a/core/bootloader/src/automerge-worker.ts +++ /dev/null @@ -1,1032 +0,0 @@ -// The automerge repo for a patchwork site, in a SharedWorker: one instance -// serves every tab and lives as long as any tab does. -// -// The service worker holds no repo. When it 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, hasHeads } from "@automerge/automerge/slim"; -// eslint-disable-next-line -// @ts-ignore — initSync is a wasm-bindgen runtime helper not in the .d.ts -import { initSync as initSubductionSync } from "@automerge/automerge-subduction/slim"; -import { WebCryptoSigner } from "@automerge/automerge-subduction/slim"; -import { makePortProvider } from "@automerge/automerge-repo/worker-port"; - -import { - Repo, - WorkerWebSocketEndpoint, - isValidAutomergeUrl, - parseAutomergeUrl, - stringifyAutomergeUrl, - type AutomergeUrl, - type DocHandle, - type DocumentId, - type PeerId, - type UrlHeads, -} from "@automerge/automerge-repo/slim"; -import { resolvePath } from "@inkandswitch/patchwork-filesystem"; - -import { IndexedDBWorkerStorageAdapter } from "@automerge/automerge-repo-storage-indexeddb/IndexedDBWorkerStorageAdapter"; -import { MessageChannelNetworkAdapter } from "@automerge/automerge-repo-network-messagechannel"; -import { WebSocketWorkerClientAdapter } from "@automerge/automerge-repo-network-websocket"; -import { - initializeAutomergeRepoKeyhive, - initKeyhiveWasm, - type AutomergeRepoKeyhive, - type SyncServerSelection, -} from "@automerge/automerge-repo-keyhive"; - -import { DEFAULT_CLASSIC_SYNC_SERVER } from "./sync-config.js"; -import { keyhiveStorageName, storagePrefix } from "./storage.js"; -import { - HANDOFF_CHANNEL, - SYNCSTATE_CHANNEL, - type HandoffCachedMessage, - type HandoffOnlineMessage, - type HandoffRequest, - type HandoffAbortMessage, - type HandoffRequestMessage, - type HandoffResponseMessage, - type SyncStateBroadcast, - type SyncStateDocMessage, - type SyncStateRequestMessage, -} from "./types.js"; - -declare const __SYNC_SERVER__: { - url: string; - keyhive?: SyncServerSelection; -}; - -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]; - -// A fresh instance means a new repo peerId and cold in-memory state, so a tab -// seeing a changed id knows to re-subscribe. Sent in `hello` and every `pong`. -const WORKER_INSTANCE_ID = Math.random().toString(36).slice(2); -const WORKER_BOOT_TIME = Date.now(); - -type Identity = { peerId: string; verifyingKey: string }; - -// `debug` reads localStorage, which a SharedWorker doesn't have, so debugging is -// toggled by a control message from a tab instead. -let debugging = false; -function log(...args: any[]) { - if (debugging) console.log("[automerge-worker]", ...args); -} - -// ── Console forwarding ───────────────────────────────────────────────── -// The SharedWorker's own console is buried in chrome://inspect, so mirror -// everything over each connected tab's control port. - -const controlPorts = new Set(); -// Logs emitted before any tab connects (wasm boot) would otherwise be lost. -const preConnectBuffer: Array<{ level: string; args: string[] }> = []; -const MAX_BUFFER = 200; - -function serializeArg(arg: any): string { - if (typeof arg === "string") return arg; - if (arg instanceof Error) return arg.stack || `${arg.name}: ${arg.message}`; - try { - return JSON.stringify(arg); - } catch { - return String(arg); - } -} - -function postToPort(port: MessagePort, message: unknown): void { - try { - port.postMessage(message); - } catch (error) { - console.warn(`sending failed`, error); - } -} - -function forwardToMainThread(level: string, rawArgs: any[]) { - const args = rawArgs.map(serializeArg); - if (!controlPorts.size) { - if (preConnectBuffer.length < MAX_BUFFER) - preConnectBuffer.push({ level, args }); - return; - } - for (const port of controlPorts) { - postToPort(port, { type: "console", level, args }); - } -} - -for (const level of ["log", "info", "warn", "error", "debug"] as const) { - const original = console[level].bind(console); - console[level] = (...args: any[]) => { - original(...args); - forwardToMainThread(level, args); - }; -} - -self.addEventListener("error", (event) => { - const e = event as ErrorEvent; - forwardToMainThread("error", [ - `uncaught error: ${e.message}`, - e.error instanceof Error ? e.error.stack : undefined, - ]); -}); - -self.addEventListener("unhandledrejection", (event) => { - const reason = (event as PromiseRejectionEvent).reason; - forwardToMainThread("error", [ - "unhandled rejection:", - reason instanceof Error ? reason.stack || reason.message : reason, - ]); -}); - -console.warn( - `[lifecycle] automerge SharedWorker started (instance ${WORKER_INSTANCE_ID})` -); - -const WATCHDOG_TICK_MS = 5_000; -let watchdogLast = Date.now(); -setInterval(() => { - const now = Date.now(); - const gap = now - watchdogLast; - watchdogLast = now; - if (gap > WATCHDOG_TICK_MS * 2) { - console.warn( - `[lifecycle] watchdog timer gap ~${Math.round(gap / 1000)}s ` + - `(expected every ${WATCHDOG_TICK_MS / 1000}s)` - ); - } -}, WATCHDOG_TICK_MS); - -// ── Per-tab sync-state subscriptions ─────────────────────────────────── -// A tab's control port subscribes to the documents it cares about and we push -// only those docs' heads down that port, so tab A never sees tab B's docs. A -// port's whole subscription set is dropped when it closes, so there's nothing -// to reference-count or time out. - -const syncWatchers = new Map>(); - -// Set once the repo's snapshot exists, so a `sync-sub` arriving during boot can -// be replayed the doc's current heads as soon as it does. -let replaySyncForPort: - ((documentId: string, port: MessagePort) => void) | null = null; - -function syncSubscribe(port: MessagePort, documentId: string): void { - let docs = syncWatchers.get(port); - if (!docs) syncWatchers.set(port, (docs = new Set())); - if (docs.has(documentId)) return; - docs.add(documentId); - replaySyncForPort?.(documentId, port); -} - -function syncUnsubscribe(port: MessagePort, documentId: string): void { - syncWatchers.get(port)?.delete(documentId); -} - -function pushSyncState(message: SyncStateDocMessage): void { - for (const [port, docs] of syncWatchers) { - if (docs.has(message.documentId)) postToPort(port, message); - } -} - -const subductionPortProvider = makePortProvider(); - -// Memoized so a construction retry reuses the endpoint instead of leaking one -// per attempt. -let subductionEndpoints: WorkerWebSocketEndpoint[] | null = null; -function getSubductionEndpoints(): WorkerWebSocketEndpoint[] { - return (subductionEndpoints ??= [ - new WorkerWebSocketEndpoint(syncServer.url, { - worker: subductionPortProvider.source, - }), - ]); -} - -type RepoHive = { repo: Repo; hive?: AutomergeRepoKeyhive }; -type BuiltRepo = RepoHive & { identity?: Identity }; - -let repoHivePromise: Promise | null = null; - -function getRepoHive(): Promise { - if (!repoHivePromise) { - repoHivePromise = setUpRepoHive(); - // Don't permanently cache a rejection (e.g. the wasm fetch failed) — clear - // the slot so the next caller retries from scratch. - repoHivePromise.catch(() => { - repoHivePromise = null; - }); - } - return repoHivePromise; -} - -async function setUpRepoHive(): Promise { - log("fetching wasm"); - const [automergeWasm, subductionWasm] = await Promise.all([ - fetch("/automerge.wasm").then((r) => r.arrayBuffer()), - fetch("/subduction.wasm").then((r) => r.arrayBuffer()), - ]); - initSubductionSync(new Uint8Array(subductionWasm)); - await initializeWasm(new Uint8Array(automergeWasm)); - log("wasm initialized"); - - const built: BuiltRepo = syncServer.keyhive - ? await buildKeyhiveRepo(syncServer.keyhive) - : await buildPlainRepo(); - - (self as any).repo = built.repo; - if (built.hive) (self as any).hive = built.hive; - if (built.identity) (self as any).syncIdentity = built.identity; - - setUpSyncStateBroadcast(built.repo, built.identity); - - // Deliberately not awaited: the network subsystem starts with only the - // subduction adapter, and the MessageChannel adapter is added later by - // connectPort, which itself awaits getRepoHive. Blocking here would deadlock - // that path and starve the handoff handler. - built.repo.networkSubsystem - .whenReady() - .then(() => log("repo network subsystem ready")); - - return { repo: built.repo, hive: built.hive }; -} - -async function buildPlainRepo(): Promise { - const signer = await WebCryptoSigner.setup(); - const identity = { - peerId: signer.peerId().toString(), - verifyingKey: ( - signer.verifyingKey() as Uint8Array & { - toHex(): string; - } - ).toHex(), - }; - const repo = new Repo({ - storage: new IndexedDBWorkerStorageAdapter(), - signer, - peerId: `automerge-worker-${Math.random().toString(36).slice(2)}` as PeerId, - async sharePolicy(peerId) { - return peerId.includes("storage-server"); - }, - enableRemoteHeadsGossiping: true, - subductionWebsocketEndpoints: getSubductionEndpoints(), - }); - console.log("[patchwork] shared-worker subduction identity:", identity); - return { repo, identity }; -} - -async function buildKeyhiveRepo( - keyhiveSyncServer: SyncServerSelection -): Promise { - initKeyhiveWasm(); - const { hive, repo } = await initializeAutomergeRepoKeyhive({ - createRepo: (config) => new Repo(config), - storage: new IndexedDBWorkerStorageAdapter(keyhiveStorageName), - peerIdSuffix: - `${storagePrefix}-worker` + 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 IndexedDBWorkerStorageAdapter(), - subductionWebsocketEndpoints: getSubductionEndpoints(), - 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 getRepoHive(); - 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; -} - -// ── Sync-state broadcast ─────────────────────────────────────────────── -// Only this worker is connected to the sync server, so it's the only place that -// learns the server's heads ("subduction-remote-heads", keyed by each Subduction -// peer's verifying-key storageId) and whether the link is up -// ("subduction-connection"). Global signals go out on SYNCSTATE_CHANNEL so any -// tab can render a sync indicator; per-document heads are addressed to -// subscribers instead (see pushSyncState). - -const RESYNC_GRACE_MS = 8_000; // must be stably diverged this long first -const RESYNC_INITIAL_DELAY_MS = 5_000; -const RESYNC_MAX_DELAY_MS = 60_000; -const RESYNC_REVIEW_INTERVAL_MS = 5_000; -const OWN_HANDLE_SCAN_INTERVAL_MS = 3_000; - -type PeerHeads = { heads: string[]; timestamp: number }; -type ResyncEntry = { - serverSig: string; - since: number; - delay: number; - lastResyncAt: number; -}; - -type SyncState = { - repo: Repo; - channel: BroadcastChannel; - identity?: Identity; - /** documentId -> storageId (verifying key) -> last-known heads */ - snapshot: Map>; - connected: boolean; - /** Directly-connected sync-server peer ids, used to judge "synced". */ - serverPeerIds: string[]; - tracked: Set; - resync: Map; -}; - -type OwnHandle = { - documentId: string; - heads: () => string[]; - on: (ev: "heads-changed", cb: () => void) => void; -}; - -let syncStateWired = false; - -function setUpSyncStateBroadcast(repo: Repo, identity?: Identity): void { - if (syncStateWired) return; - syncStateWired = true; - - const state: SyncState = { - repo, - channel: new BroadcastChannel(SYNCSTATE_CHANNEL), - identity, - snapshot: new Map(), - connected: repo.isSubductionConnected(), - serverPeerIds: [], - tracked: new Set(), - resync: new Map(), - }; - - postWhoAmI(state); - - replaySyncForPort = (documentId, port) => replayDoc(state, documentId, port); - for (const [port, docs] of syncWatchers) { - for (const documentId of docs) replayDoc(state, documentId, port); - } - - repo.on( - "subduction-remote-heads", - ({ documentId, storageId, heads, timestamp }) => { - recordHeads(state, documentId, storageId, [...heads], timestamp); - // A doc the server reported is one we hold, so advertise our heads for it - // too. Only this doc: a full scan per event is O(all handles) and goes - // quadratic during sync bursts. The tick covers general discovery. - const handle = repo.handles[documentId as DocumentId]; - if (handle) trackOwnHandle(state, handle as never); - reviewResync(state, documentId); - } - ); - - repo.on("subduction-connection", ({ connected }) => { - state.connected = connected; - postConnection(state); - if (connected) void refreshServerPeers(state); - }); - - // A BroadcastChannel never receives its own posts, so this only sees tabs' - // requests. Only the global signals are replayed; a late tab gets per-doc - // heads by subscribing. - state.channel.addEventListener("message", (event: MessageEvent) => { - if ((event.data as SyncStateRequestMessage)?.type !== "request") return; - postWhoAmI(state); - postConnection(state); - }); - - void refreshServerPeers(state); - scanOwnHandles(state); - - if (!identity) return; - // Subduction-pushed docs don't surface via the "document" event, so discover - // them by re-scanning repo.handles on a tick. - setInterval(() => scanOwnHandles(state), OWN_HANDLE_SCAN_INTERVAL_MS); - // The "stuck" case is precisely when no head events are firing, so the - // grace/backoff timers can only advance on a tick. - setInterval(() => reviewAllResync(state), RESYNC_REVIEW_INTERVAL_MS); -} - -function postWhoAmI(state: SyncState): void { - if (!state.identity) return; - state.channel.postMessage({ - type: "whoami", - peerId: state.identity.peerId, - verifyingKey: state.identity.verifyingKey, - } satisfies SyncStateBroadcast); -} - -function postConnection(state: SyncState): void { - state.channel.postMessage({ - type: "connection", - connected: state.connected, - serverPeerIds: state.serverPeerIds, - } satisfies SyncStateBroadcast); -} - -function recordHeads( - state: SyncState, - documentId: string, - storageId: string, - heads: string[], - timestamp: number -): void { - let byStorage = state.snapshot.get(documentId); - if (!byStorage) state.snapshot.set(documentId, (byStorage = new Map())); - byStorage.set(storageId, { heads, timestamp }); - pushSyncState({ - type: "sync-state", - documentId, - storageId, - heads, - timestamp, - }); -} - -function replayDoc( - state: SyncState, - documentId: string, - port: MessagePort -): void { - const byStorage = state.snapshot.get(documentId); - if (!byStorage) return; - for (const [storageId, { heads, timestamp }] of byStorage) { - postToPort(port, { - type: "sync-state", - documentId, - storageId, - heads, - timestamp, - } satisfies SyncStateDocMessage); - } -} - -/** The peer list is empty until the handshake finishes, so retry briefly. */ -async function refreshServerPeers(state: SyncState): Promise { - for (let attempt = 0; attempt < 6; attempt++) { - try { - const ids = await state.repo.connectedSubductionPeerIds(); - if (ids.length > 0) { - state.serverPeerIds = ids; - postConnection(state); - return; - } - } catch { - // No subduction source yet. - } - await new Promise((r) => setTimeout(r, 500)); - } -} - -// Advertise this worker's own heads for every doc it holds, so the worker hop is -// visible on every document. No-op on the keyhive path, which has no identity. - -function broadcastOwnHeads(state: SyncState, handle: OwnHandle): void { - if (!state.identity) return; - let heads: string[]; - try { - heads = [...handle.heads()]; - } catch { - return; // handle not ready - } - recordHeads( - state, - handle.documentId, - state.identity.peerId, - heads, - Date.now() - ); - reviewResync(state, handle.documentId); -} - -function trackOwnHandle(state: SyncState, handle: OwnHandle): void { - if (!state.identity || state.tracked.has(handle.documentId)) return; - state.tracked.add(handle.documentId); - handle.on("heads-changed", () => broadcastOwnHeads(state, handle)); - broadcastOwnHeads(state, handle); -} - -function scanOwnHandles(state: SyncState): void { - if (!state.identity) return; - for (const handle of Object.values(state.repo.handles)) { - trackOwnHandle(state, handle as never); - } -} - -function serverHeadsFor(state: SyncState, documentId: string): UrlHeads { - const byStorage = state.snapshot.get(documentId); - if (!byStorage) return [] as unknown as UrlHeads; - const heads = new Set(); - for (const [storageId, entry] of byStorage) { - if (state.serverPeerIds.includes(storageId)) { - for (const head of entry.heads) heads.add(head); - } - } - return [...heads] as UrlHeads; -} - -function reviewResync(state: SyncState, documentId: string): void { - if (!state.identity || !state.connected) { - state.resync.delete(documentId); - return; - } - const handle = state.repo.handles[documentId as DocumentId]; - if (!handle) return; - - const serverHeads = serverHeadsFor(state, documentId); - if (serverHeads.length === 0) { - state.resync.delete(documentId); // nothing to compare against - return; - } - - // The server advertises subduction sedimentree heads (loose-commit and - // fragment-boundary commit ids), which are NOT the Automerge frontier, so - // never compare them to handle.heads() for equality. Ask instead whether we - // already hold every commit the server advertises. - let haveAll: boolean; - try { - haveAll = handle.containsHeads(serverHeads); - } catch { - return; // doc not ready, or an undecodable head - } - if (haveAll) { - state.resync.delete(documentId); - return; - } - - // Behind. Key the grace timer on the server heads alone, so your own edits - // churning don't keep resetting it. - const serverSig = [...serverHeads].sort().join(","); - const now = Date.now(); - const prev = state.resync.get(documentId); - if (!prev || prev.serverSig !== serverSig) { - // First sighting, or the server made progress: restart the clock. - state.resync.set(documentId, { - serverSig, - since: now, - delay: RESYNC_INITIAL_DELAY_MS, - lastResyncAt: 0, - }); - return; - } - if (now - prev.since < RESYNC_GRACE_MS) return; - if (now - prev.lastResyncAt < prev.delay) return; - - log("re-syncing behind doc", documentId); - try { - state.repo.resyncSubduction(documentId as DocumentId); - } catch (e) { - log("resyncSubduction failed", e); - } - prev.lastResyncAt = now; - prev.delay = Math.min(prev.delay * 2, RESYNC_MAX_DELAY_MS); -} - -function reviewAllResync(state: SyncState): void { - if (!state.identity) return; - for (const documentId of state.snapshot.keys()) - reviewResync(state, documentId); - for (const id of [...state.resync.keys()]) { - if (!state.snapshot.has(id)) state.resync.delete(id); - } -} - -// ── Tab connections ──────────────────────────────────────────────────── -// Each tab connects with a control port and opens repo MessageChannel ports -// through it. `adapter` is what was registered with the network subsystem (the -// MessageChannel adapter, or the keyhive wrapper around it); `mcAdapter` is -// always the underlying MessageChannel adapter, so the port itself can be -// disconnected. - -type RepoChannel = { - adapter: { disconnect(): void }; - mcAdapter: MessageChannelNetworkAdapter; - port: MessagePort; -}; -type Connection = { channels: Set }; - -function dropRepoChannel(repo: Repo, channel: RepoChannel) { - // removeNetworkAdapter pulls the adapter out of networkSubsystem.adapters and - // calls disconnect(), which for the MessageChannel adapter emits the - // close/peer-disconnected events that clear #adaptersByPeer. - try { - repo.networkSubsystem.removeNetworkAdapter(channel.adapter as any); - } catch (err) { - console.error("removeNetworkAdapter failed", err); - } - // On the keyhive path the registered adapter is a wrapper, so make sure the - // underlying port is disconnected and closed too. - try { - channel.mcAdapter.disconnect(); - } catch {} - try { - channel.port.close(); - } catch {} -} - -async function dropConnection(connection: Connection) { - if (!connection.channels.size || !repoHivePromise) return; - const { repo } = await getRepoHive(); - log(`tab gone — removing ${connection.channels.size} network adapter(s)`); - for (const channel of connection.channels) dropRepoChannel(repo, channel); - connection.channels.clear(); -} - -async function connectPort(port: MessagePort, connection: Connection) { - const { hive, repo } = await getRepoHive(); - const mcAdapter = new MessageChannelNetworkAdapter(port, { - useWeakRef: true, - }); - - if (!hive) { - repo.networkSubsystem.addNetworkAdapter(mcAdapter); - connection.channels.add({ adapter: mcAdapter, mcAdapter, port }); - return; - } - - const adapter = hive.createKeyhiveNetworkAdapter(mcAdapter, { - onlyShareWithSyncServer: false, - periodicallyRequestSync: false, - syncRequestInterval: 2000, - }); - - adapter.on("message", (msg: any) => { - if (msg.type !== "sync" && msg.type !== "request") return; - if (!msg.documentId) return; - const handle = repo.handles[msg.documentId]; - if (handle && handle.state !== "unavailable") return; - repo.findWithProgress(`automerge:${msg.documentId}` as AutomergeUrl); - repo.shareConfigChanged(); - }); - - (adapter as any).on("ingest-remote", () => { - hive.notifySameAgentKeyhiveChange(); - (hive.networkAdapter as any).syncKeyhive?.(); - repo.shareConfigChanged(); - }); - - repo.networkSubsystem.addNetworkAdapter(adapter); - connection.channels.add({ adapter, mcAdapter, port }); -} - -function handleControlMessage( - event: MessageEvent, - controlPort: MessagePort, - connection: Connection -) { - const data = event.data; - - switch (data?.type) { - case "port": { - log("received repo channel"); - const [repoPort] = event.ports; - connectPort(repoPort, connection).then( - () => controlPort.postMessage({ type: "port-ready", id: data.id }), - (err) => { - console.error("connectPort failed", err); - // Tell the tab so it doesn't hang until its timeout. - controlPort.postMessage({ - type: "port-failed", - id: data.id, - error: String(err), - }); - } - ); - return; - } - - case "sync-sub": - if (typeof data.documentId === "string") { - syncSubscribe(controlPort, data.documentId); - } - return; - - case "sync-unsub": - if (typeof data.documentId === "string") { - syncUnsubscribe(controlPort, data.documentId); - } - return; - - case "debug": - debugging = data.debug; - log("automerge worker debugging enabled"); - return; - - case "connect-classic-sync": { - 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(); - } - ); - return; - } - - case "ping": - controlPort.postMessage({ - type: "pong", - id: data.id, - instanceId: WORKER_INSTANCE_ID, - }); - return; - } -} - -self.addEventListener("connect", (event) => { - const controlPort = (event as MessageEvent).ports[0]; - const connection: Connection = { channels: new Set() }; - - controlPort.addEventListener("message", (messageEvent) => { - handleControlMessage(messageEvent as MessageEvent, controlPort, connection); - }); - - // The tab side runs donatePort; the messages are channel-tagged so they - // coexist with the control protocol above. - subductionPortProvider.attachClient(controlPort); - - // Fires when the owning page is destroyed. Browsers without the close event - // fall back to the adapters' lazy useWeakRef cleanup. - controlPort.addEventListener("close", () => { - controlPorts.delete(controlPort); - syncWatchers.delete(controlPort); - void dropConnection(connection); - }); - - controlPort.start(); - - controlPort.postMessage({ - type: "hello", - instanceId: WORKER_INSTANCE_ID, - bootTime: WORKER_BOOT_TIME, - }); - - controlPorts.add(controlPort); - for (const { level, args } of preConnectBuffer.splice(0)) { - postToPort(controlPort, { type: "console", level, args }); - } -}); - -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(); - }); -} - -/** - * 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 getRepoHive(); - 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) { - const folder = await repo.find(maybeAutomergeUrl, { signal }); - const url = stringifyAutomergeUrl({ documentId, heads: folder.heads() }); - const location = `/${encodeURIComponent(url)}${path.length ? `/${path.join("/")}` : ""}`; - return Response.redirect(location, 307); - } - - const baseHandle = await repo.find(stringifyAutomergeUrl({ documentId }), { - signal, - }); - if (!(await waitForHeads(baseHandle, hexHeads ?? [], signal))) { - throw new AbortHandoff( - `heads not found for ${maybeAutomergeUrl} within ${RESOLVE_TIMEOUT_MS}ms` - ); - } - - const resolved = await resolvePath( - repo, - baseHandle.view(heads), - path.map(decodeURIComponent) - ); - if (!resolved) { - throw new Error( - `couldn't resolve ${path.join("/")} in folder at ${maybeAutomergeUrl}` - ); - } - - 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 - ) - ); -} - -async function handleHandoffRequest(message: HandoffRequestMessage) { - const { id, cachename, 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; - } - - 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)); - } -} - -handoffChannel.addEventListener("message", (event) => { - if (event.data?.type === "request") { - void handleHandoffRequest(event.data as HandoffRequestMessage); - } -}); - -// Announce ourselves so the service worker can re-broadcast handoff requests -// sent while we were booting. -handoffChannel.postMessage({ type: "online" } satisfies HandoffOnlineMessage); diff --git a/core/bootloader/src/externals-list.ts b/core/bootloader/src/externals-list.ts index 38e5bec9..7597f675 100644 --- a/core/bootloader/src/externals-list.ts +++ b/core/bootloader/src/externals-list.ts @@ -6,11 +6,7 @@ const externals = [ "@automerge/automerge/slim", "@automerge/automerge-repo", "@automerge/automerge-repo/slim", - // Port-donation plumbing for WorkerWebSocketEndpoint: tabs spawn the shared - // proxy entry and donate its port to the automerge worker (Chrome can't - // spawn workers from inside a SharedWorker). See setup.ts/automerge-worker.ts. - "@automerge/automerge-repo/worker-port", - "@automerge/automerge-repo/subduction-websocket-worker-shared", + "@automerge/automerge-repo-network-broadcastchannel", "@automerge/automerge-repo-network-messagechannel", "@automerge/automerge-repo-network-websocket", "@automerge/automerge-repo-storage-indexeddb", diff --git a/core/bootloader/src/setup.ts b/core/bootloader/src/setup.ts index 734ff0c6..2d512d36 100644 --- a/core/bootloader/src/setup.ts +++ b/core/bootloader/src/setup.ts @@ -1,8 +1,6 @@ import type { - ServiceWorkerRepoChannelListener, SetupServiceWorkerOptions, SetupServiceWorkerResult, - SyncStateDocMessage, } from "./types.js"; import { readClassicSyncServer, @@ -10,23 +8,16 @@ import { } from "./sync-config.js"; import debug from "debug"; import { - donatePort, - isWorkerErrorMessage, -} from "@automerge/automerge-repo/worker-port"; + forwardWorkerConsole, + lifecycleLog, + sharedWorkerHandle, +} from "./shared-worker-lifecycle.js"; + +export { lifecycleLog }; const serviceWorkerDebugging = debug.enabled("patchwork:serviceworker"); const workerDebugging = debug.enabled("patchwork:automergeworker"); -export const lifecycleLog = debug("patchwork:lifecycle"); - -function describeErrorEvent(event: Event): string { - const error = event as ErrorEvent; - const where = error.filename - ? ` (${error.filename}:${error.lineno}:${error.colno})` - : ""; - return `${error.message || String(event)}${where}`; -} - // The version is cleared on every boot, so the steady state is // DEFAULT_CACHE_NAME. bumpServiceWorkerCache is a dev escape hatch: it moves // the worker to a throwaway cache now, and the next boot both reverts the name @@ -65,348 +56,28 @@ function installServiceWorkerLogForwarding(): void { }); } -// The automerge repo lives in a SharedWorker. one instance serves every -// tab. Browsers might kill a SharedWorker under memory pressure, so we -// heartbeat it and rebuild everything if it dies. - -let automergeWorkerPath = "/automerge-worker.js"; -let automergeWorker: SharedWorker | undefined; -// A repo port opened against instance N is stale once instance N+1 exists — its -// channel ends in a dead worker — so deliveries are guarded on generation. -let workerGeneration = 0; -let disposeWorkerDeathDetection: (() => void) | undefined; -const repoChannelListeners = new Set(); -let recoveringWorker = false; -let lastWorkerRecoveryAt = 0; -// Below this spacing, skip: if the fresh worker is dead too, its own heartbeat -// re-triggers recovery later rather than spinning in a tight loop. -const RECOVERY_MIN_INTERVAL_MS = 15_000; -let nextRepoChannelId = 0; - -// Chrome can't spawn workers inside a SharedWorker, so each tab offers this -// proxy's port to the automerge worker, which requests one via its port -// provider. Being a SharedWorker itself, the proxy — and the donated -// worker↔worker port — outlives the donor tab. -const SUBDUCTION_IO_WORKER_URL = - "/packages/@automerge/automerge-repo/subduction-websocket-worker-shared.js"; - -export function getAutomergeWorker(): SharedWorker { - if (automergeWorker) return automergeWorker; - - workerGeneration++; - const worker = new SharedWorker(automergeWorkerPath, { - name: "patchwork-automerge", - type: "module", - }); - automergeWorker = worker; - - // Fires when a message can't be structured-deserialized. Silent otherwise: - // the message is dropped, which looks identical to a worker that never - // replied. - worker.port.addEventListener("messageerror", (event) => { - console.error( - "[automerge-worker] undeserializable message from worker:", - event - ); - }); - // Control replies come back on this port, and we listen with - // addEventListener rather than onmessage, so it needs start(). - worker.port.start(); - worker.port.addEventListener("message", handleWorkerMessage); - worker.port.postMessage({ type: "debug", debug: workerDebugging }); - - donatePort(worker.port, createSubductionIoPort); - disposeWorkerDeathDetection = installWorkerDeathDetection(worker); - return worker; -} - -function handleWorkerMessage(event: MessageEvent): void { - const data = event.data; - - if (data?.type === "sync-state") { - dispatchSyncState(data as SyncStateDocMessage); - return; - } - - // Crash/skew reports relayed from the subduction io proxy (e.g. a protocol - // mismatch from a stale SW-cached worker chunk). These otherwise only exist - // in chrome://inspect. - if (isWorkerErrorMessage(data)) { - console.error("[subduction-io]", data); - return; - } - - if (data?.type !== "console") return; - const { level, args } = data; - if ( - !lifecycleLog.enabled && - typeof args?.[0] === "string" && - args[0].includes("[lifecycle]") - ) { - return; - } - const write = (console as any)[level] ?? console.log; - // The worker's logs carry %c directives in args[0] with CSS in the following - // args, so the tag has to go inside the format string or the CSS prints raw. - if (typeof args[0] === "string") { - write(`[automerge-worker] ${args[0]}`, ...args.slice(1)); - } else { - write("[automerge-worker]", ...args); - } -} - -function createSubductionIoPort(): MessagePort { - const io = new SharedWorker(SUBDUCTION_IO_WORKER_URL, { - type: "module", - name: "subduction-websocket", - }); - // This worker carries the websocket to the sync server, so a load failure - // stops sync with no other symptom. - io.addEventListener("error", (event) => { - console.error( - `[subduction-io] failed to load/run ${SUBDUCTION_IO_WORKER_URL}:`, - describeErrorEvent(event) - ); - }); - io.port.addEventListener("messageerror", (event) => { - console.error("[subduction-io] undeserializable message:", event); - }); - return io.port; -} - -/** - * Build a replacement worker and re-wire everything a live tab holds against - * it: console forwarding and port donation (both re-done by - * getAutomergeWorker), the per-doc sync-state subscriptions, and every - * subscriber's repo port. The new instance boots with cold state. - */ -async function recoverAutomergeWorker( - reason: string, - deadWorker: SharedWorker -): Promise { - if (deadWorker !== automergeWorker) return; - if (recoveringWorker) return; - const now = Date.now(); - if (now - lastWorkerRecoveryAt < RECOVERY_MIN_INTERVAL_MS) return; - recoveringWorker = true; - lastWorkerRecoveryAt = now; - lifecycleLog("recreating the automerge SharedWorker (%s)", reason); - - try { - disposeWorkerDeathDetection?.(); - disposeWorkerDeathDetection = undefined; - automergeWorker = undefined; - try { - deadWorker.port.close(); - } catch {} - - const fresh = getAutomergeWorker(); - for (const documentId of syncStateListeners.keys()) { - fresh.port.postMessage({ type: "sync-sub", documentId }); - } - for (const listener of repoChannelListeners) { - try { - const generation = workerGeneration; - const port = await openRepoChannel(); - // Replaced again while we waited — the newer recovery re-delivers. - if (generation !== workerGeneration) break; - await listener(port); - } catch (err) { - console.error( - "failed to re-wire a repo channel after worker recovery", - err - ); - } - } - } finally { - recoveringWorker = false; - } -} - -// A silent port is not proof of death: the worker may still be evaluating its -// module graph, or be busy with wasm/sync work. In both cases every queued -// message — including the repo ports the network adapters ride on — is -// delivered once it catches up, and tearing the port down would lose them. So -// silence only starts a non-destructive probe: a second connection to the same -// instance. Only if the probe gets a `hello` while this port stays silent do we -// know the instance is alive but our port is stranded, and recover. -const HEARTBEAT_MS = 5_000; -const HEARTBEAT_TIMEOUT_MS = 25_000; -// An idle worker hellos within milliseconds of connecting, so before first -// contact the budget is tighter — probing early rescues stranded boots fast. -const FIRST_CONTACT_TIMEOUT_MS = 4_000; -// After a slow boot both connections hello at roughly the same moment and -// cross-port delivery order isn't guaranteed, so give the suspect this long to -// also speak before concluding it's stranded. -const PROBE_GRACE_MS = 500; - -function installWorkerDeathDetection(worker: SharedWorker): () => void { - let instanceId: string | undefined; - let lastHeardAt = Date.now(); - let warnedUnresponsive = false; - let warnedSendFailed = false; - let disposed = false; - let probe: SharedWorker | undefined; - let seq = 0; - - const closeProbe = () => { - if (!probe) return; - try { - probe.port.close(); - } catch {} - probe = undefined; - }; - - worker.port.addEventListener("message", (event: MessageEvent) => { - const data = event.data; - if (data?.type !== "hello" && data?.type !== "pong") return; - lastHeardAt = Date.now(); - warnedUnresponsive = false; - closeProbe(); - if (instanceId === undefined) { - instanceId = data.instanceId; - lifecycleLog( - "automerge SharedWorker instance %s (via %s)", - data.instanceId, - data.type - ); - } else if (data.instanceId && data.instanceId !== instanceId) { - lifecycleLog( - "automerge SharedWorker instance changed (instance %s, was %s)", - data.instanceId, - instanceId - ); - instanceId = data.instanceId; - } - }); - - worker.port.addEventListener("close", () => { - if (disposed) return; - lifecycleLog("automerge SharedWorker control port closed"); - void recoverAutomergeWorker("control port closed", worker); - }); - - // Not gated on the debug namespace: a worker that fails to load never replies - // to anything, and this is the only signal that says so. - worker.addEventListener("error", (event) => { - console.error("automerge SharedWorker error:", describeErrorEvent(event)); - }); - - const startProbe = (reason: string) => { - if (probe || disposed) return; - lifecycleLog( - "automerge SharedWorker %s; probing with a second connection", - reason - ); - const startedAt = Date.now(); - const p = new SharedWorker(automergeWorkerPath, { - name: "patchwork-automerge", - type: "module", - }); - probe = p; - p.port.start(); - p.port.addEventListener("message", (event: MessageEvent) => { - if (event.data?.type !== "hello") return; - setTimeout(() => { - if (disposed || probe !== p) return; - closeProbe(); - // The suspect spoke while the probe ran: it was merely busy, and - // everything queued on it has been delivered. - if (lastHeardAt >= startedAt) return; - void recoverAutomergeWorker( - `port unresponsive on a live worker (${reason}; probe confirmed)`, - worker - ); - }, PROBE_GRACE_MS); - }); - // No hello on the probe means the instance is loading or busy. The probe - // waits indefinitely rather than tearing anything down on a timer. - }; - - const heartbeat = setInterval(() => { - try { - worker.port.postMessage({ type: "ping", id: ++seq }); - } catch (error) { - // Without this a failed send is indistinguishable from a dead worker. - if (!warnedSendFailed) { - warnedSendFailed = true; - console.error("automerge SharedWorker ping send threw", error); - } - } - - const neverHeard = instanceId === undefined; - const silentMs = Date.now() - lastHeardAt; - const timeoutMs = neverHeard - ? FIRST_CONTACT_TIMEOUT_MS - : HEARTBEAT_TIMEOUT_MS; - if (silentMs <= timeoutMs) return; - - // First contact probes regardless of visibility: SharedWorkers don't - // suspend with the tab, and the probe destroys nothing. Post-contact - // silence defers to visibility, since a hidden page's throttling can fake - // it. - const visible = - typeof document === "undefined" || document.visibilityState === "visible"; - if (!neverHeard && !visible) return; - - const seconds = Math.round(silentMs / 1000); - const reason = neverHeard - ? `no hello ~${seconds}s after connecting` - : `no pong for ~${seconds}s`; - if (!warnedUnresponsive) { - warnedUnresponsive = true; - lifecycleLog("automerge SharedWorker %s (tab visible)", reason); - } - startProbe(reason); - }, HEARTBEAT_MS); - - return () => { - disposed = true; - clearInterval(heartbeat); - closeProbe(); - }; -} - -// Ref-counted locally so several callers in this tab can watch the same doc -// with a single worker subscription. -type SyncStateListener = (update: SyncStateDocMessage) => void; -const syncStateListeners = new Map>(); +// ── The automerge worker ─────────────────────────────────────────────── +// A SharedWorker holding the Repo that resolves `automerge:` URLs for the +// service worker. Tabs don't sync through it — each tab is its own node — but +// each tab keeps it alive and heartbeats it, so it's here rather than in the +// service worker, which can't own one. -function dispatchSyncState(update: SyncStateDocMessage): void { - for (const listener of syncStateListeners.get(update.documentId) ?? []) { - try { - listener(update); - } catch (err) { - console.error("sync-state listener threw", err); - } - } -} +let automergeProtocolHandlerWorkerPath = + "/automerge-protocol-handler-worker.js"; -export function subscribeSyncState( - documentId: string, - listener: SyncStateListener -): () => void { - const worker = getAutomergeWorker(); - let listeners = syncStateListeners.get(documentId); - if (!listeners) { - syncStateListeners.set(documentId, (listeners = new Set())); - worker.port.postMessage({ type: "sync-sub", documentId }); +const automergeProtocolHandlerWorker = sharedWorkerHandle( + "patchwork-automerge-protocol-handler", + () => automergeProtocolHandlerWorkerPath, + { + debugging: workerDebugging, + onMessage(event) { + forwardWorkerConsole("automerge-protocol-handler-worker", event.data); + }, } - listeners.add(listener); +); - let active = true; - return () => { - if (!active) return; - active = false; - const set = syncStateListeners.get(documentId); - if (!set) return; - set.delete(listener); - if (set.size > 0) return; - syncStateListeners.delete(documentId); - // Unsubscribe from whichever instance is current: recovery replays - // subscriptions onto a new worker, so it may not be the one captured above. - automergeWorker?.port.postMessage({ type: "sync-unsub", documentId }); - }; +export function getAutomergeProtocolHandlerWorker(): SharedWorker { + return automergeProtocolHandlerWorker.get(); } export function connectClassicSync( @@ -419,7 +90,6 @@ export function connectClassicSync( ); } - const worker = getAutomergeWorker(); const { port1, port2 } = new MessageChannel(); return new Promise((resolve, reject) => { const timeout = setTimeout(() => { @@ -433,69 +103,14 @@ export function connectClassicSync( else reject(new Error(event.data?.error ?? "connect-classic-sync failed")); }; - worker.port.postMessage({ type: "connect-classic-sync", server: url }, [ - port2, - ]); - }); -} - -function sendRepoPort(id: number): MessagePort { - const { port1, port2 } = new MessageChannel(); - getAutomergeWorker().port.postMessage({ type: "port", id }, [port2]); - return port1; -} - -/** - * Wait for the worker to confirm its repo is constructed. The MessageChannel - * adapter's whenReady() force-resolves after 100ms regardless of the other - * end's state, so it can't serve as a readiness signal on first boot, when the - * worker still has to fetch wasm and build its repo. - */ -function awaitPortReady(control: MessagePort, id: number): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - clearTimeout(timeout); - control.removeEventListener("message", listener); - }; - const listener = (event: MessageEvent) => { - if (event.data?.id !== id) return; - if (event.data.type === "port-ready") { - cleanup(); - resolve(); - } else if (event.data.type === "port-failed") { - cleanup(); - reject(new Error(`automerge worker init failed: ${event.data.error}`)); - } - }; - control.addEventListener("message", listener); - const timeout = setTimeout(() => { - cleanup(); - reject(new Error("automerge worker port-ready timeout")); - }, 30_000); - }); -} - -async function openRepoChannel(): Promise { - const id = ++nextRepoChannelId; - const ready = awaitPortReady(getAutomergeWorker().port, id); - const port = sendRepoPort(id); - try { - await ready; - } catch (err) { - // Surface the problem and let the rest of the site come up rather than - // hanging on a blank page. - console.warn( - "proceeding without worker ready ack:", - err instanceof Error ? err.message : err + automergeProtocolHandlerWorker.post( + { type: "connect-classic-sync", server: url }, + [port2] ); - } - return port; + }); } -/** Open a fresh repo sync port to the automerge worker (dev console). */ -function getRepoChannel(): MessagePort { - return sendRepoPort(++nextRepoChannelId); -} +// ── Boot ─────────────────────────────────────────────────────────────── function waitForActive(reg: ServiceWorkerRegistration): Promise { if (reg.active) return Promise.resolve(reg.active); @@ -528,11 +143,11 @@ export default async function setupServiceWorker( // default eviction. void navigator.storage?.persist?.().catch(() => {}); - if (options?.workerPath) automergeWorkerPath = options.workerPath; + if (options?.workerPath) + automergeProtocolHandlerWorkerPath = options.workerPath; - // Start the automerge worker now so it boots wasm and its repo while the - // service worker installs. - const shared = getAutomergeWorker(); + // Start it now so it boots wasm while the service worker installs. + const shared = automergeProtocolHandlerWorker.get(); const reg = await navigator.serviceWorker.register( options?.path ?? "/service-worker.js", @@ -567,27 +182,7 @@ export default async function setupServiceWorker( "background: #fcf2f0; color: #333; border: 2px solid; border-radius: 4px" ); - return { - shared, - connectClassicSync, - getRepoChannel, - subscribeSyncState, - // Called once with the boot port. If the automerge worker later dies and is - // recreated, the listener is called again with a fresh port — treat every - // call as "(re)wire your repo's sync onto this port". - async subscribeToRepoChannel(listener: ServiceWorkerRepoChannelListener) { - repoChannelListeners.add(listener); - const generation = workerGeneration; - const port = await openRepoChannel(); - // If the worker was replaced while this channel was opening, recovery has - // already delivered a good port to this listener — drop the stale one - // rather than wiring the repo to a dead channel. - if (generation === workerGeneration) await listener(port); - return () => { - repoChannelListeners.delete(listener); - }; - }, - }; + return { shared, connectClassicSync }; } (window as any).bumpServiceWorkerCache = bumpServiceWorkerCache; diff --git a/core/bootloader/src/shared-worker-lifecycle.ts b/core/bootloader/src/shared-worker-lifecycle.ts new file mode 100644 index 00000000..ae1982bb --- /dev/null +++ b/core/bootloader/src/shared-worker-lifecycle.ts @@ -0,0 +1,289 @@ +import debug from "debug"; + +export const lifecycleLog = debug("patchwork:lifecycle"); + +function describeErrorEvent(event: Event): string { + const error = event as ErrorEvent; + const where = error.filename + ? ` (${error.filename}:${error.lineno}:${error.colno})` + : ""; + return `${error.message || String(event)}${where}`; +} + +// A silent port is not proof of death: the worker may still be evaluating its +// module graph, or be busy with wasm/sync work. In both cases every queued +// message is delivered once it catches up, and tearing the port down would lose +// them. So silence only starts a non-destructive probe: a second connection to +// the same instance. Only if the probe gets a `hello` while this port stays +// silent do we know the instance is alive but our port is stranded, and +// recover. +const HEARTBEAT_MS = 5_000; +const HEARTBEAT_TIMEOUT_MS = 25_000; +// An idle worker hellos within milliseconds of connecting, so before first +// contact the budget is tighter — probing early rescues stranded boots fast. +const FIRST_CONTACT_TIMEOUT_MS = 4_000; +// After a slow boot both connections hello at roughly the same moment and +// cross-port delivery order isn't guaranteed, so give the suspect this long to +// also speak before concluding it's stranded. +const PROBE_GRACE_MS = 500; +// Below this spacing, skip: if the fresh worker is dead too, its own heartbeat +// re-triggers recovery later rather than spinning in a tight loop. +const RECOVERY_MIN_INTERVAL_MS = 15_000; + +export type SharedWorkerHandle = { + readonly name: string; + /** The current instance, spawning one if there isn't a live one. */ + get(): SharedWorker; + /** Send on the current instance's control port. */ + post(message: unknown, transfer?: Transferable[]): void; + /** + * The worker died and was replaced. Anything held against the old instance — + * a port, a subscription — is stranded; the new one boots with cold state. + */ + onRecreated(listener: () => void): () => void; +}; + +/** + * A SharedWorker a tab keeps alive: spawned on demand, heartbeated, and rebuilt + * if the browser kills it (which it may, under memory pressure). + */ +export function sharedWorkerHandle( + name: string, + /** Read on every spawn, so a site can set the path after this is built. */ + path: () => string, + { + debugging, + onMessage, + onSpawn, + }: { + debugging: boolean; + onMessage: (event: MessageEvent) => void; + onSpawn?: (worker: SharedWorker) => void; + } +): SharedWorkerHandle { + let current: SharedWorker | undefined; + let disposeDeathDetection: (() => void) | undefined; + let recovering = false; + let lastRecoveryAt = 0; + const recreatedListeners = new Set<() => void>(); + + const get = (): SharedWorker => { + if (current) return current; + + const worker = new SharedWorker(path(), { name, type: "module" }); + current = worker; + + // Fires when a message can't be structured-deserialized. Silent otherwise: + // the message is dropped, which looks identical to a worker that never + // replied. + worker.port.addEventListener("messageerror", (event) => { + console.error(`[${name}] undeserializable message from worker:`, event); + }); + // Replies come back on this port, and we listen with addEventListener + // rather than onmessage, so it needs start(). + worker.port.start(); + worker.port.addEventListener("message", onMessage); + worker.port.postMessage({ type: "debug", debug: debugging }); + + onSpawn?.(worker); + disposeDeathDetection = installDeathDetection(worker); + return worker; + }; + + async function recover(reason: string, dead: SharedWorker): Promise { + if (dead !== current) return; + if (recovering) return; + const now = Date.now(); + if (now - lastRecoveryAt < RECOVERY_MIN_INTERVAL_MS) return; + recovering = true; + lastRecoveryAt = now; + lifecycleLog("recreating the %s SharedWorker (%s)", name, reason); + + try { + disposeDeathDetection?.(); + disposeDeathDetection = undefined; + current = undefined; + try { + dead.port.close(); + } catch {} + + get(); + for (const listener of recreatedListeners) { + try { + listener(); + } catch (error) { + console.error(`[${name}] recreated listener threw`, error); + } + } + } finally { + recovering = false; + } + } + + function installDeathDetection(worker: SharedWorker): () => void { + let instanceId: string | undefined; + let lastHeardAt = Date.now(); + let warnedUnresponsive = false; + let warnedSendFailed = false; + let disposed = false; + let probe: SharedWorker | undefined; + let seq = 0; + + const closeProbe = () => { + if (!probe) return; + try { + probe.port.close(); + } catch {} + probe = undefined; + }; + + worker.port.addEventListener("message", (event: MessageEvent) => { + const data = event.data; + if (data?.type !== "hello" && data?.type !== "pong") return; + lastHeardAt = Date.now(); + warnedUnresponsive = false; + closeProbe(); + if (instanceId === undefined) { + instanceId = data.instanceId; + lifecycleLog( + "%s SharedWorker instance %s (via %s)", + name, + data.instanceId, + data.type + ); + } else if (data.instanceId && data.instanceId !== instanceId) { + lifecycleLog( + "%s SharedWorker instance changed (instance %s, was %s)", + name, + data.instanceId, + instanceId + ); + instanceId = data.instanceId; + } + }); + + worker.port.addEventListener("close", () => { + if (disposed) return; + lifecycleLog("%s SharedWorker control port closed", name); + void recover("control port closed", worker); + }); + + // Not gated on the debug namespace: a worker that fails to load never + // replies to anything, and this is the only signal that says so. + worker.addEventListener("error", (event) => { + console.error(`${name} SharedWorker error:`, describeErrorEvent(event)); + }); + + const startProbe = (reason: string) => { + if (probe || disposed) return; + lifecycleLog( + "%s SharedWorker %s; probing with a second connection", + name, + reason + ); + const startedAt = Date.now(); + const p = new SharedWorker(path(), { name, type: "module" }); + probe = p; + p.port.start(); + p.port.addEventListener("message", (event: MessageEvent) => { + if (event.data?.type !== "hello") return; + setTimeout(() => { + if (disposed || probe !== p) return; + closeProbe(); + // The suspect spoke while the probe ran: it was merely busy, and + // everything queued on it has been delivered. + if (lastHeardAt >= startedAt) return; + void recover( + `port unresponsive on a live worker (${reason}; probe confirmed)`, + worker + ); + }, PROBE_GRACE_MS); + }); + // No hello on the probe means the instance is loading or busy. The probe + // waits indefinitely rather than tearing anything down on a timer. + }; + + const heartbeat = setInterval(() => { + try { + worker.port.postMessage({ type: "ping", id: ++seq }); + } catch (error) { + // Without this a failed send is indistinguishable from a dead worker. + if (!warnedSendFailed) { + warnedSendFailed = true; + console.error(`${name} SharedWorker ping send threw`, error); + } + } + + const neverHeard = instanceId === undefined; + const silentMs = Date.now() - lastHeardAt; + const timeoutMs = neverHeard + ? FIRST_CONTACT_TIMEOUT_MS + : HEARTBEAT_TIMEOUT_MS; + if (silentMs <= timeoutMs) return; + + // First contact probes regardless of visibility: SharedWorkers don't + // suspend with the tab, and the probe destroys nothing. Post-contact + // silence defers to visibility, since a hidden page's throttling can fake + // it. + const visible = + typeof document === "undefined" || + document.visibilityState === "visible"; + if (!neverHeard && !visible) return; + + const seconds = Math.round(silentMs / 1000); + const reason = neverHeard + ? `no hello ~${seconds}s after connecting` + : `no pong for ~${seconds}s`; + if (!warnedUnresponsive) { + warnedUnresponsive = true; + lifecycleLog("%s SharedWorker %s (tab visible)", name, reason); + } + startProbe(reason); + }, HEARTBEAT_MS); + + return () => { + disposed = true; + clearInterval(heartbeat); + closeProbe(); + }; + } + + return { + name, + get, + post(message, transfer) { + get().port.postMessage(message, transfer ?? []); + }, + onRecreated(listener) { + recreatedListeners.add(listener); + return () => { + recreatedListeners.delete(listener); + }; + }, + }; +} + +/** + * Mirror a worker's forwarded console output into this tab's console, since a + * SharedWorker's own console is only visible in chrome://inspect. + */ +export function forwardWorkerConsole(name: string, data: any): boolean { + if (data?.type !== "console") return false; + const { level, args } = data; + if ( + !lifecycleLog.enabled && + typeof args?.[0] === "string" && + args[0].includes("[lifecycle]") + ) { + return true; + } + const write = (console as any)[level] ?? console.log; + // The worker's logs carry %c directives in args[0] with CSS in the following + // args, so the tag has to go inside the format string or the CSS prints raw. + if (typeof args[0] === "string") { + write(`[${name}] ${args[0]}`, ...args.slice(1)); + } else { + write(`[${name}]`, ...args); + } + return true; +} diff --git a/core/bootloader/src/siblings.ts b/core/bootloader/src/siblings.ts new file mode 100644 index 00000000..9843aafe --- /dev/null +++ b/core/bootloader/src/siblings.ts @@ -0,0 +1,32 @@ +import type { RepoConfig } from "@automerge/automerge-repo/slim"; +import { BroadcastChannelNetworkAdapter } from "@automerge/automerge-repo-network-broadcastchannel"; +import { storagePrefix } from "./storage.js"; + +type SubductionAdapters = NonNullable; + +/** + * Every Repo on this origin — each tab's, and the automerge protocol handler + * worker's — is a full Subduction node with its own storage and its own + * sync-server socket. Siblings would still meet through the server, + * eventually; this meets them over a BroadcastChannel so an edit in one tab + * lands in the others in the time it takes to post a message, online or not. + * + * What crosses the channel is Subduction: transport frames between two nodes, + * authenticated by each one's signer. No classic automerge sync runs here, so + * the adapters go to `new Repo({ subductionAdapters })` rather than to the + * network subsystem. + * + * A BroadcastChannel is a mesh in which every node sees every other one, and + * Subduction's handshake has an initiator and a responder, so the role is + * "mesh": for each pair, the node whose peer id sorts lower speaks first. + */ +export function siblingAdapters(): SubductionAdapters { + const serviceName = `${storagePrefix}-siblings`; + return [ + { + adapter: new BroadcastChannelNetworkAdapter({ channelName: serviceName }), + serviceName, + role: "mesh", + }, + ]; +} diff --git a/core/bootloader/src/types.ts b/core/bootloader/src/types.ts index 8d06656d..d14fd7d8 100644 --- a/core/bootloader/src/types.ts +++ b/core/bootloader/src/types.ts @@ -6,90 +6,6 @@ */ export const HANDOFF_CHANNEL = "@patchwork/handoff"; -/** - * BroadcastChannel on which the automerge shared worker announces remote - * heads it learns about from the sync server. Any tab can listen to stay - * informed of sync progress without repo-to-repo gossiping. - */ -export const SYNCSTATE_CHANNEL = "@patchwork/syncstate"; - -/** - * Worker → tabs: the worker's Subduction link to the sync server flipped. - * `serverPeerIds` are the directly-connected sync-server peer ids (their - * verifying keys), so a tab can tell which peer rows are *the server* and - * judge "synced" against them specifically. - */ -export interface SyncStateConnectionMessage { - type: "connection"; - connected: boolean; - serverPeerIds: string[]; -} - -/** - * Worker → tabs: the shared worker's own Subduction identity, so a tab can - * tell which peer rows are "us". `peerId` is `signer.peerId().toString()` (the - * value that shows up as a peer id); `verifyingKey` is its hex Ed25519 key. - */ -export interface SyncStateWhoAmIMessage { - type: "whoami"; - peerId: string; - verifyingKey: string; -} - -// What the worker broadcasts on SYNCSTATE_CHANNEL: only the *global* signals -// now. Per-document heads are addressed to subscribers over the control port -// instead (see SyncStateDocMessage) rather than fanned out to every tab. -export type SyncStateBroadcast = - | SyncStateConnectionMessage - | SyncStateWhoAmIMessage; - -/** - * Tab → worker: please replay the current global sync signals (whoami + - * connection) so a freshly-opened tab can orient immediately. Per-document - * heads are no longer replayed here — a tab subscribes to the specific docs it - * cares about over its control port instead (see {@link SyncSubscribeMessage}). - */ -export interface SyncStateRequestMessage { - type: "request"; - /** @deprecated ignored — per-doc state is delivered via sync-sub now. */ - documentId?: string; -} - -// ── Per-tab sync-state subscription (over the SharedWorker control port) ── -// -// The broadcast SyncState* messages above are global (connection/whoami). -// Per-document heads, by contrast, are addressed: a tab subscribes its control -// port to just the documents it cares about and the worker pushes only those -// docs' heads back down that port. The worker drops a port's whole -// subscription set automatically when the port closes (the tab went away), so -// there's no reference counting or heartbeat to leak. - -/** Tab → worker: start pushing me this document's heads (replays current state). */ -export interface SyncSubscribeMessage { - type: "sync-sub"; - documentId: string; -} - -/** Tab → worker: stop pushing me this document's heads. */ -export interface SyncUnsubscribeMessage { - type: "sync-unsub"; - documentId: string; -} - -/** - * Worker → tab (control port): a peer's heads for a subscribed document — the - * worker's own (keyed by its peerId) or a Subduction peer's (keyed by its - * verifying-key storageId). Same payload as the old broadcast remote-heads - * message, but delivered only to the tabs that asked for this document. - */ -export interface SyncStateDocMessage { - type: "sync-state"; - documentId: string; - storageId: string; - heads: string[]; - timestamp: number; -} - /** * The special URL to resolve, plus enough of the {@link Request} the service * worker is holding that the automerge worker can construct one that @@ -180,9 +96,7 @@ export interface HandoffAbortMessage { } export type HandoffReplyMessage = - | HandoffCachedMessage - | HandoffResponseMessage - | HandoffAbortMessage; + HandoffCachedMessage | HandoffResponseMessage | HandoffAbortMessage; /** * Automerge worker → world: broadcast once on startup so the service worker @@ -200,34 +114,14 @@ export type SetupServiceWorkerOptions = { path?: string; /** * The public path to the automerge shared worker file. - * Defaults to `/automerge-worker.js` + * Defaults to `/automerge-protocol-handler-worker.js` */ workerPath?: string; }; -export type ServiceWorkerRepoChannelListener = ( - port: MessagePort -) => void | Promise; - export type SetupServiceWorkerResult = { shared?: SharedWorker; kill?: () => void; /** Open a classic Automerge sync WebSocket from the automerge worker. */ connectClassicSync: (server?: string) => Promise; - subscribeToRepoChannel: ( - listener: ServiceWorkerRepoChannelListener - ) => Promise<() => void>; - /** Open a fresh repo sync port to the automerge worker (dev console). */ - getRepoChannel: () => MessagePort; - /** - * Watch one document's sync heads (this tab's own and each Subduction peer's, - * as the worker learns them). Calls `listener` on every update for that doc, - * replaying the current state on subscribe. Returns an unsubscribe function; - * the worker stops pushing the doc once the last local watcher drops it (and - * automatically if this tab goes away). - */ - subscribeSyncState: ( - documentId: string, - listener: (update: SyncStateDocMessage) => void - ) => () => void; }; diff --git a/core/bootloader/src/worker-control.ts b/core/bootloader/src/worker-control.ts new file mode 100644 index 00000000..384ac760 --- /dev/null +++ b/core/bootloader/src/worker-control.ts @@ -0,0 +1,130 @@ +// The control protocol every patchwork SharedWorker speaks with its tabs: +// console forwarding, a `hello` on connect, ping/pong for the tab's death +// detection, and a debug toggle. Everything else is the worker's own business +// and arrives through `onMessage`. + +/** A fresh instance means cold in-memory state, so tabs watch this. */ +const WORKER_INSTANCE_ID = Math.random().toString(36).slice(2); +const WORKER_BOOT_TIME = Date.now(); + +const MAX_BUFFER = 200; + +export function postToPort(port: MessagePort, message: unknown): void { + try { + port.postMessage(message); + } catch (error) { + console.warn("sending failed", error); + } +} + +function serializeArg(arg: unknown): string { + if (typeof arg === "string") return arg; + if (arg instanceof Error) return arg.stack || `${arg.name}: ${arg.message}`; + try { + return JSON.stringify(arg); + } catch { + return String(arg); + } +} + +export function startWorkerControl( + name: string, + handlers: { + onConnect?: (port: MessagePort) => void; + onMessage?: (data: any, port: MessagePort, event: MessageEvent) => void; + onClose?: (port: MessagePort) => void; + } = {} +): { log: (...args: unknown[]) => void } { + const ports = new Set(); + // Logs emitted before any tab connects (wasm boot) would otherwise be lost. + const preConnect: Array<{ level: string; args: string[] }> = []; + // `debug` reads localStorage, which a SharedWorker doesn't have, so this is + // toggled by a control message from a tab instead. + let debugging = false; + + // A SharedWorker's own console is buried in chrome://inspect, so mirror + // everything over each connected tab's control port. + const forward = (level: string, rawArgs: unknown[]) => { + const args = rawArgs.map(serializeArg); + if (!ports.size) { + if (preConnect.length < MAX_BUFFER) preConnect.push({ level, args }); + return; + } + for (const port of ports) + postToPort(port, { type: "console", level, args }); + }; + + for (const level of ["log", "info", "warn", "error", "debug"] as const) { + const original = console[level].bind(console); + console[level] = (...args: unknown[]) => { + original(...args); + forward(level, args); + }; + } + + self.addEventListener("error", (event) => { + const e = event as ErrorEvent; + forward("error", [ + `uncaught error: ${e.message}`, + e.error instanceof Error ? e.error.stack : undefined, + ]); + }); + + self.addEventListener("unhandledrejection", (event) => { + const reason = (event as PromiseRejectionEvent).reason; + forward("error", [ + "unhandled rejection:", + reason instanceof Error ? reason.stack || reason.message : reason, + ]); + }); + + self.addEventListener("connect", (event) => { + const port = (event as MessageEvent).ports[0]; + handlers.onConnect?.(port); + + port.addEventListener("message", (messageEvent) => { + const data = (messageEvent as MessageEvent).data; + if (data?.type === "ping") { + postToPort(port, { + type: "pong", + id: data.id, + instanceId: WORKER_INSTANCE_ID, + }); + return; + } + if (data?.type === "debug") { + debugging = data.debug; + return; + } + handlers.onMessage?.(data, port, messageEvent as MessageEvent); + }); + + // Fires when the owning page is destroyed. + port.addEventListener("close", () => { + ports.delete(port); + handlers.onClose?.(port); + }); + + port.start(); + postToPort(port, { + type: "hello", + instanceId: WORKER_INSTANCE_ID, + bootTime: WORKER_BOOT_TIME, + }); + + ports.add(port); + for (const { level, args } of preConnect.splice(0)) { + postToPort(port, { type: "console", level, args }); + } + }); + + console.warn( + `[lifecycle] ${name} SharedWorker started (instance ${WORKER_INSTANCE_ID})` + ); + + return { + log: (...args: unknown[]) => { + if (debugging) console.log(`[${name}]`, ...args); + }, + }; +} diff --git a/core/patchwork/src/index.ts b/core/patchwork/src/index.ts index 1e48f411..f35a1ab9 100644 --- a/core/patchwork/src/index.ts +++ b/core/patchwork/src/index.ts @@ -1,8 +1,9 @@ /** * One import for a Patchwork site. * - * `setup(options)` constructs the Repo, wires up the automerge-worker port, - * loads plugins via the ModuleWatcher, resolves the user's account document, + * `setup(options)` constructs the Repo, wires up the + * automerge-protocol-handler-worker port, loads plugins via the + * ModuleWatcher, resolves the user's account document, * installs the router, and resolves with the site's runtime API — `repo`, * `create`, `open`, `find`, `packages`, `plugins`, `sw` — which is what a * site assigns to `window.patchwork`. @@ -15,12 +16,13 @@ * Pulls in DOM- and plugin-layer dependencies, so it is for a browser site's * `main.ts` only. Non-UI consumers should import * `@inkandswitch/patchwork-bootloader` directly, which does SW registration - * and the automerge-worker handoff and nothing else. + * and the automerge-protocol-handler-worker handoff and nothing else. */ import { type AutomergeUrl, type DocHandle, - MessageChannelNetworkAdapter, + type DocumentId, + type StorageId, Repo, } from "@automerge/vanillajs/slim"; import * as Automerge from "@automerge/automerge/slim"; @@ -52,13 +54,9 @@ import type { Patchwork, PatchworkOptions, SignerIdentity, + SyncStateDocMessage, } from "./types.js"; -import { - createRepo, - firstRepoPort, - initWasm, - removeAdapterFor, -} from "./repo.js"; +import { createRepo, initWasm } from "./repo.js"; import { createRouter, type Router } from "./router.js"; import { createDefaultAccount } from "./createAccount.js"; @@ -130,46 +128,13 @@ async function doSetup(options: PatchworkOptions): Promise { let hive: AutomergeRepoKeyhive | undefined; let repo: Repo; let signerIdentity: SignerIdentity | undefined; - // Called with a fresh port when the automerge worker dies and is recreated. - // Assigned once the repo exists. - let onWorkerPortRenewed: ((port: MessagePort) => void) | undefined; if (options.repo) { log("using provided Repo"); repo = options.repo; hive = options.hive; } else { - const workerPort = await firstRepoPort(sw, (port) => { - if (onWorkerPortRenewed) onWorkerPortRenewed(port); - else { - console.warn( - "automerge worker port renewed before the repo existed; dropping it" - ); - } - }); - - let workerAdapter = new MessageChannelNetworkAdapter(workerPort); - ({ repo, hive, signerIdentity } = await createRepo(workerAdapter)); - - // The worker was recreated with cold state: wire the repo onto the fresh - // port and drop the adapter stranded on the dead one. - const bootHive = hive; - onWorkerPortRenewed = (port) => { - const fresh = new MessageChannelNetworkAdapter(port); - // Mirror the boot wiring: a keyhive repo talks to the worker through a - // keyhive adapter wrapped around the message channel. - const registered = bootHive - ? bootHive.createKeyhiveNetworkAdapter(fresh, { - onlyShareWithSyncServer: false, - periodicallyRequestSync: false, - syncRequestInterval: 2000, - }) - : fresh; - repo.networkSubsystem.addNetworkAdapter(registered as any); - removeAdapterFor(repo, workerAdapter, registered); - workerAdapter = fresh; - lifecycleLog("repo re-wired to the recreated automerge worker"); - }; + ({ repo, hive, signerIdentity } = await createRepo()); } // Dev-console / tool-runtime globals (e2e and loaded tools read these). The @@ -181,8 +146,6 @@ async function doSetup(options: PatchworkOptions): Promise { AutomergeRepo as typeof import("@automerge/automerge-repo"); if (hive) window.hive = hive; - await repo.networkSubsystem.whenReady(); - log("networkSubsystem ready"); (hive?.networkAdapter as any)?.syncKeyhive?.(); registerRepoProviderElement(repo as any); @@ -261,8 +224,8 @@ async function doSetup(options: PatchworkOptions): Promise { plugins, sw: { connectClassicSync: sw.connectClassicSync, - subscribeToRepoChannel: sw.subscribeToRepoChannel, - subscribeSyncState: sw.subscribeSyncState, + subscribeSyncState: (documentId, listener) => + subscribeSyncState(repo, documentId, listener), }, async create(type: string, init?: (doc: D) => void) { @@ -433,6 +396,50 @@ function installLifecycleLogging(): void { ); } +// The tab's own Repo hears the server's heads directly, so this is a filter +// over its `subduction-remote-heads` event. The current value is replayed +// from the handle's sync info, when the server has reported any. +function subscribeSyncState( + repo: Repo, + documentId: string, + listener: (update: SyncStateDocMessage) => void +): () => void { + const onHeads = (payload: { + documentId: string; + storageId: string; + heads: readonly string[]; + timestamp: number; + }) => { + if (payload.documentId !== documentId) return; + listener({ + type: "sync-state", + documentId, + storageId: payload.storageId, + heads: [...payload.heads], + timestamp: payload.timestamp, + }); + }; + repo.on("subduction-remote-heads", onHeads); + + void (async () => { + const handle = repo.handles[documentId as DocumentId]; + if (!handle) return; + for (const storageId of await repo.connectedSubductionPeerIds()) { + const info = handle.getSyncInfo(storageId as StorageId); + if (info) { + onHeads({ + documentId, + storageId, + heads: info.lastHeads, + timestamp: info.lastSyncTimestamp, + }); + } + } + })(); + + return () => repo.off("subduction-remote-heads", onHeads); +} + // ── Named exports ──────────────────────────────────────────────────────── export { createRepo, initWasm } from "./repo.js"; @@ -448,4 +455,5 @@ export type { PatchworkOptions, ServiceWorkerApi, SignerIdentity, + SyncStateDocMessage, } from "./types.js"; diff --git a/core/patchwork/src/repo.ts b/core/patchwork/src/repo.ts index 532b2709..868d2550 100644 --- a/core/patchwork/src/repo.ts +++ b/core/patchwork/src/repo.ts @@ -1,22 +1,17 @@ -import { - initializeWasm, - MessageChannelNetworkAdapter, - Repo, - type AutomergeUrl, -} from "@automerge/vanillajs/slim"; +import { initializeWasm, Repo } from "@automerge/vanillajs/slim"; import { IndexedDBWorkerStorageAdapter } from "@automerge/automerge-repo-storage-indexeddb/IndexedDBWorkerStorageAdapter"; +import { siblingAdapters } from "@inkandswitch/patchwork-bootloader/siblings"; import * as AutomergeRepo from "@automerge/automerge-repo/slim"; import { initKeyhiveWasm, - initializeLegacyAutomergeRepoKeyhive, - type AutomergeRepoKeyhiveBase, + initializeAutomergeRepoKeyhive, + type AutomergeRepoKeyhive, type SyncServerSelection, } from "@automerge/automerge-repo-keyhive"; // eslint-disable-next-line // @ts-ignore — initSync is a wasm-bindgen runtime helper not in the .d.ts import { initSync as initSubductionSync } from "@automerge/automerge-subduction/slim"; import { MemorySigner } from "@automerge/automerge-subduction/slim"; -import setupServiceWorker from "@inkandswitch/patchwork-bootloader"; import { keyhiveStorageName, storagePrefix, @@ -53,28 +48,33 @@ export function initWasm(): Promise { return wasmReady; } -export async function createRepo( - workerAdapter: MessageChannelNetworkAdapter -): Promise<{ +export type TabRepo = { repo: Repo; - hive?: AutomergeRepoKeyhiveBase; + hive?: AutomergeRepoKeyhive; signerIdentity?: SignerIdentity; -}> { +}; + +/** + * The tab's own node: this origin's IndexedDB, a socket to the sync server, + * and the siblings channel to every other Repo on the origin. Nothing is + * shared with other tabs except the database underneath. + */ +export async function createRepo(): Promise { if (syncServer.keyhive) { log("setting up keyhive"); initKeyhiveWasm(); - const { hive, repo } = await initializeLegacyAutomergeRepoKeyhive({ + const { hive, repo } = await initializeAutomergeRepoKeyhive({ createRepo: (repoConfig) => new Repo(repoConfig), storage: new IndexedDBWorkerStorageAdapter(keyhiveStorageName), peerIdSuffix: storagePrefix + Math.random().toString(36).slice(2), - networkAdapter: workerAdapter, automaticArchiveIngestion: true, cachingMode: "periodic", - onlyShareWithSyncServer: false, // ARK selects the relay via `syncServer`, defaulting to "subduction". syncServer: syncServer.keyhive, repo: { storage: new IndexedDBWorkerStorageAdapter(), + subductionWebsocketEndpoints: [syncServer.url], + subductionAdapters: siblingAdapters(), enableRemoteHeadsGossiping: true, }, }); @@ -82,20 +82,17 @@ export async function createRepo( return { repo, hive }; } - // An explicit signer, rather than the Repo's internal default, so the tab's - // identity can be exposed on window.patchwork. The tab never connects via - // Subduction, so this id never goes on the wire. + // The signer is explicit rather than the Repo's internal default so the + // identity the tab presents to the server can be shown on window.patchwork. const signer = new MemorySigner(); const repo = new Repo({ - network: [workerAdapter], - storage: new IndexedDBWorkerStorageAdapter(), signer, - async sharePolicy(peerId) { - return peerId.includes("automerge-worker"); - }, - enableRemoteHeadsGossiping: true, + storage: new IndexedDBWorkerStorageAdapter(), peerId: `${storagePrefix}-tab-${crypto.randomUUID()}` as AutomergeRepo.PeerId, + subductionWebsocketEndpoints: [syncServer.url], + subductionAdapters: siblingAdapters(), + enableRemoteHeadsGossiping: true, }); const signerIdentity = { peerId: signer.peerId().toString(), @@ -108,46 +105,3 @@ export async function createRepo( log("repo created, tab subduction identity:", signerIdentity); return { repo, signerIdentity }; } - -/** - * Resolve with the first repo port the worker delivers, calling `onRenewed` for - * every later one. - * - * subscribeToRepoChannel is deliberately not awaited: it resolves only after - * the boot channel's port-ready handshake, which can take its full 30s timeout - * against a stranded worker connection. Boot blocks on the first *delivered* - * port instead — if the boot channel stalls, worker recovery hands the listener - * a good port long before that timeout. - */ -export function firstRepoPort( - sw: Awaited>, - onRenewed: (port: MessagePort) => void -): Promise { - return new Promise((resolve) => { - let seen = false; - void sw.subscribeToRepoChannel((port) => { - if (seen) return onRenewed(port); - seen = true; - resolve(port); - }); - }); -} - -/** Drop the adapter sitting on the dead worker port, leaving `keep` in place. */ -export function removeAdapterFor( - repo: Repo, - stale: MessageChannelNetworkAdapter, - keep: unknown -): void { - for (const adapter of [...repo.networkSubsystem.adapters]) { - if (adapter === keep) continue; - // The keyhive wrapper keeps the wrapped adapter on `.networkAdapter`. - const base = (adapter as any).networkAdapter ?? adapter; - if (base !== stale) continue; - try { - repo.networkSubsystem.removeNetworkAdapter(adapter as any); - } catch (err) { - console.error("failed to remove stale worker network adapter", err); - } - } -} diff --git a/core/patchwork/src/site-kit/sync-servers.ts b/core/patchwork/src/site-kit/sync-servers.ts index 98fb5d93..da3927bf 100644 --- a/core/patchwork/src/site-kit/sync-servers.ts +++ b/core/patchwork/src/site-kit/sync-servers.ts @@ -2,9 +2,10 @@ import type { PatchworkSiteOptions } from "./options.js"; import type { SyncServerSelection } from "@automerge/automerge-repo-keyhive"; // Mirrors core/bootloader/src/sync-config.ts's DEFAULT_CLASSIC_SYNC_SERVER -// and automerge-worker.ts's SUBDUCTION_SYNC_URL selection — kept here as -// plain constants (rather than importing those runtime modules) since this -// only needs the hostnames, not the browser-only logic that reads them. +// and automerge-protocol-handler-worker.ts's SUBDUCTION_SYNC_URL selection — +// kept here as plain constants (rather than importing those runtime modules) +// since this only needs the hostnames, not the browser-only logic that reads +// them. export const DEFAULT_SYNC_SERVERS = { classic: "wss://sync3.automerge.org", subduction: "wss://subduction.sync.inkandswitch.com", diff --git a/core/patchwork/src/types.ts b/core/patchwork/src/types.ts index abc74207..ffdc70f0 100644 --- a/core/patchwork/src/types.ts +++ b/core/patchwork/src/types.ts @@ -8,21 +8,27 @@ import type { AccountCreator, AccountDoc, } from "@inkandswitch/patchwork-plugins"; -import type { - ServiceWorkerRepoChannelListener, - SyncStateDocMessage, -} from "@inkandswitch/patchwork-bootloader/types"; import type * as pluginsNS from "@inkandswitch/patchwork-plugins"; export type PluginsApi = typeof pluginsNS; export type SignerIdentity = { peerId: string; verifyingKey: string }; +/** The sync server's last-known heads for one document, as it reports them. */ +export interface SyncStateDocMessage { + type: "sync-state"; + documentId: string; + storageId: string; + heads: string[]; + timestamp: number; +} + export interface ServiceWorkerApi { connectClassicSync: (server?: string) => Promise; - subscribeToRepoChannel: ( - listener: ServiceWorkerRepoChannelListener - ) => Promise<() => void>; + /** + * Watch one document's heads at the sync server. Calls `listener` with + * what the server holds now, then on every update. Returns unsubscribe. + */ subscribeSyncState: ( documentId: string, listener: (update: SyncStateDocMessage) => void @@ -91,7 +97,8 @@ export interface PatchworkOptions { /** * Bring your own Repo. When provided, setup skips wasm initialization and * repo creation entirely — you are responsible for having initialized - * automerge/subduction and wired the automerge-worker port yourself. + * automerge/subduction and wired the automerge-protocol-handler-worker + * port yourself. */ repo?: Repo; diff --git a/core/patchwork/src/vite/config-plugin.ts b/core/patchwork/src/vite/config-plugin.ts index ed9e61e5..5ee5c4dd 100644 --- a/core/patchwork/src/vite/config-plugin.ts +++ b/core/patchwork/src/vite/config-plugin.ts @@ -40,9 +40,7 @@ export function buildDefines( * used to hand-write in its own vite.config.ts. Each is switched off * individually via the matching `false` option. */ -export function config( - options: PatchworkVitePluginOptions = {} -): Plugin { +export function config(options: PatchworkVitePluginOptions = {}): Plugin { return { name: "@patchwork/config", config() { @@ -87,11 +85,11 @@ export function config( }, }; }, - // The shared automerge-worker's chunk imports bypass the page's service - // worker, so offline boot needs the browser's HTTP cache to serve them - // without revalidating. Content hashes make that safe; a new build gets - // new URLs. Production gets this from the generated _headers file — vite - // preview doesn't read that, so mirror it here. + // The shared automerge-protocol-handler-worker's chunk imports bypass the + // page's service worker, so offline boot needs the browser's HTTP cache to + // serve them without revalidating. Content hashes make that safe; a new + // build gets new URLs. Production gets this from the generated _headers + // file — vite preview doesn't read that, so mirror it here. configurePreviewServer(server) { if ( options.netlify === false || @@ -101,10 +99,7 @@ export function config( } server.middlewares.use((req, res, next) => { if (req.url?.startsWith("/assets/")) { - res.setHeader( - "Cache-Control", - "public, max-age=31536000, immutable" - ); + res.setHeader("Cache-Control", "public, max-age=31536000, immutable"); } next(); }); diff --git a/core/patchwork/src/vite/service-worker-plugin.ts b/core/patchwork/src/vite/service-worker-plugin.ts index 1dbc16fd..90750744 100644 --- a/core/patchwork/src/vite/service-worker-plugin.ts +++ b/core/patchwork/src/vite/service-worker-plugin.ts @@ -19,8 +19,9 @@ export const workers = [ fileName: "service-worker.js", }, { - specifier: "@inkandswitch/patchwork-bootloader/automerge-worker", - fileName: "automerge-worker.js", + specifier: + "@inkandswitch/patchwork-bootloader/automerge-protocol-handler-worker", + fileName: "automerge-protocol-handler-worker.js", }, { specifier: "@inkandswitch/patchwork-bootloader/module-loader-worker", diff --git a/packages/e2e/README.md b/packages/e2e/README.md index 7dde9b31..7509361b 100644 --- a/packages/e2e/README.md +++ b/packages/e2e/README.md @@ -118,7 +118,8 @@ lookup missed entries when the request was the wasm offline boot 503'd — it now falls back to a url-keyed match — and hashed `/assets/*` get `Cache-Control: immutable` (netlify `_headers` + a preview middleware) so the browser's HTTP cache can serve the shared -automerge-worker's chunk imports offline, which bypass the page's SW. +automerge-protocol-handler-worker's chunk imports offline, which bypass the +page's SW. Heads-up: repeated full-suite runs can get the machine's IP temporarily rate-limited by netlify (the full-UI tests fetch the whole base module diff --git a/patches/@automerge__automerge-repo@2.6.0-subduction.48.patch b/patches/@automerge__automerge-repo@2.6.0-subduction.48.patch new file mode 100644 index 00000000..8779f3ee --- /dev/null +++ b/patches/@automerge__automerge-repo@2.6.0-subduction.48.patch @@ -0,0 +1,137 @@ +diff --git a/dist/Repo.d.ts b/dist/Repo.d.ts +index 15d4ca65aa613fb567335f09c8fcb8cb9c4a954a..0f89be0e8b7e19ebb706d88685593259ddf67c89 100644 +--- a/dist/Repo.d.ts ++++ b/dist/Repo.d.ts +@@ -259,8 +259,11 @@ export interface RepoConfig { + adapter: NetworkAdapterInterface; + serviceName: string; + /** Whether to initiate ("connect") or accept ("accept") the subduction +- * handshake for peers on this adapter. Defaults to "connect". */ +- role?: "connect" | "accept"; ++ * handshake for peers on this adapter. "mesh" decides per peer, for ++ * adapters where every node sees every other one (e.g. a ++ * BroadcastChannel): the side whose peer id sorts lower initiates. ++ * Defaults to "connect". */ ++ role?: "connect" | "accept" | "mesh"; + }[]; + /** + * Tunable timeouts for the Subduction sync engine and its +diff --git a/dist/subduction/AdapterConnections.d.ts b/dist/subduction/AdapterConnections.d.ts +index 55231beca6494ed4ab0eaeb45383b822bcea5fe3..419c4ec3d17dc9f521292d780d293f5b1aafa8ce 100644 +--- a/dist/subduction/AdapterConnections.d.ts ++++ b/dist/subduction/AdapterConnections.d.ts +@@ -10,6 +10,6 @@ export declare class AdapterConnections implements ConnectionManager { + generation(): number; + onChange(callback: () => void): void; + shutdown(): void; +- addAdapter(adapter: NetworkAdapterInterface, serviceName: string, role: "connect" | "accept"): void; ++ addAdapter(adapter: NetworkAdapterInterface, serviceName: string, role: "connect" | "accept" | "mesh"): void; + } + //# sourceMappingURL=AdapterConnections.d.ts.map +\ No newline at end of file +diff --git a/dist/subduction/AdapterConnections.js b/dist/subduction/AdapterConnections.js +index 40f525a6be944be10dd5e707c73f6af48d7ffde1..51d27e89d02439b4724eb830aceb9339c65a761c 100644 +--- a/dist/subduction/AdapterConnections.js ++++ b/dist/subduction/AdapterConnections.js +@@ -57,11 +57,12 @@ export class AdapterConnections { + try { + const subduction = await this.#subduction; + const transport = new NetworkAdapterTransport(adapter, this.#localPeerId, peerId); +- if (role === "accept") { +- await subduction.acceptTransport(transport, serviceName); ++ const initiate = role === "mesh" ? this.#localPeerId < peerId : role !== "accept"; ++ if (initiate) { ++ await subduction.connectTransport(transport, serviceName); + } + else { +- await subduction.connectTransport(transport, serviceName); ++ await subduction.acceptTransport(transport, serviceName); + } + } + catch { +diff --git a/dist/subduction/SubductionConnections.js b/dist/subduction/SubductionConnections.js +index 4760a1899a8c0127db4840d5de090f6825211d97..5b9d19d4dd1cfb0cb242f140d5401af53aadd6a9 100644 +--- a/dist/subduction/SubductionConnections.js ++++ b/dist/subduction/SubductionConnections.js +@@ -15,7 +15,9 @@ export class SubductionConnections { + // ── ConnectionManager interface ───────────────────────────────────── + isConnecting() { + for (const state of this.#connectionStates.values()) { +- if (state === "connecting") ++ // "awaiting-reconnect" counts: the loop is between attempts, not ++ // given up, so a query should wait rather than report unavailable. ++ if (state === "connecting" || state === "awaiting-reconnect") + return true; + } + return false; +diff --git a/src/Repo.ts b/src/Repo.ts +index 0b408d6bda6582078083949c2fea63cfd0ef5e1b..c3df442b29c5a6ecd59a9464b3a30c4f59e3adc4 100644 +--- a/src/Repo.ts ++++ b/src/Repo.ts +@@ -1060,8 +1060,11 @@ export interface RepoConfig { + adapter: NetworkAdapterInterface + serviceName: string + /** Whether to initiate ("connect") or accept ("accept") the subduction +- * handshake for peers on this adapter. Defaults to "connect". */ +- role?: "connect" | "accept" ++ * handshake for peers on this adapter. "mesh" decides per peer, for ++ * adapters where every node sees every other one (e.g. a ++ * BroadcastChannel): the side whose peer id sorts lower initiates. ++ * Defaults to "connect". */ ++ role?: "connect" | "accept" | "mesh" + }[] + + /** +diff --git a/src/subduction/AdapterConnections.ts b/src/subduction/AdapterConnections.ts +index 4426f11291a00b230332c63d77c683bb7ab02d24..e1fb64697b94637cd9dd968a6cebe8f5e06d89b2 100644 +--- a/src/subduction/AdapterConnections.ts ++++ b/src/subduction/AdapterConnections.ts +@@ -52,7 +52,7 @@ export class AdapterConnections implements ConnectionManager { + addAdapter( + adapter: NetworkAdapterInterface, + serviceName: string, +- role: "connect" | "accept" ++ role: "connect" | "accept" | "mesh" + ) { + this.#adapters.push(adapter) + adapter.on("peer-candidate", ({ peerId }) => { +@@ -77,7 +77,7 @@ export class AdapterConnections implements ConnectionManager { + adapter: NetworkAdapterInterface, + serviceName: string, + peerId: PeerId, +- role: "connect" | "accept" ++ role: "connect" | "accept" | "mesh" + ) { + try { + const subduction = await this.#subduction +@@ -86,10 +86,12 @@ export class AdapterConnections implements ConnectionManager { + this.#localPeerId, + peerId + ) +- if (role === "accept") { +- await subduction.acceptTransport(transport, serviceName) +- } else { ++ const initiate = ++ role === "mesh" ? this.#localPeerId < peerId : role !== "accept" ++ if (initiate) { + await subduction.connectTransport(transport, serviceName) ++ } else { ++ await subduction.acceptTransport(transport, serviceName) + } + } catch { + // Transport connection failed (e.g. peer disconnected during handshake) +diff --git a/src/subduction/SubductionConnections.ts b/src/subduction/SubductionConnections.ts +index 0d5510500c1e3a8e0d84b6f9b5f87f42096ee8e3..6b8376be24f932f21ab3ddc4b76bc81ff39a5030 100644 +--- a/src/subduction/SubductionConnections.ts ++++ b/src/subduction/SubductionConnections.ts +@@ -28,7 +28,9 @@ export class SubductionConnections implements ConnectionManager { + + isConnecting(): boolean { + for (const state of this.#connectionStates.values()) { +- if (state === "connecting") return true ++ // "awaiting-reconnect" counts: the loop is between attempts, not given ++ // up, so a query should wait rather than report unavailable. ++ if (state === "connecting" || state === "awaiting-reconnect") return true + } + return false + } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9f98e3d7..68eccdc1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -35,9 +35,10 @@ catalogs: version: 5.9.3 overrides: - '@automerge/automerge': 3.3.2 + '@automerge/automerge': 3.4.1 '@automerge/automerge-repo': 2.6.0-subduction.48 '@automerge/automerge-repo-keyhive': 0.5.0-alpha.7 + '@automerge/automerge-repo-network-broadcastchannel': 2.6.0-subduction.48 '@automerge/automerge-repo-network-messagechannel': 2.6.0-subduction.48 '@automerge/automerge-repo-network-websocket': 2.6.0-subduction.48 '@automerge/automerge-repo-react-hooks': 2.6.0-subduction.48 @@ -48,6 +49,9 @@ overrides: '@keyhive/keyhive': 0.1.0-alpha.8 solid-automerge: ^2.0.1 +patchedDependencies: + '@automerge/automerge-repo@2.6.0-subduction.48': 37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4 + importers: .: @@ -74,14 +78,17 @@ importers: core/bootloader: dependencies: '@automerge/automerge': - specifier: 3.3.2 - version: 3.3.2 + specifier: 3.4.1 + version: 3.4.1 '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-keyhive': specifier: 0.5.0-alpha.7 version: 0.5.0-alpha.7(ws@8.21.1) + '@automerge/automerge-repo-network-broadcastchannel': + specifier: 2.6.0-subduction.48 + version: 2.6.0-subduction.48 '@automerge/automerge-repo-network-messagechannel': specifier: 2.6.0-subduction.48 version: 2.6.0-subduction.48 @@ -157,11 +164,11 @@ importers: version: 4.4.3 devDependencies: '@automerge/automerge': - specifier: 3.3.2 - version: 3.3.2 + specifier: 3.4.1 + version: 3.4.1 '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-keyhive': specifier: 0.5.0-alpha.7 version: 0.5.0-alpha.7(ws@8.21.1) @@ -202,11 +209,11 @@ importers: core/filesystem: dependencies: '@automerge/automerge': - specifier: 3.3.2 - version: 3.3.2 + specifier: 3.4.1 + version: 3.4.1 '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) debug: specifier: ^4.4.3 version: 4.4.3 @@ -233,11 +240,11 @@ importers: core/patchwork: dependencies: '@automerge/automerge': - specifier: 3.3.2 - version: 3.3.2 + specifier: 3.4.1 + version: 3.4.1 '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-keyhive': specifier: 0.5.0-alpha.7 version: 0.5.0-alpha.7(ws@8.21.1) @@ -307,11 +314,11 @@ importers: version: 2.0.3 devDependencies: '@automerge/automerge': - specifier: 3.3.2 - version: 3.3.2 + specifier: 3.4.1 + version: 3.4.1 '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-keyhive': specifier: 0.5.0-alpha.7 version: 0.5.0-alpha.7(ws@8.21.1) @@ -338,11 +345,11 @@ importers: packages/edge-handles: devDependencies: '@automerge/automerge': - specifier: 3.3.2 - version: 3.3.2 + specifier: 3.4.1 + version: 3.4.1 '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-subduction': specifier: 0.16.1 version: 0.16.1 @@ -357,7 +364,7 @@ importers: devDependencies: '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) typescript: specifier: ^5.9.3 version: 5.9.3 @@ -370,7 +377,7 @@ importers: devDependencies: '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-react-hooks': specifier: 2.6.0-subduction.48 version: 2.6.0-subduction.48(react-dom@18.3.1(react@18.3.1))(react@18.3.1) @@ -392,10 +399,10 @@ importers: devDependencies: '@automerge/automerge-repo': specifier: 2.6.0-subduction.48 - version: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) solid-automerge: specifier: ^2.0.1 - version: 2.0.1(@automerge/automerge-repo@2.6.0-subduction.48)(solid-js@1.9.14) + version: 2.0.1(@automerge/automerge-repo@2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4))(solid-js@1.9.14) solid-js: specifier: ^1.9.13 version: 1.9.14 @@ -403,6 +410,40 @@ importers: specifier: ^5.9.3 version: 5.9.3 + sites/bench: + dependencies: + '@automerge/automerge-repo': + specifier: 2.6.0-subduction.48 + version: 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) + '@automerge/automerge-repo-network-broadcastchannel': + specifier: 2.6.0-subduction.48 + version: 2.6.0-subduction.48 + '@automerge/automerge-repo-storage-indexeddb': + specifier: 2.6.0-subduction.48 + version: 2.6.0-subduction.48 + '@automerge/automerge-subduction': + specifier: 0.16.1 + version: 0.16.1 + '@inkandswitch/patchwork': + specifier: workspace:* + version: link:../../core/patchwork + '@inkandswitch/patchwork-bootloader': + specifier: workspace:* + version: link:../../core/bootloader + devDependencies: + '@playwright/test': + specifier: 1.59.1 + version: 1.59.1 + '@types/node': + specifier: 'catalog:' + version: 24.13.3 + typescript: + specifier: 'catalog:' + version: 5.9.3 + vite: + specifier: ^7.3.5 + version: 7.3.6(@types/node@24.13.3) + packages: '@automerge/automerge-repo-keyhive@0.5.0-alpha.7': @@ -439,8 +480,8 @@ packages: '@automerge/automerge-subduction@0.16.1': resolution: {integrity: sha512-alH7U4eYn0O7sT6hNLv7CJhNODJi+QD6qfRQTNfZCxL3Q5JZdJ8lFpdvXUkBf+I3bu0GLvVlTaYPWGrBihjpIw==} - '@automerge/automerge@3.3.2': - resolution: {integrity: sha512-9vCdCL7pdQwUra66SBxPVHr+/t9epXKni9KDeak2rNBFMzABVh2u6gSpcwxi3jkR5qr047jVqZv9DlrOfVxLFw==} + '@automerge/automerge@3.4.1': + resolution: {integrity: sha512-zsZpbs/iDPvp+ZojIYd+gxmbcPVz2Xbkcx778G8zrt3E0zS+6saHJOm666lOuZyNRlTV4wHw9qzGTKueedeCsQ==} '@automerge/vanillajs@2.6.0-subduction.48': resolution: {integrity: sha512-pmgwTtukitG5kY60cJ7x527IQgeG0NSnoSU35Is1xmswS+05QYpZ1vVdeyKNlGcIz2Kttdvn4vzm/45h0eicYQ==} @@ -1996,7 +2037,7 @@ snapshots: '@automerge/automerge-repo-keyhive@0.5.0-alpha.7(ws@8.21.1)': dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-network-websocket': 2.6.0-subduction.48 '@automerge/automerge-subduction': 0.16.1 '@keyhive/keyhive': 0.1.0-alpha.8 @@ -2012,7 +2053,7 @@ snapshots: '@automerge/automerge-repo-network-broadcastchannel@2.6.0-subduction.48': dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) transitivePeerDependencies: - bufferutil - supports-color @@ -2020,7 +2061,7 @@ snapshots: '@automerge/automerge-repo-network-messagechannel@2.6.0-subduction.48': dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) eventemitter3: 5.0.4 transitivePeerDependencies: - bufferutil @@ -2029,7 +2070,7 @@ snapshots: '@automerge/automerge-repo-network-websocket@2.6.0-subduction.48': dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) cbor-x: 1.6.4 debug: 4.4.3 eventemitter3: 5.0.4 @@ -2041,8 +2082,8 @@ snapshots: '@automerge/automerge-repo-react-hooks@2.6.0-subduction.48(react-dom@18.3.1(react@18.3.1))(react@18.3.1)': dependencies: - '@automerge/automerge': 3.3.2 - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge': 3.4.1 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) eventemitter3: 5.0.4 react: 18.3.1 react-dom: 18.3.1(react@18.3.1) @@ -2053,15 +2094,15 @@ snapshots: '@automerge/automerge-repo-storage-indexeddb@2.6.0-subduction.48': dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) transitivePeerDependencies: - bufferutil - supports-color - utf-8-validate - '@automerge/automerge-repo@2.6.0-subduction.48': + '@automerge/automerge-repo@2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4)': dependencies: - '@automerge/automerge': 3.3.2 + '@automerge/automerge': 3.4.1 '@automerge/automerge-subduction': 0.16.1 bs58check: 4.0.0 cbor-x: 1.6.4 @@ -2079,11 +2120,11 @@ snapshots: '@automerge/automerge-subduction@0.16.1': {} - '@automerge/automerge@3.3.2': {} + '@automerge/automerge@3.4.1': {} '@automerge/vanillajs@2.6.0-subduction.48': dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@automerge/automerge-repo-network-broadcastchannel': 2.6.0-subduction.48 '@automerge/automerge-repo-network-messagechannel': 2.6.0-subduction.48 '@automerge/automerge-repo-network-websocket': 2.6.0-subduction.48 @@ -3307,9 +3348,9 @@ snapshots: slash@3.0.0: {} - solid-automerge@2.0.1(@automerge/automerge-repo@2.6.0-subduction.48)(solid-js@1.9.14): + solid-automerge@2.0.1(@automerge/automerge-repo@2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4))(solid-js@1.9.14): dependencies: - '@automerge/automerge-repo': 2.6.0-subduction.48 + '@automerge/automerge-repo': 2.6.0-subduction.48(patch_hash=37266639cf8001b55f69d72d72274d4c512898628937a4e0c8b96cb0b8dd67f4) '@solid-primitives/utils': 6.4.1(solid-js@1.9.14) cabbages: 0.2.10 solid-js: 1.9.14 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 46376bb1..59edfb89 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -9,6 +9,7 @@ overrides: "@automerge/automerge": "catalog:" "@automerge/automerge-repo": "catalog:" "@automerge/automerge-repo-keyhive": "catalog:" + "@automerge/automerge-repo-network-broadcastchannel": "catalog:" "@automerge/automerge-repo-network-messagechannel": "catalog:" "@automerge/automerge-repo-network-websocket": "catalog:" "@automerge/automerge-repo-react-hooks": "catalog:" @@ -26,9 +27,10 @@ minimumReleaseAgeExclude: - solid-automerge catalog: - "@automerge/automerge": 3.3.2 + "@automerge/automerge": 3.4.1 "@automerge/automerge-repo": 2.6.0-subduction.48 "@automerge/automerge-repo-keyhive": 0.5.0-alpha.7 + "@automerge/automerge-repo-network-broadcastchannel": 2.6.0-subduction.48 "@automerge/automerge-repo-network-messagechannel": 2.6.0-subduction.48 "@automerge/automerge-repo-network-websocket": 2.6.0-subduction.48 "@automerge/automerge-repo-react-hooks": 2.6.0-subduction.48 @@ -54,3 +56,6 @@ catalog: allowBuilds: cbor-extract: true esbuild: true + +patchedDependencies: + '@automerge/automerge-repo@2.6.0-subduction.48': patches/@automerge__automerge-repo@2.6.0-subduction.48.patch diff --git a/sites/bench/.gitignore b/sites/bench/.gitignore new file mode 100644 index 00000000..bb278d96 --- /dev/null +++ b/sites/bench/.gitignore @@ -0,0 +1,2 @@ +bench-results/ +dist/ diff --git a/sites/bench/README.md b/sites/bench/README.md new file mode 100644 index 00000000..a989afb9 --- /dev/null +++ b/sites/bench/README.md @@ -0,0 +1,46 @@ +# bench + +Topology benchmarks for the tab ↔ storage ↔ sync-server arrangement. Not +tests: nothing here gates anything. + +```sh +pnpm --filter patchwork-bench bench # build, run, print the table +pnpm --filter patchwork-bench bench:headed +``` + +Chromium only (`pnpm exec playwright install chromium` once). Talks to the +real sync server named in the site build, so the server columns need network. +Results land in `bench-results/results.md` and `results.jsonl`. + +## Modes + +The page at `/` builds one Repo and nothing else — no shell, no account, no +package list — in one of three shapes, chosen by `?mode=`: + +| mode | storage | server socket | tabs meet via | +| --- | --- | --- | --- | +| `patchwork` | each tab, same IndexedDB | one per tab | the siblings BroadcastChannel, then the server | +| `pertab` | each tab, same IndexedDB | one per tab | the server (or IndexedDB) | +| `pertab-bc` | each tab, same IndexedDB | one per tab | a hand-rolled BroadcastChannel, then the server | + +`patchwork` is `createRepo()` as shipped, plus the automerge worker that +resolves URLs for the service worker; the other two are bare Repos built in +the page, kept as the baseline the shipped path is measured against. +`?server=none` runs the bare modes with no socket. + +## What's measured + +- `boot.spec` — navigation start → `window.repo`, → server connected, for 1/3/10 + tabs; renderer memory (macOS physical footprint via `footprint`, RSS + elsewhere) and process count once they're all up. +- `sync.spec` — find a doc a sibling just created (and whether the first + `find()` settled unavailable); edit → seen in the other tabs; edit → server + holds our heads. Medians over 10 edits. +- `storage.spec` — a second tab finds the first's doc through storage alone; + two tabs edit the same doc and close, does a third see everything. +- `offline.spec` — both tabs edit with the network cut, then it returns. +- `churn.spec` — close the tab that booted everything, check the rest still sync. + +Cross-tab timings use epoch milliseconds, since `performance.now()` counts from +each page's own navigation start. `find()` in the helpers retries on +"unavailable" and reports how many tries it took. diff --git a/sites/bench/bench.config.ts b/sites/bench/bench.config.ts new file mode 100644 index 00000000..c782f450 --- /dev/null +++ b/sites/bench/bench.config.ts @@ -0,0 +1,33 @@ +import { defineConfig, devices } from "@playwright/test"; + +// Benchmarks, not tests: one worker, no retries, chromium only, so the numbers +// come from an otherwise idle browser. `pnpm bench` builds first. +const PORT = Number(process.env.PORT ?? 5199); + +export default defineConfig({ + testDir: "./tests", + timeout: 180_000, + workers: 1, + retries: 0, + fullyParallel: false, + reporter: [["list"]], + globalSetup: "./tests/global-setup.ts", + globalTeardown: "./tests/global-teardown.ts", + outputDir: "bench-results/artifacts", + use: { + ...devices["Desktop Chrome"], + // Full chromium in new-headless mode, not the headless shell: the shell + // lacks measureUserAgentSpecificMemory. + channel: "chromium", + baseURL: `http://localhost:${PORT}`, + serviceWorkers: "allow", + }, + projects: [{ name: "chromium" }], + webServer: { + command: "pnpm preview", + url: `http://localhost:${PORT}`, + timeout: 60_000, + reuseExistingServer: true, + env: { PORT: String(PORT) }, + }, +}); diff --git a/sites/bench/package.json b/sites/bench/package.json new file mode 100644 index 00000000..81140f5a --- /dev/null +++ b/sites/bench/package.json @@ -0,0 +1,26 @@ +{ + "name": "patchwork-bench", + "private": true, + "type": "module", + "description": "Topology benchmarks: shared subduction worker vs a subduction node per tab.", + "scripts": { + "build": "vite build", + "preview": "vite preview", + "bench": "vite build && playwright test -c bench.config.ts", + "bench:headed": "vite build && playwright test -c bench.config.ts --headed" + }, + "dependencies": { + "@automerge/automerge-repo": "catalog:", + "@automerge/automerge-repo-network-broadcastchannel": "catalog:", + "@automerge/automerge-repo-storage-indexeddb": "catalog:", + "@automerge/automerge-subduction": "catalog:", + "@inkandswitch/patchwork": "workspace:*", + "@inkandswitch/patchwork-bootloader": "workspace:*" + }, + "devDependencies": { + "@playwright/test": "1.59.1", + "@types/node": "catalog:", + "typescript": "catalog:", + "vite": "^7.3.5" + } +} diff --git a/sites/bench/src/main.ts b/sites/bench/src/main.ts new file mode 100644 index 00000000..549c8929 --- /dev/null +++ b/sites/bench/src/main.ts @@ -0,0 +1,153 @@ +// A page that builds one Repo in one of three topologies and exposes enough +// for the playwright specs in ../tests to time it. No UI, no account, no +// package list: the shell is out of scope here. +// +// ?mode=patchwork what patchwork does: createRepo() — the tab's own +// subduction node with this origin's IndexedDB, its own +// server socket and the siblings BroadcastChannel — plus +// the automerge worker that resolves URLs for the service +// worker +// ?mode=pertab the bare node: storage + socket, no siblings channel, no +// workers, so tabs only meet through the server (or the +// database) +// ?mode=pertab-bc pertab plus the siblings channel, hand-rolled +// +// `?server=none` runs the bare modes with no socket at all, so tabs can only +// meet through IndexedDB (and the BroadcastChannel). +import { + Repo, + type AutomergeUrl, + type DocHandle, + type PeerId, +} from "@automerge/automerge-repo/slim"; +import { IndexedDBWorkerStorageAdapter } from "@automerge/automerge-repo-storage-indexeddb/IndexedDBWorkerStorageAdapter"; +import { BroadcastChannelNetworkAdapter } from "@automerge/automerge-repo-network-broadcastchannel"; +import { MemorySigner } from "@automerge/automerge-subduction/slim"; +import { createRepo, initWasm } from "@inkandswitch/patchwork"; +import setupServiceWorker from "@inkandswitch/patchwork-bootloader"; + +declare const __SYNC_SERVER__: { url: string }; + +type Mode = "patchwork" | "pertab" | "pertab-bc"; + +const params = new URLSearchParams(location.search); +const mode = (params.get("mode") ?? "patchwork") as Mode; +const serverUrl = params.get("server") ?? __SYNC_SERVER__.url; + +const marks: Record = {}; +const mark = (name: string) => (marks[name] = performance.now()); + +const serverHeads = new Map(); +let serverPeerIds = new Set(); +const online = Promise.withResolvers(); + +function sameHeads(a: string[], b: string[]): boolean { + return a.length === b.length && a.every((head) => b.includes(head)); +} + +async function build(): Promise { + mark("start"); + await initWasm(); + mark("wasm"); + + let repo: Repo; + if (mode === "patchwork") { + const sw = await setupServiceWorker(); + if (!sw) throw new Error("no service worker"); + mark("workers"); + ({ repo } = await createRepo()); + } else { + repo = new Repo({ + signer: new MemorySigner(), + storage: new IndexedDBWorkerStorageAdapter(), + peerId: `bench-tab-${crypto.randomUUID()}` as PeerId, + subductionWebsocketEndpoints: serverUrl === "none" ? [] : [serverUrl], + network: + mode === "pertab-bc" + ? [new BroadcastChannelNetworkAdapter({ channelName: "bench" })] + : [], + enableRemoteHeadsGossiping: true, + }); + } + mark("repo"); + + if (serverUrl === "none") online.resolve(performance.now()); + repo.on("subduction-connection", async ({ connected }) => { + if (!connected) return; + serverPeerIds = new Set(await repo.connectedSubductionPeerIds()); + online.resolve(performance.now()); + }); + repo.on("subduction-remote-heads", ({ documentId, storageId, heads }) => { + if (!serverPeerIds.has(storageId)) return; + serverHeads.set(documentId, [...heads]); + }); + return repo; +} + +// `find` settles as unavailable when every source has said no, and a sibling +// tab's brand-new doc may not have reached those sources yet. Retrying is what +// an app would have to do; the attempt count is reported so the benches can +// say how often it was needed. +async function find( + url: string, + timeoutMs = 30_000 +): Promise<{ handle: DocHandle>; attempts: number }> { + const deadline = performance.now() + timeoutMs; + for (let attempts = 1; ; attempts++) { + try { + const handle = await window.repo.find>( + url as AutomergeUrl + ); + await handle.whenReady(); + return { handle, attempts }; + } catch (error) { + if (performance.now() > deadline) throw error; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + } +} + +window.bench = { + mode, + marks, + find, + online: () => online.promise, + serverPeerIds: () => [...serverPeerIds], + serverHeads: (documentId) => serverHeads.get(documentId), + // Resolves with the epoch time the server was seen holding exactly the heads + // the document has right now. Polled: a few ms of slop is fine here. + async serverConfirmed(url, timeoutMs = 30_000) { + const { handle } = await find(url); + const target = [...handle.heads()]; + const deadline = performance.now() + timeoutMs; + for (;;) { + const seen = serverHeads.get(handle.documentId); + if (seen && sameHeads(seen, target)) { + return performance.timeOrigin + performance.now(); + } + if (performance.now() > deadline) { + throw new Error(`server never confirmed ${url}: saw ${seen}`); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + }, +}; + +window.repo = await build(); +mark("ready"); +document.body.textContent = `${mode}: ready in ${Math.round(marks.ready - marks.start)}ms`; + +declare global { + interface Window { + repo: Repo; + bench: { + mode: Mode; + marks: Record; + find: typeof find; + online: () => Promise; + serverPeerIds: () => string[]; + serverHeads: (documentId: string) => string[] | undefined; + serverConfirmed: (url: string, timeoutMs?: number) => Promise; + }; + } +} diff --git a/sites/bench/tests/bench.ts b/sites/bench/tests/bench.ts new file mode 100644 index 00000000..c0ec413f --- /dev/null +++ b/sites/bench/tests/bench.ts @@ -0,0 +1,187 @@ +import { appendFileSync } from "node:fs"; +import { execFileSync } from "node:child_process"; +import type { Browser, BrowserContext, Page } from "@playwright/test"; + +export type Mode = "patchwork" | "pertab" | "pertab-bc"; +export const MODES: Mode[] = ["patchwork", "pertab", "pertab-bc"]; + +export const RESULTS = "bench-results/results.jsonl"; + +export type Result = { + metric: string; + mode: Mode; + tabs?: number; + value: number | boolean; + unit: "ms" | "MB" | "n" | "ok"; +}; + +export function record(result: Result): void { + appendFileSync(RESULTS, JSON.stringify(result) + "\n"); +} + +export function median(values: number[]): number { + const sorted = [...values].sort((a, b) => a - b); + const mid = Math.floor(sorted.length / 2); + return sorted.length % 2 + ? sorted[mid] + : (sorted[mid - 1] + sorted[mid]) / 2; +} + +export async function openTab( + context: BrowserContext, + mode: Mode, + { server }: { server?: string } = {} +): Promise { + const page = await context.newPage(); + const query = new URLSearchParams({ mode }); + if (server !== undefined) query.set("server", server); + await page.goto(`/?${query}`); + await page.waitForFunction(() => window.repo != null, null, { + timeout: 60_000, + }); + return page; +} + +export function marks(page: Page): Promise> { + return page.evaluate(() => window.bench.marks); +} + +export function online(page: Page): Promise { + return page.evaluate(() => window.bench.online()); +} + +export function createDoc(page: Page, value: object): Promise { + return page.evaluate((value) => { + const handle = window.repo.create>(); + handle.change((d) => Object.assign(d, value)); + return handle.url; + }, value); +} + +/** + * Time until `find()` has the doc in this tab, and how many tries it took: + * more than one means a first find settled as unavailable. + */ +export function timeFind( + page: Page, + url: string +): Promise<{ ms: number; attempts: number }> { + return page.evaluate(async (url) => { + const started = performance.now(); + const { attempts } = await window.bench.find(url); + return { ms: performance.now() - started, attempts }; + }, url); +} + +// Cross-page timings use epoch ms: performance.now() counts from each page's +// own navigation start, so it can't be compared between tabs. + +/** Set a field and return the time the change was made. */ +export function setField( + page: Page, + url: string, + field: string, + value: unknown +): Promise { + return page.evaluate( + async ([url, field, value]) => { + const { handle } = await window.bench.find(url); + handle.change((d) => { + d[field] = value; + }); + return performance.timeOrigin + performance.now(); + }, + [url, field, value] as const + ); +} + +/** Resolves with the time this tab saw `field === value`. */ +export function awaitField( + page: Page, + url: string, + field: string, + value: unknown, + timeoutMs = 30_000 +): Promise { + return page.evaluate( + ([url, field, value, timeoutMs]) => + new Promise(async (resolve, reject) => { + const { handle } = await window.bench.find(url).catch((error) => { + reject(error); + throw error; + }); + const check = () => { + if (handle.doc()?.[field] !== value) return false; + handle.off("change", check); + resolve(performance.timeOrigin + performance.now()); + return true; + }; + if (check()) return; + handle.on("change", check); + setTimeout(() => { + handle.off("change", check); + reject(new Error(`${field} never became ${value}`)); + }, timeoutMs); + }), + [url, field, value, timeoutMs] as const + ); +} + +export function serverConfirmed(page: Page, url: string): Promise { + return page.evaluate((url) => window.bench.serverConfirmed(url), url); +} + +export function getField(page: Page, url: string, field: string): Promise { + return page.evaluate( + async ([url, field]) => { + const { handle } = await window.bench.find(url); + return handle.doc()[field] as T; + }, + [url, field] as const + ); +} + +/** + * Memory of every renderer process in the browser, in MB — tabs, their + * dedicated workers, and the shared workers, wherever Chrome placed them. + * Nothing in a page can see across processes (measureUserAgentSpecificMemory + * only covers the caller's own agent cluster), so the pids come from CDP and + * the sizes from the OS: physical footprint on macOS, the same number Activity + * Monitor shows, and plain RSS elsewhere, which overcounts shared mappings + * per process and so flatters whichever topology has fewer processes. + */ +export async function rendererMemory( + browser: Browser +): Promise<{ mb: number; processes: number }> { + const session = await browser.newBrowserCDPSession(); + const { processInfo } = (await session.send("SystemInfo.getProcessInfo")) as { + processInfo: Array<{ type: string; id: number }>; + }; + await session.detach(); + const pids = processInfo + .filter((process) => process.type === "renderer") + .map((process) => String(process.id)); + if (!pids.length) return { mb: 0, processes: 0 }; + + if (process.platform === "darwin") { + const out = execFileSync("footprint", pids.flatMap((pid) => ["-p", pid]), { + encoding: "utf8", + }); + let mb = 0; + for (const [, size, unit] of out.matchAll( + /phys_footprint:\s+([\d.]+)\s*(KB|MB|GB)/g + )) { + mb += Number(size) * { KB: 1 / 1024, MB: 1, GB: 1024 }[unit]!; + } + return { mb: Math.round(mb), processes: pids.length }; + } + + const rss = execFileSync("ps", ["-o", "rss=", "-p", pids.join(",")], { + encoding: "utf8", + }); + const kb = rss + .split("\n") + .filter(Boolean) + .reduce((sum, line) => sum + Number(line.trim()), 0); + return { mb: Math.round(kb / 1024), processes: pids.length }; +} diff --git a/sites/bench/tests/boot.spec.ts b/sites/bench/tests/boot.spec.ts new file mode 100644 index 00000000..30c09c81 --- /dev/null +++ b/sites/bench/tests/boot.spec.ts @@ -0,0 +1,60 @@ +import { test } from "@playwright/test"; +import { MODES, marks, online, openTab, record, rendererMemory } from "./bench.js"; + +// Cold boot: navigation start to `window.repo`, then to the server link being +// up (performance.now() is relative to navigation start, so the marks are +// already the numbers wanted). Memory is read once every tab is up. +// The first tab pays everything; later tabs show what a warm origin saves. +for (const mode of MODES) { + for (const tabs of [1, 3, 10]) { + test(`${mode}: boot ${tabs} tab(s)`, async ({ browser, context }) => { + const pages = []; + for (let i = 0; i < tabs; i++) pages.push(await openTab(context, mode)); + + const first = await marks(pages[0]); + record({ + metric: "boot → repo, first tab", + mode, + tabs, + value: first.ready, + unit: "ms", + }); + if (tabs > 1) { + const last = await marks(pages[tabs - 1]); + record({ + metric: "boot → repo, last tab", + mode, + tabs, + value: last.ready, + unit: "ms", + }); + } + record({ + metric: "boot → server connected, first tab", + mode, + tabs, + value: await online(pages[0]), + unit: "ms", + }); + + await Promise.all(pages.map((page) => online(page))); + // Let storage flushes and the first sync rounds settle first. + await pages[0].waitForTimeout(2_000); + const memory = await rendererMemory(browser); + record({ + metric: "renderer memory, all tabs + workers", + mode, + tabs, + value: memory.mb, + unit: "MB", + }); + record({ + metric: "renderer processes", + mode, + tabs, + value: memory.processes, + unit: "n", + }); + }); + } +} diff --git a/sites/bench/tests/churn.spec.ts b/sites/bench/tests/churn.spec.ts new file mode 100644 index 00000000..54f0b8b6 --- /dev/null +++ b/sites/bench/tests/churn.spec.ts @@ -0,0 +1,51 @@ +import { expect, test } from "@playwright/test"; +import { + MODES, + awaitField, + createDoc, + online, + openTab, + record, + serverConfirmed, + setField, +} from "./bench.js"; + +// Close the tab that booted everything and check the survivors still sync. +// In patchwork mode that tab spawned the automerge worker; in every mode it +// owned a storage worker mid-write. +for (const mode of MODES) { + test(`${mode}: closing the first tab doesn't strand the rest`, async ({ + context, + }) => { + const first = await openTab(context, mode); + const b = await openTab(context, mode); + const c = await openTab(context, mode); + await Promise.all([online(first), online(b), online(c)]); + const url = await createDoc(first, { n: 0 }); + await awaitField(b, url, "n", 0); + await awaitField(c, url, "n", 0); + + await first.close(); + + const started = Date.now(); + const ok = await Promise.all([ + awaitField(c, url, "n", 1).then(() => true, () => false), + setField(b, url, "n", 1).then(() => serverConfirmed(b, url)).then(() => true, () => false), + ]).then((results) => results.every(Boolean)); + record({ + metric: "sync still works after the first tab closes", + mode, + value: ok, + unit: "ok", + }); + if (ok) { + record({ + metric: "edit → seen + confirmed, after first tab closed", + mode, + value: Date.now() - started, + unit: "ms", + }); + } + expect(ok).toBe(true); + }); +} diff --git a/sites/bench/tests/global-setup.ts b/sites/bench/tests/global-setup.ts new file mode 100644 index 00000000..1093c6e8 --- /dev/null +++ b/sites/bench/tests/global-setup.ts @@ -0,0 +1,8 @@ +import { mkdirSync, writeFileSync } from "node:fs"; +import { dirname } from "node:path"; +import { RESULTS } from "./bench.js"; + +export default function globalSetup(): void { + mkdirSync(dirname(RESULTS), { recursive: true }); + writeFileSync(RESULTS, ""); +} diff --git a/sites/bench/tests/global-teardown.ts b/sites/bench/tests/global-teardown.ts new file mode 100644 index 00000000..d2990860 --- /dev/null +++ b/sites/bench/tests/global-teardown.ts @@ -0,0 +1,40 @@ +import { readFileSync, writeFileSync } from "node:fs"; +import { MODES, RESULTS, type Result } from "./bench.js"; + +// One row per metric, one column per mode, so the three topologies read side +// by side. Also written to bench-results/results.md. +export default function globalTeardown(): void { + const results: Result[] = readFileSync(RESULTS, "utf8") + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line)); + if (!results.length) return; + + const rows = new Map>>(); + for (const { metric, mode, tabs, value, unit } of results) { + const key = tabs === undefined ? metric : `${metric} (${tabs} tabs)`; + const row = rows.get(key) ?? {}; + row[mode] = + unit === "ok" + ? value + ? "ok" + : "FAIL" + : unit === "n" + ? String(value) + : `${Math.round(Number(value))} ${unit}`; + rows.set(key, row); + } + + const header = ["metric", ...MODES]; + const lines = [ + `| ${header.join(" | ")} |`, + `| ${header.map(() => "---").join(" | ")} |`, + ...[...rows].map( + ([key, row]) => + `| ${key} | ${MODES.map((mode) => row[mode] ?? "–").join(" | ")} |` + ), + ]; + const table = lines.join("\n"); + writeFileSync("bench-results/results.md", table + "\n"); + console.log("\n" + table + "\n"); +} diff --git a/sites/bench/tests/offline.spec.ts b/sites/bench/tests/offline.spec.ts new file mode 100644 index 00000000..032e0373 --- /dev/null +++ b/sites/bench/tests/offline.spec.ts @@ -0,0 +1,67 @@ +import { expect, test } from "@playwright/test"; +import { + awaitField, + createDoc, + getField, + online, + openTab, + record, + serverConfirmed, + setField, + type Mode, +} from "./bench.js"; + +// Both tabs edit while the network is cut, then it comes back. Playwright's +// offline emulation applies to page targets, which covers every mode now that +// each tab owns its socket. +const MODES: Mode[] = ["patchwork", "pertab", "pertab-bc"]; + +for (const mode of MODES) { + test(`${mode}: concurrent offline edits converge on reconnect`, async ({ + context, + }) => { + const a = await openTab(context, mode); + const b = await openTab(context, mode); + await Promise.all([online(a), online(b)]); + const url = await createDoc(a, { x: 0, y: 0 }); + await awaitField(b, url, "x", 0); + await serverConfirmed(a, url); + + await context.setOffline(true); + await setField(a, url, "x", 1); + await setField(b, url, "y", 1); + await a.waitForTimeout(2_000); + await context.setOffline(false); + + const reconnected = Date.now(); + const [seenY, seenX] = await Promise.all([ + awaitField(a, url, "y", 1, 60_000).then(() => true, () => false), + awaitField(b, url, "x", 1, 60_000).then(() => true, () => false), + ]); + const converged = Date.now() - reconnected; + const confirmed = await serverConfirmed(a, url).then(() => true, () => false); + + record({ + metric: "offline edits in two tabs both survive reconnect", + mode, + value: seenY && seenX && confirmed, + unit: "ok", + }); + if (seenY && seenX) { + record({ metric: "reconnect → tabs converged", mode, value: converged, unit: "ms" }); + } + + const c = await openTab(context, mode); + const [x, y] = await Promise.all([ + getField(c, url, "x"), + getField(c, url, "y"), + ]); + expect({ seenY, seenX, confirmed, x, y }).toEqual({ + seenY: true, + seenX: true, + confirmed: true, + x: 1, + y: 1, + }); + }); +} diff --git a/sites/bench/tests/storage.spec.ts b/sites/bench/tests/storage.spec.ts new file mode 100644 index 00000000..ddf6dc64 --- /dev/null +++ b/sites/bench/tests/storage.spec.ts @@ -0,0 +1,101 @@ +import { expect, test } from "@playwright/test"; +import { + MODES, + createDoc, + getField, + openTab, + record, + setField, + type Mode, +} from "./bench.js"; + +const EDITS = 20; + +// The second-writer question. The bare modes run with no server, so a tab can +// only see another's work through the IndexedDB they both write. The patchwork +// mode's server is build-time, so it keeps its socket; the siblings channel is +// what carries the edits there, and closing tabs tests storage all the same. +const server = (mode: Mode) => (mode === "patchwork" ? undefined : "none"); + +async function flush(page: import("@playwright/test").Page) { + await page.evaluate(() => window.repo.flush()); +} + +for (const mode of MODES) { + test(`${mode}: a doc written by one tab is found by the next`, async ({ + context, + }) => { + const a = await openTab(context, mode, { server: server(mode) }); + const url = await createDoc(a, { n: 0 }); + for (let i = 1; i <= EDITS; i++) await setField(a, url, "n", i); + await flush(a); + + const b = await openTab(context, mode, { server: server(mode) }); + const started = Date.now(); + const seen = await getField(b, url, "n").catch(() => undefined); + record({ + metric: "second tab finds first tab's doc (no server)", + mode, + value: seen === EDITS, + unit: "ok", + }); + if (seen === EDITS) { + record({ + metric: "second tab find, from storage", + mode, + value: Date.now() - started, + unit: "ms", + }); + } + expect(seen).toBe(EDITS); + }); + + // Both tabs close right after their last edit, as a user would. With tabs + // alive the worker ends up with everything (checked separately); this asks + // whether edits still in flight when the tab goes away make it. + test(`${mode}: two tabs write the same doc and close; a third reads it`, async ({ + context, + }) => { + const a = await openTab(context, mode, { server: server(mode) }); + const b = await openTab(context, mode, { server: server(mode) }); + const url = await createDoc(a, { a: 0, b: 0 }); + await flush(a); + await getField(b, url, "a"); + + for (let i = 1; i <= EDITS; i++) { + await Promise.all([ + setField(a, url, "a", i), + setField(b, url, "b", i), + ]); + } + await Promise.all([flush(a), flush(b)]); + await a.close(); + await b.close(); + + // Read once, then again after a pause: the first says whether the doc is + // complete on arrival, the second whether the rest was merely late or is + // gone with the tabs that made it. + const c = await openTab(context, mode, { server: server(mode) }); + const read = () => + Promise.all([ + getField(c, url, "a").catch(() => undefined), + getField(c, url, "b").catch(() => undefined), + ]); + const first = await read(); + await c.waitForTimeout(5_000); + const [fromA, fromB] = await read(); + record({ + metric: "edits from tabs closed right after editing all survive", + mode, + value: first[0] === EDITS && first[1] === EDITS, + unit: "ok", + }); + record({ + metric: "…or at least 5s later", + mode, + value: fromA === EDITS && fromB === EDITS, + unit: "ok", + }); + expect({ fromA, fromB }).toEqual({ fromA: EDITS, fromB: EDITS }); + }); +} diff --git a/sites/bench/tests/sync.spec.ts b/sites/bench/tests/sync.spec.ts new file mode 100644 index 00000000..51359752 --- /dev/null +++ b/sites/bench/tests/sync.spec.ts @@ -0,0 +1,69 @@ +import { test } from "@playwright/test"; +import { + MODES, + awaitField, + createDoc, + median, + online, + openTab, + record, + serverConfirmed, + setField, + timeFind, +} from "./bench.js"; + +const TABS = 3; +const EDITS = 10; + +// Tab A creates and edits; tabs B.. find and watch. Each latency is a median +// over EDITS rounds. +for (const mode of MODES) { + test(`${mode}: cross-tab and server latency`, async ({ context }) => { + const pages = []; + for (let i = 0; i < TABS; i++) pages.push(await openTab(context, mode)); + await Promise.all(pages.map((page) => online(page))); + const [a, ...others] = pages; + + const url = await createDoc(a, { counter: 0 }); + // Per-tab modes with no local fan-out only learn of the doc via the + // server, so the first find includes a server round trip by design. + const finds = await Promise.all(others.map((page) => timeFind(page, url))); + record({ + metric: "find a doc another tab just created", + mode, + value: median(finds.map((find) => find.ms)), + unit: "ms", + }); + record({ + metric: "…and the first find() didn't settle unavailable", + mode, + value: finds.every((find) => find.attempts === 1), + unit: "ok", + }); + + const propagation: number[] = []; + const confirmation: number[] = []; + for (let i = 1; i <= EDITS; i++) { + const seen = others.map((page) => awaitField(page, url, "counter", i)); + const edited = await setField(a, url, "counter", i); + const [arrived, confirmed] = await Promise.all([ + Promise.all(seen), + serverConfirmed(a, url), + ]); + propagation.push(Math.max(...arrived) - edited); + confirmation.push(confirmed - edited); + } + record({ + metric: `edit → seen in ${TABS - 1} other tabs`, + mode, + value: median(propagation), + unit: "ms", + }); + record({ + metric: "edit → server holds our heads", + mode, + value: median(confirmation), + unit: "ms", + }); + }); +} diff --git a/sites/bench/tsconfig.json b/sites/bench/tsconfig.json new file mode 100644 index 00000000..71a34486 --- /dev/null +++ b/sites/bench/tsconfig.json @@ -0,0 +1,13 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "bundler", + "lib": ["ES2024", "DOM", "DOM.Iterable"], + "strict": true, + "noEmit": true, + "skipLibCheck": true, + "types": ["vite/client"] + }, + "include": ["src", "tests", "vite.config.ts", "bench.config.ts"] +} diff --git a/sites/bench/vite.config.ts b/sites/bench/vite.config.ts new file mode 100644 index 00000000..17a1dbeb --- /dev/null +++ b/sites/bench/vite.config.ts @@ -0,0 +1,22 @@ +import { defineConfig } from "vite"; +import patchwork from "@inkandswitch/patchwork/vite"; + +export default defineConfig({ + plugins: [ + patchwork({ + title: "bench", + storagePrefix: "bench", + manifest: false, + netlify: false, + buildInfo: false, + // Cross-origin isolation unlocks performance.measureUserAgentSpecificMemory, + // which is how the benches attribute memory to workers. + preview: { + headers: { + "Cross-Origin-Opener-Policy": "same-origin", + "Cross-Origin-Embedder-Policy": "credentialless", + }, + }, + }), + ], +});