reth_provider/bal/
rocksdb.rs

1use crate::providers::RocksDBProvider;
2use alloy_eip7928::BAL_RETENTION_PERIOD_SLOTS;
3use alloy_eips::NumHash;
4use alloy_primitives::{BlockHash, BlockNumber, Bytes};
5use parking_lot::RwLock;
6use reth_db_api::{
7    models::{StoredBlockAccessList, StoredBlockAccessListKey},
8    table::{Decode, Decompress},
9    tables, DatabaseError,
10};
11use reth_prune_types::PruneMode;
12use reth_storage_api::{
13    BalNotification, BalNotificationStream, BalStore, GetBlockAccessListLimit, RawBal,
14};
15use reth_storage_errors::provider::{ProviderError, ProviderResult};
16use reth_tokio_util::EventSender;
17use std::{
18    collections::{BTreeMap, BTreeSet, HashMap},
19    sync::Arc,
20};
21
22/// Number of recent blocks kept in the in-memory `RocksDB` BAL buffer.
23const DEFAULT_BAL_BUFFER_RETENTION_DISTANCE: u64 = 32;
24
25/// RocksDB-backed BAL store.
26///
27/// Persisted BALs are keyed by `(block_number, block_hash)` for ordered pruning and indexed by
28/// block hash for direct [`BalStore`] lookups. Validated BALs enter a shared in-memory buffer
29/// first; [`BalStore::flush`] makes confirmed canonical entries durable while retaining them in the
30/// read cache until its shorter retention window expires.
31#[derive(Clone)]
32pub struct RocksDBBalStore {
33    /// Number of recent blocks retained in the in-memory buffer.
34    buffer_retention_distance: u64,
35    /// `RocksDB` provider used for persisted BAL reads and writes.
36    rocksdb: RocksDBProvider,
37    /// Shared recent-read cache and pending-write state.
38    buffer: Arc<RwLock<RocksDBBalStoreBuffer>>,
39    /// Broadcasts BAL insert notifications.
40    notifications: EventSender<BalNotification>,
41}
42
43impl RocksDBBalStore {
44    /// Creates a new store with the EIP-defined retention distance.
45    pub fn new(rocksdb: RocksDBProvider) -> Self {
46        Self::with_buffer_retention_distance(rocksdb, DEFAULT_BAL_BUFFER_RETENTION_DISTANCE)
47    }
48
49    /// Creates a new store that retains buffered BALs for the given block distance.
50    ///
51    /// This does not change the EIP-defined retention distance for persisted BALs.
52    pub fn with_buffer_retention_distance(rocksdb: RocksDBProvider, blocks: u64) -> Self {
53        Self {
54            buffer_retention_distance: blocks,
55            rocksdb,
56            buffer: Arc::new(RwLock::new(RocksDBBalStoreBuffer::default())),
57            notifications: EventSender::new(super::DEFAULT_BAL_NOTIFICATION_CHANNEL_SIZE),
58        }
59    }
60
61    #[cfg(test)]
62    const fn rocksdb_provider(&self) -> &RocksDBProvider {
63        &self.rocksdb
64    }
65
66    fn keys_to_prune(&self, tip: BlockNumber) -> ProviderResult<Vec<StoredBlockAccessListKey>> {
67        let retention = PruneMode::Distance(BAL_RETENTION_PERIOD_SLOTS);
68        let mut keys = Vec::new();
69        let iter = self.rocksdb.raw_key_iter_from::<tables::BlockAccessLists>(
70            StoredBlockAccessListKey::first_at_number(0),
71        )?;
72
73        for key_bytes in iter {
74            let key_bytes = key_bytes?;
75            let key = StoredBlockAccessListKey::decode(&key_bytes)
76                .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
77            if !retention.should_prune(key.number(), tip) {
78                break
79            }
80            keys.push(key);
81        }
82
83        Ok(keys)
84    }
85
86    fn delete_keys(&self, keys: &[StoredBlockAccessListKey]) -> ProviderResult<usize> {
87        if keys.is_empty() {
88            return Ok(0)
89        }
90
91        let mut batch = self.rocksdb.batch();
92        for key in keys {
93            batch.delete::<tables::BlockAccessLists>(*key)?;
94            batch.delete::<tables::BlockAccessListBlockNumbers>(key.hash())?;
95        }
96        batch.commit()?;
97        Ok(keys.len())
98    }
99
100    fn read_one_from_disk(&self, key: StoredBlockAccessListKey) -> ProviderResult<Option<Bytes>> {
101        let Some(value) = self.rocksdb.get_raw::<tables::BlockAccessLists>(key)? else {
102            return Ok(None)
103        };
104        let stored = StoredBlockAccessList::decompress(&value)
105            .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
106        Ok(Some(stored.into_raw()))
107    }
108
109    fn read_one_by_hash(&self, block_hash: BlockHash) -> ProviderResult<Option<Bytes>> {
110        if let Some(bal) = self.buffer.read().get_by_hash(block_hash) {
111            return Ok(Some(bal))
112        }
113
114        let Some(block_number) =
115            self.rocksdb.get::<tables::BlockAccessListBlockNumbers>(block_hash)?
116        else {
117            return Ok(None)
118        };
119        self.read_one_from_disk(StoredBlockAccessListKey::new(block_number, block_hash))
120    }
121}
122
123impl std::fmt::Debug for RocksDBBalStore {
124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        f.debug_struct("RocksDBBalStore")
126            .field("buffer_retention_distance", &self.buffer_retention_distance)
127            .field("rocksdb", &self.rocksdb)
128            .finish_non_exhaustive()
129    }
130}
131
132impl BalStore for RocksDBBalStore {
133    fn insert(&self, block: NumHash, bal: RawBal) -> ProviderResult<()> {
134        let mut buffer = self.buffer.write();
135        buffer.insert(block, bal.clone());
136        drop(buffer);
137
138        self.notifications.notify(BalNotification::new(block, bal));
139        Ok(())
140    }
141
142    fn insert_many(&self, entries: Vec<(NumHash, RawBal)>) -> ProviderResult<()> {
143        if entries.is_empty() {
144            return Ok(())
145        }
146
147        let mut buffer = self.buffer.write();
148        buffer.entries.reserve(entries.len());
149        for (block, bal) in &entries {
150            buffer.insert(*block, bal.clone());
151        }
152        drop(buffer);
153
154        for (block, bal) in entries {
155            self.notifications.notify(BalNotification::new(block, bal));
156        }
157        Ok(())
158    }
159
160    fn flush(&self, blocks: &[NumHash]) -> ProviderResult<()> {
161        let pending = {
162            let mut buffer = self.buffer.write();
163            buffer.mark_canonical(blocks);
164            buffer.canonical_pending_entries()
165        };
166        if !pending.is_empty() {
167            let mut batch = self.rocksdb.batch();
168            for (key, bal) in &pending {
169                let value = StoredBlockAccessList::new_unchecked(bal.hash(), bal.as_raw().clone());
170                batch.put::<tables::BlockAccessLists>(*key, &value)?;
171                batch.put::<tables::BlockAccessListBlockNumbers>(key.hash(), &key.number())?;
172            }
173            batch.commit()?;
174
175            self.buffer.write().remove_flushed_pending(&pending);
176        }
177
178        if let Some(tip) = blocks.iter().map(|block| block.number).max() {
179            self.buffer.write().prune(self.buffer_retention_distance, tip);
180        }
181        Ok(())
182    }
183
184    fn prune(&self, tip: BlockNumber) -> ProviderResult<usize> {
185        let keys = self.keys_to_prune(tip)?;
186        let pruned = self.delete_keys(&keys)?;
187        self.buffer.write().remove_keys(&keys);
188        Ok(pruned)
189    }
190
191    fn get_by_hashes(&self, block_hashes: &[BlockHash]) -> ProviderResult<Vec<Option<Bytes>>> {
192        block_hashes.iter().map(|hash| self.read_one_by_hash(*hash)).collect()
193    }
194
195    fn append_by_hashes_with_limit(
196        &self,
197        block_hashes: &[BlockHash],
198        limit: GetBlockAccessListLimit,
199        out: &mut Vec<Option<Bytes>>,
200    ) -> ProviderResult<()> {
201        let mut size = 0;
202        for block_hash in block_hashes {
203            let bal = self.read_one_by_hash(*block_hash)?;
204            size += bal.as_ref().map_or(1, |bytes| bytes.len());
205            out.push(bal);
206
207            if limit.exceeds(size) {
208                break
209            }
210        }
211        Ok(())
212    }
213
214    fn bal_stream(&self) -> BalNotificationStream {
215        self.notifications.new_listener()
216    }
217}
218
219/// Shared in-memory state for recent reads and writes awaiting canonical confirmation.
220///
221/// Successful flushes clear only the pending-write state. Cached entries remain available until
222/// the buffer retention window evicts them.
223#[derive(Debug, Default)]
224struct RocksDBBalStoreBuffer {
225    /// Hash index for serving recent hash-only lookups.
226    entries: HashMap<BlockHash, RocksDBBalEntry>,
227    /// Block-number index for pruning buffered entries.
228    hashes_by_number: BTreeMap<BlockNumber, Vec<BlockHash>>,
229    /// Validated BALs waiting to be confirmed canonical and flushed.
230    pending: BTreeMap<StoredBlockAccessListKey, RawBal>,
231    /// Pending writes confirmed canonical, including writes retained for retry after failure.
232    canonical_pending: BTreeSet<StoredBlockAccessListKey>,
233}
234
235impl RocksDBBalStoreBuffer {
236    fn insert(&mut self, block: NumHash, bal: RawBal) {
237        let pending = bal.clone();
238        if let Some(entry) =
239            self.entries.insert(block.hash, RocksDBBalEntry { block_number: block.number, bal })
240        {
241            self.remove_hash_from_number(entry.block_number, block.hash);
242            self.pending.remove(&StoredBlockAccessListKey::new(entry.block_number, block.hash));
243            self.canonical_pending
244                .remove(&StoredBlockAccessListKey::new(entry.block_number, block.hash));
245        }
246
247        self.hashes_by_number.entry(block.number).or_default().push(block.hash);
248        self.pending.insert(StoredBlockAccessListKey::new(block.number, block.hash), pending);
249    }
250
251    /// Marks exact pending block identities as eligible for the next flush.
252    fn mark_canonical(&mut self, blocks: &[NumHash]) {
253        self.canonical_pending.extend(
254            blocks
255                .iter()
256                .map(|block| StoredBlockAccessListKey::new(block.number, block.hash))
257                .filter(|key| self.pending.contains_key(key)),
258        );
259    }
260
261    /// Snapshots confirmed writes so `RocksDB` I/O can run without holding the buffer lock.
262    fn canonical_pending_entries(&self) -> Vec<(StoredBlockAccessListKey, RawBal)> {
263        self.canonical_pending
264            .iter()
265            .filter_map(|key| self.pending.get(key).map(|bal| (*key, bal.clone())))
266            .collect()
267    }
268
269    fn keys_to_prune(
270        &self,
271        retention_distance: u64,
272        tip: BlockNumber,
273    ) -> Vec<StoredBlockAccessListKey> {
274        let prune_mode = PruneMode::Distance(retention_distance);
275        self.hashes_by_number
276            .iter()
277            .take_while(|(block_number, _)| prune_mode.should_prune(**block_number, tip))
278            .flat_map(|(block_number, hashes)| {
279                hashes.iter().map(move |hash| StoredBlockAccessListKey::new(*block_number, *hash))
280            })
281            .filter(|key| !self.canonical_pending.contains(key))
282            .collect()
283    }
284
285    fn get_by_hash(&self, hash: BlockHash) -> Option<Bytes> {
286        self.entries.get(&hash).map(|entry| entry.bal.as_raw().clone())
287    }
288
289    /// Clears pending state only if it still matches the snapshot written to `RocksDB`.
290    ///
291    /// A concurrent replacement for the same key must remain pending for a later flush.
292    fn remove_flushed_pending(&mut self, flushed: &[(StoredBlockAccessListKey, RawBal)]) {
293        for (key, bal) in flushed {
294            let pending_matches =
295                self.pending.get(key).is_some_and(|pending| pending.as_raw() == bal.as_raw());
296            if pending_matches {
297                self.pending.remove(key);
298                self.canonical_pending.remove(key);
299            }
300        }
301    }
302
303    fn prune(&mut self, retention_distance: u64, tip: BlockNumber) {
304        let keys = self.keys_to_prune(retention_distance, tip);
305        self.remove_keys(&keys);
306    }
307
308    fn remove_keys(&mut self, keys: &[StoredBlockAccessListKey]) -> usize {
309        let mut removed = 0;
310        for key in keys {
311            let block = NumHash::new(key.number(), key.hash());
312            let pending_removed = self.pending.remove(key).is_some();
313            self.canonical_pending.remove(key);
314            let entry_removed = if self
315                .entries
316                .get(&block.hash)
317                .is_some_and(|entry| entry.block_number == block.number)
318            {
319                self.entries.remove(&block.hash).is_some()
320            } else {
321                false
322            };
323
324            if entry_removed {
325                self.remove_hash_from_number(block.number, block.hash);
326            }
327            removed += usize::from(pending_removed || entry_removed);
328        }
329        removed
330    }
331
332    fn remove_hash_from_number(&mut self, block_number: BlockNumber, block_hash: BlockHash) {
333        let empty = self.hashes_by_number.get_mut(&block_number).is_some_and(|hashes| {
334            hashes.retain(|hash| *hash != block_hash);
335            hashes.is_empty()
336        });
337        if empty {
338            self.hashes_by_number.remove(&block_number);
339        }
340    }
341}
342
343/// Buffered BAL entry with its block number.
344#[derive(Debug)]
345struct RocksDBBalEntry {
346    /// Block number for this hash-indexed BAL.
347    block_number: BlockNumber,
348    /// Raw BAL payload.
349    bal: RawBal,
350}
351
352#[cfg(test)]
353mod tests {
354    use super::*;
355    use crate::providers::RocksDBBuilder;
356    use alloy_primitives::B256;
357    use tokio_stream::StreamExt;
358
359    fn test_store() -> (tempfile::TempDir, RocksDBBalStore) {
360        let dir = tempfile::tempdir().unwrap();
361        let rocksdb = RocksDBBuilder::new(dir.path()).with_default_tables().build().unwrap();
362        (dir, RocksDBBalStore::new(rocksdb))
363    }
364
365    fn disk_bal(store: &RocksDBBalStore, block: NumHash) -> Option<Bytes> {
366        store
367            .rocksdb_provider()
368            .get_raw::<tables::BlockAccessLists>(StoredBlockAccessListKey::new(
369                block.number,
370                block.hash,
371            ))
372            .unwrap()
373            .map(|value| StoredBlockAccessList::decompress(&value).unwrap().into_raw())
374    }
375
376    fn read_many(store: &RocksDBBalStore, blocks: &[NumHash]) -> Vec<Option<Bytes>> {
377        blocks.iter().map(|block| store.get_by_hash(block.hash).unwrap()).collect()
378    }
379
380    #[test]
381    fn inserts_and_reads_by_hash() {
382        let (_dir, store) = test_store();
383        let hash = B256::random();
384        let missing = NumHash::new(1, B256::random());
385        let bal = Bytes::from_static(&[0xc1, 0x01]);
386
387        store.insert(NumHash::new(1, hash), RawBal::from(bal.clone())).unwrap();
388
389        assert_eq!(read_many(&store, &[NumHash::new(1, hash), missing]), vec![Some(bal), None]);
390    }
391
392    #[test]
393    fn hash_lookup_reads_persisted_bal_through_store() {
394        let (_dir, store) = test_store();
395        let block = NumHash::new(1, B256::random());
396        let bal = Bytes::from_static(&[0xc1, 0x01]);
397
398        store.insert(block, RawBal::from(bal.clone())).unwrap();
399        store.flush(&[block]).unwrap();
400
401        let store_with_empty_buffer = RocksDBBalStore::new(store.rocksdb_provider().clone());
402        assert_eq!(store_with_empty_buffer.get_by_hash(block.hash).unwrap(), Some(bal));
403    }
404
405    #[test]
406    fn flush_prunes_buffer_retention() {
407        let (_dir, store) = test_store();
408        let old = NumHash::new(1, B256::with_last_byte(1));
409        let retained =
410            NumHash::new(DEFAULT_BAL_BUFFER_RETENTION_DISTANCE + 2, B256::with_last_byte(2));
411        let old_bal = Bytes::from_static(&[0xc1, 0x01]);
412        let retained_bal = Bytes::from_static(&[0xc1, 0x02]);
413
414        store.insert(old, RawBal::from(old_bal.clone())).unwrap();
415        store.insert(retained, RawBal::from(retained_bal.clone())).unwrap();
416
417        assert_eq!(
418            store.get_by_hashes(&[old.hash, retained.hash]).unwrap(),
419            vec![Some(old_bal), Some(retained_bal.clone())]
420        );
421
422        store.flush(&[retained]).unwrap();
423
424        assert_eq!(
425            store.get_by_hashes(&[old.hash, retained.hash]).unwrap(),
426            vec![None, Some(retained_bal)]
427        );
428        assert_eq!(disk_bal(&store, old), None);
429    }
430
431    #[test]
432    fn flush_prunes_only_durable_cache_entries() {
433        let (_dir, store) = test_store();
434        let old = NumHash::new(1, B256::with_last_byte(1));
435        let tip = NumHash::new(DEFAULT_BAL_BUFFER_RETENTION_DISTANCE + 2, B256::with_last_byte(2));
436        let old_bal = Bytes::from_static(&[0xc1, 0x01]);
437
438        store.insert(old, RawBal::from(old_bal.clone())).unwrap();
439        store.flush(&[old]).unwrap();
440        store.flush(&[tip]).unwrap();
441
442        assert!(!store.buffer.read().entries.contains_key(&old.hash));
443        assert_eq!(disk_bal(&store, old), Some(old_bal.clone()));
444        assert_eq!(store.get_by_hash(old.hash).unwrap(), Some(old_bal));
445    }
446
447    #[test]
448    fn configured_buffer_retention_distance_is_used() {
449        let dir = tempfile::tempdir().unwrap();
450        let rocksdb = RocksDBBuilder::new(dir.path()).with_default_tables().build().unwrap();
451        let store = RocksDBBalStore::with_buffer_retention_distance(rocksdb, 64);
452        let old = NumHash::new(1, B256::with_last_byte(1));
453        let tip = NumHash::new(34, B256::with_last_byte(2));
454
455        store.insert(old, RawBal::from(Bytes::from_static(&[0xc1, 0x01]))).unwrap();
456        store.flush(&[old]).unwrap();
457        store.flush(&[tip]).unwrap();
458
459        assert!(store.buffer.read().entries.contains_key(&old.hash));
460    }
461
462    #[test]
463    fn flush_writes_only_requested_pending_bals() {
464        let (_dir, store) = test_store();
465        let block_1 = NumHash::new(1, B256::with_last_byte(1));
466        let block_1_fork = NumHash::new(1, B256::with_last_byte(9));
467        let block_2 = NumHash::new(2, B256::with_last_byte(2));
468        let bal_1 = Bytes::from_static(&[0xc1, 0x01]);
469        let bal_1_fork = Bytes::from_static(&[0xc1, 0x09]);
470        let bal_2 = Bytes::from_static(&[0xc1, 0x02]);
471
472        store.insert(block_1, RawBal::from(bal_1.clone())).unwrap();
473        store.insert(block_1_fork, RawBal::from(bal_1_fork.clone())).unwrap();
474        store.insert(block_2, RawBal::from(bal_2.clone())).unwrap();
475
476        store.flush(&[block_1]).unwrap();
477
478        assert_eq!(disk_bal(&store, block_1), Some(bal_1.clone()));
479        assert_eq!(disk_bal(&store, block_1_fork), None);
480        assert_eq!(disk_bal(&store, block_2), None);
481        assert_eq!(
482            store.get_by_hashes(&[block_1.hash, block_1_fork.hash, block_2.hash]).unwrap(),
483            vec![Some(bal_1.clone()), Some(bal_1_fork), Some(bal_2)]
484        );
485
486        let store_with_empty_buffer = RocksDBBalStore::new(store.rocksdb_provider().clone());
487        assert_eq!(
488            store_with_empty_buffer
489                .get_by_hashes(&[block_1.hash, block_1_fork.hash, block_2.hash])
490                .unwrap(),
491            vec![Some(bal_1), None, None]
492        );
493    }
494
495    #[test]
496    fn sparse_numbers_are_valid() {
497        let (_dir, store) = test_store();
498        let hash_a = B256::with_last_byte(1);
499        let hash_b = B256::with_last_byte(2);
500        let bal_a = Bytes::from_static(&[0xc1, 0x01]);
501        let bal_b = Bytes::from_static(&[0xc1, 0x02]);
502
503        let block_a = NumHash::new(2, hash_a);
504        let block_b = NumHash::new(200, hash_b);
505
506        store.insert(block_a, RawBal::from(bal_a.clone())).unwrap();
507        store.flush(&[block_a]).unwrap();
508        store.insert(block_b, RawBal::from(bal_b.clone())).unwrap();
509        store.flush(&[block_b]).unwrap();
510
511        assert_eq!(read_many(&store, &[block_a, block_b]), vec![Some(bal_a), Some(bal_b)]);
512    }
513
514    #[test]
515    fn missing_and_empty_bal_are_distinct() {
516        let (_dir, store) = test_store();
517        let empty_hash = B256::with_last_byte(1);
518        let missing_hash = B256::with_last_byte(2);
519        let empty_bal = Bytes::from_static(&[0xc0]);
520
521        store.insert(NumHash::new(1, empty_hash), RawBal::from(empty_bal.clone())).unwrap();
522
523        assert_eq!(
524            read_many(&store, &[NumHash::new(1, empty_hash), NumHash::new(1, missing_hash)]),
525            vec![Some(empty_bal), None]
526        );
527    }
528
529    #[test]
530    fn prune_uses_eip_retention() {
531        let (_dir, store) = test_store();
532        let old_hash = B256::with_last_byte(1);
533        let retained_hash = B256::with_last_byte(2);
534        let retained_bal = Bytes::from_static(&[0xc1, 0x02]);
535        let tip = BAL_RETENTION_PERIOD_SLOTS + 2;
536
537        store
538            .insert(NumHash::new(1, old_hash), RawBal::from(Bytes::from_static(&[0xc1, 0x01])))
539            .unwrap();
540        store.insert(NumHash::new(2, retained_hash), RawBal::from(retained_bal.clone())).unwrap();
541        store.flush(&[NumHash::new(1, old_hash), NumHash::new(2, retained_hash)]).unwrap();
542
543        assert_eq!(store.prune(tip).unwrap(), 1);
544        assert_eq!(disk_bal(&store, NumHash::new(1, old_hash)), None);
545        assert_eq!(disk_bal(&store, NumHash::new(2, retained_hash)), Some(retained_bal.clone()));
546        assert_eq!(
547            read_many(&store, &[NumHash::new(1, old_hash), NumHash::new(2, retained_hash)]),
548            vec![None, Some(retained_bal)]
549        );
550    }
551
552    #[test]
553    fn stored_payload_hash_is_not_reverified() {
554        let (_dir, store) = test_store();
555        let block = NumHash::new(1, B256::with_last_byte(1));
556        let key = StoredBlockAccessListKey::new(block.number, block.hash);
557        let value = StoredBlockAccessList::new_unchecked(B256::ZERO, Bytes::from_static(&[0xc0]));
558
559        store.rocksdb_provider().put::<tables::BlockAccessLists>(key, &value).unwrap();
560        store
561            .rocksdb_provider()
562            .put::<tables::BlockAccessListBlockNumbers>(block.hash, &block.number)
563            .unwrap();
564
565        assert_eq!(store.get_by_hash(block.hash).unwrap(), Some(Bytes::from_static(&[0xc0])));
566    }
567
568    #[test]
569    fn canonical_pending_entries_are_retried() {
570        let mut buffer = RocksDBBalStoreBuffer::default();
571        let first = NumHash::new(1, B256::with_last_byte(1));
572        let second = NumHash::new(2, B256::with_last_byte(2));
573        buffer.insert(first, RawBal::from(Bytes::from_static(&[0xc1, 0x01])));
574        buffer.mark_canonical(&[first]);
575
576        assert_eq!(buffer.canonical_pending_entries().len(), 1);
577
578        // A failed flush leaves the canonical entry pending for the next attempt.
579        buffer.insert(second, RawBal::from(Bytes::from_static(&[0xc1, 0x02])));
580        buffer.mark_canonical(&[second]);
581        let retry = buffer.canonical_pending_entries();
582
583        assert_eq!(
584            retry.iter().map(|(key, _)| NumHash::new(key.number(), key.hash())).collect::<Vec<_>>(),
585            vec![first, second]
586        );
587    }
588
589    #[tokio::test]
590    async fn insert_notifies_subscribers() {
591        let (_dir, store) = test_store();
592        let mut stream = store.bal_stream();
593        let block = NumHash::new(1, B256::with_last_byte(1));
594        let bal = RawBal::from(Bytes::from_static(&[0xc0]));
595
596        store.insert(block, bal.clone()).unwrap();
597
598        assert_eq!(stream.next().await.unwrap(), BalNotification::new(block, bal));
599    }
600}
reth_provider/bal/rocksdb.rs

reth_provider/bal/
rocksdb.rs
