优化后端数据处理

README.md

@@ -125,6 +125,12 @@ database_url = "postgres://nodeflare:password@127.0.0.1:5432/nodeflare?sslmode=d
 - 恢复建议使用相同版本的 NodeFlare；`pg_dump` 与 SQLite `.backup` 可作为补充。
 - 迁移会覆盖目标库已有数据并自动更新 `database_url`，完成后重启生效。
 
+历史指标按节点设置的保存间隔聚合（默认 60 秒）：CPU、负载、内存和 GPU 保存采样均值，CPU 另存最小/最大值，网速和磁盘 I/O 保留窗口峰值，流量计数保留末值。断网补报按窗口分别写入，重复重传不会重复计数；前台实时状态仍使用最新采样。
+
+Agent 保留未确认的本地采样，只有后端事务提交后才清理相应缓存。建议先更新后端再更新 Agent；双方均更新后启用完整的分批补报与聚合，单边更新时保持旧协议兼容。历史表会自动迁移，旧快照按单个采样读取，也可恢复升级前的 ZIP 备份。
+
+历史保留天数保持原设置（默认 30 天），指标和延迟记录仍共用该期限。过期清理每分钟运行，每条语句最多处理 1000 条记录，轮流处理各表，每轮最多 50 批并在两秒预算耗尽后停止启动新批次；已开始的语句会执行完成。SQLite 每五分钟尝试小批量增量回收，数据库占用包含主文件、WAL 和 SHM。完整空间回收仍由后台手动触发。
+
 ## 从源码开发
 
 ```bash

agent/src/live_batch.rs

@@ -5,7 +5,7 @@ use super::Report;
 pub(crate) const MAX_LIVE_BATCH_BYTES: usize = 768 * 1024;
 
 pub(crate) fn batch_len(queue: &VecDeque<Report>) -> usize {
-    let mut encoded_bytes = 64_usize;
+    let mut encoded_bytes = 192_usize;
     let mut count = 0_usize;
     for report in queue {
         let Ok(report_bytes) = serde_json::to_vec(report) else {
@@ -26,31 +26,11 @@ pub(crate) fn batch_len(queue: &VecDeque<Report>) -> usize {
     count.min(super::LIVE_QUEUE_CAPACITY)
 }
 
-pub(crate) fn latest(queue: &VecDeque<Report>) -> Vec<Report> {
-    let mut encoded_bytes = 64_usize;
-    let mut reports = Vec::new();
-    for report in queue.iter().rev() {
-        let Ok(report_bytes) = serde_json::to_vec(report) else {
-            break;
-        };
-        let next_bytes = encoded_bytes
-            .saturating_add(report_bytes.len())
-            .saturating_add(usize::from(!reports.is_empty()));
-        if next_bytes > MAX_LIVE_BATCH_BYTES {
-            break;
-        }
-        encoded_bytes = next_bytes;
-        reports.push(report.clone());
-    }
-    reports.reverse();
-    reports
-}
-
 #[cfg(test)]
 mod tests {
     use std::collections::VecDeque;
 
-    use super::{MAX_LIVE_BATCH_BYTES, batch_len, latest};
+    use super::{MAX_LIVE_BATCH_BYTES, batch_len};
     use crate::Report;
 
     #[test]
@@ -62,14 +42,16 @@ mod tests {
         let queue = std::iter::repeat_n(report, 10).collect::<VecDeque<_>>();
         let count = batch_len(&queue);
         assert!((1..10).contains(&count));
-        let payload =
-            crate::live_update_payload(&queue.iter().take(count).cloned().collect::<Vec<_>>())
-                .unwrap();
+        let payload = crate::live_update_payload(
+            &queue.iter().take(count).cloned().collect::<Vec<_>>(),
+            Some(true),
+        )
+        .unwrap();
         assert!(payload.len() <= MAX_LIVE_BATCH_BYTES);
     }
 
     #[test]
-    fn persistence_batch_keeps_the_newest_samples() {
+    fn persistence_batch_keeps_the_oldest_unconfirmed_samples() {
         let queue = (0..10)
             .map(|timestamp| Report {
                 timestamp,
@@ -77,14 +59,47 @@ mod tests {
                 ..Report::default()
             })
             .collect::<VecDeque<_>>();
-        let reports = latest(&queue);
+        let reports = crate::live_persistence_batch(&queue, -1);
         assert!(!reports.is_empty());
         assert!(reports.len() < queue.len());
-        assert_eq!(reports.last().unwrap().timestamp, 9);
+        assert_eq!(reports.first().unwrap().timestamp, 0);
         assert!(
             reports
                 .windows(2)
                 .all(|pair| pair[0].timestamp < pair[1].timestamp)
         );
+        let mut persisted = -1;
+        let mut received = Vec::new();
+        loop {
+            let batch = crate::live_persistence_batch(&queue, persisted);
+            if batch.is_empty() {
+                break;
+            }
+            let message = crate::live_update_payload(&batch, Some(true)).unwrap();
+            assert!(message.len() <= MAX_LIVE_BATCH_BYTES);
+            persisted = batch.last().unwrap().timestamp;
+            received.extend(batch.into_iter().map(|report| report.timestamp));
+        }
+        assert_eq!(received, (0..10).collect::<Vec<_>>());
+        let legacy = crate::legacy_persistence_batch(&queue, -1);
+        assert_eq!(legacy.last().unwrap().timestamp, 9);
+        assert!(legacy.len() < queue.len());
+    }
+
+    #[test]
+    fn caps_batch_count_and_preserves_legacy_payload_shape() {
+        let queue = (0..1000)
+            .map(|timestamp| Report {
+                timestamp,
+                ..Report::default()
+            })
+            .collect::<VecDeque<_>>();
+        assert!(batch_len(&queue) <= crate::LIVE_QUEUE_CAPACITY);
+        let reports = crate::live_persistence_batch(&queue, 100);
+        assert_eq!(reports.first().unwrap().timestamp, 101);
+        let message = crate::live_update_payload(&reports, None).unwrap();
+        let value: serde_json::Value = serde_json::from_str(&message).unwrap();
+        assert!(value.get("persist").is_none());
+        assert_eq!(value.as_object().unwrap().len(), 3);
     }
 }

agent/src/main.rs

@@ -1746,7 +1746,7 @@ fn report_batch_id(reports: &[Report]) -> String {
 
 type LiveSocket = tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<TcpStream>>;
 
-fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option<i64>)> {
+fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option<i64>, bool)> {
     let mut request = endpoint.into_client_request()?;
     request
         .headers_mut()
@@ -1770,7 +1770,11 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option<i64>)
         .get("date")
         .and_then(|value| value.to_str().ok())
         .and_then(|value| clock_offset_from_http_date(value, started_ms, ended_ms));
-    Ok((socket, clock_offset_ms))
+    let persistence_batches = response
+        .headers()
+        .get("x-nodeflare-persistence-batches")
+        .is_some_and(|value| value == "1");
+    Ok((socket, clock_offset_ms, persistence_batches))
 }
 
 fn set_live_read_timeout(socket: &mut LiveSocket, timeout: Option<Duration>) -> io::Result<()> {
@@ -2535,12 +2539,16 @@ fn wait_for_live_ack(
     }
 }
 
-fn live_update_payload(reports: &[Report]) -> Result<String> {
-    Ok(serde_json::to_string(&serde_json::json!({
+fn live_update_payload(reports: &[Report], persist: Option<bool>) -> Result<String> {
+    let mut message = serde_json::json!({
         "type": "update",
         "batchId": report_batch_id(reports),
         "samples": reports,
-    }))?)
+    });
+    if let Some(persist) = persist {
+        message["persist"] = serde_json::json!(persist);
+    }
+    Ok(serde_json::to_string(&message)?)
 }
 
 fn observe_persisted_through(target: &AtomicI64, ack: &LiveAck) {
@@ -2573,7 +2581,21 @@ fn live_persistence_batch(queue: &VecDeque<Report>, timestamp: i64) -> Vec<Repor
         .filter(|report| report.timestamp > timestamp)
         .cloned()
         .collect::<VecDeque<_>>();
-    live_batch::latest(&candidates)
+    let count = live_batch::batch_len(&candidates);
+    candidates.into_iter().take(count).collect()
+}
+
+fn legacy_persistence_batch(queue: &VecDeque<Report>, timestamp: i64) -> Vec<Report> {
+    let candidates = queue
+        .iter()
+        .rev()
+        .filter(|report| report.timestamp > timestamp)
+        .cloned()
+        .collect::<VecDeque<_>>();
+    let count = live_batch::batch_len(&candidates);
+    let mut batch = candidates.into_iter().take(count).collect::<Vec<_>>();
+    batch.reverse();
+    batch
 }
 
 fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
@@ -2586,6 +2608,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
         stats,
     } = worker;
     let mut socket: Option<LiveSocket> = None;
+    let mut persistence_batches = false;
     let mut wss_interval = configured_interval
         .lock()
         .map_or(Duration::from_secs(1), |interval| *interval);
@@ -2602,7 +2625,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
     'sender: loop {
         if socket.is_none() {
             match connect_live(endpoint, token) {
-                Ok((mut connected, offset_ms)) => {
+                Ok((mut connected, offset_ms, supports_persistence_batches)) => {
                     observe_clock(&clock, offset_ms);
                     if set_live_read_timeout(&mut connected, Some(LIVE_HINT_READ_TIMEOUT)).is_err()
                     {
@@ -2610,6 +2633,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
                         continue;
                     }
                     socket = Some(connected);
+                    persistence_batches = supports_persistence_batches;
                     remote_executor.reset_delivery();
                     accepted_through = persisted_through.load(Ordering::Acquire);
                     next_send_at = Instant::now();
@@ -2636,18 +2660,24 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
         let Ok(mut queue) = queue_lock.lock() else {
             return;
         };
-        let batch = loop {
+        let (batch, persist, more_pending) = loop {
             let now = Instant::now();
             let persisted = persisted_through.load(Ordering::Acquire);
             if now >= next_probe_at {
-                let persistence_batch = live_persistence_batch(&queue, persisted);
+                let persistence_batch = if persistence_batches {
+                    live_persistence_batch(&queue, persisted)
+                } else {
+                    legacy_persistence_batch(&queue, persisted)
+                };
                 if !persistence_batch.is_empty() {
-                    break persistence_batch;
+                    let last = persistence_batch.last().unwrap().timestamp;
+                    let more_pending = queue.iter().any(|report| report.timestamp > last);
+                    break (persistence_batch, true, more_pending);
                 }
             }
             let unsent = live_batch_after(&queue, accepted_through);
             if !unsent.is_empty() && now >= next_send_at {
-                break unsent;
+                break (unsent, false, false);
             }
 
             let wake_at = if !unsent.is_empty() {
@@ -2732,7 +2762,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
             wss_interval = (*interval).clamp(Duration::from_secs(1), Duration::from_secs(60));
         }
 
-        let payload = match live_update_payload(&batch) {
+        let payload = match live_update_payload(&batch, persistence_batches.then_some(persist)) {
             Ok(payload) => payload,
             Err(error) => {
                 eprintln!("live payload encode failed: {error}");
@@ -2771,6 +2801,14 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
                         wss_interval = ack_wss_interval(&ack);
                         next_send_at = Instant::now() + wss_interval;
                         next_probe_at = Instant::now() + ack_persist_interval(&ack);
+                        if persist
+                            && more_pending
+                            && batch
+                                .last()
+                                .is_some_and(|report| ack.persisted_through_ts >= report.timestamp)
+                        {
+                            next_probe_at = Instant::now();
+                        }
                     }
                 }
                 Ok(LiveRead::Config(config)) => {
@@ -3769,21 +3807,25 @@ mod tests {
 
     #[test]
     fn encodes_realtime_samples_as_a_batch() {
-        let payload = live_update_payload(&[
-            Report {
-                timestamp: 10,
-                cpu: 20.0,
-                ..Report::default()
-            },
-            Report {
-                timestamp: 15,
-                cpu: 30.0,
-                ..Report::default()
-            },
-        ])
+        let payload = live_update_payload(
+            &[
+                Report {
+                    timestamp: 10,
+                    cpu: 20.0,
+                    ..Report::default()
+                },
+                Report {
+                    timestamp: 15,
+                    cpu: 30.0,
+                    ..Report::default()
+                },
+            ],
+            Some(false),
+        )
         .unwrap();
         let value: serde_json::Value = serde_json::from_str(&payload).unwrap();
         assert_eq!(value["type"], "update");
+        assert_eq!(value["persist"], false);
         assert_eq!(value["samples"].as_array().unwrap().len(), 2);
         assert_eq!(value["samples"][1]["cpu"], 30.0);
     }

backend/migrations/postgres/02_metric_history_aggregates.sql

@@ -0,0 +1,14 @@
+ALTER TABLE metric_history ADD COLUMN sample_count BIGINT NOT NULL DEFAULT 1 CHECK(sample_count > 0);
+ALTER TABLE metric_history ADD COLUMN first_timestamp BIGINT NOT NULL DEFAULT 0;
+ALTER TABLE metric_history ADD COLUMN last_timestamp BIGINT NOT NULL DEFAULT 0;
+ALTER TABLE metric_history ADD COLUMN cpu_min DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN cpu_max DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN mem_used_max BIGINT;
+ALTER TABLE metric_history ADD COLUMN memory_avg DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN memory_min DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN disk_avg DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN disk_min DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN net_in_avg DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN net_in_min DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN net_out_avg DOUBLE PRECISION;
+ALTER TABLE metric_history ADD COLUMN net_out_min DOUBLE PRECISION;

backend/migrations/sqlite/02_metric_history_aggregates.sql

@@ -0,0 +1,14 @@
+ALTER TABLE metric_history ADD COLUMN sample_count INTEGER NOT NULL DEFAULT 1 CHECK(sample_count > 0);
+ALTER TABLE metric_history ADD COLUMN first_timestamp INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE metric_history ADD COLUMN last_timestamp INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE metric_history ADD COLUMN cpu_min REAL;
+ALTER TABLE metric_history ADD COLUMN cpu_max REAL;
+ALTER TABLE metric_history ADD COLUMN mem_used_max INTEGER;
+ALTER TABLE metric_history ADD COLUMN memory_avg REAL;
+ALTER TABLE metric_history ADD COLUMN memory_min REAL;
+ALTER TABLE metric_history ADD COLUMN disk_avg REAL;
+ALTER TABLE metric_history ADD COLUMN disk_min REAL;
+ALTER TABLE metric_history ADD COLUMN net_in_avg REAL;
+ALTER TABLE metric_history ADD COLUMN net_in_min REAL;
+ALTER TABLE metric_history ADD COLUMN net_out_avg REAL;
+ALTER TABLE metric_history ADD COLUMN net_out_min REAL;

backend/src/backup.rs

@@ -754,9 +754,20 @@ async fn restore_table<R: Read + Seek + Send>(
         bail!("{table} 备份表头无效");
     }
     let header: BackupTableHeader = serde_json::from_str(line.trim_end())?;
-    if header.table != table || header.columns != columns {
+    let legacy_history = table == "metric_history"
+        && columns.len()
+            == header.columns.len() + crate::db::queries::HISTORY_AGGREGATE_COLUMNS.len()
+        && columns.starts_with(&header.columns)
+        && columns[header.columns.len()..]
+            .iter()
+            .map(|column| column.name.as_str())
+            .eq(crate::db::queries::HISTORY_AGGREGATE_COLUMNS);
+    if header.table != table || (header.columns != columns && !legacy_history) {
         bail!("{table} 备份结构与当前数据库不兼容");
     }
+    // Original snapshot archives omit the new aggregate columns; their SQL
+    // defaults preserve each old row as a single sample.
+    let columns = &header.columns;
 
     let mut restored = 0_usize;
     let mut batch = Vec::with_capacity(INSERT_BATCH_ROWS);
@@ -894,6 +905,58 @@ mod tests {
         assert_eq!(site_name, "Migrated");
     }
 
+    #[tokio::test]
+    async fn restores_legacy_snapshot_archives_and_round_trips_aggregates() {
+        let source = crate::db::connect("sqlite::memory:").await.unwrap();
+        let target = crate::db::connect("sqlite::memory:").await.unwrap();
+        let original = sqlx::migrate::Migrator::with_migrations(
+            crate::db::SQLITE_MIGRATOR
+                .iter()
+                .filter(|migration| migration.version == 1)
+                .cloned()
+                .collect(),
+        );
+        original.run(source.pool()).await.unwrap();
+        target.migrate().await.unwrap();
+        sqlx::query(
+            "INSERT INTO settings(key,value) VALUES \
+            ('admin_username','admin'),('admin_password_hash','hash'), \
+            ('password_client_salt','salt'),('password_scheme','argon2-client-pbkdf2-v1')",
+        )
+        .execute(source.pool())
+        .await
+        .unwrap();
+        sqlx::query("INSERT INTO servers(id,name,token_hash,created_at,updated_at) VALUES ('legacy','Legacy','token',1,1)")
+            .execute(source.pool()).await.unwrap();
+        sqlx::query("INSERT INTO metric_history(server_id,timestamp,cpu,mem_used) VALUES ('legacy',?,75.0,500)")
+            .bind(crate::db::now()).execute(source.pool()).await.unwrap();
+        assert_eq!(copy_database(&source, &target).await.unwrap(), 6);
+        let points = crate::db::queries::history(&target, "legacy", 1)
+            .await
+            .unwrap();
+        assert_eq!(points.len(), 1);
+        assert_eq!(points[0].cpu_min, 75.0);
+        assert_eq!(points[0].cpu_max, 75.0);
+        assert_eq!(points[0].mem_used_max, 500);
+        assert_eq!(points[0].sample_count, 1);
+        sqlx::query("UPDATE metric_history SET sample_count=4,cpu=50.0,cpu_min=10.0,cpu_max=90.0, \
+            memory_avg=50.0,memory_min=10.0,net_in_avg=20.0,net_in_min=1.0,first_timestamp=?,last_timestamp=?")
+            .bind(crate::db::now() - 3).bind(crate::db::now()).execute(target.pool()).await.unwrap();
+        source.migrate().await.unwrap();
+        assert_eq!(copy_database(&target, &source).await.unwrap(), 6);
+        let row = sqlx::query(
+            "SELECT sample_count,cpu_min,cpu_max,memory_avg,memory_min FROM metric_history",
+        )
+        .fetch_one(source.pool())
+        .await
+        .unwrap();
+        assert_eq!(row.try_get::<i64, _>("sample_count").unwrap(), 4);
+        assert_eq!(row.try_get::<f64, _>("cpu_min").unwrap(), 10.0);
+        assert_eq!(row.try_get::<f64, _>("cpu_max").unwrap(), 90.0);
+        assert_eq!(row.try_get::<f64, _>("memory_avg").unwrap(), 50.0);
+        assert_eq!(row.try_get::<f64, _>("memory_min").unwrap(), 10.0);
+    }
+
     #[tokio::test]
     async fn sqlite_backup_round_trip_restores_data_and_clears_sessions() {
         let db = crate::db::connect("sqlite::memory:").await.unwrap();

backend/src/db/mod.rs

@@ -10,6 +10,7 @@ use sqlx::sqlite::SqliteConnectOptions;
 use sqlx::{AnyPool, Row};
 use std::collections::HashMap;
 use std::fs::OpenOptions;
+use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::{Mutex, OnceLock};
 use std::time::Duration;
@@ -18,7 +19,7 @@ pub const SECRET_MASK: &str = "********";
 const PASSWORD_SCHEME: &str = "argon2-client-pbkdf2-v1";
 const SESSION_ACTIVITY_WRITE_INTERVAL_SECONDS: i64 = 60;
 
-static SQLITE_MIGRATOR: Migrator = sqlx::migrate!("./migrations/sqlite");
+pub(super) static SQLITE_MIGRATOR: Migrator = sqlx::migrate!("./migrations/sqlite");
 static POSTGRES_MIGRATOR: Migrator = sqlx::migrate!("./migrations/postgres");
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -40,6 +41,7 @@ impl DatabaseKind {
 pub struct Database {
     pool: AnyPool,
     kind: DatabaseKind,
+    sqlite_path: Option<PathBuf>,
 }
 
 #[derive(Debug, Clone)]
@@ -79,10 +81,11 @@ impl Database {
                 let free_pages = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count")
                     .fetch_one(&self.pool)
                     .await?;
-                (
-                    page_count.saturating_mul(page_size),
-                    Some(free_pages.saturating_mul(page_size)),
-                )
+                let size = match self.sqlite_path.as_deref() {
+                    Some(path) => sqlite_storage_size(path).await?,
+                    None => page_count.saturating_mul(page_size),
+                };
+                (size, Some(free_pages.saturating_mul(page_size)))
             }
             DatabaseKind::Postgres => (
                 sqlx::query_scalar::<_, i64>("SELECT pg_database_size(current_database())::BIGINT")
@@ -102,13 +105,19 @@ impl Database {
     pub async fn reclaim_space(&self) -> Result<()> {
         match self.kind {
             DatabaseKind::Sqlite => {
+                let mut connection = self.pool.acquire().await?;
                 sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
-                    .execute(&self.pool)
+                    .execute(&mut *connection)
+                    .await?;
+                sqlx::query("PRAGMA auto_vacuum = INCREMENTAL")
+                    .execute(&mut *connection)
+                    .await?;
+                sqlx::query("VACUUM").execute(&mut *connection).await?;
+                sqlx::query("PRAGMA optimize")
+                    .execute(&mut *connection)
                     .await?;
-                sqlx::query("VACUUM").execute(&self.pool).await?;
-                sqlx::query("PRAGMA optimize").execute(&self.pool).await?;
                 sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
-                    .execute(&self.pool)
+                    .execute(&mut *connection)
                     .await?;
             }
             DatabaseKind::Postgres => {
@@ -123,13 +132,38 @@ impl Database {
     pub async fn optimize(&self) -> Result<()> {
         if self.kind == DatabaseKind::Sqlite {
             sqlx::query("PRAGMA optimize").execute(&self.pool).await?;
-            sqlx::query("PRAGMA incremental_vacuum(256)")
-                .execute(&self.pool)
-                .await?;
         }
         Ok(())
     }
 
+    pub async fn reclaim_incremental(&self) -> Result<()> {
+        if self.kind != DatabaseKind::Sqlite {
+            return Ok(());
+        }
+        let mut connection = self.pool.acquire().await?;
+        let free_pages = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count")
+            .fetch_one(&mut *connection)
+            .await?;
+        if free_pages == 0 {
+            return Ok(());
+        }
+        let page_budget = (free_pages / 8).clamp(256, 8192).min(free_pages);
+        let started = std::time::Instant::now();
+        for _ in 0..(page_budget + 127) / 128 {
+            if started.elapsed() >= Duration::from_millis(200) {
+                break;
+            }
+            sqlx::query("PRAGMA incremental_vacuum(128)")
+                .execute(&mut *connection)
+                .await?;
+            tokio::task::yield_now().await;
+        }
+        sqlx::query("PRAGMA wal_checkpoint(PASSIVE)")
+            .execute(&mut *connection)
+            .await?;
+        Ok(())
+    }
+
     pub fn sql(&self, statement: &'static str) -> &'static str {
         if self.kind == DatabaseKind::Sqlite {
             return statement;
@@ -295,7 +329,35 @@ pub async fn connect(database_url: &str) -> Result<Database> {
         });
     }
     let pool = options.connect(database_url).await?;
-    Ok(Database { pool, kind })
+    let sqlite_path = if kind == DatabaseKind::Sqlite {
+        let databases = sqlx::query("PRAGMA database_list").fetch_all(&pool).await?;
+        databases.iter().find_map(|row| {
+            let name: String = row.try_get("name").ok()?;
+            let file: String = row.try_get("file").ok()?;
+            (name == "main" && !file.is_empty()).then(|| PathBuf::from(file))
+        })
+    } else {
+        None
+    };
+    Ok(Database {
+        pool,
+        kind,
+        sqlite_path,
+    })
+}
+
+async fn sqlite_storage_size(path: &std::path::Path) -> Result<i64> {
+    let mut size = 0_i64;
+    for suffix in ["", "-wal", "-shm"] {
+        let mut candidate = path.as_os_str().to_os_string();
+        candidate.push(suffix);
+        match tokio::fs::metadata(&candidate).await {
+            Ok(metadata) => size = size.saturating_add(metadata.len().min(i64::MAX as u64) as i64),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error.into()),
+        }
+    }
+    Ok(size)
 }
 
 pub fn database_kind(database_url: &str) -> Result<DatabaseKind> {
@@ -904,7 +966,88 @@ mod tests {
         assert_eq!(stats.kind, "sqlite");
         assert!(stats.size_bytes > 0);
         assert!(stats.reclaimable_bytes.is_some());
+        let expected = ["nodeflare.db", "nodeflare.db-wal", "nodeflare.db-shm"]
+            .into_iter()
+            .map(|name| {
+                std::fs::metadata(directory.path().join(name))
+                    .unwrap()
+                    .len() as i64
+            })
+            .sum::<i64>();
+        assert_eq!(stats.size_bytes, expected);
+        assert!(stats.size_bytes > std::fs::metadata(&path).unwrap().len() as i64);
         db.reclaim_space().await.unwrap();
+        db.pool().close().await;
+    }
+
+    #[tokio::test]
+    async fn incremental_reclaim_releases_pages_without_full_vacuum() {
+        let directory = tempfile::tempdir().unwrap();
+        let db = connect(&format!(
+            "sqlite://{}",
+            directory.path().join("reclaim.db").display()
+        ))
+        .await
+        .unwrap();
+        db.migrate().await.unwrap();
+        sqlx::query("CREATE TABLE reclaim_test(id INTEGER PRIMARY KEY, data BLOB)")
+            .execute(db.pool())
+            .await
+            .unwrap();
+        sqlx::query(
+            "WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM seq WHERE n<3000) \
+            INSERT INTO reclaim_test SELECT n,zeroblob(4096) FROM seq",
+        )
+        .execute(db.pool())
+        .await
+        .unwrap();
+        sqlx::query("DELETE FROM reclaim_test WHERE id>1")
+            .execute(db.pool())
+            .await
+            .unwrap();
+        let before = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count")
+            .fetch_one(db.pool())
+            .await
+            .unwrap();
+        assert!(before > 256);
+        db.reclaim_incremental().await.unwrap();
+        let after = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count")
+            .fetch_one(db.pool())
+            .await
+            .unwrap();
+        assert!(after < before, "before={before}, after={after}");
+        assert!(before - after <= 8192);
+        assert_eq!(
+            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM reclaim_test")
+                .fetch_one(db.pool())
+                .await
+                .unwrap(),
+            1
+        );
+        db.pool().close().await;
+    }
+
+    #[tokio::test]
+    async fn aggregate_migration_preserves_existing_history() {
+        let db = connect("sqlite::memory:").await.unwrap();
+        let original = Migrator::with_migrations(
+            SQLITE_MIGRATOR
+                .iter()
+                .filter(|migration| migration.version == 1)
+                .cloned()
+                .collect(),
+        );
+        original.run(db.pool()).await.unwrap();
+        sqlx::query("INSERT INTO servers(id,name,token_hash,created_at,updated_at) VALUES ('old','Old','hash',1,1)")
+            .execute(db.pool()).await.unwrap();
+        sqlx::query("INSERT INTO metric_history(server_id,timestamp,cpu,mem_used) VALUES ('old',?,73.0,100)")
+            .bind(now()).execute(db.pool()).await.unwrap();
+        db.migrate().await.unwrap();
+        let point = queries::history(&db, "old", 1).await.unwrap().remove(0);
+        assert_eq!(point.cpu, 73.0);
+        assert_eq!(point.cpu_min, 73.0);
+        assert_eq!(point.cpu_max, 73.0);
+        assert_eq!(point.sample_count, 1);
     }
 
     #[tokio::test]

backend/src/main.rs

@@ -408,6 +408,11 @@ fn spawn_maintenance(state: Arc<AppState>) {
                 {
                     tracing::error!(%error, "database optimization failed");
                 }
+                if maintenance_runs.is_multiple_of(5)
+                    && let Err(error) = state.db.reclaim_incremental().await
+                {
+                    tracing::error!(%error, "incremental database reclaim failed");
+                }
             }
             if let Err(error) = exchange::refresh(&state.db, &state.http, false).await {
                 tracing::warn!(%error, "scheduled exchange-rate refresh failed");

backend/src/models.rs

@@ -286,10 +286,14 @@ pub struct ServerView {
 pub struct HistoryPoint {
     pub timestamp: i64,
     pub cpu: f64,
+    pub cpu_min: f64,
+    pub cpu_max: f64,
+    pub sample_count: i64,
     pub load1: f64,
     pub load5: f64,
     pub load15: f64,
     pub mem_used: i64,
+    pub mem_used_max: i64,
     pub mem_total: i64,
     pub swap_used: i64,
     pub swap_total: i64,

backend/src/websocket/agent.rs

@@ -30,6 +30,8 @@ struct UpdateMessage {
     #[serde(rename = "batchId")]
     batch_id: String,
     samples: Vec<AgentReport>,
+    #[serde(default)]
+    persist: Option<bool>,
 }
 
 #[derive(Deserialize)]
@@ -86,9 +88,15 @@ pub async fn handle(
     };
     let remote_ip = client_ip(&headers, peer, &state.config.trusted_proxies);
     let token = token.to_string();
-    ws.max_message_size(MAX_AGENT_MESSAGE_BYTES)
+    let mut response = ws
+        .max_message_size(MAX_AGENT_MESSAGE_BYTES)
         .max_frame_size(MAX_AGENT_MESSAGE_BYTES)
-        .on_upgrade(move |socket| run(socket, state, identity, remote_ip, token))
+        .on_upgrade(move |socket| run(socket, state, identity, remote_ip, token));
+    response.headers_mut().insert(
+        "x-nodeflare-persistence-batches",
+        axum::http::HeaderValue::from_static("1"),
+    );
+    response
 }
 
 async fn run(
@@ -233,6 +241,7 @@ async fn handle_text(
                 &update.batch_id,
                 &update.samples,
                 remote_ip,
+                update.persist,
             )
             .await
             {

frontend/src/types.ts

@@ -221,10 +221,14 @@ export interface AdminServer extends ServerSummary {
 export interface HistoryPoint {
   timestamp: number;
   cpu: number;
+  cpu_min?: number;
+  cpu_max?: number;
+  sample_count?: number;
   load1: number;
   load5: number;
   load15: number;
   mem_used: number;
+  mem_used_max?: number;
   mem_total: number;
   swap_used: number;
   swap_total: number;