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

bsv_overlay_engine/
meter.rs
