Fix MutationQueue issue resulting in re-sending acknowledged writes. (#559)

getNextMutationBatchAfterBatchId() was not respecting highestAcknowledgedBatchId and therefore we were resending writes if they had been acknowledged but not removed (aka the held write acks case). This showed up when a user disabled / enabled the network as reported here and I've included a spec test to emulate that case: firebase/firebase-ios-sdk#772

Author: mikelehen

packages/firestore/CHANGELOG.md

@@ -3,6 +3,9 @@
   neither succeeds nor fails within 10 seconds, the SDK will consider itself
   "offline", causing get() calls to resolve with cached results, rather than
   continuing to wait.
+- [fixed] Fixed a potential race condition after calling `enableNetwork()` that
+  could result in a "Mutation batchIDs must be acknowledged in order" assertion
+  crash.
 
 # 0.3.2
 - [fixed] Fixed a regression in Firebase JS release 4.9.0 that could in certain

packages/firestore/src/local/indexeddb_mutation_queue.ts

@@ -255,14 +255,20 @@ export class IndexedDbMutationQueue implements MutationQueue {
     transaction: PersistenceTransaction,
     batchId: BatchId
   ): PersistencePromise<MutationBatch | null> {
-    const range = IDBKeyRange.lowerBound(this.keyForBatchId(batchId + 1));
+    // All batches with batchId <= this.metadata.lastAcknowledgedBatchId have
+    // been acknowledged so the first unacknowledged batch after batchID will
+    // have a batchID larger than both of these values.
+    const nextBatchId =
+      Math.max(batchId, this.metadata.lastAcknowledgedBatchId) + 1;
+
+    const range = IDBKeyRange.lowerBound(this.keyForBatchId(nextBatchId));
     let foundBatch: MutationBatch | null = null;
     return mutationsStore(transaction)
       .iterate({ range }, (key, dbBatch, control) => {
         if (dbBatch.userId === this.userId) {
           assert(
-            dbBatch.batchId > batchId,
-            'Should have found mutation after ' + batchId
+            dbBatch.batchId >= nextBatchId,
+            'Should have found mutation after ' + nextBatchId
           );
           foundBatch = this.serializer.fromDbMutationBatch(dbBatch);
         }

packages/firestore/src/local/memory_mutation_queue.ts

@@ -182,11 +182,11 @@ export class MemoryMutationQueue implements MutationQueue {
     // All batches with batchId <= this.highestAcknowledgedBatchId have been
     // acknowledged so the first unacknowledged batch after batchID will have a
     // batchID larger than both of these values.
-    batchId = Math.max(batchId + 1, this.highestAcknowledgedBatchId);
+    const nextBatchId = Math.max(batchId, this.highestAcknowledgedBatchId) + 1;
 
     // The requested batchId may still be out of range so normalize it to the
     // start of the queue.
-    const rawIndex = this.indexOfBatchId(batchId);
+    const rawIndex = this.indexOfBatchId(nextBatchId);
     let index = rawIndex < 0 ? 0 : rawIndex;
 
     // Finally return the first non-tombstone batch.

packages/firestore/test/unit/local/mutation_queue.test.ts

@@ -40,6 +40,7 @@ import {
 
 import * as persistenceHelpers from './persistence_test_helpers';
 import { TestMutationQueue } from './test_mutation_queue';
+import { addEqualityMatcher } from '../../util/equality_matcher';
 
 let persistence: Persistence;
 let mutationQueue: TestMutationQueue;
@@ -120,6 +121,8 @@ describe('IndexedDbMutationQueue', () => {
  * implementations.
  */
 function genericMutationQueueTests() {
+  addEqualityMatcher();
+
   beforeEach(() => {
     mutationQueue = new TestMutationQueue(
       persistence,
@@ -324,6 +327,24 @@ function genericMutationQueueTests() {
     expect(notFound).to.be.null;
   });
 
+  it('getNextMutationBatchAfterBatchId() skips acknowledged batches', async () => {
+    const batches = await createBatches(3);
+    expect(
+      await mutationQueue.getNextMutationBatchAfterBatchId(BATCHID_UNKNOWN)
+    ).to.deep.equal(batches[0]);
+
+    await mutationQueue.acknowledgeBatch(batches[0], emptyByteString());
+    expect(
+      await mutationQueue.getNextMutationBatchAfterBatchId(BATCHID_UNKNOWN)
+    ).to.deep.equal(batches[1]);
+    expect(
+      await mutationQueue.getNextMutationBatchAfterBatchId(batches[0].batchId)
+    ).to.deep.equal(batches[1]);
+    expect(
+      await mutationQueue.getNextMutationBatchAfterBatchId(batches[1].batchId)
+    ).to.deep.equal(batches[2]);
+  });
+
   it('can getAllMutationBatchesThroughBatchId()', async () => {
     const batches = await createBatches(10);
     await makeHolesInBatches([2, 6, 7], batches);

packages/firestore/test/unit/specs/spec_test_runner.ts

@@ -825,17 +825,17 @@ abstract class TestRunner {
   private validateStateExpectations(expectation: StateExpectation): void {
     if (expectation) {
       if ('numOutstandingWrites' in expectation) {
-        expect(this.remoteStore.outstandingWrites()).to.deep.equal(
+        expect(this.remoteStore.outstandingWrites()).to.equal(
           expectation.numOutstandingWrites
         );
       }
       if ('writeStreamRequestCount' in expectation) {
-        expect(this.connection.writeStreamRequestCount).to.deep.equal(
+        expect(this.connection.writeStreamRequestCount).to.equal(
           expectation.writeStreamRequestCount
         );
       }
       if ('watchStreamRequestCount' in expectation) {
-        expect(this.connection.watchStreamRequestCount).to.deep.equal(
+        expect(this.connection.watchStreamRequestCount).to.equal(
           expectation.watchStreamRequestCount
         );
       }

packages/firestore/test/unit/specs/write_spec.test.ts

@@ -394,6 +394,59 @@ describeSpec('Writes:', [], () => {
     );
   });
 
+  specTest(
+    'Held writes are not re-sent after disable/enable network.',
+    [],
+    () => {
+      const query = Query.atPath(path('collection'));
+      const docALocal = doc(
+        'collection/a',
+        0,
+        { v: 1 },
+        { hasLocalMutations: true }
+      );
+      const docA = doc('collection/a', 1000, { v: 1 });
+
+      return (
+        spec()
+          .userListens(query)
+          .watchAcksFull(query, 500)
+          .expectEvents(query, {})
+          .userSets('collection/a', { v: 1 })
+          .expectEvents(query, {
+            hasPendingWrites: true,
+            added: [docALocal]
+          })
+          // ack write but without a watch event.
+          .writeAcks(1000)
+
+          // handshake + write = 2 requests
+          .expectWriteStreamRequestCount(2)
+
+          .disableNetwork()
+          .expectEvents(query, {
+            hasPendingWrites: true,
+            fromCache: true
+          })
+
+          // handshake + write + close = 3 requests
+          .expectWriteStreamRequestCount(3)
+
+          .enableNetwork()
+          .expectActiveTargets({ query, resumeToken: 'resume-token-500' })
+
+          // acked write should /not/ have been resent, so count should still be 3
+          .expectWriteStreamRequestCount(3)
+
+          // Finally watch catches up.
+          .watchAcksFull(query, 2000, docA)
+          .expectEvents(query, {
+            metadata: [docA]
+          })
+      );
+    }
+  );
+
   specTest(
     'Held writes are released when there are no queries left.',
     [],