feat(all): 数据库模块化拆分与版本化迁移、任务引擎命名体系收敛、物理输出校验加固与用户配置接通
- server/db: 拆 4929 行 db.rs 单体为 db/ 目录，migrations.rs 引入 PRAGMA user_version
版本化迁移运行器（M1~M13）
- 任务引擎 Phase 6/7b/7c 改名收敛：EngineStageConfig→PhaseConfig、StagePolicy→ResumePolicy、
Converged→Completed、删除 task_type 列、success_method 拆 tlusty_/synspec_ 双列、
新增 tlusty_status/synspec_status 半失败阶段守卫
- 科学正确性加固：conv_check 任意行 NaN/Inf/溢出判无效（0 行容忍）、新增 spec_is_valid
校验 SYNSPEC 脏谱、itek_history 逐次迭代全量保真、fmt_abn powf 溢出饱和
- 用户配置真正接通：tlusty_chain/tlusty_input 由死字段经 调度器→TaskSpec→executor→runner
透传生效；config 加载期 validate + deny_unknown_fields + 解析失败记 warn
- 调度修复：H1 活锁（pending_strategies 跳过已失败策略）、种子查找错误不再静默降级冷启动
- dashboard: 阶段配置面板 tlusty_stage/synspec_stage、"已完成"标签、迭代诊断展示
- docs: 新增 database_refactor_design.md，同步 database/api/PIPELINE/workflow_detail
This commit is contained in:
+132
-20
@@ -1,5 +1,5 @@
|
||||
use anyhow::Result;
|
||||
use common::config::SynspecConfig;
|
||||
use common::config::{ChainStep, SynspecInput, TlustyInput};
|
||||
use common::embedded::{ensure_specific_data_files, RuntimePaths};
|
||||
use common::models::{ModelSummary, TaskSpec};
|
||||
use common::result_filter::is_result_worthy;
|
||||
@@ -157,22 +157,49 @@ pub async fn execute_task(
|
||||
// 3. （slot_work_dir 已在种子下载前提前创建，种子私有副本亦已落盘于沙盒内。）
|
||||
|
||||
// 反序列化工作流携带的 SYNSPEC 数值参数（波长范围等）。None → runner 用硬编码默认。
|
||||
let synspec_cfg: Option<SynspecConfig> = task
|
||||
let synspec_cfg: Option<SynspecInput> = task
|
||||
.synspec_params
|
||||
.as_ref()
|
||||
.and_then(|v| serde_json::from_value::<SynspecConfig>(v.clone()).ok());
|
||||
.and_then(|v| serde_json::from_value::<SynspecInput>(v.clone()).ok());
|
||||
|
||||
let runner = ExecutionRunner::new(runtime, slot_work_dir.clone());
|
||||
// 执行链来源（优先级）：
|
||||
// 1. TaskSpec.tlusty_chain_params（用户在 YAML `tlusty_chain:` 配置的多阶段 ChainStep
|
||||
// 数组，由 scheduler 序列化注入）——非空时优先使用，使用户能细粒度控制 niter/chmax/
|
||||
// metals 等阶段参数。
|
||||
// 2. default_chain_for_strategy(current_strategy) 兜底——按策略名（cold_run/seed_step）
|
||||
// 选预设默认链（runner.rs 的 default_cold_chain / default_seed_chain）。
|
||||
// 历史：Phase 6 起仅用 default 链（用户 config.chain 被忽略，是死字段）；本次接通后
|
||||
// 用户配置真正生效，default 链降级为兜底。旧 MQ payload（无 tlusty_chain_params 字段）
|
||||
// 反序列化为 None → 回退 default 链，行为与旧版完全一致（向后兼容）。
|
||||
let chain = resolve_execution_chain(
|
||||
current_strategy,
|
||||
&task.tlusty_chain_params,
|
||||
&task.task_id.to_string(),
|
||||
);
|
||||
// TLUSTY 输入文件全局参数（NFREAD/ions 表/nst extra_keys 等）。
|
||||
// None → runner 用代码内硬编码默认（向后兼容）。
|
||||
let tlusty_input =
|
||||
task.tlusty_input_params.as_ref().and_then(|v| {
|
||||
match serde_json::from_value::<TlustyInput>(v.clone()) {
|
||||
Ok(t) => Some(t),
|
||||
Err(e) => {
|
||||
warn!(
|
||||
"任务 {} 的 tlusty_input_params 反序列化失败，回退默认输入: {}",
|
||||
task.task_id, e
|
||||
);
|
||||
None
|
||||
}
|
||||
}
|
||||
});
|
||||
let summary = runner
|
||||
.run_model_with_timeout(
|
||||
&task.params,
|
||||
// 用权威的 point_name（DB grid_points.name 列，源精度正确）作为模型名，
|
||||
// 而非 task.params.model_name()（后者经 DB REAL 列回读已丢精度 "5.0"→"5"）。
|
||||
&task.point_name,
|
||||
task.task_type.clone(),
|
||||
// custom_chain 恒为 None：执行链由 task_type 决定（ColdRun→default_cold_chain、
|
||||
// SeedStep→default_seed_chain），节点端不再做「缺种子回退冷启动链」的降级。
|
||||
None,
|
||||
current_strategy,
|
||||
Some(chain),
|
||||
seed_atmos_path.as_deref(),
|
||||
synspec_cfg.as_ref(),
|
||||
// 阶段独立配置开关（见 docs/task_engine_decoupling_design.md §5）。
|
||||
@@ -180,17 +207,18 @@ pub async fn execute_task(
|
||||
task.synspec_config.enabled,
|
||||
task.timeout_sec,
|
||||
shutdown,
|
||||
tlusty_input.as_ref(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
info!(
|
||||
"完成计算任务 {} (网格点: {}, 收敛状态: {})",
|
||||
task.task_id, task.point_name, summary.converged
|
||||
"完成计算任务 {} (网格点: {}, 结果可用: {})",
|
||||
task.task_id, task.point_name, summary.result_valid
|
||||
);
|
||||
|
||||
// Read seed bytes if converged and clean
|
||||
// Read seed bytes if result usable and clean
|
||||
let mut seed_bytes: Option<Vec<u8>> = None;
|
||||
if summary.converged && !summary.atmosphere_has_nan {
|
||||
if summary.result_valid && !summary.atmosphere_has_nan {
|
||||
let model_sub_dir = slot_work_dir.join(&summary.name);
|
||||
let candidates = [
|
||||
model_sub_dir.join(format!("{}.7", summary.name)),
|
||||
@@ -350,15 +378,23 @@ pub async fn save_result_artifacts(result_dir: &Path, slot_work_dir: &Path, name
|
||||
);
|
||||
}
|
||||
|
||||
/// 归档目录不做数量上限治理：所有已算网格点的完整产物（.spec/.cont/.iden/各阶段
|
||||
/// 快照/日志/种子二进制等）一律永久保留，避免 LRU 淘汰导致科学产物丢失
|
||||
/// （2026-08-02 修正：撤销 1bfa240 引入的 MAX_RESULT_MODELS=200 LRU 上限）。
|
||||
// 归档目录不做数量上限治理：所有已算网格点的完整产物（.spec/.cont/.iden/各阶段
|
||||
// 快照/日志/种子二进制等）一律永久保留，避免 LRU 淘汰导致科学产物丢失
|
||||
// （2026-08-02 修正：撤销 1bfa240 引入的 MAX_RESULT_MODELS=200 LRU 上限）。
|
||||
|
||||
/// `.seed_cache/` 内保留的 `.seed.7` 文件上限。超过则按 mtime 删除最旧的。
|
||||
/// 典型网格内活跃种子点数量有限，8 足以覆盖常用邻域且把磁盘占用控制在 ~8 个种子文件。
|
||||
const MAX_SEED_CACHE_FILES: usize = 8;
|
||||
///
|
||||
/// 审查修复 #N5：上限可经环境变量 `DCTS_SEED_CACHE_MAX` 覆盖（默认 8）。多工作流或密集
|
||||
/// 网格下常用邻域种子可能超过 8 个，硬编码上限会导致反复从 server 下载，增加负载。
|
||||
fn seed_cache_max_files() -> usize {
|
||||
std::env::var("DCTS_SEED_CACHE_MAX")
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.filter(|n: &usize| *n > 0)
|
||||
.unwrap_or(8)
|
||||
}
|
||||
|
||||
/// LRU 清理种子缓存目录：当 `.seed.7` 文件数超过 `MAX_SEED_CACHE_FILES` 时，
|
||||
/// LRU 清理种子缓存目录：当 `.seed.7` 文件数超过上限（`DCTS_SEED_CACHE_MAX`，默认 8）时，
|
||||
/// 按 mtime 升序删除最旧的若干个，直到不超过上限。仅统计 `.seed.7`，忽略 `.tmp` 中间文件。
|
||||
/// 任何 IO 错误均降级为 warn，不阻断主流程。
|
||||
pub async fn cleanup_seed_cache(seed_dir: &Path) {
|
||||
@@ -391,13 +427,14 @@ pub async fn cleanup_seed_cache(seed_dir: &Path) {
|
||||
Err(_) => return,
|
||||
};
|
||||
|
||||
if entries.len() <= MAX_SEED_CACHE_FILES {
|
||||
let max_files = seed_cache_max_files();
|
||||
if entries.len() <= max_files {
|
||||
return;
|
||||
}
|
||||
|
||||
// 按 mtime 升序（最旧在前），删除超出上限的最旧文件
|
||||
entries.sort_by_key(|(mtime, _)| *mtime);
|
||||
let to_remove = entries.len().saturating_sub(MAX_SEED_CACHE_FILES);
|
||||
let to_remove = entries.len().saturating_sub(max_files);
|
||||
for (_, path) in entries.into_iter().take(to_remove) {
|
||||
match tokio::fs::remove_file(&path).await {
|
||||
Ok(()) => info!("LRU 清理种子缓存文件: {}", path.display()),
|
||||
@@ -411,6 +448,43 @@ pub async fn cleanup_seed_cache(seed_dir: &Path) {
|
||||
}
|
||||
}
|
||||
|
||||
/// 解析任务要执行的大气链（冷启动链 / 种子热启动链）。
|
||||
///
|
||||
/// 优先级（修复回归）：
|
||||
/// - `current_strategy == "seed_step"` → 强制 `default_seed_chain()`(seed_nc→nl)。
|
||||
/// 自定义 `tlusty_chain` 是冷启动链：首步 lte 的 `ltgray=T` 会删除 fort.8、丢弃已下载的
|
||||
/// 热启动种子(runner.rs 阶段 fort.8 准备逻辑)。scheduler 在派发与回退两条路径都注入
|
||||
/// 同一个 `tlusty_chain`，若 seed_step 也沿用自定义链，会把种子回退退化成本地冷启动，
|
||||
/// 丢失热启动语义。故种子链固定走内置默认，仅在 cold_run 等冷策略下信任用户自定义链。
|
||||
/// - 其余策略 → 优先 TaskSpec.tlusty_chain_params(用户 YAML `tlusty_chain:` 配置)，非空即用；
|
||||
/// 为空/反序列化失败 → `default_chain_for_strategy(current_strategy)` 兜底。
|
||||
fn resolve_execution_chain(
|
||||
current_strategy: &str,
|
||||
tlusty_chain_params: &Option<serde_json::Value>,
|
||||
task_id: &str,
|
||||
) -> Vec<ChainStep> {
|
||||
if current_strategy == "seed_step" {
|
||||
common::runner::default_seed_chain()
|
||||
} else {
|
||||
tlusty_chain_params
|
||||
.as_ref()
|
||||
.and_then(
|
||||
|v| match serde_json::from_value::<Vec<ChainStep>>(v.clone()) {
|
||||
Ok(c) => Some(c),
|
||||
Err(e) => {
|
||||
warn!(
|
||||
"任务 {} 的 tlusty_chain_params 反序列化失败，回退 default 链: {}",
|
||||
task_id, e
|
||||
);
|
||||
None
|
||||
}
|
||||
},
|
||||
)
|
||||
.filter(|c| !c.is_empty())
|
||||
.unwrap_or_else(|| common::runner::default_chain_for_strategy(current_strategy))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -429,7 +503,7 @@ mod tests {
|
||||
name: params.model_name(),
|
||||
params,
|
||||
stages: vec![],
|
||||
converged: true,
|
||||
result_valid: true,
|
||||
final_max_relc: Some(0.0005),
|
||||
final_chmax: None,
|
||||
seed: None,
|
||||
@@ -442,6 +516,44 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn labels(chain: &[ChainStep]) -> Vec<String> {
|
||||
chain.iter().map(|s| s.label.clone()).collect()
|
||||
}
|
||||
|
||||
/// P1 回归防护：seed_step 即使注入自定义冷启动链，也必须强制走种子热启动默认链
|
||||
/// （seed_nc→nl），否则首步 lte(ltgray=T) 会删 fort.8、丢弃已下载种子，回退退化成本地冷启动。
|
||||
#[test]
|
||||
fn seed_step_ignores_custom_cold_chain() {
|
||||
let custom = serde_json::json!([
|
||||
{"label": "lte", "lte": "T", "ltgray": "T", "ilvlin": 0, "niter": 0},
|
||||
{"label": "nc", "lte": "F", "ltgray": "F", "ilvlin": 0, "niter": 10},
|
||||
{"label": "nl", "lte": "F", "ltgray": "F", "ilvlin": 100, "niter": 100},
|
||||
]);
|
||||
|
||||
let chain = resolve_execution_chain("seed_step", &Some(custom), "t1");
|
||||
assert_eq!(labels(&chain), vec!["seed_nc", "nl"]);
|
||||
// 首步必须是非灰 LTE（ltgray=F），否则会删 fort.8 丢弃种子。
|
||||
assert_eq!(chain[0].ltgray, "F");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cold_run_uses_custom_chain_when_provided() {
|
||||
let custom = serde_json::json!([
|
||||
{"label": "lte", "lte": "T", "ltgray": "T", "ilvlin": 0, "niter": 0},
|
||||
{"label": "nc", "lte": "F", "ltgray": "F", "ilvlin": 0, "niter": 10},
|
||||
{"label": "nl", "lte": "F", "ltgray": "F", "ilvlin": 100, "niter": 100},
|
||||
]);
|
||||
|
||||
let chain = resolve_execution_chain("cold_run", &Some(custom), "t2");
|
||||
assert_eq!(labels(&chain), vec!["lte", "nc", "nl"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cold_run_falls_back_to_default_when_no_custom_chain() {
|
||||
let chain = resolve_execution_chain("cold_run", &None, "t3");
|
||||
assert_eq!(labels(&chain), vec!["lte", "nc", "nl"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_cleanup_slot_work_dir() {
|
||||
let temp_dir =
|
||||
|
||||
@@ -12,8 +12,14 @@ use tracing::{info, warn};
|
||||
/// 失败导致 synspec 无输入"的级联场景）；TLUSTY 关闭 → 归因 SYNSPEC（synspec-only 任务
|
||||
/// 的 converged 由 synspec_rc 决定）。
|
||||
/// - `converged=true` 但 synspec 有错误/非零 rc：归因 SYNSPEC（大气已收敛、光谱失败）。
|
||||
///
|
||||
/// 已知盲区（审查 #N1，文档化取舍）：TLUSTY 关闭 + `converged=false` 时归因 SYNSPEC，
|
||||
/// 但此时大气来自外部既有产物（非本任务计算），`converged=false` 更可能意味着外部大气
|
||||
/// 文件损坏/NaN/缺失，而非光谱合成本身的问题。归因到 synspec 触发的 synspec 链重试对
|
||||
/// 「外部大气损坏」无济于事（重试仍用同一损坏大气）。当前仍按 synspec 归因是保守选择
|
||||
/// （至少触发一次重试暴露问题），运维需结合 note 里的 atmosphere_has_nan 标志甄别。
|
||||
fn infer_failed_stage(task: &TaskSpec, summary: &ModelSummary) -> Option<String> {
|
||||
if !summary.converged {
|
||||
if !summary.result_valid {
|
||||
return if task.tlusty_config.enabled {
|
||||
Some("tlusty".to_string())
|
||||
} else {
|
||||
@@ -37,7 +43,7 @@ fn infer_failed_stage(task: &TaskSpec, summary: &ModelSummary) -> Option<String>
|
||||
/// 保存 .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 {
|
||||
if s.result_valid && !synspec_failed {
|
||||
TaskStatus::Completed
|
||||
} else {
|
||||
TaskStatus::Failed
|
||||
@@ -66,7 +72,7 @@ pub async fn report_result(
|
||||
) = match exec_res {
|
||||
Ok((s, s_bytes)) => (
|
||||
derive_report_status(&s),
|
||||
s.converged,
|
||||
s.result_valid,
|
||||
s.final_max_relc,
|
||||
s.atmosphere_has_nan,
|
||||
s.elapsed_sec,
|
||||
@@ -94,7 +100,7 @@ pub async fn report_result(
|
||||
params: Some(task.params.clone()),
|
||||
node_id: node_id.to_string(),
|
||||
status,
|
||||
converged,
|
||||
result_valid: converged,
|
||||
max_relc,
|
||||
atmosphere_has_nan: atmo_has_nan,
|
||||
elapsed_sec,
|
||||
@@ -196,16 +202,16 @@ mod tests {
|
||||
logn: (-4.0).into(),
|
||||
logo: (-4.0).into(),
|
||||
},
|
||||
tlusty_config: common::models::EngineStageConfig {
|
||||
tlusty_config: common::models::PhaseConfig {
|
||||
enabled: tlusty_enabled,
|
||||
..common::models::EngineStageConfig::default_tlusty()
|
||||
..common::models::PhaseConfig::default_tlusty()
|
||||
},
|
||||
..TaskSpec::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn mk_summary(
|
||||
converged: bool,
|
||||
result_valid: bool,
|
||||
synspec_rc: Option<i32>,
|
||||
synspec_error: Option<&str>,
|
||||
) -> ModelSummary {
|
||||
@@ -220,7 +226,7 @@ mod tests {
|
||||
logo: (-4.0).into(),
|
||||
},
|
||||
stages: vec![],
|
||||
converged,
|
||||
result_valid,
|
||||
final_max_relc: None,
|
||||
final_chmax: None,
|
||||
seed: None,
|
||||
@@ -302,4 +308,27 @@ mod tests {
|
||||
TaskStatus::Failed
|
||||
);
|
||||
}
|
||||
|
||||
/// 漏洞1修复验证（端到端判定链路）：.spec 内容校验失败时，runner 同时设置
|
||||
/// synspec_rc=Some(1) + synspec_error=Some("spec...")。验证 derive_report_status 和
|
||||
/// infer_failed_stage 在双字段同时非空/非零时都判 synspec 失败 → Failed + 归因 synspec。
|
||||
/// 这是漏洞1失败信号从 runner 一路流到 reporter 判定的关键节点。
|
||||
#[test]
|
||||
fn test_spec_invalid_drives_failed() {
|
||||
// 模拟 spec_is_valid 命中：converged=true（大气正常）+ spec 脏（双字段）
|
||||
let dirty_spec_summary =
|
||||
mk_summary(true, Some(1), Some("spec 含 NaN/Inf/溢出行 (共 2 行)"));
|
||||
// 任务整体判 Failed（不再被误报 Completed → 脏谱归档）
|
||||
assert_eq!(
|
||||
derive_report_status(&dirty_spec_summary),
|
||||
TaskStatus::Failed,
|
||||
"spec 校验失败（双字段）应判 Failed"
|
||||
);
|
||||
// 归因 synspec（触发 synspec 策略链回退）
|
||||
assert_eq!(
|
||||
infer_failed_stage(&mk_task(true), &dirty_spec_summary),
|
||||
Some("synspec".to_string()),
|
||||
"spec 校验失败应归因 synspec 以触发回退"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+117
-9
@@ -6,9 +6,11 @@ use common::embedded::RuntimePaths;
|
||||
use common::models::{NodeHeartbeatRequest, NodeHeartbeatResponse, NodeRegisterRequest, TaskSpec};
|
||||
use reqwest::Client;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashSet;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::{AtomicI32, AtomicUsize, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicI32, AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::time::{sleep, Duration};
|
||||
use tracing::{info, warn};
|
||||
|
||||
@@ -32,10 +34,65 @@ impl Drop for SlotGuard {
|
||||
}
|
||||
}
|
||||
|
||||
/// 后台沙盒 GC 间隔（秒）：长寿命 Worker 运行期间周期清理孤儿 `task_*` 目录。
|
||||
/// 孤儿来源：`cleanup_slot_work_dir` 反复失败（权限/文件锁）或任务进程异常残留。
|
||||
/// 启动清理（run() 开头）只处理「上次崩溃」的残留；运行期累积靠本 GC 兜底。
|
||||
/// 精确跳过 `active_tasks` 中的在途任务，故无需担心误删运行中沙盒。
|
||||
const SANDBOX_GC_INTERVAL_SECS: u64 = 3600;
|
||||
|
||||
/// 周期清理 work_dir 下非活跃的 `task_*` 沙盒目录。
|
||||
///
|
||||
/// - 目录名 `task_{uuid}`，后缀精确匹配 `active_tasks`（在途任务 task_id 集合）；
|
||||
/// 命中的在途任务跳过，其余视为孤儿删除（best-effort，失败 warn 下轮重试）。
|
||||
/// - 与启动清理（run() 开头删全部）互补：运行期若单 slot 清理反复失败，孤儿子目录
|
||||
/// 不会累积写满磁盘，由本任务周期性回收。
|
||||
/// - 仅在 shutdown 标志置位时退出（每 tick 检查一次），随节点优雅退出。
|
||||
async fn sandbox_gc_loop(
|
||||
work_dir: PathBuf,
|
||||
active_tasks: Arc<Mutex<HashSet<String>>>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
) {
|
||||
let mut ticker = tokio::time::interval(Duration::from_secs(SANDBOX_GC_INTERVAL_SECS));
|
||||
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
if shutdown.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
let active: HashSet<String> = active_tasks.lock().await.iter().cloned().collect();
|
||||
if let Ok(mut rd) = tokio::fs::read_dir(&work_dir).await {
|
||||
while let Ok(Some(entry)) = rd.next_entry().await {
|
||||
let name = entry.file_name();
|
||||
let name_str = name.to_string_lossy();
|
||||
if is_orphan_task_dir(&name_str, &active) {
|
||||
let p = entry.path();
|
||||
match tokio::fs::remove_dir_all(&p).await {
|
||||
Ok(_) => warn!("GC 清理孤儿沙盒: {}", p.display()),
|
||||
Err(e) => warn!("GC 清理孤儿沙盒 {} 失败（下轮重试）: {}", p.display(), e),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 判断 work_dir 下的一个条目是否为「应回收的孤儿沙盒」。
|
||||
/// `task_{uuid}` 且 uuid 不在 `active_tasks`（在途任务）→ 孤儿；其它一律非孤儿。
|
||||
fn is_orphan_task_dir(entry_name: &str, active: &HashSet<String>) -> bool {
|
||||
match entry_name.strip_prefix("task_") {
|
||||
Some(suffix) => !active.contains(suffix),
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// 领用请求的归一化结果。
|
||||
///
|
||||
/// 区分「被管理员停用」与「暂无任务」：前者节点保持存活、空闲待命（拉长轮询），
|
||||
/// 后者按常规节奏轮询。服务端返回 `{"status":"disabled"}` 映射为 `Disabled`。
|
||||
///
|
||||
/// `TaskSpec` 体积远大于空枚举项，属可接受的形态（node 端一次性消费，非热循环持有）；
|
||||
/// 装箱会改变所有 match 点的解构方式且无实际收益，故允许该 lint。
|
||||
#[allow(clippy::large_enum_variant)]
|
||||
enum ClaimOutcome {
|
||||
/// 成功领用到任务。
|
||||
Task(TaskSpec),
|
||||
@@ -62,6 +119,12 @@ pub struct NodeWorker {
|
||||
/// （见 docs/dynamic_cpu_slots_design.md §3.2）。用 `Arc<AtomicUsize>` 在心跳线程与
|
||||
/// 领用线程间共享，无需 Mutex：单写者（心跳线程）单读者（领用线程），原子读写即可。
|
||||
effective_max_slots: Arc<AtomicUsize>,
|
||||
/// 全局优雅退出标志：信号线程、心跳线程、claim_task 任一发现需要终止时置 true，
|
||||
/// 主循环据此停止领用新任务并走优雅退出（等待在途任务 ≤30s + 归档沙盒）。
|
||||
///
|
||||
/// 审查修复 #M5：此前心跳/claim_task 遇 token 失效直接 `std::process::exit(1)`，
|
||||
/// 跳过主循环的优雅退出逻辑，导致在途长任务结果静默丢失。改为置此标志让主循环感知。
|
||||
shutting_down: Arc<std::sync::atomic::AtomicBool>,
|
||||
}
|
||||
|
||||
impl NodeWorker {
|
||||
@@ -73,6 +136,7 @@ impl NodeWorker {
|
||||
runtime,
|
||||
active_slots: Arc::new(AtomicI32::new(0)),
|
||||
effective_max_slots: Arc::new(AtomicUsize::new(physical)),
|
||||
shutting_down: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -218,6 +282,7 @@ impl NodeWorker {
|
||||
let hb_physical = self.config.max_slots;
|
||||
let hb_interval = self.config.heartbeat_sec;
|
||||
let hb_runtime_dir = self.config.runtime_dir.clone();
|
||||
let hb_shutting_down = self.shutting_down.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let sys_arc = std::sync::Arc::new(std::sync::Mutex::new(sysinfo::System::new_all()));
|
||||
@@ -275,20 +340,22 @@ impl NodeWorker {
|
||||
match hb_client.post(&hb_url).json(&req).send().await {
|
||||
Ok(resp) => {
|
||||
let status = resp.status();
|
||||
// 401/403：token 失效（被重发覆盖）。与 claim_task 口径统一：直接退出进程，
|
||||
// 避免心跳线程持续发被拒请求刷日志、占用服务端限流计数。心跳通常比
|
||||
// claim 更高频，往往先于 claim_task 发现 token 失效。
|
||||
// 401/403：token 失效（被重发覆盖）。与 claim_task 口径统一：
|
||||
// 审查修复 #M5：原 `std::process::exit(1)` 跳过主循环优雅退出，
|
||||
// 在途长任务结果静默丢失。现置 shutting_down 让主循环停止领用新任务、
|
||||
// 等待在途任务完成（≤30s）并归档沙盒后再退出。
|
||||
if status.as_u16() == 401 || status.as_u16() == 403 {
|
||||
tracing::error!(
|
||||
"节点 {} 心跳被服务端拒绝 (HTTP {})：node token 已失效（已被重发覆盖）。\n\
|
||||
恢复方式：把管理员重发的新 token 明文写入 {} 后重启节点，\n\
|
||||
或删除该文件后重启以重新提交注册申请等待审批。\n\
|
||||
进程将退出，依赖编排系统重启。",
|
||||
已触发优雅退出：等待在途任务完成后进程退出，依赖编排系统重启。",
|
||||
hb_node_id,
|
||||
status,
|
||||
token_file_display(&hb_runtime_dir)
|
||||
);
|
||||
std::process::exit(1);
|
||||
hb_shutting_down.store(true, Ordering::Release);
|
||||
break;
|
||||
}
|
||||
// 解析心跳响应体，提取管理员配额（见 docs/dynamic_cpu_slots_design.md
|
||||
// §3.2）：计算生效配额并更新 effective_max_slots。反序列化失败不致命
|
||||
@@ -348,7 +415,18 @@ impl NodeWorker {
|
||||
}
|
||||
}
|
||||
|
||||
let shutting_down = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
// 运行期沙盒 GC：追踪在途任务 task_id 集合，后台周期清理孤儿 task_* 目录
|
||||
//（cleanup_slot_work_dir 反复失败/进程异常残留的累积治理，见 sandbox_gc_loop）。
|
||||
let active_tasks: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
|
||||
tokio::spawn(sandbox_gc_loop(
|
||||
work_dir.clone(),
|
||||
active_tasks.clone(),
|
||||
self.shutting_down.clone(),
|
||||
));
|
||||
|
||||
// 复用 self.shutting_down（结构体字段）：心跳线程、claim_task、信号线程、主循环
|
||||
// 共享同一标志。任一来源触发优雅退出，主循环都能感知（审查修复 #M5）。
|
||||
let shutting_down = self.shutting_down.clone();
|
||||
let shutdown_signal = shutting_down.clone();
|
||||
|
||||
// 信号处理：同时监听 SIGINT（Ctrl+C）与 SIGTERM。
|
||||
@@ -428,6 +506,8 @@ impl NodeWorker {
|
||||
let result_dir = result_dir.clone();
|
||||
let slots_counter = self.active_slots.clone();
|
||||
let shutdown = shutting_down.clone();
|
||||
// 在途任务 task_id 集合（沙盒 GC 跳过用）：spawn 前登记、任务结束移除。
|
||||
let active_tasks = active_tasks.clone();
|
||||
|
||||
// 在 spawn 前同步自增，与容量检查紧邻成原子操作：避免
|
||||
// 「检查通过 → spawn 排队 → 回循环再检查时计数尚未自增」的竞态
|
||||
@@ -438,6 +518,8 @@ impl NodeWorker {
|
||||
let slot_guard = SlotGuard {
|
||||
counter: slots_counter.clone(),
|
||||
};
|
||||
// GC 活跃集登记（在 spawn 前，避免 GC 误删尚未建目录的在途任务）。
|
||||
active_tasks.lock().await.insert(task.task_id.to_string());
|
||||
|
||||
tokio::spawn(async move {
|
||||
// 把守卫移入 future，确保任务结束（含 panic）时归还 slot。
|
||||
@@ -509,6 +591,9 @@ impl NodeWorker {
|
||||
);
|
||||
}
|
||||
}
|
||||
// 任务结束（含 panic/清理失败）：从 GC 活跃集移除，下轮 GC 可回收
|
||||
// 清理失败残留的孤儿沙盒。
|
||||
active_tasks.lock().await.remove(&task.task_id.to_string());
|
||||
// _slot_guard 在此作用域结束时 drop，归还活动 slot 计数。
|
||||
});
|
||||
}
|
||||
@@ -581,11 +666,14 @@ impl NodeWorker {
|
||||
"领用任务被服务端拒绝 (HTTP {})：node token 已失效（已被重发覆盖）。\n\
|
||||
恢复方式：把管理员重发的新 token 明文写入 {} 后重启节点，\n\
|
||||
或删除该文件后重启以重新提交注册申请等待审批。\n\
|
||||
进程将退出，依赖编排系统重启。",
|
||||
已触发优雅退出：等待在途任务完成后进程退出，依赖编排系统重启。",
|
||||
status,
|
||||
token_file_display(&self.config.runtime_dir)
|
||||
);
|
||||
std::process::exit(1);
|
||||
// 审查修复 #M5：与心跳线程口径统一，置 shutting_down 让主循环走优雅退出，
|
||||
// 而非 exit(1) 跳过在途任务的结果上报与沙盒归档。
|
||||
self.shutting_down.store(true, Ordering::Release);
|
||||
return Ok(ClaimOutcome::Empty);
|
||||
}
|
||||
|
||||
if status.is_server_error() {
|
||||
@@ -612,3 +700,23 @@ impl NodeWorker {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// 沙盒 GC 孤儿判定（#4）：`task_{uuid}` 且 uuid 不在活跃集 → 孤儿；在途/非沙盒目录 → 否。
|
||||
#[test]
|
||||
fn orphan_gc_predicate_skips_active_and_non_sandbox() {
|
||||
let active: HashSet<String> = ["task-1111-aaaa".to_string()].into_iter().collect();
|
||||
// 在途任务 → 非孤儿（GC 跳过）。
|
||||
assert!(!is_orphan_task_dir("task_task-1111-aaaa", &active));
|
||||
// 孤儿（uuid 不在活跃集）→ 回收。
|
||||
assert!(is_orphan_task_dir("task_task-2222-bbbb", &active));
|
||||
// 非沙盒目录（.seed_cache / 其它）→ 永不动。
|
||||
assert!(!is_orphan_task_dir(".seed_cache", &active));
|
||||
assert!(!is_orphan_task_dir("result_dir", &active));
|
||||
// 空活跃集：所有 task_* 都是孤儿（等价重启清理）。
|
||||
assert!(is_orphan_task_dir("task_abc", &HashSet::new()));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user