diff --git a/readme.md b/readme.md index b0ef529..9342a92 100644 --- a/readme.md +++ b/readme.md @@ -98,7 +98,7 @@ only guard. Every service method a controller calls carries one role marker, on the method or on its class (the method wins): `@RequiresRole(Role.X)`, `@Public` (no login, e.g. landing stats; the public profile checks its ticket) or `@Internal` (only Auth's service token). A Spring Security method interceptor checks it while a request is served; -scheduled jobs and `@Async` work run without a user and are not checked. A denied call answers `401` without a +scheduled jobs and background work run without a user and are not checked. A denied call answers `401` without a login, else `403`. The URL rules in `SecurityConfig` stay as the outer guard. ArchUnit fails the build for an unmarked method, two markers, a controller calling an `@Async` method (the check would never run), and a service method calling another service's method that needs more. @@ -221,7 +221,7 @@ Inside a module: `api` (controllers, DTOs), `service` (business logic), `domain` Afterwards it writes every user's pile size of the day to `backlog_snapshot`, also for users without Steam, and dates finished games by the unlock of their ending achievement (`FinishedEstimateService`, `user_game.finished_at_estimate`). -3. `LibrarySyncWorker` runs in the background. `user_steam_sync` lets only one run per user start, and none for +3. `LibrarySyncWorker` runs on the background pool. `user_steam_sync` lets only one run per user start, and none for a user in `deleted_user`. 4. The worker loads the wishlist and the owned games from Steam. It updates `steam_game` and `user_game`, adds the playtime growth of the day to `user_game_playtime_day`, and soft-deletes Steam games the user no longer @@ -239,6 +239,38 @@ The frontend polls `GET /api/library/sync-status` until the run ends. At 01:30 `ImageRepairService` repairs the pictures reported through `POST /api/library/broken-images` (table `broken_image`, one row per game and kind). Reports only count for Steam games the user owns. +### Background work + +Background work runs on two pools (`AsyncConfig`; every `@Async` names one, ArchUnit checks it): + +- The default pool (4 threads, queue 100) runs the syncs and the time-to-beat and mood refreshes a page view + starts. A page view queues only games without a fresh row. When the queue is full, the task is dropped with + a warning, and the next view or sync asks again. Known limit: a dropped sync still answers `202`; the user + starts it again. +- The follow-up pool (2 threads, no limit on the queue) runs the achievement runs after a sync and every speedrun + refresh. They hold a thread for minutes because of the Steam and speedrun.com pace, so they do not take the page + views' threads. A task is small (a user or a few appids), and none is dropped. + +Imports have their own two threads, and `game_import` is their queue. Scheduled jobs (nightly sync, image repair, +import watchdog) run on the scheduler. + +```mermaid +flowchart LR + View["page view, add, merge"] -->|"stale games only"| Pool["default pool
4 threads, queue 100"] + Sync["sync request"] --> Pool + Synced["sync, after commit"] --> FollowUp["follow-up pool
2 threads"] + View -->|"speedrun"| FollowUp + Pool --> Apis["Steam, IGDB, speedrun.com"] + FollowUp --> Apis + Import["POST /imports"] --> Lanes["GameImportWorker
2 own threads"] + Cron["scheduler"] --> Jobs["nightly sync, image repair, import watchdog"] +``` + +Core runs as one instance. The pools, the speedrun queue, the per-user achievement runs and the achievement locks +live in memory: a second instance would ask Steam and IGDB twice, and the database would stay consistent (sync +claim, upserts). A restart drops what waits in the pools; it is all refresh work and is asked again. Imports survive +a restart. + ### Corrections by an auditor An AUDITOR (active role) corrects a game's name, header image, speedrun link, moods, endless flag and genres for diff --git a/src/main/java/core_service/configuration/AsyncConfig.java b/src/main/java/core_service/configuration/AsyncConfig.java index cbcc1c0..55188f1 100644 --- a/src/main/java/core_service/configuration/AsyncConfig.java +++ b/src/main/java/core_service/configuration/AsyncConfig.java @@ -1,9 +1,46 @@ package core_service.configuration; +import core_service.shared.background.BackgroundPools; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +/** + * The two pools of {@link BackgroundPools}. A dropped task is a refresh of data with a TTL: the next view or sync + * asks again. + */ @Configuration @EnableAsync +@Slf4j public class AsyncConfig { + + // A task holds a connection only for its short transactions. At most 6 background tasks, plus the scheduler and + // the 2 import lanes, use Hikari's 10 connections at once; requests share the rest. + static final int THREADS = 4; + static final int QUEUE = 100; + // The achievement runs share one Steam pace and speedrun has one drainer: more threads would only wait. + static final int FOLLOW_UP_THREADS = 2; + + @Bean(name = BackgroundPools.DEFAULT) + public ThreadPoolTaskExecutor applicationTaskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(THREADS); + executor.setMaxPoolSize(THREADS); + executor.setQueueCapacity(QUEUE); + executor.setThreadNamePrefix("background-"); + executor.setRejectedExecutionHandler((task, pool) -> log.warn(pool.isShutdown() + ? "Background pool stopped, task dropped" : "Background queue full, task dropped")); + return executor; + } + + @Bean(name = BackgroundPools.SYNC_FOLLOW_UP) + public ThreadPoolTaskExecutor syncFollowUpExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(FOLLOW_UP_THREADS); + executor.setMaxPoolSize(FOLLOW_UP_THREADS); + executor.setThreadNamePrefix("sync-follow-up-"); + return executor; + } } diff --git a/src/main/java/core_service/modules/library/service/AchievementSyncService.java b/src/main/java/core_service/modules/library/service/AchievementSyncService.java index 190f5fa..09bcbbf 100644 --- a/src/main/java/core_service/modules/library/service/AchievementSyncService.java +++ b/src/main/java/core_service/modules/library/service/AchievementSyncService.java @@ -15,6 +15,7 @@ import core_service.modules.library.repo.UserGameRepository; import core_service.modules.steam_adapter.api.dto.response.SteamPlayerAchievementsResponse; import core_service.modules.steam_adapter.api.dto.response.SteamSchemaForGameResponse; import core_service.modules.steam_adapter.service.SteamApiService; +import core_service.shared.background.BackgroundPools; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Async; @@ -93,7 +94,7 @@ public class AchievementSyncService { } } - @Async + @Async(BackgroundPools.SYNC_FOLLOW_UP) @TransactionalEventListener public void onLibrarySynced(LibrarySyncedEvent event) { syncLibrary(event.userId(), event.steamId()); diff --git a/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java b/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java index 0c40751..88b174c 100644 --- a/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java +++ b/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java @@ -37,6 +37,7 @@ import java.util.Set; import java.util.stream.Collectors; import core_service.modules.library.domain.UserSteamSync.Status; +import core_service.shared.background.BackgroundPools; /** * Worker bean. Runs the actual Steam fetch + DB upserts. Separate bean so the Spring proxy applies @Async @@ -68,7 +69,7 @@ public class LibrarySyncWorker { @Value("${app.library.zone:Europe/Berlin}") private ZoneId zone; - @Async + @Async(BackgroundPools.DEFAULT) public void runSync(Long userId, String steamId) { sync(userId, steamId); } diff --git a/src/main/java/core_service/modules/library/service/MoodService.java b/src/main/java/core_service/modules/library/service/MoodService.java index 5e21f16..36599f7 100644 --- a/src/main/java/core_service/modules/library/service/MoodService.java +++ b/src/main/java/core_service/modules/library/service/MoodService.java @@ -7,15 +7,13 @@ import core_service.modules.library.repo.UserGameRepository; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; -import org.springframework.scheduling.annotation.Async; +import org.springframework.core.task.TaskExecutor; import org.springframework.stereotype.Service; -import java.time.Duration; import java.time.Instant; import java.util.Collection; import java.util.Map; import java.util.Set; -import java.util.TreeSet; /** * Stores the IGDB genres of games and the {@link Mood}s from IGDB's genres, themes and game modes, the same way @@ -29,15 +27,14 @@ public class MoodService { private final IgdbApiService igdbApiService; private final GameMoodsRepository gameMoodsRepository; private final UserGameRepository userGameRepository; + private final TaskExecutor applicationTaskExecutor; @Value("${app.library.mood-ttl-days:30}") private long ttlDays; public void refresh(Collection appids) { if (!igdbApiService.isEnabled() || appids.isEmpty()) return; - // Sorted: parallel syncs then lock the upserted rows in the same order and cannot deadlock. - Set stale = new TreeSet<>(appids); - stale.removeAll(gameMoodsRepository.findFreshAppids(appids, Instant.now().minus(Duration.ofDays(ttlDays)))); + Set stale = stale(appids); if (stale.isEmpty()) return; IgdbLookup.Answer answer = IgdbLookup.byAppid(stale, igdbApiService::getGameTags, @@ -62,36 +59,31 @@ public class MoodService { /** * Same as {@link #refresh}, but the request does not wait for IGDB, and an error does not fail it. */ - @Async public void refreshInBackground(Collection appids) { - try { - refresh(appids); - } catch (Exception e) { - log.warn("Moods of {} not refreshed: {}", appids, e.getMessage()); - } + if (!igdbApiService.isEnabled()) return; + Refreshes.staleInBackground(applicationTaskExecutor, "Moods of " + appids, () -> appids, this::stale, + this::refresh); } /** * For the genres and moods of the stats page: a user without a Steam sync gets them only here. */ - @Async public void refreshListedInBackground(Long userId) { - try { - refresh(userGameRepository.findAllListedAppids(userId)); - } catch (Exception e) { - log.warn("Moods of the library of user {} not refreshed: {}", userId, e.getMessage()); - } + if (!igdbApiService.isEnabled()) return; + Refreshes.staleInBackground(applicationTaskExecutor, "Moods of the library of user " + userId, + () -> userGameRepository.findAllListedAppids(userId), this::stale, this::refresh); } /** * For the picker: a user without a Steam sync gets moods only here, and a Steam user before the next sync. */ - @Async public void refreshBacklogInBackground(Long userId) { - try { - refresh(userGameRepository.findBacklogAppids(userId)); - } catch (Exception e) { - log.warn("Backlog moods of user {} not refreshed: {}", userId, e.getMessage()); - } + if (!igdbApiService.isEnabled()) return; + Refreshes.staleInBackground(applicationTaskExecutor, "Backlog moods of user " + userId, + () -> userGameRepository.findBacklogAppids(userId), this::stale, this::refresh); + } + + private Set stale(Collection appids) { + return Refreshes.stale(appids, gameMoodsRepository::findFreshAppids, ttlDays); } } diff --git a/src/main/java/core_service/modules/library/service/Refreshes.java b/src/main/java/core_service/modules/library/service/Refreshes.java new file mode 100644 index 0000000..0288447 --- /dev/null +++ b/src/main/java/core_service/modules/library/service/Refreshes.java @@ -0,0 +1,64 @@ +package core_service.modules.library.service; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.core.task.TaskExecutor; + +import java.time.Duration; +import java.time.Instant; +import java.util.Collection; +import java.util.Set; +import java.util.TreeSet; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +/** + * The shared steps of the IGDB and speedrun.com refreshes: which games have no fresh row, and the background run. + */ +@Slf4j +final class Refreshes { + + private Refreshes() { + } + + /** + * Sorted: parallel syncs then lock the upserted rows in the same order and cannot deadlock. + * + * @param findFreshAppids the appids among the given ones with a row newer than the instant + */ + static Set stale(Collection appids, BiFunction, Instant, Set> findFreshAppids, + long ttlDays) { + Set stale = new TreeSet<>(appids); + stale.removeAll(findFreshAppids.apply(appids, Instant.now().minus(Duration.ofDays(ttlDays)))); + return stale; + } + + /** + * Checks on the calling thread which games are stale and refreshes only those in the background, so a fresh game + * takes no place in the queue. No error reaches the caller: the next view or sync asks again. + * + * @param appids read on the calling thread + */ + static void staleInBackground(TaskExecutor executor, String what, Supplier> appids, + Function, Set> stale, Consumer> refresh) { + try { + Collection candidates = appids.get(); + if (candidates.isEmpty()) return; + Set staleAppids = stale.apply(candidates); + if (staleAppids.isEmpty()) return; + // ponytail: views of one stale game in the same second ask once each; an in-flight set if that hurts + executor.execute(() -> quietly(what, () -> refresh.accept(staleAppids))); + } catch (Exception e) { + log.warn("{} not refreshed: {}", what, e.getMessage()); + } + } + + static void quietly(String what, Runnable refresh) { + try { + refresh.run(); + } catch (Exception e) { + log.warn("{} not refreshed: {}", what, e.getMessage()); + } + } +} 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 03e0369..c693a31 100644 --- a/src/main/java/core_service/modules/library/service/SpeedrunService.java +++ b/src/main/java/core_service/modules/library/service/SpeedrunService.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.shared.background.BackgroundPools; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; +import org.springframework.core.task.TaskExecutor; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import org.springframework.transaction.event.TransactionalEventListener; @@ -46,6 +48,7 @@ public class SpeedrunService { private final GameSpeedrunRepository gameSpeedrunRepository; private final SteamGameRepository steamGameRepository; private final UserGameRepository userGameRepository; + private final TaskExecutor syncFollowUpExecutor; private final ConcurrentSkipListSet pending = new ConcurrentSkipListSet<>(); private final AtomicBoolean running = new AtomicBoolean(); @@ -71,28 +74,27 @@ public class SpeedrunService { } } - @Async + @Async(BackgroundPools.SYNC_FOLLOW_UP) @TransactionalEventListener public void onLibrarySynced(LibrarySyncedEvent event) { - refreshBacklogInBackground(event.userId()); + Refreshes.quietly("Speedrun games of the backlog of user " + event.userId(), + () -> refresh(userGameRepository.findBacklogAppids(event.userId()))); } - @Async public void refreshBacklogInBackground(Long userId) { - try { - refresh(userGameRepository.findBacklogAppids(userId)); - } catch (Exception e) { - log.warn("Speedrun games of the backlog of user {} not refreshed: {}", userId, e.getMessage()); - } + if (!speedrunApiService.isEnabled()) return; + Refreshes.staleInBackground(syncFollowUpExecutor, "Speedrun games of the backlog of user " + userId, + () -> userGameRepository.findBacklogAppids(userId), this::stale, this::refresh); } - @Async public void refreshInBackground(Collection appids) { - try { - refresh(appids); - } catch (Exception e) { - log.warn("Speedrun games of {} not refreshed: {}", appids, e.getMessage()); - } + if (!speedrunApiService.isEnabled()) return; + Refreshes.staleInBackground(syncFollowUpExecutor, "Speedrun games of " + appids, () -> appids, this::stale, + this::refresh); + } + + private Set stale(Collection appids) { + return Refreshes.stale(appids, gameSpeedrunRepository::findFreshAppids, ttlDays); } private void drain() { diff --git a/src/main/java/core_service/modules/library/service/TimeToBeatService.java b/src/main/java/core_service/modules/library/service/TimeToBeatService.java index 10b056c..0c32d0a 100644 --- a/src/main/java/core_service/modules/library/service/TimeToBeatService.java +++ b/src/main/java/core_service/modules/library/service/TimeToBeatService.java @@ -5,15 +5,13 @@ import core_service.modules.library.repo.GameTimeToBeatRepository; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; -import org.springframework.scheduling.annotation.Async; +import org.springframework.core.task.TaskExecutor; import org.springframework.stereotype.Service; -import java.time.Duration; import java.time.Instant; import java.util.Collection; import java.util.Map; import java.util.Set; -import java.util.TreeSet; @Service @RequiredArgsConstructor @@ -22,20 +20,19 @@ public class TimeToBeatService { private final IgdbApiService igdbApiService; private final GameTimeToBeatRepository gameTimeToBeatRepository; + private final TaskExecutor applicationTaskExecutor; @Value("${app.library.time-to-beat-ttl-days:30}") private long ttlDays; /** - * Same as {@link #refresh}, but the request does not wait for IGDB, and an error does not fail it. + * Same as {@link #refresh}, but the request does not wait for IGDB, and an error does not fail it. Only stale apps + * take a place in the background queue. */ - @Async public void refreshInBackground(Collection appids) { - try { - refresh(appids); - } catch (Exception e) { - log.warn("Time to beat of {} not refreshed: {}", appids, e.getMessage()); - } + if (!igdbApiService.isEnabled()) return; + Refreshes.staleInBackground(applicationTaskExecutor, "Time to beat of " + appids, () -> appids, this::stale, + this::refresh); } /** @@ -44,9 +41,7 @@ public class TimeToBeatService { */ public void refresh(Collection appids) { if (!igdbApiService.isEnabled() || appids.isEmpty()) return; - // Sorted: parallel syncs then lock the upserted rows in the same order and cannot deadlock. - Set stale = new TreeSet<>(appids); - stale.removeAll(gameTimeToBeatRepository.findFreshAppids(appids, Instant.now().minus(Duration.ofDays(ttlDays)))); + Set stale = stale(appids); if (stale.isEmpty()) return; IgdbLookup.Answer answer = IgdbLookup.byAppid(stale, igdbApiService::getTimeToBeat, @@ -66,4 +61,8 @@ public class TimeToBeatService { } } } + + private Set stale(Collection appids) { + return Refreshes.stale(appids, gameTimeToBeatRepository::findFreshAppids, ttlDays); + } } diff --git a/src/main/java/core_service/shared/background/BackgroundPools.java b/src/main/java/core_service/shared/background/BackgroundPools.java new file mode 100644 index 0000000..f346d65 --- /dev/null +++ b/src/main/java/core_service/shared/background/BackgroundPools.java @@ -0,0 +1,22 @@ +package core_service.shared.background; + +import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration; + +/** + * The bean names of the two background pools (configuration.AsyncConfig). Every {@code @Async} names one of them: + * with the scheduler there are two TaskExecutor beans, and a bare {@code @Async} would run unbounded. + */ +public final class BackgroundPools { + + /** Syncs and the refreshes a page view starts: short tasks, a full queue drops the task. */ + public static final String DEFAULT = TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME; + + /** + * The achievement runs after a sync and every speedrun refresh: they hold a thread for minutes (Steam and + * speedrun.com pace), so page views keep the default pool. Small tasks, nothing is dropped. + */ + public static final String SYNC_FOLLOW_UP = "syncFollowUpExecutor"; + + private BackgroundPools() { + } +} diff --git a/src/main/java/core_service/shared/security/RequiresRole.java b/src/main/java/core_service/shared/security/RequiresRole.java index 3adb3fa..92dd5f3 100644 --- a/src/main/java/core_service/shared/security/RequiresRole.java +++ b/src/main/java/core_service/shared/security/RequiresRole.java @@ -10,7 +10,7 @@ import java.lang.annotation.Target; * The lowest active role that may call this service method, or every method of this class. A method marker * replaces the class marker. Every service method a controller calls carries one of {@link RequiresRole}, * {@link Public} or {@link Internal} (ArchitectureTest). Checked only while an HTTP request is served: scheduled - * jobs and {@code @Async} work run without a user. + * jobs and background work (the pools in {@code AsyncConfig}) run without a user. */ @Documented @Retention(RetentionPolicy.RUNTIME) diff --git a/src/test/java/core_service/architecture/ArchitectureTest.java b/src/test/java/core_service/architecture/ArchitectureTest.java index 110c671..0c39925 100644 --- a/src/test/java/core_service/architecture/ArchitectureTest.java +++ b/src/test/java/core_service/architecture/ArchitectureTest.java @@ -1,15 +1,24 @@ package core_service.architecture; +import core_service.shared.background.BackgroundPools; import core_service.shared.test.AbstractArchitectureTest; import com.tngtech.archunit.base.DescribedPredicate; import com.tngtech.archunit.core.domain.JavaClass; +import com.tngtech.archunit.core.domain.JavaMethod; import com.tngtech.archunit.core.importer.ImportOption; import com.tngtech.archunit.junit.AnalyzeClasses; import com.tngtech.archunit.junit.ArchTest; +import com.tngtech.archunit.lang.ArchCondition; import com.tngtech.archunit.lang.ArchRule; +import com.tngtech.archunit.lang.ConditionEvents; +import com.tngtech.archunit.lang.SimpleConditionEvent; import com.tngtech.archunit.library.dependencies.SlicesRuleDefinition; +import org.springframework.scheduling.annotation.Async; + +import java.util.Set; import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.classes; +import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.methods; import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses; /** @@ -93,6 +102,24 @@ class ArchitectureTest extends AbstractArchitectureTest { .should().haveSimpleNameEndingWith("Repository") .because("Repository interfaces are the persistence boundary; naming makes them explicit."); + @ArchTest + static final ArchRule async_methods_should_name_a_pool = methods() + .that().areAnnotatedWith(Async.class) + .should(new ArchCondition<>("name a pool of BackgroundPools") { + @Override + public void check(JavaMethod method, ConditionEvents events) { + String pool = method.getAnnotationOfType(Async.class).value(); + boolean named = Set.of(BackgroundPools.DEFAULT, BackgroundPools.SYNC_FOLLOW_UP).contains(pool); + events.add(new SimpleConditionEvent(method, named, method.getFullName() + " runs on '" + pool + "'")); + } + }) + .because("A bare @Async runs on an unbounded executor (BackgroundPools)."); + + @ArchTest + static final ArchRule no_class_level_async = noClasses() + .should().beAnnotatedWith(Async.class) + .because("@Async on a class names no pool for its methods; put it on each method."); + // =================================================================================== // 3. SHARED / CONFIGURATION RULES // =================================================================================== diff --git a/src/test/java/core_service/configuration/AsyncConfigTest.java b/src/test/java/core_service/configuration/AsyncConfigTest.java new file mode 100644 index 0000000..a3bd41f --- /dev/null +++ b/src/test/java/core_service/configuration/AsyncConfigTest.java @@ -0,0 +1,72 @@ +package core_service.configuration; + +import core_service.shared.test.AbstractUnitTest; +import org.junit.jupiter.api.Test; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class AsyncConfigTest extends AbstractUnitTest { + + @Test + void fullQueue_DropsTheTaskWithoutFailingTheCaller() throws InterruptedException { + ThreadPoolTaskExecutor executor = new AsyncConfig().applicationTaskExecutor(); + executor.initialize(); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch started = new CountDownLatch(AsyncConfig.THREADS); + AtomicBoolean droppedRan = new AtomicBoolean(); + try { + for (int i = 0; i < AsyncConfig.THREADS; i++) { + executor.execute(() -> { + started.countDown(); + awaitQuietly(release); + }); + } + assertTrue(started.await(5, TimeUnit.SECONDS)); + for (int i = 0; i < AsyncConfig.QUEUE; i++) { + executor.execute(() -> { }); + } + + assertDoesNotThrow(() -> executor.execute(() -> droppedRan.set(true))); + + assertEquals(AsyncConfig.QUEUE, executor.getQueueSize()); + } finally { + release.countDown(); + executor.shutdown(); + } + assertTrue(executor.getThreadPoolExecutor().awaitTermination(5, TimeUnit.SECONDS)); + assertFalse(droppedRan.get()); + } + + @Test + void followUpPool_KeepsEveryTask() throws InterruptedException { + ThreadPoolTaskExecutor executor = new AsyncConfig().syncFollowUpExecutor(); + executor.initialize(); + int tasks = AsyncConfig.QUEUE * 3; + CountDownLatch done = new CountDownLatch(tasks); + try { + for (int i = 0; i < tasks; i++) { + executor.execute(done::countDown); + } + + assertTrue(done.await(5, TimeUnit.SECONDS)); + } finally { + executor.shutdown(); + } + } + + private static void awaitQuietly(CountDownLatch latch) { + try { + latch.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/src/test/java/core_service/modules/library/service/MoodServiceTest.java b/src/test/java/core_service/modules/library/service/MoodServiceTest.java index 2cc419f..e1b57c5 100644 --- a/src/test/java/core_service/modules/library/service/MoodServiceTest.java +++ b/src/test/java/core_service/modules/library/service/MoodServiceTest.java @@ -8,6 +8,9 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; import org.mockito.Mock; +import org.mockito.Spy; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; import org.springframework.test.util.ReflectionTestUtils; import java.time.Instant; @@ -38,6 +41,9 @@ class MoodServiceTest extends AbstractUnitTest { @Mock private UserGameRepository userGameRepository; + @Spy + private TaskExecutor applicationTaskExecutor = new SyncTaskExecutor(); + @InjectMocks private MoodService service; @@ -72,6 +78,16 @@ class MoodServiceTest extends AbstractUnitTest { verify(gameMoodsRepository).upsert(eq(427520L), any(), any(), eq(true), any(Instant.class)); } + @Test + void refreshInBackground_AllFresh_QueuesNothing() { + when(igdbApiService.isEnabled()).thenReturn(true); + when(gameMoodsRepository.findFreshAppids(anyCollection(), any())).thenReturn(Set.of(620L)); + + service.refreshInBackground(List.of(620L)); + + verify(applicationTaskExecutor, never()).execute(any()); + } + @Test void refreshBacklogInBackground_RefreshesTheBacklogOfTheUser() { when(igdbApiService.isEnabled()).thenReturn(true); 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 f448fb5..845fbd6 100644 --- a/src/test/java/core_service/modules/library/service/SpeedrunServiceTest.java +++ b/src/test/java/core_service/modules/library/service/SpeedrunServiceTest.java @@ -11,6 +11,9 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; import org.mockito.Mock; +import org.mockito.Spy; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; import org.springframework.dao.DataAccessResourceFailureException; import java.time.Instant; @@ -44,6 +47,8 @@ class SpeedrunServiceTest extends AbstractUnitTest { private SteamGameRepository steamGameRepository; @Mock private UserGameRepository userGameRepository; + @Spy + private TaskExecutor syncFollowUpExecutor = new SyncTaskExecutor(); @InjectMocks private SpeedrunService service; @@ -222,4 +227,14 @@ class SpeedrunServiceTest extends AbstractUnitTest { verifyNoInteractions(gameSpeedrunRepository, steamGameRepository); } + + @Test + void onLibrarySynced_RefreshesTheBacklogOnTheListenerThread() { + when(userGameRepository.findBacklogAppids(7L)).thenReturn(List.of()); + + service.onLibrarySynced(new LibrarySyncedEvent(7L, "76561198000000000")); + + verify(userGameRepository).findBacklogAppids(7L); + verify(syncFollowUpExecutor, never()).execute(any()); + } } diff --git a/src/test/java/core_service/modules/library/service/TimeToBeatServiceTest.java b/src/test/java/core_service/modules/library/service/TimeToBeatServiceTest.java index 19840b0..0368356 100644 --- a/src/test/java/core_service/modules/library/service/TimeToBeatServiceTest.java +++ b/src/test/java/core_service/modules/library/service/TimeToBeatServiceTest.java @@ -7,6 +7,10 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; import org.mockito.Mock; +import org.mockito.Spy; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.dao.QueryTimeoutException; +import org.springframework.core.task.TaskExecutor; import org.springframework.test.util.ReflectionTestUtils; import java.time.Instant; @@ -15,6 +19,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyCollection; import static org.mockito.ArgumentMatchers.eq; @@ -31,6 +36,9 @@ class TimeToBeatServiceTest extends AbstractUnitTest { @Mock private GameTimeToBeatRepository gameTimeToBeatRepository; + @Spy + private TaskExecutor applicationTaskExecutor = new SyncTaskExecutor(); + @InjectMocks private TimeToBeatService service; @@ -53,6 +61,37 @@ class TimeToBeatServiceTest extends AbstractUnitTest { verify(gameTimeToBeatRepository, never()).upsert(eq(220L), any(), any(), any(), any()); } + @Test + void refreshInBackground_AllFresh_QueuesNothing() { + when(igdbApiService.isEnabled()).thenReturn(true); + when(gameTimeToBeatRepository.findFreshAppids(anyCollection(), any())).thenReturn(Set.of(220L, 620L)); + + service.refreshInBackground(List.of(220L, 620L)); + + verify(applicationTaskExecutor, never()).execute(any()); + } + + @Test + void refreshInBackground_StaleCheckFails_DoesNotFailTheCaller() { + when(igdbApiService.isEnabled()).thenReturn(true); + when(gameTimeToBeatRepository.findFreshAppids(anyCollection(), any())).thenThrow(new QueryTimeoutException("slow")); + + assertDoesNotThrow(() -> service.refreshInBackground(List.of(620L))); + verify(applicationTaskExecutor, never()).execute(any()); + } + + @Test + void refreshInBackground_QueuesOnlyTheStaleApps() { + when(igdbApiService.isEnabled()).thenReturn(true); + when(gameTimeToBeatRepository.findFreshAppids(anyCollection(), any())).thenReturn(Set.of(220L)); + when(igdbApiService.getTimeToBeat(Set.of(620L))).thenReturn(Optional.of(Map.of())); + + service.refreshInBackground(List.of(220L, 620L)); + + verify(igdbApiService).getTimeToBeat(Set.of(620L)); + verify(gameTimeToBeatRepository).upsert(eq(620L), isNull(), isNull(), isNull(), any(Instant.class)); + } + @Test void refresh_GamesOffSteam_AreAskedByIgdbIdAndStoredUnderTheirKey() { when(igdbApiService.isEnabled()).thenReturn(true);