bsv_overlay_engine/
page_batch.rs

1//! The sync's storage per page (engine 0.2.2, the owner's word of
2//! 2026-10-15 on #58's finding): a GASP page's admissions are written in one
3//! storage batch at the page's end, and the reads they need are made for the
4//! whole page at once.
5//!
6//! The engine installs a [`PageBuffer`] around a peer's sync future
7//! ([`scope`], poll-scoped like [`crate::meter`], so a concurrent request
8//! of the same isolate never sees it). The sync opens it at each page and
9//! closes it at the page's end. While it is open the engine's storage
10//! wrapper (`load_storage::MeteredStorage`):
11//!
12//! - queues the admission writes ([`PageWrite`]): an output with its
13//!   transaction, an applied row, and a backend's own statement ([`defer`],
14//!   the lookup index rows of the Worker);
15//! - answers `find_output` (topic given, no spent filter, no BEEF) and
16//!   `does_applied_transaction_exist` from the page's own writes, the page's
17//!   prefetch ([`prefetch`]) and the answers already read in the page;
18//! - flushes the queue before every other write and before every read of
19//!   the tables the queue writes (a barrier), so the store sees the same
20//!   writes in the same order as one write at a time would have written them.
21//!
22//! The queue stays in the buffer until the backend answers the batch `Ok`: a
23//! batch that fails, or a sync dropped while its batch is in flight, leaves
24//! it there, and the engine flushes it again after the sync or fails the
25//! page. Every queued statement is idempotent (insert or ignore, the BEEF's
26//! replace), so a batch applied twice writes once.
27use crate::storage::{Storage, StorageError};
28use crate::types::{AppliedTransaction, Output};
29use std::{
30    any::Any,
31    cell::RefCell,
32    collections::{HashMap, HashSet},
33    future::Future,
34    pin::Pin,
35    rc::Rc,
36    task::{Context, Poll},
37};
38
39/// One write of a page's batch, in the order the admission made it.
40#[derive(Clone)]
41pub enum PageWrite {
42    /// `Storage::insert_output_with_origin(output, gasp)`.
43    Output { output: Rc<Output>, gasp: bool },
44    /// `Storage::insert_applied_transaction`.
45    Applied(AppliedTransaction),
46    /// A statement of the backend's own (a lookup index row), opaque to the
47    /// engine: the backend that deferred it applies it in its batch.
48    Backend(Rc<dyn Any>),
49}
50
51/// The page's answer to a topic output read it knows: the output, or none
52/// held.
53pub(crate) struct Known(pub Option<Output>);
54
55/// The key of a topic output read: `(txid, output_index, topic)`.
56type OutputKey = (String, u32, String);
57
58#[derive(Default)]
59struct State {
60    open: bool,
61    writes: Vec<PageWrite>,
62    /// Answers of `find_output(txid, vout, Some(topic), None, false)` known
63    /// in this page: prefetched, read, or written by the page.
64    outputs: HashMap<OutputKey, Option<Output>>,
65    /// Answers of `does_applied_transaction_exist`, keyed `(txid, topic)`.
66    applied: HashMap<(String, String), bool>,
67}
68
69/// A peer sync's page buffer (see the module).
70#[derive(Clone, Default)]
71pub struct PageBuffer(Rc<RefCell<State>>, Rc<RefCell<Option<Bound>>>);
72
73/// The bound of one batch: the engine's finalize submit budget.
74type Bound = (crate::engine::SleepFactory, u64);
75
76impl PageBuffer {
77    /// Bound each batch as one finalize submit's storage call is bounded
78    /// (bsv-low #559): a batch with no answer in `budget_ms` is a fault.
79    pub fn with_bound(self, sleep: crate::engine::SleepFactory, budget_ms: u64) -> Self {
80        *self.1.borrow_mut() = Some((sleep, budget_ms));
81        self
82    }
83
84    /// Start buffering: the sync's page begins.
85    pub fn open(&self) {
86        let mut state = self.0.borrow_mut();
87        state.open = true;
88        // Every queued write's key stays answered while it is queued.
89        if state.writes.is_empty() {
90            state.outputs.clear();
91            state.applied.clear();
92        }
93    }
94
95    /// Stop buffering. Writes still queued stay until a flush takes them.
96    pub fn close(&self) {
97        let mut state = self.0.borrow_mut();
98        state.open = false;
99        if state.writes.is_empty() {
100            state.outputs.clear();
101            state.applied.clear();
102        }
103    }
104
105    pub fn is_open(&self) -> bool {
106        self.0.borrow().open
107    }
108
109    /// Writes queued and not yet applied.
110    pub fn pending(&self) -> usize {
111        self.0.borrow().writes.len()
112    }
113
114    pub(crate) fn queue_output(&self, output: &Output, gasp: bool) {
115        let mut state = self.0.borrow_mut();
116        let mut held = output.clone();
117        held.beef = None;
118        state.outputs.insert(
119            (
120                output.txid.clone(),
121                output.output_index,
122                output.topic.clone(),
123            ),
124            Some(held),
125        );
126        state.writes.push(PageWrite::Output {
127            output: Rc::new(output.clone()),
128            gasp,
129        });
130    }
131
132    pub(crate) fn queue_applied(&self, tx: &AppliedTransaction) {
133        let mut state = self.0.borrow_mut();
134        state
135            .applied
136            .insert((tx.txid.clone(), tx.topic.clone()), true);
137        state.writes.push(PageWrite::Applied(tx.clone()));
138    }
139
140    /// The page's answer for a topic output read, if it has one.
141    pub(crate) fn output(&self, txid: &str, vout: u32, topic: &str) -> Option<Known> {
142        self.0
143            .borrow()
144            .outputs
145            .get(&(txid.to_string(), vout, topic.to_string()))
146            .cloned()
147            .map(Known)
148    }
149
150    pub(crate) fn remember_output(
151        &self,
152        txid: &str,
153        vout: u32,
154        topic: &str,
155        answer: Option<Output>,
156    ) {
157        self.0
158            .borrow_mut()
159            .outputs
160            .entry((txid.to_string(), vout, topic.to_string()))
161            .or_insert(answer);
162    }
163
164    pub(crate) fn applied(&self, tx: &AppliedTransaction) -> Option<bool> {
165        self.0
166            .borrow()
167            .applied
168            .get(&(tx.txid.clone(), tx.topic.clone()))
169            .copied()
170    }
171
172    pub(crate) fn remember_applied(&self, tx: &AppliedTransaction, held: bool) {
173        self.0
174            .borrow_mut()
175            .applied
176            .entry((tx.txid.clone(), tx.topic.clone()))
177            .or_insert(held);
178    }
179
180    /// Drop the queued writes unapplied: the page failed.
181    pub fn discard(&self) {
182        let mut state = self.0.borrow_mut();
183        state.writes.clear();
184        state.outputs.clear();
185        state.applied.clear();
186    }
187
188    /// Forget every answer (a write the page does not model changed rows).
189    pub(crate) fn forget_answers(&self) {
190        let mut state = self.0.borrow_mut();
191        state.outputs.clear();
192        state.applied.clear();
193    }
194
195    /// Apply the queued writes as one batch through `storage` (the engine's
196    /// own wrapper, which meters it). The writes leave the queue only when
197    /// the batch answers `Ok`.
198    pub async fn flush(&self, storage: &dyn Storage) -> Result<(), StorageError> {
199        let writes = self.0.borrow().writes.clone();
200        if writes.is_empty() {
201            return Ok(());
202        }
203        let bound = self.1.borrow().clone();
204        match bound {
205            Some((sleep, budget_ms)) => {
206                crate::gasp::race_or_deadline(storage.write_batch(&writes), sleep(budget_ms))
207                    .await
208                    .ok_or_else(|| {
209                        StorageError::Database(format!(
210                            "the page's batch had no answer within {budget_ms} ms"
211                        ))
212                    })??;
213            }
214            None => {
215                storage.write_batch(&writes).await?;
216            }
217        }
218        let mut state = self.0.borrow_mut();
219        state.writes.drain(..writes.len());
220        Ok(())
221    }
222}
223
224thread_local! {
225    static CURRENT: RefCell<Option<PageBuffer>> = const { RefCell::new(None) };
226}
227
228struct Restore(Option<PageBuffer>);
229impl Drop for Restore {
230    fn drop(&mut self) {
231        CURRENT.with(|c| *c.borrow_mut() = self.0.take());
232    }
233}
234
235/// `future` with `buffer` installed while it is polled.
236pub struct Scoped<F> {
237    buffer: PageBuffer,
238    future: Pin<Box<F>>,
239}
240
241pub fn scope<F: Future>(buffer: PageBuffer, future: F) -> Scoped<F> {
242    Scoped {
243        buffer,
244        future: Box::pin(future),
245    }
246}
247
248impl<F: Future> Future for Scoped<F> {
249    type Output = F::Output;
250    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
251        let _restore = Restore(CURRENT.with(|c| c.borrow_mut().replace(self.buffer.clone())));
252        self.future.as_mut().poll(cx)
253    }
254}
255
256/// The installed buffer, whether or not a page is open.
257pub fn installed() -> Option<PageBuffer> {
258    CURRENT.with(|c| c.borrow().clone())
259}
260
261/// The installed buffer while a page is open.
262pub fn current() -> Option<PageBuffer> {
263    installed().filter(PageBuffer::is_open)
264}
265
266/// Queue a backend's own statement in the open page, or hand it back when no
267/// page is open (the caller then runs it as before).
268pub fn defer(statement: Rc<dyn Any>) -> Result<(), Rc<dyn Any>> {
269    match current() {
270        Some(page) => {
271            page.0
272                .borrow_mut()
273                .writes
274                .push(PageWrite::Backend(statement));
275            Ok(())
276        }
277        None => Err(statement),
278    }
279}
280
281/// Read, for the open page, what its admissions will ask: which of `inputs`
282/// (`(txid, vout)`) are outputs of `topic`, and which of `txids` are applied
283/// in it. Two reads for the page; an answer already known in the page (its
284/// own write included) is kept. A read that faults prefetches nothing: each
285/// question is then asked as before.
286pub async fn prefetch(
287    storage: &dyn Storage,
288    topic: &str,
289    inputs: &[(String, u32)],
290    txids: &[String],
291) {
292    let Some(page) = current() else {
293        return;
294    };
295    let inputs: Vec<_> = {
296        let mut seen = HashSet::new();
297        inputs
298            .iter()
299            .filter(|(txid, vout)| page.output(txid, *vout, topic).is_none())
300            .filter(|key| seen.insert((*key).clone()))
301            .cloned()
302            .collect()
303    };
304    if !inputs.is_empty() {
305        if let Ok(held) = storage.find_topic_outputs(topic, &inputs).await {
306            let held: HashMap<_, _> = held
307                .into_iter()
308                .map(|o| ((o.txid.clone(), o.output_index), o))
309                .collect();
310            for (txid, vout) in &inputs {
311                let answer = held.get(&(txid.clone(), *vout)).cloned();
312                page.remember_output(txid, *vout, topic, answer);
313            }
314        }
315    }
316    let txids: Vec<_> = {
317        let mut seen = HashSet::new();
318        txids
319            .iter()
320            .filter(|txid| {
321                page.applied(&AppliedTransaction {
322                    txid: (*txid).clone(),
323                    topic: topic.to_string(),
324                })
325                .is_none()
326            })
327            .filter(|txid| seen.insert((*txid).clone()))
328            .cloned()
329            .collect()
330    };
331    if !txids.is_empty() {
332        if let Ok(applied) = storage.find_applied_transactions(topic, &txids).await {
333            let applied: HashSet<_> = applied.into_iter().collect();
334            for txid in txids {
335                let held = applied.contains(&txid);
336                page.remember_applied(
337                    &AppliedTransaction {
338                        txid,
339                        topic: topic.to_string(),
340                    },
341                    held,
342                );
343            }
344        }
345    }
346}
bsv_overlay_engine/page_batch.rs

bsv_overlay_engine/
page_batch.rs
