mirror of
https://github.com/iflytek/skillhub.git
synced 2026-10-15 04:18:20 +00:00
fix(authoring): protect draft transactions and validation lifecycle
Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>
This commit is contained in:
XiaoSeS
parent
8a69827286
commit
642e780f1e
31 changed files with 914 additions and 155 deletions
29
docs/27-authoring-drafts.md
Normal file
29
docs/27-authoring-drafts.md
Normal file
|
|
@ -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 为准；本文件不替代接口定义。
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -139,12 +139,7 @@ public class SkillAuthoringController extends BaseApiController {
|
|||
Set<String> 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")
|
||||
|
|
|
|||
|
|
@ -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<String> platformRoles) {
|
||||
ValidationRun run = runService.getOwnedRun(runId, userId, platformRoles);
|
||||
int lastSeq = resolveCursor(afterSeq, lastEventId);
|
||||
List<ValidationEvent> 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) {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
) {
|
||||
|
|
|
|||
|
|
@ -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<ValidationEvent> replay) {
|
||||
public SseEmitter subscribe(Long runId, int lastSeq, Supplier<List<ValidationEvent>> 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<EmitterSession> 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<EmitterSession> 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<EmitterSession> sessions = sessionsByRun.remove(runId);
|
||||
if (sessions == null) {
|
||||
return;
|
||||
}
|
||||
for (EmitterSession session : sessions) {
|
||||
session.emitter().complete();
|
||||
}
|
||||
}
|
||||
|
||||
private void detach(Long runId, EmitterSession session) {
|
||||
List<EmitterSession> sessions = sessionsByRun.get(runId);
|
||||
if (sessions != null) {
|
||||
sessions.remove(session);
|
||||
synchronized (sessionsByRun) {
|
||||
List<EmitterSession> 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<Integer, PendingEvent> 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) { }
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Long, AtomicBoolean> cancelFlags = new ConcurrentHashMap<>();
|
||||
|
||||
/** Serializes the RUN_FINISHED emission so it happens exactly once per run. */
|
||||
private final Map<Long, Object> terminalLocks = new ConcurrentHashMap<>();
|
||||
private final ScheduledExecutorService timer;
|
||||
private final AuthoringProperties properties;
|
||||
private final Map<Long, FutureTask<Void>> 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<Void> 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<String> 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<Void> task = tasks.get(runId);
|
||||
if (task != null) { task.cancel(true); }
|
||||
return run;
|
||||
}
|
||||
|
||||
|
|
@ -88,63 +113,77 @@ public class ValidationRunOrchestrator {
|
|||
Map<String, Object> summary) {
|
||||
runService.settleIfActive(runId, status, errorCount, warningCount, summary);
|
||||
emitTerminalOnce(runId);
|
||||
if (status == ValidationRunStatus.TIMED_OUT) {
|
||||
FutureTask<Void> 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<PackageEntry> 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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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=Достигнут предел подключений к событиям проверки. Закройте существующее подключение и повторите попытку.
|
||||
|
|
|
|||
|
|
@ -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=校验事件连接数量已达上限，请关闭已有连接后重试。
|
||||
|
|
|
|||
|
|
@ -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<String, byte[]> 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<Boolean> 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<Boolean> 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<ValidationEvent> 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<ValidationEvent> 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<Boolean> 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 <T> List<T> concurrently(int workers, Callable<T> action) throws Exception {
|
||||
CountDownLatch ready = new CountDownLatch(workers);
|
||||
CountDownLatch start = new CountDownLatch(1);
|
||||
try (var executor = Executors.newFixedThreadPool(workers)) {
|
||||
List<Future<T>> 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<T> results = new ArrayList<>();
|
||||
for (Future<T> 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);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ public interface SkillDraftRepository {
|
|||
|
||||
Optional<SkillDraft> findById(Long id);
|
||||
|
||||
Optional<SkillDraft> findByIdForUpdate(Long id);
|
||||
|
||||
List<SkillDraft> findByOwnerIdOrderByUpdatedAtDesc(String ownerId);
|
||||
|
||||
List<SkillDraft> findByNamespaceId(Long namespaceId);
|
||||
|
|
|
|||
|
|
@ -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<String> platformRoles) {
|
||||
SkillDraft draft = getOwnedDraft(draftId, userId, platformRoles);
|
||||
SkillDraft draft = getOwnedDraftForUpdate(draftId, userId, platformRoles);
|
||||
List<String> 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<String> 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<String> 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<FilePatch> patches,
|
||||
Integer expectedRevision, Set<String> 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<PackageEntry> materializeEntries(Long draftId) {
|
||||
public record MaterializedSnapshot(String name, String requirement, List<PackageEntry> 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<PackageEntry> 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<PackageEntry> 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<String> 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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ public class ValidationRunService {
|
|||
*/
|
||||
@Transactional
|
||||
public ValidationRun startRun(Long draftId, String userId, Set<String> 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<ValidationRun> settleIfActive(Long runId, ValidationRunStatus terminalStatus,
|
||||
int errorCount, int warningCount,
|
||||
Map<String, Object> 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<ValidationRun> settleIfQueued(Long runId, int errorCount, int warningCount,
|
||||
Map<String, Object> 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<String, Object> 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<ValidationEvent> 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<String, Object> 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()));
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -15,6 +15,8 @@ public interface ValidationRunRepository {
|
|||
|
||||
Optional<ValidationRun> findByIdForUpdate(Long id);
|
||||
|
||||
Optional<Long> findDraftIdById(Long id);
|
||||
|
||||
List<ValidationRun> findByDraftIdOrderByCreatedAtDesc(Long draftId);
|
||||
|
||||
/** Latest run recorded for an exact draft revision, used by the submit gate. */
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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<SkillDraft, Long>, SkillDraftRepository {
|
||||
|
||||
@Override
|
||||
@Query(value = "SELECT * FROM skill_draft WHERE id = :id FOR UPDATE", nativeQuery = true)
|
||||
Optional<SkillDraft> findByIdForUpdate(@Param("id") Long id);
|
||||
|
||||
List<SkillDraft> findByOwnerIdOrderByUpdatedAtDesc(String ownerId);
|
||||
|
||||
List<SkillDraft> findByNamespaceId(Long namespaceId);
|
||||
|
|
|
|||
|
|
@ -25,6 +25,10 @@ public interface ValidationRunJpaRepository extends JpaRepository<ValidationRun,
|
|||
@Query(value = "SELECT * FROM validation_run WHERE id = :id FOR UPDATE", nativeQuery = true)
|
||||
Optional<ValidationRun> findByIdForUpdate(@Param("id") Long id);
|
||||
|
||||
@Override
|
||||
@Query(value = "SELECT draft_id FROM validation_run WHERE id = :id", nativeQuery = true)
|
||||
Optional<Long> findDraftIdById(@Param("id") Long id);
|
||||
|
||||
List<ValidationRun> findByDraftIdOrderByCreatedAtDesc(Long draftId);
|
||||
|
||||
Optional<ValidationRun> findFirstByDraftIdAndDraftRevisionOrderByCreatedAtDesc(
|
||||
|
|
|
|||
1
web/src/api/generated/schema.d.ts
vendored
1
web/src/api/generated/schema.d.ts
vendored
|
|
@ -8071,6 +8071,7 @@ export interface components {
|
|||
size?: number;
|
||||
contentType?: string;
|
||||
content?: string;
|
||||
encoding?: string;
|
||||
};
|
||||
ApiResponseListDraftFileResponse: {
|
||||
/** Format: int32 */
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue