Fix #1273 AsyncMethodsRateLimiter does not handle ratelimitted errors…

… properly (#1274)

slack-api-client/src/main/java/com/slack/api/methods/impl/AsyncMethodsRateLimiter.java

@@ -7,6 +7,7 @@
 import com.slack.api.rate_limits.WaitTime;
 import com.slack.api.rate_limits.WaitTimeCalculator;
 import com.slack.api.rate_limits.metrics.MetricsDatastore;
+import com.slack.api.rate_limits.metrics.RequestPace;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.Optional;
@@ -19,6 +20,7 @@ public class AsyncMethodsRateLimiter implements RateLimiter {
     private final MetricsDatastore metricsDatastore;
     private final MethodsCustomRateLimitResolver customRateLimitResolver;
     private final WaitTimeCalculator waitTimeCalculator;
+    private final String executorName;
 
     public MetricsDatastore getMetricsDatastore() {
         return metricsDatastore;
@@ -28,10 +30,17 @@ public AsyncMethodsRateLimiter(MethodsConfig config) {
         this.metricsDatastore = config.getMetricsDatastore();
         this.customRateLimitResolver = config.getCustomRateLimitResolver();
         this.waitTimeCalculator = new MethodsWaitTimeCalculator(config);
+        this.executorName = config.getExecutorName();
     }
 
     @Override
     public WaitTime acquireWaitTime(String teamId, String methodName) {
+        Optional<Long> rateLimitedEpochMillis = waitTimeCalculator
+                .getRateLimitedMethodRetryEpochMillis(executorName, teamId, methodName);
+        if (rateLimitedEpochMillis.isPresent()) {
+            long millisToWait = rateLimitedEpochMillis.get() - System.currentTimeMillis();
+            return new WaitTime(millisToWait, RequestPace.RateLimited);
+        }
         return waitTimeCalculator.calculateWaitTime(
                 teamId,
                 methodName,

slack-api-client/src/test/java/test_locally/api/methods/RateLimitedTest.java

@@ -0,0 +1,90 @@
+package test_locally.api.methods;
+
+import com.slack.api.Slack;
+import com.slack.api.SlackConfig;
+import com.slack.api.methods.MethodsClient;
+import com.slack.api.methods.SlackApiException;
+import com.slack.api.rate_limits.metrics.MetricsDatastore;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import util.MockSlackApiServer;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
+import static util.MockSlackApi.RateLimitedToken;
+
+@Slf4j
+public class RateLimitedTest {
+
+    MockSlackApiServer server = new MockSlackApiServer();
+    SlackConfig config = new SlackConfig();
+    Slack slack = Slack.getInstance(config);
+
+    String executorName = RateLimitedTest.class.getCanonicalName();
+
+    @Before
+    public void setup() throws Exception {
+        server.start();
+        config.getMethodsConfig().setExecutorName(executorName);
+        config.synchronizeMetricsDatabases();
+        config.setMethodsEndpointUrlPrefix(server.getMethodsEndpointPrefix());
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        server.stop();
+    }
+
+    @Test
+    public void sync() throws Exception {
+        MethodsClient client = slack.methods(RateLimitedToken);
+        try {
+            client.usersList(r -> r);
+        } catch (SlackApiException e) {
+            assertThat(e.getResponse().code(), is(429));
+            assertThat(e.getResponse().header("Retry-After"), is("3"));
+            MetricsDatastore datastore = config.getMethodsConfig().getMetricsDatastore();
+            log.debug("stats: {}", datastore.getAllStats());
+
+            Integer numOfRequests = datastore.getNumberOfLastMinuteRequests(
+                    executorName,
+                    "T1234567",
+                    "users.list");
+            assertThat(numOfRequests, is(1));
+
+            Long millisToResume = datastore.getRateLimitedMethodRetryEpochMillis(
+                    executorName,
+                    "T1234567",
+                    "users.list");
+            assertThat(millisToResume, is(greaterThan(0L)));
+        }
+    }
+
+    @Test
+    public void async() throws Exception {
+        try {
+            slack.methodsAsync(RateLimitedToken).usersList(r -> r).get(2, TimeUnit.SECONDS);
+        } catch (Exception ee) {
+            MetricsDatastore datastore = config.getMethodsConfig().getMetricsDatastore();
+            log.debug("stats: {}", datastore.getAllStats());
+
+            Integer numOfRequests = datastore.getNumberOfLastMinuteRequests(
+                    executorName,
+                    "T1234567",
+                    "users.list");
+            assertThat(numOfRequests, is(1));
+
+            Long millisToResume = datastore.getRateLimitedMethodRetryEpochMillis(
+                    executorName,
+                    "T1234567",
+                    "users.list");
+            assertThat(millisToResume, is(greaterThan(0L)));
+        }
+    }
+
+}

slack-api-client/src/test/java/util/MockSlackApi.java

@@ -23,6 +23,7 @@ public class MockSlackApi extends HttpServlet {
     public static final String ValidFunctionToken = "xwfp-this-is-valid";
     public static final String ExpiredToken = "xoxb-this-is-expired";
     public static final String InvalidToken = "xoxb-this-is-INVALID";
+    public static final String RateLimitedToken = "xoxb-this-is-ratelimited";
 
     private final FileReader reader = new FileReader("../json-logs/samples/api/");
 
@@ -44,6 +45,28 @@ protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws I
                 resp.getWriter().write("{\"ok\":false,\"error\":\"not_authed\"}");
                 resp.setContentType("application/json");
                 return;
+            } else if (authorizationHeader.startsWith("Bearer " + RateLimitedToken)) {
+                if (methodName.equals("auth.test")) {
+                    String body = "{\n" +
+                            "  \"ok\": true,\n" +
+                            "  \"url\": \"https://java-slack-sdk-test.slack.com/\",\n" +
+                            "  \"team\": \"java-slack-sdk-test\",\n" +
+                            "  \"user\": \"test_user\",\n" +
+                            "  \"team_id\": \"T1234567\",\n" +
+                            "  \"user_id\": \"U1234567\",\n" +
+                            "  \"bot_id\": \"B12345678\",\n" +
+                            "  \"enterprise_id\": \"E12345678\"\n" +
+                            "}";
+                    resp.setStatus(200);
+                    resp.getWriter().write(body);
+                    resp.setContentType("application/json");
+                    return;
+                }
+                resp.setStatus(429);
+                resp.getWriter().write("{\"ok\":false,\"error\":\"ratelimited\"}");
+                resp.setHeader("Retry-After", "3");
+                resp.setContentType("application/json");
+                return;
             } else if (!authorizationHeader.startsWith("Bearer " + ValidToken)
                     && !authorizationHeader.startsWith("Bearer " + ValidFunctionToken)) {
                 resp.setStatus(200);