From 1c9d415a1e554d0604e2b465666d64192adde752 Mon Sep 17 00:00:00 2001 From: Worph Date: Sun, 13 Sep 2026 14:30:54 +0200 Subject: [PATCH] improvement --- Cargo.toml | 2 +- README.md | 47 +++++++++++-- src/hash.rs | 184 +++++++++++++++++++++++++++++++++++++++++++++++++- src/lib.rs | 2 + src/plugin.rs | 94 ++++++++++++++++++++++++++ src/serve.rs | 178 +++++++++++++++++++++++++++++++++++++++++++++++- 6 files changed, 496 insertions(+), 11 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 660ba1e..766e57d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-feeder-sdk" -version = "0.1.0" +version = "1.3.0" edition = "2021" rust-version = "1.85" license = "MIT" diff --git a/README.md b/README.md index 9756eb6..c96a22c 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ bytes; it never joins the swarm. ```toml [dependencies] -meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v0.1.0" } +meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.3.0" } ``` Not on crates.io yet — the contract is still moving, so consumers pin a tag. @@ -47,8 +47,32 @@ Four required methods; everything else has a default: | `compute_outcomes` | resolve a `record_id` to `HashOutcome`s | `bytes: None` for metadata-only sources | Defaults you can override: `handle_query_stream`, `handle_fetch`, `health`, -`served_file_types`, `served_content_kinds`, `config_schema`, `config_values`, -`get_blob`. +`served_file_types`, `served_content_kinds`, `package`, `redeems`, +`config_schema`, `config_values`, `get_blob`. + +### Redeeming locators (1.3.0) + +A plugin can **redeem** a locator family — turn a locator cid into the bytes it +names, usually spending a metered quota (a Newznab `.nzb` grab, an OpenSubtitles +download): + +1. Return a `RedeemClaim` from `redeems()` — `RedeemClaim::nzb_release(hosts)` or + `RedeemClaim::provider_file(sources)` — reflecting the *current* config (no + credential → no claim). Served per plugin in `GET /manifest` and, for polling, + at `GET /redeems` → `{"redeems": {"": [claim…]}}`. +2. Name a stable `package()` — the `/files/plugin//` folder the gateway + stores the bytes under. +3. In `compute_outcomes`, accept the locator cid as `record_id` and return ONE + `Sha2_256` outcome: `hash = hash::compute_ipfs_cid(&bytes)`, `bytes`, + `file_extension`, and optionally a `record` whose fields the gateway merges + onto the locator record. Return `NotFound` when the cid is not yours (e.g. no + key for that host) so the gateway can try the next claimer; `RateLimited` when + the quota is spent. + +The gateway calls this only on a real play, one claimer at a time, and writes +the bytes' cid onto the locator record under the claim's `field` (`manifest` / +`file`), so each file's quota is spent once. Decode helpers: +`hash::decode_nzb_release_cid`, `hash::decode_provider_file_cid`, `hash::codec_of`. ### The routing gate @@ -101,6 +125,9 @@ into the bitswap blockstore. The last two carry an identity multihash. Anything that dispatches on the multihash instead of the **codec** will silently mis-rank them. +`hash` also encodes/decodes the redeemable locators: `nzb-release` (`0x1005`) +and `provider-file` (`0x100A`, `(source, id)` — `docs/cid-formats.md` §8). + ## Layout | module | what | @@ -121,13 +148,19 @@ multihash instead of the **codec** will silently mis-rank them. cargo build && cargo test ``` -To work on the SDK from inside a feeder checkout without pushing a tag, add a -gitignored `.cargo/config.toml` in the feeder: +To test a feeder against an SDK change **before its tag is pushed**: cargo will +not resolve an unpushed tag, not even under a `[patch]` override ("failed to +find tag"). Build a scratch copy of the feeder (outside its git tree) with the +dependency swapped for a path, and run `cargo test` there: -```toml -paths = ["../meta-feeder-sdk"] +```bash +mkdir -p /tmp/feeder && (cd meta-feeder-x && tar cf - --exclude=./target --exclude=./.git .) | (cd /tmp/feeder && tar xf -) +sed -i 's#meta-feeder-sdk = { git = .*#meta-feeder-sdk = { path = "/abs/path/to/meta-feeder-sdk" }#' /tmp/feeder/feeder-plugin/*/Cargo.toml +(cd /tmp/feeder && cargo test) ``` +Never commit the path form: each feeder repo is its own Docker build context. + ## Context Design rationale, the feeder roster, and the naming rules live in the meta-root: diff --git a/src/hash.rs b/src/hash.rs index ea5b24d..bbb0da9 100644 --- a/src/hash.rs +++ b/src/hash.rs @@ -206,6 +206,76 @@ pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option Option<(String, String)> { + let digest = identity_locator_digest(cid, NZB_RELEASE_CODEC)?; + let mut pos = 0usize; + let base_len = read_varint(&digest, &mut pos)? as usize; + let rest = digest.get(pos..)?; + if rest.len() <= base_len { + return None; // need the base AND at least one id byte + } + let (base, id) = rest.split_at(base_len); + let api_base = std::str::from_utf8(base).ok()?.to_string(); + let id_hex: String = id.iter().map(|b| format!("{b:02x}")).collect(); + Some((api_base, id_hex)) +} + +/// The multicodec of a base32 (`b…`) CIDv1, or `None` for anything else +/// (CIDv0 `Qm…`, other multibases, garbage). Dispatch on this — never on the +/// multihash, which is identity (`0x00`) for every locator family. +pub fn codec_of(cid: &str) -> Option { + let wire = base32_lower_decode(cid.strip_prefix('b')?)?; + let mut pos = 0usize; + if read_varint(&wire, &mut pos)? != 1 { + return None; + } + read_varint(&wire, &mut pos) +} + +/// The digest of a well-formed identity-multihash CIDv1 carrying `codec`, or +/// `None`. Shared by the locator decoders. +fn identity_locator_digest(cid: &str, codec: u64) -> Option> { + let wire = base32_lower_decode(cid.strip_prefix('b')?)?; + let mut pos = 0usize; + if read_varint(&wire, &mut pos)? != 1 + || read_varint(&wire, &mut pos)? != codec + || read_varint(&wire, &mut pos)? != 0x00 + { + return None; + } + let len = read_varint(&wire, &mut pos)? as usize; + let digest = wire.get(pos..)?; + (digest.len() == len).then(|| digest.to_vec()) +} + +/// Unsigned LEB128 varint at `*pos`, advancing it. +fn read_varint(bytes: &[u8], pos: &mut usize) -> Option { + let mut value = 0u64; + let mut shift = 0u32; + loop { + let byte = *bytes.get(*pos)?; + *pos += 1; + value |= u64::from(byte & 0x7f) << shift; + if byte & 0x80 == 0 { + return Some(value); + } + shift += 7; + if shift >= 64 { + return None; + } + } +} + /// Hex-decode a Newznab release id. The guid parser only keeps hex runs, so the /// id is all-hex by construction; odd-length ids are left-padded with a `0` /// nibble. Returns `None` on any non-hex byte (defensive). @@ -399,8 +469,39 @@ pub fn compute_ext_play_cid(url: &str) -> Option { Some(identity_locator_cid(EXT_PLAY_CODEC, digest)) } +/// Custom multicodec for a fetchable **`provider-file`** locator — +/// `docs/cid-formats.md` §8. One codec for every provider whose bytes are +/// fetched the same way; the provider lives inside the digest: +/// `varint(len(source)) ‖ source ‖ id`. Framing byte-identical to `card` +/// (`0x1007`) — only the codec differs — but the byte path *redeems* it (through +/// the claiming feeder's `/compute`) instead of 404ing. +pub const PROVIDER_FILE_CODEC: u64 = 0x100A; + +/// Encode a `provider-file` CID from a registered source token +/// (`opensubtitles`) and that source's id (`file:7061834`). `None` on an empty +/// part or a digest past 64 bytes — the caller drops that row rather than emit +/// an address no peer can parse. +pub fn compute_provider_file_cid(source: &str, id: &str) -> Option { + kinded_locator_cid(PROVIDER_FILE_CODEC, source, id) +} + +/// Inverse of [`compute_provider_file_cid`]: `(source, id)`. Matches the +/// **codec slot** — a `card` with identical framing does not decode here. +pub fn decode_provider_file_cid(cid: &str) -> Option<(String, String)> { + let digest = identity_locator_digest(cid, PROVIDER_FILE_CODEC)?; + let mut pos = 0usize; + let source_len = read_varint(&digest, &mut pos)? as usize; + let source = digest.get(pos..pos.checked_add(source_len)?)?; + let id = digest.get(pos + source_len..)?; + Some(( + String::from_utf8(source.to_vec()).ok()?, + String::from_utf8(id.to_vec()).ok()?, + )) +} + /// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used -/// by [`compute_card_cid`] and [`compute_yt_video_cid`]. +/// by [`compute_card_cid`], [`compute_yt_video_cid`] and +/// [`compute_provider_file_cid`]. fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option { if kind.is_empty() || id.is_empty() { return None; @@ -1389,3 +1490,84 @@ mod roundtrip_tests { assert_eq!(decode_url_cid(&card), None); } } + +#[cfg(test)] +mod redeem_locator_tests { + use super::*; + + /// `docs/cid-formats.md` §8.2 golden vector — moved here from + /// meta-feeder-opensubtitles' local `locator.rs`; the vector proves the + /// move changed no bytes. + #[test] + fn provider_file_matches_the_documented_vector() { + assert_eq!( + compute_provider_file_cid("opensubtitles", "file:7061834").as_deref(), + Some("bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq") + ); + } + + #[test] + fn provider_file_round_trips_and_refuses_other_codecs() { + let cid = compute_provider_file_cid("opensubtitles", "file:12658754").unwrap(); + assert_eq!( + decode_provider_file_cid(&cid), + Some(("opensubtitles".to_string(), "file:12658754".to_string())) + ); + // Identical framing, different codec: a card must not decode as a file. + let card = compute_card_cid("tmdb", "tv:95479").unwrap(); + assert_eq!(card, "bagdsaaanar2g2zdcor3duojvgq3ts"); + assert_eq!(decode_provider_file_cid(&card), None); + assert_eq!(decode_provider_file_cid("not-a-cid"), None); + assert_eq!(decode_provider_file_cid(""), None); + } + + #[test] + fn provider_file_digest_ceiling_is_64_bytes() { + // varint(13) + "opensubtitles" = 14 bytes, so the id may use 50. + assert!(compute_provider_file_cid("opensubtitles", &"x".repeat(50)).is_some()); + assert!(compute_provider_file_cid("opensubtitles", &"x".repeat(51)).is_none()); + } + + #[test] + fn nzb_release_round_trips() { + let cid = compute_nzb_release_cid("api.nzbgeek.info/api", "0123abcdef").unwrap(); + assert_eq!( + decode_nzb_release_cid(&cid), + Some(("api.nzbgeek.info/api".to_string(), "0123abcdef".to_string())) + ); + // A host:port authority survives too. + let cid = compute_nzb_release_cid("127.0.0.1:8080", "ff00").unwrap(); + assert_eq!( + decode_nzb_release_cid(&cid), + Some(("127.0.0.1:8080".to_string(), "ff00".to_string())) + ); + } + + /// The encoder left-pads an odd-length id, so the decode is the padded form. + #[test] + fn nzb_release_odd_length_id_comes_back_padded() { + let cid = compute_nzb_release_cid("api.nzb.life", "abc").unwrap(); + assert_eq!( + decode_nzb_release_cid(&cid), + Some(("api.nzb.life".to_string(), "0abc".to_string())) + ); + } + + #[test] + fn nzb_release_refuses_other_codecs() { + let card = compute_card_cid("tmdb", "tv:95479").unwrap(); + assert_eq!(decode_nzb_release_cid(&card), None); + assert_eq!(decode_nzb_release_cid("bafkreigarbage"), None); + } + + #[test] + fn codec_of_reads_the_codec_slot() { + let nzb = compute_nzb_release_cid("api.nzb.life", "ab").unwrap(); + assert_eq!(codec_of(&nzb), Some(NZB_RELEASE_CODEC)); + let file = compute_provider_file_cid("opensubtitles", "file:1").unwrap(); + assert_eq!(codec_of(&file), Some(PROVIDER_FILE_CODEC)); + assert_eq!(codec_of(&compute_ipfs_cid(b"hello")), Some(0x55)); + assert_eq!(codec_of("QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG"), None); + assert_eq!(codec_of(""), None); + } +} diff --git a/src/lib.rs b/src/lib.rs index 8d6a242..18b1c64 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,10 +45,12 @@ pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher}; pub use meta_core::FeederStore; pub use plugin::{ upstream_id_field, ConfigError, FeederPlugin, HashKind, HashOutcome, PluginRegistry, + RedeemClaim, }; pub use query::{GatewayQuery, GatewaySearchEvent, GatewayWireError, Negation, RangeFilter}; pub use serve::{ configure_plugins, router, serve_feeders, ComputeRequest, ComputeResponse, HashKindDto, HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse, + RedeemsResponse, }; pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth}; diff --git a/src/plugin.rs b/src/plugin.rs index f1a36b0..d1bfbf0 100644 --- a/src/plugin.rs +++ b/src/plugin.rs @@ -17,6 +17,7 @@ use std::path::Path; use async_trait::async_trait; use futures::stream::BoxStream; +use serde::{Deserialize, Serialize}; // Re-exported so plugins can `use meta_feeder_sdk::plugin::GatewayQuery` (the // gateway's original `plugin.rs` re-exported it the same way). @@ -95,6 +96,66 @@ pub enum HashKind { YtVideo, } +/// `RedeemClaim::codec` for a Newznab release locator (`0x1005`). +pub const REDEEM_CODEC_NZB_RELEASE: &str = "nzb-release"; +/// `RedeemClaim::codec` for a `provider-file` locator (`0x100A`). +pub const REDEEM_CODEC_PROVIDER_FILE: &str = "provider-file"; +/// `RedeemClaim::field` naming the `.nzb` manifest pointer on a release record. +pub const POINTER_FIELD_MANIFEST: &str = "manifest"; +/// `RedeemClaim::field` naming the fetched-file pointer on a locator record. +pub const POINTER_FIELD_FILE: &str = "file"; + +/// A plugin's claim that it can **redeem** a locator family: turn a locator cid +/// into the bytes it names, through `POST /compute` with `record_id` = the +/// locator cid. +/// +/// Redeeming usually spends a metered upstream quota (a Newznab `t=get` grab, +/// an OpenSubtitles `/download`), so the gateway routes a redeem only to the +/// plugins claiming the cid's codec, one at a time, and only on a real play. It +/// stores the returned bytes and writes their cid onto the locator record under +/// [`field`](Self::field), so the quota is spent once per file. +/// +/// Served at `GET /manifest` (per plugin) and `GET /redeems`. Reflects the +/// plugin's *current* config: a plugin with no credential for a family claims +/// nothing for it. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct RedeemClaim { + /// Locator family: [`REDEEM_CODEC_NZB_RELEASE`] or [`REDEEM_CODEC_PROVIDER_FILE`]. + pub codec: String, + /// Pointer field the gateway writes the redeemed bytes' cid under: + /// [`POINTER_FIELD_MANIFEST`] or [`POINTER_FIELD_FILE`]. + pub field: String, + /// Indexer hosts this plugin holds a key for (`nzb-release`) — bare + /// authority, lowercase, never the key itself. + #[serde(default)] + pub hosts: Vec, + /// Provider sources this plugin redeems (`provider-file`), e.g. `opensubtitles`. + #[serde(default)] + pub sources: Vec, +} + +impl RedeemClaim { + /// Claim `nzb-release` locators for these indexer hosts. + pub fn nzb_release(hosts: Vec) -> Self { + Self { + codec: REDEEM_CODEC_NZB_RELEASE.to_string(), + field: POINTER_FIELD_MANIFEST.to_string(), + hosts, + sources: Vec::new(), + } + } + + /// Claim `provider-file` locators for these sources. + pub fn provider_file(sources: Vec) -> Self { + Self { + codec: REDEEM_CODEC_PROVIDER_FILE.to_string(), + field: POINTER_FIELD_FILE.to_string(), + hosts: Vec::new(), + sources, + } + } +} + /// Startup-only configuration error returned by [`FeederPlugin::configure`]. #[derive(Debug, thiserror::Error)] pub enum ConfigError { @@ -215,6 +276,20 @@ pub trait FeederPlugin: Send + Sync + 'static { &[] } + /// Folder under meta-core's `/files/plugin/` where the gateway stores bytes + /// this plugin redeems (e.g. `meta-feeder-usenet`). Stable across + /// deployments — unlike the feeder's compose service name. Default: none. + fn package(&self) -> Option<&'static str> { + None + } + + /// Locator families this plugin can redeem through + /// [`compute_outcomes`](FeederPlugin::compute_outcomes) — see + /// [`RedeemClaim`]. Computed from the current config. Default: none. + fn redeems(&self) -> Vec { + Vec::new() + } + /// Self-describing config schema. The feeder SDK serves this at /// `GET /config/schema` and renders a generic form from it — the gateway and /// UI carry no per-plugin field knowledge. Default: no configuration. @@ -255,4 +330,23 @@ mod tests { assert!(msg.contains("scihub")); assert!(msg.contains("mirrors")); } + + /// The JSON keys are the gateway/meta-share contract — pin them. + #[test] + fn redeem_claim_wire_shape() { + let claim = RedeemClaim::nzb_release(vec!["api.nzb.life".into()]); + assert_eq!( + serde_json::to_value(&claim).unwrap(), + serde_json::json!({ + "codec": "nzb-release", + "field": "manifest", + "hosts": ["api.nzb.life"], + "sources": [] + }) + ); + // hosts/sources may be omitted on the wire. + let parsed: RedeemClaim = + serde_json::from_str(r#"{"codec":"provider-file","field":"file"}"#).unwrap(); + assert_eq!(parsed, RedeemClaim::provider_file(vec![])); + } } diff --git a/src/serve.rs b/src/serve.rs index 9fc407f..57a8a1c 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -10,6 +10,17 @@ //! - `POST /compute` → [`ComputeRequest`] → [`ComputeResponse`] //! - `GET /fetch/{upstream_id}/{record_id}` → streamed bytes (or 404) //! - `GET /blob/{upstream_id}/{cid}` → bytes (or 404) +//! - `GET /redeems` → [`RedeemsResponse`] (current redeem claims) +//! +//! ## Redeeming a locator +//! +//! A plugin that advertises a [`RedeemClaim`] answers `POST /compute` with +//! `record_id` = a locator cid of that family (a Newznab `nzb-release`, an +//! OpenSubtitles `provider-file`) by fetching the bytes it names — usually +//! spending a metered quota — and returning ONE `Sha2_256` outcome carrying +//! them. `NotFound` means "not mine" (e.g. no key for that indexer host), so the +//! gateway tries the next claimer. The claims are re-read from `GET /redeems` +//! so a config change shows up without restarting the gateway. //! //! ## Bytes & the v1 simplification //! @@ -21,7 +32,7 @@ //! documented follow-up — those outcomes are metadata-only anyway //! (`bytes: None`), so they never ride this path. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::net::SocketAddr; use std::path::{Path as FsPath, PathBuf}; use std::sync::Arc; @@ -41,7 +52,7 @@ use serde_json::Value; use tracing::{info, warn}; use crate::config::{self, CONFIG_PAGE_HTML}; -use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome}; +use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim}; use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError}; use crate::types::{DiscoveryRecord, GatewayError, PluginHealth}; @@ -94,6 +105,22 @@ pub struct PluginManifest { pub id: String, pub served_file_types: Vec, pub served_content_kinds: Vec, + /// [`FeederPlugin::package`] — the `/files/plugin//` folder redeemed + /// bytes are stored under. `null` when the plugin redeems nothing. + #[serde(default)] + pub package: Option, + /// [`FeederPlugin::redeems`] at the time the manifest was served. + #[serde(default)] + pub redeems: Vec, +} + +/// `GET /redeems` — every hosted plugin's current [`RedeemClaim`]s, keyed by +/// `upstream_id` (an empty list for a plugin that claims nothing). Polled by +/// the gateway, so a credential added on the config page is advertised without +/// a gateway restart. +#[derive(Serialize, Deserialize, Clone, Debug, Default)] +pub struct RedeemsResponse { + pub redeems: BTreeMap>, } /// `GET /manifest` — the feeder's hosted plugins + their served types. The @@ -224,6 +251,17 @@ impl AppState { /// status semantics the gateway's `RemoteFeederPlugin` will translate back /// into a `GatewayError` on the core side. fn gateway_error_response(e: GatewayError) -> Response { + // A 429 carries the wait as a `Retry-After` header too, so the gateway can + // forward it to the client (a redeem blocked on a spent quota) instead of + // parsing the message. + if let GatewayError::RateLimited { retry_after_s } = &e { + return ( + StatusCode::TOO_MANY_REQUESTS, + [(axum::http::header::RETRY_AFTER, retry_after_s.to_string())], + format!("rate limited; retry in {retry_after_s}s"), + ) + .into_response(); + } let (status, msg) = match e { GatewayError::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()), GatewayError::RateLimited { retry_after_s } => ( @@ -251,6 +289,8 @@ async fn manifest(State(state): State) -> Json { .iter() .map(|s| s.to_string()) .collect(), + package: p.package().map(str::to_string), + redeems: p.redeems(), }) .collect(); plugins.sort_by(|a, b| a.id.cmp(&b.id)); @@ -260,6 +300,16 @@ async fn manifest(State(state): State) -> Json { }) } +async fn redeems(State(state): State) -> Json { + Json(RedeemsResponse { + redeems: state + .plugins + .values() + .map(|p| (p.upstream_id().to_string(), p.redeems())) + .collect(), + }) +} + async fn health(State(state): State) -> Json { let mut entries: Vec = state .plugins @@ -492,6 +542,7 @@ pub fn router( .route("/compute", post(compute)) .route("/fetch/:upstream_id/:record_id", get(fetch)) .route("/blob/:upstream_id/:cid", get(blob)) + .route("/redeems", get(redeems)) .route("/config", get(config_page)) .route("/config/schema", get(config_schema)) .route("/config/values", get(config_values_get).put(config_values_put)) @@ -541,6 +592,9 @@ struct DegradedPlugin { reason: String, file_types: &'static [&'static str], content_kinds: &'static [&'static str], + /// Preserved so the manifest keeps naming the plugin's folder. A degraded + /// plugin claims no redeems (it cannot redeem anything). + package: Option<&'static str>, /// Preserved from the wrapped plugin so a degraded/unconfigured feeder still /// serves its config form — otherwise the dashboard shows "no configuration /// required" and the operator can never fix the very config that's missing. @@ -586,6 +640,10 @@ impl FeederPlugin for DegradedPlugin { self.content_kinds } + fn package(&self) -> Option<&'static str> { + self.package + } + fn config_schema(&self) -> crate::config::ConfigSchema { self.schema.clone() } @@ -608,6 +666,7 @@ pub fn configure_plugins( // borrow across `configure(&mut self)` / the move into `out`. let file_types = plugin.served_file_types(); let content_kinds = plugin.served_content_kinds(); + let package = plugin.package(); // Preserve the config surface so a degraded stand-in still serves the // real config form (see DegradedPlugin.schema). let schema = plugin.config_schema(); @@ -627,6 +686,7 @@ pub fn configure_plugins( reason, file_types, content_kinds, + package, schema: schema.clone(), }) }; @@ -759,4 +819,118 @@ mod tests { assert!(matches!(events[1], GatewaySearchEvent::EnrichPatch { .. })); assert!(matches!(events[2], GatewaySearchEvent::Done)); } + + /// A plugin that redeems `nzb-release` locators for one host. + struct StubRedeemer; + + #[async_trait] + impl FeederPlugin for StubRedeemer { + fn upstream_id(&self) -> &'static str { + "redeemer" + } + fn configure(&mut self, _cache_dir: &FsPath) -> Result<(), ConfigError> { + Ok(()) + } + async fn handle_query( + &self, + _q: &GatewayQuery, + _n: usize, + ) -> Result, GatewayError> { + Ok(vec![]) + } + async fn compute_outcomes( + &self, + _record_id: &str, + ) -> Result, GatewayError> { + Ok(vec![]) + } + fn package(&self) -> Option<&'static str> { + Some("meta-feeder-stub") + } + fn redeems(&self) -> Vec { + vec![RedeemClaim::nzb_release(vec!["api.example".into()])] + } + } + + async fn serve_stubs() -> String { + let mut plugins: HashMap> = HashMap::new(); + plugins.insert("redeemer".to_string(), Arc::new(StubRedeemer)); + plugins.insert("stub".to_string(), Arc::new(StubStreaming)); + let app = router(plugins, "test".to_string(), std::env::temp_dir()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{addr}") + } + + /// The manifest carries `package` + `redeems` per plugin; a plugin that + /// claims nothing serializes `null` + `[]`. + #[tokio::test] + async fn manifest_carries_package_and_redeems() { + let base = serve_stubs().await; + let m: serde_json::Value = reqwest::get(format!("{base}/manifest")) + .await + .unwrap() + .json() + .await + .unwrap(); + let redeemer = &m["plugins"][0]; + assert_eq!(redeemer["id"], "redeemer"); + assert_eq!(redeemer["package"], "meta-feeder-stub"); + assert_eq!( + redeemer["redeems"], + serde_json::json!([{ + "codec": "nzb-release", "field": "manifest", + "hosts": ["api.example"], "sources": [] + }]) + ); + let stub = &m["plugins"][1]; + assert_eq!(stub["package"], serde_json::Value::Null); + assert_eq!(stub["redeems"], serde_json::json!([])); + } + + #[tokio::test] + async fn redeems_route_keys_claims_by_upstream_id() { + let base = serve_stubs().await; + let r: RedeemsResponse = reqwest::get(format!("{base}/redeems")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + r.redeems.get("redeemer"), + Some(&vec![RedeemClaim::nzb_release(vec!["api.example".into()])]) + ); + assert_eq!(r.redeems.get("stub"), Some(&vec![])); + } + + /// The gateway forwards a feeder 429's `Retry-After` — pin that it is set. + #[test] + fn rate_limited_sets_retry_after() { + let resp = gateway_error_response(GatewayError::RateLimited { retry_after_s: 3600 }); + assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + resp.headers() + .get(axum::http::header::RETRY_AFTER) + .and_then(|v| v.to_str().ok()), + Some("3600") + ); + // Other errors carry no Retry-After. + let resp = gateway_error_response(GatewayError::NotFound); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + assert!(resp.headers().get(axum::http::header::RETRY_AFTER).is_none()); + } + + /// A manifest from a pre-1.3 feeder (no `package` / `redeems`) still parses. + #[test] + fn old_manifest_without_redeem_fields_parses() { + let m: PluginManifest = serde_json::from_str( + r#"{"id":"x","served_file_types":[],"served_content_kinds":[]}"#, + ) + .unwrap(); + assert!(m.package.is_none() && m.redeems.is_empty()); + } }