FSTLocalStore.mm 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560
  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. #import "FIRTimestamp.h"
  18. #import "Firestore/Source/Core/FSTListenSequence.h"
  19. #import "Firestore/Source/Core/FSTQuery.h"
  20. #import "Firestore/Source/Core/FSTSnapshotVersion.h"
  21. #import "Firestore/Source/Local/FSTGarbageCollector.h"
  22. #import "Firestore/Source/Local/FSTLocalDocumentsView.h"
  23. #import "Firestore/Source/Local/FSTLocalViewChanges.h"
  24. #import "Firestore/Source/Local/FSTLocalWriteResult.h"
  25. #import "Firestore/Source/Local/FSTMutationQueue.h"
  26. #import "Firestore/Source/Local/FSTPersistence.h"
  27. #import "Firestore/Source/Local/FSTQueryCache.h"
  28. #import "Firestore/Source/Local/FSTQueryData.h"
  29. #import "Firestore/Source/Local/FSTReferenceSet.h"
  30. #import "Firestore/Source/Local/FSTRemoteDocumentCache.h"
  31. #import "Firestore/Source/Local/FSTRemoteDocumentChangeBuffer.h"
  32. #import "Firestore/Source/Model/FSTDocument.h"
  33. #import "Firestore/Source/Model/FSTDocumentDictionary.h"
  34. #import "Firestore/Source/Model/FSTDocumentKey.h"
  35. #import "Firestore/Source/Model/FSTMutation.h"
  36. #import "Firestore/Source/Model/FSTMutationBatch.h"
  37. #import "Firestore/Source/Remote/FSTRemoteEvent.h"
  38. #import "Firestore/Source/Util/FSTAssert.h"
  39. #import "Firestore/Source/Util/FSTLogger.h"
  40. #include "Firestore/core/src/firebase/firestore/auth/user.h"
  41. #include "Firestore/core/src/firebase/firestore/core/target_id_generator.h"
  42. using firebase::firestore::auth::User;
  43. using firebase::firestore::core::TargetIdGenerator;
  44. NS_ASSUME_NONNULL_BEGIN
  45. @interface FSTLocalStore ()
  46. /** Manages our in-memory or durable persistence. */
  47. @property(nonatomic, strong, readonly) id<FSTPersistence> persistence;
  48. /** The set of all mutations that have been sent but not yet been applied to the backend. */
  49. @property(nonatomic, strong) id<FSTMutationQueue> mutationQueue;
  50. /** The set of all cached remote documents. */
  51. @property(nonatomic, strong) id<FSTRemoteDocumentCache> remoteDocumentCache;
  52. /** The "local" view of all documents (layering mutationQueue on top of remoteDocumentCache). */
  53. @property(nonatomic, strong) FSTLocalDocumentsView *localDocuments;
  54. /** The set of document references maintained by any local views. */
  55. @property(nonatomic, strong) FSTReferenceSet *localViewReferences;
  56. /**
  57. * The garbage collector collects documents that should no longer be cached (e.g. if they are no
  58. * longer retained by the above reference sets and the garbage collector is performing eager
  59. * collection).
  60. */
  61. @property(nonatomic, strong) id<FSTGarbageCollector> garbageCollector;
  62. /** Maps a query to the data about that query. */
  63. @property(nonatomic, strong) id<FSTQueryCache> queryCache;
  64. /** Maps a targetID to data about its query. */
  65. @property(nonatomic, strong) NSMutableDictionary<NSNumber *, FSTQueryData *> *targetIDs;
  66. @property(nonatomic, strong) FSTListenSequence *listenSequence;
  67. /**
  68. * A heldBatchResult is a mutation batch result (from a write acknowledgement) that arrived before
  69. * the watch stream got notified of a snapshot that includes the write.  So we "hold" it until
  70. * the watch stream catches up. It ensures that the local write remains visible (latency
  71. * compensation) and doesn't temporarily appear reverted because the watch stream is slower than
  72. * the write stream and so wasn't reflecting it.
  73. *
  74. * NOTE: Eventually we want to move this functionality into the remote store.
  75. */
  76. @property(nonatomic, strong) NSMutableArray<FSTMutationBatchResult *> *heldBatchResults;
  77. @end
  78. @implementation FSTLocalStore {
  79. /** Used to generate targetIDs for queries tracked locally. */
  80. TargetIdGenerator _targetIDGenerator;
  81. }
  82. - (instancetype)initWithPersistence:(id<FSTPersistence>)persistence
  83. garbageCollector:(id<FSTGarbageCollector>)garbageCollector
  84. initialUser:(const User &)initialUser {
  85. if (self = [super init]) {
  86. _persistence = persistence;
  87. _mutationQueue = [persistence mutationQueueForUser:initialUser];
  88. _remoteDocumentCache = [persistence remoteDocumentCache];
  89. _queryCache = [persistence queryCache];
  90. _localDocuments = [FSTLocalDocumentsView viewWithRemoteDocumentCache:_remoteDocumentCache
  91. mutationQueue:_mutationQueue];
  92. _localViewReferences = [[FSTReferenceSet alloc] init];
  93. _garbageCollector = garbageCollector;
  94. [_garbageCollector addGarbageSource:_queryCache];
  95. [_garbageCollector addGarbageSource:_localViewReferences];
  96. [_garbageCollector addGarbageSource:_mutationQueue];
  97. _targetIDs = [NSMutableDictionary dictionary];
  98. _heldBatchResults = [NSMutableArray array];
  99. _targetIDGenerator = TargetIdGenerator::LocalStoreTargetIdGenerator(0);
  100. }
  101. return self;
  102. }
  103. - (void)start {
  104. [self startMutationQueue];
  105. [self startQueryCache];
  106. }
  107. - (void)startMutationQueue {
  108. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Start MutationQueue"];
  109. [self.mutationQueue startWithGroup:group];
  110. // If we have any leftover mutation batch results from a prior run, just drop them.
  111. // TODO(http://b/33446471): We probably need to repopulate heldBatchResults or similar instead,
  112. // but that is not straightforward since we're not persisting the write ack versions.
  113. [self.heldBatchResults removeAllObjects];
  114. // TODO(mikelehen): This is the only usage of getAllMutationBatchesThroughBatchId:. Consider
  115. // removing it in favor of a getAcknowledgedBatches method.
  116. FSTBatchID highestAck = [self.mutationQueue highestAcknowledgedBatchID];
  117. if (highestAck != kFSTBatchIDUnknown) {
  118. NSArray<FSTMutationBatch *> *batches =
  119. [self.mutationQueue allMutationBatchesThroughBatchID:highestAck];
  120. if (batches.count > 0) {
  121. // NOTE: This could be more efficient if we had a removeBatchesThroughBatchID, but this set
  122. // should be very small and this code should go away eventually.
  123. [self.mutationQueue removeMutationBatches:batches group:group];
  124. }
  125. }
  126. [self.persistence commitGroup:group];
  127. }
  128. - (void)startQueryCache {
  129. [self.queryCache start];
  130. FSTTargetID targetID = [self.queryCache highestTargetID];
  131. _targetIDGenerator = TargetIdGenerator::LocalStoreTargetIdGenerator(targetID);
  132. FSTListenSequenceNumber sequenceNumber = [self.queryCache highestListenSequenceNumber];
  133. self.listenSequence = [[FSTListenSequence alloc] initStartingAfter:sequenceNumber];
  134. }
  135. - (void)shutdown {
  136. [self.mutationQueue shutdown];
  137. [self.remoteDocumentCache shutdown];
  138. [self.queryCache shutdown];
  139. }
  140. - (FSTMaybeDocumentDictionary *)userDidChange:(const User &)user {
  141. // Swap out the mutation queue, grabbing the pending mutation batches before and after.
  142. NSArray<FSTMutationBatch *> *oldBatches = [self.mutationQueue allMutationBatches];
  143. [self.mutationQueue shutdown];
  144. [self.garbageCollector removeGarbageSource:self.mutationQueue];
  145. self.mutationQueue = [self.persistence mutationQueueForUser:user];
  146. [self.garbageCollector addGarbageSource:self.mutationQueue];
  147. [self startMutationQueue];
  148. NSArray<FSTMutationBatch *> *newBatches = [self.mutationQueue allMutationBatches];
  149. // Recreate our LocalDocumentsView using the new MutationQueue.
  150. self.localDocuments = [FSTLocalDocumentsView viewWithRemoteDocumentCache:self.remoteDocumentCache
  151. mutationQueue:self.mutationQueue];
  152. // Union the old/new changed keys.
  153. FSTDocumentKeySet *changedKeys = [FSTDocumentKeySet keySet];
  154. for (NSArray<FSTMutationBatch *> *batches in @[ oldBatches, newBatches ]) {
  155. for (FSTMutationBatch *batch in batches) {
  156. for (FSTMutation *mutation in batch.mutations) {
  157. changedKeys = [changedKeys setByAddingObject:mutation.key];
  158. }
  159. }
  160. }
  161. // Return the set of all (potentially) changed documents as the result of the user change.
  162. return [self.localDocuments documentsForKeys:changedKeys];
  163. }
  164. - (FSTLocalWriteResult *)locallyWriteMutations:(NSArray<FSTMutation *> *)mutations {
  165. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Locally write mutations"];
  166. FIRTimestamp *localWriteTime = [FIRTimestamp timestamp];
  167. FSTMutationBatch *batch = [self.mutationQueue addMutationBatchWithWriteTime:localWriteTime
  168. mutations:mutations
  169. group:group];
  170. [self.persistence commitGroup:group];
  171. FSTDocumentKeySet *keys = [batch keys];
  172. FSTMaybeDocumentDictionary *changedDocuments = [self.localDocuments documentsForKeys:keys];
  173. return [FSTLocalWriteResult resultForBatchID:batch.batchID changes:changedDocuments];
  174. }
  175. - (FSTMaybeDocumentDictionary *)acknowledgeBatchWithResult:(FSTMutationBatchResult *)batchResult {
  176. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Acknowledge batch"];
  177. id<FSTMutationQueue> mutationQueue = self.mutationQueue;
  178. [mutationQueue acknowledgeBatch:batchResult.batch
  179. streamToken:batchResult.streamToken
  180. group:group];
  181. FSTDocumentKeySet *affected;
  182. if ([self shouldHoldBatchResultWithVersion:batchResult.commitVersion]) {
  183. [self.heldBatchResults addObject:batchResult];
  184. affected = [FSTDocumentKeySet keySet];
  185. } else {
  186. FSTRemoteDocumentChangeBuffer *remoteDocuments =
  187. [FSTRemoteDocumentChangeBuffer changeBufferWithCache:self.remoteDocumentCache];
  188. affected =
  189. [self releaseBatchResults:@[ batchResult ] group:group remoteDocuments:remoteDocuments];
  190. [remoteDocuments applyToWriteGroup:group];
  191. }
  192. [self.persistence commitGroup:group];
  193. [self.mutationQueue performConsistencyCheck];
  194. return [self.localDocuments documentsForKeys:affected];
  195. }
  196. - (FSTMaybeDocumentDictionary *)rejectBatchID:(FSTBatchID)batchID {
  197. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Reject batch"];
  198. FSTMutationBatch *toReject = [self.mutationQueue lookupMutationBatch:batchID];
  199. FSTAssert(toReject, @"Attempt to reject nonexistent batch!");
  200. FSTBatchID lastAcked = [self.mutationQueue highestAcknowledgedBatchID];
  201. FSTAssert(batchID > lastAcked, @"Acknowledged batches can't be rejected.");
  202. FSTDocumentKeySet *affected = [self removeMutationBatch:toReject group:group];
  203. [self.persistence commitGroup:group];
  204. [self.mutationQueue performConsistencyCheck];
  205. return [self.localDocuments documentsForKeys:affected];
  206. }
  207. - (nullable NSData *)lastStreamToken {
  208. return [self.mutationQueue lastStreamToken];
  209. }
  210. - (void)setLastStreamToken:(nullable NSData *)streamToken {
  211. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Set stream token"];
  212. [self.mutationQueue setLastStreamToken:streamToken group:group];
  213. [self.persistence commitGroup:group];
  214. }
  215. - (FSTSnapshotVersion *)lastRemoteSnapshotVersion {
  216. return [self.queryCache lastRemoteSnapshotVersion];
  217. }
  218. - (FSTMaybeDocumentDictionary *)applyRemoteEvent:(FSTRemoteEvent *)remoteEvent {
  219. id<FSTQueryCache> queryCache = self.queryCache;
  220. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Apply remote event"];
  221. FSTRemoteDocumentChangeBuffer *remoteDocuments =
  222. [FSTRemoteDocumentChangeBuffer changeBufferWithCache:self.remoteDocumentCache];
  223. [remoteEvent.targetChanges enumerateKeysAndObjectsUsingBlock:^(
  224. NSNumber *targetIDNumber, FSTTargetChange *change, BOOL *stop) {
  225. FSTTargetID targetID = targetIDNumber.intValue;
  226. // Do not ref/unref unassigned targetIDs - it may lead to leaks.
  227. FSTQueryData *queryData = self.targetIDs[targetIDNumber];
  228. if (!queryData) {
  229. return;
  230. }
  231. FSTTargetMapping *mapping = change.mapping;
  232. if (mapping) {
  233. // First make sure that all references are deleted.
  234. if ([mapping isKindOfClass:[FSTResetMapping class]]) {
  235. FSTResetMapping *reset = (FSTResetMapping *)mapping;
  236. [queryCache removeMatchingKeysForTargetID:targetID group:group];
  237. [queryCache addMatchingKeys:reset.documents forTargetID:targetID group:group];
  238. } else if ([mapping isKindOfClass:[FSTUpdateMapping class]]) {
  239. FSTUpdateMapping *update = (FSTUpdateMapping *)mapping;
  240. [queryCache removeMatchingKeys:update.removedDocuments forTargetID:targetID group:group];
  241. [queryCache addMatchingKeys:update.addedDocuments forTargetID:targetID group:group];
  242. } else {
  243. FSTFail(@"Unknown mapping type: %@", mapping);
  244. }
  245. }
  246. // Update the resume token if the change includes one. Don't clear any preexisting value.
  247. NSData *resumeToken = change.resumeToken;
  248. if (resumeToken.length > 0) {
  249. queryData = [queryData queryDataByReplacingSnapshotVersion:change.snapshotVersion
  250. resumeToken:resumeToken];
  251. self.targetIDs[targetIDNumber] = queryData;
  252. [self.queryCache updateQueryData:queryData group:group];
  253. }
  254. }];
  255. // TODO(klimt): This could probably be an NSMutableDictionary.
  256. __block FSTDocumentKeySet *changedDocKeys = [FSTDocumentKeySet keySet];
  257. [remoteEvent.documentUpdates
  258. enumerateKeysAndObjectsUsingBlock:^(FSTDocumentKey *key, FSTMaybeDocument *doc, BOOL *stop) {
  259. changedDocKeys = [changedDocKeys setByAddingObject:key];
  260. FSTMaybeDocument *existingDoc = [remoteDocuments entryForKey:key];
  261. // Make sure we don't apply an old document version to the remote cache, though we
  262. // make an exception for [SnapshotVersion noVersion] which can happen for manufactured
  263. // events (e.g. in the case of a limbo document resolution failing).
  264. if (!existingDoc || [doc.version isEqual:[FSTSnapshotVersion noVersion]] ||
  265. [doc.version compare:existingDoc.version] != NSOrderedAscending) {
  266. [remoteDocuments addEntry:doc];
  267. } else {
  268. FSTLog(
  269. @"FSTLocalStore Ignoring outdated watch update for %@. "
  270. "Current version: %@ Watch version: %@",
  271. key, existingDoc.version, doc.version);
  272. }
  273. // The document might be garbage because it was unreferenced by everything.
  274. // Make sure to mark it as garbage if it is...
  275. [self.garbageCollector addPotentialGarbageKey:key];
  276. }];
  277. // HACK: The only reason we allow omitting snapshot version is so we can synthesize remote events
  278. // when we get permission denied errors while trying to resolve the state of a locally cached
  279. // document that is in limbo.
  280. FSTSnapshotVersion *lastRemoteVersion = [self.queryCache lastRemoteSnapshotVersion];
  281. FSTSnapshotVersion *remoteVersion = remoteEvent.snapshotVersion;
  282. if (![remoteVersion isEqual:[FSTSnapshotVersion noVersion]]) {
  283. FSTAssert([remoteVersion compare:lastRemoteVersion] != NSOrderedAscending,
  284. @"Watch stream reverted to previous snapshot?? (%@ < %@)", remoteVersion,
  285. lastRemoteVersion);
  286. [self.queryCache setLastRemoteSnapshotVersion:remoteVersion group:group];
  287. }
  288. FSTDocumentKeySet *releasedWriteKeys =
  289. [self releaseHeldBatchResultsWithGroup:group remoteDocuments:remoteDocuments];
  290. [remoteDocuments applyToWriteGroup:group];
  291. [self.persistence commitGroup:group];
  292. // Union the two key sets.
  293. __block FSTDocumentKeySet *keysToRecalc = changedDocKeys;
  294. [releasedWriteKeys enumerateObjectsUsingBlock:^(FSTDocumentKey *key, BOOL *stop) {
  295. keysToRecalc = [keysToRecalc setByAddingObject:key];
  296. }];
  297. return [self.localDocuments documentsForKeys:keysToRecalc];
  298. }
  299. - (void)notifyLocalViewChanges:(NSArray<FSTLocalViewChanges *> *)viewChanges {
  300. FSTReferenceSet *localViewReferences = self.localViewReferences;
  301. for (FSTLocalViewChanges *view in viewChanges) {
  302. FSTQueryData *queryData = [self.queryCache queryDataForQuery:view.query];
  303. FSTAssert(queryData, @"Local view changes contain unallocated query.");
  304. FSTTargetID targetID = queryData.targetID;
  305. [localViewReferences addReferencesToKeys:view.addedKeys forID:targetID];
  306. [localViewReferences removeReferencesToKeys:view.removedKeys forID:targetID];
  307. }
  308. }
  309. - (nullable FSTMutationBatch *)nextMutationBatchAfterBatchID:(FSTBatchID)batchID {
  310. return [self.mutationQueue nextMutationBatchAfterBatchID:batchID];
  311. }
  312. - (nullable FSTMaybeDocument *)readDocument:(FSTDocumentKey *)key {
  313. return [self.localDocuments documentForKey:key];
  314. }
  315. - (FSTQueryData *)allocateQuery:(FSTQuery *)query {
  316. FSTQueryData *cached = [self.queryCache queryDataForQuery:query];
  317. FSTTargetID targetID;
  318. FSTListenSequenceNumber sequenceNumber = [self.listenSequence next];
  319. if (cached) {
  320. // This query has been listened to previously, so reuse the previous targetID.
  321. // TODO(mcg): freshen last accessed date?
  322. targetID = cached.targetID;
  323. } else {
  324. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Allocate query"];
  325. targetID = _targetIDGenerator.NextId();
  326. cached = [[FSTQueryData alloc] initWithQuery:query
  327. targetID:targetID
  328. listenSequenceNumber:sequenceNumber
  329. purpose:FSTQueryPurposeListen];
  330. [self.queryCache addQueryData:cached group:group];
  331. [self.persistence commitGroup:group];
  332. }
  333. // Sanity check to ensure that even when resuming a query it's not currently active.
  334. FSTBoxedTargetID *boxedTargetID = @(targetID);
  335. FSTAssert(!self.targetIDs[boxedTargetID], @"Tried to allocate an already allocated query: %@",
  336. query);
  337. self.targetIDs[boxedTargetID] = cached;
  338. return cached;
  339. }
  340. - (void)releaseQuery:(FSTQuery *)query {
  341. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Release query"];
  342. FSTQueryData *queryData = [self.queryCache queryDataForQuery:query];
  343. FSTAssert(queryData, @"Tried to release nonexistent query: %@", query);
  344. [self.localViewReferences removeReferencesForID:queryData.targetID];
  345. if (self.garbageCollector.isEager) {
  346. [self.queryCache removeQueryData:queryData group:group];
  347. }
  348. [self.targetIDs removeObjectForKey:@(queryData.targetID)];
  349. // If this was the last watch target, then we won't get any more watch snapshots, so we should
  350. // release any held batch results.
  351. if ([self.targetIDs count] == 0) {
  352. FSTRemoteDocumentChangeBuffer *remoteDocuments =
  353. [FSTRemoteDocumentChangeBuffer changeBufferWithCache:self.remoteDocumentCache];
  354. [self releaseHeldBatchResultsWithGroup:group remoteDocuments:remoteDocuments];
  355. [remoteDocuments applyToWriteGroup:group];
  356. }
  357. [self.persistence commitGroup:group];
  358. }
  359. - (FSTDocumentDictionary *)executeQuery:(FSTQuery *)query {
  360. return [self.localDocuments documentsMatchingQuery:query];
  361. }
  362. - (FSTDocumentKeySet *)remoteDocumentKeysForTarget:(FSTTargetID)targetID {
  363. return [self.queryCache matchingKeysForTargetID:targetID];
  364. }
  365. - (void)collectGarbage {
  366. // Call collectGarbage regardless of whether isGCEnabled so the referenceSet doesn't continue to
  367. // accumulate the garbage keys.
  368. NSSet<FSTDocumentKey *> *garbage = [self.garbageCollector collectGarbage];
  369. if (garbage.count > 0) {
  370. FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Garbage Collection"];
  371. for (FSTDocumentKey *key in garbage) {
  372. [self.remoteDocumentCache removeEntryForKey:key group:group];
  373. }
  374. [self.persistence commitGroup:group];
  375. }
  376. }
  377. /**
  378. * Releases all the held mutation batches up to the current remote version received, and
  379. * applies their mutations to the docs in the remote documents cache.
  380. *
  381. * @return the set of keys of docs that were modified by those writes.
  382. */
  383. - (FSTDocumentKeySet *)releaseHeldBatchResultsWithGroup:(FSTWriteGroup *)group
  384. remoteDocuments:
  385. (FSTRemoteDocumentChangeBuffer *)remoteDocuments {
  386. NSMutableArray<FSTMutationBatchResult *> *toRelease = [NSMutableArray array];
  387. for (FSTMutationBatchResult *batchResult in self.heldBatchResults) {
  388. if (![self isRemoteUpToVersion:batchResult.commitVersion]) {
  389. break;
  390. }
  391. [toRelease addObject:batchResult];
  392. }
  393. if (toRelease.count == 0) {
  394. return [FSTDocumentKeySet keySet];
  395. } else {
  396. [self.heldBatchResults removeObjectsInRange:NSMakeRange(0, toRelease.count)];
  397. return [self releaseBatchResults:toRelease group:group remoteDocuments:remoteDocuments];
  398. }
  399. }
  400. - (BOOL)isRemoteUpToVersion:(FSTSnapshotVersion *)version {
  401. // If there are no watch targets, then we won't get remote snapshots, and are always "up-to-date."
  402. return [version compare:self.queryCache.lastRemoteSnapshotVersion] != NSOrderedDescending ||
  403. self.targetIDs.count == 0;
  404. }
  405. - (BOOL)shouldHoldBatchResultWithVersion:(FSTSnapshotVersion *)version {
  406. // Check if watcher isn't up to date or prior results are already held.
  407. return ![self isRemoteUpToVersion:version] || self.heldBatchResults.count > 0;
  408. }
  409. - (FSTDocumentKeySet *)releaseBatchResults:(NSArray<FSTMutationBatchResult *> *)batchResults
  410. group:(FSTWriteGroup *)group
  411. remoteDocuments:(FSTRemoteDocumentChangeBuffer *)remoteDocuments {
  412. NSMutableArray<FSTMutationBatch *> *batches = [NSMutableArray array];
  413. for (FSTMutationBatchResult *batchResult in batchResults) {
  414. [self applyBatchResult:batchResult toRemoteDocuments:remoteDocuments];
  415. [batches addObject:batchResult.batch];
  416. }
  417. return [self removeMutationBatches:batches group:group];
  418. }
  419. - (FSTDocumentKeySet *)removeMutationBatch:(FSTMutationBatch *)batch group:(FSTWriteGroup *)group {
  420. return [self removeMutationBatches:@[ batch ] group:group];
  421. }
  422. /** Removes all the mutation batches named in the given array. */
  423. - (FSTDocumentKeySet *)removeMutationBatches:(NSArray<FSTMutationBatch *> *)batches
  424. group:(FSTWriteGroup *)group {
  425. // TODO(klimt): Could this be an NSMutableDictionary?
  426. __block FSTDocumentKeySet *affectedDocs = [FSTDocumentKeySet keySet];
  427. for (FSTMutationBatch *batch in batches) {
  428. for (FSTMutation *mutation in batch.mutations) {
  429. FSTDocumentKey *key = mutation.key;
  430. affectedDocs = [affectedDocs setByAddingObject:key];
  431. }
  432. }
  433. [self.mutationQueue removeMutationBatches:batches group:group];
  434. return affectedDocs;
  435. }
  436. - (void)applyBatchResult:(FSTMutationBatchResult *)batchResult
  437. toRemoteDocuments:(FSTRemoteDocumentChangeBuffer *)remoteDocuments {
  438. FSTMutationBatch *batch = batchResult.batch;
  439. FSTDocumentKeySet *docKeys = batch.keys;
  440. [docKeys enumerateObjectsUsingBlock:^(FSTDocumentKey *docKey, BOOL *stop) {
  441. FSTMaybeDocument *_Nullable remoteDoc = [remoteDocuments entryForKey:docKey];
  442. FSTMaybeDocument *_Nullable doc = remoteDoc;
  443. FSTSnapshotVersion *ackVersion = batchResult.docVersions[docKey];
  444. FSTAssert(ackVersion, @"docVersions should contain every doc in the write.");
  445. if (!doc || [doc.version compare:ackVersion] == NSOrderedAscending) {
  446. doc = [batch applyTo:doc documentKey:docKey mutationBatchResult:batchResult];
  447. if (!doc) {
  448. FSTAssert(!remoteDoc, @"Mutation batch %@ applied to document %@ resulted in nil.", batch,
  449. remoteDoc);
  450. } else {
  451. [remoteDocuments addEntry:doc];
  452. }
  453. }
  454. }];
  455. }
  456. @end
  457. NS_ASSUME_NONNULL_END