FSTLocalStore.mm 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568
  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/FSTLocalStore.h"
  17. #include <set>
  18. #import "FIRTimestamp.h"
  19. #import "Firestore/Source/Core/FSTListenSequence.h"
  20. #import "Firestore/Source/Core/FSTQuery.h"
  21. #import "Firestore/Source/Local/FSTLocalDocumentsView.h"
  22. #import "Firestore/Source/Local/FSTLocalViewChanges.h"
  23. #import "Firestore/Source/Local/FSTLocalWriteResult.h"
  24. #import "Firestore/Source/Local/FSTMutationQueue.h"
  25. #import "Firestore/Source/Local/FSTPersistence.h"
  26. #import "Firestore/Source/Local/FSTQueryCache.h"
  27. #import "Firestore/Source/Local/FSTQueryData.h"
  28. #import "Firestore/Source/Local/FSTReferenceSet.h"
  29. #import "Firestore/Source/Local/FSTRemoteDocumentCache.h"
  30. #import "Firestore/Source/Model/FSTDocument.h"
  31. #import "Firestore/Source/Model/FSTDocumentDictionary.h"
  32. #import "Firestore/Source/Model/FSTMutation.h"
  33. #import "Firestore/Source/Model/FSTMutationBatch.h"
  34. #import "Firestore/Source/Remote/FSTRemoteEvent.h"
  35. #include "Firestore/core/src/firebase/firestore/auth/user.h"
  36. #include "Firestore/core/src/firebase/firestore/core/target_id_generator.h"
  37. #include "Firestore/core/src/firebase/firestore/model/document_key.h"
  38. #include "Firestore/core/src/firebase/firestore/model/snapshot_version.h"
  39. #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
  40. #include "Firestore/core/src/firebase/firestore/util/log.h"
  41. using firebase::firestore::auth::User;
  42. using firebase::firestore::core::TargetIdGenerator;
  43. using firebase::firestore::model::BatchId;
  44. using firebase::firestore::model::DocumentKey;
  45. using firebase::firestore::model::DocumentKeySet;
  46. using firebase::firestore::model::DocumentVersionMap;
  47. using firebase::firestore::model::ListenSequenceNumber;
  48. using firebase::firestore::model::SnapshotVersion;
  49. using firebase::firestore::model::TargetId;
  50. NS_ASSUME_NONNULL_BEGIN
  51. /**
  52. * The maximum time to leave a resume token buffered without writing it out. This value is
  53. * arbitrary: it's long enough to avoid several writes (possibly indefinitely if updates come more
  54. * frequently than this) but short enough that restarting after crashing will still have a pretty
  55. * recent resume token.
  56. */
  57. static const int64_t kResumeTokenMaxAgeSeconds = 5 * 60; // 5 minutes
  58. @interface FSTLocalStore ()
  59. /** Manages our in-memory or durable persistence. */
  60. @property(nonatomic, strong, readonly) id<FSTPersistence> persistence;
  61. /** The set of all mutations that have been sent but not yet been applied to the backend. */
  62. @property(nonatomic, strong) id<FSTMutationQueue> mutationQueue;
  63. /** The set of all cached remote documents. */
  64. @property(nonatomic, strong) id<FSTRemoteDocumentCache> remoteDocumentCache;
  65. /** The "local" view of all documents (layering mutationQueue on top of remoteDocumentCache). */
  66. @property(nonatomic, strong) FSTLocalDocumentsView *localDocuments;
  67. /** The set of document references maintained by any local views. */
  68. @property(nonatomic, strong) FSTReferenceSet *localViewReferences;
  69. /** Maps a query to the data about that query. */
  70. @property(nonatomic, strong) id<FSTQueryCache> queryCache;
  71. /** Maps a targetID to data about its query. */
  72. @property(nonatomic, strong) NSMutableDictionary<NSNumber *, FSTQueryData *> *targetIDs;
  73. /**
  74. * A heldBatchResult is a mutation batch result (from a write acknowledgement) that arrived before
  75. * the watch stream got notified of a snapshot that includes the write.  So we "hold" it until
  76. * the watch stream catches up. It ensures that the local write remains visible (latency
  77. * compensation) and doesn't temporarily appear reverted because the watch stream is slower than
  78. * the write stream and so wasn't reflecting it.
  79. *
  80. * NOTE: Eventually we want to move this functionality into the remote store.
  81. */
  82. @property(nonatomic, strong) NSMutableArray<FSTMutationBatchResult *> *heldBatchResults;
  83. @end
  84. @implementation FSTLocalStore {
  85. /** Used to generate targetIDs for queries tracked locally. */
  86. TargetIdGenerator _targetIDGenerator;
  87. }
  88. - (instancetype)initWithPersistence:(id<FSTPersistence>)persistence
  89. initialUser:(const User &)initialUser {
  90. if (self = [super init]) {
  91. _persistence = persistence;
  92. _mutationQueue = [persistence mutationQueueForUser:initialUser];
  93. _remoteDocumentCache = [persistence remoteDocumentCache];
  94. _queryCache = [persistence queryCache];
  95. _localDocuments = [FSTLocalDocumentsView viewWithRemoteDocumentCache:_remoteDocumentCache
  96. mutationQueue:_mutationQueue];
  97. _localViewReferences = [[FSTReferenceSet alloc] init];
  98. [_persistence.referenceDelegate addInMemoryPins:_localViewReferences];
  99. _targetIDs = [NSMutableDictionary dictionary];
  100. _heldBatchResults = [NSMutableArray array];
  101. _targetIDGenerator = TargetIdGenerator::LocalStoreTargetIdGenerator(0);
  102. }
  103. return self;
  104. }
  105. - (void)start {
  106. [self startMutationQueue];
  107. TargetId targetID = [self.queryCache highestTargetID];
  108. _targetIDGenerator = TargetIdGenerator::LocalStoreTargetIdGenerator(targetID);
  109. }
  110. - (void)startMutationQueue {
  111. self.persistence.run("Start MutationQueue", [&]() {
  112. [self.mutationQueue start];
  113. // If we have any leftover mutation batch results from a prior run, just drop them.
  114. // TODO(http://b/33446471): We probably need to repopulate heldBatchResults or similar instead,
  115. // but that is not straightforward since we're not persisting the write ack versions.
  116. [self.heldBatchResults removeAllObjects];
  117. // TODO(mikelehen): This is the only usage of getAllMutationBatchesThroughBatchId:. Consider
  118. // removing it in favor of a getAcknowledgedBatches method.
  119. BatchId highestAck = [self.mutationQueue highestAcknowledgedBatchID];
  120. if (highestAck != kFSTBatchIDUnknown) {
  121. NSArray<FSTMutationBatch *> *batches =
  122. [self.mutationQueue allMutationBatchesThroughBatchID:highestAck];
  123. if (batches.count > 0) {
  124. // NOTE: This could be more efficient if we had a removeBatchesThroughBatchID, but this set
  125. // should be very small and this code should go away eventually.
  126. [self.mutationQueue removeMutationBatches:batches];
  127. }
  128. }
  129. });
  130. }
  131. - (FSTMaybeDocumentDictionary *)userDidChange:(const User &)user {
  132. // Swap out the mutation queue, grabbing the pending mutation batches before and after.
  133. NSArray<FSTMutationBatch *> *oldBatches = self.persistence.run(
  134. "OldBatches",
  135. [&]() -> NSArray<FSTMutationBatch *> * { return [self.mutationQueue allMutationBatches]; });
  136. self.mutationQueue = [self.persistence mutationQueueForUser:user];
  137. [self startMutationQueue];
  138. return self.persistence.run("NewBatches", [&]() -> FSTMaybeDocumentDictionary * {
  139. NSArray<FSTMutationBatch *> *newBatches = [self.mutationQueue allMutationBatches];
  140. // Recreate our LocalDocumentsView using the new MutationQueue.
  141. self.localDocuments =
  142. [FSTLocalDocumentsView viewWithRemoteDocumentCache:self.remoteDocumentCache
  143. mutationQueue:self.mutationQueue];
  144. // Union the old/new changed keys.
  145. DocumentKeySet changedKeys;
  146. for (NSArray<FSTMutationBatch *> *batches in @[ oldBatches, newBatches ]) {
  147. for (FSTMutationBatch *batch in batches) {
  148. for (FSTMutation *mutation in batch.mutations) {
  149. changedKeys = changedKeys.insert(mutation.key);
  150. }
  151. }
  152. }
  153. // Return the set of all (potentially) changed documents as the result of the user change.
  154. return [self.localDocuments documentsForKeys:changedKeys];
  155. });
  156. }
  157. - (FSTLocalWriteResult *)locallyWriteMutations:(NSArray<FSTMutation *> *)mutations {
  158. return self.persistence.run("Locally write mutations", [&]() -> FSTLocalWriteResult * {
  159. FIRTimestamp *localWriteTime = [FIRTimestamp timestamp];
  160. FSTMutationBatch *batch =
  161. [self.mutationQueue addMutationBatchWithWriteTime:localWriteTime mutations:mutations];
  162. DocumentKeySet keys = [batch keys];
  163. FSTMaybeDocumentDictionary *changedDocuments = [self.localDocuments documentsForKeys:keys];
  164. return [FSTLocalWriteResult resultForBatchID:batch.batchID changes:changedDocuments];
  165. });
  166. }
  167. - (FSTMaybeDocumentDictionary *)acknowledgeBatchWithResult:(FSTMutationBatchResult *)batchResult {
  168. return self.persistence.run("Acknowledge batch", [&]() -> FSTMaybeDocumentDictionary * {
  169. id<FSTMutationQueue> mutationQueue = self.mutationQueue;
  170. [mutationQueue acknowledgeBatch:batchResult.batch streamToken:batchResult.streamToken];
  171. DocumentKeySet affected;
  172. if ([self shouldHoldBatchResultWithVersion:batchResult.commitVersion]) {
  173. [self.heldBatchResults addObject:batchResult];
  174. } else {
  175. affected = [self releaseBatchResults:@[ batchResult ]];
  176. }
  177. [self.mutationQueue performConsistencyCheck];
  178. return [self.localDocuments documentsForKeys:affected];
  179. });
  180. }
  181. - (FSTMaybeDocumentDictionary *)rejectBatchID:(BatchId)batchID {
  182. return self.persistence.run("Reject batch", [&]() -> FSTMaybeDocumentDictionary * {
  183. FSTMutationBatch *toReject = [self.mutationQueue lookupMutationBatch:batchID];
  184. HARD_ASSERT(toReject, "Attempt to reject nonexistent batch!");
  185. BatchId lastAcked = [self.mutationQueue highestAcknowledgedBatchID];
  186. HARD_ASSERT(batchID > lastAcked, "Acknowledged batches can't be rejected.");
  187. DocumentKeySet affected = [self removeMutationBatch:toReject];
  188. [self.mutationQueue performConsistencyCheck];
  189. return [self.localDocuments documentsForKeys:affected];
  190. });
  191. }
  192. - (nullable NSData *)lastStreamToken {
  193. return [self.mutationQueue lastStreamToken];
  194. }
  195. - (void)setLastStreamToken:(nullable NSData *)streamToken {
  196. self.persistence.run("Set stream token",
  197. [&]() { [self.mutationQueue setLastStreamToken:streamToken]; });
  198. }
  199. - (const SnapshotVersion &)lastRemoteSnapshotVersion {
  200. return [self.queryCache lastRemoteSnapshotVersion];
  201. }
  202. - (FSTMaybeDocumentDictionary *)applyRemoteEvent:(FSTRemoteEvent *)remoteEvent {
  203. return self.persistence.run("Apply remote event", [&]() -> FSTMaybeDocumentDictionary * {
  204. // TODO(gsoltis): move the sequence number into the reference delegate.
  205. ListenSequenceNumber sequenceNumber = self.persistence.currentSequenceNumber;
  206. id<FSTQueryCache> queryCache = self.queryCache;
  207. DocumentKeySet authoritativeUpdates;
  208. for (const auto &entry : remoteEvent.targetChanges) {
  209. TargetId targetID = entry.first;
  210. FSTBoxedTargetID *boxedTargetID = @(targetID);
  211. FSTTargetChange *change = entry.second;
  212. // Do not ref/unref unassigned targetIDs - it may lead to leaks.
  213. FSTQueryData *queryData = self.targetIDs[boxedTargetID];
  214. if (!queryData) {
  215. continue;
  216. }
  217. // When a global snapshot contains updates (either add or modify) we can completely trust
  218. // these updates as authoritative and blindly apply them to our cache (as a defensive measure
  219. // to promote self-healing in the unfortunate case that our cache is ever somehow corrupted /
  220. // out-of-sync).
  221. //
  222. // If the document is only updated while removing it from a target then watch isn't obligated
  223. // to send the absolute latest version: it can send the first version that caused the document
  224. // not to match.
  225. for (const DocumentKey &key : change.addedDocuments) {
  226. authoritativeUpdates = authoritativeUpdates.insert(key);
  227. }
  228. for (const DocumentKey &key : change.modifiedDocuments) {
  229. authoritativeUpdates = authoritativeUpdates.insert(key);
  230. }
  231. [queryCache removeMatchingKeys:change.removedDocuments forTargetID:targetID];
  232. [queryCache addMatchingKeys:change.addedDocuments forTargetID:targetID];
  233. // Update the resume token if the change includes one. Don't clear any preexisting value.
  234. // Bump the sequence number as well, so that documents being removed now are ordered later
  235. // than documents that were previously removed from this target.
  236. NSData *resumeToken = change.resumeToken;
  237. if (resumeToken.length > 0) {
  238. FSTQueryData *oldQueryData = queryData;
  239. queryData = [queryData queryDataByReplacingSnapshotVersion:remoteEvent.snapshotVersion
  240. resumeToken:resumeToken
  241. sequenceNumber:sequenceNumber];
  242. self.targetIDs[boxedTargetID] = queryData;
  243. if ([self shouldPersistQueryData:queryData oldQueryData:oldQueryData change:change]) {
  244. [self.queryCache updateQueryData:queryData];
  245. }
  246. }
  247. }
  248. // TODO(klimt): This could probably be an NSMutableDictionary.
  249. DocumentKeySet changedDocKeys;
  250. const DocumentKeySet &limboDocuments = remoteEvent.limboDocumentChanges;
  251. for (const auto &kv : remoteEvent.documentUpdates) {
  252. const DocumentKey &key = kv.first;
  253. FSTMaybeDocument *doc = kv.second;
  254. changedDocKeys = changedDocKeys.insert(key);
  255. FSTMaybeDocument *existingDoc = [self.remoteDocumentCache entryForKey:key];
  256. // If a document update isn't authoritative, make sure we don't apply an old document version
  257. // to the remote cache. We make an exception for SnapshotVersion.MIN which can happen for
  258. // manufactured events (e.g. in the case of a limbo document resolution failing).
  259. if (!existingDoc || doc.version == SnapshotVersion::None() ||
  260. authoritativeUpdates.contains(doc.key) || doc.version >= existingDoc.version) {
  261. [self.remoteDocumentCache addEntry:doc];
  262. } else {
  263. LOG_DEBUG(
  264. "FSTLocalStore Ignoring outdated watch update for %s. "
  265. "Current version: %s Watch version: %s",
  266. key.ToString(), existingDoc.version.timestamp().ToString(),
  267. doc.version.timestamp().ToString());
  268. }
  269. // If this was a limbo resolution, make sure we mark when it was accessed.
  270. if (limboDocuments.contains(key)) {
  271. [self.persistence.referenceDelegate limboDocumentUpdated:key];
  272. }
  273. }
  274. // HACK: The only reason we allow omitting snapshot version is so we can synthesize remote
  275. // events when we get permission denied errors while trying to resolve the state of a locally
  276. // cached document that is in limbo.
  277. const SnapshotVersion &lastRemoteVersion = [self.queryCache lastRemoteSnapshotVersion];
  278. const SnapshotVersion &remoteVersion = remoteEvent.snapshotVersion;
  279. if (remoteVersion != SnapshotVersion::None()) {
  280. HARD_ASSERT(remoteVersion >= lastRemoteVersion,
  281. "Watch stream reverted to previous snapshot?? (%s < %s)",
  282. remoteVersion.timestamp().ToString(), lastRemoteVersion.timestamp().ToString());
  283. [self.queryCache setLastRemoteSnapshotVersion:remoteVersion];
  284. }
  285. DocumentKeySet releasedWriteKeys = [self releaseHeldBatchResults];
  286. // Union the two key sets.
  287. DocumentKeySet keysToRecalc = changedDocKeys;
  288. for (const DocumentKey &key : releasedWriteKeys) {
  289. keysToRecalc = keysToRecalc.insert(key);
  290. }
  291. return [self.localDocuments documentsForKeys:keysToRecalc];
  292. });
  293. }
  294. /**
  295. * Returns YES if the newQueryData should be persisted during an update of an active target.
  296. * QueryData should always be persisted when a target is being released and should not call this
  297. * function.
  298. *
  299. * While the target is active, QueryData updates can be omitted when nothing about the target has
  300. * changed except metadata like the resume token or snapshot version. Occasionally it's worth the
  301. * extra write to prevent these values from getting too stale after a crash, but this doesn't have
  302. * to be too frequent.
  303. */
  304. - (BOOL)shouldPersistQueryData:(FSTQueryData *)newQueryData
  305. oldQueryData:(FSTQueryData *)oldQueryData
  306. change:(FSTTargetChange *)change {
  307. // Avoid clearing any existing value
  308. if (newQueryData.resumeToken.length == 0) return NO;
  309. // Any resume token is interesting if there isn't one already.
  310. if (oldQueryData.resumeToken.length == 0) return YES;
  311. // Don't allow resume token changes to be buffered indefinitely. This allows us to be reasonably
  312. // up-to-date after a crash and avoids needing to loop over all active queries on shutdown.
  313. // Especially in the browser we may not get time to do anything interesting while the current
  314. // tab is closing.
  315. int64_t newSeconds = newQueryData.snapshotVersion.timestamp().seconds();
  316. int64_t oldSeconds = oldQueryData.snapshotVersion.timestamp().seconds();
  317. int64_t timeDelta = newSeconds - oldSeconds;
  318. if (timeDelta >= kResumeTokenMaxAgeSeconds) return YES;
  319. // Otherwise if the only thing that has changed about a target is its resume token then it's not
  320. // worth persisting. Note that the RemoteStore keeps an in-memory view of the currently active
  321. // targets which includes the current resume token, so stream failure or user changes will still
  322. // use an up-to-date resume token regardless of what we do here.
  323. size_t changes = change.addedDocuments.size() + change.modifiedDocuments.size() +
  324. change.removedDocuments.size();
  325. return changes > 0;
  326. }
  327. - (void)notifyLocalViewChanges:(NSArray<FSTLocalViewChanges *> *)viewChanges {
  328. self.persistence.run("NotifyLocalViewChanges", [&]() {
  329. FSTReferenceSet *localViewReferences = self.localViewReferences;
  330. for (FSTLocalViewChanges *viewChange in viewChanges) {
  331. for (const DocumentKey &key : viewChange.removedKeys) {
  332. [self->_persistence.referenceDelegate removeReference:key];
  333. }
  334. [localViewReferences addReferencesToKeys:viewChange.addedKeys forID:viewChange.targetID];
  335. [localViewReferences removeReferencesToKeys:viewChange.removedKeys forID:viewChange.targetID];
  336. }
  337. });
  338. }
  339. - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(BatchId)batchID {
  340. FSTMutationBatch *result =
  341. self.persistence.run("NextMutationBatchAfterBatchID", [&]() -> FSTMutationBatch * {
  342. return [self.mutationQueue nextMutationBatchAfterBatchID:batchID];
  343. });
  344. return result;
  345. }
  346. - (nullable FSTMaybeDocument *)readDocument:(const DocumentKey &)key {
  347. return self.persistence.run("ReadDocument", [&]() -> FSTMaybeDocument *_Nullable {
  348. return [self.localDocuments documentForKey:key];
  349. });
  350. }
  351. - (FSTQueryData *)allocateQuery:(FSTQuery *)query {
  352. FSTQueryData *queryData = self.persistence.run("Allocate query", [&]() -> FSTQueryData * {
  353. FSTQueryData *cached = [self.queryCache queryDataForQuery:query];
  354. // TODO(mcg): freshen last accessed date if cached exists?
  355. if (!cached) {
  356. cached = [[FSTQueryData alloc] initWithQuery:query
  357. targetID:_targetIDGenerator.NextId()
  358. listenSequenceNumber:self.persistence.currentSequenceNumber
  359. purpose:FSTQueryPurposeListen];
  360. [self.queryCache addQueryData:cached];
  361. }
  362. return cached;
  363. });
  364. // Sanity check to ensure that even when resuming a query it's not currently active.
  365. FSTBoxedTargetID *boxedTargetID = @(queryData.targetID);
  366. HARD_ASSERT(!self.targetIDs[boxedTargetID], "Tried to allocate an already allocated query: %s",
  367. query);
  368. self.targetIDs[boxedTargetID] = queryData;
  369. return queryData;
  370. }
  371. - (void)releaseQuery:(FSTQuery *)query {
  372. self.persistence.run("Release query", [&]() {
  373. FSTQueryData *queryData = [self.queryCache queryDataForQuery:query];
  374. HARD_ASSERT(queryData, "Tried to release nonexistent query: %s", query);
  375. TargetId targetID = queryData.targetID;
  376. FSTBoxedTargetID *boxedTargetID = @(targetID);
  377. FSTQueryData *cachedQueryData = self.targetIDs[boxedTargetID];
  378. if (cachedQueryData.snapshotVersion > queryData.snapshotVersion) {
  379. // If we've been avoiding persisting the resumeToken (see shouldPersistQueryData for
  380. // conditions and rationale) we need to persist the token now because there will no
  381. // longer be an in-memory version to fall back on.
  382. queryData = cachedQueryData;
  383. [self.queryCache updateQueryData:queryData];
  384. }
  385. [self.localViewReferences removeReferencesForID:targetID];
  386. [self.targetIDs removeObjectForKey:boxedTargetID];
  387. [self.persistence.referenceDelegate removeTarget:queryData];
  388. // If this was the last watch target, then we won't get any more watch snapshots, so we should
  389. // release any held batch results.
  390. if ([self.targetIDs count] == 0) {
  391. [self releaseHeldBatchResults];
  392. }
  393. });
  394. }
  395. - (FSTDocumentDictionary *)executeQuery:(FSTQuery *)query {
  396. return self.persistence.run("ExecuteQuery", [&]() -> FSTDocumentDictionary * {
  397. return [self.localDocuments documentsMatchingQuery:query];
  398. });
  399. }
  400. - (DocumentKeySet)remoteDocumentKeysForTarget:(TargetId)targetID {
  401. return self.persistence.run("RemoteDocumentKeysForTarget", [&]() -> DocumentKeySet {
  402. return [self.queryCache matchingKeysForTargetID:targetID];
  403. });
  404. }
  405. /**
  406. * Releases all the held mutation batches up to the current remote version received, and
  407. * applies their mutations to the docs in the remote documents cache.
  408. *
  409. * @return the set of keys of docs that were modified by those writes.
  410. */
  411. - (DocumentKeySet)releaseHeldBatchResults {
  412. NSMutableArray<FSTMutationBatchResult *> *toRelease = [NSMutableArray array];
  413. for (FSTMutationBatchResult *batchResult in self.heldBatchResults) {
  414. if (![self isRemoteUpToVersion:batchResult.commitVersion]) {
  415. break;
  416. }
  417. [toRelease addObject:batchResult];
  418. }
  419. if (toRelease.count == 0) {
  420. return DocumentKeySet{};
  421. } else {
  422. [self.heldBatchResults removeObjectsInRange:NSMakeRange(0, toRelease.count)];
  423. return [self releaseBatchResults:toRelease];
  424. }
  425. }
  426. - (BOOL)isRemoteUpToVersion:(const SnapshotVersion &)version {
  427. // If there are no watch targets, then we won't get remote snapshots, and are always "up-to-date."
  428. return version <= self.queryCache.lastRemoteSnapshotVersion || self.targetIDs.count == 0;
  429. }
  430. - (BOOL)shouldHoldBatchResultWithVersion:(const SnapshotVersion &)version {
  431. // Check if watcher isn't up to date or prior results are already held.
  432. return ![self isRemoteUpToVersion:version] || self.heldBatchResults.count > 0;
  433. }
  434. - (DocumentKeySet)releaseBatchResults:(NSArray<FSTMutationBatchResult *> *)batchResults {
  435. NSMutableArray<FSTMutationBatch *> *batches = [NSMutableArray array];
  436. for (FSTMutationBatchResult *batchResult in batchResults) {
  437. [self applyBatchResult:batchResult];
  438. [batches addObject:batchResult.batch];
  439. }
  440. return [self removeMutationBatches:batches];
  441. }
  442. - (DocumentKeySet)removeMutationBatch:(FSTMutationBatch *)batch {
  443. return [self removeMutationBatches:@[ batch ]];
  444. }
  445. /** Removes all the mutation batches named in the given array. */
  446. - (DocumentKeySet)removeMutationBatches:(NSArray<FSTMutationBatch *> *)batches {
  447. DocumentKeySet affectedDocs;
  448. for (FSTMutationBatch *batch in batches) {
  449. for (FSTMutation *mutation in batch.mutations) {
  450. const DocumentKey &key = mutation.key;
  451. affectedDocs = affectedDocs.insert(key);
  452. }
  453. }
  454. [self.mutationQueue removeMutationBatches:batches];
  455. return affectedDocs;
  456. }
  457. - (void)applyBatchResult:(FSTMutationBatchResult *)batchResult {
  458. FSTMutationBatch *batch = batchResult.batch;
  459. DocumentKeySet docKeys = batch.keys;
  460. const DocumentVersionMap &versions = batchResult.docVersions;
  461. for (const DocumentKey &docKey : docKeys) {
  462. FSTMaybeDocument *_Nullable remoteDoc = [self.remoteDocumentCache entryForKey:docKey];
  463. FSTMaybeDocument *_Nullable doc = remoteDoc;
  464. auto ackVersionIter = versions.find(docKey);
  465. HARD_ASSERT(ackVersionIter != versions.end(),
  466. "docVersions should contain every doc in the write.");
  467. const SnapshotVersion &ackVersion = ackVersionIter->second;
  468. if (!doc || doc.version < ackVersion) {
  469. doc = [batch applyTo:doc documentKey:docKey mutationBatchResult:batchResult];
  470. if (!doc) {
  471. HARD_ASSERT(!remoteDoc, "Mutation batch %s applied to document %s resulted in nil.", batch,
  472. remoteDoc);
  473. } else {
  474. [self.remoteDocumentCache addEntry:doc];
  475. }
  476. }
  477. }
  478. }
  479. @end
  480. NS_ASSUME_NONNULL_END