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

1 files changed, 14 insertions, 32 deletions

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];