Skip to content

refactor: make MessagePool address resolution at finality non-blocking #7334

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sudo-shashank merged 3 commits into from
Jul 10, 2026
Merged

refactor: make MessagePool address resolution at finality non-blocking #7334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5b4c969
Use non-blocking resolve_to_deterministic_address_at_finality_async
sudo-shashank Jul 9, 2026
a908fb8
fmt
sudo-shashank Jul 9, 2026
9583c1f
rpc call with retry with error log
sudo-shashank Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/chain_sync/chain_follower.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
28 changes: 21 additions & 7 deletions src/dev/subcommands/tests_cmd/helpers.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,11 +303,25 @@ pub async fn rpc_call_opt(method: &str, params: Value) -> anyhow::Result<Option<
}
}

/// Like [`rpc_call_opt`] but treats a missing `result` as an error.
pub async fn rpc_call(method: &str, params: Value) -> anyhow::Result<Value> {
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<Value> {
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),
}
}
Comment thread
sudo-shashank marked this conversation as resolved.
}

/// Extract a CID string from either a Lotus `{ "/": "bafy..." }` map or a
Expand Down Expand Up @@ -362,7 +376,7 @@ pub fn mpool_nonce(address: &str) -> anyhow::Result<u64> {

/// Pending message nonces for `address` via `Filecoin.MpoolPending`.
pub async fn pending_nonces_for(address: &str) -> anyhow::Result<Vec<u64>> {
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}"))?;
Expand All @@ -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<String> {
let result = rpc_call(
let result = rpc_call_with_retry(
"Filecoin.FilecoinAddressToEthAddress",
json!([address, "pending"]),
)
Expand Down
2 changes: 1 addition & 1 deletion src/dev/subcommands/tests_cmd/wallet.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
59 changes: 31 additions & 28 deletions src/message_pool/msgpool/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
);
}
Expand Down Expand Up @@ -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);

Expand All @@ -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]
Expand Down Expand Up @@ -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);

Expand All @@ -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);

Expand All @@ -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);

Expand All @@ -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);
Expand All @@ -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]
Expand All @@ -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);
}
Expand All @@ -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!(
Expand Down Expand Up @@ -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.
Expand All @@ -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);
Expand All @@ -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());
Expand Down
Loading
Loading