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, RocksDBProvider};
356    use alloy_primitives::B256;
357    use tokio_stream::StreamExt;
358
359    fn test_rocksdb(dir: &tempfile::TempDir) -> RocksDBProvider {
360        RocksDBBuilder::new(dir.path())
361            .with_table::<tables::BlockAccessLists>()
362            .with_table::<tables::BlockAccessListBlockNumbers>()
363            .build()
364            .unwrap()
365    }
366
367    fn test_store() -> (tempfile::TempDir, RocksDBBalStore) {
368        let dir = tempfile::tempdir().unwrap();
369        let rocksdb = test_rocksdb(&dir);
370        (dir, RocksDBBalStore::new(rocksdb))
371    }
372
373    fn disk_bal(store: &RocksDBBalStore, block: NumHash) -> Option<Bytes> {
374        store
375            .rocksdb_provider()
376            .get_raw::<tables::BlockAccessLists>(StoredBlockAccessListKey::new(
377                block.number,
378                block.hash,
379            ))
380            .unwrap()
381            .map(|value| StoredBlockAccessList::decompress(&value).unwrap().into_raw())
382    }
383
384    fn read_many(store: &RocksDBBalStore, blocks: &[NumHash]) -> Vec<Option<Bytes>> {
385        blocks.iter().map(|block| store.get_by_hash(block.hash).unwrap()).collect()
386    }
387
388    #[test]
389    fn inserts_and_reads_by_hash() {
390        let (_dir, store) = test_store();
391        let hash = B256::random();
392        let missing = NumHash::new(1, B256::random());
393        let bal = Bytes::from_static(&[0xc1, 0x01]);
394
395        store.insert(NumHash::new(1, hash), RawBal::from(bal.clone())).unwrap();
396
397        assert_eq!(read_many(&store, &[NumHash::new(1, hash), missing]), vec![Some(bal), None]);
398    }
399
400    #[test]
401    fn hash_lookup_reads_persisted_bal_through_store() {
402        let (_dir, store) = test_store();
403        let block = NumHash::new(1, B256::random());
404        let bal = Bytes::from_static(&[0xc1, 0x01]);
405
406        store.insert(block, RawBal::from(bal.clone())).unwrap();
407        store.flush(&[block]).unwrap();
408
409        let store_with_empty_buffer = RocksDBBalStore::new(store.rocksdb_provider().clone());
410        assert_eq!(store_with_empty_buffer.get_by_hash(block.hash).unwrap(), Some(bal));
411    }
412
413    #[test]
414    fn flush_prunes_buffer_retention() {
415        let (_dir, store) = test_store();
416        let old = NumHash::new(1, B256::with_last_byte(1));
417        let retained =
418            NumHash::new(DEFAULT_BAL_BUFFER_RETENTION_DISTANCE + 2, B256::with_last_byte(2));
419        let old_bal = Bytes::from_static(&[0xc1, 0x01]);
420        let retained_bal = Bytes::from_static(&[0xc1, 0x02]);
421
422        store.insert(old, RawBal::from(old_bal.clone())).unwrap();
423        store.insert(retained, RawBal::from(retained_bal.clone())).unwrap();
424
425        assert_eq!(
426            store.get_by_hashes(&[old.hash, retained.hash]).unwrap(),
427            vec![Some(old_bal), Some(retained_bal.clone())]
428        );
429
430        store.flush(&[retained]).unwrap();
431
432        assert_eq!(
433            store.get_by_hashes(&[old.hash, retained.hash]).unwrap(),
434            vec![None, Some(retained_bal)]
435        );
436        assert_eq!(disk_bal(&store, old), None);
437    }
438
439    #[test]
440    fn flush_prunes_only_durable_cache_entries() {
441        let (_dir, store) = test_store();
442        let old = NumHash::new(1, B256::with_last_byte(1));
443        let tip = NumHash::new(DEFAULT_BAL_BUFFER_RETENTION_DISTANCE + 2, B256::with_last_byte(2));
444        let old_bal = Bytes::from_static(&[0xc1, 0x01]);
445
446        store.insert(old, RawBal::from(old_bal.clone())).unwrap();
447        store.flush(&[old]).unwrap();
448        store.flush(&[tip]).unwrap();
449
450        assert!(!store.buffer.read().entries.contains_key(&old.hash));
451        assert_eq!(disk_bal(&store, old), Some(old_bal.clone()));
452        assert_eq!(store.get_by_hash(old.hash).unwrap(), Some(old_bal));
453    }
454
455    #[test]
456    fn configured_buffer_retention_distance_is_used() {
457        let dir = tempfile::tempdir().unwrap();
458        let rocksdb = test_rocksdb(&dir);
459        let store = RocksDBBalStore::with_buffer_retention_distance(rocksdb, 64);
460        let old = NumHash::new(1, B256::with_last_byte(1));
461        let tip = NumHash::new(34, B256::with_last_byte(2));
462
463        store.insert(old, RawBal::from(Bytes::from_static(&[0xc1, 0x01]))).unwrap();
464        store.flush(&[old]).unwrap();
465        store.flush(&[tip]).unwrap();
466
467        assert!(store.buffer.read().entries.contains_key(&old.hash));
468    }
469
470    #[test]
471    fn flush_writes_only_requested_pending_bals() {
472        let (_dir, store) = test_store();
473        let block_1 = NumHash::new(1, B256::with_last_byte(1));
474        let block_1_fork = NumHash::new(1, B256::with_last_byte(9));
475        let block_2 = NumHash::new(2, B256::with_last_byte(2));
476        let bal_1 = Bytes::from_static(&[0xc1, 0x01]);
477        let bal_1_fork = Bytes::from_static(&[0xc1, 0x09]);
478        let bal_2 = Bytes::from_static(&[0xc1, 0x02]);
479
480        store.insert(block_1, RawBal::from(bal_1.clone())).unwrap();
481        store.insert(block_1_fork, RawBal::from(bal_1_fork.clone())).unwrap();
482        store.insert(block_2, RawBal::from(bal_2.clone())).unwrap();
483
484        store.flush(&[block_1]).unwrap();
485
486        assert_eq!(disk_bal(&store, block_1), Some(bal_1.clone()));
487        assert_eq!(disk_bal(&store, block_1_fork), None);
488        assert_eq!(disk_bal(&store, block_2), None);
489        assert_eq!(
490            store.get_by_hashes(&[block_1.hash, block_1_fork.hash, block_2.hash]).unwrap(),
491            vec![Some(bal_1.clone()), Some(bal_1_fork), Some(bal_2)]
492        );
493
494        let store_with_empty_buffer = RocksDBBalStore::new(store.rocksdb_provider().clone());
495        assert_eq!(
496            store_with_empty_buffer
497                .get_by_hashes(&[block_1.hash, block_1_fork.hash, block_2.hash])
498                .unwrap(),
499            vec![Some(bal_1), None, None]
500        );
501    }
502
503    #[test]
504    fn sparse_numbers_are_valid() {
505        let (_dir, store) = test_store();
506        let hash_a = B256::with_last_byte(1);
507        let hash_b = B256::with_last_byte(2);
508        let bal_a = Bytes::from_static(&[0xc1, 0x01]);
509        let bal_b = Bytes::from_static(&[0xc1, 0x02]);
510
511        let block_a = NumHash::new(2, hash_a);
512        let block_b = NumHash::new(200, hash_b);
513
514        store.insert(block_a, RawBal::from(bal_a.clone())).unwrap();
515        store.flush(&[block_a]).unwrap();
516        store.insert(block_b, RawBal::from(bal_b.clone())).unwrap();
517        store.flush(&[block_b]).unwrap();
518
519        assert_eq!(read_many(&store, &[block_a, block_b]), vec![Some(bal_a), Some(bal_b)]);
520    }
521
522    #[test]
523    fn missing_and_empty_bal_are_distinct() {
524        let (_dir, store) = test_store();
525        let empty_hash = B256::with_last_byte(1);
526        let missing_hash = B256::with_last_byte(2);
527        let empty_bal = Bytes::from_static(&[0xc0]);
528
529        store.insert(NumHash::new(1, empty_hash), RawBal::from(empty_bal.clone())).unwrap();
530
531        assert_eq!(
532            read_many(&store, &[NumHash::new(1, empty_hash), NumHash::new(1, missing_hash)]),
533            vec![Some(empty_bal), None]
534        );
535    }
536
537    #[test]
538    fn prune_uses_eip_retention() {
539        let (_dir, store) = test_store();
540        let old_hash = B256::with_last_byte(1);
541        let retained_hash = B256::with_last_byte(2);
542        let retained_bal = Bytes::from_static(&[0xc1, 0x02]);
543        let tip = BAL_RETENTION_PERIOD_SLOTS + 2;
544
545        store
546            .insert(NumHash::new(1, old_hash), RawBal::from(Bytes::from_static(&[0xc1, 0x01])))
547            .unwrap();
548        store.insert(NumHash::new(2, retained_hash), RawBal::from(retained_bal.clone())).unwrap();
549        store.flush(&[NumHash::new(1, old_hash), NumHash::new(2, retained_hash)]).unwrap();
550
551        assert_eq!(store.prune(tip).unwrap(), 1);
552        assert_eq!(disk_bal(&store, NumHash::new(1, old_hash)), None);
553        assert_eq!(disk_bal(&store, NumHash::new(2, retained_hash)), Some(retained_bal.clone()));
554        assert_eq!(
555            read_many(&store, &[NumHash::new(1, old_hash), NumHash::new(2, retained_hash)]),
556            vec![None, Some(retained_bal)]
557        );
558    }
559
560    #[test]
561    fn stored_payload_hash_is_not_reverified() {
562        let (_dir, store) = test_store();
563        let block = NumHash::new(1, B256::with_last_byte(1));
564        let key = StoredBlockAccessListKey::new(block.number, block.hash);
565        let value = StoredBlockAccessList::new_unchecked(B256::ZERO, Bytes::from_static(&[0xc0]));
566
567        store.rocksdb_provider().put::<tables::BlockAccessLists>(key, &value).unwrap();
568        store
569            .rocksdb_provider()
570            .put::<tables::BlockAccessListBlockNumbers>(block.hash, &block.number)
571            .unwrap();
572
573        assert_eq!(store.get_by_hash(block.hash).unwrap(), Some(Bytes::from_static(&[0xc0])));
574    }
575
576    #[test]
577    fn canonical_pending_entries_are_retried() {
578        let mut buffer = RocksDBBalStoreBuffer::default();
579        let first = NumHash::new(1, B256::with_last_byte(1));
580        let second = NumHash::new(2, B256::with_last_byte(2));
581        buffer.insert(first, RawBal::from(Bytes::from_static(&[0xc1, 0x01])));
582        buffer.mark_canonical(&[first]);
583
584        assert_eq!(buffer.canonical_pending_entries().len(), 1);
585
586        // A failed flush leaves the canonical entry pending for the next attempt.
587        buffer.insert(second, RawBal::from(Bytes::from_static(&[0xc1, 0x02])));
588        buffer.mark_canonical(&[second]);
589        let retry = buffer.canonical_pending_entries();
590
591        assert_eq!(
592            retry.iter().map(|(key, _)| NumHash::new(key.number(), key.hash())).collect::<Vec<_>>(),
593            vec![first, second]
594        );
595    }
596
597    #[tokio::test]
598    async fn insert_notifies_subscribers() {
599        let (_dir, store) = test_store();
600        let mut stream = store.bal_stream();
601        let block = NumHash::new(1, B256::with_last_byte(1));
602        let bal = RawBal::from(Bytes::from_static(&[0xc0]));
603
604        store.insert(block, bal.clone()).unwrap();
605
606        assert_eq!(stream.next().await.unwrap(), BalNotification::new(block, bal));
607    }
608}
reth_provider/bal/rocksdb.rs

reth_provider/bal/
rocksdb.rs
