diff --git a/Cargo.toml b/Cargo.toml index 0570607..2e7852d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-transport-ipfs" -version = "1.0.0" +version = "1.1.0" edition = "2021" rust-version = "1.85" license = "MIT" @@ -19,7 +19,7 @@ path = "src/main.rs" # The transport-plugin contract. Pinned by TAG, never by path — a path dep # resolves outside this repo, which is the Docker build context. See the # Dockerfile for building against an unreleased SDK. -meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.4.0", features = ["transport"] } +meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.5.0", features = ["transport"] } # libp2p — transport only: identify (address exchange, load-bearing for bitswap # reachability), kad (DHT provider records), mdns (LAN discovery) + bitswap. @@ -56,7 +56,7 @@ multihash-codetable = { version = "0.1", features = ["sha2", "sha3"] } [dev-dependencies] tempfile = "3" -meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.4.0", features = ["testkit"] } +meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.5.0", features = ["testkit"] } # --------------------------------------------------------------------------- # Workaround for the yanked transitive `core2` (all versions). `multihash- diff --git a/Dockerfile b/Dockerfile index 177b976..462f32e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -44,8 +44,8 @@ COPY --from=builder /meta-transport-ipfs /usr/local/bin/meta-transport-ipfs # Same uid as meta-share: the plugins share its /data volume. RUN useradd --system --uid 10001 --no-create-home metashare \ - && mkdir -p /data/meta-share \ - && chown metashare:metashare /data/meta-share + && mkdir -p /data/meta-share /state \ + && chown metashare:metashare /data/meta-share /state USER metashare WORKDIR /data/meta-share @@ -57,7 +57,7 @@ EXPOSE 3000 4001 ENV HTTP_LISTEN=0.0.0.0:3000 \ P2P_LISTEN=/ip4/0.0.0.0/tcp/4001 \ META_SHARE_DATA=/data/meta-share \ - META_SHARE_CONFIG_DIR=/config \ + META_SHARE_PLUGIN_STATE_DIR=/state \ RUST_LOG=info,meta_transport_ipfs=info,libp2p=warn,beetswap=warn,yamux=warn ENTRYPOINT ["/usr/local/bin/meta-transport-ipfs"] diff --git a/README.md b/README.md index 5b89e41..f8e5a8e 100644 --- a/README.md +++ b/README.md @@ -23,11 +23,20 @@ meta-share's variables, unchanged: `P2P_LISTEN`, `PUBLIC_ADDR`, `KAD_*`, `META_SHARE_BLOCKSTORE_CACHE_MB`, `META_SHARE_INGRESS_*`, `META_SHARE_LOOSE_BLOCK_MAX_BYTES`, `META_SHARE_LOCAL_FILES_PATH`, `META_SHARE_FILES_PATH`, `META_SHARE_SEED_DHT_PROVIDE`, `META_SHARE_DATA`, plus -`META_SHARE_HULL_URL`. `meta_core_url` / `peer_url` come from the hull's -`settings.json` (mount `/config` read-only), env as the first-boot fallback. +`META_SHARE_HULL_URL`. `meta_core_url` / `peer_url` are asked from the hull at +boot (`GET /internal/network`), env as the fallback. -Start it **after** the hull is healthy: the hull migrates its material index out -of `ipfs/blocks.redb` on its first boot after the split (redb is single-process). +**Own settings** (config plane, `META_SHARE_PLUGIN_STATE_DIR`, default `/state` → +`config.json`): cohort namespace, DHT/pinned bootstrap peers, mDNS, DHT announce, +max connections, background completion. Edited from meta-share's dashboard +(Plugins → ipfs → configure); `config.json` wins and is written onto the env names +above at boot, so env only seeds it. Saving restarts the plugin. `PUBLIC_ADDR` / +`P2P_LISTEN` stay env-only (they follow the published port and the box's IP). + +Boot order is handled here, not by compose: before opening `ipfs/blocks.redb` the +plugin waits (≤ 60 s) for the hull's `/health`, because the hull migrates its +material index out of that file on its first boot after the split and redb is +single-process. ## Build diff --git a/src/config.rs b/src/config.rs index bd28b60..ec52019 100644 --- a/src/config.rs +++ b/src/config.rs @@ -268,7 +268,10 @@ pub struct Config { } impl Config { - pub fn from_env() -> Result { + /// `net`: the hull's endpoints (`GET /internal/network`), or + /// [`NetworkSettings::from_env`](crate::settings::NetworkSettings::from_env) + /// when the hull did not answer. + pub fn from_env(net: crate::settings::NetworkSettings) -> Result { // Reachability posture first — the listen addr, the kad bootstrap // list, and whether we publish provider records all fall out of it. // `P2P_ANNOUNCE` is the deprecated spelling of `PUBLIC_ADDR`. @@ -323,13 +326,7 @@ impl Config { )), }; - // The network endpoints are the hull's settings (`settings.json`, - // seeded once from env by the hull). Read-only here; env fallback when - // the file doesn't exist yet. - let config_dir: PathBuf = std::env::var("META_SHARE_CONFIG_DIR") - .unwrap_or_else(|_| "/config".to_string()) - .into(); - let net = crate::settings::NetworkSettings::load(&config_dir); + // The network endpoints are the hull's settings, handed in by `main`. let peer_api_url = net .peer_url .as_deref() diff --git a/src/main.rs b/src/main.rs index aeacbd5..6182e2a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,9 +6,12 @@ //! `cache/`. Serves the transport contract on `HTTP_LISTEN` (internal only) and //! the public `/ipfs/:cid` gateway, which the hull relays. //! -//! The swarm announces the **hull's** URL in identify (`baseUrl=`, from the -//! hull's `settings.json`), so peers and gateways keep calling meta-share's -//! public API exactly as before — the plugin is invisible on the wire. +//! The swarm announces the **hull's** URL in identify (`baseUrl=`, asked from the +//! hull at boot), so peers and gateways keep calling meta-share's public API +//! exactly as before — the plugin is invisible on the wire. +//! +//! Own settings: [`plane`] (`/config.json`, edited from meta-share's +//! dashboard), overlaid onto the env names the modules read. mod api; mod blockstore; @@ -19,6 +22,7 @@ mod ingress; mod ingress_commit; mod ipfs_chunk; mod material; +mod plane; mod plugin; mod settings; mod share; @@ -35,7 +39,8 @@ use std::sync::Arc; use anyhow::{Context, Result}; use libp2p::identity; -use meta_feeder_sdk::transport::{serve_transport, FocusView, HullClient}; +use meta_feeder_sdk::transport::config::state_dir_from_env; +use meta_feeder_sdk::transport::{serve_transport, ConfigPlane, FocusView, HullClient}; use tracing::{debug, info, warn}; use tracing_subscriber::EnvFilter; @@ -46,7 +51,35 @@ async fn main() -> Result<()> { }); tracing_subscriber::fmt().with_env_filter(filter).with_target(false).init(); - let cfg = config::Config::from_env()?; + // Own settings first: `config.json` wins over the env it was seeded from, + // written onto the env names every module reads — before any of them does. + let plane = Arc::new(ConfigPlane::new(plane::schema(), &state_dir_from_env(), plane::seed())); + plane::overlay_env(&plane.effective()); + + let http = reqwest::Client::builder() + .user_agent(concat!("meta-transport-ipfs/", env!("CARGO_PKG_VERSION"))) + .timeout(std::time::Duration::from_secs(30)) + .build() + .context("build shared reqwest client")?; + let hull = HullClient::from_env(http.clone()); + + // Wait for the hull before opening `ipfs/blocks.redb`: on its first boot + // after the split it migrates the material index out of that file, and redb + // is single-process. Compose `depends_on` used to order this; separate + // store apps cannot. Bounded — a hull that stays down must not wedge us. + if hull.is_enabled() && !hull.wait_ready(std::time::Duration::from_secs(60)).await { + warn!("meta-share (hull) did not answer within 60s; starting anyway"); + } + let net = match hull.network().await { + Ok(Some(n)) => settings::NetworkSettings::from(n), + Ok(None) => settings::NetworkSettings::from_env(), + Err(e) => { + warn!(error = %e, "could not ask the hull for its endpoints; using env"); + settings::NetworkSettings::from_env() + } + }; + + let cfg = config::Config::from_env(net)?; let keypair = identity::Keypair::generate_ed25519(); let local_peer_id = libp2p::PeerId::from(keypair.public()); let agent_version = swarm::build_agent_version(cfg.peer_api_url.as_deref()); @@ -61,17 +94,13 @@ async fn main() -> Result<()> { kad_provide_enabled = cfg.kad.provide_enabled, seed_dht_provide = cfg.seed_dht_provide, agent_version = %agent_version, + config = %plane.path().display(), "starting meta-transport-ipfs" ); let raw_block_store = blockstore::open_redb_blockstore(&cfg.data_dir) .await .context("open ipfs blockstore")?; - let http = reqwest::Client::builder() - .user_agent(concat!("meta-transport-ipfs/", env!("CARGO_PKG_VERSION"))) - .timeout(std::time::Duration::from_secs(30)) - .build() - .context("build shared reqwest client")?; let webdav_url_cache = Arc::new(tokio::sync::OnceCell::new()); let redb_handle = raw_block_store.raw_db(); let ingress = Arc::new(ingress::IngressRegistry::from_data_dir(&cfg.data_dir)); @@ -132,7 +161,7 @@ async fn main() -> Result<()> { peer_directory, focus, http: http.clone(), - hull: HullClient::from_env(http), + hull, data_dir: cfg.data_dir.clone(), meta_core_url: cfg.meta_core_url.clone(), webdav_url_cache, @@ -149,5 +178,5 @@ async fn main() -> Result<()> { } ingress_commit::spawn(Arc::clone(&state)); - serve_transport(Arc::new(plugin::IpfsPlugin { state }), cfg.http_addr).await + serve_transport(Arc::new(plugin::IpfsPlugin { state, config: plane }), cfg.http_addr).await } diff --git a/src/plane.rs b/src/plane.rs new file mode 100644 index 0000000..099a28b --- /dev/null +++ b/src/plane.rs @@ -0,0 +1,167 @@ +//! The plugin's own settings (config plane), edited from meta-share's dashboard +//! (Plugins → ipfs → configure) and stored in `/config.json`. +//! +//! Every knob here was — and still is — an env var read by the module that uses +//! it (`config.rs`, `swarm/`, `ingress_commit.rs`). Rather than re-thread each +//! one, [`overlay_env`] writes the effective values onto those same env names +//! once at boot, before anything reads them: `config.json` wins, the env seeds +//! it. Infra stays env-only: `PUBLIC_ADDR` / `P2P_LISTEN` (bound to the port +//! compose publishes and the box's IP) and the data dirs. + +use meta_feeder_sdk::transport::config::read; +use meta_feeder_sdk::{ConfigField, ConfigSchema}; +use serde_json::{json, Value}; + +const GIB: u64 = 1024 * 1024 * 1024; +const DEFAULT_NAMESPACE: &str = "metamesh-share-default"; + +pub fn schema() -> ConfigSchema { + ConfigSchema { + fields: vec![ + ConfigField::text("kad_namespace", "Cohort namespace") + .with_help("Peers only see others with the same namespace. Change it on every node of a private group."), + ConfigField::list("kad_bootstrap_peers", "DHT bootstrap peers") + .with_help("Multiaddrs used to join the public IPFS DHT. Empty = the app's default list. `none` disables the DHT."), + ConfigField::list("bootstrap_peers", "Pinned peers") + .with_help("Multiaddrs this node keeps dialled (e.g. a known gateway)."), + ConfigField::bool("enable_mdns", "LAN discovery (mDNS)"), + ConfigField::bool("seed_dht_provide", "Announce seeds on the DHT") + .with_help("Publish provider records for what this node seeds, so remote peers can find it."), + ConfigField::number("max_conns", "Max connections"), + ConfigField::bool("ingress_commit", "Background completion") + .with_help("Finish partially fetched IPFS content in the background."), + ConfigField::number("ingress_commit_max_concurrent", "Concurrent completions"), + ConfigField::number("ingress_commit_min_free_gib", "Disk floor (GiB)") + .with_help("Background completion stands down below this much free disk."), + ], + } +} + +fn env(key: &str) -> Option { + std::env::var(key).ok().map(|s| s.trim().to_string()).filter(|s| !s.is_empty()) +} + +fn env_list(key: &str) -> Vec { + env(key) + .map(|s| s.split(',').map(|p| p.trim().to_string()).filter(|p| !p.is_empty()).collect()) + .unwrap_or_default() +} + +fn env_flag(key: &str, default: bool) -> bool { + match env(key) { + Some(v) => !matches!(v.to_ascii_lowercase().as_str(), "0" | "false" | "no" | "off"), + None => default, + } +} + +fn env_u64(key: &str, default: u64) -> u64 { + env(key).and_then(|v| v.parse().ok()).unwrap_or(default) +} + +/// The values a plugin with no `config.json` runs with. Booleans are explicit +/// (the page's checkbox always submits one), so each carries the default the +/// reading module would apply. +pub fn seed() -> Value { + // `config.rs` defaults the seed announce to "public mode", i.e. a PUBLIC_ADDR. + let public = env("PUBLIC_ADDR").or_else(|| env("P2P_ANNOUNCE")).is_some(); + json!({ + "kad_namespace": env("KAD_NAMESPACE").unwrap_or_else(|| DEFAULT_NAMESPACE.to_string()), + "kad_bootstrap_peers": env_list("KAD_BOOTSTRAP_PEERS"), + "bootstrap_peers": env_list("BOOTSTRAP_PEERS"), + "enable_mdns": env_flag("ENABLE_MDNS", true), + "seed_dht_provide": env_flag("META_SHARE_SEED_DHT_PROVIDE", public), + "max_conns": env_u64("META_SHARE_MAX_CONNS", 128), + "ingress_commit": env_flag("META_SHARE_INGRESS_COMMIT", true), + "ingress_commit_max_concurrent": env_u64("META_SHARE_INGRESS_COMMIT_MAX_CONCURRENT", 2), + "ingress_commit_min_free_gib": env_u64("META_SHARE_INGRESS_COMMIT_MIN_FREE_BYTES", 5 * GIB) / GIB, + }) +} + +/// `(env name, value)` pairs the effective config sets. An unset/blank value +/// is skipped, leaving the env (or the module's default) in force. +pub fn env_overrides(v: &Value) -> Vec<(&'static str, String)> { + let mut out = Vec::new(); + let bool_s = |b: bool| if b { "true".to_string() } else { "false".to_string() }; + if let Some(s) = read::text(v, "kad_namespace") { + out.push(("KAD_NAMESPACE", s)); + } + if let Some(l) = read::list(v, "kad_bootstrap_peers").filter(|l| !l.is_empty()) { + // `none` is the documented "no DHT" spelling; config.rs reads set-but-empty that way. + let joined = if l.iter().any(|p| p.eq_ignore_ascii_case("none")) { String::new() } else { l.join(",") }; + out.push(("KAD_BOOTSTRAP_PEERS", joined)); + } + if let Some(l) = read::list(v, "bootstrap_peers").filter(|l| !l.is_empty()) { + out.push(("BOOTSTRAP_PEERS", l.join(","))); + } + if let Some(b) = read::flag(v, "enable_mdns") { + out.push(("ENABLE_MDNS", bool_s(b))); + } + if let Some(b) = read::flag(v, "seed_dht_provide") { + out.push(("META_SHARE_SEED_DHT_PROVIDE", bool_s(b))); + } + if let Some(n) = read::uint(v, "max_conns").filter(|n| *n > 0) { + out.push(("META_SHARE_MAX_CONNS", n.to_string())); + } + if let Some(b) = read::flag(v, "ingress_commit") { + out.push(("META_SHARE_INGRESS_COMMIT", bool_s(b))); + } + if let Some(n) = read::uint(v, "ingress_commit_max_concurrent").filter(|n| *n > 0) { + out.push(("META_SHARE_INGRESS_COMMIT_MAX_CONCURRENT", n.to_string())); + } + if let Some(n) = read::uint(v, "ingress_commit_min_free_gib") { + out.push(("META_SHARE_INGRESS_COMMIT_MIN_FREE_BYTES", n.saturating_mul(GIB).to_string())); + } + out +} + +/// Apply [`env_overrides`] to this process's environment. Call once at the top +/// of `main`, before any config is read and before other tasks run. +pub fn overlay_env(v: &Value) { + for (k, val) in env_overrides(v) { + std::env::set_var(k, val); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn get<'a>(o: &'a [(&'static str, String)], k: &str) -> Option<&'a str> { + o.iter().find(|(n, _)| *n == k).map(|(_, v)| v.as_str()) + } + + #[test] + fn page_values_map_onto_the_env_names_the_modules_read() { + let o = env_overrides(&json!({ + "kad_namespace": "private-cohort", + "kad_bootstrap_peers": ["/dns4/a/tcp/4001/p2p/X", "/dns4/b/tcp/4001/p2p/Y"], + "bootstrap_peers": [], + "enable_mdns": false, + "seed_dht_provide": true, + "max_conns": null, + "ingress_commit": true, + "ingress_commit_max_concurrent": 4, + "ingress_commit_min_free_gib": 10, + })); + assert_eq!(get(&o, "KAD_NAMESPACE"), Some("private-cohort")); + assert_eq!(get(&o, "KAD_BOOTSTRAP_PEERS"), Some("/dns4/a/tcp/4001/p2p/X,/dns4/b/tcp/4001/p2p/Y")); + assert_eq!(get(&o, "BOOTSTRAP_PEERS"), None, "an empty list keeps the env"); + assert_eq!(get(&o, "ENABLE_MDNS"), Some("false")); + assert_eq!(get(&o, "META_SHARE_MAX_CONNS"), None, "blank number keeps the env"); + assert_eq!(get(&o, "META_SHARE_INGRESS_COMMIT_MIN_FREE_BYTES"), Some("10737418240")); + } + + #[test] + fn none_disables_the_dht() { + let o = env_overrides(&json!({ "kad_bootstrap_peers": ["none"] })); + assert_eq!(get(&o, "KAD_BOOTSTRAP_PEERS"), Some("")); + } + + #[test] + fn every_seeded_key_is_in_the_schema() { + let keys: Vec = schema().fields.into_iter().map(|f| f.key).collect(); + for k in seed().as_object().unwrap().keys() { + assert!(keys.contains(k), "{k} missing from schema"); + } + } +} diff --git a/src/plugin.rs b/src/plugin.rs index c76cd46..3caada1 100644 --- a/src/plugin.rs +++ b/src/plugin.rs @@ -17,11 +17,11 @@ use meta_feeder_sdk::transport::ipfs::{ RelPath, ShareContainer, ShareLibrary, Shared, }; use meta_feeder_sdk::transport::{ - ApiError, Capabilities, Deleted, FocusView, Health, Job, Lane, Manifest, ReconcileReport, + ApiError, Capabilities, ConfigPlane, Deleted, FocusView, Health, Job, Lane, Manifest, ReconcileReport, ReconcileRequest, TransportPlugin, CONTRACT_VERSION, }; use serde::Deserialize; -use tracing::debug; +use tracing::{debug, warn}; use crate::api::files::{bitswap, bytes_to_response, PLAYER_HEADER}; use crate::api::AppState; @@ -34,6 +34,7 @@ const IMPORT_BODY_LIMIT: usize = 600 * 1024 * 1024; pub struct IpfsPlugin { pub state: Arc, + pub config: Arc, } fn lane_of(headers: &HeaderMap) -> Lane { @@ -53,6 +54,7 @@ impl TransportPlugin for IpfsPlugin { version: env!("CARGO_PKG_VERSION").into(), contract: CONTRACT_VERSION, capabilities: Capabilities { fetch: true, share: true }, + config: true, } } @@ -110,6 +112,10 @@ impl TransportPlugin for IpfsPlugin { &self.state.focus } + fn config(&self) -> Option> { + Some(Arc::clone(&self.config)) + } + fn extra_routes(self: Arc) -> Router { let gateway = Router::new() .route("/ipfs/:cid", get(crate::api::ipfs_gateway::get_ipfs)) @@ -144,6 +150,7 @@ impl TransportPlugin for IpfsPlugin { .route("/ipfs-tier/share/container", post(share_container)) .route("/ipfs-tier/stats/blockstore", get(blockstore_stats)) .route("/ipfs-tier/debug/:cid", get(debug_block)) + .route("/ipfs-tier/restart", post(restart)) .with_state(self) .merge(gateway) } @@ -422,3 +429,15 @@ async fn debug_block(State(p): S, Path(cid): Path) -> Response { pub fn bytes_response(bytes: Vec, range: Option<&HeaderValue>) -> Response { bytes_to_response(bytes, range) } + +/// The hull changed the endpoints this plugin announces (`peer_url`) or reads +/// through (`meta_core_url`): exit so `restart: unless-stopped` re-execs us and +/// `main` asks the hull again. +async fn restart() -> StatusCode { + tokio::spawn(async { + tokio::time::sleep(Duration::from_millis(300)).await; + warn!("restarting to pick up the hull's new network settings"); + std::process::exit(0); + }); + StatusCode::NO_CONTENT +} diff --git a/src/settings.rs b/src/settings.rs index 944ba40..ba36082 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -1,11 +1,9 @@ -//! The `network` section of the hull's `settings.json`, read-only. +//! The hull's two endpoints this plugin needs: `peer_url` (announced in +//! identify as `baseUrl=`) and `meta_core_url` (library byte reads). //! -//! The hull owns the file (it seeds it from the deprecated env vars on first -//! boot and edits it from the dashboard); this plugin only needs the two -//! endpoints, and falls back to the same env-derived defaults the hull would -//! seed when the file doesn't exist yet. - -use std::path::Path; +//! `main` asks the hull (`GET /internal/network`) at boot; when it does not +//! answer, [`NetworkSettings::from_env`] recomputes the same env-derived values +//! the hull seeds its own settings from. use serde::Deserialize; use tracing::warn; @@ -21,20 +19,6 @@ pub struct NetworkSettings { } impl NetworkSettings { - /// `/settings.json` → `network`, else [`Self::from_env`]. - pub fn load(config_dir: &Path) -> Self { - #[derive(Deserialize, Default)] - #[serde(default)] - struct File { - network: Option, - } - std::fs::read(config_dir.join("settings.json")) - .ok() - .and_then(|b| serde_json::from_slice::(&b).ok()) - .and_then(|f| f.network) - .unwrap_or_else(Self::from_env) - } - /// The hull's first-boot seed, recomputed: `META_CORE_URL`, and /// `INTERNAL_BASE_URL` → `BASE_URL` → [`DEFAULT_PEER_URL`]. pub fn from_env() -> Self { @@ -63,22 +47,21 @@ fn env_nonempty(key: &str) -> Option { .filter(|s| !s.is_empty()) } +impl From for NetworkSettings { + fn from(n: meta_feeder_sdk::transport::NetworkInfo) -> Self { + Self { + meta_core_url: n.meta_core_url, + peer_url: n.peer_url, + } + } +} + #[cfg(test)] mod tests { use super::*; #[test] - fn reads_the_network_section_of_the_hulls_file() { - let d = std::env::temp_dir().join(format!("ipfs-settings-{}", std::process::id())); - std::fs::create_dir_all(&d).unwrap(); - std::fs::write( - d.join("settings.json"), - r#"{"network":{"meta_core_url":"http://mc:9000","peer_url":"https://share.example"},"usenet":{"nntp_host":"x"}}"#, - ) - .unwrap(); - let n = NetworkSettings::load(&d); - assert_eq!(n.meta_core_url.as_deref(), Some("http://mc:9000")); - assert_eq!(n.peer_url.as_deref(), Some("https://share.example")); - let _ = std::fs::remove_dir_all(&d); + fn the_env_fallback_always_has_a_peer_url() { + assert!(NetworkSettings::from_env().peer_url.is_some()); } }