urma_node/
engine_sync.rs

1//! Peer synchronization bookkeeping. Coverage is a local claim about which
2//! peer listings were fully admitted; full-block scans remain the audit.
3use crate::{
4    Error,
5    engine::Engine,
6    model::{ArchiveStatus, ChainPoint, Omission},
7};
8use std::{
9    sync::atomic::{AtomicBool, Ordering},
10    time::{Duration, Instant},
11};
12
13/// One catch-up pass: how many blocks it committed and how it ended.
14pub struct CatchUp {
15    pub scanned: u64,
16    pub outcome: Result<ArchiveStatus, Error>,
17}
18
19impl Engine {
20    /// Scan selected blocks one after another until the selected tip, `blocks`
21    /// commits, `time` elapsed or cancellation, which is checked between
22    /// blocks. Each block is still its own verified, atomic commit; the first
23    /// error ends the pass and keeps every block committed before it.
24    pub fn scan_catch_up(&self, cancelled: &AtomicBool, blocks: u64, time: Duration) -> CatchUp {
25        let deadline = Instant::now() + time;
26        let mut scanned = 0;
27        let mut last = match self.status() {
28            Ok(status) => status,
29            Err(error) => {
30                return CatchUp {
31                    scanned,
32                    outcome: Err(error),
33                };
34            }
35        };
36        while scanned < blocks && Instant::now() < deadline && !cancelled.load(Ordering::Acquire) {
37            match self.scan_next_until(cancelled) {
38                Ok(next) if next.scanned_blocks == last.scanned_blocks => {
39                    last = next;
40                    break;
41                }
42                Ok(next) => {
43                    scanned += 1;
44                    last = next;
45                }
46                Err(error) => {
47                    return CatchUp {
48                        scanned,
49                        outcome: Err(error),
50                    };
51                }
52            }
53        }
54        CatchUp {
55            scanned,
56            outcome: Ok(last),
57        }
58    }
59
60    /// Record that `peer` (its PeerId string) listed every placement in
61    /// selected [from, to] and each listed placement was admitted, already
62    /// held, or answered Missing (listed-only).
63    /// `generation` is the one observed before fetching; a reorg since is Stale.
64    /// admit_record keeps multipart children even in Pruned mode, so synced
65    /// components stay until cache maintenance evicts them.
66    pub fn record_peer_coverage(
67        &self,
68        peer: &str,
69        from: u64,
70        to: u64,
71        generation: u64,
72    ) -> Result<(), Error> {
73        self.archive()?
74            .record_peer_coverage(peer, from, to, generation)
75    }
76
77    /// First height after the peer's contiguous coverage from app start.
78    pub fn peer_cursor(&self, peer: &str) -> Result<u64, Error> {
79        self.archive()?.peer_cursor(peer, 0)
80    }
81
82    /// As `peer_cursor`, starting at the peer's own scan floor when that is
83    /// above app start: heights below it are never asked of this peer.
84    pub fn peer_cursor_from(&self, peer: &str, floor: u64) -> Result<u64, Error> {
85        self.archive()?.peer_cursor(peer, floor)
86    }
87
88    pub fn peer_synced_through(&self) -> Result<Option<ChainPoint>, Error> {
89        self.archive()?.peer_synced_through()
90    }
91
92    pub fn omitting_peers(&self) -> Result<Vec<String>, Error> {
93        self.archive()?.omitting_peers()
94    }
95
96    pub fn peer_omissions(&self) -> Result<Vec<Omission>, Error> {
97        self.archive()?.peer_omissions()
98    }
99
100    /// The `limit` highest selected omissions, ascending; bounded in SQL.
101    pub fn peer_omissions_latest(&self, limit: u32) -> Result<Vec<Omission>, Error> {
102        self.archive()?.peer_omissions_latest(limit)
103    }
104
105    /// Chain peers the source holds now; 0 without a source (offline).
106    pub fn chain_peers(&self) -> Result<u64, Error> {
107        match &self.source {
108            Some(source) => source.chain_peers(),
109            None => Ok(0),
110        }
111    }
112
113    /// Absolute header-sync target; `None` while unknown or offline.
114    pub fn chain_target(&self) -> Option<u64> {
115        self.source
116            .as_ref()
117            .and_then(|source| source.chain_target())
118    }
119}
urma_node/engine_sync.rs

urma_node/
engine_sync.rs
