From ea4625e5dad22b2e643197eba64b1587e87d00f3 Mon Sep 17 00:00:00 2001 From: chuan Date: Mon, 10 Aug 2026 13:35:22 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0=20RocksDB=20?= =?UTF-8?q?=E6=A3=80=E6=9F=A5=E7=82=B9=E5=A4=87=E4=BB=BD=E5=92=8C=E6=81=A2?= =?UTF-8?q?=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 7 + TODOS.md | 6 +- dht-search.example.toml | 7 + dht-search/Cargo.toml | 1 + dht-search/README.md | 28 ++ dht-search/src/api/handlers.rs | 8 + dht-search/src/api/mod.rs | 6 +- dht-search/src/api/response.rs | 7 + dht-search/src/app.rs | 19 + dht-search/src/backup.rs | 498 +++++++++++++++++++++++++++ dht-search/src/config.rs | 76 +++- dht-search/src/error.rs | 2 + dht-search/src/main.rs | 29 +- dht-search/src/storage/mod.rs | 4 +- dht-search/src/storage/repository.rs | 5 + dht-search/src/storage/rocks.rs | 52 ++- 16 files changed, 740 insertions(+), 15 deletions(-) create mode 100644 dht-search/src/backup.rs diff --git a/Cargo.lock b/Cargo.lock index 2b92e91..37b4b66 100644 --- a/Cargo.lock +++ b/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" diff --git a/TODOS.md b/TODOS.md index fc53048..6cc6c30 100644 --- a/TODOS.md +++ b/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 内部资源指标并根据实测继续调优 diff --git a/dht-search.example.toml b/dht-search.example.toml index 24e5210..f5d8c9b 100644 --- a/dht-search.example.toml +++ b/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 diff --git a/dht-search/Cargo.toml b/dht-search/Cargo.toml index 5cebb48..b90b943 100644 --- a/dht-search/Cargo.toml +++ b/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 } diff --git a/dht-search/README.md b/dht-search/README.md index d2e1e1b..7d9964b 100644 --- a/dht-search/README.md +++ b/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 前执行两层资源与结构校验 diff --git a/dht-search/src/api/handlers.rs b/dht-search/src/api/handlers.rs index 75d3a26..466eb4d 100644 --- a/dht-search/src/api/handlers.rs +++ b/dht-search/src/api/handlers.rs @@ -36,12 +36,20 @@ pub(crate) async fn stats(State(state): State) -> Json 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, diff --git a/dht-search/src/api/mod.rs b/dht-search/src/api/mod.rs index 52a6790..d45864a 100644 --- a/dht-search/src/api/mod.rs +++ b/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, 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() diff --git a/dht-search/src/api/response.rs b/dht-search/src/api/response.rs index cea5b49..812866c 100644 --- a/dht-search/src/api/response.rs +++ b/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, + 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, pub(crate) disk_minimum_free_bytes: u64, diff --git a/dht-search/src/app.rs b/dht-search/src/app.rs index e8367e6..04bf00f 100644 --- a/dht-search/src/app.rs +++ b/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 { diff --git a/dht-search/src/backup.rs b/dht-search/src/backup.rs new file mode 100644 index 0000000..0bca8e5 --- /dev/null +++ b/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, +} + +#[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, + 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, +} + +#[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 { + 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, + 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, + 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, + 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 { + 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 { + 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, 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 { + 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 { + 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 { + 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); + } +} diff --git a/dht-search/src/config.rs b/dht-search/src/config.rs index f791bc6..709a91f 100644 --- a/dht-search/src/config.rs +++ b/dht-search/src/config.rs @@ -21,6 +21,14 @@ pub(crate) struct Cli { data_dir: Option, #[arg(long)] run_duration_secs: Option, + #[arg(long)] + restore_checkpoint: Option, +} + +#[derive(Debug)] +pub(crate) struct StartupConfig { + pub(crate) app: AppConfig, + pub(crate) restore_checkpoint: Option, } #[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 { + pub(crate) fn load(self) -> Result { 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(_)))); + } } diff --git a/dht-search/src/error.rs b/dht-search/src/error.rs index cc9b924..24649af 100644 --- a/dht-search/src/error.rs +++ b/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}")] diff --git a/dht-search/src/main.rs b/dht-search/src/main.rs index 9c984fc..100a533 100644 --- a/dht-search/src/main.rs +++ b/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); diff --git a/dht-search/src/storage/mod.rs b/dht-search/src/storage/mod.rs index 06c3698..070a13f 100644 --- a/dht-search/src/storage/mod.rs +++ b/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; diff --git a/dht-search/src/storage/repository.rs b/dht-search/src/storage/repository.rs index 8c27a84..79e21bf 100644 --- a/dht-search/src/storage/repository.rs +++ b/dht-search/src/storage/repository.rs @@ -92,6 +92,11 @@ pub struct ContentVariants { pub records: Vec, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CheckpointSummary { + pub records: u64, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum VerificationPriority { Normal, diff --git a/dht-search/src/storage/rocks.rs b/dht-search/src/storage/rocks.rs index de4041d..a37ffe0 100644 --- a/dht-search/src/storage/rocks.rs +++ b/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) -> Result<(), StorageError> { + Checkpoint::new(&self.db)?.create_checkpoint(path)?; + Ok(()) + } + + pub fn validate_checkpoint(path: impl AsRef) -> Result { + 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();