Switch FSTLevelDBQueryCache to use transactions (#942)

* 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

* Fix an include, but mostly try to trigger a travis build

6 files changed, 44 insertions, 39 deletions

diff --git a/Firestore/Source/Local/FSTLevelDB.h b/Firestore/Source/Local/FSTLevelDB.h
index 77abb3d..b03b65f 100644
--- a/Firestore/Source/Local/FSTLevelDB.h
+++ b/Firestore/Source/Local/FSTLevelDB.h
@@ -20,6 +20,7 @@
 
 #import "Firestore/Source/Local/FSTPersistence.h"
 #include "Firestore/core/src/firebase/firestore/core/database_info.h"
+#include "Firestore/core/src/firebase/firestore/local/leveldb_transaction.h"
 #include "leveldb/db.h"
 
 @class FSTLocalSerializer;
@@ -96,6 +97,8 @@ NS_ASSUME_NONNULL_BEGIN
 /** The native db pointer, allocated during start. */
 @property(nonatomic, assign, readonly) std::shared_ptr<leveldb::DB> ptr;
 
+@property(nonatomic, readonly) firebase::firestore::local::LevelDbTransaction *currentTransaction;
+
 @end
 
 NS_ASSUME_NONNULL_END
diff --git a/Firestore/Source/Local/FSTLevelDB.mm b/Firestore/Source/Local/FSTLevelDB.mm
index a7ee99d..6efbff1 100644
--- a/Firestore/Source/Local/FSTLevelDB.mm
+++ b/Firestore/Source/Local/FSTLevelDB.mm
@@ -203,6 +203,11 @@ using leveldb::WriteOptions;
   return database;
 }
 
+- (LevelDbTransaction *)currentTransaction {
+  FSTAssert(_transaction != nullptr, @"Attempting to access transaction before one has started");
+  return _transaction.get();
+}
+
 #pragma mark - Persistence Factory methods
 
 - (id<FSTMutationQueue>)mutationQueueForUser:(const User &)user {
@@ -210,7 +215,7 @@ using leveldb::WriteOptions;
 }
 
 - (id<FSTQueryCache>)queryCache {
-  return [[FSTLevelDBQueryCache alloc] initWithDB:_ptr serializer:self.serializer];
+  return [[FSTLevelDBQueryCache alloc] initWithDB:self serializer:self.serializer];
 }
 
 - (id<FSTRemoteDocumentCache>)remoteDocumentCache {
diff --git a/Firestore/Source/Local/FSTLevelDBMigrations.mm b/Firestore/Source/Local/FSTLevelDBMigrations.mm
index 7f521d1..cf06c9f 100644
--- a/Firestore/Source/Local/FSTLevelDBMigrations.mm
+++ b/Firestore/Source/Local/FSTLevelDBMigrations.mm
@@ -66,7 +66,7 @@ static void SaveVersion(FSTLevelDBSchemaVersion version, LevelDbTransaction *tra
  * It assumes the metadata has already been written and is able to be read in this transaction.
  */
 static void AddTargetCount(LevelDbTransaction *transaction) {
-  std::unique_ptr<LevelDbTransaction::Iterator> it(transaction->NewIterator());
+  auto it = transaction->NewIterator();
   std::string start_key = [FSTLevelDBTargetKey keyPrefix];
   it->Seek(start_key);
 
diff --git a/Firestore/Source/Local/FSTLevelDBQueryCache.h b/Firestore/Source/Local/FSTLevelDBQueryCache.h
index d955b02..2cd6758 100644
--- a/Firestore/Source/Local/FSTLevelDBQueryCache.h
+++ b/Firestore/Source/Local/FSTLevelDBQueryCache.h
@@ -22,6 +22,7 @@
 #include "Firestore/core/src/firebase/firestore/local/leveldb_transaction.h"
 #include "leveldb/db.h"
 
+@class FSTLevelDB;
 @class FSTLocalSerializer;
 @class FSTPBTargetGlobal;
 @protocol FSTGarbageCollector;
@@ -53,7 +54,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/FSTLevelDBQueryCache.mm b/Firestore/Source/Local/FSTLevelDBQueryCache.mm
index 95b032e..c7eecc4 100644
--- a/Firestore/Source/Local/FSTLevelDBQueryCache.mm
+++ b/Firestore/Source/Local/FSTLevelDBQueryCache.mm
@@ -16,9 +16,6 @@
 
 #import "Firestore/Source/Local/FSTLevelDBQueryCache.h"
 
-#include <leveldb/db.h>
-#include <leveldb/write_batch.h>
-
 #import "Firestore/Protos/objc/firestore/local/Target.pbobjc.h"
 #import "Firestore/Source/Core/FSTQuery.h"
 #import "Firestore/Source/Local/FSTLevelDB.h"
@@ -28,17 +25,15 @@
 #import "Firestore/Source/Local/FSTWriteGroup.h"
 #import "Firestore/Source/Model/FSTDocumentKey.h"
 #import "Firestore/Source/Util/FSTAssert.h"
+#include "absl/strings/match.h"
 
 NS_ASSUME_NONNULL_BEGIN
 
 using firebase::firestore::local::LevelDbTransaction;
 using Firestore::StringView;
 using leveldb::DB;
-using leveldb::Iterator;
-using leveldb::ReadOptions;
 using leveldb::Slice;
 using leveldb::Status;
-using leveldb::WriteOptions;
 
 @interface FSTLevelDBQueryCache ()
 
@@ -50,8 +45,7 @@ using leveldb::WriteOptions;
 @end
 
 @implementation FSTLevelDBQueryCache {
-  // The DB pointer is shared with all cooperating LevelDB-related objects.
-  std::shared_ptr<DB> _db;
+  FSTLevelDB *_db;
 
   /**
    * The last received snapshot version. This is part of `metadata` but we store it separately to
@@ -107,7 +101,7 @@ using leveldb::WriteOptions;
   return proto;
 }
 
-- (instancetype)initWithDB:(std::shared_ptr<DB>)db serializer:(FSTLocalSerializer *)serializer {
+- (instancetype)initWithDB:(FSTLevelDB *)db serializer:(FSTLocalSerializer *)serializer {
   if (self = [super init]) {
     FSTAssert(db, @"db must not be NULL");
     _db = db;
@@ -117,7 +111,8 @@ using leveldb::WriteOptions;
 }
 
 - (void)start {
-  FSTPBTargetGlobal *metadata = [FSTLevelDBQueryCache readTargetMetadataFromDB:_db];
+  // TODO(gsoltis): switch this usage of ptr to currentTransaction
+  FSTPBTargetGlobal *metadata = [FSTLevelDBQueryCache readTargetMetadataFromDB:_db.ptr];
   FSTAssert(
       metadata != nil,
       @"Found nil metadata, expected schema to be at version 0 which ensures metadata existence");
@@ -148,7 +143,6 @@ using leveldb::WriteOptions;
 }
 
 - (void)shutdown {
-  _db.reset();
 }
 
 - (void)saveQueryData:(FSTQueryData *)queryData group:(FSTWriteGroup *)group {
@@ -221,9 +215,10 @@ using leveldb::WriteOptions;
  * Parses the given bytes as an FSTPBTarget protocol buffer and then converts to the equivalent
  * query data.
  */
-- (FSTQueryData *)decodedTargetWithSlice:(Slice)slice {
-  NSData *data =
-      [[NSData alloc] initWithBytesNoCopy:(void *)slice.data() length:slice.size() freeWhenDone:NO];
+- (FSTQueryData *)decodeTarget:(absl::string_view)encoded {
+  NSData *data = [[NSData alloc] initWithBytesNoCopy:(void *)encoded.data()
+                                              length:encoded.size()
+                                        freeWhenDone:NO];
 
   NSError *error;
   FSTPBTarget *proto = [FSTPBTarget parseFromData:data error:&error];
@@ -239,7 +234,7 @@ using leveldb::WriteOptions;
   // Note that this is a scan rather than a get because canonicalIDs are not required to be unique
   // per target.
   Slice canonicalID = StringView(query.canonicalID);
-  std::unique_ptr<Iterator> indexItererator(_db->NewIterator([FSTLevelDB standardReadOptions]));
+  auto indexItererator = _db.currentTransaction->NewIterator();
   std::string indexPrefix = [FSTLevelDBQueryTargetKey keyPrefixWithCanonicalID:canonicalID];
   indexItererator->Seek(indexPrefix);
 
@@ -247,15 +242,13 @@ using leveldb::WriteOptions;
   // unique and ordered, so when scanning a table prefixed by exactly one canonicalID, all the
   // targetIDs will be unique and in order.
   std::string targetPrefix = [FSTLevelDBTargetKey keyPrefix];
-  std::unique_ptr<Iterator> targetIterator(_db->NewIterator([FSTLevelDB standardReadOptions]));
+  auto targetIterator = _db.currentTransaction->NewIterator();
 
   FSTLevelDBQueryTargetKey *rowKey = [[FSTLevelDBQueryTargetKey alloc] init];
   for (; indexItererator->Valid(); indexItererator->Next()) {
-    Slice indexKey = indexItererator->key();
-
     // Only consider rows matching exactly the specific canonicalID of interest.
-    if (!indexKey.starts_with(indexPrefix) || ![rowKey decodeKey:indexKey] ||
-        canonicalID != rowKey.canonicalID) {
+    if (!absl::StartsWith(indexItererator->key(), indexPrefix) ||
+        ![rowKey decodeKey:indexItererator->key()] || canonicalID != rowKey.canonicalID) {
       // End of this canonicalID's possible targets.
       break;
     }
@@ -272,13 +265,13 @@ using leveldb::WriteOptions;
       FSTFail(
           @"Dangling query-target reference found: "
           @"%@ points to %@; seeking there found %@",
-          [FSTLevelDBKey descriptionForKey:indexKey], [FSTLevelDBKey descriptionForKey:targetKey],
-          foundKeyDescription);
+          [FSTLevelDBKey descriptionForKey:indexItererator->key()],
+          [FSTLevelDBKey descriptionForKey:targetKey], foundKeyDescription);
     }
 
     // Finally after finding a potential match, check that the query is actually equal to the
     // requested query.
-    FSTQueryData *target = [self decodedTargetWithSlice:targetIterator->value()];
+    FSTQueryData *target = [self decodeTarget:targetIterator->value()];
     if ([target.query isEqual:query]) {
       return target;
     }
@@ -319,12 +312,12 @@ using leveldb::WriteOptions;
 
 - (void)removeMatchingKeysForTargetID:(FSTTargetID)targetID group:(FSTWriteGroup *)group {
   std::string indexPrefix = [FSTLevelDBTargetDocumentKey keyPrefixWithTargetID:targetID];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator([FSTLevelDB standardReadOptions]));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   FSTLevelDBTargetDocumentKey *rowKey = [[FSTLevelDBTargetDocumentKey alloc] init];
   for (; indexIterator->Valid(); indexIterator->Next()) {
-    Slice indexKey = indexIterator->key();
+    absl::string_view indexKey = indexIterator->key();
 
     // Only consider rows matching this specific targetID.
     if (![rowKey decodeKey:indexKey] || rowKey.targetID != targetID) {
@@ -342,13 +335,13 @@ using leveldb::WriteOptions;
 
 - (FSTDocumentKeySet *)matchingKeysForTargetID:(FSTTargetID)targetID {
   std::string indexPrefix = [FSTLevelDBTargetDocumentKey keyPrefixWithTargetID:targetID];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator([FSTLevelDB standardReadOptions]));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   FSTDocumentKeySet *result = [FSTDocumentKeySet keySet];
   FSTLevelDBTargetDocumentKey *rowKey = [[FSTLevelDBTargetDocumentKey alloc] init];
   for (; indexIterator->Valid(); indexIterator->Next()) {
-    Slice indexKey = indexIterator->key();
+    absl::string_view indexKey = indexIterator->key();
 
     // Only consider rows matching this specific targetID.
     if (![rowKey decodeKey:indexKey] || rowKey.targetID != targetID) {
@@ -365,7 +358,7 @@ using leveldb::WriteOptions;
 
 - (BOOL)containsKey:(FSTDocumentKey *)key {
   std::string indexPrefix = [FSTLevelDBDocumentTargetKey keyPrefixWithResourcePath:key.path];
-  std::unique_ptr<Iterator> indexIterator(_db->NewIterator([FSTLevelDB standardReadOptions]));
+  auto indexIterator = _db.currentTransaction->NewIterator();
   indexIterator->Seek(indexPrefix);
 
   if (indexIterator->Valid()) {
diff --git a/Firestore/Source/Local/FSTLocalStore.mm b/Firestore/Source/Local/FSTLocalStore.mm
index a5b85cd..f61a20f 100644
--- a/Firestore/Source/Local/FSTLocalStore.mm
+++ b/Firestore/Source/Local/FSTLocalStore.mm
@@ -31,6 +31,7 @@
 #import "Firestore/Source/Local/FSTReferenceSet.h"
 #import "Firestore/Source/Local/FSTRemoteDocumentCache.h"
 #import "Firestore/Source/Local/FSTRemoteDocumentChangeBuffer.h"
+#import "Firestore/Source/Local/FSTWriteGroup.h"
 #import "Firestore/Source/Model/FSTDocument.h"
 #import "Firestore/Source/Model/FSTDocumentDictionary.h"
 #import "Firestore/Source/Model/FSTDocumentKey.h"
@@ -371,6 +372,7 @@ NS_ASSUME_NONNULL_BEGIN
 
 - (void)notifyLocalViewChanges:(NSArray<FSTLocalViewChanges *> *)viewChanges {
   FSTReferenceSet *localViewReferences = self.localViewReferences;
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"NotifyLocalViewChanges"];
   for (FSTLocalViewChanges *view in viewChanges) {
     FSTQueryData *queryData = [self.queryCache queryDataForQuery:view.query];
     FSTAssert(queryData, @"Local view changes contain unallocated query.");
@@ -378,6 +380,7 @@ NS_ASSUME_NONNULL_BEGIN
     [localViewReferences addReferencesToKeys:view.addedKeys forID:targetID];
     [localViewReferences removeReferencesToKeys:view.removedKeys forID:targetID];
   }
+  [self.persistence commitGroup:group];
 }
 
 - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(FSTBatchID)batchID {
@@ -389,6 +392,7 @@ NS_ASSUME_NONNULL_BEGIN
 }
 
 - (FSTQueryData *)allocateQuery:(FSTQuery *)query {
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Allocate query"];
   FSTQueryData *cached = [self.queryCache queryDataForQuery:query];
   FSTTargetID targetID;
   FSTListenSequenceNumber sequenceNumber = [self.listenSequence next];
@@ -397,18 +401,14 @@ NS_ASSUME_NONNULL_BEGIN
     // TODO(mcg): freshen last accessed date?
     targetID = cached.targetID;
   } else {
-    FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Allocate query"];
-
     targetID = _targetIDGenerator.NextId();
     cached = [[FSTQueryData alloc] initWithQuery:query
                                         targetID:targetID
                             listenSequenceNumber:sequenceNumber
                                          purpose:FSTQueryPurposeListen];
     [self.queryCache addQueryData:cached group:group];
-
-    [self.persistence commitGroup:group];
   }
-
+  [self.persistence commitGroup:group];
   // Sanity check to ensure that even when resuming a query it's not currently active.
   FSTBoxedTargetID *boxedTargetID = @(targetID);
   FSTAssert(!self.targetIDs[boxedTargetID], @"Tried to allocate an already allocated query: %@",
@@ -448,20 +448,23 @@ NS_ASSUME_NONNULL_BEGIN
 }
 
 - (FSTDocumentKeySet *)remoteDocumentKeysForTarget:(FSTTargetID)targetID {
-  return [self.queryCache matchingKeysForTargetID:targetID];
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"RemoteDocumentKeysForTarget"];
+  FSTDocumentKeySet *keySet = [self.queryCache matchingKeysForTargetID:targetID];
+  [self.persistence commitGroup:group];
+  return keySet;
 }
 
 - (void)collectGarbage {
+  FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Garbage Collection"];
   // Call collectGarbage regardless of whether isGCEnabled so the referenceSet doesn't continue to
   // accumulate the garbage keys.
   NSSet<FSTDocumentKey *> *garbage = [self.garbageCollector collectGarbage];
   if (garbage.count > 0) {
-    FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Garbage Collection"];
     for (FSTDocumentKey *key in garbage) {
       [self.remoteDocumentCache removeEntryForKey:key group:group];
     }
-    [self.persistence commitGroup:group];
   }
+  [self.persistence commitGroup:group];
 }
 
 /**