diff --git a/docs/27-authoring-drafts.md b/docs/27-authoring-drafts.md new file mode 100644 index 00000000..b2219a99 --- /dev/null +++ b/docs/27-authoring-drafts.md @@ -0,0 +1,29 @@ +# 草稿与结构校验 API + +这是创作平台的第一层：创建私有草稿、维护文件和运行结构校验。尚不包含网页工作台、模型执行、MCP 探测、自动修复确认或发布集成。草稿校验通过不等于技能已发布，也不代表运行效果或安全扫描通过。 + +## 权限与文件 + +接口同时支持 `/api/v1/authoring` 与 `/api/web/authoring`；复用平台认证及浏览器 CSRF 规则。草稿与运行的读取、编辑和事件订阅仅允许所有者或超级管理员。创建草稿还要求目标空间可写，并满足已有空间成员权限。 + +`POST /drafts` 创建草稿并生成 `SKILL.md` 模板。文件操作在 `/drafts/{draftId}/files`，读取内容在 `/drafts/{draftId}/files/content?path=...`。保存请求可带 `expectedRevision`；版本不匹配返回冲突。同一草稿的写入串行化，文件仍受技能包路径、扩展名、单文件大小、文件数量和总大小限制。 + +文件保存和读取的 `encoding` 为 `text` 或 `base64`。合法 UTF-8 内容返回 `text`，非 UTF-8 字节返回 `base64`；调用方按返回的 encoding 解码，保存时使用相同 encoding，不能将图片等二进制内容当普通字符串处理。 + +删除草稿先提交数据库事务，再清理私有对象；数据库回滚不会删除仍被引用的文件。提交后存储清理失败会留下不可达的私有对象，并记录清理失败，不假装数据库删除失败。 + +## 运行与事件 + +`POST /drafts/{draftId}/runs` 为当前草稿 revision 创建运行。同一草稿最多有一个活动运行。执行时核验 revision 并捕获稳定的文件内容；排队期间草稿被修改则该运行失败，不能把新版内容冒充旧版的检测结果。已捕获内容不随之后的编辑改变；只有成功且草稿 revision 仍匹配的运行才更新草稿校验标记。 + +运行状态、事件和 findings 在 `/runs/{runId}` 及其子接口读取。SSE 在 `/runs/{runId}/events/stream`，可用 `afterSeq` 或 `Last-Event-ID` 恢复，按连续序号发送回放和实时事件。只有真正终态才能记录 `RUN_FINISHED`，重复完成不会产生第二条终态事件。连接或缓冲超过限制时关闭／拒绝连接，客户端可凭最后确认的事件编号重连，或通过轮询获取持久化记录。 + +执行器为有界线程池；队列满时已创建的排队运行会受控失败，不由 HTTP 请求线程代为执行，也不留下永久排队记录。取消或超时先使运行落入终态，阻止后续事件／finding 写入，再协作中断本机任务。中断不保证硬停止不响应中断的计算或外部 I/O，不能视为进程级沙箱。 + +## 配置与迁移 + +部署属性为 `skillhub.authoring.executor-threads`、`run-timeout-ms`、`stale-run-minutes` 与 `workspace-root`。默认执行线程 4、执行期限 900,000 毫秒、遗留活动运行清理阈值 30 分钟；期限从任务实际执行开始计算，遗留清理是恢复机制。修改启动配置需要重启应用。维护任务不会因数据库暂时失败就删除工作目录。 + +数据库迁移为未发布的 `V69__authoring_platform.sql`。正 revision、非负大小／计数及状态转换在业务代码中验证；数据库保留主键、外键和唯一性结构约束。头像／邀请 PR #947 使用 V70–V74，须在 V69 合入后验证其升级顺序，避免把较低编号的迁移留到已应用较高版本之后。 + +API 的完整字段、操作与响应以服务生成的 OpenAPI 为准；本文件不替代接口定义。 diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringExecutorConfig.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringExecutorConfig.java index 79771dfa..d0ff8bf5 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringExecutorConfig.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringExecutorConfig.java @@ -1,6 +1,8 @@ package com.iflytek.skillhub.config; import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -14,6 +16,17 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @Configuration public class AuthoringExecutorConfig { + @Bean(name = "authoringValidationTimer", destroyMethod = "shutdownNow") + public ScheduledExecutorService authoringValidationTimer() { + ScheduledThreadPoolExecutor timer = new ScheduledThreadPoolExecutor(1, runnable -> { + Thread thread = new Thread(runnable, "authoring-validation-timeout"); + thread.setDaemon(true); + return thread; + }); + timer.setRemoveOnCancelPolicy(true); + return timer; + } + @Bean(name = "authoringValidationExecutor") public Executor authoringValidationExecutor(AuthoringProperties properties) { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); @@ -21,8 +34,7 @@ public class AuthoringExecutorConfig { executor.setMaxPoolSize(Math.max(1, properties.getExecutorThreads())); executor.setQueueCapacity(100); executor.setThreadNamePrefix("authoring-validation-"); - // a full queue runs the task on the submitting thread instead of dropping runs - executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); executor.initialize(); return executor; } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringProperties.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringProperties.java index 5d0c07a1..2379f079 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringProperties.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/config/AuthoringProperties.java @@ -17,7 +17,7 @@ public class AuthoringProperties { /** Concurrent validation runs executed in-process. */ private int executorThreads = 4; - /** Hard wall-clock cap for one whole validation run. */ + /** Deadline after execution starts; interrupts cooperatively and blocks late writes. */ private long runTimeoutMs = 900_000; /** Active runs older than this are swept to TIMED_OUT by the maintenance task. */ diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/SkillAuthoringController.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/SkillAuthoringController.java index 741a5ca0..ac0a3bf5 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/SkillAuthoringController.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/SkillAuthoringController.java @@ -139,12 +139,7 @@ public class SkillAuthoringController extends BaseApiController { Set platformRoles) { SkillDraftService.FileContent content = draftService.readFile(draftId, userId, path, platformRoles); - return ok("response.success", new DraftFileContentResponse( - content.file().getFilePath(), - content.file().getSha256(), - content.file().getSize(), - content.file().getContentType(), - content.asText())); + return ok("response.success", DraftFileContentResponse.from(content)); } @Operation(operationId = "deleteDraftFile", summary = "Delete one draft file") diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/ValidationStreamController.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/ValidationStreamController.java index 92e6ead2..b8e3c55c 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/ValidationStreamController.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/controller/authoring/ValidationStreamController.java @@ -1,12 +1,10 @@ package com.iflytek.skillhub.controller.authoring; import com.iflytek.skillhub.domain.authoring.service.ValidationRunService; -import com.iflytek.skillhub.domain.authoring.validation.ValidationEvent; import com.iflytek.skillhub.domain.authoring.validation.ValidationRun; import com.iflytek.skillhub.service.authoring.ValidationEventBroadcaster; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; -import java.util.List; import java.util.Set; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.GetMapping; @@ -49,8 +47,8 @@ public class ValidationStreamController { Set platformRoles) { ValidationRun run = runService.getOwnedRun(runId, userId, platformRoles); int lastSeq = resolveCursor(afterSeq, lastEventId); - List replay = runService.listEvents(run.getId(), lastSeq); - return broadcaster.subscribe(run.getId(), lastSeq, replay); + return broadcaster.subscribe(run.getId(), lastSeq, + () -> runService.listEvents(run.getId(), lastSeq)); } private int resolveCursor(Integer afterSeq, String lastEventId) { diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/DraftFileContentResponse.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/DraftFileContentResponse.java index a1d283c0..3cb219ec 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/DraftFileContentResponse.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/DraftFileContentResponse.java @@ -1,11 +1,36 @@ package com.iflytek.skillhub.dto; -/** Draft file content, always delivered as text (draft files are text formats). */ +import com.iflytek.skillhub.domain.authoring.service.SkillDraftService; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; +import java.util.Base64; + +/** UTF-8 text stays editable; other file bytes use reversible base64 transport. */ public record DraftFileContentResponse( String path, String sha256, Long size, String contentType, - String content + String content, + String encoding ) { + public static DraftFileContentResponse from(SkillDraftService.FileContent fileContent) { + String content; + String encoding; + try { + content = StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .decode(ByteBuffer.wrap(fileContent.content())).toString(); + encoding = "text"; + } catch (CharacterCodingException exception) { + content = Base64.getEncoder().encodeToString(fileContent.content()); + encoding = "base64"; + } + var file = fileContent.file(); + return new DraftFileContentResponse(file.getFilePath(), file.getSha256(), file.getSize(), + file.getContentType(), content, encoding); + } } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/SaveDraftFileRequest.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/SaveDraftFileRequest.java index 88c82ab0..00bc0db9 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/SaveDraftFileRequest.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/dto/SaveDraftFileRequest.java @@ -2,6 +2,7 @@ package com.iflytek.skillhub.dto; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Pattern; /** * Request body for saving one draft file. Text is the default encoding; base64 @@ -10,7 +11,7 @@ import jakarta.validation.constraints.NotNull; public record SaveDraftFileRequest( @NotBlank(message = "{error.badRequest}") String path, @NotNull(message = "{error.badRequest}") String content, - String encoding, + @Pattern(regexp = "(?i)(text|base64)", message = "{error.badRequest}") String encoding, String contentType, Integer expectedRevision ) { diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcaster.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcaster.java index 0dc5eaeb..e5240c00 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcaster.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcaster.java @@ -8,6 +8,9 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.TreeMap; +import java.util.function.Supplier; +import com.iflytek.skillhub.domain.shared.exception.DomainConflictException; import org.springframework.http.MediaType; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -42,32 +45,39 @@ public class ValidationEventBroadcaster { * @param lastSeq the subscriber's last seen sequence number (Last-Event-ID) * @param replay persisted events after {@code lastSeq}, oldest first */ - public SseEmitter subscribe(Long runId, int lastSeq, List replay) { + public SseEmitter subscribe(Long runId, int lastSeq, Supplier> replay) { EmitterSession session = new EmitterSession(lastSeq); - sessionsByRun.computeIfAbsent(runId, key -> new CopyOnWriteArrayList<>()).add(session); + synchronized (sessionsByRun) { + int total = sessionsByRun.values().stream().mapToInt(List::size).sum(); + List existing = sessionsByRun.get(runId); + if (total >= 256 || (existing != null && existing.size() >= 16)) { + throw new DomainConflictException("error.authoring.stream.tooManySubscribers"); + } + sessionsByRun.computeIfAbsent(runId, key -> new CopyOnWriteArrayList<>()).add(session); + } Runnable detach = () -> detach(runId, session); session.emitter().onCompletion(detach); session.emitter().onTimeout(detach); session.emitter().onError(error -> detach.run()); - - boolean terminal = false; - for (ValidationEvent event : replay) { - if (!session.send(event.getSeq(), envelope(event))) { - detach(runId, session); - return session.emitter(); + try { + // Registration precedes the DB read; live events buffer until replay ends. + for (ValidationEvent event : replay.get()) { + if (!session.accept(event, envelope(event))) { + detach.run(); + return session.emitter(); + } } - if (event.getEventType() == ValidationEventType.RUN_FINISHED) { - terminal = true; + if (!session.finishReplay()) { + detach.run(); } - } - if (terminal) { - // run already finished: close the stream so clients stop reconnecting + } catch (RuntimeException exception) { + detach.run(); session.emitter().complete(); + throw exception; } return session.emitter(); } - /** Pushes one persisted event to every live subscriber of the run. */ public void publish(Long runId, ValidationEvent event) { List sessions = sessionsByRun.get(runId); if (sessions == null) { @@ -75,28 +85,27 @@ public class ValidationEventBroadcaster { } String envelope = envelope(event); for (EmitterSession session : sessions) { - if (!session.send(event.getSeq(), envelope)) { + if (!session.accept(event, envelope)) { detach(runId, session); } } } - /** Sends the terminal event and completes all subscribers of a finished run. */ public void completeRun(Long runId, ValidationEvent terminalEvent) { + // The terminal may race replay or a lower live sequence. Only close once + // every preceding event has actually been delivered. publish(runId, terminalEvent); - List sessions = sessionsByRun.remove(runId); - if (sessions == null) { - return; - } - for (EmitterSession session : sessions) { - session.emitter().complete(); - } } private void detach(Long runId, EmitterSession session) { - List sessions = sessionsByRun.get(runId); - if (sessions != null) { - sessions.remove(session); + synchronized (sessionsByRun) { + List sessions = sessionsByRun.get(runId); + if (sessions != null) { + sessions.remove(session); + if (sessions.isEmpty()) { + sessionsByRun.remove(runId); + } + } } } @@ -117,37 +126,60 @@ public class ValidationEventBroadcaster { /** One subscriber connection with its own delivery cursor and serialized writes. */ static final class EmitterSession { - - private final SseEmitter emitter; + private final SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MS); + private final TreeMap pending = new TreeMap<>(); private int lastSeq; + private boolean replaying = true; + private boolean closed; EmitterSession(int lastSeq) { - this.emitter = new SseEmitter(EMITTER_TIMEOUT_MS); this.lastSeq = lastSeq; } - SseEmitter emitter() { - return emitter; - } + SseEmitter emitter() { return emitter; } + synchronized int lastSeq() { return lastSeq; } - int lastSeq() { - return lastSeq; - } - - /** Sends unless the event predates this session's cursor; false means the pipe broke. */ - synchronized boolean send(int seq, String envelope) { - if (seq <= lastSeq) { - return true; - } - try { - emitter.send(SseEmitter.event() - .id(String.valueOf(seq)) - .data(envelope, MediaType.APPLICATION_JSON)); - lastSeq = seq; - return true; - } catch (IOException | IllegalStateException exception) { + synchronized boolean accept(ValidationEvent event, String envelope) { + if (closed) { return false; } + if (event.getSeq() <= lastSeq) { return true; } + pending.putIfAbsent(event.getSeq(), new PendingEvent(envelope, + event.getEventType() == ValidationEventType.RUN_FINISHED)); + if (pending.size() > 1024) { + closed = true; + emitter.complete(); return false; } + return replaying || drain(); } + + synchronized boolean finishReplay() { + replaying = false; + return drain(); + } + + private boolean drain() { + while (!closed && pending.containsKey(lastSeq + 1)) { + PendingEvent event = pending.remove(lastSeq + 1); + try { + emitter.send(SseEmitter.event().id(String.valueOf(lastSeq + 1)) + .data(event.envelope(), MediaType.APPLICATION_JSON)); + lastSeq++; + if (event.terminal()) { + closed = true; + pending.clear(); + emitter.complete(); + return false; + } + } catch (IOException | IllegalStateException exception) { + closed = true; + pending.clear(); + emitter.complete(); + return false; + } + } + return !closed; + } + + private record PendingEvent(String envelope, boolean terminal) { } } } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestrator.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestrator.java index fe20a821..2a0c36ab 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestrator.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestrator.java @@ -1,6 +1,6 @@ package com.iflytek.skillhub.service.authoring; -import com.iflytek.skillhub.domain.authoring.SkillDraft; +import com.iflytek.skillhub.config.AuthoringProperties; import com.iflytek.skillhub.domain.authoring.service.DraftStructureValidator; import com.iflytek.skillhub.domain.authoring.service.SkillDraftService; import com.iflytek.skillhub.domain.authoring.service.ValidationRunService; @@ -13,14 +13,17 @@ import com.iflytek.skillhub.domain.authoring.validation.ValidationLayer; import com.iflytek.skillhub.domain.authoring.validation.ValidationRun; import com.iflytek.skillhub.domain.authoring.validation.ValidationRunStatus; import com.iflytek.skillhub.domain.shared.exception.DomainConflictException; -import com.iflytek.skillhub.domain.skill.validation.PackageEntry; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.FutureTask; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.RejectedExecutionException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; @@ -45,38 +48,60 @@ public class ValidationRunOrchestrator { private final ValidationEventBroadcaster broadcaster; private final Executor executor; - private final Map cancelFlags = new ConcurrentHashMap<>(); - - /** Serializes the RUN_FINISHED emission so it happens exactly once per run. */ - private final Map terminalLocks = new ConcurrentHashMap<>(); + private final ScheduledExecutorService timer; + private final AuthoringProperties properties; + private final Map> tasks = new ConcurrentHashMap<>(); public ValidationRunOrchestrator(ValidationRunService runService, SkillDraftService draftService, DraftStructureValidator structureValidator, ValidationEventBroadcaster broadcaster, - @Qualifier("authoringValidationExecutor") Executor executor) { + @Qualifier("authoringValidationExecutor") Executor executor, + @Qualifier("authoringValidationTimer") ScheduledExecutorService timer, + AuthoringProperties properties) { this.runService = runService; this.draftService = draftService; this.structureValidator = structureValidator; this.broadcaster = broadcaster; this.executor = executor; + this.timer = timer; + this.properties = properties; } /** Enqueues one QUEUED run for execution. */ public void submit(Long runId) { - executor.execute(() -> executeRun(runId)); + FutureTask task = new FutureTask<>(() -> { executeRun(runId); return null; }) { + @Override protected void done() { tasks.remove(runId, this); } + }; + if (tasks.putIfAbsent(runId, task) != null) { + return; + } + try { + executor.execute(task); + } catch (RejectedExecutionException exception) { + tasks.remove(runId, task); + // The run was persisted before enqueue. Settle rather than leave it QUEUED. + runService.settleIfQueued(runId, 1, 0, + Map.of("reason", "validation executor is busy; start a new run")); + emitTerminalOnce(runId); + } } /** - * Requests cooperative cancellation. QUEUED runs settle immediately; running runs - * settle when their executor observes the flag. + * Settles cancellation immediately, fences further writes, then interrupts + * the local worker cooperatively. Interrupt-insensitive work may still finish computing. */ public ValidationRun requestCancel(Long runId, String userId, Set platformRoles) { ValidationRun run = runService.requestCancel(runId, userId, platformRoles); - cancelFlags.computeIfAbsent(runId, key -> new AtomicBoolean()).set(true); - if (run.getStatus().isTerminal()) { + if (run.isActive()) { + settle(runId, ValidationRunStatus.CANCELLED, run.getErrorCount(), run.getWarningCount(), + Map.of("reason", "cancelled by user")); + run = runService.getRun(runId); + } else { emitTerminalOnce(runId); } + FutureTask task = tasks.get(runId); + if (task != null) { task.cancel(true); } return run; } @@ -88,63 +113,77 @@ public class ValidationRunOrchestrator { Map summary) { runService.settleIfActive(runId, status, errorCount, warningCount, summary); emitTerminalOnce(runId); + if (status == ValidationRunStatus.TIMED_OUT) { + FutureTask task = tasks.get(runId); + if (task != null) { task.cancel(true); } + } } // ---------------------------------------------------------------- execution private void executeRun(Long runId) { + ScheduledFuture deadline = null; + boolean claimed = false; try { ValidationRun run = runService.claimForExecution(runId); if (run.getStatus().isTerminal()) { emitTerminalOnce(runId); return; } + claimed = true; + deadline = timer.schedule(() -> { + try { + settle(runId, ValidationRunStatus.TIMED_OUT, 0, 0, + Map.of("reason", "validation execution deadline exceeded")); + } catch (RuntimeException exception) { + log.error("Could not settle validation deadline for run {}", runId, exception); + } + }, Math.max(1, properties.getRunTimeoutMs()), TimeUnit.MILLISECONDS); executePipeline(run); } catch (DomainConflictException exception) { - // cancelled while queued, or already claimed/settled elsewhere - log.info("Validation run {} no longer claimable: {}", runId, exception.getMessage()); - emitTerminalOnce(runId); + if (claimed) { + settle(runId, ValidationRunStatus.FAILED, 1, 0, + Map.of("reason", "draft revision changed or run already settled")); + } else { + // Duplicate submit cannot append a fake terminal to another active worker. + emitTerminalOnce(runId); + } } catch (Exception exception) { log.error("Validation run {} crashed", runId, exception); settle(runId, ValidationRunStatus.FAILED, 1, 0, Map.of("internalError", truncate(String.valueOf(exception.getMessage()), 500))); } finally { - cancelFlags.remove(runId); - terminalLocks.remove(runId); + if (deadline != null) { deadline.cancel(false); } } } private void executePipeline(ValidationRun run) { Long runId = run.getId(); - SkillDraft draft = draftService.getDraft(run.getDraftId()); - List entries = draftService.materializeEntries(draft.getId()); + SkillDraftService.MaterializedSnapshot snapshot = + draftService.materializeSnapshot(run.getDraftId(), run.getDraftRevision()); RunContext context = new RunContext(runId); emit(runId, ValidationEventType.RUN_STARTED, null, Map.of( - "draftId", draft.getId(), - "draftName", draft.getName(), + "draftId", run.getDraftId(), + "draftName", snapshot.name(), "draftRevision", run.getDraftRevision())); try { // ---- STRUCTURE layer emit(runId, ValidationEventType.PHASE_STARTED, "STRUCTURE", Map.of()); DraftStructureValidator.StructureReport structure = - structureValidator.validate(draft.getName(), draft.getRequirement(), entries); + structureValidator.validate(snapshot.name(), snapshot.requirement(), snapshot.entries()); structure.findings().forEach(context::recordFinding); emit(runId, ValidationEventType.PHASE_FINISHED, "STRUCTURE", Map.of( "findings", structure.findings().size(), "errors", context.errorCount(ValidationLayer.STRUCTURE), "warnings", context.warningCount(ValidationLayer.STRUCTURE))); - if (cancelFlags.getOrDefault(runId, new AtomicBoolean(false)).get()) { - settle(runId, ValidationRunStatus.CANCELLED, context.errorCount, context.warningCount, - summary(context)); - return; - } - settle(runId, context.errorCount == 0 ? ValidationRunStatus.SUCCEEDED : ValidationRunStatus.FAILED, context.errorCount, context.warningCount, summary(context)); + } catch (DomainConflictException exception) { + throw exception; } catch (Exception exception) { log.error("Validation pipeline for run {} failed", runId, exception); settle(runId, ValidationRunStatus.FAILED, context.errorCount + 1, context.warningCount, @@ -161,20 +200,8 @@ public class ValidationRunOrchestrator { /** Appends the RUN_FINISHED event exactly once per run, then completes SSE subscribers. */ private void emitTerminalOnce(Long runId) { - Object lock = terminalLocks.computeIfAbsent(runId, key -> new Object()); - synchronized (lock) { - for (ValidationEvent event : runService.listEvents(runId, null)) { - if (event.getEventType() == ValidationEventType.RUN_FINISHED) { - broadcaster.completeRun(runId, event); - return; - } - } - ValidationRun run = runService.getRun(runId); - ValidationEvent terminal = runService.appendEvent(runId, - ValidationEventType.RUN_FINISHED, null, Map.of( - "status", run.getStatus().name(), - "errorCount", run.getErrorCount(), - "warningCount", run.getWarningCount())); + ValidationEvent terminal = runService.appendTerminalEvent(runId); + if (terminal != null) { broadcaster.completeRun(runId, terminal); } } diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTask.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTask.java index f7c687c8..2ebfc8aa 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTask.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTask.java @@ -1,6 +1,7 @@ package com.iflytek.skillhub.task; import com.iflytek.skillhub.config.AuthoringProperties; +import com.iflytek.skillhub.domain.shared.exception.DomainNotFoundException; import com.iflytek.skillhub.domain.authoring.validation.ValidationRun; import com.iflytek.skillhub.domain.authoring.validation.ValidationRunStatus; import com.iflytek.skillhub.domain.authoring.service.ValidationRunService; @@ -95,9 +96,10 @@ public class ValidationRunMaintenanceTask { } }); } - } catch (Exception exception) { - // unknown run id (workspace of a deleted draft) — remove the directory too + } catch (DomainNotFoundException exception) { deleteQuietly(directory); + } catch (IOException | RuntimeException exception) { + log.warn("Cannot check validation workspace {}; retaining it", directory, exception); } } diff --git a/server/skillhub-app/src/main/resources/db/migration/V69__authoring_platform.sql b/server/skillhub-app/src/main/resources/db/migration/V69__authoring_platform.sql index d986818a..63943958 100644 --- a/server/skillhub-app/src/main/resources/db/migration/V69__authoring_platform.sql +++ b/server/skillhub-app/src/main/resources/db/migration/V69__authoring_platform.sql @@ -18,8 +18,7 @@ CREATE TABLE skill_draft ( submitted_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - version BIGINT NOT NULL DEFAULT 0, - CONSTRAINT chk_skill_draft_revision CHECK (revision > 0) + version BIGINT NOT NULL DEFAULT 0 ); -- Name uniqueness is enforced case-insensitively per owner. @@ -44,8 +43,7 @@ CREATE TABLE draft_file ( storage_key VARCHAR(768) NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - CONSTRAINT uq_draft_file_path UNIQUE (draft_id, file_path), - CONSTRAINT chk_draft_file_size CHECK (size >= 0) + CONSTRAINT uq_draft_file_path UNIQUE (draft_id, file_path) ); CREATE INDEX idx_draft_file_draft @@ -82,9 +80,7 @@ CREATE TABLE validation_run ( started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - version BIGINT NOT NULL DEFAULT 0, - CONSTRAINT chk_validation_run_error_count CHECK (error_count >= 0), - CONSTRAINT chk_validation_run_warning_count CHECK (warning_count >= 0) + version BIGINT NOT NULL DEFAULT 0 ); CREATE INDEX idx_validation_run_draft_created @@ -121,9 +117,7 @@ CREATE TABLE validation_finding ( suggestion JSONB, status VARCHAR(16) NOT NULL DEFAULT 'OPEN', applied_revision INT, - created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - CONSTRAINT chk_validation_finding_status - CHECK (status IN ('OPEN', 'APPLIED', 'DISMISSED')) + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_validation_finding_run_status diff --git a/server/skillhub-app/src/main/resources/messages.properties b/server/skillhub-app/src/main/resources/messages.properties index ef5c8661..55aaec57 100644 --- a/server/skillhub-app/src/main/resources/messages.properties +++ b/server/skillhub-app/src/main/resources/messages.properties @@ -319,3 +319,5 @@ error.authoring.finding.notFound=Validation finding not found: {0} error.authoring.finding.noSuggestion=Finding {0} has no applicable fix suggestion error.authoring.finding.applied=The fix for finding {0} was already applied error.authoring.finding.suggestionUnreadable=The fix suggestion of finding {0} could not be read + +error.authoring.stream.tooManySubscribers=Too many validation event subscribers. Close an existing connection and retry. diff --git a/server/skillhub-app/src/main/resources/messages_ru.properties b/server/skillhub-app/src/main/resources/messages_ru.properties index c919ca8e..48fcfdb6 100644 --- a/server/skillhub-app/src/main/resources/messages_ru.properties +++ b/server/skillhub-app/src/main/resources/messages_ru.properties @@ -262,3 +262,5 @@ error.authoring.finding.notFound=Результат проверки не най error.authoring.finding.noSuggestion=Для результата {0} нет применимой рекомендации error.authoring.finding.applied=Исправление результата {0} уже применено error.authoring.finding.suggestionUnreadable=Не удалось прочитать рекомендацию для результата {0} + +error.authoring.stream.tooManySubscribers=Достигнут предел подключений к событиям проверки. Закройте существующее подключение и повторите попытку. diff --git a/server/skillhub-app/src/main/resources/messages_zh.properties b/server/skillhub-app/src/main/resources/messages_zh.properties index 8b6ac4a3..a32c45fe 100644 --- a/server/skillhub-app/src/main/resources/messages_zh.properties +++ b/server/skillhub-app/src/main/resources/messages_zh.properties @@ -321,3 +321,5 @@ error.authoring.finding.notFound=验证问题项不存在：{0} error.authoring.finding.noSuggestion=问题项 {0} 没有可应用的修复建议 error.authoring.finding.applied=问题项 {0} 的修复已被应用 error.authoring.finding.suggestionUnreadable=问题项 {0} 的修复建议无法解析 + +error.authoring.stream.tooManySubscribers=校验事件连接数量已达上限，请关闭已有连接后重试。 diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/AuthoringDraftFlowIntegrationTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/AuthoringDraftFlowIntegrationTest.java index 51bad052..22cfeb3a 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/AuthoringDraftFlowIntegrationTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/AuthoringDraftFlowIntegrationTest.java @@ -7,6 +7,12 @@ import com.iflytek.skillhub.domain.authoring.service.ValidationRunService; import com.iflytek.skillhub.domain.authoring.validation.ValidationFinding; import com.iflytek.skillhub.domain.authoring.validation.ValidationRun; import com.iflytek.skillhub.domain.authoring.validation.ValidationRunStatus; +import com.iflytek.skillhub.domain.authoring.validation.ValidationEvent; +import com.iflytek.skillhub.domain.authoring.validation.ValidationEventType; +import com.iflytek.skillhub.domain.authoring.validation.FindingDraft; +import com.iflytek.skillhub.domain.authoring.validation.FindingSeverity; +import com.iflytek.skillhub.domain.authoring.validation.ValidationLayer; +import com.iflytek.skillhub.domain.shared.exception.DomainConflictException; import com.iflytek.skillhub.domain.namespace.Namespace; import com.iflytek.skillhub.domain.namespace.NamespaceMember; import com.iflytek.skillhub.domain.namespace.NamespaceMemberRepository; @@ -23,6 +29,14 @@ import java.time.Duration; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.ArrayList; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; @@ -73,6 +87,8 @@ class AuthoringDraftFlowIntegrationTest { private static final String USER = "author-user-1"; private static final String NS_SLUG = "authoring-test-ns"; + @Autowired private ObjectStorageService objectStorage; + @Autowired private PlatformTransactionManager transactionManager; @Autowired private SkillDraftService draftService; @Autowired private ValidationRunService runService; @Autowired private ValidationRunOrchestrator orchestrator; @@ -92,6 +108,9 @@ class AuthoringDraftFlowIntegrationTest { static final class InMemoryObjectStorage implements ObjectStorageService { private final Map objects = new ConcurrentHashMap<>(); + private volatile String blockedKey; + private final CountDownLatch readEntered = new CountDownLatch(1); + private final CountDownLatch releaseRead = new CountDownLatch(1); @Override public void putObject(String key, InputStream data, long size, String contentType) { @@ -104,6 +123,17 @@ class AuthoringDraftFlowIntegrationTest { @Override public InputStream getObject(String key) { + if (key.equals(blockedKey)) { + readEntered.countDown(); + try { + if (!releaseRead.await(30, TimeUnit.SECONDS)) { + throw new IllegalStateException("Blocked test storage read was not released"); + } + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Test storage read interrupted", exception); + } + } byte[] content = objects.get(key); if (content == null) { throw new IllegalArgumentException("missing object: " + key); @@ -220,6 +250,159 @@ class AuthoringDraftFlowIntegrationTest { "text/markdown", 1, null)); } + @Test + void rolledBackDeletionRetainsReadableDraftFiles() { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "delete-rollback", null, null); + new TransactionTemplate(transactionManager).executeWithoutResult(transaction -> { + draftService.deleteDraft(draft.getId(), USER, null); + transaction.setRollbackOnly(); + }); + assertThat(draftService.readFile(draft.getId(), USER, "SKILL.md", null).asText()) + .contains("name: delete-rollback"); + draftService.deleteDraft(draft.getId(), USER, null); + assertThrows(com.iflytek.skillhub.domain.shared.exception.DomainNotFoundException.class, + () -> draftService.getDraft(draft.getId())); + } + + @Test + void simultaneousStartsCreateExactlyOneActiveRun() throws Exception { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "concurrent-start", null, null); + List results = concurrently(6, () -> { + try { + runService.startRun(draft.getId(), USER, null); + return true; + } catch (DomainConflictException expected) { + return false; + } + }); + assertThat(results).filteredOn(Boolean::booleanValue).hasSize(1); + assertThat(runService.listRuns(draft.getId())).hasSize(1); + } + + @Test + void simultaneousSavesRejectStaleRevisionRatherThanLosingContent() throws Exception { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "concurrent-save", null, null); + List results = concurrently(4, () -> { + try { + draftService.saveFile(draft.getId(), USER, "SKILL.md", validSkillMdBytes(), + "text/markdown", 1, null); + return true; + } catch (DomainConflictException expected) { + return false; + } + }); + assertThat(results).filteredOn(Boolean::booleanValue).hasSize(1); + assertThat(draftService.getDraft(draft.getId()).getRevision()).isEqualTo(2); + assertThat(draftService.readFile(draft.getId(), USER, "SKILL.md", null).content()) + .isEqualTo(validSkillMdBytes()); + } + + @Test + void snapshotRejectsChangedRevisionAndRetainsCapturedContent() { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "snapshot-draft", null, null); + var snapshot = draftService.materializeSnapshot(draft.getId(), 1); + draftService.saveFile(draft.getId(), USER, "SKILL.md", validSkillMdBytes(), + "text/markdown", 1, null); + assertThrows(DomainConflictException.class, () -> draftService.materializeSnapshot(draft.getId(), 1)); + draftService.deleteDraft(draft.getId(), USER, null); + assertThat(new String(snapshot.entries().get(0).content(), StandardCharsets.UTF_8)) + .contains("name: snapshot-draft"); + } + + @Test + void concurrentEventsRemainOrderedAndTerminalEventIsIdempotent() throws Exception { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "concurrent-events", null, null); + ValidationRun run = runService.startRun(draft.getId(), USER, null); + assertThat(runService.appendTerminalEvent(run.getId())).isNull(); + assertThrows(DomainConflictException.class, () -> runService.appendEvent(run.getId(), + ValidationEventType.RUN_FINISHED, null, Map.of("status", "SUCCEEDED"))); + List appended = concurrently(8, () -> runService.appendEvent(run.getId(), + ValidationEventType.PHASE_STARTED, "STRUCTURE", Map.of())); + assertThat(appended).extracting(ValidationEvent::getSeq) + .containsExactlyInAnyOrder(1, 2, 3, 4, 5, 6, 7, 8); + runService.requestCancel(run.getId(), USER, null); + List terminal = concurrently(6, () -> runService.appendTerminalEvent(run.getId())); + assertThat(terminal).extracting(ValidationEvent::getId).containsOnly(terminal.get(0).getId()); + assertThat(terminal.get(0).getPayload()).containsEntry("status", "CANCELLED"); + assertThat(runService.listEvents(run.getId(), null)).hasSize(9); + assertThrows(DomainConflictException.class, () -> runService.appendEvent(run.getId(), + ValidationEventType.PHASE_STARTED, "STRUCTURE", Map.of())); + assertThrows(DomainConflictException.class, () -> runService.recordFinding(run.getId(), + new FindingDraft(ValidationLayer.STRUCTURE, "LATE", FindingSeverity.ERROR, + "SKILL.md", null, "late finding", null))); + assertThat(runService.listFindings(run.getId())).isEmpty(); + } + + @Test + void simultaneousSettlementReturnsOneWinnerAndQueueRejectionCannotFailRunningRun() throws Exception { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "concurrent-settle", null, null); + ValidationRun run = runService.startRun(draft.getId(), USER, null); + runService.claimForExecution(run.getId()); + assertThat(runService.settleIfQueued(run.getId(), 1, 0, Map.of())).isEmpty(); + assertThat(runService.getRun(run.getId()).getStatus()).isEqualTo(ValidationRunStatus.RUNNING); + List results = concurrently(6, () -> runService.settleIfActive(run.getId(), + ValidationRunStatus.TIMED_OUT, 0, 0, Map.of()).isPresent()); + assertThat(results).filteredOn(Boolean::booleanValue).hasSize(1); + assertThat(runService.getRun(run.getId()).getStatus()).isEqualTo(ValidationRunStatus.TIMED_OUT); + } + + @Test + void timeoutSettlesBeforeBlockedSnapshotStorageReadCompletes() throws Exception { + SkillDraft draft = draftService.createDraft(NS_SLUG, USER, "blocked-snapshot-timeout", null, null); + ValidationRun run = runService.startRun(draft.getId(), USER, null); + runService.claimForExecution(run.getId()); + InMemoryObjectStorage storage = (InMemoryObjectStorage) objectStorage; + storage.blockedKey = draftService.listFiles(draft.getId()).get(0).getStorageKey(); + try (var executor = Executors.newFixedThreadPool(2)) { + try { + var snapshot = executor.submit(() -> + draftService.materializeSnapshot(draft.getId(), run.getDraftRevision())); + assertThat(storage.readEntered.await(5, TimeUnit.SECONDS)).isTrue(); + var timeout = executor.submit(() -> runService.settleIfActive(run.getId(), + ValidationRunStatus.TIMED_OUT, 0, 0, Map.of("reason", "deadline"))); + assertThat(timeout.get(5, TimeUnit.SECONDS)).isPresent(); + assertThat(runService.getRun(run.getId()).getStatus()).isEqualTo(ValidationRunStatus.TIMED_OUT); + assertThat(snapshot.isDone()).isFalse(); + assertThrows(DomainConflictException.class, () -> runService.recordFinding(run.getId(), + FindingDraft.error(ValidationLayer.STRUCTURE, "LATE", "late finding"))); + assertThat(runService.listFindings(run.getId())).isEmpty(); + storage.releaseRead.countDown(); + assertThat(snapshot.get(5, TimeUnit.SECONDS).entries()).hasSize(1); + assertThrows(DomainConflictException.class, () -> runService.appendEvent(run.getId(), + ValidationEventType.PHASE_STARTED, "STRUCTURE", Map.of())); + } finally { + storage.blockedKey = null; + storage.releaseRead.countDown(); + } + } + } + + private List concurrently(int workers, Callable action) throws Exception { + CountDownLatch ready = new CountDownLatch(workers); + CountDownLatch start = new CountDownLatch(1); + try (var executor = Executors.newFixedThreadPool(workers)) { + List> futures = new ArrayList<>(); + for (int i = 0; i < workers; i++) { + futures.add(executor.submit(() -> { + ready.countDown(); + if (!start.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("Concurrent test did not start"); + } + return action.call(); + })); + } + assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue(); + start.countDown(); + List results = new ArrayList<>(); + for (Future future : futures) { + results.add(future.get(20, TimeUnit.SECONDS)); + } + return results; + } finally { + start.countDown(); + } + } + private String validSkillMd() { return "---\nname: draft-flow\ndescription: summarizes documents\n---\n\n# draft-flow\n"; } @@ -231,8 +414,10 @@ class AuthoringDraftFlowIntegrationTest { private ValidationRun awaitTerminal(Long runId) { long deadline = System.currentTimeMillis() + 60_000; while (System.currentTimeMillis() < deadline) { - if (runService.getRun(runId).getStatus().isTerminal()) { - return runService.getRun(runId); + ValidationRun run = runService.getRun(runId); + if (run.getStatus().isTerminal() && runService.listEvents(runId, null).stream() + .anyMatch(event -> event.getEventType() == ValidationEventType.RUN_FINISHED)) { + return run; } try { Thread.sleep(200); diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/SkillAuthoringControllerAuthorizationTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/SkillAuthoringControllerAuthorizationTest.java index 9d84cc21..f15dd9b6 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/SkillAuthoringControllerAuthorizationTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/authoring/SkillAuthoringControllerAuthorizationTest.java @@ -13,6 +13,8 @@ import com.iflytek.skillhub.domain.user.UserAccount; import com.iflytek.skillhub.domain.user.UserAccountRepository; import java.util.List; import java.util.Set; +import java.nio.file.Files; +import java.nio.file.Path; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; @@ -157,4 +159,13 @@ class SkillAuthoringControllerAuthorizationTest { .andExpect(status().isForbidden()); } } + + @Test + void openApiDescribesBinarySafeDraftContent() throws Exception { + String schema = mockMvc.perform(get("/v3/api-docs").with(asUser(ADMIN, "SUPER_ADMIN"))) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.components.schemas.DraftFileContentResponse.properties.encoding").exists()) + .andReturn().getResponse().getContentAsString(); + Files.writeString(Path.of("target", "authoring-openapi.json"), schema); + } } diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/dto/DraftFileContentResponseTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/dto/DraftFileContentResponseTest.java new file mode 100644 index 00000000..ba485327 --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/dto/DraftFileContentResponseTest.java @@ -0,0 +1,35 @@ +package com.iflytek.skillhub.dto; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.iflytek.skillhub.domain.authoring.DraftFile; +import com.iflytek.skillhub.domain.authoring.service.SkillDraftService; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import org.junit.jupiter.api.Test; + +class DraftFileContentResponseTest { + @Test + void keepsValidUtf8TextEditable() { + byte[] bytes = "技能 draft".getBytes(StandardCharsets.UTF_8); + var response = response(bytes); + assertThat(response.encoding()).isEqualTo("text"); + assertThat(response.content().getBytes(StandardCharsets.UTF_8)).isEqualTo(bytes); + } + + @Test + void binaryContentCanBeReadAndSavedWithoutChangingItsBytes() { + byte[] bytes = {(byte) 0x89, 0x50, 0x4e, 0x47, 0, (byte) 0xff}; + var response = response(bytes); + var request = new SaveDraftFileRequest("image.png", response.content(), response.encoding(), + "image/png", 1); + assertThat(request.isBase64()).isTrue(); + assertThat(Base64.getDecoder().decode(request.content())).isEqualTo(bytes); + } + + private DraftFileContentResponse response(byte[] bytes) { + var file = new DraftFile(1L, "file", "digest", (long) bytes.length, + "application/octet-stream", "drafts/1/digest"); + return DraftFileContentResponse.from(new SkillDraftService.FileContent(file, bytes)); + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcasterTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcasterTest.java new file mode 100644 index 00000000..4be41f15 --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/authoring/ValidationEventBroadcasterTest.java @@ -0,0 +1,65 @@ +package com.iflytek.skillhub.service.authoring; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.iflytek.skillhub.domain.authoring.validation.ValidationEvent; +import com.iflytek.skillhub.domain.authoring.validation.ValidationEventType; +import com.iflytek.skillhub.domain.shared.exception.DomainConflictException; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ValidationEventBroadcasterTest { + @Test + void replayDoesNotLoseLowerSequencesWhenLiveDeliveryWinsTheRace() { + ValidationEventBroadcaster.EmitterSession session = new ValidationEventBroadcaster.EmitterSession(0); + session.accept(event(3, ValidationEventType.PHASE_FINISHED), "live-3"); + session.accept(event(1, ValidationEventType.RUN_STARTED), "replay-1"); + session.accept(event(2, ValidationEventType.PHASE_STARTED), "replay-2"); + assertThat(session.lastSeq()).isZero(); + assertThat(session.finishReplay()).isTrue(); + assertThat(session.lastSeq()).isEqualTo(3); + } + + @Test + void outOfOrderTerminalWaitsForTheMissingEvent() { + ValidationEventBroadcaster.EmitterSession session = new ValidationEventBroadcaster.EmitterSession(0); + session.finishReplay(); + session.accept(event(2, ValidationEventType.RUN_FINISHED), "terminal"); + assertThat(session.lastSeq()).isZero(); + assertThat(session.accept(event(1, ValidationEventType.RUN_STARTED), "first")).isFalse(); + assertThat(session.lastSeq()).isEqualTo(2); + } + + @Test + void registrationPrecedesDatabaseReplayAndTerminalDetachesTheEmptyRun() { + ValidationEventBroadcaster broadcaster = new ValidationEventBroadcaster(new ObjectMapper()); + broadcaster.subscribe(1L, 0, () -> { + broadcaster.completeRun(1L, event(2, ValidationEventType.RUN_FINISHED)); + return List.of(event(1, ValidationEventType.RUN_STARTED)); + }); + assertThat((Map) ReflectionTestUtils.getField(broadcaster, "sessionsByRun")).isEmpty(); + } + + @Test + void subscriberCountAndGapBufferAreBounded() { + ValidationEventBroadcaster broadcaster = new ValidationEventBroadcaster(new ObjectMapper()); + for (int i = 0; i < 16; i++) { broadcaster.subscribe(1L, 0, List::of); } + assertThatThrownBy(() -> broadcaster.subscribe(1L, 0, List::of)) + .isInstanceOf(DomainConflictException.class); + ValidationEventBroadcaster.EmitterSession session = new ValidationEventBroadcaster.EmitterSession(0); + for (int seq = 2; seq <= 1025; seq++) { + assertThat(session.accept(event(seq, ValidationEventType.FINDING), "gap")).isTrue(); + } + assertThat(session.accept(event(1026, ValidationEventType.FINDING), "overflow")).isFalse(); + } + + private static ValidationEvent event(int seq, ValidationEventType type) { + ValidationEvent event = new ValidationEvent(1L, seq, type, null, Map.of()); + ReflectionTestUtils.setField(event, "createdAt", Instant.EPOCH); + return event; + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestratorTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestratorTest.java new file mode 100644 index 00000000..9c15aa19 --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/authoring/ValidationRunOrchestratorTest.java @@ -0,0 +1,140 @@ +package com.iflytek.skillhub.service.authoring; + +import com.iflytek.skillhub.config.AuthoringProperties; +import com.iflytek.skillhub.domain.authoring.service.DraftStructureValidator; +import com.iflytek.skillhub.domain.authoring.service.SkillDraftService; +import com.iflytek.skillhub.domain.authoring.service.ValidationRunService; +import com.iflytek.skillhub.domain.authoring.validation.ValidationEvent; +import com.iflytek.skillhub.domain.authoring.validation.ValidationEventType; +import com.iflytek.skillhub.domain.authoring.validation.ValidationRun; +import com.iflytek.skillhub.domain.authoring.validation.ValidationRunStatus; +import com.iflytek.skillhub.domain.shared.exception.DomainConflictException; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class ValidationRunOrchestratorTest { + private final ValidationRunService runs = mock(ValidationRunService.class); + private final SkillDraftService drafts = mock(SkillDraftService.class); + private final DraftStructureValidator validator = mock(DraftStructureValidator.class); + private final ValidationEventBroadcaster broadcaster = mock(ValidationEventBroadcaster.class); + + @Test + void duplicateClaimCannotEmitAnActiveRunAsFinished() { + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + try { + when(runs.claimForExecution(1L)).thenThrow(new DomainConflictException("already running")); + when(runs.appendTerminalEvent(1L)).thenReturn(null); + orchestrator(Runnable::run, timer, 1000).submit(1L); + verify(broadcaster, never()).completeRun(anyLong(), any()); + verify(runs, never()).appendEvent(anyLong(), eq(ValidationEventType.RUN_FINISHED), any(), any()); + } finally { timer.shutdownNow(); } + } + + @Test + void rejectionSettlesOnlyQueuedRunsWithoutExecutingOnRequestThread() { + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + try { + orchestrator(task -> { throw new RejectedExecutionException(); }, timer, 1000).submit(1L); + verify(runs).settleIfQueued(eq(1L), eq(1), eq(0), anyMap()); + verify(runs, never()).claimForExecution(anyLong()); + verify(runs, never()).settleIfActive(anyLong(), any(), anyInt(), anyInt(), anyMap()); + } finally { timer.shutdownNow(); } + } + + @Test + void executionDeadlineSettlesEvenWhenValidatorIgnoresInterrupts() throws Exception { + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + CountDownLatch validating = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch timedOut = new CountDownLatch(1); + CountDownLatch workerEnded = new CountDownLatch(1); + AtomicBoolean terminal = new AtomicBoolean(); + ValidationRun run = new ValidationRun(2L, 1, "owner"); + ReflectionTestUtils.setField(run, "id", 1L); + run.markPreparing(); + run.markRunning(Instant.now()); + when(runs.claimForExecution(1L)).thenReturn(run); + when(drafts.materializeSnapshot(2L, 1)).thenReturn( + new SkillDraftService.MaterializedSnapshot("test", "requirement", List.of())); + when(runs.appendEvent(eq(1L), any(), any(), anyMap())).thenAnswer(invocation -> { + if (terminal.get()) { + workerEnded.countDown(); + throw new DomainConflictException("already terminal"); + } + return new ValidationEvent(1L, 1, invocation.getArgument(1), null, Map.of()); + }); + doAnswer(invocation -> { + terminal.set(true); + timedOut.countDown(); + return java.util.Optional.empty(); + }).when(runs).settleIfActive(eq(1L), eq(ValidationRunStatus.TIMED_OUT), anyInt(), anyInt(), anyMap()); + when(validator.validate(anyString(), anyString(), anyList())).thenAnswer(invocation -> { + validating.countDown(); + boolean released = false; + while (!released) { + try { released = release.await(2, TimeUnit.SECONDS); } + catch (InterruptedException ignored) { /* intentionally uncooperative validator */ } + } + return new DraftStructureValidator.StructureReport(List.of(), true); + }); + try { + orchestrator(executor, timer, 100).submit(1L); + assertThat(validating.await(2, TimeUnit.SECONDS)).isTrue(); + assertThat(timedOut.await(2, TimeUnit.SECONDS)).isTrue(); + release.countDown(); + assertThat(workerEnded.await(2, TimeUnit.SECONDS)).isTrue(); + verify(runs, never()).recordFinding(anyLong(), any()); + verify(runs, never()).settleIfActive(eq(1L), eq(ValidationRunStatus.SUCCEEDED), anyInt(), anyInt(), anyMap()); + } finally { release.countDown(); executor.shutdownNow(); timer.shutdownNow(); } + } + + @Test + void snapshotRevisionConflictAfterClaimSettlesInsteadOfLeavingRunning() { + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + try { + ValidationRun run = new ValidationRun(2L, 1, "owner"); + ReflectionTestUtils.setField(run, "id", 1L); + run.markPreparing(); + run.markRunning(Instant.now()); + when(runs.claimForExecution(1L)).thenReturn(run); + when(drafts.materializeSnapshot(2L, 1)).thenThrow(new DomainConflictException("revision changed")); + orchestrator(Runnable::run, timer, 1000).submit(1L); + verify(runs).settleIfActive(eq(1L), eq(ValidationRunStatus.FAILED), eq(1), eq(0), anyMap()); + } finally { timer.shutdownNow(); } + } + + @Test + void completedCancelDoesNotAllocatePermanentControlState() { + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + try { + ValidationRun run = new ValidationRun(2L, 1, "owner"); + run.finish(ValidationRunStatus.SUCCEEDED, 0, 0, Map.of(), Instant.now()); + when(runs.requestCancel(eq(1L), eq("owner"), anySet())).thenReturn(run); + ValidationRunOrchestrator orchestrator = orchestrator(Runnable::run, timer, 1000); + orchestrator.requestCancel(1L, "owner", java.util.Set.of()); + assertThat((Map) ReflectionTestUtils.getField(orchestrator, "tasks")).isEmpty(); + verify(runs, never()).settleIfActive(anyLong(), any(), anyInt(), anyInt(), anyMap()); + } finally { timer.shutdownNow(); } + } + + private ValidationRunOrchestrator orchestrator(java.util.concurrent.Executor executor, + ScheduledExecutorService timer, long timeout) { + AuthoringProperties properties = new AuthoringProperties(); + properties.setRunTimeoutMs(timeout); + return new ValidationRunOrchestrator(runs, drafts, validator, broadcaster, executor, timer, properties); + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTaskTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTaskTest.java new file mode 100644 index 00000000..fddf3f14 --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/task/ValidationRunMaintenanceTaskTest.java @@ -0,0 +1,35 @@ +package com.iflytek.skillhub.task; + +import com.iflytek.skillhub.config.AuthoringProperties; +import com.iflytek.skillhub.domain.authoring.service.ValidationRunService; +import com.iflytek.skillhub.domain.shared.exception.DomainNotFoundException; +import com.iflytek.skillhub.service.authoring.ValidationRunOrchestrator; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +class ValidationRunMaintenanceTaskTest { + @TempDir Path root; + + @Test + void databaseFailureRetainsWorkspaceButConfirmedMissingRunDeletesIt() throws Exception { + Path unavailable = Files.createDirectory(root.resolve("run-1")); + Path missing = Files.createDirectory(root.resolve("run-2")); + Files.writeString(unavailable.resolve("input.md"), "retained"); + Files.writeString(missing.resolve("input.md"), "orphan"); + ValidationRunService runs = mock(ValidationRunService.class); + when(runs.findActiveRunsCreatedBefore(any())).thenReturn(List.of()); + when(runs.getRun(1L)).thenThrow(new IllegalStateException("database unavailable")); + when(runs.getRun(2L)).thenThrow(new DomainNotFoundException("missing")); + AuthoringProperties properties = new AuthoringProperties(); + properties.setWorkspaceRoot(root.toString()); + new ValidationRunMaintenanceTask(runs, mock(ValidationRunOrchestrator.class), properties).sweep(); + assertThat(unavailable.resolve("input.md")).exists(); + assertThat(missing).doesNotExist(); + } +} diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/DraftFile.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/DraftFile.java index 1ff28c06..5df5bdfb 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/DraftFile.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/DraftFile.java @@ -56,6 +56,9 @@ public class DraftFile { String contentType, String storageKey) { this.draftId = draftId; this.filePath = filePath; + if (size == null || size < 0) { + throw new IllegalArgumentException("Draft file size must be nonnegative"); + } this.sha256 = sha256; this.size = size; this.contentType = contentType; @@ -75,6 +78,9 @@ public class DraftFile { } public void updateContent(String sha256, Long size, String contentType, String storageKey) { + if (size == null || size < 0) { + throw new IllegalArgumentException("Draft file size must be nonnegative"); + } this.sha256 = sha256; this.size = size; this.contentType = contentType; diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraft.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraft.java index c8f3f842..24c069b8 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraft.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraft.java @@ -100,6 +100,9 @@ public class SkillDraft { * is still considered validated. */ public void applyContentChange(int newRevision, String newContentDigest) { + if (newRevision <= 0) { + throw new IllegalArgumentException("Draft revision must be positive"); + } this.revision = newRevision; this.contentDigest = newContentDigest; } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraftRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraftRepository.java index 87611eb6..6cadaabd 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraftRepository.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/SkillDraftRepository.java @@ -12,6 +12,8 @@ public interface SkillDraftRepository { Optional findById(Long id); + Optional findByIdForUpdate(Long id); + List findByOwnerIdOrderByUpdatedAtDesc(String ownerId); List findByNamespaceId(Long namespaceId); diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/SkillDraftService.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/SkillDraftService.java index 17474859..0d2a344e 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/SkillDraftService.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/SkillDraftService.java @@ -34,6 +34,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; /** * Domain service for skill authoring drafts: creation with a scaffold, content-addressed @@ -138,21 +140,23 @@ public class SkillDraftService { @Transactional public void deleteDraft(Long draftId, String userId, Set platformRoles) { - SkillDraft draft = getOwnedDraft(draftId, userId, platformRoles); + SkillDraft draft = getOwnedDraftForUpdate(draftId, userId, platformRoles); List storageKeys = fileRepository.findByDraftIdOrderByFilePath(draftId).stream() .map(DraftFile::getStorageKey) .toList(); - if (!storageKeys.isEmpty()) { - try { - objectStorageService.deleteObjects(storageKeys); - } catch (Exception exception) { - // storage garbage is not worth blocking a draft deletion; keys are - // namespaced under drafts/{draftId} and become unreachable - log.warn("Failed to clean up draft storage for draft {}: {}", - draftId, exception.getMessage()); - } - } draftRepository.delete(draft); + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + if (!storageKeys.isEmpty()) { + try { + objectStorageService.deleteObjects(storageKeys); + } catch (RuntimeException exception) { + log.warn("Failed to clean up storage after deleting draft {}", draftId, exception); + } + } + } + }); } // ---------------------------------------------------------------- file editing @@ -165,7 +169,7 @@ public class SkillDraftService { public SaveFileOutcome saveFile(Long draftId, String userId, String rawPath, byte[] content, String contentType, Integer expectedRevision, Set platformRoles) { - SkillDraft draft = getOwnedDraft(draftId, userId, platformRoles); + SkillDraft draft = getOwnedDraftForUpdate(draftId, userId, platformRoles); assertRevisionMatches(draft, expectedRevision); String path = normalizeAndCheckPath(rawPath); @@ -211,7 +215,7 @@ public class SkillDraftService { @Transactional public void deleteFile(Long draftId, String userId, String rawPath, Integer expectedRevision, Set platformRoles) { - SkillDraft draft = getOwnedDraft(draftId, userId, platformRoles); + SkillDraft draft = getOwnedDraftForUpdate(draftId, userId, platformRoles); assertRevisionMatches(draft, expectedRevision); String path = normalizeAndCheckPath(rawPath); DraftFile file = fileRepository.findByDraftIdAndFilePath(draftId, path) @@ -263,7 +267,7 @@ public class SkillDraftService { @Transactional public SkillDraft applyPatch(Long draftId, String userId, List patches, Integer expectedRevision, Set platformRoles) { - SkillDraft draft = getOwnedDraft(draftId, userId, platformRoles); + SkillDraft draft = getOwnedDraftForUpdate(draftId, userId, platformRoles); assertRevisionMatches(draft, expectedRevision); if (patches == null || patches.isEmpty()) { throw new DomainBadRequestException("error.authoring.patch.empty"); @@ -307,17 +311,32 @@ public class SkillDraftService { * Builds streaming package entries from the draft's current files. Used by the * validation orchestrator (snapshot) and the submit flow (publish input). */ - @Transactional(readOnly = true) - public List materializeEntries(Long draftId) { + public record MaterializedSnapshot(String name, String requirement, List entries) {} + + /** Locks the draft while checking its revision and reading its immutable content. */ + @Transactional + public MaterializedSnapshot materializeSnapshot(Long draftId, Integer expectedRevision) { + SkillDraft draft = draftRepository.findByIdForUpdate(draftId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.draft.notFound", draftId)); + assertRevisionMatches(draft, expectedRevision); List entries = new ArrayList<>(); for (DraftFile file : fileRepository.findByDraftIdOrderByFilePath(draftId)) { - entries.add(PackageEntry.streaming( - file.getFilePath(), - file.getSize(), - file.getContentType(), - () -> objectStorageService.getObject(file.getStorageKey()))); + try (InputStream input = objectStorageService.getObject(file.getStorageKey())) { + byte[] content = input.readNBytes(Math.toIntExact(file.getSize()) + 1); + if (content.length != file.getSize()) { + throw new IllegalStateException("Draft file size changed: " + file.getFilePath()); + } + entries.add(new PackageEntry(file.getFilePath(), content, file.getSize(), file.getContentType())); + } catch (IOException exception) { + throw new IllegalStateException("Failed to snapshot draft file: " + file.getFilePath(), exception); + } } - return entries; + return new MaterializedSnapshot(draft.getName(), draft.getRequirement(), List.copyOf(entries)); + } + + @Transactional + public List materializeEntries(Long draftId) { + return materializeSnapshot(draftId, null).entries(); } @Transactional(readOnly = true) @@ -342,6 +361,13 @@ public class SkillDraftService { // ---------------------------------------------------------------- internals + private SkillDraft getOwnedDraftForUpdate(Long draftId, String userId, Set platformRoles) { + SkillDraft draft = draftRepository.findByIdForUpdate(draftId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.draft.notFound", draftId)); + assertOwner(draft, userId, platformRoles); + return draft; + } + private DraftFile storeFile(Long draftId, String path, byte[] content, String contentType) { return new DraftFile(draftId, path, sha256Hex(content), (long) content.length, contentType, storageKey(draftId, content)); @@ -370,6 +396,10 @@ public class SkillDraftService { private String resolveContentType(String path, String provided) { if (provided != null && !provided.isBlank()) { + if (provided.length() > 128) { + throw new DomainBadRequestException("error.authoring.file.contentMismatch", path, + "content type exceeds 128 characters"); + } return provided; } String lower = path.toLowerCase(Locale.ROOT); @@ -399,6 +429,9 @@ public class SkillDraftService { } catch (IllegalArgumentException exception) { throw new DomainBadRequestException("error.authoring.file.path.invalid", rawPath); } + if (normalized.length() > 512) { + throw new DomainBadRequestException("error.authoring.file.path.invalid", rawPath); + } if (!SkillPackagePolicy.hasAllowedExtension(normalized)) { throw new DomainBadRequestException("error.authoring.file.extension.disallowed", normalized); } diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/ValidationRunService.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/ValidationRunService.java index f2e21c96..7b8ce953 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/ValidationRunService.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/service/ValidationRunService.java @@ -64,7 +64,7 @@ public class ValidationRunService { */ @Transactional public ValidationRun startRun(Long draftId, String userId, Set platformRoles) { - SkillDraft draft = draftRepository.findById(draftId) + SkillDraft draft = draftRepository.findByIdForUpdate(draftId) .orElseThrow(() -> new DomainNotFoundException("error.authoring.draft.notFound", draftId)); assertOwner(draft, userId, platformRoles); if (runRepository.findByDraftIdOrderByCreatedAtDesc(draftId).stream() @@ -133,6 +133,15 @@ public class ValidationRunService { public Optional settleIfActive(Long runId, ValidationRunStatus terminalStatus, int errorCount, int warningCount, Map summary) { + // Only successful validation updates the draft. Timeout/failure must not wait + // for snapshot object-storage IO while it holds the draft lock. + SkillDraft draft = null; + if (terminalStatus == ValidationRunStatus.SUCCEEDED) { + Long draftId = runRepository.findDraftIdById(runId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.run.notFound", runId)); + draft = draftRepository.findByIdForUpdate(draftId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.draft.notFound", draftId)); + } ValidationRun run = runRepository.findByIdForUpdate(runId) .orElseThrow(() -> new DomainNotFoundException("error.authoring.run.notFound", runId)); if (run.getStatus().isTerminal()) { @@ -141,31 +150,82 @@ public class ValidationRunService { run.finish(terminalStatus, errorCount, warningCount, summary, clock.instant()); ValidationRun settled = runRepository.save(run); - if (terminalStatus == ValidationRunStatus.SUCCEEDED && errorCount == 0) { - draftRepository.findById(settled.getDraftId()).ifPresent(draft -> { - if (draft.getRevision().equals(settled.getDraftRevision())) { - draft.markValidated(settled.getDraftRevision(), settled.getId()); - draftRepository.save(draft); - } - }); + if (terminalStatus == ValidationRunStatus.SUCCEEDED && errorCount == 0 + && draft.getRevision().equals(settled.getDraftRevision())) { + draft.markValidated(settled.getDraftRevision(), settled.getId()); + draftRepository.save(draft); } return Optional.of(settled); } + /** Queue rejection may fail an unclaimed run, never a duplicate already executing. */ + @Transactional + public Optional settleIfQueued(Long runId, int errorCount, int warningCount, + Map summary) { + Long draftId = runRepository.findDraftIdById(runId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.run.notFound", runId)); + draftRepository.findByIdForUpdate(draftId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.draft.notFound", draftId)); + ValidationRun run = lockRun(runId); + if (run.getStatus() != ValidationRunStatus.QUEUED) { + return Optional.empty(); + } + run.finish(ValidationRunStatus.FAILED, errorCount, warningCount, summary, clock.instant()); + return Optional.of(runRepository.save(run)); + } + // ---------------------------------------------------------------- events & findings /** Appends one event with the next sequence number; commits independently. */ @Transactional(propagation = Propagation.REQUIRES_NEW) public ValidationEvent appendEvent(Long runId, ValidationEventType type, String phase, Map payload) { - int nextSeq = eventRepository.findMaxSeqByRunId(runId).map(seq -> seq + 1).orElse(1); - return eventRepository.save( - new ValidationEvent(runId, nextSeq, type, phase, payload)); + ValidationRun run = lockRun(runId); + requireActive(run); + if (type == ValidationEventType.RUN_FINISHED) { + throw new DomainConflictException("error.authoring.run.terminal", runId, run.getStatus()); + } + return saveEvent(runId, type, phase, payload); } - /** Records one finding and returns it; commits independently. */ + /** Persists exactly one event from the actual terminal state; active runs return null. */ + @Transactional(propagation = Propagation.REQUIRES_NEW) + public ValidationEvent appendTerminalEvent(Long runId) { + ValidationRun run = lockRun(runId); + if (run.isActive()) { + return null; + } + Optional existing = eventRepository.findByRunIdOrderBySeqAsc(runId).stream() + .filter(event -> event.getEventType() == ValidationEventType.RUN_FINISHED).findFirst(); + if (existing.isPresent()) { + return existing.get(); + } + return saveEvent(runId, ValidationEventType.RUN_FINISHED, null, Map.of( + "status", run.getStatus().name(), "errorCount", run.getErrorCount(), + "warningCount", run.getWarningCount())); + } + + private ValidationEvent saveEvent(Long runId, ValidationEventType type, String phase, + Map payload) { + int nextSeq = eventRepository.findMaxSeqByRunId(runId).map(seq -> seq + 1).orElse(1); + return eventRepository.save(new ValidationEvent(runId, nextSeq, type, phase, payload)); + } + + private ValidationRun lockRun(Long runId) { + return runRepository.findByIdForUpdate(runId) + .orElseThrow(() -> new DomainNotFoundException("error.authoring.run.notFound", runId)); + } + + private void requireActive(ValidationRun run) { + if (!run.isActive()) { + throw new DomainConflictException("error.authoring.run.terminal", run.getId(), run.getStatus()); + } + } + + /** Records one finding only while the run is active; serialized with settlement. */ @Transactional(propagation = Propagation.REQUIRES_NEW) public ValidationFinding recordFinding(Long runId, FindingDraft draft) { + requireActive(lockRun(runId)); return findingRepository.save(new ValidationFinding( runId, draft.layer(), draft.ruleCode(), draft.severity(), draft.filePath(), draft.location(), draft.message(), draft.suggestion())); diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRun.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRun.java index ecb72396..05149daa 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRun.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRun.java @@ -110,6 +110,9 @@ public class ValidationRun { if (terminalStatus == null || !terminalStatus.isTerminal()) { throw new IllegalArgumentException("Terminal status required: " + terminalStatus); } + if (errorCount < 0 || warningCount < 0) { + throw new IllegalArgumentException("Finding counts must be nonnegative"); + } requireActive("finish"); status = terminalStatus; this.errorCount = errorCount; diff --git a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRunRepository.java b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRunRepository.java index 689f464d..514ef42e 100644 --- a/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRunRepository.java +++ b/server/skillhub-domain/src/main/java/com/iflytek/skillhub/domain/authoring/validation/ValidationRunRepository.java @@ -15,6 +15,8 @@ public interface ValidationRunRepository { Optional findByIdForUpdate(Long id); + Optional findDraftIdById(Long id); + List findByDraftIdOrderByCreatedAtDesc(Long draftId); /** Latest run recorded for an exact draft revision, used by the submit gate. */ diff --git a/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/authoring/service/AuthoringEntityValidationTest.java b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/authoring/service/AuthoringEntityValidationTest.java new file mode 100644 index 00000000..7cc0e193 --- /dev/null +++ b/server/skillhub-domain/src/test/java/com/iflytek/skillhub/domain/authoring/service/AuthoringEntityValidationTest.java @@ -0,0 +1,47 @@ +package com.iflytek.skillhub.domain.authoring.service; + +import com.iflytek.skillhub.domain.authoring.DraftFile; +import com.iflytek.skillhub.domain.authoring.SkillDraft; +import com.iflytek.skillhub.domain.authoring.validation.ValidationRun; +import com.iflytek.skillhub.domain.authoring.validation.ValidationRunStatus; +import java.time.Instant; +import java.util.Map; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** Mutable authoring rules are guarded in code rather than migration CHECK constraints. */ +class AuthoringEntityValidationTest { + + @Test + void invalidRevisionDoesNotChangeDraftContent() { + SkillDraft draft = new SkillDraft(1L, "owner", "draft", null, "original"); + assertThrows(IllegalArgumentException.class, () -> draft.applyContentChange(0, "invalid")); + assertThat(draft.getRevision()).isEqualTo(1); + assertThat(draft.getContentDigest()).isEqualTo("original"); + } + + @Test + void negativeFileSizeIsRejectedOnConstructionAndUpdate() { + assertThrows(IllegalArgumentException.class, + () -> new DraftFile(1L, "SKILL.md", "hash", -1L, "text/markdown", "key")); + DraftFile file = new DraftFile(1L, "SKILL.md", "hash", 0L, "text/markdown", "key"); + assertThrows(IllegalArgumentException.class, + () -> file.updateContent("new-hash", -1L, "text/markdown", "new-key")); + assertThat(file.getSize()).isZero(); + assertThat(file.getSha256()).isEqualTo("hash"); + assertThat(file.getStorageKey()).isEqualTo("key"); + } + + @Test + void invalidFindingCountsCannotSettleRun() { + ValidationRun run = new ValidationRun(1L, 1, "owner"); + assertThrows(IllegalArgumentException.class, () -> run.finish( + ValidationRunStatus.FAILED, -1, 0, Map.of(), Instant.now())); + assertThrows(IllegalArgumentException.class, () -> run.finish( + ValidationRunStatus.FAILED, 0, -1, Map.of(), Instant.now())); + assertThat(run.getStatus()).isEqualTo(ValidationRunStatus.QUEUED); + assertThat(run.getFinishedAt()).isNull(); + } +} diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/SkillDraftJpaRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/SkillDraftJpaRepository.java index db139890..bb67d8e2 100644 --- a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/SkillDraftJpaRepository.java +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/SkillDraftJpaRepository.java @@ -5,6 +5,8 @@ import com.iflytek.skillhub.domain.authoring.SkillDraftRepository; import java.util.List; import java.util.Optional; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; import org.springframework.stereotype.Repository; /** @@ -13,6 +15,10 @@ import org.springframework.stereotype.Repository; @Repository public interface SkillDraftJpaRepository extends JpaRepository, SkillDraftRepository { + @Override + @Query(value = "SELECT * FROM skill_draft WHERE id = :id FOR UPDATE", nativeQuery = true) + Optional findByIdForUpdate(@Param("id") Long id); + List findByOwnerIdOrderByUpdatedAtDesc(String ownerId); List findByNamespaceId(Long namespaceId); diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ValidationRunJpaRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ValidationRunJpaRepository.java index d9706ea1..2fd75e8e 100644 --- a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ValidationRunJpaRepository.java +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/ValidationRunJpaRepository.java @@ -25,6 +25,10 @@ public interface ValidationRunJpaRepository extends JpaRepository findByIdForUpdate(@Param("id") Long id); + @Override + @Query(value = "SELECT draft_id FROM validation_run WHERE id = :id", nativeQuery = true) + Optional findDraftIdById(@Param("id") Long id); + List findByDraftIdOrderByCreatedAtDesc(Long draftId); Optional findFirstByDraftIdAndDraftRevisionOrderByCreatedAtDesc( diff --git a/web/src/api/generated/schema.d.ts b/web/src/api/generated/schema.d.ts index 6b7f10a9..f03228da 100644 --- a/web/src/api/generated/schema.d.ts +++ b/web/src/api/generated/schema.d.ts @@ -8071,6 +8071,7 @@ export interface components { size?: number; contentType?: string; content?: string; + encoding?: string; }; ApiResponseListDraftFileResponse: { /** Format: int32 */