scriptoriumadmin/skillhub
1
0
Fork 0
mirror of https://github.com/iflytek/skillhub.git synced 2026-10-06 02:48:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #809 from iflytek/codex/fix/issue-808-scan-pending-bound

Browse source 
fix(scanner): bound unavailable task recovery
This commit is contained in:
XiaoSeS 2026-09-03 20:10:23 +08:00 • committed by GitHub
commit 4128801c68
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 483 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
docs/security-scanning.md
View file

@ -49,6 +49,7 @@ skillhub:
key: skillhub:scan:requests
group: skillhub-scanners
reclaim-min-idle: PT16M
max-unavailable-age: PT1H
```
Important environment variables:
@ -60,6 +61,7 @@ Important environment variables:
- `SKILLHUB_SCAN_STREAM_KEY`
- `SKILLHUB_SCAN_STREAM_GROUP`
- `SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE`
- `SKILLHUB_SECURITY_STREAM_MAX_UNAVAILABLE_AGE`
Scanner-side optional environment variables:
@ -134,7 +136,11 @@ Response fields include:
## Failure Semantics
- scan task retries are handled by `AbstractStreamConsumer`
- final failure marks the version as `SCAN_FAILED`
- scanner connection failures, HTTP 429, and HTTP 5xx remain pending for automatic recovery
- unavailable tasks older than `max-unavailable-age` are marked `SCAN_FAILED`, acknowledged, and removed from the Redis Stream
- the timeout is evaluated during pending reclaim; terminal handling can occur roughly one `reclaim-min-idle` plus one `reclaim-interval` after the configured age
- terminal failures retain a failure reason in the security audit response for operators and authorized users
- other final failures mark the version as `SCAN_FAILED` after retry exhaustion
- even after scan failure, a review task is still created so the package does not get stuck forever
This keeps the existing human review path intact while making scanner failures visible.

9
server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedisStreamConfig.java
View file

@ -8,12 +8,13 @@ import com.iflytek.skillhub.observability.MessageObservationSupport;
import com.iflytek.skillhub.storage.ObjectStorageService;
import com.iflytek.skillhub.stream.RedissonScanTaskProducer;
import com.iflytek.skillhub.stream.ScanTaskConsumer;
import java.time.Clock;
import java.time.Duration;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.Duration;
@Configuration
@ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true")
@ -40,6 +41,9 @@ public class RedisStreamConfig {
@Value("${skillhub.security.scanner.retry-max-attempts:3}")
private int maxRetryAttempts;
@Value("${skillhub.security.stream.max-unavailable-age:PT1H}")
private Duration maxUnavailableAge;
@Bean
public RedissonScanTaskProducer redisScanTaskProducer(
RedissonClient redissonClient,
@ -55,6 +59,7 @@ public class RedisStreamConfig {
SkillVersionRepository skillVersionRepository,
ScanTaskProducer scanTaskProducer,
ObjectStorageService objectStorageService,
Clock clock,
MessageObservationSupport messageObservationSupport) {
return new ScanTaskConsumer(
redissonClient,
@ -70,6 +75,8 @@ public class RedisStreamConfig {
reclaimBatchSize,
reclaimInterval,
maxRetryAttempts,
maxUnavailableAge,
clock,
messageObservationSupport
);
}

1
server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/portal/SecurityAuditController.java
View file

@ -121,6 +121,7 @@ public class SecurityAuditController extends BaseApiController {
audit.getFindingsCount(),
deserializeFindings(audit.getFindings()),
audit.getScanDurationSeconds(),
audit.getFailureReason(),
audit.getScannedAt(),
audit.getCreatedAt()
);

1
server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/SecurityAuditResponse.java
View file

@ -16,6 +16,7 @@ public record SecurityAuditResponse(
Integer findingsCount,
List<SecurityFinding> findings,
Double scanDurationSeconds,
String failureReason,
Instant scannedAt,
Instant createdAt
) {

14
server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/AbstractStreamConsumer.java
View file

@ -247,15 +247,17 @@ public abstract class AbstractStreamConsumer<T> {
}
private void handleFailure(T payload, int retryCount, Exception e) {
if (retryCount < maxRetryCount()) {
if (shouldRetry(payload, e, retryCount)) {
// Retry publication remains inside the current consumer scope, so the new producer
// span and message carrier continue the original trace.
retryMessage(payload, retryCount + 1);
return;
}
markFailed(payload, truncateError(
taskDisplayName() + " failed (retried " + retryCount + " times): " + e.getMessage()
));
markFailed(payload, truncateError(finalFailureReason(payload, e, retryCount)));
}
protected String finalFailureReason(T payload, Exception error, int retryCount) {
return taskDisplayName() + " failed (retried " + retryCount + " times): " + error.getMessage();
}
protected int parseRetryCount(Map<String, String> data) {
@ -293,6 +295,10 @@ public abstract class AbstractStreamConsumer<T> {
return DEFAULT_MAX_RETRY_COUNT;
}
protected boolean shouldRetry(T payload, Exception error, int retryCount) {
return retryCount < maxRetryCount();
}
protected boolean shouldDeferFailure(T payload, Exception error) {
return false;
}

127
server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java
View file

@ -21,12 +21,18 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.time.Clock;
import java.time.DateTimeException;
import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.Map;
import java.util.Objects;
public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.ScanTaskPayload> {
private static final Path SCAN_TEMP_DIR = Paths.get("/tmp/skillhub-scans").toAbsolutePath().normalize();
private static final Duration DEFAULT_MAX_UNAVAILABLE_AGE = Duration.ofHours(1);
private static final Duration MAX_CLOCK_SKEW = Duration.ofMinutes(5);
private final RedissonClient redissonClient;
private final SecurityScanner securityScanner;
@ -35,6 +41,8 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
private final ScanTaskProducer scanTaskProducer;
private final ObjectStorageService objectStorageService;
private final int maxRetryAttempts;
private final Duration maxUnavailableAge;
private final Clock clock;
public ScanTaskConsumer(RedissonClient redissonClient,
String streamKey,
@ -53,6 +61,8 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
this.scanTaskProducer = scanTaskProducer;
this.objectStorageService = objectStorageService;
this.maxRetryAttempts = 3;
this.maxUnavailableAge = DEFAULT_MAX_UNAVAILABLE_AGE;
this.clock = Clock.systemUTC();
}
public ScanTaskConsumer(RedissonClient redissonClient,
@ -68,6 +78,8 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
int reclaimBatchSize,
Duration reclaimInterval,
int maxRetryAttempts,
Duration maxUnavailableAge,
Clock clock,
MessageObservationSupport messageObservationSupport) {
super(
redissonClient,
@ -86,6 +98,11 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
this.scanTaskProducer = scanTaskProducer;
this.objectStorageService = objectStorageService;
this.maxRetryAttempts = maxRetryAttempts;
if (maxUnavailableAge == null || maxUnavailableAge.isZero() || maxUnavailableAge.isNegative()) {
throw new IllegalArgumentException("maxUnavailableAge must be positive");
}
this.maxUnavailableAge = maxUnavailableAge;
this.clock = Objects.requireNonNull(clock, "clock");
}
@Override
@ -101,14 +118,34 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
@Override
protected boolean shouldDeferFailure(ScanTaskPayload payload, Exception error) {
return error instanceof ConcurrentScanInProgressException
|| (error instanceof SecurityScanException scanError && scanError.isScannerUnavailable());
|| (isScannerUnavailable(error) && !hasUnavailableRecoveryExpired(payload));
}
@Override
protected boolean shouldRetry(ScanTaskPayload payload, Exception error, int retryCount) {
if (isScannerUnavailable(error) && hasUnavailableRecoveryExpired(payload)) {
return false;
}
return super.shouldRetry(payload, error, retryCount);
}
@Override
protected String finalFailureReason(ScanTaskPayload payload, Exception error, int retryCount) {
log.error("Security scan failed after retries: taskId={}, versionId={}, scanner={}, retryCount={}",
payload.taskId(), payload.versionId(), payload.scannerType(), retryCount, error);
if (isScannerUnavailable(error) && hasUnavailableRecoveryExpired(payload)) {
return "Security scanner did not recover before the configured timeout. "
+ "Retry after scanner availability is restored.";
}
return "Security scan failed after automatic retries. Retry the scan or contact an administrator.";
}
@Override
protected void markDeferred(ScanTaskPayload payload, Exception error) {
cleanupRetryTempPath(payload);
log.warn("Scanner unavailable; keeping task pending for later recovery: taskId={}, versionId={}, reason={}",
payload.taskId(), payload.versionId(), error.getMessage());
log.warn("Scanner unavailable; keeping task pending for later recovery: taskId={}, versionId={}, "
+ "taskAge={}, maxUnavailableAge={}, reason={}",
payload.taskId(), payload.versionId(), taskAge(payload), maxUnavailableAge, error.getMessage());
}
@Override
@ -136,7 +173,8 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
blankToNull(data.get("skillPath")),
blankToNull(data.get("bundleKey")),
scannerType,
parseRetryCount(data)
parseRetryCount(data),
parseCreatedAtMillis(messageId, data.get("createdAtMillis"))
);
} catch (NumberFormatException e) {
return null;
@ -210,24 +248,24 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
@Override
protected void markFailed(ScanTaskPayload payload, String error) {
log.error("Security scan task failed permanently: taskId={}, versionId={}, scanner={}, source={}, error={}",
log.error("Security scan task failed permanently: taskId={}, versionId={}, scanner={}, source={}, "
+ "taskAge={}, maxUnavailableAge={}, error={}",
payload.taskId(),
payload.versionId(),
payload.scannerType(),
payload.sourceDescription(),
taskAge(payload),
maxUnavailableAge,
error);
try {
skillVersionRepository.findById(payload.versionId())
.filter(version -> version.getStatus() == SkillVersionStatus.SCANNING)
.ifPresent(version -> {
version.setStatus(SkillVersionStatus.SCAN_FAILED);
skillVersionRepository.save(version);
});
securityScanService.processScanFailure(
payload.taskId(), payload.versionId(), payload.scannerType(), error);
} finally {
cleanupTempPath(payload.cleanupPath());
}
}
@Override
protected void retryMessage(ScanTaskPayload payload, int retryCount) {
log.warn("Retrying security scan task: taskId={}, versionId={}, scanner={}, nextRetryCount={}, source={}",
@ -315,6 +353,55 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
return value == null || value.isBlank() ? null : value;
}
private boolean isScannerUnavailable(Exception error) {
return error instanceof SecurityScanException scanError && scanError.isScannerUnavailable();
}
private boolean hasUnavailableRecoveryExpired(ScanTaskPayload payload) {
Instant now = clock.instant();
long createdAtMillis = payload.createdAtMillis();
if (createdAtMillis <= 0 || createdAtMillis > now.plus(MAX_CLOCK_SKEW).toEpochMilli()) {
return true;
}
try {
return !Instant.ofEpochMilli(createdAtMillis).plus(maxUnavailableAge).isAfter(now);
} catch (DateTimeException | ArithmeticException ignored) {
return true;
}
}
private Duration taskAge(ScanTaskPayload payload) {
try {
Duration age = Duration.between(Instant.ofEpochMilli(payload.createdAtMillis()), clock.instant());
return age.isNegative() ? Duration.ZERO : age;
} catch (DateTimeException | ArithmeticException ignored) {
return maxUnavailableAge;
}
}
private long parseCreatedAtMillis(String messageId, String value) {
Long createdAt = parsePositiveLong(value);
if (createdAt != null) {
return createdAt;
}
int separator = messageId.indexOf('-');
String redisTimestamp = separator >= 0 ? messageId.substring(0, separator) : messageId;
Long fallback = parsePositiveLong(redisTimestamp);
return fallback != null ? fallback : 0L;
}
private Long parsePositiveLong(String value) {
if (value == null || value.isBlank()) {
return null;
}
try {
long parsed = Long.parseLong(value);
return parsed > 0 ? parsed : null;
} catch (NumberFormatException ignored) {
return null;
}
}
protected static final class ScanTaskPayload {
private final String taskId;
private final Long versionId;
@ -322,11 +409,12 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
private final String bundleKey;
private final ScannerType scannerType;
private final int retryCount;
private final long createdAtMillis;
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);
this(taskId, versionId, skillPath, bundleKey, scannerType, 0, System.currentTimeMillis());
}
protected ScanTaskPayload(String taskId,
@ -335,12 +423,23 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
String bundleKey,
ScannerType scannerType,
int retryCount) {
this(taskId, versionId, skillPath, bundleKey, scannerType, retryCount, System.currentTimeMillis());
}
protected ScanTaskPayload(String taskId,
Long versionId,
String skillPath,
String bundleKey,
ScannerType scannerType,
int retryCount,
long createdAtMillis) {
this.taskId = taskId;
this.versionId = versionId;
this.skillPath = skillPath;
this.bundleKey = bundleKey;
this.scannerType = scannerType;
this.retryCount = retryCount;
this.createdAtMillis = createdAtMillis;
}
protected String taskId() {
@ -367,6 +466,10 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
return retryCount;
}
protected long createdAtMillis() {
return createdAtMillis;
}
protected void markWorkingSkillPath(String workingSkillPath) {
this.workingSkillPath = workingSkillPath;
}

2
server/skillhub-app/src/main/resources/application.yml
View file

@ -228,6 +228,8 @@ skillhub:
custom-policy-path: ${SKILLHUB_SCANNER_CUSTOM_POLICY_PATH:}
fail-on-severity: ${SKILLHUB_SCANNER_FAIL_ON_SEVERITY:high}
stream:
# Keep temporary scanner outages recoverable, but do not retain Redis Pending entries forever.
max-unavailable-age: ${SKILLHUB_SECURITY_STREAM_MAX_UNAVAILABLE_AGE:PT1H}
key: ${SKILLHUB_SCAN_STREAM_KEY:skillhub:scan:requests}
group: ${SKILLHUB_SCAN_STREAM_GROUP:skillhub-scanners}
reclaim-enabled: ${SKILLHUB_SCAN_STREAM_RECLAIM_ENABLED:true}

2
server/skillhub-app/src/main/resources/db/migration/V48__security_audit_failure_reason.sql Normal file
View file

@ -0,0 +1,2 @@
ALTER TABLE security_audit
ADD COLUMN failure_reason VARCHAR(1000);

7
server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerLoggingTest.java
View file

@ -222,6 +222,13 @@ class ScanTaskConsumerLoggingTest {
@Override
public void processScanResult(Long versionId, ScannerType scannerType, SecurityScanResponse response) {
}
@Override
public void processScanFailure(String taskId,
Long versionId,
ScannerType scannerType,
String reason) {
}
}
private static final class TestProducer implements ScanTaskProducer {

223
server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerTest.java
View file

@ -33,8 +33,10 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@ -96,17 +98,13 @@ class ScanTaskConsumerTest {
}
@Test
void markFailed_setsScanFailedWithoutChangingReviewTaskAndCleansTempFile() throws Exception {
SkillVersion version = new SkillVersion(8L, "1.0.0", "publisher-1");
setField(version, "id", 42L);
version.setStatus(SkillVersionStatus.SCANNING);
InMemorySkillVersionRepository skillVersionRepository = new InMemorySkillVersionRepository(version);
void markFailed_recordsExactAttemptAndCleansTempFile() throws Exception {
StubSecurityScanService securityScanService = new StubSecurityScanService();
InMemoryReviewTaskRepository reviewTaskRepository = new InMemoryReviewTaskRepository();
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
new StubSecurityScanner(),
new StubSecurityScanService(),
skillVersionRepository,
securityScanService,
new InMemorySkillVersionRepository(),
new InMemoryScanTaskProducer(),
new InMemoryObjectStorageService()
);
@ -122,7 +120,8 @@ class ScanTaskConsumerTest {
consumer.invokeMarkFailed(payload, "scan failed");
assertThat(skillVersionRepository.savedVersion.getStatus()).isEqualTo(SkillVersionStatus.SCAN_FAILED);
assertThat(securityScanService.failedTaskId).isEqualTo("task-2");
assertThat(securityScanService.failedVersionId).isEqualTo(42L);
assertThat(reviewTaskRepository.savedTask).isNull();
assertThat(reviewTaskRepository.deletedTask).isNull();
assertThat(Files.exists(tempFile)).isFalse();
@ -217,9 +216,10 @@ class ScanTaskConsumerTest {
securityScanner.failure = new IllegalStateException("scanner unavailable");
InMemoryScanTaskProducer producer = new InMemoryScanTaskProducer();
InMemorySkillVersionRepository repository = new InMemorySkillVersionRepository();
StubSecurityScanService scanService = new StubSecurityScanService();
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
securityScanner,
new StubSecurityScanService(),
scanService,
repository,
producer,
objectStorageService
@ -361,6 +361,7 @@ class ScanTaskConsumerTest {
"taskId", "task-timeout",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", String.valueOf(System.currentTimeMillis()),
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
@ -370,6 +371,140 @@ class ScanTaskConsumerTest {
verify(consumer.stream, never()).ack("skillhub-scanners", messageId);
}
@Test
void handleMessage_whenScannerRemainsUnavailablePastRecoveryWindow_failsAndRemovesDelivery() {
StubSecurityScanner securityScanner = new StubSecurityScanner();
securityScanner.failure = new SecurityScanException(
"scanner timed out", new HttpClientException("request timed out", new java.util.concurrent.TimeoutException()));
SkillVersion version = new SkillVersion(8L, "1.0.0", "publisher-1");
try {
setField(version, "id", 42L);
} catch (Exception e) {
throw new AssertionError(e);
}
version.setStatus(SkillVersionStatus.SCANNING);
InMemorySkillVersionRepository repository = new InMemorySkillVersionRepository(version);
StubSecurityScanService scanService = new StubSecurityScanService();
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
securityScanner,
scanService,
repository,
new InMemoryScanTaskProducer(),
new InMemoryObjectStorageService(),
Clock.fixed(Instant.parse("2026-09-03T08:00:00Z"), ZoneOffset.UTC),
Duration.ofHours(1)
);
StreamMessageId messageId = new StreamMessageId(13, 0);
when(consumer.stream.ack("skillhub-scanners", messageId)).thenReturn(1L);
consumer.handleMessage(messageId, Map.of(
"taskId", "task-expired-timeout",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", String.valueOf(Instant.parse("2026-09-03T06:59:59Z").toEpochMilli()),
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
assertThat(scanService.failedTaskId).isEqualTo("task-expired-timeout");
assertThat(scanService.failedReason).contains("Retry after scanner availability is restored");
verify(consumer.stream).ack("skillhub-scanners", messageId);
verify(consumer.stream).remove(messageId);
}
@Test
void handleMessage_whenScannerUnavailableBeforeRecoveryDeadline_keepsDeliveryPending() {
StubSecurityScanner securityScanner = unavailableScanner();
SkillVersion version = scanningVersion(42L);
InMemorySkillVersionRepository repository = new InMemorySkillVersionRepository(version);
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
securityScanner,
new StubSecurityScanService(),
repository,
new InMemoryScanTaskProducer(),
new InMemoryObjectStorageService(),
Clock.fixed(Instant.parse("2026-09-03T08:00:00Z"), ZoneOffset.UTC),
Duration.ofHours(1)
);
StreamMessageId messageId = new StreamMessageId(14, 0);
consumer.handleMessage(messageId, Map.of(
"taskId", "task-before-deadline",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", String.valueOf(Instant.parse("2026-09-03T07:00:01Z").toEpochMilli()),
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.SCANNING);
assertThat(repository.savedVersion).isNull();
verify(consumer.stream, never()).ack("skillhub-scanners", messageId);
}
@Test
void handleMessage_whenTaskTimestampIsMalformed_usesRedisEntryTimeForExpiry() {
StubSecurityScanner securityScanner = unavailableScanner();
SkillVersion version = scanningVersion(42L);
InMemorySkillVersionRepository repository = new InMemorySkillVersionRepository(version);
StubSecurityScanService scanService = new StubSecurityScanService();
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
securityScanner,
scanService,
repository,
new InMemoryScanTaskProducer(),
new InMemoryObjectStorageService(),
Clock.fixed(Instant.parse("2026-09-03T08:00:00Z"), ZoneOffset.UTC),
Duration.ofHours(1)
);
StreamMessageId messageId = new StreamMessageId(
Instant.parse("2026-09-03T06:00:00Z").toEpochMilli(), 0);
when(consumer.stream.ack("skillhub-scanners", messageId)).thenReturn(1L);
consumer.handleMessage(messageId, Map.of(
"taskId", "task-malformed-timestamp",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", "not-a-number",
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
assertThat(scanService.failedTaskId).isEqualTo("task-malformed-timestamp");
verify(consumer.stream).remove(messageId);
}
@Test
void handleMessage_whenFailureWasRecordedButAckFails_redeliveryOnlyCompletesAck() {
StubSecurityScanner securityScanner = unavailableScanner();
StubSecurityScanService scanService = new StubSecurityScanService();
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
securityScanner,
scanService,
new InMemorySkillVersionRepository(scanningVersion(42L)),
new InMemoryScanTaskProducer(),
new InMemoryObjectStorageService(),
Clock.fixed(Instant.parse("2026-09-03T08:00:00Z"), ZoneOffset.UTC),
Duration.ofHours(1)
);
StreamMessageId messageId = new StreamMessageId(15, 0);
Map<String, String> task = Map.of(
"taskId", "task-ack-recovery",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", String.valueOf(Instant.parse("2026-09-03T06:00:00Z").toEpochMilli()),
"scannerType", ScannerType.SKILL_SCANNER.getValue()
);
when(consumer.stream.ack("skillhub-scanners", messageId))
.thenThrow(new IllegalStateException("redis unavailable"))
.thenReturn(1L);
assertThatThrownBy(() -> consumer.handleMessage(messageId, task))
.isInstanceOf(IllegalStateException.class)
.hasMessage("redis unavailable");
consumer.handleMessage(messageId, task);
assertThat(securityScanner.invocations).isEqualTo(1);
verify(consumer.stream).remove(messageId);
}
@Test
void processBusiness_whenScannerFails_releasesProcessingLock() {
StubSecurityScanner securityScanner = new StubSecurityScanner();
@ -392,6 +527,24 @@ class ScanTaskConsumerTest {
verify(processingLock).unlock();
}
private StubSecurityScanner unavailableScanner() {
StubSecurityScanner scanner = new StubSecurityScanner();
scanner.failure = new SecurityScanException(
"scanner timed out", new HttpClientException("request timed out", new java.util.concurrent.TimeoutException()));
return scanner;
}
private SkillVersion scanningVersion(Long id) {
SkillVersion version = new SkillVersion(8L, "1.0.0", "publisher-1");
try {
setField(version, "id", id);
} catch (Exception e) {
throw new AssertionError(e);
}
version.setStatus(SkillVersionStatus.SCANNING);
return version;
}
private void setField(Object target, String fieldName, Object value) throws Exception {
Field field = target.getClass().getDeclaredField(fieldName);
field.setAccessible(true);
@ -436,6 +589,35 @@ class ScanTaskConsumerTest {
this.stream = mock(RStream.class);
}
@SuppressWarnings("unchecked")
private TestableScanTaskConsumer(SecurityScanner securityScanner,
SecurityScanService securityScanService,
SkillVersionRepository skillVersionRepository,
ScanTaskProducer scanTaskProducer,
ObjectStorageService objectStorageService,
Clock clock,
Duration maxUnavailableAge) {
super(
redissonClient(availableProcessingLock()),
"skillhub:scan:requests",
"skillhub-scanners",
securityScanner,
securityScanService,
skillVersionRepository,
scanTaskProducer,
objectStorageService,
true,
Duration.ofMinutes(16),
20,
Duration.ofSeconds(30),
3,
maxUnavailableAge,
clock,
new MessageObservationSupport(ObservationRegistry.NOOP, new RequestIdAccessor())
);
this.stream = mock(RStream.class);
}
@SuppressWarnings("unchecked")
private TestableScanTaskConsumer(SecurityScanner securityScanner,
SecurityScanService securityScanService,
@ -496,9 +678,11 @@ class ScanTaskConsumerTest {
private SecurityScanRequest lastRequest;
private SecurityScanResponse response;
private RuntimeException failure;
private int invocations;
@Override
public SecurityScanResponse scan(SecurityScanRequest request) {
invocations++;
this.lastRequest = request;
if (failure != null) {
throw failure;
@ -521,6 +705,10 @@ class ScanTaskConsumerTest {
private Long lastVersionId;
private ScannerType lastScannerType;
private SecurityScanResponse lastResponse;
private String failedTaskId;
private Long failedVersionId;
private String failedReason;
private boolean processed;
private StubSecurityScanService() {
super(null, null, task -> {
@ -533,6 +721,19 @@ class ScanTaskConsumerTest {
this.lastScannerType = scannerType;
this.lastResponse = response;
}
@Override
public void processScanFailure(String taskId, Long versionId, ScannerType scannerType, String reason) {
this.failedTaskId = taskId;
this.failedVersionId = versionId;
this.failedReason = reason;
this.processed = true;
}
@Override
public boolean isTaskAlreadyProcessed(String taskId) {
return processed && taskId.equals(failedTaskId);
}
}
private static final class InMemorySkillVersionRepository implements SkillVersionRepository {

19
server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/security/SecurityAudit.java
View file

@ -56,6 +56,9 @@ public class SecurityAudit {
@Column(name = "scan_duration_seconds")
private Double scanDurationSeconds;
@Column(name = "failure_reason", length = 1000)
private String failureReason;
@Column(name = "scanned_at")
private Instant scannedAt;
@ -131,6 +134,10 @@ public class SecurityAudit {
return scanDurationSeconds;
}
public String getFailureReason() {
return failureReason;
}
public Instant getScannedAt() {
return scannedAt;
}
@ -171,6 +178,18 @@ public class SecurityAudit {
this.scannedAt = scannedAt;
}
public void markFailed(Instant failedAt, String reason) {
this.failureReason = truncate(reason, 1000);
this.scannedAt = failedAt;
}
private String truncate(String value, int maxLength) {
if (value == null || value.length() <= maxLength) {
return value;
}
return value.substring(0, maxLength);
}
public Instant getDeletedAt() {
return deletedAt;
}

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

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

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

@ -115,6 +115,33 @@ public class SecurityScanService {
&& auditRepository.existsByTaskIdAndScannedAtIsNotNull(taskId);
}
@Transactional
public void processScanFailure(String taskId, Long versionId, ScannerType scannerType, String reason) {
SecurityAudit audit = auditRepository.findByTaskId(taskId)
.filter(candidate -> candidate.getSkillVersionId().equals(versionId))
.filter(candidate -> candidate.getScannerType() == scannerType)
.orElseThrow(() -> new IllegalStateException("SecurityAudit not found for taskId=" + taskId));
if (audit.getScannedAt() != null) {
return;
}
audit.markFailed(Instant.now(Clock.systemUTC()), reason);
auditRepository.save(audit);
boolean currentAttempt = auditRepository
.findLatestActiveByVersionIdAndScannerType(versionId, scannerType)
.map(latest -> taskId.equals(latest.getTaskId()))
.orElse(false);
if (!currentAttempt) {
return;
}
skillVersionRepository.findById(versionId)
.filter(version -> version.getStatus() == SkillVersionStatus.SCANNING)
.ifPresent(version -> {
version.setStatus(SkillVersionStatus.SCAN_FAILED);
skillVersionRepository.save(version);
});
}
@Transactional
public void processScanResult(Long versionId, ScannerType scannerType, SecurityScanResponse response) {
SecurityAudit audit = auditRepository.findLatestActiveByVersionIdAndScannerType(versionId, scannerType)

39
server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/security/SecurityScanServiceTest.java
View file

@ -292,6 +292,45 @@ class SecurityScanServiceTest {
assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.PUBLISHED);
}
@Test
void processScanFailure_marksExactCurrentAttemptAndVersionFailed() throws Exception {
SecurityAudit audit = new SecurityAudit(42L, ScannerType.SKILL_SCANNER, "task-current");
SkillVersion version = new SkillVersion(8L, "1.0.0", "publisher-1");
setId(version, 42L);
version.setStatus(SkillVersionStatus.SCANNING);
given(auditRepository.findByTaskId("task-current")).willReturn(Optional.of(audit));
given(auditRepository.findLatestActiveByVersionIdAndScannerType(42L, ScannerType.SKILL_SCANNER))
.willReturn(Optional.of(audit));
given(skillVersionRepository.findById(42L)).willReturn(Optional.of(version));
service.processScanFailure("task-current", 42L, ScannerType.SKILL_SCANNER, "scanner unavailable");
assertThat(audit.getScannedAt()).isNotNull();
assertThat(audit.getFailureReason()).isEqualTo("scanner unavailable");
assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.SCAN_FAILED);
verify(auditRepository).save(audit);
verify(skillVersionRepository).save(version);
}
@Test
void processScanFailure_forStaleAttemptDoesNotFailCurrentVersion() throws Exception {
SecurityAudit stale = new SecurityAudit(42L, ScannerType.SKILL_SCANNER, "task-stale");
SecurityAudit current = new SecurityAudit(42L, ScannerType.SKILL_SCANNER, "task-current");
SkillVersion version = new SkillVersion(8L, "1.0.0", "publisher-1");
setId(version, 42L);
version.setStatus(SkillVersionStatus.SCANNING);
given(auditRepository.findByTaskId("task-stale")).willReturn(Optional.of(stale));
given(auditRepository.findLatestActiveByVersionIdAndScannerType(42L, ScannerType.SKILL_SCANNER))
.willReturn(Optional.of(current));
service.processScanFailure("task-stale", 42L, ScannerType.SKILL_SCANNER, "stale failure");
assertThat(stale.getScannedAt()).isNotNull();
assertThat(stale.getFailureReason()).isEqualTo("stale failure");
assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.SCANNING);
verify(skillVersionRepository, never()).save(any());
}
@Test
void processScanResult_shouldNotChangeStatusWhenVersionAlreadyPublished() {
SecurityAudit audit = new SecurityAudit(42L, ScannerType.SKILL_SCANNER);

3
server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/SecurityAuditJpaRepository.java
View file

@ -20,6 +20,9 @@ public interface SecurityAuditJpaRepository extends JpaRepository<SecurityAudit,
@Override
Optional<SecurityAudit> findByScanId(String scanId);
@Override
Optional<SecurityAudit> findByTaskId(String taskId);
@Override
boolean existsBySkillVersionId(Long skillVersionId);

1
web/src/api/generated/schema.d.ts vendored
View file

@ -5309,6 +5309,7 @@ export interface components {
findings?: components["schemas"]["SecurityFinding"][];
/** Format: double */
scanDurationSeconds?: number;
failureReason?: string;
/** Format: date-time */
scannedAt?: string;
/** Format: date-time */

8
web/src/features/security-audit/display-state.ts
View file

@ -1,14 +1,14 @@
import type { SecurityAuditDisplayState, SecurityAuditRecord } from './types'
export function getSecurityAuditDisplayState(
audit: Pick<SecurityAuditRecord, 'scannedAt' | 'verdict'>,
audit: Pick<SecurityAuditRecord, 'scannedAt' | 'verdict' | 'failureReason'>,
versionStatus?: string
): SecurityAuditDisplayState {
if (audit.failureReason || versionStatus === 'SCAN_FAILED') {
return 'SCAN_FAILED'
}
if (audit.scannedAt) {
return audit.verdict
}
if (versionStatus === 'SCAN_FAILED') {
return 'SCAN_FAILED'
}
return 'SCANNING'
}

15
web/src/features/security-audit/security-audit-section.test.tsx
View file

@ -8,8 +8,11 @@ vi.mock('react-i18next', async () => {
return {
...actual,
useTranslation: () => ({
t: (key: string, values?: Record<string, unknown>) =>
values?.count !== undefined ? `${key}:${values.count}` : key,
t: (key: string, values?: Record<string, unknown>) => {
if (values?.count !== undefined) return `${key}:${values.count}`
if (values?.reason !== undefined) return `${key}:${values.reason}`
return key
},
i18n: { language: 'en' },
}),
}
@ -26,6 +29,7 @@ function createAudit(overrides: Partial<SecurityAuditRecord> = {}): SecurityAudi
findingsCount: 0,
findings: [],
scanDurationSeconds: null,
failureReason: null,
scannedAt: '2026-03-20T10:00:00Z',
createdAt: '2026-03-20T10:00:00Z',
...overrides,
@ -109,13 +113,18 @@ describe('SecurityAuditSection', () => {
})
it('renders scan failed status when version scan failed before audit completion', () => {
mockAudits = [createAudit({ verdict: 'SUSPICIOUS', scannedAt: null })]
mockAudits = [createAudit({
verdict: 'SUSPICIOUS',
scannedAt: '2026-03-20T10:00:00Z',
failureReason: 'Scanner remained unavailable',
})]
mockIsLoading = false
const html = renderToStaticMarkup(<SecurityAuditSection skillId={1} versionId={10} versionStatus="SCAN_FAILED" />)
expect(html).toContain('securityAudit.statusScanFailed')
expect(html).not.toContain('securityAudit.statusScanning')
expect(html).toContain('Scanner remained unavailable')
})
it('renders the findings count', () => {

6
web/src/features/security-audit/security-audit-section.tsx
View file

@ -86,6 +86,12 @@ function ScannerCard({ audit, versionStatus }: { audit: SecurityAuditRecord; ver
</div>
</div>
{audit.failureReason && (
<p className="text-sm text-destructive">
{t('securityAudit.failureReason', { reason: audit.failureReason })}
</p>
)}
{sortedFindings.length > 0 && (
<>
<Button

1
web/src/features/security-audit/security-audit-summary.test.tsx
View file

@ -26,6 +26,7 @@ function createAudit(overrides: Partial<SecurityAuditRecord> = {}): SecurityAudi
findingsCount: 0,
findings: [],
scanDurationSeconds: null,
failureReason: null,
scannedAt: '2026-03-20T10:00:00Z',
createdAt: '2026-03-20T10:00:00Z',
...overrides,

1
web/src/features/security-audit/types.ts
View file

@ -26,6 +26,7 @@ export interface SecurityAuditRecord {
findingsCount: number
findings: SecurityFinding[]
scanDurationSeconds: number | null
failureReason: string | null
scannedAt: string | null
createdAt: string
}

1
web/src/i18n/locales/en.json
View file

@ -1588,6 +1588,7 @@
"scanDuration": "{{seconds}}s",
"statusScanning": "Scanning",
"statusScanFailed": "Scan Failed",
"failureReason": "Reason: {{reason}}",
"remediation": "Remediation",
"viewDetails": "View Details",
"verdict": {

1
web/src/i18n/locales/ru.json
View file

@ -1619,6 +1619,7 @@
"scanDuration": "{{seconds}} с",
"statusScanning": "Сканирование",
"statusScanFailed": "Сканирование не удалось",
"failureReason": "Причина: {{reason}}",
"remediation": "Рекомендации",
"viewDetails": "Подробности",
"verdict": {

1
web/src/i18n/locales/zh.json
View file

@ -1587,6 +1587,7 @@
"scanDuration": "{{seconds}}s",
"statusScanning": "扫描中",
"statusScanFailed": "扫描失败",
"failureReason": "失败原因：{{reason}}",
"remediation": "修复建议",
"viewDetails": "查看详情",
"verdict": {