meta-feeder-sdk 1.8.1: filename feature (filename_meta without the feeder stack)

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "meta-feeder-sdk"
-version = "1.4.0"
+version = "1.8.1"
 edition = "2021"
 rust-version = "1.85"
 license = "MIT"
@@ -32,48 +32,68 @@ publish = false
 # of here so feeders that don't need them don't pay for them.
 
 [features]
+default = ["feeder"]
+# Everything a feeder sidecar (and the transport contract) builds on: the
+# FeederPlugin trait, hashing, the redb cache, the axum harness. On by default
+# so existing consumers build exactly what they always did.
+feeder = [
+    "beacon", "filename",
+    "dep:axum", "dep:hyper", "dep:reqwest", "dep:base64", "dep:futures",
+    "dep:async-trait", "dep:anyhow", "dep:thiserror", "dep:sha2", "dep:bytes",
+    "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.
+beacon = ["dep:libc"]
 # The meta-share transport-plugin contract (`src/transport/`): the hull's
 # client, the plugins' axum harness, the focus replica, shared error/range
 # types. Off by default so feeders build exactly what they always did.
-transport = ["dep:cid", "dep:libc", "dep:tokio-util"]
+transport = ["feeder", "dep:cid", "dep:tokio-util"]
 # Conformance checks any transport implementation must pass.
 testkit = ["transport"]
 
 [dependencies]
 # Async runtime + HTTP server (the serve_feeders harness)
 tokio = { version = "1.40", features = ["full"] }
-axum = { version = "0.7", features = ["json"] }
-hyper = { version = "1", features = ["full"] }
-reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"], default-features = false }
+axum = { version = "0.7", features = ["json"], optional = true }
+hyper = { version = "1", features = ["full"], optional = true }
+reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"], default-features = false, optional = true }
 
 # Serialization
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-base64 = "0.22"
+base64 = { version = "0.22", optional = true }
 
 # Utilities
-futures = "0.3"
-async-trait = "0.1"
-anyhow = "1"
-thiserror = "1"
+futures = { version = "0.3", optional = true }
+async-trait = { version = "0.1", optional = true }
+anyhow = { version = "1", optional = true }
+thiserror = { version = "1", optional = true }
 tracing = "0.1"
-sha2 = "0.10"
-bytes = "1"
+sha2 = { version = "0.10", optional = true }
+bytes = { version = "1", optional = true }
 # Title-tag parsing (shared filename_meta — season/episode/quality/codec/languages).
-regex = "1"
+regex = { version = "1", optional = true }
 
 # Per-plugin midhash/blob cache. Schema is identical to meta-share v1 — see
 # cache.rs. Pure-Rust single-process embedded store.
-redb = "2"
+redb = { version = "2", optional = true }
 
 # CID parsing for the transport contract only (`transport::cid`). Optional so
 # feeders don't pull it.
 cid = { version = "0.11", optional = true }
-# statvfs(2) for `transport::fsutil::available_bytes`.
+# Raw sockets for `beacon` (per-interface multicast) and statvfs(2) for
+# `transport::fsutil::available_bytes`.
 libc = { version = "0.2", optional = true }
 # `ReaderStream` for `transport::file`.
 tokio-util = { version = "0.7", features = ["io"], optional = true }
 
 [dev-dependencies]
 tempfile = "3"
+tower = { version = "0.4", features = ["util"] }
 tokio = { version = "1.40", features = ["full", "test-util"] }

README.md

@@ -35,6 +35,42 @@ async fn main() -> anyhow::Result<()> {
 A **service is a `Vec`** — grouping plugins into one binary is a deployment
 choice, not an architectural one.
 
+### Discovery (1.6.0, scan 1.7.0)
+
+`serve_feeders` and `serve_transport` advertise the process over **beacon v2**
+(UDP multicast `239.255.99.1:9099`, spec: meta-root
+`docs/project-architecture/beacon-v2.md`) as soon as the port is bound — a
+feeder as `metamesh.feeder/<upstream_id>` (one cap per hosted plugin), a
+transport as `metamesh.transport/<id>@<contract>` — with `endpoints.http`,
+`endpoints.manifest = /manifest` and `rev` = a hash of the served manifest
+(re-advertised when it changes). Nothing to call. Env:
+
+| Env | Default | |
+|---|---|---|
+| `ENABLE_UDP_DISCOVERY` | `true` | `false` turns advertising off |
+| `BEACON_ADVERTISE_URL` | `http://$HOSTNAME:<port>` | `endpoints.http` |
+| `BEACON_BINDS` | unset | the one consumer instance allowed to list this plugin |
+| `BEACON_GROUP` / `BEACON_PORT` / `BEACON_INTERVAL_MS` | `239.255.99.1` / `9099` / `10000` | |
+
+`beacon::scan(node, pattern, configured, wait)` (1.7.0) is what a consumer's
+**Scan** button runs: probe everyone, wait, then classify every resource
+matching `pattern` as `configured` / `addable` / `bound-elsewhere` against the
+consumer's `(name, url)` list — the `ScanReport` JSON the shared
+`<meta-beacon-scan>` element renders. Transports advertise as
+`meta-transport-<id>` (`SERVICE_NAME` overrides), feeders as
+`meta-feeder-<first upstream>`.
+
+The `beacon` module is also usable on its own — the meta-* services depend on
+the SDK for nothing else:
+
+```toml
+meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.7.0", default-features = false, features = ["beacon"] }
+```
+
+Features: `feeder` (default — the whole feeder harness; implies `beacon`),
+`beacon` (tokio/serde/tracing/libc only), `transport` (implies `feeder`),
+`testkit`.
+
 ## Writing a plugin
 
 Four required methods; everything else has a default:
@@ -141,6 +177,7 @@ and `provider-file` (`0x100A`, `(source, id)` — `docs/cid-formats.md` §8).
 | `hash` | the content-addressing families above |
 | `filename_meta` / `lang` | release-tag parsing helpers |
 | `enrich`, `meta_core`, `common` | MetaMesh-internal; not part of the plugin contract |
+| `beacon` | beacon v2 local discovery: `BeaconNode`, `Resource`, `cap_matches`, the MetaMesh `caps` vocabulary |
 
 ## Development
 

src/beacon/mod.rs

@@ -0,0 +1,165 @@
+//! Beacon v2 — local resource advertise / discover over UDP multicast
+//! (`239.255.99.1:9099`).
+//!
+//! Every meta-* service and every SDK-hosted plugin runs one [`BeaconNode`]: it
+//! advertises this process's [`Resource`]s, each tagged with capability strings
+//! (`metamesh.service/meta-sort`, `metamesh.transport/nzb@1`, …), and keeps a
+//! live view of everyone else's, which consumers filter by capability
+//! ([`cap_matches`]).
+//!
+//! The normative spec is meta-root's `docs/project-architecture/beacon-v2.md`
+//! (the Go / TypeScript / Python ports are written against it too). This module
+//! needs only the `beacon` feature, so a service can depend on the SDK with
+//! `default-features = false, features = ["beacon"]` and pull in none of the
+//! feeder machinery.
+
+mod node;
+mod proto;
+mod scan;
+
+pub use node::{enabled_from_env, hostname, BeaconNode, Config, SeenNode, SeenResource};
+pub use scan::{classify, scan, suggested_name, Candidate, CandidateState, ScanReport, ScanSummary};
+pub use proto::{
+    cap_matches, rev_of, Message, NodeInfo, Resource, DEFAULT_GROUP, DEFAULT_INTERVAL, DEFAULT_PORT,
+    LIVENESS_FACTOR, MAX_DATAGRAM, PROTO, VERSION,
+};
+
+/// Capability vocabulary of the MetaMesh profile (spec Part 2).
+pub mod caps {
+    /// meta-core; its resource carries `data.urls`.
+    pub const CORE: &str = "metamesh.core";
+    /// Prefix of `metamesh.service/<name>`.
+    pub const SERVICE_PREFIX: &str = "metamesh.service/";
+    /// Every service.
+    pub const ANY_SERVICE: &str = "metamesh.service/*";
+    /// Prefix of `metamesh.transport/<id>@<contract>`.
+    pub const TRANSPORT_PREFIX: &str = "metamesh.transport/";
+    /// Every transport plugin.
+    pub const ANY_TRANSPORT: &str = "metamesh.transport/*";
+    /// Prefix of `metamesh.feeder/<upstream_id>`.
+    pub const FEEDER_PREFIX: &str = "metamesh.feeder/";
+    /// Every feeder upstream.
+    pub const ANY_FEEDER: &str = "metamesh.feeder/*";
+    /// Prefix of `metamesh.enrich/<plugin_id>` (meta-sort enrichment plugins).
+    pub const ENRICH_PREFIX: &str = "metamesh.enrich/";
+    /// Every meta-sort enrichment plugin.
+    pub const ANY_ENRICH: &str = "metamesh.enrich/*";
+
+    pub fn service(name: &str) -> String {
+        format!("{SERVICE_PREFIX}{name}")
+    }
+
+    pub fn transport(id: &str, contract: u32) -> String {
+        format!("{TRANSPORT_PREFIX}{id}@{contract}")
+    }
+
+    pub fn feeder(upstream_id: &str) -> String {
+        format!("{FEEDER_PREFIX}{upstream_id}")
+    }
+
+    pub fn enrich(plugin_id: &str) -> String {
+        format!("{ENRICH_PREFIX}{plugin_id}")
+    }
+}
+
+/// The resource every meta-* service advertises: id `service`, cap
+/// `metamesh.service/<name>`, `endpoints.ui` = its browser-facing URL.
+pub fn service_resource(name: &str, ui_url: Option<&str>) -> Resource {
+    let mut r = Resource::new("service", [caps::service(name)]);
+    if let Some(u) = ui_url {
+        r = r.with_endpoint("ui", u);
+    }
+    r
+}
+
+/// `BEACON_ADVERTISE_URL`, else `http://$HOSTNAME:<port>` — what a plugin
+/// advertises as `endpoints.http`.
+pub fn advertise_url(listen_port: u16) -> String {
+    std::env::var("BEACON_ADVERTISE_URL")
+        .ok()
+        .map(|s| s.trim().trim_end_matches('/').to_string())
+        .filter(|s| !s.is_empty())
+        .unwrap_or_else(|| format!("http://{}:{}", hostname(), listen_port))
+}
+
+/// `BEACON_BINDS`, when set.
+pub fn binds_from_env() -> Option<String> {
+    std::env::var("BEACON_BINDS").ok().map(|s| s.trim().to_string()).filter(|s| !s.is_empty())
+}
+
+/// Advertise a plugin process: spawn a [`BeaconNode`] named `name` whose single
+/// resource is whatever `build` returns, and re-run `build` every interval so a
+/// changed manifest (`rev`) is re-advertised at once. Used by both SDK
+/// harnesses; returns `None` (and logs) when discovery is disabled or the
+/// socket can't be set up — a plugin keeps serving either way.
+pub fn advertise_plugin<F>(name: &str, version: &str, build: F) -> Option<BeaconNode>
+where
+    F: Fn() -> Resource + Send + Sync + 'static,
+{
+    if !enabled_from_env() {
+        tracing::info!("beacon: disabled (ENABLE_UDP_DISCOVERY)");
+        return None;
+    }
+    let cfg = Config::from_env(name).version(version).resource(build());
+    let interval = cfg.interval;
+    let node = match BeaconNode::spawn(cfg) {
+        Ok(n) => n,
+        Err(e) => {
+            tracing::warn!(error = %e, "beacon: could not start; plugin will not be discoverable");
+            return None;
+        }
+    };
+    let refresher = node.clone();
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(interval);
+        tick.tick().await;
+        loop {
+            tick.tick().await;
+            refresher.set_resources(vec![build()]).await;
+        }
+    });
+    Some(node)
+}
+
+/// The `/api/neighbors` JSON every service serves (spec Part 2, "HTTP
+/// surface"). Each row is a [`SeenNode`] plus `caps` and a v1-compatible
+/// `baseUrl` (the `metamesh.service/*` resource's `endpoints.ui`).
+///
+/// `all` returns every instance instead of one row per name; `cap` keeps only
+/// nodes owning a resource that matches it.
+pub fn neighbors_json(
+    node: Option<&BeaconNode>,
+    current: &str,
+    all: bool,
+    cap: Option<&str>,
+) -> serde_json::Value {
+    let Some(node) = node else {
+        return serde_json::json!({
+            "current": current, "enabled": false, "count": 0, "neighbors": [],
+        });
+    };
+    let rows: Vec<serde_json::Value> = if all { node.nodes() } else { node.nodes_by_name() }
+        .into_iter()
+        .filter(|n| cap.map_or(true, |c| n.matches(c)))
+        .map(|n| neighbor_row(&n))
+        .collect();
+    serde_json::json!({
+        "current": current,
+        "enabled": true,
+        "count": rows.len(),
+        "self": neighbor_row(&node.self_node()),
+        "neighbors": rows,
+    })
+}
+
+fn neighbor_row(n: &SeenNode) -> serde_json::Value {
+    let mut v = serde_json::to_value(n).unwrap_or_default();
+    if let Some(obj) = v.as_object_mut() {
+        obj.insert("caps".into(), serde_json::json!(n.caps()));
+        let base = n.resource(caps::ANY_SERVICE).and_then(|r| r.endpoint("ui"));
+        if let Some(b) = base {
+            obj.insert("baseUrl".into(), serde_json::Value::String(b));
+        }
+    }
+    v
+}

src/beacon/node.rs

@@ -0,0 +1,719 @@
+//! `BeaconNode` — advertise this process's resources and keep a live view of
+//! everyone else's.
+//!
+//! Ported from meta-discovery v1's `meshdisco.rs`, keeping the two details that
+//! fail *silently* when wrong:
+//!
+//! 1. **Per-interface fan-out.** A socket that joins with `INADDR_ANY` and sends
+//!    via the default route covers exactly one interface. `metashare-app` sits
+//!    on `pcs` *and* `metamesh-mesh`; `metawatch-*` on `metawatch` *and*
+//!    `metamesh-mesh`. We enumerate interfaces and join/send on each, using
+//!    `ip_mreqn`'s `imr_ifindex`.
+//! 2. **Expiry at read time.** No reaper owns correctness; the announce loop's
+//!    prune only exists to fire change notifications for nodes that went quiet.
+//!
+//! Raw `libc` rather than `socket2` / `if-addrs`: every consumer already links
+//! libc, and this is the whole of what we need from either crate.
+
+use std::collections::HashMap;
+use std::io;
+use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
+use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, RwLock};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use serde::Serialize;
+use tokio::net::UdpSocket;
+use tokio::sync::watch;
+use tracing::{debug, info, warn};
+
+use super::proto::{
+    Message, NodeInfo, Resource, DEFAULT_GROUP, DEFAULT_INTERVAL, DEFAULT_PORT, LIVENESS_FACTOR,
+    MAX_DATAGRAM, READ_BUFFER, TYPE_ADVERTISE, TYPE_BYE, TYPE_PROBE,
+};
+
+/// How a node is set up. Build with [`Config::new`] / [`Config::from_env`].
+#[derive(Clone, Debug)]
+pub struct Config {
+    pub node: NodeInfo,
+    pub resources: Vec<Resource>,
+    pub group: String,
+    pub port: u16,
+    pub interval: Duration,
+}
+
+impl Config {
+    /// Defaults: this host's name as the instance, the standard group/port,
+    /// a 10 s interval, no resources.
+    pub fn new(name: impl Into<String>) -> Self {
+        Self {
+            node: NodeInfo {
+                name: name.into(),
+                instance: hostname(),
+                version: None,
+                status: None,
+            },
+            resources: Vec::new(),
+            group: DEFAULT_GROUP.into(),
+            port: DEFAULT_PORT,
+            interval: DEFAULT_INTERVAL,
+        }
+    }
+
+    /// [`Config::new`] with `BEACON_GROUP` / `BEACON_PORT` / `BEACON_INTERVAL_MS`
+    /// applied.
+    pub fn from_env(name: impl Into<String>) -> Self {
+        let mut c = Self::new(name);
+        if let Some(g) = env_nonempty("BEACON_GROUP") {
+            c.group = g;
+        }
+        if let Some(p) = env_nonempty("BEACON_PORT").and_then(|v| v.parse().ok()) {
+            c.port = p;
+        }
+        if let Some(ms) = env_nonempty("BEACON_INTERVAL_MS").and_then(|v| v.parse::<u64>().ok()) {
+            if ms > 0 {
+                c.interval = Duration::from_millis(ms);
+            }
+        }
+        c
+    }
+
+    pub fn version(mut self, v: impl Into<String>) -> Self {
+        self.node.version = Some(v.into());
+        self
+    }
+
+    pub fn instance(mut self, i: impl Into<String>) -> Self {
+        self.node.instance = i.into();
+        self
+    }
+
+    pub fn resource(mut self, r: Resource) -> Self {
+        self.resources.push(r);
+        self
+    }
+}
+
+/// A node heard on the wire (or this node itself, via [`BeaconNode::self_node`]).
+#[derive(Clone, Debug, Serialize)]
+pub struct SeenNode {
+    #[serde(flatten)]
+    pub node: NodeInfo,
+    pub resources: Vec<Resource>,
+    /// Source address taken from the packet, never from the payload.
+    pub addr: String,
+    /// Unix seconds.
+    #[serde(rename = "lastSeen")]
+    pub last_seen: f64,
+}
+
+impl SeenNode {
+    /// Every cap of every resource, in order, deduplicated.
+    pub fn caps(&self) -> Vec<String> {
+        let mut out: Vec<String> = Vec::new();
+        for c in self.resources.iter().flat_map(|r| r.caps.iter()) {
+            if !out.contains(c) {
+                out.push(c.clone());
+            }
+        }
+        out
+    }
+
+    /// True if any resource matches `pattern`.
+    pub fn matches(&self, pattern: &str) -> bool {
+        self.resources.iter().any(|r| r.matches(pattern))
+    }
+
+    /// The first resource matching `pattern`.
+    pub fn resource(&self, pattern: &str) -> Option<&Resource> {
+        self.resources.iter().find(|r| r.matches(pattern))
+    }
+}
+
+/// One resource together with the node advertising it.
+#[derive(Clone, Debug, Serialize)]
+pub struct SeenResource {
+    pub node: NodeInfo,
+    pub resource: Resource,
+    pub addr: String,
+    #[serde(rename = "lastSeen")]
+    pub last_seen: f64,
+}
+
+struct Inner {
+    group: Ipv4Addr,
+    port: u16,
+    interval: Duration,
+    node: RwLock<NodeInfo>,
+    resources: RwLock<Vec<Resource>>,
+    socket: UdpSocket,
+    ifindexes: Vec<(String, u32)>,
+    seen: Mutex<HashMap<String, SeenNode>>,
+    changes: watch::Sender<u64>,
+    stopped: AtomicBool,
+}
+
+/// A running beacon v2 participant. Cheap to clone.
+#[derive(Clone)]
+pub struct BeaconNode {
+    inner: Arc<Inner>,
+}
+
+impl BeaconNode {
+    /// Bind, join the group on every eligible interface, and start the read and
+    /// advertise loops. Must be called inside a tokio runtime.
+    ///
+    /// A join failure on one interface is logged and skipped rather than fatal —
+    /// a host that blocks multicast must degrade, not fail to boot.
+    pub fn spawn(cfg: Config) -> io::Result<BeaconNode> {
+        let group: Ipv4Addr = cfg.group.parse().map_err(|_| {
+            io::Error::new(io::ErrorKind::InvalidInput, format!("bad group {}", cfg.group))
+        })?;
+
+        let std_sock = bind_reuse(cfg.port)?;
+        set_multicast_ttl(&std_sock, 1)?;
+
+        let ifindexes = eligible_interfaces();
+        let mut joined = 0usize;
+        for (name, idx) in &ifindexes {
+            match join_group(&std_sock, group, *idx) {
+                Ok(()) => joined += 1,
+                Err(e) => warn!(iface = %name, error = %e, "beacon: join failed"),
+            }
+        }
+        info!(
+            group = %cfg.group, port = cfg.port, name = %cfg.node.name,
+            instance = %cfg.node.instance, resources = cfg.resources.len(),
+            joined, total = ifindexes.len(),
+            "beacon: listening"
+        );
+
+        std_sock.set_nonblocking(true)?;
+        let socket = UdpSocket::from_std(std_sock)?;
+        let (changes, _) = watch::channel(0u64);
+
+        let node = BeaconNode {
+            inner: Arc::new(Inner {
+                group,
+                port: cfg.port,
+                interval: cfg.interval,
+                node: RwLock::new(cfg.node),
+                resources: RwLock::new(cfg.resources),
+                socket,
+                ifindexes,
+                seen: Mutex::new(HashMap::new()),
+                changes,
+                stopped: AtomicBool::new(false),
+            }),
+        };
+
+        let reader = node.clone();
+        tokio::spawn(async move { reader.read_loop().await });
+
+        let advertiser = node.clone();
+        tokio::spawn(async move {
+            advertiser.advertise().await;
+            advertiser.probe(&[]).await;
+            let mut tick = tokio::time::interval(advertiser.inner.interval);
+            tick.tick().await; // the immediate first tick
+            loop {
+                tick.tick().await;
+                if advertiser.inner.stopped.load(Ordering::Relaxed) {
+                    break;
+                }
+                advertiser.advertise().await;
+                advertiser.prune();
+            }
+        });
+
+        Ok(node)
+    }
+
+    /// The instance name this node advertises as.
+    pub fn instance(&self) -> String {
+        self.inner.node.read().unwrap().instance.clone()
+    }
+
+    /// Replace this node's resources and re-advertise at once (e.g. a manifest
+    /// `rev` changed). A no-op if nothing changed.
+    pub async fn set_resources(&self, resources: Vec<Resource>) {
+        {
+            let mut cur = self.inner.resources.write().unwrap();
+            if *cur == resources {
+                return;
+            }
+            *cur = resources;
+        }
+        self.advertise().await;
+    }
+
+    /// Set `node.status` (`starting` / `running`) and re-advertise.
+    pub async fn set_status(&self, status: &str) {
+        {
+            let mut n = self.inner.node.write().unwrap();
+            if n.status.as_deref() == Some(status) {
+                return;
+            }
+            n.status = Some(status.to_string());
+        }
+        self.advertise().await;
+    }
+
+    /// Multicast a probe. Every node owning a resource matching one of `want`
+    /// (every node, when `want` is empty) replies at once — what makes the
+    /// protocol usable on a boot path instead of waiting out an interval.
+    pub async fn probe(&self, want: &[String]) {
+        let msg = Message::probe(&self.instance(), want);
+        self.multicast(&msg).await;
+    }
+
+    /// Say goodbye: multicast `bye` and stop advertising. Call on clean
+    /// shutdown so neighbours drop us at once instead of after the TTL.
+    pub async fn bye(&self) {
+        if self.inner.stopped.swap(true, Ordering::Relaxed) {
+            return;
+        }
+        let node = self.inner.node.read().unwrap().clone();
+        self.multicast(&Message::bye(node)).await;
+    }
+
+    /// Fires (the counter increments) whenever the set of live nodes or any
+    /// node's resources change — new node, changed advertise, bye, expiry.
+    /// Consumers re-read [`BeaconNode::resources`] on each change.
+    pub fn subscribe(&self) -> watch::Receiver<u64> {
+        self.inner.changes.subscribe()
+    }
+
+    /// Every live node, sorted by (name, instance).
+    pub fn nodes(&self) -> Vec<SeenNode> {
+        let cutoff = self.cutoff();
+        let guard = self.inner.seen.lock().unwrap();
+        let mut out: Vec<SeenNode> = guard.values().filter(|n| n.last_seen >= cutoff).cloned().collect();
+        drop(guard);
+        out.sort_by(|a, b| a.node.name.cmp(&b.node.name).then(a.node.instance.cmp(&b.node.instance)));
+        out
+    }
+
+    /// One row per node name, most recently seen instance wins — the nav
+    /// menu's shape.
+    pub fn nodes_by_name(&self) -> Vec<SeenNode> {
+        let mut best: HashMap<String, SeenNode> = HashMap::new();
+        for n in self.nodes() {
+            match best.get(&n.node.name) {
+                Some(cur) if cur.last_seen >= n.last_seen => {}
+                _ => {
+                    best.insert(n.node.name.clone(), n);
+                }
+            }
+        }
+        let mut out: Vec<SeenNode> = best.into_values().collect();
+        out.sort_by(|a, b| a.node.name.cmp(&b.node.name));
+        out
+    }
+
+    /// Every live resource matching `pattern`, with its node.
+    pub fn resources(&self, pattern: &str) -> Vec<SeenResource> {
+        self.nodes()
+            .into_iter()
+            .flat_map(|n| {
+                let SeenNode { node, resources, addr, last_seen } = n;
+                resources.into_iter().filter(|r| r.matches(pattern)).map(move |resource| SeenResource {
+                    node: node.clone(),
+                    resource,
+                    addr: addr.clone(),
+                    last_seen,
+                })
+            })
+            .collect()
+    }
+
+    /// This node as a row, so a UI can render itself without waiting for its
+    /// own echo.
+    pub fn self_node(&self) -> SeenNode {
+        SeenNode {
+            node: self.inner.node.read().unwrap().clone(),
+            resources: self.inner.resources.read().unwrap().clone(),
+            addr: String::new(),
+            last_seen: now_secs(),
+        }
+    }
+
+    // -- internals ----------------------------------------------------------
+
+    fn cutoff(&self) -> f64 {
+        now_secs() - self.inner.interval.as_secs_f64() * LIVENESS_FACTOR as f64
+    }
+
+    fn bump(&self) {
+        self.inner.changes.send_modify(|g| *g = g.wrapping_add(1));
+    }
+
+    /// Drop expired nodes; notify if any went.
+    fn prune(&self) {
+        let cutoff = self.cutoff();
+        let removed = {
+            let mut guard = self.inner.seen.lock().unwrap();
+            let before = guard.len();
+            guard.retain(|_, n| n.last_seen >= cutoff);
+            before != guard.len()
+        };
+        if removed {
+            self.bump();
+        }
+    }
+
+    fn advertise_msg(&self) -> Message {
+        Message::advertise(
+            self.inner.node.read().unwrap().clone(),
+            self.inner.resources.read().unwrap().clone(),
+        )
+    }
+
+    async fn advertise(&self) {
+        if self.inner.stopped.load(Ordering::Relaxed) {
+            return;
+        }
+        let msg = self.advertise_msg();
+        self.multicast(&msg).await;
+    }
+
+    fn encode(msg: &Message) -> Option<Vec<u8>> {
+        match serde_json::to_vec(msg) {
+            Ok(b) => {
+                if b.len() > MAX_DATAGRAM {
+                    warn!(bytes = b.len(), max = MAX_DATAGRAM, "beacon: datagram over the fragmentation-safe size");
+                }
+                Some(b)
+            }
+            Err(e) => {
+                warn!(error = %e, "beacon: encode failed");
+                None
+            }
+        }
+    }
+
+    /// Writes once per interface. Letting the routing table choose would reach
+    /// exactly one network — the failure this protocol exists to avoid.
+    async fn multicast(&self, msg: &Message) {
+        let Some(body) = Self::encode(msg) else { return };
+        let dst = SocketAddr::V4(SocketAddrV4::new(self.inner.group, self.inner.port));
+        for (name, idx) in &self.inner.ifindexes {
+            if let Err(e) = set_multicast_if(&self.inner.socket, *idx) {
+                warn!(iface = %name, error = %e, "beacon: select interface failed");
+                continue;
+            }
+            if let Err(e) = self.inner.socket.send_to(&body, dst).await {
+                warn!(iface = %name, error = %e, "beacon: send failed");
+            }
+        }
+    }
+
+    async fn reply_to(&self, dst: SocketAddr) {
+        let Some(body) = Self::encode(&self.advertise_msg()) else { return };
+        if let Err(e) = self.inner.socket.send_to(&body, dst).await {
+            debug!(error = %e, "beacon: reply failed");
+        }
+    }
+
+    async fn read_loop(self) {
+        let mut buf = vec![0u8; READ_BUFFER];
+        loop {
+            let (n, src) = match self.inner.socket.recv_from(&mut buf).await {
+                Ok(v) => v,
+                Err(e) => {
+                    debug!(error = %e, "beacon: read error");
+                    continue;
+                }
+            };
+            let Some(msg) = Message::parse(&buf[..n]) else { continue };
+            let me = self.instance();
+
+            match msg.msg_type.as_str() {
+                TYPE_PROBE => {
+                    if msg.from.as_deref() == Some(me.as_str()) || self.inner.stopped.load(Ordering::Relaxed) {
+                        continue;
+                    }
+                    let wanted = msg.want.is_empty()
+                        || self.inner.resources.read().unwrap().iter().any(|r| r.matches_any(&msg.want));
+                    if wanted {
+                        self.reply_to(src).await;
+                    }
+                }
+                TYPE_ADVERTISE | TYPE_BYE => {
+                    let Some(node) = msg.node else { continue };
+                    if node.instance == me {
+                        continue; // our own multicast echo
+                    }
+                    let changed = if msg.msg_type == TYPE_BYE {
+                        self.inner.seen.lock().unwrap().remove(&node.instance).is_some()
+                    } else {
+                        let cutoff = self.cutoff();
+                        let mut guard = self.inner.seen.lock().unwrap();
+                        let changed = match guard.get(&node.instance) {
+                            Some(prev) => {
+                                prev.last_seen < cutoff || prev.node != node || prev.resources != msg.resources
+                            }
+                            None => true,
+                        };
+                        guard.insert(
+                            node.instance.clone(),
+                            SeenNode {
+                                node,
+                                resources: msg.resources,
+                                addr: src.ip().to_string(),
+                                last_seen: now_secs(),
+                            },
+                        );
+                        changed
+                    };
+                    if changed {
+                        self.bump();
+                    }
+                }
+                _ => {}
+            }
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Socket plumbing.
+// ---------------------------------------------------------------------------
+
+fn bind_reuse(port: u16) -> io::Result<std::net::UdpSocket> {
+    unsafe {
+        let fd: RawFd = libc::socket(libc::AF_INET, libc::SOCK_DGRAM, 0);
+        if fd < 0 {
+            return Err(io::Error::last_os_error());
+        }
+        // Must be set before bind. Lets several listeners share the port on one
+        // host (a test, a sidecar, a beacon v1 responder in the same netns).
+        let one: libc::c_int = 1;
+        for opt in [libc::SO_REUSEADDR, libc::SO_REUSEPORT] {
+            libc::setsockopt(
+                fd,
+                libc::SOL_SOCKET,
+                opt,
+                &one as *const _ as *const libc::c_void,
+                std::mem::size_of::<libc::c_int>() as libc::socklen_t,
+            );
+        }
+        let addr = libc::sockaddr_in {
+            sin_family: libc::AF_INET as libc::sa_family_t,
+            sin_port: port.to_be(),
+            sin_addr: libc::in_addr { s_addr: libc::INADDR_ANY.to_be() },
+            sin_zero: [0; 8],
+        };
+        if libc::bind(
+            fd,
+            &addr as *const _ as *const libc::sockaddr,
+            std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
+        ) < 0
+        {
+            let err = io::Error::last_os_error();
+            libc::close(fd);
+            return Err(err);
+        }
+        Ok(std::net::UdpSocket::from_raw_fd(fd))
+    }
+}
+
+fn set_multicast_ttl(sock: &std::net::UdpSocket, ttl: libc::c_int) -> io::Result<()> {
+    // TTL 1: link-local only, never routed off the segment.
+    let rc = unsafe {
+        libc::setsockopt(
+            sock.as_raw_fd(),
+            libc::IPPROTO_IP,
+            libc::IP_MULTICAST_TTL,
+            &ttl as *const _ as *const libc::c_void,
+            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
+        )
+    };
+    if rc < 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+/// `ip_mreqn` — `imr_ifindex` is what makes this a per-interface join instead
+/// of "whatever the routing table picks".
+fn mreqn(group: Ipv4Addr, ifindex: u32) -> libc::ip_mreqn {
+    libc::ip_mreqn {
+        imr_multiaddr: libc::in_addr { s_addr: u32::from(group).to_be() },
+        imr_address: libc::in_addr { s_addr: libc::INADDR_ANY.to_be() },
+        imr_ifindex: ifindex as libc::c_int,
+    }
+}
+
+fn join_group(sock: &std::net::UdpSocket, group: Ipv4Addr, ifindex: u32) -> io::Result<()> {
+    let req = mreqn(group, ifindex);
+    let rc = unsafe {
+        libc::setsockopt(
+            sock.as_raw_fd(),
+            libc::IPPROTO_IP,
+            libc::IP_ADD_MEMBERSHIP,
+            &req as *const _ as *const libc::c_void,
+            std::mem::size_of::<libc::ip_mreqn>() as libc::socklen_t,
+        )
+    };
+    if rc < 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+fn set_multicast_if(sock: &UdpSocket, ifindex: u32) -> io::Result<()> {
+    let req = mreqn(Ipv4Addr::UNSPECIFIED, ifindex);
+    let rc = unsafe {
+        libc::setsockopt(
+            sock.as_raw_fd(),
+            libc::IPPROTO_IP,
+            libc::IP_MULTICAST_IF,
+            &req as *const _ as *const libc::c_void,
+            std::mem::size_of::<libc::ip_mreqn>() as libc::socklen_t,
+        )
+    };
+    if rc < 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+/// Every interface except loopback, as (name, ifindex).
+fn eligible_interfaces() -> Vec<(String, u32)> {
+    let mut out = Vec::new();
+    unsafe {
+        let list = libc::if_nameindex();
+        if list.is_null() {
+            warn!("beacon: could not enumerate interfaces");
+            return out;
+        }
+        let mut p = list;
+        while (*p).if_index != 0 && !(*p).if_name.is_null() {
+            let name = std::ffi::CStr::from_ptr((*p).if_name).to_string_lossy().into_owned();
+            if name != "lo" {
+                out.push((name, (*p).if_index));
+            }
+            p = p.add(1);
+        }
+        libc::if_freenameindex(list);
+    }
+    out
+}
+
+fn env_nonempty(key: &str) -> Option<String> {
+    std::env::var(key).ok().map(|s| s.trim().to_string()).filter(|s| !s.is_empty())
+}
+
+/// `$HOSTNAME`, else `/etc/hostname`, else `unknown`.
+pub fn hostname() -> String {
+    env_nonempty("HOSTNAME")
+        .or_else(|| {
+            std::fs::read_to_string("/etc/hostname")
+                .ok()
+                .map(|s| s.trim().to_string())
+                .filter(|s| !s.is_empty())
+        })
+        .unwrap_or_else(|| "unknown".into())
+}
+
+pub(crate) fn now_secs() -> f64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs_f64())
+        .unwrap_or(0.0)
+}
+
+/// `ENABLE_UDP_DISCOVERY`, permissive: only `false`/`0`/`no`/`off` disable.
+pub fn enabled_from_env() -> bool {
+    match std::env::var("ENABLE_UDP_DISCOVERY") {
+        Ok(v) => !matches!(v.trim().to_ascii_lowercase().as_str(), "false" | "0" | "no" | "off"),
+        Err(_) => true,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Two nodes on loopback-only hosts can't multicast to each other, so
+    /// exercise the receive path directly by unicasting into a node's port.
+    async fn node_on(port: u16, instance: &str, resources: Vec<Resource>) -> BeaconNode {
+        let mut cfg = Config::new("test").instance(instance);
+        cfg.port = port;
+        cfg.interval = Duration::from_millis(200);
+        cfg.resources = resources;
+        BeaconNode::spawn(cfg).unwrap()
+    }
+
+    async fn send(port: u16, msg: &Message) {
+        let s = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        s.send_to(&serde_json::to_vec(msg).unwrap(), ("127.0.0.1", port)).await.unwrap();
+    }
+
+    fn free_port() -> u16 {
+        std::net::UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()
+    }
+
+    fn info(instance: &str) -> NodeInfo {
+        NodeInfo { name: "peer".into(), instance: instance.into(), ..Default::default() }
+    }
+
+    #[tokio::test]
+    async fn advertise_bye_and_expiry() {
+        let port = free_port();
+        let node = node_on(port, "me", vec![]).await;
+        let mut rx = node.subscribe();
+
+        let r = Resource::new("nzb", ["metamesh.transport/nzb@1"]);
+        send(port, &Message::advertise(info("peer-1"), vec![r])).await;
+        tokio::time::timeout(Duration::from_secs(2), rx.changed()).await.unwrap().unwrap();
+        assert_eq!(node.resources("metamesh.transport/*").len(), 1);
+        assert!(node.resources("metamesh.core").is_empty());
+
+        send(port, &Message::bye(info("peer-1"))).await;
+        tokio::time::timeout(Duration::from_secs(2), rx.changed()).await.unwrap().unwrap();
+        assert!(node.nodes().is_empty());
+
+        send(port, &Message::advertise(info("peer-2"), vec![])).await;
+        tokio::time::timeout(Duration::from_secs(2), rx.changed()).await.unwrap().unwrap();
+        assert_eq!(node.nodes().len(), 1);
+        // 3 × 200 ms liveness, then the prune on the next tick notifies.
+        tokio::time::timeout(Duration::from_secs(3), rx.changed()).await.unwrap().unwrap();
+        assert!(node.nodes().is_empty());
+    }
+
+    #[tokio::test]
+    async fn own_echo_and_v1_ignored() {
+        let port = free_port();
+        let node = node_on(port, "me", vec![]).await;
+        send(port, &Message::advertise(info("me"), vec![])).await;
+        let s = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        s.send_to(br#"{"type":"announce","name":"x","tools":[]}"#, ("127.0.0.1", port)).await.unwrap();
+        s.send_to(br#"{"v":1,"type":"announce","name":"meta-sort","instance":"a"}"#, ("127.0.0.1", port))
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        assert!(node.nodes().is_empty());
+    }
+
+    #[tokio::test]
+    async fn probe_reply_honours_want() {
+        let port = free_port();
+        let _node = node_on(port, "me", vec![Resource::new("service", ["metamesh.service/meta-sort"])]).await;
+        let s = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let mut buf = vec![0u8; READ_BUFFER];
+
+        let miss = Message::probe("cli", &["metamesh.core".to_string()]);
+        s.send_to(&serde_json::to_vec(&miss).unwrap(), ("127.0.0.1", port)).await.unwrap();
+        assert!(tokio::time::timeout(Duration::from_millis(300), s.recv_from(&mut buf)).await.is_err());
+
+        let hit = Message::probe("cli", &["metamesh.service/*".to_string()]);
+        s.send_to(&serde_json::to_vec(&hit).unwrap(), ("127.0.0.1", port)).await.unwrap();
+        let (n, _) = tokio::time::timeout(Duration::from_secs(2), s.recv_from(&mut buf)).await.unwrap().unwrap();
+        let m = Message::parse(&buf[..n]).unwrap();
+        assert_eq!(m.msg_type, TYPE_ADVERTISE);
+        assert_eq!(m.node.unwrap().instance, "me");
+    }
+}

src/beacon/proto.rs

@@ -0,0 +1,312 @@
+//! Beacon v2 wire types and capability matching.
+//!
+//! The normative spec is meta-root's `docs/project-architecture/beacon-v2.md`.
+
+use std::collections::BTreeMap;
+use std::hash::{Hash, Hasher};
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+
+pub const PROTO: &str = "beacon";
+pub const VERSION: u8 = 2;
+/// Beacon v1's group and port, on purpose — one well-known endpoint for every
+/// kind of local resource. v1 and v2 coexist because v2 never sends v1's
+/// `discovery` / `announce` types.
+pub const DEFAULT_GROUP: &str = "239.255.99.1";
+pub const DEFAULT_PORT: u16 = 9099;
+pub const DEFAULT_INTERVAL: Duration = Duration::from_secs(10);
+/// A node unheard for `interval * this` is dropped (evaluated at read time).
+pub const LIVENESS_FACTOR: u32 = 3;
+/// Receive buffer. Senders stay under [`MAX_DATAGRAM`].
+pub const READ_BUFFER: usize = 8192;
+/// Above this a datagram risks IP fragmentation; the sender logs a warning.
+pub const MAX_DATAGRAM: usize = 1400;
+
+pub const TYPE_PROBE: &str = "probe";
+pub const TYPE_ADVERTISE: &str = "advertise";
+pub const TYPE_BYE: &str = "bye";
+
+/// Who is advertising.
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct NodeInfo {
+    /// What the software is (`meta-sort`, `meta-transport-nzb`).
+    pub name: String,
+    /// Unique per running copy — the container hostname.
+    pub instance: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub version: Option<String>,
+    /// `starting` | `running`. Absent reads as `running`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub status: Option<String>,
+}
+
+/// One advertised resource.
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
+pub struct Resource {
+    /// Unique within the node.
+    pub id: String,
+    pub caps: Vec<String>,
+    /// Well-known names: `http`, `ui`, `manifest`, `mcp`. A value starting with
+    /// `/` is relative to `http` — see [`Resource::endpoint`].
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub endpoints: BTreeMap<String, String>,
+    /// Opaque revision of the resource's details (e.g. its manifest).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub rev: Option<String>,
+    /// The one consumer instance this resource belongs to. Absent = anyone.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub binds: Option<String>,
+    /// Profile-defined, per capability.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub data: Option<serde_json::Value>,
+}
+
+impl Resource {
+    pub fn new(id: impl Into<String>, caps: impl IntoIterator<Item = impl Into<String>>) -> Self {
+        Self {
+            id: id.into(),
+            caps: caps.into_iter().map(Into::into).collect(),
+            ..Default::default()
+        }
+    }
+
+    pub fn with_endpoint(mut self, name: impl Into<String>, url: impl Into<String>) -> Self {
+        let url = url.into();
+        if !url.is_empty() {
+            self.endpoints.insert(name.into(), url);
+        }
+        self
+    }
+
+    /// True if any of this resource's caps matches `pattern`.
+    pub fn matches(&self, pattern: &str) -> bool {
+        self.caps.iter().any(|c| cap_matches(pattern, c))
+    }
+
+    /// True if any cap matches any of `patterns`. An empty list matches.
+    pub fn matches_any(&self, patterns: &[String]) -> bool {
+        patterns.is_empty() || patterns.iter().any(|p| self.matches(p))
+    }
+
+    /// `binds` is absent, or names `instance`.
+    pub fn usable_by(&self, instance: &str) -> bool {
+        self.binds.as_deref().map_or(true, |b| b == instance)
+    }
+
+    /// The named endpoint as an absolute URL: a value starting with `/` is
+    /// joined onto `endpoints.http`.
+    pub fn endpoint(&self, name: &str) -> Option<String> {
+        let v = self.endpoints.get(name)?;
+        if v.starts_with('/') {
+            let base = self.endpoints.get("http")?;
+            Some(format!("{}{}", base.trim_end_matches('/'), v))
+        } else {
+            Some(v.clone())
+        }
+    }
+}
+
+/// Every v2 datagram.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Message {
+    pub proto: String,
+    pub v: u8,
+    #[serde(rename = "type")]
+    pub msg_type: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub node: Option<NodeInfo>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub resources: Vec<Resource>,
+    /// Probe only: informational sender instance.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub from: Option<String>,
+    /// Probe only: reply only if a resource matches one of these.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub want: Vec<String>,
+}
+
+impl Message {
+    fn base(msg_type: &str) -> Self {
+        Self {
+            proto: PROTO.into(),
+            v: VERSION,
+            msg_type: msg_type.into(),
+            node: None,
+            resources: Vec::new(),
+            from: None,
+            want: Vec::new(),
+        }
+    }
+
+    pub fn probe(from: &str, want: &[String]) -> Self {
+        Self {
+            from: Some(from.to_string()),
+            want: want.to_vec(),
+            ..Self::base(TYPE_PROBE)
+        }
+    }
+
+    pub fn advertise(node: NodeInfo, resources: Vec<Resource>) -> Self {
+        Self {
+            node: Some(node),
+            resources,
+            ..Self::base(TYPE_ADVERTISE)
+        }
+    }
+
+    pub fn bye(node: NodeInfo) -> Self {
+        Self {
+            node: Some(node),
+            ..Self::base(TYPE_BYE)
+        }
+    }
+
+    /// Decode a datagram, returning `None` for anything that is not a
+    /// well-formed beacon v2 message — beacon v1 traffic (no envelope), other
+    /// versions, unknown types, or an advertise/bye without a node. The group
+    /// is shared, so all of that is silently ignored.
+    pub fn parse(bytes: &[u8]) -> Option<Self> {
+        let m: Message = serde_json::from_slice(bytes).ok()?;
+        if m.proto != PROTO || m.v != VERSION {
+            return None;
+        }
+        match m.msg_type.as_str() {
+            TYPE_PROBE => Some(m),
+            TYPE_ADVERTISE | TYPE_BYE => {
+                let n = m.node.as_ref()?;
+                if n.instance.is_empty() || n.name.is_empty() {
+                    return None;
+                }
+                Some(m)
+            }
+            _ => None,
+        }
+    }
+}
+
+/// Split `base@contract`. A non-numeric or missing contract is `None`.
+fn split_contract(s: &str) -> (&str, Option<&str>) {
+    match s.rsplit_once('@') {
+        Some((base, c)) => (base, Some(c)),
+        None => (s, None),
+    }
+}
+
+/// Does capability `cap` satisfy `pattern`?
+///
+/// - `*` matches everything.
+/// - `@N` on the pattern must equal the cap's contract; without one the cap's
+///   contract is ignored.
+/// - `x/*` matches any variant of `x`; otherwise the bases must be equal.
+pub fn cap_matches(pattern: &str, cap: &str) -> bool {
+    if pattern == "*" {
+        return true;
+    }
+    let (pbase, pcontract) = split_contract(pattern);
+    let (cbase, ccontract) = split_contract(cap);
+    if let Some(pc) = pcontract {
+        if ccontract != Some(pc) {
+            return false;
+        }
+    }
+    match pbase.strip_suffix("/*") {
+        Some(prefix) => cbase
+            .strip_prefix(prefix)
+            .is_some_and(|rest| rest.starts_with('/') && rest.len() > 1),
+        None => pbase == cbase,
+    }
+}
+
+/// A short opaque revision for any serialisable value — what a harness puts in
+/// [`Resource::rev`] so consumers re-fetch a manifest only when it changed.
+/// Stable for a given build; not cryptographic and not meant to be.
+pub fn rev_of<T: Serialize>(value: &T) -> String {
+    let bytes = serde_json::to_vec(value).unwrap_or_default();
+    let mut h = std::collections::hash_map::DefaultHasher::new();
+    bytes.hash(&mut h);
+    format!("{:016x}", h.finish())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn wildcard_matches_any_variant_only() {
+        assert!(cap_matches("metamesh.transport/*", "metamesh.transport/nzb@1"));
+        assert!(cap_matches("metamesh.transport/*", "metamesh.transport/ipfs"));
+        assert!(!cap_matches("metamesh.transport/*", "metamesh.transport"));
+        assert!(!cap_matches("metamesh.transport/*", "metamesh.transportx/a"));
+        assert!(!cap_matches("metamesh.transport/*", "metamesh.transport/"));
+        assert!(cap_matches("*", "anything@3"));
+    }
+
+    #[test]
+    fn contract_rules() {
+        assert!(cap_matches("metamesh.transport/nzb", "metamesh.transport/nzb@1"));
+        assert!(cap_matches("metamesh.transport/nzb", "metamesh.transport/nzb@2"));
+        assert!(cap_matches("metamesh.transport/nzb@2", "metamesh.transport/nzb@2"));
+        assert!(!cap_matches("metamesh.transport/nzb@2", "metamesh.transport/nzb@1"));
+        assert!(!cap_matches("metamesh.transport/nzb@2", "metamesh.transport/nzb"));
+        assert!(cap_matches("metamesh.transport/*@1", "metamesh.transport/nzb@1"));
+        assert!(!cap_matches("metamesh.transport/*@1", "metamesh.transport/nzb@2"));
+    }
+
+    #[test]
+    fn exact_match() {
+        assert!(cap_matches("mcp", "mcp"));
+        assert!(!cap_matches("mcp", "mcp/x"));
+        assert!(!cap_matches("metamesh.core", "metamesh.core2"));
+        assert!(!cap_matches("metamesh.transport/nzb", "metamesh.transport/ipfs@1"));
+    }
+
+    #[test]
+    fn parse_rejects_v1_and_foreign_traffic() {
+        // beacon v1 aggregator probe / server announce
+        assert!(Message::parse(br#"{"type":"discovery"}"#).is_none());
+        assert!(Message::parse(br#"{"type":"announce","name":"x","tools":[]}"#).is_none());
+        // meta-discovery v1
+        assert!(Message::parse(br#"{"v":1,"type":"announce","name":"meta-sort"}"#).is_none());
+        // wrong version / unknown type / not json
+        assert!(Message::parse(br#"{"proto":"beacon","v":3,"type":"probe"}"#).is_none());
+        assert!(Message::parse(br#"{"proto":"beacon","v":2,"type":"announce"}"#).is_none());
+        assert!(Message::parse(b"\x00\x01garbage").is_none());
+        // advertise without a node
+        assert!(Message::parse(br#"{"proto":"beacon","v":2,"type":"advertise"}"#).is_none());
+    }
+
+    #[test]
+    fn roundtrip_advertise() {
+        let r = Resource::new("nzb", ["metamesh.transport/nzb@1"])
+            .with_endpoint("http", "http://metashare-nzb:3000")
+            .with_endpoint("manifest", "/manifest");
+        let m = Message::advertise(
+            NodeInfo { name: "meta-transport-nzb".into(), instance: "metashare-nzb".into(), ..Default::default() },
+            vec![r],
+        );
+        let bytes = serde_json::to_vec(&m).unwrap();
+        let back = Message::parse(&bytes).unwrap();
+        assert_eq!(back.msg_type, TYPE_ADVERTISE);
+        let r = &back.resources[0];
+        assert_eq!(r.endpoint("manifest").as_deref(), Some("http://metashare-nzb:3000/manifest"));
+        assert!(r.matches("metamesh.transport/*"));
+    }
+
+    #[test]
+    fn binds() {
+        let mut r = Resource::new("nzb", ["metamesh.transport/nzb@1"]);
+        assert!(r.usable_by("anyone"));
+        r.binds = Some("metashare-app".into());
+        assert!(r.usable_by("metashare-app"));
+        assert!(!r.usable_by("metawatch-share"));
+    }
+
+    #[test]
+    fn probe_filter() {
+        let r = Resource::new("service", ["metamesh.service/meta-sort"]);
+        assert!(r.matches_any(&[]));
+        assert!(r.matches_any(&["metamesh.service/*".into()]));
+        assert!(!r.matches_any(&["metamesh.core".into()]));
+    }
+}

src/beacon/scan.rs

@@ -0,0 +1,234 @@
+//! On-demand discovery scan — what a consumer's "Scan" button runs.
+//!
+//! Probe everyone, wait a moment for the replies, then classify every resource
+//! matching the consumer's capability pattern against what the consumer has
+//! already configured. The report is the same JSON in every consumer
+//! (meta-share, meta-gateway, meta-sort), rendered by the shared
+//! `<meta-beacon-scan>` element.
+
+use std::time::{Duration, Instant};
+
+use serde::Serialize;
+
+use super::node::{BeaconNode, SeenResource};
+
+/// How a discovered resource relates to this consumer.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum CandidateState {
+    /// Its URL is already in the consumer's list.
+    Configured,
+    /// Can be added.
+    Addable,
+    /// `binds` names another consumer; shown but not addable.
+    BoundElsewhere,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Candidate {
+    pub instance: String,
+    pub name: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub version: Option<String>,
+    pub resource_id: String,
+    pub caps: Vec<String>,
+    /// `endpoints.http` — what the consumer would register.
+    pub url: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub binds: Option<String>,
+    pub state: CandidateState,
+    /// The list entry that already holds this URL (state `configured`).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub configured_as: Option<String>,
+    /// A list-safe name to register it under.
+    pub suggested_name: String,
+}
+
+#[derive(Clone, Debug, Default, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ScanSummary {
+    /// Every node heard on the segment, whatever it advertises.
+    pub nodes: usize,
+    /// Resources matching the consumer's pattern.
+    pub capable: usize,
+    pub configured: usize,
+    pub bound_elsewhere: usize,
+    pub addable: usize,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ScanReport {
+    pub cap: String,
+    /// This consumer's instance (what `binds` is compared against).
+    #[serde(rename = "self")]
+    pub self_instance: String,
+    pub duration_ms: u64,
+    pub summary: ScanSummary,
+    pub candidates: Vec<Candidate>,
+}
+
+fn norm(u: &str) -> String {
+    u.trim().trim_end_matches('/').to_string()
+}
+
+/// Lower-case, `[a-z0-9._-]` only — accepted by every consumer's name rules.
+pub fn suggested_name(instance: &str) -> String {
+    let s: String = instance
+        .trim()
+        .to_ascii_lowercase()
+        .chars()
+        .map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' { c } else { '-' })
+        .collect();
+    if s.is_empty() { "plugin".into() } else { s }
+}
+
+/// Pure classification, for tests and for callers that already hold a view.
+/// `configured` is `(name, url)` of every entry the consumer has.
+pub fn classify(
+    pattern: &str,
+    self_instance: &str,
+    nodes: usize,
+    resources: Vec<SeenResource>,
+    configured: &[(String, String)],
+) -> ScanReport {
+    let mut summary = ScanSummary { nodes, ..Default::default() };
+    let mut candidates = Vec::new();
+    for r in resources {
+        // A resource without an http endpoint cannot be registered anywhere.
+        let Some(url) = r.resource.endpoint("http") else { continue };
+        let url = norm(&url);
+        summary.capable += 1;
+        let configured_as = configured.iter().find(|(_, u)| norm(u) == url).map(|(n, _)| n.clone());
+        let state = if configured_as.is_some() {
+            summary.configured += 1;
+            CandidateState::Configured
+        } else if !r.resource.usable_by(self_instance) {
+            summary.bound_elsewhere += 1;
+            CandidateState::BoundElsewhere
+        } else {
+            summary.addable += 1;
+            CandidateState::Addable
+        };
+        candidates.push(Candidate {
+            suggested_name: suggested_name(&r.node.instance),
+            instance: r.node.instance,
+            name: r.node.name,
+            version: r.node.version,
+            resource_id: r.resource.id,
+            caps: r.resource.caps,
+            url,
+            binds: r.resource.binds,
+            state,
+            configured_as,
+        });
+    }
+    // Addable first (what the operator came for), then configured, then the rest.
+    let rank = |s: CandidateState| match s {
+        CandidateState::Addable => 0,
+        CandidateState::Configured => 1,
+        CandidateState::BoundElsewhere => 2,
+    };
+    candidates.sort_by(|a, b| rank(a.state).cmp(&rank(b.state)).then(a.instance.cmp(&b.instance)));
+    ScanReport {
+        cap: pattern.to_string(),
+        self_instance: self_instance.to_string(),
+        duration_ms: 0,
+        summary,
+        candidates,
+    }
+}
+
+/// Probe every node, wait `wait` for the replies, then classify what matches
+/// `pattern`. A node that stays silent through the wait but is still inside
+/// its liveness window is counted too — the view is "who is here", not "who
+/// answered this probe".
+pub async fn scan(
+    node: &BeaconNode,
+    pattern: &str,
+    configured: &[(String, String)],
+    wait: Duration,
+) -> ScanReport {
+    let started = Instant::now();
+    node.probe(&[]).await;
+    tokio::time::sleep(wait).await;
+    let mut report = classify(
+        pattern,
+        &node.instance(),
+        node.nodes().len(),
+        node.resources(pattern),
+        configured,
+    );
+    report.duration_ms = started.elapsed().as_millis() as u64;
+    report
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::beacon::{NodeInfo, Resource};
+
+    fn seen(instance: &str, cap: &str, url: Option<&str>, binds: Option<&str>) -> SeenResource {
+        let mut r = Resource::new("x", [cap]);
+        if let Some(u) = url {
+            r = r.with_endpoint("http", u);
+        }
+        r.binds = binds.map(Into::into);
+        SeenResource {
+            node: NodeInfo { name: "plugin".into(), instance: instance.into(), ..Default::default() },
+            resource: r,
+            addr: "10.0.0.1".into(),
+            last_seen: 0.0,
+        }
+    }
+
+    #[test]
+    fn classifies_configured_addable_and_bound_elsewhere() {
+        let resources = vec![
+            seen("metashare-nzb", "metamesh.transport/nzb@1", Some("http://metashare-nzb:3000/"), Some("metashare-app")),
+            seen("qbit", "metamesh.transport/torrent@1", Some("http://qbit:3000"), None),
+            seen("other-ipfs", "metamesh.transport/ipfs@1", Some("http://other-ipfs:3000"), Some("metawatch-share")),
+            seen("no-endpoint", "metamesh.transport/ipfs@1", None, None),
+        ];
+        let configured = vec![("nzb".to_string(), "http://metashare-nzb:3000".to_string())];
+        let r = classify("metamesh.transport/*", "metashare-app", 10, resources, &configured);
+        assert_eq!(r.summary.nodes, 10);
+        assert_eq!(r.summary.capable, 3, "a resource without an http endpoint is not a candidate");
+        assert_eq!((r.summary.configured, r.summary.addable, r.summary.bound_elsewhere), (1, 1, 1));
+        assert_eq!(r.candidates[0].instance, "qbit", "addable rows sort first");
+        assert_eq!(r.candidates[1].configured_as.as_deref(), Some("nzb"));
+        assert_eq!(r.candidates[2].state, CandidateState::BoundElsewhere);
+        let json = serde_json::to_value(&r).unwrap();
+        assert_eq!(json["self"], "metashare-app");
+        assert_eq!(json["summary"]["boundElsewhere"], 1);
+        assert_eq!(json["candidates"][2]["state"], "bound-elsewhere");
+    }
+
+    #[test]
+    fn suggested_names_are_list_safe() {
+        assert_eq!(suggested_name("MetaShare NZB"), "metashare-nzb");
+        assert_eq!(suggested_name(""), "plugin");
+    }
+
+    #[tokio::test]
+    async fn scan_reports_what_the_node_heard() {
+        use crate::beacon::{BeaconNode, Config, Message};
+        let port = std::net::UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
+        let mut cfg = Config::new("consumer").instance("me");
+        cfg.port = port;
+        let node = BeaconNode::spawn(cfg).unwrap();
+        let ad = Message::advertise(
+            NodeInfo { name: "meta-feeder-tmdb".into(), instance: "tmdb-feeder".into(), ..Default::default() },
+            vec![Resource::new("feeder", ["metamesh.feeder/tmdb"]).with_endpoint("http", "http://tmdb-feeder:8080")],
+        );
+        let s = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        s.send_to(&serde_json::to_vec(&ad).unwrap(), ("127.0.0.1", port)).await.unwrap();
+        let r = scan(&node, "metamesh.feeder/*", &[], Duration::from_millis(200)).await;
+        assert_eq!(r.summary.nodes, 1);
+        assert_eq!(r.summary.addable, 1);
+        assert_eq!(r.candidates[0].url, "http://tmdb-feeder:8080");
+        assert_eq!(r.candidates[0].suggested_name, "tmdb-feeder");
+        assert!(r.duration_ms >= 200);
+    }
+}

src/config_page.html

@@ -3,7 +3,7 @@
 <head>
 <meta charset="utf-8" />
 <meta name="viewport" content="width=device-width, initial-scale=1" />
-<title>feeder config</title>
+<title>plugin config</title>
 <style>
   :root { color-scheme: dark; }
   * { box-sizing: border-box; }
@@ -47,7 +47,7 @@
 </head>
 <body>
   <h1 id="title">Plugin configuration</h1>
-  <p class="sub">Secrets are write-only — set values are shown as “set”, never echoed back. Saving requires a feeder restart to take effect. Fields marked <span class="req">*</span> are required.</p>
+  <p class="sub">Secrets are write-only — set values are shown as “set”, never echoed back. Saving restarts the plugin to apply. Fields marked <span class="req">*</span> are required.</p>
   <div id="banner" class="banner"></div>
   <form id="form"></form>
   <div class="bar" id="bar" style="display:none">
@@ -220,14 +220,14 @@ async function save() {
     });
     if (!res.ok) {
       const detail = (await res.text().catch(() => "")).trim();
-      throw new Error(detail ? `the feeder rejected it (HTTP ${res.status}): ${detail}`
-                             : `the feeder returned HTTP ${res.status}`);
+      throw new Error(detail ? `the plugin rejected it (HTTP ${res.status}): ${detail}`
+                             : `the plugin returned HTTP ${res.status}`);
     }
     const j = await res.json().catch(() => ({}));
     if (j.reloading) {
       // The feeder is restarting itself (~2-3s) to apply the change. Poll it
       // back up, then refresh the form so secret pills reflect stored state.
-      banner("warn", "Saved — reloading the feeder to apply (~3 s)…");
+      banner("warn", "Saved — reloading the plugin to apply (~3 s)…");
       let tries = 6;
       const poll = async () => {
         try { await load(); banner("ok", "Reloaded — new config applied."); }
@@ -235,7 +235,7 @@ async function save() {
       };
       setTimeout(poll, 3000);
     } else {
-      if (j.restart_required) banner("warn", "Saved. Restart the feeder container for changes to take effect.");
+      if (j.restart_required) banner("warn", "Saved. Restart the plugin container for changes to take effect.");
       else banner("ok", "Saved.");
       // Re-fetch so secret pills reflect the new stored state.
       await load();
@@ -243,7 +243,7 @@ async function save() {
   } catch (e) {
     const msg = (e && e.message) ? e.message : String(e);
     const friendly = /Failed to fetch|NetworkError|load failed/i.test(msg)
-      ? "couldn't reach the feeder — it may be restarting; wait a moment and try again."
+      ? "couldn't reach the plugin — it may be restarting; wait a moment and try again."
       : msg;
     banner("err", "Could not save: " + friendly);
   } finally {

src/lib.rs

@@ -12,6 +12,10 @@
 //! With the `transport` feature it also carries the **transport-plugin**
 //! contract ([`transport`]): the same split, applied to meta-share — a
 //! protocol-agnostic hull and one sidecar per byte-moving protocol.
+//!
+//! [`beacon`] is beacon v2 local discovery. Both harnesses advertise through it;
+//! with `default-features = false, features = ["beacon"]` it is all a meta-*
+//! service pulls in.
 
 // `redb::Error` is a large enum; the cache wrappers (`cache.rs`) return it
 // directly rather than boxing on every embedded-DB call. Boxing each would be
@@ -25,38 +29,66 @@
 /// `metacore-app` service the AppStore feeder apps ship on the shared network.
 pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
 
+#[cfg(feature = "beacon")]
+pub mod beacon;
+
+#[cfg(feature = "feeder")]
 pub mod budget;
+#[cfg(feature = "feeder")]
 pub mod cache;
+#[cfg(feature = "feeder")]
 pub mod common;
+#[cfg(feature = "feeder")]
 pub mod config;
+#[cfg(feature = "feeder")]
 pub mod domain;
+#[cfg(feature = "feeder")]
 pub mod enrich;
+#[cfg(feature = "filename")]
 pub mod filename_meta;
+#[cfg(feature = "feeder")]
 pub mod hash;
+#[cfg(feature = "feeder")]
 pub mod lang;
+#[cfg(feature = "feeder")]
 pub mod licence;
+#[cfg(feature = "feeder")]
 pub mod meta_core;
+#[cfg(feature = "feeder")]
 pub mod plugin;
+#[cfg(feature = "feeder")]
 pub mod query;
+#[cfg(feature = "feeder")]
 pub mod query_eval;
+#[cfg(feature = "feeder")]
 pub mod serve;
 #[cfg(feature = "transport")]
 pub mod transport;
+#[cfg(feature = "feeder")]
 pub mod types;
 
+#[cfg(feature = "feeder")]
 pub use budget::RateBudget;
+#[cfg(feature = "feeder")]
 pub use config::{ConfigField, ConfigSchema, FieldKind};
+#[cfg(feature = "feeder")]
 pub use licence::licence_from_url;
+#[cfg(feature = "feeder")]
 pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher};
+#[cfg(feature = "feeder")]
 pub use meta_core::FeederStore;
+#[cfg(feature = "feeder")]
 pub use plugin::{
     upstream_id_field, ConfigError, FeederPlugin, HashKind, HashOutcome, PluginRegistry,
     RedeemClaim,
 };
+#[cfg(feature = "feeder")]
 pub use query::{GatewayQuery, GatewaySearchEvent, GatewayWireError, Negation, RangeFilter};
+#[cfg(feature = "feeder")]
 pub use serve::{
     configure_plugins, router, serve_feeders, ComputeRequest, ComputeResponse, HashKindDto,
     HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse,
     RedeemsResponse,
 };
+#[cfg(feature = "feeder")]
 pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth};

src/serve.rs

@@ -278,8 +278,15 @@ fn gateway_error_response(e: GatewayError) -> Response {
 // -------- handlers ---------------------------------------------------------
 
 async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
-    let mut plugins: Vec<PluginManifest> = state
-        .plugins
+    Json(build_manifest(&state.plugins, &state.version))
+}
+
+/// The `/manifest` body — also what the beacon advertise's `rev` is a hash of.
+fn build_manifest(
+    plugins: &HashMap<String, Arc<dyn FeederPlugin>>,
+    version: &str,
+) -> ManifestResponse {
+    let mut plugins: Vec<PluginManifest> = plugins
         .values()
         .map(|p| PluginManifest {
             id: p.upstream_id().to_string(),
@@ -294,10 +301,27 @@ async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
         })
         .collect();
     plugins.sort_by(|a, b| a.id.cmp(&b.id));
-    Json(ManifestResponse {
-        feeder_version: state.version.clone(),
+    ManifestResponse {
+        feeder_version: version.to_string(),
         plugins,
-    })
+    }
+}
+
+/// The feeder's beacon v2 resource: one `metamesh.feeder/<upstream_id>` cap per
+/// hosted plugin, the manifest endpoint, and `rev` = hash of the manifest.
+fn beacon_resource(
+    plugins: &HashMap<String, Arc<dyn FeederPlugin>>,
+    version: &str,
+    listen_port: u16,
+) -> crate::beacon::Resource {
+    use crate::beacon::{advertise_url, binds_from_env, caps, rev_of, Resource};
+    let m = build_manifest(plugins, version);
+    let mut r = Resource::new("feeder", m.plugins.iter().map(|p| caps::feeder(&p.id)))
+        .with_endpoint("http", advertise_url(listen_port))
+        .with_endpoint("manifest", "/manifest");
+    r.rev = Some(rev_of(&m));
+    r.binds = binds_from_env();
+    r
 }
 
 async fn redeems(State(state): State<AppState>) -> Json<RedeemsResponse> {
@@ -568,7 +592,8 @@ pub async fn serve_feeders(
     let state_dir = state_dir.into();
     let configured = configure_plugins(plugins, &state_dir)?;
     let version = env!("CARGO_PKG_VERSION").to_string();
-    let app = router(configured, version, state_dir.clone());
+    let advertised = configured.clone();
+    let app = router(configured, version.clone(), state_dir.clone());
 
     let listener = tokio::net::TcpListener::bind(listen).await?;
     info!(
@@ -576,6 +601,18 @@ pub async fn serve_feeders(
         %listen,
         "feeder listening"
     );
+    // Beacon v2: advertise once the port is bound, so a consumer that hears us
+    // can connect. The node lives for the process.
+    let port = listen.port();
+    let name = std::env::var("SERVICE_NAME").ok().filter(|s| !s.is_empty()).unwrap_or_else(|| {
+        let mut ids: Vec<&String> = advertised.keys().collect();
+        ids.sort();
+        ids.first().map_or_else(|| "meta-feeder".to_string(), |id| format!("meta-feeder-{id}"))
+    });
+    let beacon_version = version.clone();
+    let _beacon = crate::beacon::advertise_plugin(&name, &version, move || {
+        beacon_resource(&advertised, &beacon_version, port)
+    });
     axum::serve(listener, app)
         .await
         .map_err(anyhow::Error::from)

src/transport/client.rs

@@ -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> {

src/transport/config.rs

@@ -0,0 +1,347 @@
+//! A transport plugin's own settings — the "config plane".
+//!
+//! Same model as a feeder's (`crate::config`): the plugin declares a
+//! [`ConfigSchema`], keeps its values in `<state_dir>/config.json`, and serves a
+//! schema-driven page that meta-share's dashboard embeds through its plugin
+//! proxy (`/api/plugins/:name/ui/config`). Keys stay on the plugin; the hull
+//! carries no protocol-specific settings.
+//!
+//! Routes (mounted by [`serve`](super::serve) when
+//! [`TransportPlugin::config`](super::plugin::TransportPlugin::config) is
+//! `Some`):
+//!
+//! - `GET /config`        → the page ([`CONFIG_PAGE_HTML`]); every call it makes
+//!   is relative to its own path, so the proxy prefix is irrelevant.
+//! - `GET /config/schema` → [`ConfigSchema`].
+//! - `GET /config/values` → effective values, secrets redacted to `<key>_set`.
+//! - `PUT /config/values` → merge (blank secret keeps the stored one), persist
+//!   atomically, then exit so the container restarts on the new values
+//!   (`{"saved":true,"reloading":true}`). Plugins read their config once at
+//!   boot — a restart is the one apply path that is right for every one of them.
+//!
+//! **Effective values** = `config.json` if it exists and parses as an object,
+//! else the `seed` the plugin derived from its env (its pre-config-plane
+//! settings source). A plugin deserializes [`ConfigPlane::effective`] into its
+//! own typed struct with serde defaults, so a missing key is never fatal.
+
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::extract::State;
+use axum::http::StatusCode;
+use axum::response::{Html, IntoResponse, Response};
+use axum::routing::get;
+use axum::{Json, Router};
+use serde_json::Value;
+use tracing::{info, warn};
+
+use crate::config::{merge, redact, ConfigSchema, CONFIG_PAGE_HTML};
+
+/// Env var naming the plugin's own writable state dir.
+pub const STATE_DIR_ENV: &str = "META_SHARE_PLUGIN_STATE_DIR";
+/// Default state dir (the plugin app's own `AppData/<app>/state` mount).
+pub const DEFAULT_STATE_DIR: &str = "/state";
+/// File name of the stored values inside the state dir.
+pub const CONFIG_FILE: &str = "config.json";
+
+/// The plugin's state dir from [`STATE_DIR_ENV`], else [`DEFAULT_STATE_DIR`].
+pub fn state_dir_from_env() -> PathBuf {
+    std::env::var(STATE_DIR_ENV)
+        .ok()
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+        .unwrap_or_else(|| DEFAULT_STATE_DIR.to_string())
+        .into()
+}
+
+/// One plugin's schema + stored values.
+pub struct ConfigPlane {
+    schema: ConfigSchema,
+    path: PathBuf,
+    seed: Value,
+    restart_on_save: bool,
+}
+
+impl ConfigPlane {
+    /// `seed` is what the plugin would run with if no `config.json` existed
+    /// (normally its env-derived settings, serialized).
+    pub fn new(schema: ConfigSchema, state_dir: &Path, seed: Value) -> Self {
+        Self {
+            schema,
+            path: state_dir.join(CONFIG_FILE),
+            seed,
+            restart_on_save: true,
+        }
+    }
+
+    /// Don't exit after a save (tests, or a plugin that hot-applies).
+    pub fn without_restart(mut self) -> Self {
+        self.restart_on_save = false;
+        self
+    }
+
+    pub fn schema(&self) -> &ConfigSchema {
+        &self.schema
+    }
+
+    pub fn path(&self) -> &Path {
+        &self.path
+    }
+
+    /// The stored values, if `config.json` exists and is a JSON object.
+    pub fn stored(&self) -> Option<Value> {
+        let bytes = std::fs::read(&self.path).ok()?;
+        match serde_json::from_slice::<Value>(&bytes) {
+            Ok(v @ Value::Object(_)) => Some(v),
+            Ok(_) | Err(_) => {
+                warn!(path = %self.path.display(), "plugin config is not a JSON object; using the seed");
+                None
+            }
+        }
+    }
+
+    /// Stored values, else the seed.
+    pub fn effective(&self) -> Value {
+        self.stored().unwrap_or_else(|| self.seed.clone())
+    }
+
+    /// Persist `values` atomically (tmp + rename), creating the state dir.
+    pub fn save(&self, values: &Value) -> std::io::Result<()> {
+        if let Some(dir) = self.path.parent() {
+            std::fs::create_dir_all(dir)?;
+        }
+        let tmp = self.path.with_extension("json.tmp");
+        let body = serde_json::to_vec_pretty(values)
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
+        std::fs::write(&tmp, body)?;
+        std::fs::rename(&tmp, &self.path)
+    }
+
+    /// Write `values` as the initial config **only** if no `config.json` exists
+    /// yet — the one-shot migration hook (e.g. the nzb plugin importing the
+    /// NNTP block meta-share used to own). Returns whether it wrote.
+    pub fn seed_file_if_absent(&self, values: &Value) -> std::io::Result<bool> {
+        if self.path.exists() {
+            return Ok(false);
+        }
+        self.save(values)?;
+        Ok(true)
+    }
+
+    /// The `/config*` routes.
+    pub fn routes(self: Arc<Self>) -> Router {
+        Router::new()
+            .route("/config", get(page))
+            .route("/config/schema", get(schema))
+            .route("/config/values", get(values_get).put(values_put))
+            .with_state(self)
+    }
+}
+
+/// Lenient readers for a config object as the page writes it: an empty number
+/// is `null`, an empty text is `""`, and a hand-edited file may hold a number as
+/// a string. Each returns `None` for "not set" so the caller falls back to its
+/// default.
+pub mod read {
+    use serde_json::Value;
+
+    /// A non-blank, trimmed string.
+    pub fn text(v: &Value, key: &str) -> Option<String> {
+        match v.get(key)? {
+            Value::String(s) => Some(s.trim().to_string()).filter(|s| !s.is_empty()),
+            Value::Number(n) => Some(n.to_string()),
+            _ => None,
+        }
+    }
+
+    pub fn number(v: &Value, key: &str) -> Option<f64> {
+        match v.get(key)? {
+            Value::Number(n) => n.as_f64(),
+            Value::String(s) => s.trim().parse().ok(),
+            _ => None,
+        }
+    }
+
+    /// A non-negative integer (fractions truncate, negatives are "not set").
+    pub fn uint(v: &Value, key: &str) -> Option<u64> {
+        number(v, key).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u64)
+    }
+
+    pub fn flag(v: &Value, key: &str) -> Option<bool> {
+        match v.get(key)? {
+            Value::Bool(b) => Some(*b),
+            Value::String(s) => match s.trim().to_ascii_lowercase().as_str() {
+                "1" | "true" | "yes" | "on" => Some(true),
+                "0" | "false" | "no" | "off" => Some(false),
+                _ => None,
+            },
+            _ => None,
+        }
+    }
+
+    /// A list of non-blank strings (a string is split on commas/newlines).
+    pub fn list(v: &Value, key: &str) -> Option<Vec<String>> {
+        let items: Vec<String> = match v.get(key)? {
+            Value::Array(a) => a.iter().filter_map(|x| x.as_str()).map(str::to_string).collect(),
+            Value::String(s) => s.split([',', '\n']).map(str::to_string).collect(),
+            _ => return None,
+        };
+        Some(items.into_iter().map(|s| s.trim().to_string()).filter(|s| !s.is_empty()).collect())
+    }
+}
+
+async fn page() -> Html<&'static str> {
+    Html(CONFIG_PAGE_HTML)
+}
+
+async fn schema(State(p): State<Arc<ConfigPlane>>) -> Json<ConfigSchema> {
+    Json(p.schema.clone())
+}
+
+async fn values_get(State(p): State<Arc<ConfigPlane>>) -> Json<Value> {
+    Json(redact(&p.effective(), &p.schema))
+}
+
+async fn values_put(
+    State(p): State<Arc<ConfigPlane>>,
+    Json(incoming): Json<Value>,
+) -> Result<Json<Value>, Response> {
+    if !incoming.is_object() {
+        return Err((StatusCode::BAD_REQUEST, "expected a JSON object").into_response());
+    }
+    let merged = merge(&p.effective(), &incoming, &p.schema);
+    p.save(&merged).map_err(|e| {
+        (StatusCode::INTERNAL_SERVER_ERROR, format!("write config: {e}")).into_response()
+    })?;
+    if !p.restart_on_save {
+        info!(path = %p.path.display(), "plugin config saved");
+        return Ok(Json(serde_json::json!({ "saved": true, "reloading": false })));
+    }
+    info!(path = %p.path.display(), "plugin config saved; restarting to apply");
+    // Let this response flush, then exit; `restart: unless-stopped` brings the
+    // plugin back and it reads config.json at boot.
+    tokio::spawn(async {
+        tokio::time::sleep(Duration::from_millis(800)).await;
+        std::process::exit(0);
+    });
+    Ok(Json(serde_json::json!({ "saved": true, "reloading": true })))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::ConfigField;
+    use axum::body::Body;
+    use axum::http::Request;
+    use tower::ServiceExt;
+
+    fn schema() -> ConfigSchema {
+        ConfigSchema {
+            fields: vec![
+                ConfigField::text("host", "Host"),
+                ConfigField::secret("pass", "Password"),
+                ConfigField::number("connections", "Connections"),
+            ],
+        }
+    }
+
+    fn plane(dir: &Path) -> Arc<ConfigPlane> {
+        Arc::new(
+            ConfigPlane::new(schema(), dir, serde_json::json!({"host":"seed.example","connections":4}))
+                .without_restart(),
+        )
+    }
+
+    async fn call(app: Router, method: &str, uri: &str, body: Option<Value>) -> (StatusCode, Value) {
+        let req = Request::builder()
+            .method(method)
+            .uri(uri)
+            .header("content-type", "application/json")
+            .body(match body {
+                Some(b) => Body::from(serde_json::to_vec(&b).unwrap()),
+                None => Body::empty(),
+            })
+            .unwrap();
+        let r = app.oneshot(req).await.unwrap();
+        let status = r.status();
+        let bytes = axum::body::to_bytes(r.into_body(), 1 << 20).await.unwrap();
+        (status, serde_json::from_slice(&bytes).unwrap_or(Value::Null))
+    }
+
+    #[tokio::test]
+    async fn seed_until_saved_then_file_wins_and_secret_is_kept() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = plane(dir.path());
+        assert_eq!(p.effective()["host"], "seed.example");
+
+        let (st, v) = call(Arc::clone(&p).routes(), "GET", "/config/values", None).await;
+        assert_eq!(st, StatusCode::OK);
+        assert_eq!(v["host"], "seed.example");
+        assert_eq!(v["pass_set"], false);
+
+        let (st, _) = call(
+            Arc::clone(&p).routes(),
+            "PUT",
+            "/config/values",
+            Some(serde_json::json!({"host":"news.example","pass":"s3cret"})),
+        )
+        .await;
+        assert_eq!(st, StatusCode::OK);
+        let stored = p.stored().unwrap();
+        assert_eq!(stored["host"], "news.example");
+        assert_eq!(stored["pass"], "s3cret");
+        assert_eq!(stored["connections"], 4, "absent field keeps the effective value");
+
+        // A blank secret keeps the stored one; the GET never leaks it.
+        call(Arc::clone(&p).routes(), "PUT", "/config/values", Some(serde_json::json!({"pass":""}))).await;
+        assert_eq!(p.stored().unwrap()["pass"], "s3cret");
+        let (_, v) = call(Arc::clone(&p).routes(), "GET", "/config/values", None).await;
+        assert!(v.get("pass").is_none());
+        assert_eq!(v["pass_set"], true);
+    }
+
+    #[tokio::test]
+    async fn schema_and_page_are_served() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = plane(dir.path());
+        let (st, v) = call(Arc::clone(&p).routes(), "GET", "/config/schema", None).await;
+        assert_eq!(st, StatusCode::OK);
+        assert_eq!(v["fields"].as_array().unwrap().len(), 3);
+        let r = p
+            .routes()
+            .oneshot(Request::get("/config").body(Body::empty()).unwrap())
+            .await
+            .unwrap();
+        assert_eq!(r.status(), StatusCode::OK);
+    }
+
+    #[test]
+    fn seed_file_if_absent_writes_once() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = plane(dir.path());
+        assert!(p.seed_file_if_absent(&serde_json::json!({"host":"a"})).unwrap());
+        assert!(!p.seed_file_if_absent(&serde_json::json!({"host":"b"})).unwrap());
+        assert_eq!(p.effective()["host"], "a");
+    }
+
+    #[test]
+    fn readers_treat_blank_as_unset() {
+        let v = serde_json::json!({"a":"", "b":" x ", "n":null, "m":"7", "f":"off", "l":["a"," ",""], "s":"p, q"});
+        assert_eq!(read::text(&v, "a"), None);
+        assert_eq!(read::text(&v, "b").as_deref(), Some("x"));
+        assert_eq!(read::uint(&v, "n"), None);
+        assert_eq!(read::uint(&v, "m"), Some(7));
+        assert_eq!(read::flag(&v, "f"), Some(false));
+        assert_eq!(read::list(&v, "l"), Some(vec!["a".to_string()]));
+        assert_eq!(read::list(&v, "s"), Some(vec!["p".to_string(), "q".to_string()]));
+        assert_eq!(read::text(&v, "missing"), None);
+    }
+
+    #[test]
+    fn a_non_object_file_falls_back_to_the_seed() {
+        let dir = tempfile::tempdir().unwrap();
+        std::fs::write(dir.path().join(CONFIG_FILE), b"[1,2]").unwrap();
+        assert_eq!(plane(dir.path()).effective()["host"], "seed.example");
+    }
+}

src/transport/dto.rs

@@ -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";
@@ -82,6 +111,11 @@ pub struct Manifest {
     pub version: String,
     pub contract: u32,
     pub capabilities: Capabilities,
+    /// Serves a config plane (`/config`, `/config/schema`, `/config/values`).
+    /// Set by the harness from [`TransportPlugin::config`](super::plugin::TransportPlugin::config);
+    /// absent in older manifests.
+    #[serde(default)]
+    pub config: bool,
 }
 
 #[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)]

src/transport/hull.rs

@@ -17,6 +17,13 @@
 //!   [`HDR_MANIFEST_CID`]. `404` = not a Usenet cid; any other error carries
 //!   `{"error"}` — the manifest could not be redeemed or fetched. Parsing and
 //!   the `nzb-posting` digest check stay in the plugin.
+//! - `GET  /internal/network` → [`NetworkInfo`]: the hull's own endpoints a
+//!   plugin needs to know (the ipfs plugin announces `peer_url` in identify).
+//! - `GET  /internal/legacy/usenet` → the NNTP block meta-share's `settings.json`
+//!   held before the nzb plugin owned its settings (`404` = none). One-shot
+//!   migration source for the nzb plugin's config plane.
+//! - `GET  /health` → `200` once the hull has finished its boot steps (the
+//!   ipfs plugin waits on it before opening `ipfs/blocks.redb`).
 //!
 //! Why callbacks and not plugin-side logic: these need what only the hull has —
 //! the record cache and meta-core, the gateway redeem, the IPFS tier. Plugins
@@ -32,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.
@@ -61,6 +69,17 @@ pub struct RecordInfo {
     pub title: Option<String>,
 }
 
+/// The hull's endpoints (`GET /internal/network`).
+#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
+pub struct NetworkInfo {
+    /// meta-core base URL the hull talks to.
+    #[serde(default)]
+    pub meta_core_url: Option<String>,
+    /// The URL other peers reach this meta-share at (announced in identify).
+    #[serde(default)]
+    pub peer_url: Option<String>,
+}
+
 /// Plugin-side client for the hull's internal API. A client with no base URL
 /// (`META_SHARE_HULL_URL=` empty, or [`HullClient::disabled`]) answers every call
 /// with "not available", which is what unit tests and a standalone plugin get.
@@ -191,6 +210,37 @@ impl HullClient {
         self.get_json(&format!("/internal/records/{cid}"), Duration::from_secs(15)).await
     }
 
+    /// The hull's endpoints.
+    pub async fn network(&self) -> Result<Option<NetworkInfo>, String> {
+        self.get_json("/internal/network", Duration::from_secs(5)).await
+    }
+
+    /// The NNTP block meta-share's own settings held before the nzb plugin had
+    /// a config plane; `Ok(None)` = there is none.
+    pub async fn legacy_usenet(&self) -> Result<Option<super::nzb::UsenetSettings>, String> {
+        self.get_json("/internal/legacy/usenet", Duration::from_secs(5)).await
+    }
+
+    /// Wait until the hull's internal listener answers `GET /health` with 2xx,
+    /// or `max` elapses. Returns whether it came up.
+    pub async fn wait_ready(&self, max: Duration) -> bool {
+        let Ok(url) = self.url("/health") else {
+            return false;
+        };
+        let deadline = tokio::time::Instant::now() + max;
+        loop {
+            if let Ok(r) = self.http.get(&url).timeout(Duration::from_secs(3)).send().await {
+                if r.status().is_success() {
+                    return true;
+                }
+            }
+            if tokio::time::Instant::now() >= deadline {
+                return false;
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    }
+
     /// [`RecordInfo`] from the hull's record cache only.
     pub async fn cached_record(&self, cid: &str) -> Result<Option<RecordInfo>, String> {
         self.get_json(&format!("/internal/records/{cid}?cached=1"), Duration::from_secs(5)).await

src/transport/ipfs.rs

@@ -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)]

src/transport/mod.rs

@@ -12,6 +12,8 @@
 //! - [`plugin::TransportPlugin`] + [`serve::serve_transport`]: the plugin side.
 //! - [`client::RemoteTransport`]: the hull side.
 //! - [`hull::HullClient`]: plugin → hull callbacks (internal port only).
+//! - [`config::ConfigPlane`]: the plugin's own settings + the config page the
+//!   hull's dashboard embeds.
 //! - [`focus::FocusView`]: the plugin's replica of the hull's playback focus.
 //! - [`dto`]: common wire types; [`error::ApiError`] / [`range`]: shared so a
 //!   relayed plugin response is byte-identical to what the monolith answered.
@@ -20,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.
 //!
@@ -29,6 +32,7 @@
 
 pub mod cid;
 pub mod client;
+pub mod config;
 pub mod dto;
 pub mod error;
 pub mod file;
@@ -39,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;
@@ -46,12 +51,13 @@ pub mod torrent;
 pub mod testkit;
 
 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};
-pub use hull::{HullClient, RecordInfo};
+pub use hull::{HullClient, NetworkInfo, RecordInfo};
 pub use plugin::TransportPlugin;
 pub use serve::serve_transport;

src/transport/plugin.rs

@@ -7,6 +7,7 @@ use axum::http::HeaderMap;
 use axum::response::Response;
 use axum::Router;
 
+use super::config::ConfigPlane;
 use super::dto::{Deleted, Health, Job, Manifest, ReconcileReport, ReconcileRequest};
 use super::error::ApiError;
 use super::focus::FocusView;
@@ -42,6 +43,13 @@ pub trait TransportPlugin: Send + Sync + 'static {
     /// The view `PUT /focus` writes into.
     fn focus(&self) -> &Arc<FocusView>;
 
+    /// The plugin's own settings, if it has any. `Some` mounts the `/config*`
+    /// routes and sets `manifest.config` — the hull's dashboard then offers a
+    /// "configure" panel that embeds `/config` through its proxy.
+    fn config(&self) -> Option<Arc<ConfigPlane>> {
+        None
+    }
+
     /// Protocol-specific routes, merged into the common router. Paths must not
     /// collide with the common ones.
     fn extra_routes(self: Arc<Self>) -> Router {

src/transport/pointer.rs

@@ -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"));
+    }
+}

src/transport/serve.rs

@@ -11,7 +11,9 @@
 //! - `PUT    /focus`           ← [`FocusSnapshot`](super::focus::FocusSnapshot), `204`
 //! - `GET    /focus`           → the snapshot the plugin currently obeys
 //!
-//! plus whatever [`TransportPlugin::extra_routes`] adds.
+//! plus the `/config*` routes when [`TransportPlugin::config`] is `Some` (see
+//! [`ConfigPlane`](super::config::ConfigPlane)), plus whatever
+//! [`TransportPlugin::extra_routes`] adds.
 
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -32,8 +34,9 @@ type Shared = Arc<dyn TransportPlugin>;
 /// The common router for `plugin`, merged with its extra routes.
 pub fn router<P: TransportPlugin>(plugin: Arc<P>) -> Router {
     let extra = Arc::clone(&plugin).extra_routes();
+    let config = plugin.config().map(|c| c.routes());
     let shared: Shared = plugin;
-    Router::new()
+    let r = Router::new()
         .route("/manifest", get(manifest))
         .route("/health", get(health))
         .route("/raw/:cid", get(raw))
@@ -42,7 +45,11 @@ pub fn router<P: TransportPlugin>(plugin: Arc<P>) -> Router {
         .route("/reconcile", post(reconcile))
         .route("/focus", get(get_focus).put(put_focus))
         .with_state(shared)
-        .merge(extra)
+        .merge(extra);
+    match config {
+        Some(c) => r.merge(c),
+        None => r,
+    }
 }
 
 /// Bind `listen` and serve `plugin` until SIGTERM / Ctrl-C.
@@ -50,16 +57,47 @@ pub async fn serve_transport<P: TransportPlugin>(
     plugin: Arc<P>,
     listen: SocketAddr,
 ) -> anyhow::Result<()> {
-    let id = plugin.manifest().id;
+    let manifest = plugin.manifest();
+    let id = manifest.id.clone();
+    let advertised = Arc::clone(&plugin);
     let app = router(plugin);
     let listener = tokio::net::TcpListener::bind(listen).await?;
     info!(%listen, plugin = %id, "transport plugin listening");
+    // Beacon v2: advertise once bound; say `bye` on clean shutdown so the hull
+    // drops us at once instead of after the liveness window.
+    let port = listen.port();
+    // `node.name` is a short machine name; the human description stays in the
+    // manifest (`implementation`), which a consumer fetches over HTTP.
+    let beacon_name = std::env::var("SERVICE_NAME")
+        .ok()
+        .filter(|s| !s.is_empty())
+        .unwrap_or_else(|| format!("meta-transport-{}", manifest.id));
+    let beacon = crate::beacon::advertise_plugin(&beacon_name, &manifest.version, move || {
+        beacon_resource(advertised.as_ref(), port)
+    });
     axum::serve(listener, app)
         .with_graceful_shutdown(shutdown_signal())
         .await?;
+    if let Some(b) = beacon {
+        b.bye().await;
+    }
     Ok(())
 }
 
+/// The transport's beacon v2 resource: id = manifest id, cap
+/// `metamesh.transport/<id>@<contract>`, `rev` = hash of the served manifest.
+fn beacon_resource<P: TransportPlugin + ?Sized>(plugin: &P, listen_port: u16) -> crate::beacon::Resource {
+    use crate::beacon::{advertise_url, binds_from_env, caps, rev_of, Resource};
+    let mut m = plugin.manifest();
+    m.config = plugin.config().is_some();
+    let mut r = Resource::new(m.id.clone(), [caps::transport(&m.id, m.contract)])
+        .with_endpoint("http", advertise_url(listen_port))
+        .with_endpoint("manifest", "/manifest");
+    r.rev = Some(rev_of(&m));
+    r.binds = binds_from_env();
+    r
+}
+
 /// Resolves on SIGTERM (docker stop) or Ctrl-C.
 pub async fn shutdown_signal() {
     let ctrl_c = async {
@@ -82,7 +120,9 @@ pub async fn shutdown_signal() {
 }
 
 async fn manifest(State(p): State<Shared>) -> Response {
-    Json(p.manifest()).into_response()
+    let mut m = p.manifest();
+    m.config = p.config().is_some();
+    Json(m).into_response()
 }
 
 async fn health(State(p): State<Shared>) -> Response {

src/transport/testkit.rs

@@ -39,8 +39,10 @@ pub async fn conformance(base: &str) -> Result<(), Vec<String>> {
         .with_control_timeout(Duration::from_secs(10));
     let mut fails = Vec::new();
 
+    let mut has_config = false;
     match t.manifest().await {
         Ok(m) => {
+            has_config = m.config;
             if m.id.is_empty() {
                 fails.push("manifest.id is empty".into());
             }
@@ -110,6 +112,31 @@ pub async fn conformance(base: &str) -> Result<(), Vec<String>> {
         Err(e) => fails.push(format!("GET /raw/not-a-cid: {e}")),
     }
 
+    if has_config {
+        match http.get(t.url("/config/schema")).send().await {
+            Ok(r) if r.status().is_success() => {
+                if r.json::<crate::config::ConfigSchema>().await.is_err() {
+                    fails.push("GET /config/schema body is not a ConfigSchema".into());
+                }
+            }
+            Ok(r) => fails.push(format!("GET /config/schema: status {}", r.status())),
+            Err(e) => fails.push(format!("GET /config/schema: {e}")),
+        }
+        match http.get(t.url("/config/values")).send().await {
+            Ok(r) if r.status().is_success() => match r.json::<serde_json::Value>().await {
+                Ok(v) if v.is_object() => {}
+                _ => fails.push("GET /config/values is not a JSON object".into()),
+            },
+            Ok(r) => fails.push(format!("GET /config/values: status {}", r.status())),
+            Err(e) => fails.push(format!("GET /config/values: {e}")),
+        }
+        match http.get(t.url("/config")).send().await {
+            Ok(r) if r.status().is_success() => {}
+            Ok(r) => fails.push(format!("GET /config: status {}", r.status())),
+            Err(e) => fails.push(format!("GET /config: {e}")),
+        }
+    }
+
     if fails.is_empty() {
         Ok(())
     } else {
@@ -130,6 +157,7 @@ mod tests {
     /// The smallest plugin that should pass: holds nothing, fetches nothing.
     struct Empty {
         focus: Arc<FocusView>,
+        config: Option<Arc<crate::transport::config::ConfigPlane>>,
     }
 
     #[async_trait]
@@ -141,6 +169,7 @@ mod tests {
                 version: "0".into(),
                 contract: CONTRACT_VERSION,
                 capabilities: Capabilities { fetch: true, share: false },
+                config: false,
             }
         }
         async fn raw(&self, _cid: String, _h: HeaderMap) -> Response {
@@ -158,11 +187,32 @@ mod tests {
         fn focus(&self) -> &Arc<FocusView> {
             &self.focus
         }
+        fn config(&self) -> Option<Arc<crate::transport::config::ConfigPlane>> {
+            self.config.clone()
+        }
     }
 
     #[tokio::test]
     async fn the_empty_plugin_conforms() {
-        let addr = spawn_local(Arc::new(Empty { focus: FocusView::new() })).await;
+        let addr = spawn_local(Arc::new(Empty { focus: FocusView::new(), config: None })).await;
         conformance(&format!("http://{addr}")).await.unwrap();
     }
+
+    #[tokio::test]
+    async fn a_plugin_with_a_config_plane_conforms_and_says_so() {
+        use crate::config::{ConfigField, ConfigSchema};
+        use crate::transport::config::ConfigPlane;
+        let dir = tempfile::tempdir().unwrap();
+        let plane = ConfigPlane::new(
+            ConfigSchema { fields: vec![ConfigField::secret("key", "Key")] },
+            dir.path(),
+            serde_json::json!({}),
+        )
+        .without_restart();
+        let addr = spawn_local(Arc::new(Empty { focus: FocusView::new(), config: Some(Arc::new(plane)) })).await;
+        let base = format!("http://{addr}");
+        conformance(&base).await.unwrap();
+        let m: Manifest = reqwest::get(format!("{base}/manifest")).await.unwrap().json().await.unwrap();
+        assert!(m.config, "the harness must advertise the config plane");
+    }
 }