diff --git a/Cargo.toml b/Cargo.toml index 766e57d..06e6a48 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-feeder-sdk" -version = "1.3.0" +version = "1.4.0" edition = "2021" rust-version = "1.85" license = "MIT" @@ -31,6 +31,14 @@ publish = false # and the libp2p wire. Keep heavy deps (librqbit, beetswap, cid, libp2p) OUT # of here so feeders that don't need them don't pay for them. +[features] +# The meta-share transport-plugin contract (`src/transport/`): the hull's +# client, the plugins' axum harness, the focus replica, shared error/range +# types. Off by default so feeders build exactly what they always did. +transport = ["dep:cid", "dep:libc", "dep:tokio-util"] +# Conformance checks any transport implementation must pass. +testkit = ["transport"] + [dependencies] # Async runtime + HTTP server (the serve_feeders harness) tokio = { version = "1.40", features = ["full"] } @@ -58,6 +66,14 @@ regex = "1" # cache.rs. Pure-Rust single-process embedded store. redb = "2" +# CID parsing for the transport contract only (`transport::cid`). Optional so +# feeders don't pull it. +cid = { version = "0.11", optional = true } +# statvfs(2) for `transport::fsutil::available_bytes`. +libc = { version = "0.2", optional = true } +# `ReaderStream` for `transport::file`. +tokio-util = { version = "0.7", features = ["io"], optional = true } + [dev-dependencies] tempfile = "3" tokio = { version = "1.40", features = ["full", "test-util"] } diff --git a/src/lib.rs b/src/lib.rs index 18b1c64..80cb5ee 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,10 @@ //! hashing-into-the-blockstore, and the meta-core store-back to itself. //! //! Deliberately libp2p-free and blockstore-free — see this crate's `Cargo.toml`. +//! +//! With the `transport` feature it also carries the **transport-plugin** +//! contract ([`transport`]): the same split, applied to meta-share — a +//! protocol-agnostic hull and one sidecar per byte-moving protocol. // `redb::Error` is a large enum; the cache wrappers (`cache.rs`) return it // directly rather than boxing on every embedded-DB call. Boxing each would be @@ -36,6 +40,8 @@ pub mod plugin; pub mod query; pub mod query_eval; pub mod serve; +#[cfg(feature = "transport")] +pub mod transport; pub mod types; pub use budget::RateBudget; diff --git a/src/transport/cid.rs b/src/transport/cid.rs new file mode 100644 index 0000000..73deee6 --- /dev/null +++ b/src/transport/cid.rs @@ -0,0 +1,368 @@ +//! CID parsing and the CID-*shape* knowledge every transport process shares. +//! +//! Not protocol logic — these are pure decoders over the multihash standard, +//! which is the one thing meta-share's hull stays tied to. Moved verbatim from +//! meta-share (`blockstore.rs` for the parse, `bt.rs` for the torrent shapes) so +//! the hull and the plugins classify a cid identically: a drift here would route +//! a torrent cid to the IPFS walker or vice versa. + +use std::str::FromStr; + +use anyhow::{Context, Result}; +use cid::CidGeneric; + +/// Multihash byte ceiling. Must match the gateway and meta-search side — it +/// feeds beetswap's `Behaviour` in the ipfs plugin. +pub const MAX_MULTIHASH_SIZE: usize = 64; + +/// The CID type every MetaMesh transport speaks. +pub type MsCid = CidGeneric; + +/// Parse a `Record.cid` into an [`MsCid`]. +/// +/// **Strict**: a bare CIDv1 multibase string and nothing else, matching +/// `cid.Decode` in meta-core's `internal/cid/rank.go`. The `:` token +/// form and CIDv0 (`Qm…`) are both rejected — see meta-share's +/// `blockstore::parse_record_cid` history for why each one bit. +pub fn parse_record_cid(cid_str: &str) -> Result { + if cid_str.contains(':') { + anyhow::bail!( + "parse record cid `{cid_str}`: the `:` token form was removed; \ + pass the bare CIDv1 (the multicodec already names the algorithm)" + ); + } + let parsed = + MsCid::from_str(cid_str).with_context(|| format!("parse record cid `{cid_str}`"))?; + if parsed.version() != cid::Version::V1 { + anyhow::bail!( + "parse record cid `{cid_str}`: only CIDv1 is supported, got {:?}", + parsed.version() + ); + } + Ok(parsed) +} + +// --------------------------------------------------------------------------- +// BitTorrent cid shapes +// --------------------------------------------------------------------------- + +/// Multihash code for sha1 — a v1 infohash is the sha1 of the info dict. +pub const MH_SHA1: u64 = 0x11; +const BT_V1_INFOHASH_LEN: usize = 20; +const BT_V2_INFOHASH_LEN: usize = 32; +/// `btih-v1-file`: one file inside a v1 torrent; digest ` ‖ varint(index)`. +pub const BTIH_V1_FILE_CODEC: u64 = 0x1001; +/// `btih-v2-file`: recognised so the error is precise, not yet fetchable. +pub const BTIH_V2_FILE_CODEC: u64 = 0x1002; + +/// Is this a whole-torrent v1 infohash cid (sha1 multihash, 20 bytes)? The raw +/// codec `0x55` is shared with IPFS raw leaves, so key off the multihash. +pub fn is_btih(mscid: &MsCid) -> bool { + let mh = mscid.hash(); + mh.code() == MH_SHA1 && mh.digest().len() == BT_V1_INFOHASH_LEN +} + +/// The raw 20-byte v1 infohash of a `btih:` cid. Errors (a `400` upstream) for +/// anything that isn't sha1/20. +pub fn infohash20_from_mscid(mscid: &MsCid) -> Result<[u8; 20]> { + let mh = mscid.hash(); + anyhow::ensure!( + mh.code() == MH_SHA1, + "not a v1 BitTorrent infohash CID (multihash code 0x{:x}, want 0x11/sha1)", + mh.code() + ); + <[u8; 20]>::try_from(mh.digest()) + .map_err(|_| anyhow::anyhow!("infohash digest is {} bytes, want 20", mh.digest().len())) +} + +/// A decoded per-file torrent cid. +pub struct BtFileRef { + pub infohash20: [u8; 20], + pub file_index: usize, +} + +/// Is this a per-file torrent cid (`0x1001` / `0x1002`)? +pub fn is_btih_file(mscid: &MsCid) -> bool { + matches!(mscid.codec(), BTIH_V1_FILE_CODEC | BTIH_V2_FILE_CODEC) +} + +/// Decode a `btih-v1-file` cid into `(infohash, file_index)`; v2 is rejected +/// pending swarm support. +pub fn decode_btih_file(mscid: &MsCid) -> Result { + let infohash_len = match mscid.codec() { + BTIH_V1_FILE_CODEC => BT_V1_INFOHASH_LEN, + BTIH_V2_FILE_CODEC => { + let _ = BT_V2_INFOHASH_LEN; // documents the v2 digest shape + anyhow::bail!("btih-v2-file CIDs are not yet supported by the fetch tier"); + } + other => anyhow::bail!("not a btih-file CID (codec 0x{other:x})"), + }; + let digest = mscid.hash().digest(); + anyhow::ensure!( + digest.len() > infohash_len, + "btih-file digest is {} bytes, need > {} (infohash + file-index varint)", + digest.len(), + infohash_len + ); + let (ih, idx_bytes) = digest.split_at(infohash_len); + let infohash20 = <[u8; 20]>::try_from(ih) + .map_err(|_| anyhow::anyhow!("btih-file infohash slice is not 20 bytes"))?; + let (file_index, consumed) = read_uvarint(idx_bytes)?; + anyhow::ensure!( + consumed == idx_bytes.len(), + "btih-file digest has {} trailing byte(s) after the file-index varint", + idx_bytes.len() - consumed + ); + let file_index = usize::try_from(file_index) + .map_err(|_| anyhow::anyhow!("btih-file index {file_index} overflows usize"))?; + Ok(BtFileRef { infohash20, file_index }) +} + +/// Minimal unsigned-LEB128 reader → `(value, bytes_consumed)`. Rejects truncated +/// and overflowing encodings. +pub fn read_uvarint(bytes: &[u8]) -> Result<(u64, usize)> { + let mut result: u64 = 0; + let mut shift: u32 = 0; + for (i, &b) in bytes.iter().enumerate() { + anyhow::ensure!(shift < 64, "varint overflows u64"); + result |= u64::from(b & 0x7f) << shift; + if b & 0x80 == 0 { + return Ok((result, i + 1)); + } + shift += 7; + } + anyhow::bail!("varint is truncated (no terminating byte)") +} + +/// Is this cid a torrent of either shape? +pub fn is_torrent_cid(mscid: &MsCid) -> bool { + is_btih(mscid) || is_btih_file(mscid) +} + +/// A torrent cid as `(infohash, file_index)`; a whole-torrent cid is file 0. +/// `None` for anything else. +pub fn torrent_ref_of_cid(cid: &str) -> Option<([u8; 20], usize)> { + let mscid = parse_record_cid(cid).ok()?; + if let Ok(r) = decode_btih_file(&mscid) { + return Some((r.infohash20, r.file_index)); + } + infohash20_from_mscid(&mscid).ok().map(|ih| (ih, 0)) +} + +/// A torrent cid's v1 infohash (whole-torrent and per-file share it). +pub fn infohash_of_cid(cid: &str) -> Option<[u8; 20]> { + torrent_ref_of_cid(cid).map(|(ih, _)| ih) +} + +/// Lowercase hex. +pub fn to_hex(bytes: &[u8]) -> String { + use std::fmt::Write; + let mut s = String::with_capacity(bytes.len() * 2); + for b in bytes { + let _ = write!(s, "{b:02x}"); + } + s +} + +/// Lowercase hex of a v1 infohash — the magnet-cache and seed-unit key. +pub fn infohash20_hex(infohash20: &[u8; 20]) -> String { + to_hex(infohash20) +} + +/// Parse a 40-char hex infohash. +pub fn infohash20_from_hex(s: &str) -> Option<[u8; 20]> { + if s.len() != 40 { + return None; + } + let mut out = [0u8; 20]; + for (i, pair) in s.as_bytes().chunks(2).enumerate() { + let hi = (pair[0] as char).to_digit(16)?; + let lo = (pair[1] as char).to_digit(16)?; + out[i] = ((hi << 4) | lo) as u8; + } + Some(out) +} + +/// Best-effort MIME from a filename extension; `application/octet-stream` when +/// unknown so the browser sniffs. +pub fn content_type_from_name(name: &str) -> String { + let ext = name.rsplit('.').next().unwrap_or("").to_ascii_lowercase(); + let ct = match ext.as_str() { + "mkv" => "video/x-matroska", + "mp4" | "m4v" => "video/mp4", + "webm" => "video/webm", + "avi" => "video/x-msvideo", + "mov" => "video/quicktime", + "ts" | "mts" | "m2ts" => "video/mp2t", + "mpg" | "mpeg" => "video/mpeg", + "wmv" => "video/x-ms-wmv", + "flv" => "video/x-flv", + "ogv" => "video/ogg", + "mp3" => "audio/mpeg", + "flac" => "audio/flac", + "aac" => "audio/aac", + "ogg" | "oga" => "audio/ogg", + "wav" => "audio/wav", + "pdf" => "application/pdf", + _ => "application/octet-stream", + }; + ct.to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn base32_lower(input: &[u8]) -> String { + const ALPHABET: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567"; + let mut out = String::new(); + let mut buffer: u64 = 0; + let mut bits: u32 = 0; + for &b in input { + buffer = (buffer << 8) | b as u64; + bits += 8; + while bits >= 5 { + bits -= 5; + out.push(ALPHABET[((buffer >> bits) & 0x1F) as usize] as char); + } + } + if bits > 0 { + out.push(ALPHABET[((buffer << (5 - bits)) & 0x1F) as usize] as char); + } + out + } + + fn make_cid(codec: u8, mh_code: u8, digest: &[u8]) -> MsCid { + let mut wire = vec![0x01u8, codec, mh_code, digest.len() as u8]; + wire.extend_from_slice(digest); + MsCid::from_str(&format!("b{}", base32_lower(&wire))).expect("cid parse") + } + + fn write_uvarint(mut v: u64, out: &mut Vec) { + loop { + let mut byte = (v & 0x7f) as u8; + v >>= 7; + if v != 0 { + byte |= 0x80; + } + out.push(byte); + if v == 0 { + break; + } + } + } + + fn make_cid_varint(codec: u64, mh_code: u64, digest: &[u8]) -> MsCid { + let mut wire = vec![0x01u8]; + write_uvarint(codec, &mut wire); + write_uvarint(mh_code, &mut wire); + write_uvarint(digest.len() as u64, &mut wire); + wire.extend_from_slice(digest); + MsCid::from_str(&format!("b{}", base32_lower(&wire))).expect("cid parse") + } + + fn bt_file_digest(infohash: &[u8], index: u64) -> Vec { + let mut d = infohash.to_vec(); + write_uvarint(index, &mut d); + d + } + + fn v1_file_cid(infohash: &[u8; 20], index: u64) -> MsCid { + make_cid_varint(BTIH_V1_FILE_CODEC, BTIH_V1_FILE_CODEC, &bt_file_digest(infohash, index)) + } + + #[test] + fn btih_cid_is_recognised_and_decodes() { + let infohash = [7u8; 20]; + let cid = make_cid(0x55, MH_SHA1 as u8, &infohash); + assert!(is_btih(&cid)); + assert_eq!(infohash20_from_mscid(&cid).unwrap(), infohash); + } + + #[test] + fn ipfs_sha256_cid_is_not_btih() { + let cid = make_cid(0x55, 0x12, &[0u8; 32]); + assert!(!is_btih(&cid)); + assert!(infohash20_from_mscid(&cid).is_err()); + } + + #[test] + fn dagpb_cid_is_not_btih() { + assert!(!is_btih(&make_cid(0x70, 0x12, &[0u8; 32]))); + } + + #[test] + fn btih_v1_file_cid_is_recognised_and_decodes() { + let infohash = [7u8; 20]; + let cid = v1_file_cid(&infohash, 4); + assert!(is_btih_file(&cid)); + assert!(!is_btih(&cid)); + let r = decode_btih_file(&cid).unwrap(); + assert_eq!(r.infohash20, infohash); + assert_eq!(r.file_index, 4); + } + + #[test] + fn btih_v1_file_index_zero_and_multibyte_roundtrip() { + assert_eq!(decode_btih_file(&v1_file_cid(&[0xab; 20], 0)).unwrap().file_index, 0); + assert_eq!(decode_btih_file(&v1_file_cid(&[0x11; 20], 200)).unwrap().file_index, 200); + } + + #[test] + fn btih_v2_file_cid_recognised_but_rejected() { + let cid = make_cid_varint(BTIH_V2_FILE_CODEC, BTIH_V2_FILE_CODEC, &bt_file_digest(&[0x22; 32], 1)); + assert!(is_btih_file(&cid)); + assert!(decode_btih_file(&cid).is_err()); + } + + #[test] + fn whole_torrent_cid_is_index_zero_and_not_a_file_cid() { + let infohash = [9u8; 20]; + let cid = make_cid(0x55, MH_SHA1 as u8, &infohash); + assert!(is_btih(&cid)); + assert!(!is_btih_file(&cid)); + assert!(decode_btih_file(&cid).is_err()); + assert_eq!(torrent_ref_of_cid(&cid.to_string()), Some((infohash, 0))); + } + + #[test] + fn decode_btih_file_rejects_missing_index_varint() { + let cid = make_cid_varint(BTIH_V1_FILE_CODEC, BTIH_V1_FILE_CODEC, &[3u8; 20]); + assert!(decode_btih_file(&cid).is_err()); + } + + #[test] + fn read_uvarint_rejects_truncated_and_reads_multibyte() { + assert!(read_uvarint(&[0x80]).is_err()); + assert_eq!(read_uvarint(&[0xAC, 0x02]).unwrap(), (300, 2)); + assert_eq!(read_uvarint(&[0x05, 0xFF]).unwrap(), (5, 1)); + } + + #[test] + fn torrent_ref_of_a_file_cid_keeps_its_index() { + let cid = v1_file_cid(&[5u8; 20], 3).to_string(); + assert_eq!(torrent_ref_of_cid(&cid), Some(([5u8; 20], 3))); + assert_eq!(infohash_of_cid(&cid), Some([5u8; 20])); + assert_eq!(torrent_ref_of_cid("not a cid"), None); + } + + #[test] + fn hex_round_trips() { + let ih = [0xabu8; 20]; + assert_eq!(infohash20_from_hex(&infohash20_hex(&ih)), Some(ih)); + assert_eq!(infohash20_from_hex("zz"), None); + } + + #[test] + fn parse_rejects_token_form_and_cidv0() { + assert!(parse_record_cid("midhash256:bafy").is_err()); + assert!(parse_record_cid("QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG").is_err()); + } + + #[test] + fn content_type_guesses_common_media() { + assert_eq!(content_type_from_name("movie.mkv"), "video/x-matroska"); + assert_eq!(content_type_from_name("clip.MP4"), "video/mp4"); + assert_eq!(content_type_from_name("noext"), "application/octet-stream"); + } +} diff --git a/src/transport/client.rs b/src/transport/client.rs new file mode 100644 index 0000000..0d78b05 --- /dev/null +++ b/src/transport/client.rs @@ -0,0 +1,317 @@ +//! The hull side of the contract: an HTTP client for one transport plugin. +//! +//! [`RemoteTransport`] covers the common routes and gives typed helpers for a +//! plugin's extra ones (`get_json`, `post_json`, …). Byte responses come back as +//! an axum [`Response`] whose body streams straight from the plugin, so the hull +//! relays a multi-GB range without buffering it. + +use std::time::Duration; + +use axum::body::Body; +use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode}; +use axum::response::{IntoResponse, Response}; +use serde::de::DeserializeOwned; +use serde::Serialize; + +use super::dto::{ + Deleted, Health, Job, Jobs, Manifest, ReconcileReport, ReconcileRequest, +}; +use super::error::ApiError; +use super::focus::FocusSnapshot; + +/// Request headers never forwarded to a plugin (hop-by-hop, or describing the +/// client's connection to the hull rather than the hull's to the plugin). +const HOP_BY_HOP: &[&str] = &[ + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "transfer-encoding", + "upgrade", + "host", + "content-length", +]; + +/// Client for one plugin at `base` (e.g. `http://metashare-torrent:3000`). +#[derive(Clone, Debug)] +pub struct RemoteTransport { + id: String, + base: String, + http: reqwest::Client, + /// For control calls (jobs, focus, reconcile). `/raw` has no overall + /// timeout — a body can legitimately stream for the length of a film. + control_timeout: Duration, +} + +impl RemoteTransport { + /// `id` names the tier in error messages (`torrent`, `nzb`, `ipfs`). + pub fn new(id: impl Into, base: impl Into, http: reqwest::Client) -> Self { + Self { + id: id.into(), + base: base.into().trim_end_matches('/').to_string(), + http, + control_timeout: Duration::from_secs(30), + } + } + + pub fn with_control_timeout(mut self, t: Duration) -> Self { + self.control_timeout = t; + self + } + + pub fn id(&self) -> &str { + &self.id + } + + pub fn base(&self) -> &str { + &self.base + } + + pub fn url(&self, path: &str) -> String { + format!("{}{}", self.base, path) + } + + fn unreachable(&self, e: impl std::fmt::Display) -> ApiError { + ApiError::Upstream(format!("{} plugin unreachable: {e}", self.id)) + } + + // ---- common routes --------------------------------------------------- + + pub async fn manifest(&self) -> Result { + self.get_json("/manifest").await + } + + pub async fn health(&self) -> Result { + let r = self + .http + .get(self.url("/health")) + .timeout(self.control_timeout) + .send() + .await + .map_err(|e| self.unreachable(e))?; + r.json::().await.map_err(|e| self.unreachable(e)) + } + + /// `GET /raw/:cid`, forwarding the client's request headers plus `extra`. + /// Returns the plugin's response as-is (any status), with its body streaming. + pub async fn raw( + &self, + cid: &str, + client_headers: &HeaderMap, + extra: &[(&str, String)], + ) -> Result { + let mut req = self.http.get(self.url(&format!("/raw/{cid}"))); + req = req.headers(forwardable(client_headers)); + for (k, v) in extra { + req = req.header(*k, v); + } + let resp = req.send().await.map_err(|e| self.unreachable(e))?; + Ok(relay(resp)) + } + + pub async fn jobs(&self) -> Result, ApiError> { + Ok(self.get_json::("/jobs").await?.jobs) + } + + pub async fn delete(&self, cid: &str, keep: &[String]) -> Result { + let r = self + .http + .delete(self.url(&format!("/jobs/{cid}"))) + .query(&[("keep", keep.join(","))]) + .timeout(self.control_timeout) + .send() + .await + .map_err(|e| self.unreachable(e))?; + decode(r).await + } + + pub async fn reconcile(&self, req: &ReconcileRequest) -> Result { + // A reconcile can delete many units; give it more room than a poll. + let r = self + .http + .post(self.url("/reconcile")) + .json(req) + .timeout(self.control_timeout * 10) + .send() + .await + .map_err(|e| self.unreachable(e))?; + decode(r).await + } + + pub async fn push_focus(&self, snap: &FocusSnapshot) -> Result<(), ApiError> { + let r = self + .http + .put(self.url("/focus")) + .json(snap) + .timeout(self.control_timeout) + .send() + .await + .map_err(|e| self.unreachable(e))?; + if r.status().is_success() { + Ok(()) + } else { + Err(ApiError::Upstream(format!("{} plugin /focus: {}", self.id, r.status()))) + } + } + + // ---- typed helpers for extra routes ------------------------------------ + + pub async fn get_json(&self, path: &str) -> Result { + let r = self + .http + .get(self.url(path)) + .timeout(self.control_timeout) + .send() + .await + .map_err(|e| self.unreachable(e))?; + decode(r).await + } + + pub async fn post_json( + &self, + path: &str, + body: &B, + ) -> Result { + self.post_json_timeout(path, body, self.control_timeout).await + } + + pub async fn post_json_timeout( + &self, + path: &str, + body: &B, + timeout: Duration, + ) -> Result { + let r = self + .http + .post(self.url(path)) + .json(body) + .timeout(timeout) + .send() + .await + .map_err(|e| self.unreachable(e))?; + decode(r).await + } + + pub async fn put_json(&self, path: &str, body: &B) -> Result<(), ApiError> { + let r = self + .http + .put(self.url(path)) + .json(body) + .timeout(self.control_timeout) + .send() + .await + .map_err(|e| self.unreachable(e))?; + if r.status().is_success() { + Ok(()) + } else { + Err(error_from(r).await) + } + } + + /// Forward an arbitrary request and relay the response verbatim — for the + /// public routes a plugin now answers (`/ipfs/:cid`, `/api/peers`, …). + pub async fn forward( + &self, + method: Method, + path_and_query: &str, + client_headers: &HeaderMap, + body: Option, + ) -> Result { + let mut req = self + .http + .request(method, self.url(path_and_query)) + .headers(forwardable(client_headers)); + if let Some(b) = body { + req = req.body(b); + } + let resp = req.send().await.map_err(|e| self.unreachable(e))?; + Ok(relay(resp)) + } +} + +/// Client request headers worth forwarding to a plugin. +fn forwardable(h: &HeaderMap) -> reqwest::header::HeaderMap { + let mut out = reqwest::header::HeaderMap::new(); + for (k, v) in h { + if HOP_BY_HOP.contains(&k.as_str()) { + continue; + } + if let (Ok(k), Ok(v)) = ( + reqwest::header::HeaderName::from_bytes(k.as_str().as_bytes()), + reqwest::header::HeaderValue::from_bytes(v.as_bytes()), + ) { + out.append(k, v); + } + } + out +} + +/// A plugin response as an axum response: same status, same headers (minus +/// hop-by-hop), body streamed. +pub fn relay(resp: reqwest::Response) -> Response { + let status = StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY); + let mut headers = HeaderMap::new(); + for (k, v) in resp.headers() { + let name = k.as_str(); + if matches!(name, "connection" | "keep-alive" | "transfer-encoding" | "trailer" | "upgrade") { + continue; + } + if let (Ok(k), Ok(v)) = ( + HeaderName::from_bytes(name.as_bytes()), + HeaderValue::from_bytes(v.as_bytes()), + ) { + headers.append(k, v); + } + } + let body = Body::from_stream(resp.bytes_stream()); + let mut out = (status, body).into_response(); + *out.headers_mut() = headers; + out +} + +async fn decode(r: reqwest::Response) -> Result { + if !r.status().is_success() { + return Err(error_from(r).await); + } + r.json::() + .await + .map_err(|e| ApiError::Upstream(format!("plugin returned malformed JSON: {e}"))) +} + +/// Map a plugin error response back onto [`ApiError`], keeping its message so +/// a relayed error reads the same as the in-process one did. +pub async fn error_from(r: reqwest::Response) -> ApiError { + let status = r.status().as_u16(); + let msg = r + .json::() + .await + .ok() + .and_then(|v| v.get("error").and_then(|e| e.as_str()).map(str::to_string)) + .unwrap_or_else(|| format!("plugin error {status}")); + match status { + 400 => ApiError::BadRequest(msg), + 404 => ApiError::NotFound, + 504 => ApiError::GatewayTimeout(msg), + 503 if msg == "swarm task is gone" => ApiError::SwarmGone, + _ => ApiError::Upstream(msg), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hop_by_hop_headers_are_not_forwarded() { + let mut h = HeaderMap::new(); + h.insert("range", HeaderValue::from_static("bytes=0-1")); + h.insert("host", HeaderValue::from_static("x")); + h.insert("connection", HeaderValue::from_static("close")); + let f = forwardable(&h); + assert!(f.contains_key("range")); + assert!(!f.contains_key("host")); + assert!(!f.contains_key("connection")); + } +} diff --git a/src/transport/dto.rs b/src/transport/dto.rs new file mode 100644 index 0000000..b9d0f67 --- /dev/null +++ b/src/transport/dto.rs @@ -0,0 +1,274 @@ +//! Wire types of the transport-plugin contract (hull ⇄ plugin, JSON over HTTP). +//! +//! Only what is common to every plugin lives here. A plugin's own extra routes +//! (ipfs `/share`, torrent `/commits`, …) carry their DTOs in the per-protocol +//! modules next to this one, so the hull and the plugin still compile against +//! one definition. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// Bumped on any breaking change to the common routes below. The hull refuses +/// a plugin whose `/manifest` reports a different major contract. +pub const CONTRACT_VERSION: u32 = 1; + +/// `X-MetaMesh-Lane`: the hull's lane decision for a `/raw` request. +pub const HDR_LANE: &str = "x-metamesh-lane"; +/// `X-MetaMesh-Player`: forwarded verbatim from the client — marks a genuine +/// play (commits, provisional focus). +pub const HDR_PLAYER: &str = "x-metamesh-player"; +/// `X-MetaMesh-Magnet`: the magnet the hull resolved for a torrent cid. +pub const HDR_MAGNET: &str = "x-metamesh-magnet"; +/// Seed facts a plugin attaches to a successful `/raw` response, so the hull +/// records the seed row *before* it relays the body (the monolith wrote the +/// row inside the fetch, ahead of the first byte). Internal: the hull strips +/// every `x-metamesh-seed-*` header before relaying. +pub const HDR_SEED_NAME: &str = "x-metamesh-seed-name"; +pub const HDR_SEED_SIZE: &str = "x-metamesh-seed-size"; +/// Prefix of the seed headers the hull strips from a relayed response. +pub const INTERNAL_HEADER_PREFIX: &str = "x-metamesh-seed-"; +/// `x-metamesh-meter: egress` on a `/raw` response: the hull floors this body +/// against the playback focus when it relays it. Set exactly where the +/// monolith metered (`focus::gated_stream`) — torrent ranges, Usenet file +/// reads, IPFS dag ranges — and *not* on single-block IPFS answers (posters), +/// which the monolith never throttled. Internal; stripped by the hull. +pub const HDR_METER: &str = "x-metamesh-meter"; +pub const METER_EGRESS: &str = "egress"; + +/// Encode free text (a file name) for an internal header value: every byte +/// outside a conservative visible-ASCII set becomes `%XX`. Inverse: +/// [`header_text_decode`]. +pub fn header_text_encode(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + for b in s.bytes() { + if b.is_ascii_alphanumeric() || b" ._-()[]+,".contains(&b) { + out.push(b as char); + } else { + out.push_str(&format!("%{b:02X}")); + } + } + out +} + +/// Inverse of [`header_text_encode`]. Invalid escapes pass through literally. +pub fn header_text_decode(s: &str) -> String { + let bytes = s.as_bytes(); + let mut out = Vec::with_capacity(bytes.len()); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + let hi = (bytes[i + 1] as char).to_digit(16); + let lo = (bytes[i + 2] as char).to_digit(16); + if let (Some(hi), Some(lo)) = (hi, lo) { + out.push((hi * 16 + lo) as u8); + i += 3; + continue; + } + } + out.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&out).into_owned() +} + +/// `GET /manifest`. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct Manifest { + /// Protocol id: `torrent` | `nzb` | `ipfs` (a competing implementation of a + /// protocol reports the same id — that's what makes it a drop-in). + pub id: String, + /// Implementation name + version, for the dashboard. + pub implementation: String, + pub version: String, + pub contract: u32, + pub capabilities: Capabilities, +} + +#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Capabilities { + /// Can fetch bytes for a cid (`/raw`). + pub fetch: bool, + /// Serves what it fetched to other peers (seeds). + pub share: bool, +} + +/// `GET /health`. +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Health { + pub ok: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +impl Health { + pub fn ok() -> Self { + Self { ok: true, detail: None } + } +} + +/// One unit of bytes a plugin holds — the eviction unit as the plugin knows it +/// (a torrent, an NZB job). `GET /jobs`. +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq)] +pub struct Job { + /// Stable unit key (a torrent's infohash hex, an nzb job's cid). + pub id: String, + /// The cids this unit answers for, as far as the plugin can name them. + #[serde(default)] + pub cids: Vec, + /// Bytes on disk right now. + pub bytes_on_disk: u64, + /// Every selected byte is present. + pub complete: bool, + /// Downloading in the background on its own (exempt from eviction). + pub filling: bool, + /// Plugin-defined state label (`live`, `paused`, `streaming`, …). + #[serde(default)] + pub state: String, + /// Plugin-specific detail the hull renders opaquely (live transfer stats). + #[serde(default, skip_serializing_if = "Value::is_null")] + pub extra: Value, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default)] +pub struct Jobs { + pub jobs: Vec, +} + +/// `DELETE /jobs/:cid` query: the *other* cids of the same unit the hull still +/// wants (sibling files of one torrent). The plugin keeps those. +#[derive(Serialize, Deserialize, Clone, Debug, Default)] +pub struct DeleteQuery { + /// Comma-separated cids. + #[serde(default)] + pub keep: String, +} + +impl DeleteQuery { + pub fn keep_list(&self) -> Vec { + self.keep + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .collect() + } +} + +/// `DELETE /jobs/:cid` → the bytes actually unlinked, which is what the hull's +/// eviction budget credits. +#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Deleted { + pub bytes_freed: u64, +} + +/// `POST /reconcile`: delete every unit no wanted cid names, and any on-disk +/// leftover no unit owns older than `grace_secs`. +#[derive(Serialize, Deserialize, Clone, Debug, Default)] +pub struct ReconcileRequest { + pub wanted: Vec, + pub grace_secs: u64, + /// Only meaningful for plugins whose state lives in a lazily-started + /// session (torrent): bring it up even if nothing asked for bytes yet, + /// because persisted units are on disk and only the session can say whose. + #[serde(default)] + pub start_session: bool, +} + +#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ReconcileReport { + /// Units deleted because nothing wanted them. + pub units_deleted: u64, + pub unit_bytes_freed: u64, + /// On-disk leftovers removed. + pub orphans_deleted: u64, + pub orphan_bytes_freed: u64, + /// The plugin couldn't tell what it owns (session down): nothing deleted. + #[serde(default)] + pub skipped: bool, +} + +/// A file inside a promoted cache container. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct PromotedFile { + /// Path relative to the container directory. + pub rel: String, + /// Absolute path on the shared `/data` volume. + pub path: String, + pub size: u64, +} + +/// Plugin → hull notifications, `POST {hull}/internal/events` (see +/// [`super::hull::HullClient`]). The hull acknowledges once it has acted, so a +/// plugin that must not proceed before the hull indexed something can wait. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Event { + /// The plugin now holds bytes under `cid`: the hull upserts a seed row. + Seed { + cid: String, + /// `torrent` | `ipfs` (the hull's `SeedKind`). + kind: String, + #[serde(default)] + name: String, + #[serde(default)] + size_bytes: u64, + /// The container the bytes live in, when there is one. + #[serde(default, skip_serializing_if = "Option::is_none")] + container: Option, + }, + /// A completed fetch was published as a cache container (`rename(2)` into + /// `cache/`). The hull indexes it in the material index. + Promoted { + container: String, + cids: Vec, + files: Vec, + }, + /// The content behind `cid` turned out to be an archive: the hull marks + /// the meta-core record `fileType=archive`. + Archive { cid: String }, + /// Bytes behind these cids are gone: the hull drops their seed rows. + Vanished { cids: Vec }, + /// What the plugin knows about `cid` changed (a new availability verdict): + /// the hull drops its cached probe grade. + ProbeStale { cid: String }, + /// A fetch finished and its file is complete at `path` (still under + /// `tmp/`). The hull indexes it for local-first reads and decides what to + /// do next (the Usenet → IPFS re-seed). + Ready { + cid: String, + path: String, + size: u64, + }, + /// Any event type this hull doesn't know yet. Ignored. + #[serde(other)] + Unknown, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn header_text_round_trips_unicode_and_reserved() { + for s in ["Plain Name.mkv", "Été 100% \"quoted\"/x.mkv", "", "%", "a%2"] { + assert_eq!(header_text_decode(&header_text_encode(s)), s); + } + assert!(header_text_encode("é/").is_ascii()); + } + + #[test] + fn delete_query_splits_and_trims() { + let q = DeleteQuery { keep: " a, ,b,".into() }; + assert_eq!(q.keep_list(), vec!["a".to_string(), "b".to_string()]); + assert!(DeleteQuery::default().keep_list().is_empty()); + } + + #[test] + fn events_are_tagged_and_unknown_tolerated() { + let e = Event::Archive { cid: "b".into() }; + let j = serde_json::to_string(&e).unwrap(); + assert_eq!(j, r#"{"type":"archive","cid":"b"}"#); + let u: Event = serde_json::from_str(r#"{"type":"from_the_future","x":1}"#).unwrap(); + assert_eq!(u, Event::Unknown); + } +} diff --git a/src/transport/error.rs b/src/transport/error.rs new file mode 100644 index 0000000..1d556dc --- /dev/null +++ b/src/transport/error.rs @@ -0,0 +1,60 @@ +//! The one error type every transport HTTP surface answers with. +//! +//! Moved verbatim from meta-share `crates/meta-share/src/api/errors.rs`. It lives +//! here, not in the hull, because a plugin's error body is relayed to meta-share's +//! public API unchanged: the JSON shape `{ "error": "" }` and the +//! status per variant are the contract every dashboard and peer depends on, and +//! the hull and its plugins must not be able to drift on it. + +use axum::{http::StatusCode, response::IntoResponse, Json}; + +#[derive(Debug)] +pub enum ApiError { + BadRequest(String), + NotFound, + SwarmGone, + /// This peer is consumer-only (no `META_CORE_URL`) so it can't satisfy a + /// raw-bytes request directly. The cross-peer proxy turns this into a + /// clearer 502 with hint text; the direct route returns 503. + ConsumerOnly, + Upstream(String), + /// Gateway-side compute exceeded its deadline. Distinct from `Upstream` so + /// the dashboard can surface a clearer "the source fetcher is slow" message + /// instead of a generic 502. + GatewayTimeout(String), +} + +impl IntoResponse for ApiError { + fn into_response(self) -> axum::response::Response { + let (status, msg) = match self { + ApiError::BadRequest(m) => (StatusCode::BAD_REQUEST, m), + ApiError::NotFound => (StatusCode::NOT_FOUND, "not found".into()), + ApiError::SwarmGone => ( + StatusCode::SERVICE_UNAVAILABLE, + "swarm task is gone".into(), + ), + ApiError::ConsumerOnly => ( + StatusCode::SERVICE_UNAVAILABLE, + "this peer has no META_CORE_URL; raw bytes are only served by the source peer".into(), + ), + ApiError::Upstream(m) => (StatusCode::BAD_GATEWAY, m), + ApiError::GatewayTimeout(m) => (StatusCode::GATEWAY_TIMEOUT, m), + }; + (status, Json(serde_json::json!({ "error": msg }))).into_response() + } +} + +impl std::fmt::Display for ApiError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ApiError::BadRequest(m) => write!(f, "bad request: {m}"), + ApiError::NotFound => write!(f, "not found"), + ApiError::SwarmGone => write!(f, "swarm task is gone"), + ApiError::ConsumerOnly => write!(f, "consumer-only peer"), + ApiError::Upstream(m) => write!(f, "upstream: {m}"), + ApiError::GatewayTimeout(m) => write!(f, "timeout: {m}"), + } + } +} + +impl std::error::Error for ApiError {} diff --git a/src/transport/file.rs b/src/transport/file.rs new file mode 100644 index 0000000..0766111 --- /dev/null +++ b/src/transport/file.rs @@ -0,0 +1,114 @@ +//! Serving a byte range straight off a file on the shared `/data` volume. +//! +//! The hull's local-first read and the usenet plugin's materialised/partial +//! file reads answer with the same headers, so they share this. Moved from +//! meta-share's `nzb::serve::serve_file_range`, minus the focus metering: the +//! caller meters the body (the hull wraps every response it serves, a plugin +//! doesn't meter its `/raw` at all — the hull does on relay). + +use std::path::Path; + +use axum::body::Body; +use axum::http::{header, HeaderMap, HeaderValue, StatusCode}; +use axum::response::{IntoResponse, Response}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; +use tokio_util::io::ReaderStream; + +use super::range::{parse_range_header, RangeOutcome}; + +/// Stream a complete on-disk file of `total` bytes, honouring a single +/// `Range:` (206) else 200; 416 for an unsatisfiable range. +/// +/// `Err` carries the IO error that stopped the open/seek, so a caller can tell +/// "the file moved under us" apart from a served response. +pub async fn serve_file_range( + path: &Path, + total: u64, + range: Option<&HeaderValue>, +) -> std::io::Result { + let ctype = content_type(path); + match parse_range_header(range, total) { + RangeOutcome::Unsatisfiable => { + let mut headers = HeaderMap::new(); + headers.insert( + header::CONTENT_RANGE, + HeaderValue::from_str(&format!("bytes */{total}")).unwrap(), + ); + Ok((StatusCode::RANGE_NOT_SATISFIABLE, headers).into_response()) + } + RangeOutcome::Ok { start, end } => { + let len = end - start + 1; + let mut file = tokio::fs::File::open(path).await?; + file.seek(std::io::SeekFrom::Start(start)).await?; + let body = Body::from_stream(ReaderStream::new(file.take(len))); + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_TYPE, HeaderValue::from_static(ctype)); + headers.insert(header::ACCEPT_RANGES, HeaderValue::from_static("bytes")); + headers.insert(header::CONTENT_LENGTH, HeaderValue::from(len)); + headers.insert( + header::CONTENT_RANGE, + HeaderValue::from_str(&format!("bytes {start}-{end}/{total}")).unwrap(), + ); + Ok((StatusCode::PARTIAL_CONTENT, headers, body).into_response()) + } + // Absent / Unsupported → full 200. + _ => { + let file = tokio::fs::File::open(path).await?; + let body = Body::from_stream(ReaderStream::new(file)); + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_TYPE, HeaderValue::from_static(ctype)); + headers.insert(header::ACCEPT_RANGES, HeaderValue::from_static("bytes")); + headers.insert(header::CONTENT_LENGTH, HeaderValue::from(total)); + Ok((StatusCode::OK, headers, body).into_response()) + } + } +} + +/// Coarse content-type from extension — enough for the player to pick a demuxer. +pub fn content_type(path: &Path) -> &'static str { + match path + .extension() + .and_then(|e| e.to_str()) + .map(|e| e.to_ascii_lowercase()) + .as_deref() + { + Some("mkv") => "video/x-matroska", + Some("mp4") | Some("m4v") => "video/mp4", + Some("webm") => "video/webm", + Some("avi") => "video/x-msvideo", + Some("mov") => "video/quicktime", + Some("ts") | Some("m2ts") => "video/mp2t", + _ => "application/octet-stream", + } +} + +#[cfg(test)] +mod tests { + use super::*; + + async fn body(r: Response) -> Vec { + axum::body::to_bytes(r.into_body(), 1 << 20).await.unwrap().to_vec() + } + + #[tokio::test] + async fn range_full_and_unsatisfiable() { + let p = std::env::temp_dir().join(format!("file-range-{}.mkv", std::process::id())); + std::fs::write(&p, b"0123456789").unwrap(); + let r = serve_file_range(&p, 10, Some(&HeaderValue::from_static("bytes=2-4"))).await.unwrap(); + assert_eq!(r.status(), StatusCode::PARTIAL_CONTENT); + assert_eq!(r.headers()[header::CONTENT_RANGE], "bytes 2-4/10"); + assert_eq!(r.headers()[header::CONTENT_TYPE], "video/x-matroska"); + assert_eq!(body(r).await, b"234"); + let r = serve_file_range(&p, 10, None).await.unwrap(); + assert_eq!(r.status(), StatusCode::OK); + assert_eq!(body(r).await, b"0123456789"); + let r = serve_file_range(&p, 10, Some(&HeaderValue::from_static("bytes=20-"))).await.unwrap(); + assert_eq!(r.status(), StatusCode::RANGE_NOT_SATISFIABLE); + assert_eq!(r.headers()[header::CONTENT_RANGE], "bytes */10"); + let _ = std::fs::remove_file(&p); + assert_eq!( + serve_file_range(&p, 10, None).await.unwrap_err().kind(), + std::io::ErrorKind::NotFound + ); + } +} diff --git a/src/transport/focus.rs b/src/transport/focus.rs new file mode 100644 index 0000000..dbf94f7 --- /dev/null +++ b/src/transport/focus.rs @@ -0,0 +1,464 @@ +//! Playback focus, as a transport plugin sees it. +//! +//! meta-share's hull owns the focus *policy*: `POST/DELETE /api/playback/:cid`, +//! the leases, the TTL reaper, which sibling CIDs make up a title. A plugin only +//! needs to *obey* it: pause a torrent that isn't focused, park a background +//! fill, floor a non-focused ingress. So the hull pushes a [`FocusSnapshot`] +//! (`PUT /focus`) on every transition, plus periodically so a restarted plugin +//! recovers, and the plugin answers its gates from a local [`FocusView`]. +//! +//! The read side mirrors meta-share's `focus::FocusGate` method for method, so +//! code moved out of meta-share keeps its call sites. Two things are +//! deliberately different from the in-process gate: +//! +//! - **No leases.** A view never expires an entry on its own; only the next +//! snapshot changes it. The hull's reaper is the single source of truth. +//! - **Per-process buckets.** `throttle_in`/`throttle_out` floor non-focused +//! traffic at `bg_rate_bytes` *per process*, where the monolith shared one +//! bucket across tiers. HTTP response bodies are still metered once, by the +//! hull, when it relays a plugin's `/raw` — plugins must not meter their own +//! `/raw` egress, or a peer pulling through the hull pays twice. + +use std::collections::HashSet; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use serde::{Deserialize, Serialize}; +use tokio::sync::watch; + +/// How often a parked background task re-checks whether the focus cleared. +/// Same value as meta-share's `focus::PARK_POLL`. +pub const PARK_POLL: Duration = Duration::from_millis(750); + +/// Burst allowance as a multiple of the per-second rate, floored at +/// [`MIN_BURST_BYTES`]. Same values as meta-share's `focus.rs`. +const BURST_SECS: u64 = 2; +const MIN_BURST_BYTES: u64 = 512 * 1024; + +/// Which bandwidth lane a byte flow belongs to. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Lane { + /// Belongs to a title a viewer is playing right now — never gated. + Focused, + /// Everything else. Parked or throttled while any title is focused; + /// unrestricted otherwise. + Background, +} + +impl Lane { + pub fn is_focused(self) -> bool { + matches!(self, Lane::Focused) + } + + /// Wire value of the `X-MetaMesh-Lane` header. + pub fn as_header(self) -> &'static str { + match self { + Lane::Focused => "focused", + Lane::Background => "background", + } + } + + /// Parse the `X-MetaMesh-Lane` header. An absent or unknown value is + /// `Focused`: a request the hull didn't classify must never be starved. + pub fn from_header(v: Option<&str>) -> Lane { + match v { + Some("background") => Lane::Background, + _ => Lane::Focused, + } + } +} + +/// One focused title: its group key plus every identity it is reachable under. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct FocusTitle { + /// The record's canonical cid — the stable per-title key. + pub group: String, + /// Every sibling cid of the title. + #[serde(default)] + pub cids: Vec, + /// The torrent infohashes among those siblings, lowercase hex, pre-decoded + /// by the hull so a torrent plugin never re-parses cids on a transition. + #[serde(default)] + pub infohashes: Vec, +} + +/// The whole focus state, pushed by the hull on `PUT /focus`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct FocusSnapshot { + /// Master switch (`META_SHARE_FOCUS_ENABLED`). Off → every gate is a no-op. + pub enabled: bool, + /// Bytes/s a non-focused flow is floored at. `0` → hard park. + pub bg_rate_bytes: u64, + #[serde(default)] + pub titles: Vec, +} + +impl Default for FocusSnapshot { + fn default() -> Self { + Self { + enabled: true, + bg_rate_bytes: 10 * 1024, + titles: Vec::new(), + } + } +} + +struct Inner { + titles: Vec, + cids: HashSet, + infohashes: HashSet<[u8; 20]>, +} + +/// A plugin's local replica of the hull's focus gate. +pub struct FocusView { + inner: Mutex, + /// `titles.len()` as an atomic, so the resting state is one relaxed load. + n: AtomicUsize, + enabled: std::sync::atomic::AtomicBool, + rate: AtomicU64, + out_bucket: TokenBucket, + in_bucket: TokenBucket, + /// Bumped on every applied snapshot that changed something. A torrent + /// plugin watches it to re-run its pause/unpause pass. + changed: watch::Sender, +} + +impl FocusView { + pub fn new() -> Arc { + let d = FocusSnapshot::default(); + let (changed, _) = watch::channel(0); + let burst = (d.bg_rate_bytes * BURST_SECS).max(MIN_BURST_BYTES); + Arc::new(Self { + inner: Mutex::new(Inner { + titles: Vec::new(), + cids: HashSet::new(), + infohashes: HashSet::new(), + }), + n: AtomicUsize::new(0), + enabled: std::sync::atomic::AtomicBool::new(d.enabled), + rate: AtomicU64::new(d.bg_rate_bytes), + out_bucket: TokenBucket::new(d.bg_rate_bytes, burst), + in_bucket: TokenBucket::new(d.bg_rate_bytes, burst), + changed, + }) + } + + /// Replace the state with `snap`. Returns `true` when anything changed (and + /// notifies [`subscribe`](Self::subscribe)rs). + pub fn apply(&self, snap: FocusSnapshot) -> bool { + let mut cids = HashSet::new(); + let mut infohashes = HashSet::new(); + for t in &snap.titles { + cids.extend(t.cids.iter().cloned()); + cids.insert(t.group.clone()); + infohashes.extend(t.infohashes.iter().filter_map(|h| decode_hex20(h))); + } + let changed = { + let mut inner = self.inner.lock().unwrap(); + let changed = inner.titles != snap.titles + || self.enabled.load(Ordering::Relaxed) != snap.enabled + || self.rate.load(Ordering::Relaxed) != snap.bg_rate_bytes; + inner.titles = snap.titles; + inner.cids = cids; + inner.infohashes = infohashes; + self.n.store(inner.titles.len(), Ordering::Relaxed); + changed + }; + self.enabled.store(snap.enabled, Ordering::Relaxed); + if self.rate.swap(snap.bg_rate_bytes, Ordering::Relaxed) != snap.bg_rate_bytes { + let burst = (snap.bg_rate_bytes * BURST_SECS).max(MIN_BURST_BYTES); + self.out_bucket.reconfigure(snap.bg_rate_bytes, burst); + self.in_bucket.reconfigure(snap.bg_rate_bytes, burst); + } + if changed { + self.changed.send_modify(|g| *g += 1); + } + changed + } + + /// The current state, as the hull last pushed it. + pub fn snapshot(&self) -> FocusSnapshot { + FocusSnapshot { + enabled: self.enabled.load(Ordering::Relaxed), + bg_rate_bytes: self.rate.load(Ordering::Relaxed), + titles: self.inner.lock().unwrap().titles.clone(), + } + } + + /// Fires after every snapshot that changed something. + pub fn subscribe(&self) -> watch::Receiver { + self.changed.subscribe() + } + + pub fn bg_rate_bytes(&self) -> u64 { + self.rate.load(Ordering::Relaxed) + } + + /// Is *any* title focused? + pub fn any(&self) -> bool { + self.enabled.load(Ordering::Relaxed) && self.n.load(Ordering::Relaxed) > 0 + } + + pub fn count(&self) -> usize { + self.n.load(Ordering::Relaxed) + } + + /// Which lane does a byte flow for `cid` belong to? Same rule as the hull: + /// nothing focused → everything is `Focused`. + pub fn lane(&self, cid: &str) -> Lane { + if !self.any() { + return Lane::Focused; + } + if self.inner.lock().unwrap().cids.contains(cid) { + Lane::Focused + } else { + Lane::Background + } + } + + pub fn is_cid_focused(&self, cid: &str) -> bool { + self.inner.lock().unwrap().cids.contains(cid) + } + + pub fn focused_infohashes(&self) -> HashSet<[u8; 20]> { + self.inner.lock().unwrap().infohashes.clone() + } + + pub fn is_infohash_focused(&self, infohash: &[u8; 20]) -> bool { + self.inner.lock().unwrap().infohashes.contains(infohash) + } + + /// Park a background task while any title is focused. + pub async fn wait_turn(&self, lane: Lane) { + if lane.is_focused() { + return; + } + while self.any() { + tokio::time::sleep(PARK_POLL).await; + } + } + + /// [`wait_turn`](Self::wait_turn) with a deadline; `false` if the focus + /// outlasted `budget`. + pub async fn wait_turn_timeout(&self, lane: Lane, budget: Duration) -> bool { + if lane.is_focused() { + return true; + } + let deadline = Instant::now() + budget; + while self.any() { + if Instant::now() >= deadline { + return false; + } + tokio::time::sleep(PARK_POLL).await; + } + true + } + + /// Floor a non-focused egress of `n` bytes (bitswap block service). + pub async fn throttle_out(&self, lane: Lane, n: u64) { + if lane.is_focused() || !self.any() { + return; + } + if self.bg_rate_bytes() == 0 { + self.wait_turn(lane).await; + return; + } + self.out_bucket.consume(n).await; + } + + /// Floor a non-focused ingress of `n` bytes that a caller is waiting on. + pub async fn throttle_in(&self, lane: Lane, n: u64) { + if lane.is_focused() || !self.any() { + return; + } + if self.bg_rate_bytes() == 0 { + self.wait_turn(lane).await; + return; + } + self.in_bucket.consume(n).await; + } +} + +/// Meter a byte stream against a view's egress floor. Same shape as +/// meta-share's `focus::gated_stream`. +pub fn gated_stream( + stream: S, + focus: Arc, + lane: Lane, +) -> impl futures::Stream> +where + S: futures::Stream>, + B: AsRef<[u8]>, +{ + use futures::StreamExt; + stream.then(move |item| { + let focus = Arc::clone(&focus); + async move { + if let Ok(bytes) = &item { + focus.throttle_out(lane, bytes.as_ref().len() as u64).await; + } + item + } + }) +} + +/// Lowercase hex of a 20-byte infohash — the wire form in [`FocusTitle`]. +pub fn infohash_hex(ih: &[u8; 20]) -> String { + let mut s = String::with_capacity(40); + for b in ih { + s.push_str(&format!("{b:02x}")); + } + s +} + +fn decode_hex20(s: &str) -> Option<[u8; 20]> { + if s.len() != 40 { + return None; + } + let mut out = [0u8; 20]; + for (i, pair) in s.as_bytes().chunks(2).enumerate() { + let hi = (pair[0] as char).to_digit(16)?; + let lo = (pair[1] as char).to_digit(16)?; + out[i] = ((hi << 4) | lo) as u8; + } + Some(out) +} + +/// Leaky bucket that always waits (never errors on `n > burst`). Copied from +/// meta-share's `focus.rs`, plus `reconfigure` for a rate pushed at runtime. +struct TokenBucket { + state: Mutex, +} + +struct BucketState { + tokens: f64, + last: Instant, + rate: u64, + burst: u64, +} + +impl TokenBucket { + fn new(rate: u64, burst: u64) -> Self { + Self { + state: Mutex::new(BucketState { + tokens: burst as f64, + last: Instant::now(), + rate, + burst, + }), + } + } + + fn reconfigure(&self, rate: u64, burst: u64) { + let mut s = self.state.lock().unwrap(); + s.rate = rate; + s.burst = burst; + s.tokens = s.tokens.min(burst as f64); + } + + async fn consume(&self, n: u64) { + let wait = { + let mut s = self.state.lock().unwrap(); + if s.rate == 0 || n == 0 { + return; + } + let now = Instant::now(); + let elapsed = now.duration_since(s.last).as_secs_f64(); + s.last = now; + s.tokens = (s.tokens + elapsed * s.rate as f64).min(s.burst as f64); + s.tokens -= n as f64; + if s.tokens >= 0.0 { + return; + } + Duration::from_secs_f64(-s.tokens / s.rate as f64) + }; + tokio::time::sleep(wait).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn snap(titles: Vec) -> FocusSnapshot { + FocusSnapshot { + titles, + ..Default::default() + } + } + + #[test] + fn nothing_focused_means_every_flow_is_focused() { + let v = FocusView::new(); + assert!(!v.any()); + assert_eq!(v.lane("bafyanything"), Lane::Focused); + } + + #[test] + fn siblings_and_group_are_focused_others_background() { + let v = FocusView::new(); + v.apply(snap(vec![FocusTitle { + group: "g".into(), + cids: vec!["a".into(), "b".into()], + infohashes: vec![infohash_hex(&[7u8; 20])], + }])); + assert!(v.any()); + assert_eq!(v.lane("a"), Lane::Focused); + assert_eq!(v.lane("g"), Lane::Focused); + assert_eq!(v.lane("z"), Lane::Background); + assert!(v.is_infohash_focused(&[7u8; 20])); + assert!(!v.is_infohash_focused(&[8u8; 20])); + } + + #[test] + fn disabled_gate_focuses_nothing() { + let v = FocusView::new(); + v.apply(FocusSnapshot { + enabled: false, + bg_rate_bytes: 1, + titles: vec![FocusTitle { + group: "g".into(), + ..Default::default() + }], + }); + assert!(!v.any()); + assert_eq!(v.lane("z"), Lane::Focused); + } + + #[test] + fn apply_reports_change_only_on_change() { + let v = FocusView::new(); + let rx = v.subscribe(); + let s = snap(vec![FocusTitle { + group: "g".into(), + ..Default::default() + }]); + assert!(v.apply(s.clone())); + assert!(!v.apply(s)); + assert_eq!(*rx.borrow(), 1); + assert!(v.apply(snap(vec![]))); + assert_eq!(*rx.borrow(), 2); + } + + #[test] + fn lane_header_round_trip_defaults_to_focused() { + assert_eq!(Lane::from_header(Some("background")), Lane::Background); + assert_eq!(Lane::from_header(Some("focused")), Lane::Focused); + assert_eq!(Lane::from_header(None), Lane::Focused); + assert_eq!(Lane::from_header(Some("junk")), Lane::Focused); + assert_eq!(Lane::from_header(Some(Lane::Background.as_header())), Lane::Background); + } + + #[test] + fn snapshot_json_round_trips() { + let s = snap(vec![FocusTitle { + group: "g".into(), + cids: vec!["a".into()], + infohashes: vec![infohash_hex(&[1u8; 20])], + }]); + let back: FocusSnapshot = serde_json::from_str(&serde_json::to_string(&s).unwrap()).unwrap(); + assert_eq!(back, s); + } +} diff --git a/src/transport/fsutil.rs b/src/transport/fsutil.rs new file mode 100644 index 0000000..0af457c --- /dev/null +++ b/src/transport/fsutil.rs @@ -0,0 +1,230 @@ +//! Filesystem helpers shared by the hull and the plugins that own a corner of +//! the shared `/data` volume: free space, on-disk size, and the grace-period +//! removal both `reclaim` passes (cache containers in the hull, `bt-temp/` in +//! the torrent plugin) are built on. Moved verbatim from meta-share +//! `reclaim.rs` / `eviction.rs`. + +use std::ffi::OsString; +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime}; + +use tracing::warn; + +/// Free space (bytes available to an unprivileged process) on the filesystem +/// backing `path`; `None` when the query fails or the platform isn't supported. +#[cfg(unix)] +pub fn available_bytes(path: &Path) -> Option { + use std::os::unix::ffi::OsStrExt; + let c_path = std::ffi::CString::new(path.as_os_str().as_bytes()).ok()?; + // SAFETY: `c_path` is a valid NUL-terminated path; `statvfs` only writes + // into the zeroed-out struct and returns 0 on success. + let mut stat: libc::statvfs = unsafe { std::mem::zeroed() }; + let rc = unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) }; + if rc != 0 { + return None; + } + #[allow(clippy::unnecessary_cast)] + Some((stat.f_bavail as u64).saturating_mul(stat.f_frsize as u64)) +} + +#[cfg(not(unix))] +pub fn available_bytes(_path: &Path) -> Option { + None +} + +/// Apparent bytes under `dir` (file lengths, subdirectories walked, symlinks +/// skipped). `0` when absent. Blocking. +pub fn dir_size_blocking(dir: &Path) -> u64 { + let mut total = 0u64; + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + let Ok(rd) = std::fs::read_dir(&d) else { continue }; + for entry in rd.flatten() { + match entry.file_type() { + Ok(ft) if ft.is_dir() => stack.push(entry.path()), + Ok(ft) if ft.is_file() => { + if let Ok(meta) = entry.metadata() { + total = total.saturating_add(meta.len()); + } + } + _ => {} + } + } + } + total +} + +/// Bytes `path` actually occupies: allocated blocks, not apparent length (a +/// streaming-only torrent's file is sparse and its length is the whole film). +/// Blocking. +pub fn disk_usage_blocking(path: &Path) -> u64 { + let mut total = 0u64; + let mut stack = vec![path.to_path_buf()]; + while let Some(p) = stack.pop() { + let Ok(meta) = std::fs::symlink_metadata(&p) else { continue }; + if meta.is_dir() { + if let Ok(rd) = std::fs::read_dir(&p) { + stack.extend(rd.flatten().map(|e| e.path())); + } + } else if meta.is_file() { + total = total.saturating_add(allocated(&meta)); + } + } + total +} + +#[cfg(unix)] +fn allocated(meta: &std::fs::Metadata) -> u64 { + use std::os::unix::fs::MetadataExt; + meta.blocks().saturating_mul(512) +} + +#[cfg(not(unix))] +fn allocated(meta: &std::fs::Metadata) -> u64 { + meta.len() +} + +/// Top-level entries of `dir` that `is_owned` doesn't claim. +pub async fn unowned_entries( + dir: &Path, + is_owned: impl Fn(&std::ffi::OsStr) -> bool, +) -> Vec<(PathBuf, OsString)> { + let mut out = Vec::new(); + let Ok(mut rd) = tokio::fs::read_dir(dir).await else { + return out; + }; + while let Ok(Some(entry)) = rd.next_entry().await { + let name = entry.file_name(); + if !is_owned(&name) { + out.push((entry.path(), name)); + } + } + out +} + +/// Outcome of [`remove_if_stale`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Removal { + Removed(u64), + TooYoung, + Failed, +} + +/// Has nothing under `path` changed for `grace`? Unreadable → `false` (leave it). +pub async fn is_stale(path: &Path, grace: Duration, now: SystemTime) -> bool { + let p = path.to_path_buf(); + let newest = tokio::task::spawn_blocking(move || newest_change_blocking(&p)) + .await + .ok() + .flatten(); + match newest { + Some(t) => now.duration_since(t).map(|age| age >= grace).unwrap_or(false), + None => false, + } +} + +/// Remove a file or directory tree once nothing under it changed for `grace`. +/// Returns the bytes it occupied on disk. +pub async fn remove_if_stale(path: &Path, grace: Duration, now: SystemTime) -> Removal { + if !grace.is_zero() && !is_stale(path, grace, now).await { + return Removal::TooYoung; + } + let p = path.to_path_buf(); + let res = tokio::task::spawn_blocking(move || { + let bytes = disk_usage_blocking(&p); + let r = match std::fs::symlink_metadata(&p) { + Ok(m) if m.is_dir() => std::fs::remove_dir_all(&p), + Ok(_) => std::fs::remove_file(&p), + Err(e) => Err(e), + }; + r.map(|()| bytes) + }) + .await; + match res { + Ok(Ok(b)) => Removal::Removed(b), + Ok(Err(e)) if e.kind() == std::io::ErrorKind::NotFound => Removal::Removed(0), + Ok(Err(e)) => { + warn!(path = %path.display(), error = %e, "reclaim: removal failed"); + Removal::Failed + } + Err(_) => Removal::Failed, + } +} + +/// The most recent modification or status change of `path` or anything under +/// it. ctime matters: a container renamed from `tmp/` into `cache/` keeps its +/// old mtime, and only ctime records the move. +pub fn newest_change_blocking(path: &Path) -> Option { + let mut newest: Option = None; + let mut stack = vec![path.to_path_buf()]; + while let Some(p) = stack.pop() { + let Ok(meta) = std::fs::symlink_metadata(&p) else { + if p == path { + return None; + } + continue; + }; + for t in [meta.modified().ok(), change_time(&meta)].into_iter().flatten() { + newest = Some(newest.map_or(t, |n| n.max(t))); + } + if meta.is_dir() { + if let Ok(rd) = std::fs::read_dir(&p) { + stack.extend(rd.flatten().map(|e| e.path())); + } + } + } + newest +} + +#[cfg(unix)] +fn change_time(meta: &std::fs::Metadata) -> Option { + use std::os::unix::fs::MetadataExt; + let secs = u64::try_from(meta.ctime()).ok()?; + Some(SystemTime::UNIX_EPOCH + Duration::from_secs(secs)) +} + +#[cfg(not(unix))] +fn change_time(_meta: &std::fs::Metadata) -> Option { + None +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn young_entries_survive_and_zero_grace_removes() { + let d = std::env::temp_dir().join(format!("fsutil-{}", std::process::id())); + std::fs::create_dir_all(d.join("a")).unwrap(); + std::fs::write(d.join("a/f"), vec![1u8; 4096]).unwrap(); + let now = SystemTime::now(); + assert_eq!( + remove_if_stale(&d.join("a"), Duration::from_secs(3600), now).await, + Removal::TooYoung + ); + assert!(matches!( + remove_if_stale(&d.join("a"), Duration::ZERO, now).await, + Removal::Removed(_) + )); + assert!(!d.join("a").exists()); + assert_eq!(remove_if_stale(&d.join("gone"), Duration::ZERO, now).await, Removal::Removed(0)); + let _ = std::fs::remove_dir_all(&d); + } + + #[tokio::test] + async fn unowned_lists_only_unclaimed_names() { + let d = std::env::temp_dir().join(format!("fsutil-own-{}", std::process::id())); + std::fs::create_dir_all(d.join("keep")).unwrap(); + std::fs::create_dir_all(d.join("drop")).unwrap(); + let got = unowned_entries(&d, |n| n == "keep").await; + assert_eq!(got.len(), 1); + assert_eq!(got[0].1, "drop"); + let _ = std::fs::remove_dir_all(&d); + } + + #[test] + fn sizes_of_a_missing_dir_are_zero() { + assert_eq!(dir_size_blocking(Path::new("/definitely/not/here")), 0); + assert_eq!(disk_usage_blocking(Path::new("/definitely/not/here")), 0); + } +} diff --git a/src/transport/hull.rs b/src/transport/hull.rs new file mode 100644 index 0000000..8385316 --- /dev/null +++ b/src/transport/hull.rs @@ -0,0 +1,226 @@ +//! The plugin → hull direction: callbacks a plugin makes into meta-share. +//! +//! The hull serves these on a **separate internal listener** +//! (`META_SHARE_INTERNAL_LISTEN`, default `0.0.0.0:3001`) that no deployment +//! publishes: they reach meta-core's records and write to them, so they must +//! never sit behind the public perimeter. A plugin finds it at +//! `META_SHARE_HULL_URL` (default `http://metashare-app:3001`). +//! +//! Routes: +//! +//! - `POST /internal/events` ← [`Event`] → `204` once the hull acted. +//! - `GET /internal/records/:cid[?cached=1]` → [`RecordInfo`] (`404` = no +//! record). `cached=1` answers from the hull's record cache only (no +//! meta-core round trip) — for display names on a hot path. +//! - `GET /internal/nzb/manifest/:cid` → the `.nzb` bytes the cid resolves to, +//! with [`HDR_MANIFEST_SOURCE`] (`record` | `redeem`) and, for `record`, +//! [`HDR_MANIFEST_CID`]. `404` = not a Usenet cid; any other error carries +//! `{"error"}` — the manifest could not be redeemed or fetched. Parsing and +//! the `nzb-posting` digest check stay in the plugin. +//! +//! Why callbacks and not plugin-side logic: these need what only the hull has — +//! the record cache and meta-core, the gateway redeem, the IPFS tier. Plugins +//! never call each other; the hull routes every such edge. + +use std::time::Duration; + +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +use super::dto::Event; + +/// Default hull callback base, matching the dev and store compose service name. +pub const DEFAULT_HULL_URL: &str = "http://metashare-app:3001"; + +/// Where the hull got a manifest: the record's `manifest` pointer, or a +/// gateway redeem. +pub const HDR_MANIFEST_SOURCE: &str = "x-metamesh-manifest-source"; +/// The content cid of a `record`-sourced manifest. +pub const HDR_MANIFEST_CID: &str = "x-metamesh-manifest-cid"; + +/// `.nzb` bytes as the hull resolved them. +#[derive(Clone, Debug)] +pub struct ManifestBytes { + pub bytes: bytes::Bytes, + /// `true` when redeemed through a gateway, `false` for the record pointer. + pub redeemed: bool, + pub manifest_cid: Option, +} + +/// What a plugin may need to know about a record. +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct RecordInfo { + /// meta-core's hash id for the record (write-back key). + #[serde(default)] + pub hash_id: Option, + /// The record's `fileName`. + #[serde(default)] + pub file_name: Option, + /// The record's display title. + #[serde(default)] + pub title: Option, +} + +/// Plugin-side client for the hull's internal API. A client with no base URL +/// (`META_SHARE_HULL_URL=` empty, or [`HullClient::disabled`]) answers every call +/// with "not available", which is what unit tests and a standalone plugin get. +#[derive(Clone, Debug)] +pub struct HullClient { + base: Option, + http: reqwest::Client, +} + +impl HullClient { + pub fn new(base: Option, http: reqwest::Client) -> Self { + let base = base + .map(|b| b.trim().trim_end_matches('/').to_string()) + .filter(|b| !b.is_empty()); + Self { base, http } + } + + /// From `META_SHARE_HULL_URL`, defaulting to [`DEFAULT_HULL_URL`]; an + /// explicitly empty value disables callbacks. + pub fn from_env(http: reqwest::Client) -> Self { + let base = match std::env::var("META_SHARE_HULL_URL") { + Ok(v) => Some(v), + Err(_) => Some(DEFAULT_HULL_URL.to_string()), + }; + Self::new(base, http) + } + + pub fn disabled() -> Self { + Self::new(None, reqwest::Client::new()) + } + + pub fn is_enabled(&self) -> bool { + self.base.is_some() + } + + fn url(&self, path: &str) -> Result { + self.base + .as_ref() + .map(|b| format!("{b}{path}")) + .ok_or_else(|| "hull callbacks disabled (META_SHARE_HULL_URL unset)".to_string()) + } + + /// Notify the hull; returns once it has acted. + pub async fn emit(&self, ev: &Event) -> Result<(), String> { + let r = self + .http + .post(self.url("/internal/events")?) + .json(ev) + .timeout(Duration::from_secs(30)) + .send() + .await + .map_err(|e| format!("hull unreachable: {e}"))?; + if r.status().is_success() { + Ok(()) + } else { + Err(format!("hull refused event: {}", r.status())) + } + } + + /// `GET` a JSON resource; `404` → `Ok(None)`. + pub async fn get_json(&self, path: &str, timeout: Duration) -> Result, String> { + let r = self + .http + .get(self.url(path)?) + .timeout(timeout) + .send() + .await + .map_err(|e| format!("hull unreachable: {e}"))?; + if r.status().as_u16() == 404 { + return Ok(None); + } + if !r.status().is_success() { + return Err(error_message(r).await); + } + r.json::().await.map(Some).map_err(|e| format!("hull sent malformed JSON: {e}")) + } + + /// `GET` raw bytes; `404` → `Ok(None)`, any other failure → the hull's + /// `{"error"}` message. + pub async fn get_bytes(&self, path: &str, timeout: Duration) -> Result, String> { + let r = self + .http + .get(self.url(path)?) + .timeout(timeout) + .send() + .await + .map_err(|e| format!("hull unreachable: {e}"))?; + if r.status().as_u16() == 404 { + return Ok(None); + } + if !r.status().is_success() { + return Err(error_message(r).await); + } + r.bytes().await.map(Some).map_err(|e| format!("hull body: {e}")) + } + + /// The `.nzb` behind a Usenet cid. `Ok(None)` = not a Usenet cid. + pub async fn nzb_manifest(&self, cid: &str, timeout: Duration) -> Result, String> { + let r = self + .http + .get(self.url(&format!("/internal/nzb/manifest/{cid}"))?) + .timeout(timeout) + .send() + .await + .map_err(|e| format!("hull unreachable: {e}"))?; + if r.status().as_u16() == 404 { + return Ok(None); + } + if !r.status().is_success() { + return Err(error_message(r).await); + } + let h = r.headers(); + let redeemed = h + .get(HDR_MANIFEST_SOURCE) + .and_then(|v| v.to_str().ok()) + .is_some_and(|v| v == "redeem"); + let manifest_cid = h + .get(HDR_MANIFEST_CID) + .and_then(|v| v.to_str().ok()) + .map(str::to_string); + let bytes = r.bytes().await.map_err(|e| format!("hull body: {e}"))?; + Ok(Some(ManifestBytes { bytes, redeemed, manifest_cid })) + } + + /// [`RecordInfo`] for `cid` (record cache, else meta-core), or `None` when + /// the hull has no record. + pub async fn record(&self, cid: &str) -> Result, String> { + self.get_json(&format!("/internal/records/{cid}"), Duration::from_secs(15)).await + } + + /// [`RecordInfo`] from the hull's record cache only. + pub async fn cached_record(&self, cid: &str) -> Result, String> { + self.get_json(&format!("/internal/records/{cid}?cached=1"), Duration::from_secs(5)).await + } +} + +async fn error_message(r: reqwest::Response) -> String { + let status = r.status(); + r.json::() + .await + .ok() + .and_then(|v| v.get("error").and_then(|e| e.as_str()).map(str::to_string)) + .unwrap_or_else(|| format!("hull error {status}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn a_disabled_client_refuses_without_network() { + let h = HullClient::disabled(); + assert!(!h.is_enabled()); + assert!(h.emit(&Event::Archive { cid: "x".into() }).await.is_err()); + assert!(h.record("x").await.is_err()); + } + + #[test] + fn blank_base_disables() { + assert!(!HullClient::new(Some(" ".into()), reqwest::Client::new()).is_enabled()); + assert!(HullClient::new(Some("http://h:3001/".into()), reqwest::Client::new()).is_enabled()); + } +} diff --git a/src/transport/ipfs.rs b/src/transport/ipfs.rs new file mode 100644 index 0000000..21bc9df --- /dev/null +++ b/src/transport/ipfs.rs @@ -0,0 +1,157 @@ +//! IPFS-specific half of the contract: the ipfs plugin's extra routes. +//! +//! The ipfs plugin owns the libp2p swarm (identify / kad / mdns / bitswap), the +//! blockstore (`ipfs/blocks.redb`: material blocks + leaf refs into files on the +//! shared volume), bitswap ingress and the peer directory. These routes are the +//! hull's facade onto that — each mirrors one blockstore / swarm call the hull +//! made in-process, so the hull's logic (seed rows, eviction, meta-core links, +//! probe grading) did not have to change shape. +//! +//! Public passthrough (the hull relays meta-share's own route verbatim): +//! +//! - `GET /ipfs/:cid` — the read-only IPFS gateway. +//! +//! Internal (all under `/ipfs-tier/`): +//! +//! | route | in-process equivalent | +//! |---|---| +//! | `GET peers` → [`PeersInfo`] | `swarm::Command::Peers` | +//! | `GET directory` → [`Directory`] | `swarm::PeerDirectory` (gateway discovery) | +//! | `GET local/:cid` (Range) → bytes / `404` | `files::bitswap::try_local_block` | +//! | `GET has/:cid` → [`Present`] | `block_store.get(cid).is_some()` | +//! | `GET complete/:cid` → [`Complete`] | `local_dag_complete` | +//! | `GET cat/:cid?max=` → bytes | `ipfs_walker::get_block` (+ dag-pb assemble) | +//! | `GET fetch/:cid?timeout_ms=` → block bytes | `block_store.get` → `bitswap_get_block` | +//! | `PUT block/:cid` ← bytes | `blockstore::put_ipfs_block` | +//! | `POST import` ← file bytes → [`Imported`] | `put_ipfs_blocks(compute_ipfs_blocks(..))` | +//! | `POST forget/:cid` | `api::seeds::remove_ipfs_blocks` | +//! | `POST forget-library` ← [`ForgetLibrary`] | `block_store.remove_library_dag` | +//! | `POST provide/:cid`, `POST unprovide/:cid` | `Command::Provide` / `StopProviding` | +//! | `POST drop-refs` ← [`DropRefs`] | `block_store.drop_refs_for_materials` | +//! | `GET first-leaf/:cid` → [`FirstLeaf`] | `block_store.first_leaf_backing` | +//! | `GET rel-path/:midhash` → [`RelPath`] | `resolver().rel_path_for` | +//! | `POST share/library` ← [`ShareLibrary`] → [`Shared`] | `ipfs_seed::seed_file_as_refs` | +//! | `POST share/container` ← [`ShareContainer`] → [`Shared`] | the seal / rebuild re-chunk | +//! | `GET stats/blockstore` | `/api/stats/blockstore` body | +//! | `GET debug/:cid` | `/api/debug/blockstore/:cid` body | +//! +//! Plugin → hull, over [`super::hull`]: [`Event::Seed`](super::Event::Seed) +//! (a bitswap fetch now seeds), [`Event::Promoted`](super::Event::Promoted) (an +//! ingress job was renamed into `cache/`), and `GET /internal/records/:cid` for +//! a display name. + +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct PeersInfo { + pub local_peer_id: String, + pub connected: Vec, +} + +/// One peer of the directory, with its freshness as *ages* (the hull rebuilds +/// its own `Instant`s from them). +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq)] +pub struct DirectoryPeer { + pub peer_id: String, + #[serde(default)] + pub base_url: Option, + /// Milliseconds since the gateway last advertised `nzbFetch`; `None` = not. + #[serde(default)] + pub nzb_fetch_age_ms: Option, + /// Milliseconds since the caps were fetched; `None` = never. + #[serde(default)] + pub caps_age_ms: Option, + /// The gateway's redeem claims, as the gateway serves them + /// (`{codec, field, hosts, sources}`). + #[serde(default)] + pub redeems: Vec, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq)] +pub struct Directory { + pub peers: Vec, +} + +#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Present { + pub present: bool, +} + +#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Complete { + pub complete: bool, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct Imported { + pub root: String, + pub size: u64, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct ForgetLibrary { + pub root: String, + pub midhash: String, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct DropRefs { + /// `(container, rel)` pairs. + pub materials: Vec<(String, String)>, +} + +/// Where a dag's first leaf points (the filestore's `LeafBacking`). +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum LeafBacking { + /// A ref into a file meta-core owns, by midhash256. + MetaCore(String), + /// A ref into a file in a cache container. + Cache, + /// The leaf's bytes are stored material (swarm-fetched, no file behind it). + Material, + /// The root or a node on the way down is absent. + Unknown, +} + +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct FirstLeaf { + pub backing: LeafBacking, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct RelPath { + pub rel: Option, +} + +/// Seed a meta-core library file as refs `(midhash, offset, len)`. +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct ShareLibrary { + /// Path relative to the files volume (what meta-core's WebDAV serves). + pub rel: String, + pub midhash: String, +} + +/// Seed a file in a cache container as refs `(container, rel, offset, len)`. +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct ShareContainer { + pub container: String, + pub rel: String, + /// Absolute path on the shared volume. + pub path: String, + /// Verify the resulting dag is fully held before answering (the seal does; + /// the rebuild doesn't). + #[serde(default)] + pub verify_complete: bool, +} + +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct Shared { + pub root: String, + pub size: u64, + /// Leaf refs written. + pub refs: u64, + /// `verify_complete` was asked and the dag is fully held locally. + #[serde(default)] + pub complete: bool, +} diff --git a/src/transport/magic.rs b/src/transport/magic.rs new file mode 100644 index 0000000..cad50c9 --- /dev/null +++ b/src/transport/magic.rs @@ -0,0 +1,230 @@ +//! Content sniffing — what the bytes *are*, independent of what they are called. +//! +//! Every classifier in the usenet plugin's `archive` keys off the filename quoted in an NZB +//! subject. A fully obfuscated posting has no filename to key off, so all of them +//! fail **open**: `subject_is_par2` misses the parity volumes, `subject_is_rar` +//! misses the archive, and `pick_media`'s `best_video.or(best_any)` then elects +//! the largest arbitrary blob as "the movie". That blob is what gets served, and +//! — far worse — what the Usenet → IPFS re-seed hashes and publishes as +//! the record's content identity. +//! +//! Names are advisory. Bytes are not. This module is the byte half. +//! +//! The resolution order mirrors `filename-tool`'s `FileTypeConfigurable` +//! (the source of truth named by `METADATA_KEYS.md` §`fileType`): sniff the +//! content first, compare with the extension when there is one, and on +//! **disagreement return `undefined` rather than guessing** — a wrong `fileType` +//! routes a file to a handler that cannot open it. + +/// A container format recognised from its leading bytes. +/// +/// Deliberately *not* exhaustive over "every format that exists" — it only needs +/// to separate payload from wrapper from parity well enough to decide whether a +/// posting has resolved. [`Magic::Unknown`] is a first-class answer, and callers +/// must treat it as "cannot prove anything", never as "bad". +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Magic { + /// RAR 1.5–4.x — signature `52 61 72 21 1A 07 00`. + Rar4, + /// RAR 5.0+ — signature `52 61 72 21 1A 07 01 00`. + Rar5, + /// par2 recovery data. **Never payload**, whatever its size. + Par2, + SevenZip, + /// A zip that is not an EPUB (includes `.cbz`, which `METADATA_KEYS.md` + /// §`fileType` classifies as `archive`). + Zip, + /// A zip whose first entry is the stored `mimetype` = `application/epub+zip`. + /// Split from [`Magic::Zip`] because an EPUB **is** the document, not a + /// wrapper around one — three correct ebook postings on watch.nsl.sh are + /// this shape and a naive "zip ⇒ archive" rule condemns all of them. + Epub, + /// Palm database holding an ebook (`BOOK`/`TEXt` type at offset 60) — the + /// `CR!…`-named MobiPocket/eReader files an indexer hands back for a book. + PalmDoc, + Matroska, + Mp4, + /// RIFF container (AVI, and also WAV — the extension disambiguates). + Riff, + Pdf, + /// Nothing recognised. Not an error: see the type-level note above. + Unknown, +} + +impl Magic { + /// The `METADATA_KEYS.md` §`fileType` value these bytes imply, or `None` + /// when the content says nothing. + /// + /// [`Magic::Riff`] is deliberately `None`: RIFF is AVI *and* WAV, i.e. a + /// video and an audio type, and guessing between them is exactly the + /// "disagreement returns undefined" case the registry legislates. + pub fn file_type(self) -> Option<&'static str> { + match self { + Magic::Matroska | Magic::Mp4 => Some("video"), + Magic::Epub | Magic::PalmDoc | Magic::Pdf => Some("document"), + Magic::Rar4 | Magic::Rar5 | Magic::SevenZip | Magic::Zip => Some("archive"), + // Parity is not a fileType — it is not payload at all. Callers route + // on `is_parity()` before they ever ask for a type. + Magic::Par2 => None, + Magic::Riff | Magic::Unknown => None, + } + } + + /// The extension these bytes imply, for naming a materialised file whose + /// posted name carried none (or carried an obfuscated one) — see + /// the usenet plugin's `naming`. + /// + /// Abstains wherever the bytes do not pick a single extension, on the same + /// discipline as [`Magic::file_type`]: [`Magic::Riff`] is AVI *and* WAV, and + /// [`Magic::PalmDoc`] is `.prc`/`.mobi`/`.pdb`. [`Magic::Par2`] abstains for + /// a stronger reason — parity is never the payload, so it never gets named. + pub fn extension(self) -> Option<&'static str> { + match self { + Magic::Matroska => Some("mkv"), + Magic::Mp4 => Some("mp4"), + Magic::Pdf => Some("pdf"), + Magic::Epub => Some("epub"), + Magic::Rar4 | Magic::Rar5 => Some("rar"), + Magic::SevenZip => Some("7z"), + Magic::Zip => Some("zip"), + Magic::Riff | Magic::PalmDoc | Magic::Par2 | Magic::Unknown => None, + } + } + + /// par2 recovery data: present to repair the payload, never to *be* it. + /// + /// One posting on watch.nsl.sh (`…gk2iwj5i…`, claiming `video`/`movie`) + /// materialised to a single 122 MB par2 file and published a dag-pb root + /// over it, because `subject_is_par2` reads the `.par2` extension the + /// obfuscated subject did not carry. + pub fn is_parity(self) -> bool { + matches!(self, Magic::Par2) + } + + /// A multi-volume wrapper that must be unwrapped before anything inside it + /// can be addressed. `Epub`/`PalmDoc` are containers in the file-format + /// sense but **not** in this one — they are the document. + pub fn is_archive(self) -> bool { + matches!(self, Magic::Rar4 | Magic::Rar5 | Magic::SevenZip | Magic::Zip) + } +} + +/// Bytes needed for a confident answer. The EPUB probe is the deepest reader: +/// local header (30) + `"mimetype"` (8) + extra field + `"application/epub+zip"` +/// (20). 128 covers any sane extra-field length. +pub const SNIFF_LEN: usize = 128; + +/// Classify a file from its leading bytes. Short input is fine — every probe +/// bounds-checks and falls through to [`Magic::Unknown`]. +pub fn sniff(head: &[u8]) -> Magic { + // RAR5 before RAR4: the RAR4 signature is a prefix of neither, but the + // shared first 6 bytes make ordering worth being explicit about. + if head.starts_with(b"Rar!\x1a\x07\x01\x00") { + return Magic::Rar5; + } + if head.starts_with(b"Rar!\x1a\x07\x00") { + return Magic::Rar4; + } + if head.starts_with(b"PAR2\x00PKT") { + return Magic::Par2; + } + if head.starts_with(b"7z\xbc\xaf\x27\x1c") { + return Magic::SevenZip; + } + if head.starts_with(b"%PDF-") { + return Magic::Pdf; + } + // EBML — Matroska and WebM share it; both demux the same way here. + if head.starts_with(b"\x1a\x45\xdf\xa3") { + return Magic::Matroska; + } + if head.len() >= 8 && &head[4..8] == b"ftyp" { + return Magic::Mp4; + } + if head.starts_with(b"RIFF") { + return Magic::Riff; + } + if head.starts_with(b"PK\x03\x04") { + return if is_epub(head) { Magic::Epub } else { Magic::Zip }; + } + if is_palm_ebook(head) { + return Magic::PalmDoc; + } + Magic::Unknown +} + +/// An EPUB is a zip whose **first** entry is an uncompressed `mimetype` member +/// holding exactly `application/epub+zip` (OCF 3, §4.2 — the rule exists so the +/// type is readable without inflating anything). That is what separates a book +/// from a `.cbz` or a scene `.zip`. +fn is_epub(head: &[u8]) -> bool { + const NAME_AT: usize = 30; + const NAME: &[u8] = b"mimetype"; + const VALUE: &[u8] = b"application/epub+zip"; + + // Stored (method 0) is mandatory for this member; anything else is not an + // OCF container even if it happens to hold the string. + if head.len() < NAME_AT + NAME.len() || u16::from_le_bytes([head[8], head[9]]) != 0 { + return false; + } + if &head[NAME_AT..NAME_AT + NAME.len()] != NAME { + return false; + } + let extra = u16::from_le_bytes([head[28], head[29]]) as usize; + let at = NAME_AT + NAME.len() + extra; + head.len() >= at + VALUE.len() && &head[at..at + VALUE.len()] == VALUE +} + +/// A Palm database (`name[32]`, then `type` at 60 and `creator` at 64) whose +/// type marks it an ebook. Covers MobiPocket (`BOOK`/`MOBI`), eReader +/// (`BOOK`/`PNRd`) and PalmDOC (`TEXt`/`REAd`). +fn is_palm_ebook(head: &[u8]) -> bool { + const TYPE_AT: usize = 60; + head.len() >= TYPE_AT + 4 && matches!(&head[TYPE_AT..TYPE_AT + 4], b"BOOK" | b"TEXt") +} + +/// The `fileType` an extension implies, for the agreement check. Mirrors the +/// table in `METADATA_KEYS.md` §`fileType`, whose own source of truth is +/// `filename-tool`'s `ExtentionsMapping.ts`. Only the values reachable from a +/// Usenet posting are listed; anything else answers `None` ("no opinion"), +/// which agrees with everything. +pub fn file_type_from_extension(name: &str) -> Option<&'static str> { + let ext = name.rsplit_once('.')?.1.to_ascii_lowercase(); + Some(match ext.as_str() { + "mkv" | "mp4" | "webm" | "avi" | "mov" | "wmv" | "flv" | "m4v" | "mpg" | "mpeg" | "3gp" + | "ts" | "m2ts" | "mts" | "vob" | "ogm" | "ogv" | "divx" | "xvid" => "video", + "mp3" | "wav" | "flac" | "aac" | "ogg" | "oga" | "opus" | "m4a" | "m4b" | "wma" | "mka" => { + "audio" + } + "pdf" | "epub" | "mobi" | "azw" | "azw3" | "djvu" | "txt" | "doc" | "docx" | "prc" => { + "document" + } + "zip" | "rar" | "7z" | "tar" | "gz" | "bz2" | "xz" | "zst" | "cbz" | "cbr" => "archive", + "srt" | "sub" | "ass" | "ssa" | "vtt" | "sbv" | "smi" => "subtitle", + "par2" => return None, // parity, not a fileType — see `Magic::is_parity` + _ => return None, + }) +} + +/// Resolve one file's `fileType` from its bytes and its name together. +/// +/// The registry's rule, verbatim: *"sniff the MIME from the first bytes; if it +/// maps to a definite value, compare it with the extension's value — agreement +/// wins, **disagreement returns `undefined`**"*. `None` here is that +/// `undefined`: it means "these two sources disagree, do not route on it", +/// which is a different and strictly weaker statement than "wrong". +pub fn resolve_file_type(head: &[u8], name: &str) -> Option<&'static str> { + let sniffed = sniff(head).file_type(); + let from_name = file_type_from_extension(name); + match (sniffed, from_name) { + (Some(a), Some(b)) if a == b => Some(a), + (Some(_), Some(_)) => None, // disagreement ⇒ undefined + (Some(a), None) => Some(a), // no usable extension: content decides + (None, Some(b)) => Some(b), // unrecognised bytes: the name is all we have + (None, None) => None, + } +} + +#[cfg(test)] +#[path = "magic_tests.rs"] +mod tests; diff --git a/src/transport/magic_tests.rs b/src/transport/magic_tests.rs new file mode 100644 index 0000000..2adf071 --- /dev/null +++ b/src/transport/magic_tests.rs @@ -0,0 +1,205 @@ +//! Tests for [`super`] — content sniffing. +//! +//! Every fixture below is **real**: the leading bytes of files materialised on +//! watch.nsl.sh, captured 2026-09-16. Synthetic headers would have agreed with +//! whatever this module happened to implement; these are the bytes that actually +//! defeated the name-based classifiers. + +use super::*; + +/// Decode a hex fixture. Panics on bad input — these are consts, not user data. +fn hx(s: &str) -> Vec { + assert!(s.len() % 2 == 0, "odd-length hex fixture"); + (0..s.len() / 2) + .map(|i| u8::from_str_radix(&s[i * 2..i * 2 + 2], 16).expect("hex")) + .collect() +} + +/// Volume 1 of 35 of the trigger posting — `[smol] Tensei Oujo … S01E05v2`, +/// `bagcsaaa5brqxa2jonz5geltmnftglnrqztt2n23topk2bsnglvhjmty`. Extension-less +/// name, RAR5, method 0 (store). The EBML header of the contained MKV is +/// visible inside these same 200 bytes, at the volume's data offset. +const RAR5_VOL1: &str = "526172211a0701002409274c1001050c010b0101bac0d7af80808080004ce244be8b0102130b94bfd7af80800004eefaa1818d8000b683021a946591800001635b736d6f6c5d2054656e736569204f756a6f20746f2054656e736169205265696a6f75206e6f204d61686f75204b616b756d6569202d205330314530357632202842442031303830702048455643204f70757329205b39394434344436465d2e6d6b760a0313184c91644af1d2231a45dfa3a34286810142f7810142f2810442f381084282886d61"; + +/// Volume 2 — a *continuation* volume. Byte-identical signature to volume 1, +/// which is why "does it start with `Rar!`" cannot tell you which volume you +/// hold, and why electing "the largest file" picks one at random. +const RAR5_VOL2: &str = "526172211a0701000d6002471101050c03010b0101bac0d7af808080800030554b778b01021b0b93bfd7af80800004eefaa1818d8000b6830236ead743800001635b736d6f6c5d2054656e736569204f756a6f20746f2054656e736169205265696a6f75206e6f204d61686f75204b616b756d6569202d205330314530357632202842442031303830702048455643204f70757329205b39394434344436465d2e6d6b760a0313184c91644af1d223da5b7aaacbc910a46f8b3a8886bc4bfaff0518a54e4a2eb791"; + +/// RAR4 head, from `…gkigqufwwvhpkbhijvufrdcshogy` (claims `video`/`movie`). +const RAR4: &str = "526172211a0700197a7311000d00000000000000d9427423916f0069ff3f067a"; + +/// par2 head, from `…gk2iwj5iy3cyzafghckcab7j2fuy` — the posting that +/// materialised to a *single* par2 file and published a dag-pb root over it. +const PAR2: &str = "5041523200504b54083b1a00000000000226e38aa627ac7ee833f4fc92637a69"; + +/// A genuine EPUB — `Brandon Sanderson - Mistborn`, `…gkdthedplbm5qwmtedo2ymyct2si`. +/// Stored `mimetype` member first, exactly as OCF requires. +const EPUB: &str = "504b0304140000000000ee523a486f61ab2c1400000014000000080000006d696d65747970656170706c69636174696f6e2f657075622b7a6970504b0304140002080800ee523a4827f0a79ab4000000"; + +/// A MobiPocket Palm database — the `CR!…`-named ebook from `…gle4fanocf5txcdocnto6dh4au4a`. +/// Type `BOOK` at offset 60, creator `MOBI` at 64. +const PALM: &str = "43522158593841474857463731314e4642394a44314443314d374b4548344d000000000057d30d1e57d3113200000000000000000000000000000000424f4f4b4d4f4249000001df"; + +/// A Matroska head from a posting that resolved correctly. +const MKV: &str = "1a45dfa3a34286810142f7810142f281"; + +// ---- the six shapes that mattered ------------------------------------------- + +#[test] +fn the_real_headers_are_each_recognised() { + assert_eq!(sniff(&hx(RAR5_VOL1)), Magic::Rar5); + assert_eq!(sniff(&hx(RAR5_VOL2)), Magic::Rar5); + assert_eq!(sniff(&hx(RAR4)), Magic::Rar4); + assert_eq!(sniff(&hx(PAR2)), Magic::Par2); + assert_eq!(sniff(&hx(EPUB)), Magic::Epub); + assert_eq!(sniff(&hx(PALM)), Magic::PalmDoc); + assert_eq!(sniff(&hx(MKV)), Magic::Matroska); +} + +/// The regression that started this: three correct ebook postings on the box are +/// zips, and a "zip ⇒ archive" rule condemns all of them. An EPUB is the +/// document, not a wrapper around one. +#[test] +fn an_epub_is_a_document_not_an_archive() { + let m = sniff(&hx(EPUB)); + assert_eq!(m.file_type(), Some("document")); + assert!(!m.is_archive(), "an epub must not route into the unwrapper"); +} + +/// A `.cbz` and a scene `.zip` share the EPUB signature for four bytes and +/// diverge at the first member. `METADATA_KEYS.md` §`fileType` puts `cbz` in +/// `archive`, so plain-zip ⇒ archive is the correct fallback. +#[test] +fn a_zip_without_the_ocf_mimetype_member_is_an_archive() { + let mut z = hx(EPUB); + // Corrupt just the mimetype *value*; everything else stays a valid zip. + let at = 38; + z[at..at + 3].copy_from_slice(b"XXX"); + assert_eq!(sniff(&z), Magic::Zip); + assert_eq!(sniff(&z).file_type(), Some("archive")); +} + +/// OCF requires the `mimetype` member to be *stored*. A deflated one is not an +/// OCF container even if the bytes happen to spell the string. +#[test] +fn a_deflated_mimetype_member_is_not_an_epub() { + let mut z = hx(EPUB); + z[8] = 8; // method = deflate + assert_eq!(sniff(&z), Magic::Zip); +} + +#[test] +fn parity_is_not_a_file_type_at_all() { + let m = sniff(&hx(PAR2)); + assert!(m.is_parity()); + assert_eq!(m.file_type(), None, "par2 must never present as payload"); +} + +#[test] +fn every_rar_generation_reads_as_an_archive() { + for f in [RAR4, RAR5_VOL1, RAR5_VOL2] { + let m = sniff(&hx(f)); + assert!(m.is_archive()); + assert_eq!(m.file_type(), Some("archive")); + assert!(!m.is_parity()); + } +} + +// ---- Unknown is an answer, not an error ------------------------------------- + +#[test] +fn unrecognised_and_truncated_input_is_unknown_not_a_panic() { + assert_eq!(sniff(b""), Magic::Unknown); + assert_eq!(sniff(b"Rar"), Magic::Unknown); + assert_eq!(sniff(b"PK\x03"), Magic::Unknown); + assert_eq!(sniff(&[0u8; 8]), Magic::Unknown); + assert_eq!(sniff(b"not a container at all"), Magic::Unknown); + assert_eq!(sniff(&hx(MKV)[..2]), Magic::Unknown); +} + +/// The holed posting on the box (`…gkisayoy…`, 837 MB of zeros) sniffs Unknown. +/// It is already refused upstream by `disk::first_hole`; this only pins that +/// Unknown carries no `fileType` claim either way. +#[test] +fn a_hole_claims_nothing() { + let zeros = vec![0u8; SNIFF_LEN]; + assert_eq!(sniff(&zeros), Magic::Unknown); + assert_eq!(sniff(&zeros).file_type(), None); +} + +/// RIFF is AVI *and* WAV — a video type and an audio type. Guessing between +/// them is precisely the disagreement case the registry legislates away. +#[test] +fn riff_is_recognised_but_declines_to_name_a_type() { + let m = sniff(b"RIFF\x00\x00\x00\x00AVI LIST"); + assert_eq!(m, Magic::Riff); + assert_eq!(m.file_type(), None); +} + +// ---- name + bytes together --------------------------------------------------- + +#[test] +fn content_decides_when_the_name_says_nothing() { + // The obfuscated case: no extension anywhere. + assert_eq!( + resolve_file_type(&hx(RAR5_VOL1), "IuYf3jBKJOukfUesqNCw0LoUc1Gqxxmilo0ICJJMeddjB6"), + Some("archive") + ); + assert_eq!(resolve_file_type(&hx(MKV), "56GqgfEUMp7RAqF94AGp"), Some("video")); +} + +#[test] +fn agreement_between_name_and_bytes_wins() { + assert_eq!(resolve_file_type(&hx(MKV), "The.Movie.2019.1080p.mkv"), Some("video")); + assert_eq!(resolve_file_type(&hx(EPUB), "book.epub"), Some("document")); +} + +/// `METADATA_KEYS.md` §`fileType`: *"disagreement returns `undefined`"*. A file +/// named `.mkv` whose bytes are a RAR is not a video and is not reliably an +/// archive either — it is a thing we refuse to route on. +#[test] +fn disagreement_returns_undefined_rather_than_guessing() { + assert_eq!(resolve_file_type(&hx(RAR5_VOL1), "episode.mkv"), None); + assert_eq!(resolve_file_type(&hx(MKV), "release.rar"), None); +} + +#[test] +fn an_unreadable_header_falls_back_to_the_extension() { + assert_eq!(resolve_file_type(b"\x00\x01\x02\x03", "movie.mkv"), Some("video")); + assert_eq!(resolve_file_type(b"", "chapter.cbz"), Some("archive")); +} + +#[test] +fn neither_source_has_an_opinion() { + assert_eq!(resolve_file_type(b"\x00\x01\x02\x03", "blob"), None); + // par2 is deliberately absent from the extension table: it is parity, and + // `Magic::is_parity` is how callers route it. + assert_eq!(file_type_from_extension("recovery.par2"), None); +} + +#[test] +fn the_extension_table_matches_the_registry_for_the_shapes_we_see() { + assert_eq!(file_type_from_extension("a.mkv"), Some("video")); + assert_eq!(file_type_from_extension("a.MKV"), Some("video")); + assert_eq!(file_type_from_extension("a.epub"), Some("document")); + assert_eq!(file_type_from_extension("a.cbz"), Some("archive")); + assert_eq!(file_type_from_extension("a.rar"), Some("archive")); + assert_eq!(file_type_from_extension("a.srt"), Some("subtitle")); + assert_eq!(file_type_from_extension("no-extension-here"), None); +} + +/// The MKV inside the trigger posting is *stored*, so its EBML header sits in +/// plaintext a few hundred bytes into volume 1. That is what makes the +/// offset-map unwrap possible at all, and it is worth pinning independently of +/// the RAR parser. +#[test] +fn the_stored_mkv_header_is_visible_inside_volume_one() { + let v1 = hx(RAR5_VOL1); + let ebml = v1 + .windows(4) + .position(|w| w == [0x1a, 0x45, 0xdf, 0xa3]) + .expect("EBML header should be plainly visible inside a stored RAR volume"); + assert_eq!(ebml, 174, "volume 1's data offset, per the RAR5 header chain"); + assert_eq!(sniff(&v1[ebml..]), Magic::Matroska); +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs new file mode 100644 index 0000000..ee68723 --- /dev/null +++ b/src/transport/mod.rs @@ -0,0 +1,57 @@ +//! **Transport plugins** — the contract between meta-share's protocol-agnostic +//! hull and the processes that actually move bytes (`meta-transport-torrent`, +//! `meta-transport-nzb`, `meta-transport-ipfs`, or a competing implementation +//! of any of them). +//! +//! Same split as gateway ⇄ feeders: the plugin finds and moves bytes for one +//! protocol; the hull keeps everything that is policy and bookkeeping — the +//! public HTTP API, the material index, seed rows, eviction, focus, meta-core. +//! Plugins never call each other and never talk to meta-core's record API; +//! every cross-tier edge goes through the hull. +//! +//! - [`plugin::TransportPlugin`] + [`serve::serve_transport`]: the plugin side. +//! - [`client::RemoteTransport`]: the hull side. +//! - [`hull::HullClient`]: plugin → hull callbacks (internal port only). +//! - [`focus::FocusView`]: the plugin's replica of the hull's playback focus. +//! - [`dto`]: common wire types; [`error::ApiError`] / [`range`]: shared so a +//! relayed plugin response is byte-identical to what the monolith answered. +//! - [`cid`]: CID parsing + cid-shape decoders, shared so every process +//! classifies a cid the same way. +//! - [`torrent`]: the torrent plugin's extra routes + magnet helpers. +//! - [`ipfs`]: the ipfs plugin's extra routes (the hull's blockstore/swarm facade). +//! - [`nzb`]: Usenet cid shapes, the usenet plugin's extra routes + settings. +//! - [`magic`]: content sniffing; [`file`]: range-serving a file off `/data`. +//! - `testkit` (feature `testkit`): conformance checks for any implementation. +//! +//! Bytes on disk: every plugin mounts the same `/data` volume as the hull, so a +//! plugin-written file is opened by the hull directly (local-first reads) and +//! `rename(2)` between `tmp/` and `cache/` stays atomic. + +pub mod cid; +pub mod client; +pub mod dto; +pub mod error; +pub mod file; +pub mod focus; +pub mod fsutil; +pub mod hull; +pub mod ipfs; +pub mod magic; +pub mod nzb; +pub mod plugin; +pub mod range; +pub mod serve; +pub mod torrent; +#[cfg(any(test, feature = "testkit"))] +pub mod testkit; + +pub use client::RemoteTransport; +pub use dto::{ + Capabilities, Deleted, Event, Health, Job, Jobs, Manifest, PromotedFile, ReconcileReport, + ReconcileRequest, CONTRACT_VERSION, +}; +pub use error::ApiError; +pub use focus::{FocusSnapshot, FocusTitle, FocusView, Lane}; +pub use hull::{HullClient, RecordInfo}; +pub use plugin::TransportPlugin; +pub use serve::serve_transport; diff --git a/src/transport/nzb.rs b/src/transport/nzb.rs new file mode 100644 index 0000000..7aaeea1 --- /dev/null +++ b/src/transport/nzb.rs @@ -0,0 +1,560 @@ +//! Usenet cid shapes, shared by the hull and the usenet transport plugin, plus +//! the plugin's extra-route DTOs and its settings. +//! +//! The codec helpers are moved verbatim from meta-share `nzb/mod.rs`; the hull +//! dispatches and grades with them, the plugin decodes with them, so they must +//! be one copy. +//! +//! Extra routes of the usenet plugin (all under `/nzb/`): +//! +//! - `GET /nzb/status/:cid` → the `kind:"nzb"` status JSON (never starts a job) +//! - `GET /nzb/jobs` → `{"jobs":[…]}` in-flight jobs (`NzbActivity` JSON) +//! - `DELETE /nzb/jobs/:cid` → `{"cancelled":bool}` +//! - `GET /nzb/capability` → [`Capability`] +//! - `GET /nzb/verdict/:cid` → `Option` remembered verdict +//! - `POST /nzb/verify/:cid` → `Option` verdict, verified if free +//! - `POST /nzb/test-nntp` ← [`NntpTest`] → `{"ok","message"}` +//! - `POST /nzb/restart` → `204`, then the process exits (settings changed) +//! - `POST /nzb/seal/:cid/claim` → [`SealClaim`] / `409` (not `Ready`, or already claimed) +//! - `POST /nzb/seal/:cid/promote`→ [`SealPrepared`] (publish into `cache/` + the hole/magic gates) +//! - `POST /nzb/seal/:cid/done` ← [`SealDone`] +//! - `POST /nzb/seal/:cid/abort` → `204` +//! +//! `/raw/:cid` may answer `409` with header [`HDR_REDIRECT_CID`] — "these bytes +//! now live under that cid (an IPFS root) and my copy is gone": the hull +//! re-dispatches to it, exactly where the monolith fell back to the blockstore. + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; + +use super::cid::MsCid; + +/// `409` + this header on `/raw`: re-dispatch the request to this cid. +pub const HDR_REDIRECT_CID: &str = "x-metamesh-redirect-cid"; + +/// `GET /nzb/capability`. +#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)] +pub struct Capability { + /// The plugin holds an NNTP pool (host+user+pass configured). + pub nntp: bool, + /// The NNTP provider host (never the account), when credentialed. + #[serde(default)] + pub provider: Option, +} + +/// `POST /nzb/test-nntp` — a candidate connection, already merged with the +/// stored secret by the hull (which owns `settings.json`). +#[derive(Serialize, Deserialize, Clone, Debug, Default)] +pub struct NntpTest { + pub host: String, + pub port: u16, + pub user: String, + pub pass: String, + pub tls: bool, +} + +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct SealClaim { + pub path: String, + pub size: u64, +} + +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct SealPrepared { + /// The published file, under `cache//`. + pub path: String, + /// Its name inside the container. + pub rel: String, + pub size: u64, +} + +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct SealDone { + pub root: String, + pub size: u64, + pub path: String, +} + +/// Usenet (NZB) retrieval settings — the `usenet` section of the hull's +/// `settings.json`. The plugin reads the file read-only at boot (the hull owns +/// it and restarts the plugin when it changes). +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(default)] +pub struct UsenetSettings { + pub nntp_host: Option, + pub nntp_port: Option, + pub nntp_user: Option, + /// Secret. + pub nntp_pass: Option, + pub nntp_tls: bool, + pub nntp_connections: u32, +} + +impl Default for UsenetSettings { + fn default() -> Self { + Self { + nntp_host: None, + nntp_port: None, + nntp_user: None, + nntp_pass: None, + nntp_tls: true, + nntp_connections: 20, + } + } +} + +fn env_nonempty(key: &str) -> Option { + std::env::var(key) + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) +} + +impl UsenetSettings { + /// DEPRECATED seed path: the legacy `META_SHARE_NNTP_*` env vars. Only + /// consulted when `settings.json` doesn't exist yet. + pub fn from_env() -> Self { + let mut s = Self::default(); + if let Some(v) = env_nonempty("META_SHARE_NNTP_HOST") { + s.nntp_host = Some(v); + } + if let Some(v) = env_nonempty("META_SHARE_NNTP_PORT").and_then(|v| v.parse().ok()) { + s.nntp_port = Some(v); + } + if let Some(v) = env_nonempty("META_SHARE_NNTP_USER") { + s.nntp_user = Some(v); + } + if let Some(v) = env_nonempty("META_SHARE_NNTP_PASS") { + s.nntp_pass = Some(v); + } + if let Ok(v) = std::env::var("META_SHARE_NNTP_TLS") { + s.nntp_tls = !matches!(v.trim(), "0" | "false" | "no"); + } + if let Some(v) = env_nonempty("META_SHARE_NNTP_CONNECTIONS").and_then(|v| v.parse().ok()) { + s.nntp_connections = v; + } + s.nntp_connections = s.nntp_connections.clamp(1, 100); + s + } + + /// The `usenet` section of `/settings.json`, or the legacy env + /// seed when the file is absent or unparseable. + pub fn load(config_dir: &std::path::Path) -> Self { + #[derive(Deserialize, Default)] + #[serde(default)] + struct File { + usenet: UsenetSettings, + } + match std::fs::read(config_dir.join("settings.json")) { + Ok(bytes) => match serde_json::from_slice::(&bytes) { + Ok(f) => f.usenet, + Err(_) => Self::from_env(), + }, + Err(_) => Self::from_env(), + } + } +} + +/// MetaMesh-private multicodec for a self-describing Newznab **release** locator +/// (`nzb-release`). Mirrors `meta-feeder-sdk`'s `hash::NZB_RELEASE_CODEC`. The +/// cid's identity multihash embeds `{host, id}` directly, so the feeder emits it +/// with **no `.nzb` download** and the credentialed peer decodes the locator +/// straight from the cid at playback — no KV side-table. +pub const NZB_RELEASE_CODEC: u64 = 0x1005; + +/// MetaMesh-private multicodec for a **self-scanned Usenet posting** +/// (`nzb-posting`). Mirrors `meta-feeder-sdk`'s `hash::NZB_POSTING_CODEC`. +/// +/// The crucial difference from [`NZB_RELEASE_CODEC`]: this is a **digest over +/// the article Message-ID set**, not a locator. It embeds no indexer host, so +/// **any peer with a plain NNTP provider can redeem it — no indexer credential**. +/// That portability is the entire point of scanning Usenet ourselves. +/// +/// Two consequences the code must respect: +/// - It is **not reversible**. There is no `decode_*` for it. The `.nzb` +/// manifest arrives out of band, as an ordinary sha2-256 cid named by the +/// record's `manifest` field (see the hull's manifest resolution). +/// - It is **not bitswap-fetchable under itself**. Bitswap derives a block's +/// cid by hashing the block; this digest is over the id set, so a want for it +/// could never be satisfied. Never offer it to the swarm. +/// +/// See `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md` §5. +pub const NZB_POSTING_CODEC: u64 = 0x1003; + +/// Does this cid address a Usenet **release locator** (codec `0x1005`) — the +/// self-describing `{host, id}` form that must be redeemed against a specific +/// indexer with a matching credential? +/// +/// Narrower than [`is_nzb_cid`] on purpose: only the locator carries a host, so +/// only the locator gates on an indexer key (held by a gateway's feeder). Use this where +/// the *host* matters; use [`is_nzb_cid`] where "should this go to the Usenet +/// transport" matters. +pub fn is_nzb_locator(mscid: &MsCid) -> bool { + mscid.codec() == NZB_RELEASE_CODEC +} + +/// Does this cid address a **self-scanned Usenet posting** (codec `0x1003`)? +pub fn is_nzb_posting(mscid: &MsCid) -> bool { + mscid.codec() == NZB_POSTING_CODEC +} + +/// Does this cid go to the **Usenet transport** at all — either form? +/// +/// This is the `/raw` dispatch predicate: both shapes end at +/// the usenet plugin's `/raw`, which resolves a manifest (differently per +/// form — see the hull's manifest resolution), NNTP-fetches the articles, and +/// materialises/streams the media. Everything downstream of manifest +/// resolution is identical for the two. +pub fn is_nzb_cid(mscid: &MsCid) -> bool { + is_nzb_locator(mscid) || is_nzb_posting(mscid) +} + +/// String convenience over [`is_nzb_cid`] — either Usenet form. Used by the +/// playback-focus endpoints, which see the bare CID string. There is no +/// locator-only counterpart: every string-typed call site that used to ask +/// "is this specifically a `0x1005` locator" now needs "is this Usenet at +/// all" — see the six dispatch sites widened alongside `nzb-posting`. +pub fn cid_is_nzb(cid: &str) -> bool { + super::cid::parse_record_cid(cid) + .map(|c| is_nzb_cid(&c)) + .unwrap_or(false) +} + +/// Re-derive an `nzb-posting` digest from a manifest's Message-IDs. +/// +/// ⚠ **MIRRORED LOGIC** — byte-for-byte equivalent to +/// `meta_feeder_sdk::hash::compute_nzb_posting_cid` (the gateway/feeder side +/// mints with it; we verify with it). The two crates cannot share a path-dep, +/// so the normalisation rule is duplicated and must be changed on both sides +/// together. A drift here does not error — it silently makes every self-scanned +/// posting fail verification and become unplayable. +/// +/// The rule: trim, strip a surrounding `<…>`, drop empties, dedupe, sort +/// lexicographically, join with `\n`, SHA-256. Case is preserved (a Message-ID +/// local part is case-sensitive per RFC 5322). +pub fn nzb_posting_digest>(message_ids: &[S]) -> [u8; 32] { + use sha2::{Digest, Sha256}; + + let mut ids: Vec<&str> = message_ids + .iter() + .map(|s| { + let t = s.as_ref().trim(); + t.strip_prefix('<') + .and_then(|r| r.strip_suffix('>')) + .unwrap_or(t) + }) + .filter(|s| !s.is_empty()) + .collect(); + ids.sort_unstable(); + ids.dedup(); + + let mut hasher = Sha256::new(); + for (i, id) in ids.iter().enumerate() { + if i > 0 { + hasher.update(b"\n"); + } + hasher.update(id.as_bytes()); + } + hasher.finalize().into() +} + +/// A decoded self-describing `nzb-release` locator: the Newznab **API base**, +/// recovered from the cid's identity-multihash digest. The release id stays in +/// the cid — only the feeder that redeems it reads it. +pub struct NzbLocator { + /// The indexer's Newznab API base, **scheme-less, authority + optional + /// path** (`api.nzb.life`, `api.nzbgeek.info/api`). Stamped by the feeder + /// from the indexer aggregator's own configured `baseUrl` — i.e. the URL + /// that aggregator itself queries, so it is redeemable by construction. + /// + /// ⚠ This used to be a bare `host` taken from the release's `` + /// **permalink**, which is only the API host by coincidence. nzb.life + /// serves both from `api.nzb.life` so it worked; nzbgeek's permalinks are on + /// `nzbgeek.info` while its API is on `api.nzbgeek.info/api`, so every + /// nzbgeek release was unredeemable — `https://nzbgeek.info/api?t=get` + /// answers 302→HTML and the grab died in `nzb xml parse`. + /// + /// Because it may carry a path it is **not** a matching key: use + /// [`authority()`](Self::authority) for that. + pub api_base: String, +} + +impl NzbLocator { + /// The host to match against the gateways' redeem claims — the authority + /// alone, path stripped. + /// + /// The base may carry a path (`api.nzbgeek.info/api`) that a claim never + /// does: operators configure, and feeders advertise, plain hosts. + pub fn authority(&self) -> &str { + match self.api_base.split_once('/') { + Some((authority, _path)) => authority, + None => &self.api_base, + } + } +} + +/// Normalise an indexer host for matching: drop any `http(s)://` scheme, trim a +/// trailing slash and surrounding whitespace, lowercase. A locator's +/// [`NzbLocator::authority`] passes through this before it is compared with the +/// hosts a gateway advertises, so a claim for `https://api.example.com/` and one +/// for `api.example.com` match the same releases. +pub fn normalize_indexer_host(raw: &str) -> String { + raw.trim() + .trim_start_matches("https://") + .trim_start_matches("http://") + .trim_end_matches('/') + .to_ascii_lowercase() +} + +/// Decode a self-describing `nzb-release` cid (codec `0x1005`) into its +/// API base. The identity-multihash digest is +/// `varint(base_len) ‖ api_base ‖ id_bytes` — the exact inverse of +/// meta-feeder-sdk's `hash::compute_nzb_release_cid`; `id_bytes` is the +/// hex-decoded release id, which only the redeeming feeder reads. Errors +/// (mapped to 4xx by the caller) on a wrong codec or a malformed digest. +/// +/// The wire shape is unchanged from when this field held a bare host — only the +/// *meaning* of those bytes widened to "authority + optional path", so old cids +/// still decode (they just name an unredeemable host; see [`NzbLocator`]). +pub fn decode_nzb_release_cid(mscid: &MsCid) -> Result { + anyhow::ensure!( + mscid.codec() == NZB_RELEASE_CODEC, + "not an nzb-release cid (codec 0x{:x}, want 0x{:x})", + mscid.codec(), + NZB_RELEASE_CODEC + ); + let digest = mscid.hash().digest(); + let (base_len, consumed) = + super::cid::read_uvarint(digest).context("nzb-release digest: base-length varint")?; + let base_len = usize::try_from(base_len) + .map_err(|_| anyhow::anyhow!("nzb-release base length {base_len} overflows usize"))?; + let rest = &digest[consumed..]; + anyhow::ensure!( + rest.len() > base_len, + "nzb-release digest truncated: base_len={base_len}, only {} byte(s) left (need base + id)", + rest.len() + ); + let base_bytes = &rest[..base_len]; + let api_base = std::str::from_utf8(base_bytes) + .context("nzb-release api base is not valid utf-8")? + .to_string(); + Ok(NzbLocator { api_base }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::cid::parse_record_cid; + + #[test] + fn settings_load_reads_the_usenet_section_and_ignores_the_rest() { + let d = std::env::temp_dir().join(format!("nzb-settings-{}", std::process::id())); + std::fs::create_dir_all(&d).unwrap(); + std::fs::write( + d.join("settings.json"), + r#"{"network":{"meta_core_url":"x"},"usenet":{"nntp_host":"news.example","nntp_user":"u","nntp_pass":"p","nntp_tls":true,"nntp_connections":7,"indexers":[1]}}"#, + ) + .unwrap(); + let s = UsenetSettings::load(&d); + assert_eq!(s.nntp_host.as_deref(), Some("news.example")); + assert_eq!(s.nntp_connections, 7); + let _ = std::fs::remove_dir_all(&d); + } + + + /// RFC 4648 base32 lowercase, no padding (multibase `b` body). + fn base32_lower(input: &[u8]) -> String { + const ALPHABET: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567"; + let mut out = String::new(); + let (mut buffer, mut bits): (u64, u32) = (0, 0); + for &b in input { + buffer = (buffer << 8) | b as u64; + bits += 8; + while bits >= 5 { + bits -= 5; + out.push(ALPHABET[((buffer >> bits) & 0x1F) as usize] as char); + } + } + if bits > 0 { + out.push(ALPHABET[((buffer << (5 - bits)) & 0x1F) as usize] as char); + } + out + } + + fn varint(mut v: u64, out: &mut Vec) { + while v >= 0x80 { + out.push((v as u8) | 0x80); + v >>= 7; + } + out.push(v as u8); + } + + /// Build a self-describing nzb-release cid exactly as meta-feeder-sdk's + /// `compute_nzb_release_cid` does: identity multihash over + /// `varint(host_len) ‖ host ‖ id_bytes`. + fn make_nzb_cid(host: &str, id_bytes: &[u8]) -> String { + let mut digest = Vec::new(); + varint(host.len() as u64, &mut digest); + digest.extend_from_slice(host.as_bytes()); + digest.extend_from_slice(id_bytes); + let mut wire = vec![0x01u8]; + varint(NZB_RELEASE_CODEC, &mut wire); + wire.push(0x00); // identity multihash + varint(digest.len() as u64, &mut wire); + wire.extend_from_slice(&digest); + format!("b{}", base32_lower(&wire)) + } + + #[test] + fn decode_round_trips_the_base() { + let cid = make_nzb_cid("api.example.com", &[0x48, 0x5c, 0x52, 0xf0]); + let mscid = parse_record_cid(&cid).expect("parse"); + assert!(is_nzb_locator(&mscid)); + let loc = decode_nzb_release_cid(&mscid).expect("decode"); + assert_eq!(loc.api_base, "api.example.com"); + // No path ⇒ authority is the whole thing. + assert_eq!(loc.authority(), "api.example.com"); + } + + /// The nzbgeek shape, and the whole reason `authority()` exists: its API + /// lives at `https://api.nzbgeek.info/api` while a feeder claims the plain + /// host. The base keeps the path; every match must drop it. + #[test] + fn a_base_with_a_path_matches_on_its_authority() { + let cid = make_nzb_cid("api.nzbgeek.info/api", &[0x19, 0xfe, 0x38, 0xe4]); + let mscid = parse_record_cid(&cid).expect("parse"); + let loc = decode_nzb_release_cid(&mscid).expect("decode"); + assert_eq!(loc.api_base, "api.nzbgeek.info/api"); + assert_eq!( + loc.authority(), + "api.nzbgeek.info", + "can_redeem / probe / redeem claims all compare on the authority" + ); + } + + /// A port is part of the authority, not a path — it must survive the split. + #[test] + fn a_port_stays_in_the_authority() { + let cid = make_nzb_cid("indexer.local:5076/api", &[0x01, 0x02]); + let mscid = parse_record_cid(&cid).expect("parse"); + let loc = decode_nzb_release_cid(&mscid).expect("decode"); + assert_eq!(loc.authority(), "indexer.local:5076"); + } + + #[test] + fn decode_rejects_wrong_codec() { + // codec 0x1001 (btih-v1-file), not an nzb-release locator. + let mut wire = vec![0x01u8]; + varint(0x1001, &mut wire); + wire.extend_from_slice(&[0x00, 0x02, 0xAB, 0xCD]); // identity mh, 2-byte digest + let cid = format!("b{}", base32_lower(&wire)); + let mscid = parse_record_cid(&cid).expect("parse"); + assert!(!is_nzb_locator(&mscid)); + assert!(decode_nzb_release_cid(&mscid).is_err()); + } + + #[test] + fn decode_rejects_truncated_digest() { + // host_len claims the whole digest, leaving no id bytes. + let cid = make_nzb_cid("api.example.com", &[]); + let mscid = parse_record_cid(&cid).expect("parse"); + assert!(decode_nzb_release_cid(&mscid).is_err()); + } + + /// Build an nzb-posting cid exactly as meta-feeder-sdk's + /// `compute_nzb_posting_cid` does: codec 0x1003 + sha2-256 over the + /// normalised, sorted, newline-joined Message-ID set. + fn make_posting_cid(message_ids: &[&str]) -> String { + let digest = nzb_posting_digest(message_ids); + let mut wire = vec![0x01u8]; + varint(NZB_POSTING_CODEC, &mut wire); + wire.push(0x12); // sha2-256 + wire.push(0x20); // 32 bytes + wire.extend_from_slice(&digest); + format!("b{}", base32_lower(&wire)) + } + + /// ⚠ THE MIRROR TEST. `nzb_posting_digest` here must agree byte-for-byte + /// with `meta_feeder_sdk::hash::compute_nzb_posting_cid`'s digest — the + /// gateway mints with one, we verify with the other, and a drift silently + /// makes every self-scanned posting unplayable rather than erroring. + /// + /// This vector is the same one pinned on the SDK side + /// (`hash::tests::nzb_posting_cid_wire_shape`): a single id `a@x.com` + /// digests to plain `sha256("a@x.com")`. + #[test] + fn posting_digest_matches_the_sdk_mint() { + use sha2::{Digest, Sha256}; + let expect: [u8; 32] = Sha256::digest(b"a@x.com").into(); + assert_eq!(nzb_posting_digest(&["a@x.com"]), expect); + } + + /// The normalisation rule, mirrored from the SDK: order-independent, + /// bracket/blank/dupe-insensitive, case-SENSITIVE. + #[test] + fn posting_digest_normalisation_matches_the_mint() { + // Order does not matter — this is why we hash a sorted set and never + // the raw `.nzb` bytes (the retired 0x1004's mistake). + assert_eq!( + nzb_posting_digest(&["c@x.com", "a@x.com", "b@x.com"]), + nzb_posting_digest(&["a@x.com", "b@x.com", "c@x.com"]) + ); + // Angle brackets, whitespace, empties and duplicates all normalise away. + assert_eq!( + nzb_posting_digest(&["a@x.com", "b@x.com"]), + nzb_posting_digest(&[" ", "b@x.com", "", " ", "a@x.com"]) + ); + // Case is preserved (RFC 5322: the local part is case-sensitive). + assert_ne!( + nzb_posting_digest(&["Abc@x.com"]), + nzb_posting_digest(&["abc@x.com"]) + ); + // A different article set is a different posting. + assert_ne!( + nzb_posting_digest(&["a@x.com", "b@x.com"]), + nzb_posting_digest(&["a@x.com", "c@x.com"]) + ); + } + + /// Both Usenet forms route to the Usenet transport, but only the `0x1005` + /// locator is host-bound. Keeping the two predicates distinct is what lets + /// `can_redeem` skip the credential check for a posting. + #[test] + fn posting_and_locator_are_both_nzb_but_only_one_is_a_locator() { + let posting = parse_record_cid(&make_posting_cid(&["a@x.com"])).expect("parse posting"); + assert!(is_nzb_posting(&posting)); + assert!(is_nzb_cid(&posting), "postings go to the Usenet transport"); + assert!( + !is_nzb_locator(&posting), + "a posting names no indexer host, so it must NOT be treated as a locator (that would send it down the credential-gated grab path)" + ); + + let locator = parse_record_cid(&make_nzb_cid("api.example.com", &[0xab, 0xcd])) + .expect("parse locator"); + assert!(is_nzb_locator(&locator)); + assert!(is_nzb_cid(&locator)); + assert!(!is_nzb_posting(&locator)); + } + + /// A posting cid must never decode as a release locator — the digest is not + /// a `{host, id}` payload, and misreading it would produce a garbage host. + #[test] + fn posting_cid_does_not_decode_as_a_locator() { + let posting = parse_record_cid(&make_posting_cid(&["a@x.com", "b@x.com"])) + .expect("parse posting"); + assert!(decode_nzb_release_cid(&posting).is_err()); + } + + /// The string convenience wrappers agree with the parsed forms, including + /// on garbage input (which must be `false`, never a panic). + #[test] + fn cid_string_helpers_agree() { + let posting = make_posting_cid(&["a@x.com"]); + assert!(cid_is_nzb(&posting)); + + let locator = make_nzb_cid("api.example.com", &[0x01]); + assert!(cid_is_nzb(&locator)); + + assert!(!cid_is_nzb("not-a-cid")); + } +} diff --git a/src/transport/plugin.rs b/src/transport/plugin.rs new file mode 100644 index 0000000..e5d14b4 --- /dev/null +++ b/src/transport/plugin.rs @@ -0,0 +1,50 @@ +//! The plugin side of the contract: what a transport implementation provides. + +use std::sync::Arc; + +use async_trait::async_trait; +use axum::http::HeaderMap; +use axum::response::Response; +use axum::Router; + +use super::dto::{Deleted, Health, Job, Manifest, ReconcileReport, ReconcileRequest}; +use super::error::ApiError; +use super::focus::FocusView; + +/// One transport implementation (BitTorrent, Usenet, IPFS, or a competing +/// implementation of one of them). [`serve_transport`](super::serve::serve_transport) +/// wraps it in the common routes; [`extra_routes`](Self::extra_routes) adds its +/// protocol-specific ones. +#[async_trait] +pub trait TransportPlugin: Send + Sync + 'static { + fn manifest(&self) -> Manifest; + + async fn health(&self) -> Health { + Health::ok() + } + + /// `GET /raw/:cid`. Must answer exactly what the in-process tier answered + /// in the monolith (status, `Content-*`, `Retry-After`, error body) — the + /// hull relays it to meta-share's public API untouched. Must **not** meter + /// its own egress: the hull does, on relay. + async fn raw(&self, cid: String, headers: HeaderMap) -> Response; + + /// `GET /jobs`. + async fn jobs(&self) -> Result, ApiError>; + + /// `DELETE /jobs/:cid?keep=…`. `keep` lists sibling cids of the same unit + /// the hull still wants. + async fn delete(&self, cid: String, keep: Vec) -> Result; + + /// `POST /reconcile`. + async fn reconcile(&self, req: ReconcileRequest) -> Result; + + /// The view `PUT /focus` writes into. + fn focus(&self) -> &Arc; + + /// Protocol-specific routes, merged into the common router. Paths must not + /// collide with the common ones. + fn extra_routes(self: Arc) -> Router { + Router::new() + } +} diff --git a/src/transport/range.rs b/src/transport/range.rs new file mode 100644 index 0000000..51f6eeb --- /dev/null +++ b/src/transport/range.rs @@ -0,0 +1,277 @@ +//! HTTP `Range:` request-header parser. +//! +//! Pure helper: takes an inbound `Range:` header value plus the total +//! resource size in bytes, returns a [`RangeOutcome`] describing how the +//! caller should respond. +//! +//! Shared by the meta-share hull and every transport plugin so the +//! `