summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/api/auth.rs6
-rw-r--r--src/api/debug.rs2
-rw-r--r--src/api/log.rs32
-rw-r--r--src/api/mod.rs11
-rw-r--r--src/api/peers.rs4
-rw-r--r--src/api/store.rs2
-rw-r--r--src/document/store.rs6
-rw-r--r--src/logging.rs58
-rw-r--r--src/main.rs20
-rw-r--r--src/remake/api.rs20
-rw-r--r--src/remake/catchup.rs8
-rw-r--r--src/remake/coordinator.rs2
-rw-r--r--src/remake/delta.rs2
-rw-r--r--src/remake/kind.rs12
-rw-r--r--src/remake/local.rs2
-rw-r--r--src/remake/mod.rs12
-rw-r--r--src/remake/quarantine.rs8
-rw-r--r--src/remake/records.rs2
-rw-r--r--src/remake/snapshot.rs8
-rw-r--r--src/remake/switch.rs4
-rw-r--r--src/state.rs70
-rw-r--r--src/swarm/auth.rs4
-rw-r--r--src/swarm/bootstrap.rs8
-rw-r--r--src/swarm/mod.rs1
-rw-r--r--src/swarm/pair.rs78
-rw-r--r--src/swarm/peer.rs28
-rw-r--r--src/swarm/sync.rs44
-rw-r--r--src/transport/firewall.rs4
-rw-r--r--src/transport/mod.rs5
-rw-r--r--src/transport/mycelium.rs114
30 files changed, 385 insertions, 192 deletions
diff --git a/src/api/auth.rs b/src/api/auth.rs
index 3926c70..70b1af5 100644
--- a/src/api/auth.rs
+++ b/src/api/auth.rs
@@ -1,14 +1,14 @@
-use argon2::{password_hash::PasswordVerifier, Argon2};
+use argon2::{Argon2, password_hash::PasswordVerifier};
use axum::extract::State;
use axum::{
body::Body,
http::{
- header::{AUTHORIZATION, WWW_AUTHENTICATE},
HeaderMap, Request, StatusCode,
+ header::{AUTHORIZATION, WWW_AUTHENTICATE},
},
response::{IntoResponse, Response},
};
-use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
+use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
use dashmap::DashMap;
use password_hash::phc::PasswordHash;
use std::{
diff --git a/src/api/debug.rs b/src/api/debug.rs
index 929d665..6a56a9c 100644
--- a/src/api/debug.rs
+++ b/src/api/debug.rs
@@ -2,7 +2,7 @@
use std::collections::HashSet;
-use axum::{body::Bytes, extract::State, http::StatusCode, Json};
+use axum::{Json, body::Bytes, extract::State, http::StatusCode};
use tracing::{info, warn};
use crate::state::AppState;
diff --git a/src/api/log.rs b/src/api/log.rs
new file mode 100644
index 0000000..5d44926
--- /dev/null
+++ b/src/api/log.rs
@@ -0,0 +1,32 @@
+//! /admin/log: read and replace the log filter of the running node.
+
+use axum::{body::Bytes, http::StatusCode};
+use tracing::{info, warn};
+
+use crate::logging;
+
+/// `GET /admin/log`: the filter in effect, as plain text in `RUST_LOG` syntax.
+pub(crate) async fn get_log_filter() -> Result<String, StatusCode> {
+ logging::filter().map_err(|error| {
+ warn!(error = %format!("{error:#}"), "Cannot read the log filter");
+ StatusCode::INTERNAL_SERVER_ERROR
+ })
+}
+
+/// `PUT /admin/log`: replace the filter; the body is plain text in `RUST_LOG` syntax,
+/// e.g. `info,continuum::swarm::sync=trace`. Answers with the filter now in effect;
+/// 400 (with the reason) if the body doesn't parse, and the old filter stays.
+pub(crate) async fn put_log_filter(body: Bytes) -> Result<String, (StatusCode, String)> {
+ let directives = std::str::from_utf8(&body)
+ .map_err(|_| {
+ (
+ StatusCode::BAD_REQUEST,
+ "The filter must be UTF-8\n".to_owned(),
+ )
+ })?
+ .trim();
+ let applied = logging::set_filter(directives)
+ .map_err(|error| (StatusCode::BAD_REQUEST, format!("{error:#}\n")))?;
+ info!(filter = %applied, "Log filter replaced");
+ Ok(applied + "\n")
+}
diff --git a/src/api/mod.rs b/src/api/mod.rs
index ea3ad41..a2bd7e0 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -2,17 +2,18 @@
pub(crate) mod auth;
mod debug;
+mod log;
mod peers;
mod store;
use axum::{
+ Router,
http::{
- header::{CONTENT_TYPE, ETAG},
HeaderValue, StatusCode,
+ header::{CONTENT_TYPE, ETAG},
},
response::{IntoResponse, Response},
routing::{delete, get, put},
- Router,
};
use serde_json::Value as JsonValue;
use sha2::{Digest, Sha256};
@@ -20,7 +21,7 @@ use tracing::{error, warn};
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;
-use crate::api::auth::{basic_auth, AuthState};
+use crate::api::auth::{AuthState, basic_auth};
use crate::document::store::StoreError;
use crate::state::AppState;
@@ -37,6 +38,10 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
.route("/stream/{*key}", get(store::stream_store_entry))
.route("/peers", get(peers::list_peers))
.route("/peers/{peer}", delete(peers::delete_peer))
+ .route(
+ "/admin/log",
+ get(log::get_log_filter).put(log::put_log_filter),
+ )
.merge(crate::remake::router::<AppState>());
if debug_api {
diff --git a/src/api/peers.rs b/src/api/peers.rs
index d60e7b3..65f8fa1 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -3,9 +3,9 @@
use std::sync::Arc;
use axum::{
+ Json,
extract::{Path, Query, State},
http::StatusCode,
- Json,
};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
@@ -16,7 +16,7 @@ use utoipa::ToSchema;
use super::store_status;
use crate::document::replica::Epoch;
use crate::document::store;
-use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState};
+use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState, unix_millis};
use crate::swarm::peer::forgotten_path;
#[derive(Debug, Serialize, ToSchema)]
diff --git a/src/api/store.rs b/src/api/store.rs
index 409a530..d6ff306 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -7,8 +7,8 @@ use axum::{
body::{Body, Bytes},
extract::{Path, Query, State},
http::{
- header::{ACCEPT, CONTENT_TYPE},
HeaderMap, HeaderValue, StatusCode,
+ header::{ACCEPT, CONTENT_TYPE},
},
response::{IntoResponse, Response},
};
diff --git a/src/document/store.rs b/src/document/store.rs
index 9db8300..9e18693 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -2,9 +2,9 @@
use anyhow::anyhow;
use automerge::{
- transaction::Transactable, ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value,
+ ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value, transaction::Transactable,
};
-use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
+use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
use serde::Serialize;
use serde_json::Value as JsonValue;
use std::collections::BTreeSet;
@@ -97,7 +97,7 @@ fn resolved_to_json(
let maps = match resolved {
Resolved::Maps(maps) => maps,
Resolved::Other(Value::Scalar(scalar), _) => {
- return scalar_to_json(scalar.clone().into_owned())
+ return scalar_to_json(scalar.clone().into_owned());
}
Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, log),
};
diff --git a/src/logging.rs b/src/logging.rs
new file mode 100644
index 0000000..4496125
--- /dev/null
+++ b/src/logging.rs
@@ -0,0 +1,58 @@
+//! Logging: a compact fmt logger whose filter can be replaced while the node runs
+//! (`/admin/log`), and, with the `tokio-console` feature, a tokio-console layer.
+//!
+//! The filter starts from `RUST_LOG`, or `info` without it. It applies to the fmt
+//! logger only, so the console layer keeps getting the runtime events it needs.
+
+use std::sync::OnceLock;
+
+use anyhow::{Result, anyhow};
+use tracing_subscriber::{EnvFilter, Registry, prelude::*, reload};
+
+type FilterHandle = reload::Handle<EnvFilter, Registry>;
+
+static FILTER: OnceLock<FilterHandle> = OnceLock::new();
+
+/// Installs the global subscriber. Call once, first thing in main.
+pub(crate) fn init() {
+ let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+ let (filter, handle) = reload::Layer::new(filter);
+
+ let registry = tracing_subscriber::registry().with(
+ tracing_subscriber::fmt::layer()
+ .compact()
+ .with_filter(filter),
+ );
+
+ #[cfg(feature = "tokio-console")]
+ let registry = registry
+ .with(console_subscriber::spawn().with_filter(EnvFilter::new("tokio=trace,runtime=trace")));
+
+ registry.init();
+ let _ = FILTER.set(handle);
+}
+
+/// The filter in effect, in `RUST_LOG` syntax.
+pub(crate) fn filter() -> Result<String> {
+ handle()?
+ .with_current(|filter| filter.to_string())
+ .map_err(|error| anyhow!("Reading the log filter: {error}"))
+}
+
+/// Replaces the filter. `directives` uses `RUST_LOG` syntax, e.g.
+/// `info,continuum::swarm::sync=trace`. An invalid one leaves the old filter in place.
+pub(crate) fn set_filter(directives: &str) -> Result<String> {
+ let filter = EnvFilter::try_new(directives)
+ .map_err(|error| anyhow!("Invalid log filter {directives:?}: {error}"))?;
+ let applied = filter.to_string();
+ handle()?
+ .reload(filter)
+ .map_err(|error| anyhow!("Replacing the log filter: {error}"))?;
+ Ok(applied)
+}
+
+fn handle() -> Result<&'static FilterHandle> {
+ FILTER
+ .get()
+ .ok_or_else(|| anyhow!("Logging is not initialised"))
+}
diff --git a/src/main.rs b/src/main.rs
index 96758ec..5abc0e2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -7,19 +7,20 @@ use std::collections::HashMap;
use std::future::IntoFuture;
use std::net::SocketAddr;
use std::path::PathBuf;
-use std::sync::atomic::AtomicBool;
use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
use std::time::Duration;
use anyhow::{Context, Result};
use automerge::AutoCommit;
use clap::Parser;
-use tokio::sync::{broadcast, mpsc, watch, RwLock};
+use tokio::sync::{RwLock, broadcast, mpsc, watch};
use tracing::{debug, error, info, trace, warn};
use url::Url;
mod api;
mod document;
+mod logging;
mod remake;
mod state;
mod swarm;
@@ -29,9 +30,9 @@ use api::auth::AuthState;
use document::replica::{Epoch, Replica};
use state::{AppState, NodeLifecycle, Timing};
use swarm::{bootstrap, peer, sync};
+use transport::Transport;
use transport::firewall::{Firewall, FirewallTransport};
use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig};
-use transport::Transport;
#[derive(Parser, Debug)]
#[command(version, about = "continuum messaging system", long_about = None)]
@@ -67,15 +68,20 @@ struct Args {
use_value_delimiter = true,
default_value = "tls://127.0.0.1:9651"
)]
+ /// Peers of the supervised Mycelium daemon (overlay links such as
+ /// `tls://host:9651`), not continuum peers.
pub mycelium_peers: Vec<String>,
+ /// Seeds: how to reach a node of the swarm to join, as a Mycelium public key, a
+ /// Mycelium IPv6 address, or a host name with an AAAA record for one. Without seeds,
+ /// the node starts a new swarm (genesis).
#[arg(long, value_name = "STRING,[STRING...]", value_delimiter = ',')]
seeds: Option<Vec<String>>,
#[arg(long, value_name = "PATH")]
htpasswd: String,
- /// Pre-shared swarm key: only peers that know it may join (raw bytes, at least 16;
+ /// Pre-shared swarm key: only nodes that know it may join (raw bytes, at least 16;
/// e.g. `head -c 32 /dev/urandom`). Without it, anyone who knows the swarm ID can.
#[arg(long, value_name = "PATH")]
swarm_key_file: Option<PathBuf>,
@@ -92,7 +98,7 @@ struct Args {
#[tokio::main]
async fn main() -> Result<()> {
- console_subscriber::init();
+ logging::init();
let args = Args::parse();
@@ -100,7 +106,7 @@ async fn main() -> Result<()> {
let swarm_key = match &args.swarm_key_file {
Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)),
None => {
- warn!("No swarm key: any peer that learns the swarm ID can join");
+ warn!("No swarm key: any node that learns the swarm ID can join");
None
}
};
@@ -281,7 +287,7 @@ async fn main() -> Result<()> {
/// Turns SIGINT and SIGTERM into a shutdown request. A second signal exits at once (the
/// supervised Mycelium still gets its parent-death signal then).
fn watch_for_shutdown() -> Result<watch::Receiver<bool>> {
- use tokio::signal::unix::{signal, SignalKind};
+ use tokio::signal::unix::{SignalKind, signal};
let mut interrupt = signal(SignalKind::interrupt())?;
let mut terminate = signal(SignalKind::terminate())?;
diff --git a/src/remake/api.rs b/src/remake/api.rs
index fbd5f54..988bb34 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -32,14 +32,14 @@ use std::time::Duration;
use automerge::ChangeHash;
use axum::{
+ Json,
body::Bytes,
extract::{Path, Query, State},
- http::{header::LOCATION, StatusCode},
+ http::{StatusCode, header::LOCATION},
response::{IntoResponse, Response},
- Json,
};
use serde::{Deserialize, Serialize};
-use serde_json::{json, Value as JsonValue};
+use serde_json::{Value as JsonValue, json};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
use uuid::Uuid;
@@ -47,9 +47,9 @@ use uuid::Uuid;
use super::coordinator;
use super::local::LocalStatus;
use super::records::{
- self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention,
+ self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, unix_millis,
};
-use super::{kind, Host};
+use super::{Host, kind};
use crate::document::store::StoreError;
#[derive(Debug, Default, Deserialize)]
@@ -220,7 +220,7 @@ async fn start_kind<H: Host>(
StatusCode::BAD_REQUEST,
format!("invalid request: {error}"),
None,
- )
+ );
}
}
};
@@ -247,7 +247,7 @@ async fn start_kind<H: Host>(
StatusCode::BAD_REQUEST,
format!("invalid {kind_name} spec: {reason}"),
None,
- )
+ );
}
};
@@ -339,7 +339,7 @@ async fn start_kind<H: Host>(
let id = Uuid::new_v4().to_string();
let grace = grace(&host);
- let mut proposal = Proposal {
+ let proposal = Proposal {
coordinator: host.node_id(),
from_epoch: host.current_epoch().into(),
base_heads: Vec::new(),
@@ -559,14 +559,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
StatusCode::CONFLICT,
"already committed; the switch cannot be undone",
Some(json!({ "remake": current })),
- )
+ );
}
Phase::Aborted | Phase::Expired => {
return error_response(
StatusCode::CONFLICT,
"already over",
Some(json!({ "remake": current })),
- )
+ );
}
}
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 4aa66e7..db25251 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -31,13 +31,13 @@ use automerge::{AutoCommit, ChangeHash};
use tokio::time::Instant;
use tracing::{error, info, warn};
+use super::Host;
use super::delta::{self, ReplayStats};
use super::kind::Translator;
use super::local::{CatchUp, LocalStatus, Rejoin};
use super::quarantine;
use super::records::unix_millis;
use super::snapshot::{self, EpochMetadata};
-use super::Host;
use crate::document::replica::Epoch;
/// Pause before trying the same target again after a failed rejoin.
@@ -444,9 +444,9 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
#[cfg(test)]
mod tests {
use super::*;
- use crate::remake::snapshot::{build, SnapshotParams};
- use automerge::{transaction::Transactable, ActorId, ROOT};
- use serde_json::{json, Value as JsonValue};
+ use crate::remake::snapshot::{SnapshotParams, build};
+ use automerge::{ActorId, ROOT, transaction::Transactable};
+ use serde_json::{Value as JsonValue, json};
/// An old-epoch document with one own change after the base, and the swarm's
/// document of the next epoch (snapshot plus one foreign change).
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index d1b3397..28140cb 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -20,9 +20,9 @@ use std::time::Duration;
use tracing::{info, warn};
+use super::Host;
use super::records::{self, Decision, Record};
use super::snapshot;
-use super::Host;
use crate::document::store::StoreError;
const TICK: Duration = Duration::from_secs(1);
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index c161e2c..d318c00 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -268,7 +268,7 @@ fn patch_path(patch: &automerge::Patch) -> Vec<String> {
#[cfg(test)]
mod tests {
use super::*;
- use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT};
+ use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, transaction::Transactable};
use serde_json::json;
fn view(doc: &AutoCommit) -> serde_json::Value {
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
index b709a05..7d1300f 100644
--- a/src/remake/kind.rs
+++ b/src/remake/kind.rs
@@ -20,7 +20,7 @@ use automerge::AutoCommit;
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
-use super::snapshot::{canonical, META_ROOT};
+use super::snapshot::{META_ROOT, canonical};
use crate::document::store::{self, StoreError};
/// Name of the kind used when a request doesn't name one.
@@ -205,7 +205,7 @@ impl RemakeKind for Replace {
_ => {
return Err(format!(
"rule for {from:?} needs exactly one of \"to\" and \"drop\""
- ))
+ ));
}
};
if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) {
@@ -399,10 +399,10 @@ mod tests {
assert!(replace(json!({ "document": [] })).is_err());
assert!(replace(json!({ "document": { META_ROOT: {} } })).is_err());
assert!(replace(json!({ "document": {}, "rules": [{ "from": "a" }] })).is_err());
- assert!(replace(
- json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] })
- )
- .is_err());
+ assert!(
+ replace(json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] }))
+ .is_err()
+ );
assert!(
replace(json!({ "document": {}, "rules": [{ "from": "a//b", "to": "c" }] })).is_err()
);
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 35a4ff1..d2ea2c2 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -12,7 +12,7 @@ use automerge::AutoCommit;
use serde::Serialize;
use tokio::time::Instant;
-use super::records::{unix_millis, Retention};
+use super::records::{Retention, unix_millis};
use super::snapshot::Snapshot;
use crate::document::replica::Epoch;
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index ff3c9c3..2429ad7 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -37,8 +37,8 @@ use std::time::Duration;
use automerge::AutoCommit;
use axum::{
- routing::{get, post},
Router,
+ routing::{get, post},
};
use tokio::sync::broadcast;
@@ -52,7 +52,7 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
/// This node's in-memory remake state.
fn remake(&self) -> &RemakeState;
- /// This node's peer ID.
+ /// This node's ID (its peers know it by this).
fn node_id(&self) -> String;
/// The epoch of the local replica.
@@ -92,12 +92,6 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
F: FnOnce(&AutoCommit) -> T + Send + 'static,
T: Send + 'static;
- /// Mutable access to the document without publishing a change (e.g. forks).
- fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
- where
- F: FnOnce(&mut AutoCommit) -> T + Send + 'static,
- T: Send + 'static;
-
/// One committed change (rolled back on error), published and replicated.
fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
where
@@ -123,8 +117,6 @@ pub(crate) struct PeerSummary {
pub(crate) id: String,
/// Not dormant.
pub(crate) active: bool,
- /// Handshake completed.
- pub(crate) established: bool,
/// The epoch the peer last advertised.
pub(crate) epoch: Option<Epoch>,
}
diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs
index deed8cd..1090e2c 100644
--- a/src/remake/quarantine.rs
+++ b/src/remake/quarantine.rs
@@ -16,22 +16,22 @@ use std::convert::Infallible;
use automerge::AutoCommit;
use axum::{
+ Json,
extract::{Path, State},
- http::{header, StatusCode},
+ http::{StatusCode, header},
response::{IntoResponse, Response},
- Json,
};
use serde::Serialize;
-use serde_json::{json, Value as JsonValue};
+use serde_json::{Value as JsonValue, json};
use tracing::{info, warn};
use uuid::Uuid;
+use super::Host;
use super::api::error_response;
use super::delta;
use super::kind::Translator;
use super::local::Quarantined;
use super::records::unix_millis;
-use super::Host;
use crate::document::replica::Epoch;
use crate::document::store;
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 600d4dc..5c844f1 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -21,7 +21,7 @@ use serde_json::Value as JsonValue;
use tracing::warn;
use super::kind::DEFAULT_KIND;
-use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES};
+use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index bd99854..efe7e37 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -24,11 +24,11 @@
use std::collections::BTreeMap;
-use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash};
-use serde_json::{json, Map as JsonMap, Value as JsonValue};
+use automerge::{ActorId, AutoCommit, ChangeHash, transaction::CommitOptions};
+use serde_json::{Map as JsonMap, Value as JsonValue, json};
use sha2::{Digest, Sha256};
-use super::kind::{by_name, RemakeKind, Translator, DEFAULT_KIND};
+use super::kind::{DEFAULT_KIND, RemakeKind, Translator, by_name};
use super::records::{EpochRecord, Policy};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
@@ -353,7 +353,7 @@ pub(super) fn canonical(value: JsonValue) -> JsonValue {
#[cfg(test)]
mod tests {
use super::*;
- use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT};
+ use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, Value, transaction::Transactable};
fn doc(actor: &[u8]) -> AutoCommit {
AutoCommit::new().with_actor(ActorId::from(actor))
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 2670f14..f4a018a 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -24,14 +24,14 @@ use automerge::{ActorId, AutoCommit, ChangeHash};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
+use super::Host;
use super::catchup;
use super::delta::{self, ReplayStats};
use super::kind::Translator;
use super::local::LocalStatus;
use super::quarantine;
-use super::records::{self, unix_millis, EpochRecord, Record};
+use super::records::{self, EpochRecord, Record, unix_millis};
use super::snapshot::{self, META_ROOT, REMAKES};
-use super::Host;
use crate::document::replica::Epoch;
use crate::document::store;
use crate::swarm::sync::paths_related;
diff --git a/src/state.rs b/src/state.rs
index 49f75dc..e1e375a 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -2,13 +2,13 @@
use std::collections::HashMap;
use std::future::Future;
-use std::sync::atomic::AtomicBool;
use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
use std::time::Duration;
-use anyhow::{bail, Result};
-use automerge::{sync, AutoCommit, ChangeHash};
-use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};
+use anyhow::{Result, bail};
+use automerge::{AutoCommit, ChangeHash, sync};
+use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
use tokio::time::Instant;
use tracing::{debug, error, warn};
@@ -110,6 +110,9 @@ pub(crate) struct PeerState {
pub(crate) sync_reply: Option<SyncReplyWait>,
/// How long to wait for it; doubles while answers stay out.
pub(crate) sync_reply_timeout: Duration,
+
+ /// When sync with this peer last started from scratch.
+ pub(crate) sync_reset_at: Instant,
}
/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
@@ -128,6 +131,10 @@ const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
/// Upper bound (Mycelium gives up on a message after five minutes).
const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
+/// How long the side that doesn't open a fresh sync waits for the other's opening
+/// before opening itself (the other one may not have reset, or its message got lost).
+const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
+
impl Default for PeerState {
fn default() -> Self {
Self {
@@ -148,6 +155,7 @@ impl Default for PeerState {
authenticated: false,
sync_reply: None,
sync_reply_timeout: SYNC_REPLY_TIMEOUT,
+ sync_reset_at: Instant::now(),
epoch: None,
}
}
@@ -162,6 +170,22 @@ impl PeerState {
self.needs_generate = false;
self.sync_reply = None;
self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+ self.sync_reset_at = Instant::now();
+ }
+
+ /// After a reset, usually both sides do it and both open the sync at once. The
+ /// openings cross, and each side answers the other's: harmless between peers that
+ /// share history, but an empty peer answers with empty heads, which Automerge takes
+ /// for "lost everything", so the whole document goes out a second time. Only the
+ /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening, up to
+ /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all.
+ pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> {
+ if we_open || self.sync_state.their_heads.is_some() {
+ return None;
+ }
+ SYNC_OPENING_GRACE
+ .checked_sub(now.saturating_duration_since(self.sync_reset_at))
+ .filter(|wait| !wait.is_zero())
}
/// A sync message went out; `sent` are the changes it carried for the first time.
@@ -178,6 +202,12 @@ impl PeerState {
}
}
+ /// Keepalives show both documents are the same: stop waiting for an answer.
+ pub(crate) fn sync_in_step(&mut self) {
+ self.sync_reply = None;
+ self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+ }
+
/// The peer answered.
pub(crate) fn sync_reply_received(&mut self) {
self.sync_reply = None;
@@ -244,11 +274,29 @@ impl PeerState {
}
/// True (and records the ping) if an established, quiet peer needs a keepalive.
- pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
+ ///
+ /// One PING and its PONG tell both sides that the other is alive, so only one side of
+ /// a pair needs to ping: the one leading the pair (`we_ping`, `swarm::pair`). Each of its PINGs
+ /// makes the other side's peer look recently heard from, so the other side never
+ /// pings while it does. If the pings stop, the other side pings once it has heard
+ /// nothing for half an interval more. Without that, both sides would get due at about
+ /// the same time (each one's quiet time starts with the same exchange) and their
+ /// PINGs would often cross.
+ pub(crate) fn take_keepalive_due(
+ &mut self,
+ timing: &Timing,
+ we_ping: bool,
+ now: Instant,
+ ) -> bool {
+ let quiet_enough = if we_ping {
+ timing.keepalive
+ } else {
+ timing.keepalive + timing.keepalive / 2
+ };
let due = self.liveness == Liveness::Active
&& self.hello_state == HelloState::Established
&& now.saturating_duration_since(self.added_at) >= timing.keepalive
- && self.quiet_for(now) >= timing.keepalive
+ && self.quiet_for(now) >= quiet_enough
&& self
.last_ping_at
.is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
@@ -759,7 +807,6 @@ impl Host for AppState {
summaries.push(PeerSummary {
id,
active: peer.liveness == Liveness::Active,
- established: peer.hello_state == HelloState::Established,
epoch: peer.epoch,
});
}
@@ -828,15 +875,6 @@ impl Host for AppState {
async move { state.render(f, |value| value).await }
}
- fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
- where
- F: FnOnce(&mut AutoCommit) -> T + Send + 'static,
- T: Send + 'static,
- {
- let state = self.clone();
- async move { state.write_doc(f).await }
- }
-
fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
where
F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs
index 8b1352c..ea95406 100644
--- a/src/swarm/auth.rs
+++ b/src/swarm/auth.rs
@@ -1,7 +1,7 @@
//! Swarm membership by pre-shared key.
//!
//! Mycelium authenticates the sender of every message (`src_pk` can't be forged), so we
-//! always know *who* talks to us. With a swarm key configured, only peers that also know
+//! always know *who* talks to us. With a swarm key configured, only nodes that also know
//! the key may take part: HELLO and HELLO_REPLY carry a tag proving it, and until a peer
//! has sent a valid one, everything else it sends is ignored.
//!
@@ -19,7 +19,7 @@ use std::path::Path;
use std::sync::Mutex;
use std::time::{Duration, Instant};
-use anyhow::{bail, Context, Result};
+use anyhow::{Context, Result, bail};
use sha2::{Digest, Sha256};
const HELLO_LABEL: &[u8] = b"continuum/hello/1";
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index b279cb1..3b6e510 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -3,7 +3,7 @@
use std::sync::atomic::Ordering;
use std::time::Duration;
-use anyhow::{bail, Result};
+use anyhow::{Result, bail};
use tracing::{debug, error, info, warn};
use uuid::Uuid;
@@ -15,7 +15,7 @@ const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500);
pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> {
if seeds.is_empty() {
- info!("Initializing genesis node (no seed peers configured)");
+ info!("Initializing genesis node (no seeds configured)");
state.initial_sync_done.store(true, Ordering::Release);
state.mark_ready(Uuid::new_v4().to_string()).await?;
return Ok(());
@@ -29,9 +29,9 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
let mut peers = Vec::new();
for seed in seeds {
- match state.transport.resolve_peer(seed).await {
+ match state.transport.resolve_seed(seed).await {
Ok(peer) => {
- debug!(seed = %seed, peer = %peer, "Resolved seed to peer");
+ debug!(seed = %seed, peer = %peer, "Resolved seed");
peers.push(peer)
}
Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"),
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index a4277c7..7b53afc 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -3,5 +3,6 @@
pub(crate) mod auth;
pub(crate) mod bootstrap;
+pub(crate) mod pair;
pub(crate) mod peer;
pub(crate) mod sync;
diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs
new file mode 100644
index 0000000..7200a91
--- /dev/null
+++ b/src/swarm/pair.rs
@@ -0,0 +1,78 @@
+//! Which side of a pair of nodes leads, for duties only one side needs to do (opening
+//! the sync after a reset, sending keepalive PINGs).
+//!
+//! Both sides decide on their own, from the two keys alone, and agree. The leader is
+//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
+//! global order would make the node with the smallest key lead every one of its pairs
+//! and the one with the largest lead none, so every such duty would pile up on the same
+//! few nodes. With the hash, each node leads about half of its pairs.
+
+use sha2::{Digest, Sha256};
+
+const LABEL: &[u8] = b"continuum/pair-leader/1";
+
+/// Whether `local` leads its pair with `peer`. Exactly one of `leads(a, b)` and
+/// `leads(b, a)` is true for any two different keys.
+pub(crate) fn leads(local: &str, peer: &str) -> bool {
+ let (low, high) = if local <= peer {
+ (local, peer)
+ } else {
+ (peer, local)
+ };
+ let mut hash = Sha256::new();
+ // Length-prefixed fields, so no two different pairs hash the same input.
+ for field in [LABEL, low.as_bytes(), high.as_bytes()] {
+ hash.update((field.len() as u32).to_be_bytes());
+ hash.update(field);
+ }
+ let leader = if hash.finalize()[0] & 1 == 0 {
+ low
+ } else {
+ high
+ };
+ leader == local
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn key(n: u32) -> String {
+ format!("{:064x}", u64::from(n).wrapping_mul(0x9e37_79b9_7f4a_7c15))
+ }
+
+ #[test]
+ fn exactly_one_side_leads() {
+ for a in 0..50 {
+ for b in 0..50 {
+ if a != b {
+ assert_ne!(leads(&key(a), &key(b)), leads(&key(b), &key(a)));
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn leading_is_spread_across_nodes() {
+ let nodes: Vec<String> = (0..64).map(key).collect();
+ let mut smallest_key_leads = 0;
+ for (i, node) in nodes.iter().enumerate() {
+ let led = nodes
+ .iter()
+ .filter(|peer| *peer != node && leads(node, peer))
+ .count();
+ // 63 pairs per node: about half each; far from all or none.
+ assert!((10..=53).contains(&led), "node {i} leads {led} of 63 pairs");
+ for peer in &nodes {
+ if node < peer && leads(node, peer) {
+ smallest_key_leads += 1;
+ }
+ }
+ }
+ // Not simply the smaller key of each pair (64 * 63 / 2 = 2016 pairs).
+ assert!(
+ (800..=1216).contains(&smallest_key_leads),
+ "{smallest_key_leads}"
+ );
+ }
+}
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index d801b0f..cec8842 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -4,7 +4,7 @@
use std::sync::Arc;
use std::time::Duration;
-use anyhow::{anyhow, bail, Context, Result};
+use anyhow::{Context, Result, anyhow, bail};
use automerge::AutoCommit;
use sha2::{Digest, Sha256};
use tokio::sync::Mutex;
@@ -14,7 +14,7 @@ use tracing::{debug, info, instrument, trace, warn};
use crate::document::replica::{Epoch, EpochRelation};
use crate::document::store::{self, StoreError};
use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState};
-use crate::swarm::sync::{handle_sync, SyncEvent};
+use crate::swarm::sync::{SyncEvent, handle_sync};
use crate::transport::{self, ParsedMessage};
const HELLO_RETRY_INITIAL: Duration = Duration::from_secs(1);
@@ -133,12 +133,21 @@ async fn heal_if_diverged(
return;
};
- let heal = peer_state_mutex.lock().await.needs_heal(
- &local.heads_digest,
- &remote.heads_digest,
- &state.timing,
- Instant::now(),
- );
+ let heal = {
+ let mut peer_state = peer_state_mutex.lock().await;
+ if local.heads_digest == remote.heads_digest {
+ // Same document: whatever answer we were waiting for, there is nothing left
+ // to send. (Automerge answers only when it has something to say, so a sync
+ // message to a peer that already has everything gets no reply at all.)
+ peer_state.sync_in_step();
+ }
+ peer_state.needs_heal(
+ &local.heads_digest,
+ &remote.heads_digest,
+ &state.timing,
+ Instant::now(),
+ )
+ };
if heal {
warn!(peer = %peer, "Documents differ and sync is idle; re-handshaking to restart sync");
@@ -401,7 +410,8 @@ pub(crate) async fn run_peer_worker(state: AppState) {
let (became_dormant, ping_due, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
let became_dormant = ready && peer_state.check_dormancy(&state.timing, now);
- let ping_due = peer_state.take_keepalive_due(&state.timing, now);
+ let we_ping = crate::swarm::pair::leads(state.local_id(), peer);
+ let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now);
let unanswered = peer_state.take_overdue_sync_reply(now);
(became_dormant, ping_due, unanswered)
};
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 0c7148e..2c2079a 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -2,11 +2,11 @@
//! messages, and the change publisher that feeds incremental streams.
use std::collections::HashSet;
-use std::sync::atomic::Ordering;
use std::sync::Arc;
+use std::sync::atomic::Ordering;
use std::time::Duration;
-use anyhow::{anyhow, bail, Context, Result};
+use anyhow::{Context, Result, anyhow, bail};
use automerge::sync::SyncDoc;
use tokio::sync::mpsc;
use tokio::time::Instant;
@@ -58,10 +58,28 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(());
}
- match &peer_state.membership {
- PeerMembership::Unknown => return Ok(()),
- PeerMembership::Member { swarm_id } => swarm_id.clone(),
+ let PeerMembership::Member { swarm_id } = &peer_state.membership else {
+ return Ok(());
+ };
+
+ if !retry_delivery {
+ let we_open = crate::swarm::pair::leads(state.local_id(), peer);
+ if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) {
+ trace!(
+ wait_ms = wait.as_millis(),
+ "Leaving the opening sync message to the peer"
+ );
+ let tx = state.sync_events.clone();
+ let peer = peer.to_owned();
+ tokio::spawn(async move {
+ tokio::time::sleep(wait).await;
+ let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
+ });
+ return Ok(());
+ }
}
+
+ swarm_id.clone()
};
let swarm_id = state
@@ -110,9 +128,19 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
peer_state.needs_generate = false;
let epoch = replica.epoch;
let sent_before = peer_state.sync_state.sent_hashes.clone();
- let sync_doc = replica.doc.sync();
- let message = sync_doc.generate_sync_message(&mut peer_state.sync_state);
- if message.is_some() {
+ let message = {
+ let sync_doc = replica.doc.sync();
+ sync_doc.generate_sync_message(&mut peer_state.sync_state)
+ };
+ // A peer whose heads are ours already has everything; Automerge
+ // answers only when it has something to say, so it will stay
+ // silent. Waiting for an answer would only end in a pointless
+ // resend.
+ let in_step = peer_state.sync_state.their_heads.as_ref()
+ == Some(&replica.doc.get_heads());
+ if in_step {
+ peer_state.sync_in_step();
+ } else if message.is_some() {
let sent = peer_state
.sync_state
.sent_hashes
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index 917fe5f..84fdc59 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -69,8 +69,8 @@ impl Transport for FirewallTransport {
self.inner.local_id()
}
- async fn resolve_peer(&self, hint: &str) -> Result<String> {
- self.inner.resolve_peer(hint).await
+ async fn resolve_seed(&self, seed: &str) -> Result<String> {
+ self.inner.resolve_seed(seed).await
}
fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 58db611..69d9e70 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -82,8 +82,9 @@ pub trait Transport: Send + Sync {
/// This node's identity on the transport (for Mycelium: its public key).
fn local_id(&self) -> &str;
- /// Resolves a user-supplied peer hint (e.g. a seed) to a peer identity.
- async fn resolve_peer(&self, hint: &str) -> Result<String>;
+ /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's
+ /// identity, as `local_id` gives it on that node.
+ async fn resolve_seed(&self, seed: &str) -> Result<String>;
fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage>;
async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>;
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 057916f..84a34ea 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -1,6 +1,6 @@
-use anyhow::{anyhow, Context, Result};
+use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
-use base64::engine::{general_purpose::STANDARD as BASE64, Engine};
+use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
use reqwest::StatusCode;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_bytes::{ByteBuf, Bytes};
@@ -179,27 +179,11 @@ impl MyceliumSetup {
}
}
+/// The part of Mycelium's `/api/v1/admin` answer we use.
#[derive(Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct MyceliumAdminInfo {
- pub node_ip: String,
pub node_pubkey: String,
- pub node_start_time: u64,
- pub node_subnet: String,
-}
-
-pub async fn fetch_admin_info(rest_url: &Url) -> Result<MyceliumAdminInfo> {
- let url = rest_url.join("/api/v1/admin")?;
- let info = reqwest::Client::builder()
- .timeout(Duration::from_secs(10))
- .build()?
- .get(url)
- .send()
- .await?
- .error_for_status()?
- .json()
- .await?;
- Ok(info)
}
#[derive(Debug, Clone)]
@@ -216,10 +200,6 @@ pub struct SupervisedMycelium {
pub config: MyceliumNodeConfig,
_key_file_handle: Option<File>,
_net_key_file_handle: Option<File>,
- key_file_path: PathBuf,
- net_key_file_path: Option<PathBuf>,
- is_ephemeral_key: bool,
- is_ephemeral_net_key: bool,
}
impl SupervisedMycelium {
@@ -232,6 +212,15 @@ impl SupervisedMycelium {
explicit_discovery_port: Option<u16>,
peers: &[String],
) -> Result<Self> {
+ // Linux only: keys are handed over in memory (memfd, never on disk), and the
+ // child dies with us (PR_SET_PDEATHSIG). Elsewhere, run myceliumd yourself.
+ if cfg!(not(target_os = "linux")) {
+ bail!(
+ "Supervising Mycelium is only supported on Linux; run myceliumd yourself \
+ and pass --mycelium-rest-url"
+ );
+ }
+
let tcp_port = match explicit_tcp {
Some(p) => p,
None => Self::find_available_port().context("Failed to allocate TCP wire port")?,
@@ -261,18 +250,17 @@ impl SupervisedMycelium {
let parent_pid = std::process::id();
- let (key_file_handle, key_file_path, node_fd, is_ephemeral_key) =
+ let (key_file_handle, key_file_path, node_fd) =
Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?;
- let (net_key_file_handle, net_key_file_path, net_fd, is_ephemeral_net_key) =
- match private_network {
- Some(net_cfg) => {
- let (h, p, fd, eph) =
- Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
- (h, Some(p), fd, eph)
- }
- None => (None, None, None, false),
- };
+ let (net_key_file_handle, net_key_file_path, net_fd) = match private_network {
+ Some(net_cfg) => {
+ let (h, p, fd) =
+ Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
+ (h, Some(p), fd)
+ }
+ None => (None, None, None),
+ };
let api_addr = format!("127.0.0.1:{}", config.api_port);
let rpc_addr = format!("127.0.0.1:{}", config.rpc_port);
@@ -370,14 +358,8 @@ impl SupervisedMycelium {
config,
_key_file_handle: key_file_handle,
_net_key_file_handle: net_key_file_handle,
- key_file_path,
- net_key_file_path,
- is_ephemeral_key,
- is_ephemeral_net_key,
};
- //let admin_info = node.poll_admin_endpoint().await?;
- //Ok((node, admin_info))
Ok(node)
}
@@ -385,14 +367,14 @@ impl SupervisedMycelium {
desc_tag: &str,
source: KeySource,
parent_pid: u32,
- ) -> Result<(Option<File>, PathBuf, Option<i32>, bool)> {
+ ) -> Result<(Option<File>, PathBuf, Option<i32>)> {
match source {
KeySource::Raw(raw_bytes) => {
let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?;
let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd));
- Ok((Some(handle), pseudo_path, Some(fd), true))
+ Ok((Some(handle), pseudo_path, Some(fd)))
}
- KeySource::Path(existing_path) => Ok((None, existing_path, None, false)),
+ KeySource::Path(existing_path) => Ok((None, existing_path, None)),
}
}
@@ -419,41 +401,14 @@ impl SupervisedMycelium {
Ok((file, fd))
}
+ // Unreachable: `spawn` refuses to run elsewhere.
#[cfg(not(target_os = "linux"))]
{
- let mut path = std::env::tempdir();
- path.push(format!("{}_{}.key", desc_tag, uuid::Uuid::new_v4()));
- let mut file = File::create(&path)?;
- file.write_all(&raw_key)?;
- file.flush()?;
- Ok((file, -1))
+ let _ = (desc_tag, raw_key);
+ bail!("In-memory key files need Linux")
}
}
- //async fn poll_admin_endpoint(&mut self) -> Result<MyceliumAdminInfo> {
- // let client = reqwest::Client::new();
- // let url = format!("http://127.0.0.1:{}/api/v1/admin", self.config.api_port);
- //
- // for _ in 0..30 {
- // if let Some(status) = self.child.try_wait()? {
- // return Err(anyhow!(
- // "myceliumd process closed prematurely with status: {}",
- // status
- // ));
- // }
- // if let Ok(resp) = client.get(&url).send().await {
- // if resp.status().is_success() {
- // let info: MyceliumAdminInfo = resp.json().await?;
- // return Ok(info);
- // }
- // }
- // tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
- // }
- // Err(anyhow!(
- // "Timeout waiting for local mycelium administration socket to respond"
- // ))
- //}
-
pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
self.child.wait().await
}
@@ -461,19 +416,8 @@ impl SupervisedMycelium {
impl Drop for SupervisedMycelium {
fn drop(&mut self) {
+ // The memfd key files go away with their handles.
let _ = self.child.start_kill();
-
- #[cfg(not(target_os = "linux"))]
- {
- if self.is_ephemeral_key {
- let _ = std::fs::remove_file(&self.key_file_path);
- }
- if self.is_ephemeral_net_key {
- if let Some(ref path) = self.net_key_file_path {
- let _ = std::fs::remove_file(path);
- }
- }
- }
}
}
@@ -546,7 +490,7 @@ impl Transport for MyceliumTransport {
&self.local_id
}
- async fn resolve_peer(&self, hint: &str) -> Result<String> {
+ async fn resolve_seed(&self, hint: &str) -> Result<String> {
if is_pubkey(hint) {
return Ok(hint.to_owned());
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 00:13:29 +0000