diff --git a/Cargo.toml b/Cargo.toml index 9312604..aac7f59 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-feeder-sdk" -version = "1.6.0" +version = "1.7.0" edition = "2021" rust-version = "1.85" license = "MIT" diff --git a/README.md b/README.md index 0d8bf55..dc3e8a9 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ 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) +### 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 @@ -52,11 +52,19 @@ transport as `metamesh.transport/@` — with `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 +`` element renders. Transports advertise as +`meta-transport-` (`SERVICE_NAME` overrides), feeders as +`meta-feeder-`. + 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"] } +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`), diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index 299e65e..2cdc574 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -15,8 +15,10 @@ 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, @@ -38,6 +40,10 @@ pub mod caps { pub const FEEDER_PREFIX: &str = "metamesh.feeder/"; /// Every feeder upstream. pub const ANY_FEEDER: &str = "metamesh.feeder/*"; + /// Prefix of `metamesh.enrich/` (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}") @@ -50,6 +56,10 @@ pub mod caps { 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 diff --git a/src/beacon/scan.rs b/src/beacon/scan.rs new file mode 100644 index 0000000..75db6fa --- /dev/null +++ b/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 +//! `` 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, + pub resource_id: String, + pub caps: Vec, + /// `endpoints.http` — what the consumer would register. + pub url: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub binds: Option, + pub state: CandidateState, + /// The list entry that already holds this URL (state `configured`). + #[serde(skip_serializing_if = "Option::is_none")] + pub configured_as: Option, + /// 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, +} + +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, + 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); + } +} diff --git a/src/transport/serve.rs b/src/transport/serve.rs index 9fca4bd..a24eb1e 100644 --- a/src/transport/serve.rs +++ b/src/transport/serve.rs @@ -66,7 +66,13 @@ pub async fn serve_transport( // 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 || { + // `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)