Implement limbo resolution throttling (#2790)

Author: dconeybe

packages/firestore/src/core/firestore_client.ts

@@ -53,6 +53,7 @@ import { OnlineState, OnlineStateSource } from './types';
 import { ViewSnapshot } from './view_snapshot';
 
 const LOG_TAG = 'FirestoreClient';
+const MAX_CONCURRENT_LIMBO_RESOLUTIONS = 100;
 
 /** DOMException error code constants. */
 const DOM_EXCEPTION_INVALID_STATE = 11;
@@ -369,7 +370,8 @@ export class FirestoreClient {
           this.localStore,
           this.remoteStore,
           this.sharedClientState,
-          user
+          user,
+          MAX_CONCURRENT_LIMBO_RESOLUTIONS
         );
 
         this.sharedClientState.onlineStateHandler = sharedClientStateOnlineStateChangedHandler;

packages/firestore/src/core/sync_engine.ts

@@ -148,10 +148,23 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     q.canonicalId()
   );
   private queriesByTarget: { [targetId: number]: Query[] } = {};
-  private limboTargetsByKey = new SortedMap<DocumentKey, TargetId>(
+  /**
+   * The keys of documents that are in limbo for which we haven't yet started a
+   * limbo resolution query.
+   */
+  private enqueuedLimboResolutions: DocumentKey[] = [];
+  /**
+   * Keeps track of the target ID for each document that is in limbo with an
+   * active target.
+   */
+  private activeLimboTargetsByKey = new SortedMap<DocumentKey, TargetId>(
     DocumentKey.comparator
   );
-  private limboResolutionsByTarget: {
+  /**
+   * 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 activeLimboResolutionsByTarget: {
     [targetId: number]: LimboResolution;
   } = {};
   private limboDocumentRefs = new ReferenceSet();
@@ -174,7 +187,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
   ) {}
 
   // Only used for testing.
@@ -400,7 +414,9 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
       const changes = await this.localStore.applyRemoteEvent(remoteEvent);
       // Update `receivedDocument` as appropriate for any limbo targets.
       objUtils.forEach(remoteEvent.targetChanges, (targetId, targetChange) => {
-        const limboResolution = this.limboResolutionsByTarget[Number(targetId)];
+        const limboResolution = this.activeLimboResolutionsByTarget[
+          Number(targetId)
+        ];
         if (limboResolution) {
           // Since this is a limbo resolution lookup, it's for a single document
           // and it could be added, modified, or removed, but not a combination.
@@ -479,13 +495,16 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     // PORTING NOTE: Multi-tab only.
     this.sharedClientState.updateQueryState(targetId, 'rejected', err);
 
-    const limboResolution = this.limboResolutionsByTarget[targetId];
+    const limboResolution = this.activeLimboResolutionsByTarget[targetId];
     const limboKey = limboResolution && limboResolution.key;
     if (limboKey) {
       // Since this query failed, we won't want to manually unlisten to it.
       // So go ahead and remove it from bookkeeping.
-      this.limboTargetsByKey = this.limboTargetsByKey.remove(limboKey);
-      delete this.limboResolutionsByTarget[targetId];
+      this.activeLimboTargetsByKey = this.activeLimboTargetsByKey.remove(
+        limboKey
+      );
+      delete this.activeLimboResolutionsByTarget[targetId];
+      this.pumpEnqueuedLimboResolutions();
 
       // TODO(klimt): We really only should do the following on permission
       // denied errors, but we don't have the cause code here.
@@ -731,15 +750,16 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
   private removeLimboTarget(key: DocumentKey): void {
     // It's possible that the target already got removed because the query failed. In that case,
     // the key won't exist in `limboTargetsByKey`. Only do the cleanup if we still have the target.
-    const limboTargetId = this.limboTargetsByKey.get(key);
+    const limboTargetId = this.activeLimboTargetsByKey.get(key);
     if (limboTargetId === null) {
       // This target already got removed, because the query failed.
       return;
     }
 
     this.remoteStore.unlisten(limboTargetId);
-    this.limboTargetsByKey = this.limboTargetsByKey.remove(key);
-    delete this.limboResolutionsByTarget[limboTargetId];
+    this.activeLimboTargetsByKey = this.activeLimboTargetsByKey.remove(key);
+    delete this.activeLimboResolutionsByTarget[limboTargetId];
+    this.pumpEnqueuedLimboResolutions();
   }
 
   private updateTrackedLimbos(
@@ -768,29 +788,54 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
 
   private trackLimboChange(limboChange: AddedLimboDocument): void {
     const key = limboChange.key;
-    if (!this.limboTargetsByKey.get(key)) {
+    if (!this.activeLimboTargetsByKey.get(key)) {
       logDebug(LOG_TAG, 'New document in limbo: ' + key);
+      this.enqueuedLimboResolutions.push(key);
+      this.pumpEnqueuedLimboResolutions();
+    }
+  }
+
+  /**
+   * Starts listens for documents in limbo that are enqueued for resolution,
+   * subject to a maximum number of concurrent resolutions.
+   *
+   * Without bounding the number of concurrent resolutions, the server can fail
+   * with "resource exhausted" errors which can lead to pathological client
+   * behavior as seen in https://github.com/firebase/firebase-js-sdk/issues/2683.
+   */
+  private pumpEnqueuedLimboResolutions(): void {
+    while (
+      this.enqueuedLimboResolutions.length > 0 &&
+      this.activeLimboTargetsByKey.size < this.maxConcurrentLimboResolutions
+    ) {
+      const key = this.enqueuedLimboResolutions.shift()!;
       const limboTargetId = this.limboTargetIdGenerator.next();
-      const query = Query.atPath(key.path);
-      this.limboResolutionsByTarget[limboTargetId] = new LimboResolution(key);
+      this.activeLimboResolutionsByTarget[limboTargetId] = new LimboResolution(
+        key
+      );
+      this.activeLimboTargetsByKey = this.activeLimboTargetsByKey.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> {
-    return this.limboTargetsByKey;
+  activeLimboDocumentResolutions(): SortedMap<DocumentKey, TargetId> {
+    return this.activeLimboTargetsByKey;
+  }
+
+  // Visible for testing
+  enqueuedLimboDocumentResolutions(): DocumentKey[] {
+    return this.enqueuedLimboResolutions;
   }
 
   private async emitNewSnapsAndNotifyLocalStore(
@@ -936,12 +981,12 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
 
   // PORTING NOTE: Multi-tab only.
   private resetLimboDocuments(): void {
-    objUtils.forEachNumber(this.limboResolutionsByTarget, targetId => {
+    objUtils.forEachNumber(this.activeLimboResolutionsByTarget, targetId => {
       this.remoteStore.unlisten(targetId);
     });
     this.limboDocumentRefs.removeAllReferences();
-    this.limboResolutionsByTarget = [];
-    this.limboTargetsByKey = new SortedMap<DocumentKey, TargetId>(
+    this.activeLimboResolutionsByTarget = [];
+    this.activeLimboTargetsByKey = new SortedMap<DocumentKey, TargetId>(
       DocumentKey.comparator
     );
   }
@@ -1138,7 +1183,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
   }
 
   getRemoteKeysForTarget(targetId: TargetId): DocumentKeySet {
-    const limboResolution = this.limboResolutionsByTarget[targetId];
+    const limboResolution = this.activeLimboResolutionsByTarget[targetId];
     if (limboResolution && limboResolution.receivedDocument) {
       return documentKeySet().add(limboResolution.key);
     } else {