From 26d9897c26d7ecc65677780b595b7e6385b19da8 Mon Sep 17 00:00:00 2001 From: imengying Date: Tue, 8 Sep 2026 02:35:52 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E8=BF=9C=E7=A8=8B=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 + backend/src/db/queries.rs | 32 +------- backend/src/main.rs | 13 +++- backend/src/routes/remote.rs | 28 ++++++- backend/src/websocket/agent.rs | 85 +++++++++++++++----- frontend/src/api.ts | 6 +- frontend/src/components/AdminPanel.tsx | 36 ++++++--- frontend/src/refresh.ts | 1 + scripts/agent-runtime-test.mjs | 37 ++++++++- scripts/websocket-smoke.mjs | 103 +++++++++++++++++++++++-- 10 files changed, 269 insertions(+), 74 deletions(-) diff --git a/README.md b/README.md index c1145b9..e0d98ea 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,8 @@ Windows 下载加速参数为 `-Mirror https://ghproxy.net`。更新会校验 Re Linux systemd 下，请通过 SSH 或本机终端重装、手动更新或卸载 Agent，不要在该 Agent 自己的“远程执行”中操作；安装脚本会在停止服务前拒绝这种调用，避免脚本随服务一起被终止。服务启动后会连续检查 10 秒，进程退出或反复重启均视为失败。 +远程执行是异步批量任务：只向当前连接的 Agent 下发命令，离线节点直接报告失败，重连后不会自动重发旧命令。多行脚本保留原始内容，执行结束后一次性返回输出；Agent 对单条命令设有 10 分钟执行上限。离开后台或退出登录不会停止已下发的命令。结果每 2 秒查询一次，持续 1 分钟后暂停自动刷新，可点击“刷新结果”继续查询；暂停查询不代表命令执行失败或已停止。 + 实时上/下行速率由 Agent 按“收发字节增量 / 实际采样秒数”计算，前台每秒合并展示最新样本，无新样本时保持原值。节点实时采样间隔默认为 1 秒；设置更长间隔时不会凭空生成每秒数据。历史保存间隔独立，默认 60 秒。 节点价格最低为 `0`，表示免费；不再接受负价格。是否隐藏节点仅由“隐藏节点”开关决定。 diff --git a/backend/src/db/queries.rs b/backend/src/db/queries.rs index 67273bf..6feb790 100644 --- a/backend/src/db/queries.rs +++ b/backend/src/db/queries.rs @@ -1293,7 +1293,7 @@ async fn cleanup_database_with_budget( let affected = if index == 0 { sqlx::query(db.sql( "UPDATE remote_tasks SET status='failed', completed_at=?, \ - result=CASE WHEN result='' THEN '任务等待超过 24 小时，已自动取消' ELSE result END \ + result=CASE WHEN result='' THEN '超过 24 小时未收到执行结果，无法确认命令状态' ELSE result END \ WHERE status IN ('pending','sent') AND id IN ( \ SELECT id FROM remote_tasks WHERE status IN ('pending','sent') AND requested_at Result Result> { - let current = now(); - let cutoff = current - REMOTE_TASK_ACTIVE_TTL_SECONDS; - let mut transaction = db.pool().begin().await?; - sqlx::query(db.sql( - "UPDATE remote_tasks SET status='failed', completed_at=?, \ - result=CASE WHEN result='' THEN '任务等待超过 24 小时，已自动取消' ELSE result END \ - WHERE server_id=? AND status IN ('pending','sent') AND requested_at, _>>() - .map_err(Into::into) -} - fn remote_task_from_row( row: &sqlx::any::AnyRow, ) -> std::result::Result { diff --git a/backend/src/main.rs b/backend/src/main.rs index 5e04f5e..914b3c9 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -129,10 +129,15 @@ impl AppState { seconds as u64 * 1000 } - pub async fn send_remote_task(&self, task: &models::RemoteTaskInfo) { - if let Some(connection) = self.agents.read().await.get(&task.server_id) { - websocket::agent::queue_remote_task(connection, task); - } + pub async fn send_remote_task( + &self, + task: &models::RemoteTaskInfo, + ) -> std::result::Result<(), &'static str> { + let agents = self.agents.read().await; + let connection = agents + .get(&task.server_id) + .ok_or("节点未连接，命令未下发")?; + websocket::agent::queue_remote_task(connection, task) } } diff --git a/backend/src/routes/remote.rs b/backend/src/routes/remote.rs index fe56222..3909b87 100644 --- a/backend/src/routes/remote.rs +++ b/backend/src/routes/remote.rs @@ -57,12 +57,12 @@ pub async fn create_task( format!("验证码尝试过多，请在 {seconds} 秒后重试"), )); } - let command = input.command.trim(); + let command = input.command.as_str(); let server_ids = remote_server_ids(&input); if server_ids.is_empty() || server_ids.len() > MAX_REMOTE_TARGETS || server_ids.iter().any(|id| id.len() > 80) - || command.is_empty() + || command.trim().is_empty() || command.len() > 16_384 { return Err(ApiResponse::bad_request("请选择 1 至 128 个节点并填写命令")); @@ -93,13 +93,35 @@ pub async fn create_task( return Err(ApiResponse::not_found(format!("节点不存在: {missing}"))); } + { + let agents = state.agents.read().await; + if !server_ids.iter().any(|id| { + agents + .get(id) + .is_some_and(|agent| !agent.sender.is_closed()) + }) { + return Err(ApiResponse::bad_request("所选节点均未连接，命令未下发")); + } + } + let mut tasks = Vec::with_capacity(server_ids.len()); for server_id in server_ids { let task = crate::db::queries::create_remote_task(&state.db, &server_id, command, &user.username) .await .map_err(ApiResponse::internal)?; - state.send_remote_task(&task).await; + if let Err(reason) = state.send_remote_task(&task).await { + crate::db::queries::update_remote_task_result( + &state.db, + &server_id, + &task.id, + "failed", + reason, + Some(-1), + ) + .await + .map_err(ApiResponse::internal)?; + } tasks.push(CreatedRemoteTask { server_id, task_id: task.id, diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index e17f37e..f0711cb 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -154,13 +154,8 @@ async fn run( .to_string(), )); } - if let Ok(tasks) = - crate::db::queries::pending_remote_tasks(&state.db, &identity.server_id).await - { - for task in tasks { - send_task(&outbound_tx, &task); - } - } + // Commands are dispatched only to the current connection. Reconnecting must + // not replay an old command; the Agent retries its persisted results itself. drop(activity); let writer = tokio::spawn(async move { @@ -358,17 +353,6 @@ async fn handle_text( } } -fn send_task(outbound: &mpsc::Sender, task: &RemoteTaskInfo) { - let _ = outbound.try_send(AgentCommand::Text( - serde_json::json!({ - "type": "remote_task", - "task_id": task.id, - "command": task.command, - }) - .to_string(), - )); -} - fn send_ack( outbound: &mpsc::Sender, result: &PersistResult, @@ -476,8 +460,24 @@ fn agent_protocol_supported(headers: &HeaderMap) -> bool { .all(|capability| capabilities.contains(capability)) } -pub fn queue_remote_task(connection: &AgentConnection, task: &RemoteTaskInfo) { - send_task(&connection.sender, task); +pub fn queue_remote_task( + connection: &AgentConnection, + task: &RemoteTaskInfo, +) -> Result<(), &'static str> { + connection + .sender + .try_send(AgentCommand::Text( + serde_json::json!({ + "type": "remote_task", + "task_id": task.id, + "command": task.command, + }) + .to_string(), + )) + .map_err(|error| match error { + mpsc::error::TrySendError::Full(_) => "节点发送队列已满，命令未下发，请稍后重试", + mpsc::error::TrySendError::Closed(_) => "节点连接已断开，命令未下发", + }) } #[cfg(test)] @@ -485,6 +485,51 @@ mod tests { use super::*; use axum::http::HeaderValue; + #[test] + fn remote_dispatch_reports_backpressure_and_disconnects() { + let (sender, mut receiver) = mpsc::channel(1); + let connection = AgentConnection { + connection_id: "test".to_string(), + sender, + report_interval: 60, + collect_interval: 1, + live_until: Arc::new(AtomicI64::new(0)), + }; + let task = RemoteTaskInfo { + id: uuid::Uuid::new_v4().to_string(), + server_id: "test-server".to_string(), + command: "\n printf test\n ".to_string(), + status: "pending".to_string(), + requested_by: "admin".to_string(), + requested_at: 0, + started_at: None, + completed_at: None, + result: String::new(), + exit_code: None, + }; + + queue_remote_task(&connection, &task).unwrap(); + assert!( + queue_remote_task(&connection, &task) + .unwrap_err() + .contains("队列已满") + ); + let AgentCommand::Text(payload) = receiver.try_recv().unwrap() else { + panic!("expected a remote command"); + }; + let payload: serde_json::Value = serde_json::from_str(&payload).unwrap(); + assert_eq!(payload["command"], task.command); + assert_eq!(payload["task_id"], task.id); + assert!(receiver.try_recv().is_err()); + + drop(receiver); + assert!( + queue_remote_task(&connection, &task) + .unwrap_err() + .contains("连接已断开") + ); + } + #[test] fn requires_the_current_agent_protocol_and_capabilities() { let mut headers = HeaderMap::new(); diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 07d5932..57b6d6e 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -185,6 +185,10 @@ export const api = { createRemoteTask: (input: RemoteTaskInput) => request<{ tasks: RemoteTaskCreated[] }>("/api/admin/remote/task", { method: "POST", body: JSON.stringify(input), + signal: AbortSignal.timeout(15_000), + }, true), + remoteTask: (id: string) => request(`/api/admin/remote/task/${encodeURIComponent(id)}`, { + cache: "no-store", + signal: AbortSignal.timeout(15_000), }, true), - remoteTask: (id: string) => request(`/api/admin/remote/task/${encodeURIComponent(id)}`, {}, true), }; diff --git a/frontend/src/components/AdminPanel.tsx b/frontend/src/components/AdminPanel.tsx index 2fafdc0..6b80165 100644 --- a/frontend/src/components/AdminPanel.tsx +++ b/frontend/src/components/AdminPanel.tsx @@ -40,7 +40,7 @@ import { ADMIN_UNAUTHORIZED_EVENT, api, ApiError } from "../api"; import { adminTabFromPath, adminTabPaths, canonicalAdminPath, type AdminTab } from "../adminRoutes"; import { formatBytes, formatByteSize, parseByteSize } from "../format"; import { derivePassword } from "../password"; -import { hasActiveRemoteTasks, isRemoteTaskActive, REMOTE_TASK_POLL_INTERVAL_MS } from "../refresh"; +import { hasActiveRemoteTasks, isRemoteTaskActive, REMOTE_TASK_POLL_INTERVAL_MS, REMOTE_TASK_POLL_TIMEOUT_MS } from "../refresh"; import { ASSET_CURRENCIES, type AdminServer, type Config, type DatabaseMigrationResult, type DatabaseStats, type ExchangeRates, type LoginSession, type RemoteTask, type ServerInput, type Settings, type Theme, type ThemeSettingField, type ThemeSettingsSchema, type ThemeSettingValue, type TotpSetup, type TotpStatus } from "../types"; import { Checkbox } from "./Checkbox"; import { TurnstileWidget } from "./TurnstileWidget"; @@ -308,6 +308,7 @@ export function AdminPanel({ const [remoteQuery, setRemoteQuery] = useState(""); const [remoteCommand, setRemoteCommand] = useState(""); const [remoteTasks, setRemoteTasks] = useState([]); + const [remotePollingUntil, setRemotePollingUntil] = useState(0); const [twoFactorStatus, setTwoFactorStatus] = useState(null); const [twoFactorSetup, setTwoFactorSetup] = useState(null); const [newPasswordConfirmation, setNewPasswordConfirmation] = useState(""); @@ -837,6 +838,9 @@ export function AdminPanel({ setTwoFactorSecretCopied(false); setLoginSessions([]); setLoginSessionsLoaded(false); setRevokingSessionId(""); setRemoteSelectedIds([]); setRemoteCommand(""); setRemoteTasks([]); setRemoteQuery(""); + remoteTasksRequestRef.current += 1; + remoteTasksRef.current = []; + setRemotePollingUntil(0); } } @@ -985,7 +989,7 @@ export function AdminPanel({ try { const code = await verificationDialog.ask("确认执行远程命令", true); if (code === null) return; - const command = remoteCommand.trim(); + const command = remoteCommand; const data = await api.createRemoteTask({ server_ids: remoteSelectedIds, command, @@ -1007,21 +1011,29 @@ export function AdminPanel({ remoteTasksRequestRef.current += 1; remoteTasksRef.current = tasks; setRemoteTasks(tasks); + setRemotePollingUntil(Date.now() + REMOTE_TASK_POLL_TIMEOUT_MS); setRemoteCommand(""); - setNotice(`已向 ${tasks.length} 台服务器下发命令`); + setNotice("已提交命令，请查看各节点的执行结果"); await refreshRemoteTasks(true, true); } catch (err) { - setError(err instanceof Error ? err.message : "下发命令失败"); + setError(err instanceof Error && err.name === "TimeoutError" + ? "下发请求超时，命令可能已提交，请先检查节点，避免重复执行" + : err instanceof Error ? err.message : "下发命令失败"); } finally { setBusy(false); } }; useEffect(() => { - if (!authenticated || tab !== "remote" || !remoteTasksActive) return; + if (!authenticated || tab !== "remote" || !remoteTasksActive || !remotePollingUntil) return; + if (Date.now() >= remotePollingUntil) { + setRemotePollingUntil(0); + return; + } let stopped = false; let running = false; let timer: number | undefined; + const timeout = window.setTimeout(() => setRemotePollingUntil(0), remotePollingUntil - Date.now()); const canPoll = () => !document.hidden && navigator.onLine !== false; const clearTimer = () => { @@ -1039,6 +1051,10 @@ export function AdminPanel({ const run = async () => { clearTimer(); if (stopped || running || !canPoll()) return; + if (Date.now() >= remotePollingUntil) { + setRemotePollingUntil(0); + return; + } running = true; try { await refreshRemoteTasks(true); @@ -1064,11 +1080,12 @@ export function AdminPanel({ return () => { stopped = true; clearTimer(); + window.clearTimeout(timeout); document.removeEventListener("visibilitychange", handleVisibility); window.removeEventListener("online", resume); window.removeEventListener("offline", pause); }; - }, [authenticated, refreshRemoteTasks, remoteTasksActive, tab]); + }, [authenticated, refreshRemoteTasks, remoteTasksActive, remotePollingUntil, tab]); if (!authChecked) { return
; @@ -1243,7 +1260,7 @@ export function AdminPanel({