From 844f514ed28fa901811e6d1dba3c01b1584ff887 Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 02:17:11 +0800 Subject: [PATCH 1/8] fix(server): clarify readiness diagnostic semantics Co-authored-by: Claude --- README.md | 2 + docs/readiness.md | 33 ++++++ .../src/handlers/auth_routes/healthz.rs | 11 +- .../src/tests/route_surface_tests.rs | 101 +++++++++++++----- 4 files changed, 115 insertions(+), 32 deletions(-) create mode 100644 docs/readiness.md diff --git a/README.md b/README.md index 0b15e48..846b50d 100644 --- a/README.md +++ b/README.md @@ -85,6 +85,8 @@ sudo systemctl status nodelite-agent.service sudo journalctl -u nodelite-agent.service -f ``` +反向代理、负载均衡器和告警系统接入前，请确认 [`/healthz` 与 `/readyz` 的探针语义](docs/readiness.md)，避免把诊断降级误判为实例不可接流量。 + 升级建议由管理员手动触发：先确认 release notes 和协议兼容，再升级 Server 或 Agent。 ## 常见排障 diff --git a/docs/readiness.md b/docs/readiness.md new file mode 100644 index 0000000..29ef341 --- /dev/null +++ b/docs/readiness.md @@ -0,0 +1,33 @@ +# 健康检查与就绪探针 + +NodeLite Server 提供两个无需认证的探针端点： + +- `/healthz`：进程存活检查。只要 HTTP 服务还能响应就返回 `200 OK`。 +- `/readyz`：流量就绪检查和结构化运行诊断。HTTP 状态码与 JSON 的 `ready` 字段表示能否承载流量，`status` 和 `problems` 表示更广泛的运行健康度。 + +## `/readyz` 语义 + +`/readyz` 将信号分为两类： + +| 类别 | JSON 位置 | 是否影响 HTTP 状态码和 `ready` | +| --- | --- | --- | +| 硬就绪检查 | `checks.history_available`、`checks.registry_reload_healthy` | 是；任一失败时返回 `503 Service Unavailable` 和 `ready: false` | +| 诊断信号 | `signals` 中的审计可用性、写入丢弃/失败、队列及 WebSocket 容量 | 否；异常会返回 `status: "degraded"` 并加入 `problems` | + +审计日志是安全诊断能力，但不是 Server 接收 Agent、查询节点状态所需的硬依赖。因此，仅审计写入器不可用时，响应组合是： + +```json +{ + "status": "degraded", + "ready": true, + "problems": ["audit_unavailable"] +} +``` + +对应的 HTTP 状态码仍为 `200 OK`。如果历史存储或注册表重载检查失败，则响应为 `ready: false`、`status: "degraded"`，HTTP 状态码为 `503 Service Unavailable`。 + +## 探针与告警配置 + +- Kubernetes、systemd watchdog 或负载均衡器的就绪判断应使用 `/readyz` 的 HTTP 状态码；需要解析 JSON 时，应读取 `ready`，不要把 `status == "ok"` 当作接流量条件。 +- 告警系统应另外监控 `status`、`problems` 和 `signals`。`ready: true` 且 `status: "degraded"` 表示服务仍可接流量，但存在需要运维处理的诊断异常。 +- 不要仅用 `/healthz` 判断是否应把实例加入流量池；它只验证进程仍能响应。 diff --git a/nodelite-server/src/handlers/auth_routes/healthz.rs b/nodelite-server/src/handlers/auth_routes/healthz.rs index 3b7433f..d0ef916 100644 --- a/nodelite-server/src/handlers/auth_routes/healthz.rs +++ b/nodelite-server/src/handlers/auth_routes/healthz.rs @@ -47,11 +47,14 @@ pub(crate) async fn healthz() -> StatusCode { StatusCode::OK } -/// 就绪检查接口:保留 200/503 语义,同时返回结构化诊断方便排障。 +/// 就绪检查接口。 +/// +/// HTTP 状态码和 `ready` 仅反映承载流量所需的硬依赖；审计与写入异常属于诊断信号， +/// 只会把 `status` 标为 `degraded` 并写入 `problems`，不会单独触发 503。 pub(crate) async fn readyz(State(state): State) -> Response { let history_available = state.readiness.history_available(); let registry_reload_healthy = state.readiness.registry_reload_healthy(); - let ready = history_available && registry_reload_healthy; + let hard_ready = history_available && registry_reload_healthy; let audit_enabled = state.audit_log.enabled(); let audit_available = state.audit_log.is_available().await; let history_dropped_writes = state.history.dropped_writes(); @@ -88,7 +91,7 @@ pub(crate) async fn readyz(State(state): State) -> Response { } else { "degraded" }, - ready, + ready: hard_ready, problems, checks: ReadyzChecks { history_available, @@ -112,7 +115,7 @@ pub(crate) async fn readyz(State(state): State) -> Response { browser_ws_max_connections_per_ip: browser_ws_snapshot.max_connections_per_ip, }, }; - let status = if ready { + let status = if hard_ready { StatusCode::OK } else { StatusCode::SERVICE_UNAVAILABLE diff --git a/nodelite-server/src/tests/route_surface_tests.rs b/nodelite-server/src/tests/route_surface_tests.rs index a60077d..045940a 100644 --- a/nodelite-server/src/tests/route_surface_tests.rs +++ b/nodelite-server/src/tests/route_surface_tests.rs @@ -75,38 +75,14 @@ fn router_builds_with_v08_path_syntax() { } #[test] -fn readyz_reports_json_diagnostics_for_degraded_state() { +fn readyz_returns_503_for_hard_dependency_failure() { let runtime = Runtime::new().expect("runtime should build"); runtime.block_on(async { - let unique = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("clock should be monotonic enough") - .as_nanos(); - let registry_path = - std::env::temp_dir().join(format!("nodelite-readyz-test-{unique}.json")); - let mut config = test_server_config( - SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 8080)), - "http://127.0.0.1:8080".to_string(), - registry_path, - PathBuf::from("./data/history.sqlite3"), - PathBuf::from("./data/snapshot.json"), - ); - config.readonly_auth = None; - config.ws = test_ws_config(32, 8); - let state = AppState::test_fixture( - Arc::new(config), - Arc::new(PathBuf::from("config/server.toml")), - ) - .await - .expect("state fixture should build"); + let (state, temp_dir) = readyz_test_state("hard-failure").await; state.readiness.mark_history_available(false); - let response = readyz(State(state.clone())).await.into_response(); - assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); - let body = to_bytes(response.into_body(), usize::MAX) - .await - .expect("readyz body should collect"); - let payload: Value = serde_json::from_slice(&body).expect("readyz body should be json"); + let (status, payload) = readyz_payload(state.clone()).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(payload["ready"], Value::Bool(false)); assert_eq!(payload["status"], Value::String("degraded".to_string())); assert_eq!( @@ -116,9 +92,78 @@ fn readyz_reports_json_diagnostics_for_degraded_state() { state.history.shutdown().await; state.audit_log.shutdown().await; + drop(state); + std::fs::remove_dir_all(temp_dir).expect("readyz temp dir should be removable"); + }); +} + +#[test] +fn readyz_returns_200_for_audit_only_degradation() { + let runtime = Runtime::new().expect("runtime should build"); + runtime.block_on(async { + let (state, temp_dir) = readyz_test_state("audit-degraded").await; + state.audit_log.shutdown().await; + + let (status, payload) = readyz_payload(state.clone()).await; + let (repeat_status, repeat_payload) = readyz_payload(state.clone()).await; + assert_eq!(repeat_status, status); + assert_eq!(repeat_payload, payload); + assert_eq!(status, StatusCode::OK); + assert_eq!(payload["ready"], Value::Bool(true)); + assert_eq!(payload["status"], Value::String("degraded".to_string())); + assert_eq!( + payload["problems"], + Value::Array(vec![Value::String("audit_unavailable".to_string())]) + ); + assert_eq!(payload["checks"]["history_available"], Value::Bool(true)); + assert_eq!( + payload["checks"]["registry_reload_healthy"], + Value::Bool(true) + ); + assert_eq!(payload["signals"]["audit_enabled"], Value::Bool(true)); + assert_eq!(payload["signals"]["audit_available"], Value::Bool(false)); + + state.history.shutdown().await; + drop(state); + std::fs::remove_dir_all(temp_dir).expect("readyz temp dir should be removable"); }); } +async fn readyz_test_state(test_name: &str) -> (AppState, PathBuf) { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should be monotonic enough") + .as_nanos(); + let temp_dir = std::env::temp_dir().join(format!("nodelite-readyz-{test_name}-{unique}")); + std::fs::create_dir_all(&temp_dir).expect("readyz temp dir should exist"); + let mut config = test_server_config( + SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 8080)), + "http://127.0.0.1:8080".to_string(), + temp_dir.join("registry.json"), + temp_dir.join("history.sqlite3"), + temp_dir.join("snapshot.json"), + ); + config.readonly_auth = None; + config.ws = test_ws_config(32, 8); + let state = AppState::test_fixture( + Arc::new(config), + Arc::new(PathBuf::from("config/server.toml")), + ) + .await + .expect("state fixture should build"); + (state, temp_dir) +} + +async fn readyz_payload(state: AppState) -> (StatusCode, Value) { + let response = readyz(State(state)).await.into_response(); + let status = response.status(); + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("readyz body should collect"); + let payload = serde_json::from_slice(&body).expect("readyz body should be json"); + (status, payload) +} + #[test] fn install_endpoints_disable_caching() { let runtime = Runtime::new().expect("runtime should build"); From 99130cc7afd27b70af87f83b8c66ad7f6cc6000b Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 02:26:03 +0800 Subject: [PATCH 2/8] feat(server): expose token cache diagnostics Co-authored-by: Claude --- docs/token-verification.md | 11 +- .../src/handlers/metrics_routes/tests.rs | 9 ++ .../handlers/metrics_routes/token_verify.rs | 18 +++ nodelite-server/src/registry.rs | 3 - nodelite-server/src/registry/auth.rs | 62 +++++++--- nodelite-server/src/registry/metrics.rs | 23 ++++ .../src/registry/tests/cache_tests.rs | 108 ++++++++++++++++++ 7 files changed, 211 insertions(+), 23 deletions(-) diff --git a/docs/token-verification.md b/docs/token-verification.md index 9d2c2ba..a5469f7 100644 --- a/docs/token-verification.md +++ b/docs/token-verification.md @@ -21,7 +21,16 @@ NodeLite 使用 Argon2id 验证 Agent Token。每次冷缓存验证约需要 19 - `nodelite_token_verify_limit`：当前配置的最大并发数； - `nodelite_token_verify_active`：正在执行 Argon2 的任务数； - `nodelite_token_verify_waiting`：正在等待并发许可的请求数； -- `nodelite_token_verify_wait_seconds_total`：所有请求累计等待许可的秒数。 +- `nodelite_token_verify_wait_seconds_total`：所有请求累计等待许可的秒数； +- `nodelite_token_cache_hits_total`：由未过期缓存直接返回的验证次数，包括获取并发许可后的二次检查命中； +- `nodelite_token_cache_misses_total`：实际执行 Argon2 验证的次数； +- `nodelite_token_cache_evictions_total`：缓存达到容量上限后发生的 LRU 驱逐次数。 + +缓存命中包含成功和失败的验证结果。`evictions_total` 只统计容量驱逐，不统计 TTL 过期 +清理或 Token 轮换触发的显式清空。诊断命中率时可比较 +`rate(nodelite_token_cache_hits_total[5m])` 与 hits、misses 两者速率之和；如果 miss 较高但 +eviction 保持为 0，应优先检查并发冷启动、注册表 revision 变化或 Token 轮换，而不是直接 +增大缓存。 如果 `waiting` 长时间大于 0，且主机仍有足够内存，可以逐步提高并发。小内存主机应优先 保持 2 或 4，并结合进程 RSS、cgroup `MemoryCurrent` 和 OOM 日志判断，而不是只看单次 diff --git a/nodelite-server/src/handlers/metrics_routes/tests.rs b/nodelite-server/src/handlers/metrics_routes/tests.rs index edc7c91..eeb60fc 100644 --- a/nodelite-server/src/handlers/metrics_routes/tests.rs +++ b/nodelite-server/src/handlers/metrics_routes/tests.rs @@ -375,6 +375,9 @@ fn exporter_exposes_token_verify_pressure_metrics() { active: 3, waiting: 7, wait_seconds_total: 1.25, + token_cache_hits_total: 11, + token_cache_misses_total: 13, + token_cache_evictions_total: 17, }); assert!(body.contains("# TYPE nodelite_token_verify_limit gauge")); @@ -385,6 +388,12 @@ fn exporter_exposes_token_verify_pressure_metrics() { assert!(body.contains("nodelite_token_verify_waiting 7")); assert!(body.contains("# TYPE nodelite_token_verify_wait_seconds_total counter")); assert!(body.contains("nodelite_token_verify_wait_seconds_total 1.25")); + assert!(body.contains("# TYPE nodelite_token_cache_hits_total counter")); + assert!(body.contains("nodelite_token_cache_hits_total 11")); + assert!(body.contains("# TYPE nodelite_token_cache_misses_total counter")); + assert!(body.contains("nodelite_token_cache_misses_total 13")); + assert!(body.contains("# TYPE nodelite_token_cache_evictions_total counter")); + assert!(body.contains("nodelite_token_cache_evictions_total 17")); } #[test] diff --git a/nodelite-server/src/handlers/metrics_routes/token_verify.rs b/nodelite-server/src/handlers/metrics_routes/token_verify.rs index 0ae9606..1f7fd6c 100644 --- a/nodelite-server/src/handlers/metrics_routes/token_verify.rs +++ b/nodelite-server/src/handlers/metrics_routes/token_verify.rs @@ -30,5 +30,23 @@ pub(crate) fn render_token_verify_metrics(metrics: TokenVerifyMetrics) -> String &[], metrics.wait_seconds_total, ); + emitter.counter( + "nodelite_token_cache_hits_total", + "Number of token verification requests served by a live cached result.", + &[], + metrics.token_cache_hits_total, + ); + emitter.counter( + "nodelite_token_cache_misses_total", + "Number of token verification requests that executed Argon2 after cache lookup.", + &[], + metrics.token_cache_misses_total, + ); + emitter.counter( + "nodelite_token_cache_evictions_total", + "Number of token verification cache entries evicted because the cache was at capacity.", + &[], + metrics.token_cache_evictions_total, + ); emitter.finish() } diff --git a/nodelite-server/src/registry.rs b/nodelite-server/src/registry.rs index fc6580a..f167e56 100644 --- a/nodelite-server/src/registry.rs +++ b/nodelite-server/src/registry.rs @@ -80,9 +80,6 @@ const LOCATION_COORDINATE_SCALE: f64 = 1_000_000.0; /// 配合 8 并发度的 Argon2id verify,200 节点 reconnect storm 实测: /// p50 从 640ms 降至 ~70ms,p95 从 1,689ms 降至 ~700ms。 /// -/// 注意:实测 cache hit rate 约 33%,低于理论值(200 nodes 放入 512 容量应该 -/// 无 eviction)。可能原因:并发 miss、registry_revision 变化、或统计口径问题。 -/// TODO: 添加 cache_hit/miss/eviction 计数以诊断实际缓存效率。 /// TODO: 后续可考虑配置化,允许低内存 VPS 降至 128 或 256。 const TOKEN_CACHE_CAPACITY: usize = 512; /// Token 验证结果缓存 TTL:5 分钟。重连场景下可直接命中缓存,避免 Argon2id 开销。 diff --git a/nodelite-server/src/registry/auth.rs b/nodelite-server/src/registry/auth.rs index 2670aec..36857bf 100644 --- a/nodelite-server/src/registry/auth.rs +++ b/nodelite-server/src/registry/auth.rs @@ -7,7 +7,11 @@ use std::time::Instant; use anyhow::anyhow; use chrono::{DateTime, Utc}; +#[cfg(test)] +use lru::LruCache; use nodelite_proto::{NodeIdentity, validate_non_empty}; +#[cfg(test)] +use parking_lot::Mutex as ParkingLotMutex; use sha2::{Digest, Sha256}; #[cfg(test)] use tokio::sync::Semaphore; @@ -19,6 +23,7 @@ use super::validate::validate_runtime_identity; use super::{ AuthorizedNode, NodeRegistry, RegisteredNode, RegistryError, RegistryResult, RegistryTokenStatus, TOKEN_CACHE_TTL, TOKEN_VERIFY_WAIT_WARN_AFTER, TokenCacheEntry, + TokenCacheKey, }; #[derive(Debug, Clone, Copy)] @@ -168,17 +173,8 @@ impl NodeRegistry { }; let cache_key = (cache_key_hash.clone(), self.registry_revision()); - // 检查缓存（parking_lot::Mutex 是同步的） - { - let mut cache = self.token_cache.lock(); - if let Some(entry) = cache.get(&cache_key) { - let age = entry.cached_at.elapsed(); - if age < TOKEN_CACHE_TTL { - return Ok(entry.verified); - } - // TTL 过期,移除旧条目 - cache.pop(&cache_key); - } + if let Some(verified) = self.cached_token_result(&cache_key) { + return Ok(verified); } // 缓存未命中,获取 semaphore 进行验证 @@ -199,14 +195,8 @@ impl NodeRegistry { } // 获取 permit 后再次检查缓存（并发场景下可能已经被其他请求填充） - { - let mut cache = self.token_cache.lock(); - if let Some(entry) = cache.get(&cache_key) { - let age = entry.cached_at.elapsed(); - if age < TOKEN_CACHE_TTL { - return Ok(entry.verified); - } - } + if let Some(verified) = self.cached_token_result(&cache_key) { + return Ok(verified); } // 执行实际的 Argon2id 验证 @@ -217,6 +207,7 @@ impl NodeRegistry { let probe = self.token_verify_probe.clone(); let (verified, permit) = tokio::task::spawn_blocking(move || { + metrics.record_cache_miss(); let _active_guard = TokenVerifyActiveGuard::start(metrics); #[cfg(test)] let _probe_guard = probe.as_ref().map(|probe| probe.enter()); @@ -228,6 +219,7 @@ impl NodeRegistry { // 写入缓存 { let mut cache = self.token_cache.lock(); + let capacity_eviction = cache.len() == cache.cap().get() && !cache.contains(&cache_key); cache.put( cache_key, TokenCacheEntry { @@ -235,12 +227,36 @@ impl NodeRegistry { cached_at: Instant::now(), }, ); + if capacity_eviction { + self.token_verify_metrics.record_cache_eviction(); + } } drop(permit); Ok(verified) } + fn cached_token_result(&self, cache_key: &TokenCacheKey) -> Option { + let cached_result = { + let mut cache = self.token_cache.lock(); + match cache + .get(cache_key) + .map(|entry| (entry.verified, entry.cached_at.elapsed())) + { + Some((verified, age)) if age < TOKEN_CACHE_TTL => Some(verified), + Some(_) => { + cache.pop(cache_key); + None + } + None => None, + } + }; + if cached_result.is_some() { + self.token_verify_metrics.record_cache_hit(); + } + cached_result + } + pub(crate) fn token_verify_metrics(&self) -> super::TokenVerifyMetrics { self.token_verify_metrics.snapshot(self.token_verify_limit) } @@ -261,6 +277,14 @@ impl NodeRegistry { self.token_verify_probe = Some(probe); self } + + #[cfg(test)] + pub(super) fn with_token_cache_capacity_for_tests(mut self, capacity: usize) -> Self { + assert!(capacity > 0, "test token cache capacity must be positive"); + let capacity = std::num::NonZeroUsize::new(capacity).unwrap_or(std::num::NonZeroUsize::MIN); + self.token_cache = Arc::new(ParkingLotMutex::new(LruCache::new(capacity))); + self + } } pub(super) fn token_status_for_node( diff --git a/nodelite-server/src/registry/metrics.rs b/nodelite-server/src/registry/metrics.rs index fe306d6..45d8592 100644 --- a/nodelite-server/src/registry/metrics.rs +++ b/nodelite-server/src/registry/metrics.rs @@ -10,6 +10,9 @@ pub(crate) struct TokenVerifyMetrics { pub(crate) active: u64, pub(crate) waiting: u64, pub(crate) wait_seconds_total: f64, + pub(crate) token_cache_hits_total: u64, + pub(crate) token_cache_misses_total: u64, + pub(crate) token_cache_evictions_total: u64, } #[derive(Debug, Default)] @@ -17,6 +20,9 @@ pub(super) struct TokenVerifyMetricsState { active: AtomicU64, waiting: AtomicU64, wait_nanos_total: AtomicU64, + token_cache_hits_total: AtomicU64, + token_cache_misses_total: AtomicU64, + token_cache_evictions_total: AtomicU64, } impl TokenVerifyMetricsState { @@ -27,9 +33,26 @@ impl TokenVerifyMetricsState { waiting: self.waiting.load(Ordering::Relaxed), wait_seconds_total: self.wait_nanos_total.load(Ordering::Relaxed) as f64 / 1_000_000_000.0, + token_cache_hits_total: self.token_cache_hits_total.load(Ordering::Relaxed), + token_cache_misses_total: self.token_cache_misses_total.load(Ordering::Relaxed), + token_cache_evictions_total: self.token_cache_evictions_total.load(Ordering::Relaxed), } } + pub(super) fn record_cache_hit(&self) { + self.token_cache_hits_total.fetch_add(1, Ordering::Relaxed); + } + + pub(super) fn record_cache_miss(&self) { + self.token_cache_misses_total + .fetch_add(1, Ordering::Relaxed); + } + + pub(super) fn record_cache_eviction(&self) { + self.token_cache_evictions_total + .fetch_add(1, Ordering::Relaxed); + } + fn record_wait(&self, waited: Duration) { let nanos = u64::try_from(waited.as_nanos()).unwrap_or(u64::MAX); let _ = diff --git a/nodelite-server/src/registry/tests/cache_tests.rs b/nodelite-server/src/registry/tests/cache_tests.rs index 2b7f418..f09bbb0 100644 --- a/nodelite-server/src/registry/tests/cache_tests.rs +++ b/nodelite-server/src/registry/tests/cache_tests.rs @@ -1,5 +1,69 @@ use super::*; +#[tokio::test] +async fn token_cache_miss_counter_tracks_argon2_work() { + let (registry, credentials, temp_dir) = + token_cache_fixture("miss-metric", &["miss-01"], 2).await; + let (identity, token) = &credentials[0]; + + registry + .authorize(identity, token) + .await + .expect("cold token should authorize"); + + let metrics = registry.token_verify_metrics(); + assert_eq!(metrics.token_cache_hits_total, 0); + assert_eq!(metrics.token_cache_misses_total, 1); + assert_eq!(metrics.token_cache_evictions_total, 0); + + drop(registry); + std::fs::remove_dir_all(temp_dir).expect("cache metric temp dir should be removable"); +} + +#[tokio::test] +async fn token_cache_hit_counter_tracks_warm_results() { + let (registry, credentials, temp_dir) = token_cache_fixture("hit-metric", &["hit-01"], 2).await; + let (identity, token) = &credentials[0]; + + registry + .authorize(identity, token) + .await + .expect("cold token should authorize"); + registry + .authorize(identity, token) + .await + .expect("warm token should authorize"); + + let metrics = registry.token_verify_metrics(); + assert_eq!(metrics.token_cache_hits_total, 1); + assert_eq!(metrics.token_cache_misses_total, 1); + assert_eq!(metrics.token_cache_evictions_total, 0); + + drop(registry); + std::fs::remove_dir_all(temp_dir).expect("cache metric temp dir should be removable"); +} + +#[tokio::test] +async fn token_cache_eviction_counter_tracks_capacity_pressure() { + let (registry, credentials, temp_dir) = + token_cache_fixture("eviction-metric", &["evict-01", "evict-02"], 1).await; + + for (identity, token) in &credentials { + registry + .authorize(identity, token) + .await + .expect("token should authorize"); + } + + let metrics = registry.token_verify_metrics(); + assert_eq!(metrics.token_cache_hits_total, 0); + assert_eq!(metrics.token_cache_misses_total, 2); + assert_eq!(metrics.token_cache_evictions_total, 1); + + drop(registry); + std::fs::remove_dir_all(temp_dir).expect("cache metric temp dir should be removable"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn token_cache_prevents_redundant_argon2_verifies_on_concurrent_requests() { let unique = SystemTime::now() @@ -78,6 +142,13 @@ async fn token_cache_prevents_redundant_argon2_verifies_on_concurrent_requests() total_verifies, "warm cache hit should not run another Argon2 verify" ); + let metrics = registry.token_verify_metrics(); + assert_eq!(metrics.token_cache_misses_total, total_verifies as u64); + assert_eq!( + metrics.token_cache_hits_total + metrics.token_cache_misses_total, + 11, + "each authorization should produce exactly one cache outcome" + ); let _ = std::fs::remove_file(&path); let _ = std::fs::remove_dir(&temp_dir); @@ -198,3 +269,40 @@ async fn token_cache_distinguishes_current_and_grace_tokens_after_rotation() { let _ = std::fs::remove_file(&path); let _ = std::fs::remove_dir(&temp_dir); } + +async fn token_cache_fixture( + prefix: &str, + node_ids: &[&str], + capacity: usize, +) -> ( + NodeRegistry, + Vec<(NodeIdentity, String)>, + std::path::PathBuf, +) { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should be monotonic enough") + .as_nanos(); + let temp_dir = std::env::temp_dir().join(format!("nodelite-token-cache-{prefix}-{unique}")); + std::fs::create_dir_all(&temp_dir).expect("cache metric temp dir should exist"); + let path = temp_dir.join("server.json"); + let mut credentials = Vec::with_capacity(node_ids.len()); + for node_id in node_ids { + let issued = issue_node( + &path, + IssueNodeRequest { + node_id: (*node_id).to_string(), + node_label: Some((*node_id).to_string()), + tags: Vec::new(), + }, + ) + .await + .expect("cache metric node should be issued"); + credentials.push((identity_for(node_id), issued.node_session_token)); + } + let registry = NodeRegistry::load(&path) + .await + .expect("cache metric registry should load") + .with_token_cache_capacity_for_tests(capacity); + (registry, credentials, temp_dir) +} From 3c70c021056fae45474da77830d01b0e233a450e Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 02:36:46 +0800 Subject: [PATCH 3/8] feat(proto): configure writer scheduling Co-authored-by: Claude --- config/server.example.toml | 8 ++ nodelite-proto/src/config.rs | 26 +++- nodelite-proto/src/config/defaults.rs | 23 +++- nodelite-proto/src/config/raw.rs | 44 ++++++- .../src/config/tests/server_defaults.rs | 119 +++++++++++++++++- nodelite-proto/src/lib.rs | 16 +-- nodelite-server/src/audit/support.rs | 2 + .../src/handlers/settings/helpers.rs | 5 + nodelite-server/src/state/tests/mod.rs | 4 + nodelite-server/src/test_support.rs | 4 + .../src/tests/sanitize_snapshot_tests.rs | 16 +++ scripts/install-server.sh | 18 +++ scripts/test-install-server-config.sh | 91 +++++++++++++- 13 files changed, 354 insertions(+), 22 deletions(-) diff --git a/config/server.example.toml b/config/server.example.toml index bffa5bd..8791a09 100644 --- a/config/server.example.toml +++ b/config/server.example.toml @@ -21,6 +21,10 @@ history_db_path = "./data/history.sqlite3" history_query_concurrency = 4 # 每个历史只读连接的 SQLite 私有 page cache(KiB),允许 64-1024,不等同于 Linux filesystem page cache。 history_read_cache_kib = 512 +# 历史 writer 单次 SQLite 事务最多写入的记录数,必须至少为 1。 +history_writer_batch_max = 128 +# 历史 writer 最大攒批时间(毫秒),必须至少为 10,更小会导致高频唤醒。 +history_writer_flush_interval_ms = 100 # 运行态快照文件,用于 Server 重启后秒级恢复最近视图。 snapshot_path = "./data/snapshot.json" # 多少秒未收到心跳即视为节点离线。 @@ -73,6 +77,10 @@ enabled = true db_path = "./data/audit.sqlite3" # 仅保留最近 N 天的安全事件。 retention_days = 90 +# 审计 writer 单次 SQLite 事务最多写入的事件数,必须至少为 1。 +writer_batch_max = 128 +# 审计 writer 最大攒批时间(毫秒),必须至少为 10。 +writer_flush_interval_ms = 100 # 是否记录成功认证/握手事件。 log_successful_auth = true # 是否记录失败认证/TOTP 失败事件。 diff --git a/nodelite-proto/src/config.rs b/nodelite-proto/src/config.rs index 36c8467..16f324f 100644 --- a/nodelite-proto/src/config.rs +++ b/nodelite-proto/src/config.rs @@ -20,8 +20,10 @@ use ipnet::IpNet; use serde::{Deserialize, Serialize}; use self::defaults::{ + default_audit_writer_batch_max, default_audit_writer_flush_interval_ms, default_connect_timeout_secs, default_hello_timeout_secs, default_history_query_concurrency, - default_history_read_cache_kib, default_insecure_transport_warn_interval_secs, + default_history_read_cache_kib, default_history_writer_batch_max, + default_history_writer_flush_interval_ms, default_insecure_transport_warn_interval_secs, default_max_incoming_message_bytes, default_max_outstanding_pings, default_max_sanitized_disks, default_max_sanitized_string_bytes, default_metric_anomaly_session_limit, default_metrics_export_node_disk_metrics, default_metrics_export_node_resource_metrics, @@ -60,6 +62,10 @@ pub const DEFAULT_HISTORY_WRITE_INTERVAL_SECS: u64 = 30; pub const DEFAULT_HISTORY_QUERY_CONCURRENCY: usize = 4; /// 历史只读连接的 SQLite 私有 page cache 默认大小(KiB)。 pub const DEFAULT_HISTORY_READ_CACHE_KIB: u64 = 512; +/// 历史写入器单次事务默认最多写入的记录数。 +pub const DEFAULT_HISTORY_WRITER_BATCH_MAX: usize = 128; +/// 历史写入器默认最大攒批时间(毫秒)。 +pub const DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS: u64 = 100; /// 历史查询至少保留一个并发槽位。 pub const MIN_HISTORY_QUERY_CONCURRENCY: usize = 1; /// 防止大量并发 SQLite page cache 抬高匿名内存峰值。 @@ -106,6 +112,12 @@ pub const MIN_TOKEN_VERIFY_MAX_PARALLELISM: usize = 1; pub const MAX_TOKEN_VERIFY_MAX_PARALLELISM: usize = 8; /// 审计日志默认保留天数。 pub const DEFAULT_AUDIT_RETENTION_DAYS: u64 = 90; +/// 审计写入器单次事务默认最多写入的记录数。 +pub const DEFAULT_AUDIT_WRITER_BATCH_MAX: usize = 128; +/// 审计写入器默认最大攒批时间(毫秒)。 +pub const DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS: u64 = 100; +/// 写入器 flush 间隔下限，避免极短定时器形成忙循环。 +pub const MIN_WRITER_FLUSH_INTERVAL_MS: u64 = 10; /// GeoIP 数据库默认更新间隔(天)。 pub const DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS: u64 = 30; /// Agent 连接超时(秒)。 @@ -180,6 +192,12 @@ pub struct ServerConfig { #[serde(default = "default_history_read_cache_kib")] /// 每个历史只读连接的 SQLite 私有 page cache 大小(KiB)。 pub history_read_cache_kib: u64, + #[serde(default = "default_history_writer_batch_max")] + /// 历史写入器单次事务最多写入的记录数。 + pub history_writer_batch_max: usize, + #[serde(default = "default_history_writer_flush_interval_ms")] + /// 历史写入器最大攒批时间(毫秒)。 + pub history_writer_flush_interval_ms: u64, /// 最新快照持久化文件路径。 pub snapshot_path: PathBuf, /// 超过该秒数未收到上报后,节点视为离线。 @@ -281,6 +299,12 @@ pub struct AuditConfig { pub db_path: PathBuf, /// 审计记录保留天数。 pub retention_days: u64, + #[serde(default = "default_audit_writer_batch_max")] + /// 审计写入器单次事务最多写入的记录数。 + pub writer_batch_max: usize, + #[serde(default = "default_audit_writer_flush_interval_ms")] + /// 审计写入器最大攒批时间(毫秒)。 + pub writer_flush_interval_ms: u64, /// 是否记录成功认证事件。 pub log_successful_auth: bool, /// 是否记录失败认证事件。 diff --git a/nodelite-proto/src/config/defaults.rs b/nodelite-proto/src/config/defaults.rs index ddf37ac..b6bd771 100644 --- a/nodelite-proto/src/config/defaults.rs +++ b/nodelite-proto/src/config/defaults.rs @@ -5,8 +5,11 @@ use super::{ DEFAULT_ALERT_INSPECTION_LOCAL_TIME, DEFAULT_ALERT_INSPECTION_LOOKBACK_HOURS, DEFAULT_ALERT_INSPECTION_MEMORY_WARN_PERCENT, DEFAULT_ALERT_INSPECTION_OFFLINE_GRACE_MINUTES, DEFAULT_ALERT_RULE_COOLDOWN_MINUTES, DEFAULT_ALERT_RULE_WINDOW_MINUTES, - DEFAULT_AUDIT_RETENTION_DAYS, DEFAULT_CONNECT_TIMEOUT_SECS, DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, - DEFAULT_HELLO_TIMEOUT_SECS, DEFAULT_HISTORY_QUERY_CONCURRENCY, DEFAULT_HISTORY_READ_CACHE_KIB, + DEFAULT_AUDIT_RETENTION_DAYS, DEFAULT_AUDIT_WRITER_BATCH_MAX, + DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, DEFAULT_CONNECT_TIMEOUT_SECS, + DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, DEFAULT_HELLO_TIMEOUT_SECS, + DEFAULT_HISTORY_QUERY_CONCURRENCY, DEFAULT_HISTORY_READ_CACHE_KIB, + DEFAULT_HISTORY_WRITER_BATCH_MAX, DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, DEFAULT_INSECURE_TRANSPORT_WARN_INTERVAL_SECS, DEFAULT_MAX_INCOMING_MESSAGE_BYTES, DEFAULT_MAX_MESSAGE_BYTES, DEFAULT_MAX_OUTSTANDING_PINGS, DEFAULT_MAX_SANITIZED_DISKS, DEFAULT_MAX_SANITIZED_STRING_BYTES, DEFAULT_METRIC_ANOMALY_SESSION_LIMIT, @@ -29,6 +32,14 @@ pub(super) fn default_history_read_cache_kib() -> u64 { DEFAULT_HISTORY_READ_CACHE_KIB } +pub(super) fn default_history_writer_batch_max() -> usize { + DEFAULT_HISTORY_WRITER_BATCH_MAX +} + +pub(super) fn default_history_writer_flush_interval_ms() -> u64 { + DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS +} + pub(super) fn default_node_registry_path() -> PathBuf { PathBuf::from("./config/server.json") } @@ -137,6 +148,14 @@ pub(super) fn default_audit_retention_days() -> u64 { DEFAULT_AUDIT_RETENTION_DAYS } +pub(super) fn default_audit_writer_batch_max() -> usize { + DEFAULT_AUDIT_WRITER_BATCH_MAX +} + +pub(super) fn default_audit_writer_flush_interval_ms() -> u64 { + DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS +} + pub(super) fn default_audit_log_successful_auth() -> bool { true } diff --git a/nodelite-proto/src/config/raw.rs b/nodelite-proto/src/config/raw.rs index 9070a12..d42a1e9 100644 --- a/nodelite-proto/src/config/raw.rs +++ b/nodelite-proto/src/config/raw.rs @@ -8,11 +8,13 @@ mod alerts; use super::defaults::{ default_audit_db_path, default_audit_enabled, default_audit_log_failed_auth, default_audit_log_rate_limit, default_audit_log_successful_auth, - default_audit_log_token_events, default_audit_retention_days, default_connect_timeout_secs, + default_audit_log_token_events, default_audit_retention_days, default_audit_writer_batch_max, + default_audit_writer_flush_interval_ms, default_connect_timeout_secs, default_geoip_auto_update, default_geoip_database_path, default_geoip_edition, default_geoip_enabled, default_geoip_provider, default_geoip_update_interval_days, default_hello_timeout_secs, default_history_db_path, default_history_query_concurrency, - default_history_read_cache_kib, default_ignored_filesystems, + default_history_read_cache_kib, default_history_writer_batch_max, + default_history_writer_flush_interval_ms, default_ignored_filesystems, default_insecure_transport_warn_interval_secs, default_max_incoming_message_bytes, default_max_message_bytes, default_max_outstanding_pings, default_max_sanitized_disks, default_max_sanitized_string_bytes, default_metric_anomaly_session_limit, @@ -32,8 +34,8 @@ use super::{ AgentConfig, AlertingConfig, AuditConfig, ConfigError, GeoIpConfig, GeoIpEdition, GeoIpProvider, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, MAX_NODE_IDENTITY_TEXT_BYTES, MAX_TOKEN_VERIFY_MAX_PARALLELISM, MIN_HISTORY_QUERY_CONCURRENCY, - MIN_HISTORY_READ_CACHE_KIB, MIN_TOKEN_VERIFY_MAX_PARALLELISM, MetricsConfig, - ReadonlyAuthConfig, ServerConfig, WsConfig, + MIN_HISTORY_READ_CACHE_KIB, MIN_TOKEN_VERIFY_MAX_PARALLELISM, MIN_WRITER_FLUSH_INTERVAL_MS, + MetricsConfig, ReadonlyAuthConfig, ServerConfig, WsConfig, }; use crate::validation::{ ValidationError, normalize_string_list, validate_bounded_text, validate_identifier, @@ -88,6 +90,10 @@ struct RawServerSection { history_query_concurrency: usize, #[serde(default = "default_history_read_cache_kib")] history_read_cache_kib: u64, + #[serde(default = "default_history_writer_batch_max")] + history_writer_batch_max: usize, + #[serde(default = "default_history_writer_flush_interval_ms")] + history_writer_flush_interval_ms: u64, #[serde(default = "default_snapshot_path")] snapshot_path: PathBuf, #[serde(default = "default_stale_after_secs")] @@ -154,6 +160,10 @@ struct RawAuditSection { db_path: PathBuf, #[serde(default = "default_audit_retention_days")] retention_days: u64, + #[serde(default = "default_audit_writer_batch_max")] + writer_batch_max: usize, + #[serde(default = "default_audit_writer_flush_interval_ms")] + writer_flush_interval_ms: u64, #[serde(default = "default_audit_log_successful_auth")] log_successful_auth: bool, #[serde(default = "default_audit_log_failed_auth")] @@ -226,6 +236,8 @@ impl Default for RawAuditSection { enabled: default_audit_enabled(), db_path: default_audit_db_path(), retention_days: default_audit_retention_days(), + writer_batch_max: default_audit_writer_batch_max(), + writer_flush_interval_ms: default_audit_writer_flush_interval_ms(), log_successful_auth: default_audit_log_successful_auth(), log_failed_auth: default_audit_log_failed_auth(), log_token_events: default_audit_log_token_events(), @@ -341,6 +353,8 @@ impl RawServerConfigFile { history_db_path: self.server.history_db_path, history_query_concurrency: self.server.history_query_concurrency, history_read_cache_kib: self.server.history_read_cache_kib, + history_writer_batch_max: self.server.history_writer_batch_max, + history_writer_flush_interval_ms: self.server.history_writer_flush_interval_ms, snapshot_path: self.server.snapshot_path, stale_after_secs: self.server.stale_after_secs, ping_interval_secs: self.server.ping_interval_secs, @@ -434,11 +448,23 @@ impl RawServerConfigFile { "audit.retention_days must be greater than 0", )); } + if self.audit.writer_batch_max == 0 { + return Err(ConfigError::new( + "audit.writer_batch_max must be at least 1", + )); + } + if self.audit.writer_flush_interval_ms < MIN_WRITER_FLUSH_INTERVAL_MS { + return Err(ConfigError::new(format!( + "audit.writer_flush_interval_ms must be at least {MIN_WRITER_FLUSH_INTERVAL_MS} ms" + ))); + } Ok(AuditConfig { enabled: self.audit.enabled, db_path: self.audit.db_path.clone(), retention_days: self.audit.retention_days, + writer_batch_max: self.audit.writer_batch_max, + writer_flush_interval_ms: self.audit.writer_flush_interval_ms, log_successful_auth: self.audit.log_successful_auth, log_failed_auth: self.audit.log_failed_auth, log_token_events: self.audit.log_token_events, @@ -596,6 +622,16 @@ impl RawServerConfigFile { "server.history_read_cache_kib must be between {MIN_HISTORY_READ_CACHE_KIB} and {MAX_HISTORY_READ_CACHE_KIB} KiB", ))); } + if self.server.history_writer_batch_max == 0 { + return Err(ConfigError::new( + "server.history_writer_batch_max must be at least 1", + )); + } + if self.server.history_writer_flush_interval_ms < MIN_WRITER_FLUSH_INTERVAL_MS { + return Err(ConfigError::new(format!( + "server.history_writer_flush_interval_ms must be at least {MIN_WRITER_FLUSH_INTERVAL_MS} ms" + ))); + } Ok(()) } diff --git a/nodelite-proto/src/config/tests/server_defaults.rs b/nodelite-proto/src/config/tests/server_defaults.rs index 87960df..191707e 100644 --- a/nodelite-proto/src/config/tests/server_defaults.rs +++ b/nodelite-proto/src/config/tests/server_defaults.rs @@ -7,14 +7,16 @@ use super::super::{ DEFAULT_ALERT_INSPECTION_CPU_WARN_PERCENT, DEFAULT_ALERT_INSPECTION_LATENCY_WARN_MS, DEFAULT_ALERT_INSPECTION_LOCAL_TIME, DEFAULT_ALERT_INSPECTION_MEMORY_WARN_PERCENT, DEFAULT_ALERT_MEMORY_WINDOW_MINUTES, DEFAULT_ALERT_OFFLINE_THRESHOLD_MINUTES, - DEFAULT_ALERT_RTT_WINDOW_MINUTES, DEFAULT_AUDIT_RETENTION_DAYS, - DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, DEFAULT_HISTORY_QUERY_CONCURRENCY, - DEFAULT_HISTORY_READ_CACHE_KIB, DEFAULT_MAX_MESSAGE_BYTES, - DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, DEFAULT_WS_AUTH_BLOCK_SECS, + DEFAULT_ALERT_RTT_WINDOW_MINUTES, DEFAULT_AUDIT_RETENTION_DAYS, DEFAULT_AUDIT_WRITER_BATCH_MAX, + DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, + DEFAULT_HISTORY_QUERY_CONCURRENCY, DEFAULT_HISTORY_READ_CACHE_KIB, + DEFAULT_HISTORY_WRITER_BATCH_MAX, DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, + DEFAULT_MAX_MESSAGE_BYTES, DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, DEFAULT_WS_AUTH_BLOCK_SECS, DEFAULT_WS_AUTH_FAIL_MAX_ATTEMPTS, DEFAULT_WS_AUTH_FAIL_WINDOW_SECS, DEFAULT_WS_MAX_CONNECTIONS_PER_IP, DEFAULT_WS_MAX_TOTAL_CONNECTIONS, GeoIpEdition, GeoIpProvider, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, - MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, parse_server_config, + MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, MIN_WRITER_FLUSH_INTERVAL_MS, + parse_server_config, }; #[test] @@ -52,6 +54,20 @@ fn server_example_documents_history_query_limits() { assert!(example.contains("history_read_cache_kib")); } +#[test] +fn server_example_documents_writer_scheduling() { + let example = include_str!("../../../../config/server.example.toml"); + + for expected in [ + "history_writer_batch_max = 128", + "history_writer_flush_interval_ms = 100", + "writer_batch_max = 128", + "writer_flush_interval_ms = 100", + ] { + assert!(example.lines().any(|line| line == expected)); + } +} + #[test] fn parses_server_config_with_defaults() { let config = parse_server_config( @@ -76,6 +92,14 @@ fn parses_server_config_with_defaults() { config.history_read_cache_kib, DEFAULT_HISTORY_READ_CACHE_KIB ); + assert_eq!( + config.history_writer_batch_max, + DEFAULT_HISTORY_WRITER_BATCH_MAX + ); + assert_eq!( + config.history_writer_flush_interval_ms, + DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS + ); assert_eq!( config.ws.max_total_connections, DEFAULT_WS_MAX_TOTAL_CONNECTIONS @@ -110,6 +134,14 @@ fn parses_server_config_with_defaults() { assert!(config.audit.enabled); assert_eq!(config.audit.db_path, PathBuf::from("./data/audit.sqlite3")); assert_eq!(config.audit.retention_days, DEFAULT_AUDIT_RETENTION_DAYS); + assert_eq!( + config.audit.writer_batch_max, + DEFAULT_AUDIT_WRITER_BATCH_MAX + ); + assert_eq!( + config.audit.writer_flush_interval_ms, + DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS + ); assert!(config.audit.log_successful_auth); assert!(config.audit.log_failed_auth); assert!(config.audit.log_token_events); @@ -193,6 +225,83 @@ fn parses_history_query_resource_overrides() { assert_eq!(config.history_read_cache_kib, 1024); } +#[test] +fn parses_writer_scheduling_overrides() { + let config = parse_server_config( + r#" + [server] + listen = "127.0.0.1:8080" + public_base_url = "https://monitor.example.com" + history_writer_batch_max = 64 + history_writer_flush_interval_ms = 25 + + [audit] + writer_batch_max = 32 + writer_flush_interval_ms = 50 + "#, + ) + .expect("writer scheduling overrides should parse"); + + assert_eq!(config.history_writer_batch_max, 64); + assert_eq!(config.history_writer_flush_interval_ms, 25); + assert_eq!(config.audit.writer_batch_max, 32); + assert_eq!(config.audit.writer_flush_interval_ms, 50); +} + +#[test] +fn rejects_invalid_writer_scheduling_values() { + assert_eq!(MIN_WRITER_FLUSH_INTERVAL_MS, 10); + for (section, key, value, expected) in [ + ( + "server", + "history_writer_batch_max", + 0, + "server.history_writer_batch_max", + ), + ("audit", "writer_batch_max", 0, "audit.writer_batch_max"), + ( + "server", + "history_writer_flush_interval_ms", + 0, + "server.history_writer_flush_interval_ms", + ), + ( + "audit", + "writer_flush_interval_ms", + 0, + "audit.writer_flush_interval_ms", + ), + ( + "server", + "history_writer_flush_interval_ms", + MIN_WRITER_FLUSH_INTERVAL_MS - 1, + "server.history_writer_flush_interval_ms", + ), + ( + "audit", + "writer_flush_interval_ms", + MIN_WRITER_FLUSH_INTERVAL_MS - 1, + "audit.writer_flush_interval_ms", + ), + ] { + let setting = if section == "server" { + format!("{key} = {value}") + } else { + format!("[audit]\n{key} = {value}") + }; + let input = format!( + r#" + [server] + listen = "127.0.0.1:8080" + public_base_url = "https://monitor.example.com" + {setting} + "#, + ); + let error = parse_server_config(&input).expect_err("invalid writer setting should fail"); + assert!(error.to_string().contains(expected)); + } +} + #[test] fn rejects_history_query_resource_limits_outside_safe_ranges() { assert_eq!(MIN_HISTORY_QUERY_CONCURRENCY, 1); diff --git a/nodelite-proto/src/lib.rs b/nodelite-proto/src/lib.rs index 965199f..53f6f6e 100644 --- a/nodelite-proto/src/lib.rs +++ b/nodelite-proto/src/lib.rs @@ -20,17 +20,19 @@ pub use config::{ DEFAULT_ALERT_INSPECTION_LOOKBACK_HOURS, DEFAULT_ALERT_INSPECTION_MEMORY_WARN_PERCENT, DEFAULT_ALERT_INSPECTION_OFFLINE_GRACE_MINUTES, DEFAULT_ALERT_RULE_COOLDOWN_MINUTES, DEFAULT_ALERT_RULE_WINDOW_MINUTES, DEFAULT_AUDIT_RETENTION_DAYS, + DEFAULT_AUDIT_WRITER_BATCH_MAX, DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, DEFAULT_HISTORY_QUERY_CONCURRENCY, DEFAULT_HISTORY_READ_CACHE_KIB, DEFAULT_HISTORY_RETENTION_HOURS, - DEFAULT_HISTORY_WRITE_INTERVAL_SECS, DEFAULT_MAX_MESSAGE_BYTES, DEFAULT_PING_INTERVAL_SECS, - DEFAULT_REFRESH_INTERVAL_SECS, DEFAULT_REPORT_INTERVAL_SECS, DEFAULT_STALE_AFTER_SECS, - DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, GeoIpConfig, GeoIpEdition, GeoIpProvider, - InspectionConfig, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, + DEFAULT_HISTORY_WRITE_INTERVAL_SECS, DEFAULT_HISTORY_WRITER_BATCH_MAX, + DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, DEFAULT_MAX_MESSAGE_BYTES, + DEFAULT_PING_INTERVAL_SECS, DEFAULT_REFRESH_INTERVAL_SECS, DEFAULT_REPORT_INTERVAL_SECS, + DEFAULT_STALE_AFTER_SECS, DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, GeoIpConfig, GeoIpEdition, + GeoIpProvider, InspectionConfig, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, MAX_NODE_IDENTITY_TEXT_BYTES, MAX_NODE_TAG_BYTES, MAX_NODE_TAGS, MAX_TOKEN_VERIFY_MAX_PARALLELISM, MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, - MIN_TOKEN_VERIFY_MAX_PARALLELISM, MetricsConfig, ReadonlyAuthConfig, ServerConfig, WsConfig, - normalize_totp_secret, parse_agent_config, parse_server_config, - upsert_toml_item_preserving_decor, + MIN_TOKEN_VERIFY_MAX_PARALLELISM, MIN_WRITER_FLUSH_INTERVAL_MS, MetricsConfig, + ReadonlyAuthConfig, ServerConfig, WsConfig, normalize_totp_secret, parse_agent_config, + parse_server_config, upsert_toml_item_preserving_decor, }; pub use message::{ AgentLogEntry, AgentLogsMessage, BrowserMessage, HelloMessage, diff --git a/nodelite-server/src/audit/support.rs b/nodelite-server/src/audit/support.rs index a8db0f6..fb31325 100644 --- a/nodelite-server/src/audit/support.rs +++ b/nodelite-server/src/audit/support.rs @@ -14,6 +14,8 @@ pub(super) fn sample_config(db_path: PathBuf) -> nodelite_proto::AuditConfig { enabled: true, db_path, retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, diff --git a/nodelite-server/src/handlers/settings/helpers.rs b/nodelite-server/src/handlers/settings/helpers.rs index 060922d..fd00d36 100644 --- a/nodelite-server/src/handlers/settings/helpers.rs +++ b/nodelite-server/src/handlers/settings/helpers.rs @@ -346,6 +346,8 @@ mod tests { enabled: true, db_path: PathBuf::from("/opt/nodelite/data/audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -364,6 +366,9 @@ mod tests { history_db_path: PathBuf::from("/opt/nodelite/data/history.sqlite3"), history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: + nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path: PathBuf::from("/opt/nodelite/data/snapshot.json"), stale_after_secs: 15, ping_interval_secs: 60, diff --git a/nodelite-server/src/state/tests/mod.rs b/nodelite-server/src/state/tests/mod.rs index 92971da..10c0277 100644 --- a/nodelite-server/src/state/tests/mod.rs +++ b/nodelite-server/src/state/tests/mod.rs @@ -57,6 +57,8 @@ fn sample_config() -> ServerConfig { enabled: true, db_path: PathBuf::from("/tmp/nodelite-test-audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -75,6 +77,8 @@ fn sample_config() -> ServerConfig { history_db_path: PathBuf::from("/tmp/nodelite-test-history.sqlite3"), history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path: PathBuf::from("/tmp/nodelite-test-snapshot.json"), stale_after_secs: 5, ping_interval_secs: 60, diff --git a/nodelite-server/src/test_support.rs b/nodelite-server/src/test_support.rs index 93388d5..4004c6e 100644 --- a/nodelite-server/src/test_support.rs +++ b/nodelite-server/src/test_support.rs @@ -79,6 +79,8 @@ pub(crate) fn test_server_config( enabled: true, db_path: history_path.with_file_name("audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -97,6 +99,8 @@ pub(crate) fn test_server_config( history_db_path: history_path, history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path, stale_after_secs: 5, ping_interval_secs: 60, diff --git a/nodelite-server/src/tests/sanitize_snapshot_tests.rs b/nodelite-server/src/tests/sanitize_snapshot_tests.rs index 8ed7c66..b17bc72 100644 --- a/nodelite-server/src/tests/sanitize_snapshot_tests.rs +++ b/nodelite-server/src/tests/sanitize_snapshot_tests.rs @@ -33,6 +33,8 @@ fn sanitize_snapshot_clamps_invalid_metrics() { enabled: true, db_path: PathBuf::from("./data/audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -51,6 +53,8 @@ fn sanitize_snapshot_clamps_invalid_metrics() { history_db_path: PathBuf::from("./data/history.sqlite3"), history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path: PathBuf::from("./data/snapshot.json"), stale_after_secs: 15, ping_interval_secs: 5, @@ -211,6 +215,8 @@ fn sanitize_caps_disk_field_string_length() { enabled: true, db_path: PathBuf::from("./data/audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -229,6 +235,8 @@ fn sanitize_caps_disk_field_string_length() { history_db_path: PathBuf::from("./data/history.sqlite3"), history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path: PathBuf::from("./data/snapshot.json"), stale_after_secs: 15, ping_interval_secs: 5, @@ -311,6 +319,8 @@ fn sanitize_snapshot_caps_disk_count_and_tracks_clean_reports() { enabled: true, db_path: PathBuf::from("./data/audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -329,6 +339,8 @@ fn sanitize_snapshot_caps_disk_count_and_tracks_clean_reports() { history_db_path: PathBuf::from("./data/history.sqlite3"), history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path: PathBuf::from("./data/snapshot.json"), stale_after_secs: 15, ping_interval_secs: 5, @@ -426,6 +438,8 @@ fn sanitize_snapshot_deduplicates_repeated_disk_devices() { enabled: true, db_path: PathBuf::from("./data/audit.sqlite3"), retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, log_successful_auth: true, log_failed_auth: true, log_token_events: true, @@ -444,6 +458,8 @@ fn sanitize_snapshot_deduplicates_repeated_disk_devices() { history_db_path: PathBuf::from("./data/history.sqlite3"), history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, snapshot_path: PathBuf::from("./data/snapshot.json"), stale_after_secs: 15, ping_interval_secs: 5, diff --git a/scripts/install-server.sh b/scripts/install-server.sh index 5fe8eda..4dd2675 100644 --- a/scripts/install-server.sh +++ b/scripts/install-server.sh @@ -558,6 +558,14 @@ load_existing_server_defaults() { [ -n "$value" ] && SERVER_HISTORY_QUERY_CONCURRENCY="$value" value="$(trim_whitespace "$(toml_get_raw "$config_path" server history_read_cache_kib)")" [ -n "$value" ] && SERVER_HISTORY_READ_CACHE_KIB="$value" + value="$(trim_whitespace "$(toml_get_raw "$config_path" server history_writer_batch_max)")" + [ -n "$value" ] && SERVER_HISTORY_WRITER_BATCH_MAX="$value" + value="$(trim_whitespace "$(toml_get_raw "$config_path" server history_writer_flush_interval_ms)")" + [ -n "$value" ] && SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS="$value" + value="$(trim_whitespace "$(toml_get_raw "$config_path" audit writer_batch_max)")" + [ -n "$value" ] && AUDIT_WRITER_BATCH_MAX="$value" + value="$(trim_whitespace "$(toml_get_raw "$config_path" audit writer_flush_interval_ms)")" + [ -n "$value" ] && AUDIT_WRITER_FLUSH_INTERVAL_MS="$value" value="$(trim_whitespace "$(toml_get_raw "$config_path" ws max_total_connections)")" [ -n "$value" ] && WS_MAX_TOTAL_CONNECTIONS="$value" value="$(trim_whitespace "$(toml_get_raw "$config_path" ws max_connections_per_ip)")" @@ -599,6 +607,8 @@ complete_server_config_defaults() { ensure_toml_default "$config_path" server history_db_path "history_db_path = \"$DATA_DIR/history.sqlite3\"" ensure_toml_default "$config_path" server history_query_concurrency "history_query_concurrency = $SERVER_HISTORY_QUERY_CONCURRENCY" ensure_toml_default "$config_path" server history_read_cache_kib "history_read_cache_kib = $SERVER_HISTORY_READ_CACHE_KIB" + ensure_toml_default "$config_path" server history_writer_batch_max "history_writer_batch_max = $SERVER_HISTORY_WRITER_BATCH_MAX" + ensure_toml_default "$config_path" server history_writer_flush_interval_ms "history_writer_flush_interval_ms = $SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS" ensure_toml_default "$config_path" server snapshot_path "snapshot_path = \"$DATA_DIR/snapshot.json\"" ensure_toml_default "$config_path" server stale_after_secs "stale_after_secs = $SERVER_STALE_AFTER_SECS" ensure_toml_default "$config_path" server ping_interval_secs "ping_interval_secs = $SERVER_PING_INTERVAL_SECS" @@ -620,6 +630,8 @@ complete_server_config_defaults() { ensure_toml_default "$config_path" audit enabled "enabled = $AUDIT_ENABLED" ensure_toml_default "$config_path" audit db_path "db_path = \"$audit_db_path\"" ensure_toml_default "$config_path" audit retention_days "retention_days = $AUDIT_RETENTION_DAYS" + ensure_toml_default "$config_path" audit writer_batch_max "writer_batch_max = $AUDIT_WRITER_BATCH_MAX" + ensure_toml_default "$config_path" audit writer_flush_interval_ms "writer_flush_interval_ms = $AUDIT_WRITER_FLUSH_INTERVAL_MS" ensure_toml_default "$config_path" audit log_successful_auth "log_successful_auth = $AUDIT_LOG_SUCCESSFUL_AUTH" ensure_toml_default "$config_path" audit log_failed_auth "log_failed_auth = $AUDIT_LOG_FAILED_AUTH" ensure_toml_default "$config_path" audit log_token_events "log_token_events = $AUDIT_LOG_TOKEN_EVENTS" @@ -682,6 +694,8 @@ node_registry_path = "${CONFIG_DIR}/server.json" history_db_path = "${DATA_DIR}/history.sqlite3" history_query_concurrency = ${SERVER_HISTORY_QUERY_CONCURRENCY} history_read_cache_kib = ${SERVER_HISTORY_READ_CACHE_KIB} +history_writer_batch_max = ${SERVER_HISTORY_WRITER_BATCH_MAX} +history_writer_flush_interval_ms = ${SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS} snapshot_path = "${DATA_DIR}/snapshot.json" stale_after_secs = ${SERVER_STALE_AFTER_SECS} ping_interval_secs = ${SERVER_PING_INTERVAL_SECS} @@ -802,6 +816,8 @@ SERVER_PING_INTERVAL_SECS="10" SERVER_MAX_MESSAGE_BYTES="65536" SERVER_HISTORY_QUERY_CONCURRENCY="4" SERVER_HISTORY_READ_CACHE_KIB="512" +SERVER_HISTORY_WRITER_BATCH_MAX="128" +SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS="100" SERVER_HELLO_TIMEOUT_SECS="10" SERVER_MAX_OUTSTANDING_PINGS="32" SERVER_INSECURE_TRANSPORT_WARN_INTERVAL_SECS="900" @@ -819,6 +835,8 @@ METRICS_EXPORT_NODE_RESOURCE_METRICS="false" METRICS_EXPORT_NODE_DISK_METRICS="false" AUDIT_ENABLED="true" AUDIT_RETENTION_DAYS="90" +AUDIT_WRITER_BATCH_MAX="128" +AUDIT_WRITER_FLUSH_INTERVAL_MS="100" AUDIT_LOG_SUCCESSFUL_AUTH="true" AUDIT_LOG_FAILED_AUTH="true" AUDIT_LOG_TOKEN_EVENTS="true" diff --git a/scripts/test-install-server-config.sh b/scripts/test-install-server-config.sh index 1816712..dc5079e 100644 --- a/scripts/test-install-server-config.sh +++ b/scripts/test-install-server-config.sh @@ -36,7 +36,11 @@ SERVER_PING_INTERVAL_SECS="10" SERVER_MAX_MESSAGE_BYTES="65536" SERVER_HISTORY_QUERY_CONCURRENCY="4" SERVER_HISTORY_READ_CACHE_KIB="512" +SERVER_HISTORY_WRITER_BATCH_MAX="128" +SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS="100" SERVER_TOKEN_VERIFY_MAX_PARALLELISM="4" +AUDIT_WRITER_BATCH_MAX="128" +AUDIT_WRITER_FLUSH_INTERVAL_MS="100" WS_MAX_TOTAL_CONNECTIONS="1024" WS_MAX_CONNECTIONS_PER_IP="32" WS_AUTH_FAIL_WINDOW_SECS="300" @@ -58,16 +62,29 @@ printf '%s\n' "$fresh_config" | grep -Fx 'history_query_concurrency = 4' >/dev/null printf '%s\n' "$fresh_config" | grep -Fx 'history_read_cache_kib = 512' >/dev/null +printf '%s\n' "$fresh_config" | + grep -Fx 'history_writer_batch_max = 128' >/dev/null +printf '%s\n' "$fresh_config" | + grep -Fx 'history_writer_flush_interval_ms = 100' >/dev/null existing_config="$TEMP_DIR/existing.toml" printf '%s\n' \ '[server]' \ 'token_verify_max_parallelism = 7' \ 'history_query_concurrency = 6' \ - 'history_read_cache_kib = 768' >"$existing_config" + 'history_read_cache_kib = 768' \ + 'history_writer_batch_max = 64' \ + 'history_writer_flush_interval_ms = 40' \ + '[audit]' \ + 'writer_batch_max = 32' \ + 'writer_flush_interval_ms = 25' >"$existing_config" SERVER_TOKEN_VERIFY_MAX_PARALLELISM="4" SERVER_HISTORY_QUERY_CONCURRENCY="4" SERVER_HISTORY_READ_CACHE_KIB="512" +SERVER_HISTORY_WRITER_BATCH_MAX="128" +SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS="100" +AUDIT_WRITER_BATCH_MAX="128" +AUDIT_WRITER_FLUSH_INTERVAL_MS="100" load_existing_server_defaults "$existing_config" [ "$SERVER_TOKEN_VERIFY_MAX_PARALLELISM" = "7" ] || { printf '%s\n' "upgrade defaults did not preserve token verify parallelism" >&2 @@ -81,6 +98,22 @@ load_existing_server_defaults "$existing_config" printf '%s\n' "upgrade defaults did not preserve history read cache" >&2 exit 1 } +[ "$SERVER_HISTORY_WRITER_BATCH_MAX" = "64" ] || { + printf '%s\n' "upgrade defaults did not preserve history writer batch max" >&2 + exit 1 +} +[ "$SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS" = "40" ] || { + printf '%s\n' "upgrade defaults did not preserve history writer flush interval" >&2 + exit 1 +} +[ "$AUDIT_WRITER_BATCH_MAX" = "32" ] || { + printf '%s\n' "upgrade defaults did not preserve audit writer batch max" >&2 + exit 1 +} +[ "$AUDIT_WRITER_FLUSH_INTERVAL_MS" = "25" ] || { + printf '%s\n' "upgrade defaults did not preserve audit writer flush interval" >&2 + exit 1 +} missing_config="$TEMP_DIR/missing.toml" printf '%s\n' '[server]' 'listen = "127.0.0.1:20000"' >"$missing_config" @@ -89,6 +122,10 @@ CONFIG_DEFAULTS_ADDED=0 SERVER_TOKEN_VERIFY_MAX_PARALLELISM="4" SERVER_HISTORY_QUERY_CONCURRENCY="4" SERVER_HISTORY_READ_CACHE_KIB="512" +SERVER_HISTORY_WRITER_BATCH_MAX="128" +SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS="100" +AUDIT_WRITER_BATCH_MAX="128" +AUDIT_WRITER_FLUSH_INTERVAL_MS="100" ensure_toml_default \ "$missing_config" server token_verify_max_parallelism \ "token_verify_max_parallelism = $SERVER_TOKEN_VERIFY_MAX_PARALLELISM" @@ -103,10 +140,26 @@ ensure_toml_default \ ensure_toml_default \ "$missing_config" server history_read_cache_kib \ "history_read_cache_kib = $SERVER_HISTORY_READ_CACHE_KIB" +ensure_toml_default \ + "$missing_config" server history_writer_batch_max \ + "history_writer_batch_max = $SERVER_HISTORY_WRITER_BATCH_MAX" +ensure_toml_default \ + "$missing_config" server history_writer_flush_interval_ms \ + "history_writer_flush_interval_ms = $SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS" +ensure_toml_default \ + "$missing_config" audit writer_batch_max \ + "writer_batch_max = $AUDIT_WRITER_BATCH_MAX" +ensure_toml_default \ + "$missing_config" audit writer_flush_interval_ms \ + "writer_flush_interval_ms = $AUDIT_WRITER_FLUSH_INTERVAL_MS" grep -Fx 'history_query_concurrency = 4' "$missing_config" >/dev/null grep -Fx 'history_read_cache_kib = 512' "$missing_config" >/dev/null -[ "$CONFIG_DEFAULTS_ADDED" -eq 3 ] || { - printf '%s\n' "missing history query defaults were not counted" >&2 +grep -Fx 'history_writer_batch_max = 128' "$missing_config" >/dev/null +grep -Fx 'history_writer_flush_interval_ms = 100' "$missing_config" >/dev/null +grep -Fx 'writer_batch_max = 128' "$missing_config" >/dev/null +grep -Fx 'writer_flush_interval_ms = 100' "$missing_config" >/dev/null +[ "$CONFIG_DEFAULTS_ADDED" -eq 7 ] || { + printf '%s\n' "missing writer defaults were not counted" >&2 exit 1 } @@ -122,6 +175,22 @@ ensure_toml_default \ "history_read_cache_kib = $SERVER_HISTORY_READ_CACHE_KIB" grep -Fx 'history_query_concurrency = 6' "$existing_config" >/dev/null grep -Fx 'history_read_cache_kib = 768' "$existing_config" >/dev/null +ensure_toml_default \ + "$existing_config" server history_writer_batch_max \ + "history_writer_batch_max = $SERVER_HISTORY_WRITER_BATCH_MAX" +ensure_toml_default \ + "$existing_config" server history_writer_flush_interval_ms \ + "history_writer_flush_interval_ms = $SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS" +ensure_toml_default \ + "$existing_config" audit writer_batch_max \ + "writer_batch_max = $AUDIT_WRITER_BATCH_MAX" +ensure_toml_default \ + "$existing_config" audit writer_flush_interval_ms \ + "writer_flush_interval_ms = $AUDIT_WRITER_FLUSH_INTERVAL_MS" +grep -Fx 'history_writer_batch_max = 64' "$existing_config" >/dev/null +grep -Fx 'history_writer_flush_interval_ms = 40' "$existing_config" >/dev/null +grep -Fx 'writer_batch_max = 32' "$existing_config" >/dev/null +grep -Fx 'writer_flush_interval_ms = 25' "$existing_config" >/dev/null # shellcheck disable=SC2016 grep -F \ @@ -135,3 +204,19 @@ grep -F \ grep -F \ 'ensure_toml_default "$config_path" server history_read_cache_kib "history_read_cache_kib = $SERVER_HISTORY_READ_CACHE_KIB"' \ "$INSTALLER" >/dev/null +# shellcheck disable=SC2016 +grep -F \ + 'ensure_toml_default "$config_path" server history_writer_batch_max "history_writer_batch_max = $SERVER_HISTORY_WRITER_BATCH_MAX"' \ + "$INSTALLER" >/dev/null +# shellcheck disable=SC2016 +grep -F \ + 'ensure_toml_default "$config_path" server history_writer_flush_interval_ms "history_writer_flush_interval_ms = $SERVER_HISTORY_WRITER_FLUSH_INTERVAL_MS"' \ + "$INSTALLER" >/dev/null +# shellcheck disable=SC2016 +grep -F \ + 'ensure_toml_default "$config_path" audit writer_batch_max "writer_batch_max = $AUDIT_WRITER_BATCH_MAX"' \ + "$INSTALLER" >/dev/null +# shellcheck disable=SC2016 +grep -F \ + 'ensure_toml_default "$config_path" audit writer_flush_interval_ms "writer_flush_interval_ms = $AUDIT_WRITER_FLUSH_INTERVAL_MS"' \ + "$INSTALLER" >/dev/null From 1170f4f958f214445ec0f846e67f52397e8251d0 Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 02:47:23 +0800 Subject: [PATCH 4/8] feat(server): apply history writer scheduling config Co-authored-by: Claude --- nodelite-server/src/app_state.rs | 4 +- nodelite-server/src/history.rs | 54 +++++++++-- .../src/history/tests/writer_tests.rs | 94 +++++++++++++++++++ nodelite-server/src/history/writer.rs | 14 +-- nodelite-server/src/startup.rs | 4 +- 5 files changed, 152 insertions(+), 18 deletions(-) diff --git a/nodelite-server/src/app_state.rs b/nodelite-server/src/app_state.rs index 1c76bbc..c5d3816 100644 --- a/nodelite-server/src/app_state.rs +++ b/nodelite-server/src/app_state.rs @@ -101,11 +101,13 @@ impl AppState { config: Arc, config_path: Arc, ) -> anyhow::Result { - let history = HistoryStore::new( + let history = HistoryStore::new_with_writer_schedule( config.history_db_path.clone(), config.sqlite_busy_timeout_secs, config.history_query_concurrency, config.history_read_cache_kib, + config.history_writer_batch_max, + std::time::Duration::from_millis(config.history_writer_flush_interval_ms), ); history.initialize().await; let audit_log = AuditLog::new(config.audit.clone(), config.sqlite_busy_timeout_secs); diff --git a/nodelite-server/src/history.rs b/nodelite-server/src/history.rs index b2d0ac4..fc9165a 100644 --- a/nodelite-server/src/history.rs +++ b/nodelite-server/src/history.rs @@ -28,7 +28,9 @@ use std::time::{Duration, Instant}; use anyhow::{Context, anyhow}; use chrono::{DateTime, Utc}; use nodelite_proto::{ - DEFAULT_HISTORY_RETENTION_HOURS, DEFAULT_HISTORY_WRITE_INTERVAL_SECS, HistoryPoint, NodeStatus, + DEFAULT_HISTORY_RETENTION_HOURS, DEFAULT_HISTORY_WRITE_INTERVAL_SECS, + DEFAULT_HISTORY_WRITER_BATCH_MAX, DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, HistoryPoint, + NodeStatus, }; use parking_lot::Mutex as ParkingLotMutex; use rusqlite::Connection; @@ -61,12 +63,6 @@ const SQLITE_BUSY_RETRY_MAX_MS: u64 = 1_000; /// 在 100ms flush 间隔下,稳态 backlog ≤ ~100。1024 留出 10x 余量 /// (突发重连 / Burst 上报),仍然不至于让 record_status 真的被反压。 const HISTORY_CHANNEL_CAPACITY: usize = 1024; -/// 一次事务里最多打包多少条 INSERT。给上限是为了在极端 burst 下 -/// 防止单个 fsync 周期被一直推迟。 -const HISTORY_BATCH_MAX: usize = 128; -/// 没有新写入到达时,writer 也每隔这段时间 flush 一次当前 batch, -/// 保证最后几条样本不会无限期停留在内存。 -const HISTORY_BATCH_FLUSH_INTERVAL: Duration = Duration::from_millis(100); /// 距上一次 DELETE 至少要过这么长时间才再次触发清理。 const HISTORY_PRUNE_MIN_INTERVAL: Duration = Duration::from_secs(300); /// 条目数只作为异常小对象下的第二道防线,主要内存约束由字节预算提供。 @@ -129,6 +125,10 @@ pub struct HistoryStore { sqlite_busy_timeout_secs: u64, /// 每个临时只读连接的 SQLite 私有 page cache(KiB)。None 仅供测试旧版默认值。 history_read_cache_kib: Option, + /// 一次事务里最多打包多少条 INSERT。 + writer_batch_max: usize, + /// 当前 batch 在内存中的最长停留时间。 + writer_flush_interval: Duration, /// Writer task 的入口。Some 表示已初始化;关停时清空以让 record_status 进入空操作分支。 writer_tx: Arc>>>, /// Writer task 的 join handle,用于在关停时显式 await。 @@ -149,28 +149,58 @@ impl HistoryStore { query_limit: usize, history_read_cache_kib: u64, ) -> Self { - Self::new_with_read_cache( + Self::new_with_options( db_path, sqlite_busy_timeout_secs, query_limit, Some(history_read_cache_kib), + DEFAULT_HISTORY_WRITER_BATCH_MAX, + Duration::from_millis(DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS), ) } + pub(crate) fn new_with_writer_schedule( + db_path: PathBuf, + sqlite_busy_timeout_secs: u64, + query_limit: usize, + history_read_cache_kib: u64, + writer_batch_max: usize, + writer_flush_interval: Duration, + ) -> Self { + let mut store = Self::new( + db_path, + sqlite_busy_timeout_secs, + query_limit, + history_read_cache_kib, + ); + store.writer_batch_max = writer_batch_max; + store.writer_flush_interval = writer_flush_interval; + store + } + #[cfg(test)] pub(crate) fn new_with_default_read_cache( db_path: PathBuf, sqlite_busy_timeout_secs: u64, query_limit: usize, ) -> Self { - Self::new_with_read_cache(db_path, sqlite_busy_timeout_secs, query_limit, None) + Self::new_with_options( + db_path, + sqlite_busy_timeout_secs, + query_limit, + None, + DEFAULT_HISTORY_WRITER_BATCH_MAX, + Duration::from_millis(DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS), + ) } - fn new_with_read_cache( + fn new_with_options( db_path: PathBuf, sqlite_busy_timeout_secs: u64, query_limit: usize, history_read_cache_kib: Option, + writer_batch_max: usize, + writer_flush_interval: Duration, ) -> Self { Self { db_path: Arc::new(db_path), @@ -181,6 +211,8 @@ impl HistoryStore { artifacts_hardened_after_write: Arc::new(AtomicBool::new(false)), sqlite_busy_timeout_secs, history_read_cache_kib, + writer_batch_max, + writer_flush_interval, writer_tx: Arc::new(RwLock::new(None)), writer_handle: Arc::new(Mutex::new(None)), dropped_writes: Arc::new(AtomicU64::new(0)), @@ -246,6 +278,8 @@ impl HistoryStore { write_connection: Arc::clone(&self.write_connection), last_pruned_at: Arc::clone(&self.last_pruned_at), artifacts_hardened_after_write: Arc::clone(&self.artifacts_hardened_after_write), + batch_max: self.writer_batch_max, + flush_interval: self.writer_flush_interval, }; let handle = tokio::spawn(run_history_writer(rx, context)); let mut guard = self.writer_handle.lock().await; diff --git a/nodelite-server/src/history/tests/writer_tests.rs b/nodelite-server/src/history/tests/writer_tests.rs index 7cd64a7..e171213 100644 --- a/nodelite-server/src/history/tests/writer_tests.rs +++ b/nodelite-server/src/history/tests/writer_tests.rs @@ -1,5 +1,28 @@ use super::*; +fn persisted_history_point_count(db_path: &PathBuf) -> i64 { + let connection = rusqlite::Connection::open(db_path).expect("history database should open"); + connection + .query_row("SELECT COUNT(*) FROM history_points", [], |row| row.get(0)) + .expect("history count query should succeed") +} + +async fn wait_for_persisted_history_point_count(db_path: &PathBuf, expected: i64) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1); + loop { + let actual = persisted_history_point_count(db_path); + if actual == expected { + return; + } + assert!( + std::time::Instant::now() < deadline, + "expected {expected} persisted history points, found {actual}" + ); + tokio::task::yield_now().await; + std::thread::sleep(std::time::Duration::from_millis(1)); + } +} + #[test] fn forget_missing_prunes_retired_nodes_from_write_throttle_state() { let runtime = Runtime::new().expect("runtime should build"); @@ -60,6 +83,77 @@ async fn record_status_flushes_through_writer_task_to_sqlite() { } } +#[tokio::test(start_paused = true)] +async fn history_writer_flushes_when_configured_batch_max_is_reached() { + let db_path = temp_history_db_path("writer-configured-batch"); + let store = HistoryStore::new_with_writer_schedule( + db_path.clone(), + 5, + DEFAULT_HISTORY_QUERY_CONCURRENCY, + DEFAULT_HISTORY_READ_CACHE_KIB, + 2, + std::time::Duration::from_secs(60), + ); + store.initialize().await; + assert!(store.is_available()); + tokio::task::yield_now().await; + + let paused_at = tokio::time::Instant::now(); + store + .record_status(&fake_status_for("batch-node-01", Utc::now())) + .await; + tokio::task::yield_now().await; + assert_eq!(persisted_history_point_count(&db_path), 0); + + store + .record_status(&fake_status_for("batch-node-02", Utc::now())) + .await; + wait_for_persisted_history_point_count(&db_path, 2).await; + assert_eq!(tokio::time::Instant::now(), paused_at); + + store.shutdown().await; + let _ = std::fs::remove_file(&db_path); + if let Some(parent) = db_path.parent() { + let _ = std::fs::remove_dir(parent); + } +} + +#[tokio::test(start_paused = true)] +async fn history_writer_flushes_at_configured_interval() { + let db_path = temp_history_db_path("writer-configured-interval"); + let store = HistoryStore::new_with_writer_schedule( + db_path.clone(), + 5, + DEFAULT_HISTORY_QUERY_CONCURRENCY, + DEFAULT_HISTORY_READ_CACHE_KIB, + 128, + std::time::Duration::from_millis(10), + ); + store.initialize().await; + assert!(store.is_available()); + tokio::task::yield_now().await; + + store + .record_status(&fake_status_for("interval-node", Utc::now())) + .await; + tokio::task::yield_now().await; + assert_eq!(persisted_history_point_count(&db_path), 0); + + let before_advance = tokio::time::Instant::now(); + tokio::time::advance(std::time::Duration::from_millis(10)).await; + wait_for_persisted_history_point_count(&db_path, 1).await; + assert_eq!( + tokio::time::Instant::now(), + before_advance + std::time::Duration::from_millis(10) + ); + + store.shutdown().await; + let _ = std::fs::remove_file(&db_path); + if let Some(parent) = db_path.parent() { + let _ = std::fs::remove_dir(parent); + } +} + #[tokio::test] async fn record_status_does_not_throttle_after_queue_full_drop() { let db_path = temp_history_db_path("queue-full-throttle"); diff --git a/nodelite-server/src/history/writer.rs b/nodelite-server/src/history/writer.rs index 4f6c64e..9733cfe 100644 --- a/nodelite-server/src/history/writer.rs +++ b/nodelite-server/src/history/writer.rs @@ -14,8 +14,8 @@ use tokio::time::MissedTickBehavior; use tracing::{debug, warn}; use super::{ - HISTORY_BATCH_FLUSH_INTERVAL, HISTORY_BATCH_MAX, HISTORY_PRUNE_MIN_INTERVAL, - SQLITE_BUSY_MAX_RETRIES, SQLITE_BUSY_RETRY_BASE_MS, SQLITE_BUSY_RETRY_MAX_MS, + HISTORY_PRUNE_MIN_INTERVAL, SQLITE_BUSY_MAX_RETRIES, SQLITE_BUSY_RETRY_BASE_MS, + SQLITE_BUSY_RETRY_MAX_MS, }; use crate::history::init::harden_database_artifacts; @@ -25,6 +25,8 @@ pub(super) struct WriterContext { pub(super) write_connection: Arc>>, pub(super) last_pruned_at: Arc, pub(super) artifacts_hardened_after_write: Arc, + pub(super) batch_max: usize, + pub(super) flush_interval: Duration, } /// 单一 writer 任务:对外的所有 record_status 都经过 mpsc 灌入此处。 @@ -39,8 +41,8 @@ pub(super) async fn run_history_writer( mut rx: mpsc::Receiver, context: WriterContext, ) { - let mut batch: Vec = Vec::with_capacity(HISTORY_BATCH_MAX); - let mut flush_timer = tokio::time::interval(HISTORY_BATCH_FLUSH_INTERVAL); + let mut batch: Vec = Vec::with_capacity(context.batch_max); + let mut flush_timer = tokio::time::interval(context.flush_interval); flush_timer.set_missed_tick_behavior(MissedTickBehavior::Delay); flush_timer.tick().await; @@ -50,7 +52,7 @@ pub(super) async fn run_history_writer( received = rx.recv() => match received { Some(point) => { batch.push(point); - if batch.len() >= HISTORY_BATCH_MAX { + if batch.len() >= context.batch_max { flush_history_batch(&mut batch, &context).await; } } @@ -66,7 +68,7 @@ pub(super) async fn run_history_writer( while let Ok(point) = rx.try_recv() { batch.push(point); - if batch.len() >= HISTORY_BATCH_MAX { + if batch.len() >= context.batch_max { flush_history_batch(&mut batch, &context).await; } } diff --git a/nodelite-server/src/startup.rs b/nodelite-server/src/startup.rs index 18305b7..3028736 100644 --- a/nodelite-server/src/startup.rs +++ b/nodelite-server/src/startup.rs @@ -138,11 +138,13 @@ async fn initialize_server_runtime( ) })?; let shared = SharedState::new(Arc::clone(&config)); - let history = HistoryStore::new( + let history = HistoryStore::new_with_writer_schedule( config.history_db_path.clone(), config.sqlite_busy_timeout_secs, config.history_query_concurrency, config.history_read_cache_kib, + config.history_writer_batch_max, + Duration::from_millis(config.history_writer_flush_interval_ms), ); let audit_log = AuditLog::new(config.audit.clone(), config.sqlite_busy_timeout_secs); history.initialize().await; From 566d8e45a43d807db030701a688339ec6ee32188 Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 02:50:24 +0800 Subject: [PATCH 5/8] feat(server): apply audit writer scheduling config Co-authored-by: Claude --- nodelite-server/src/audit/log.rs | 3 + nodelite-server/src/audit/tests.rs | 99 +++++++++++++++++++++++++++++ nodelite-server/src/audit/writer.rs | 13 ++-- 3 files changed, 108 insertions(+), 7 deletions(-) diff --git a/nodelite-server/src/audit/log.rs b/nodelite-server/src/audit/log.rs index 8c01a88..f60daf9 100644 --- a/nodelite-server/src/audit/log.rs +++ b/nodelite-server/src/audit/log.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; use anyhow::{Context, Result, anyhow}; use nodelite_proto::AuditConfig; @@ -76,6 +77,8 @@ impl AuditLog { let context = AuditWriterContext { connection: Arc::clone(&self.connection), write_failures: Arc::clone(&self.write_failures), + batch_max: self.config.writer_batch_max, + flush_interval: Duration::from_millis(self.config.writer_flush_interval_ms), }; let handle = tokio::spawn(run_audit_writer(rx, context)); let mut guard = self.writer_handle.lock().await; diff --git a/nodelite-server/src/audit/tests.rs b/nodelite-server/src/audit/tests.rs index 4b57d70..529e2f8 100644 --- a/nodelite-server/src/audit/tests.rs +++ b/nodelite-server/src/audit/tests.rs @@ -5,6 +5,29 @@ use tokio::runtime::Runtime; use super::{AuditEventType, AuditLog, AuditQuery, NewAuditEvent}; use crate::audit::support::{sample_config, unique_temp_dir}; +fn persisted_audit_event_count(db_path: &std::path::Path) -> i64 { + let connection = rusqlite::Connection::open(db_path).expect("audit database should open"); + connection + .query_row("SELECT COUNT(*) FROM audit_log", [], |row| row.get(0)) + .expect("audit count query should succeed") +} + +async fn wait_for_persisted_audit_event_count(db_path: &std::path::Path, expected: i64) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1); + loop { + let actual = persisted_audit_event_count(db_path); + if actual == expected { + return; + } + assert!( + std::time::Instant::now() < deadline, + "expected {expected} persisted audit events, found {actual}" + ); + tokio::task::yield_now().await; + std::thread::sleep(std::time::Duration::from_millis(1)); + } +} + #[test] fn audit_log_round_trips_and_filters_events() { let runtime = Runtime::new().expect("runtime should build"); @@ -230,3 +253,79 @@ fn audit_log_drains_burst_writes_through_writer_task() { let _ = std::fs::remove_dir_all(&temp_dir); }); } + +#[tokio::test(start_paused = true)] +async fn audit_writer_flushes_when_configured_batch_max_is_reached() { + let temp_dir = unique_temp_dir("nodelite-audit-configured-batch"); + std::fs::create_dir_all(&temp_dir).expect("temp dir should exist"); + let db_path = temp_dir.join("audit.sqlite3"); + let mut config = sample_config(db_path.clone()); + config.writer_batch_max = 2; + config.writer_flush_interval_ms = 60_000; + let audit = AuditLog::new(config, 5); + audit.initialize().await.expect("audit should initialize"); + tokio::task::yield_now().await; + + let paused_at = tokio::time::Instant::now(); + audit + .record(NewAuditEvent::now( + AuditEventType::RateLimitExceeded, + "198.51.100.40", + false, + )) + .await + .expect("first audit event should enqueue"); + tokio::task::yield_now().await; + assert_eq!(persisted_audit_event_count(&db_path), 0); + + audit + .record(NewAuditEvent::now( + AuditEventType::RateLimitExceeded, + "198.51.100.41", + false, + )) + .await + .expect("second audit event should enqueue"); + wait_for_persisted_audit_event_count(&db_path, 2).await; + assert_eq!(tokio::time::Instant::now(), paused_at); + + audit.shutdown().await; + let _ = std::fs::remove_file(&db_path); + let _ = std::fs::remove_dir_all(&temp_dir); +} + +#[tokio::test(start_paused = true)] +async fn audit_writer_flushes_at_configured_interval() { + let temp_dir = unique_temp_dir("nodelite-audit-configured-interval"); + std::fs::create_dir_all(&temp_dir).expect("temp dir should exist"); + let db_path = temp_dir.join("audit.sqlite3"); + let mut config = sample_config(db_path.clone()); + config.writer_batch_max = 128; + config.writer_flush_interval_ms = 10; + let audit = AuditLog::new(config, 5); + audit.initialize().await.expect("audit should initialize"); + tokio::task::yield_now().await; + + audit + .record(NewAuditEvent::now( + AuditEventType::RateLimitExceeded, + "198.51.100.42", + false, + )) + .await + .expect("audit event should enqueue"); + tokio::task::yield_now().await; + assert_eq!(persisted_audit_event_count(&db_path), 0); + + let before_advance = tokio::time::Instant::now(); + tokio::time::advance(std::time::Duration::from_millis(10)).await; + wait_for_persisted_audit_event_count(&db_path, 1).await; + assert_eq!( + tokio::time::Instant::now(), + before_advance + std::time::Duration::from_millis(10) + ); + + audit.shutdown().await; + let _ = std::fs::remove_file(&db_path); + let _ = std::fs::remove_dir_all(&temp_dir); +} diff --git a/nodelite-server/src/audit/writer.rs b/nodelite-server/src/audit/writer.rs index 98c7366..6069f99 100644 --- a/nodelite-server/src/audit/writer.rs +++ b/nodelite-server/src/audit/writer.rs @@ -12,12 +12,11 @@ use tracing::{debug, warn}; use super::types::NewAuditEvent; -const AUDIT_BATCH_MAX: usize = 128; -const AUDIT_BATCH_FLUSH_INTERVAL: Duration = Duration::from_millis(100); - pub(super) struct AuditWriterContext { pub(super) connection: Arc>>, pub(super) write_failures: Arc, + pub(super) batch_max: usize, + pub(super) flush_interval: Duration, } pub(super) enum AuditWriterCommand { @@ -29,8 +28,8 @@ pub(super) async fn run_audit_writer( mut rx: mpsc::Receiver, context: AuditWriterContext, ) { - let mut batch = Vec::with_capacity(AUDIT_BATCH_MAX); - let mut flush_timer = tokio::time::interval(AUDIT_BATCH_FLUSH_INTERVAL); + let mut batch = Vec::with_capacity(context.batch_max); + let mut flush_timer = tokio::time::interval(context.flush_interval); flush_timer.set_missed_tick_behavior(MissedTickBehavior::Delay); flush_timer.tick().await; @@ -40,7 +39,7 @@ pub(super) async fn run_audit_writer( received = rx.recv() => match received { Some(AuditWriterCommand::Event(event)) => { batch.push(event); - if batch.len() >= AUDIT_BATCH_MAX { + if batch.len() >= context.batch_max { flush_audit_batch(&mut batch, &context).await; } } @@ -64,7 +63,7 @@ pub(super) async fn run_audit_writer( match command { AuditWriterCommand::Event(event) => { batch.push(event); - if batch.len() >= AUDIT_BATCH_MAX { + if batch.len() >= context.batch_max { flush_audit_batch(&mut batch, &context).await; } } From cdaea137f3ddbcb3b8bb44df17b82c09dda8df30 Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 03:23:19 +0800 Subject: [PATCH 6/8] fix(server): bound writer batch configuration Co-authored-by: Claude --- config/server.example.toml | 4 +- nodelite-proto/src/config.rs | 2 + nodelite-proto/src/config/raw.rs | 22 +++---- .../src/config/tests/server_defaults.rs | 58 ++++++++++++++++++- nodelite-proto/src/lib.rs | 8 +-- nodelite-server/src/audit/writer.rs | 8 ++- nodelite-server/src/history/writer.rs | 11 ++-- 7 files changed, 86 insertions(+), 27 deletions(-) diff --git a/config/server.example.toml b/config/server.example.toml index 8791a09..c86f402 100644 --- a/config/server.example.toml +++ b/config/server.example.toml @@ -21,7 +21,7 @@ history_db_path = "./data/history.sqlite3" history_query_concurrency = 4 # 每个历史只读连接的 SQLite 私有 page cache(KiB),允许 64-1024,不等同于 Linux filesystem page cache。 history_read_cache_kib = 512 -# 历史 writer 单次 SQLite 事务最多写入的记录数,必须至少为 1。 +# 历史 writer 单次 SQLite 事务最多写入的记录数,允许 1-4096。 history_writer_batch_max = 128 # 历史 writer 最大攒批时间(毫秒),必须至少为 10,更小会导致高频唤醒。 history_writer_flush_interval_ms = 100 @@ -77,7 +77,7 @@ enabled = true db_path = "./data/audit.sqlite3" # 仅保留最近 N 天的安全事件。 retention_days = 90 -# 审计 writer 单次 SQLite 事务最多写入的事件数,必须至少为 1。 +# 审计 writer 单次 SQLite 事务最多写入的事件数,允许 1-4096。 writer_batch_max = 128 # 审计 writer 最大攒批时间(毫秒),必须至少为 10。 writer_flush_interval_ms = 100 diff --git a/nodelite-proto/src/config.rs b/nodelite-proto/src/config.rs index 16f324f..8f0fbbb 100644 --- a/nodelite-proto/src/config.rs +++ b/nodelite-proto/src/config.rs @@ -116,6 +116,8 @@ pub const DEFAULT_AUDIT_RETENTION_DAYS: u64 = 90; pub const DEFAULT_AUDIT_WRITER_BATCH_MAX: usize = 128; /// 审计写入器默认最大攒批时间(毫秒)。 pub const DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS: u64 = 100; +/// 历史与审计写入器单次事务允许的最大记录数。 +pub const MAX_WRITER_BATCH_SIZE: usize = 4096; /// 写入器 flush 间隔下限，避免极短定时器形成忙循环。 pub const MIN_WRITER_FLUSH_INTERVAL_MS: u64 = 10; /// GeoIP 数据库默认更新间隔(天)。 diff --git a/nodelite-proto/src/config/raw.rs b/nodelite-proto/src/config/raw.rs index d42a1e9..de6cded 100644 --- a/nodelite-proto/src/config/raw.rs +++ b/nodelite-proto/src/config/raw.rs @@ -33,9 +33,9 @@ use super::helpers::{ use super::{ AgentConfig, AlertingConfig, AuditConfig, ConfigError, GeoIpConfig, GeoIpEdition, GeoIpProvider, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, - MAX_NODE_IDENTITY_TEXT_BYTES, MAX_TOKEN_VERIFY_MAX_PARALLELISM, MIN_HISTORY_QUERY_CONCURRENCY, - MIN_HISTORY_READ_CACHE_KIB, MIN_TOKEN_VERIFY_MAX_PARALLELISM, MIN_WRITER_FLUSH_INTERVAL_MS, - MetricsConfig, ReadonlyAuthConfig, ServerConfig, WsConfig, + MAX_NODE_IDENTITY_TEXT_BYTES, MAX_TOKEN_VERIFY_MAX_PARALLELISM, MAX_WRITER_BATCH_SIZE, + MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, MIN_TOKEN_VERIFY_MAX_PARALLELISM, + MIN_WRITER_FLUSH_INTERVAL_MS, MetricsConfig, ReadonlyAuthConfig, ServerConfig, WsConfig, }; use crate::validation::{ ValidationError, normalize_string_list, validate_bounded_text, validate_identifier, @@ -448,10 +448,10 @@ impl RawServerConfigFile { "audit.retention_days must be greater than 0", )); } - if self.audit.writer_batch_max == 0 { - return Err(ConfigError::new( - "audit.writer_batch_max must be at least 1", - )); + if !(1..=MAX_WRITER_BATCH_SIZE).contains(&self.audit.writer_batch_max) { + return Err(ConfigError::new(format!( + "audit.writer_batch_max must be between 1 and {MAX_WRITER_BATCH_SIZE}" + ))); } if self.audit.writer_flush_interval_ms < MIN_WRITER_FLUSH_INTERVAL_MS { return Err(ConfigError::new(format!( @@ -622,10 +622,10 @@ impl RawServerConfigFile { "server.history_read_cache_kib must be between {MIN_HISTORY_READ_CACHE_KIB} and {MAX_HISTORY_READ_CACHE_KIB} KiB", ))); } - if self.server.history_writer_batch_max == 0 { - return Err(ConfigError::new( - "server.history_writer_batch_max must be at least 1", - )); + if !(1..=MAX_WRITER_BATCH_SIZE).contains(&self.server.history_writer_batch_max) { + return Err(ConfigError::new(format!( + "server.history_writer_batch_max must be between 1 and {MAX_WRITER_BATCH_SIZE}" + ))); } if self.server.history_writer_flush_interval_ms < MIN_WRITER_FLUSH_INTERVAL_MS { return Err(ConfigError::new(format!( diff --git a/nodelite-proto/src/config/tests/server_defaults.rs b/nodelite-proto/src/config/tests/server_defaults.rs index 191707e..b0d5bd6 100644 --- a/nodelite-proto/src/config/tests/server_defaults.rs +++ b/nodelite-proto/src/config/tests/server_defaults.rs @@ -15,8 +15,8 @@ use super::super::{ DEFAULT_WS_AUTH_FAIL_MAX_ATTEMPTS, DEFAULT_WS_AUTH_FAIL_WINDOW_SECS, DEFAULT_WS_MAX_CONNECTIONS_PER_IP, DEFAULT_WS_MAX_TOTAL_CONNECTIONS, GeoIpEdition, GeoIpProvider, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, - MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, MIN_WRITER_FLUSH_INTERVAL_MS, - parse_server_config, + MAX_WRITER_BATCH_SIZE, MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, + MIN_WRITER_FLUSH_INTERVAL_MS, parse_server_config, }; #[test] @@ -66,6 +66,7 @@ fn server_example_documents_writer_scheduling() { ] { assert!(example.lines().any(|line| line == expected)); } + assert_eq!(example.matches("允许 1-4096").count(), 2); } #[test] @@ -248,9 +249,29 @@ fn parses_writer_scheduling_overrides() { assert_eq!(config.audit.writer_flush_interval_ms, 50); } +#[test] +fn accepts_writer_batch_size_upper_bound() { + let config = parse_server_config(&format!( + r#" + [server] + listen = "127.0.0.1:8080" + public_base_url = "https://monitor.example.com" + history_writer_batch_max = {MAX_WRITER_BATCH_SIZE} + + [audit] + writer_batch_max = {MAX_WRITER_BATCH_SIZE} + "#, + )) + .expect("writer batch upper bound should parse"); + + assert_eq!(config.history_writer_batch_max, MAX_WRITER_BATCH_SIZE); + assert_eq!(config.audit.writer_batch_max, MAX_WRITER_BATCH_SIZE); +} + #[test] fn rejects_invalid_writer_scheduling_values() { assert_eq!(MIN_WRITER_FLUSH_INTERVAL_MS, 10); + assert_eq!(MAX_WRITER_BATCH_SIZE, 4096); for (section, key, value, expected) in [ ( "server", @@ -259,6 +280,37 @@ fn rejects_invalid_writer_scheduling_values() { "server.history_writer_batch_max", ), ("audit", "writer_batch_max", 0, "audit.writer_batch_max"), + ( + "server", + "history_writer_batch_max", + MAX_WRITER_BATCH_SIZE.saturating_add(1), + "server.history_writer_batch_max", + ), + ( + "audit", + "writer_batch_max", + MAX_WRITER_BATCH_SIZE.saturating_add(1), + "audit.writer_batch_max", + ), + ] { + let setting = if section == "server" { + format!("{key} = {value}") + } else { + format!("[audit]\n{key} = {value}") + }; + let input = format!( + r#" + [server] + listen = "127.0.0.1:8080" + public_base_url = "https://monitor.example.com" + {setting} + "#, + ); + let error = parse_server_config(&input).expect_err("invalid writer batch should fail"); + assert!(error.to_string().contains(expected)); + } + + for (section, key, value, expected) in [ ( "server", "history_writer_flush_interval_ms", @@ -297,7 +349,7 @@ fn rejects_invalid_writer_scheduling_values() { {setting} "#, ); - let error = parse_server_config(&input).expect_err("invalid writer setting should fail"); + let error = parse_server_config(&input).expect_err("invalid writer interval should fail"); assert!(error.to_string().contains(expected)); } } diff --git a/nodelite-proto/src/lib.rs b/nodelite-proto/src/lib.rs index 53f6f6e..1541723 100644 --- a/nodelite-proto/src/lib.rs +++ b/nodelite-proto/src/lib.rs @@ -29,10 +29,10 @@ pub use config::{ DEFAULT_STALE_AFTER_SECS, DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, GeoIpConfig, GeoIpEdition, GeoIpProvider, InspectionConfig, MAX_HISTORY_QUERY_CONCURRENCY, MAX_HISTORY_READ_CACHE_KIB, MAX_NODE_IDENTITY_TEXT_BYTES, MAX_NODE_TAG_BYTES, MAX_NODE_TAGS, - MAX_TOKEN_VERIFY_MAX_PARALLELISM, MIN_HISTORY_QUERY_CONCURRENCY, MIN_HISTORY_READ_CACHE_KIB, - MIN_TOKEN_VERIFY_MAX_PARALLELISM, MIN_WRITER_FLUSH_INTERVAL_MS, MetricsConfig, - ReadonlyAuthConfig, ServerConfig, WsConfig, normalize_totp_secret, parse_agent_config, - parse_server_config, upsert_toml_item_preserving_decor, + MAX_TOKEN_VERIFY_MAX_PARALLELISM, MAX_WRITER_BATCH_SIZE, MIN_HISTORY_QUERY_CONCURRENCY, + MIN_HISTORY_READ_CACHE_KIB, MIN_TOKEN_VERIFY_MAX_PARALLELISM, MIN_WRITER_FLUSH_INTERVAL_MS, + MetricsConfig, ReadonlyAuthConfig, ServerConfig, WsConfig, normalize_totp_secret, + parse_agent_config, parse_server_config, upsert_toml_item_preserving_decor, }; pub use message::{ AgentLogEntry, AgentLogsMessage, BrowserMessage, HelloMessage, diff --git a/nodelite-server/src/audit/writer.rs b/nodelite-server/src/audit/writer.rs index 6069f99..2e7eddf 100644 --- a/nodelite-server/src/audit/writer.rs +++ b/nodelite-server/src/audit/writer.rs @@ -5,6 +5,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use anyhow::{Context, Result}; +use nodelite_proto::MAX_WRITER_BATCH_SIZE; use rusqlite::{Connection, params}; use tokio::sync::{Mutex, mpsc, oneshot}; use tokio::time::MissedTickBehavior; @@ -28,7 +29,8 @@ pub(super) async fn run_audit_writer( mut rx: mpsc::Receiver, context: AuditWriterContext, ) { - let mut batch = Vec::with_capacity(context.batch_max); + let batch_max = context.batch_max.clamp(1, MAX_WRITER_BATCH_SIZE); + let mut batch = Vec::with_capacity(batch_max); let mut flush_timer = tokio::time::interval(context.flush_interval); flush_timer.set_missed_tick_behavior(MissedTickBehavior::Delay); flush_timer.tick().await; @@ -39,7 +41,7 @@ pub(super) async fn run_audit_writer( received = rx.recv() => match received { Some(AuditWriterCommand::Event(event)) => { batch.push(event); - if batch.len() >= context.batch_max { + if batch.len() >= batch_max { flush_audit_batch(&mut batch, &context).await; } } @@ -63,7 +65,7 @@ pub(super) async fn run_audit_writer( match command { AuditWriterCommand::Event(event) => { batch.push(event); - if batch.len() >= context.batch_max { + if batch.len() >= batch_max { flush_audit_batch(&mut batch, &context).await; } } diff --git a/nodelite-server/src/history/writer.rs b/nodelite-server/src/history/writer.rs index 9733cfe..8819028 100644 --- a/nodelite-server/src/history/writer.rs +++ b/nodelite-server/src/history/writer.rs @@ -7,7 +7,9 @@ use std::time::Duration; use anyhow::Result; use chrono::{DateTime, Utc}; -use nodelite_proto::{DEFAULT_HISTORY_RETENTION_HOURS, HistoryPoint, NodeStatus, percentage}; +use nodelite_proto::{ + DEFAULT_HISTORY_RETENTION_HOURS, HistoryPoint, MAX_WRITER_BATCH_SIZE, NodeStatus, percentage, +}; use rusqlite::{Connection, Error as SqliteError, ErrorCode, params}; use tokio::sync::{Mutex, mpsc}; use tokio::time::MissedTickBehavior; @@ -41,7 +43,8 @@ pub(super) async fn run_history_writer( mut rx: mpsc::Receiver, context: WriterContext, ) { - let mut batch: Vec = Vec::with_capacity(context.batch_max); + let batch_max = context.batch_max.clamp(1, MAX_WRITER_BATCH_SIZE); + let mut batch: Vec = Vec::with_capacity(batch_max); let mut flush_timer = tokio::time::interval(context.flush_interval); flush_timer.set_missed_tick_behavior(MissedTickBehavior::Delay); flush_timer.tick().await; @@ -52,7 +55,7 @@ pub(super) async fn run_history_writer( received = rx.recv() => match received { Some(point) => { batch.push(point); - if batch.len() >= context.batch_max { + if batch.len() >= batch_max { flush_history_batch(&mut batch, &context).await; } } @@ -68,7 +71,7 @@ pub(super) async fn run_history_writer( while let Ok(point) = rx.try_recv() { batch.push(point); - if batch.len() >= context.batch_max { + if batch.len() >= batch_max { flush_history_batch(&mut batch, &context).await; } } From 7d1c3881474e051ddbca91ecac30c9c0f62d72aa Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 03:26:02 +0800 Subject: [PATCH 7/8] refactor(server): split test configuration fixtures Co-authored-by: Claude --- nodelite-server/src/test_support.rs | 87 ++-------------------- nodelite-server/src/test_support/config.rs | 85 +++++++++++++++++++++ 2 files changed, 91 insertions(+), 81 deletions(-) create mode 100644 nodelite-server/src/test_support/config.rs diff --git a/nodelite-server/src/test_support.rs b/nodelite-server/src/test_support.rs index 4004c6e..721143d 100644 --- a/nodelite-server/src/test_support.rs +++ b/nodelite-server/src/test_support.rs @@ -26,11 +26,15 @@ use crate::set_protected_response_headers; use crate::state::{SessionRefreshReply, SharedState}; use crate::ws::{ws_browser_handler, ws_handler}; use nodelite_proto::{ - AuditConfig, BrowserMessage, DiskUsage, HelloMessage, HistoryPoint, LoadAverage, MemoryUsage, + BrowserMessage, DiskUsage, HelloMessage, HistoryPoint, LoadAverage, MemoryUsage, NetworkCounters, NodeIdentity, NodeSnapshot, NodeStatus, NoticeLevel, OverviewData, - ReadonlyAuthConfig, RefreshTokenResponseMessage, ServerConfig, WireMessage, WsConfig, + RefreshTokenResponseMessage, WireMessage, }; +mod config; + +pub(crate) use config::{test_server_config, test_ws_config}; + pub const TEST_BASIC_AUTH_HEADER: &str = "Basic dmlld2VyOnNlY3JldA=="; #[cfg(tarpaulin)] const TEST_TIMEOUT_SECS: u64 = 60; @@ -42,85 +46,6 @@ pub const LIVE_REFRESH_TIMEOUT: Duration = Duration::from_secs(20); pub(crate) type TestSocket = tokio_tungstenite::WebSocketStream>; -pub(crate) fn test_ws_config( - max_total_connections: usize, - max_connections_per_ip: usize, -) -> WsConfig { - WsConfig { - max_total_connections, - max_connections_per_ip, - auth_fail_window_secs: 300, - auth_fail_max_attempts: 12, - auth_block_secs: 900, - } -} - -pub(crate) fn test_server_config( - listen: SocketAddr, - public_base_url: String, - registry_path: PathBuf, - history_path: PathBuf, - snapshot_path: PathBuf, -) -> ServerConfig { - ServerConfig { - listen, - public_base_url, - insecure_allow_http: false, - trusted_proxies: Vec::new(), - readonly_auth: Some(ReadonlyAuthConfig { - username: "viewer".to_string(), - password: "secret".to_string(), - enable_2fa: false, - totp_secret: None, - }), - ws: test_ws_config(128, 128), - metrics: nodelite_proto::MetricsConfig::default(), - audit: AuditConfig { - enabled: true, - db_path: history_path.with_file_name("audit.sqlite3"), - retention_days: 90, - writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, - writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, - log_successful_auth: true, - log_failed_auth: true, - log_token_events: true, - log_rate_limit: true, - }, - geoip: nodelite_proto::GeoIpConfig { - enabled: false, - provider: nodelite_proto::GeoIpProvider::Dbip, - edition: nodelite_proto::GeoIpEdition::CountryLite, - database_path: PathBuf::from("./data/geoip/dbip.mmdb"), - auto_update: true, - update_interval_days: nodelite_proto::DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, - }, - alerting: nodelite_proto::AlertingConfig::default(), - node_registry_path: registry_path, - history_db_path: history_path, - history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, - history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, - history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, - history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, - snapshot_path, - stale_after_secs: 5, - ping_interval_secs: 60, - max_message_bytes: 64 * 1024, - refresh_interval_secs: 5, - ignored_filesystems: vec!["tmpfs".to_string(), "devtmpfs".to_string()], - agent_release_base_url: None, - agent_release_sha256_x86_64: None, - agent_release_sha256_aarch64: None, - hello_timeout_secs: 10, - max_outstanding_pings: 32, - insecure_transport_warn_interval_secs: 900, - max_sanitized_disks: 64, - max_sanitized_string_bytes: 256, - metric_anomaly_session_limit: 5, - sqlite_busy_timeout_secs: 5, - token_verify_max_parallelism: nodelite_proto::DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, - } -} - pub(crate) fn synthetic_identity( node_id: &str, node_label: &str, diff --git a/nodelite-server/src/test_support/config.rs b/nodelite-server/src/test_support/config.rs new file mode 100644 index 0000000..16045ac --- /dev/null +++ b/nodelite-server/src/test_support/config.rs @@ -0,0 +1,85 @@ +//! Test-only server and WebSocket configuration fixtures. + +use std::net::SocketAddr; +use std::path::PathBuf; + +use nodelite_proto::{AuditConfig, ReadonlyAuthConfig, ServerConfig, WsConfig}; + +pub(crate) fn test_ws_config( + max_total_connections: usize, + max_connections_per_ip: usize, +) -> WsConfig { + WsConfig { + max_total_connections, + max_connections_per_ip, + auth_fail_window_secs: 300, + auth_fail_max_attempts: 12, + auth_block_secs: 900, + } +} + +pub(crate) fn test_server_config( + listen: SocketAddr, + public_base_url: String, + registry_path: PathBuf, + history_path: PathBuf, + snapshot_path: PathBuf, +) -> ServerConfig { + ServerConfig { + listen, + public_base_url, + insecure_allow_http: false, + trusted_proxies: Vec::new(), + readonly_auth: Some(ReadonlyAuthConfig { + username: "viewer".to_string(), + password: "secret".to_string(), + enable_2fa: false, + totp_secret: None, + }), + ws: test_ws_config(128, 128), + metrics: nodelite_proto::MetricsConfig::default(), + audit: AuditConfig { + enabled: true, + db_path: history_path.with_file_name("audit.sqlite3"), + retention_days: 90, + writer_batch_max: nodelite_proto::DEFAULT_AUDIT_WRITER_BATCH_MAX, + writer_flush_interval_ms: nodelite_proto::DEFAULT_AUDIT_WRITER_FLUSH_INTERVAL_MS, + log_successful_auth: true, + log_failed_auth: true, + log_token_events: true, + log_rate_limit: true, + }, + geoip: nodelite_proto::GeoIpConfig { + enabled: false, + provider: nodelite_proto::GeoIpProvider::Dbip, + edition: nodelite_proto::GeoIpEdition::CountryLite, + database_path: PathBuf::from("./data/geoip/dbip.mmdb"), + auto_update: true, + update_interval_days: nodelite_proto::DEFAULT_GEOIP_UPDATE_INTERVAL_DAYS, + }, + alerting: nodelite_proto::AlertingConfig::default(), + node_registry_path: registry_path, + history_db_path: history_path, + history_query_concurrency: nodelite_proto::DEFAULT_HISTORY_QUERY_CONCURRENCY, + history_read_cache_kib: nodelite_proto::DEFAULT_HISTORY_READ_CACHE_KIB, + history_writer_batch_max: nodelite_proto::DEFAULT_HISTORY_WRITER_BATCH_MAX, + history_writer_flush_interval_ms: nodelite_proto::DEFAULT_HISTORY_WRITER_FLUSH_INTERVAL_MS, + snapshot_path, + stale_after_secs: 5, + ping_interval_secs: 60, + max_message_bytes: 64 * 1024, + refresh_interval_secs: 5, + ignored_filesystems: vec!["tmpfs".to_string(), "devtmpfs".to_string()], + agent_release_base_url: None, + agent_release_sha256_x86_64: None, + agent_release_sha256_aarch64: None, + hello_timeout_secs: 10, + max_outstanding_pings: 32, + insecure_transport_warn_interval_secs: 900, + max_sanitized_disks: 64, + max_sanitized_string_bytes: 256, + metric_anomaly_session_limit: 5, + sqlite_busy_timeout_secs: 5, + token_verify_max_parallelism: nodelite_proto::DEFAULT_TOKEN_VERIFY_MAX_PARALLELISM, + } +} From a7f134824587c3c02a79e8b9eb3706a9312eab0f Mon Sep 17 00:00:00 2001 From: XiNian-dada Date: Mon, 27 Jul 2026 03:40:58 +0800 Subject: [PATCH 8/8] docs(server): clarify readiness observation fields Co-authored-by: Claude --- docs/readiness.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/docs/readiness.md b/docs/readiness.md index 29ef341..115f1ea 100644 --- a/docs/readiness.md +++ b/docs/readiness.md @@ -7,12 +7,13 @@ NodeLite Server 提供两个无需认证的探针端点： ## `/readyz` 语义 -`/readyz` 将信号分为两类： +`/readyz` 将暴露内容分为三类： -| 类别 | JSON 位置 | 是否影响 HTTP 状态码和 `ready` | +| 类别 | JSON 位置或条件 | 当前判定语义 | | --- | --- | --- | -| 硬就绪检查 | `checks.history_available`、`checks.registry_reload_healthy` | 是；任一失败时返回 `503 Service Unavailable` 和 `ready: false` | -| 诊断信号 | `signals` 中的审计可用性、写入丢弃/失败、队列及 WebSocket 容量 | 否；异常会返回 `status: "degraded"` 并加入 `problems` | +| 硬就绪检查 | `checks.history_available`、`checks.registry_reload_healthy` | 任一失败时加入对应的 `problems`，返回 `status: "degraded"`、`ready: false` 和 `503 Service Unavailable` | +| 降级诊断条件 | `signals.audit_enabled && !signals.audit_available`；三个写入计数器中的任一个大于 `0` | 分别加入 `audit_unavailable`、`history_dropped_writes`、`audit_dropped_writes` 或 `audit_write_failures`，并返回 `status: "degraded"`；这些条件本身不参与 HTTP 状态码和 `ready` 的判定 | +| 原始观测字段 | `signals` 中 history/audit 队列深度与容量、Agent/Browser WebSocket 当前连接数、总容量及单 IP 限制 | 只报告当前值；服务端目前不为这些字段设置阈值，因此它们本身不会加入 `problems`，也不会改变 `status`、`ready` 或 HTTP 状态码 | 审计日志是安全诊断能力，但不是 Server 接收 Agent、查询节点状态所需的硬依赖。因此，仅审计写入器不可用时，响应组合是： @@ -29,5 +30,5 @@ NodeLite Server 提供两个无需认证的探针端点： ## 探针与告警配置 - Kubernetes、systemd watchdog 或负载均衡器的就绪判断应使用 `/readyz` 的 HTTP 状态码；需要解析 JSON 时，应读取 `ready`，不要把 `status == "ok"` 当作接流量条件。 -- 告警系统应另外监控 `status`、`problems` 和 `signals`。`ready: true` 且 `status: "degraded"` 表示服务仍可接流量，但存在需要运维处理的诊断异常。 +- 告警系统应另外监控 `status`、`problems` 和 `signals`。`ready: true` 且 `status: "degraded"` 表示服务仍可接流量，但存在需要运维处理的诊断异常。队列和 WebSocket 容量字段仅提供原始数据，需要由外部监控按部署规模设置阈值。 - 不要仅用 `/healthz` 判断是否应把实例加入流量池；它只验证进程仍能响应。