diff --git a/bin/ntx-builder/src/commands/mod.rs b/bin/ntx-builder/src/commands/mod.rs index 1e2a2ad74..60126d3ca 100644 --- a/bin/ntx-builder/src/commands/mod.rs +++ b/bin/ntx-builder/src/commands/mod.rs @@ -12,6 +12,7 @@ use miden_node_utils::formatting::format_endpoint; use miden_node_utils::fs::ensure_empty_directory; use miden_node_utils::genesis::{OfficialNetwork, fetch_genesis_block, read_genesis_block}; use miden_node_utils::shutdown::CancellationToken; +use miden_protocol::account::AccountId; use tokio::net::TcpListener; use tonic::metadata::AsciiMetadataValue; use url::Url; @@ -26,6 +27,7 @@ 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_PRIORITY_ACCOUNTS: &str = "MIDEN_NODE_NTX_BUILDER_PRIORITY_ACCOUNTS"; 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"; @@ -114,6 +116,19 @@ pub enum NtxBuilderCommand { )] max_concurrent_txs: usize, + /// Network account served before every other account, such as the native faucet. + /// + /// Repeat the flag to prioritize several accounts. Keep the list shorter than + /// `--max-concurrent-txs` so a slot always remains for the other accounts. + #[arg( + long = "priority-account", + env = ENV_PRIORITY_ACCOUNTS, + value_delimiter = ',', + value_parser = parse_account_id, + value_name = "ACCOUNT_ID" + )] + priority_accounts: Vec, + /// Maximum number of VM execution cycles allowed for a single network transaction. /// /// Network transactions that exceed this limit will fail. Defaults to 2^18 (262.144) @@ -246,6 +261,7 @@ impl NtxBuilderCommand { tx_prover_timeout, script_cache_size, max_concurrent_txs, + priority_accounts, max_tx_cycles, tx_expiration_delta, sqlite_connection_pool_size, @@ -269,6 +285,7 @@ impl NtxBuilderCommand { tx_prover.timeout = humantime::Duration::from(tx_prover_timeout).to_string(), rpc.authentication.configured = rpc_auth_header_value.is_some(), ntx_builder.max_concurrent_txs = max_concurrent_txs, + account.ids.count = priority_accounts.len(), 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() @@ -287,6 +304,7 @@ impl NtxBuilderCommand { .with_tx_prover_timeout(tx_prover_timeout) .with_script_cache_size(script_cache_size) .with_max_concurrent_txs(max_concurrent_txs) + .with_priority_accounts(priority_accounts) .with_max_cycles(max_tx_cycles) .with_tx_expiration_delta(tx_expiration_delta) .with_sqlite_connection_pool_size(sqlite_connection_pool_size); @@ -324,3 +342,8 @@ async fn read_bootstrap_genesis_block( _ => unreachable!("clap requires exactly one genesis block source"), } } + +/// Parses a network account id from its hex representation. +fn parse_account_id(value: &str) -> anyhow::Result { + AccountId::from_hex(value).map_err(Into::into) +} diff --git a/bin/ntx-builder/src/db/mod.rs b/bin/ntx-builder/src/db/mod.rs index 50a15368e..811caaba4 100644 --- a/bin/ntx-builder/src/db/mod.rs +++ b/bin/ntx-builder/src/db/mod.rs @@ -125,11 +125,12 @@ impl NtxDbReader { max_note_attempts: usize, block_num: BlockNumber, busy: Vec, + priority: Vec, limit: usize, ) -> Result, DatabaseError> { self.reader .read("ready_accounts", move |tx| { - queries::ready_accounts(tx, max_note_attempts, block_num, &busy, limit) + queries::ready_accounts(tx, max_note_attempts, block_num, &busy, &priority, limit) }) .await } diff --git a/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs b/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs index 898a9d58b..d897225e9 100644 --- a/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs +++ b/bin/ntx-builder/src/db/queries/ready_accounts/mod.rs @@ -11,18 +11,27 @@ const SQL: &str = include_str!("ready_accounts.sql"); /// longest-waiting one first. /// /// `busy` names the accounts to skip: those with a running attempt or an in-flight transaction. +/// `priority` names the accounts to serve first; the rest are served longest-waiting first. #[expect(clippy::cast_possible_wrap)] pub fn ready_accounts( tx: &ReadTx<'_>, max_attempts: usize, block_num: BlockNumber, busy: &[AccountId], + priority: &[AccountId], limit: usize, ) -> Result, DatabaseError> { let busy = InList::from_values(busy.iter().copied()); + let priority = InList::from_values(priority.iter().copied()); tx.query( SQL, - &[&(max_attempts as i64), &block_num.to_raw_sql(), &busy, &(limit as i64)], + &[ + &(max_attempts as i64), + &block_num.to_raw_sql(), + &busy, + &priority, + &(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 index f81dbe40e..0cd977461 100644 --- a/bin/ntx-builder/src/db/queries/ready_accounts/ready_accounts.sql +++ b/bin/ntx-builder/src/db/queries/ready_accounts/ready_accounts.sql @@ -10,5 +10,6 @@ WHERE n.committed_at IS NULL 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 +ORDER BY (n.account_id IN (SELECT value FROM rarray(?4))) DESC, + MIN(n.next_eligible_block) ASC +LIMIT ?5 diff --git a/bin/ntx-builder/src/db/queries/tests.rs b/bin/ntx-builder/src/db/queries/tests.rs index 391525ed3..9791531bd 100644 --- a/bin/ntx-builder/src/db/queries/tests.rs +++ b/bin/ntx-builder/src/db/queries/tests.rs @@ -432,7 +432,7 @@ async fn stale_eligibility_is_reported_and_corrected() { db.update_note_eligibility(available.stale_eligibility).await.unwrap(); assert!( - db.ready_accounts(30, BlockNumber::from(100), vec![], 10) + db.ready_accounts(30, BlockNumber::from(100), vec![], vec![], 10) .await .unwrap() .is_empty(), @@ -458,14 +458,14 @@ async fn ready_accounts_respect_the_eligibility_column() { let eligible_from = db.note_eligibility(note.as_note().id()).await.unwrap(); assert!( - db.ready_accounts(30, eligible_from.parent().unwrap(), vec![], 10) + db.ready_accounts(30, eligible_from.parent().unwrap(), vec![], 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(), + db.ready_accounts(30, eligible_from, vec![], vec![], 10).await.unwrap(), vec![account_id], "the account is selected again exactly at the stored block", ); @@ -612,7 +612,10 @@ async fn ready_accounts_are_distinct_and_exclude_consumed_and_capped_notes() { .unwrap(); } - let ready = db.ready_accounts(30, BlockNumber::from(1000), vec![], 10).await.unwrap(); + let ready = db + .ready_accounts(30, BlockNumber::from(1000), vec![], vec![], 10) + .await + .unwrap(); assert_eq!(ready, vec![alice], "only alice has a pending note within its attempt budget"); } @@ -650,17 +653,62 @@ async fn ready_accounts_are_limited_and_least_recently_attempted_first() { .unwrap(); assert_eq!( - db.ready_accounts(30, BlockNumber::from(1000), vec![], 1).await.unwrap(), + db.ready_accounts(30, BlockNumber::from(1000), vec![], 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(), + db.ready_accounts(30, BlockNumber::from(1000), vec![stale], vec![], 1) + .await + .unwrap(), vec![recent], "an excluded account is skipped in favour of the next one", ); } +/// A priority account is served before every other ready account, whatever the wait ordering says. +#[tokio::test] +async fn ready_accounts_serve_priority_accounts_first() { + let (db, _dir) = test_setup().await; + let ordinary = mock_network_account_id(); + let prioritized = mock_network_account_id_seeded(42); + + for account_id in [ordinary, prioritized] { + db.upsert_account_for_test(account_id, mock_account(account_id), mock_transaction_id(1)) + .await + .unwrap(); + } + + let ordinary_note = mock_single_target_note(ordinary, 1); + let prioritized_note = mock_single_target_note(prioritized, 2); + db.insert_network_notes(vec![ordinary_note, prioritized_note.clone()]) + .await + .unwrap(); + + // The priority account has waited least, so the wait ordering alone would serve it last. + db.notes_failed( + vec![(prioritized_note.as_note().nullifier(), test_note_error("boom"))], + BlockNumber::from(9), + ) + .await + .unwrap(); + + assert_eq!( + db.ready_accounts(30, BlockNumber::from(1000), vec![], vec![prioritized], 1) + .await + .unwrap(), + vec![prioritized], + "the priority account takes the only free slot", + ); + assert_eq!( + db.ready_accounts(30, BlockNumber::from(1000), vec![], vec![prioritized], 2) + .await + .unwrap(), + vec![prioritized, ordinary], + "with room for both, the priority account still comes first", + ); +} + // SUBMITTED-TX LANDING // ================================================================================================ @@ -787,7 +835,7 @@ async fn discard_notes_pins_attempts_to_cap_and_drops_from_pending() { "a discarded note must not be selectable", ); assert!( - !db.ready_accounts(30, BlockNumber::from(1000), vec![], 10) + !db.ready_accounts(30, BlockNumber::from(1000), vec![], vec![], 10) .await .unwrap() .contains(&account_id), diff --git a/bin/ntx-builder/src/lib.rs b/bin/ntx-builder/src/lib.rs index 7fb5e2aac..5cdbd1e15 100644 --- a/bin/ntx-builder/src/lib.rs +++ b/bin/ntx-builder/src/lib.rs @@ -11,9 +11,10 @@ use builder::BlockStream; use chain_state::ChainState; use clients::{RemoteTransactionProver, RpcClient}; use miden_node_store::genesis::GenesisBlock; -use miden_node_tracing::{ErrorReport, debug}; +use miden_node_tracing::{ErrorReport, debug, warn}; use miden_node_utils::lru_cache::LruCache; use miden_node_utils::shutdown::CancellationToken; +use miden_protocol::account::AccountId; use tonic::metadata::AsciiMetadataValue; use url::Url; @@ -198,6 +199,12 @@ pub struct NtxBuilderConfig { /// to be committed do not count against this limit. pub max_concurrent_txs: usize, + /// Network accounts served before every other account, such as the native faucet. + /// + /// Each one still holds at most one attempt slot. Keep the list shorter than + /// [`Self::max_concurrent_txs`] so a slot always remains for the other accounts. + pub priority_accounts: Vec, + /// Maximum number of network notes a single transaction is allowed to consume. Sponsorship /// notes count against this budget. pub max_notes_per_tx: NonZeroUsize, @@ -248,6 +255,7 @@ impl NtxBuilderConfig { grpc_timeout: DEFAULT_GRPC_TIMEOUT, script_cache_size: DEFAULT_SCRIPT_CACHE_SIZE, max_concurrent_txs: DEFAULT_MAX_CONCURRENT_TXS, + priority_accounts: Vec::new(), max_notes_per_tx: DEFAULT_MAX_NOTES_PER_TX, max_note_attempts: DEFAULT_MAX_NOTE_ATTEMPTS, max_block_count: DEFAULT_MAX_BLOCK_COUNT, @@ -302,6 +310,13 @@ impl NtxBuilderConfig { self } + /// Sets the network accounts served before every other account. + #[must_use] + pub fn with_priority_accounts(mut self, accounts: Vec) -> Self { + self.priority_accounts = accounts; + self + } + /// Sets the maximum number of notes per transaction. /// /// # Panics @@ -508,6 +523,24 @@ impl NtxBuilderConfig { }, }; - Ok(Scheduler::new(ctx, self.max_concurrent_txs, self.tx_expiration_delta)) + // With as many priority accounts as slots, the other accounts only run when a priority + // account has nothing to do. + if !self.priority_accounts.is_empty() + && self.priority_accounts.len() >= self.max_concurrent_txs + { + warn!( + target: LOG_TARGET, + "priority accounts can occupy every attempt slot; raise --max-concurrent-txs", + account.ids.count = self.priority_accounts.len(), + ntx_builder.max_concurrent_txs = self.max_concurrent_txs + ); + } + + Ok(Scheduler::new( + ctx, + self.max_concurrent_txs, + self.tx_expiration_delta, + self.priority_accounts.clone(), + )) } } diff --git a/bin/ntx-builder/src/scheduler.rs b/bin/ntx-builder/src/scheduler.rs index e63933688..7b143a3bb 100644 --- a/bin/ntx-builder/src/scheduler.rs +++ b/bin/ntx-builder/src/scheduler.rs @@ -67,6 +67,10 @@ pub struct Scheduler { /// 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, + + /// Accounts served before every other account. Each one still holds at most one slot, because + /// an account with a running attempt is excluded from selection. + priority_accounts: Vec, } impl Scheduler { @@ -74,6 +78,7 @@ impl Scheduler { ctx: AttemptContext, max_concurrent_txs: usize, tx_expiration_delta: NonZeroU16, + priority_accounts: Vec, ) -> Self { Self { ctx, @@ -82,6 +87,7 @@ impl Scheduler { in_flight: HashMap::new(), max_concurrent_txs, tx_expiration_delta, + priority_accounts, } } @@ -112,9 +118,16 @@ impl Scheduler { let ready = self .ctx .db - .ready_accounts(self.ctx.config.max_note_attempts, block_num, busy, free) + .ready_accounts( + self.ctx.config.max_note_attempts, + block_num, + busy, + self.priority_accounts.clone(), + free, + ) .await .context("failed to query accounts ready for a transaction attempt")?; + let dispatched = ready.len(); for account_id in ready { let ctx = self.ctx.clone(); let chain = chain.clone(); @@ -128,6 +141,18 @@ impl Scheduler { ); } + // Neither the attempt concurrency nor the submitted-but-uncommitted backlog is otherwise + // observable from outside the process. + debug!( + target: LOG_TARGET, + "network transaction pipeline", + reference_block.number = block_num, + attempt.dispatched.count = dispatched, + attempt.running.count = self.running.len(), + transaction.in_flight.count = self.in_flight.len(), + attempt.slots.count = self.max_concurrent_txs + ); + Ok(()) } @@ -318,7 +343,7 @@ mod tests { 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) + (Scheduler::new(ctx, 4, NonZeroU16::new(30).unwrap(), Vec::new()), db, dir) } /// Effects for a block carrying nothing but its header. @@ -433,6 +458,7 @@ mod tests { 30, BlockNumber::from(1), vec![in_flight_account], + Vec::new(), scheduler.max_concurrent_txs, ) .await @@ -453,7 +479,10 @@ mod tests { .unwrap(); assert!( - db.ready_accounts(30, BlockNumber::from(1), vec![], 4).await.unwrap().is_empty(), + db.ready_accounts(30, BlockNumber::from(1), vec![], vec![], 4) + .await + .unwrap() + .is_empty(), "a note targeting an account with no committed state is not dispatchable", ); @@ -466,7 +495,7 @@ mod tests { .unwrap(); assert_eq!( - db.ready_accounts(30, BlockNumber::from(1), vec![], 4).await.unwrap(), + db.ready_accounts(30, BlockNumber::from(1), vec![], vec![], 4).await.unwrap(), vec![account_id], "the account becomes dispatchable once its state is committed", );