diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java index 31e6550c..f32a2876 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java @@ -11,6 +11,7 @@ import com.iflytek.skillhub.domain.skill.SkillVersionRepository; import com.iflytek.skillhub.domain.skill.SkillVersionStatus; import com.iflytek.skillhub.observability.MessageObservationSupport; import com.iflytek.skillhub.storage.ObjectStorageService; +import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import java.io.IOException; @@ -26,6 +27,7 @@ import java.util.Map; public class ScanTaskConsumer extends AbstractStreamConsumer { private static final Path SCAN_TEMP_DIR = Paths.get("/tmp/skillhub-scans").toAbsolutePath().normalize(); + private final RedissonClient redissonClient; private final SecurityScanner securityScanner; private final SecurityScanService securityScanService; private final SkillVersionRepository skillVersionRepository; @@ -42,6 +44,7 @@ public class ScanTaskConsumer extends AbstractStreamConsumer candidates = new LinkedHashMap<>(); - repository.findPendingDue(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); - repository.findExpiredLeases(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); - for (ScanTaskOutbox outbox : candidates.values()) { - if (!outbox.claim(now, lease)) continue; - repository.saveAndFlush(outbox); + for (ScanTaskOutbox outbox : repository.findDispatchable(now, batchSize)) { + if (!outbox.claim(now, lease)) { + continue; + } try { producer.publishScanTask(outbox.toScanTask()); outbox.markSent(Instant.now(clock)); repository.save(outbox); } catch (Exception e) { - Duration delay = retryDelay(outbox.getRetryCount() + 1); - outbox.markRetry(Instant.now(clock), delay, e.toString()); - repository.save(outbox); - log.warn("Failed to publish scan task; will retry taskId={}, retryCount={}, nextDelay={}", - outbox.getTaskId(), outbox.getRetryCount(), delay, e); + handlePublishFailure(outbox, e); } } } + private void handlePublishFailure(ScanTaskOutbox outbox, Exception error) { + Instant now = Instant.now(clock); + int nextAttempt = outbox.getRetryCount() + 1; + if (nextAttempt >= maxAttempts) { + outbox.markFailed(now, error.toString()); + repository.save(outbox); + versionRepository.findById(outbox.getVersionId()) + .filter(version -> version.getStatus() == SkillVersionStatus.SCANNING) + .ifPresent(version -> { + version.setStatus(SkillVersionStatus.SCAN_FAILED); + versionRepository.save(version); + }); + log.error("Scan task publish failed permanently: taskId={}, versionId={}, attempts={}", + outbox.getTaskId(), outbox.getVersionId(), outbox.getRetryCount(), error); + return; + } + Duration delay = retryDelay(nextAttempt); + outbox.markRetry(now, delay, error.toString()); + repository.save(outbox); + log.warn("Failed to publish scan task; will retry taskId={}, retryCount={}, nextDelay={}", + outbox.getTaskId(), outbox.getRetryCount(), delay, error); + } + @Scheduled(cron = "0 20 2 * * ?") @Transactional public void cleanupSent() { int deleted = repository.deleteSentBefore(Instant.now(clock).minus(Duration.ofDays(7))); - if (deleted > 0) log.info("Cleaned up {} sent scan outbox records", deleted); + if (deleted > 0) { + log.info("Cleaned up {} sent scan outbox records", deleted); + } } private Duration retryDelay(int retryCount) { long seconds = Math.min(maxBackoff.toSeconds(), 1L << Math.min(retryCount, 16)); return Duration.ofSeconds(Math.max(seconds, 1)); } -} \ No newline at end of file +} diff --git a/server/skillhub-app/src/main/resources/db/migration/V45__scan_task_outbox_metadata.sql b/server/skillhub-app/src/main/resources/db/migration/V45__scan_task_outbox_metadata.sql new file mode 100644 index 00000000..b07bb8e8 --- /dev/null +++ b/server/skillhub-app/src/main/resources/db/migration/V45__scan_task_outbox_metadata.sql @@ -0,0 +1,2 @@ +ALTER TABLE scan_task_outbox + ADD COLUMN metadata JSONB NOT NULL DEFAULT '{}'::jsonb; diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerLoggingTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerLoggingTest.java index 8f473b65..50e73da8 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerLoggingTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerLoggingTest.java @@ -20,6 +20,7 @@ import com.iflytek.skillhub.storage.ObjectStorageService; import io.micrometer.observation.ObservationRegistry; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import org.redisson.api.RLock; import org.redisson.api.RStream; import org.redisson.api.RedissonClient; import org.redisson.api.StreamMessageId; @@ -35,6 +36,7 @@ import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; class ScanTaskConsumerLoggingTest { @@ -150,6 +152,15 @@ class ScanTaskConsumerLoggingTest { } } + private static RedissonClient redissonClientWithAvailableProcessingLock() { + RedissonClient redissonClient = mock(RedissonClient.class); + RLock processingLock = mock(RLock.class); + when(redissonClient.getLock(org.mockito.ArgumentMatchers.anyString())).thenReturn(processingLock); + when(processingLock.tryLock()).thenReturn(true); + when(processingLock.isHeldByCurrentThread()).thenReturn(true); + return redissonClient; + } + private static final class TestableLoggingConsumer extends ScanTaskConsumer { private final RStream stream = mock(RStream.class); @@ -159,7 +170,7 @@ class ScanTaskConsumerLoggingTest { ScanTaskProducer scanTaskProducer, ObjectStorageService objectStorageService) { super( - mock(RedissonClient.class), + redissonClientWithAvailableProcessingLock(), "skillhub:scan:requests", "skillhub-scanners", securityScanner, diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerTest.java index 7f8cabaa..8456d11d 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerTest.java @@ -20,6 +20,7 @@ import com.iflytek.skillhub.storage.ObjectStorageService; import com.iflytek.skillhub.storage.ObjectMetadata; import io.micrometer.observation.ObservationRegistry; import org.junit.jupiter.api.Test; +import org.redisson.api.RLock; import org.redisson.api.RStream; import org.redisson.api.RedissonClient; import org.redisson.api.StreamMessageId; @@ -38,7 +39,11 @@ import java.util.Map; import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; class ScanTaskConsumerTest { private static final Path SCAN_TEMP_DIR = Path.of("/tmp/skillhub-scans"); @@ -268,6 +273,60 @@ class ScanTaskConsumerTest { assertThat(listScanTempFiles(versionId)).isEmpty(); } + @Test + void processBusiness_whenTaskIsAlreadyInFlight_skipsScanAndPreservesSharedTempPath() throws Exception { + Files.createDirectories(SCAN_TEMP_DIR); + Path tempDir = Files.createTempDirectory(SCAN_TEMP_DIR, "scan-task-consumer-inflight"); + Path skillFile = Files.writeString(tempDir.resolve("SKILL.md"), "# demo"); + StubSecurityScanner securityScanner = new StubSecurityScanner(); + RLock processingLock = mock(RLock.class); + when(processingLock.tryLock()).thenReturn(false); + TestableScanTaskConsumer consumer = new TestableScanTaskConsumer( + securityScanner, + new StubSecurityScanService(), + new InMemorySkillVersionRepository(), + new InMemoryScanTaskProducer(), + new InMemoryObjectStorageService(), + redissonClient(processingLock) + ); + ScanTaskConsumer.ScanTaskPayload payload = new ScanTaskConsumer.ScanTaskPayload( + "task-inflight", 42L, tempDir.toString(), null, ScannerType.SKILL_SCANNER); + + try { + consumer.invokeProcessBusiness(payload); + consumer.invokeMarkCompleted(payload); + + assertThat(securityScanner.lastRequest).isNull(); + assertThat(skillFile).exists(); + verify(processingLock, never()).unlock(); + } finally { + Files.deleteIfExists(skillFile); + Files.deleteIfExists(tempDir); + } + } + + @Test + void processBusiness_whenScannerFails_releasesProcessingLock() { + StubSecurityScanner securityScanner = new StubSecurityScanner(); + securityScanner.failure = new IllegalStateException("scanner unavailable"); + RLock processingLock = availableProcessingLock(); + TestableScanTaskConsumer consumer = new TestableScanTaskConsumer( + securityScanner, + new StubSecurityScanService(), + new InMemorySkillVersionRepository(), + new InMemoryScanTaskProducer(), + new InMemoryObjectStorageService(), + redissonClient(processingLock) + ); + ScanTaskConsumer.ScanTaskPayload payload = new ScanTaskConsumer.ScanTaskPayload( + "task-failure", 42L, "/tmp/failure", null, ScannerType.SKILL_SCANNER); + + assertThatThrownBy(() -> consumer.invokeProcessBusiness(payload)) + .isInstanceOf(IllegalStateException.class) + .hasMessage("scanner unavailable"); + verify(processingLock).unlock(); + } + private void setField(Object target, String fieldName, Object value) throws Exception { Field field = target.getClass().getDeclaredField(fieldName); field.setAccessible(true); @@ -299,7 +358,28 @@ class ScanTaskConsumerTest { ScanTaskProducer scanTaskProducer, ObjectStorageService objectStorageService) { super( - mock(RedissonClient.class), + redissonClient(availableProcessingLock()), + "skillhub:scan:requests", + "skillhub-scanners", + securityScanner, + securityScanService, + skillVersionRepository, + scanTaskProducer, + objectStorageService, + new MessageObservationSupport(ObservationRegistry.NOOP, new RequestIdAccessor()) + ); + this.stream = mock(RStream.class); + } + + @SuppressWarnings("unchecked") + private TestableScanTaskConsumer(SecurityScanner securityScanner, + SecurityScanService securityScanService, + SkillVersionRepository skillVersionRepository, + ScanTaskProducer scanTaskProducer, + ObjectStorageService objectStorageService, + RedissonClient redissonClient) { + super( + redissonClient, "skillhub:scan:requests", "skillhub-scanners", securityScanner, @@ -334,6 +414,19 @@ class ScanTaskConsumerTest { } } + private static RLock availableProcessingLock() { + RLock processingLock = mock(RLock.class); + when(processingLock.tryLock()).thenReturn(true); + when(processingLock.isHeldByCurrentThread()).thenReturn(true); + return processingLock; + } + + private static RedissonClient redissonClient(RLock processingLock) { + RedissonClient redissonClient = mock(RedissonClient.class); + when(redissonClient.getLock(org.mockito.ArgumentMatchers.anyString())).thenReturn(processingLock); + return redissonClient; + } + private static final class StubSecurityScanner implements SecurityScanner { private SecurityScanRequest lastRequest; private SecurityScanResponse response; diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java index eae1e0ec..eafa3b6f 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java @@ -3,8 +3,12 @@ package com.iflytek.skillhub.task; import com.iflytek.skillhub.domain.security.ScanTask; import com.iflytek.skillhub.domain.security.ScanTaskOutbox; import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus; import com.iflytek.skillhub.domain.security.ScanTaskProducer; import com.iflytek.skillhub.domain.security.ScannerType; +import com.iflytek.skillhub.domain.skill.SkillVersion; +import com.iflytek.skillhub.domain.skill.SkillVersionRepository; +import com.iflytek.skillhub.domain.skill.SkillVersionStatus; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; @@ -15,35 +19,122 @@ import java.time.Duration; import java.time.Instant; import java.time.ZoneOffset; import java.util.List; +import java.util.Optional; 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.BDDMockito.given; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; @ExtendWith(MockitoExtension.class) class ScanTaskOutboxDispatcherTest { @Mock ScanTaskOutboxRepository repository; @Mock ScanTaskProducer producer; + @Mock SkillVersionRepository versionRepository; @Test void failedRedisPublishLeavesTaskPendingForRetry() { - Clock clock = Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC); - ScanTaskOutbox outbox = new ScanTaskOutbox( - new ScanTask("task-1", 1L, "/tmp/1", null, "user", 1L, - java.util.Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()))); - given(repository.findPendingDue(any(), any(Integer.class))).willReturn(List.of(outbox)); - given(repository.findExpiredLeases(any(), any(Integer.class))).willReturn(List.of()); + ScanTaskOutbox outbox = outbox("task-1", 1L); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any()); - ScanTaskOutboxDispatcher dispatcher = new ScanTaskOutboxDispatcher( - repository, producer, clock, 50, Duration.ofMinutes(2), Duration.ofMinutes(5)); - dispatcher.dispatch(); + dispatcher(10).dispatch(); - assertThat(outbox.getStatus()).isEqualTo(com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); assertThat(outbox.getRetryCount()).isEqualTo(1); verify(producer).publishScanTask(any()); - verify(repository).saveAndFlush(outbox); + verify(repository).save(outbox); } -} \ No newline at end of file + + @Test + void successfulPublishMarksTaskSentWithoutChangingVersion() { + ScanTaskOutbox outbox = outbox("task-success", 3L); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); + + dispatcher(10).dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT); + assertThat(outbox.getRetryCount()).isZero(); + verify(producer).publishScanTask(any()); + verify(repository).save(outbox); + verifyNoInteractions(versionRepository); + } + + @Test + void lastPublishAttemptMarksOutboxAndVersionFailed() { + ScanTaskOutbox outbox = outbox("task-2", 2L); + SkillVersion version = new SkillVersion(9L, "1.0.0", "user"); + version.setStatus(SkillVersionStatus.SCANNING); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); + given(versionRepository.findById(2L)).willReturn(Optional.of(version)); + doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any()); + + dispatcher(1).dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.FAILED); + assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.SCAN_FAILED); + verify(versionRepository).save(version); + } + + @Test + void lastPublishAttemptDoesNotOverwriteTerminalVersionStatus() { + ScanTaskOutbox outbox = outbox("task-published", 4L); + SkillVersion version = new SkillVersion(9L, "1.0.0", "user"); + version.setStatus(SkillVersionStatus.PUBLISHED); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); + given(versionRepository.findById(4L)).willReturn(Optional.of(version)); + doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any()); + + dispatcher(1).dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.FAILED); + assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.PUBLISHED); + verify(versionRepository, never()).save(version); + } + + @Test + void expiredLeaseCanBeReclaimedAndPublished() { + ScanTaskOutbox outbox = outbox("task-expired", 5L); + assertThat(outbox.claim(Instant.parse("2025-12-31T23:00:00Z"), Duration.ofMinutes(2))).isTrue(); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); + + dispatcher(10).dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT); + verify(producer).publishScanTask(any()); + } + + @Test + void staleFinderResultInTerminalStateIsIgnored() { + ScanTaskOutbox outbox = outbox("task-sent", 6L); + outbox.markSent(Instant.parse("2025-12-31T23:00:00Z")); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); + + dispatcher(10).dispatch(); + + verifyNoInteractions(producer); + verify(repository, never()).save(outbox); + } + + @Test + void maxAttemptsMustBePositive() { + assertThatThrownBy(() -> dispatcher(0)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("maxAttempts"); + } + + private ScanTaskOutboxDispatcher dispatcher(int maxAttempts) { + Clock clock = Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC); + return new ScanTaskOutboxDispatcher(repository, producer, versionRepository, clock, + 50, maxAttempts, Duration.ofMinutes(2), Duration.ofMinutes(5)); + } + + private ScanTaskOutbox outbox(String taskId, Long versionId) { + return new ScanTaskOutbox(new ScanTask(taskId, versionId, "/tmp/" + versionId, null, "user", 1L, + java.util.Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()))); + } +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java index 8f93fce1..7da371fb 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java @@ -10,6 +10,8 @@ import jakarta.persistence.Id; import jakarta.persistence.PrePersist; import jakarta.persistence.Table; import jakarta.persistence.Version; +import org.hibernate.annotations.JdbcTypeCode; +import org.hibernate.type.SqlTypes; import java.time.Clock; import java.time.Duration; @@ -31,6 +33,9 @@ public class ScanTaskOutbox { private String bundleKey; @Column(name = "publisher_id", length = 255) private String publisherId; + @JdbcTypeCode(SqlTypes.JSON) + @Column(name = "metadata", nullable = false, columnDefinition = "jsonb") + private Map metadata; @Enumerated(EnumType.STRING) @Column(nullable = false, length = 20) private ScanTaskOutboxStatus status; @Column(name = "retry_count", nullable = false) @@ -56,22 +61,25 @@ public class ScanTaskOutbox { this.skillPath = task.skillPath(); this.bundleKey = task.bundleKey(); this.publisherId = task.publisherId(); + this.metadata = task.metadata() == null ? Map.of() : Map.copyOf(task.metadata()); this.status = ScanTaskOutboxStatus.PENDING; - this.nextAttemptAt = Instant.now(Clock.systemUTC()); + Instant taskCreatedAt = Instant.ofEpochMilli(task.createdAtMillis()); + this.nextAttemptAt = taskCreatedAt; + this.createdAt = taskCreatedAt; + this.updatedAt = taskCreatedAt; } @PrePersist protected void onCreate() { Instant now = Instant.now(Clock.systemUTC()); - createdAt = now; - updatedAt = now; + if (createdAt == null) createdAt = now; + if (updatedAt == null) updatedAt = now; if (nextAttemptAt == null) nextAttemptAt = now; } public ScanTask toScanTask() { return new ScanTask(taskId, versionId, skillPath, bundleKey, publisherId, - createdAt == null ? System.currentTimeMillis() : createdAt.toEpochMilli(), - Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + createdAt.toEpochMilli(), metadata == null ? Map.of() : Map.copyOf(metadata)); } public boolean claim(Instant now, Duration lease) { @@ -90,12 +98,20 @@ public class ScanTaskOutbox { updatedAt = now; } + public void markFailed(Instant now, String error) { + retryCount++; + status = ScanTaskOutboxStatus.FAILED; + leaseUntil = null; + lastError = truncateError(error); + updatedAt = now; + } + public void markRetry(Instant now, Duration delay, String error) { retryCount++; status = ScanTaskOutboxStatus.PENDING; nextAttemptAt = now.plus(delay); leaseUntil = null; - lastError = error == null ? null : error.substring(0, Math.min(error.length(), 2000)); + lastError = truncateError(error); updatedAt = now; } @@ -106,5 +122,10 @@ public class ScanTaskOutbox { public int getRetryCount() { return retryCount; } public Instant getNextAttemptAt() { return nextAttemptAt; } public Instant getLeaseUntil() { return leaseUntil; } + + private String truncateError(String error) { + return error == null ? null : error.substring(0, Math.min(error.length(), 2000)); + } + public Instant getCreatedAt() { return createdAt; } } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java index 43c5c24e..7ed29c5d 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java @@ -5,9 +5,7 @@ import java.util.List; public interface ScanTaskOutboxRepository { ScanTaskOutbox save(ScanTaskOutbox outbox); - ScanTaskOutbox saveAndFlush(ScanTaskOutbox outbox); - List findPendingDue(Instant now, int limit); - List findExpiredLeases(Instant now, int limit); + List findDispatchable(Instant now, int limit); int deleteSentBefore(Instant cutoff); int deleteByVersionId(Long versionId); } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java index 5efccbd0..df12ee8e 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java @@ -207,6 +207,9 @@ public class SecurityScanService { */ @Transactional public void softDeleteByVersionId(Long versionId) { + if (scanTaskOutboxRepository != null) { + scanTaskOutboxRepository.deleteByVersionId(versionId); + } List audits = auditRepository.findAllActiveBySkillVersionId(versionId); if (audits.isEmpty()) { log.debug("No active security audits to soft-delete for versionId={}", versionId); diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java index 968fd1e8..373baaf6 100644 --- a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java @@ -12,7 +12,9 @@ class ScanTaskOutboxTest { @Test void claimAndMarkSentProducesStableTaskPayload() { ScanTask task = new ScanTask("task-1", 7L, "/tmp/7", null, "u1", 123L, - Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + Map.of( + "scannerType", ScannerType.SKILL_SCANNER.getValue(), + "futureAttribute", "preserved")); ScanTaskOutbox outbox = new ScanTaskOutbox(task); Instant now = Instant.parse("2026-01-01T00:00:00Z"); @@ -21,8 +23,22 @@ class ScanTaskOutboxTest { outbox.markSent(now.plusSeconds(1)); assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT); - assertThat(outbox.toScanTask().taskId()).isEqualTo("task-1"); - assertThat(outbox.toScanTask().versionId()).isEqualTo(7L); + assertThat(outbox.toScanTask()).isEqualTo(task); + } + + @Test + void exhaustedPublishAttemptsMoveTaskToFailed() { + ScanTaskOutbox outbox = new ScanTaskOutbox( + new ScanTask("task-failed", 9L, null, "bundle.zip", null, 1L, Map.of())); + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + outbox.claim(now, Duration.ofMinutes(2)); + + outbox.markFailed(now, "permanent failure"); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.FAILED); + assertThat(outbox.getRetryCount()).isEqualTo(1); + assertThat(outbox.getLeaseUntil()).isNull(); + assertThat(outbox.claim(now.plusSeconds(1), Duration.ofMinutes(2))).isFalse(); } @Test @@ -37,4 +53,4 @@ class ScanTaskOutboxTest { assertThat(outbox.getRetryCount()).isEqualTo(1); assertThat(outbox.getNextAttemptAt()).isEqualTo(now.plusSeconds(5)); } -} \ No newline at end of file +} diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java index ce0b247e..f0e056bf 100644 --- a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java @@ -44,4 +44,16 @@ class SecurityScanOutboxTest { assertThat(outbox.getValue().getVersionId()).isEqualTo(42L); assertThat(outbox.getValue().getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); } -} \ No newline at end of file + + @Test + void softDeleteRemovesPendingOutboxEvenWhenNoActiveAuditExists() { + given(auditRepository.findAllActiveBySkillVersionId(42L)).willReturn(List.of()); + SecurityScanService service = new SecurityScanService(auditRepository, versionRepository, producer, + new ObjectMapper(), "upload", true, outboxRepository); + + service.softDeleteByVersionId(42L); + + verify(outboxRepository).deleteByVersionId(42L); + verify(auditRepository, never()).saveAll(org.mockito.ArgumentMatchers.anyList()); + } +} diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java index 34cd6be0..84b08993 100644 --- a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java @@ -2,8 +2,6 @@ package com.iflytek.skillhub.infra.jpa; import com.iflytek.skillhub.domain.security.ScanTaskOutbox; import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; -import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus; -import org.springframework.data.domain.PageRequest; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; @@ -16,16 +14,15 @@ import java.util.List; @Repository public interface ScanTaskOutboxJpaRepository extends JpaRepository, ScanTaskOutboxRepository { @Override - default List findPendingDue(Instant now, int limit) { - return findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( - ScanTaskOutboxStatus.PENDING, now, PageRequest.of(0, limit)); - } - - @Override - default List findExpiredLeases(Instant now, int limit) { - return findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( - ScanTaskOutboxStatus.SENDING, now, PageRequest.of(0, limit)); - } + @Query(value = """ + SELECT * FROM scan_task_outbox + WHERE (status = 'PENDING' AND next_attempt_at <= :now) + OR (status = 'SENDING' AND lease_until < :now) + ORDER BY created_at + LIMIT :limit + FOR UPDATE SKIP LOCKED + """, nativeQuery = true) + List findDispatchable(@Param("now") Instant now, @Param("limit") int limit); @Override @Modifying @@ -34,10 +31,4 @@ public interface ScanTaskOutboxJpaRepository extends JpaRepository findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( - ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); - - List findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( - ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); } \ No newline at end of file