refactor: remove invariant panic paths

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,15 +331,14 @@ impl LedgerReader {
     }
 
     /// Builds the block response requested by the caller.
-    fn populate_block_response(
+    fn block_response(
         &mut self,
         superblock: &Superblock,
         request: &BlockPayload,
         span: Span,
     ) -> Result<BlockResponse> {
         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)),
         }
     }
 

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<Vec<u8>>),
     /// Transactions verified by Ingest while Control was occupied.
     Verified(Vec<VerifiedTransaction>),
-    /// 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<IngestExit> {
         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);
             }
         }
     }

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,

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<Self> {
+    pub(crate) fn try_new_legacy_or_v0(bytes: &[u8], offset: &mut usize) -> Result<Self> {
         // 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);

solana/transaction-view/src/transaction_frame.rs

@@ -53,7 +53,7 @@ impl TransactionFrame {
     fn try_new_as_legacy_or_v0(bytes: &[u8]) -> Result<Self> {
         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::<Hash>(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)?,
         };