• I completed the next plan item and updated the checklist. Queue visibility-timeout tuning is now marked done in docs/video-clipping-service-implementation-plan.md:716.

The main change is in src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java:52: the DB queue now reclaims stale PROCESSING messages after a configurable visibility timeout instead of leaving them
  stuck forever after a worker crash. I added video-clipping.database-queue.visibility-timeout-ms with a default of 900000 ms in src/main/java/org/example/videoclips/config/VideoClippingProperties.java:171 and src/main/
  resources/application.properties:19, plus coverage for stale-claim recovery in src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java:91. I also added the tuning note and sizing rule in docs/
  queue-visibility-timeout-tuning.md:1.

docs/queue-visibility-timeout-tuning.md

@@ -0,0 +1,32 @@
+# Queue Visibility Timeout Tuning
+
+This document captures the implementation and sizing guidance for performance plan item `Queue visibility-timeout tuning`.
+
+Implemented behavior:
+
+- Database-backed queue messages now use a configurable visibility timeout via `video-clipping.database-queue.visibility-timeout-ms`.
+- The default is `900000` ms (`15` minutes).
+- Messages left in `PROCESSING` past the visibility timeout are reclaimed on the next poll and retried.
+- This prevents permanently stuck jobs after worker crashes or process termination between claim and completion.
+
+Sizing rule:
+
+- Start with `visibility-timeout-ms = 3x` the observed `p99` end-to-end job runtime for the dominant workload.
+- Include source-object materialization, FFmpeg runtime, clip upload, and persistence overhead.
+- Keep `retry-backoff-ms` materially smaller than the visibility timeout so normal retries happen before stale-claim recovery becomes the dominant path.
+
+Current baseline rationale:
+
+- The checked-in FFmpeg and object-storage benchmarks show local workloads completing in seconds, not minutes.
+- The database queue does not extend visibility during active processing, so the default is intentionally conservative.
+- `15` minutes gives room for slower `EXACT` jobs, degraded object-storage bandwidth, and moderate node contention without causing premature duplicate delivery.
+
+How to validate:
+
+```bash
+mvn -q -Dtest=DatabaseBackedClipJobQueueAdapterTest test
+```
+
+Operational note:
+
+- If production measurements show long-running jobs regularly exceeding the configured timeout, either raise `visibility-timeout-ms` or implement active lease extension before increasing worker concurrency.

docs/video-clipping-service-implementation-plan.md

@@ -713,7 +713,7 @@ Keep the domain independent from Spring framework details. Adapters implement st
 1. [x] Worker CPU/disk benchmarks.
 2. [x] FFmpeg preset benchmarks.
 3. [x] Object-storage bandwidth tests.
-4. [ ] Queue visibility-timeout tuning.
+4. [x] Queue visibility-timeout tuning.
 5. [ ] Autoscaling policies.
 6. [ ] Cost model by video minute.
 

src/main/java/org/example/videoclips/config/VideoClippingProperties.java

@@ -170,6 +170,9 @@ public class VideoClippingProperties {
         @Min(1)
         private int batchSize = 5;
 
+        @Min(1000)
+        private long visibilityTimeoutMs = 900000;
+
         @Min(1)
         private int maxAttempts = 3;
 
@@ -192,6 +195,14 @@ public class VideoClippingProperties {
             this.batchSize = batchSize;
         }
 
+        public long getVisibilityTimeoutMs() {
+            return visibilityTimeoutMs;
+        }
+
+        public void setVisibilityTimeoutMs(long visibilityTimeoutMs) {
+            this.visibilityTimeoutMs = visibilityTimeoutMs;
+        }
+
         public int getMaxAttempts() {
             return maxAttempts;
         }

src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 
@@ -53,24 +54,17 @@ public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
     public void poll() {
         int remaining = properties.getDatabaseQueue().getBatchSize();
         Instant now = Instant.now();
-        List<QueueMessageEntity> messages = queueMessageJpaRepository
+        List<QueueMessageEntity> messages = claimableMessages(now, remaining);
-                .findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", now);
         for (QueueMessageEntity message : messages) {
             if (remaining-- <= 0) {
                 break;
             }
-            message.setStatus("PROCESSING");
+            claimMessage(message, now);
-            message.setClaimedAt(now);
-            message.setAttemptCount(message.getAttemptCount() + 1);
-            queueMessageJpaRepository.save(message);
-            ClipJob job = repository.findJobById(message.getJobId());
-            if (job != null) {
-                repository.saveJob(job.withAttemptCount(message.getAttemptCount()));
-            }
             try {
                 clipProcessor.processNow(message.getJobId(), false);
                 message.setStatus("COMPLETED");
                 message.setLastError(null);
+                message.setClaimedAt(now);
             } catch (Exception ex) {
                 message.setLastError(ex.getMessage());
                 if (message.getAttemptCount() >= properties.getDatabaseQueue().getMaxAttempts()) {
@@ -86,4 +80,36 @@ public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
             queueMessageJpaRepository.save(message);
         }
     }
+
+    private List<QueueMessageEntity> claimableMessages(Instant now, int batchSize) {
+        List<QueueMessageEntity> messages = new ArrayList<>(
+                queueMessageJpaRepository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", now)
+        );
+        if (messages.size() >= batchSize) {
+            return messages;
+        }
+        Instant expiredClaimCutoff = now.minusMillis(properties.getDatabaseQueue().getVisibilityTimeoutMs());
+        List<QueueMessageEntity> expiredClaims = queueMessageJpaRepository
+                .findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc("PROCESSING", expiredClaimCutoff);
+        for (QueueMessageEntity expiredClaim : expiredClaims) {
+            if (messages.stream().noneMatch(message -> message.getId().equals(expiredClaim.getId()))) {
+                messages.add(expiredClaim);
+            }
+            if (messages.size() >= batchSize) {
+                break;
+            }
+        }
+        return messages;
+    }
+
+    private void claimMessage(QueueMessageEntity message, Instant now) {
+        message.setStatus("PROCESSING");
+        message.setClaimedAt(now);
+        message.setAttemptCount(message.getAttemptCount() + 1);
+        queueMessageJpaRepository.save(message);
+        ClipJob job = repository.findJobById(message.getJobId());
+        if (job != null) {
+            repository.saveJob(job.withAttemptCount(message.getAttemptCount()));
+        }
+    }
 }

src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java

@@ -8,4 +8,6 @@ import java.util.List;
 public interface QueueMessageJpaRepository extends JpaRepository<QueueMessageEntity, String> {
 
     List<QueueMessageEntity> findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(String status, java.time.Instant availableAt);
+
+    List<QueueMessageEntity> findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc(String status, java.time.Instant claimedAt);
 }

src/main/resources/application.properties

@@ -18,6 +18,7 @@ video-clipping.s3.bucket=video-clipping-dev
 video-clipping.s3.path-style=true
 video-clipping.database-queue.poll-interval-ms=1000
 video-clipping.database-queue.batch-size=5
+video-clipping.database-queue.visibility-timeout-ms=900000
 video-clipping.database-queue.max-attempts=3
 video-clipping.database-queue.retry-backoff-ms=5000
 video-clipping.ffmpeg.ffmpeg-binary=ffmpeg

src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java

@@ -39,6 +39,8 @@ class DatabaseBackedClipJobQueueAdapterTest {
 
         when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
                 .thenReturn(List.of(firstAttempt));
+        when(repository.findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc(any(), any()))
+                .thenReturn(List.of());
         doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_1", false);
 
         DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, videoClippingRepository, clipProcessor, properties);
@@ -71,6 +73,8 @@ class DatabaseBackedClipJobQueueAdapterTest {
 
         when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
                 .thenReturn(List.of(finalAttempt));
+        when(repository.findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc(any(), any()))
+                .thenReturn(List.of());
         doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_2", false);
 
         DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, videoClippingRepository, clipProcessor, properties);
@@ -83,4 +87,36 @@ class DatabaseBackedClipJobQueueAdapterTest {
         verify(repository, times(2)).save(finalAttempt);
         verify(clipProcessor).markTerminalFailure("job_2", 3, "processing failed");
     }
+
+    @Test
+    void reclaimsExpiredProcessingMessagesAfterVisibilityTimeout() {
+        QueueMessageJpaRepository repository = mock(QueueMessageJpaRepository.class);
+        VideoClippingRepository videoClippingRepository = mock(VideoClippingRepository.class);
+        ClipProcessor clipProcessor = mock(ClipProcessor.class);
+        VideoClippingProperties properties = new VideoClippingProperties();
+        properties.getDatabaseQueue().setVisibilityTimeoutMs(5000);
+
+        QueueMessageEntity expiredClaim = new QueueMessageEntity();
+        expiredClaim.setId("msg_3");
+        expiredClaim.setJobId("job_3");
+        expiredClaim.setStatus("PROCESSING");
+        expiredClaim.setAttemptCount(1);
+        expiredClaim.setAvailableAt(Instant.now().minusSeconds(30));
+        expiredClaim.setClaimedAt(Instant.now().minusSeconds(10));
+
+        when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
+                .thenReturn(List.of());
+        when(repository.findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc(any(), any()))
+                .thenReturn(List.of(expiredClaim));
+
+        DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, videoClippingRepository, clipProcessor, properties);
+
+        adapter.poll();
+
+        assertEquals("COMPLETED", expiredClaim.getStatus());
+        assertEquals(2, expiredClaim.getAttemptCount());
+        assertEquals(null, expiredClaim.getLastError());
+        verify(clipProcessor).processNow("job_3", false);
+        verify(repository, times(2)).save(expiredClaim);
+    }
 }