Switch LevelDB MutationQueue and RemoteDocumentCache to use transactions (#968)

* Start work on leveldb transactions

* Style

* Working API. Not plumbed in yet

* Move files into correct place

* Wrangling file locations and associations

* Tests pass

* Add some comments

* style

* Fix copyright

* Rewrite iterator internals to handle deletion-while-iterating. Also add tests for same

* Switch to strings instead of slices

* Style

* More style fixes

* Start switching writegroup over

* Swap out write group tracking for transaction usage

* Style

* Response to feedback before updating docs

* Style

* Add comment

* Initialize version_

* Satisfy the linter

* Start switching writegroup over

* Swap out write group tracking for transaction usage

* Style

* Checkpoint before implementing BatchDescription

* Style

* Initial plumbing for leveldb local parts

* Add model::BatchId

* Port leveldb_key.{h,cc}

* Add string StartsWith

* Add leveldb_key_test.cc to the project

* Revert back to using leveldb::Slice for read/describe

These operations universally operate on keys obtained from leveldb so
it's actually unhelpful to force all the callers to make
absl::string_views from them.

* Everything passing

* Drop unused function

* Style

* STart work on reads

* Swap reads in queryCache to use transactions

* Fix up tests of querycache

* Drop commented out code

* Cleanup

* Style

* Fix up for passing tests

* style

* Renaming

* Style

* Start work on ToString for transactions

* Add ToString() method to LevelDbTransaction

* Style

* lint

* Fix includes, drop runTransaction

* current_transaction -> currentTransaction

* LevelDbTransaction::NewIterator now returns a unique_ptr

* Style

* Revert addition of util::StartsWith

* Add log line

* Style

* Add log line

* Style

* Add debug log line for commits, drop unused BatchDescription

* STart work on reads

* Swap reads in queryCache to use transactions

* Start on remote documents

* Transition mutation queue and remote documents to use transactions

* Style

* Make everything pass

* Make everything pass

* Make it compile

* Style

* Style

* Revert name change, use DefaultReadOptions()

* Style

* Handle iterators returning bad statuses

11 files changed, 175 insertions, 134 deletions

diff --git a/Firestore/Example/Tests/Local/FSTLevelDBMigrationsTests.mm b/Firestore/Example/Tests/Local/FSTLevelDBMigrationsTests.mm
index 4f0332e..ad892d8 100644
--- a/Firestore/Example/Tests/Local/FSTLevelDBMigrationsTests.mm
+++ b/Firestore/Example/Tests/Local/FSTLevelDBMigrationsTests.mm
@@ -21,9 +21,8 @@
 #import "Firestore/Source/Local/FSTLevelDBKey.h"
 #import "Firestore/Source/Local/FSTLevelDBMigrations.h"
 #import "Firestore/Source/Local/FSTLevelDBQueryCache.h"
-#include "leveldb/db.h"
-
 #include "Firestore/core/src/firebase/firestore/util/ordered_code.h"
+#include "leveldb/db.h"
 
 #import "Firestore/Example/Tests/Local/FSTPersistenceTestHelpers.h"
 
diff --git a/Firestore/Example/Tests/Local/FSTMutationQueueTests.mm b/Firestore/Example/Tests/Local/FSTMutationQueueTests.mm
index cad62cf..014adb1 100644
--- a/Firestore/Example/Tests/Local/FSTMutationQueueTests.mm
+++ b/Firestore/Example/Tests/Local/FSTMutationQueueTests.mm
@@ -57,22 +57,32 @@ NS_ASSUME_NONNULL_BEGIN
 - (void)testCountBatches {
   if ([self isTestBaseClass]) return;
 
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Count batches"];
   XCTAssertEqual(0, [self batchCount]);
   XCTAssertTrue([self.mutationQueue isEmpty]);
+  [self.persistence commitGroup:group];
 
   FSTMutationBatch *batch1 = [self addMutationBatch];
+  group = [self.persistence startGroupWithAction:@"Count batches"];
   XCTAssertEqual(1, [self batchCount]);
   XCTAssertFalse([self.mutationQueue isEmpty]);
+  [self.persistence commitGroup:group];
 
   FSTMutationBatch *batch2 = [self addMutationBatch];
+  group = [self.persistence startGroupWithAction:@"Count batches"];
   XCTAssertEqual(2, [self batchCount]);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batch2 ]];
+  group = [self.persistence startGroupWithAction:@"Count batches"];
   XCTAssertEqual(1, [self batchCount]);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batch1 ]];
+  group = [self.persistence startGroupWithAction:@"Count batches"];
   XCTAssertEqual(0, [self batchCount]);
   XCTAssertTrue([self.mutationQueue isEmpty]);
+  [self.persistence commitGroup:group];
 }
 
 - (void)testAcknowledgeBatchID {
@@ -114,10 +124,10 @@ NS_ASSUME_NONNULL_BEGIN
   FSTWriteGroup *group = [self.persistence startGroupWithAction:NSStringFromSelector(_cmd)];
   [self.mutationQueue acknowledgeBatch:batch1 streamToken:nil group:group];
   [self.mutationQueue removeMutationBatches:@[ batch1 ] group:group];
-  [self.persistence commitGroup:group];
 
   XCTAssertEqual([self batchCount], 0);
   XCTAssertEqual([self.mutationQueue highestAcknowledgedBatchID], batch1.batchID);
+  [self.persistence commitGroup:group];
 }
 
 - (void)testHighestAcknowledgedBatchIDNeverExceedsNextBatchID {
@@ -166,13 +176,16 @@ NS_ASSUME_NONNULL_BEGIN
   if ([self isTestBaseClass]) return;
 
   // Searching on an empty queue should not find a non-existent batch
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"LookupMutationBatch"];
   FSTMutationBatch *notFound = [self.mutationQueue lookupMutationBatch:42];
+  [self.persistence commitGroup:group];
   XCTAssertNil(notFound);
 
   NSMutableArray<FSTMutationBatch *> *batches = [self createBatches:10];
   NSArray<FSTMutationBatch *> *removed = [self makeHoles:@[ @2, @6, @7 ] inBatches:batches];
 
   // After removing, a batch should not be found
+  group = [self.persistence startGroupWithAction:@"LookupMutationBatch"];
   for (NSUInteger i = 0; i < removed.count; i++) {
     notFound = [self.mutationQueue lookupMutationBatch:removed[i].batchID];
     XCTAssertNil(notFound);
@@ -187,6 +200,7 @@ NS_ASSUME_NONNULL_BEGIN
   // Even on a nonempty queue searching should not find a non-existent batch
   notFound = [self.mutationQueue lookupMutationBatch:42];
   XCTAssertNil(notFound);
+  [self.persistence commitGroup:group];
 }
 
 - (void)testNextMutationBatchAfterBatchID {
@@ -198,6 +212,7 @@ NS_ASSUME_NONNULL_BEGIN
   NSArray<FSTMutationBatch *> *afters = @[ batches[3], batches[8], batches[8] ];
   NSArray<FSTMutationBatch *> *removed = [self makeHoles:@[ @2, @6, @7 ] inBatches:batches];
 
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"NextMutationBatchAfterBatchID"];
   for (NSUInteger i = 0; i < batches.count - 1; i++) {
     FSTMutationBatch *current = batches[i];
     FSTMutationBatch *next = batches[i + 1];
@@ -219,22 +234,29 @@ NS_ASSUME_NONNULL_BEGIN
   FSTMutationBatch *last = batches[batches.count - 1];
   FSTMutationBatch *notFound = [self.mutationQueue nextMutationBatchAfterBatchID:last.batchID];
   XCTAssertNil(notFound);
+  [self.persistence commitGroup:group];
 }
 
 - (void)testNextMutationBatchAfterBatchIDSkipsAcknowledgedBatches {
   if ([self isTestBaseClass]) return;
 
   NSMutableArray<FSTMutationBatch *> *batches = [self createBatches:3];
+  FSTWriteGroup *group = [self.persistence
+      startGroupWithAction:@"NextMutationBatchAfterBatchIDSkipsAcknowledgedBatches"];
   XCTAssertEqualObjects([self.mutationQueue nextMutationBatchAfterBatchID:kFSTBatchIDUnknown],
                         batches[0]);
+  [self.persistence commitGroup:group];
 
   [self acknowledgeBatch:batches[0]];
+  group = [self.persistence
+      startGroupWithAction:@"NextMutationBatchAfterBatchIDSkipsAcknowledgedBatches"];
   XCTAssertEqualObjects([self.mutationQueue nextMutationBatchAfterBatchID:kFSTBatchIDUnknown],
                         batches[1]);
   XCTAssertEqualObjects([self.mutationQueue nextMutationBatchAfterBatchID:batches[0].batchID],
                         batches[1]);
   XCTAssertEqualObjects([self.mutationQueue nextMutationBatchAfterBatchID:batches[1].batchID],
                         batches[2]);
+  [self.persistence commitGroup:group];
 }
 
 - (void)testAllMutationBatchesThroughBatchID {
@@ -245,6 +267,8 @@ NS_ASSUME_NONNULL_BEGIN
 
   NSArray<FSTMutationBatch *> *found, *expected;
 
+  FSTWriteGroup *group =
+      [self.persistence startGroupWithAction:@"AllMutationBatchesThroughBatchID"];
   found = [self.mutationQueue allMutationBatchesThroughBatchID:batches[0].batchID - 1];
   XCTAssertEqualObjects(found, (@[]));
 
@@ -253,6 +277,7 @@ NS_ASSUME_NONNULL_BEGIN
     expected = [batches subarrayWithRange:NSMakeRange(0, i + 1)];
     XCTAssertEqualObjects(found, expected, @"for index %lu", (unsigned long)i);
   }
+  [self.persistence commitGroup:group];
 }
 
 - (void)testAllMutationBatchesAffectingDocumentKey {
@@ -283,12 +308,12 @@ NS_ASSUME_NONNULL_BEGIN
                                                     group:group];
     [batches addObject:batch];
   }
-  [self.persistence commitGroup:group];
 
   NSArray<FSTMutationBatch *> *expected = @[ batches[1], batches[2] ];
   NSArray<FSTMutationBatch *> *matches =
       [self.mutationQueue allMutationBatchesAffectingDocumentKey:testutil::Key("foo/bar")];
 
+  [self.persistence commitGroup:group];
   XCTAssertEqualObjects(matches, expected);
 }
 
@@ -320,12 +345,12 @@ NS_ASSUME_NONNULL_BEGIN
                                                     group:group];
     [batches addObject:batch];
   }
-  [self.persistence commitGroup:group];
 
   NSArray<FSTMutationBatch *> *expected = @[ batches[1], batches[2], batches[4] ];
   FSTQuery *query = FSTTestQuery("foo");
   NSArray<FSTMutationBatch *> *matches =
       [self.mutationQueue allMutationBatchesAffectingQuery:query];
+  [self.persistence commitGroup:group];
 
   XCTAssertEqualObjects(matches, expected);
 }
@@ -338,48 +363,68 @@ NS_ASSUME_NONNULL_BEGIN
 
   [self removeMutationBatches:@[ batches[0] ]];
   [batches removeObjectAtIndex:0];
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   XCTAssertEqual([self batchCount], 9);
+  [self.persistence commitGroup:group];
 
   NSArray<FSTMutationBatch *> *found;
 
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
   XCTAssertEqualObjects(found, batches);
   XCTAssertEqual(found.count, 9);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[0], batches[1], batches[2] ]];
   [batches removeObjectsInRange:NSMakeRange(0, 3)];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   XCTAssertEqual([self batchCount], 6);
+  [self.persistence commitGroup:group];
 
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
   XCTAssertEqualObjects(found, batches);
   XCTAssertEqual(found.count, 6);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[batches.count - 1] ]];
   [batches removeObjectAtIndex:batches.count - 1];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   XCTAssertEqual([self batchCount], 5);
+  [self.persistence commitGroup:group];
 
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
   XCTAssertEqualObjects(found, batches);
   XCTAssertEqual(found.count, 5);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[3] ]];
   [batches removeObjectAtIndex:3];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   XCTAssertEqual([self batchCount], 4);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[1] ]];
   [batches removeObjectAtIndex:1];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   XCTAssertEqual([self batchCount], 3);
+  [self.persistence commitGroup:group];
 
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
   XCTAssertEqualObjects(found, batches);
   XCTAssertEqual(found.count, 3);
   XCTAssertFalse([self.mutationQueue isEmpty]);
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:batches];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatches"];
   found = [self.mutationQueue allMutationBatchesThroughBatchID:last.batchID];
   XCTAssertEqualObjects(found, @[]);
   XCTAssertEqual(found.count, 0);
   XCTAssertTrue([self.mutationQueue isEmpty]);
+  [self.persistence commitGroup:group];
 }
 
 - (void)testRemoveMutationBatchesEmitsGarbageEvents {
@@ -399,29 +444,42 @@ NS_ASSUME_NONNULL_BEGIN
   ]];
 
   [self removeMutationBatches:@[ batches[0] ]];
+  FSTWriteGroup *group =
+      [self.persistence startGroupWithAction:@"RemoveMutationBatchesEmitsGarbageEvents"];
   std::set<DocumentKey> garbage = [garbageCollector collectGarbage];
   XCTAssertEqual(garbage, std::set<DocumentKey>({}));
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[1] ]];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatchesEmitsGarbageEvents"];
   garbage = [garbageCollector collectGarbage];
   XCTAssertEqual(garbage, std::set<DocumentKey>({testutil::Key("foo/ba")}));
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[5] ]];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatchesEmitsGarbageEvents"];
   garbage = [garbageCollector collectGarbage];
   XCTAssertEqual(garbage, std::set<DocumentKey>({testutil::Key("bar/baz")}));
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[2], batches[3] ]];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatchesEmitsGarbageEvents"];
   garbage = [garbageCollector collectGarbage];
   XCTAssertEqual(garbage,
                  std::set<DocumentKey>({testutil::Key("foo/bar"), testutil::Key("foo/bar2")}));
+  [self.persistence commitGroup:group];
 
   [batches addObject:[self addMutationBatchWithKey:@"foo/bar/suffix/baz"]];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatchesEmitsGarbageEvents"];
   garbage = [garbageCollector collectGarbage];
   XCTAssertEqual(garbage, std::set<DocumentKey>({}));
+  [self.persistence commitGroup:group];
 
   [self removeMutationBatches:@[ batches[4], batches[6] ]];
+  group = [self.persistence startGroupWithAction:@"RemoveMutationBatchesEmitsGarbageEvents"];
   garbage = [garbageCollector collectGarbage];
   XCTAssertEqual(garbage, std::set<DocumentKey>({testutil::Key("foo/bar/suffix/baz")}));
+  [self.persistence commitGroup:group];
 }
 
 - (void)testStreamToken {
diff --git a/Firestore/Example/Tests/Local/FSTRemoteDocumentCacheTests.mm b/Firestore/Example/Tests/Local/FSTRemoteDocumentCacheTests.mm
index 16b7185..4f5595e 100644
--- a/Firestore/Example/Tests/Local/FSTRemoteDocumentCacheTests.mm
+++ b/Firestore/Example/Tests/Local/FSTRemoteDocumentCacheTests.mm
@@ -120,6 +120,7 @@ static const int kVersion = 42;
   [self setTestDocumentAtPath:"c/1"];
 
   FSTQuery *query = FSTTestQuery("b");
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"DocumentsMatchingQuery"];
   FSTDocumentDictionary *results = [self.remoteDocumentCache documentsMatchingQuery:query];
   NSArray *expected =
       @[ FSTTestDoc("b/1", kVersion, _kDocData, NO), FSTTestDoc("b/2", kVersion, _kDocData, NO) ];
@@ -127,6 +128,7 @@ static const int kVersion = 42;
   for (FSTDocument *doc in expected) {
     XCTAssertEqualObjects([results objectForKey:doc.key], doc);
   }
+  [self.persistence commitGroup:group];
 }
 
 #pragma mark - Helpers
@@ -144,7 +146,10 @@ static const int kVersion = 42;
 }
 
 - (FSTMaybeDocument *_Nullable)readEntryAtPath:(const absl::string_view)path {
-  return [self.remoteDocumentCache entryForKey:testutil::Key(path)];
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"ReadEntryAtPath"];
+  FSTMaybeDocument *result = [self.remoteDocumentCache entryForKey:testutil::Key(path)];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (void)removeEntryAtPath:(const absl::string_view)path {
diff --git a/Firestore/Example/Tests/Local/FSTRemoteDocumentChangeBufferTests.mm b/Firestore/Example/Tests/Local/FSTRemoteDocumentChangeBufferTests.mm
index 36d3661..d0466f9 100644
--- a/Firestore/Example/Tests/Local/FSTRemoteDocumentChangeBufferTests.mm
+++ b/Firestore/Example/Tests/Local/FSTRemoteDocumentChangeBufferTests.mm
@@ -68,8 +68,10 @@ NS_ASSUME_NONNULL_BEGIN
 }
 
 - (void)testReadUnchangedEntry {
+  FSTWriteGroup *group = [_db startGroupWithAction:@"ReadUnchangedEntry"];
   XCTAssertEqualObjects([_remoteDocumentBuffer entryForKey:FSTTestDocKey(@"coll/a")],
                         _kInitialADoc);
+  [_db commitGroup:group];
 }
 
 - (void)testAddEntryAndReadItBack {
@@ -78,24 +80,28 @@ NS_ASSUME_NONNULL_BEGIN
   XCTAssertEqualObjects([_remoteDocumentBuffer entryForKey:FSTTestDocKey(@"coll/a")], newADoc);
 
   // B should still be unchanged.
+  FSTWriteGroup *group = [_db startGroupWithAction:@"AddEntryAndReadItBack"];
   XCTAssertEqualObjects([_remoteDocumentBuffer entryForKey:FSTTestDocKey(@"coll/b")],
                         _kInitialBDoc);
+  [_db commitGroup:group];
 }
 
 - (void)testApplyChanges {
   FSTMaybeDocument *newADoc = FSTTestDoc("coll/a", 43, @{@"new" : @"data"}, NO);
   [_remoteDocumentBuffer addEntry:newADoc];
+  FSTWriteGroup *group = [_db startGroupWithAction:@"Apply Changes"];
   XCTAssertEqualObjects([_remoteDocumentBuffer entryForKey:FSTTestDocKey(@"coll/a")], newADoc);
 
   // Reading directly against the cache should still yield the old result.
   XCTAssertEqualObjects([_remoteDocumentCache entryForKey:FSTTestDocKey(@"coll/a")], _kInitialADoc);
+  [_db commitGroup:group];
 
-  FSTWriteGroup *group = [_db startGroupWithAction:@"Apply changes"];
+  group = [_db startGroupWithAction:@"Apply changes"];
   [_remoteDocumentBuffer applyToWriteGroup:group];
-  [_db commitGroup:group];
 
   // Reading against the cache should now yield the new result.
   XCTAssertEqualObjects([_remoteDocumentCache entryForKey:FSTTestDocKey(@"coll/a")], newADoc);
+  [_db commitGroup:group];
 }
 
 - (void)testMethodsThrowAfterApply {
diff --git a/Firestore/Source/Local/FSTLevelDB.mm b/Firestore/Source/Local/FSTLevelDB.mm
index 6efbff1..2edccb4 100644
--- a/Firestore/Source/Local/FSTLevelDB.mm
+++ b/Firestore/Source/Local/FSTLevelDB.mm
@@ -211,7 +211,7 @@ using leveldb::WriteOptions;
 #pragma mark - Persistence Factory methods
 
 - (id<FSTMutationQueue>)mutationQueueForUser:(const User &)user {
-  return [FSTLevelDBMutationQueue mutationQueueWithUser:user db:_ptr serializer:self.serializer];
+  return [FSTLevelDBMutationQueue mutationQueueWithUser:user db:self serializer:self.serializer];
 }
 
 - (id<FSTQueryCache>)queryCache {
@@ -219,7 +219,7 @@ using leveldb::WriteOptions;
 }
 
 - (id<FSTRemoteDocumentCache>)remoteDocumentCache {
-  return [[FSTLevelDBRemoteDocumentCache alloc] initWithDB:_ptr serializer:self.serializer];
+  return [[FSTLevelDBRemoteDocumentCache alloc] initWithDB:self serializer:self.serializer];
 }
 
 - (FSTWriteGroup *)startGroupWithAction:(NSString *)action {
diff --git a/Firestore/Source/Local/FSTLevelDBMutationQueue.h b/Firestore/Source/Local/FSTLevelDBMutationQueue.h
index 3f1bd51..034738f 100644
--- a/Firestore/Source/Local/FSTLevelDBMutationQueue.h
+++ b/Firestore/Source/Local/FSTLevelDBMutationQueue.h
@@ -44,7 +44,7 @@ NS_ASSUME_NONNULL_BEGIN
  * @param db The LevelDB in which to create the queue.
  */
 + (instancetype)mutationQueueWithUser:(const firebase::firestore::auth::User &)user
-                                   db:(std::shared_ptr<leveldb::DB>)db
+                                   db:(FSTLevelDB *)db
                            serializer:(FSTLocalSerializer *)serializer;
 
 /**
diff --git a/Firestore/Source/Local/FSTLevelDBMutationQueue.mm b/Firestore/Source/Local/FSTLevelDBMutationQueue.mm
index 221d56f..6d4bd29 100644
--- a/Firestore/Source/Local/FSTLevelDBMutationQueue.mm
+++ b/Firestore/Source/Local/FSTLevelDBMutationQueue.mm
@@ -16,11 +16,13 @@
 
 #import "Firestore/Source/Local/FSTLevelDBMutationQueue.h"
 
-#include <leveldb/db.h>
-#include <leveldb/write_batch.h>
 #include <set>
 #include <string>
 
+#include <absl/strings/match.h>
+#include <leveldb/db.h>
+#include <leveldb/write_batch.h>
+
 #import "Firestore/Protos/objc/firestore/local/Mutation.pbobjc.h"
 #import "Firestore/Source/Core/FSTQuery.h"
 #import "Firestore/Source/Local/FSTLevelDB.h"
@@ -32,6 +34,7 @@
 #import "Firestore/Source/Util/FSTAssert.h"
 
 #include "Firestore/core/src/firebase/firestore/auth/user.h"
+#include "Firestore/core/src/firebase/firestore/local/leveldb_transaction.h"
 #include "Firestore/core/src/firebase/firestore/model/document_key.h"
 #include "Firestore/core/src/firebase/firestore/model/resource_path.h"
 #include "Firestore/core/src/firebase/firestore/util/string_apple.h"
@@ -40,6 +43,7 @@
 NS_ASSUME_NONNULL_BEGIN
 
 namespace util = firebase::firestore::util;
+using firebase::firestore::local::LevelDbTransaction;
 using Firestore::StringView;
 using firebase::firestore::auth::User;
 using firebase::firestore::model::DocumentKey;
@@ -55,7 +59,7 @@ using leveldb::WriteOptions;
 @interface FSTLevelDBMutationQueue ()
 
 - (instancetype)initWithUserID:(NSString *)userID
-                            db:(std::shared_ptr<DB>)db
+                            db:(FSTLevelDB *)db
                     serializer:(FSTLocalSerializer *)serializer NS_DESIGNATED_INITIALIZER;
 
 /** The normalized userID (e.g. nil UID => @"" userID) used in our LevelDB keys. */
@@ -77,24 +81,12 @@ using leveldb::WriteOptions;
 
 @end
 
-/**
- * Returns a standard set of read options.
- *
- * For now this is paranoid, but perhaps disable that in production builds.
- */
-static ReadOptions StandardReadOptions() {
-  ReadOptions options;
-  options.verify_checksums = true;
-  return options;
-}
-
 @implementation FSTLevelDBMutationQueue {
-  // The DB pointer is shared with all cooperating LevelDB-related objects.
-  std::shared_ptr<DB> _db;
+  FSTLevelDB *_db;
 }
 
 + (instancetype)mutationQueueWithUser:(const User &)user
-                                   db:(std::shared_ptr<DB>)db
+                                   db:(FSTLevelDB *)db
                            serializer:(FSTLocalSerializer *)serializer {
   NSString *userID = user.is_authenticated() ? util::WrapNSString(user.uid()) : @"";
 
@@ -102,7 +94,7 @@ static ReadOptions StandardReadOptions() {
 }
 
 - (instancetype)initWithUserID:(NSString *)userID
-                            db:(std::shared_ptr<DB>)db
+                            db:(FSTLevelDB *)db
                     serializer:(FSTLocalSerializer *)serializer {
   if (self = [super init]) {
     _userID = [userID copy];
@@ -113,7 +105,7 @@ static ReadOptions StandardReadOptions() {
 }
 
 - (void)startWithGroup:(FSTWriteGroup *)group {
-  FSTBatchID nextBatchID = [FSTLevelDBMutationQueue loadNextBatchIDFromDB:_db];
+  FSTBatchID nextBatchID = [FSTLevelDBMutationQueue loadNextBatchIDFromDB:_db.ptr];
 
   // On restart, nextBatchId may end up lower than lastAcknowledgedBatchId since it's computed from
   // the queue contents, and there may be no mutations in the queue. In this case, we need to reset
@@ -142,11 +134,12 @@ static ReadOptions StandardReadOptions() {
 }
 
 - (void)shutdown {
-  _db.reset();
 }
 
 + (FSTBatchID)loadNextBatchIDFromDB:(std::shared_ptr<DB>)db {
-  std::unique_ptr<Iterator> it(db->NewIterator(StandardReadOptions()));
+  // TODO(gsoltis): implement Prev() and SeekToLast() on LevelDbTransaction::Iterator, then port
+  // this to a transaction.
+  std::unique_ptr<Iterator> it(db->NewIterator(LevelDbTransaction::DefaultReadOptions()));
 
   auto tableKey = [FSTLevelDBMutationKey keyPrefix];
 
@@ -209,19 +202,14 @@ static ReadOptions StandardReadOptions() {
 - (BOOL)isEmpty {
   std::string userKey = [FSTLevelDBMutationKey keyPrefixWithUserID:self.userID];
 
-  std::unique_ptr<Iterator> it(_db->NewIterator(StandardReadOptions()));
+  auto it = _db.currentTransaction->NewIterator();
   it->Seek(userKey);
 
   BOOL empty = YES;
-  if (it->Valid() && it->key().starts_with(userKey)) {
+  if (it->Valid() && absl::StartsWith(it->key(), userKey)) {
     empty = NO;
   }
 
-  Status status = it->status();
-  if (!status.ok()) {
-    FSTFail(@"isEmpty failed with status: %s", status.ToString().c_str());
-  }
-
   return empty;
 }
 
@@ -260,7 +248,7 @@ static ReadOptions StandardReadOptions() {
 
 - (nullable FSTPBMutationQueue *)metadataForKey:(const std::string &)key {
   std::string value;
-  Status status = _db->Get(StandardReadOptions(), key, &value);
+  Status status = _db.currentTransaction->Get(key, &value);
   if (status.ok()) {
     return [self parsedMetadata:value];
   } else if (status.IsNotFound()) {
@@ -304,7 +292,7 @@ static ReadOptions StandardReadOptions() {
   std::string key = [self mutationKeyForBatchID:batchID];
 
   std::string value;
-  Status status = _db->Get(StandardReadOptions(), key, &value);
+  Status status = _db.currentTransaction->Get(key, &value);
   if (!status.ok()) {
     if (status.IsNotFound()) {
       return nil;
@@ -323,15 +311,9 @@ static ReadOptions StandardReadOptions() {
   FSTBatchID nextBatchID = MAX(batchID, self.metadata.lastAcknowledgedBatchId) + 1;
 
   std::string key = [self mutationKeyForBatchID:nextBatchID];
-  std::unique_ptr<Iterator> it(_db->NewIterator(StandardReadOptions()));
+  auto it = _db.currentTransaction->NewIterator();
   it->Seek(key);
 
-  Status status = it->status();
-  if (!status.ok()) {
-    FSTFail(@"Seek to mutation batch (%@, %d) failed with status: %s", self.userID, batchID,
-            status.ToString().c_str());
-  }
-
   FSTLevelDBMutationKey *rowKey = [[FSTLevelDBMutationKey alloc] init];
   if (!it->Valid() || ![rowKey decodeKey:it->key()]) {
     // Past the last row in the DB or out of the mutations table
@@ -351,7 +333,7 @@ static ReadOptions StandardReadOptions() {
   std::string userKey = [FSTLevelDBMutationKey keyPrefixWithUserID:self.userID];
   const char *userID = [self.userID UTF8String];
 
-  std::unique_ptr<Iterator> it(_db->NewIterator(StandardReadOptions()));
+  auto it = _db.currentTransaction->NewIterator();
   it->Seek(userKey);
 
   NSMutableArray *result = [NSMutableArray array];
@@ -368,12 +350,6 @@ static ReadOptions StandardReadOptions() {
     [result addObject:[self decodedMutationBatch:it->value()]];
   }
 
-  Status status = it->status();
-  if (!status.ok()) {
-    FSTFail(@"Find all mutations through mutation batch (%@, %d) failed with status: %s",
-            self.userID, batchID, status.ToString().c_str());
-  }
-
   return result;
 }
 
@@ -384,26 +360,25 @@ static ReadOptions StandardReadOptions() {
   // Scan the document-mutation index starting with a prefix starting with the given documentKey.
   std::string indexPrefix = [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID
                                                                   resourcePath:documentKey.path()];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator(StandardReadOptions()));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   // Simultaneously scan the mutation queue. This works because each (key, batchID) pair is unique
   // and ordered, so when scanning a table prefixed by exactly key, all the batchIDs encountered
   // will be unique and in order.
   std::string mutationsPrefix = [FSTLevelDBMutationKey keyPrefixWithUserID:userID];
-  std::unique_ptr<Iterator> mutationIterator(_db->NewIterator(StandardReadOptions()));
+  auto mutationIterator = _db.currentTransaction->NewIterator();
 
   NSMutableArray *result = [NSMutableArray array];
   FSTLevelDBDocumentMutationKey *rowKey = [[FSTLevelDBDocumentMutationKey alloc] init];
   for (; indexIterator->Valid(); indexIterator->Next()) {
-    Slice indexKey = indexIterator->key();
-
     // Only consider rows matching exactly the specific key of interest. Note that because we order
     // by path first, and we order terminators before path separators, we'll encounter all the
     // index rows for documentKey contiguously. In particular, all the rows for documentKey will
     // occur before any rows for documents nested in a subcollection beneath documentKey so we can
     // stop as soon as we hit any such row.
-    if (!indexKey.starts_with(indexPrefix) || ![rowKey decodeKey:indexKey] ||
+    if (!absl::StartsWith(indexIterator->key(), indexPrefix) ||
+        ![rowKey decodeKey:indexIterator->key()] ||
         DocumentKey{rowKey.documentKey} != documentKey) {
       break;
     }
@@ -420,8 +395,8 @@ static ReadOptions StandardReadOptions() {
       FSTFail(
           @"Dangling document-mutation reference found: "
           @"%@ points to %@; seeking there found %@",
-          [FSTLevelDBKey descriptionForKey:indexKey], [FSTLevelDBKey descriptionForKey:mutationKey],
-          foundKeyDescription);
+          [FSTLevelDBKey descriptionForKey:indexIterator->key()],
+          [FSTLevelDBKey descriptionForKey:mutationKey], foundKeyDescription);
     }
 
     [result addObject:[self decodedMutationBatch:mutationIterator->value()]];
@@ -451,7 +426,7 @@ static ReadOptions StandardReadOptions() {
   // unique nor in order. This means an efficient simultaneous scan isn't possible.
   std::string indexPrefix =
       [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID resourcePath:queryPath];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator(StandardReadOptions()));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   NSMutableArray *result = [NSMutableArray array];
@@ -465,9 +440,8 @@ static ReadOptions StandardReadOptions() {
   // numbers of keys but > 30% faster for larger numbers of keys.
   std::set<FSTBatchID> uniqueBatchIds;
   for (; indexIterator->Valid(); indexIterator->Next()) {
-    Slice indexKey = indexIterator->key();
-
-    if (!indexKey.starts_with(indexPrefix) || ![rowKey decodeKey:indexKey]) {
+    if (!absl::StartsWith(indexIterator->key(), indexPrefix) ||
+        ![rowKey decodeKey:indexIterator->key()]) {
       break;
     }
 
@@ -483,7 +457,7 @@ static ReadOptions StandardReadOptions() {
 
   // Given an ordered set of unique batchIDs perform a skipping scan over the main table to find
   // the mutation batches.
-  std::unique_ptr<Iterator> mutationIterator(_db->NewIterator(StandardReadOptions()));
+  auto mutationIterator = _db.currentTransaction->NewIterator();
 
   for (FSTBatchID batchID : uniqueBatchIds) {
     std::string mutationKey = [FSTLevelDBMutationKey keyWithUserID:userID batchID:batchID];
@@ -507,19 +481,14 @@ static ReadOptions StandardReadOptions() {
 - (NSArray<FSTMutationBatch *> *)allMutationBatches {
   std::string userKey = [FSTLevelDBMutationKey keyPrefixWithUserID:self.userID];
 
-  std::unique_ptr<Iterator> it(_db->NewIterator(StandardReadOptions()));
+  auto it = _db.currentTransaction->NewIterator();
   it->Seek(userKey);
 
   NSMutableArray *result = [NSMutableArray array];
-  for (; it->Valid() && it->key().starts_with(userKey); it->Next()) {
+  for (; it->Valid() && absl::StartsWith(it->key(), userKey); it->Next()) {
     [result addObject:[self decodedMutationBatch:it->value()]];
   }
 
-  Status status = it->status();
-  if (!status.ok()) {
-    FSTFail(@"Find all mutation batches failed with status: %s", status.ToString().c_str());
-  }
-
   return result;
 }
 
@@ -527,7 +496,7 @@ static ReadOptions StandardReadOptions() {
   NSString *userID = self.userID;
   id<FSTGarbageCollector> garbageCollector = self.garbageCollector;
 
-  std::unique_ptr<Iterator> checkIterator(_db->NewIterator(StandardReadOptions()));
+  auto checkIterator = _db.currentTransaction->NewIterator();
 
   for (FSTMutationBatch *batch in batches) {
     FSTBatchID batchID = batch.batchID;
@@ -561,20 +530,18 @@ static ReadOptions StandardReadOptions() {
 
   // Verify that there are no entries in the document-mutation index if the queue is empty.
   std::string indexPrefix = [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator(StandardReadOptions()));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   NSMutableArray<NSString *> *danglingMutationReferences = [NSMutableArray array];
 
   for (; indexIterator->Valid(); indexIterator->Next()) {
-    Slice indexKey = indexIterator->key();
-
     // Only consider rows matching this index prefix for the current user.
-    if (!indexKey.starts_with(indexPrefix)) {
+    if (!absl::StartsWith(indexIterator->key(), indexPrefix)) {
       break;
     }
 
-    [danglingMutationReferences addObject:[FSTLevelDBKey descriptionForKey:indexKey]];
+    [danglingMutationReferences addObject:[FSTLevelDBKey descriptionForKey:indexIterator->key()]];
   }
 
   FSTAssert(danglingMutationReferences.count == 0,
@@ -605,9 +572,10 @@ static ReadOptions StandardReadOptions() {
   return proto;
 }
 
-- (FSTMutationBatch *)decodedMutationBatch:(Slice)slice {
-  NSData *data =
-      [[NSData alloc] initWithBytesNoCopy:(void *)slice.data() length:slice.size() freeWhenDone:NO];
+- (FSTMutationBatch *)decodedMutationBatch:(absl::string_view)encoded {
+  NSData *data = [[NSData alloc] initWithBytesNoCopy:(void *)encoded.data()
+                                              length:encoded.size()
+                                        freeWhenDone:NO];
 
   NSError *error;
   FSTPBWriteBatch *proto = [FSTPBWriteBatch parseFromData:data error:&error];
@@ -623,17 +591,16 @@ static ReadOptions StandardReadOptions() {
 - (BOOL)containsKey:(const DocumentKey &)documentKey {
   std::string indexPrefix = [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID
                                                                   resourcePath:documentKey.path()];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator(StandardReadOptions()));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   if (indexIterator->Valid()) {
     FSTLevelDBDocumentMutationKey *rowKey = [[FSTLevelDBDocumentMutationKey alloc] init];
-    Slice iteratorKey = indexIterator->key();
 
     // Check both that the key prefix matches and that the decoded document key is exactly the key
     // we're looking for.
-    if (iteratorKey.starts_with(indexPrefix) && [rowKey decodeKey:iteratorKey] &&
-        DocumentKey{rowKey.documentKey} == documentKey) {
+    if (absl::StartsWith(indexIterator->key(), indexPrefix) &&
+        [rowKey decodeKey:indexIterator->key()] && DocumentKey{rowKey.documentKey} == documentKey) {
       return YES;
     }
   }
diff --git a/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.h b/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.h
index 20942e2..381d308 100644
--- a/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.h
+++ b/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.h
@@ -19,8 +19,8 @@
 #include <memory>
 
 #import "Firestore/Source/Local/FSTRemoteDocumentCache.h"
-#include "leveldb/db.h"
 
+@class FSTLevelDB;
 @class FSTLocalSerializer;
 
 NS_ASSUME_NONNULL_BEGIN
@@ -35,7 +35,7 @@ NS_ASSUME_NONNULL_BEGIN
  *
  * @param db The leveldb in which to create the cache.
  */
-- (instancetype)initWithDB:(std::shared_ptr<leveldb::DB>)db
+- (instancetype)initWithDB:(FSTLevelDB *)db
                 serializer:(FSTLocalSerializer *)serializer NS_DESIGNATED_INITIALIZER;
 
 @end
diff --git a/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.mm b/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.mm
index cd14390..4ddd29f 100644
--- a/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.mm
+++ b/Firestore/Source/Local/FSTLevelDBRemoteDocumentCache.mm
@@ -22,6 +22,7 @@
 
 #import "Firestore/Protos/objc/firestore/local/MaybeDocument.pbobjc.h"
 #import "Firestore/Source/Core/FSTQuery.h"
+#import "Firestore/Source/Local/FSTLevelDB.h"
 #import "Firestore/Source/Local/FSTLevelDBKey.h"
 #import "Firestore/Source/Local/FSTLocalSerializer.h"
 #import "Firestore/Source/Local/FSTWriteGroup.h"
@@ -29,18 +30,16 @@
 #import "Firestore/Source/Model/FSTDocumentDictionary.h"
 #import "Firestore/Source/Model/FSTDocumentSet.h"
 #import "Firestore/Source/Util/FSTAssert.h"
+#include "Firestore/core/src/firebase/firestore/local/leveldb_transaction.h"
 
 #include "Firestore/core/src/firebase/firestore/model/document_key.h"
 
 NS_ASSUME_NONNULL_BEGIN
 
+using firebase::firestore::local::LevelDbTransaction;
 using firebase::firestore::model::DocumentKey;
 using leveldb::DB;
-using leveldb::Iterator;
-using leveldb::ReadOptions;
-using leveldb::Slice;
 using leveldb::Status;
-using leveldb::WriteOptions;
 
 @interface FSTLevelDBRemoteDocumentCache ()
 
@@ -48,23 +47,11 @@ using leveldb::WriteOptions;
 
 @end
 
-/**
- * Returns a standard set of read options.
- *
- * For now this is paranoid, but perhaps disable that in production builds.
- */
-static ReadOptions StandardReadOptions() {
-  ReadOptions options;
-  options.verify_checksums = true;
-  return options;
-}
-
 @implementation FSTLevelDBRemoteDocumentCache {
-  // The DB pointer is shared with all cooperating LevelDB-related objects.
-  std::shared_ptr<DB> _db;
+  FSTLevelDB *_db;
 }
 
-- (instancetype)initWithDB:(std::shared_ptr<DB>)db serializer:(FSTLocalSerializer *)serializer {
+- (instancetype)initWithDB:(FSTLevelDB *)db serializer:(FSTLocalSerializer *)serializer {
   if (self = [super init]) {
     _db = db;
     _serializer = serializer;
@@ -73,7 +60,6 @@ static ReadOptions StandardReadOptions() {
 }
 
 - (void)shutdown {
-  _db.reset();
 }
 
 - (void)addEntry:(FSTMaybeDocument *)document group:(FSTWriteGroup *)group {
@@ -89,11 +75,11 @@ static ReadOptions StandardReadOptions() {
 - (nullable FSTMaybeDocument *)entryForKey:(const DocumentKey &)documentKey {
   std::string key = [FSTLevelDBRemoteDocumentKey keyWithDocumentKey:documentKey];
   std::string value;
-  Status status = _db->Get(StandardReadOptions(), key, &value);
+  Status status = _db.currentTransaction->Get(key, &value);
   if (status.IsNotFound()) {
     return nil;
   } else if (status.ok()) {
-    return [self decodedMaybeDocument:value withKey:documentKey];
+    return [self decodeMaybeDocument:value withKey:documentKey];
   } else {
     FSTFail(@"Fetch document for key (%s) failed with status: %s", documentKey.ToString().c_str(),
             status.ToString().c_str());
@@ -106,13 +92,13 @@ static ReadOptions StandardReadOptions() {
   // Documents are ordered by key, so we can use a prefix scan to narrow down
   // the documents we need to match the query against.
   std::string startKey = [FSTLevelDBRemoteDocumentKey keyPrefixWithResourcePath:query.path];
-  std::unique_ptr<Iterator> it(_db->NewIterator(StandardReadOptions()));
+  auto it = _db.currentTransaction->NewIterator();
   it->Seek(startKey);
 
   FSTLevelDBRemoteDocumentKey *currentKey = [[FSTLevelDBRemoteDocumentKey alloc] init];
   for (; it->Valid() && [currentKey decodeKey:it->key()]; it->Next()) {
     FSTMaybeDocument *maybeDoc =
-        [self decodedMaybeDocument:it->value() withKey:currentKey.documentKey];
+        [self decodeMaybeDocument:it->value() withKey:currentKey.documentKey];
     if (!query.path.IsPrefixOf(maybeDoc.key.path())) {
       break;
     } else if ([maybeDoc isKindOfClass:[FSTDocument class]]) {
@@ -120,12 +106,6 @@ static ReadOptions StandardReadOptions() {
     }
   }
 
-  Status status = it->status();
-  if (!status.ok()) {
-    FSTFail(@"Find documents matching query (%@) failed with status: %s", query,
-            status.ToString().c_str());
-  }
-
   return results;
 }
 
@@ -133,9 +113,11 @@ static ReadOptions StandardReadOptions() {
   return [FSTLevelDBRemoteDocumentKey keyWithDocumentKey:key];
 }
 
-- (FSTMaybeDocument *)decodedMaybeDocument:(Slice)slice withKey:(const DocumentKey &)documentKey {
-  NSData *data =
-      [[NSData alloc] initWithBytesNoCopy:(void *)slice.data() length:slice.size() freeWhenDone:NO];
+- (FSTMaybeDocument *)decodeMaybeDocument:(absl::string_view)encoded
+                                  withKey:(const DocumentKey &)documentKey {
+  NSData *data = [[NSData alloc] initWithBytesNoCopy:(void *)encoded.data()
+                                              length:encoded.size()
+                                        freeWhenDone:NO];
 
   NSError *error;
   FSTPBMaybeDocument *proto = [FSTPBMaybeDocument parseFromData:data error:&error];
diff --git a/Firestore/Source/Local/FSTLocalStore.mm b/Firestore/Source/Local/FSTLocalStore.mm
index 1bcc6d0..2ea3328 100644
--- a/Firestore/Source/Local/FSTLocalStore.mm
+++ b/Firestore/Source/Local/FSTLocalStore.mm
@@ -172,7 +172,9 @@ NS_ASSUME_NONNULL_BEGIN
 
 - (FSTMaybeDocumentDictionary *)userDidChange:(const User &)user {
   // Swap out the mutation queue, grabbing the pending mutation batches before and after.
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"OldBatches"];
   NSArray<FSTMutationBatch *> *oldBatches = [self.mutationQueue allMutationBatches];
+  [self.persistence commitGroup:group];
 
   [self.mutationQueue shutdown];
   [self.garbageCollector removeGarbageSource:self.mutationQueue];
@@ -182,6 +184,7 @@ NS_ASSUME_NONNULL_BEGIN
 
   [self startMutationQueue];
 
+  group = [self.persistence startGroupWithAction:@"NewBatches"];
   NSArray<FSTMutationBatch *> *newBatches = [self.mutationQueue allMutationBatches];
 
   // Recreate our LocalDocumentsView using the new MutationQueue.
@@ -199,7 +202,9 @@ NS_ASSUME_NONNULL_BEGIN
   }
 
   // Return the set of all (potentially) changed documents as the result of the user change.
-  return [self.localDocuments documentsForKeys:changedKeys];
+  FSTMaybeDocumentDictionary *result = [self.localDocuments documentsForKeys:changedKeys];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (FSTLocalWriteResult *)locallyWriteMutations:(NSArray<FSTMutation *> *)mutations {
@@ -208,10 +213,9 @@ NS_ASSUME_NONNULL_BEGIN
   FSTMutationBatch *batch = [self.mutationQueue addMutationBatchWithWriteTime:localWriteTime
                                                                     mutations:mutations
                                                                         group:group];
-  [self.persistence commitGroup:group];
-
   FSTDocumentKeySet *keys = [batch keys];
   FSTMaybeDocumentDictionary *changedDocuments = [self.localDocuments documentsForKeys:keys];
+  [self.persistence commitGroup:group];
   return [FSTLocalWriteResult resultForBatchID:batch.batchID changes:changedDocuments];
 }
 
@@ -237,10 +241,11 @@ NS_ASSUME_NONNULL_BEGIN
     [remoteDocuments applyToWriteGroup:group];
   }
 
-  [self.persistence commitGroup:group];
   [self.mutationQueue performConsistencyCheck];
 
-  return [self.localDocuments documentsForKeys:affected];
+  FSTMaybeDocumentDictionary *result = [self.localDocuments documentsForKeys:affected];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (FSTMaybeDocumentDictionary *)rejectBatchID:(FSTBatchID)batchID {
@@ -254,10 +259,11 @@ NS_ASSUME_NONNULL_BEGIN
 
   FSTDocumentKeySet *affected = [self removeMutationBatch:toReject group:group];
 
-  [self.persistence commitGroup:group];
   [self.mutationQueue performConsistencyCheck];
 
-  return [self.localDocuments documentsForKeys:affected];
+  FSTMaybeDocumentDictionary *result = [self.localDocuments documentsForKeys:affected];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (nullable NSData *)lastStreamToken {
@@ -362,15 +368,15 @@ NS_ASSUME_NONNULL_BEGIN
 
   [remoteDocuments applyToWriteGroup:group];
 
-  [self.persistence commitGroup:group];
-
   // Union the two key sets.
   __block FSTDocumentKeySet *keysToRecalc = changedDocKeys;
   [releasedWriteKeys enumerateObjectsUsingBlock:^(FSTDocumentKey *key, BOOL *stop) {
     keysToRecalc = [keysToRecalc setByAddingObject:key];
   }];
 
-  return [self.localDocuments documentsForKeys:keysToRecalc];
+  FSTMaybeDocumentDictionary *result = [self.localDocuments documentsForKeys:keysToRecalc];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (void)notifyLocalViewChanges:(NSArray<FSTLocalViewChanges *> *)viewChanges {
@@ -387,11 +393,17 @@ NS_ASSUME_NONNULL_BEGIN
 }
 
 - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(FSTBatchID)batchID {
-  return [self.mutationQueue nextMutationBatchAfterBatchID:batchID];
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"nextMutationBatchAfterBatchID"];
+  FSTMutationBatch *result = [self.mutationQueue nextMutationBatchAfterBatchID:batchID];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (nullable FSTMaybeDocument *)readDocument:(const DocumentKey &)key {
-  return [self.localDocuments documentForKey:key];
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"ReadDocument"];
+  FSTMaybeDocument *result = [self.localDocuments documentForKey:key];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (FSTQueryData *)allocateQuery:(FSTQuery *)query {
@@ -447,7 +459,10 @@ NS_ASSUME_NONNULL_BEGIN
 }
 
 - (FSTDocumentDictionary *)executeQuery:(FSTQuery *)query {
-  return [self.localDocuments documentsMatchingQuery:query];
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"ExecuteQuery"];
+  FSTDocumentDictionary *result = [self.localDocuments documentsMatchingQuery:query];
+  [self.persistence commitGroup:group];
+  return result;
 }
 
 - (FSTDocumentKeySet *)remoteDocumentKeysForTarget:(FSTTargetID)targetID {
diff --git a/Firestore/core/src/firebase/firestore/local/leveldb_transaction.cc b/Firestore/core/src/firebase/firestore/local/leveldb_transaction.cc
index f7d39b2..966ccf2 100644
--- a/Firestore/core/src/firebase/firestore/local/leveldb_transaction.cc
+++ b/Firestore/core/src/firebase/firestore/local/leveldb_transaction.cc
@@ -71,8 +71,14 @@ void LevelDbTransaction::Iterator::UpdateCurrent() {
 
 void LevelDbTransaction::Iterator::Seek(const std::string& key) {
   db_iter_->Seek(key);
+  FIREBASE_ASSERT_MESSAGE(db_iter_->status().ok(),
+                          "leveldb iterator reported an error: %s",
+                          db_iter_->status().ToString().c_str());
   for (; db_iter_->Valid() && IsDeleted(db_iter_->key()); db_iter_->Next()) {
   }
+  FIREBASE_ASSERT_MESSAGE(db_iter_->status().ok(),
+                          "leveldb iterator reported an error: %s",
+                          db_iter_->status().ToString().c_str());
   mutations_iter_ = txn_->mutations_.lower_bound(key);
   UpdateCurrent();
   last_version_ = txn_->version_;
@@ -109,6 +115,9 @@ void LevelDbTransaction::Iterator::AdvanceLDB() {
   do {
     db_iter_->Next();
   } while (db_iter_->Valid() && IsDeleted(db_iter_->key()));
+  FIREBASE_ASSERT_MESSAGE(db_iter_->status().ok(),
+                          "leveldb iterator reported an error: %s",
+                          db_iter_->status().ToString().c_str());
 }
 
 void LevelDbTransaction::Iterator::Next() {