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 fce9bcc2e..3b1b2c669 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.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 @@ 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 ScanTaskConsumer(RedissonClient redissonClient, ObjectStorageService objectStorageService, MessageObservationSupport messageObservationSupport) { super(redissonClient, streamKey, groupName, messageObservationSupport); + this.redissonClient = redissonClient; this.securityScanner = securityScanner; this.securityScanService = securityScanService; this.skillVersionRepository = skillVersionRepository; @@ -72,6 +75,7 @@ public ScanTaskConsumer(RedissonClient redissonClient, reclaimInterval, messageObservationSupport ); + this.redissonClient = redissonClient; this.securityScanner = securityScanner; this.securityScanService = securityScanService; this.skillVersionRepository = skillVersionRepository; @@ -128,17 +132,49 @@ protected void markProcessing(ScanTaskPayload payload) { @Override protected void processBusiness(ScanTaskPayload payload) { + if (securityScanService.isTaskAlreadyProcessed(payload.taskId())) { + log.info("Skipping already processed security scan task: taskId={}, versionId={}", payload.taskId(), payload.versionId()); + return; + } + RLock processingLock = redissonClient.getLock("skillhub:scan:processing:" + payload.taskId()); + boolean acquired = false; + try { + acquired = processingLock.tryLock(); + if (!acquired) { + log.info("Skipping concurrently processed security scan task: taskId={}, versionId={}", + payload.taskId(), payload.versionId()); + payload.skipCleanup(); + // A normal return is treated as success by AbstractStreamConsumer and ACKs + // the Redis entry. Requeue through the common failure path instead, so a + // reclaimed duplicate cannot erase the only durable delivery while the active + // scanner still owns the task lock. + throw new ConcurrentScanInProgressException(payload.taskId()); + } + if (securityScanService.isTaskAlreadyProcessed(payload.taskId())) { + return; + } + executeScan(payload); + } finally { + if (acquired && processingLock.isHeldByCurrentThread()) { + processingLock.unlock(); + } + } + } + + private void executeScan(ScanTaskPayload payload) { String skillPath = resolveWorkingSkillPath(payload); SecurityScanRequest request = new SecurityScanRequest( - payload.taskId(), - payload.versionId(), - skillPath, - Map.of() - ); + payload.taskId(), payload.versionId(), skillPath, Map.of()); SecurityScanResponse response = securityScanner.scan(request); securityScanService.processScanResult(payload.versionId(), payload.scannerType(), response); } + private static final class ConcurrentScanInProgressException extends RuntimeException { + private ConcurrentScanInProgressException(String taskId) { + super("Security scan is already in progress: taskId=" + taskId); + } + } + @Override protected void markCompleted(ScanTaskPayload payload) { cleanupTempPath(payload.cleanupPath()); @@ -259,6 +295,7 @@ protected static final class ScanTaskPayload { private final ScannerType scannerType; private final int retryCount; private String workingSkillPath; + private boolean cleanupEnabled = true; protected ScanTaskPayload(String taskId, Long versionId, String skillPath, String bundleKey, ScannerType scannerType) { this(taskId, versionId, skillPath, bundleKey, scannerType, 0); @@ -307,9 +344,16 @@ protected void markWorkingSkillPath(String workingSkillPath) { } protected String cleanupPath() { + if (!cleanupEnabled) { + return null; + } return workingSkillPath != null ? workingSkillPath : skillPath; } + protected void skipCleanup() { + cleanupEnabled = false; + } + protected String workingSkillPath() { return workingSkillPath; } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcher.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcher.java new file mode 100644 index 000000000..242a32d20 --- /dev/null +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcher.java @@ -0,0 +1,109 @@ +package com.iflytek.skillhub.task; + +import com.iflytek.skillhub.domain.security.ScanTaskOutbox; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import com.iflytek.skillhub.domain.security.ScanTaskProducer; +import com.iflytek.skillhub.domain.skill.SkillVersionRepository; +import com.iflytek.skillhub.domain.skill.SkillVersionStatus; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; + +@Component +@ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true") +public class ScanTaskOutboxDispatcher { + private static final Logger log = LoggerFactory.getLogger(ScanTaskOutboxDispatcher.class); + + private final ScanTaskOutboxRepository repository; + private final ScanTaskProducer producer; + private final SkillVersionRepository versionRepository; + private final Clock clock; + private final int batchSize; + private final int maxAttempts; + private final Duration lease; + private final Duration maxBackoff; + + public ScanTaskOutboxDispatcher(ScanTaskOutboxRepository repository, + ScanTaskProducer producer, + SkillVersionRepository versionRepository, + Clock clock, + @Value("${skillhub.security.outbox.batch-size:50}") int batchSize, + @Value("${skillhub.security.outbox.max-attempts:10}") int maxAttempts, + @Value("${skillhub.security.outbox.lease:PT2M}") Duration lease, + @Value("${skillhub.security.outbox.max-backoff:PT5M}") Duration maxBackoff) { + this.repository = repository; + this.producer = producer; + this.versionRepository = versionRepository; + this.clock = clock; + this.batchSize = batchSize; + if (maxAttempts < 1) { + throw new IllegalArgumentException("maxAttempts must be at least 1"); + } + this.maxAttempts = maxAttempts; + this.lease = lease; + this.maxBackoff = maxBackoff; + } + + @Scheduled(fixedDelayString = "${skillhub.security.outbox.dispatch-interval-ms:5000}") + @Transactional + public void dispatch() { + Instant now = Instant.now(clock); + 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) { + 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); + } + } + + private Duration retryDelay(int retryCount) { + long seconds = Math.min(maxBackoff.toSeconds(), 1L << Math.min(retryCount, 16)); + return Duration.ofSeconds(Math.max(seconds, 1)); + } +} diff --git a/server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql b/server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql new file mode 100644 index 000000000..2b56fb36c --- /dev/null +++ b/server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql @@ -0,0 +1,28 @@ +CREATE TABLE scan_task_outbox ( + id BIGSERIAL PRIMARY KEY, + task_id VARCHAR(100) NOT NULL, + version_id BIGINT NOT NULL, + skill_path VARCHAR(1000), + bundle_key VARCHAR(1000), + publisher_id VARCHAR(255), + status VARCHAR(20) NOT NULL, + retry_count INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ NOT NULL, + lease_until TIMESTAMPTZ, + last_error VARCHAR(2000), + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + entity_version BIGINT NOT NULL DEFAULT 0, + CONSTRAINT uk_scan_task_outbox_task_id UNIQUE (task_id), + CONSTRAINT ck_scan_task_outbox_status CHECK (status IN ('PENDING', 'SENDING', 'SENT', 'FAILED')) +); + +CREATE INDEX idx_scan_task_outbox_pending + ON scan_task_outbox (status, next_attempt_at, created_at); +CREATE INDEX idx_scan_task_outbox_lease + ON scan_task_outbox (status, lease_until); +CREATE INDEX idx_scan_task_outbox_version + ON scan_task_outbox (version_id); + +ALTER TABLE security_audit ADD COLUMN task_id VARCHAR(100); +CREATE INDEX idx_security_audit_task_id ON security_audit (task_id); \ 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 000000000..b07bb8e84 --- /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 8f473b656..50e73da86 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 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 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 @@ private void setVersionId(SkillVersion version, Long id) { } } + 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 @@ private TestableLoggingConsumer(SecurityScanner securityScanner, 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 7f8cabaaa..07d7d9d65 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.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.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,88 @@ void handleMessage_retryableBundleDownloadFailure_requeuesTaskWithoutLeakingTemp 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 { + assertThatThrownBy(() -> consumer.invokeProcessBusiness(payload)) + .isInstanceOf(RuntimeException.class) + .hasMessage("Security scan is already in progress: taskId=task-inflight"); + + assertThat(securityScanner.lastRequest).isNull(); + assertThat(skillFile).exists(); + verify(processingLock, never()).unlock(); + } finally { + Files.deleteIfExists(skillFile); + Files.deleteIfExists(tempDir); + } + } + + @Test + void handleMessage_whenTaskLockIsHeld_republishesInsteadOfDroppingDelivery() { + StubSecurityScanner securityScanner = new StubSecurityScanner(); + InMemoryScanTaskProducer producer = new InMemoryScanTaskProducer(); + RLock processingLock = mock(RLock.class); + when(processingLock.tryLock()).thenReturn(false); + TestableScanTaskConsumer consumer = new TestableScanTaskConsumer( + securityScanner, + new StubSecurityScanService(), + new InMemorySkillVersionRepository(), + producer, + new InMemoryObjectStorageService(), + redissonClient(processingLock) + ); + + consumer.handleMessage(new StreamMessageId(11, 0), Map.of( + "taskId", "task-reclaimed", + "versionId", "42", + "skillPath", "/tmp/skillhub-scans/42", + "scannerType", ScannerType.SKILL_SCANNER.getValue() + )); + + assertThat(producer.publishedTask.taskId()).isEqualTo("task-reclaimed"); + assertThat(producer.publishedTask.metadata()).containsEntry("retryCount", "1"); + verify(consumer.stream).ack("skillhub-scanners", new StreamMessageId(11, 0)); + } + + @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 +386,28 @@ private TestableScanTaskConsumer(SecurityScanner securityScanner, 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 +442,19 @@ private void invokeRetryMessage(ScanTaskPayload payload, int retryCount) { } } + 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 new file mode 100644 index 000000000..eafa3b6f3 --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java @@ -0,0 +1,140 @@ +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; +import org.mockito.junit.jupiter.MockitoExtension; + +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 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() { + 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()); + + dispatcher(10).dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getRetryCount()).isEqualTo(1); + verify(producer).publishScanTask(any()); + verify(repository).save(outbox); + } + + @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 new file mode 100644 index 000000000..7da371fbb --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java @@ -0,0 +1,131 @@ +package com.iflytek.skillhub.domain.security; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.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; +import java.time.Instant; +import java.util.Map; + +@Entity +@Table(name = "scan_task_outbox") +public class ScanTaskOutbox { + @Id @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + @Column(name = "task_id", nullable = false, unique = true, length = 100) + private String taskId; + @Column(name = "version_id", nullable = false) + private Long versionId; + @Column(name = "skill_path", length = 1000) + private String skillPath; + @Column(name = "bundle_key", length = 1000) + 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) + private int retryCount; + @Column(name = "next_attempt_at", nullable = false) + private Instant nextAttemptAt; + @Column(name = "lease_until") + private Instant leaseUntil; + @Column(name = "last_error", length = 2000) + private String lastError; + @Column(name = "created_at", nullable = false, updatable = false) + private Instant createdAt; + @Column(name = "updated_at", nullable = false) + private Instant updatedAt; + @Version @Column(nullable = false) + private long entityVersion; + + protected ScanTaskOutbox() { } + + public ScanTaskOutbox(ScanTask task) { + this.taskId = task.taskId(); + this.versionId = task.versionId(); + 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; + 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()); + 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.toEpochMilli(), metadata == null ? Map.of() : Map.copyOf(metadata)); + } + + public boolean claim(Instant now, Duration lease) { + if (status != ScanTaskOutboxStatus.PENDING + && !(status == ScanTaskOutboxStatus.SENDING && leaseUntil != null && leaseUntil.isBefore(now))) return false; + status = ScanTaskOutboxStatus.SENDING; + leaseUntil = now.plus(lease); + updatedAt = now; + return true; + } + + public void markSent(Instant now) { + status = ScanTaskOutboxStatus.SENT; + leaseUntil = null; + lastError = null; + 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 = truncateError(error); + updatedAt = now; + } + + public Long getId() { return id; } + public String getTaskId() { return taskId; } + public Long getVersionId() { return versionId; } + public ScanTaskOutboxStatus getStatus() { return status; } + 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 new file mode 100644 index 000000000..7ed29c5db --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java @@ -0,0 +1,11 @@ +package com.iflytek.skillhub.domain.security; + +import java.time.Instant; +import java.util.List; + +public interface ScanTaskOutboxRepository { + ScanTaskOutbox save(ScanTaskOutbox outbox); + 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/ScanTaskOutboxStatus.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxStatus.java new file mode 100644 index 000000000..dc0fd83a3 --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxStatus.java @@ -0,0 +1,8 @@ +package com.iflytek.skillhub.domain.security; + +public enum ScanTaskOutboxStatus { + PENDING, + SENDING, + SENT, + FAILED +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java index 055287f07..0b7844de7 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java @@ -26,6 +26,9 @@ public class SecurityAudit { @Column(name = "skill_version_id", nullable = false) private Long skillVersionId; + @Column(name = "task_id", length = 100) + private String taskId; + @Column(name = "scan_id", length = 100) private String scanId; @@ -66,8 +69,13 @@ protected SecurityAudit() { } public SecurityAudit(Long skillVersionId, ScannerType scannerType) { + this(skillVersionId, scannerType, null); + } + + public SecurityAudit(Long skillVersionId, ScannerType scannerType, String taskId) { this.skillVersionId = skillVersionId; this.scannerType = scannerType; + this.taskId = taskId; this.verdict = SecurityVerdict.SUSPICIOUS; this.isSafe = false; this.findingsCount = 0; @@ -91,6 +99,10 @@ public String getScanId() { return scanId; } + public String getTaskId() { + return taskId; + } + public ScannerType getScannerType() { return scannerType; } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java index 2ae168564..9bde6ea6a 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java @@ -12,6 +12,8 @@ public interface SecurityAuditRepository { Optional findByScanId(String scanId); + boolean existsByTaskIdAndScannedAtIsNotNull(String taskId); + boolean existsBySkillVersionId(Long skillVersionId); /** 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 2d4550649..df12ee8e6 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 @@ -9,6 +9,7 @@ import com.iflytek.skillhub.domain.skill.validation.PackageEntry; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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; @@ -32,23 +33,36 @@ public class SecurityScanService { private final SecurityAuditRepository auditRepository; private final SkillVersionRepository skillVersionRepository; + private final ScanTaskOutboxRepository scanTaskOutboxRepository; private final ScanTaskProducer scanTaskProducer; private final ObjectMapper objectMapper; private final String scanMode; private final boolean enabled; + @Autowired public SecurityScanService(SecurityAuditRepository auditRepository, SkillVersionRepository skillVersionRepository, ScanTaskProducer scanTaskProducer, ObjectMapper objectMapper, @Value("${skillhub.security.scanner.mode:local}") String scanMode, - @Value("${skillhub.security.scanner.enabled:false}") boolean enabled) { + @Value("${skillhub.security.scanner.enabled:false}") boolean enabled, + ScanTaskOutboxRepository scanTaskOutboxRepository) { this.auditRepository = auditRepository; this.skillVersionRepository = skillVersionRepository; this.scanTaskProducer = scanTaskProducer; this.objectMapper = objectMapper; this.scanMode = scanMode; this.enabled = enabled; + this.scanTaskOutboxRepository = scanTaskOutboxRepository; + } + + public SecurityScanService(SecurityAuditRepository auditRepository, + SkillVersionRepository skillVersionRepository, + ScanTaskProducer scanTaskProducer, + ObjectMapper objectMapper, + String scanMode, + boolean enabled) { + this(auditRepository, skillVersionRepository, scanTaskProducer, objectMapper, scanMode, enabled, null); } public boolean isEnabled() { @@ -74,7 +88,6 @@ public void triggerScan(Long versionId, List entries, String publi packagePath = saveTempDirectory(versionId, entries).toString(); } // Always create a new audit record — supports multiple rounds per version - auditRepository.save(new SecurityAudit(versionId, ScannerType.SKILL_SCANNER)); final ScanTask scanTask = new ScanTask( UUID.randomUUID().toString(), versionId, @@ -84,9 +97,12 @@ public void triggerScan(Long versionId, List entries, String publi System.currentTimeMillis(), Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()) ); - // The stream consumer must not observe this task before skill_version / - // security_audit rows are committed and visible. - TransactionCommitCallbacks.afterCommitOrNow(() -> scanTaskProducer.publishScanTask(scanTask)); + auditRepository.save(new SecurityAudit(versionId, ScannerType.SKILL_SCANNER, scanTask.taskId())); + if (scanTaskOutboxRepository != null) { + scanTaskOutboxRepository.save(new ScanTaskOutbox(scanTask)); + } else { + TransactionCommitCallbacks.afterCommitOrNow(() -> scanTaskProducer.publishScanTask(scanTask)); + } // Only transition to SCANNING if the version is not already published (auto-publish flow) if (version.getStatus() != SkillVersionStatus.PUBLISHED) { version.setStatus(SkillVersionStatus.SCANNING); @@ -94,6 +110,11 @@ public void triggerScan(Long versionId, List entries, String publi } } + public boolean isTaskAlreadyProcessed(String taskId) { + return taskId != null && auditRepository != null + && auditRepository.existsByTaskIdAndScannedAtIsNotNull(taskId); + } + @Transactional public void processScanResult(Long versionId, ScannerType scannerType, SecurityScanResponse response) { SecurityAudit audit = auditRepository.findLatestActiveByVersionIdAndScannerType(versionId, scannerType) @@ -186,6 +207,9 @@ private String safeZipEntryName(String entryPath) { */ @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); @@ -203,5 +227,8 @@ public void softDeleteByVersionId(Long versionId) { @Transactional public void hardDeleteByVersionId(Long versionId) { auditRepository.deleteBySkillVersionId(versionId); + if (scanTaskOutboxRepository != null) { + scanTaskOutboxRepository.deleteByVersionId(versionId); + } } } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java new file mode 100644 index 000000000..8abf6ca7e --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java @@ -0,0 +1,2 @@ +/** Security scanning domain model and durable task dispatch ports. */ +package com.iflytek.skillhub.domain.security; 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 new file mode 100644 index 000000000..373baaf60 --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java @@ -0,0 +1,56 @@ +package com.iflytek.skillhub.domain.security; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class ScanTaskOutboxTest { + @Test + void claimAndMarkSentProducesStableTaskPayload() { + ScanTask task = new ScanTask("task-1", 7L, "/tmp/7", null, "u1", 123L, + Map.of( + "scannerType", ScannerType.SKILL_SCANNER.getValue(), + "futureAttribute", "preserved")); + ScanTaskOutbox outbox = new ScanTaskOutbox(task); + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + + assertThat(outbox.claim(now, Duration.ofMinutes(2))).isTrue(); + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENDING); + outbox.markSent(now.plusSeconds(1)); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT); + 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 + void failedPublishReturnsToPendingWithBackoffAndTruncatesError() { + ScanTaskOutbox outbox = new ScanTaskOutbox( + new ScanTask("task-2", 8L, null, "packages/1/8/bundle.zip", null, 1L, Map.of())); + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + outbox.claim(now, Duration.ofMinutes(2)); + outbox.markRetry(now, Duration.ofSeconds(5), "x".repeat(5000)); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getRetryCount()).isEqualTo(1); + assertThat(outbox.getNextAttemptAt()).isEqualTo(now.plusSeconds(5)); + } +} 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 new file mode 100644 index 000000000..f0e056bf1 --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java @@ -0,0 +1,59 @@ +package com.iflytek.skillhub.domain.security; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.iflytek.skillhub.domain.skill.SkillVersion; +import com.iflytek.skillhub.domain.skill.SkillVersionRepository; +import com.iflytek.skillhub.domain.skill.validation.PackageEntry; +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 java.lang.reflect.Field; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +@ExtendWith(MockitoExtension.class) +class SecurityScanOutboxTest { + @Mock SecurityAuditRepository auditRepository; + @Mock SkillVersionRepository versionRepository; + @Mock ScanTaskProducer producer; + @Mock ScanTaskOutboxRepository outboxRepository; + + @Test + void triggerPersistsAuditStateAndOutboxWithoutPublishingInsideTransaction() throws Exception { + SkillVersion version = new SkillVersion(9L, "1.0.0", "publisher"); + Field id = SkillVersion.class.getDeclaredField("id"); + id.setAccessible(true); + id.set(version, 42L); + given(versionRepository.findById(42L)).willReturn(Optional.of(version)); + SecurityScanService service = new SecurityScanService(auditRepository, versionRepository, producer, + new ObjectMapper(), "upload", true, outboxRepository); + + service.triggerScan(42L, List.of(new PackageEntry("SKILL.md", new byte[0], 0, "text/markdown")), "publisher"); + + ArgumentCaptor outbox = ArgumentCaptor.forClass(ScanTaskOutbox.class); + verify(outboxRepository).save(outbox.capture()); + verify(producer, never()).publishScanTask(org.mockito.ArgumentMatchers.any()); + assertThat(outbox.getValue().getVersionId()).isEqualTo(42L); + assertThat(outbox.getValue().getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); + } + + @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 new file mode 100644 index 000000000..84b089939 --- /dev/null +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java @@ -0,0 +1,34 @@ +package com.iflytek.skillhub.infra.jpa; + +import com.iflytek.skillhub.domain.security.ScanTaskOutbox; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import org.springframework.stereotype.Repository; + +import java.time.Instant; +import java.util.List; + +@Repository +public interface ScanTaskOutboxJpaRepository extends JpaRepository, ScanTaskOutboxRepository { + @Override + @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 + @Query("DELETE FROM ScanTaskOutbox o WHERE o.status = com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.SENT AND o.updatedAt < :cutoff") + int deleteSentBefore(@Param("cutoff") Instant cutoff); + + @Override + int deleteByVersionId(Long versionId); +} \ No newline at end of file