diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/config/JobOutboxProperties.java b/backend-java/src/main/java/com/fengting/aigcforensics/config/JobOutboxProperties.java new file mode 100644 index 0000000..31637c1 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/config/JobOutboxProperties.java @@ -0,0 +1,64 @@ +package com.fengting.aigcforensics.config; + +import java.time.Duration; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "app.jobs.outbox") +public class JobOutboxProperties { + + private boolean dispatcherEnabled = true; + private long pollDelayMs = 500; + private Duration staleAfter = Duration.ofMinutes(1); + private int maxAttempts = 5; + private Duration baseRetryDelay = Duration.ofSeconds(1); + private Duration maxRetryDelay = Duration.ofMinutes(1); + + public boolean isDispatcherEnabled() { + return dispatcherEnabled; + } + + public void setDispatcherEnabled(boolean dispatcherEnabled) { + this.dispatcherEnabled = dispatcherEnabled; + } + + public long getPollDelayMs() { + return pollDelayMs; + } + + public void setPollDelayMs(long pollDelayMs) { + this.pollDelayMs = pollDelayMs; + } + + public Duration getStaleAfter() { + return staleAfter; + } + + public void setStaleAfter(Duration staleAfter) { + this.staleAfter = staleAfter; + } + + public int getMaxAttempts() { + return maxAttempts; + } + + public void setMaxAttempts(int maxAttempts) { + this.maxAttempts = maxAttempts; + } + + public Duration getBaseRetryDelay() { + return baseRetryDelay; + } + + public void setBaseRetryDelay(Duration baseRetryDelay) { + this.baseRetryDelay = baseRetryDelay; + } + + public Duration getMaxRetryDelay() { + return maxRetryDelay; + } + + public void setMaxRetryDelay(Duration maxRetryDelay) { + this.maxRetryDelay = maxRetryDelay; + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/controller/ApiExceptionHandler.java b/backend-java/src/main/java/com/fengting/aigcforensics/controller/ApiExceptionHandler.java index ca9ba13..c126186 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/controller/ApiExceptionHandler.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/controller/ApiExceptionHandler.java @@ -9,6 +9,7 @@ import org.springframework.web.multipart.support.MissingServletRequestPartException; import com.fengting.aigcforensics.dto.error.ErrorResponse; +import com.fengting.aigcforensics.domain.InvalidJobOutboxStateException; import com.fengting.aigcforensics.service.ResourceNotFoundException; @RestControllerAdvice @@ -20,6 +21,12 @@ public ErrorResponse handleBadRequest(IllegalArgumentException exception) { return new ErrorResponse(exception.getMessage(), Instant.now()); } + @ExceptionHandler(InvalidJobOutboxStateException.class) + @ResponseStatus(HttpStatus.CONFLICT) + public ErrorResponse handleConflict(InvalidJobOutboxStateException exception) { + return new ErrorResponse(exception.getMessage(), Instant.now()); + } + @ExceptionHandler(MissingServletRequestPartException.class) @ResponseStatus(HttpStatus.BAD_REQUEST) public ErrorResponse handleMissingPart(MissingServletRequestPartException exception) { diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/controller/JobOutboxOperationsController.java b/backend-java/src/main/java/com/fengting/aigcforensics/controller/JobOutboxOperationsController.java new file mode 100644 index 0000000..8be434a --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/controller/JobOutboxOperationsController.java @@ -0,0 +1,65 @@ +package com.fengting.aigcforensics.controller; + +import java.util.List; + +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import com.fengting.aigcforensics.domain.JobOutboxEvent; +import com.fengting.aigcforensics.domain.JobOutboxStatus; +import com.fengting.aigcforensics.dto.operations.JobOutboxEventResponse; +import com.fengting.aigcforensics.service.JobOutboxService; + +@RestController +@RequestMapping("/api/operations/job-outbox") +public class JobOutboxOperationsController { + + private static final int DEFAULT_LIMIT = 50; + private static final int MAX_LIMIT = 100; + + private final JobOutboxService outboxService; + + public JobOutboxOperationsController(JobOutboxService outboxService) { + this.outboxService = outboxService; + } + + @GetMapping + public List list( + @RequestParam(defaultValue = "FAILED") JobOutboxStatus status, + @RequestParam(defaultValue = "50") int limit) { + validateLimit(limit); + return outboxService.list(status, limit).stream() + .map(this::toResponse) + .toList(); + } + + @PostMapping("/{eventId}/replay") + public JobOutboxEventResponse replay(@PathVariable String eventId) { + return toResponse(outboxService.replay(eventId)); + } + + private void validateLimit(int limit) { + if (limit < 1 || limit > MAX_LIMIT) { + throw new IllegalArgumentException("limit must be between 1 and 100"); + } + } + + private JobOutboxEventResponse toResponse(JobOutboxEvent event) { + return new JobOutboxEventResponse( + event.getEventId(), + event.getEventType(), + event.getAggregateType(), + event.getAggregateId(), + event.getStatus(), + event.getAttemptCount(), + event.getAvailableAt(), + event.getLastError(), + event.getCreatedAt(), + event.getUpdatedAt(), + event.getPublishedAt()); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/domain/InvalidJobOutboxStateException.java b/backend-java/src/main/java/com/fengting/aigcforensics/domain/InvalidJobOutboxStateException.java new file mode 100644 index 0000000..bcc9425 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/domain/InvalidJobOutboxStateException.java @@ -0,0 +1,8 @@ +package com.fengting.aigcforensics.domain; + +public class InvalidJobOutboxStateException extends IllegalStateException { + + public InvalidJobOutboxStateException(String message) { + super(message); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEvent.java b/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEvent.java new file mode 100644 index 0000000..862181f --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEvent.java @@ -0,0 +1,234 @@ +package com.fengting.aigcforensics.domain; + +import java.time.Instant; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import jakarta.persistence.Version; + +@Entity +@Table(name = "job_outbox") +public class JobOutboxEvent { + + private static final int MAX_ERROR_LENGTH = 2048; + private static final String DETECTION_AGGREGATE_TYPE = "DETECTION_TASK"; + + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + + @Column(name = "event_id", nullable = false, unique = true, length = 64) + private String eventId; + + @Enumerated(EnumType.STRING) + @Column(name = "event_type", nullable = false, length = 64) + private JobOutboxEventType eventType; + + @Column(name = "aggregate_type", nullable = false, length = 64) + private String aggregateType; + + @Column(name = "aggregate_id", nullable = false, length = 64) + private String aggregateId; + + @Column(name = "payload_json", nullable = false, columnDefinition = "text") + private String payloadJson; + + @Enumerated(EnumType.STRING) + @Column(name = "status", nullable = false, length = 32) + private JobOutboxStatus status; + + @Column(name = "attempt_count", nullable = false) + private int attemptCount; + + @Column(name = "available_at") + private Instant availableAt; + + @Column(name = "last_error", length = MAX_ERROR_LENGTH) + private String lastError; + + @Column(name = "created_at", nullable = false) + private Instant createdAt; + + @Column(name = "updated_at", nullable = false) + private Instant updatedAt; + + @Column(name = "published_at") + private Instant publishedAt; + + @Version + @Column(name = "lock_version", nullable = false) + private long lockVersion; + + protected JobOutboxEvent() { + } + + private JobOutboxEvent( + String eventId, + JobOutboxEventType eventType, + String aggregateType, + String aggregateId, + String payloadJson, + Instant now) { + this.eventId = requireText(eventId, "eventId"); + this.eventType = eventType; + this.aggregateType = requireText(aggregateType, "aggregateType"); + this.aggregateId = requireText(aggregateId, "aggregateId"); + this.payloadJson = requireText(payloadJson, "payloadJson"); + this.status = JobOutboxStatus.PENDING; + this.availableAt = now; + this.createdAt = now; + this.updatedAt = now; + } + + public static JobOutboxEvent detectionRequested( + String eventId, + String taskId, + String payloadJson, + Instant now) { + return new JobOutboxEvent( + eventId, + JobOutboxEventType.DETECTION_REQUESTED, + DETECTION_AGGREGATE_TYPE, + taskId, + payloadJson, + now); + } + + public void claim(Instant claimedAt) { + requireStatus(JobOutboxStatus.PENDING, "claim"); + if (availableAt == null || availableAt.isAfter(claimedAt)) { + throw new InvalidJobOutboxStateException( + "Outbox event is not available for publishing: " + eventId); + } + status = JobOutboxStatus.PUBLISHING; + attemptCount++; + updatedAt = claimedAt; + } + + public void markPublished(Instant completedAt) { + requireStatus(JobOutboxStatus.PUBLISHING, "mark published"); + status = JobOutboxStatus.PUBLISHED; + publishedAt = completedAt; + availableAt = null; + lastError = null; + updatedAt = completedAt; + } + + public void markPublishFailed(String error, Instant retryAt, int maxAttempts, Instant failedAt) { + requireStatus(JobOutboxStatus.PUBLISHING, "mark publish failed"); + if (maxAttempts < 1) { + throw new IllegalArgumentException("maxAttempts must be at least 1"); + } + lastError = truncateError(error); + updatedAt = failedAt; + if (attemptCount >= maxAttempts) { + status = JobOutboxStatus.FAILED; + availableAt = null; + return; + } + status = JobOutboxStatus.PENDING; + availableAt = retryAt; + } + + public void recoverStaleClaim(Instant recoveredAt) { + requireStatus(JobOutboxStatus.PUBLISHING, "recover stale claim"); + status = JobOutboxStatus.PENDING; + availableAt = recoveredAt; + lastError = "Recovered stale publishing claim"; + updatedAt = recoveredAt; + } + + public void replay(Instant replayedAt) { + if (status != JobOutboxStatus.PUBLISHED && status != JobOutboxStatus.FAILED) { + throw new InvalidJobOutboxStateException( + "Only terminal outbox events can be replayed: " + eventId); + } + status = JobOutboxStatus.PENDING; + attemptCount = 0; + availableAt = replayedAt; + lastError = null; + publishedAt = null; + updatedAt = replayedAt; + } + + private void requireStatus(JobOutboxStatus expected, String action) { + if (status != expected) { + throw new InvalidJobOutboxStateException( + "Cannot " + action + " outbox event " + eventId + " while status is " + status); + } + } + + private static String requireText(String value, String fieldName) { + if (value == null || value.isBlank()) { + throw new IllegalArgumentException(fieldName + " must not be blank"); + } + return value; + } + + private String truncateError(String error) { + String message = error == null || error.isBlank() ? "Unknown publish failure" : error; + return message.length() <= MAX_ERROR_LENGTH ? message : message.substring(0, MAX_ERROR_LENGTH); + } + + public Long getId() { + return id; + } + + public String getEventId() { + return eventId; + } + + public JobOutboxEventType getEventType() { + return eventType; + } + + public String getAggregateType() { + return aggregateType; + } + + public String getAggregateId() { + return aggregateId; + } + + public String getPayloadJson() { + return payloadJson; + } + + public JobOutboxStatus getStatus() { + return status; + } + + public int getAttemptCount() { + return attemptCount; + } + + public Instant getAvailableAt() { + return availableAt; + } + + public String getLastError() { + return lastError; + } + + public Instant getCreatedAt() { + return createdAt; + } + + public Instant getUpdatedAt() { + return updatedAt; + } + + public Instant getPublishedAt() { + return publishedAt; + } + + public long getLockVersion() { + return lockVersion; + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEventType.java b/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEventType.java new file mode 100644 index 0000000..357a99e --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEventType.java @@ -0,0 +1,5 @@ +package com.fengting.aigcforensics.domain; + +public enum JobOutboxEventType { + DETECTION_REQUESTED +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxStatus.java b/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxStatus.java new file mode 100644 index 0000000..ea2fd8a --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxStatus.java @@ -0,0 +1,8 @@ +package com.fengting.aigcforensics.domain; + +public enum JobOutboxStatus { + PENDING, + PUBLISHING, + PUBLISHED, + FAILED +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/dto/operations/JobOutboxEventResponse.java b/backend-java/src/main/java/com/fengting/aigcforensics/dto/operations/JobOutboxEventResponse.java new file mode 100644 index 0000000..7c7b03c --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/dto/operations/JobOutboxEventResponse.java @@ -0,0 +1,20 @@ +package com.fengting.aigcforensics.dto.operations; + +import java.time.Instant; + +import com.fengting.aigcforensics.domain.JobOutboxEventType; +import com.fengting.aigcforensics.domain.JobOutboxStatus; + +public record JobOutboxEventResponse( + String eventId, + JobOutboxEventType eventType, + String aggregateType, + String aggregateId, + JobOutboxStatus status, + int attemptCount, + Instant availableAt, + String lastError, + Instant createdAt, + Instant updatedAt, + Instant publishedAt) { +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/repository/DetectionTaskRepository.java b/backend-java/src/main/java/com/fengting/aigcforensics/repository/DetectionTaskRepository.java index 369c488..6052e04 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/repository/DetectionTaskRepository.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/repository/DetectionTaskRepository.java @@ -4,12 +4,21 @@ import java.util.Optional; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Lock; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; import com.fengting.aigcforensics.domain.DetectionTask; +import jakarta.persistence.LockModeType; + public interface DetectionTaskRepository extends JpaRepository { Optional findByTaskId(String taskId); + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query("select task from DetectionTask task where task.taskId = :taskId") + Optional findByTaskIdForUpdate(@Param("taskId") String taskId); + List findAllByOrderByCreatedAtDesc(); } diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/repository/JobOutboxEventRepository.java b/backend-java/src/main/java/com/fengting/aigcforensics/repository/JobOutboxEventRepository.java new file mode 100644 index 0000000..8dbaeca --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/repository/JobOutboxEventRepository.java @@ -0,0 +1,56 @@ +package com.fengting.aigcforensics.repository; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; + +import org.springframework.data.domain.Pageable; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Lock; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; + +import com.fengting.aigcforensics.domain.JobOutboxEvent; +import com.fengting.aigcforensics.domain.JobOutboxEventType; +import com.fengting.aigcforensics.domain.JobOutboxStatus; + +import jakarta.persistence.LockModeType; + +public interface JobOutboxEventRepository extends JpaRepository { + + Optional findByEventId(String eventId); + + Optional findByEventTypeAndAggregateId( + JobOutboxEventType eventType, + String aggregateId); + + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query("select event from JobOutboxEvent event where event.eventId = :eventId") + Optional findByEventIdForUpdate(@Param("eventId") String eventId); + + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query(""" + select event from JobOutboxEvent event + where event.status = :status and event.availableAt <= :now + order by event.createdAt asc + """) + List findClaimable( + @Param("status") JobOutboxStatus status, + @Param("now") Instant now, + Pageable pageable); + + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query(""" + select event from JobOutboxEvent event + where event.status = :status and event.updatedAt <= :staleBefore + order by event.updatedAt asc + """) + List findStalePublishing( + @Param("status") JobOutboxStatus status, + @Param("staleBefore") Instant staleBefore, + Pageable pageable); + + List findByStatusOrderByCreatedAtDesc( + JobOutboxStatus status, + Pageable pageable); +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobMessage.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobMessage.java index 4c83bf1..189c9cc 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobMessage.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobMessage.java @@ -1,4 +1,8 @@ package com.fengting.aigcforensics.service; -public record DetectionJobMessage(String messageId, String taskId) { +public record DetectionJobMessage( + String messageId, + String eventId, + int eventVersion, + String taskId) { } diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobQueue.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobQueue.java index 552b7a5..5d35d07 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobQueue.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobQueue.java @@ -2,5 +2,5 @@ public interface DetectionJobQueue { - void enqueue(String taskId); + void enqueue(DetectionJobRequest request); } diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobRequest.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobRequest.java new file mode 100644 index 0000000..8b825ae --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobRequest.java @@ -0,0 +1,25 @@ +package com.fengting.aigcforensics.service; + +import java.time.Instant; + +public record DetectionJobRequest( + String eventId, + int eventVersion, + String taskId, + Instant occurredAt) { + + public DetectionJobRequest { + if (eventId == null || eventId.isBlank()) { + throw new IllegalArgumentException("eventId must not be blank"); + } + if (eventVersion < 1) { + throw new IllegalArgumentException("eventVersion must be at least 1"); + } + if (taskId == null || taskId.isBlank()) { + throw new IllegalArgumentException("taskId must not be blank"); + } + if (occurredAt == null) { + throw new IllegalArgumentException("occurredAt must not be null"); + } + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobService.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobService.java index 1fb5a7e..088f5af 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobService.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobService.java @@ -11,23 +11,24 @@ public class DetectionJobService { private final DetectionTaskRepository detectionTaskRepository; - private final DetectionJobQueue detectionJobQueue; + private final JobOutboxService jobOutboxService; public DetectionJobService( DetectionTaskRepository detectionTaskRepository, - DetectionJobQueue detectionJobQueue) { + JobOutboxService jobOutboxService) { this.detectionTaskRepository = detectionTaskRepository; - this.detectionJobQueue = detectionJobQueue; + this.jobOutboxService = jobOutboxService; } - @Transactional(readOnly = true) + @Transactional public DetectionTask submit(String taskId) { - DetectionTask task = detectionTaskRepository.findByTaskId(taskId) + DetectionTask task = detectionTaskRepository.findByTaskIdForUpdate(taskId) .orElseThrow(() -> new ResourceNotFoundException("Detection task not found: " + taskId)); - if (task.getStatus() != DetectionStatus.QUEUED && task.getStatus() != DetectionStatus.FAILED) { - return task; + if (task.getStatus() == DetectionStatus.QUEUED) { + jobOutboxService.scheduleDetection(taskId); + } else if (task.getStatus() == DetectionStatus.FAILED) { + jobOutboxService.replayDetection(taskId); } - detectionJobQueue.enqueue(taskId); return task; } } diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxDispatcher.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxDispatcher.java new file mode 100644 index 0000000..fa218bb --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxDispatcher.java @@ -0,0 +1,45 @@ +package com.fengting.aigcforensics.service; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import com.fengting.aigcforensics.domain.JobOutboxEvent; + +@Service +@ConditionalOnProperty( + prefix = "app.jobs.outbox", + name = "dispatcher-enabled", + havingValue = "true", + matchIfMissing = true) +public class JobOutboxDispatcher { + + private final JobOutboxService outboxService; + private final JobOutboxPublisher publisher; + + public JobOutboxDispatcher(JobOutboxService outboxService, JobOutboxPublisher publisher) { + this.outboxService = outboxService; + this.publisher = publisher; + } + + @Scheduled(fixedDelayString = "${app.jobs.outbox.poll-delay-ms:500}") + public void pollOnce() { + outboxService.claimNext().ifPresent(this::publish); + } + + private void publish(JobOutboxEvent event) { + try { + publisher.publish(event); + outboxService.markPublished(event.getEventId()); + } catch (RuntimeException exception) { + outboxService.markPublishFailed(event.getEventId(), failureMessage(exception)); + } + } + + private String failureMessage(RuntimeException exception) { + if (exception.getMessage() == null || exception.getMessage().isBlank()) { + return exception.getClass().getSimpleName(); + } + return exception.getMessage(); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxPublisher.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxPublisher.java new file mode 100644 index 0000000..69dfd46 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxPublisher.java @@ -0,0 +1,49 @@ +package com.fengting.aigcforensics.service; + +import org.springframework.stereotype.Service; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fengting.aigcforensics.domain.JobOutboxEvent; +import com.fengting.aigcforensics.domain.JobOutboxEventType; + +@Service +public class JobOutboxPublisher { + + private static final int DETECTION_EVENT_VERSION = 1; + + private final DetectionJobQueue detectionJobQueue; + private final ObjectMapper objectMapper; + + public JobOutboxPublisher(DetectionJobQueue detectionJobQueue, ObjectMapper objectMapper) { + this.detectionJobQueue = detectionJobQueue; + this.objectMapper = objectMapper; + } + + public void publish(JobOutboxEvent event) { + if (event.getEventType() != JobOutboxEventType.DETECTION_REQUESTED) { + throw new IllegalArgumentException("Unsupported outbox event type: " + event.getEventType()); + } + DetectionRequestedPayload payload = readDetectionPayload(event.getPayloadJson()); + if (!event.getAggregateId().equals(payload.taskId())) { + throw new IllegalArgumentException( + "Outbox payload taskId does not match aggregate id: " + event.getEventId()); + } + detectionJobQueue.enqueue(new DetectionJobRequest( + event.getEventId(), + DETECTION_EVENT_VERSION, + payload.taskId(), + event.getCreatedAt())); + } + + private DetectionRequestedPayload readDetectionPayload(String payloadJson) { + try { + return objectMapper.readValue(payloadJson, DetectionRequestedPayload.class); + } catch (JsonProcessingException exception) { + throw new IllegalArgumentException("Invalid detection outbox payload", exception); + } + } + + private record DetectionRequestedPayload(String taskId) { + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxService.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxService.java new file mode 100644 index 0000000..3138df8 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxService.java @@ -0,0 +1,155 @@ +package com.fengting.aigcforensics.service; + +import java.time.Clock; +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.domain.PageRequest; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fengting.aigcforensics.config.JobOutboxProperties; +import com.fengting.aigcforensics.domain.JobOutboxEvent; +import com.fengting.aigcforensics.domain.JobOutboxEventType; +import com.fengting.aigcforensics.domain.JobOutboxStatus; +import com.fengting.aigcforensics.repository.JobOutboxEventRepository; + +@Service +public class JobOutboxService { + + private static final int STALE_RECOVERY_BATCH_SIZE = 100; + + private final JobOutboxEventRepository repository; + private final ObjectMapper objectMapper; + private final JobOutboxProperties properties; + private final OutboxBackoffPolicy backoffPolicy; + private final Clock clock; + + @Autowired + public JobOutboxService( + JobOutboxEventRepository repository, + ObjectMapper objectMapper, + JobOutboxProperties properties) { + this(repository, objectMapper, properties, Clock.systemUTC()); + } + + JobOutboxService( + JobOutboxEventRepository repository, + ObjectMapper objectMapper, + JobOutboxProperties properties, + Clock clock) { + this.repository = repository; + this.objectMapper = objectMapper; + this.properties = properties; + this.backoffPolicy = new OutboxBackoffPolicy( + properties.getBaseRetryDelay(), + properties.getMaxRetryDelay()); + this.clock = clock; + } + + @Transactional + public JobOutboxEvent scheduleDetection(String taskId) { + return repository.findByEventTypeAndAggregateId( + JobOutboxEventType.DETECTION_REQUESTED, + taskId) + .orElseGet(() -> repository.save(newDetectionEvent(taskId))); + } + + @Transactional + public JobOutboxEvent replayDetection(String taskId) { + Optional existing = repository.findByEventTypeAndAggregateId( + JobOutboxEventType.DETECTION_REQUESTED, + taskId); + if (existing.isEmpty()) { + return repository.save(newDetectionEvent(taskId)); + } + JobOutboxEvent event = existing.get(); + if (event.getStatus() == JobOutboxStatus.PUBLISHED || event.getStatus() == JobOutboxStatus.FAILED) { + event.replay(Instant.now(clock)); + } + return event; + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public Optional claimNext() { + Instant now = Instant.now(clock); + Instant staleBefore = now.minus(properties.getStaleAfter()); + repository.findStalePublishing( + JobOutboxStatus.PUBLISHING, + staleBefore, + PageRequest.of(0, STALE_RECOVERY_BATCH_SIZE)) + .forEach(event -> event.recoverStaleClaim(now)); + + List candidates = repository.findClaimable( + JobOutboxStatus.PENDING, + now, + PageRequest.of(0, 1)); + if (candidates.isEmpty()) { + return Optional.empty(); + } + JobOutboxEvent event = candidates.get(0); + event.claim(now); + return Optional.of(event); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void markPublished(String eventId) { + JobOutboxEvent event = findForUpdate(eventId); + event.markPublished(Instant.now(clock)); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void markPublishFailed(String eventId, String error) { + JobOutboxEvent event = findForUpdate(eventId); + Instant now = Instant.now(clock); + Instant retryAt = now.plus(backoffPolicy.delay(eventId, event.getAttemptCount())); + event.markPublishFailed(error, retryAt, properties.getMaxAttempts(), now); + } + + @Transactional + public JobOutboxEvent replay(String eventId) { + JobOutboxEvent event = findForUpdate(eventId); + event.replay(Instant.now(clock)); + return event; + } + + @Transactional(readOnly = true) + public List list(JobOutboxStatus status, int limit) { + return repository.findByStatusOrderByCreatedAtDesc(status, PageRequest.of(0, limit)); + } + + private JobOutboxEvent newDetectionEvent(String taskId) { + Instant now = Instant.now(clock); + return JobOutboxEvent.detectionRequested( + newExternalId("event"), + taskId, + serializePayload(new DetectionRequestedPayload(taskId)), + now); + } + + private String serializePayload(DetectionRequestedPayload payload) { + try { + return objectMapper.writeValueAsString(payload); + } catch (JsonProcessingException exception) { + throw new IllegalStateException("Failed to serialize outbox payload", exception); + } + } + + private JobOutboxEvent findForUpdate(String eventId) { + return repository.findByEventIdForUpdate(eventId) + .orElseThrow(() -> new ResourceNotFoundException("Outbox event not found: " + eventId)); + } + + private String newExternalId(String prefix) { + return prefix + "_" + UUID.randomUUID().toString().replace("-", ""); + } + + private record DetectionRequestedPayload(String taskId) { + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/OutboxBackoffPolicy.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/OutboxBackoffPolicy.java new file mode 100644 index 0000000..d5b8214 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/OutboxBackoffPolicy.java @@ -0,0 +1,43 @@ +package com.fengting.aigcforensics.service; + +import java.time.Duration; + +public class OutboxBackoffPolicy { + + private static final int MIN_JITTER_PERMILLE = 800; + private static final int JITTER_RANGE = 401; + + private final Duration baseDelay; + private final Duration maxDelay; + + public OutboxBackoffPolicy(Duration baseDelay, Duration maxDelay) { + if (baseDelay == null || baseDelay.isNegative() || baseDelay.isZero()) { + throw new IllegalArgumentException("baseDelay must be positive"); + } + if (maxDelay == null || maxDelay.compareTo(baseDelay) < 0) { + throw new IllegalArgumentException("maxDelay must be at least baseDelay"); + } + this.baseDelay = baseDelay; + this.maxDelay = maxDelay; + } + + public Duration delay(String eventId, int attemptCount) { + if (eventId == null || eventId.isBlank()) { + throw new IllegalArgumentException("eventId must not be blank"); + } + if (attemptCount < 1) { + throw new IllegalArgumentException("attemptCount must be at least 1"); + } + + long baseMillis = baseDelay.toMillis(); + long maxMillis = maxDelay.toMillis(); + int shift = Math.min(attemptCount - 1, 30); + long exponential = baseMillis > (Long.MAX_VALUE >> shift) + ? Long.MAX_VALUE + : baseMillis << shift; + long capped = Math.min(exponential, maxMillis); + int jitterPermille = MIN_JITTER_PERMILLE + Math.floorMod(eventId.hashCode(), JITTER_RANGE); + long jittered = capped * jitterPermille / 1000; + return Duration.ofMillis(Math.max(1, Math.min(jittered, maxMillis))); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/RedisDetectionJobQueue.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/RedisDetectionJobQueue.java index e2d9a48..25316e9 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/service/RedisDetectionJobQueue.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/RedisDetectionJobQueue.java @@ -1,5 +1,8 @@ package com.fengting.aigcforensics.service; +import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -27,6 +30,10 @@ public class RedisDetectionJobQueue implements DetectionJobQueue, DetectionJobConsumer { private static final String TASK_ID_FIELD = "taskId"; + private static final String EVENT_ID_FIELD = "eventId"; + private static final String EVENT_VERSION_FIELD = "eventVersion"; + private static final String OCCURRED_AT_FIELD = "occurredAt"; + private static final int SUPPORTED_EVENT_VERSION = 1; private static final String ORIGINAL_MESSAGE_ID_FIELD = "originalMessageId"; private static final String DELIVERY_COUNT_FIELD = "deliveryCount"; private static final String REASON_FIELD = "reason"; @@ -43,18 +50,28 @@ public RedisDetectionJobQueue( } @Override - public void enqueue(String taskId) { - String submittedKey = submittedKey(taskId); + public void enqueue(DetectionJobRequest request) { + String submittedKey = submittedKey(request.eventId()); Boolean acquired = redisTemplate.opsForValue() .setIfAbsent(submittedKey, "1", properties.getSubmittedTtl()); + if (acquired == null) { + throw new IllegalStateException("Redis deduplication returned no result for event " + request.eventId()); + } if (Boolean.FALSE.equals(acquired)) { return; } try { - redisTemplate.opsForStream().add(StreamRecords.newRecord() - .ofMap(Map.of(TASK_ID_FIELD, taskId)) + RecordId recordId = redisTemplate.opsForStream().add(StreamRecords.newRecord() + .ofMap(Map.of( + EVENT_ID_FIELD, request.eventId(), + EVENT_VERSION_FIELD, String.valueOf(request.eventVersion()), + TASK_ID_FIELD, request.taskId(), + OCCURRED_AT_FIELD, request.occurredAt().toString())) .withStreamKey(properties.getStreamKey())); + if (recordId == null) { + throw new IllegalStateException("Redis stream returned no record id for event " + request.eventId()); + } } catch (RuntimeException exception) { redisTemplate.delete(submittedKey); throw exception; @@ -96,8 +113,8 @@ public void acknowledge(DetectionJobMessage message) { properties.getStreamKey(), properties.getGroupName(), RecordId.of(message.messageId())); - if (!message.taskId().isBlank()) { - redisTemplate.delete(submittedKey(message.taskId())); + if (!message.eventId().isBlank()) { + redisTemplate.delete(submittedKey(message.eventId())); } } @@ -159,25 +176,83 @@ private Optional> claim(RecordId recordId) { } private Optional toMessage(MapRecord record) { - Object taskId = record.getValue().get(TASK_ID_FIELD); - if (taskId == null) { - acknowledge(new DetectionJobMessage(record.getId().getValue(), "")); + String messageId = record.getId().getValue(); + String eventId = field(record, EVENT_ID_FIELD); + String taskId = field(record, TASK_ID_FIELD); + String versionText = field(record, EVENT_VERSION_FIELD); + String occurredAt = field(record, OCCURRED_AT_FIELD); + + if (eventId.isBlank()) { + moveInvalidToDeadLetter(record, "missing event id"); + return Optional.empty(); + } + if (taskId.isBlank()) { + moveInvalidToDeadLetter(record, "missing task id"); + return Optional.empty(); + } + + int eventVersion; + try { + eventVersion = Integer.parseInt(versionText); + } catch (NumberFormatException exception) { + moveInvalidToDeadLetter(record, "invalid event version: " + versionText); + return Optional.empty(); + } + if (eventVersion != SUPPORTED_EVENT_VERSION) { + moveInvalidToDeadLetter(record, "unsupported event version: " + eventVersion); + return Optional.empty(); + } + try { + Instant.parse(occurredAt); + } catch (DateTimeParseException exception) { + moveInvalidToDeadLetter(record, "invalid occurredAt: " + occurredAt); return Optional.empty(); } - return Optional.of(new DetectionJobMessage(record.getId().getValue(), taskId.toString())); + return Optional.of(new DetectionJobMessage(messageId, eventId, eventVersion, taskId)); } private void moveToDeadLetter(MapRecord record, long deliveryCount) { String messageId = record.getId().getValue(); - Object taskId = record.getValue().get(TASK_ID_FIELD); + String eventId = field(record, EVENT_ID_FIELD); + String taskId = field(record, TASK_ID_FIELD); redisTemplate.opsForStream().add(StreamRecords.newRecord() .ofMap(Map.of( - TASK_ID_FIELD, taskId == null ? "" : taskId.toString(), + EVENT_ID_FIELD, eventId, + TASK_ID_FIELD, taskId, ORIGINAL_MESSAGE_ID_FIELD, messageId, DELIVERY_COUNT_FIELD, String.valueOf(deliveryCount), REASON_FIELD, "max delivery attempts exceeded")) .withStreamKey(properties.getDeadLetterStreamKey())); - acknowledge(new DetectionJobMessage(messageId, taskId == null ? "" : taskId.toString())); + acknowledgeRecord(messageId, eventId); + } + + private void moveInvalidToDeadLetter(MapRecord record, String reason) { + String messageId = record.getId().getValue(); + String eventId = field(record, EVENT_ID_FIELD); + Map deadLetter = new LinkedHashMap<>(); + deadLetter.put(EVENT_ID_FIELD, eventId); + deadLetter.put(TASK_ID_FIELD, field(record, TASK_ID_FIELD)); + deadLetter.put(ORIGINAL_MESSAGE_ID_FIELD, messageId); + deadLetter.put(REASON_FIELD, reason); + redisTemplate.opsForStream().add(StreamRecords.newRecord() + .ofMap(deadLetter) + .withStreamKey(properties.getDeadLetterStreamKey())); + acknowledgeRecord(messageId, eventId); + } + + private void acknowledgeRecord(String messageId, String eventId) { + redisTemplate.opsForStream().acknowledge( + properties.getStreamKey(), + properties.getGroupName(), + RecordId.of(messageId)); + if (!eventId.isBlank()) { + redisTemplate.delete(submittedKey(eventId)); + } + } + + private String field(MapRecord record, String fieldName) { + Object value = record.getValue().get(fieldName); + return value == null ? "" : value.toString(); } private boolean isGroupAlreadyExistsError(RedisSystemException exception) { @@ -192,7 +267,7 @@ private boolean isStreamMissingError(RedisSystemException exception) { return message.contains("requires the key to exist") || message.contains("no such key"); } - private String submittedKey(String taskId) { - return properties.getSubmittedKeyPrefix() + taskId; + private String submittedKey(String eventId) { + return properties.getSubmittedKeyPrefix() + eventId; } } diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 778b664..0d8d0d6 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -22,6 +22,14 @@ server: app: storage: root: ${STORAGE_ROOT:storage} + jobs: + outbox: + dispatcher-enabled: ${APP_JOBS_OUTBOX_DISPATCHER_ENABLED:true} + poll-delay-ms: ${APP_JOBS_OUTBOX_POLL_DELAY_MS:500} + stale-after: ${APP_JOBS_OUTBOX_STALE_AFTER:1m} + max-attempts: ${APP_JOBS_OUTBOX_MAX_ATTEMPTS:5} + base-retry-delay: ${APP_JOBS_OUTBOX_BASE_RETRY_DELAY:1s} + max-retry-delay: ${APP_JOBS_OUTBOX_MAX_RETRY_DELAY:1m} detection: jobs: worker-enabled: ${APP_DETECTION_JOBS_WORKER_ENABLED:true} diff --git a/backend-java/src/main/resources/db/migration/V5__add_job_outbox.sql b/backend-java/src/main/resources/db/migration/V5__add_job_outbox.sql new file mode 100644 index 0000000..a942cc0 --- /dev/null +++ b/backend-java/src/main/resources/db/migration/V5__add_job_outbox.sql @@ -0,0 +1,20 @@ +create table job_outbox ( + id bigint generated by default as identity primary key, + event_id varchar(64) not null unique, + event_type varchar(64) not null, + aggregate_type varchar(64) not null, + aggregate_id varchar(64) not null, + payload_json text not null, + status varchar(32) not null, + attempt_count integer not null default 0, + available_at timestamp with time zone, + last_error varchar(2048), + created_at timestamp with time zone not null, + updated_at timestamp with time zone not null, + published_at timestamp with time zone, + lock_version bigint not null default 0, + constraint uq_job_outbox_aggregate_event unique (event_type, aggregate_id) +); + +create index idx_job_outbox_claim on job_outbox(status, available_at, created_at); +create index idx_job_outbox_stale on job_outbox(status, updated_at); diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/controller/JobOutboxOperationsControllerTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/controller/JobOutboxOperationsControllerTest.java new file mode 100644 index 0000000..baf4306 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/controller/JobOutboxOperationsControllerTest.java @@ -0,0 +1,100 @@ +package com.fengting.aigcforensics.controller; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import java.time.Instant; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; +import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.test.web.servlet.MockMvc; + +import com.fengting.aigcforensics.domain.JobOutboxEvent; +import com.fengting.aigcforensics.domain.JobOutboxStatus; +import com.fengting.aigcforensics.domain.InvalidJobOutboxStateException; +import com.fengting.aigcforensics.service.JobOutboxService; +import com.fengting.aigcforensics.service.ResourceNotFoundException; + +@WebMvcTest(JobOutboxOperationsController.class) +class JobOutboxOperationsControllerTest { + + private static final Instant CREATED_AT = Instant.parse("2026-07-11T00:00:00Z"); + + @Autowired + private MockMvc mockMvc; + + @MockitoBean + private JobOutboxService outboxService; + + @Test + void listsBoundedOutboxMetadata() throws Exception { + JobOutboxEvent event = event(); + event.claim(CREATED_AT); + event.markPublishFailed("redis unavailable", CREATED_AT.plusSeconds(1), 1, CREATED_AT); + when(outboxService.list(JobOutboxStatus.FAILED, 50)).thenReturn(List.of(event)); + + mockMvc.perform(get("/api/operations/job-outbox").param("status", "FAILED")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$[0].eventId").value("event-001")) + .andExpect(jsonPath("$[0].eventType").value("DETECTION_REQUESTED")) + .andExpect(jsonPath("$[0].aggregateId").value("task-001")) + .andExpect(jsonPath("$[0].status").value("FAILED")) + .andExpect(jsonPath("$[0].lastError").value("redis unavailable")) + .andExpect(jsonPath("$[0].payloadJson").doesNotExist()); + } + + @Test + void rejectsListLimitAboveOneHundred() throws Exception { + mockMvc.perform(get("/api/operations/job-outbox") + .param("status", "FAILED") + .param("limit", "101")) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.message").value("limit must be between 1 and 100")); + } + + @Test + void replaysTerminalEvent() throws Exception { + when(outboxService.replay("event-001")).thenReturn(event()); + + mockMvc.perform(post("/api/operations/job-outbox/{eventId}/replay", "event-001")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.eventId").value("event-001")) + .andExpect(jsonPath("$.status").value("PENDING")); + } + + @Test + void returnsNotFoundForMissingEvent() throws Exception { + when(outboxService.replay("event-missing")) + .thenThrow(new ResourceNotFoundException("Outbox event not found: event-missing")); + + mockMvc.perform(post("/api/operations/job-outbox/{eventId}/replay", "event-missing")) + .andExpect(status().isNotFound()) + .andExpect(jsonPath("$.message").value("Outbox event not found: event-missing")); + } + + @Test + void returnsConflictForNonTerminalReplay() throws Exception { + when(outboxService.replay(any())) + .thenThrow(new InvalidJobOutboxStateException( + "Only terminal outbox events can be replayed: event-001")); + + mockMvc.perform(post("/api/operations/job-outbox/{eventId}/replay", "event-001")) + .andExpect(status().isConflict()) + .andExpect(jsonPath("$.message").value("Only terminal outbox events can be replayed: event-001")); + } + + private JobOutboxEvent event() { + return JobOutboxEvent.detectionRequested( + "event-001", + "task-001", + "{\"taskId\":\"task-001\"}", + CREATED_AT); + } +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/domain/JobOutboxEventTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/domain/JobOutboxEventTest.java new file mode 100644 index 0000000..cd27037 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/domain/JobOutboxEventTest.java @@ -0,0 +1,125 @@ +package com.fengting.aigcforensics.domain; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.Instant; + +import org.junit.jupiter.api.Test; + +class JobOutboxEventTest { + + private static final Instant CREATED_AT = Instant.parse("2026-07-11T00:00:00Z"); + + @Test + void createsPendingDetectionEvent() { + JobOutboxEvent event = newEvent(); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(event.getAttemptCount()).isZero(); + assertThat(event.getAvailableAt()).isEqualTo(CREATED_AT); + assertThat(event.getPayloadJson()).isEqualTo("{\"taskId\":\"task-001\"}"); + } + + @Test + void claimsPendingEventWhenAvailable() { + JobOutboxEvent event = newEvent(); + Instant claimedAt = CREATED_AT.plusSeconds(1); + + event.claim(claimedAt); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PUBLISHING); + assertThat(event.getAttemptCount()).isEqualTo(1); + assertThat(event.getUpdatedAt()).isEqualTo(claimedAt); + } + + @Test + void rejectsClaimBeforeAvailableTime() { + JobOutboxEvent event = newEvent(); + event.claim(CREATED_AT); + event.markPublishFailed("redis unavailable", CREATED_AT.plusSeconds(10), 5, CREATED_AT.plusSeconds(1)); + + assertThatThrownBy(() -> event.claim(CREATED_AT.plusSeconds(5))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("not available"); + } + + @Test + void marksPublishedEvent() { + JobOutboxEvent event = claimedEvent(); + Instant publishedAt = CREATED_AT.plusSeconds(2); + + event.markPublished(publishedAt); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PUBLISHED); + assertThat(event.getPublishedAt()).isEqualTo(publishedAt); + assertThat(event.getLastError()).isNull(); + } + + @Test + void schedulesRetryAfterPublishFailure() { + JobOutboxEvent event = claimedEvent(); + Instant failedAt = CREATED_AT.plusSeconds(2); + Instant retryAt = CREATED_AT.plusSeconds(10); + + event.markPublishFailed("redis unavailable", retryAt, 5, failedAt); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(event.getAvailableAt()).isEqualTo(retryAt); + assertThat(event.getLastError()).isEqualTo("redis unavailable"); + assertThat(event.getUpdatedAt()).isEqualTo(failedAt); + } + + @Test + void marksEventFailedAfterRetryBudgetIsExhausted() { + JobOutboxEvent event = newEvent(); + for (int attempt = 1; attempt <= 5; attempt++) { + Instant attemptedAt = CREATED_AT.plusSeconds(attempt); + event.claim(attemptedAt); + event.markPublishFailed("redis unavailable", attemptedAt.plusSeconds(1), 5, attemptedAt); + } + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.FAILED); + assertThat(event.getAttemptCount()).isEqualTo(5); + } + + @Test + void recoversStalePublishingEvent() { + JobOutboxEvent event = claimedEvent(); + Instant recoveredAt = CREATED_AT.plusSeconds(60); + + event.recoverStaleClaim(recoveredAt); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(event.getAvailableAt()).isEqualTo(recoveredAt); + assertThat(event.getLastError()).contains("stale"); + } + + @Test + void replaysTerminalEvent() { + JobOutboxEvent event = claimedEvent(); + event.markPublished(CREATED_AT.plusSeconds(2)); + Instant replayedAt = CREATED_AT.plusSeconds(3); + + event.replay(replayedAt); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(event.getAttemptCount()).isZero(); + assertThat(event.getAvailableAt()).isEqualTo(replayedAt); + assertThat(event.getPublishedAt()).isNull(); + } + + private JobOutboxEvent claimedEvent() { + JobOutboxEvent event = newEvent(); + event.claim(CREATED_AT.plusSeconds(1)); + return event; + } + + private JobOutboxEvent newEvent() { + return JobOutboxEvent.detectionRequested( + "event-001", + "task-001", + "{\"taskId\":\"task-001\"}", + CREATED_AT); + } +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobServiceTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobServiceTest.java index c5ceb98..c4371dc 100644 --- a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobServiceTest.java +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobServiceTest.java @@ -1,11 +1,11 @@ package com.fengting.aigcforensics.service; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.time.Instant; -import java.util.ArrayList; -import java.util.List; import java.util.Optional; import org.junit.jupiter.api.Test; @@ -23,44 +23,45 @@ class DetectionJobServiceTest { @Mock private DetectionTaskRepository detectionTaskRepository; + @Mock + private JobOutboxService jobOutboxService; + @Test void submitsQueuedTaskToQueue() { - CapturingDetectionJobQueue jobQueue = new CapturingDetectionJobQueue(); DetectionTask task = task("task-001", DetectionStatus.QUEUED); - when(detectionTaskRepository.findByTaskId("task-001")).thenReturn(Optional.of(task)); + when(detectionTaskRepository.findByTaskIdForUpdate("task-001")).thenReturn(Optional.of(task)); DetectionTask submitted = new DetectionJobService( detectionTaskRepository, - jobQueue).submit("task-001"); + jobOutboxService).submit("task-001"); assertThat(submitted).isSameAs(task); - assertThat(jobQueue.taskIds).containsExactly("task-001"); + verify(jobOutboxService).scheduleDetection("task-001"); } @Test void doesNotSubmitCompletedTask() { - CapturingDetectionJobQueue jobQueue = new CapturingDetectionJobQueue(); DetectionTask task = task("task-001", DetectionStatus.QUEUED); task.markCompleted(Instant.parse("2026-07-07T00:00:01Z")); - when(detectionTaskRepository.findByTaskId("task-001")).thenReturn(Optional.of(task)); + when(detectionTaskRepository.findByTaskIdForUpdate("task-001")).thenReturn(Optional.of(task)); - new DetectionJobService(detectionTaskRepository, jobQueue) + new DetectionJobService(detectionTaskRepository, jobOutboxService) .submit("task-001"); - assertThat(jobQueue.taskIds).isEmpty(); + verify(jobOutboxService, never()).scheduleDetection("task-001"); + verify(jobOutboxService, never()).replayDetection("task-001"); } @Test void submitsFailedTaskForManualRetry() { - CapturingDetectionJobQueue jobQueue = new CapturingDetectionJobQueue(); DetectionTask task = task("task-001", DetectionStatus.QUEUED); task.markFailed("temporary model outage", Instant.parse("2026-07-07T00:00:01Z")); - when(detectionTaskRepository.findByTaskId("task-001")).thenReturn(Optional.of(task)); + when(detectionTaskRepository.findByTaskIdForUpdate("task-001")).thenReturn(Optional.of(task)); - new DetectionJobService(detectionTaskRepository, jobQueue) + new DetectionJobService(detectionTaskRepository, jobOutboxService) .submit("task-001"); - assertThat(jobQueue.taskIds).containsExactly("task-001"); + verify(jobOutboxService).replayDetection("task-001"); } private DetectionTask task(String taskId, DetectionStatus status) { @@ -70,14 +71,4 @@ private DetectionTask task(String taskId, DetectionStatus status) { status, Instant.parse("2026-07-07T00:00:00Z")); } - - private static class CapturingDetectionJobQueue implements DetectionJobQueue { - - private final List taskIds = new ArrayList<>(); - - @Override - public void enqueue(String taskId) { - taskIds.add(taskId); - } - } } diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobWorkerTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobWorkerTest.java index 2bfc31d..055c1bb 100644 --- a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobWorkerTest.java +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobWorkerTest.java @@ -22,7 +22,7 @@ class DetectionJobWorkerTest { @Test void pollsJobRunsDetectionAndAcknowledgesMessage() { - DetectionJobMessage message = new DetectionJobMessage("message-001", "task-001"); + DetectionJobMessage message = new DetectionJobMessage("message-001", "event-001", 1, "task-001"); when(detectionJobConsumer.poll()).thenReturn(Optional.of(message)); new DetectionJobWorker(detectionJobConsumer, detectionExecutionService).pollOnce(); diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxDispatcherTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxDispatcherTest.java new file mode 100644 index 0000000..5235239 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxDispatcherTest.java @@ -0,0 +1,68 @@ +package com.fengting.aigcforensics.service; + +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.time.Instant; +import java.util.Optional; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.fengting.aigcforensics.domain.JobOutboxEvent; + +@ExtendWith(MockitoExtension.class) +class JobOutboxDispatcherTest { + + @Mock + private JobOutboxService outboxService; + + @Mock + private JobOutboxPublisher publisher; + + @Test + void marksClaimedEventPublishedAfterQueueAcceptsIt() { + JobOutboxEvent event = event(); + when(outboxService.claimNext()).thenReturn(Optional.of(event)); + + new JobOutboxDispatcher(outboxService, publisher).pollOnce(); + + verify(publisher).publish(event); + verify(outboxService).markPublished("event-001"); + verify(outboxService, never()).markPublishFailed("event-001", "redis unavailable"); + } + + @Test + void recordsRetryableFailureWhenPublishThrows() { + JobOutboxEvent event = event(); + when(outboxService.claimNext()).thenReturn(Optional.of(event)); + doThrow(new IllegalStateException("redis unavailable")).when(publisher).publish(event); + + new JobOutboxDispatcher(outboxService, publisher).pollOnce(); + + verify(outboxService).markPublishFailed("event-001", "redis unavailable"); + verify(outboxService, never()).markPublished("event-001"); + } + + @Test + void doesNothingWhenNoEventIsAvailable() { + when(outboxService.claimNext()).thenReturn(Optional.empty()); + + new JobOutboxDispatcher(outboxService, publisher).pollOnce(); + + verifyNoInteractions(publisher); + } + + private JobOutboxEvent event() { + return JobOutboxEvent.detectionRequested( + "event-001", + "task-001", + "{\"taskId\":\"task-001\"}", + Instant.parse("2026-07-11T00:00:00Z")); + } +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxPublisherTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxPublisherTest.java new file mode 100644 index 0000000..3b8a120 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxPublisherTest.java @@ -0,0 +1,65 @@ +package com.fengting.aigcforensics.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.verify; + +import java.time.Instant; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fengting.aigcforensics.domain.JobOutboxEvent; + +@ExtendWith(MockitoExtension.class) +class JobOutboxPublisherTest { + + @Mock + private DetectionJobQueue detectionJobQueue; + + @Test + void publishesDetectionRequestedEvent() { + JobOutboxEvent event = event("{\"taskId\":\"task-001\"}"); + JobOutboxPublisher publisher = new JobOutboxPublisher(detectionJobQueue, new ObjectMapper()); + + publisher.publish(event); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(DetectionJobRequest.class); + verify(detectionJobQueue).enqueue(requestCaptor.capture()); + assertThat(requestCaptor.getValue()).isEqualTo(new DetectionJobRequest( + "event-001", + 1, + "task-001", + Instant.parse("2026-07-11T00:00:00Z"))); + } + + @Test + void rejectsMalformedDetectionPayload() { + JobOutboxPublisher publisher = new JobOutboxPublisher(detectionJobQueue, new ObjectMapper()); + + assertThatThrownBy(() -> publisher.publish(event("not-json"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("payload"); + } + + @Test + void rejectsPayloadForDifferentAggregate() { + JobOutboxPublisher publisher = new JobOutboxPublisher(detectionJobQueue, new ObjectMapper()); + + assertThatThrownBy(() -> publisher.publish(event("{\"taskId\":\"task-other\"}"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("aggregate"); + } + + private JobOutboxEvent event(String payloadJson) { + return JobOutboxEvent.detectionRequested( + "event-001", + "task-001", + payloadJson, + Instant.parse("2026-07-11T00:00:00Z")); + } +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxServiceTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxServiceTest.java new file mode 100644 index 0000000..3b74be2 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxServiceTest.java @@ -0,0 +1,154 @@ +package com.fengting.aigcforensics.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Optional; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.data.domain.Pageable; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fengting.aigcforensics.config.JobOutboxProperties; +import com.fengting.aigcforensics.domain.JobOutboxEvent; +import com.fengting.aigcforensics.domain.JobOutboxEventType; +import com.fengting.aigcforensics.domain.JobOutboxStatus; +import com.fengting.aigcforensics.repository.JobOutboxEventRepository; + +@ExtendWith(MockitoExtension.class) +class JobOutboxServiceTest { + + private static final Instant NOW = Instant.parse("2026-07-11T00:00:00Z"); + + @Mock + private JobOutboxEventRepository repository; + + private JobOutboxService service; + + @BeforeEach + void setUp() { + JobOutboxProperties properties = new JobOutboxProperties(); + properties.setStaleAfter(Duration.ofMinutes(1)); + properties.setMaxAttempts(5); + properties.setBaseRetryDelay(Duration.ofSeconds(1)); + properties.setMaxRetryDelay(Duration.ofMinutes(1)); + service = new JobOutboxService( + repository, + new ObjectMapper(), + properties, + Clock.fixed(NOW, ZoneOffset.UTC)); + } + + @Test + void schedulesNewDetectionEvent() { + when(repository.findByEventTypeAndAggregateId( + JobOutboxEventType.DETECTION_REQUESTED, + "task-001")) + .thenReturn(Optional.empty()); + when(repository.save(any(JobOutboxEvent.class))) + .thenAnswer(invocation -> invocation.getArgument(0)); + + JobOutboxEvent scheduled = service.scheduleDetection("task-001"); + + assertThat(scheduled.getEventId()).startsWith("event_"); + assertThat(scheduled.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(scheduled.getPayloadJson()).isEqualTo("{\"taskId\":\"task-001\"}"); + } + + @Test + void reusesExistingDetectionEventForDuplicateSubmission() { + JobOutboxEvent existing = event("event-001", "task-001"); + when(repository.findByEventTypeAndAggregateId( + JobOutboxEventType.DETECTION_REQUESTED, + "task-001")) + .thenReturn(Optional.of(existing)); + + assertThat(service.scheduleDetection("task-001")).isSameAs(existing); + verify(repository, never()).save(any()); + } + + @Test + void replaysTerminalEventForFailedTaskRetry() { + JobOutboxEvent existing = event("event-001", "task-001"); + existing.claim(NOW); + existing.markPublished(NOW); + when(repository.findByEventTypeAndAggregateId( + JobOutboxEventType.DETECTION_REQUESTED, + "task-001")) + .thenReturn(Optional.of(existing)); + + JobOutboxEvent replayed = service.replayDetection("task-001"); + + assertThat(replayed.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(replayed.getAvailableAt()).isEqualTo(NOW); + } + + @Test + void recoversStaleClaimAndClaimsOldestAvailableEvent() { + JobOutboxEvent stale = event("event-stale", "task-stale"); + stale.claim(NOW.minusSeconds(120)); + JobOutboxEvent pending = event("event-next", "task-next"); + when(repository.findStalePublishing( + eq(JobOutboxStatus.PUBLISHING), + eq(NOW.minusSeconds(60)), + any(Pageable.class))) + .thenReturn(List.of(stale)); + when(repository.findClaimable( + eq(JobOutboxStatus.PENDING), + eq(NOW), + any(Pageable.class))) + .thenReturn(List.of(pending)); + + Optional claimed = service.claimNext(); + + assertThat(stale.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(claimed).contains(pending); + assertThat(pending.getStatus()).isEqualTo(JobOutboxStatus.PUBLISHING); + } + + @Test + void recordsRetryWithBackoffAfterPublishFailure() { + JobOutboxEvent event = event("event-001", "task-001"); + event.claim(NOW); + when(repository.findByEventIdForUpdate("event-001")).thenReturn(Optional.of(event)); + + service.markPublishFailed("event-001", "redis unavailable"); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(event.getAvailableAt()).isAfter(NOW); + assertThat(event.getLastError()).isEqualTo("redis unavailable"); + } + + @Test + void marksClaimedEventPublished() { + JobOutboxEvent event = event("event-001", "task-001"); + event.claim(NOW); + when(repository.findByEventIdForUpdate("event-001")).thenReturn(Optional.of(event)); + + service.markPublished("event-001"); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PUBLISHED); + assertThat(event.getPublishedAt()).isEqualTo(NOW); + } + + private JobOutboxEvent event(String eventId, String taskId) { + return JobOutboxEvent.detectionRequested( + eventId, + taskId, + "{\"taskId\":\"" + taskId + "\"}", + NOW.minusSeconds(180)); + } +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/OutboxBackoffPolicyTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/OutboxBackoffPolicyTest.java new file mode 100644 index 0000000..c2f74ee --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/OutboxBackoffPolicyTest.java @@ -0,0 +1,34 @@ +package com.fengting.aigcforensics.service; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; + +class OutboxBackoffPolicyTest { + + @Test + void doublesDelayAndCapsAtConfiguredMaximum() { + OutboxBackoffPolicy policy = new OutboxBackoffPolicy( + Duration.ofSeconds(1), + Duration.ofSeconds(8)); + + assertThat(policy.delay("event-001", 1)) + .isBetween(Duration.ofMillis(800), Duration.ofMillis(1200)); + assertThat(policy.delay("event-001", 2)) + .isBetween(Duration.ofMillis(1600), Duration.ofMillis(2400)); + assertThat(policy.delay("event-001", 8)) + .isBetween(Duration.ofMillis(6400), Duration.ofMillis(9600)); + } + + @Test + void returnsStableJitterForSameEventAndAttempt() { + OutboxBackoffPolicy policy = new OutboxBackoffPolicy( + Duration.ofSeconds(1), + Duration.ofMinutes(1)); + + assertThat(policy.delay("event-001", 3)) + .isEqualTo(policy.delay("event-001", 3)); + } +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/RedisDetectionJobQueueTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/RedisDetectionJobQueueTest.java index aaefa3e..0bbd385 100644 --- a/backend-java/src/test/java/com/fengting/aigcforensics/service/RedisDetectionJobQueueTest.java +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/RedisDetectionJobQueueTest.java @@ -1,6 +1,7 @@ package com.fengting.aigcforensics.service; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.never; @@ -8,6 +9,7 @@ import static org.mockito.Mockito.when; import java.time.Duration; +import java.time.Instant; import java.util.List; import java.util.Map; import java.util.Optional; @@ -31,6 +33,7 @@ import org.springframework.data.domain.Range; import org.springframework.data.redis.core.StreamOperations; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.ValueOperations; import com.fengting.aigcforensics.config.DetectionJobRedisProperties; @@ -44,6 +47,84 @@ class RedisDetectionJobQueueTest { @Mock private StreamOperations streamOperations; + @Mock + private ValueOperations valueOperations; + + @Test + void enqueuesVersionedEventEnvelope() { + DetectionJobRedisProperties properties = new DetectionJobRedisProperties(); + DetectionJobRequest request = request("event-001", "task-001"); + when(redisTemplate.opsForValue()).thenReturn(valueOperations); + when(valueOperations.setIfAbsent( + properties.getSubmittedKeyPrefix() + "event-001", + "1", + properties.getSubmittedTtl())) + .thenReturn(true); + when(redisTemplate.opsForStream()).thenReturn(streamOperations); + when(streamOperations.add(any(MapRecord.class))).thenReturn(RecordId.of("message-001")); + + new RedisDetectionJobQueue(redisTemplate, properties).enqueue(request); + + ArgumentCaptor> recordCaptor = ArgumentCaptor.forClass(MapRecord.class); + verify(streamOperations).add(recordCaptor.capture()); + assertThat(recordCaptor.getValue().getValue()) + .containsEntry("eventId", "event-001") + .containsEntry("eventVersion", "1") + .containsEntry("taskId", "task-001") + .containsEntry("occurredAt", "2026-07-11T00:00:00Z"); + } + + @Test + void doesNotEnqueueDuplicateEventId() { + DetectionJobRedisProperties properties = new DetectionJobRedisProperties(); + when(redisTemplate.opsForValue()).thenReturn(valueOperations); + when(valueOperations.setIfAbsent( + properties.getSubmittedKeyPrefix() + "event-001", + "1", + properties.getSubmittedTtl())) + .thenReturn(false); + + new RedisDetectionJobQueue(redisTemplate, properties).enqueue(request("event-001", "task-001")); + + verify(redisTemplate, never()).opsForStream(); + } + + @Test + void rejectsIndeterminateDeduplicationResult() { + DetectionJobRedisProperties properties = new DetectionJobRedisProperties(); + when(redisTemplate.opsForValue()).thenReturn(valueOperations); + when(valueOperations.setIfAbsent( + properties.getSubmittedKeyPrefix() + "event-001", + "1", + properties.getSubmittedTtl())) + .thenReturn(null); + + assertThatThrownBy(() -> new RedisDetectionJobQueue(redisTemplate, properties) + .enqueue(request("event-001", "task-001"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("deduplication"); + verify(redisTemplate, never()).opsForStream(); + } + + @Test + void rejectsMissingStreamRecordIdAndReleasesDeduplicationKey() { + DetectionJobRedisProperties properties = new DetectionJobRedisProperties(); + when(redisTemplate.opsForValue()).thenReturn(valueOperations); + when(valueOperations.setIfAbsent( + properties.getSubmittedKeyPrefix() + "event-001", + "1", + properties.getSubmittedTtl())) + .thenReturn(true); + when(redisTemplate.opsForStream()).thenReturn(streamOperations); + when(streamOperations.add(any(MapRecord.class))).thenReturn(null); + + assertThatThrownBy(() -> new RedisDetectionJobQueue(redisTemplate, properties) + .enqueue(request("event-001", "task-001"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("record id"); + verify(redisTemplate).delete(properties.getSubmittedKeyPrefix() + "event-001"); + } + @Test void returnsEmptyWhenPollingBeforeStreamExists() { DetectionJobRedisProperties properties = new DetectionJobRedisProperties(); @@ -73,7 +154,7 @@ void claimsStalePendingMessageBeforeReadingNewMessages() { MapRecord claimedRecord = record( properties.getStreamKey(), "message-001", - Map.of("taskId", "task-001")); + eventFields("event-001", "1", "task-001")); when(redisTemplate.opsForStream()).thenReturn(streamOperations); when(streamOperations.pending( @@ -91,7 +172,7 @@ void claimsStalePendingMessageBeforeReadingNewMessages() { Optional message = new RedisDetectionJobQueue(redisTemplate, properties).poll(); - assertThat(message).contains(new DetectionJobMessage("message-001", "task-001")); + assertThat(message).contains(new DetectionJobMessage("message-001", "event-001", 1, "task-001")); verify(streamOperations, never()).read( any(Consumer.class), any(StreamReadOptions.class), @@ -106,7 +187,7 @@ void movesOverDeliveredPendingMessageToDeadLetter() { MapRecord claimedRecord = record( properties.getStreamKey(), "message-001", - Map.of("taskId", "task-001")); + eventFields("event-001", "1", "task-001")); when(redisTemplate.opsForStream()).thenReturn(streamOperations); when(streamOperations.pending( @@ -130,6 +211,7 @@ void movesOverDeliveredPendingMessageToDeadLetter() { verify(streamOperations).add(deadLetterRecordCaptor.capture()); assertThat(deadLetterRecordCaptor.getValue().getStream()).isEqualTo(properties.getDeadLetterStreamKey()); assertThat(deadLetterRecordCaptor.getValue().getValue()) + .containsEntry("eventId", "event-001") .containsEntry("taskId", "task-001") .containsEntry("originalMessageId", "message-001") .containsEntry("deliveryCount", "3"); @@ -139,6 +221,38 @@ void movesOverDeliveredPendingMessageToDeadLetter() { RecordId.of("message-001")); } + @Test + void movesUnsupportedEventVersionToDeadLetter() { + DetectionJobRedisProperties properties = new DetectionJobRedisProperties(); + PendingMessage pendingMessage = pendingMessage(properties, "message-001", Duration.ofMinutes(6), 1); + MapRecord claimedRecord = record( + properties.getStreamKey(), + "message-001", + eventFields("event-001", "2", "task-001")); + when(redisTemplate.opsForStream()).thenReturn(streamOperations); + when(streamOperations.pending( + eq(properties.getStreamKey()), + eq(properties.getGroupName()), + any(Range.class), + eq((long) properties.getPendingClaimBatchSize()))) + .thenReturn(new PendingMessages(properties.getGroupName(), List.of(pendingMessage))); + when(streamOperations.claim( + eq(properties.getStreamKey()), + eq(properties.getGroupName()), + eq(properties.getConsumerName()), + any(XClaimOptions.class))) + .thenReturn(List.of(claimedRecord)); + + assertThat(new RedisDetectionJobQueue(redisTemplate, properties).poll()).isEmpty(); + + ArgumentCaptor> deadLetterRecordCaptor = + ArgumentCaptor.forClass(MapRecord.class); + verify(streamOperations).add(deadLetterRecordCaptor.capture()); + assertThat(deadLetterRecordCaptor.getValue().getValue()) + .containsEntry("eventId", "event-001") + .containsEntry("reason", "unsupported event version: 2"); + } + private PendingMessage pendingMessage( DetectionJobRedisProperties properties, String messageId, @@ -161,4 +275,20 @@ private MapRecord record( .withId(RecordId.of(messageId)) .withStreamKey(streamKey); } + + private DetectionJobRequest request(String eventId, String taskId) { + return new DetectionJobRequest( + eventId, + 1, + taskId, + Instant.parse("2026-07-11T00:00:00Z")); + } + + private Map eventFields(String eventId, String version, String taskId) { + return Map.of( + "eventId", eventId, + "eventVersion", version, + "taskId", taskId, + "occurredAt", "2026-07-11T00:00:00Z"); + } } diff --git a/backend-java/src/test/resources/application-test.yml b/backend-java/src/test/resources/application-test.yml index c274206..f22bf03 100644 --- a/backend-java/src/test/resources/application-test.yml +++ b/backend-java/src/test/resources/application-test.yml @@ -20,6 +20,9 @@ logging: org.springframework: warn app: + jobs: + outbox: + dispatcher-enabled: false detection: jobs: worker-enabled: false diff --git a/docs/README.md b/docs/README.md index 35acc7a..fac2c15 100644 --- a/docs/README.md +++ b/docs/README.md @@ -24,6 +24,8 @@ Use these when setting up or operating the project. services are running. - [Full-Stack Evaluation Demo](fullstack-evaluation-demo.md): admin UI to Java backend evaluation workflow without model weights. +- [Reliable Job Dispatch](reliable-job-dispatch.md): transactional outbox, + Redis delivery, failure semantics, inspection, and replay. ## Architecture And Contracts diff --git a/docs/async-detection-jobs.md b/docs/async-detection-jobs.md index cd93428..6e0f831 100644 --- a/docs/async-detection-jobs.md +++ b/docs/async-detection-jobs.md @@ -13,8 +13,10 @@ debugging. ## Async Run -`/run-async` submits a queued task to the Redis-backed worker queue and immediately -returns `202 Accepted` with the current task snapshot. Clients should poll: +`/run-async` writes or replays a durable PostgreSQL outbox event in the same +transaction used to validate the task, then immediately returns `202 Accepted` +with the current task snapshot. A separate dispatcher publishes eligible +outbox events to Redis. Clients should poll: ```text GET /api/detections/{taskId} @@ -29,12 +31,13 @@ The current implementation uses Redis Stream: - stream key: `detection:jobs` - consumer group: `detection-workers` -- submitted lock prefix: `detection:jobs:submitted:` +- submitted event prefix: `detection:jobs:submitted:` - dead-letter stream key: `detection:jobs:dead-letter` -Redis keeps task submission outside the backend process, so the backend can be -restarted without losing queued work. Duplicate submissions are guarded with a -short-lived Redis lock per task id. +PostgreSQL is the source of truth for publication work. Redis keeps delivered +jobs outside the backend process, while a stable event id and short-lived Redis +key suppress immediate duplicate publication. Delivery remains at least once; +the detection task state is the final idempotency boundary. ## Reliability Behavior @@ -47,6 +50,10 @@ to the dead-letter stream and acknowledged in the main stream. This keeps a bad message from blocking the queue while preserving enough metadata for operations: - `taskId` +- `eventId` - `originalMessageId` - `deliveryCount` - `reason` + +See [Reliable Job Dispatch](reliable-job-dispatch.md) for outbox states, +configuration, failure scenarios, inspection, and replay. diff --git a/docs/project-worklog.md b/docs/project-worklog.md index c008f4b..ac53c8b 100644 --- a/docs/project-worklog.md +++ b/docs/project-worklog.md @@ -574,26 +574,72 @@ Redis publication before evaluation is connected to real inference. --- -## Next Recommended Work +### 2026-07-11: Reliable Detection Job Dispatch -Start Stage 1 from the production foundation design with a reliable dispatch -branch: +Branch: ```text feature/reliable-job-dispatch ``` +What changed: + +- Added Flyway migration V5 and a durable PostgreSQL outbox state machine. +- Changed asynchronous submission from a direct Redis write to transactional + outbox scheduling while locking the detection task against concurrent + submission. +- Added a versioned Redis event envelope, stable event-id deduplication, stale + claim recovery, bounded retry with jitter, and permanent failure state. +- Added bounded operations APIs for inspection and explicit terminal-event + replay without exposing raw payload JSON. +- Added a detailed reliability and recovery runbook. + +Why: + +- The previous request-to-Redis call was a database/queue dual write. +- Real systems must survive Redis outages and process crashes without losing + accepted business work. +- The same outbox foundation can later dispatch evaluation work without model + weights or additional infrastructure products. + +Verification: + +- TDD RED/GREEN cycles covered the domain state machine, transaction service, + Redis envelope, dispatcher, operations API, concurrent submission lock, and + indeterminate Redis responses. +- Java suite increased from 47 baseline tests to 79 passing tests. +- Frontend tests (8), lint, and production build passed. +- Model-service tests (6) and full-stack smoke tests (3) passed without model + weights. + +Deferred: + +- Real PostgreSQL/Redis restart and concurrency tests remain assigned to + `test/postgres-redis-testcontainers` when Docker is available. +- The model HTTP call still runs inside its database transaction and is the + next production-boundary change. +- Operations endpoints remain trusted-network only until authentication is in + scope. + +--- + +## Next Recommended Work + +Continue Stage 1 by shortening detection execution transactions: + +```text +feature/short-lived-execution-transactions +``` + Scope: -- Persist task dispatch requests in PostgreSQL in the same transaction as the - business command. -- Publish pending outbox records to Redis with a versioned event envelope. -- Add idempotent publication, bounded retry, stale-claim recovery, inspection, - and explicit replay behavior. -- Document queue failure and recovery operations. +- Claim a detection task in a short transaction. +- Invoke the model service outside any database transaction. +- Persist success or failure in a second short transaction. +- Add protection against stale execution attempts overwriting newer state. Reason: -The current database-to-Redis call is a dual write. Reliable dispatch provides -a reusable foundation for detection and evaluation, demonstrates a real -production consistency problem, and can be implemented without model weights. +`DetectionExecutionService` currently keeps a database transaction open across +the model HTTP call. Removing that boundary prevents long-held database +connections and locks while retaining durable attempt state. diff --git a/docs/reliable-job-dispatch.md b/docs/reliable-job-dispatch.md new file mode 100644 index 0000000..ce51242 --- /dev/null +++ b/docs/reliable-job-dispatch.md @@ -0,0 +1,172 @@ +# Reliable Job Dispatch + +This runbook describes how detection commands move durably from PostgreSQL to +Redis Streams and how to inspect or recover failed publication. + +## Guarantee + +The system provides at-least-once job delivery: + +- An accepted asynchronous submission has a detection task and outbox event in + PostgreSQL. +- Redis unavailability does not remove the durable publication request. +- Failed publication is retried with bounded exponential backoff and jitter. +- A process crash while publishing leaves a stale claim that another + dispatcher can recover. +- Duplicate publication or delivery is allowed. The detection task state is the + final idempotency boundary. + +The system does not claim exactly-once delivery. Redis and PostgreSQL do not +share a distributed transaction. + +## Data Flow + +```text +POST /api/detections/{taskId}/run-async + -> lock detection_task row + -> write/replay job_outbox event in the same PostgreSQL transaction + -> return 202 Accepted + +JobOutboxDispatcher + -> claim one PENDING event in a short PostgreSQL transaction + -> publish versioned event to Redis outside a database transaction + -> mark PUBLISHED in a new short transaction + -> or persist retry/FAILED state in a new short transaction + +DetectionJobWorker + -> read/claim Redis consumer-group message + -> run idempotent detection workflow + -> persist prediction/report/task state + -> acknowledge Redis message +``` + +The outbox dispatcher and detection worker are intentionally separate. A +published event can wait in Redis while model execution is unavailable. + +## Event Contract + +Redis stream `detection:jobs` receives these fields: + +| Field | Meaning | +| --- | --- | +| `eventId` | Stable outbox identifier used for transport deduplication | +| `eventVersion` | Contract version; currently `1` | +| `taskId` | Detection task identifier | +| `occurredAt` | UTC time at which the outbox event was created | + +Malformed events and unsupported versions move to +`detection:jobs:dead-letter`. Dead-letter entries retain the event id, task id, +original Redis message id, delivery count when available, and reason. + +## Outbox States + +| State | Meaning | +| --- | --- | +| `PENDING` | Waiting until `availableAt` and eligible for claiming | +| `PUBLISHING` | Claimed by a dispatcher; Redis publication may be in progress | +| `PUBLISHED` | Redis accepted the event and returned a stream record id | +| `FAILED` | Publication retry budget is exhausted; operator action required | + +`PUBLISHING` events older than `app.jobs.outbox.stale-after` return to +`PENDING`. Republishing is safe because Redis uses the stable event id as its +submitted-key and the task workflow is terminal-state idempotent. + +## Configuration + +| Environment variable | Default | Purpose | +| --- | --- | --- | +| `APP_JOBS_OUTBOX_DISPATCHER_ENABLED` | `true` | Enable PostgreSQL-to-Redis dispatcher | +| `APP_JOBS_OUTBOX_POLL_DELAY_MS` | `500` | Delay between dispatcher polls | +| `APP_JOBS_OUTBOX_STALE_AFTER` | `1m` | Age at which a publishing claim is recoverable | +| `APP_JOBS_OUTBOX_MAX_ATTEMPTS` | `5` | Publication attempt budget | +| `APP_JOBS_OUTBOX_BASE_RETRY_DELAY` | `1s` | Initial retry delay | +| `APP_JOBS_OUTBOX_MAX_RETRY_DELAY` | `1m` | Maximum retry delay | + +Redis consumer recovery continues to use the +`APP_DETECTION_JOBS_REDIS_*` settings documented in +[Async Detection Jobs](async-detection-jobs.md). + +## Inspect Events + +List the newest failed publication records: + +```powershell +Invoke-RestMethod ` + -Uri 'http://localhost:8080/api/operations/job-outbox?status=FAILED&limit=50' +``` + +Valid status filters are `PENDING`, `PUBLISHING`, `PUBLISHED`, and `FAILED`. +The limit must be between 1 and 100. Responses omit `payloadJson` so the +operations endpoint does not expose event data unnecessarily. + +The endpoint currently has no authentication because authentication and RBAC +are outside the MVP boundary. Do not expose it to the public internet. Restrict +the backend to a trusted development or server network until access control is +implemented. + +## Replay A Failed Event + +Inspect the event and task first. Then replay a terminal outbox event: + +```powershell +Invoke-RestMethod ` + -Method Post ` + -Uri 'http://localhost:8080/api/operations/job-outbox/event_xxx/replay' +``` + +Only `PUBLISHED` and `FAILED` events can be replayed. Replaying an active event +returns `409 Conflict`. Replaying a published event is appropriate when the +associated detection task is `FAILED` and an operator has confirmed a manual +retry. + +## Failure Scenarios + +### Redis unavailable during publication + +Expected behavior: + +1. The outbox event returns to `PENDING` with `lastError` populated. +2. `attemptCount` increments. +3. `availableAt` moves forward according to capped exponential backoff. +4. The detection task remains queryable in PostgreSQL. + +Restore Redis and wait for the next eligible attempt. Do not create a second +detection task merely to replace the lost publication. + +### Backend crashes after Redis accepts the event + +The outbox event may remain `PUBLISHING`. After the stale interval, another +dispatcher claims and republishes it. Redis suppresses the same event id while +its submitted-key exists. If the key has expired, the worker may receive a +duplicate; a completed detection task returns without creating another report. + +### Worker crashes after reading Redis + +The message remains in the consumer group's pending entries list. After +`app.detection.jobs.redis.pending-idle-timeout`, another consumer claims it. +Messages exceeding the Redis delivery budget move to the dead-letter stream. + +### Outbox reaches FAILED + +Check `lastError`, PostgreSQL health, Redis health, and configuration. Repair +the cause before calling the replay endpoint. A replay resets publication +attempt state; it does not alter the detection task result. + +## Verification Boundary + +This branch verifies state transitions, scheduling, event serialization, +dispatch orchestration, Redis adapter behavior, operations endpoints, and +Flyway/JPA compatibility with automated tests. Real PostgreSQL and Redis +failure/restart tests are deferred to `test/postgres-redis-testcontainers`, +because the current CI uses H2 for fast repository tests and does not provision +Docker services. + +Model weights, CUDA, and evaluation inference are unrelated to this dispatch +guarantee and remain deferred. + +Normal duplicate publication is suppressed by the stable event id, and a +duplicate delivered after task completion returns without repeating model +work. Atomic ownership when two independently injected messages target the +same non-terminal task is deferred to +`feature/short-lived-execution-transactions`. Until that work is merged, do +not manually append duplicate task messages to the Redis stream. diff --git a/docs/superpowers/plans/2026-07-11-reliable-job-dispatch.md b/docs/superpowers/plans/2026-07-11-reliable-job-dispatch.md new file mode 100644 index 0000000..4073061 --- /dev/null +++ b/docs/superpowers/plans/2026-07-11-reliable-job-dispatch.md @@ -0,0 +1,457 @@ +# Reliable Job Dispatch Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Persist detection dispatch requests transactionally in PostgreSQL and publish them to Redis Streams with recoverable, idempotent at-least-once delivery. + +**Architecture:** `DetectionJobService` writes a durable `JobOutboxEvent` instead of calling Redis. A scheduled dispatcher claims one event in a short transaction, publishes a versioned `DetectionJobRequest`, and records success or bounded retry state in a second transaction. PostgreSQL is authoritative; Redis remains the delivery transport, and the detection worker remains idempotent through task terminal-state checks. + +**Tech Stack:** Java 21, Spring Boot 3.5, Spring Data JPA, Flyway, PostgreSQL, Redis Streams, Jackson, JUnit 5, AssertJ, Mockito. + +## Global Constraints + +- Keep the Java backend as a modular monolith; do not add Kafka, CDC, or a new deployable service. +- PostgreSQL owns business and outbox state; Redis owns transport state only. +- Delivery semantics are at least once, never described as exactly once. +- Do not hold a database transaction while publishing to Redis. +- Use Flyway forward migrations; do not edit merged migrations. +- Do not change frontend behavior or download model weights in this branch. +- Develop every production behavior test-first and observe the expected failure before implementation. +- Preserve existing API routes and the current `202 Accepted` response body. + +--- + +## File Structure + +New production files: + +- `domain/JobOutboxEvent.java`: durable event aggregate and state transitions. +- `domain/JobOutboxStatus.java`: `PENDING`, `PUBLISHING`, `PUBLISHED`, `FAILED`. +- `domain/JobOutboxEventType.java`: initially `DETECTION_REQUESTED`. +- `repository/JobOutboxEventRepository.java`: locked claiming and operations queries. +- `config/JobOutboxProperties.java`: retry and stale-claim configuration. +- `service/OutboxBackoffPolicy.java`: capped exponential delay with deterministic jitter. +- `service/JobOutboxService.java`: scheduling, claiming, completion, failure, and replay transactions. +- `service/JobOutboxPublisher.java`: converts outbox records into queue requests. +- `service/JobOutboxDispatcher.java`: scheduled orchestration outside database transactions. +- `service/DetectionJobRequest.java`: versioned Redis event payload. +- `dto/operations/JobOutboxEventResponse.java`: bounded operations response. +- `controller/JobOutboxOperationsController.java`: inspect and replay endpoints. +- `db/migration/V5__add_job_outbox.sql`: schema, constraints, and claim indexes. +- `docs/reliable-job-dispatch.md`: behavior and recovery runbook. + +Modified production files: + +- `service/DetectionJobService.java`: schedule durable work instead of writing Redis. +- `service/DetectionJobQueue.java`: accept `DetectionJobRequest`. +- `service/DetectionJobMessage.java`: expose event id and version to the consumer boundary. +- `service/RedisDetectionJobQueue.java`: publish and read the versioned envelope. +- `application.yml`: outbox defaults and separate dispatcher scheduling. +- `docs/README.md` and `docs/project-worklog.md`: link and record the feature. + +## Task 1: Outbox Schema And Domain State Machine + +**Files:** + +- Create: `backend-java/src/main/resources/db/migration/V5__add_job_outbox.sql` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxStatus.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEventType.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/domain/JobOutboxEvent.java` +- Create: `backend-java/src/test/java/com/fengting/aigcforensics/domain/JobOutboxEventTest.java` + +**Interfaces:** + +- Produces: `JobOutboxEvent.detectionRequested(String eventId, String taskId, String payloadJson, Instant now)`. +- Produces: `claim`, `markPublished`, `markPublishFailed`, `recoverStaleClaim`, and `replay` state transitions. + +- [ ] **Step 1: Write failing state-transition tests** + +```java +@Test +void claimsPendingEventWhenAvailable() { + JobOutboxEvent event = eventAt("2026-07-11T00:00:00Z"); + + event.claim(Instant.parse("2026-07-11T00:00:01Z")); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PUBLISHING); + assertThat(event.getAttemptCount()).isEqualTo(1); +} + +@Test +void schedulesRetryAfterPublishFailure() { + JobOutboxEvent event = claimedEvent(); + Instant retryAt = Instant.parse("2026-07-11T00:00:10Z"); + + event.markPublishFailed("redis unavailable", retryAt, 5, retryAt); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.PENDING); + assertThat(event.getAvailableAt()).isEqualTo(retryAt); +} + +@Test +void marksEventFailedAfterRetryBudgetIsExhausted() { + JobOutboxEvent event = eventClaimedFiveTimes(); + + event.markPublishFailed("redis unavailable", Instant.now(), 5, Instant.now()); + + assertThat(event.getStatus()).isEqualTo(JobOutboxStatus.FAILED); +} +``` + +- [ ] **Step 2: Verify RED** + +Run: + +```powershell +cd backend-java +mvn -B -Dtest=JobOutboxEventTest test +``` + +Expected: compilation fails because `JobOutboxEvent` does not exist. + +- [ ] **Step 3: Add the migration and minimal domain implementation** + +The migration creates `job_outbox` with a unique `event_id`, a unique +`(event_type, aggregate_id)` pair, status and scheduling fields, timestamps, +and `lock_version`. Add indexes on `(status, available_at, created_at)` and +`(status, updated_at)`. + +State methods reject invalid transitions with `IllegalStateException`, cap +`last_error` at 2048 characters, and update `updated_at` on every transition. + +- [ ] **Step 4: Verify GREEN** + +Run the focused test and then: + +```powershell +mvn -B test +``` + +Expected: all Java tests pass. + +- [ ] **Step 5: Commit** + +```powershell +git add backend-java/src/main/resources/db/migration/V5__add_job_outbox.sql backend-java/src/main/java/com/fengting/aigcforensics/domain backend-java/src/test/java/com/fengting/aigcforensics/domain/JobOutboxEventTest.java +git commit -m "feat: add durable job outbox state" +``` + +## Task 2: Repository, Backoff, And Transactional Scheduling + +**Files:** + +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/repository/JobOutboxEventRepository.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/config/JobOutboxProperties.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/service/OutboxBackoffPolicy.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxService.java` +- Create: `backend-java/src/test/java/com/fengting/aigcforensics/service/OutboxBackoffPolicyTest.java` +- Create: `backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxServiceTest.java` +- Modify: `backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobService.java` +- Modify: `backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobServiceTest.java` + +**Interfaces:** + +- Produces: `JobOutboxService.scheduleDetection(String taskId)`. +- Produces: `Optional claimNext()`. +- Produces: `markPublished`, `markPublishFailed`, and `replay` methods using `REQUIRES_NEW` transactions. +- Consumes: `DetectionJobService.submit(String taskId)` eligibility rules. + +- [ ] **Step 1: Write failing backoff tests** + +```java +@Test +void doublesDelayAndCapsAtConfiguredMaximum() { + OutboxBackoffPolicy policy = new OutboxBackoffPolicy(Duration.ofSeconds(1), Duration.ofSeconds(8)); + + assertThat(policy.delay("event-1", 1)).isBetween(Duration.ofMillis(800), Duration.ofMillis(1200)); + assertThat(policy.delay("event-1", 8)).isLessThanOrEqualTo(Duration.ofMillis(9600)); +} +``` + +- [ ] **Step 2: Verify RED, implement deterministic jitter, and verify GREEN** + +Use an event-id hash to produce a stable factor between `0.8` and `1.2`; this +keeps tests deterministic while preventing all events from retrying together. + +- [ ] **Step 3: Write failing scheduling and claiming tests** + +Tests prove that scheduling writes one event, duplicate scheduling reuses the +same event, failed-task submission replays the existing event, claiming uses a +pessimistic lock query, and stale `PUBLISHING` records become claimable. + +- [ ] **Step 4: Verify RED** + +Expected: `DetectionJobServiceTest` fails because it still depends on +`DetectionJobQueue` and `JobOutboxService` is absent. + +- [ ] **Step 5: Implement repository and service** + +Repository signatures: + +```java +Optional findByEventId(String eventId); +Optional findByEventTypeAndAggregateId(JobOutboxEventType eventType, String aggregateId); + +@Lock(LockModeType.PESSIMISTIC_WRITE) +@Query("select event from JobOutboxEvent event where event.status = :status and event.availableAt <= :now order by event.createdAt") +List findClaimable(JobOutboxStatus status, Instant now, Pageable pageable); + +List findByStatusOrderByCreatedAtDesc(JobOutboxStatus status, Pageable pageable); +``` + +`scheduleDetection` serializes `{"taskId":"..."}` through `ObjectMapper`, +creates a stable `event_` UUID for a new task, and calls `replay(now)` only when +the existing event is terminal and the detection task is eligible for retry. + +- [ ] **Step 6: Change detection submission to the outbox boundary** + +```java +@Transactional +public DetectionTask submit(String taskId) { + DetectionTask task = findTask(taskId); + if (task.getStatus() == DetectionStatus.QUEUED || task.getStatus() == DetectionStatus.FAILED) { + jobOutboxService.scheduleDetection(taskId); + } + return task; +} +``` + +- [ ] **Step 7: Verify focused and full Java tests** + +```powershell +mvn -B -Dtest=OutboxBackoffPolicyTest,JobOutboxServiceTest,DetectionJobServiceTest test +mvn -B test +``` + +- [ ] **Step 8: Commit** + +```powershell +git add backend-java/src/main backend-java/src/test +git commit -m "feat: schedule detection jobs transactionally" +``` + +## Task 3: Versioned Redis Event Envelope + +**Files:** + +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobRequest.java` +- Modify: `backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobQueue.java` +- Modify: `backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobMessage.java` +- Modify: `backend-java/src/main/java/com/fengting/aigcforensics/service/RedisDetectionJobQueue.java` +- Modify: `backend-java/src/test/java/com/fengting/aigcforensics/service/RedisDetectionJobQueueTest.java` +- Modify: `backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionJobWorkerTest.java` + +**Interfaces:** + +- Produces: `DetectionJobRequest(String eventId, int eventVersion, String taskId, Instant occurredAt)`. +- Produces: Redis fields `eventId`, `eventVersion`, `taskId`, and `occurredAt`. +- Preserves: `DetectionJobWorker` executes by `taskId` and acknowledges only after persistence. + +- [ ] **Step 1: Write failing Redis queue tests** + +Tests capture the added stream record and assert all four fields. Add tests for +missing event id, unsupported event version, duplicate event id, and dead-letter +metadata retaining the original event id. + +- [ ] **Step 2: Verify RED** + +```powershell +mvn -B -Dtest=RedisDetectionJobQueueTest,DetectionJobWorkerTest test +``` + +Expected: compilation fails against the old `enqueue(String taskId)` contract. + +- [ ] **Step 3: Implement the envelope** + +Use the event id for the Redis submitted-key deduplication key. Validate +`eventVersion == 1` when reading. Invalid envelopes are acknowledged and copied +to the dead-letter stream with a stable reason rather than passed to the +worker. + +- [ ] **Step 4: Verify GREEN and full Java suite** + +```powershell +mvn -B -Dtest=RedisDetectionJobQueueTest,DetectionJobWorkerTest test +mvn -B test +``` + +- [ ] **Step 5: Commit** + +```powershell +git add backend-java/src/main/java/com/fengting/aigcforensics/service backend-java/src/test/java/com/fengting/aigcforensics/service +git commit -m "feat: version detection queue events" +``` + +## Task 4: Dispatcher And Publish Recovery + +**Files:** + +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxPublisher.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxDispatcher.java` +- Create: `backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxPublisherTest.java` +- Create: `backend-java/src/test/java/com/fengting/aigcforensics/service/JobOutboxDispatcherTest.java` +- Modify: `backend-java/src/main/resources/application.yml` +- Modify: `backend-java/src/test/resources/application-test.yml` + +**Interfaces:** + +- `JobOutboxPublisher.publish(JobOutboxEvent event)` supports + `DETECTION_REQUESTED` and rejects unknown event types. +- `JobOutboxDispatcher.pollOnce()` owns no transaction and coordinates the + short transactional methods on `JobOutboxService`. + +- [ ] **Step 1: Write failing publisher tests** + +Tests prove payload parsing, version `1`, event metadata propagation, and +malformed payload rejection. + +- [ ] **Step 2: Verify RED and implement publisher** + +Parse payload with Jackson into a private `DetectionRequestedPayload` record +and call `DetectionJobQueue.enqueue(DetectionJobRequest)`. + +- [ ] **Step 3: Write failing dispatcher tests** + +```java +@Test +void marksClaimedEventPublishedAfterQueueAcceptsIt() { + when(outboxService.claimNext()).thenReturn(Optional.of(event)); + + dispatcher.pollOnce(); + + verify(publisher).publish(event); + verify(outboxService).markPublished(event.getEventId()); +} + +@Test +void recordsRetryableFailureWhenPublishThrows() { + when(outboxService.claimNext()).thenReturn(Optional.of(event)); + doThrow(new IllegalStateException("redis unavailable")).when(publisher).publish(event); + + dispatcher.pollOnce(); + + verify(outboxService).markPublishFailed(event.getEventId(), "redis unavailable"); +} +``` + +- [ ] **Step 4: Verify RED and implement dispatcher** + +Enable scheduling with `app.jobs.outbox.dispatcher-enabled`; disable it in the +test profile. Do not annotate `pollOnce` with `@Transactional`. + +- [ ] **Step 5: Verify GREEN and full Java suite** + +- [ ] **Step 6: Commit** + +```powershell +git add backend-java/src/main/java/com/fengting/aigcforensics/service backend-java/src/main/resources/application.yml backend-java/src/test +git commit -m "feat: dispatch outbox jobs to redis" +``` + +## Task 5: Inspection And Explicit Replay API + +**Files:** + +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/dto/operations/JobOutboxEventResponse.java` +- Create: `backend-java/src/main/java/com/fengting/aigcforensics/controller/JobOutboxOperationsController.java` +- Create: `backend-java/src/test/java/com/fengting/aigcforensics/controller/JobOutboxOperationsControllerTest.java` +- Modify: `backend-java/src/main/java/com/fengting/aigcforensics/service/JobOutboxService.java` + +**Interfaces:** + +- `GET /api/operations/job-outbox?status=FAILED&limit=50` returns at most 100 + newest matching records. +- `POST /api/operations/job-outbox/{eventId}/replay` returns the replayed event. + +- [ ] **Step 1: Write failing MVC tests** + +Test valid filtering, default limit 50, limit rejection above 100, missing event +404, valid replay, and invalid-state rejection. + +- [ ] **Step 2: Verify RED** + +```powershell +mvn -B -Dtest=JobOutboxOperationsControllerTest test +``` + +- [ ] **Step 3: Implement bounded query, DTO mapping, and replay** + +The DTO exposes identifiers, type, aggregate, status, attempts, scheduling and +publication timestamps, and the last sanitized error. It never exposes raw +payload JSON. + +- [ ] **Step 4: Verify GREEN and full Java suite** + +- [ ] **Step 5: Commit** + +```powershell +git add backend-java/src/main backend-java/src/test +git commit -m "feat: expose outbox recovery operations" +``` + +## Task 6: Runbook, Worklog, And Full Verification + +**Files:** + +- Create: `docs/reliable-job-dispatch.md` +- Modify: `docs/README.md` +- Modify: `docs/async-detection-jobs.md` +- Modify: `docs/project-worklog.md` + +**Interfaces:** + +- Produces: operator commands and failure semantics for local, WSL, and server + environments. + +- [ ] **Step 1: Write the runbook** + +Document the submission, outbox, dispatcher, Redis, worker, and acknowledgement +sequence. Include SQL/HTTP inspection, replay, stale-claim recovery, retry +budget, dead-letter behavior, and explicit at-least-once semantics. + +- [ ] **Step 2: Update durable docs** + +Link the runbook, record branch purpose and verification, and replace any text +that still claims direct request-to-Redis submission. + +- [ ] **Step 3: Run fresh verification** + +```powershell +npm run test +npm run lint +npm run build +cd backend-java +mvn -B test +cd ..\model-services\nonescape-mini +& 'D:\workspace\develop\.venv-model-service\Scripts\python.exe' -m pytest +cd ..\.. +python -m pytest tools/tests +git diff --check +``` + +Expected: 0 failures in every suite and no whitespace errors. + +- [ ] **Step 4: Review the final diff against the design** + +Check specifically for database transactions around Redis publication, raw +payload leakage, unbounded error strings, duplicate-event behavior, missing +indexes, unsupported event versions, and undocumented failure states. + +- [ ] **Step 5: Commit documentation** + +```powershell +git add docs +git commit -m "docs: explain reliable job dispatch" +``` + +- [ ] **Step 6: Push, open a ready PR, wait for CI, and merge only when all required checks succeed** + +PR title: + +```text +Add reliable detection job dispatch +``` + +The PR body must state the at-least-once guarantee, transactional boundary, +recovery behavior, test evidence, deferred Testcontainers coverage, and absence +of model-weight changes.