bsv_overlay_engine/stream_sizing.rs
1//! What a read of a BEEF through the streaming reader will hold, estimated
2//! from the frame's LENGTHS AND COUNTS before anything is allocated (bsv-low
3//! #585, the doors lens E585-D12-L3).
4//!
5//! bsv-rs 0.4.0's `BeefStream` hands out one element at a time, and an
6//! element is not its wire bytes: a BUMP leaf of 2 to 38 bytes is a 56 byte
7//! `Leaf` and, while the root is computed, an entry in four tables; a raw
8//! transaction is its bytes and a record per input and per output. A reader
9//! that keeps a table per element adds an entry per element. So the memory of
10//! a read follows the element COUNT and a BUMP's LEAVES, not the body's
11//! bytes: measured by the lens, natively, 77 MB for a 9.9 MB body of 900,000
12//! minimal transactions and 108 MB for a 9.8 MB body holding one BUMP of 2^18
13//! leaves, each beside the body in a 128 MB isolate.
14//!
15//! [`estimate`] walks the frame with no allocation and no hash (it reads the
16//! varints and steps over everything else), adds what the caller says each
17//! thing costs ([`StreamCharges`]), and stops at the first element that takes
18//! the estimate past the caller's limit. The caller then does not open the
19//! stream. It is a BUDGET, never a refusal: what a caller does with a breach
20//! is its own (the script door answers "the network judges").
21//!
22//! The estimate is an UPPER bound by construction, not a measurement: every
23//! growing buffer is charged three times its contents (a `Vec` that doubles
24//! holds the old and the new buffer while it moves), every hash table three
25//! times its buckets at the fullest load (7/8), and the sizes are the native
26//! ones (a `usize` is 8 bytes; wasm32's are smaller). The pins hold the
27//! measured peak under it for each shape.
28
29use bsv_rs::transaction::{ATOMIC_BEEF, BEEF_V1, BEEF_V2};
30
31use crate::script_door::Fields;
32
33/// What a reader holds per thing of the stream, in bytes.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub struct StreamCharges {
36 /// Kept for the whole read, per raw transaction.
37 pub kept_tx: u64,
38 /// Kept for the whole read, per input of a raw transaction.
39 pub kept_input: u64,
40 /// Kept for the whole read, per output of a raw transaction.
41 pub kept_output: u64,
42 /// Kept for the whole read, per txid a BUMP carries at level 0.
43 pub kept_proven: u64,
44 /// In hand, one element at a time: per byte of a raw transaction.
45 pub tx_byte: u64,
46 /// In hand: per input of the raw transaction.
47 pub tx_input: u64,
48 /// In hand: per output of the raw transaction.
49 pub tx_output: u64,
50 /// In hand: per leaf of the BUMP, at any level.
51 pub bump_leaf: u64,
52}
53
54impl StreamCharges {
55 /// The stream's own element and nothing kept: what `BeefStream` holds
56 /// while it builds and hands out one element.
57 ///
58 /// A raw transaction: its bytes in a growing buffer (3x), an `InputRef`
59 /// of 64 bytes an input and an `OutputRef` of 32 an output, each in a
60 /// growing buffer (192, 96). A BUMP leaf: the `Leaf` of 56 bytes in a
61 /// growing buffer (168) and, while `bump_root` runs, a path node of 48
62 /// twice and an entry in three tables of 49, 41 and 41 bytes a bucket at
63 /// up to 16/7 buckets an entry (96 + 112 + 94 + 94): 564, taken as 576.
64 pub const STREAM: StreamCharges = StreamCharges {
65 kept_tx: 0,
66 kept_input: 0,
67 kept_output: 0,
68 kept_proven: 0,
69 tx_byte: 3,
70 tx_input: 192,
71 tx_output: 96,
72 bump_leaf: 576,
73 };
74}
75
76/// The estimate of one read.
77#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78pub struct StreamEstimate {
79 /// What is kept and the heaviest element in hand, of everything read up
80 /// to where the estimate stopped.
81 pub bytes: u64,
82 /// The estimate passed the limit: the offset of the element that took it
83 /// there. Nothing after it was read.
84 pub over_at: Option<usize>,
85 /// The frame was followed to its end. `false` with no `over_at`: the
86 /// bytes are not a BEEF frame this reader follows (the stream refuses
87 /// them too), and `bytes` is of what came before.
88 pub read: bool,
89}
90
91/// What a reader of the stream does with the estimate, before it opens the
92/// stream. ONE rule for every reader (the script door and the census; the
93/// delta lens E585-D12-DELTA-N1, Rule 10).
94#[derive(Debug, Clone, Copy, PartialEq, Eq)]
95pub enum Admission {
96 /// The frame was followed to its end within the limit: open the stream.
97 Open,
98 /// The estimate passed the limit: do NOT open the stream.
99 Over,
100 /// The frame was not followed to its end, under the limit. The stream
101 /// refuses every such frame today (the pin
102 /// `e585f_l3_the_sizing_follows_every_frame_the_stream_reads`, re-run at
103 /// every bsv-rs bump), so the reader opens it and takes its refusal; a
104 /// read the stream DOES make of such bytes was not estimated, and the
105 /// reader answers it as over the limit, after the fact.
106 Unfollowed,
107}
108
109impl StreamEstimate {
110 /// The rule of [`Admission`].
111 #[must_use]
112 pub fn admission(&self) -> Admission {
113 if self.over_at.is_some() {
114 Admission::Over
115 } else if self.read {
116 Admission::Open
117 } else {
118 Admission::Unfollowed
119 }
120 }
121}
122
123/// Estimates what a read of `body` holds under `charges`, stopping at the
124/// first element that takes it past `limit`.
125#[must_use]
126pub fn estimate(body: &[u8], charges: &StreamCharges, limit: u64) -> StreamEstimate {
127 let mut sizer = Sizer {
128 f: Fields { raw: body, at: 0 },
129 charges,
130 limit,
131 kept: 0,
132 largest: 0,
133 over_at: None,
134 };
135 let followed = sizer.frame().is_some();
136 StreamEstimate {
137 bytes: sizer.kept.saturating_add(sizer.largest),
138 over_at: sizer.over_at,
139 read: followed,
140 }
141}
142
143struct Sizer<'a> {
144 f: Fields<'a>,
145 charges: &'a StreamCharges,
146 limit: u64,
147 kept: u64,
148 /// The heaviest element in hand so far.
149 largest: u64,
150 over_at: Option<usize>,
151}
152
153impl Sizer<'_> {
154 /// An element at `at` holds `bytes` in hand: `None` once that is past
155 /// the limit.
156 fn in_hand(&mut self, at: usize, bytes: u64) -> Option<()> {
157 self.largest = self.largest.max(bytes);
158 if self.kept.saturating_add(self.largest) > self.limit {
159 self.over_at = Some(at);
160 return None;
161 }
162 Some(())
163 }
164
165 fn frame(&mut self) -> Option<()> {
166 let mut word = self.f.u32()?;
167 if word == ATOMIC_BEEF {
168 self.f.take(32)?;
169 word = self.f.u32()?;
170 }
171 let v2 = match word {
172 BEEF_V1 => false,
173 BEEF_V2 => true,
174 _ => return None,
175 };
176 for _ in 0..self.f.varint()? {
177 self.bump()?;
178 }
179 for _ in 0..self.f.varint()? {
180 if v2 {
181 match *self.f.take(1)?.first()? {
182 0 => self.tx()?,
183 1 => {
184 self.f.varint()?;
185 self.tx()?;
186 }
187 2 => {
188 self.f.take(32)?;
189 }
190 _ => return None,
191 }
192 } else {
193 self.tx()?;
194 match *self.f.take(1)?.first()? {
195 0 => {}
196 1 => {
197 self.f.varint()?;
198 }
199 _ => return None,
200 }
201 }
202 }
203 Some(())
204 }
205
206 /// One BUMP, charged leaf by leaf: a BUMP cut short is stopped where a
207 /// whole one would be.
208 fn bump(&mut self) -> Option<()> {
209 let at = self.f.at;
210 self.f.varint()?;
211 let levels = *self.f.take(1)?.first()?;
212 let mut leaves = 0u64;
213 for level in 0..levels {
214 for _ in 0..self.f.varint()? {
215 self.f.varint()?;
216 let hashed = match *self.f.take(1)?.first()? {
217 1 => false,
218 0 | 2 => {
219 self.f.take(32)?;
220 true
221 }
222 _ => return None,
223 };
224 leaves += 1;
225 if level == 0 && hashed {
226 self.kept = self.kept.saturating_add(self.charges.kept_proven);
227 }
228 self.in_hand(at, leaves.saturating_mul(self.charges.bump_leaf))?;
229 }
230 }
231 Some(())
232 }
233
234 /// One raw transaction, charged input by input and output by output.
235 fn tx(&mut self) -> Option<()> {
236 let at = self.f.at;
237 self.f.take(4)?;
238 let (mut inputs, mut outputs) = (0u64, 0u64);
239 for _ in 0..self.f.varint()? {
240 self.f.take(36)?;
241 self.f.script()?;
242 self.f.take(4)?;
243 inputs += 1;
244 self.tx_in_hand(at, inputs, outputs)?;
245 }
246 for _ in 0..self.f.varint()? {
247 self.f.take(8)?;
248 self.f.script()?;
249 outputs += 1;
250 self.tx_in_hand(at, inputs, outputs)?;
251 }
252 self.f.take(4)?;
253 self.tx_in_hand(at, inputs, outputs)?;
254 self.kept = self
255 .kept
256 .saturating_add(self.charges.kept_tx)
257 .saturating_add(inputs.saturating_mul(self.charges.kept_input))
258 .saturating_add(outputs.saturating_mul(self.charges.kept_output));
259 self.in_hand(at, 0)
260 }
261
262 fn tx_in_hand(&mut self, at: usize, inputs: u64, outputs: u64) -> Option<()> {
263 let c = self.charges;
264 let bytes = ((self.f.at - at) as u64)
265 .saturating_mul(c.tx_byte)
266 .saturating_add(inputs.saturating_mul(c.tx_input))
267 .saturating_add(outputs.saturating_mul(c.tx_output));
268 self.in_hand(at, bytes)
269 }
270}