2 Commits
8 changed files with 323 additions and 10 deletions
+7 -3
View File
@@ -1,6 +1,6 @@
[package]
name = "meta-feeder-sdk"
version = "1.7.0"
version = "1.8.1"
edition = "2021"
rust-version = "1.85"
license = "MIT"
@@ -37,11 +37,15 @@ default = ["feeder"]
# FeederPlugin trait, hashing, the redb cache, the axum harness. On by default
# so existing consumers build exactly what they always did.
feeder = [
"beacon",
"beacon", "filename",
"dep:axum", "dep:hyper", "dep:reqwest", "dep:base64", "dep:futures",
"dep:async-trait", "dep:anyhow", "dep:thiserror", "dep:sha2", "dep:bytes",
"dep:regex", "dep:redb",
"dep:redb",
]
# Release-name parsing (`src/filename_meta.rs`: clean_title, season/episode,
# quality, codec). Needs only `regex`, so a client (meta-watch) can derive a
# display name from a raw release string without the feeder machinery.
filename = ["dep:regex"]
# Beacon v2 local discovery (`src/beacon/`). Needs only tokio/serde/tracing/libc,
# so a meta-* service takes the SDK with `default-features = false,
# features = ["beacon"]` and pulls in none of the feeder machinery.
+1 -1
View File
@@ -44,7 +44,7 @@ pub mod config;
pub mod domain;
#[cfg(feature = "feeder")]
pub mod enrich;
#[cfg(feature = "feeder")]
#[cfg(feature = "filename")]
pub mod filename_meta;
#[cfg(feature = "feeder")]
pub mod hash;
+19
View File
@@ -156,6 +156,25 @@ impl RemoteTransport {
}
}
/// `PUT /manifests/:cid`: hand a manifest-consuming plugin (contract 2) the
/// manifest a pointer resolved to, after it answered `428` +
/// [`super::dto::HDR_NEEDS_MANIFEST`].
pub async fn push_manifest(&self, cid: &str, manifest: bytes::Bytes) -> Result<(), ApiError> {
let r = self
.http
.put(self.url(&format!("/manifests/{cid}")))
.body(manifest)
.timeout(self.control_timeout)
.send()
.await
.map_err(|e| self.unreachable(e))?;
if r.status().is_success() {
Ok(())
} else {
Err(error_from(r).await)
}
}
// ---- typed helpers for extra routes ------------------------------------
pub async fn get_json<T: DeserializeOwned>(&self, path: &str) -> Result<T, ApiError> {
+31 -2
View File
@@ -9,8 +9,37 @@ 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;
/// a plugin whose `/manifest` reports a contract outside
/// [`MIN_CONTRACT`]`..=`[`CONTRACT_VERSION`].
///
/// - **2**: a plugin that consumes a manifest (nzb) no longer pulls it from the
/// hull. It answers `/raw` (and its verify route) with `428` +
/// [`HDR_NEEDS_MANIFEST`] until the hull has pushed one with
/// `PUT /manifests/:cid`.
pub const CONTRACT_VERSION: u32 = 2;
/// The oldest contract the hull still drives (contract-1 nzb plugins pull their
/// manifest from `/internal/nzb/manifest/:cid`, kept for one release).
pub const MIN_CONTRACT: u32 = 1;
/// `428` marker: this plugin needs the cid's manifest pushed before it can
/// answer. The hull resolves the pointer, `PUT`s `/manifests/:cid`, and retries.
pub const HDR_NEEDS_MANIFEST: &str = "x-metamesh-needs-manifest";
/// Why a pointer could not be resolved, on a non-2xx `/ipfs-tier/resolve`
/// answer: [`RESOLVE_NO_GATEWAY`], [`RESOLVE_UNCLAIMED`], [`RESOLVE_QUOTA`],
/// [`RESOLVE_NOT_FOUND`] or [`RESOLVE_UPSTREAM`].
pub const HDR_RESOLVE_ERROR: &str = "x-metamesh-resolve-error";
/// No gateway is known at all yet (discovery is eventual): retry.
pub const RESOLVE_NO_GATEWAY: &str = "no-gateway";
/// Gateways are known, but none claims this pointer's key.
pub const RESOLVE_UNCLAIMED: &str = "unclaimed";
/// The provider's quota is spent: retry after `Retry-After`.
pub const RESOLVE_QUOTA: &str = "quota";
/// Every claiming gateway answered not found (or nothing is stored and the
/// pointer cannot be redeemed).
pub const RESOLVE_NOT_FOUND: &str = "not-found";
/// A gateway failed, or answered bytes that did not verify.
pub const RESOLVE_UPSTREAM: &str = "upstream";
/// `X-MetaMesh-Lane`: the hull's lane decision for a `/raw` request.
pub const HDR_LANE: &str = "x-metamesh-lane";
+4 -3
View File
@@ -39,10 +39,11 @@ 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.
/// Where a pointer's bytes came from: `store` (already held here, named by the
/// record's pointer field), `redeem` (a gateway), or `record` (contract-1
/// callback naming for `store`).
pub const HDR_MANIFEST_SOURCE: &str = "x-metamesh-manifest-source";
/// The content cid of a `record`-sourced manifest.
/// The content cid the pointer resolved to.
pub const HDR_MANIFEST_CID: &str = "x-metamesh-manifest-cid";
/// `.nzb` bytes as the hull resolved them.
+10
View File
@@ -17,6 +17,7 @@
//! |---|---|
//! | `GET peers` → [`PeersInfo`] | `swarm::Command::Peers` |
//! | `GET directory` → [`Directory`] | `swarm::PeerDirectory` (gateway discovery) |
//! | `GET resolve/:cid?hint=` → bytes | a [pointer](super::pointer)'s bytes: the stored copy `hint` names, else a gateway redeem (see below) |
//! | `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` |
@@ -40,6 +41,15 @@
//! ingress job was renamed into `cache/`), and `GET /internal/records/:cid` for
//! a display name.
//! **Resolving a pointer** (`GET resolve/:cid`) is the one route that reaches
//! outside the swarm: a pointer is answered only by a gateway holding the key,
//! over plain HTTP at the gateway's base URL. Gateways come from config (the
//! local one, pinned) and from the swarm directory; the fastest one whose redeem
//! claim covers the pointer is asked first. `2xx` carries the bytes plus
//! [`HDR_MANIFEST_SOURCE`](super::hull::HDR_MANIFEST_SOURCE) (`store` | `redeem`)
//! and [`HDR_MANIFEST_CID`](super::hull::HDR_MANIFEST_CID); a failure carries
//! [`HDR_RESOLVE_ERROR`](super::dto::HDR_RESOLVE_ERROR).
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
+3 -1
View File
@@ -22,6 +22,7 @@
//! - [`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.
//! - [`pointer`]: which cids are pointers (resolved before their bytes exist).
//! - [`magic`]: content sniffing; [`file`]: range-serving a file off `/data`.
//! - `testkit` (feature `testkit`): conformance checks for any implementation.
//!
@@ -42,6 +43,7 @@ pub mod ipfs;
pub mod magic;
pub mod nzb;
pub mod plugin;
pub mod pointer;
pub mod range;
pub mod serve;
pub mod torrent;
@@ -52,7 +54,7 @@ pub use client::RemoteTransport;
pub use config::ConfigPlane;
pub use dto::{
Capabilities, Deleted, Event, Health, Job, Jobs, Manifest, PromotedFile, ReconcileReport,
ReconcileRequest, CONTRACT_VERSION,
ReconcileRequest, CONTRACT_VERSION, MIN_CONTRACT,
};
pub use error::ApiError;
pub use focus::{FocusSnapshot, FocusTitle, FocusView, Lane};
+248
View File
@@ -0,0 +1,248 @@
//! **Pointer cids** — cids that name something to fetch rather than hash it.
//!
//! A pointer's multihash is `identity`: the cid *contains* the address (an
//! indexer host + release id, a provider source + file id), not a digest of the
//! bytes. Bitswap cannot answer one, because there is no hash to verify against.
//! It is resolved instead: by a manifest this peer already stored (the record's
//! pointer field names its content cid), or by a **redeem** at a gateway whose
//! feeder holds the credential for `key`.
//!
//! This module is the single place that knows which codecs are pointers, so the
//! hull ("resolve it first") and the ipfs plugin ("here is how") agree.
//!
//! | codec | what it points at | redeem key | record field |
//! |---|---|---|---|
//! | `0x1005` nzb-release | a `.nzb` at an indexer | indexer host | `manifest` |
//! | `0x1003` nzb-posting | a self-scanned `.nzb` | — (record only) | `manifest` |
//! | `0x100A` provider-file | a file a provider holds | provider source | `file` |
use anyhow::{Context, Result};
use super::cid::{parse_record_cid, read_uvarint};
use super::nzb::{
decode_nzb_release_cid, normalize_indexer_host, NZB_POSTING_CODEC, NZB_RELEASE_CODEC,
};
/// MetaMesh-private multicodec for a provider-held file (`docs/cid-formats.md` §8).
pub const PROVIDER_FILE_CODEC: u64 = 0x100A;
/// Redeem-claim codec names, as gateways advertise them in
/// `GET /api/gateway/plugins` (`redeems[].codec`).
pub const REDEEM_NZB_RELEASE: &str = "nzb-release";
pub const REDEEM_PROVIDER_FILE: &str = "provider-file";
/// Record fields a redeem merges onto the pointer's record: the content cid of
/// what it resolved to.
pub const FIELD_MANIFEST: &str = "manifest";
pub const FIELD_FILE: &str = "file";
/// Identity multihash code: the digest bytes are the payload verbatim.
const IDENTITY_MULTIHASH: u64 = 0x00;
/// A decoded pointer.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Pointer {
/// The redeem-claim codec, or `None` when no gateway can redeem it (a
/// self-scanned posting resolves from the record alone).
pub redeem: Option<RedeemKey>,
/// The record field naming the resolved content cid.
pub field: &'static str,
/// What the resolved bytes are: a manifest a byte plugin consumes, or the
/// file itself.
pub kind: PointerKind,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RedeemKey {
/// [`REDEEM_NZB_RELEASE`] or [`REDEEM_PROVIDER_FILE`].
pub codec: &'static str,
/// The indexer host / provider source a gateway's claim must cover.
pub key: String,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PointerKind {
/// A `.nzb`: handed to the nzb plugin, which fetches the bytes.
NzbManifest,
/// The file itself (a subtitle): served as is.
File,
}
/// Decode `cid` as a pointer.
///
/// - `None` — not a pointer: an ordinary content cid, or unparseable.
/// - `Some(Err(_))` — the codec says pointer but the payload is malformed; the
/// caller answers `400`.
pub fn decode(cid: &str) -> Option<Result<Pointer>> {
match codec_of(cid)? {
NZB_RELEASE_CODEC => Some(decode_nzb_release(cid)),
NZB_POSTING_CODEC => Some(Ok(Pointer {
redeem: None,
field: FIELD_MANIFEST,
kind: PointerKind::NzbManifest,
})),
PROVIDER_FILE_CODEC => Some(decode_provider_file(cid).map(|(source, _id)| Pointer {
redeem: Some(RedeemKey { codec: REDEEM_PROVIDER_FILE, key: source }),
field: FIELD_FILE,
kind: PointerKind::File,
})),
_ => None,
}
}
/// Is `cid` a pointer (well-formed or not)?
pub fn is_pointer(cid: &str) -> bool {
decode(cid).is_some()
}
fn decode_nzb_release(cid: &str) -> Result<Pointer> {
let mscid = parse_record_cid(cid)?;
let locator = decode_nzb_release_cid(&mscid)?;
Ok(Pointer {
redeem: Some(RedeemKey {
codec: REDEEM_NZB_RELEASE,
key: normalize_indexer_host(locator.authority()),
}),
field: FIELD_MANIFEST,
kind: PointerKind::NzbManifest,
})
}
/// Decode a `provider-file` cid into `(source, id)`. Errors on a different
/// codec or a malformed payload. Works on the raw multibase string because a
/// long provider id can exceed the 64-byte multihash a typed cid parse allows.
pub fn decode_provider_file(cid: &str) -> Result<(String, String)> {
let bytes = raw_cid_bytes(cid).context("provider-file: not a base32 CIDv1")?;
let mut at = 0usize;
let next = |at: &mut usize, what: &str| -> Result<u64> {
let (v, n) = read_uvarint(&bytes[*at..])
.with_context(|| format!("provider-file: {what} varint"))?;
*at += n;
Ok(v)
};
anyhow::ensure!(next(&mut at, "version")? == 1, "provider-file: not a CIDv1");
anyhow::ensure!(next(&mut at, "codec")? == PROVIDER_FILE_CODEC, "not a provider-file cid");
let mh = next(&mut at, "multihash-code")?;
anyhow::ensure!(
mh == IDENTITY_MULTIHASH,
"provider-file locator must use the identity multihash (0x00), got 0x{mh:x}"
);
let len = next(&mut at, "digest-length")?;
anyhow::ensure!(
(bytes.len() - at) as u64 == len,
"provider-file digest length {len} != {} trailing byte(s)",
bytes.len() - at
);
let source_len = next(&mut at, "source-length")?;
let rest = &bytes[at..];
let source_len = usize::try_from(source_len)
.ok()
.filter(|n| *n <= rest.len())
.context("provider-file: source length overruns the digest")?;
let (source, id) = rest.split_at(source_len);
let source = std::str::from_utf8(source).context("provider-file source is not utf-8")?;
let id = std::str::from_utf8(id).context("provider-file id is not utf-8")?;
anyhow::ensure!(
!source.is_empty() && !id.is_empty(),
"provider-file locator needs both a source and an id"
);
Ok((source.to_string(), id.to_string()))
}
/// The codec of a base32 CIDv1 string, without the 64-byte multihash limit.
fn codec_of(cid: &str) -> Option<u64> {
let bytes = raw_cid_bytes(cid)?;
let (version, n) = read_uvarint(&bytes).ok()?;
if version != 1 {
return None;
}
read_uvarint(&bytes[n..]).ok().map(|(codec, _)| codec)
}
/// Decode a `b`-prefixed lowercase base32 (RFC 4648, no padding) CID string.
fn raw_cid_bytes(cid: &str) -> Option<Vec<u8>> {
let body = cid.strip_prefix('b')?;
let mut out = Vec::with_capacity(body.len() * 5 / 8);
let (mut acc, mut bits) = (0u32, 0u32);
for c in body.bytes() {
let v = match c {
b'a'..=b'z' => c - b'a',
b'2'..=b'7' => c - b'2' + 26,
_ => return None,
} as u32;
acc = (acc << 5) | v;
bits += 5;
if bits >= 8 {
bits -= 8;
out.push((acc >> bits) as u8);
acc &= (1 << bits) - 1;
}
}
Some(out)
}
#[cfg(test)]
mod tests {
use super::*;
/// The Ragna Crimson release from watch.nsl.sh (2026-10-04).
const NZBGEEK: &str = "bagcsaabfcrqxa2jonz5gez3fmvvs42lomzxs6ylqngxuch22nhfhl4v4vmyhiegah2ka";
fn b32(bytes: &[u8]) -> String {
const A: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567";
let (mut acc, mut bits, mut s) = (0u32, 0u32, String::from("b"));
for &b in bytes {
acc = (acc << 8) | b as u32;
bits += 8;
while bits >= 5 {
bits -= 5;
s.push(A[((acc >> bits) & 31) as usize] as char);
}
acc &= (1 << bits) - 1;
}
if bits > 0 {
s.push(A[((acc << (5 - bits)) & 31) as usize] as char);
}
s
}
fn provider_file(source: &str, id: &str) -> String {
let mut digest = vec![source.len() as u8];
digest.extend_from_slice(source.as_bytes());
digest.extend_from_slice(id.as_bytes());
// version 1, codec 0x100A (varint 0x8a 0x20), identity, length
let mut wire = vec![0x01, 0x8a, 0x20, 0x00, digest.len() as u8];
wire.extend_from_slice(&digest);
b32(&wire)
}
#[test]
fn an_nzb_release_is_a_pointer_keyed_by_its_indexer_host() {
let p = decode(NZBGEEK).unwrap().unwrap();
assert_eq!(
p.redeem,
Some(RedeemKey { codec: REDEEM_NZB_RELEASE, key: "api.nzbgeek.info".into() })
);
assert_eq!(p.field, FIELD_MANIFEST);
assert_eq!(p.kind, PointerKind::NzbManifest);
}
#[test]
fn a_provider_file_is_a_pointer_keyed_by_its_source() {
let cid = provider_file("opensubtitles", "file:7061834");
assert_eq!(
decode_provider_file(&cid).unwrap(),
("opensubtitles".into(), "file:7061834".into())
);
let p = decode(&cid).unwrap().unwrap();
assert_eq!(p.redeem.unwrap().key, "opensubtitles");
assert_eq!(p.field, FIELD_FILE);
assert_eq!(p.kind, PointerKind::File);
}
#[test]
fn a_content_cid_is_not_a_pointer() {
assert!(!is_pointer("bafkreigh2akiscaildcqabsyg3dfr6chu3fgpregiymsck7e7aqa4s52zy"));
assert!(!is_pointer("not a cid"));
}
}