diff --git a/docs/queue-visibility-timeout-tuning.md b/docs/queue-visibility-timeout-tuning.md new file mode 100644 index 0000000..19df96d --- /dev/null +++ b/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. diff --git a/docs/video-clipping-service-implementation-plan.md b/docs/video-clipping-service-implementation-plan.md index 31b000d..1b3ab9a 100644 --- a/docs/video-clipping-service-implementation-plan.md +++ b/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. diff --git a/src/main/java/org/example/videoclips/config/VideoClippingProperties.java b/src/main/java/org/example/videoclips/config/VideoClippingProperties.java index 0b0c980..82ba8e1 100644 --- a/src/main/java/org/example/videoclips/config/VideoClippingProperties.java +++ b/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; } diff --git a/src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java b/src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java index 0cf5b23..ec6960c 100644 --- a/src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java +++ b/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 messages = queueMessageJpaRepository - .findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", now); + List messages = claimableMessages(now, remaining); for (QueueMessageEntity message : messages) { if (remaining-- <= 0) { break; } - 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())); - } + claimMessage(message, now); 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 claimableMessages(Instant now, int batchSize) { + List messages = new ArrayList<>( + queueMessageJpaRepository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", now) + ); + if (messages.size() >= batchSize) { + return messages; + } + Instant expiredClaimCutoff = now.minusMillis(properties.getDatabaseQueue().getVisibilityTimeoutMs()); + List 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())); + } + } } diff --git a/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java b/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java index e292e54..f53477d 100644 --- a/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java +++ b/src/main/java/org/example/videoclips/queue/spring/QueueMessageJpaRepository.java @@ -8,4 +8,6 @@ import java.util.List; public interface QueueMessageJpaRepository extends JpaRepository { List findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(String status, java.time.Instant availableAt); + + List findTop10ByStatusAndClaimedAtLessThanEqualOrderByClaimedAtAsc(String status, java.time.Instant claimedAt); } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index ee5bb83..750e1fd 100644 --- a/src/main/resources/application.properties +++ b/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 diff --git a/src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java b/src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java index 9750b48..3a7acab 100644 --- a/src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java +++ b/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); + } }