reth_transaction_pool/blobstore/
mem.rs

1use crate::blobstore::{BlobStore, BlobStoreCleanupStat, BlobStoreError, BlobStoreSize};
2use alloy_eips::{
3    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
4    eip7594::{BlobCellMask, BlobTransactionSidecarVariant, Cell},
5};
6use alloy_primitives::{map::B256Map, B128, B256};
7use parking_lot::RwLock;
8use std::sync::Arc;
9
10/// An in-memory blob store.
11#[derive(Clone, Debug, Default, PartialEq)]
12pub struct InMemoryBlobStore {
13    inner: Arc<InMemoryBlobStoreInner>,
14}
15
16impl InMemoryBlobStore {
17    /// Look up EIP-7594 blobs by their versioned hashes.
18    ///
19    /// This returns a result vector with the **same length and order** as the input
20    /// `versioned_hashes`. Each element is `Some(BlobAndProofV2)` if the blob is available, or
21    /// `None` if it is missing or an older sidecar version.
22    fn get_by_versioned_hashes_eip7594(
23        &self,
24        versioned_hashes: &[B256],
25    ) -> Vec<Option<BlobAndProofV2>> {
26        let mut result = vec![None; versioned_hashes.len()];
27        let mut missing_count = result.len();
28        for blob_sidecar in self.inner.store.read().values() {
29            if let Some(blob_sidecar) = blob_sidecar.as_eip7594() {
30                for (hash_idx, match_result) in
31                    blob_sidecar.match_versioned_hashes(versioned_hashes)
32                {
33                    let slot = &mut result[hash_idx];
34                    if slot.is_none() {
35                        missing_count -= 1;
36                    }
37                    *slot = Some(match_result);
38                }
39            }
40
41            // Return early if all blobs are found.
42            if missing_count == 0 {
43                // since versioned_hashes may have duplicates, we double check here
44                if result.iter().all(|blob| blob.is_some()) {
45                    break;
46                }
47            }
48        }
49        result
50    }
51
52    /// Look up EIP-7594 blob cells by their versioned hashes.
53    fn get_by_versioned_hashes_cells_eip7594(
54        &self,
55        versioned_hashes: &[B256],
56        indices_bitarray: B128,
57    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
58        let cell_mask = BlobCellMask::new(indices_bitarray);
59        let mut result = vec![None; versioned_hashes.len()];
60        let mut missing_count = result.len();
61        let blob_sidecars = self.inner.store.read().values().cloned().collect::<Vec<_>>();
62        for blob_sidecar in blob_sidecars {
63            if let Some(blob_sidecar) = blob_sidecar.as_eip7594() {
64                for (hash_idx, match_result) in blob_sidecar
65                    .match_versioned_hashes_cells(versioned_hashes, cell_mask)
66                    .map_err(|err| BlobStoreError::Other(Box::new(err)))?
67                {
68                    let slot = &mut result[hash_idx];
69                    if slot.is_none() {
70                        missing_count -= 1;
71                    }
72                    *slot = Some(match_result);
73                }
74            }
75
76            if missing_count == 0 && result.iter().all(Option::is_some) {
77                break;
78            }
79        }
80        Ok(result)
81    }
82}
83
84#[derive(Debug, Default)]
85struct InMemoryBlobStoreInner {
86    /// Storage for all blob data.
87    store: RwLock<B256Map<Arc<BlobTransactionSidecarVariant>>>,
88    size_tracker: BlobStoreSize,
89}
90
91impl PartialEq for InMemoryBlobStoreInner {
92    fn eq(&self, other: &Self) -> bool {
93        self.store.read().eq(&*other.store.read())
94    }
95}
96
97impl BlobStore for InMemoryBlobStore {
98    fn insert(&self, tx: B256, data: BlobTransactionSidecarVariant) -> Result<(), BlobStoreError> {
99        let mut store = self.inner.store.write();
100        self.inner.size_tracker.add_size(insert_size(&mut store, tx, data));
101        self.inner.size_tracker.update_len(store.len());
102        Ok(())
103    }
104
105    fn insert_all(
106        &self,
107        txs: Vec<(B256, BlobTransactionSidecarVariant)>,
108    ) -> Result<(), BlobStoreError> {
109        if txs.is_empty() {
110            return Ok(())
111        }
112        let mut store = self.inner.store.write();
113        let mut total_add = 0;
114        for (tx, data) in txs {
115            let add = insert_size(&mut store, tx, data);
116            total_add += add;
117        }
118        self.inner.size_tracker.add_size(total_add);
119        self.inner.size_tracker.update_len(store.len());
120        Ok(())
121    }
122
123    fn delete(&self, tx: B256) -> Result<(), BlobStoreError> {
124        let mut store = self.inner.store.write();
125        let sub = remove_size(&mut store, &tx);
126        self.inner.size_tracker.sub_size(sub);
127        self.inner.size_tracker.update_len(store.len());
128        Ok(())
129    }
130
131    fn delete_all(&self, txs: Vec<B256>) -> Result<(), BlobStoreError> {
132        if txs.is_empty() {
133            return Ok(())
134        }
135        let mut store = self.inner.store.write();
136        let mut total_sub = 0;
137        for tx in txs {
138            total_sub += remove_size(&mut store, &tx);
139        }
140        self.inner.size_tracker.sub_size(total_sub);
141        self.inner.size_tracker.update_len(store.len());
142        Ok(())
143    }
144
145    fn cleanup(&self) -> BlobStoreCleanupStat {
146        BlobStoreCleanupStat::default()
147    }
148
149    // Retrieves the decoded blob data for the given transaction hash.
150    fn get(&self, tx: B256) -> Result<Option<Arc<BlobTransactionSidecarVariant>>, BlobStoreError> {
151        Ok(self.inner.store.read().get(&tx).cloned())
152    }
153
154    fn contains(&self, tx: B256) -> Result<bool, BlobStoreError> {
155        Ok(self.inner.store.read().contains_key(&tx))
156    }
157
158    fn get_all(
159        &self,
160        txs: Vec<B256>,
161    ) -> Result<Vec<(B256, Arc<BlobTransactionSidecarVariant>)>, BlobStoreError> {
162        let store = self.inner.store.read();
163        Ok(txs.into_iter().filter_map(|tx| store.get(&tx).map(|item| (tx, item.clone()))).collect())
164    }
165
166    fn get_exact(
167        &self,
168        txs: Vec<B256>,
169    ) -> Result<Vec<Arc<BlobTransactionSidecarVariant>>, BlobStoreError> {
170        if txs.is_empty() {
171            return Ok(Vec::new());
172        }
173        let store = self.inner.store.read();
174        txs.into_iter()
175            .map(|tx| store.get(&tx).cloned().ok_or(BlobStoreError::MissingSidecar(tx)))
176            .collect()
177    }
178
179    fn get_by_versioned_hashes_v1(
180        &self,
181        versioned_hashes: &[B256],
182    ) -> Result<Vec<Option<BlobAndProofV1>>, BlobStoreError> {
183        let mut result = vec![None; versioned_hashes.len()];
184        for blob_sidecar in self.inner.store.read().values() {
185            if let Some(blob_sidecar) = blob_sidecar.as_eip4844() {
186                for (hash_idx, match_result) in
187                    blob_sidecar.match_versioned_hashes(versioned_hashes)
188                {
189                    result[hash_idx] = Some(match_result);
190                }
191            }
192
193            // Return early if all blobs are found.
194            if result.iter().all(|blob| blob.is_some()) {
195                break;
196            }
197        }
198        Ok(result)
199    }
200
201    fn get_by_versioned_hashes_v2(
202        &self,
203        versioned_hashes: &[B256],
204    ) -> Result<Option<Vec<BlobAndProofV2>>, BlobStoreError> {
205        let result = self.get_by_versioned_hashes_eip7594(versioned_hashes);
206        if result.iter().all(|blob| blob.is_some()) {
207            Ok(Some(result.into_iter().map(Option::unwrap).collect()))
208        } else {
209            Ok(None)
210        }
211    }
212
213    fn get_by_versioned_hashes_v3(
214        &self,
215        versioned_hashes: &[B256],
216    ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError> {
217        Ok(self.get_by_versioned_hashes_eip7594(versioned_hashes))
218    }
219
220    fn get_by_versioned_hashes_v4(
221        &self,
222        versioned_hashes: &[B256],
223        indices_bitarray: B128,
224    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
225        self.get_by_versioned_hashes_cells_eip7594(versioned_hashes, indices_bitarray)
226    }
227
228    fn has_versioned_hashes(&self, versioned_hashes: &[B256]) -> Result<Vec<bool>, BlobStoreError> {
229        let mut result = vec![false; versioned_hashes.len()];
230        for blob_sidecar in self.inner.store.read().values() {
231            for available_hash in blob_sidecar.versioned_hashes() {
232                for (idx, requested_hash) in versioned_hashes.iter().enumerate() {
233                    if !result[idx] && *requested_hash == available_hash {
234                        result[idx] = true;
235                    }
236                }
237            }
238
239            if result.iter().all(|available| *available) {
240                break;
241            }
242        }
243        Ok(result)
244    }
245
246    fn get_cells(
247        &self,
248        tx: B256,
249        indices_bitarray: B128,
250    ) -> Result<Option<Vec<Cell>>, BlobStoreError> {
251        let Some(sidecar) = self.get(tx)? else {
252            return Ok(None);
253        };
254
255        let Some(sidecar) = sidecar.as_eip7594() else {
256            return Ok(None);
257        };
258
259        sidecar
260            .compute_matching_cells(BlobCellMask::new(indices_bitarray))
261            .map(Some)
262            .map_err(|err| BlobStoreError::Other(Box::new(err)))
263    }
264
265    fn data_size_hint(&self) -> Option<usize> {
266        Some(self.inner.size_tracker.data_size())
267    }
268
269    fn blobs_len(&self) -> usize {
270        self.inner.size_tracker.blobs_len()
271    }
272}
273
274/// Removes the given blob from the store and returns the size of the blob that was removed.
275#[inline]
276fn remove_size(store: &mut B256Map<Arc<BlobTransactionSidecarVariant>>, tx: &B256) -> usize {
277    store.remove(tx).map(|rem| rem.size()).unwrap_or_default()
278}
279
280/// Inserts the given blob into the store and returns the size of the blob that was added.
281///
282/// We don't need to handle the size updates for replacements because transactions are unique.
283#[inline]
284fn insert_size(
285    store: &mut B256Map<Arc<BlobTransactionSidecarVariant>>,
286    tx: B256,
287    blob: BlobTransactionSidecarVariant,
288) -> usize {
289    let add = blob.size();
290    store.insert(tx, Arc::new(blob));
291    add
292}
293
294#[cfg(test)]
295mod tests {
296    use super::*;
297    use alloy_consensus::BlobTransactionSidecar;
298    use alloy_eips::{
299        eip4844::{kzg_to_versioned_hash, Blob, BlobAndProofV2, Bytes48},
300        eip7594::{
301            BlobTransactionSidecarEip7594, BlobTransactionSidecarVariant, CELLS_PER_EXT_BLOB,
302        },
303    };
304
305    fn eip7594_single_blob_sidecar() -> (BlobTransactionSidecarVariant, B256, BlobAndProofV2) {
306        let blob = Blob::default();
307        let commitment = Bytes48::default();
308        let cell_proofs = vec![Bytes48::default(); CELLS_PER_EXT_BLOB];
309
310        let versioned_hash = kzg_to_versioned_hash(commitment.as_slice());
311
312        let expected =
313            BlobAndProofV2 { blob: Box::new(Blob::default()), proofs: cell_proofs.clone() };
314        let sidecar = BlobTransactionSidecarEip7594::new(vec![blob], vec![commitment], cell_proofs);
315
316        (BlobTransactionSidecarVariant::Eip7594(sidecar), versioned_hash, expected)
317    }
318
319    fn eip4844_single_blob_sidecar() -> (BlobTransactionSidecarVariant, B256) {
320        let blob = Blob::default();
321        let commitment = Bytes48::from([1u8; 48]);
322        let proof = Bytes48::default();
323        let versioned_hash = kzg_to_versioned_hash(commitment.as_slice());
324        let sidecar = BlobTransactionSidecar {
325            blobs: vec![blob],
326            commitments: vec![commitment],
327            proofs: vec![proof],
328        };
329
330        (BlobTransactionSidecarVariant::Eip4844(sidecar), versioned_hash)
331    }
332
333    #[test]
334    fn mem_has_blobs_returns_ordered_availability() {
335        let store = InMemoryBlobStore::default();
336
337        let (eip7594_sidecar, eip7594_hash, _) = eip7594_single_blob_sidecar();
338        let (eip4844_sidecar, eip4844_hash) = eip4844_single_blob_sidecar();
339        store.insert(B256::random(), eip7594_sidecar).unwrap();
340        store.insert(B256::random(), eip4844_sidecar).unwrap();
341
342        let request = vec![eip7594_hash, B256::ZERO, eip4844_hash, eip7594_hash];
343        assert_eq!(store.has_versioned_hashes(&request).unwrap(), vec![true, false, true, true]);
344    }
345
346    #[test]
347    fn mem_get_blobs_v3_returns_partial_results() {
348        let store = InMemoryBlobStore::default();
349
350        let (sidecar, versioned_hash, expected) = eip7594_single_blob_sidecar();
351        store.insert(B256::random(), sidecar).unwrap();
352
353        assert_ne!(versioned_hash, B256::ZERO);
354
355        let request = vec![versioned_hash, B256::ZERO];
356        let v2 = store.get_by_versioned_hashes_v2(&request).unwrap();
357        assert!(v2.is_none(), "v2 must return null if any requested blob is missing");
358
359        let v3 = store.get_by_versioned_hashes_v3(&request).unwrap();
360        assert_eq!(v3, vec![Some(expected), None]);
361    }
362
363    #[test]
364    fn mem_get_blobs_v4_returns_requested_cells() {
365        let store = InMemoryBlobStore::default();
366
367        let (sidecar, versioned_hash, _) = eip7594_single_blob_sidecar();
368        store.insert(B256::random(), sidecar).unwrap();
369
370        let indices_bitarray = B128::from((1u128 << 0) | (1u128 << 7));
371        let request = vec![versioned_hash, B256::ZERO];
372
373        let v4 = store.get_by_versioned_hashes_v4(&request, indices_bitarray).unwrap();
374        assert_eq!(v4.len(), request.len());
375        assert!(v4[1].is_none());
376
377        let cells_and_proofs = v4[0].as_ref().unwrap();
378        assert_eq!(cells_and_proofs.blob_cells.len(), 2);
379        assert_eq!(cells_and_proofs.proofs.len(), 2);
380        assert!(cells_and_proofs.blob_cells.iter().all(Option::is_some));
381        assert_eq!(cells_and_proofs.proofs, vec![Some(Bytes48::default()); 2]);
382    }
383
384    #[test]
385    fn mem_get_cells_returns_requested_cells() {
386        let store = InMemoryBlobStore::default();
387
388        let tx_hash = B256::random();
389        let (sidecar, versioned_hash, _) = eip7594_single_blob_sidecar();
390        store.insert(tx_hash, sidecar).unwrap();
391
392        let indices_bitarray = B128::from((1u128 << 0) | (1u128 << 7));
393        let expected = store
394            .get_by_versioned_hashes_v4(&[versioned_hash], indices_bitarray)
395            .unwrap()
396            .pop()
397            .unwrap()
398            .unwrap()
399            .blob_cells
400            .into_iter()
401            .collect::<Option<Vec<_>>>()
402            .unwrap();
403
404        assert_eq!(store.get_cells(tx_hash, indices_bitarray).unwrap(), Some(expected));
405    }
406}
reth_transaction_pool/blobstore/mem.rs

reth_transaction_pool/blobstore/
mem.rs
