Response status updates

analytics-core/src/main/java/com/segment/analytics/http/SegmentService.java

@@ -4,11 +4,13 @@
 import okhttp3.HttpUrl;
 import retrofit2.Call;
 import retrofit2.http.Body;
+import retrofit2.http.Header;
 import retrofit2.http.POST;
 import retrofit2.http.Url;
 
 /** REST interface for the Segment API. */
 public interface SegmentService {
   @POST
-  Call<UploadResponse> upload(@Url HttpUrl uploadUrl, @Body Batch batch);
+  Call<UploadResponse> upload(
+      @Header("X-Retry-Count") int retryCount, @Url HttpUrl uploadUrl, @Body Batch batch);
 }

analytics/src/main/java/com/segment/analytics/Analytics.java

@@ -397,7 +397,8 @@ public Analytics build() {
         maximumQueueSizeInBytes = MESSAGE_QUEUE_MAX_BYTE_SIZE;
       }
       if (maximumFlushAttempts == 0) {
-        maximumFlushAttempts = 3;
+        // Adjusted upward to accomodate shorter max retry backoff.
+        maximumFlushAttempts = 1000;
       }
       if (messageTransformers == null) {
         messageTransformers = Collections.emptyList();
@@ -435,7 +436,7 @@ public void log(String message) {
       OkHttpClient.Builder builder =
           client
               .newBuilder()
-              .addInterceptor(new AnalyticsRequestInterceptor(userAgent))
+              .addInterceptor(new AnalyticsRequestInterceptor(writeKey, userAgent))
               .addInterceptor(interceptor);
 
       if (forceTlsV1) {

analytics/src/main/java/com/segment/analytics/AnalyticsRequestInterceptor.java

@@ -2,22 +2,31 @@
 
 import jakarta.annotation.Nonnull;
 import java.io.IOException;
+import okhttp3.Credentials;
 import okhttp3.Interceptor;
 import okhttp3.Request;
 
 class AnalyticsRequestInterceptor implements Interceptor {
+  private static final String AUTHORIZATION_HEADER = "Authorization";
   private static final String USER_AGENT_HEADER = "User-Agent";
 
+  private final @Nonnull String writeKey;
   private final @Nonnull String userAgent;
 
-  AnalyticsRequestInterceptor(@Nonnull String userAgent) {
+  AnalyticsRequestInterceptor(@Nonnull String writeKey, @Nonnull String userAgent) {
+    this.writeKey = writeKey;
     this.userAgent = userAgent;
   }
 
   @Override
   public okhttp3.Response intercept(Chain chain) throws IOException {
     Request request = chain.request();
-    Request newRequest = request.newBuilder().addHeader(USER_AGENT_HEADER, userAgent).build();
+    Request newRequest =
+        request
+            .newBuilder()
+            .addHeader(AUTHORIZATION_HEADER, Credentials.basic(writeKey, ""))
+            .addHeader(USER_AGENT_HEADER, userAgent)
+            .build();
 
     return chain.proceed(newRequest);
   }

analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java

@@ -46,6 +46,8 @@ public class AnalyticsClient {
   private static final String instanceId = UUID.randomUUID().toString();
   private static final int WAIT_FOR_THREAD_COMPLETE_S = 5;
   private static final int TERMINATION_TIMEOUT_S = 1;
+  private static final long MAX_RETRY_AFTER_SECONDS = 300L;
+
 
   static {
     Map<String, String> library = new LinkedHashMap<>();
@@ -409,8 +411,8 @@ public void run() {
   static class BatchUploadTask implements Runnable {
     private static final Backo BACKO =
         Backo.builder() //
-            .base(TimeUnit.SECONDS, 15) //
-            .cap(TimeUnit.HOURS, 1) //
+            .base(TimeUnit.MILLISECONDS, 100) //
+            .cap(TimeUnit.MINUTES, 1) //
             .jitter(1) //
             .build();
 
@@ -438,12 +440,37 @@ private void notifyCallbacksWithException(Batch batch, Exception exception) {
       }
     }
 
-    /** Returns {@code true} to indicate a batch should be retried. {@code false} otherwise. */
-    boolean upload() {
+    private enum RetryStrategy {
+      NONE,
+      BACKOFF,
+      RETRY_AFTER
+    }
+
+    private static final class UploadResult {
+      final RetryStrategy strategy;
+      final long retryAfterSeconds;
+
+      UploadResult(RetryStrategy strategy) {
+        this(strategy, 0L);
+      }
+
+      UploadResult(RetryStrategy strategy, long retryAfterSeconds) {
+        this.strategy = strategy;
+        this.retryAfterSeconds = retryAfterSeconds;
+      }
+    }
+
+    /**
+     * Perform a single upload attempt.
+     *
+     * @param attempt overall number of attempts so far (1-based)
+     */
+    UploadResult upload(int attempt) {
       client.log.print(VERBOSE, "Uploading batch %s.", batch.sequence());
 
       try {
-        Call<UploadResponse> call = client.service.upload(client.uploadUrl, batch);
+        Call<UploadResponse> call;
+        call = client.service.upload(attempt - 1, client.uploadUrl, batch);
         Response<UploadResponse> response = call.execute();
 
         if (response.isSuccessful()) {
@@ -455,59 +482,148 @@ boolean upload() {
             }
           }
 
-          return false;
+          return new UploadResult(RetryStrategy.NONE);
         }
 
         int status = response.code();
-        if (is5xx(status)) {
+
+        if (isStatusRetryAfterEligible(status)) {
+          String retryAfterHeader = response.headers().get("Retry-After");
+          Long retryAfterSeconds = parseRetryAfterSeconds(retryAfterHeader);
+          if (retryAfterSeconds != null) {
+            client.log.print(
+                DEBUG,
+                "Could not upload batch %s due to status %s with Retry-After %s seconds. Retrying after delay.",
+                batch.sequence(),
+                status,
+                retryAfterSeconds);
+            return new UploadResult(RetryStrategy.RETRY_AFTER, retryAfterSeconds);
+          }
           client.log.print(
-              DEBUG, "Could not upload batch %s due to server error. Retrying.", batch.sequence());
-          return true;
-        } else if (status == 429) {
+              DEBUG,
+              "Status %s did not have a valid Retry-After header.",
+              batch.sequence(),
+              status);
+        }
+
+        if (isStatusRetryWithBackoff(status)) {
           client.log.print(
-              DEBUG, "Could not upload batch %s due to rate limiting. Retrying.", batch.sequence());
-          return true;
+              DEBUG,
+              "Could not upload batch %s due to retryable status %s. Retrying with backoff.",
+              batch.sequence(),
+              status);
+          return new UploadResult(RetryStrategy.BACKOFF);
         }
 
-        client.log.print(DEBUG, "Could not upload batch %s. Giving up.", batch.sequence());
+        client.log.print(
+            DEBUG,
+            "Could not upload batch %s due to non-retryable status %s. Giving up.",
+            batch.sequence(),
+            status);
         notifyCallbacksWithException(batch, new IOException(response.errorBody().string()));
 
-        return false;
+        return new UploadResult(RetryStrategy.NONE);
       } catch (IOException error) {
         client.log.print(DEBUG, error, "Could not upload batch %s. Retrying.", batch.sequence());
 
-        return true;
+        return new UploadResult(RetryStrategy.BACKOFF);
       } catch (Exception exception) {
         client.log.print(DEBUG, "Could not upload batch %s. Giving up.", batch.sequence());
 
         notifyCallbacksWithException(batch, exception);
 
-        return false;
+        return new UploadResult(RetryStrategy.NONE);
+      }
+    }
+
+    private static boolean isStatusRetryAfterEligible(int status) {
+      return status == 429 || status == 408 || status == 503;
+    }
+
+    private static Long parseRetryAfterSeconds(String headerValue) {
+      if (headerValue == null) {
+        return null;
+      }
+      headerValue = headerValue.trim();
+      if (headerValue.isEmpty()) {
+        return null;
+      }
+      try {
+        long seconds = Long.parseLong(headerValue);
+        if (seconds <= 0L) {
+          return null;
+        }
+        if (seconds > MAX_RETRY_AFTER_SECONDS) {
+          return MAX_RETRY_AFTER_SECONDS;
+        }
+        return seconds;
+      } catch (NumberFormatException ignored) {
+        return null;
       }
     }
 
     @Override
     public void run() {
-      int attempt = 0;
-      for (; attempt <= maxRetries; attempt++) {
-        boolean retry = upload();
-        if (!retry) return;
+      int totalAttempts = 0; // counts every HTTP attempt (for header and error message)
+      int backoffAttempts = 0; // counts attempts that consume backoff-based retries
+      int maxBackoffAttempts = maxRetries + 1; // preserve existing semantics
+
+      while (true) {
+        totalAttempts++;
+        UploadResult result = upload(totalAttempts);
+
+        if (result.strategy == RetryStrategy.NONE) {
+          return;
+        }
+
+        if (result.strategy == RetryStrategy.RETRY_AFTER) {
+          try {
+            TimeUnit.SECONDS.sleep(result.retryAfterSeconds);
+          } catch (InterruptedException e) {
+            client.log.print(
+                DEBUG,
+                "Thread interrupted while waiting for Retry-After for batch %s.",
+                batch.sequence());
+            Thread.currentThread().interrupt();
+            return;
+          }
+          // Do not count Retry-After based retries against maxRetries.
+          continue;
+        }
+
+        // BACKOFF strategy
+        backoffAttempts++;
+        if (backoffAttempts >= maxBackoffAttempts) {
+          break;
+        }
+
         try {
-          backo.sleep(attempt);
+          backo.sleep(backoffAttempts - 1);
         } catch (InterruptedException e) {
           client.log.print(
               DEBUG, "Thread interrupted while backing off for batch %s.", batch.sequence());
+          Thread.currentThread().interrupt();
           return;
         }
       }
 
       client.log.print(ERROR, "Could not upload batch %s. Retries exhausted.", batch.sequence());
       notifyCallbacksWithException(
-          batch, new IOException(Integer.toString(attempt) + " retries exhausted"));
+          batch, new IOException(Integer.toString(totalAttempts) + " retries exhausted"));
     }
 
-    private static boolean is5xx(int status) {
-      return status >= 500 && status < 600;
+    private static boolean isStatusRetryWithBackoff(int status) {
+      // Explicitly retry these client errors
+      if (status == 408 || status == 410 || status == 429 || status == 460) {
+        return true;
+      }
+
+      // Retry all other 5xx errors except 501, 505, and 511
+      if (status >= 500 && status < 600) {
+        return status != 501 && status != 505 && status != 511;
+      }
+
+      return false;
     }
   }
 

analytics/src/test/java/com/segment/analytics/AnalyticsRequestInterceptorTest.java

@@ -16,14 +16,16 @@ public class AnalyticsRequestInterceptorTest {
 
   @Test
   public void testInterceptor() throws IOException {
-    AnalyticsRequestInterceptor interceptor = new AnalyticsRequestInterceptor("userAgent");
+    AnalyticsRequestInterceptor interceptor =
+        new AnalyticsRequestInterceptor("writeKey", "userAgent");
 
     final Request request = new Request.Builder().url("https://api.segment.io").get().build();
 
     Chain chain =
         new ChainAdapter(request, mockConnection) {
           @Override
           public Response proceed(Request request) throws IOException {
+            assertThat(request.header("Authorization"), Is.is("Basic d3JpdGVLZXk6"));
             assertThat(request.header("User-Agent"), Is.is("userAgent"));
             return null;
           }