sui_core/
admission_queue.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use crate::authority::authority_per_epoch_store::AuthorityPerEpochStore;
5use crate::consensus_adapter::ConsensusAdapter;
6use arc_swap::ArcSwap;
7use mysten_common::debug_fatal;
8use mysten_metrics::{COUNT_BUCKETS, spawn_monitored_task};
9use prometheus::{
10    Histogram, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Registry,
11    register_histogram_vec_with_registry, register_histogram_with_registry,
12    register_int_counter_vec_with_registry, register_int_counter_with_registry,
13    register_int_gauge_vec_with_registry, register_int_gauge_with_registry,
14};
15use std::collections::{BTreeMap, HashMap, VecDeque};
16use std::net::IpAddr;
17use std::sync::atomic::{AtomicUsize, Ordering};
18use std::sync::{Arc, Mutex};
19use std::time::{Duration, Instant};
20use sui_macros::handle_fail_point_if;
21use sui_network::tonic;
22use sui_types::error::{SuiError, SuiErrorKind, SuiResult};
23use sui_types::messages_consensus::{
24    ConsensusPosition, ConsensusTransaction, ConsensusTransactionKey,
25};
26use tokio::sync::{mpsc, oneshot};
27use tracing::debug;
28
29/// A transaction (or soft bundle) waiting in the admission queue for consensus submission.
30pub struct QueueEntry {
31    pub gas_price: u64,
32    pub transactions: Vec<ConsensusTransaction>,
33    pub position_sender: oneshot::Sender<Result<Vec<ConsensusPosition>, tonic::Status>>,
34    pub submitter_client_addr: Option<IpAddr>,
35    pub enqueue_time: Instant,
36}
37
38/// What `PriorityAdmissionQueue::pop_batch_while` does with an examined entry.
39pub enum PopAction {
40    /// Pop the entry into the included partition.
41    Include,
42    /// Pop the entry into the excluded partition.
43    Exclude,
44    /// Leave the entry queued and stop iterating.
45    Stop,
46}
47
48pub trait AdmissionQueueEntry {
49    fn gas_price(&self) -> u64;
50    fn transaction_keys(&self) -> impl Iterator<Item = ConsensusTransactionKey>;
51    fn notify_evicted(self, min_gas_price: u64);
52    fn notify_rejected(self, min_gas_price: u64);
53}
54
55/// Outcome of [`PriorityAdmissionQueue::try_insert`]. Any displaced entry is
56/// carried here unnotified so that the caller can deliver the notification
57/// outside of any held locks.
58#[must_use = "call `notify` to deliver the outcome to the displaced entry"]
59pub struct InsertOutcome<E> {
60    outcome: Option<Outcome<E>>,
61    created_at: &'static std::panic::Location<'static>,
62}
63
64enum Outcome<E> {
65    Inserted {
66        /// False if an entry sharing one of the transaction keys was already queued.
67        newly_inserted: bool,
68        /// The lowest-priced entry evicted to make room, if any.
69        evicted: Option<E>,
70        /// The inserted entry's gas price.
71        gas_price: u64,
72    },
73    /// Rejected: the queue is full and `min_gas_price` was not met.
74    Rejected { entry: E, min_gas_price: u64 },
75}
76
77impl<E: AdmissionQueueEntry> InsertOutcome<E> {
78    #[track_caller]
79    fn new(outcome: Outcome<E>) -> Self {
80        Self {
81            outcome: Some(outcome),
82            created_at: std::panic::Location::caller(),
83        }
84    }
85
86    /// Notifies the displaced entry, if any, then reports the insert result.
87    /// `try_insert(e).notify()` is equivalent to `insert(e)`.
88    pub fn notify(mut self) -> SuiResult<bool> {
89        match self
90            .outcome
91            .take()
92            .expect("outcome is pending until notified")
93        {
94            Outcome::Inserted {
95                newly_inserted,
96                evicted,
97                gas_price,
98            } => {
99                if let Some(evicted) = evicted {
100                    evicted.notify_evicted(gas_price);
101                }
102                Ok(newly_inserted)
103            }
104            Outcome::Rejected {
105                entry,
106                min_gas_price,
107            } => {
108                entry.notify_rejected(min_gas_price);
109                Err(
110                    SuiErrorKind::TransactionRejectedDueToOutbiddingDuringCongestion {
111                        min_gas_price,
112                    }
113                    .into(),
114                )
115            }
116        }
117    }
118}
119
120impl<E> Drop for InsertOutcome<E> {
121    fn drop(&mut self) {
122        if self.outcome.is_some() && !std::thread::panicking() {
123            debug_fatal!(
124                "InsertOutcome from try_insert at {} dropped without notify",
125                self.created_at
126            );
127        }
128    }
129}
130
131impl AdmissionQueueEntry for QueueEntry {
132    fn gas_price(&self) -> u64 {
133        self.gas_price
134    }
135
136    fn transaction_keys(&self) -> impl Iterator<Item = ConsensusTransactionKey> {
137        self.transactions.iter().map(ConsensusTransaction::key)
138    }
139
140    fn notify_evicted(self, min_gas_price: u64) {
141        let _ = self
142            .position_sender
143            .send(Err(tonic::Status::from(SuiError::from(
144                SuiErrorKind::TransactionRejectedDueToOutbiddingDuringCongestion { min_gas_price },
145            ))));
146    }
147
148    fn notify_rejected(self, _min_gas_price: u64) {
149        // Nothing to do here in push mode. The rejected caller receives the
150        // outbid error via the insert result instead.
151    }
152}
153
154impl QueueEntry {
155    #[cfg(test)]
156    pub fn new_for_test(
157        gas_price: u64,
158        position_sender: oneshot::Sender<Result<Vec<ConsensusPosition>, tonic::Status>>,
159    ) -> Self {
160        Self {
161            gas_price,
162            transactions: vec![],
163            position_sender,
164            submitter_client_addr: None,
165            enqueue_time: Instant::now(),
166        }
167    }
168}
169
170/// Prometheus metrics for the admission queue.
171pub struct AdmissionQueueMetrics {
172    // Pull mode intentionally reuses these admission metrics for its user lane because
173    // both modes implement the same admission policy and are mutually exclusive on a
174    // validator. This preserves dashboard continuity when switching modes.
175    pub queue_depth: IntGauge,
176    pub queue_wait_latency: HistogramVec,
177    pub evictions: IntCounter,
178    pub rejections: IntCounter,
179    pub duplicate_inserts: IntCounter,
180
181    pub pool_depth: IntGaugeVec,
182    pub pool_bytes: IntGaugeVec,
183    pub pool_taken_per_proposal: Histogram,
184    pub pool_requeued_on_dropped_ack: IntCounter,
185    pub pool_gc_notified: IntCounter,
186    pub pool_waiting_inserts: IntGauge,
187    pub pool_already_processed: IntCounterVec,
188    pub pool_commit_latency: HistogramVec,
189    pub pool_abandoned: IntCounterVec,
190}
191
192impl AdmissionQueueMetrics {
193    pub fn new(registry: &Registry) -> Self {
194        Self {
195            queue_depth: register_int_gauge_with_registry!(
196                "admission_queue_depth",
197                "Current number of entries in the admission priority queue",
198                registry,
199            )
200            .unwrap(),
201            queue_wait_latency: register_histogram_vec_with_registry!(
202                "admission_queue_wait_latency",
203                "Time a submission spends waiting in the admission queue or transaction pool before being drained or proposed",
204                &["lane"],
205                mysten_metrics::SUBSECOND_LATENCY_SEC_BUCKETS.to_vec(),
206                registry,
207            )
208            .unwrap(),
209            evictions: register_int_counter_with_registry!(
210                "admission_queue_evictions",
211                "Number of entries evicted from the admission queue by higher gas price transactions",
212                registry,
213            )
214            .unwrap(),
215            rejections: register_int_counter_with_registry!(
216                "admission_queue_rejections",
217                "Number of transactions rejected because the queue was full and their gas price was too low",
218                registry,
219            )
220            .unwrap(),
221            duplicate_inserts: register_int_counter_with_registry!(
222                "admission_queue_duplicate_inserts",
223                "Transactions admitted to the queue whose ConsensusTransactionKey duplicated an entry already present. Tallied as spam for DoS protection.",
224                registry,
225            )
226            .unwrap(),
227            pool_depth: register_int_gauge_vec_with_registry!(
228                "consensus_transaction_pool_depth",
229                "Current number of entries in each consensus transaction pool lane",
230                &["lane"],
231                registry,
232            )
233            .unwrap(),
234            pool_bytes: register_int_gauge_vec_with_registry!(
235                "consensus_transaction_pool_bytes",
236                "Current serialized transaction bytes in each consensus transaction pool lane",
237                &["lane"],
238                registry,
239            )
240            .unwrap(),
241            pool_taken_per_proposal: register_histogram_with_registry!(
242                "consensus_transaction_pool_taken_per_proposal",
243                "Transactions taken from the consensus transaction pool per proposal",
244                COUNT_BUCKETS.to_vec(),
245                registry,
246            )
247            .unwrap(),
248            pool_requeued_on_dropped_ack: register_int_counter_with_registry!(
249                "consensus_transaction_pool_requeued_on_dropped_ack",
250                "Entries requeued because a proposal acknowledgement was dropped",
251                registry,
252            )
253            .unwrap(),
254            pool_gc_notified: register_int_counter_with_registry!(
255                "consensus_transaction_pool_gc_notified",
256                "Block-status subscribers notified that their block was garbage collected",
257                registry,
258            )
259            .unwrap(),
260            pool_waiting_inserts: register_int_gauge_with_registry!(
261                "consensus_transaction_pool_waiting_inserts",
262                "Pool submissions waiting for the matching epoch pool to become available",
263                registry,
264            )
265            .unwrap(),
266            pool_already_processed: register_int_counter_vec_with_registry!(
267                "consensus_transaction_pool_already_processed",
268                "User submissions not proposed because they were already processed elsewhere, by the stage that detected it and the path that processed them",
269                &["stage", "method"],
270                registry,
271            )
272            .unwrap(),
273            pool_commit_latency: register_histogram_vec_with_registry!(
274                "consensus_transaction_pool_commit_latency",
275                "Time from insert into the transaction pool to commit of the block that proposed the entry",
276                &["lane"],
277                mysten_metrics::LATENCY_SEC_BUCKETS.to_vec(),
278                registry,
279            )
280            .unwrap(),
281            pool_abandoned: register_int_counter_vec_with_registry!(
282                "consensus_transaction_pool_abandoned",
283                "Pool entries dropped at proposal time because their submitter stopped waiting",
284                &["lane"],
285                registry,
286            )
287            .unwrap(),
288        }
289    }
290
291    pub fn new_for_tests() -> Self {
292        Self::new(&Registry::new())
293    }
294}
295
296/// Bounded priority queue that orders transactions by gas price. Uses a BTreeMap
297/// for efficient access at both ends: lowest gas price (for eviction) and highest
298/// gas price (for draining to consensus). Entries at the same gas price are FIFO.
299pub struct PriorityAdmissionQueue<E: AdmissionQueueEntry> {
300    capacity: usize,
301    map: BTreeMap<u64, VecDeque<E>>,
302    /// Number of queue entries per transaction key, for duplicate detection.
303    queued_keys: HashMap<ConsensusTransactionKey, u32>,
304    total_len: usize,
305    metrics: Arc<AdmissionQueueMetrics>,
306}
307
308impl<E: AdmissionQueueEntry> PriorityAdmissionQueue<E> {
309    pub fn new(capacity: usize, metrics: Arc<AdmissionQueueMetrics>) -> Self {
310        Self {
311            capacity,
312            map: BTreeMap::new(),
313            queued_keys: HashMap::new(),
314            total_len: 0,
315            metrics,
316        }
317    }
318
319    pub fn len(&self) -> usize {
320        self.total_len
321    }
322
323    pub fn min_gas_price(&self) -> Option<u64> {
324        self.map.first_key_value().map(|(&k, _)| k)
325    }
326
327    /// On success, returns `Ok(true)` or `Ok(false)` to indicate whether the
328    /// value was newly inserted. Returns `Err` if the queue was full and the
329    /// tx's gas price was not high enough to evict an existing entry.
330    pub fn insert(&mut self, entry: E) -> SuiResult<bool> {
331        self.try_insert(entry).notify()
332    }
333
334    /// Same as `insert`, without notifying displaced entries. The evicted entry (on
335    /// success) or the refused entry itself (on rejection) is handed back in the
336    /// outcome for the caller to notify manually.
337    #[track_caller]
338    pub fn try_insert(&mut self, entry: E) -> InsertOutcome<E> {
339        let keys: Vec<_> = entry.transaction_keys().collect();
340        let newly_inserted = !keys.iter().any(|k| self.queued_keys.contains_key(k));
341        if !newly_inserted {
342            self.metrics.duplicate_inserts.inc();
343        }
344
345        let gas_price = entry.gas_price();
346        if self.total_len < self.capacity {
347            self.push_entry(entry, keys, false);
348            return InsertOutcome::new(Outcome::Inserted {
349                newly_inserted,
350                evicted: None,
351                gas_price,
352            });
353        }
354
355        let min_gas_price = self.min_gas_price().unwrap();
356        if gas_price > min_gas_price {
357            let evicted = self.evict_lowest();
358            self.push_entry(entry, keys, false);
359            self.metrics.evictions.inc();
360            return InsertOutcome::new(Outcome::Inserted {
361                newly_inserted,
362                evicted: Some(evicted),
363                gas_price,
364            });
365        }
366
367        self.metrics.rejections.inc();
368        InsertOutcome::new(Outcome::Rejected {
369            entry,
370            min_gas_price,
371        })
372    }
373
374    /// Pop up to `count` entries, highest gas price first.
375    /// Within the same gas price, entries are returned in FIFO order.
376    pub fn pop_batch(&mut self, count: usize) -> Vec<E> {
377        let mut remaining = count;
378        self.pop_batch_while(|_| {
379            if remaining == 0 {
380                PopAction::Stop
381            } else {
382                remaining -= 1;
383                PopAction::Include
384            }
385        })
386        .0
387    }
388
389    pub fn into_entries(mut self) -> Vec<E> {
390        let len = self.len();
391        self.pop_batch(len)
392    }
393
394    /// Pop entries highest gas price first (FIFO within a price level) until
395    /// `action` returns `Stop` or the queue is empty. The entry that stopped
396    /// iteration stays queued. Popped entries are returned partitioned into
397    /// those the callback chose to `Include` and those to `Exclude`.
398    pub fn pop_batch_while(
399        &mut self,
400        mut action: impl FnMut(&mut E) -> PopAction,
401    ) -> (Vec<E>, Vec<E>) {
402        let mut included = Vec::new();
403        let mut excluded = Vec::new();
404        'levels: while let Some(mut last) = self.map.last_entry() {
405            let deque = last.get_mut();
406            while let Some(entry) = deque.front_mut() {
407                let action = action(entry);
408                if matches!(action, PopAction::Stop) {
409                    break 'levels;
410                }
411                let entry = deque.pop_front().expect("front entry must exist");
412                self.total_len -= 1;
413                match action {
414                    PopAction::Include => included.push(entry),
415                    PopAction::Exclude => excluded.push(entry),
416                    PopAction::Stop => unreachable!("Stop breaks out above"),
417                }
418            }
419            last.remove();
420        }
421        for entry in included.iter().chain(&excluded) {
422            self.remove_keys(entry);
423        }
424        self.metrics.queue_depth.set(self.total_len as i64);
425        (included, excluded)
426    }
427
428    pub fn is_empty(&self) -> bool {
429        self.total_len == 0
430    }
431
432    pub fn reinsert_front(&mut self, entry: E) {
433        let keys: Vec<_> = entry.transaction_keys().collect();
434        self.push_entry(entry, keys, true);
435    }
436
437    fn push_entry(&mut self, entry: E, keys: Vec<ConsensusTransactionKey>, front: bool) {
438        for key in keys {
439            *self.queued_keys.entry(key).or_insert(0) += 1;
440        }
441        let level = self.map.entry(entry.gas_price()).or_default();
442        if front {
443            level.push_front(entry);
444        } else {
445            level.push_back(entry);
446        }
447        self.total_len += 1;
448        self.metrics.queue_depth.set(self.total_len as i64);
449    }
450
451    fn evict_lowest(&mut self) -> E {
452        let evicted = {
453            let mut first = self
454                .map
455                .first_entry()
456                .expect("evict_lowest called on empty queue");
457            let deque = first.get_mut();
458            let evicted = deque.pop_front().unwrap();
459            if deque.is_empty() {
460                first.remove();
461            }
462            evicted
463        };
464        self.remove_keys(&evicted);
465        self.total_len -= 1;
466        evicted
467    }
468
469    fn remove_keys(&mut self, entry: &E) {
470        for key in entry.transaction_keys() {
471            let std::collections::hash_map::Entry::Occupied(mut slot) = self.queued_keys.entry(key)
472            else {
473                debug_fatal!("remove_keys on absent key");
474                continue;
475            };
476            *slot.get_mut() -= 1;
477            if *slot.get() == 0 {
478                slot.remove();
479            }
480        }
481    }
482}
483
484/// Command sent from RPC handlers to the admission queue actor via mpsc channel.
485struct InsertCommand {
486    entry: QueueEntry,
487    response: oneshot::Sender<SuiResult<bool>>,
488}
489
490/// Cloneable handle for submitting transactions to the admission queue actor.
491/// Held by RPC handlers; the actor runs in a separate spawned task.
492#[derive(Clone)]
493pub struct AdmissionQueueHandle {
494    sender: mpsc::Sender<InsertCommand>,
495    /// The moment the queue last submitted an entry to consensus.
496    last_drain: Arc<Mutex<Instant>>,
497    queue_depth: Arc<AtomicUsize>,
498    failover_timeout: Duration,
499}
500
501impl AdmissionQueueHandle {
502    /// Returns true if the queue has been non-empty for longer than
503    /// `failover_timeout` without any drain to consensus. Callers should
504    /// bypass the queue entirely when this is true.
505    pub fn failover_tripped(&self) -> bool {
506        if self.queue_depth.load(Ordering::Relaxed) == 0 {
507            return false;
508        }
509        self.last_drain.lock().unwrap().elapsed() > self.failover_timeout
510    }
511
512    /// Returns `(position_receiver, newly_inserted)` on admission. Returns `Err` on outbid
513    /// rejection.
514    pub async fn try_insert(
515        &self,
516        gas_price: u64,
517        transactions: Vec<ConsensusTransaction>,
518        submitter_client_addr: Option<IpAddr>,
519    ) -> SuiResult<(
520        oneshot::Receiver<Result<Vec<ConsensusPosition>, tonic::Status>>,
521        bool,
522    )> {
523        let (position_tx, position_rx) = oneshot::channel();
524        let entry = QueueEntry {
525            gas_price,
526            transactions,
527            position_sender: position_tx,
528            submitter_client_addr,
529            enqueue_time: Instant::now(),
530        };
531
532        let (resp_tx, resp_rx) = oneshot::channel();
533        let cmd = InsertCommand {
534            entry,
535            response: resp_tx,
536        };
537
538        self.sender
539            .send(cmd)
540            .await
541            .map_err(|_| SuiError::from(SuiErrorKind::TooManyTransactionsPendingConsensus))?;
542
543        let newly_inserted = resp_rx
544            .await
545            .map_err(|_| SuiError::from(SuiErrorKind::TooManyTransactionsPendingConsensus))??;
546
547        Ok((position_rx, newly_inserted))
548    }
549}
550
551/// Manages the lifecycle of per-epoch admission queue actors.
552/// Holds immutable config and shared metrics; call `spawn()` each epoch
553/// with the new epoch store to create a fresh actor and handle.
554pub struct AdmissionQueueManager {
555    capacity: usize,
556    failover_timeout: Duration,
557    metrics: Arc<AdmissionQueueMetrics>,
558    consensus_adapter: Arc<ConsensusAdapter>,
559    slot_freed_notify: Arc<tokio::sync::Notify>,
560}
561
562impl AdmissionQueueManager {
563    pub fn new(
564        consensus_adapter: Arc<ConsensusAdapter>,
565        metrics: Arc<AdmissionQueueMetrics>,
566        capacity_fraction: f64,
567        failover_timeout: Duration,
568        slot_freed_notify: Arc<tokio::sync::Notify>,
569    ) -> Self {
570        let max_pending = consensus_adapter.max_pending_transactions();
571        let capacity = (max_pending as f64 * capacity_fraction) as usize;
572        assert!(
573            capacity > 0,
574            "admission_queue_capacity_fraction ({capacity_fraction}) * max_pending_transactions ({max_pending}) must be > 0"
575        );
576        Self {
577            capacity,
578            failover_timeout,
579            metrics,
580            consensus_adapter,
581            slot_freed_notify,
582        }
583    }
584
585    pub fn new_for_tests(
586        consensus_adapter: Arc<ConsensusAdapter>,
587        slot_freed_notify: Arc<tokio::sync::Notify>,
588    ) -> Self {
589        Self {
590            capacity: 10_000,
591            failover_timeout: Duration::from_secs(30),
592            metrics: Arc::new(AdmissionQueueMetrics::new_for_tests()),
593            consensus_adapter,
594            slot_freed_notify,
595        }
596    }
597
598    pub fn metrics(&self) -> &Arc<AdmissionQueueMetrics> {
599        &self.metrics
600    }
601
602    /// Spawns a new per-epoch admission queue actor and returns a handle.
603    /// The previous actor shuts down when its handle is dropped.
604    pub fn spawn(&self, epoch_store: Arc<AuthorityPerEpochStore>) -> AdmissionQueueHandle {
605        let last_drain = Arc::new(Mutex::new(Instant::now()));
606        let queue_depth = Arc::new(AtomicUsize::new(0));
607
608        let (sender, receiver) = mpsc::channel(self.capacity.max(1024));
609
610        let event_loop = AdmissionQueueEventLoop {
611            receiver,
612            queue: PriorityAdmissionQueue::new(self.capacity, self.metrics.clone()),
613            consensus_adapter: self.consensus_adapter.clone(),
614            slot_freed_notify: self.slot_freed_notify.clone(),
615            epoch_store,
616            last_drain: last_drain.clone(),
617            queue_depth: queue_depth.clone(),
618            last_published_depth: 0,
619        };
620        spawn_monitored_task!(event_loop.run());
621
622        AdmissionQueueHandle {
623            sender,
624            last_drain,
625            queue_depth,
626            failover_timeout: self.failover_timeout,
627        }
628    }
629}
630
631/// Shared handle to a live admission queue. Holds the manager (for spawning a
632/// fresh per-epoch actor on reconfig), the per-epoch `ArcSwap` handle, and the
633/// cached (config-derived) bypass threshold. Cloned cheaply by `Arc`; passed
634/// both to `ValidatorService` (for hot-path routing) and through
635/// `ValidatorComponents` (for epoch rotation).
636#[derive(Clone)]
637pub struct AdmissionQueueContext {
638    manager: Arc<AdmissionQueueManager>,
639    swap: Arc<ArcSwap<AdmissionQueueHandle>>,
640}
641
642impl AdmissionQueueContext {
643    pub fn spawn(
644        manager: Arc<AdmissionQueueManager>,
645        epoch_store: Arc<AuthorityPerEpochStore>,
646    ) -> Self {
647        let initial_handle = manager.spawn(epoch_store);
648        let swap = Arc::new(ArcSwap::new(Arc::new(initial_handle)));
649        Self { manager, swap }
650    }
651
652    /// Spawns a new per-epoch actor and atomically replaces the current handle.
653    /// The old actor shuts down when its handle is dropped.
654    pub fn rotate_for_epoch(&self, epoch_store: Arc<AuthorityPerEpochStore>) {
655        self.swap.store(Arc::new(self.manager.spawn(epoch_store)));
656    }
657
658    pub(crate) fn load(&self) -> arc_swap::Guard<Arc<AdmissionQueueHandle>> {
659        self.swap.load()
660    }
661}
662
663/// Per-epoch event loop that owns the priority queue and drains entries
664/// to consensus as capacity becomes available.
665struct AdmissionQueueEventLoop {
666    receiver: mpsc::Receiver<InsertCommand>,
667    queue: PriorityAdmissionQueue<QueueEntry>,
668    consensus_adapter: Arc<ConsensusAdapter>,
669    slot_freed_notify: Arc<tokio::sync::Notify>,
670    epoch_store: Arc<AuthorityPerEpochStore>,
671    last_drain: Arc<Mutex<Instant>>,
672    queue_depth: Arc<AtomicUsize>,
673    last_published_depth: usize,
674}
675
676impl AdmissionQueueEventLoop {
677    pub async fn run(mut self) {
678        loop {
679            self.process_pending_inserts();
680            self.publish_queue_depth();
681
682            if !handle_fail_point_if("admission_queue_disable_drain")
683                && !self.queue.is_empty()
684                && self.has_consensus_capacity()
685            {
686                self.drain_batch();
687                self.publish_queue_depth();
688                continue;
689            }
690
691            if self.queue.is_empty() {
692                // Nothing to drain — just wait for a new insert.
693                match self.receiver.recv().await {
694                    Some(cmd) => self.handle_insert(cmd),
695                    None => {
696                        debug!("Admission queue actor shutting down");
697                        break;
698                    }
699                }
700                continue;
701            }
702
703            // Queue has entries but consensus is at capacity. Wait for either
704            // a new insert or a freed inflight slot.
705            // Register the notified future BEFORE re-checking capacity to avoid
706            // missing notifications.
707            let notify = self.slot_freed_notify.clone();
708            let slot_freed = notify.notified();
709            tokio::pin!(slot_freed);
710
711            self.process_pending_inserts();
712            if !handle_fail_point_if("admission_queue_disable_drain")
713                && !self.queue.is_empty()
714                && self.has_consensus_capacity()
715            {
716                continue;
717            }
718
719            tokio::select! {
720                biased;
721
722                result = self.receiver.recv() => {
723                    match result {
724                        Some(cmd) => self.handle_insert(cmd),
725                        None => {
726                            debug!("Admission queue actor shutting down");
727                            break;
728                        }
729                    }
730                }
731
732                _ = &mut slot_freed => {}
733            }
734        }
735    }
736
737    fn publish_queue_depth(&mut self) {
738        let len = self.queue.len();
739        if len != self.last_published_depth {
740            self.queue_depth.store(len, Ordering::Relaxed);
741            self.last_published_depth = len;
742        }
743    }
744
745    fn process_pending_inserts(&mut self) {
746        while let Ok(cmd) = self.receiver.try_recv() {
747            self.handle_insert(cmd);
748        }
749    }
750
751    fn has_consensus_capacity(&self) -> bool {
752        self.consensus_adapter.num_inflight_transactions()
753            < u64::try_from(self.consensus_adapter.max_pending_transactions()).unwrap()
754    }
755
756    fn drain_batch(&mut self) {
757        let max_pending = u64::try_from(self.consensus_adapter.max_pending_transactions()).unwrap();
758        let available =
759            max_pending.saturating_sub(self.consensus_adapter.num_inflight_transactions());
760        let entries = self.queue.pop_batch(usize::try_from(available).unwrap());
761        if entries.is_empty() {
762            return;
763        }
764        for entry in entries {
765            self.queue
766                .metrics
767                .queue_wait_latency
768                .with_label_values(&["user"])
769                .observe(entry.enqueue_time.elapsed().as_secs_f64());
770            let adapter = self.consensus_adapter.clone();
771            let es = self.epoch_store.clone();
772            spawn_monitored_task!(submit_queue_entry(entry, adapter, es));
773        }
774        *self.last_drain.lock().unwrap() = Instant::now();
775    }
776
777    fn handle_insert(&mut self, cmd: InsertCommand) {
778        let _ = cmd.response.send(self.queue.insert(cmd.entry));
779    }
780}
781
782async fn submit_queue_entry(
783    entry: QueueEntry,
784    consensus_adapter: Arc<ConsensusAdapter>,
785    epoch_store: Arc<AuthorityPerEpochStore>,
786) {
787    let _ = entry.position_sender.send(
788        consensus_adapter
789            .submit_and_get_positions(
790                entry.transactions,
791                &epoch_store,
792                entry.submitter_client_addr,
793            )
794            .await
795            .map_err(tonic::Status::from),
796    );
797}
798
799#[cfg(test)]
800mod tests {
801    use super::*;
802
803    fn make_test_entry(
804        gas_price: u64,
805    ) -> (
806        QueueEntry,
807        oneshot::Receiver<Result<Vec<ConsensusPosition>, tonic::Status>>,
808    ) {
809        let (tx, rx) = oneshot::channel();
810        (QueueEntry::new_for_test(gas_price, tx), rx)
811    }
812
813    #[cfg(debug_assertions)]
814    #[test]
815    #[should_panic(expected = "dropped without notify")]
816    fn dropped_insert_outcome_without_notify_panics() {
817        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
818        let mut q = PriorityAdmissionQueue::new(1, metrics);
819        let (entry, _rx) = make_test_entry(100);
820        let _ = q.try_insert(entry);
821    }
822
823    #[test]
824    fn try_insert_notify_matches_insert() {
825        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
826        let mut q = PriorityAdmissionQueue::new(1, metrics);
827        let (low, mut low_rx) = make_test_entry(100);
828        let (high, _) = make_test_entry(200);
829        let (rejected, mut rejected_rx) = make_test_entry(50);
830
831        assert!(q.try_insert(low).notify().unwrap());
832        // Eviction is delivered only by `notify`.
833        let outcome = q.try_insert(high);
834        assert!(low_rx.try_recv().is_err());
835        assert!(outcome.notify().unwrap());
836        assert!(matches!(
837            low_rx.try_recv(),
838            Ok(Err(status)) if status.message().contains("minimum gas price required: 200")
839        ));
840
841        assert!(matches!(
842            q.try_insert(rejected).notify().unwrap_err().as_inner(),
843            SuiErrorKind::TransactionRejectedDueToOutbiddingDuringCongestion { min_gas_price: 200 }
844        ));
845        assert_eq!(q.len(), 1);
846        assert!(rejected_rx.try_recv().is_err());
847    }
848
849    fn build_queue(capacity: usize, gas_prices: &[u64]) -> PriorityAdmissionQueue<QueueEntry> {
850        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
851        let mut q = PriorityAdmissionQueue::new(capacity, metrics);
852        for &gp in gas_prices {
853            let (entry, _) = make_test_entry(gp);
854            q.insert(entry).unwrap();
855        }
856        q
857    }
858
859    #[test]
860    fn test_insert_within_capacity() {
861        let q = build_queue(3, &[100, 200, 50]);
862        assert_eq!(q.len(), 3);
863    }
864
865    #[test]
866    fn test_pop_batch_while_partitions_and_stops() {
867        let mut q = build_queue(10, &[100, 200, 200, 50, 75]);
868
869        let (included, excluded) = q.pop_batch_while(|entry| match entry.gas_price {
870            price if price < 80 => PopAction::Stop,
871            price if price < 150 => PopAction::Exclude,
872            _ => PopAction::Include,
873        });
874        assert_eq!(
875            included.iter().map(|e| e.gas_price).collect::<Vec<_>>(),
876            vec![200, 200]
877        );
878        assert_eq!(
879            excluded.iter().map(|e| e.gas_price).collect::<Vec<_>>(),
880            vec![100]
881        );
882        // The entry that stopped iteration stays queued, as does everything behind it.
883        assert_eq!(q.len(), 2);
884        assert_eq!(q.min_gas_price(), Some(50));
885        let (rest, _) = q.pop_batch_while(|_| PopAction::Include);
886        assert_eq!(rest.len(), 2);
887        assert!(q.is_empty());
888    }
889
890    #[test]
891    fn test_eviction_when_full() {
892        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
893        let mut q = PriorityAdmissionQueue::new(2, metrics);
894
895        let (e1, mut r1) = make_test_entry(100);
896        let (e2, _) = make_test_entry(200);
897        let (e3, _) = make_test_entry(300);
898
899        q.insert(e1).unwrap();
900        q.insert(e2).unwrap();
901        assert_eq!(q.len(), 2);
902
903        assert!(q.insert(e3).is_ok());
904        assert_eq!(q.len(), 2);
905        // Evicted entry's caller receives an explicit outbid error.
906        let r1_result = r1.try_recv().expect("evicted entry must be signalled");
907        assert!(matches!(r1_result, Err(ref status) if status.message().contains("outbid")));
908    }
909
910    #[test]
911    fn test_rejection_when_full_and_low_price() {
912        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
913        let mut q = PriorityAdmissionQueue::new(2, metrics);
914
915        let (e1, _) = make_test_entry(100);
916        let (e2, _) = make_test_entry(200);
917        let (e3, mut r3) = make_test_entry(50);
918
919        q.insert(e1).unwrap();
920        q.insert(e2).unwrap();
921
922        assert!(matches!(
923            q.insert(e3).unwrap_err().as_inner(),
924            SuiErrorKind::TransactionRejectedDueToOutbiddingDuringCongestion { min_gas_price: 100 }
925        ));
926        assert_eq!(q.len(), 2);
927        assert!(r3.try_recv().is_err());
928    }
929
930    #[test]
931    fn test_pop_batch() {
932        let mut q = build_queue(5, &[100, 300, 200]);
933        let batch = q.pop_batch(2);
934        assert_eq!(batch.len(), 2);
935        assert_eq!(q.len(), 1);
936    }
937
938    #[test]
939    fn test_min_gas_price() {
940        let q = build_queue(5, &[200, 100, 300]);
941        assert_eq!(q.min_gas_price(), Some(100));
942    }
943
944    #[test]
945    fn test_gasless_tx_evicted_first() {
946        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
947        let mut q = PriorityAdmissionQueue::new(2, metrics);
948
949        let (gasless, mut r_gasless) = make_test_entry(0);
950        let (normal, _) = make_test_entry(1000);
951        let (high, _) = make_test_entry(2000);
952
953        q.insert(gasless).unwrap();
954        q.insert(normal).unwrap();
955
956        assert!(q.insert(high).is_ok());
957        let gasless_result = r_gasless
958            .try_recv()
959            .expect("evicted gasless entry must be signalled");
960        assert!(matches!(gasless_result, Err(ref status) if status.message().contains("outbid")));
961        assert_eq!(q.min_gas_price(), Some(1000));
962    }
963
964    #[test]
965    fn test_pop_batch_returns_highest_gas_price_first() {
966        let mut q = build_queue(5, &[100, 500, 200, 400, 300]);
967        let batch = q.pop_batch(5);
968        let gas_prices: Vec<u64> = batch.iter().map(|e| e.gas_price).collect();
969        assert_eq!(gas_prices, vec![500, 400, 300, 200, 100]);
970    }
971
972    #[test]
973    fn test_equal_gas_price_rejected_when_full() {
974        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
975        let mut q = PriorityAdmissionQueue::new(1, metrics);
976
977        let (e1, _) = make_test_entry(100);
978        let (e2, _) = make_test_entry(100);
979
980        q.insert(e1).unwrap();
981        assert!(matches!(
982            q.insert(e2).unwrap_err().as_inner(),
983            SuiErrorKind::TransactionRejectedDueToOutbiddingDuringCongestion { min_gas_price: 100 }
984        ));
985    }
986
987    fn make_dup_entry(
988        gas_price: u64,
989        tx: ConsensusTransaction,
990    ) -> (
991        QueueEntry,
992        oneshot::Receiver<Result<Vec<ConsensusPosition>, tonic::Status>>,
993    ) {
994        let (position_tx, position_rx) = oneshot::channel();
995        let entry = QueueEntry {
996            gas_price,
997            transactions: vec![tx],
998            position_sender: position_tx,
999            submitter_client_addr: None,
1000            enqueue_time: Instant::now(),
1001        };
1002        (entry, position_rx)
1003    }
1004
1005    #[test]
1006    fn test_duplicate_transaction_admitted_and_flagged() {
1007        use sui_types::base_types::AuthorityName;
1008
1009        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
1010        let mut q = PriorityAdmissionQueue::new(10, metrics);
1011
1012        let tx = ConsensusTransaction::new_end_of_publish(AuthorityName::ZERO);
1013
1014        let (entry1, _rx1) = make_dup_entry(100, tx.clone());
1015        assert!(q.insert(entry1).unwrap());
1016        assert_eq!(q.len(), 1);
1017
1018        // Same transaction again — admitted, but flagged as not-fresh so the
1019        // RPC layer can tally it as spam for DoS protection.
1020        let (entry2, _rx2) = make_dup_entry(100, tx.clone());
1021        assert!(!q.insert(entry2).unwrap());
1022        assert_eq!(q.len(), 2);
1023    }
1024
1025    #[test]
1026    fn test_duplicate_key_counter_decrements_on_pop() {
1027        use sui_types::base_types::AuthorityName;
1028
1029        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
1030        let mut q = PriorityAdmissionQueue::new(10, metrics);
1031
1032        let tx = ConsensusTransaction::new_end_of_publish(AuthorityName::ZERO);
1033
1034        // Insert two copies of the same tx.
1035        let (entry1, _rx1) = make_dup_entry(100, tx.clone());
1036        q.insert(entry1).unwrap();
1037        let (entry2, _rx2) = make_dup_entry(100, tx.clone());
1038        assert!(!q.insert(entry2).unwrap());
1039
1040        // Pop one copy. The key's counter should drop to 1 — a fresh insert
1041        // should still be flagged as not-fresh against the remaining copy.
1042        let batch = q.pop_batch(1);
1043        assert_eq!(batch.len(), 1);
1044        let (entry3, _rx3) = make_dup_entry(100, tx.clone());
1045        assert!(!q.insert(entry3).unwrap());
1046        assert_eq!(q.len(), 2);
1047
1048        // Drain both remaining entries. The counter should hit 0 and the key
1049        // should be removed — a subsequent insert is fresh again.
1050        let _ = q.pop_batch(q.len());
1051        assert!(q.is_empty());
1052        let (entry4, _rx4) = make_dup_entry(100, tx);
1053        assert!(q.insert(entry4).unwrap());
1054    }
1055
1056    #[test]
1057    fn test_duplicate_key_counter_decrements_on_evict() {
1058        use sui_types::base_types::AuthorityName;
1059
1060        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
1061        let mut q = PriorityAdmissionQueue::new(2, metrics);
1062
1063        let tx = ConsensusTransaction::new_end_of_publish(AuthorityName::ZERO);
1064
1065        // Fill queue with two copies of `tx` at price 100.
1066        let (entry1, _rx1) = make_dup_entry(100, tx.clone());
1067        q.insert(entry1).unwrap();
1068        let (entry2, _rx2) = make_dup_entry(100, tx.clone());
1069        q.insert(entry2).unwrap();
1070        assert_eq!(q.len(), 2);
1071
1072        // Evict one dup with a higher-priced non-dup.
1073        let (filler, _) = make_test_entry(200);
1074        q.insert(filler).unwrap();
1075        assert_eq!(q.len(), 2);
1076
1077        // Evict the remaining dup with another non-dup. After both dups are
1078        // evicted, the counter should hit 0 and re-inserting `tx` is not a
1079        // duplicate.
1080        let (filler2, _) = make_test_entry(300);
1081        q.insert(filler2).unwrap();
1082
1083        let (entry3, _rx3) = make_dup_entry(500, tx);
1084        assert!(q.insert(entry3).unwrap());
1085    }
1086
1087    #[test]
1088    fn test_duplicate_key_counter_restored_on_reinsert_front() {
1089        use sui_types::base_types::AuthorityName;
1090
1091        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
1092        let mut q = PriorityAdmissionQueue::new(10, metrics);
1093
1094        let tx = ConsensusTransaction::new_end_of_publish(AuthorityName::ZERO);
1095
1096        let (entry1, _rx1) = make_dup_entry(100, tx.clone());
1097        q.insert(entry1).unwrap();
1098
1099        // Popping removes the key; reinserting the popped entry (the dropped-proposal
1100        // requeue path) must restore it, so a copy of `tx` is flagged as a duplicate.
1101        let popped = q.pop_batch(1).pop().unwrap();
1102        q.reinsert_front(popped);
1103        assert_eq!(q.len(), 1);
1104        let (entry2, _rx2) = make_dup_entry(100, tx.clone());
1105        assert!(!q.insert(entry2).unwrap());
1106
1107        // Draining everything zeroes the counter and the same tx is fresh again.
1108        let _ = q.pop_batch(q.len());
1109        assert!(q.is_empty());
1110        let (entry3, _rx3) = make_dup_entry(100, tx);
1111        assert!(q.insert(entry3).unwrap());
1112    }
1113
1114    #[tokio::test]
1115    async fn test_actor_shuts_down_when_handle_dropped() {
1116        use crate::authority::test_authority_builder::TestAuthorityBuilder;
1117        use crate::checkpoints::CheckpointStore;
1118        use crate::consensus_adapter::ConsensusAdapterMetrics;
1119        use crate::mysticeti_adapter::LazyMysticetiClient;
1120        use sui_types::base_types::AuthorityName;
1121
1122        let state = TestAuthorityBuilder::new().build().await;
1123        let epoch_store = state.epoch_store_for_testing().clone();
1124        let consensus_adapter = Arc::new(ConsensusAdapter::new(
1125            Arc::new(LazyMysticetiClient::new()),
1126            CheckpointStore::new_for_tests(),
1127            AuthorityName::ZERO,
1128            100_000,
1129            100_000,
1130            ConsensusAdapterMetrics::new_test(),
1131            Arc::new(tokio::sync::Notify::new()),
1132        ));
1133
1134        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
1135        let (sender, receiver) = mpsc::channel(100);
1136        let slot_freed_notify = Arc::new(tokio::sync::Notify::new());
1137
1138        let event_loop = AdmissionQueueEventLoop {
1139            receiver,
1140            queue: PriorityAdmissionQueue::new(100, metrics),
1141            consensus_adapter,
1142            slot_freed_notify,
1143            epoch_store,
1144            last_drain: Arc::new(Mutex::new(Instant::now())),
1145            queue_depth: Arc::new(AtomicUsize::new(0)),
1146            last_published_depth: 0,
1147        };
1148
1149        let handle = tokio::spawn(event_loop.run());
1150
1151        // Drop the sender — this closes the channel.
1152        drop(sender);
1153
1154        // The actor should exit promptly.
1155        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
1156            .await
1157            .expect("actor did not shut down within timeout")
1158            .expect("actor task panicked");
1159    }
1160
1161    async fn build_consensus_adapter(
1162        max_pending_transactions: usize,
1163    ) -> (
1164        Arc<ConsensusAdapter>,
1165        Arc<AuthorityPerEpochStore>,
1166        Arc<tokio::sync::Notify>,
1167    ) {
1168        use crate::authority::test_authority_builder::TestAuthorityBuilder;
1169        use crate::checkpoints::CheckpointStore;
1170        use crate::consensus_adapter::ConsensusAdapterMetrics;
1171        use crate::mysticeti_adapter::LazyMysticetiClient;
1172        use sui_types::base_types::AuthorityName;
1173
1174        let state = TestAuthorityBuilder::new().build().await;
1175        let epoch_store = state.epoch_store_for_testing().clone();
1176        let slot_freed_notify = Arc::new(tokio::sync::Notify::new());
1177        let adapter = Arc::new(ConsensusAdapter::new(
1178            Arc::new(LazyMysticetiClient::new()),
1179            CheckpointStore::new_for_tests(),
1180            AuthorityName::ZERO,
1181            max_pending_transactions,
1182            100_000,
1183            ConsensusAdapterMetrics::new_test(),
1184            slot_freed_notify.clone(),
1185        ));
1186        (adapter, epoch_store, slot_freed_notify)
1187    }
1188
1189    #[tokio::test]
1190    async fn test_failover_tripped_when_actor_stalls() {
1191        // Construct a handle with a tiny failover window and no running actor.
1192        // Failover requires queue_depth > 0, so simulate a non-empty queue.
1193        let handle = AdmissionQueueHandle {
1194            sender: mpsc::channel(1).0,
1195            last_drain: Arc::new(Mutex::new(Instant::now())),
1196            queue_depth: Arc::new(AtomicUsize::new(1)),
1197            failover_timeout: Duration::from_millis(10),
1198        };
1199        assert!(!handle.failover_tripped());
1200        tokio::time::sleep(Duration::from_millis(30)).await;
1201        assert!(handle.failover_tripped());
1202
1203        // An empty queue is never a failover, even if last_drain is stale.
1204        handle.queue_depth.store(0, Ordering::Relaxed);
1205        assert!(!handle.failover_tripped());
1206    }
1207
1208    #[tokio::test]
1209    async fn test_idle_actor_does_not_trip_failover() {
1210        // A healthy actor with an empty queue must never trip failover, even
1211        // after long idle periods while blocked on `receiver.recv()`.
1212        let (adapter, epoch_store, notify) = build_consensus_adapter(100_000).await;
1213        let manager = AdmissionQueueManager::new(
1214            adapter,
1215            Arc::new(AdmissionQueueMetrics::new_for_tests()),
1216            0.5,
1217            Duration::from_millis(10),
1218            notify,
1219        );
1220        let handle = manager.spawn(epoch_store);
1221        tokio::time::sleep(Duration::from_millis(50)).await;
1222        assert!(
1223            !handle.failover_tripped(),
1224            "idle actor with empty queue must not trip failover"
1225        );
1226    }
1227
1228    /// If `drain_batch` is entered but consensus has zero slots available
1229    /// (the inflight count raced past `max_pending_transactions` between the
1230    /// `has_consensus_capacity` check and the read inside `drain_batch`), no
1231    /// entries are popped and `last_drain` must NOT advance — otherwise a
1232    /// truly stuck drainer would be hidden from the failover check.
1233    #[tokio::test]
1234    async fn test_drain_batch_does_not_bump_last_drain_when_no_slots() {
1235        let (adapter, epoch_store, notify) = build_consensus_adapter(0).await;
1236        let metrics = Arc::new(AdmissionQueueMetrics::new_for_tests());
1237        let (_sender, receiver) = mpsc::channel(10);
1238
1239        let mut queue = PriorityAdmissionQueue::new(10, metrics.clone());
1240        let (entry, _rx) = make_test_entry(100);
1241        assert!(queue.insert(entry).is_ok());
1242        assert_eq!(queue.len(), 1);
1243
1244        let last_drain = Arc::new(Mutex::new(Instant::now()));
1245        let before = *last_drain.lock().unwrap();
1246
1247        let mut event_loop = AdmissionQueueEventLoop {
1248            receiver,
1249            queue,
1250            consensus_adapter: adapter,
1251            slot_freed_notify: notify,
1252            epoch_store,
1253            last_drain: last_drain.clone(),
1254            queue_depth: Arc::new(AtomicUsize::new(0)),
1255            last_published_depth: 0,
1256        };
1257
1258        // Sleep so that if drain_batch erroneously stamps Instant::now() the
1259        // stored value would differ from `before`.
1260        tokio::time::sleep(Duration::from_millis(20)).await;
1261
1262        event_loop.drain_batch();
1263
1264        assert_eq!(event_loop.queue.len(), 1, "no entries should be drained");
1265        assert_eq!(
1266            *last_drain.lock().unwrap(),
1267            before,
1268            "last_drain must not advance when drain_batch drained nothing"
1269        );
1270    }
1271}
sui_core/admission_queue.rs

sui_core/
admission_queue.rs
