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

bsv_overlay_engine/
meter.rs
