feat: 增加 RocksDB 检查点备份和恢复

Cargo.lock

@@ -656,6 +656,7 @@ dependencies = [
  "blake3",
  "clap",
  "dht-crawler",
+ "dunce",
  "fs2",
  "hex",
  "regex",
@@ -716,6 +717,12 @@ version = "2.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "117240f60069e65410b3ae1bb213295bd828f707b5bec6596a1afc8793ce0cbc"
 
+[[package]]
+name = "dunce"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
+
 [[package]]
 name = "either"
 version = "1.17.0"

TODOS.md

@@ -283,7 +283,7 @@
 - [ ] 记录 Tantivy IndexWriter 内存和 commit 延迟
 - [ ] 根据实测调整批量大小队列容量和并发
 - [x] 增加带排空阶段恢复滞回和探测失败保护的磁盘只读降级策略
-- [ ] 增加数据库备份检查点和恢复验证
+- [x] 增加在线 RocksDB 检查点保留上限只读校验和带旧库保留的离线恢复
 - [x] 增加可配置的终端日志滚动文件日志和保留文件上限
 - [x] 验证间歇运行和正常退出恢复
 - [x] 完成本机约七小时真实持续运行并确认采集索引和搜索服务可用
@@ -295,7 +295,7 @@
 - [ ] 内存使用在目标上限内稳定
 - [ ] 队列和缓存不会随运行时间无限增长
 - [x] 磁盘不足时能够拒绝新任务排空持久化队列并保留搜索能力
-- [ ] 备份可以在独立目录恢复并搜索
+- [x] 备份可以在独立目录恢复并从 RocksDB 重建索引搜索
 - [ ] 连续运行期间没有数据格式损坏和不可恢复任务
 
 ## 阶段七 部署和运维
@@ -336,4 +336,4 @@
 
 完成二十四小时持续运行并继续观察私有内存 Metadata 成功率候选队列深度和每条成功 Metadata 的网络成本
 
-随后增加 RocksDB 检查点恢复
+随后补充 RocksDB 和 Tantivy 内部资源指标并根据实测继续调优

dht-search.example.toml

@@ -13,6 +13,13 @@ check_interval_secs = 10
 minimum_free_bytes = 5368709120
 resume_free_bytes = 6442450944
 
+[backup]
+enabled = true
+directory = "data/backups"
+interval_secs = 21600
+retain_checkpoints = 3
+create_on_start = true
+
 [logging]
 directory = "data/logs"
 file_enabled = true

dht-search/Cargo.toml

@@ -18,6 +18,7 @@ axum = "0.8.9"
 blake3 = "1.8.5"
 clap = { version = "4.5", features = ["derive"] }
 dht-crawler = { path = "../dht-crawler", features = ["metrics"] }
+dunce = "1.0"
 hex = "0.4"
 fs2 = "0.4"
 rocksdb = { version = "0.24.0", default-features = false, features = ["bindgen-runtime", "lz4"], optional = true }

dht-search/README.md

@@ -118,6 +118,34 @@ Windows 下索引每五秒批量提交 临时文件占用会自动指数退避
 
 开发时需要直接观察终端日志可以设置 `console_enabled = true` 文件日志和终端日志不能同时关闭
 
+### RocksDB 检查点备份和恢复
+
+RocksDB 是唯一权威数据源 应用使用 RocksDB 原生 Checkpoint API 在线生成一致快照 备份期间采集可以继续运行 Tantivy 不进入备份因为它可以从 RocksDB 完整重建
+
+| 配置项 | 默认值 | 作用 |
+|---|---:|---|
+| `backup.enabled` | `true` | 是否启用自动检查点 |
+| `backup.directory` | `data/backups` | 检查点目录 相对主配置文件解析 |
+| `backup.interval_secs` | `21600` | 每 6 小时创建一次检查点 |
+| `backup.retain_checkpoints` | `3` | 保留最近 3 个自动检查点 |
+| `backup.create_on_start` | `true` | 每次启动后立即创建一次检查点 |
+
+备份目录与数据目录位于同一磁盘时 RocksDB 会尽量通过硬链接减少复制开销 放到其他磁盘时可能复制全部数据库文件 创建前会检查备份磁盘剩余空间 磁盘保护期间自动跳过而不会阻塞服务
+
+`/stats` 返回检查点成功失败跳过清理数量 最近成功时间耗时和记录数量 自动清理只处理名称严格匹配 `checkpoint-` 加二十位时间戳的直接子目录 不会删除手工目录文件或符号链接
+
+恢复必须在服务停止后执行 数据目录独占锁会阻止运行中的服务和恢复命令同时操作
+
+```powershell
+target\release\dht-search.exe `
+  --config dht-search.example.toml `
+  --restore-checkpoint data\backups\checkpoint-00000001775400000000
+```
+
+恢复命令会先以只读方式校验源检查点 再复制到数据目录内的暂存目录并二次校验 然后切换 `rocksdb` 目录 原数据库不会删除而是保留为 `rocksdb.pre-restore-*` 方便人工回滚 Tantivy 作为派生数据会被删除 下次正常启动自动从恢复后的 RocksDB 重建
+
+确认恢复数据无误后可以人工删除 `rocksdb.pre-restore-*` 释放空间 不要在服务运行时移动或删除这些目录
+
 ### Metadata 安全限制
 
 应用会在 Metadata 下载和进入 RocksDB 前执行两层资源与结构校验

dht-search/src/api/handlers.rs

@@ -36,12 +36,20 @@ pub(crate) async fn stats(State(state): State<ApiState>) -> Json<StatsResponse>
     let observability = state.dht_stats.observability_snapshot();
     let persistence = state.persistence.snapshot();
     let disk = state.disk_guard.snapshot();
+    let backup = state.backup_stats.snapshot();
     let filtered = persistence.filtered;
     let verification = state
         .verification
         .as_ref()
         .map(|ingress| ingress.stats().snapshot());
     Json(StatsResponse {
+        backup_created: backup.created,
+        backup_failed: backup.failed,
+        backup_skipped: backup.skipped,
+        backup_pruned: backup.pruned,
+        backup_last_success_at: backup.last_success_at,
+        backup_last_duration_millis: backup.last_duration_millis,
+        backup_latest_records: backup.latest_records,
         disk_state: disk.mode.as_str(),
         disk_available_bytes: disk.available_bytes,
         disk_minimum_free_bytes: disk.minimum_free_bytes,

dht-search/src/api/mod.rs

@@ -13,7 +13,8 @@ use tokio_util::sync::CancellationToken;
 use tower_http::services::{ServeDir, ServeFile};
 
 use crate::{
-    crawler::pipeline::PersistenceIngress, disk_guard::DiskGuard, verification::VerificationIngress,
+    backup::BackupStats, crawler::pipeline::PersistenceIngress, disk_guard::DiskGuard,
+    verification::VerificationIngress,
 };
 
 #[derive(Clone)]
@@ -24,6 +25,7 @@ pub(crate) struct ApiState {
     pub(crate) persistence: PersistenceIngress,
     pub(crate) verification: Option<VerificationIngress>,
     pub(crate) disk_guard: DiskGuard,
+    pub(crate) backup_stats: BackupStats,
 }
 
 pub(crate) async fn serve(
@@ -133,6 +135,7 @@ mod tests {
                 persistence: persistence.ingress.clone(),
                 verification: Some(verification),
                 disk_guard,
+                backup_stats: BackupStats::default(),
             },
             web_dir,
         );
@@ -164,6 +167,7 @@ mod tests {
         assert_eq!(json["metadata_filtered_too_many_files"], 0);
         assert_eq!(json["disk_state"], "normal");
         assert!(json["disk_available_bytes"].is_null());
+        assert_eq!(json["backup_created"], 0);
 
         let response = app
             .clone()

dht-search/src/api/response.rs

@@ -16,6 +16,13 @@ pub(crate) struct ErrorResponse {
 
 #[derive(Debug, Serialize)]
 pub(crate) struct StatsResponse {
+    pub(crate) backup_created: u64,
+    pub(crate) backup_failed: u64,
+    pub(crate) backup_skipped: u64,
+    pub(crate) backup_pruned: u64,
+    pub(crate) backup_last_success_at: Option<u64>,
+    pub(crate) backup_last_duration_millis: u64,
+    pub(crate) backup_latest_records: u64,
     pub(crate) disk_state: &'static str,
     pub(crate) disk_available_bytes: Option<u64>,
     pub(crate) disk_minimum_free_bytes: u64,

dht-search/src/app.rs

@@ -16,6 +16,7 @@ use tokio_util::sync::CancellationToken;
 
 use crate::{
     api::{self, ApiState},
+    backup::{self, BackupStats},
     config::AppConfig,
     crawler::pipeline::PersistencePipeline,
     disk_guard::{self, DiskGuard},
@@ -25,6 +26,7 @@ use crate::{
 
 pub(crate) async fn run(config: AppConfig) -> Result<(), AppError> {
     std::fs::create_dir_all(&config.data_dir)?;
+    let _data_lock = backup::acquire_data_lock(&config.data_dir)?;
     let disk_guard = DiskGuard::new(&config.disk_guard);
     let database_path = config.data_dir.join("rocksdb");
     let metadata_limits = config.metadata_limits();
@@ -57,6 +59,18 @@ pub(crate) async fn run(config: AppConfig) -> Result<(), AppError> {
         ingress.clone(),
         disk_cancel.clone(),
     ));
+    let backup_cancel = CancellationToken::new();
+    let (backup_stats, backup_task) = if config.backup.enabled {
+        let (stats, task) = backup::start(
+            repository.clone(),
+            config.backup.clone(),
+            disk_guard.clone(),
+            backup_cancel.clone(),
+        );
+        (stats, Some(task))
+    } else {
+        (BackupStats::default(), None)
+    };
 
     let options = config.dht_options();
     let server = DHTServer::new(options.clone()).await?;
@@ -211,6 +225,7 @@ pub(crate) async fn run(config: AppConfig) -> Result<(), AppError> {
             persistence: persistence.ingress.clone(),
             verification: verification_ingress,
             disk_guard: disk_guard.clone(),
+            backup_stats,
         },
         api_cancel.clone(),
     ));
@@ -255,6 +270,10 @@ pub(crate) async fn run(config: AppConfig) -> Result<(), AppError> {
     };
 
     verification_cancel.cancel();
+    backup_cancel.cancel();
+    if let Some(task) = backup_task {
+        let _ = task.await;
+    }
     drop(verification_fatal_guard);
     if let Some(task) = verification_task {
         match task.await {

dht-search/src/backup.rs

@@ -0,0 +1,498 @@
+// 负责在线创建 RocksDB 检查点限制备份数量并执行带旧库保留的离线恢复
+
+use std::{
+    fs::{self, File, OpenOptions},
+    path::{Path, PathBuf},
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering},
+    },
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+
+use dht_search::storage::{CheckpointSummary, RocksTorrentRepository, StorageError};
+use fs2::FileExt;
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+
+use crate::{config::BackupConfig, disk_guard::DiskGuard};
+
+const CHECKPOINT_PREFIX: &str = "checkpoint-";
+const CHECKPOINT_DIGITS: usize = 20;
+const DATA_LOCK_FILE: &str = ".dht-search.lock";
+
+pub(crate) struct DataDirectoryLock {
+    _file: File,
+}
+
+#[derive(Clone, Default)]
+pub(crate) struct BackupStats {
+    inner: Arc<BackupStatsInner>,
+}
+
+#[derive(Default)]
+struct BackupStatsInner {
+    created: AtomicU64,
+    failed: AtomicU64,
+    skipped: AtomicU64,
+    pruned: AtomicU64,
+    last_success_at: AtomicU64,
+    last_duration_millis: AtomicU64,
+    latest_records: AtomicU64,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct BackupSnapshot {
+    pub(crate) created: u64,
+    pub(crate) failed: u64,
+    pub(crate) skipped: u64,
+    pub(crate) pruned: u64,
+    pub(crate) last_success_at: Option<u64>,
+    pub(crate) last_duration_millis: u64,
+    pub(crate) latest_records: u64,
+}
+
+#[derive(Debug)]
+pub(crate) struct RestoreOutcome {
+    pub(crate) records: u64,
+    pub(crate) previous_database: Option<PathBuf>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum BackupError {
+    #[error("备份文件操作失败: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("RocksDB 检查点操作失败: {0}")]
+    Storage(#[from] StorageError),
+    #[error("数据目录正被另一个服务或恢复进程使用: {0}")]
+    DataDirectoryLocked(PathBuf),
+    #[error("检查点路径必须是独立目录且不能位于当前 RocksDB 内部")]
+    UnsafeCheckpointPath,
+    #[error("检查点包含不支持的符号链接或特殊文件: {0}")]
+    UnsafeCheckpointEntry(PathBuf),
+    #[error("无法生成唯一的检查点目录名")]
+    CheckpointNameExhausted,
+    #[error("恢复暂存目录已经存在: {0}")]
+    RestoreStagingExists(PathBuf),
+    #[error("数据库目录切换失败且旧数据库回滚失败: {0}")]
+    RestoreRollbackFailed(String),
+}
+
+pub(crate) fn acquire_data_lock(data_dir: &Path) -> Result<DataDirectoryLock, BackupError> {
+    fs::create_dir_all(data_dir)?;
+    let path = data_dir.join(DATA_LOCK_FILE);
+    let file = OpenOptions::new()
+        .create(true)
+        .truncate(false)
+        .read(true)
+        .write(true)
+        .open(&path)?;
+    file.try_lock_exclusive()
+        .map_err(|_| BackupError::DataDirectoryLocked(path))?;
+    Ok(DataDirectoryLock { _file: file })
+}
+
+pub(crate) fn start(
+    repository: Arc<RocksTorrentRepository>,
+    config: BackupConfig,
+    disk_guard: DiskGuard,
+    cancel: CancellationToken,
+) -> (BackupStats, JoinHandle<()>) {
+    let stats = BackupStats::default();
+    let task_stats = stats.clone();
+    let task = tokio::spawn(async move {
+        run(repository, config, disk_guard, task_stats, cancel).await;
+    });
+    (stats, task)
+}
+
+async fn run(
+    repository: Arc<RocksTorrentRepository>,
+    config: BackupConfig,
+    disk_guard: DiskGuard,
+    stats: BackupStats,
+    cancel: CancellationToken,
+) {
+    if config.create_on_start {
+        create_one(
+            repository.clone(),
+            config.clone(),
+            disk_guard.clone(),
+            stats.clone(),
+        )
+        .await;
+    }
+    let start = tokio::time::Instant::now() + Duration::from_secs(config.interval_secs);
+    let mut ticker = tokio::time::interval_at(start, Duration::from_secs(config.interval_secs));
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        tokio::select! {
+            _ = cancel.cancelled() => break,
+            _ = ticker.tick() => {
+                create_one(
+                    repository.clone(),
+                    config.clone(),
+                    disk_guard.clone(),
+                    stats.clone(),
+                ).await;
+            }
+        }
+    }
+}
+
+async fn create_one(
+    repository: Arc<RocksTorrentRepository>,
+    config: BackupConfig,
+    disk_guard: DiskGuard,
+    stats: BackupStats,
+) {
+    let Some(permit) = disk_guard.begin_new_write() else {
+        stats.inner.skipped.fetch_add(1, Ordering::Relaxed);
+        tracing::warn!("磁盘处于保护状态并跳过 RocksDB 检查点");
+        return;
+    };
+    if let Err(error) = fs::create_dir_all(&config.directory) {
+        stats.inner.failed.fetch_add(1, Ordering::Relaxed);
+        tracing::error!(%error, "无法创建检查点目录");
+        return;
+    }
+    let minimum_free_bytes = disk_guard.snapshot().minimum_free_bytes;
+    match fs2::available_space(&config.directory) {
+        Ok(available) if available < minimum_free_bytes => {
+            stats.inner.skipped.fetch_add(1, Ordering::Relaxed);
+            tracing::warn!(
+                available,
+                minimum_free_bytes,
+                "备份磁盘空间不足并跳过检查点"
+            );
+            return;
+        }
+        Err(error) => {
+            stats.inner.skipped.fetch_add(1, Ordering::Relaxed);
+            tracing::warn!(%error, "无法确认备份磁盘剩余空间并跳过检查点");
+            return;
+        }
+        Ok(_) => {}
+    }
+    let started = Instant::now();
+    let result = tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        create_checkpoint(&repository, &config, unix_timestamp_millis())
+    })
+    .await;
+    match result {
+        Ok(Ok((summary, pruned))) => {
+            let now = unix_timestamp();
+            stats.inner.created.fetch_add(1, Ordering::Relaxed);
+            stats.inner.pruned.fetch_add(pruned, Ordering::Relaxed);
+            stats.inner.last_success_at.store(now, Ordering::Relaxed);
+            stats.inner.last_duration_millis.store(
+                started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64,
+                Ordering::Relaxed,
+            );
+            stats
+                .inner
+                .latest_records
+                .store(summary.records, Ordering::Relaxed);
+            tracing::info!(records = summary.records, pruned, "RocksDB 检查点创建完成");
+        }
+        Ok(Err(error)) => {
+            stats.inner.failed.fetch_add(1, Ordering::Relaxed);
+            tracing::error!(%error, "RocksDB 检查点创建失败");
+        }
+        Err(error) => {
+            stats.inner.failed.fetch_add(1, Ordering::Relaxed);
+            tracing::error!(%error, "RocksDB 检查点任务异常");
+        }
+    }
+}
+
+fn create_checkpoint(
+    repository: &RocksTorrentRepository,
+    config: &BackupConfig,
+    timestamp_millis: u128,
+) -> Result<(CheckpointSummary, u64), BackupError> {
+    fs::create_dir_all(&config.directory)?;
+    let checkpoint_path = unique_checkpoint_path(&config.directory, timestamp_millis)?;
+    repository.create_checkpoint(&checkpoint_path)?;
+    let summary = RocksTorrentRepository::validate_checkpoint(&checkpoint_path)?;
+    let pruned = prune_checkpoints(&config.directory, config.retain_checkpoints)?;
+    Ok((summary, pruned))
+}
+
+fn unique_checkpoint_path(
+    directory: &Path,
+    timestamp_millis: u128,
+) -> Result<PathBuf, BackupError> {
+    for offset in 0..1_000_u128 {
+        let value = timestamp_millis.saturating_add(offset);
+        let path = directory.join(format!("{CHECKPOINT_PREFIX}{value:0CHECKPOINT_DIGITS$}"));
+        if !path.exists() {
+            return Ok(path);
+        }
+    }
+    Err(BackupError::CheckpointNameExhausted)
+}
+
+fn prune_checkpoints(directory: &Path, retain: usize) -> Result<u64, BackupError> {
+    let mut checkpoints = checkpoint_directories(directory)?;
+    checkpoints.sort_unstable_by_key(|(timestamp, _)| *timestamp);
+    let remove_count = checkpoints.len().saturating_sub(retain);
+    for (_, path) in checkpoints.into_iter().take(remove_count) {
+        fs::remove_dir_all(path)?;
+    }
+    Ok(remove_count as u64)
+}
+
+fn checkpoint_directories(directory: &Path) -> Result<Vec<(u128, PathBuf)>, BackupError> {
+    let mut checkpoints = Vec::new();
+    for entry in fs::read_dir(directory)? {
+        let entry = entry?;
+        let file_type = entry.file_type()?;
+        if !file_type.is_dir() || file_type.is_symlink() {
+            continue;
+        }
+        let name = entry.file_name();
+        let Some(name) = name.to_str() else { continue };
+        let Some(timestamp) = parse_checkpoint_name(name) else {
+            continue;
+        };
+        checkpoints.push((timestamp, entry.path()));
+    }
+    Ok(checkpoints)
+}
+
+fn parse_checkpoint_name(name: &str) -> Option<u128> {
+    let digits = name.strip_prefix(CHECKPOINT_PREFIX)?;
+    if digits.len() != CHECKPOINT_DIGITS || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
+        return None;
+    }
+    digits.parse().ok()
+}
+
+pub(crate) fn restore(data_dir: &Path, checkpoint: &Path) -> Result<RestoreOutcome, BackupError> {
+    let _lock = acquire_data_lock(data_dir)?;
+    let source = dunce::canonicalize(checkpoint)?;
+    let database_path = data_dir.join("rocksdb");
+    let database_absolute = absolute_path(&database_path)?;
+    if !source.is_dir() || source == database_absolute || source.starts_with(&database_absolute) {
+        return Err(BackupError::UnsafeCheckpointPath);
+    }
+    let source_summary = RocksTorrentRepository::validate_checkpoint(&source)?;
+    let timestamp = unix_timestamp_millis();
+    let staging = data_dir.join(format!("rocksdb.restore-staging-{timestamp:020}"));
+    if staging.exists() {
+        return Err(BackupError::RestoreStagingExists(staging));
+    }
+    copy_directory(&source, &staging)?;
+    let staged_summary = RocksTorrentRepository::validate_checkpoint(&staging)?;
+    if staged_summary != source_summary {
+        return Err(BackupError::Storage(StorageError::CorruptContentGroup));
+    }
+
+    let tantivy_path = data_dir.join("tantivy");
+    if tantivy_path.exists() {
+        fs::remove_dir_all(&tantivy_path)?;
+    }
+    let previous_database = if database_path.exists() {
+        let previous = data_dir.join(format!("rocksdb.pre-restore-{timestamp:020}"));
+        fs::rename(&database_path, &previous)?;
+        Some(previous)
+    } else {
+        None
+    };
+    if let Err(error) = fs::rename(&staging, &database_path) {
+        if let Some(previous) = &previous_database
+            && let Err(rollback) = fs::rename(previous, &database_path)
+        {
+            return Err(BackupError::RestoreRollbackFailed(format!(
+                "切换错误 {error}; 回滚错误 {rollback}"
+            )));
+        }
+        return Err(BackupError::Io(error));
+    }
+    Ok(RestoreOutcome {
+        records: staged_summary.records,
+        previous_database,
+    })
+}
+
+fn copy_directory(source: &Path, destination: &Path) -> Result<(), BackupError> {
+    fs::create_dir(destination)?;
+    for entry in fs::read_dir(source)? {
+        let entry = entry?;
+        let source_path = entry.path();
+        let destination_path = destination.join(entry.file_name());
+        let file_type = entry.file_type()?;
+        if file_type.is_symlink() {
+            return Err(BackupError::UnsafeCheckpointEntry(source_path));
+        }
+        if file_type.is_dir() {
+            copy_directory(&source_path, &destination_path)?;
+        } else if file_type.is_file() {
+            fs::copy(&source_path, &destination_path)?;
+        } else {
+            return Err(BackupError::UnsafeCheckpointEntry(source_path));
+        }
+    }
+    Ok(())
+}
+
+fn absolute_path(path: &Path) -> Result<PathBuf, std::io::Error> {
+    if path.is_absolute() {
+        Ok(path.to_path_buf())
+    } else {
+        Ok(std::env::current_dir()?.join(path))
+    }
+}
+
+impl BackupStats {
+    pub(crate) fn snapshot(&self) -> BackupSnapshot {
+        let last_success_at = self.inner.last_success_at.load(Ordering::Relaxed);
+        BackupSnapshot {
+            created: self.inner.created.load(Ordering::Relaxed),
+            failed: self.inner.failed.load(Ordering::Relaxed),
+            skipped: self.inner.skipped.load(Ordering::Relaxed),
+            pruned: self.inner.pruned.load(Ordering::Relaxed),
+            last_success_at: (last_success_at != 0).then_some(last_success_at),
+            last_duration_millis: self.inner.last_duration_millis.load(Ordering::Relaxed),
+            latest_records: self.inner.latest_records.load(Ordering::Relaxed),
+        }
+    }
+}
+
+fn unix_timestamp() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
+fn unix_timestamp_millis() -> u128 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis()
+}
+
+#[cfg(test)]
+mod tests {
+    use dht_crawler::{FileInfo, TorrentInfo};
+    use dht_search::{domain::TorrentRecord, search::SearchEngine, storage::TorrentRepository};
+    use tempfile::TempDir;
+
+    use super::*;
+
+    fn record(byte: u8, name: &str) -> TorrentRecord {
+        TorrentRecord::try_from(TorrentInfo {
+            info_hash: format!("{byte:02x}").repeat(20),
+            magnet_link: String::new(),
+            name: name.into(),
+            total_size: 42,
+            files: vec![FileInfo {
+                path: format!("{name}.bin"),
+                size: 42,
+            }],
+            piece_length: 16_384,
+            peers: Vec::new(),
+            timestamp: 10,
+        })
+        .unwrap()
+    }
+
+    #[test]
+    fn data_directory_lock_rejects_a_second_owner() {
+        let directory = TempDir::new().unwrap();
+        let first = acquire_data_lock(directory.path()).unwrap();
+        assert!(matches!(
+            acquire_data_lock(directory.path()),
+            Err(BackupError::DataDirectoryLocked(_))
+        ));
+        drop(first);
+        acquire_data_lock(directory.path()).unwrap();
+    }
+
+    #[test]
+    fn retention_prunes_only_recognized_checkpoint_directories() {
+        let directory = TempDir::new().unwrap();
+        for timestamp in 1..=4_u128 {
+            fs::create_dir(directory.path().join(format!(
+                "{CHECKPOINT_PREFIX}{timestamp:0CHECKPOINT_DIGITS$}"
+            )))
+            .unwrap();
+        }
+        fs::create_dir(directory.path().join("checkpoint-manual")).unwrap();
+        fs::write(
+            directory
+                .path()
+                .join(format!("{CHECKPOINT_PREFIX}{:0CHECKPOINT_DIGITS$}", 5)),
+            "not a directory",
+        )
+        .unwrap();
+
+        assert_eq!(prune_checkpoints(directory.path(), 2).unwrap(), 2);
+        assert_eq!(checkpoint_directories(directory.path()).unwrap().len(), 2);
+        assert!(directory.path().join("checkpoint-manual").is_dir());
+    }
+
+    #[test]
+    fn repeated_checkpoints_keep_only_the_configured_latest_snapshots() {
+        let directory = TempDir::new().unwrap();
+        let repository = RocksTorrentRepository::open(directory.path().join("rocksdb")).unwrap();
+        let config = BackupConfig {
+            enabled: true,
+            directory: directory.path().join("backups"),
+            interval_secs: 1,
+            retain_checkpoints: 2,
+            create_on_start: true,
+        };
+        for index in 1..=3_u8 {
+            repository
+                .upsert(record(index, &format!("record-{index}")))
+                .unwrap();
+            let (summary, _) = create_checkpoint(&repository, &config, u128::from(index)).unwrap();
+            assert_eq!(summary.records, u64::from(index));
+        }
+
+        let checkpoints = checkpoint_directories(&config.directory).unwrap();
+        assert_eq!(checkpoints.len(), 2);
+        assert!(checkpoints.iter().all(|(timestamp, _)| *timestamp >= 2));
+    }
+
+    #[test]
+    fn restore_keeps_previous_database_and_forces_search_rebuild() {
+        let directory = TempDir::new().unwrap();
+        let data_dir = directory.path().join("data");
+        fs::create_dir_all(&data_dir).unwrap();
+        let database_path = data_dir.join("rocksdb");
+        let checkpoint_path = directory.path().join("checkpoint");
+        let first = record(1, "checkpoint-first");
+        let second = record(2, "newer-second");
+        {
+            let repository = RocksTorrentRepository::open(&database_path).unwrap();
+            repository.upsert(first.clone()).unwrap();
+            repository.create_checkpoint(&checkpoint_path).unwrap();
+            repository.upsert(second.clone()).unwrap();
+        }
+        fs::create_dir(data_dir.join("tantivy")).unwrap();
+        fs::write(data_dir.join("tantivy/old-index"), "derived").unwrap();
+
+        let outcome = restore(&data_dir, &checkpoint_path).unwrap();
+        assert_eq!(outcome.records, 1);
+        assert!(!data_dir.join("tantivy").exists());
+        let restored = RocksTorrentRepository::open(&database_path).unwrap();
+        assert!(restored.get(first.info_hash).unwrap().is_some());
+        assert!(restored.get(second.info_hash).unwrap().is_none());
+        let previous_path = outcome.previous_database.unwrap();
+        let previous = RocksTorrentRepository::open(previous_path).unwrap();
+        assert!(previous.get(first.info_hash).unwrap().is_some());
+        assert!(previous.get(second.info_hash).unwrap().is_some());
+        drop(previous);
+
+        let search = SearchEngine::open(data_dir.join("tantivy")).unwrap();
+        restored.prepare_full_reindex().unwrap();
+        while search.index_pending(&restored, 100, 20).unwrap() > 0 {}
+        assert_eq!(search.search("checkpoint-first", 0, 10).unwrap().total, 1);
+        assert_eq!(search.search("newer-second", 0, 10).unwrap().total, 0);
+    }
+}

dht-search/src/config.rs

@@ -21,6 +21,14 @@ pub(crate) struct Cli {
     data_dir: Option<PathBuf>,
     #[arg(long)]
     run_duration_secs: Option<u64>,
+    #[arg(long)]
+    restore_checkpoint: Option<PathBuf>,
+}
+
+#[derive(Debug)]
+pub(crate) struct StartupConfig {
+    pub(crate) app: AppConfig,
+    pub(crate) restore_checkpoint: Option<PathBuf>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -36,6 +44,7 @@ pub(crate) struct AppConfig {
     pub(crate) metadata_limits: MetadataLimitsConfig,
     pub(crate) dht: DhtConfig,
     pub(crate) disk_guard: DiskGuardConfig,
+    pub(crate) backup: BackupConfig,
     pub(crate) logging: LoggingConfig,
     pub(crate) http: HttpConfig,
     pub(crate) verification: VerificationConfig,
@@ -92,6 +101,16 @@ pub(crate) struct LoggingConfig {
     pub(crate) file_prefix: String,
 }
 
+#[derive(Debug, Clone, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub(crate) struct BackupConfig {
+    pub(crate) enabled: bool,
+    pub(crate) directory: PathBuf,
+    pub(crate) interval_secs: u64,
+    pub(crate) retain_checkpoints: usize,
+    pub(crate) create_on_start: bool,
+}
+
 #[derive(Debug, Clone, Copy, Default, Deserialize)]
 #[serde(rename_all = "kebab-case")]
 pub(crate) enum LogRotation {
@@ -133,7 +152,7 @@ pub(crate) enum NetworkMode {
 }
 
 impl Cli {
-    pub(crate) fn load(self) -> Result<AppConfig, AppError> {
+    pub(crate) fn load(self) -> Result<StartupConfig, AppError> {
         let config_path = absolute_path(&self.config)?;
         let mut config = if config_path.exists() {
             let contents = fs::read_to_string(&config_path)?;
@@ -163,12 +182,23 @@ impl Cli {
             config.logging.directory = base.join(&config.logging.directory);
         }
         config.logging.directory = normalize_absolute(config.logging.directory)?;
+        if config.backup.directory.is_relative() {
+            config.backup.directory = base.join(&config.backup.directory);
+        }
+        config.backup.directory = normalize_absolute(config.backup.directory)?;
         if config.http.web_dir.is_relative() {
             config.http.web_dir = base.join(&config.http.web_dir);
         }
         config.http.web_dir = normalize_absolute(config.http.web_dir)?;
         config.validate()?;
-        Ok(config)
+        let restore_checkpoint = self
+            .restore_checkpoint
+            .map(normalize_absolute)
+            .transpose()?;
+        Ok(StartupConfig {
+            app: config,
+            restore_checkpoint,
+        })
     }
 }
 
@@ -296,6 +326,19 @@ impl AppConfig {
                 "磁盘检查间隔必须大于零且恢复阈值必须大于保护阈值".to_owned(),
             ));
         }
+        if self.backup.interval_secs == 0 || self.backup.retain_checkpoints == 0 {
+            return Err(AppError::Config(
+                "备份间隔和检查点保留数量必须大于零".to_owned(),
+            ));
+        }
+        let database_path = self.data_dir.join("rocksdb");
+        if self.backup.directory == database_path
+            || self.backup.directory.starts_with(&database_path)
+        {
+            return Err(AppError::Config(
+                "检查点目录不能位于 RocksDB 数据库目录内部".to_owned(),
+            ));
+        }
         if !self.logging.file_enabled && !self.logging.console_enabled {
             return Err(AppError::Config(
                 "文件日志和终端日志不能同时关闭".to_owned(),
@@ -347,6 +390,7 @@ impl Default for AppConfig {
             metadata_limits: MetadataLimitsConfig::default(),
             dht: DhtConfig::default(),
             disk_guard: DiskGuardConfig::default(),
+            backup: BackupConfig::default(),
             logging: LoggingConfig::default(),
             http: HttpConfig::default(),
             verification: VerificationConfig::default(),
@@ -425,6 +469,18 @@ impl Default for LoggingConfig {
     }
 }
 
+impl Default for BackupConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            directory: PathBuf::from("data/backups"),
+            interval_secs: 6 * 60 * 60,
+            retain_checkpoints: 3,
+            create_on_start: true,
+        }
+    }
+}
+
 impl Default for VerificationConfig {
     fn default() -> Self {
         Self {
@@ -498,19 +554,25 @@ mod tests {
         let directory = TempDir::new().unwrap();
         let config_path = directory.path().join("service.toml");
         fs::write(&config_path, "data_dir = 'state'").unwrap();
-        let config = Cli {
+        let startup = Cli {
             config: config_path,
             data_dir: None,
             run_duration_secs: None,
+            restore_checkpoint: None,
         }
         .load()
         .unwrap();
+        let config = startup.app;
         assert_eq!(config.data_dir, directory.path().join("state"));
         assert_eq!(
             config.content_filter_file,
             directory.path().join("content-filters.toml")
         );
         assert_eq!(config.logging.directory, directory.path().join("data/logs"));
+        assert_eq!(
+            config.backup.directory,
+            directory.path().join("data/backups")
+        );
         assert_eq!(config.http.web_dir, directory.path().join("web/dist"));
     }
 
@@ -523,6 +585,7 @@ mod tests {
             config: config_path,
             data_dir: None,
             run_duration_secs: None,
+            restore_checkpoint: None,
         }
         .load()
         .unwrap_err();
@@ -564,4 +627,11 @@ mod tests {
         config.logging.file_prefix = "../service".into();
         assert!(matches!(config.validate(), Err(AppError::Config(_))));
     }
+
+    #[test]
+    fn backup_directory_cannot_be_inside_rocksdb() {
+        let mut config = AppConfig::default();
+        config.backup.directory = config.data_dir.join("rocksdb/checkpoints");
+        assert!(matches!(config.validate(), Err(AppError::Config(_))));
+    }
 }

dht-search/src/error.rs

@@ -8,6 +8,8 @@ pub(crate) enum AppError {
     Toml(#[from] toml::de::Error),
     #[error("内容过滤配置无效: {0}")]
     ContentFilter(#[from] dht_search::domain::ContentFilterError),
+    #[error("备份或恢复失败: {0}")]
+    Backup(#[from] crate::backup::BackupError),
     #[error("配置无效: {0}")]
     Config(String),
     #[error("DHT 服务失败: {0}")]

dht-search/src/main.rs

@@ -4,6 +4,7 @@ use clap::Parser;
 
 mod api;
 mod app;
+mod backup;
 mod config;
 mod crawler;
 mod disk_guard;
@@ -16,21 +17,43 @@ mod verification;
 
 #[tokio::main]
 async fn main() {
-    let config = match config::Cli::parse().load() {
+    let startup = match config::Cli::parse().load() {
         Ok(config) => config,
         Err(error) => {
             eprintln!("无法加载配置: {error}");
             std::process::exit(1);
         }
     };
-    let _telemetry = match telemetry::init(&config.logging) {
+    let restore_requested = startup.restore_checkpoint.is_some();
+    let mut logging = startup.app.logging.clone();
+    if restore_requested {
+        logging.file_enabled = false;
+        logging.console_enabled = true;
+    }
+    let _telemetry = match telemetry::init(&logging) {
         Ok(guard) => guard,
         Err(error) => {
             eprintln!("无法初始化日志: {error}");
             std::process::exit(1);
         }
     };
-    let result = app::run(config).await;
+    let result = if let Some(checkpoint) = startup.restore_checkpoint {
+        backup::restore(&startup.app.data_dir, &checkpoint)
+            .map(|outcome| {
+                tracing::info!(
+                    records = outcome.records,
+                    previous_database = ?outcome.previous_database,
+                    "RocksDB 检查点恢复完成 下次正常启动将重建搜索索引"
+                );
+                println!("恢复完成: {} 条记录", outcome.records);
+                if let Some(previous) = outcome.previous_database {
+                    println!("原数据库保留在: {}", previous.display());
+                }
+            })
+            .map_err(error::AppError::from)
+    } else {
+        app::run(startup.app).await
+    };
     if let Err(error) = result {
         tracing::error!(%error, "dht-search 退出");
         std::process::exit(1);

dht-search/src/storage/mod.rs

@@ -6,8 +6,8 @@ mod repository;
 mod rocks;
 
 pub use repository::{
-    ContentGroupTask, ContentVariants, StorageError, TorrentRepository, UpsertOutcome,
-    VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
+    CheckpointSummary, ContentGroupTask, ContentVariants, StorageError, TorrentRepository,
+    UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
 };
 #[cfg(feature = "rocksdb-storage")]
 pub use rocks::RocksTorrentRepository;

dht-search/src/storage/repository.rs

@@ -92,6 +92,11 @@ pub struct ContentVariants {
     pub records: Vec<TorrentRecord>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct CheckpointSummary {
+    pub records: u64,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum VerificationPriority {
     Normal,

dht-search/src/storage/rocks.rs

@@ -8,7 +8,7 @@ use std::{
 
 use rocksdb::{
     BlockBasedOptions, Cache, DB, DBCompressionType, Direction, IteratorMode, Options,
-    SliceTransform, WriteBatch,
+    SliceTransform, WriteBatch, checkpoint::Checkpoint,
 };
 
 use crate::domain::{
@@ -27,8 +27,8 @@ use super::{
         verification_locator_key, verification_task_key, verification_task_prefix,
     },
     repository::{
-        ContentGroupTask, ContentVariants, StorageError, TorrentRepository, UpsertOutcome,
-        VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
+        CheckpointSummary, ContentGroupTask, ContentVariants, StorageError, TorrentRepository,
+        UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
     },
 };
 
@@ -98,6 +98,32 @@ impl RocksTorrentRepository {
         self.content_filter_changed
     }
 
+    pub fn create_checkpoint(&self, path: impl AsRef<Path>) -> Result<(), StorageError> {
+        Checkpoint::new(&self.db)?.create_checkpoint(path)?;
+        Ok(())
+    }
+
+    pub fn validate_checkpoint(path: impl AsRef<Path>) -> Result<CheckpointSummary, StorageError> {
+        let options = Options::default();
+        let database = DB::open_for_read_only(&options, path, false)?;
+        match database.get(DATABASE_FORMAT_KEY)? {
+            Some(value) if value.as_slice() == DATABASE_FORMAT_VALUE => {}
+            _ => return Err(StorageError::IncompatibleDatabaseFormat),
+        }
+        let prefix = torrent_prefix();
+        let iterator = database.iterator(IteratorMode::From(&prefix, Direction::Forward));
+        let mut records = 0_u64;
+        for entry in iterator {
+            let (key, value) = entry?;
+            if !key.starts_with(&prefix) {
+                break;
+            }
+            Self::decode(&value)?;
+            records = records.saturating_add(1);
+        }
+        Ok(CheckpointSummary { records })
+    }
+
     fn initialize_format(&self) -> Result<(), StorageError> {
         match self.db.get(DATABASE_FORMAT_KEY)? {
             None => self
@@ -896,6 +922,26 @@ mod tests {
         assert_eq!(repository.get(expected.info_hash).unwrap(), Some(expected));
     }
 
+    #[test]
+    fn checkpoint_is_a_consistent_snapshot_and_can_be_opened_read_only() {
+        let directory = TempDir::new().unwrap();
+        let database_path = directory.path().join("rocksdb");
+        let checkpoint_path = directory.path().join("checkpoint");
+        let repository = RocksTorrentRepository::open(&database_path).unwrap();
+        let first = test_record(1, 10);
+        let second = test_record(2, 20);
+        repository.upsert(first.clone()).unwrap();
+        repository.create_checkpoint(&checkpoint_path).unwrap();
+        repository.upsert(second.clone()).unwrap();
+
+        let summary = RocksTorrentRepository::validate_checkpoint(&checkpoint_path).unwrap();
+        assert_eq!(summary.records, 1);
+        drop(repository);
+        let restored = RocksTorrentRepository::open(&checkpoint_path).unwrap();
+        assert!(restored.get(first.info_hash).unwrap().is_some());
+        assert!(restored.get(second.info_hash).unwrap().is_none());
+    }
+
     #[test]
     fn filter_keeps_raw_files_but_returns_effective_view() {
         let directory = TempDir::new().unwrap();