Remove Mutation tombstones (#1354)

Author: schmidt-sebastian

packages/firestore/src/local/memory_mutation_queue.ts

@@ -71,6 +71,10 @@ export class MemoryMutationQueue implements MutationQueue {
     );
 
     const batchIndex = this.indexOfExistingBatchId(batchId, 'acknowledged');
+    assert(
+      batchIndex === 0,
+      'Can only acknowledge the first batch in the mutation queue'
+    );
 
     // Verify that the batch in the queue is the one to be acknowledged.
     const check = this.mutationQueue[batchIndex];
@@ -81,10 +85,6 @@ export class MemoryMutationQueue implements MutationQueue {
         ', got batch ' +
         check.batchId
     );
-    assert(
-      !check.isTombstone(),
-      "Can't acknowledge a previously removed batch"
-    );
 
     this.highestAcknowledgedBatchId = batchId;
     this.lastStreamToken = streamToken;
@@ -149,17 +149,15 @@ export class MemoryMutationQueue implements MutationQueue {
   ): PersistencePromise<DocumentKeySet | null> {
     const mutationBatch = this.findMutationBatch(batchId);
     assert(mutationBatch != null, 'Failed to find local mutation batch.');
-    return PersistencePromise.resolve(
-      !mutationBatch!.isTombstone() ? mutationBatch!.keys() : null
+    return PersistencePromise.resolve<DocumentKeySet | null>(
+      mutationBatch!.keys()
     );
   }
 
   getNextMutationBatchAfterBatchId(
     transaction: PersistenceTransaction,
     batchId: BatchId
   ): PersistencePromise<MutationBatch | null> {
-    const size = this.mutationQueue.length;
-
     // 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.
@@ -168,24 +166,16 @@ export class MemoryMutationQueue implements MutationQueue {
     // The requested batchId may still be out of range so normalize it to the
     // start of the queue.
     const rawIndex = this.indexOfBatchId(nextBatchId);
-    let index = rawIndex < 0 ? 0 : rawIndex;
-
-    // Finally return the first non-tombstone batch.
-    for (; index < size; index++) {
-      const batch = this.mutationQueue[index];
-      if (!batch.isTombstone()) {
-        return PersistencePromise.resolve<MutationBatch | null>(batch);
-      }
-    }
-    return PersistencePromise.resolve<MutationBatch | null>(null);
+    const index = rawIndex < 0 ? 0 : rawIndex;
+    return PersistencePromise.resolve(
+      this.mutationQueue.length > index ? this.mutationQueue[index] : null
+    );
   }
 
   getAllMutationBatches(
     transaction: PersistenceTransaction
   ): PersistencePromise<MutationBatch[]> {
-    return PersistencePromise.resolve(
-      this.getAllLiveMutationBatchesBeforeIndex(this.mutationQueue.length)
-    );
+    return PersistencePromise.resolve(this.mutationQueue.slice());
   }
 
   getAllMutationBatchesAffectingDocumentKey(
@@ -298,27 +288,10 @@ export class MemoryMutationQueue implements MutationQueue {
     // first entry in the queue.
     const batchIndex = this.indexOfExistingBatchId(batch.batchId, 'removed');
     assert(
-      this.mutationQueue[batchIndex].batchId === batch.batchId,
-      'Removed batches must exist in the queue'
+      batchIndex === 0,
+      'Can only remove the first entry of the mutation queue'
     );
-
-    // Only actually remove batches if removing at the front of the queue.
-    // Previously rejected batches may have left tombstones in the queue, so
-    // expand the removal range to include any tombstones.
-    if (batchIndex === 0) {
-      let endIndex = 1;
-      for (; endIndex < this.mutationQueue.length; endIndex++) {
-        const batch = this.mutationQueue[endIndex];
-        if (!batch.isTombstone()) {
-          break;
-        }
-      }
-      this.mutationQueue.splice(0, endIndex);
-    } else {
-      this.mutationQueue[batchIndex] = this.mutationQueue[
-        batchIndex
-      ].toTombstone();
-    }
+    this.mutationQueue.shift();
 
     let references = this.batchesByDocumentKey;
     return PersistencePromise.forEach(batch.mutations, mutation => {
@@ -358,26 +331,6 @@ export class MemoryMutationQueue implements MutationQueue {
     return PersistencePromise.resolve();
   }
 
-  /**
-   * A private helper that collects all the mutations batches in the queue up to
-   * but not including the given endIndex. All tombstones in the queue are
-   * excluded.
-   */
-  private getAllLiveMutationBatchesBeforeIndex(
-    endIndex: number
-  ): MutationBatch[] {
-    const result: MutationBatch[] = [];
-
-    for (let i = 0; i < endIndex; i++) {
-      const batch = this.mutationQueue[i];
-      if (!batch.isTombstone()) {
-        result.push(batch);
-      }
-    }
-
-    return result;
-  }
-
   /**
    * Finds the index of the given batchId in the mutation queue and asserts that
    * the resulting index is within the bounds of the queue.
@@ -430,6 +383,6 @@ export class MemoryMutationQueue implements MutationQueue {
 
     const batch = this.mutationQueue[index];
     assert(batch.batchId === batchId, 'If found batch must match');
-    return batch.isTombstone() ? null : batch;
+    return batch;
   }
 }

packages/firestore/src/model/mutation_batch.ts

@@ -39,7 +39,9 @@ export class MutationBatch {
     public batchId: BatchId,
     public localWriteTime: Timestamp,
     public mutations: Mutation[]
-  ) {}
+  ) {
+    assert(mutations.length > 0, 'Cannot create an empty mutation batch');
+  }
 
   /**
    * Applies all the mutations in this MutationBatch to the specified document
@@ -129,23 +131,6 @@ export class MutationBatch {
       misc.arrayEquals(this.mutations, other.mutations)
     );
   }
-
-  /**
-   * Returns true if this mutation batch has already been removed from the
-   * mutation queue.
-   *
-   * Note that not all implementations of the MutationQueue necessarily use
-   * tombstones as part of their implementation and generally speaking no code
-   * outside the mutation queues should really care about this.
-   */
-  isTombstone(): boolean {
-    return this.mutations.length === 0;
-  }
-
-  /** Converts this batch into a tombstone */
-  toTombstone(): MutationBatch {
-    return new MutationBatch(this.batchId, this.localWriteTime, []);
-  }
 }
 
 /** The result of applying a mutation batch to the backend. */

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

@@ -109,26 +109,22 @@ function genericMutationQueueTests(): void {
   }
 
   /**
-   * Removes entries from from the given a batches and returns them.
+   * Removes the first n entries from the given batches and returns them.
    *
-   * @param holes An array of indexes in the batches array; in increasing order.
-   * Indexes are relative to the original state of the batches array, not any
-   * intermediate state that might occur.
+   * @param n The number of batches to remove.
    * @param batches The array to mutate, removing entries from it.
    * @return A new array containing all the entries that were removed from
    * batches.
    */
-  async function makeHolesInBatches(
-    holes: number[],
+  async function removeFirstBatches(
+    n: number,
     batches: MutationBatch[]
   ): Promise<MutationBatch[]> {
     const removed: MutationBatch[] = [];
-    for (let i = 0; i < holes.length; i++) {
-      const index = holes[i] - i;
-      const batch = batches[index];
+    for (let i = 0; i < n; i++) {
+      const batch = batches[0];
       await mutationQueue.removeMutationBatch(batch);
-
-      batches.splice(index, 1);
+      batches.shift();
       removed.push(batch);
     }
     return removed;
@@ -145,10 +141,10 @@ function genericMutationQueueTests(): void {
     const batch2 = await addMutationBatch();
     expect(await mutationQueue.countBatches()).to.equal(2);
 
-    await mutationQueue.removeMutationBatch(batch2);
+    await mutationQueue.removeMutationBatch(batch1);
     expect(await mutationQueue.countBatches()).to.equal(1);
 
-    await mutationQueue.removeMutationBatch(batch1);
+    await mutationQueue.removeMutationBatch(batch2);
     expect(await mutationQueue.countBatches()).to.equal(0);
   });
 
@@ -168,7 +164,7 @@ function genericMutationQueueTests(): void {
     expect(notFound).to.be.null;
 
     const batches = await createBatches(10);
-    const removed = await makeHolesInBatches([2, 6, 7], batches);
+    const removed = await removeFirstBatches(3, batches);
 
     // After removing, a batch should not be found
     for (const batch of removed) {
@@ -189,10 +185,7 @@ function genericMutationQueueTests(): void {
 
   it('can getNextMutationBatchAfterBatchId()', async () => {
     const batches = await createBatches(10);
-
-    // This is an array of successors assuming the removals below will happen:
-    const afters = [batches[3], batches[8], batches[8]];
-    const removed = await makeHolesInBatches([2, 6, 7], batches);
+    const removed = await removeFirstBatches(3, batches);
 
     for (let i = 0; i < batches.length - 1; i++) {
       const current = batches[i];
@@ -205,7 +198,7 @@ function genericMutationQueueTests(): void {
 
     for (let i = 0; i < removed.length; i++) {
       const current = removed[i];
-      const next = afters[i];
+      const next = batches[0];
       const found = await mutationQueue.getNextMutationBatchAfterBatchId(
         current.batchId
       );
@@ -372,20 +365,20 @@ function genericMutationQueueTests(): void {
     expectEqualArrays(found, batches);
     expect(found.length).to.equal(6);
 
-    await mutationQueue.removeMutationBatch(batches[batches.length - 1]);
-    batches.splice(batches.length - 1, 1);
+    await mutationQueue.removeMutationBatch(batches[0]);
+    batches.shift();
     expect(await mutationQueue.countBatches()).to.equal(5);
 
     found = await mutationQueue.getAllMutationBatches();
     expectEqualArrays(found, batches);
     expect(found.length).to.equal(5);
 
-    await mutationQueue.removeMutationBatch(batches[3]);
-    batches.splice(3, 1);
+    await mutationQueue.removeMutationBatch(batches[0]);
+    batches.shift();
     expect(await mutationQueue.countBatches()).to.equal(4);
 
-    await mutationQueue.removeMutationBatch(batches[1]);
-    batches.splice(1, 1);
+    await mutationQueue.removeMutationBatch(batches[0]);
+    batches.shift();
     expect(await mutationQueue.countBatches()).to.equal(3);
 
     found = await mutationQueue.getAllMutationBatches();