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}
180impl Run {
181    pub fn new(fetcher: Option<std::rc::Rc<dyn crate::gasp::AncestorFetcher>>) -> Self {
182        Self {
183            fetcher: fetcher
184                .map(|f| std::rc::Rc::new(crate::sync_fetcher::SyncFetcher::new(f)) as _),
185            expired: std::cell::RefCell::default(),
186        }
187    }
188    pub fn expired(&self, topic: &str) -> ExpiredMemo {
189        self.expired
190            .borrow_mut()
191            .entry(topic.to_string())
192            .or_default()
193            .clone()
194    }
195}
196
197impl GASPSyncResult {
198    pub(crate) fn append_slice(&mut self, slice: Self) {
199        for topic in slice.topics_filtered {
200            if !self.topics_filtered.contains(&topic) {
201                self.topics_filtered.push(topic);
202            }
203        }
204        for (name, mut incoming) in slice.topics_synced {
205            let Some(topic) = self.topics_synced.get_mut(&name) else {
206                self.topics_synced.insert(name, incoming);
207                continue;
208            };
209            topic.cost.add(incoming.cost);
210            macro_rules! sum {
211                ($($field:ident),+ $(,)?) => { $(topic.$field = topic.$field.saturating_add(incoming.$field);)+ };
212            }
213            sum!(
214                roots_proven,
215                roots_unproven,
216                expired_refused,
217                pruned_inputs,
218                discarded_graphs,
219                anchor_walk_held_graphs,
220                finalized_graphs,
221                deadline_dropped_graphs,
222                deferred_graphs,
223                resumed_graphs,
224                converged_graphs,
225                stalled_graphs,
226                held_back_graphs,
227                backed_off_graphs,
228                waiting_graphs,
229                idle_counted,
230                unanchorable_graphs,
231                unanchorable_cleared,
232                strikes_refused,
233                dead_graphs,
234                expired_swept
235            );
236            topic.root_bounds.append(&mut incoming.root_bounds);
237            topic.peers_skipped.append(&mut incoming.peers_skipped);
238            topic.errors.append(&mut incoming.errors);
239            topic.cursor_moves.append(&mut incoming.cursor_moves);
240            topic.dropped_graphs.append(&mut incoming.dropped_graphs);
241            topic.cursor_held_by = incoming.cursor_held_by;
242            for fault in incoming.faulted_roots {
243                if !topic
244                    .faulted_roots
245                    .iter()
246                    .any(|f| f.peer == fault.peer && f.outpoint == fault.outpoint)
247                {
248                    topic.faulted_roots.push(fault);
249                }
250            }
251            for (peer, result) in incoming.peer_results {
252                let held = topic.peer_results.entry(peer).or_default();
253                held.cost.add(result.cost);
254                held.slices += result.slices;
255                held.pages += result.pages;
256                held.reached_tip = result.reached_tip;
257                held.cursor = result.cursor;
258            }
259        }
260    }
261}
bsv_overlay_engine/backfill.rs

bsv_overlay_engine/
backfill.rs
