DINO-1093: update the requeuing logic to also consider problems who h…

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

js/src/app/admin/_components/incomplete/RequeueIncompleteQuestionsModal.tsx

@@ -34,8 +34,10 @@ export default function RequeueIncompleteQuestionsModal() {
       >
         <form onSubmit={onSubmit}>
           <Text mt={12}>
-            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.
           </Text>
           <Text mt={12}>
             Each job calls LeetCode. If jobs have been failing because the auth

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

src/main/java/org/patinanetwork/codebloom/api/admin/AdminController.java

@@ -263,8 +263,9 @@ public ResponseEntity<ApiResponder<List<QuestionWithUserDto>>> 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<ApiResponder<Empty>> requeueIncompleteQuestions(

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.
      *
-     * <p>Safe to call concurrently: the active-job unique index rejects duplicates, which are counted as skipped.
+     * <p>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<Question> questions = questionRepository.getAllIncompleteQuestionsWithNoJob();
+        int requeuedCount = jobRepository.requeueStaleProcessingJobs(
+                StandardizedOffsetDateTime.now().minusHours(1));
 
-        if (questions.isEmpty()) {
-            log.info("All questions have complete information.");
+        ArrayList<Question> 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;
     }
 }

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

src/main/java/org/patinanetwork/codebloom/common/db/repos/job/JobSqlRepository.java

@@ -147,6 +147,24 @@ public List<Job> 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 = """

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