newton_prover_transporter/

lib.rs

//! transporter crate for cross-chain operator table updates

/// operator table calculation
pub mod calculator;
/// transporter configuration
pub mod config;
/// source chain event listener
pub mod listener;
/// operator table updater
pub mod updater;

use alloy::{primitives::FixedBytes, providers::Provider};
use calculator::WeightCalculator;
use config::TransporterConfig;
use eigensdk::common::get_provider;
use eyre::Result;
use listener::{SourceChainListener, TransportTrigger};
use newton_prover_core::config::NewtonAvsConfig;
use std::time::Duration;
use tokio::{sync::mpsc, time};
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
use updater::TableUpdater;

pub use calculator::{OperatorSetQueryResult, WeightCalculator as TransporterWeightCalculator};
pub use updater::TableUpdater as TransporterTableUpdater;

struct WeightTransportState<P> {
    trigger_rx: mpsc::Receiver<TransportTrigger>,
    calculator: WeightCalculator,
    updater: TableUpdater,
    table_calculator: alloy::primitives::Address,
    operator_set_id: u32,
    source_provider: P,
}

/// transport operator table updates to the destination chain
pub async fn transport_update(
    calculator: &WeightCalculator,
    updater: &TableUpdater,
    table_calculator: alloy::primitives::Address,
    operator_set_id: u32,
    provider: &impl Provider,
) -> Result<()> {
    let operator_set_query = calculator.query_operator_set(operator_set_id, table_calculator).await?;

    let tables = calculator.calculate_operator_tables(&[operator_set_query]);

    if tables.is_empty() {
        info!("no operator tables to update");
        return Ok(());
    }

    let latest_block = provider.get_block_number().await?;
    // use latest - 1 to avoid GlobalTableRootInFuture when destination chain's clock
    // lags the source chain's latest block timestamp
    let reference_block = latest_block.saturating_sub(1);
    let block = provider
        .get_block_by_number(reference_block.into())
        .await?
        .ok_or_else(|| eyre::eyre!("failed to get latest block"))?;

    let reference_timestamp: u32 = block
        .header
        .timestamp
        .try_into()
        .map_err(|_| eyre::eyre!("block timestamp {} exceeds u32 max", block.header.timestamp))?;
    let reference_block_number: u32 = reference_block
        .try_into()
        .map_err(|_| eyre::eyre!("block number {} exceeds u32 max", reference_block))?;

    let mut leaves: Vec<FixedBytes<32>> = Vec::with_capacity(tables.len());
    for table in &tables {
        let table_bytes = calculator.encode_operator_table(table);
        let leaf = updater.calculate_operator_table_leaf(table_bytes).await?;
        leaves.push(leaf);
    }

    let (root, proofs) = calculator.build_merkle_tree(&leaves);

    updater
        .confirm_global_root(root, reference_timestamp, reference_block_number)
        .await?;

    for (i, (table, proof)) in tables.iter().zip(proofs.iter()).enumerate() {
        let table_bytes = calculator.encode_operator_table(table);
        updater
            .update_operator_table(reference_timestamp, root, i as u32, proof.clone(), table_bytes)
            .await?;
    }

    Ok(())
}

/// Shared setup for transporter operations: builds calculator, updater, and loads source chain state.
fn setup_transport(
    config: &NewtonAvsConfig<TransporterConfig>,
) -> Result<(WeightCalculator, TableUpdater, alloy::primitives::Address, String)> {
    let source_chain_id = config.get_source_chain_id();
    let source_rpc = config
        .rpc
        .get(source_chain_id)
        .ok_or_else(|| eyre::eyre!("no RPC config for source chain {}", source_chain_id))?;
    let source_rpc_url = source_rpc.http.clone();

    let source_multichain = if source_chain_id != config.chain_id {
        use newton_prover_core::config::contracts::SourceChainMultichainContracts;
        SourceChainMultichainContracts::load(source_chain_id, &config.env)?
    } else {
        return Err(eyre::eyre!(
            "source and destination chain are the same - multichain contracts not applicable"
        ));
    };

    let calculator = WeightCalculator::new(
        source_rpc_url.clone(),
        config.contracts.avs.newton_prover_service_manager,
        source_multichain.cross_chain_registry,
    );

    let dest_rpc = config
        .rpc
        .get(config.chain_id)
        .ok_or_else(|| eyre::eyre!("no RPC config for chain {}", config.chain_id))?;

    let dest_multichain = config
        .contracts
        .destination_multichain()
        .map_err(|e| eyre::eyre!("destination_multichain not configured: {}", e))?;

    let updater = TableUpdater::new(
        config.service.signer.clone(),
        dest_multichain.operator_table_updater,
        dest_rpc.http.clone(),
    )?;

    Ok((calculator, updater, source_multichain.table_calculator, source_rpc_url))
}

/// Perform a single one-shot transport update for a given destination chain config.
///
/// This extracts the setup from `run()` and calls `transport_update()` exactly once,
/// then returns. Useful for ad-hoc syncs triggered by CLI commands.
pub async fn sync_once(config: NewtonAvsConfig<TransporterConfig>) -> Result<()> {
    let (calculator, updater, table_calculator, source_rpc_url) = setup_transport(&config)?;
    let source_provider = get_provider(&source_rpc_url);

    let source_chain_id = config.get_source_chain_id();
    let dest_chain_id = config.chain_id;

    info!(
        source_chain_id = source_chain_id,
        dest_chain_id = dest_chain_id,
        "performing one-shot transport sync"
    );

    let start = std::time::Instant::now();
    let result = transport_update(
        &calculator,
        &updater,
        table_calculator,
        config.service.operator_set_id,
        &source_provider,
    )
    .await;

    let duration = start.elapsed().as_secs_f64();
    newton_prover_metrics::record_transporter_sync_duration(source_chain_id, dest_chain_id, duration);
    match &result {
        Ok(()) => newton_prover_metrics::inc_transporter_sync_success(source_chain_id, dest_chain_id),
        Err(_) => newton_prover_metrics::inc_transporter_sync_failure(source_chain_id, dest_chain_id),
    }

    result
}

/// run the transporter service
pub async fn run(config: NewtonAvsConfig<TransporterConfig>, cancellation_token: CancellationToken) -> Result<()> {
    let (calculator, updater, table_calculator, source_rpc_url) = setup_transport(&config)?;

    let source_chain_id = config.get_source_chain_id();
    let dest_chain_id = config.chain_id;
    let source_rpc = config
        .rpc
        .get(source_chain_id)
        .ok_or_else(|| eyre::eyre!("no RPC config for source chain {}", source_chain_id))?;
    let source_ws_url = source_rpc.ws.clone();

    let listener = SourceChainListener::new(source_ws_url, config.contracts.eigenlayer.allocation_manager);
    let trigger_rx = listener.start(cancellation_token.clone()).await;

    let source_provider = get_provider(&source_rpc_url);

    let mut state = WeightTransportState {
        trigger_rx,
        calculator,
        updater,
        table_calculator,
        operator_set_id: config.service.operator_set_id,
        source_provider,
    };

    // Perform an explicit initial sync on startup to ensure destination chains
    // have up-to-date operator state immediately after deploy, rather than
    // waiting for the first periodic interval tick.
    info!(
        source_chain_id = source_chain_id,
        dest_chain_id = dest_chain_id,
        "performing initial startup sync"
    );
    let start = std::time::Instant::now();
    let result = transport_update(
        &state.calculator,
        &state.updater,
        state.table_calculator,
        state.operator_set_id,
        &state.source_provider,
    )
    .await;
    let duration = start.elapsed().as_secs_f64();
    newton_prover_metrics::record_transporter_sync_duration(source_chain_id, dest_chain_id, duration);
    match result {
        Ok(()) => {
            info!(
                source_chain_id = source_chain_id,
                dest_chain_id = dest_chain_id,
                duration_secs = duration,
                "initial startup sync completed"
            );
            newton_prover_metrics::inc_transporter_sync_success(source_chain_id, dest_chain_id);
        }
        Err(e) => {
            error!(
                source_chain_id = source_chain_id,
                dest_chain_id = dest_chain_id,
                error = %e,
                "initial startup sync failed, continuing with event loop"
            );
            newton_prover_metrics::inc_transporter_sync_failure(source_chain_id, dest_chain_id);
        }
    }

    let mut interval = time::interval(Duration::from_secs(config.service.update_frequency));
    // Skip the first immediate tick since we already performed the initial sync above
    interval.tick().await;

    loop {
        tokio::select! {
            _ = cancellation_token.cancelled() => {
                tracing::info!("received shutdown signal, stopping transporter");
                break Ok(());
            }

            _ = interval.tick() => {
                info!("updating stake weights according to tempo");
                let start = std::time::Instant::now();
                let result = transport_update(
                    &state.calculator,
                    &state.updater,
                    state.table_calculator,
                    state.operator_set_id,
                    &state.source_provider,
                ).await;
                let duration = start.elapsed().as_secs_f64();
                newton_prover_metrics::record_transporter_sync_duration(source_chain_id, dest_chain_id, duration);
                match result {
                    Ok(()) => newton_prover_metrics::inc_transporter_sync_success(source_chain_id, dest_chain_id),
                    Err(e) => {
                        newton_prover_metrics::inc_transporter_sync_failure(source_chain_id, dest_chain_id);
                        error!("failed to transport update: {}", e);
                    }
                }
            }

            Some(trigger) = state.trigger_rx.recv() => {
                let trigger_name = match &trigger {
                    TransportTrigger::OperatorSlashed { .. } => "operator_slashed",
                    TransportTrigger::OperatorAdded { .. } => "operator_added",
                    TransportTrigger::OperatorRemoved { .. } => "operator_removed",
                    TransportTrigger::Cadence => "cadence",
                };
                match trigger {
                    TransportTrigger::OperatorSlashed { .. } |
                    TransportTrigger::OperatorAdded { .. } |
                    TransportTrigger::OperatorRemoved { .. } => {
                        info!("forced update triggered by {}", trigger_name);
                        newton_prover_metrics::inc_transporter_event_triggered_sync(source_chain_id, dest_chain_id, trigger_name);
                        let start = std::time::Instant::now();
                        let result = transport_update(
                            &state.calculator,
                            &state.updater,
                            state.table_calculator,
                            state.operator_set_id,
                            &state.source_provider,
                        ).await;
                        let duration = start.elapsed().as_secs_f64();
                        newton_prover_metrics::record_transporter_sync_duration(source_chain_id, dest_chain_id, duration);
                        match result {
                            Ok(()) => newton_prover_metrics::inc_transporter_sync_success(source_chain_id, dest_chain_id),
                            Err(e) => {
                                newton_prover_metrics::inc_transporter_sync_failure(source_chain_id, dest_chain_id);
                                error!("failed to transport update: {}", e);
                            }
                        }
                        interval.reset();
                    }
                    TransportTrigger::Cadence => {}
                }
            }
        }
    }
}