diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/bootstrap.rs | 28 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 63 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 42 |
3 files changed, 65 insertions, 68 deletions
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); } } |