feat: Automatically recover from frozen live streams (#2569)

- Closes: #2099

src/utils/concurrency/keyed-subscription-manager.ts

@@ -1,7 +1,7 @@
 import PQueue from 'p-queue';
 
-type UnsubscribeCallback = () => Promise<void>;
-type SubscribeCallback = () => Promise<UnsubscribeCallback>;
+type AsyncUnsubscribeCallback = () => Promise<void>;
+type SubscribeCallback = () => Promise<AsyncUnsubscribeCallback>;
 
 /**
  * Extracts the key from a request. Used by `KeyedSubscriptionManager` and any
@@ -22,7 +22,7 @@ export type GetKeyCallback<R, K> = (request: R) => K;
  */
 export class KeyedSubscriptionManager<K, R> {
   private _requests: R[] = [];
-  private _unsubscribers = new Map<K, UnsubscribeCallback>();
+  private _unsubscribers = new Map<K, AsyncUnsubscribeCallback>();
   private _queues = new Map<K, PQueue>();
   private _getKeyCallback: GetKeyCallback<R, K>;
 

src/utils/initializer/initializer.ts

@@ -1,6 +1,9 @@
 import { allPromises } from '../basic';
 
-type InitializationCallback = () => Promise<void>;
+// An initializer returns `false` when it could not complete its work in this
+// attempt; the aspect is then left uninitialized so a later attempt retries.
+// Any other result marks the aspect initialized.
+type InitializationCallback = () => Promise<boolean | void>;
 
 /**
  * Manages initialization state and runs initializers.
@@ -21,31 +24,34 @@ export class Initializer {
 
   public async initializeMultipleIfNecessary(
     aspects: Record<string, InitializationCallback>,
-  ): Promise<void> {
-    await allPromises(
+  ): Promise<boolean> {
+    const results = await allPromises(
       Object.entries(aspects),
       async ([aspect, options]) => await this.initializeIfNecessary(aspect, options),
     );
+    return results.every(Boolean);
   }
 
+  // Returns whether the aspect is initialized once this attempt completes.
   public async initializeIfNecessary(
     aspect: string,
     initializer?: InitializationCallback,
-  ): Promise<void> {
+  ): Promise<boolean> {
     if (this._initialized.has(aspect)) {
-      return;
+      return true;
     }
     const generation = this._generation.get(aspect) ?? 0;
-    if (initializer) {
-      await initializer();
+    if (initializer && (await initializer()) === false) {
+      return false;
     }
     // If `uninitialize()` ran while we were awaiting, a newer attempt has taken
     // over -- throw this result away (don't mark it initialized) so a stale
     // result can't leave the card stuck, and a fresh attempt runs next time.
     if ((this._generation.get(aspect) ?? 0) !== generation) {
-      return;
+      return false;
     }
     this._initialized.add(aspect);
+    return true;
   }
 
   public uninitialize(aspect: string): void {