1import {2 createUpgradeCoordinator,3 type UpgradeChannel,4 upgradeChannelName,5 type UpgradeCoordinatorOptions,6 type UpgradeLockManager,7 upgradeLockName,8} from "./upgrade-coordination.ts";9import { assert } from "./test-assert.ts";1011// A shared in-memory BroadcastChannel bus: every coordinator built over one12// bus behaves like a tab on the same origin and scope.13class FakeChannelBus {14 readonly channels = new Map<string, Set<FakeChannel>>();1516 open(name: string): FakeChannel {17 const channel = new FakeChannel(this, name);18 let peers = this.channels.get(name);19 if (!peers) {20 peers = new Set();21 this.channels.set(name, peers);22 }23 peers.add(channel);24 return channel;25 }2627 broadcast(from: FakeChannel, name: string, message: unknown): void {28 for (const peer of this.channels.get(name) ?? []) {29 // BroadcastChannel never delivers to the posting port.30 if (peer !== from) peer.deliver(message);31 }32 }33}3435class FakeChannel implements UpgradeChannel {36 readonly #listeners = new Set<(event: MessageEvent) => void>();37 constructor(private readonly bus: FakeChannelBus, private readonly name: string) {}38 postMessage(message: unknown): void {39 this.bus.broadcast(this, this.name, message);40 }41 addEventListener(_type: "message", listener: (event: MessageEvent) => void): void {42 this.#listeners.add(listener);43 }44 deliver(message: unknown): void {45 for (const listener of this.#listeners) {46 listener({ data: message } as MessageEvent);47 }48 }49 close(): void {50 this.#listeners.clear();51 this.bus.channels.get(this.name)?.delete(this);52 }53}5455// A minimal Web Locks implementation: shared requests coexist, an exclusive56// request waits for every earlier holder, and later requests queue behind a57// pending exclusive one — the semantics the quiescence handshake relies on.58type LockRequest = {59 mode: "shared" | "exclusive";60 grant: () => void;61};6263class FakeLockManager implements UpgradeLockManager {64 readonly #queues = new Map<string, { holders: LockRequest[]; queue: LockRequest[] }>();6566 request<T>(67 name: string,68 options: { mode: "shared" | "exclusive" },69 callback: (lock: unknown) => Promise<T> | T,70 ): Promise<T> {71 let state = this.#queues.get(name);72 if (!state) {73 state = { holders: [], queue: [] };74 this.#queues.set(name, state);75 }76 const lockState = state;77 return new Promise<T>((resolve, reject) => {78 const request: LockRequest = {79 mode: options.mode,80 grant: () => {81 lockState.holders.push(request);82 void Promise.resolve()83 .then(() => callback({}))84 .then(resolve, reject)85 .finally(() => {86 lockState.holders.splice(lockState.holders.indexOf(request), 1);87 this.#drain(lockState);88 });89 },90 };91 lockState.queue.push(request);92 this.#drain(lockState);93 });94 }9596 #drain(state: { holders: LockRequest[]; queue: LockRequest[] }): void {97 while (state.queue.length > 0) {98 const next = state.queue[0];99 if (next.mode === "exclusive") {100 if (state.holders.length > 0) return;101 state.queue.shift();102 next.grant();103 return;104 }105 if (state.holders.some((holder) => holder.mode === "exclusive")) return;106 state.queue.shift();107 next.grant();108 }109 }110}111112function tab(113 bus: FakeChannelBus,114 locks: FakeLockManager,115 overrides: Partial<UpgradeCoordinatorOptions> = {},116) {117 return createUpgradeCoordinator({118 scopePath: "/app/",119 channel: (name) => bus.open(name),120 locks: () => locks,121 ...overrides,122 });123}124125async function flush(times = 3): Promise<void> {126 for (let i = 0; i < times; i += 1) await Promise.resolve();127}128129Deno.test("channel and lock names are scope-keyed like the worker's cache names", () => {130 assert(131 upgradeChannelName("/field-notes/") === "lofi-scope-field-notes-upgrade",132 "channel name is not scope-keyed",133 );134 assert(135 upgradeLockName("/field-notes/") === "lofi-scope-field-notes-writes",136 "lock name is not scope-keyed",137 );138 assert(upgradeChannelName("/") === "lofi-scope-root-upgrade", "root scope key drifted");139});140141Deno.test("two tabs: the swap waits for the sibling's in-flight write to settle", async () => {142 const bus = new FakeChannelBus();143 const locks = new FakeLockManager();144 const initiating = tab(bus, locks);145 const sibling = tab(bus, locks);146 try {147 // The sibling has a write in flight (its shared lock is held).148 const releaseWrite = await sibling.acquireWriteLock();149 let swapped = false;150 const announce = initiating.announceUpgrade().then(() => swapped = true);151 await flush();152 assert(sibling.writesPaused(), "the sibling did not pause writes on the announcement");153 assert(initiating.writesPaused(), "the initiating tab did not pause its own writes");154 assert(!swapped, "the swap proceeded while a write was in flight");155 releaseWrite();156 await announce;157 assert(swapped, "the swap did not proceed once writes settled");158 } finally {159 initiating.dispose();160 sibling.dispose();161 }162});163164Deno.test("two tabs: writes queued during the swap window resume after activation", async () => {165 const bus = new FakeChannelBus();166 const locks = new FakeLockManager();167 const initiating = tab(bus, locks);168 const sibling = tab(bus, locks);169 try {170 await initiating.announceUpgrade();171 let acquired = false;172 const write = sibling.acquireWriteLock().then((release) => {173 acquired = true;174 return release;175 });176 await flush();177 assert(!acquired, "a paused sibling accepted a new write during the swap window");178 // Activation reached the sibling (controllerchange); its window ends.179 sibling.notifyActivation();180 (await write)();181 assert(acquired, "the sibling's queued write did not resume after activation");182 } finally {183 initiating.dispose();184 sibling.dispose();185 }186});187188Deno.test("an abandoned swap window resumes writes at the bounded timeout", async () => {189 const bus = new FakeChannelBus();190 const locks = new FakeLockManager();191 const timers: Array<() => void> = [];192 const sibling = tab(bus, locks, {193 setTimeout: (callback) => {194 timers.push(callback);195 return timers.length;196 },197 clearTimeout: () => undefined,198 });199 const initiating = tab(bus, locks);200 try {201 await initiating.announceUpgrade();202 assert(sibling.writesPaused(), "the sibling did not pause writes");203 // The worker swap never happens; the pause-window timer expires.204 for (const expire of timers.splice(0)) expire();205 assert(!sibling.writesPaused(), "the pause window did not expire");206 (await sibling.acquireWriteLock())();207 } finally {208 initiating.dispose();209 sibling.dispose();210 }211});212213Deno.test("quiescence is bounded when a shared holder never settles", async () => {214 const bus = new FakeChannelBus();215 const locks = new FakeLockManager();216 const timers: Array<() => void> = [];217 const initiating = tab(bus, locks, {218 setTimeout: (callback) => {219 timers.push(callback);220 return timers.length;221 },222 clearTimeout: () => undefined,223 });224 const sibling = tab(bus, locks);225 try {226 await sibling.acquireWriteLock(); // never released227 let swapped = false;228 const announce = initiating.announceUpgrade().then(() => swapped = true);229 await flush();230 assert(!swapped, "the swap did not wait at all");231 for (const expire of timers.splice(0)) expire();232 await announce;233 assert(swapped, "a hung write blocked the swap past its bound");234 } finally {235 initiating.dispose();236 sibling.dispose();237 }238});239240Deno.test("without Web Locks and BroadcastChannel the coordinator degrades safely", async () => {241 const coordinator = createUpgradeCoordinator({242 scopePath: "/app/",243 channel: () => undefined,244 locks: () => undefined,245 });246 try {247 (await coordinator.acquireWriteLock())();248 await coordinator.announceUpgrade();249 assert(coordinator.writesPaused(), "the local pause window is still expected");250 coordinator.notifyActivation();251 assert(!coordinator.writesPaused(), "activation did not end the local pause");252 } finally {253 coordinator.dispose();254 }255});