1 Commits
9 changed files with 598 additions and 24 deletions
+5
View File
@@ -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 1. **The swarm is transport-only**: identify/kad/mdns/bitswap. No kamilata, no
gossipsub, no request_response — search is meta-search's. 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. 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 3. **`core2` is patched** in `Cargo.toml` (every libp2p-0.56 bitswap impl needs
the yanked `^0.4.0`); mirror changes in meta-gateway. the yanked `^0.4.0`); mirror changes in meta-gateway.
Generated
+1 -1
View File
@@ -1990,7 +1990,7 @@ dependencies = [
[[package]] [[package]]
name = "meta-transport-ipfs" name = "meta-transport-ipfs"
version = "1.2.2" version = "1.3.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait", "async-trait",
+3 -3
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "meta-transport-ipfs" name = "meta-transport-ipfs"
version = "1.2.2" version = "1.3.0"
edition = "2021" edition = "2021"
rust-version = "1.85" rust-version = "1.85"
license = "MIT" license = "MIT"
@@ -19,7 +19,7 @@ path = "src/main.rs"
# The transport-plugin contract. Pinned by TAG, never by path — a path dep # 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 # resolves outside this repo, which is the Docker build context. See the
# Dockerfile for building against an unreleased SDK. # 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 # libp2p — transport only: identify (address exchange, load-bearing for bitswap
# reachability), kad (DHT provider records), mdns (LAN discovery) + bitswap. # reachability), kad (DHT provider records), mdns (LAN discovery) + bitswap.
@@ -56,7 +56,7 @@ multihash-codetable = { version = "0.1", features = ["sha2", "sha3"] }
[dev-dependencies] [dev-dependencies]
tempfile = "3" 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- # Workaround for the yanked transitive `core2` (all versions). `multihash-
+2
View File
@@ -38,6 +38,8 @@ pub struct AppState {
/// default = public mode). Decided here, where `PUBLIC_ADDR` is known; the /// default = public mode). Decided here, where `PUBLIC_ADDR` is known; the
/// hull always asks and this gates. /// hull always asks and this gates.
pub seed_dht_provide: bool, pub seed_dht_provide: bool,
/// Gateways that can redeem pointers, fastest first (`/ipfs-tier/resolve`).
pub gateways: Arc<crate::gateways::Gateways>,
} }
/// "This peer now seeds `cid`" — the hull writes the seed row (and decides on /// "This peer now seeds `cid`" — the hull writes the seed row (and decides on
+205
View File
@@ -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<String>,
/// Pinned base URL → (claims, fetched at).
pinned_claims: Mutex<HashMap<String, (Vec<RedeemClaim>, Instant)>>,
/// Base URL → last caps-fetch round-trip.
rtt: Mutex<HashMap<String, Duration>>,
}
fn norm(url: &str) -> String {
url.trim().trim_end_matches('/').to_string()
}
impl Gateways {
pub fn new(pinned: impl IntoIterator<Item = String>) -> Self {
let mut list: Vec<String> = 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<Vec<RedeemClaim>>, 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<String> {
let mut out: Vec<String> = 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<String>) -> Vec<String> {
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<AppState>) {
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"]
);
}
}
+4
View File
@@ -18,12 +18,14 @@ mod blockstore;
mod config; mod config;
mod filestore; mod filestore;
mod gateway_discovery; mod gateway_discovery;
mod gateways;
mod ingress; mod ingress;
mod ingress_commit; mod ingress_commit;
mod ipfs_chunk; mod ipfs_chunk;
mod material; mod material;
mod plane; mod plane;
mod plugin; mod plugin;
mod resolve;
mod settings; mod settings;
mod share; mod share;
mod store; mod store;
@@ -166,7 +168,9 @@ async fn main() -> Result<()> {
meta_core_url: cfg.meta_core_url.clone(), meta_core_url: cfg.meta_core_url.clone(),
webdav_url_cache, webdav_url_cache,
seed_dht_provide: cfg.seed_dht_provide, 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/`, // Bitswap ingress: sweep partial fetches a previous process left in `tmp/`,
// then promote/fill on a timer. // then promote/fill on a timer.
+6
View File
@@ -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."), .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") ConfigField::list("bootstrap_peers", "Pinned peers")
.with_help("Multiaddrs this node keeps dialled (e.g. a known gateway)."), .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("enable_mdns", "LAN discovery (mDNS)"),
ConfigField::bool("seed_dht_provide", "Announce seeds on the DHT") 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."), .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_namespace": env("KAD_NAMESPACE").unwrap_or_else(|| DEFAULT_NAMESPACE.to_string()),
"kad_bootstrap_peers": env_list("KAD_BOOTSTRAP_PEERS"), "kad_bootstrap_peers": env_list("KAD_BOOTSTRAP_PEERS"),
"bootstrap_peers": env_list("BOOTSTRAP_PEERS"), "bootstrap_peers": env_list("BOOTSTRAP_PEERS"),
"gateway_urls": env_list("META_SHARE_GATEWAY_URLS"),
"enable_mdns": env_flag("ENABLE_MDNS", true), "enable_mdns": env_flag("ENABLE_MDNS", true),
"seed_dht_provide": env_flag("META_SHARE_SEED_DHT_PROVIDE", public), "seed_dht_provide": env_flag("META_SHARE_SEED_DHT_PROVIDE", public),
"max_conns": env_u64("META_SHARE_MAX_CONNS", 128), "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()) { if let Some(l) = read::list(v, "bootstrap_peers").filter(|l| !l.is_empty()) {
out.push(("BOOTSTRAP_PEERS", l.join(","))); 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") { if let Some(b) = read::flag(v, "enable_mdns") {
out.push(("ENABLE_MDNS", bool_s(b))); out.push(("ENABLE_MDNS", bool_s(b)));
} }
+21 -18
View File
@@ -119,6 +119,7 @@ impl TransportPlugin for IpfsPlugin {
fn extra_routes(self: Arc<Self>) -> Router { fn extra_routes(self: Arc<Self>) -> Router {
let gateway = Router::new() let gateway = Router::new()
.route("/ipfs/:cid", get(crate::api::ipfs_gateway::get_ipfs)) .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)); .with_state(Arc::clone(&self.state));
Router::new() Router::new()
.route("/ipfs-tier/peers", get(peers)) .route("/ipfs-tier/peers", get(peers))
@@ -216,26 +217,28 @@ async fn cat(State(p): S, Path(cid): Path<String>, Query(q): Query<CatQuery>) ->
Err(e) => return bad_cid(e), Err(e) => return bad_cid(e),
}; };
let max = q.max.unwrap_or_else(crate::api::ipfs_walker::max_body_bytes); 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 { match read_object(&p.state, &mscid, max).await {
Ok(b) => b, Ok(bytes) => (StatusCode::OK, bytes).into_response(),
Err(e) => return ApiError::Upstream(format!("{e:#}")).into_response(), Err(e) => 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!( /// A block, or a dag-pb file assembled, up to `max` bytes — local, else bitswap.
"{cid} is {} bytes, over the {max}-byte ceiling", pub(crate) async fn read_object(
state: &AppState,
mscid: &crate::blockstore::MsCid,
max: usize,
) -> anyhow::Result<Vec<u8>> {
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() block.len()
)) );
.into_response(); Ok(block)
}
block
};
(StatusCode::OK, bytes).into_response()
} }
#[derive(Deserialize)] #[derive(Deserialize)]
+349
View File
@@ -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<u64> },
/// 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<String>,
}
pub async fn resolve(
State(state): State<Arc<AppState>>,
Path(cid): Path<String>,
Query(q): Query<ResolveQuery>,
) -> 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<Vec<u8>> {
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<u8>, 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<u64>, 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<u8>), RedeemError> {
let mut quota: Option<Option<u64>> = None;
let mut upstream: Option<String> = 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<u8>), 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<u8>) {
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(|_| ())),
}
}
}