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

Preserve scan delivery on lock contention through the retry path and add reclaim regression coverage.

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<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 class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
                             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 class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
                 reclaimInterval,
                 messageObservationSupport
         );
+        this.redissonClient = redissonClient;
         this.securityScanner = securityScanner;
         this.securityScanService = securityScanService;
         this.skillVersionRepository = skillVersionRepository;
@@ -128,17 +132,49 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
 
     @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 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
         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 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
         }
 
         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 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<String, String> 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,

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,88 @@ 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 {
+            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 @@ 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 +442,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;

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

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<String, String> 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; }
+}

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<ScanTaskOutbox> findDispatchable(Instant now, int limit);
+    int deleteSentBefore(Instant cutoff);
+    int deleteByVersionId(Long versionId);
+}

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

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 @@ public class 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 class SecurityAudit {
         return scanId;
     }
 
+    public String getTaskId() {
+        return taskId;
+    }
+
     public ScannerType getScannerType() {
         return scannerType;
     }

server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAuditRepository.java

@@ -12,6 +12,8 @@ public interface SecurityAuditRepository {
 
     Optional<SecurityAudit> findByScanId(String scanId);
 
+    boolean existsByTaskIdAndScannedAtIsNotNull(String taskId);
+
     boolean existsBySkillVersionId(Long skillVersionId);
 
     /**

server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java

@@ -9,6 +9,7 @@ import com.iflytek.skillhub.domain.skill.SkillVersionStatus;
 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 class SecurityScanService {
             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 class SecurityScanService {
                 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 class SecurityScanService {
         }
     }
 
+    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 @@ public class SecurityScanService {
      */
     @Transactional
     public void softDeleteByVersionId(Long versionId) {
+        if (scanTaskOutboxRepository != null) {
+            scanTaskOutboxRepository.deleteByVersionId(versionId);
+        }
         List<SecurityAudit> audits = auditRepository.findAllActiveBySkillVersionId(versionId);
         if (audits.isEmpty()) {
             log.debug("No active security audits to soft-delete for versionId={}", versionId);
@@ -203,5 +227,8 @@ public class SecurityScanService {
     @Transactional
     public void hardDeleteByVersionId(Long versionId) {
         auditRepository.deleteBySkillVersionId(versionId);
+        if (scanTaskOutboxRepository != null) {
+            scanTaskOutboxRepository.deleteByVersionId(versionId);
+        }
     }
 }

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;

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

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

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