diff --git a/webui/src/api/oplog.ts b/webui/src/api/oplog.ts index c45e5deb..de29b345 100644 --- a/webui/src/api/oplog.ts +++ b/webui/src/api/oplog.ts @@ -1,21 +1,18 @@ import { Operation, OperationEvent, - OperationEventSchema, OperationStatus, } from "../../gen/ts/v1/operations_pb"; import { GetOperationsRequest } from "../../gen/ts/v1/service_pb"; -import { fromBinary, toBinary } from "@bufbuild/protobuf"; import { backrestService } from "./client"; -import { createSharedStream } from "./streams/sharedStream"; +import { createTabStream } from "./streams/tabStream"; -// Operation-event stream, shared across tabs (see sharedStream). onResync means -// reset + refetch; consumers load their own initial state via getOperations(). -export const operationsStream = createSharedStream({ +// Operation-event stream, held by the most recently focused tab (see +// tabStream). onConnectOrResync means reset + refetch; consumers load their +// own initial state via getOperations(). +export const operationsStream = createTabStream({ name: "backrest:operations", connect: (signal) => backrestService.getOperationEvents({}, { signal }), - encode: (event) => toBinary(OperationEventSchema, event), - decode: (bytes) => fromBinary(OperationEventSchema, bytes), }); export const getOperations = async ( diff --git a/webui/src/api/streams/sharedStream.test.ts b/webui/src/api/streams/sharedStream.test.ts deleted file mode 100644 index 405a1503..00000000 --- a/webui/src/api/streams/sharedStream.test.ts +++ /dev/null @@ -1,356 +0,0 @@ -import { - afterEach, - beforeEach, - describe, - expect, - it, - vi, - type Mock, -} from "vitest"; -import { createSharedStream, type StreamSubscriber } from "./sharedStream"; - -// --------------------------------------------------------------------------- -// Mocks: a shared in-process registry lets multiple SharedStream instances in a -// single test stand in for multiple browser tabs of the same origin. -// --------------------------------------------------------------------------- - -// --- BroadcastChannel: delivers to sibling channels of the same name, never to -// the sender (matching real semantics). --- -const bcRegistry = new Map>(); - -class MockBroadcastChannel { - onmessage: ((ev: { data: unknown }) => void) | null = null; - constructor(public readonly name: string) { - if (!bcRegistry.has(name)) bcRegistry.set(name, new Set()); - bcRegistry.get(name)!.add(this); - } - postMessage(data: unknown) { - for (const ch of bcRegistry.get(this.name) ?? []) { - if (ch !== this && ch.onmessage) { - const handler = ch.onmessage; - queueMicrotask(() => handler({ data })); - } - } - } - close() { - bcRegistry.get(this.name)?.delete(this); - } -} - -// --- Web Locks: a single exclusive holder per name, FIFO queue, signal aborts a -// still-queued request (matching real semantics). --- -interface QueuedLock { - cb: () => Promise; - resolve: (v: unknown) => void; - reject: (e: unknown) => void; - signal?: AbortSignal; -} - -class MockLockManager { - private held = new Set(); - private queues = new Map(); - - request( - name: string, - opts: { signal?: AbortSignal }, - cb: () => Promise, - ): Promise { - return new Promise((resolve, reject) => { - const entry: QueuedLock = { cb, resolve, reject, signal: opts.signal }; - if (opts.signal?.aborted) { - reject(new DOMException("aborted", "AbortError")); - return; - } - opts.signal?.addEventListener("abort", () => { - const q = this.queues.get(name); - if (q) { - const i = q.indexOf(entry); - if (i >= 0) { - q.splice(i, 1); - reject(new DOMException("aborted", "AbortError")); - } - } - }); - if (!this.held.has(name)) { - void this.grant(name, entry); - } else { - if (!this.queues.has(name)) this.queues.set(name, []); - this.queues.get(name)!.push(entry); - } - }); - } - - private async grant(name: string, entry: QueuedLock) { - this.held.add(name); - try { - entry.resolve(await entry.cb()); - } catch (e) { - entry.reject(e); - } finally { - this.held.delete(name); - const q = this.queues.get(name); - while (q && q.length) { - const next = q.shift()!; - if (!next.signal?.aborted) { - void this.grant(name, next); - break; - } - } - } - } -} - -// A never-resolving stream body that unwinds when its signal aborts — stands in -// for a long-lived server-stream that stays open. -const openUntilAborted = (signal: AbortSignal): Promise => - new Promise((resolve) => { - if (signal.aborted) return resolve(); - signal.addEventListener("abort", () => resolve(), { once: true }); - }); - -// Trivial codec over a { id } message. -type Msg = { id: number }; -const encode = (m: Msg) => new TextEncoder().encode(JSON.stringify(m)); -const decode = (b: Uint8Array): Msg => JSON.parse(new TextDecoder().decode(b)); - -const spySubscriber = (): StreamSubscriber & { - onMessage: Mock; - onConnectOrResync: Mock; -} => ({ - onMessage: vi.fn(), - onConnectOrResync: vi.fn(), -}); - -beforeEach(() => { - bcRegistry.clear(); - vi.stubGlobal("BroadcastChannel", MockBroadcastChannel); -}); - -afterEach(() => { - vi.unstubAllGlobals(); - // Remove any navigator.locks we defined. - if ("locks" in navigator) { - delete (navigator as unknown as { locks?: unknown }).locks; - } -}); - -const installLocks = () => { - Object.defineProperty(navigator, "locks", { - value: new MockLockManager(), - configurable: true, - writable: true, - }); -}; - -describe("sharedStream — shared (leader/follower) mode", () => { - beforeEach(() => installLocks()); - - it("elects a single leader; followers receive forwarded messages and never open their own upstream", async () => { - const connectA = vi.fn((signal: AbortSignal) => - (async function* () { - yield { id: 1 }; - await openUntilAborted(signal); - })(), - ); - const connectB = vi.fn((signal: AbortSignal) => - (async function* () { - yield { id: 99 }; - await openUntilAborted(signal); - })(), - ); - - const streamA = createSharedStream({ - name: "t1", - connect: connectA, - encode, - decode, - }); - const streamB = createSharedStream({ - name: "t1", - connect: connectB, - encode, - decode, - }); - - const subA = spySubscriber(); - const subB = spySubscriber(); - // A subscribes first, so A wins the lock and leads. - const disposeA = streamA.subscribe(subA); - await Promise.resolve(); - const disposeB = streamB.subscribe(subB); - - // Follower B receives the leader's message over the bus. - await vi.waitFor(() => - expect(subB.onMessage).toHaveBeenCalledWith({ id: 1 }), - ); - // Leader A delivered it locally too. - expect(subA.onMessage).toHaveBeenCalledWith({ id: 1 }); - // B never opened its own upstream stream. - expect(connectB).not.toHaveBeenCalled(); - expect(connectA).toHaveBeenCalledTimes(1); - - disposeA(); - disposeB(); - }); - - it("fails over to another tab when the leader stops", async () => { - const connectA = vi.fn((signal: AbortSignal) => - (async function* () { - yield { id: 1 }; - await openUntilAborted(signal); - })(), - ); - const connectB = vi.fn((signal: AbortSignal) => - (async function* () { - yield { id: 2 }; - await openUntilAborted(signal); - })(), - ); - - const streamA = createSharedStream({ - name: "t2", - connect: connectA, - encode, - decode, - }); - const streamB = createSharedStream({ - name: "t2", - connect: connectB, - encode, - decode, - }); - - const disposeA = streamA.subscribe(spySubscriber()); - await Promise.resolve(); - const subB = spySubscriber(); - streamB.subscribe(subB); - - await vi.waitFor(() => expect(connectA).toHaveBeenCalledTimes(1)); - expect(connectB).not.toHaveBeenCalled(); - - // Leader leaves — B should acquire the lock and open its own upstream. - disposeA(); - await vi.waitFor(() => expect(connectB).toHaveBeenCalledTimes(1)); - await vi.waitFor(() => - expect(subB.onMessage).toHaveBeenCalledWith({ id: 2 }), - ); - }); - - it("tells a late-joining follower to reload itself instead of replaying state", async () => { - const idle = (signal: AbortSignal) => - (async function* () { - await openUntilAborted(signal); - })(); - - const connectA = vi.fn(idle); - const streamA = createSharedStream({ - name: "t5", - connect: connectA, - encode, - decode, - }); - const streamB = createSharedStream({ - name: "t5", - connect: vi.fn(idle), - encode, - decode, - }); - - streamA.subscribe(spySubscriber()); - // Wait until A is the leader (its upstream opened) before B joins. - await vi.waitFor(() => expect(connectA).toHaveBeenCalledTimes(1)); - - const subB = spySubscriber(); - streamB.subscribe(subB); - - // B is told to (re)load from the API on join; the stream carries no - // snapshot, so B receives no forwarded messages from the idle leader. - expect(subB.onConnectOrResync).toHaveBeenCalledTimes(1); - await Promise.resolve(); - expect(subB.onMessage).not.toHaveBeenCalled(); - }); -}); - -describe("sharedStream — fallback mode (no Web Locks)", () => { - // navigator.locks is absent here → each tab runs its own upstream stream. - - it("delivers messages and fires onConnectOrResync on subscribe and each (re)connect", async () => { - let call = 0; - const connect = vi.fn((signal: AbortSignal) => { - call++; - const n = call; - return (async function* () { - if (n === 1) { - yield { id: 1 }; - return; // stream ends → triggers a reconnect - } - yield { id: 2 }; - await openUntilAborted(signal); - })(); - }); - - const stream = createSharedStream({ - name: "t3", - connect, - encode, - decode, - backoffMs: 5, - }); - - const sub = spySubscriber(); - const dispose = stream.subscribe(sub); - - await vi.waitFor(() => - expect(sub.onMessage).toHaveBeenCalledWith({ id: 2 }), - ); - expect(sub.onMessage).toHaveBeenCalledWith({ id: 1 }); - // Once on subscribe, then again on the reconnect gap — the first connect is - // already covered by the subscribe-time load. - expect(sub.onConnectOrResync).toHaveBeenCalledTimes(2); - - dispose(); - }); - - it("fires onConnectOrResync when a backgrounded tab (hidden past the threshold) returns to the foreground", async () => { - let hidden = false; - Object.defineProperty(document, "hidden", { - configurable: true, - get: () => hidden, - }); - - const connect = vi.fn((signal: AbortSignal) => - (async function* () { - yield { id: 1 }; - await openUntilAborted(signal); - })(), - ); - const stream = createSharedStream({ - name: "t4", - connect, - encode, - decode, - hiddenEligibilityMs: 20, - }); - - const sub = spySubscriber(); - const dispose = stream.subscribe(sub); - await vi.waitFor(() => - expect(sub.onMessage).toHaveBeenCalledWith({ id: 1 }), - ); - const before = sub.onConnectOrResync.mock.calls.length; - - // Hidden past the threshold, then visible again → catch-up reload. - hidden = true; - document.dispatchEvent(new Event("visibilitychange")); - await new Promise((r) => setTimeout(r, 40)); - hidden = false; - document.dispatchEvent(new Event("visibilitychange")); - - await vi.waitFor(() => - expect(sub.onConnectOrResync.mock.calls.length).toBe(before + 1), - ); - - dispose(); - delete (document as unknown as { hidden?: unknown }).hidden; - }); -}); diff --git a/webui/src/api/streams/sharedStream.ts b/webui/src/api/streams/sharedStream.ts deleted file mode 100644 index 9ce5ab3c..00000000 --- a/webui/src/api/streams/sharedStream.ts +++ /dev/null @@ -1,356 +0,0 @@ -// Leader-elected shared stream. Backrest is usually served over http:// (h2c), -// where browsers use HTTP/1.1 and cap ~6 connections/origin across all tabs, so -// a long-lived server-stream per tab exhausts the pool. Here one tab (the leader, -// via Web Locks) holds the upstream stream and rebroadcasts messages to the rest -// over a BroadcastChannel; followers hold no upstream connection. Falls back to a -// per-tab stream when Web Locks/BroadcastChannel are unavailable (e.g. Chrome -// Android). - -/** A tab hidden at least this long stops being eligible to lead. */ -export const BACKGROUND_ELIGIBILITY_MS = 120_000; - -const DEFAULT_BACKOFF_MS = 5_000; - -export type StreamStatus = "live" | "reconnecting" | "offline"; - -export interface StreamSubscriber { - onMessage(msg: T): void; - /** Fired whenever this subscriber (re)joins a live stream: on subscribe, on - * reconnect/leader-handoff, and on foreground return past the hidden - * threshold. The stream carries only deltas, so consumers (re)load their - * full initial state from the API here. */ - onConnectOrResync(): void; - onStatus?(status: StreamStatus): void; -} - -export interface SharedStreamOpts { - /** Used for both the lock name and the BroadcastChannel name. */ - name: string; - connect: (signal: AbortSignal) => AsyncIterable; - encode: (msg: T) => Uint8Array; - decode: (bytes: Uint8Array) => T; - backoffMs?: number; - /** How long a hidden tab stays eligible to lead. Overridable for tests. */ - hiddenEligibilityMs?: number; -} - -export interface SharedStream { - /** First subscribe starts the stream, last unsubscribe tears it down. */ - subscribe(sub: StreamSubscriber): () => void; -} - -// Uint8Array is structured-cloneable, so encoded proto payloads pass through the -// BroadcastChannel unchanged. -type BusMsg = - | { t: "m"; d: Uint8Array } // forwarded message - | { t: "r" }; // leader (re)connected — reload from the API - -const hasWebLocks = (): boolean => - typeof navigator !== "undefined" && "locks" in navigator; -const hasBroadcastChannel = (): boolean => - typeof BroadcastChannel !== "undefined"; -const hasDocument = (): boolean => typeof document !== "undefined"; - -export function createSharedStream( - opts: SharedStreamOpts, -): SharedStream { - return new SharedStreamImpl(opts); -} - -class SharedStreamImpl implements SharedStream { - private readonly backoffMs: number; - private readonly shared: boolean; - private readonly eligibility: Eligibility; - - private readonly subscribers = new Set>(); - - // Non-null while running; its signal aborts the leader loop / election on the - // last unsubscribe. Doubles as the started/stopped flag. - private run: AbortController | null = null; - - private bc: BroadcastChannel | null = null; - private status: StreamStatus | null = null; - private hasBeenLive = false; - - constructor(private readonly opts: SharedStreamOpts) { - this.backoffMs = opts.backoffMs ?? DEFAULT_BACKOFF_MS; - this.shared = hasWebLocks() && hasBroadcastChannel(); - this.eligibility = new Eligibility( - opts.hiddenEligibilityMs ?? BACKGROUND_ELIGIBILITY_MS, - () => this.deliverConnectOrResync(), - ); - } - - subscribe(sub: StreamSubscriber): () => void { - this.subscribers.add(sub); - if (!this.run) this.start(); - // A late joiner missed everything so far; have it load initial state now. - this.fireConnectOrResync(sub); - return () => { - this.subscribers.delete(sub); - if (this.subscribers.size === 0) this.stop(); - }; - } - - // --- lifecycle ---------------------------------------------------------- - - private start() { - const run = new AbortController(); - this.run = run; - this.hasBeenLive = false; - this.eligibility.start(); - - if (this.shared) { - this.bc = new BroadcastChannel(this.opts.name); - this.bc.onmessage = (e) => this.handleBusMessage(e.data as BusMsg); - void this.runElection(run.signal); - } else { - void this.leaderStreamLoop(run.signal); - } - } - - private stop() { - this.run?.abort(); - this.run = null; - this.status = null; - - // Unblocks anything waiting on eligibility / leadership so the election can - // observe the abort and unwind. - this.eligibility.stop(); - - if (this.bc) { - this.bc.onmessage = null; - this.bc.close(); - this.bc = null; - } - } - - // --- leader election ---------------------------------------------------- - - private async runElection(runSignal: AbortSignal) { - while (!runSignal.aborted) { - await this.eligibility.waitUntilEligible(); - if (runSignal.aborted) return; - - // One signal serves both roles: a Web Locks signal aborts a still-queued - // request (so we leave the queue when we lose eligibility), and once the - // lock is held the leader loop exits on the same signal, resolving the - // callback and releasing the lock. stop() also aborts it via - // eligibility.stop(), so teardown unwinds a held lock too. - const signal = this.eligibility.ineligibleSignal(); - try { - await navigator.locks.request(this.opts.name, { signal }, () => - this.leaderStreamLoop(signal), - ); - } catch (err) { - // AbortError just means we left the queue; otherwise log and re-contend. - if ((err as Error)?.name !== "AbortError") { - console.warn(`[sharedStream:${this.opts.name}] lock error`, err); - } - } - } - } - - private async leaderStreamLoop(signal: AbortSignal) { - while (!signal.aborted) { - this.emitStatus("reconnecting"); - try { - let first = true; - for await (const msg of this.opts.connect(signal)) { - if (first) { - first = false; - this.goLive(); - } - this.deliverMessage(msg); - this.post({ t: "m", d: this.opts.encode(msg) }); - } - } catch (err) { - if (!signal.aborted) { - console.warn(`[sharedStream:${this.opts.name}] stream error`, err); - } - } - if (signal.aborted) break; - this.emitStatus("offline"); - await abortableDelay(this.backoffMs, signal); - } - } - - // --- data plane --------------------------------------------------------- - - private handleBusMessage(msg: BusMsg) { - switch (msg.t) { - case "m": - this.hasBeenLive = true; - this.emitStatus("live"); - this.deliverMessage(this.opts.decode(msg.d)); - break; - case "r": - this.emitStatus("live"); - this.deliverConnectOrResync(); - break; - } - } - - // Leader (re)connected its upstream. On the first connect every subscriber - // already loaded via subscribe, so only a reconnect/handoff is a real gap - // worth reloading for. - private goLive() { - if (this.hasBeenLive) { - this.post({ t: "r" }); - this.deliverConnectOrResync(); - } - this.hasBeenLive = true; - this.emitStatus("live"); - } - - private deliverMessage(msg: T) { - for (const sub of this.subscribers) { - try { - sub.onMessage(msg); - } catch (e) { - console.warn(`[sharedStream:${this.opts.name}] onMessage threw`, e); - } - } - } - - private deliverConnectOrResync() { - for (const sub of this.subscribers) this.fireConnectOrResync(sub); - } - - private fireConnectOrResync(sub: StreamSubscriber) { - try { - sub.onConnectOrResync(); - } catch (e) { - console.warn(`[sharedStream:${this.opts.name}] onConnectOrResync threw`, e); - } - } - - private emitStatus(status: StreamStatus) { - if (this.status === status) return; - this.status = status; - for (const sub of this.subscribers) sub.onStatus?.(status); - } - - private post(msg: BusMsg) { - this.bc?.postMessage(msg); - } - -} - -// Owns tab visibility and the hidden-eligibility timer, exposing the two things -// leader election needs: a promise that resolves while eligible, and a one-shot -// signal that aborts when eligibility is lost. A tab hidden past the threshold -// stops being eligible to lead so a foreground tab can take over. -class Eligibility { - private eligible = true; - private hiddenTimer: ReturnType | null = null; - private readonly waiters: Array<() => void> = []; - private readonly pending = new Set(); - private readonly onVisibilityChange = () => this.handleVisibilityChange(); - - constructor( - private readonly hiddenMs: number, - // Fired on each ineligible -> eligible transition (never on the initial - // eligible state), letting the owner run a catch-up after being throttled. - private readonly onRegainEligible: () => void, - ) {} - - start() { - if (hasDocument()) { - this.eligible = !document.hidden; // hidden-but-recent is still eligible - if (document.hidden) this.armHiddenTimer(); - document.addEventListener("visibilitychange", this.onVisibilityChange); - } else { - this.eligible = true; - } - } - - stop() { - if (hasDocument()) { - document.removeEventListener("visibilitychange", this.onVisibilityChange); - } - this.clearHiddenTimer(); - // Resolve any eligibility wait and abort any in-flight leadership signal so - // the election can unwind. - this.eligible = true; - this.drainWaiters(); - this.abortPending(); - } - - // Resolves immediately if eligible, else on the next transition to eligible - // (or on stop). - waitUntilEligible(): Promise { - if (this.eligible) return Promise.resolve(); - return new Promise((resolve) => this.waiters.push(resolve)); - } - - // A fresh one-shot signal that aborts when eligibility is next lost (or on - // stop). - ineligibleSignal(): AbortSignal { - const ac = new AbortController(); - this.pending.add(ac); - return ac.signal; - } - - private handleVisibilityChange() { - if (document.hidden) { - this.armHiddenTimer(); - return; - } - this.clearHiddenTimer(); - this.setEligible(true); - } - - private setEligible(next: boolean) { - if (this.eligible === next) return; - this.eligible = next; - if (next) { - this.drainWaiters(); - this.onRegainEligible(); - } else { - this.abortPending(); - } - } - - private armHiddenTimer() { - this.clearHiddenTimer(); - this.hiddenTimer = setTimeout(() => this.setEligible(false), this.hiddenMs); - } - - private clearHiddenTimer() { - if (this.hiddenTimer !== null) { - clearTimeout(this.hiddenTimer); - this.hiddenTimer = null; - } - } - - private drainWaiters() { - const waiters = this.waiters.splice(0); - for (const w of waiters) w(); - } - - private abortPending() { - for (const ac of this.pending) ac.abort(); - this.pending.clear(); - } -} - -/** A setTimeout that also resolves early if `signal` aborts. */ -function abortableDelay(ms: number, signal: AbortSignal): Promise { - return new Promise((resolve) => { - if (signal.aborted) return resolve(); - const cleanup = () => { - clearTimeout(timer); - signal.removeEventListener("abort", onAbort); - }; - const onAbort = () => { - cleanup(); - resolve(); - }; - const timer = setTimeout(() => { - cleanup(); - resolve(); - }, ms); - signal.addEventListener("abort", onAbort, { once: true }); - }); -} diff --git a/webui/src/api/streams/tabStream.test.ts b/webui/src/api/streams/tabStream.test.ts new file mode 100644 index 00000000..6ddd1d5a --- /dev/null +++ b/webui/src/api/streams/tabStream.test.ts @@ -0,0 +1,252 @@ +import { + afterEach, + beforeEach, + describe, + expect, + it, + vi, + type Mock, +} from "vitest"; +import { createTabStream, type StreamSubscriber } from "./tabStream"; + +// Delivers to sibling channels of the same name, never to the sender +// (matching real semantics). A shared registry lets multiple TabStream +// instances in one test stand in for multiple tabs, and lets tests inject +// claim/release messages as if from another tab. +const bcRegistry = new Map>(); + +class MockBroadcastChannel { + onmessage: ((ev: { data: unknown }) => void) | null = null; + constructor(public readonly name: string) { + if (!bcRegistry.has(name)) bcRegistry.set(name, new Set()); + bcRegistry.get(name)!.add(this); + } + postMessage(data: unknown) { + for (const ch of bcRegistry.get(this.name) ?? []) { + if (ch !== this && ch.onmessage) { + const handler = ch.onmessage; + queueMicrotask(() => handler({ data })); + } + } + } + close() { + bcRegistry.get(this.name)?.delete(this); + } +} + +// Posts a message into a named channel as if from another tab. +const postAs = (name: string, data: unknown) => { + const ch = new MockBroadcastChannel(name); + ch.postMessage(data); + ch.close(); +}; + +// A never-resolving stream body that unwinds when its signal aborts — stands +// in for a long-lived server-stream that stays open. +const openUntilAborted = (signal: AbortSignal): Promise => + new Promise((resolve) => { + if (signal.aborted) return resolve(); + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + +type Msg = { id: number }; + +const streamOf = (msg: Msg) => + vi.fn((signal: AbortSignal) => + (async function* () { + yield msg; + await openUntilAborted(signal); + })(), + ); + +const spySubscriber = (): StreamSubscriber & { + onMessage: Mock; + onConnectOrResync: Mock; +} => ({ + onMessage: vi.fn(), + onConnectOrResync: vi.fn(), +}); + +const tick = (ms = 2) => new Promise((r) => setTimeout(r, ms)); + +beforeEach(() => { + bcRegistry.clear(); + vi.stubGlobal("BroadcastChannel", MockBroadcastChannel); +}); + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("tabStream", () => { + it("streams, reconnects with backoff, and fires onConnectOrResync on subscribe and each connect attempt", async () => { + let call = 0; + const connect = vi.fn((signal: AbortSignal) => { + call++; + const n = call; + return (async function* () { + if (n === 1) { + yield { id: 1 }; + return; // stream ends → triggers a reconnect + } + yield { id: 2 }; + await openUntilAborted(signal); + })(); + }); + + const stream = createTabStream({ + name: "t1", + connect, + backoffMs: 5, + }); + + const sub = spySubscriber(); + const dispose = stream.subscribe(sub); + + await vi.waitFor(() => + expect(sub.onMessage).toHaveBeenCalledWith({ id: 2 }), + ); + expect(sub.onMessage).toHaveBeenCalledWith({ id: 1 }); + // Once on subscribe, then once per connect attempt. + expect(sub.onConnectOrResync).toHaveBeenCalledTimes(3); + + dispose(); + }); + + it("hands the stream to the most recently started tab", async () => { + const connectA = streamOf({ id: 1 }); + const connectB = streamOf({ id: 2 }); + const streamA = createTabStream({ name: "t2", connect: connectA }); + const streamB = createTabStream({ name: "t2", connect: connectB }); + + const subA = spySubscriber(); + const disposeA = streamA.subscribe(subA); + await vi.waitFor(() => + expect(subA.onMessage).toHaveBeenCalledWith({ id: 1 }), + ); + + // B starts later, so its claim is newer: A stands down, B streams. + await tick(); + const subB = spySubscriber(); + const disposeB = streamB.subscribe(subB); + await vi.waitFor(() => + expect(subB.onMessage).toHaveBeenCalledWith({ id: 2 }), + ); + const aSignal = connectA.mock.calls[0][0]; + expect(aSignal.aborted).toBe(true); + expect(connectA).toHaveBeenCalledTimes(1); // A did not reconnect + + disposeA(); + disposeB(); + }); + + it("stands down for a newer claim and reclaims on that tab's release", async () => { + const connect = streamOf({ id: 1 }); + const stream = createTabStream({ name: "t3", connect, backoffMs: 5 }); + + const sub = spySubscriber(); + const dispose = stream.subscribe(sub); + await vi.waitFor(() => + expect(sub.onMessage).toHaveBeenCalledWith({ id: 1 }), + ); + + // Another tab claims: this tab's connection aborts and stays down. + postAs("t3", { t: "claim", ts: Date.now() + 1000, id: "other" }); + await vi.waitFor(() => + expect(connect.mock.calls[0][0].aborted).toBe(true), + ); + await tick(20); + expect(connect).toHaveBeenCalledTimes(1); + + // That tab goes away: this tab reclaims and reconnects. + postAs("t3", { t: "release", id: "other" }); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => + expect(sub.onMessage).toHaveBeenCalledTimes(2), + ); + + dispose(); + }); + + it("releases its claim on teardown so another tab can reclaim", async () => { + const connectA = streamOf({ id: 1 }); + const connectB = streamOf({ id: 2 }); + const streamA = createTabStream({ name: "t4", connect: connectA }); + const streamB = createTabStream({ name: "t4", connect: connectB }); + + const disposeA = streamA.subscribe(spySubscriber()); + await tick(); + const subB = spySubscriber(); + const disposeB = streamB.subscribe(subB); + await vi.waitFor(() => + expect(subB.onMessage).toHaveBeenCalledWith({ id: 2 }), + ); + + // B (the streamer) tears down: A reclaims and starts streaming. + disposeB(); + await vi.waitFor(() => expect(connectA).toHaveBeenCalledTimes(2)); + + disposeA(); + }); + + it("stops streaming when hidden past the grace period and reconnects with a reload on return", async () => { + let hidden = false; + Object.defineProperty(document, "hidden", { + configurable: true, + get: () => hidden, + }); + + const connect = streamOf({ id: 1 }); + const stream = createTabStream({ + name: "t5", + connect, + hiddenGraceMs: 20, + }); + + const sub = spySubscriber(); + const dispose = stream.subscribe(sub); + await vi.waitFor(() => + expect(sub.onMessage).toHaveBeenCalledWith({ id: 1 }), + ); + const before = sub.onConnectOrResync.mock.calls.length; + + hidden = true; + document.dispatchEvent(new Event("visibilitychange")); + await vi.waitFor(() => + expect(connect.mock.calls[0][0].aborted).toBe(true), + ); + + hidden = false; + document.dispatchEvent(new Event("visibilitychange")); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(2)); + expect(sub.onConnectOrResync.mock.calls.length).toBe(before + 1); + + dispose(); + delete (document as unknown as { hidden?: unknown }).hidden; + }); + + it("streams per tab when BroadcastChannel is unavailable", async () => { + vi.stubGlobal("BroadcastChannel", undefined); + + const connectA = streamOf({ id: 1 }); + const connectB = streamOf({ id: 2 }); + const streamA = createTabStream({ name: "t6", connect: connectA }); + const streamB = createTabStream({ name: "t6", connect: connectB }); + + const subA = spySubscriber(); + const subB = spySubscriber(); + const disposeA = streamA.subscribe(subA); + const disposeB = streamB.subscribe(subB); + + // No claims can be heard, so both stream for themselves. + await vi.waitFor(() => + expect(subA.onMessage).toHaveBeenCalledWith({ id: 1 }), + ); + await vi.waitFor(() => + expect(subB.onMessage).toHaveBeenCalledWith({ id: 2 }), + ); + + disposeA(); + disposeB(); + }); +}); diff --git a/webui/src/api/streams/tabStream.ts b/webui/src/api/streams/tabStream.ts new file mode 100644 index 00000000..bce6684f --- /dev/null +++ b/webui/src/api/streams/tabStream.ts @@ -0,0 +1,295 @@ +// One upstream server-stream per stream name, held by whichever Backrest tab +// the user touched most recently. Backrest is usually served over http:// +// (h2c), where browsers use HTTP/1.1 and cap ~6 connections/origin across all +// tabs, so a stream per tab exhausts the pool — and a full pool hangs rather +// than errors. +// +// Rules: +// 1. A tab may stream while it is visible (brief hides are forgiven for +// hiddenGraceMs) and holds the newest claim among Backrest tabs. +// 2. A tab claims by broadcasting {ts, id} when it gains focus or becomes +// visible. The newest claim wins, ties broken by id, so all tabs agree +// on the winner. A closing tab broadcasts a release so a visible +// survivor can reclaim without waiting for a click. +// 3. Every connect attempt starts by telling subscribers to reload from the +// API, covering any deltas missed since the previous connection. Reloads +// are idempotent and the reconnect backoff bounds their rate. +// +// The channel carries only claim/release messages, never data. If +// BroadcastChannel is missing or a message is lost, tabs merely fail to stand +// down: extra connections, never wrong data. A tab that stood down keeps its +// last-loaded state and catches up when it next claims. + +/** A tab hidden at least this long stops streaming until it's next visible. */ +export const HIDDEN_GRACE_MS = 120_000; + +const DEFAULT_BACKOFF_MS = 5_000; + +export interface StreamSubscriber { + onMessage(msg: T): void; + /** Reload full state from the API: fired on subscribe and at the start of + * every connect attempt (rule 3 above). */ + onConnectOrResync(): void; +} + +export interface TabStreamOpts { + /** Names the claim channel; shared by all tabs streaming the same thing. */ + name: string; + connect: (signal: AbortSignal) => AsyncIterable; + backoffMs?: number; + /** Overridable for tests. */ + hiddenGraceMs?: number; +} + +export interface TabStream { + /** First subscribe starts the stream, last unsubscribe tears it down. */ + subscribe(sub: StreamSubscriber): () => void; +} + +export function createTabStream(opts: TabStreamOpts): TabStream { + return new TabStreamImpl(opts); +} + +interface Claim { + ts: number; + id: string; +} + +type ClaimMsg = + | { t: "claim"; ts: number; id: string } + | { t: "release"; id: string }; + +// Tabs share a clock (same machine), so timestamps are comparable; the id +// tiebreak only matters for same-millisecond claims and just has to be a rule +// every tab applies identically. +const newerClaim = (a: Claim, b: Claim) => + a.ts !== b.ts ? a.ts > b.ts : a.id > b.id; + +/** Tracks rule 1: is this tab visible and holding the newest claim? */ +class Eligibility { + private readonly id = Math.random().toString(36).slice(2); + /** Newest claim seen from any tab, ours included. */ + private latest: Claim = { ts: 0, id: this.id }; + private visible = true; + private hiddenTimer: ReturnType | null = null; + private channel: BroadcastChannel | null = null; + + /** Single consumer (the stream loop); fired on any possible change. */ + onChange: (() => void) | null = null; + + constructor( + name: string, + private readonly hiddenGraceMs: number, + ) { + if (typeof document === "undefined") return; // non-browser: always eligible + this.visible = !document.hidden; + document.addEventListener("visibilitychange", this.onVisibilityChange); + window.addEventListener("focus", this.claim); + window.addEventListener("pageshow", this.claimIfVisible); + window.addEventListener("pagehide", this.release); + if (typeof BroadcastChannel !== "undefined") { + try { + this.channel = new BroadcastChannel(name); + this.channel.onmessage = (e) => this.onBusMessage(e.data as ClaimMsg); + } catch (err) { + console.warn(`[tabStream:${name}] BroadcastChannel unusable`, err); + } + } + this.claimIfVisible(); + } + + get eligible(): boolean { + return this.visible && this.latest.id === this.id; + } + + stop() { + if (typeof document !== "undefined") { + document.removeEventListener("visibilitychange", this.onVisibilityChange); + window.removeEventListener("focus", this.claim); + window.removeEventListener("pageshow", this.claimIfVisible); + window.removeEventListener("pagehide", this.release); + } + if (this.hiddenTimer !== null) clearTimeout(this.hiddenTimer); + this.release(); + if (this.channel) { + this.channel.onmessage = null; + this.channel.close(); + this.channel = null; + } + this.onChange = null; + } + + /** Resolves on the next possible eligibility change, or when signal aborts. */ + changed(signal: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal.aborted) return resolve(); + const done = () => { + this.onChange = null; + signal.removeEventListener("abort", done); + resolve(); + }; + this.onChange = done; + signal.addEventListener("abort", done, { once: true }); + }); + } + + private readonly claim = () => { + const mine: Claim = { ts: Date.now(), id: this.id }; + if (newerClaim(mine, this.latest)) this.latest = mine; + this.channel?.postMessage({ t: "claim", ...mine } satisfies ClaimMsg); + this.onChange?.(); + }; + + private readonly claimIfVisible = () => { + if (this.visible) this.claim(); + }; + + private readonly release = () => { + this.channel?.postMessage({ t: "release", id: this.id } satisfies ClaimMsg); + }; + + private onBusMessage(msg: ClaimMsg) { + if (msg.t === "claim") { + const theirs: Claim = { ts: msg.ts, id: msg.id }; + if (newerClaim(theirs, this.latest)) { + this.latest = theirs; + this.onChange?.(); + } + } else if (msg.id === this.latest.id) { + // The streaming tab went away: void its claim, then reclaim if visible. + // Concurrent reclaims all broadcast and newest-claim-wins settles on one. + this.latest = { ts: 0, id: this.id }; + this.claimIfVisible(); + } + } + + private readonly onVisibilityChange = () => { + if (document.hidden) { + this.hiddenTimer ??= setTimeout(() => { + this.hiddenTimer = null; + this.visible = false; + this.onChange?.(); + }, this.hiddenGraceMs); + } else { + if (this.hiddenTimer !== null) clearTimeout(this.hiddenTimer); + this.hiddenTimer = null; + this.visible = true; + this.claim(); + } + }; +} + +class TabStreamImpl implements TabStream { + private readonly backoffMs: number; + private readonly hiddenGraceMs: number; + private readonly subscribers = new Set>(); + + // Non-null while running; aborting tears down the loop and eligibility. + private run: AbortController | null = null; + private eligibility: Eligibility | null = null; + + constructor(private readonly opts: TabStreamOpts) { + this.backoffMs = opts.backoffMs ?? DEFAULT_BACKOFF_MS; + this.hiddenGraceMs = opts.hiddenGraceMs ?? HIDDEN_GRACE_MS; + } + + subscribe(sub: StreamSubscriber): () => void { + this.subscribers.add(sub); + if (!this.run) { + this.run = new AbortController(); + this.eligibility = new Eligibility(this.opts.name, this.hiddenGraceMs); + void this.main(this.run.signal, this.eligibility); + } + // A late joiner missed everything so far; have it load initial state now. + this.fireConnectOrResync(sub); + return () => { + this.subscribers.delete(sub); + if (this.subscribers.size === 0) { + this.run?.abort(); + this.run = null; + this.eligibility?.stop(); + this.eligibility = null; + } + }; + } + + // The whole algorithm: while eligible, hold the stream; park when another + // tab claims it or this one is hidden too long; reload around every gap. + private async main(signal: AbortSignal, eligibility: Eligibility) { + while (!signal.aborted) { + if (!eligibility.eligible) { + await eligibility.changed(signal); + continue; + } + + // Aborts the connection when eligibility may have been lost or on + // teardown; the loop re-checks the real state either way. + const conn = new AbortController(); + const abortConn = () => conn.abort(); + signal.addEventListener("abort", abortConn, { once: true }); + eligibility.onChange = () => { + if (!eligibility.eligible) conn.abort(); + }; + try { + this.deliverConnectOrResync(); // rule 3: reload around every gap + for await (const msg of this.opts.connect(conn.signal)) { + this.deliverMessage(msg); + } + } catch (err) { + if (!conn.signal.aborted) { + console.warn(`[tabStream:${this.opts.name}] stream error`, err); + } + } finally { + eligibility.onChange = null; + signal.removeEventListener("abort", abortConn); + } + + // Parked or torn down: the top of the loop handles both. Otherwise the + // stream itself ended or errored, so back off before reconnecting. + if (signal.aborted || !eligibility.eligible) continue; + await abortableDelay(this.backoffMs, signal); + } + } + + private deliverMessage(msg: T) { + for (const sub of this.subscribers) { + try { + sub.onMessage(msg); + } catch (e) { + console.warn(`[tabStream:${this.opts.name}] onMessage threw`, e); + } + } + } + + private deliverConnectOrResync() { + for (const sub of this.subscribers) this.fireConnectOrResync(sub); + } + + private fireConnectOrResync(sub: StreamSubscriber) { + try { + sub.onConnectOrResync(); + } catch (e) { + console.warn(`[tabStream:${this.opts.name}] onConnectOrResync threw`, e); + } + } +} + +/** A setTimeout that also resolves early if `signal` aborts. */ +function abortableDelay(ms: number, signal: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal.aborted) return resolve(); + const cleanup = () => { + clearTimeout(timer); + signal.removeEventListener("abort", onAbort); + }; + const onAbort = () => { + cleanup(); + resolve(); + }; + const timer = setTimeout(() => { + cleanup(); + resolve(); + }, ms); + signal.addEventListener("abort", onAbort, { once: true }); + }); +} diff --git a/webui/src/state/peerStates.ts b/webui/src/state/peerStates.ts index 8d637104..8fb25967 100644 --- a/webui/src/state/peerStates.ts +++ b/webui/src/state/peerStates.ts @@ -1,9 +1,8 @@ import { useEffect, useState } from "react"; -import { PeerState, PeerStateSchema } from "../../gen/ts/v1sync/syncservice_pb"; +import { PeerState } from "../../gen/ts/v1sync/syncservice_pb"; import { Config } from "../../gen/ts/v1/config_pb"; -import { fromBinary, toBinary } from "@bufbuild/protobuf"; import { syncStateService } from "../api/client"; -import { createSharedStream } from "../api/streams/sharedStream"; +import { createTabStream } from "../api/streams/tabStream"; import { useConfig } from "../app/provider"; // Type intersection to combine properties from Repo and RepoMetadata @@ -27,15 +26,13 @@ const notifySubscribers = () => { }, 100); }; -// Peer-sync stream, shared across tabs (see sharedStream). Started only while a -// consumer wants it and, via the config gate in useSyncStates, never opens for -// the common no-peer user. -const peerStatesStream = createSharedStream({ +// Peer-sync stream, held by the most recently focused tab (see tabStream). +// Started only while a consumer wants it and, via the config gate in +// useSyncStates, never opens for the common no-peer user. +const peerStatesStream = createTabStream({ name: "backrest:peer-states", connect: (signal) => syncStateService.getPeerSyncStatesStream({ subscribe: true }, { signal }), - encode: (state) => toBinary(PeerStateSchema, state), - decode: (bytes) => fromBinary(PeerStateSchema, bytes), }); // Reload the full current peer-state set from the API. A non-subscribing stream