fmq
1bfa240cb0
核心变更：
1. GridAxisValue 源精度命名
- 新增 GridAxisValue 类型，携带 f64 数值 + YAML 源书写文本（Deref 透明兼容算术）
- config.rs 绕过 serde_yaml 归一化，逐 token 捕获轴值原文（logg: 5.0 → g5.0）
- runner/executor/scheduler 全链路改用 DB TEXT 列权威 point_name，
修复 REAL 列回读丢精度导致的 model_name 错配
2. 工作流执行可观测台
- 新增 stats/progress/points 三组 API（进度时间序列、经验速率 ETA、
停滞预警、逐点明细分页、收敛性热力图数据）
- 新增 workflow_progress_snapshots 表 + tasks/grid_points 耗时列
- runner 携带 last_iter/worst_depth/n_depths 进 conv.json
- 前端新增 hash 路由、工作流详情页（概览/网格点/收敛分析三 Tab）、YAML 编辑器
3. 节点停用/启用管理
- 新增 disabled 状态 + disable/enable API；停用节点保持心跳但停止分发，
worker 空闲待命而非退出；移除 revoke API，token 失效统一走重发覆盖；
移除 host_name 字段
4. 白名单结果归档
- 新增 result_filter 模块，只归档有语义产物，丢弃 Tlusty 中间单元（~2MB/模型）
- executor 原子写入归档 + 200 点 LRU 上限
5. 历史数据导入
- sync_seeds 重写为 import_results：经 /admin/import_seed 标记 converged +
按新版命名迁移产物树
6. 部署与目录重规划
- data/results→seeds、data/archive→result + migrate_data_dirs.sh
- deploy.sh 增强（SSH 复用、Profile、远程 env）；Dockerfile 瘦身
7. 文档同步更新 api/database/architecture/deployment
397 lines
16 KiB
Rust
397 lines
16 KiB
Rust
use super::{AppState, AuthenticatedNode};
|
||
use axum::{
|
||
extract::{Extension, Multipart, Query, State},
|
||
response::IntoResponse,
|
||
Json,
|
||
};
|
||
use common::models::{GridPointParams, ModelSummary, TaskReport, TaskStatus};
|
||
use serde::Deserialize;
|
||
use serde_json::json;
|
||
use std::path::Path;
|
||
use tokio::fs;
|
||
use tracing::{info, warn};
|
||
|
||
use axum::http::StatusCode;
|
||
|
||
pub async fn claim_task(
|
||
State(state): State<AppState>,
|
||
Extension(auth_node): Extension<AuthenticatedNode>,
|
||
) -> Result<impl IntoResponse, crate::api::AppError> {
|
||
// 管理员手动停用拦截：被停用的节点保持在线但不再分发任务。
|
||
// 返回 HTTP 200 + {"status":"disabled"}（绝不能用 403 —— worker 见 403 会判定
|
||
// token 失效而 exit(1)，停用是运维意图而非凭据失效，应让 worker 空闲待命）。
|
||
match state.db.is_node_disabled(&auth_node.node_id).await {
|
||
Ok(true) => {
|
||
return Ok((
|
||
StatusCode::OK,
|
||
Json(json!({"status": "disabled", "task": null})),
|
||
));
|
||
}
|
||
Ok(false) => {}
|
||
Err(e) => {
|
||
tracing::error!("查询节点停用状态异常: {}", e);
|
||
return Err(crate::api::AppError::Internal(e));
|
||
}
|
||
}
|
||
|
||
// 领用时记录任务归属：pop_task 写入 claimed_by_node_id，
|
||
// report 阶段据此校验「上报者确为领用者」，杜绝跨节点伪造结果。
|
||
match state.queue.pop_task(&auth_node.node_id).await {
|
||
Ok(Some(task)) => {
|
||
// 多工作流分区：mark_grid_point_running 须带 workflow_name，避免按 name 全局更新
|
||
// 误改其他工作流的同名点。TaskSpec.workflow_name 在调度时已绑定。
|
||
let wf = task.workflow_name.as_deref().unwrap_or("");
|
||
if let Err(e) = state.db.mark_grid_point_running(&task.point_name, wf).await {
|
||
warn!("领用任务 {} 后同步变更为 running 状态遇到异常: {}. 后置 stale 定时自取检索引索将介入修复维护", task.task_id, e);
|
||
}
|
||
Ok((StatusCode::OK, Json(json!({"status": "ok", "task": task}))))
|
||
}
|
||
Ok(None) => Ok((
|
||
StatusCode::OK,
|
||
Json(json!({"status": "empty", "task": null})),
|
||
)),
|
||
Err(e) => {
|
||
tracing::error!("领用任务数据库异常: {}", e);
|
||
Err(crate::api::AppError::Internal(e))
|
||
}
|
||
}
|
||
}
|
||
|
||
pub async fn report_task(
|
||
State(state): State<AppState>,
|
||
Extension(auth_node): Extension<AuthenticatedNode>,
|
||
mut multipart: Multipart,
|
||
) -> Result<impl IntoResponse, crate::api::AppError> {
|
||
let mut report_json: Option<TaskReport> = None;
|
||
let mut seed_file_data: Option<Vec<u8>> = None;
|
||
|
||
while let Ok(Some(field)) = multipart.next_field().await {
|
||
let field_name = field.name().unwrap_or("").to_string();
|
||
if field_name == "report" {
|
||
if let Ok(bytes) = field.bytes().await {
|
||
if let Ok(report) = serde_json::from_slice::<TaskReport>(&bytes) {
|
||
report_json = Some(report);
|
||
}
|
||
}
|
||
} else if field_name == "seed_file" {
|
||
if let Ok(bytes) = field.bytes().await {
|
||
seed_file_data = Some(bytes.to_vec());
|
||
}
|
||
}
|
||
}
|
||
|
||
let mut report = match report_json {
|
||
Some(r) => r,
|
||
None => {
|
||
return Err(crate::api::AppError::BadRequest(
|
||
"请求中缺少 report 字段".to_string(),
|
||
));
|
||
}
|
||
};
|
||
|
||
// ── 任务归属校验（S1 核心，防跨节点伪造结果投毒）──
|
||
// 1. 该 task_id 必须由当前鉴权 node 领用（claim 时记录的 claimed_by_node_id 匹配）。
|
||
// 2. 上报的 point_name 必须与该 task 绑定的 point_name 一致（防跨点上报）。
|
||
// 3. 忽略 body 里声称的 node_id，统一以鉴权 node_id 写库（修复审计归因断裂）。
|
||
// 4. 取 task 绑定的 workflow_name，用于定向更新该工作流的 grid_points（多工作流分区）。
|
||
let (claimed_point, claimed_workflow) = match state
|
||
.queue
|
||
.verify_task_claim(&report.task_id.to_string(), &auth_node.node_id)
|
||
.await
|
||
{
|
||
Ok(Some((p, w))) => (p, w),
|
||
Ok(None) => {
|
||
warn!(
|
||
"任务归属校验失败：node={} 上报 task_id={} 但未领用或已被清理",
|
||
auth_node.node_id, report.task_id
|
||
);
|
||
return Err(crate::api::AppError::Forbidden(
|
||
"任务未由本节点领用或已上报过".to_string(),
|
||
));
|
||
}
|
||
Err(e) => {
|
||
tracing::error!("校验任务归属数据库异常: {}", e);
|
||
return Err(crate::api::AppError::Internal(e));
|
||
}
|
||
};
|
||
if claimed_point != report.point_name {
|
||
warn!(
|
||
"任务点名校验失败：task_id={} 领用 point={} 但上报 point={}",
|
||
report.task_id, claimed_point, report.point_name
|
||
);
|
||
return Err(crate::api::AppError::Forbidden(
|
||
"上报的网格点与领用任务不匹配".to_string(),
|
||
));
|
||
}
|
||
// 统一以鉴权 node_id 覆盖 body 里的 node_id，保证归因可信
|
||
report.node_id = auth_node.node_id.clone();
|
||
// workflow_name 以领用记录为准（claim 时从 TaskSpec 落库），body 无权声称。
|
||
let workflow_name = claimed_workflow.unwrap_or_default();
|
||
|
||
let name = report.point_name.clone();
|
||
|
||
if !super::workflow::is_valid_point_name(&name) {
|
||
warn!(
|
||
"拒绝可能包含路径穿越或特殊非常规编码号攻击的网格点名称请求: {}",
|
||
name
|
||
);
|
||
return Err(crate::api::AppError::BadRequest(
|
||
"非法的网格点名称参数".to_string(),
|
||
));
|
||
}
|
||
|
||
let params = match extract_params(&report) {
|
||
Some(p) => p,
|
||
None => {
|
||
warn!(
|
||
"网格点 {} 汇报数据解析失败: 无法解析 params 或 summary_json",
|
||
name
|
||
);
|
||
return Err(crate::api::AppError::BadRequest(
|
||
"无法解析 params 或 summary_json".to_string(),
|
||
));
|
||
}
|
||
};
|
||
|
||
// Record in DB（带 workflow_name 定向更新该工作流的 grid_points）
|
||
if let Err(e) = state.db.record_task_report(&report, &workflow_name).await {
|
||
// DB 错误细节进日志，对客户端只返回通用消息（避免泄露表结构/内部错误给未授权方）
|
||
tracing::error!("记录网格点 {} 任务结果到数据库失败: {}", name, e);
|
||
return Err(crate::api::AppError::Internal(e));
|
||
}
|
||
|
||
// Clean up task from task_queue table to prevent queue DB bloat
|
||
if let Err(e) = state.queue.remove_task(&report.task_id.to_string()).await {
|
||
tracing::warn!(
|
||
"从任务队列中清理已上报任务记录 {} 失败: {}",
|
||
report.task_id,
|
||
e
|
||
);
|
||
}
|
||
|
||
// 采用原子写入模式保持 conv.json 与核心二进制数据完整落地后才揭晓真实文件名
|
||
let model_dir = Path::new(&state.seeds_dir).join(&name);
|
||
if fs::create_dir_all(&model_dir).await.is_ok() {
|
||
let conv_tmp = model_dir.join(format!("conv.json.{}.tmp", uuid::Uuid::new_v4().simple()));
|
||
let conv_path = model_dir.join("conv.json");
|
||
if fs::write(&conv_tmp, &report.summary_json).await.is_ok() {
|
||
let _ = fs::rename(&conv_tmp, &conv_path).await;
|
||
}
|
||
|
||
// Save seed file .7 using atomic temporary writing strategy
|
||
if report.converged && !report.atmosphere_has_nan {
|
||
if let Some(bytes) = seed_file_data {
|
||
let seed_tmp =
|
||
model_dir.join(format!("{}.7.{}.tmp", name, uuid::Uuid::new_v4().simple()));
|
||
let seed_path = model_dir.join(format!("{}.7", name));
|
||
if fs::write(&seed_tmp, bytes).await.is_ok()
|
||
&& fs::rename(&seed_tmp, &seed_path).await.is_ok()
|
||
{
|
||
info!(
|
||
"成功保持原子写入落地并保存网格点 {} 的收敛种子文件: {}",
|
||
name,
|
||
seed_path.display()
|
||
);
|
||
let _ = state
|
||
.db
|
||
.insert_seed(¶ms, &seed_path.to_string_lossy())
|
||
.await;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if !report.converged
|
||
|| report.atmosphere_has_nan
|
||
|| report.status == TaskStatus::Failed
|
||
|| report.status == TaskStatus::Timeout
|
||
{
|
||
// Task did not succeed -> check if seed_step fallback should be triggered
|
||
info!("网格点 {} 计算未成功完成，检查种子回退机制...", name);
|
||
if let Err(e) = state
|
||
.scheduler
|
||
.trigger_seed_step_fallback(¶ms, &name, &workflow_name)
|
||
.await
|
||
{
|
||
warn!("网格点 {} 触发种子回退机制失败: {}", name, e);
|
||
}
|
||
}
|
||
|
||
Ok((
|
||
StatusCode::OK,
|
||
Json(json!({"status": "ok", "message": "上报成功"})),
|
||
))
|
||
}
|
||
|
||
fn extract_params(report: &TaskReport) -> Option<GridPointParams> {
|
||
if let Some(ref p) = report.params {
|
||
return Some(p.clone());
|
||
}
|
||
serde_json::from_str::<ModelSummary>(&report.summary_json)
|
||
.ok()
|
||
.map(|summary| summary.params)
|
||
}
|
||
|
||
/// `/admin/import_seed` 的查询参数。
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct ImportSeedQuery {
|
||
/// 目标工作流名（导入到此工作流的 grid_points）。缺省归入 `imported` 工作流。
|
||
#[serde(default = "default_import_workflow")]
|
||
pub workflow: String,
|
||
}
|
||
|
||
fn default_import_workflow() -> String {
|
||
"imported".to_string()
|
||
}
|
||
|
||
/// 历史种子导入端点（Admin 鉴权）。
|
||
///
|
||
/// 供 `tools/import_results` 把旧版单机 `run_grid.py` 产物（`conv.json` + `.7` 大气文件）
|
||
/// 批量回灌进 DCTS。与 `/task/report` 的关键区别：
|
||
/// - **跳过任务归属校验**（`verify_task_claim`）：历史数据无领用语义，导入端点不经过
|
||
/// claim/report 队列，直接幂等落库。
|
||
/// - **`point_name` 取旧 `conv.json` 的 `name` 字段**（Python `gen_input5.model_name`
|
||
/// 生成的源精度真名，如 `t20000_g5.0_...`），而非从数值重推——保证迁移逐字符保真。
|
||
/// - **真实 `max_relc`** 取自 `summary.final_max_relc`（旧版已记录），不硬编码。
|
||
///
|
||
/// 幂等：`upsert_grid_point` 用 `ON CONFLICT DO NOTHING`，`.7`/`conv.json` 原子覆盖写，
|
||
/// 可重复运行。
|
||
pub async fn import_seed(
|
||
State(state): State<AppState>,
|
||
Query(query): Query<ImportSeedQuery>,
|
||
mut multipart: Multipart,
|
||
) -> Result<impl IntoResponse, crate::api::AppError> {
|
||
let mut summary_json: Option<String> = None;
|
||
let mut seed_file_data: Option<Vec<u8>> = None;
|
||
|
||
while let Ok(Some(field)) = multipart.next_field().await {
|
||
let field_name = field.name().unwrap_or("").to_string();
|
||
if field_name == "report" {
|
||
if let Ok(bytes) = field.bytes().await {
|
||
summary_json = Some(String::from_utf8_lossy(&bytes).to_string());
|
||
}
|
||
} else if field_name == "seed_file" {
|
||
if let Ok(bytes) = field.bytes().await {
|
||
seed_file_data = Some(bytes.to_vec());
|
||
}
|
||
}
|
||
}
|
||
|
||
let summary_json = match summary_json {
|
||
Some(s) => s,
|
||
None => {
|
||
return Err(crate::api::AppError::BadRequest(
|
||
"请求中缺少 report 字段（旧版 conv.json 内容）".to_string(),
|
||
));
|
||
}
|
||
};
|
||
|
||
// 解析旧版 conv.json（ModelSummary 结构）取 name / params / 收敛状态 / 真实 max_relc。
|
||
let summary: ModelSummary = match serde_json::from_str(&summary_json) {
|
||
Ok(s) => s,
|
||
Err(e) => {
|
||
warn!("历史种子导入：conv.json 解析失败: {}", e);
|
||
return Err(crate::api::AppError::BadRequest(
|
||
"conv.json 解析失败，非合法 ModelSummary".to_string(),
|
||
));
|
||
}
|
||
};
|
||
|
||
// point_name 优先用旧 conv.json 的 name（源精度真名）；回退到 params 规范名。
|
||
let name = if !summary.name.is_empty() {
|
||
summary.name.clone()
|
||
} else {
|
||
summary.params.model_name()
|
||
};
|
||
|
||
// 名称合法性校验（防路径穿越），与 report_task 同口径。
|
||
if !super::workflow::is_valid_point_name(&name) {
|
||
warn!("历史种子导入：拒绝非法网格点名称: {}", name);
|
||
return Err(crate::api::AppError::BadRequest(
|
||
"非法的网格点名称参数".to_string(),
|
||
));
|
||
}
|
||
|
||
let workflow_name = query.workflow;
|
||
let params = summary.params.clone();
|
||
let converged = summary.converged && !summary.atmosphere_has_nan;
|
||
let max_relc = summary.final_max_relc;
|
||
|
||
// 1. 幂等写入 grid_points（ON CONFLICT DO NOTHING）：无需事先 start 工作流。
|
||
// 用权威 name（旧 conv.json 的源精度真名），而非从 params 重推——导入路径的
|
||
// params 来自旧 JSON（无源文本，model_name() 会失真）。
|
||
if let Err(e) = state
|
||
.db
|
||
.upsert_grid_point_named(&name, ¶ms, 0, &workflow_name)
|
||
.await
|
||
{
|
||
tracing::error!("历史种子导入：upsert grid_points {} 失败: {}", name, e);
|
||
return Err(crate::api::AppError::Internal(e));
|
||
}
|
||
|
||
// 2. 落地 conv.json（原子 tmp→rename）。
|
||
let model_dir = Path::new(&state.seeds_dir).join(&name);
|
||
if fs::create_dir_all(&model_dir).await.is_ok() {
|
||
let conv_tmp = model_dir.join(format!("conv.json.{}.tmp", uuid::Uuid::new_v4().simple()));
|
||
let conv_path = model_dir.join("conv.json");
|
||
if fs::write(&conv_tmp, &summary_json).await.is_ok() {
|
||
let _ = fs::rename(&conv_tmp, &conv_path).await;
|
||
}
|
||
|
||
// 3. 收敛且干净才写 .7 + 入种子库（与 report_task 同口径）。
|
||
if converged {
|
||
if let Some(bytes) = seed_file_data {
|
||
let seed_tmp =
|
||
model_dir.join(format!("{}.7.{}.tmp", name, uuid::Uuid::new_v4().simple()));
|
||
let seed_path = model_dir.join(format!("{}.7", name));
|
||
if fs::write(&seed_tmp, bytes).await.is_ok()
|
||
&& fs::rename(&seed_tmp, &seed_path).await.is_ok()
|
||
{
|
||
info!(
|
||
"历史种子导入：网格点 {} 收敛种子已落地: {} (max_relc={:?})",
|
||
name,
|
||
seed_path.display(),
|
||
max_relc
|
||
);
|
||
let _ = state
|
||
.db
|
||
.insert_seed_named(&name, ¶ms, &seed_path.to_string_lossy())
|
||
.await;
|
||
}
|
||
} else {
|
||
warn!(
|
||
"历史种子导入：网格点 {} 声称收敛但未上传 seed_file，跳过种子写入",
|
||
name
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
// 4. 更新 grid_points 状态：收敛→converged(success_method='imported')；否则维持 pending
|
||
// 让正常调度处理（导入未收敛点无意义，但记录其尝试）。
|
||
if converged {
|
||
if let Err(e) = state
|
||
.db
|
||
.mark_grid_point_imported(&name, &workflow_name, Some(summary.elapsed_sec))
|
||
.await
|
||
{
|
||
warn!("历史种子导入：标记 {} 为 converged 失败: {}", name, e);
|
||
}
|
||
}
|
||
|
||
info!(
|
||
"历史种子导入完成：网格点 {} (workflow={}, converged={}, max_relc={:?})",
|
||
name, workflow_name, converged, max_relc
|
||
);
|
||
|
||
Ok((
|
||
StatusCode::OK,
|
||
Json(json!({
|
||
"status": "ok",
|
||
"point_name": name,
|
||
"converged": converged,
|
||
"max_relc": max_relc,
|
||
})),
|
||
))
|
||
}
|