Linghong
b19b76af56
新增 Web Project runtime job、持久日志、lease、取消、expired、stale 和 active preview guard 状态机 接入 api-server Web Project runtime worker 与 TempDirBuildRuntime 构建执行链路 补齐 SpacetimeDB procedure、spacetime-client facade、shared contracts 和前端 web-project client 契约 更新 /editor/agent 的 runtime job 恢复、日志回填、SSE 重连、取消按钮和 active preview 刷新恢复 新增 P2 dev smoke 脚本，并让完整 npm run dev 默认以 all 角色启动 P2 worker 补充 P2 自动化测试、浏览器 smoke 验收记录、开发运维文档和 Hermes 踩坑记忆
473 lines
16 KiB
Rust
473 lines
16 KiB
Rust
use std::{
|
||
future::Future,
|
||
path::PathBuf,
|
||
pin::Pin,
|
||
process::Stdio,
|
||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||
};
|
||
|
||
use shared_contracts::web_project::{WebProjectPreviewBuild, WebProjectPreviewBuildStatus};
|
||
use shared_kernel::new_uuid_simple_string;
|
||
use spacetime_client::{
|
||
WebProjectPreviewBuildGetRecordInput, WebProjectPreviewBuildUpdateRecordInput,
|
||
WebProjectRuntimeJobRecord, WebProjectSnapshotGetRecordInput, WebProjectSnapshotRecord,
|
||
};
|
||
use tokio::{
|
||
io::{AsyncReadExt, AsyncWriteExt},
|
||
process::Command,
|
||
};
|
||
use web_project_runner::{
|
||
WebProjectBuildInput, WebProjectBuildOutput, WebProjectBuildStatus, WebProjectRunnerFile,
|
||
};
|
||
|
||
use crate::{state::AppState, web_project};
|
||
|
||
const WEB_PROJECT_RUNNER_TIMEOUT: Duration = Duration::from_secs(300);
|
||
|
||
// P2-06 先保留子进程 runner 边界；后续 Docker / gVisor / microVM 只替换 SandboxRuntime 实现。
|
||
pub(crate) trait SandboxRuntime: Send + Sync {
|
||
fn run_preview_build<'a>(
|
||
&'a self,
|
||
state: &'a AppState,
|
||
build: &'a WebProjectPreviewBuild,
|
||
snapshot: WebProjectSnapshotRecord,
|
||
) -> Pin<Box<dyn Future<Output = Result<WebProjectBuildOutput, String>> + Send + 'a>>;
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy, Default)]
|
||
pub(crate) struct TempDirBuildRuntime;
|
||
|
||
impl SandboxRuntime for TempDirBuildRuntime {
|
||
fn run_preview_build<'a>(
|
||
&'a self,
|
||
state: &'a AppState,
|
||
build: &'a WebProjectPreviewBuild,
|
||
snapshot: WebProjectSnapshotRecord,
|
||
) -> Pin<Box<dyn Future<Output = Result<WebProjectBuildOutput, String>> + Send + 'a>> {
|
||
Box::pin(invoke_runner_process(state, build, snapshot))
|
||
}
|
||
}
|
||
|
||
#[derive(Debug, Clone)]
|
||
pub(crate) struct PreviewBuildRunOutput {
|
||
pub build: WebProjectPreviewBuild,
|
||
pub runner_output: WebProjectBuildOutput,
|
||
pub started_at_micros: i64,
|
||
pub finished_at_micros: i64,
|
||
}
|
||
|
||
#[derive(Debug, Clone)]
|
||
pub(crate) struct PreviewBuildFinishPlan {
|
||
pub succeeded: bool,
|
||
pub logs: Vec<String>,
|
||
pub artifact_id: Option<String>,
|
||
pub preview_token_id: Option<String>,
|
||
pub preview_url: Option<String>,
|
||
pub error_summary: Option<String>,
|
||
}
|
||
|
||
pub(crate) async fn run_preview_build_with_temp_dir_runtime(
|
||
state: &AppState,
|
||
job: &WebProjectRuntimeJobRecord,
|
||
) -> Result<PreviewBuildRunOutput, String> {
|
||
let runtime = TempDirBuildRuntime;
|
||
run_preview_build_with_runtime(state, job, &runtime).await
|
||
}
|
||
|
||
async fn run_preview_build_with_runtime<R: SandboxRuntime>(
|
||
state: &AppState,
|
||
job: &WebProjectRuntimeJobRecord,
|
||
runtime: &R,
|
||
) -> Result<PreviewBuildRunOutput, String> {
|
||
let preview_build_id = job
|
||
.preview_build_id
|
||
.as_deref()
|
||
.ok_or_else(|| "preview_build runtime job 缺少 preview_build_id".to_string())?;
|
||
let build = get_preview_build(state, &job.owner_user_id, preview_build_id).await?;
|
||
validate_runtime_job_preview_build(job, &build)?;
|
||
if is_terminal_preview_build_status(&build.status) {
|
||
return Err(format!(
|
||
"preview build {preview_build_id} 已终态，不能重复执行"
|
||
));
|
||
}
|
||
|
||
let started_at_micros = current_utc_micros();
|
||
let running_build = mark_preview_build_running(state, &job.owner_user_id, &build).await?;
|
||
let snapshot = state
|
||
.spacetime_client()
|
||
.get_web_project_snapshot(WebProjectSnapshotGetRecordInput {
|
||
project_id: job.project_id.clone(),
|
||
snapshot_id: Some(job.snapshot_id.clone()),
|
||
owner_user_id: job.owner_user_id.clone(),
|
||
})
|
||
.await
|
||
.map_err(|error| format!("hydrate snapshot 失败：{error}"))?
|
||
.snapshot;
|
||
let runner_output = runtime
|
||
.run_preview_build(state, &running_build, snapshot)
|
||
.await?;
|
||
let finished_at_micros = current_utc_micros();
|
||
|
||
Ok(PreviewBuildRunOutput {
|
||
build: running_build,
|
||
runner_output,
|
||
started_at_micros,
|
||
finished_at_micros,
|
||
})
|
||
}
|
||
|
||
pub(crate) async fn finish_preview_build_from_runner_output(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
running_build: &WebProjectPreviewBuild,
|
||
started_at_micros: i64,
|
||
finished_at_micros: i64,
|
||
runner_output: &WebProjectBuildOutput,
|
||
) -> Result<Option<String>, String> {
|
||
let finish_plan = plan_preview_build_finish_from_runner_output(
|
||
state,
|
||
running_build,
|
||
finished_at_micros,
|
||
runner_output,
|
||
);
|
||
let updated = state
|
||
.spacetime_client()
|
||
.update_web_project_preview_build(WebProjectPreviewBuildUpdateRecordInput {
|
||
job_id: running_build.job_id.clone(),
|
||
owner_user_id: owner_user_id.to_string(),
|
||
status: if finish_plan.succeeded {
|
||
"succeeded"
|
||
} else {
|
||
"failed"
|
||
}
|
||
.to_string(),
|
||
logs: finish_plan.logs,
|
||
artifact_id: finish_plan.artifact_id.clone(),
|
||
preview_token_id: finish_plan.preview_token_id,
|
||
preview_url: finish_plan.preview_url.clone(),
|
||
error_summary: finish_plan.error_summary.clone(),
|
||
started_at_micros: Some(started_at_micros),
|
||
finished_at_micros: Some(finished_at_micros),
|
||
updated_at_micros: finished_at_micros,
|
||
})
|
||
.await
|
||
.map_err(|error| error.to_string())?;
|
||
let build = web_project::build_from_record(updated.build);
|
||
web_project::publish_build_event(
|
||
state,
|
||
&build,
|
||
build
|
||
.error_summary
|
||
.as_deref()
|
||
.or(Some(if finish_plan.succeeded {
|
||
"构建完成"
|
||
} else {
|
||
"构建失败"
|
||
})),
|
||
);
|
||
|
||
Ok(finish_plan.error_summary)
|
||
}
|
||
|
||
pub(crate) fn plan_preview_build_finish_from_runner_output(
|
||
state: &AppState,
|
||
running_build: &WebProjectPreviewBuild,
|
||
finished_at_micros: i64,
|
||
runner_output: &WebProjectBuildOutput,
|
||
) -> PreviewBuildFinishPlan {
|
||
let succeeded = runner_output.status == WebProjectBuildStatus::Succeeded
|
||
&& runner_output.artifact_id.is_some();
|
||
let missing_artifact_error = (runner_output.status == WebProjectBuildStatus::Succeeded
|
||
&& runner_output.artifact_id.is_none())
|
||
.then(|| "web-project-runner 成功但缺少 artifact_id".to_string());
|
||
let preview_token_id = succeeded.then(|| build_preview_token_id(finished_at_micros));
|
||
let preview_url = preview_token_id
|
||
.as_ref()
|
||
.map(|token| build_preview_url(&state.config.web_project_preview_public_base_url, token));
|
||
let error_summary = missing_artifact_error.or_else(|| runner_output.error_summary.clone());
|
||
let mut logs = running_build.logs.clone();
|
||
logs.extend(runner_output.logs.clone());
|
||
logs.push(if succeeded {
|
||
"runtime worker: 构建完成，预览地址已生成".to_string()
|
||
} else {
|
||
"runtime worker: 构建失败".to_string()
|
||
});
|
||
|
||
PreviewBuildFinishPlan {
|
||
succeeded,
|
||
logs,
|
||
artifact_id: runner_output.artifact_id.clone().filter(|_| succeeded),
|
||
preview_token_id,
|
||
preview_url,
|
||
error_summary,
|
||
}
|
||
}
|
||
|
||
pub(crate) async fn fail_preview_build_for_runtime_worker(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
preview_build_id: &str,
|
||
error: &str,
|
||
) -> Result<WebProjectPreviewBuild, String> {
|
||
finish_preview_build_with_terminal_status(
|
||
state,
|
||
owner_user_id,
|
||
preview_build_id,
|
||
"failed",
|
||
"runtime worker: 构建任务失败",
|
||
error,
|
||
)
|
||
.await
|
||
}
|
||
|
||
pub(crate) async fn cancel_preview_build_for_runtime_worker(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
preview_build_id: &str,
|
||
) -> Result<WebProjectPreviewBuild, String> {
|
||
finish_preview_build_with_terminal_status(
|
||
state,
|
||
owner_user_id,
|
||
preview_build_id,
|
||
"cancelled",
|
||
"runtime worker: 构建任务已取消",
|
||
"构建任务已取消",
|
||
)
|
||
.await
|
||
}
|
||
|
||
async fn finish_preview_build_with_terminal_status(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
preview_build_id: &str,
|
||
status: &str,
|
||
log_message: &str,
|
||
error: &str,
|
||
) -> Result<WebProjectPreviewBuild, String> {
|
||
let build = get_preview_build(state, owner_user_id, preview_build_id).await?;
|
||
let now = current_utc_micros();
|
||
let mut logs = build.logs.clone();
|
||
logs.push(log_message.to_string());
|
||
logs.push(format!("runtime worker: {error}"));
|
||
let updated = state
|
||
.spacetime_client()
|
||
.update_web_project_preview_build(WebProjectPreviewBuildUpdateRecordInput {
|
||
job_id: build.job_id.clone(),
|
||
owner_user_id: owner_user_id.to_string(),
|
||
status: status.to_string(),
|
||
logs,
|
||
artifact_id: None,
|
||
preview_token_id: None,
|
||
preview_url: None,
|
||
error_summary: Some(error.to_string()),
|
||
started_at_micros: None,
|
||
finished_at_micros: Some(now),
|
||
updated_at_micros: now,
|
||
})
|
||
.await
|
||
.map_err(|error| error.to_string())?;
|
||
let build = web_project::build_from_record(updated.build);
|
||
web_project::publish_build_event(state, &build, build.error_summary.as_deref());
|
||
Ok(build)
|
||
}
|
||
|
||
async fn mark_preview_build_running(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
build: &WebProjectPreviewBuild,
|
||
) -> Result<WebProjectPreviewBuild, String> {
|
||
let started_at_micros = current_utc_micros();
|
||
let mut logs = build.logs.clone();
|
||
logs.push("runtime worker: hydrate snapshot".to_string());
|
||
logs.push("构建任务开始执行".to_string());
|
||
let updated = state
|
||
.spacetime_client()
|
||
.update_web_project_preview_build(WebProjectPreviewBuildUpdateRecordInput {
|
||
job_id: build.job_id.clone(),
|
||
owner_user_id: owner_user_id.to_string(),
|
||
status: "running".to_string(),
|
||
logs,
|
||
artifact_id: None,
|
||
preview_token_id: None,
|
||
preview_url: None,
|
||
error_summary: None,
|
||
started_at_micros: Some(started_at_micros),
|
||
finished_at_micros: None,
|
||
updated_at_micros: started_at_micros,
|
||
})
|
||
.await
|
||
.map_err(|error| error.to_string())?;
|
||
let running_build = web_project::build_from_record(updated.build);
|
||
web_project::publish_build_event(state, &running_build, Some("构建任务开始执行"));
|
||
Ok(running_build)
|
||
}
|
||
|
||
async fn get_preview_build(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
preview_build_id: &str,
|
||
) -> Result<WebProjectPreviewBuild, String> {
|
||
let mutation = state
|
||
.spacetime_client()
|
||
.get_web_project_preview_build(WebProjectPreviewBuildGetRecordInput {
|
||
job_id: preview_build_id.to_string(),
|
||
owner_user_id: owner_user_id.to_string(),
|
||
})
|
||
.await
|
||
.map_err(|error| error.to_string())?;
|
||
Ok(web_project::build_from_record(mutation.build))
|
||
}
|
||
|
||
fn validate_runtime_job_preview_build(
|
||
job: &WebProjectRuntimeJobRecord,
|
||
build: &WebProjectPreviewBuild,
|
||
) -> Result<(), String> {
|
||
if build.project_id != job.project_id
|
||
|| build.snapshot_id != job.snapshot_id
|
||
|| build.owner_user_id != job.owner_user_id
|
||
{
|
||
return Err(format!(
|
||
"runtime job {} 与 preview build {} 绑定信息不一致",
|
||
job.job_id, build.job_id
|
||
));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
fn is_terminal_preview_build_status(status: &WebProjectPreviewBuildStatus) -> bool {
|
||
matches!(
|
||
status,
|
||
WebProjectPreviewBuildStatus::Succeeded
|
||
| WebProjectPreviewBuildStatus::Failed
|
||
| WebProjectPreviewBuildStatus::Cancelled
|
||
| WebProjectPreviewBuildStatus::Expired
|
||
| WebProjectPreviewBuildStatus::Stale
|
||
)
|
||
}
|
||
|
||
async fn invoke_runner_process(
|
||
state: &AppState,
|
||
build: &WebProjectPreviewBuild,
|
||
snapshot: WebProjectSnapshotRecord,
|
||
) -> Result<WebProjectBuildOutput, String> {
|
||
let runner_bin = resolve_runner_binary(state)?;
|
||
let input = WebProjectBuildInput {
|
||
job_id: build.job_id.clone(),
|
||
project_id: build.project_id.clone(),
|
||
snapshot_id: build.snapshot_id.clone(),
|
||
files: snapshot
|
||
.files
|
||
.into_iter()
|
||
.map(|file| WebProjectRunnerFile {
|
||
path: file.path,
|
||
content: file.content,
|
||
})
|
||
.collect(),
|
||
artifact_root: state.config.web_project_artifact_root.clone(),
|
||
};
|
||
let input_json = serde_json::to_vec(&input).map_err(|error| error.to_string())?;
|
||
let mut command = Command::new(&runner_bin);
|
||
command
|
||
.stdin(Stdio::piped())
|
||
.stdout(Stdio::piped())
|
||
.stderr(Stdio::piped())
|
||
.kill_on_drop(true);
|
||
let mut child = command
|
||
.spawn()
|
||
.map_err(|error| format!("启动 web-project-runner 失败：{error}"))?;
|
||
if let Some(mut stdin) = child.stdin.take() {
|
||
stdin
|
||
.write_all(&input_json)
|
||
.await
|
||
.map_err(|error| format!("写入 runner 输入失败：{error}"))?;
|
||
}
|
||
let mut stdout = child
|
||
.stdout
|
||
.take()
|
||
.ok_or_else(|| "web-project-runner stdout 未打开".to_string())?;
|
||
let mut stderr = child
|
||
.stderr
|
||
.take()
|
||
.ok_or_else(|| "web-project-runner stderr 未打开".to_string())?;
|
||
let stdout_task = tokio::spawn(async move {
|
||
let mut bytes = Vec::new();
|
||
stdout
|
||
.read_to_end(&mut bytes)
|
||
.await
|
||
.map(|_| bytes)
|
||
.map_err(|error| error.to_string())
|
||
});
|
||
let stderr_task = tokio::spawn(async move {
|
||
let mut bytes = Vec::new();
|
||
stderr
|
||
.read_to_end(&mut bytes)
|
||
.await
|
||
.map(|_| bytes)
|
||
.map_err(|error| error.to_string())
|
||
});
|
||
let status = match tokio::time::timeout(WEB_PROJECT_RUNNER_TIMEOUT, child.wait()).await {
|
||
Ok(result) => result.map_err(|error| format!("等待 runner 退出失败：{error}"))?,
|
||
Err(_) => {
|
||
let _ = child.kill().await;
|
||
let _ = child.wait().await;
|
||
let _ = stdout_task.await;
|
||
let _ = stderr_task.await;
|
||
return Err("web-project-runner 执行超时".to_string());
|
||
}
|
||
};
|
||
let stdout = stdout_task
|
||
.await
|
||
.map_err(|error| format!("读取 runner stdout 任务失败：{error}"))?
|
||
.map_err(|error| format!("读取 runner stdout 失败：{error}"))?;
|
||
let stderr = stderr_task
|
||
.await
|
||
.map_err(|error| format!("读取 runner stderr 任务失败：{error}"))?
|
||
.map_err(|error| format!("读取 runner stderr 失败：{error}"))?;
|
||
if !status.success() {
|
||
let stderr = String::from_utf8_lossy(&stderr);
|
||
return Ok(WebProjectBuildOutput {
|
||
job_id: build.job_id.clone(),
|
||
project_id: build.project_id.clone(),
|
||
snapshot_id: build.snapshot_id.clone(),
|
||
status: WebProjectBuildStatus::Failed,
|
||
artifact_id: None,
|
||
artifact_path: None,
|
||
error_summary: Some(format!("web-project-runner 退出失败：{}", stderr.trim())),
|
||
logs: vec![stderr.to_string()],
|
||
});
|
||
}
|
||
serde_json::from_slice::<WebProjectBuildOutput>(&stdout)
|
||
.map_err(|error| format!("runner 输出不是合法 JSON：{error}"))
|
||
}
|
||
|
||
fn resolve_runner_binary(state: &AppState) -> Result<PathBuf, String> {
|
||
if let Some(path) = state.config.web_project_runner_bin.as_ref() {
|
||
return Ok(path.clone());
|
||
}
|
||
let current_exe = std::env::current_exe().map_err(|error| error.to_string())?;
|
||
let exe_name = if cfg!(windows) {
|
||
"web-project-runner.exe"
|
||
} else {
|
||
"web-project-runner"
|
||
};
|
||
Ok(current_exe
|
||
.parent()
|
||
.map(|parent| parent.join(exe_name))
|
||
.unwrap_or_else(|| PathBuf::from(exe_name)))
|
||
}
|
||
|
||
fn build_preview_token_id(issued_at_micros: i64) -> String {
|
||
format!("wpt_{}_{}", issued_at_micros, new_uuid_simple_string())
|
||
}
|
||
|
||
fn build_preview_url(base_url: &str, preview_token_id: &str) -> String {
|
||
format!("{}/p/{preview_token_id}/", base_url.trim_end_matches('/'))
|
||
}
|
||
|
||
fn current_utc_micros() -> i64 {
|
||
let duration = SystemTime::now()
|
||
.duration_since(UNIX_EPOCH)
|
||
.expect("system clock should be after unix epoch");
|
||
i64::try_from(duration.as_micros()).expect("current unix micros should fit in i64")
|
||
}
|