diff --git a/src/main/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessService.java b/src/main/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessService.java index 762d8d8bd..0aba00e7d 100644 --- a/src/main/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessService.java +++ b/src/main/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessService.java @@ -33,7 +33,7 @@ @Profile("!ci | thread") public class LeetcodeQuestionProcessService { - private static final ReentrantLock LOCK = new ReentrantLock(); + private final ReentrantLock lock = new ReentrantLock(); private static final int MAX_JOBS_PER_RUN = 10; private static final long REQUESTS_OVER_TIME = 1L; @@ -89,7 +89,7 @@ private List claimBatch(final int maxSize) { @Scheduled(initialDelay = 0, fixedDelay = 30, timeUnit = TimeUnit.MINUTES) @Async public CompletableFuture drainQueue() { - if (!LOCK.tryLock()) { + if (!lock.tryLock()) { log.info("thread attempted to drain queue, but queue is already being drained."); return CompletableFuture.completedFuture(Empty.of()); } @@ -119,7 +119,7 @@ public CompletableFuture drainQueue() { } } } finally { - LOCK.unlock(); + lock.unlock(); } return CompletableFuture.completedFuture(Empty.of()); } diff --git a/src/test/java/org/patinanetwork/codebloom/common/utils/lock/QueueLockTest.java b/src/test/java/org/patinanetwork/codebloom/common/utils/lock/QueueLockTest.java index e7c0fbbb2..0ae118ae0 100644 --- a/src/test/java/org/patinanetwork/codebloom/common/utils/lock/QueueLockTest.java +++ b/src/test/java/org/patinanetwork/codebloom/common/utils/lock/QueueLockTest.java @@ -4,13 +4,11 @@ import static org.mockito.Mockito.*; import io.github.bucket4j.BlockingBucket; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -75,59 +73,67 @@ void acquireShouldAppendToEndOfQueue() throws InterruptedException { @Test @Timeout(value = 5, unit = TimeUnit.SECONDS) - void acquireFastShouldAppendToStartOfQueue() throws InterruptedException { - CountDownLatch tickerCallLatch = new CountDownLatch(1); - CountDownLatch releaseTicker = new CountDownLatch(1); + void acquireFastShouldAppendToStartOfQueue() throws Exception { + var tickerStarted = new CountDownLatch(1); + var releaseTicker = new CountDownLatch(1); + var releaseNormal = new CountDownLatch(1); + var normalSelected = new CountDownLatch(1); + var calls = new AtomicInteger(); doAnswer(invocation -> { - tickerCallLatch.countDown(); - releaseTicker.await(); + int call = calls.incrementAndGet(); + if (call == 1) { + tickerStarted.countDown(); + releaseTicker.await(); + } else if (call == 3) { + normalSelected.countDown(); + releaseNormal.await(); + } return null; }) .when(bucket) .consume(1); - executor.submit(() -> { - try { + try { + var initial = executor.submit(() -> { queueLock.acquire(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }); - assertTrue(tickerCallLatch.await(2, TimeUnit.SECONDS)); - - List orderedCalls = Collections.synchronizedList(new ArrayList<>()); - CountDownLatch doneLatch = new CountDownLatch(2); + return null; + }); + assertTrue(tickerStarted.await(2, TimeUnit.SECONDS)); - executor.submit(() -> { - try { + var normal = executor.submit(() -> { queueLock.acquire(); - orderedCalls.add("T2"); - doneLatch.countDown(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }); + return null; + }); + awaitQueueSize(1); - Thread.sleep(100); - - executor.submit(() -> { - try { + var fast = executor.submit(() -> { queueLock.acquireFast(); - orderedCalls.add("T3"); - doneLatch.countDown(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }); - - Thread.sleep(100); - - releaseTicker.countDown(); + return null; + }); + awaitQueueSize(2); + + releaseTicker.countDown(); + fast.get(2, TimeUnit.SECONDS); + assertTrue(normalSelected.await(2, TimeUnit.SECONDS)); + assertFalse(normal.isDone(), "Normal request must wait until the fast request is released"); + + releaseNormal.countDown(); + normal.get(2, TimeUnit.SECONDS); + initial.get(2, TimeUnit.SECONDS); + verify(bucket, times(3)).consume(1); + } finally { + releaseTicker.countDown(); + releaseNormal.countDown(); + } + } - assertTrue(doneLatch.await(2, TimeUnit.SECONDS)); - assertEquals("T3", orderedCalls.get(0)); - assertEquals("T2", orderedCalls.get(1)); + private void awaitQueueSize(int expected) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + while (queueLock.queue.size() != expected && System.nanoTime() < deadline) { + Thread.sleep(1); + } + assertEquals(expected, queueLock.queue.size(), "Requests did not enter the queue in time"); } @Test diff --git a/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceTest.java b/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceTest.java index 44217bcb0..3416b8f90 100644 --- a/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceTest.java +++ b/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceTest.java @@ -17,19 +17,24 @@ import org.patinanetwork.codebloom.common.db.models.question.QuestionDifficulty; import org.patinanetwork.codebloom.common.db.repos.job.JobRepository; import org.patinanetwork.codebloom.common.db.repos.question.QuestionRepository; +import org.patinanetwork.codebloom.common.leetcode.throttled.ThrottledLeetcodeClient; import org.patinanetwork.codebloom.common.time.StandardizedOffsetDateTime; import org.patinanetwork.codebloom.config.NoJdaRequired; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.bean.override.mockito.MockitoBean; -@SpringBootTest +@SpringBootTest(properties = {"codebloom.scheduling.enabled=false", "codebloom.notify.enabled=false"}) @ActiveProfiles({"ci", "thread"}) @TestInstance(TestInstance.Lifecycle.PER_CLASS) @DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) public class LeetcodeQuestionProcessServiceTest extends NoJdaRequired { + @MockitoBean + private ThrottledLeetcodeClient leetcodeClient; + private final JobRepository jobRepository; private final LeetcodeQuestionProcessService service; private final QuestionRepository questionRepository; @@ -177,7 +182,9 @@ void jobStatusTransitionValid() { @Test void drainQueueValid() { - service.drainQueue(); + service.drainQueue().join(); + assertEquals( + JobStatus.COMPLETE, jobRepository.findJobById(testJob.getId()).getStatus()); } // TODO: (TAN-32) re-enable diff --git a/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceUnitTest.java b/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceUnitTest.java index 9438ad0e8..8e7c34071 100644 --- a/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceUnitTest.java +++ b/src/test/java/org/patinanetwork/codebloom/scheduled/leetcode/LeetcodeQuestionProcessServiceUnitTest.java @@ -5,6 +5,8 @@ import java.util.List; import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -65,6 +67,48 @@ private void runQueue() { .join(); } + @Test + void independentServiceCanDrainWhileAnotherInstanceIsRunning() { + var otherJobs = mock(JobRepository.class); + when(otherJobs.findIncompleteJobs(10)).thenReturn(List.of()); + var otherService = new LeetcodeQuestionProcessService(otherJobs, client, questions, bank); + when(jobs.findIncompleteJobs(10)).thenAnswer(invocation -> { + CompletableFuture.runAsync(() -> otherService.drainQueue().join()).get(5, TimeUnit.SECONDS); + return List.of(); + }); + + runQueue(); + + verify(otherJobs).findIncompleteJobs(10); + } + + @Test + void sameServiceSkipsConcurrentDrain() { + var service = new LeetcodeQuestionProcessService(jobs, client, questions, bank); + when(jobs.findIncompleteJobs(10)).thenAnswer(invocation -> { + CompletableFuture.runAsync(() -> service.drainQueue().join()).get(5, TimeUnit.SECONDS); + return List.of(); + }); + + service.drainQueue().join(); + + verify(jobs).findIncompleteJobs(10); + } + + @Test + void failedDrainReleasesInstanceLock() { + var service = new LeetcodeQuestionProcessService(jobs, client, questions, bank); + when(jobs.findIncompleteJobs(10)) + .thenThrow(new IllegalStateException("Database unavailable")) + .thenReturn(List.of()); + + assertThrows(IllegalStateException.class, () -> service.drainQueue().join()); + assertDoesNotThrow(() -> + CompletableFuture.runAsync(() -> service.drainQueue().join()).get(5, TimeUnit.SECONDS)); + + verify(jobs, times(2)).findIncompleteJobs(10); + } + @ParameterizedTest @NullAndEmptySource @ValueSource(strings = {" ", "\t\n"})