Don't allow rate limit responses to end all group sends.

app/src/main/java/org/thoughtcrime/securesms/jobs/PushGroupSendJob.java

@@ -405,8 +405,13 @@ public final class PushGroupSendJob extends PushSendJob {
 
       RetrieveProfileJob.enqueue(mismatchRecipientIds);
     } else if (!networkFailures.isEmpty()) {
-      Log.w(TAG, "Retrying because there were " + networkFailures.size() + " network failures.");
+      long retryAfter = results.stream()
-      throw new RetryLaterException();
+                               .filter(r -> r.getRateLimitFailure() != null)
+                               .map(r -> r.getRateLimitFailure().getRetryAfterMilliseconds().or(-1L))
+                               .max(Long::compare)
+                               .orElse(-1L);
+      Log.w(TAG, "Retrying because there were " + networkFailures.size() + " network failures. retryAfter: " + retryAfter);
+      throw new RetryLaterException(retryAfter);
     }
   }
 

libsignal/service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java

@@ -75,6 +75,7 @@ import org.whispersystems.signalservice.api.push.exceptions.NonSuccessfulRespons
 import org.whispersystems.signalservice.api.push.exceptions.NotFoundException;
 import org.whispersystems.signalservice.api.push.exceptions.ProofRequiredException;
 import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException;
+import org.whispersystems.signalservice.api.push.exceptions.RateLimitException;
 import org.whispersystems.signalservice.api.push.exceptions.ServerRejectedException;
 import org.whispersystems.signalservice.api.push.exceptions.UnregisteredUserException;
 import org.whispersystems.signalservice.api.services.AttachmentService;
@@ -1623,6 +1624,9 @@ public class SignalServiceMessageSender {
         } else if (e.getCause() instanceof ProofRequiredException) {
           Log.w(TAG, e);
           results.add(SendMessageResult.proofRequiredFailure(recipient, (ProofRequiredException) e.getCause()));
+        } else if (e.getCause() instanceof RateLimitException) {
+          Log.w(TAG, e);
+          results.add(SendMessageResult.rateLimitFailure(recipient, (RateLimitException) e.getCause()));
         } else {
           throw new IOException(e);
         }

libsignal/service/src/main/java/org/whispersystems/signalservice/api/messages/SendMessageResult.java

@@ -5,7 +5,7 @@ import org.whispersystems.libsignal.IdentityKey;
 import org.whispersystems.libsignal.util.guava.Optional;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
 import org.whispersystems.signalservice.api.push.exceptions.ProofRequiredException;
-import org.whispersystems.signalservice.internal.push.SignalServiceProtos;
+import org.whispersystems.signalservice.api.push.exceptions.RateLimitException;
 import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Content;
 
 import java.util.List;
@@ -18,25 +18,30 @@ public class SendMessageResult {
   private final boolean                unregisteredFailure;
   private final IdentityFailure        identityFailure;
   private final ProofRequiredException proofRequiredFailure;
+  private final RateLimitException     rateLimitFailure;
 
   public static SendMessageResult success(SignalServiceAddress address, List<Integer> devices, boolean unidentified, boolean needsSync, long duration, Optional<Content> content) {
-    return new SendMessageResult(address, new Success(unidentified, needsSync, duration, content, devices), false, false, null, null);
+    return new SendMessageResult(address, new Success(unidentified, needsSync, duration, content, devices), false, false, null, null, null);
   }
 
   public static SendMessageResult networkFailure(SignalServiceAddress address) {
-    return new SendMessageResult(address, null, true, false, null, null);
+    return new SendMessageResult(address, null, true, false, null, null, null);
   }
 
   public static SendMessageResult unregisteredFailure(SignalServiceAddress address) {
-    return new SendMessageResult(address, null, false, true, null, null);
+    return new SendMessageResult(address, null, false, true, null, null, null);
   }
 
   public static SendMessageResult identityFailure(SignalServiceAddress address, IdentityKey identityKey) {
-    return new SendMessageResult(address, null, false, false, new IdentityFailure(identityKey), null);
+    return new SendMessageResult(address, null, false, false, new IdentityFailure(identityKey), null, null);
   }
 
   public static SendMessageResult proofRequiredFailure(SignalServiceAddress address, ProofRequiredException proofRequiredException) {
-    return new SendMessageResult(address, null, false, false, null, proofRequiredException);
+    return new SendMessageResult(address, null, false, false, null, proofRequiredException, null);
+  }
+
+  public static SendMessageResult rateLimitFailure(SignalServiceAddress address, RateLimitException rateLimitException) {
+    return new SendMessageResult(address, null, false, false, null, null, rateLimitException);
   }
 
   public SignalServiceAddress getAddress() {
@@ -52,7 +57,7 @@ public class SendMessageResult {
   }
 
   public boolean isNetworkFailure() {
-    return networkFailure || proofRequiredFailure != null;
+    return networkFailure || proofRequiredFailure != null || rateLimitFailure != null;
   }
 
   public boolean isUnregisteredFailure() {
@@ -67,19 +72,25 @@ public class SendMessageResult {
     return proofRequiredFailure;
   }
 
+  public RateLimitException getRateLimitFailure() {
+    return rateLimitFailure;
+  }
+
   private SendMessageResult(SignalServiceAddress address,
                             Success success,
                             boolean networkFailure,
                             boolean unregisteredFailure,
                             IdentityFailure identityFailure,
-                            ProofRequiredException proofRequiredFailure)
+                            ProofRequiredException proofRequiredFailure,
+                            RateLimitException rateLimitFailure)
   {
-    this.address             = address;
+    this.address              = address;
-    this.success             = success;
+    this.success              = success;
-    this.networkFailure      = networkFailure;
+    this.networkFailure       = networkFailure;
-    this.unregisteredFailure = unregisteredFailure;
+    this.unregisteredFailure  = unregisteredFailure;
-    this.identityFailure     = identityFailure;
+    this.identityFailure      = identityFailure;
     this.proofRequiredFailure = proofRequiredFailure;
+    this.rateLimitFailure     = rateLimitFailure;
   }
 
   public static class Success {

libsignal/service/src/main/java/org/whispersystems/signalservice/api/push/exceptions/RateLimitException.java

@@ -6,8 +6,21 @@
 
 package org.whispersystems.signalservice.api.push.exceptions;
 
+import org.whispersystems.libsignal.util.guava.Optional;
+
 public class RateLimitException extends NonSuccessfulResponseCodeException {
-  public RateLimitException(String s) {
+  private final Optional<Long> retryAfterMilliseconds;
-    super(413, s);
+
+  public RateLimitException(int status, String message) {
+    this(status, message, Optional.absent());
+  }
+
+  public RateLimitException(int status, String message, Optional<Long> retryAfterMilliseconds) {
+    super(status, message);
+    this.retryAfterMilliseconds = retryAfterMilliseconds;
+  }
+
+  public Optional<Long> getRetryAfterMilliseconds() {
+    return retryAfterMilliseconds;
   }
 }

libsignal/service/src/main/java/org/whispersystems/signalservice/internal/push/PushServiceSocket.java

@@ -124,6 +124,7 @@ import org.whispersystems.signalservice.internal.util.Util;
 import org.whispersystems.signalservice.internal.util.concurrent.FutureTransformers;
 import org.whispersystems.signalservice.internal.util.concurrent.ListenableFuture;
 import org.whispersystems.signalservice.internal.util.concurrent.SettableFuture;
+import org.whispersystems.signalservice.internal.websocket.DefaultErrorMapper;
 import org.whispersystems.util.Base64;
 import org.whispersystems.util.Base64UrlSafe;
 
@@ -1678,8 +1679,11 @@ public class PushServiceSocket {
 
     switch (responseCode) {
       case 413:
-      case 429:
+      case 429: {
-        throw new RateLimitException("Rate limit exceeded: " + responseCode);
+        long           retryAfterLong = Util.parseLong(response.header("Retry-After"), -1);
+        Optional<Long> retryAfter     = retryAfterLong != -1 ? Optional.of(TimeUnit.SECONDS.toMillis(retryAfterLong)) : Optional.absent();
+        throw new RateLimitException(responseCode, "Rate limit exceeded: " + responseCode, retryAfter);
+      }
       case 401:
       case 403:
         throw new AuthorizationFailedException(responseCode, "Authorization failed!");
@@ -1884,7 +1888,7 @@ public class PushServiceSocket {
       case 409:
         throw new RemoteAttestationResponseExpiredException("Remote attestation response expired");
       case 429:
-        throw new RateLimitException("Rate limit exceeded: " + response.code());
+        throw new RateLimitException(response.code(), "Rate limit exceeded: " + response.code());
     }
 
     throw new NonSuccessfulResponseCodeException(response.code(), "Response: " + response);
@@ -1981,7 +1985,7 @@ public class PushServiceSocket {
             throw new ConflictException();
           }
         case 429:
-          throw new RateLimitException("Rate limit exceeded: " + response.code());
+          throw new RateLimitException(response.code(), "Rate limit exceeded: " + response.code());
         case 499:
           throw new DeprecatedVersionException();
       }

libsignal/service/src/main/java/org/whispersystems/signalservice/internal/util/Util.java

@@ -157,4 +157,12 @@ public class Util {
       return defaultValue;
     }
   }
+
+  public static long parseLong(String longString, long defaultValue) {
+    try {
+      return Long.parseLong(longString);
+    } catch (NumberFormatException e) {
+      return defaultValue;
+    }
+  }
 }

libsignal/service/src/main/java/org/whispersystems/signalservice/internal/websocket/DefaultErrorMapper.java

@@ -1,6 +1,7 @@
 package org.whispersystems.signalservice.internal.websocket;
 
 import org.whispersystems.libsignal.util.guava.Function;
+import org.whispersystems.libsignal.util.guava.Optional;
 import org.whispersystems.signalservice.api.push.exceptions.AuthorizationFailedException;
 import org.whispersystems.signalservice.api.push.exceptions.CaptchaRequiredException;
 import org.whispersystems.signalservice.api.push.exceptions.DeprecatedVersionException;
@@ -27,6 +28,7 @@ import org.whispersystems.signalservice.internal.util.Util;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * A default implementation of a {@link ErrorMapper} that can parse most known application
@@ -100,8 +102,11 @@ public final class DefaultErrorMapper implements ErrorMapper {
           return e;
         }
       case 413:
-      case 429:
+      case 429: {
-        return new RateLimitException("Rate limit exceeded: " + status);
+        long           retryAfterLong = Util.parseLong(getHeader.apply("Retry-After"), -1);
+        Optional<Long> retryAfter     = retryAfterLong != -1 ? Optional.of(TimeUnit.SECONDS.toMillis(retryAfterLong)) : Optional.absent();
+        return new RateLimitException(status, "Rate limit exceeded: " + status, retryAfter);
+      }
       case 417:
         return new ExpectationFailedException();
       case 423: