fix(speedrun): pause queue on 429/420 instead of dropping it

speedrun.com returns 429 or 420 when we ask too fast. These were
counted as errors: 3 in a row dropped both queues. Now the thread
pauses and retries the same game. Rate: 1 request per 2 s. Refs #48

readme.md

@@ -72,7 +72,9 @@ To Be Played is split into three repositories:
   least one verified run. An edition that finds nothing tries its IGDB parent game with the parent's year, else its
   name without an edition suffix. Several same-name games and no year to decide: no link. The game detail returns
   the speedrun.com link as `speedrunUrl`. speedrun.com has no Steam or IGDB link and no key, so one queue works for
-  all users, with up to 3 requests per game and 1 request per second; a result is kept 30 days. Migration V47 asks
+  all users, with up to 3 requests per game and 1 request every 2 seconds; a result is kept 30 days. A rate limit answer
+  (429 or 420) pauses the queue (its Retry-After, else 1 min doubling up to 30 min) and asks the same game again; the
+  queue is never dropped for it. Migration V47 asks
   every stored match again once at boot, behind user requests; links set by an auditor are never touched.
 - **Achievements**: after every library sync a background job fetches the achievement schema of each owned game
   and the user's unlocks of each game whose playtime changed. One Steam call per second, so user requests keep half

src/main/java/core_service/modules/library/service/SpeedrunService.java

@@ -10,6 +10,7 @@ import core_service.modules.library.repo.GameSpeedrunRepository;
 import core_service.modules.library.repo.SteamGameRepository;
 import core_service.modules.library.repo.UserGameRepository;
 import core_service.modules.speedrun_adapter.service.SpeedrunApiService;
+import core_service.modules.speedrun_adapter.service.SpeedrunThrottledException;
 import core_service.shared.background.BackgroundPools;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -42,6 +43,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
  * year of its IGDB parent game, a game without a parent by its name without an edition suffix. A sure hit without
  * verified runs ends the lookup. See {@link SpeedrunMatch} for "sure".
  * <p>
+ * A rate limit answer (429, 420) pauses the thread (Retry-After, else 1 min doubling up to 30 min) and asks the same
+ * game again; it never drops the queues.
+ * <p>
  * {@link #recheckMarked} asks the rows a migration marked (checked_at = epoch) again. Those games wait behind every
  * game a user asked for.
  */
@@ -55,6 +59,8 @@ public class SpeedrunService {
     static final int MAX_ERRORS_IN_A_ROW = 3;
     // A small batch, so a user request waits for a few games and not for 50.
     static final int RECHECK_BATCH = 10;
+    static final Duration FIRST_PAUSE = Duration.ofMinutes(1);
+    static final Duration MAX_PAUSE = Duration.ofMinutes(30);
 
     private final SpeedrunApiService speedrunApiService;
     private final IgdbApiService igdbApiService;
@@ -75,7 +81,7 @@ public class SpeedrunService {
 
     /**
      * Queues the games without a fresh row and returns; the speedrun thread works the queue off. A game takes up to 3
-     * requests, and there is 1 request per second.
+     * requests, and there is 1 request every 2 seconds.
      */
     public void refresh(Collection<Long> appids) {
         if (!speedrunApiService.isEnabled() || appids.isEmpty()) return;
@@ -107,7 +113,7 @@ public class SpeedrunService {
                 running.set(false);
             }
             // Games can be queued after the last poll and before the flag was cleared.
-            if (pending.isEmpty() && recheck.isEmpty()) return;
+            if (Thread.currentThread().isInterrupted() || (pending.isEmpty() && recheck.isEmpty())) return;
         }
     }
 
@@ -136,6 +142,7 @@ public class SpeedrunService {
 
     private void drain() {
         int errorsInARow = 0;
+        int throttlesInARow = 0;
         while (!pending.isEmpty() || !recheck.isEmpty()) {
             boolean forUsers = !pending.isEmpty();
             List<Long> batch = forUsers ? poll(pending, BATCH) : poll(recheck, RECHECK_BATCH);
@@ -165,12 +172,25 @@ public class SpeedrunService {
                 boolean stored;
                 try {
                     stored = checkAndStore(appid, name, years.get().getOrDefault(appid, Set.of()), parents.get().get(appid));
+                } catch (SpeedrunThrottledException e) {
+                    Duration pause = pauseFor(e.getRetryAfter(), ++throttlesInARow);
+                    log.warn("speedrun.com rate limit, pausing {} ({} games queued)", pause,
+                            pending.size() + recheck.size() + batch.size() - i);
+                    try {
+                        pause(pause);
+                    } catch (InterruptedException stop) {
+                        Thread.currentThread().interrupt();
+                        return;
+                    }
+                    i--; // same game again, after the user check at the loop head
+                    continue;
                 } catch (Exception e) {
                     log.warn("Speedrun game of appid={} not stored: {}", appid, e.getMessage());
                     stored = false;
                 }
                 if (stored) {
                     errorsInARow = 0;
+                    throttlesInARow = 0;
                 } else if (++errorsInARow >= MAX_ERRORS_IN_A_ROW) {
                     log.warn("Speedrun check failed {} times in a row, {} queued games dropped", errorsInARow, pending.size());
                     pending.clear();
@@ -181,6 +201,22 @@ public class SpeedrunService {
         }
     }
 
+    /**
+     * @param retryAfter the pause speedrun.com asked for; null when it did not say
+     * @param throttlesInARow throttled answers since the last stored game, at least 1
+     */
+    static Duration pauseFor(Duration retryAfter, int throttlesInARow) {
+        if (retryAfter != null) {
+            return Duration.ofSeconds(Math.clamp(retryAfter.toSeconds(), 1, MAX_PAUSE.toSeconds()));
+        }
+        Duration backoff = FIRST_PAUSE.multipliedBy(1L << Math.min(throttlesInARow - 1, 5));
+        return backoff.compareTo(MAX_PAUSE) > 0 ? MAX_PAUSE : backoff;
+    }
+
+    void pause(Duration pause) throws InterruptedException {
+        Thread.sleep(pause);
+    }
+
     private static List<Long> poll(ConcurrentSkipListSet<Long> queue, int size) {
         List<Long> batch = new ArrayList<>();
         Long next;

src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunApiService.java

@@ -7,9 +7,12 @@ import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.HttpHeaders;
 import org.springframework.stereotype.Service;
+import org.springframework.web.client.HttpStatusCodeException;
 import org.springframework.web.client.RestClient;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 
@@ -45,6 +48,8 @@ public class SpeedrunApiService {
     /**
      * Games whose name looks like the given one, in speedrun.com's order, without mods and category extensions.
      * Empty when the adapter is off or the request failed.
+     *
+     * @throws SpeedrunThrottledException on a rate limit answer (429, 420)
      */
     public Optional<List<Game>> findGames(String name) {
         if (!enabled || name == null || name.isBlank()) {
@@ -67,6 +72,7 @@ public class SpeedrunApiService {
                     .map(game -> new Game(game.id(), game.names().international().trim(), game.released(), game.weblink()))
                     .toList());
         } catch (Exception e) {
+            throwIfThrottle(e);
             log.warn("speedrun.com search for \"{}\" failed: {}", name, e.getMessage());
             return Optional.empty();
         }
@@ -74,6 +80,8 @@ public class SpeedrunApiService {
 
     /**
      * Empty when the adapter is off or the request failed.
+     *
+     * @throws SpeedrunThrottledException on a rate limit answer (429, 420)
      */
     public Optional<Boolean> hasVerifiedRuns(String gameId) {
         if (!enabled) {
@@ -90,8 +98,26 @@ public class SpeedrunApiService {
             }
             return Optional.of(!runs.data().isEmpty());
         } catch (Exception e) {
+            throwIfThrottle(e);
             log.warn("speedrun.com runs of game {} failed: {}", gameId, e.getMessage());
             return Optional.empty();
         }
     }
+
+    private static void throwIfThrottle(Exception e) {
+        if (!(e instanceof HttpStatusCodeException http)) return;
+        int status = http.getStatusCode().value();
+        if (status != 429 && status != 420) return;
+        Duration retryAfter = null;
+        HttpHeaders headers = http.getResponseHeaders();
+        String header = headers == null ? null : headers.getFirst(HttpHeaders.RETRY_AFTER);
+        if (header != null) {
+            try {
+                retryAfter = Duration.ofSeconds(Long.parseLong(header.trim()));
+            } catch (NumberFormatException ignored) {
+                // an HTTP-date: the caller backs off on its own
+            }
+        }
+        throw new SpeedrunThrottledException(retryAfter);
+    }
 }

src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunThrottledException.java

@@ -0,0 +1,22 @@
+package core_service.modules.speedrun_adapter.service;
+
+import lombok.Getter;
+
+import java.time.Duration;
+
+/**
+ * speedrun.com asked us to slow down: 429 (its Cloudflare) or 420 (the documented API limit).
+ */
+@Getter
+public class SpeedrunThrottledException extends RuntimeException {
+
+    /**
+     * From the Retry-After header; null when it is missing or not a number of seconds.
+     */
+    private final Duration retryAfter;
+
+    public SpeedrunThrottledException(Duration retryAfter) {
+        super("speedrun.com rate limit");
+        this.retryAfter = retryAfter;
+    }
+}

src/main/resources/application.yml

@@ -135,9 +135,9 @@ resilience4j:
         limit-refresh-period: 1s
         timeout-duration: 5s
       speedrunApi:
-        # speedrun.com allows 100 requests/min per IP.
+        # speedrun.com documents 100 requests/min per IP, but its Cloudflare answered 429 (1015) at 60/min: 30/min.
         limit-for-period: 1
-        limit-refresh-period: 1s
+        limit-refresh-period: 2s
         timeout-duration: 30s
       igdbApi:
         # IGDB allows 4 requests/s. Taken per HTTP request.

src/test/java/core_service/modules/library/service/SpeedrunServiceTest.java

@@ -8,9 +8,11 @@ import core_service.modules.library.repo.GameSpeedrunRepository;
 import core_service.modules.library.repo.SteamGameRepository;
 import core_service.modules.library.repo.UserGameRepository;
 import core_service.modules.speedrun_adapter.service.SpeedrunApiService;
+import core_service.modules.speedrun_adapter.service.SpeedrunThrottledException;
 import core_service.shared.test.AbstractUnitTest;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.Spy;
@@ -18,14 +20,17 @@ import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.dao.DataAccessResourceFailureException;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.LongStream;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -59,14 +64,26 @@ class SpeedrunServiceTest extends AbstractUnitTest {
     @Spy
     private TaskExecutor speedrunExecutor = new SyncTaskExecutor();
     private SpeedrunService service;
+    private final List<Duration> pauses = new ArrayList<>();
+    private PauseHook onPause = () -> {};
+
+    private interface PauseHook {
+        void run() throws InterruptedException;
+    }
 
     private static final String LINK = "https://www.speedrun.com/prey2017";
 
     @BeforeEach
     void setUp() {
         service = new SpeedrunService(speedrunApiService, igdbApiService, gameSpeedrunRepository, gameEditionRepository,
-                steamGameRepository, userGameRepository, syncFollowUpExecutor, speedrunExecutor);
-        when(speedrunApiService.isEnabled()).thenReturn(true);
+                steamGameRepository, userGameRepository, syncFollowUpExecutor, speedrunExecutor) {
+            @Override
+            void pause(Duration pause) throws InterruptedException {
+                pauses.add(pause);
+                onPause.run();
+            }
+        };
+        lenient().when(speedrunApiService.isEnabled()).thenReturn(true);
     }
 
     private void names(long... appids) {
@@ -282,6 +299,105 @@ class SpeedrunServiceTest extends AbstractUnitTest {
         verify(speedrunApiService, times(SpeedrunService.MAX_ERRORS_IN_A_ROW)).findGames("Prey");
     }
 
+    private static final Optional<List<SpeedrunApiService.Game>> NO_HITS = Optional.of(List.of());
+
+    @Test
+    void refresh_Throttled_PausesAndRetriesTheSameGame() {
+        names(620);
+        when(speedrunApiService.findGames("Prey")).thenThrow(new SpeedrunThrottledException(null)).thenReturn(NO_HITS);
+
+        service.refresh(List.of(620L));
+
+        assertThat(pauses).containsExactly(Duration.ofMinutes(1));
+        verify(speedrunApiService, times(2)).findGames("Prey");
+        verify(gameSpeedrunRepository).upsert(eq(620L), isNull(), isNull(), any());
+    }
+
+    @Test
+    void refresh_Throttled_KeepsTheQueue() {
+        names(1, 2, 3, 4, 5);
+        when(speedrunApiService.findGames("Prey")).thenThrow(new SpeedrunThrottledException(null),
+                new SpeedrunThrottledException(null), new SpeedrunThrottledException(null),
+                new SpeedrunThrottledException(null)).thenReturn(NO_HITS);
+
+        service.refresh(List.of(1L, 2L, 3L, 4L, 5L));
+
+        verify(gameSpeedrunRepository, times(5)).upsert(any(), isNull(), isNull(), any());
+    }
+
+    @Test
+    void refresh_ThrottledWithRetryAfter_PausesThatLong() {
+        names(620);
+        when(speedrunApiService.findGames("Prey")).thenThrow(new SpeedrunThrottledException(Duration.ofSeconds(90)))
+                .thenReturn(NO_HITS);
+
+        service.refresh(List.of(620L));
+
+        assertThat(pauses).containsExactly(Duration.ofSeconds(90));
+    }
+
+    @Test
+    void refresh_ThrottledAgain_PauseGrowsUpToTheCap() {
+        names(620);
+        SpeedrunThrottledException throttled = new SpeedrunThrottledException(null);
+        when(speedrunApiService.findGames("Prey")).thenThrow(throttled, throttled, throttled, throttled, throttled,
+                throttled, throttled).thenReturn(NO_HITS);
+
+        service.refresh(List.of(620L));
+
+        assertThat(pauses).containsExactly(Duration.ofMinutes(1), Duration.ofMinutes(2), Duration.ofMinutes(4),
+                Duration.ofMinutes(8), Duration.ofMinutes(16), Duration.ofMinutes(30), Duration.ofMinutes(30));
+    }
+
+    @Test
+    void refresh_StoredGame_ResetsTheBackoff() {
+        names(1, 2);
+        SpeedrunThrottledException throttled = new SpeedrunThrottledException(null);
+        when(speedrunApiService.findGames("Prey")).thenThrow(throttled).thenReturn(NO_HITS)
+                .thenThrow(throttled).thenReturn(NO_HITS);
+
+        service.refresh(List.of(1L, 2L));
+
+        assertThat(pauses).containsExactly(Duration.ofMinutes(1), Duration.ofMinutes(1));
+    }
+
+    @Test
+    void refresh_ThrottledReCheck_UserRequestQueuedDuringThePause_IsServedFirst() {
+        names(1, 2, 999);
+        when(gameSpeedrunRepository.findAppidsToRecheck(Instant.EPOCH)).thenReturn(List.of(1L, 2L));
+        when(speedrunApiService.findGames("Prey")).thenThrow(new SpeedrunThrottledException(null)).thenReturn(NO_HITS);
+        onPause = () -> service.refresh(List.of(999L));
+
+        service.recheckMarked();
+
+        InOrder order = inOrder(gameSpeedrunRepository);
+        order.verify(gameSpeedrunRepository).upsert(eq(999L), isNull(), isNull(), any());
+        order.verify(gameSpeedrunRepository).upsert(eq(1L), isNull(), isNull(), any());
+        order.verify(gameSpeedrunRepository).upsert(eq(2L), isNull(), isNull(), any());
+    }
+
+    @Test
+    @Timeout(value = 5, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void refresh_InterruptedDuringPause_StopsAndKeepsInterruptFlag() {
+        names(1, 2, 999);
+        when(speedrunApiService.findGames("Prey")).thenThrow(new SpeedrunThrottledException(null));
+        onPause = () -> {
+            service.refresh(List.of(999L)); // a user asks during the pause
+            throw new InterruptedException();
+        };
+
+        service.refresh(List.of(1L, 2L));
+
+        assertThat(Thread.interrupted()).isTrue();
+        verify(speedrunApiService, times(1)).findGames("Prey");
+    }
+
+    @Test
+    void pauseFor_RetryAfterIsClampedToTheCap() {
+        assertThat(SpeedrunService.pauseFor(Duration.ofHours(2), 1)).isEqualTo(SpeedrunService.MAX_PAUSE);
+        assertThat(SpeedrunService.pauseFor(Duration.ZERO, 1)).isEqualTo(Duration.ofSeconds(1));
+    }
+
     @Test
     void refresh_CalledDuringARun_QueuesTheGameForTheRunningThread() {
         names(620, 220);

src/test/java/core_service/modules/speedrun_adapter/service/SpeedrunApiServiceTest.java

@@ -4,18 +4,23 @@ import core_service.shared.test.AbstractUnitTest;
 import io.github.resilience4j.ratelimiter.RateLimiterRegistry;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpStatusCode;
 import org.springframework.http.MediaType;
 import org.springframework.test.web.client.MockRestServiceServer;
 import org.springframework.web.client.RestClient;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.springframework.test.web.client.match.MockRestRequestMatchers.queryParam;
 import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
 import static org.springframework.test.web.client.response.MockRestResponseCreators.withServerError;
+import static org.springframework.test.web.client.response.MockRestResponseCreators.withStatus;
+import static org.springframework.test.web.client.response.MockRestResponseCreators.withTooManyRequests;
 import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess;
 
 class SpeedrunApiServiceTest extends AbstractUnitTest {
@@ -66,6 +71,29 @@ class SpeedrunApiServiceTest extends AbstractUnitTest {
         assertThat(service.findGames("Celeste")).isEmpty();
     }
 
+    @Test
+    void findGames_TooManyRequests_ThrowsThrottledWithRetryAfter() {
+        server.expect(requestTo(containsString("/games"))).andRespond(withTooManyRequests(120));
+
+        assertThatThrownBy(() -> service.findGames("Celeste")).isInstanceOfSatisfying(SpeedrunThrottledException.class,
+                e -> assertThat(e.getRetryAfter()).isEqualTo(Duration.ofSeconds(120)));
+    }
+
+    @Test
+    void findGames_TooManyRequestsWithoutRetryAfter_ThrowsThrottledWithoutDelay() {
+        server.expect(requestTo(containsString("/games"))).andRespond(withTooManyRequests());
+
+        assertThatThrownBy(() -> service.findGames("Celeste")).isInstanceOfSatisfying(SpeedrunThrottledException.class,
+                e -> assertThat(e.getRetryAfter()).isNull());
+    }
+
+    @Test
+    void hasVerifiedRuns_Status420_ThrowsThrottled() {
+        server.expect(requestTo(containsString("/runs"))).andRespond(withStatus(HttpStatusCode.valueOf(420)));
+
+        assertThatThrownBy(() -> service.hasVerifiedRuns("o1y9j9v6")).isInstanceOf(SpeedrunThrottledException.class);
+    }
+
     @Test
     void findGames_Disabled_CallsNothing() {
         service = create(false);