bsv_overlay_engine/
meter.rs

1//! Poll-scoped counters: concurrent futures cannot charge each other's runs.
2use serde::{Deserialize, Serialize};
3use std::{
4    cell::{Cell, RefCell},
5    future::Future,
6    pin::Pin,
7    rc::Rc,
8    task::{Context, Poll},
9};
10
11/// Storage figures count attempted trait calls, including failures. They are
12/// operations, not SQL rows; discovery lookups count as peer requests.
13#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
14#[serde(default)]
15pub struct SyncCost {
16    pub calls_out: u64,
17    pub peer_requests: u64,
18    pub courier_rungs: u64,
19    pub storage_reads: u64,
20    pub storage_writes: u64,
21    pub d1_queries: u64,
22    pub courier_not_found_memo_hits: u64,
23    /// Requests the peers' remembered absence saved (engine 0.1.15): an
24    /// input the peer answered it does not hold, needed again in the same
25    /// sync, or a root whose counted absence waits out its gap or backoff.
26    pub peer_absent_memo_hits: u64,
27    /// Batch node requests sent (engine 0.2.1, #58): each is one body, also
28    /// counted in `calls_out` and `peer_requests`.
29    pub batch_requests: u64,
30    /// Nodes the batch answers served, each in place of a single request.
31    pub batch_nodes: u64,
32    /// Page batches the storage applied (engine 0.2.2): each is one
33    /// storage write, also counted in `storage_writes`.
34    pub storage_batches: u64,
35    /// The statements those batches ran, each in place of a write of its own.
36    pub storage_statements: u64,
37    pub ms: u64,
38    pub outputs_walked: u64,
39    pub ms_per_1000_outputs: Option<u64>,
40}
41
42impl SyncCost {
43    pub fn since(self, before: Self) -> Self {
44        let mut cost = Self {
45            calls_out: self.calls_out.saturating_sub(before.calls_out),
46            peer_requests: self.peer_requests.saturating_sub(before.peer_requests),
47            courier_rungs: self.courier_rungs.saturating_sub(before.courier_rungs),
48            storage_reads: self.storage_reads.saturating_sub(before.storage_reads),
49            storage_writes: self.storage_writes.saturating_sub(before.storage_writes),
50            d1_queries: self.d1_queries.saturating_sub(before.d1_queries),
51            courier_not_found_memo_hits: self
52                .courier_not_found_memo_hits
53                .saturating_sub(before.courier_not_found_memo_hits),
54            peer_absent_memo_hits: self
55                .peer_absent_memo_hits
56                .saturating_sub(before.peer_absent_memo_hits),
57            batch_requests: self.batch_requests.saturating_sub(before.batch_requests),
58            batch_nodes: self.batch_nodes.saturating_sub(before.batch_nodes),
59            storage_batches: self.storage_batches.saturating_sub(before.storage_batches),
60            storage_statements: self
61                .storage_statements
62                .saturating_sub(before.storage_statements),
63            ms: self.ms.saturating_sub(before.ms),
64            outputs_walked: self.outputs_walked.saturating_sub(before.outputs_walked),
65            ms_per_1000_outputs: None,
66        };
67        cost.refresh_speed();
68        cost
69    }
70
71    pub fn refresh_speed(&mut self) {
72        self.ms_per_1000_outputs = (self.outputs_walked != 0).then(|| {
73            ((u128::from(self.ms) * 1000) / u128::from(self.outputs_walked))
74                .min(u128::from(u64::MAX)) as u64
75        });
76    }
77
78    pub fn add(&mut self, other: Self) {
79        self.calls_out = self.calls_out.saturating_add(other.calls_out);
80        self.peer_requests = self.peer_requests.saturating_add(other.peer_requests);
81        self.courier_rungs = self.courier_rungs.saturating_add(other.courier_rungs);
82        self.storage_reads = self.storage_reads.saturating_add(other.storage_reads);
83        self.storage_writes = self.storage_writes.saturating_add(other.storage_writes);
84        self.d1_queries = self.d1_queries.saturating_add(other.d1_queries);
85        self.courier_not_found_memo_hits = self
86            .courier_not_found_memo_hits
87            .saturating_add(other.courier_not_found_memo_hits);
88        self.peer_absent_memo_hits = self
89            .peer_absent_memo_hits
90            .saturating_add(other.peer_absent_memo_hits);
91        self.batch_requests = self.batch_requests.saturating_add(other.batch_requests);
92        self.batch_nodes = self.batch_nodes.saturating_add(other.batch_nodes);
93        self.storage_batches = self.storage_batches.saturating_add(other.storage_batches);
94        self.storage_statements = self
95            .storage_statements
96            .saturating_add(other.storage_statements);
97        self.ms = self.ms.saturating_add(other.ms);
98        self.outputs_walked = self.outputs_walked.saturating_add(other.outputs_walked);
99        self.refresh_speed();
100    }
101}
102
103struct Data {
104    cost: Cell<SyncCost>,
105    started: f64,
106    clock: Rc<dyn Fn() -> f64>,
107}
108#[derive(Clone)]
109pub struct LoadMeter(Rc<Data>);
110impl Default for LoadMeter {
111    fn default() -> Self {
112        Self::with_clock(Rc::new(crate::engine::current_timestamp_ms))
113    }
114}
115impl LoadMeter {
116    /// Inject a monotonic millisecond clock for deterministic host tests.
117    pub fn with_clock(clock: Rc<dyn Fn() -> f64>) -> Self {
118        Self(Rc::new(Data {
119            cost: Cell::default(),
120            started: clock(),
121            clock,
122        }))
123    }
124    pub fn snapshot(&self) -> SyncCost {
125        let mut cost = self.0.cost.get();
126        cost.ms = ((self.0.clock)() - self.0.started).max(0.0) as u64;
127        cost.refresh_speed();
128        cost
129    }
130    pub fn scope<F: Future>(&self, future: F) -> Measured<F> {
131        Measured {
132            meter: self.clone(),
133            future: Box::pin(future),
134        }
135    }
136}
137thread_local! {
138    static CURRENT: RefCell<Option<LoadMeter>> = const { RefCell::new(None) };
139    static ROOT: RefCell<Option<Rc<RootLimit>>> = const { RefCell::new(None) };
140}
141struct Restore(Option<LoadMeter>);
142impl Drop for Restore {
143    fn drop(&mut self) {
144        CURRENT.with(|c| *c.borrow_mut() = self.0.take());
145    }
146}
147pub struct Measured<F> {
148    meter: LoadMeter,
149    future: Pin<Box<F>>,
150}
151impl<F: Future> Future for Measured<F> {
152    type Output = F::Output;
153    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
154        let _restore = Restore(CURRENT.with(|c| c.borrow_mut().replace(self.meter.clone())));
155        self.future.as_mut().poll(cx)
156    }
157}
158pub fn snapshot() -> SyncCost {
159    CURRENT.with(|c| {
160        c.borrow()
161            .as_ref()
162            .map_or_else(SyncCost::default, LoadMeter::snapshot)
163    })
164}
165fn note(f: impl FnOnce(&mut SyncCost)) {
166    CURRENT.with(|c| {
167        if let Some(m) = c.borrow().as_ref() {
168            let mut v = m.0.cost.get();
169            f(&mut v);
170            m.0.cost.set(v);
171        }
172    });
173}
174pub fn walk_output() {
175    note(|v| v.outputs_walked += 1);
176}
177pub fn peer_request() {
178    note(|v| {
179        v.calls_out += 1;
180        v.peer_requests += 1;
181    });
182}
183/// One batch node request (engine 0.2.1): a peer request like any other.
184pub fn batch_request() {
185    note(|v| {
186        v.calls_out += 1;
187        v.peer_requests += 1;
188        v.batch_requests += 1;
189    });
190}
191pub fn batch_nodes(count: u64) {
192    note(|v| v.batch_nodes = v.batch_nodes.saturating_add(count));
193}
194pub fn storage_read() {
195    note(|v| v.storage_reads += 1);
196}
197pub fn storage_write() {
198    note(|v| v.storage_writes += 1);
199}
200/// One page batch applied (engine 0.2.2), of `statements` statements; its
201/// storage write is counted by [`storage_write`].
202pub fn storage_batch(statements: u64) {
203    note(|v| {
204        v.storage_batches += 1;
205        v.storage_statements = v.storage_statements.saturating_add(statements);
206    });
207}
208/// Count attempted SQL statements at the worker adapter, including batch members.
209pub fn d1_queries(count: u64) {
210    note(|v| v.d1_queries = v.d1_queries.saturating_add(count));
211}
212pub fn not_found_memo_hit() {
213    note(|v| v.courier_not_found_memo_hits += 1);
214}
215pub fn peer_absent_memo_hit() {
216    note(|v| v.peer_absent_memo_hits += 1);
217}
218
219/// Twelve courier HTTP attempts at at most 1,000 ms each fit in a 15,000 ms
220/// graph slice, leaving 3,000 ms for peer/storage work. With the 30,000 ms
221/// peer slice and 9,900 ms resume share, the page retains 20,100 ms. These
222/// are stop bounds, not promises about finalize/storage latency.
223pub const ROOT_MAX_COURIER_RUNGS: u32 = 12;
224pub const COURIER_RUNG_TIMEOUT_MS: u64 = 1_000;
225pub const ROOT_MAX_MS: u64 = 15_000;
226#[derive(Default)]
227struct RootLimit {
228    calls: Cell<u32>,
229    exhausted: Cell<bool>,
230}
231pub struct RootScope<F> {
232    limit: Rc<RootLimit>,
233    future: Pin<Box<F>>,
234}
235pub fn root_scope<F: Future>(future: F) -> RootScope<F> {
236    RootScope {
237        limit: Rc::default(),
238        future: Box::pin(future),
239    }
240}
241struct RestoreRoot(Option<Rc<RootLimit>>);
242impl Drop for RestoreRoot {
243    fn drop(&mut self) {
244        ROOT.with(|c| *c.borrow_mut() = self.0.take());
245    }
246}
247impl<F: Future> Future for RootScope<F> {
248    type Output = F::Output;
249    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
250        let _restore = RestoreRoot(ROOT.with(|c| c.borrow_mut().replace(self.limit.clone())));
251        self.future.as_mut().poll(cx)
252    }
253}
254pub fn root_scoped() -> bool {
255    ROOT.with(|c| c.borrow().is_some())
256}
257pub fn root_exhausted() -> bool {
258    ROOT.with(|c| c.borrow().as_ref().is_some_and(|r| r.exhausted.get()))
259}
260/// Charge before awaiting HTTP. A refused charge sends no request.
261pub fn courier_request() -> Result<(), &'static str> {
262    let allowed = ROOT.with(|c| {
263        c.borrow().as_ref().is_none_or(|r| {
264            if r.calls.get() >= ROOT_MAX_COURIER_RUNGS {
265                r.exhausted.set(true);
266                false
267            } else {
268                r.calls.set(r.calls.get() + 1);
269                true
270            }
271        })
272    });
273    if !allowed {
274        return Err("root courier rung budget exhausted");
275    }
276    note(|v| {
277        v.calls_out += 1;
278        v.courier_rungs += 1;
279    });
280    Ok(())
281}
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286    #[tokio::test]
287    async fn load_0112_cancelled_run_keeps_its_attempts() {
288        let meter = LoadMeter::default();
289        let mut task = Box::pin(meter.scope(async {
290            peer_request();
291            storage_write();
292            std::future::pending::<()>().await;
293        }));
294        std::future::poll_fn(|cx| {
295            assert!(task.as_mut().poll(cx).is_pending());
296            Poll::Ready(())
297        })
298        .await;
299        drop(task);
300        assert_eq!(
301            (
302                meter.snapshot().peer_requests,
303                meter.snapshot().storage_writes
304            ),
305            (1, 1)
306        );
307        assert_eq!(snapshot(), SyncCost::default());
308    }
309
310    #[tokio::test]
311    async fn load_0112_poll_scopes_keep_interleaved_runs_separate() {
312        let a = LoadMeter::default();
313        let b = LoadMeter::default();
314        let ((), ()) = tokio::join!(
315            a.scope(async {
316                peer_request();
317                tokio::task::yield_now().await;
318                storage_read();
319            }),
320            b.scope(async {
321                storage_write();
322                tokio::task::yield_now().await;
323                courier_request().unwrap();
324            })
325        );
326        assert_eq!(
327            (
328                a.snapshot().peer_requests,
329                a.snapshot().storage_reads,
330                a.snapshot().storage_writes
331            ),
332            (1, 1, 0)
333        );
334        assert_eq!(
335            (
336                b.snapshot().courier_rungs,
337                b.snapshot().storage_writes,
338                b.snapshot().storage_reads
339            ),
340            (1, 1, 0)
341        );
342        assert_eq!(snapshot(), SyncCost::default());
343    }
344}
bsv_overlay_engine/meter.rs

bsv_overlay_engine/
meter.rs
