diff --git a/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.test.tsx b/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.test.tsx index cb414acbc..09772ffe9 100644 --- a/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.test.tsx +++ b/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.test.tsx @@ -37,7 +37,7 @@ describe("RequeueIncompleteQuestionsModal", () => { ); expect( - await screen.findByText(/refresh the LeetCode token first/), + await screen.findByRole("button", { name: "Requeue" }), ).toBeInTheDocument(); expect(requestCount).toBe(0); }); diff --git a/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.tsx b/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.tsx index 5c3bc9903..ce31796ac 100644 --- a/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.tsx +++ b/js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.tsx @@ -34,8 +34,10 @@ export default function RequeueIncompleteQuestionsModal() { >
- Creates a new processing job for every incomplete question that does - not already have an active job. + Queues incomplete questions without an active job and restarts jobs + stuck processing for at least one hour with a fresh retry budget. + Already queued jobs and jobs processing for less than an hour are + left unchanged. Each job calls LeetCode. If jobs have been failing because the auth diff --git a/js/src/lib/api/queries/incomplete/index.ts b/js/src/lib/api/queries/incomplete/index.ts index 473d2b081..82d77eb77 100644 --- a/js/src/lib/api/queries/incomplete/index.ts +++ b/js/src/lib/api/queries/incomplete/index.ts @@ -26,7 +26,7 @@ async function fetchIncompleteQuestions({ } /** - * Queue a new job for every incomplete question without an active job. + * Restart jobs stuck processing for at least one hour and queue incomplete questions without an active job. */ export const useRequeueIncompleteQuestionsMutation = () => { const queryClient = useQueryClient(); diff --git a/src/main/java/org/patinanetwork/codebloom/api/admin/AdminController.java b/src/main/java/org/patinanetwork/codebloom/api/admin/AdminController.java index 606c68948..3244b17a4 100644 --- a/src/main/java/org/patinanetwork/codebloom/api/admin/AdminController.java +++ b/src/main/java/org/patinanetwork/codebloom/api/admin/AdminController.java @@ -263,8 +263,9 @@ public ResponseEntity>> getIncompleteQues } @Operation(summary = "Requeue incomplete questions", description = """ - Creates a new processing job for every incomplete question that does not already have an active job. Questions \ - that already have an INCOMPLETE or PROCESSING job are skipped. Only accessible to admins. + Restarts jobs stuck in PROCESSING for at least one hour with a fresh retry budget and creates a new job for \ + every incomplete question without an active job. INCOMPLETE jobs and more recent PROCESSING jobs are skipped. \ + The queued count includes recovered jobs. Only accessible to admins. """) @PostMapping("/questions/incomplete/requeue") public ResponseEntity> requeueIncompleteQuestions( diff --git a/src/main/java/org/patinanetwork/codebloom/common/components/IncompleteQuestionRequeuer.java b/src/main/java/org/patinanetwork/codebloom/common/components/IncompleteQuestionRequeuer.java index 5d19e76a3..a8503ea96 100644 --- a/src/main/java/org/patinanetwork/codebloom/common/components/IncompleteQuestionRequeuer.java +++ b/src/main/java/org/patinanetwork/codebloom/common/components/IncompleteQuestionRequeuer.java @@ -23,22 +23,26 @@ public IncompleteQuestionRequeuer(final QuestionRepository questionRepository, f } /** - * Creates a new job for every incomplete question that has no active job. + * Restarts jobs stuck in PROCESSING for at least one hour and creates a new job for every incomplete question that + * has no active job. * - *

Safe to call concurrently: the active-job unique index rejects duplicates, which are counted as skipped. + *

Concurrent calls only count jobs they recover or create: recovery checks the current status, and the + * active-job unique index rejects duplicate inserts. * - * @return number of jobs actually created. + * @return number of jobs recovered or created. */ public int requeueIncompleteQuestions() { - ArrayList questions = questionRepository.getAllIncompleteQuestionsWithNoJob(); + int requeuedCount = jobRepository.requeueStaleProcessingJobs( + StandardizedOffsetDateTime.now().minusHours(1)); - if (questions.isEmpty()) { - log.info("All questions have complete information."); + ArrayList incompleteQuestionsWithNoJob = questionRepository.getAllIncompleteQuestionsWithNoJob(); + + if (requeuedCount == 0 && incompleteQuestionsWithNoJob.isEmpty()) { + log.info("No incomplete questions need requeueing."); return 0; } - int queued = 0; - for (Question question : questions) { + for (Question question : incompleteQuestionsWithNoJob) { String questionId = question.getId(); log.warn("Question with id of {} does not have complete information.", questionId); Job newJob = Job.builder() @@ -49,11 +53,11 @@ public int requeueIncompleteQuestions() { if (jobRepository.createJob(newJob)) { log.info("Added job for question of id {}", questionId); - queued++; + requeuedCount++; } else { log.info("Question of id {} already has an active job, skipping", questionId); } } - return queued; + return requeuedCount; } } diff --git a/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepository.java b/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepository.java index d5669db11..b41a7a1ec 100644 --- a/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepository.java +++ b/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepository.java @@ -1,5 +1,6 @@ package org.patinanetwork.codebloom.common.db.repos.job; +import java.time.OffsetDateTime; import java.util.List; import java.util.Optional; import org.patinanetwork.codebloom.common.db.models.job.Job; @@ -49,6 +50,14 @@ public interface JobRepository { */ List findIncompleteJobs(int maxJobs); + /** + * Restarts PROCESSING jobs whose processing began at or before the cutoff, with zero attempts and no processing or + * completion timestamps. Recovered jobs are immediately eligible for processing. + * + * @return number of jobs recovered + */ + int requeueStaleProcessingJobs(OffsetDateTime cutoff); + /** * Updates an existing job in the database. * diff --git a/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobSqlRepository.java b/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobSqlRepository.java index d0246c7e3..ef7febdcf 100644 --- a/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobSqlRepository.java +++ b/src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobSqlRepository.java @@ -147,6 +147,24 @@ public List findIncompleteJobs(final int maxJobs) { .list(); } + @Override + public int requeueStaleProcessingJobs(final OffsetDateTime cutoff) { + String sql = """ + UPDATE "Job" + SET + status = 'INCOMPLETE', + attempts = 0, + "processedAt" = NULL, + "completedAt" = NULL, + "nextAttemptAt" = NOW() + WHERE + status = 'PROCESSING' + AND "processedAt" <= :cutoff + """; + + return jdbcClient.sql(sql).param("cutoff", cutoff).update(); + } + @Override public boolean updateJob(final Job job) { String sql = """ diff --git a/src/test/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepositoryTest.java b/src/test/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepositoryTest.java index bd94924dd..23d3649b0 100644 --- a/src/test/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepositoryTest.java +++ b/src/test/java/org/patinanetwork/codebloom/common/db/repos/job/JobRepositoryTest.java @@ -7,10 +7,17 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import java.time.OffsetDateTime; +import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -19,13 +26,16 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; import org.junit.jupiter.api.TestMethodOrder; +import org.patinanetwork.codebloom.common.components.IncompleteQuestionRequeuer; import org.patinanetwork.codebloom.common.db.models.job.Job; import org.patinanetwork.codebloom.common.db.models.job.JobStatus; import org.patinanetwork.codebloom.common.db.repos.BaseRepositoryTest; +import org.patinanetwork.codebloom.common.db.repos.question.QuestionRepository; import org.patinanetwork.codebloom.common.time.StandardizedOffsetDateTime; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.dao.DuplicateKeyException; +import org.springframework.transaction.annotation.Transactional; @SpringBootTest @TestInstance(TestInstance.Lifecycle.PER_CLASS) @@ -160,4 +170,107 @@ void testOnlyOneActiveJobPerQuestion() { } } } + + @Test + @Transactional + void recoveryResetsOnlyProcessingJobsAtOrBeforeCutoff() { + jobRepository.deleteAllJobs(); + OffsetDateTime cutoff = StandardizedOffsetDateTime.now().minusHours(1); + Job stale = recoveryJob(JobStatus.PROCESSING, Optional.of(cutoff.minusSeconds(1))); + Job boundary = recoveryJob(JobStatus.PROCESSING, Optional.of(cutoff)); + Job recent = recoveryJob(JobStatus.PROCESSING, Optional.of(cutoff.plusSeconds(1))); + Job queued = recoveryJob(JobStatus.INCOMPLETE, Optional.of(cutoff.minusHours(1))); + Job complete = recoveryJob(JobStatus.COMPLETE, Optional.of(cutoff.minusHours(1))); + Job unknownStart = recoveryJob(JobStatus.PROCESSING, Optional.empty()); + + OffsetDateTime beforeRecovery = StandardizedOffsetDateTime.now(); + assertEquals(2, jobRepository.requeueStaleProcessingJobs(cutoff)); + + for (Job original : List.of(stale, boundary)) { + Job recovered = jobRepository.findJobById(original.getId()).orElseThrow(); + assertEquals(JobStatus.INCOMPLETE, recovered.getStatus()); + assertEquals(0, recovered.getAttempts()); + assertTrue(recovered.getProcessedAt().isEmpty()); + assertTrue(recovered.getCompletedAt().isEmpty()); + assertFalse(recovered.getNextAttemptAt().isAfter(beforeRecovery)); + assertEquals(original.getCreatedAt(), recovered.getCreatedAt()); + assertEquals(List.of(recovered), jobRepository.findJobsByQuestionId(original.getQuestionId())); + assertTrue(jobRepository.findIncompleteJobs(10).contains(recovered)); + } + for (Job unchanged : List.of(recent, queued, complete, unknownStart)) { + assertEquals(unchanged, jobRepository.findJobById(unchanged.getId()).orElseThrow()); + } + assertEquals(0, jobRepository.requeueStaleProcessingJobs(cutoff)); + } + + @Test + @Transactional + void requeueCountsRecoveredJobsWithoutCreatingNewJobs() { + jobRepository.deleteAllJobs(); + Job stale = recoveryJob( + JobStatus.PROCESSING, + Optional.of(StandardizedOffsetDateTime.now().minusHours(2))); + QuestionRepository questions = mock(QuestionRepository.class); + when(questions.getAllIncompleteQuestionsWithNoJob()).thenReturn(new ArrayList<>()); + var requeuer = new IncompleteQuestionRequeuer(questions, jobRepository); + + assertEquals(1, requeuer.requeueIncompleteQuestions()); + Job recovered = jobRepository.findJobById(stale.getId()).orElseThrow(); + assertEquals(JobStatus.INCOMPLETE, recovered.getStatus()); + assertEquals(0, recovered.getAttempts()); + assertEquals(List.of(recovered), jobRepository.findJobsByQuestionId(stale.getQuestionId())); + assertEquals(0, requeuer.requeueIncompleteQuestions()); + } + + @Test + void concurrentRequeuesRecoverAndCountTheStaleJobOnlyOnce() throws Exception { + Job stale = recoveryJob( + JobStatus.PROCESSING, + Optional.of(StandardizedOffsetDateTime.now().minusHours(2))); + QuestionRepository questions = mock(QuestionRepository.class); + when(questions.getAllIncompleteQuestionsWithNoJob()).thenReturn(new ArrayList<>()); + var firstRequeuer = new IncompleteQuestionRequeuer(questions, jobRepository); + var secondRequeuer = new IncompleteQuestionRequeuer(questions, jobRepository); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + + try (var executor = Executors.newFixedThreadPool(2)) { + var results = List.of(firstRequeuer, secondRequeuer).stream() + .map(requeuer -> executor.submit(() -> { + ready.countDown(); + assertTrue(start.await(10, TimeUnit.SECONDS)); + return requeuer.requeueIncompleteQuestions(); + })) + .toList(); + assertTrue(ready.await(10, TimeUnit.SECONDS)); + start.countDown(); + int firstCount = results.get(0).get(10, TimeUnit.SECONDS); + int secondCount = results.get(1).get(10, TimeUnit.SECONDS); + + assertEquals( + List.of(0, 1), + List.of(firstCount, secondCount).stream().sorted().toList()); + Job recovered = jobRepository.findJobById(stale.getId()).orElseThrow(); + assertEquals(JobStatus.INCOMPLETE, recovered.getStatus()); + assertEquals(0, recovered.getAttempts()); + assertEquals(List.of(recovered), jobRepository.findJobsByQuestionId(stale.getQuestionId())); + } finally { + start.countDown(); + jobRepository.deleteJobById(stale.getId()); + } + } + + private Job recoveryJob(final JobStatus status, final Optional processedAt) { + Job job = Job.builder() + .questionId(UUID.randomUUID().toString()) + .status(status) + .build(); + assertTrue(jobRepository.createJob(job)); + job.setAttempts(3); + job.setProcessedAt(processedAt); + job.setCompletedAt(Optional.of(StandardizedOffsetDateTime.now().minusMinutes(30))); + job.setNextAttemptAt(StandardizedOffsetDateTime.now().plusHours(2)); + assertTrue(jobRepository.updateJob(job)); + return jobRepository.findJobById(job.getId()).orElseThrow(); + } }