refactor(library): two bounded background pools

Every @Async names a pool: with the scheduler there are two TaskExecutor beans, and a bare @Async ran on an unbounded executor. Page views queue only stale games; the long runs after a sync get their own pool.

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
     // ===================================================================================

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();
+        }
+    }
+}

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);

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());
+    }
 }

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);