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        self.probe_commit_sources();
74        // Discovery keeps connections and the book fresh before inventory
75        // sync uses them; both are independent of the sync thread.
76        self.discover_peers();
77        self.sync_next_inventory();
78        self.acquire_next_dependency();
79        // Registered projections use already scanned data. They never
80        // create registrations or acquire missing history.
81        if !self.stopped() {
82            let result = self.engine.maintain_next_names(&self.stop);
83            record(&self.names_error, &result);
84        }
85        if !self.stopped() {
86            self.retain_next_object();
87        }
88        // Local cache upkeep never opens a source.
89        if !self.stopped() {
90            let result = self.engine.maintain_cache(self.cache_bytes, &self.stop);
91            if let Ok(mut status) = self.cache_status.lock() {
92                match result {
93                    Ok(receipt) => {
94                        status.receipt = Some(receipt);
95                        status.error = None;
96                    }
97                    Err(cause) => {
98                        status.error = Some(RecordedError::now(cause));
99                    }
100                }
101            }
102        }
103    }
104
105    /// One object slice from the last selected view; a sync failure does not
106    /// suppress it, and its error stays separately visible.
107    fn retain_next_object(&self) {
108        let Some((scratch, slice)) = &self.retention else {
109            return;
110        };
111        let owned = self.record_sources();
112        let sources: Vec<&dyn RecordSource> = owned
113            .iter()
114            .map(|source| -> &dyn RecordSource { source.as_ref() })
115            .collect();
116        let result = self
117            .engine
118            .retain_next_object(&sources, *slice, scratch, &self.stop);
119        record(&self.retention_error, &result);
120    }
121
122    /// Commit-source liveness for status, at each source's own cadence and
123    /// bound; only a broken status table is worth a dependency error here.
124    fn probe_commit_sources(&self) {
125        let Some(sources) = &self.commit_sources else {
126            return;
127        };
128        if self.stopped() {
129            return;
130        }
131        if let Err(error) = sources.probe() {
132            record(&self.dependency_error, &Err::<(), _>(error));
133        }
134    }
135
136    fn acquire_next_dependency(&self) {
137        let Some(sources) = &self.commit_sources else {
138            return;
139        };
140        if self.stopped() {
141            return;
142        }
143        let result = self.engine.resolve_next_dependency(sources, &self.stop);
144        if self.stopped() {
145            return;
146        }
147        record(&self.dependency_error, &result);
148    }
149}
urma_node/service/worker.rs

urma_node/service/
worker.rs
