use AsyncQueue instead of calling setTimeout directly

Author: dconeybe

packages/firestore/src/remote/remote_store.ts

@@ -28,7 +28,7 @@ import { TargetData } from '../local/target_data';
 import { MutationResult } from '../model/mutation';
 import { MutationBatch, MutationBatchResult } from '../model/mutation_batch';
 import { debugAssert, debugCast } from '../util/assert';
-import { AsyncQueue } from '../util/async_queue';
+import { AsyncQueue, DelayedOperation, TimerId } from '../util/async_queue';
 import { ByteString } from '../util/byte_string';
 import { FirestoreError } from '../util/error';
 import { logDebug } from '../util/log';
@@ -77,6 +77,11 @@ const enum OfflineCause {
   Shutdown
 }
 
+export interface WritePipelineEntry {
+  mutationBatch: MutationBatch;
+  sent: boolean;
+}
+
 /**
  * RemoteStore - An interface to remotely stored data, basically providing a
  * wrapper around the Datastore that is more reliable for the rest of the
@@ -124,11 +129,15 @@ class RemoteStoreImpl implements RemoteStore {
    * purely based on order, and so we can just shift() writes from the front of
    * the writePipeline as we receive responses.
    */
-  writePipeline: MutationBatch[] = [];
-
-  sentWrites = new WeakSet<MutationBatch>();
+  writePipeline: WritePipelineEntry[] = [];
 
-  writeTimeoutId: ReturnType<typeof setTimeout> | null = null;
+  /**
+   * The operation that is enqueued to send writes from the `writePipeline` that
+   * have not yet been sent. The delay is used to enable batching of writes,
+   * reducing the overall number of HTTP requests that need to be made to the
+   * backend.
+   */
+  sendWritesOperation: DelayedOperation<void> | null = null;
 
   /**
    * A mapping of watched targets that the client cares about tracking and the
@@ -678,7 +687,7 @@ export async function fillWritePipeline(
   let lastBatchIdRetrieved =
     remoteStoreImpl.writePipeline.length > 0
       ? remoteStoreImpl.writePipeline[remoteStoreImpl.writePipeline.length - 1]
-          .batchId
+          .mutationBatch.batchId
       : BATCHID_UNKNOWN;
 
   while (canAddToWritePipeline(remoteStoreImpl)) {
@@ -736,25 +745,33 @@ function addToWritePipeline(
     canAddToWritePipeline(remoteStoreImpl),
     'addToWritePipeline called when pipeline is full'
   );
-  remoteStoreImpl.writePipeline.push(batch);
+  remoteStoreImpl.writePipeline.push({
+    mutationBatch: batch,
+    sent: false
+  });
 
-  if (remoteStoreImpl.writeTimeoutId !== null) {
+  if (remoteStoreImpl.sendWritesOperation !== null) {
     return;
   }
 
-  remoteStoreImpl.writeTimeoutId = setTimeout(() => {
-    remoteStoreImpl.writeTimeoutId = null;
-    const writeStream = ensureWriteStream(remoteStoreImpl);
-    if (writeStream.isOpen() && writeStream.handshakeComplete) {
-      for (const curBatch of remoteStoreImpl.writePipeline) {
-        if (remoteStoreImpl.sentWrites.has(curBatch)) {
-          continue;
+  remoteStoreImpl.sendWritesOperation =
+    remoteStoreImpl.asyncQueue.enqueueAfterDelay(
+      TimerId.WriteStreamSendDelay,
+      200,
+      () => {
+        remoteStoreImpl.sendWritesOperation = null;
+        const writeStream = ensureWriteStream(remoteStoreImpl);
+        if (writeStream.isOpen() && writeStream.handshakeComplete) {
+          for (const pipelineEntry of remoteStoreImpl.writePipeline) {
+            if (!pipelineEntry.sent) {
+              writeStream.writeMutations(pipelineEntry.mutationBatch.mutations);
+              pipelineEntry.sent = true;
+            }
+          }
         }
-        writeStream.writeMutations(curBatch.mutations);
-        remoteStoreImpl.sentWrites.add(curBatch);
+        return Promise.resolve();
       }
-    }
-  }, 200);
+    );
 }
 
 function shouldStartWriteStream(remoteStoreImpl: RemoteStoreImpl): boolean {
@@ -784,9 +801,9 @@ async function onWriteHandshakeComplete(
 ): Promise<void> {
   const writeStream = ensureWriteStream(remoteStoreImpl);
   // Send the write pipeline now that the stream is established.
-  for (const batch of remoteStoreImpl.writePipeline) {
-    writeStream.writeMutations(batch.mutations);
-    remoteStoreImpl.sentWrites.add(batch);
+  for (const pipelineEntry of remoteStoreImpl.writePipeline) {
+    writeStream.writeMutations(pipelineEntry.mutationBatch.mutations);
+    pipelineEntry.sent = true;
   }
 }
 
@@ -801,8 +818,12 @@ async function onMutationResult(
     remoteStoreImpl.writePipeline.length > 0,
     'Got result for empty write pipeline'
   );
-  const batch = remoteStoreImpl.writePipeline.shift()!;
-  const success = MutationBatchResult.from(batch, commitVersion, results);
+  const pipelineEntry = remoteStoreImpl.writePipeline.shift()!;
+  const success = MutationBatchResult.from(
+    pipelineEntry.mutationBatch,
+    commitVersion,
+    results
+  );
 
   debugAssert(
     !!remoteStoreImpl.remoteSyncer.applySuccessfulWrite,
@@ -853,7 +874,7 @@ async function handleWriteError(
   if (isPermanentWriteError(error.code)) {
     // This was a permanent error, the request itself was the problem
     // so it's not going to succeed if we resend it.
-    const batch = remoteStoreImpl.writePipeline.shift()!;
+    const pipelineEntry = remoteStoreImpl.writePipeline.shift()!;
 
     // In this case it's also unlikely that the server itself is melting
     // down -- this was just a bad request so inhibit backoff on the next
@@ -865,7 +886,10 @@ async function handleWriteError(
       'rejectFailedWrite() not set'
     );
     await executeWithRecovery(remoteStoreImpl, () =>
-      remoteStoreImpl.remoteSyncer.rejectFailedWrite!(batch.batchId, error)
+      remoteStoreImpl.remoteSyncer.rejectFailedWrite!(
+        pipelineEntry.mutationBatch.batchId,
+        error
+      )
     );
 
     // It's possible that with the completion of this mutation

packages/firestore/src/util/async_queue.ts

@@ -48,6 +48,7 @@ export const enum TimerId {
   ListenStreamConnectionBackoff = 'listen_stream_connection_backoff',
   WriteStreamIdle = 'write_stream_idle',
   WriteStreamConnectionBackoff = 'write_stream_connection_backoff',
+  WriteStreamSendDelay = 'write_stream_send_delay',
   HealthCheckTimeout = 'health_check_timeout',
 
   /**