gateway_tests/

main.rs

#![deny(clippy::pedantic)]

use std::collections::BTreeMap;
use std::fs::{remove_dir_all, remove_file};
use std::ops::ControlFlow;
use std::path::PathBuf;
use std::str::FromStr;
use std::time::Duration;
use std::{env, ffi};

use anyhow::Context;
use clap::{Parser, Subcommand};
use devimint::cli::cleanup_on_exit;
use devimint::envs::FM_DATA_DIR_ENV;
use devimint::external::{Bitcoind, Esplora};
use devimint::federation::Federation;
use devimint::util::{ProcessManager, almost_equal, poll, poll_with_timeout};
use devimint::version_constants::{VERSION_0_10_0_ALPHA, VERSION_0_12_0_ALPHA};
use devimint::{Gatewayd, LightningNode, cli, util};
use fedimint_core::config::FederationId;
use fedimint_core::time::now;
use fedimint_core::{Amount, BitcoinAmountOrAll, bitcoin, default_esplora_server};
use fedimint_gateway_common::{FederationInfo, PaymentDetails, PaymentKind, PaymentStatus};
use fedimint_logging::LOG_TEST;
use fedimint_testing_core::node_type::LightningNodeType;
use itertools::Itertools;
use tracing::info;

#[derive(Parser)]
struct GatewayTestOpts {
    #[clap(subcommand)]
    test: GatewayTest,
}

#[derive(Debug, Clone, Subcommand)]
#[allow(clippy::enum_variant_names)]
enum GatewayTest {
    ConfigTest {
        #[arg(long = "gw-type")]
        gateway_type: LightningNodeType,
    },
    BackupRestoreTest,
    LiquidityTest,
    EsploraTest,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let opts = GatewayTestOpts::parse();
    match opts.test {
        GatewayTest::ConfigTest { gateway_type } => Box::pin(config_test(gateway_type)).await,
        GatewayTest::BackupRestoreTest => Box::pin(backup_restore_test()).await,
        GatewayTest::LiquidityTest => Box::pin(liquidity_test()).await,
        GatewayTest::EsploraTest => esplora_test().await,
    }
}

async fn backup_restore_test() -> anyhow::Result<()> {
    Box::pin(
        devimint::run_devfed_test().call(|dev_fed, process_mgr| async move {
            let gw = if devimint::util::supports_lnv2() {
                dev_fed.gw_ldk_connected().await?
            } else {
                dev_fed.gw_lnd_registered().await?
            };

            let fed = dev_fed.fed().await?;
            fed.pegin_gateways(10_000_000, vec![gw]).await?;

            let mnemonic = gw.client().get_mnemonic().await?.mnemonic;

            // Recover without a backup
            info!(target: LOG_TEST, "Wiping gateway and recovering without a backup...");
            let ln = gw.ln.clone();
            let new_gw = stop_and_recover_gateway(
                process_mgr.clone(),
                mnemonic.clone(),
                gw.to_owned(),
                ln.clone(),
                fed,
            )
            .await?;

            // Recover with a backup
            info!(target: LOG_TEST, "Wiping gateway and recovering with a backup...");
            info!(target: LOG_TEST, "Creating backup...");
            new_gw.client().backup_to_fed(fed).await?;
            stop_and_recover_gateway(process_mgr, mnemonic, new_gw, ln, fed).await?;

            info!(target: LOG_TEST, "backup_restore_test successful");
            Ok(())
        }),
    )
    .await
}

async fn stop_and_recover_gateway(
    process_mgr: ProcessManager,
    mnemonic: Vec<String>,
    old_gw: Gatewayd,
    new_ln: LightningNode,
    fed: &Federation,
) -> anyhow::Result<Gatewayd> {
    let gateway_balances = old_gw.client().get_balances().await?;
    let before_onchain_balance = gateway_balances.onchain_balance_sats;

    // Stop the Gateway
    let gw_type = old_gw.ln.ln_type();
    let gw_name = old_gw.gw_name.clone();
    let old_gw_index = old_gw.gateway_index;
    old_gw.terminate().await?;
    info!(target: LOG_TEST, "Terminated Gateway");

    // Delete the gateway's database
    let data_dir: PathBuf = env::var(FM_DATA_DIR_ENV)
        .expect("Data dir is not set")
        .parse()
        .expect("Could not parse data dir");
    let gw_db = data_dir.join(gw_name.clone()).join("gatewayd.db");
    if gw_db.is_file() {
        // db is single file on redb
        remove_file(gw_db)?;
    } else {
        remove_dir_all(gw_db)?;
    }
    info!(target: LOG_TEST, "Deleted the Gateway's database");

    if gw_type == LightningNodeType::Ldk {
        // Delete LDK's database as well
        let ldk_data_dir = data_dir.join(gw_name).join("ldk_node");
        remove_dir_all(ldk_data_dir)?;
        info!(target: LOG_TEST, "Deleted LDK's database");
    }

    let seed = mnemonic.join(" ");
    // TODO: Audit that the environment access only happens in single-threaded code.
    unsafe { std::env::set_var("FM_GATEWAY_MNEMONIC", seed) };
    let new_gw = Gatewayd::new(&process_mgr, new_ln, old_gw_index).await?;
    let new_mnemonic = new_gw.client().get_mnemonic().await?.mnemonic;
    assert_eq!(mnemonic, new_mnemonic);
    info!(target: LOG_TEST, "Verified mnemonic is the same after creating new Gateway");

    let federations = serde_json::from_value::<Vec<FederationInfo>>(
        new_gw.client().get_info().await?["federations"].clone(),
    )?;
    assert_eq!(0, federations.len());
    info!(target: LOG_TEST, "Verified new Gateway has no federations");

    new_gw.client().recover_fed(fed).await?;

    let gateway_balances = new_gw.client().get_balances().await?;
    let ecash_balance = gateway_balances
        .ecash_balances
        .first()
        .expect("Should have one joined federation");
    almost_equal(
        ecash_balance.ecash_balance_msats.sats_round_down(),
        10_000_000,
        10,
    )
    .unwrap();
    let after_onchain_balance = gateway_balances.onchain_balance_sats;
    assert_eq!(before_onchain_balance, after_onchain_balance);
    info!(target: LOG_TEST, "Verified balances after recovery");

    Ok(new_gw)
}

/// Test that sets and verifies configurations within the gateway
#[allow(clippy::too_many_lines)]
async fn config_test(gw_type: LightningNodeType) -> anyhow::Result<()> {
    Box::pin(
        devimint::run_devfed_test()
            .num_feds(2)
            .call(|dev_fed, process_mgr| async move {
                let gw = match gw_type {
                    LightningNodeType::Lnd => dev_fed.gw_lnd_registered().await?,
                    LightningNodeType::Ldk => dev_fed.gw_ldk_connected().await?,
                };

                // Try to connect to already connected federation
                let invite_code = dev_fed.fed().await?.invite_code()?;
                gw.client().connect_fed(invite_code).await.expect_err("Connecting to the same federation succeeded");
                info!(target: LOG_TEST, "Verified that gateway couldn't connect to already connected federation");

                // Change the routing fees for a specific federation
                let fed_id = dev_fed.fed().await?.calculate_federation_id();
                gw.client().set_federation_routing_fee(fed_id.clone(), 20, 20000)
                    .await?;

                let lightning_fee = gw.client().get_lightning_fee(fed_id.clone()).await?;
                assert_eq!(
                    lightning_fee.base.msats, 20,
                    "Federation base msat is not 20"
                );
                assert_eq!(
                    lightning_fee.parts_per_million, 20000,
                    "Federation proportional millionths is not 20000"
                );
                info!(target: LOG_TEST, "Verified per-federation routing fees changed");

                let info_value = gw.client().get_info().await?;
                let federations = info_value["federations"]
                    .as_array()
                    .expect("federations is an array");
                assert_eq!(
                    federations.len(),
                    1,
                    "Gateway did not have one connected federation"
                );

                // Get the federation's config and verify it parses correctly
                gw.client().client_config(fed_id.clone()).await?;

                // Spawn new federation
                let bitcoind = dev_fed.bitcoind().await?;
                let new_fed = Federation::new(
                    &process_mgr,
                    bitcoind.clone(),
                    false,
                    false,
                    1,
                    "config-test".to_string(),
                )
                .await?;
                let new_fed_id = new_fed.calculate_federation_id();
                info!(target: LOG_TEST, "Successfully spawned new federation");

                let new_invite_code = new_fed.invite_code()?;
                gw.client().connect_fed(new_invite_code.clone()).await?;

                let default_base = 0;
                let default_ppm = 0;

                let lightning_fee = gw.client().get_lightning_fee(new_fed_id.clone()).await?;
                assert_eq!(
                    lightning_fee.base.msats, default_base,
                    "Default Base msat for new federation was not correct"
                );
                assert_eq!(
                    lightning_fee.parts_per_million, default_ppm,
                    "Default Base msat for new federation was not correct"
                );

                info!(target: LOG_TEST, federation_id = %new_fed_id, "Verified new federation");

                // Peg-in sats to gw for the new fed
                let pegin_amount = Amount::from_msats(10_000_000);
                new_fed
                    .pegin_gateways(pegin_amount.sats_round_down(), vec![gw])
                    .await?;

                // Verify `info` returns multiple federations
                let info_value = gw.client().get_info().await?;
                let federations = info_value["federations"]
                    .as_array()
                    .expect("federations is an array");

                assert_eq!(
                    federations.len(),
                    2,
                    "Gateway did not have two connected federations"
                );

                let federation_fake_scids =
                    serde_json::from_value::<Option<BTreeMap<u64, FederationId>>>(
                        info_value
                            .get("channels")
                            .or_else(|| info_value.get("federation_fake_scids"))
                            .expect("field  exists")
                            .to_owned(),
                    )
                    .expect("cannot parse")
                    .expect("should have scids");

                assert_eq!(
                    federation_fake_scids.keys().copied().collect::<Vec<u64>>(),
                    vec![1, 2]
                );

                let first_fed_info = federations
                    .iter()
                    .find(|i| {
                        *i["federation_id"]
                            .as_str()
                            .expect("should parse as str")
                            .to_string()
                            == fed_id
                    })
                    .expect("Could not find federation");

                let second_fed_info = federations
                    .iter()
                    .find(|i| {
                        *i["federation_id"]
                            .as_str()
                            .expect("should parse as str")
                            .to_string()
                            == new_fed_id
                    })
                    .expect("Could not find federation");

                let first_fed_balance_msat =
                    serde_json::from_value::<Amount>(first_fed_info["balance_msat"].clone())
                        .expect("fed should have balance");

                let second_fed_balance_msat =
                    serde_json::from_value::<Amount>(second_fed_info["balance_msat"].clone())
                        .expect("fed should have balance");

                assert_eq!(first_fed_balance_msat, Amount::ZERO);
                almost_equal(second_fed_balance_msat.msats, pegin_amount.msats, 10_000).unwrap();

                let fed_id = FederationId::from_str(&fed_id).expect("invalid Federation ID");
                let fed_info = gw.client().leave_federation(fed_id).await?;
                assert_eq!(serde_json::from_value::<FederationId>(fed_info["federation_id"].clone())?, fed_id);
                assert_eq!(fed_info["config"]["federation_index"].as_u64().expect("Was not u64"), 1);
                gw.client().leave_federation(fed_id).await.expect_err("Successfully left a federation twice");

                let new_fed_id = FederationId::from_str(&new_fed_id).expect("invalid Federation ID");
                let fed_info = gw.client().leave_federation(new_fed_id).await?;
                assert_eq!(serde_json::from_value::<FederationId>(fed_info["federation_id"].clone())?, new_fed_id);
                assert_eq!(fed_info["config"]["federation_index"].as_u64().expect("Was not u64"), 2);

                // Rejoin new federation, verify that the balance is the same
                let fed_info = gw.client().connect_fed(new_invite_code).await?;
                assert_eq!(second_fed_balance_msat, Amount::from_msats(fed_info["balance_msat"].as_u64().expect("Balance should be present")));

                if gw.gatewayd_version >= *VERSION_0_10_0_ALPHA {
                    // Try to get the info over iroh
                    info!(target: LOG_TEST, gatewayd_version = %gw.gatewayd_version, "Getting info over iroh");
                    gw.client().with_iroh().get_info().await?;
                }

                info!(target: LOG_TEST, "Gateway configuration test successful");
                Ok(())
            }),
    )
    .await
}

/// Test that verifies the various liquidity tools (onchain, lightning, ecash)
/// work correctly.
#[allow(clippy::too_many_lines)]
async fn liquidity_test() -> anyhow::Result<()> {
    devimint::run_devfed_test()
        .call(|dev_fed, _process_mgr| async move {
            let federation = dev_fed.fed().await?;

            if !devimint::util::supports_lnv2() {
                info!(target: LOG_TEST, "LNv2 is not supported, which is necessary for LDK GW and liquidity test");
                return Ok(());
            }

            let gw_lnd = dev_fed.gw_lnd_registered().await?;
            let gw_ldk = dev_fed.gw_ldk_connected().await?;
            let gw_ldk_second = dev_fed.gw_ldk_second_connected().await?;
            let gateways = [gw_lnd, gw_ldk].to_vec();

            let gateway_matrix = gateways
                .iter()
                .cartesian_product(gateways.iter())
                .filter(|(a, b)| a.ln.ln_type() != b.ln.ln_type());

            info!(target: LOG_TEST, "Pegging-in gateways...");
            federation
                .pegin_gateways(1_000_000, gateways.clone())
                .await?;

            info!(target: LOG_TEST, "Testing ecash payments between gateways...");
            for (gw_send, gw_receive) in gateway_matrix.clone() {
                info!(
                    target: LOG_TEST,
                    gw_send = %gw_send.ln.ln_type(),
                    gw_receive = %gw_receive.ln.ln_type(),
                    "Testing ecash payment",
                );

                let fed_id = federation.calculate_federation_id();
                let prev_send_ecash_balance = gw_send.client().ecash_balance(fed_id.clone()).await?;
                let prev_receive_ecash_balance = gw_receive.client().ecash_balance(fed_id.clone()).await?;
                let ecash = gw_send.client().send_ecash(fed_id.clone(), 500_000).await?;
                gw_receive.client().receive_ecash(ecash).await?;
                let after_send_ecash_balance = gw_send.client().ecash_balance(fed_id.clone()).await?;
                almost_equal(
                    prev_send_ecash_balance - 500_000,
                    after_send_ecash_balance,
                    if util::supports_mint_v2() { 2_000 } else { 512 },
                )
                .expect("Balances were not almost equal");

                poll_with_timeout(
                    "receive ecash balance",
                    Duration::from_secs(30),
                    || async {
                        let balance = gw_receive.client().ecash_balance(fed_id.clone()).await
                            .map_err(ControlFlow::Break)?;
                        almost_equal(prev_receive_ecash_balance + 500_000, balance, 2_000)
                            .map_err(|e| ControlFlow::Continue(anyhow::anyhow!(e)))
                    },
                )
                .await?;
            }

            info!(target: LOG_TEST, "Testing payments between gateways...");
            for (gw_send, gw_receive) in gateway_matrix.clone() {
                info!(
                    target: LOG_TEST,
                    gw_send = %gw_send.ln.ln_type(),
                    gw_receive = %gw_receive.ln.ln_type(),
                    "Testing lightning payment",
                );

                let invoice = gw_receive.client().create_invoice(1_000_000).await?;
                gw_send.client().pay_invoice(invoice).await?;
            }

            let start = now() - Duration::from_mins(5);
            let end = now() + Duration::from_mins(5);
            info!(target: LOG_TEST, "Verifying list of transactions");
            let lnd_transactions = gw_lnd.client().list_transactions(start, end).await?;
            // One inbound and one outbound transaction
            assert_eq!(lnd_transactions.len(), 2);

            let ldk_transactions = gw_ldk.client().list_transactions(start, end).await?;
            assert_eq!(ldk_transactions.len(), 2);

            // Verify that transactions are filtered by time
            let start = now() - Duration::from_mins(10);
            let end = now() - Duration::from_mins(5);
            let lnd_transactions = gw_lnd.client().list_transactions(start, end).await?;
            assert_eq!(lnd_transactions.len(), 0);

            info!(target: LOG_TEST, "Testing paying Bolt12 Offers...");
            let offer_with_amount = gw_ldk_second.client().create_offer(Some(Amount::from_msats(10_000_000))).await?;
            gw_ldk.client().pay_offer(offer_with_amount, None).await?;
            assert!(get_transaction(gw_ldk_second, PaymentKind::Bolt12Offer, Amount::from_msats(10_000_000), PaymentStatus::Succeeded).await.is_some());

            let offer_without_amount = gw_ldk.client().create_offer(None).await?;
            gw_ldk_second.client().pay_offer(offer_without_amount.clone(), Some(Amount::from_msats(5_000_000))).await?;
            assert!(get_transaction(gw_ldk, PaymentKind::Bolt12Offer, Amount::from_msats(5_000_000), PaymentStatus::Succeeded).await.is_some());

            // Cannot pay an offer without an amount without specifying an amount
            gw_ldk_second.client().pay_offer(offer_without_amount.clone(), None).await.expect_err("Cannot pay amountless offer without specifying an amount");

            // Verify we can pay the offer again
            gw_ldk_second.client().pay_offer(offer_without_amount, Some(Amount::from_msats(3_000_000))).await?;
            assert!(get_transaction(gw_ldk, PaymentKind::Bolt12Offer, Amount::from_msats(3_000_000), PaymentStatus::Succeeded).await.is_some());

            // `set-channel-fees` was added in 0.12.0-alpha for both the gateway
            // API and the CLI. Skip the test against any older gateway/CLI binary
            // to keep this test backwards-compatible with prior releases.
            let gateway_cli_version = util::GatewayCli::version_or_default().await;
            let all_gateways_support_fees = gateways
                .iter()
                .all(|gw| gw.gatewayd_version >= *VERSION_0_12_0_ALPHA);
            if gateway_cli_version >= *VERSION_0_12_0_ALPHA && all_gateways_support_fees {
                info!(target: LOG_TEST, "Testing updating channel fees on both gateways...");
                for gw in &gateways {
                    let channels = gw.client().list_channels().await?;
                    let channel = channels
                        .into_iter()
                        .find(|c| c.funding_outpoint.is_some())
                        .with_context(|| {
                            format!(
                                "{} gateway has no channel with a known funding outpoint",
                                gw.ln.ln_type(),
                            )
                        })?;
                    let funding_outpoint = channel.funding_outpoint.expect("filtered above");

                    // Pick values that are unlikely to collide with any backend default.
                    let new_base_fee_msat = 12_345u64;
                    let new_parts_per_million = 678u64;

                    gw.client()
                        .set_channel_fees(
                            funding_outpoint,
                            new_base_fee_msat,
                            new_parts_per_million,
                        )
                        .await?;

                    // Both backends report local config synchronously, but poll briefly
                    // in case the LND policy update needs a moment to be visible to
                    // `fee_report`.
                    poll_with_timeout(
                        "channel fees reflect updated values",
                        Duration::from_secs(15),
                        || async {
                            let updated = gw
                                .client()
                                .list_channels()
                                .await
                                .map_err(ControlFlow::Continue)?
                                .into_iter()
                                .find(|c| c.funding_outpoint == Some(funding_outpoint))
                                .ok_or_else(|| {
                                    ControlFlow::Break(anyhow::anyhow!(
                                        "channel disappeared after fee update"
                                    ))
                                })?;
                            if updated.base_fee_msat == Some(new_base_fee_msat)
                                && updated.parts_per_million == Some(new_parts_per_million)
                            {
                                Ok(())
                            } else {
                                Err(ControlFlow::Continue(anyhow::anyhow!(
                                    "{} gateway still reports base={:?}, ppm={:?}",
                                    gw.ln.ln_type(),
                                    updated.base_fee_msat,
                                    updated.parts_per_million,
                                )))
                            }
                        },
                    )
                    .await?;
                }
            } else {
                info!(
                    target: LOG_TEST,
                    gateway_cli_version = %gateway_cli_version,
                    "Skipping set-channel-fees test (requires gateway >= 0.12.0-alpha)"
                );
            }

            info!(target: LOG_TEST, "Pegging-out gateways...");
            federation
                .pegout_gateways(500_000_000, gateways.clone())
                .await?;

            info!(target: LOG_TEST, "Testing only admin can send onchain...");
            let send_result = gw_lnd.client().with_password("secondbest").send_onchain(dev_fed.bitcoind().await?, BitcoinAmountOrAll::All, 10).await;
            assert!(send_result.is_err(), "Only admins can send onchain");

            info!(target: LOG_TEST, "Testing sending onchain...");
            let bitcoind = dev_fed.bitcoind().await?;
            for gw in &gateways {
                let txid = gw
                    .client()
                    .send_onchain(dev_fed.bitcoind().await?, BitcoinAmountOrAll::All, 10)
                    .await?;
                bitcoind.poll_get_transaction(txid).await?;
            }

            info!(target: LOG_TEST, "Testing closing all channels...");

            // Gracefully close one of LND's channel's
            let gw_ldk_pubkey = gw_ldk.client().lightning_pubkey().await?;
            gw_lnd.client().close_channel(gw_ldk_pubkey, false).await?;

            // Force close remaining channels on every gateway and wait
            for gw in &gateways {
                gw.client()
                    .close_all_channels(true, Duration::from_secs(30))
                    .await?;
            }

            Ok(())
        })
        .await
}

async fn esplora_test() -> anyhow::Result<()> {
    let args = cli::CommonArgs::parse_from::<_, ffi::OsString>(vec![]);
    let (process_mgr, task_group) = cli::setup(args).await?;
    cleanup_on_exit(
        async {
            info!("Spawning bitcoind...");
            let bitcoind = Bitcoind::new(&process_mgr, false).await?;
            info!("Spawning esplora...");
            let _esplora = Esplora::new(&process_mgr, bitcoind).await?;
            let network = bitcoin::Network::from_str(&process_mgr.globals.FM_GATEWAY_NETWORK)
                .expect("Could not parse network");
            let esplora_port = process_mgr.globals.FM_PORT_ESPLORA.to_string();
            let esplora = default_esplora_server(network, Some(esplora_port));
            unsafe {
                std::env::remove_var("FM_BITCOIND_URL");
                std::env::set_var("FM_ESPLORA_URL", esplora.url.to_string());
            }
            info!("Spawning ldk gateway...");
            let ldk = Gatewayd::new(
                &process_mgr,
                LightningNode::Ldk {
                    name: "gateway-ldk-esplora".to_string(),
                    gw_port: process_mgr.globals.FM_PORT_GW_LDK,
                    ldk_port: process_mgr.globals.FM_PORT_LDK,
                    metrics_port: process_mgr.globals.FM_PORT_GW_LDK_METRICS,
                },
                0,
            )
            .await?;

            info!("Waiting for ldk gatewy to be ready...");
            poll("Waiting for LDK to be ready", || async {
                let info = ldk
                    .client()
                    .get_info()
                    .await
                    .map_err(ControlFlow::Continue)?;
                let state: String = serde_json::from_value(info["gateway_state"].clone())
                    .expect("Could not get gateway state");
                if state == "Running" {
                    Ok(())
                } else {
                    Err(ControlFlow::Continue(anyhow::anyhow!(
                        "Gateway not running"
                    )))
                }
            })
            .await?;

            ldk.client().get_ln_onchain_address().await?;
            info!(target:LOG_TEST, "ldk gateway successfully spawned and connected to esplora");
            Ok(())
        },
        task_group,
    )
    .await?;
    Ok(())
}

async fn get_transaction(
    gateway: &Gatewayd,
    kind: PaymentKind,
    amount: Amount,
    status: PaymentStatus,
) -> Option<PaymentDetails> {
    let transactions = gateway
        .client()
        .list_transactions(
            now() - Duration::from_mins(5),
            now() + Duration::from_mins(5),
        )
        .await
        .ok()?;
    transactions.into_iter().find(|details| {
        details.payment_kind == kind && details.amount == amount && details.status == status
    })
}