diff --git a/bin/ntx-builder/README.md b/bin/ntx-builder/README.md index 53db0f2bcc..4ed4ef60f1 100644 --- a/bin/ntx-builder/README.md +++ b/bin/ntx-builder/README.md @@ -6,9 +6,10 @@ layout. ## Role -The network transaction builder syncs blocks from an upstream node, tracking network notes and accounts. It starts -per-account workers for network accounts with pending work. Each worker selects viable notes, constructs a transaction, -proves it, and submits the proven transaction back through the upstream node's RPC API. +The network transaction builder syncs blocks from an upstream node, tracking network notes and accounts. On every +committed block it picks the network accounts that have pending notes and spawns one short-lived transaction attempt per +account, up to a configured concurrency limit. Each attempt selects viable notes, constructs a transaction, proves it, +and submits the proven transaction back through the upstream node's RPC API. The builder can use a remote transaction prover through `miden-remote-prover`, or fall back to in-process proving where appropriate for local development. It also exposes an internal gRPC API that the node RPC component can use for @@ -22,11 +23,10 @@ be exposed through the public RPC API. ## Benchmarks -`benches/large_account.rs` measures the cost of a large network account, which each actor keeps fully resident for its -lifetime and reloads from the database on start and after every expired submission. It synthesizes a network account -with a populated storage map and reports resident/peak heap, serialized size, and per-operation timings. Per-candidate -cost is not a factor: the account is shared via `Arc` and advanced with `Arc::make_mut`, and `PartialAccount::from` is -constant-time in the map size for existing accounts. +`benches/large_account.rs` measures the cost of a large network account, which every transaction attempt loads from the +database. It synthesizes a network account with a populated storage map and reports resident/peak heap, serialized size, +and per-operation timings. Per-candidate cost is not a factor: the account is shared via `Arc`, and +`PartialAccount::from` is constant-time in the map size for existing accounts. ```bash # Default sizes (1k, 10k, 100k entries). diff --git a/bin/ntx-builder/benches/large_account.rs b/bin/ntx-builder/benches/large_account.rs index 12028c685e..b530df37d7 100644 --- a/bin/ntx-builder/benches/large_account.rs +++ b/bin/ntx-builder/benches/large_account.rs @@ -1,9 +1,9 @@ //! Benchmark: building a network transaction against a very large account. //! -//! The `ntx-builder` keeps the native network `Account` fully resident in memory for the lifetime of -//! an actor. For an account with a large storage map (e.g. a million entries) that could become -//! untenable. This benchmark measures the costs so we can decide whether lazy/partial native-account -//! loading is worth implementing. +//! Every `ntx-builder` transaction attempt loads the native network `Account` from the database. For +//! an account with a large storage map (e.g. a million entries) that could become untenable. This +//! benchmark measures the costs so we can decide whether lazy/partial native-account loading is +//! worth implementing. use std::alloc::{GlobalAlloc, Layout, System}; use std::collections::BTreeSet; diff --git a/bin/ntx-builder/src/actor/mod.rs b/bin/ntx-builder/src/actor/mod.rs deleted file mode 100644 index d24d20aff0..0000000000 --- a/bin/ntx-builder/src/actor/mod.rs +++ /dev/null @@ -1,1016 +0,0 @@ -use std::num::{NonZeroU16, NonZeroUsize}; -use std::sync::Arc; -use std::time::Duration; - -use anyhow::Context; -use futures::FutureExt; -use miden_node_tracing::{debug, error, info, miden_instrument}; -use miden_node_utils::lru_cache::LruCache; -use miden_node_utils::shutdown::CancellationToken; -use miden_protocol::Word; -use miden_protocol::account::{Account, AccountId, AccountPatch}; -use miden_protocol::block::BlockNumber; -use miden_protocol::note::{Note, NoteScript, Nullifier}; -use miden_protocol::transaction::{TransactionArgs, TransactionId}; -use tokio::sync::{Semaphore, mpsc, watch}; - -use crate::candidate::TransactionCandidate; -use crate::chain_state::SharedChainState; -use crate::clients::{RemoteTransactionProver, RpcClient}; -use crate::coordinator::AccountView; -use crate::db::NtxDbReader; -use crate::selection::{ - Selection, - attribute_failed_notes, - log_deferred_notes, - log_oversized_notes, - select_candidate, -}; -use crate::{LOG_TARGET, NoteError, execute}; - -// ACTOR REQUESTS -// ================================================================================================ - -/// A request sent from an account actor to the coordinator via a shared mpsc channel. -pub enum ActorRequest { - /// One or more notes failed during transaction execution and should have their attempt counters - /// incremented. The actor waits for the coordinator to acknowledge the DB write via the oneshot - /// channel, preventing race conditions where the actor could re-select the same notes before - /// the failure is persisted. - NotesFailed { - failed_notes: Vec<(Nullifier, NoteError)>, - block_num: BlockNumber, - ack_tx: tokio::sync::oneshot::Sender<()>, - }, - /// One or more notes were proven to exceed the per-tx cycle budget on their own and can never - /// be consumed. They should be marked permanently unconsumable (discarded) so they are not - /// re-selected. The actor waits for the DB write to be acknowledged before re-selecting notes. - NotesDiscarded { - nullifiers: Vec, - block_num: BlockNumber, - ack_tx: tokio::sync::oneshot::Sender<()>, - }, - /// A note script was fetched from the remote RPC service and should be persisted to the local - /// DB. - CacheNoteScript { script_root: Word, script: NoteScript }, -} - -// ACTOR SUB-STRUCTS -// ================================================================================================ - -/// gRPC clients used by an account actor to interact with the node's services. -#[derive(Clone)] -pub struct GrpcClients { - /// Client for interacting with the RPC service in order to load account state. - pub rpc: RpcClient, - /// Client for remote transaction proving. - pub prover: RemoteTransactionProver, -} - -/// Shared state read (and written, in the case of `db`) by all account actors. -#[derive(Clone)] -pub struct State { - /// Local database for account state, notes, and transaction tracking. - pub db: NtxDbReader, - /// The latest chain state. A single chain state is shared among all actors. - pub chain: Arc, - /// Shared LRU cache for storing retrieved note scripts to avoid repeated RPC calls. - pub script_cache: LruCache, - /// [`TransactionArgs`] used by every network transaction. - /// - /// These are constant and are therefore prebuilt and cloned for each transaction. - /// - /// Currently this contains the transaction expiration script. - pub tx_args: TransactionArgs, -} - -/// Per-actor configuration knobs. -#[derive(Debug, Clone, Copy)] -pub struct ActorConfig { - /// Maximum number of notes per transaction. Sponsorship notes count against this budget. - pub max_notes_per_tx: NonZeroUsize, - /// Maximum number of note execution attempts before dropping a note. - pub max_note_attempts: usize, - /// Duration after which an idle actor will deactivate. - pub idle_timeout: Duration, - /// Maximum number of VM execution cycles for network transactions. - pub max_cycles: u32, - /// Number of blocks after which a submitted transaction expires. Set as the on-chain expiration - /// delta and reused as the `WaitForBlock` retry timeout. - pub tx_expiration_delta: NonZeroU16, - /// Initial sleep applied between per-request retries on transient infrastructure failures - /// (prover unreachable, RPC transport error, RPC gRPC hiccup). Doubles each retry up to - /// [`Self::request_backoff_max`]. - pub request_backoff_initial: Duration, - /// Upper bound on the per-request retry backoff sleep. - pub request_backoff_max: Duration, -} - -// ACCOUNT ACTOR CONTEXT -// ================================================================================================ - -/// Contains resources shared by all account actors. The coordinator uses this to spawn new actors. -#[derive(Clone)] -pub struct AccountActorContext { - pub clients: GrpcClients, - pub state: State, - pub config: ActorConfig, - /// Channel for sending requests to the coordinator (via the builder loop). - pub request_tx: mpsc::Sender, -} - -#[cfg(test)] -impl AccountActorContext { - /// Creates a minimal `AccountActorContext` suitable for unit tests. - /// - /// The URLs are fake and actors spawned with this context will fail on their first gRPC call, - /// but this is sufficient for testing coordinator logic (registry, deactivation, etc.). - pub fn test(db: &NtxDbReader) -> Self { - use miden_protocol::crypto::merkle::mmr::{Forest, MmrPeaks, PartialMmr}; - use url::Url; - - use crate::chain_state::SharedChainState; - use crate::clients::RpcClient; - use crate::test_utils::mock_block_header; - - let url = Url::parse("http://127.0.0.1:1").unwrap(); - let block_header = mock_block_header(0_u32.into()); - let trusted_validator_signing_keys = block_header.validator_config().keys().to_vec(); - let chain_mmr = PartialMmr::from_peaks( - MmrPeaks::new(Forest::new(0).expect("forest 0 is valid"), vec![]).unwrap(), - ); - let chain_state = Arc::new(SharedChainState::new(block_header, chain_mmr)); - let (request_tx, _request_rx) = mpsc::channel(1); - let tx_args = crate::selection::build_tx_args(NonZeroU16::new(30).unwrap()); - - Self { - clients: GrpcClients { - rpc: RpcClient::new( - url.clone(), - miden_protocol::Word::default(), - trusted_validator_signing_keys, - Duration::from_secs(10), - Duration::from_millis(100), - Duration::from_secs(30), - ) - .expect("rpc client should be constructed"), - prover: RemoteTransactionProver::new(url.clone(), Duration::from_secs(10)) - .expect("prover client should be constructed"), - }, - state: State { - db: db.clone(), - chain: chain_state, - script_cache: LruCache::new(NonZeroUsize::new(1).unwrap()), - tx_args, - }, - config: ActorConfig { - max_notes_per_tx: NonZeroUsize::new(1).unwrap(), - max_note_attempts: 1, - idle_timeout: Duration::from_mins(1), - max_cycles: 1 << 18, - tx_expiration_delta: NonZeroU16::new(30).unwrap(), - request_backoff_initial: Duration::from_millis(1), - request_backoff_max: Duration::from_millis(10), - }, - request_tx, - } - } -} - -// ACTOR MODE -// ================================================================================================ - -/// The mode of operation that the account actor is currently performing. -#[derive(Debug)] -enum ActorMode { - /// No notes targeting this account are currently available. The actor sleeps on the idle - /// timeout and awaits a coordinator notification to re-evaluate. - NoViableNotes, - /// Notes are available for consumption. The actor acquires a transaction permit and submits a - /// candidate. - NotesAvailable, - /// A network transaction has been submitted; the actor waits for it to land in a committed - /// block. Landing is detected from the pushed [`AccountView`]: the coordinator reports the - /// latest transaction id committed against each network account (mirroring - /// `accounts.last_tx_id`), so the actor checks whether its own submitted id is the account's - /// latest. On landing it applies `pending_patch` to its in-memory account, avoiding a re-read - /// of the full account from the database. - WaitForBlock { - /// Id of the network transaction the actor submitted. - submitted_tx_id: TransactionId, - /// Chain tip block number at submission. With [`ActorConfig::tx_expiration_delta`] this - /// bounds how long the actor waits before retrying. - submitted_at: BlockNumber, - /// The account patch the submitted transaction produced, applied to the in-memory account - /// once the transaction lands. - pending_patch: AccountPatch, - }, -} - -// ACCOUNT ACTOR -// ================================================================================================ - -/// A long-running asynchronous task that handles the complete lifecycle of network transaction -/// processing. Each actor operates independently and is managed by a single coordinator that -/// spawns, monitors, and messages all actors. -/// -/// ## Core Responsibilities -/// -/// - **State Management**: Tracks the account's committed state in memory, advancing it from the -/// [`AccountView`] the coordinator pushes after each block. -/// - **Transaction Selection**: Selects viable notes and constructs a [`TransactionCandidate`] -/// based on current chain state and a DB query for the account's available notes. -/// - **Transaction Execution**: Executes selected transactions using either local or remote -/// proving. -/// - **Chain Integration**: Reacts to per-account [`AccountView`] updates pushed by the coordinator -/// to stay synchronized with the network state. -/// -/// ## Lifecycle -/// -/// 1. **Initialization**: Loads the committed account state (guaranteed to exist, since the -/// coordinator only spawns actors for committed accounts), then checks DB for available notes. -/// 2. **Event Loop**: Re-evaluates state from the pushed [`AccountView`] and executes transactions. -/// 3. **Transaction Processing**: Selects, executes, proves, and submits transactions through RPC. -/// 4. **State Updates**: Committed-chain updates are persisted to DB and reflected in the view -/// before actors observe them. -/// 5. **Shutdown**: Terminates gracefully on idle timeout (only when it has no pending notes), or -/// returns an error on unrecoverable failures. -/// -/// ## Concurrency -/// -/// Each actor runs in its own async task and communicates with other system components through -/// shared state. The coordinator signals state changes by pushing an [`AccountView`] over a watch -/// channel; the actor exits of its own accord when idle for longer than -/// [`ActorConfig::idle_timeout`]. -pub struct AccountActor { - /// The network account this actor is responsible for. - account_id: AccountId, - /// gRPC clients used by the actor. - clients: GrpcClients, - /// Shared state accessed by the actor. - state: State, - /// Per-actor configuration knobs. - config: ActorConfig, - /// Channel for sending requests to the coordinator. - request: mpsc::Sender, -} - -impl AccountActor { - /// Constructs a new account actor with the given configuration. - pub fn new(account_id: AccountId, actor_context: &AccountActorContext) -> Self { - Self { - account_id, - clients: actor_context.clients.clone(), - state: actor_context.state.clone(), - config: actor_context.config, - request: actor_context.request_tx.clone(), - } - } - - /// Runs the account actor, processing notifications and managing state until shutdown. - /// - /// The return value signals the shutdown category to the coordinator: - /// - /// - `Ok(())`: intentional shutdown (idle timeout). - /// - `Err(_)`: crash (database error, semaphore failure, or any other bug). - pub async fn run( - self, - semaphore: Arc, - mut view_rx: watch::Receiver, - shutdown: CancellationToken, - ) -> anyhow::Result<()> { - let account_id = self.account_id; - - // Load the account once and keep it in memory for the actor's lifetime, advancing it from - // the delta of each transaction the actor itself lands. The coordinator only spawns actors - // for accounts whose creation has been committed, so the account must exist. Held in an - // `Arc` so building a transaction candidate shares this account rather than deep-cloning it - // (expensive for large storage maps). The actor is the sole writer and advances it via - // `Arc::make_mut`, which is cheap because the account is never mutated while a candidate is - // in flight (execution is awaited to completion before any patch/reload). - let mut account = Arc::new( - self.state - .db - .get_account(account_id) - .await - .context("failed to load committed account")? - .context("no committed state for the account; the coordinator must only spawn actors for committed accounts")?, - ); - - // Determine initial mode by querying the DB for available notes. `next_retry_block` records - // when a currently-ineligible note (awaiting backoff or an execution-hint window) becomes - // eligible, so the actor can wait for that block instead of re-querying every block. - let block_num = self.state.chain.chain_tip_block_number(); - let availability = self - .state - .db - .available_notes(account_id, block_num, self.config.max_note_attempts) - .await - .context("failed to check for available notes")?; - let mut next_retry_block = availability.next_retry_block; - let mut mode = if availability.eligible.is_empty() { - ActorMode::NoViableNotes - } else { - ActorMode::NotesAvailable - }; - - // Local cursor over the view's monotone note counter. Mark the spawn-time view as seen so - // the first `changed()` corresponds to the next committed block. - let mut notes_cursor = view_rx.borrow_and_update().notes_seen; - - // Absolute instant at which the actor deactivates if it has done no real work. The - // coordinator pushes a view to every actor on every committed block, so a relative timer - // would restart on each update and a workless actor would never expire on an active chain. - // The deadline is only pushed back when the actor actually executes a transaction. - let mut idle_deadline = tokio::time::Instant::now() + self.config.idle_timeout; - - loop { - // Acquire an execution permit only when there are notes to process. - let tx_permit_acquisition = match mode { - ActorMode::NoViableNotes | ActorMode::WaitForBlock { .. } => { - std::future::pending().boxed() - }, - ActorMode::NotesAvailable => semaphore.acquire().boxed(), - }; - - // The idle timer only ticks while there is nothing to do. - let idle_timeout_sleep = match mode { - ActorMode::NoViableNotes if next_retry_block.is_none() => { - tokio::time::sleep_until(idle_deadline).boxed() - }, - _ => std::future::pending().boxed(), - }; - - tokio::select! { - // Check shutdown first, then poll the view before the idle timer, so cancellation - // wins and a pending update is always processed rather than racing an idle - // shutdown. Tokio native. - biased; - - () = shutdown.cancelled() => return Ok(()), - // A committed block updated this account's view: the submission may have landed - // (advancing the in-memory account by its own delta) or expired, or new notes / a - // due retry may make work available. All of this is answered in memory. - changed = view_rx.changed() => { - changed.context("coordinator dropped the account view channel")?; - let view = view_rx.borrow_and_update().clone(); - mode = self - .reevaluate_mode(&mut account, mode, &view, &mut notes_cursor, next_retry_block) - .await?; - }, - // Execute a transaction once a permit is available. - permit = tx_permit_acquisition => { - let _permit = permit.context("semaphore closed")?; - let chain_state = self.state.chain.get_cloned(); - let block_num = chain_state.chain_tip_header.block_num(); - let Selection { candidate, rejected, next_retry_block: retry } = - select_candidate( - &self.state.db, - &account, - chain_state, - self.config.max_notes_per_tx, - self.config.max_note_attempts, - ) - .await?; - // Selection reports the notes the account's allowlist rejects; the actor owns - // the write, as it does for every other note failure. - self.mark_notes_failed(&rejected, block_num).await; - next_retry_block = retry; - mode = match candidate { - Some(candidate) => { - let next = self - .execute_transactions(account_id, candidate, &mut account) - .await?; - // The actor did real work; push the idle deadline back. - idle_deadline = tokio::time::Instant::now() + self.config.idle_timeout; - next - }, - None => ActorMode::NoViableNotes, - }; - } - // Idle timeout: actor has been idle too long, deactivate. - () = idle_timeout_sleep => { - debug!( - target: LOG_TARGET, - "Account actor deactivated due to idle timeout", - account.id = account_id - ); - return Ok(()); - } - } - } - } - - /// Decides the actor's next mode after the coordinator pushes a fresh [`AccountView`], advancing - /// the in-memory account when the actor's own transaction lands. - /// - /// - In `NotesAvailable`, keep the mode so the pending permit acquisition can complete. - /// - In `NoViableNotes`, advance to `NotesAvailable` only if the view shows new notes (its - /// counter moved past `notes_cursor`) or a scheduled retry is due (`next_retry_block` reached); - /// otherwise stay idle without touching the DB. - /// - In `WaitForBlock`, use the view rather than a DB query: - /// - If `last_committed_tx` equals the actor's submitted id, the transaction landed: apply its - /// `pending_patch` to the in-memory account and resume selection. - /// - Else if `tx_expiration_delta` blocks have passed since submission, the submission expired: - /// reload the account from the DB (in case a different transaction changed it while we - /// waited) and resume selection. - /// - Otherwise keep waiting. - async fn reevaluate_mode( - &self, - account: &mut Arc, - mode: ActorMode, - view: &AccountView, - notes_cursor: &mut u64, - next_retry_block: Option, - ) -> anyhow::Result { - let next = match mode { - // A permit acquisition is already in flight; let it complete rather than cancel it. - ActorMode::NotesAvailable => ActorMode::NotesAvailable, - - // Resume selection only when there is a reason to: new notes arrived, or a previously - // ineligible note's backoff/hint window is now due. Otherwise stay idle, no DB query. - ActorMode::NoViableNotes => { - let new_work = view.notes_seen > *notes_cursor; - let retry_due = next_retry_block.is_some_and(|block| view.chain_tip >= block); - if new_work || retry_due { - ActorMode::NotesAvailable - } else { - ActorMode::NoViableNotes - } - }, - - // Waiting on a submission: detect landing or expiry from the view, not the DB. - ActorMode::WaitForBlock { - submitted_tx_id, - submitted_at, - pending_patch, - } => { - let elapsed = view.chain_tip.checked_sub(submitted_at.as_u32()).unwrap_or_default(); - if view.last_committed_tx == Some(submitted_tx_id) { - // The landed transaction is the one we executed, so the committed state is our - // in-memory account plus the patch it produced. `make_mut` does not clone here: - // the candidate that shared this `Arc` was dropped when its execution - // completed, so the actor holds the only reference. - Arc::make_mut(account) - .apply_patch(&pending_patch) - .context("failed to apply landed transaction patch to in-memory account")?; - info!( - target: LOG_TARGET, - "submitted transaction landed; advanced in-memory account by its patch", - account.id = self.account_id, - transaction.id = submitted_tx_id - ); - ActorMode::NotesAvailable - } else if elapsed.as_u32() >= u32::from(self.config.tx_expiration_delta.get()) { - info!( - target: LOG_TARGET, - "submitted transaction expired", - account.id = self.account_id, - transaction.submitted_at = submitted_at, - tip.number = view.chain_tip, - transaction.expiration_delta = self.config.tx_expiration_delta.get() - ); - // The submission did not land. Reload the authoritative account in case a - // different transaction changed it while we waited, then resume selection. - if let Some(latest) = self - .state - .db - .get_account(self.account_id) - .await - .context("failed to reload account after submission expiry")? - { - *account = Arc::new(latest); - } - ActorMode::NotesAvailable - } else { - ActorMode::WaitForBlock { - submitted_tx_id, - submitted_at, - pending_patch, - } - } - }, - }; - - // Whenever the actor resumes selection it accounts for every note seen so far, so sync the - // cursor to the view's counter in that one place. - if matches!(next, ActorMode::NotesAvailable) { - *notes_cursor = view.notes_seen; - } - Ok(next) - } - - /// Execute a transaction candidate and mark notes as failed as required. - /// - /// Returns the new actor mode based on the execution result. - /// - /// Transient infrastructure failures (prover unreachable, RPC transport hiccup, RPC gRPC - /// error) are retried inside [`execute::NtxContext::execute_transaction`]. - /// Any error reaching this method is therefore terminal for the candidate: the batch's notes - /// are marked failed and the actor moves on. - /// - /// On a submission rejection (`NtxError::Submission`), `account` is refreshed in place from the - /// DB before returning: the rejection usually means the in-memory account diverged from the - /// committed chain, so the next selection must build on the authoritative state rather than - /// re-declaring the stale commitment. - #[miden_instrument( - name = "ntx.actor.execute_transactions", - fields(account.id = account_id), - )] - async fn execute_transactions( - &self, - account_id: AccountId, - tx_candidate: TransactionCandidate, - account: &mut Arc, - ) -> anyhow::Result { - let block_num = tx_candidate.chain_tip_header.block_num(); - - // Execute the selected transaction. - let context = execute::NtxContext::new( - self.clients.prover.clone(), - self.clients.rpc.clone(), - self.state.script_cache.clone(), - self.state.db.clone(), - self.config.max_cycles, - self.state.tx_args.clone(), - self.config.request_backoff_initial, - self.config.request_backoff_max, - ); - - let sponsored_notes = tx_candidate.notes.clone(); - // Failures of a sponsorship note are attributed to the feature note of its bundle: - // sponsorship notes have no row in the `notes` table, so the feature note carries the - // attempt tracking for its whole bundle. - let sponsor_to_feature = tx_candidate.sponsor_to_feature_nullifier(); - let account_id = tx_candidate.account.id(); - let note_ids: Vec<_> = sponsored_notes - .iter() - .flat_map(|sponsored| { - std::iter::once(sponsored.feature.as_note().id()) - .chain(sponsored.sponsorships.iter().map(Note::id)) - }) - .collect(); - info!( - target: LOG_TARGET, - "executing network transaction", - account.id = account_id, - note.ids = note_ids.as_slice(), - note.count = note_ids.len() - ); - - let execution_result = context.execute_transaction(tx_candidate).await; - Ok(match execution_result { - Ok(execute::NtxExecutionResult { - tx_id, - account_patch, - failed_notes, - deferred_notes, - oversized_notes, - fetched_scripts, - }) => { - // `filter_notes` has already partitioned the failed notes: - // - `deferred_notes` were dropped only because the combined per-tx cycle budget was - // exhausted but are consumable on their own. They are left un-penalized so - // `available_notes` re-selects them next round, letting a large note land in its - // own transaction once its batch-mates commit. - // - `oversized_notes` exceed the per-tx cycle budget on their own and can never be - // consumed. They are discarded immediately so they stop being re-selected. - // - `failed_notes` are genuine consumability failures and are penalized as usual. - info!( - target: LOG_TARGET, - "network transaction executed", - account.id = account_id, - transaction.id = tx_id, - note.failed.count = failed_notes.len(), - note.deferred.count = deferred_notes.len(), - note.oversized.count = oversized_notes.len() - ); - self.cache_note_scripts(fetched_scripts).await; - - log_deferred_notes(deferred_notes); - - // Only feature notes are discarded permanently. An oversized sponsorship (its - // isolated re-check runs the reclaim path, so this is unexpected) is charged to its - // feature note as a regular failure instead: the feature itself may still be - // consumable with a different sponsorship. - let (oversized_sponsorships, oversized_features): (Vec<_>, Vec<_>) = - oversized_notes - .into_iter() - .partition(|f| sponsor_to_feature.contains_key(&f.note().id())); - - let mut to_penalize = failed_notes; - to_penalize.extend(oversized_sponsorships); - let failed_notes = attribute_failed_notes(to_penalize, &sponsor_to_feature); - self.mark_notes_failed(&failed_notes, block_num).await; - - let nullifiers = log_oversized_notes(oversized_features); - self.discard_notes(&nullifiers, block_num).await; - - // A non-empty successful set is guaranteed by `filter_notes` (it returns - // `AllNotesFailed` otherwise), so a transaction was always submitted here and - // carries real work to wait for. - ActorMode::WaitForBlock { - submitted_tx_id: tx_id, - submitted_at: block_num, - pending_patch: account_patch, - } - }, - // Transaction execution failed. - Err(err) => { - error!( - &err, - target: LOG_TARGET, - "network transaction failed", - account.id = account_id, - note.ids = note_ids.as_slice() - ); - - // A rejected submission (e.g. an account-commitment mismatch) means our in-memory - // account has diverged from the committed chain. Reload it from the DB below so the - // next selection builds on the authoritative state instead of re-declaring the same - // stale commitment. Resumption is gated on the next committed block by - // `NoViableNotes`, so this cannot hot-loop. - let submission_rejected = matches!(err, execute::NtxError::Submission(_)); - - // For `AllNotesFailed`, use the per-note errors which contain the specific reason - // each note failed (e.g. consumability check details). Whole-transaction errors are - // recorded against the feature notes only: sponsorships have no row in the `notes` - // table. - let failed_notes: Vec<_> = match err { - execute::NtxError::AllNotesFailed(per_note) => { - attribute_failed_notes(per_note, &sponsor_to_feature) - }, - other => { - let error: NoteError = Arc::new(other); - sponsored_notes - .iter() - .map(|sponsored| { - let feature = sponsored.feature.as_note(); - info!( - error.as_ref(), - target: LOG_TARGET, - "note failed: transaction execution error", - note.id = feature.id(), - note.nullifier = feature.nullifier() - ); - (feature.nullifier(), error.clone()) - }) - .collect() - }, - }; - self.mark_notes_failed(&failed_notes, block_num).await; - - if submission_rejected { - if let Some(latest) = self - .state - .db - .get_account(self.account_id) - .await - .context("failed to reload account after a rejected submission")? - { - info!( - target: LOG_TARGET, - "reloaded account from the database after a rejected submission", - account.id = account_id - ); - *account = Arc::new(latest); - } - } - - ActorMode::NoViableNotes - }, - }) - } - - /// Sends requests to the coordinator to cache note scripts fetched from the remote RPC service. - async fn cache_note_scripts(&self, scripts: Vec<(Word, NoteScript)>) { - for (script_root, script) in scripts { - if self - .request - .send(ActorRequest::CacheNoteScript { script_root, script }) - .await - .is_err() - { - break; - } - } - } - - /// Sends a request to the coordinator to mark notes as failed and waits for the DB write to - /// complete. This prevents a race condition where the actor could re-select the same notes - /// before the failure counts are updated in the database. - async fn mark_notes_failed( - &self, - failed_notes: &[(Nullifier, NoteError)], - block_num: BlockNumber, - ) { - // Avoid an empty coordinator round-trip (and DB write-transaction) on the common - // no-failures path. - if failed_notes.is_empty() { - return; - } - let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); - if self - .request - .send(ActorRequest::NotesFailed { - failed_notes: failed_notes.to_vec(), - block_num, - ack_tx, - }) - .await - .is_err() - { - return; - } - // Wait for the coordinator to confirm the DB write. - let _ = ack_rx.await; - } - - /// Sends a request to the coordinator to discard notes (mark them permanently unconsumable) and - /// waits for the DB write to complete. Like [`Self::mark_notes_failed`], the acknowledgement - /// prevents the actor from re-selecting the notes before the write lands. - async fn discard_notes(&self, nullifiers: &[Nullifier], block_num: BlockNumber) { - // Avoid an empty coordinator round-trip (and DB write-transaction) on the common - // no-oversized-notes path. - if nullifiers.is_empty() { - return; - } - let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); - if self - .request - .send(ActorRequest::NotesDiscarded { - nullifiers: nullifiers.to_vec(), - block_num, - ack_tx, - }) - .await - .is_err() - { - return; - } - // Wait for the coordinator to confirm the DB write. - let _ = ack_rx.await; - } -} - -#[cfg(test)] -mod tests { - - use miden_protocol::ONE; - use miden_protocol::account::{Account, AccountPatch, AccountStoragePatch, AccountVaultPatch}; - use tokio::sync::watch; - - use super::*; - use crate::test_utils::{mock_account, mock_network_account_id, mock_transaction_id}; - - /// Builds a valid nonce-only [`AccountPatch`] that advances `account` by a single nonce. - fn nonce_bump_patch(account: &Account) -> AccountPatch { - AccountPatch::new( - account.id(), - AccountStoragePatch::default(), - AccountVaultPatch::default(), - None, - Some(account.nonce() + ONE), - ) - .expect("a nonce-only patch is valid") - } - - /// Builds an actor wired to `db` for the given account. - fn test_actor(db: &NtxDbReader, account: &Account) -> AccountActor { - let ctx = AccountActorContext::test(db); - AccountActor::new(account.id(), &ctx) - } - - /// Builds an [`AccountView`] for driving `reevaluate_mode` directly. - fn view( - chain_tip: u32, - last_committed_tx: Option, - notes_seen: u64, - ) -> AccountView { - AccountView { - chain_tip: chain_tip.into(), - last_committed_tx, - notes_seen, - } - } - - /// When the submitted transaction lands (its id is the view's latest committed tx), the actor - /// advances its in-memory account by exactly the patch the transaction produced. - #[tokio::test] - async fn landing_advances_in_memory_account_by_its_patch() { - let (db, _dir) = crate::db::test_setup().await; - let account = mock_account(mock_network_account_id()); - let submitted = mock_transaction_id(7); - - let patch = nonce_bump_patch(&account); - let mut expected = account.clone(); - expected.apply_patch(&patch).unwrap(); - - let actor = test_actor(&db, &account); - let mut in_memory = Arc::new(account.clone()); - let mut notes_cursor = 0; - // The view reports our submission as the account's latest committed transaction. - let view = view(1, Some(submitted), 0); - let mode = actor - .reevaluate_mode( - &mut in_memory, - ActorMode::WaitForBlock { - submitted_tx_id: submitted, - submitted_at: 0_u32.into(), - pending_patch: patch, - }, - &view, - &mut notes_cursor, - None, - ) - .await - .unwrap(); - - assert!(matches!(mode, ActorMode::NotesAvailable), "a landed tx must resume selection"); - assert_eq!( - in_memory.to_commitment(), - expected.to_commitment(), - "the in-memory account must be advanced by the landed tx's patch", - ); - } - - /// While the submission has neither landed nor expired, the actor keeps waiting and leaves its - /// in-memory account untouched. - #[tokio::test] - async fn pending_submission_keeps_waiting_without_touching_account() { - let (db, _dir) = crate::db::test_setup().await; - let account = mock_account(mock_network_account_id()); - - // The view shows no committed tx for the account (submission has not landed) and a tip well - // within `tx_expiration_delta` of the submission block, so it has not expired either. - let actor = test_actor(&db, &account); - let mut in_memory = Arc::new(account.clone()); - let mut notes_cursor = 0; - let submitted = mock_transaction_id(7); - let view = view(1, None, 0); - let mode = actor - .reevaluate_mode( - &mut in_memory, - ActorMode::WaitForBlock { - submitted_tx_id: submitted, - submitted_at: 0_u32.into(), - pending_patch: nonce_bump_patch(&account), - }, - &view, - &mut notes_cursor, - None, - ) - .await - .unwrap(); - - match mode { - ActorMode::WaitForBlock { submitted_tx_id, .. } => { - assert_eq!(submitted_tx_id, submitted, "the actor must keep waiting on its own tx"); - }, - other => panic!("expected to stay in WaitForBlock, got {other:?}"), - } - assert_eq!( - in_memory.to_commitment(), - account.to_commitment(), - "a still-pending submission must not change the in-memory account", - ); - } - - /// An idle actor must not re-select (and so must not hit the DB) on a view that only advances - /// the chain tip: no new notes arrived and no scheduled retry is due. - #[tokio::test] - async fn idle_actor_ignores_view_without_new_work() { - let (db, _dir) = crate::db::test_setup().await; - let account = mock_account(mock_network_account_id()); - let actor = test_actor(&db, &account); - let mut in_memory = Arc::new(account.clone()); - let mut notes_cursor = 3; - - // notes_seen matches the cursor (no new notes) and there is no pending retry. - let view = view(10, None, 3); - let mode = actor - .reevaluate_mode( - &mut in_memory, - ActorMode::NoViableNotes, - &view, - &mut notes_cursor, - None, - ) - .await - .unwrap(); - - assert!( - matches!(mode, ActorMode::NoViableNotes), - "no new notes and no due retry must leave the actor idle", - ); - assert_eq!(notes_cursor, 3, "the cursor is untouched while the actor stays idle"); - } - - /// New notes (the view's counter moving past the local cursor) wake an idle actor. - #[tokio::test] - async fn new_notes_wake_idle_actor() { - let (db, _dir) = crate::db::test_setup().await; - let account = mock_account(mock_network_account_id()); - let actor = test_actor(&db, &account); - let mut in_memory = Arc::new(account.clone()); - let mut notes_cursor = 3; - - let view = view(10, None, 4); - let mode = actor - .reevaluate_mode( - &mut in_memory, - ActorMode::NoViableNotes, - &view, - &mut notes_cursor, - None, - ) - .await - .unwrap(); - - assert!(matches!(mode, ActorMode::NotesAvailable), "a new note must trigger a re-select"); - assert_eq!(notes_cursor, 4, "the cursor advances to the observed note count"); - } - - /// A scheduled retry wakes an idle actor exactly when the chain tip reaches `next_retry_block`, - /// and not before. This is how backoff/hint retries fire without a new note arriving. - #[tokio::test] - async fn due_retry_wakes_idle_actor_at_its_block() { - let (db, _dir) = crate::db::test_setup().await; - let account = mock_account(mock_network_account_id()); - let actor = test_actor(&db, &account); - let mut in_memory = Arc::new(account.clone()); - - // Tip below the retry block: stay idle. - let mut notes_cursor = 0; - let early = actor - .reevaluate_mode( - &mut in_memory, - ActorMode::NoViableNotes, - &view(9, None, 0), - &mut notes_cursor, - Some(10_u32.into()), - ) - .await - .unwrap(); - assert!(matches!(early, ActorMode::NoViableNotes), "a retry is not due before its block"); - - // Tip reaches the retry block: re-select. - let due = actor - .reevaluate_mode( - &mut in_memory, - ActorMode::NoViableNotes, - &view(10, None, 0), - &mut notes_cursor, - Some(10_u32.into()), - ) - .await - .unwrap(); - assert!(matches!(due, ActorMode::NotesAvailable), "a due retry must trigger a re-select"); - } - - /// The idle timeout must still fire while the coordinator keeps pushing a view every block. The - /// coordinator updates every actor's view on every committed block, so a workless actor would - /// never expire if updates reset the idle timer. The deadline is absolute and only pushed back - /// by real work, so repeated view updates cannot keep a no-work actor resident indefinitely. - #[tokio::test] - async fn idle_timeout_fires_despite_repeated_view_updates() { - let (db, _dir) = crate::db::test_setup().await; - // A real network account with a populated allowlist, so re-evaluation on each wake reaches - // a clean "no viable notes" outcome instead of erroring on a missing allowlist slot. - let (account, _) = crate::test_utils::mock_network_account_update(); - let account_id = account.id(); - - // Seed the committed account but no notes, so the actor starts and stays in NoViableNotes - // with no pending retry: it remains genuinely note-less and the idle timer ticks. - db.upsert_account_for_test(account_id, account.clone(), mock_transaction_id(1)) - .await - .unwrap(); - - let mut ctx = AccountActorContext::test(&db); - // Short idle timeout keeps the test fast. - ctx.config.idle_timeout = Duration::from_millis(300); - - let actor = AccountActor::new(account_id, &ctx); - let (view_tx, view_rx) = watch::channel(view(0, None, 0)); - let semaphore = Arc::new(Semaphore::new(1)); - let handle = tokio::spawn(actor.run(semaphore, view_rx, CancellationToken::new())); - - // Push a view update far more often than the idle timeout, advancing only the chain tip (no - // new notes), for longer than the test's deadline. With a relative timer every update would - // restart it and the actor would never deactivate, failing the timeout below. - let notifier = tokio::spawn(async move { - loop { - view_tx.send_modify(|v| v.chain_tip = v.chain_tip.child()); - tokio::time::sleep(Duration::from_millis(50)).await; - } - }); - - let result = tokio::time::timeout(Duration::from_secs(3), handle) - .await - .expect("actor must deactivate on idle timeout despite repeated view updates") - .expect("actor task should not panic"); - assert!(result.is_ok(), "idle deactivation is a clean shutdown"); - - notifier.abort(); - } -} diff --git a/bin/ntx-builder/src/attempt.rs b/bin/ntx-builder/src/attempt.rs new file mode 100644 index 0000000000..ba72810fc2 --- /dev/null +++ b/bin/ntx-builder/src/attempt.rs @@ -0,0 +1,357 @@ +//! One transaction attempt against one network account. + +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Context; +use miden_node_tracing::{error, info, miden_instrument}; +use miden_node_utils::lru_cache::LruCache; +use miden_protocol::Word; +use miden_protocol::account::AccountId; +use miden_protocol::block::BlockNumber; +use miden_protocol::note::{Note, NoteScript, Nullifier}; +use miden_protocol::transaction::{TransactionArgs, TransactionId}; + +use crate::candidate::TransactionCandidate; +use crate::chain_state::ChainState; +use crate::clients::{RemoteTransactionProver, RpcClient}; +use crate::db::NtxDbReader; +use crate::execute::{NtxError, NtxExecutionResult}; +use crate::selection::{ + Selection, + attribute_failed_notes, + log_deferred_notes, + log_oversized_notes, + select_candidate, +}; +use crate::{LOG_TARGET, NoteError, execute}; + +// ATTEMPT CONTEXT +// ================================================================================================ + +/// gRPC clients used by an attempt to interact with the node's services. +#[derive(Clone)] +pub struct GrpcClients { + /// Client for interacting with the RPC service in order to load account state. + pub rpc: RpcClient, + /// Client for remote transaction proving. + pub prover: RemoteTransactionProver, +} + +/// Per-attempt configuration knobs. +#[derive(Debug, Clone, Copy)] +pub struct AttemptConfig { + /// Maximum number of notes per transaction. Sponsorship notes count against this budget. + pub max_notes_per_tx: NonZeroUsize, + /// Maximum number of note execution attempts before dropping a note. + pub max_note_attempts: usize, + /// Maximum number of VM execution cycles for network transactions. + pub max_cycles: u32, + /// Initial sleep applied between per-request retries on transient infrastructure failures + /// (prover unreachable, RPC transport error, RPC gRPC hiccup). Doubles each retry up to + /// [`Self::request_backoff_max`]. + pub request_backoff_initial: Duration, + /// Upper bound on the per-request retry backoff sleep. + pub request_backoff_max: Duration, +} + +/// Resources every attempt shares. +#[derive(Clone)] +pub struct AttemptContext { + /// gRPC clients used by the attempt. + pub clients: GrpcClients, + /// Read-only database handle. An attempt performs no writes. + pub db: NtxDbReader, + /// Shared LRU cache for note scripts retrieved over RPC. + pub script_cache: LruCache, + /// [`TransactionArgs`] used by every network transaction. These are constant and are therefore + /// prebuilt once and cloned per transaction. + pub tx_args: TransactionArgs, + /// Per-attempt configuration knobs. + pub config: AttemptConfig, +} + +#[cfg(test)] +impl AttemptContext { + /// Creates a minimal [`AttemptContext`] for tests. + pub fn test(db: &NtxDbReader) -> Self { + use url::Url; + + let url = Url::parse("http://127.0.0.1:1").unwrap(); + let block_header = crate::test_utils::mock_block_header(0_u32.into()); + let trusted_validator_signing_keys = block_header.validator_config().keys().to_vec(); + + Self { + clients: GrpcClients { + rpc: RpcClient::new( + url.clone(), + Word::default(), + trusted_validator_signing_keys, + Duration::from_secs(10), + Duration::from_millis(1), + Duration::from_millis(10), + ) + .expect("rpc client should be constructed"), + prover: RemoteTransactionProver::new(url, Duration::from_secs(10)) + .expect("prover client should be constructed"), + }, + db: db.clone(), + script_cache: LruCache::new(NonZeroUsize::new(1).unwrap()), + tx_args: crate::selection::build_tx_args( + std::num::NonZeroU16::new(30).expect("literal is non-zero"), + ), + config: AttemptConfig { + max_notes_per_tx: NonZeroUsize::new(20).expect("literal is non-zero"), + max_note_attempts: 30, + max_cycles: 1 << 18, + request_backoff_initial: Duration::from_millis(1), + request_backoff_max: Duration::from_millis(10), + }, + } + } +} + +// ATTEMPT OUTCOME +// ================================================================================================ + +/// Note bookkeeping an attempt produced. The scheduler persists all of it, whatever the +/// [`AttemptResult`]. +#[derive(Default)] +pub struct NoteUpdates { + /// Notes whose attempt counter must be incremented, keyed by the nullifier the failure is + /// recorded under. + pub failed: Vec<(Nullifier, NoteError)>, + /// Notes that can never be consumed and must be marked permanently unconsumable. + pub discarded: Vec, + /// Corrected eligibility blocks for notes whose stored block was too permissive. Persisting + /// these is what stops the account being selected again for a note it cannot attempt. + pub eligibility: Vec<(Nullifier, BlockNumber)>, + /// Note scripts fetched over RPC that should be persisted to the local cache. + pub scripts: Vec<(Word, NoteScript)>, +} + +/// How an attempt ended. +pub enum AttemptResult { + /// A transaction was proven and submitted. The account is now in flight and is not retried + /// until the transaction commits or expires. + Submitted { tx_id: TransactionId }, + /// Selection produced no candidate. Any note whose stored eligibility was wrong is corrected + /// through [`NoteUpdates::eligibility`], so the account is not selected again for it. + NoWork, + /// The candidate failed for a reason attributed to its notes, which are named in + /// [`NoteUpdates::failed`]. + Failed, + /// The attempt could not be attributed to any note: the account or its notes could not be read. + /// No note is penalized, because the failure is not the notes' fault. + Aborted(anyhow::Error), +} + +/// What an attempt did, and what the scheduler must persist as a result. +pub struct AttemptOutcome { + /// The account the attempt ran against. + pub account_id: AccountId, + /// Reference block of the attempt. Note bookkeeping is recorded against it. + pub block_num: BlockNumber, + /// Note bookkeeping to persist, populated on every path. + pub notes: NoteUpdates, + /// How the attempt ended. + pub result: AttemptResult, +} + +// ATTEMPT +// ================================================================================================ + +/// Runs a single transaction attempt for `account_id`. +#[miden_instrument( + name = "ntx.attempt", + fields( + account.id = account_id, + reference_block.number = chain.chain_tip_header.block_num(), + ), +)] +pub async fn attempt( + ctx: AttemptContext, + account_id: AccountId, + chain: ChainState, +) -> AttemptOutcome { + let block_num = chain.chain_tip_header.block_num(); + match run(&ctx, account_id, chain).await { + Ok(outcome) => outcome, + Err(err) => AttemptOutcome { + account_id, + block_num, + notes: NoteUpdates::default(), + result: AttemptResult::Aborted(err), + }, + } +} + +/// The attempt body. Errors here are infrastructure failures that cannot be charged to a note. +async fn run( + ctx: &AttemptContext, + account_id: AccountId, + chain: ChainState, +) -> anyhow::Result { + let block_num = chain.chain_tip_header.block_num(); + + // The scheduler only picks accounts whose creation is committed, so a missing row means the + // account state was removed between the query and this read. Report no work rather than an + // error: there is nothing to penalize. + let Some(account) = ctx + .db + .get_account(account_id) + .await + .context("failed to load committed account")? + else { + return Ok(AttemptOutcome { + account_id, + block_num, + notes: NoteUpdates::default(), + result: AttemptResult::NoWork, + }); + }; + let account = Arc::new(account); + + let Selection { candidate, rejected, stale_eligibility } = select_candidate( + &ctx.db, + &account, + chain, + ctx.config.max_notes_per_tx, + ctx.config.max_note_attempts, + ) + .await?; + + // Notes outside the account's allowlist are penalized, and notes whose stored eligibility was + // too permissive are corrected, whether or not a candidate was built. + let mut notes = NoteUpdates { + failed: rejected, + eligibility: stale_eligibility, + ..NoteUpdates::default() + }; + + let Some(candidate) = candidate else { + return Ok(AttemptOutcome { + account_id, + block_num, + notes, + result: AttemptResult::NoWork, + }); + }; + + let result = execute_candidate(ctx, account_id, candidate, &mut notes).await; + + Ok(AttemptOutcome { account_id, block_num, notes, result }) +} + +/// Executes, proves and submits `candidate`, recording the note bookkeeping into `notes`. +async fn execute_candidate( + ctx: &AttemptContext, + account_id: AccountId, + candidate: TransactionCandidate, + notes: &mut NoteUpdates, +) -> AttemptResult { + // Failures of a sponsorship note are attributed to the feature note of its bundle: sponsorship + // notes have no row in the `notes` table, so the feature note carries the attempt tracking for + // its whole bundle. + let sponsor_to_feature = candidate.sponsor_to_feature_nullifier(); + let note_ids: Vec<_> = candidate + .notes + .iter() + .flat_map(|sponsored| { + std::iter::once(sponsored.feature.as_note().id()) + .chain(sponsored.sponsorships.iter().map(Note::id)) + }) + .collect(); + let feature_nullifiers: Vec<_> = candidate + .notes + .iter() + .map(|sponsored| sponsored.feature.as_note().nullifier()) + .collect(); + + info!( + target: LOG_TARGET, + "executing network transaction", + account.id = account_id, + note.ids = note_ids.as_slice(), + note.count = note_ids.len() + ); + + let context = execute::NtxContext::new( + ctx.clients.prover.clone(), + ctx.clients.rpc.clone(), + ctx.script_cache.clone(), + ctx.db.clone(), + ctx.config.max_cycles, + ctx.tx_args.clone(), + ctx.config.request_backoff_initial, + ctx.config.request_backoff_max, + ); + + match context.execute_transaction(candidate).await { + Ok(NtxExecutionResult { + tx_id, + failed_notes, + deferred_notes, + oversized_notes, + fetched_scripts, + }) => { + info!( + target: LOG_TARGET, + "network transaction executed", + account.id = account_id, + transaction.id = tx_id, + note.failed.count = failed_notes.len(), + note.deferred.count = deferred_notes.len(), + note.oversized.count = oversized_notes.len() + ); + notes.scripts = fetched_scripts; + + log_deferred_notes(deferred_notes); + + let (oversized_sponsorships, oversized_features): (Vec<_>, Vec<_>) = oversized_notes + .into_iter() + .partition(|f| sponsor_to_feature.contains_key(&f.note().id())); + + let mut to_penalize = failed_notes; + to_penalize.extend(oversized_sponsorships); + notes.failed.extend(attribute_failed_notes(to_penalize, &sponsor_to_feature)); + notes.discarded = log_oversized_notes(oversized_features); + + AttemptResult::Submitted { tx_id } + }, + Err(err) => { + error!( + &err, + target: LOG_TARGET, + "network transaction failed", + account.id = account_id, + note.ids = note_ids.as_slice() + ); + + let failed = match err { + NtxError::AllNotesFailed(per_note) => { + attribute_failed_notes(per_note, &sponsor_to_feature) + }, + other => { + let error: NoteError = Arc::new(other); + feature_nullifiers + .into_iter() + .map(|nullifier| { + info!( + error.as_ref(), + target: LOG_TARGET, + "note failed: transaction execution error", + note.nullifier = nullifier + ); + (nullifier, error.clone()) + }) + .collect() + }, + }; + notes.failed.extend(failed); + + AttemptResult::Failed + }, + } +} diff --git a/bin/ntx-builder/src/builder.rs b/bin/ntx-builder/src/builder.rs index 96489b24e5..ade9e55b71 100644 --- a/bin/ntx-builder/src/builder.rs +++ b/bin/ntx-builder/src/builder.rs @@ -1,33 +1,29 @@ use std::pin::Pin; -use std::sync::Arc; use anyhow::Context; use futures::Stream; use miden_node_tracing::{info, miden_instrument}; use miden_node_utils::shutdown::CancellationToken; use miden_node_utils::tasks::Tasks; -use miden_protocol::account::AccountId; use miden_protocol::block::{BlockNumber, SignedBlock}; use tokio::net::TcpListener; -use tokio::sync::mpsc; use tokio_stream::StreamExt; -use crate::actor::ActorRequest; -use crate::chain_state::SharedChainState; +use crate::attempt::AttemptOutcome; +use crate::chain_state::ChainState; use crate::clients::RpcError; use crate::committed_block::CommittedBlockEffects; -use crate::coordinator::Coordinator; use crate::db::NtxDbWriter; +use crate::scheduler::Scheduler; use crate::server::NtxBuilderRpcServer; use crate::{LOG_TARGET, NtxBuilderConfig}; /// Discriminator returned by the steady-state `select!` so the dispatch can run on a fully-owned -/// `&mut self` instead of three concurrent borrows. The `Block` variant is boxed since a -/// `SignedBlock` dwarfs the other two payloads. +/// `&mut self` instead of two concurrent borrows. The `Block` variant is boxed since a +/// `SignedBlock` dwarfs the other payloads. enum SteadyStateAction { Block(Box>>), - Request(Option), - Respawn(Option), + Completion(anyhow::Result), Shutdown, } @@ -44,16 +40,13 @@ pub(crate) type BlockStream = /// Network transaction builder component. /// -/// Runs in three phases: -/// 1. **Catch-up**: drain the committed-block subscription, applying each block to the local DB -/// and in-memory chain, until the local tip matches the node-reported `committed_chain_tip` -/// (signaled by `is_synced` flipping to `true`). No actors run. -/// 2. **Boundary**: query the DB for accounts with carry-over pending notes (e.g. from a previous -/// process) and spawn an actor for each. -/// 3. **Steady-state**: on every subsequent committed block, apply the effects, advance the chain, -/// and have the coordinator spawn-if-missing for newly-targeted accounts then wake every active -/// actor. Concurrently drain actor requests (`NotesFailed`, `CacheNoteScript`) so the actors' -/// DB writes happen serialized through the builder. +/// Runs in two phases: +/// 1. **Catch-up**: drain the committed-block subscription, applying each block to the local DB and +/// in-memory chain, until the local tip matches the node-reported `committed_chain_tip` +/// (signaled by `is_synced` flipping to `true`). No transaction attempts run. +/// 2. **Steady-state**: on every committed block, apply the effects, advance the chain, resolve the +/// scheduler's in-flight transactions against the block, and fill the free attempt slots. +/// Concurrently reap finished attempts, persisting the note bookkeeping each one reports. pub struct NetworkTransactionBuilder { /// Configuration for the builder. config: NtxBuilderConfig, @@ -63,13 +56,10 @@ pub struct NetworkTransactionBuilder { block_stream: BlockStream, /// Highest block number applied to the DB so far. last_applied_block: BlockNumber, - /// In-memory partial chain shared with every spawned actor through the coordinator. - chain: Arc, - /// Lifecycle owner for `AccountActor` instances. - coordinator: Coordinator, - /// Channel receiving DB-side requests (note-failed bookkeeping, script-cache persistence) from - /// spawned actors. Drained in the steady-state loop so writes happen through the builder. - actor_request_rx: mpsc::Receiver, + /// In-memory partial chain. + chain: ChainState, + /// Owner of the transaction attempts and of the in-flight transaction set. + scheduler: Scheduler, /// `false` until the first applied block whose `committed_chain_tip` matches the just-applied /// block number. Stays `true` afterwards. is_synced: bool, @@ -81,9 +71,8 @@ impl NetworkTransactionBuilder { db: NtxDbWriter, block_stream: BlockStream, last_applied_block: BlockNumber, - chain: Arc, - coordinator: Coordinator, - actor_request_rx: mpsc::Receiver, + chain: ChainState, + scheduler: Scheduler, ) -> Self { Self { config, @@ -91,8 +80,7 @@ impl NetworkTransactionBuilder { block_stream, last_applied_block, chain, - coordinator, - actor_request_rx, + scheduler, is_synced: false, } } @@ -153,38 +141,24 @@ impl NetworkTransactionBuilder { } } - // Phase 2: spawn an actor for every account with carry-over pending notes. Accounts whose - // creation has not been committed yet have their spawn deferred by the coordinator. - let max_note_attempts = self.config.max_note_attempts; - let pending_accounts = self - .db - .accounts_with_pending_notes(max_note_attempts) - .await - .context("failed to load accounts with pending notes at catch-up")?; - info!( - target: LOG_TARGET, - "spawning actors for accounts with carry-over pending notes", - account.ids.count = pending_accounts.len() - ); - for account_id in pending_accounts { - self.coordinator.spawn_actor_when_committed(account_id).await?; - } + // Phase 2: work the accounts that have pending notes, one attempt per free slot, driven by + // committed blocks and by the completion of earlier attempts. + self.scheduler.dispatch(&self.chain).await?; - // Phase 3: drive actors per committed block, plus serialize their DB writes. loop { // Split `&mut self` into disjoint borrows so each `select!` arm holds only the one // field it polls. The action is materialised and self is released before the body // dispatches the work via the regular `&mut self` methods. let action = { let block_stream = &mut self.block_stream; - let actor_request_rx = &mut self.actor_request_rx; - let coordinator = &mut self.coordinator; + let scheduler = &mut self.scheduler; tokio::select! { () = shutdown.cancelled() => SteadyStateAction::Shutdown, block = block_stream.next() => SteadyStateAction::Block(Box::new(block)), - request = actor_request_rx.recv() => SteadyStateAction::Request(request), - respawn = coordinator.next() => SteadyStateAction::Respawn(respawn?), + completion = scheduler.next_completion() => { + SteadyStateAction::Completion(completion) + }, } }; @@ -192,28 +166,18 @@ impl NetworkTransactionBuilder { SteadyStateAction::Block(block) => { let (block, committed_tip) = (*block).context("block stream ended")?.context("block stream failed")?; - let (effects, sponsored_accounts) = + let effects = self.apply_committed_block_with_effects(block, committed_tip).await?; - self.coordinator.handle_committed_block(&effects, &sponsored_accounts).await?; + self.scheduler.handle_committed_block(&effects); + self.scheduler.dispatch(&self.chain).await?; }, - SteadyStateAction::Request(request) => { - let Some(request) = request else { - anyhow::bail!("actor request channel closed unexpectedly"); - }; - handle_actor_request(&self.db, request, self.config.max_note_attempts).await?; - }, - SteadyStateAction::Respawn(respawn) => { - if let Some(account_id) = respawn { - info!( - target: LOG_TARGET, - "respawning actor that shut down with a pending notification", - account.id = account_id - ); - self.coordinator.spawn_actor(account_id); + SteadyStateAction::Completion(outcome) => { + if self.scheduler.handle_completion(&self.db, outcome?).await? { + self.scheduler.dispatch(&self.chain).await?; } }, SteadyStateAction::Shutdown => { - self.coordinator.shutdown().await?; + self.scheduler.shutdown().await; return Ok(()); }, } @@ -239,10 +203,9 @@ impl NetworkTransactionBuilder { self.apply_committed_block_with_effects(block, committed_tip).await.map(drop) } - /// Applies a committed block and returns the computed `CommittedBlockEffects`, plus the - /// accounts whose pending feature notes gained a sponsorship in this block (one entry per - /// sponsorship), so the steady-state loop can hand both to the coordinator without re-deriving - /// them from the signed block. + /// Applies a committed block and returns the computed [`CommittedBlockEffects`], so the caller + /// can resolve the scheduler's in-flight transactions against the same effects without + /// re-deriving them from the signed block. #[miden_instrument( name = "ntx.builder.apply_committed_block", fields( @@ -254,7 +217,7 @@ impl NetworkTransactionBuilder { &mut self, block: SignedBlock, committed_tip: BlockNumber, - ) -> anyhow::Result<(CommittedBlockEffects, Vec)> { + ) -> anyhow::Result { let header = block.header().clone(); let block_num = header.block_num(); @@ -266,43 +229,13 @@ impl NetworkTransactionBuilder { let next_mmr = self.chain.current_mmr(); let effects_for_db = effects.clone(); - let sponsored_accounts = self - .db + self.db .apply_committed_block(effects_for_db, next_mmr) .await .context("failed to apply committed block to DB")?; self.last_applied_block = block_num; - Ok((effects, sponsored_accounts)) - } -} - -/// Handles a single actor request then acknowledges the actor. All writes go through the -/// framework's single writer connection, so the actors' reads cannot starve them. -async fn handle_actor_request( - db: &NtxDbWriter, - request: ActorRequest, - max_note_attempts: usize, -) -> anyhow::Result<()> { - match request { - ActorRequest::NotesFailed { failed_notes, block_num, ack_tx } => { - db.notes_failed(failed_notes, block_num) - .await - .context("failed to persist note failure")?; - let _ = ack_tx.send(()); - }, - ActorRequest::NotesDiscarded { nullifiers, block_num, ack_tx } => { - db.discard_notes(nullifiers, block_num, max_note_attempts) - .await - .context("failed to persist note discard")?; - let _ = ack_tx.send(()); - }, - ActorRequest::CacheNoteScript { script_root, script } => { - db.insert_note_scripts(script_root, script) - .await - .context("failed to cache note script")?; - }, + Ok(effects) } - Ok(()) } diff --git a/bin/ntx-builder/src/candidate.rs b/bin/ntx-builder/src/candidate.rs index a8470fc2b6..b6cd913aff 100644 --- a/bin/ntx-builder/src/candidate.rs +++ b/bin/ntx-builder/src/candidate.rs @@ -87,10 +87,9 @@ fn sponsorship_amount(note: &Note) -> AssetAmount { pub struct TransactionCandidate { /// The current inflight state of the account. /// - /// Wrapped in `Arc` so building a candidate shares the actor's resident account instead of - /// deep-cloning it (which, for accounts with large storage maps, is expensive). The account is - /// only ever read during execution; the actor advances its own copy via `Arc::make_mut` once - /// the candidate has been consumed. + /// Wrapped in `Arc` so building a candidate shares the account the attempt loaded instead of + /// deep-cloning it, which is expensive for accounts with large storage maps. The account is + /// only ever read during execution. pub account: Arc, /// The sponsored feature notes selected for this transaction: each feature note addressed to diff --git a/bin/ntx-builder/src/chain_state.rs b/bin/ntx-builder/src/chain_state.rs index 811fa1f69f..f02ab7d5c5 100644 --- a/bin/ntx-builder/src/chain_state.rs +++ b/bin/ntx-builder/src/chain_state.rs @@ -1,7 +1,7 @@ -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use miden_node_tracing::debug; -use miden_protocol::block::{BlockHeader, BlockNumber}; +use miden_protocol::block::BlockHeader; use miden_protocol::crypto::merkle::mmr::PartialMmr; use miden_protocol::transaction::PartialBlockchain; @@ -10,8 +10,9 @@ use crate::LOG_TARGET; // CHAIN STATE // ================================================================================================ -/// Contains information about the chain that is relevant to the [`NetworkTransactionBuilder`] and -/// all account actors managed by the [`Coordinator`]. +/// Contains information about the chain that is relevant to the +/// [`NetworkTransactionBuilder`](crate::NetworkTransactionBuilder). +/// /// /// The chain MMR stored here contains: /// - The MMR peaks. @@ -82,35 +83,3 @@ impl ChainState { Arc::make_mut(&mut self.chain_mmr).prune_to(..pruned_block_height.into()); } } - -/// A thread-safe wrapper around [`ChainState`] that can be shared across multiple actors. -/// -/// The API guarantees that the lock cannot be held across await points. -pub struct SharedChainState(RwLock); - -impl SharedChainState { - pub fn new(chain_tip_header: BlockHeader, chain_mmr: PartialMmr) -> Self { - Self(RwLock::new(ChainState::new(chain_tip_header, chain_mmr))) - } - - pub(crate) fn chain_tip_block_number(&self) -> BlockNumber { - self.0.read().expect("chain state lock poisoned").chain_tip_header.block_num() - } - - /// Returns a clone of the current partial chain MMR. Cheap enough for per-block persistence - /// since the MMR is bounded by `max_block_count` headers. - pub(crate) fn current_mmr(&self) -> PartialMmr { - self.0.read().expect("chain state lock poisoned").current_mmr() - } - - pub(crate) fn update_chain_tip(&self, tip: BlockHeader, max_block_count: usize) { - self.0 - .write() - .expect("chain state lock poisoned") - .update_chain_tip(tip, max_block_count); - } - - pub(crate) fn get_cloned(&self) -> ChainState { - self.0.read().expect("chain state lock poisoned").clone() - } -} diff --git a/bin/ntx-builder/src/commands/mod.rs b/bin/ntx-builder/src/commands/mod.rs index ce1bc8e84f..1e2a2ad746 100644 --- a/bin/ntx-builder/src/commands/mod.rs +++ b/bin/ntx-builder/src/commands/mod.rs @@ -25,11 +25,12 @@ const ENV_RPC_AUTH_HEADER_VALUE: &str = "MIDEN_NODE_NTX_BUILDER_RPC_AUTH_HEADER_ const ENV_TX_PROVER_URL: &str = "MIDEN_NODE_NTX_BUILDER_NTX_PROVER_URL"; const ENV_TX_PROVER_TIMEOUT: &str = "MIDEN_NODE_NTX_BUILDER_NTX_PROVER_TIMEOUT"; const ENV_SCRIPT_CACHE_SIZE: &str = "MIDEN_NODE_NTX_BUILDER_SCRIPT_CACHE_SIZE"; +const ENV_MAX_CONCURRENT_TXS: &str = "MIDEN_NODE_NTX_BUILDER_MAX_CONCURRENT_TXS"; const ENV_MAX_CYCLES: &str = "MIDEN_NODE_NTX_BUILDER_MAX_CYCLES"; const ENV_TX_EXPIRATION_DELTA: &str = "MIDEN_NODE_NTX_BUILDER_TX_EXPIRATION_DELTA"; const ENV_SQLITE_CONNECTION_POOL_SIZE: &str = "MIDEN_NODE_NTX_BUILDER_SQLITE_CONNECTION_POOL_SIZE"; -const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_mins(5); +const DEFAULT_MAX_CONCURRENT_TXS: usize = 16; const DEFAULT_GRPC_TIMEOUT: Duration = Duration::from_secs(10); const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(10); const DEFAULT_TX_PROVER_TIMEOUT: Duration = Duration::from_secs(10); @@ -104,23 +105,14 @@ pub enum NtxBuilderCommand { )] script_cache_size: NonZeroUsize, - /// Duration after which an idle network account will deactivate. - /// - /// An account is considered idle once it has no viable notes to consume. - /// A deactivated account will reactivate if targeted with new notes. + /// Maximum number of network transactions computed concurrently. #[arg( - long = "idle-timeout", - default_value = &duration_to_human_readable_string(DEFAULT_IDLE_TIMEOUT), - value_parser = humantime::parse_duration, - value_name = "DURATION" + long = "max-concurrent-txs", + env = ENV_MAX_CONCURRENT_TXS, + default_value_t = DEFAULT_MAX_CONCURRENT_TXS, + value_name = "NUM" )] - idle_timeout: Duration, - - /// Maximum number of crashes before an account deactivated. - /// - /// Once this limit is reached, no new transactions will be created for this account. - #[arg(long = "max-account-crashes", default_value_t = 10, value_name = "NUM")] - max_account_crashes: usize, + max_concurrent_txs: usize, /// Maximum number of VM execution cycles allowed for a single network transaction. /// @@ -253,8 +245,7 @@ impl NtxBuilderCommand { tx_prover_url, tx_prover_timeout, script_cache_size, - idle_timeout, - max_account_crashes, + max_concurrent_txs, max_tx_cycles, tx_expiration_delta, sqlite_connection_pool_size, @@ -277,7 +268,7 @@ impl NtxBuilderCommand { tx_prover.endpoint = format_endpoint(&tx_prover_url), tx_prover.timeout = humantime::Duration::from(tx_prover_timeout).to_string(), rpc.authentication.configured = rpc_auth_header_value.is_some(), - ntx_builder.idle_timeout = humantime::Duration::from(idle_timeout).to_string(), + ntx_builder.max_concurrent_txs = max_concurrent_txs, ntx_builder.max_cycles = max_tx_cycles, ntx_builder.tx_expiration_delta = tx_expiration_delta.get(), db.sqlite.connection_pool_size = sqlite_connection_pool_size.get() @@ -295,8 +286,7 @@ impl NtxBuilderCommand { .with_rpc_timeout(rpc_timeout) .with_tx_prover_timeout(tx_prover_timeout) .with_script_cache_size(script_cache_size) - .with_idle_timeout(idle_timeout) - .with_max_account_crashes(max_account_crashes) + .with_max_concurrent_txs(max_concurrent_txs) .with_max_cycles(max_tx_cycles) .with_tx_expiration_delta(tx_expiration_delta) .with_sqlite_connection_pool_size(sqlite_connection_pool_size); diff --git a/bin/ntx-builder/src/committed_block.rs b/bin/ntx-builder/src/committed_block.rs index 5e36f93bd0..cd01a2a1d6 100644 --- a/bin/ntx-builder/src/committed_block.rs +++ b/bin/ntx-builder/src/committed_block.rs @@ -6,13 +6,12 @@ use miden_protocol::note::Nullifier; use miden_protocol::transaction::{OutputNote, TransactionId}; use miden_standards::note::AccountTargetNetworkNote; -use crate::db::queries::account_effect::NetworkAccountEffect; use crate::sponsorship::SponsorshipNote; /// Network-relevant state extracted from a committed [`SignedBlock`]. /// -/// Produced once per committed block on the ntx-builder side. Downstream code (DB layer, -/// coordinator) applies the contained effects to local state. +/// Produced once per committed block on the ntx-builder side. The DB layer applies the contained +/// effects to local state, and the scheduler reads them to resolve its in-flight transactions. #[derive(Debug, Clone)] pub struct CommittedBlockEffects { pub header: BlockHeader, @@ -24,7 +23,7 @@ pub struct CommittedBlockEffects { pub network_account_updates: Vec<(AccountId, AccountUpdateDetails)>, /// Transaction id paired with the account it updated, for every transaction in the block. /// `apply_committed_block` uses this to record the latest landed transaction per network - /// account so actors can confirm their own submitted transaction landed. + /// account, and the scheduler uses it to confirm that its own submission landed. pub account_transactions: Vec<(AccountId, TransactionId)>, } @@ -91,26 +90,12 @@ impl CommittedBlockEffects { } } - /// Returns the ids of the network accounts created by this block. - /// - /// The coordinator uses this to release actor spawns that were deferred until the account's - /// creation transaction committed. - pub fn created_network_accounts(&self) -> impl Iterator + '_ { - self.network_account_updates.iter().filter_map(|(account_id, details)| { - matches!( - NetworkAccountEffect::from_protocol(details), - Some(NetworkAccountEffect::Created(_)) - ) - .then_some(*account_id) - }) - } - /// The latest transaction committed against each account in this block. /// /// `account_transactions` is in block order, so collecting into a map keeps the last /// transaction per account. Both `apply_committed_block` (to persist `accounts.last_tx_id`) and - /// the coordinator (to populate each [`AccountView`](crate::coordinator)'s `last_committed_tx`) - /// derive landing state from this single definition, so the two never disagree. + /// the scheduler (to detect that a submitted transaction landed) derive landing state from this + /// single definition, so the two never disagree. pub fn latest_tx_per_account(&self) -> HashMap { self.account_transactions.iter().copied().collect() } diff --git a/bin/ntx-builder/src/coordinator.rs b/bin/ntx-builder/src/coordinator.rs deleted file mode 100644 index 65396b854f..0000000000 --- a/bin/ntx-builder/src/coordinator.rs +++ /dev/null @@ -1,638 +0,0 @@ -use std::collections::{HashMap, HashSet}; -use std::sync::Arc; - -use anyhow::Context; -use miden_node_tracing::{debug, error, info, miden_instrument, warn}; -use miden_node_utils::shutdown::CancellationToken; -use miden_protocol::account::AccountId; -use miden_protocol::block::BlockNumber; -use miden_protocol::transaction::TransactionId; -use miden_standards::note::AccountTargetNetworkNote; -use tokio::sync::{Semaphore, watch}; -use tokio::task::JoinSet; - -use crate::LOG_TARGET; -#[cfg(test)] -use crate::actor::ActorRequest; -use crate::actor::{AccountActor, AccountActorContext}; -use crate::committed_block::CommittedBlockEffects; -#[cfg(test)] -use crate::db::NtxDbWriter; - -// ACCOUNT VIEW -// ================================================================================================ - -/// Per-account state the coordinator pushes to an actor on every committed block. -/// -/// Every field is cumulative, so an actor that wakes after several blocks reads the latest view and -/// answers entirely in memory: "did my submission land" (`last_committed_tx`), "has it expired" -/// (`chain_tip` vs the submission block), and "is there new work" (`notes_seen` vs a local cursor). -/// The view is intentionally bounded: a single latest-tx slot and a monotone counter, never a -/// growing list. -#[derive(Clone, Debug)] -pub(crate) struct AccountView { - /// Chain tip as of the latest committed block. Advances every block, so the view always changes - /// and every actor wakes (cheaply, in memory) once per block. - pub chain_tip: BlockNumber, - /// Latest transaction committed against this account, mirroring `accounts.last_tx_id`. An actor - /// waiting on its submission compares this against its own transaction id to confirm landing. - pub last_committed_tx: Option, - /// Monotone count of network notes seen targeting this account since the actor was spawned. A - /// local cursor on the actor side answers "is there new work" without a DB query. - pub notes_seen: u64, -} - -// ACTOR HANDLE -// ================================================================================================ - -/// Handle to an account actor spawned by the coordinator. -struct ActorHandle { - /// Sender half of the per-account [`AccountView`] watch channel. The coordinator updates the - /// view on every committed block; the actor awaits changes on its receiver and re-evaluates its - /// state from the pushed data rather than querying the DB. - view_tx: watch::Sender, -} - -impl ActorHandle { - fn new(view_tx: watch::Sender) -> Self { - Self { view_tx } - } -} - -// COORDINATOR -// ================================================================================================ - -/// Lifecycle owner for [`AccountActor`] instances driven by committed blocks. -/// -/// The coordinator owns the actor-side context (gRPC clients, shared chain state, script cache, -/// per-actor config), the actor task join set, and a registry mapping each network account to a -/// notify handle. The builder calls into the coordinator at two moments: -/// -/// 1. At the catch-up boundary, to spawn one actor per account returned by -/// `Db::accounts_with_pending_notes()`. -/// 2. On every committed block in steady state, via [`Coordinator::handle_committed_block`], which -/// spawns missing actors for accounts that just received new network notes and pushes a fresh -/// [`AccountView`] to every active actor so it can re-evaluate its state in memory. -/// -/// Actors only operate on committed account state, so spawning is restricted to accounts whose -/// creation has been committed: a spawn requested for a not-yet-committed account is deferred -/// until the block carrying the account's creation arrives. -/// -/// State changes are pushed through a per-account [`watch`] channel: intermediate views are -/// coalesced (an actor busy for several blocks only ever sees the latest). Actors that crash -/// repeatedly are deactivated after `max_account_crashes` failures. -pub struct Coordinator { - /// Mapping of network account IDs to their view-channel handles. - actor_registry: HashMap, - - /// Join set tracking each spawned actor task; used to detect intentional shutdowns vs. crashes. - actor_join_set: JoinSet<(AccountId, anyhow::Result<()>)>, - - /// Shared transaction-execution semaphore handed to each spawned actor. - semaphore: Arc, - - /// Shared resources needed to spawn an actor. Stored on the coordinator so spawns at runtime - /// don't need the builder to plumb context through every call site. - actor_context: AccountActorContext, - - /// Tracks the number of crashes per account actor. - /// - /// When an actor shuts down due to a DB error, its crash count is incremented. Once - /// the count reaches `max_account_crashes`, the account is deactivated and no new actor - /// will be spawned for it. - crash_counts: HashMap, - - /// Maximum number of crashes an account actor is allowed before being deactivated. - max_account_crashes: usize, - - /// Accounts targeted by network notes whose creation transaction has not been committed yet. - /// - /// Their actor spawn is deferred until a committed block carries the account's creation, at - /// which point [`Coordinator::handle_committed_block`] promotes them to a real actor. - pending_spawns: HashSet, - - /// Cancellation signal shared by all actors spawned by this coordinator. - shutdown: CancellationToken, -} - -impl Coordinator { - /// Creates a new coordinator with the specified transaction concurrency limit and the per- - /// account crash threshold. - pub fn new( - max_inflight_transactions: usize, - max_account_crashes: usize, - actor_context: AccountActorContext, - shutdown: CancellationToken, - ) -> Self { - Self { - actor_registry: HashMap::new(), - actor_join_set: JoinSet::new(), - semaphore: Arc::new(Semaphore::new(max_inflight_transactions)), - actor_context, - crash_counts: HashMap::new(), - max_account_crashes, - pending_spawns: HashSet::new(), - shutdown, - } - } - - /// Spawns a new actor to manage the state of the provided network account. - /// - /// This method creates a new [`AccountActor`] instance for the specified account origin - /// and adds it to the coordinator's management system. The actor will be responsible for - /// processing transactions and managing state for the network account. - #[miden_instrument( - name = "ntx.builder.spawn_actor", - fields(account.id = account_id), - )] - pub fn spawn_actor(&mut self, account_id: AccountId) { - if let Some(&count) = self.crash_counts.get(&account_id) - && count >= self.max_account_crashes - { - warn!( - target: LOG_TARGET, - "Account deactivated due to repeated crashes, skipping actor spawn", - account.id = account_id, - account.crashes.count = count - ); - return; - } - - if self.actor_registry.contains_key(&account_id) { - error!( - anyhow::anyhow!("account actor already exists"), - target: LOG_TARGET, - "Account actor already exists", - account.id = account_id - ); - return; - } - - let initial_view = AccountView { - chain_tip: self.actor_context.state.chain.chain_tip_block_number(), - last_committed_tx: None, - notes_seen: 0, - }; - let (view_tx, view_rx) = watch::channel(initial_view); - let actor = AccountActor::new(account_id, &self.actor_context); - let handle = ActorHandle::new(view_tx); - - let semaphore = self.semaphore.clone(); - let shutdown = self.shutdown.clone(); - self.actor_join_set.spawn(Box::pin(async move { - (account_id, actor.run(semaphore, view_rx, shutdown).await) - })); - - self.actor_registry.insert(account_id, handle); - debug!( - target: LOG_TARGET, - "Created actor for account", - account.id = account_id - ); - } - - /// Spawns an actor for the given account if its committed state exists in the DB; otherwise - /// defers the spawn until the block carrying the account's creation arrives. - /// - /// Actors only operate on committed account state, so spawning earlier would only produce an - /// actor idling for the creation transaction to commit. - pub async fn spawn_actor_when_committed( - &mut self, - account_id: AccountId, - ) -> anyhow::Result<()> { - if self.actor_registry.contains_key(&account_id) { - return Ok(()); - } - - let committed = self - .actor_context - .state - .db - .account_exists(account_id) - .await - .context("failed to check for committed account state")?; - - if committed { - self.spawn_actor(account_id); - } else { - info!( - target: LOG_TARGET, - "deferring actor spawn until the account's creation is committed", - account.id = account_id - ); - self.pending_spawns.insert(account_id); - } - Ok(()) - } - - /// Reacts to a committed block: spawns actors for any newly-targeted network accounts whose - /// committed state exists (deferring the rest until their creation commits), releases deferred - /// spawns for accounts created by this block, and pushes a fresh [`AccountView`] to every - /// active actor so it can re-evaluate its state in memory. - /// - /// `sponsored_accounts` names the accounts whose pending feature notes gained a sponsorship in - /// this block (one entry per sponsorship, resolved by `apply_committed_block`). They are woken - /// exactly like accounts targeted by a new network note: a feature note that selection skipped - /// for lacking a sponsorship becomes viable when its sponsorship arrives. - pub async fn handle_committed_block( - &mut self, - effects: &CommittedBlockEffects, - sponsored_accounts: &[AccountId], - ) -> anyhow::Result<()> { - // Accounts created by this block release any spawn deferred on their creation. - for account_id in effects.created_network_accounts() { - if self.pending_spawns.remove(&account_id) { - self.spawn_actor(account_id); - } - } - - let mut targeted: HashSet = effects - .network_notes - .iter() - .map(AccountTargetNetworkNote::target_account_id) - .collect(); - targeted.extend(sponsored_accounts.iter().copied()); - for account_id in &targeted { - self.spawn_actor_when_committed(*account_id).await?; - } - - // Push the block's effects to every active actor. The latest transaction per account is the - // same map `apply_committed_block` uses for `accounts.last_tx_id`, so the pushed - // `last_committed_tx` agrees with the persisted state; the per-account note counts feed the - // `notes_seen` work counter. A sponsorship for a pending feature note counts as work just - // like a new note. - let chain_tip = effects.header.block_num(); - let latest_tx = effects.latest_tx_per_account(); - let mut new_notes: HashMap = HashMap::new(); - for note in &effects.network_notes { - *new_notes.entry(note.target_account_id()).or_default() += 1; - } - for account_id in sponsored_accounts { - *new_notes.entry(*account_id).or_default() += 1; - } - - for (account_id, handle) in &self.actor_registry { - let committed_tx = latest_tx.get(account_id).copied(); - let notes = new_notes.get(account_id).copied().unwrap_or(0); - handle.view_tx.send_modify(|view| { - view.chain_tip = chain_tip; - if let Some(tx) = committed_tx { - view.last_committed_tx = Some(tx); - } - view.notes_seen += notes; - }); - } - Ok(()) - } - - /// Waits for the next actor to complete and handles the outcome. - /// - /// Returns `Some(account_id)` if an actor should be respawned (because work reappeared for the - /// account between its last view observation and its idle shutdown), or `None` otherwise. If no - /// actors are currently running, this method waits indefinitely until new actors are spawned. - pub async fn next(&mut self) -> anyhow::Result> { - let actor_result = self.actor_join_set.join_next().await; - match actor_result { - Some(Ok((account_id, Ok(())))) => { - // Actor shut down intentionally on idle timeout, which only happens when it had no - // pending notes. Reap it, then respawn if a block committed between its last - // observation and its exit added (or re-armed) work for the account: that view - // update went to a now-dropped receiver and would otherwise wait for the next - // block. - self.actor_registry.remove(&account_id); - let should_respawn = self.account_has_pending_notes(account_id).await?; - Ok(should_respawn.then_some(account_id)) - }, - Some(Ok((account_id, Err(err)))) => { - let count = self.crash_counts.entry(account_id).or_insert(0); - *count += 1; - error!( - &err, - target: LOG_TARGET, - "Account actor crashed", - account.id = account_id - ); - self.actor_registry.remove(&account_id); - Ok(None) - }, - Some(Err(err)) => { - error!(&err, target: LOG_TARGET, "Actor task failed"); - Ok(None) - }, - None => { - // There are no actors to wait for. Wait indefinitely until actors are spawned. - std::future::pending().await - }, - } - } - - /// Returns `true` if the account has any pending notes: eligible now, or awaiting a backoff or - /// execution-hint window. Used to decide whether to respawn an actor that just idle-timed-out. - async fn account_has_pending_notes(&self, account_id: AccountId) -> anyhow::Result { - self.actor_context - .state - .db - .account_has_pending_notes(account_id, self.actor_context.config.max_note_attempts) - .await - .context("failed to check pending notes when reaping an idle actor") - } - - /// Waits for all currently running actors to exit after cancellation. - pub async fn shutdown(&mut self) -> anyhow::Result<()> { - while let Some(result) = self.actor_join_set.join_next().await { - match result { - Ok((_account_id, Ok(()))) => {}, - Ok((account_id, Err(err))) => { - return Err(err).with_context(|| { - format!("account actor {account_id} failed during shutdown") - }); - }, - Err(err) if err.is_cancelled() => {}, - Err(err) => return Err(err).context("account actor failed to join"), - } - } - self.actor_registry.clear(); - Ok(()) - } -} - -#[cfg(test)] -impl Coordinator { - /// Creates a coordinator with default settings backed by a temp DB. Returns the coordinator, - /// the temp dir holding the DB file, and the actor request receiver (drop it to discard, or - /// drive it from the test to inspect actor requests). - pub async fn test() - -> (Self, NtxDbWriter, tempfile::TempDir, tokio::sync::mpsc::Receiver) { - let (db, dir) = crate::db::test_setup().await; - let (tx, rx) = tokio::sync::mpsc::channel(8); - let mut actor_context = AccountActorContext::test(&db); - actor_context.request_tx = tx; - (Self::new(4, 10, actor_context, CancellationToken::new()), db, dir, rx) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::test_utils::*; - - /// Registers a dummy actor handle (no real actor task) in the coordinator's registry and - /// returns the view receiver so the test can observe what the coordinator pushes. - fn register_dummy_actor( - coordinator: &mut Coordinator, - account_id: AccountId, - ) -> watch::Receiver { - let (view_tx, view_rx) = watch::channel(AccountView { - chain_tip: BlockNumber::GENESIS, - last_committed_tx: None, - notes_seen: 0, - }); - coordinator.actor_registry.insert(account_id, ActorHandle::new(view_tx)); - view_rx - } - - /// Seeds a committed row for `account_id` so the coordinator's spawn check sees the account. - async fn seed_committed_account(db: &NtxDbWriter, account_id: AccountId) { - db.upsert_account_for_test(account_id, mock_account(account_id), mock_transaction_id(0)) - .await - .unwrap(); - } - - #[tokio::test] - async fn handle_committed_block_spawns_for_committed_note_target() { - let (mut coordinator, db, _dir, _rx) = Coordinator::test().await; - - let target_id = mock_network_account_id(); - seed_committed_account(&db, target_id).await; - - let note = mock_single_target_note(target_id, 10); - let effects = CommittedBlockEffects { - header: mock_block_header(1_u32.into()), - network_notes: vec![note], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![], - account_transactions: vec![], - }; - - coordinator.handle_committed_block(&effects, &[]).await.unwrap(); - - assert!( - coordinator.actor_registry.contains_key(&target_id), - "a committed account targeted by a note should get a fresh actor", - ); - } - - #[tokio::test] - async fn handle_committed_block_defers_spawn_until_account_creation_commits() { - let (mut coordinator, db, _dir, _rx) = Coordinator::test().await; - - let (account, details) = mock_network_account_update(); - let account_id = account.id(); - - // A note targets the account before its creation transaction has been committed. - let note = mock_single_target_note(account_id, 10); - let effects = CommittedBlockEffects { - header: mock_block_header(1_u32.into()), - network_notes: vec![note], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![], - account_transactions: vec![], - }; - coordinator.handle_committed_block(&effects, &[]).await.unwrap(); - - assert!( - !coordinator.actor_registry.contains_key(&account_id), - "an account without committed state must not get an actor", - ); - assert!( - coordinator.pending_spawns.contains(&account_id), - "the spawn must be deferred until the account's creation commits", - ); - - // The creation commits in a later block; the builder persists the block's effects to the DB - // before handing them to the coordinator. - db.upsert_account_for_test(account_id, account, mock_transaction_id(0)) - .await - .unwrap(); - let effects = CommittedBlockEffects { - header: mock_block_header(2_u32.into()), - network_notes: vec![], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![(account_id, details)], - account_transactions: vec![], - }; - coordinator.handle_committed_block(&effects, &[]).await.unwrap(); - - assert!( - coordinator.actor_registry.contains_key(&account_id), - "the block committing the account's creation must release the deferred spawn", - ); - assert!( - coordinator.pending_spawns.is_empty(), - "a released spawn must leave the pending set", - ); - } - - #[tokio::test] - async fn handle_committed_block_does_not_spawn_for_account_update_only() { - let (mut coordinator, _db, _dir, _rx) = Coordinator::test().await; - - let updated_id = mock_network_account_id(); - let effects = CommittedBlockEffects { - header: mock_block_header(1_u32.into()), - network_notes: vec![], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![( - updated_id, - miden_protocol::account::AccountUpdateDetails::Private, - )], - account_transactions: vec![], - }; - - coordinator.handle_committed_block(&effects, &[]).await.unwrap(); - - assert!( - !coordinator.actor_registry.contains_key(&updated_id), - "an account update without a new note should not trigger an actor spawn", - ); - } - - #[tokio::test] - async fn spawn_actor_skips_deactivated_account() { - let (mut coordinator, _db, _dir, _rx) = Coordinator::test().await; - - let account_id = mock_network_account_id(); - coordinator.crash_counts.insert(account_id, coordinator.max_account_crashes); - - coordinator.spawn_actor(account_id); - - assert!( - !coordinator.actor_registry.contains_key(&account_id), - "deactivated account should not have an actor in the registry", - ); - } - - #[tokio::test] - async fn spawn_actor_allows_below_threshold() { - let (mut coordinator, _db, _dir, _rx) = Coordinator::test().await; - - let account_id = mock_network_account_id(); - coordinator - .crash_counts - .insert(account_id, coordinator.max_account_crashes.saturating_sub(1)); - - coordinator.spawn_actor(account_id); - - assert!( - coordinator.actor_registry.contains_key(&account_id), - "account below crash threshold should have an actor in the registry", - ); - } - - #[tokio::test] - async fn handle_committed_block_pushes_view_to_existing_actors() { - let (mut coordinator, db, _dir, _rx) = Coordinator::test().await; - - let bystander = mock_network_account_id(); - let mut bystander_rx = register_dummy_actor(&mut coordinator, bystander); - // Mark the initial view as seen so the post-block update is observable as a change. - let _ = bystander_rx.borrow_and_update(); - - let target = mock_network_account_id_seeded(42); - seed_committed_account(&db, target).await; - let note = mock_single_target_note(target, 10); - let effects = CommittedBlockEffects { - header: mock_block_header(1_u32.into()), - network_notes: vec![note], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![], - account_transactions: vec![], - }; - - coordinator.handle_committed_block(&effects, &[]).await.unwrap(); - - assert!( - bystander_rx.has_changed().unwrap(), - "every registered actor should receive a view update on a committed block", - ); - let view = bystander_rx.borrow_and_update(); - assert_eq!(view.chain_tip, 1_u32.into(), "the view carries the new chain tip"); - assert_eq!(view.notes_seen, 0, "a bystander targeted by no note sees no new work"); - drop(view); - - assert!( - coordinator.actor_registry.contains_key(&target), - "freshly-targeted account should get an actor", - ); - } - - /// The pushed view carries the account's latest committed transaction (for landing detection) - /// and a bumped note counter (for the work signal). - #[tokio::test] - async fn handle_committed_block_view_carries_landing_and_new_notes() { - let (mut coordinator, _db, _dir, _rx) = Coordinator::test().await; - - let account_id = mock_network_account_id(); - // A dummy handle for the targeted account so the coordinator updates it in place instead of - // spawning a real actor (which would own the receiver and hide it from the test). - let mut rx = register_dummy_actor(&mut coordinator, account_id); - let _ = rx.borrow_and_update(); - - let tx_id = mock_transaction_id(5); - let note = mock_single_target_note(account_id, 10); - let effects = CommittedBlockEffects { - header: mock_block_header(3_u32.into()), - network_notes: vec![note], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![], - account_transactions: vec![(account_id, tx_id)], - }; - - coordinator.handle_committed_block(&effects, &[]).await.unwrap(); - - let view = rx.borrow_and_update(); - assert_eq!(view.chain_tip, 3_u32.into()); - assert_eq!( - view.last_committed_tx, - Some(tx_id), - "the account's latest committed tx is pushed for in-memory landing detection", - ); - assert_eq!(view.notes_seen, 1, "one note targeting the account bumps the work counter"); - } - - /// A sponsorship arriving for an account's pending feature note counts as new work: the feature - /// note may have been skipped for lacking a sponsorship, and this wakes the actor for a - /// re-selection. - #[tokio::test] - async fn handle_committed_block_sponsorship_wakeup_bumps_notes_seen() { - let (mut coordinator, _db, _dir, _rx) = Coordinator::test().await; - - let account_id = mock_network_account_id(); - let mut rx = register_dummy_actor(&mut coordinator, account_id); - let _ = rx.borrow_and_update(); - - // The block carries no network note for the account; only a sponsorship resolved to it. - let effects = CommittedBlockEffects { - header: mock_block_header(1_u32.into()), - network_notes: vec![], - sponsorship_notes: vec![], - nullifiers: vec![], - network_account_updates: vec![], - account_transactions: vec![], - }; - - coordinator.handle_committed_block(&effects, &[account_id]).await.unwrap(); - - let view = rx.borrow_and_update(); - assert_eq!( - view.notes_seen, 1, - "a sponsorship for a pending feature note bumps the work counter", - ); - } -} diff --git a/bin/ntx-builder/src/db/migrations/004_note_eligibility.sql b/bin/ntx-builder/src/db/migrations/004_note_eligibility.sql index f39f24d448..547a053ef1 100644 --- a/bin/ntx-builder/src/db/migrations/004_note_eligibility.sql +++ b/bin/ntx-builder/src/db/migrations/004_note_eligibility.sql @@ -1,4 +1,5 @@ --- Materializes note eligibility so the scheduler can ask for the ready accounts. +-- Materializes note eligibility so the scheduler can ask for the ready accounts with one indexed +-- query. ALTER TABLE notes ADD COLUMN next_eligible_block BIGINT NOT NULL DEFAULT 0 CHECK (next_eligible_block BETWEEN 0 AND 0xFFFFFFFF); diff --git a/bin/ntx-builder/src/db/mod.rs b/bin/ntx-builder/src/db/mod.rs index b55a395da5..50a15368e6 100644 --- a/bin/ntx-builder/src/db/mod.rs +++ b/bin/ntx-builder/src/db/mod.rs @@ -86,7 +86,7 @@ miden_node_db::impl_blob_codec!(GenesisValidatorKeys); /// Read-only handle to the ntx-builder database. /// /// Wraps the framework [`DbReader`] and exposes every read query as a method. Cloneable, and handed -/// to read-only components (the gRPC server, the coordinator, and actors); it has no write methods, +/// to read-only components (the gRPC server and the transaction attempts); it has no write methods, /// so those components cannot mutate the database. #[derive(Clone)] pub(crate) struct NtxDbReader { @@ -118,17 +118,18 @@ impl NtxDbReader { .await } - /// Returns `true` if the account has any pending (unconsumed, within attempt budget) note. Used - /// by the coordinator to decide whether to respawn an actor that just idle-timed-out, without - /// loading or deserializing the notes themselves. - pub(crate) async fn account_has_pending_notes( + /// Returns up to `limit` accounts that are ready for a transaction attempt, longest-waiting + /// first. + pub(crate) async fn ready_accounts( &self, - account_id: AccountId, - max_attempts: usize, - ) -> Result { + max_note_attempts: usize, + block_num: BlockNumber, + busy: Vec, + limit: usize, + ) -> Result, DatabaseError> { self.reader - .read("account_has_pending_notes", move |tx| { - queries::account_has_pending_notes(tx, account_id, max_attempts) + .read("ready_accounts", move |tx| { + queries::ready_accounts(tx, max_note_attempts, block_num, &busy, limit) }) .await } @@ -155,29 +156,9 @@ impl NtxDbReader { self.reader.read("select_chain_state", queries::select_chain_state).await } - pub(crate) async fn account_exists( - &self, - account_id: AccountId, - ) -> Result { - self.reader - .read("account_exists", move |tx| db::queries::account_exists(tx, account_id)) - .await - } - - pub(crate) async fn accounts_with_pending_notes( - &self, - max_note_attempts: usize, - ) -> Result, DatabaseError> { - self.reader - .read("accounts_with_pending_notes", move |tx| { - queries::accounts_with_pending_notes(tx, max_note_attempts) - }) - .await - } - - /// The committed-transaction landing check reads `last_committed_tx` from the `AccountView` the - /// coordinator pushes, so this read accessor is only used by tests to verify that - /// `upsert_account` persists `accounts.last_tx_id` correctly. + /// The scheduler detects a landed transaction from the block's own transaction list, so this + /// read accessor is only used by tests to verify that `upsert_account` persists + /// `accounts.last_tx_id` correctly. #[cfg(test)] pub(crate) async fn account_last_tx( &self, @@ -258,14 +239,12 @@ impl NtxDbWriter { .await } - /// Applies a committed block's effects and returns the accounts whose pending feature notes - /// gained a sponsorship in this block (one entry per sponsorship), so the coordinator can wake - /// their actors. + /// Applies a committed block's effects in a single write transaction. pub(crate) async fn apply_committed_block( &self, effects: CommittedBlockEffects, chain_mmr: PartialMmr, - ) -> Result, DatabaseError> { + ) -> Result<(), DatabaseError> { self.writer .write("apply_committed_block", move |tx| { queries::apply_committed_block(tx, &effects, &chain_mmr) @@ -302,6 +281,19 @@ impl NtxDbWriter { .await } + /// Stores the corrected eligibility block of notes whose exact hint and backoff check disagrees + /// with the stored one. + pub(crate) async fn update_note_eligibility( + &self, + eligibility: Vec<(Nullifier, BlockNumber)>, + ) -> Result<(), DatabaseError> { + self.writer + .write("update_note_eligibility", move |tx| { + queries::update_note_eligibility(tx, &eligibility) + }) + .await + } + pub(crate) async fn insert_note_scripts( &self, script_root: Word, @@ -476,8 +468,8 @@ impl NtxDbReader { /// still reach the database exclusively through the wrapper. #[cfg(test)] impl NtxDbWriter { - /// Seeds a committed account row (and its `last_tx_id`) for tests that exercise the actor's - /// landing detection without driving a full committed block. + /// Seeds a committed account row (and its `last_tx_id`) for tests that need a committed account + /// without driving a full committed block. pub(crate) async fn upsert_account_for_test( &self, account_id: AccountId, diff --git a/bin/ntx-builder/src/db/queries/account_effect.rs b/bin/ntx-builder/src/db/queries/account_effect.rs index 4ef4af901e..ec59e15ac4 100644 --- a/bin/ntx-builder/src/db/queries/account_effect.rs +++ b/bin/ntx-builder/src/db/queries/account_effect.rs @@ -25,8 +25,8 @@ impl NetworkAccountEffect { .map(|na| NetworkAccountEffect::Created(na.into_account())) }, AccountUpdateDetails::Public(update) => { - // Partial updates carry no storage we can inspect here. Forward them as updates and - // let the coordinator's actor registry filter to known network accounts. + // Partial updates carry no storage we can inspect here. Forward them as updates; + // `apply_committed_block` drops the ones whose account is not tracked locally. Some(NetworkAccountEffect::Updated(update.clone())) }, } diff --git a/bin/ntx-builder/src/db/queries/account_exists/account_exists.sql b/bin/ntx-builder/src/db/queries/account_exists/account_exists.sql deleted file mode 100644 index ad724b31c9..0000000000 --- a/bin/ntx-builder/src/db/queries/account_exists/account_exists.sql +++ /dev/null @@ -1,2 +0,0 @@ --- Returns whether a committed state for the given account is tracked locally. -SELECT EXISTS (SELECT 1 FROM accounts WHERE account_id = ?1) diff --git a/bin/ntx-builder/src/db/queries/account_exists/mod.rs b/bin/ntx-builder/src/db/queries/account_exists/mod.rs deleted file mode 100644 index 21b7a8ac85..0000000000 --- a/bin/ntx-builder/src/db/queries/account_exists/mod.rs +++ /dev/null @@ -1,16 +0,0 @@ -//! Checks whether a network account is tracked locally. - -use miden_node_db::DatabaseError; -use miden_node_db::sqlite::ReadTx; -use miden_protocol::account::AccountId; - -const SQL: &str = include_str!("account_exists.sql"); - -/// Returns `true` if a committed state for the given account is tracked locally. -pub fn account_exists(tx: &ReadTx<'_>, account_id: AccountId) -> Result { - Ok(tx - .query(SQL, &[&account_id], |row| row.get::(0))? - .into_iter() - .next() - .unwrap_or(false)) -} diff --git a/bin/ntx-builder/src/db/queries/account_has_pending_notes/account_has_pending_notes.sql b/bin/ntx-builder/src/db/queries/account_has_pending_notes/account_has_pending_notes.sql deleted file mode 100644 index d465bdf6a5..0000000000 --- a/bin/ntx-builder/src/db/queries/account_has_pending_notes/account_has_pending_notes.sql +++ /dev/null @@ -1,6 +0,0 @@ --- Returns whether the account has any pending note: unconsumed and within the per-note attempt --- budget. Tests for existence in SQL and deserializes nothing. -SELECT EXISTS ( - SELECT 1 FROM notes - WHERE account_id = ?1 AND committed_at IS NULL AND attempt_count < ?2 -) diff --git a/bin/ntx-builder/src/db/queries/account_has_pending_notes/mod.rs b/bin/ntx-builder/src/db/queries/account_has_pending_notes/mod.rs deleted file mode 100644 index f50a467d5c..0000000000 --- a/bin/ntx-builder/src/db/queries/account_has_pending_notes/mod.rs +++ /dev/null @@ -1,25 +0,0 @@ -//! Cheap existence check for an account's pending notes. - -use miden_node_db::DatabaseError; -use miden_node_db::sqlite::ReadTx; -use miden_protocol::account::AccountId; - -const SQL: &str = include_str!("account_has_pending_notes.sql"); - -/// Returns `true` if the account has any pending note: unconsumed and within the per-note attempt -/// budget. This is the cheap equivalent of "does [`available_notes`](super::available_notes) return -/// a note that is eligible or awaiting a retry window" (every row passing this filter is one or the -/// other), but it tests for existence in SQL and deserializes nothing. The coordinator uses it to -/// decide whether to respawn an actor that just idle-timed-out. -#[expect(clippy::cast_possible_wrap)] -pub fn account_has_pending_notes( - tx: &ReadTx<'_>, - account_id: AccountId, - max_attempts: usize, -) -> Result { - Ok(tx - .query(SQL, &[&account_id, &(max_attempts as i64)], |row| row.get::(0))? - .into_iter() - .next() - .unwrap_or(false)) -} diff --git a/bin/ntx-builder/src/db/queries/accounts_with_pending_notes/accounts_with_pending_notes.sql b/bin/ntx-builder/src/db/queries/accounts_with_pending_notes/accounts_with_pending_notes.sql deleted file mode 100644 index 369d022755..0000000000 --- a/bin/ntx-builder/src/db/queries/accounts_with_pending_notes/accounts_with_pending_notes.sql +++ /dev/null @@ -1,4 +0,0 @@ --- Returns the distinct set of network accounts that currently have at least one pending note --- (unconsumed and within the per-note attempt budget). -SELECT DISTINCT account_id FROM notes -WHERE committed_at IS NULL AND attempt_count < ?1 diff --git a/bin/ntx-builder/src/db/queries/accounts_with_pending_notes/mod.rs b/bin/ntx-builder/src/db/queries/accounts_with_pending_notes/mod.rs deleted file mode 100644 index 7f1e8b9f86..0000000000 --- a/bin/ntx-builder/src/db/queries/accounts_with_pending_notes/mod.rs +++ /dev/null @@ -1,17 +0,0 @@ -//! Returns the network accounts that currently have pending notes. - -use miden_node_db::DatabaseError; -use miden_node_db::sqlite::ReadTx; -use miden_protocol::account::AccountId; - -const SQL: &str = include_str!("accounts_with_pending_notes.sql"); - -/// Returns the distinct set of network accounts that currently have at least one pending note -/// (unconsumed and within the per-note attempt budget). -#[expect(clippy::cast_possible_wrap)] -pub fn accounts_with_pending_notes( - tx: &ReadTx<'_>, - max_attempts: usize, -) -> Result, DatabaseError> { - tx.query(SQL, &[&(max_attempts as i64)], |row| row.get::(0)) -} diff --git a/bin/ntx-builder/src/db/queries/available_notes/available_notes.sql b/bin/ntx-builder/src/db/queries/available_notes/available_notes.sql index b6e9bf8ef6..0eb2f24cf9 100644 --- a/bin/ntx-builder/src/db/queries/available_notes/available_notes.sql +++ b/bin/ntx-builder/src/db/queries/available_notes/available_notes.sql @@ -1,4 +1,7 @@ -- Selects unconsumed notes for the account (a row exists only while a note is unconsumed) whose --- `attempt_count` is below the cap. Execution-hint and backoff filtering are applied in Rust. +-- `attempt_count` is below the cap and whose stored eligibility block has passed. SELECT note_data, attempt_count, last_attempt FROM notes -WHERE account_id = ?1 AND committed_at IS NULL AND attempt_count < ?2 +WHERE account_id = ?1 + AND committed_at IS NULL + AND attempt_count < ?2 + AND next_eligible_block <= ?3 diff --git a/bin/ntx-builder/src/db/queries/available_notes/mod.rs b/bin/ntx-builder/src/db/queries/available_notes/mod.rs index dc1c7bfbdb..240103d211 100644 --- a/bin/ntx-builder/src/db/queries/available_notes/mod.rs +++ b/bin/ntx-builder/src/db/queries/available_notes/mod.rs @@ -4,33 +4,27 @@ use miden_node_db::sqlite::ReadTx; use miden_node_db::{DatabaseError, SqlTypeConvert}; use miden_protocol::account::AccountId; use miden_protocol::block::BlockNumber; -use miden_protocol::note::Note; +use miden_protocol::note::{Note, Nullifier}; use miden_standards::note::AccountTargetNetworkNote; use crate::db::eligibility::{has_backoff_passed, hint_floor, note_recheck_block}; const SQL: &str = include_str!("available_notes.sql"); -/// Notes available for consumption by an account, plus a hint for when to look again. +/// Notes available for consumption by an account, plus the corrections its eligibility filter +/// needs. pub struct AvailableNotes { /// Notes that are eligible for consumption at the queried block. pub eligible: Vec, - /// Earliest block at which a currently-ineligible (but still alive) note becomes eligible, - /// or `None` if the account has no pending notes awaiting backoff or an execution-hint window. + /// Notes whose stored eligibility block has passed but which the exact hint and backoff check + /// still rejects, paired with the block they actually become eligible at. /// - /// Actors use this to avoid re-querying the DB on every block: a `NoViableNotes` actor only - /// re-selects once the chain tip reaches this block (or a new note arrives), and an actor with - /// `None` here has no pending notes at all and may deactivate on idle timeout. - pub next_retry_block: Option, + /// The caller persists these. Without that write the account keeps being selected for a note + /// that cannot be attempted (see [`crate::db::eligibility`]). + pub stale_eligibility: Vec<(Nullifier, BlockNumber)>, } /// Returns notes available for consumption by a given account. -/// -/// Selects unconsumed notes for the account (a row exists only while a note is unconsumed) whose -/// `attempt_count` is below the cap, then applies execution-hint and backoff filtering in Rust. -/// Notes filtered out by backoff or an execution-hint window are still alive and become eligible at -/// a later block; the earliest such block is returned as [`AvailableNotes::next_retry_block`] so the -/// caller can schedule a single re-check instead of polling every block. #[expect(clippy::cast_possible_wrap)] pub fn available_notes( tx: &ReadTx<'_>, @@ -38,12 +32,13 @@ pub fn available_notes( block_num: BlockNumber, max_attempts: usize, ) -> Result { - let rows = tx.query(SQL, &[&account_id, &(max_attempts as i64)], |row| { - Ok((row.get::(0)?, row.get::(1)?, row.get::>(2)?)) - })?; + let rows = + tx.query(SQL, &[&account_id, &(max_attempts as i64), &block_num.to_raw_sql()], |row| { + Ok((row.get::(0)?, row.get::(1)?, row.get::>(2)?)) + })?; let mut eligible = Vec::new(); - let mut next_retry_block: Option = None; + let mut stale_eligibility = Vec::new(); for (note, attempt_count, last_attempt) in rows { #[expect(clippy::cast_sign_loss)] let attempt_count = attempt_count as usize; @@ -66,10 +61,9 @@ pub fn available_notes( backoff_ok, hint_ok, ); - next_retry_block = - Some(next_retry_block.map_or(recheck, |earliest| earliest.min(recheck))); + stale_eligibility.push((note.as_note().nullifier(), recheck)); } } - Ok(AvailableNotes { eligible, next_retry_block }) + Ok(AvailableNotes { eligible, stale_eligibility }) } diff --git a/bin/ntx-builder/src/db/queries/mod.rs b/bin/ntx-builder/src/db/queries/mod.rs index 37d0ec8356..8e7a9df29b 100644 --- a/bin/ntx-builder/src/db/queries/mod.rs +++ b/bin/ntx-builder/src/db/queries/mod.rs @@ -8,7 +8,6 @@ use miden_node_db::DatabaseError; use miden_node_db::sqlite::WriteTx; use miden_protocol::Word; -use miden_protocol::account::AccountId; use miden_protocol::block::BlockNumber; use miden_protocol::crypto::merkle::mmr::PartialMmr; use miden_protocol::transaction::TransactionId; @@ -18,23 +17,14 @@ use crate::db::queries::account_effect::NetworkAccountEffect; pub(crate) mod account_effect; -mod account_exists; -pub use account_exists::account_exists; - -mod account_has_pending_notes; -pub use account_has_pending_notes::account_has_pending_notes; - -// The committed-transaction landing check reads `last_committed_tx` from the `AccountView` the -// coordinator pushes, so this read accessor is only used by tests to verify that `upsert_account` -// persists `accounts.last_tx_id` correctly. +// The scheduler detects a landed transaction from the block's own transaction list, so this read +// accessor is only used by tests to verify that `upsert_account` persists `accounts.last_tx_id` +// correctly. #[cfg(test)] mod account_last_tx; #[cfg(test)] pub use account_last_tx::account_last_tx; -mod accounts_with_pending_notes; -pub use accounts_with_pending_notes::accounts_with_pending_notes; - mod available_notes; pub use available_notes::{AvailableNotes, available_notes}; @@ -71,6 +61,9 @@ pub use mark_sponsorships_consumed::mark_sponsorships_consumed; mod notes_failed; pub use notes_failed::notes_failed; +mod ready_accounts; +pub use ready_accounts::ready_accounts; + mod reset_sponsored_notes; pub use reset_sponsored_notes::reset_sponsored_notes; @@ -83,12 +76,12 @@ pub use select_genesis_commitment::select_genesis_commitment; mod select_genesis_validator_keys; pub use select_genesis_validator_keys::select_genesis_validator_keys; -mod sponsored_accounts; -pub use sponsored_accounts::get_target_account_ids_for_sponsor_notes; - mod sponsorships_for_pending_notes; pub use sponsorships_for_pending_notes::select_sponsorships_for_pending_notes; +mod update_note_eligibility; +pub use update_note_eligibility::update_note_eligibility; + mod update_chain_state_tip; pub use update_chain_state_tip::update_chain_state_tip; @@ -113,26 +106,17 @@ mod tests; /// - Updates the singleton `chain_state` row's tip with the new block header and the /// post-application chain MMR. /// -/// Returns the accounts whose pending feature notes gained a sponsorship in this block (one entry -/// per sponsorship), so the coordinator can wake their actors: a feature note skipped for lacking a -/// sponsorship becomes viable when its sponsorship arrives later. -/// -/// The account upserts apply each block's network-account effects to the local store so an actor's -/// post-expiry reload sees the authoritative committed state. The recorded `accounts.last_tx_id` and -/// the `last_committed_tx` the coordinator pushes to actors both derive from the block's +/// The account upserts apply each block's network-account effects to the local store, so the next +/// attempt for an account reads its authoritative committed state. The recorded +/// `accounts.last_tx_id` and the scheduler's landing check both derive from the block's /// `account_transactions`, so they agree on which transaction last touched each account. pub fn apply_committed_block( tx: &WriteTx<'_>, effects: &CommittedBlockEffects, chain_mmr: &PartialMmr, -) -> Result, DatabaseError> { - // Derive each account's latest transaction from the effects that the coordinator uses. The - // stored `accounts.last_tx_id` and `AccountView::last_committed_tx` values must agree. Each - // non-genesis account update has an originating transaction in the same block. Genesis account - // updates use the zero sentinel because genesis contains no transactions. - // - // Landing detection reads `AccountView::last_committed_tx`. The database column records the - // same committed state. +) -> Result<(), DatabaseError> { + // Each non-genesis account update has an originating transaction in the same block. Genesis + // account updates use the zero sentinel because genesis contains no transactions. let last_tx = effects.latest_tx_per_account(); let is_genesis = effects.header.block_num() == BlockNumber::GENESIS; @@ -174,12 +158,11 @@ pub fn apply_committed_block( mark_notes_consumed(tx, &effects.nullifiers, block_num)?; mark_sponsorships_consumed(tx, &effects.nullifiers, block_num)?; - // Both of these run after the consumption marks so a feature note consumed in this same block - // is neither woken nor made eligible again. - let sponsored = get_target_account_ids_for_sponsor_notes(tx, &effects.sponsorship_notes)?; + // Applied after the consumption marks so a feature note consumed in this same block is not made + // eligible again. reset_sponsored_notes(tx, &effects.sponsorship_notes, block_num)?; update_chain_state_tip(tx, effects.header.block_num(), &effects.header, chain_mmr)?; - Ok(sponsored) + Ok(()) } diff --git a/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs b/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs new file mode 100644 index 0000000000..898a9d58ba --- /dev/null +++ b/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs @@ -0,0 +1,28 @@ +//! Selects the network accounts that are ready for a transaction attempt. + +use miden_node_db::sqlite::{InList, ReadTx}; +use miden_node_db::{DatabaseError, SqlTypeConvert}; +use miden_protocol::account::AccountId; +use miden_protocol::block::BlockNumber; + +const SQL: &str = include_str!("ready_accounts.sql"); + +/// Returns up to `limit` accounts that are ready for a transaction attempt at `block_num`, the +/// longest-waiting one first. +/// +/// `busy` names the accounts to skip: those with a running attempt or an in-flight transaction. +#[expect(clippy::cast_possible_wrap)] +pub fn ready_accounts( + tx: &ReadTx<'_>, + max_attempts: usize, + block_num: BlockNumber, + busy: &[AccountId], + limit: usize, +) -> Result, DatabaseError> { + let busy = InList::from_values(busy.iter().copied()); + tx.query( + SQL, + &[&(max_attempts as i64), &block_num.to_raw_sql(), &busy, &(limit as i64)], + |row| row.get::(0), + ) +} diff --git a/bin/ntx-builder/src/db/queries/ready_accounts/ready_accounts.sql b/bin/ntx-builder/src/db/queries/ready_accounts/ready_accounts.sql new file mode 100644 index 0000000000..f81dbe40ef --- /dev/null +++ b/bin/ntx-builder/src/db/queries/ready_accounts/ready_accounts.sql @@ -0,0 +1,14 @@ +-- Selects the network accounts that are ready for a transaction attempt: accounts whose creation +-- is committed and which have at least one pending note (unconsumed, within the per-note attempt +-- budget, and past its stored eligibility block). + +SELECT n.account_id +FROM notes n +JOIN accounts a ON a.account_id = n.account_id +WHERE n.committed_at IS NULL + AND n.attempt_count < ?1 + AND n.next_eligible_block <= ?2 + AND n.account_id NOT IN (SELECT value FROM rarray(?3)) +GROUP BY n.account_id +ORDER BY MIN(n.next_eligible_block) ASC +LIMIT ?4 diff --git a/bin/ntx-builder/src/db/queries/sponsored_accounts/mod.rs b/bin/ntx-builder/src/db/queries/sponsored_accounts/mod.rs deleted file mode 100644 index b217e86ff6..0000000000 --- a/bin/ntx-builder/src/db/queries/sponsored_accounts/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -//! Resolves the accounts whose pending feature notes just gained a sponsorship. - -use miden_node_db::DatabaseError; -use miden_node_db::sqlite::{InList, WriteTx}; -use miden_protocol::account::AccountId; - -use crate::sponsorship::SponsorshipNote; - -const SQL: &str = include_str!("sponsored_account.sql"); - -/// Returns, for each sponsorship, the account targeted by the pending feature note it is bound to. -/// -/// The result may name the same account several times (once per sponsorship); the coordinator -/// counts every occurrence towards the account's work counter. -pub fn get_target_account_ids_for_sponsor_notes( - tx: &WriteTx<'_>, - sponsorships: &[SponsorshipNote], -) -> Result, DatabaseError> { - let feature_note_ids = - InList::from_values(sponsorships.iter().map(SponsorshipNote::feature_note_id)); - - tx.query(SQL, &[&feature_note_ids], |row| row.get::(0)) -} diff --git a/bin/ntx-builder/src/db/queries/sponsored_accounts/sponsored_account.sql b/bin/ntx-builder/src/db/queries/sponsored_accounts/sponsored_account.sql deleted file mode 100644 index b26b5e2ad7..0000000000 --- a/bin/ntx-builder/src/db/queries/sponsored_accounts/sponsored_account.sql +++ /dev/null @@ -1,7 +0,0 @@ --- Resolves, in one query, the accounts targeted by the still-unconsumed feature notes that the --- supplied FEE_SPONSORSHIP notes are bound to. Joining against rarray preserves duplicate feature --- IDs, so the result still contains one account occurrence per sponsorship. -SELECT feature.account_id -FROM rarray(?1) AS sponsored -JOIN notes AS feature ON feature.note_id = sponsored.value -WHERE feature.committed_at IS NULL diff --git a/bin/ntx-builder/src/db/queries/tests.rs b/bin/ntx-builder/src/db/queries/tests.rs index 4ec4c7c837..391525ed3c 100644 --- a/bin/ntx-builder/src/db/queries/tests.rs +++ b/bin/ntx-builder/src/db/queries/tests.rs @@ -240,42 +240,6 @@ async fn sponsorships_for_pending_notes_binds_by_feature_note_not_tag() { assert_eq!(pending[&feature.as_note().id()].len(), 1); } -/// `apply_committed_block` reports one wakeup per sponsorship whose feature note is known and still -/// pending; sponsorships for consumed or unknown feature notes wake nobody. -#[tokio::test] -async fn apply_committed_block_returns_sponsored_account_wakeups() { - let (db, _dir) = test_setup().await; - let account_id = mock_network_account_id(); - let pending = mock_single_target_note(account_id, 1); - let consumed = mock_single_target_note(account_id, 2); - db.insert_network_notes(vec![pending.clone(), consumed.clone()]).await.unwrap(); - db.mark_notes_consumed(vec![consumed.as_note().nullifier()], BlockNumber::from(1)) - .await - .unwrap(); - - let effects = CommittedBlockEffects { - header: mock_block_header(BlockNumber::from(2)), - network_notes: vec![], - sponsorship_notes: vec![ - sponsorship_for(account_id, pending.as_note().id(), 3), - sponsorship_for(account_id, pending.as_note().id(), 6), - sponsorship_for(account_id, consumed.as_note().id(), 4), - sponsorship_for(account_id, NoteId::from_raw(Word::from([9, 9, 9, 9u32])), 5), - ], - nullifiers: vec![], - network_account_updates: vec![], - account_transactions: vec![], - }; - - let wakeups = db.apply_committed_block(effects, PartialMmr::default()).await.unwrap(); - - assert_eq!( - wakeups, - vec![account_id, account_id], - "each sponsorship bound to the pending feature note wakes its account once", - ); -} - // NOTE ELIGIBILITY // ================================================================================================ // @@ -439,6 +403,74 @@ async fn reset_sponsored_notes_skips_consumed_feature_notes() { ); } +/// The stored block can be too permissive: a periodic window that was open when the value was +/// written closes again later. Selection detects that and reports the correction, which +/// `update_note_eligibility` persists so the account stops being selected for the note. +#[tokio::test] +async fn stale_eligibility_is_reported_and_corrected() { + let (db, _dir) = test_setup().await; + let account_id = mock_network_account_id(); + let hint = NoteExecutionHint::on_block_slot(8, 4, 0); + let note = mock_single_target_note_with_hint(account_id, 1, hint); + db.insert_network_notes(vec![note.clone()]).await.unwrap(); + + assert_eq!( + db.note_eligibility(note.as_note().id()).await, + Some(BlockNumber::GENESIS), + "the window is open at the block that created the note", + ); + + // The window has closed again by block 100, so the stored value is now too permissive. + let available = db.available_notes(account_id, BlockNumber::from(100), 30).await.unwrap(); + assert!(available.eligible.is_empty(), "the exact check rejects the closed window"); + assert_eq!( + available.stale_eligibility, + vec![(note.as_note().nullifier(), BlockNumber::from(256))], + "selection reports the block at which the window opens again", + ); + + db.update_note_eligibility(available.stale_eligibility).await.unwrap(); + + assert!( + db.ready_accounts(30, BlockNumber::from(100), vec![], 10) + .await + .unwrap() + .is_empty(), + "once corrected, the account is no longer selected for this note", + ); +} + +/// The eligibility column, not just the attempt cap, keeps a backed-off account out of selection. +#[tokio::test] +async fn ready_accounts_respect_the_eligibility_column() { + let (db, _dir) = test_setup().await; + let account_id = mock_network_account_id(); + db.upsert_account_for_test(account_id, mock_account(account_id), mock_transaction_id(1)) + .await + .unwrap(); + let note = mock_single_target_note(account_id, 1); + db.insert_network_notes(vec![note.clone()]).await.unwrap(); + + let failed_at = BlockNumber::from(50); + db.notes_failed(vec![(note.as_note().nullifier(), test_note_error("boom"))], failed_at) + .await + .unwrap(); + let eligible_from = db.note_eligibility(note.as_note().id()).await.unwrap(); + + assert!( + db.ready_accounts(30, eligible_from.parent().unwrap(), vec![], 10) + .await + .unwrap() + .is_empty(), + "the account is not selected before its note becomes eligible", + ); + assert_eq!( + db.ready_accounts(30, eligible_from, vec![], 10).await.unwrap(), + vec![account_id], + "the account is selected again exactly at the stored block", + ); +} + // AVAILABLE NOTES + BACKOFF // ================================================================================================ @@ -540,16 +572,22 @@ async fn note_script_cache_roundtrip() { db.insert_note_scripts(root, script).await.unwrap(); } -// ACCOUNTS WITH PENDING NOTES +// READY ACCOUNTS // ================================================================================================ #[tokio::test] -async fn accounts_with_pending_notes_distinct_and_filters_consumed_and_capped() { +async fn ready_accounts_are_distinct_and_exclude_consumed_and_capped_notes() { let (db, _dir) = test_setup().await; let alice = mock_network_account_id(); let bob = mock_network_account_id_seeded(42); let carol = mock_network_account_id_seeded(99); + for account_id in [alice, bob, carol] { + db.upsert_account_for_test(account_id, mock_account(account_id), mock_transaction_id(1)) + .await + .unwrap(); + } + let alice_note_1 = mock_single_target_note(alice, 1); let alice_note_2 = mock_single_target_note(alice, 2); let bob_note = mock_single_target_note(bob, 3); @@ -559,12 +597,12 @@ async fn accounts_with_pending_notes_distinct_and_filters_consumed_and_capped() .await .unwrap(); - // Alice has two notes — must still appear exactly once (DISTINCT). Bob's only note is already - // consumed — exclude. + // Alice has two notes and must still appear exactly once. Bob's only note is already consumed, + // so he is excluded. db.mark_notes_consumed(vec![bob_note.as_note().nullifier()], BlockNumber::from(7)) .await .unwrap(); - // Carol's note has hit the attempt cap — exclude. + // Carol's note has hit the attempt cap, so she is excluded. for _ in 0..30 { db.notes_failed( vec![(carol_note.as_note().nullifier(), test_note_error("boom"))], @@ -574,9 +612,53 @@ async fn accounts_with_pending_notes_distinct_and_filters_consumed_and_capped() .unwrap(); } - let pending = db.accounts_with_pending_notes(30).await.unwrap(); - assert_eq!(pending.len(), 1, "only alice should remain pending"); - assert_eq!(pending[0], alice); + let ready = db.ready_accounts(30, BlockNumber::from(1000), vec![], 10).await.unwrap(); + assert_eq!(ready, vec![alice], "only alice has a pending note within its attempt budget"); +} + +/// The limit caps the returned accounts, and the account whose note has waited longest comes first, +/// so attempt slots rotate over the accounts that have work. +#[tokio::test] +async fn ready_accounts_are_limited_and_least_recently_attempted_first() { + let (db, _dir) = test_setup().await; + let recent = mock_network_account_id(); + let stale = mock_network_account_id_seeded(42); + + for account_id in [recent, stale] { + db.upsert_account_for_test(account_id, mock_account(account_id), mock_transaction_id(1)) + .await + .unwrap(); + } + + let recent_note = mock_single_target_note(recent, 1); + let stale_note = mock_single_target_note(stale, 2); + db.insert_network_notes(vec![recent_note.clone(), stale_note.clone()]) + .await + .unwrap(); + + db.notes_failed( + vec![(stale_note.as_note().nullifier(), test_note_error("older"))], + BlockNumber::from(1), + ) + .await + .unwrap(); + db.notes_failed( + vec![(recent_note.as_note().nullifier(), test_note_error("newer"))], + BlockNumber::from(9), + ) + .await + .unwrap(); + + assert_eq!( + db.ready_accounts(30, BlockNumber::from(1000), vec![], 1).await.unwrap(), + vec![stale], + "the least recently attempted account is served first", + ); + assert_eq!( + db.ready_accounts(30, BlockNumber::from(1000), vec![stale], 1).await.unwrap(), + vec![recent], + "an excluded account is skipped in favour of the next one", + ); } // SUBMITTED-TX LANDING @@ -636,8 +718,8 @@ async fn apply_committed_block_seeds_genesis_network_account() { db.get_account(account_id).await.unwrap().is_some(), "genesis account should be seeded" ); - // The seeded account carries the zero sentinel: no transaction produced it. An actor never - // submits the zero id, so this can never be mistaken for a landed transaction. + // The seeded account carries the zero sentinel: no transaction produced it. No submitted + // transaction has the zero id, so this can never be mistaken for a landed transaction. assert_eq!( db.account_last_tx(account_id).await.unwrap(), Some(TransactionId::from_raw(Word::empty())), @@ -705,7 +787,10 @@ async fn discard_notes_pins_attempts_to_cap_and_drops_from_pending() { "a discarded note must not be selectable", ); assert!( - !db.accounts_with_pending_notes(30).await.unwrap().contains(&account_id), - "an account whose only note was discarded must not count as pending", + !db.ready_accounts(30, BlockNumber::from(1000), vec![], 10) + .await + .unwrap() + .contains(&account_id), + "an account whose only note was discarded must not be ready for an attempt", ); } diff --git a/bin/ntx-builder/src/db/queries/update_note_eligibility/mod.rs b/bin/ntx-builder/src/db/queries/update_note_eligibility/mod.rs new file mode 100644 index 0000000000..6277c137b8 --- /dev/null +++ b/bin/ntx-builder/src/db/queries/update_note_eligibility/mod.rs @@ -0,0 +1,19 @@ +//! Corrects the stored eligibility block of notes whose exact check disagrees with it. + +use miden_node_db::sqlite::WriteTx; +use miden_node_db::{DatabaseError, SqlTypeConvert}; +use miden_protocol::block::BlockNumber; +use miden_protocol::note::Nullifier; + +const SQL: &str = include_str!("update_note_eligibility.sql"); + +/// Stores the corrected eligibility block for each note. +pub fn update_note_eligibility( + tx: &WriteTx<'_>, + eligibility: &[(Nullifier, BlockNumber)], +) -> Result<(), DatabaseError> { + for (nullifier, eligible_from) in eligibility { + tx.execute(SQL, &[nullifier, &eligible_from.to_raw_sql()])?; + } + Ok(()) +} diff --git a/bin/ntx-builder/src/db/queries/update_note_eligibility/update_note_eligibility.sql b/bin/ntx-builder/src/db/queries/update_note_eligibility/update_note_eligibility.sql new file mode 100644 index 0000000000..82b2968252 --- /dev/null +++ b/bin/ntx-builder/src/db/queries/update_note_eligibility/update_note_eligibility.sql @@ -0,0 +1,5 @@ +-- Corrects the stored eligibility block of one note. + +UPDATE notes +SET next_eligible_block = ?2 +WHERE nullifier = ?1 AND committed_at IS NULL diff --git a/bin/ntx-builder/src/execute.rs b/bin/ntx-builder/src/execute.rs index 8700eb4e79..786f0cb7a1 100644 --- a/bin/ntx-builder/src/execute.rs +++ b/bin/ntx-builder/src/execute.rs @@ -18,7 +18,6 @@ use miden_protocol::Word; use miden_protocol::account::{ Account, AccountId, - AccountPatch, AccountStorageHeader, PartialAccount, StorageMapKey, @@ -119,7 +118,7 @@ fn is_transient_rpc_error(err: &RpcError) -> bool { } /// Maximum number of retries applied to a single transient request before the error is propagated -/// to the actor-level retry. +/// to the caller. const MAX_REQUEST_RETRIES: usize = 20; /// Builds the [`ExponentialBuilder`] used to back off retries on transient request failures. @@ -142,9 +141,6 @@ fn log_transient_retry(operation: &'static str, err: &E, s pub struct NtxExecutionResult { /// ID of the submitted transaction. pub tx_id: TransactionId, - /// The account patch the transaction produced, applied to the actor's in-memory account once - /// the transaction lands. - pub account_patch: AccountPatch, /// Notes that failed consumability filtering for a genuine reason (not a cycle-budget drop). /// Their attempt counters should be incremented. pub failed_notes: Vec, @@ -162,7 +158,7 @@ pub struct NtxExecutionResult { } /// The outcome of consumability filtering: the executable set plus the failed notes partitioned by -/// how the actor should treat them. See [`NtxExecutionResult`] for the meaning of each bucket. +/// how the caller should treat them. See [`NtxExecutionResult`] for the meaning of each bucket. struct FilteredNotes { successful: InputNotes, failed: Vec, @@ -203,7 +199,7 @@ impl NtxContext { /// Creates a new [`NtxContext`] instance. #[expect( clippy::too_many_arguments, - reason = "execution context aggregates actor resources" + reason = "execution context aggregates the resources of one attempt" )] pub fn new( prover: RemoteTransactionProver, @@ -261,9 +257,9 @@ impl NtxContext { /// /// # Returns /// - /// On success, returns an [`NtxExecutionResult`] containing the transaction ID, the account - /// delta the transaction produced, any notes that failed during filtering, and note scripts - /// fetched from the remote RPC service that should be persisted to the local DB cache. + /// On success, returns an [`NtxExecutionResult`] containing the transaction ID, any notes that + /// failed during filtering, and note scripts fetched from the remote RPC service that should be + /// persisted to the local DB cache. /// /// # Errors /// @@ -336,9 +332,9 @@ impl NtxContext { .await .unwrap_or_else(|err| std::panic::resume_unwind(err.into_panic()))?; - // Destructure the executed tx into its parts; the actor applies the account patch - // to its in-memory account once this transaction lands in a committed block. - let (tx_inputs, _, account_patch, _) = executed_tx.into_parts(); + // The committed account state is derived from the block that lands this + // transaction, so the patch the execution produced is not needed here. + let (tx_inputs, ..) = executed_tx.into_parts(); // Prove transaction. let proven_tx = Box::pin(self.prove(&tx_inputs)).await?; @@ -348,7 +344,6 @@ impl NtxContext { Ok(NtxExecutionResult { tx_id: proven_tx.id(), - account_patch, failed_notes, deferred_notes, oversized_notes, @@ -594,7 +589,7 @@ impl NtxContext { /// Submits the transaction through the RPC service. /// /// Transient gRPC failures (`Unavailable`, `DeadlineExceeded`, ...) are retried in-place; - /// content-rejection codes escape on the first attempt so the actor can mark the batch failed. + /// content-rejection codes escape on the first attempt so the caller can mark the batch failed. #[miden_instrument( target = COMPONENT, name = "ntx.execute_transaction.submit", @@ -712,7 +707,7 @@ fn partition_cycle_limited(failed: Vec) -> (Vec, Vec, reference_block: BlockHeader, protocol_config: ProtocolConfig, @@ -725,8 +720,8 @@ struct NtxDataStore { script_cache: LruCache, /// Local database for persistent note script. db: NtxDbReader, - /// Scripts fetched from the remote RPC service during execution, to be persisted by the - /// coordinator. + /// Scripts fetched from the remote RPC service during execution. They are reported in the + /// execution result and persisted by the scheduler, which owns the database writer. fetched_scripts: Arc>>, /// Maps storage map roots to storage slot names. /// @@ -999,7 +994,7 @@ impl DataStore for NtxDataStore { })?; if let Some(script) = maybe_script { - // Collect for later persistence by the coordinator. + // Collect so the scheduler can persist the script after the attempt returns. self.fetched_scripts .lock() .expect("fetched scripts lock poisoned") diff --git a/bin/ntx-builder/src/lib.rs b/bin/ntx-builder/src/lib.rs index 21cfce0ca9..7fb5e2aacf 100644 --- a/bin/ntx-builder/src/lib.rs +++ b/bin/ntx-builder/src/lib.rs @@ -8,32 +8,31 @@ use std::time::Duration; use anyhow::Context; use builder::BlockStream; -use chain_state::SharedChainState; +use chain_state::ChainState; use clients::{RemoteTransactionProver, RpcClient}; use miden_node_store::genesis::GenesisBlock; use miden_node_tracing::{ErrorReport, debug}; use miden_node_utils::lru_cache::LruCache; use miden_node_utils::shutdown::CancellationToken; -use tokio::sync::mpsc; use tonic::metadata::AsciiMetadataValue; use url::Url; -use crate::actor::{AccountActorContext, ActorConfig, GrpcClients, State}; -use crate::coordinator::Coordinator; +use crate::attempt::{AttemptConfig, AttemptContext, GrpcClients}; use crate::db::NtxDbReader; +use crate::scheduler::Scheduler; pub(crate) type NoteError = Arc; -mod actor; mod allowlist; +mod attempt; mod builder; mod candidate; mod chain_state; mod clients; mod committed_block; -mod coordinator; pub(crate) mod db; mod execute; +mod scheduler; mod selection; pub mod server; mod sponsorship; @@ -119,18 +118,15 @@ pub const LOG_TARGET: &str = "user::miden-ntx-builder"; 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 should be in progress concurrently. +/// Default maximum number of network transactions which are computed concurrently. /// -/// This only counts transactions which are being computed locally and does not include -/// uncommitted transactions in the mempool. +/// This bounds the transaction attempts running 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 channel capacity for account loading through RPC. -const DEFAULT_ACCOUNT_CHANNEL_CAPACITY: usize = 1_000; - /// Default maximum number of attempts to execute a failing note before dropping it. const DEFAULT_MAX_NOTE_ATTEMPTS: usize = 30; @@ -138,9 +134,6 @@ const DEFAULT_MAX_NOTE_ATTEMPTS: usize = 30; const DEFAULT_SCRIPT_CACHE_SIZE: NonZeroUsize = NonZeroUsize::new(1_000).expect("literal is non-zero"); -/// Default duration after which an idle network account actor will deactivate. -const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_mins(5); - /// Default per-request timeout for node RPC requests. const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(10); @@ -150,9 +143,6 @@ 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 maximum number of crashes an account actor is allowed before being deactivated. -const DEFAULT_MAX_ACCOUNT_CRASHES: usize = 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`]. @@ -169,8 +159,9 @@ 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 retry timeout an actor -/// waits in `WaitForBlock` before resubmitting. Must be within the kernel's `1..=u16::MAX` range. +/// Used both as the on-chain transaction expiration delta and as the local timeout after which the +/// scheduler releases the account for a new attempt. Must be within the kernel's `1..=u16::MAX` +/// range. const DEFAULT_TX_EXPIRATION_DELTA: NonZeroU16 = NonZeroU16::new(30).unwrap(); // CONFIGURATION @@ -203,8 +194,8 @@ pub struct NtxBuilderConfig { /// repeated gRPC calls. pub script_cache_size: NonZeroUsize, - /// Maximum number of network transactions which should be in progress concurrently across all - /// account actors. + /// Maximum number of network transactions computed concurrently. Submitted transactions waiting + /// to be committed do not count against this limit. pub max_concurrent_txs: usize, /// Maximum number of network notes a single transaction is allowed to consume. Sponsorship @@ -218,20 +209,6 @@ pub struct NtxBuilderConfig { /// Maximum number of blocks to keep in the chain MMR. Older blocks are pruned. pub max_block_count: usize, - /// Channel capacity for loading accounts through RPC during startup. - pub account_channel_capacity: usize, - - /// Duration after which an idle network account will deactivate. - /// - /// An account is considered idle once it has no viable notes to consume. - /// A deactivated account will reactivate if targeted with new notes. - pub idle_timeout: Duration, - - /// Maximum number of crashes before an account deactivated. - /// - /// Once this limit is reached, no new transactions will be created for this account. - pub max_account_crashes: 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. @@ -239,8 +216,9 @@ pub struct NtxBuilderConfig { 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 `WaitForBlock` retry timeout. Must be - /// within `1..=u16::MAX` (enforced by the transaction kernel). + /// transaction expiration delta and reused as the local timeout after which the scheduler + /// releases the account for a new attempt. 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. @@ -273,9 +251,6 @@ impl NtxBuilderConfig { max_notes_per_tx: DEFAULT_MAX_NOTES_PER_TX, max_note_attempts: DEFAULT_MAX_NOTE_ATTEMPTS, max_block_count: DEFAULT_MAX_BLOCK_COUNT, - account_channel_capacity: DEFAULT_ACCOUNT_CHANNEL_CAPACITY, - idle_timeout: DEFAULT_IDLE_TIMEOUT, - max_account_crashes: DEFAULT_MAX_ACCOUNT_CRASHES, max_cycles: DEFAULT_MAX_TX_CYCLES, tx_expiration_delta: DEFAULT_TX_EXPIRATION_DELTA, request_backoff_initial: DEFAULT_REQUEST_BACKOFF_INITIAL, @@ -358,29 +333,6 @@ impl NtxBuilderConfig { self } - /// Sets the account channel capacity for startup loading. - #[must_use] - pub fn with_account_channel_capacity(mut self, capacity: usize) -> Self { - self.account_channel_capacity = capacity; - self - } - - /// Sets the idle timeout for actors. - /// - /// Actors that remain idle (no viable notes) for this duration will be deactivated. - #[must_use] - pub fn with_idle_timeout(mut self, timeout: Duration) -> Self { - self.idle_timeout = timeout; - self - } - - /// Sets the maximum number of crashes before an account actor is deactivated. - #[must_use] - pub fn with_max_account_crashes(mut self, max: usize) -> Self { - self.max_account_crashes = 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 { @@ -388,8 +340,8 @@ impl NtxBuilderConfig { self } - /// Sets the transaction expiration delta (in blocks). Also bounds the actor's `WaitForBlock` - /// retry timeout. + /// 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; @@ -430,8 +382,8 @@ impl NtxBuilderConfig { shutdown: CancellationToken, ) -> anyhow::Result { // Set up the database connection pool. Writes are serialized by the framework's single - // dedicated writer connection, so block application never contends with the account actors - // (which only read) for the shared reader pool. + // dedicated writer connection, so block application never contends with the transaction + // attempts (which only read) for the shared reader pool. let db = db::load_with_pool_size( self.database_filepath.clone(), self.sqlite_connection_pool_size, @@ -517,10 +469,9 @@ impl NtxBuilderConfig { // block that the builder has not applied. let block_stream: BlockStream = Box::pin(rpc.block_subscription_reconnecting(block_from)); - let chain = Arc::new(SharedChainState::new(header, mmr)); + let chain = ChainState::new(header, mmr); - let (coordinator, actor_request_rx) = - self.build_coordinator(rpc, db.reader(), chain.clone(), shutdown)?; + let scheduler = self.build_scheduler(rpc, db.reader())?; Ok(NetworkTransactionBuilder::new( self, @@ -528,26 +479,16 @@ impl NtxBuilderConfig { block_stream, last_applied_block, chain, - coordinator, - actor_request_rx, + scheduler, )) } - /// Builds the actor [`Coordinator`] and the channel over which spawned actors send their DB - /// writes back to the builder's event loop. + /// Builds the [`Scheduler`] that owns the transaction attempts. /// - /// The receiver is owned by the builder loop; the sender is cloned into every spawned actor so - /// all actor-side DB writes serialize through the loop. - fn build_coordinator( - &self, - rpc: RpcClient, - db: NtxDbReader, - chain: Arc, - shutdown: CancellationToken, - ) -> anyhow::Result<(Coordinator, mpsc::Receiver)> { - let (request_tx, actor_request_rx) = mpsc::channel(self.account_channel_capacity); - let tx_args = selection::build_tx_args(self.tx_expiration_delta); - let actor_context = AccountActorContext { + /// The attempt context it carries is cloned into every spawned attempt, 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 { + let ctx = AttemptContext { clients: GrpcClients { rpc, prover: RemoteTransactionProver::new( @@ -555,30 +496,18 @@ impl NtxBuilderConfig { self.tx_prover_timeout, )?, }, - state: State { - db, - chain, - script_cache: LruCache::new(self.script_cache_size), - tx_args, - }, - config: ActorConfig { + db, + script_cache: LruCache::new(self.script_cache_size), + tx_args: selection::build_tx_args(self.tx_expiration_delta), + config: AttemptConfig { max_notes_per_tx: self.max_notes_per_tx, max_note_attempts: self.max_note_attempts, - idle_timeout: self.idle_timeout, max_cycles: self.max_cycles, - tx_expiration_delta: self.tx_expiration_delta, request_backoff_initial: self.request_backoff_initial, request_backoff_max: self.request_backoff_max, }, - request_tx, }; - let coordinator = Coordinator::new( - self.max_concurrent_txs, - self.max_account_crashes, - actor_context, - shutdown, - ); - Ok((coordinator, actor_request_rx)) + Ok(Scheduler::new(ctx, self.max_concurrent_txs, self.tx_expiration_delta)) } } diff --git a/bin/ntx-builder/src/scheduler.rs b/bin/ntx-builder/src/scheduler.rs new file mode 100644 index 0000000000..e639336884 --- /dev/null +++ b/bin/ntx-builder/src/scheduler.rs @@ -0,0 +1,549 @@ +//! Block-driven scheduler for network transaction attempts. +//! +//! The scheduler owns the attempt tasks and decides which accounts to work on. It keeps no +//! per-account state beyond the in-flight set, which records the transactions it has submitted but +//! not yet seen committed. Everything else that decides "what to work on next" is read from the +//! database on every dispatch. + +use std::collections::HashMap; +use std::num::NonZeroU16; + +use anyhow::Context; +use miden_node_tracing::{debug, error, info, miden_instrument}; +use miden_node_utils::formatting::format_opt; +use miden_protocol::account::AccountId; +use miden_protocol::block::BlockNumber; +use miden_protocol::transaction::TransactionId; +use tokio::task::{Id, JoinSet}; + +use crate::LOG_TARGET; +use crate::attempt::{AttemptContext, AttemptOutcome, AttemptResult, NoteUpdates, attempt}; +use crate::chain_state::ChainState; +use crate::committed_block::CommittedBlockEffects; +use crate::db::NtxDbWriter; + +// SCHEDULER STATE +// ================================================================================================ + +/// A transaction the scheduler submitted and has not yet seen committed. +struct Inflight { + /// Id of the submitted transaction, compared against each block's committed transactions. + tx_id: TransactionId, + /// Chain tip at submission. With the transaction expiration delta this bounds how long the + /// account stays blocked when the submission never lands. + submitted_at: BlockNumber, +} + +// SCHEDULER +// ================================================================================================ + +/// Spawns and reaps network transaction attempts. +/// +/// The scheduler is driven from the builder's event loop at three moments: +/// +/// 1. On every committed block, [`Scheduler::handle_committed_block`] resolves the in-flight set +/// (landed or expired) and [`Scheduler::dispatch`] fills the free attempt slots. +/// 2. Whenever an attempt completes, [`Scheduler::handle_completion`] persists what the attempt +/// reported and says whether the freed slot should be refilled immediately. +/// 3. On shutdown, [`Scheduler::shutdown`] aborts the outstanding attempts. +pub struct Scheduler { + /// Resources cloned into every spawned attempt. + ctx: AttemptContext, + + /// The spawned attempt tasks. + tasks: JoinSet, + + /// Accounts with a running attempt, keyed by task id. These occupy the attempt slots. + running: HashMap, + + /// Accounts with a submitted transaction awaiting commitment. These do not occupy an attempt + /// slot (no work is being computed locally) but are excluded from selection so an account never + /// has two transactions in flight. + in_flight: HashMap, + + /// Maximum number of attempts computed concurrently. + max_concurrent_txs: usize, + + /// Number of blocks after which a submitted transaction expires. An in-flight entry older than + /// this is dropped, which releases the account for a new attempt. + tx_expiration_delta: NonZeroU16, +} + +impl Scheduler { + pub fn new( + ctx: AttemptContext, + max_concurrent_txs: usize, + tx_expiration_delta: NonZeroU16, + ) -> Self { + Self { + ctx, + tasks: JoinSet::new(), + running: HashMap::new(), + in_flight: HashMap::new(), + max_concurrent_txs, + tx_expiration_delta, + } + } + + /// Fills the free attempt slots with accounts that have pending notes. + /// + /// Accounts with a running attempt or an in-flight transaction are excluded, so an account + /// never has two attempts or two submitted transactions at once. `chain` fixes the reference + /// block for every attempt spawned here. + #[miden_instrument( + name = "ntx.scheduler.dispatch", + fields(tip.number = chain.chain_tip_header.block_num()), + err, + )] + pub async fn dispatch(&mut self, chain: &ChainState) -> anyhow::Result<()> { + let free = self.max_concurrent_txs.saturating_sub(self.running.len()); + if free == 0 { + return Ok(()); + } + + let busy = self + .running + .values() + .copied() + .chain(self.in_flight.keys().copied()) + .collect::>(); + + let block_num = chain.chain_tip_header.block_num(); + let ready = self + .ctx + .db + .ready_accounts(self.ctx.config.max_note_attempts, block_num, busy, free) + .await + .context("failed to query accounts ready for a transaction attempt")?; + for account_id in ready { + let ctx = self.ctx.clone(); + let chain = chain.clone(); + let handle = self.tasks.spawn(attempt(ctx, account_id, chain)); + self.running.insert(handle.id(), account_id); + debug!( + target: LOG_TARGET, + "dispatched a network transaction attempt", + account.id = account_id, + reference_block.number = block_num + ); + } + + Ok(()) + } + + /// Resolves the in-flight set against a committed block. + /// + /// An entry is dropped when the block commits its transaction (the account is free to work + /// again on the state that transaction produced) or when the submission has been outstanding + /// for longer than the expiration delta, in which case it can no longer land on-chain. + pub fn handle_committed_block(&mut self, effects: &CommittedBlockEffects) { + let tip = effects.header.block_num(); + let committed = effects.latest_tx_per_account(); + let expiration_delta = u32::from(self.tx_expiration_delta.get()); + + self.in_flight.retain(|account_id, inflight| { + if committed.get(account_id) == Some(&inflight.tx_id) { + info!( + target: LOG_TARGET, + "submitted network transaction landed", + account.id = *account_id, + transaction.id = inflight.tx_id, + block.number = tip + ); + return false; + } + + let elapsed = tip.checked_sub(inflight.submitted_at.as_u32()).unwrap_or_default(); + if elapsed.as_u32() >= expiration_delta { + info!( + target: LOG_TARGET, + "submitted network transaction expired", + account.id = *account_id, + transaction.id = inflight.tx_id, + transaction.submitted_at = inflight.submitted_at, + tip.number = tip, + transaction.expiration_delta = expiration_delta + ); + return false; + } + + true + }); + } + + /// Waits for the next attempt to complete. + /// + /// Waits indefinitely while no attempt is running, so this is safe to poll in a `select!` + /// alongside the block stream. An attempt task that does not return an outcome has panicked, + /// which is a bug rather than a state the scheduler can recover from, so the error is + /// propagated and ends the event loop. + pub async fn next_completion(&mut self) -> anyhow::Result { + loop { + match self.tasks.join_next_with_id().await { + Some(Ok((id, outcome))) => { + self.running.remove(&id); + return Ok(outcome); + }, + Some(Err(err)) => { + let account_id = self.running.remove(&err.id()); + // Cancelled tasks were aborted on shutdown. + if err.is_cancelled() { + continue; + } + return Err(err).with_context(|| { + format!( + "network transaction attempt failed for account {}", + format_opt(account_id.as_ref()) + ) + }); + }, + // No attempt is running. Wait until one is spawned and completes. + None => std::future::pending().await, + } + } + } + + /// Persists what an attempt reported and returns whether the freed slot should be refilled now. + /// + /// A slot is refilled immediately after an attempt that made progress, so a completed + /// transaction does not leave capacity idle until the next block. An attempt that found no + /// viable work, or that could not run at all, does not trigger a refill: the state that + /// selected its account has not changed, so an immediate re-dispatch could pick the same + /// account again. + pub async fn handle_completion( + &mut self, + db: &NtxDbWriter, + outcome: AttemptOutcome, + ) -> anyhow::Result { + let AttemptOutcome { account_id, block_num, notes, result } = outcome; + self.persist_note_updates(db, block_num, notes).await?; + + match result { + AttemptResult::Submitted { tx_id } => { + info!( + target: LOG_TARGET, + "network transaction submitted; account is in flight", + account.id = account_id, + transaction.id = tx_id, + transaction.submitted_at = block_num + ); + self.in_flight.insert(account_id, Inflight { tx_id, submitted_at: block_num }); + Ok(true) + }, + AttemptResult::Failed => Ok(true), + AttemptResult::NoWork => { + debug!( + target: LOG_TARGET, + "no viable notes for account", + account.id = account_id + ); + Ok(false) + }, + AttemptResult::Aborted(err) => { + error!( + &err, + target: LOG_TARGET, + "network transaction attempt could not run", + account.id = account_id + ); + Ok(false) + }, + } + } + + /// Writes the note bookkeeping an attempt produced. + async fn persist_note_updates( + &self, + db: &NtxDbWriter, + block_num: BlockNumber, + notes: NoteUpdates, + ) -> anyhow::Result<()> { + let NoteUpdates { failed, discarded, scripts, eligibility } = notes; + + // Applied before the failures so a note that is both corrected and penalized keeps the + // backoff block the penalty computes, which is the later of the two. + if !eligibility.is_empty() { + db.update_note_eligibility(eligibility) + .await + .context("failed to correct note eligibility")?; + } + if !failed.is_empty() { + db.notes_failed(failed, block_num) + .await + .context("failed to persist note failures")?; + } + if !discarded.is_empty() { + db.discard_notes(discarded, block_num, self.ctx.config.max_note_attempts) + .await + .context("failed to discard notes")?; + } + for (script_root, script) in scripts { + db.insert_note_scripts(script_root, script) + .await + .context("failed to cache note script")?; + } + + Ok(()) + } + + /// Aborts every outstanding attempt and waits for the tasks to finish. + /// + /// An aborted attempt loses its note bookkeeping. Its notes stay pending, and a transaction it + /// already submitted either lands (its notes are then marked consumed from the committed block) + /// or expires on-chain. + pub async fn shutdown(&mut self) { + self.tasks.shutdown().await; + self.running.clear(); + self.in_flight.clear(); + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use miden_protocol::account::AccountUpdateDetails; + + use super::*; + use crate::NoteError; + use crate::test_utils::{ + mock_block_header, + mock_network_account_id, + mock_network_account_id_seeded, + mock_transaction_id, + }; + + /// Builds a scheduler backed by a temp database. The attempt context points at unreachable + /// endpoints, which is enough for the scheduling logic under test (no attempt is spawned). + async fn test_scheduler() -> (Scheduler, NtxDbWriter, tempfile::TempDir) { + let (db, dir) = crate::db::test_setup().await; + let ctx = AttemptContext::test(&db.reader()); + (Scheduler::new(ctx, 4, NonZeroU16::new(30).unwrap()), db, dir) + } + + /// Effects for a block carrying nothing but its header. + fn empty_effects(block_num: u32) -> CommittedBlockEffects { + CommittedBlockEffects { + header: mock_block_header(block_num.into()), + network_notes: vec![], + sponsorship_notes: vec![], + nullifiers: vec![], + network_account_updates: vec![], + account_transactions: vec![], + } + } + + #[tokio::test] + async fn landed_transaction_releases_the_account() { + let (mut scheduler, _db, _dir) = test_scheduler().await; + let account_id = mock_network_account_id(); + let tx_id = mock_transaction_id(1); + + scheduler + .in_flight + .insert(account_id, Inflight { tx_id, submitted_at: 1_u32.into() }); + + let mut effects = empty_effects(2); + effects.account_transactions = vec![(account_id, tx_id)]; + scheduler.handle_committed_block(&effects); + + assert!( + scheduler.in_flight.is_empty(), + "the block committing the submission must release the account", + ); + } + + /// A block that commits a *different* transaction for the account leaves the entry in place: + /// the submission may still land in a later block. + #[tokio::test] + async fn unrelated_transaction_keeps_the_account_in_flight() { + let (mut scheduler, _db, _dir) = test_scheduler().await; + let account_id = mock_network_account_id(); + + scheduler.in_flight.insert( + account_id, + Inflight { + tx_id: mock_transaction_id(1), + submitted_at: 1_u32.into(), + }, + ); + + let mut effects = empty_effects(2); + effects.account_transactions = vec![(account_id, mock_transaction_id(9))]; + scheduler.handle_committed_block(&effects); + + assert!(scheduler.in_flight.contains_key(&account_id)); + } + + #[tokio::test] + async fn expired_submission_releases_the_account() { + let (mut scheduler, _db, _dir) = test_scheduler().await; + let account_id = mock_network_account_id(); + + scheduler.in_flight.insert( + account_id, + Inflight { + tx_id: mock_transaction_id(1), + submitted_at: 1_u32.into(), + }, + ); + + // One block short of the delta: still waiting. + scheduler.handle_committed_block(&empty_effects(30)); + assert!(scheduler.in_flight.contains_key(&account_id)); + + // The delta has now fully elapsed, so the submission can no longer land. + scheduler.handle_committed_block(&empty_effects(31)); + assert!(scheduler.in_flight.is_empty()); + } + + /// The slot budget counts running attempts only. An in-flight transaction blocks its own + /// account but does not consume a slot, because no work is being computed for it locally. + #[tokio::test] + async fn in_flight_accounts_are_excluded_but_do_not_occupy_a_slot() { + let (mut scheduler, db, _dir) = test_scheduler().await; + let in_flight_account = mock_network_account_id(); + let other_account = mock_network_account_id_seeded(42); + + for account_id in [in_flight_account, other_account] { + db.upsert_account_for_test( + account_id, + crate::test_utils::mock_account(account_id), + mock_transaction_id(0), + ) + .await + .unwrap(); + db.insert_network_notes(vec![crate::test_utils::mock_single_target_note( + account_id, 1, + )]) + .await + .unwrap(); + } + + scheduler.in_flight.insert( + in_flight_account, + Inflight { + tx_id: mock_transaction_id(1), + submitted_at: 1_u32.into(), + }, + ); + + let ready = db + .ready_accounts( + 30, + BlockNumber::from(1), + vec![in_flight_account], + scheduler.max_concurrent_txs, + ) + .await + .unwrap(); + + assert_eq!(ready, vec![other_account], "an in-flight account is not selected again"); + } + + /// An account with pending notes but no committed state is not a candidate. The join against + /// `accounts` in the query is the only gate on account state. + #[tokio::test] + async fn uncommitted_accounts_are_not_ready() { + let (_scheduler, db, _dir) = test_scheduler().await; + let account_id = mock_network_account_id(); + + db.insert_network_notes(vec![crate::test_utils::mock_single_target_note(account_id, 1)]) + .await + .unwrap(); + + assert!( + db.ready_accounts(30, BlockNumber::from(1), vec![], 4).await.unwrap().is_empty(), + "a note targeting an account with no committed state is not dispatchable", + ); + + db.upsert_account_for_test( + account_id, + crate::test_utils::mock_account(account_id), + mock_transaction_id(0), + ) + .await + .unwrap(); + + assert_eq!( + db.ready_accounts(30, BlockNumber::from(1), vec![], 4).await.unwrap(), + vec![account_id], + "the account becomes dispatchable once its state is committed", + ); + } + + /// The note bookkeeping an attempt reports is persisted whatever the result, and a submission + /// puts the account in flight. + #[tokio::test] + async fn submitted_outcome_persists_notes_and_records_the_submission() { + let (mut scheduler, db, _dir) = test_scheduler().await; + let account_id = mock_network_account_id(); + let failed_note = crate::test_utils::mock_single_target_note(account_id, 1); + let discarded_note = crate::test_utils::mock_single_target_note(account_id, 2); + db.insert_network_notes(vec![failed_note.clone(), discarded_note.clone()]) + .await + .unwrap(); + + let tx_id = mock_transaction_id(3); + let error: NoteError = Arc::new(std::io::Error::other("boom")); + let outcome = AttemptOutcome { + account_id, + block_num: 7_u32.into(), + notes: NoteUpdates { + failed: vec![(failed_note.as_note().nullifier(), error)], + discarded: vec![discarded_note.as_note().nullifier()], + eligibility: vec![], + scripts: vec![], + }, + result: AttemptResult::Submitted { tx_id }, + }; + + let refill = scheduler.handle_completion(&db, outcome).await.unwrap(); + + assert!(refill, "a submission frees a slot that should be refilled immediately"); + assert!(scheduler.in_flight.contains_key(&account_id)); + + let failed = db.get_note_status(failed_note.as_note().id()).await.unwrap().unwrap(); + assert_eq!(failed.attempt_count, 1); + let discarded = db.get_note_status(discarded_note.as_note().id()).await.unwrap().unwrap(); + assert_eq!(discarded.attempt_count, 30, "a discarded note is pinned to the attempt cap"); + } + + /// An attempt that found no viable work must not trigger an immediate re-dispatch, because + /// nothing about its account changed. + #[tokio::test] + async fn no_work_outcome_does_not_refill_the_slot() { + let (mut scheduler, db, _dir) = test_scheduler().await; + let outcome = AttemptOutcome { + account_id: mock_network_account_id(), + block_num: 1_u32.into(), + notes: NoteUpdates::default(), + result: AttemptResult::NoWork, + }; + + let refill = scheduler.handle_completion(&db, outcome).await.unwrap(); + + assert!(!refill); + assert!(scheduler.in_flight.is_empty()); + } + + /// A block that only updates an account (without a note for it) leaves the in-flight set alone. + #[tokio::test] + async fn account_update_alone_does_not_resolve_an_in_flight_entry() { + let (mut scheduler, _db, _dir) = test_scheduler().await; + let account_id = mock_network_account_id(); + scheduler.in_flight.insert( + account_id, + Inflight { + tx_id: mock_transaction_id(1), + submitted_at: 1_u32.into(), + }, + ); + + let mut effects = empty_effects(2); + effects.network_account_updates = vec![(account_id, AccountUpdateDetails::Private)]; + scheduler.handle_committed_block(&effects); + + assert!(scheduler.in_flight.contains_key(&account_id)); + } +} diff --git a/bin/ntx-builder/src/selection.rs b/bin/ntx-builder/src/selection.rs index 169aed8039..bae551d32d 100644 --- a/bin/ntx-builder/src/selection.rs +++ b/bin/ntx-builder/src/selection.rs @@ -56,9 +56,10 @@ pub(crate) struct Selection { /// Notes dropped because the account does not allowlist their script root. They can never be /// consumed by this account, so the caller must penalize them. pub rejected: Vec<(Nullifier, NoteError)>, - /// Earliest block at which a currently-ineligible note becomes eligible, or `None` when the - /// account has no pending note awaiting a backoff or execution-hint window. - pub next_retry_block: Option, + /// Notes whose stored eligibility block has passed while the exact hint and backoff check still + /// rejects them, paired with the block they really become eligible at. The caller persists + /// these; see [`crate::db::eligibility`]. + pub stale_eligibility: Vec<(Nullifier, BlockNumber)>, } /// Selects a transaction candidate for `account` by querying its available notes. @@ -80,7 +81,7 @@ pub(crate) async fn select_candidate( .available_notes(account_id, block_num, max_note_attempts) .await .context("failed to query DB for available notes")?; - let next_retry_block = availability.next_retry_block; + let stale_eligibility = availability.stale_eligibility; let partitioned_notes = partition_by_allowlist(account.as_ref(), availability.eligible) .context("failed to read network account note allowlist")?; @@ -142,17 +143,10 @@ pub(crate) async fn select_candidate( } if selected.is_empty() { - // Notes just dropped by the allowlist re-enter eligibility through backoff, so ask for a - // re-check on the next block rather than reporting the account as having no pending work. - let next_retry_block = if rejected.is_empty() { - next_retry_block - } else { - Some(next_retry_block.map_or(block_num.child(), |block| block.min(block_num.child()))) - }; return Ok(Selection { candidate: None, rejected, - next_retry_block, + stale_eligibility, }); } @@ -166,7 +160,7 @@ pub(crate) async fn select_candidate( chain_mmr, }), rejected, - next_retry_block, + stale_eligibility, }) } @@ -467,10 +461,9 @@ end"; assert!(selection.candidate.is_none(), "a non-allowlisted note is never selected"); assert_eq!(selection.rejected.len(), 1, "the note is reported for the caller to penalize"); assert_eq!(selection.rejected[0].0, note.as_note().nullifier()); - assert_eq!( - selection.next_retry_block, - Some(BlockNumber::from(1)), - "a rejected note schedules a re-check on the next block", + assert!( + selection.stale_eligibility.is_empty(), + "the note was eligible; penalizing it is what moves its eligibility", ); } diff --git a/crates/tracing/src/attribute.rs b/crates/tracing/src/attribute.rs index a9c7b7a11c..0fee45fe12 100644 --- a/crates/tracing/src/attribute.rs +++ b/crates/tracing/src/attribute.rs @@ -60,6 +60,7 @@ const NUMBER_FIELD_NAMES: &[&str] = &[ "mempool.transactions.unbatched", "mempool.transactions.uncommitted", "note.tag", + "ntx_builder.max_concurrent_txs", "ntx_builder.max_cycles", "ntx_builder.tx_expiration_delta", "port",