diff --git a/readme.md b/readme.md index 10e94a9..b080faf 100644 --- a/readme.md +++ b/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 diff --git a/src/main/java/core_service/modules/library/service/SpeedrunService.java b/src/main/java/core_service/modules/library/service/SpeedrunService.java index 0ff3926..023f1b3 100644 --- a/src/main/java/core_service/modules/library/service/SpeedrunService.java +++ b/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". *

+ * 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. + *

* {@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 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 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 poll(ConcurrentSkipListSet queue, int size) { List batch = new ArrayList<>(); Long next; diff --git a/src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunApiService.java b/src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunApiService.java index 86eaebb..7784837 100644 --- a/src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunApiService.java +++ b/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> 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 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(); } } -} \ No newline at end of file + + 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); + } +} diff --git a/src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunThrottledException.java b/src/main/java/core_service/modules/speedrun_adapter/service/SpeedrunThrottledException.java new file mode 100644 index 0000000..f57ff07 --- /dev/null +++ b/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; + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 67c1e59..6f0914e 100644 --- a/src/main/resources/application.yml +++ b/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. diff --git a/src/test/java/core_service/modules/library/service/SpeedrunServiceTest.java b/src/test/java/core_service/modules/library/service/SpeedrunServiceTest.java index e8d829c..fbbc68d 100644 --- a/src/test/java/core_service/modules/library/service/SpeedrunServiceTest.java +++ b/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 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> 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); diff --git a/src/test/java/core_service/modules/speedrun_adapter/service/SpeedrunApiServiceTest.java b/src/test/java/core_service/modules/speedrun_adapter/service/SpeedrunApiServiceTest.java index dc7f73b..245c8c8 100644 --- a/src/test/java/core_service/modules/speedrun_adapter/service/SpeedrunApiServiceTest.java +++ b/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);