From 7f8b8170b927ab6b1773d6c97afe52ba7d8469cd Mon Sep 17 00:00:00 2001 From: Worph Date: Mon, 5 Oct 2026 12:06:09 +0000 Subject: [PATCH] meta-transport-ipfs 1.3.0: resolve pointers (GET /ipfs-tier/resolve: stored copy, else HTTP redeem at the fastest claiming gateway; config-pinned local gateways), SDK v1.8.0 --- CLAUDE.md | 5 + Cargo.lock | 2 +- Cargo.toml | 6 +- src/api/mod.rs | 2 + src/gateways.rs | 205 ++++++++++++++++++++++++++++ src/main.rs | 4 + src/plane.rs | 6 + src/plugin.rs | 43 +++--- src/resolve.rs | 349 ++++++++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 598 insertions(+), 24 deletions(-) create mode 100644 src/gateways.rs create mode 100644 src/resolve.rs diff --git a/CLAUDE.md b/CLAUDE.md index 4b8c5d5..90b731e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -7,6 +7,11 @@ Behaviour must stay identical to the monolith's swarm / blockstore / ingress. 1. **The swarm is transport-only**: identify/kad/mdns/bitswap. No kamilata, no gossipsub, no request_response — search is meta-search's. + Resolving a **pointer** (`GET /ipfs-tier/resolve/:cid`, `resolve.rs`) is not a + swarm protocol: it redeems over plain HTTP at a gateway's base URL, chosen by + `gateways.rs` from the config-pinned local gateways plus the swarm directory, + fastest first. Never move it onto libp2p, and never let bitswap answer a + pointer — its multihash is `identity`, there is nothing to verify. 2. **`MAX_MULTIHASH_SIZE = 64`** must match meta-gateway and meta-search. 3. **`core2` is patched** in `Cargo.toml` (every libp2p-0.56 bitswap impl needs the yanked `^0.4.0`); mirror changes in meta-gateway. diff --git a/Cargo.lock b/Cargo.lock index 2f3ae4d..eccdc93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1990,7 +1990,7 @@ dependencies = [ [[package]] name = "meta-transport-ipfs" -version = "1.2.2" +version = "1.3.0" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index ccc3c8c..cb5a4b4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-transport-ipfs" -version = "1.2.2" +version = "1.3.0" edition = "2021" rust-version = "1.85" license = "MIT" @@ -19,7 +19,7 @@ path = "src/main.rs" # The transport-plugin contract. Pinned by TAG, never by path — a path dep # resolves outside this repo, which is the Docker build context. See the # Dockerfile for building against an unreleased SDK. -meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.7.0", features = ["transport"] } +meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.8.0", features = ["transport"] } # libp2p — transport only: identify (address exchange, load-bearing for bitswap # reachability), kad (DHT provider records), mdns (LAN discovery) + bitswap. @@ -56,7 +56,7 @@ multihash-codetable = { version = "0.1", features = ["sha2", "sha3"] } [dev-dependencies] tempfile = "3" -meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.7.0", features = ["testkit"] } +meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.8.0", features = ["testkit"] } # --------------------------------------------------------------------------- # Workaround for the yanked transitive `core2` (all versions). `multihash- diff --git a/src/api/mod.rs b/src/api/mod.rs index 19109f6..df4e7cd 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -38,6 +38,8 @@ pub struct AppState { /// default = public mode). Decided here, where `PUBLIC_ADDR` is known; the /// hull always asks and this gates. pub seed_dht_provide: bool, + /// Gateways that can redeem pointers, fastest first (`/ipfs-tier/resolve`). + pub gateways: Arc, } /// "This peer now seeds `cid`" — the hull writes the seed row (and decides on diff --git a/src/gateways.rs b/src/gateways.rs new file mode 100644 index 0000000..c2da21a --- /dev/null +++ b/src/gateways.rs @@ -0,0 +1,205 @@ +//! Which gateways can redeem a pointer, fastest first. +//! +//! Two sources, merged by base URL: +//! +//! - **Pinned** gateways from config (`gateway_urls`, seeded from +//! `META_SHARE_GATEWAY_URLS`) — the local one. Their redeem claims come +//! straight from `GET {url}/api/gateway/plugins`, so they never depend on the +//! libp2p link (watch.nsl.sh 2026-10-04: a lost swarm link left every NZB +//! "unredeemable" though the gateway sat on the same Docker network). +//! - **Swarm** gateways from the peer directory (identify `gateways=` + a caps +//! fetch) — remote ones. +//! +//! Order is by the round-trip of the last caps fetch, which the refresh task +//! times for every known gateway. Nothing ranks the local gateway first by rule: +//! it is simply the closest, so it wins on latency. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use tracing::debug; + +use crate::api::AppState; +use crate::gateway_discovery::{fetch_gateway_caps, RedeemClaim}; +use crate::swarm::PeerDirectory; + +/// How often pinned gateways' claims and every gateway's latency are refreshed. +const REFRESH_EVERY: Duration = Duration::from_secs(30); +/// A pinned gateway's claims are trusted this long after the last fetch. +const CLAIMS_TTL: Duration = Duration::from_secs(600); +const FETCH_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Default)] +pub struct Gateways { + pinned: Vec, + /// Pinned base URL → (claims, fetched at). + pinned_claims: Mutex, Instant)>>, + /// Base URL → last caps-fetch round-trip. + rtt: Mutex>, +} + +fn norm(url: &str) -> String { + url.trim().trim_end_matches('/').to_string() +} + +impl Gateways { + pub fn new(pinned: impl IntoIterator) -> Self { + let mut list: Vec = Vec::new(); + for u in pinned.into_iter().map(|u| norm(&u)).filter(|u| !u.is_empty()) { + if !list.contains(&u) { + list.push(u); + } + } + Self { pinned: list, ..Self::default() } + } + + /// From `META_SHARE_GATEWAY_URLS` (comma-separated), which the config plane + /// overlays at boot. + pub fn from_env() -> Self { + Self::new( + std::env::var("META_SHARE_GATEWAY_URLS") + .unwrap_or_default() + .split(',') + .map(str::to_string), + ) + } + + pub fn pinned(&self) -> &[String] { + &self.pinned + } + + fn record(&self, url: &str, claims: Option>, rtt: Duration) { + self.rtt.lock().unwrap().insert(norm(url), rtt); + if let Some(c) = claims { + self.pinned_claims.lock().unwrap().insert(norm(url), (c, Instant::now())); + } + } + + /// Gateways whose claim covers (`codec`, `key`), fastest first. A gateway + /// with no measured latency yet goes after every measured one. + pub fn candidates(&self, dir: &PeerDirectory, codec: &str, key: &str) -> Vec { + let mut out: Vec = Vec::new(); + { + let claims = self.pinned_claims.lock().unwrap(); + for url in &self.pinned { + let covered = claims.get(url).is_some_and(|(c, at)| { + at.elapsed() <= CLAIMS_TTL && c.iter().any(|c| c.covers(codec, key)) + }); + if covered { + out.push(url.clone()); + } + } + } + for url in dir.gateways_redeeming(codec, key) { + let url = norm(&url); + if !out.contains(&url) { + out.push(url); + } + } + self.by_latency(out) + } + + fn by_latency(&self, mut urls: Vec) -> Vec { + let rtt = self.rtt.lock().unwrap(); + // Stable sort: unmeasured gateways keep their order (pinned first). + urls.sort_by_key(|u| rtt.get(u).copied().unwrap_or(Duration::MAX)); + urls + } + + /// Is any gateway known at all — a pinned one that answered, or a swarm one? + /// Tells "nobody claims this" apart from "discovery hasn't finished". + pub fn knows_any(&self, dir: &PeerDirectory) -> bool { + let pinned_answered = self + .pinned_claims + .lock() + .unwrap() + .values() + .any(|(_, at)| at.elapsed() <= CLAIMS_TTL); + pinned_answered || dir.knows_a_gateway() + } +} + +/// Refresh pinned claims and every gateway's latency, now and every +/// [`REFRESH_EVERY`]. +pub fn spawn_refresh(state: Arc) { + tokio::spawn(async move { + loop { + refresh(&state).await; + tokio::time::sleep(REFRESH_EVERY).await; + } + }); +} + +async fn refresh(state: &AppState) { + let gws = &state.gateways; + let mut urls: Vec<(String, bool)> = gws.pinned().iter().map(|u| (u.clone(), true)).collect(); + for p in state.peer_directory.snapshot() { + if p.redeems.is_empty() { + continue; + } + if let Some(u) = p.base_url.as_deref().map(norm) { + if !urls.iter().any(|(x, _)| *x == u) { + urls.push((u, false)); + } + } + } + for (url, pinned) in urls { + let t0 = Instant::now(); + match fetch_gateway_caps(&state.http, &url, FETCH_TIMEOUT).await { + Ok(caps) => gws.record(&url, pinned.then(|| caps.redeem_claims()), t0.elapsed()), + Err(e) => debug!(gateway = %url, error = %e, "gateways: caps fetch failed"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::gateway_discovery::REDEEM_NZB_RELEASE; + + fn claim(host: &str) -> RedeemClaim { + serde_json::from_value(serde_json::json!({ + "codec": REDEEM_NZB_RELEASE, "field": "manifest", "hosts": [host], "sources": [] + })) + .unwrap() + } + + #[test] + fn a_pinned_gateway_counts_only_for_what_it_claims() { + let g = Gateways::new(["http://local:3000/".to_string()]); + let dir = PeerDirectory::new(); + assert!(g.candidates(&dir, REDEEM_NZB_RELEASE, "api.nzbgeek.info").is_empty()); + assert!(!g.knows_any(&dir)); + g.record("http://local:3000", Some(vec![claim("api.nzbgeek.info")]), Duration::from_millis(3)); + assert_eq!( + g.candidates(&dir, REDEEM_NZB_RELEASE, "api.nzbgeek.info"), + vec!["http://local:3000".to_string()] + ); + assert!(g.candidates(&dir, REDEEM_NZB_RELEASE, "api.nzb.life").is_empty()); + assert!(g.knows_any(&dir)); + } + + #[test] + fn the_fastest_gateway_comes_first() { + let g = Gateways::new(["http://a".to_string(), "http://b".to_string(), "http://c".to_string()]); + g.record("http://a", Some(vec![claim("h")]), Duration::from_millis(80)); + g.record("http://b", Some(vec![claim("h")]), Duration::from_millis(2)); + // c claims but was never timed separately from its claims fetch: still timed. + g.record("http://c", Some(vec![claim("h")]), Duration::from_millis(20)); + assert_eq!( + g.candidates(&PeerDirectory::new(), REDEEM_NZB_RELEASE, "h"), + vec!["http://b", "http://c", "http://a"] + ); + } + + #[test] + fn unmeasured_gateways_go_last_in_their_given_order() { + let g = Gateways::new([]); + g.rtt.lock().unwrap().insert("http://timed".into(), Duration::from_millis(500)); + assert_eq!( + g.by_latency(vec!["http://x".into(), "http://timed".into(), "http://y".into()]), + vec!["http://timed", "http://x", "http://y"] + ); + } +} diff --git a/src/main.rs b/src/main.rs index 6182e2a..087921a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,12 +18,14 @@ mod blockstore; mod config; mod filestore; mod gateway_discovery; +mod gateways; mod ingress; mod ingress_commit; mod ipfs_chunk; mod material; mod plane; mod plugin; +mod resolve; mod settings; mod share; mod store; @@ -166,7 +168,9 @@ async fn main() -> Result<()> { meta_core_url: cfg.meta_core_url.clone(), webdav_url_cache, seed_dht_provide: cfg.seed_dht_provide, + gateways: Arc::new(gateways::Gateways::from_env()), }); + gateways::spawn_refresh(Arc::clone(&state)); // Bitswap ingress: sweep partial fetches a previous process left in `tmp/`, // then promote/fill on a timer. diff --git a/src/plane.rs b/src/plane.rs index 099a28b..b169560 100644 --- a/src/plane.rs +++ b/src/plane.rs @@ -24,6 +24,8 @@ pub fn schema() -> ConfigSchema { .with_help("Multiaddrs used to join the public IPFS DHT. Empty = the app's default list. `none` disables the DHT."), ConfigField::list("bootstrap_peers", "Pinned peers") .with_help("Multiaddrs this node keeps dialled (e.g. a known gateway)."), + ConfigField::list("gateway_urls", "Local gateways") + .with_help("Base URLs of gateways reached over HTTP without the swarm (e.g. http://metagateway-app:3000). They resolve NZB and subtitle pointers even when the libp2p link is down."), ConfigField::bool("enable_mdns", "LAN discovery (mDNS)"), ConfigField::bool("seed_dht_provide", "Announce seeds on the DHT") .with_help("Publish provider records for what this node seeds, so remote peers can find it."), @@ -68,6 +70,7 @@ pub fn seed() -> Value { "kad_namespace": env("KAD_NAMESPACE").unwrap_or_else(|| DEFAULT_NAMESPACE.to_string()), "kad_bootstrap_peers": env_list("KAD_BOOTSTRAP_PEERS"), "bootstrap_peers": env_list("BOOTSTRAP_PEERS"), + "gateway_urls": env_list("META_SHARE_GATEWAY_URLS"), "enable_mdns": env_flag("ENABLE_MDNS", true), "seed_dht_provide": env_flag("META_SHARE_SEED_DHT_PROVIDE", public), "max_conns": env_u64("META_SHARE_MAX_CONNS", 128), @@ -93,6 +96,9 @@ pub fn env_overrides(v: &Value) -> Vec<(&'static str, String)> { if let Some(l) = read::list(v, "bootstrap_peers").filter(|l| !l.is_empty()) { out.push(("BOOTSTRAP_PEERS", l.join(","))); } + if let Some(l) = read::list(v, "gateway_urls").filter(|l| !l.is_empty()) { + out.push(("META_SHARE_GATEWAY_URLS", l.join(","))); + } if let Some(b) = read::flag(v, "enable_mdns") { out.push(("ENABLE_MDNS", bool_s(b))); } diff --git a/src/plugin.rs b/src/plugin.rs index 3caada1..42ccf8a 100644 --- a/src/plugin.rs +++ b/src/plugin.rs @@ -119,6 +119,7 @@ impl TransportPlugin for IpfsPlugin { fn extra_routes(self: Arc) -> Router { let gateway = Router::new() .route("/ipfs/:cid", get(crate::api::ipfs_gateway::get_ipfs)) + .route("/ipfs-tier/resolve/:cid", get(crate::resolve::resolve)) .with_state(Arc::clone(&self.state)); Router::new() .route("/ipfs-tier/peers", get(peers)) @@ -216,26 +217,28 @@ async fn cat(State(p): S, Path(cid): Path, Query(q): Query) -> Err(e) => return bad_cid(e), }; let max = q.max.unwrap_or_else(crate::api::ipfs_walker::max_body_bytes); - let block = match crate::api::ipfs_walker::get_block(&p.state, &mscid, Lane::Focused).await { - Ok(b) => b, - Err(e) => return ApiError::Upstream(format!("{e:#}")).into_response(), - }; - let bytes = if mscid.codec() == crate::api::ipfs_walker::DAGPB_CODEC { - match crate::api::ipfs_walker::assemble_dagpb_file(&p.state, &block, max, Lane::Focused).await { - Ok(b) => b, - Err(e) => return ApiError::Upstream(format!("{e:#}")).into_response(), - } - } else { - if block.len() > max { - return ApiError::Upstream(format!( - "{cid} is {} bytes, over the {max}-byte ceiling", - block.len() - )) - .into_response(); - } - block - }; - (StatusCode::OK, bytes).into_response() + match read_object(&p.state, &mscid, max).await { + Ok(bytes) => (StatusCode::OK, bytes).into_response(), + Err(e) => ApiError::Upstream(format!("{e:#}")).into_response(), + } +} + +/// A block, or a dag-pb file assembled, up to `max` bytes — local, else bitswap. +pub(crate) async fn read_object( + state: &AppState, + mscid: &crate::blockstore::MsCid, + max: usize, +) -> anyhow::Result> { + let block = crate::api::ipfs_walker::get_block(state, mscid, Lane::Focused).await?; + if mscid.codec() == crate::api::ipfs_walker::DAGPB_CODEC { + return crate::api::ipfs_walker::assemble_dagpb_file(state, &block, max, Lane::Focused).await; + } + anyhow::ensure!( + block.len() <= max, + "{mscid} is {} bytes, over the {max}-byte ceiling", + block.len() + ); + Ok(block) } #[derive(Deserialize)] diff --git a/src/resolve.rs b/src/resolve.rs new file mode 100644 index 0000000..645817e --- /dev/null +++ b/src/resolve.rs @@ -0,0 +1,349 @@ +//! `GET /ipfs-tier/resolve/:cid?hint=` — a [pointer](meta_feeder_sdk::transport::pointer)'s bytes. +//! +//! A pointer (an `nzb-release`'s `.nzb`, a `provider-file`'s subtitle) costs +//! quota on an account this peer does not hold, so only a gateway whose feeder +//! holds the key can answer it — never bitswap: the cid is an address, not a +//! hash of the bytes. +//! +//! 1. **Stored.** `hint` is the content cid the record's pointer field names +//! (the hull reads it). Held locally → served, no network. +//! 2. **Redeemed.** Otherwise `POST {gateway}/api/file/:cid/redeem` at each +//! gateway whose claim covers the pointer, fastest first +//! ([`crate::gateways`]). The gateway names the bytes' content cid; we +//! re-derive it with the kubo-identical chunker and refuse a mismatch, then +//! store the blocks. A repeat redeem is answered from the gateway's own store +//! without spending quota, so a second box pays nothing. +//! +//! The hull owns what follows — the seed row, the permanent copy in meta-core, +//! the record link — so this route only fetches, verifies and stores blocks. +//! Moved here from the hull's `redeem.rs` (meta-share 2.4.0), unchanged in +//! behaviour: ordering, timeout, size cap, verification, error ranking. + +use std::sync::Arc; +use std::time::Duration; + +use axum::extract::{Path, Query, State}; +use axum::http::{header::RETRY_AFTER, HeaderValue, StatusCode}; +use axum::response::{IntoResponse, Response}; +use meta_feeder_sdk::transport::dto::{ + HDR_RESOLVE_ERROR, RESOLVE_NOT_FOUND, RESOLVE_NO_GATEWAY, RESOLVE_QUOTA, RESOLVE_UNCLAIMED, + RESOLVE_UPSTREAM, +}; +use meta_feeder_sdk::transport::hull::{HDR_MANIFEST_CID, HDR_MANIFEST_SOURCE}; +use meta_feeder_sdk::transport::pointer; +use serde::Deserialize; +use tracing::{debug, warn}; + +use crate::api::AppState; +use crate::blockstore::parse_record_cid; +use crate::ipfs_chunk::IpfsBlocks; + +/// Response header naming the redeemed bytes' content cid. +pub const CONTENT_CID_HEADER: &str = "x-metamesh-content-cid"; + +/// One redeem's wall-clock budget: the gateway makes a metered upstream call and +/// a store-back before it answers. +const REDEEM_TIMEOUT: Duration = Duration::from_secs(60); + +/// Ceiling on a redeemed body. A 100k-segment `.nzb` is ~16 MiB; a subtitle is +/// kilobytes. Bounds a broken or hostile gateway. +pub const MAX_REDEEM_BYTES: usize = 32 * 1024 * 1024; + +#[derive(Debug)] +pub enum RedeemError { + /// Every claiming gateway answered `404`. + NotFound, + /// A gateway refused on quota; `retry_after` in seconds. + Quota { retry_after: Option }, + /// Transport failure, unexpected status, or a body that doesn't hash to the + /// cid the gateway named. + Upstream(String), +} + +#[derive(Deserialize)] +pub struct ResolveQuery { + #[serde(default)] + hint: Option, +} + +pub async fn resolve( + State(state): State>, + Path(cid): Path, + Query(q): Query, +) -> Response { + let ptr = match pointer::decode(&cid) { + Some(Ok(p)) => p, + Some(Err(e)) => return refuse(StatusCode::BAD_REQUEST, RESOLVE_NOT_FOUND, None, format!("{e:#}")), + None => { + return refuse( + StatusCode::BAD_REQUEST, + RESOLVE_NOT_FOUND, + None, + format!("`{cid}` is not a pointer cid"), + ) + } + }; + + if let Some(hint) = q.hint.as_deref().filter(|h| !h.is_empty()) { + if let Some(bytes) = stored(&state, hint).await { + debug!(cid, hint, "resolve: served the stored copy"); + return answer(bytes, "store", hint); + } + } + + let Some(key) = ptr.redeem else { + return refuse( + StatusCode::NOT_FOUND, + RESOLVE_NOT_FOUND, + None, + format!("`{cid}` resolves only from a stored copy, and none is held here"), + ); + }; + let targets = state.gateways.candidates(&state.peer_directory, key.codec, &key.key); + if targets.is_empty() { + return if state.gateways.knows_any(&state.peer_directory) { + refuse( + StatusCode::NOT_FOUND, + RESOLVE_UNCLAIMED, + None, + format!("no known gateway redeems {} `{}`", key.codec, key.key), + ) + } else { + refuse( + StatusCode::SERVICE_UNAVAILABLE, + RESOLVE_NO_GATEWAY, + Some(2), + "no gateway reachable yet".into(), + ) + }; + } + + match redeem_from(&state.http, &targets, &cid).await { + Ok((blocks, body)) => { + let root = blocks.root.clone(); + if let Err(e) = crate::blockstore::put_ipfs_blocks(state.block_store.as_ref(), &blocks).await { + // The bytes are verified; failing to keep them only costs a + // repeat redeem, which the gateway answers from its store. + warn!(cid, root, error = %format!("{e:#}"), "resolve: storing the redeemed blocks failed"); + } + debug!(cid, root, size = body.len(), "resolve: redeemed through a gateway"); + answer(body, "redeem", &root) + } + Err(RedeemError::NotFound) => refuse( + StatusCode::NOT_FOUND, + RESOLVE_NOT_FOUND, + None, + "every gateway that claims it answered not found".into(), + ), + Err(RedeemError::Quota { retry_after }) => refuse( + StatusCode::SERVICE_UNAVAILABLE, + RESOLVE_QUOTA, + Some(retry_after.unwrap_or(3600)), + "the provider's quota is spent".into(), + ), + Err(RedeemError::Upstream(e)) => refuse(StatusCode::BAD_GATEWAY, RESOLVE_UPSTREAM, None, e), + } +} + +/// The stored copy `hint` names, when its whole dag is held locally. +async fn stored(state: &AppState, hint: &str) -> Option> { + let mscid = parse_record_cid(hint).ok()?; + if !crate::share::local_dag_complete(state, &mscid).await { + return None; + } + crate::plugin::read_object(state, &mscid, MAX_REDEEM_BYTES).await.ok() +} + +fn answer(bytes: Vec, source: &'static str, root: &str) -> Response { + let mut r = (StatusCode::OK, bytes).into_response(); + r.headers_mut().insert(HDR_MANIFEST_SOURCE, HeaderValue::from_static(source)); + if let Ok(v) = HeaderValue::from_str(root) { + r.headers_mut().insert(HDR_MANIFEST_CID, v); + } + r +} + +fn refuse(status: StatusCode, why: &'static str, retry_after: Option, msg: String) -> Response { + let mut r = (status, msg).into_response(); + r.headers_mut().insert(HDR_RESOLVE_ERROR, HeaderValue::from_static(why)); + if let Some(s) = retry_after { + r.headers_mut().insert(RETRY_AFTER, HeaderValue::from(s)); + } + r +} + +/// Try each target in turn and return the first verified answer. +/// +/// A `404` moves on silently. A quota refusal or an upstream failure also moves +/// on (another gateway may hold its own quota) but is remembered: when nobody +/// delivers, quota outranks an upstream error, which outranks not-found. +pub(crate) async fn redeem_from( + http: &reqwest::Client, + targets: &[String], + cid: &str, +) -> Result<(IpfsBlocks, Vec), RedeemError> { + let mut quota: Option> = None; + let mut upstream: Option = None; + for base in targets { + match redeem_one(http, base, cid).await { + Ok(v) => return Ok(v), + Err(RedeemError::Quota { retry_after }) => { + quota.get_or_insert(retry_after); + } + Err(RedeemError::Upstream(e)) => { + debug!(cid, gateway = %base, error = %e, "redeem: gateway failed; trying the next"); + upstream.get_or_insert(e); + } + Err(RedeemError::NotFound) => {} + } + } + if let Some(retry_after) = quota { + return Err(RedeemError::Quota { retry_after }); + } + if let Some(e) = upstream { + return Err(RedeemError::Upstream(e)); + } + Err(RedeemError::NotFound) +} + +async fn redeem_one( + http: &reqwest::Client, + base: &str, + cid: &str, +) -> Result<(IpfsBlocks, Vec), RedeemError> { + let url = format!("{}/api/file/{}/redeem", base.trim_end_matches('/'), cid); + let resp = http + .post(&url) + .timeout(REDEEM_TIMEOUT) + .send() + .await + .map_err(|e| RedeemError::Upstream(format!("POST {url}: {e}")))?; + match resp.status() { + s if s.is_success() => {} + reqwest::StatusCode::NOT_FOUND => return Err(RedeemError::NotFound), + reqwest::StatusCode::TOO_MANY_REQUESTS => { + let retry_after = resp + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.trim().parse().ok()); + return Err(RedeemError::Quota { retry_after }); + } + s => return Err(RedeemError::Upstream(format!("POST {url} answered {s}"))), + } + let advertised = resp + .headers() + .get(CONTENT_CID_HEADER) + .and_then(|v| v.to_str().ok()) + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()) + .ok_or_else(|| RedeemError::Upstream(format!("POST {url}: no {CONTENT_CID_HEADER} header")))?; + if resp.content_length().is_some_and(|n| n > MAX_REDEEM_BYTES as u64) { + return Err(RedeemError::Upstream(format!( + "POST {url}: body over the {MAX_REDEEM_BYTES}-byte ceiling" + ))); + } + let body = resp + .bytes() + .await + .map_err(|e| RedeemError::Upstream(format!("POST {url}: body: {e}")))? + .to_vec(); + if body.len() > MAX_REDEEM_BYTES { + return Err(RedeemError::Upstream(format!( + "POST {url}: body over the {MAX_REDEEM_BYTES}-byte ceiling" + ))); + } + // Kubo-identical chunking is CPU-bound; keep it off the async runtime. + let (blocks, body) = + tokio::task::spawn_blocking(move || (crate::ipfs_chunk::compute_ipfs_blocks(&body), body)) + .await + .map_err(|e| RedeemError::Upstream(format!("ipfs chunker task: {e}")))?; + if blocks.root != advertised { + // Without this any gateway could answer with unrelated bytes and we + // would store them as the subtitle or the `.nzb`. + return Err(RedeemError::Upstream(format!( + "POST {url}: body hashes to {} but the gateway named {advertised}", + blocks.root + ))); + } + Ok((blocks, body)) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::HeaderMap; + use axum::routing::post; + + const BODY: &[u8] = b"1\n00:00:01,000 --> 00:00:02,000\nhi\n"; + + /// A loopback gateway: `/ok` answers truthfully, `/lie` names the wrong cid, + /// `/quota` refuses, `/missing` 404s. + async fn gateway() -> String { + async fn answer(bytes: &'static [u8], cid: String) -> (HeaderMap, Vec) { + let mut h = HeaderMap::new(); + h.insert(CONTENT_CID_HEADER, HeaderValue::from_str(&cid).unwrap()); + (h, bytes.to_vec()) + } + let good = crate::ipfs_chunk::compute_ipfs_blocks(BODY).root; + let other = crate::ipfs_chunk::compute_ipfs_blocks(b"something else").root; + let app = axum::Router::new() + .route("/ok/api/file/:cid/redeem", post(move || answer(BODY, good.clone()))) + .route("/lie/api/file/:cid/redeem", post(move || answer(BODY, other.clone()))) + .route( + "/quota/api/file/:cid/redeem", + post(|| async { + let mut h = HeaderMap::new(); + h.insert(RETRY_AFTER, HeaderValue::from_static("120")); + (StatusCode::TOO_MANY_REQUESTS, h, "quota") + }), + ) + .route("/missing/api/file/:cid/redeem", post(|| async { StatusCode::NOT_FOUND })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + base + } + + const LOCATOR: &str = "bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq"; + + #[tokio::test] + async fn a_verified_answer_comes_back_with_its_blocks() { + let gw = gateway().await; + let (blocks, body) = redeem_from(&reqwest::Client::new(), &[format!("{gw}/ok")], LOCATOR) + .await + .expect("redeemed"); + assert_eq!(body, BODY); + assert_eq!(blocks.root, crate::ipfs_chunk::compute_ipfs_blocks(&body).root); + } + + #[tokio::test] + async fn a_body_that_does_not_hash_to_the_named_cid_is_refused() { + let gw = gateway().await; + match redeem_from(&reqwest::Client::new(), &[format!("{gw}/lie")], LOCATOR).await { + Err(RedeemError::Upstream(e)) => assert!(e.contains("hashes to"), "{e}"), + other => panic!("expected an upstream refusal, got {:?}", other.map(|_| ())), + } + } + + #[tokio::test] + async fn a_not_found_moves_on_to_the_next_gateway() { + let gw = gateway().await; + let targets = [format!("{gw}/missing"), format!("{gw}/ok")]; + assert!(redeem_from(&reqwest::Client::new(), &targets, LOCATOR).await.is_ok()); + } + + #[tokio::test] + async fn quota_outranks_not_found_when_nobody_delivers() { + let gw = gateway().await; + let targets = [format!("{gw}/missing"), format!("{gw}/quota")]; + match redeem_from(&reqwest::Client::new(), &targets, LOCATOR).await { + Err(RedeemError::Quota { retry_after }) => assert_eq!(retry_after, Some(120)), + other => panic!("expected quota, got {:?}", other.map(|_| ())), + } + match redeem_from(&reqwest::Client::new(), &[format!("{gw}/missing")], LOCATOR).await { + Err(RedeemError::NotFound) => {} + other => panic!("expected not found, got {:?}", other.map(|_| ())), + } + } +}