Refactor pendingWrites / write pipeline. (#217)

* Rename receiveListener => listener.

This matches JS and I'm not sure why "receive" was added to the name. The
listener is used for received data but also for close notifications, etc.

* Refactor pendingWrites / write pipeline.

[Port of firebase/firebase-js-sdk#972]

No functional changes. Just renames, moves, added comments, etc.

* pendingWrites => writePipeline
* canWriteMutations() renamed canAddToWritePipeline()
* commit() => addToWritePipeline()
* lastBatchSeen removed since we can compute it on demand from
  writePipeline.
* Removed cleanUpWriteStreamState() and instead inlined it in
  disableNetworkInternal() since I didn't like the non-symmetry with
  cleanUpWatchStreamState() which is called every time the stream closes.
* Lots of comment cleanup.

Author: mikelehen

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java

@@ -152,7 +152,7 @@ public void run() {
   private State state = State.Initial;
   private ClientCall<ReqT, RespT> call;
   final ExponentialBackoff backoff;
-  CallbackT receiveListener;
+  CallbackT listener;
   StreamObserver streamObserver;
 
   AbstractStream(
@@ -190,7 +190,7 @@ public boolean isOpen() {
   @Override
   public void start(CallbackT listener) {
     workerQueue.verifyIsCurrentThread();
-    hardAssert(receiveListener == null, "Receive listener still set");
+    hardAssert(this.listener == null, "Receive listener still set");
     hardAssert(call == null, "Last call still set");
     hardAssert(idleTimer == null, "Idle timer still set");
 
@@ -201,7 +201,7 @@ public void start(CallbackT listener) {
 
     hardAssert(state == State.Initial, "Already started");
 
-    receiveListener = listener;
+    this.listener = listener;
 
     streamObserver = new StreamObserver();
     call = firestoreChannel.runBidiStreamingRpc(methodDescriptor, streamObserver);
@@ -219,7 +219,7 @@ public void start(CallbackT listener) {
           // in the auth state.
           if (state == State.Auth) {
             state = State.Open;
-            receiveListener.onOpen();
+            this.listener.onOpen();
           }
         });
   }
@@ -238,7 +238,7 @@ public void start(CallbackT listener) {
    * 'State.Stop'.
    *
    * @param finalState the intended state of the stream after closing.
-   * @param status the status to emit to the receiveListener.
+   * @param status the status to emit to the listener.
    */
   private void close(State finalState, Status status) {
     workerQueue.verifyIsCurrentThread();
@@ -292,14 +292,14 @@ private void close(State finalState, Status status) {
       call = null;
     }
 
-    // This state must be assigned before calling receiveListener.onClose to allow the callback to
+    // This state must be assigned before calling listener.onClose to allow the callback to
     // inhibit backoff or otherwise manipulate the state in its non-started state.
     this.state = finalState;
 
     // If the caller explicitly requested a stream stop, don't notify them of a closing stream (it
     // could trigger undesirable recovery logic, etc.).
-    CallbackT receiveListener = this.receiveListener;
-    this.receiveListener = null;
+    CallbackT receiveListener = this.listener;
+    this.listener = null;
     if (finalState != State.Stop) {
       receiveListener.onClose(status);
     }

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/RemoteStore.java

@@ -40,11 +40,11 @@
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import java.util.ArrayDeque;
+import java.util.Deque;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Queue;
 
 /**
  * RemoteStore handles all interaction with the backend through a simple, clean interface. This
@@ -129,9 +129,21 @@ public interface RemoteStoreCallback {
   @Nullable private WriteStream writeStream;
   @Nullable private WatchChangeAggregator watchChangeAggregator;
 
-  private int lastBatchSeen;
-
-  private final Queue<MutationBatch> pendingWrites;
+  /**
+   * A list of up to MAX_PENDING_WRITES writes that we have fetched from the LocalStore via
+   * fillWritePipeline() and have or will send to the write stream.
+   *
+   * <p>Whenever writePipeline.length > 0 the RemoteStore will attempt to start or restart the write
+   * stream. When the stream is established the writes in the pipeline will be sent in order.
+   *
+   * <p>Writes remain in writePipeline until they are acknowledged by the backend and thus will
+   * automatically be re-sent if the stream is interrupted / restarted before they're acknowledged.
+   *
+   * <p>Write responses from the backend are linked to their originating request purely based on
+   * order, and so we can just poll() writes from the front of the writePipeline as we receive
+   * responses.
+   */
+  private final Deque<MutationBatch> writePipeline;
 
   public RemoteStore(
       RemoteStoreCallback remoteStoreCallback,
@@ -143,8 +155,7 @@ public RemoteStore(
     this.datastore = datastore;
 
     listenTargets = new HashMap<>();
-    lastBatchSeen = MutationBatch.UNKNOWN;
-    pendingWrites = new ArrayDeque<>();
+    writePipeline = new ArrayDeque<>();
 
     onlineStateTracker =
         new OnlineStateTracker(workerQueue, remoteStoreCallback::handleOnlineStateChange);
@@ -188,7 +199,10 @@ private void disableNetworkInternal() {
     writeStream.stop();
 
     cleanUpWatchStreamState();
-    cleanUpWriteStreamState();
+    if (!writePipeline.isEmpty()) {
+      Logger.debug(LOG_TAG, "Stopping write stream with %d pending writes", writePipeline.size());
+      writePipeline.clear();
+    }
 
     writeStream = null;
     watchStream = null;
@@ -290,7 +304,7 @@ private void sendUnwatchRequest(int targetId) {
    * pending writes.
    */
   private boolean shouldStartWriteStream() {
-    return isNetworkEnabled() && !writeStream.isStarted() && !pendingWrites.isEmpty();
+    return isNetworkEnabled() && !writeStream.isStarted() && !writePipeline.isEmpty();
   }
 
   /**
@@ -301,12 +315,6 @@ private boolean shouldStartWatchStream() {
     return isNetworkEnabled() && !watchStream.isStarted() && !listenTargets.isEmpty();
   }
 
-  private void cleanUpWriteStreamState() {
-    lastBatchSeen = MutationBatch.UNKNOWN;
-    Logger.debug(LOG_TAG, "Stopping write stream with " + pendingWrites.size() + " pending writes");
-    pendingWrites.clear();
-  }
-
   private void cleanUpWatchStreamState() {
     // If the connection is closed then we'll never get a snapshot version for the accumulated
     // changes and so we'll never be able to complete the batch. When we start up again the server
@@ -488,49 +496,46 @@ private void processTargetError(WatchTargetChange targetChange) {
   // Write Stream
 
   /**
-   * Notifies that there are new mutations to process in the queue. This is typically called by
-   * SyncEngine after it has sent mutations to LocalStore.
+   * Attempts to fill our write pipeline with writes from the LocalStore.
    *
-   * <p>In response the remote store will pull mutations from the local store until the datastore
-   * instance reports that it cannot accept further in-progress writes. This mechanism serves to
-   * maintain a pipeline of in-flight requests between the Datastore and the server that applies
-   * them.
+   * <p>Called internally to bootstrap or refill the write pipeline and by SyncEngine whenever there
+   * are new mutations to process.
+   *
+   * <p>Starts the write stream if necessary.
    */
   public void fillWritePipeline() {
     if (isNetworkEnabled()) {
-      while (canWriteMutations()) {
-        MutationBatch batch = localStore.getNextMutationBatch(lastBatchSeen);
+      int lastBatchIdRetrieved =
+          writePipeline.isEmpty() ? MutationBatch.UNKNOWN : writePipeline.getLast().getBatchId();
+      while (canAddToWritePipeline()) {
+        MutationBatch batch = localStore.getNextMutationBatch(lastBatchIdRetrieved);
         if (batch == null) {
           break;
         }
-        commitBatch(batch);
+        addToWritePipeline(batch);
+        lastBatchIdRetrieved = batch.getBatchId();
       }
 
-      if (pendingWrites.isEmpty()) {
-        writeStream.markIdle();
-      }
+      writeStream.markIdle();
     }
   }
 
   /**
-   * Returns true if the backend can accept additional write requests.
-   *
-   * <p>When sending mutations to the write stream (e.g. in fillWritePipeline()), call this method
-   * first to check if more mutations can be sent.
-   *
-   * <p>Currently the only thing that can prevent the backend from accepting write requests is if
-   * there are too many requests already outstanding. As writes complete the backend will be able to
-   * accept more.
+   * Returns true if we can add to the write pipeline (i.e. it is not full and the network is
+   * enabled).
    */
-  private boolean canWriteMutations() {
-    return isNetworkEnabled() && pendingWrites.size() < MAX_PENDING_WRITES;
+  private boolean canAddToWritePipeline() {
+    return isNetworkEnabled() && writePipeline.size() < MAX_PENDING_WRITES;
   }
 
-  private void commitBatch(MutationBatch mutationBatch) {
-    hardAssert(canWriteMutations(), "commitBatch called when mutations can't be written");
-    lastBatchSeen = mutationBatch.getBatchId();
+  /**
+   * Queues additional writes to be sent to the write stream, sending them immediately if the write
+   * stream is established, else starting the write stream if it is not yet started.
+   */
+  private void addToWritePipeline(MutationBatch mutationBatch) {
+    hardAssert(canAddToWritePipeline(), "addToWritePipeline called when pipeline is full");
 
-    pendingWrites.add(mutationBatch);
+    writePipeline.add(mutationBatch);
 
     if (shouldStartWriteStream()) {
       startWriteStream();
@@ -576,19 +581,8 @@ private void handleWriteStreamHandshakeComplete() {
     // Record the stream token.
     localStore.setLastStreamToken(writeStream.getLastStreamToken());
 
-    /*
-     * Drain pending writes.
-     *
-     * Note that at this point pendingWrites contains mutations that have already been accepted by
-     * fillWritePipeline/commitBatch. If the pipeline is full, canWriteMutations will be false,
-     * despite the fact that we actually need to send mutations over.
-     *
-     * This also means that this method indirectly respects the limits imposed by canWriteMutations
-     * since writes can't be added to the pendingWrites array when canWriteMutations is false. If
-     * the limits imposed by canWriteMutations actually protect us from DOSing ourselves then those
-     * limits won't be exceeded here and we'll continue to make progress.
-     */
-    for (MutationBatch batch : pendingWrites) {
+    // Send the write pipeline now that stream is established.
+    for (MutationBatch batch : writePipeline) {
       writeStream.writeMutations(batch.getMutations());
     }
   }
@@ -599,8 +593,8 @@ private void handleWriteStreamHandshakeComplete() {
   private void handleWriteStreamMutationResults(
       SnapshotVersion commitVersion, List<MutationResult> results) {
     // This is a response to a write containing mutations and should be correlated to the first
-    // pending write.
-    MutationBatch batch = pendingWrites.poll();
+    // write in our write pipeline.
+    MutationBatch batch = writePipeline.poll();
 
     MutationBatchResult mutationBatchResult =
         MutationBatchResult.create(batch, commitVersion, results, writeStream.getLastStreamToken());
@@ -617,7 +611,7 @@ private void handleWriteStreamClose(Status status) {
 
     // If the write stream closed due to an error, invoke the error callbacks if there are pending
     // writes.
-    if (!status.isOk() && !pendingWrites.isEmpty()) {
+    if (!status.isOk() && !writePipeline.isEmpty()) {
       // TODO: handle UNAUTHENTICATED status, see go/firestore-client-errors
       if (writeStream.isHandshakeComplete()) {
         // This error affects the actual writes
@@ -658,7 +652,7 @@ private void handleWriteError(Status status) {
     if (Datastore.isPermanentWriteError(status)) {
       // If this was a permanent error, the request itself was the problem so it's not going
       // to succeed if we resend it.
-      MutationBatch batch = pendingWrites.poll();
+      MutationBatch batch = writePipeline.poll();
 
       // 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 restart

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/WatchStream.java

@@ -100,6 +100,6 @@ public void onNext(com.google.firestore.v1beta1.ListenResponse listenResponse) {
 
     WatchChange watchChange = serializer.decodeWatchChange(listenResponse);
     SnapshotVersion snapshotVersion = serializer.decodeVersionFromListenResponse(listenResponse);
-    receiveListener.onWatchChange(snapshotVersion, watchChange);
+    listener.onWatchChange(snapshotVersion, watchChange);
   }
 }

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/WriteStream.java

@@ -161,7 +161,7 @@ public void onNext(WriteResponse response) {
       // The first response is the handshake response
       handshakeComplete = true;
 
-      receiveListener.onHandshakeComplete();
+      listener.onHandshakeComplete();
     } else {
       // A successful first write response means the stream is healthy,
       // Note, that we could consider a successful handshake healthy, however,
@@ -176,7 +176,7 @@ public void onNext(WriteResponse response) {
         com.google.firestore.v1beta1.WriteResult result = response.getWriteResults(i);
         results.add(serializer.decodeMutationResult(result, commitVersion));
       }
-      receiveListener.onWriteResponse(commitVersion, results);
+      listener.onWriteResponse(commitVersion, results);
     }
   }
 }

firebase-firestore/src/test/java/com/google/firebase/firestore/remote/MockDatastore.java

@@ -55,7 +55,7 @@ private class MockWatchStream extends WatchStream {
     @Override
     public void start(WatchStream.Callback callback) {
       hardAssert(!open, "Trying to start already started watch stream");
-      receiveListener = callback;
+      listener = callback;
       streamObserver = new WatchStream.StreamObserver();
       open = true;
       callback.onOpen();
@@ -105,7 +105,7 @@ public void unwatchTarget(int targetId) {
     /** Injects a stream failure as though it had come from the backend. */
     void failStream(Status status) {
       open = false;
-      receiveListener.onClose(status);
+      listener.onClose(status);
     }
 
     /** Injects a watch change as though it had come from the backend. */
@@ -129,7 +129,7 @@ void writeWatchChange(WatchChange change, SnapshotVersion snapshotVersion) {
           snapshotVersion = SnapshotVersion.NONE;
         }
       }
-      receiveListener.onWatchChange(snapshotVersion, change);
+      listener.onWatchChange(snapshotVersion, change);
     }
   }
 
@@ -146,7 +146,7 @@ private class MockWriteStream extends WriteStream {
     @Override
     public void start(WriteStream.Callback callback) {
       hardAssert(!open, "Trying to start already started write stream");
-      receiveListener = callback;
+      listener = callback;
       streamObserver = new WriteStream.StreamObserver();
       handshakeComplete = false;
       open = true;
@@ -169,7 +169,7 @@ public void writeHandshake() {
       hardAssert(!handshakeComplete, "Handshake already completed");
       writeStreamRequestCount += 1;
       handshakeComplete = true;
-      receiveListener.onHandshakeComplete();
+      listener.onHandshakeComplete();
     }
 
     @Override
@@ -180,13 +180,13 @@ public void writeMutations(List<Mutation> mutations) {
 
     /** Injects a write ack as though it had come from the backend in response to a write. */
     void ackWrite(SnapshotVersion commitVersion, List<MutationResult> results) {
-      receiveListener.onWriteResponse(commitVersion, results);
+      listener.onWriteResponse(commitVersion, results);
     }
 
     /** Injects a stream failure as though it had come from the backend. */
     void failStream(Status status) {
       open = false;
-      receiveListener.onClose(status);
+      listener.onClose(status);
     }
 
     /** Returns a previous write that had been "sent to the backend". */