From ab8f359f93d2c2117c05ee73935e299ae76ab050 Mon Sep 17 00:00:00 2001 From: imengying Date: Tue, 8 Sep 2026 01:15:34 +0800 Subject: [PATCH] Fixes and improvements --- .github/workflows/release.yml | 9 +- README.md | 6 +- agent/agent.sh | 46 +++++++- agent/install-freebsd.sh | 5 +- agent/install-macos.sh | 5 +- agent/src/main.rs | 132 ++++++++++++++++----- backend/src/activity.rs | 86 ++++++++++++++ backend/src/backup.rs | 148 +++++++++++++++++++----- backend/src/main.rs | 33 ++++-- backend/src/middleware.rs | 64 ++++++++++- backend/src/notify.rs | 110 +++++++++++++++++- backend/src/routes/admin.rs | 46 +++----- backend/src/websocket/agent.rs | 33 ++++-- frontend/src/api.ts | 5 +- frontend/src/components/AdminPanel.tsx | 8 +- frontend/src/install.test.ts | 112 ++++++++++++++++++ frontend/src/refresh.ts | 2 +- frontend/src/styles/admin.css | 1 + install.sh | 5 +- scripts/agent-runtime-test.mjs | 153 +++++++++++++++++++++++++ scripts/smoke-test.sh | 23 +++- 21 files changed, 901 insertions(+), 131 deletions(-) create mode 100644 backend/src/activity.rs create mode 100644 scripts/agent-runtime-test.mjs diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f9baa52..a303b62 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -103,7 +103,10 @@ jobs: - name: Build frontend run: bun run build:frontend - name: Check shell scripts - run: sh -n install.sh agent/agent.sh agent/install-macos.sh agent/install-freebsd.sh scripts/*.sh start.sh dev.sh + run: | + for script in install.sh agent/agent.sh agent/install-macos.sh agent/install-freebsd.sh scripts/*.sh start.sh dev.sh; do + sh -n "$script" || exit "$?" + done - name: Check PowerShell scripts shell: pwsh run: | @@ -129,6 +132,8 @@ jobs: run: CARGO_TARGET_DIR=target/agent cargo test --locked --manifest-path agent/Cargo.toml - name: Build Agent for smoke test run: CARGO_TARGET_DIR=target/agent cargo build --locked --manifest-path agent/Cargo.toml + - name: Test Agent update and connection recovery + run: MONITOR_AGENT_BINARY=target/agent/debug/nodeflare-agent node scripts/agent-runtime-test.mjs - name: Build backend run: CARGO_TARGET_DIR=target/backend cargo build --locked --manifest-path backend/Cargo.toml - name: Smoke test SQLite and PostgreSQL @@ -208,6 +213,8 @@ jobs: MONITOR_ADMIN_USERNAME=sqlite-smoke \ MONITOR_ADMIN_PASSWORD=SmokePassword123 \ MONITOR_AGENT_BINARY="$GITHUB_WORKSPACE/target/agent/debug/nodeflare-agent" \ + MONITOR_MIGRATION_URL="sqlite://$postgres_dir/round-trip.db" \ + MONITOR_MIGRATION_KIND=sqlite \ sh scripts/smoke-test.sh server: diff --git a/README.md b/README.md index 933d9f3..c1145b9 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,8 @@ Unblock-File "$env:TEMP\nodeflare-agent-install.ps1" Windows 下载加速参数为 `-Mirror https://ghproxy.net`。更新会校验 Release 摘要和程序版本，启动失败会回滚；无需重新输入 Token。 +Linux systemd 下，请通过 SSH 或本机终端重装、手动更新或卸载 Agent，不要在该 Agent 自己的“远程执行”中操作；安装脚本会在停止服务前拒绝这种调用，避免脚本随服务一起被终止。服务启动后会连续检查 10 秒，进程退出或反复重启均视为失败。 + 实时上/下行速率由 Agent 按“收发字节增量 / 实际采样秒数”计算，前台每秒合并展示最新样本，无新样本时保持原值。节点实时采样间隔默认为 1 秒；设置更长间隔时不会凭空生成每秒数据。历史保存间隔独立，默认 60 秒。 节点价格最低为 `0`，表示免费；不再接受负价格。是否隐藏节点仅由“隐藏节点”开关决定。 @@ -167,8 +169,10 @@ database_url = "postgres://nodeflare:password@127.0.0.1:5432/nodeflare?sslmode=d - 备份含设置、节点、监控历史、通知、主题（含文件）、远程任务与安全配置；会话等临时数据不导出，恢复后需重新登录。 - 导出、恢复、迁移前需验证当前 TOTP，未启用时为管理员密码。 -- 迁移会覆盖目标库已有数据并自动更新 `database_url`，完成后重启生效。 +- 迁移会覆盖目标库已有数据并自动更新 `database_url`，保留全部 Agent Token（含额外安装 Token）。迁移完成后暂停业务写入，须重启服务才恢复；期间仍可登录后台执行重启。 - 节点主 Token 保留；“下载 Agent”额外生成的安装 Token 不进入数据库备份，使用这些 Token 的节点恢复后需重新生成并执行安装命令，单独 `--update` 不会更换 Token。 +- ZIP 备份上限为 512 MiB，解压后不超过 4 GiB；导出也会检查这些上限，超限时请减少历史数据后重试。 +- 新节点首次有效上报前不发送离线告警；后续按最近实时上报时间判断，后端重启后会等待一个离线告警周期供 Agent 重连。 - 历史指标默认保留 30 天，过期数据自动清理；完整空间回收在后台手动触发。 - 升级建议先更新服务端再更新 Agent；恢复备份建议使用相同版本。 diff --git a/agent/agent.sh b/agent/agent.sh index efb511c..b7818ae 100644 --- a/agent/agent.sh +++ b/agent/agent.sh @@ -131,6 +131,31 @@ detect_init_system() { fi } +ensure_not_agent_service() { + [ "$init_system" = systemd ] || return 0 + [ -r "/proc/$$/cgroup" ] || fail "无法检查当前进程所属服务，已停止安装" + while IFS=: read -r cgroup_id cgroup_controllers cgroup_path; do + case "$cgroup_path/" in + */"$SERVICE_NAME.service"/*) + fail "不能在当前 Agent 的远程执行中重装、更新或卸载自身，请通过 SSH 执行；服务未停止" + ;; + esac + done < "/proc/$$/cgroup" +} + +verify_agent_started() { + started_pid=$(systemctl show -p MainPID --value "$SERVICE_NAME") || return 1 + case "$started_pid" in ''|*[!0-9]*|0) return 1 ;; esac + attempt=1 + while [ "$attempt" -le 10 ]; do + sleep 1 + systemctl is-active --quiet "$SERVICE_NAME" || return 1 + current_pid=$(systemctl show -p MainPID --value "$SERVICE_NAME") || return 1 + [ "$current_pid" = "$started_pid" ] || return 1 + attempt=$((attempt + 1)) + done +} + ensure_curl() { command -v curl >/dev/null 2>&1 && return log "未找到 curl，正在通过系统包管理器安装" @@ -225,13 +250,17 @@ install_agent() { fi init_system=$(detect_init_system) [ "$init_system" != "unknown" ] || fail "未检测到正在运行的 systemd 或 OpenRC" + ensure_not_agent_service mkdir -p "$INSTALL_DIR" "$STATE_ROOT" "$STATE_DIR" chmod 755 "$INSTALL_DIR" chmod 700 "$STATE_ROOT" chmod 750 "$STATE_DIR" temporary="$INSTALL_DIR/.agent.$$.download" - trap cleanup_agent_install EXIT HUP INT TERM + trap cleanup_agent_install EXIT + trap 'exit 129' HUP + trap 'exit 130' INT + trap 'exit 143' TERM artifact="agent-linux-$arch-$libc" release_api="https://api.github.com/repos/imengying/NodeFlare/releases/latest" log "正在获取 GitHub 最新正式版本（$artifact）" @@ -302,8 +331,16 @@ install_agent() { fi rollback_agent=true case "$init_system" in - systemd) systemctl stop "$SERVICE_NAME" 2>/dev/null || true ;; - openrc) rc-service "$SERVICE_NAME" stop 2>/dev/null || true ;; + systemd) + systemctl stop "$SERVICE_NAME" 2>/dev/null || { + [ "$had_service" = false ] || fail "无法停止已有 Agent 服务，未替换程序" + } + ;; + openrc) + rc-service "$SERVICE_NAME" stop 2>/dev/null || { + [ "$had_service" = false ] || fail "无法停止已有 Agent 服务，未替换程序" + } + ;; esac mv "$temporary" "$AGENT_FILE" log "正在启动 $init_system 服务" @@ -328,7 +365,7 @@ install_agent() { chmod 600 "$SERVICE_FILE" systemctl daemon-reload systemctl enable --now --quiet "$SERVICE_NAME" - systemctl is-active --quiet "$SERVICE_NAME" || { + verify_agent_started || { systemctl status "$SERVICE_NAME" --no-pager >&2 || true fail "服务启动失败，请查看上方状态信息" } @@ -429,6 +466,7 @@ uninstall_agent() { [ "$(id -u)" -eq 0 ] || fail "请使用 root 权限执行卸载" log "正在停止并移除 Agent 服务" init_system=$(detect_init_system) + ensure_not_agent_service case "$init_system" in systemd) systemctl disable --now "$SERVICE_NAME" 2>/dev/null || true ;; openrc) diff --git a/agent/install-freebsd.sh b/agent/install-freebsd.sh index 338cf24..82da72f 100755 --- a/agent/install-freebsd.sh +++ b/agent/install-freebsd.sh @@ -213,7 +213,10 @@ chmod 755 "$INSTALL_DIR" chmod 750 "$STATE_ROOT" chmod 750 "$STATE_DIR" temporary="$INSTALL_DIR/.agent.$$.download" -trap cleanup_agent_install EXIT HUP INT TERM +trap cleanup_agent_install EXIT +trap 'exit 129' HUP +trap 'exit 130' INT +trap 'exit 143' TERM artifact="agent-freebsd-$arch" release_api="https://api.github.com/repos/imengying/NodeFlare/releases/latest" log "正在获取 GitHub 最新正式版本（$artifact）" diff --git a/agent/install-macos.sh b/agent/install-macos.sh index ea2b620..160bbbf 100644 --- a/agent/install-macos.sh +++ b/agent/install-macos.sh @@ -187,7 +187,10 @@ chmod 755 "$INSTALL_DIR" chmod 750 "$STATE_ROOT" chmod 750 "$STATE_DIR" temporary="$INSTALL_DIR/.agent.$$.download" -trap cleanup_agent_install EXIT HUP INT TERM +trap cleanup_agent_install EXIT +trap 'exit 129' HUP +trap 'exit 130' INT +trap 'exit 143' TERM artifact="agent-macos-aarch64" release_api="https://api.github.com/repos/imengying/NodeFlare/releases/latest" log "正在获取 GitHub 最新正式版本（$artifact）" diff --git a/agent/src/main.rs b/agent/src/main.rs index 6f81e3b..cfc2c43 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -23,7 +23,7 @@ use sha2::{Digest, Sha256}; #[cfg(any(target_os = "windows", target_os = "macos", target_os = "freebsd"))] use sysinfo::{Disks, Networks, ProcessRefreshKind, ProcessesToUpdate, System}; use tungstenite::client::IntoClientRequest; -use tungstenite::{Error as WebSocketError, Message, connect}; +use tungstenite::{Error as WebSocketError, Message, client_tls}; #[cfg(not(any( target_os = "linux", @@ -57,6 +57,7 @@ const UPDATE_CHECK_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60); const UPDATE_RETRY_INTERVAL: Duration = Duration::from_secs(30 * 60); const UPDATE_CHECK_JITTER_MAX_SECONDS: u64 = 30 * 60; const LIVE_RECONNECT_DELAY: Duration = Duration::from_secs(3); +const LIVE_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const LIVE_QUEUE_CAPACITY: usize = 720; const MAX_PENDING_SPOOL_BYTES: u64 = 64 * 1024 * 1024; const MAX_PENDING_SPOOL_LINE_BYTES: usize = 1024 * 1024; @@ -1747,23 +1748,48 @@ fn report_batch_id(reports: &[Report]) -> String { type LiveSocket = tungstenite::WebSocket>; fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option)> { - let mut request = endpoint.into_client_request()?; - request - .headers_mut() - .insert("Authorization", format!("Bearer {token}").parse()?); - request - .headers_mut() - .insert("User-Agent", format!("nodeflare-agent/{VERSION}").parse()?); - request.headers_mut().insert( - "X-NodeFlare-Agent-Protocol", - AGENT_PROTOCOL_VERSION.parse()?, - ); - request.headers_mut().insert( - "X-NodeFlare-Agent-Capabilities", - AGENT_CAPABILITIES.parse()?, - ); let started_ms = unix_timestamp_millis(); - let (socket, response) = connect(request)?; + let deadline = Instant::now() + LIVE_CONNECT_TIMEOUT; + let mut url = url::Url::parse(endpoint)?; + let mut redirects = 0; + let (socket, response) = loop { + let mut request = url.as_str().into_client_request()?; + request + .headers_mut() + .insert("Authorization", format!("Bearer {token}").parse()?); + request + .headers_mut() + .insert("User-Agent", format!("nodeflare-agent/{VERSION}").parse()?); + request.headers_mut().insert( + "X-NodeFlare-Agent-Protocol", + AGENT_PROTOCOL_VERSION.parse()?, + ); + request.headers_mut().insert( + "X-NodeFlare-Agent-Capabilities", + AGENT_CAPABILITIES.parse()?, + ); + let stream = connect_live_stream(&url, deadline)?; + match client_tls(request, stream) { + Ok(connected) => break connected, + Err(tungstenite::HandshakeError::Failure(WebSocketError::Http(response))) + if response.status().is_redirection() && redirects < 3 => + { + let location = response + .headers() + .get("location") + .ok_or("WebSocket redirect has no location")? + .to_str()?; + let redirected = url.join(location)?; + // Never forward the Agent token to a different origin. + if redirected.origin() != url.origin() { + return Err("WebSocket redirect must stay on the configured origin".into()); + } + url = redirected; + redirects += 1; + } + Err(error) => return Err(format!("WebSocket handshake failed: {error}").into()), + } + }; let ended_ms = unix_timestamp_millis(); let clock_offset_ms = response .headers() @@ -1783,6 +1809,47 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option) Ok((socket, clock_offset_ms)) } +fn connect_live_stream(url: &url::Url, deadline: Instant) -> Result { + let host = url + .host_str() + .ok_or("WebSocket endpoint has no hostname")? + .trim_start_matches('[') + .trim_end_matches(']'); + let port = url + .port_or_known_default() + .ok_or("WebSocket endpoint has no port")?; + let addresses = (host, port).to_socket_addrs()?; + let mut connected = None; + let mut last_error = io::Error::new( + ErrorKind::NotFound, + "WebSocket hostname resolved to no addresses", + ); + for address in addresses { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err( + io::Error::new(ErrorKind::TimedOut, "WebSocket connection timed out").into(), + ); + } + match TcpStream::connect_timeout(&address, remaining.min(Duration::from_secs(3))) { + Ok(stream) => { + connected = Some(stream); + break; + } + Err(error) => last_error = error, + } + } + let stream = connected.ok_or(last_error)?; + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(io::Error::new(ErrorKind::TimedOut, "WebSocket connection timed out").into()); + } + stream.set_nodelay(true)?; + stream.set_read_timeout(Some(remaining))?; + stream.set_write_timeout(Some(remaining))?; + Ok(stream) +} + fn set_live_read_timeout(socket: &mut LiveSocket, timeout: Option) -> io::Result<()> { match socket.get_mut() { tungstenite::stream::MaybeTlsStream::Plain(stream) => stream.set_read_timeout(timeout), @@ -3331,6 +3398,23 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { { return Err("WSS report persistence acknowledgement timed out".into()); } + + // Check after ACK pruning, before the next sample refills the pending queue. + if !once + && !print_only + && config.auto_update + && pending_samples.is_empty() + && Instant::now() >= next_update_check + { + match update(&agent, &config.agent_mirror) { + Ok(true) => return Ok(()), + Ok(false) => next_update_check = Instant::now() + UPDATE_CHECK_INTERVAL, + Err(error) => { + eprintln!("agent update failed: {error}"); + next_update_check = Instant::now() + UPDATE_RETRY_INTERVAL; + } + } + } pending_results.extend(latency_executor.drain()); let current = Instant::now(); let due = config @@ -3415,20 +3499,6 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { ); } - if !once - && config.auto_update - && pending_samples.is_empty() - && Instant::now() >= next_update_check - { - match update(&agent, &config.agent_mirror) { - Ok(true) => return Ok(()), - Ok(false) => next_update_check = Instant::now() + UPDATE_CHECK_INTERVAL, - Err(error) => { - eprintln!("agent update failed: {error}"); - next_update_check = Instant::now() + UPDATE_RETRY_INTERVAL; - } - } - } if Instant::now() >= next_stats_log { stats.log_and_reset(); next_stats_log = Instant::now() + RUNTIME_STATS_INTERVAL; diff --git a/backend/src/activity.rs b/backend/src/activity.rs new file mode 100644 index 0000000..4fb2967 --- /dev/null +++ b/backend/src/activity.rs @@ -0,0 +1,86 @@ +use crate::routes::ApiResponse; +use axum::http::StatusCode; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; + +#[derive(Default)] +pub struct DatabaseActivity { + access: RwLock<()>, + restart_required: AtomicBool, +} + +impl DatabaseActivity { + pub fn read(&self) -> Result, ApiResponse> { + self.access.try_read().map_err(|_| { + ApiResponse::error( + StatusCode::SERVICE_UNAVAILABLE, + "数据库正在维护，请稍后重试", + ) + }) + } + + // Normal writers share access; maintenance drains and excludes all of them. + pub fn write(&self) -> Result, ApiResponse> { + let access = self.read()?; + self.ensure_writable()?; + Ok(access) + } + + pub async fn maintenance(&self) -> Result, ApiResponse> { + let access = self.access.write().await; + self.ensure_writable()?; + Ok(access) + } + + fn ensure_writable(&self) -> Result<(), ApiResponse> { + if self.restart_required() { + return Err(ApiResponse::error( + StatusCode::SERVICE_UNAVAILABLE, + "数据库迁移已完成，请重启服务后继续修改", + )); + } + Ok(()) + } + + pub fn require_restart(&self) { + self.restart_required.store(true, Ordering::Release); + } + + pub fn restart_required(&self) -> bool { + self.restart_required.load(Ordering::Acquire) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::poll; + use std::task::Poll; + + #[tokio::test] + async fn maintenance_drains_writes_and_rejects_new_access() { + let activity = DatabaseActivity::default(); + let writer = activity.write().unwrap(); + let maintenance = activity.maintenance(); + tokio::pin!(maintenance); + assert!(matches!(poll!(&mut maintenance), Poll::Pending)); + assert!(activity.read().is_err()); + assert!(activity.write().is_err()); + drop(writer); + let exclusive = maintenance.await.unwrap(); + assert!(activity.write().is_err()); + drop(exclusive); + assert!(activity.write().is_ok()); + } + + #[tokio::test] + async fn migration_keeps_business_writes_frozen_until_restart() { + let activity = DatabaseActivity::default(); + let exclusive = activity.maintenance().await.unwrap(); + activity.require_restart(); + drop(exclusive); + assert!(activity.read().is_ok()); + assert!(activity.write().is_err()); + assert!(activity.maintenance().await.is_err()); + } +} diff --git a/backend/src/backup.rs b/backend/src/backup.rs index 30112f6..2d50ff5 100644 --- a/backend/src/backup.rs +++ b/backend/src/backup.rs @@ -21,6 +21,27 @@ const INSERT_BATCH_ROWS: usize = 100; const MAX_NDJSON_LINE_BYTES: usize = 16 * 1024 * 1024; const THEME_BACKUP_FILE_MAX_BYTES: u64 = 32 * 1024 * 1024; +#[derive(Debug)] +pub struct BackupSizeExceeded; + +impl std::fmt::Display for BackupSizeExceeded { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .write_str("备份超过恢复上限（ZIP 512 MiB、解压后 4 GiB），请减少历史数据后重新备份") + } +} + +impl std::error::Error for BackupSizeExceeded {} + +fn validate_backup_size(compressed: u64, extracted: u64) -> Result<()> { + if compressed > DATABASE_BACKUP_MAX_BYTES as u64 + || extracted > DATABASE_BACKUP_MAX_EXTRACTED_BYTES + { + return Err(BackupSizeExceeded.into()); + } + Ok(()) +} + const BACKUP_TABLES: &[&str] = &[ "settings", "servers", @@ -248,9 +269,9 @@ async fn table_columns(db: &Database, table: &str) -> Result> .collect() } -async fn database_schema(db: &Database) -> Result>> { - let mut schema = Vec::with_capacity(BACKUP_TABLES.len()); - for table in BACKUP_TABLES { +async fn database_schema(db: &Database, tables: &[&str]) -> Result>> { + let mut schema = Vec::with_capacity(tables.len()); + for table in tables { schema.push(table_columns(db, table).await?); } Ok(schema) @@ -299,8 +320,12 @@ fn row_values(row: &AnyRow, columns: &[BackupColumn]) -> Result> { .collect() } -async fn export(db: &Database, theme_dir: Option<&Path>) -> Result { - let schema = database_schema(db).await?; +async fn export( + db: &Database, + theme_dir: Option<&Path>, + tables: &[&str], +) -> Result { + let schema = database_schema(db, tables).await?; let theme_files = collect_theme_files(theme_dir)?; let created_at = crate::db::now(); let manifest = BackupManifest { @@ -312,10 +337,7 @@ async fn export(db: &Database, theme_dir: Option<&Path>) -> Result "postgresql", } .to_string(), - tables: BACKUP_TABLES - .iter() - .map(|table| (*table).to_string()) - .collect(), + tables: tables.iter().map(|table| (*table).to_string()).collect(), theme_files: theme_files .iter() .map(|(relative, _, _)| relative.clone()) @@ -325,7 +347,10 @@ async fn export(db: &Database, theme_dir: Option<&Path>) -> Result) -> Result) -> Result) -> Result Result { - export(db, Some(theme_dir)).await + export(db, Some(theme_dir), BACKUP_TABLES).await } -fn validate_archive(archive: &mut ZipArchive) -> Result { +fn validate_archive( + archive: &mut ZipArchive, + tables: &[&str], +) -> Result { if archive.is_empty() || archive.len() > DATABASE_BACKUP_MAX_ENTRIES { bail!("备份 ZIP 文件条目数量无效"); } @@ -418,7 +450,7 @@ fn validate_archive(archive: &mut ZipArchive) -> Result>() @@ -426,7 +458,7 @@ fn validate_archive(archive: &mut ZipArchive) -> Result>(); @@ -454,9 +486,7 @@ fn validate_archive(archive: &mut ZipArchive) -> Result DATABASE_BACKUP_MAX_EXTRACTED_BYTES { - bail!("备份 ZIP 解压后不能超过 4 GiB"); - } + validate_backup_size(0, extracted_bytes)?; } if seen != expected { bail!("备份 ZIP 缺少必要文件"); @@ -792,15 +822,16 @@ async fn restore( db: &Database, source: R, theme_dir: Option<&Path>, + tables: &[&str], ) -> Result { let mut archive = ZipArchive::new(source).context("文件不是有效的 ZIP")?; - let manifest = validate_archive(&mut archive)?; + let manifest = validate_archive(&mut archive, tables)?; let staged_themes = match theme_dir { Some(theme_dir) => Some(extract_theme_files(&mut archive, &manifest, theme_dir)?), None if manifest.theme_files.is_empty() => None, None => bail!("数据库迁移备份不能包含主题文件"), }; - let schema = database_schema(db).await?; + let schema = database_schema(db, tables).await?; let mut transaction = db.pool().begin().await?; if db.is_postgres() { @@ -816,7 +847,7 @@ async fn restore( } let mut restored = 0_usize; - for (table, columns) in BACKUP_TABLES.iter().zip(schema.iter()) { + for (table, columns) in tables.iter().zip(schema.iter()) { restored += restore_table(&mut archive, db, &mut transaction, table, columns).await?; } // Current-format backups can contain snapshot rows with unset aggregates. @@ -878,18 +909,29 @@ pub async fn restore_archive(db: &Database, theme_dir: &Path, archive: &[u8]) -> if archive.is_empty() || archive.len() > DATABASE_BACKUP_MAX_BYTES { bail!("数据库备份 ZIP 大小无效"); } - restore(db, Cursor::new(archive), Some(theme_dir)).await + restore(db, Cursor::new(archive), Some(theme_dir), BACKUP_TABLES).await } pub async fn copy_database(source: &Database, target: &Database) -> Result { - let archive = export(source, None).await?; - restore(target, archive.file, None).await + let mut tables = BACKUP_TABLES.to_vec(); + tables.push("server_install_tokens"); + let archive = export(source, None, &tables).await?; + restore(target, archive.file, None, &tables).await } #[cfg(test)] mod tests { use super::*; + #[test] + fn exported_archives_must_fit_both_restore_limits() { + let compressed = DATABASE_BACKUP_MAX_BYTES as u64; + let extracted = DATABASE_BACKUP_MAX_EXTRACTED_BYTES; + assert!(validate_backup_size(compressed, extracted).is_ok()); + assert!(validate_backup_size(compressed + 1, 0).is_err()); + assert!(validate_backup_size(0, extracted + 1).is_err()); + } + #[tokio::test] async fn copies_database_through_a_streamed_archive() { let source = crate::db::connect("sqlite::memory:").await.unwrap(); @@ -917,6 +959,56 @@ mod tests { assert_eq!(site_name, "Migrated"); } + #[tokio::test] + async fn external_backups_keep_the_current_format_and_token_policy() { + let db = crate::db::connect("sqlite::memory:").await.unwrap(); + db.migrate().await.unwrap(); + sqlx::query( + "INSERT INTO settings(key,value) VALUES \ + ('admin_username','admin'),('admin_password_hash','hash'), \ + ('password_client_salt','salt'),('password_scheme','argon2-client-pbkdf2-v1')", + ) + .execute(db.pool()) + .await + .unwrap(); + sqlx::query("INSERT INTO servers(id,name,token_hash,created_at,updated_at) VALUES ('node','Node',?,1,1)") + .bind(crate::auth::token_hash("primary-token")) + .execute(db.pool()).await.unwrap(); + sqlx::query("INSERT INTO server_install_tokens(token_hash,server_id,created_at) VALUES (?,'node',1)") + .bind(crate::auth::token_hash("install-token")) + .execute(db.pool()).await.unwrap(); + let directory = tempfile::tempdir().unwrap(); + let themes = directory.path().join("themes"); + let mut exported = export_archive(&db, &themes).await.unwrap(); + let mut bytes = Vec::new(); + exported.file.read_to_end(&mut bytes).unwrap(); + let manifest = validate_archive( + &mut ZipArchive::new(Cursor::new(&bytes)).unwrap(), + BACKUP_TABLES, + ) + .unwrap(); + assert_eq!(manifest.version, 2); + assert!( + manifest + .excluded_ephemeral_tables + .iter() + .any(|table| table == "server_install_tokens") + ); + restore_archive(&db, &themes, &bytes).await.unwrap(); + assert!( + crate::db::queries::agent_identity(&db, "primary-token") + .await + .unwrap() + .is_some() + ); + assert!( + crate::db::queries::agent_identity(&db, "install-token") + .await + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn restores_current_format_snapshots_and_round_trips_aggregates() { let source = crate::db::connect("sqlite::memory:").await.unwrap(); @@ -940,7 +1032,7 @@ mod tests { sqlx::query("INSERT INTO server_install_tokens(token_hash,server_id,created_at) VALUES (?,'node',1)") .bind(crate::auth::token_hash("install-token")) .execute(source.pool()).await.unwrap(); - assert_eq!(copy_database(&source, &target).await.unwrap(), 6); + assert_eq!(copy_database(&source, &target).await.unwrap(), 7); let server = sqlx::query("SELECT price,hidden FROM servers WHERE id='node'") .fetch_one(target.pool()) .await @@ -957,7 +1049,7 @@ mod tests { crate::db::queries::agent_identity(&target, "install-token") .await .unwrap() - .is_none() + .is_some() ); let points = crate::db::queries::history(&target, "node", 1) .await @@ -990,7 +1082,7 @@ mod tests { memory_avg=50.0,memory_min=10.0,net_in_avg=20.0,net_in_min=1.0,first_timestamp=?,last_timestamp=?") .bind(crate::db::now() - 3).bind(crate::db::now()).execute(target.pool()).await.unwrap(); source.migrate().await.unwrap(); - assert_eq!(copy_database(&target, &source).await.unwrap(), 6); + assert_eq!(copy_database(&target, &source).await.unwrap(), 7); let row = sqlx::query( "SELECT sample_count,cpu_min,cpu_max,memory_avg,memory_min FROM metric_history", ) @@ -1014,7 +1106,7 @@ mod tests { crate::db::queries::agent_identity(&source, "install-token") .await .unwrap() - .is_none() + .is_some() ); } diff --git a/backend/src/main.rs b/backend/src/main.rs index 7f85686..5e04f5e 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -1,3 +1,4 @@ +mod activity; mod auth; mod backup; mod config; @@ -23,7 +24,7 @@ use clap::Parser; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::Ordering; use tokio::sync::{Mutex, RwLock, Semaphore, broadcast, watch}; use tower_http::trace::TraceLayer; @@ -36,11 +37,12 @@ pub struct AppState { pub database_overridden: bool, pub http: reqwest::Client, pub agents: RwLock>, + pub last_agent_reports: RwLock>, + pub started_at: i64, pub dashboard_tx: broadcast::Sender, pub database_maintenance: Mutex<()>, pub theme_operations: Mutex<()>, - pub database_maintenance_active: AtomicBool, - pub database_restart_required: AtomicBool, + pub database_activity: activity::DatabaseActivity, pub restart_tx: watch::Sender, pub login_attempts: security::AttemptLimiter, pub sensitive_attempts: security::AttemptLimiter, @@ -202,11 +204,12 @@ async fn main() -> Result<()> { database_overridden, http, agents: RwLock::new(HashMap::new()), + last_agent_reports: RwLock::new(HashMap::new()), + started_at: db::now(), dashboard_tx, database_maintenance: Mutex::new(()), theme_operations: Mutex::new(()), - database_maintenance_active: AtomicBool::new(false), - database_restart_required: AtomicBool::new(false), + database_activity: activity::DatabaseActivity::default(), restart_tx, login_attempts: security::AttemptLimiter::new( 5, @@ -357,6 +360,10 @@ async fn main() -> Result<()> { .merge(protected) .fallback(routes::site::handle) .layer(DefaultBodyLimit::max(2 * 1024 * 1024)) + .layer(axum_middleware::from_fn_with_state( + Arc::clone(&state), + middleware::database_activity, + )) .layer(TraceLayer::new_for_http()) .layer(axum_middleware::from_fn(middleware::security_headers)) .with_state(Arc::clone(&state)); @@ -383,6 +390,10 @@ fn spawn_maintenance(state: Arc) { let mut maintenance_runs = 0_u64; loop { interval.tick().await; + let _maintenance = state.database_maintenance.lock().await; + let Ok(_activity) = state.database_activity.write() else { + continue; + }; if let Err(error) = db::cleanup_auth(&state.db).await { tracing::error!(%error, "session cleanup failed"); } @@ -396,12 +407,20 @@ fn spawn_maintenance(state: Arc) { continue; } }; - if let Err(error) = notify::run_periodic(&state.db, &state.http, &settings).await { + let last_reports = state.last_agent_reports.read().await.clone(); + if let Err(error) = notify::run_periodic( + &state.db, + &state.http, + &settings, + &last_reports, + state.started_at, + ) + .await + { tracing::error!(%error, "notification maintenance failed"); } maintenance_runs = maintenance_runs.wrapping_add(1); { - let _maintenance = state.database_maintenance.lock().await; if let Err(error) = db::queries::cleanup_database(&state.db, settings.history_retention_days).await { diff --git a/backend/src/middleware.rs b/backend/src/middleware.rs index ca3c632..dccdd4a 100644 --- a/backend/src/middleware.rs +++ b/backend/src/middleware.rs @@ -1,6 +1,6 @@ use axum::{ extract::{Request, State}, - http::{HeaderName, HeaderValue, header}, + http::{HeaderName, HeaderValue, Method, header}, middleware::Next, response::{IntoResponse, Response}, }; @@ -9,6 +9,48 @@ use std::sync::Arc; use crate::AppState; use crate::routes::{ApiResponse, bearer_or_cookie}; +fn maintenance_route(path: &str) -> bool { + matches!( + path, + "/api/admin/database/migrate" + | "/api/admin/database/restore" + | "/api/admin/database/reclaim" + | "/api/admin/database/backup" + ) +} + +fn business_write(method: &Method, path: &str) -> bool { + !matches!(*method, Method::GET | Method::HEAD | Method::OPTIONS) + && !matches!( + path, + "/api/admin/login" + | "/api/admin/logout" + | "/api/turnstile/verify" + | "/api/admin/database/restart" + ) +} + +pub async fn database_activity( + State(state): State>, + request: Request, + next: Next, +) -> Response { + // Maintenance handlers acquire exclusive access after authentication/body parsing. + if maintenance_route(request.uri().path()) { + return next.run(request).await; + } + let access = if business_write(request.method(), request.uri().path()) { + state.database_activity.write() + } else { + state.database_activity.read() + }; + let _access = match access { + Ok(access) => access, + Err(error) => return error.into_response(), + }; + next.run(request).await +} + pub async fn auth_middleware( State(state): State>, mut request: Request, @@ -71,3 +113,23 @@ pub async fn security_headers(request: Request, next: Next) -> Response { ); response } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pending_migration_allows_login_and_restart_but_not_business_changes() { + assert!(!business_write(&Method::POST, "/api/admin/login")); + assert!(!business_write( + &Method::POST, + "/api/admin/database/restart" + )); + assert!(!business_write(&Method::GET, "/api/admin/database")); + assert!(business_write(&Method::PATCH, "/api/admin/settings")); + assert!(business_write(&Method::POST, "/api/admin/remote/task")); + assert!(business_write(&Method::DELETE, "/api/admin/servers/node")); + assert!(maintenance_route("/api/admin/database/migrate")); + assert!(!maintenance_route("/api/admin/database/restart")); + } +} diff --git a/backend/src/notify.rs b/backend/src/notify.rs index 9b9962a..2b7ac0c 100644 --- a/backend/src/notify.rs +++ b/backend/src/notify.rs @@ -2,6 +2,7 @@ use crate::db::{self, Database, Settings, queries}; use crate::models::TelegramSettingsView; use anyhow::{Context, Result}; use serde_json::Value; +use std::collections::HashMap; use std::time::Duration; pub async fn test_telegram(db: &Database, client: &reqwest::Client) -> Result<()> { @@ -78,22 +79,29 @@ pub async fn run_periodic( db: &Database, client: &reqwest::Client, settings: &Settings, + last_reports: &HashMap, + started_at: i64, ) -> Result<()> { if !settings.notification_enabled { return Ok(()); } let current = db::now(); for server in queries::list_servers(db, true).await? { - if !server.offline_notify_disabled { - let offline = server.timestamp.is_none_or(|timestamp| { - current.saturating_sub(timestamp) >= settings.offline_alert_minutes * 60 - }); + if !server.offline_notify_disabled + && let Some((offline, last_seen)) = offline_status( + server.timestamp, + last_reports.get(&server.id).copied(), + started_at, + current, + settings.offline_alert_minutes * 60, + ) + { let key = format!("offline:{}", server.id); if queries::update_alert_state( db, &key, offline, - &serde_json::json!({"timestamp": server.timestamp}), + &serde_json::json!({"timestamp": last_seen}), ) .await? { @@ -179,6 +187,22 @@ pub async fn run_periodic( Ok(()) } +fn offline_status( + persisted_at: Option, + received_at: Option, + started_at: i64, + current: i64, + delay: i64, +) -> Option<(bool, i64)> { + // No status transition until the first report or the end of the restart grace period. + let last_seen = received_at.or(persisted_at)?; + if received_at.is_none() && current.saturating_sub(started_at) < delay { + return None; + } + let elapsed = current.saturating_sub(last_seen); + Some((elapsed >= delay, last_seen)) +} + fn traffic_milestones(start: i64) -> Vec { let mut milestones = vec![start.clamp(50, 100)]; while *milestones.last().unwrap_or(&100) < 100 { @@ -277,7 +301,81 @@ fn render(template: &str, title: &str, server: &str, message: &str) -> String { #[cfg(test)] mod tests { - use super::traffic_milestones; + use super::{offline_status, traffic_milestones}; + + #[tokio::test] + async fn offline_alerts_start_only_after_the_first_valid_report() { + let db = crate::db::connect("sqlite::memory:").await.unwrap(); + db.migrate().await.unwrap(); + sqlx::query("INSERT INTO servers(id,name,token_hash,created_at,updated_at) VALUES ('node','Node','token',1,1)") + .execute(db.pool()).await.unwrap(); + let mut settings = crate::db::load_settings(&db).await.unwrap(); + settings.notification_enabled = true; + settings.offline_alert_minutes = 1; + let _ = rustls::crypto::ring::default_provider().install_default(); + let client = reqwest::Client::new(); + let now = crate::db::now(); + let mut reports = std::collections::HashMap::new(); + super::run_periodic(&db, &client, &settings, &reports, now - 3_600) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM alert_states") + .fetch_one(db.pool()) + .await + .unwrap(), + 0 + ); + for (received_at, expected_active) in [(now, 0), (now - 120, 1), (now, 0)] { + reports.insert("node".to_string(), received_at); + super::run_periodic(&db, &client, &settings, &reports, now - 3_600) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT active FROM alert_states WHERE state_key='offline:node'" + ) + .fetch_one(db.pool()) + .await + .unwrap(), + expected_active + ); + } + } + + #[test] + fn unreported_nodes_never_trigger_offline_alerts() { + assert_eq!(offline_status(None, None, 100, 10_000, 300), None); + } + + #[test] + fn live_reports_override_old_persisted_history() { + assert_eq!( + offline_status(Some(100), Some(999), 100, 1_000, 300), + Some((false, 999)) + ); + assert_eq!( + offline_status(None, Some(999), 100, 1_000, 300), + Some((false, 999)) + ); + assert_eq!( + offline_status(Some(100), Some(600), 100, 1_000, 300), + Some((true, 600)) + ); + } + + #[test] + fn backend_restart_allows_agents_time_to_reconnect() { + assert_eq!(offline_status(Some(100), None, 900, 1_000, 300), None); + assert_eq!( + offline_status(Some(100), Some(999), 900, 1_000, 300), + Some((false, 999)) + ); + assert_eq!( + offline_status(Some(100), None, 900, 1_200, 300), + Some((true, 100)) + ); + } #[test] fn traffic_alerts_advance_in_five_percent_steps() { diff --git a/backend/src/routes/admin.rs b/backend/src/routes/admin.rs index 7797fba..2770caa 100644 --- a/backend/src/routes/admin.rs +++ b/backend/src/routes/admin.rs @@ -19,7 +19,6 @@ use sha2::{Digest, Sha256}; use std::collections::HashSet; use std::net::{Ipv4Addr, SocketAddr}; use std::sync::Arc; -use std::sync::atomic::Ordering; use tokio_util::io::ReaderStream; use url::Url; @@ -34,14 +33,6 @@ pub struct DatabaseMigrationInput { database_url: String, } -struct DatabaseOperationFlag<'a>(&'a std::sync::atomic::AtomicBool); - -impl Drop for DatabaseOperationFlag<'_> { - fn drop(&mut self) { - self.0.store(false, Ordering::Release); - } -} - pub async fn servers_get(State(state): State>) -> Result { let servers = crate::db::queries::list_servers(&state.db, true) .await @@ -93,6 +84,7 @@ pub async fn server_delete( return Err(ApiResponse::not_found("节点不存在")); } state.disconnect_agent(&id).await; + state.last_agent_reports.write().await.remove(&id); Ok(StatusCode::NO_CONTENT.into_response()) } @@ -106,6 +98,7 @@ pub async fn servers_delete( .map_err(ApiResponse::internal)?; for id in input.ids { state.disconnect_agent(&id).await; + state.last_agent_reports.write().await.remove(&id); } Ok(StatusCode::NO_CONTENT.into_response()) } @@ -657,16 +650,13 @@ pub async fn exchange_refresh(State(state): State>) -> Result>) -> Result { let mut database = state.db.stats().await.map_err(ApiResponse::internal)?; - database.restart_required = state.database_restart_required.load(Ordering::Acquire); + database.restart_required = state.database_activity.restart_required(); Ok(Json(database).into_response()) } pub async fn database_reclaim(State(state): State>) -> Result { let _maintenance = state.database_maintenance.lock().await; - state - .database_maintenance_active - .store(true, Ordering::Release); - let _operation = DatabaseOperationFlag(&state.database_maintenance_active); + let _activity = state.database_activity.maintenance().await?; state.disconnect_agents().await; let before = state.db.stats().await.map_err(ApiResponse::internal)?; @@ -676,7 +666,7 @@ pub async fn database_reclaim(State(state): State>) -> Result>) -> Result { - if !state.database_restart_required.load(Ordering::Acquire) { + if !state.database_activity.restart_required() { return Err(ApiResponse::bad_request("当前没有等待生效的数据库迁移")); } state.disconnect_agents().await; @@ -767,10 +752,17 @@ pub async fn database_backup( ) -> Result { super::auth::require_sensitive_headers(&state, &user, &headers).await?; let _maintenance = state.database_maintenance.lock().await; + let _activity = state.database_activity.maintenance().await?; let _themes = state.theme_operations.lock().await; let archive = crate::backup::export_archive(&state.db, &state.config.theme_dir) .await - .map_err(ApiResponse::internal)?; + .map_err(|error| { + if error.is::() { + ApiResponse::unprocessable(error.to_string()) + } else { + ApiResponse::internal(error) + } + })?; let disposition = HeaderValue::from_str(&format!("attachment; filename=\"{}\"", archive.filename)) .map_err(ApiResponse::internal)?; @@ -827,16 +819,14 @@ pub async fn database_restore( })?; let _maintenance = state.database_maintenance.lock().await; + let _activity = state.database_activity.maintenance().await?; let _themes = state.theme_operations.lock().await; - state - .database_maintenance_active - .store(true, Ordering::Release); - let _operation = DatabaseOperationFlag(&state.database_maintenance_active); state.disconnect_agents().await; let restored_rows = crate::backup::restore_archive(&state.db, &state.config.theme_dir, &archive) .await .map_err(|error| ApiResponse::unprocessable(format!("恢复失败：{error}")))?; + state.last_agent_reports.write().await.clear(); Ok(Json(serde_json::json!({"restored_rows": restored_rows})).into_response()) } diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index 375bbaf..e17f37e 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -11,7 +11,7 @@ use futures_util::{SinkExt, StreamExt}; use serde::Deserialize; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::atomic::AtomicI64; use tokio::sync::mpsc; const MAX_AGENT_MESSAGE_BYTES: usize = 2 * 1024 * 1024; @@ -58,13 +58,10 @@ pub async fn handle( ConnectInfo(peer): ConnectInfo, headers: HeaderMap, ) -> Response { - if state.database_maintenance_active.load(Ordering::Acquire) { - return ApiResponse::error( - axum::http::StatusCode::SERVICE_UNAVAILABLE, - "数据库正在维护，请稍后重试", - ) - .into_response(); - } + let _activity = match state.database_activity.write() { + Ok(access) => access, + Err(error) => return error.into_response(), + }; if !agent_protocol_supported(&headers) { return ApiResponse::error( axum::http::StatusCode::UPGRADE_REQUIRED, @@ -87,10 +84,11 @@ pub async fn handle( }; let remote_ip = client_ip(&headers, peer, &state.config.trusted_proxies); let token = token.to_string(); + let connection_state = Arc::clone(&state); let mut response = ws .max_message_size(MAX_AGENT_MESSAGE_BYTES) .max_frame_size(MAX_AGENT_MESSAGE_BYTES) - .on_upgrade(move |socket| run(socket, state, identity, remote_ip, token)); + .on_upgrade(move |socket| run(socket, connection_state, identity, remote_ip, token)); response.headers_mut().insert( AGENT_PROTOCOL_HEADER, axum::http::HeaderValue::from_static(AGENT_PROTOCOL_VERSION), @@ -107,6 +105,10 @@ async fn run( ) { let connection_id = uuid::Uuid::new_v4().to_string(); let (mut websocket_tx, mut websocket_rx) = socket.split(); + let Ok(activity) = state.database_activity.write() else { + let _ = websocket_tx.send(Message::Close(None)).await; + return; + }; let (outbound_tx, mut outbound_rx) = mpsc::channel::(256); let previous = state.agents.write().await.insert( identity.server_id.clone(), @@ -159,6 +161,7 @@ async fn run( send_task(&outbound_tx, &task); } } + drop(activity); let writer = tokio::spawn(async move { while let Some(command) = outbound_rx.recv().await { @@ -217,10 +220,10 @@ async fn handle_text( outbound: &mpsc::Sender, text: &str, ) { - if state.database_maintenance_active.load(Ordering::Acquire) { + let Ok(_activity) = state.database_activity.write() else { send_persistence_error(outbound, 0); return; - } + }; let Ok(value) = serde_json::from_str::(text) else { return; }; @@ -245,6 +248,11 @@ async fn handle_text( .await { Ok(result) => { + state + .last_agent_reports + .write() + .await + .insert(identity.server_id.clone(), crate::db::now()); let next_wss_report_after_ms = state.agent_wss_interval_ms(&identity.server_id).await; send_ack(outbound, &result, next_wss_report_after_ms); @@ -255,6 +263,9 @@ async fn handle_text( let state = Arc::clone(state); let server_id = identity.server_id.clone(); tokio::spawn(async move { + let Ok(_activity) = state.database_activity.write() else { + return; + }; let settings = match crate::db::load_settings(&state.db).await { Ok(settings) => settings, Err(error) => { diff --git a/frontend/src/api.ts b/frontend/src/api.ts index a4dd5bf..07d5932 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1,4 +1,4 @@ -import type { AdminServer, AlertRule, AlertRuleInput, Bootstrap, Config, DatabaseMigrationResult, DatabaseStats, ExchangeRates, HistoryPoint, LatencySample, LatencyTask, LatencyTaskInput, LatencyTestPoint, LoginSession, RemoteTask, RemoteTaskCreated, RemoteTaskInput, Server, ServerInput, Settings, TelegramSettings, TelegramSettingsInput, Theme, ThemeSettingsSchema, TotpSetup, TotpStatus } from "./types"; +import type { AdminServer, AlertRule, AlertRuleInput, Bootstrap, Config, DatabaseMigrationResult, DatabaseStats, ExchangeRates, HistoryPoint, LatencySample, LatencyTask, LatencyTaskInput, LatencyTestPoint, LoginSession, RemoteTask, RemoteTaskCreated, RemoteTaskInput, ServerInput, Settings, TelegramSettings, TelegramSettingsInput, Theme, ThemeSettingsSchema, TotpSetup, TotpStatus } from "./types"; export const ADMIN_UNAUTHORIZED_EVENT = "nodeflare:admin-unauthorized"; @@ -51,7 +51,6 @@ export const api = { signal: AbortSignal.timeout(15_000), }), refreshExchangeRates: () => request("/api/admin/exchange-rates/refresh", { method: "POST" }, true), - servers: () => request<{ servers: Server[] }>("/api/servers"), adminServers: () => request<{ servers: AdminServer[] }>("/api/admin/servers", {}, true), history: (id: string, hours: number) => request<{ points: HistoryPoint[] }>(`/api/history/${encodeURIComponent(id)}?hours=${hours}`, { cache: "no-store", signal: AbortSignal.timeout(15_000) }), @@ -123,8 +122,6 @@ export const api = { { method: "POST", body: JSON.stringify(input) }, true, ), - rotateServerToken: (id: string) => - request<{ agent_token: string }>(`/api/admin/servers/${encodeURIComponent(id)}/token`, { method: "POST" }, true), createAgentInstallToken: (id: string) => request<{ agent_token: string }>(`/api/admin/servers/${encodeURIComponent(id)}/agent-token`, { method: "POST" }, true), updateServer: (id: string, input: ServerInput) => diff --git a/frontend/src/components/AdminPanel.tsx b/frontend/src/components/AdminPanel.tsx index aea1af7..2fafdc0 100644 --- a/frontend/src/components/AdminPanel.tsx +++ b/frontend/src/components/AdminPanel.tsx @@ -650,7 +650,7 @@ export function AdminPanel({ if (!databaseUrl) return; const source = database?.kind === "postgresql" ? "PostgreSQL" : "SQLite"; const target = database?.kind === "postgresql" ? "SQLite" : "PostgreSQL"; - if (!window.confirm(`将 ${source} 迁移到 ${target}？目标库中已有的 NodeFlare 数据会被覆盖。`)) return; + if (!window.confirm(`将 ${source} 迁移到 ${target}？目标库中已有的 NodeFlare 数据会被覆盖，完成后业务写入将暂停，直至重启服务。`)) return; setBusy(true); setError(""); setNotice(""); try { const proof = await sensitiveProof("迁移数据库"); @@ -660,7 +660,7 @@ export function AdminPanel({ setDatabaseMigrationResult(result); setDatabase((current) => current ? { ...current, restart_required: result.restart_required } : current); const targetName = result.target_kind === "postgresql" ? "PostgreSQL" : "SQLite"; - setNotice(`数据库已迁移到 ${targetName}，共 ${result.migrated_rows.toLocaleString()} 行（${formatBytes(result.size_bytes)}）`); + setNotice(`数据库已迁移到 ${targetName}，共 ${result.migrated_rows.toLocaleString()} 行（${formatBytes(result.size_bytes)}），业务写入已暂停，请重启服务`); } catch (reason) { setError(reason instanceof Error ? reason.message : "数据库迁移失败"); } finally { setBusy(false); } } @@ -1227,10 +1227,10 @@ export function AdminPanel({
数据库大小{database ? formatBytes(database.size_bytes) : "读取中..."}{database ? {database.kind === "postgresql" ? "PostgreSQL" : "SQLite"}{database.reclaimable_bytes ? ` · 可回收 ${formatBytes(database.reclaimable_bytes)}` : ""} : null}
数据库迁移

将当前 {database?.kind === "postgresql" ? "PostgreSQL" : "SQLite"} 数据复制到 {database?.kind === "postgresql" ? "SQLite" : "PostgreSQL"}，完成后自动更新配置，重启服务后生效。

- {database?.restart_required ?
迁移完成，等待重启{databaseMigrationResult ? `已复制 ${databaseMigrationResult.migrated_rows.toLocaleString()} 行，新数据库大小 ${formatBytes(databaseMigrationResult.size_bytes)}` : "配置已更新，重启后切换到新数据库"}
:
} + {database?.restart_required ?
迁移完成，业务写入已暂停{databaseMigrationResult ? `已复制 ${databaseMigrationResult.migrated_rows.toLocaleString()} 行，新数据库大小 ${formatBytes(databaseMigrationResult.size_bytes)}，请重启服务` : "配置已更新，重启后切换到新数据库并恢复写入"}
:
}
-
每日汇率

{exchangeRates ? `${exchangeRates.source} · ${exchangeRates.date || "等待首次更新"}${exchangeRates.stale ? " · 数据待更新" : ""}` : "正在读取汇率快照"}

{exchangeRatesExpanded ? : null}
+
每日汇率

{exchangeRates ? `${exchangeRates.source} · ${exchangeRates.date || "等待首次更新"}${exchangeRates.stale ? " · 数据待更新" : ""}` : "正在读取汇率快照"}

{exchangeRatesExpanded ? : null}
{exchangeRatesExpanded ? exchangeRates ?
{ASSET_CURRENCIES.filter((currency) => currency !== "CNY").map((currency) => )}
币种1 CNY 可兑换
{currency}{exchangeRates.rates[currency]?.toLocaleString(undefined, { maximumFractionDigits: 6 }) ?? "--"}
:
尚未读取
: null}
: null} diff --git a/frontend/src/install.test.ts b/frontend/src/install.test.ts index 518bee7..4568711 100644 --- a/frontend/src/install.test.ts +++ b/frontend/src/install.test.ts @@ -7,6 +7,24 @@ import { spawnSync } from "node:child_process"; const installer = readFileSync(new URL("../../install.sh", import.meta.url), "utf8"); const agentInstaller = readFileSync(new URL("../../agent/agent.sh", import.meta.url), "utf8"); +test("CI checks each shell script and stops on a later syntax error", () => { + const workflow = Bun.YAML.parse(readFileSync(new URL("../../.github/workflows/release.yml", import.meta.url), "utf8")) as { + jobs: { quality: { steps: { name?: string; run?: string }[] } }; + }; + const command = workflow.jobs.quality.steps.find((step) => step.name === "Check shell scripts")?.run; + expect(command).toBeDefined(); + const result = spawnSync("sh", ["-c", ` + sh() { + [ "$#" -eq 2 ] && [ "$1" = -n ] || return 98 + printf '%s\\n' "$2" + [ "$2" != agent/agent.sh ] || return 19 + } + ${command} + `], { encoding: "utf8" }); + expect(result.status).toBe(19); + expect(result.stdout).toBe("install.sh\nagent/agent.sh\n"); +}); + function shellFunctions(...names: string[]) { return names.map((name) => { const match = installer.match(new RegExp(`^${name}\\(\\) \\{[\\s\\S]*?^\\}`, "m")); @@ -271,3 +289,97 @@ describe("agent installer manual update", () => { expect(rejected.status).toBe(1); }); }); + +describe("agent installer service safety", () => { + test("a failed stop of an existing service prevents replacement", () => { + const stop = agentInstaller.slice(agentInstaller.indexOf(" rollback_agent=true\n"), agentInstaller.indexOf(' mv "$temporary" "$AGENT_FILE"')); + expect(stop).toContain("systemctl stop"); + for (const init of ["systemd", "openrc"]) { + const result = spawnSync("sh", ["-c", ` + set -eu + ${agentShellFunctions("fail")} + init_system=$TEST_INIT + had_service=true + SERVICE_NAME=nodeflare-agent + systemctl() { return 1; } + rc-service() { return 1; } + ${stop} + printf 'replaced' + `], { env: { ...process.env, TEST_INIT: init }, encoding: "utf8" }); + expect(result.status).toBe(1); + expect(result.stdout).toBe(""); + expect(result.stderr).toContain("未替换程序"); + } + }); + + test("refuses to stop its own systemd service from remote execution", () => { + const directory = mkdtempSync(join(tmpdir(), "nodeflare-agent-cgroup-")); + const path = join(directory, "cgroup"); + try { + for (const [cgroup, safe] of [ + ["0::/system.slice/nodeflare-agent.service\n", false], + ["1:name=systemd:/system.slice/nodeflare-agent.service/remote\n", false], + ["0::/user.slice/session-10.scope\n", true], + ["0::/system.slice/nodeflare.service\n", true], + ["0::/system.slice/other-nodeflare-agent.service\n", true], + ] as const) { + writeFileSync(path, cgroup); + const result = spawnSync("sh", ["-c", ` + set -eu + init_system=systemd + SERVICE_NAME=nodeflare-agent + ${agentShellFunctions("fail", "ensure_not_agent_service").replaceAll('"/proc/$$/cgroup"', '"$TEST_CGROUP_FILE"')} + ensure_not_agent_service + printf 'safe-to-stop' + `], { env: { ...process.env, TEST_CGROUP_FILE: path }, encoding: "utf8" }); + expect(result.status).toBe(safe ? 0 : 1); + expect(result.stdout).toBe(safe ? "safe-to-stop" : ""); + if (!safe) expect(result.stderr).toContain("服务未停止"); + } + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); + + for (const scenario of ["healthy", "exits", "restarts", "query-fails"]) { + test(`checks continuous agent startup: ${scenario}`, () => { + const result = spawnSync("sh", ["-c", ` + set -eu + SERVICE_NAME=nodeflare-agent + tick=0 + sleep() { tick=$((tick + 1)); } + systemctl() { + case "$1" in + is-active) [ "$SCENARIO" != exits ] || [ "$tick" -lt 2 ] ;; + show) + [ "$SCENARIO" != query-fails ] || return 1 + if [ "$SCENARIO" = restarts ] && [ "$tick" -ge 2 ]; then printf '456'; else printf '123'; fi + ;; + *) return 99 ;; + esac + } + ${agentShellFunctions("verify_agent_started")} + if verify_agent_started; then printf '%s' "$tick"; else exit 1; fi + `], { env: { ...process.env, SCENARIO: scenario }, encoding: "utf8" }); + expect(result.status).toBe(scenario === "healthy" ? 0 : 1); + if (scenario === "healthy") expect(result.stdout).toBe("10"); + }); + } + + for (const file of ["../../install.sh", "../../agent/agent.sh", "../../agent/install-macos.sh", "../../agent/install-freebsd.sh"]) { + test(`${file} exits on termination and runs rollback once`, () => { + const source = readFileSync(new URL(file, import.meta.url), "utf8"); + const traps = source.match(/^[ \t]*trap (?:cleanup|cleanup_agent_install) EXIT\n[ \t]*trap 'exit 129' HUP\n[ \t]*trap 'exit 130' INT\n[ \t]*trap 'exit 143' TERM/m)?.[0]; + expect(traps).toBeDefined(); + const result = spawnSync("sh", ["-c", ` + cleanup() { printf 'rollback'; } + cleanup_agent_install() { cleanup; } + ${traps} + kill -TERM "$$" + printf 'continued-after-termination' + `], { encoding: "utf8" }); + expect(result.status).toBe(143); + expect(result.stdout).toBe("rollback"); + }); + } +}); diff --git a/frontend/src/refresh.ts b/frontend/src/refresh.ts index 3e9748f..413a291 100644 --- a/frontend/src/refresh.ts +++ b/frontend/src/refresh.ts @@ -3,7 +3,7 @@ import type { RemoteTask } from "./types"; export const BOOTSTRAP_POLL_INTERVAL_MS = 15_000; export const BOOTSTRAP_LIVE_SYNC_INTERVAL_MS = 5 * 60_000; export const REMOTE_TASK_POLL_INTERVAL_MS = 2_000; -export const LIVE_FLUSH_INTERVAL_MS = 1_000; +const LIVE_FLUSH_INTERVAL_MS = 1_000; export interface LiveFlushScheduler { schedule: () => void; diff --git a/frontend/src/styles/admin.css b/frontend/src/styles/admin.css index ea83218..dad610c 100644 --- a/frontend/src/styles/admin.css +++ b/frontend/src/styles/admin.css @@ -428,6 +428,7 @@ @media (max-width: 639px) { .remote-submit { width: 100%; } + .usage-head { align-items: stretch; flex-direction: column; } .database-section-head .section-actions, .usage-head-actions { width: 100%; } .database-section-head .section-actions > label { width: 100%; } .database-section-head .section-actions > button, .usage-head-actions > button { flex: 1; } diff --git a/install.sh b/install.sh index 0691518..2dff08c 100755 --- a/install.sh +++ b/install.sh @@ -153,7 +153,10 @@ cleanup() { [ -z "$download_dir" ] || rm -rf "$download_dir" } -trap cleanup EXIT HUP INT TERM +trap cleanup EXIT +trap 'exit 129' HUP +trap 'exit 130' INT +trap 'exit 143' TERM toml_escape() { printf '%s' "$1" | sed -e 's/\\/\\\\/g' -e 's/"/\\"/g' diff --git a/scripts/agent-runtime-test.mjs b/scripts/agent-runtime-test.mjs new file mode 100644 index 0000000..8f0f9ef --- /dev/null +++ b/scripts/agent-runtime-test.mjs @@ -0,0 +1,153 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { createHash } from "node:crypto"; +import { once } from "node:events"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createServer as createHttpServer } from "node:http"; +import { createServer as createTcpServer } from "node:net"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { WebSocketServer } from "ws"; + +const binary = resolve(process.env.MONITOR_AGENT_BINARY ?? "agent/target/debug/nodeflare-agent"); +const directory = mkdtempSync(join(tmpdir(), "nodeflare-agent-runtime-")); +const agents = []; +const sockets = new Set(); +const releaseRequests = []; +const proxy = createHttpServer((_request, response) => { + response.writeHead(502); + response.end(); +}); +proxy.on("connect", (request, socket) => { + if (request.url === "api.github.com:443") releaseRequests.push(Date.now()); + socket.end("HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); +}); +let websocketServer; +let upgradeServer; +let stalledServer; + +async function waitUntil(predicate, timeout, message) { + const deadline = Date.now() + timeout; + while (!predicate()) { + assert(Date.now() < deadline, message); + await delay(50); + } +} + +function startAgent(port, token, name) { + const proxyUrl = `http://127.0.0.1:${proxy.address().port}`; + const child = spawn(binary, ["-e", `http://127.0.0.1:${port}`, "-t", token], { + env: { + ...process.env, + NODEFLARE_STATE_DIR: join(directory, name), + HTTP_PROXY: proxyUrl, http_proxy: proxyUrl, + HTTPS_PROXY: proxyUrl, https_proxy: proxyUrl, + ALL_PROXY: proxyUrl, all_proxy: proxyUrl, + NO_PROXY: "", no_proxy: "", + }, + stdio: ["ignore", "ignore", "pipe"], + }); + const running = { child, exited: once(child, "exit"), stderr: "" }; + child.stderr.on("data", (chunk) => { running.stderr += chunk; }); + agents.push(running); + return running; +} + +async function stopAgent(running) { + if (running.child.exitCode === null && running.child.signalCode === null) { + running.child.kill("SIGTERM"); + const timeout = setTimeout(() => running.child.kill("SIGKILL"), 3_000); + await running.exited; + clearTimeout(timeout); + } +} + +try { + proxy.listen(0, "127.0.0.1"); + await once(proxy, "listening"); + + console.log("agent runtime: automatic update during one-second sampling"); + websocketServer = new WebSocketServer({ noServer: true }); + upgradeServer = createHttpServer(); + let redirected = false; + upgradeServer.on("upgrade", (request, socket, head) => { + if (request.url === "/api/agent/ws") { + redirected = true; + socket.end("HTTP/1.1 307 Temporary Redirect\r\nLocation: /live\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + return; + } + websocketServer.handleUpgrade(request, socket, head, (connection) => { + websocketServer.emit("connection", connection); + }); + }); + upgradeServer.listen(0, "127.0.0.1"); + await once(upgradeServer, "listening"); + websocketServer.on("headers", (headers) => headers.push("X-NodeFlare-Agent-Protocol: 2")); + let samples = 0; + websocketServer.on("connection", (socket) => { + let persistedThrough = 0; + socket.send(JSON.stringify({ + type: "config", ts: Math.floor(Date.now() / 1_000), + config: { + report_interval: 15, collect_interval: 1, network_interface: "", + agent_mirror: "", auto_update: true, latency_tasks: [], + }, + })); + socket.on("message", (raw) => { + const message = JSON.parse(raw.toString()); + if (message.type !== "update") return; + samples += message.samples.length; + if (message.persist) persistedThrough = Math.max(...message.samples.map((sample) => sample.timestamp)); + socket.send(JSON.stringify({ + type: "ack", ts: Math.floor(Date.now() / 1_000), + persisted: message.persist, persistenceError: false, + persistedThroughTs: persistedThrough, + nextPersistAfterMs: 1_000, nextWssReportAfterMs: 1_000, realtimeHint: false, + })); + }); + }); + let token; + for (let index = 0; index < 100_000; index++) { + const candidate = `runtime-update-${index}`; + if (createHash("sha256").update(candidate).digest().readBigUInt64BE(0) % 1_801n === 2n) { + token = candidate; + break; + } + } + assert(token, "Unable to create deterministic update jitter"); + const updating = startAgent(upgradeServer.address().port, token, "update"); + await waitUntil(() => releaseRequests.length > 0, 12_000, "Automatic update was starved by sampling"); + assert(samples > 0, "Update check must follow acknowledged samples"); + assert(redirected, "Same-origin WebSocket redirects must remain supported"); + await stopAgent(updating); + + console.log("agent runtime: stalled handshake times out and reconnects"); + let connections = 0; + let handshakeReceived = false; + stalledServer = createTcpServer((socket) => { + connections++; + sockets.add(socket); + socket.on("data", (data) => { handshakeReceived ||= data.toString().includes("/api/agent/ws"); }); + socket.on("error", () => {}); + socket.on("close", () => sockets.delete(socket)); + }); + stalledServer.listen(0, "127.0.0.1"); + await once(stalledServer, "listening"); + const stalled = startAgent(stalledServer.address().port, "runtime-stalled-handshake", "stalled"); + await waitUntil(() => connections >= 2, 20_000, "Stalled WebSocket handshake did not reconnect"); + assert(handshakeReceived, "The test must reach the WebSocket handshake"); + assert(stalled.stderr.includes("live connection failed"), "Handshake failure must be visible"); + console.log("Agent runtime tests passed"); +} finally { + for (const running of agents) await stopAgent(running); + for (const socket of sockets) socket.destroy(); + if (websocketServer) { + for (const socket of websocketServer.clients) socket.terminate(); + await new Promise((resolveClose) => websocketServer.close(resolveClose)); + } + if (stalledServer?.listening) await new Promise((resolveClose) => stalledServer.close(resolveClose)); + if (upgradeServer?.listening) await new Promise((resolveClose) => upgradeServer.close(resolveClose)); + if (proxy.listening) await new Promise((resolveClose) => proxy.close(resolveClose)); + rmSync(directory, { recursive: true, force: true }); +} diff --git a/scripts/smoke-test.sh b/scripts/smoke-test.sh index 8324d7f..1ad8fb7 100644 --- a/scripts/smoke-test.sh +++ b/scripts/smoke-test.sh @@ -62,6 +62,7 @@ alert_rule_id= backup_file= agent_pid= agent_state_dir= +migration_completed=false cleanup() { if [ -n "$agent_pid" ]; then kill "$agent_pid" 2>/dev/null || true @@ -69,7 +70,7 @@ cleanup() { fi [ -z "$agent_state_dir" ] || rm -rf "$agent_state_dir" [ -z "$backup_file" ] || rm -f -- "$backup_file" - if [ "${MONITOR_KEEP_RESOURCES:-0}" != "1" ] && [ -n "$admin_token" ]; then + if [ "$migration_completed" = false ] && [ "${MONITOR_KEEP_RESOURCES:-0}" != "1" ] && [ -n "$admin_token" ]; then if [ -n "$alert_rule_id" ]; then monitor_curl --silent --show-error -H "Authorization: Bearer $admin_token" \ -X DELETE "$MONITOR_BASE_URL/api/admin/alert-rules/$alert_rule_id" >/dev/null || true @@ -348,6 +349,26 @@ if [ -n "${MONITOR_MIGRATION_URL:-}" ]; then --data "$migration_payload" "$MONITOR_BASE_URL/api/admin/database/migrate" | \ jq -e --arg kind "${MONITOR_MIGRATION_KIND:-postgresql}" \ '.migrated_rows > 0 and .target_kind == $kind and .size_bytes > 0 and .restart_required == true' >/dev/null + migration_completed=true + step "database migration write freeze" + request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/admin/database" | \ + jq -e '.restart_required == true' >/dev/null + frozen_settings_status=$(monitor_curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/json' -X PATCH \ + --data "$settings_payload" "$MONITOR_BASE_URL/api/admin/settings") + [ "$frozen_settings_status" = "503" ] + frozen_delete_status=$(monitor_curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + -H "Authorization: Bearer $admin_token" -X DELETE "$MONITOR_BASE_URL/api/admin/servers/$server_id") + [ "$frozen_delete_status" = "503" ] + frozen_agent_status=$(monitor_curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + -H "Authorization: Bearer $agent_token" -H 'X-NodeFlare-Agent-Protocol: 2' \ + -H 'Connection: Upgrade' -H 'Upgrade: websocket' -H 'Sec-WebSocket-Version: 13' \ + -H 'Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==' "$MONITOR_BASE_URL/api/agent/ws") + [ "$frozen_agent_status" = "503" ] + request -H "Authorization: Bearer $admin_token" -X POST "$MONITOR_BASE_URL/api/admin/logout" >/dev/null + admin_token=$(login_admin | jq -er '.token') + request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/admin/database" | \ + jq -e '.restart_required == true' >/dev/null fi echo "Smoke test passed"