Skip to content

Commit a0ee257

Browse files
5d5d
authored
fix(datastore): sync pending mutation events with latest synced metadata (#3377)
* apply metadata of incoming mutation event which has pending mutations * sync mutation event with latest synced version * add doc for reconcile * resolve comments * remove redundant logic * update test case to verify the latest synced version is applied to api request * resolve comments
1 parent 756548b commit a0ee257

File tree

11 files changed

+186
-677
lines changed

11 files changed

+186
-677
lines changed

‎AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/MutationSync/AWSMutationDatabaseAdapter/AWSMutationDatabaseAdapter+MutationEventIngester.swift‎

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -33,21 +33,6 @@ extension AWSMutationDatabaseAdapter: MutationEventIngester {
3333
func resolveConflictsThenSave(mutationEvent: MutationEvent,
3434
storageAdapter: StorageEngineAdapter,
3535
completion: @escaping (Result<MutationEvent, DataStoreError>)->Void) {
36-
37-
// We don't want to query MutationSync<AnyModel> because a) we already have the model, and b) delete mutations
38-
// are submitted *after* the delete has already been applied to the local data store, meaning there is no model
39-
// to query.
40-
var mutationEvent = mutationEvent
41-
do {
42-
// TODO: Refactor this so that it's clear that the storage engine is not responsible for setting the version
43-
// perhaps as simple as renaming to `submit(unversionedMutationEvent:)` or similar
44-
let syncMetadata = try storageAdapter.queryMutationSyncMetadata(for: mutationEvent.modelId,
45-
modelName: mutationEvent.modelName)
46-
mutationEvent.version = syncMetadata?.version
47-
} catch {
48-
completion(.failure(DataStoreError(error: error)))
49-
}
50-
5136
MutationEvent.pendingMutationEvents(
5237
forMutationEvent: mutationEvent,
5338
storageAdapter: storageAdapter) { result in
@@ -208,8 +193,6 @@ extension AWSMutationDatabaseAdapter: MutationEventIngester {
208193
}
209194
resolvedEvent.mutationType = updatedMutationType
210195

211-
resolvedEvent.version = candidate.version
212-
213196
return resolvedEvent
214197
}
215198

‎AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -204,6 +204,7 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
204204
Task {
205205
let syncMutationToCloudOperation = await SyncMutationToCloudOperation(
206206
mutationEvent: mutationEvent,
207+
getLatestSyncMetadata: { try? self.storageAdapter.queryMutationSyncMetadata(for: mutationEvent.modelId, modelName: mutationEvent.modelName) },
207208
api: api,
208209
authModeStrategy: authModeStrategy
209210
) { [weak self] result in
@@ -257,12 +258,7 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
257258
return
258259
}
259260
reconciliationQueue.offer([mutationSync], modelName: mutationEvent.modelName)
260-
MutationEvent.reconcilePendingMutationEventsVersion(
261-
sent: mutationEvent,
262-
received: mutationSync,
263-
storageAdapter: storageAdapter) { _ in
264-
self.completeProcessingEvent(mutationEvent, mutationSync: mutationSync)
265-
}
261+
completeProcessingEvent(mutationEvent, mutationSync: mutationSync)
266262
} else {
267263
completeProcessingEvent(mutationEvent)
268264
}

‎AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/SyncMutationToCloudOperation.swift‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
1919

2020
private weak var api: APICategoryGraphQLBehaviorExtended?
2121
private let mutationEvent: MutationEvent
22+
private let getLatestSyncMetadata: () -> MutationSyncMetadata?
2223
private let completion: GraphQLOperation<MutationSync<AnyModel>>.ResultListener
2324
private let requestRetryablePolicy: RequestRetryablePolicy
2425

@@ -31,13 +32,15 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
3132
private var authTypesIterator: AWSAuthorizationTypeIterator?
3233

3334
init(mutationEvent: MutationEvent,
35+
getLatestSyncMetadata: @escaping () -> MutationSyncMetadata?,
3436
api: APICategoryGraphQLBehaviorExtended,
3537
authModeStrategy: AuthModeStrategy,
3638
networkReachabilityPublisher: AnyPublisher<ReachabilityUpdate, Never>? = nil,
3739
currentAttemptNumber: Int = 1,
3840
requestRetryablePolicy: RequestRetryablePolicy? = RequestRetryablePolicy(),
3941
completion: @escaping GraphQLOperation<MutationSync<AnyModel>>.ResultListener) async {
4042
self.mutationEvent = mutationEvent
43+
self.getLatestSyncMetadata = getLatestSyncMetadata
4144
self.api = api
4245
self.networkReachabilityPublisher = networkReachabilityPublisher
4346
self.completion = completion
@@ -57,6 +60,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
5760

5861
override func main() {
5962
log.verbose(#function)
63+
6064
sendMutationToCloud(withAuthType: authTypesIterator?.next())
6165
}
6266

@@ -108,6 +112,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
108112
mutationType: GraphQLMutationType,
109113
authType: AWSAuthorizationType? = nil
110114
) -> GraphQLRequest<MutationSync<AnyModel>>? {
115+
let latestSyncMetadata = getLatestSyncMetadata()
111116
var request: GraphQLRequest<MutationSync<AnyModel>>
112117

113118
do {
@@ -128,7 +133,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
128133
request = GraphQLRequest<MutationSyncResult>.deleteMutation(of: model,
129134
modelSchema: modelSchema,
130135
where: graphQLFilter,
131-
version: mutationEvent.version)
136+
version: latestSyncMetadata?.version)
132137
case .update:
133138
let model = try mutationEvent.decodeModel()
134139
guard let modelSchema = ModelRegistry.modelSchema(from: mutationEvent.modelName) else {
@@ -140,7 +145,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
140145
request = GraphQLRequest<MutationSyncResult>.updateMutation(of: model,
141146
modelSchema: modelSchema,
142147
where: graphQLFilter,
143-
version: mutationEvent.version)
148+
version: latestSyncMetadata?.version)
144149
case .create:
145150
let model = try mutationEvent.decodeModel()
146151
guard let modelSchema = ModelRegistry.modelSchema(from: mutationEvent.modelName) else {
@@ -151,7 +156,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
151156
}
152157
request = GraphQLRequest<MutationSyncResult>.createMutation(of: model,
153158
modelSchema: modelSchema,
154-
version: mutationEvent.version)
159+
version: latestSyncMetadata?.version)
155160
}
156161
} catch {
157162
let apiError = APIError.unknown("Couldn't decode model", "", error)

‎AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/ReconcileAndLocalSaveOperation.swift‎

Lines changed: 103 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,14 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
113113

114114
// MARK: - Responder methods
115115

116+
/// The reconcile function incorporates incoming mutation events into the local database through the following steps:
117+
/// 1. Retrieve the local metadata of the models.
118+
/// 2. Generate dispositions based on incoming mutation events and local metadata.
119+
/// 3. Categorize dispositions into:
120+
/// 3.1 Apply metadata only for those with existing pending mutations.
121+
/// 3.1.1 Notify the count of these incoming mutation events as dropped items.
122+
/// 3.2 Apply incoming mutation and metadata for those without existing pending mutations.
123+
/// 4. Notify the final result.
116124
func reconcile(remoteModels: [RemoteModel]) {
117125
guard !isCancelled else {
118126
log.info("\(#function) - cancelled, aborting")
@@ -133,16 +141,21 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
133141

134142
do {
135143
try storageAdapter.transaction {
136-
queryPendingMutations(withModels: remoteModels.map(\.model))
144+
self.queryLocalMetadata(remoteModels)
137145
.subscribe(on: workQueue)
138-
.flatMap { mutationEvents -> Future<([RemoteModel], [LocalMetadata]), DataStoreError> in
139-
let remoteModelsToApply = self.reconcile(remoteModels, pendingMutations: mutationEvents)
140-
return self.queryLocalMetadata(remoteModelsToApply)
146+
.map { (remoteModels, localMetadatas) in
147+
self.getDispositions(for: remoteModels, localMetadatas: localMetadatas)
141148
}
142-
.flatMap { (remoteModelsToApply, localMetadatas) -> Future<Void, DataStoreError> in
143-
let dispositions = self.getDispositions(for: remoteModelsToApply,
144-
localMetadatas: localMetadatas)
145-
return self.applyRemoteModelsDispositions(dispositions)
149+
.flatMap { dispositions in
150+
self.queryPendingMutations(withModels: dispositions.map(\.remoteModel.model))
151+
.map { pendingMutations in (pendingMutations, dispositions) }
152+
}
153+
.map { (pendingMutations, dispositions) in
154+
self.separateDispositions(pendingMutations: pendingMutations, dispositions: dispositions)
155+
}
156+
.flatMap { (dispositions, dispositionOnlyApplyMetadata) in
157+
self.waitAllPublisherFinishes(publishers: dispositionOnlyApplyMetadata.map(self.saveMetadata(disposition:)))
158+
.flatMap { _ in self.applyRemoteModelsDispositions(dispositions) }
146159
}
147160
.sink(
148161
receiveCompletion: {
@@ -203,15 +216,27 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
203216
}
204217
}
205218

206-
func reconcile(_ remoteModels: [RemoteModel], pendingMutations: [MutationEvent]) -> [RemoteModel] {
207-
guard !remoteModels.isEmpty else {
208-
return []
219+
func separateDispositions(
220+
pendingMutations: [MutationEvent],
221+
dispositions: [RemoteSyncReconciler.Disposition]
222+
) -> ([RemoteSyncReconciler.Disposition], [RemoteSyncReconciler.Disposition]) {
223+
guard !dispositions.isEmpty else {
224+
return ([], [])
209225
}
210226

211-
let remoteModelsToApply = RemoteSyncReconciler.filter(remoteModels,
212-
pendingMutations: pendingMutations)
213-
notifyDropped(count: remoteModels.count - remoteModelsToApply.count)
214-
return remoteModelsToApply
227+
228+
let pendingMutationModelIds = Set(pendingMutations.map(\.modelId))
229+
230+
let dispositionsToApply = dispositions.filter {
231+
!pendingMutationModelIds.contains($0.remoteModel.model.identifier)
232+
}
233+
234+
let dispositionsOnlyApplyMetadata = dispositions.filter {
235+
pendingMutationModelIds.contains($0.remoteModel.model.identifier)
236+
}
237+
238+
notifyDropped(count: dispositionsOnlyApplyMetadata.count)
239+
return (dispositionsToApply, dispositionsOnlyApplyMetadata)
215240
}
216241

217242
func queryLocalMetadata(_ remoteModels: [RemoteModel]) -> Future<([RemoteModel], [LocalMetadata]), DataStoreError> {
@@ -269,24 +294,16 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
269294
disposition: RemoteSyncReconciler.Disposition
270295
) -> AnyPublisher<Result<Void, DataStoreError>, Never> {
271296
let operation: Future<ApplyRemoteModelResult, DataStoreError>
272-
let mutationType: MutationEvent.MutationType
273297
switch disposition {
274-
case .create(let remoteModel):
275-
operation = self.save(storageAdapter: storageAdapter, remoteModel: remoteModel)
276-
mutationType = .create
277-
case .update(let remoteModel):
278-
operation = self.save(storageAdapter: storageAdapter, remoteModel: remoteModel)
279-
mutationType = .update
280-
case .delete(let remoteModel):
281-
operation = self.delete(storageAdapter: storageAdapter, remoteModel: remoteModel)
282-
mutationType = .delete
298+
case .create, .update:
299+
operation = self.save(storageAdapter: storageAdapter, remoteModel: disposition.remoteModel)
300+
case .delete:
301+
operation = self.delete(storageAdapter: storageAdapter, remoteModel: disposition.remoteModel)
283302
}
284303

285304
return operation
286-
.flatMap { applyResult in
287-
self.saveMetadata(storageAdapter: storageAdapter, applyResult: applyResult, mutationType: mutationType)
288-
}
289-
.map {_ in Result.success(()) }
305+
.flatMap { self.saveMetadata(storageAdapter: storageAdapter, result: $0, mutationType: disposition.mutationType) }
306+
.map { _ in Result.success(()) }
290307
.catch { Just<Result<Void, DataStoreError>>(.failure($0))}
291308
.eraseToAnyPublisher()
292309
}
@@ -315,15 +332,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
315332
applyRemoteModelsDisposition(storageAdapter: storageAdapter, disposition: $0)
316333
}
317334

318-
return Future { promise in
319-
Publishers.MergeMany(publishers)
320-
.collect()
321-
.sink { _ in
322-
// This stream will never fail, as we wrapped error in the result type.
323-
promise(.successfulVoid)
324-
} receiveValue: { _ in }
325-
.store(in: &self.cancellables)
326-
}
335+
return self.waitAllPublisherFinishes(publishers: publishers)
327336
}
328337

329338
enum ApplyRemoteModelResult {
@@ -359,8 +368,10 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
359368
}
360369
}
361370

362-
private func save(storageAdapter: StorageEngineAdapter,
363-
remoteModel: RemoteModel) -> Future<ApplyRemoteModelResult, DataStoreError> {
371+
private func save(
372+
storageAdapter: StorageEngineAdapter,
373+
remoteModel: RemoteModel
374+
) -> Future<ApplyRemoteModelResult, DataStoreError> {
364375
Future<ApplyRemoteModelResult, DataStoreError> { promise in
365376
storageAdapter.save(untypedModel: remoteModel.model.instance, eagerLoad: self.isEagerLoad) { response in
366377
switch response {
@@ -388,27 +399,50 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
388399
}
389400
}
390401

391-
private func saveMetadata(storageAdapter: StorageEngineAdapter,
392-
applyResult: ApplyRemoteModelResult,
393-
mutationType: MutationEvent.MutationType) -> Future<Void, DataStoreError> {
394-
Future<Void, DataStoreError> { promise in
395-
guard case let .applied(inProcessModel) = applyResult else {
396-
promise(.successfulVoid)
397-
return
398-
}
402+
private func saveMetadata(
403+
disposition: RemoteSyncReconciler.Disposition
404+
) -> AnyPublisher<Void, Never> {
405+
guard let storageAdapter = self.storageAdapter else {
406+
return Just(()).eraseToAnyPublisher()
407+
}
408+
return saveMetadata(storageAdapter: storageAdapter, remoteModel: disposition.remoteModel, mutationType: disposition.mutationType)
409+
.map { _ in () }
410+
.catch { _ in Just(()) }
411+
.eraseToAnyPublisher()
412+
}
399413

400-
storageAdapter.save(inProcessModel.syncMetadata,
401-
condition: nil,
402-
eagerLoad: self.isEagerLoad) { result in
403-
switch result {
404-
case .failure(let dataStoreError):
405-
self.notifyDropped(error: dataStoreError)
406-
promise(.failure(dataStoreError))
407-
case .success(let syncMetadata):
414+
private func saveMetadata(
415+
storageAdapter: StorageEngineAdapter,
416+
result: ApplyRemoteModelResult,
417+
mutationType: MutationEvent.MutationType
418+
) -> AnyPublisher<Void, DataStoreError> {
419+
if case let .applied(inProcessModel) = result {
420+
return self.saveMetadata(storageAdapter: storageAdapter, remoteModel: inProcessModel, mutationType: mutationType)
421+
.handleEvents( receiveOutput: { syncMetadata in
408422
let appliedModel = MutationSync(model: inProcessModel.model, syncMetadata: syncMetadata)
409423
self.notify(savedModel: appliedModel, mutationType: mutationType)
410-
promise(.successfulVoid)
411-
}
424+
}, receiveCompletion: { completion in
425+
if case .failure(let error) = completion {
426+
self.notifyDropped(error: error)
427+
}
428+
})
429+
.map { _ in () }
430+
.eraseToAnyPublisher()
431+
432+
}
433+
return Just(()).setFailureType(to: DataStoreError.self).eraseToAnyPublisher()
434+
}
435+
436+
private func saveMetadata(
437+
storageAdapter: StorageEngineAdapter,
438+
remoteModel: RemoteModel,
439+
mutationType: MutationEvent.MutationType
440+
) -> Future<MutationSyncMetadata, DataStoreError> {
441+
Future { promise in
442+
storageAdapter.save(remoteModel.syncMetadata,
443+
condition: nil,
444+
eagerLoad: self.isEagerLoad) { result in
445+
promise(result)
412446
}
413447
}
414448
}
@@ -454,6 +488,17 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
454488
private static func unfulfilledDataStoreError(name: String = #function) -> DataStoreError {
455489
.unknown("\(name) did not fulfill promise", AmplifyErrorMessages.shouldNotHappenReportBugToAWS(), nil)
456490
}
491+
492+
private func waitAllPublisherFinishes<T>(publishers: [AnyPublisher<T, Never>]) -> Future<Void, DataStoreError> {
493+
Future { promise in
494+
Publishers.MergeMany(publishers)
495+
.collect()
496+
.sink(receiveCompletion: { _ in
497+
promise(.successfulVoid)
498+
}, receiveValue: { _ in })
499+
.store(in: &self.cancellables)
500+
}
501+
}
457502
}
458503

459504
extension ReconcileAndLocalSaveOperation: DefaultLogger {

‎AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/RemoteSyncReconciler.swift‎

Lines changed: 12 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -16,31 +16,24 @@ struct RemoteSyncReconciler {
1616
case create(RemoteModel)
1717
case update(RemoteModel)
1818
case delete(RemoteModel)
19-
}
2019

21-
/// Filter the incoming `remoteModels` against the pending mutations.
22-
/// If there is a matching pending mutation, drop the remote model.
23-
///
24-
/// - Parameters:
25-
/// - remoteModels: models retrieved from the remote store
26-
/// - pendingMutations: pending mutations from the outbox
27-
/// - Returns: remote models to be applied
28-
static func filter(_ remoteModels: [RemoteModel],
29-
pendingMutations: [MutationEvent]) -> [RemoteModel] {
30-
guard !pendingMutations.isEmpty else {
31-
return remoteModels
20+
var remoteModel: RemoteModel {
21+
switch self {
22+
case .create(let model), .update(let model), .delete(let model):
23+
return model
24+
}
3225
}
3326

34-
let pendingMutationModelIdsArr = pendingMutations.map { mutationEvent in
35-
mutationEvent.modelId
36-
}
37-
let pendingMutationModelIds = Set(pendingMutationModelIdsArr)
38-
39-
return remoteModels.filter { remoteModel in
40-
!pendingMutationModelIds.contains(remoteModel.model.identifier)
27+
var mutationType: MutationEvent.MutationType {
28+
switch self {
29+
case .create: return .create
30+
case .update: return .update
31+
case .delete: return .delete
32+
}
4133
}
4234
}
4335

36+
4437
/// Reconciles the incoming `remoteModels` against the local metadata to get the disposition
4538
///
4639
/// - Parameters:

0 commit comments

Comments
 (0)