Skip to content

perf: pack withdrawal queue bounds into one slot #693

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

Draft
0xalpharush wants to merge 1 commit into
base: main
Choose a base branch
from
+224 −183
Draft

perf: pack withdrawal queue bounds into one slot #693

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
5 changes: 3 additions & 2 deletions crates/contracts/src/precompiles/zone_portal.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,8 +180,9 @@ crate::sol! {
function blockHash() external view returns (bytes32);
function currentDepositQueueHash() external view returns (bytes32);
function lastSyncedTempoBlockNumber() external view returns (uint64);
function withdrawalQueueHead() external view returns (uint256);
function withdrawalQueueTail() external view returns (uint256);
function withdrawalQueueHead() external view returns (uint64);
function withdrawalQueueTail() external view returns (uint64);
function withdrawalQueueBounds() external view returns (uint64 head, uint64 tail);
function withdrawalQueueSlot(uint256 physicalSlot) external view returns (bytes32);
function genesisTempoBlockNumber() external view returns (uint64);
function calculateDepositFee() external view returns (uint128 fee);
Expand Down
8 changes: 4 additions & 4 deletions crates/node/assets/zone-dev-genesis.json
View file
Open in desktop

Large diffs are not rendered by default.

5 changes: 2 additions & 3 deletions crates/node/tests/it/restart_e2e.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,8 @@ async fn portal_block_hash(
/// Read the portal's withdrawal queue head and tail.
async fn portal_queue_state(l1: &L1TestNode, portal_address: Address) -> eyre::Result<(u64, u64)> {
let portal = ZonePortal::new(portal_address, l1.provider());
let head: U256 = portal.withdrawalQueueHead().call().await?;
let tail: U256 = portal.withdrawalQueueTail().call().await?;
Ok((head.to::<u64>(), tail.to::<u64>()))
let bounds = portal.withdrawalQueueBounds().call().await?;
Ok((bounds.head, bounds.tail))
}

/// Count `BatchSubmitted` events on the portal.
Expand Down
17 changes: 7 additions & 10 deletions crates/sequencer/src/monitor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -938,6 +938,7 @@ mod tests {
use alloy_network::Network;
use alloy_primitives::{Bytes, U256};
use alloy_rpc_types_eth::{Block, Header};
use alloy_sol_types::SolValue;
use alloy_transport::mock::Asserter;
use tempo_alloy::rpc::TempoHeaderResponse;
use tempo_primitives::TempoHeader;
Expand Down Expand Up @@ -965,8 +966,8 @@ mod tests {
Bytes::copy_from_slice(value.as_slice())
}

fn abi_encode_u64(value: u64) -> Bytes {
Bytes::copy_from_slice(&U256::from(value).to_be_bytes::<32>())
fn abi_encode_queue_bounds(head: u64, tail: u64) -> Bytes {
Bytes::from((head, tail).abi_encode_params())
}

fn mock_block(hash: B256, number: u64) -> <TempoNetwork as Network>::BlockResponse {
Expand Down Expand Up @@ -1072,8 +1073,7 @@ mod tests {
let confirmed_deposit_hash = B256::repeat_byte(0x33);

l1.push_success(&abi_encode_b256(portal_hash));
l1.push_success(&abi_encode_u64(7));
l1.push_success(&abi_encode_u64(7));
l1.push_success(&abi_encode_queue_bounds(7, 7));

zone.push_success(&Some(mock_block(portal_hash, confirmed_zone_block)));
zone.push_success(&abi_encode_b256(confirmed_deposit_hash));
Expand All @@ -1098,8 +1098,7 @@ mod tests {
let confirmed_deposit_hash = B256::repeat_byte(0x33);

l1.push_success(&abi_encode_b256(portal_hash));
l1.push_success(&abi_encode_u64(7));
l1.push_success(&abi_encode_u64(7));
l1.push_success(&abi_encode_queue_bounds(7, 7));

zone.push_success(&Some(mock_block(portal_hash, confirmed_zone_block)));
zone.push_success(&abi_encode_b256(confirmed_deposit_hash));
Expand Down Expand Up @@ -1141,8 +1140,7 @@ mod tests {
let confirmed_deposit_hash = B256::repeat_byte(0x33);

l1.push_success(&abi_encode_b256(portal_hash));
l1.push_failure_msg("head read failed");
l1.push_failure_msg("tail read failed");
l1.push_failure_msg("queue bounds read failed");

zone.push_success(&Some(mock_block(portal_hash, confirmed_zone_block)));
zone.push_success(&abi_encode_b256(confirmed_deposit_hash));
Expand Down Expand Up @@ -1185,8 +1183,7 @@ mod tests {

l1.push_success(&abi_encode_b256(portal_hash));
l1.push_success(&abi_encode_b256(portal_hash));
l1.push_success(&abi_encode_u64(7));
l1.push_success(&abi_encode_u64(7));
l1.push_success(&abi_encode_queue_bounds(7, 7));

zone.push_success(&Some(mock_block(portal_hash, confirmed_zone_block)));
zone.push_success(&abi_encode_b256(confirmed_deposit_hash));
Expand Down
60 changes: 37 additions & 23 deletions crates/sequencer/src/settlement.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -485,11 +485,7 @@ impl BatchSubmitter {

/// Read the current logical withdrawal queue tail from the ZonePortal on L1.
pub async fn read_portal_withdrawal_queue_tail(&self) -> Result<u64> {
let tail = self.portal.withdrawalQueueTail().call().await?;
let tail: u64 = tail
.try_into()
.map_err(|_| eyre::eyre!("withdrawal queue tail overflow"))?;
Ok(tail)
Ok(self.portal.withdrawalQueueTail().call().await?)
}

/// Read the current withdrawal batch index from the ZonePortal on L1.
Expand All @@ -499,22 +495,21 @@ impl BatchSubmitter {

/// Read the current withdrawal queue head from the ZonePortal on L1.
pub async fn read_portal_withdrawal_queue_head(&self) -> Result<u64> {
let head = self.portal.withdrawalQueueHead().call().await?;
let head: u64 = head
.try_into()
.map_err(|_| eyre::eyre!("withdrawal queue head overflow"))?;
Ok(head)
Ok(self.portal.withdrawalQueueHead().call().await?)
}

/// Read the portal's withdrawal queue bounds from one L1 state snapshot.
async fn read_portal_withdrawal_queue_bounds(&self) -> Result<(u64, u64)> {
let bounds = self.portal.withdrawalQueueBounds().call().await?;
Ok((bounds.head, bounds.tail))
}

/// Check if the withdrawal queue has capacity for another batch.
///
/// The portal uses a ring buffer with 100 slots. Returns an error if the
/// queue is full (`tail - head >= 100`).
pub async fn check_withdrawal_queue_capacity(&self) -> Result<()> {
let (head, tail) = tokio::try_join!(
self.read_portal_withdrawal_queue_head(),
self.read_portal_withdrawal_queue_tail(),
)?;
let (head, tail) = self.read_portal_withdrawal_queue_bounds().await?;
if tail.saturating_sub(head) >= WITHDRAWAL_QUEUE_CAPACITY {
return Err(eyre::eyre!(
"withdrawal queue full ({} pending slots, capacity {})",
Expand Down Expand Up @@ -551,10 +546,7 @@ impl BatchSubmitter {
outbox_address: Address,
) -> Result<BTreeMap<u64, Vec<abi::Withdrawal>>> {
// Step 1: read pending slot range from the L1 portal.
let (head, tail) = tokio::try_join!(
self.read_portal_withdrawal_queue_head(),
self.read_portal_withdrawal_queue_tail(),
)?;
let (head, tail) = self.read_portal_withdrawal_queue_bounds().await?;

if head >= tail {
info!(head, tail, "No pending withdrawals to restore");
Expand Down Expand Up @@ -622,10 +614,7 @@ impl BatchSubmitter {
.await?;

// Guard: verify the queue didn't change during the multi-RPC replay.
let (head2, tail2) = tokio::try_join!(
self.read_portal_withdrawal_queue_head(),
self.read_portal_withdrawal_queue_tail(),
)?;
let (head2, tail2) = self.read_portal_withdrawal_queue_bounds().await?;

if head2 != head || tail2 != tail {
eyre::bail!(
Expand Down Expand Up @@ -1079,7 +1068,10 @@ pub(crate) struct ZoneBlockSnapshot {
mod tests {
use super::*;
use crate::abi;
use alloy_primitives::{B256, address};
use alloy_primitives::{B256, Bytes, address};
use alloy_provider::ProviderBuilder;
use alloy_sol_types::SolValue;
use alloy_transport::mock::Asserter;

fn test_withdrawal(to: Address, amount: u128) -> abi::Withdrawal {
abi::Withdrawal {
Expand All @@ -1096,6 +1088,10 @@ mod tests {
}
}

fn abi_encode_queue_bounds(head: u64, tail: u64) -> Bytes {
Bytes::from((head, tail).abi_encode_params())
}

#[test]
fn batch_anchor_config_validates_effective_window() {
let config = BatchAnchorConfig::new(10, 4).unwrap();
Expand All @@ -1108,6 +1104,24 @@ mod tests {
assert!(BatchAnchorConfig::new(10, 11).is_err());
}

#[tokio::test]
async fn withdrawal_queue_bounds_use_one_contract_call() {
let asserter = Asserter::new();
let provider = ProviderBuilder::new_with_network::<TempoNetwork>()
.connect_mocked_client(asserter.clone())
.erased();
let submitter = BatchSubmitter::new(Address::ZERO, provider, 0);

asserter.push_success(&abi_encode_queue_bounds(7, 19));
let bounds = submitter
.read_portal_withdrawal_queue_bounds()
.await
.unwrap();

assert_eq!(bounds, (7, 19));
assert!(asserter.read_q().is_empty());
}

#[test]
fn find_offset_no_withdrawals_processed() {
let w0 = test_withdrawal(address!("0x0000000000000000000000000000000000000001"), 100);
Expand Down
68 changes: 31 additions & 37 deletions crates/sequencer/src/withdrawals.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,12 @@ impl WithdrawalProcessor {
}
}

/// Read the portal's withdrawal queue bounds from one L1 state snapshot.
async fn read_queue_bounds(&self) -> eyre::Result<(u64, u64)> {
let bounds = self.portal.withdrawalQueueBounds().call().await?;
Ok((bounds.head, bounds.tail))
}

/// Run the processor loop. This method never returns under normal operation.
///
/// Waits for a notification from the batch submitter (or a fallback timeout) before
Expand Down Expand Up @@ -346,42 +352,36 @@ impl WithdrawalProcessor {
async fn process_queue(&mut self) -> eyre::Result<()> {
// loop through all the slots
loop {
let head_call = self.portal.withdrawalQueueHead();
let tail_call = self.portal.withdrawalQueueTail();
let (head, tail): (U256, U256) = tokio::try_join!(head_call.call(), tail_call.call())?;
let (head, tail) = self.read_queue_bounds().await?;

let head_val: u64 = head.try_into().map_err(|_| eyre::eyre!("head overflow"))?;
let tail_val: u64 = tail.try_into().map_err(|_| eyre::eyre!("tail overflow"))?;
let StoreSnapshot {
batch_count: store_batch_count,
first_slot: store_first_slot,
last_slot: store_last_slot,
prev_slot: prev_store_slot,
next_slot: next_store_slot,
withdrawals,
} = self.capture_store_snapshot(head_val);
self.record_queue_metrics(head_val, tail_val, store_batch_count);
} = self.capture_store_snapshot(head);
self.record_queue_metrics(head, tail, store_batch_count);

if head_val == tail_val {
if head == tail {
debug!("Withdrawal queue empty, nothing to process");
return Ok(());
}

let pending_slots = tail_val - head_val;
let pending_slots = tail - head;
info!(
head = head_val,
tail = tail_val,
pending_slots,
"Withdrawal queue has pending slots"
head,
tail, pending_slots, "Withdrawal queue has pending slots"
);

let withdrawals = match withdrawals {
Some(w) if !w.is_empty() => w,
_ => {
self.repair_notify.notify_one();
warn!(
slot = head_val,
tail = tail_val,
slot = head,
tail,
pending_slots,
store_batches = store_batch_count,
store_first_slot,
Expand All @@ -398,23 +398,20 @@ impl WithdrawalProcessor {
// withdrawals the portal has already consumed.
let slot_hash = self
.portal
.withdrawalQueueSlot(U256::from(head_val % WITHDRAWAL_QUEUE_CAPACITY))
.withdrawalQueueSlot(U256::from(head % WITHDRAWAL_QUEUE_CAPACITY))
.call()
.await?;

if slot_hash == EMPTY_SENTINEL {
// The slot was fully consumed and head advanced between our reads.
// Re-check on the next cycle.
debug!(
slot = head_val,
"Head slot already consumed; skipping cycle"
);
debug!(slot = head, "Head slot already consumed; skipping cycle");
return Ok(());
}

let Some(offset) = find_processed_offset(&withdrawals, slot_hash) else {
error!(
slot = head_val,
slot = head,
on_chain_slot_hash = %slot_hash,
store_queue_hash = %abi::Withdrawal::queue_hash(&withdrawals),
"Store data does not match the head slot's on-chain hash; requesting repair"
Expand All @@ -425,7 +422,7 @@ impl WithdrawalProcessor {

if offset > 0 {
info!(
slot = head_val,
slot = head,
processed = offset,
remaining = withdrawals.len() - offset,
"Trimmed withdrawals already consumed by the portal"
Expand All @@ -437,15 +434,15 @@ impl WithdrawalProcessor {
// Defensive: queue_hash never produces B256::ZERO for a pending head
// slot, but if it happens drop the stale batch and wait for the portal.
warn!(
slot = head_val,
slot = head,
"Head slot fully processed but head not advanced"
);
self.store.lock().remove_batch(head_val);
self.store.lock().remove_batch(head);
return Ok(());
}

info!(
slot = head_val,
slot = head,
count = remaining.len(),
"Processing withdrawal batch"
);
Expand All @@ -455,7 +452,7 @@ impl WithdrawalProcessor {
let remaining_queue = compute_remaining_queue(remaining, i + 1);
let outcome = self
.submit_and_confirm(
head_val,
head,
offset + i,
remaining.len(),
withdrawal,
Expand All @@ -482,10 +479,10 @@ impl WithdrawalProcessor {

// All withdrawals in this slot confirmed — safe to remove. Continue
// the loop to drain any further pending slots.
self.store.lock().remove_batch(head_val);
self.store.lock().remove_batch(head);

info!(
slot = head_val,
slot = head,
count = remaining.len(),
"Batch fully processed and removed from store"
);
Expand Down Expand Up @@ -676,7 +673,7 @@ pub fn spawn_withdrawal_processor(
mod tests {
use super::*;
use crate::abi::EMPTY_SENTINEL;
use alloy_primitives::{Bytes, U256, address, keccak256};
use alloy_primitives::{Bytes, address, keccak256};
use alloy_provider::{Provider, ProviderBuilder};
use alloy_sol_types::SolValue;
use alloy_transport::mock::Asserter;
Expand All @@ -689,8 +686,8 @@ mod tests {
.erased()
}

fn abi_encode_u64(value: u64) -> Bytes {
Bytes::copy_from_slice(&U256::from(value).to_be_bytes::<32>())
fn abi_encode_queue_bounds(head: u64, tail: u64) -> Bytes {
Bytes::from((head, tail).abi_encode_params())
}

fn test_withdrawal(to: Address, amount: u128) -> abi::Withdrawal {
Expand Down Expand Up @@ -912,8 +909,7 @@ mod tests {
#[tokio::test]
async fn process_queue_requests_monitor_resync_when_head_slot_missing() {
let l1 = Asserter::new();
l1.push_success(&abi_encode_u64(51));
l1.push_success(&abi_encode_u64(71));
l1.push_success(&abi_encode_queue_bounds(51, 71));

let repair_notify = Arc::new(Notify::new());
let mut processor = test_processor(
Expand All @@ -934,8 +930,7 @@ mod tests {
async fn process_queue_requests_repair_when_store_data_mismatches_slot_hash() {
let l1 = Asserter::new();
// head = 5, tail = 6, slot hash that matches no suffix of the stored batch.
l1.push_success(&abi_encode_u64(5));
l1.push_success(&abi_encode_u64(6));
l1.push_success(&abi_encode_queue_bounds(5, 6));
l1.push_success(&abi_encode_b256(B256::repeat_byte(0xde)));

let store = SharedWithdrawalStore::new();
Expand Down Expand Up @@ -963,8 +958,7 @@ mod tests {
let l1 = Asserter::new();
// head = 5, tail = 6, slot already contains EMPTY_SENTINEL (head advanced
// between our head read and the slot read).
l1.push_success(&abi_encode_u64(5));
l1.push_success(&abi_encode_u64(6));
l1.push_success(&abi_encode_queue_bounds(5, 6));
l1.push_success(&abi_encode_b256(EMPTY_SENTINEL));

let store = SharedWithdrawalStore::new();
Expand Down
Loading
Loading