zakura 1.0.4

//! Lightwalletd sync functions.

use std::{
    net::SocketAddr,
    sync::atomic::{AtomicBool, Ordering},
    time::{Duration, Instant},
};

use tempfile::TempDir;

use zakura_node_services::rpc_client::RpcRequestClient;
use zakura_test::prelude::*;

use crate::common::{launch::ZakuradTestDirExt, test_type::TestType};

/// The amount of time we wait between each tip check.
pub const TIP_CHECK_RATE_LIMIT: Duration = Duration::from_secs(60);

/// Wait for lightwalletd to sync to Zebra's tip.
///
/// If `wait_for_zakurad_mempool` is `true`, wait for Zebra to activate its mempool.
/// If `wait_for_zakurad_tip` is `true`, also wait for Zebra to sync to the network chain tip.
#[tracing::instrument]
pub fn wait_for_zakurad_and_lightwalletd_sync<
    P: ZakuradTestDirExt + std::fmt::Debug + std::marker::Send + 'static,
>(
    mut lightwalletd: TestChild<TempDir>,
    lightwalletd_rpc_port: u16,
    mut zakurad: TestChild<P>,
    zakura_rpc_address: SocketAddr,
    test_type: TestType,
    wait_for_zakurad_mempool: bool,
    wait_for_zakurad_tip: bool,
) -> Result<(TestChild<TempDir>, TestChild<P>)> {
    let is_zakurad_finished = AtomicBool::new(!wait_for_zakurad_tip);
    let is_lightwalletd_finished = AtomicBool::new(false);

    let is_zakurad_finished = &is_zakurad_finished;
    let is_lightwalletd_finished = &is_lightwalletd_finished;

    // TODO: split these closures out into their own functions

    // Check Zebra's logs for errors.
    // Optionally waits until Zebra has synced to the tip, based on `wait_for_zakurad_tip`.
    //
    // `lightwalletd` syncs can take a long time,
    // so we need to check that `zakurad` has synced to the tip in parallel.
    let zakurad_mut = &mut zakurad;
    let zakurad_wait_fn = || -> Result<_> {
        // When we are near the tip, the mempool should activate at least once
        if wait_for_zakurad_mempool {
            tracing::info!(
                ?test_type,
                "waiting for zakurad to activate the mempool when it gets near the tip..."
            );
            zakurad_mut.expect_stdout_line_matches("activating mempool")?;
        }

        // When we are near the tip, this message is logged multiple times
        if wait_for_zakurad_tip {
            tracing::info!(?test_type, "waiting for zakurad to sync to the tip...");
            zakurad_mut.expect_stdout_line_matches(crate::common::sync::SYNC_FINISHED_REGEX)?;
        }

        // Tell the other thread that Zebra has finished
        is_zakurad_finished.store(true, Ordering::SeqCst);

        tracing::info!(
            ?test_type,
            "zakurad is waiting for lightwalletd to sync to the tip..."
        );
        while !is_lightwalletd_finished.load(Ordering::SeqCst) {
            // Just keep checking the Zebra logs for errors...
            if wait_for_zakurad_tip {
                // Make sure the sync is still finished, this is logged every minute or so.
                zakurad_mut.expect_stdout_line_matches(crate::common::sync::SYNC_FINISHED_REGEX)?;
            } else {
                // Use sync progress logs, which are logged every minute or so.
                zakurad_mut.expect_stdout_line_matches(crate::common::sync::SYNC_PROGRESS_REGEX)?;
            }
        }

        Ok(zakurad_mut)
    };

    // Wait until `lightwalletd` has synced to Zebra's tip.
    // Calls `lightwalletd`'s gRPCs and Zebra's JSON-RPCs.
    // Also checks `lightwalletd`'s logs for errors.
    //
    // `zakurad` syncs can take a long time,
    // so we need to check that `lightwalletd` has synced to the tip in parallel.
    let lightwalletd_mut = &mut lightwalletd;
    let lightwalletd_wait_fn = || -> Result<_> {
        tracing::info!(
            ?test_type,
            "lightwalletd is waiting for zakurad to sync to the tip..."
        );
        while !is_zakurad_finished.load(Ordering::SeqCst) {
            // Just keep checking the `lightwalletd` logs for errors.
            // It usually logs something every 30-90 seconds,
            // but there's no specific message we need to wait for here.
            assert!(
                lightwalletd_mut.wait_for_stdout_line(None),
                "lightwalletd output unexpectedly finished early",
            );
        }

        tracing::info!(?test_type, "waiting for lightwalletd to sync to the tip...");
        while !are_zakurad_and_lightwalletd_tips_synced(zakura_rpc_address, lightwalletd_mut)? {
            let previous_check = Instant::now();

            // To improve performance, only check the tips occasionally
            while previous_check.elapsed() < TIP_CHECK_RATE_LIMIT {
                assert!(
                    lightwalletd_mut.wait_for_stdout_line(None),
                    "lightwalletd output unexpectedly finished early",
                );
            }
        }

        // Tell the other thread that `lightwalletd` has finished
        is_lightwalletd_finished.store(true, Ordering::SeqCst);

        Ok(lightwalletd_mut)
    };

    // Run both threads in parallel, automatically propagating any panics to this thread.
    std::thread::scope(|s| {
        // Launch the sync-waiting threads
        let zakurad_thread = s.spawn(|| {
            let zakurad_result = zakurad_wait_fn();
            is_zakurad_finished.store(true, Ordering::SeqCst);

            zakurad_result.expect("test failed while waiting for zakurad to sync");
        });

        let lightwalletd_thread = s.spawn(|| {
            let lightwalletd_result = lightwalletd_wait_fn();
            is_lightwalletd_finished.store(true, Ordering::SeqCst);

            lightwalletd_result.expect("test failed while waiting for lightwalletd to sync.");
        });

        // Mark the sync-waiting threads as finished if they fail or panic.
        // This tells the other thread that it can exit.
        //
        // TODO: use `panic::catch_unwind()` instead,
        //       when `&mut zakura_test::command::TestChild<TempDir>` is unwind-safe
        s.spawn(|| {
            let zakurad_result = zakurad_thread.join();
            is_zakurad_finished.store(true, Ordering::SeqCst);

            zakurad_result.expect("test panicked or failed while waiting for zakurad to sync");
        });
        s.spawn(|| {
            let lightwalletd_result = lightwalletd_thread.join();
            is_lightwalletd_finished.store(true, Ordering::SeqCst);

            lightwalletd_result
                .expect("test panicked or failed while waiting for lightwalletd to sync");
        });
    });

    Ok((lightwalletd, zakurad))
}

/// Returns `Ok(true)` if zakurad and lightwalletd are both at the same height.
#[tracing::instrument]
pub fn are_zakurad_and_lightwalletd_tips_synced(
    zakura_rpc_address: SocketAddr,
    lightwalletd: &mut TestChild<TempDir>,
) -> Result<bool> {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?;

    rt.block_on(async {
        // We are going to try getting the current lightwalletd height by reading the lightwalletd logs.
        let mut lightwalletd_next_height = 1;

        // Only go forward on getting next height from lightwalletd logs if we find the line we are interested in.
        //
        // TODO: move this blocking code out of the async executor.
        // The executor could block all tasks and futures while this code is running.
        // That's ok for now, but it might cause test hangs or failures if we spawn tasks, select(), or join().
        if let Ok(line) = lightwalletd.expect_stdout_line_matches("Waiting for block: [0-9]+") {
            let line_json: serde_json::Value = serde_json::from_str(line.as_str())
                .expect("captured lightwalletd logs are always valid json");
            let msg = line_json["msg"]
                .as_str()
                .expect("`msg` field is always a valid string");

            // Block number is the last word of the message. We rely on that specific for this to work.
            let last = msg
                .split(' ')
                .next_back()
                .expect("always possible to get the last word of a separated by space string");
            lightwalletd_next_height = last
                .parse()
                .expect("the last word is always the block number so it can be parsed to i32 ");
        }

        // The last height in lightwalletd is the one the program is expecting minus one.
        let lightwalletd_tip_height = (lightwalletd_next_height - 1) as u64;

        // Get the block tip from zakurad
        let client = RpcRequestClient::new(zakura_rpc_address);
        let zakurad_blockchain_info = client
            .text_from_call("getblockchaininfo", "[]".to_string())
            .await?;
        let zakurad_blockchain_info: serde_json::Value =
            serde_json::from_str(&zakurad_blockchain_info)?;
        let zakurad_tip_height = zakurad_blockchain_info["result"]["blocks"]
            .as_u64()
            .expect("unexpected block height: doesn't fit in u64");

        if lightwalletd_tip_height != zakurad_tip_height {
            tracing::info!(
                lightwalletd_tip_height,
                zakurad_tip_height,
                zakura_rpc_address = ?zakura_rpc_address,
                "lightwalletd tip is behind Zakura tip, waiting for sync",
            );
        } else {
            tracing::debug!(
                lightwalletd_tip_height,
                zakurad_tip_height,
                zakura_rpc_address = ?zakura_rpc_address,
                "lightwalletd tip matches Zakura tip",
            );
        }

        Ok(lightwalletd_tip_height == zakurad_tip_height)
    })
}