From 4006669063b25afa498b27e41cbd39cdb82e919a Mon Sep 17 00:00:00 2001 From: fengting124 Date: Sat, 11 Jul 2026 07:29:40 +0800 Subject: [PATCH 1/6] docs: plan short-lived execution transactions --- ...7-11-short-lived-execution-transactions.md | 126 ++++++++++++++++++ 1 file changed, 126 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-11-short-lived-execution-transactions.md diff --git a/docs/superpowers/plans/2026-07-11-short-lived-execution-transactions.md b/docs/superpowers/plans/2026-07-11-short-lived-execution-transactions.md new file mode 100644 index 0000000..00c6254 --- /dev/null +++ b/docs/superpowers/plans/2026-07-11-short-lived-execution-transactions.md @@ -0,0 +1,126 @@ +# Short-Lived Detection Execution Transactions + +**Branch:** `feature/short-lived-execution-transactions` + +## Goal + +Remove model HTTP calls from database transactions while preserving durable, +race-safe detection state. A stale model response must never overwrite a newer +execution attempt. + +## Current Risk + +`DetectionExecutionService.runDetection` is transactional around task reads, +all model HTTP calls, prediction writes, report creation, and the final task +update. A slow or unavailable model therefore holds a database connection and +transaction for the full network timeout. Concurrent deliveries also have no +durable execution ownership token. + +## Transaction Model + +```text +short transaction A + -> lock detection_task + -> return terminal/busy when no execution should start + -> otherwise assign execution_token, increment attempt count, + set execution_lease_until, and mark INFERENCING + -> snapshot asset and enabled model configuration + +no database transaction + -> call each model endpoint + -> collect immutable inference results in memory + +short transaction B + -> lock detection_task + -> compare execution_token + -> stale token: discard the complete result set + -> matching token: persist predictions and report atomically, then COMPLETE + +short failure transaction + -> lock detection_task + -> matching token: mark FAILED + -> stale token: preserve the newer attempt +``` + +The execution lease is not an exactly-once mechanism. It permits recovery when +a process dies after claiming a task. The token comparison is the fencing +mechanism that prevents an expired attempt from committing after a replacement +attempt has started. + +## Behavioral Rules + +- `COMPLETED` tasks are terminal and do not execute again. +- `QUEUED` and `FAILED` tasks may start a new attempt. +- `INFERENCING` with a live lease is busy and must not be acknowledged by the + asynchronous worker. +- `INFERENCING` with an expired lease may be reclaimed with a new token. +- Missing enabled models fail the claimed task without making a network call. +- Predictions and the report commit in one transaction. +- Partial model success is not persisted when a later model fails. +- A stale success or failure result performs no writes. +- Synchronous callers receive the current task snapshot when work is already + active; asynchronous messages remain pending for later terminal cleanup. + +## Schema + +Add Flyway V6 fields to `detection_task`: + +- `execution_token varchar(64)` +- `execution_lease_until timestamp with time zone` +- `execution_attempt_count integer not null default 0` + +Add an index on `(status, execution_lease_until)` for recovery scans and future +operations tooling. + +## Java Boundaries + +- `DetectionExecutionService`: non-transactional coordinator only. +- `DetectionExecutionTransactionService`: public transactional claim, + completion, and failure methods. Keeping it as a separate Spring bean avoids + proxy self-invocation mistakes. +- `DetectionExecutionPlan`: immutable snapshot used outside persistence scope. +- `DetectionExecutionOutcome`: tells queue workers whether a message is safe + to acknowledge. + +## TDD Sequence + +1. Add domain tests for claim, live-lease rejection, expired-lease reclaim, + token match, and stale-token rejection. +2. Add V6 and map the new fields on `DetectionTask`. +3. Add transaction-service tests for claim snapshots and fenced completion. +4. Refactor the coordinator and prove the model client runs without an active + Spring transaction. +5. Update the worker so busy work remains unacknowledged while terminal, + successful, and failed attempts are acknowledged. +6. Add operator documentation and worklog evidence. + +## Verification + +Run before opening the PR: + +```powershell +cd backend-java +mvn -B test + +cd .. +npm run test +npm run lint +npm run build + +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 +``` + +Then push this branch, open one focused PR, wait for all CI jobs, and squash +merge only when the PR is clean and green. + +## Deferred + +- Model weights and GPU runtime setup. +- Lease heartbeat for model calls longer than the configured lease. +- Docker-backed crash/restart tests with PostgreSQL and Redis. +- Applying the same pattern to batch evaluation execution. From 28aceb8eb0c92909453eff9cfea0f727cf6f2179 Mon Sep 17 00:00:00 2001 From: fengting124 Date: Sat, 11 Jul 2026 07:32:08 +0800 Subject: [PATCH 2/6] feat: add fenced detection execution leases --- .../aigcforensics/domain/DetectionTask.java | 84 +++++++++++++++++ .../V6__add_detection_execution_lease.sql | 6 ++ .../domain/DetectionTaskTest.java | 92 +++++++++++++++++++ 3 files changed, 182 insertions(+) create mode 100644 backend-java/src/main/resources/db/migration/V6__add_detection_execution_lease.sql create mode 100644 backend-java/src/test/java/com/fengting/aigcforensics/domain/DetectionTaskTest.java diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/domain/DetectionTask.java b/backend-java/src/main/java/com/fengting/aigcforensics/domain/DetectionTask.java index fe9db51..515d206 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/domain/DetectionTask.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/domain/DetectionTask.java @@ -1,6 +1,7 @@ package com.fengting.aigcforensics.domain; import java.time.Instant; +import java.util.Objects; import jakarta.persistence.Column; import jakarta.persistence.Entity; @@ -41,6 +42,15 @@ public class DetectionTask { @Column(name = "completed_at") private Instant completedAt; + @Column(name = "execution_token", length = 64) + private String executionToken; + + @Column(name = "execution_lease_until") + private Instant executionLeaseUntil; + + @Column(name = "execution_attempt_count", nullable = false) + private int executionAttemptCount; + protected DetectionTask() { } @@ -59,12 +69,74 @@ public void markStarted(Instant startedAt) { public void markCompleted(Instant completedAt) { this.status = DetectionStatus.COMPLETED; this.completedAt = completedAt; + clearExecutionOwnership(); } public void markFailed(String failureReason, Instant completedAt) { this.status = DetectionStatus.FAILED; this.failureReason = failureReason; this.completedAt = completedAt; + clearExecutionOwnership(); + } + + public boolean claimExecution(String token, Instant now, Instant leaseUntil) { + requireValidClaim(token, now, leaseUntil); + if (status == DetectionStatus.COMPLETED || hasLiveExecutionLease(now)) { + return false; + } + + status = DetectionStatus.INFERENCING; + executionToken = token; + executionLeaseUntil = leaseUntil; + executionAttemptCount++; + startedAt = now; + completedAt = null; + failureReason = null; + return true; + } + + public boolean completeExecution(String token, Instant completedAt) { + if (!ownsExecution(token)) { + return false; + } + markCompleted(Objects.requireNonNull(completedAt, "completedAt must not be null")); + return true; + } + + public boolean failExecution(String token, String reason, Instant completedAt) { + if (!ownsExecution(token)) { + return false; + } + markFailed(reason, Objects.requireNonNull(completedAt, "completedAt must not be null")); + return true; + } + + public boolean ownsExecution(String token) { + return status == DetectionStatus.INFERENCING + && executionToken != null + && executionToken.equals(token); + } + + private boolean hasLiveExecutionLease(Instant now) { + return status == DetectionStatus.INFERENCING + && executionLeaseUntil != null + && executionLeaseUntil.isAfter(now); + } + + private void requireValidClaim(String token, Instant now, Instant leaseUntil) { + if (token == null || token.isBlank()) { + throw new IllegalArgumentException("execution token must not be blank"); + } + Objects.requireNonNull(now, "now must not be null"); + Objects.requireNonNull(leaseUntil, "leaseUntil must not be null"); + if (!leaseUntil.isAfter(now)) { + throw new IllegalArgumentException("execution lease must expire after claim time"); + } + } + + private void clearExecutionOwnership() { + executionToken = null; + executionLeaseUntil = null; } public Long getId() { @@ -98,4 +170,16 @@ public Instant getStartedAt() { public Instant getCompletedAt() { return completedAt; } + + public String getExecutionToken() { + return executionToken; + } + + public Instant getExecutionLeaseUntil() { + return executionLeaseUntil; + } + + public int getExecutionAttemptCount() { + return executionAttemptCount; + } } diff --git a/backend-java/src/main/resources/db/migration/V6__add_detection_execution_lease.sql b/backend-java/src/main/resources/db/migration/V6__add_detection_execution_lease.sql new file mode 100644 index 0000000..ee18cad --- /dev/null +++ b/backend-java/src/main/resources/db/migration/V6__add_detection_execution_lease.sql @@ -0,0 +1,6 @@ +alter table detection_task add column execution_token varchar(64); +alter table detection_task add column execution_lease_until timestamp with time zone; +alter table detection_task add column execution_attempt_count integer not null default 0; + +create index idx_detection_task_execution_lease + on detection_task(status, execution_lease_until); diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/domain/DetectionTaskTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/domain/DetectionTaskTest.java new file mode 100644 index 0000000..e84f760 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/domain/DetectionTaskTest.java @@ -0,0 +1,92 @@ +package com.fengting.aigcforensics.domain; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Instant; + +import org.junit.jupiter.api.Test; + +class DetectionTaskTest { + + private static final Instant NOW = Instant.parse("2026-07-11T00:00:00Z"); + + @Test + void claimsQueuedTaskAndRecordsAttemptLease() { + DetectionTask task = queuedTask(); + + boolean claimed = task.claimExecution("attempt-1", NOW, NOW.plusSeconds(60)); + + assertThat(claimed).isTrue(); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.INFERENCING); + assertThat(task.getExecutionToken()).isEqualTo("attempt-1"); + assertThat(task.getExecutionLeaseUntil()).isEqualTo(NOW.plusSeconds(60)); + assertThat(task.getExecutionAttemptCount()).isEqualTo(1); + } + + @Test + void rejectsClaimWhileExecutionLeaseIsLive() { + DetectionTask task = queuedTask(); + task.claimExecution("attempt-1", NOW, NOW.plusSeconds(60)); + + boolean claimed = task.claimExecution( + "attempt-2", + NOW.plusSeconds(30), + NOW.plusSeconds(90)); + + assertThat(claimed).isFalse(); + assertThat(task.getExecutionToken()).isEqualTo("attempt-1"); + assertThat(task.getExecutionAttemptCount()).isEqualTo(1); + } + + @Test + void reclaimsExpiredExecutionWithNewFencingToken() { + DetectionTask task = queuedTask(); + task.claimExecution("attempt-1", NOW, NOW.plusSeconds(60)); + + boolean claimed = task.claimExecution( + "attempt-2", + NOW.plusSeconds(61), + NOW.plusSeconds(121)); + + assertThat(claimed).isTrue(); + assertThat(task.getExecutionToken()).isEqualTo("attempt-2"); + assertThat(task.getExecutionAttemptCount()).isEqualTo(2); + } + + @Test + void staleAttemptCannotCompleteOrFailNewerExecution() { + DetectionTask task = queuedTask(); + task.claimExecution("attempt-1", NOW, NOW.plusSeconds(60)); + task.claimExecution("attempt-2", NOW.plusSeconds(61), NOW.plusSeconds(121)); + + assertThat(task.completeExecution("attempt-1", NOW.plusSeconds(70))).isFalse(); + assertThat(task.failExecution("attempt-1", "late failure", NOW.plusSeconds(70))).isFalse(); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.INFERENCING); + assertThat(task.getExecutionToken()).isEqualTo("attempt-2"); + } + + @Test + void matchingAttemptCompletesAndClearsLease() { + DetectionTask task = queuedTask(); + task.claimExecution("attempt-1", NOW, NOW.plusSeconds(60)); + + boolean completed = task.completeExecution("attempt-1", NOW.plusSeconds(10)); + + assertThat(completed).isTrue(); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.COMPLETED); + assertThat(task.getExecutionToken()).isNull(); + assertThat(task.getExecutionLeaseUntil()).isNull(); + } + + @Test + void completedTaskCannotBeClaimedAgain() { + DetectionTask task = queuedTask(); + task.markCompleted(NOW); + + assertThat(task.claimExecution("attempt-1", NOW, NOW.plusSeconds(60))).isFalse(); + } + + private DetectionTask queuedTask() { + return new DetectionTask("task-1", "asset-1", DetectionStatus.QUEUED, NOW.minusSeconds(10)); + } +} From 12b03f946f7ea86ec9d26f0df8bd80db174599f5 Mon Sep 17 00:00:00 2001 From: fengting124 Date: Sat, 11 Jul 2026 07:35:10 +0800 Subject: [PATCH 3/6] feat: add short detection transaction boundaries --- .../service/DetectionExecutionClaim.java | 17 ++ .../DetectionExecutionClaimStatus.java | 8 + .../service/DetectionExecutionOutcome.java | 13 ++ .../service/DetectionExecutionPlan.java | 16 ++ .../DetectionExecutionTransactionService.java | 218 ++++++++++++++++++ .../service/DetectionModelResult.java | 8 + .../service/DetectionModelTarget.java | 7 + ...ectionExecutionTransactionServiceTest.java | 196 ++++++++++++++++ 8 files changed, 483 insertions(+) create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaim.java create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaimStatus.java create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionOutcome.java create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionPlan.java create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionService.java create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelResult.java create mode 100644 backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelTarget.java create mode 100644 backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaim.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaim.java new file mode 100644 index 0000000..9869335 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaim.java @@ -0,0 +1,17 @@ +package com.fengting.aigcforensics.service; + +public record DetectionExecutionClaim( + DetectionExecutionClaimStatus status, + DetectionExecutionPlan plan) { + + public static DetectionExecutionClaim claimed(DetectionExecutionPlan plan) { + return new DetectionExecutionClaim(DetectionExecutionClaimStatus.CLAIMED, plan); + } + + public static DetectionExecutionClaim withoutPlan(DetectionExecutionClaimStatus status) { + if (status == DetectionExecutionClaimStatus.CLAIMED) { + throw new IllegalArgumentException("claimed execution requires a plan"); + } + return new DetectionExecutionClaim(status, null); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaimStatus.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaimStatus.java new file mode 100644 index 0000000..8a85aac --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionClaimStatus.java @@ -0,0 +1,8 @@ +package com.fengting.aigcforensics.service; + +public enum DetectionExecutionClaimStatus { + CLAIMED, + BUSY, + TERMINAL, + FAILED +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionOutcome.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionOutcome.java new file mode 100644 index 0000000..270c6c8 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionOutcome.java @@ -0,0 +1,13 @@ +package com.fengting.aigcforensics.service; + +public enum DetectionExecutionOutcome { + COMPLETED, + FAILED, + TERMINAL, + BUSY, + STALE; + + public boolean shouldAcknowledge() { + return this != BUSY; + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionPlan.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionPlan.java new file mode 100644 index 0000000..dad694c --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionPlan.java @@ -0,0 +1,16 @@ +package com.fengting.aigcforensics.service; + +import java.nio.file.Path; +import java.util.List; + +public record DetectionExecutionPlan( + String executionToken, + String taskId, + String assetId, + Path assetPath, + List models) { + + public DetectionExecutionPlan { + models = List.copyOf(models); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionService.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionService.java new file mode 100644 index 0000000..dc1b373 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionService.java @@ -0,0 +1,218 @@ +package com.fengting.aigcforensics.service; + +import java.nio.file.Path; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Comparator; +import java.util.List; +import java.util.UUID; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import com.fengting.aigcforensics.domain.DetectionReport; +import com.fengting.aigcforensics.domain.DetectionStatus; +import com.fengting.aigcforensics.domain.DetectionTask; +import com.fengting.aigcforensics.domain.MediaAsset; +import com.fengting.aigcforensics.domain.ModelLabel; +import com.fengting.aigcforensics.domain.ModelPrediction; +import com.fengting.aigcforensics.domain.ModelRegistry; +import com.fengting.aigcforensics.domain.ReportVerdict; +import com.fengting.aigcforensics.domain.RiskLevel; +import com.fengting.aigcforensics.repository.DetectionReportRepository; +import com.fengting.aigcforensics.repository.DetectionTaskRepository; +import com.fengting.aigcforensics.repository.MediaAssetRepository; +import com.fengting.aigcforensics.repository.ModelPredictionRepository; +import com.fengting.aigcforensics.repository.ModelRegistryRepository; + +@Service +public class DetectionExecutionTransactionService { + + private final DetectionTaskRepository taskRepository; + private final MediaAssetRepository assetRepository; + private final ModelRegistryRepository modelRepository; + private final ModelPredictionRepository predictionRepository; + private final DetectionReportRepository reportRepository; + private final Duration leaseDuration; + private final Clock clock; + + @Autowired + public DetectionExecutionTransactionService( + DetectionTaskRepository taskRepository, + MediaAssetRepository assetRepository, + ModelRegistryRepository modelRepository, + ModelPredictionRepository predictionRepository, + DetectionReportRepository reportRepository, + @Value("${app.detection.execution.lease-duration:5m}") Duration leaseDuration) { + this( + taskRepository, + assetRepository, + modelRepository, + predictionRepository, + reportRepository, + leaseDuration, + Clock.systemUTC()); + } + + DetectionExecutionTransactionService( + DetectionTaskRepository taskRepository, + MediaAssetRepository assetRepository, + ModelRegistryRepository modelRepository, + ModelPredictionRepository predictionRepository, + DetectionReportRepository reportRepository, + Duration leaseDuration, + Clock clock) { + if (leaseDuration.isZero() || leaseDuration.isNegative()) { + throw new IllegalArgumentException("execution lease duration must be positive"); + } + this.taskRepository = taskRepository; + this.assetRepository = assetRepository; + this.modelRepository = modelRepository; + this.predictionRepository = predictionRepository; + this.reportRepository = reportRepository; + this.leaseDuration = leaseDuration; + this.clock = clock; + } + + @Transactional + public DetectionExecutionClaim claim(String taskId) { + DetectionTask task = findTaskForUpdate(taskId); + if (task.getStatus() == DetectionStatus.COMPLETED) { + return DetectionExecutionClaim.withoutPlan(DetectionExecutionClaimStatus.TERMINAL); + } + + Instant now = Instant.now(clock); + String token = newExternalId("execution"); + if (!task.claimExecution(token, now, now.plus(leaseDuration))) { + return DetectionExecutionClaim.withoutPlan(DetectionExecutionClaimStatus.BUSY); + } + + MediaAsset asset = assetRepository.findByAssetId(task.getAssetId()) + .orElseThrow(() -> new ResourceNotFoundException("Media asset not found: " + task.getAssetId())); + List models = modelRepository.findByEnabledTrueOrderByWeightDesc(); + if (models.isEmpty()) { + task.failExecution(token, "No enabled model is available", now); + return DetectionExecutionClaim.withoutPlan(DetectionExecutionClaimStatus.FAILED); + } + + List targets = models.stream() + .map(model -> new DetectionModelTarget( + model.getModelId(), + model.getEndpointUrl(), + model.getDefaultThreshold())) + .toList(); + return DetectionExecutionClaim.claimed(new DetectionExecutionPlan( + token, + task.getTaskId(), + asset.getAssetId(), + Path.of(asset.getStoragePath()), + targets)); + } + + @Transactional + public DetectionExecutionOutcome complete( + String taskId, + String executionToken, + List results) { + if (results.isEmpty()) { + throw new IllegalArgumentException("detection completion requires at least one model result"); + } + + DetectionTask task = findTaskForUpdate(taskId); + if (!task.ownsExecution(executionToken)) { + return DetectionExecutionOutcome.STALE; + } + + Instant now = Instant.now(clock); + List predictions = results.stream() + .map(result -> toPrediction(taskId, result, now)) + .toList(); + predictionRepository.saveAll(predictions); + reportRepository.save(buildReport(taskId, results, now)); + task.completeExecution(executionToken, now); + return DetectionExecutionOutcome.COMPLETED; + } + + @Transactional + public DetectionExecutionOutcome fail(String taskId, String executionToken, String failureReason) { + DetectionTask task = findTaskForUpdate(taskId); + if (!task.failExecution(executionToken, failureReason, Instant.now(clock))) { + return DetectionExecutionOutcome.STALE; + } + return DetectionExecutionOutcome.FAILED; + } + + private DetectionTask findTaskForUpdate(String taskId) { + return taskRepository.findByTaskIdForUpdate(taskId) + .orElseThrow(() -> new ResourceNotFoundException("Detection task not found: " + taskId)); + } + + private ModelPrediction toPrediction(String taskId, DetectionModelResult result, Instant createdAt) { + return new ModelPrediction( + newExternalId("prediction"), + taskId, + result.target().modelId(), + result.inference().modelVersion(), + result.inference().rawScore(), + result.inference().normalizedScore(), + result.inference().label(), + result.target().threshold(), + result.inference().latencyMs(), + result.inference().rawResponseJson(), + createdAt); + } + + private DetectionReport buildReport( + String taskId, + List results, + Instant createdAt) { + DetectionModelResult strongest = results.stream() + .max(Comparator.comparingDouble(result -> result.inference().normalizedScore())) + .orElseThrow(); + ModelLabel label = strongest.inference().label(); + double confidence = strongest.inference().normalizedScore(); + ReportVerdict verdict = toVerdict(label); + RiskLevel riskLevel = toRiskLevel(label, confidence); + String summary = "Model " + strongest.target().modelId() + + " classified the image as " + label + + " with confidence " + String.format("%.2f", confidence) + "."; + String reportJson = "{\"verdict\":\"" + verdict + + "\",\"confidence\":" + confidence + + ",\"riskLevel\":\"" + riskLevel + "\"}"; + + return new DetectionReport( + newExternalId("report"), + taskId, + verdict, + confidence, + summary, + riskLevel, + reportJson, + createdAt); + } + + private ReportVerdict toVerdict(ModelLabel label) { + return switch (label) { + case AUTHENTIC -> ReportVerdict.LIKELY_AUTHENTIC; + case SYNTHETIC -> ReportVerdict.LIKELY_SYNTHETIC; + case UNCERTAIN -> ReportVerdict.UNCERTAIN; + }; + } + + private RiskLevel toRiskLevel(ModelLabel label, double score) { + if (label == ModelLabel.SYNTHETIC && score >= 0.8) { + return RiskLevel.HIGH; + } + if (label == ModelLabel.SYNTHETIC || label == ModelLabel.UNCERTAIN) { + return RiskLevel.MEDIUM; + } + return RiskLevel.LOW; + } + + private String newExternalId(String prefix) { + return prefix + "_" + UUID.randomUUID().toString().replace("-", ""); + } +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelResult.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelResult.java new file mode 100644 index 0000000..f9fce35 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelResult.java @@ -0,0 +1,8 @@ +package com.fengting.aigcforensics.service; + +import com.fengting.aigcforensics.client.ModelInferenceResult; + +public record DetectionModelResult( + DetectionModelTarget target, + ModelInferenceResult inference) { +} diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelTarget.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelTarget.java new file mode 100644 index 0000000..3d56d55 --- /dev/null +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionModelTarget.java @@ -0,0 +1,7 @@ +package com.fengting.aigcforensics.service; + +public record DetectionModelTarget( + String modelId, + String endpointUrl, + double threshold) { +} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java new file mode 100644 index 0000000..7f9f18e --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java @@ -0,0 +1,196 @@ +package com.fengting.aigcforensics.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.nio.file.Path; +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.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.fengting.aigcforensics.client.ModelInferenceResult; +import com.fengting.aigcforensics.domain.DetectionReport; +import com.fengting.aigcforensics.domain.DetectionStatus; +import com.fengting.aigcforensics.domain.DetectionTask; +import com.fengting.aigcforensics.domain.MediaAsset; +import com.fengting.aigcforensics.domain.ModelLabel; +import com.fengting.aigcforensics.domain.ModelPrediction; +import com.fengting.aigcforensics.domain.ModelRegistry; +import com.fengting.aigcforensics.repository.DetectionReportRepository; +import com.fengting.aigcforensics.repository.DetectionTaskRepository; +import com.fengting.aigcforensics.repository.MediaAssetRepository; +import com.fengting.aigcforensics.repository.ModelPredictionRepository; +import com.fengting.aigcforensics.repository.ModelRegistryRepository; + +@ExtendWith(MockitoExtension.class) +class DetectionExecutionTransactionServiceTest { + + private static final Instant NOW = Instant.parse("2026-07-11T00:00:00Z"); + + @Mock + private DetectionTaskRepository taskRepository; + @Mock + private MediaAssetRepository assetRepository; + @Mock + private ModelRegistryRepository modelRepository; + @Mock + private ModelPredictionRepository predictionRepository; + @Mock + private DetectionReportRepository reportRepository; + + private DetectionExecutionTransactionService service; + + @BeforeEach + void setUp() { + service = new DetectionExecutionTransactionService( + taskRepository, + assetRepository, + modelRepository, + predictionRepository, + reportRepository, + Duration.ofMinutes(5), + Clock.fixed(NOW, ZoneOffset.UTC)); + } + + @Test + void claimsTaskAndReturnsDetachedExecutionPlan() { + DetectionTask task = queuedTask(); + when(taskRepository.findByTaskIdForUpdate("task-1")).thenReturn(Optional.of(task)); + when(assetRepository.findByAssetId("asset-1")).thenReturn(Optional.of(asset())); + when(modelRepository.findByEnabledTrueOrderByWeightDesc()).thenReturn(List.of(model())); + + DetectionExecutionClaim claim = service.claim("task-1"); + + assertThat(claim.status()).isEqualTo(DetectionExecutionClaimStatus.CLAIMED); + assertThat(claim.plan()).isNotNull(); + assertThat(claim.plan().executionToken()).isNotBlank(); + assertThat(claim.plan().assetPath()).isEqualTo(Path.of("uploads/image.png")); + assertThat(claim.plan().models()) + .extracting(DetectionModelTarget::modelId) + .containsExactly("model-1"); + assertThat(task.getExecutionLeaseUntil()).isEqualTo(NOW.plus(Duration.ofMinutes(5))); + } + + @Test + void returnsBusyWithoutLoadingExecutionInputsForLiveLease() { + DetectionTask task = queuedTask(); + task.claimExecution("active", NOW.minusSeconds(10), NOW.plusSeconds(30)); + when(taskRepository.findByTaskIdForUpdate("task-1")).thenReturn(Optional.of(task)); + + DetectionExecutionClaim claim = service.claim("task-1"); + + assertThat(claim.status()).isEqualTo(DetectionExecutionClaimStatus.BUSY); + verify(assetRepository, never()).findByAssetId(any()); + verify(modelRepository, never()).findByEnabledTrueOrderByWeightDesc(); + } + + @Test + void marksClaimFailedWhenNoEnabledModelExists() { + DetectionTask task = queuedTask(); + when(taskRepository.findByTaskIdForUpdate("task-1")).thenReturn(Optional.of(task)); + when(assetRepository.findByAssetId("asset-1")).thenReturn(Optional.of(asset())); + when(modelRepository.findByEnabledTrueOrderByWeightDesc()).thenReturn(List.of()); + + DetectionExecutionClaim claim = service.claim("task-1"); + + assertThat(claim.status()).isEqualTo(DetectionExecutionClaimStatus.FAILED); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.FAILED); + assertThat(task.getFailureReason()).isEqualTo("No enabled model is available"); + } + + @Test + void staleCompletionDoesNotPersistResults() { + DetectionTask task = queuedTask(); + task.claimExecution("new-token", NOW, NOW.plusSeconds(60)); + when(taskRepository.findByTaskIdForUpdate("task-1")).thenReturn(Optional.of(task)); + + DetectionExecutionOutcome outcome = service.complete( + "task-1", + "old-token", + List.of(result())); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.STALE); + verify(predictionRepository, never()).saveAll(any()); + verify(reportRepository, never()).save(any()); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.INFERENCING); + } + + @Test + void matchingCompletionPersistsPredictionsAndReportAtomically() { + DetectionTask task = queuedTask(); + task.claimExecution("token-1", NOW, NOW.plusSeconds(60)); + when(taskRepository.findByTaskIdForUpdate("task-1")).thenReturn(Optional.of(task)); + + DetectionExecutionOutcome outcome = service.complete( + "task-1", + "token-1", + List.of(result())); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.COMPLETED); + ArgumentCaptor> predictions = ArgumentCaptor.forClass(List.class); + verify(predictionRepository).saveAll(predictions.capture()); + assertThat(predictions.getValue()).hasSize(1); + ArgumentCaptor report = ArgumentCaptor.forClass(DetectionReport.class); + verify(reportRepository).save(report.capture()); + assertThat(report.getValue().getTaskId()).isEqualTo("task-1"); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.COMPLETED); + } + + private DetectionTask queuedTask() { + return new DetectionTask("task-1", "asset-1", DetectionStatus.QUEUED, NOW.minusSeconds(60)); + } + + private MediaAsset asset() { + return new MediaAsset( + "asset-1", + "image.png", + "image/png", + 100, + "a".repeat(64), + 100, + 100, + "uploads/image.png", + null, + NOW.minusSeconds(60)); + } + + private ModelRegistry model() { + return new ModelRegistry( + "model-1", + "Model 1", + "IMAGE_FORENSICS", + "v1", + "http://model:5010", + true, + 0.5, + 1.0, + "test model", + NOW, + NOW); + } + + private DetectionModelResult result() { + return new DetectionModelResult( + new DetectionModelTarget("model-1", "http://model:5010", 0.5), + new ModelInferenceResult( + "v1", + 0.9, + 0.9, + ModelLabel.SYNTHETIC, + 25, + "{\"score\":0.9}")); + } +} From 3e783ca7ef442303179a1158cc1adb81a5356fa8 Mon Sep 17 00:00:00 2001 From: fengting124 Date: Sat, 11 Jul 2026 07:38:00 +0800 Subject: [PATCH 4/6] refactor: run model inference outside transactions --- .../service/DetectionExecutionService.java | 182 ++++-------------- .../service/DetectionJobWorker.java | 6 +- .../src/main/resources/application.yml | 2 + .../controller/DetectionControllerTest.java | 2 + .../DetectionExecutionServiceTest.java | 115 +++++++++++ .../service/DetectionJobWorkerTest.java | 14 ++ 6 files changed, 178 insertions(+), 143 deletions(-) create mode 100644 backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionService.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionService.java index d1add14..9074bf0 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionService.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionExecutionService.java @@ -1,177 +1,77 @@ package com.fengting.aigcforensics.service; -import java.nio.file.Path; -import java.time.Clock; -import java.time.Instant; -import java.util.Comparator; +import java.util.ArrayList; import java.util.List; -import java.util.UUID; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; import com.fengting.aigcforensics.client.ModelInferenceClient; import com.fengting.aigcforensics.client.ModelInferenceRequest; import com.fengting.aigcforensics.client.ModelInferenceResult; -import com.fengting.aigcforensics.domain.DetectionReport; -import com.fengting.aigcforensics.domain.DetectionStatus; -import com.fengting.aigcforensics.domain.DetectionTask; -import com.fengting.aigcforensics.domain.MediaAsset; -import com.fengting.aigcforensics.domain.ModelLabel; -import com.fengting.aigcforensics.domain.ModelPrediction; -import com.fengting.aigcforensics.domain.ModelRegistry; -import com.fengting.aigcforensics.domain.ReportVerdict; -import com.fengting.aigcforensics.domain.RiskLevel; -import com.fengting.aigcforensics.repository.DetectionReportRepository; -import com.fengting.aigcforensics.repository.DetectionTaskRepository; -import com.fengting.aigcforensics.repository.MediaAssetRepository; -import com.fengting.aigcforensics.repository.ModelPredictionRepository; -import com.fengting.aigcforensics.repository.ModelRegistryRepository; @Service public class DetectionExecutionService { - private final DetectionTaskRepository detectionTaskRepository; - private final MediaAssetRepository mediaAssetRepository; - private final ModelRegistryRepository modelRegistryRepository; - private final ModelPredictionRepository modelPredictionRepository; - private final DetectionReportRepository detectionReportRepository; + private static final int MAX_FAILURE_REASON_LENGTH = 2048; + + private final DetectionExecutionTransactionService transactionService; private final ModelInferenceClient modelInferenceClient; - private final Clock clock; - @Autowired public DetectionExecutionService( - DetectionTaskRepository detectionTaskRepository, - MediaAssetRepository mediaAssetRepository, - ModelRegistryRepository modelRegistryRepository, - ModelPredictionRepository modelPredictionRepository, - DetectionReportRepository detectionReportRepository, + DetectionExecutionTransactionService transactionService, ModelInferenceClient modelInferenceClient) { - this( - detectionTaskRepository, - mediaAssetRepository, - modelRegistryRepository, - modelPredictionRepository, - detectionReportRepository, - modelInferenceClient, - Clock.systemUTC()); - } - - DetectionExecutionService( - DetectionTaskRepository detectionTaskRepository, - MediaAssetRepository mediaAssetRepository, - ModelRegistryRepository modelRegistryRepository, - ModelPredictionRepository modelPredictionRepository, - DetectionReportRepository detectionReportRepository, - ModelInferenceClient modelInferenceClient, - Clock clock) { - this.detectionTaskRepository = detectionTaskRepository; - this.mediaAssetRepository = mediaAssetRepository; - this.modelRegistryRepository = modelRegistryRepository; - this.modelPredictionRepository = modelPredictionRepository; - this.detectionReportRepository = detectionReportRepository; + this.transactionService = transactionService; this.modelInferenceClient = modelInferenceClient; - this.clock = clock; } - @Transactional - public void runDetection(String taskId) { - DetectionTask task = detectionTaskRepository.findByTaskId(taskId) - .orElseThrow(() -> new ResourceNotFoundException("Detection task not found: " + taskId)); - if (task.getStatus() == DetectionStatus.COMPLETED) { - return; + public DetectionExecutionOutcome runDetection(String taskId) { + DetectionExecutionClaim claim = transactionService.claim(taskId); + if (claim.status() != DetectionExecutionClaimStatus.CLAIMED) { + return toOutcome(claim.status()); } - MediaAsset asset = mediaAssetRepository.findByAssetId(task.getAssetId()) - .orElseThrow(() -> new ResourceNotFoundException("Media asset not found: " + task.getAssetId())); - List enabledModels = modelRegistryRepository.findByEnabledTrueOrderByWeightDesc(); - if (enabledModels.isEmpty()) { - task.markFailed("No enabled model is available", Instant.now(clock)); - return; - } - - task.markStarted(Instant.now(clock)); + DetectionExecutionPlan plan = claim.plan(); + List results = new ArrayList<>(plan.models().size()); try { - for (ModelRegistry model : enabledModels) { - ModelInferenceResult result = modelInferenceClient.predict( - model.getEndpointUrl(), + for (DetectionModelTarget model : plan.models()) { + ModelInferenceResult inference = modelInferenceClient.predict( + model.endpointUrl(), new ModelInferenceRequest( - task.getTaskId(), - asset.getAssetId(), - Path.of(asset.getStoragePath()), - model.getDefaultThreshold())); - modelPredictionRepository.save(toPrediction(task, model, result)); + plan.taskId(), + plan.assetId(), + plan.assetPath(), + model.threshold())); + results.add(new DetectionModelResult(model, inference)); } - DetectionReport report = buildReport(task.getTaskId()); - detectionReportRepository.save(report); - task.markCompleted(Instant.now(clock)); } catch (RuntimeException exception) { - task.markFailed(exception.getMessage(), Instant.now(clock)); + return transactionService.fail( + plan.taskId(), + plan.executionToken(), + failureReason(exception)); } - } - private ModelPrediction toPrediction(DetectionTask task, ModelRegistry model, ModelInferenceResult result) { - Instant now = Instant.now(clock); - return new ModelPrediction( - newExternalId("prediction"), - task.getTaskId(), - model.getModelId(), - result.modelVersion(), - result.rawScore(), - result.normalizedScore(), - result.label(), - model.getDefaultThreshold(), - result.latencyMs(), - result.rawResponseJson(), - now); + return transactionService.complete( + plan.taskId(), + plan.executionToken(), + List.copyOf(results)); } - private DetectionReport buildReport(String taskId) { - List predictions = modelPredictionRepository.findByTaskIdOrderByCreatedAtAsc(taskId); - ModelPrediction strongestPrediction = predictions.stream() - .max(Comparator.comparingDouble(ModelPrediction::getNormalizedScore)) - .orElseThrow(() -> new IllegalStateException("No model prediction was produced")); - ReportVerdict verdict = toVerdict(strongestPrediction.getLabel()); - RiskLevel riskLevel = toRiskLevel(strongestPrediction.getLabel(), strongestPrediction.getNormalizedScore()); - double confidence = strongestPrediction.getNormalizedScore(); - String summary = "Model " + strongestPrediction.getModelId() - + " classified the image as " + strongestPrediction.getLabel() - + " with confidence " + String.format("%.2f", confidence) + "."; - String reportJson = "{\"verdict\":\"" + verdict - + "\",\"confidence\":" + confidence - + ",\"riskLevel\":\"" + riskLevel + "\"}"; - - return new DetectionReport( - newExternalId("report"), - taskId, - verdict, - confidence, - summary, - riskLevel, - reportJson, - Instant.now(clock)); - } - - private ReportVerdict toVerdict(ModelLabel label) { - return switch (label) { - case AUTHENTIC -> ReportVerdict.LIKELY_AUTHENTIC; - case SYNTHETIC -> ReportVerdict.LIKELY_SYNTHETIC; - case UNCERTAIN -> ReportVerdict.UNCERTAIN; + private DetectionExecutionOutcome toOutcome(DetectionExecutionClaimStatus status) { + return switch (status) { + case BUSY -> DetectionExecutionOutcome.BUSY; + case TERMINAL -> DetectionExecutionOutcome.TERMINAL; + case FAILED -> DetectionExecutionOutcome.FAILED; + case CLAIMED -> throw new IllegalArgumentException("claimed status must include an execution plan"); }; } - private RiskLevel toRiskLevel(ModelLabel label, double score) { - if (label == ModelLabel.SYNTHETIC && score >= 0.8) { - return RiskLevel.HIGH; + private String failureReason(RuntimeException exception) { + String message = exception.getMessage(); + if (message == null || message.isBlank()) { + message = exception.getClass().getSimpleName(); } - if (label == ModelLabel.SYNTHETIC || label == ModelLabel.UNCERTAIN) { - return RiskLevel.MEDIUM; - } - return RiskLevel.LOW; - } - - private String newExternalId(String prefix) { - return prefix + "_" + UUID.randomUUID().toString().replace("-", ""); + return message.length() <= MAX_FAILURE_REASON_LENGTH + ? message + : message.substring(0, MAX_FAILURE_REASON_LENGTH); } } diff --git a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobWorker.java b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobWorker.java index 5d24b67..03d673a 100644 --- a/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobWorker.java +++ b/backend-java/src/main/java/com/fengting/aigcforensics/service/DetectionJobWorker.java @@ -25,8 +25,10 @@ public DetectionJobWorker( @Scheduled(fixedDelayString = "${app.detection.jobs.poll-delay-ms:1000}") public void pollOnce() { detectionJobConsumer.poll().ifPresent(message -> { - detectionExecutionService.runDetection(message.taskId()); - detectionJobConsumer.acknowledge(message); + DetectionExecutionOutcome outcome = detectionExecutionService.runDetection(message.taskId()); + if (outcome.shouldAcknowledge()) { + detectionJobConsumer.acknowledge(message); + } }); } } diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 0d8d0d6..704a3c7 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -31,6 +31,8 @@ app: base-retry-delay: ${APP_JOBS_OUTBOX_BASE_RETRY_DELAY:1s} max-retry-delay: ${APP_JOBS_OUTBOX_MAX_RETRY_DELAY:1m} detection: + execution: + lease-duration: ${APP_DETECTION_EXECUTION_LEASE_DURATION:5m} jobs: worker-enabled: ${APP_DETECTION_JOBS_WORKER_ENABLED:true} poll-delay-ms: ${APP_DETECTION_JOBS_POLL_DELAY_MS:1000} diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/controller/DetectionControllerTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/controller/DetectionControllerTest.java index af6739c..c88d0e0 100644 --- a/backend-java/src/test/java/com/fengting/aigcforensics/controller/DetectionControllerTest.java +++ b/backend-java/src/test/java/com/fengting/aigcforensics/controller/DetectionControllerTest.java @@ -25,6 +25,7 @@ import org.springframework.mock.web.MockMultipartFile; import org.springframework.test.context.ActiveProfiles; import org.springframework.test.context.TestPropertySource; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.test.web.servlet.MockMvc; import com.fasterxml.jackson.databind.JsonNode; @@ -275,6 +276,7 @@ public void checkHealth(String endpointUrl) { @Override public ModelInferenceResult predict(String endpointUrl, ModelInferenceRequest request) { + assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isFalse(); if (failNextCall.getAndSet(false)) { throw new ModelInferenceException("model service unavailable"); } diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java new file mode 100644 index 0000000..4f2a273 --- /dev/null +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java @@ -0,0 +1,115 @@ +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.nio.file.Path; +import java.util.List; + +import org.junit.jupiter.api.BeforeEach; +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.fengting.aigcforensics.client.ModelInferenceClient; +import com.fengting.aigcforensics.client.ModelInferenceException; +import com.fengting.aigcforensics.client.ModelInferenceRequest; +import com.fengting.aigcforensics.client.ModelInferenceResult; +import com.fengting.aigcforensics.domain.ModelLabel; + +@ExtendWith(MockitoExtension.class) +class DetectionExecutionServiceTest { + + @Mock + private DetectionExecutionTransactionService transactionService; + @Mock + private ModelInferenceClient modelInferenceClient; + + private DetectionExecutionService service; + + @BeforeEach + void setUp() { + service = new DetectionExecutionService(transactionService, modelInferenceClient); + } + + @Test + void executesClaimedPlanAndCommitsCollectedResults() { + DetectionExecutionPlan plan = plan(); + when(transactionService.claim("task-1")).thenReturn(DetectionExecutionClaim.claimed(plan)); + when(modelInferenceClient.predict(eq("http://model:5010"), any())) + .thenReturn(inferenceResult()); + when(transactionService.complete(eq("task-1"), eq("token-1"), any())) + .thenReturn(DetectionExecutionOutcome.COMPLETED); + + DetectionExecutionOutcome outcome = service.runDetection("task-1"); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.COMPLETED); + ArgumentCaptor request = ArgumentCaptor.forClass(ModelInferenceRequest.class); + verify(modelInferenceClient).predict(eq("http://model:5010"), request.capture()); + assertThat(request.getValue().taskId()).isEqualTo("task-1"); + assertThat(request.getValue().imagePath()).isEqualTo(Path.of("uploads/image.png")); + verify(transactionService).complete(eq("task-1"), eq("token-1"), any()); + } + + @Test + void recordsFailureInSeparateTransactionWhenModelCallFails() { + when(transactionService.claim("task-1")).thenReturn(DetectionExecutionClaim.claimed(plan())); + when(modelInferenceClient.predict(eq("http://model:5010"), any())) + .thenThrow(new ModelInferenceException("model unavailable")); + when(transactionService.fail("task-1", "token-1", "model unavailable")) + .thenReturn(DetectionExecutionOutcome.FAILED); + + DetectionExecutionOutcome outcome = service.runDetection("task-1"); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.FAILED); + verify(transactionService).fail("task-1", "token-1", "model unavailable"); + verify(transactionService, never()).complete(any(), any(), any()); + } + + @Test + void returnsBusyWithoutCallingModel() { + when(transactionService.claim("task-1")).thenReturn( + DetectionExecutionClaim.withoutPlan(DetectionExecutionClaimStatus.BUSY)); + + DetectionExecutionOutcome outcome = service.runDetection("task-1"); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.BUSY); + verify(modelInferenceClient, never()).predict(any(), any()); + } + + @Test + void returnsTerminalWithoutCallingModel() { + when(transactionService.claim("task-1")).thenReturn( + DetectionExecutionClaim.withoutPlan(DetectionExecutionClaimStatus.TERMINAL)); + + DetectionExecutionOutcome outcome = service.runDetection("task-1"); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.TERMINAL); + verify(modelInferenceClient, never()).predict(any(), any()); + } + + private DetectionExecutionPlan plan() { + return new DetectionExecutionPlan( + "token-1", + "task-1", + "asset-1", + Path.of("uploads/image.png"), + List.of(new DetectionModelTarget("model-1", "http://model:5010", 0.5))); + } + + private ModelInferenceResult inferenceResult() { + return new ModelInferenceResult( + "v1", + 0.9, + 0.9, + ModelLabel.SYNTHETIC, + 20, + "{\"score\":0.9}"); + } +} 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 055c1bb..73cbd61 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 @@ -24,6 +24,8 @@ class DetectionJobWorkerTest { void pollsJobRunsDetectionAndAcknowledgesMessage() { DetectionJobMessage message = new DetectionJobMessage("message-001", "event-001", 1, "task-001"); when(detectionJobConsumer.poll()).thenReturn(Optional.of(message)); + when(detectionExecutionService.runDetection("task-001")) + .thenReturn(DetectionExecutionOutcome.COMPLETED); new DetectionJobWorker(detectionJobConsumer, detectionExecutionService).pollOnce(); @@ -31,6 +33,18 @@ void pollsJobRunsDetectionAndAcknowledgesMessage() { verify(detectionJobConsumer).acknowledge(message); } + @Test + void leavesMessagePendingWhenTaskHasLiveExecutionLease() { + DetectionJobMessage message = new DetectionJobMessage("message-001", "event-001", 1, "task-001"); + when(detectionJobConsumer.poll()).thenReturn(Optional.of(message)); + when(detectionExecutionService.runDetection("task-001")) + .thenReturn(DetectionExecutionOutcome.BUSY); + + new DetectionJobWorker(detectionJobConsumer, detectionExecutionService).pollOnce(); + + verify(detectionJobConsumer, never()).acknowledge(message); + } + @Test void skipsWorkWhenNoJobIsAvailable() { when(detectionJobConsumer.poll()).thenReturn(Optional.empty()); From bd0f978d25f533c1bfb40e109ec95d2a711564ad Mon Sep 17 00:00:00 2001 From: fengting124 Date: Sat, 11 Jul 2026 07:39:15 +0800 Subject: [PATCH 5/6] test: cover stale and partial detection execution --- .../DetectionExecutionServiceTest.java | 25 +++++++++++++++++++ ...ectionExecutionTransactionServiceTest.java | 17 +++++++++++++ 2 files changed, 42 insertions(+) diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java index 4f2a273..e098967 100644 --- a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionServiceTest.java @@ -72,6 +72,31 @@ void recordsFailureInSeparateTransactionWhenModelCallFails() { verify(transactionService, never()).complete(any(), any(), any()); } + @Test + void doesNotCommitPartialResultsWhenLaterModelFails() { + DetectionExecutionPlan plan = new DetectionExecutionPlan( + "token-1", + "task-1", + "asset-1", + Path.of("uploads/image.png"), + List.of( + new DetectionModelTarget("model-1", "http://model-1:5010", 0.5), + new DetectionModelTarget("model-2", "http://model-2:5010", 0.6))); + when(transactionService.claim("task-1")).thenReturn(DetectionExecutionClaim.claimed(plan)); + when(modelInferenceClient.predict(eq("http://model-1:5010"), any())) + .thenReturn(inferenceResult()); + when(modelInferenceClient.predict(eq("http://model-2:5010"), any())) + .thenThrow(new ModelInferenceException("second model unavailable")); + when(transactionService.fail("task-1", "token-1", "second model unavailable")) + .thenReturn(DetectionExecutionOutcome.FAILED); + + DetectionExecutionOutcome outcome = service.runDetection("task-1"); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.FAILED); + verify(transactionService, never()).complete(any(), any(), any()); + verify(transactionService).fail("task-1", "token-1", "second model unavailable"); + } + @Test void returnsBusyWithoutCallingModel() { when(transactionService.claim("task-1")).thenReturn( diff --git a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java index 7f9f18e..40053e5 100644 --- a/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java +++ b/backend-java/src/test/java/com/fengting/aigcforensics/service/DetectionExecutionTransactionServiceTest.java @@ -128,6 +128,23 @@ void staleCompletionDoesNotPersistResults() { assertThat(task.getStatus()).isEqualTo(DetectionStatus.INFERENCING); } + @Test + void staleFailureDoesNotChangeNewerExecution() { + DetectionTask task = queuedTask(); + task.claimExecution("new-token", NOW, NOW.plusSeconds(60)); + when(taskRepository.findByTaskIdForUpdate("task-1")).thenReturn(Optional.of(task)); + + DetectionExecutionOutcome outcome = service.fail( + "task-1", + "old-token", + "late model failure"); + + assertThat(outcome).isEqualTo(DetectionExecutionOutcome.STALE); + assertThat(task.getStatus()).isEqualTo(DetectionStatus.INFERENCING); + assertThat(task.getFailureReason()).isNull(); + assertThat(task.getExecutionToken()).isEqualTo("new-token"); + } + @Test void matchingCompletionPersistsPredictionsAndReportAtomically() { DetectionTask task = queuedTask(); From ea2d5be31ffd2be6649d9c38def16592766370c0 Mon Sep 17 00:00:00 2001 From: fengting124 Date: Sat, 11 Jul 2026 07:41:25 +0800 Subject: [PATCH 6/6] docs: explain detection execution leases --- docs/README.md | 2 + docs/async-detection-jobs.md | 5 ++ docs/detection-execution-leases.md | 113 +++++++++++++++++++++++++++++ docs/project-worklog.md | 59 ++++++++++++--- docs/reliable-job-dispatch.md | 7 +- 5 files changed, 173 insertions(+), 13 deletions(-) create mode 100644 docs/detection-execution-leases.md diff --git a/docs/README.md b/docs/README.md index fac2c15..24647d8 100644 --- a/docs/README.md +++ b/docs/README.md @@ -26,6 +26,8 @@ Use these when setting up or operating the project. backend evaluation workflow without model weights. - [Reliable Job Dispatch](reliable-job-dispatch.md): transactional outbox, Redis delivery, failure semantics, inspection, and replay. +- [Detection Execution Leases](detection-execution-leases.md): short + transactions, fencing tokens, lease recovery, and queue acknowledgement. ## Architecture And Contracts diff --git a/docs/async-detection-jobs.md b/docs/async-detection-jobs.md index 6e0f831..6bae867 100644 --- a/docs/async-detection-jobs.md +++ b/docs/async-detection-jobs.md @@ -57,3 +57,8 @@ message from blocking the queue while preserving enough metadata for operations: See [Reliable Job Dispatch](reliable-job-dispatch.md) for outbox states, configuration, failure scenarios, inspection, and replay. + +Worker-side model execution uses a database lease and fencing token. A message +for a task with a live execution lease remains pending instead of being +acknowledged. See [Detection Execution Leases](detection-execution-leases.md) +for transaction and recovery semantics. diff --git a/docs/detection-execution-leases.md b/docs/detection-execution-leases.md new file mode 100644 index 0000000..7d53f90 --- /dev/null +++ b/docs/detection-execution-leases.md @@ -0,0 +1,113 @@ +# Detection Execution Leases + +This runbook describes how a detection worker owns model execution without +holding a database transaction during network inference. + +## Guarantee + +Each attempt receives a random execution token and a time-bounded lease. The +token is a fencing value: + +- only the current token may persist predictions, a report, or failure state; +- a live lease rejects another attempt as `BUSY`; +- an expired lease can be replaced after a worker crash; +- a late response from the replaced attempt is discarded; +- all predictions and the final report commit atomically. + +The lease does not provide exactly-once model invocation. Two model calls can +overlap when an earlier call outlives its lease, but only the newest owner can +change durable task results. + +## Transaction Boundaries + +```text +claim transaction + -> lock detection_task + -> assign execution_token and execution_lease_until + -> increment execution_attempt_count + -> snapshot asset and enabled model configuration + +outside a transaction + -> invoke model endpoints + -> collect results in memory + +completion transaction + -> lock detection_task + -> verify execution_token + -> save every prediction and one report + -> mark task COMPLETED and clear ownership + +failure transaction + -> lock detection_task + -> verify execution_token + -> mark task FAILED and clear ownership +``` + +Keeping the coordinator and transactional persistence service as separate +Spring beans is intentional. It ensures Spring's transaction proxy is applied +and avoids self-invocation silently bypassing transaction annotations. + +## Configuration + +| Environment variable | Default | Purpose | +| --- | --- | --- | +| `APP_DETECTION_EXECUTION_LEASE_DURATION` | `5m` | Time before an unfinished attempt may be replaced | + +Choose a lease longer than normal end-to-end inference latency, including all +enabled models. A short lease improves crash recovery but increases the chance +of overlapping calls. A long lease reduces overlap but delays recovery. + +The current implementation has no lease heartbeat. Do not set the lease below +the measured high-percentile inference time. Heartbeats are deferred until a +real model runtime provides representative latency data. + +## Queue Acknowledgement + +The Redis worker acknowledges these outcomes: + +- `COMPLETED`: results committed; +- `FAILED`: the current attempt recorded a failure; +- `TERMINAL`: the task was already complete; +- `STALE`: a newer execution owns or finished the task. + +It does not acknowledge `BUSY`. The message remains in the consumer group's +pending list and can be claimed again after the configured Redis pending idle +timeout. Once the active attempt reaches a terminal state, redelivery becomes +safe to acknowledge. + +## Failure Scenarios + +### Worker exits during model inference + +The task remains `INFERENCING`. After `execution_lease_until`, a redelivered +message can assign a new token and execute again. No manual database update is +required. + +### Old worker returns after lease replacement + +The old completion transaction sees a token mismatch and performs no writes. +The worker may acknowledge its own Redis message because the newer attempt is +now the durable owner. + +### One of several models fails + +Results are collected in memory. The service writes no partial predictions; +it records the task failure in a short transaction. A later explicit replay +starts a clean attempt. + +### Completion transaction fails + +The exception escapes the worker, so the Redis message is not acknowledged. +The task remains recoverable after its lease expires. Database errors are not +misreported as model failures. + +## Verification Boundary + +Automated tests cover lease state transitions, active and expired claims, +stale success and failure fencing, partial multi-model failure, queue +acknowledgement, Flyway/JPA validation, and an integration assertion that model +inference runs without an active Spring transaction. + +Real process termination and PostgreSQL/Redis restart tests remain assigned to +the Docker-backed integration-test branch. Model weights are not required for +these transaction guarantees. diff --git a/docs/project-worklog.md b/docs/project-worklog.md index ac53c8b..f436e6a 100644 --- a/docs/project-worklog.md +++ b/docs/project-worklog.md @@ -623,23 +623,64 @@ Deferred: --- +### 2026-07-11: Short-Lived Detection Execution Transactions + +```text +feature/short-lived-execution-transactions +``` + +What changed: + +- Added Flyway V6 execution token, lease expiry, and attempt count fields. +- Split task claim, external model invocation, and result persistence into + separate transaction boundaries. +- Added fencing-token checks so stale success or failure callbacks cannot + overwrite a newer attempt. +- Kept partial multi-model results in memory and committed predictions and the + report atomically only after every model succeeded. +- Changed the Redis worker to leave `BUSY` messages pending for safe redelivery. + +Why: + +- Network inference must not hold database connections or row locks. +- A crashed worker needs bounded recovery without accepting late stale writes. +- The guarantee is useful with a heuristic adapter now and remains valid when + GPU-backed model latency is introduced later. + +Verification: + +- Domain tests cover live lease rejection, expiry recovery, and token fencing. +- Service tests cover detached plans, stale writes, multi-model partial + failure, and acknowledgement decisions. +- Spring integration tests assert the model client runs without an active + transaction and validate Flyway V6 against JPA. +- Java suite reached 97 passing tests before final cross-project verification. + +Deferred: + +- Lease heartbeat until representative real-model latency is available. +- Docker-backed process crash and PostgreSQL/Redis restart tests. +- Applying the same execution boundary to batch evaluation. + +--- + ## Next Recommended Work -Continue Stage 1 by shortening detection execution transactions: +Continue the production foundation with the upload trust boundary: ```text -feature/short-lived-execution-transactions +feature/upload-trust-boundary ``` Scope: -- 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. +- Validate file signatures and decoded image content instead of trusting MIME + type or filename. +- Apply decoded pixel and dimension limits against decompression bombs. +- Separate quarantine, accepted storage, and cleanup behavior. +- Define deterministic rejection responses and security-focused tests. Reason: -`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. +Uploads are the system's primary untrusted input. Hardening this boundary adds +real security depth without requiring Docker, GPU hardware, or model weights. diff --git a/docs/reliable-job-dispatch.md b/docs/reliable-job-dispatch.md index ce51242..72416ce 100644 --- a/docs/reliable-job-dispatch.md +++ b/docs/reliable-job-dispatch.md @@ -166,7 +166,6 @@ 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. +work. Concurrent ownership of a non-terminal task is protected by the +execution lease and fencing token described in +[Detection Execution Leases](detection-execution-leases.md).