fix: http/1.1 limit on long polling connections prevents >3 tabs from receiving operation streams

Author: garethgeorge

webui/src/api/logState.ts

@@ -1,6 +1,5 @@
 import {
   Operation,
-  OperationEvent,
   OperationEventType,
   OperationStatus,
 } from "../../gen/ts/v1/operations_pb";
@@ -9,11 +8,7 @@ import {
   GetOperationsRequestSchema,
   OpSelector,
 } from "../../gen/ts/v1/service_pb";
-import {
-  getOperations,
-  subscribeToOperations,
-  unsubscribeFromOperations,
-} from "./oplog";
+import { getOperations, operationsStream } from "./oplog";
 import { create } from "@bufbuild/protobuf";
 
 type Subscriber = (
@@ -28,35 +23,23 @@ export const syncStateFromRequest = (
   onError?: (e: Error) => void,
   onInitialLoad?: () => void,
 ): (() => void) => {
-  getOperations(req)
-    .then((res) => {
-      state.add(...res);
-      onInitialLoad?.();
-    })
-    .catch((e) => {
-      if (onError) {
-        onError(e);
-      }
-      onInitialLoad?.();
-    });
-
-  const cbHelper = (event?: OperationEvent, err?: Error) => {
-    if (err) {
-      if (onError) {
-        onError(err);
-      }
-      state.reset();
+  // The stream carries only deltas, which may have gaps across a connect or
+  // resync, so (re)load the full set from the API and rebuild from scratch.
+  const load = () => {
+    state.reset();
+    getOperations(req)
+      .then((res) => {
+        state.add(...res);
+        onInitialLoad?.();
+      })
+      .catch((e) => {
+        onError?.(e);
+        onInitialLoad?.();
+      });
+  };
 
-      getOperations(req)
-        .then((res) => {
-          state.add(...res);
-        })
-        .catch((e) => {
-          if (onError) {
-            onError(e);
-          }
-        });
-    } else if (event) {
+  return operationsStream.subscribe({
+    onMessage: (event) => {
       switch (event.event.case) {
         case "createdOperations":
         case "updatedOperations":
@@ -70,13 +53,9 @@ export const syncStateFromRequest = (
           state.removeIDs(...event.event.value.values);
           break;
       }
-    }
-  };
-
-  subscribeToOperations(cbHelper);
-  return () => {
-    unsubscribeFromOperations(cbHelper);
-  };
+    },
+    onConnectOrResync: load,
+  });
 };
 
 // getStatus returns the status of the last N operations that belong to a single snapshot.

webui/src/api/oplog.ts

@@ -1,45 +1,22 @@
 import {
   Operation,
   OperationEvent,
-  OperationEventType,
+  OperationEventSchema,
   OperationStatus,
 } from "../../gen/ts/v1/operations_pb";
-import { GetOperationsRequest, OpSelector } from "../../gen/ts/v1/service_pb";
-import {
-  BackupProgressEntry,
-  ResticSnapshot,
-  RestoreProgressEntry,
-} from "../../gen/ts/v1/restic_pb";
-import { EmptySchema } from "../../gen/ts/types/value_pb";
-import { create } from "@bufbuild/protobuf";
+import { GetOperationsRequest } from "../../gen/ts/v1/service_pb";
+import { fromBinary, toBinary } from "@bufbuild/protobuf";
 import { backrestService } from "./client";
-import { useEffect, useState } from "react";
-
-const subscribers: ((event?: OperationEvent, err?: Error) => void)[] = [];
+import { createSharedStream } from "./streams/sharedStream";
 
-// Start fetching and emitting operations.
-(async () => {
-  while (true) {
-    let nextConnWaitUntil = new Date().getTime() + 5000;
-    try {
-      for await (const event of backrestService.getOperationEvents({})) {
-        console.log("operation event", event);
-        subscribers.forEach((subscriber) => subscriber(event, undefined));
-      }
-    } catch (e: any) {
-      console.warn("operations stream died with exception: ", e);
-      let waitRemaining = nextConnWaitUntil - new Date().getTime();
-      if (waitRemaining < 0) {
-        subscribers.forEach((subscriber) =>
-          subscriber(undefined, e instanceof Error ? e : new Error(String(e))),
-        );
-      }
-    }
-    await new Promise((accept, _) =>
-      setTimeout(accept, nextConnWaitUntil - new Date().getTime()),
-    );
-  }
-})();
+// 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>({
+  name: "backrest:operations",
+  connect: (signal) => backrestService.getOperationEvents({}, { signal }),
+  encode: (event) => toBinary(OperationEventSchema, event),
+  decode: (bytes) => fromBinary(OperationEventSchema, bytes),
+});
 
 export const getOperations = async (
   req: GetOperationsRequest,
@@ -48,30 +25,6 @@ export const getOperations = async (
   return opList.operations || [];
 };
 
-export const subscribeToOperations = (
-  callback: (event?: OperationEvent, err?: Error) => void,
-) => {
-  subscribers.push(callback);
-  console.log(
-    "subscribed to operations, subscriber count: ",
-    subscribers.length,
-  );
-};
-
-export const unsubscribeFromOperations = (
-  callback: (event?: OperationEvent, err?: Error) => void,
-) => {
-  const index = subscribers.indexOf(callback);
-  if (index > -1) {
-    subscribers[index] = subscribers[subscribers.length - 1];
-    subscribers.pop();
-  }
-  console.log(
-    "unsubscribed from operations, subscriber count: ",
-    subscribers.length,
-  );
-};
-
 export const shouldHideOperation = (operation: Operation) => {
   // Hide successful backups with no snapshot ID (e.g., --skip-if-unchanged)
   // but NOT dry run backups which intentionally have no snapshot

webui/src/api/resourceStatus.tsx

@@ -1,19 +1,15 @@
 import { useEffect, useState } from "react";
-import {
-  Operation,
-  OperationEvent,
-  OperationStatus,
-} from "../../gen/ts/v1/operations_pb";
+import { Operation, OperationStatus } from "../../gen/ts/v1/operations_pb";
 import { OpSelector } from "../../gen/ts/v1/service_pb";
-import { subscribeToOperations, unsubscribeFromOperations } from "./oplog";
+import { operationsStream } from "./oplog";
 import { getStatusForSelector, matchSelector } from "./logState";
 import { debounce } from "../lib/util";
 
 // Module-level shared state: all registered selectors and their cached statuses.
 const selectors = new Map<string, { selector: OpSelector; refCount: number }>();
 const statuses = new Map<string, OperationStatus>();
 const listeners = new Set<() => void>();
-let subscribed = false;
+let unsubscribe: (() => void) | null = null;
 
 const notify = () => {
   for (const l of listeners) l();
@@ -62,19 +58,6 @@ const refreshMatching = (ops: Operation[]) => {
   }
 };
 
-const handleEvent = (event?: OperationEvent, _err?: Error) => {
-  if (!event || !event.event) return;
-  switch (event.event.case) {
-    case "createdOperations":
-    case "updatedOperations":
-      refreshMatching(event.event.value.operations);
-      break;
-    case "deletedOperations":
-      refreshAll();
-      break;
-  }
-};
-
 const register = (selector: OpSelector): string => {
   const key = JSON.stringify(selector);
   const existing = selectors.get(key);
@@ -84,9 +67,22 @@ const register = (selector: OpSelector): string => {
   }
   selectors.set(key, { selector, refCount: 1 });
   fetchStatus(key, selector);
-  if (!subscribed) {
-    subscribed = true;
-    subscribeToOperations(handleEvent);
+  if (!unsubscribe) {
+    unsubscribe = operationsStream.subscribe({
+      onMessage: (event) => {
+        switch (event.event.case) {
+          case "createdOperations":
+          case "updatedOperations":
+            refreshMatching(event.event.value.operations);
+            break;
+          case "deletedOperations":
+            refreshAll();
+            break;
+        }
+      },
+      // May have missed events; re-fetch every tracked selector.
+      onConnectOrResync: () => refreshAll(),
+    });
   }
   return key;
 };
@@ -99,9 +95,9 @@ const unregister = (key: string) => {
     selectors.delete(key);
     statuses.delete(key);
     notify();
-    if (selectors.size === 0 && subscribed) {
-      subscribed = false;
-      unsubscribeFromOperations(handleEvent);
+    if (selectors.size === 0 && unsubscribe) {
+      unsubscribe();
+      unsubscribe = null;
       flushPending.cancel();
       pendingKeys.clear();
     }