FSTLevelDBMutationQueue.mm 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658
  1. /*
  2. * Copyright 2017 Google
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #import "Firestore/Source/Local/FSTLevelDBMutationQueue.h"
  17. #include <memory>
  18. #include <set>
  19. #include <string>
  20. #import "Firestore/Protos/objc/firestore/local/Mutation.pbobjc.h"
  21. #import "Firestore/Source/Core/FSTQuery.h"
  22. #import "Firestore/Source/Local/FSTLevelDB.h"
  23. #import "Firestore/Source/Local/FSTLevelDBKey.h"
  24. #import "Firestore/Source/Local/FSTLocalSerializer.h"
  25. #import "Firestore/Source/Model/FSTMutation.h"
  26. #import "Firestore/Source/Model/FSTMutationBatch.h"
  27. #include "Firestore/core/src/firebase/firestore/auth/user.h"
  28. #include "Firestore/core/src/firebase/firestore/local/leveldb_transaction.h"
  29. #include "Firestore/core/src/firebase/firestore/model/resource_path.h"
  30. #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
  31. #include "Firestore/core/src/firebase/firestore/util/string_apple.h"
  32. #include "Firestore/core/src/firebase/firestore/util/string_util.h"
  33. #include "absl/strings/match.h"
  34. #include "leveldb/db.h"
  35. #include "leveldb/write_batch.h"
  36. NS_ASSUME_NONNULL_BEGIN
  37. namespace util = firebase::firestore::util;
  38. using firebase::firestore::local::LevelDbTransaction;
  39. using Firestore::StringView;
  40. using firebase::firestore::auth::User;
  41. using firebase::firestore::model::DocumentKey;
  42. using firebase::firestore::model::DocumentKeySet;
  43. using firebase::firestore::model::ResourcePath;
  44. using leveldb::DB;
  45. using leveldb::Iterator;
  46. using leveldb::ReadOptions;
  47. using leveldb::Slice;
  48. using leveldb::Status;
  49. using leveldb::WriteBatch;
  50. using leveldb::WriteOptions;
  51. @interface FSTLevelDBMutationQueue ()
  52. - (instancetype)initWithUserID:(NSString *)userID
  53. db:(FSTLevelDB *)db
  54. serializer:(FSTLocalSerializer *)serializer NS_DESIGNATED_INITIALIZER;
  55. /** The normalized userID (e.g. nil UID => @"" userID) used in our LevelDB keys. */
  56. @property(nonatomic, strong, readonly) NSString *userID;
  57. /**
  58. * Next value to use when assigning sequential IDs to each mutation batch.
  59. *
  60. * NOTE: There can only be one FSTLevelDBMutationQueue for a given db at a time, hence it is safe
  61. * to track nextBatchID as an instance-level property. Should we ever relax this constraint we'll
  62. * need to revisit this.
  63. */
  64. @property(nonatomic, assign) FSTBatchID nextBatchID;
  65. /** A write-through cache copy of the metadata describing the current queue. */
  66. @property(nonatomic, strong, nullable) FSTPBMutationQueue *metadata;
  67. @property(nonatomic, strong, readonly) FSTLocalSerializer *serializer;
  68. @end
  69. @implementation FSTLevelDBMutationQueue {
  70. FSTLevelDB *_db;
  71. }
  72. + (instancetype)mutationQueueWithUser:(const User &)user
  73. db:(FSTLevelDB *)db
  74. serializer:(FSTLocalSerializer *)serializer {
  75. NSString *userID = user.is_authenticated() ? util::WrapNSString(user.uid()) : @"";
  76. return [[FSTLevelDBMutationQueue alloc] initWithUserID:userID db:db serializer:serializer];
  77. }
  78. - (instancetype)initWithUserID:(NSString *)userID
  79. db:(FSTLevelDB *)db
  80. serializer:(FSTLocalSerializer *)serializer {
  81. if (self = [super init]) {
  82. _userID = [userID copy];
  83. _db = db;
  84. _serializer = serializer;
  85. }
  86. return self;
  87. }
  88. - (void)start {
  89. FSTBatchID nextBatchID = [FSTLevelDBMutationQueue loadNextBatchIDFromDB:_db.ptr];
  90. // On restart, nextBatchId may end up lower than lastAcknowledgedBatchId since it's computed from
  91. // the queue contents, and there may be no mutations in the queue. In this case, we need to reset
  92. // lastAcknowledgedBatchId (which is safe since the queue must be empty).
  93. std::string key = [self keyForCurrentMutationQueue];
  94. FSTPBMutationQueue *metadata = [self metadataForKey:key];
  95. if (!metadata) {
  96. metadata = [FSTPBMutationQueue message];
  97. // proto3's default value for lastAcknowledgedBatchId is zero, but that would consider the first
  98. // entry in the queue to be acknowledged without that acknowledgement actually happening.
  99. metadata.lastAcknowledgedBatchId = kFSTBatchIDUnknown;
  100. } else {
  101. FSTBatchID lastAcked = metadata.lastAcknowledgedBatchId;
  102. if (lastAcked >= nextBatchID) {
  103. HARD_ASSERT([self isEmpty], "Reset nextBatchID is only possible when the queue is empty");
  104. lastAcked = kFSTBatchIDUnknown;
  105. metadata.lastAcknowledgedBatchId = lastAcked;
  106. _db.currentTransaction->Put([self keyForCurrentMutationQueue], metadata);
  107. }
  108. }
  109. self.nextBatchID = nextBatchID;
  110. self.metadata = metadata;
  111. }
  112. + (FSTBatchID)loadNextBatchIDFromDB:(DB *)db {
  113. // TODO(gsoltis): implement Prev() and SeekToLast() on LevelDbTransaction::Iterator, then port
  114. // this to a transaction.
  115. std::unique_ptr<Iterator> it(db->NewIterator(LevelDbTransaction::DefaultReadOptions()));
  116. auto tableKey = [FSTLevelDBMutationKey keyPrefix];
  117. FSTLevelDBMutationKey *rowKey = [[FSTLevelDBMutationKey alloc] init];
  118. FSTBatchID maxBatchID = kFSTBatchIDUnknown;
  119. BOOL moreUserIDs = NO;
  120. std::string nextUserID;
  121. it->Seek(tableKey);
  122. if (it->Valid() && [rowKey decodeKey:it->key()]) {
  123. moreUserIDs = YES;
  124. nextUserID = rowKey.userID;
  125. }
  126. // This loop assumes that nextUserId contains the next username at the start of the iteration.
  127. while (moreUserIDs) {
  128. // Compute the first key after the last mutation for nextUserID.
  129. auto userEnd = [FSTLevelDBMutationKey keyPrefixWithUserID:nextUserID];
  130. userEnd = util::PrefixSuccessor(userEnd);
  131. // Seek to that key with the intent of finding the boundary between nextUserID's mutations
  132. // and the one after that (if any).
  133. it->Seek(userEnd);
  134. // At this point there are three possible cases to handle differently. Each case must prepare
  135. // the next iteration (by assigning to nextUserID or setting moreUserIDs = NO) and seek the
  136. // iterator to the last row in the current user's mutation sequence.
  137. if (!it->Valid()) {
  138. // The iterator is past the last row altogether (there are no additional userIDs and now
  139. // rows in any table after mutations). The last row will have the highest batchID.
  140. moreUserIDs = NO;
  141. it->SeekToLast();
  142. } else if ([rowKey decodeKey:it->key()]) {
  143. // The iterator is valid and the key decoded successfully so the next user was just decoded.
  144. nextUserID = rowKey.userID;
  145. it->Prev();
  146. } else {
  147. // The iterator is past the end of the mutations table but there are other rows.
  148. moreUserIDs = NO;
  149. it->Prev();
  150. }
  151. // In all the cases above there was at least one row for the current user and each case has
  152. // set things up such that iterator points to it.
  153. if (![rowKey decodeKey:it->key()]) {
  154. HARD_FAIL("There should have been a key previous to %s", userEnd);
  155. }
  156. if (rowKey.batchID > maxBatchID) {
  157. maxBatchID = rowKey.batchID;
  158. }
  159. }
  160. return maxBatchID + 1;
  161. }
  162. - (BOOL)isEmpty {
  163. std::string userKey = [FSTLevelDBMutationKey keyPrefixWithUserID:self.userID];
  164. auto it = _db.currentTransaction->NewIterator();
  165. it->Seek(userKey);
  166. BOOL empty = YES;
  167. if (it->Valid() && absl::StartsWith(it->key(), userKey)) {
  168. empty = NO;
  169. }
  170. return empty;
  171. }
  172. - (FSTBatchID)highestAcknowledgedBatchID {
  173. return self.metadata.lastAcknowledgedBatchId;
  174. }
  175. - (void)acknowledgeBatch:(FSTMutationBatch *)batch streamToken:(nullable NSData *)streamToken {
  176. FSTBatchID batchID = batch.batchID;
  177. HARD_ASSERT(batchID > self.highestAcknowledgedBatchID,
  178. "Mutation batchIDs must be acknowledged in order");
  179. FSTPBMutationQueue *metadata = self.metadata;
  180. metadata.lastAcknowledgedBatchId = batchID;
  181. metadata.lastStreamToken = streamToken;
  182. _db.currentTransaction->Put([self keyForCurrentMutationQueue], metadata);
  183. }
  184. - (nullable NSData *)lastStreamToken {
  185. return self.metadata.lastStreamToken;
  186. }
  187. - (void)setLastStreamToken:(nullable NSData *)streamToken {
  188. FSTPBMutationQueue *metadata = self.metadata;
  189. metadata.lastStreamToken = streamToken;
  190. _db.currentTransaction->Put([self keyForCurrentMutationQueue], metadata);
  191. }
  192. - (std::string)keyForCurrentMutationQueue {
  193. return [FSTLevelDBMutationQueueKey keyWithUserID:self.userID];
  194. }
  195. - (nullable FSTPBMutationQueue *)metadataForKey:(const std::string &)key {
  196. std::string value;
  197. Status status = _db.currentTransaction->Get(key, &value);
  198. if (status.ok()) {
  199. return [self parsedMetadata:value];
  200. } else if (status.IsNotFound()) {
  201. return nil;
  202. } else {
  203. HARD_FAIL("metadataForKey: failed loading key %s with status: %s", key, status.ToString());
  204. }
  205. }
  206. - (FSTMutationBatch *)addMutationBatchWithWriteTime:(FIRTimestamp *)localWriteTime
  207. mutations:(NSArray<FSTMutation *> *)mutations {
  208. FSTBatchID batchID = self.nextBatchID;
  209. self.nextBatchID += 1;
  210. FSTMutationBatch *batch = [[FSTMutationBatch alloc] initWithBatchID:batchID
  211. localWriteTime:localWriteTime
  212. mutations:mutations];
  213. std::string key = [self mutationKeyForBatch:batch];
  214. _db.currentTransaction->Put(key, [self.serializer encodedMutationBatch:batch]);
  215. NSString *userID = self.userID;
  216. // Store an empty value in the index which is equivalent to serializing a GPBEmpty message. In the
  217. // future if we wanted to store some other kind of value here, we can parse these empty values as
  218. // with some other protocol buffer (and the parser will see all default values).
  219. std::string emptyBuffer;
  220. for (FSTMutation *mutation in mutations) {
  221. key = [FSTLevelDBDocumentMutationKey keyWithUserID:userID
  222. documentKey:mutation.key
  223. batchID:batchID];
  224. _db.currentTransaction->Put(key, emptyBuffer);
  225. }
  226. return batch;
  227. }
  228. - (nullable FSTMutationBatch *)lookupMutationBatch:(FSTBatchID)batchID {
  229. std::string key = [self mutationKeyForBatchID:batchID];
  230. std::string value;
  231. Status status = _db.currentTransaction->Get(key, &value);
  232. if (!status.ok()) {
  233. if (status.IsNotFound()) {
  234. return nil;
  235. }
  236. HARD_FAIL("Lookup mutation batch (%s, %s) failed with status: %s", self.userID, batchID,
  237. status.ToString());
  238. }
  239. return [self decodedMutationBatch:value];
  240. }
  241. - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(FSTBatchID)batchID {
  242. // All batches with batchID <= self.metadata.lastAcknowledgedBatchId have been acknowledged so
  243. // the first unacknowledged batch after batchID will have a batchID larger than both of these
  244. // values.
  245. FSTBatchID nextBatchID = MAX(batchID, self.metadata.lastAcknowledgedBatchId) + 1;
  246. std::string key = [self mutationKeyForBatchID:nextBatchID];
  247. auto it = _db.currentTransaction->NewIterator();
  248. it->Seek(key);
  249. FSTLevelDBMutationKey *rowKey = [[FSTLevelDBMutationKey alloc] init];
  250. if (!it->Valid() || ![rowKey decodeKey:it->key()]) {
  251. // Past the last row in the DB or out of the mutations table
  252. return nil;
  253. }
  254. if (rowKey.userID != [self.userID UTF8String]) {
  255. // Jumped past the last mutation for this user
  256. return nil;
  257. }
  258. HARD_ASSERT(rowKey.batchID >= nextBatchID, "Should have found mutation after %s", nextBatchID);
  259. return [self decodedMutationBatch:it->value()];
  260. }
  261. - (NSArray<FSTMutationBatch *> *)allMutationBatchesThroughBatchID:(FSTBatchID)batchID {
  262. std::string userKey = [FSTLevelDBMutationKey keyPrefixWithUserID:self.userID];
  263. const char *userID = [self.userID UTF8String];
  264. auto it = _db.currentTransaction->NewIterator();
  265. it->Seek(userKey);
  266. NSMutableArray *result = [NSMutableArray array];
  267. FSTLevelDBMutationKey *rowKey = [[FSTLevelDBMutationKey alloc] init];
  268. for (; it->Valid() && [rowKey decodeKey:it->key()]; it->Next()) {
  269. if (rowKey.userID != userID) {
  270. // End of this user's mutations
  271. break;
  272. } else if (rowKey.batchID > batchID) {
  273. // This mutation is past what we're looking for
  274. break;
  275. }
  276. [result addObject:[self decodedMutationBatch:it->value()]];
  277. }
  278. return result;
  279. }
  280. - (NSArray<FSTMutationBatch *> *)allMutationBatchesAffectingDocumentKey:
  281. (const DocumentKey &)documentKey {
  282. NSString *userID = self.userID;
  283. // Scan the document-mutation index starting with a prefix starting with the given documentKey.
  284. std::string indexPrefix = [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID
  285. resourcePath:documentKey.path()];
  286. auto indexIterator = _db.currentTransaction->NewIterator();
  287. indexIterator->Seek(indexPrefix);
  288. // Simultaneously scan the mutation queue. This works because each (key, batchID) pair is unique
  289. // and ordered, so when scanning a table prefixed by exactly key, all the batchIDs encountered
  290. // will be unique and in order.
  291. std::string mutationsPrefix = [FSTLevelDBMutationKey keyPrefixWithUserID:userID];
  292. auto mutationIterator = _db.currentTransaction->NewIterator();
  293. NSMutableArray *result = [NSMutableArray array];
  294. FSTLevelDBDocumentMutationKey *rowKey = [[FSTLevelDBDocumentMutationKey alloc] init];
  295. for (; indexIterator->Valid(); indexIterator->Next()) {
  296. // Only consider rows matching exactly the specific key of interest. Index rows have this
  297. // form (with markers in brackets):
  298. //
  299. // <User>user <Path>collection <Path>doc <BatchId>2 <Terminator>
  300. // <User>user <Path>collection <Path>doc <BatchId>3 <Terminator>
  301. // <User>user <Path>collection <Path>doc <Path>sub <Path>doc <BatchId>3 <Terminator>
  302. //
  303. // Note that Path markers sort after BatchId markers so this means that when searching for
  304. // collection/doc, all the entries for it will be contiguous in the table, allowing a break
  305. // after any mismatch.
  306. if (!absl::StartsWith(indexIterator->key(), indexPrefix) ||
  307. ![rowKey decodeKey:indexIterator->key()] ||
  308. DocumentKey{rowKey.documentKey} != documentKey) {
  309. break;
  310. }
  311. // Each row is a unique combination of key and batchID, so this foreign key reference can
  312. // only occur once.
  313. std::string mutationKey = [FSTLevelDBMutationKey keyWithUserID:userID batchID:rowKey.batchID];
  314. mutationIterator->Seek(mutationKey);
  315. if (!mutationIterator->Valid() || mutationIterator->key() != mutationKey) {
  316. NSString *foundKeyDescription = @"the end of the table";
  317. if (mutationIterator->Valid()) {
  318. foundKeyDescription = [FSTLevelDBKey descriptionForKey:mutationIterator->key()];
  319. }
  320. HARD_FAIL(
  321. "Dangling document-mutation reference found: "
  322. "%s points to %s; seeking there found %s",
  323. [FSTLevelDBKey descriptionForKey:indexIterator->key()],
  324. [FSTLevelDBKey descriptionForKey:mutationKey], foundKeyDescription);
  325. }
  326. [result addObject:[self decodedMutationBatch:mutationIterator->value()]];
  327. }
  328. return result;
  329. }
  330. - (NSArray<FSTMutationBatch *> *)allMutationBatchesAffectingDocumentKeys:
  331. (const DocumentKeySet &)documentKeys {
  332. NSString *userID = self.userID;
  333. // Take a pass through the document keys and collect the set of unique mutation batchIDs that
  334. // affect them all. Some batches can affect more than one key.
  335. std::set<FSTBatchID> batchIDs;
  336. auto indexIterator = _db.currentTransaction->NewIterator();
  337. FSTLevelDBDocumentMutationKey *rowKey = [[FSTLevelDBDocumentMutationKey alloc] init];
  338. for (const DocumentKey &documentKey : documentKeys) {
  339. std::string indexPrefix =
  340. [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:userID resourcePath:documentKey.path()];
  341. for (indexIterator->Seek(indexPrefix); indexIterator->Valid(); indexIterator->Next()) {
  342. // Only consider rows matching exactly the specific key of interest. Index rows have this
  343. // form (with markers in brackets):
  344. //
  345. // <User>user <Path>collection <Path>doc <BatchId>2 <Terminator>
  346. // <User>user <Path>collection <Path>doc <BatchId>3 <Terminator>
  347. // <User>user <Path>collection <Path>doc <Path>sub <Path>doc <BatchId>3 <Terminator>
  348. //
  349. // Note that Path markers sort after BatchId markers so this means that when searching for
  350. // collection/doc, all the entries for it will be contiguous in the table, allowing a break
  351. // after any mismatch.
  352. if (!absl::StartsWith(indexIterator->key(), indexPrefix) ||
  353. ![rowKey decodeKey:indexIterator->key()] ||
  354. DocumentKey{rowKey.documentKey} != documentKey) {
  355. break;
  356. }
  357. batchIDs.insert(rowKey.batchID);
  358. }
  359. }
  360. return [self allMutationBatchesWithBatchIDs:batchIDs];
  361. }
  362. - (NSArray<FSTMutationBatch *> *)allMutationBatchesAffectingQuery:(FSTQuery *)query {
  363. HARD_ASSERT(![query isDocumentQuery], "Document queries shouldn't go down this path");
  364. NSString *userID = self.userID;
  365. const ResourcePath &queryPath = query.path;
  366. size_t immediateChildrenPathLength = queryPath.size() + 1;
  367. // TODO(mcg): Actually implement a single-collection query
  368. //
  369. // This is actually executing an ancestor query, traversing the whole subtree below the
  370. // collection which can be horrifically inefficient for some structures. The right way to
  371. // solve this is to implement the full value index, but that's not in the cards in the near
  372. // future so this is the best we can do for the moment.
  373. //
  374. // Since we don't yet index the actual properties in the mutations, our current approach is to
  375. // just return all mutation batches that affect documents in the collection being queried.
  376. //
  377. // Unlike allMutationBatchesAffectingDocumentKey, this iteration will scan the document-mutation
  378. // index for more than a single document so the associated batchIDs will be neither necessarily
  379. // unique nor in order. This means an efficient simultaneous scan isn't possible.
  380. std::string indexPrefix =
  381. [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:userID resourcePath:queryPath];
  382. auto indexIterator = _db.currentTransaction->NewIterator();
  383. indexIterator->Seek(indexPrefix);
  384. FSTLevelDBDocumentMutationKey *rowKey = [[FSTLevelDBDocumentMutationKey alloc] init];
  385. // Collect up unique batchIDs encountered during a scan of the index. Use a set<FSTBatchID> to
  386. // accumulate batch IDs so they can be traversed in order in a scan of the main table.
  387. //
  388. // This method is faster than performing lookups of the keys with _db->Get and keeping a hash of
  389. // batchIDs that have already been looked up. The performance difference is minor for small
  390. // numbers of keys but > 30% faster for larger numbers of keys.
  391. std::set<FSTBatchID> uniqueBatchIDs;
  392. for (; indexIterator->Valid(); indexIterator->Next()) {
  393. if (!absl::StartsWith(indexIterator->key(), indexPrefix) ||
  394. ![rowKey decodeKey:indexIterator->key()]) {
  395. break;
  396. }
  397. // Rows with document keys more than one segment longer than the query path can't be matches.
  398. // For example, a query on 'rooms' can't match the document /rooms/abc/messages/xyx.
  399. // TODO(mcg): we'll need a different scanner when we implement ancestor queries.
  400. if (rowKey.documentKey.path.size() != immediateChildrenPathLength) {
  401. continue;
  402. }
  403. uniqueBatchIDs.insert(rowKey.batchID);
  404. }
  405. return [self allMutationBatchesWithBatchIDs:uniqueBatchIDs];
  406. }
  407. /**
  408. * Constructs an array of matching batches, sorted by batchID to ensure that multiple mutations
  409. * affecting the same document key are applied in order.
  410. */
  411. - (NSArray<FSTMutationBatch *> *)allMutationBatchesWithBatchIDs:
  412. (const std::set<FSTBatchID> &)batchIDs {
  413. NSMutableArray *result = [NSMutableArray array];
  414. NSString *userID = self.userID;
  415. // Given an ordered set of unique batchIDs perform a skipping scan over the main table to find
  416. // the mutation batches.
  417. auto mutationIterator = _db.currentTransaction->NewIterator();
  418. for (FSTBatchID batchID : batchIDs) {
  419. std::string mutationKey = [FSTLevelDBMutationKey keyWithUserID:userID batchID:batchID];
  420. mutationIterator->Seek(mutationKey);
  421. if (!mutationIterator->Valid() || mutationIterator->key() != mutationKey) {
  422. NSString *foundKeyDescription = @"the end of the table";
  423. if (mutationIterator->Valid()) {
  424. foundKeyDescription = [FSTLevelDBKey descriptionForKey:mutationIterator->key()];
  425. }
  426. HARD_FAIL(
  427. "Dangling document-mutation reference found: "
  428. "Missing batch %s; seeking there found %s",
  429. [FSTLevelDBKey descriptionForKey:mutationKey], foundKeyDescription);
  430. }
  431. [result addObject:[self decodedMutationBatch:mutationIterator->value()]];
  432. }
  433. return result;
  434. }
  435. - (NSArray<FSTMutationBatch *> *)allMutationBatches {
  436. std::string userKey = [FSTLevelDBMutationKey keyPrefixWithUserID:self.userID];
  437. auto it = _db.currentTransaction->NewIterator();
  438. it->Seek(userKey);
  439. NSMutableArray *result = [NSMutableArray array];
  440. for (; it->Valid() && absl::StartsWith(it->key(), userKey); it->Next()) {
  441. [result addObject:[self decodedMutationBatch:it->value()]];
  442. }
  443. return result;
  444. }
  445. - (void)removeMutationBatches:(NSArray<FSTMutationBatch *> *)batches {
  446. NSString *userID = self.userID;
  447. id<FSTGarbageCollector> garbageCollector = self.garbageCollector;
  448. auto checkIterator = _db.currentTransaction->NewIterator();
  449. for (FSTMutationBatch *batch in batches) {
  450. FSTBatchID batchID = batch.batchID;
  451. std::string key = [FSTLevelDBMutationKey keyWithUserID:userID batchID:batchID];
  452. // As a sanity check, verify that the mutation batch exists before deleting it.
  453. checkIterator->Seek(key);
  454. HARD_ASSERT(checkIterator->Valid(), "Mutation batch %s did not exist",
  455. [FSTLevelDBKey descriptionForKey:key]);
  456. HARD_ASSERT(key == checkIterator->key(), "Mutation batch %s not found; found %s",
  457. [FSTLevelDBKey descriptionForKey:key],
  458. [FSTLevelDBKey descriptionForKey:checkIterator->key()]);
  459. _db.currentTransaction->Delete(key);
  460. for (FSTMutation *mutation in batch.mutations) {
  461. key = [FSTLevelDBDocumentMutationKey keyWithUserID:userID
  462. documentKey:mutation.key
  463. batchID:batchID];
  464. _db.currentTransaction->Delete(key);
  465. [_db.referenceDelegate removeMutationReference:mutation.key];
  466. [garbageCollector addPotentialGarbageKey:mutation.key];
  467. }
  468. }
  469. }
  470. - (void)performConsistencyCheck {
  471. if (![self isEmpty]) {
  472. return;
  473. }
  474. // Verify that there are no entries in the document-mutation index if the queue is empty.
  475. std::string indexPrefix = [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID];
  476. auto indexIterator = _db.currentTransaction->NewIterator();
  477. indexIterator->Seek(indexPrefix);
  478. NSMutableArray<NSString *> *danglingMutationReferences = [NSMutableArray array];
  479. for (; indexIterator->Valid(); indexIterator->Next()) {
  480. // Only consider rows matching this index prefix for the current user.
  481. if (!absl::StartsWith(indexIterator->key(), indexPrefix)) {
  482. break;
  483. }
  484. [danglingMutationReferences addObject:[FSTLevelDBKey descriptionForKey:indexIterator->key()]];
  485. }
  486. HARD_ASSERT(danglingMutationReferences.count == 0,
  487. "Document leak -- detected dangling mutation references when queue "
  488. "is empty. Dangling keys: %s",
  489. danglingMutationReferences);
  490. }
  491. - (std::string)mutationKeyForBatch:(FSTMutationBatch *)batch {
  492. return [FSTLevelDBMutationKey keyWithUserID:self.userID batchID:batch.batchID];
  493. }
  494. - (std::string)mutationKeyForBatchID:(FSTBatchID)batchID {
  495. return [FSTLevelDBMutationKey keyWithUserID:self.userID batchID:batchID];
  496. }
  497. /** Parses the MutationQueue metadata from the given LevelDB row contents. */
  498. - (FSTPBMutationQueue *)parsedMetadata:(Slice)slice {
  499. NSData *data =
  500. [[NSData alloc] initWithBytesNoCopy:(void *)slice.data() length:slice.size() freeWhenDone:NO];
  501. NSError *error;
  502. FSTPBMutationQueue *proto = [FSTPBMutationQueue parseFromData:data error:&error];
  503. if (!proto) {
  504. HARD_FAIL("FSTPBMutationQueue failed to parse: %s", error);
  505. }
  506. return proto;
  507. }
  508. - (FSTMutationBatch *)decodedMutationBatch:(absl::string_view)encoded {
  509. NSData *data = [[NSData alloc] initWithBytesNoCopy:(void *)encoded.data()
  510. length:encoded.size()
  511. freeWhenDone:NO];
  512. NSError *error;
  513. FSTPBWriteBatch *proto = [FSTPBWriteBatch parseFromData:data error:&error];
  514. if (!proto) {
  515. HARD_FAIL("FSTPBMutationBatch failed to parse: %s", error);
  516. }
  517. return [self.serializer decodedMutationBatch:proto];
  518. }
  519. #pragma mark - FSTGarbageSource implementation
  520. - (BOOL)containsKey:(const DocumentKey &)documentKey {
  521. std::string indexPrefix = [FSTLevelDBDocumentMutationKey keyPrefixWithUserID:self.userID
  522. resourcePath:documentKey.path()];
  523. auto indexIterator = _db.currentTransaction->NewIterator();
  524. indexIterator->Seek(indexPrefix);
  525. if (indexIterator->Valid()) {
  526. FSTLevelDBDocumentMutationKey *rowKey = [[FSTLevelDBDocumentMutationKey alloc] init];
  527. // Check both that the key prefix matches and that the decoded document key is exactly the key
  528. // we're looking for.
  529. if (absl::StartsWith(indexIterator->key(), indexPrefix) &&
  530. [rowKey decodeKey:indexIterator->key()] && DocumentKey{rowKey.documentKey} == documentKey) {
  531. return YES;
  532. }
  533. }
  534. return NO;
  535. }
  536. @end
  537. NS_ASSUME_NONNULL_END