feat(ntx-builder): serve priority network accounts first

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<AccountId>,
+
         /// 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> {
+    AccountId::from_hex(value).map_err(Into::into)
+}

bin/ntx-builder/src/db/mod.rs

@@ -125,11 +125,12 @@ impl NtxDbReader {
         max_note_attempts: usize,
         block_num: BlockNumber,
         busy: Vec<AccountId>,
+        priority: Vec<AccountId>,
         limit: usize,
     ) -> Result<Vec<AccountId>, 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
     }

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<Vec<AccountId>, 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::<AccountId>(0),
     )
 }

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

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),

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<AccountId>,
+
     /// Maximum number of network notes a single transaction is allowed to consume. Sponsorship
     /// notes count against this budget.
     pub max_notes_per_tx: NonZeroUsize,
@@ -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<AccountId>) -> 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(),
+        ))
     }
 }