diff --git a/Cargo.toml b/Cargo.toml index 7c69d61..9312604 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-feeder-sdk" -version = "1.5.0" +version = "1.6.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"] } -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 } diff --git a/README.md b/README.md index c96a22c..0d8bf55 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,34 @@ 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) + +`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/` (one cap per hosted plugin), a +transport as `metamesh.transport/@` — 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:` | `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` | | + +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.6.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 +169,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 diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs new file mode 100644 index 0000000..299e65e --- /dev/null +++ b/src/beacon/mod.rs @@ -0,0 +1,155 @@ +//! 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; + +pub use node::{enabled_from_env, hostname, BeaconNode, Config, SeenNode, SeenResource}; +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/`. + pub const SERVICE_PREFIX: &str = "metamesh.service/"; + /// Every service. + pub const ANY_SERVICE: &str = "metamesh.service/*"; + /// Prefix of `metamesh.transport/@`. + pub const TRANSPORT_PREFIX: &str = "metamesh.transport/"; + /// Every transport plugin. + pub const ANY_TRANSPORT: &str = "metamesh.transport/*"; + /// Prefix of `metamesh.feeder/`. + pub const FEEDER_PREFIX: &str = "metamesh.feeder/"; + /// Every feeder upstream. + pub const ANY_FEEDER: &str = "metamesh.feeder/*"; + + 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}") + } +} + +/// The resource every meta-* service advertises: id `service`, cap +/// `metamesh.service/`, `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:` — 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 { + 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(name: &str, version: &str, build: F) -> Option +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 = 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 +} diff --git a/src/beacon/node.rs b/src/beacon/node.rs new file mode 100644 index 0000000..8e4235b --- /dev/null +++ b/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, + 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) -> 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) -> 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::().ok()) { + if ms > 0 { + c.interval = Duration::from_millis(ms); + } + } + c + } + + pub fn version(mut self, v: impl Into) -> Self { + self.node.version = Some(v.into()); + self + } + + pub fn instance(mut self, i: impl Into) -> 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, + /// 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 { + let mut out: Vec = 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, + resources: RwLock>, + socket: UdpSocket, + ifindexes: Vec<(String, u32)>, + seen: Mutex>, + changes: watch::Sender, + stopped: AtomicBool, +} + +/// A running beacon v2 participant. Cheap to clone. +#[derive(Clone)] +pub struct BeaconNode { + inner: Arc, +} + +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 { + 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) { + { + 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 { + self.inner.changes.subscribe() + } + + /// Every live node, sorted by (name, instance). + pub fn nodes(&self) -> Vec { + let cutoff = self.cutoff(); + let guard = self.inner.seen.lock().unwrap(); + let mut out: Vec = 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 { + let mut best: HashMap = 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 = 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 { + 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> { + 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 { + 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::() 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::() 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::() 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::() 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::() 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 { + 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) -> 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"); + } +} diff --git a/src/beacon/proto.rs b/src/beacon/proto.rs new file mode 100644 index 0000000..504e881 --- /dev/null +++ b/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, + /// `starting` | `running`. Absent reads as `running`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status: Option, +} + +/// One advertised resource. +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)] +pub struct Resource { + /// Unique within the node. + pub id: String, + pub caps: Vec, + /// 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, + /// Opaque revision of the resource's details (e.g. its manifest). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rev: Option, + /// The one consumer instance this resource belongs to. Absent = anyone. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub binds: Option, + /// Profile-defined, per capability. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub data: Option, +} + +impl Resource { + pub fn new(id: impl Into, caps: impl IntoIterator>) -> Self { + Self { + id: id.into(), + caps: caps.into_iter().map(Into::into).collect(), + ..Default::default() + } + } + + pub fn with_endpoint(mut self, name: impl Into, url: impl Into) -> 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 { + 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, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub resources: Vec, + /// Probe only: informational sender instance. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub from: Option, + /// Probe only: reply only if a resource matches one of these. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub want: Vec, +} + +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) -> 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 { + 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(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()])); + } +} diff --git a/src/lib.rs b/src/lib.rs index 80cb5ee..1792edf 100644 --- a/src/lib.rs +++ b/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}; diff --git a/src/serve.rs b/src/serve.rs index 57a8a1c..d03bb4e 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -278,8 +278,15 @@ fn gateway_error_response(e: GatewayError) -> Response { // -------- handlers --------------------------------------------------------- async fn manifest(State(state): State) -> Json { - let mut plugins: Vec = 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>, + version: &str, +) -> ManifestResponse { + let mut plugins: Vec = plugins .values() .map(|p| PluginManifest { id: p.upstream_id().to_string(), @@ -294,10 +301,27 @@ async fn manifest(State(state): State) -> Json { }) .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/` cap per +/// hosted plugin, the manifest endpoint, and `rev` = hash of the manifest. +fn beacon_resource( + plugins: &HashMap>, + 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) -> Json { @@ -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) diff --git a/src/transport/serve.rs b/src/transport/serve.rs index a41955e..9fca4bd 100644 --- a/src/transport/serve.rs +++ b/src/transport/serve.rs @@ -57,16 +57,41 @@ pub async fn serve_transport( plugin: Arc

, 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(); + let beacon = crate::beacon::advertise_plugin(&manifest.implementation, &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/@`, `rev` = hash of the served manifest. +fn beacon_resource(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 {