summaryrefslogtreecommitdiffstats
path: root/src/swarm/sync.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/sync.rs')
-rw-r--r--src/swarm/sync.rs42
1 files changed, 21 insertions, 21 deletions
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);
}
}