fix(reliability): keep recovery retry time on one clock

src/main/java/com/clearfolio/viewer/service/DefaultConversionWorker.java

@@ -186,7 +186,7 @@ public int recoverPendingJobs(Instant now, Duration processingLeaseTimeout) {
         List<ConversionJob> recoverableJobs = repository.findRecoverableJobs(now, staleProcessingBefore);
         recoverableJobs.forEach(job -> {
             if (job.getStatus() == ConversionJobStatus.PROCESSING) {
-                stateStore.scheduleRetry(job.getJobId(), "worker lease expired; retry queued", Instant.now());
+                stateStore.scheduleRetry(job.getJobId(), "worker lease expired; retry queued", now);
             }
             enqueue(job.getJobId());
         });

src/test/java/com/clearfolio/viewer/service/DefaultConversionWorkerRecoveryClockTest.java

@@ -0,0 +1,91 @@
+package com.clearfolio.viewer.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.junit.jupiter.api.Test;
+
+import com.clearfolio.viewer.artifact.InMemoryArtifactStore;
+import com.clearfolio.viewer.artifact.PdfBoxArtifactGenerator;
+import com.clearfolio.viewer.config.ConversionProperties;
+import com.clearfolio.viewer.model.ConversionJob;
+import com.clearfolio.viewer.repository.ConversionJobStateStore;
+import com.clearfolio.viewer.repository.InMemoryConversionJobRepository;
+
+/**
+ * Verifies that startup recovery uses one caller-supplied clock observation.
+ */
+class DefaultConversionWorkerRecoveryClockTest {
+
+    @Test
+    void staleProcessingRetryUsesTheRecoveryEvaluationTimestamp() {
+        InMemoryConversionJobRepository repository = new InMemoryConversionJobRepository();
+        CapturingStateStore stateStore = new CapturingStateStore(repository);
+        ConversionJob staleProcessing = new ConversionJob(
+                UUID.randomUUID(),
+                "stale.docx",
+                "application/octet-stream",
+                "hash-recovery-clock",
+                10L,
+                3
+        );
+        assertTrue(staleProcessing.markProcessing("worker exited"));
+        repository.save(staleProcessing);
+
+        Instant recoveryNow = Instant.now().plus(Duration.ofDays(1));
+        DefaultConversionWorker worker = new DefaultConversionWorker(
+                repository,
+                stateStore,
+                command -> { },
+                new InMemoryArtifactStore(),
+                new PdfBoxArtifactGenerator(),
+                new ConversionProperties(),
+                id -> "/artifacts/" + id + ".pdf"
+        );
+
+        int recovered = worker.recoverPendingJobs(recoveryNow, Duration.ofSeconds(60));
+
+        assertEquals(1, recovered);
+        assertEquals(recoveryNow, stateStore.retryAt);
+    }
+
+    private static final class CapturingStateStore implements ConversionJobStateStore {
+        private final ConversionJobStateStore delegate;
+        private Instant retryAt;
+
+        private CapturingStateStore(ConversionJobStateStore delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public Optional<ConversionJob> claimForProcessing(UUID jobId, Instant now) {
+            return delegate.claimForProcessing(jobId, now);
+        }
+
+        @Override
+        public void scheduleRetry(UUID jobId, String message, Instant retryAt) {
+            this.retryAt = retryAt;
+            delegate.scheduleRetry(jobId, message, retryAt);
+        }
+
+        @Override
+        public void markSucceeded(UUID jobId, String resourcePath, String message) {
+            delegate.markSucceeded(jobId, resourcePath, message);
+        }
+
+        @Override
+        public void markDeadLettered(UUID jobId, String message) {
+            delegate.markDeadLettered(jobId, message);
+        }
+
+        @Override
+        public boolean retryDeadLettered(UUID jobId, String operatorId) {
+            return delegate.retryDeadLettered(jobId, operatorId);
+        }
+    }
+}

src/test/java/com/clearfolio/viewer/service/DefaultConversionWorkerTest.java

@@ -607,7 +607,6 @@ void scheduleRetryImmediatelyRequeuesWhenRetryTimeAlreadyPassed() throws Excepti
     void recoverPendingJobsRequeuesDueSubmittedAndStaleProcessingJobs() {
         InMemoryConversionJobRepository repository = new InMemoryConversionJobRepository();
         ConversionProperties conversionProperties = new ConversionProperties();
-        Instant recoveryNow = Instant.now().plusSeconds(120);
 
         ConversionJob dueSubmitted = new ConversionJob(
                 UUID.randomUUID(),
@@ -625,7 +624,6 @@ void recoverPendingJobsRequeuesDueSubmittedAndStaleProcessingJobs() {
                 10L,
                 3
         );
-        futureRetry.markRetryScheduled("retry later", recoveryNow.plusSeconds(30));
         ConversionJob staleProcessing = new ConversionJob(
                 UUID.randomUUID(),
                 "stale.docx",
@@ -635,6 +633,8 @@ void recoverPendingJobsRequeuesDueSubmittedAndStaleProcessingJobs() {
                 3
         );
         assertTrue(staleProcessing.markProcessing("worker exited"));
+        Instant recoveryNow = staleProcessing.getStartedAt().plusNanos(1);
+        futureRetry.markRetryScheduled("retry later", recoveryNow.plusSeconds(30));
         repository.save(dueSubmitted);
         repository.save(futureRetry);
         repository.save(staleProcessing);
@@ -652,7 +652,7 @@ void recoverPendingJobsRequeuesDueSubmittedAndStaleProcessingJobs() {
                 }
         );
 
-        int recovered = worker.recoverPendingJobs(recoveryNow, Duration.ofSeconds(60));
+        int recovered = worker.recoverPendingJobs(recoveryNow, Duration.ZERO);
 
         assertEquals(2, recovered);
         assertEquals(2, attempts.get());