feat(search): add live content filtering and contextual file order

This commit is contained in:
chuan
2026-08-11 09:49:12 +08:00
parent 418fee42c0
commit e74ebf3a95
40 changed files with 1671 additions and 802 deletions
+1 -1
View File
@@ -15,7 +15,7 @@ opencodes/ 不参与构建且不得修改的参考项目
RocksDB 是唯一权威数据源 Tantivy 索引可以从 RocksDB 完整重建
Tantivy 使用代际影子索引完成全量重建 过滤规则或索引文档结构变化时旧索引继续提供搜索 新索引完整校验后通过原子指针切换 重建状态和进度可以在搜索提示与系统诊断页查看
Tantivy 使用代际影子索引处理 Schema 文档格式损坏和缺失等全量重建 用户过滤规则通过可恢复扫描只增量更新真正变化的搜索文档 两类进度可以在搜索提示与系统诊断页查看
应用全部配置和内容隐藏规则统一位于 [`config.toml`](config.toml)
+8 -4
View File
@@ -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] 增加配置查询完整校验原子保存并发修订和统一重启提示
+30 -27
View File
@@ -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"
+8 -6
View File
@@ -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 文件占用造成的清理失败只记录警告而不影响服务
+12 -1
View File
@@ -145,6 +145,7 @@ pub(crate) async fn stats(State(state): State<ApiState>) -> Json<StatsResponse>
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<ApiState>,
Json(request): Json<ConfigUpdateRequest>,
) -> Result<Json<ConfigSnapshot>, 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)
+26 -3
View File
@@ -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(
+5
View File
@@ -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<SearchSort>,
}
#[derive(Debug, Deserialize)]
+9 -3
View File
@@ -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<TorrentFile>,
@@ -166,12 +169,13 @@ impl From<TorrentRecord> 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,
+13 -3
View File
@@ -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<dyn TorrentRepository> = 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 {
+4 -8
View File
@@ -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]
+15 -48
View File
@@ -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<String>,
pub(crate) file_path_patterns: Vec<String>,
pub(crate) torrent_name_patterns: Vec<String>,
pub(crate) file_patterns: Vec<String>,
}
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<FileFilterRule> {
fn normalize_patterns(patterns: &[String]) -> Vec<String> {
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 {
+22
View File
@@ -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();
+7 -1
View File
@@ -297,7 +297,13 @@ mod tests {
Ok(None)
}
fn mark_indexed(&self, _: &[u8; 32], _: u64) -> Result<bool, StorageError> {
fn mark_indexed(
&self,
_: &[u8; 32],
_: u64,
_: [u8; 32],
_: bool,
) -> Result<bool, StorageError> {
Ok(true)
}
+13 -2
View File
@@ -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<dyn TorrentRepository> = 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)
+2
View File
@@ -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<u64>,
pub(crate) block_cache_bytes: Option<u64>,
pub(crate) memtable_bytes: Option<u64>,
pub(crate) pending_compaction_bytes: Option<u64>,
+148 -288
View File
@@ -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<FileFilterRule>,
}
#[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<String>,
pub file_patterns: Vec<String>,
}
#[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<CompiledRule>,
torrent_name_rules: Vec<CompiledPattern>,
file_rules: Vec<CompiledPattern>,
}
#[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<Self, ContentFilterError> {
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<FilterOutcome, TorrentRecordError> {
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<TorrentRecord> {
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<Vec<CompiledPattern>, 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::<String>().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<FileFilterRule>) -> 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 { .. })
));
}
}
+1 -4
View File
@@ -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;
+9
View File
@@ -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<u64>,
}
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,
}
}
+430
View File
@@ -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<String>,
}
#[derive(Clone)]
pub(crate) struct FilterRuntime {
repository: Arc<RocksTorrentRepository>,
target: watch::Sender<Arc<ContentFilter>>,
status: Arc<RwLock<FilterStatus>>,
}
impl FilterRuntime {
#[cfg(test)]
pub(crate) fn for_test(repository: Arc<RocksTorrentRepository>) -> 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<RocksTorrentRepository>,
target: Arc<ContentFilter>,
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<ContentFilter>) {
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<RocksTorrentRepository>,
mut target: watch::Receiver<Arc<ContentFilter>>,
status: Arc<RwLock<FilterStatus>>,
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<RocksTorrentRepository>,
filter: Arc<ContentFilter>,
baseline: bool,
target: &watch::Receiver<Arc<ContentFilter>>,
status: &Arc<RwLock<FilterStatus>>,
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<RwLock<FilterStatus>>, 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<RwLock<FilterStatus>>,
state: FilterState,
total: u64,
scanned: u64,
changed: u64,
pending: u64,
error: Option<String>,
) {
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");
}
}
+6 -6
View File
@@ -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;
+2
View File
@@ -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;
+1 -1
View File
@@ -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);
+219
View File
@@ -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<TorrentFile>,
query: &str,
mode: SearchMode,
sort: Option<SearchSort>,
) -> Result<Vec<TorrentFile>, 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<String>),
Pattern(Regex),
}
impl FileMatcher {
fn compile(query: &str, mode: SearchMode) -> Result<Self, FileOrderError> {
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<SearchSort>) -> 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::<String>().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(&regex::escape(&character.to_string()));
escaped = false;
} else {
match character {
'\\' => escaped = true,
'*' => output.push_str(".*"),
'?' => output.push('.'),
_ => output.push_str(&regex::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");
}
}
+56 -16
View File
@@ -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<usize, IndexingError> {
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();
+2
View File
@@ -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,
+5 -11
View File
@@ -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<SearchBootstrap, SearchError> {
pub(crate) fn open(data_dir: &Path) -> Result<SearchBootstrap, SearchError> {
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);
+19 -6
View File
@@ -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],
+4 -2
View File
@@ -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")]
+55 -1
View File
@@ -46,7 +46,27 @@ pub trait TorrentRepository: Send + Sync {
now: u64,
) -> Result<Option<ContentGroup>, StorageError>;
fn mark_indexed(&self, content_key: &[u8; 32], revision: u64) -> Result<bool, StorageError>;
fn index_document(
&self,
content_key: &[u8; 32],
now: u64,
) -> Result<IndexDocument, StorageError> {
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<bool, StorageError>;
fn prepare_full_reindex(&self) -> Result<u64, StorageError>;
@@ -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<ContentGroup>,
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")]
+120 -73
View File
@@ -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<ContentFilter>,
content_filter_changed: bool,
canonical_filter: Arc<ContentFilter>,
content_filter: RwLock<Arc<ContentFilter>>,
inventory: InventoryState,
}
@@ -64,6 +68,20 @@ struct InventoryDelta {
}
impl RocksTorrentRepository {
fn current_filter(&self) -> Arc<ContentFilter> {
self.content_filter
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
pub(crate) fn set_content_filter(&self, filter: Arc<ContentFilter>) {
*self
.content_filter
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner) = filter;
}
fn encode(record: &TorrentRecord) -> Result<Vec<u8>, 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<Option<ProjectionState>, StorageError> {
self.db
.get(filter_projection_key(content_key))?
.map(|bytes| decode_projection(&bytes))
.transpose()
}
fn encode_group(state: ContentGroupState) -> Result<Vec<u8>, 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<Vec<InfoHash>, 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<Option<TorrentRecord>, 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<UpsertOutcome, StorageError> {
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<Option<crate::domain::ContentGroup>, 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<bool, StorageError> {
fn index_document(
&self,
content_key: &[u8; 32],
now: u64,
) -> Result<IndexDocument, StorageError> {
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<bool, StorageError> {
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<ContentVariants, StorageError> {
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<ProjectionState, StorageError> {
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<Option<u64>, StorageError> {
db.get(key)?
.map(|value| {
+228 -140
View File
@@ -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::<u64>();
#[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<bool, StorageError> {
const MIGRATION_BATCH_SIZE: usize = 1_000;
pub(crate) fn applied_filter_fingerprint(&self) -> Result<Option<[u8; 32]>, 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<ContentFilter>,
baseline: bool,
limit: usize,
now: u64,
) -> Result<FilterScanBatch, StorageError> {
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<u64, StorageError> {
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<u64, StorageError> {
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<Option<ContentGroup>, 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<u64, StorageError> {
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)
}
+4 -9
View File
@@ -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<StorageDiagnostics, StorageError> {
Ok(StorageDiagnostics {
block_cache_bytes: self.db.property_int_value("rocksdb.block-cache-usage")?,
+103 -90
View File
@@ -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<ContentFilter> {
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());
+5 -16
View File
@@ -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);
}
@@ -113,7 +113,7 @@ onBeforeUnmount(() => {
<div><h3 class="font-semibold">文件详情</h3><p class="mt-1 text-xs text-muted-foreground"> {{ detail.file_count.toLocaleString() }} 个文件</p></div>
<div class="ml-auto flex flex-wrap items-center justify-end gap-2">
<AppSelect :model-value="String(detail.file_limit)" :options="filePageSizeOptions" label="每页文件数量" @update:model-value="changeFilePageSize" />
<AppPagination v-if="detail.file_count > detail.file_limit" :disabled="filesLoading" :page="filePage" :page-size="detail.file_limit" :total="detail.file_count" @update:page="changeFilePage" />
<AppPagination v-if="detail.visible_file_count > detail.file_limit" :disabled="filesLoading" :page="filePage" :page-size="detail.file_limit" :total="detail.visible_file_count" @update:page="changeFilePage" />
</div>
</div>
<div class="relative overflow-hidden rounded-xl border bg-card">
+3 -2
View File
@@ -49,8 +49,9 @@ export function search(input: SearchInput, signal?: AbortSignal): Promise<Search
return request<SearchPage>(`/search?${params}`, signal)
}
export function getTorrent(infoHash: string, fileOffset = 0, fileLimit = 100, signal?: AbortSignal): Promise<TorrentDetail> {
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<TorrentDetail> {
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<TorrentDetail>(`/torrents/${encodeURIComponent(infoHash)}?${params}`, signal)
}
+8 -8
View File
@@ -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('配置尚未保存,
</label>
</div>
<div v-else class="grid gap-4 p-5 md:grid-cols-2">
<label class="field-label">文件名过滤<textarea v-model="fileNamePatterns" class="filter-textarea" spellcheck="false" placeholder="每行一条通配符&#10;*_____padding_file_*" @input="syncContentFilter" /></label>
<label class="field-label">文件路径过滤<textarea v-model="filePathPatterns" class="filter-textarea" spellcheck="false" placeholder="每行一条通配符&#10;*.pad/*" @input="syncContentFilter" /></label>
<label class="field-label">种子标题<textarea v-model="torrentNamePatterns" class="filter-textarea" spellcheck="false" placeholder="每行一条规则&#10;*【加QQ *】*&#10;regex:【加QQ\\s+[^】]+】" @input="syncContentFilter" /></label>
<label class="field-label">种子内部文件<textarea v-model="filePatterns" class="filter-textarea" spellcheck="false" placeholder="同时匹配文件名和完整路径&#10;*_____padding_file_*&#10;regex:(^|/)\\.pad/.*" @input="syncContentFilter" /></label>
</div>
</section>
</TabsContent>
+30 -11
View File
@@ -39,16 +39,16 @@ const rangeOptions = [
const sample = computed(() => current.value?.sample ?? null)
const samples = computed(() => history.value?.samples ?? [])
function chartOption(series: Array<{ name: string; values: Array<number | null> }>, suffix: string): EChartsOption {
function chartOption(series: Array<{ name: string; values: Array<number | null> }>, suffix: string, precision = 1, scale = false): EChartsOption {
const labels = samples.value.map((item) => new Date(item.captured_at * 1000).toLocaleString('zh-CN', { hour: '2-digit', minute: '2-digit', month: '2-digit', day: '2-digit' }))
return {
animation: false,
color: ['#2563eb', '#10b981', '#f59e0b', '#8b5cf6'],
grid: { left: 48, right: 16, top: 36, bottom: 30 },
legend: { top: 0, textStyle: { color: '#737373' } },
tooltip: { trigger: 'axis', valueFormatter: (value) => `${Number(value).toFixed(1)} ${suffix}` },
tooltip: { trigger: 'axis', valueFormatter: (value) => `${Number(value).toLocaleString('zh-CN', { maximumFractionDigits: precision })} ${suffix}` },
xAxis: { type: 'category', data: labels, boundaryGap: false, axisLabel: { color: '#737373', hideOverlap: true }, axisLine: { lineStyle: { color: '#a3a3a3' } } },
yAxis: { type: 'value', axisLabel: { color: '#737373', formatter: `{value} ${suffix}` }, splitLine: { lineStyle: { color: 'rgba(115,115,115,.15)' } } },
yAxis: { type: 'value', scale, axisLabel: { color: '#737373', formatter: `{value} ${suffix}` }, splitLine: { lineStyle: { color: 'rgba(115,115,115,.15)' } } },
series: series.map((item) => ({ name: item.name, type: 'line', data: item.values, showSymbol: false, smooth: true, connectNulls: true, lineStyle: { width: 2 }, areaStyle: { opacity: 0.05 } })),
}
}
@@ -108,11 +108,9 @@ const metadataPeerSummary = computed(() => {
return `成功率 ${successRate.toFixed(2)}%`
})
const httpChart = computed(() => chartOption([
{ name: 'HTTP 请求', values: perSecond(samples.value, (item) => item.http.requests) },
{ name: '客户端错误', values: perSecond(samples.value, (item) => item.http.client_errors) },
{ name: '服务端错误', values: perSecond(samples.value, (item) => item.http.server_errors) },
], '/s'))
const torrentChart = computed(() => chartOption([
{ name: '已保存种子', values: samples.value.map((item) => item.storage.stored_torrents) },
], '个', 0, true))
function bytesToMiB(value: number | null): number | null {
return value === null ? null : Number((value / 1_048_576).toFixed(2))
@@ -178,7 +176,6 @@ function indexStateLabel(state: ServiceStats['index']['state']): string {
}
function indexReasonLabel(reason: ServiceStats['index']['reason']): string {
if (reason === 'content_filter') return '内容过滤规则变化'
if (reason === 'schema') return '索引结构变化'
if (reason === 'document_format') return '索引文档格式变化'
if (reason === 'missing') return '索引缺失'
@@ -186,6 +183,14 @@ function indexReasonLabel(reason: ServiceStats['index']['reason']): string {
return '首次构建'
}
function filterStateLabel(state: ServiceStats['filter']['state']): string {
if (state === 'ready') return '过滤规则已同步'
if (state === 'baselining') return '正在建立过滤状态基线'
if (state === 'scanning') return '正在扫描内容组'
if (state === 'applying') return '正在提交变化文档'
return '过滤规则应用失败'
}
async function loadStats() {
if (statsRequest) return
statsRequest = true
@@ -227,7 +232,7 @@ onBeforeUnmount(() => {
<article class="diagnostic-card"><HardDrive /><span>磁盘</span><b class="flex items-center gap-2"><i class="size-1.5 rounded-full" :class="stats.disk_state === 'normal' ? 'bg-emerald-500' : 'bg-amber-500'" />{{ diskStateLabel(stats.disk_state) }}<small class="font-normal text-muted-foreground">{{ stats.disk_available_bytes === null ? '剩余未知' : `剩余 ${formatBytes(stats.disk_available_bytes)}` }}</small></b></article>
<article class="diagnostic-card"><Network /><span>DHT 节点</span><b>{{ stats.nodes.toLocaleString() }}</b></article>
<article class="diagnostic-card"><Gauge /><span>Metadata 下载中</span><b>{{ stats.metadata_in_flight.toLocaleString() }}</b></article>
<article class="diagnostic-card"><Database /><span>本次新收录</span><b>{{ stats.persistence_inserted.toLocaleString() }}</b></article>
<article class="diagnostic-card"><Database /><span>已保存种子</span><b>{{ stats.index.stored_torrents.toLocaleString() }}</b></article>
<article class="diagnostic-card"><Database /><span>搜索索引</span><b class="flex items-center gap-2">{{ stats.index.active_documents.toLocaleString() }}<small class="font-normal text-muted-foreground">{{ indexStateLabel(stats.index.state) }}</small></b></article>
</div>
@@ -245,6 +250,20 @@ onBeforeUnmount(() => {
</div>
<p v-if="stats.index.error" class="mt-3 text-xs text-destructive">{{ stats.index.error }}</p>
</section>
<section v-if="stats.filter.state !== 'ready'" class="mt-4 rounded-xl border bg-card p-4 shadow-sm">
<div class="flex flex-wrap items-center justify-between gap-2">
<div><h2 class="text-sm font-medium">{{ filterStateLabel(stats.filter.state) }}</h2><p class="mt-1 text-xs text-muted-foreground">详情立即使用最新规则搜索结果在后台逐步更新</p></div>
<strong class="text-sm">{{ stats.filter.progress_percent.toFixed(1) }}%</strong>
</div>
<div class="mt-3 h-1.5 overflow-hidden rounded-full bg-secondary"><div class="h-full rounded-full bg-primary transition-[width]" :style="{ width: `${Math.min(100, stats.filter.progress_percent)}%` }" /></div>
<div class="mt-3 grid grid-cols-2 gap-3 text-xs text-muted-foreground sm:grid-cols-4">
<span>内容组 <b class="ml-1 font-medium text-foreground">{{ stats.filter.total_groups.toLocaleString() }}</b></span>
<span>已扫描 <b class="ml-1 font-medium text-foreground">{{ stats.filter.scanned_groups.toLocaleString() }}</b></span>
<span>发生变化 <b class="ml-1 font-medium text-foreground">{{ stats.filter.changed_groups.toLocaleString() }}</b></span>
<span>待提交 <b class="ml-1 font-medium text-foreground">{{ stats.filter.pending_documents.toLocaleString() }}</b></span>
</div>
<p v-if="stats.filter.error" class="mt-3 text-xs text-destructive">{{ stats.filter.error }}</p>
</section>
</section>
<section class="mt-6">
@@ -257,7 +276,7 @@ onBeforeUnmount(() => {
<article class="chart-card"><div class="chart-card-header"><h2>RocksDB 内存与压缩</h2><p>SST {{ sample.storage.live_sst_bytes === null ? '未知' : formatBytes(sample.storage.live_sst_bytes) }}</p></div><VChart class="diagnostic-chart" :option="storageChart" autoresize /></article>
<article class="chart-card"><div class="chart-card-header"><h2>网络与 Peer 下载吞吐</h2></div><VChart class="diagnostic-chart" :option="throughputChart" autoresize /></article>
<article class="chart-card"><div class="chart-card-header"><h2>Peer 失败原因构成</h2><p>{{ metadataPeerSummary }}</p></div><VChart class="diagnostic-chart" :option="metadataFailureChart" autoresize /></article>
<article class="chart-card lg:col-span-2"><div class="chart-card-header"><h2>HTTP 请求与错误</h2><p>当前并发 {{ sample.http.active_requests }} · P95 {{ sample.http.latency_p95_millis === null ? '暂无' : `${sample.http.latency_p95_millis} ms` }}</p></div><VChart class="diagnostic-chart" :option="httpChart" autoresize /></article>
<article class="chart-card lg:col-span-2"><div class="chart-card-header"><h2>种子数量趋势</h2><p>{{ sample.storage.stored_torrents === null ? '等待新采样' : `当前 ${sample.storage.stored_torrents.toLocaleString()} ` }}</p></div><VChart class="diagnostic-chart" :option="torrentChart" autoresize /></article>
</div>
</section>
</template>
+20 -7
View File
@@ -9,7 +9,7 @@ import { Button } from '@/components/ui/button'
import { AppPagination } from '@/components/ui/pagination'
import { AppSelect } from '@/components/ui/select'
import { getStats, getTorrent, getVariants, search } from '@/lib/api'
import type { ContentVariants, IndexStatus, SearchHit, SearchPage, SearchSort, TorrentDetail } from '@/types/api'
import type { ContentVariants, FilterStatus, IndexStatus, SearchHit, SearchPage, SearchSort, TorrentDetail } from '@/types/api'
const query = ref('')
const submittedQuery = ref('')
@@ -30,6 +30,8 @@ const selectedHash = ref('')
const detail = ref<TorrentDetail | null>(null)
const variants = ref<ContentVariants | null>(null)
const indexStatus = ref<IndexStatus | null>(null)
const filterStatus = ref<FilterStatus | null>(null)
const detailSearchContext = ref<{ query: string; sort: SearchSort }>({ query: '', sort: 'latest' })
const supportedFilePageSizes = [25, 50, 100, 200]
const savedFilePageSize = Number(localStorage.getItem('dht-search-file-page-size'))
const detailFilePageSize = ref(supportedFilePageSizes.includes(savedFilePageSize) ? savedFilePageSize : 100)
@@ -66,6 +68,14 @@ const indexNotice = computed(() => {
if (status.state === 'initializing') return `搜索索引正在初始化${progress},当前结果可能不完整`
return `搜索索引正在后台更新${progress},新收录内容将在完成后出现`
})
const filterNotice = computed(() => {
const status = filterStatus.value
if (!status || status.state === 'ready') return ''
if (status.state === 'failed') return '过滤规则应用失败,请前往系统页面查看'
if (status.state === 'baselining') return `正在建立过滤状态基线 ${status.progress_percent.toFixed(1)}%`
if (status.state === 'applying') return `过滤规则已扫描完成,剩余 ${status.pending_documents.toLocaleString()} 个搜索文档待更新`
return `过滤规则正在后台应用 ${status.progress_percent.toFixed(1)}%,搜索结果会逐步更新`
})
function updateBrowserUrl() {
const params = new URLSearchParams()
if (submittedQuery.value) params.set('q', submittedQuery.value)
@@ -91,11 +101,13 @@ async function loadSearch() {
async function loadIndexStatus() {
try {
indexStatus.value = (await getStats()).index
const stats = await getStats()
indexStatus.value = stats.index
filterStatus.value = stats.filter
} catch {
// 使
} finally {
const delay = indexStatus.value?.state === 'ready' ? 30_000 : 5_000
const delay = indexStatus.value?.state === 'ready' && filterStatus.value?.state === 'ready' ? 30_000 : 5_000
indexTimer = window.setTimeout(loadIndexStatus, delay)
}
}
@@ -135,6 +147,7 @@ function resetSearch() {
}
async function openDetail(hit: SearchHit) {
detailSearchContext.value = { query: submittedQuery.value, sort: sort.value }
selectedHash.value = hit.info_hash
detailOpen.value = true
detailLoading.value = true
@@ -145,7 +158,7 @@ async function openDetail(hit: SearchHit) {
detailController?.abort()
detailController = new AbortController()
try {
const [torrent, contentVariants] = await Promise.all([getTorrent(hit.info_hash, 0, detailFilePageSize.value, detailController.signal), getVariants(hit.content_key, detailController.signal)])
const [torrent, contentVariants] = await Promise.all([getTorrent(hit.info_hash, 0, detailFilePageSize.value, detailSearchContext.value.query, detailSearchContext.value.sort, detailController.signal), getVariants(hit.content_key, detailController.signal)])
detail.value = torrent
variants.value = contentVariants
} catch (cause) {
@@ -163,7 +176,7 @@ async function loadFilePage(fileOffset: number) {
detailFilesError.value = ''
const infoHash = detail.value.info_hash
try {
const next = await getTorrent(infoHash, fileOffset, detailFilePageSize.value, filesController.signal)
const next = await getTorrent(infoHash, fileOffset, detailFilePageSize.value, detailSearchContext.value.query, detailSearchContext.value.sort, filesController.signal)
if (selectedHash.value === infoHash) detail.value = next
} catch (cause) {
if ((cause as Error).name !== 'AbortError') detailFilesError.value = cause instanceof Error ? cause.message : '文件列表加载失败'
@@ -216,9 +229,9 @@ onBeforeUnmount(() => {
</div>
</form>
<RouterLink v-if="indexNotice" class="mt-3 flex items-center gap-2 rounded-lg border border-amber-500/25 bg-amber-500/8 px-3 py-2 text-sm text-amber-800 transition-colors hover:bg-amber-500/12 dark:text-amber-300" to="/system">
<RouterLink v-if="indexNotice || filterNotice" class="mt-3 flex items-center gap-2 rounded-lg border border-amber-500/25 bg-amber-500/8 px-3 py-2 text-sm text-amber-800 transition-colors hover:bg-amber-500/12 dark:text-amber-300" to="/system">
<AlertTriangle class="size-4 shrink-0" />
<span>{{ indexNotice }}</span>
<span>{{ indexNotice || filterNotice }}</span>
</RouterLink>
<div v-if="loading" class="flex min-h-72 items-center justify-center text-sm text-muted-foreground"><LoaderCircle class="mr-2 size-5 animate-spin" />正在搜索</div>
+17 -3
View File
@@ -61,6 +61,7 @@ export interface TorrentDetail {
name: string
total_size: number
file_count: number
visible_file_count: number
file_offset: number
file_limit: number
files: TorrentFile[]
@@ -133,13 +134,14 @@ export interface ServiceStats {
persistence_queue: number
indexed_documents: number
index: IndexStatus
filter: FilterStatus
verification_queue: number
verification_succeeded: number
verification_failed: number
}
export type IndexState = 'ready' | 'initializing' | 'rebuilding' | 'switching' | 'blocked' | 'failed'
export type IndexRebuildReason = 'initial' | 'content_filter' | 'schema' | 'document_format' | 'missing' | 'corrupt'
export type IndexRebuildReason = 'initial' | 'schema' | 'document_format' | 'missing' | 'corrupt'
export interface IndexStatus {
state: IndexState
@@ -156,6 +158,17 @@ export interface IndexStatus {
error: string | null
}
export type FilterState = 'baselining' | 'scanning' | 'applying' | 'ready' | 'failed'
export interface FilterStatus {
state: FilterState
total_groups: number
scanned_groups: number
changed_groups: number
pending_documents: number
progress_percent: number
error: string | null
}
export interface ProcessDiagnostics {
resident_memory_bytes: number | null
private_memory_bytes: number | null
@@ -165,6 +178,7 @@ export interface ProcessDiagnostics {
}
export interface StorageDiagnostics {
stored_torrents: number | null
block_cache_bytes: number | null
memtable_bytes: number | null
pending_compaction_bytes: number | null
@@ -241,8 +255,8 @@ export interface DiagnosticHistory {
export type NetworkMode = 'ipv4-only' | 'ipv6-only' | 'dual-stack'
export type LogRotation = 'minutely' | 'hourly' | 'daily' | 'never'
export interface ContentFilterConfig {
file_name_patterns: string[]
file_path_patterns: string[]
torrent_name_patterns: string[]
file_patterns: string[]
}
export interface AppConfigDto {