SK-3026 retry logic update

Author: skyflow-bharti

flowvault/src/main/java/com/skyflow/Skyflow.java

@@ -55,6 +55,8 @@ public static final class SkyflowClientBuilder extends BaseSkyflowClientBuilder<
         private Integer readTimeout;
         private Integer writeTimeout;
         private Integer maxRetries;
+        private Long initialRetryDelayMillis;
+        private Long maxRetryDelayMillis;
 
         @Override
         protected void validateVaultConfig(VaultConfig vaultConfig) throws SkyflowException {
@@ -65,7 +67,8 @@ protected void validateVaultConfig(VaultConfig vaultConfig) throws SkyflowExcept
         protected void onVaultConfigAdded(VaultConfig vaultConfig) throws SkyflowException {
             VaultController controller = new VaultController(vaultConfig, this.skyflowCredentials);
             controller.setCommonHttpConfig(this.timeout, this.connectTimeout, this.readTimeout,
-                    this.writeTimeout, this.maxRetries);
+                    this.writeTimeout, this.maxRetries, this.initialRetryDelayMillis,
+                    this.maxRetryDelayMillis);
             this.vaultClientsMap.put(vaultConfig.getVaultId(), controller);
             LogUtil.printInfoLog(Utils.parameterizedString(InfoLogs.VAULT_CONTROLLER_INITIALIZED.getLog(), vaultConfig.getVaultId()));
         }
@@ -82,7 +85,8 @@ protected void onVaultConfigUpdated(VaultConfig updatedConfig) throws SkyflowExc
                 updated.setVaultConfig(updatedConfig);
             }
             updated.setCommonHttpConfig(this.timeout, this.connectTimeout, this.readTimeout,
-                    this.writeTimeout, this.maxRetries);
+                    this.writeTimeout, this.maxRetries, this.initialRetryDelayMillis,
+                    this.maxRetryDelayMillis);
         }
 
         @Override
@@ -142,6 +146,12 @@ private void carryVaultOverrides(VaultConfig incoming) throws SkyflowException {
             if (incoming.getMaxRetries() != null) {
                 merged.setMaxRetries(incoming.getMaxRetries());
             }
+            if (incoming.getInitialRetryDelayMillis() != null) {
+                merged.setInitialRetryDelayMillis(incoming.getInitialRetryDelayMillis());
+            }
+            if (incoming.getMaxRetryDelayMillis() != null) {
+                merged.setMaxRetryDelayMillis(incoming.getMaxRetryDelayMillis());
+            }
             // The HTTP settings above are resolved lazily on the next request, but the URL is
             // resolved once in the VaultClient constructor — which already ran with the old value.
             if (incoming.getVaultUrl() != null) {
@@ -232,11 +242,38 @@ public SkyflowClientBuilder maxRetries(int maxRetries) {
             return this;
         }
 
+        /**
+         * Backoff before the first retry, in milliseconds. Default 500. Only applies when
+         * {@code maxRetries} is greater than zero.
+         * <p>
+         * <b>Precedence:</b> a vault that sets {@link VaultConfig#setInitialRetryDelayMillis(Long)}
+         * wins; this value applies only to vaults that leave it unset.
+         */
+        public SkyflowClientBuilder initialRetryDelayMillis(long initialRetryDelayMillis) {
+            this.initialRetryDelayMillis = initialRetryDelayMillis;
+            propagateHttpConfig();
+            return this;
+        }
+
+        /**
+         * Ceiling the exponential backoff grows to, in milliseconds. Default 2000. Only applies
+         * when {@code maxRetries} is greater than zero.
+         * <p>
+         * <b>Precedence:</b> a vault that sets {@link VaultConfig#setMaxRetryDelayMillis(Long)}
+         * wins; this value applies only to vaults that leave it unset.
+         */
+        public SkyflowClientBuilder maxRetryDelayMillis(long maxRetryDelayMillis) {
+            this.maxRetryDelayMillis = maxRetryDelayMillis;
+            propagateHttpConfig();
+            return this;
+        }
+
         /** Push the current client-wide HTTP settings onto every vault controller built so far. */
         private void propagateHttpConfig() {
             for (VaultController vault : this.vaultClientsMap.values()) {
                 vault.setCommonHttpConfig(this.timeout, this.connectTimeout, this.readTimeout,
-                        this.writeTimeout, this.maxRetries);
+                        this.writeTimeout, this.maxRetries, this.initialRetryDelayMillis,
+                        this.maxRetryDelayMillis);
             }
         }
 

flowvault/src/main/java/com/skyflow/VaultClient.java

@@ -5,9 +5,9 @@
 import com.skyflow.errors.SkyflowException;
 import com.skyflow.generated.rest.ApiClient;
 import com.skyflow.generated.rest.ApiClientBuilder;
-import com.skyflow.generated.rest.core.RetryInterceptor;
 import com.skyflow.generated.rest.resources.flowservice.FlowserviceClient;
 import com.skyflow.generated.rest.resources.records.RecordsClient;
+import com.skyflow.utils.SkyflowRetryInterceptor;
 import com.skyflow.utils.Utils;
 
 import java.util.concurrent.TimeUnit;
@@ -25,10 +25,14 @@ public class VaultClient extends BaseVaultClient<VaultConfig> {
     private Integer commonReadTimeout;
     private Integer commonWriteTimeout;
     private Integer commonMaxRetries;
+    private Long commonInitialRetryDelayMillis;
+    private Long commonMaxRetryDelayMillis;
     // SDK defaults, used when neither the vault-level nor the client-wide value is set.
     private static final int DEFAULT_TIMEOUT_SECONDS = 60;
     // Retries OFF by default (opt-in) so non-idempotent bulk writes aren't replayed automatically.
     private static final int DEFAULT_MAX_RETRIES = 0;
+    private static final long DEFAULT_INITIAL_RETRY_DELAY_MILLIS = 500L;
+    private static final long DEFAULT_MAX_RETRY_DELAY_MILLIS = 2000L;
 
     protected VaultClient(VaultConfig vaultConfig, Credentials credentials) throws SkyflowException {
         super(vaultConfig, credentials);
@@ -42,12 +46,15 @@ protected VaultClient(VaultConfig vaultConfig, Credentials credentials) throws S
      * client and ApiClient so the next call rebuilds them with the new values.
      */
     protected void setCommonHttpConfig(Integer timeout, Integer connectTimeout, Integer readTimeout,
-                                       Integer writeTimeout, Integer maxRetries) {
+                                       Integer writeTimeout, Integer maxRetries,
+                                       Long initialRetryDelayMillis, Long maxRetryDelayMillis) {
         this.commonTimeout = timeout;
         this.commonConnectTimeout = connectTimeout;
         this.commonReadTimeout = readTimeout;
         this.commonWriteTimeout = writeTimeout;
         this.commonMaxRetries = maxRetries;
+        this.commonInitialRetryDelayMillis = initialRetryDelayMillis;
+        this.commonMaxRetryDelayMillis = maxRetryDelayMillis;
         this.sharedHttpClient = null;
         this.apiClient = null;
     }
@@ -60,6 +67,14 @@ private static int resolveInt(Integer vaultLevel, Integer clientLevel, int defau
         return clientLevel != null ? clientLevel : defaultValue;
     }
 
+    /** Resolve a long setting: vault-level override, else client-wide default, else the SDK default. */
+    private static long resolveLong(Long vaultLevel, Long clientLevel, long defaultValue) {
+        if (vaultLevel != null) {
+            return vaultLevel;
+        }
+        return clientLevel != null ? clientLevel : defaultValue;
+    }
+
     /**
      * Resolve an optional setting: vault-level override, else client-wide default, else null.
      * Null means "not configured" — the caller leaves the underlying HTTP client default in place.
@@ -134,6 +149,10 @@ protected void updateExecutorInHTTP() {
         if (sharedHttpClient == null) {
             int timeoutSeconds = resolveInt(vaultConfig.getTimeout(), commonTimeout, DEFAULT_TIMEOUT_SECONDS);
             int maxRetries = resolveInt(vaultConfig.getMaxRetries(), commonMaxRetries, DEFAULT_MAX_RETRIES);
+            long initialRetryDelayMillis = resolveLong(vaultConfig.getInitialRetryDelayMillis(),
+                    commonInitialRetryDelayMillis, DEFAULT_INITIAL_RETRY_DELAY_MILLIS);
+            long maxRetryDelayMillis = resolveLong(vaultConfig.getMaxRetryDelayMillis(),
+                    commonMaxRetryDelayMillis, DEFAULT_MAX_RETRY_DELAY_MILLIS);
             // Per-attempt timeouts: null => leave OkHttp's built-in default (backward compatible).
             Integer connectTimeout = resolveNullableInt(vaultConfig.getConnectTimeout(), commonConnectTimeout);
             Integer readTimeout = resolveNullableInt(vaultConfig.getReadTimeout(), commonReadTimeout);
@@ -145,7 +164,7 @@ protected void updateExecutorInHTTP() {
                     .callTimeout(timeoutSeconds, TimeUnit.SECONDS)
                     // OUTER: retries. Must wrap the auth interceptor so each attempt re-reads the
                     // (possibly refreshed) bearer token rather than replaying a stale one.
-                    .addInterceptor(new RetryInterceptor(maxRetries))
+                    .addInterceptor(new SkyflowRetryInterceptor(maxRetries, initialRetryDelayMillis, maxRetryDelayMillis))
                     .addInterceptor(chain -> {  // INNER: auth
                         Request requestWithAuth = chain.request().newBuilder()
                                 .header("Authorization", "Bearer " + this.token)

flowvault/src/main/java/com/skyflow/config/VaultConfig.java

@@ -21,6 +21,8 @@ public class VaultConfig extends BaseVaultConfig {
     private Integer readTimeout;     // per-attempt response-read timeout, in seconds
     private Integer writeTimeout;    // per-attempt request-write timeout, in seconds
     private Integer maxRetries;      // retry attempts after the first failure
+    private Long initialRetryDelayMillis; // backoff before the first retry, in milliseconds
+    private Long maxRetryDelayMillis;     // ceiling the exponential backoff grows to, in milliseconds
 
     public VaultConfig() {
         super();
@@ -30,6 +32,8 @@ public VaultConfig() {
         this.readTimeout = null;
         this.writeTimeout = null;
         this.maxRetries = null;
+        this.initialRetryDelayMillis = null;
+        this.maxRetryDelayMillis = null;
     }
 
     public String getVaultUrl() {
@@ -114,4 +118,35 @@ public void setMaxRetries(Integer maxRetries) {
         this.maxRetries = maxRetries;
     }
 
+
+    public Long getInitialRetryDelayMillis() {
+        return initialRetryDelayMillis;
+    }
+
+    /**
+     * Backoff before the first retry, in milliseconds, for this vault.
+     * <p>
+     * Takes precedence over the client-wide {@code Skyflow.builder().initialRetryDelayMillis(...)}.
+     * Leave unset (null) to inherit that value, or the SDK default of 500 ms if it is also unset.
+     * Only applies when {@code maxRetries} is greater than zero.
+     */
+    public void setInitialRetryDelayMillis(Long initialRetryDelayMillis) {
+        this.initialRetryDelayMillis = initialRetryDelayMillis;
+    }
+
+    public Long getMaxRetryDelayMillis() {
+        return maxRetryDelayMillis;
+    }
+
+    /**
+     * Ceiling the exponential backoff grows to, in milliseconds, for this vault.
+     * <p>
+     * Takes precedence over the client-wide {@code Skyflow.builder().maxRetryDelayMillis(...)}.
+     * Leave unset (null) to inherit that value, or the SDK default of 2000 ms if it is also unset.
+     * Only applies when {@code maxRetries} is greater than zero.
+     */
+    public void setMaxRetryDelayMillis(Long maxRetryDelayMillis) {
+        this.maxRetryDelayMillis = maxRetryDelayMillis;
+    }
+
 }

flowvault/src/main/java/com/skyflow/utils/SkyflowRetryInterceptor.java

@@ -0,0 +1,97 @@
+package com.skyflow.utils;
+
+import okhttp3.Interceptor;
+import okhttp3.Response;
+
+import java.io.IOException;
+import java.util.Random;
+
+/**
+ * Retries failed requests with exponential backoff and jitter.
+ * <p>
+ * This exists as hand-written code rather than using the generated
+ * {@code com.skyflow.generated.rest.core.RetryInterceptor} because that one only accepts a retry
+ * count — it has no way to configure the backoff delays that {@code VaultConfig} exposes. It also
+ * keeps its backoff counter on the interceptor instance, so a single shared instance exhausts its
+ * retry budget once for the whole client rather than once per request; this implementation keeps
+ * that state per call.
+ * <p>
+ * Retries the same statuses the generated interceptor does: 408, 429, and any 5xx.
+ */
+public final class SkyflowRetryInterceptor implements Interceptor {
+
+    /** Fraction of the computed delay applied as random jitter, to avoid synchronised retries. */
+    private static final double JITTER_FACTOR = 0.2;
+
+    private final int maxRetries;
+    private final long initialRetryDelayMillis;
+    private final long maxRetryDelayMillis;
+    private final Random random = new Random();
+
+    public SkyflowRetryInterceptor(int maxRetries, long initialRetryDelayMillis, long maxRetryDelayMillis) {
+        if (maxRetries < 0) {
+            throw new IllegalArgumentException("maxRetries must be non-negative");
+        }
+        if (initialRetryDelayMillis < 0) {
+            throw new IllegalArgumentException("initialRetryDelayMillis must be non-negative");
+        }
+        if (maxRetryDelayMillis < 0) {
+            throw new IllegalArgumentException("maxRetryDelayMillis must be non-negative");
+        }
+        this.maxRetries = maxRetries;
+        this.initialRetryDelayMillis = initialRetryDelayMillis;
+        this.maxRetryDelayMillis = maxRetryDelayMillis;
+    }
+
+    @Override
+    public Response intercept(Chain chain) throws IOException {
+        Response response = chain.proceed(chain.request());
+        // Retry budget is scoped to this call, not to the interceptor instance.
+        for (int attempt = 1; attempt <= maxRetries && shouldRetry(response.code()); attempt++) {
+            sleep(backoffMillis(attempt));
+            response.close();
+            response = chain.proceed(chain.request());
+        }
+        return response;
+    }
+
+    /** Exponential growth from the initial delay, capped at the maximum, then jittered. */
+    long backoffMillis(int attempt) {
+        long delay = initialRetryDelayMillis;
+        for (int i = 1; i < attempt && delay < maxRetryDelayMillis; i++) {
+            delay = delay > maxRetryDelayMillis / 2 ? maxRetryDelayMillis : delay * 2;
+        }
+        delay = Math.min(delay, maxRetryDelayMillis);
+        long jitter = (long) (delay * JITTER_FACTOR);
+        if (jitter <= 0) {
+            return delay;
+        }
+        // delay +/- up to JITTER_FACTOR, never negative.
+        return Math.max(0, delay - jitter + random.nextInt((int) Math.min(2 * jitter + 1, Integer.MAX_VALUE)));
+    }
+
+    static boolean shouldRetry(int statusCode) {
+        return statusCode == 408 || statusCode == 429 || statusCode >= 500;
+    }
+
+    private static void sleep(long millis) throws IOException {
+        try {
+            Thread.sleep(millis);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Interrupted while waiting to retry request", e);
+        }
+    }
+
+    public int getMaxRetries() {
+        return maxRetries;
+    }
+
+    public long getInitialRetryDelayMillis() {
+        return initialRetryDelayMillis;
+    }
+
+    public long getMaxRetryDelayMillis() {
+        return maxRetryDelayMillis;
+    }
+}