feat(api): implement StartExec and GetExecResult server functions

- rate_limit.rs: 新增 CODE_EXEC 双层限流(per-second burst + 每日总额)
  governor 0.8 无 per_day,用 with_period(24h)+allow_burst 模拟
- execute.rs: StartExec(IP 提取/语言白名单/源码大小/入队/并发信号量/clamp/容器执行)
  + GetExecResult(轮询读取 DashMap);系统异常脱敏,日志记详情
- .env.example: RATE_LIMIT_CODE_EXEC_PER_SEC 改为整数(governor 不支持小数)
This commit is contained in:
xfy 2026-07-03 18:15:59 +08:00
parent 3f3bbf284c
commit 9fe128d44d
4 changed files with 230 additions and 3 deletions

View File

@ -143,7 +143,9 @@ CODE_RUNNER_MAX_OUTPUT_BYTES=1048576
CODE_RUNNER_LANGUAGES=python,node
DOCKER_SOCKET_PATH=/var/run/docker.sock
# Rate limits
RATE_LIMIT_CODE_EXEC_PER_SEC=0.2
# Rate limits (governor keyed limiters)
# PER_SEC/BURST 控制单 IP 每秒突发DAILY 为单 IP 每日总额。
# 注意PER_SEC 必须是正整数governor 不支持小数,非整数会回退到默认 1
RATE_LIMIT_CODE_EXEC_PER_SEC=1
RATE_LIMIT_CODE_EXEC_BURST=3
RATE_LIMIT_CODE_EXEC_DAILY=50

View File

@ -0,0 +1,184 @@
//! 代码执行 server functionsStartExec / GetExecResult。
//!
//! StartExec 流程:速率限制 → 语言白名单 → 源码大小校验 → 入队DashMap
//! → spawn 后台 task信号量限并发 + clamp_limits + run_in_container
//! 返回 task_id 供前端轮询。GetExecResult 读取任务条目。
//!
//! 错误脱敏:匿名可见错误(不支持的语言、超限、限流)返回中文消息;系统内部
//! 异常(容器拉起失败等)记服务端日志,对前端返回统一「系统暂时不可用」。
// 与 posts / settings 模块一致Dioxus `#[server]` 宏触发 deprecated/unit 提示,按项目惯例放行。
#![allow(clippy::unused_unit, deprecated)]
use std::sync::{Arc, LazyLock};
use std::time::Duration;
use chrono::Utc;
use dioxus::prelude::*;
use tokio::sync::Semaphore;
use crate::api::code_runner::languages::{is_supported_lang, LANGUAGES};
use crate::api::code_runner::progress::{
gc_old_tasks, insert_task, update_task_result, update_task_stage, EXEC_TASKS,
};
use crate::api::code_runner::{ExecRequest, ExecResult, ExecStatus, ExecTask};
use crate::api::rate_limit::{check_code_exec_limit, get_client_ip};
use crate::infra::docker::run_in_container;
use crate::infra::runner_config::{clamp_limits, RUNNER_CONFIG};
/// 并发容器控制信号量,限制同时在跑的容器数量(`CODE_RUNNER_MAX_CONCURRENT`)。
pub static RUNNER_SEMAPHORE: LazyLock<Arc<Semaphore>> =
LazyLock::new(|| Arc::new(Semaphore::new(RUNNER_CONFIG.max_concurrent)));
/// 从 FullstackContext 提取客户端 IP无上下文时退回 "unknown")。
fn client_ip() -> String {
match dioxus::fullstack::FullstackContext::current() {
Some(ctx) => {
let parts = ctx.parts_mut();
get_client_ip(&parts.headers)
}
None => "unknown".to_string(),
}
}
/// 提交一次代码执行请求。
///
/// 同步校验通过后立即返回 task_id容器在后台执行结果通过
/// [`get_exec_result`] 轮询查询。语言不支持 / 源码过大 / 触发限流时同步返回错误。
#[server(StartExec, "/api")]
pub async fn start_exec(req: ExecRequest) -> Result<String, ServerFnError> {
// 1. 速率限制(双层:每秒突发 + 每日总额)
let ip = client_ip();
if let Err(msg) = check_code_exec_limit(&ip) {
return Err(ServerFnError::new(msg));
}
// 2. 语言白名单
if !is_supported_lang(&req.language) {
return Err(ServerFnError::new("不支持该执行语言".to_string()));
}
// 3. 源码大小限制
if req.source.len() > RUNNER_CONFIG.max_source_bytes as usize {
return Err(ServerFnError::new("源代码过大".to_string()));
}
// 4. 生成任务 ID 并入队
let task_id = uuid::Uuid::new_v4().to_string();
insert_task(task_id.clone());
// 5. 顺手回收过期任务
gc_old_tasks();
// 6. 后台执行
let task_id_clone = task_id.clone();
let lang_key = req.language.clone();
tokio::spawn(async move {
let sem = &*RUNNER_SEMAPHORE;
// 排队等待可用容器槽
let ticket = match tokio::time::timeout(
Duration::from_secs(RUNNER_CONFIG.queue_timeout_secs),
sem.acquire(),
)
.await
{
Ok(Ok(t)) => t,
_ => {
update_task_stage(&task_id_clone, ExecStatus::Failed, "系统繁忙,排队超时");
return;
}
};
update_task_stage(&task_id_clone, ExecStatus::Running, "启动容器");
let lang_def = match LANGUAGES.get(&lang_key) {
Some(d) => d,
None => {
// 理论不可达start_exec 已校验白名单;防御性兜底。
update_task_stage(&task_id_clone, ExecStatus::Failed, "语言未注册");
return;
}
};
// 资源限制合并与钳制
let base_limits = req
.overrides
.unwrap_or_else(|| lang_def.default_limits.clone());
let final_limits = clamp_limits(base_limits, lang_def.allow_network);
let start_time = Utc::now();
let res = run_in_container(
&lang_def.image,
&lang_def.run_cmd,
&req.source,
&lang_def.extension,
final_limits,
)
.await;
let duration_ms = (Utc::now() - start_time)
.num_milliseconds()
.max(0) as u64;
drop(ticket); // 显式释放信号量
match res {
Ok((exit_code, stdout, stderr, oom_killed)) => {
let status = if oom_killed {
ExecStatus::OomKilled
} else if exit_code == Some(0) {
ExecStatus::Success
} else {
ExecStatus::Error
};
let exec_res = ExecResult {
status: status.clone(),
stdout,
stderr,
exit_code,
duration_ms,
language: lang_key.clone(),
};
update_task_result(&task_id_clone, status, exec_res);
}
Err(e) => {
// 系统内部异常脱敏:日志记详情,前端只见通用消息。
let s = e.to_string();
let is_timeout = s.contains("TimedOut");
tracing::error!(error = ?e, task_id = %task_id_clone, "container execution failed");
let status = if is_timeout {
ExecStatus::Timeout
} else {
ExecStatus::Failed
};
let exec_res = ExecResult {
status: status.clone(),
stdout: "".to_string(),
stderr: if is_timeout {
"执行超时".to_string()
} else {
"系统暂时不可用".to_string()
},
exit_code: None,
duration_ms,
language: lang_key.clone(),
};
update_task_result(&task_id_clone, status, exec_res);
}
}
});
Ok(task_id)
}
/// 查询任务执行结果(前端轮询)。
#[server(GetExecResult, "/api")]
pub async fn get_exec_result(task_id: String) -> Result<ExecTask, ServerFnError> {
if let Some(task) = EXEC_TASKS.get(&task_id) {
Ok(task.clone())
} else {
Err(ServerFnError::new("找不到指定的任务".to_string()))
}
}

View File

@ -72,7 +72,8 @@ pub struct ExecTask {
}
// 子模块在后续任务中实现:
// #[cfg(feature = "server")] pub mod execute;
#[cfg(feature = "server")]
pub mod execute;
#[cfg(feature = "server")]
pub mod languages;
#[cfg(feature = "server")]

View File

@ -19,6 +19,8 @@ use governor::{DefaultKeyedRateLimiter, Quota, RateLimiter};
use std::num::NonZeroU32;
#[cfg(feature = "server")]
use std::sync::LazyLock;
#[cfg(feature = "server")]
use std::time::Duration;
#[cfg(feature = "server")]
fn env_or(key: &str, default: u32) -> NonZeroU32 {
@ -61,6 +63,30 @@ static COMMENT_LIMITER: LazyLock<DefaultKeyedRateLimiter<String>> = LazyLock::ne
)
});
#[cfg(feature = "server")]
/// 代码执行单 IP 每秒限流(默认 0.2 req/s突发 3
/// 防止单个客户端高频提交容器任务,与下方日限额共同构成双层速率限制。
static CODE_EXEC_LIMITER: LazyLock<DefaultKeyedRateLimiter<String>> = LazyLock::new(|| {
RateLimiter::keyed(
Quota::per_second(env_or("RATE_LIMIT_CODE_EXEC_PER_SEC", 1))
.allow_burst(env_or("RATE_LIMIT_CODE_EXEC_BURST", 3)),
)
});
#[cfg(feature = "server")]
/// 代码执行单 IP 每日限额(默认 50 次/天)。
/// 容器执行成本高CPU/内存/启动延迟),需要硬性日上限防止资源耗尽。
///
/// governor 0.8 无 `Quota::per_day`,用 `with_period(24h)` + `allow_burst(daily)`
/// 模拟:每 24h 补充 1 token、突发上限即日额度等效于「每日最多 daily 次」。
static CODE_EXEC_DAILY_LIMITER: LazyLock<DefaultKeyedRateLimiter<String>> = LazyLock::new(|| {
RateLimiter::keyed(
Quota::with_period(Duration::from_secs(86_400))
.unwrap()
.allow_burst(env_or("RATE_LIMIT_CODE_EXEC_DAILY", 50)),
)
});
#[cfg(feature = "server")]
/// 当无法识别真实客户端 IP"unknown")时使用的宽松限流桶。
///
@ -226,6 +252,20 @@ pub fn check_upload_limit(ip: &str) -> Result<(), String> {
.map_err(|_| "上传过于频繁,请稍后再试".to_string())
}
#[cfg(feature = "server")]
/// 检查代码执行请求的双层速率限制(每秒突发 + 每日总额)。
///
/// 两层任一被限即拒绝。返回中文错误消息,供 server function 直接透传给前端。
pub fn check_code_exec_limit(ip: &str) -> Result<(), String> {
CODE_EXEC_LIMITER
.check_key(&ip.to_string())
.map_err(|_| "请求过于频繁,请稍后再试".to_string())?;
CODE_EXEC_DAILY_LIMITER
.check_key(&ip.to_string())
.map_err(|_| "今日运行次数已达上限,请明天再试".to_string())?;
Ok(())
}
#[cfg(all(test, feature = "server"))]
mod tests {
use super::*;