reth_storage_api/
bal.rs

1use alloc::{sync::Arc, vec::Vec};
2use alloy_eip7928::bal::DecodedBal;
3pub use alloy_eip7928::bal::RawBal;
4use alloy_eips::NumHash;
5use alloy_primitives::{BlockHash, BlockNumber, Bytes};
6use reth_storage_errors::provider::ProviderResult;
7
8/// Notification emitted when a new BAL is inserted into the store.
9#[derive(Clone, Debug, PartialEq, Eq)]
10pub struct BalNotification {
11    /// Number and hash of the block the BAL belongs to.
12    pub num_hash: NumHash,
13    /// Raw BAL RLP payload.
14    pub bal: RawBal,
15}
16
17impl BalNotification {
18    /// Creates a new [`BalNotification`].
19    pub const fn new(num_hash: NumHash, bal: RawBal) -> Self {
20        Self { num_hash, bal }
21    }
22}
23
24#[cfg(feature = "std")]
25pub use self::subscriptions::BalNotificationStream;
26
27#[cfg(feature = "std")]
28mod subscriptions {
29    use super::BalNotification;
30
31    /// A stream of [`BalNotification`]s.
32    pub type BalNotificationStream = reth_tokio_util::EventStream<BalNotification>;
33}
34
35/// Store for Block Access Lists (BALs).
36///
37/// This abstraction intentionally does not prescribe where BALs live. Implementations may keep
38/// recent BALs in memory, read canonical BALs from static files, or compose multiple tiers behind
39/// a single interface.
40#[auto_impl::auto_impl(&, Arc, Box)]
41pub trait BalStore: Send + Sync + 'static {
42    /// Insert the BAL for the given block.
43    ///
44    /// Implementations may buffer inserts. Call [`Self::flush`] when pending BALs need to be made
45    /// durable.
46    fn insert(&self, num_hash: NumHash, bal: RawBal) -> ProviderResult<()>;
47
48    /// Insert multiple BALs.
49    ///
50    /// The default implementation preserves the behavior of repeated [`Self::insert`] calls.
51    fn insert_many(&self, entries: Vec<(NumHash, RawBal)>) -> ProviderResult<()> {
52        for (num_hash, bal) in entries {
53            self.insert(num_hash, bal)?;
54        }
55        Ok(())
56    }
57
58    /// Flushes pending BALs for the given canonical blocks to the backing store.
59    ///
60    /// In-memory implementations may treat this as a no-op.
61    fn flush(&self, _blocks: &[NumHash]) -> ProviderResult<()> {
62        Ok(())
63    }
64
65    /// Prunes expired BALs according to the store's retention policy and the given chain tip.
66    ///
67    /// Returns the number of BALs pruned.
68    fn prune(&self, tip: BlockNumber) -> ProviderResult<usize>;
69
70    /// Fetch BALs for the given block hashes.
71    ///
72    /// The returned vector must align with `block_hashes`.
73    fn get_by_hashes(&self, block_hashes: &[BlockHash]) -> ProviderResult<Vec<Option<Bytes>>>;
74
75    /// Fetches the BAL for the given block hash.
76    fn get_by_hash(&self, block_hash: BlockHash) -> ProviderResult<Option<Bytes>> {
77        Ok(self.get_by_hashes(&[block_hash])?.into_iter().next().flatten())
78    }
79
80    /// Fetches and decodes the BAL for the given block hash.
81    fn get_decoded_by_hash(&self, block_hash: BlockHash) -> ProviderResult<Option<DecodedBal>> {
82        self.get_by_hash(block_hash)?
83            .map(DecodedBal::from_rlp_bytes)
84            .transpose()
85            .map_err(Into::into)
86    }
87
88    /// Fetch BAL response entries for the given block hashes, stopping after the soft limit is
89    /// exceeded.
90    ///
91    /// Entries are returned in request order. Unavailable BALs are represented as `None`. The
92    /// limit is soft: the entry that exceeds the limit is included.
93    fn get_by_hashes_with_limit(
94        &self,
95        block_hashes: &[BlockHash],
96        limit: GetBlockAccessListLimit,
97    ) -> ProviderResult<Vec<Option<Bytes>>> {
98        let mut out = Vec::new();
99        self.append_by_hashes_with_limit(block_hashes, limit, &mut out)?;
100        out.shrink_to_fit();
101        Ok(out)
102    }
103
104    /// Extends the given vector with BAL response entries for the given hashes.
105    ///
106    /// This adheres to the expected behavior of [`Self::get_by_hashes_with_limit`].
107    fn append_by_hashes_with_limit(
108        &self,
109        block_hashes: &[BlockHash],
110        limit: GetBlockAccessListLimit,
111        out: &mut Vec<Option<Bytes>>,
112    ) -> ProviderResult<()> {
113        let mut size = 0;
114        for bal in self.get_by_hashes(block_hashes)? {
115            size += bal.as_ref().map_or(1, |bytes| bytes.len());
116            out.push(bal);
117
118            if limit.exceeds(size) {
119                break
120            }
121        }
122        Ok(())
123    }
124
125    /// Returns a stream of BAL insert notifications.
126    ///
127    /// Notifications are emitted only after a BAL has been successfully inserted into the store.
128    /// They do not imply canonicality.
129    #[cfg(feature = "std")]
130    fn bal_stream(&self) -> BalNotificationStream;
131}
132
133/// The limit to enforce for [`BalStore::get_by_hashes_with_limit`].
134#[derive(Debug, Clone, Copy, Eq, PartialEq)]
135pub enum GetBlockAccessListLimit {
136    /// No limit, return all BALs.
137    None,
138    /// Enforce a size limit on the returned BALs, for example 2MB.
139    ResponseSizeSoftLimit(usize),
140}
141
142impl GetBlockAccessListLimit {
143    /// Returns true if the given size exceeds the limit.
144    #[inline]
145    pub const fn exceeds(&self, size: usize) -> bool {
146        match self {
147            Self::None => false,
148            Self::ResponseSizeSoftLimit(limit) => size > *limit,
149        }
150    }
151}
152
153/// Clone-friendly façade around a BAL store implementation.
154#[derive(Clone)]
155pub struct BalStoreHandle {
156    inner: Arc<dyn BalStore>,
157}
158
159impl BalStoreHandle {
160    /// Creates a new [`BalStoreHandle`] from the given implementation.
161    pub fn new(inner: impl BalStore) -> Self {
162        Self { inner: Arc::new(inner) }
163    }
164
165    /// Creates a [`BalStoreHandle`] backed by [`NoopBalStore`].
166    pub fn noop() -> Self {
167        Self::new(NoopBalStore)
168    }
169
170    /// Insert the BAL for the given block.
171    #[inline]
172    pub fn insert(&self, num_hash: NumHash, bal: RawBal) -> ProviderResult<()> {
173        self.inner.insert(num_hash, bal)
174    }
175
176    /// Insert multiple BALs.
177    #[inline]
178    pub fn insert_many(&self, entries: Vec<(NumHash, RawBal)>) -> ProviderResult<()> {
179        self.inner.insert_many(entries)
180    }
181
182    /// Flushes pending BALs for the given canonical blocks to the backing store.
183    #[inline]
184    pub fn flush(&self, blocks: &[NumHash]) -> ProviderResult<()> {
185        self.inner.flush(blocks)
186    }
187
188    /// Prunes expired BALs according to the store's retention policy and the given chain tip.
189    #[inline]
190    pub fn prune(&self, tip: BlockNumber) -> ProviderResult<usize> {
191        self.inner.prune(tip)
192    }
193
194    /// Fetch BALs for the given block hashes.
195    #[inline]
196    pub fn get_by_hashes(&self, block_hashes: &[BlockHash]) -> ProviderResult<Vec<Option<Bytes>>> {
197        self.inner.get_by_hashes(block_hashes)
198    }
199
200    /// Fetches the BAL for the given block hash.
201    #[inline]
202    pub fn get_by_hash(&self, block_hash: BlockHash) -> ProviderResult<Option<Bytes>> {
203        self.inner.get_by_hash(block_hash)
204    }
205
206    /// Fetches and decodes the BAL for the given block hash.
207    #[inline]
208    pub fn get_decoded_by_hash(&self, block_hash: BlockHash) -> ProviderResult<Option<DecodedBal>> {
209        self.inner.get_decoded_by_hash(block_hash)
210    }
211
212    /// Fetch BAL response entries for the given block hashes, stopping after the soft limit is
213    /// exceeded.
214    #[inline]
215    pub fn get_by_hashes_with_limit(
216        &self,
217        block_hashes: &[BlockHash],
218        limit: GetBlockAccessListLimit,
219    ) -> ProviderResult<Vec<Option<Bytes>>> {
220        self.inner.get_by_hashes_with_limit(block_hashes, limit)
221    }
222
223    /// Extends the given vector with BAL response entries for the given hashes.
224    #[inline]
225    pub fn append_by_hashes_with_limit(
226        &self,
227        block_hashes: &[BlockHash],
228        limit: GetBlockAccessListLimit,
229        out: &mut Vec<Option<Bytes>>,
230    ) -> ProviderResult<()> {
231        self.inner.append_by_hashes_with_limit(block_hashes, limit, out)
232    }
233
234    /// Returns a stream of BAL insert notifications.
235    #[cfg(feature = "std")]
236    #[inline]
237    pub fn bal_stream(&self) -> BalNotificationStream {
238        self.inner.bal_stream()
239    }
240}
241
242impl Default for BalStoreHandle {
243    fn default() -> Self {
244        Self::noop()
245    }
246}
247
248impl core::fmt::Debug for BalStoreHandle {
249    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
250        f.debug_struct("BalStoreHandle").finish_non_exhaustive()
251    }
252}
253
254/// Provider-side access to BAL storage.
255#[auto_impl::auto_impl(&, Arc)]
256pub trait BalProvider {
257    /// Returns the configured BAL store handle.
258    fn bal_store(&self) -> &BalStoreHandle;
259
260    /// Fetches the BAL for the given block hash.
261    fn get_bal_by_hash(&self, block_hash: BlockHash) -> ProviderResult<Option<Bytes>> {
262        self.bal_store().get_by_hash(block_hash)
263    }
264
265    /// Fetches BALs for the given block hashes.
266    fn get_bals_by_hashes(&self, block_hashes: &[BlockHash]) -> ProviderResult<Vec<Option<Bytes>>> {
267        self.bal_store().get_by_hashes(block_hashes)
268    }
269
270    /// Fetches BAL response entries for the given block hashes, stopping after the soft limit is
271    /// exceeded.
272    fn get_bals_by_hashes_with_limit(
273        &self,
274        block_hashes: &[BlockHash],
275        limit: GetBlockAccessListLimit,
276    ) -> ProviderResult<Vec<Option<Bytes>>> {
277        self.bal_store().get_by_hashes_with_limit(block_hashes, limit)
278    }
279}
280
281/// No-op BAL store used as the default wiring target until a concrete implementation is injected.
282#[derive(Debug, Default, Clone, Copy)]
283pub struct NoopBalStore;
284
285impl BalStore for NoopBalStore {
286    fn insert(&self, _num_hash: NumHash, _bal: RawBal) -> ProviderResult<()> {
287        Ok(())
288    }
289
290    fn insert_many(&self, _entries: Vec<(NumHash, RawBal)>) -> ProviderResult<()> {
291        Ok(())
292    }
293
294    fn prune(&self, _tip: BlockNumber) -> ProviderResult<usize> {
295        Ok(0)
296    }
297
298    fn get_by_hashes(&self, block_hashes: &[BlockHash]) -> ProviderResult<Vec<Option<Bytes>>> {
299        Ok(block_hashes.iter().map(|_| None).collect())
300    }
301
302    fn append_by_hashes_with_limit(
303        &self,
304        block_hashes: &[BlockHash],
305        limit: GetBlockAccessListLimit,
306        out: &mut Vec<Option<Bytes>>,
307    ) -> ProviderResult<()> {
308        let mut size = 0;
309        for _ in block_hashes {
310            size += 1;
311            out.push(None);
312
313            if limit.exceeds(size) {
314                break
315            }
316        }
317        Ok(())
318    }
319
320    #[cfg(feature = "std")]
321    fn bal_stream(&self) -> BalNotificationStream {
322        reth_tokio_util::EventSender::new(1).new_listener()
323    }
324}
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329    use alloy_primitives::B256;
330    #[cfg(feature = "std")]
331    use tokio_stream::StreamExt;
332
333    const EMPTY_LIST_CODE: u8 = 0xc0;
334
335    #[test]
336    fn noop_store_returns_empty_results() {
337        let store = BalStoreHandle::default();
338        let hashes = [B256::random(), B256::random()];
339
340        let by_hash = store.get_by_hashes(&hashes).unwrap();
341
342        assert_eq!(by_hash, vec![None, None]);
343        assert!(store.get_by_hash(B256::random()).unwrap().is_none());
344        assert_eq!(store.prune(10).unwrap(), 0);
345    }
346
347    #[test]
348    fn noop_provider_returns_empty_results() {
349        let provider = crate::noop::NoopProvider::default();
350        let hashes = [B256::random(), B256::random()];
351
352        assert_eq!(provider.get_bals_by_hashes(&hashes).unwrap(), vec![None, None]);
353        assert_eq!(
354            provider
355                .get_bals_by_hashes_with_limit(
356                    &hashes,
357                    GetBlockAccessListLimit::ResponseSizeSoftLimit(0),
358                )
359                .unwrap(),
360            vec![None]
361        );
362        assert!(provider.get_bal_by_hash(B256::random()).unwrap().is_none());
363    }
364
365    #[test]
366    fn noop_store_flush_is_noop() {
367        let store = BalStoreHandle::default();
368
369        store.flush(&[]).unwrap();
370    }
371
372    #[test]
373    fn noop_store_decoded_lookup_returns_none() {
374        let store = BalStoreHandle::default();
375
376        assert!(store.get_decoded_by_hash(B256::random()).unwrap().is_none());
377    }
378
379    #[test]
380    fn decoded_lookup_decodes_raw_bal() {
381        let hash = B256::random();
382        let raw_bal = Bytes::from_static(&[EMPTY_LIST_CODE]);
383        let store = BalStoreHandle::new(TestBalStore { hash, raw_bal: raw_bal.clone() });
384
385        assert_eq!(store.get_by_hash(hash).unwrap(), Some(raw_bal.clone()));
386
387        let decoded = store.get_decoded_by_hash(hash).unwrap().unwrap();
388
389        assert_eq!(decoded.as_raw(), &raw_bal);
390    }
391
392    #[test]
393    fn noop_store_limited_lookup_returns_prefix() {
394        let store = BalStoreHandle::default();
395        let hashes = [B256::random(), B256::random(), B256::random()];
396
397        let limited = store
398            .get_by_hashes_with_limit(&hashes, GetBlockAccessListLimit::ResponseSizeSoftLimit(1))
399            .unwrap();
400
401        assert_eq!(limited, vec![None, None]);
402    }
403
404    #[test]
405    fn block_access_list_limit() {
406        let limit_none = GetBlockAccessListLimit::None;
407        assert!(!limit_none.exceeds(usize::MAX));
408
409        let size_limit_2mb = GetBlockAccessListLimit::ResponseSizeSoftLimit(2 * 1024 * 1024);
410        assert!(!size_limit_2mb.exceeds(1024 * 1024));
411        assert!(!size_limit_2mb.exceeds(2 * 1024 * 1024));
412        assert!(size_limit_2mb.exceeds(3 * 1024 * 1024));
413    }
414
415    #[cfg(feature = "std")]
416    #[tokio::test]
417    async fn noop_store_stream_is_empty() {
418        let store = BalStoreHandle::default();
419        let mut stream = store.bal_stream();
420
421        assert!(stream.next().await.is_none());
422    }
423
424    #[derive(Debug)]
425    struct TestBalStore {
426        hash: B256,
427        raw_bal: Bytes,
428    }
429
430    impl BalStore for TestBalStore {
431        fn insert(&self, _num_hash: NumHash, _bal: RawBal) -> ProviderResult<()> {
432            Ok(())
433        }
434
435        fn prune(&self, _tip: BlockNumber) -> ProviderResult<usize> {
436            Ok(0)
437        }
438
439        fn get_by_hashes(&self, block_hashes: &[BlockHash]) -> ProviderResult<Vec<Option<Bytes>>> {
440            Ok(block_hashes
441                .iter()
442                .map(|hash| (*hash == self.hash).then(|| self.raw_bal.clone()))
443                .collect())
444        }
445
446        #[cfg(feature = "std")]
447        fn bal_stream(&self) -> BalNotificationStream {
448            reth_tokio_util::EventSender::new(1).new_listener()
449        }
450    }
451}
reth_storage_api/bal.rs

reth_storage_api/
bal.rs
