diff --git a/Cargo.lock b/Cargo.lock index 2b9f8bc..5e2c115 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -658,9 +658,11 @@ dependencies = [ "dht-crawler", "dunce", "fs2", + "getrandom 0.4.3", "hex", "libc", "regex", + "reqwest", "rmp-serde", "rocksdb", "rusqlite", diff --git a/Dockerfile b/Dockerfile index c67eeea..4447fab 100644 --- a/Dockerfile +++ b/Dockerfile @@ -66,6 +66,7 @@ RUN groupadd --gid 10001 dht-search \ dht-search \ && install -d -o 10001 -g 10001 \ /dht-search \ + /dht-search/collector \ /dht-search/data \ /dht-search/web @@ -76,6 +77,7 @@ COPY --chmod=0644 --chown=10001:10001 config.toml /dht-search/config.toml USER 10001:10001 EXPOSE 8080/tcp +EXPOSE 8081/tcp EXPOSE 12313/udp STOPSIGNAL SIGTERM diff --git a/README.md b/README.md index 0349bae..369ecc2 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ 这是一个用于持续发现持久化索引和搜索 BitTorrent DHT 元数据的 Rust 服务 -当前已经具备可即时启停的 DHT 采集 Metadata 下载 RocksDB 精确去重 Tantivy 全文搜索 内容聚合 可用性验证 HTTP API Web 搜索界面 运行诊断 配置管理 备份恢复和 Docker 部署能力 +当前已经具备可即时启停的分布式 DHT 采集 Metadata 下载 RocksDB 精确去重 Tantivy 全文搜索 内容聚合 可用性验证 HTTP API Web 搜索界面 运行诊断 配置管理 备份恢复和 Docker 部署能力 ## 项目结构 @@ -21,7 +21,34 @@ Tantivy 使用代际影子索引处理 Schema 文档格式损坏和缺失等全 应用全部配置和内容隐藏规则统一位于 [`config.toml`](config.toml) -Web 右上角的无线电图标可以即时停止或恢复 DHT 持续采集 状态会写回 `dht.enabled` 关闭后不会建立 DHT 和 Metadata 网络任务 但现有 RocksDB 数据仍会继续补建索引并提供本地搜索 +诊断页默认聚合全部采集器的 DHT 下载和失败原因指标 也可以切换到单个采集器查看其历史吞吐和进程资源 独立的采集器页面负责节点状态启停与每节点 DHT 配置 关闭采集器不会影响已有 RocksDB 数据补建索引和本地搜索 + +## 分布式运行 + +同一个程序通过 `[service].role` 支持三种运行角色 + +| 角色 | 行为 | +|---|---| +| `standalone` | 默认单机模式 同时运行存储搜索和本机采集器 | +| `coordinator` | 唯一存储搜索协调器 不启动 DHT | +| `collector` | 只运行 DHT 采集有效性验证和持久待发送箱 | + +协调器和采集器必须使用相同的 `DHT_SEARCH_CLUSTER_TOKEN` 环境变量 值至少三十二个字符 内部接口使用独立端口且不应通过公开 Caddy 站点暴露 + +采集器第一次连接时自动生成并持久化节点 ID 协调器保存每个节点的期望配置和修订号 Metadata 在采集器本地 RocksDB 待发送箱落盘后才确认下载成功 协调器断开或积压达到高水位时采集器会暂停并在恢复后继续 采集器心跳同时上报 DHT Peer 失败分类队列和当前进程资源 协调器把集群汇总与逐节点快照写入可删除的诊断历史 + +本地启动一个协调器和两个采集器 + +```shell +export DHT_SEARCH_CLUSTER_TOKEN="请替换为至少三十二个字符的随机令牌" +docker compose -f compose.cluster.yaml up -d --build +``` + +PowerShell 使用 `$env:DHT_SEARCH_CLUSTER_TOKEN = "请替换为至少三十二个字符的随机令牌"` + +打开 `http://127.0.0.1:18080/system` 查看两个节点 再到配置页分别启用和修改参数 + +内部协议使用批量 infohash 租约避免多个采集器重复下载 RocksDB 仍是唯一权威数据源 租约丢失或上传重试只影响额外网络开销 不会生成重复记录 ## 本地运行 @@ -41,6 +68,7 @@ scripts\run.bat | HTTP API | `http://127.0.0.1:8080` | | 健康检查 | `http://127.0.0.1:8080/health` | | 就绪检查 | `http://127.0.0.1:8080/ready` | +| 内部采集接口 | `127.0.0.1:8081` | 也可以分别启动 diff --git a/TODOS.md b/TODOS.md index 598dedb..c3d9e99 100644 --- a/TODOS.md +++ b/TODOS.md @@ -84,6 +84,7 @@ - [ ] 在应用启动前检查数据目录配置文件和关键子目录的读写权限 - [ ] 使用同一版本完成二十四小时连续运行 - [ ] 使用同一版本完成七天连续运行 +- [ ] 在真实独立设备上部署一个协调器和至少两个采集器完成二十四小时小流量验收 - [ ] 记录连续运行期间的私有内存队列深度 Metadata 成功率磁盘增长和索引提交状态 - [ ] 在 `README.md` 补充简洁的备份恢复和常见故障排查步骤 @@ -96,8 +97,4 @@ ## 当前下一步 -部署紧凑文件名 Schema 并观察远端影子索引的重建进度峰值磁盘占用最终体积和常用查询延迟 - -同时继续观察六小时动态状态分桶 使用诊断中的安排刷新抑制刷新实际写入和跳过写入数量确认效果 - -本次 Schema 更新已经删除停用字段和重复代表标题 后续动态热度字段与稳定排序键需要评估是否能够避免再次改变 Schema +在真实独立设备上小流量部署分布式采集 验证隧道断开恢复每节点资源占用待发送箱增长和有效性验证任务转移 diff --git a/compose.cluster.yaml b/compose.cluster.yaml new file mode 100644 index 0000000..8361596 --- /dev/null +++ b/compose.cluster.yaml @@ -0,0 +1,53 @@ +name: dht-search-cluster + +services: + coordinator: + image: ${DHT_CLUSTER_IMAGE:-dht-search:cluster-dev} + build: + context: . + dockerfile: Dockerfile + init: true + command: ["--config", "/dht-search/config.toml", "--role", "coordinator", "--http-listen", "0.0.0.0:8080", "--web-dir", "/dht-search/web", "--ingest-listen", "0.0.0.0:8081"] + environment: + DHT_SEARCH_CLUSTER_TOKEN: ${DHT_SEARCH_CLUSTER_TOKEN:?请设置至少三十二个字符的集群令牌} + ports: + - "127.0.0.1:18080:8080/tcp" + volumes: + - ./config.toml:/dht-search/config.toml + - cluster-coordinator-data:/dht-search/data + restart: unless-stopped + + collector-a: + image: ${DHT_CLUSTER_IMAGE:-dht-search:cluster-dev} + init: true + command: ["--config", "/dht-search/config.toml", "--role", "collector", "--coordinator-url", "http://coordinator:8081", "--collector-name", "采集器 A", "--collector-state-dir", "/dht-search/collector", "--no-file-logging"] + environment: + DHT_SEARCH_CLUSTER_TOKEN: ${DHT_SEARCH_CLUSTER_TOKEN:?请设置至少三十二个字符的集群令牌} + ports: + - "12314:12313/udp" + volumes: + - ./config.toml:/dht-search/config.toml:ro + - cluster-collector-a:/dht-search/collector + depends_on: + - coordinator + restart: unless-stopped + + collector-b: + image: ${DHT_CLUSTER_IMAGE:-dht-search:cluster-dev} + init: true + command: ["--config", "/dht-search/config.toml", "--role", "collector", "--coordinator-url", "http://coordinator:8081", "--collector-name", "采集器 B", "--collector-state-dir", "/dht-search/collector", "--no-file-logging"] + environment: + DHT_SEARCH_CLUSTER_TOKEN: ${DHT_SEARCH_CLUSTER_TOKEN:?请设置至少三十二个字符的集群令牌} + ports: + - "12315:12313/udp" + volumes: + - ./config.toml:/dht-search/config.toml:ro + - cluster-collector-b:/dht-search/collector + depends_on: + - coordinator + restart: unless-stopped + +volumes: + cluster-coordinator-data: + cluster-collector-a: + cluster-collector-b: diff --git a/config.toml b/config.toml index 322a8e0..6192e1f 100644 --- a/config.toml +++ b/config.toml @@ -4,6 +4,23 @@ stats_interval_secs = 10 index_batch_size = 1024 index_interval_millis = 5000 +[service] +role = "standalone" + +[coordinator] +ingest_listen = "127.0.0.1:8081" +lease_secs = 120 +offline_after_secs = 15 +max_leases = 1000000 + +[collector] +coordinator_url = "http://127.0.0.1:8081" +name = "本机采集器" +state_dir = "data/collector" +outbox_max_bytes = 2147483648 +outbox_max_items = 100000 +heartbeat_secs = 2 + [content_filter] torrent_name_patterns = ["*【加QQ *】*"] file_patterns = [ diff --git a/src/crawler/src/scheduler.rs b/src/crawler/src/scheduler.rs index 098a0cb..8ce0a98 100644 --- a/src/crawler/src/scheduler.rs +++ b/src/crawler/src/scheduler.rs @@ -14,6 +14,7 @@ use metrics::{counter, gauge, histogram}; use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::panic::{AssertUnwindSafe, catch_unwind}; +use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::time::{Duration, Instant}; @@ -35,8 +36,13 @@ mod pending_queue; use peer_race::race_peer_fetches; use pending_queue::{PeerCandidate, PendingHashQueue, QueuePushKind, QueuedHash}; -/// Callback returning whether the application accepted a downloaded torrent. -pub type TorrentAckCallback = Box bool + Send + Sync + 'static>; +/// Callback returning asynchronously whether the application durably accepted a downloaded torrent. +pub type TorrentAckCallback = Box< + dyn Fn(TorrentInfo) -> Pin + Send>> + + Send + + Sync + + 'static, +>; /// Callback invoked once when an admitted Metadata job reaches a terminal state. pub type MetadataCompletionCallback = Box; /// Asynchronous InfoHash admission callback. @@ -550,9 +556,14 @@ impl MetadataScheduler { .as_secs(), }; - let accepted = callback.load_full().is_some_and(|callback| { - catch_unwind(AssertUnwindSafe(|| callback(metadata))).unwrap_or(false) - }); + let accepted = if let Some(callback) = callback.load_full() { + match catch_unwind(AssertUnwindSafe(|| callback(metadata))) { + Ok(future) => future.await, + Err(_) => false, + } + } else { + false + }; let status = if accepted { MetadataFetchCompletionStatus::Accepted } else { diff --git a/src/crawler/src/server.rs b/src/crawler/src/server.rs index 273cc9e..3f51050 100644 --- a/src/crawler/src/server.rs +++ b/src/crawler/src/server.rs @@ -298,9 +298,13 @@ impl DHTServer { where F: Fn(TorrentInfo) + Send + Sync + 'static, { + let callback = Arc::new(callback); let callback: Arc = Arc::new(Box::new(move |torrent| { - callback(torrent); - true + let callback = callback.clone(); + Box::pin(async move { + callback(torrent); + true + }) })); self.torrent_callback.store(Some(callback)); } @@ -316,7 +320,22 @@ impl DHTServer { where F: Fn(TorrentInfo) -> bool + Send + Sync + 'static, { - let callback: Arc = Arc::new(Box::new(callback)); + let callback = Arc::new(callback); + let callback: Arc = Arc::new(Box::new(move |torrent| { + let callback = callback.clone(); + Box::pin(async move { callback(torrent) }) + })); + self.torrent_callback.store(Some(callback)); + } + + /// Registers an asynchronous callback that acknowledges durable application delivery. + pub fn on_torrent_with_async_ack(&self, callback: F) + where + F: Fn(TorrentInfo) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + let callback: Arc = + Arc::new(Box::new(move |torrent| Box::pin(callback(torrent)))); self.torrent_callback.store(Some(callback)); } diff --git a/src/search/Cargo.toml b/src/search/Cargo.toml index ffb855a..c581414 100644 --- a/src/search/Cargo.toml +++ b/src/search/Cargo.toml @@ -21,7 +21,9 @@ dht-crawler = { path = "../crawler", features = ["metrics"] } dunce = "1.0" hex = "0.4" fs2 = "0.4" +getrandom = "0.4" rocksdb = { version = "0.24.0", default-features = false, features = ["bindgen-runtime", "lz4"], optional = true } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } rmp-serde = "1.3" regex = "1.12" rusqlite = { version = "0.40.2", features = ["bundled"], optional = true } diff --git a/src/search/README.md b/src/search/README.md index cfa394f..977d5c8 100644 --- a/src/search/README.md +++ b/src/search/README.md @@ -163,7 +163,7 @@ SQLite 使用 WAL 和单独 writer 线程 原始采样超过 24 小时自动删 除 `dht.enabled` 外当前版本不在线修改正在运行的 DHT 存储索引和监听器 保存其他配置后返回 `restart_required = true` 并在重启服务后统一生效 命令行覆盖字段也会单独返回并继续优先于文件配置 -Web 右上角采集开关通过 `/crawler` 即时停止或重新创建 DHT 运行时并把状态持久化到 `dht.enabled` 关闭时同步停止 Metadata 下载和按需有效性验证 RocksDB Tantivy HTTP 和 Web 保持运行 +诊断页通过 `/collectors` 选择集群总览或单个采集器并展示对应 DHT Peer 和进程指标 独立采集器页面负责按节点启停与修改配置 每个节点拥有独立修订号并由协调器集中持久化 通过 Web 保存会按 DTO 重新生成 TOML 原有手写注释不会保留 管理接口默认随 HTTP 服务提供 因此生产部署不应把 `/config` 暴露到不受信任的公网入口 @@ -279,7 +279,7 @@ GET /torrents/{infohash}?file_offset=0&file_limit=100 详情文件列表默认返回 100 条且单次最多 200 条 使用 `file_offset` 翻页避免超大种子一次向浏览器返回全部文件 -生产 Web 页面使用 `/` `/system` 和 `/settings` 三个路由 分别提供搜索运行诊断和配置管理 API 路径继续保持独立避免单页回退冲突 +生产 Web 页面使用 `/` `/system` `/workers` 和 `/settings` 四个路由 分别提供搜索运行诊断采集器管理和服务配置 API 路径继续保持独立避免单页回退冲突 ## 数据恢复 diff --git a/src/search/src/api/handlers.rs b/src/search/src/api/handlers.rs index 1df0ce8..fbd70b3 100644 --- a/src/search/src/api/handlers.rs +++ b/src/search/src/api/handlers.rs @@ -19,15 +19,13 @@ use axum::{ use super::{ ApiState, - request::{ - ContentVariantsRequest, CrawlerUpdateRequest, DiagnosticHistoryRequest, SearchRequest, - TorrentRequest, - }, + request::{ContentVariantsRequest, DiagnosticHistoryRequest, SearchRequest, TorrentRequest}, response::{ - ContentVariantsResponse, CrawlerStatusResponse, ErrorResponse, StatsResponse, - StatusResponse, TorrentResponse, TorrentVariantResponse, + ContentVariantsResponse, ErrorResponse, StatsResponse, StatusResponse, TorrentResponse, + TorrentVariantResponse, }, }; +use crate::cluster::protocol::{CollectorConfigUpdate, CollectorView}; use crate::config::{ConfigServiceError, ConfigSnapshot, ConfigUpdateRequest}; use crate::diagnostics::{CurrentDiagnosticsResponse, DiagnosticHistory, HistoryResolution}; @@ -56,6 +54,17 @@ pub(crate) async fn stats(State(state): State) -> Json let refresh_diagnostics = state.repository.index_refresh_diagnostics(); let search_diagnostics = state.search.diagnostics(); let index = state.search.status(storage); + let collector_metrics = state + .coordinator + .as_ref() + .and_then(|coordinator| { + coordinator + .collectors() + .ok() + .map(|nodes| (coordinator, nodes)) + }) + .filter(|(_, nodes)| !nodes.is_empty()) + .and_then(|(coordinator, _)| coordinator.aggregate_metrics().ok()); Json(StatsResponse { http_active_requests: http.active_requests, http_requests: http.requests, @@ -80,8 +89,16 @@ pub(crate) async fn stats(State(state): State) -> Json disk_probe_failures: disk.probe_failures, disk_transitions: disk.transitions, disk_rejected_new_work: disk.rejected_new_work, - nodes: dht.node_pool_size, - udp_tx_packets: observability.udp_tx_packets, + nodes: collector_metrics + .as_ref() + .map_or(dht.node_pool_size, |metrics| { + metrics.nodes.min(usize::MAX as u64) as usize + }), + udp_tx_packets: collector_metrics + .as_ref() + .map_or(observability.udp_tx_packets, |metrics| { + metrics.udp_tx_packets + }), find_node_queries: dht .queries_new .saturating_add(dht.queries_revisit) @@ -104,29 +121,97 @@ pub(crate) async fn stats(State(state): State) -> Json sampled_hashes_filtered: dht.sample_infohashes_hashes_filtered, sampled_hashes_duplicate: dht.sample_infohashes_hashes_duplicate, sampled_hashes_dropped: dht.sample_infohashes_hashes_dropped, - metadata_peer_attempts: dht.metadata_peer_attempts, - metadata_in_flight: dht.metadata_in_flight, - metadata_ok: dht.metadata_peer_succeeded, + metadata_peer_attempts: collector_metrics + .as_ref() + .map_or(dht.metadata_peer_attempts, |metrics| { + metrics.metadata_peer_attempts + }), + metadata_in_flight: collector_metrics + .as_ref() + .map_or(dht.metadata_in_flight, |metrics| { + metrics.metadata_in_flight.min(usize::MAX as u64) as usize + }), + metadata_ok: collector_metrics + .as_ref() + .map_or(dht.metadata_peer_succeeded, |metrics| { + metrics.metadata_succeeded + }), metadata_ok_from_announce: dht.metadata_success_from_announce, metadata_ok_from_sample_direct: dht.metadata_success_from_sample_direct, metadata_ok_from_sample_snapshot: dht.metadata_success_from_sample_snapshot, metadata_ok_from_active_lookup: dht.metadata_success_from_active_lookup, - metadata_failed: dht.metadata_peer_failed, - metadata_peer_canceled: dht.metadata_peer_canceled, - metadata_peer_failure_cache_hits: dht.metadata_peer_failure_cache_hits, - metadata_peer_failure_cache_entries: dht.metadata_peer_failure_cache_entries, - metadata_failure_timeout: observability.metadata_failure_timeout, - metadata_failure_connect: observability.metadata_failure_connect, - metadata_failure_no_extension: observability.metadata_failure_no_extension, - metadata_failure_send: observability.metadata_failure_send, - metadata_failure_size_limit: observability.metadata_failure_size_limit, - metadata_failure_sha1: observability.metadata_failure_sha1, - metadata_failure_parse: observability.metadata_failure_parse, - metadata_failure_other: observability.metadata_failure_other, - metadata_filtered: observability - .metadata_failure_size_limit + metadata_failed: collector_metrics + .as_ref() + .map_or(dht.metadata_peer_failed, |metrics| metrics.metadata_failed), + metadata_peer_canceled: collector_metrics + .as_ref() + .map_or(dht.metadata_peer_canceled, |metrics| { + metrics.metadata_peer_canceled + }), + metadata_peer_failure_cache_hits: collector_metrics + .as_ref() + .map_or(dht.metadata_peer_failure_cache_hits, |metrics| { + metrics.metadata_peer_failure_cache_hits + }), + metadata_peer_failure_cache_entries: collector_metrics.as_ref().map_or( + dht.metadata_peer_failure_cache_entries, + |metrics| { + metrics + .metadata_peer_failure_cache_entries + .min(usize::MAX as u64) as usize + }, + ), + metadata_failure_timeout: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_timeout, |metrics| { + metrics.metadata_failure_timeout + }), + metadata_failure_connect: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_connect, |metrics| { + metrics.metadata_failure_connect + }), + metadata_failure_no_extension: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_no_extension, |metrics| { + metrics.metadata_failure_no_extension + }), + metadata_failure_send: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_send, |metrics| { + metrics.metadata_failure_send + }), + metadata_failure_size_limit: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_size_limit, |metrics| { + metrics.metadata_failure_size_limit + }), + metadata_failure_sha1: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_sha1, |metrics| { + metrics.metadata_failure_sha1 + }), + metadata_failure_parse: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_parse, |metrics| { + metrics.metadata_failure_parse + }), + metadata_failure_other: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_other, |metrics| { + metrics.metadata_failure_other + }), + metadata_filtered: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_size_limit, |metrics| { + metrics.metadata_failure_size_limit + }) .saturating_add(filtered.total()), - metadata_filtered_too_large: observability.metadata_failure_size_limit, + metadata_filtered_too_large: collector_metrics + .as_ref() + .map_or(observability.metadata_failure_size_limit, |metrics| { + metrics.metadata_failure_size_limit + }), metadata_filtered_invalid_info_hash: filtered .count(MetadataRejectionReason::InvalidInfoHash), metadata_filtered_empty_name: filtered.count(MetadataRejectionReason::EmptyName), @@ -166,42 +251,83 @@ pub(crate) async fn stats(State(state): State) -> Json }) } -pub(crate) async fn crawler_status(State(state): State) -> Json { - let status = state.crawler.status(); - Json(CrawlerStatusResponse { - enabled: status.enabled, - transitioning: status.transitioning, - }) +pub(crate) async fn collectors( + State(state): State, +) -> Result>, ApiError> { + let coordinator = state + .coordinator + .as_ref() + .ok_or_else(|| ApiError::bad_request("当前运行模式没有启用分布式协调器"))?; + coordinator + .collectors() + .map(Json) + .map_err(ApiError::internal) } -pub(crate) async fn crawler_update( +pub(crate) async fn collector_update( State(state): State, - Json(request): Json, -) -> Result, ApiError> { - let previous = state.crawler.status().enabled; - if previous != request.enabled { - state - .crawler - .set_enabled(request.enabled) + Path(node_id): Path, + Json(update): Json, +) -> Result, ApiError> { + let coordinator = state + .coordinator + .as_ref() + .ok_or_else(|| ApiError::bad_request("当前运行模式没有启用分布式协调器"))?; + let previous_collector = coordinator + .collectors() + .map_err(ApiError::internal)? + .into_iter() + .find(|collector| collector.node_id == node_id) + .ok_or_else(|| ApiError::not_found("没有找到该采集器"))?; + let updated = coordinator + .update_collector(&node_id, update) + .map_err(|error| { + if error.contains("已经被其他请求修改") { + ApiError::conflict(error) + } else { + ApiError::internal(error) + } + })? + .ok_or_else(|| ApiError::not_found("没有找到该采集器"))?; + if state.local_collector_id.as_deref() == Some(node_id.as_str()) { + let previous = state.crawler.status().enabled; + if previous != updated.desired.dht.enabled + && let Err(error) = state.crawler.set_enabled(updated.desired.dht.enabled).await + { + rollback_collector(coordinator, &updated, &previous_collector); + return Err(ApiError::internal(error.to_string())); + } + let config = state.config.clone(); + let dht = updated.desired.dht.clone(); + if let Err(error) = tokio::task::spawn_blocking(move || config.set_dht_config(dht)) .await - .map_err(|error| ApiError::internal(format!("切换 DHT 采集状态失败: {error}")))?; - } - let config = state.config.clone(); - let enabled = request.enabled; - let persisted = tokio::task::spawn_blocking(move || config.set_dht_enabled(enabled)) - .await - .map_err(|error| ApiError::internal(format!("采集状态保存任务失败: {error}")))?; - if let Err(error) = persisted { - if previous != request.enabled { - let _ = state.crawler.set_enabled(previous).await; + .map_err(|error| ApiError::internal(error.to_string()))? + { + if previous != updated.desired.dht.enabled { + let _ = state.crawler.set_enabled(previous).await; + } + rollback_collector(coordinator, &updated, &previous_collector); + return Err(ApiError::internal(error.to_string())); } - return Err(ApiError::internal(error.to_string())); } - let status = state.crawler.status(); - Ok(Json(CrawlerStatusResponse { - enabled: status.enabled, - transitioning: status.transitioning, - })) + Ok(Json(updated)) +} + +fn rollback_collector( + coordinator: &crate::cluster::coordinator::CoordinatorRuntime, + updated: &CollectorView, + previous: &CollectorView, +) { + if let Err(error) = coordinator.update_collector( + &updated.node_id, + CollectorConfigUpdate { + revision: updated.revision, + name: previous.name.clone(), + desired: previous.desired.clone(), + }, + ) { + tracing::error!(%error, node_id = %updated.node_id, "回滚本机采集器期望配置失败"); + } } pub(crate) async fn diagnostics_current( diff --git a/src/search/src/api/mod.rs b/src/search/src/api/mod.rs index 84885a3..5f2bd84 100644 --- a/src/search/src/api/mod.rs +++ b/src/search/src/api/mod.rs @@ -37,6 +37,8 @@ pub(crate) struct ApiState { pub(crate) config: ConfigService, pub(crate) filter: crate::filter_worker::FilterRuntime, pub(crate) http_stats: HttpStats, + pub(crate) coordinator: Option, + pub(crate) local_collector_id: Option, } pub(crate) async fn serve( @@ -60,9 +62,10 @@ fn router(state: ApiState, web_dir: PathBuf) -> Router { .route("/health", get(handlers::health)) .route("/ready", get(handlers::ready)) .route("/stats", get(handlers::stats)) + .route("/collectors", get(handlers::collectors)) .route( - "/crawler", - get(handlers::crawler_status).put(handlers::crawler_update), + "/collectors/{node_id}", + axum::routing::put(handlers::collector_update), ) .route("/diagnostics/current", get(handlers::diagnostics_current)) .route("/diagnostics/history", get(handlers::diagnostics_history)) @@ -181,6 +184,8 @@ mod tests { .unwrap(), filter: crate::filter_worker::FilterRuntime::for_test(repository.clone()), http_stats: HttpStats::default(), + coordinator: None, + local_collector_id: None, }, web_dir, ); @@ -236,23 +241,6 @@ mod tests { .is_some_and(|value| value >= 3) ); - let response = app - .clone() - .oneshot( - Request::builder() - .uri("/crawler") - .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["enabled"], false); - assert_eq!(json["transitioning"], false); - let response = app .clone() .oneshot( diff --git a/src/search/src/api/request.rs b/src/search/src/api/request.rs index 08d9707..b949e61 100644 --- a/src/search/src/api/request.rs +++ b/src/search/src/api/request.rs @@ -6,11 +6,6 @@ use crate::{ }; use serde::Deserialize; -#[derive(Debug, Deserialize)] -pub(crate) struct CrawlerUpdateRequest { - pub(crate) enabled: bool, -} - fn default_limit() -> usize { 20 } diff --git a/src/search/src/api/response.rs b/src/search/src/api/response.rs index 114d47b..cbd93ae 100644 --- a/src/search/src/api/response.rs +++ b/src/search/src/api/response.rs @@ -18,12 +18,6 @@ pub(crate) struct ErrorResponse { pub(crate) error: String, } -#[derive(Debug, Serialize)] -pub(crate) struct CrawlerStatusResponse { - pub(crate) enabled: bool, - pub(crate) transitioning: bool, -} - #[derive(Debug, Serialize)] pub(crate) struct StatsResponse { pub(crate) http_active_requests: u64, diff --git a/src/search/src/app.rs b/src/search/src/app.rs index f8d9201..17deebb 100644 --- a/src/search/src/app.rs +++ b/src/search/src/app.rs @@ -11,7 +11,7 @@ use tokio_util::sync::CancellationToken; use crate::{ api::{self, ApiState}, backup::{self, BackupStats}, - config::{AppConfig, ConfigService}, + config::{AppConfig, ConfigService, ServiceRole}, crawler::{pipeline::PersistencePipeline, runtime::CrawlerRuntime}, diagnostics::{DiagnosticSources, DiagnosticsRuntime, HttpStats}, disk_guard::{self, DiskGuard}, @@ -32,6 +32,25 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res metadata_limits.rule_id(), content_filter.clone(), )?); + let coordinator_token = match config.service.role { + ServiceRole::Coordinator => Some(crate::cluster::cluster_token()?), + ServiceRole::Standalone => std::env::var(crate::cluster::CLUSTER_TOKEN_ENV).ok(), + ServiceRole::Collector => None, + }; + let coordinator = if config.service.role == ServiceRole::Collector { + None + } else { + Some(crate::cluster::coordinator::CoordinatorRuntime::open( + &config.data_dir, + repository.clone(), + metadata_limits, + &config.coordinator, + config.verification.lease_secs, + coordinator_token + .as_deref() + .unwrap_or("standalone-local-control-token"), + )?) + }; let search_bootstrap = SearchRuntime::open(&config.data_dir)?; let search = search_bootstrap.runtime.clone(); disk_guard.probe(&config.data_dir, 0); @@ -77,12 +96,112 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res disk_guard.clone(), config.verification.clone(), ); - if config.dht.enabled { + if config.service.role == ServiceRole::Coordinator && config.verification.enabled { + crawler.install_verification(crate::verification::VerificationIngress::queue_only( + repository.clone(), + config.verification.queue_capacity, + disk_guard.clone(), + )); + } + if config.service.role == ServiceRole::Standalone && config.dht.enabled { crawler.set_enabled(true).await?; } + let local_collector_id = if config.service.role == ServiceRole::Standalone { + let digest = blake3::hash(config.data_dir.to_string_lossy().as_bytes()); + let node_id = hex::encode(&digest.as_bytes()[..16]); + let coordinator = coordinator.as_ref().expect("standalone coordinator"); + coordinator + .register_local( + node_id.clone(), + config.collector.name.clone(), + crate::cluster::protocol::CollectorDesiredConfig { + dht: config.dht.clone(), + verification_enabled: config.verification.enabled, + verification_max_active: config.verification.max_active, + }, + ) + .map_err(AppError::Cluster)?; + Some(node_id) + } else { + None + }; + + let coordinator_cancel = CancellationToken::new(); + let coordinator_task = coordinator + .as_ref() + .filter(|_| config.service.role == ServiceRole::Coordinator || coordinator_token.is_some()) + .map(|runtime| { + let runtime = runtime.clone(); + let listen = config.coordinator.ingest_listen; + let cancel = coordinator_cancel.clone(); + tokio::spawn(async move { runtime.serve(listen, cancel).await }) + }); + let local_heartbeat_task = local_collector_id.as_ref().map(|node_id| { + let node_id = node_id.clone(); + let coordinator = coordinator.clone().expect("standalone coordinator"); + let crawler = crawler.clone(); + let persistence = persistence.ingress.clone(); + let cancel = coordinator_cancel.clone(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(Duration::from_secs(2)); + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = ticker.tick() => { + let dht = crawler.stats(); + let snapshot = dht.snapshot(); + let observability = dht.observability_snapshot(); + let process = crate::diagnostics::process::snapshot(); + let persistence = persistence.snapshot(); + let revision = coordinator + .collectors() + .ok() + .and_then(|collectors| collectors.into_iter().find(|item| item.node_id == node_id)) + .map_or(0, |item| item.revision); + coordinator.heartbeat_local( + node_id.clone(), + revision, + crate::cluster::protocol::CollectorMetrics { + state: if crawler.status().enabled { "running" } else { "disabled" }.to_owned(), + nodes: snapshot.node_pool_size as u64, + metadata_in_flight: snapshot.metadata_in_flight as u64, + metadata_succeeded: snapshot.metadata_peer_succeeded, + metadata_failed: snapshot.metadata_peer_failed, + udp_tx_packets: observability.udp_tx_packets, + metadata_peer_attempts: snapshot.metadata_peer_attempts, + metadata_peer_canceled: snapshot.metadata_peer_canceled, + metadata_peer_failure_cache_hits: snapshot.metadata_peer_failure_cache_hits, + metadata_peer_failure_cache_entries: snapshot.metadata_peer_failure_cache_entries as u64, + metadata_failure_timeout: observability.metadata_failure_timeout, + metadata_failure_connect: observability.metadata_failure_connect, + metadata_failure_no_extension: observability.metadata_failure_no_extension, + metadata_failure_send: observability.metadata_failure_send, + metadata_failure_size_limit: observability.metadata_failure_size_limit, + metadata_failure_sha1: observability.metadata_failure_sha1, + metadata_failure_parse: observability.metadata_failure_parse, + metadata_failure_other: observability.metadata_failure_other, + sample_queue_depth: snapshot.sample_candidate_queue_depth as u64, + sample_queue_capacity: snapshot.sample_candidate_queue_capacity as u64, + resident_memory_bytes: process.resident_memory_bytes, + private_memory_bytes: process.private_memory_bytes, + cpu_time_millis: process.cpu_time_millis, + thread_count: process.thread_count, + handle_count: process.handle_count, + outbox_items: persistence.queue_depth as u64, + outbox_bytes: 0, + last_error: None, + }, + ); + } + } + } + }) + }); + tracing::info!( dht_enabled = config.dht.enabled, + role = ?config.service.role, dht_port = config.dht.port, data_dir = %config.data_dir.display(), persistence_queue_capacity = config.persistence_queue_capacity, @@ -121,6 +240,7 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res persistence: persistence.ingress.clone(), disk_guard: disk_guard.clone(), http: http_stats.clone(), + coordinator: coordinator.clone(), }, ) { Ok(runtime) => runtime, @@ -147,6 +267,8 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res config: config_service, filter: filter_runtime, http_stats, + coordinator: coordinator.clone(), + local_collector_id: local_collector_id.clone(), }, api_cancel.clone(), )); @@ -175,6 +297,7 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res let mut shutdown_error = None; diagnostics.request_shutdown(); + coordinator_cancel.cancel(); filter_cancel.cancel(); let _ = filter_task.await; backup_cancel.cancel(); @@ -229,6 +352,16 @@ pub(crate) async fn run(config: AppConfig, config_service: ConfigService) -> Res } tracing::error!(%error, "关闭阶段发生附加错误"); } + if let Some(task) = coordinator_task { + match task.await { + Ok(Ok(())) => {} + Ok(Err(error)) => tracing::warn!(%error, "采集器内部接口关闭失败"), + Err(error) => tracing::warn!(%error, "采集器内部接口任务异常"), + } + } + if let Some(task) = local_heartbeat_task { + let _ = task.await; + } tracing::info!("dht-search 已安全停止"); run_result } diff --git a/src/search/src/cluster/collector.rs b/src/search/src/cluster/collector.rs new file mode 100644 index 0000000..f044ffb --- /dev/null +++ b/src/search/src/cluster/collector.rs @@ -0,0 +1,682 @@ +// 负责运行远程采集器并协调 DHT 租约持久待发送箱配置同步和验证任务 + +use std::{ + collections::{HashMap, HashSet}, + net::SocketAddr, + str::FromStr, + sync::{ + Arc, Mutex, RwLock, + atomic::{AtomicBool, AtomicU64, Ordering}, + }, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use dht_crawler::{DHTServer, DhtRuntimeStats}; +use tokio::{task::JoinHandle, time::MissedTickBehavior}; +use tokio_util::sync::CancellationToken; + +use crate::{ + config::{AppConfig, dht_options_from}, + domain::{InfoHash, VerificationResult}, + error::AppError, + shutdown, +}; + +use super::{ + cluster_token, + outbox::{Outbox, OutboxError}, + protocol::{ + AdmissionRequest, AdmissionResponse, AdmissionStatus, CollectorDesiredConfig, + CollectorMetrics, ControlResponse, HeartbeatRequest, MetadataEnvelope, + MetadataSubmitResponse, PROTOCOL_VERSION, RegisterRequest, VerificationClaimRequest, + VerificationClaimResponse, VerificationJob, VerificationResultRequest, + }, +}; + +#[derive(Clone)] +struct CoordinatorClient { + http: reqwest::Client, + base_url: String, + token: Arc, +} + +struct CollectorShared { + node_id: String, + client: CoordinatorClient, + outbox: Arc, + leases: Mutex>, + stats: RwLock, + server: tokio::sync::RwLock>, + applied_revision: AtomicU64, + connected: AtomicBool, + capacity_paused: AtomicBool, + last_error: RwLock>, +} + +pub(crate) async fn run(config: AppConfig) -> Result<(), AppError> { + std::fs::create_dir_all(&config.collector.state_dir)?; + let outbox = Arc::new( + Outbox::open( + config.collector.state_dir.join("outbox"), + config.collector.outbox_max_items, + config.collector.outbox_max_bytes, + ) + .map_err(|error| AppError::Cluster(error.to_string()))?, + ); + let node_id = outbox + .identity() + .map_err(|error| AppError::Cluster(error.to_string()))?; + let client = CoordinatorClient::new(&config.collector.coordinator_url, cluster_token()?)?; + let initial = CollectorDesiredConfig { + dht: config.dht.clone(), + verification_enabled: config.verification.enabled, + verification_max_active: config.verification.max_active, + }; + let register = RegisterRequest { + protocol_version: PROTOCOL_VERSION, + node_id: node_id.clone(), + name: config.collector.name.clone(), + initial_config: initial, + }; + let (control, initially_connected) = match client.register(®ister).await { + Ok(control) => { + outbox + .store_control(&control) + .map_err(|error| AppError::Cluster(error.to_string()))?; + (control, true) + } + Err(error) => ( + outbox + .load_control() + .map_err(|error| AppError::Cluster(error.to_string()))? + .ok_or_else(|| { + AppError::Cluster(format!("首次注册采集器失败且没有缓存配置: {error}")) + })?, + false, + ), + }; + let shared = Arc::new(CollectorShared { + node_id, + client, + outbox, + leases: Mutex::new(HashMap::new()), + stats: RwLock::new(DhtRuntimeStats::default()), + server: tokio::sync::RwLock::new(None), + applied_revision: AtomicU64::new(0), + connected: AtomicBool::new(initially_connected), + capacity_paused: AtomicBool::new(false), + last_error: RwLock::new(None), + }); + let cancel = CancellationToken::new(); + let (control_tx, control_rx) = tokio::sync::watch::channel(control); + let heartbeat = tokio::spawn(heartbeat_loop( + shared.clone(), + register, + control_tx, + config.collector.heartbeat_secs, + cancel.clone(), + )); + let sender = tokio::spawn(sender_loop(shared.clone(), cancel.clone())); + let verifier = tokio::spawn(verification_loop( + shared.clone(), + control_rx.clone(), + config.verification.max_peer_attempts, + cancel.clone(), + )); + let dht = tokio::spawn(dht_loop( + shared.clone(), + control_rx, + config.metadata_limits.max_metadata_bytes, + cancel.clone(), + )); + + tracing::info!(node_id = %shared.node_id, "远程 DHT 采集器启动"); + shutdown::signal().await; + cancel.cancel(); + if let Some(server) = shared.server.write().await.take() { + server.shutdown(); + } + for task in [heartbeat, sender, verifier, dht] { + if let Err(error) = task.await { + tracing::warn!(%error, "采集器后台任务异常结束"); + } + } + tracing::info!(node_id = %shared.node_id, "远程 DHT 采集器已停止"); + Ok(()) +} + +impl CoordinatorClient { + fn new(base_url: &str, token: String) -> Result { + let http = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(15)) + .build() + .map_err(|error| AppError::Cluster(error.to_string()))?; + Ok(Self { + http, + base_url: base_url.trim_end_matches('/').to_owned(), + token: Arc::from(token), + }) + } + + async fn register(&self, request: &RegisterRequest) -> Result { + self.post_json("/internal/register", request).await + } + + async fn heartbeat(&self, request: &HeartbeatRequest) -> Result { + self.post_json("/internal/heartbeat", request).await + } + + async fn admit(&self, node_id: &str, hashes: Vec) -> Result { + self.post_json( + "/internal/admissions", + &AdmissionRequest { + protocol_version: PROTOCOL_VERSION, + node_id: node_id.to_owned(), + info_hashes: hashes, + }, + ) + .await + } + + async fn submit(&self, envelope: &MetadataEnvelope) -> Result { + let bytes = rmp_serde::to_vec_named(envelope).map_err(|error| error.to_string())?; + let response = self + .http + .post(format!("{}/internal/metadata", self.base_url)) + .bearer_auth(self.token.as_ref()) + .header(reqwest::header::CONTENT_TYPE, "application/msgpack") + .body(bytes) + .send() + .await + .map_err(|error| error.to_string())?; + decode_response(response).await + } + + async fn claim_verification( + &self, + node_id: &str, + limit: usize, + ) -> Result { + self.post_json( + "/internal/verification/claim", + &VerificationClaimRequest { + protocol_version: PROTOCOL_VERSION, + node_id: node_id.to_owned(), + limit, + }, + ) + .await + } + + async fn finish_verification( + &self, + request: &VerificationResultRequest, + ) -> Result { + self.post_json("/internal/verification/result", request) + .await + } + + async fn post_json( + &self, + path: &str, + body: &T, + ) -> Result { + let response = self + .http + .post(format!("{}{path}", self.base_url)) + .bearer_auth(self.token.as_ref()) + .json(body) + .send() + .await + .map_err(|error| error.to_string())?; + decode_response(response).await + } +} + +async fn decode_response( + response: reqwest::Response, +) -> Result { + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + return Err(format!("协调器返回 {status}: {body}")); + } + response.json().await.map_err(|error| error.to_string()) +} + +async fn heartbeat_loop( + shared: Arc, + register: RegisterRequest, + control_tx: tokio::sync::watch::Sender, + interval_secs: u64, + cancel: CancellationToken, +) { + let mut ticker = tokio::time::interval(Duration::from_secs(interval_secs)); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = ticker.tick() => { + let request = HeartbeatRequest { + protocol_version: PROTOCOL_VERSION, + node_id: shared.node_id.clone(), + applied_revision: shared.applied_revision.load(Ordering::Relaxed), + metrics: metrics(&shared), + }; + let result = match shared.client.heartbeat(&request).await { + Ok(control) => Ok(control), + Err(_) => shared.client.register(®ister).await, + }; + match result { + Ok(control) => { + shared.connected.store(true, Ordering::Release); + clear_error(&shared); + if let Err(error) = shared.outbox.store_control(&control) { + set_error(&shared, error.to_string()); + } + if control_tx.send(control).is_err() { + break; + } + } + Err(error) => { + shared.connected.store(false, Ordering::Release); + set_error(&shared, error); + } + } + } + } + } +} + +async fn sender_loop(shared: Arc, cancel: CancellationToken) { + let mut ticker = tokio::time::interval(Duration::from_millis(100)); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = ticker.tick() => { + let outbox = shared.outbox.clone(); + let now = unix_timestamp(); + let item = match tokio::task::spawn_blocking(move || outbox.next_ready(now)).await { + Ok(Ok(item)) => item, + Ok(Err(error)) => { + set_error(&shared, error.to_string()); + continue; + } + Err(error) => { + set_error(&shared, error.to_string()); + continue; + } + }; + let Some(item) = item else { continue }; + match shared.client.submit(&item.envelope).await { + Ok(_) => { + let outbox = shared.outbox.clone(); + match tokio::task::spawn_blocking(move || outbox.acknowledge(&item)).await { + Ok(Ok(())) => {} + Ok(Err(error)) => set_error(&shared, error.to_string()), + Err(error) => set_error(&shared, error.to_string()), + } + } + Err(error) => { + set_error(&shared, error); + let outbox = shared.outbox.clone(); + let _ = tokio::task::spawn_blocking(move || outbox.retry(item, now)).await; + } + } + } + } + } +} + +async fn dht_loop( + shared: Arc, + mut control_rx: tokio::sync::watch::Receiver, + max_metadata_bytes: usize, + cancel: CancellationToken, +) { + let mut active_revision = u64::MAX; + let mut runner: Option> = None; + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = control_rx.changed() => {} + _ = ticker.tick() => {} + } + let control = control_rx.borrow().clone(); + let high = shared.outbox.high_watermark(); + if high { + shared.capacity_paused.store(true, Ordering::Release); + } else if shared.outbox.below_low_watermark() { + shared.capacity_paused.store(false, Ordering::Release); + } + let should_run = control.desired.dht.enabled + && shared.connected.load(Ordering::Acquire) + && !shared.capacity_paused.load(Ordering::Acquire); + let running = shared.server.read().await.is_some(); + if running && (!should_run || active_revision != control.revision) { + if let Some(server) = shared.server.write().await.take() { + server.shutdown(); + } + if let Some(task) = runner.take() { + let _ = task.await; + } + *shared + .stats + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = DhtRuntimeStats::default(); + } + if should_run && shared.server.read().await.is_none() { + match DHTServer::new(dht_options_from(&control.desired.dht, max_metadata_bytes)).await { + Ok(server) => { + configure_callbacks(&server, shared.clone()); + *shared + .stats + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + server.runtime_stats(); + *shared.server.write().await = Some(server.clone()); + let running_server = server.clone(); + runner = Some(tokio::spawn(async move { + if let Err(error) = running_server.start().await { + tracing::error!(%error, "采集器 DHT 运行时停止"); + } + })); + active_revision = control.revision; + shared + .applied_revision + .store(control.revision, Ordering::Release); + clear_error(&shared); + } + Err(error) => set_error(&shared, error.to_string()), + } + } else if !should_run { + shared + .applied_revision + .store(control.revision, Ordering::Release); + active_revision = control.revision; + } + } + if let Some(server) = shared.server.write().await.take() { + server.shutdown(); + } + if let Some(task) = runner { + let _ = task.await; + } +} + +fn configure_callbacks(server: &DHTServer, shared: Arc) { + let sampled = shared.clone(); + server.on_sampled_hashes(move |hashes| { + let shared = sampled.clone(); + async move { + let text: Vec<_> = hashes.iter().map(hex::encode).collect(); + let response = match shared.client.admit(&shared.node_id, text).await { + Ok(response) => response, + Err(error) => { + set_error(&shared, error); + return Vec::new(); + } + }; + let mut granted = HashSet::new(); + let mut leases = shared + .leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for decision in response.decisions { + if decision.status == AdmissionStatus::Granted { + if let Some(token) = decision.lease_token { + leases.insert(decision.info_hash.clone(), token); + } + granted.insert(decision.info_hash); + } + } + hashes + .into_iter() + .filter(|hash| granted.contains(&hex::encode(hash))) + .collect() + } + }); + + let gate = shared.clone(); + server.on_metadata_fetch(move |hash| { + let shared = gate.clone(); + async move { + if shared + .leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains_key(&hash) + { + return true; + } + match shared + .client + .admit(&shared.node_id, vec![hash.clone()]) + .await + { + Ok(response) => response.decisions.into_iter().any(|decision| { + if decision.status == AdmissionStatus::Granted { + if let Some(token) = decision.lease_token { + shared + .leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(hash.clone(), token); + } + true + } else { + false + } + }), + Err(error) => { + set_error(&shared, error); + false + } + } + } + }); + + let sink = shared; + server.on_torrent_with_async_ack(move |torrent| { + let shared = sink.clone(); + async move { + let lease_token = shared + .leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&torrent.info_hash); + let envelope = MetadataEnvelope { + protocol_version: PROTOCOL_VERSION, + node_id: shared.node_id.clone(), + lease_token, + torrent, + }; + let outbox = shared.outbox.clone(); + match tokio::task::spawn_blocking(move || outbox.append(envelope)).await { + Ok(Ok(_)) => true, + Ok(Err(error)) => { + if matches!(error, OutboxError::Full) { + shared.capacity_paused.store(true, Ordering::Release); + } + set_error(&shared, error.to_string()); + false + } + Err(error) => { + set_error(&shared, error.to_string()); + false + } + } + } + }); +} + +async fn verification_loop( + shared: Arc, + control_rx: tokio::sync::watch::Receiver, + max_peer_attempts: usize, + cancel: CancellationToken, +) { + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = ticker.tick() => {} + } + let control = control_rx.borrow().clone(); + if !control.desired.verification_enabled || !shared.connected.load(Ordering::Acquire) { + continue; + } + let Some(server) = shared.server.read().await.clone() else { + continue; + }; + let response = match shared + .client + .claim_verification(&shared.node_id, control.desired.verification_max_active) + .await + { + Ok(response) => response, + Err(error) => { + set_error(&shared, error); + continue; + } + }; + let mut tasks = tokio::task::JoinSet::new(); + for job in response.jobs { + let server = server.clone(); + tasks.spawn(async move { verify_one(server, job, max_peer_attempts).await }); + } + while let Some(result) = tasks.join_next().await { + let Ok(Ok((job, result))) = result else { + continue; + }; + let request = VerificationResultRequest { + protocol_version: PROTOCOL_VERSION, + node_id: shared.node_id.clone(), + lease_token: job.lease_token, + info_hash: job.info_hash, + result, + }; + if let Err(error) = shared.client.finish_verification(&request).await { + set_error(&shared, error); + } + } + } +} + +async fn verify_one( + server: DHTServer, + job: VerificationJob, + max_peer_attempts: usize, +) -> Result<(VerificationJob, VerificationResult), String> { + let hash = InfoHash::from_str(&job.info_hash).map_err(|error| error.to_string())?; + let dht_peers = server + .lookup_peers(*hash.as_bytes()) + .await + .map(|result| result.peers) + .unwrap_or_default(); + let mut unique = HashSet::new(); + let peers: Vec = dht_peers + .into_iter() + .chain(job.source_peers.iter().filter_map(|peer| peer.parse().ok())) + .filter(|peer| unique.insert(*peer)) + .collect(); + let mut handshakes = tokio::task::JoinSet::new(); + for peer in peers.iter().copied().take(max_peer_attempts) { + let server = server.clone(); + let bytes = *hash.as_bytes(); + handshakes.spawn(async move { server.verify_peer_handshake(bytes, peer).await }); + } + let mut reachable = 0_u32; + while let Some(result) = handshakes.join_next().await { + if result.unwrap_or(false) { + reachable = reachable.saturating_add(1); + } + } + Ok(( + job, + VerificationResult { + verified_at: unix_timestamp(), + discovered_peers: peers.len().min(u32::MAX as usize) as u32, + reachable_peers: reachable, + }, + )) +} + +fn metrics(shared: &CollectorShared) -> CollectorMetrics { + let runtime_stats = shared + .stats + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let stats = runtime_stats.snapshot(); + let observability = runtime_stats.observability_snapshot(); + let process = crate::diagnostics::process::snapshot(); + let outbox = shared.outbox.snapshot(); + CollectorMetrics { + state: if shared.capacity_paused.load(Ordering::Acquire) { + "outbox_full" + } else if !shared.connected.load(Ordering::Acquire) { + "disconnected" + } else if stats.node_pool_size > 0 { + "running" + } else { + "idle" + } + .to_owned(), + nodes: stats.node_pool_size as u64, + metadata_in_flight: stats.metadata_in_flight as u64, + metadata_succeeded: stats.metadata_peer_succeeded, + metadata_failed: stats.metadata_peer_failed, + udp_tx_packets: observability.udp_tx_packets, + metadata_peer_attempts: stats.metadata_peer_attempts, + metadata_peer_canceled: stats.metadata_peer_canceled, + metadata_peer_failure_cache_hits: stats.metadata_peer_failure_cache_hits, + metadata_peer_failure_cache_entries: stats.metadata_peer_failure_cache_entries as u64, + metadata_failure_timeout: observability.metadata_failure_timeout, + metadata_failure_connect: observability.metadata_failure_connect, + metadata_failure_no_extension: observability.metadata_failure_no_extension, + metadata_failure_send: observability.metadata_failure_send, + metadata_failure_size_limit: observability.metadata_failure_size_limit, + metadata_failure_sha1: observability.metadata_failure_sha1, + metadata_failure_parse: observability.metadata_failure_parse, + metadata_failure_other: observability.metadata_failure_other, + sample_queue_depth: stats.sample_candidate_queue_depth as u64, + sample_queue_capacity: stats.sample_candidate_queue_capacity as u64, + resident_memory_bytes: process.resident_memory_bytes, + private_memory_bytes: process.private_memory_bytes, + cpu_time_millis: process.cpu_time_millis, + thread_count: process.thread_count, + handle_count: process.handle_count, + outbox_items: outbox.items, + outbox_bytes: outbox.bytes, + last_error: shared + .last_error + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(), + } +} + +fn set_error(shared: &CollectorShared, error: String) { + *shared + .last_error + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(error); +} + +fn clear_error(shared: &CollectorShared) { + *shared + .last_error + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = None; +} + +fn unix_timestamp() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} diff --git a/src/search/src/cluster/coordinator.rs b/src/search/src/cluster/coordinator.rs new file mode 100644 index 0000000..a11395d --- /dev/null +++ b/src/search/src/cluster/coordinator.rs @@ -0,0 +1,959 @@ +// 负责提供带认证的采集器准入入库心跳配置和验证任务内部接口 + +use std::{ + collections::HashMap, + net::SocketAddr, + str::FromStr, + sync::{Arc, Mutex, RwLock}, + time::{SystemTime, UNIX_EPOCH}, +}; + +use axum::{ + Json, Router, + body::Bytes, + extract::{DefaultBodyLimit, Request, State}, + http::{HeaderMap, StatusCode}, + middleware::{self, Next}, + response::{IntoResponse, Response}, + routing::post, +}; +use serde::Serialize; +use tokio_util::sync::CancellationToken; + +use crate::{ + config::CoordinatorConfig, + crawler::mapper::metadata_candidate, + domain::{InfoHash, MetadataLimits, RejectedMetadata, TorrentRecord}, + storage::{RocksTorrentRepository, TorrentRepository}, +}; + +use super::{ + protocol::{ + AdmissionDecision, AdmissionRequest, AdmissionResponse, AdmissionStatus, + CollectorConfigUpdate, CollectorMetrics, CollectorView, ControlResponse, HeartbeatRequest, + MetadataEnvelope, MetadataSubmitResponse, MetadataSubmitStatus, PROTOCOL_VERSION, + RegisterRequest, VerificationClaimRequest, VerificationClaimResponse, VerificationJob, + VerificationResultRequest, + }, + registry::{CollectorRecord, CollectorRegistry}, +}; + +#[derive(Clone)] +pub(crate) struct CoordinatorRuntime { + inner: Arc, +} + +struct CoordinatorState { + token_hash: [u8; 32], + repository: Arc, + registry: Arc, + metadata_limits: MetadataLimits, + acquisition_lease_secs: u64, + verification_lease_secs: u64, + offline_after_secs: u64, + max_leases: usize, + leases: Mutex>, + verification_leases: Mutex>, + live: RwLock>, +} + +#[derive(Clone)] +struct LiveCollector { + last_seen: u64, + applied_revision: u64, + metrics: CollectorMetrics, +} + +#[derive(Clone)] +struct AcquisitionLease { + token: String, + expires_at: u64, +} + +#[derive(Debug, Serialize)] +struct InternalError { + error: String, +} + +type HandlerResult = Result, (StatusCode, Json)>; + +impl CoordinatorRuntime { + pub(crate) fn open( + data_dir: &std::path::Path, + repository: Arc, + metadata_limits: MetadataLimits, + config: &CoordinatorConfig, + verification_lease_secs: u64, + token: &str, + ) -> Result { + let registry = CollectorRegistry::open(data_dir.join("cluster")) + .map_err(|error| crate::error::AppError::Cluster(error.to_string()))?; + Ok(Self { + inner: Arc::new(CoordinatorState { + token_hash: *blake3::hash(token.as_bytes()).as_bytes(), + repository, + registry: Arc::new(registry), + metadata_limits, + acquisition_lease_secs: config.lease_secs, + verification_lease_secs, + offline_after_secs: config.offline_after_secs, + max_leases: config.max_leases, + leases: Mutex::new(HashMap::new()), + verification_leases: Mutex::new(HashMap::new()), + live: RwLock::new(HashMap::new()), + }), + }) + } + + pub(crate) async fn serve( + &self, + listen: SocketAddr, + cancel: CancellationToken, + ) -> std::io::Result<()> { + let router = internal_router(self.clone()); + let listener = tokio::net::TcpListener::bind(listen).await?; + tracing::info!(%listen, "采集器内部接口启动"); + axum::serve(listener, router) + .with_graceful_shutdown(cancel.cancelled_owned()) + .await + } + + pub(crate) fn collectors(&self) -> Result, String> { + let records = self + .inner + .registry + .list() + .map_err(|error| error.to_string())?; + let live = self + .inner + .live + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let now = unix_timestamp(); + Ok(records + .into_iter() + .map(|record| { + let node_id = record.node_id.clone(); + view( + record, + live.get(&node_id), + now, + self.inner.offline_after_secs, + ) + }) + .collect()) + } + + pub(crate) fn aggregate_metrics(&self) -> Result { + let collectors = self.collectors()?; + let mut aggregate = CollectorMetrics { + state: if collectors.iter().any(|collector| collector.online) { + "running".to_owned() + } else { + "idle".to_owned() + }, + ..CollectorMetrics::default() + }; + for collector in collectors { + let metrics = collector.metrics; + if collector.online { + aggregate.nodes = aggregate.nodes.saturating_add(metrics.nodes); + aggregate.metadata_in_flight = aggregate + .metadata_in_flight + .saturating_add(metrics.metadata_in_flight); + aggregate.metadata_peer_failure_cache_entries = aggregate + .metadata_peer_failure_cache_entries + .saturating_add(metrics.metadata_peer_failure_cache_entries); + aggregate.sample_queue_depth = aggregate + .sample_queue_depth + .saturating_add(metrics.sample_queue_depth); + aggregate.sample_queue_capacity = aggregate + .sample_queue_capacity + .saturating_add(metrics.sample_queue_capacity); + aggregate.resident_memory_bytes = sum_optional( + aggregate.resident_memory_bytes, + metrics.resident_memory_bytes, + ); + aggregate.private_memory_bytes = + sum_optional(aggregate.private_memory_bytes, metrics.private_memory_bytes); + aggregate.cpu_time_millis = + sum_optional(aggregate.cpu_time_millis, metrics.cpu_time_millis); + aggregate.thread_count = sum_optional(aggregate.thread_count, metrics.thread_count); + aggregate.handle_count = sum_optional(aggregate.handle_count, metrics.handle_count); + } + aggregate.metadata_succeeded = aggregate + .metadata_succeeded + .saturating_add(metrics.metadata_succeeded); + aggregate.metadata_failed = aggregate + .metadata_failed + .saturating_add(metrics.metadata_failed); + aggregate.udp_tx_packets = aggregate + .udp_tx_packets + .saturating_add(metrics.udp_tx_packets); + aggregate.metadata_peer_attempts = aggregate + .metadata_peer_attempts + .saturating_add(metrics.metadata_peer_attempts); + aggregate.metadata_peer_canceled = aggregate + .metadata_peer_canceled + .saturating_add(metrics.metadata_peer_canceled); + aggregate.metadata_peer_failure_cache_hits = aggregate + .metadata_peer_failure_cache_hits + .saturating_add(metrics.metadata_peer_failure_cache_hits); + aggregate.metadata_failure_timeout = aggregate + .metadata_failure_timeout + .saturating_add(metrics.metadata_failure_timeout); + aggregate.metadata_failure_connect = aggregate + .metadata_failure_connect + .saturating_add(metrics.metadata_failure_connect); + aggregate.metadata_failure_no_extension = aggregate + .metadata_failure_no_extension + .saturating_add(metrics.metadata_failure_no_extension); + aggregate.metadata_failure_send = aggregate + .metadata_failure_send + .saturating_add(metrics.metadata_failure_send); + aggregate.metadata_failure_size_limit = aggregate + .metadata_failure_size_limit + .saturating_add(metrics.metadata_failure_size_limit); + aggregate.metadata_failure_sha1 = aggregate + .metadata_failure_sha1 + .saturating_add(metrics.metadata_failure_sha1); + aggregate.metadata_failure_parse = aggregate + .metadata_failure_parse + .saturating_add(metrics.metadata_failure_parse); + aggregate.metadata_failure_other = aggregate + .metadata_failure_other + .saturating_add(metrics.metadata_failure_other); + aggregate.outbox_items = aggregate.outbox_items.saturating_add(metrics.outbox_items); + aggregate.outbox_bytes = aggregate.outbox_bytes.saturating_add(metrics.outbox_bytes); + } + Ok(aggregate) + } + + pub(crate) fn update_collector( + &self, + node_id: &str, + update: CollectorConfigUpdate, + ) -> Result, String> { + if update.name.trim().is_empty() { + return Err("采集器名称不能为空".to_owned()); + } + let expected_revision = update.revision; + let record = self + .inner + .registry + .update(node_id, expected_revision, update.name, update.desired) + .map_err(|error| error.to_string())?; + if record + .as_ref() + .is_some_and(|record| record.revision != expected_revision.saturating_add(1)) + { + return Err("采集器配置已经被其他请求修改 请重新加载".to_owned()); + } + let live = self + .inner + .live + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Ok(record.map(|record| { + view( + record, + live.get(node_id), + unix_timestamp(), + self.inner.offline_after_secs, + ) + })) + } + + pub(crate) fn register_local( + &self, + node_id: String, + name: String, + desired: super::protocol::CollectorDesiredConfig, + ) -> Result { + let record = self + .inner + .registry + .register(&RegisterRequest { + protocol_version: PROTOCOL_VERSION, + node_id: node_id.clone(), + name: name.clone(), + initial_config: desired.clone(), + }) + .map_err(|error| error.to_string())?; + if record.name == name && record.desired == desired { + return Ok(record); + } + self.inner + .registry + .update(&node_id, record.revision, name, desired) + .map_err(|error| error.to_string())? + .ok_or_else(|| "本机采集器注册记录意外消失".to_owned()) + } + + pub(crate) fn heartbeat_local( + &self, + node_id: String, + applied_revision: u64, + metrics: CollectorMetrics, + ) { + self.inner + .live + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert( + node_id, + LiveCollector { + last_seen: unix_timestamp(), + applied_revision, + metrics, + }, + ); + } +} + +fn sum_optional(left: Option, right: Option) -> Option { + match (left, right) { + (Some(left), Some(right)) => Some(left.saturating_add(right)), + (Some(value), None) | (None, Some(value)) => Some(value), + (None, None) => None, + } +} + +fn internal_router(runtime: CoordinatorRuntime) -> Router { + Router::new() + .route("/internal/register", post(register)) + .route("/internal/heartbeat", post(heartbeat)) + .route("/internal/admissions", post(admit)) + .route("/internal/metadata", post(submit_metadata)) + .route("/internal/verification/claim", post(claim_verification)) + .route("/internal/verification/result", post(finish_verification)) + .layer(DefaultBodyLimit::max(16 * 1024 * 1024)) + .layer(middleware::from_fn_with_state( + runtime.clone(), + authenticate_request, + )) + .with_state(runtime) +} + +async fn authenticate_request( + State(runtime): State, + request: Request, + next: Next, +) -> Response { + if let Err(error) = authorize(&runtime, request.headers()) { + return error.into_response(); + } + next.run(request).await +} + +async fn register( + State(runtime): State, + headers: HeaderMap, + Json(request): Json, +) -> HandlerResult { + authorize(&runtime, &headers)?; + validate_version(request.protocol_version)?; + validate_node_id(&request.node_id)?; + if request.name.trim().is_empty() { + return Err(bad_request("采集器名称不能为空")); + } + let registry = runtime.inner.registry.clone(); + let record = tokio::task::spawn_blocking(move || registry.register(&request)) + .await + .map_err(|error| internal(error.to_string()))? + .map_err(|error| internal(error.to_string()))?; + Ok(Json(control_response(record))) +} + +async fn heartbeat( + State(runtime): State, + headers: HeaderMap, + Json(request): Json, +) -> HandlerResult { + authorize(&runtime, &headers)?; + validate_version(request.protocol_version)?; + let record = runtime + .inner + .registry + .get(&request.node_id) + .map_err(|error| internal(error.to_string()))? + .ok_or_else(|| bad_request("采集器尚未注册"))?; + runtime + .inner + .live + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert( + request.node_id, + LiveCollector { + last_seen: unix_timestamp(), + applied_revision: request.applied_revision, + metrics: request.metrics, + }, + ); + Ok(Json(control_response(record))) +} + +async fn admit( + State(runtime): State, + headers: HeaderMap, + Json(request): Json, +) -> HandlerResult { + authorize(&runtime, &headers)?; + validate_version(request.protocol_version)?; + ensure_registered(&runtime, &request.node_id)?; + if request.info_hashes.is_empty() || request.info_hashes.len() > 512 { + return Err(bad_request("准入批量必须包含一到五百一十二个 infohash")); + } + let mut hashes = Vec::with_capacity(request.info_hashes.len()); + for hash in request.info_hashes { + hashes.push(InfoHash::from_str(&hash).map_err(|error| bad_request(error.to_string()))?); + } + hashes.sort_unstable_by_key(|hash| *hash.as_bytes()); + hashes.dedup(); + let state = runtime.inner.clone(); + let decisions = tokio::task::spawn_blocking(move || admit_blocking(&state, hashes)) + .await + .map_err(|error| internal(error.to_string()))? + .map_err(|error| internal(error.to_string()))?; + Ok(Json(AdmissionResponse { decisions })) +} + +fn admit_blocking( + state: &CoordinatorState, + hashes: Vec, +) -> Result, crate::storage::StorageError> { + let now = unix_timestamp(); + let mut leases = state + .leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + leases.retain(|_, lease| lease.expires_at > now); + let mut decisions = Vec::with_capacity(hashes.len()); + for hash in hashes { + let text = hash.to_string(); + if state.repository.observe_existing(hash, now)? { + decisions.push(decision(text, AdmissionStatus::Known, None)); + continue; + } + if let Some(lease) = leases.get(&hash) { + decisions.push(AdmissionDecision { + info_hash: text, + status: AdmissionStatus::Leased, + lease_token: None, + expires_at: Some(lease.expires_at), + }); + continue; + } + if leases.len() >= state.max_leases { + decisions.push(decision(text, AdmissionStatus::Busy, None)); + continue; + } + let token = random_token().unwrap_or_else(|_| format!("{}-{now}", hash)); + let expires_at = now.saturating_add(state.acquisition_lease_secs); + leases.insert( + hash, + AcquisitionLease { + token: token.clone(), + expires_at, + }, + ); + decisions.push(AdmissionDecision { + info_hash: text, + status: AdmissionStatus::Granted, + lease_token: Some(token), + expires_at: Some(expires_at), + }); + } + Ok(decisions) +} + +async fn submit_metadata( + State(runtime): State, + headers: HeaderMap, + body: Bytes, +) -> HandlerResult { + authorize(&runtime, &headers)?; + let envelope: MetadataEnvelope = + rmp_serde::from_slice(&body).map_err(|error| bad_request(error.to_string()))?; + validate_version(envelope.protocol_version)?; + ensure_registered(&runtime, &envelope.node_id)?; + let state = runtime.inner.clone(); + let response = tokio::task::spawn_blocking(move || ingest_metadata(&state, envelope)) + .await + .map_err(|error| internal(error.to_string()))? + .map_err(|error| internal(error.to_string()))?; + Ok(Json(response)) +} + +fn ingest_metadata( + state: &CoordinatorState, + mut envelope: MetadataEnvelope, +) -> Result { + let hash = match InfoHash::from_str(&envelope.torrent.info_hash) { + Ok(hash) => hash, + Err(error) => { + return Ok(MetadataSubmitResponse { + status: MetadataSubmitStatus::Rejected, + reason: Some(error.to_string()), + }); + } + }; + if state.repository.contains(hash)? { + remove_lease(state, hash, envelope.lease_token.as_deref()); + return Ok(MetadataSubmitResponse { + status: MetadataSubmitStatus::AlreadyKnown, + reason: None, + }); + } + envelope.torrent.timestamp = unix_timestamp(); + match TorrentRecord::try_from_with_limits( + metadata_candidate(envelope.torrent), + state.metadata_limits, + ) { + Ok(record) => { + let inserted = state.repository.insert_if_absent(record)?; + remove_lease(state, hash, envelope.lease_token.as_deref()); + Ok(MetadataSubmitResponse { + status: if inserted { + MetadataSubmitStatus::Inserted + } else { + MetadataSubmitStatus::AlreadyKnown + }, + reason: None, + }) + } + Err(error) => { + if state.repository.rejection(hash)?.is_none() { + state.repository.record_rejection(RejectedMetadata::new( + hash, + error.rejection_reason(), + state.metadata_limits.rule_id(), + unix_timestamp(), + ))?; + } + remove_lease(state, hash, envelope.lease_token.as_deref()); + Ok(MetadataSubmitResponse { + status: MetadataSubmitStatus::Rejected, + reason: Some(error.to_string()), + }) + } + } +} + +async fn claim_verification( + State(runtime): State, + headers: HeaderMap, + Json(request): Json, +) -> HandlerResult { + authorize(&runtime, &headers)?; + validate_version(request.protocol_version)?; + ensure_registered(&runtime, &request.node_id)?; + let state = runtime.inner.clone(); + let limit = request.limit.clamp(1, 32); + let jobs = tokio::task::spawn_blocking(move || claim_verification_blocking(&state, limit)) + .await + .map_err(|error| internal(error.to_string()))? + .map_err(|error| internal(error.to_string()))?; + Ok(Json(VerificationClaimResponse { jobs })) +} + +fn claim_verification_blocking( + state: &CoordinatorState, + limit: usize, +) -> Result, crate::storage::StorageError> { + let mut jobs = Vec::with_capacity(limit); + for _ in 0..limit { + let Some(request) = state + .repository + .claim_verification(unix_timestamp(), state.verification_lease_secs)? + else { + break; + }; + let token = random_token().unwrap_or_else(|_| request.info_hash.to_string()); + let source_peers = state + .repository + .get(request.info_hash)? + .map(|record| record.source_peers) + .unwrap_or_default(); + state + .verification_leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(token.clone(), request.info_hash); + jobs.push(VerificationJob { + info_hash: request.info_hash.to_string(), + lease_token: token, + source_peers, + }); + } + Ok(jobs) +} + +async fn finish_verification( + State(runtime): State, + headers: HeaderMap, + Json(request): Json, +) -> HandlerResult { + authorize(&runtime, &headers)?; + validate_version(request.protocol_version)?; + ensure_registered(&runtime, &request.node_id)?; + let hash = + InfoHash::from_str(&request.info_hash).map_err(|error| bad_request(error.to_string()))?; + let leased = runtime + .inner + .verification_leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&request.lease_token); + if leased != Some(hash) { + return Err(( + StatusCode::CONFLICT, + Json(InternalError { + error: "验证任务租约已失效".to_owned(), + }), + )); + } + let repository = runtime.inner.repository.clone(); + tokio::task::spawn_blocking(move || repository.finish_verification(hash, request.result)) + .await + .map_err(|error| internal(error.to_string()))? + .map_err(|error| internal(error.to_string()))?; + Ok(Json(serde_json::json!({"status": "accepted"}))) +} + +fn authorize( + runtime: &CoordinatorRuntime, + headers: &HeaderMap, +) -> Result<(), (StatusCode, Json)> { + let Some(value) = headers.get(axum::http::header::AUTHORIZATION) else { + return Err(unauthorized()); + }; + let Ok(value) = value.to_str() else { + return Err(unauthorized()); + }; + let Some(token) = value.strip_prefix("Bearer ") else { + return Err(unauthorized()); + }; + if blake3::hash(token.as_bytes()).as_bytes() != &runtime.inner.token_hash { + return Err(unauthorized()); + } + Ok(()) +} + +fn ensure_registered( + runtime: &CoordinatorRuntime, + node_id: &str, +) -> Result<(), (StatusCode, Json)> { + validate_node_id(node_id)?; + if runtime + .inner + .registry + .get(node_id) + .map_err(|error| internal(error.to_string()))? + .is_none() + { + return Err(bad_request("采集器尚未注册")); + } + Ok(()) +} + +fn validate_version(version: u16) -> Result<(), (StatusCode, Json)> { + if version != PROTOCOL_VERSION { + return Err(( + StatusCode::UPGRADE_REQUIRED, + Json(InternalError { + error: format!("协议版本不兼容 当前版本为 {PROTOCOL_VERSION}"), + }), + )); + } + Ok(()) +} + +fn validate_node_id(node_id: &str) -> Result<(), (StatusCode, Json)> { + if node_id.len() != 32 || !node_id.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(bad_request("采集器节点 ID 无效")); + } + Ok(()) +} + +fn control_response(record: CollectorRecord) -> ControlResponse { + ControlResponse { + protocol_version: PROTOCOL_VERSION, + revision: record.revision, + desired: record.desired, + } +} + +fn view( + record: CollectorRecord, + live: Option<&LiveCollector>, + now: u64, + offline_after: u64, +) -> CollectorView { + CollectorView { + node_id: record.node_id, + name: record.name, + revision: record.revision, + applied_revision: live.map_or(0, |live| live.applied_revision), + desired: record.desired, + online: live.is_some_and(|live| now.saturating_sub(live.last_seen) <= offline_after), + last_seen: live.map(|live| live.last_seen), + metrics: live.map_or_else(CollectorMetrics::default, |live| live.metrics.clone()), + } +} + +fn decision( + info_hash: String, + status: AdmissionStatus, + lease: Option, +) -> AdmissionDecision { + AdmissionDecision { + info_hash, + status, + lease_token: lease.as_ref().map(|lease| lease.token.clone()), + expires_at: lease.map(|lease| lease.expires_at), + } +} + +fn remove_lease(state: &CoordinatorState, hash: InfoHash, token: Option<&str>) { + let mut leases = state + .leases + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if leases + .get(&hash) + .is_some_and(|lease| token.is_none() || token == Some(lease.token.as_str())) + { + leases.remove(&hash); + } +} + +fn random_token() -> Result { + let mut bytes = [0_u8; 16]; + getrandom::fill(&mut bytes)?; + Ok(hex::encode(bytes)) +} + +fn unix_timestamp() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +fn unauthorized() -> (StatusCode, Json) { + ( + StatusCode::UNAUTHORIZED, + Json(InternalError { + error: "内部接口令牌无效".to_owned(), + }), + ) +} + +fn bad_request(message: impl Into) -> (StatusCode, Json) { + ( + StatusCode::BAD_REQUEST, + Json(InternalError { + error: message.into(), + }), + ) +} + +fn internal(message: impl Into) -> (StatusCode, Json) { + tracing::error!(error = %message.into(), "采集器内部接口失败"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(InternalError { + error: "内部接口处理失败".to_owned(), + }), + ) +} + +#[cfg(test)] +mod tests { + use axum::{ + body::{Body, to_bytes}, + http::Request, + }; + use tempfile::TempDir; + use tower::ServiceExt; + + use super::*; + use crate::{ + cluster::protocol::CollectorDesiredConfig, + config::{AppConfigDto, CoordinatorConfig}, + domain::MetadataLimits, + storage::{RocksTorrentRepository, TorrentRepository}, + }; + + const TOKEN: &str = "0123456789abcdef0123456789abcdef"; + const NODE_ID: &str = "00112233445566778899aabbccddeeff"; + + fn runtime(directory: &TempDir) -> (CoordinatorRuntime, Arc) { + let repository = + Arc::new(RocksTorrentRepository::open(directory.path().join("rocks")).unwrap()); + let runtime = CoordinatorRuntime::open( + directory.path(), + repository.clone(), + MetadataLimits::default(), + &CoordinatorConfig::default(), + 60, + TOKEN, + ) + .unwrap(); + (runtime, repository) + } + + async fn json_request( + app: Router, + path: &str, + body: &T, + ) -> axum::response::Response { + app.oneshot( + Request::post(path) + .header("authorization", format!("Bearer {TOKEN}")) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(body).unwrap())) + .unwrap(), + ) + .await + .unwrap() + } + + #[tokio::test] + async fn one_lease_and_idempotent_metadata_are_enforced() { + let directory = TempDir::new().unwrap(); + let (runtime, repository) = runtime(&directory); + let app = internal_router(runtime); + let desired = CollectorDesiredConfig { + dht: AppConfigDto::default().dht, + verification_enabled: true, + verification_max_active: 2, + }; + let response = json_request( + app.clone(), + "/internal/register", + &RegisterRequest { + protocol_version: PROTOCOL_VERSION, + node_id: NODE_ID.to_owned(), + name: "test".to_owned(), + initial_config: desired, + }, + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + + let hash = "01".repeat(20); + let request = AdmissionRequest { + protocol_version: PROTOCOL_VERSION, + node_id: NODE_ID.to_owned(), + info_hashes: vec![hash.clone()], + }; + let response = json_request(app.clone(), "/internal/admissions", &request).await; + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let first: AdmissionResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(first.decisions[0].status, AdmissionStatus::Granted); + let lease_token = first.decisions[0].lease_token.clone(); + + let response = json_request(app.clone(), "/internal/admissions", &request).await; + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let second: AdmissionResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(second.decisions[0].status, AdmissionStatus::Leased); + + let envelope = MetadataEnvelope { + protocol_version: PROTOCOL_VERSION, + node_id: NODE_ID.to_owned(), + lease_token, + torrent: dht_crawler::TorrentInfo { + info_hash: hash.clone(), + magnet_link: format!("magnet:?xt=urn:btih:{hash}"), + name: "example".to_owned(), + total_size: 1, + files: vec![dht_crawler::FileInfo { + path: "example.iso".to_owned(), + size: 1, + }], + piece_length: 16_384, + peers: Vec::new(), + timestamp: 1, + }, + }; + let bytes = rmp_serde::to_vec_named(&envelope).unwrap(); + for expected in [ + MetadataSubmitStatus::Inserted, + MetadataSubmitStatus::AlreadyKnown, + ] { + let response = app + .clone() + .oneshot( + Request::post("/internal/metadata") + .header("authorization", format!("Bearer {TOKEN}")) + .header("content-type", "application/msgpack") + .body(Body::from(bytes.clone())) + .unwrap(), + ) + .await + .unwrap(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let submitted: MetadataSubmitResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(submitted.status, expected); + } + let info_hash = InfoHash::from_str(&hash).unwrap(); + let record = repository.get(info_hash).unwrap().unwrap(); + assert_eq!(record.seen_count, 1); + } + + #[tokio::test] + async fn internal_routes_require_the_shared_token() { + let directory = TempDir::new().unwrap(); + let (runtime, _) = runtime(&directory); + let response = internal_router(runtime) + .oneshot( + Request::post("/internal/register") + .header("content-type", "application/json") + .body(Body::from("{}")) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + + #[test] + fn aggregate_metrics_sum_counters_and_online_gauges() { + let directory = TempDir::new().unwrap(); + let (runtime, _) = runtime(&directory); + let config = AppConfigDto::default(); + let desired = CollectorDesiredConfig { + dht: config.dht, + verification_enabled: config.verification.enabled, + verification_max_active: config.verification.max_active, + }; + for (node_id, nodes, succeeded, timeout) in [ + (NODE_ID, 11, 3, 7), + ("ffeeddccbbaa99887766554433221100", 13, 5, 9), + ] { + let record = runtime + .register_local(node_id.to_owned(), node_id.to_owned(), desired.clone()) + .unwrap(); + runtime.heartbeat_local( + node_id.to_owned(), + record.revision, + CollectorMetrics { + nodes, + metadata_succeeded: succeeded, + metadata_failure_timeout: timeout, + resident_memory_bytes: Some(100), + ..CollectorMetrics::default() + }, + ); + } + + let aggregate = runtime.aggregate_metrics().unwrap(); + assert_eq!(aggregate.nodes, 24); + assert_eq!(aggregate.metadata_succeeded, 8); + assert_eq!(aggregate.metadata_failure_timeout, 16); + assert_eq!(aggregate.resident_memory_bytes, Some(200)); + } +} diff --git a/src/search/src/cluster/mod.rs b/src/search/src/cluster/mod.rs new file mode 100644 index 0000000..64fe7a6 --- /dev/null +++ b/src/search/src/cluster/mod.rs @@ -0,0 +1,23 @@ +// 负责组合分布式采集协议协调器采集器和节点注册能力 + +pub(crate) mod collector; +pub(crate) mod coordinator; +pub(crate) mod outbox; +pub(crate) mod protocol; +pub(crate) mod registry; + +pub(crate) const CLUSTER_TOKEN_ENV: &str = "DHT_SEARCH_CLUSTER_TOKEN"; + +pub(crate) fn cluster_token() -> Result { + let token = std::env::var(CLUSTER_TOKEN_ENV).map_err(|_| { + crate::error::AppError::Config(format!( + "分布式模式需要通过环境变量 {CLUSTER_TOKEN_ENV} 提供内部接口令牌" + )) + })?; + if token.trim().len() < 32 { + return Err(crate::error::AppError::Config(format!( + "环境变量 {CLUSTER_TOKEN_ENV} 至少需要三十二个字符" + ))); + } + Ok(token) +} diff --git a/src/search/src/cluster/outbox.rs b/src/search/src/cluster/outbox.rs new file mode 100644 index 0000000..c47f198 --- /dev/null +++ b/src/search/src/cluster/outbox.rs @@ -0,0 +1,283 @@ +// 负责持久保存采集器尚未送达的 Metadata 并实施字节和条目双重上限 + +use std::{path::Path, sync::Mutex}; + +use rocksdb::{DB, Direction, IteratorMode, WriteBatch}; +use serde::{Deserialize, Serialize}; + +use super::protocol::MetadataEnvelope; + +const IDENTITY_KEY: &[u8] = b"\0identity"; +const NEXT_SEQUENCE_KEY: &[u8] = b"\0next-sequence"; +const CONTROL_KEY: &[u8] = b"\0control"; +const ITEM_PREFIX: u8 = b'o'; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum OutboxError { + #[error("待发送箱数据库失败: {0}")] + Rocks(#[from] rocksdb::Error), + #[error("待发送箱编码失败: {0}")] + Encode(#[from] rmp_serde::encode::Error), + #[error("待发送箱解码失败: {0}")] + Decode(#[from] rmp_serde::decode::Error), + #[error("待发送箱已达到容量上限")] + Full, + #[error("无法生成采集器身份: {0}")] + Random(#[from] getrandom::Error), +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct OutboxItem { + pub(crate) sequence: u64, + pub(crate) envelope: MetadataEnvelope, + pub(crate) encoded_bytes: u64, + pub(crate) attempts: u32, + pub(crate) next_attempt_at: u64, +} + +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct OutboxSnapshot { + pub(crate) items: u64, + pub(crate) bytes: u64, +} + +struct OutboxState { + next_sequence: u64, + items: u64, + bytes: u64, +} + +pub(crate) struct Outbox { + db: DB, + state: Mutex, + max_items: u64, + max_bytes: u64, +} + +impl Outbox { + pub(crate) fn open( + path: impl AsRef, + max_items: usize, + max_bytes: u64, + ) -> Result { + let mut options = rocksdb::Options::default(); + options.create_if_missing(true); + let db = DB::open(&options, path)?; + let next_sequence = db + .get(NEXT_SEQUENCE_KEY)? + .as_deref() + .and_then(|value| value.try_into().ok()) + .map(u64::from_be_bytes) + .unwrap_or(0); + let mut items = 0_u64; + let mut bytes = 0_u64; + for entry in db.iterator(IteratorMode::From(&[ITEM_PREFIX], Direction::Forward)) { + let (key, value) = entry?; + if key.first().copied() != Some(ITEM_PREFIX) { + break; + } + let item: OutboxItem = rmp_serde::from_slice(&value)?; + items = items.saturating_add(1); + bytes = bytes.saturating_add(item.encoded_bytes); + } + Ok(Self { + db, + state: Mutex::new(OutboxState { + next_sequence, + items, + bytes, + }), + max_items: max_items.min(u64::MAX as usize) as u64, + max_bytes, + }) + } + + pub(crate) fn identity(&self) -> Result { + let _state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(bytes) = self.db.get(IDENTITY_KEY)? { + return Ok(String::from_utf8_lossy(&bytes).into_owned()); + } + let mut random = [0_u8; 16]; + getrandom::fill(&mut random)?; + let identity = hex::encode(random); + self.db.put(IDENTITY_KEY, identity.as_bytes())?; + Ok(identity) + } + + pub(crate) fn append(&self, envelope: MetadataEnvelope) -> Result { + let encoded_bytes = rmp_serde::to_vec_named(&envelope)?.len() as u64; + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.items >= self.max_items + || state.bytes.saturating_add(encoded_bytes) > self.max_bytes + { + return Err(OutboxError::Full); + } + let sequence = state.next_sequence; + let item = OutboxItem { + sequence, + envelope, + encoded_bytes, + attempts: 0, + next_attempt_at: 0, + }; + let bytes = rmp_serde::to_vec_named(&item)?; + let mut batch = WriteBatch::default(); + batch.put(item_key(sequence), bytes); + batch.put(NEXT_SEQUENCE_KEY, sequence.saturating_add(1).to_be_bytes()); + self.db.write(batch)?; + state.next_sequence = sequence.saturating_add(1); + state.items = state.items.saturating_add(1); + state.bytes = state.bytes.saturating_add(encoded_bytes); + Ok(sequence) + } + + pub(crate) fn load_control( + &self, + ) -> Result, OutboxError> { + self.db + .get(CONTROL_KEY)? + .map(|bytes| rmp_serde::from_slice(&bytes).map_err(OutboxError::from)) + .transpose() + } + + pub(crate) fn store_control( + &self, + control: &super::protocol::ControlResponse, + ) -> Result<(), OutboxError> { + self.db + .put(CONTROL_KEY, rmp_serde::to_vec_named(control)?)?; + Ok(()) + } + + pub(crate) fn next_ready(&self, now: u64) -> Result, OutboxError> { + for entry in self + .db + .iterator(IteratorMode::From(&[ITEM_PREFIX], Direction::Forward)) + { + let (key, value) = entry?; + if key.first().copied() != Some(ITEM_PREFIX) { + break; + } + let item: OutboxItem = rmp_serde::from_slice(&value)?; + if item.next_attempt_at <= now { + return Ok(Some(item)); + } + } + Ok(None) + } + + pub(crate) fn acknowledge(&self, item: &OutboxItem) -> Result<(), OutboxError> { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + self.db.delete(item_key(item.sequence))?; + state.items = state.items.saturating_sub(1); + state.bytes = state.bytes.saturating_sub(item.encoded_bytes); + Ok(()) + } + + pub(crate) fn retry(&self, mut item: OutboxItem, now: u64) -> Result<(), OutboxError> { + item.attempts = item.attempts.saturating_add(1); + let exponent = item.attempts.min(6); + let delay = 1_u64 << exponent; + item.next_attempt_at = now.saturating_add(delay.min(60)); + self.db + .put(item_key(item.sequence), rmp_serde::to_vec_named(&item)?)?; + Ok(()) + } + + pub(crate) fn snapshot(&self) -> OutboxSnapshot { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + OutboxSnapshot { + items: state.items, + bytes: state.bytes, + } + } + + pub(crate) fn high_watermark(&self) -> bool { + let snapshot = self.snapshot(); + snapshot.items.saturating_mul(10) >= self.max_items.saturating_mul(9) + || snapshot.bytes.saturating_mul(10) >= self.max_bytes.saturating_mul(9) + } + + pub(crate) fn below_low_watermark(&self) -> bool { + let snapshot = self.snapshot(); + snapshot.items.saturating_mul(10) <= self.max_items.saturating_mul(7) + && snapshot.bytes.saturating_mul(10) <= self.max_bytes.saturating_mul(7) + } +} + +fn item_key(sequence: u64) -> [u8; 9] { + let mut key = [0_u8; 9]; + key[0] = ITEM_PREFIX; + key[1..].copy_from_slice(&sequence.to_be_bytes()); + key +} + +#[cfg(test)] +mod tests { + use dht_crawler::TorrentInfo; + use tempfile::TempDir; + + use super::*; + use crate::cluster::protocol::PROTOCOL_VERSION; + + fn envelope(hash: &str) -> MetadataEnvelope { + MetadataEnvelope { + protocol_version: PROTOCOL_VERSION, + node_id: "node".to_owned(), + lease_token: None, + torrent: TorrentInfo { + info_hash: hash.to_owned(), + magnet_link: format!("magnet:?xt=urn:btih:{hash}"), + name: "example".to_owned(), + total_size: 1, + files: vec![dht_crawler::FileInfo { + path: "example.txt".to_owned(), + size: 1, + }], + piece_length: 16_384, + peers: Vec::new(), + timestamp: 1, + }, + } + } + + #[test] + fn items_and_identity_survive_reopen() { + let directory = TempDir::new().unwrap(); + let identity; + { + let outbox = Outbox::open(directory.path(), 10, 1_000_000).unwrap(); + identity = outbox.identity().unwrap(); + outbox.append(envelope(&"01".repeat(20))).unwrap(); + } + let outbox = Outbox::open(directory.path(), 10, 1_000_000).unwrap(); + assert_eq!(outbox.identity().unwrap(), identity); + assert_eq!(outbox.snapshot().items, 1); + let item = outbox.next_ready(1).unwrap().unwrap(); + outbox.acknowledge(&item).unwrap(); + assert_eq!(outbox.snapshot().items, 0); + } + + #[test] + fn capacity_is_enforced_before_write() { + let directory = TempDir::new().unwrap(); + let outbox = Outbox::open(directory.path(), 1, 1_000_000).unwrap(); + outbox.append(envelope(&"01".repeat(20))).unwrap(); + assert!(matches!( + outbox.append(envelope(&"02".repeat(20))), + Err(OutboxError::Full) + )); + } +} diff --git a/src/search/src/cluster/protocol.rs b/src/search/src/cluster/protocol.rs new file mode 100644 index 0000000..89b8f42 --- /dev/null +++ b/src/search/src/cluster/protocol.rs @@ -0,0 +1,167 @@ +// 负责定义采集器与协调器之间稳定且有版本检查的传输数据 + +use serde::{Deserialize, Serialize}; + +use crate::{config::DhtConfig, domain::VerificationResult}; + +pub(crate) const PROTOCOL_VERSION: u16 = 2; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct CollectorDesiredConfig { + pub(crate) dht: DhtConfig, + pub(crate) verification_enabled: bool, + pub(crate) verification_max_active: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct RegisterRequest { + pub(crate) protocol_version: u16, + pub(crate) node_id: String, + pub(crate) name: String, + pub(crate) initial_config: CollectorDesiredConfig, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct ControlResponse { + pub(crate) protocol_version: u16, + pub(crate) revision: u64, + pub(crate) desired: CollectorDesiredConfig, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct CollectorMetrics { + pub(crate) state: String, + pub(crate) nodes: u64, + pub(crate) metadata_in_flight: u64, + pub(crate) metadata_succeeded: u64, + pub(crate) metadata_failed: u64, + pub(crate) udp_tx_packets: u64, + pub(crate) metadata_peer_attempts: u64, + pub(crate) metadata_peer_canceled: u64, + pub(crate) metadata_peer_failure_cache_hits: u64, + pub(crate) metadata_peer_failure_cache_entries: u64, + pub(crate) metadata_failure_timeout: u64, + pub(crate) metadata_failure_connect: u64, + pub(crate) metadata_failure_no_extension: u64, + pub(crate) metadata_failure_send: u64, + pub(crate) metadata_failure_size_limit: u64, + pub(crate) metadata_failure_sha1: u64, + pub(crate) metadata_failure_parse: u64, + pub(crate) metadata_failure_other: u64, + pub(crate) sample_queue_depth: u64, + pub(crate) sample_queue_capacity: u64, + pub(crate) resident_memory_bytes: Option, + pub(crate) private_memory_bytes: Option, + pub(crate) cpu_time_millis: Option, + pub(crate) thread_count: Option, + pub(crate) handle_count: Option, + pub(crate) outbox_items: u64, + pub(crate) outbox_bytes: u64, + pub(crate) last_error: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct HeartbeatRequest { + pub(crate) protocol_version: u16, + pub(crate) node_id: String, + pub(crate) applied_revision: u64, + pub(crate) metrics: CollectorMetrics, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct AdmissionRequest { + pub(crate) protocol_version: u16, + pub(crate) node_id: String, + pub(crate) info_hashes: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum AdmissionStatus { + Known, + Granted, + Leased, + Busy, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct AdmissionDecision { + pub(crate) info_hash: String, + pub(crate) status: AdmissionStatus, + pub(crate) lease_token: Option, + pub(crate) expires_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct AdmissionResponse { + pub(crate) decisions: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct MetadataEnvelope { + pub(crate) protocol_version: u16, + pub(crate) node_id: String, + pub(crate) lease_token: Option, + pub(crate) torrent: dht_crawler::TorrentInfo, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum MetadataSubmitStatus { + Inserted, + AlreadyKnown, + Rejected, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct MetadataSubmitResponse { + pub(crate) status: MetadataSubmitStatus, + pub(crate) reason: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct VerificationClaimRequest { + pub(crate) protocol_version: u16, + pub(crate) node_id: String, + pub(crate) limit: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct VerificationJob { + pub(crate) info_hash: String, + pub(crate) lease_token: String, + pub(crate) source_peers: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct VerificationClaimResponse { + pub(crate) jobs: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct VerificationResultRequest { + pub(crate) protocol_version: u16, + pub(crate) node_id: String, + pub(crate) lease_token: String, + pub(crate) info_hash: String, + pub(crate) result: VerificationResult, +} + +#[derive(Debug, Clone, Serialize)] +pub(crate) struct CollectorView { + pub(crate) node_id: String, + pub(crate) name: String, + pub(crate) revision: u64, + pub(crate) applied_revision: u64, + pub(crate) desired: CollectorDesiredConfig, + pub(crate) online: bool, + pub(crate) last_seen: Option, + pub(crate) metrics: CollectorMetrics, +} + +#[derive(Debug, Clone, Deserialize)] +pub(crate) struct CollectorConfigUpdate { + pub(crate) revision: u64, + pub(crate) name: String, + pub(crate) desired: CollectorDesiredConfig, +} diff --git a/src/search/src/cluster/registry.rs b/src/search/src/cluster/registry.rs new file mode 100644 index 0000000..0953741 --- /dev/null +++ b/src/search/src/cluster/registry.rs @@ -0,0 +1,126 @@ +// 负责使用独立 RocksDB 保存采集器身份和带修订号的期望配置 + +use std::{path::Path, sync::Mutex}; + +use rocksdb::{DB, Direction, IteratorMode}; +use serde::{Deserialize, Serialize}; + +use super::protocol::{CollectorDesiredConfig, RegisterRequest}; + +const NODE_PREFIX: u8 = b'n'; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum RegistryError { + #[error("采集器注册数据库失败: {0}")] + Rocks(#[from] rocksdb::Error), + #[error("采集器注册记录编码失败: {0}")] + Encode(#[from] rmp_serde::encode::Error), + #[error("采集器注册记录解码失败: {0}")] + Decode(#[from] rmp_serde::decode::Error), +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct CollectorRecord { + pub(crate) node_id: String, + pub(crate) name: String, + pub(crate) revision: u64, + pub(crate) desired: CollectorDesiredConfig, +} + +pub(crate) struct CollectorRegistry { + db: DB, + write_lock: Mutex<()>, +} + +impl CollectorRegistry { + pub(crate) fn open(path: impl AsRef) -> Result { + let mut options = rocksdb::Options::default(); + options.create_if_missing(true); + Ok(Self { + db: DB::open(&options, path)?, + write_lock: Mutex::new(()), + }) + } + + pub(crate) fn register( + &self, + request: &RegisterRequest, + ) -> Result { + let _guard = self + .write_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(record) = self.get(&request.node_id)? { + return Ok(record); + } + let record = CollectorRecord { + node_id: request.node_id.clone(), + name: request.name.trim().to_owned(), + revision: 1, + desired: request.initial_config.clone(), + }; + self.db.put(node_key(&record.node_id), encode(&record)?)?; + Ok(record) + } + + pub(crate) fn get(&self, node_id: &str) -> Result, RegistryError> { + self.db + .get(node_key(node_id))? + .map(|bytes| decode(&bytes)) + .transpose() + } + + pub(crate) fn list(&self) -> Result, RegistryError> { + let mut records = Vec::new(); + for entry in self + .db + .iterator(IteratorMode::From(&[NODE_PREFIX], Direction::Forward)) + { + let (key, value) = entry?; + if key.first().copied() != Some(NODE_PREFIX) { + break; + } + records.push(decode(&value)?); + } + Ok(records) + } + + pub(crate) fn update( + &self, + node_id: &str, + expected_revision: u64, + name: String, + desired: CollectorDesiredConfig, + ) -> Result, RegistryError> { + let _guard = self + .write_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut record) = self.get(node_id)? else { + return Ok(None); + }; + if record.revision != expected_revision { + return Ok(Some(record)); + } + record.name = name.trim().to_owned(); + record.desired = desired; + record.revision = record.revision.saturating_add(1); + self.db.put(node_key(node_id), encode(&record)?)?; + Ok(Some(record)) + } +} + +fn node_key(node_id: &str) -> Vec { + let mut key = Vec::with_capacity(1 + node_id.len()); + key.push(NODE_PREFIX); + key.extend_from_slice(node_id.as_bytes()); + key +} + +fn encode(record: &CollectorRecord) -> Result, RegistryError> { + Ok(rmp_serde::to_vec_named(record)?) +} + +fn decode(bytes: &[u8]) -> Result { + Ok(rmp_serde::from_slice(bytes)?) +} diff --git a/src/search/src/config.rs b/src/search/src/config.rs index 59599a4..95ca046 100644 --- a/src/search/src/config.rs +++ b/src/search/src/config.rs @@ -16,9 +16,11 @@ use clap::Parser; use crate::error::AppError; pub(crate) use model::{ - AppConfigDto, BackupConfig, DiagnosticsConfig, LogRotation, LoggingConfig, VerificationConfig, + AppConfigDto, BackupConfig, CoordinatorConfig, DhtConfig, DiagnosticsConfig, LogRotation, + LoggingConfig, ServiceRole, VerificationConfig, }; pub(crate) use runtime::AppConfig; +pub(crate) use runtime::dht_options_from; pub(crate) use service::{ConfigService, ConfigServiceError, ConfigSnapshot, ConfigUpdateRequest}; pub(crate) use store::{ConfigStore, TomlConfigStore}; @@ -37,6 +39,16 @@ pub(crate) struct Cli { no_file_logging: bool, #[arg(long)] restore_checkpoint: Option, + #[arg(long)] + role: Option, + #[arg(long)] + ingest_listen: Option, + #[arg(long)] + coordinator_url: Option, + #[arg(long)] + collector_name: Option, + #[arg(long)] + collector_state_dir: Option, } #[derive(Debug)] @@ -70,6 +82,35 @@ impl Cli { dto.logging.file_enabled = false; command_line_overrides.push("logging.file_enabled".to_owned()); } + if let Some(role) = self.role { + dto.service.role = match role.as_str() { + "standalone" => ServiceRole::Standalone, + "coordinator" => ServiceRole::Coordinator, + "collector" => ServiceRole::Collector, + _ => { + return Err(AppError::Config( + "role 必须是 standalone coordinator 或 collector".to_owned(), + )); + } + }; + command_line_overrides.push("service.role".to_owned()); + } + if let Some(listen) = self.ingest_listen { + dto.coordinator.ingest_listen = listen; + command_line_overrides.push("coordinator.ingest_listen".to_owned()); + } + if let Some(url) = self.coordinator_url { + dto.collector.coordinator_url = url; + command_line_overrides.push("collector.coordinator_url".to_owned()); + } + if let Some(name) = self.collector_name { + dto.collector.name = name; + command_line_overrides.push("collector.name".to_owned()); + } + if let Some(path) = self.collector_state_dir { + dto.collector.state_dir = path; + command_line_overrides.push("collector.state_dir".to_owned()); + } let base = config_path .parent() .ok_or_else(|| AppError::Config("配置文件没有父目录".to_owned()))?; @@ -157,6 +198,11 @@ mod tests { web_dir: None, no_file_logging: false, restore_checkpoint: None, + role: None, + ingest_listen: None, + coordinator_url: None, + collector_name: None, + collector_state_dir: None, } .load() .unwrap(); @@ -186,6 +232,11 @@ mod tests { web_dir: None, no_file_logging: false, restore_checkpoint: None, + role: None, + ingest_listen: None, + coordinator_url: None, + collector_name: None, + collector_state_dir: None, } .load() .unwrap_err(); @@ -209,6 +260,11 @@ mod tests { web_dir: Some(PathBuf::from("/dht-search/web")), no_file_logging: true, restore_checkpoint: None, + role: None, + ingest_listen: None, + coordinator_url: None, + collector_name: None, + collector_state_dir: None, } .load() .unwrap(); diff --git a/src/search/src/config/model.rs b/src/search/src/config/model.rs index ccaa04e..c15e956 100644 --- a/src/search/src/config/model.rs +++ b/src/search/src/config/model.rs @@ -9,6 +9,9 @@ use crate::domain::ContentFilterConfig; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(default, deny_unknown_fields)] pub(crate) struct AppConfigDto { + pub(crate) service: ServiceConfig, + pub(crate) coordinator: CoordinatorConfig, + pub(crate) collector: CollectorConfig, pub(crate) data_dir: PathBuf, pub(crate) content_filter: ContentFilterConfigDto, pub(crate) persistence_queue_capacity: usize, @@ -24,6 +27,41 @@ pub(crate) struct AppConfigDto { pub(crate) verification: VerificationConfig, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum ServiceRole { + #[default] + Standalone, + Coordinator, + Collector, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub(crate) struct ServiceConfig { + pub(crate) role: ServiceRole, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub(crate) struct CoordinatorConfig { + pub(crate) ingest_listen: SocketAddr, + pub(crate) lease_secs: u64, + pub(crate) offline_after_secs: u64, + pub(crate) max_leases: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub(crate) struct CollectorConfig { + pub(crate) coordinator_url: String, + pub(crate) name: String, + pub(crate) state_dir: PathBuf, + pub(crate) outbox_max_bytes: u64, + pub(crate) outbox_max_items: usize, + pub(crate) heartbeat_secs: u64, +} + #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(default, deny_unknown_fields)] pub(crate) struct ContentFilterConfigDto { @@ -146,6 +184,9 @@ pub(crate) enum NetworkMode { impl Default for AppConfigDto { fn default() -> Self { Self { + service: ServiceConfig::default(), + coordinator: CoordinatorConfig::default(), + collector: CollectorConfig::default(), data_dir: PathBuf::from("data"), content_filter: default_content_filter(), persistence_queue_capacity: 8_192, @@ -163,6 +204,38 @@ impl Default for AppConfigDto { } } +impl Default for ServiceConfig { + fn default() -> Self { + Self { + role: ServiceRole::Standalone, + } + } +} + +impl Default for CoordinatorConfig { + fn default() -> Self { + Self { + ingest_listen: SocketAddr::from(([127, 0, 0, 1], 8081)), + lease_secs: 120, + offline_after_secs: 15, + max_leases: 1_000_000, + } + } +} + +impl Default for CollectorConfig { + fn default() -> Self { + Self { + coordinator_url: "http://127.0.0.1:8081".to_owned(), + name: "本机采集器".to_owned(), + state_dir: PathBuf::from("data/collector"), + outbox_max_bytes: 2 * 1024 * 1024 * 1024, + outbox_max_items: 100_000, + heartbeat_secs: 2, + } + } +} + fn default_content_filter() -> ContentFilterConfigDto { ContentFilterConfigDto { torrent_name_patterns: vec!["*【加QQ *】*".to_owned()], diff --git a/src/search/src/config/runtime.rs b/src/search/src/config/runtime.rs index 52e5366..4a4fe91 100644 --- a/src/search/src/config/runtime.rs +++ b/src/search/src/config/runtime.rs @@ -22,6 +22,7 @@ impl AppConfig { dto.backup.directory = resolve_path(base, dto.backup.directory); dto.diagnostics.database = resolve_path(base, dto.diagnostics.database); dto.http.web_dir = resolve_path(base, dto.http.web_dir); + dto.collector.state_dir = resolve_path(base, dto.collector.state_dir); let config = Self(dto); config.validate()?; Ok(config) @@ -32,57 +33,7 @@ impl AppConfig { } pub(crate) fn dht_options(&self) -> DHTOptions { - let defaults = DHTOptions::default(); - DHTOptions { - port: self.dht.port, - netmode: self.dht.netmode.into(), - hash_queue_capacity: self.dht.hash_queue_capacity, - max_outbound_queries_per_second: self.dht.max_outbound_queries_per_second, - outbound_query_burst: self.dht.outbound_query_burst, - metadata: MetadataOptions { - timeout_secs: self.dht.metadata_timeout_secs, - max_queue_size: self.dht.metadata_queue_capacity, - max_worker_count: self.dht.metadata_workers, - max_connects_per_second: self.dht.metadata_connects_per_second, - max_metadata_size_bytes: self.metadata_limits.max_metadata_bytes, - ..defaults.metadata - }, - peer_lookup: PeerLookupOptions { - max_lookups_per_second: self.dht.peer_lookups_per_second, - burst: self.dht.peer_lookups_per_second.max(1), - max_active_lookups: self.dht.peer_lookup_max_active, - }, - sample_infohashes: SampleInfohashesOptions { - max_queries_per_second: self.dht.sample_queries_per_second, - burst: self.dht.sample_queries_per_second.max(1), - max_in_flight: self.dht.sample_max_in_flight, - new_node_sample_percent: self.dht.sample_new_node_percent, - candidate_queue_capacity: self.dht.sample_candidate_queue_capacity, - fallback_to_iterative: self.dht.sample_fallback_to_iterative, - ..defaults.sample_infohashes - }, - crawl: CrawlOptions { - pool: PoolOptions { - ..defaults.crawl.pool - }, - rate_limit: RateLimitOptions { - max_find_node_rate_per_sec: self.dht.find_node_queries_per_second, - burst: self.dht.outbound_query_burst, - max_in_flight: self.dht.find_node_max_in_flight, - max_new_destinations_per_minute: self.dht.new_destinations_per_minute, - ..defaults.crawl.rate_limit - }, - bootstrap: BootstrapOptions { - ..defaults.crawl.bootstrap - }, - target: TargetOptions { - ..defaults.crawl.target - }, - scheduler: SchedulerOptions { - ..defaults.crawl.scheduler - }, - }, - } + dht_options_from(&self.dht, self.metadata_limits.max_metadata_bytes) } pub(crate) fn metadata_limits(&self) -> MetadataLimits { @@ -101,6 +52,24 @@ impl AppConfig { "persistence_queue_capacity 必须大于零".to_owned(), )); } + if self.coordinator.lease_secs == 0 + || self.coordinator.offline_after_secs == 0 + || self.coordinator.max_leases == 0 + { + return Err(AppError::Config( + "协调器租约离线判断和租约容量必须大于零".to_owned(), + )); + } + if self.collector.name.trim().is_empty() + || self.collector.coordinator_url.trim().is_empty() + || self.collector.outbox_max_bytes == 0 + || self.collector.outbox_max_items == 0 + || self.collector.heartbeat_secs == 0 + { + return Err(AppError::Config( + "采集器名称地址待发送箱容量和心跳间隔必须有效".to_owned(), + )); + } if self.stats_interval_secs == 0 { return Err(AppError::Config( "stats_interval_secs 必须大于零".to_owned(), @@ -191,6 +160,63 @@ impl AppConfig { } } +pub(crate) fn dht_options_from( + config: &super::model::DhtConfig, + max_metadata_bytes: usize, +) -> DHTOptions { + let defaults = DHTOptions::default(); + DHTOptions { + port: config.port, + netmode: config.netmode.into(), + hash_queue_capacity: config.hash_queue_capacity, + max_outbound_queries_per_second: config.max_outbound_queries_per_second, + outbound_query_burst: config.outbound_query_burst, + metadata: MetadataOptions { + timeout_secs: config.metadata_timeout_secs, + max_queue_size: config.metadata_queue_capacity, + max_worker_count: config.metadata_workers, + max_connects_per_second: config.metadata_connects_per_second, + max_metadata_size_bytes: max_metadata_bytes, + ..defaults.metadata + }, + peer_lookup: PeerLookupOptions { + max_lookups_per_second: config.peer_lookups_per_second, + burst: config.peer_lookups_per_second.max(1), + max_active_lookups: config.peer_lookup_max_active, + }, + sample_infohashes: SampleInfohashesOptions { + max_queries_per_second: config.sample_queries_per_second, + burst: config.sample_queries_per_second.max(1), + max_in_flight: config.sample_max_in_flight, + new_node_sample_percent: config.sample_new_node_percent, + candidate_queue_capacity: config.sample_candidate_queue_capacity, + fallback_to_iterative: config.sample_fallback_to_iterative, + ..defaults.sample_infohashes + }, + crawl: CrawlOptions { + pool: PoolOptions { + ..defaults.crawl.pool + }, + rate_limit: RateLimitOptions { + max_find_node_rate_per_sec: config.find_node_queries_per_second, + burst: config.outbound_query_burst, + max_in_flight: config.find_node_max_in_flight, + max_new_destinations_per_minute: config.new_destinations_per_minute, + ..defaults.crawl.rate_limit + }, + bootstrap: BootstrapOptions { + ..defaults.crawl.bootstrap + }, + target: TargetOptions { + ..defaults.crawl.target + }, + scheduler: SchedulerOptions { + ..defaults.crawl.scheduler + }, + }, + } +} + impl Deref for AppConfig { type Target = AppConfigDto; diff --git a/src/search/src/config/service.rs b/src/search/src/config/service.rs index 10359c1..ceea615 100644 --- a/src/search/src/config/service.rs +++ b/src/search/src/config/service.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use crate::error::AppError; -use super::{AppConfig, AppConfigDto, ConfigStore}; +use super::{AppConfig, AppConfigDto, ConfigStore, DhtConfig}; #[derive(Debug, Clone, Serialize)] pub(crate) struct ConfigSnapshot { @@ -123,20 +123,17 @@ impl ConfigService { Ok(self.snapshot_from(&state)) } - pub(crate) fn set_dht_enabled( + pub(crate) fn set_dht_config( &self, - enabled: bool, + dht: DhtConfig, ) -> Result { let mut state = self .inner .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if state.config.dht.enabled == enabled { - return Ok(self.snapshot_from(&state)); - } let mut candidate = state.config.clone(); - candidate.dht.enabled = enabled; + candidate.dht = dht; AppConfig::resolve(candidate.clone(), &self.inner.base) .map_err(|error| ConfigServiceError::Validation(error.to_string()))?; let candidate_revision = revision(&candidate) @@ -210,20 +207,6 @@ mod tests { assert_eq!(stored, Some(config)); } - #[test] - fn live_crawler_toggle_is_persisted_without_requiring_restart() { - let directory = TempDir::new().unwrap(); - let service = service(&directory); - let updated = service.set_dht_enabled(false).unwrap(); - assert!(!updated.config.dht.enabled); - assert!(!updated.restart_required); - let stored = TomlConfigStore::new(directory.path().join("service.toml")) - .load() - .unwrap() - .unwrap(); - assert!(!stored.dht.enabled); - } - #[test] fn content_filter_update_is_live_and_does_not_require_restart() { let directory = TempDir::new().unwrap(); diff --git a/src/search/src/crawler/runtime.rs b/src/search/src/crawler/runtime.rs index 4b6e5c5..f88d98c 100644 --- a/src/search/src/crawler/runtime.rs +++ b/src/search/src/crawler/runtime.rs @@ -27,7 +27,6 @@ use crate::disk_guard::DiskGuard; #[derive(Debug, Clone, Copy)] pub(crate) struct CrawlerRuntimeStatus { pub(crate) enabled: bool, - pub(crate) transitioning: bool, } #[derive(Clone, Default)] @@ -65,7 +64,6 @@ struct CrawlerRuntimeInner { stats: CrawlerStats, verification: RwLock>, enabled: AtomicBool, - transitioning: AtomicBool, state: tokio::sync::Mutex, } @@ -95,7 +93,6 @@ impl CrawlerRuntime { stats: CrawlerStats::default(), verification: RwLock::new(None), enabled: AtomicBool::new(false), - transitioning: AtomicBool::new(false), state: tokio::sync::Mutex::new(RunningState::default()), }), } @@ -104,7 +101,6 @@ impl CrawlerRuntime { pub(crate) fn status(&self) -> CrawlerRuntimeStatus { CrawlerRuntimeStatus { enabled: self.inner.enabled.load(Ordering::Acquire), - transitioning: self.inner.transitioning.load(Ordering::Acquire), } } @@ -125,14 +121,11 @@ impl CrawlerRuntime { } pub(crate) async fn set_enabled(&self, enabled: bool) -> Result<(), AppError> { - self.inner.transitioning.store(true, Ordering::Release); - let result = if enabled { + if enabled { self.start().await } else { self.stop().await - }; - self.inner.transitioning.store(false, Ordering::Release); - result + } } pub(crate) async fn shutdown(&self) { @@ -225,14 +218,18 @@ impl CrawlerRuntime { Ok(()) } - #[cfg(test)] - pub(crate) fn set_verification_for_test(&self, ingress: VerificationIngress) { + pub(crate) fn install_verification(&self, ingress: VerificationIngress) { *self .inner .verification .write() .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(ingress); } + + #[cfg(test)] + pub(crate) fn set_verification_for_test(&self, ingress: VerificationIngress) { + self.install_verification(ingress); + } } fn configure_callbacks( diff --git a/src/search/src/diagnostics/mod.rs b/src/search/src/diagnostics/mod.rs index 6935881..c9b6e3e 100644 --- a/src/search/src/diagnostics/mod.rs +++ b/src/search/src/diagnostics/mod.rs @@ -2,7 +2,7 @@ mod http; mod model; -mod process; +pub(crate) mod process; mod store; use std::{ @@ -17,6 +17,7 @@ use std::{ }; use crate::{ + cluster::coordinator::CoordinatorRuntime, search::SearchRuntime, storage::{RocksTorrentRepository, StorageDiagnostics as RocksDiagnostics, TorrentRepository}, }; @@ -29,10 +30,13 @@ use crate::{ }; pub(crate) use http::HttpStats; +use model::{ + CollectorDiagnostics, DiagnosticsStatus, RuntimeDiagnostics, SearchDiagnostics, + StorageDiagnostics, +}; pub(crate) use model::{ CurrentDiagnosticsResponse, DiagnosticHistory, DiagnosticSample, HistoryResolution, }; -use model::{DiagnosticsStatus, RuntimeDiagnostics, SearchDiagnostics, StorageDiagnostics}; use store::{DiagnosticStore, DiagnosticStoreError}; #[derive(Clone)] @@ -43,6 +47,7 @@ pub(crate) struct DiagnosticSources { pub(crate) persistence: PersistenceIngress, pub(crate) disk_guard: DiskGuard, pub(crate) http: HttpStats, + pub(crate) coordinator: Option, } #[derive(Clone)] @@ -271,6 +276,42 @@ fn collect_sample(sources: &DiagnosticSources, session_started_at: u64) -> Diagn }); let inventory = sources.repository.index_inventory(); let search = sources.search.diagnostics(); + let collectors = sources + .coordinator + .as_ref() + .and_then(|coordinator| coordinator.collectors().ok()) + .unwrap_or_default(); + let aggregate = sources + .coordinator + .as_ref() + .filter(|_| !collectors.is_empty()) + .and_then(|coordinator| coordinator.aggregate_metrics().ok()); + let runtime = aggregate.as_ref().map_or_else( + || RuntimeDiagnostics { + nodes: saturating_u64(dht.node_pool_size), + udp_tx_packets: observability.udp_tx_packets, + metadata_in_flight: saturating_u64(dht.metadata_in_flight), + metadata_succeeded: dht.metadata_peer_succeeded, + metadata_failed: dht.metadata_peer_failed, + sample_queue_depth: saturating_u64(dht.sample_candidate_queue_depth), + sample_queue_capacity: saturating_u64(dht.sample_candidate_queue_capacity), + persistence_queue_depth: saturating_u64(persistence.queue_depth), + indexed_documents: search.documents, + disk_available_bytes: disk.available_bytes, + }, + |metrics| RuntimeDiagnostics { + nodes: metrics.nodes, + udp_tx_packets: metrics.udp_tx_packets, + metadata_in_flight: metrics.metadata_in_flight, + metadata_succeeded: metrics.metadata_succeeded, + metadata_failed: metrics.metadata_failed, + sample_queue_depth: metrics.sample_queue_depth, + sample_queue_capacity: metrics.sample_queue_capacity, + persistence_queue_depth: saturating_u64(persistence.queue_depth), + indexed_documents: search.documents, + disk_available_bytes: disk.available_bytes, + }, + ); DiagnosticSample { captured_at: unix_timestamp(), session_started_at, @@ -298,18 +339,16 @@ fn collect_sample(sources: &DiagnosticSources, session_started_at: u64) -> Diagn documents_skipped: search.documents_skipped, }, http: sources.http.snapshot(), - runtime: RuntimeDiagnostics { - nodes: saturating_u64(dht.node_pool_size), - udp_tx_packets: observability.udp_tx_packets, - metadata_in_flight: saturating_u64(dht.metadata_in_flight), - metadata_succeeded: dht.metadata_peer_succeeded, - metadata_failed: dht.metadata_peer_failed, - sample_queue_depth: saturating_u64(dht.sample_candidate_queue_depth), - sample_queue_capacity: saturating_u64(dht.sample_candidate_queue_capacity), - persistence_queue_depth: saturating_u64(persistence.queue_depth), - indexed_documents: search.documents, - disk_available_bytes: disk.available_bytes, - }, + runtime, + collectors: collectors + .into_iter() + .map(|collector| CollectorDiagnostics { + node_id: collector.node_id, + name: collector.name, + online: collector.online, + metrics: collector.metrics, + }) + .collect(), } } @@ -374,6 +413,7 @@ mod tests { persistence: persistence.ingress.clone(), disk_guard, http: HttpStats::default(), + coordinator: None, }, ) .unwrap(); diff --git a/src/search/src/diagnostics/model.rs b/src/search/src/diagnostics/model.rs index 8a2ffdb..31f7b99 100644 --- a/src/search/src/diagnostics/model.rs +++ b/src/search/src/diagnostics/model.rs @@ -2,6 +2,8 @@ use serde::{Deserialize, Serialize}; +use crate::cluster::protocol::CollectorMetrics; + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub(crate) struct DiagnosticSample { pub(crate) captured_at: u64, @@ -12,6 +14,16 @@ pub(crate) struct DiagnosticSample { #[serde(default)] pub(crate) http: HttpDiagnostics, pub(crate) runtime: RuntimeDiagnostics, + #[serde(default)] + pub(crate) collectors: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct CollectorDiagnostics { + pub(crate) node_id: String, + pub(crate) name: String, + pub(crate) online: bool, + pub(crate) metrics: CollectorMetrics, } #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] diff --git a/src/search/src/diagnostics/store.rs b/src/search/src/diagnostics/store.rs index 015c37c..aa291bd 100644 --- a/src/search/src/diagnostics/store.rs +++ b/src/search/src/diagnostics/store.rs @@ -168,6 +168,7 @@ mod tests { search: SearchDiagnostics::default(), http: HttpDiagnostics::default(), runtime: RuntimeDiagnostics::default(), + collectors: Vec::new(), } } diff --git a/src/search/src/entry.rs b/src/search/src/entry.rs index 8ace92d..82e496b 100644 --- a/src/search/src/entry.rs +++ b/src/search/src/entry.rs @@ -2,7 +2,7 @@ use clap::Parser; -use crate::{backup, config, error, telemetry}; +use crate::{backup, cluster, config, error, telemetry}; pub async fn run_cli() -> std::process::ExitCode { let startup = match config::Cli::parse().load() { @@ -38,6 +38,8 @@ pub async fn run_cli() -> std::process::ExitCode { } }) .map_err(error::AppError::from) + } else if startup.app.service.role == config::ServiceRole::Collector { + cluster::collector::run(startup.app).await } else { crate::app::run(startup.app, startup.config_service).await }; diff --git a/src/search/src/error.rs b/src/search/src/error.rs index 6453ff5..643e820 100644 --- a/src/search/src/error.rs +++ b/src/search/src/error.rs @@ -28,4 +28,6 @@ pub(crate) enum AppError { IndexWorker(String), #[error("运行诊断失败: {0}")] Diagnostics(String), + #[error("分布式采集失败: {0}")] + Cluster(String), } diff --git a/src/search/src/lib.rs b/src/search/src/lib.rs index 8bfc47c..843733b 100644 --- a/src/search/src/lib.rs +++ b/src/search/src/lib.rs @@ -7,6 +7,8 @@ mod app; #[cfg(feature = "rocksdb-storage")] mod backup; #[cfg(feature = "rocksdb-storage")] +mod cluster; +#[cfg(feature = "rocksdb-storage")] mod config; #[cfg(feature = "rocksdb-storage")] mod crawler; diff --git a/src/search/src/storage/repository.rs b/src/search/src/storage/repository.rs index 7be03cc..6189091 100644 --- a/src/search/src/storage/repository.rs +++ b/src/search/src/storage/repository.rs @@ -17,6 +17,13 @@ pub trait TorrentRepository: Send + Sync { fn upsert(&self, observation: TorrentRecord) -> Result; + fn insert_if_absent(&self, observation: TorrentRecord) -> Result { + if self.contains(observation.info_hash)? { + return Ok(false); + } + self.upsert(observation).map(|_| true) + } + fn rejection(&self, info_hash: InfoHash) -> Result, StorageError>; fn record_rejection(&self, rejection: RejectedMetadata) -> Result<(), StorageError>; diff --git a/src/search/src/storage/rocks.rs b/src/search/src/storage/rocks.rs index 54ccaf0..0e269f6 100644 --- a/src/search/src/storage/rocks.rs +++ b/src/search/src/storage/rocks.rs @@ -414,6 +414,49 @@ impl TorrentRepository for RocksTorrentRepository { Ok(UpsertOutcome::Inserted) } + fn insert_if_absent(&self, mut observation: TorrentRecord) -> Result { + 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() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if self.get(observation.info_hash)?.is_some() { + return Ok(false); + } + let mut batch = WriteBatch::default(); + batch.put( + torrent_key(observation.info_hash), + Self::encode(&observation)?, + ); + batch.delete(rejected_metadata_key(observation.info_hash)); + let mut delta = InventoryDelta { + stored_torrents: 1, + ..InventoryDelta::default() + }; + if observation.searchable { + batch.put( + content_member_key(&observation.content_key, observation.info_hash), + [], + ); + let group_delta = self.dirty_group(&mut batch, &observation.content_key, true)?; + delta.searchable_groups = group_delta.searchable_groups; + delta.pending_documents = group_delta.pending_documents; + } + self.commit_inventory_batch(batch, delta)?; + Ok(true) + } + fn rejection(&self, info_hash: InfoHash) -> Result, StorageError> { self.db .get(rejected_metadata_key(info_hash))? diff --git a/src/search/src/verification.rs b/src/search/src/verification.rs index 91568fc..51e5eaf 100644 --- a/src/search/src/verification.rs +++ b/src/search/src/verification.rs @@ -106,6 +106,19 @@ impl VerificationIngress { ) } + pub(crate) fn queue_only( + repository: Arc, + capacity: usize, + disk_guard: DiskGuard, + ) -> Self { + let stats = VerificationStats::default(); + stats.inner.queue_depth.store( + repository.verification_queue_len().unwrap_or_default() as u64, + Ordering::Relaxed, + ); + Self::new(repository, capacity, stats, disk_guard) + } + pub(crate) async fn enqueue(&self, hashes: Vec, priority: VerificationPriority) { if hashes.is_empty() { return; diff --git a/src/web/src/App.vue b/src/web/src/App.vue index 19202b4..dcdae50 100644 --- a/src/web/src/App.vue +++ b/src/web/src/App.vue @@ -1,20 +1,16 @@ @@ -86,11 +49,6 @@ onMounted(() => {
-
diff --git a/src/web/src/lib/api.ts b/src/web/src/lib/api.ts index c497e10..764aca2 100644 --- a/src/web/src/lib/api.ts +++ b/src/web/src/lib/api.ts @@ -2,7 +2,8 @@ import type { ContentVariants, ConfigSnapshot, ConfigUpdateRequest, - CrawlerStatus, + CollectorConfigUpdate, + CollectorView, CurrentDiagnostics, DiagnosticHistory, SearchPage, @@ -80,14 +81,14 @@ export function updateConfig(input: ConfigUpdateRequest, signal?: AbortSignal): }) } -export function getCrawlerStatus(signal?: AbortSignal): Promise { - return request('/crawler', signal) +export function getCollectors(signal?: AbortSignal): Promise { + return request('/collectors', signal) } -export function updateCrawlerStatus(enabled: boolean, signal?: AbortSignal): Promise { - return request('/crawler', signal, { +export function updateCollector(nodeId: string, input: CollectorConfigUpdate, signal?: AbortSignal): Promise { + return request(`/collectors/${encodeURIComponent(nodeId)}`, signal, { method: 'PUT', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ enabled }), + body: JSON.stringify(input), }) } diff --git a/src/web/src/pages/CollectorsPage.vue b/src/web/src/pages/CollectorsPage.vue new file mode 100644 index 0000000..a07425a --- /dev/null +++ b/src/web/src/pages/CollectorsPage.vue @@ -0,0 +1,165 @@ + + + diff --git a/src/web/src/pages/ConfigPage.vue b/src/web/src/pages/ConfigPage.vue index 66763a0..e796dc1 100644 --- a/src/web/src/pages/ConfigPage.vue +++ b/src/web/src/pages/ConfigPage.vue @@ -27,11 +27,6 @@ interface ConfigSection { const sections: ConfigSection[] = [ { title: '设置', fields: [ - { path: 'dht.netmode', label: '网络模式', description: '选择 DHT 使用的 IP 协议族', type: 'select', options: [{ value: 'ipv4-only', label: '仅 IPv4' }, { value: 'ipv6-only', label: '仅 IPv6' }, { value: 'dual-stack', label: '双栈' }] }, - { path: 'dht.max_outbound_queries_per_second', label: 'UDP 总速率', description: '所有主动 DHT 查询的每秒硬上限', type: 'number' }, - { path: 'dht.metadata_timeout_secs', label: 'Metadata 超时', description: '单次 Metadata 下载的总超时秒数', type: 'number' }, - { path: 'dht.metadata_workers', label: 'Metadata Worker', description: '同时处理 Metadata 下载的最大任务数', type: 'number' }, - { path: 'dht.metadata_connects_per_second', label: 'Metadata 建连速率', description: '每秒发起 TCP 连接的硬上限', type: 'number' }, { path: 'metadata_limits.max_metadata_bytes', label: '最大 Metadata', description: '单条 Metadata 允许占用的最大空间', type: 'bytes' }, { path: 'metadata_limits.max_files', label: '最大文件数', description: '单个种子允许包含的文件数量', type: 'number' }, { path: 'persistence_queue_capacity', label: '持久化队列容量', description: 'Metadata 到 RocksDB 的有界队列', type: 'number' }, @@ -218,7 +213,7 @@ onBeforeRouteLeave(() => !dirty.value || window.confirm('配置尚未保存， -
+