From e74ebf3a9544eb47ac0e8dbdec4e6e4a37ddad79 Mon Sep 17 00:00:00 2001 From: chuan Date: Tue, 11 Aug 2026 09:49:12 +0800 Subject: [PATCH] feat(search): add live content filtering and contextual file order --- README.md | 2 +- TODOS.md | 12 +- config.toml | 57 +-- src/search/README.md | 14 +- src/search/src/api/handlers.rs | 13 +- src/search/src/api/mod.rs | 29 +- src/search/src/api/request.rs | 5 + src/search/src/api/response.rs | 12 +- src/search/src/app.rs | 16 +- src/search/src/config.rs | 12 +- src/search/src/config/model.rs | 63 +-- src/search/src/config/service.rs | 22 + src/search/src/crawler/pipeline.rs | 8 +- src/search/src/diagnostics/mod.rs | 15 +- src/search/src/diagnostics/model.rs | 2 + src/search/src/domain/content_filter.rs | 436 ++++++------------ src/search/src/domain/mod.rs | 5 +- src/search/src/domain/torrent.rs | 9 + src/search/src/filter_worker.rs | 430 +++++++++++++++++ src/search/src/index_worker.rs | 12 +- src/search/src/lib.rs | 2 + src/search/src/search/document.rs | 2 +- src/search/src/search/file_order.rs | 219 +++++++++ src/search/src/search/indexer.rs | 72 ++- src/search/src/search/mod.rs | 2 + src/search/src/search/runtime.rs | 16 +- src/search/src/storage/keys.rs | 25 +- src/search/src/storage/mod.rs | 6 +- src/search/src/storage/repository.rs | 56 ++- src/search/src/storage/rocks.rs | 193 +++++--- .../src/storage/rocks/filter_migration.rs | 368 +++++++++------ src/search/src/storage/rocks/lifecycle.rs | 13 +- src/search/src/storage/rocks/tests.rs | 193 ++++---- src/search/tests/storage_search_flow.rs | 21 +- .../src/components/TorrentDetailDialog.vue | 2 +- src/web/src/lib/api.ts | 5 +- src/web/src/pages/ConfigPage.vue | 16 +- src/web/src/pages/DiagnosticsPage.vue | 41 +- src/web/src/pages/SearchPage.vue | 27 +- src/web/src/types/api.ts | 20 +- 40 files changed, 1671 insertions(+), 802 deletions(-) create mode 100644 src/search/src/filter_worker.rs create mode 100644 src/search/src/search/file_order.rs diff --git a/README.md b/README.md index 57675b7..2e5f10e 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ opencodes/ 不参与构建且不得修改的参考项目 RocksDB 是唯一权威数据源 Tantivy 索引可以从 RocksDB 完整重建 -Tantivy 使用代际影子索引完成全量重建 过滤规则或索引文档结构变化时旧索引继续提供搜索 新索引完整校验后通过原子指针切换 重建状态和进度可以在搜索提示与系统诊断页查看 +Tantivy 使用代际影子索引处理 Schema 文档格式损坏和缺失等全量重建 用户过滤规则通过可恢复扫描只增量更新真正变化的搜索文档 两类进度都可以在搜索提示与系统诊断页查看 应用全部配置和内容隐藏规则统一位于 [`config.toml`](config.toml) diff --git a/TODOS.md b/TODOS.md index 669f4d0..c52bdd6 100644 --- a/TODOS.md +++ b/TODOS.md @@ -182,6 +182,7 @@ - [x] 使用 Bun Vue TypeScript Vite Tailwind CSS 和 shadcn-vue 建立 Web 基础环境 - [x] 实现简单现代并适配移动端的单页搜索界面 - [x] 接入搜索排序分页详情内容变体和磁力链接复制 +- [x] 详情文件分页优先展示匹配搜索条件的文件并继承外层大小或名称排序 - [x] 实现名称别名和文件路径的有限状态自动机正则搜索 - [x] 根据输入语法自动识别普通文本通配符和正则表达式并移除独立模式开关 - [x] 品牌入口可清除搜索查询排序分页和详情状态并返回主页 @@ -231,10 +232,12 @@ - [x] `/stats` 返回验证队列发现握手成功失败和拒绝指标 - [ ] 统计真实数据的 infohash 重复率和内容重复率 -- [x] 在统一 `config.toml` 中定义文件名和文件路径隐藏规则 -- [x] 使用文件名和文件路径双文本框按行管理不区分大小写的通配符规则 -- [x] 保留 RocksDB 原始文件列表并为详情统计搜索和内容聚合生成有效内容视图 -- [x] 使用规则指纹在配置变化时重算内容组并从 RocksDB 重建 Tantivy +- [x] 在统一 `config.toml` 中定义种子标题和内部文件隐藏规则 +- [x] 使用标题与内部文件双文本框按行管理不区分大小写的通配符和 `regex:` 规则 +- [x] 保留 RocksDB 原始文件列表并将用户过滤从内容指纹和内容组身份中解耦 +- [x] 使用内容组过滤投影哈希只增量更新真正变化的 Tantivy 文档 +- [x] 持久化过滤扫描游标并支持连续修改采用最新规则和跨重启恢复 +- [x] 在搜索页和系统诊断页展示过滤基线扫描更新提交和失败状态 - [x] 默认隐藏 BitComet padding 文件以及 `.pad` 和 `.____padding_file` 填充目录 - [x] 全部文件被隐藏的 Metadata 只保留原始记录且不进入公开索引 - [ ] 根据真实垃圾数据决定是否增加种子名称扩展名和大小准入规则 @@ -361,6 +364,7 @@ - [x] 将配置拆分为可序列化 DTO TOML 读取适配器和运行时解析结果 - [x] 使用独立 SQLite 建立有界运行诊断历史存储 - [x] 采集进程 RocksDB Tantivy DHT 队列和磁盘资源快照 +- [x] 在系统诊断中持续展示已保存种子总量并以历史趋势替代低价值 HTTP 请求图表 - [x] 提供当前诊断快照和原始或分钟历史查询接口 - [x] 将 HTTP 请求并发客户端错误服务端错误和延迟分布写入诊断历史 - [x] 增加配置查询完整校验原子保存并发修订和统一重启提示 diff --git a/config.toml b/config.toml index e7b19c7..322a8e0 100644 --- a/config.toml +++ b/config.toml @@ -1,32 +1,17 @@ -# 定义 dht-search 的全部运行配置 - data_dir = "data" persistence_queue_capacity = 8192 stats_interval_secs = 10 index_batch_size = 1024 index_interval_millis = 5000 -[backup] -enabled = false -directory = "data/backups" -interval_secs = 21600 -retain_checkpoints = 3 -create_on_start = true - -[diagnostics] -enabled = true -database = "data/diagnostics.sqlite3" -sample_interval_secs = 10 -raw_retention_hours = 24 -minute_retention_days = 30 -queue_capacity = 128 - -[logging] -directory = "data/logs" -file_enabled = true -rotation = "daily" -retain_files = 7 -file_prefix = "dht-search" +[content_filter] +torrent_name_patterns = ["*【加QQ *】*"] +file_patterns = [ + "*_____padding_file_*", + "*.pad/*", + "*.____padding_file/*", + "*如无法下载扫码福利基地*", +] [metadata_limits] max_metadata_bytes = 10485760 @@ -35,10 +20,6 @@ max_name_bytes = 1024 max_path_bytes = 4096 max_path_depth = 64 -[content_filter] -file_name_patterns = ["*_____padding_file_*"] -file_path_patterns = ["*.pad/*", "*.____padding_file/*"] - [dht] enabled = false port = 12313 @@ -61,6 +42,28 @@ find_node_queries_per_second = 10 find_node_max_in_flight = 100 new_destinations_per_minute = 12000 +[backup] +enabled = false +directory = "data/backups" +interval_secs = 21600 +retain_checkpoints = 3 +create_on_start = true + +[diagnostics] +enabled = true +database = "data/diagnostics.sqlite3" +sample_interval_secs = 10 +raw_retention_hours = 24 +minute_retention_days = 30 +queue_capacity = 128 + +[logging] +directory = "data/logs" +file_enabled = true +rotation = "daily" +retain_files = 7 +file_prefix = "dht-search" + [http] listen = "127.0.0.1:8080" web_dir = "src/web/dist" diff --git a/src/search/README.md b/src/search/README.md index 7517357..a5196e5 100644 --- a/src/search/README.md +++ b/src/search/README.md @@ -194,15 +194,17 @@ Invoke-RestMethod http://127.0.0.1:8080/stats | ConvertTo-Json -Depth 5 ### 无效文件隐藏规则 -`config.toml` 的 `content_filter` 区域控制哪些文件不参与详情展示 搜索 文件数量 有效大小和内容聚合 默认规则会隐藏 BitComet `_____padding_file_` 文件以及 `.pad` 和 `.____padding_file` 填充目录 +`config.toml` 的 `content_filter` 区域控制哪些种子标题和内部文件不参与详情与搜索 默认内部规则会隐藏 BitComet `_____padding_file_` 文件以及 `.pad` 和 `.____padding_file` 填充目录 -RocksDB 始终保存完整原始 Metadata 隐藏规则不会删除文件或种子 修改或回滚规则后应用会根据规则指纹重新计算内容组并从 RocksDB 重建 Tantivy +RocksDB 始终保存完整原始 Metadata 用户隐藏规则不会删除文件或种子也不会改变 `content_key` 和内容组成员关系 标题命中或全部内部文件被隐藏时只隐藏对应 infohash 版本 同组其他版本仍可展示 -配置只包含 `file_name_patterns` 和 `file_path_patterns` 两组不区分大小写的通配符 Web 使用左右两个多行文本框编辑并按行切分 空行和重复规则自动忽略 +配置只包含 `torrent_name_patterns` 和 `file_patterns` 两组规则 内部文件规则同时检查 basename 和规范化完整路径 Web 使用两个多行文本框编辑并按行切分 空行和重复规则自动忽略 -通配符中 `*` 表示任意长度字符 `?` 表示一个字符并匹配完整字段 文件路径在匹配前统一使用 `/` 分隔符 +每行默认是不区分大小写的通配符 `*` 表示任意长度字符 `?` 表示一个字符并匹配完整字段 使用 `regex:` 前缀可以编写 Rust `regex` 语法的正则表达式 -如果一个 Metadata 的全部文件都被隐藏 原始记录仍保留在 RocksDB 但不会进入搜索索引或公开详情 +规则保存后详情立即使用最新投影 后台扫描内容组并以投影哈希只提交真正变化的 Tantivy 文档 扫描游标和待删除任务持久化且允许连续保存时自动收敛到最新规则 `/stats.filter` 与 Web 系统页展示进度 + +内部文件隐藏后列表只返回可见文件但总大小和原始文件数量保持 Metadata 原值 `visible_file_count` 单独用于详情分页 ### 采样去重和 Peer 查找 @@ -284,7 +286,7 @@ GET /torrents/{infohash}?file_offset=0&file_limit=100 RocksDB 是权威数据源而 Tantivy 是可重建索引 -当过滤规则 Tantivy Schema 或索引文档格式变化时 应用在独立代际目录构建影子索引 旧索引继续提供搜索且新收录内容在切换后统一可见 影子索引清空 RocksDB 待索引状态并通过文档数校验后原子更新活动指针 +当 Tantivy Schema 或索引文档格式变化以及索引缺失损坏时 应用在独立代际目录构建影子索引 旧索引继续提供搜索且新收录内容在切换后统一可见 影子索引清空 RocksDB 待索引状态并通过文档数校验后原子更新活动指针 影子索引通过内部构建清单跨重启恢复 构建失败磁盘保护或进程退出不会删除活动索引 没有旧索引时会提供正在初始化的部分结果并明确标记结果尚不完整 切换成功后立即清理旧索引 Windows 文件占用造成的清理失败只记录警告而不影响服务 diff --git a/src/search/src/api/handlers.rs b/src/search/src/api/handlers.rs index 7cac743..8c1b0e6 100644 --- a/src/search/src/api/handlers.rs +++ b/src/search/src/api/handlers.rs @@ -145,6 +145,7 @@ pub(crate) async fn stats(State(state): State) -> Json persistence_queue: persistence.queue_depth, indexed_documents: state.search.num_docs(), index, + filter: state.filter.status(), 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), @@ -236,6 +237,7 @@ pub(crate) async fn config_update( State(state): State, Json(request): Json, ) -> Result, ApiError> { + let previous = state.config.snapshot(); let config = state.config.clone(); let snapshot = tokio::task::spawn_blocking(move || config.update(request)) .await @@ -245,6 +247,12 @@ pub(crate) async fn config_update( ConfigServiceError::Validation(_) => ApiError::bad_request(error.to_string()), ConfigServiceError::Persistence(_) => ApiError::internal(error.to_string()), })?; + if snapshot.config.content_filter != previous.config.content_filter { + let filter = + crate::domain::ContentFilter::compile(snapshot.config.content_filter.to_domain()) + .map_err(|error| ApiError::bad_request(error.to_string()))?; + state.filter.update(std::sync::Arc::new(filter)); + } Ok(Json(snapshot)) } @@ -288,7 +296,7 @@ pub(crate) async fn search( }; let content_key = if let Some(info_hash) = content_key { let repository = state.repository.clone(); - let record = tokio::task::spawn_blocking(move || repository.get(info_hash)) + let record = tokio::task::spawn_blocking(move || repository.get_visible(info_hash)) .await .map_err(|error| ApiError::internal(error.to_string()))? .map_err(|error| ApiError::internal(error.to_string()))?; @@ -403,6 +411,9 @@ pub(crate) async fn torrent( .map_err(|error| ApiError::internal(error.to_string()))? .map_err(|error| ApiError::internal(error.to_string()))? .ok_or_else(|| ApiError::not_found("没有找到该 infohash"))?; + let mut record = record; + record.files = crate::search::order_files(record.files, &request.q, request.mode, request.sort) + .map_err(|error| ApiError::bad_request(error.to_string()))?; if let Some(verification) = &verification { verification .enqueue(vec![info_hash], VerificationPriority::High) diff --git a/src/search/src/api/mod.rs b/src/search/src/api/mod.rs index 8984f68..2c58568 100644 --- a/src/search/src/api/mod.rs +++ b/src/search/src/api/mod.rs @@ -35,6 +35,7 @@ pub(crate) struct ApiState { pub(crate) backup_stats: BackupStats, pub(crate) diagnostics: DiagnosticsHandle, pub(crate) config: ConfigService, + pub(crate) filter: crate::filter_worker::FilterRuntime, pub(crate) http_stats: HttpStats, } @@ -178,6 +179,7 @@ mod tests { Vec::new(), ) .unwrap(), + filter: crate::filter_worker::FilterRuntime::for_test(repository.clone()), http_stats: HttpStats::default(), }, web_dir, @@ -226,6 +228,7 @@ mod tests { assert_eq!(json["index"]["state"], "ready"); assert_eq!(json["index"]["active_documents"], 1); assert_eq!(json["index"]["pending_documents"], 0); + assert_eq!(json["filter"]["state"], "ready"); assert_eq!(json["http_active_requests"], 1); assert!( json["http_requests"] @@ -295,8 +298,7 @@ mod tests { .unwrap(); let original_revision = config_snapshot["revision"].as_str().unwrap().to_owned(); let mut invalid_filter = config_snapshot["config"].clone(); - invalid_filter["content_filter"]["file_name_patterns"][0] = - serde_json::json!("x".repeat(1_025)); + invalid_filter["content_filter"]["file_patterns"][0] = serde_json::json!("x".repeat(1_025)); let response = app .clone() .oneshot( @@ -498,13 +500,34 @@ mod tests { .unwrap() .starts_with("magnet:?xt=") ); - assert_eq!(json["files"][0]["path"], "movie.mkv"); + assert_eq!(json["files"][0]["path"], "extras/1.txt"); assert_eq!(json["files"].as_array().unwrap().len(), 100); assert_eq!(json["file_count"], 205); + assert_eq!(json["visible_file_count"], 205); assert_eq!(json["file_offset"], 0); assert_eq!(json["file_limit"], 100); assert_eq!(repository.verification_queue_len().unwrap(), 1); + let response = app + .clone() + .oneshot( + Request::builder() + .uri(format!( + "/torrents/{}?q=movie&sort=latest&file_limit=1", + InfoHash::from_bytes([1; 20]) + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let json: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(json["files"][0]["path"], "movie.mkv"); + assert_eq!(json["file_limit"], 1); + let response = app .clone() .oneshot( diff --git a/src/search/src/api/request.rs b/src/search/src/api/request.rs index 7c57388..f9a67a6 100644 --- a/src/search/src/api/request.rs +++ b/src/search/src/api/request.rs @@ -53,6 +53,11 @@ pub(crate) struct TorrentRequest { pub(crate) file_offset: usize, #[serde(default = "default_file_limit")] pub(crate) file_limit: usize, + #[serde(default)] + pub(crate) q: String, + #[serde(default)] + pub(crate) mode: SearchMode, + pub(crate) sort: Option, } #[derive(Debug, Deserialize)] diff --git a/src/search/src/api/response.rs b/src/search/src/api/response.rs index f202db2..933fb3a 100644 --- a/src/search/src/api/response.rs +++ b/src/search/src/api/response.rs @@ -2,6 +2,7 @@ use crate::{ domain::{Availability, Heat, TorrentFile, TorrentRecord}, + filter_worker::FilterStatus, search::IndexStatus, }; use serde::Serialize; @@ -109,6 +110,7 @@ pub(crate) struct StatsResponse { pub(crate) persistence_queue: usize, pub(crate) indexed_documents: u64, pub(crate) index: IndexStatus, + pub(crate) filter: FilterStatus, pub(crate) verification_queue: u64, pub(crate) verification_accepted: u64, pub(crate) verification_deduplicated: u64, @@ -127,6 +129,7 @@ pub(crate) struct TorrentResponse { pub(crate) name: String, pub(crate) total_size: u64, pub(crate) file_count: usize, + pub(crate) visible_file_count: usize, pub(crate) file_offset: usize, pub(crate) file_limit: usize, pub(crate) files: Vec, @@ -166,12 +169,13 @@ impl From for TorrentVariantResponse { fn from(record: TorrentRecord) -> Self { let info_hash = record.info_hash.to_string(); let heat = record.heat(unix_timestamp()); + let file_count = record.original_file_count().min(usize::MAX as u64) as usize; Self { magnet_link: format!("magnet:?xt=urn:btih:{info_hash}"), info_hash, name: record.name, total_size: record.total_size, - file_count: record.files.len(), + file_count, first_seen: record.first_seen, last_seen: record.last_seen, seen_count: record.seen_count, @@ -189,8 +193,9 @@ impl TorrentResponse { ) -> Self { let info_hash = record.info_hash.to_string(); let heat = record.heat(unix_timestamp()); - let file_count = record.files.len(); - let file_offset = file_offset.min(file_count); + let file_count = record.original_file_count().min(usize::MAX as u64) as usize; + let visible_file_count = record.files.len(); + let file_offset = file_offset.min(visible_file_count); let files = record .files .into_iter() @@ -203,6 +208,7 @@ impl TorrentResponse { name: record.name, total_size: record.total_size, file_count, + visible_file_count, file_offset, file_limit, files, diff --git a/src/search/src/app.rs b/src/search/src/app.rs index 9dd1675..f8d9201 100644 --- a/src/search/src/app.rs +++ b/src/search/src/app.rs @@ -16,6 +16,7 @@ use crate::{ diagnostics::{DiagnosticSources, DiagnosticsRuntime, HttpStats}, disk_guard::{self, DiskGuard}, error::AppError, + filter_worker::FilterRuntime, index_worker, monitor, shutdown, }; @@ -29,12 +30,18 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res let repository = Arc::new(RocksTorrentRepository::open_with_rules( &database_path, metadata_limits.rule_id(), - content_filter, + content_filter.clone(), )?); - let search_bootstrap = - SearchRuntime::open(&config.data_dir, repository.content_filter_changed())?; + let search_bootstrap = SearchRuntime::open(&config.data_dir)?; let search = search_bootstrap.runtime.clone(); disk_guard.probe(&config.data_dir, 0); + let filter_cancel = CancellationToken::new(); + let (filter_runtime, filter_task) = FilterRuntime::start( + repository.clone(), + content_filter, + disk_guard.clone(), + filter_cancel.clone(), + ); let repository_api: Arc = repository.clone(); let mut persistence = PersistencePipeline::start( repository_api, @@ -138,6 +145,7 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res backup_stats, diagnostics: diagnostics.handle(), config: config_service, + filter: filter_runtime, http_stats, }, api_cancel.clone(), @@ -167,6 +175,8 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res let mut shutdown_error = None; diagnostics.request_shutdown(); + filter_cancel.cancel(); + let _ = filter_task.await; backup_cancel.cancel(); crawler.shutdown().await; if let Some(task) = backup_task { diff --git a/src/search/src/config.rs b/src/search/src/config.rs index 449cde4..59599a4 100644 --- a/src/search/src/config.rs +++ b/src/search/src/config.rs @@ -239,29 +239,25 @@ mod tests { #[test] fn invalid_embedded_content_filter_is_rejected() { let mut dto = AppConfigDto::default(); - dto.content_filter.file_name_patterns[0] = "x".repeat(1_025); + dto.content_filter.file_patterns[0] = "x".repeat(1_025); assert!(matches!(resolve(dto), Err(AppError::ContentFilter(_)))); } #[test] fn content_filter_patterns_ignore_blank_duplicate_and_case() { let mut dto = AppConfigDto::default(); - dto.content_filter.file_name_patterns = vec![ + dto.content_filter.file_patterns = vec![ String::new(), "*PADDING_FILE*".to_owned(), "*padding_file*".to_owned(), ]; let config = resolve(dto).unwrap(); - assert_eq!(config.content_filter.to_domain().file_rules.len(), 3); + assert_eq!(config.content_filter.to_domain().file_patterns.len(), 1); let filter = config.content_filter().unwrap(); - assert!(filter.is_hidden(&crate::domain::TorrentFile { + assert!(filter.is_file_hidden(&crate::domain::TorrentFile { path: "release/Padding_File_1".to_owned(), size: 1, })); - assert!(filter.is_hidden(&crate::domain::TorrentFile { - path: "release/.pad/1".to_owned(), - size: 1, - })); } #[test] diff --git a/src/search/src/config/model.rs b/src/search/src/config/model.rs index 6c83611..ccaa04e 100644 --- a/src/search/src/config/model.rs +++ b/src/search/src/config/model.rs @@ -4,9 +4,7 @@ use std::{net::SocketAddr, path::PathBuf}; use serde::{Deserialize, Serialize}; -use crate::domain::{ - ContentFilterConfig, FileFilterRule, FileMatchField, FileMatchKind, FileRuleAction, -}; +use crate::domain::ContentFilterConfig; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(default, deny_unknown_fields)] @@ -29,25 +27,15 @@ pub(crate) struct AppConfigDto { #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(default, deny_unknown_fields)] pub(crate) struct ContentFilterConfigDto { - pub(crate) file_name_patterns: Vec, - pub(crate) file_path_patterns: Vec, + pub(crate) torrent_name_patterns: Vec, + pub(crate) file_patterns: Vec, } impl ContentFilterConfigDto { pub(crate) fn to_domain(&self) -> ContentFilterConfig { - let mut file_rules = pattern_rules( - "file-name", - FileMatchField::FileName, - &self.file_name_patterns, - ); - file_rules.extend(pattern_rules( - "file-path", - FileMatchField::FilePath, - &self.file_path_patterns, - )); ContentFilterConfig { - version: 1, - file_rules, + torrent_name_patterns: normalize_patterns(&self.torrent_name_patterns), + file_patterns: normalize_patterns(&self.file_patterns), } } } @@ -177,46 +165,25 @@ impl Default for AppConfigDto { fn default_content_filter() -> ContentFilterConfigDto { ContentFilterConfigDto { - file_name_patterns: vec!["*_____padding_file_*".to_owned()], - file_path_patterns: vec!["*.pad/*".to_owned(), "*.____padding_file/*".to_owned()], + torrent_name_patterns: vec!["*【加QQ *】*".to_owned()], + file_patterns: vec![ + "*_____padding_file_*".to_owned(), + "*.pad/*".to_owned(), + "*.____padding_file/*".to_owned(), + ], } } -fn pattern_rules( - id_prefix: &str, - field: FileMatchField, - patterns: &[String], -) -> Vec { +fn normalize_patterns(patterns: &[String]) -> Vec { let mut patterns: Vec<_> = patterns .iter() .map(|pattern| pattern.trim()) .filter(|pattern| !pattern.is_empty()) - .map(|pattern| { - if field == FileMatchField::FilePath { - pattern.replace('\\', "/").to_lowercase() - } else { - pattern.to_lowercase() - } - }) + .map(ToString::to_string) .collect(); - patterns.sort_unstable(); - patterns.dedup(); + patterns.sort_unstable_by_key(|pattern| pattern.to_lowercase()); + patterns.dedup_by(|left, right| left.eq_ignore_ascii_case(right)); patterns - .into_iter() - .map(|value| { - let fingerprint = blake3::hash(format!("{id_prefix}\0{value}").as_bytes()); - FileFilterRule { - id: format!("{id_prefix}-{fingerprint}"), - enabled: true, - field, - match_kind: FileMatchKind::Wildcard, - value, - case_sensitive: false, - action: FileRuleAction::Hide, - reason: String::new(), - } - }) - .collect() } impl Default for MetadataLimitsConfig { diff --git a/src/search/src/config/service.rs b/src/search/src/config/service.rs index 1e5033c..10359c1 100644 --- a/src/search/src/config/service.rs +++ b/src/search/src/config/service.rs @@ -164,6 +164,7 @@ impl ConfigService { fn restart_required(current: &AppConfigDto, startup: &AppConfigDto) -> bool { let mut comparable = current.clone(); comparable.dht.enabled = startup.dht.enabled; + comparable.content_filter = startup.content_filter.clone(); comparable != *startup } @@ -223,6 +224,27 @@ mod tests { assert!(!stored.dht.enabled); } + #[test] + fn content_filter_update_is_live_and_does_not_require_restart() { + let directory = TempDir::new().unwrap(); + let service = service(&directory); + let current = service.snapshot(); + let mut config = current.config; + config + .content_filter + .torrent_name_patterns + .push("*广告*".into()); + + let updated = service + .update(ConfigUpdateRequest { + revision: current.revision, + config, + }) + .unwrap(); + + assert!(!updated.restart_required); + } + #[test] fn stale_revision_cannot_overwrite_a_newer_update() { let directory = TempDir::new().unwrap(); diff --git a/src/search/src/crawler/pipeline.rs b/src/search/src/crawler/pipeline.rs index 8a1af79..b7535ec 100644 --- a/src/search/src/crawler/pipeline.rs +++ b/src/search/src/crawler/pipeline.rs @@ -297,7 +297,13 @@ mod tests { Ok(None) } - fn mark_indexed(&self, _: &[u8; 32], _: u64) -> Result { + fn mark_indexed( + &self, + _: &[u8; 32], + _: u64, + _: [u8; 32], + _: bool, + ) -> Result { Ok(true) } diff --git a/src/search/src/diagnostics/mod.rs b/src/search/src/diagnostics/mod.rs index d2eec38..d588900 100644 --- a/src/search/src/diagnostics/mod.rs +++ b/src/search/src/diagnostics/mod.rs @@ -18,7 +18,7 @@ use std::{ use crate::{ search::SearchRuntime, - storage::{RocksTorrentRepository, StorageDiagnostics as RocksDiagnostics}, + storage::{RocksTorrentRepository, StorageDiagnostics as RocksDiagnostics, TorrentRepository}, }; use tokio_util::sync::CancellationToken; @@ -269,12 +269,14 @@ fn collect_sample(sources: &DiagnosticSources, session_started_at: u64) -> Diagn tracing::warn!(%error, "读取 RocksDB 诊断属性失败"); RocksDiagnostics::default() }); + let inventory = sources.repository.index_inventory(); let search = sources.search.diagnostics(); DiagnosticSample { captured_at: unix_timestamp(), session_started_at, process: process::snapshot(), storage: StorageDiagnostics { + stored_torrents: Some(inventory.stored_torrents), block_cache_bytes: storage.block_cache_bytes, memtable_bytes: storage.memtable_bytes, pending_compaction_bytes: storage.pending_compaction_bytes, @@ -341,6 +343,9 @@ mod tests { let directory = TempDir::new().unwrap(); let repository = Arc::new(RocksTorrentRepository::open(directory.path().join("rocksdb")).unwrap()); + repository + .upsert(crate::domain::test_record(1, unix_timestamp())) + .unwrap(); let repository_trait: Arc = repository.clone(); let disk_guard = DiskGuard::new(); let persistence = PersistencePipeline::start( @@ -378,7 +383,13 @@ mod tests { let current = handle.current(); assert!(current.status.enabled); assert_eq!(current.status.write_failures, 0); - assert!(current.sample.is_some()); + assert_eq!( + current + .sample + .as_ref() + .and_then(|sample| sample.storage.stored_torrents), + Some(1) + ); let now = unix_timestamp(); let history = handle .history(HistoryResolution::Raw, now.saturating_sub(5), now) diff --git a/src/search/src/diagnostics/model.rs b/src/search/src/diagnostics/model.rs index 0ef3816..c0eda37 100644 --- a/src/search/src/diagnostics/model.rs +++ b/src/search/src/diagnostics/model.rs @@ -25,6 +25,8 @@ pub(crate) struct ProcessDiagnostics { #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] pub(crate) struct StorageDiagnostics { + #[serde(default)] + pub(crate) stored_torrents: Option, pub(crate) block_cache_bytes: Option, pub(crate) memtable_bytes: Option, pub(crate) pending_compaction_bytes: Option, diff --git a/src/search/src/domain/content_filter.rs b/src/search/src/domain/content_filter.rs index bca67e2..9dc1c4e 100644 --- a/src/search/src/domain/content_filter.rs +++ b/src/search/src/domain/content_filter.rs @@ -1,274 +1,173 @@ -// 负责定义可配置的无效文件识别规则和面向用户的有效内容视图 - -use std::collections::HashSet; +// 负责定义附属内容过滤规则和不修改权威记录的可见投影 use regex::{Regex, RegexBuilder}; use serde::{Deserialize, Serialize}; use unicode_normalization::UnicodeNormalization; -use super::{TorrentFile, TorrentRecord, TorrentRecordError, content_key}; +use super::{TorrentFile, TorrentRecord}; -const FILTER_FORMAT_VERSION: u64 = 1; const MAX_RULES: usize = 256; const MAX_PATTERN_BYTES: usize = 1_024; +const REGEX_PREFIX: &str = "regex:"; +const DEFAULT_FILE_PATTERNS: [&str; 3] = + ["*_____padding_file_*", "*.pad/*", "*.____padding_file/*"]; -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] pub struct ContentFilterConfig { - pub version: u64, - #[serde(default)] - pub file_rules: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct FileFilterRule { - pub id: String, - #[serde(default = "default_true")] - pub enabled: bool, - pub field: FileMatchField, - #[serde(rename = "match")] - pub match_kind: FileMatchKind, - pub value: String, - #[serde(default)] - pub case_sensitive: bool, - pub action: FileRuleAction, - pub reason: String, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "kebab-case")] -pub enum FileMatchField { - FileName, - FilePath, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "kebab-case")] -pub enum FileMatchKind { - Exact, - Prefix, - Suffix, - Contains, - Wildcard, - Regex, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "kebab-case")] -pub enum FileRuleAction { - Hide, -} - -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct FilterOutcome { - pub hidden_files: usize, - pub searchable: bool, + pub torrent_name_patterns: Vec, + pub file_patterns: Vec, } #[derive(Debug, thiserror::Error)] pub enum ContentFilterError { - #[error("内容过滤规则版本 {0} 不受支持")] - UnsupportedVersion(u64), #[error("内容过滤规则数量 {actual} 超过上限 {limit}")] TooManyRules { actual: usize, limit: usize }, - #[error("内容过滤规则 ID 不能为空")] - EmptyRuleId, - #[error("内容过滤规则 ID 重复: {0}")] - DuplicateRuleId(String), - #[error("内容过滤规则 {0} 的匹配值不能为空")] - EmptyPattern(String), - #[error("内容过滤规则 {id} 的匹配值超过 {limit} 字节")] - PatternTooLong { id: String, limit: usize }, - #[error("内容过滤规则 {id} 的正则表达式无效: {source}")] - InvalidRegex { id: String, source: regex::Error }, + #[error("内容过滤规则不能为空")] + EmptyPattern, + #[error("内容过滤规则超过 {limit} 字节")] + PatternTooLong { limit: usize }, + #[error("内容过滤正则表达式无效: {source}")] + InvalidRegex { source: regex::Error }, #[error("内容过滤规则无法生成稳定指纹: {0}")] Fingerprint(serde_json::Error), } #[derive(Debug)] pub struct ContentFilter { + config: ContentFilterConfig, fingerprint: [u8; 32], - rules: Vec, + torrent_name_rules: Vec, + file_rules: Vec, } #[derive(Debug)] -struct CompiledRule { - field: FileMatchField, - case_sensitive: bool, - matcher: CompiledMatcher, -} - -#[derive(Debug)] -enum CompiledMatcher { - Exact(String), - Prefix(String), - Suffix(String), - Contains(String), - Pattern(Regex), +enum CompiledPattern { + Wildcard(Regex), + Regex(Regex), } impl ContentFilter { pub fn compile(config: ContentFilterConfig) -> Result { - if config.version != FILTER_FORMAT_VERSION { - return Err(ContentFilterError::UnsupportedVersion(config.version)); - } - if config.file_rules.len() > MAX_RULES { + let total = config + .torrent_name_patterns + .len() + .saturating_add(config.file_patterns.len()); + if total > MAX_RULES { return Err(ContentFilterError::TooManyRules { - actual: config.file_rules.len(), + actual: total, limit: MAX_RULES, }); } + let torrent_name_rules = compile_patterns(&config.torrent_name_patterns)?; + let file_rules = compile_patterns(&config.file_patterns)?; let fingerprint = *blake3::hash(&serde_json::to_vec(&config).map_err(ContentFilterError::Fingerprint)?) .as_bytes(); - let mut ids = HashSet::with_capacity(config.file_rules.len()); - let mut rules = Vec::new(); - for rule in config.file_rules { - if rule.id.trim().is_empty() { - return Err(ContentFilterError::EmptyRuleId); - } - if !ids.insert(rule.id.clone()) { - return Err(ContentFilterError::DuplicateRuleId(rule.id)); - } - if rule.value.is_empty() { - return Err(ContentFilterError::EmptyPattern(rule.id)); - } - if rule.value.len() > MAX_PATTERN_BYTES { - return Err(ContentFilterError::PatternTooLong { - id: rule.id, - limit: MAX_PATTERN_BYTES, - }); - } - if !rule.enabled { - continue; - } - let mut value = normalize(&rule.value, rule.case_sensitive); - if rule.field == FileMatchField::FilePath && rule.match_kind != FileMatchKind::Regex { - value = value.replace('\\', "/"); - } - let matcher = match rule.match_kind { - FileMatchKind::Exact => CompiledMatcher::Exact(value), - FileMatchKind::Prefix => CompiledMatcher::Prefix(value), - FileMatchKind::Suffix => CompiledMatcher::Suffix(value), - FileMatchKind::Contains => CompiledMatcher::Contains(value), - FileMatchKind::Wildcard => CompiledMatcher::Pattern( - RegexBuilder::new(&glob_regex(&value)) - .case_insensitive(false) - .build() - .map_err(|source| ContentFilterError::InvalidRegex { - id: rule.id.clone(), - source, - })?, - ), - FileMatchKind::Regex => CompiledMatcher::Pattern( - RegexBuilder::new(&rule.value) - .case_insensitive(!rule.case_sensitive) - .build() - .map_err(|source| ContentFilterError::InvalidRegex { - id: rule.id.clone(), - source, - })?, - ), - }; - rules.push(CompiledRule { - field: rule.field, - case_sensitive: rule.case_sensitive || rule.match_kind == FileMatchKind::Regex, - matcher, - }); - } - Ok(Self { fingerprint, rules }) + Ok(Self { + config, + fingerprint, + torrent_name_rules, + file_rules, + }) + } + + pub fn legacy_default() -> Self { + Self::compile(ContentFilterConfig { + torrent_name_patterns: Vec::new(), + file_patterns: DEFAULT_FILE_PATTERNS + .iter() + .map(ToString::to_string) + .collect(), + }) + .expect("默认内容过滤规则必须有效") + } + + pub fn config(&self) -> &ContentFilterConfig { + &self.config } pub fn fingerprint(&self) -> [u8; 32] { self.fingerprint } - pub fn apply_derivatives( - &self, - record: &mut TorrentRecord, - ) -> Result { - let visible: Vec<_> = record - .files - .iter() - .filter(|file| !self.is_hidden(file)) - .cloned() - .collect(); - let hidden_files = record.files.len().saturating_sub(visible.len()); - record.searchable = !visible.is_empty(); - record.content_key = if record.searchable { - content_key(&visible)? - } else { - [0; 32] - }; - Ok(FilterOutcome { - hidden_files, - searchable: record.searchable, - }) - } - pub fn public_record(&self, record: &TorrentRecord) -> Option { - if !record.searchable { + if !record.searchable || self.is_torrent_hidden(&record.name) { return None; } let mut public = record.clone(); - public.files.retain(|file| !self.is_hidden(file)); - if public.files.is_empty() { - return None; - } - public.total_size = public - .files - .iter() - .fold(0_u64, |total, file| total.saturating_add(file.size)); - Some(public) + public.original_file_count = Some(record.files.len() as u64); + public.files.retain(|file| !self.is_file_hidden(file)); + (!public.files.is_empty()).then_some(public) } - pub fn is_hidden(&self, file: &TorrentFile) -> bool { - self.rules.iter().any(|rule| rule.matches(file)) + pub fn is_torrent_hidden(&self, name: &str) -> bool { + matches_any(&self.torrent_name_rules, &normalize(name)) + } + + pub fn is_file_hidden(&self, file: &TorrentFile) -> bool { + let path = normalize(&file.path).replace('\\', "/"); + let name = path.rsplit('/').next().unwrap_or(path.as_str()); + self.file_rules + .iter() + .any(|rule| rule.is_match(name) || rule.is_match(&path)) } } impl Default for ContentFilter { fn default() -> Self { - Self::compile(ContentFilterConfig { - version: FILTER_FORMAT_VERSION, - file_rules: Vec::new(), + Self::legacy_default() + } +} + +impl CompiledPattern { + fn is_match(&self, value: &str) -> bool { + match self { + Self::Wildcard(regex) | Self::Regex(regex) => regex.is_match(value), + } + } +} + +fn compile_patterns(patterns: &[String]) -> Result, ContentFilterError> { + patterns + .iter() + .map(|pattern| { + let pattern = pattern.trim(); + if pattern.is_empty() { + return Err(ContentFilterError::EmptyPattern); + } + if pattern.len() > MAX_PATTERN_BYTES { + return Err(ContentFilterError::PatternTooLong { + limit: MAX_PATTERN_BYTES, + }); + } + if let Some(regex) = pattern.strip_prefix(REGEX_PREFIX) { + if regex.trim().is_empty() { + return Err(ContentFilterError::EmptyPattern); + } + return RegexBuilder::new(regex.trim()) + .case_insensitive(true) + .unicode(true) + .build() + .map(CompiledPattern::Regex) + .map_err(|source| ContentFilterError::InvalidRegex { source }); + } + RegexBuilder::new(&glob_regex(&normalize(pattern))) + .case_insensitive(false) + .unicode(true) + .build() + .map(CompiledPattern::Wildcard) + .map_err(|source| ContentFilterError::InvalidRegex { source }) }) - .expect("空内容过滤规则必须有效") - } + .collect() } -impl CompiledRule { - fn matches(&self, file: &TorrentFile) -> bool { - let target = match self.field { - FileMatchField::FileName => file - .path - .rsplit(['/', '\\']) - .next() - .unwrap_or(file.path.as_str()), - FileMatchField::FilePath => &file.path, - }; - let mut target = normalize(target, self.case_sensitive); - if self.field == FileMatchField::FilePath { - target = target.replace('\\', "/"); - } - match &self.matcher { - CompiledMatcher::Exact(value) => target == *value, - CompiledMatcher::Prefix(value) => target.starts_with(value), - CompiledMatcher::Suffix(value) => target.ends_with(value), - CompiledMatcher::Contains(value) => target.contains(value), - CompiledMatcher::Pattern(pattern) => pattern.is_match(&target), - } - } +fn matches_any(patterns: &[CompiledPattern], value: &str) -> bool { + patterns.iter().any(|pattern| pattern.is_match(value)) } -fn normalize(value: &str, case_sensitive: bool) -> String { - let normalized: String = value.nfkc().collect(); - if case_sensitive { - normalized - } else { - normalized.to_lowercase() - } +fn normalize(value: &str) -> String { + value.nfkc().collect::().to_lowercase() } fn glob_regex(pattern: &str) -> String { @@ -294,115 +193,76 @@ fn glob_regex(pattern: &str) -> String { output } -const fn default_true() -> bool { - true -} - #[cfg(test)] mod tests { use super::*; use crate::domain::test_record; - fn filter(rules: Vec) -> ContentFilter { + fn filter(torrent: &[&str], files: &[&str]) -> ContentFilter { ContentFilter::compile(ContentFilterConfig { - version: 1, - file_rules: rules, + torrent_name_patterns: torrent.iter().map(ToString::to_string).collect(), + file_patterns: files.iter().map(ToString::to_string).collect(), }) .unwrap() } - fn rule(field: FileMatchField, match_kind: FileMatchKind, value: &str) -> FileFilterRule { - FileFilterRule { - id: format!("{field:?}-{match_kind:?}"), - enabled: true, - field, - match_kind, - value: value.into(), - case_sensitive: false, - action: FileRuleAction::Hide, - reason: "测试".into(), - } - } - #[test] - fn prefix_rule_hides_bitcomet_padding_case_insensitively() { - let filter = filter(vec![rule( - FileMatchField::FileName, - FileMatchKind::Prefix, - "_____padding_file_", - )]); - assert!(filter.is_hidden(&TorrentFile { - path: "目录/_____PADDING_FILE_1_请升级____".into(), - size: 16, - })); - } - - #[test] - fn wildcard_matches_the_whole_selected_field() { - let filter = filter(vec![rule( - FileMatchField::FilePath, - FileMatchKind::Wildcard, - ".pad/*", - )]); - assert!(filter.is_hidden(&TorrentFile { - path: ".pad/123".into(), - size: 1, - })); - assert!(!filter.is_hidden(&TorrentFile { - path: "movie.pad/123".into(), + fn wildcard_and_regex_are_case_insensitive() { + let filter = filter(&["*【加QQ *】*"], &[r"regex:(^|/)\.pad/.*"]); + assert!(filter.is_torrent_hidden("电影【加qq 123456】")); + assert!(filter.is_file_hidden(&TorrentFile { + path: "Release/.PAD/1".into(), size: 1, })); } #[test] - fn regex_rule_matches_nested_libtorrent_padding_directory() { - let filter = filter(vec![rule( - FileMatchField::FilePath, - FileMatchKind::Regex, - r"(^|/)\.____padding_file/", - )]); - assert!(filter.is_hidden(&TorrentFile { - path: "release/.____padding_file/47".into(), + fn file_rule_matches_name_or_full_path() { + let filter = filter(&[], &["*padding_file*", "*.pad/*"]); + assert!(filter.is_file_hidden(&TorrentFile { + path: "release/_____padding_file_1".into(), size: 1, })); - assert!(!filter.is_hidden(&TorrentFile { - path: "release/real_padding_file.txt".into(), + assert!(filter.is_file_hidden(&TorrentFile { + path: "release/.pad/1".into(), size: 1, })); } #[test] - fn public_record_keeps_raw_record_unchanged() { - let filter = filter(vec![rule( - FileMatchField::FileName, - FileMatchKind::Prefix, - "_____padding_file_", - )]); + fn public_projection_keeps_original_statistics() { + let filter = filter(&[], &["*padding*"]); let mut record = test_record(1, 1); record.files.push(TorrentFile { - path: "_____padding_file_1_".into(), + path: "padding.bin".into(), size: 100, }); record.total_size += 100; - filter.apply_derivatives(&mut record).unwrap(); let public = filter.public_record(&record).unwrap(); - assert_eq!(record.files.len(), 2); - assert_eq!(record.total_size, 142); + assert_eq!(public.total_size, 142); + assert_eq!(public.original_file_count(), 2); assert_eq!(public.files.len(), 1); - assert_eq!(public.total_size, 42); } #[test] - fn record_with_only_hidden_files_is_not_searchable() { - let filter = filter(vec![rule( - FileMatchField::FileName, - FileMatchKind::Prefix, - "_____padding_file_", - )]); + fn hidden_title_or_all_hidden_files_remove_only_the_record() { + let title_filter = filter(&["*广告*"], &[]); let mut record = test_record(1, 1); - record.files[0].path = "_____padding_file_1_".into(); - let outcome = filter.apply_derivatives(&mut record).unwrap(); - assert!(!outcome.searchable); - assert!(filter.public_record(&record).is_none()); + record.name = "广告资源".into(); + assert!(title_filter.public_record(&record).is_none()); + + let file_filter = filter(&[], &["*"]); + assert!(file_filter.public_record(&test_record(1, 1)).is_none()); + } + + #[test] + fn invalid_regex_is_rejected() { + assert!(matches!( + ContentFilter::compile(ContentFilterConfig { + torrent_name_patterns: vec!["regex:[".into()], + file_patterns: Vec::new(), + }), + Err(ContentFilterError::InvalidRegex { .. }) + )); } } diff --git a/src/search/src/domain/mod.rs b/src/search/src/domain/mod.rs index a232dde..13326cf 100644 --- a/src/search/src/domain/mod.rs +++ b/src/search/src/domain/mod.rs @@ -7,10 +7,7 @@ mod info_hash; mod metadata; mod torrent; -pub use content_filter::{ - ContentFilter, ContentFilterConfig, ContentFilterError, FileFilterRule, FileMatchField, - FileMatchKind, FileRuleAction, FilterOutcome, -}; +pub use content_filter::{ContentFilter, ContentFilterConfig, ContentFilterError}; pub use content_group::ContentGroup; #[cfg(any(feature = "rocksdb-storage", test))] pub(crate) use content_group::ContentGroupBuilder; diff --git a/src/search/src/domain/torrent.rs b/src/search/src/domain/torrent.rs index 8c83aa7..3ea2c00 100644 --- a/src/search/src/domain/torrent.rs +++ b/src/search/src/domain/torrent.rs @@ -90,6 +90,8 @@ pub struct TorrentRecord { pub activity_score_millis: u64, #[serde(default)] pub activity_updated_at: u64, + #[serde(skip)] + pub(crate) original_file_count: Option, } pub(crate) struct NewTorrentRecord { @@ -121,9 +123,15 @@ impl TorrentRecord { availability: new.availability, activity_score_millis: ACTIVITY_SCALE, activity_updated_at: new.timestamp, + original_file_count: None, } } + pub fn original_file_count(&self) -> u64 { + self.original_file_count + .unwrap_or_else(|| self.files.len().min(u64::MAX as usize) as u64) + } + pub fn try_from_with_limits( info: MetadataCandidate, limits: MetadataLimits, @@ -243,6 +251,7 @@ pub(crate) fn test_record(hash_byte: u8, timestamp: u64) -> TorrentRecord { availability: Availability::default(), activity_score_millis: ACTIVITY_SCALE, activity_updated_at: timestamp, + original_file_count: None, } } diff --git a/src/search/src/filter_worker.rs b/src/search/src/filter_worker.rs new file mode 100644 index 0000000..fccd47b --- /dev/null +++ b/src/search/src/filter_worker.rs @@ -0,0 +1,430 @@ +// 负责热更新附属过滤规则并协调可恢复的增量内容组扫描 + +use std::{ + sync::{Arc, RwLock}, + time::Duration, +}; + +use serde::Serialize; +use tokio::sync::watch; +use tokio_util::sync::CancellationToken; + +use crate::{ + disk_guard::DiskGuard, + domain::ContentFilter, + storage::{RocksTorrentRepository, TorrentRepository}, +}; + +const SCAN_BATCH_SIZE: usize = 256; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum FilterState { + Baselining, + Scanning, + Applying, + Ready, + Failed, +} + +#[derive(Debug, Clone, Serialize)] +pub(crate) struct FilterStatus { + pub(crate) state: FilterState, + pub(crate) total_groups: u64, + pub(crate) scanned_groups: u64, + pub(crate) changed_groups: u64, + pub(crate) pending_documents: u64, + pub(crate) progress_percent: f64, + pub(crate) error: Option, +} + +#[derive(Clone)] +pub(crate) struct FilterRuntime { + repository: Arc, + target: watch::Sender>, + status: Arc>, +} + +impl FilterRuntime { + #[cfg(test)] + pub(crate) fn for_test(repository: Arc) -> Self { + let filter = Arc::new(ContentFilter::legacy_default()); + repository.set_content_filter(filter.clone()); + let (target, _) = watch::channel(filter); + Self { + repository, + target, + status: Arc::new(RwLock::new(FilterStatus { + state: FilterState::Ready, + total_groups: 0, + scanned_groups: 0, + changed_groups: 0, + pending_documents: 0, + progress_percent: 100.0, + error: None, + })), + } + } + + pub(crate) fn start( + repository: Arc, + target: Arc, + disk_guard: DiskGuard, + cancel: CancellationToken, + ) -> (Self, tokio::task::JoinHandle<()>) { + let (sender, receiver) = watch::channel(target.clone()); + let status = Arc::new(RwLock::new(FilterStatus { + state: FilterState::Baselining, + total_groups: repository.index_inventory().searchable_groups, + scanned_groups: 0, + changed_groups: 0, + pending_documents: 0, + progress_percent: 0.0, + error: None, + })); + let runtime = Self { + repository: repository.clone(), + target: sender, + status: status.clone(), + }; + let task = tokio::spawn(run(repository, receiver, status, disk_guard, cancel)); + (runtime, task) + } + + pub(crate) fn update(&self, filter: Arc) { + self.repository.set_content_filter(filter.clone()); + self.target.send_replace(filter); + } + + pub(crate) fn status(&self) -> FilterStatus { + self.status + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } +} + +async fn run( + repository: Arc, + mut target: watch::Receiver>, + status: Arc>, + disk_guard: DiskGuard, + cancel: CancellationToken, +) { + loop { + if cancel.is_cancelled() { + return; + } + let target_filter = target.borrow().clone(); + let applied = match repository.applied_filter_fingerprint() { + Ok(applied) => applied, + Err(error) => { + set_failed(&status, error.to_string()); + wait_or_cancel(&cancel).await; + continue; + } + }; + if applied.is_none() { + let legacy = Arc::new(ContentFilter::legacy_default()); + if let Err(error) = apply_filter( + repository.clone(), + legacy, + true, + &target, + &status, + &disk_guard, + &cancel, + ) + .await + { + set_failed(&status, error); + wait_or_cancel(&cancel).await; + continue; + } + continue; + } + if applied == Some(target_filter.fingerprint()) { + set_status( + &status, + FilterState::Ready, + repository.index_inventory().searchable_groups, + repository.index_inventory().searchable_groups, + 0, + 0, + None, + ); + tokio::select! { + _ = cancel.cancelled() => return, + changed = target.changed() => { + if changed.is_err() { return; } + } + } + continue; + } + if let Err(error) = apply_filter( + repository.clone(), + target_filter, + false, + &target, + &status, + &disk_guard, + &cancel, + ) + .await + { + set_failed(&status, error); + wait_or_cancel(&cancel).await; + } + } +} + +async fn apply_filter( + repository: Arc, + filter: Arc, + baseline: bool, + target: &watch::Receiver>, + status: &Arc>, + disk_guard: &DiskGuard, + cancel: &CancellationToken, +) -> Result<(), String> { + let fingerprint = filter.fingerprint(); + repository + .prepare_filter_scan(fingerprint) + .map_err(|error| error.to_string())?; + let total = repository.index_inventory().searchable_groups; + let state = if baseline { + FilterState::Baselining + } else { + FilterState::Scanning + }; + loop { + if cancel.is_cancelled() { + return Ok(()); + } + if !baseline && target.borrow().fingerprint() != fingerprint { + return Ok(()); + } + let Some(_permit) = disk_guard.begin_new_write() else { + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + }; + let repository_task = repository.clone(); + let filter_task = filter.clone(); + let batch = tokio::task::spawn_blocking(move || { + repository_task.scan_filter_batch( + filter_task, + baseline, + SCAN_BATCH_SIZE, + unix_timestamp(), + ) + }) + .await + .map_err(|error| error.to_string())? + .map_err(|error| error.to_string())?; + let (scanned, changed) = repository + .filter_scan_counters() + .map_err(|error| error.to_string())?; + let pending = repository + .filter_pending_len() + .map_err(|error| error.to_string())?; + set_status(status, state, total, scanned, changed, pending, None); + if batch.finished { + break; + } + tokio::task::yield_now().await; + } + if !baseline { + loop { + if cancel.is_cancelled() || target.borrow().fingerprint() != fingerprint { + return Ok(()); + } + let pending = repository + .filter_pending_len() + .map_err(|error| error.to_string())?; + let (scanned, changed) = repository + .filter_scan_counters() + .map_err(|error| error.to_string())?; + set_status( + status, + FilterState::Applying, + total, + scanned, + changed, + pending, + None, + ); + if pending == 0 { + break; + } + tokio::select! { + _ = cancel.cancelled() => return Ok(()), + _ = tokio::time::sleep(Duration::from_millis(250)) => {} + } + } + } + repository + .finish_filter_scan(fingerprint) + .map_err(|error| error.to_string())?; + Ok(()) +} + +fn set_failed(status: &Arc>, error: String) { + let mut current = status + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + current.state = FilterState::Failed; + current.error = Some(error); +} + +fn set_status( + status: &Arc>, + state: FilterState, + total: u64, + scanned: u64, + changed: u64, + pending: u64, + error: Option, +) { + let progress_percent = if total == 0 { + 100.0 + } else { + (scanned.min(total) as f64 / total as f64 * 100.0).clamp(0.0, 100.0) + }; + *status + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = FilterStatus { + state, + total_groups: total, + scanned_groups: scanned, + changed_groups: changed, + pending_documents: pending, + progress_percent, + error, + }; +} + +async fn wait_or_cancel(cancel: &CancellationToken) { + tokio::select! { + _ = cancel.cancelled() => {}, + _ = tokio::time::sleep(Duration::from_secs(1)) => {}, + } +} + +fn unix_timestamp() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + domain::{ContentFilterConfig, test_record}, + search::SearchEngine, + storage::TorrentRepository, + }; + use tempfile::TempDir; + + #[tokio::test] + async fn live_filter_deletes_and_restores_only_the_changed_document() { + let directory = TempDir::new().unwrap(); + let repository = + Arc::new(RocksTorrentRepository::open(directory.path().join("rocksdb")).unwrap()); + let mut record = test_record(1, 10); + record.name = "需要隐藏的标题".into(); + let content_key = record.content_key; + repository.upsert(record.clone()).unwrap(); + let search = SearchEngine::open(directory.path().join("tantivy")).unwrap(); + search.index_pending(repository.as_ref(), 10, 20).unwrap(); + assert_eq!(search.search("需要隐藏", 0, 10).unwrap().total, 1); + + let cancel = CancellationToken::new(); + let (runtime, task) = FilterRuntime::start( + repository.clone(), + Arc::new(ContentFilter::legacy_default()), + DiskGuard::new(), + cancel.clone(), + ); + let legacy_fingerprint = ContentFilter::legacy_default().fingerprint(); + wait_until_ready( + &runtime, + &search, + repository.as_ref(), + 1, + legacy_fingerprint, + ) + .await; + let hidden = Arc::new( + ContentFilter::compile(ContentFilterConfig { + torrent_name_patterns: vec!["*需要隐藏*".into()], + file_patterns: Vec::new(), + }) + .unwrap(), + ); + runtime.update(hidden.clone()); + wait_until_ready( + &runtime, + &search, + repository.as_ref(), + 0, + hidden.fingerprint(), + ) + .await; + assert_eq!(search.search("需要隐藏", 0, 10).unwrap().total, 0); + assert_eq!( + repository + .get(record.info_hash) + .unwrap() + .unwrap() + .content_key, + content_key + ); + + let visible = Arc::new(ContentFilter::legacy_default()); + runtime.update(visible.clone()); + wait_until_ready( + &runtime, + &search, + repository.as_ref(), + 1, + visible.fingerprint(), + ) + .await; + assert_eq!(search.search("需要隐藏", 0, 10).unwrap().total, 1); + + runtime.update(hidden); + runtime.update(visible.clone()); + wait_until_ready( + &runtime, + &search, + repository.as_ref(), + 1, + visible.fingerprint(), + ) + .await; + cancel.cancel(); + task.await.unwrap(); + } + + async fn wait_until_ready( + runtime: &FilterRuntime, + search: &SearchEngine, + repository: &RocksTorrentRepository, + expected_documents: u64, + expected_fingerprint: [u8; 32], + ) { + for _ in 0..200 { + search.index_pending(repository, 10, 30).unwrap(); + if runtime.status().state == FilterState::Ready + && search.num_docs() == expected_documents + && repository.applied_filter_fingerprint().unwrap() == Some(expected_fingerprint) + { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("filter runtime did not become ready"); + } +} diff --git a/src/search/src/index_worker.rs b/src/search/src/index_worker.rs index 6a02959..6458ee2 100644 --- a/src/search/src/index_worker.rs +++ b/src/search/src/index_worker.rs @@ -132,13 +132,16 @@ async fn rebuild_index( 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 { + let expected_documents = repository + .visible_group_count(unix_timestamp()) + .map_err(|error| error.to_string())?; + if target.num_docs() != expected_documents { consistency_retries = consistency_retries.saturating_add(1); if consistency_retries > 1 { return Err(format!( "影子索引文档数不一致 indexed={} expected={}", target.num_docs(), - inventory.searchable_groups + expected_documents )); } let rebuild_repository = repository.clone(); @@ -153,10 +156,7 @@ async fn rebuild_index( runtime .finish_rebuild(session, target) .map_err(|error| error.to_string())?; - tracing::info!( - documents = inventory.searchable_groups, - "影子搜索索引已原子切换" - ); + tracing::info!(documents = expected_documents, "影子搜索索引已原子切换"); return Ok(true); } tokio::task::yield_now().await; diff --git a/src/search/src/lib.rs b/src/search/src/lib.rs index b1b2330..8bfc47c 100644 --- a/src/search/src/lib.rs +++ b/src/search/src/lib.rs @@ -20,6 +20,8 @@ mod entry; #[cfg(feature = "rocksdb-storage")] mod error; #[cfg(feature = "rocksdb-storage")] +mod filter_worker; +#[cfg(feature = "rocksdb-storage")] mod index_worker; #[cfg(feature = "rocksdb-storage")] mod monitor; diff --git a/src/search/src/search/document.rs b/src/search/src/search/document.rs index 3a7df9f..c58996b 100644 --- a/src/search/src/search/document.rs +++ b/src/search/src/search/document.rs @@ -71,7 +71,7 @@ pub(crate) fn from_group(group: &ContentGroup, fields: SearchFields) -> TantivyD document.add_text(fields.extensions, extension); } document.add_u64(fields.total_size, record.total_size); - document.add_u64(fields.file_count, record.files.len() as u64); + document.add_u64(fields.file_count, record.original_file_count()); document.add_u64(fields.first_seen, group.first_seen); document.add_u64(fields.last_seen, group.last_seen); document.add_u64(fields.seen_count, group.seen_count); diff --git a/src/search/src/search/file_order.rs b/src/search/src/search/file_order.rs new file mode 100644 index 0000000..6727e71 --- /dev/null +++ b/src/search/src/search/file_order.rs @@ -0,0 +1,219 @@ +// 负责根据搜索上下文对详情文件执行相关性优先且分页稳定的排序 + +use std::cmp::Ordering; + +use regex::{Regex, RegexBuilder}; +use unicode_normalization::UnicodeNormalization; + +use crate::domain::TorrentFile; + +use super::{SearchMode, SearchSort}; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum FileOrderError { + #[error("文件相关性正则表达式无效: {0}")] + InvalidRegex(#[from] regex::Error), +} + +pub(crate) fn order_files( + files: Vec, + query: &str, + mode: SearchMode, + sort: Option, +) -> Result, FileOrderError> { + let matcher = FileMatcher::compile(query, mode)?; + let secondary = SecondaryOrder::from_search_sort(sort); + let mut ranked: Vec<_> = files + .into_iter() + .map(|file| RankedFile::new(file, &matcher)) + .collect(); + ranked.sort_unstable_by(|left, right| { + right + .relevance + .cmp(&left.relevance) + .then_with(|| secondary.compare(left, right)) + .then_with(|| left.name.cmp(&right.name)) + .then_with(|| left.path.cmp(&right.path)) + .then_with(|| left.file.size.cmp(&right.file.size)) + }); + Ok(ranked.into_iter().map(|ranked| ranked.file).collect()) +} + +struct RankedFile { + file: TorrentFile, + name: String, + path: String, + relevance: u8, +} + +impl RankedFile { + fn new(file: TorrentFile, matcher: &FileMatcher) -> Self { + let path = normalize(&file.path).replace('\\', "/"); + let name = path.rsplit('/').next().unwrap_or(path.as_str()).to_owned(); + let relevance = if matcher.is_match(&name) { + 2 + } else if matcher.is_match(&path) { + 1 + } else { + 0 + }; + Self { + file, + name, + path, + relevance, + } + } +} + +enum FileMatcher { + None, + Plain(Vec), + Pattern(Regex), +} + +impl FileMatcher { + fn compile(query: &str, mode: SearchMode) -> Result { + let query = query.trim(); + if query.is_empty() || query == "*" { + return Ok(Self::None); + } + if mode == SearchMode::Regex { + return RegexBuilder::new(&query.to_lowercase()) + .case_insensitive(true) + .unicode(true) + .build() + .map(Self::Pattern) + .map_err(FileOrderError::from); + } + if query.contains('*') || query.contains('?') { + return RegexBuilder::new(&wildcard_regex(&normalize(query))) + .unicode(true) + .build() + .map(Self::Pattern) + .map_err(FileOrderError::from); + } + let terms = normalize(query) + .split_whitespace() + .map(ToOwned::to_owned) + .collect(); + Ok(Self::Plain(terms)) + } + + fn is_match(&self, value: &str) -> bool { + match self { + Self::None => false, + Self::Plain(terms) => terms.iter().all(|term| value.contains(term)), + Self::Pattern(regex) => regex.is_match(value), + } + } +} + +#[derive(Clone, Copy)] +enum SecondaryOrder { + Name, + SizeAscending, + SizeDescending, +} + +impl SecondaryOrder { + fn from_search_sort(sort: Option) -> Self { + match sort { + Some(SearchSort::SizeAsc) => Self::SizeAscending, + Some(SearchSort::SizeDesc) => Self::SizeDescending, + _ => Self::Name, + } + } + + fn compare(self, left: &RankedFile, right: &RankedFile) -> Ordering { + match self { + Self::Name => left.name.cmp(&right.name), + Self::SizeAscending => left.file.size.cmp(&right.file.size), + Self::SizeDescending => right.file.size.cmp(&left.file.size), + } + } +} + +fn normalize(value: &str) -> String { + value.nfkc().collect::().to_lowercase() +} + +fn wildcard_regex(pattern: &str) -> String { + let mut output = String::from("^"); + let mut escaped = false; + for character in pattern.chars() { + if escaped { + output.push_str(®ex::escape(&character.to_string())); + escaped = false; + } else { + match character { + '\\' => escaped = true, + '*' => output.push_str(".*"), + '?' => output.push('.'), + _ => output.push_str(®ex::escape(&character.to_string())), + } + } + } + if escaped { + output.push_str(r"\\"); + } + output.push('$'); + output +} + +#[cfg(test)] +mod tests { + use super::*; + + fn file(path: &str, size: u64) -> TorrentFile { + TorrentFile { + path: path.to_owned(), + size, + } + } + + #[test] + fn related_file_names_and_paths_are_promoted_before_name_order() { + let files = vec![ + file("z/readme.txt", 1), + file("ubuntu/docs/a.txt", 2), + file("a/ubuntu.iso", 3), + ]; + let ordered = order_files(files, "ubuntu", SearchMode::Text, None).unwrap(); + assert_eq!(ordered[0].path, "a/ubuntu.iso"); + assert_eq!(ordered[1].path, "ubuntu/docs/a.txt"); + assert_eq!(ordered[2].path, "z/readme.txt"); + } + + #[test] + fn size_sort_is_secondary_to_relevance() { + let files = vec![ + file("unrelated-large.bin", 1_000), + file("match-small.iso", 1), + file("match-large.iso", 10), + file("unrelated-small.bin", 2), + ]; + let ordered = + order_files(files, "match", SearchMode::Text, Some(SearchSort::SizeDesc)).unwrap(); + assert_eq!(ordered[0].path, "match-large.iso"); + assert_eq!(ordered[1].path, "match-small.iso"); + assert_eq!(ordered[2].path, "unrelated-large.bin"); + assert_eq!(ordered[3].path, "unrelated-small.bin"); + } + + #[test] + fn wildcard_and_regex_follow_search_mode() { + let files = vec![file("a/movie.mkv", 1), file("b/movie.iso", 2)]; + let wildcard = order_files( + files.clone(), + "*.iso", + SearchMode::Text, + Some(SearchSort::SizeAsc), + ) + .unwrap(); + assert_eq!(wildcard[0].path, "b/movie.iso"); + + let regex = order_files(files, r"movie\.mkv$", SearchMode::Regex, None).unwrap(); + assert_eq!(regex[0].path, "a/movie.mkv"); + } +} diff --git a/src/search/src/search/indexer.rs b/src/search/src/search/indexer.rs index 27d8c06..42686fc 100644 --- a/src/search/src/search/indexer.rs +++ b/src/search/src/search/indexer.rs @@ -9,8 +9,9 @@ use std::{ time::Instant, }; +#[cfg(test)] use crate::domain::ContentGroup; -use crate::storage::TorrentRepository; +use crate::storage::{IndexDocument, TorrentRepository}; use tantivy::{ DocAddress, Index, IndexReader, IndexWriter, Order, ReloadPolicy, Searcher, TantivyDocument, Term, @@ -158,8 +159,8 @@ impl SearchEngine { }) } - fn index_groups(&self, groups: &[ContentGroup]) -> Result<(), SearchError> { - if groups.is_empty() { + fn index_documents(&self, documents: &[IndexDocument]) -> Result<(), SearchError> { + if documents.is_empty() { return Ok(()); } let started = Instant::now(); @@ -178,12 +179,14 @@ impl SearchEngine { ); } let writer = writer.as_mut().expect("search writer was initialized"); - for group in groups { + for document in documents { writer.delete_term(Term::from_field_text( fields.content_key, - &hex::encode(group.content_key), + &hex::encode(document.content_key), )); - writer.add_document(super::document::from_group(group, fields))?; + if let Some(group) = &document.group { + writer.add_document(super::document::from_group(group, fields))?; + } } writer.commit()?; self.inner.reader.reload()?; @@ -199,7 +202,7 @@ impl SearchEngine { .last_commit_at .store(unix_timestamp(), AtomicOrdering::Relaxed); self.inner.last_commit_documents.store( - groups.len().min(u64::MAX as usize) as u64, + documents.len().min(u64::MAX as usize) as u64, AtomicOrdering::Relaxed, ); } else { @@ -210,6 +213,20 @@ impl SearchEngine { result } + #[cfg(test)] + fn index_groups(&self, groups: &[ContentGroup]) -> Result<(), SearchError> { + let documents: Vec<_> = groups + .iter() + .cloned() + .map(|group| IndexDocument { + content_key: group.content_key, + projection_hash: crate::storage::projection_hash(Some(&group)), + group: Some(group), + }) + .collect(); + self.index_documents(&documents) + } + pub fn num_docs(&self) -> u64 { self.inner.reader.searcher().num_docs() } @@ -251,17 +268,20 @@ impl SearchEngine { now: u64, ) -> Result { let tasks = repository.pending_index(limit)?; - let mut groups = Vec::with_capacity(tasks.len()); + let mut documents = Vec::with_capacity(tasks.len()); for task in &tasks { - if let Some(group) = repository.content_group(&task.content_key, now)? { - groups.push(group); - } + documents.push(repository.index_document(&task.content_key, now)?); } - self.index_groups(&groups)?; - for task in &tasks { - repository.mark_indexed(&task.content_key, task.revision)?; + self.index_documents(&documents)?; + for (task, document) in tasks.iter().zip(&documents) { + repository.mark_indexed( + &task.content_key, + task.revision, + document.projection_hash, + document.group.is_some(), + )?; } - Ok(groups.len()) + Ok(documents.len()) } pub fn search( @@ -434,6 +454,7 @@ mod tests { availability: crate::domain::Availability::default(), activity_score_millis: 1_000, activity_updated_at: 20, + original_file_count: None, } } @@ -463,7 +484,7 @@ mod tests { let directory = TempDir::new().unwrap(); let engine = SearchEngine::open(directory.path()).unwrap(); let mut record = record(); - index_records(&engine, &[record.clone()]); + index_records(&engine, std::slice::from_ref(&record)); record.seen_count = 4; index_records(&engine, &[record]); @@ -473,6 +494,25 @@ mod tests { assert_eq!(page.hits[0].seen_count, 4); } + #[test] + fn delete_only_document_removes_a_hidden_group() { + let directory = TempDir::new().unwrap(); + let engine = SearchEngine::open(directory.path()).unwrap(); + let record = record(); + index_records(&engine, std::slice::from_ref(&record)); + assert_eq!(engine.search("ubuntu", 0, 10).unwrap().total, 1); + + engine + .index_documents(&[IndexDocument { + content_key: record.content_key, + group: None, + projection_hash: [9; 32], + }]) + .unwrap(); + + assert_eq!(engine.search("ubuntu", 0, 10).unwrap().total, 0); + } + #[test] fn writer_is_created_only_when_a_document_is_committed() { let directory = TempDir::new().unwrap(); diff --git a/src/search/src/search/mod.rs b/src/search/src/search/mod.rs index 02ee73f..99a0001 100644 --- a/src/search/src/search/mod.rs +++ b/src/search/src/search/mod.rs @@ -1,12 +1,14 @@ // 负责暴露全文搜索抽象并隐藏 Tantivy 的具体实现细节 mod document; +mod file_order; mod filter; mod indexer; mod query; mod runtime; mod schema; +pub(crate) use file_order::order_files; pub use indexer::{SearchDiagnostics, SearchEngine}; pub use query::{ AvailabilitySummary, SearchHit, SearchMode, SearchOptions, SearchPage, SearchSort, diff --git a/src/search/src/search/runtime.rs b/src/search/src/search/runtime.rs index a403d06..f075016 100644 --- a/src/search/src/search/runtime.rs +++ b/src/search/src/search/runtime.rs @@ -36,7 +36,6 @@ pub enum IndexState { #[serde(rename_all = "snake_case")] pub enum IndexRebuildReason { Initial, - ContentFilter, Schema, DocumentFormat, Missing, @@ -133,10 +132,7 @@ impl SearchRuntime { ) } - pub(crate) fn open( - data_dir: &Path, - content_filter_changed: bool, - ) -> Result { + pub(crate) fn open(data_dir: &Path) -> Result { let managed_root = data_dir.join(MANAGED_DIRECTORY); let generations = managed_root.join(GENERATIONS_DIRECTORY); let legacy_path = data_dir.join("tantivy"); @@ -164,9 +160,7 @@ impl SearchRuntime { .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() { + let reason = if active_path.is_none() { Some(IndexRebuildReason::Initial) } else if active.is_none() { Some(IndexRebuildReason::Corrupt) @@ -695,7 +689,7 @@ mod tests { legacy.index_pending(&repository, 10, 20).unwrap(); drop(legacy); - let bootstrap = SearchRuntime::open(directory.path(), false).unwrap(); + let bootstrap = SearchRuntime::open(directory.path()).unwrap(); assert_eq!(bootstrap.runtime.num_docs(), 1); assert_eq!( bootstrap @@ -755,7 +749,7 @@ mod tests { legacy.index_pending(&repository, 10, 20).unwrap(); drop(legacy); - let first = SearchRuntime::open(directory.path(), false).unwrap(); + let first = SearchRuntime::open(directory.path()).unwrap(); let mut session = first.rebuild.unwrap(); let total = repository.prepare_full_reindex().unwrap(); first.runtime.mark_prepared(&mut session, total).unwrap(); @@ -764,7 +758,7 @@ mod tests { drop(first.target); drop(first.runtime); - let resumed = SearchRuntime::open(directory.path(), false).unwrap(); + let resumed = SearchRuntime::open(directory.path()).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/storage/keys.rs b/src/search/src/storage/keys.rs index 62e553c..c3d6666 100644 --- a/src/search/src/storage/keys.rs +++ b/src/search/src/storage/keys.rs @@ -5,8 +5,11 @@ use crate::domain::InfoHash; pub(crate) const DATABASE_FORMAT_KEY: &[u8] = b"\x00database-format"; 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 FILTER_APPLIED_FINGERPRINT_KEY: &[u8] = b"\x00filter-applied-fingerprint"; +pub(crate) const FILTER_TARGET_FINGERPRINT_KEY: &[u8] = b"\x00filter-target-fingerprint"; +pub(crate) const FILTER_SCAN_CURSOR_KEY: &[u8] = b"\x00filter-scan-cursor"; +pub(crate) const FILTER_SCAN_SCANNED_KEY: &[u8] = b"\x00filter-scan-scanned"; +pub(crate) const FILTER_SCAN_CHANGED_KEY: &[u8] = b"\x00filter-scan-changed"; 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"; @@ -19,6 +22,8 @@ const VERIFICATION_HIGH_PREFIX: u8 = b'h'; const VERIFICATION_NORMAL_PREFIX: u8 = b'n'; const VERIFICATION_LEASE_PREFIX: u8 = b'l'; const VERIFICATION_LOCATOR_PREFIX: u8 = b'v'; +const FILTER_PROJECTION_PREFIX: u8 = b'i'; +const FILTER_PENDING_PREFIX: u8 = b'f'; pub(crate) fn torrent_key(info_hash: InfoHash) -> [u8; 1 + InfoHash::BYTE_LEN] { prefixed_info_hash(TORRENT_PREFIX, info_hash) @@ -48,6 +53,18 @@ pub(crate) fn pending_index_prefix() -> [u8; 1] { [PENDING_INDEX_PREFIX] } +pub(crate) fn filter_projection_key(content_key: &[u8; 32]) -> [u8; 1 + 32] { + prefixed_content_key(FILTER_PROJECTION_PREFIX, content_key) +} + +pub(crate) fn filter_pending_key(content_key: &[u8; 32]) -> [u8; 1 + 32] { + prefixed_content_key(FILTER_PENDING_PREFIX, content_key) +} + +pub(crate) fn filter_pending_prefix() -> [u8; 1] { + [FILTER_PENDING_PREFIX] +} + pub(crate) fn verification_task_key( high_priority: bool, requested_at: u64, @@ -129,10 +146,6 @@ pub(crate) fn content_members_prefix(content_key: &[u8; 32]) -> [u8; 1 + 32] { key } -pub(crate) fn content_member_prefix() -> [u8; 1] { - [CONTENT_PREFIX] -} - pub(crate) fn decode_content_member_info_hash( key: &[u8], content_key: &[u8; 32], diff --git a/src/search/src/storage/mod.rs b/src/search/src/storage/mod.rs index 70e85d3..3958dec 100644 --- a/src/search/src/storage/mod.rs +++ b/src/search/src/storage/mod.rs @@ -6,9 +6,11 @@ mod repository; #[cfg(feature = "rocksdb-storage")] mod rocks; +#[cfg(test)] +pub(crate) use repository::projection_hash; pub use repository::{ - CheckpointSummary, ContentGroupTask, ContentVariants, IndexInventory, StorageDiagnostics, - StorageError, TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome, + CheckpointSummary, ContentGroupTask, ContentVariants, IndexDocument, IndexInventory, + StorageDiagnostics, StorageError, TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, VerificationRequest, }; #[cfg(feature = "rocksdb-storage")] diff --git a/src/search/src/storage/repository.rs b/src/search/src/storage/repository.rs index 388f4ea..70d8568 100644 --- a/src/search/src/storage/repository.rs +++ b/src/search/src/storage/repository.rs @@ -46,7 +46,27 @@ pub trait TorrentRepository: Send + Sync { now: u64, ) -> Result, StorageError>; - fn mark_indexed(&self, content_key: &[u8; 32], revision: u64) -> Result; + fn index_document( + &self, + content_key: &[u8; 32], + now: u64, + ) -> Result { + let group = self.content_group(content_key, now)?; + let projection_hash = projection_hash(group.as_ref()); + Ok(IndexDocument { + content_key: *content_key, + group, + projection_hash, + }) + } + + fn mark_indexed( + &self, + content_key: &[u8; 32], + revision: u64, + projection_hash: [u8; 32], + visible: bool, + ) -> Result; fn prepare_full_reindex(&self) -> Result; @@ -97,6 +117,13 @@ pub struct ContentGroupTask { pub revision: u64, } +#[derive(Debug, Clone, PartialEq)] +pub struct IndexDocument { + pub content_key: [u8; 32], + pub group: Option, + pub projection_hash: [u8; 32], +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct ContentVariants { pub total: u64, @@ -142,6 +169,33 @@ pub enum UpsertOutcome { Updated { seen_count: u64 }, } +pub(crate) fn projection_hash(group: Option<&ContentGroup>) -> [u8; 32] { + let mut hasher = blake3::Hasher::new(); + match group { + None => { + hasher.update(b"hidden"); + } + Some(group) => { + hasher.update(b"visible"); + hasher.update(group.representative.info_hash.as_bytes()); + hasher.update(group.representative.name.as_bytes()); + hasher.update(&group.representative.original_file_count().to_be_bytes()); + hasher.update(&group.variant_count.to_be_bytes()); + for alias in &group.aliases { + hasher.update(&(alias.len() as u64).to_be_bytes()); + hasher.update(alias.as_bytes()); + } + for file in &group.representative.files { + hasher.update(&(file.path.len() as u64).to_be_bytes()); + hasher.update(file.path.as_bytes()); + hasher.update(&file.size.to_be_bytes()); + } + hasher.update(b"end"); + } + } + *hasher.finalize().as_bytes() +} + #[derive(Debug, thiserror::Error)] pub enum StorageError { #[cfg(feature = "rocksdb-storage")] diff --git a/src/search/src/storage/rocks.rs b/src/search/src/storage/rocks.rs index 3f1b3e5..06b096c 100644 --- a/src/search/src/storage/rocks.rs +++ b/src/search/src/storage/rocks.rs @@ -1,30 +1,28 @@ // 负责实现 RocksDB 打开配置批量写入精确查询和关闭流程 use std::sync::{ - Arc, Mutex, + Arc, Mutex, RwLock, atomic::{AtomicU64, Ordering}, }; use rocksdb::{DB, Direction, IteratorMode, WriteBatch}; -use crate::domain::{ - ContentFilter, ContentGroupBuilder, InfoHash, RejectedMetadata, TorrentRecord, - VerificationResult, -}; +use crate::domain::{ContentFilter, InfoHash, RejectedMetadata, TorrentRecord, VerificationResult}; 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, torrent_prefix, - verification_lease_key, verification_lease_prefix, verification_locator_key, - verification_task_key, verification_task_prefix, + decode_verification_lease, decode_verification_task, filter_pending_key, + filter_projection_key, pending_index_key, 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, IndexInventory, StorageError, TorrentRepository, - UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, VerificationRequest, + ContentGroupTask, ContentVariants, IndexDocument, IndexInventory, StorageError, + TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, + VerificationRequest, projection_hash, }, }; @@ -40,12 +38,18 @@ struct ContentGroupState { member_count: u64, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ProjectionState { + hash: [u8; 32], + visible: bool, +} + pub struct RocksTorrentRepository { db: DB, write_lock: Mutex<()>, rejection_rule_id: [u8; 32], - content_filter: Arc, - content_filter_changed: bool, + canonical_filter: Arc, + content_filter: RwLock>, inventory: InventoryState, } @@ -64,6 +68,20 @@ struct InventoryDelta { } impl RocksTorrentRepository { + fn current_filter(&self) -> Arc { + self.content_filter + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + pub(crate) fn set_content_filter(&self, filter: Arc) { + *self + .content_filter + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = filter; + } + fn encode(record: &TorrentRecord) -> Result, StorageError> { rmp_serde::to_vec_named(record).map_err(Into::into) } @@ -80,6 +98,16 @@ impl RocksTorrentRepository { rmp_serde::from_slice(bytes).map_err(Into::into) } + fn projection_state( + &self, + content_key: &[u8; 32], + ) -> Result, StorageError> { + self.db + .get(filter_projection_key(content_key))? + .map(|bytes| decode_projection(&bytes)) + .transpose() + } + fn encode_group(state: ContentGroupState) -> Result, StorageError> { rmp_serde::to_vec_named(&state).map_err(Into::into) } @@ -239,32 +267,6 @@ impl RocksTorrentRepository { }; Ok(Some((key, info_hash))) } - - fn content_member_hashes( - &self, - content_key: &[u8; 32], - limit: usize, - ) -> Result, StorageError> { - if limit == 0 { - return Ok(Vec::new()); - } - let prefix = content_members_prefix(content_key); - let iterator = self - .db - .iterator(IteratorMode::From(&prefix, Direction::Forward)); - let mut hashes = Vec::with_capacity(limit.min(1024)); - for entry in iterator { - let (key, _) = entry?; - let Some(info_hash) = decode_content_member_info_hash(&key, content_key) else { - break; - }; - hashes.push(info_hash); - if hashes.len() == limit { - break; - } - } - Ok(hashes) - } } impl TorrentRepository for RocksTorrentRepository { @@ -276,13 +278,25 @@ impl TorrentRepository for RocksTorrentRepository { } fn get_visible(&self, info_hash: InfoHash) -> Result, StorageError> { + let filter = self.current_filter(); Ok(self .get(info_hash)? - .and_then(|record| self.content_filter.public_record(&record))) + .and_then(|record| filter.public_record(&record))) } fn upsert(&self, mut observation: TorrentRecord) -> Result { - self.content_filter.apply_derivatives(&mut observation)?; + let visible: Vec<_> = observation + .files + .iter() + .filter(|file| !self.canonical_filter.is_file_hidden(file)) + .cloned() + .collect(); + observation.searchable = !visible.is_empty(); + observation.content_key = if observation.searchable { + crate::domain::content_key(&visible)? + } else { + [0; 32] + }; let _guard = self .write_lock .lock() @@ -491,33 +505,30 @@ impl TorrentRepository for RocksTorrentRepository { content_key: &[u8; 32], now: u64, ) -> Result, StorageError> { - let mut builder = ContentGroupBuilder::new(*content_key, now); - let prefix = content_members_prefix(content_key); - let iterator = self - .db - .iterator(IteratorMode::From(&prefix, Direction::Forward)); - let mut found = false; - for entry in iterator { - let (key, _) = entry?; - let Some(info_hash) = decode_content_member_info_hash(&key, content_key) else { - break; - }; - if let Some(record) = self - .get(info_hash)? - .and_then(|record| self.content_filter.public_record(&record)) - { - builder.push(record); - found = true; - } - } - if found { - Ok(builder.finish()) - } else { - Ok(None) - } + let filter = self.current_filter(); + self.content_group_with_filter(content_key, now, &filter) } - fn mark_indexed(&self, content_key: &[u8; 32], revision: u64) -> Result { + fn index_document( + &self, + content_key: &[u8; 32], + now: u64, + ) -> Result { + let group = self.content_group(content_key, now)?; + Ok(IndexDocument { + content_key: *content_key, + projection_hash: projection_hash(group.as_ref()), + group, + }) + } + + fn mark_indexed( + &self, + content_key: &[u8; 32], + revision: u64, + projection_hash: [u8; 32], + visible: bool, + ) -> Result { let _guard = self .write_lock .lock() @@ -531,6 +542,14 @@ 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)); + batch.delete(filter_pending_key(content_key)); + batch.put( + filter_projection_key(content_key), + encode_projection(ProjectionState { + hash: projection_hash, + visible, + }), + ); self.commit_inventory_batch( batch, InventoryDelta { @@ -598,17 +617,26 @@ impl TorrentRepository for RocksTorrentRepository { offset: usize, limit: usize, ) -> Result { - let total = self - .group_state(content_key)? - .map_or(0, |state| state.member_count); - let hashes = self.content_member_hashes(content_key, offset.saturating_add(limit))?; + let filter = self.current_filter(); + let prefix = content_members_prefix(content_key); + let iterator = self + .db + .iterator(IteratorMode::From(&prefix, Direction::Forward)); let mut records = Vec::with_capacity(limit); - for info_hash in hashes.into_iter().skip(offset).take(limit) { + let mut total = 0_u64; + for entry in iterator { + let (key, _) = entry?; + let Some(info_hash) = decode_content_member_info_hash(&key, content_key) else { + break; + }; if let Some(record) = self .get(info_hash)? - .and_then(|record| self.content_filter.public_record(&record)) + .and_then(|record| filter.public_record(&record)) { - records.push(record); + if total >= offset as u64 && records.len() < limit { + records.push(record); + } + total = total.saturating_add(1); } } Ok(ContentVariants { total, records }) @@ -788,6 +816,25 @@ impl TorrentRepository for RocksTorrentRepository { } } +fn encode_projection(state: ProjectionState) -> [u8; 33] { + let mut bytes = [0_u8; 33]; + bytes[0] = u8::from(state.visible); + bytes[1..].copy_from_slice(&state.hash); + bytes +} + +fn decode_projection(bytes: &[u8]) -> Result { + if bytes.len() != 33 || bytes[0] > 1 { + return Err(StorageError::CorruptContentGroup); + } + Ok(ProjectionState { + visible: bytes[0] == 1, + hash: bytes[1..] + .try_into() + .map_err(|_| StorageError::CorruptContentGroup)?, + }) +} + fn read_counter(db: &DB, key: &[u8]) -> Result, StorageError> { db.get(key)? .map(|value| { diff --git a/src/search/src/storage/rocks/filter_migration.rs b/src/search/src/storage/rocks/filter_migration.rs index 39e0851..49b6e41 100644 --- a/src/search/src/storage/rocks/filter_migration.rs +++ b/src/search/src/storage/rocks/filter_migration.rs @@ -1,149 +1,76 @@ -// 负责内容过滤规则变更后的派生字段内容组和待索引标记重建 +// 负责持久化过滤投影基线并增量扫描真正发生变化的内容组 + +use std::sync::Arc; -use super::{ContentGroupState, RocksTorrentRepository}; -use crate::storage::{ - StorageError, - keys::{ - CONTENT_FILTER_FINGERPRINT_KEY, CONTENT_FILTER_MIGRATION_KEY, content_group_key, - content_group_prefix, content_member_key, content_member_prefix, pending_index_key, - pending_index_prefix, torrent_prefix, - }, -}; use rocksdb::{Direction, IteratorMode, WriteBatch}; -use std::collections::BTreeMap; + +use super::{ProjectionState, RocksTorrentRepository, encode_projection}; +use crate::{ + domain::{ContentFilter, ContentGroup, ContentGroupBuilder}, + storage::{StorageError, TorrentRepository, keys::*, repository::projection_hash}, +}; + +const COUNTER_BYTES: usize = std::mem::size_of::(); + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct FilterScanBatch { + pub(crate) scanned: u64, + pub(crate) changed: u64, + pub(crate) finished: bool, +} impl RocksTorrentRepository { - pub(super) fn synchronize_content_filter(&self) -> Result { - const MIGRATION_BATCH_SIZE: usize = 1_000; + pub(crate) fn applied_filter_fingerprint(&self) -> Result, StorageError> { + self.db + .get(FILTER_APPLIED_FINGERPRINT_KEY)? + .map(|bytes| decode_fingerprint(&bytes)) + .transpose() + } - let fingerprint = self.content_filter.fingerprint(); - if self.db.get(CONTENT_FILTER_MIGRATION_KEY)?.is_none() - && self - .db - .get(CONTENT_FILTER_FINGERPRINT_KEY)? - .is_some_and(|stored| stored.as_slice() == fingerprint) - { - return Ok(false); - } - - let _guard = self - .write_lock - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - self.db.put(CONTENT_FILTER_MIGRATION_KEY, fingerprint)?; - self.clear_prefix(content_member_prefix(), MIGRATION_BATCH_SIZE)?; - self.clear_prefix(content_group_prefix(), MIGRATION_BATCH_SIZE)?; - self.clear_prefix(pending_index_prefix(), MIGRATION_BATCH_SIZE)?; - - let prefix = torrent_prefix(); - let iterator = self + pub(crate) fn prepare_filter_scan(&self, fingerprint: [u8; 32]) -> Result<(), StorageError> { + if self .db - .iterator(IteratorMode::From(&prefix, Direction::Forward)); + .get(FILTER_TARGET_FINGERPRINT_KEY)? + .is_some_and(|value| value.as_slice() == fingerprint) + { + return Ok(()); + } let mut batch = WriteBatch::default(); - let mut groups = BTreeMap::<[u8; 32], u64>::new(); - let mut batch_records = 0_usize; - let mut records = 0_u64; - let mut hidden_files = 0_u64; - let mut hidden_records = 0_u64; - for entry in iterator { - let (key, value) = entry?; - if !key.starts_with(&prefix) { - break; - } - let mut record = Self::decode(&value)?; - let old_content_key = record.content_key; - let old_searchable = record.searchable; - let outcome = self.content_filter.apply_derivatives(&mut record)?; - records = records.saturating_add(1); - hidden_files = hidden_files.saturating_add(outcome.hidden_files as u64); - if !outcome.searchable { - hidden_records = hidden_records.saturating_add(1); - } - if old_content_key != record.content_key || old_searchable != record.searchable { - batch.put(&key, Self::encode(&record)?); - } - if record.searchable { - batch.put( - content_member_key(&record.content_key, record.info_hash), - [], - ); - *groups.entry(record.content_key).or_default() += 1; - } - batch_records += 1; - if batch_records == MIGRATION_BATCH_SIZE { - self.write_filter_migration_batch(batch, &groups)?; - batch = WriteBatch::default(); - groups.clear(); - batch_records = 0; - } - } - if batch_records > 0 { - self.write_filter_migration_batch(batch, &groups)?; - } - self.restore_pending_markers(MIGRATION_BATCH_SIZE)?; - let mut finish = WriteBatch::default(); - finish.put(CONTENT_FILTER_FINGERPRINT_KEY, fingerprint); - finish.delete(CONTENT_FILTER_MIGRATION_KEY); - self.db.write(finish)?; - tracing::info!( - records, - hidden_files, - hidden_records, - "内容过滤规则已更新并重建内容组" - ); - Ok(true) - } - - fn clear_prefix(&self, prefix: [u8; 1], limit: usize) -> Result<(), StorageError> { - loop { - let iterator = self - .db - .iterator(IteratorMode::From(&prefix, Direction::Forward)); - let mut batch = WriteBatch::default(); - let mut deleted = 0_usize; - for entry in iterator.take(limit) { - let (key, _) = entry?; - if !key.starts_with(&prefix) { - break; - } - batch.delete(key); - deleted += 1; - } - if deleted == 0 { - return Ok(()); - } - self.db.write(batch)?; - } - } - - fn write_filter_migration_batch( - &self, - mut batch: WriteBatch, - groups: &BTreeMap<[u8; 32], u64>, - ) -> Result<(), StorageError> { - for (content_key, added) in groups { - let member_count = self - .group_state(content_key)? - .map_or(*added, |state| state.member_count.saturating_add(*added)); - batch.put( - content_group_key(content_key), - Self::encode_group(ContentGroupState { - revision: 1, - member_count, - })?, - ); - } + batch.put(FILTER_TARGET_FINGERPRINT_KEY, fingerprint); + batch.delete(FILTER_SCAN_CURSOR_KEY); + batch.put(FILTER_SCAN_SCANNED_KEY, 0_u64.to_be_bytes()); + batch.put(FILTER_SCAN_CHANGED_KEY, 0_u64.to_be_bytes()); self.db.write(batch)?; Ok(()) } - fn restore_pending_markers(&self, batch_size: usize) -> Result<(), StorageError> { + pub(crate) fn scan_filter_batch( + &self, + filter: Arc, + baseline: bool, + limit: usize, + now: u64, + ) -> Result { + if limit == 0 { + return Ok(FilterScanBatch::default()); + } + let cursor = self.db.get(FILTER_SCAN_CURSOR_KEY)?; let prefix = content_group_prefix(); + let start = cursor + .as_deref() + .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok()) + .map(|content_key| content_group_key(&content_key)) + .unwrap_or_else(|| { + let mut key = [0_u8; 33]; + key[0] = prefix[0]; + key + }); let iterator = self .db - .iterator(IteratorMode::From(&prefix, Direction::Forward)); - let mut batch = WriteBatch::default(); - let mut batch_len = 0_usize; + .iterator(IteratorMode::From(&start, Direction::Forward)); + let mut scanned = 0_u64; + let mut changed = 0_u64; + let mut last = None; for entry in iterator { let (key, _) = entry?; if !key.starts_with(&prefix) { @@ -152,17 +79,178 @@ impl RocksTorrentRepository { let content_key: [u8; 32] = key[1..] .try_into() .map_err(|_| StorageError::CorruptContentGroup)?; - batch.put(pending_index_key(&content_key), 1_u64.to_be_bytes()); - batch_len += 1; - if batch_len == batch_size { - self.db.write(batch)?; - batch = WriteBatch::default(); - batch_len = 0; + if cursor.as_deref() == Some(content_key.as_slice()) { + continue; + } + let group = self.content_group_with_filter(&content_key, now, &filter)?; + let hash = projection_hash(group.as_ref()); + if baseline { + if self.db.get(pending_index_key(&content_key))?.is_none() + && self.projection_state(&content_key)?.is_none() + { + self.db.put( + filter_projection_key(&content_key), + encode_projection(ProjectionState { + hash, + visible: group.is_some(), + }), + )?; + } + } else if self + .projection_state(&content_key)? + .is_none_or(|state| state.hash != hash || state.visible != group.is_some()) + { + self.enqueue_filter_group(content_key)?; + changed = changed.saturating_add(1); + } + scanned = scanned.saturating_add(1); + last = Some(content_key); + if scanned as usize == limit { + break; } } - if batch_len > 0 { - self.db.write(batch)?; + let finished = scanned < limit as u64; + let mut batch = WriteBatch::default(); + if let Some(content_key) = last { + batch.put(FILTER_SCAN_CURSOR_KEY, content_key); } + let total_scanned = self + .read_counter(FILTER_SCAN_SCANNED_KEY)? + .saturating_add(scanned); + let total_changed = self + .read_counter(FILTER_SCAN_CHANGED_KEY)? + .saturating_add(changed); + batch.put(FILTER_SCAN_SCANNED_KEY, total_scanned.to_be_bytes()); + batch.put(FILTER_SCAN_CHANGED_KEY, total_changed.to_be_bytes()); + self.db.write(batch)?; + Ok(FilterScanBatch { + scanned, + changed, + finished, + }) + } + + pub(crate) fn finish_filter_scan(&self, fingerprint: [u8; 32]) -> Result<(), StorageError> { + let mut batch = WriteBatch::default(); + batch.put(FILTER_APPLIED_FINGERPRINT_KEY, fingerprint); + batch.delete(FILTER_TARGET_FINGERPRINT_KEY); + batch.delete(FILTER_SCAN_CURSOR_KEY); + self.db.write(batch)?; Ok(()) } + + pub(crate) fn filter_scan_counters(&self) -> Result<(u64, u64), StorageError> { + Ok(( + self.read_counter(FILTER_SCAN_SCANNED_KEY)?, + self.read_counter(FILTER_SCAN_CHANGED_KEY)?, + )) + } + + pub(crate) fn filter_pending_len(&self) -> Result { + let prefix = filter_pending_prefix(); + let iterator = self + .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) + } + + pub(crate) fn visible_group_count(&self, now: u64) -> Result { + let filter = self.current_filter(); + let prefix = content_group_prefix(); + let iterator = self + .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; + } + let content_key: [u8; 32] = key[1..] + .try_into() + .map_err(|_| StorageError::CorruptContentGroup)?; + if self + .content_group_with_filter(&content_key, now, &filter)? + .is_some() + { + count = count.saturating_add(1); + } + } + Ok(count) + } + + pub(crate) fn content_group_with_filter( + &self, + content_key: &[u8; 32], + now: u64, + filter: &ContentFilter, + ) -> Result, StorageError> { + let mut builder = ContentGroupBuilder::new(*content_key, now); + let prefix = content_members_prefix(content_key); + let iterator = self + .db + .iterator(IteratorMode::From(&prefix, Direction::Forward)); + let mut found = false; + for entry in iterator { + let (key, _) = entry?; + let Some(info_hash) = decode_content_member_info_hash(&key, content_key) else { + break; + }; + if let Some(record) = self + .get(info_hash)? + .and_then(|record| filter.public_record(&record)) + { + builder.push(record); + found = true; + } + } + if found { + Ok(builder.finish()) + } else { + Ok(None) + } + } + + fn enqueue_filter_group(&self, content_key: [u8; 32]) -> Result<(), StorageError> { + let _guard = self + .write_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut batch = WriteBatch::default(); + let delta = self.dirty_group(&mut batch, &content_key, false)?; + batch.put(filter_pending_key(&content_key), []); + self.commit_inventory_batch(batch, delta) + } + + fn read_counter(&self, key: &[u8]) -> Result { + self.db + .get(key)? + .map(|bytes| { + if bytes.len() != COUNTER_BYTES { + return Err(StorageError::CorruptContentGroup); + } + Ok(u64::from_be_bytes( + bytes + .as_slice() + .try_into() + .map_err(|_| StorageError::CorruptContentGroup)?, + )) + }) + .transpose() + .map(|value| value.unwrap_or(0)) + } +} + +fn decode_fingerprint(bytes: &[u8]) -> Result<[u8; 32], StorageError> { + bytes + .try_into() + .map_err(|_| StorageError::CorruptContentGroup) } diff --git a/src/search/src/storage/rocks/lifecycle.rs b/src/search/src/storage/rocks/lifecycle.rs index 88f6064..ee71745 100644 --- a/src/search/src/storage/rocks/lifecycle.rs +++ b/src/search/src/storage/rocks/lifecycle.rs @@ -48,24 +48,19 @@ impl RocksTorrentRepository { options.set_prefix_extractor(SliceTransform::create_fixed_prefix(1)); options.set_max_open_files(256); - let mut repository = Self { + let repository = Self { db: DB::open(&options, path)?, write_lock: Mutex::new(()), rejection_rule_id, - content_filter, - content_filter_changed: false, + canonical_filter: Arc::new(ContentFilter::legacy_default()), + content_filter: std::sync::RwLock::new(content_filter), inventory: Default::default(), }; repository.initialize_format()?; - repository.content_filter_changed = repository.synchronize_content_filter()?; - repository.initialize_inventory(repository.content_filter_changed)?; + repository.initialize_inventory(false)?; Ok(repository) } - pub fn content_filter_changed(&self) -> bool { - self.content_filter_changed - } - pub fn diagnostics(&self) -> Result { Ok(StorageDiagnostics { block_cache_bytes: self.db.property_int_value("rocksdb.block-cache-usage")?, diff --git a/src/search/src/storage/rocks/tests.rs b/src/search/src/storage/rocks/tests.rs index 475c72d..c914b77 100644 --- a/src/search/src/storage/rocks/tests.rs +++ b/src/search/src/storage/rocks/tests.rs @@ -5,10 +5,10 @@ use std::sync::Arc; use tempfile::TempDir; use crate::domain::{ - AvailabilityStatus, ContentFilter, ContentFilterConfig, FileFilterRule, FileMatchField, - FileMatchKind, FileRuleAction, MetadataLimits, TorrentFile, VerificationResult, test_record, + AvailabilityStatus, ContentFilter, ContentFilterConfig, MetadataLimits, TorrentFile, + VerificationResult, test_record, }; -use crate::storage::keys::{CONTENT_FILTER_MIGRATION_KEY, DATABASE_FORMAT_KEY}; +use crate::storage::keys::DATABASE_FORMAT_KEY; use rocksdb::Options; use super::*; @@ -16,17 +16,8 @@ use super::*; fn padding_filter() -> Arc { Arc::new( ContentFilter::compile(ContentFilterConfig { - version: 1, - file_rules: vec![FileFilterRule { - id: "padding".into(), - enabled: true, - field: FileMatchField::FileName, - match_kind: FileMatchKind::Prefix, - value: "_____padding_file_".into(), - case_sensitive: false, - action: FileRuleAction::Hide, - reason: "测试".into(), - }], + torrent_name_patterns: Vec::new(), + file_patterns: vec!["*_____padding_file_*".into()], }) .unwrap(), ) @@ -65,7 +56,7 @@ fn index_inventory_is_exact_and_survives_reopen() { let task = repository.pending_index(1).unwrap()[0]; assert!( repository - .mark_indexed(&task.content_key, task.revision) + .mark_indexed(&task.content_key, task.revision, [0; 32], true) .unwrap() ); assert_eq!(repository.index_inventory().pending_documents, 0); @@ -118,95 +109,117 @@ fn filter_keeps_raw_files_but_returns_effective_view() { assert_eq!(raw.total_size, 142); let visible = repository.get_visible(record.info_hash).unwrap().unwrap(); assert_eq!(visible.files.len(), 1); - assert_eq!(visible.total_size, 42); + assert_eq!(visible.total_size, 142); + assert_eq!(visible.original_file_count(), 2); assert_eq!(repository.pending_index(10).unwrap().len(), 1); } #[test] -fn changed_filter_rebuilds_groups_without_deleting_raw_metadata() { +fn changed_filter_enqueues_only_a_changed_group_without_rekeying_metadata() { let directory = TempDir::new().unwrap(); + let repository = RocksTorrentRepository::open(directory.path()).unwrap(); let mut record = test_record(2, 10); - record.files.push(TorrentFile { - path: "_____padding_file_1_".into(), - size: 100, - }); - record.total_size += 100; - let old_content_key; - { - let repository = RocksTorrentRepository::open(directory.path()).unwrap(); - repository.upsert(record.clone()).unwrap(); - old_content_key = repository + record.name = "普通资源".into(); + repository.upsert(record.clone()).unwrap(); + let mut clean = test_record(3, 10); + clean.files[0].path = "clean.bin".into(); + clean.content_key = crate::domain::content_key(&clean.files).unwrap(); + repository.upsert(clean.clone()).unwrap(); + for task in repository.pending_index(10).unwrap() { + let document = repository.index_document(&task.content_key, 20).unwrap(); + repository + .mark_indexed( + &task.content_key, + task.revision, + document.projection_hash, + document.group.is_some(), + ) + .unwrap(); + } + let filter = Arc::new( + ContentFilter::compile(ContentFilterConfig { + torrent_name_patterns: vec!["*普通*".into()], + file_patterns: Vec::new(), + }) + .unwrap(), + ); + repository.set_content_filter(filter.clone()); + repository + .prepare_filter_scan(filter.fingerprint()) + .unwrap(); + assert!( + repository + .scan_filter_batch(filter, false, 10, 30) + .unwrap() + .finished + ); + assert_eq!(repository.pending_index(10).unwrap().len(), 1); + assert_eq!( + repository .get(record.info_hash) .unwrap() .unwrap() - .content_key; - } - - let repository = RocksTorrentRepository::open_with_rules( - directory.path(), - MetadataLimits::default().rule_id(), - padding_filter(), - ) - .unwrap(); - assert!(repository.content_filter_changed()); - let raw = repository.get(record.info_hash).unwrap().unwrap(); - assert_eq!(raw.files.len(), 2); - assert_ne!(raw.content_key, old_content_key); - assert_eq!( - repository - .content_variants(&old_content_key, 0, 10) - .unwrap() - .total, - 0 + .content_key, + record.content_key ); - assert_eq!( - repository - .content_variants(&raw.content_key, 0, 10) - .unwrap() - .total, - 1 - ); - drop(repository); - - let rolled_back = RocksTorrentRepository::open(directory.path()).unwrap(); - assert!(rolled_back.content_filter_changed()); - let restored = rolled_back.get_visible(record.info_hash).unwrap().unwrap(); - assert_eq!(restored.files.len(), 2); - assert_eq!(restored.total_size, 142); - assert_eq!(restored.content_key, old_content_key); } #[test] -fn unfinished_filter_migration_is_rebuilt_even_when_fingerprint_matches() { +fn filter_scan_with_unchanged_projection_does_not_reindex() { let directory = TempDir::new().unwrap(); - let record = test_record(4, 10); - { - let repository = RocksTorrentRepository::open(directory.path()).unwrap(); - repository.upsert(record.clone()).unwrap(); - repository - .db - .put(CONTENT_FILTER_MIGRATION_KEY, b"interrupted") - .unwrap(); - repository - .db - .delete(content_member_key(&record.content_key, record.info_hash)) - .unwrap(); - } + let repository = RocksTorrentRepository::open(directory.path()).unwrap(); + let record = test_record(5, 10); + repository.upsert(record).unwrap(); + let task = repository.pending_index(1).unwrap()[0]; + let document = repository.index_document(&task.content_key, 20).unwrap(); + repository + .mark_indexed( + &task.content_key, + task.revision, + document.projection_hash, + true, + ) + .unwrap(); + let filter = Arc::new(ContentFilter::legacy_default()); + repository + .prepare_filter_scan(filter.fingerprint()) + .unwrap(); - let recovered = RocksTorrentRepository::open(directory.path()).unwrap(); - assert!(recovered.content_filter_changed()); - assert_eq!( - recovered - .content_variants(&record.content_key, 0, 10) - .unwrap() - .total, - 1 + let scan = repository.scan_filter_batch(filter, false, 10, 30).unwrap(); + + assert_eq!(scan.changed, 0); + assert!(repository.pending_index(10).unwrap().is_empty()); +} + +#[test] +fn title_filter_hides_only_matching_variant_in_a_group() { + let directory = TempDir::new().unwrap(); + let repository = RocksTorrentRepository::open(directory.path()).unwrap(); + let clean = test_record(6, 10); + let mut advertisement = test_record(7, 20); + advertisement.content_key = clean.content_key; + advertisement.name = "电影【加QQ 123456】".into(); + repository.upsert(clean.clone()).unwrap(); + repository.upsert(advertisement.clone()).unwrap(); + let filter = Arc::new( + ContentFilter::compile(ContentFilterConfig { + torrent_name_patterns: vec!["*【加QQ *】*".into()], + file_patterns: Vec::new(), + }) + .unwrap(), ); - assert!(recovered.pending_index(10).unwrap().len() == 1); + repository.set_content_filter(filter); + + let group = repository + .content_group(&clean.content_key, 30) + .unwrap() + .unwrap(); + + assert_eq!(group.variant_count, 1); + assert_eq!(group.representative.info_hash, clean.info_hash); assert!( - recovered - .db - .get(CONTENT_FILTER_MIGRATION_KEY) + repository + .get_visible(advertisement.info_hash) .unwrap() .is_none() ); @@ -409,7 +422,7 @@ fn marking_indexed_is_atomic_with_removing_pending_marker() { let task = repository.pending_index(10).unwrap()[0]; assert!( repository - .mark_indexed(&record.content_key, task.revision) + .mark_indexed(&record.content_key, task.revision, [0; 32], true) .unwrap() ); @@ -428,7 +441,7 @@ fn stale_index_revision_cannot_clear_a_newer_update() { assert!( !repository - .mark_indexed(&record.content_key, stale.revision) + .mark_indexed(&record.content_key, stale.revision, [0; 32], true) .unwrap() ); let current = repository.pending_index(10).unwrap()[0]; @@ -464,7 +477,7 @@ fn full_reindex_restores_pending_markers_for_every_record() { repository.upsert(second.clone()).unwrap(); for task in repository.pending_index(10).unwrap() { repository - .mark_indexed(&task.content_key, task.revision) + .mark_indexed(&task.content_key, task.revision, [0; 32], true) .unwrap(); } assert!(repository.pending_index(10).unwrap().is_empty()); diff --git a/src/search/tests/storage_search_flow.rs b/src/search/tests/storage_search_flow.rs index b528b21..523e763 100644 --- a/src/search/tests/storage_search_flow.rs +++ b/src/search/tests/storage_search_flow.rs @@ -3,10 +3,7 @@ #![cfg(feature = "rocksdb-storage")] use dht_search::{ - domain::{ - ContentFilter, ContentFilterConfig, FileFilterRule, FileMatchField, FileMatchKind, - FileRuleAction, MetadataCandidate, TorrentFile, TorrentRecord, - }, + domain::{ContentFilter, ContentFilterConfig, MetadataCandidate, TorrentFile, TorrentRecord}, search::SearchEngine, storage::{RocksTorrentRepository, TorrentRepository, UpsertOutcome}, }; @@ -54,17 +51,8 @@ fn content_filter_excludes_padding_from_search_and_visible_details() { let directory = TempDir::new().unwrap(); let filter = Arc::new( ContentFilter::compile(ContentFilterConfig { - version: 1, - file_rules: vec![FileFilterRule { - id: "bitcomet-padding".into(), - enabled: true, - field: FileMatchField::FileName, - match_kind: FileMatchKind::Prefix, - value: "_____padding_file_".into(), - case_sensitive: false, - action: FileRuleAction::Hide, - reason: "测试".into(), - }], + torrent_name_patterns: Vec::new(), + file_patterns: vec!["*_____padding_file_*".into()], }) .unwrap(), ); @@ -101,5 +89,6 @@ fn content_filter_excludes_padding_from_search_and_visible_details() { assert_eq!(search.search("*.mkv", 0, 10).unwrap().total, 1); let visible = repository.get_visible(record.info_hash).unwrap().unwrap(); assert_eq!(visible.files.len(), 1); - assert_eq!(visible.total_size, 42); + assert_eq!(visible.total_size, 142); + assert_eq!(visible.original_file_count(), 2); } diff --git a/src/web/src/components/TorrentDetailDialog.vue b/src/web/src/components/TorrentDetailDialog.vue index 8b73dc0..c87ae17 100644 --- a/src/web/src/components/TorrentDetailDialog.vue +++ b/src/web/src/components/TorrentDetailDialog.vue @@ -113,7 +113,7 @@ onBeforeUnmount(() => {

文件详情

共 {{ detail.file_count.toLocaleString() }} 个文件

- +
diff --git a/src/web/src/lib/api.ts b/src/web/src/lib/api.ts index 8d38b8a..25e14cc 100644 --- a/src/web/src/lib/api.ts +++ b/src/web/src/lib/api.ts @@ -49,8 +49,9 @@ export function search(input: SearchInput, signal?: AbortSignal): Promise(`/search?${params}`, signal) } -export function getTorrent(infoHash: string, fileOffset = 0, fileLimit = 100, signal?: AbortSignal): Promise { - const params = new URLSearchParams({ file_offset: String(fileOffset), file_limit: String(fileLimit) }) +export function getTorrent(infoHash: string, fileOffset = 0, fileLimit = 100, query = '', sort: SearchSort = 'latest', signal?: AbortSignal): Promise { + const params = new URLSearchParams({ file_offset: String(fileOffset), file_limit: String(fileLimit), q: query.trim(), sort }) + if (isLikelyRegex(query)) params.set('mode', 'regex') return request(`/torrents/${encodeURIComponent(infoHash)}?${params}`, signal) } diff --git a/src/web/src/pages/ConfigPage.vue b/src/web/src/pages/ConfigPage.vue index 77a1d49..66763a0 100644 --- a/src/web/src/pages/ConfigPage.vue +++ b/src/web/src/pages/ConfigPage.vue @@ -59,8 +59,8 @@ const saving = ref(false) const error = ref('') const saved = ref(false) const metadataUnit = ref('MiB') -const fileNamePatterns = ref('') -const filePathPatterns = ref('') +const torrentNamePatterns = ref('') +const filePatterns = ref('') const dirty = computed(() => config.value !== null && JSON.stringify(config.value) !== original.value) function fieldValue(path: string): unknown { @@ -125,15 +125,15 @@ function splitPatterns(value: string): string[] { function syncContentFilter() { if (!config.value) return - config.value.content_filter.file_name_patterns = splitPatterns(fileNamePatterns.value) - config.value.content_filter.file_path_patterns = splitPatterns(filePathPatterns.value) + config.value.content_filter.torrent_name_patterns = splitPatterns(torrentNamePatterns.value) + config.value.content_filter.file_patterns = splitPatterns(filePatterns.value) saved.value = false } function syncEditorState(next: AppConfigDto) { metadataUnit.value = selectByteUnit(next.metadata_limits.max_metadata_bytes) - fileNamePatterns.value = next.content_filter.file_name_patterns.join('\n') - filePathPatterns.value = next.content_filter.file_path_patterns.join('\n') + torrentNamePatterns.value = next.content_filter.torrent_name_patterns.join('\n') + filePatterns.value = next.content_filter.file_patterns.join('\n') } async function load() { @@ -219,8 +219,8 @@ onBeforeRouteLeave(() => !dirty.value || window.confirm('配置尚未保存，
-