urma_node/service/
overlay.rs

1//! Overlay steps of the Service round: peer discovery, then inventory sync.
2//! Each is optional, stop-guarded and keeps its own recorded error.
3use super::Service;
4use crate::{
5    Error,
6    model::{BookEntry, OverlayStatus},
7    peer::SyncPeers,
8    peer_discovery::Discovery,
9    peer_sync::{PeerSync, Scheduler},
10    recorded::{record, unix_seconds},
11    retrieval::RecordSource,
12};
13use std::sync::{Arc, atomic::Ordering};
14
15impl Service {
16    /// Overlay peers as record sources: every retrieval call takes the
17    /// current window as separate sources, so each package is verified on
18    /// its own and a bad one only moves retrieval on to the next peer.
19    pub fn with_record_peers(mut self, peers: SyncPeers) -> Self {
20        self.record_peers = Some(peers);
21        self
22    }
23
24    pub(super) fn record_sources(&self) -> Vec<Arc<dyn RecordSource + Send + Sync>> {
25        let mut sources = self.sources.clone();
26        let peers = self.record_peers.iter().flat_map(SyncPeers::record_sources);
27        sources.extend(peers.map(|peer| -> Arc<dyn RecordSource + Send + Sync> { Arc::new(peer) }));
28        sources
29    }
30
31    pub(super) fn record_source_count(&self) -> usize {
32        let peers = self
33            .record_peers
34            .as_ref()
35            .map_or(0, SyncPeers::record_count);
36        self.sources.len() + peers
37    }
38
39    /// Fast inventory synchronization from configured overlay peers. Only
40    /// run(true) schedules it; configuring peers does not dial or fetch.
41    pub fn with_peer_sync(mut self, sync: PeerSync) -> Result<Self, Error> {
42        self.peer_sync = Some(Scheduler::new(sync)?);
43        Ok(self)
44    }
45
46    /// Automatic peer discovery from seeds and peer address exchange. Only
47    /// run(true) schedules it, as its own bounded step before inventory sync.
48    pub fn with_discovery(mut self, discovery: Discovery) -> Result<Self, Error> {
49        self.discovery = Some(discovery);
50        Ok(self)
51    }
52
53    pub(super) fn discover_peers(&self) {
54        let Some(discovery) = &self.discovery else {
55            return;
56        };
57        if self.stop.load(Ordering::Acquire) {
58            return;
59        }
60        let result = self
61            .share_sync_backoff(discovery)
62            .and_then(|()| discovery.step(&self.engine, &self.stop));
63        if self.stop.load(Ordering::Acquire) {
64            return;
65        }
66        record(&self.discovery_error, &result);
67    }
68
69    /// Discovery must not redial a peer sync is holding off, or it would
70    /// refill the slot an eviction or idle release just freed.
71    fn share_sync_backoff(&self, discovery: &Discovery) -> Result<(), Error> {
72        let Some(scheduler) = &self.peer_sync else {
73            return Ok(());
74        };
75        discovery.hold_off(scheduler.backed_off(unix_seconds()?)?)
76    }
77
78    pub(super) fn sync_next_inventory(&self) {
79        let Some(scheduler) = &self.peer_sync else {
80            return;
81        };
82        if self.stop.load(Ordering::Acquire) {
83            return;
84        }
85        let result = scheduler
86            .step(&self.engine, &self.stop)
87            .and_then(|()| self.penalize_in_book(scheduler));
88        if self.stop.load(Ordering::Acquire) {
89            return;
90        }
91        record(&self.inventory_error, &result);
92    }
93
94    /// Peers the scheduler backed off also back off in the book, so discovery
95    /// stops dialing them; without discovery there is no book to tell.
96    fn penalize_in_book(&self, scheduler: &Scheduler) -> Result<(), Error> {
97        let penalized = scheduler.drain_penalized()?;
98        match &self.discovery {
99            Some(discovery) => discovery.penalize(&penalized),
100            None => Ok(()),
101        }
102    }
103
104    /// The sync view with discovery counters attached when both run.
105    pub(super) fn overlay_status(&self) -> Result<Option<OverlayStatus>, Error> {
106        let Some(scheduler) = &self.peer_sync else {
107            return Ok(None);
108        };
109        let mut overlay = scheduler.status(&self.engine)?;
110        if let Some(discovery) = &self.discovery {
111            overlay.discovery = Some(discovery.status()?);
112        }
113        Ok(Some(overlay))
114    }
115
116    /// `None` without discovery: the route answers 404 "overlay disabled".
117    pub(super) fn overlay_book(&self) -> Result<Option<Vec<BookEntry>>, Error> {
118        self.discovery.as_ref().map(Discovery::book).transpose()
119    }
120}
urma_node/service/overlay.rs

urma_node/service/
overlay.rs
