summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/peers.rs7
-rw-r--r--src/state.rs40
-rw-r--r--src/swarm/bootstrap.rs7
-rw-r--r--src/swarm/peer.rs24
-rw-r--r--src/swarm/sync.rs12
5 files changed, 47 insertions, 43 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs
index 65f8fa1..6637da4 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -16,7 +16,7 @@ use utoipa::ToSchema;
use super::store_status;
use crate::document::replica::Epoch;
use crate::document::store;
-use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState, unix_millis};
+use crate::state::{AppState, HelloState, Liveness, PeerState, unix_millis};
use crate::swarm::peer::forgotten_path;
#[derive(Debug, Serialize, ToSchema)]
@@ -112,10 +112,7 @@ pub(crate) async fn list_peers(
peers.push(PeerStatus {
public_key,
- validated_swarm: match &ps.membership {
- PeerMembership::Unknown => None,
- PeerMembership::Member { swarm_id } => Some(swarm_id.clone()),
- },
+ validated_swarm: ps.standing.swarm_id().map(str::to_owned),
status,
handshake_complete: ps.hello_state == HelloState::Established,
last_seen_secs_ago: ps
diff --git a/src/state.rs b/src/state.rs
index ccc8253..33bc9b5 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -25,13 +25,32 @@ pub(crate) enum NodeLifecycle {
Ready { swarm_id: String },
}
+/// How far a peer got into the swarm, as far as we're concerned.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub(crate) enum PeerMembership {
+pub(crate) enum Standing {
+ /// Nothing proven yet. With a swarm key, everything but a handshake from it is
+ /// ignored (`swarm::peer::admit`).
#[default]
- Unknown,
- Member {
- swarm_id: String,
- },
+ Stranger,
+ /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key.
+ Admitted,
+ /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it.
+ Member { swarm_id: String },
+}
+
+impl Standing {
+ /// The swarm a member belongs to; None before its HELLO_REPLY.
+ pub(crate) fn swarm_id(&self) -> Option<&str> {
+ match self {
+ Self::Member { swarm_id } => Some(swarm_id),
+ Self::Stranger | Self::Admitted => None,
+ }
+ }
+
+ /// Proved the swarm key (meaningful only with one).
+ pub(crate) fn is_admitted(&self) -> bool {
+ !matches!(self, Self::Stranger)
+ }
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -67,7 +86,7 @@ pub(crate) struct PendingOutbound {
#[derive(Clone)]
pub(crate) struct PeerState {
- pub(crate) membership: PeerMembership,
+ pub(crate) standing: Standing,
pub(crate) sync_state: sync::State,
pub(crate) hello_state: HelloState,
@@ -102,10 +121,6 @@ pub(crate) struct PeerState {
/// Consecutive sync exchanges that changed nothing (livelock detection).
pub(crate) stalled_exchanges: u32,
- /// The peer proved the swarm key in its HELLO or HELLO_REPLY (always false and
- /// irrelevant without a swarm key).
- pub(crate) authenticated: bool,
-
/// The answer we expect to our last sync message(s), if any.
pub(crate) sync_reply: Option<SyncReplyWait>,
/// How long to wait for it; doubles while answers stay out.
@@ -144,7 +159,7 @@ const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
impl Default for PeerState {
fn default() -> Self {
Self {
- membership: PeerMembership::default(),
+ standing: Standing::default(),
sync_state: sync::State::new(),
hello_state: HelloState::default(),
reply_owed: false,
@@ -158,7 +173,6 @@ impl Default for PeerState {
last_sync_rx: None,
last_resync_at: None,
stalled_exchanges: 0,
- authenticated: false,
sync_reply: None,
sync_reply_timeout: SYNC_REPLY_TIMEOUT,
sync_reset_at: Instant::now(),
@@ -668,7 +682,7 @@ impl AppState {
for (id, peer_state_mutex) in &peers {
let mut peer_state = peer_state_mutex.lock().await;
let syncs_with_us = peer_state.liveness == Liveness::Active
- && matches!(peer_state.membership, PeerMembership::Member { .. })
+ && matches!(peer_state.standing, Standing::Member { .. })
&& peer_state.epoch == Some(local);
peer_state.epoch_unannounced = syncs_with_us;
if !syncs_with_us {
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 3b6e510..edadce2 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -7,7 +7,7 @@ use anyhow::{Result, bail};
use tracing::{debug, error, info, warn};
use uuid::Uuid;
-use crate::state::{AppState, PeerMembership};
+use crate::state::AppState;
use crate::swarm::peer::{remember_peer, send_hello_once};
const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300);
@@ -68,10 +68,7 @@ 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_swarm = match &peer_state.membership {
- PeerMembership::Unknown => None,
- PeerMembership::Member { swarm_id } => Some(swarm_id.as_str()),
- };
+ let validated_swarm = peer_state.standing.swarm_id();
peer_diagnostics.push(format!(
"{peer}: validated_swarm={:?}, epoch={:?}, pending_inbound_sync={}",
validated_swarm,
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index cec8842..9de0ce0 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -13,7 +13,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::state::{AppState, HelloState, Liveness, PeerState, Standing};
use crate::swarm::sync::{SyncEvent, handle_sync};
use crate::transport::{self, ParsedMessage};
@@ -479,7 +479,7 @@ 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 { swarm_id: existing } = &peer_state.membership {
+ if let Standing::Member { swarm_id: existing } = &peer_state.standing {
if existing != &swarm_id {
bail!(
"Peer {peer} changed swarm identity from {} to {}",
@@ -494,8 +494,8 @@ async fn establish_peer(
state.set_expected_swarm_id(swarm_id.clone()).await?;
}
- if matches!(peer_state.membership, PeerMembership::Unknown) {
- peer_state.membership = PeerMembership::Member {
+ if !matches!(peer_state.standing, Standing::Member { .. }) {
+ peer_state.standing = Standing::Member {
swarm_id: swarm_id.clone(),
};
}
@@ -518,10 +518,7 @@ async fn establish_peer(
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 { swarm_id } => Some(swarm_id.clone()),
- }
+ peer_state.standing.swarm_id().map(str::to_owned)
}
// --- epochs ------------------------------------------------------------------
@@ -609,8 +606,8 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
remember_peer(state, peer).await;
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- if !peer_state.authenticated {
- peer_state.authenticated = true;
+ if peer_state.standing == Standing::Stranger {
+ peer_state.standing = Standing::Admitted;
debug!(peer = %peer, "Peer proved the swarm key");
}
}
@@ -622,11 +619,10 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
// we stop syncing our document to it, not just listening to it.
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- if peer_state.authenticated {
+ if peer_state.standing.is_admitted() {
warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership");
}
- peer_state.authenticated = false;
- peer_state.membership = PeerMembership::Unknown;
+ peer_state.standing = Standing::Stranger;
peer_state.hello_state = HelloState::NotSent;
peer_state.reset_sync();
}
@@ -643,7 +639,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
}
None => {
let admitted = match state.get_peer_state(peer).await {
- Some(peer_state) => peer_state.lock().await.authenticated,
+ Some(peer_state) => peer_state.lock().await.standing.is_admitted(),
None => false,
};
if !admitted {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 567b1ea..22d6600 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -12,7 +12,7 @@ use tokio::sync::mpsc;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
-use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
+use crate::state::{AppState, Liveness, PeerState, PendingOutbound, Standing};
use crate::swarm::peer::{peer_swarm, send_hello_once};
use crate::transport::{self, ParsedMessage};
@@ -58,7 +58,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(());
}
- let PeerMembership::Member { swarm_id } = &peer_state.membership else {
+ let Standing::Member { swarm_id } = &peer_state.standing else {
return Ok(());
};
@@ -249,8 +249,8 @@ pub(crate) async fn handle_sync(
let needs_hello = {
let mut peer_state = peer_state_mutex.lock().await;
- match &peer_state.membership {
- PeerMembership::Unknown => {
+ match &peer_state.standing {
+ Standing::Stranger | Standing::Admitted => {
let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| {
pending.message == envelope.message
&& pending.swarm_id == envelope.swarm_id
@@ -268,13 +268,13 @@ pub(crate) async fn handle_sync(
}
true
}
- PeerMembership::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
+ Standing::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
bail!(
"Sync swarm mismatch for peer {peer}: validated_swarm={swarm_id}, envelope_swarm={}",
envelope.swarm_id
);
}
- PeerMembership::Member { .. } => false,
+ Standing::Member { .. } => false,
}
};
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-25 15:50:20 +0000