beacon scan (1.7.0): probe + classify configured/addable/bound-elsewhere for consumer Scan cards; metamesh.enrich caps; transports advertise as meta-transport-<id>

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "meta-feeder-sdk"
-version = "1.5.0"
+version = "1.7.0"
 edition = "2021"
 rust-version = "1.85"
 license = "MIT"
@@ -32,44 +32,59 @@ 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",
+    "dep:axum", "dep:hyper", "dep:reqwest", "dep:base64", "dep:futures",
+    "dep:async-trait", "dep:anyhow", "dep:thiserror", "dep:sha2", "dep:bytes",
+    "dep:regex", "dep:redb",
+]
+# 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"] }
+axum = { version = "0.7", features = ["json"], optional = true }
-hyper = { version = "1", features = ["full"] }
+hyper = { version = "1", features = ["full"], optional = true }
-reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"], default-features = false }
+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"
+futures = { version = "0.3", optional = true }
-async-trait = "0.1"
+async-trait = { version = "0.1", optional = true }
-anyhow = "1"
+anyhow = { version = "1", optional = true }
-thiserror = "1"
+thiserror = { version = "1", optional = true }
 tracing = "0.1"
-sha2 = "0.10"
+sha2 = { version = "0.10", optional = true }
-bytes = "1"
+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 }

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/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 = "feeder")]
 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
+    Json(build_manifest(&state.plugins, &state.version))
-        .plugins
+}
+
+/// 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 {
+    ManifestResponse {
-        feeder_version: state.version.clone(),
+        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/serve.rs

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