forest/chain/store/
chain_store.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use super::{
5    Error,
6    index::{ChainIndex, ResolveNullTipset},
7    tipset_tracker::TipsetTracker,
8};
9use crate::networks::{ChainConfig, Height};
10use crate::prelude::*;
11use crate::rpc::chain::PathChange;
12use crate::rpc::{
13    chain::ChainGetTipSetFinalityStatus,
14    eth::{eth_tx_from_signed_eth_message, types::EthHash},
15};
16use crate::shim::clock::ChainEpoch;
17use crate::shim::{
18    address::Address, executor::Receipt, message::Message, state_tree::StateTree,
19    version::NetworkVersion,
20};
21use crate::state_manager::ExecutedTipset;
22use crate::utils::db::{BlockstoreExt, CborStoreExt};
23use crate::utils::publisher::Publisher;
24use crate::{
25    blocks::{CachingBlockHeader, Tipset, TipsetKey, TxMeta},
26    db::{DbImpl, EthMappingsStoreExt as _, HeaviestTipsetKeyProvider},
27    message::{ChainMessage, SignedMessage},
28};
29use crate::{fil_cns, utils::cache::SizeTrackingCache};
30use crate::{
31    interpreter::{BlockMessages, VMTrace},
32    rpc::chain::PathChanges,
33};
34use ahash::HashMap;
35use arc_swap::{ArcSwap, ArcSwapOption};
36use fil_actors_shared::fvm_ipld_amt::Amtv0 as Amt;
37use fvm_ipld_encoding::CborStore;
38use nonzero_ext::nonzero;
39use serde::{Serialize, de::DeserializeOwned};
40use std::{
41    num::NonZeroUsize,
42    sync::atomic::{self, AtomicI64},
43};
44
45// Assume a tipset has 5 blocks on average, we cache 1-day-worth of validated blocks. (5 * 2 * 60 * 24 = 14400)
46const VALIDATED_BLOCKS_CACHE_SIZE: NonZeroUsize = nonzero!(14400usize);
47
48/// Disambiguate the type to signify that we are expecting a delta and not an actual epoch/height
49/// while maintaining the same type.
50pub type ChainEpochDelta = ChainEpoch;
51
52/// Outcome of [`ChainStore::resolve_to_deterministic_address_at_finality`].
53pub enum AtFinalityResolution {
54    /// Resolved against a tipset at least `chain_finality` epochs behind the
55    /// requested one. The mapping is identical on every possible future
56    /// chain, so it is safe to memoize by actor ID alone.
57    ReorgStable(Address),
58    /// The chain is younger than finality, so the address was resolved
59    /// against the requested tipset itself and may change across a reorg.
60    Unstable(Address),
61}
62
63impl AtFinalityResolution {
64    pub fn into_address(self) -> Address {
65        match self {
66            Self::ReorgStable(addr) | Self::Unstable(addr) => addr,
67        }
68    }
69}
70
71/// `Enum` for `pubsub` channel that defines message type variant and data
72/// contained in message type.
73pub type HeadChange = PathChange<Tipset>;
74
75pub type HeadChanges = PathChanges<Tipset>;
76
77/// Stores chain data such as heaviest tipset and cached tipset info at each
78/// epoch. This structure is thread-safe, and all caches are wrapped in a mutex
79/// to allow a consistent `ChainStore` to be shared across tasks.
80pub struct ChainStore {
81    /// Publisher for head change events
82    head_changes: Publisher<HeadChanges>,
83
84    /// Heaviest tipset cache
85    heaviest_tipset: Arc<ArcSwap<Tipset>>,
86
87    /// F3 finalized tipset cache
88    f3_finalized_tipset: Arc<ArcSwapOption<Tipset>>,
89
90    /// EC calculator finalized epoch cache
91    ec_calculator_finalized_epoch: Arc<AtomicI64>,
92
93    /// Used as a cache for tipset `lookbacks`.
94    chain_index: ChainIndex,
95
96    /// Tracks blocks for the purpose of forming tipsets.
97    tipset_tracker: TipsetTracker<DbImpl>,
98
99    /// Genesis tipset.
100    genesis: Tipset,
101
102    /// validated blocks
103    pub(crate) validated_blocks: SizeTrackingCache<CidWrapper, ()>,
104
105    /// Needed by the Ethereum mapping.
106    chain_config: Arc<ChainConfig>,
107
108    /// Cache for messages in tipsets, keyed by tipset key.
109    messages_in_tipset_cache: MessagesInTipsetCache,
110
111    /// Head key of the last clean [`Self::repair_tipset_lookup`] scan, for debouncing.
112    last_clean_lookup_repair_head: Arc<ArcSwapOption<TipsetKey>>,
113}
114
115impl ShallowClone for ChainStore {
116    fn shallow_clone(&self) -> Self {
117        Self {
118            head_changes: self.head_changes.clone(),
119            heaviest_tipset: self.heaviest_tipset.shallow_clone(),
120            f3_finalized_tipset: self.f3_finalized_tipset.shallow_clone(),
121            ec_calculator_finalized_epoch: self.ec_calculator_finalized_epoch.shallow_clone(),
122            chain_index: self.chain_index.shallow_clone(),
123            tipset_tracker: self.tipset_tracker.shallow_clone(),
124            genesis: self.genesis.shallow_clone(),
125            validated_blocks: self.validated_blocks.shallow_clone(),
126            chain_config: self.chain_config.shallow_clone(),
127            messages_in_tipset_cache: self.messages_in_tipset_cache.shallow_clone(),
128            last_clean_lookup_repair_head: self.last_clean_lookup_repair_head.shallow_clone(),
129        }
130    }
131}
132
133impl ChainStore {
134    pub fn new(
135        db: impl Into<DbImpl>,
136        chain_config: Arc<ChainConfig>,
137        genesis: impl Into<Tipset>,
138    ) -> anyhow::Result<Self> {
139        let db = db.into();
140        let genesis = genesis.into();
141        anyhow::ensure!(genesis.epoch() == 0, "genesis tipset must be at epoch 0");
142        let head = if let Some(head_tsk) = db
143            .heaviest_tipset_key()
144            .context("failed to load head tipset key")?
145        {
146            Tipset::load_required(&db, &head_tsk)
147                .with_context(|| format!("failed to load head tipset with key {head_tsk}"))?
148        } else {
149            genesis.shallow_clone()
150        };
151        let heaviest_tipset = Arc::new(ArcSwap::from_pointee(head.shallow_clone()));
152        let f3_finalized_tipset: Arc<ArcSwapOption<Tipset>> = Default::default();
153        let chain_index = ChainIndex::new(db.shallow_clone(), genesis.shallow_clone());
154        let ec_calculator_finalized_epoch = Arc::new(AtomicI64::new(
155            ChainGetTipSetFinalityStatus::get_ec_finality_epoch(&chain_index, &chain_config, &head),
156        ));
157        let chain_index = chain_index.with_is_epoch_finalized(Arc::new({
158            let ec_calculator_finalized_epoch = ec_calculator_finalized_epoch.shallow_clone();
159            move |epoch| {
160                let finalized = ec_calculator_finalized_epoch.load(atomic::Ordering::Acquire);
161                epoch <= finalized
162            }
163        }));
164        Ok(Self {
165            head_changes: Publisher::default(),
166            chain_index,
167            tipset_tracker: TipsetTracker::new(db, chain_config.clone()),
168            heaviest_tipset,
169            f3_finalized_tipset,
170            ec_calculator_finalized_epoch,
171            genesis,
172            validated_blocks: SizeTrackingCache::new_with_metrics(
173                "validated_blocks",
174                VALIDATED_BLOCKS_CACHE_SIZE,
175            ),
176            chain_config,
177            messages_in_tipset_cache: Default::default(),
178            last_clean_lookup_repair_head: Default::default(),
179        })
180    }
181
182    /// Verifies and repairs the lookup table over the last `chain_finality` epochs of the
183    /// head's lineage, returning the number of wrong entries fixed. Clean scans are
184    /// debounced per head key, bounding the cost of validation-failure bursts while any
185    /// head change (including a same-epoch reorg) allows a re-scan.
186    pub fn repair_tipset_lookup(&self) -> anyhow::Result<usize> {
187        let head = self.heaviest_tipset();
188        if self.last_clean_lookup_repair_head.load().as_deref() == Some(head.key()) {
189            return Ok(0);
190        }
191        let n_repaired = self.chain_index.repair_tipset_lookup_window(
192            &head,
193            self.chain_config.policy.chain_finality,
194            self.ec_calculator_finalized_epoch(),
195        )?;
196        if n_repaired == 0 {
197            self.last_clean_lookup_repair_head
198                .store(Some(Arc::new(head.key().clone())));
199        }
200        Ok(n_repaired)
201    }
202
203    /// Sets F3 finalized tipset
204    pub fn set_f3_finalized_tipset(&self, ts: Tipset) {
205        self.f3_finalized_tipset.store(Some(ts.into()));
206    }
207
208    /// Gets F3 finalized tipset
209    pub fn f3_finalized_tipset(&self) -> Option<Tipset> {
210        self.f3_finalized_tipset
211            .load()
212            .as_ref()
213            .map(|ts| ts.as_ref().shallow_clone())
214    }
215
216    /// Gets the EC calculator finalized epoch
217    pub fn ec_calculator_finalized_epoch(&self) -> ChainEpoch {
218        self.ec_calculator_finalized_epoch
219            .load(atomic::Ordering::Acquire)
220    }
221
222    /// Cache for messages in tipsets, keyed by tipset key.
223    pub fn messages_in_tipset_cache(&self) -> &MessagesInTipsetCache {
224        &self.messages_in_tipset_cache
225    }
226
227    /// Sets heaviest tipset
228    pub fn set_heaviest_tipset(&self, head: Tipset) -> Result<(), Error> {
229        head.key().save(self.db())?;
230        self.db().set_heaviest_tipset_key(head.key())?;
231
232        let finalized_epoch = ChainGetTipSetFinalityStatus::get_ec_finality_epoch(
233            self.chain_index(),
234            self.chain_config(),
235            &head,
236        );
237        self.ec_calculator_finalized_epoch
238            .store(finalized_epoch, atomic::Ordering::Release);
239
240        // Update the tipset lookup table.
241        if let Err(e) = self
242            .chain_index
243            .update_tipset_lookup_for_finalized_head(&head, finalized_epoch)
244        {
245            error!("failed to update tipset lookup table: {e:#?}");
246        }
247        // Fix stale lookups at null rounds which could be caused by chain reorg.
248        if let Err(e) = self.chain_index.cleanup_stale_lookup_at_new_head(&head) {
249            error!("failed to cleanup stale null round lookups: {e:#?}");
250        }
251
252        let old_head = self.heaviest_tipset.swap(head.shallow_clone().into());
253        if old_head.key() != head.key() && self.head_changes.has_subscribers() {
254            let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key())
255            {
256                Ok(changes) => changes,
257                Err(e) => {
258                    // Do not warn when the old head is genesis
259                    if old_head.epoch() > 0 {
260                        error!("failed to get chain path changes: {e:#}");
261                    }
262                    // Fallback to single apply
263                    PathChanges {
264                        applies: vec![head],
265                        reverts: vec![],
266                    }
267                }
268            };
269            if !changes.is_empty() {
270                self.head_changes.publish(changes);
271            }
272        }
273
274        Ok(())
275    }
276
277    /// Adds a block header to the tipset tracker, which tracks valid headers.
278    pub fn add_to_tipset_tracker(&self, header: &CachingBlockHeader) {
279        self.tipset_tracker.add(header);
280    }
281
282    /// Writes pending tipset block headers to data store and updates heaviest tipset
283    /// with other compatible pending tracked headers when it's heavier than
284    /// the current head.
285    /// Returns whether the heaviest tipset is updated.
286    pub fn maybe_update_pending_head(&self, ts: &Tipset) -> Result<bool, Error> {
287        persist_objects(self.db(), ts.block_headers().iter())?;
288        // Expand tipset to include other compatible blocks at the epoch.
289        let expanded = self.expand_tipset(ts.min_ticket_block().clone())?;
290        self.maybe_update_heaviest(expanded)
291    }
292
293    /// Reads the `TipsetKey` from the blockstore for `EthAPI` queries.
294    pub fn get_required_tipset_key(&self, hash: &EthHash) -> Result<TipsetKey, Error> {
295        Ok(TipsetKey::load(self.db(), &hash.to_cid())?)
296    }
297
298    /// Writes with timestamp the `Hash` to `Cid` mapping to the blockstore for `EthAPI` queries.
299    pub fn put_mapping(&self, k: EthHash, v: Cid, timestamp: u64) -> Result<(), Error> {
300        self.db().write_obj(&k, &(v, timestamp))?;
301        Ok(())
302    }
303
304    /// Like [`Self::put_mapping`], but only overwrites an existing entry when the incoming
305    /// `timestamp` is strictly newer than the stored one. Used by index backfill (which runs
306    /// concurrently with the live head indexer) so that walking historical tipsets cannot
307    /// clobber a mapping written for a newer tipset with an older one.
308    pub fn put_mapping_if_newer(&self, k: EthHash, v: Cid, timestamp: u64) -> Result<(), Error> {
309        if let Some((_, existing_timestamp)) = self.db().read_obj::<(Cid, u64)>(&k)?
310            && existing_timestamp >= timestamp
311        {
312            return Ok(());
313        }
314        self.put_mapping(k, v, timestamp)
315    }
316
317    /// Reads the `Cid` from the blockstore for `EthAPI` queries.
318    pub fn get_mapping(&self, hash: &EthHash) -> Result<Option<Cid>, Error> {
319        Ok(self.db().read_obj::<(Cid, u64)>(hash)?.map(|(cid, _)| cid))
320    }
321
322    /// Expands tipset to tipset with all other headers in the same epoch using
323    /// the tipset tracker.
324    fn expand_tipset(&self, header: CachingBlockHeader) -> Result<Tipset, Error> {
325        self.tipset_tracker.expand(header)
326    }
327
328    /// Returns the genesis block header.
329    pub fn genesis_block_header(&self) -> &CachingBlockHeader {
330        self.genesis.min_ticket_block()
331    }
332
333    /// Returns the genesis tipset.
334    pub fn genesis_tipset(&self) -> Tipset {
335        self.genesis.shallow_clone()
336    }
337
338    /// Returns the currently tracked heaviest tipset.
339    pub fn heaviest_tipset(&self) -> Tipset {
340        self.heaviest_tipset.load().as_ref().shallow_clone()
341    }
342
343    /// Subscribes to head changes with an unbounded, lossless queue. Use this for
344    /// consumers that must not miss any head change (e.g. the chain/message indexers).
345    pub fn subscribe_head_changes(&self) -> flume::Receiver<HeadChanges> {
346        self.head_changes.subscribe()
347    }
348
349    /// Subscribes to head changes with a bounded queue that drops events for this
350    /// subscriber alone once it falls `cap` behind. Use this for best-effort consumers
351    /// that must not be able to grow memory without bound, e.g. ones driven by an
352    /// untrusted RPC client's read rate.
353    pub fn subscribe_head_changes_bounded(&self, cap: usize) -> flume::Receiver<HeadChanges> {
354        self.head_changes.subscribe_bounded(cap)
355    }
356
357    /// Returns a borrowed key-value store instance.
358    pub fn db(&self) -> &DbImpl {
359        self.chain_index().db()
360    }
361
362    /// Returns an owned key-value store instance.
363    pub fn db_owned(&self) -> DbImpl {
364        self.chain_index().db_owned()
365    }
366
367    /// Returns the chain index
368    pub fn chain_index(&self) -> &ChainIndex {
369        &self.chain_index
370    }
371
372    /// Returns the chain configuration
373    pub fn chain_config(&self) -> &Arc<ChainConfig> {
374        &self.chain_config
375    }
376
377    /// Resolves `addr` to its deterministic (public-key or delegated) form
378    /// using the state at `chain_finality` epochs behind `ts`. Falls back to
379    /// `ts` itself when the chain is younger than finality; the returned
380    /// [`AtFinalityResolution`] says which of the two happened.
381    ///
382    /// Matches the logic at <https://github.com/filecoin-project/lotus/blob/v1.35.1/chain/stmgr/stmgr.go#L361>
383    pub fn resolve_to_deterministic_address_at_finality(
384        &self,
385        addr: &Address,
386        ts: &Tipset,
387    ) -> anyhow::Result<AtFinalityResolution> {
388        use crate::shim::address::Protocol::*;
389        match addr.protocol() {
390            BLS | Secp256k1 | Delegated => Ok(AtFinalityResolution::ReorgStable(*addr)),
391            ID => {
392                let finality_deep = ts.epoch() > self.chain_config().policy.chain_finality;
393                let lookback_ts = if finality_deep {
394                    self.chain_index().load_required_tipset_by_height_blocking(
395                        ts.epoch() - self.chain_config().policy.chain_finality,
396                        ts.shallow_clone(),
397                        ResolveNullTipset::TakeOlder,
398                    )?
399                } else {
400                    ts.shallow_clone()
401                };
402                let state = StateTree::new_from_root(self.db(), lookback_ts.parent_state())?;
403                let resolved = state.resolve_to_deterministic_address(self.db(), *addr)?;
404                Ok(if finality_deep {
405                    AtFinalityResolution::ReorgStable(resolved)
406                } else {
407                    AtFinalityResolution::Unstable(resolved)
408                })
409            }
410            Actor => anyhow::bail!("Cannot resolve actor address to key address"),
411        }
412    }
413
414    /// Lotus often treats an empty [`TipsetKey`] as shorthand for "the heaviest tipset".
415    /// You may opt-in to that behavior by calling this method with [`None`].
416    ///
417    /// This calls fails if the tipset is missing or invalid.
418    #[tracing::instrument(skip_all)]
419    pub fn load_required_tipset_or_heaviest<'a>(
420        &self,
421        maybe_key: impl Into<Option<&'a TipsetKey>>,
422    ) -> Result<Tipset, Error> {
423        match maybe_key.into() {
424            Some(key) => self.chain_index.load_required_tipset(key),
425            None => Ok(self.heaviest_tipset()),
426        }
427    }
428
429    /// Returns [`None`] when `ts` has no known child on the current heaviest chain
430    /// (e.g. `ts` is the chain head). Blockstore errors are returned as [`Err`].
431    pub async fn load_child_tipset(&self, ts: &Tipset) -> Result<Option<Tipset>, Error> {
432        let head = self.heaviest_tipset();
433        if head.parents() == ts.key() {
434            Ok(Some(head))
435        } else if head.epoch() > ts.epoch() {
436            match self
437                .chain_index()
438                .tipset_by_height(ts.epoch() + 1, head, ResolveNullTipset::TakeNewer)
439                .await?
440            {
441                Some(maybe_child) if maybe_child.parents() == ts.key() => Ok(Some(maybe_child)),
442                _ => Ok(None),
443            }
444        } else {
445            Ok(None)
446        }
447    }
448
449    /// Determines if provided tipset is heavier than existing known heaviest
450    /// tipset.
451    /// Returns whether the heaviest tipset is updated.
452    fn maybe_update_heaviest(&self, ts: Tipset) -> Result<bool, Error> {
453        // Calculate heaviest weight before matching to avoid deadlock with mutex
454        let heaviest_weight = fil_cns::weight(self.db(), &self.heaviest_tipset())?;
455
456        let new_weight = fil_cns::weight(self.db(), &ts)?;
457        let curr_weight = heaviest_weight;
458
459        if new_weight > curr_weight {
460            self.set_heaviest_tipset(ts)?;
461            Ok(true)
462        } else {
463            Ok(false)
464        }
465    }
466
467    /// Checks metadata file if block has already been validated.
468    pub fn is_block_validated(&self, cid: &Cid) -> bool {
469        let validated = self.validated_blocks.get(cid).is_some();
470        if validated {
471            trace!("Block {cid} was previously validated");
472        }
473        validated
474    }
475
476    /// Marks block as validated in the metadata file.
477    pub fn mark_block_as_validated(&self, cid: &Cid) {
478        self.validated_blocks.insert((*cid).into(), ());
479    }
480
481    pub fn unmark_block_as_validated(&self, cid: &Cid) {
482        self.validated_blocks.remove(cid);
483    }
484
485    /// Retrieves ordered valid messages from a `Tipset`. This will only include
486    /// messages that will be passed through the VM.
487    pub fn messages_for_tipset(&self, ts: &Tipset) -> Result<Arc<Vec<ChainMessage>>, Error> {
488        Ok(self
489            .messages_in_tipset_cache()
490            .get_or_insert_with(ts.key(), || {
491                let bmsgs = BlockMessages::for_tipset(self.db(), ts)?;
492                anyhow::Ok(
493                    bmsgs
494                        .into_iter()
495                        .flat_map(|bm| bm.messages)
496                        .collect_vec()
497                        .into(),
498                )
499            })?)
500    }
501
502    /// Gets look-back tipset (and state-root of that tipset) for block
503    /// validations.
504    ///
505    /// The look-back tipset for a round is the tipset with epoch `round -
506    /// chain_finality`. [Chain
507    /// finality](https://docs.filecoin.io/reference/general/glossary/#finality)
508    /// is usually 900. The `heaviest_tipset` is a reference point in the
509    /// blockchain. It must be a child of the look-back tipset.
510    pub fn get_lookback_tipset_for_round_blocking(
511        chain_index: &ChainIndex,
512        chain_config: &Arc<ChainConfig>,
513        heaviest_tipset: &Tipset,
514        round: ChainEpoch,
515    ) -> Result<(Tipset, Cid), Error> {
516        let version = chain_config.network_version(round);
517        let lb = if version <= NetworkVersion::V3 {
518            ChainEpoch::from(10)
519        } else {
520            chain_config.policy.chain_finality
521        };
522        // The subtraction, not the result, is what must be guarded, as in Lotus:
523        // <https://github.com/filecoin-project/lotus/blob/v1.35.1/chain/stmgr/utils.go#L167>
524        let lbr = if round > lb { round - lb } else { 0 };
525
526        // More null blocks than our lookback
527        if lbr >= heaviest_tipset.epoch() {
528            // Legitimate while the 10-block WinningPoSt lookback was active (nv <= 3),
529            // where >9 consecutive null rounds could push lbr past ts.epoch().
530            // Any lookback at genesis should resolve to genesis.
531            if version <= NetworkVersion::V3 || heaviest_tipset.epoch() == 0 {
532                let genesis_timestamp = chain_index.genesis().min_ticket_block().timestamp;
533                let beacon = Arc::new(chain_config.get_beacon_schedule(genesis_timestamp));
534                let ExecutedTipset { state_root, .. } =
535                    crate::state_manager::apply_block_messages_blocking(
536                        chain_index.shallow_clone(),
537                        chain_config.shallow_clone(),
538                        beacon,
539                        // Using shared WASM engine here as creating new WASM engines is expensive
540                        &crate::shim::machine::GLOBAL_MULTI_ENGINE,
541                        heaviest_tipset.clone(),
542                        crate::state_manager::NO_CALLBACK,
543                        VMTrace::NotTraced,
544                    )
545                    .map_err(|e| Error::Other(e.to_string()))?;
546                return Ok((heaviest_tipset.clone(), state_root));
547            } else {
548                return Err(Error::LookbackHeightOverflow {
549                    lookback_height: lbr,
550                    base_height: heaviest_tipset.epoch(),
551                });
552            }
553        }
554
555        let next_ts = chain_index
556            .load_required_tipset_by_height_blocking(
557                lbr + 1,
558                heaviest_tipset.clone(),
559                ResolveNullTipset::TakeNewer,
560            )
561            .map_err(|e| Error::Other(format!("Could not get tipset by height {e:?}")))?;
562        if lbr > next_ts.epoch() {
563            return Err(Error::Other(format!(
564                "failed to find non-null tipset {:?} {} which is known to exist, found {:?} {}",
565                heaviest_tipset.key(),
566                heaviest_tipset.epoch(),
567                next_ts.key(),
568                next_ts.epoch()
569            )));
570        }
571        let lbts = chain_index
572            .load_required_tipset(next_ts.parents())
573            .map_err(|e| Error::Other(format!("Could not get tipset from keys {e:?}")))?;
574        Ok((lbts, *next_ts.parent_state()))
575    }
576
577    pub async fn get_lookback_tipset_for_round(
578        chain_index: ChainIndex,
579        chain_config: Arc<ChainConfig>,
580        heaviest_tipset: Tipset,
581        round: ChainEpoch,
582    ) -> Result<(Tipset, Cid), Error> {
583        tokio::task::spawn_blocking(move || {
584            Self::get_lookback_tipset_for_round_blocking(
585                &chain_index,
586                &chain_config,
587                &heaviest_tipset,
588                round,
589            )
590        })
591        .await?
592    }
593
594    /// Filter [`SignedMessage`]'s to keep only the most recent ones, then write corresponding entries to the Ethereum mapping.
595    ///
596    /// When `compare_timestamps` is `true`, existing entries are only overwritten by strictly
597    /// newer ones (see [`Self::put_mapping_if_newer`]). The live head indexer passes `false` to
598    /// keep its blind-write fast path, while index backfill passes `true` so that historical
599    /// writes do not clobber newer mappings written concurrently by the head indexer.
600    pub fn process_signed_messages(
601        &self,
602        messages: &[(SignedMessage, u64)],
603        compare_timestamps: bool,
604    ) -> anyhow::Result<()> {
605        let eth_txs: Vec<(EthHash, Cid, u64, usize)> = messages
606            .iter()
607            .enumerate()
608            .filter_map(|(i, (smsg, timestamp))| {
609                if let Ok((_, tx)) =
610                    eth_tx_from_signed_eth_message(smsg, self.chain_config.eth_chain_id)
611                {
612                    if let Ok(hash) = tx.eth_hash() {
613                        // newest messages are the ones with lowest index
614                        Some((hash.into(), smsg.cid(), *timestamp, i))
615                    } else {
616                        None
617                    }
618                } else {
619                    None
620                }
621            })
622            .collect();
623        let filtered = filter_lowest_index(eth_txs);
624        let num_entries = filtered.len();
625
626        // write back
627        for (k, v, timestamp) in filtered.into_iter() {
628            trace!("Insert mapping {} => {}", k, v);
629            if compare_timestamps {
630                self.put_mapping_if_newer(k, v, timestamp)?;
631            } else {
632                self.put_mapping(k, v, timestamp)?;
633            }
634        }
635        trace!("Wrote {} entries in Ethereum mapping", num_entries);
636        Ok(())
637    }
638
639    pub fn headers_delegated_messages<'a>(
640        &self,
641        headers: impl Iterator<Item = &'a CachingBlockHeader>,
642    ) -> anyhow::Result<Vec<(SignedMessage, u64)>> {
643        let mut delegated_messages = vec![];
644
645        // Hygge is the start of Ethereum support in the FVM (through the FEVM actor).
646        // Before this height, no notion of an Ethereum-like API existed.
647        let filtered_headers =
648            headers.filter(|bh| bh.epoch >= self.chain_config.epoch(Height::Hygge));
649
650        for bh in filtered_headers {
651            if let Ok((_, secp_cids)) = block_messages(self.db(), bh) {
652                let mut messages: Vec<_> = secp_cids
653                    .into_iter()
654                    .filter(|msg| msg.is_delegated())
655                    .map(|m| (m, bh.timestamp))
656                    .collect();
657                delegated_messages.append(&mut messages);
658            }
659        }
660
661        Ok(delegated_messages)
662    }
663}
664
665fn filter_lowest_index(values: Vec<(EthHash, Cid, u64, usize)>) -> Vec<(EthHash, Cid, u64)> {
666    let map: HashMap<EthHash, (Cid, u64, usize)> = values.into_iter().fold(
667        HashMap::default(),
668        |mut acc, (hash, cid, timestamp, index)| {
669            acc.entry(hash)
670                .and_modify(|&mut (_, _, ref mut min_index)| {
671                    if index < *min_index {
672                        *min_index = index;
673                    }
674                })
675                .or_insert((cid, timestamp, index));
676            acc
677        },
678    );
679
680    map.into_iter()
681        .map(|(hash, (cid, timestamp, _))| (hash, cid, timestamp))
682        .collect()
683}
684
685/// Returns a Tuple of BLS messages of type `UnsignedMessage` and SECP messages
686/// of type `SignedMessage`
687pub fn block_messages<DB>(
688    db: &DB,
689    bh: &CachingBlockHeader,
690) -> Result<(Vec<Message>, Vec<SignedMessage>), Error>
691where
692    DB: Blockstore,
693{
694    let (bls_cids, secpk_cids) = read_msg_cids(db, bh)?;
695
696    let bls_msgs: Vec<Message> = messages_from_cids(db, &bls_cids)?;
697    let secp_msgs: Vec<SignedMessage> = messages_from_cids(db, &secpk_cids)?;
698
699    Ok((bls_msgs, secp_msgs))
700}
701
702/// Returns a tuple of `UnsignedMessage` and `SignedMessages` from their CID
703pub fn block_messages_from_cids<DB>(
704    db: &DB,
705    bls_cids: &[Cid],
706    secp_cids: &[Cid],
707) -> Result<(Vec<Message>, Vec<SignedMessage>), Error>
708where
709    DB: Blockstore,
710{
711    let bls_msgs: Vec<Message> = messages_from_cids(db, bls_cids)?;
712    let secp_msgs: Vec<SignedMessage> = messages_from_cids(db, secp_cids)?;
713
714    Ok((bls_msgs, secp_msgs))
715}
716
717/// Returns a tuple of CIDs for both unsigned and signed messages
718pub fn read_msg_cids<DB>(
719    db: &DB,
720    block_header: &CachingBlockHeader,
721) -> Result<(Vec<Cid>, Vec<Cid>), Error>
722where
723    DB: Blockstore,
724{
725    let msg_cid = &block_header.messages;
726    if let Some(roots) = db.get_cbor::<TxMeta>(msg_cid)? {
727        let bls_cids = read_amt_cids(db, &roots.bls_message_root)?;
728        let secpk_cids = read_amt_cids(db, &roots.secp_message_root)?;
729        Ok((bls_cids, secpk_cids))
730    } else {
731        Err(Error::UndefinedKey(format!(
732            "no msg root with cid {msg_cid} at epoch {} in block {}",
733            block_header.epoch,
734            block_header.cid(),
735        )))
736    }
737}
738
739/// Persists slice of `serializable` objects to `blockstore`.
740pub fn persist_objects<'a, DB, C>(
741    db: &DB,
742    headers: impl Iterator<Item = &'a C>,
743) -> Result<(), Error>
744where
745    DB: Blockstore,
746    C: 'a + Serialize,
747{
748    for chunk in &headers.chunks(256) {
749        db.bulk_put(chunk, DB::default_code())?;
750    }
751    Ok(())
752}
753
754/// Returns a vector of CIDs from provided root CID
755fn read_amt_cids<DB>(db: &DB, root: &Cid) -> Result<Vec<Cid>, Error>
756where
757    DB: Blockstore,
758{
759    let amt = Amt::<Cid, _>::load(root, db)?;
760
761    let mut cids = Vec::with_capacity(amt.count() as usize);
762    amt.for_each_cacheless(|_, c| {
763        cids.push(*c);
764        Ok(())
765    })?;
766
767    Ok(cids)
768}
769
770/// Attempts to de-serialize to unsigned message or signed message and then
771/// returns it as a [`ChainMessage`].
772pub fn get_chain_message<DB>(db: &DB, key: &Cid) -> Result<ChainMessage, Error>
773where
774    DB: Blockstore,
775{
776    db.get_cbor(key)?
777        .ok_or_else(|| Error::UndefinedKey(key.to_string()))
778}
779
780/// A cache structure mapping tipset keys to messages. The regular [`messages_for_tipset`], based
781/// on performance measurements, is resource-intensive and can be a bottleneck for certain
782/// use-cases. This cache is intended to be used with a complementary function;
783/// [`messages_for_tipset_with_cache`].
784#[derive(derive_more::Deref)]
785pub struct MessagesInTipsetCache(SizeTrackingCache<TipsetKey, Arc<Vec<ChainMessage>>>);
786
787impl MessagesInTipsetCache {
788    pub fn new(capacity: NonZeroUsize) -> Self {
789        Self(SizeTrackingCache::new_with_metrics(
790            "msg_in_tipset",
791            capacity,
792        ))
793    }
794
795    /// Reads the intended cache size for this process from the environment or uses the default.
796    fn read_cache_size() -> NonZeroUsize {
797        // Arbitrary number, can be adjusted
798        const DEFAULT: NonZeroUsize = nonzero!(8192usize); // maximum ~40MiB on mainnet
799        std::env::var("FOREST_MESSAGES_IN_TIPSET_CACHE_SIZE")
800            .ok()
801            .and_then(|s| s.parse().ok())
802            .unwrap_or(DEFAULT)
803    }
804}
805
806impl Default for MessagesInTipsetCache {
807    fn default() -> Self {
808        Self::new(Self::read_cache_size())
809    }
810}
811
812impl ShallowClone for MessagesInTipsetCache {
813    fn shallow_clone(&self) -> Self {
814        Self(self.deref().shallow_clone())
815    }
816}
817
818/// Returns messages from key-value store based on a slice of [`Cid`]s.
819pub fn messages_from_cids<DB, T>(db: &DB, keys: &[Cid]) -> Result<Vec<T>, Error>
820where
821    DB: Blockstore,
822    T: DeserializeOwned,
823{
824    keys.iter().map(|k| message_from_cid(db, k)).collect()
825}
826
827/// Returns message from key-value store based on a [`Cid`].
828pub fn message_from_cid<DB, T>(db: &DB, key: &Cid) -> Result<T, Error>
829where
830    DB: Blockstore,
831    T: DeserializeOwned,
832{
833    db.get_cbor(key)?
834        .ok_or_else(|| Error::UndefinedKey(key.to_string()))
835}
836
837/// Returns parent message receipt given `block_header` and message index.
838pub fn get_parent_receipt(
839    db: &impl Blockstore,
840    block_header: &CachingBlockHeader,
841    i: usize,
842) -> Result<Option<Receipt>, Error> {
843    Ok(Receipt::get_receipt(
844        db,
845        &block_header.message_receipts,
846        i as u64,
847    )?)
848}
849
850#[cfg(test)]
851mod tests {
852    use super::*;
853    use crate::utils::multihash::prelude::*;
854    use crate::{blocks::RawBlockHeader, shim::address::Address};
855    use fvm_ipld_encoding::DAG_CBOR;
856
857    #[test]
858    fn genesis_test() {
859        let db = Arc::new(crate::db::MemoryDB::default());
860        let chain_config = Arc::new(ChainConfig::default());
861
862        let gen_block = CachingBlockHeader::new(RawBlockHeader {
863            miner_address: Address::new_id(0),
864            state_root: Cid::new_v1(DAG_CBOR, MultihashCode::Identity.digest(&[])),
865            epoch: 0,
866            weight: 2u32.into(),
867            messages: Cid::new_v1(DAG_CBOR, MultihashCode::Identity.digest(&[])),
868            message_receipts: Cid::new_v1(DAG_CBOR, MultihashCode::Identity.digest(&[])),
869            ..Default::default()
870        });
871        let gen_ts = Tipset::from(&gen_block);
872        let cs = ChainStore::new(db, chain_config, gen_ts.shallow_clone()).unwrap();
873
874        assert_eq!(cs.genesis_tipset(), gen_ts);
875        assert_eq!(cs.genesis_block_header(), &gen_block);
876    }
877
878    #[test]
879    fn block_validation_cache_basic() {
880        let db = DbImpl::from(Arc::new(crate::db::MemoryDB::default()));
881        let chain_config = Arc::new(ChainConfig::default());
882        let gen_block = CachingBlockHeader::new(RawBlockHeader {
883            miner_address: Address::new_id(0),
884            ..Default::default()
885        });
886
887        let cs = ChainStore::new(db, chain_config, gen_block).unwrap();
888
889        let cid = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[1, 2, 3]));
890        assert!(!cs.is_block_validated(&cid));
891
892        cs.mark_block_as_validated(&cid);
893        assert!(cs.is_block_validated(&cid));
894    }
895
896    #[test]
897    fn put_mapping_if_newer_keeps_newest() {
898        let db = DbImpl::from(Arc::new(crate::db::MemoryDB::default()));
899        let chain_config = Arc::new(ChainConfig::default());
900        let gen_block = CachingBlockHeader::new(RawBlockHeader {
901            miner_address: Address::new_id(0),
902            ..Default::default()
903        });
904        let cs = ChainStore::new(db, chain_config, gen_block).unwrap();
905
906        let hash = EthHash::default();
907        let older = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[1]));
908        let newer = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[2]));
909
910        // Seed with the newer (higher timestamp) entry, as the live head indexer would.
911        cs.put_mapping(hash, newer, 100).unwrap();
912
913        // A backfill writing an older (lower timestamp) entry must not clobber it.
914        cs.put_mapping_if_newer(hash, older, 50).unwrap();
915        assert_eq!(cs.get_mapping(&hash).unwrap(), Some(newer));
916
917        // An equal timestamp must also not overwrite.
918        cs.put_mapping_if_newer(hash, older, 100).unwrap();
919        assert_eq!(cs.get_mapping(&hash).unwrap(), Some(newer));
920
921        // A strictly newer entry wins.
922        let newest = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[3]));
923        cs.put_mapping_if_newer(hash, newest, 200).unwrap();
924        assert_eq!(cs.get_mapping(&hash).unwrap(), Some(newest));
925
926        // Writing into an empty slot always succeeds.
927        let fresh_hash = EthHash(ethereum_types::H256::repeat_byte(0xab));
928        cs.put_mapping_if_newer(fresh_hash, older, 1).unwrap();
929        assert_eq!(cs.get_mapping(&fresh_hash).unwrap(), Some(older));
930    }
931
932    #[test]
933    fn test_messages_in_tipset_cache() {
934        let cache = MessagesInTipsetCache::new(nonzero!(2_usize));
935        let key1 = TipsetKey::from(nunny::vec![Cid::new_v1(
936            DAG_CBOR,
937            MultihashCode::Blake2b256.digest(&[1])
938        )]);
939        assert!(cache.get(&key1).is_none());
940
941        let msgs = Arc::new(vec![Message::default().into()]);
942        cache.insert(key1.clone(), msgs.clone());
943        assert_eq!(&msgs, &cache.get(&key1).unwrap());
944
945        let inserter_executed: std::sync::atomic::AtomicBool =
946            std::sync::atomic::AtomicBool::new(false);
947        let key_inserter = || {
948            inserter_executed.store(true, std::sync::atomic::Ordering::Relaxed);
949            anyhow::Ok(msgs.clone())
950        };
951
952        assert_eq!(
953            &msgs,
954            &cache.get_or_insert_with(&key1, key_inserter).unwrap()
955        );
956        assert!(!inserter_executed.load(std::sync::atomic::Ordering::Relaxed));
957
958        let key2 = TipsetKey::from(nunny::vec![Cid::new_v1(
959            DAG_CBOR,
960            MultihashCode::Blake2b256.digest(&[2])
961        )]);
962
963        assert!(cache.get(&key2).is_none());
964        assert_eq!(
965            &msgs,
966            &cache.get_or_insert_with(&key2, key_inserter).unwrap()
967        );
968        assert!(inserter_executed.load(std::sync::atomic::Ordering::Relaxed));
969    }
970}
forest/chain/store/chain_store.rs

forest/chain/store/
chain_store.rs
