Remove Mutation tombstones (#2052)

* Remove Mutation tombstones

* Review comments

Author: schmidt-sebastian

Firestore/Example/Tests/Local/FSTMutationQueueTests.mm

@@ -71,10 +71,10 @@ - (void)testCountBatches {
     FSTMutationBatch *batch2 = [self addMutationBatch];
     XCTAssertEqual(2, [self batchCount]);
 
-    [self.mutationQueue removeMutationBatch:batch2];
+    [self.mutationQueue removeMutationBatch:batch1];
     XCTAssertEqual(1, [self batchCount]);
 
-    [self.mutationQueue removeMutationBatch:batch1];
+    [self.mutationQueue removeMutationBatch:batch2];
     XCTAssertEqual(0, [self batchCount]);
     XCTAssertTrue([self.mutationQueue isEmpty]);
   });
@@ -98,10 +98,9 @@ - (void)testAcknowledgeBatchID {
     XCTAssertEqual([self.mutationQueue highestAcknowledgedBatchID], kFSTBatchIDUnknown);
 
     [self.mutationQueue acknowledgeBatch:batch1 streamToken:nil];
-    [self.mutationQueue acknowledgeBatch:batch2 streamToken:nil];
-    XCTAssertEqual([self.mutationQueue highestAcknowledgedBatchID], batch2.batchID);
-
     [self.mutationQueue removeMutationBatch:batch1];
+
+    [self.mutationQueue acknowledgeBatch:batch2 streamToken:nil];
     XCTAssertEqual([self.mutationQueue highestAcknowledgedBatchID], batch2.batchID);
 
     [self.mutationQueue removeMutationBatch:batch2];
@@ -139,10 +138,10 @@ - (void)testHighestAcknowledgedBatchIDNeverExceedsNextBatchID {
 
   self.persistence.run("testHighestAcknowledgedBatchIDNeverExceedsNextBatchID", [&]() {
     [self.mutationQueue acknowledgeBatch:batch1 streamToken:nil];
+    [self.mutationQueue removeMutationBatch:batch1];
     [self.mutationQueue acknowledgeBatch:batch2 streamToken:nil];
     XCTAssertEqual([self.mutationQueue highestAcknowledgedBatchID], batch2.batchID);
 
-    [self.mutationQueue removeMutationBatch:batch1];
     [self.mutationQueue removeMutationBatch:batch2];
     XCTAssertEqual([self.mutationQueue highestAcknowledgedBatchID], batch2.batchID);
   });
@@ -186,7 +185,7 @@ - (void)testLookupMutationBatch {
     XCTAssertNil(notFound);
 
     NSMutableArray<FSTMutationBatch *> *batches = [self createBatches:10];
-    NSArray<FSTMutationBatch *> *removed = [self makeHoles:@[ @2, @6, @7 ] inBatches:batches];
+    NSArray<FSTMutationBatch *> *removed = [self removeFirstBatches:3 inBatches:batches];
 
     // After removing, a batch should not be found
     for (NSUInteger i = 0; i < removed.count; i++) {
@@ -211,10 +210,7 @@ - (void)testNextMutationBatchAfterBatchID {
 
   self.persistence.run("testNextMutationBatchAfterBatchID", [&]() {
     NSMutableArray<FSTMutationBatch *> *batches = [self createBatches:10];
-
-    // This is an array of successors assuming the removals below will happen:
-    NSArray<FSTMutationBatch *> *afters = @[ batches[3], batches[8], batches[8] ];
-    NSArray<FSTMutationBatch *> *removed = [self makeHoles:@[ @2, @6, @7 ] inBatches:batches];
+    NSArray<FSTMutationBatch *> *removed = [self removeFirstBatches:3 inBatches:batches];
 
     for (NSUInteger i = 0; i < batches.count - 1; i++) {
       FSTMutationBatch *current = batches[i];
@@ -225,7 +221,7 @@ - (void)testNextMutationBatchAfterBatchID {
 
     for (NSUInteger i = 0; i < removed.count; i++) {
       FSTMutationBatch *current = removed[i];
-      FSTMutationBatch *next = afters[i];
+      FSTMutationBatch *next = batches[0];
       FSTMutationBatch *found = [self.mutationQueue nextMutationBatchAfterBatchID:current.batchID];
       XCTAssertEqual(found.batchID, next.batchID);
     }
@@ -262,26 +258,6 @@ - (void)testNextMutationBatchAfterBatchIDSkipsAcknowledgedBatches {
   });
 }
 
-- (void)testAllMutationBatchesThroughBatchID {
-  if ([self isTestBaseClass]) return;
-
-  self.persistence.run("testAllMutationBatchesThroughBatchID", [&]() {
-    NSMutableArray<FSTMutationBatch *> *batches = [self createBatches:10];
-    [self makeHoles:@[ @2, @6, @7 ] inBatches:batches];
-
-    NSArray<FSTMutationBatch *> *found, *expected;
-
-    found = [self.mutationQueue allMutationBatchesThroughBatchID:batches[0].batchID - 1];
-    XCTAssertEqualObjects(found, (@[]));
-
-    for (NSUInteger i = 0; i < batches.count; i++) {
-      found = [self.mutationQueue allMutationBatchesThroughBatchID:batches[i].batchID];
-      expected = [batches subarrayWithRange:NSMakeRange(0, i + 1)];
-      XCTAssertEqualObjects(found, expected, @"for index %lu", (unsigned long)i);
-    }
-  });
-}
-
 - (void)testAllMutationBatchesAffectingDocumentKey {
   if ([self isTestBaseClass]) return;
 
@@ -443,12 +419,11 @@ - (void)testRemoveMutationBatches {
     [self.mutationQueue removeMutationBatch:batches[0]];
     [batches removeObjectAtIndex:0];
 
-    FSTMutationBatch *last = batches[batches.count - 1];
     XCTAssertEqual([self batchCount], 9);
 
     NSArray<FSTMutationBatch *> *found;
 
-    found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
+    found = [self.mutationQueue allMutationBatches];
     XCTAssertEqualObjects(found, batches);
     XCTAssertEqual(found.count, 9);
 
@@ -458,35 +433,35 @@ - (void)testRemoveMutationBatches {
     [batches removeObjectsInRange:NSMakeRange(0, 3)];
     XCTAssertEqual([self batchCount], 6);
 
-    found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
+    found = [self.mutationQueue allMutationBatches];
     XCTAssertEqualObjects(found, batches);
     XCTAssertEqual(found.count, 6);
 
-    [self.mutationQueue removeMutationBatch:batches[batches.count - 1]];
-    [batches removeObjectAtIndex:batches.count - 1];
+    [self.mutationQueue removeMutationBatch:batches[0]];
+    [batches removeObjectAtIndex:0];
     XCTAssertEqual([self batchCount], 5);
 
-    found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
+    found = [self.mutationQueue allMutationBatches];
     XCTAssertEqualObjects(found, batches);
     XCTAssertEqual(found.count, 5);
 
-    [self.mutationQueue removeMutationBatch:batches[3]];
-    [batches removeObjectAtIndex:3];
+    [self.mutationQueue removeMutationBatch:batches[0]];
+    [batches removeObjectAtIndex:0];
     XCTAssertEqual([self batchCount], 4);
 
-    [self.mutationQueue removeMutationBatch:batches[1]];
-    [batches removeObjectAtIndex:1];
+    [self.mutationQueue removeMutationBatch:batches[0]];
+    [batches removeObjectAtIndex:0];
     XCTAssertEqual([self batchCount], 3);
 
-    found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
+    found = [self.mutationQueue allMutationBatches];
     XCTAssertEqualObjects(found, batches);
     XCTAssertEqual(found.count, 3);
     XCTAssertFalse([self.mutationQueue isEmpty]);
 
     for (FSTMutationBatch *batch in batches) {
       [self.mutationQueue removeMutationBatch:batch];
     }
-    found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
+    found = [self.mutationQueue allMutationBatches];
     XCTAssertEqualObjects(found, @[]);
     XCTAssertEqual(found.count, 0);
     XCTAssertTrue([self.mutationQueue isEmpty]);
@@ -554,24 +529,21 @@ - (NSUInteger)batchCount {
 }
 
 /**
- * Removes entries from from the given @a batches and returns them.
+ * Removes the first n entries from the 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 @a batches.
  */
-- (NSArray<FSTMutationBatch *> *)makeHoles:(NSArray<NSNumber *> *)holes
-                                 inBatches:(NSMutableArray<FSTMutationBatch *> *)batches {
-  NSMutableArray<FSTMutationBatch *> *removed = [NSMutableArray array];
-  for (NSUInteger i = 0; i < holes.count; i++) {
-    NSUInteger index = holes[i].unsignedIntegerValue - i;
-    FSTMutationBatch *batch = batches[index];
+- (NSArray<FSTMutationBatch *> *)removeFirstBatches:(NSUInteger)n
+                                          inBatches:(NSMutableArray<FSTMutationBatch *> *)batches {
+  NSArray<FSTMutationBatch *> *removed = [batches subarrayWithRange:NSMakeRange(0, n)];
+  [batches removeObjectsInRange:NSMakeRange(0, n)];
+
+  [removed enumerateObjectsUsingBlock:^(FSTMutationBatch *batch, NSUInteger idx, BOOL *stop) {
     [self.mutationQueue removeMutationBatch:batch];
+  }];
 
-    [batches removeObjectAtIndex:index];
-    [removed addObject:batch];
-  }
   return removed;
 }
 

Firestore/Source/Local/FSTLevelDBMutationQueue.mm

@@ -327,29 +327,6 @@ - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(BatchId)batchID {
   return [self decodedMutationBatch:it->value()];
 }
 
-- (NSArray<FSTMutationBatch *> *)allMutationBatchesThroughBatchID:(BatchId)batchID {
-  std::string userKey = LevelDbMutationKey::KeyPrefix(_userID);
-
-  auto it = _db.currentTransaction->NewIterator();
-  it->Seek(userKey);
-
-  NSMutableArray *result = [NSMutableArray array];
-  LevelDbMutationKey rowKey;
-  for (; it->Valid() && rowKey.Decode(it->key()); it->Next()) {
-    if (rowKey.user_id() != _userID) {
-      // End of this user's mutations
-      break;
-    } else if (rowKey.batch_id() > batchID) {
-      // This mutation is past what we're looking for
-      break;
-    }
-
-    [result addObject:[self decodedMutationBatch:it->value()]];
-  }
-
-  return result;
-}
-
 - (NSArray<FSTMutationBatch *> *)allMutationBatchesAffectingDocumentKey:
     (const DocumentKey &)documentKey {
   // Scan the document-mutation index starting with a prefix starting with the given documentKey.

Firestore/Source/Local/FSTMemoryMutationQueue.mm

@@ -131,12 +131,12 @@ - (void)acknowledgeBatch:(FSTMutationBatch *)batch streamToken:(nullable NSData
               "Mutation batchIDs must be acknowledged in order");
 
   NSInteger batchIndex = [self indexOfExistingBatchID:batchID action:@"acknowledged"];
+  HARD_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.
   FSTMutationBatch *check = queue[(NSUInteger)batchIndex];
   HARD_ASSERT(batchID == check.batchID, "Queue ordering failure: expected batch %s, got batch %s",
               batchID, check.batchID);
-  HARD_ASSERT(![check isTombstone], "Can't acknowledge a previously removed batch");
 
   self.highestAcknowledgedBatchID = batchID;
   self.lastStreamToken = streamToken;
@@ -182,12 +182,11 @@ - (nullable FSTMutationBatch *)lookupMutationBatch:(BatchId)batchID {
 
   FSTMutationBatch *batch = queue[(NSUInteger)index];
   HARD_ASSERT(batch.batchID == batchID, "If found batch must match");
-  return [batch isTombstone] ? nil : batch;
+  return batch;
 }
 
 - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(BatchId)batchID {
   NSMutableArray<FSTMutationBatch *> *queue = self.queue;
-  NSUInteger count = queue.count;
 
   // All batches with batchID <= self.highestAcknowledgedBatchID have been acknowledged so the
   // first unacknowledged batch after batchID will have a batchID larger than both of these values.
@@ -196,37 +195,11 @@ - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(BatchId)batchID {
   // The requested batchID may still be out of range so normalize it to the start of the queue.
   NSInteger rawIndex = [self indexOfBatchID:nextBatchID];
   NSUInteger index = rawIndex < 0 ? 0 : (NSUInteger)rawIndex;
-
-  // Finally return the first non-tombstone batch.
-  for (; index < count; index++) {
-    FSTMutationBatch *batch = queue[index];
-    if (![batch isTombstone]) {
-      return batch;
-    }
-  }
-
-  return nil;
+  return queue.count > index ? queue[index] : nil;
 }
 
 - (NSArray<FSTMutationBatch *> *)allMutationBatches {
-  return [self allLiveMutationBatchesBeforeIndex:self.queue.count];
-}
-
-- (NSArray<FSTMutationBatch *> *)allMutationBatchesThroughBatchID:(BatchId)batchID {
-  NSMutableArray<FSTMutationBatch *> *queue = self.queue;
-  NSUInteger count = queue.count;
-
-  NSInteger endIndex = [self indexOfBatchID:batchID];
-  if (endIndex < 0) {
-    endIndex = 0;
-  } else if (endIndex >= count) {
-    endIndex = count;
-  } else {
-    // The endIndex is in the queue so increment to pull everything in the queue including it.
-    endIndex += 1;
-  }
-
-  return [self allLiveMutationBatchesBeforeIndex:(NSUInteger)endIndex];
+  return [[self queue] copy];
 }
 
 - (NSArray<FSTMutationBatch *> *)allMutationBatchesAffectingDocumentKey:
@@ -328,31 +301,14 @@ - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(BatchId)batchID {
 
 - (void)removeMutationBatch:(FSTMutationBatch *)batch {
   NSMutableArray<FSTMutationBatch *> *queue = self.queue;
-  NSUInteger queueCount = queue.count;
   BatchId batchID = batch.batchID;
 
   // Find the position of the first batch for removal. This need not be the first entry in the
   // queue.
   NSUInteger batchIndex = [self indexOfExistingBatchID:batchID action:@"removed"];
-  HARD_ASSERT(queue[batchIndex].batchID == batchID, "Removed batches must exist in the 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) {
-    int endIndex = 1;
-    for (; endIndex < queueCount; endIndex++) {
-      FSTMutationBatch *batch = queue[endIndex];
-      if (![batch isTombstone]) {
-        break;
-      }
-    }
+  HARD_ASSERT(batchIndex == 0, "Can only remove the first entry of the mutation queue");
 
-    NSUInteger length = endIndex - batchIndex;
-    [queue removeObjectsInRange:NSMakeRange(batchIndex, length)];
-
-  } else {
-    queue[batchIndex] = [queue[batchIndex] toTombstone];
-  }
+  [queue removeObjectAtIndex:0];
 
   // Remove entries from the index too.
   FSTImmutableSortedSet<FSTDocumentReference *> *references = self.batchesByDocumentKey;
@@ -388,25 +344,6 @@ - (BOOL)containsKey:(const DocumentKey &)key {
 
 #pragma mark - Helpers
 
-/**
- * A private helper that collects all the mutation batches in the queue up to but not including
- * the given endIndex. All tombstones in the queue are excluded.
- */
-- (NSArray<FSTMutationBatch *> *)allLiveMutationBatchesBeforeIndex:(NSUInteger)endIndex {
-  NSMutableArray<FSTMutationBatch *> *result = [NSMutableArray arrayWithCapacity:endIndex];
-
-  NSUInteger index = 0;
-  for (FSTMutationBatch *batch in self.queue) {
-    if (index++ >= endIndex) break;
-
-    if (![batch isTombstone]) {
-      [result addObject:batch];
-    }
-  }
-
-  return result;
-}
-
 /**
  * Finds the index of the given batchID in the mutation queue. This operation is O(1).
  *

Firestore/Source/Local/FSTMutationQueue.h

@@ -92,21 +92,6 @@ NS_ASSUME_NONNULL_BEGIN
 // cheaply, we should replace this.
 - (NSArray<FSTMutationBatch *> *)allMutationBatches;
 
-/**
- * Finds all mutations with a batchID less than or equal to the given batchID.
- *
- * Generally the caller should be asking for the next unacknowledged batchID and the number of
- * acknowledged batches should be very small when things are functioning well.
- *
- * @param batchID The batch to search through.
- *
- * @return an NSArray containing all batches with matching batchIDs.
- */
-// TODO(mcg): This should really return NSEnumerator and the caller should be adjusted to only
-// loop through these once.
-- (NSArray<FSTMutationBatch *> *)allMutationBatchesThroughBatchID:
-    (firebase::firestore::model::BatchId)batchID;
-
 /**
  * Finds all mutation batches that could @em possibly affect the given document key. Not all
  * mutations in a batch will necessarily affect the document key, so when looping through the

Firestore/Source/Model/FSTMutationBatch.h

@@ -87,18 +87,6 @@ extern const firebase::firestore::model::BatchId kFSTBatchIDUnknown;
     applyToLocalDocument:(FSTMaybeDocument *_Nullable)maybeDoc
              documentKey:(const firebase::firestore::model::DocumentKey &)documentKey;
 
-/**
- * Returns YES if this mutation batch has already been removed from the mutation queue.
- *
- * Note that not all implementations of the FSTMutationQueue necessarily use tombstones as a part
- * of their implementation and generally speaking no code outside the mutation queues should really
- * care about this.
- */
-- (BOOL)isTombstone;
-
-/** Converts this batch to a tombstone. */
-- (FSTMutationBatch *)toTombstone;
-
 /** Returns the set of unique keys referenced by all mutations in the batch. */
 - (firebase::firestore::model::DocumentKeySet)keys;