diff options
Diffstat (limited to 'src/state.rs')
| -rw-r--r-- | src/state.rs | 83 |
1 files changed, 68 insertions, 15 deletions
diff --git a/src/state.rs b/src/state.rs index df053a6..785eee8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -583,8 +583,8 @@ pub(crate) struct AppState { /// Per-node remake state (prebuilt snapshots, switch status, archive). pub(crate) remake: Arc<RemakeState>, - /// Pre-shared key every member must prove in the handshake; None admits anyone who - /// knows the swarm ID. + /// Pre-shared key every member must prove in the handshake; None admits any node + /// that reaches us. pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, /// Random per run of this node; our HELLOs carry it and replies must echo it. @@ -597,6 +597,11 @@ pub(crate) struct AppState { pub(crate) alias: Option<String>, /// Members alive within this window vote on remakes (`--electorate-window-days`). pub(crate) electorate_window: Duration, + /// Chunked transfers served and fetched (`swarm::blob`). + pub(crate) blobs: Arc<crate::swarm::blob::Blobs>, + /// HTTP API users who may write protocol data through the store API + /// (`--protocol-writers`). + pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>, /// The effective configuration with sources, imported values left out /// (`config::Loaded::report`), for `GET /admin/config`. pub(crate) config: Arc<serde_json::Value>, @@ -608,6 +613,9 @@ pub(crate) struct AppState { pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>, /// The neighbour we are pulling changes from, and since when (`swarm::relay`). pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>, + /// An announcement round is scheduled, for changes from these peers + /// (`swarm::relay`). + pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>, /// Our neighbours, once known (`swarm::topology`). pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>, @@ -638,6 +646,18 @@ impl AppState { self.transport.local_id() } + /// Peers that can serve our epoch's data: active, ready members on our epoch + /// (`PeerSummary::serves_epoch`). + pub(crate) async fn active_peers_on_local_epoch(&self) -> Vec<String> { + let epoch = self.local_epoch(); + self.peer_summaries() + .await + .into_iter() + .filter(|peer| peer.serves_epoch(epoch)) + .map(|peer| peer.id) + .collect() + } + pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> { self.rtt .lock() @@ -798,9 +818,14 @@ impl AppState { T: Send + 'static, { let doc = Arc::clone(&self.doc); - tokio::task::spawn_blocking(move || f(&mut doc.blocking_write().doc)) - .await - .expect("document write task panicked") + tokio::task::spawn_blocking(move || { + let mut replica = doc.blocking_write(); + let out = f(&mut replica.doc); + replica.refresh_heads(); + out + }) + .await + .expect("document write task panicked") } /// Mutates the document on the blocking pool as one Automerge change, then @@ -835,6 +860,7 @@ impl AppState { replica.doc.rollback(); } } + replica.refresh_heads(); result }; // lock released before notifying @@ -927,8 +953,8 @@ impl AppState { /// anyway with `unready_live`. /// /// With `with_heads`, a live read also returns the document's heads at the moment of - /// the read (for clients that base later requests on what they read). That needs the - /// write lock, so it is opt-in. Stale reads have no heads. + /// the read (for clients that base later requests on what they read). Stale reads + /// have no heads. pub(crate) async fn render_read<T, R>( &self, unready_live: bool, @@ -978,14 +1004,14 @@ impl AppState { if !with_heads { return (self.render(read, finish).await, None); } - // AutoCommit::get_heads takes &mut self (it closes a pending transaction). - let (data, mut heads) = self - .write_doc(move |doc| { - let heads = doc.get_heads(); - (read(doc), heads) - }) - .await; - heads.sort(); + // Heads and data under the same read lock (the heads are cached: `Replica`). + let doc = Arc::clone(&self.doc); + let (data, heads) = tokio::task::spawn_blocking(move || { + let replica = doc.blocking_read(); + (read(&replica.doc), replica.heads().to_vec()) + }) + .await + .expect("document read task panicked"); let rendered = tokio::task::spawn_blocking(move || finish(data)) .await .expect("document read task panicked"); @@ -1029,6 +1055,29 @@ impl Host for AppState { crate::swarm::members::electorate(doc, now_ms, self.electorate_window) } + fn fetch_change( + &self, + hash: ChangeHash, + ) -> impl Future<Output = Result<automerge::Change>> + Send { + let state = self.clone(); + async move { + let peers = state.active_peers_on_local_epoch().await; + crate::swarm::blob::fetch_change(&state, hash, peers).await + } + } + + fn fetch_content(&self, hash: [u8; 32]) -> impl Future<Output = Result<Vec<u8>>> + Send { + let state = self.clone(); + async move { + let peers = state.active_peers_on_local_epoch().await; + crate::swarm::blob::fetch_content(&state, hash, peers).await + } + } + + fn request_sync(&self) { + crate::swarm::source::resync_everyone(self); + } + fn is_node_ready(&self) -> impl Future<Output = bool> + Send { let state = self.clone(); async move { state.is_ready().await } @@ -1051,6 +1100,9 @@ impl Host for AppState { summaries.push(PeerSummary { id, active: peer.liveness == Liveness::Active, + member: matches!(peer.standing, Standing::Member { .. }) + && peer.hello_state == HelloState::Established, + ready: peer.reports_ready, epoch: peer.epoch, }); } @@ -1155,6 +1207,7 @@ impl Host for AppState { let new_doc = f(&mut replica.doc)?; let old_doc = std::mem::replace(&mut replica.doc, new_doc); replica.epoch = epoch; + replica.refresh_heads(); // Updated under the same lock, so the cached epoch never disagrees // with the document for anyone holding the lock. state.epoch_tx.send_replace(epoch); |