FSTLocalStore.mm 23 KB

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