Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(sequencer): remove unwrap from app utilized mempool logic #1772

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(sequencer): remove unwrap from app utilized mempool logic #1772

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 1 addition & 5 deletions crates/astria-sequencer/src/app/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -582,11 +582,7 @@ impl App {
let mut current_tx_group = Group::BundleableGeneral;

// get copy of transactions to execute from mempool
let pending_txs = self
.mempool
.builder_queue(&self.state)
.await
.expect("failed to fetch pending transactions");
let pending_txs = self.mempool.builder_queue(&self.state).await;
Copy link
Member

@SuperFluffy SuperFluffy Oct 31, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rereading the surrounding code, I would have honestly been fine with just wrap_err'oring this (without the other changes).

Tracing the error back up it eventually still panics (execute_transactions_prepare_proposal -> prepare_proposal -> handle_prepare_proposal -> handle_request -> run). But we could fix that by returning the error back up to the server, which would be cleaner (not this PR but followup).


let mut unused_count = pending_txs.len();
for (tx_hash, tx) in pending_txs {
Expand Down
2 changes: 1 addition & 1 deletion crates/astria-sequencer/src/mempool/benchmarks.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ fn builder_queue<T: MempoolSize>(bencher: divan::Bencher) {
.with_inputs(|| init_mempool::<T>())
.bench_values(move |mempool| {
runtime.block_on(async {
mempool.builder_queue(&mock_state).await.unwrap();
mempool.builder_queue(&mock_state).await;
});
});
}
Expand Down
37 changes: 8 additions & 29 deletions crates/astria-sequencer/src/mempool/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,7 +294,7 @@ impl Mempool {
pub(crate) async fn builder_queue<S: accounts::StateReadExt>(
&self,
state: &S,
) -> Result<Vec<([u8; 32], Arc<Transaction>)>> {
) -> Vec<([u8; 32], Arc<Transaction>)> {
self.pending.read().await.builder_queue(state).await
}

Expand Down Expand Up @@ -649,10 +649,7 @@ mod tests {

// grab building queue, should return transactions [1,2] since [0] was below and [4] is
// gapped
let builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let builder_queue = mempool.builder_queue(&mock_state).await;

// see contains first two transactions that should be pending
assert_eq!(builder_queue[0].1.nonce(), 1, "nonce should be one");
Expand Down Expand Up @@ -682,10 +679,7 @@ mod tests {
assert_eq!(mempool.len().await, 1);

// see transaction [4] properly promoted
let mut builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let mut builder_queue = mempool.builder_queue(&mock_state).await;
let (_, returned_tx) = builder_queue.pop().expect("should return last transaction");
assert_eq!(returned_tx.nonce(), 4, "nonce should be four");
}
Expand Down Expand Up @@ -730,10 +724,7 @@ mod tests {
1,
);

let builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let builder_queue = mempool.builder_queue(&mock_state).await;
assert_eq!(
builder_queue.len(),
1,
Expand All @@ -752,10 +743,7 @@ mod tests {
mempool.run_maintenance(&mock_state, false).await;

// see builder queue now contains them
let builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let builder_queue = mempool.builder_queue(&mock_state).await;
assert_eq!(
builder_queue.len(),
3,
Expand Down Expand Up @@ -804,10 +792,7 @@ mod tests {
1,
);

let builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let builder_queue = mempool.builder_queue(&mock_state).await;
assert_eq!(
builder_queue.len(),
4,
Expand All @@ -824,10 +809,7 @@ mod tests {
mempool.run_maintenance(&mock_state, false).await;

// see builder queue now contains single transactions
let builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let builder_queue = mempool.builder_queue(&mock_state).await;
assert_eq!(
builder_queue.len(),
1,
Expand All @@ -847,10 +829,7 @@ mod tests {

mempool.run_maintenance(&mock_state, false).await;

let builder_queue = mempool
.builder_queue(&mock_state)
.await
.expect("failed to get builder queue");
let builder_queue = mempool.builder_queue(&mock_state).await;
assert_eq!(
builder_queue.len(),
3,
Expand Down
42 changes: 26 additions & 16 deletions crates/astria-sequencer/src/mempool/transactions_container.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ use astria_core::{
use astria_eyre::eyre::{
eyre,
Result,
WrapErr as _,
};
use tokio::time::{
Duration,
Expand Down Expand Up @@ -773,7 +772,7 @@ impl PendingTransactions {
pub(super) async fn builder_queue<S: accounts::StateReadExt>(
&self,
state: &S,
) -> Result<Vec<([u8; 32], Arc<Transaction>)>> {
) -> Vec<([u8; 32], Arc<Transaction>)> {
// Used to hold the values in Vec for sorting.
struct QueueEntry {
tx: Arc<Transaction>,
Expand All @@ -784,10 +783,27 @@ impl PendingTransactions {
let mut queue = Vec::with_capacity(self.len());
// Add all transactions to the queue.
for (address, account_txs) in &self.txs {
let current_account_nonce = state
.get_account_nonce(address)
.await
.wrap_err("failed to fetch account nonce for builder queue")?;
let current_account_nonce = match state.get_account_nonce(address).await {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if this function returns an error, it's pretty severe (either we can't read from state, and the app will crash soon anyways, or we can't deserialize the storage value which would have caused an error earlier) so i think expect()ing here is fine

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

expecting is not a replacement for error handling. expecting should only be used for when system invariants are so badly violated that you can't recover from them at all, which is not the case here.

Ok(nonce) => nonce,
Err(error) => {
error!(address = %telemetry::display::base64(address), "failed to fetch account nonce for builder queue: {error:#}");
// use the lowest nonce in pending as the current nonce, this should be the
// same value as the nonce returned by the state. the pending queue shouldn't
// be empty, if it is we can continue to the next account
Comment on lines +790 to +792
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a good point. I think we should actually just use this technique rather than passing in the state. That would simplify this method and allow it to not be async.

We could change constructing the current account nonce to:

let Some(current_account_nonce) = account_txs.current_account_nonce() else {
    error!(
        address = %telemetry::display::base64(address),
        "pending queue is empty during builder queue step"
    );
    continue;
};

and add this to the impl for PendingTransactionsForAccount:

fn current_account_nonce(&self) -> Option<u32> {
    self.txs.first_key_value().map(|(nonce, _)| *nonce)
}

if let Some(nonce) = account_txs
.txs()
.values()
.map(TimemarkedTransaction::nonce)
.min()
{
nonce
} else {
error!(address = %telemetry::display::base64(address), "pending queue is empty during builder queue step");
continue;
}
}
};

for ttx in account_txs.txs.values() {
let priority = match ttx.priority(current_account_nonce) {
Ok(priority) => priority,
Expand All @@ -811,11 +827,11 @@ impl PendingTransactions {
// Sort the queue and return the relevant data. Note that the sorted queue will be ordered
// from lowest to highest priority, so we need to reverse the order before returning.
queue.sort_unstable_by_key(|entry| entry.priority);
Ok(queue
queue
.into_iter()
.rev()
.map(|entry| (entry.tx_hash, entry.tx))
.collect())
.collect()
}
}

Expand Down Expand Up @@ -2011,10 +2027,7 @@ mod tests {
);

// get builder queue
let builder_queue = pending_txs
.builder_queue(&mock_state)
.await
.expect("building builders queue should work");
let builder_queue = pending_txs.builder_queue(&mock_state).await;
assert_eq!(
builder_queue.len(),
3,
Expand Down Expand Up @@ -2291,10 +2304,7 @@ mod tests {

// get the builder queue
// note: the account nonces are set to zero when not initialized in the mock state
let builder_queue = pending_txs
.builder_queue(&mock_state)
.await
.expect("building builders queue should work");
let builder_queue = pending_txs.builder_queue(&mock_state).await;

// check that the transactions are in the expected order
let (first_tx_hash, _) = builder_queue[0];
Expand Down
Loading