miden-ntx-builder 0.17.2

// Required by code generated by the upstream `#[instrument]` macro.
extern crate miden_node_tracing as tracing;

use std::num::{NonZeroU16, NonZeroUsize};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use anyhow::Context;
use builder::BlockStream;
use chain_state::ChainState;
use clients::{RemoteTransactionProver, RpcClient};
use miden_node_store::genesis::GenesisBlock;
use miden_node_tracing::{ErrorReport, debug, warn};
use miden_node_utils::lru_cache::LruCache;
use miden_node_utils::shutdown::CancellationToken;
use miden_protocol::account::AccountId;
use tonic::metadata::AsciiMetadataValue;
use url::Url;

use crate::db::NtxDbReader;
use crate::network_transaction::{
    GrpcClients,
    NetworkTransactionConfig,
    NetworkTransactionContext,
};
use crate::scheduler::{Scheduler, SchedulerConfig};

pub(crate) type NoteError = Arc<dyn ErrorReport + Send + Sync>;

mod allowlist;
mod builder;
mod candidate;
mod chain_state;
mod clients;
mod committed_block;
pub(crate) mod db;
mod execute;
mod network_transaction;
mod scheduler;
mod selection;
pub mod server;
mod sponsorship;

#[cfg(test)]
pub(crate) mod test_utils;

pub use builder::NetworkTransactionBuilder;

// BOOTSTRAP
// =================================================================================================

/// Bootstraps the ntx-builder database at `database_filepath` with the genesis block.
///
/// After this completes the singleton chain-state row exists at the genesis block number, so
/// [`NtxBuilderConfig`] startup can always resume from a persisted chain state instead of consuming
/// the genesis block from the subscription.
///
/// Returns an error if the database has already been bootstrapped.
pub async fn bootstrap(database_filepath: PathBuf, genesis: &GenesisBlock) -> anyhow::Result<()> {
    db::bootstrap(database_filepath, genesis.inner()).await
}

/// Applies pending migrations to the ntx-builder database at `database_filepath`.
pub fn migrate(database_filepath: impl AsRef<Path>) -> anyhow::Result<()> {
    db::migrate(database_filepath).context("failed to apply ntx-builder database migrations")
}

#[cfg(test)]
mod bootstrap_tests {
    use miden_node_store::genesis::{GenesisBlock, GenesisState};
    use miden_node_utils::fee::{test_fee_params, test_protocol_config};
    use miden_protocol::block::{BlockSignatures, SignedBlock};
    use miden_protocol::crypto::dsa::ecdsa_k256_keccak::SigningKey;

    #[tokio::test]
    async fn bootstrap_accepts_genesis_artifact_with_protocol_config() {
        use miden_node_utils::genesis::read_genesis_block;

        let genesis = GenesisState::new(
            Vec::new(),
            test_fee_params(),
            0,
            crate::test_utils::mock_genesis_block().header().validator_config().clone(),
            test_protocol_config(),
        )
        .into_block()
        .unwrap();
        let root = tempfile::tempdir().unwrap();
        let path = root.path().join("genesis.dat");
        fs_err::write(&path, miden_node_persistence::encode(&genesis)).unwrap();
        let decoded = read_genesis_block(&path).unwrap();
        assert_eq!(decoded.protocol_config(), genesis.protocol_config());
        let database_path = root.path().join("ntx.sqlite3");
        super::bootstrap(database_path.clone(), &decoded).await.unwrap();
        assert!(database_path.is_file());
    }

    #[test]
    fn genesis_block_rejects_signed_block() {
        let (header, body, _) = crate::test_utils::mock_genesis_block().into_parts();
        let signature = SigningKey::new().sign(header.commitment());
        let signatures = BlockSignatures::new(vec![signature]).unwrap();
        let block = SignedBlock::new_unchecked(header, body, signatures);

        let err = GenesisBlock::new(block, test_protocol_config())
            .expect_err("signed genesis block should fail");

        assert!(err.to_string().contains("must not carry signatures"), "unexpected error: {err}");
    }
}

// CONSTANTS
// =================================================================================================

const COMPONENT: &str = "miden-ntx-builder";

// Tracing target used for user-visible events.
pub const LOG_TARGET: &str = "user::miden-ntx-builder";

/// Default maximum number of network notes a network transaction is allowed to consume.
const DEFAULT_MAX_NOTES_PER_TX: NonZeroUsize = NonZeroUsize::new(20).expect("literal is non-zero");
const _: () = assert!(DEFAULT_MAX_NOTES_PER_TX.get() <= miden_tx::MAX_NUM_CHECKER_NOTES);

/// Default maximum number of network transactions which are computed concurrently.
///
/// This bounds the network transactions built locally. It does not include submitted
/// transactions which are waiting to be committed: those hold no local compute.
const DEFAULT_MAX_CONCURRENT_TXS: usize = 4;

/// Default maximum number of blocks to keep in the chain MMR.
const DEFAULT_MAX_BLOCK_COUNT: usize = 4;

/// Default maximum number of attempts to execute a failing note before dropping it.
const DEFAULT_MAX_NOTE_ATTEMPTS: usize = 30;

/// Default script cache size.
const DEFAULT_SCRIPT_CACHE_SIZE: NonZeroUsize =
    NonZeroUsize::new(1_000).expect("literal is non-zero");

/// Default per-request timeout for node RPC requests.
const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(10);

/// Default per-request timeout for the remote transaction prover.
const DEFAULT_TX_PROVER_TIMEOUT: Duration = Duration::from_secs(10);

/// Default timeout for gRPC requests served by the network transaction builder.
const DEFAULT_GRPC_TIMEOUT: Duration = Duration::from_secs(10);

/// Default initial sleep applied between per-request retries on transient infrastructure failures
/// (downed prover, transport error, RPC crash, RPC gRPC hiccup). Doubles on each retry up to
/// [`DEFAULT_REQUEST_BACKOFF_MAX`].
const DEFAULT_REQUEST_BACKOFF_INITIAL: Duration = Duration::from_millis(100);

/// Default upper bound on the per-request retry backoff sleep.
const DEFAULT_REQUEST_BACKOFF_MAX: Duration = Duration::from_secs(30);

/// Default maximum number of VM execution cycles allowed for a network transaction.
///
/// This limits the computational cost of network transactions. The protocol maximum is
/// `1 << 29` but network transactions should be much cheaper.
const DEFAULT_MAX_TX_CYCLES: u32 = 1 << 19;

/// Default number of blocks after which a submitted network transaction expires.
///
/// Used both as the on-chain transaction expiration delta and as the local timeout after which the
/// scheduler releases the account for a new build. Must be within the kernel's `1..=u16::MAX`
/// range.
const DEFAULT_TX_EXPIRATION_DELTA: NonZeroU16 = NonZeroU16::new(30).unwrap();

// CONFIGURATION
// =================================================================================================

/// Configuration for the Network Transaction Builder.
///
/// This struct contains all the settings needed to create and run a `NetworkTransactionBuilder`.
#[derive(Debug, Clone)]
pub struct NtxBuilderConfig {
    /// Address of the node RPC gRPC server.
    pub rpc_url: Url,

    /// Timeout for requests to the node RPC service.
    pub rpc_timeout: Duration,

    /// Optional auth header value injected into internal RPC requests.
    pub rpc_auth_header: Option<AsciiMetadataValue>,

    /// Address of the remote transaction prover.
    pub tx_prover_url: Url,

    /// Timeout for requests to the remote transaction prover.
    pub tx_prover_timeout: Duration,

    /// Timeout for gRPC requests served by the network transaction builder.
    pub grpc_timeout: Duration,

    /// Size of the LRU cache for note scripts. Scripts are fetched through RPC and cached to avoid
    /// repeated gRPC calls.
    pub script_cache_size: NonZeroUsize,

    /// Maximum number of network transactions computed concurrently. Submitted transactions waiting
    /// to be committed do not count against this limit.
    pub max_concurrent_txs: usize,

    /// Network accounts served before every other account, such as the native faucet.
    ///
    /// Each one still holds at most one attempt slot. Keep the list shorter than
    /// [`Self::max_concurrent_txs`] so a slot always remains for the other accounts.
    pub priority_accounts: Vec<AccountId>,

    /// Maximum number of network notes a single transaction is allowed to consume. Sponsorship
    /// notes count against this budget.
    pub max_notes_per_tx: NonZeroUsize,

    /// Maximum number of attempts to execute a failing note before dropping it. Notes use
    /// exponential backoff between attempts.
    pub max_note_attempts: usize,

    /// Maximum number of blocks to keep in the chain MMR. Older blocks are pruned.
    pub max_block_count: usize,

    /// Maximum number of VM execution cycles allowed for a single network transaction.
    ///
    /// Network transactions that exceed this limit will fail with an execution error.
    /// Defaults to 2^18 cycles.
    pub max_cycles: u32,

    /// Number of blocks after which a submitted network transaction expires. Set as the on-chain
    /// transaction expiration delta and reused as the local timeout after which the scheduler
    /// releases the account for a new build. Must be within `1..=u16::MAX` (enforced by the
    /// transaction kernel).
    pub tx_expiration_delta: NonZeroU16,

    /// Initial sleep applied between per-request retries on transient infrastructure failures (e.g.
    /// prover unreachable, RPC crash, transport error, RPC gRPC hiccup). Doubles on each retry up
    /// to [`Self::request_backoff_max`]. Per-note `attempt_count` is *not* advanced while retries
    /// are in progress.
    pub request_backoff_initial: Duration,

    /// Upper bound on the per-request retry backoff sleep.
    pub request_backoff_max: Duration,

    /// Path to the SQLite database file used for persistent state.
    pub database_filepath: PathBuf,

    /// Maximum number of SQLite connections in the database connection pool.
    pub sqlite_connection_pool_size: NonZeroUsize,
}

impl NtxBuilderConfig {
    pub fn new(rpc_url: Url, tx_prover_url: Url, database_filepath: PathBuf) -> Self {
        Self {
            rpc_url,
            rpc_timeout: DEFAULT_RPC_TIMEOUT,
            rpc_auth_header: None,
            tx_prover_url,
            tx_prover_timeout: DEFAULT_TX_PROVER_TIMEOUT,
            grpc_timeout: DEFAULT_GRPC_TIMEOUT,
            script_cache_size: DEFAULT_SCRIPT_CACHE_SIZE,
            max_concurrent_txs: DEFAULT_MAX_CONCURRENT_TXS,
            priority_accounts: Vec::new(),
            max_notes_per_tx: DEFAULT_MAX_NOTES_PER_TX,
            max_note_attempts: DEFAULT_MAX_NOTE_ATTEMPTS,
            max_block_count: DEFAULT_MAX_BLOCK_COUNT,
            max_cycles: DEFAULT_MAX_TX_CYCLES,
            tx_expiration_delta: DEFAULT_TX_EXPIRATION_DELTA,
            request_backoff_initial: DEFAULT_REQUEST_BACKOFF_INITIAL,
            request_backoff_max: DEFAULT_REQUEST_BACKOFF_MAX,
            database_filepath,
            sqlite_connection_pool_size: miden_node_db::default_connection_pool_size(),
        }
    }

    /// Sets the optional auth header value to inject into internal RPC requests.
    #[must_use]
    pub fn with_rpc_auth_header(mut self, value: AsciiMetadataValue) -> Self {
        self.rpc_auth_header = Some(value);
        self
    }

    /// Sets the timeout for requests to the node RPC service.
    #[must_use]
    pub fn with_rpc_timeout(mut self, timeout: Duration) -> Self {
        self.rpc_timeout = timeout;
        self
    }

    /// Sets the timeout for requests to the remote transaction prover.
    #[must_use]
    pub fn with_tx_prover_timeout(mut self, timeout: Duration) -> Self {
        self.tx_prover_timeout = timeout;
        self
    }

    /// Sets the timeout for gRPC requests served by the network transaction builder.
    #[must_use]
    pub fn with_grpc_timeout(mut self, timeout: Duration) -> Self {
        self.grpc_timeout = timeout;
        self
    }

    /// Sets the script cache size.
    #[must_use]
    pub fn with_script_cache_size(mut self, size: NonZeroUsize) -> Self {
        self.script_cache_size = size;
        self
    }

    /// Sets the maximum number of concurrent transactions.
    #[must_use]
    pub fn with_max_concurrent_txs(mut self, max: usize) -> Self {
        self.max_concurrent_txs = max;
        self
    }

    /// Sets the network accounts served before every other account.
    #[must_use]
    pub fn with_priority_accounts(mut self, accounts: Vec<AccountId>) -> Self {
        self.priority_accounts = accounts;
        self
    }

    /// Sets the maximum number of notes per transaction.
    ///
    /// # Panics
    ///
    /// Panics if `max` exceeds `miden_tx::MAX_NUM_CHECKER_NOTES`.
    #[must_use]
    pub fn with_max_notes_per_tx(mut self, max: NonZeroUsize) -> Self {
        assert!(
            max.get() <= miden_tx::MAX_NUM_CHECKER_NOTES,
            "max_notes_per_tx ({}) exceeds MAX_NUM_CHECKER_NOTES ({})",
            max,
            miden_tx::MAX_NUM_CHECKER_NOTES
        );
        self.max_notes_per_tx = max;
        self
    }

    /// Sets the maximum number of note execution attempts.
    #[must_use]
    pub fn with_max_note_attempts(mut self, max: usize) -> Self {
        self.max_note_attempts = max;
        self
    }

    /// Sets the maximum number of blocks to keep in the chain MMR.
    #[must_use]
    pub fn with_max_block_count(mut self, max: usize) -> Self {
        self.max_block_count = max;
        self
    }

    /// Sets the maximum number of VM execution cycles for network transactions.
    #[must_use]
    pub fn with_max_cycles(mut self, max: u32) -> Self {
        self.max_cycles = max;
        self
    }

    /// Sets the transaction expiration delta (in blocks). Also bounds how long the scheduler keeps
    /// an account blocked on a submission that never lands.
    #[must_use]
    pub fn with_tx_expiration_delta(mut self, delta: NonZeroU16) -> Self {
        self.tx_expiration_delta = delta;
        self
    }

    /// Sets the per-request retry backoff bounds (initial sleep and cap) used when retrying
    /// transient infrastructure failures inside a single network transaction build.
    #[must_use]
    pub fn with_request_backoff(mut self, initial: Duration, max: Duration) -> Self {
        self.request_backoff_initial = initial;
        self.request_backoff_max = max;
        self
    }

    /// Sets the SQLite connection pool size.
    #[must_use]
    pub fn with_sqlite_connection_pool_size(mut self, size: NonZeroUsize) -> Self {
        self.sqlite_connection_pool_size = size;
        self
    }

    /// Builds and initializes the network transaction builder.
    ///
    /// Loads the in-memory chain state from the persisted header + chain MMR and opens a
    /// committed-block subscription against the node RPC service starting at `persisted_tip + 1`.
    /// The database must have been bootstrapped with the genesis block beforehand (see
    /// [`crate::bootstrap`]).
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The DB cannot be opened or the schema verification fails
    /// - The DB has not been bootstrapped (no persisted chain state)
    /// - The RPC connection fails (after retries)
    /// - The remote header or protocol config does not match the persisted chain tip
    pub async fn build(
        self,
        shutdown: CancellationToken,
    ) -> anyhow::Result<NetworkTransactionBuilder> {
        // Set up the database connection pool. Writes are serialized by the framework's single
        // dedicated writer connection, so block application never contends with the transaction
        // builds (which only read) for the shared reader pool.
        let db = db::load_with_pool_size(
            self.database_filepath.clone(),
            self.sqlite_connection_pool_size,
        )
        .await?;

        // Get the genesis commitment to send in the accept header
        let genesis_commitment = db
            .select_genesis_commitment()
            .await
            .context("failed to read genesis commitment")?
            .context(
                "ntx-builder database has not been bootstrapped; \
                 run `miden-ntx-builder bootstrap` first",
            )?;
        let genesis_validator_keys = db
            .select_genesis_validator_keys()
            .await
            .context("failed to read genesis validator keys")?
            .context("genesis validator keys are missing; re-bootstrap the NTX builder database")?;
        let trusted_validator_signing_keys = genesis_validator_keys.keys().to_vec();

        let rpc = match self.rpc_auth_header.clone() {
            Some(rpc_auth_header_value) => RpcClient::new_with_auth(
                self.rpc_url.clone(),
                Some(rpc_auth_header_value),
                genesis_commitment,
                trusted_validator_signing_keys,
                self.rpc_timeout,
                self.request_backoff_initial,
                self.request_backoff_max,
            ),
            None => RpcClient::new(
                self.rpc_url.clone(),
                genesis_commitment,
                trusted_validator_signing_keys,
                self.rpc_timeout,
                self.request_backoff_initial,
                self.request_backoff_max,
            ),
        }?;
        rpc.sealer()
            .await
            .context("failed to initialize the transaction inputs sealer")?;

        // The transaction prover is otherwise contacted only once an account has work, so monitor
        // its lazy endpoint from startup to make a bad configuration visible immediately.
        let tx_prover_monitor = miden_node_proto::clients::Builder::new(self.tx_prover_url.clone())
            .with_tls()?
            .with_timeout(self.tx_prover_timeout)
            .without_metadata_version()
            .without_metadata_genesis()
            .without_auth_header()
            .with_otel_context_injection();
        let monitor_shutdown = shutdown.clone();
        let _tx_prover_monitor = tokio::spawn(async move {
            tx_prover_monitor
                .monitor::<miden_node_proto::clients::RemoteProverClient>(
                    "transaction-prover",
                    monitor_shutdown,
                )
                .await;
        });

        // The database is bootstrapped with the genesis block before startup (see
        // `miden-ntx-builder bootstrap`), so a persisted chain state is always present. Load it and
        // resume the subscription from the block after the last applied one.
        let (last_applied_block, header, mmr) =
            db.select_chain_state().await.context("failed to read chain state")?.context(
                "ntx-builder database has not been bootstrapped; \
                 run `miden-ntx-builder bootstrap` first",
            )?;
        let protocol_config = rpc
            .protocol_config_for_header(&header)
            .await
            .context("failed to verify the persisted chain tip and protocol config")?;

        let block_from = last_applied_block.child();

        debug!(
            target: LOG_TARGET,
            "ntx-builder opening committed-block subscription",
            block.from = block_from
        );

        // The stream reconnects when the node closes the subscription. It resumes from the next
        // block that the builder has not applied.
        let block_stream: BlockStream = Box::pin(rpc.block_subscription_reconnecting(block_from));

        let chain = ChainState::new(header, mmr, protocol_config);

        let scheduler = self.build_scheduler(rpc, db.reader())?;

        Ok(NetworkTransactionBuilder::new(
            self,
            db,
            block_stream,
            last_applied_block,
            chain,
            scheduler,
        ))
    }

    /// Builds the [`Scheduler`] that owns the network transaction builds.
    ///
    /// The context it carries is cloned into every spawned build, so the gRPC clients and the note
    /// script cache are shared rather than rebuilt per transaction.
    fn build_scheduler(&self, rpc: RpcClient, db: NtxDbReader) -> anyhow::Result<Scheduler> {
        let ctx = NetworkTransactionContext {
            clients: GrpcClients {
                rpc,
                prover: RemoteTransactionProver::new(
                    self.tx_prover_url.clone(),
                    self.tx_prover_timeout,
                )?,
            },
            db,
            script_cache: LruCache::new(self.script_cache_size),
            tx_args: selection::build_tx_args(self.tx_expiration_delta),
            config: NetworkTransactionConfig {
                max_notes_per_tx: self.max_notes_per_tx,
                max_note_attempts: self.max_note_attempts,
                max_cycles: self.max_cycles,
                request_backoff_initial: self.request_backoff_initial,
                request_backoff_max: self.request_backoff_max,
            },
        };

        // With as many priority accounts as slots, the other accounts only run when a priority
        // account has nothing to do.
        if !self.priority_accounts.is_empty()
            && self.priority_accounts.len() >= self.max_concurrent_txs
        {
            warn!(
                target: LOG_TARGET,
                "priority accounts can occupy every build slot; raise --max-concurrent-txs",
                account.ids.count = self.priority_accounts.len(),
                ntx_builder.max_concurrent_txs = self.max_concurrent_txs
            );
        }

        let config = SchedulerConfig {
            max_concurrent_txs: self.max_concurrent_txs,
            tx_expiration_delta: self.tx_expiration_delta,
            priority_accounts: self.priority_accounts.clone(),
        };

        Ok(Scheduler::new(ctx, config))
    }
}