From 13d3a3b5d696066e66cde65e719501b1d2aa54ae Mon Sep 17 00:00:00 2001 From: chuan Date: Mon, 10 Aug 2026 22:55:28 +0800 Subject: [PATCH] feat(search): implement shadow index rebuilding --- README.md | 4 + TODOS.md | 9 + src/search/README.md | 11 +- src/search/src/api/handlers.rs | 2 + src/search/src/api/mod.rs | 11 +- src/search/src/api/response.rs | 6 +- src/search/src/app.rs | 14 +- src/search/src/backup.rs | 4 + src/search/src/bin/dht-benchmark/config.rs | 16 + src/search/src/bin/dht-benchmark/dataset.rs | 10 +- src/search/src/bin/dht-benchmark/report.rs | 1 + src/search/src/bin/dht-benchmark/runner.rs | 10 +- src/search/src/crawler/runtime.rs | 8 +- src/search/src/diagnostics/mod.rs | 10 +- src/search/src/index_worker.rs | 135 +++- src/search/src/search/document.rs | 58 +- src/search/src/search/filter.rs | 57 +- src/search/src/search/indexer.rs | 143 +++- src/search/src/search/mod.rs | 5 + src/search/src/search/runtime.rs | 772 ++++++++++++++++++++ src/search/src/search/schema.rs | 39 + src/search/src/storage/keys.rs | 3 + src/search/src/storage/mod.rs | 6 +- src/search/src/storage/repository.rs | 11 + src/search/src/storage/rocks.rs | 231 +++++- src/search/src/storage/rocks/lifecycle.rs | 2 + src/search/src/storage/rocks/tests.rs | 29 + src/web/src/pages/DiagnosticsPage.vue | 35 +- src/web/src/pages/SearchPage.vue | 37 +- src/web/src/types/api.ts | 19 + 30 files changed, 1589 insertions(+), 109 deletions(-) create mode 100644 src/search/src/search/runtime.rs diff --git a/README.md b/README.md index d7c0e1f..aa21a26 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,8 @@ opencodes/ 不参与构建且不得修改的参考项目 RocksDB 是唯一权威数据源 Tantivy 索引可以从 RocksDB 完整重建 +Tantivy 使用代际影子索引完成全量重建 过滤规则或索引文档结构变化时旧索引继续提供搜索 新索引完整校验后通过原子指针切换 重建状态和进度可以在搜索提示与系统诊断页查看 + 应用全部配置和内容隐藏规则统一位于 [`config.toml`](config.toml) Web 右上角的无线电图标可以即时停止或恢复 DHT 持续采集 状态会写回 `dht.enabled` 关闭后不会建立 DHT 和 Metadata 网络任务 但现有 RocksDB 数据仍会继续补建索引并提供本地搜索 @@ -143,6 +145,8 @@ bun run build 百万级基准中普通搜索过滤排序精确哈希索引吞吐磁盘和峰值内存均达到当前目标 大命中集合正则仍是继续扩大规模前最值得优化的查询路径 +大型种子会先按文件大小降序和规范化路径稳定排序 最多索引 2048 个文件且完整路径文本总量不超过 256 KiB 以优先覆盖主体内容并限制极端 Metadata 的索引放大 + ## 相关文档 - 当前实施状态和后续计划见 [`TODOS.md`](TODOS.md) diff --git a/TODOS.md b/TODOS.md index 9bc153e..89453ed 100644 --- a/TODOS.md +++ b/TODOS.md @@ -148,12 +148,18 @@ - [x] 实现相同 `content_key` 结果精确折叠和变体分页 - [x] 实现从 RocksDB 全量重建 Tantivy 索引 - [x] 支持索引结构不兼容时直接重建 +- [x] 使用影子索引保留旧搜索并在完整校验后原子切换 +- [x] 持久化影子索引构建清单并支持跨重启继续重建 +- [x] 暴露种子内容组待索引数量和全量重建状态进度 +- [x] 按文件大小为大型种子选择最多 2048 个文件并限制路径文本预算 +- [x] 优化完整名称别名文件名路径的相关性权重并使用热度时间稳定同分结果 ### 验收标准 - [x] 新写入记录在目标延迟内可搜索 - [x] 搜索索引删除后可以从 RocksDB 完整重建 - [x] 索引过程中异常退出不会永久丢失文档 +- [x] 全量重建期间旧索引继续提供完整旧结果且新数据在原子切换后可见 - [x] 百万级测试数据常用查询延迟达到 `README.md` 记录的目标 ## 阶段四 HTTP 搜索服务 @@ -362,6 +368,9 @@ - [x] 将主配置和内容过滤规则合并为唯一 `config.toml` - [x] 将 Docker 测试和性能基线核心内容合并到根目录 `README.md` - [x] 将 Xray 透明代理异常环境的判断处理和回滚方案独立到 `docs` +- [x] 实现可恢复的影子索引原子切换和 Web 重建进度 +- [x] 将大型种子文件路径覆盖扩大到按大小选择的 2048 个文件和 256 KiB +- [x] 使用文本优先热度时间同分的稳定相关性排序 完成二十四小时持续运行并继续观察私有内存 Metadata 成功率候选队列深度和每条成功 Metadata 的网络成本 diff --git a/src/search/README.md b/src/search/README.md index d685558..7517357 100644 --- a/src/search/README.md +++ b/src/search/README.md @@ -284,7 +284,15 @@ GET /torrents/{infohash}?file_offset=0&file_limit=100 RocksDB 是权威数据源而 Tantivy 是可重建索引 -当 Tantivy 目录不存在或结构不匹配时应用会直接创建新索引并从 RocksDB 的内容组状态完成全量重建 +当过滤规则 Tantivy Schema 或索引文档格式变化时 应用在独立代际目录构建影子索引 旧索引继续提供搜索且新收录内容在切换后统一可见 影子索引清空 RocksDB 待索引状态并通过文档数校验后原子更新活动指针 + +影子索引通过内部构建清单跨重启恢复 构建失败磁盘保护或进程退出不会删除活动索引 没有旧索引时会提供正在初始化的部分结果并明确标记结果尚不完整 切换成功后立即清理旧索引 Windows 文件占用造成的清理失败只记录警告而不影响服务 + +`/stats` 的 `index` 字段区分已保存 infohash 可搜索内容组活动索引文档影子索引文档和待索引数量 并返回重建原因状态进度开始完成时间和错误 + +每个内容组按文件大小降序和规范化路径升序选择最多 2048 个文件 完整路径文本总预算为 256 KiB 被选择文件的名称路径和扩展名参与搜索 文件总数大小和详情仍使用全部可见文件 + +默认相关性优先完整种子名称 名称片段 别名 文件名和完整路径 热度与最后发现时间只用于文本同分结果 用户显式选择的时间热度大小和发现次数排序不变 项目当前处于开发阶段 持久化结构变化时直接清理测试数据重新采集 不维护旧测试数据库兼容层 @@ -320,6 +328,7 @@ cargo run --release -p dht-search --bin dht-benchmark -- ` | `--records` | `10000` | 生成记录数量 上限一千万 | | `--generation-batch-size` | `1000` | 单批生成并暂存在内存的记录数量 | | `--index-batch-size` | `1000` | 每次 Tantivy 提交的内容文档数量 | +| `--files-per-record` | 不指定 | 为大型文件集合基准固定每条记录的文件数量 范围 1 到 20000 | | `--index-max-retries` | `20` | Windows 临时 IO 错误的最大连续重试次数 | | `--query-iterations` | `50` | 每类查询正式采样次数 | | `--query-warmup` | `5` | 每类查询预热次数 | diff --git a/src/search/src/api/handlers.rs b/src/search/src/api/handlers.rs index b462a85..7cac743 100644 --- a/src/search/src/api/handlers.rs +++ b/src/search/src/api/handlers.rs @@ -52,6 +52,7 @@ pub(crate) async fn stats(State(state): State) -> Json .crawler .verification() .map(|ingress| ingress.stats().snapshot()); + let index = state.search.status(state.repository.index_inventory()); Json(StatsResponse { http_active_requests: http.active_requests, http_requests: http.requests, @@ -143,6 +144,7 @@ pub(crate) async fn stats(State(state): State) -> Json persistence_rejected_full: persistence.rejected_full, persistence_queue: persistence.queue_depth, indexed_documents: state.search.num_docs(), + index, verification_queue: verification.map_or(0, |stats| stats.queue_depth), verification_accepted: verification.map_or(0, |stats| stats.accepted), verification_deduplicated: verification.map_or(0, |stats| stats.deduplicated), diff --git a/src/search/src/api/mod.rs b/src/search/src/api/mod.rs index 346c3fd..8984f68 100644 --- a/src/search/src/api/mod.rs +++ b/src/search/src/api/mod.rs @@ -6,7 +6,7 @@ mod response; use std::{net::SocketAddr, path::PathBuf, sync::Arc}; -use crate::{search::SearchEngine, storage::TorrentRepository}; +use crate::{search::SearchRuntime, storage::TorrentRepository}; use axum::{ Router, extract::{Request, State}, @@ -28,7 +28,7 @@ use crate::{ #[derive(Clone)] pub(crate) struct ApiState { pub(crate) repository: Arc, - pub(crate) search: SearchEngine, + pub(crate) search: SearchRuntime, pub(crate) crawler: CrawlerRuntime, pub(crate) persistence: PersistenceIngress, pub(crate) disk_guard: DiskGuard, @@ -90,7 +90,7 @@ mod tests { use crate::{ domain::{InfoHash, MetadataCandidate, MetadataLimits, TorrentFile, TorrentRecord}, - search::SearchEngine, + search::{SearchEngine, SearchRuntime}, storage::{RocksTorrentRepository, TorrentRepository}, }; use axum::{ @@ -166,7 +166,7 @@ mod tests { let app = router( ApiState { repository: repository_trait, - search, + search: SearchRuntime::from_engine(search), crawler, persistence: persistence.ingress.clone(), disk_guard, @@ -223,6 +223,9 @@ mod tests { assert_eq!(json["disk_state"], "normal"); assert!(json["disk_available_bytes"].is_null()); assert_eq!(json["backup_created"], 0); + assert_eq!(json["index"]["state"], "ready"); + assert_eq!(json["index"]["active_documents"], 1); + assert_eq!(json["index"]["pending_documents"], 0); assert_eq!(json["http_active_requests"], 1); assert!( json["http_requests"] diff --git a/src/search/src/api/response.rs b/src/search/src/api/response.rs index 0e2a20a..f202db2 100644 --- a/src/search/src/api/response.rs +++ b/src/search/src/api/response.rs @@ -1,6 +1,9 @@ // 负责定义稳定的 HTTP 响应模型和领域对象转换边界 -use crate::domain::{Availability, Heat, TorrentFile, TorrentRecord}; +use crate::{ + domain::{Availability, Heat, TorrentFile, TorrentRecord}, + search::IndexStatus, +}; use serde::Serialize; use std::time::{SystemTime, UNIX_EPOCH}; @@ -105,6 +108,7 @@ pub(crate) struct StatsResponse { pub(crate) persistence_rejected_full: u64, pub(crate) persistence_queue: usize, pub(crate) indexed_documents: u64, + pub(crate) index: IndexStatus, pub(crate) verification_queue: u64, pub(crate) verification_accepted: u64, pub(crate) verification_deduplicated: u64, diff --git a/src/search/src/app.rs b/src/search/src/app.rs index 8267beb..9dd1675 100644 --- a/src/search/src/app.rs +++ b/src/search/src/app.rs @@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration}; use crate::{ - search::SearchEngine, + search::SearchRuntime, storage::{RocksTorrentRepository, TorrentRepository}, }; use tokio_util::sync::CancellationToken; @@ -31,12 +31,9 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res metadata_limits.rule_id(), content_filter, )?); - let search_path = config.data_dir.join("tantivy"); - let (search, search_created) = if repository.content_filter_changed() { - (SearchEngine::recreate(&search_path)?, true) - } else { - SearchEngine::open_with_status(&search_path)? - }; + let search_bootstrap = + SearchRuntime::open(&config.data_dir, repository.content_filter_changed())?; + let search = search_bootstrap.runtime.clone(); disk_guard.probe(&config.data_dir, 0); let repository_api: Arc = repository.clone(); let mut persistence = PersistencePipeline::start( @@ -97,11 +94,10 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res let (index_fatal_tx, mut index_fatal) = tokio::sync::oneshot::channel(); let index_task = tokio::spawn(index_worker::run( repository.clone(), - search.clone(), + search_bootstrap, index_worker::IndexWorkerOptions { batch_size: config.index_batch_size, interval: Duration::from_millis(config.index_interval_millis), - prepare_full_reindex: search_created, }, disk_guard.clone(), index_cancel.clone(), diff --git a/src/search/src/backup.rs b/src/search/src/backup.rs index bc34677..91654a3 100644 --- a/src/search/src/backup.rs +++ b/src/search/src/backup.rs @@ -294,6 +294,10 @@ pub(crate) fn restore(data_dir: &Path, checkpoint: &Path) -> Result, #[arg(long, default_value = "benchmark-data")] pub(crate) output_dir: PathBuf, #[arg(long)] @@ -44,6 +46,12 @@ impl Args { if self.duplicate_every == 1 { return Err("duplicate-every 必须是零或至少为二".into()); } + if self + .files_per_record + .is_some_and(|files| !(1..=20_000).contains(&files)) + { + return Err("files-per-record 必须在 1 到 20000 之间".into()); + } Ok(()) } } @@ -57,4 +65,12 @@ mod tests { let args = Args::parse_from(["benchmark", "--duplicate-every", "1"]); assert!(args.validate().is_err()); } + + #[test] + fn validates_large_file_dataset_size() { + let valid = Args::parse_from(["benchmark", "--files-per-record", "2048"]); + assert!(valid.validate().is_ok()); + let invalid = Args::parse_from(["benchmark", "--files-per-record", "20001"]); + assert!(invalid.validate().is_err()); + } } diff --git a/src/search/src/bin/dht-benchmark/dataset.rs b/src/search/src/bin/dht-benchmark/dataset.rs index adcd485..592fdd9 100644 --- a/src/search/src/bin/dht-benchmark/dataset.rs +++ b/src/search/src/bin/dht-benchmark/dataset.rs @@ -9,9 +9,17 @@ pub(crate) const BASE_TIMESTAMP: u64 = 1_700_000_000; pub(crate) fn generate_record( index: usize, duplicate_every: usize, +) -> Result> { + generate_record_with_files(index, duplicate_every, None) +} + +pub(crate) fn generate_record_with_files( + index: usize, + duplicate_every: usize, + files_per_record: Option, ) -> Result> { let content_id = content_id(index, duplicate_every); - let file_count = 1 + content_id % 4; + let file_count = files_per_record.unwrap_or(1 + content_id % 4); let mut files = Vec::with_capacity(file_count); let main_size = 64 * 1024 * 1024 + (content_id as u64 % 8_192) * 1_048_576; files.push(TorrentFile { diff --git a/src/search/src/bin/dht-benchmark/report.rs b/src/search/src/bin/dht-benchmark/report.rs index 86d4878..b6a6d39 100644 --- a/src/search/src/bin/dht-benchmark/report.rs +++ b/src/search/src/bin/dht-benchmark/report.rs @@ -13,6 +13,7 @@ pub(crate) struct BenchmarkReport { pub(crate) records: usize, pub(crate) indexed_documents: u64, pub(crate) duplicate_every: usize, + pub(crate) files_per_record: Option, pub(crate) generation_seconds: f64, pub(crate) rocksdb_write_seconds: f64, pub(crate) rocksdb_records_per_second: f64, diff --git a/src/search/src/bin/dht-benchmark/runner.rs b/src/search/src/bin/dht-benchmark/runner.rs index b348060..bc23e22 100644 --- a/src/search/src/bin/dht-benchmark/runner.rs +++ b/src/search/src/bin/dht-benchmark/runner.rs @@ -13,7 +13,7 @@ use dht_search::{ use super::{ config::Args, - dataset::{BASE_TIMESTAMP, expected_document_count, generate_record}, + dataset::{BASE_TIMESTAMP, expected_document_count, generate_record_with_files}, metrics::{benchmark_query, directory_size, peak_memory_bytes, rate, ratio}, report::{BenchmarkReport, format_integer, print_query, print_summary}, workload::query_cases, @@ -43,6 +43,9 @@ pub(crate) fn run(args: Args) -> Result<(), Box> { println!("警告: debug 模式仅用于流程验证 性能结论必须使用 --release"); } println!("数据量: {}", format_integer(args.records as u64)); + if let Some(files) = args.files_per_record { + println!("每条文件数: {}", format_integer(files as u64)); + } println!("运行目录: {}", run_dir.display()); let rocksdb_dir = run_dir.join("rocksdb"); @@ -73,6 +76,7 @@ pub(crate) fn run(args: Args) -> Result<(), Box> { records: args.records, indexed_documents, duplicate_every: args.duplicate_every, + files_per_record: args.files_per_record, generation_seconds: generation_duration.as_secs_f64(), rocksdb_write_seconds: write_duration.as_secs_f64(), rocksdb_records_per_second: rate(args.records as u64, write_duration), @@ -113,7 +117,9 @@ fn populate( let batch_end = (batch_start + args.generation_batch_size).min(args.records); let generation_started = Instant::now(); let records: Vec<_> = (batch_start..batch_end) - .map(|index| generate_record(index, args.duplicate_every)) + .map(|index| { + generate_record_with_files(index, args.duplicate_every, args.files_per_record) + }) .collect::>()?; generation_duration += generation_started.elapsed(); diff --git a/src/search/src/crawler/runtime.rs b/src/search/src/crawler/runtime.rs index 0a155c2..4b6e5c5 100644 --- a/src/search/src/crawler/runtime.rs +++ b/src/search/src/crawler/runtime.rs @@ -214,10 +214,10 @@ impl CrawlerRuntime { Err(error) => tracing::warn!(%error, "有效性验证运行时任务异常"), } } - if let Some(task) = state.server_task.take() { - if let Err(error) = task.await { - tracing::warn!(%error, "DHT 采集运行时任务异常"); - } + if let Some(task) = state.server_task.take() + && let Err(error) = task.await + { + tracing::warn!(%error, "DHT 采集运行时任务异常"); } self.inner.stats.replace(DhtRuntimeStats::default()); self.inner.enabled.store(false, Ordering::Release); diff --git a/src/search/src/diagnostics/mod.rs b/src/search/src/diagnostics/mod.rs index d1374c1..d2eec38 100644 --- a/src/search/src/diagnostics/mod.rs +++ b/src/search/src/diagnostics/mod.rs @@ -17,7 +17,7 @@ use std::{ }; use crate::{ - search::SearchEngine, + search::SearchRuntime, storage::{RocksTorrentRepository, StorageDiagnostics as RocksDiagnostics}, }; use tokio_util::sync::CancellationToken; @@ -38,7 +38,7 @@ use store::{DiagnosticStore, DiagnosticStoreError}; #[derive(Clone)] pub(crate) struct DiagnosticSources { pub(crate) repository: Arc, - pub(crate) search: SearchEngine, + pub(crate) search: SearchRuntime, pub(crate) dht: CrawlerStats, pub(crate) persistence: PersistenceIngress, pub(crate) disk_guard: DiskGuard, @@ -328,7 +328,7 @@ fn unix_timestamp() -> u64 { #[cfg(test)] mod tests { use crate::{ - search::SearchEngine, + search::{SearchEngine, SearchRuntime}, storage::{RocksTorrentRepository, TorrentRepository}, }; use tempfile::TempDir; @@ -358,7 +358,9 @@ mod tests { config, DiagnosticSources { repository, - search: SearchEngine::open(directory.path().join("tantivy")).unwrap(), + search: SearchRuntime::from_engine( + SearchEngine::open(directory.path().join("tantivy")).unwrap(), + ), dht: CrawlerStats::default(), persistence: persistence.ingress.clone(), disk_guard, diff --git a/src/search/src/index_worker.rs b/src/search/src/index_worker.rs index b2a1f27..6a02959 100644 --- a/src/search/src/index_worker.rs +++ b/src/search/src/index_worker.rs @@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration}; use crate::{ - search::SearchEngine, + search::{RebuildSession, SearchBootstrap, SearchEngine, SearchRuntime}, storage::{RocksTorrentRepository, TorrentRepository}, }; use tokio_util::sync::CancellationToken; @@ -13,18 +13,43 @@ use crate::disk_guard::DiskGuard; pub(crate) struct IndexWorkerOptions { pub(crate) batch_size: usize, pub(crate) interval: Duration, - pub(crate) prepare_full_reindex: bool, } pub(crate) async fn run( repository: Arc, - search: SearchEngine, + bootstrap: SearchBootstrap, options: IndexWorkerOptions, disk_guard: DiskGuard, cancel: CancellationToken, fatal: tokio::sync::oneshot::Sender, ) -> Result<(), String> { - let mut prepare_full_reindex = options.prepare_full_reindex; + let SearchBootstrap { + runtime: search, + target, + rebuild, + } = bootstrap; + if let Some(rebuild) = rebuild { + match rebuild_index( + repository.clone(), + search.clone(), + target, + rebuild, + options.batch_size, + disk_guard.clone(), + cancel.clone(), + ) + .await + { + Ok(true) => {} + Ok(false) => return Ok(()), + Err(error) => { + tracing::error!(%error, "影子搜索索引重建失败 旧索引继续提供服务"); + search.mark_failed(error); + cancel.cancelled().await; + return Ok(()); + } + } + } let mut ticker = tokio::time::interval(options.interval); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let mut consecutive_retries = 0_u32; @@ -35,18 +60,8 @@ pub(crate) async fn run( let Some(_permit) = disk_guard.begin_new_write() else { continue; }; - if prepare_full_reindex { - let rebuild_repository = repository.clone(); - let records = tokio::task::spawn_blocking(move || { - rebuild_repository.prepare_full_reindex() - }) - .await - .map_err(|error| error.to_string())? - .map_err(|error| error.to_string())?; - tracing::info!(records, "检测到新搜索索引并准备全量重建"); - prepare_full_reindex = false; - } - match index_one_batch(repository.clone(), search.clone(), options.batch_size).await { + let active = search.active_engine().map_err(|error| error.to_string())?; + match index_one_batch(repository.clone(), active, options.batch_size).await { Ok(count) => { consecutive_retries = 0; if count > 0 { @@ -71,7 +86,93 @@ pub(crate) async fn run( } } - drain_before_shutdown(repository, search, options.batch_size, disk_guard).await + let active = search.active_engine().map_err(|error| error.to_string())?; + drain_before_shutdown(repository, active, options.batch_size, disk_guard).await +} + +async fn rebuild_index( + repository: Arc, + runtime: SearchRuntime, + target: SearchEngine, + mut session: RebuildSession, + batch_size: usize, + disk_guard: DiskGuard, + cancel: CancellationToken, +) -> Result { + let rebuild_batch_size = batch_size.clamp(1, 64); + if !session.prepared { + let rebuild_repository = repository.clone(); + let records = + tokio::task::spawn_blocking(move || rebuild_repository.prepare_full_reindex()) + .await + .map_err(|error| error.to_string())? + .map_err(|error| error.to_string())?; + runtime + .mark_prepared(&mut session, records) + .map_err(|error| error.to_string())?; + tracing::info!(records, reason = ?session.reason, "影子搜索索引已准备全量重建"); + } + + let mut retries = 0_u32; + let mut consistency_retries = 0_u8; + loop { + if cancel.is_cancelled() { + return Ok(false); + } + let Some(_permit) = disk_guard.begin_new_write() else { + runtime.mark_blocked(); + tokio::select! { + _ = cancel.cancelled() => return Ok(false), + _ = tokio::time::sleep(Duration::from_secs(1)) => continue, + } + }; + match index_one_batch(repository.clone(), target.clone(), rebuild_batch_size).await { + Ok(count) => { + retries = 0; + let inventory = repository.index_inventory(); + runtime.update_building(target.num_docs(), inventory.searchable_groups); + if count == 0 && inventory.pending_documents == 0 { + if target.num_docs() != inventory.searchable_groups { + consistency_retries = consistency_retries.saturating_add(1); + if consistency_retries > 1 { + return Err(format!( + "影子索引文档数不一致 indexed={} expected={}", + target.num_docs(), + inventory.searchable_groups + )); + } + let rebuild_repository = repository.clone(); + tokio::task::spawn_blocking(move || { + rebuild_repository.prepare_full_reindex() + }) + .await + .map_err(|error| error.to_string())? + .map_err(|error| error.to_string())?; + continue; + } + runtime + .finish_rebuild(session, target) + .map_err(|error| error.to_string())?; + tracing::info!( + documents = inventory.searchable_groups, + "影子搜索索引已原子切换" + ); + return Ok(true); + } + tokio::task::yield_now().await; + } + Err(IndexBatchError::Retryable(error)) => { + retries = retries.saturating_add(1); + let delay = retry_delay(retries); + tracing::warn!(%error, retry = retries, delay_ms = delay.as_millis(), "影子搜索索引遇到临时 I/O 错误"); + tokio::select! { + _ = cancel.cancelled() => return Ok(false), + _ = tokio::time::sleep(delay) => {} + } + } + Err(IndexBatchError::Fatal(error)) => return Err(error), + } + } } async fn drain_before_shutdown( diff --git a/src/search/src/search/document.rs b/src/search/src/search/document.rs index 375c006..d145171 100644 --- a/src/search/src/search/document.rs +++ b/src/search/src/search/document.rs @@ -5,7 +5,7 @@ use std::{collections::BTreeSet, path::Path}; use tantivy::{TantivyDocument, schema::Value}; use unicode_normalization::UnicodeNormalization; -use crate::domain::{AvailabilityStatus, ContentGroup, Heat, TorrentRecord}; +use crate::domain::{AvailabilityStatus, ContentGroup, Heat}; use super::{ SearchError, @@ -14,31 +14,63 @@ use super::{ }; pub(crate) fn from_group(group: &ContentGroup, fields: SearchFields) -> TantivyDocument { - const MAX_INDEXED_FILES: usize = 512; - const MAX_PATH_TEXT_BYTES: usize = 32 * 1024; + const MAX_INDEXED_FILES: usize = 2_048; + const MAX_PATH_TEXT_BYTES: usize = 256 * 1024; let record = &group.representative; let mut document = TantivyDocument::default(); document.add_text(fields.info_hash, record.info_hash.to_string()); - document.add_text(fields.name, normalize_bounded(&record.name, 512)); - document.add_text(fields.regex_text, normalize_bounded(&record.name, 512)); + let normalized_name = normalize_bounded(&record.name, 512); + document.add_text(fields.name, &normalized_name); + if let Some(field) = fields.exact_name { + document.add_text(field, &normalized_name); + } + document.add_text(fields.regex_text, &normalized_name); document.add_text(fields.display_name, &record.name); for alias in &group.aliases { let alias = normalize_bounded(alias, 512); document.add_text(fields.aliases, &alias); + if let Some(field) = fields.exact_aliases { + document.add_text(field, &alias); + } document.add_text(fields.regex_text, alias); } let mut indexed_path_bytes = 0_usize; - for file in record.files.iter().take(MAX_INDEXED_FILES) { - let path = normalize_bounded(&file.path, 512); + let mut files: Vec<_> = record + .files + .iter() + .map(|file| (file, normalize_bounded(&file.path, 512))) + .collect(); + files.sort_unstable_by(|(left, left_path), (right, right_path)| { + right + .size + .cmp(&left.size) + .then_with(|| left_path.cmp(right_path)) + }); + let mut indexed_files = Vec::with_capacity(files.len().min(MAX_INDEXED_FILES)); + for (file, path) in files.into_iter().take(MAX_INDEXED_FILES) { if indexed_path_bytes.saturating_add(path.len()) > MAX_PATH_TEXT_BYTES { - break; + continue; } indexed_path_bytes += path.len(); + let file_name = Path::new(&path) + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or(&path); + if let Some(field) = fields.file_names { + document.add_text(field, file_name); + } + if let Some(field) = fields.exact_file_names { + document.add_text(field, file_name); + } document.add_text(fields.files_text, &path); - document.add_text(fields.regex_text, path); + document.add_text(fields.regex_text, &path); + if file_name != path.as_str() { + document.add_text(fields.regex_text, file_name); + } + indexed_files.push(file); } - for extension in extensions(record) { + for extension in extensions(&indexed_files) { document.add_text(fields.extensions, extension); } document.add_u64(fields.total_size, record.total_size); @@ -123,11 +155,9 @@ fn normalize_bounded(value: &str, max_chars: usize) -> String { .to_lowercase() } -fn extensions(record: &TorrentRecord) -> BTreeSet { - record - .files +fn extensions(files: &[&crate::domain::TorrentFile]) -> BTreeSet { + files .iter() - .take(512) .filter_map(|file| Path::new(&file.path).extension()) .filter_map(|extension| extension.to_str()) .map(str::to_lowercase) diff --git a/src/search/src/search/filter.rs b/src/search/src/search/filter.rs index db81982..2f0f6a9 100644 --- a/src/search/src/search/filter.rs +++ b/src/search/src/search/filter.rs @@ -191,7 +191,44 @@ fn text_query(query: &str, fields: SearchFields) -> Box { if terms.is_empty() { return Box::new(AllQuery); } - let mut required = Vec::with_capacity(terms.len()); + let mut required = Vec::with_capacity(terms.len() + 1); + let mut exact = Vec::<(Occur, Box)>::new(); + if let Some(field) = fields.exact_name { + exact.push(( + Occur::Should, + Box::new(BoostQuery::new( + Box::new(TermQuery::new( + Term::from_field_text(field, &normalized), + IndexRecordOption::Basic, + )), + 12.0, + )), + )); + } + if let Some(field) = fields.exact_aliases { + exact.push(( + Occur::Should, + Box::new(BoostQuery::new( + Box::new(TermQuery::new( + Term::from_field_text(field, &normalized), + IndexRecordOption::Basic, + )), + 8.0, + )), + )); + } + if let Some(field) = fields.exact_file_names { + exact.push(( + Occur::Should, + Box::new(BoostQuery::new( + Box::new(TermQuery::new( + Term::from_field_text(field, &normalized), + IndexRecordOption::Basic, + )), + 5.0, + )), + )); + } for term in terms { let alternatives: Vec<(Occur, Box)> = vec![ ( @@ -201,7 +238,7 @@ fn text_query(query: &str, fields: SearchFields) -> Box { Term::from_field_text(fields.name, &term), IndexRecordOption::WithFreqs, )), - 3.0, + 4.0, )), ), ( @@ -214,6 +251,19 @@ fn text_query(query: &str, fields: SearchFields) -> Box { 2.0, )), ), + ( + Occur::Should, + Box::new(BoostQuery::new( + Box::new(TermQuery::new( + Term::from_field_text( + fields.file_names.unwrap_or(fields.files_text), + &term, + ), + IndexRecordOption::WithFreqs, + )), + 1.5, + )), + ), ( Occur::Should, Box::new(TermQuery::new( @@ -227,6 +277,9 @@ fn text_query(query: &str, fields: SearchFields) -> Box { Box::new(BooleanQuery::new(alternatives)) as Box, )); } + if !exact.is_empty() { + required.push((Occur::Should, Box::new(BooleanQuery::new(exact)))); + } Box::new(BooleanQuery::new(required)) } diff --git a/src/search/src/search/indexer.rs b/src/search/src/search/indexer.rs index bdbf7f4..28b1bbb 100644 --- a/src/search/src/search/indexer.rs +++ b/src/search/src/search/indexer.rs @@ -23,7 +23,7 @@ use tantivy::{ use super::{ IndexingError, SearchError, query::{SearchOptions, SearchPage, SearchSort}, - schema::{MIXED_NGRAM_TOKENIZER, SearchFields, build_schema}, + schema::{MIXED_NGRAM_TOKENIZER, SearchFields, build_schema, fields_from_schema}, }; const INDEX_WRITER_MEMORY_BYTES: usize = 64 * 1024 * 1024; @@ -47,6 +47,7 @@ pub struct SearchDiagnostics { } struct SearchInner { + index_schema: tantivy::schema::Schema, reader: IndexReader, writer: Mutex, fields: SearchFields, @@ -77,7 +78,7 @@ impl SearchEngine { .map_err(|error| SearchError::Directory(error.to_string()))?; let mut directory = MmapDirectory::open(&path) .map_err(|error| SearchError::Directory(error.to_string()))?; - let (expected_schema, fields) = build_schema(); + let (expected_schema, _) = build_schema(); let exists = Index::exists(&directory).map_err(|error| SearchError::Directory(error.to_string()))?; let mut created = !exists; @@ -99,6 +100,24 @@ impl SearchEngine { } else { Index::open_or_create(directory, expected_schema)? }; + let fields = fields_from_schema(&index.schema())?; + Ok((Self::from_index(index, fields)?, created)) + } + + pub(crate) fn open_existing(path: impl AsRef) -> Result { + let path = path.as_ref(); + let directory = + MmapDirectory::open(path).map_err(|error| SearchError::Directory(error.to_string()))?; + let index = Index::open(directory)?; + let fields = fields_from_schema(&index.schema())?; + Self::from_index(index, fields) + } + + pub(crate) fn has_current_schema(&self) -> bool { + self.inner.index_schema == build_schema().0 + } + + fn from_index(index: Index, fields: SearchFields) -> Result { let analyzer = TextAnalyzer::builder(NgramTokenizer::all_ngrams(1, 20)?) .filter(LowerCaser) .build(); @@ -108,21 +127,19 @@ impl SearchEngine { .reload_policy(ReloadPolicy::OnCommitWithDelay) .try_into()?; let writer = index.writer_with_num_threads(1, INDEX_WRITER_MEMORY_BYTES)?; - Ok(( - Self { - inner: Arc::new(SearchInner { - reader, - writer: Mutex::new(writer), - fields, - commits: AtomicU64::new(0), - commit_failures: AtomicU64::new(0), - last_commit_at: AtomicU64::new(0), - last_commit_duration_millis: AtomicU64::new(0), - last_commit_documents: AtomicU64::new(0), - }), - }, - created, - )) + Ok(Self { + inner: Arc::new(SearchInner { + index_schema: index.schema(), + reader, + writer: Mutex::new(writer), + fields, + commits: AtomicU64::new(0), + commit_failures: AtomicU64::new(0), + last_commit_at: AtomicU64::new(0), + last_commit_duration_millis: AtomicU64::new(0), + last_commit_documents: AtomicU64::new(0), + }), + }) } fn index_groups(&self, groups: &[ContentGroup]) -> Result<(), SearchError> { @@ -241,9 +258,22 @@ impl SearchEngine { query.as_ref(), &TopDocs::with_limit(limit) .and_offset(offset) - .order_by_score(), + .tweak_score(|segment| { + let heat = segment + .fast_fields() + .u64("heat_score") + .expect("heat_score fast field exists") + .first_or_default_col(0); + let last_seen = segment + .fast_fields() + .u64("last_seen") + .expect("last_seen fast field exists") + .first_or_default_col(0); + move |doc, score| (score, heat.get_val(doc), last_seen.get_val(doc)) + }), )? .into_iter() + .map(|((score, _, _), address)| (score, address)) .collect(), SearchSort::Latest => sorted_documents( &searcher, @@ -416,6 +446,83 @@ mod tests { assert_eq!(engine.search("ubuntu.iso", 0, 10).unwrap().total, 1); } + #[test] + fn large_torrents_index_the_largest_2048_files() { + let directory = TempDir::new().unwrap(); + let engine = SearchEngine::open(directory.path()).unwrap(); + let mut record = record(); + record.files = (0..2_050) + .map(|index| TorrentFile { + path: format!("collection/item-{index:04}.bin"), + size: index as u64 + 1, + }) + .collect(); + record.total_size = record.files.iter().map(|file| file.size).sum(); + record.content_key = crate::domain::content_key(&record.files).unwrap(); + index_records(&engine, &[record]); + + assert_eq!(engine.search("item-0002", 0, 10).unwrap().total, 1); + assert_eq!(engine.search("item-0000", 0, 10).unwrap().total, 0); + } + + #[test] + fn large_path_text_obeys_the_256_kib_budget() { + let directory = TempDir::new().unwrap(); + let engine = SearchEngine::open(directory.path()).unwrap(); + let suffix = "x".repeat(210); + let mut record = record(); + record.files = (0..1_500) + .map(|index| TorrentFile { + path: format!("collection/largest-{index:04}-{suffix}.bin"), + size: index as u64 + 1, + }) + .collect(); + record.total_size = record.files.iter().map(|file| file.size).sum(); + record.content_key = crate::domain::content_key(&record.files).unwrap(); + index_records(&engine, &[record]); + + assert_eq!(engine.search("largest-1499", 0, 10).unwrap().total, 1); + assert_eq!(engine.search("largest-0000", 0, 10).unwrap().total, 0); + } + + #[test] + fn exact_title_ranks_above_a_file_name_match() { + let directory = TempDir::new().unwrap(); + let engine = SearchEngine::open(directory.path()).unwrap(); + let mut title = record(); + title.name = "Ubuntu".into(); + let mut file = record(); + file.info_hash = InfoHash::from_bytes([2; 20]); + file.content_key = [3; 32]; + file.name = "Linux Collection".into(); + file.files[0].path = "images/ubuntu.iso".into(); + index_records(&engine, &[file, title.clone()]); + + let page = engine.search("ubuntu", 0, 10).unwrap(); + assert_eq!(page.total, 2); + assert_eq!(page.hits[0].info_hash, title.info_hash.to_string()); + } + + #[test] + fn equal_text_scores_use_heat_then_last_seen() { + let directory = TempDir::new().unwrap(); + let engine = SearchEngine::open(directory.path()).unwrap(); + let now = unix_timestamp(); + let mut older = record(); + older.name = "Equal Search Name".into(); + older.activity_updated_at = now; + older.activity_score_millis = 1_000; + older.last_seen = 100; + let mut newer = older.clone(); + newer.info_hash = InfoHash::from_bytes([4; 20]); + newer.content_key = [4; 32]; + newer.last_seen = 200; + index_records(&engine, &[older, newer.clone()]); + + let page = engine.search("equal search", 0, 10).unwrap(); + assert_eq!(page.hits[0].info_hash, newer.info_hash.to_string()); + } + #[test] fn mixed_substrings_match_chinese_and_release_names() { let directory = TempDir::new().unwrap(); diff --git a/src/search/src/search/mod.rs b/src/search/src/search/mod.rs index 91cfe28..02ee73f 100644 --- a/src/search/src/search/mod.rs +++ b/src/search/src/search/mod.rs @@ -4,12 +4,15 @@ mod document; mod filter; mod indexer; mod query; +mod runtime; mod schema; pub use indexer::{SearchDiagnostics, SearchEngine}; pub use query::{ AvailabilitySummary, SearchHit, SearchMode, SearchOptions, SearchPage, SearchSort, }; +pub use runtime::{IndexRebuildReason, IndexState, IndexStatus, SearchRuntime}; +pub(crate) use runtime::{RebuildSession, SearchBootstrap}; #[derive(Debug, thiserror::Error)] pub enum SearchError { @@ -19,6 +22,8 @@ pub enum SearchError { Directory(String), #[error("搜索文档缺少字段 {0}")] MissingField(&'static str), + #[error("搜索索引尚未可用")] + Unavailable, } impl SearchError { diff --git a/src/search/src/search/runtime.rs b/src/search/src/search/runtime.rs new file mode 100644 index 0000000..ad46834 --- /dev/null +++ b/src/search/src/search/runtime.rs @@ -0,0 +1,772 @@ +// 负责管理活动与影子 Tantivy 代际并提供可恢复的原子切换状态 + +use std::{ + fs::{self, OpenOptions}, + io::Write, + path::{Path, PathBuf}, + sync::{Arc, RwLock}, + time::{SystemTime, UNIX_EPOCH}, +}; + +use serde::{Deserialize, Serialize}; + +use crate::storage::IndexInventory; + +use super::{SearchDiagnostics, SearchEngine, SearchError, SearchOptions, SearchPage}; + +const INDEX_DOCUMENT_VERSION: u32 = 2; +const MANAGED_DIRECTORY: &str = "search-index"; +const GENERATIONS_DIRECTORY: &str = "generations"; +const CURRENT_FILE: &str = "CURRENT"; +const BUILDING_FILE: &str = "BUILDING.json"; +const GENERATION_FILE: &str = "generation.json"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IndexState { + Ready, + Initializing, + Rebuilding, + Switching, + Blocked, + Failed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IndexRebuildReason { + Initial, + ContentFilter, + Schema, + DocumentFormat, + Missing, + Corrupt, +} + +#[derive(Debug, Clone, Serialize)] +pub struct IndexStatus { + pub state: IndexState, + pub reason: Option, + pub stored_torrents: u64, + pub searchable_groups: u64, + pub active_documents: u64, + pub building_documents: u64, + pub target_documents: u64, + pub pending_documents: u64, + pub progress_percent: f64, + pub started_at: Option, + pub last_completed_at: Option, + pub error: Option, +} + +#[derive(Clone)] +pub struct SearchRuntime { + inner: Arc, +} + +struct SearchRuntimeInner { + active: RwLock>, + lifecycle: RwLock, + managed_root: PathBuf, + legacy_path: PathBuf, +} + +#[derive(Debug, Clone)] +struct LifecycleStatus { + state: IndexState, + reason: Option, + building_documents: u64, + target_documents: u64, + started_at: Option, + last_completed_at: Option, + error: Option, +} + +pub(crate) struct SearchBootstrap { + pub(crate) runtime: SearchRuntime, + pub(crate) target: SearchEngine, + pub(crate) rebuild: Option, +} + +#[derive(Debug, Clone)] +pub(crate) struct RebuildSession { + generation: String, + path: PathBuf, + previous_path: Option, + pub(crate) prepared: bool, + pub(crate) reason: IndexRebuildReason, + started_at: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct BuildingManifest { + generation: String, + reason: IndexRebuildReason, + document_version: u32, + prepared: bool, + target_documents: u64, + started_at: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct GenerationManifest { + document_version: u32, +} + +impl SearchRuntime { + #[cfg(test)] + pub(crate) fn from_engine(engine: SearchEngine) -> Self { + let documents = engine.num_docs(); + Self::new( + Some(engine), + PathBuf::new(), + PathBuf::new(), + LifecycleStatus { + state: IndexState::Ready, + reason: None, + building_documents: 0, + target_documents: documents, + started_at: None, + last_completed_at: None, + error: None, + }, + ) + } + + pub(crate) fn open( + data_dir: &Path, + content_filter_changed: bool, + ) -> Result { + let managed_root = data_dir.join(MANAGED_DIRECTORY); + let generations = managed_root.join(GENERATIONS_DIRECTORY); + let legacy_path = data_dir.join("tantivy"); + fs::create_dir_all(&generations).map_err(directory_error)?; + + let current_generation = read_trimmed(managed_root.join(CURRENT_FILE))?; + let active_path = current_generation + .as_ref() + .map(|generation| generations.join(generation)) + .or_else(|| legacy_path.exists().then(|| legacy_path.clone())); + let mut active_error = None; + let active = + active_path + .as_ref() + .and_then(|path| match SearchEngine::open_existing(path) { + Ok(engine) => Some(engine), + Err(error) => { + tracing::error!(path = %path.display(), %error, "现有搜索索引无法打开"); + active_error = Some("现有搜索索引无法打开 请查看服务日志".to_owned()); + None + } + }); + let active_version = current_generation + .as_ref() + .and_then(|generation| read_generation(&generations.join(generation)).ok()) + .map(|manifest| manifest.document_version); + + let reason = if content_filter_changed { + Some(IndexRebuildReason::ContentFilter) + } else if active_path.is_none() { + Some(IndexRebuildReason::Initial) + } else if active.is_none() { + Some(IndexRebuildReason::Corrupt) + } else if active + .as_ref() + .is_some_and(|engine| !engine.has_current_schema()) + { + Some(IndexRebuildReason::Schema) + } else if active_version != Some(INDEX_DOCUMENT_VERSION) { + Some(IndexRebuildReason::DocumentFormat) + } else { + None + }; + + if reason.is_none() { + let active = active.expect("ready index has an active engine"); + let _ = fs::remove_file(managed_root.join(BUILDING_FILE)); + cleanup_obsolete( + &generations, + current_generation.as_deref(), + None, + &legacy_path, + ); + let runtime = Self::new( + Some(active.clone()), + managed_root, + legacy_path, + LifecycleStatus { + state: IndexState::Ready, + reason: None, + building_documents: 0, + target_documents: active.num_docs(), + started_at: None, + last_completed_at: None, + error: None, + }, + ); + return Ok(SearchBootstrap { + runtime, + target: active, + rebuild: None, + }); + } + + let reason = reason.expect("rebuild reason exists"); + let previous_path = active_path.filter(|_| active.is_some()); + let existing_build = read_building(&managed_root) + .ok() + .flatten() + .filter(|manifest| { + manifest.document_version == INDEX_DOCUMENT_VERSION && manifest.reason == reason + }); + let (manifest, target) = if let Some(manifest) = existing_build { + let path = generations.join(&manifest.generation); + match SearchEngine::open_existing(&path) { + Ok(engine) if engine.has_current_schema() => (manifest, engine), + _ => create_generation(&managed_root, &generations, reason)?, + } + } else { + create_generation(&managed_root, &generations, reason)? + }; + cleanup_obsolete( + &generations, + current_generation.as_deref(), + Some(&manifest.generation), + &legacy_path, + ); + let building_documents = target.num_docs(); + let state = if previous_path.is_some() { + IndexState::Rebuilding + } else { + IndexState::Initializing + }; + let runtime_active = active.or_else(|| Some(target.clone())); + let runtime = Self::new( + runtime_active, + managed_root, + legacy_path, + LifecycleStatus { + state, + reason: Some(reason), + building_documents, + target_documents: manifest.target_documents, + started_at: Some(manifest.started_at), + last_completed_at: None, + error: active_error, + }, + ); + let generation = manifest.generation.clone(); + Ok(SearchBootstrap { + runtime, + target, + rebuild: Some(RebuildSession { + generation, + path: generations.join(&manifest.generation), + previous_path, + prepared: manifest.prepared, + reason, + started_at: manifest.started_at, + }), + }) + } + + fn new( + active: Option, + managed_root: PathBuf, + legacy_path: PathBuf, + lifecycle: LifecycleStatus, + ) -> Self { + Self { + inner: Arc::new(SearchRuntimeInner { + active: RwLock::new(active), + lifecycle: RwLock::new(lifecycle), + managed_root, + legacy_path, + }), + } + } + + pub fn search_with(&self, options: SearchOptions) -> Result { + self.active_engine()?.search_with(options) + } + + pub fn num_docs(&self) -> u64 { + self.inner + .active + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .map_or(0, SearchEngine::num_docs) + } + + pub fn diagnostics(&self) -> SearchDiagnostics { + self.inner + .active + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .map_or_else(SearchDiagnostics::default, SearchEngine::diagnostics) + } + + pub fn status(&self, inventory: IndexInventory) -> IndexStatus { + let lifecycle = self + .inner + .lifecycle + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + let target_documents = if lifecycle.state == IndexState::Ready { + inventory.searchable_groups + } else { + lifecycle.target_documents + }; + let building_documents = if lifecycle.state == IndexState::Ready { + 0 + } else { + lifecycle.building_documents + }; + let progress = if lifecycle.state == IndexState::Ready { + 100.0 + } else if target_documents == 0 { + 0.0 + } else { + building_documents.min(target_documents) as f64 * 100.0 / target_documents as f64 + }; + IndexStatus { + state: lifecycle.state, + reason: lifecycle.reason, + stored_torrents: inventory.stored_torrents, + searchable_groups: inventory.searchable_groups, + active_documents: self.num_docs(), + building_documents, + target_documents, + pending_documents: inventory.pending_documents, + progress_percent: (progress * 100.0).round() / 100.0, + started_at: lifecycle.started_at, + last_completed_at: lifecycle.last_completed_at, + error: lifecycle.error, + } + } + + pub(crate) fn active_engine(&self) -> Result { + self.inner + .active + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + .ok_or(SearchError::Unavailable) + } + + pub(crate) fn mark_prepared( + &self, + session: &mut RebuildSession, + target_documents: u64, + ) -> Result<(), SearchError> { + session.prepared = true; + write_building( + &self.inner.managed_root, + &BuildingManifest { + generation: session.generation.clone(), + reason: session.reason, + document_version: INDEX_DOCUMENT_VERSION, + prepared: true, + target_documents, + started_at: session.started_at, + }, + )?; + let mut lifecycle = self + .inner + .lifecycle + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + lifecycle.target_documents = target_documents; + lifecycle.error = None; + Ok(()) + } + + pub(crate) fn update_building(&self, documents: u64, target: u64) { + let mut lifecycle = self + .inner + .lifecycle + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + lifecycle.state = if lifecycle.state == IndexState::Initializing { + IndexState::Initializing + } else { + IndexState::Rebuilding + }; + lifecycle.building_documents = documents; + lifecycle.target_documents = target; + lifecycle.error = None; + } + + pub(crate) fn mark_blocked(&self) { + self.inner + .lifecycle + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .state = IndexState::Blocked; + } + + pub(crate) fn mark_failed(&self, _error: String) { + let mut lifecycle = self + .inner + .lifecycle + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + lifecycle.state = IndexState::Failed; + lifecycle.error = Some("影子索引构建失败 请查看服务日志".to_owned()); + } + + pub(crate) fn finish_rebuild( + &self, + session: RebuildSession, + target: SearchEngine, + ) -> Result<(), SearchError> { + { + let mut lifecycle = self + .inner + .lifecycle + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + lifecycle.state = IndexState::Switching; + } + atomic_write( + &self.inner.managed_root.join(CURRENT_FILE), + format!("{}\n", session.generation).as_bytes(), + )?; + let previous = self + .inner + .active + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .replace(target.clone()); + drop(previous); + let _ = fs::remove_file(self.inner.managed_root.join(BUILDING_FILE)); + { + let mut lifecycle = self + .inner + .lifecycle + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + lifecycle.state = IndexState::Ready; + lifecycle.reason = None; + lifecycle.building_documents = target.num_docs(); + lifecycle.target_documents = target.num_docs(); + lifecycle.started_at = None; + lifecycle.last_completed_at = Some(unix_timestamp()); + lifecycle.error = None; + } + if let Some(previous_path) = session.previous_path + && previous_path != session.path + && is_safe_index_path( + &previous_path, + &self.inner.managed_root, + &self.inner.legacy_path, + ) + && let Err(error) = fs::remove_dir_all(&previous_path) + { + tracing::warn!(path = %previous_path.display(), %error, "旧搜索索引暂时无法清理"); + } + Ok(()) + } +} + +fn create_generation( + managed_root: &Path, + generations: &Path, + reason: IndexRebuildReason, +) -> Result<(BuildingManifest, SearchEngine), SearchError> { + let generation = format!("g-{:020}-{}", unix_timestamp_millis(), std::process::id()); + let path = generations.join(&generation); + let engine = SearchEngine::open(&path)?; + atomic_write( + &path.join(GENERATION_FILE), + &serde_json::to_vec_pretty(&GenerationManifest { + document_version: INDEX_DOCUMENT_VERSION, + }) + .map_err(|error| SearchError::Directory(error.to_string()))?, + )?; + let manifest = BuildingManifest { + generation, + reason, + document_version: INDEX_DOCUMENT_VERSION, + prepared: false, + target_documents: 0, + started_at: unix_timestamp(), + }; + write_building(managed_root, &manifest)?; + Ok((manifest, engine)) +} + +fn read_generation(path: &Path) -> Result { + let bytes = fs::read(path.join(GENERATION_FILE)).map_err(directory_error)?; + serde_json::from_slice(&bytes).map_err(|error| SearchError::Directory(error.to_string())) +} + +fn read_building(root: &Path) -> Result, SearchError> { + let path = root.join(BUILDING_FILE); + if !path.exists() { + return Ok(None); + } + let bytes = fs::read(path).map_err(directory_error)?; + serde_json::from_slice(&bytes) + .map(Some) + .map_err(|error| SearchError::Directory(error.to_string())) +} + +fn write_building(root: &Path, manifest: &BuildingManifest) -> Result<(), SearchError> { + let bytes = serde_json::to_vec_pretty(manifest) + .map_err(|error| SearchError::Directory(error.to_string()))?; + atomic_write(&root.join(BUILDING_FILE), &bytes) +} + +fn read_trimmed(path: PathBuf) -> Result, SearchError> { + if !path.exists() { + return Ok(None); + } + let value = fs::read_to_string(path).map_err(directory_error)?; + let value = value.trim(); + Ok((!value.is_empty()).then(|| value.to_owned())) +} + +fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), SearchError> { + let parent = path + .parent() + .ok_or_else(|| SearchError::Directory("索引状态文件没有父目录".to_owned()))?; + fs::create_dir_all(parent).map_err(directory_error)?; + let temporary = path.with_extension(format!("tmp-{}", std::process::id())); + let result = (|| { + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(&temporary)?; + file.write_all(bytes)?; + file.sync_all()?; + drop(file); + replace_file(&temporary, path)?; + #[cfg(unix)] + fs::File::open(parent)?.sync_all()?; + Ok::<(), std::io::Error>(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary); + } + result.map_err(directory_error) +} + +#[cfg(windows)] +fn replace_file(source: &Path, destination: &Path) -> std::io::Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW, + }; + + let source: Vec = source.as_os_str().encode_wide().chain(Some(0)).collect(); + let destination: Vec = destination + .as_os_str() + .encode_wide() + .chain(Some(0)) + .collect(); + if unsafe { + MoveFileExW( + source.as_ptr(), + destination.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + } == 0 + { + Err(std::io::Error::last_os_error()) + } else { + Ok(()) + } +} + +#[cfg(not(windows))] +fn replace_file(source: &Path, destination: &Path) -> std::io::Result<()> { + fs::rename(source, destination) +} + +fn is_safe_index_path(path: &Path, root: &Path, legacy: &Path) -> bool { + let safe_location = path == legacy || path.starts_with(root.join(GENERATIONS_DIRECTORY)); + safe_location + && fs::symlink_metadata(path) + .map(|metadata| !metadata.file_type().is_symlink()) + .unwrap_or(false) +} + +fn cleanup_obsolete( + generations: &Path, + current: Option<&str>, + building: Option<&str>, + legacy: &Path, +) { + if current.is_some() + && legacy.exists() + && fs::symlink_metadata(legacy) + .map(|metadata| !metadata.file_type().is_symlink()) + .unwrap_or(false) + && let Err(error) = fs::remove_dir_all(legacy) + { + tracing::warn!(path = %legacy.display(), %error, "遗留搜索索引暂时无法清理"); + } + let Ok(entries) = fs::read_dir(generations) else { + return; + }; + for entry in entries.flatten() { + let name = entry.file_name(); + let name = name.to_string_lossy(); + if current == Some(name.as_ref()) || building == Some(name.as_ref()) { + continue; + } + let path = entry.path(); + let Ok(file_type) = entry.file_type() else { + continue; + }; + if !file_type.is_dir() || file_type.is_symlink() { + continue; + } + if let Err(error) = fs::remove_dir_all(&path) { + tracing::warn!(path = %path.display(), %error, "过期搜索索引代际暂时无法清理"); + } + } +} + +fn directory_error(error: std::io::Error) -> SearchError { + SearchError::Directory(error.to_string()) +} + +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 tempfile::TempDir; + + use crate::{ + domain::{TorrentFile, content_key, test_record}, + storage::{RocksTorrentRepository, TorrentRepository}, + }; + + use super::*; + + fn second_record() -> crate::domain::TorrentRecord { + let mut record = test_record(2, 20); + record.name = "Second Release".into(); + record.files = vec![TorrentFile { + path: "second/release.iso".into(), + size: 84, + }]; + record.total_size = 84; + record.content_key = content_key(&record.files).unwrap(); + record + } + + fn prepare_and_finish( + repository: &RocksTorrentRepository, + runtime: &SearchRuntime, + target: &SearchEngine, + mut session: RebuildSession, + ) { + let total = repository.prepare_full_reindex().unwrap(); + runtime.mark_prepared(&mut session, total).unwrap(); + while target.index_pending(repository, 1, 30).unwrap() > 0 {} + runtime.finish_rebuild(session, target.clone()).unwrap(); + } + + #[test] + fn old_index_serves_until_shadow_switches() { + let directory = TempDir::new().unwrap(); + let repository = RocksTorrentRepository::open(directory.path().join("rocksdb")).unwrap(); + repository.upsert(test_record(1, 10)).unwrap(); + let legacy = SearchEngine::open(directory.path().join("tantivy")).unwrap(); + legacy.index_pending(&repository, 10, 20).unwrap(); + drop(legacy); + + let bootstrap = SearchRuntime::open(directory.path(), false).unwrap(); + assert_eq!(bootstrap.runtime.num_docs(), 1); + assert_eq!( + bootstrap + .runtime + .search_with(SearchOptions { + query: "Example".into(), + limit: 10, + ..SearchOptions::default() + }) + .unwrap() + .total, + 1 + ); + repository.upsert(second_record()).unwrap(); + assert_eq!( + bootstrap + .runtime + .search_with(SearchOptions { + query: "Second".into(), + limit: 10, + ..SearchOptions::default() + }) + .unwrap() + .total, + 0 + ); + + prepare_and_finish( + &repository, + &bootstrap.runtime, + &bootstrap.target, + bootstrap.rebuild.unwrap(), + ); + assert_eq!(bootstrap.runtime.num_docs(), 2); + assert_eq!( + bootstrap + .runtime + .search_with(SearchOptions { + query: "Second".into(), + limit: 10, + ..SearchOptions::default() + }) + .unwrap() + .total, + 1 + ); + assert!(!directory.path().join("tantivy").exists()); + } + + #[test] + fn prepared_shadow_resumes_without_restarting_from_zero() { + let directory = TempDir::new().unwrap(); + let repository = RocksTorrentRepository::open(directory.path().join("rocksdb")).unwrap(); + repository.upsert(test_record(1, 10)).unwrap(); + repository.upsert(second_record()).unwrap(); + let legacy = SearchEngine::open(directory.path().join("tantivy")).unwrap(); + legacy.index_pending(&repository, 10, 20).unwrap(); + drop(legacy); + + let first = SearchRuntime::open(directory.path(), false).unwrap(); + let mut session = first.rebuild.unwrap(); + let total = repository.prepare_full_reindex().unwrap(); + first.runtime.mark_prepared(&mut session, total).unwrap(); + assert_eq!(first.target.index_pending(&repository, 1, 30).unwrap(), 1); + assert_eq!(first.target.num_docs(), 1); + drop(first.target); + drop(first.runtime); + + let resumed = SearchRuntime::open(directory.path(), false).unwrap(); + assert!(resumed.rebuild.as_ref().unwrap().prepared); + assert_eq!(resumed.target.num_docs(), 1); + assert_eq!(repository.index_inventory().pending_documents, 1); + } +} diff --git a/src/search/src/search/schema.rs b/src/search/src/search/schema.rs index f3f340e..8e8bcba 100644 --- a/src/search/src/search/schema.rs +++ b/src/search/src/search/schema.rs @@ -10,8 +10,12 @@ pub(crate) const MIXED_NGRAM_TOKENIZER: &str = "dht_mixed_ngram"; pub(crate) struct SearchFields { pub(crate) info_hash: Field, pub(crate) name: Field, + pub(crate) exact_name: Option, pub(crate) display_name: Field, pub(crate) aliases: Field, + pub(crate) exact_aliases: Option, + pub(crate) file_names: Option, + pub(crate) exact_file_names: Option, pub(crate) files_text: Field, pub(crate) regex_text: Field, pub(crate) extensions: Field, @@ -37,8 +41,12 @@ pub(crate) fn build_schema() -> (Schema, SearchFields) { .set_index_option(IndexRecordOption::WithFreqsAndPositions), ); let name = builder.add_text_field("name", indexed_text.clone()); + let exact_name = builder.add_text_field("exact_name", STRING); let display_name = builder.add_text_field("display_name", STORED); let aliases = builder.add_text_field("aliases", indexed_text.clone()); + let exact_aliases = builder.add_text_field("exact_aliases", STRING); + let file_names = builder.add_text_field("file_names", indexed_text.clone()); + let exact_file_names = builder.add_text_field("exact_file_names", STRING); let files_text = builder.add_text_field("files_text", indexed_text); let regex_text = builder.add_text_field("regex_text", STRING); let extensions = builder.add_text_field("extensions", STRING); @@ -59,8 +67,12 @@ pub(crate) fn build_schema() -> (Schema, SearchFields) { SearchFields { info_hash, name, + exact_name: Some(exact_name), display_name, aliases, + exact_aliases: Some(exact_aliases), + file_names: Some(file_names), + exact_file_names: Some(exact_file_names), files_text, regex_text, extensions, @@ -78,3 +90,30 @@ pub(crate) fn build_schema() -> (Schema, SearchFields) { }, ) } + +pub(crate) fn fields_from_schema(schema: &Schema) -> tantivy::Result { + Ok(SearchFields { + info_hash: schema.get_field("info_hash")?, + name: schema.get_field("name")?, + exact_name: schema.get_field("exact_name").ok(), + display_name: schema.get_field("display_name")?, + aliases: schema.get_field("aliases")?, + exact_aliases: schema.get_field("exact_aliases").ok(), + file_names: schema.get_field("file_names").ok(), + exact_file_names: schema.get_field("exact_file_names").ok(), + files_text: schema.get_field("files_text")?, + regex_text: schema.get_field("regex_text")?, + extensions: schema.get_field("extensions")?, + total_size: schema.get_field("total_size")?, + file_count: schema.get_field("file_count")?, + first_seen: schema.get_field("first_seen")?, + last_seen: schema.get_field("last_seen")?, + seen_count: schema.get_field("seen_count")?, + content_key: schema.get_field("content_key")?, + availability_status: schema.get_field("availability_status")?, + reachable_peers: schema.get_field("reachable_peers")?, + last_verified_at: schema.get_field("last_verified_at")?, + heat_score: schema.get_field("heat_score")?, + variant_count: schema.get_field("variant_count")?, + }) +} diff --git a/src/search/src/storage/keys.rs b/src/search/src/storage/keys.rs index a9d7f0a..62e553c 100644 --- a/src/search/src/storage/keys.rs +++ b/src/search/src/storage/keys.rs @@ -7,6 +7,9 @@ pub(crate) const DATABASE_FORMAT_VALUE: &[u8] = b"dht-search"; pub(crate) const VERIFICATION_QUEUE_COUNT_KEY: &[u8] = b"\x00verification-queue-count"; pub(crate) const CONTENT_FILTER_FINGERPRINT_KEY: &[u8] = b"\x00content-filter-fingerprint"; pub(crate) const CONTENT_FILTER_MIGRATION_KEY: &[u8] = b"\x00content-filter-migration"; +pub(crate) const TORRENT_COUNT_KEY: &[u8] = b"\x00torrent-count"; +pub(crate) const CONTENT_GROUP_COUNT_KEY: &[u8] = b"\x00content-group-count"; +pub(crate) const PENDING_INDEX_COUNT_KEY: &[u8] = b"\x00pending-index-count"; const TORRENT_PREFIX: u8 = b't'; const REJECTED_METADATA_PREFIX: u8 = b'r'; const CONTENT_PREFIX: u8 = b'c'; diff --git a/src/search/src/storage/mod.rs b/src/search/src/storage/mod.rs index 543729f..70e85d3 100644 --- a/src/search/src/storage/mod.rs +++ b/src/search/src/storage/mod.rs @@ -7,9 +7,9 @@ mod repository; mod rocks; pub use repository::{ - CheckpointSummary, ContentGroupTask, ContentVariants, StorageDiagnostics, StorageError, - TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, - VerificationRequest, + CheckpointSummary, ContentGroupTask, ContentVariants, IndexInventory, StorageDiagnostics, + StorageError, TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome, + VerificationPriority, VerificationRequest, }; #[cfg(feature = "rocksdb-storage")] pub use rocks::RocksTorrentRepository; diff --git a/src/search/src/storage/repository.rs b/src/search/src/storage/repository.rs index b9dedc8..388f4ea 100644 --- a/src/search/src/storage/repository.rs +++ b/src/search/src/storage/repository.rs @@ -78,6 +78,17 @@ pub trait TorrentRepository: Send + Sync { ) -> Result<(), StorageError>; fn verification_queue_len(&self) -> Result; + + fn index_inventory(&self) -> IndexInventory { + IndexInventory::default() + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)] +pub struct IndexInventory { + pub stored_torrents: u64, + pub searchable_groups: u64, + pub pending_documents: u64, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src/search/src/storage/rocks.rs b/src/search/src/storage/rocks.rs index 56951fd..3f1b3e5 100644 --- a/src/search/src/storage/rocks.rs +++ b/src/search/src/storage/rocks.rs @@ -1,6 +1,9 @@ // 负责实现 RocksDB 打开配置批量写入精确查询和关闭流程 -use std::sync::{Arc, Mutex}; +use std::sync::{ + Arc, Mutex, + atomic::{AtomicU64, Ordering}, +}; use rocksdb::{DB, Direction, IteratorMode, WriteBatch}; @@ -11,16 +14,17 @@ use crate::domain::{ use super::{ keys::{ + CONTENT_GROUP_COUNT_KEY, PENDING_INDEX_COUNT_KEY, TORRENT_COUNT_KEY, VERIFICATION_QUEUE_COUNT_KEY, content_group_key, content_group_prefix, content_member_key, content_members_prefix, decode_content_member_info_hash, decode_pending_content_key, decode_verification_lease, decode_verification_task, pending_index_key, - pending_index_prefix, rejected_metadata_key, torrent_key, verification_lease_key, - verification_lease_prefix, verification_locator_key, verification_task_key, - verification_task_prefix, + pending_index_prefix, rejected_metadata_key, torrent_key, torrent_prefix, + verification_lease_key, verification_lease_prefix, verification_locator_key, + verification_task_key, verification_task_prefix, }, repository::{ - ContentGroupTask, ContentVariants, StorageError, TorrentRepository, UpsertOutcome, - VerificationEnqueueOutcome, VerificationPriority, VerificationRequest, + ContentGroupTask, ContentVariants, IndexInventory, StorageError, TorrentRepository, + UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, VerificationRequest, }, }; @@ -42,6 +46,21 @@ pub struct RocksTorrentRepository { rejection_rule_id: [u8; 32], content_filter: Arc, content_filter_changed: bool, + inventory: InventoryState, +} + +#[derive(Default)] +struct InventoryState { + stored_torrents: AtomicU64, + searchable_groups: AtomicU64, + pending_documents: AtomicU64, +} + +#[derive(Debug, Clone, Copy, Default)] +struct InventoryDelta { + stored_torrents: u64, + searchable_groups: u64, + pending_documents: i64, } impl RocksTorrentRepository { @@ -80,8 +99,10 @@ impl RocksTorrentRepository { batch: &mut WriteBatch, content_key: &[u8; 32], inserted: bool, - ) -> Result<(), StorageError> { - let mut state = self.group_state(content_key)?.unwrap_or(ContentGroupState { + ) -> Result { + let current = self.group_state(content_key)?; + let pending = self.db.get(pending_index_key(content_key))?.is_some(); + let mut state = current.unwrap_or(ContentGroupState { revision: 0, member_count: 0, }); @@ -91,6 +112,102 @@ impl RocksTorrentRepository { } batch.put(content_group_key(content_key), Self::encode_group(state)?); batch.put(pending_index_key(content_key), state.revision.to_be_bytes()); + Ok(InventoryDelta { + searchable_groups: u64::from(current.is_none()), + pending_documents: i64::from(!pending), + ..InventoryDelta::default() + }) + } + + fn inventory_snapshot(&self) -> IndexInventory { + IndexInventory { + stored_torrents: self.inventory.stored_torrents.load(Ordering::Relaxed), + searchable_groups: self.inventory.searchable_groups.load(Ordering::Relaxed), + pending_documents: self.inventory.pending_documents.load(Ordering::Relaxed), + } + } + + fn commit_inventory_batch( + &self, + mut batch: WriteBatch, + delta: InventoryDelta, + ) -> Result<(), StorageError> { + let current = self.inventory_snapshot(); + let next = IndexInventory { + stored_torrents: current + .stored_torrents + .saturating_add(delta.stored_torrents), + searchable_groups: current + .searchable_groups + .saturating_add(delta.searchable_groups), + pending_documents: if delta.pending_documents >= 0 { + current + .pending_documents + .saturating_add(delta.pending_documents as u64) + } else { + current + .pending_documents + .saturating_sub(delta.pending_documents.unsigned_abs()) + }, + }; + batch.put(TORRENT_COUNT_KEY, next.stored_torrents.to_be_bytes()); + batch.put( + CONTENT_GROUP_COUNT_KEY, + next.searchable_groups.to_be_bytes(), + ); + batch.put( + PENDING_INDEX_COUNT_KEY, + next.pending_documents.to_be_bytes(), + ); + self.db.write(batch)?; + self.set_inventory(next); + Ok(()) + } + + fn set_inventory(&self, inventory: IndexInventory) { + self.inventory + .stored_torrents + .store(inventory.stored_torrents, Ordering::Relaxed); + self.inventory + .searchable_groups + .store(inventory.searchable_groups, Ordering::Relaxed); + self.inventory + .pending_documents + .store(inventory.pending_documents, Ordering::Relaxed); + } + + fn initialize_inventory(&self, force_scan: bool) -> Result<(), StorageError> { + if !force_scan + && let (Some(stored), Some(groups), Some(pending)) = ( + read_counter(&self.db, TORRENT_COUNT_KEY)?, + read_counter(&self.db, CONTENT_GROUP_COUNT_KEY)?, + read_counter(&self.db, PENDING_INDEX_COUNT_KEY)?, + ) + { + self.set_inventory(IndexInventory { + stored_torrents: stored, + searchable_groups: groups, + pending_documents: pending, + }); + return Ok(()); + } + let inventory = IndexInventory { + stored_torrents: count_prefix(&self.db, torrent_prefix())?, + searchable_groups: count_prefix(&self.db, content_group_prefix())?, + pending_documents: count_prefix(&self.db, pending_index_prefix())?, + }; + let mut batch = WriteBatch::default(); + batch.put(TORRENT_COUNT_KEY, inventory.stored_torrents.to_be_bytes()); + batch.put( + CONTENT_GROUP_COUNT_KEY, + inventory.searchable_groups.to_be_bytes(), + ); + batch.put( + PENDING_INDEX_COUNT_KEY, + inventory.pending_documents.to_be_bytes(), + ); + self.db.write(batch)?; + self.set_inventory(inventory); Ok(()) } @@ -174,10 +291,12 @@ impl TorrentRepository for RocksTorrentRepository { current.observe_again(observation.last_seen, &observation.source_peers); let mut batch = WriteBatch::default(); batch.put(torrent_key(current.info_hash), Self::encode(¤t)?); - if current.searchable { - self.dirty_group(&mut batch, ¤t.content_key, false)?; - } - self.db.write(batch)?; + let delta = if current.searchable { + self.dirty_group(&mut batch, ¤t.content_key, false)? + } else { + InventoryDelta::default() + }; + self.commit_inventory_batch(batch, delta)?; return Ok(UpsertOutcome::Updated { seen_count: current.seen_count, }); @@ -189,14 +308,20 @@ impl TorrentRepository for RocksTorrentRepository { Self::encode(&observation)?, ); batch.delete(rejected_metadata_key(observation.info_hash)); + let mut delta = InventoryDelta { + stored_torrents: 1, + ..InventoryDelta::default() + }; if observation.searchable { batch.put( content_member_key(&observation.content_key, observation.info_hash), [], ); - self.dirty_group(&mut batch, &observation.content_key, true)?; + let group_delta = self.dirty_group(&mut batch, &observation.content_key, true)?; + delta.searchable_groups = group_delta.searchable_groups; + delta.pending_documents = group_delta.pending_documents; } - self.db.write(batch)?; + self.commit_inventory_batch(batch, delta)?; Ok(UpsertOutcome::Inserted) } @@ -254,10 +379,12 @@ impl TorrentRepository for RocksTorrentRepository { record.observe_again(observed_at, &[]); let mut batch = WriteBatch::default(); batch.put(torrent_key(info_hash), Self::encode(&record)?); - if record.searchable { - self.dirty_group(&mut batch, &record.content_key, false)?; - } - self.db.write(batch)?; + let delta = if record.searchable { + self.dirty_group(&mut batch, &record.content_key, false)? + } else { + InventoryDelta::default() + }; + self.commit_inventory_batch(batch, delta)?; Ok(true) } @@ -314,10 +441,17 @@ impl TorrentRepository for RocksTorrentRepository { unknown.push(*info_hash); } if updated > 0 { + let mut delta = InventoryDelta::default(); for content_key in dirty_groups { - self.dirty_group(&mut batch, &content_key, false)?; + let group_delta = self.dirty_group(&mut batch, &content_key, false)?; + delta.searchable_groups = delta + .searchable_groups + .saturating_add(group_delta.searchable_groups); + delta.pending_documents = delta + .pending_documents + .saturating_add(group_delta.pending_documents); } - self.db.write(batch)?; + self.commit_inventory_batch(batch, delta)?; } Ok(unknown) } @@ -397,7 +531,13 @@ impl TorrentRepository for RocksTorrentRepository { let mut batch = WriteBatch::default(); batch.put(content_group_key(content_key), Self::encode_group(state)?); batch.delete(pending_index_key(content_key)); - self.db.write(batch)?; + self.commit_inventory_batch( + batch, + InventoryDelta { + pending_documents: -1, + ..InventoryDelta::default() + }, + )?; Ok(true) } @@ -439,6 +579,16 @@ impl TorrentRepository for RocksTorrentRepository { if batch_len > 0 { self.db.write(batch)?; } + let inventory = self.inventory_snapshot(); + let mut batch = WriteBatch::default(); + batch.put( + PENDING_INDEX_COUNT_KEY, + inventory.searchable_groups.to_be_bytes(), + ); + self.db.write(batch)?; + self.inventory + .pending_documents + .store(inventory.searchable_groups, Ordering::Relaxed); Ok(total) } @@ -615,21 +765,52 @@ impl TorrentRepository for RocksTorrentRepository { let count = self.verification_queue_len_inner()?.saturating_sub(1); let mut batch = WriteBatch::default(); batch.put(torrent_key(info_hash), Self::encode(&record)?); - if record.searchable { - self.dirty_group(&mut batch, &record.content_key, false)?; - } + let delta = if record.searchable { + self.dirty_group(&mut batch, &record.content_key, false)? + } else { + InventoryDelta::default() + }; if let Some(queued_key) = queued_key { batch.delete(queued_key); } batch.delete(locator_key); batch.put(VERIFICATION_QUEUE_COUNT_KEY, (count as u64).to_be_bytes()); - self.db.write(batch)?; + self.commit_inventory_batch(batch, delta)?; Ok(()) } fn verification_queue_len(&self) -> Result { self.verification_queue_len_inner() } + + fn index_inventory(&self) -> IndexInventory { + self.inventory_snapshot() + } +} + +fn read_counter(db: &DB, key: &[u8]) -> Result, StorageError> { + db.get(key)? + .map(|value| { + value + .as_slice() + .try_into() + .map(u64::from_be_bytes) + .map_err(|_| StorageError::CorruptContentGroup) + }) + .transpose() +} + +fn count_prefix(db: &DB, prefix: [u8; N]) -> Result { + let iterator = db.iterator(IteratorMode::From(&prefix, Direction::Forward)); + let mut count = 0_u64; + for entry in iterator { + let (key, _) = entry?; + if !key.starts_with(&prefix) { + break; + } + count = count.saturating_add(1); + } + Ok(count) } #[cfg(test)] diff --git a/src/search/src/storage/rocks/lifecycle.rs b/src/search/src/storage/rocks/lifecycle.rs index 931a863..88f6064 100644 --- a/src/search/src/storage/rocks/lifecycle.rs +++ b/src/search/src/storage/rocks/lifecycle.rs @@ -54,9 +54,11 @@ impl RocksTorrentRepository { rejection_rule_id, content_filter, content_filter_changed: false, + inventory: Default::default(), }; repository.initialize_format()?; repository.content_filter_changed = repository.synchronize_content_filter()?; + repository.initialize_inventory(repository.content_filter_changed)?; Ok(repository) } diff --git a/src/search/src/storage/rocks/tests.rs b/src/search/src/storage/rocks/tests.rs index 79fdce2..475c72d 100644 --- a/src/search/src/storage/rocks/tests.rs +++ b/src/search/src/storage/rocks/tests.rs @@ -47,6 +47,35 @@ fn record_survives_close_and_reopen() { assert_eq!(repository.get(expected.info_hash).unwrap(), Some(expected)); } +#[test] +fn index_inventory_is_exact_and_survives_reopen() { + let directory = TempDir::new().unwrap(); + { + let repository = RocksTorrentRepository::open(directory.path()).unwrap(); + repository.upsert(test_record(1, 10)).unwrap(); + repository.upsert(test_record(2, 20)).unwrap(); + assert_eq!( + repository.index_inventory(), + IndexInventory { + stored_torrents: 2, + searchable_groups: 1, + pending_documents: 1, + } + ); + let task = repository.pending_index(1).unwrap()[0]; + assert!( + repository + .mark_indexed(&task.content_key, task.revision) + .unwrap() + ); + assert_eq!(repository.index_inventory().pending_documents, 0); + } + let reopened = RocksTorrentRepository::open(directory.path()).unwrap(); + assert_eq!(reopened.index_inventory().stored_torrents, 2); + assert_eq!(reopened.index_inventory().searchable_groups, 1); + assert_eq!(reopened.index_inventory().pending_documents, 0); +} + #[test] fn checkpoint_is_a_consistent_snapshot_and_can_be_opened_read_only() { let directory = TempDir::new().unwrap(); diff --git a/src/web/src/pages/DiagnosticsPage.vue b/src/web/src/pages/DiagnosticsPage.vue index ee56f23..7a2d1c9 100644 --- a/src/web/src/pages/DiagnosticsPage.vue +++ b/src/web/src/pages/DiagnosticsPage.vue @@ -168,6 +168,24 @@ function diskStateLabel(state: ServiceStats['disk_state']): string { return '只读保护' } +function indexStateLabel(state: ServiceStats['index']['state']): string { + if (state === 'ready') return '已同步' + if (state === 'initializing') return '初始化中' + if (state === 'rebuilding') return '重建中' + if (state === 'switching') return '切换中' + if (state === 'blocked') return '已暂停' + return '失败' +} + +function indexReasonLabel(reason: ServiceStats['index']['reason']): string { + if (reason === 'content_filter') return '内容过滤规则变化' + if (reason === 'schema') return '索引结构变化' + if (reason === 'document_format') return '索引文档格式变化' + if (reason === 'missing') return '索引缺失' + if (reason === 'corrupt') return '索引损坏' + return '首次构建' +} + async function loadStats() { if (statsRequest) return statsRequest = true @@ -210,8 +228,23 @@ onBeforeUnmount(() => {
DHT 节点{{ stats.nodes.toLocaleString() }}
Metadata 下载中{{ stats.metadata_in_flight.toLocaleString() }}
本次新收录{{ stats.persistence_inserted.toLocaleString() }}
-
已索引内容{{ stats.indexed_documents.toLocaleString() }}
+
搜索索引{{ stats.index.active_documents.toLocaleString() }}{{ indexStateLabel(stats.index.state) }}
+ +
+
+

{{ indexStateLabel(stats.index.state) }}

{{ indexReasonLabel(stats.index.reason) }} · {{ stats.index.state === 'initializing' ? '当前结果可能不完整' : '当前索引继续提供搜索' }}

+ {{ stats.index.progress_percent.toFixed(1) }}% +
+
+
+ 已保存种子 {{ stats.index.stored_torrents.toLocaleString() }} + 可搜索内容 {{ stats.index.searchable_groups.toLocaleString() }} + 影子索引 {{ stats.index.building_documents.toLocaleString() }} + 待处理 {{ stats.index.pending_documents.toLocaleString() }} +
+

{{ stats.index.error }}

+
diff --git a/src/web/src/pages/SearchPage.vue b/src/web/src/pages/SearchPage.vue index c7970d0..e29f6e1 100644 --- a/src/web/src/pages/SearchPage.vue +++ b/src/web/src/pages/SearchPage.vue @@ -1,14 +1,15 @@ @@ -190,6 +216,11 @@ onBeforeUnmount(() => { + + + {{ indexNotice }} + +
正在搜索

无法完成搜索

{{ error }}

diff --git a/src/web/src/types/api.ts b/src/web/src/types/api.ts index bfb7467..d04c5c4 100644 --- a/src/web/src/types/api.ts +++ b/src/web/src/types/api.ts @@ -132,11 +132,30 @@ export interface ServiceStats { persistence_updated: number persistence_queue: number indexed_documents: number + index: IndexStatus verification_queue: number verification_succeeded: number verification_failed: number } +export type IndexState = 'ready' | 'initializing' | 'rebuilding' | 'switching' | 'blocked' | 'failed' +export type IndexRebuildReason = 'initial' | 'content_filter' | 'schema' | 'document_format' | 'missing' | 'corrupt' + +export interface IndexStatus { + state: IndexState + reason: IndexRebuildReason | null + stored_torrents: number + searchable_groups: number + active_documents: number + building_documents: number + target_documents: number + pending_documents: number + progress_percent: number + started_at: number | null + last_completed_at: number | null + error: string | null +} + export interface ProcessDiagnostics { resident_memory_bytes: number | null private_memory_bytes: number | null