fmq/DCTS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(all): 任务引擎双阶段解耦、僵尸涡旋修复、动态 CPU 配额与前端详情页重构

Browse Source 
将 TLUSTY/SYNSPEC 拆为各自独立的 enabled/policy/strategies 阶段，
以策略链自动弹栈取代单级 seed_step 布尔回退；定向修复 2026-08-02
僵尸任务涡旋事故；新增节点并发配额热调；前端详情页从 1412 行巨型
视图拆为薄控制器 + detail 子模块，并补齐工具层与单测。

引擎与调度（task_engine_decoupling_design.md）
- models.rs: 新增 StagePolicy / EngineStageConfig / TaskSpec 阶段字段、
  normalize_compat() 校正旧版在途消息策略链、failed_stage 归因
- scheduler.rs: resolve_dispatchable_chain 派发门控、
  trigger_strategy_fallback 按 failed_stage 精确弹栈；启动期
  force_recompute/skip_converged(默认)/skip_failed 三策略
- db.rs: tasks 表 +7 列持久化阶段配置；终态守卫
  (mark_grid_point_running 仅 pending/queued→running；
  record_task_report 拒绝迟到失败翻黑 converged)；策略弹栈快照

僵尸涡旋修复（runbook-20260802-zombie-vortex-fix.md）
- 全链路跨库活性交叉校验：派发/claim/孤儿回收/回退统一查 MQ 队列活性，
  活则放行、死则清僵尸，结构性消除"每点重复派发"
- stop/重启卫生：清队列同步 delete_tasks_by_ids，杜绝遗留 pending 行
- report_task: 幂等吸收 + 409 区分迟到冗余结果，仅 state_changed 时回退
- MQ: NULL workflow_name 回填 __legacy__、requeue 后迟到上报被 403 竞态修复

动态 CPU 配额（dynamic_cpu_slots_design.md）
- admin.rs: POST /admin/nodes/:id/quota（Option<Option<i32>> 区分
  缺字段/显式 null）；nodes 表 +admin_max_slots
- worker.rs: effective_max_slots = min(admin, physical)，心跳下发原子生效

科学产物保全（tlusty_result_artifacts.md）
- runner.rs: SYNSPEC 启动前快照 fort.12/fort.14 → .bfac/.emflux 防覆盖
- 半失败点(大气收敛+光谱失败)改判 Failed 并写入 note；仅 SYNSPEC
  场景不再恒判失败；撤销归档 LRU 200 上限改为永久保留
- executor.rs: 透传 synspec_params 数值参数（此前固定 None）

前端（dashboard/）
- workflowDetail.js 1412→328 行，拆出 views/detail/{ctx,overview,
  pointsTable,parSets,pointPanel}.js，AbortController 治理监听/请求生命周期
- 删除 wfActions.js，新增 wfEnginePanel.js（双阶段三维配置编辑面板）
- 新增 utils/{errors,format,icons,polling,yamlStage}.js 纯函数模块
- 路由级动态 import 代码分割；节点配额三点菜单 + Modal 管理
- 首次引入 node:test 单测（format/polling/yamlStage/psCache，644 行）
- 系统性补齐 a11y：skip-link、ARIA、Tab 键盘漫游、toast 关闭、退出动画

文档与工具
- 新增 6 篇设计/调研：引擎解耦、动态配额、涡旋 runbook、
  光谱正确性分析、收敛判断、产物归档
- PIPELINE/design/api/database 等协同重写为分布式 C/S 架构口径
- scripts/fetch_results.sh 跨节点产物备份；import_results 按 cno 升序导入
- workflows/sdB_cno.yaml: 新增 tlusty/synspec_stage 配置块，修正 wstart 笔误
This commit is contained in:
fmq
2026-08-04 23:40:52 +08:00
parent c8fd24b120
commit cd370d88e7
74 changed files with 13386 additions and 3033 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/node/src/executor.rs
+153 -222
View File
@@ -1,7 +1,8 @@
use anyhow::Result;
use common::result_filter::is_result_worthy;
use common::config::SynspecConfig;
use common::embedded::{ensure_specific_data_files, RuntimePaths};
use common::models::{ModelSummary, TaskSpec, TaskType};
use common::models::{ModelSummary, TaskSpec};
use common::result_filter::is_result_worthy;
use common::runner::ExecutionRunner;
use reqwest::Client;
use std::path::{Path, PathBuf};
@@ -12,6 +13,7 @@ pub async fn execute_task(
server_url: &str,
runtime: &RuntimePaths,
work_dir: &Path,
result_dir: &Path,
task: &TaskSpec,
shutdown: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
) -> Result<(ModelSummary, Option<Vec<u8>>)> {
@@ -54,75 +56,112 @@ pub async fn execute_task(
let slot_work_dir = work_dir.join(format!("task_{}", task.task_id));
tokio::fs::create_dir_all(&slot_work_dir).await?;
// 2. If seed_step, download seed .7 file from server using atomic file rename.
// 调度入口已改为「冷启动优先」，SeedStep 仅作为冷启动失败后的救援任务出现，服务端
// 派发前已经 find_best_seed_from_db 确认种子存在于 DB。因此下载失败（缺种子名/HTTP
// 错误/网络异常/响应体读取失败）按硬错误处理，直接失败该任务（fail-fast），不再无种
// 子继续运行：default_seed_chain 首阶段 seed_nc 的 ltgray="F" 依赖 fort.8，无种子时
// Tlusty 在无初始大气下运行必然崩溃。任务失败后由服务端走既有上报路径，
// has_seed_step_attempt 阻止重复回退，网格点保持 failed 终态。
if task.task_type == TaskType::SeedStep {
let seed_name = task.seed_point_name.as_deref().ok_or_else(|| {
anyhow::anyhow!(
"SeedStep 任务 {} 缺少 seed_point_name，无法热启动",
task.point_name
)
})?;
let seed_url = format!("{}/api/seed/{}", server_url, seed_name);
info!("正在从服务端下载种子大气文件: {}", seed_url);
// 2. 种子大气获取（见 docs/task_engine_decoupling_design.md §5）：
// - TLUSTY 启用 + 策略为 seed_step：下载近邻种子 .7 作热启动种子（既有逻辑）。
// - TLUSTY 关闭（仅 SYNSPEC 场景）：需拉取目标点 .7 大气作光谱合成输入。
// 顺序：本地 result 归档 → server 拉取。
// 注：不查沙盒本地——TLUSTY 与 SYNSPEC 在同一任务内串行，种子获取先于 runner，
// 全新 slot 内不可能已有目标大气；重试任务的 slot 亦全新（task_id 唯一）。
let tlusty_enabled = task.tlusty_config.enabled;
let current_strategy = task.tlusty_config.current_strategy("cold_run");
let needs_seed_download = (tlusty_enabled && current_strategy == "seed_step")
|| (!tlusty_enabled && task.synspec_config.enabled);
let resp = client.get(&seed_url).send().await.map_err(|e| {
anyhow::anyhow!(
"SeedStep 任务 {} 下载种子文件 {} 失败: {}",
task.point_name,
seed_url,
e
)
})?;
if !resp.status().is_success() {
anyhow::bail!(
"SeedStep 任务 {} 下载种子文件 {} 失败: HTTP {}",
task.point_name,
seed_url,
resp.status()
);
if needs_seed_download {
let seed_name = if !tlusty_enabled {
// 仅 SYNSPEC 场景：大气来自目标点本身（atmosphere_ref 或 point_name）。
task.atmosphere_ref
.clone()
.unwrap_or_else(|| task.point_name.clone())
} else {
// SeedStep 热启动：大气来自近邻种子点。
task.seed_point_name.clone().ok_or_else(|| {
anyhow::anyhow!(
"SeedStep 任务 {} 缺少 seed_point_name，无法热启动",
task.point_name
)
})?
};
// 仅 SYNSPEC 场景先查本地 result 归档目录（节点此前算过同点大气，避免 server 拉取）。
if !tlusty_enabled {
let archived = result_dir
.join(&task.point_name)
.join(format!("{}.7", task.point_name));
if archived.is_file() {
info!(
"SYNSPEC-only：在本地 result 归档找到大气 {}，复用避免 server 拉取",
archived.display()
);
seed_atmos_path = Some(archived);
}
}
let bytes = resp.bytes().await.map_err(|e| {
anyhow::anyhow!(
"SeedStep 任务 {} 读取种子文件 {} 响应体失败: {}",
task.point_name,
seed_url,
e
)
})?;
let temp_seed_dir = work_dir.join(".seed_cache");
tokio::fs::create_dir_all(&temp_seed_dir).await?;
// LRU 上限清理：下载新种子前，删除最旧的超出 MAX_SEED_CACHE_FILES 的
// .seed.7 文件，防止长期运行后不同种子点累积到 GB 级。同名种子会被
// 覆盖写，真正累积的维度是「不同 seed_name」的数量。
cleanup_seed_cache(&temp_seed_dir).await;
let tmp_path = temp_seed_dir.join(format!(
"{}.{}.tmp",
seed_name,
uuid::Uuid::new_v4().simple()
));
let final_seed_path = temp_seed_dir.join(format!("{}.seed.7", seed_name));
tokio::fs::write(&tmp_path, bytes).await?;
tokio::fs::rename(&tmp_path, &final_seed_path).await?;
// 归档无 → 向 server 拉取。
if seed_atmos_path.is_none() {
let seed_url = format!("{}/api/seed/{}", server_url, seed_name);
info!(
"正在从服务端下载大气文件 ({}, 用途: {}): {}",
seed_name,
if tlusty_enabled {
"TLUSTY 热启动种子"
} else {
"SYNSPEC 输入大气"
},
seed_url
);
// 关键：复制一份种子到本任务沙盒私有副本，让 seed_atmos_path
// 指向私有副本而非共享缓存。此后 runner 的 current_seed 全程
// 只引用沙盒内文件，与 .seed_cache 完全解耦——这样 LRU 清理
// （含并发竞争）即便删掉该缓存文件，也不会破坏正在使用该种子
// 的 in-flight 任务。.seed_cache 退化为纯粹的下载去重缓存。
let private_seed = slot_work_dir.join("seed_atmos.seed.7");
tokio::fs::copy(&final_seed_path, &private_seed).await?;
seed_atmos_path = Some(private_seed);
let resp = client.get(&seed_url).send().await.map_err(|e| {
anyhow::anyhow!(
"任务 {} 下载大气文件 {} 失败: {}",
task.point_name,
seed_url,
e
)
})?;
if !resp.status().is_success() {
anyhow::bail!(
"任务 {} 下载大气文件 {} 失败: HTTP {}",
task.point_name,
seed_url,
resp.status()
);
}
let bytes = resp.bytes().await.map_err(|e| {
anyhow::anyhow!(
"任务 {} 读取大气文件 {} 响应体失败: {}",
task.point_name,
seed_url,
e
)
})?;
let temp_seed_dir = work_dir.join(".seed_cache");
tokio::fs::create_dir_all(&temp_seed_dir).await?;
cleanup_seed_cache(&temp_seed_dir).await;
let tmp_path = temp_seed_dir.join(format!(
"{}.{}.tmp",
seed_name,
uuid::Uuid::new_v4().simple()
));
let final_seed_path = temp_seed_dir.join(format!("{}.seed.7", seed_name));
tokio::fs::write(&tmp_path, bytes).await?;
tokio::fs::rename(&tmp_path, &final_seed_path).await?;
let private_seed = slot_work_dir.join("seed_atmos.seed.7");
tokio::fs::copy(&final_seed_path, &private_seed).await?;
seed_atmos_path = Some(private_seed);
}
}
// 3. （slot_work_dir 已在种子下载前提前创建，种子私有副本亦已落盘于沙盒内。）
// 反序列化工作流携带的 SYNSPEC 数值参数（波长范围等）。None → runner 用硬编码默认。
let synspec_cfg: Option<SynspecConfig> = task
.synspec_params
.as_ref()
.and_then(|v| serde_json::from_value::<SynspecConfig>(v.clone()).ok());
let runner = ExecutionRunner::new(runtime, slot_work_dir.clone());
let summary = runner
.run_model_with_timeout(
@@ -135,7 +174,10 @@ pub async fn execute_task(
// SeedStep→default_seed_chain），节点端不再做「缺种子回退冷启动链」的降级。
None,
seed_atmos_path.as_deref(),
None,
synspec_cfg.as_ref(),
// 阶段独立配置开关（见 docs/task_engine_decoupling_design.md §5）。
task.tlusty_config.enabled,
task.synspec_config.enabled,
task.timeout_sec,
shutdown,
)
@@ -199,7 +241,8 @@ pub async fn cleanup_slot_work_dir(slot_work_dir: &Path) -> Result<()> {
///
/// 保留内容（详见 [`is_result_worthy`]）：
/// - 裸名：`conv.json`、`fort.8`（synspec 输入大气）、`fort.55`（synspec 控制卡）
/// - 科学核心：`<name>.7/.spec/.cont/.iden/.log`
/// - 科学核心：`<name>.7/.spec/.cont/.iden/.log`，以及 runner 在 SYNSPEC 覆盖前快照的
/// TLUSTY 最终产物：`<name>.bfac`（b 因子/非 LTE 偏离因子）、`<name>.emflux`（出射谱 λ–Fλ）
/// - 阶段快照：`<name>.<label>.5/.6/.err/.nst/.7`
/// - 收敛诊断：`<name>.<label>_chmax*.9`（**唯一保留的 .9**）
///
@@ -211,11 +254,7 @@ pub async fn cleanup_slot_work_dir(slot_work_dir: &Path) -> Result<()> {
/// `name` 为网格点权威名：取自 summary.name（runner 现用 task.point_name 作权威名），
/// 严重失败（runner 抛 Err、无 summary）时回退到 task.point_name，确保失败任务的
/// 排错日志也能落盘。
pub async fn save_result_artifacts(
result_dir: &Path,
slot_work_dir: &Path,
name: &str,
) {
pub async fn save_result_artifacts(result_dir: &Path, slot_work_dir: &Path, name: &str) {
let src_dir = slot_work_dir.join(name);
if !src_dir.is_dir() {
// 模型子目录不存在（极早期失败），无可归档内容
@@ -276,7 +315,11 @@ pub async fn save_result_artifacts(
let dest_path = dest_dir.join(&file_name);
// 原子写入：先拷到 .result.tmp.<uuid> 再 rename，防止中途崩溃产生半截文件
let tmp_path = dest_dir.join(format!("{}.result.tmp.{}", file_name, uuid::Uuid::new_v4().simple()));
let tmp_path = dest_dir.join(format!(
"{}.result.tmp.{}",
file_name,
uuid::Uuid::new_v4().simple()
));
match tokio::fs::copy(&path, &tmp_path).await {
Ok(_) => {
if let Err(e) = tokio::fs::rename(&tmp_path, &dest_path).await {
@@ -284,9 +327,7 @@ pub async fn save_result_artifacts(
let _ = tokio::fs::remove_file(&tmp_path).await;
warn!(
"归档网格点 {} 的文件 {} rename 失败: {}",
name,
file_name,
e
name, file_name, e
);
continue;
}
@@ -294,12 +335,7 @@ pub async fn save_result_artifacts(
}
Err(e) => {
let _ = tokio::fs::remove_file(&tmp_path).await;
warn!(
"归档网格点 {} 的文件 {} 拷贝失败: {}",
name,
file_name,
e
);
warn!("归档网格点 {} 的文件 {} 拷贝失败: {}", name, file_name, e);
}
}
}
@@ -314,52 +350,9 @@ pub async fn save_result_artifacts(
);
}
/// 归档目录保留的网格点（子目录）数量上限。超过则按 mtime 删除最旧的。
/// 200 足以覆盖中等规模网格的完整归档；更大网格可经环境变量或常量调整。
const MAX_RESULT_MODELS: usize = 200;
/// LRU 治理归档目录：当网格点子目录数超过 `MAX_RESULT_MODELS` 时，
/// 按 mtime 升序删除最旧的若干个子目录，直到不超过上限。
/// 仅统计子目录（每个对应一个网格点），忽略散落文件。错误降级为 warn，不阻断主流程。
pub async fn cleanup_result_dir(result_dir: &Path) {
let mut entries: Vec<(std::time::SystemTime, PathBuf)> =
match tokio::fs::read_dir(result_dir).await {
Ok(mut rd) => {
let mut v = Vec::new();
while let Ok(Some(entry)) = rd.next_entry().await {
let path = entry.path();
// 仅纳入子目录（网格点归档目录），跳过散落文件
if !path.is_dir() {
continue;
}
let meta = match entry.metadata().await {
Ok(m) => m,
Err(_) => continue,
};
let mtime = meta.modified().unwrap_or(std::time::SystemTime::UNIX_EPOCH);
v.push((mtime, path));
}
v
}
// 归档目录不存在或不可读：无操作（首次归档尚未创建）
Err(_) => return,
};
if entries.len() <= MAX_RESULT_MODELS {
return;
}
// 按 mtime 升序（最旧在前），删除超出上限的最旧子目录
entries.sort_by_key(|(mtime, _)| *mtime);
let to_remove = entries.len().saturating_sub(MAX_RESULT_MODELS);
for (_, path) in entries.into_iter().take(to_remove) {
if let Err(e) = tokio::fs::remove_dir_all(&path).await {
warn!("LRU 清理归档目录 {} 失败: {}", path.display(), e);
} else {
info!("LRU 清理归档目录: {}", path.display());
}
}
}
/// 归档目录不做数量上限治理：所有已算网格点的完整产物（.spec/.cont/.iden/各阶段
/// 快照/日志/种子二进制等）一律永久保留，避免 LRU 淘汰导致科学产物丢失
/// （2026-08-02 修正：撤销 1bfa240 引入的 MAX_RESULT_MODELS=200 LRU 上限）。
/// `.seed_cache/` 内保留的 `.seed.7` 文件上限。超过则按 mtime 删除最旧的。
/// 典型网格内活跃种子点数量有限，8 足以覆盖常用邻域且把磁盘占用控制在 ~8 个种子文件。
@@ -475,40 +468,42 @@ mod tests {
// 应被归档的白名单产物（科学核心 + 阶段快照 + 收敛诊断 + 裸名保留）
let kept_files = [
format!("{}.7", summary.name), // 最终大气
format!("{}.spec", summary.name), // 合成光谱
format!("{}.cont", summary.name), // 连续谱
format!("{}.iden", summary.name), // 谱线证认
format!("{}.log", summary.name), // synspec 日志
"conv.json".to_string(), // 摘要
"fort.8".to_string(), // synspec 输入大气（裸名保留）
"fort.55".to_string(), // synspec 控制卡（裸名保留）
format!("{}.nl.7", summary.name), // nl 阶段大气快照
format!("{}.nc.7", summary.name), // nc 阶段大气快照
format!("{}.nl.5", summary.name), // nl 阶段输入卡快照
format!("{}.nl.6", summary.name), // nl 阶段输出日志快照
format!("{}.nl.err", summary.name), // nl 阶段错误日志快照
format!("{}.nl.nst", summary.name), // nl 阶段控制卡快照
format!("{}.nc.nst", summary.name), // nc 阶段控制卡快照
format!("{}.7", summary.name), // 最终大气
format!("{}.spec", summary.name), // 合成光谱
format!("{}.cont", summary.name), // 连续谱
format!("{}.iden", summary.name), // 谱线证认
format!("{}.log", summary.name), // synspec 日志
format!("{}.bfac", summary.name), // TLUSTY 最终 b 因子（快照 fort.12）
format!("{}.emflux", summary.name), // TLUSTY 最终出射谱（快照 fort.14）
"conv.json".to_string(), // 摘要
"fort.8".to_string(), // synspec 输入大气（裸名保留）
"fort.55".to_string(), // synspec 控制卡（裸名保留）
format!("{}.nl.7", summary.name), // nl 阶段大气快照
format!("{}.nc.7", summary.name), // nc 阶段大气快照
format!("{}.nl.5", summary.name), // nl 阶段输入卡快照
format!("{}.nl.6", summary.name), // nl 阶段输出日志快照
format!("{}.nl.err", summary.name), // nl 阶段错误日志快照
format!("{}.nl.nst", summary.name), // nl 阶段控制卡快照
format!("{}.nc.nst", summary.name), // nc 阶段控制卡快照
format!("{}.nl_chmax0.001.9", summary.name), // nl 收敛诊断（唯一保留的 .9）
];
// 应被白名单过滤掉的文件：Tlusty 中间单元、裸的 runner 已清理文件、
// 无 _chmax 的重复 .9 快照、未知后缀
let skipped_files: [String; 14] = [
"fort.1".to_string(), // 空单元
"fort.13".to_string(), // Tlusty NLTE 跃迁频率网格
"fort.18".to_string(), // Tlusty 大气结构内部表
"fort.22".to_string(), // Tlusty 中间大气副本
"fort.82".to_string(), // Tlusty 运行时诊断表
"fort.95".to_string(), // Tlusty 旧模型定义副本
"fort.84".to_string(), // NATOMS 崩溃缓存
"residue.tmp".to_string(), // 原子写入残留
"nst".to_string(), // 裸 nst（runner 已改名为 <name>.<label>.nst）
"fort.9".to_string(), // 裸 fort.9（runner 已删，内容在 _chmax.9）
"fort.12".to_string(), // 裸 fort.12（已 copy 为 .iden）
"fort.17".to_string(), // 裸 fort.17（已 copy 为 .cont）
format!("{}.nl.9", summary.name), // 无 _chmax 的 .9 快照（与 _chmax.9 重复）
format!("{}.unknown", summary.name), // 未知后缀
"fort.1".to_string(), // 空单元
"fort.13".to_string(), // Tlusty 出射辐射场 (FREQ/FLUX/FH)，未快照，丢弃
"fort.18".to_string(), // Tlusty 大气结构内部表
"fort.22".to_string(), // Tlusty 中间大气副本
"fort.82".to_string(), // Tlusty 运行时诊断表
"fort.95".to_string(), // Tlusty 旧模型定义副本
"fort.84".to_string(), // NATOMS 崩溃缓存
"residue.tmp".to_string(), // 原子写入残留
"nst".to_string(), // 裸 nst（runner 已改名为 <name>.<label>.nst）
"fort.9".to_string(), // 裸 fort.9（runner 已删，内容在 _chmax.9）
"fort.12".to_string(), // 裸 fort.12（已 copy 为 .iden）
"fort.17".to_string(), // 裸 fort.17（已 copy 为 .cont）
format!("{}.nl.9", summary.name), // 无 _chmax 的 .9 快照（与 _chmax.9 重复）
format!("{}.unknown", summary.name), // 未知后缀
];
for f in kept_files.iter() {
tokio::fs::write(model_dir.join(f), "payload")
@@ -535,19 +530,11 @@ mod tests {
assert!(dest_dir.is_dir(), "归档目标目录应被创建");
// 验证白名单产物都被拷贝
for f in &kept_files {
assert!(
dest_dir.join(f).is_file(),
"白名单产物 {} 应被归档",
f
);
assert!(dest_dir.join(f).is_file(), "白名单产物 {} 应被归档", f);
}
// 验证非白名单文件未进归档
for f in &skipped_files {
assert!(
!dest_dir.join(f).exists(),
"非白名单文件 {} 应被跳过",
f
);
assert!(!dest_dir.join(f).exists(), "非白名单文件 {} 应被跳过", f);
}
#[cfg(unix)]
{
@@ -561,65 +548,9 @@ mod tests {
let mut rd = tokio::fs::read_dir(&dest_dir).await.unwrap();
while let Ok(Some(e)) = rd.next_entry().await {
let name = e.file_name().to_string_lossy().to_string();
assert!(
!name.contains(".result.tmp"),
"不应残留 tmp 文件: {}",
name
);
assert!(!name.contains(".result.tmp"), "不应残留 tmp 文件: {}", name);
}
let _ = tokio::fs::remove_dir_all(&root).await;
}
/// 验证 LRU 治理：超过上限时按 mtime 删最旧的子目录
#[tokio::test]
async fn test_cleanup_result_dir() {
let result_dir =
std::env::temp_dir().join(format!("test_result_lru_{}", uuid::Uuid::new_v4()));
tokio::fs::create_dir_all(&result_dir).await.unwrap();
// 创建 MAX+10 个子目录，按创建顺序递增 mtime（每个 sleep 制造可测的时间差）。
// model_0000 最早创建（最旧），model_0209 最新创建。
let total = MAX_RESULT_MODELS + 10;
for i in 0..total {
let dir = result_dir.join(format!("model_{:04}", i));
tokio::fs::create_dir_all(&dir).await.unwrap();
tokio::fs::write(dir.join("marker"), format!("{}", i))
.await
.unwrap();
// 10ms 间隔足以让多数文件系统的 mtime 分辨出先后顺序
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
cleanup_result_dir(&result_dir).await;
let mut remaining: Vec<String> = Vec::new();
let mut rd = tokio::fs::read_dir(&result_dir).await.unwrap();
while let Ok(Some(e)) = rd.next_entry().await {
if e.path().is_dir() {
remaining.push(e.file_name().to_string_lossy().to_string());
}
}
// 清理后剩余数量应恰为上限
assert_eq!(
remaining.len(),
MAX_RESULT_MODELS,
"清理后应剩余 {} 个，实际 {} 个",
MAX_RESULT_MODELS,
remaining.len()
);
// 最旧的那批（model_0000~model_0009）应被删除，最新的 MAX 个应保留
remaining.sort();
assert!(
!remaining.contains(&"model_0000".to_string()),
"最旧的 model_0000 应被 LRU 删除"
);
assert!(
remaining.contains(&format!("model_{:04}", total - 1)),
"最新的 model_{:04} 应被保留",
total - 1
);
let _ = tokio::fs::remove_dir_all(&result_dir).await;
}
}
crates/node/src/reporter.rs
+211 -27
View File
@@ -4,6 +4,46 @@ use reqwest::multipart::{Form, Part};
use reqwest::Client;
use tracing::{info, warn};
/// 失败阶段归因（见 docs/task_engine_decoupling_design.md §4.2）：从 summary + 任务阶段开关
/// 推断本次失败发生在 TLUSTY 还是 SYNSPEC，供服务端按对应策略链弹栈回退。
///
/// 判定顺序（顺序敏感）：
/// - `converged=false` 时：大气未收敛。若 TLUSTY 启用 → 归因 TLUSTY（含"TLUSTY 产出大气
/// 失败导致 synspec 无输入"的级联场景）；TLUSTY 关闭 → 归因 SYNSPEC（synspec-only 任务
/// 的 converged 由 synspec_rc 决定）。
/// - `converged=true` 但 synspec 有错误/非零 rc：归因 SYNSPEC（大气已收敛、光谱失败）。
fn infer_failed_stage(task: &TaskSpec, summary: &ModelSummary) -> Option<String> {
if !summary.converged {
return if task.tlusty_config.enabled {
Some("tlusty".to_string())
} else {
Some("synspec".to_string())
};
}
if summary.synspec_error.is_some() || matches!(summary.synspec_rc, Some(rc) if rc != 0) {
return Some("synspec".to_string());
}
None
}
/// 任务整体成败判定（修复审查 #2 后续）。
///
/// 旧实现 `if s.converged { Completed } else { Failed }` 只看大气收敛：当 TLUSTY 收敛但
/// SYNSPEC 失败（rc≠0 / synspec_error）时，`converged` 仍为 true（它是大气收敛标志），
/// 任务被误报为 Completed → 服务端 `record_task_report` 按 converged 吸收为终态：
/// 半失败点（大气好、光谱坏）既不回退重试，错误也无从查证。
///
/// 新判定：**半失败 = 失败**。`converged` 仍保留 true 随上报（大气产物有效，服务端仍会
/// 保存 .7 作种子），但 status=Failed 让服务端弹 synspec 链重试，且错误对运维可见。
fn derive_report_status(s: &ModelSummary) -> TaskStatus {
let synspec_failed = s.synspec_error.is_some() || matches!(s.synspec_rc, Some(rc) if rc != 0);
if s.converged && !synspec_failed {
TaskStatus::Completed
} else {
TaskStatus::Failed
}
}
pub async fn report_result(
client: &Client,
server_url: &str,
@@ -13,33 +53,40 @@ pub async fn report_result(
) -> Result<()> {
let report_url = format!("{}/api/task/report", server_url);
let (status, converged, max_relc, atmo_has_nan, elapsed_sec, err_msg, summary_json, seed_bytes) =
match exec_res {
Ok((s, s_bytes)) => (
if s.converged {
TaskStatus::Completed
} else {
TaskStatus::Failed
},
s.converged,
s.final_max_relc,
s.atmosphere_has_nan,
s.elapsed_sec,
s.note.clone(),
serde_json::to_string(&s).unwrap_or_default(),
s_bytes,
),
Err(e) => (
TaskStatus::Failed,
false,
None,
false,
0.0,
Some(e.clone()),
serde_json::json!({"error": e}).to_string(),
None,
),
};
let (
status,
converged,
max_relc,
atmo_has_nan,
elapsed_sec,
err_msg,
summary_json,
seed_bytes,
failed_stage,
) = match exec_res {
Ok((s, s_bytes)) => (
derive_report_status(&s),
s.converged,
s.final_max_relc,
s.atmosphere_has_nan,
s.elapsed_sec,
s.note.clone(),
serde_json::to_string(&s).unwrap_or_default(),
s_bytes,
infer_failed_stage(task, &s),
),
Err(e) => (
TaskStatus::Failed,
false,
None,
false,
0.0,
Some(e.clone()),
serde_json::json!({"error": e}).to_string(),
None,
None,
),
};
let report = TaskReport {
task_id: task.task_id,
@@ -53,6 +100,7 @@ pub async fn report_result(
elapsed_sec,
error_message: err_msg,
summary_json,
failed_stage,
};
let report_bytes = serde_json::to_vec(&report)?;
@@ -93,6 +141,17 @@ pub async fn report_result(
);
anyhow::bail!("node token 失效 (HTTP {})，结果未上报", status);
}
// M3 修复：409 = 任务已被其他节点重新领用（超时重投后重领）或已由他节点结算。
// 这是迟到的冗余结果，服务端据此弃掉本次上报——**不是凭据问题**，切勿误报
// "token 失效"误导运维换 token；也无需重试（重试必再 409）。产物仍会归档，
// 返回 Ok 让 worker 正常收尾（归档 + 清理沙盒 + 归还 slot）。
if status.as_u16() == 409 {
info!(
"任务 {} 上报被服务端拒绝 (HTTP 409)：任务已被其他节点重新领用或已结算，本次结果丢弃（不重试，产物仍归档）",
task.task_id
);
return Ok(());
}
warn!(
"向服务端上报任务 {} 结果失败 (尝试 {}/{}): HTTP {}",
task.task_id, attempt, max_attempts, status
@@ -119,3 +178,128 @@ pub async fn report_result(
task.task_id
)
}
#[cfg(test)]
mod tests {
use super::*;
use common::models::GridPointParams;
fn mk_task(tlusty_enabled: bool) -> TaskSpec {
TaskSpec {
task_id: uuid::Uuid::new_v4(),
point_name: "t20000_g5.0_he-2_c-4_n-4_o-4".to_string(),
params: GridPointParams {
teff: 20000.0.into(),
logg: 5.0.into(),
loghe: (-2.0).into(),
logc: (-4.0).into(),
logn: (-4.0).into(),
logo: (-4.0).into(),
},
tlusty_config: common::models::EngineStageConfig {
enabled: tlusty_enabled,
..common::models::EngineStageConfig::default_tlusty()
},
..TaskSpec::default()
}
}
fn mk_summary(
converged: bool,
synspec_rc: Option<i32>,
synspec_error: Option<&str>,
) -> ModelSummary {
ModelSummary {
name: "t20000_g5.0_he-2_c-4_n-4_o-4".to_string(),
params: GridPointParams {
teff: 20000.0.into(),
logg: 5.0.into(),
loghe: (-2.0).into(),
logc: (-4.0).into(),
logn: (-4.0).into(),
logo: (-4.0).into(),
},
stages: vec![],
converged,
final_max_relc: None,
final_chmax: None,
seed: None,
atmosphere_has_nan: false,
synspec_rc,
synspec_error: synspec_error.map(|s| s.to_string()),
synspec_sec: None,
elapsed_sec: 10.0,
note: None,
}
}
/// 失败阶段归因（docs/task_engine_decoupling_design.md §4.2 语义）：
/// 大气未收敛 → TLUSTY（tlusty 启用时）；tlusty 关闭 → SYNSPEC；
/// 大气已收敛但 synspec 失败 → SYNSPEC；全成功 → None。
#[test]
fn test_infer_failed_stage() {
// tlusty 启用 + 大气未收敛（synspec 无输入级联）→ tlusty
assert_eq!(
infer_failed_stage(
&mk_task(true),
&mk_summary(false, None, Some("No atmosphere .7 produced"))
),
Some("tlusty".to_string())
);
// tlusty 关闭 + synspec 失败（converged=false）→ synspec
assert_eq!(
infer_failed_stage(&mk_task(false), &mk_summary(false, Some(1), None)),
Some("synspec".to_string())
);
// tlusty 启用 + 大气收敛但 synspec rc≠0 → synspec
assert_eq!(
infer_failed_stage(&mk_task(true), &mk_summary(true, Some(2), None)),
Some("synspec".to_string())
);
// tlusty 启用 + 大气收敛 + synspec_error（如 fort.8 写入失败）→ synspec
assert_eq!(
infer_failed_stage(
&mk_task(true),
&mk_summary(true, None, Some("fort.8 copy failed"))
),
Some("synspec".to_string())
);
// 全成功 → None
assert_eq!(
infer_failed_stage(&mk_task(true), &mk_summary(true, Some(0), None)),
None
);
}
/// 半失败判定（修复审查 #2 后续）：
/// 大气收敛 + 光谱失败 → 任务整体 Failed（否则服务端吸收为 converged 终态，无重试）；
/// 大气收敛 + 光谱成功 → Completed；大气未收敛 → Failed。
#[test]
fn test_derive_report_status_semi_failed() {
// 大气收敛 + synspec rc≠0 → Failed（半失败点不再被误报成功）
assert_eq!(
derive_report_status(&mk_summary(true, Some(2), None)),
TaskStatus::Failed
);
// 大气收敛 + synspec_error（如 fort.8 写入失败）→ Failed
assert_eq!(
derive_report_status(&mk_summary(true, None, Some("fort.8 copy failed"))),
TaskStatus::Failed
);
// 大气收敛 + synspec 成功（rc=0）→ Completed
assert_eq!(
derive_report_status(&mk_summary(true, Some(0), None)),
TaskStatus::Completed
);
// TLUSTY-only 模式（synspec 关闭，无 rc 无 error）+ 大气收敛 → Completed
assert_eq!(
derive_report_status(&mk_summary(true, None, None)),
TaskStatus::Completed
);
// 大气未收敛（无论光谱状态）→ Failed
assert_eq!(
derive_report_status(&mk_summary(false, Some(0), None)),
TaskStatus::Failed
);
}
}
crates/node/src/worker.rs
+62 -13
View File
@@ -3,11 +3,11 @@ use crate::reporter::report_result;
use anyhow::Result;
use common::config::NodeConfig;
use common::embedded::RuntimePaths;
use common::models::{NodeHeartbeatRequest, NodeRegisterRequest, TaskSpec};
use common::models::{NodeHeartbeatRequest, NodeHeartbeatResponse, NodeRegisterRequest, TaskSpec};
use reqwest::Client;
use serde_json::Value;
use std::path::PathBuf;
use std::sync::atomic::{AtomicI32, Ordering};
use std::sync::atomic::{AtomicI32, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::time::{sleep, Duration};
use tracing::{info, warn};
@@ -56,15 +56,23 @@ pub struct NodeWorker {
client: Client,
runtime: RuntimePaths,
active_slots: Arc<AtomicI32>,
/// 生效并发槽位上限：心跳线程据此写入，领用主循环据此读取。
///
/// 取自 `min(admin_max_slots, physical_max_slots)`；管理员未设配额时等于物理上限
/// （见 docs/dynamic_cpu_slots_design.md §3.2）。用 `Arc<AtomicUsize>` 在心跳线程与
/// 领用线程间共享，无需 Mutex：单写者（心跳线程）单读者（领用线程），原子读写即可。
effective_max_slots: Arc<AtomicUsize>,
}
impl NodeWorker {
pub fn new(config: NodeConfig, runtime: RuntimePaths, client: Client) -> Self {
let physical = config.max_slots;
Self {
config,
client,
runtime,
active_slots: Arc::new(AtomicI32::new(0)),
effective_max_slots: Arc::new(AtomicUsize::new(physical)),
}
}
@@ -206,6 +214,8 @@ impl NodeWorker {
let hb_url = format!("{}/api/node/heartbeat", self.config.server_url);
let hb_node_id = self.config.node_id.clone();
let hb_slots = self.active_slots.clone();
let hb_effective = self.effective_max_slots.clone();
let hb_physical = self.config.max_slots;
let hb_interval = self.config.heartbeat_sec;
let hb_runtime_dir = self.config.runtime_dir.clone();
@@ -280,6 +290,34 @@ impl NodeWorker {
);
std::process::exit(1);
}
// 解析心跳响应体，提取管理员配额（见 docs/dynamic_cpu_slots_design.md
// §3.2）：计算生效配额并更新 effective_max_slots。反序列化失败不致命
// ——旧服务端响应体可能不含 admin_max_slots 字段，保持现有配额不变即可；
// 但真正解析失败（服务端改了响应形状/网关篡改 body）须留日志，
// 否则运维无法定位「配额为何没生效」（审查修复 N1）。
if status.is_success() {
match resp.json::<NodeHeartbeatResponse>().await {
Ok(hb_resp) => {
let target = match hb_resp.admin_max_slots {
Some(admin_limit) => {
std::cmp::min(admin_limit.max(0) as usize, hb_physical)
}
None => hb_physical,
};
let prev = hb_effective.swap(target, Ordering::AcqRel);
if prev != target {
info!(
"节点 {} 并发槽位配额已调整: {} → {}（admin_max_slots={:?}, physical={}）",
hb_node_id, prev, target, hb_resp.admin_max_slots, hb_physical
);
}
}
Err(e) => warn!(
"节点 {} 心跳响应解析失败 (HTTP {})，保持当前配额: {}",
hb_node_id, status, e
),
}
}
}
Err(e) => warn!("节点 {} 心跳上报失败: {}", hb_node_id, e),
}
@@ -375,7 +413,7 @@ impl NodeWorker {
}
let active = self.active_slots.load(Ordering::Acquire);
if (active as usize) < self.config.max_slots {
if (active as usize) < self.effective_max_slots.load(Ordering::Acquire) {
match self.claim_task().await {
Ok(ClaimOutcome::Task(task)) => {
if was_disconnected {
@@ -397,39 +435,50 @@ impl NodeWorker {
slots_counter.fetch_add(1, Ordering::AcqRel);
// RAII 守卫仅负责 drop 时 -1（含 panic 展栈），杜绝 spawned future
// panic 导致的活动 slot 永久泄漏。
let slot_guard = SlotGuard { counter: slots_counter.clone() };
let slot_guard = SlotGuard {
counter: slots_counter.clone(),
};
tokio::spawn(async move {
// 把守卫移入 future，确保任务结束（含 panic）时归还 slot。
let _slot_guard = slot_guard;
let slot_work_dir = work_dir.join(format!("task_{}", task.task_id));
let res =
execute_task(&client, &server_url, &runtime, &work_dir, &task, Some(shutdown.clone()))
.await
.map_err(|e| e.to_string());
let res = execute_task(
&client,
&server_url,
&runtime,
&work_dir,
&result_dir,
&task,
Some(shutdown.clone()),
)
.await
.map_err(|e| e.to_string());
// 在上报前克隆 summary，供上报成功后的归档使用（归档目录名取
// summary.name，与 executor 内部 model_sub_dir 路径口径一致）。
// 严重失败（runner 抛 Err）时 summary_opt=None，此时回退到 task.point_name
// （二者均源自 params.model_name()）作为归档目录名，确保失败任务的
// 排错日志也能落盘而非随沙盒删除丢失。
let summary_opt =
res.as_ref().ok().map(|(s, _)| s.clone());
let result_name =
summary_opt.as_ref().map(|s| s.name.clone()).unwrap_or_else(|| task.point_name.clone());
let summary_opt = res.as_ref().ok().map(|(s, _)| s.clone());
let result_name = summary_opt
.as_ref()
.map(|s| s.name.clone())
.unwrap_or_else(|| task.point_name.clone());
let report_res =
report_result(&client, &server_url, &node_id, &task, res).await;
if report_res.is_ok() {
// 上报成功后、清理沙盒前，先把完整产物归档到持久目录，
// 避免随沙盒删除丢失（.spec/.cont/.iden/各阶段快照/日志等）。
// 归档不做 LRU 淘汰：所有已算网格点产物一律永久保留
// （2026-08-02 修正：撤销 1bfa240 引入的 200 目录上限）。
crate::executor::save_result_artifacts(
&result_dir,
&slot_work_dir,
&result_name,
)
.await;
crate::executor::cleanup_result_dir(&result_dir).await;
if let Err(e) =
crate::executor::cleanup_slot_work_dir(&slot_work_dir).await
{