Fix calibnet wallet test flakiness

tests/calibnet_mpool_tools.rs

@@ -9,9 +9,48 @@
 #[path = "common/calibnet_wallet_helpers.rs"]
 mod helpers;
 
+use anyhow::Context as _;
 use helpers::*;
+use rstest::rstest;
+use serde_json::json;
 use serial_test::serial;
 
+/// Run `forest-cli <args>` and return trimmed stdout.
+fn forest_cli(args: &[&str]) -> anyhow::Result<String> {
+    Ok(String::from_utf8(run_command("forest-cli", args)?)?
+        .trim()
+        .to_string())
+}
+
+/// Next nonce for an address
+fn mpool_nonce(address: &str) -> anyhow::Result<u64> {
+    let out = forest_cli(&["mpool", "nonce", address])?;
+    out.parse::<u64>()
+        .with_context(|| format!("invalid mpool nonce output: {out}"))
+}
+
+/// Poll until `address` has a pending message at `nonce`.
+async fn poll_until_pending_nonce(address: &str, nonce: u64) -> anyhow::Result<()> {
+    let label = format!("pending nonce {nonce} for {address}");
+    let address = address.to_string();
+    poll(&label, || async {
+        let result = rpc_call("Filecoin.MpoolPending", json!([null])).await?;
+        let pending = result
+            .as_array()
+            .with_context(|| format!("expected MpoolPending array, got {result}"))?
+            .iter()
+            .any(|entry| {
+                let Some(msg) = entry.get("Message") else {
+                    return false;
+                };
+                msg.get("From").and_then(|v| v.as_str()) == Some(address.as_str())
+                    && msg.get("Nonce").and_then(|v| v.as_u64()) == Some(nonce)
+            });
+        Ok(pending.then_some(()))
+    })
+    .await
+}
+
 #[tokio::test]
 #[serial]
 async fn mpool_nonce_fix_auto_unblocks_pending() {
@@ -40,13 +79,16 @@ async fn mpool_nonce_fix_auto_unblocks_pending() {
     );
 }
 
+#[rstest]
+#[case::local(Backend::Local)]
+#[case::remote(Backend::Remote)]
 #[tokio::test]
 #[serial]
-async fn mpool_replace_auto_unblocks_pending() {
+async fn mpool_replace_auto_unblocks_pending(#[case] backend: Backend) {
     let addr = FOREST_TEST_PRELOADED_ADDRESS.as_str();
     let nonce = mpool_nonce(addr).unwrap();
 
-    let cid = send_from(addr, addr, FIL_AMT, Backend::Local).unwrap();
+    let cid = backend.send(addr, addr, FIL_ZERO).unwrap();
     poll_until_pending_nonce(addr, nonce).await.unwrap();
 
     forest_cli(&[
@@ -61,7 +103,7 @@ async fn mpool_replace_auto_unblocks_pending() {
     .unwrap();
 
     assert!(
-        poll_until_state_search_msg(&cid).await.is_ok(),
+        wait_for_msg(&cid).await.is_ok(),
         "mpool replace --auto should replace message {cid} from {addr} at nonce {nonce}."
     );
 }

tests/calibnet_wallet.rs

@@ -10,31 +10,99 @@
 #[path = "common/calibnet_wallet_helpers.rs"]
 mod helpers;
 
+use std::io::Write as _;
+
+use anyhow::Context as _;
 use helpers::*;
 use rstest::rstest;
 use serde_json::json;
+use tempfile::NamedTempFile;
+use tokio::sync::OnceCell;
+
+/// Test amount to be transferred between accounts in wallet tests.
+const FIL_AMT: &str = "500 atto FIL";
+/// Amount to seed a freshly-created delegated wallet.
+const DELEGATE_FUND_AMT: &str = "3 micro FIL";
+
+static FUNDED_DELEGATED: OnceCell<String> = OnceCell::const_new();
+
+/// Delegated signer: create once on local, fund locally, mirror to remote.
+async fn funded_delegated_addr() -> &'static str {
+    let addr = FUNDED_DELEGATED
+        .get_or_try_init(|| async {
+            let addr = Backend::Local.run(&["new", "delegated"]).unwrap();
+            let fund_msg = Backend::Local
+                .send(&FOREST_TEST_PRELOADED_ADDRESS, &addr, DELEGATE_FUND_AMT)
+                .unwrap();
+            eprintln!("delegated funding send to {addr} msg: {fund_msg}");
+            wait_for_msg(&fund_msg).await.unwrap();
+            let funded = poll_until_funded(&addr, Backend::Local).await.unwrap();
+            eprintln!("delegated wallet {addr} funded balance: {funded}");
+
+            let mirrored = import_wallet(&addr, Backend::Local, Backend::Remote).unwrap();
+            assert_eq!(mirrored, addr, "mirror mismatch: {mirrored} != {addr}");
+            Ok::<_, anyhow::Error>(addr)
+        })
+        .await
+        .unwrap();
+    addr.as_str()
+}
+
+impl Backend {
+    /// Exact on-chain balance of `address`.
+    fn balance(self, address: &str) -> anyhow::Result<String> {
+        self.run(&["balance", address, "--exact-balance"])
+    }
+}
+
+/// Poll until the balance reported for `address` differs from `baseline`.
+async fn poll_until_changed(
+    address: &str,
+    baseline: &str,
+    backend: Backend,
+) -> anyhow::Result<String> {
+    let label = format!("{backend:?} balance change for {address}");
+    let baseline = baseline.to_string();
+    poll(&label, || async {
+        let bal = backend.balance(address)?;
+        Ok((bal != baseline).then_some(bal))
+    })
+    .await
+}
+
+/// Poll until the balance reported for `address` is no longer [`FIL_ZERO`].
+async fn poll_until_funded(address: &str, backend: Backend) -> anyhow::Result<String> {
+    poll_until_changed(address, FIL_ZERO, backend).await
+}
+
+/// Import an address from one backend into another.
+fn import_wallet(address: &str, from: Backend, to: Backend) -> anyhow::Result<String> {
+    let raw = from.run_raw(&["export", address])?;
+    let mut file = NamedTempFile::new_in(std::env::temp_dir())
+        .context("failed to create temp file for wallet export")?;
+    file.write_all(&raw)?;
+    file.flush()?;
+    let path = file
+        .path()
+        .to_str()
+        .context("temp path is not valid UTF-8")?;
+
+    if to == from {
+        to.run(&["delete", address])?;
+    }
+    to.run(&["import", path])
+}
 
 #[rstest]
 #[case::local(Backend::Local)]
 #[case::remote(Backend::Remote)]
 #[tokio::test]
 async fn export_import_roundtrip(#[case] backend: Backend) {
-    let addr = wallet(backend, &["new"]).unwrap();
-    let exported = export_to_temp_file(&addr, backend).unwrap();
-    let path = exported
-        .path()
-        .to_str()
-        .expect("temp path is not valid UTF-8");
-
-    let deleted = wallet(backend, &["delete", &addr]).unwrap();
-    eprintln!("delete output ({}): {deleted}", backend.label());
-
-    let imported = wallet(backend, &["import", path]).unwrap();
+    let addr = backend.run(&["new"]).unwrap();
+    let imported = import_wallet(&addr, backend, backend).unwrap();
     assert_eq!(
-        imported,
-        addr,
-        "round-trip mismatch on {} backend: {imported} != {addr}",
-        backend.label(),
+        imported, addr,
+        "round-trip mismatch on {backend:?}: {imported} != {addr}",
     );
 }
 
@@ -51,18 +119,24 @@ async fn market_add_balance_message_on_chain() {
     )
     .await
     .unwrap();
-    let msg_cid = cid_from_lotus_json_result(&result).unwrap();
-    poll_until_state_search_msg(&msg_cid).await.unwrap();
+    let msg_cid = result
+        .get("/")
+        .and_then(|v| v.as_str())
+        .expect("MarketAddBalance should return a CID");
+    wait_for_msg(msg_cid).await.unwrap();
 }
 
 #[rstest]
 #[case::local(Backend::Local)]
 #[case::remote(Backend::Remote)]
 #[tokio::test]
 async fn send_to_filecoin_address(#[case] backend: Backend) {
-    let target = wallet(backend, &["new"]).unwrap();
-    let msg = send_from(&FOREST_TEST_PRELOADED_ADDRESS, &target, FIL_AMT, backend).unwrap();
-    eprintln!("send to {target} ({}) msg: {msg}", backend.label());
+    let target = backend.run(&["new"]).unwrap();
+    let msg = backend
+        .send(&FOREST_TEST_PRELOADED_ADDRESS, &target, FIL_AMT)
+        .unwrap();
+    eprintln!("send to {target} ({backend:?}) msg: {msg}");
+    wait_for_msg(&msg).await.unwrap();
     let funded = poll_until_funded(&target, backend).await.unwrap();
     eprintln!("{target} funded balance: {funded}");
 }
@@ -72,17 +146,26 @@ async fn send_to_filecoin_address(#[case] backend: Backend) {
 #[case::remote(Backend::Remote)]
 #[tokio::test]
 async fn send_to_eth_equivalent(#[case] backend: Backend) {
-    let target = wallet(backend, &["new"]).unwrap();
-    let initial_msg = send_from(&FOREST_TEST_PRELOADED_ADDRESS, &target, FIL_AMT, backend).unwrap();
-    eprintln!(
-        "initial send to {target} ({}) msg: {initial_msg}",
-        backend.label(),
-    );
+    let target = backend.run(&["new"]).unwrap();
+    let initial_msg = backend
+        .send(&FOREST_TEST_PRELOADED_ADDRESS, &target, FIL_AMT)
+        .unwrap();
+    eprintln!("initial send to {target} ({backend:?}) msg: {initial_msg}");
+    wait_for_msg(&initial_msg).await.unwrap();
     let baseline = poll_until_funded(&target, backend).await.unwrap();
 
-    let eth = filecoin_to_eth(&target).await.unwrap();
-    let eth_msg = send_from(&FOREST_TEST_PRELOADED_ADDRESS, &eth, FIL_AMT, backend).unwrap();
+    let eth_result = rpc_call(
+        "Filecoin.FilecoinAddressToEthAddress",
+        json!([&target, "pending"]),
+    )
+    .await
+    .unwrap();
+    let eth = eth_result.as_str().expect("expected string ETH address");
+    let eth_msg = backend
+        .send(&FOREST_TEST_PRELOADED_ADDRESS, eth, FIL_AMT)
+        .unwrap();
     eprintln!("send to ETH {eth} (mapped from {target}) msg: {eth_msg}");
+    wait_for_msg(&eth_msg).await.unwrap();
 
     let updated = poll_until_changed(&target, &baseline, backend)
         .await
@@ -98,10 +181,10 @@ async fn send_to_eth_equivalent(#[case] backend: Backend) {
 #[case::remote(Backend::Remote)]
 #[tokio::test]
 async fn wallet_delete(#[case] backend: Backend) {
-    let addr = wallet(backend, &["new"]).unwrap();
-    let deleted = wallet(backend, &["delete", &addr]).unwrap();
-    eprintln!("delete output ({}): {deleted}", backend.label());
-    let listing = wallet(backend, &["list"]).unwrap();
+    let addr = backend.run(&["new"]).unwrap();
+    let deleted = backend.run(&["delete", &addr]).unwrap();
+    eprintln!("delete output ({backend:?}): {deleted}");
+    let listing = backend.run(&["list"]).unwrap();
     assert!(
         !listing.contains(&addr),
         "deleted wallet {addr} still appears in `list`:\n{listing}",
@@ -114,14 +197,12 @@ async fn wallet_delete(#[case] backend: Backend) {
 #[tokio::test]
 async fn delegated_send(#[case] target_backend: Backend) {
     let funded = funded_delegated_addr().await;
-    let target = wallet(target_backend, &["new", "delegated"]).unwrap();
+    let target = target_backend.run(&["new", "delegated"]).unwrap();
     // Baseline `FIL_ZERO` ⇒ first credit; otherwise expect a balance delta.
-    let baseline = balance(&target, target_backend).unwrap();
-    let msg = send_from(funded, &target, FIL_AMT, Backend::Local).unwrap();
-    eprintln!(
-        "delegated send to {target} ({}) msg: {msg}",
-        target_backend.label(),
-    );
+    let baseline = target_backend.balance(&target).unwrap();
+    let msg = Backend::Local.send(funded, &target, FIL_AMT).unwrap();
+    eprintln!("delegated send to {target} ({target_backend:?}) msg: {msg}");
+    wait_for_msg(&msg).await.unwrap();
     let observed = if baseline == FIL_ZERO {
         poll_until_funded(&target, target_backend).await.unwrap()
     } else {
@@ -138,10 +219,11 @@ async fn delegated_send(#[case] target_backend: Backend) {
 #[tokio::test]
 async fn delegated_remote_send() {
     let funded = funded_delegated_addr().await;
-    let target = wallet(Backend::Remote, &["new", "delegated"]).unwrap();
-    let baseline = balance(&target, Backend::Remote).unwrap();
-    let msg = send_from(funded, &target, FIL_AMT, Backend::Remote).unwrap();
+    let target = Backend::Remote.run(&["new", "delegated"]).unwrap();
+    let baseline = Backend::Remote.balance(&target).unwrap();
+    let msg = Backend::Remote.send(funded, &target, FIL_AMT).unwrap();
     eprintln!("delegated --remote-wallet send to {target} msg: {msg}");
+    wait_for_msg(&msg).await.unwrap();
     let observed = if baseline == FIL_ZERO {
         poll_until_funded(&target, Backend::Remote).await.unwrap()
     } else {