diff --git a/ledger/src/reader.rs b/ledger/src/reader.rs index 1ce6d382..569f6bac 100644 --- a/ledger/src/reader.rs +++ b/ledger/src/reader.rs @@ -229,7 +229,7 @@ impl LedgerReader { break; } let Some((superblock, span)) = position else { return Ok(None) }; - self.populate_block_response(&superblock, request, span).map(Some) + self.block_response(&superblock, request, span).map(Some) } /// Reads blocks and indexed transaction statuses in descending slot order. @@ -331,7 +331,7 @@ impl LedgerReader { } /// Builds the block response requested by the caller. - fn populate_block_response( + fn block_response( &mut self, superblock: &Superblock, request: &BlockPayload, @@ -339,7 +339,6 @@ impl LedgerReader { ) -> Result { let BlockParams { slot, details } = request.params; let block = self.block_entry(superblock, slot, span)?; - if matches!(details, BlockDetails::None) { return Ok(BlockResponse::Bare(block)); } @@ -371,7 +370,7 @@ impl LedgerReader { signatures, })) } - BlockDetails::None => unreachable!(), + BlockDetails::None => Ok(BlockResponse::Bare(block)), } } diff --git a/replicator/src/client.rs b/replicator/src/client.rs index 765c9b59..0b30af70 100644 --- a/replicator/src/client.rs +++ b/replicator/src/client.rs @@ -14,10 +14,10 @@ use engine::{ use flume::{Receiver, Sender, TrySendError}; use ledger::{ Superblock, - schema::{Block, OwnedBlockstoreEntry, blockstore}, + schema::{Block, OwnedBlockstoreEntry, SuperblockSeal, blockstore}, }; use nucleus::{ - KB, + KB, Slot, ledger::{ACCOUNTSDB_SNAPSHOT_FILE, BlockstorePosition}, runtime::BarrierHandle, shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, @@ -68,8 +68,12 @@ enum ReplicationMessage { Unverified(Vec>), /// Transactions verified by Ingest while Control was occupied. Verified(Vec), - /// Non-transaction entry fenced behind every preceding batch. - Entry(OwnedBlockstoreEntry), + /// Block boundary fenced behind every preceding batch. + Block(Block), + /// Superblock seal fenced behind every preceding batch. + Superblock(SuperblockSeal), + /// Volatile-state reset fenced behind every preceding batch. + Reset(Slot), } /// Why connection-scoped Ingest stopped without a terminal replication error. @@ -203,14 +207,32 @@ impl ReplicationClient { self.schedule(verifier.verify(batch)?).await?; } ReplicationMessage::Verified(batch) => self.schedule(batch).await?, - ReplicationMessage::Entry(entry) => { - let boundary = matches!(entry, OwnedBlockstoreEntry::Block(_)); - self.process(entry).await?; - if boundary && draining { + ReplicationMessage::Block(block) => { + let (external, guard) = ExternalBlock::new(block); + self.pacer.send(external).await.map_err(EngineError::from)?; + let pending = time::timeout(IO_TIMEOUT, self.blocks.recv()); + let observed = pending.await?.ok_or(ReplicationError::BlockStreamClosed)?; + if block != observed { + let error = EngineError::Replay(ReplayError::BlockhashMismatch(block.slot)); + return Err(error.into()); + } + guard.await.map_err(EngineError::from)?; + if draining { let (_guard, position) = self.resume().await?; return Ok(ControlExit::Boundary(position)); } } + ReplicationMessage::Superblock(expected) => { + let observed = self.superblocks().sealed(); + if observed != expected { + error!(?expected, ?observed, "replication state mismatch detected"); + metrics::client_state_mismatch(); + return Err(EngineError::Replay(ReplayError::StateMismatch).into()); + } + } + ReplicationMessage::Reset(slot) => { + self.engine.replay(OwnedBlockstoreEntry::Reset(slot)).await?; + } } } } @@ -223,36 +245,6 @@ impl ReplicationClient { Ok(()) } - /// Applies and validates one non-transaction stream entry. - async fn process(&mut self, entry: OwnedBlockstoreEntry) -> Result<()> { - match entry { - OwnedBlockstoreEntry::Block(block) => { - let (external, guard) = ExternalBlock::new(block); - self.pacer.send(external).await.map_err(EngineError::from)?; - let pending = time::timeout(IO_TIMEOUT, self.blocks.recv()); - let observed = pending.await?.ok_or(ReplicationError::BlockStreamClosed)?; - if block != observed { - let error = ReplayError::BlockhashMismatch(block.slot); - Err(EngineError::from(error))?; - } - guard.await.map_err(EngineError::from)?; - } - OwnedBlockstoreEntry::Superblock(expected) => { - let observed = self.superblocks().sealed(); - if observed != expected { - error!(?expected, ?observed, "replication state mismatch detected"); - metrics::client_state_mismatch(); - Err(EngineError::Replay(ReplayError::StateMismatch))?; - } - } - OwnedBlockstoreEntry::Reset(slot) => { - self.engine.replay(OwnedBlockstoreEntry::Reset(slot)).await?; - } - OwnedBlockstoreEntry::Transaction(_) => unreachable!("Ingest batches transactions"), - } - Ok(()) - } - async fn resume(&self) -> Result<(BarrierHandle, BlockstorePosition)> { let guard = self.barrier().await?; self.superblocks().sync(false)?; @@ -359,18 +351,8 @@ impl Ingest { /// Decodes until transport loss, terminal failure, or Control exit. fn run(mut self) -> Result { loop { - match blockstore::decode(&mut self.stream) { - Ok(OwnedBlockstoreEntry::Transaction(transaction)) => { - self.batch.push(transaction); - if self.batch.is_full() && !self.flush()? { - return Ok(IngestExit::Stopped); - } - } - Ok(entry) => { - if !self.flush()? || self.tx.send(ReplicationMessage::Entry(entry)).is_err() { - return Ok(IngestExit::Stopped); - } - } + let entry = match blockstore::decode(&mut self.stream) { + Ok(entry) => entry, Err(wincode::error::ReadError::Io(error)) => { if !self.flush()? { return Ok(IngestExit::Stopped); @@ -383,6 +365,21 @@ impl Ingest { } return Err(wincode::Error::from(error).into()); } + }; + let message = match entry { + OwnedBlockstoreEntry::Transaction(transaction) => { + self.batch.push(transaction); + if self.batch.is_full() && !self.flush()? { + return Ok(IngestExit::Stopped); + } + continue; + } + OwnedBlockstoreEntry::Block(block) => ReplicationMessage::Block(block), + OwnedBlockstoreEntry::Superblock(seal) => ReplicationMessage::Superblock(seal), + OwnedBlockstoreEntry::Reset(slot) => ReplicationMessage::Reset(slot), + }; + if !self.flush()? || self.tx.send(message).is_err() { + return Ok(IngestExit::Stopped); } } } diff --git a/solana/transaction-view/src/address_table_lookup_frame.rs b/solana/transaction-view/src/address_table_lookup_frame.rs index e274bbd5..abfc7d9d 100644 --- a/solana/transaction-view/src/address_table_lookup_frame.rs +++ b/solana/transaction-view/src/address_table_lookup_frame.rs @@ -50,7 +50,7 @@ const MAX_ATLS_PER_PACKET: u8 = ((PACKET_DATA_SIZE - MIN_SIZED_PACKET_WITH_ATLS) / MIN_SIZED_ATL) as u8; /// Contains metadata about the address table lookups in a transaction packet. -#[derive(Debug)] +#[derive(Debug, Default)] pub(crate) struct AddressTableLookupFrame { /// The number of address table lookups in the transaction. pub(crate) num_address_table_lookups: u8, diff --git a/solana/transaction-view/src/message_header_frame.rs b/solana/transaction-view/src/message_header_frame.rs index b37c2018..33b1adb4 100644 --- a/solana/transaction-view/src/message_header_frame.rs +++ b/solana/transaction-view/src/message_header_frame.rs @@ -27,7 +27,7 @@ pub(crate) struct MessageHeaderFrame { impl MessageHeaderFrame { #[inline(always)] - pub(crate) fn try_new(bytes: &[u8], offset: &mut usize) -> Result { + pub(crate) fn try_new_legacy_or_v0(bytes: &[u8], offset: &mut usize) -> Result { // Get the message offset. let message_offset = try_u32_offset(*offset)?; @@ -69,21 +69,21 @@ mod tests { fn test_invalid_version() { let bytes = [0b1000_0001]; let mut offset = 0; - assert!(MessageHeaderFrame::try_new(&bytes, &mut offset).is_err()); + assert!(MessageHeaderFrame::try_new_legacy_or_v0(&bytes, &mut offset).is_err()); } #[test] fn test_legacy_transaction_missing_header_byte() { let bytes = [5, 0]; let mut offset = 0; - assert!(MessageHeaderFrame::try_new(&bytes, &mut offset).is_err()); + assert!(MessageHeaderFrame::try_new_legacy_or_v0(&bytes, &mut offset).is_err()); } #[test] fn test_legacy_transaction_valid() { let bytes = [5, 1, 2]; let mut offset = 0; - let header = MessageHeaderFrame::try_new(&bytes, &mut offset).unwrap(); + let header = MessageHeaderFrame::try_new_legacy_or_v0(&bytes, &mut offset).unwrap(); assert!(matches!(header.version, TransactionVersion::Legacy)); assert_eq!(header.num_required_signatures, 5); assert_eq!(header.num_readonly_signed_accounts, 1); @@ -94,14 +94,14 @@ mod tests { fn test_v0_transaction_missing_header_byte() { let bytes = [MESSAGE_VERSION_PREFIX, 5, 1]; let mut offset = 0; - assert!(MessageHeaderFrame::try_new(&bytes, &mut offset).is_err()); + assert!(MessageHeaderFrame::try_new_legacy_or_v0(&bytes, &mut offset).is_err()); } #[test] fn test_v0_transaction_valid() { let bytes = [MESSAGE_VERSION_PREFIX, 5, 1, 2]; let mut offset = 0; - let header = MessageHeaderFrame::try_new(&bytes, &mut offset).unwrap(); + let header = MessageHeaderFrame::try_new_legacy_or_v0(&bytes, &mut offset).unwrap(); assert!(matches!(header.version, TransactionVersion::V0)); assert_eq!(header.num_required_signatures, 5); assert_eq!(header.num_readonly_signed_accounts, 1); diff --git a/solana/transaction-view/src/transaction_frame.rs b/solana/transaction-view/src/transaction_frame.rs index 219c3568..7328b38b 100644 --- a/solana/transaction-view/src/transaction_frame.rs +++ b/solana/transaction-view/src/transaction_frame.rs @@ -53,7 +53,7 @@ impl TransactionFrame { fn try_new_as_legacy_or_v0(bytes: &[u8]) -> Result { let mut offset = 0; let signature = SignatureFrame::try_new(bytes, &mut offset)?; - let message_header = MessageHeaderFrame::try_new(bytes, &mut offset)?; + let message_header = MessageHeaderFrame::try_new_legacy_or_v0(bytes, &mut offset)?; let static_account_keys = StaticAccountKeysFrame::try_new(bytes, &mut offset)?; // The recent blockhash is the first account key after the static @@ -63,17 +63,10 @@ impl TransactionFrame { advance_offset_for_type::(bytes, &mut offset)?; let instructions = InstructionsFrame::try_new_for_legacy_and_v0(bytes, &mut offset)?; - let address_table_lookup = match message_header.version { - TransactionVersion::Legacy => AddressTableLookupFrame { - num_address_table_lookups: 0, - offset: 0, - total_writable_lookup_accounts: 0, - total_readonly_lookup_accounts: 0, - }, - TransactionVersion::V0 => AddressTableLookupFrame::try_new(bytes, &mut offset)?, - TransactionVersion::V1 | TransactionVersion::Magicblock => { - unreachable!("unexpected variant") - } + let address_table_lookup = if matches!(message_header.version, TransactionVersion::V0) { + AddressTableLookupFrame::try_new(bytes, &mut offset)? + } else { + AddressTableLookupFrame::default() }; // Verify that the entire transaction was parsed. @@ -177,12 +170,7 @@ impl TransactionFrame { recent_blockhash_offset, instructions, // Don't have ATL in txv1 - address_table_lookup: AddressTableLookupFrame { - num_address_table_lookups: 0, - offset: 0, - total_writable_lookup_accounts: 0, - total_readonly_lookup_accounts: 0, - }, + address_table_lookup: AddressTableLookupFrame::default(), transaction_config_frame, data_len: try_u32_offset(offset)?, };