mirror of
https://github.com/iflytek/skillhub.git
synced 2026-10-10 03:27:54 +00:00
fix(security-scan): harden outbox retries and concurrency
Signed-off-by: bbdu3 <ergouyang854@gmail.com>
This commit is contained in:
bbdu3
parent
28031ca99c
commit
15566650f0
12 changed files with 362 additions and 66 deletions
|
|
@ -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;
|
||||
|
|
@ -132,13 +136,31 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
|
|||
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 @@ 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);
|
||||
|
|
@ -311,9 +334,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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ 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;
|
||||
|
|
@ -14,8 +16,6 @@ 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")
|
||||
|
|
@ -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 class ScanTaskOutboxDispatcher {
|
|||
@Transactional
|
||||
public void dispatch() {
|
||||
Instant now = Instant.now(clock);
|
||||
Map<String, ScanTaskOutbox> 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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,2 @@
|
|||
ALTER TABLE scan_task_outbox
|
||||
ADD COLUMN metadata JSONB NOT NULL DEFAULT '{}'::jsonb;
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import com.iflytek.skillhub.storage.ObjectStorageService;
|
|||
import com.iflytek.skillhub.storage.ObjectMetadata;
|
||||
import io.micrometer.observation.ObservationRegistry;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.redisson.api.RLock;
|
||||
import org.redisson.api.RStream;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.redisson.api.StreamMessageId;
|
||||
|
|
@ -38,7 +39,11 @@ import java.util.Map;
|
|||
import java.util.Optional;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
class ScanTaskConsumerTest {
|
||||
private static final Path SCAN_TEMP_DIR = Path.of("/tmp/skillhub-scans");
|
||||
|
|
@ -268,6 +273,60 @@ class ScanTaskConsumerTest {
|
|||
assertThat(listScanTempFiles(versionId)).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void processBusiness_whenTaskIsAlreadyInFlight_skipsScanAndPreservesSharedTempPath() throws Exception {
|
||||
Files.createDirectories(SCAN_TEMP_DIR);
|
||||
Path tempDir = Files.createTempDirectory(SCAN_TEMP_DIR, "scan-task-consumer-inflight");
|
||||
Path skillFile = Files.writeString(tempDir.resolve("SKILL.md"), "# demo");
|
||||
StubSecurityScanner securityScanner = new StubSecurityScanner();
|
||||
RLock processingLock = mock(RLock.class);
|
||||
when(processingLock.tryLock()).thenReturn(false);
|
||||
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
|
||||
securityScanner,
|
||||
new StubSecurityScanService(),
|
||||
new InMemorySkillVersionRepository(),
|
||||
new InMemoryScanTaskProducer(),
|
||||
new InMemoryObjectStorageService(),
|
||||
redissonClient(processingLock)
|
||||
);
|
||||
ScanTaskConsumer.ScanTaskPayload payload = new ScanTaskConsumer.ScanTaskPayload(
|
||||
"task-inflight", 42L, tempDir.toString(), null, ScannerType.SKILL_SCANNER);
|
||||
|
||||
try {
|
||||
consumer.invokeProcessBusiness(payload);
|
||||
consumer.invokeMarkCompleted(payload);
|
||||
|
||||
assertThat(securityScanner.lastRequest).isNull();
|
||||
assertThat(skillFile).exists();
|
||||
verify(processingLock, never()).unlock();
|
||||
} finally {
|
||||
Files.deleteIfExists(skillFile);
|
||||
Files.deleteIfExists(tempDir);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void processBusiness_whenScannerFails_releasesProcessingLock() {
|
||||
StubSecurityScanner securityScanner = new StubSecurityScanner();
|
||||
securityScanner.failure = new IllegalStateException("scanner unavailable");
|
||||
RLock processingLock = availableProcessingLock();
|
||||
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
|
||||
securityScanner,
|
||||
new StubSecurityScanService(),
|
||||
new InMemorySkillVersionRepository(),
|
||||
new InMemoryScanTaskProducer(),
|
||||
new InMemoryObjectStorageService(),
|
||||
redissonClient(processingLock)
|
||||
);
|
||||
ScanTaskConsumer.ScanTaskPayload payload = new ScanTaskConsumer.ScanTaskPayload(
|
||||
"task-failure", 42L, "/tmp/failure", null, ScannerType.SKILL_SCANNER);
|
||||
|
||||
assertThatThrownBy(() -> consumer.invokeProcessBusiness(payload))
|
||||
.isInstanceOf(IllegalStateException.class)
|
||||
.hasMessage("scanner unavailable");
|
||||
verify(processingLock).unlock();
|
||||
}
|
||||
|
||||
private void setField(Object target, String fieldName, Object value) throws Exception {
|
||||
Field field = target.getClass().getDeclaredField(fieldName);
|
||||
field.setAccessible(true);
|
||||
|
|
@ -299,7 +358,28 @@ class ScanTaskConsumerTest {
|
|||
ScanTaskProducer scanTaskProducer,
|
||||
ObjectStorageService objectStorageService) {
|
||||
super(
|
||||
mock(RedissonClient.class),
|
||||
redissonClient(availableProcessingLock()),
|
||||
"skillhub:scan:requests",
|
||||
"skillhub-scanners",
|
||||
securityScanner,
|
||||
securityScanService,
|
||||
skillVersionRepository,
|
||||
scanTaskProducer,
|
||||
objectStorageService,
|
||||
new MessageObservationSupport(ObservationRegistry.NOOP, new RequestIdAccessor())
|
||||
);
|
||||
this.stream = mock(RStream.class);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private TestableScanTaskConsumer(SecurityScanner securityScanner,
|
||||
SecurityScanService securityScanService,
|
||||
SkillVersionRepository skillVersionRepository,
|
||||
ScanTaskProducer scanTaskProducer,
|
||||
ObjectStorageService objectStorageService,
|
||||
RedissonClient redissonClient) {
|
||||
super(
|
||||
redissonClient,
|
||||
"skillhub:scan:requests",
|
||||
"skillhub-scanners",
|
||||
securityScanner,
|
||||
|
|
@ -334,6 +414,19 @@ class ScanTaskConsumerTest {
|
|||
}
|
||||
}
|
||||
|
||||
private static RLock availableProcessingLock() {
|
||||
RLock processingLock = mock(RLock.class);
|
||||
when(processingLock.tryLock()).thenReturn(true);
|
||||
when(processingLock.isHeldByCurrentThread()).thenReturn(true);
|
||||
return processingLock;
|
||||
}
|
||||
|
||||
private static RedissonClient redissonClient(RLock processingLock) {
|
||||
RedissonClient redissonClient = mock(RedissonClient.class);
|
||||
when(redissonClient.getLock(org.mockito.ArgumentMatchers.anyString())).thenReturn(processingLock);
|
||||
return redissonClient;
|
||||
}
|
||||
|
||||
private static final class StubSecurityScanner implements SecurityScanner {
|
||||
private SecurityScanRequest lastRequest;
|
||||
private SecurityScanResponse response;
|
||||
|
|
|
|||
|
|
@ -3,8 +3,12 @@ package com.iflytek.skillhub.task;
|
|||
import com.iflytek.skillhub.domain.security.ScanTask;
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskOutbox;
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository;
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus;
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskProducer;
|
||||
import com.iflytek.skillhub.domain.security.ScannerType;
|
||||
import com.iflytek.skillhub.domain.skill.SkillVersion;
|
||||
import com.iflytek.skillhub.domain.skill.SkillVersionRepository;
|
||||
import com.iflytek.skillhub.domain.skill.SkillVersionStatus;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.Mock;
|
||||
|
|
@ -15,35 +19,122 @@ import java.time.Duration;
|
|||
import java.time.Instant;
|
||||
import java.time.ZoneOffset;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.BDDMockito.given;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ScanTaskOutboxDispatcherTest {
|
||||
@Mock ScanTaskOutboxRepository repository;
|
||||
@Mock ScanTaskProducer producer;
|
||||
@Mock SkillVersionRepository versionRepository;
|
||||
|
||||
@Test
|
||||
void failedRedisPublishLeavesTaskPendingForRetry() {
|
||||
Clock clock = Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC);
|
||||
ScanTaskOutbox outbox = new ScanTaskOutbox(
|
||||
new ScanTask("task-1", 1L, "/tmp/1", null, "user", 1L,
|
||||
java.util.Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue())));
|
||||
given(repository.findPendingDue(any(), any(Integer.class))).willReturn(List.of(outbox));
|
||||
given(repository.findExpiredLeases(any(), any(Integer.class))).willReturn(List.of());
|
||||
ScanTaskOutbox outbox = outbox("task-1", 1L);
|
||||
given(repository.findDispatchable(any(), any(Integer.class))).willReturn(List.of(outbox));
|
||||
doThrow(new IllegalStateException("redis unavailable")).when(producer).publishScanTask(any());
|
||||
ScanTaskOutboxDispatcher dispatcher = new ScanTaskOutboxDispatcher(
|
||||
repository, producer, clock, 50, Duration.ofMinutes(2), Duration.ofMinutes(5));
|
||||
|
||||
dispatcher.dispatch();
|
||||
dispatcher(10).dispatch();
|
||||
|
||||
assertThat(outbox.getStatus()).isEqualTo(com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus.PENDING);
|
||||
assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.PENDING);
|
||||
assertThat(outbox.getRetryCount()).isEqualTo(1);
|
||||
verify(producer).publishScanTask(any());
|
||||
verify(repository).saveAndFlush(outbox);
|
||||
verify(repository).save(outbox);
|
||||
}
|
||||
}
|
||||
|
||||
@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())));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,6 +10,8 @@ import jakarta.persistence.Id;
|
|||
import jakarta.persistence.PrePersist;
|
||||
import jakarta.persistence.Table;
|
||||
import jakarta.persistence.Version;
|
||||
import org.hibernate.annotations.JdbcTypeCode;
|
||||
import org.hibernate.type.SqlTypes;
|
||||
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
|
|
@ -31,6 +33,9 @@ public class ScanTaskOutbox {
|
|||
private String bundleKey;
|
||||
@Column(name = "publisher_id", length = 255)
|
||||
private String publisherId;
|
||||
@JdbcTypeCode(SqlTypes.JSON)
|
||||
@Column(name = "metadata", nullable = false, columnDefinition = "jsonb")
|
||||
private Map<String, String> metadata;
|
||||
@Enumerated(EnumType.STRING) @Column(nullable = false, length = 20)
|
||||
private ScanTaskOutboxStatus status;
|
||||
@Column(name = "retry_count", nullable = false)
|
||||
|
|
@ -56,22 +61,25 @@ public class ScanTaskOutbox {
|
|||
this.skillPath = task.skillPath();
|
||||
this.bundleKey = task.bundleKey();
|
||||
this.publisherId = task.publisherId();
|
||||
this.metadata = task.metadata() == null ? Map.of() : Map.copyOf(task.metadata());
|
||||
this.status = ScanTaskOutboxStatus.PENDING;
|
||||
this.nextAttemptAt = Instant.now(Clock.systemUTC());
|
||||
Instant taskCreatedAt = Instant.ofEpochMilli(task.createdAtMillis());
|
||||
this.nextAttemptAt = taskCreatedAt;
|
||||
this.createdAt = taskCreatedAt;
|
||||
this.updatedAt = taskCreatedAt;
|
||||
}
|
||||
|
||||
@PrePersist
|
||||
protected void onCreate() {
|
||||
Instant now = Instant.now(Clock.systemUTC());
|
||||
createdAt = now;
|
||||
updatedAt = now;
|
||||
if (createdAt == null) createdAt = now;
|
||||
if (updatedAt == null) updatedAt = now;
|
||||
if (nextAttemptAt == null) nextAttemptAt = now;
|
||||
}
|
||||
|
||||
public ScanTask toScanTask() {
|
||||
return new ScanTask(taskId, versionId, skillPath, bundleKey, publisherId,
|
||||
createdAt == null ? System.currentTimeMillis() : createdAt.toEpochMilli(),
|
||||
Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()));
|
||||
createdAt.toEpochMilli(), metadata == null ? Map.of() : Map.copyOf(metadata));
|
||||
}
|
||||
|
||||
public boolean claim(Instant now, Duration lease) {
|
||||
|
|
@ -90,12 +98,20 @@ public class ScanTaskOutbox {
|
|||
updatedAt = now;
|
||||
}
|
||||
|
||||
public void markFailed(Instant now, String error) {
|
||||
retryCount++;
|
||||
status = ScanTaskOutboxStatus.FAILED;
|
||||
leaseUntil = null;
|
||||
lastError = truncateError(error);
|
||||
updatedAt = now;
|
||||
}
|
||||
|
||||
public void markRetry(Instant now, Duration delay, String error) {
|
||||
retryCount++;
|
||||
status = ScanTaskOutboxStatus.PENDING;
|
||||
nextAttemptAt = now.plus(delay);
|
||||
leaseUntil = null;
|
||||
lastError = error == null ? null : error.substring(0, Math.min(error.length(), 2000));
|
||||
lastError = truncateError(error);
|
||||
updatedAt = now;
|
||||
}
|
||||
|
||||
|
|
@ -106,5 +122,10 @@ public class ScanTaskOutbox {
|
|||
public int getRetryCount() { return retryCount; }
|
||||
public Instant getNextAttemptAt() { return nextAttemptAt; }
|
||||
public Instant getLeaseUntil() { return leaseUntil; }
|
||||
|
||||
private String truncateError(String error) {
|
||||
return error == null ? null : error.substring(0, Math.min(error.length(), 2000));
|
||||
}
|
||||
|
||||
public Instant getCreatedAt() { return createdAt; }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,9 +5,7 @@ import java.util.List;
|
|||
|
||||
public interface ScanTaskOutboxRepository {
|
||||
ScanTaskOutbox save(ScanTaskOutbox outbox);
|
||||
ScanTaskOutbox saveAndFlush(ScanTaskOutbox outbox);
|
||||
List<ScanTaskOutbox> findPendingDue(Instant now, int limit);
|
||||
List<ScanTaskOutbox> findExpiredLeases(Instant now, int limit);
|
||||
List<ScanTaskOutbox> findDispatchable(Instant now, int limit);
|
||||
int deleteSentBefore(Instant cutoff);
|
||||
int deleteByVersionId(Long versionId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -207,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);
|
||||
|
|
|
|||
|
|
@ -12,7 +12,9 @@ class ScanTaskOutboxTest {
|
|||
@Test
|
||||
void claimAndMarkSentProducesStableTaskPayload() {
|
||||
ScanTask task = new ScanTask("task-1", 7L, "/tmp/7", null, "u1", 123L,
|
||||
Map.of("scannerType", ScannerType.SKILL_SCANNER.getValue()));
|
||||
Map.of(
|
||||
"scannerType", ScannerType.SKILL_SCANNER.getValue(),
|
||||
"futureAttribute", "preserved"));
|
||||
ScanTaskOutbox outbox = new ScanTaskOutbox(task);
|
||||
Instant now = Instant.parse("2026-01-01T00:00:00Z");
|
||||
|
||||
|
|
@ -21,8 +23,22 @@ class ScanTaskOutboxTest {
|
|||
outbox.markSent(now.plusSeconds(1));
|
||||
|
||||
assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.SENT);
|
||||
assertThat(outbox.toScanTask().taskId()).isEqualTo("task-1");
|
||||
assertThat(outbox.toScanTask().versionId()).isEqualTo(7L);
|
||||
assertThat(outbox.toScanTask()).isEqualTo(task);
|
||||
}
|
||||
|
||||
@Test
|
||||
void exhaustedPublishAttemptsMoveTaskToFailed() {
|
||||
ScanTaskOutbox outbox = new ScanTaskOutbox(
|
||||
new ScanTask("task-failed", 9L, null, "bundle.zip", null, 1L, Map.of()));
|
||||
Instant now = Instant.parse("2026-01-01T00:00:00Z");
|
||||
outbox.claim(now, Duration.ofMinutes(2));
|
||||
|
||||
outbox.markFailed(now, "permanent failure");
|
||||
|
||||
assertThat(outbox.getStatus()).isEqualTo(ScanTaskOutboxStatus.FAILED);
|
||||
assertThat(outbox.getRetryCount()).isEqualTo(1);
|
||||
assertThat(outbox.getLeaseUntil()).isNull();
|
||||
assertThat(outbox.claim(now.plusSeconds(1), Duration.ofMinutes(2))).isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -37,4 +53,4 @@ class ScanTaskOutboxTest {
|
|||
assertThat(outbox.getRetryCount()).isEqualTo(1);
|
||||
assertThat(outbox.getNextAttemptAt()).isEqualTo(now.plusSeconds(5));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -44,4 +44,16 @@ class SecurityScanOutboxTest {
|
|||
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());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,8 +2,6 @@ package com.iflytek.skillhub.infra.jpa;
|
|||
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskOutbox;
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskOutboxRepository;
|
||||
import com.iflytek.skillhub.domain.security.ScanTaskOutboxStatus;
|
||||
import org.springframework.data.domain.PageRequest;
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
import org.springframework.data.jpa.repository.Modifying;
|
||||
import org.springframework.data.jpa.repository.Query;
|
||||
|
|
@ -16,16 +14,15 @@ import java.util.List;
|
|||
@Repository
|
||||
public interface ScanTaskOutboxJpaRepository extends JpaRepository<ScanTaskOutbox, Long>, ScanTaskOutboxRepository {
|
||||
@Override
|
||||
default List<ScanTaskOutbox> findPendingDue(Instant now, int limit) {
|
||||
return findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc(
|
||||
ScanTaskOutboxStatus.PENDING, now, PageRequest.of(0, limit));
|
||||
}
|
||||
|
||||
@Override
|
||||
default List<ScanTaskOutbox> 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<ScanTaskOutbox> findDispatchable(@Param("now") Instant now, @Param("limit") int limit);
|
||||
|
||||
@Override
|
||||
@Modifying
|
||||
|
|
@ -34,10 +31,4 @@ public interface ScanTaskOutboxJpaRepository extends JpaRepository<ScanTaskOutbo
|
|||
|
||||
@Override
|
||||
int deleteByVersionId(Long versionId);
|
||||
|
||||
List<ScanTaskOutbox> findByStatusAndNextAttemptAtLessThanEqualOrderByCreatedAtAsc(
|
||||
ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable);
|
||||
|
||||
List<ScanTaskOutbox> findByStatusAndLeaseUntilBeforeOrderByCreatedAtAsc(
|
||||
ScanTaskOutboxStatus status, Instant now, org.springframework.data.domain.Pageable pageable);
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue