Implement limbo resolution throttling.

packages/firestore/src/core/sync_engine.ts

@@ -148,9 +148,13 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     q.canonicalId()
   );
   private queriesByTarget: { [targetId: number]: Query[] } = {};
+  /** The keys of documents that are in limbo for which we haven't yet started a limbo resolution query. */
+  private limboListenQueue: DocumentKey[] = [];
+  /** Keeps track of the target ID for each document that is in limbo with an active target. */
   private limboTargetsByKey = new SortedMap<DocumentKey, TargetId>(
     DocumentKey.comparator
   );
+  /** Keeps track of the information about an active limbo resolution for each active target ID that was started for the purpose of limbo resolution. */
   private limboResolutionsByTarget: {
     [targetId: number]: LimboResolution;
   } = {};
@@ -174,7 +178,8 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     private remoteStore: RemoteStore,
     // PORTING NOTE: Manages state synchronization in multi-tab environments.
     private sharedClientState: SharedClientState,
-    private currentUser: User
+    private currentUser: User,
+    private maxConcurrentLimboResolutions: number = 100
   ) {}
 
   // Only used for testing.
@@ -486,6 +491,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
       // So go ahead and remove it from bookkeeping.
       this.limboTargetsByKey = this.limboTargetsByKey.remove(limboKey);
       delete this.limboResolutionsByTarget[targetId];
+      this.pumpLimboResolutionListenQueue();
 
       // TODO(klimt): We really only should do the following on permission
       // denied errors, but we don't have the cause code here.
@@ -740,6 +746,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     this.remoteStore.unlisten(limboTargetId);
     this.limboTargetsByKey = this.limboTargetsByKey.remove(key);
     delete this.limboResolutionsByTarget[limboTargetId];
+    this.pumpLimboResolutionListenQueue();
   }
 
   private updateTrackedLimbos(
@@ -770,29 +777,63 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     const key = limboChange.key;
     if (!this.limboTargetsByKey.get(key)) {
       log.debug(LOG_TAG, 'New document in limbo: ' + key);
+      this.limboListenQueue.push(key);
+      this.pumpLimboResolutionListenQueue();
+    }
+  }
+
+  /**
+   * Starts listens for documents in limbo that are enqueued for resolution.
+   *
+   * When a document goes into limbo it is enqueued for resolution. This method
+   * repeatedly removes entries from the limbo resolution queue and starts a
+   * listen for them until either (1) the queue is empty, meaning that all
+   * documents that were in limbo either have active listens or have been
+   * resolved, or (2) the maximum number of concurrent limbo resolution listens
+   * has been reached.
+   *
+   * This method is invoked every time an entry is added to the limbo
+   * resolution queue and every time that a limbo resolution listen completes
+   * (either successfully or unsuccessfully). This ensures that all documents in
+   * limbo are eventually resolved.
+   *
+   * A maximum number of concurrent limbo resolution listens was implemented to
+   * prevent an unbounded number of active limbo resolution listens that can
+   * exhaust server resources and result in "resource exhausted" errors.
+   */
+  private pumpLimboResolutionListenQueue(): void {
+    while (
+      this.limboListenQueue.length > 0 &&
+      this.limboTargetsByKey.size < this.maxConcurrentLimboResolutions
+    ) {
+      const key = this.limboListenQueue.shift()!;
       const limboTargetId = this.limboTargetIdGenerator.next();
-      const query = Query.atPath(key.path);
       this.limboResolutionsByTarget[limboTargetId] = new LimboResolution(key);
+      this.limboTargetsByKey = this.limboTargetsByKey.insert(
+        key,
+        limboTargetId
+      );
       this.remoteStore.listen(
         new TargetData(
-          query.toTarget(),
+          Query.atPath(key.path).toTarget(),
           limboTargetId,
           TargetPurpose.LimboResolution,
           ListenSequence.INVALID
         )
       );
-      this.limboTargetsByKey = this.limboTargetsByKey.insert(
-        key,
-        limboTargetId
-      );
     }
   }
 
   // Visible for testing
-  currentLimboDocs(): SortedMap<DocumentKey, TargetId> {
+  activeLimboDocumentResolutions(): SortedMap<DocumentKey, TargetId> {
     return this.limboTargetsByKey;
   }
 
+  // Visible for testing
+  documentsEnqueuedForLimboResolution(): DocumentKey[] {
+    return this.limboListenQueue;
+  }
+
   private async emitNewSnapsAndNotifyLocalStore(
     changes: MaybeDocumentMap,
     remoteEvent?: RemoteEvent