fix: simplified workaround to long polling connection limits, avoids issues with suspended tabs on mobile

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<OperationEvent>({
+// 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<OperationEvent>({
   name: "backrest:operations",
   connect: (signal) => backrestService.getOperationEvents({}, { signal }),
-  encode: (event) => toBinary(OperationEventSchema, event),
-  decode: (bytes) => fromBinary(OperationEventSchema, bytes),
 });
 
 export const getOperations = async (

webui/src/api/streams/sharedStream.test.ts

@@ -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<string, Set<MockBroadcastChannel>>();
-
-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<unknown>;
-  resolve: (v: unknown) => void;
-  reject: (e: unknown) => void;
-  signal?: AbortSignal;
-}
-
-class MockLockManager {
-  private held = new Set<string>();
-  private queues = new Map<string, QueuedLock[]>();
-
-  request(
-    name: string,
-    opts: { signal?: AbortSignal },
-    cb: () => Promise<unknown>,
-  ): Promise<unknown> {
-    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<void> =>
-  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<Msg> & {
-  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<Msg>({
-      name: "t1",
-      connect: connectA,
-      encode,
-      decode,
-    });
-    const streamB = createSharedStream<Msg>({
-      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<Msg>({
-      name: "t2",
-      connect: connectA,
-      encode,
-      decode,
-    });
-    const streamB = createSharedStream<Msg>({
-      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<Msg>({
-      name: "t5",
-      connect: connectA,
-      encode,
-      decode,
-    });
-    const streamB = createSharedStream<Msg>({
-      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<Msg>({
-      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<Msg>({
-      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;
-  });
-});

webui/src/api/streams/sharedStream.ts

@@ -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<T> {
-  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<T> {
-  /** Used for both the lock name and the BroadcastChannel name. */
-  name: string;
-  connect: (signal: AbortSignal) => AsyncIterable<T>;
-  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<T> {
-  /** First subscribe starts the stream, last unsubscribe tears it down. */
-  subscribe(sub: StreamSubscriber<T>): () => 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<T>(
-  opts: SharedStreamOpts<T>,
-): SharedStream<T> {
-  return new SharedStreamImpl(opts);
-}
-
-class SharedStreamImpl<T> implements SharedStream<T> {
-  private readonly backoffMs: number;
-  private readonly shared: boolean;
-  private readonly eligibility: Eligibility;
-
-  private readonly subscribers = new Set<StreamSubscriber<T>>();
-
-  // 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<T>) {
-    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<T>): () => 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<T>) {
-    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<typeof setTimeout> | null = null;
-  private readonly waiters: Array<() => void> = [];
-  private readonly pending = new Set<AbortController>();
-  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<void> {
-    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<void> {
-  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 });
-  });
-}

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<string, Set<MockBroadcastChannel>>();
+
+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<void> =>
+  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<Msg> & {
+  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<Msg>({
+      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<Msg>({ name: "t2", connect: connectA });
+    const streamB = createTabStream<Msg>({ 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<Msg>({ 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<Msg>({ name: "t4", connect: connectA });
+    const streamB = createTabStream<Msg>({ 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<Msg>({
+      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<Msg>({ name: "t6", connect: connectA });
+    const streamB = createTabStream<Msg>({ 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();
+  });
+});

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<T> {
+  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<T> {
+  /** Names the claim channel; shared by all tabs streaming the same thing. */
+  name: string;
+  connect: (signal: AbortSignal) => AsyncIterable<T>;
+  backoffMs?: number;
+  /** Overridable for tests. */
+  hiddenGraceMs?: number;
+}
+
+export interface TabStream<T> {
+  /** First subscribe starts the stream, last unsubscribe tears it down. */
+  subscribe(sub: StreamSubscriber<T>): () => void;
+}
+
+export function createTabStream<T>(opts: TabStreamOpts<T>): TabStream<T> {
+  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<typeof setTimeout> | 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<void> {
+    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<T> implements TabStream<T> {
+  private readonly backoffMs: number;
+  private readonly hiddenGraceMs: number;
+  private readonly subscribers = new Set<StreamSubscriber<T>>();
+
+  // 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<T>) {
+    this.backoffMs = opts.backoffMs ?? DEFAULT_BACKOFF_MS;
+    this.hiddenGraceMs = opts.hiddenGraceMs ?? HIDDEN_GRACE_MS;
+  }
+
+  subscribe(sub: StreamSubscriber<T>): () => 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<T>) {
+    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<void> {
+  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 });
+  });
+}

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<PeerState>({
+// 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<PeerState>({
   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