forest/dev/subcommands/tests_cmd/

helpers.rs

// Copyright 2019-2026 ChainSafe Systems
// SPDX-License-Identifier: Apache-2.0, MIT

use std::future::Future;
use std::io::Write as _;
use std::process::Command;
use std::sync::LazyLock;
use std::time::Duration;

use anyhow::{Context as _, bail};
use serde_json::{Value, json};
use tempfile::NamedTempFile;
use tokio::sync::OnceCell;

/// Funded preloaded address from env `FOREST_TEST_PRELOADED_ADDRESS` (`forest_wallet_init` in `scripts/tests/harness.sh`).
pub static FOREST_TEST_PRELOADED_ADDRESS: LazyLock<String> = LazyLock::new(|| {
    std::env::var("FOREST_TEST_PRELOADED_ADDRESS")
        .ok()
        .map(|s| s.trim().to_owned())
        .filter(|s| !s.is_empty())
        .expect("FOREST_TEST_PRELOADED_ADDRESS must be set")
});

/// Test amount to be transferred between accounts in wallet tests.
pub const FIL_AMT: &str = "500 atto FIL";
/// Sentinel `forest-wallet balance --exact-balance` returns for an unfunded address.
pub const FIL_ZERO: &str = "0 FIL";
/// Amount to seed a freshly-created delegated wallet.
pub const DELEGATE_FUND_AMT: &str = "30 micro FIL";

/// Maximum time to wait for a polled condition before failing the test.
pub const POLL_TIMEOUT: Duration = Duration::from_secs(600);
/// Delay between poll attempts.
pub const POLL_WAIT_TIME: Duration = Duration::from_secs(1);

/// Selects which `forest-wallet` keystore an operation targets.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum Backend {
    Local,
    Remote,
}

impl Backend {
    fn extra_args(self) -> &'static [&'static str] {
        match self {
            Self::Local => &[],
            Self::Remote => &["--remote-wallet"],
        }
    }

    pub fn label(self) -> &'static str {
        match self {
            Self::Local => "local",
            Self::Remote => "remote",
        }
    }
}

/// Run `forest-wallet [--remote-wallet] <args>` and return trimmed stdout.
pub fn wallet(backend: Backend, args: &[&str]) -> anyhow::Result<String> {
    Ok(String::from_utf8(run_wallet_raw(backend, args)?)?
        .trim()
        .to_string())
}

/// Same as [`wallet`] but yields raw stdout bytes.
pub fn run_wallet_raw(backend: Backend, args: &[&str]) -> anyhow::Result<Vec<u8>> {
    let mut full = Vec::with_capacity(backend.extra_args().len() + args.len());
    full.extend_from_slice(backend.extra_args());
    full.extend_from_slice(args);
    run("forest-wallet", &full)
}

/// Runs `program args...` and returns raw stdout, with stderr surfaced on failure.
fn run(program: &str, args: &[&str]) -> anyhow::Result<Vec<u8>> {
    let output = Command::new(program)
        .args(args)
        .output()
        .with_context(|| format!("failed to spawn `{program}`"))?;
    if !output.status.success() {
        bail!(
            "`{program} {}` failed (status={}): {}",
            args.join(" "),
            output.status,
            String::from_utf8_lossy(&output.stderr)
        );
    }
    Ok(output.stdout)
}

/// [`run`] with stdout as trimmed UTF-8.
fn run_str(program: &str, args: &[&str]) -> anyhow::Result<String> {
    Ok(String::from_utf8(run(program, args)?)?.trim().to_owned())
}

/// Export `address` from the chosen backend into a temp file ready to feed
/// back to `forest-wallet import`.
pub fn export_to_temp_file(address: &str, backend: Backend) -> anyhow::Result<NamedTempFile> {
    let raw = run_wallet_raw(backend, &["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()?;
    Ok(file)
}

pub fn balance(address: &str, backend: Backend) -> anyhow::Result<String> {
    wallet(backend, &["balance", address, "--exact-balance"])
}

/// Send with `--from`. `backend` chooses the signing keystore
/// (local file vs `--remote-wallet`).
pub fn send_from(from: &str, to: &str, amount: &str, backend: Backend) -> anyhow::Result<String> {
    send_from_and_maybe_wait(from, to, amount, backend, true)
}

pub fn send_from_no_wait(
    from: &str,
    to: &str,
    amount: &str,
    backend: Backend,
) -> anyhow::Result<String> {
    send_from_and_maybe_wait(from, to, amount, backend, false)
}

fn send_from_and_maybe_wait(
    from: &str,
    to: &str,
    amount: &str,
    backend: Backend,
    wait: bool,
) -> anyhow::Result<String> {
    let mut args = vec!["send", to, amount, "--from", from];
    if wait {
        args.extend(["--wait-confidence", "0", "--wait-timeout", "10m"]);
    }
    wallet(backend, &args)
}

/// Max attempts for [`rpc_call_with_retry`].
const RPC_RETRIES: usize = 3;
/// Delay between [`rpc_call_with_retry`] attempts; one block-time at calibnet
/// cadence is enough for the daemon's state snapshot to refresh.
const RPC_RETRY_DELAY: Duration = Duration::from_secs(15);

/// Poll until `try_check` returns `Some` or [`POLL_TIMEOUT`] elapses, sleeping
/// [`POLL_WAIT_TIME`] between attempts.
async fn poll<F, Fut, T>(label: &str, mut try_check: F) -> anyhow::Result<T>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = anyhow::Result<Option<T>>>,
{
    let started = tokio::time::Instant::now();
    let mut attempt = 0u32;
    loop {
        attempt += 1;
        eprintln!("Polling {label} attempt {attempt}");
        if let Some(value) = try_check().await? {
            return Ok(value);
        }
        if started.elapsed() >= POLL_TIMEOUT {
            bail!("Timed out waiting for {label} after {POLL_TIMEOUT:?}");
        }
        let remaining = POLL_TIMEOUT.saturating_sub(started.elapsed());
        tokio::time::sleep(POLL_WAIT_TIME.min(remaining)).await;
    }
}

/// Poll until the balance reported for `address` differs from `baseline`.
pub async fn poll_until_changed(
    address: &str,
    baseline: &str,
    backend: Backend,
) -> anyhow::Result<String> {
    let label = format!("{} balance change for {address}", backend.label());
    let baseline = baseline.to_string();
    poll(&label, || async {
        let bal = balance(address, backend)?;
        Ok((bal != baseline).then_some(bal))
    })
    .await
}

/// Poll until the balance reported for `address` is no longer [`FIL_ZERO`].
pub async fn poll_until_funded(address: &str, backend: Backend) -> anyhow::Result<String> {
    poll_until_changed(address, FIL_ZERO, backend).await
}

/// Run a `lotus` command, retrying while it fails with the transient mpool `check has failed`
/// (the mpool briefly lags the chain head just after the sender is funded, so a submit can be
/// rejected until it catches up). Any other failure propagates immediately.
pub async fn lotus_exec_retrying_mpool(args: &[&str]) -> anyhow::Result<String> {
    poll(&format!("lotus {}", args.join(" ")), || async {
        match lotus_exec(args) {
            Ok(out) => Ok(Some(out)),
            Err(e) if format!("{e:#}").contains("check has failed") => Ok(None),
            Err(e) => Err(e),
        }
    })
    .await
}

/// Delegated signer: create once on local, fund locally, mirror to remote
/// for tests that query or sign.
pub async fn funded_delegated_addr() -> &'static str {
    static FUNDED_DELEGATED: OnceCell<String> = OnceCell::const_new();

    FUNDED_DELEGATED
        .get_or_try_init(|| async {
            let addr = wallet(Backend::Local, &["new", "delegated"]).unwrap();
            let fund_msg = send_from(
                &FOREST_TEST_PRELOADED_ADDRESS,
                &addr,
                DELEGATE_FUND_AMT,
                Backend::Local,
            )
            .unwrap();
            eprintln!("delegated funding send to {addr} msg: {fund_msg}");
            for backend in [Backend::Local, Backend::Remote] {
                let funded = poll_until_funded(&addr, backend).await.unwrap();
                eprintln!(
                    "delegated wallet {addr} funded balance: {funded} ({})",
                    backend.label()
                );
            }

            let exported = export_to_temp_file(&addr, Backend::Local).unwrap();
            let path = exported
                .path()
                .to_str()
                .expect("temp path is not valid UTF-8");
            let mirrored = wallet(Backend::Remote, &["import", path]).unwrap();
            assert_eq!(mirrored, addr, "mirror mismatch: {mirrored} != {addr}");
            Ok::<_, anyhow::Error>(addr)
        })
        .await
        .unwrap()
        .as_str()
}

static HTTP: LazyLock<reqwest::Client> = LazyLock::new(|| {
    reqwest::Client::builder()
        .timeout(Duration::from_secs(120))
        .build()
        .expect("failed to build reqwest client")
});

/// Cached `(token, http_url)` parsed once from `FULLNODE_API_INFO`.
static API: LazyLock<anyhow::Result<(String, String)>> = LazyLock::new(|| {
    let raw = std::env::var("FULLNODE_API_INFO").context("FULLNODE_API_INFO env var not set")?;
    let (token, multiaddr) = raw
        .split_once(':')
        .context("FULLNODE_API_INFO must be `<token>:<multiaddr>`")?;
    let parts: Vec<&str> = multiaddr.split('/').collect();
    let host = parts
        .get(2)
        .filter(|s| !s.is_empty())
        .with_context(|| format!("missing host in multiaddr `{multiaddr}`"))?;
    let port = parts
        .get(4)
        .filter(|s| !s.is_empty())
        .with_context(|| format!("missing port in multiaddr `{multiaddr}`"))?;
    Ok((token.to_string(), format!("http://{host}:{port}/rpc/v1")))
});

fn api() -> anyhow::Result<&'static (String, String)> {
    API.as_ref()
        .map_err(|e| anyhow::anyhow!("FULLNODE_API_INFO unavailable: {e}"))
}

/// Typed client for the Lotus node on the docker devnet, whose read methods need no token.
pub fn lotus_client() -> anyhow::Result<crate::rpc::Client> {
    let port = std::env::var("LOTUS_RPC_PORT")
        .context("LOTUS_RPC_PORT not set; source the devnet test harness")?;
    Ok(crate::rpc::Client::from_url(
        format!("http://127.0.0.1:{port}/").parse()?,
    ))
}

/// Typed client for the Forest node under test, from `FULLNODE_API_INFO`.
pub fn forest_client() -> anyhow::Result<crate::rpc::Client> {
    crate::rpc::Client::default_or_from_env(None)
}

pub fn docker(args: &[&str]) -> anyhow::Result<String> {
    run_str("docker", args).context("is the devnet up?")
}

/// Runs `lotus <args>` inside the `lotus` container on the docker devnet.
pub fn lotus_exec(args: &[&str]) -> anyhow::Result<String> {
    let mut full = vec!["exec", "lotus", "lotus"];
    full.extend_from_slice(args);
    docker(&full)
}

/// POST a JSON-RPC v1 request and return the `result` field, or `None` if
/// the server responded without one.
pub async fn rpc_call_opt(method: &str, params: Value) -> anyhow::Result<Option<Value>> {
    let (token, url) = api()?;
    let body = json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": method,
        "params": params,
    });
    let resp: Value = HTTP
        .post(url)
        .bearer_auth(token)
        .json(&body)
        .send()
        .await
        .with_context(|| format!("POST {url} for {method}"))?
        .error_for_status()
        .with_context(|| format!("HTTP error from {method}"))?
        .json()
        .await
        .with_context(|| format!("decoding JSON-RPC response for {method}"))?;
    if let Some(err) = resp.get("error").filter(|e| !e.is_null()) {
        bail!("RPC error from {method}: {err}");
    }
    match resp.get("result") {
        None => Ok(None),
        Some(v) if v.is_null() => Ok(None),
        Some(v) => Ok(Some(v.clone())),
    }
}

/// 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 < RPC_RETRIES => {
                eprintln!(
                    "error: {e:?} {method} failed on attempt {attempt}/{RPC_RETRIES}, retrying"
                );
                tokio::time::sleep(RPC_RETRY_DELAY).await;
                attempt += 1;
            }
            Err(e) => return Err(e),
        }
    }
}

/// Extract a CID string from either a Lotus `{ "/": "bafy..." }` map or a
/// plain string.
pub fn cid_from_lotus_json_result(result: &Value) -> anyhow::Result<String> {
    if let Some(s) = result.as_str() {
        return Ok(s.to_string());
    }
    result
        .get("/")
        .and_then(|v| v.as_str())
        .map(str::to_owned)
        .with_context(|| format!("expected CID (lotus JSON or string), got {result}"))
}

/// Poll `Filecoin.StateSearchMsg` until the message is mined or [`POLL_TIMEOUT`] elapses.
pub async fn poll_until_state_search_msg(msg_cid: &str) -> anyhow::Result<()> {
    let label = format!("StateSearchMsg for {msg_cid}");
    poll(&label, || async {
        let params = json!([[], { "/": msg_cid }, 800_i64, true]);
        Ok((rpc_call_opt("Filecoin.StateSearchMsg", params)
            .await?
            .is_some())
        .then_some(()))
    })
    .await
}

pub fn forest_cli(args: &[&str]) -> anyhow::Result<String> {
    run_str("forest-cli", args)
}

/// Next nonce for an address
pub 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}"))
}

/// Pending message nonces for `address` via `Filecoin.MpoolPending`.
pub async fn pending_nonces_for(address: &str) -> anyhow::Result<Vec<u64>> {
    let result = rpc_call_with_retry("Filecoin.MpoolPending", json!([null])).await?;
    let entries = result
        .as_array()
        .with_context(|| format!("expected MpoolPending array, got {result}"))?;
    Ok(entries
        .iter()
        .filter_map(|entry| {
            let msg = entry.get("Message")?;
            (msg.get("From")?.as_str()? == address).then_some(msg.get("Nonce")?.as_u64()?)
        })
        .collect())
}

/// Poll until `address` has a pending message at `nonce`.
pub 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 nonces = pending_nonces_for(&address).await?;
        Ok(nonces.contains(&nonce).then_some(()))
    })
    .await
}

/// 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_with_retry(
        "Filecoin.FilecoinAddressToEthAddress",
        json!([address, "pending"]),
    )
    .await?;
    result
        .as_str()
        .map(str::to_owned)
        .with_context(|| format!("expected string ETH address, got {result}"))
}

pub fn block_on<F: Future + Send + 'static>(future: F) -> F::Output
where
    F::Output: Send + 'static,
{
    std::thread::spawn(|| {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(future)
    })
    .join()
    // Preserve the panic message instead of `unwrap`'s `Any { .. }`.
    .unwrap_or_else(|payload| std::panic::resume_unwind(payload))
}