diff options
Diffstat (limited to 'src/transport/mycelium')
| -rw-r--r-- | src/transport/mycelium/mod.rs | 251 |
1 files changed, 202 insertions, 49 deletions
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 5e8dc05..9078134 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -91,9 +91,145 @@ pub(crate) struct MyceliumTransport { local_id: String, } +/// This transport's options, checked when the configuration is loaded +/// (`config::Settings::check`): a daemon elsewhere (`mycelium-rest-url`) by a URL, and +/// none of the options only a supervised one takes with it; a private network with +/// both its key and its name. +pub(crate) fn check(settings: &crate::config::Settings) -> Result<()> { + if let Some(url) = &settings.mycelium_rest_url { + Url::parse(url).with_context(|| format!("mycelium-rest-url {url:?}"))?; + if let Some((name, _)) = supervised_only(settings).iter().find(|(_, set)| *set) { + anyhow::bail!( + "{name} is for a supervised Mycelium daemon and can't be combined with mycelium-rest-url" + ); + } + } + if settings.mycelium_key_create && settings.mycelium_key_file.is_none() { + anyhow::bail!("mycelium-key-create creates the key at mycelium-key-file; set that too"); + } + // A private network has both; one alone would be ignored or invented. + if settings.mycelium_network_key_file.is_some() != settings.mycelium_network_name.is_some() { + anyhow::bail!("mycelium-network-key-file and mycelium-network-name go together"); + } + Ok(()) +} + +/// The options only a supervised Mycelium daemon takes, and whether each is set: the +/// other side of `config::CONFLICTS`' `mycelium-rest-url` (a test keeps them the same). +pub(crate) fn supervised_only(settings: &crate::config::Settings) -> [(&'static str, bool); 5] { + [ + ("mycelium-key-file", settings.mycelium_key_file.is_some()), + ("mycelium-key-create", settings.mycelium_key_create), + ( + "mycelium-network-key-file", + settings.mycelium_network_key_file.is_some(), + ), + ( + "mycelium-network-name", + settings.mycelium_network_name.is_some(), + ), + ( + "mycelium-tcp-listen-port", + settings.mycelium_tcp_listen_port.is_some(), + ), + ] +} + +/// The transport as `settings` configure it (`transport::start`): its daemon supervised +/// (started, which can take a while) or one running elsewhere. None if `stop` came +/// first. +pub(crate) async fn start( + settings: &crate::config::Settings, + mut stop: tokio::sync::watch::Receiver<bool>, +) -> Result<Option<MyceliumTransport>> { + // The daemon's key, 32 raw bytes (as `daemon` hands it an ephemeral one): read, + // or made, here, so a missing file fails before the daemon starts. + if let Some(path) = &settings.mycelium_key_file { + let (bytes, created) = crate::files::key_file( + path, + "Mycelium key", + settings.mycelium_key_create, + "mycelium-key-create", + "a new Mycelium address", + 32, + )?; + if bytes.len() != 32 { + anyhow::bail!( + "The Mycelium key in {} has {} bytes; it has 32", + path.display(), + bytes.len() + ); + } + if created { + tracing::info!(path = %path.display(), "Created a new Mycelium key"); + } + } + let spawn_config = MyceliumSpawnConfig { + rest_url_opt: settings + .mycelium_rest_url + .as_deref() + .and_then(|url| Url::parse(url).ok()), + key_file_opt: settings.mycelium_key_file.clone(), + network_key_file: settings.mycelium_network_key_file.clone(), + network_name: settings.mycelium_network_name.clone(), + tcp_port: settings.mycelium_tcp_listen_port, + peers: settings.mycelium_peers.clone(), + }; + let setup = tokio::select! { + setup = MyceliumSetup::bootstrap(spawn_config) => setup?, + _ = stop.wait_for(|stop| *stop) => return Ok(None), + }; + // Without a password it may carry (the option is secret). + let mut shown_url = setup.rest_url.clone(); + let _ = shown_url.set_password(None); + tracing::debug!(url = shown_url.to_string(), "Using Mycelium REST API"); + tokio::select! { + transport = MyceliumTransport::connect(setup) => Ok(Some(transport?)), + _ = stop.wait_for(|stop| *stop) => Ok(None), + } +} + +/// How long the daemon's API may take to come up: a daemon just started listens a +/// moment later (`admin_info`). +const API_READY_WITHIN: Duration = Duration::from_secs(30); + +/// The daemon's identity, from its API at `url` (`/api/v1/admin`), once the API answers: +/// a daemon we just started doesn't listen yet, so refused connections are tried again +/// (50 ms, doubling up to a second) for `API_READY_WITHIN`; anything else is an error. +async fn admin_info(client: &reqwest::Client, url: Url) -> Result<MyceliumAdminInfo> { + let deadline = tokio::time::Instant::now() + API_READY_WITHIN; + let mut wait = Duration::from_millis(50); + loop { + match client.get(url.clone()).send().await { + Ok(response) => { + return response + .error_for_status()? + .json() + .await + .context("Querying local Mycelium identity"); + } + Err(error) if error.is_connect() && tokio::time::Instant::now() + wait < deadline => { + tracing::debug!(url = %url, error = %error, "Mycelium's API doesn't answer yet; trying again"); + tokio::time::sleep(wait).await; + wait = (wait * 2).min(Duration::from_secs(1)); + } + Err(error) => { + return Err(error).with_context(|| { + format!( + "Querying local Mycelium identity (no answer within {API_READY_WITHIN:?})" + ) + }); + } + } + } +} + impl MyceliumTransport { pub(crate) async fn connect(setup: MyceliumSetup) -> Result<Self> { let rest_url = setup.rest_url.clone(); + // reqwest's TLS has no crypto provider of its own (`Cargo.toml`): ring, for the + // whole process, unless one is installed already. + let _ = rustls::crypto::ring::default_provider().install_default(); let client = reqwest::Client::builder() .timeout(REQUEST_TIMEOUT) @@ -102,14 +238,7 @@ impl MyceliumTransport { .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30)) .build()?; - let admin: MyceliumAdminInfo = client - .get(rest_url.join("/api/v1/admin")?) - .send() - .await? - .error_for_status()? - .json() - .await - .context("Querying local Mycelium identity")?; + let admin = admin_info(&client, rest_url.join("/api/v1/admin")?).await?; let transport = Self { setup: tokio::sync::Mutex::new(setup), @@ -246,31 +375,17 @@ impl Transport for MyceliumTransport { .into()); } - // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing - // a message that never arrived. - if tracing::enabled!(tracing::Level::DEBUG) { - let Ok(body) = response + // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing a + // message that never arrived. The message itself is logged above, in the paths + // layer (`transport::paths`). + if tracing::enabled!(tracing::Level::TRACE) { + let id = response .bytes() .await - .map_err(anyhow::Error::from) - .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?)) - else { - tracing::debug!(to = %Addressed(SCHEME, &message.peer), "Sent message (couldn't decode response from Mycelium REST API)"); - return Ok(()); - }; - let Ok(outbound) = serde_json::from_slice::<OutboundMessage>(&message.body) else { - tracing::debug!(to = %Addressed(SCHEME, &message.peer), body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); - return Ok(()); - }; - let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { - tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?outbound, "Sent message (couldn't decode outbound message)"); - return Ok(()); - }; - let Some(id) = body.get("id").and_then(|id| id.as_str()) else { - tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); - return Ok(()); - }; - tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?parsed, id = %id, "Sent message"); + .ok() + .and_then(|bytes| serde_json::from_slice::<serde_json::Value>(&bytes).ok()) + .and_then(|body| body.get("id")?.as_str().map(str::to_owned)); + tracing::trace!(to = %Addressed(SCHEME, &message.peer), id = ?id, "Mycelium took the message"); } Ok(()) } @@ -300,12 +415,11 @@ impl Transport for MyceliumTransport { while let Some(inbound_message) = rx.recv().await { match parse_message(&inbound_message) { Ok((sender, parsed)) => { - tracing::debug!( + // The message itself is logged above, in the paths layer. + tracing::trace!( from = %Addressed(SCHEME, &inbound_message.src_pk), - message = ?parsed, id = %inbound_message.id, - //payload_bytes = raw_bytes.len(), - "Received message" + "Mycelium delivered a message" ); let payload_len = base64_decoded_len(&inbound_message.payload); if is_probe(&parsed) { @@ -464,6 +578,13 @@ pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<Enco }) } +/// Mycelium's envelope, for simulations that run on it (`transport::sim::Codec`). +#[cfg(test)] +pub(crate) const CODEC: crate::transport::sim::Codec = crate::transport::sim::Codec { + encode: encode_message, + decode: decode_message, +}; + /// The message in a body made by [`encode_message`]. #[cfg(test)] pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> { @@ -562,10 +683,10 @@ pub(crate) enum TransportError { #[cfg(test)] mod tests { use super::*; + use crate::document::ChangeHash; use crate::transport::{ BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, MAX_LISTED_HEADS, Status, }; - use automerge::ChangeHash; /// Mycelium splits messages into chunks of this size. const MYCELIUM_CHUNK: usize = 1300; @@ -595,6 +716,7 @@ mod tests { epoch_seq: u32::MAX, ready: true, syncs_with_you: true, + leaving: false, }; let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap(); // Mycelium's message data: a length byte, the topic, then our payload. @@ -657,20 +779,51 @@ mod tests { )); } + /// Every kind decodes from Mycelium's API body (its envelope: the topic, the payload + /// in base64) to what was encoded. #[test] - fn messages_round_trip_through_the_api_body() { - let status = ParsedMessage::Status(Status { - heads: HeadsSummary::of(vec![ChangeHash([1; 32])]), - epoch: Epoch::GENESIS, - epoch_seq: 0, - ready: true, - syncs_with_you: true, - }); - let encoded = encode_message("peer", &status).unwrap(); - assert!(matches!( - decode_message(&encoded.body).unwrap(), - ParsedMessage::Status(_) - )); + fn every_kind_round_trips_through_the_api_body() { + for message in wire::samples() { + let encoded = encode_message("peer", &message).unwrap(); + assert_eq!(encoded.kind, message.kind()); + let decoded = decode_message(&encoded.body).unwrap(); + assert_eq!( + wire::encode(&decoded).unwrap(), + wire::encode(&message).unwrap(), + "{message:?}" + ); + assert_eq!(encoded.payload_len, wire::encode(&message).unwrap().len()); + } + } + + /// A swarm whose every message goes through Mycelium's envelope: joining, writes + /// reaching everyone, and concurrent writes merging, as the simulations do with the + /// wire encoding alone. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_swarm_converges_over_the_envelope() { + use crate::simtest::{Sim, read, write}; + use crate::swarm::bootstrap::Start; + use serde_json::json; + const SOON: Duration = Duration::from_secs(10); + + let sim = Sim::with_codec(11, CODEC); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], SOON).await; + write(b, "users/bob", json!({"name": "Bob"})).await; + sim.wait_until("c has b's write", SOON, move || async move { + read(c, "users/bob/name").await == Some(json!("Bob")) + }) + .await; + write(a, "users/alice", json!(1)).await; + write(c, "users/carol", json!(3)).await; + sim.wait_converged(&[a, b, c], SOON).await; + for node in [a, b, c] { + assert_eq!(read(node, "users/alice").await, Some(json!(1))); + assert_eq!(read(node, "users/carol").await, Some(json!(3))); + } } #[test] |