fix(scan): prevent scan tasks from being lost after transaction commit

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<ScanTaskConsumer.ScanTaskPayload> {
     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,13 +132,35 @@ 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();
+                return;
+            }
+            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);
     }
@@ -259,6 +285,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 +334,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;
         }

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));
+    }
+}

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);

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;

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<String, String> 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,