feat(datastore): Retry sync (#1414)

* Added retry logic on SyncProcessor. https://github.com/orgs/aws-amplify/projects/62#card-56626619

* Updated retry on syncprocessor to retry only on recoverable error.

Updated retry on syncprocessor to retry only on recoverable error.

* Adding irrecoverable error to APIException

* Updating unit test

* Making retry class generic

* Update settings.gradle

Co-authored-by: Richard McClellan <[email protected]>

* Made retrier generic but still using Rx Java

* Made retrier generic but still using Rx Java

* Update aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/RequestRetry.kt

Co-authored-by: Richard McClellan <[email protected]>

* PR suggestions

* Update aws-api/src/main/java/com/amplifyframework/api/aws/AppSyncGraphQLOperation.java

Co-authored-by: Richard McClellan <[email protected]>

* PR suggestions

* PR suggestions

* PR suggestions

* More unit test

* Update aws-datastore/src/main/java/com/amplifyframework/datastore/appsync/AppSyncClient.java

Co-authored-by: Rafael Juliano <[email protected]>

* Update aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/RetryHandler.kt

Co-authored-by: Rafael Juliano <[email protected]>

* PR suggestions!

* checkstyle fixes

* PR Suggestions

* Removing timeout from Orchestrator blocking await.

* Checkstyle fixes.

* Checkstyle fixes.

* Checkstyle fix.

* Checkstyle fix.

* Adding test for retry on subscription disposed.

* Adding maxDelay to retry logic.

* Added isDisposed check on the emitter in retryhandler before retrying and emitting error.

* Added retry on sync enable as a property on builder in awsdatastore plugin constructor.

* adding retry enable as a builder method in constructor

* checkstyle

* Update aws-datastore/src/main/java/com/amplifyframework/datastore/AWSDataStorePlugin.java

Co-authored-by: Rafael Juliano <[email protected]>

* Added code to get syncRetry from userprovidedconfiguration.getDoSyncRetry()

* Push to start git checks.

Co-authored-by: Richard McClellan <[email protected]>
Co-authored-by: Rafael Juliano <[email protected]>

Author: 3 people

aws-api/src/main/java/com/amplifyframework/api/aws/AppSyncGraphQLOperation.java

@@ -50,7 +50,8 @@
 public final class AppSyncGraphQLOperation<R> extends GraphQLOperation<R> {
     private static final Logger LOG = Amplify.Logging.forNamespace("amplify:aws-api");
     private static final String CONTENT_TYPE = "application/json";
-
+    private static final int START_OF_CLIENT_ERROR_CODE = 400;
+    private static final int END_OF_CLIENT_ERROR_CODE = 499;
     private final String endpoint;
     private final OkHttpClient client;
     private final Consumer<GraphQLResponse<R>> onResponse;
@@ -142,6 +143,11 @@ public void onResponse(@NonNull Call call, @NonNull Response response) {
                     return;
                 }
             }
+            if (response.code() >= START_OF_CLIENT_ERROR_CODE && response.code() <= END_OF_CLIENT_ERROR_CODE) {
+                onFailure.accept(new ApiException
+                        .NonRetryableException("OkHttp client request failed.", "Irrecoverable error")
+                );
+            }
 
             try {
                 onResponse.accept(wrapResponse(jsonResponse));

aws-datastore/build.gradle

@@ -35,7 +35,9 @@ dependencies {
     testImplementation dependency.mockito
     testImplementation dependency.robolectric
     testImplementation dependency.androidx.test.core
+    testImplementation 'io.mockk:mockk:1.11.0'
 
+    androidTestImplementation dependency.mockito
     androidTestImplementation project(path: ':testmodels')
     androidTestImplementation project(path: ':testutils')
     androidTestImplementation project(path: ':aws-api')

aws-datastore/src/main/java/com/amplifyframework/datastore/AWSDataStorePlugin.java

@@ -85,6 +85,8 @@ public final class AWSDataStorePlugin extends DataStorePlugin<Void> {
 
     private final AuthModeStrategyType authModeStrategy;
 
+    private final boolean isSyncRetryEnabled;
+
     private AWSDataStorePlugin(
             @NonNull ModelProvider modelProvider,
             @NonNull ModelSchemaRegistry modelSchemaRegistry,
@@ -93,16 +95,19 @@ private AWSDataStorePlugin(
         this.sqliteStorageAdapter = SQLiteStorageAdapter.forModels(modelSchemaRegistry, modelProvider);
         this.categoryInitializationsPending = new CountDownLatch(1);
         this.authModeStrategy = AuthModeStrategyType.DEFAULT;
+        this.userProvidedConfiguration = userProvidedConfiguration;
+        this.isSyncRetryEnabled = userProvidedConfiguration != null && userProvidedConfiguration.getDoSyncRetry();
         // Used to interrogate plugins, to understand if sync should be automatically turned on
         this.orchestrator = new Orchestrator(
             modelProvider,
             modelSchemaRegistry,
             sqliteStorageAdapter,
             AppSyncClient.via(api),
             () -> pluginConfiguration,
-            () -> api.getPlugins().isEmpty() ? Orchestrator.State.LOCAL_ONLY : Orchestrator.State.SYNC_VIA_API
+            () -> api.getPlugins().isEmpty() ? Orchestrator.State.LOCAL_ONLY : Orchestrator.State.SYNC_VIA_API,
+                isSyncRetryEnabled
         );
-        this.userProvidedConfiguration = userProvidedConfiguration;
+
     }
 
     private AWSDataStorePlugin(@NonNull Builder builder) throws DataStoreException {
@@ -115,6 +120,7 @@ private AWSDataStorePlugin(@NonNull Builder builder) throws DataStoreException {
         this.authModeStrategy = builder.authModeStrategy == null ?
             AuthModeStrategyType.DEFAULT :
             builder.authModeStrategy;
+        this.isSyncRetryEnabled = builder.isSyncRetryEnabled;
         ApiCategory api = builder.apiCategory == null ? Amplify.API : builder.apiCategory;
         this.userProvidedConfiguration = builder.dataStoreConfiguration;
         this.sqliteStorageAdapter = builder.storageAdapter == null ?
@@ -129,7 +135,8 @@ private AWSDataStorePlugin(@NonNull Builder builder) throws DataStoreException {
             sqliteStorageAdapter,
             AppSyncClient.via(api, this.authModeStrategy),
             () -> pluginConfiguration,
-            () -> api.getPlugins().isEmpty() ? Orchestrator.State.LOCAL_ONLY : Orchestrator.State.SYNC_VIA_API
+            () -> api.getPlugins().isEmpty() ? Orchestrator.State.LOCAL_ONLY : Orchestrator.State.SYNC_VIA_API,
+            isSyncRetryEnabled
         );
     }
 
@@ -626,6 +633,7 @@ public static final class Builder {
         private ApiCategory apiCategory;
         private AuthModeStrategyType authModeStrategy;
         private LocalStorageAdapter storageAdapter;
+        private boolean isSyncRetryEnabled;
 
         private Builder() {}
 
@@ -693,6 +701,16 @@ public Builder authModeStrategy(AuthModeStrategyType authModeStrategy) {
             return this;
         }
 
+        /**
+         * Enables Retry on DataStore sync engine.
+         * @param isSyncRetryEnabled is sync retry enabled.
+         * @return An implementation of the {@link ModelProvider} interface.
+         */
+        public Builder isSyncRetryEnabled(Boolean isSyncRetryEnabled) {
+            this.isSyncRetryEnabled = isSyncRetryEnabled;
+            return this;
+        }
+
         /**
          * Builds the DataStore plugin.
          * @return An instance of the DataStore plugin ready for use.

aws-datastore/src/main/java/com/amplifyframework/datastore/DataStoreConfiguration.java

@@ -43,11 +43,14 @@ public final class DataStoreConfiguration {
     static final int DEFAULT_SYNC_MAX_RECORDS = 10_000;
     @VisibleForTesting 
     static final int DEFAULT_SYNC_PAGE_SIZE = 1_000;
+    @VisibleForTesting
+    static final boolean DEFAULT_DO_SYNC_RETRY = false;
 
     private final DataStoreErrorHandler errorHandler;
     private final DataStoreConflictHandler conflictHandler;
     private final Integer syncMaxRecords;
     private final Integer syncPageSize;
+    private final boolean doSyncRetry;
     private final Map<String, DataStoreSyncExpression> syncExpressions;
     private final Long syncIntervalInMinutes;
 
@@ -58,6 +61,7 @@ private DataStoreConfiguration(Builder builder) {
         this.syncPageSize = builder.syncPageSize;
         this.syncIntervalInMinutes = builder.syncIntervalInMinutes;
         this.syncExpressions = builder.syncExpressions;
+        this.doSyncRetry = builder.doSyncRetry;
     }
 
     /**
@@ -111,6 +115,7 @@ public static DataStoreConfiguration defaults() throws DataStoreException {
             .syncInterval(DEFAULT_SYNC_INTERVAL_MINUTES, TimeUnit.MINUTES)
             .syncPageSize(DEFAULT_SYNC_PAGE_SIZE)
             .syncMaxRecords(DEFAULT_SYNC_MAX_RECORDS)
+                .doSyncRetry(DEFAULT_DO_SYNC_RETRY)
             .build();
     }
 
@@ -174,6 +179,15 @@ public Integer getSyncPageSize() {
         return this.syncPageSize;
     }
 
+    /**
+     * Gets the boolean for enabling retry on sync failure
+     * a sync operation.
+     * @return Desired size of a page of results from an AppSync sync response
+     */
+    public boolean getDoSyncRetry() {
+        return this.doSyncRetry;
+    }
+
     /**
      * Returns the Map of all {@link DataStoreSyncExpression}s used to filter data received from AppSync, either during
      * a sync or over the real-time subscription.
@@ -247,6 +261,7 @@ public static final class Builder {
         private Long syncIntervalInMinutes;
         private Integer syncMaxRecords;
         private Integer syncPageSize;
+        private boolean doSyncRetry;
         private Map<String, DataStoreSyncExpression> syncExpressions;
         private boolean ensureDefaults;
         private JSONObject pluginJson;
@@ -313,6 +328,17 @@ public Builder syncMaxRecords(@IntRange(from = 0) Integer syncMaxRecords) {
             return Builder.this;
         }
 
+        /**
+         * Sets the retry enabled on datastore sync.
+         * @param doSyncRetry Is retry enabled on datastore sync
+         * @return Current builder instance
+         */
+        @NonNull
+        public Builder doSyncRetry(boolean doSyncRetry) {
+            this.doSyncRetry = doSyncRetry;
+            return Builder.this;
+        }
+
         /**
          * Sets the number of items requested in each page of sync results.
          * @param syncPageSize Number of items requested per page in sync operation
@@ -410,6 +436,7 @@ private void applyUserProvidedConfiguration() {
             syncMaxRecords = getValueOrDefault(userProvidedConfiguration.getSyncMaxRecords(), syncMaxRecords);
             syncPageSize = getValueOrDefault(userProvidedConfiguration.getSyncPageSize(), syncPageSize);
             syncExpressions = userProvidedConfiguration.getSyncExpressions();
+            doSyncRetry = getValueOrDefault(userProvidedConfiguration.getDoSyncRetry(), doSyncRetry);
         }
 
         private static <T> T getValueOrDefault(T value, T defaultValue) {

aws-datastore/src/main/java/com/amplifyframework/datastore/appsync/AppSyncClient.java

@@ -119,9 +119,8 @@ public <T extends Model> Cancelable sync(
         };
         final Consumer<ApiException> failureConsumer =
             failure -> onFailure.accept(new DataStoreException(
-                "Failure performing sync query to AppSync.",
-                failure, AmplifyException.TODO_RECOVERY_SUGGESTION
-            ));
+                        "Failure performing sync query to AppSync.",
+                        failure, AmplifyException.TODO_RECOVERY_SUGGESTION));
 
         final Cancelable cancelable = api.query(request, responseConsumer, failureConsumer);
         if (cancelable != null) {

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

@@ -84,14 +84,16 @@ public final class Orchestrator {
      *        The reference to the variable returned by the provider only get set after the plugin's
      *        {@link AWSDataStorePlugin#configure(JSONObject, Context)} is invoked by Amplify.
      * @param targetState The desired state of operation - online, or offline
+     * @param isSyncRetryEnabled enable or disable the SyncProcessor retry
      */
     public Orchestrator(
             @NonNull final ModelProvider modelProvider,
             @NonNull final ModelSchemaRegistry modelSchemaRegistry,
             @NonNull final LocalStorageAdapter localStorageAdapter,
             @NonNull final AppSync appSync,
             @NonNull final DataStoreConfigurationProvider dataStoreConfigurationProvider,
-            @NonNull final Supplier<State> targetState) {
+            @NonNull final Supplier<State> targetState,
+            final boolean isSyncRetryEnabled) {
         Objects.requireNonNull(modelSchemaRegistry);
         Objects.requireNonNull(modelProvider);
         Objects.requireNonNull(appSync);
@@ -120,6 +122,8 @@ public Orchestrator(
             .merger(merger)
             .dataStoreConfigurationProvider(dataStoreConfigurationProvider)
             .queryPredicateProvider(queryPredicateProvider)
+            .retryHandler(new RetryHandler())
+                .isSyncRetryEnabled(isSyncRetryEnabled)
             .build();
         this.subscriptionProcessor = SubscriptionProcessor.builder()
                 .appSync(appSync)
@@ -264,15 +268,12 @@ private void transitionToApiSync() throws DataStoreException {
     private void startObservingStorageChanges() throws DataStoreException {
         LOG.info("Starting to observe local storage changes.");
         try {
-            boolean subscribed = mutationOutbox.load()
+            mutationOutbox.load()
                 .andThen(Completable.create(emitter -> {
                     storageObserver.startObservingStorageChanges(emitter::onComplete);
                     LOG.info("Setting currentState to LOCAL_ONLY");
                     currentState.set(State.LOCAL_ONLY);
-                })).blockingAwait(LOCAL_OP_TIMEOUT_SECONDS, TimeUnit.SECONDS);
-            if (!subscribed) {
-                throw new TimeoutException("Timed out while preparing local-only mode.");
-            }
+                })).blockingAwait();
         } catch (Throwable throwable) {
             throw new DataStoreException("Timed out while starting to observe storage changes.",
                 throwable,
@@ -292,7 +293,6 @@ private void stopObservingStorageChanges() {
 
     /**
      * Start syncing models to and from a remote API.
-     * @return A Completable that succeeds when API sync is enabled.
      */
     private void startApiSync() {
         LOG.info("Setting currentState to SYNC_VIA_API");

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

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package com.amplifyframework.datastore.syncengine;
+
+import com.amplifyframework.core.Amplify;
+import com.amplifyframework.datastore.utils.ErrorInspector;
+import com.amplifyframework.logging.Logger;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.core.SingleEmitter;
+
+/**
+ * Class for retrying call on failure on a single.
+ */
+public class RetryHandler {
+
+    private static final Logger LOG = Amplify.Logging.forNamespace("amplify:aws-datastore");
+    private static final int MAX_EXPONENT_VALUE = 8;
+    private static final int JITTER_FACTOR_VALUE = 100;
+    private static final int MAX_ATTEMPTS_VALUE = 3;
+    private static final int MAX_DELAY_S_VALUE = 5 * 60;
+    private final int maxExponent;
+    private final int jitterFactor;
+    private final int maxAttempts;
+    private final int maxDelayS;
+
+    /**
+     * Constructor to inject constants for unit testing.
+     * @param maxExponent maxExponent backoff can go to.
+     * @param jitterFactor jitterFactor for backoff.
+     * @param maxAttempts max attempt for retrying.
+     * @param maxDelayS max delay for retrying.
+     */
+    public RetryHandler(int maxExponent,
+                        int jitterFactor,
+                        int maxAttempts,
+                        int maxDelayS) {
+
+        this.maxExponent = maxExponent;
+        this.jitterFactor = jitterFactor;
+        this.maxAttempts = maxAttempts;
+        this.maxDelayS = maxDelayS;
+    }
+
+    /**
+     * Parameter less constructor.
+     */
+    public RetryHandler() {
+        maxExponent = MAX_EXPONENT_VALUE;
+        jitterFactor = JITTER_FACTOR_VALUE;
+        maxAttempts = MAX_ATTEMPTS_VALUE;
+        maxDelayS = MAX_DELAY_S_VALUE;
+    }
+
+    /**
+     * retry.
+     * @param single single to be retried.
+     * @param skipExceptions exceptions which should not be retried.
+     * @param <T> The type for single.
+     * @return single of type T.
+     */
+    public <T> Single<T> retry(Single<T> single, List<Class<? extends Throwable>> skipExceptions) {
+        return Single.create(emitter -> call(single, emitter, 0L, maxAttempts, skipExceptions));
+    }
+
+    @SuppressWarnings("ResultOfMethodCallIgnored")
+    private <T> void call(
+            Single<T> single,
+            SingleEmitter<T> emitter,
+            Long delayInSeconds,
+            int attemptsLeft,
+            List<Class<? extends Throwable>> skipExceptions) {
+        single.delaySubscription(delayInSeconds, TimeUnit.SECONDS)
+                .subscribe(emitter::onSuccess, error -> {
+                    if (!emitter.isDisposed()) {
+                        LOG.verbose("Retry attempts left " + attemptsLeft + ". exception type:" + error.getClass());
+                        if (attemptsLeft == 0 || ErrorInspector.contains(error, skipExceptions)) {
+                            emitter.onError(error);
+                        } else {
+                            call(single, emitter, jitteredDelaySec(attemptsLeft),
+                                    attemptsLeft - 1, skipExceptions);
+                        }
+                    } else {
+                        LOG.verbose("The subscribing channel is disposed.");
+                    }
+                });
+    }
+
+
+    /**
+     * Method returns jittered delay time in seconds.
+     * @param attemptsLeft number of attempts left.
+     * @return delay in seconds.
+     */
+    long jitteredDelaySec(int attemptsLeft) {
+        int numAttempt = maxAttempts - (maxAttempts - attemptsLeft);
+        double waitTimeSeconds =
+                Math.min(maxDelayS, Math.pow(2, ((numAttempt) % maxExponent))
+                        + jitterFactor * Math.random());
+        LOG.debug("Wait time is " + waitTimeSeconds + " seconds before retrying");
+        return (long) waitTimeSeconds;
+    }
+}