chore(core, proto): migrate byte slices from Vec to Bytes (#1319)

## Summary
Changed protobuf, astria-core, and dependent code to utilize generic
`Bytes` instead of the default `Vec<u8>` to utilize zero-copy
deserialization from Protobuf and make type conversion cheaper.

## Background
Previously, some components were utilizing `Bytes` and some were
utilizing `Vec<u8>` as the generated Rust types when converting from
Protobuf `bytes`.

## Changes
- Adjusted Protobuf compiler to use `Bytes` for all Astria paths.
- Edited conversions to/from `Raw` in `astria-core` to utilize
`prost::bytes::Bytes`.
- Adjusted code which relied on `Vec<u8>`.

## Testing
Passing all tests.

## Related Issues
closes #674

crates/astria-composer/src/collectors/geth.rs

@@ -236,7 +236,7 @@ impl Geth {
                         let data = tx.rlp().to_vec();
                         let seq_action = SequenceAction {
                             rollup_id,
-                            data,
+                            data: data.into(),
                             fee_asset: fee_asset.clone(),
                         };
 

crates/astria-composer/src/executor/tests.rs

@@ -13,7 +13,10 @@ use astria_core::{
 };
 use astria_eyre::eyre;
 use once_cell::sync::Lazy;
-use prost::Message;
+use prost::{
+    bytes::Bytes,
+    Message as _,
+};
 use sequencer_client::SignedTransaction;
 use serde_json::json;
 use tempfile::NamedTempFile;
@@ -81,7 +84,7 @@ static TELEMETRY: Lazy<()> = Lazy::new(|| {
 fn sequence_action() -> SequenceAction {
     SequenceAction {
         rollup_id: RollupId::new([0; ROLLUP_ID_LEN]),
-        data: vec![],
+        data: Bytes::new(),
         fee_asset: "nria".parse().unwrap(),
     }
 }
@@ -411,7 +414,7 @@ async fn bundle_triggered_by_block_timer() {
     // send two sequence actions to the executor, both small enough to fit in a single bundle
     // without filling it
     let seq0 = SequenceAction {
-        data: vec![0u8; cfg.max_bytes_per_bundle / 4],
+        data: vec![0u8; cfg.max_bytes_per_bundle / 4].into(),
         ..sequence_action()
     };
 
@@ -498,13 +501,13 @@ async fn two_seq_actions_single_bundle() {
     // send two sequence actions to the executor, both small enough to fit in a single bundle
     // without filling it
     let seq0 = SequenceAction {
-        data: vec![0u8; cfg.max_bytes_per_bundle / 4],
+        data: vec![0u8; cfg.max_bytes_per_bundle / 4].into(),
         ..sequence_action()
     };
 
     let seq1 = SequenceAction {
         rollup_id: RollupId::new([1; ROLLUP_ID_LEN]),
-        data: vec![1u8; cfg.max_bytes_per_bundle / 4],
+        data: vec![1u8; cfg.max_bytes_per_bundle / 4].into(),
         ..sequence_action()
     };
 

crates/astria-composer/src/test_utils.rs

@@ -16,7 +16,7 @@ fn encoded_len(action: &SequenceAction) -> usize {
 pub(crate) fn sequence_action_with_n_bytes(n: usize) -> SequenceAction {
     SequenceAction {
         rollup_id: RollupId::new([0; ROLLUP_ID_LEN]),
-        data: vec![0; n],
+        data: vec![0; n].into(),
         fee_asset: "nria"
             .parse::<asset::Denom>()
             .unwrap()

crates/astria-composer/tests/blackbox/grpc_collector.rs

@@ -11,6 +11,7 @@ use astria_core::{
     primitive::v1::RollupId,
 };
 use ethers::prelude::Transaction;
+use prost::bytes::Bytes;
 
 use crate::helper::{
     mount_broadcast_tx_sync_invalid_nonce_mock,
@@ -45,8 +46,8 @@ async fn tx_from_one_rollup_is_received_by_sequencer() {
     .unwrap();
     composer_client
         .submit_rollup_transaction(SubmitRollupTransactionRequest {
-            rollup_id: rollup_id.as_ref().to_vec(),
-            data: tx.rlp().to_vec(),
+            rollup_id: Bytes::copy_from_slice(rollup_id.as_ref()),
+            data: Bytes::copy_from_slice(&tx.rlp()),
         })
         .await
         .expect("rollup transactions should have been submitted successfully to grpc collector");
@@ -107,8 +108,8 @@ async fn invalid_nonce_causes_resubmission_under_different_nonce() {
     .unwrap();
     composer_client
         .submit_rollup_transaction(SubmitRollupTransactionRequest {
-            rollup_id: rollup_id.as_ref().to_vec(),
-            data: tx.rlp().to_vec(),
+            rollup_id: Bytes::copy_from_slice(rollup_id.as_ref()),
+            data: Bytes::copy_from_slice(&tx.rlp()),
         })
         .await
         .expect("rollup transactions should have been submitted successfully to grpc collector");
@@ -162,8 +163,8 @@ async fn single_rollup_tx_payload_integrity() {
     .unwrap();
     composer_client
         .submit_rollup_transaction(SubmitRollupTransactionRequest {
-            rollup_id: rollup_id.as_ref().to_vec(),
-            data: tx.rlp().to_vec(),
+            rollup_id: Bytes::copy_from_slice(rollup_id.as_ref()),
+            data: Bytes::copy_from_slice(&tx.rlp()),
         })
         .await
         .expect("rollup transactions should have been submitted successfully to grpc collector");

crates/astria-conductor/src/celestia/block_verifier.rs

@@ -228,6 +228,7 @@ mod test {
             celestia::UncheckedSubmittedMetadata,
         },
     };
+    use bytes::Bytes;
     use prost::Message as _;
     use sequencer_client::{
         tendermint::{
@@ -337,9 +338,9 @@ mod test {
                 seconds: 1,
                 nanos: 0,
             }),
-            data_hash: data_hash.to_vec(),
-            rollup_transactions_root: rollup_transactions_root.to_vec(),
-            proposer_address: proposer_address.as_bytes().to_vec(),
+            data_hash: Bytes::copy_from_slice(&data_hash),
+            rollup_transactions_root: Bytes::copy_from_slice(&rollup_transactions_root),
+            proposer_address: Bytes::copy_from_slice(proposer_address.as_bytes()),
         };
         let header = SequencerBlockHeader::try_from_raw(header).unwrap();
 
@@ -359,7 +360,7 @@ mod test {
 
     #[tokio::test]
     async fn validate_sequencer_blob_with_chain_ids() {
-        let test_tx = b"test-tx".to_vec();
+        let test_tx = Bytes::from_static(b"test-tx");
         let rollup_id = RollupId::from_unhashed_bytes(b"test-chain");
         let grouped_txs = BTreeMap::from([(rollup_id, vec![test_tx.clone()])]);
         let rollup_transactions_tree =
@@ -382,9 +383,9 @@ mod test {
                 seconds: 1,
                 nanos: 0,
             }),
-            data_hash: data_hash.to_vec(),
-            rollup_transactions_root: rollup_transactions_root.to_vec(),
-            proposer_address: proposer_address.as_bytes().to_vec(),
+            data_hash: Bytes::copy_from_slice(&data_hash),
+            rollup_transactions_root: Bytes::copy_from_slice(&rollup_transactions_root),
+            proposer_address: Bytes::copy_from_slice(proposer_address.as_bytes()),
         };
         let header = SequencerBlockHeader::try_from_raw(header).unwrap();
 

crates/astria-conductor/src/celestia/mod.rs

@@ -13,6 +13,7 @@ use astria_eyre::eyre::{
     bail,
     WrapErr as _,
 };
+use bytes::Bytes;
 use celestia_types::nmt::Namespace;
 use futures::{
     future::{
@@ -101,7 +102,7 @@ pub(crate) struct ReconstructedBlock {
     pub(crate) celestia_height: u64,
     pub(crate) block_hash: [u8; 32],
     pub(crate) header: SequencerBlockHeader,
-    pub(crate) transactions: Vec<Vec<u8>>,
+    pub(crate) transactions: Vec<Bytes>,
 }
 
 impl ReconstructedBlock {

crates/astria-conductor/src/executor/client.rs

@@ -117,14 +117,14 @@ impl Client {
     pub(super) async fn execute_block_with_retry(
         &mut self,
         prev_block_hash: Bytes,
-        transactions: Vec<Vec<u8>>,
+        transactions: Vec<Bytes>,
         timestamp: Timestamp,
     ) -> eyre::Result<Block> {
         use prost::Message;
 
         let transactions = transactions
             .into_iter()
-            .map(|tx| RollupData::decode(tx.as_slice()))
+            .map(RollupData::decode)
             .collect::<Result<_, _>>()
             .wrap_err("failed to decode tx bytes as RollupData")?;
 

crates/astria-conductor/src/executor/mod.rs

@@ -686,7 +686,7 @@ struct ExecutableBlock {
     hash: [u8; 32],
     height: SequencerHeight,
     timestamp: pbjson_types::Timestamp,
-    transactions: Vec<Vec<u8>>,
+    transactions: Vec<Bytes>,
 }
 
 impl ExecutableBlock {