From 2190e52a9c9975f23f13092b525d470e5045228c Mon Sep 17 00:00:00 2001 From: bbdu3 Date: Mon, 24 Aug 2026 16:57:29 +0800 Subject: [PATCH 1/5] fix security-scan Signed-off-by: bbdu3 --- .../skillhub/stream/ScanTaskConsumer.java | 4 + .../task/ScanTaskOutboxDispatcher.java | 81 +++++++++++++ .../db/migration/V44__scan_task_outbox.sql | 28 +++++ .../task/ScanTaskOutboxDispatcherTest.java | 49 ++++++++ .../domain/security/ScanTaskOutbox.java | 110 ++++++++++++++++++ .../security/ScanTaskOutboxRepository.java | 13 +++ .../domain/security/ScanTaskOutboxStatus.java | 8 ++ .../domain/security/SecurityAudit.java | 12 ++ .../security/SecurityAuditRepository.java | 2 + .../domain/security/SecurityScanService.java | 33 +++++- .../domain/security/package-info.java | 2 + .../domain/security/ScanTaskOutboxTest.java | 40 +++++++ .../security/SecurityScanOutboxTest.java | 47 ++++++++ .../jpa/ScanTaskOutboxJpaRepository.java | 43 +++++++ 14 files changed, 467 insertions(+), 5 deletions(-) create mode 100644 server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcher.java create mode 100644 server/skillhub-app/src/main/resources/db/migration/V44__scan_task_outbox.sql create mode 100644 server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java create mode 100644 server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java create mode 100644 server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java create mode 100644 server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxStatus.java create mode 100644 server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/package-info.java create mode 100644 server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java create mode 100644 server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java create mode 100644 server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java 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..31e6550cc 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 @@ -128,6 +128,10 @@ 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; + } String skillPath = resolveWorkingSkillPath(payload); SecurityScanRequest request = new SecurityScanRequest( payload.taskId(), 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..efe87b8a4 --- /dev/null +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcher.java @@ -0,0 +1,81 @@ +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 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; +import java.util.LinkedHashMap; +import java.util.Map; + +@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 Clock clock; + private final int batchSize; + private final Duration lease; + private final Duration maxBackoff; + + public ScanTaskOutboxDispatcher(ScanTaskOutboxRepository repository, + ScanTaskProducer producer, + Clock clock, + @Value("${skillhub.security.outbox.batch-size:50}") int batchSize, + @Value("${skillhub.security.outbox.lease:PT2M}") Duration lease, + @Value("${skillhub.security.outbox.max-backoff:PT5M}") Duration maxBackoff) { + this.repository = repository; + this.producer = producer; + this.clock = clock; + this.batchSize = batchSize; + this.lease = lease; + this.maxBackoff = maxBackoff; + } + + @Scheduled(fixedDelayString = "${skillhub.security.outbox.dispatch-interval-ms:5000}") + @Transactional + public void dispatch() { + Instant now = Instant.now(clock); + Map candidates = new LinkedHashMap<>(); + repository.findPendingDue(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); + repository.findExpiredLeases(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); + for (ScanTaskOutbox outbox : candidates.values()) { + if (!outbox.claim(now, lease)) continue; + repository.saveAndFlush(outbox); + try { + producer.publishScanTask(outbox.toScanTask()); + outbox.markSent(Instant.now(clock)); + repository.save(outbox); + } catch (Exception e) { + Duration delay = retryDelay(outbox.getRetryCount() + 1); + outbox.markRetry(Instant.now(clock), delay, e.toString()); + repository.save(outbox); + log.warn("Failed to publish scan task; will retry taskId={}, retryCount={}, nextDelay={}", + outbox.getTaskId(), outbox.getRetryCount(), delay, e); + } + } + } + + @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)); + } +} \ No newline at end of file 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/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..eae1e0eca --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java @@ -0,0 +1,49 @@ +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.ScanTaskProducer; +import com.iflytek.skillhub.domain.security.ScannerType; +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 static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; + +@ExtendWith(MockitoExtension.class) +class ScanTaskOutboxDispatcherTest { + @Mock ScanTaskOutboxRepository repository; + @Mock ScanTaskProducer producer; + + @Test + void failedRedisPublishLeavesTaskPendingForRetry() { + Clock clock = Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC); + ScanTaskOutbox outbox = new ScanTaskOutbox( + new ScanTask("task-1", 1L, "/tmp/1", null, "user", 1L, + java.util.Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()))); + given(repository.findPendingDue(any(), any(Integer.class))).willReturn(List.of(outbox)); + given(repository.findExpiredLeases(any(), any(Integer.class))).willReturn(List.of()); + doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any()); + ScanTaskOutboxDispatcher dispatcher = new ScanTaskOutboxDispatcher( + repository, producer, clock, 50, Duration.ofMinutes(2), Duration.ofMinutes(5)); + + dispatcher.dispatch(); + + assertThat(outbox.getStatus()).isEqualTo(com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getRetryCount()).isEqualTo(1); + verify(producer).publishScanTask(any()); + verify(repository).saveAndFlush(outbox); + } +} \ No newline at end of file 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..8f93fce17 --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java @@ -0,0 +1,110 @@ +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 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; + @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.status = ScanTaskOutboxStatus.PENDING; + this.nextAttemptAt = Instant.now(Clock.systemUTC()); + } + + @PrePersist + protected void onCreate() { + Instant now = Instant.now(Clock.systemUTC()); + createdAt = now; + updatedAt = now; + if (nextAttemptAt == null) nextAttemptAt = now; + } + + public ScanTask toScanTask() { + return new ScanTask(taskId, versionId, skillPath, bundleKey, publisherId, + createdAt == null ? System.currentTimeMillis() : createdAt.toEpochMilli(), + Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + } + + 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 markRetry(Instant now, Duration delay, String error) { + retryCount++; + status = ScanTaskOutboxStatus.PENDING; + nextAttemptAt = now.plus(delay); + leaseUntil = null; + lastError = error == null ? null : error.substring(0, Math.min(error.length(), 2000)); + 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; } + 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..43c5c24e1 --- /dev/null +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java @@ -0,0 +1,13 @@ +package com.iflytek.skillhub.domain.security; + +import java.time.Instant; +import java.util.List; + +public interface ScanTaskOutboxRepository { + ScanTaskOutbox save(ScanTaskOutbox outbox); + ScanTaskOutbox saveAndFlush(ScanTaskOutbox outbox); + List findPendingDue(Instant now, int limit); + List findExpiredLeases(Instant now, int limit); + 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..43aa8962a 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,10 @@ public void triggerScan(Long versionId, List entries, String publi } } + public boolean isTaskAlreadyProcessed(String taskId) { + return taskId != null && auditRepository.existsByTaskIdAndScannedAtIsNotNull(taskId); + } + @Transactional public void processScanResult(Long versionId, ScannerType scannerType, SecurityScanResponse response) { SecurityAudit audit = auditRepository.findLatestActiveByVersionIdAndScannerType(versionId, scannerType) @@ -203,5 +223,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..968fd1e89 --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java @@ -0,0 +1,40 @@ +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())); + 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().taskId()).isEqualTo("task-1"); + assertThat(outbox.toScanTask().versionId()).isEqualTo(7L); + } + + @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)); + } +} \ No newline at end of file diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java new file mode 100644 index 000000000..ce0b247e7 --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java @@ -0,0 +1,47 @@ +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); + } +} \ No newline at end of file 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..04f7eb4e6 --- /dev/null +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java @@ -0,0 +1,43 @@ +package com.iflytek.skillhub.infra.jpa; + +import com.iflytek.skillhub.domain.security.ScanTaskOutbox; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +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 + default List findPendingDue(Instant now, int limit) { + return findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( + ScanTaskOutboxStatus.PENDING, now, PageRequest.of(0, limit)); + } + + @Override + default List findExpiredLeases(Instant now, int limit) { + return findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( + ScanTaskOutboxStatus.SENDING, now, PageRequest.of(0, limit)); + } + + @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 + void deleteByVersionId(Long versionId); + + List findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( + ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); + + List findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( + ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); +} \ No newline at end of file From 6c55d7adf4ad1c487612c380324b49e7cc7fd849 Mon Sep 17 00:00:00 2001 From: bbdu3 Date: Mon, 24 Aug 2026 17:36:20 +0800 Subject: [PATCH 2/5] fix(security-scan): align outbox repository delete signature Signed-off-by: bbdu3 --- .../iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java index 04f7eb4e6..34cd6be09 100644 --- a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java @@ -33,7 +33,7 @@ default List findExpiredLeases(Instant now, int limit) { int deleteSentBefore(@Param("cutoff") Instant cutoff); @Override - void deleteByVersionId(Long versionId); + int deleteByVersionId(Long versionId); List findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); From 28031ca99c8ca2cd42b2ad43cb818e729485d07b Mon Sep 17 00:00:00 2001 From: bbdu3 Date: Mon, 24 Aug 2026 18:05:57 +0800 Subject: [PATCH 3/5] fix(security-scan): guard idempotency check without audit repository Signed-off-by: bbdu3 --- .../iflytek/skillhub/domain/security/SecurityScanService.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 43aa8962a..5efccbd06 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 @@ -111,7 +111,8 @@ public void triggerScan(Long versionId, List entries, String publi } public boolean isTaskAlreadyProcessed(String taskId) { - return taskId != null && auditRepository.existsByTaskIdAndScannedAtIsNotNull(taskId); + return taskId != null && auditRepository != null + && auditRepository.existsByTaskIdAndScannedAtIsNotNull(taskId); } @Transactional From 15566650f006323e24a6e32d7eb5ae0c61cd89e6 Mon Sep 17 00:00:00 2001 From: bbdu3 Date: Tue, 25 Aug 2026 10:28:38 +0800 Subject: [PATCH 4/5] fix(security-scan): harden outbox retries and concurrency Signed-off-by: bbdu3 --- .../skillhub/stream/ScanTaskConsumer.java | 40 +++++- .../task/ScanTaskOutboxDispatcher.java | 58 ++++++--- .../V45__scan_task_outbox_metadata.sql | 2 + .../stream/ScanTaskConsumerLoggingTest.java | 13 +- .../skillhub/stream/ScanTaskConsumerTest.java | 95 ++++++++++++++- .../task/ScanTaskOutboxDispatcherTest.java | 115 ++++++++++++++++-- .../domain/security/ScanTaskOutbox.java | 33 ++++- .../security/ScanTaskOutboxRepository.java | 4 +- .../domain/security/SecurityScanService.java | 3 + .../domain/security/ScanTaskOutboxTest.java | 24 +++- .../security/SecurityScanOutboxTest.java | 14 ++- .../jpa/ScanTaskOutboxJpaRepository.java | 27 ++-- 12 files changed, 362 insertions(+), 66 deletions(-) create mode 100644 server/skillhub-app/src/main/resources/db/migration/V45__scan_task_outbox_metadata.sql 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 31e6550cc..f32a28768 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; @@ -132,13 +136,31 @@ protected void processBusiness(ScanTaskPayload payload) { 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); } @@ -263,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); @@ -311,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; } 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 index efe87b8a4..242a32d20 100644 --- 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 @@ -3,6 +3,8 @@ 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; @@ -14,8 +16,6 @@ import java.time.Clock; import java.time.Duration; import java.time.Instant; -import java.util.LinkedHashMap; -import java.util.Map; @Component @ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true") @@ -24,21 +24,30 @@ public class ScanTaskOutboxDispatcher { 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; } @@ -47,35 +56,54 @@ public ScanTaskOutboxDispatcher(ScanTaskOutboxRepository repository, @Transactional public void dispatch() { Instant now = Instant.now(clock); - Map candidates = new LinkedHashMap<>(); - repository.findPendingDue(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); - repository.findExpiredLeases(now, batchSize).forEach(o -> candidates.put(o.getTaskId(), o)); - for (ScanTaskOutbox outbox : candidates.values()) { - if (!outbox.claim(now, lease)) continue; - repository.saveAndFlush(outbox); + for (ScanTaskOutbox outbox : repository.findDispatchable(now, batchSize)) { + if (!outbox.claim(now, lease)) { + continue; + } try { producer.publishScanTask(outbox.toScanTask()); outbox.markSent(Instant.now(clock)); repository.save(outbox); } catch (Exception e) { - Duration delay = retryDelay(outbox.getRetryCount() + 1); - outbox.markRetry(Instant.now(clock), delay, e.toString()); - repository.save(outbox); - log.warn("Failed to publish scan task; will retry taskId={}, retryCount={}, nextDelay={}", - outbox.getTaskId(), outbox.getRetryCount(), delay, e); + handlePublishFailure(outbox, e); } } } + private void handlePublishFailure(ScanTaskOutbox outbox, Exception error) { + Instant now = Instant.now(clock); + int nextAttempt = outbox.getRetryCount() + 1; + if (nextAttempt >= maxAttempts) { + outbox.markFailed(now, error.toString()); + repository.save(outbox); + versionRepository.findById(outbox.getVersionId()) + .filter(version -> version.getStatus() == SkillVersionStatus.SCANNING) + .ifPresent(version -> { + version.setStatus(SkillVersionStatus.SCAN_FAILED); + versionRepository.save(version); + }); + log.error("Scan task publish failed permanently: taskId={}, versionId={}, attempts={}", + outbox.getTaskId(), outbox.getVersionId(), outbox.getRetryCount(), error); + return; + } + Duration delay = retryDelay(nextAttempt); + outbox.markRetry(now, delay, error.toString()); + repository.save(outbox); + log.warn("Failed to publish scan task; will retry taskId={}, retryCount={}, nextDelay={}", + outbox.getTaskId(), outbox.getRetryCount(), delay, error); + } + @Scheduled(cron = "0 20 2 * * ?") @Transactional public void cleanupSent() { int deleted = repository.deleteSentBefore(Instant.now(clock).minus(Duration.ofDays(7))); - if (deleted > 0) log.info("Cleaned up {} sent scan outbox records", deleted); + if (deleted > 0) { + log.info("Cleaned up {} sent scan outbox records", deleted); + } } private Duration retryDelay(int retryCount) { long seconds = Math.min(maxBackoff.toSeconds(), 1L << Math.min(retryCount, 16)); return Duration.ofSeconds(Math.max(seconds, 1)); } -} \ No newline at end of file +} diff --git a/server/skillhub-app/src/main/resources/db/migration/V45__scan_task_outbox_metadata.sql b/server/skillhub-app/src/main/resources/db/migration/V45__scan_task_outbox_metadata.sql new file mode 100644 index 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..8456d11d0 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,60 @@ 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 { + consumer.invokeProcessBusiness(payload); + consumer.invokeMarkCompleted(payload); + + assertThat(securityScanner.lastRequest).isNull(); + assertThat(skillFile).exists(); + verify(processingLock, never()).unlock(); + } finally { + Files.deleteIfExists(skillFile); + Files.deleteIfExists(tempDir); + } + } + + @Test + void processBusiness_whenScannerFails_releasesProcessingLock() { + StubSecurityScanner securityScanner = new StubSecurityScanner(); + securityScanner.failure = new IllegalStateException("scanner unavailable"); + RLock processingLock = availableProcessingLock(); + TestableScanTaskConsumer consumer = new TestableScanTaskConsumer( + securityScanner, + new StubSecurityScanService(), + new InMemorySkillVersionRepository(), + new InMemoryScanTaskProducer(), + new InMemoryObjectStorageService(), + redissonClient(processingLock) + ); + ScanTaskConsumer.ScanTaskPayload payload = new ScanTaskConsumer.ScanTaskPayload( + "task-failure", 42L, "/tmp/failure", null, ScannerType.SKILL_SCANNER); + + assertThatThrownBy(() -> consumer.invokeProcessBusiness(payload)) + .isInstanceOf(IllegalStateException.class) + .hasMessage("scanner unavailable"); + verify(processingLock).unlock(); + } + private void setField(Object target, String fieldName, Object value) throws Exception { Field field = target.getClass().getDeclaredField(fieldName); field.setAccessible(true); @@ -299,7 +358,28 @@ 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 +414,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 index eae1e0eca..eafa3b6f3 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ScanTaskOutboxDispatcherTest.java @@ -3,8 +3,12 @@ import com.iflytek.skillhub.domain.security.ScanTask; import com.iflytek.skillhub.domain.security.ScanTaskOutbox; import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; +import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus; import com.iflytek.skillhub.domain.security.ScanTaskProducer; import com.iflytek.skillhub.domain.security.ScannerType; +import com.iflytek.skillhub.domain.skill.SkillVersion; +import com.iflytek.skillhub.domain.skill.SkillVersionRepository; +import com.iflytek.skillhub.domain.skill.SkillVersionStatus; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; @@ -15,35 +19,122 @@ import java.time.Instant; import java.time.ZoneOffset; import java.util.List; +import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; @ExtendWith(MockitoExtension.class) class ScanTaskOutboxDispatcherTest { @Mock ScanTaskOutboxRepository repository; @Mock ScanTaskProducer producer; + @Mock SkillVersionRepository versionRepository; @Test void failedRedisPublishLeavesTaskPendingForRetry() { - Clock clock = Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC); - ScanTaskOutbox outbox = new ScanTaskOutbox( - new ScanTask("task-1", 1L, "/tmp/1", null, "user", 1L, - java.util.Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()))); - given(repository.findPendingDue(any(), any(Integer.class))).willReturn(List.of(outbox)); - given(repository.findExpiredLeases(any(), any(Integer.class))).willReturn(List.of()); + ScanTaskOutbox outbox = outbox("task-1", 1L); + given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox)); doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any()); - ScanTaskOutboxDispatcher dispatcher = new ScanTaskOutboxDispatcher( - repository, producer, clock, 50, Duration.ofMinutes(2), Duration.ofMinutes(5)); - dispatcher.dispatch(); + dispatcher(10).dispatch(); - assertThat(outbox.getStatus()).isEqualTo(com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.PENDING); + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); assertThat(outbox.getRetryCount()).isEqualTo(1); verify(producer).publishScanTask(any()); - verify(repository).saveAndFlush(outbox); + verify(repository).save(outbox); + } + + @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()))); } -} \ No newline at end of file +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java index 8f93fce17..7da371fbb 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutbox.java @@ -10,6 +10,8 @@ import jakarta.persistence.PrePersist; import jakarta.persistence.Table; import jakarta.persistence.Version; +import org.hibernate.annotations.JdbcTypeCode; +import org.hibernate.type.SqlTypes; import java.time.Clock; import java.time.Duration; @@ -31,6 +33,9 @@ public class ScanTaskOutbox { private String bundleKey; @Column(name = "publisher_id", length = 255) private String publisherId; + @JdbcTypeCode(SqlTypes.JSON) + @Column(name = "metadata", nullable = false, columnDefinition = "jsonb") + private Map metadata; @Enumerated(EnumType.STRING) @Column(nullable = false, length = 20) private ScanTaskOutboxStatus status; @Column(name = "retry_count", nullable = false) @@ -56,22 +61,25 @@ public ScanTaskOutbox(ScanTask task) { this.skillPath = task.skillPath(); this.bundleKey = task.bundleKey(); this.publisherId = task.publisherId(); + this.metadata = task.metadata() == null ? Map.of() : Map.copyOf(task.metadata()); this.status = ScanTaskOutboxStatus.PENDING; - this.nextAttemptAt = Instant.now(Clock.systemUTC()); + Instant taskCreatedAt = Instant.ofEpochMilli(task.createdAtMillis()); + this.nextAttemptAt = taskCreatedAt; + this.createdAt = taskCreatedAt; + this.updatedAt = taskCreatedAt; } @PrePersist protected void onCreate() { Instant now = Instant.now(Clock.systemUTC()); - createdAt = now; - updatedAt = now; + if (createdAt == null) createdAt = now; + if (updatedAt == null) updatedAt = now; if (nextAttemptAt == null) nextAttemptAt = now; } public ScanTask toScanTask() { return new ScanTask(taskId, versionId, skillPath, bundleKey, publisherId, - createdAt == null ? System.currentTimeMillis() : createdAt.toEpochMilli(), - Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + createdAt.toEpochMilli(), metadata == null ? Map.of() : Map.copyOf(metadata)); } public boolean claim(Instant now, Duration lease) { @@ -90,12 +98,20 @@ public void markSent(Instant now) { updatedAt = now; } + public void markFailed(Instant now, String error) { + retryCount++; + status = ScanTaskOutboxStatus.FAILED; + leaseUntil = null; + lastError = truncateError(error); + updatedAt = now; + } + public void markRetry(Instant now, Duration delay, String error) { retryCount++; status = ScanTaskOutboxStatus.PENDING; nextAttemptAt = now.plus(delay); leaseUntil = null; - lastError = error == null ? null : error.substring(0, Math.min(error.length(), 2000)); + lastError = truncateError(error); updatedAt = now; } @@ -106,5 +122,10 @@ public void markRetry(Instant now, Duration delay, String error) { public int getRetryCount() { return retryCount; } public Instant getNextAttemptAt() { return nextAttemptAt; } public Instant getLeaseUntil() { return leaseUntil; } + + private String truncateError(String error) { + return error == null ? null : error.substring(0, Math.min(error.length(), 2000)); + } + public Instant getCreatedAt() { return createdAt; } } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java index 43c5c24e1..7ed29c5db 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxRepository.java @@ -5,9 +5,7 @@ public interface ScanTaskOutboxRepository { ScanTaskOutbox save(ScanTaskOutbox outbox); - ScanTaskOutbox saveAndFlush(ScanTaskOutbox outbox); - List findPendingDue(Instant now, int limit); - List findExpiredLeases(Instant now, int limit); + List findDispatchable(Instant now, int limit); int deleteSentBefore(Instant cutoff); int deleteByVersionId(Long versionId); } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityScanService.java index 5efccbd06..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 @@ -207,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); diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java index 968fd1e89..373baaf60 100644 --- a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/ScanTaskOutboxTest.java @@ -12,7 +12,9 @@ class ScanTaskOutboxTest { @Test void claimAndMarkSentProducesStableTaskPayload() { ScanTask task = new ScanTask("task-1", 7L, "/tmp/7", null, "u1", 123L, - Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())); + Map.of( + "scannerType", ScannerType.SKILL_SCANNER.getValue(), + "futureAttribute", "preserved")); ScanTaskOutbox outbox = new ScanTaskOutbox(task); Instant now = Instant.parse("2026-01-01T00:00:00Z"); @@ -21,8 +23,22 @@ void claimAndMarkSentProducesStableTaskPayload() { outbox.markSent(now.plusSeconds(1)); assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT); - assertThat(outbox.toScanTask().taskId()).isEqualTo("task-1"); - assertThat(outbox.toScanTask().versionId()).isEqualTo(7L); + assertThat(outbox.toScanTask()).isEqualTo(task); + } + + @Test + void exhaustedPublishAttemptsMoveTaskToFailed() { + ScanTaskOutbox outbox = new ScanTaskOutbox( + new ScanTask("task-failed", 9L, null, "bundle.zip", null, 1L, Map.of())); + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + outbox.claim(now, Duration.ofMinutes(2)); + + outbox.markFailed(now, "permanent failure"); + + assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.FAILED); + assertThat(outbox.getRetryCount()).isEqualTo(1); + assertThat(outbox.getLeaseUntil()).isNull(); + assertThat(outbox.claim(now.plusSeconds(1), Duration.ofMinutes(2))).isFalse(); } @Test @@ -37,4 +53,4 @@ void failedPublishReturnsToPendingWithBackoffAndTruncatesError() { assertThat(outbox.getRetryCount()).isEqualTo(1); assertThat(outbox.getNextAttemptAt()).isEqualTo(now.plusSeconds(5)); } -} \ No newline at end of file +} diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java index ce0b247e7..f0e056bf1 100644 --- a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanOutboxTest.java @@ -44,4 +44,16 @@ void triggerPersistsAuditStateAndOutboxWithoutPublishingInsideTransaction() thro assertThat(outbox.getValue().getVersionId()).isEqualTo(42L); assertThat(outbox.getValue().getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING); } -} \ No newline at end of file + + @Test + void softDeleteRemovesPendingOutboxEvenWhenNoActiveAuditExists() { + given(auditRepository.findAllActiveBySkillVersionId(42L)).willReturn(List.of()); + SecurityScanService service = new SecurityScanService(auditRepository, versionRepository, producer, + new ObjectMapper(), "upload", true, outboxRepository); + + service.softDeleteByVersionId(42L); + + verify(outboxRepository).deleteByVersionId(42L); + verify(auditRepository, never()).saveAll(org.mockito.ArgumentMatchers.anyList()); + } +} diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java index 34cd6be09..84b089939 100644 --- a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ScanTaskOutboxJpaRepository.java @@ -2,8 +2,6 @@ import com.iflytek.skillhub.domain.security.ScanTaskOutbox; import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository; -import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus; -import org.springframework.data.domain.PageRequest; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; @@ -16,16 +14,15 @@ @Repository public interface ScanTaskOutboxJpaRepository extends JpaRepository, ScanTaskOutboxRepository { @Override - default List findPendingDue(Instant now, int limit) { - return findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( - ScanTaskOutboxStatus.PENDING, now, PageRequest.of(0, limit)); - } - - @Override - default List findExpiredLeases(Instant now, int limit) { - return findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( - ScanTaskOutboxStatus.SENDING, now, PageRequest.of(0, limit)); - } + @Query(value = """ + SELECT * FROM scan_task_outbox + WHERE (status = 'PENDING' AND next_attempt_at <= :now) + OR (status = 'SENDING' AND lease_until < :now) + ORDER BY created_at + LIMIT :limit + FOR UPDATE SKIP LOCKED + """, nativeQuery = true) + List findDispatchable(@Param("now") Instant now, @Param("limit") int limit); @Override @Modifying @@ -34,10 +31,4 @@ default List findExpiredLeases(Instant now, int limit) { @Override int deleteByVersionId(Long versionId); - - List findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc( - ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); - - List findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc( - ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable); } \ No newline at end of file From 3ef425a5a452ea4a02d44d6eefb1d0ae144deb11 Mon Sep 17 00:00:00 2001 From: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com> Date: Wed, 26 Aug 2026 17:22:35 +0800 Subject: [PATCH 5/5] fix(scan): retry reclaimed task lock contention Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com> --- .../skillhub/stream/ScanTaskConsumer.java | 12 ++++++- .../skillhub/stream/ScanTaskConsumerTest.java | 32 +++++++++++++++++-- 2 files changed, 41 insertions(+), 3 deletions(-) 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 f32a28768..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 @@ -144,7 +144,11 @@ protected void processBusiness(ScanTaskPayload payload) { log.info("Skipping concurrently processed security scan task: taskId={}, versionId={}", payload.taskId(), payload.versionId()); payload.skipCleanup(); - return; + // 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; @@ -165,6 +169,12 @@ private void executeScan(ScanTaskPayload payload) { 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()); 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 8456d11d0..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 @@ -293,8 +293,9 @@ void processBusiness_whenTaskIsAlreadyInFlight_skipsScanAndPreservesSharedTempPa "task-inflight", 42L, tempDir.toString(), null, ScannerType.SKILL_SCANNER); try { - consumer.invokeProcessBusiness(payload); - consumer.invokeMarkCompleted(payload); + assertThatThrownBy(() -> consumer.invokeProcessBusiness(payload)) + .isInstanceOf(RuntimeException.class) + .hasMessage("Security scan is already in progress: taskId=task-inflight"); assertThat(securityScanner.lastRequest).isNull(); assertThat(skillFile).exists(); @@ -305,6 +306,33 @@ void processBusiness_whenTaskIsAlreadyInFlight_skipsScanAndPreservesSharedTempPa } } + @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();