bsv_overlay_engine/
backfill.rs

1//! Bounded admin backfills. Continuations select work; persisted cursors own progress.
2
3use crate::{engine::GASPSyncResult, meter::SyncCost, sync_load::SyncFilter};
4use serde::{Deserialize, Serialize};
5
6/// A request stays within the existing 240 s sync belt with 60 s for finalization
7/// and accounting. HTTP duration itself is unbounded while connected; paid Workers
8/// default to 30 s CPU and 10,000 subrequests, with 128 MiB per isolate and six
9/// simultaneous connections. Paid D1 allows 1,000 queries per invocation. Stop at
10/// 600 measured SQL statements or 500 trait/network operations, leaving room for
11/// schema startup, an in-flight guarded submit and bookkeeping.
12/// These are conservative application bounds, not a CPU or SQL-row estimate.
13/// Source: https://developers.cloudflare.com/workers/platform/limits/
14/// Source: https://developers.cloudflare.com/d1/platform/limits/
15pub const BACKFILL_REQUEST_MS: u64 = 180_000;
16pub const BACKFILL_OPERATIONS: u64 = 500;
17pub const BACKFILL_D1_QUERIES: u64 = 600;
18pub const BACKFILL_MAX_SLICES: u64 = 64;
19
20#[derive(Debug, Clone, Default, Deserialize)]
21#[serde(deny_unknown_fields)]
22pub struct SyncRequest {
23    pub topics: Option<Vec<String>>,
24    pub peers: Option<Vec<String>>,
25    pub until: Option<Until>,
26    pub budget_ms: Option<u64>,
27    pub continuation: Option<String>,
28}
29
30#[derive(Debug, Clone, Copy, Deserialize)]
31#[serde(rename_all = "snake_case")]
32pub enum Until {
33    Tip,
34}
35
36impl SyncRequest {
37    pub fn parse(body: &str) -> Result<Self, String> {
38        let request = if body.trim().is_empty() {
39            Self::default()
40        } else {
41            serde_json::from_str(body).map_err(|e| e.to_string())?
42        };
43        request.validate()?;
44        Ok(request)
45    }
46
47    pub fn filter(&self) -> SyncFilter {
48        SyncFilter {
49            topics: self.topics.clone(),
50            peers: self.peers.clone(),
51        }
52    }
53
54    pub fn validate(&self) -> Result<(), String> {
55        if self.until.is_none() {
56            if self.budget_ms.is_some() || self.continuation.is_some() {
57                return Err("budget_ms and continuation require until: tip".into());
58            }
59            return Ok(());
60        }
61        let (topic, peer) = self.pair()?;
62        if self.budget_ms.is_none_or(|n| n == 0) {
63            return Err("until: tip requires a positive integer budget_ms".into());
64        }
65        if let Some(token) = &self.continuation {
66            let token = Continuation::decode(token)?;
67            if token.topic != topic || token.peer != peer {
68                return Err("continuation belongs to another topic or peer".into());
69            }
70        }
71        Ok(())
72    }
73
74    pub fn pair(&self) -> Result<(&str, &str), String> {
75        match (self.topics.as_deref(), self.peers.as_deref()) {
76            (Some([topic]), Some([peer]))
77                if !topic.is_empty()
78                    && topic.len() <= 256
79                    && !peer.is_empty()
80                    && peer.len() <= 1024 =>
81            {
82                Ok((topic, peer))
83            }
84            _ => Err(
85                "until: tip requires one topic (1..256 bytes) and one peer (1..1024 bytes)".into(),
86            ),
87        }
88    }
89}
90
91#[derive(Debug, Clone, Serialize, Deserialize)]
92#[serde(deny_unknown_fields)]
93pub struct Continuation {
94    version: u8,
95    pub topic: String,
96    pub peer: String,
97    pub cursor: u64,
98}
99
100impl Continuation {
101    pub fn new(topic: &str, peer: &str, cursor: u64) -> Self {
102        Self {
103            version: 1,
104            topic: topic.into(),
105            peer: peer.into(),
106            cursor,
107        }
108    }
109
110    pub fn encode(&self) -> String {
111        serde_json::to_string(self).expect("continuation has only scalar fields")
112    }
113
114    pub fn decode(token: &str) -> Result<Self, String> {
115        if token.len() > 16_384 {
116            return Err("continuation exceeds 16384 bytes".into());
117        }
118        let token: Self = serde_json::from_str(token).map_err(|_| "invalid continuation")?;
119        if token.version != 1 {
120            return Err("unsupported continuation version".into());
121        }
122        Ok(token)
123    }
124}
125
126#[derive(Debug, Clone, Default, Serialize, Deserialize)]
127pub struct PeerSyncResult {
128    #[serde(flatten)]
129    pub cost: SyncCost,
130    pub slices: u64,
131    pub pages: u64,
132    pub reached_tip: bool,
133    pub cursor: u64,
134}
135
136#[derive(Debug, Clone, Serialize)]
137pub struct BackfillResult {
138    #[serde(flatten)]
139    pub sync: GASPSyncResult,
140    pub slices: u64,
141    pub reached_tip: bool,
142    pub stop_reason: String,
143    pub continuation: Option<String>,
144}
145
146#[derive(Clone)]
147pub(crate) struct Control {
148    pub meter: crate::meter::LoadMeter,
149    pub budget_ms: u64,
150}
151
152impl Control {
153    pub fn remaining_ms(&self) -> u64 {
154        self.budget_ms.saturating_sub(self.meter.snapshot().ms)
155    }
156
157    pub fn stop_reason(&self) -> Option<&'static str> {
158        let cost = self.meter.snapshot();
159        if cost.ms >= self.budget_ms {
160            Some("wall_budget")
161        } else if cost.d1_queries >= BACKFILL_D1_QUERIES
162            || cost
163                .calls_out
164                .saturating_add(cost.storage_reads)
165                .saturating_add(cost.storage_writes)
166                >= BACKFILL_OPERATIONS
167        {
168            Some("request_allowance")
169        } else {
170            None
171        }
172    }
173}
174
175type ExpiredMemo = std::rc::Rc<std::cell::RefCell<std::collections::HashSet<String>>>;
176pub(crate) struct Run {
177    pub fetcher: Option<std::rc::Rc<dyn crate::gasp::AncestorFetcher>>,
178    expired: std::cell::RefCell<std::collections::HashMap<String, ExpiredMemo>>,
179    /// The peers asked one node at a time for the rest of this run (engine
180    /// 0.2.1, #58), by peer: never persisted, so a peer that gains the batch
181    /// route is asked for it on the next run.
182    single_only: std::cell::RefCell<std::collections::HashMap<String, SingleOnly>>,
183}
184type SingleOnly = std::rc::Rc<std::cell::Cell<bool>>;
185impl Run {
186    pub fn new(fetcher: Option<std::rc::Rc<dyn crate::gasp::AncestorFetcher>>) -> Self {
187        Self {
188            fetcher: fetcher
189                .map(|f| std::rc::Rc::new(crate::sync_fetcher::SyncFetcher::new(f)) as _),
190            expired: std::cell::RefCell::default(),
191            single_only: std::cell::RefCell::default(),
192        }
193    }
194    pub fn single_only(&self, peer: &str) -> SingleOnly {
195        self.single_only
196            .borrow_mut()
197            .entry(peer.to_string())
198            .or_default()
199            .clone()
200    }
201    pub fn expired(&self, topic: &str) -> ExpiredMemo {
202        self.expired
203            .borrow_mut()
204            .entry(topic.to_string())
205            .or_default()
206            .clone()
207    }
208}
209
210impl GASPSyncResult {
211    pub(crate) fn append_slice(&mut self, slice: Self) {
212        for topic in slice.topics_filtered {
213            if !self.topics_filtered.contains(&topic) {
214                self.topics_filtered.push(topic);
215            }
216        }
217        for (name, mut incoming) in slice.topics_synced {
218            let Some(topic) = self.topics_synced.get_mut(&name) else {
219                self.topics_synced.insert(name, incoming);
220                continue;
221            };
222            topic.cost.add(incoming.cost);
223            macro_rules! sum {
224                ($($field:ident),+ $(,)?) => { $(topic.$field = topic.$field.saturating_add(incoming.$field);)+ };
225            }
226            sum!(
227                roots_proven,
228                roots_unproven,
229                expired_refused,
230                pruned_inputs,
231                discarded_graphs,
232                anchor_walk_held_graphs,
233                finalized_graphs,
234                deadline_dropped_graphs,
235                deferred_graphs,
236                resumed_graphs,
237                converged_graphs,
238                stalled_graphs,
239                held_back_graphs,
240                backed_off_graphs,
241                waiting_graphs,
242                idle_counted,
243                unanchorable_graphs,
244                unanchorable_cleared,
245                absent_struck,
246                strikes_refused,
247                dead_graphs,
248                expired_swept,
249                memo_folded
250            );
251            topic.root_bounds.append(&mut incoming.root_bounds);
252            topic.peers_skipped.append(&mut incoming.peers_skipped);
253            topic.errors.append(&mut incoming.errors);
254            topic.route_faults.append(&mut incoming.route_faults);
255            topic.cursor_moves.append(&mut incoming.cursor_moves);
256            topic.dropped_graphs.append(&mut incoming.dropped_graphs);
257            topic.cursor_held_by = incoming.cursor_held_by;
258            for fault in incoming.faulted_roots {
259                if !topic
260                    .faulted_roots
261                    .iter()
262                    .any(|f| f.peer == fault.peer && f.outpoint == fault.outpoint)
263                {
264                    topic.faulted_roots.push(fault);
265                }
266            }
267            for (peer, result) in incoming.peer_results {
268                let held = topic.peer_results.entry(peer).or_default();
269                held.cost.add(result.cost);
270                held.slices += result.slices;
271                held.pages += result.pages;
272                held.reached_tip = result.reached_tip;
273                held.cursor = result.cursor;
274            }
275        }
276    }
277}
bsv_overlay_engine/backfill.rs

bsv_overlay_engine/
backfill.rs
