urma_node/service/
worker.rs

1//! Background work of run(true), on two threads so neither waits on the other.
2//! The sync thread selects headers and scans blocks back to back while behind
3//! the selected tip, pausing only after a round without scan progress. The
4//! maintenance thread runs discovery, peer inventory, parent acquisition,
5//! Names, object retention and cache upkeep, pausing after every round.
6use super::Service;
7use crate::{
8    engine_sync::CatchUp, recorded::RecordedError, recorded::record, retrieval::RecordSource,
9};
10use std::{sync::atomic::Ordering, time::Duration};
11
12impl Service {
13    /// Most blocks one sync round scans before it re-selects headers.
14    pub const CATCH_UP_BLOCKS: u64 = 256;
15    /// Longest scanning stretch of one sync round.
16    pub const CATCH_UP_TIME: Duration = Duration::from_secs(5);
17    const PAUSE_STEPS: u32 = 20;
18    const PAUSE_STEP: Duration = Duration::from_millis(100);
19
20    fn stopped(&self) -> bool {
21        self.stop.load(Ordering::Acquire)
22    }
23
24    /// Two seconds in short steps, ending early on stop.
25    fn pause(&self) {
26        for _ in 0..Self::PAUSE_STEPS {
27            if self.stopped() {
28                return;
29            }
30            std::thread::sleep(Self::PAUSE_STEP);
31        }
32    }
33
34    pub(super) fn sync_loop(&self) {
35        while !self.stopped() {
36            if !self.sync_round() {
37                self.pause();
38            }
39        }
40    }
41
42    pub(super) fn maintenance_loop(&self) {
43        while !self.stopped() {
44            self.maintenance_round();
45            self.pause();
46        }
47    }
48
49    /// Headers, then a bounded catch-up scan and one cached parent retry.
50    /// Returns whether any block was scanned.
51    fn sync_round(&self) -> bool {
52        let catch_up = match self.engine.synchronize_headers_until(&self.stop) {
53            Ok(_) => {
54                self.engine
55                    .scan_catch_up(&self.stop, Self::CATCH_UP_BLOCKS, Self::CATCH_UP_TIME)
56            }
57            Err(error) => CatchUp {
58                scanned: 0,
59                outcome: Err(error),
60            },
61        };
62        let result = catch_up.outcome.and_then(|status| {
63            if !self.stopped() {
64                self.engine.resolve_cached_dependency()?;
65            }
66            Ok(status)
67        });
68        record(&self.sync_error, &result);
69        catch_up.scanned > 0
70    }
71
72    fn maintenance_round(&self) {
73        // Discovery keeps connections and the book fresh before inventory
74        // sync uses them; both are independent of the sync thread.
75        self.discover_peers();
76        self.sync_next_inventory();
77        self.acquire_next_dependency();
78        // Registered projections use already scanned data. They never
79        // create registrations or acquire missing history.
80        if !self.stopped() {
81            let result = self.engine.maintain_next_names(&self.stop);
82            record(&self.names_error, &result);
83        }
84        if !self.stopped() {
85            self.retain_next_object();
86        }
87        // Local cache upkeep never opens a source.
88        if !self.stopped() {
89            let result = self.engine.maintain_cache(self.cache_bytes, &self.stop);
90            if let Ok(mut status) = self.cache_status.lock() {
91                match result {
92                    Ok(receipt) => {
93                        status.receipt = Some(receipt);
94                        status.error = None;
95                    }
96                    Err(cause) => {
97                        status.error = Some(RecordedError::now(cause));
98                    }
99                }
100            }
101        }
102    }
103
104    /// One object slice from the last selected view; a sync failure does not
105    /// suppress it, and its error stays separately visible.
106    fn retain_next_object(&self) {
107        let Some((scratch, slice)) = &self.retention else {
108            return;
109        };
110        let owned = self.record_sources();
111        let sources: Vec<&dyn RecordSource> = owned
112            .iter()
113            .map(|source| -> &dyn RecordSource { source.as_ref() })
114            .collect();
115        let result = self
116            .engine
117            .retain_next_object(&sources, *slice, scratch, &self.stop);
118        record(&self.retention_error, &result);
119    }
120
121    fn acquire_next_dependency(&self) {
122        let Some(sources) = &self.commit_sources else {
123            return;
124        };
125        if self.stopped() {
126            return;
127        }
128        let result = self.engine.resolve_next_dependency(sources, &self.stop);
129        if self.stopped() {
130            return;
131        }
132        record(&self.dependency_error, &result);
133    }
134}
urma_node/service/worker.rs

urma_node/service/
worker.rs
