forest/state_manager/
message_search.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use super::*;
5use crate::blocks::TipsetKey;
6use crate::message::MessageRead as _;
7use ahash::HashSet;
8use parking_lot::RwLock;
9use std::sync::OnceLock;
10use std::time::Duration;
11use tokio::sync::broadcast::error::RecvError;
12use tokio_util::sync::CancellationToken;
13use tracing::warn;
14
15impl StateManager {
16    /// Check if tipset had executed the message, by loading the receipt based
17    /// on the index of the message in the block.
18    fn tipset_executed_message(
19        &self,
20        tipset: &Tipset,
21        message: &ChainMessage,
22        allow_replaced: bool,
23    ) -> Result<Option<Receipt>, Error> {
24        if tipset.epoch() == 0 {
25            return Ok(None);
26        }
27        let message_from_address = message.from();
28        let message_sequence = message.sequence();
29        // Load parent state.
30        let pts = self
31            .chain_index()
32            .load_required_tipset(tipset.parents())
33            .map_err(|err| Error::Other(format!("Failed to load tipset: {err}")))?;
34        let messages = self
35            .cs
36            .messages_for_tipset(&pts)
37            .map_err(|err| Error::Other(format!("Failed to load messages for tipset: {err}")))?;
38        messages
39            .iter()
40            .enumerate()
41            // iterate in reverse because we going backwards through the chain
42            .rev()
43            .filter(|(_, s)| {
44                s.sequence() == message_sequence
45                    && s.from() == message_from_address
46                    && s.equal_call(message)
47            })
48            .map(|(index, m)| {
49                // A replacing message is a message with a different CID,
50                // any of Gas values, and different signature, but with all
51                // other parameters matching (source/destination, nonce, params, etc.)
52                if !allow_replaced && message.cid() != m.cid(){
53                    Err(Error::Other(format!(
54                        "found message with equal nonce and call params but different CID. wanted {}, found: {}, nonce: {}, from: {}",
55                        message.cid(),
56                        m.cid(),
57                        message.sequence(),
58                        message.from(),
59                    )))
60                } else {
61                    let block_header = tipset.block_headers().first();
62                    crate::chain::get_parent_receipt(
63                        self.db(),
64                        block_header,
65                        index,
66                    )
67                        .map_err(|err| Error::Other(format!("Failed to get parent receipt (message_receipts={}, index={index}, error={err})", block_header.message_receipts)))
68                }
69            })
70            .next()
71            .unwrap_or(Ok(None))
72    }
73
74    fn check_search_blocking(
75        &self,
76        mut current: Tipset,
77        message: &ChainMessage,
78        lookback_max_epoch: ChainEpoch,
79        allow_replaced: bool,
80        cancellation_token: &CancellationToken,
81    ) -> Result<Option<(Tipset, Receipt)>, Error> {
82        let message_from_address = message.from();
83        let message_sequence = message.sequence();
84        let current_actor_state = self
85            .get_required_actor(&message_from_address, *current.parent_state())
86            .map_err(Error::state)?;
87        // The sender's nonce only grows, so once it is at or below the message
88        // nonce the message cannot have been executed yet. Walking back would
89        // only end at the sender's creation or at pruned state.
90        if current_actor_state.sequence <= message_sequence {
91            return Ok(None);
92        }
93        let message_from_id = self.lookup_required_id(&message_from_address, &current)?;
94
95        while !cancellation_token.is_cancelled() && current.epoch() >= lookback_max_epoch {
96            let parent_tipset = self
97                .chain_index()
98                .load_required_tipset(current.parents())
99                .map_err(|err| {
100                    Error::Other(format!(
101                        "failed to load tipset during msg wait searchback: {err:}"
102                    ))
103                })?;
104
105            let parent_actor_state = self
106                .get_actor(&message_from_id, *parent_tipset.parent_state())
107                .map_err(|e| Error::State(e.to_string()))?;
108
109            match parent_actor_state {
110                // The nonce is still above the message nonce at the parent, so
111                // the message executed strictly earlier; keep walking back.
112                Some(state) if state.sequence > message_sequence => current = parent_tipset,
113                // The nonce crossed the message nonce between the parent and
114                // `current` (or the sender did not exist yet), so only `current`
115                // can have executed the message. No receipt there means a
116                // replacing message was executed instead.
117                _ => {
118                    return Ok(self
119                        .tipset_executed_message(&current, message, allow_replaced)?
120                        .map(|receipt| (current, receipt)));
121                }
122            }
123        }
124
125        Ok(None)
126    }
127
128    /// Searches backwards through the chain for a message receipt.
129    fn search_back_for_message_blocking(
130        &self,
131        current: Tipset,
132        message: &ChainMessage,
133        look_back_limit: Option<ChainEpoch>,
134        allow_replaced: Option<bool>,
135        cancellation_token: &CancellationToken,
136    ) -> Result<Option<(Tipset, Receipt)>, Error> {
137        let current_epoch = current.epoch();
138        let allow_replaced = allow_replaced.unwrap_or(true);
139
140        let Some(max_lookback_epoch_inclusive) =
141            Self::max_lookback_epoch_inclusive(current_epoch, look_back_limit)
142        else {
143            return Ok(None);
144        };
145
146        self.check_search_blocking(
147            current,
148            message,
149            max_lookback_epoch_inclusive,
150            allow_replaced,
151            cancellation_token,
152        )
153    }
154
155    //. Calculates the max lookback epoch (inclusive lower bound) for the search.
156    pub fn max_lookback_epoch_inclusive(
157        current_epoch: ChainEpoch,
158        look_back_limit: Option<ChainEpoch>,
159    ) -> Option<ChainEpoch> {
160        match look_back_limit {
161            // No search: limit = 0 means search 0 epochs
162            Some(0) => None,
163            // Limited search: calculate the inclusive lower bound, clamped to genesis
164            // Example: limit=5 at epoch=1000 → min_epoch=996, searches [996,1000] = 5 epochs
165            // Example: limit=2000 at epoch=1000 → min_epoch=0, searches [0,1000] = 1001 epochs (all available)
166            Some(limit) if limit > 0 => Some((current_epoch - limit + 1).max(0)),
167            // Search all the way to genesis (epoch 0)
168            _ => Some(0),
169        }
170    }
171
172    /// Returns a message receipt from a given tipset and message CID.
173    pub fn get_receipt_blocking(
174        &self,
175        tipset: Tipset,
176        msg: Cid,
177        cancellation_token: &CancellationToken,
178    ) -> Result<Receipt, Error> {
179        let m = crate::chain::get_chain_message(self.db(), &msg)
180            .map_err(|e| Error::Other(e.to_string()))?;
181        let message_receipt = self.tipset_executed_message(&tipset, &m, true)?;
182        if let Some(receipt) = message_receipt {
183            return Ok(receipt);
184        }
185
186        let maybe_tuple =
187            self.search_back_for_message_blocking(tipset, &m, None, None, cancellation_token)?;
188        let message_receipt = maybe_tuple
189            .ok_or_else(|| {
190                Error::Other("Could not get receipt from search back message".to_string())
191            })?
192            .1;
193        Ok(message_receipt)
194    }
195
196    pub async fn wait_for_message_with_timeout(
197        &self,
198        msg_cid: Cid,
199        confidence: i64,
200        look_back_limit: Option<ChainEpoch>,
201        allow_replaced: Option<bool>,
202        timeout: Duration,
203    ) -> Result<(Tipset, Receipt), Error> {
204        let cancellation_token = CancellationToken::new();
205        let _cancellation_token_drop_guard = cancellation_token.drop_guard_ref();
206        tokio::time::timeout(
207            timeout,
208            self.wait_for_message(
209                msg_cid,
210                confidence,
211                look_back_limit,
212                allow_replaced,
213                &cancellation_token,
214            ),
215        )
216        .await
217        .map_err(|_| {
218            Error::other(format!(
219                "wait_for_message timed out after {}",
220                humantime::format_duration(timeout)
221            ))
222        })?
223    }
224
225    /// `WaitForMessage` blocks until a message appears on chain. It looks
226    /// backwards in the chain to see if this has already happened. It
227    /// guarantees that the message has been on chain for at least
228    /// confidence epochs without being reverted before returning.
229    /// Returns an error when cancelled.
230    pub async fn wait_for_message(
231        &self,
232        msg_cid: Cid,
233        confidence: i64,
234        look_back_limit: Option<ChainEpoch>,
235        allow_replaced: Option<bool>,
236        cancellation_token: &CancellationToken,
237    ) -> Result<(Tipset, Receipt), Error> {
238        let message = Arc::new(
239            crate::chain::get_chain_message(self.db(), &msg_cid)
240                .map_err(|err| Error::Other(format!("failed to load message {err:}")))?,
241        );
242        // Subscribe to head changes before sampling the head so that a reorg
243        // between sampling and subscribing cannot be missed. Otherwise a revert
244        // of the sampled head could go unseen and a reverted receipt could be
245        // released after `confidence` epochs.
246        let mut head_changes_rx = self.cs.subscribe_head_changes();
247        let current_ts = self.heaviest_tipset();
248        let maybe_message_receipt =
249            self.tipset_executed_message(&current_ts, &message, allow_replaced.unwrap_or(true))?;
250        // If the message already executed at the current head, return right
251        // away only when no confidence is required; otherwise seed it as a
252        // candidate for the head-change loop to release after `confidence` epochs.
253        let initial_candidate = match maybe_message_receipt {
254            Some(receipt) if confidence == 0 => return Ok((current_ts, receipt)),
255            Some(receipt) => Some((current_ts.shallow_clone(), receipt)),
256            None => None,
257        };
258
259        // For immediate search back response
260        let (search_back_tx, search_back_rx) = flume::bounded(1);
261        let search_back_candidate: Arc<OnceLock<(Tipset, Receipt)>> = Default::default();
262        let reverted: Arc<RwLock<HashSet<TipsetKey>>> = Arc::new(RwLock::new(HashSet::default()));
263        // Search back task
264        tokio::task::spawn_blocking({
265            let sm = self.shallow_clone();
266            let message = message.shallow_clone();
267            // Cloning tx to avoid all senders being dropped to make `search_back_rx.recv_async()` wait
268            let search_back_tx = search_back_tx.clone();
269            let search_back_candidate = search_back_candidate.shallow_clone();
270            let reverted = reverted.shallow_clone();
271            let cancellation_token = cancellation_token.clone();
272            move || {
273                if let Ok(Some((ts, receipt))) = sm
274                    .search_back_for_message_blocking(
275                        current_ts,
276                        &message,
277                        look_back_limit,
278                        allow_replaced,
279                        &cancellation_token,
280                    )
281                    .inspect_err(|e| {
282                        tracing::warn!("failed to search back for message: {e}");
283                    })
284                    && !reverted.read().contains(ts.key())
285                {
286                    if sm.heaviest_tipset().epoch() >= ts.epoch() + confidence {
287                        _ = search_back_tx.send((ts, receipt)).inspect_err(|e| {
288                            tracing::warn!("failed to send to search_back_tx: {e}");
289                        });
290                    } else {
291                        _ = search_back_candidate.set((ts, receipt)).inspect_err(|_| {
292                            tracing::warn!("failed to send to set search_back_candidate");
293                        });
294                    }
295                }
296            }
297        });
298
299        // Wait for message to be included in head change.
300        let subscriber_poll = tokio::task::spawn({
301            let cancellation_token = cancellation_token.clone();
302            let search_back_candidate = search_back_candidate.shallow_clone();
303            let reverted = reverted.shallow_clone();
304            let sm = self.shallow_clone();
305            async move {
306                let mut candidate: Option<(Tipset, Receipt)> = initial_candidate;
307                while !cancellation_token.is_cancelled() {
308                    match head_changes_rx.recv().await {
309                        Ok(head_changes) => {
310                            for reverted_ts in head_changes.reverts {
311                                reverted.write().insert(reverted_ts.key().clone());
312
313                                if candidate
314                                    .as_ref()
315                                    .is_some_and(|(ts, _)| ts.key() == reverted_ts.key())
316                                {
317                                    candidate = None;
318                                }
319                            }
320                            for applied_ts in head_changes.applies {
321                                reverted.write().remove(applied_ts.key());
322
323                                // Return if `search_back_candidate` meets confidence requirement
324                                if let Some((candidate_ts, candidate_receipt)) =
325                                    search_back_candidate.get()
326                                    && applied_ts.epoch() >= candidate_ts.epoch() + confidence
327                                    && !reverted.read().contains(candidate_ts.key())
328                                {
329                                    return Ok((
330                                        candidate_ts.shallow_clone(),
331                                        candidate_receipt.clone(),
332                                    ));
333                                }
334
335                                // Return if the candidate meets confidence requirement
336                                if let Some((candidate_ts, _)) = &candidate
337                                    && applied_ts.epoch() >= candidate_ts.epoch() + confidence
338                                    && let Some(candidate) = candidate
339                                {
340                                    return Ok(candidate);
341                                }
342
343                                let maybe_receipt = sm.tipset_executed_message(
344                                    &applied_ts,
345                                    &message,
346                                    allow_replaced.unwrap_or(true),
347                                )?;
348                                if let Some(receipt) = maybe_receipt {
349                                    if confidence == 0 {
350                                        // Return if there's no confidence requirement
351                                        return Ok((applied_ts, receipt));
352                                    } else {
353                                        // Otherwise set it as candidate
354                                        candidate = Some((applied_ts, receipt));
355                                    }
356                                }
357                            }
358                        }
359                        Err(RecvError::Lagged(i)) => {
360                            warn!(
361                                "wait for message head change subscriber lagged, skipped {} events",
362                                i
363                            );
364                        }
365                        Err(RecvError::Closed) => break,
366                    }
367                }
368                Err(Error::other("cancelled"))
369            }
370        });
371
372        // Await on first future to finish.
373        tokio::select! {
374            res = subscriber_poll => {
375                res.context("tokio join error")?
376            }
377            res = search_back_rx.recv_async()  => {
378                Ok(res.context("channel receive error")?)
379            }
380            _ = cancellation_token.cancelled() => {
381                Err(Error::other("cancelled"))
382            }
383        }
384    }
385
386    pub async fn search_for_message(
387        &self,
388        from: Option<Tipset>,
389        msg_cid: Cid,
390        look_back_limit: Option<i64>,
391        allow_replaced: Option<bool>,
392        cancellation_token: &CancellationToken,
393    ) -> Result<Option<(Tipset, Receipt)>, Error> {
394        let from = from.unwrap_or_else(|| self.heaviest_tipset());
395        let message = crate::chain::get_chain_message(self.db(), &msg_cid)
396            .map_err(|err| Error::Other(format!("failed to load message {err}")))?;
397        let maybe_message_receipt =
398            self.tipset_executed_message(&from, &message, allow_replaced.unwrap_or(true))?;
399        if let Some(r) = maybe_message_receipt {
400            Ok(Some((from, r)))
401        } else {
402            tokio::task::spawn_blocking({
403                let this = self.shallow_clone();
404                let cancellation_token = cancellation_token.clone();
405                move || {
406                    this.search_back_for_message_blocking(
407                        from,
408                        &message,
409                        look_back_limit,
410                        allow_replaced,
411                        &cancellation_token,
412                    )
413                }
414            })
415            .await?
416        }
417    }
418}
419
420#[cfg(test)]
421mod tests {
422    use super::*;
423    use crate::blocks::{
424        CachingBlockHeader, Chain4U, HeaderBuilder, RawBlockHeader, TxMeta, chain4u,
425    };
426    use crate::chain::ChainStore;
427    use crate::db::MemoryDB;
428    use crate::networks::ChainConfig;
429    use crate::shim::address::Address;
430    use crate::shim::econ::TokenAmount;
431    use crate::shim::message::Message;
432    use crate::shim::state_tree::{ActorState, StateTree, StateTreeVersion};
433    use crate::utils::db::CborStoreExt as _;
434    use fil_actors_shared::fvm_ipld_amt::Amtv0;
435    use fvm_ipld_blockstore::Blockstore;
436
437    const SENDER: Address = Address::new_id(100);
438
439    fn state_root_with_sender_nonce(db: &Arc<MemoryDB>, sequence: u64) -> Cid {
440        let mut state_tree = StateTree::new(db, StateTreeVersion::V5).unwrap();
441        state_tree
442            .set_actor(
443                &SENDER,
444                ActorState::new(
445                    Cid::default(),
446                    Cid::default(),
447                    TokenAmount::default(),
448                    sequence,
449                    None,
450                ),
451            )
452            .unwrap();
453        state_tree.flush().unwrap()
454    }
455
456    fn tx_meta(db: &impl Blockstore, message: Cid) -> Cid {
457        let bls_message_root = Amtv0::new_from_iter(db, [message]).unwrap();
458        let secp_message_root = Amtv0::new_from_iter(db, std::iter::empty::<Cid>()).unwrap();
459        db.put_cbor_default(&TxMeta {
460            bls_message_root,
461            secp_message_root,
462        })
463        .unwrap()
464    }
465
466    fn receipts_root(db: &impl Blockstore) -> Cid {
467        let receipt = fvm_shared4::receipt::Receipt {
468            exit_code: fvm_shared4::error::ExitCode::OK,
469            return_data: Default::default(),
470            gas_used: 0,
471            events_root: None,
472        };
473        Amtv0::new_from_iter(db, [receipt]).unwrap()
474    }
475
476    fn message_with_nonce(sequence: u64) -> Message {
477        Message {
478            from: SENDER,
479            to: Address::new_id(101),
480            sequence,
481            ..Default::default()
482        }
483    }
484
485    fn state_manager_with_replaced_message_at_head(db: &Arc<MemoryDB>) -> (StateManager, Cid) {
486        let message = message_with_nonce(5);
487        let msg_cid = db.put_cbor_default(&message).unwrap();
488        let replacement = Message {
489            gas_limit: 1,
490            ..message
491        };
492        let replacement_cid = db.put_cbor_default(&replacement).unwrap();
493
494        let root_before = state_root_with_sender_nonce(db, 5);
495        let root_after = state_root_with_sender_nonce(db, 6);
496        let messages = tx_meta(db, replacement_cid);
497        let receipts = receipts_root(db);
498        let c4u = Chain4U::with_blockstore(db.clone());
499        chain4u! {
500            in c4u;
501            [genesis = HeaderBuilder::new().with_timestamp(7777)]
502            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
503            -> [_e2 = HeaderBuilder::new()
504                    .with_state_root(root_before)
505                    .with_messages(messages)]
506            -> head @ [_e3 = HeaderBuilder::new()
507                    .with_state_root(root_after)
508                    .with_message_receipts(receipts)]
509        };
510        (state_manager_with_head(db.clone(), genesis, head), msg_cid)
511    }
512
513    fn state_manager_with_head(
514        db: Arc<MemoryDB>,
515        genesis: &RawBlockHeader,
516        head: &Tipset,
517    ) -> StateManager {
518        let chain_store = ChainStore::new(
519            db,
520            Arc::new(ChainConfig::default()),
521            CachingBlockHeader::new(genesis.clone()),
522        )
523        .unwrap();
524        chain_store.set_heaviest_tipset(head.clone()).unwrap();
525        StateManager::new(chain_store).unwrap()
526    }
527
528    /// Chain where the sender's nonce is `actor_nonce` at every epoch and the
529    /// genesis state is unavailable, like state pruned by GC. Searching must
530    /// not walk into the missing state.
531    async fn search_pending(
532        actor_nonce: u64,
533        message_nonce: u64,
534    ) -> Result<Option<(Tipset, Receipt)>, Error> {
535        let db = Arc::new(MemoryDB::default());
536        let root = state_root_with_sender_nonce(&db, actor_nonce);
537        let c4u = Chain4U::with_blockstore(db.clone());
538        chain4u! {
539            in c4u;
540            [genesis = HeaderBuilder::new().with_timestamp(7777)]
541            -> [_e1 = HeaderBuilder::new().with_state_root(root)]
542            -> [_e2 = HeaderBuilder::new().with_state_root(root)]
543            -> head @ [_e3 = HeaderBuilder::new().with_state_root(root)]
544        };
545        let state_manager = state_manager_with_head(db.clone(), genesis, head);
546
547        let message = message_with_nonce(message_nonce);
548        let msg_cid = db.put_cbor_default(&message).unwrap();
549
550        state_manager
551            .search_for_message(None, msg_cid, None, Some(true), &CancellationToken::new())
552            .await
553    }
554
555    #[tokio::test]
556    async fn search_returns_none_for_message_with_future_nonce() {
557        let result = search_pending(5, 10).await.unwrap();
558        assert!(result.is_none());
559    }
560
561    #[tokio::test]
562    async fn search_returns_none_for_pending_message_at_current_nonce() {
563        let result = search_pending(5, 5).await.unwrap();
564        assert!(result.is_none());
565    }
566
567    #[tokio::test]
568    async fn search_returns_none_for_pending_message_from_fresh_sender() {
569        let result = search_pending(0, 0).await.unwrap();
570        assert!(result.is_none());
571    }
572
573    /// The sender's nonce crossed the message nonce, but a replacing message
574    /// with a different call executed instead: the searched message was never
575    /// executed, so the result is `None`, not an error.
576    #[tokio::test]
577    async fn search_returns_none_for_replaced_message() {
578        let db = Arc::new(MemoryDB::default());
579        let message = message_with_nonce(5);
580        let msg_cid = db.put_cbor_default(&message).unwrap();
581
582        let root_before = state_root_with_sender_nonce(&db, 5);
583        let root_after = state_root_with_sender_nonce(&db, 6);
584        let c4u = Chain4U::with_blockstore(db.clone());
585        chain4u! {
586            in c4u;
587            [genesis = HeaderBuilder::new().with_timestamp(7777)]
588            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
589            -> [_e2 = HeaderBuilder::new().with_state_root(root_before)]
590            -> [_e3 = HeaderBuilder::new().with_state_root(root_after)]
591            -> head @ [_e4 = HeaderBuilder::new().with_state_root(root_after)]
592        };
593        let state_manager = state_manager_with_head(db.clone(), genesis, head);
594
595        let result = state_manager
596            .search_for_message(None, msg_cid, None, Some(true), &CancellationToken::new())
597            .await
598            .unwrap();
599        assert!(result.is_none());
600    }
601
602    #[tokio::test]
603    async fn search_finds_executed_message() {
604        let db = Arc::new(MemoryDB::default());
605        let message = message_with_nonce(5);
606        let msg_cid = db.put_cbor_default(&message).unwrap();
607
608        let root_before = state_root_with_sender_nonce(&db, 5);
609        let root_after = state_root_with_sender_nonce(&db, 6);
610        let messages = tx_meta(&db, msg_cid);
611        let receipts = receipts_root(&db);
612        let c4u = Chain4U::with_blockstore(db.clone());
613        chain4u! {
614            in c4u;
615            [genesis = HeaderBuilder::new().with_timestamp(7777)]
616            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
617            -> [_e2 = HeaderBuilder::new()
618                    .with_state_root(root_before)
619                    .with_messages(messages)]
620            -> [_e3 = HeaderBuilder::new()
621                    .with_state_root(root_after)
622                    .with_message_receipts(receipts)]
623            -> head @ [_e4 = HeaderBuilder::new().with_state_root(root_after)]
624        };
625        let state_manager = state_manager_with_head(db.clone(), genesis, head);
626
627        let (tipset, receipt) = state_manager
628            .search_for_message(None, msg_cid, None, Some(true), &CancellationToken::new())
629            .await
630            .unwrap()
631            .expect("executed message should be found");
632        assert_eq!(tipset.epoch(), 3);
633        assert!(receipt.exit_code().is_success());
634    }
635
636    /// Searching from an explicit tipset only covers executions at or below
637    /// it, like in Lotus, even when the message executed later in the chain.
638    #[tokio::test]
639    async fn search_from_older_tipset_ignores_later_execution() {
640        let db = Arc::new(MemoryDB::default());
641        let message = message_with_nonce(5);
642        let msg_cid = db.put_cbor_default(&message).unwrap();
643
644        let root_before = state_root_with_sender_nonce(&db, 5);
645        let root_after = state_root_with_sender_nonce(&db, 6);
646        let messages = tx_meta(&db, msg_cid);
647        let receipts = receipts_root(&db);
648        let c4u = Chain4U::with_blockstore(db.clone());
649        chain4u! {
650            in c4u;
651            [genesis = HeaderBuilder::new().with_timestamp(7777)]
652            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
653            -> from @ [_e2 = HeaderBuilder::new()
654                    .with_state_root(root_before)
655                    .with_messages(messages)]
656            -> [_e3 = HeaderBuilder::new()
657                    .with_state_root(root_after)
658                    .with_message_receipts(receipts)]
659            -> head @ [_e4 = HeaderBuilder::new().with_state_root(root_after)]
660        };
661        let state_manager = state_manager_with_head(db.clone(), genesis, head);
662
663        let result = state_manager
664            .search_for_message(
665                Some(from.clone()),
666                msg_cid,
667                None,
668                Some(true),
669                &CancellationToken::new(),
670            )
671            .await
672            .unwrap();
673        assert!(result.is_none());
674    }
675
676    /// A replacing message executed on chain. When replacements are
677    /// disallowed, `wait_for_message` must forward `allow_replaced = false`
678    /// and surface an error instead of silently returning the replacement's
679    /// receipt.
680    #[tokio::test]
681    async fn wait_for_message_rejects_replaced_when_disallowed() {
682        let db = Arc::new(MemoryDB::default());
683        let (state_manager, msg_cid) = state_manager_with_replaced_message_at_head(&db);
684
685        let result = state_manager
686            .wait_for_message(msg_cid, 0, None, Some(false), &CancellationToken::new())
687            .await;
688        let err = result.expect_err("replaced message should be rejected");
689        assert!(err.to_string().contains("different CID"), "{err}");
690    }
691
692    /// The same replacing message is accepted when replacements are allowed,
693    /// returning the receipt at the head tipset.
694    #[tokio::test]
695    async fn wait_for_message_accepts_replaced_when_allowed() {
696        let db = Arc::new(MemoryDB::default());
697        let (state_manager, msg_cid) = state_manager_with_replaced_message_at_head(&db);
698
699        let (tipset, receipt) = state_manager
700            .wait_for_message(msg_cid, 0, None, Some(true), &CancellationToken::new())
701            .await
702            .expect("replaced message should be accepted");
703        assert_eq!(tipset.epoch(), 3);
704        assert!(receipt.exit_code().is_success());
705    }
706
707    /// The message executed as of the head tipset and no confidence is
708    /// required, so `wait_for_message` returns the head hit immediately.
709    #[tokio::test]
710    async fn wait_for_message_head_hit_returns_immediately_with_zero_confidence() {
711        let db = Arc::new(MemoryDB::default());
712        let message = message_with_nonce(5);
713        let msg_cid = db.put_cbor_default(&message).unwrap();
714
715        let root_before = state_root_with_sender_nonce(&db, 5);
716        let root_after = state_root_with_sender_nonce(&db, 6);
717        let messages = tx_meta(&db, msg_cid);
718        let receipts = receipts_root(&db);
719        let c4u = Chain4U::with_blockstore(db.clone());
720        chain4u! {
721            in c4u;
722            [genesis = HeaderBuilder::new().with_timestamp(7777)]
723            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
724            -> [_e2 = HeaderBuilder::new()
725                    .with_state_root(root_before)
726                    .with_messages(messages)]
727            -> head @ [_e3 = HeaderBuilder::new()
728                    .with_state_root(root_after)
729                    .with_message_receipts(receipts)]
730        };
731        let state_manager = state_manager_with_head(db.clone(), genesis, head);
732
733        let (tipset, receipt) = state_manager
734            .wait_for_message(msg_cid, 0, None, Some(true), &CancellationToken::new())
735            .await
736            .expect("head hit should return immediately with zero confidence");
737        assert_eq!(tipset.epoch(), 3);
738        assert!(receipt.exit_code().is_success());
739    }
740
741    /// The message executed as of the head tipset, but a positive confidence is
742    /// requested and the head never advances, so the head hit must not be
743    /// returned early; the call times out instead.
744    #[tokio::test]
745    async fn wait_for_message_head_hit_waits_for_confidence() {
746        let db = Arc::new(MemoryDB::default());
747        let message = message_with_nonce(5);
748        let msg_cid = db.put_cbor_default(&message).unwrap();
749
750        let root_before = state_root_with_sender_nonce(&db, 5);
751        let root_after = state_root_with_sender_nonce(&db, 6);
752        let messages = tx_meta(&db, msg_cid);
753        let receipts = receipts_root(&db);
754        let c4u = Chain4U::with_blockstore(db.clone());
755        chain4u! {
756            in c4u;
757            [genesis = HeaderBuilder::new().with_timestamp(7777)]
758            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
759            -> [_e2 = HeaderBuilder::new()
760                    .with_state_root(root_before)
761                    .with_messages(messages)]
762            -> head @ [_e3 = HeaderBuilder::new()
763                    .with_state_root(root_after)
764                    .with_message_receipts(receipts)]
765        };
766        let state_manager = state_manager_with_head(db.clone(), genesis, head);
767
768        let result = state_manager
769            .wait_for_message_with_timeout(msg_cid, 2, None, Some(true), Duration::from_millis(300))
770            .await;
771        assert!(result.is_err());
772    }
773
774    /// The message executed as of the head tipset with a positive confidence.
775    /// Once the head advances by `confidence` epochs without a revert, the head
776    /// hit is returned.
777    #[tokio::test]
778    async fn wait_for_message_head_hit_returns_after_confidence_reached() {
779        let db = Arc::new(MemoryDB::default());
780        let message = message_with_nonce(5);
781        let msg_cid = db.put_cbor_default(&message).unwrap();
782
783        let root_before = state_root_with_sender_nonce(&db, 5);
784        let root_after = state_root_with_sender_nonce(&db, 6);
785        let messages = tx_meta(&db, msg_cid);
786        let receipts = receipts_root(&db);
787        let c4u = Chain4U::with_blockstore(db.clone());
788        chain4u! {
789            in c4u;
790            [genesis = HeaderBuilder::new().with_timestamp(7777)]
791            -> [_e1 = HeaderBuilder::new().with_state_root(root_before)]
792            -> [_e2 = HeaderBuilder::new()
793                    .with_state_root(root_before)
794                    .with_messages(messages)]
795            -> exec @ [_e3 = HeaderBuilder::new()
796                    .with_state_root(root_after)
797                    .with_message_receipts(receipts)]
798            -> next4 @ [_e4 = HeaderBuilder::new().with_state_root(root_after)]
799            -> next5 @ [_e5 = HeaderBuilder::new().with_state_root(root_after)]
800        };
801        let state_manager = state_manager_with_head(db.clone(), genesis, exec);
802
803        let token = CancellationToken::new();
804        let mut wait =
805            Box::pin(state_manager.wait_for_message(msg_cid, 2, None, Some(true), &token));
806
807        // Poll while the head is at epoch 3 to seed the candidate and subscribe,
808        // then advance to epoch 4. Confidence 2 is unmet in both cases, so the
809        // future must stay pending.
810        for advance_to in [None, Some(next4)] {
811            if let Some(ts) = advance_to {
812                state_manager
813                    .chain_store()
814                    .set_heaviest_tipset(ts.clone())
815                    .unwrap();
816            }
817            assert!(
818                tokio::time::timeout(Duration::from_millis(100), &mut wait)
819                    .await
820                    .is_err(),
821                "confidence 2 must not be reached before epoch 5"
822            );
823        }
824
825        // Epoch 5 reaches confidence 2; the future resolves.
826        state_manager
827            .chain_store()
828            .set_heaviest_tipset(next5.clone())
829            .unwrap();
830        let (tipset, receipt) = tokio::time::timeout(Duration::from_secs(5), &mut wait)
831            .await
832            .expect("should resolve once confidence is reached")
833            .expect("head hit should be returned after confidence reached");
834        assert_eq!(tipset.epoch(), 3);
835        assert!(receipt.exit_code().is_success());
836    }
837}
forest/state_manager/message_search.rs

forest/state_manager/
message_search.rs
