forest/message_pool/
msg_chain.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3#![allow(clippy::indexing_slicing)]
4use std::{
5    cmp::Ordering,
6    mem,
7    ops::{Index, IndexMut},
8};
9
10use crate::message::{MessageRead as _, SignedMessage};
11use crate::networks::ChainConfig;
12use crate::shim::{
13    address::Address,
14    econ::TokenAmount,
15    gas::{Gas, price_list_by_network_version},
16};
17use crate::{
18    blocks::{BLOCK_MESSAGE_LIMIT, Tipset},
19    shim::crypto::SignatureType,
20};
21use ahash::HashMap;
22use num_traits::Zero;
23use slotmap::{SlotMap, new_key_type};
24use tracing::warn;
25
26use super::errors::Error;
27use crate::message_pool::{
28    provider::Provider,
29    utils::{get_gas_perf, get_gas_reward},
30};
31
32new_key_type! {
33    pub struct NodeKey;
34}
35
36/// Chains is an abstraction of a list of message chain nodes.
37/// It wraps a `SlotMap` instance. `key_vec` is an additional requirement in
38/// order to satisfy optimal `msg` selection use cases, such as iteration in
39/// insertion order. The `SlotMap` serves as a lookup table for nodes to get
40/// around the borrow checker rules. Each `MsgChainNode` contains only pointers
41/// as `NodeKey` to the entries in the map With this design, we get around the
42/// borrow checker rule issues when implementing the optimal selection
43/// algorithm.
44pub(in crate::message_pool) struct Chains {
45    pub map: SlotMap<NodeKey, MsgChainNode>,
46    pub key_vec: Vec<NodeKey>,
47}
48
49impl Chains {
50    // Sort by effective perf with cmp_effective
51    pub(in crate::message_pool) fn sort_effective(&mut self) {
52        let mut chains = mem::take(&mut self.key_vec);
53        chains.sort_by(|a, b| {
54            let a = self.map.get(*a).unwrap();
55            let b = self.map.get(*b).unwrap();
56            a.cmp_effective(b)
57        });
58        let _ = mem::replace(&mut self.key_vec, chains);
59    }
60
61    // Sort by effective `perf` on a range
62    pub(in crate::message_pool) fn sort_range_effective(
63        &mut self,
64        range: std::ops::RangeFrom<usize>,
65    ) {
66        let mut chains = mem::take(&mut self.key_vec);
67        chains[range].sort_by(|a, b| {
68            self.map
69                .get(*a)
70                .unwrap()
71                .cmp_effective(self.map.get(*b).unwrap())
72        });
73        let _ = mem::replace(&mut self.key_vec, chains);
74    }
75
76    /// Retrieves the `msg` chain node by the given `NodeKey` along with the
77    /// data required from previous chain (if exists) to set effective
78    /// performance of this node.
79    pub(in crate::message_pool) fn get_mut_with_prev_eff(
80        &mut self,
81        k: NodeKey,
82    ) -> (Option<&mut MsgChainNode>, Option<(f64, u64)>) {
83        let node = self.map.get(k);
84        let prev = if let Some(node) = node {
85            if let Some(prev_key) = node.prev {
86                let prev_node = self.map.get(prev_key).unwrap();
87                Some((prev_node.eff_perf, prev_node.gas_limit))
88            } else {
89                None
90            }
91        } else {
92            None
93        };
94
95        let node = self.map.get_mut(k);
96        (node, prev)
97    }
98
99    /// Retrieves the `msg` chain node by the given `NodeKey`
100    pub(in crate::message_pool) fn get(&self, k: NodeKey) -> Option<&MsgChainNode> {
101        self.map.get(k)
102    }
103}
104
105impl Chains {
106    pub(in crate::message_pool) fn new() -> Self {
107        Self {
108            map: SlotMap::with_key(),
109            key_vec: vec![],
110        }
111    }
112
113    /// Pushes a `msg` chain node into slot map and places the key in the
114    /// `node_vec` passed as parameter.
115    pub(in crate::message_pool) fn push_with(
116        &mut self,
117        cur_chain: MsgChainNode,
118        node_vec: &mut Vec<NodeKey>,
119    ) {
120        let key = self.map.insert(cur_chain);
121        node_vec.push(key);
122    }
123
124    /// Sorts the chains with `compare` method. If rev is true, sorts in
125    /// descending order.
126    pub(in crate::message_pool) fn sort(&mut self, rev: bool) {
127        // replace dance to get around borrow checker
128        let mut chains = mem::take(&mut self.key_vec);
129        chains.sort_by(|a, b| {
130            let a = self.map.get(*a).unwrap();
131            let b = self.map.get(*b).unwrap();
132            if rev { b.compare(a) } else { a.compare(b) }
133        });
134        let _ = mem::replace(&mut self.key_vec, chains);
135    }
136
137    /// Retrieves the `msg` chain node by the given `NodeKey`
138    pub(in crate::message_pool) fn get_mut(&mut self, k: NodeKey) -> Option<&mut MsgChainNode> {
139        self.map.get_mut(k)
140    }
141
142    /// Retrieves the `msg` chain node at the given index
143    pub(in crate::message_pool) fn get_mut_at(&mut self, i: usize) -> Option<&mut MsgChainNode> {
144        let key = self.key_vec.get(i)?;
145        self.get_mut(*key)
146    }
147
148    // Retrieves a msg chain node at the given index in the provided NodeKey vec
149    pub(in crate::message_pool) fn get_from(&self, i: usize, vec: &[NodeKey]) -> &MsgChainNode {
150        #[allow(clippy::indexing_slicing)]
151        self.map.get(vec[i]).unwrap()
152    }
153
154    // Retrieves a msg chain node at the given index in the provided NodeKey vec
155    pub(in crate::message_pool) fn get_mut_from(
156        &mut self,
157        i: usize,
158        vec: &[NodeKey],
159    ) -> &mut MsgChainNode {
160        #[allow(clippy::indexing_slicing)]
161        self.map.get_mut(vec[i]).unwrap()
162    }
163
164    // Retrieves the node key at the given index
165    pub(in crate::message_pool) fn get_key_at(&self, i: usize) -> Option<NodeKey> {
166        self.key_vec.get(i).copied()
167    }
168
169    /// Retrieves the `msg` chain node at the given index. Returns `None` if index is out-of-bounds.
170    pub(in crate::message_pool) fn get_at(&self, i: usize) -> Option<&MsgChainNode> {
171        self.map.get(self.get_key_at(i)?)
172    }
173
174    /// Retrieves the amount of items.
175    pub(in crate::message_pool) fn len(&self) -> usize {
176        self.map.len()
177    }
178
179    /// Returns true is the chain is empty and otherwise. We check the map as
180    /// the source of truth as `key_vec` can be extended time to time.
181    pub(in crate::message_pool) fn is_empty(&self) -> bool {
182        self.map.is_empty()
183    }
184
185    /// Removes messages from the given index and resets effective `perfs`
186    #[tracing::instrument(skip_all, level = "debug")]
187    pub(in crate::message_pool) fn trim_msgs_at(
188        &mut self,
189        idx: usize,
190        gas_limit: u64,
191        msg_limit: usize,
192        base_fee: &TokenAmount,
193    ) {
194        let prev = match idx {
195            0 => None,
196            _ => self
197                .get_at(idx - 1)
198                .map(|prev| (prev.eff_perf, prev.gas_limit)),
199        };
200        let chain_node = self.get_mut_at(idx).unwrap();
201        let mut i = chain_node.msgs.len() as i64 - 1;
202
203        while i >= 0
204            && (chain_node.gas_limit > gas_limit
205                || chain_node.gas_perf < 0.0
206                || i >= msg_limit as i64)
207        {
208            #[allow(clippy::indexing_slicing)]
209            let msg = &chain_node.msgs[i as usize];
210            let gas_reward = get_gas_reward(msg, base_fee);
211            chain_node.gas_reward -= gas_reward;
212            chain_node.gas_limit = chain_node.gas_limit.saturating_sub(msg.gas_limit());
213            if chain_node.gas_limit > 0 {
214                chain_node.gas_perf = get_gas_perf(&chain_node.gas_reward, chain_node.gas_limit);
215                if chain_node.bp != 0.0 {
216                    chain_node.set_eff_perf(prev);
217                }
218            } else {
219                chain_node.gas_perf = 0.0;
220                chain_node.eff_perf = 0.0;
221            }
222            i -= 1;
223        }
224
225        if i < 0 {
226            chain_node.msgs.clear();
227            chain_node.valid = false;
228        } else {
229            chain_node.msgs.truncate(i as usize + 1);
230        }
231
232        let next = chain_node.next;
233        if next.is_some() {
234            self.invalidate(next);
235        }
236    }
237
238    pub(in crate::message_pool) fn bubble_down_after_trim(&mut self, from: usize) {
239        let mut j = from;
240        while j < self.key_vec.len().saturating_sub(1) {
241            #[allow(clippy::indexing_slicing)]
242            if self[j].compare(&self[j + 1]) == Ordering::Less {
243                break;
244            }
245            self.key_vec.swap(j, j + 1);
246            j += 1;
247        }
248    }
249
250    pub(in crate::message_pool) fn invalidate(&mut self, mut key: Option<NodeKey>) {
251        let mut next_keys = vec![];
252
253        while let Some(nk) = key {
254            let chain_node = self.map.get(nk).unwrap();
255            next_keys.push(nk);
256            key = chain_node.next;
257        }
258
259        for k in next_keys.iter().rev() {
260            if let Some(node) = self.map.get_mut(*k) {
261                node.valid = false;
262                node.msgs.clear();
263                node.next = None;
264            }
265        }
266    }
267
268    /// Drops nodes which are no longer valid after the merge step
269    pub(in crate::message_pool) fn drop_invalid(&mut self, key_vec: &mut Vec<NodeKey>) {
270        let mut valid_keys = vec![];
271        for k in key_vec.iter() {
272            if self.map.get(*k).map(|n| n.valid).unwrap() {
273                valid_keys.push(*k);
274            } else {
275                self.map.remove(*k);
276            }
277        }
278
279        *key_vec = valid_keys;
280    }
281}
282
283impl Index<usize> for Chains {
284    type Output = MsgChainNode;
285    fn index(&self, i: usize) -> &Self::Output {
286        self.get_at(i).unwrap()
287    }
288}
289
290impl IndexMut<usize> for Chains {
291    fn index_mut(&mut self, i: usize) -> &mut Self::Output {
292        #[allow(clippy::indexing_slicing)]
293        self.map.get_mut(self.key_vec[i]).unwrap()
294    }
295}
296
297/// Represents a node in the `MsgChain`.
298#[derive(Clone, Debug)]
299pub struct MsgChainNode {
300    pub msgs: Vec<SignedMessage>,
301    pub gas_reward: TokenAmount,
302    pub gas_limit: u64,
303    pub gas_perf: f64,
304    pub eff_perf: f64,
305    pub bp: f64,
306    pub parent_offset: f64,
307    pub valid: bool,
308    pub merged: bool,
309    pub next: Option<NodeKey>,
310    pub prev: Option<NodeKey>,
311    pub sig_type: Option<SignatureType>,
312}
313
314impl MsgChainNode {
315    pub fn compare(&self, other: &Self) -> Ordering {
316        if approx_cmp(self.gas_perf, other.gas_perf) == Ordering::Greater
317            || approx_cmp(self.gas_perf, other.gas_perf) == Ordering::Equal
318                && self.gas_reward.cmp(&other.gas_reward) == Ordering::Greater
319        {
320            return Ordering::Greater;
321        }
322
323        Ordering::Less
324    }
325
326    pub fn set_eff_perf(&mut self, prev: Option<(f64, u64)>) {
327        let mut eff_perf = self.gas_perf * self.bp;
328        if let Some(prev) = prev
329            && eff_perf > 0.0
330        {
331            let prev_eff_perf = prev.0;
332            let prev_gas_limit = prev.1;
333            let eff_perf_with_parent = (eff_perf * self.gas_limit as f64
334                + prev_eff_perf * prev_gas_limit as f64)
335                / (self.gas_limit + prev_gas_limit) as f64;
336            self.parent_offset = eff_perf - eff_perf_with_parent;
337            eff_perf = eff_perf_with_parent;
338        }
339        self.eff_perf = eff_perf;
340    }
341}
342
343impl MsgChainNode {
344    pub(in crate::message_pool) fn cmp_effective(&self, other: &Self) -> Ordering {
345        // Highest priority: merged
346        // Comment from Lotus: move merged chains to the front so we can discard them earlier
347        // Note: both cases need to be checked to ensure total ordering. Without it, the standard
348        // libraries' sorting methods may panic (since Rust 1.81).
349        match (self.merged, other.merged) {
350            (true, false) => return Ordering::Greater,
351            (false, true) => return Ordering::Less,
352            _ => {}
353        }
354
355        if self.gas_perf >= 0.0 && other.gas_perf < 0.0
356            || self.eff_perf > other.eff_perf
357            || (approx_cmp(self.eff_perf, other.eff_perf) == Ordering::Equal
358                && self.gas_perf > other.gas_perf)
359            || (approx_cmp(self.eff_perf, other.eff_perf) == Ordering::Equal
360                && approx_cmp(self.gas_perf, other.gas_perf) == Ordering::Equal
361                && self.gas_reward > other.gas_reward)
362        {
363            return Ordering::Greater;
364        }
365
366        Ordering::Less
367    }
368
369    pub fn set_null_effective_perf(&mut self) {
370        if self.gas_perf < 0.0 {
371            self.eff_perf = self.gas_perf;
372        } else {
373            self.eff_perf = 0.0;
374        }
375    }
376}
377
378impl std::default::Default for MsgChainNode {
379    fn default() -> Self {
380        Self {
381            msgs: vec![],
382            gas_reward: TokenAmount::zero(),
383            gas_limit: 0,
384            gas_perf: 0.0,
385            eff_perf: 0.0,
386            bp: 0.0,
387            parent_offset: 0.0,
388            valid: true,
389            merged: false,
390            next: None,
391            prev: None,
392            sig_type: None,
393        }
394    }
395}
396
397pub(in crate::message_pool) fn create_message_chains<T>(
398    api: &T,
399    actor: &Address,
400    mset: &HashMap<u64, SignedMessage>,
401    base_fee: &TokenAmount,
402    ts: &Tipset,
403    chains: &mut Chains,
404    chain_config: &ChainConfig,
405) -> Result<(), Error>
406where
407    T: Provider,
408{
409    // collect all messages and sort
410    let mut msgs: Vec<SignedMessage> = mset.values().cloned().collect();
411    msgs.sort_by_key(|v| v.sequence());
412
413    // sanity checks:
414    // - there can be no gaps in nonces, starting from the current actor nonce if
415    //   there is a gap, drop messages after the gap, we can't include them
416    // - all messages must have minimum gas and the total gas for the candidate
417    //   messages cannot exceed the block limit; drop all messages that exceed the
418    //   limit
419    // - the total gasReward cannot exceed the actor's balance; drop all messages
420    //   that exceed the balance
421    let Ok(actor_state) = api.get_actor_after(actor, ts) else {
422        tracing::warn!("failed to load actor state, not building chain for {actor}");
423        return Ok(());
424    };
425    let mut cur_seq = actor_state.sequence;
426    let mut balance: TokenAmount = TokenAmount::from(&actor_state.balance);
427
428    let mut gas_limit = 0;
429    let mut skip = 0;
430    let mut i = 0;
431    let mut rewards = Vec::with_capacity(msgs.len());
432
433    while let Some(m) = msgs.get(i) {
434        if m.sequence() < cur_seq {
435            warn!(
436                "encountered message from actor {} with nonce {} less than the current nonce {}",
437                actor,
438                m.sequence(),
439                cur_seq
440            );
441            skip += 1;
442            i += 1;
443            continue;
444        }
445
446        if m.sequence() != cur_seq {
447            break;
448        }
449        cur_seq += 1;
450
451        let network_version = chain_config.network_version(ts.epoch());
452
453        let min_gas = price_list_by_network_version(network_version)
454            .on_chain_message(m.chain_length()?)
455            .total();
456
457        if Gas::new(m.gas_limit()) < min_gas {
458            break;
459        }
460        gas_limit += m.gas_limit();
461        if gas_limit > crate::shim::econ::BLOCK_GAS_LIMIT {
462            break;
463        }
464
465        let required = m.required_funds();
466        if balance < required {
467            break;
468        }
469
470        balance -= required;
471        let value = m.value();
472        balance -= value;
473
474        let gas_reward = get_gas_reward(m, base_fee);
475        rewards.push(gas_reward);
476        i += 1;
477    }
478
479    // check we have a sane set of messages to construct the chains
480    let mut msgs = if i > skip {
481        #[allow(clippy::indexing_slicing)]
482        msgs[skip..i].to_vec()
483    } else {
484        return Ok(());
485    };
486
487    // if we have more messages from this sender than can fit in a block, drop the extra ones
488    if msgs.len() > BLOCK_MESSAGE_LIMIT {
489        warn!(
490            "dropping {} messages from {actor} as they exceed the block message limit of {BLOCK_MESSAGE_LIMIT}",
491            msgs.len() - BLOCK_MESSAGE_LIMIT,
492        );
493        msgs.truncate(BLOCK_MESSAGE_LIMIT);
494    };
495
496    let mut cur_chain = MsgChainNode::default();
497    let mut node_vec = vec![];
498
499    let new_chain = |m: SignedMessage, reward: &TokenAmount| -> MsgChainNode {
500        let gl = m.gas_limit();
501        let sig_type = Some(m.signature().sig_type);
502        MsgChainNode {
503            msgs: vec![m],
504            gas_reward: reward.clone(),
505            gas_limit: gl,
506            gas_perf: get_gas_perf(reward, gl),
507            eff_perf: 0.0,
508            bp: 0.0,
509            parent_offset: 0.0,
510            valid: true,
511            merged: false,
512            prev: None,
513            next: None,
514            sig_type,
515        }
516    };
517
518    // creates msg chain nodes in chunks based on gas_perf obtained from the current
519    // chain's gas limit.
520    for (i, (m, reward)) in msgs.into_iter().zip(rewards.iter()).enumerate() {
521        if i == 0 {
522            cur_chain = new_chain(m, reward);
523            continue;
524        }
525
526        let gas_reward = cur_chain.gas_reward.clone() + reward;
527        let gas_limit = cur_chain.gas_limit + m.gas_limit();
528        let gas_perf = get_gas_perf(&gas_reward, gas_limit);
529
530        // try to add the message to the current chain -- if it decreases the gasPerf,
531        // then make a new chain
532        if gas_perf < cur_chain.gas_perf {
533            chains.push_with(cur_chain, &mut node_vec);
534            cur_chain = new_chain(m, reward);
535        } else {
536            cur_chain.msgs.push(m);
537            cur_chain.gas_reward = gas_reward;
538            cur_chain.gas_limit = gas_limit;
539            cur_chain.gas_perf = gas_perf;
540        }
541    }
542
543    chains.push_with(cur_chain, &mut node_vec);
544
545    // merge chains to maintain the invariant: higher gas perf nodes on the front.
546    loop {
547        let mut merged = 0;
548        for i in (1..node_vec.len()).rev() {
549            if chains.get_from(i, &node_vec).gas_perf >= chains.get_from(i - 1, &node_vec).gas_perf
550            {
551                // copy messages
552                let chain_i_msg = chains.get_from(i, &node_vec).msgs.clone();
553                chains
554                    .get_mut_from(i - 1, &node_vec)
555                    .msgs
556                    .extend(chain_i_msg);
557
558                // set gas reward
559                let chain_i_gas_reward = chains.get_from(i, &node_vec).gas_reward.clone();
560                chains.get_mut_from(i - 1, &node_vec).gas_reward += chain_i_gas_reward;
561
562                // set gas limit
563                let chain_i_gas_limit = chains.get_from(i, &node_vec).gas_limit;
564                chains.get_mut_from(i - 1, &node_vec).gas_limit += chain_i_gas_limit;
565
566                // set gas perf
567                let chain_i_gas_perf = get_gas_perf(
568                    &chains.get_from(i - 1, &node_vec).gas_reward,
569                    chains.get_from(i - 1, &node_vec).gas_limit,
570                );
571                chains.get_mut_from(i - 1, &node_vec).gas_perf = chain_i_gas_perf;
572                // invalidate the current chain as it is merged with the prev chain
573                chains.get_mut_from(i, &node_vec).valid = false;
574                merged += 1;
575            }
576        }
577
578        if merged == 0 {
579            break;
580        }
581
582        chains.drop_invalid(&mut node_vec);
583    }
584
585    if node_vec.len() > 1 {
586        for (&k1, &k2) in node_vec.iter().zip(node_vec.iter().skip(1)) {
587            // link next pointers
588            let n1 = chains
589                .get_mut(k1)
590                .ok_or_else(|| Error::Other(format!("{k1:?} should present in `chains`")))?;
591            n1.next = Some(k2);
592            // Should we link or clear n1.prev as well?
593
594            // link prev pointers
595            let n2 = chains
596                .get_mut(k2)
597                .ok_or_else(|| Error::Other(format!("{k2:?} should present in `chains`")))?;
598            n2.prev = Some(k1);
599            // Should we link or clear n2.next as well?
600        }
601    }
602
603    // Update the main chain key_vec with this node_vec
604    chains.key_vec.extend(node_vec);
605
606    Ok(())
607}
608
609fn approx_cmp(a: f64, b: f64) -> Ordering {
610    if (a - b).abs() <= (a * f64::EPSILON).abs() {
611        Ordering::Equal
612    } else {
613        a.partial_cmp(&b).unwrap()
614    }
615}
forest/message_pool/msg_chain.rs

forest/message_pool/
msg_chain.rs
