1use super::circulating_supply::GenesisInfo;
5use super::*;
6use crate::interpreter::{BlockMessages, ExecutionContext, VM, VMTrace};
7use crate::prelude::*;
8use crate::shim::message::Message;
9use crate::state_migration::run_state_migrations;
10use anyhow::{bail, ensure};
11use fil_actors_shared::fvm_ipld_amt::{Amt, Amtv0};
12use tracing::{error, info, instrument, warn};
13
14enum StateRecomputePolicy {
15 Allowed,
16 Disallowed,
17}
18
19impl StateManager {
20 pub async fn load_tipset_state(&self, ts: &Tipset) -> anyhow::Result<TipsetState> {
22 if let Some(state) = self.cache.get_map(ts.key(), |et| et.into()) {
23 Ok(state)
24 } else {
25 match self.chain_store().load_child_tipset(ts).await? {
26 Some(receipt_ts) => Ok(TipsetState {
27 state_root: *receipt_ts.parent_state(),
28 receipt_root: *receipt_ts.parent_message_receipts(),
29 }),
30 None => Ok(self.load_executed_tipset(ts).await?.into()),
31 }
32 }
33 }
34
35 pub fn clear_tipset_state_caches(&self) {
39 self.cache.clear();
40 self.trace_cache.clear();
41 }
42
43 pub fn repair_tipset_lookup(&self) -> anyhow::Result<usize> {
47 let n_repaired = self.cs.repair_tipset_lookup()?;
48 if n_repaired > 0 {
49 self.clear_tipset_state_caches();
50 }
51 Ok(n_repaired)
52 }
53
54 fn rpc_state_recompute_policy() -> StateRecomputePolicy {
57 crate::def_is_env_truthy!(
58 enable_state_computation,
59 "FOREST_ETH_RPC_COMPUTE_STATE_ON_INDEX_MISS"
60 );
61
62 if enable_state_computation() {
63 StateRecomputePolicy::Allowed
64 } else {
65 StateRecomputePolicy::Disallowed
66 }
67 }
68
69 pub async fn load_executed_tipset_for_rpc(
71 &self,
72 ts: &Tipset,
73 ) -> anyhow::Result<ExecutedTipset> {
74 self.load_executed_tipset_with_cache(ts, Self::rpc_state_recompute_policy())
75 .await
76 }
77
78 pub async fn load_executed_tipset_with_receipt(
82 &self,
83 msg_ts: &Tipset,
84 receipt_ts: &Tipset,
85 ) -> anyhow::Result<ExecutedTipset> {
86 self.cache
87 .get_or_insert_async(msg_ts.key(), async move {
88 self.load_executed_tipset_inner(
89 msg_ts,
90 Some(receipt_ts),
91 Self::rpc_state_recompute_policy(),
92 )
93 .await
94 })
95 .await
96 }
97
98 pub async fn load_executed_tipset(&self, ts: &Tipset) -> anyhow::Result<ExecutedTipset> {
100 self.load_executed_tipset_with_cache(ts, StateRecomputePolicy::Allowed)
101 .await
102 }
103
104 pub async fn load_executed_tipset_uncached(
107 &self,
108 ts: &Tipset,
109 allow_state_compute: bool,
110 ) -> anyhow::Result<ExecutedTipset> {
111 let policy = if allow_state_compute {
112 StateRecomputePolicy::Allowed
113 } else {
114 StateRecomputePolicy::Disallowed
115 };
116 let receipt_ts = self.chain_store().load_child_tipset(ts).await?;
117 self.load_executed_tipset_inner(ts, receipt_ts.as_ref(), policy)
118 .await
119 }
120
121 async fn load_executed_tipset_with_cache(
122 &self,
123 ts: &Tipset,
124 policy: StateRecomputePolicy,
125 ) -> anyhow::Result<ExecutedTipset> {
126 if ts.epoch() >= self.heaviest_tipset().epoch()
128 && let Some(cached) = self.cache.get(ts.key())
129 {
130 if StateTree::new_from_root(self.db(), &cached.state_root).is_ok() {
131 return Ok(cached);
132 } else {
133 self.cache.remove(ts.key());
134 }
135 }
136 self.cache
137 .get_or_insert_async(ts.key(), async move {
138 let receipt_ts = self.chain_store().load_child_tipset(ts).await?;
139 self.load_executed_tipset_inner(ts, receipt_ts.as_ref(), policy)
140 .await
141 })
142 .await
143 }
144
145 async fn load_executed_tipset_inner(
146 &self,
147 msg_ts: &Tipset,
148 receipt_ts: Option<&Tipset>,
150 policy: StateRecomputePolicy,
151 ) -> anyhow::Result<ExecutedTipset> {
152 let state_compute_disallow_error = || {
153 format!(
154 "failed to load tipset state output and recomputation is disallowed, epoch={}, key={}",
155 msg_ts.epoch(),
156 msg_ts.key()
157 )
158 };
159
160 if let Some(receipt_ts) = receipt_ts {
161 anyhow::ensure!(
162 msg_ts.key() == receipt_ts.parents(),
163 "message tipset should be the parent of message receipt tipset"
164 );
165 }
166 let allow_state_compute = matches!(policy, StateRecomputePolicy::Allowed);
167 let mut recomputed = false;
168 let (state_root, receipt_root, receipts) = match receipt_ts.and_then(|ts| {
169 let receipt_root = *ts.parent_message_receipts();
170 Receipt::get_receipts(self.cs.db(), receipt_root)
171 .ok()
172 .map(|r| (*ts.parent_state(), receipt_root, r))
173 }) {
174 Some((state_root, receipt_root, receipts)) => (state_root, receipt_root, receipts),
175 None => {
176 if !allow_state_compute {
177 anyhow::bail!(state_compute_disallow_error());
178 }
179 let state_output = self
180 .compute_tipset_state(msg_ts.shallow_clone(), NO_CALLBACK, VMTrace::NotTraced)
181 .await?;
182 recomputed = true;
183 (
184 state_output.state_root,
185 state_output.receipt_root,
186 Receipt::get_receipts(self.cs.db(), state_output.receipt_root)?,
187 )
188 }
189 };
190
191 let messages = self.chain_store().messages_for_tipset(msg_ts)?;
192 anyhow::ensure!(
193 messages.len() == receipts.len(),
194 "mismatching message and receipt counts ({} messages, {} receipts)",
195 messages.len(),
196 receipts.len()
197 );
198 let mut executed_messages = Vec::with_capacity(messages.len());
199 for (message, receipt) in messages.iter().cloned().zip(receipts) {
200 let events = if let Some(events_root) = receipt.events_root() {
201 Some(match StampedEvent::get_events(self.cs.db(), &events_root) {
202 Ok(events) => events,
203 Err(e) if recomputed => return Err(e),
204 Err(_) => {
205 if !allow_state_compute {
206 anyhow::bail!(state_compute_disallow_error());
207 }
208 self.compute_tipset_state(
209 msg_ts.shallow_clone(),
210 NO_CALLBACK,
211 VMTrace::NotTraced,
212 )
213 .await?;
214 recomputed = true;
215 StampedEvent::get_events(self.cs.db(), &events_root)?
216 }
217 })
218 } else {
219 None
220 };
221 executed_messages.push(ExecutedMessage {
222 message,
223 receipt,
224 events,
225 });
226 }
227
228 if recomputed
230 && let Err(e) = crate::rpc::eth::store_block_logs_bloom(
231 self,
232 msg_ts,
233 &state_root,
234 &executed_messages,
235 )
236 {
237 warn!(
238 "failed to store block logs bloom for tipset {}: {e:#}",
239 msg_ts.key()
240 );
241 }
242
243 Ok(ExecutedTipset {
244 state_root,
245 receipt_root,
246 executed_messages: Arc::new(executed_messages),
247 })
248 }
249
250 pub async fn compute_tipset_state(
272 &self,
273 tipset: Tipset,
274 callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()> + Send + 'static>,
275 enable_tracing: VMTrace,
276 ) -> Result<ExecutedTipset, Error> {
277 let this = self.shallow_clone();
278 tokio::task::spawn_blocking(move || {
279 this.compute_tipset_state_blocking(tipset, callback, enable_tracing)
280 })
281 .await?
282 }
283
284 pub fn compute_tipset_state_blocking(
286 &self,
287 tipset: Tipset,
288 callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()>>,
289 enable_tracing: VMTrace,
290 ) -> Result<ExecutedTipset, Error> {
291 let epoch = tipset.epoch();
292 let has_callback = callback.is_some();
293 info!(
294 "Evaluating tipset: EPOCH={epoch}, blocks={}, tsk={}",
295 tipset.len(),
296 tipset.key(),
297 );
298 Ok(apply_block_messages_blocking(
299 self.chain_index().shallow_clone(),
300 self.chain_config().shallow_clone(),
301 self.beacon_schedule().shallow_clone(),
302 &self.engine,
303 tipset,
304 callback,
305 enable_tracing,
306 )
307 .map_err(|e| {
308 if has_callback {
309 e
310 } else {
311 e.context(format!("Failed to compute tipset state@{epoch}"))
312 }
313 })?)
314 }
315
316 #[instrument(skip_all)]
317 pub async fn compute_state(
318 &self,
319 height: ChainEpoch,
320 messages: Vec<Message>,
321 tipset: Tipset,
322 callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()> + Send + 'static>,
323 enable_tracing: VMTrace,
324 ) -> Result<ExecutedTipset, Error> {
325 let this = self.shallow_clone();
326 tokio::task::spawn_blocking(move || {
327 this.compute_state_blocking(height, messages, tipset, callback, enable_tracing)
328 })
329 .await?
330 }
331
332 #[tracing::instrument(skip_all)]
334 pub fn compute_state_blocking(
335 &self,
336 height: ChainEpoch,
337 messages: Vec<Message>,
338 tipset: Tipset,
339 callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()>>,
340 enable_tracing: VMTrace,
341 ) -> Result<ExecutedTipset, Error> {
342 Ok(compute_state_blocking(
343 height,
344 messages,
345 tipset,
346 self.chain_index().shallow_clone(),
347 self.chain_config().shallow_clone(),
348 self.beacon_schedule().shallow_clone(),
349 &self.engine,
350 callback,
351 enable_tracing,
352 )?)
353 }
354}
355
356pub fn validate_tipsets_blocking<T>(
357 chain_index: &ChainIndex,
358 chain_config: &Arc<ChainConfig>,
359 beacon: &Arc<BeaconSchedule>,
360 engine: &MultiEngine,
361 tipsets: T,
362) -> anyhow::Result<()>
363where
364 T: Iterator<Item = Tipset> + Send,
365{
366 for (child, parent) in tipsets.tuple_windows() {
371 info!(height = parent.epoch(), "compute parent state");
372 let ExecutedTipset {
373 state_root: actual_state,
374 receipt_root: actual_receipt,
375 ..
376 } = apply_block_messages_blocking(
377 chain_index.shallow_clone(),
378 chain_config.shallow_clone(),
379 beacon.shallow_clone(),
380 engine,
381 parent,
382 NO_CALLBACK,
383 VMTrace::NotTraced,
384 )
385 .context("couldn't compute tipset state")?;
386 let expected_receipt = child.min_ticket_block().message_receipts;
387 let expected_state = child.parent_state();
388 if (expected_state, expected_receipt) != (&actual_state, actual_receipt) {
389 error!(
390 height = child.epoch(),
391 ?expected_state,
392 ?expected_receipt,
393 ?actual_state,
394 ?actual_receipt,
395 "state mismatch"
396 );
397 bail!("state mismatch");
398 }
399 }
400 Ok(())
401}
402
403pub(in crate::state_manager) struct TipsetExecutor<'a> {
408 tipset: Tipset,
409 rand: ChainRand,
410 chain_config: Arc<ChainConfig>,
411 chain_index: ChainIndex,
412 genesis_info: GenesisInfo,
413 engine: &'a MultiEngine,
414}
415
416impl<'a> TipsetExecutor<'a> {
417 pub(in crate::state_manager) fn new(
418 chain_index: ChainIndex,
419 chain_config: Arc<ChainConfig>,
420 beacon: Arc<BeaconSchedule>,
421 engine: &'a MultiEngine,
422 tipset: Tipset,
423 ) -> Self {
424 let rand = ChainRand::new(
425 chain_config.shallow_clone(),
426 tipset.shallow_clone(),
427 chain_index.shallow_clone(),
428 beacon,
429 );
430 let genesis_info = GenesisInfo::from_chain_config(chain_config.shallow_clone());
431 Self {
432 tipset,
433 rand,
434 chain_config,
435 chain_index,
436 genesis_info,
437 engine,
438 }
439 }
440
441 pub(in crate::state_manager) fn create_vm(
442 &self,
443 state_root: Cid,
444 epoch: ChainEpoch,
445 timestamp: u64,
446 trace: VMTrace,
447 ) -> anyhow::Result<VM> {
448 let circ_supply = self.genesis_info.get_vm_circulating_supply(
449 epoch,
450 self.chain_index.db(),
451 &state_root,
452 )?;
453 VM::new(
454 ExecutionContext {
455 heaviest_tipset: self.tipset.shallow_clone(),
456 state_tree_root: state_root,
457 epoch,
458 rand: Box::new(self.rand.shallow_clone()),
459 base_fee: self.tipset.min_ticket_block().parent_base_fee.clone(),
460 circ_supply,
461 chain_config: self.chain_config.shallow_clone(),
462 chain_index: self.chain_index.shallow_clone(),
463 timestamp,
464 },
465 self.engine,
466 trace,
467 )
468 }
469
470 pub(in crate::state_manager) fn prepare_parent_state_blocking<F>(
473 &self,
474 genesis_timestamp: u64,
475 null_epoch_trace: VMTrace,
476 cron_callback: &mut Option<F>,
477 ) -> anyhow::Result<(Cid, ChainEpoch, Vec<BlockMessages>)>
478 where
479 F: FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()>,
480 {
481 use crate::shim::clock::EPOCH_DURATION_SECONDS;
482
483 let mut parent_state = *self.tipset.parent_state();
484 let parent_epoch = self
485 .chain_index
486 .load_required_tipset(self.tipset.parents())?
487 .epoch();
488 let epoch = self.tipset.epoch();
489
490 for epoch_i in parent_epoch..epoch {
491 if epoch_i > parent_epoch {
492 let timestamp = genesis_timestamp + ((EPOCH_DURATION_SECONDS * epoch_i) as u64);
493 parent_state = stacker::grow(64 << 20, || -> anyhow::Result<Cid> {
494 let mut vm =
495 self.create_vm(parent_state, epoch_i, timestamp, null_epoch_trace)?;
496 if let Err(e) = vm.run_cron(epoch_i, cron_callback.as_mut()) {
497 error!("Beginning of epoch cron failed to run: {e:#}");
498 return Err(e);
499 }
500 vm.flush()
501 })?;
502 }
503 if let Some(new_state) = run_state_migrations(
504 epoch_i,
505 &self.chain_config,
506 self.chain_index.db(),
507 &parent_state,
508 )? {
509 parent_state = new_state;
510 }
511 }
512
513 let block_messages = BlockMessages::for_tipset(self.chain_index.db(), &self.tipset)?;
514 Ok((parent_state, epoch, block_messages))
515 }
516}
517
518#[allow(clippy::too_many_arguments)]
595pub fn apply_block_messages_blocking(
596 chain_index: ChainIndex,
597 chain_config: Arc<ChainConfig>,
598 beacon: Arc<BeaconSchedule>,
599 engine: &MultiEngine,
600 tipset: Tipset,
601 mut callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()>>,
602 enable_tracing: VMTrace,
603) -> anyhow::Result<ExecutedTipset> {
604 let genesis_timestamp = chain_index.genesis().min_ticket_block().timestamp;
613 if tipset.epoch() == 0 {
614 let message_receipts = tipset.min_ticket_block().message_receipts;
619 return Ok(ExecutedTipset {
620 state_root: *tipset.parent_state(),
621 receipt_root: message_receipts,
622 executed_messages: vec![].into(),
623 });
624 }
625
626 let exec = TipsetExecutor::new(
627 chain_index.shallow_clone(),
628 chain_config,
629 beacon,
630 engine,
631 tipset.shallow_clone(),
632 );
633
634 let (parent_state, epoch, block_messages) =
637 exec.prepare_parent_state_blocking(genesis_timestamp, enable_tracing, &mut callback)?;
638
639 stacker::grow(64 << 20, || -> anyhow::Result<ExecutedTipset> {
642 let mut vm = exec.create_vm(parent_state, epoch, tipset.min_timestamp(), enable_tracing)?;
643
644 let (receipts, events, events_roots) =
646 vm.apply_block_messages(&block_messages, epoch, callback)?;
647
648 let receipt_root = Amtv0::new_from_iter(chain_index.db(), receipts.iter())?;
650
651 for (events, events_root) in events.iter().zip(events_roots.iter()) {
653 if let Some(events) = events {
654 let event_root =
655 events_root.context("events root should be present when events present")?;
656 let derived_event_root = Amt::new_from_iter_with_bit_width(
658 chain_index.db(),
659 EVENTS_AMT_BITWIDTH,
660 events.iter(),
661 )
662 .map_err(|e| Error::Other(format!("failed to store events AMT: {e}")))?;
663
664 ensure!(
666 derived_event_root == event_root,
667 "Events AMT root mismatch: derived={derived_event_root}, actual={event_root}."
668 );
669 }
670 }
671
672 let state_root = vm.flush()?;
673
674 let messages: Vec<ChainMessage> = block_messages
676 .into_iter()
677 .flat_map(|bm| bm.messages)
678 .collect_vec();
679 anyhow::ensure!(
680 messages.len() == receipts.len() && messages.len() == events.len(),
681 "length of messages, receipts, and events should match",
682 );
683 Ok(ExecutedTipset {
684 state_root,
685 receipt_root,
686 executed_messages: messages
687 .into_iter()
688 .zip(receipts)
689 .zip(events)
690 .map(|((message, receipt), events)| ExecutedMessage {
691 message,
692 receipt,
693 events,
694 })
695 .collect_vec()
696 .into(),
697 })
698 })
699}
700
701#[allow(clippy::too_many_arguments)]
702pub(in crate::state_manager) fn compute_state_blocking(
703 _height: ChainEpoch,
704 messages: Vec<Message>,
705 tipset: Tipset,
706 chain_index: ChainIndex,
707 chain_config: Arc<ChainConfig>,
708 beacon: Arc<BeaconSchedule>,
709 engine: &MultiEngine,
710 callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()>>,
711 enable_tracing: VMTrace,
712) -> anyhow::Result<ExecutedTipset> {
713 if !messages.is_empty() {
714 anyhow::bail!("Applying messages is not yet implemented.");
715 }
716
717 let output = apply_block_messages_blocking(
718 chain_index,
719 chain_config,
720 beacon,
721 engine,
722 tipset,
723 callback,
724 enable_tracing,
725 )?;
726
727 Ok(output)
728}