fix(aws-datastore): idempotent deletions from merger (#1014)

When deletions are merged to the local store, the storage adapter might
throw an error, if there's nothing left to delete.

The existing code tries to work around this by first checking if there
is a matching instance before attempting to delete it.

Instead, we will swallow deletion failures and log a verbose message.

Resolves: #1011

Author: jamesonwilliams

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/Merger.java

@@ -17,17 +17,14 @@
 
 import androidx.annotation.NonNull;
 
-import com.amplifyframework.core.Action;
 import com.amplifyframework.core.Amplify;
 import com.amplifyframework.core.Consumer;
 import com.amplifyframework.core.NoOpConsumer;
 import com.amplifyframework.core.model.Model;
-import com.amplifyframework.core.model.query.Where;
 import com.amplifyframework.core.model.query.predicate.QueryPredicates;
 import com.amplifyframework.datastore.DataStoreChannelEventName;
 import com.amplifyframework.datastore.appsync.ModelMetadata;
 import com.amplifyframework.datastore.appsync.ModelWithMetadata;
-import com.amplifyframework.datastore.appsync.SerializedModel;
 import com.amplifyframework.datastore.storage.LocalStorageAdapter;
 import com.amplifyframework.datastore.storage.StorageItemChange;
 import com.amplifyframework.hub.HubChannel;
@@ -75,13 +72,14 @@ <T extends Model> Completable merge(ModelWithMetadata<T> modelWithMetadata) {
 
     /**
      * Merge an item back into the local store, using a default strategy.
+     * TODO: Change this method to return a Maybe, and remove the Consumer argument.
      * @param modelWithMetadata A model, combined with metadata about it
-     * @param storageItemChangeConsumer A callback invoked when the merge method saves or deletes the model.
+     * @param changeTypeConsumer A callback invoked when the merge method saves or deletes the model.
      * @param <T> Type of model
      * @return A completable operation to merge the model
      */
-    <T extends Model> Completable merge(ModelWithMetadata<T> modelWithMetadata,
-                                        Consumer<StorageItemChange<T>> storageItemChangeConsumer) {
+    <T extends Model> Completable merge(
+            ModelWithMetadata<T> modelWithMetadata, Consumer<StorageItemChange.Type> changeTypeConsumer) {
         ModelMetadata metadata = modelWithMetadata.getSyncMetadata();
         boolean isDelete = Boolean.TRUE.equals(metadata.isDeleted());
         int incomingVersion = metadata.getVersion() == null ? -1 : metadata.getVersion();
@@ -103,7 +101,7 @@ <T extends Model> Completable merge(ModelWithMetadata<T> modelWithMetadata,
             .filter(currentVersion -> currentVersion == -1 || incomingVersion > currentVersion)
             // If we should merge, then do so now, starting with the model data.
             .flatMapCompletable(shouldMerge ->
-                (isDelete ? delete(model, storageItemChangeConsumer) : save(model, storageItemChangeConsumer))
+                (isDelete ? delete(model, changeTypeConsumer) : save(model, changeTypeConsumer))
                     .andThen(save(metadata, NoOpConsumer.create()))
             )
             // Let the world know that we've done a good thing.
@@ -128,63 +126,35 @@ private <T extends Model> void announceSuccessfulMerge(ModelWithMetadata<T> mode
     }
 
     // Delete a model.
-    private <T extends Model> Completable delete(T model, Consumer<StorageItemChange<T>> onStorageItemChange) {
-        return Completable.defer(() -> Completable.create(emitter -> {
-            // First, check if the thing exists.
-            // If we don't, we'll get an exception saying basically,
-            // "failed to delete a non-existing thing."
-            ifPresent(model,
-                () -> localStorageAdapter.delete(
-                    model,
-                    StorageItemChange.Initiator.SYNC_ENGINE,
-                    QueryPredicates.all(),
-                    storageItemChange -> {
-                        onStorageItemChange.accept(storageItemChange);
-                        emitter.onComplete();
-                    },
-                    emitter::onError
-                ),
-                emitter::onComplete
-            );
-        }));
+    private <T extends Model> Completable delete(T model, Consumer<StorageItemChange.Type> changeTypeConsumer) {
+        return Completable.create(emitter ->
+            localStorageAdapter.delete(model, StorageItemChange.Initiator.SYNC_ENGINE, QueryPredicates.all(),
+                storageItemChange -> {
+                    changeTypeConsumer.accept(storageItemChange.type());
+                    emitter.onComplete();
+                },
+                failure -> {
+                    LOG.verbose(
+                        "Failed to delete a model while merging. Perhaps it was already gone? "
+                        + android.util.Log.getStackTraceString(failure)
+                    );
+                    changeTypeConsumer.accept(StorageItemChange.Type.DELETE);
+                    emitter.onComplete();
+                }
+            )
+        );
     }
 
     // Create or update a model.
-    private <T extends Model> Completable save(T model, Consumer<StorageItemChange<T>> onStorageItemChange) {
-        return Completable.defer(() -> Completable.create(emitter ->
-            localStorageAdapter.save(
-                model,
-                StorageItemChange.Initiator.SYNC_ENGINE,
-                QueryPredicates.all(),
+    private <T extends Model> Completable save(T model, Consumer<StorageItemChange.Type> changeTypeConsumer) {
+        return Completable.create(emitter ->
+            localStorageAdapter.save(model, StorageItemChange.Initiator.SYNC_ENGINE, QueryPredicates.all(),
                 storageItemChange -> {
-                    onStorageItemChange.accept(storageItemChange);
+                    changeTypeConsumer.accept(storageItemChange.type());
                     emitter.onComplete();
                 },
                 emitter::onError
             )
-        ));
-    }
-
-    /**
-     * If the DataStore contains a model instance, then perform an action.
-     * Otherwise, perform some other action.
-     * @param model A model that might exist in local storage
-     * @param onPresent If there is a match, perform this action
-     * @param onNotPresent If there is NOT a match, perform this action as a fallback
-     */
-    private void ifPresent(Model model, Action onPresent, Action onNotPresent) {
-        final String modelName;
-        if (model instanceof SerializedModel) {
-            modelName = ((SerializedModel) model).getModelName();
-        } else {
-            modelName = model.getClass().getSimpleName();
-        }
-        localStorageAdapter.query(modelName, Where.id(model.getId()), iterator -> {
-            if (iterator.hasNext()) {
-                onPresent.call();
-            } else {
-                onNotPresent.call();
-            }
-        }, failure -> onNotPresent.call());
+        );
     }
 }

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/ModelSyncMetricsAccumulator.java

@@ -15,11 +15,8 @@
 
 package com.amplifyframework.datastore.syncengine;
 
-import com.amplifyframework.core.Amplify;
-import com.amplifyframework.core.model.Model;
 import com.amplifyframework.datastore.events.ModelSyncedEvent;
 import com.amplifyframework.datastore.storage.StorageItemChange;
-import com.amplifyframework.logging.Logger;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -30,7 +27,6 @@
  * by operation type for a given model.
  */
 final class ModelSyncMetricsAccumulator {
-    private static final Logger LOG = Amplify.Logging.forNamespace("amplify:aws-datastore");
     private final Map<StorageItemChange.Type, AtomicInteger> syncMetrics;
     private final String modelClassName;
 
@@ -62,9 +58,9 @@ public ModelSyncedEvent toModelSyncedEvent(SyncType syncType) {
 
     /**
      * Increments the counter for a given change type.
-     * @param itemChange The change type to increment.
+     * @param changeType The change type to increment.
      */
-    public void increment(StorageItemChange<? extends Model> itemChange) {
-        syncMetrics.get(itemChange.type()).incrementAndGet();
+    public void increment(StorageItemChange.Type changeType) {
+        syncMetrics.get(changeType).incrementAndGet();
     }
 }