diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index 66c5f6eacca2..aeab65e65776 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -272,7 +272,7 @@ async fn chain_follower( get_full_tipset(&network, cs, None, &key).await } PubsubMessage::Message(m) => { - if let Err(why) = mem_pool.add(m) { + if let Err(why) = mem_pool.add(m).await { debug!("Received invalid GossipSub message: {}", why); } continue; diff --git a/src/dev/subcommands/tests_cmd/helpers.rs b/src/dev/subcommands/tests_cmd/helpers.rs index 9f1cac7735f0..dd38c9392aa0 100644 --- a/src/dev/subcommands/tests_cmd/helpers.rs +++ b/src/dev/subcommands/tests_cmd/helpers.rs @@ -303,11 +303,25 @@ pub async fn rpc_call_opt(method: &str, params: Value) -> anyhow::Result anyhow::Result { - rpc_call_opt(method, params) - .await? - .with_context(|| format!("missing `result` in response for {method}")) +/// Like [`rpc_call_opt`] but treats a missing `result` as an error and retries. +pub async fn rpc_call_with_retry(method: &str, params: Value) -> anyhow::Result { + let mut attempt = 1; + loop { + let result = rpc_call_opt(method, params.clone()).await.and_then(|opt| { + opt.with_context(|| format!("missing `result` in response for {method}")) + }); + match result { + Ok(v) => return Ok(v), + Err(e) if attempt < SEND_RETRIES => { + eprintln!( + "error: {e:?} {method} failed on attempt {attempt}/{SEND_RETRIES}, retrying" + ); + tokio::time::sleep(SEND_RETRY_DELAY).await; + attempt += 1; + } + Err(e) => return Err(e), + } + } } /// Extract a CID string from either a Lotus `{ "/": "bafy..." }` map or a @@ -362,7 +376,7 @@ pub fn mpool_nonce(address: &str) -> anyhow::Result { /// Pending message nonces for `address` via `Filecoin.MpoolPending`. pub async fn pending_nonces_for(address: &str) -> anyhow::Result> { - let result = rpc_call("Filecoin.MpoolPending", json!([null])).await?; + let result = rpc_call_with_retry("Filecoin.MpoolPending", json!([null])).await?; let entries = result .as_array() .with_context(|| format!("expected MpoolPending array, got {result}"))?; @@ -389,7 +403,7 @@ pub async fn poll_until_pending_nonce(address: &str, nonce: u64) -> anyhow::Resu /// Resolve the ETH equivalent of a Filecoin address via /// `Filecoin.FilecoinAddressToEthAddress`. pub async fn filecoin_to_eth(address: &str) -> anyhow::Result { - let result = rpc_call( + let result = rpc_call_with_retry( "Filecoin.FilecoinAddressToEthAddress", json!([address, "pending"]), ) diff --git a/src/dev/subcommands/tests_cmd/wallet.rs b/src/dev/subcommands/tests_cmd/wallet.rs index 989c662b58dc..14116e064eb1 100644 --- a/src/dev/subcommands/tests_cmd/wallet.rs +++ b/src/dev/subcommands/tests_cmd/wallet.rs @@ -93,7 +93,7 @@ async fn export_import_roundtrip(backend: Backend) { async fn market_add_balance_message_on_chain() { const ATTO_FIL: &str = "23"; - let result = rpc_call( + let result = rpc_call_with_retry( "Filecoin.MarketAddBalance", serde_json::json!([ FOREST_TEST_PRELOADED_ADDRESS.as_str(), diff --git a/src/message_pool/msgpool/mod.rs b/src/message_pool/msgpool/mod.rs index 9aa6f75f3a2c..c248f95b3ee9 100644 --- a/src/message_pool/msgpool/mod.rs +++ b/src/message_pool/msgpool/mod.rs @@ -118,10 +118,10 @@ pub mod tests { let (last, body) = smsg_vec.split_last().unwrap(); for smsg in body { - mpool.add(smsg.clone()).unwrap(); + mpool.add(smsg.clone()).await.unwrap(); } assert_eq!( - mpool.add(last.clone()), + mpool.add(last.clone()).await, Err(Error::TooManyPendingMessages(sender.to_string(), true)) ); } @@ -194,11 +194,11 @@ pub mod tests { } mpool.api.inner.lock().set_state_sequence(&sender, 0); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 0); - mpool.add(smsg_vec[0].clone()).unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 1); - mpool.add(smsg_vec[1].clone()).unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 2); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 0); + mpool.add(smsg_vec[0].clone()).await.unwrap(); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 1); + mpool.add(smsg_vec[1].clone()).await.unwrap(); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 2); let a = mock_block(1, 1); @@ -208,7 +208,7 @@ pub mod tests { .await .unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 2); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 2); } #[tokio::test] @@ -240,10 +240,10 @@ pub mod tests { drop(api_temp); } - mpool.add(smsg_vec[0].clone()).unwrap(); - mpool.add(smsg_vec[1].clone()).unwrap(); - mpool.add(smsg_vec[2].clone()).unwrap(); - mpool.add(smsg_vec[3].clone()).unwrap(); + mpool.add(smsg_vec[0].clone()).await.unwrap(); + mpool.add(smsg_vec[1].clone()).await.unwrap(); + mpool.add(smsg_vec[2].clone()).await.unwrap(); + mpool.add(smsg_vec[3].clone()).await.unwrap(); mpool.api.set_state_sequence(&sender, 0); @@ -252,7 +252,7 @@ pub mod tests { .await .unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 4); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 4); mpool.api.set_state_sequence(&sender, 1); @@ -261,7 +261,7 @@ pub mod tests { .await .unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 4); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 4); mpool.api.set_state_sequence(&sender, 0); @@ -270,7 +270,7 @@ pub mod tests { .await .unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 4); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 4); let (p, _) = mpool.pending(); assert_eq!(p.len(), 3); @@ -291,12 +291,12 @@ pub mod tests { let target = Address::new_id(1001); for i in 0..3 { let msg = create_smsg(&target, &key_addr, wallet.borrow_mut(), i, 1000000, 1); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); } // get_sequence with ID address should see the same pending messages - assert_eq!(mpool.get_sequence(&id_addr).unwrap(), 3); - assert_eq!(mpool.get_sequence(&key_addr).unwrap(), 3); + assert_eq!(mpool.get_sequence(&id_addr).await.unwrap(), 3); + assert_eq!(mpool.get_sequence(&key_addr).await.unwrap(), 3); } #[tokio::test] @@ -314,18 +314,20 @@ pub mod tests { let target = Address::new_id(1001); for i in 0..2 { let msg = create_smsg(&target, &key_addr, wallet.borrow_mut(), i, 1000000, 1); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); } // pending_for with ID address should find messages added under key address let msgs = mpool .pending_for(&id_addr) + .await .expect("should find pending messages"); assert_eq!(msgs.len(), 2); // pending_for with key address should also work let msgs2 = mpool .pending_for(&key_addr) + .await .expect("should find pending messages"); assert_eq!(msgs2.len(), 2); } @@ -342,7 +344,7 @@ pub mod tests { // Create a message with the ID address as sender and a fake signature let msg = create_fake_smsg(&mpool, &Address::new_id(1001), &id_addr, 0, 1000000, 1); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); // Pending map should be keyed by key_addr, not id_addr. assert!( @@ -372,9 +374,9 @@ pub mod tests { let target = Address::new_id(1001); let msg0 = create_smsg(&target, &key_addr, wallet.borrow_mut(), 0, 1000000, 1); let msg1 = create_smsg(&target, &key_addr, wallet.borrow_mut(), 1, 1000000, 1); - mpool.add(msg0.clone()).unwrap(); - mpool.add(msg1).unwrap(); - assert_eq!(mpool.get_sequence(&key_addr).unwrap(), 2); + mpool.add(msg0.clone()).await.unwrap(); + mpool.add(msg1).await.unwrap(); + assert_eq!(mpool.get_sequence(&key_addr).await.unwrap(), 2); // Block messages are stored under the key_addr (as would appear on chain). // The head_change remove path resolves addresses before touching pending. @@ -388,9 +390,10 @@ pub mod tests { .unwrap(); // msg0 was applied on chain, msg1 remains pending - assert_eq!(mpool.get_sequence(&id_addr).unwrap(), 2); + assert_eq!(mpool.get_sequence(&id_addr).await.unwrap(), 2); let msgs = mpool .pending_for(&key_addr) + .await .expect("should have remaining msg"); assert_eq!(msgs.len(), 1); assert_eq!(msgs[0].sequence(), 1); @@ -413,13 +416,13 @@ pub mod tests { smsg_vec.push(msg); } - assert_eq!(mpool.get_sequence(&sender).unwrap(), 0); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 0); mpool.push(smsg_vec[0].clone()).await.unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 1); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 1); mpool.push(smsg_vec[1].clone()).await.unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 2); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 2); mpool.push(smsg_vec[2].clone()).await.unwrap(); - assert_eq!(mpool.get_sequence(&sender).unwrap(), 3); + assert_eq!(mpool.get_sequence(&sender).await.unwrap(), 3); let header = mock_block(1, 1); let tipset = Tipset::from(&header.clone()); diff --git a/src/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index 44208bf30392..f1c43bd43366 100644 --- a/src/message_pool/msgpool/msg_pool.rs +++ b/src/message_pool/msgpool/msg_pool.rs @@ -18,7 +18,7 @@ use crate::message_pool::{ BASE_FEE_LOWER_BOUND_FACTOR_CONSERVATIVE, events::MpoolSubscriber, pending_store::PendingStore, recover_sig, republish::RepublishState, }, - provider::Provider, + provider::{Provider, ProviderExt}, utils::get_base_fee_lower_bound, }; use crate::networks::{ChainConfig, NEWEST_NETWORK_VERSION}; @@ -147,8 +147,8 @@ impl ShallowClone for MessagePool { /// Resolve an address to its key form, checking the cache first. /// Non-ID addresses are returned unchanged. -pub(in crate::message_pool) fn resolve_to_key( - api: &T, +pub(in crate::message_pool) async fn resolve_to_key( + api: &Arc, key_cache: &IdToAddressCache, addr: &Address, cur_ts: &Tipset, @@ -159,7 +159,9 @@ pub(in crate::message_pool) fn resolve_to_key( { return Ok(resolved); } - let resolved = api.resolve_to_deterministic_address_at_finality(addr, cur_ts)?; + let resolved = api + .resolve_to_deterministic_address_at_finality_async(*addr, cur_ts.clone()) + .await?; if let Some(id) = id { key_cache.insert(id, resolved); } @@ -175,19 +177,25 @@ where self.cur_tipset.read().clone() } - pub(in crate::message_pool) fn resolve_to_key( + pub(in crate::message_pool) async fn resolve_to_key( &self, addr: &Address, cur_ts: &Tipset, - ) -> Result { - resolve_to_key(self.api.as_ref(), &self.caches.key, addr, cur_ts) + ) -> Result + where + T: Send + Sync + 'static, + { + resolve_to_key(&self.api, &self.caches.key, addr, cur_ts).await } /// Record the resolved-key sender of a locally-submitted message so the /// republish loop can find it on its next sweep. - fn add_local(&self, m: &SignedMessage) -> Result<(), Error> { + async fn add_local(&self, m: &SignedMessage) -> Result<(), Error> + where + T: Send + Sync + 'static, + { let cur_ts = self.current_tipset(); - let resolved = self.resolve_to_key(&m.from(), &cur_ts)?; + let resolved = self.resolve_to_key(&m.from(), &cur_ts).await?; self.local_addrs.write().insert(resolved); Ok(()) } @@ -198,10 +206,13 @@ where &self, msg: SignedMessage, trust_policy: TrustPolicy, - ) -> Result { + ) -> Result + where + T: Send + Sync + 'static, + { let cid = msg.cid(); - let publish = self.add_to_pool(msg.clone(), true, trust_policy)?; - self.add_local(&msg)?; + let publish = self.add_to_pool(msg.clone(), true, trust_policy).await?; + self.add_local(&msg).await?; if publish { self.publish_pubsub(&msg).await?; } @@ -225,19 +236,28 @@ where } /// Push a signed message to the `MessagePool` from an trusted source. - pub async fn push(&self, msg: SignedMessage) -> Result { + pub async fn push(&self, msg: SignedMessage) -> Result + where + T: Send + Sync + 'static, + { self.push_internal(msg, TrustPolicy::Trusted).await } /// Push a signed message to the `MessagePool` from an untrusted source. - pub async fn push_untrusted(&self, msg: SignedMessage) -> Result { + pub async fn push_untrusted(&self, msg: SignedMessage) -> Result + where + T: Send + Sync + 'static, + { self.push_internal(msg, TrustPolicy::Untrusted).await } /// Insert a message received via gossip. Runs full validation. Does /// not publish back to the network. - pub fn add(&self, msg: SignedMessage) -> Result<(), Error> { - self.add_to_pool(msg, false, TrustPolicy::Trusted)?; + pub async fn add(&self, msg: SignedMessage) -> Result<(), Error> + where + T: Send + Sync + 'static, + { + self.add_to_pool(msg, false, TrustPolicy::Trusted).await?; Ok(()) } @@ -246,16 +266,19 @@ where /// Returns `publish: bool` — `true` when the message should be gossiped /// after insertion; `false` when a local sender's message failed the /// soft base-fee floor (kept locally, not broadcast). - pub(in crate::message_pool) fn validate_for_pool( + pub(in crate::message_pool) async fn validate_for_pool( &self, msg: &SignedMessage, cur_ts: &Tipset, local: bool, - ) -> Result { + ) -> Result + where + T: Send + Sync + 'static, + { validate_static(msg)?; validate_signature(msg, &self.caches.sig_val, self.chain_config.eth_chain_id)?; - let expected_sequence = self.get_state_sequence(&msg.from(), cur_ts)?; + let expected_sequence = self.get_state_sequence(&msg.from(), cur_ts).await?; let sender_actor = self.api.get_actor_after(&msg.from(), cur_ts)?; validate_with_state( @@ -271,20 +294,24 @@ where /// Validate `msg` and insert it into the pending pool. /// /// Returns `publish: bool` (see [`Self::validate_for_pool`]). - pub(in crate::message_pool) fn add_to_pool( + pub(in crate::message_pool) async fn add_to_pool( &self, msg: SignedMessage, local: bool, trust_policy: TrustPolicy, - ) -> Result { + ) -> Result + where + T: Send + Sync + 'static, + { let cur_ts = self.current_tipset(); - let publish = self.validate_for_pool(&msg, &cur_ts, local)?; + let publish = self.validate_for_pool(&msg, &cur_ts, local).await?; let strictness = if local { StrictnessPolicy::Relaxed } else { StrictnessPolicy::Strict }; - self.add_to_pool_unchecked(&cur_ts, msg, trust_policy, strictness)?; + self.add_to_pool_unchecked(&cur_ts, msg, trust_policy, strictness) + .await?; Ok(publish) } @@ -292,13 +319,16 @@ where /// (size, sig, base-fee, sender-actor checks). The reorg replay path /// uses this directly to restore reverted messages even when they no /// longer pass the add-time filters. - pub(in crate::message_pool) fn add_to_pool_unchecked( + pub(in crate::message_pool) async fn add_to_pool_unchecked( &self, cur_ts: &Tipset, msg: SignedMessage, trust_policy: TrustPolicy, strictness: StrictnessPolicy, - ) -> Result<(), Error> { + ) -> Result<(), Error> + where + T: Send + Sync + 'static, + { if msg.signature().signature_type() == SignatureType::Bls { self.caches .bls_sig @@ -310,20 +340,23 @@ where self.api .put_message(&ChainMessage::Unsigned(msg.message().clone().into()))?; - let sequence = self.get_state_sequence(&msg.from(), cur_ts)?; - let resolved_from = self.resolve_to_key(&msg.from(), cur_ts)?; + let sequence = self.get_state_sequence(&msg.from(), cur_ts).await?; + let resolved_from = self.resolve_to_key(&msg.from(), cur_ts).await?; self.pending .insert(resolved_from, msg, sequence, trust_policy, strictness) } /// Get the sequence for a given address, return Error if there is a failure /// to retrieve the respective sequence. - pub fn get_sequence(&self, addr: &Address) -> Result { + pub async fn get_sequence(&self, addr: &Address) -> Result + where + T: Send + Sync + 'static, + { let cur_ts = self.current_tipset(); - let sequence = self.get_state_sequence(addr, &cur_ts)?; + let sequence = self.get_state_sequence(addr, &cur_ts).await?; - let resolved = self.resolve_to_key(addr, &cur_ts).ok(); + let resolved = self.resolve_to_key(addr, &cur_ts).await.ok(); let mset = resolved .and_then(|r| self.pending.snapshot_for(&r)) .or_else(|| self.pending.snapshot_for(addr)); @@ -341,11 +374,14 @@ where /// Get the state nonce for an address in `cur_ts`, accounting for /// messages already included in that tipset. Cached by `(TipsetKey, /// Address)`. - pub(in crate::message_pool) fn get_state_sequence( + pub(in crate::message_pool) async fn get_state_sequence( &self, addr: &Address, cur_ts: &Tipset, - ) -> Result { + ) -> Result + where + T: Send + Sync + 'static, + { let nk = StateNonceCacheKey { tipset_key: cur_ts.key().clone(), addr: *addr, @@ -358,15 +394,17 @@ where let actor = self.api.get_actor_after(addr, cur_ts)?; let mut next_nonce = actor.sequence; - if let (Ok(resolved), Ok(messages)) = ( - self.resolve_to_key(addr, cur_ts) - .inspect_err(|e| tracing::warn!(%addr, "failed to resolve address to key: {e:#}")), - self.api - .messages_for_tipset(cur_ts) - .inspect_err(|e| tracing::warn!("failed to get messages for tipset: {e:#}")), - ) { + let resolved = self + .resolve_to_key(addr, cur_ts) + .await + .inspect_err(|e| tracing::warn!(%addr, "failed to resolve address to key: {e:#}")); + let messages = self + .api + .messages_for_tipset(cur_ts) + .inspect_err(|e| tracing::warn!("failed to get messages for tipset: {e:#}")); + if let (Ok(resolved), Ok(messages)) = (resolved, messages) { for msg in messages.iter() { - if let Ok(from) = self.resolve_to_key(&msg.from(), cur_ts).inspect_err( + if let Ok(from) = self.resolve_to_key(&msg.from(), cur_ts).await.inspect_err( |e| tracing::warn!(from = %msg.from(), "failed to resolve message sender: {e:#}"), ) && from == resolved { @@ -405,10 +443,14 @@ where /// Return a Vector of signed messages for a given from address. This vector /// will be sorted by each `message`'s sequence. If no corresponding /// messages found, return None result type. - pub fn pending_for(&self, a: &Address) -> Option> { + pub async fn pending_for(&self, a: &Address) -> Option> + where + T: Send + Sync + 'static, + { let cur_ts = self.current_tipset(); let resolved = self .resolve_to_key(a, &cur_ts) + .await .inspect_err(|e| tracing::debug!(%a, "pending_for: failed to resolve address: {e:#}")) .ok()?; let mset = self.pending.snapshot_for(&resolved)?; @@ -694,12 +736,14 @@ mod tests { ..ShimMessage::default() }; let msg = SignedMessage::mock_bls_signed_message(message); - let res = mpool.add_to_pool_unchecked( - &cur_ts, - msg, - TrustPolicy::Trusted, - StrictnessPolicy::Relaxed, - ); + let res = mpool + .add_to_pool_unchecked( + &cur_ts, + msg, + TrustPolicy::Trusted, + StrictnessPolicy::Relaxed, + ) + .await; assert!(res.is_ok()); } @@ -710,7 +754,7 @@ mod tests { let ts = mpool.current_tipset(); let bls_addr = Address::new_bls(&[1u8; 48]).unwrap(); - let result = mpool.resolve_to_key(&bls_addr, &ts).unwrap(); + let result = mpool.resolve_to_key(&bls_addr, &ts).await.unwrap(); assert_eq!(result, bls_addr); assert_eq!( mpool.caches.key.len(), @@ -729,7 +773,7 @@ mod tests { let (mpool, _services) = make_test_mpool(api); let ts = mpool.current_tipset(); - let result = mpool.resolve_to_key(&id_addr, &ts).unwrap(); + let result = mpool.resolve_to_key(&id_addr, &ts).await.unwrap(); assert_eq!(result, key_addr); assert_eq!( mpool.caches.key.len(), @@ -738,7 +782,7 @@ mod tests { ); // Second call should hit the cache (no API call needed) - let result2 = mpool.resolve_to_key(&id_addr, &ts).unwrap(); + let result2 = mpool.resolve_to_key(&id_addr, &ts).await.unwrap(); assert_eq!(result2, key_addr); } @@ -767,6 +811,7 @@ mod tests { TrustPolicy::Trusted, StrictnessPolicy::Relaxed, ) + .await .unwrap(); assert!( @@ -806,6 +851,7 @@ mod tests { TrustPolicy::Trusted, StrictnessPolicy::Relaxed, ) + .await .unwrap(); } @@ -814,8 +860,8 @@ mod tests { .get_actor_after(&id_addr, &cur_ts) .unwrap() .sequence; - let resolved_for_id = mpool.resolve_to_key(&id_addr, &cur_ts).unwrap(); - let resolved_for_key = mpool.resolve_to_key(&key_addr, &cur_ts).unwrap(); + let resolved_for_id = mpool.resolve_to_key(&id_addr, &cur_ts).await.unwrap(); + let resolved_for_key = mpool.resolve_to_key(&key_addr, &cur_ts).await.unwrap(); assert_eq!(resolved_for_id, resolved_for_key); let next_seq = mpool @@ -844,7 +890,7 @@ mod tests { let (mpool, _services) = make_test_mpool(api); - let nonce = mpool.get_state_sequence(&sender, &ts).unwrap(); + let nonce = mpool.get_state_sequence(&sender, &ts).await.unwrap(); assert_eq!( nonce, 8, "should account for non-consecutive tipset message at nonce 7" @@ -874,13 +920,13 @@ mod tests { let (mpool, _services) = make_test_mpool(api); - let nonce_a = mpool.get_state_sequence(&addr_a, &ts).unwrap(); + let nonce_a = mpool.get_state_sequence(&addr_a, &ts).await.unwrap(); assert_eq!( nonce_a, 0, "addr_a nonce should be unaffected by addr_b's messages" ); - let nonce_b = mpool.get_state_sequence(&addr_b, &ts).unwrap(); + let nonce_b = mpool.get_state_sequence(&addr_b, &ts).await.unwrap(); assert_eq!( nonce_b, 3, "addr_b nonce should reflect its tipset messages" @@ -903,12 +949,12 @@ mod tests { let (mpool, _services) = make_test_mpool(api); - let nonce1 = mpool.get_state_sequence(&sender, &ts).unwrap(); + let nonce1 = mpool.get_state_sequence(&sender, &ts).await.unwrap(); assert_eq!(nonce1, 6); // Mutate the underlying state; the cache should still return the old value. mpool.api.set_state_sequence(&sender, 99); - let nonce2 = mpool.get_state_sequence(&sender, &ts).unwrap(); + let nonce2 = mpool.get_state_sequence(&sender, &ts).await.unwrap(); assert_eq!( nonce2, 6, "second call should return the cached value, not re-read state" @@ -928,7 +974,7 @@ mod tests { let block_a = mock_block(1, 1); let ts_a = Tipset::from(&block_a); - let nonce_a = mpool.get_state_sequence(&sender, &ts_a).unwrap(); + let nonce_a = mpool.get_state_sequence(&sender, &ts_a).await.unwrap(); assert_eq!(nonce_a, 10); // Different tipset should be a cache miss and re-read state. @@ -936,7 +982,7 @@ mod tests { let block_b = mock_block(2, 2); let ts_b = Tipset::from(&block_b); - let nonce_b = mpool.get_state_sequence(&sender, &ts_b).unwrap(); + let nonce_b = mpool.get_state_sequence(&sender, &ts_b).await.unwrap(); assert_eq!( nonce_b, 20, "different tipset should miss the cache and read fresh state" diff --git a/src/message_pool/msgpool/provider.rs b/src/message_pool/msgpool/provider.rs index 8f90a45272bd..7d9614460465 100644 --- a/src/message_pool/msgpool/provider.rs +++ b/src/message_pool/msgpool/provider.rs @@ -144,7 +144,6 @@ impl Provider for ChainStore { } } -#[allow(dead_code)] pub trait ProviderExt { /// Non-blocking version of [`Provider::resolve_to_deterministic_address_at_finality`] async fn resolve_to_deterministic_address_at_finality_async( diff --git a/src/message_pool/msgpool/reorg.rs b/src/message_pool/msgpool/reorg.rs index 128779a5304f..3b114e0b4b29 100644 --- a/src/message_pool/msgpool/reorg.rs +++ b/src/message_pool/msgpool/reorg.rs @@ -18,7 +18,7 @@ use ahash::{HashMap, HashMapExt}; impl MessagePool where - T: Provider + 'static, + T: Provider + Send + Sync + 'static, { /// Revert and/or apply tipsets to the message pool. /// @@ -74,13 +74,15 @@ where }; for msg in smsgs { - self.remove_applied_from_pool(&msg.from(), msg.sequence(), &mut rmsgs, &ts)?; + self.remove_applied_from_pool(&msg.from(), msg.sequence(), &mut rmsgs, &ts) + .await?; if !repub && self.republish.was_republished(&msg.cid()) { repub = true; } } for msg in msgs { - self.remove_applied_from_pool(&msg.from, msg.sequence, &mut rmsgs, &ts)?; + self.remove_applied_from_pool(&msg.from, msg.sequence, &mut rmsgs, &ts) + .await?; if !repub && self.republish.was_republished(&msg.cid()) { repub = true; } @@ -95,12 +97,15 @@ where let cur_ts = self.cur_tipset.read().shallow_clone(); for (_, hm) in rmsgs { for (_, msg) in hm { - if let Err(e) = self.add_to_pool_unchecked( - &cur_ts, - msg, - TrustPolicy::Trusted, - StrictnessPolicy::Relaxed, - ) { + if let Err(e) = self + .add_to_pool_unchecked( + &cur_ts, + msg, + TrustPolicy::Trusted, + StrictnessPolicy::Relaxed, + ) + .await + { tracing::error!("Failed to read message from reorg to mpool: {}", e); } } @@ -113,7 +118,7 @@ where /// message isn't there, fall back to removing it from the real pending /// pool. Used by [`Self::apply_head_change`] when an applied tipset /// includes a message that we hadn't yet seen reverted. - fn remove_applied_from_pool( + async fn remove_applied_from_pool( &self, from: &Address, sequence: u64, @@ -126,6 +131,7 @@ where .is_none() && let Ok(resolved) = self .resolve_to_key(from, ts) + .await .inspect_err(|e| tracing::debug!(%from, "remove: failed to resolve address: {e:#}")) { let _ = self.pending.remove(&resolved, sequence, true); diff --git a/src/message_pool/msgpool/selection.rs b/src/message_pool/msgpool/selection.rs index 1e663dbe1239..b1d1c293ffb2 100644 --- a/src/message_pool/msgpool/selection.rs +++ b/src/message_pool/msgpool/selection.rs @@ -916,11 +916,11 @@ mod test_selection { // order his messages first for i in 0..10 { let m = create_smsg(&a2, &a1, &mut w1, i, TEST_GAS_LIMIT, 2 * i + 1); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } for i in 0..10 { let m = create_smsg(&a1, &a2, &mut w2, i, TEST_GAS_LIMIT, i + 1); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } let msgs = mpool.select_messages(&ts, 1.0).unwrap(); @@ -990,9 +990,11 @@ mod test_selection { TEST_GAS_LIMIT, 2 * i + 200, )) + .await .unwrap(); mpool .add(create_smsg(&a1, &a2, &mut w2, i, TEST_GAS_LIMIT, i + 1)) + .await .unwrap(); } // select messages in the last tipset; this should include the missed messages @@ -1074,6 +1076,7 @@ mod test_selection { TrustPolicy::Trusted, StrictnessPolicy::Relaxed, ) + .await .unwrap(); let before = mpool.pending.snapshot(); @@ -1150,7 +1153,7 @@ mod test_selection { TEST_GAS_LIMIT, (1 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); let m = create_fake_smsg( &mpool, &a1, @@ -1159,7 +1162,7 @@ mod test_selection { TEST_GAS_LIMIT, (1 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } let msgs = mpool.select_messages(&ts, 1.0).unwrap(); @@ -1186,7 +1189,7 @@ mod test_selection { // create a larger than selectable chain for i in 0..BLOCK_MESSAGE_LIMIT { let msg = create_fake_smsg(&mpool, &address, &address, i as u64, 200_000, 100); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); } let msgs = mpool.select_messages(&ts, 1.0).unwrap(); @@ -1233,7 +1236,7 @@ mod test_selection { 300_000, 100, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); // higher has price, those should be preferred and fill the block up to // the [`CBOR_GEN_LIMIT`] messages. let msg = create_smsg( @@ -1244,7 +1247,7 @@ mod test_selection { 300_000, 1000, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); } let msgs = mpool.select_messages(&ts, 1.0).unwrap(); // check that the gas limit is not exceeded @@ -1296,7 +1299,7 @@ mod test_selection { 300_000, 100, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); // higher has price, those should be preferred and fill the block up to // the [`CBOR_GEN_LIMIT`] messages. let msg = create_smsg( @@ -1307,7 +1310,7 @@ mod test_selection { 300_000, 100, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); } // address_1 8192th message is worth more than address_2 8192th message @@ -1319,7 +1322,7 @@ mod test_selection { 300_000, 1000, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); let msg = create_smsg( &address_1, @@ -1329,7 +1332,7 @@ mod test_selection { 300_000, 100, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); counter += 1; @@ -1342,7 +1345,7 @@ mod test_selection { 400_000, 1_000_000, ); - mpool.add(msg).unwrap(); + mpool.add(msg).await.unwrap(); let msgs = mpool.select_messages(&ts, 1.0).unwrap(); // check that the gas limit is not exceeded @@ -1405,7 +1408,7 @@ mod test_selection { TEST_GAS_LIMIT, (1 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); let m = create_smsg( &a1, &a2, @@ -1414,7 +1417,7 @@ mod test_selection { TEST_GAS_LIMIT, (1 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } let msgs = mpool.select_messages(&ts, 1.0).unwrap(); @@ -1477,7 +1480,7 @@ mod test_selection { TEST_GAS_LIMIT, (1 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } let msgs = mpool.select_messages(&ts, 0.25).unwrap(); @@ -1529,7 +1532,7 @@ mod test_selection { TEST_GAS_LIMIT, (200000 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); let m = create_fake_smsg( &mpool, &a1, @@ -1538,7 +1541,7 @@ mod test_selection { TEST_GAS_LIMIT, (190000 + i % 3 + bias) as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } let msgs = mpool.select_messages(&ts, 0.1).unwrap(); @@ -1626,7 +1629,7 @@ mod test_selection { TEST_GAS_LIMIT, premium as u64, ); - mpool.add(m).unwrap(); + mpool.add(m).await.unwrap(); } } diff --git a/src/message_pool/nonce_tracker.rs b/src/message_pool/nonce_tracker.rs index b5341dfbd890..231ed91164e8 100644 --- a/src/message_pool/nonce_tracker.rs +++ b/src/message_pool/nonce_tracker.rs @@ -24,7 +24,7 @@ impl NonceTracker { } /// Acquire the global lock, assign a nonce, sign, and push to `mpool`. - pub async fn sign_and_push( + pub async fn sign_and_push( &self, mpool: &MessagePool, mut message: Message, @@ -33,7 +33,7 @@ impl NonceTracker { ) -> anyhow::Result { let _guard = self.lock.lock().await; - let nonce = mpool.get_sequence(&message.from)?; + let nonce = mpool.get_sequence(&message.from).await?; message.sequence = nonce; let smsg = sign_message(key, &message, eth_chain_id)?; diff --git a/src/rpc/methods/eth.rs b/src/rpc/methods/eth.rs index fbe758a01fca..240ad2c0efaa 100644 --- a/src/rpc/methods/eth.rs +++ b/src/rpc/methods/eth.rs @@ -2432,7 +2432,7 @@ impl RpcMethod<2> for EthGetTransactionCount { let addr = sender.to_filecoin_address()?; match block_param { BlockNumberOrHash::PredefinedBlock(Predefined::Pending) => { - Ok(EthUint64(ctx.mpool.get_sequence(&addr)?)) + Ok(EthUint64(ctx.mpool.get_sequence(&addr).await?)) } _ => { let resolver = TipsetResolver::new(&ctx, Self::api_path(ext)?); diff --git a/src/rpc/methods/gas.rs b/src/rpc/methods/gas.rs index de39530fc2d6..407991db4898 100644 --- a/src/rpc/methods/gas.rs +++ b/src/rpc/methods/gas.rs @@ -221,7 +221,7 @@ impl GasEstimateGasLimit { .resolve_to_deterministic_address(msg.from, &curr_ts) .await?; - let pending = data.mpool.pending_for(&from_a); + let pending = data.mpool.pending_for(&from_a).await; let prior_messages: Arc> = pending .map(|s| s.into_iter().map(Into::into).collect_vec()) .unwrap_or_default() diff --git a/src/rpc/methods/mpool.rs b/src/rpc/methods/mpool.rs index 6154bc3fe1aa..338b4612a3d4 100644 --- a/src/rpc/methods/mpool.rs +++ b/src/rpc/methods/mpool.rs @@ -84,7 +84,7 @@ impl RpcMethod<1> for MpoolGetNonce { (address,): Self::Params, _: &http::Extensions, ) -> Result { - Ok(ctx.mpool.get_sequence(&address)?) + Ok(ctx.mpool.get_sequence(&address).await?) } }