From 2f3d191a12b6e03ef1845c202d15c7b90736558c Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 04:24:52 +0100 Subject: unify peers snapshot logic --- src/state/host.rs | 20 ++++---------------- src/state/mod.rs | 13 +++++++++++++ 2 files changed, 17 insertions(+), 16 deletions(-) (limited to 'src/state') diff --git a/src/state/host.rs b/src/state/host.rs index 9b149b8..fd40bf3 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -7,11 +7,11 @@ use std::time::Duration; use anyhow::Result; use automerge::{AutoCommit, ChangeHash}; -use tokio::sync::{Mutex, broadcast}; +use tokio::sync::broadcast; use tokio::time::Instant; use tracing::debug; -use super::{AppState, HelloState, Liveness, NodeLifecycle, PeerState, StaleView, Standing}; +use super::{AppState, HelloState, Liveness, NodeLifecycle, StaleView, Standing}; use crate::document::replica::Epoch; use crate::locks::MutexExt; use crate::remake::{Host, PeerSummary, RemakeState}; @@ -22,13 +22,7 @@ impl AppState { /// document, so start over with all of them. Peers still on the old epoch are /// skipped by the sync worker until they switch too. async fn after_epoch_switch(&self) { - let peers: Vec<(String, Arc>)> = self - .peer_states - .read() - .await - .iter() - .map(|(id, peer)| (id.clone(), Arc::clone(peer))) - .collect(); + let peers = self.peers_snapshot().await; // Sync states were reset under the document lock by the swap itself. self.publish_change(); @@ -126,13 +120,7 @@ impl Host for AppState { fn peer_summaries(&self) -> impl Future> + Send { let state = self.clone(); async move { - let peers: Vec<(String, Arc>)> = state - .peer_states - .read() - .await - .iter() - .map(|(id, peer)| (id.clone(), Arc::clone(peer))) - .collect(); + let peers = state.peers_snapshot().await; let mut summaries = Vec::with_capacity(peers.len()); for (id, peer) in peers { diff --git a/src/state/mod.rs b/src/state/mod.rs index 3336b00..1c31c61 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -280,6 +280,19 @@ impl AppState { Ok(()) } + /// Every known peer with its state, copied out of the peer map. Lock a peer's state + /// only from such a copy, never while holding the map's lock: the map would then + /// wait for the peer, and with it everything that only needs the map (a lookup, a + /// new peer). + pub(crate) async fn peers_snapshot(&self) -> Vec<(String, Arc>)> { + self.peer_states + .read() + .await + .iter() + .map(|(peer, state)| (peer.clone(), Arc::clone(state))) + .collect() + } + /// Looks up a peer without creating it. Background work (sync, retries, /// keepalives) must use this so a forgotten peer isn't resurrected. pub(crate) async fn get_peer_state(&self, pk: &str) -> Option>> { -- cgit v1.3.1