scriptoriumadmin/skillhub
1
0
Fork 0
mirror of https://github.com/iflytek/skillhub.git synced 2026-09-06 08:15:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(scanner): expire unavailable pending tasks

Browse source 
Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>
This commit is contained in:
XiaoSeS 2026-09-03 17:45:56 +08:00
parent 2e697d1581
commit 4bfb5e2692
6 changed files with 235 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
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,9 @@ 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
- 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
);
}

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

@ -247,7 +247,7 @@ 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);
@ -293,6 +293,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;
}

107
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,23 @@ 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 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 +162,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,11 +237,14 @@ 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())
@ -315,6 +345,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 +401,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 +415,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 +458,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}

117
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;
@ -361,6 +363,7 @@ class ScanTaskConsumerTest {
"taskId", "task-timeout",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", String.valueOf(System.currentTimeMillis()),
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
@ -388,7 +391,9 @@ class ScanTaskConsumerTest {
new StubSecurityScanService(),
repository,
new InMemoryScanTaskProducer(),
new InMemoryObjectStorageService()
new InMemoryObjectStorageService(),
Clock.fixed(Instant.parse("2026-09-03T08:00:00Z"), ZoneOffset.UTC),
Duration.ofHours(1)
);
StreamMessageId messageId = new StreamMessageId(13, 0);
@ -397,7 +402,7 @@ class ScanTaskConsumerTest {
"taskId", "task-expired-timeout",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"createdAtMillis", "1",
"createdAtMillis", String.valueOf(Instant.parse("2026-09-03T06:59:59Z").toEpochMilli()),
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
@ -407,6 +412,65 @@ class ScanTaskConsumerTest {
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);
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(
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(version.getStatus()).isEqualTo(SkillVersionStatus.SCAN_FAILED);
verify(consumer.stream).remove(messageId);
}
@Test
void processBusiness_whenScannerFails_releasesProcessingLock() {
StubSecurityScanner securityScanner = new StubSecurityScanner();
@ -429,6 +493,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);
@ -473,6 +555,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,