summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-21 14:51:51 +0200
committertv <tv@krebsco.de>2026-10-21 14:51:51 +0200
commit98ca55ee593c1d073241a295c9aba513e8e96e99 (patch)
tree0e9d361b832cb3e0c0b69c760c40ae1187d23c9a /src
parent540a344b811885551ad5ed374e0abe3411554b9d (diff)
rename cluster to swarm everywhere!
Diffstat (limited to 'src')
-rw-r--r--src/api/peers.rs10
-rw-r--r--src/document/replica.rs4
-rw-r--r--src/main.rs10
-rw-r--r--src/remake/api.rs4
-rw-r--r--src/remake/catchup.rs28
-rw-r--r--src/remake/coordinator.rs2
-rw-r--r--src/remake/mod.rs4
-rw-r--r--src/remake/snapshot.rs16
-rw-r--r--src/state.rs106
-rw-r--r--src/swarm/bootstrap.rs28
-rw-r--r--src/swarm/peer.rs63
-rw-r--r--src/swarm/sync.rs42
-rw-r--r--src/transport/mod.rs4
-rw-r--r--src/transport/mycelium.rs18
14 files changed, 162 insertions, 177 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs
index 5308eb9..d60e7b3 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -47,7 +47,7 @@ impl From<Epoch> for EpochInfo {
#[serde(rename_all = "camelCase")]
pub(crate) struct PeerStatus {
public_key: String,
- validated_cluster: Option<String>,
+ validated_swarm: Option<String>,
status: PeerLivenessStatus,
handshake_complete: bool,
last_seen_secs_ago: Option<u64>,
@@ -112,9 +112,9 @@ pub(crate) async fn list_peers(
peers.push(PeerStatus {
public_key,
- validated_cluster: match &ps.membership {
+ validated_swarm: match &ps.membership {
PeerMembership::Unknown => None,
- PeerMembership::Member { cluster_id } => Some(cluster_id.clone()),
+ PeerMembership::Member { swarm_id } => Some(swarm_id.clone()),
},
status,
handshake_complete: ps.hello_state == HelloState::Established,
@@ -144,7 +144,7 @@ pub(crate) async fn list_peers(
("force" = Option<bool>, Query, description = "Also forget a peer that is currently alive; it returns on its next message")
),
responses(
- (status = 204, description = "Cluster-wide forget recorded"),
+ (status = 204, description = "Swarm-wide forget recorded"),
(status = 400, description = "Not a public key, or the local node's own key"),
(status = 409, description = "Peer is alive; pass force=true"),
(status = 503, description = "Node is not ready")
@@ -183,7 +183,7 @@ pub(crate) async fn delete_peer(
.map_err(|e| store_status(e, "Failed to write forget record"))?;
if state.peer_states.write().await.remove(&peer).is_some() {
- warn!(peer = %peer, "Peer forgotten cluster-wide by operator");
+ warn!(peer = %peer, "Peer forgotten swarm-wide by operator");
}
Ok(StatusCode::NO_CONTENT)
diff --git a/src/document/replica.rs b/src/document/replica.rs
index ff3af4c..af807ea 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -11,7 +11,7 @@ use std::ops::{Deref, DerefMut};
use automerge::AutoCommit;
-/// One generation of the cluster document.
+/// One generation of the swarm document.
///
/// * `generation` orders epochs: every remake increments it, so a node can tell
/// whether a peer is behind or ahead.
@@ -36,7 +36,7 @@ pub enum EpochRelation {
}
impl Epoch {
- /// Clusters that have never been compacted.
+ /// Swarms that have never been compacted.
pub const GENESIS: Epoch = Epoch {
generation: 0,
id: [0; 32],
diff --git a/src/main.rs b/src/main.rs
index d9d24f3..ef9bf54 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -149,7 +149,7 @@ async fn main() -> Result<()> {
doc: Arc::new(RwLock::new(Replica::new(initial_epoch, AutoCommit::new()))),
epoch_tx,
lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining {
- discovered_cluster_id: None,
+ discovered_swarm_id: None,
})),
peer_states: Arc::new(RwLock::new(HashMap::new())),
transport: Arc::new(FirewallTransport::new(
@@ -194,15 +194,15 @@ async fn main() -> Result<()> {
trace!("Peer worker task spawned");
// Switches to committed epochs and catches up with newer ones. Starts before
- // bootstrapping: a node joining a compacted cluster can only finish its initial sync
+ // bootstrapping: a node joining a compacted swarm can only finish its initial sync
// after fetching the current epoch from a peer.
tokio::spawn(remake::run_worker(state.clone()));
trace!("Remake worker task spawned");
tokio::select! {
- joined = bootstrap::bootstrap(&state, &seeds) => joined.context("Joining cluster")?,
+ joined = bootstrap::bootstrap(&state, &seeds) => joined.context("Joining swarm")?,
_ = stopped(stop.clone()) => {
- info!("Stopped while joining the cluster");
+ info!("Stopped while joining the swarm");
return Ok(());
}
}
@@ -213,7 +213,7 @@ async fn main() -> Result<()> {
tokio::spawn(async move { sync::run_change_publisher(publisher_state).await });
info!(
- cluster_id = ?state.cluster_id().await,
+ swarm_id = ?state.swarm_id().await,
ready = state.is_ready().await,
"Bootstrap completed"
);
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 7f62b48..fbd5f54 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -4,7 +4,7 @@
//! /admin/compaction` is a shortcut, kept from before). The receiving
//! node becomes its coordinator. `202` with the report, or with `?wait=true` the final
//! report once decided. A dropped request does not abort anything. `?force=true`
-//! accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters).
+//! accepts a `settleSecs` below the dormancy threshold (tests, single-site swarms).
//! * `GET /admin/remakes` lists all remakes.
//! * `GET /admin/remakes/{id}` reports one; `?wait=true` long-polls until decided
//! (bounded by the proposal's own expiry).
@@ -22,7 +22,7 @@
//! "settleSecs": 480, "switchDelaySecs": 60 }`
//!
//! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs`
-//! below one keepalive interval: for tests and debugging, not for real clusters.
+//! below one keepalive interval: for tests and debugging, not for real swarms.
//!
//! `kind` is `compact` (no spec) or `replace`, whose spec is
//! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index c8878cc..4aa66e7 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -11,7 +11,7 @@
//! exactly like a new node joining;
//! 3. once caught up with a peer, checks that the history starts with the epoch's
//! snapshot change;
-//! 4. decides about its own changes the cluster never saw (none, for a new node):
+//! 4. decides about its own changes the swarm never saw (none, for a new node):
//! * the metadata of the epoch that followed the node's own records which changes
//! went into it (per-actor clock); the rest are this node's missing changes;
//! * within that epoch's replay window they are replayed (state per path, as in a
@@ -374,7 +374,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E
(None, _) => Plan::Quarantine {
paths,
reason: format!(
- "the metadata of the epoch after {previous} is missing, so it is unknown which of {} local changes the cluster has",
+ "the metadata of the epoch after {previous} is missing, so it is unknown which of {} local changes the swarm has",
missing.own
),
},
@@ -386,7 +386,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E
Plan::Quarantine {
paths,
reason: format!(
- "{} local changes are outside the replay window ({gap} epochs behind, cluster moved on {}s ago)",
+ "{} local changes are outside the replay window ({gap} epochs behind, swarm moved on {}s ago)",
missing.own,
away_ms / 1000
),
@@ -448,7 +448,7 @@ mod tests {
use automerge::{transaction::Transactable, ActorId, ROOT};
use serde_json::{json, Value as JsonValue};
- /// An old-epoch document with one own change after the base, and the cluster's
+ /// 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).
fn scenario(policy: &JsonValue) -> (AutoCommit, AutoCommit, Epoch) {
let mut old = AutoCommit::new().with_actor(ActorId::from(&b"late-node"[..]));
@@ -472,19 +472,19 @@ mod tests {
.unwrap();
let target = snapshot.epoch;
- let mut cluster = snapshot.doc.with_actor(ActorId::from(&b"cluster-node"[..]));
- cluster.put(ROOT, "theirs", 1_i64).unwrap();
- cluster.commit();
+ let mut swarm = snapshot.doc.with_actor(ActorId::from(&b"swarm-node"[..]));
+ swarm.put(ROOT, "theirs", 1_i64).unwrap();
+ swarm.commit();
old.put(ROOT, "mine", 1_i64).unwrap();
old.commit();
- (old, cluster, target)
+ (old, swarm, target)
}
#[test]
fn replays_within_the_window() {
- let (old, cluster, target) = scenario(&json!({}));
- match prepare(cluster, old, Epoch::GENESIS, target) {
+ let (old, swarm, target) = scenario(&json!({}));
+ match prepare(swarm, old, Epoch::GENESIS, target) {
Prepared::Ready {
plan: Plan::Replay(paths, _),
own_missing,
@@ -499,8 +499,8 @@ mod tests {
#[test]
fn quarantines_outside_the_window() {
- let (old, cluster, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
- match prepare(cluster, old, Epoch::GENESIS, target) {
+ let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+ match prepare(swarm, old, Epoch::GENESIS, target) {
Prepared::Ready {
plan: Plan::Quarantine { paths, .. },
..
@@ -511,9 +511,9 @@ mod tests {
#[test]
fn new_node_has_nothing_to_replay() {
- let (_, cluster, target) = scenario(&json!({}));
+ let (_, swarm, target) = scenario(&json!({}));
assert!(matches!(
- prepare(cluster, AutoCommit::new(), Epoch::GENESIS, target),
+ prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target),
Prepared::Ready {
plan: Plan::Nothing,
had_history: false,
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index b84b0a9..d1b3397 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -1,4 +1,4 @@
-//! The coordinator: the node that received the proposal watches the cluster during the
+//! The coordinator: the node that received the proposal watches the swarm during the
//! settle period and then writes the decision.
//!
//! It aborts if, at any point before deciding:
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 3cfaf0b..ff3c9c3 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -1,4 +1,4 @@
-//! Remakes: coordinated moves of the whole cluster to a new document, a new epoch.
+//! Remakes: coordinated moves of the whole swarm to a new document, a new epoch.
//! Compaction is one: the new document has the old one's content without its history.
//!
//! * `kind`: what differs between kinds of remake (building the new document, and
@@ -66,7 +66,7 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
/// Known (not forgotten) peers.
fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
- /// Stops serving requests (back to `Joining`, keeping the cluster ID), as while a
+ /// Stops serving requests (back to `Joining`, keeping the swarm ID), as while a
/// rejoin syncs a fresh document. Returns whether the node was serving.
fn suspend(&self) -> impl Future<Output = bool> + Send;
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index 507d8ac..bd99854 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -139,7 +139,7 @@ pub(crate) struct EpochMetadata {
/// Highest sequence number per actor of `previous` contained in the snapshot.
pub(crate) clock: BTreeMap<String, u64>,
pub(crate) policy: Policy,
- /// When the remake was proposed (unix ms): the point the cluster moved on.
+ /// When the remake was proposed (unix ms): the point the swarm moved on.
pub(crate) proposed_at: u64,
/// The remake kind and its translation input, for writes made before the epoch.
pub(crate) kind: String,
@@ -206,7 +206,7 @@ fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, St
/// the same generation. Returns the metadata.
///
/// This does not prove the snapshot's content matches the old epoch; the hash commits
-/// to it, and the peer is a cluster member trusted as much as for ordinary sync.
+/// to it, and the peer is a swarm member trusted as much as for ordinary sync.
pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<EpochMetadata, String> {
let root = ChangeHash(epoch.id);
let Some(change) = doc.get_change_by_hash(&root) else {
@@ -653,14 +653,14 @@ mod tests {
let policy = json!({});
let mut snap = build(&mut late, &params(&base, &policy)).unwrap();
- // Meanwhile the cluster moved on in the new epoch.
- let mut cluster = snap
+ // Meanwhile the swarm moved on in the new epoch.
+ let mut swarm = snap
.doc
.fork()
- .with_actor(ActorId::from(&b"cluster-node"[..]));
- cluster.put(ROOT, "new", 1_i64).unwrap();
- cluster.commit();
- let bytes = cluster.save();
+ .with_actor(ActorId::from(&b"swarm-node"[..]));
+ swarm.put(ROOT, "new", 1_i64).unwrap();
+ swarm.commit();
+ let bytes = swarm.save();
late.put(ROOT, "mine", "late").unwrap();
late.commit();
diff --git a/src/state.rs b/src/state.rs
index 81a3e6a..736f55b 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -20,13 +20,9 @@ use crate::transport::firewall::Firewall;
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum NodeLifecycle {
- // TODO maybe rename discovered_cluster_id to cluster_id if that doesn't impair readability.
- Joining {
- discovered_cluster_id: Option<String>,
- },
- Ready {
- cluster_id: String,
- },
+ // TODO maybe rename discovered_swarm_id to swarm_id if that doesn't impair readability.
+ Joining { discovered_swarm_id: Option<String> },
+ Ready { swarm_id: String },
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
@@ -34,7 +30,7 @@ pub(crate) enum PeerMembership {
#[default]
Unknown,
Member {
- cluster_id: String,
+ swarm_id: String,
},
}
@@ -75,7 +71,7 @@ pub(crate) struct PeerState {
pub(crate) sync_state: sync::State,
pub(crate) hello_state: HelloState,
- /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no cluster ID yet,
+ /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no swarm ID yet,
/// or sending failed).
pub(crate) reply_owed: bool,
@@ -302,7 +298,7 @@ impl PeerState {
}
/// Heard from recently. A returning node counts as alive from its first message,
- /// before the handshake finishes, so a cluster-wide forget can't make it flap.
+ /// before the handshake finishes, so a swarm-wide forget can't make it flap.
pub(crate) fn is_alive(&self, timing: &Timing, now: Instant) -> bool {
self.liveness == Liveness::Active
&& self
@@ -324,9 +320,9 @@ impl Timing {
// Environment | Keepalive | Notes
// --------------------------------------------+-----------+------------------------------------------------------------
// Test harness | 1-2 s | Dormancy after about 4–8 s and healing within seconds keep tests short.
- // LAN or small cluster (up to ~50 nodes) | 15-30 s | Under 3.5 messages/s per node at 50 nodes and 30 s. Current setting.
+ // LAN or small swarm (up to ~50 nodes) | 15-30 s | Under 3.5 messages/s per node at 50 nodes and 30 s. Current setting.
// WAN, mobile or metered links, or 100+ nodes | 60-120 s | At 100 nodes and 30 s it’s about 6–7 messages/s per node. Fine, but rarely worth it on battery or metered links.
- // Several hundred nodes or more | - | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the cluster, which calls for partial views (SWIM-style random probing and syncing with a subset of peers).
+ // Several hundred nodes or more | - | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the swarm, which calls for partial views (SWIM-style random probing and syncing with a subset of peers).
pub(crate) fn from_keepalive(keepalive: Duration) -> Self {
Self {
keepalive,
@@ -348,8 +344,8 @@ pub(crate) struct AppState {
pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>,
- // Serializes cluster-ID establishment, peer membership, and readiness transitions.
- //[concurrent]cluster_validation: Arc<Mutex<()>>,
+ // Serializes swarm-ID establishment, peer membership, and readiness transitions.
+ //[concurrent]swarm_validation: Arc<Mutex<()>>,
pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>,
pub(crate) transport: Arc<dyn transport::Transport>,
pub(crate) firewall: Arc<Firewall>,
@@ -388,94 +384,86 @@ impl AppState {
matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. })
}
- pub(crate) async fn cluster_id(&self) -> Option<String> {
+ pub(crate) async fn swarm_id(&self) -> Option<String> {
match &*self.lifecycle.read().await {
- NodeLifecycle::Ready { cluster_id } => Some(cluster_id.clone()),
+ NodeLifecycle::Ready { swarm_id } => Some(swarm_id.clone()),
NodeLifecycle::Joining {
- discovered_cluster_id,
- } => discovered_cluster_id.clone(),
+ discovered_swarm_id,
+ } => discovered_swarm_id.clone(),
}
}
- // Record the expected cluster without making the node Ready. Bootstrap owns readiness.
- pub(crate) async fn set_expected_cluster_id(&self, cluster_id: String) -> Result<()> {
- //[concurrent]let _guard = self.cluster_validation.lock().await;
+ // Record the expected swarm without making the node Ready. Bootstrap owns readiness.
+ pub(crate) async fn set_expected_swarm_id(&self, swarm_id: String) -> Result<()> {
+ //[concurrent]let _guard = self.swarm_validation.lock().await;
let mut lifecycle = self.lifecycle.write().await;
match &*lifecycle {
NodeLifecycle::Joining {
- discovered_cluster_id: Some(existing),
- } if existing != &cluster_id => {
+ discovered_swarm_id: Some(existing),
+ } if existing != &swarm_id => {
bail!(
- "Conflicting expected cluster IDs: expected {}, received {}",
+ "Conflicting expected swarm IDs: expected {}, received {}",
existing,
- cluster_id
+ swarm_id
);
}
NodeLifecycle::Joining {
- discovered_cluster_id: None,
+ discovered_swarm_id: None,
} => {
let old_lifecycle = lifecycle.clone();
*lifecycle = NodeLifecycle::Joining {
- discovered_cluster_id: Some(cluster_id),
+ discovered_swarm_id: Some(swarm_id),
};
debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
Ok(())
}
NodeLifecycle::Joining {
- discovered_cluster_id: Some(_),
+ discovered_swarm_id: Some(_),
} => {
warn!("Node already Joining with the same identity");
Ok(())
}
- NodeLifecycle::Ready {
- cluster_id: existing,
- } if existing == &cluster_id => {
+ NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
warn!("Node already Ready with the same identity");
Ok(())
}
- NodeLifecycle::Ready {
- cluster_id: existing,
- } => {
+ NodeLifecycle::Ready { swarm_id: existing } => {
bail!(
- "Cannot change established cluster ID: existing {}, received {}",
+ "Cannot change established swarm ID: existing {}, received {}",
existing,
- cluster_id
+ swarm_id
);
}
}
}
// Only bootstrap should call this, after validating document metadata.
- pub(crate) async fn mark_ready(&self, cluster_id: String) -> Result<()> {
- //[concurrent]let _guard = self.cluster_validation.lock().await;
+ pub(crate) async fn mark_ready(&self, swarm_id: String) -> Result<()> {
+ //[concurrent]let _guard = self.swarm_validation.lock().await;
let mut lifecycle = self.lifecycle.write().await;
match &*lifecycle {
NodeLifecycle::Joining {
- discovered_cluster_id: Some(expected),
- } if expected != &cluster_id => {
+ discovered_swarm_id: Some(expected),
+ } if expected != &swarm_id => {
bail!(
- "Cannot mark node ready: expected cluster {}, received {}",
+ "Cannot mark node ready: expected swarm {}, received {}",
expected,
- cluster_id
+ swarm_id
);
}
NodeLifecycle::Joining { .. } => {
let old_lifecycle = lifecycle.clone();
- *lifecycle = NodeLifecycle::Ready { cluster_id };
+ *lifecycle = NodeLifecycle::Ready { swarm_id };
debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
}
- NodeLifecycle::Ready {
- cluster_id: existing,
- } if existing == &cluster_id => {
+ NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
debug!("Node already ready with the same identity");
}
- NodeLifecycle::Ready {
- cluster_id: existing,
- } => {
+ NodeLifecycle::Ready { swarm_id: existing } => {
bail!(
- "Cannot change established cluster ID: existing {}, received {}",
+ "Cannot change established swarm ID: existing {}, received {}",
existing,
- cluster_id
+ swarm_id
);
}
}
@@ -774,13 +762,13 @@ impl Host for AppState {
let state = self.clone();
async move {
let mut lifecycle = state.lifecycle.write().await;
- let NodeLifecycle::Ready { cluster_id } = &*lifecycle else {
+ let NodeLifecycle::Ready { swarm_id } = &*lifecycle else {
return false;
};
- let cluster_id = cluster_id.clone();
- debug!(cluster_id = %cluster_id, "Node suspended (rejoining)");
+ let swarm_id = swarm_id.clone();
+ debug!(swarm_id = %swarm_id, "Node suspended (rejoining)");
*lifecycle = NodeLifecycle::Joining {
- discovered_cluster_id: Some(cluster_id),
+ discovered_swarm_id: Some(swarm_id),
};
true
}
@@ -791,12 +779,12 @@ impl Host for AppState {
async move {
let mut lifecycle = state.lifecycle.write().await;
if let NodeLifecycle::Joining {
- discovered_cluster_id: Some(cluster_id),
+ discovered_swarm_id: Some(swarm_id),
} = &*lifecycle
{
- let cluster_id = cluster_id.clone();
- debug!(cluster_id = %cluster_id, "Node resumed");
- *lifecycle = NodeLifecycle::Ready { cluster_id };
+ let swarm_id = swarm_id.clone();
+ debug!(swarm_id = %swarm_id, "Node resumed");
+ *lifecycle = NodeLifecycle::Ready { swarm_id };
}
}
}
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 47fadfa..b279cb1 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -1,4 +1,4 @@
-//! Joining a cluster (or creating one as the genesis node).
+//! Joining a swarm (or creating one as the genesis node).
use std::sync::atomic::Ordering;
use std::time::Duration;
@@ -23,7 +23,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
info!(
timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(),
- "Joining existing cluster"
+ "Joining existing swarm"
);
let mut peers = Vec::new();
@@ -51,12 +51,12 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
let deadline = tokio::time::Instant::now() + BOOTSTRAP_TIMEOUT;
loop {
- if let Some(cluster_id) = state.cluster_id().await {
- // A node joining a compacted cluster rejoins the newer epoch first; it is
+ if let Some(swarm_id) = state.swarm_id().await {
+ // A node joining a compacted swarm rejoins the newer epoch first; it is
// ready only once that rejoin verified the epoch.
if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
- state.mark_ready(cluster_id.clone()).await?;
- info!(cluster_id = %cluster_id, "Node is ready");
+ state.mark_ready(swarm_id.clone()).await?;
+ info!(swarm_id = %swarm_id, "Node is ready");
return Ok(());
}
}
@@ -68,22 +68,22 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
for (peer, peer_state_mutex) in &peer_states {
let peer_state = peer_state_mutex.lock().await;
- let validated_cluster = match &peer_state.membership {
+ let validated_swarm = match &peer_state.membership {
PeerMembership::Unknown => None,
- PeerMembership::Member { cluster_id } => Some(cluster_id.as_str()),
+ PeerMembership::Member { swarm_id } => Some(swarm_id.as_str()),
};
peer_diagnostics.push(format!(
- "{peer}: validated_cluster={:?}, epoch={:?}, pending_inbound_sync={}",
- validated_cluster,
+ "{peer}: validated_swarm={:?}, epoch={:?}, pending_inbound_sync={}",
+ validated_swarm,
peer_state.epoch,
peer_state.pending_inbound_sync.len()
))
}
- let local_cluster = state.cluster_id().await;
+ let local_swarm = state.swarm_id().await;
let local_epoch = state.local_epoch();
let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire);
error!(
- local_cluster = ?local_cluster,
+ local_swarm = ?local_swarm,
local_epoch = %local_epoch,
initial_sync_done,
peers = %peer_diagnostics.join("; "),
@@ -91,9 +91,9 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
"Bootstrap timed out"
);
bail!(
- "Bootstrap timed out after {:?}; local_cluster={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.",
+ "Bootstrap timed out after {:?}; local_swarm={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.",
BOOTSTRAP_TIMEOUT,
- local_cluster,
+ local_swarm,
local_epoch,
initial_sync_done,
peer_diagnostics.join("; ")
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 6184427..53a4731 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -1,5 +1,5 @@
//! Peers: handshake (HELLO / HELLO_REPLY), liveness and keepalives, self-healing,
-//! cluster-wide forget records, and dispatch of inbound messages.
+//! swarm-wide forget records, and dispatch of inbound messages.
use std::sync::Arc;
use std::time::Duration;
@@ -70,7 +70,7 @@ async fn apply_forget_records(state: &AppState) {
if alive {
cancel.push(pk);
} else if state.peer_states.write().await.remove(&pk).is_some() {
- info!(peer = %pk, "Forgetting peer per cluster-wide record");
+ info!(peer = %pk, "Forgetting peer per swarm-wide record");
}
}
@@ -84,7 +84,7 @@ async fn apply_forget_records(state: &AppState) {
.mutate_doc(move |doc| {
for pk in &cancel {
match store::delete_path(&mut *doc, &forgotten_path(pk), false) {
- Ok(()) => info!(peer = %pk, "Peer is alive; cancelling cluster-wide forget"),
+ Ok(()) => info!(peer = %pk, "Peer is alive; cancelling swarm-wide forget"),
Err(StoreError::NotFound) => {}
Err(error) => {
warn!(peer = %pk, error = %error, "Failed to cancel forget record")
@@ -267,10 +267,10 @@ async fn ensure_handshake(state: &AppState, peer: &str) {
}
}
-/// Sends a HELLO_REPLY if one is owed and our cluster ID is known.
+/// Sends a HELLO_REPLY if one is owed and our swarm ID is known.
/// Otherwise it stays owed and the peer worker tries again.
async fn send_owed_hello_reply(state: &AppState, peer: &str) {
- if state.cluster_id().await.is_none() {
+ if state.swarm_id().await.is_none() {
return;
}
@@ -297,10 +297,10 @@ async fn flush_owed_hello_replies(state: &AppState) {
}
async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
- let cluster_id = state
- .cluster_id()
+ let swarm_id = state
+ .swarm_id()
.await
- .ok_or_else(|| anyhow!("Cannot send HELLO_REPLY without an established cluster ID"))?;
+ .ok_or_else(|| anyhow!("Cannot send HELLO_REPLY without an established swarm ID"))?;
let candidates: Vec<(String, Arc<Mutex<PeerState>>)> = state
.peer_states
@@ -322,7 +322,7 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
}
let reply = transport::HelloReply {
- cluster_id,
+ swarm_id,
peers,
epoch: state.local_epoch(),
};
@@ -441,16 +441,16 @@ pub(crate) async fn run_peer_worker(state: AppState) {
async fn establish_peer(
state: &AppState,
peer: &str,
- cluster_id: String,
+ swarm_id: String,
) -> Result<Vec<transport::SyncEnvelope>> {
- let expected_cluster = state.cluster_id().await;
+ let expected_swarm = state.swarm_id().await;
- if let Some(expected) = &expected_cluster {
- if expected != &cluster_id {
+ if let Some(expected) = &expected_swarm {
+ if expected != &swarm_id {
bail!(
- "Incompatible cluster at {peer}: expected {}, received {}",
+ "Incompatible swarm at {peer}: expected {}, received {}",
expected,
- cluster_id
+ swarm_id
);
}
}
@@ -458,27 +458,24 @@ async fn establish_peer(
let peer_state_mutex = state.get_or_create_peer_state(peer).await;
let mut peer_state = peer_state_mutex.lock().await;
- if let PeerMembership::Member {
- cluster_id: existing,
- } = &peer_state.membership
- {
- if existing != &cluster_id {
+ if let PeerMembership::Member { swarm_id: existing } = &peer_state.membership {
+ if existing != &swarm_id {
bail!(
- "Peer {peer} changed cluster identity from {} to {}",
+ "Peer {peer} changed swarm identity from {} to {}",
existing,
- cluster_id
+ swarm_id
);
}
}
// Lock order: peer mutex -> lifecycle.
- if expected_cluster.is_none() {
- state.set_expected_cluster_id(cluster_id.clone()).await?;
+ if expected_swarm.is_none() {
+ state.set_expected_swarm_id(swarm_id.clone()).await?;
}
if matches!(peer_state.membership, PeerMembership::Unknown) {
peer_state.membership = PeerMembership::Member {
- cluster_id: cluster_id.clone(),
+ swarm_id: swarm_id.clone(),
};
}
@@ -492,17 +489,17 @@ async fn establish_peer(
let pending = std::mem::take(&mut peer_state.pending_inbound_sync);
- info!(peer = %peer, cluster_id = %cluster_id, pending_messages = pending.len(), "Peer membership established");
+ info!(peer = %peer, swarm_id = %swarm_id, pending_messages = pending.len(), "Peer membership established");
Ok(pending)
}
#[instrument(skip(state), fields(peer = %peer))]
-pub(crate) async fn peer_cluster(state: &AppState, peer: &str) -> Option<String> {
+pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> {
let peer_state_mutex = state.get_peer_state(peer).await?;
let peer_state = peer_state_mutex.lock().await;
match &peer_state.membership {
PeerMembership::Unknown => None,
- PeerMembership::Member { cluster_id } => Some(cluster_id.clone()),
+ PeerMembership::Member { swarm_id } => Some(swarm_id.clone()),
}
}
@@ -607,7 +604,7 @@ pub(crate) async fn handle_message(
peer_state.reply_owed = true;
}
- // Sends now if we know our cluster ID; otherwise stays owed.
+ // Sends now if we know our swarm ID; otherwise stays owed.
send_owed_hello_reply(state, &peer).await;
// Our own HELLO goes out regardless of whether the reply could be sent.
@@ -621,11 +618,11 @@ pub(crate) async fn handle_message(
.send(SyncEvent::PeerNeedsSync(peer.clone()));
}
transport::ParsedMessage::HelloReply(reply) => {
- let pending = establish_peer(state, &peer, reply.cluster_id.clone()).await?;
+ let pending = establish_peer(state, &peer, reply.swarm_id.clone()).await?;
discover_peers(state, &peer, &reply.peers).await;
- // Our cluster ID may have just become known: pay any replies we owe.
+ // Our swarm ID may have just become known: pay any replies we owe.
flush_owed_hello_replies(state).await;
if !pending.is_empty() {
@@ -657,7 +654,7 @@ pub(crate) async fn handle_message(
handle_sync(state, &peer, envelope).await?;
}
transport::ParsedMessage::Ping(remote) => {
- if reactivated || peer_cluster(state, &peer).await.is_none() {
+ if reactivated || peer_swarm(state, &peer).await.is_none() {
ensure_handshake(state, &peer).await;
}
let local = local_keepalive(state).await;
@@ -671,7 +668,7 @@ pub(crate) async fn handle_message(
}
}
transport::ParsedMessage::Pong(remote) => {
- if reactivated || peer_cluster(state, &peer).await.is_none() {
+ if reactivated || peer_swarm(state, &peer).await.is_none() {
ensure_handshake(state, &peer).await;
}
let local = local_keepalive(state).await;
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 04d5b15..0c7148e 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -13,7 +13,7 @@ use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
-use crate::swarm::peer::{peer_cluster, send_hello_once};
+use crate::swarm::peer::{peer_swarm, send_hello_once};
use crate::transport::{self, ParsedMessage};
const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500);
@@ -40,7 +40,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(());
};
- let validated_cluster = {
+ let validated_swarm = {
let peer_state = peer_state_mutex.lock().await;
if matches!(peer_state.liveness, Liveness::Dormant { .. }) {
@@ -60,17 +60,17 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
match &peer_state.membership {
PeerMembership::Unknown => return Ok(()),
- PeerMembership::Member { cluster_id } => cluster_id.clone(),
+ PeerMembership::Member { swarm_id } => swarm_id.clone(),
}
};
- let cluster_id = state
- .cluster_id()
+ let swarm_id = state
+ .swarm_id()
.await
- .ok_or_else(|| anyhow!("Cannot sync without an established cluster ID"))?;
+ .ok_or_else(|| anyhow!("Cannot sync without an established swarm ID"))?;
- if validated_cluster != cluster_id {
- bail!("Peer {peer} belongs to a different cluster");
+ if validated_swarm != swarm_id {
+ bail!("Peer {peer} belongs to a different swarm");
}
let existing = {
@@ -127,7 +127,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(None);
};
let envelope = transport::SyncEnvelope {
- cluster_id,
+ swarm_id,
epoch,
message,
};
@@ -221,7 +221,7 @@ pub(crate) async fn handle_sync(
PeerMembership::Unknown => {
let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| {
pending.message == envelope.message
- && pending.cluster_id == envelope.cluster_id
+ && pending.swarm_id == envelope.swarm_id
&& pending.epoch == envelope.epoch
});
if !duplicate {
@@ -236,10 +236,10 @@ pub(crate) async fn handle_sync(
}
true
}
- PeerMembership::Member { cluster_id } if cluster_id != &envelope.cluster_id => {
+ PeerMembership::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
bail!(
- "Sync cluster mismatch for peer {peer}: validated_cluster={cluster_id}, envelope_cluster={}",
- envelope.cluster_id
+ "Sync swarm mismatch for peer {peer}: validated_swarm={swarm_id}, envelope_swarm={}",
+ envelope.swarm_id
);
}
PeerMembership::Member { .. } => false,
@@ -255,15 +255,15 @@ pub(crate) async fn handle_sync(
return Ok(());
}
- let expected_cluster = state
- .cluster_id()
+ let expected_swarm = state
+ .swarm_id()
.await
- .ok_or_else(|| anyhow!("Missing expected cluster ID after peer membership"))?;
- if envelope.cluster_id != expected_cluster {
+ .ok_or_else(|| anyhow!("Missing expected swarm ID after peer membership"))?;
+ if envelope.swarm_id != expected_swarm {
bail!(
- "Incompatible cluster from {peer}: expected {}, received {}",
- expected_cluster,
- envelope.cluster_id
+ "Incompatible swarm from {peer}: expected {}, received {}",
+ expected_swarm,
+ envelope.swarm_id
);
}
@@ -462,7 +462,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
};
for peer in peer_keys {
- if peer_cluster(&state, &peer).await.is_some() {
+ if peer_swarm(&state, &peer).await.is_some() {
dirty_peers.insert(peer);
}
}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 6accf81..0afb5ee 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -47,14 +47,14 @@ pub struct HelloRequest {
#[derive(Debug, Clone)]
pub struct HelloReply {
- pub cluster_id: String,
+ pub swarm_id: String,
pub peers: Vec<String>,
pub epoch: Epoch,
}
#[derive(Debug, Clone)]
pub struct SyncEnvelope {
- pub cluster_id: String,
+ pub swarm_id: String,
/// Epoch of the document the message was generated from.
pub epoch: Epoch,
pub message: Message,
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index a8a0e10..194f998 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -203,7 +203,7 @@ pub async fn fetch_admin_info(rest_url: &Url) -> Result<MyceliumAdminInfo> {
}
#[derive(Debug, Clone)]
-pub struct ClusterNodeConfig {
+pub struct MyceliumNodeConfig {
pub tcp_port: u16,
pub api_port: u16,
pub rpc_port: u16,
@@ -213,7 +213,7 @@ pub struct ClusterNodeConfig {
pub struct SupervisedMycelium {
child: Child,
- pub config: ClusterNodeConfig,
+ pub config: MyceliumNodeConfig,
_key_file_handle: Option<File>,
_net_key_file_handle: Option<File>,
key_file_path: PathBuf,
@@ -251,7 +251,7 @@ impl SupervisedMycelium {
p => p,
};
- let config = ClusterNodeConfig {
+ let config = MyceliumNodeConfig {
tcp_port,
api_port,
rpc_port,
@@ -856,7 +856,7 @@ pub struct HelloRequestShadow {
#[derive(Serialize, Deserialize)]
pub struct HelloReplyShadow<'a> {
// Cow allows zero-copy references when saving, but creates an owned String when loading
- pub cluster_id: Cow<'a, str>,
+ pub swarm_id: Cow<'a, str>,
#[serde(default)]
pub peers: Vec<String>,
#[serde(default)]
@@ -865,7 +865,7 @@ pub struct HelloReplyShadow<'a> {
#[derive(Serialize, Deserialize)]
pub struct SyncEnvelopeShadow<'a> {
- pub cluster_id: Cow<'a, str>,
+ pub swarm_id: Cow<'a, str>,
#[serde(with = "automerge_msg_codec")]
pub message: automerge::sync::Message,
#[serde(default)]
@@ -917,7 +917,7 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr
// --- HelloReply ---
pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
let shadow = HelloReplyShadow {
- cluster_id: Cow::Borrowed(&msg.cluster_id),
+ swarm_id: Cow::Borrowed(&msg.swarm_id),
peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
epoch: epoch_to_shadow(&msg.epoch),
};
@@ -926,7 +926,7 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
Ok(HelloReply {
- cluster_id: shadow.cluster_id.into_owned(),
+ swarm_id: shadow.swarm_id.into_owned(),
peers: shadow.peers,
epoch: epoch_from_shadow(shadow.epoch)?,
})
@@ -935,7 +935,7 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError>
// --- SyncEnvelope ---
pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> {
let shadow = SyncEnvelopeShadow {
- cluster_id: Cow::Borrowed(&msg.cluster_id),
+ swarm_id: Cow::Borrowed(&msg.swarm_id),
message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
epoch: epoch_to_shadow(&msg.epoch),
};
@@ -944,7 +944,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::
pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
Ok(SyncEnvelope {
- cluster_id: shadow.cluster_id.into_owned(),
+ swarm_id: shadow.swarm_id.into_owned(),
epoch: epoch_from_shadow(shadow.epoch)?,
message: shadow.message,
})