chore(sequencer)!: exclusively use Borsh encoding for stored data (EN…

…G-768) (#1492)

## Summary
This PR primarily changes the encoding format of all data being written
to storage to Borsh.

## Background
We currently have a variety of different encoding formats, and this can
be confusing, sub-optimal (in terms of storage space consumed and
serialization/deserialization performance) and potentially problematic
as e.g. JSON-encoding leaves a lot of flexibility in how the actual
serialized data can look.

This PR is part one of three which aim to improve the performance and
quality of the storage component. As such, the APIs of the various
`StateReadExt` and `StateWriteExt` extension traits were updated
slightly in preparation for the upcoming changes. In broad terms, for
getters this meant having ref parameters rather than value ones (even
for copyable types like `[u8; 32]` this is significantly more
performant), and for "putters", parameters which are used for DB keys
are refs, while the DB value parameters become values, since in the next
PR these values will be added to a cache.

## Changes
- Added a new `storage` module. This will ultimately contain our own
equivalents of the `cnidarium` types, but for now consists only of a
collection of submodules for types currently being written to storage.
There is a top-level enum `StoredValue` which becomes the only type
being written to storage by our own code.
- To accommodate for converting between the storage types and the
corresponding domain types defined in `astria-core` and `astria-merkle`,
some of these have been provided with constructors named
`unchecked_from_parts`. This allows the type to be constructed from a
trusted source like our own DB, skipping any validation steps which
might otherwise be done.
- Updated all `StateReadExt` and `StateWriteExt` traits to use the new
`StoredValue`, which internally uses Borsh encoding.
- Updated the APIs of all these extension traits as described above.
This change resulted in a slew of similar updates to avoid needless
copying `[u8; 32]` and `[u8; 20]` arrays, and that has unfortunately
made the PR larger than expected.
- A few core types which currently derive or implement `serde` traits
have had that removed, since it only existed to support JSON-encoding
the type for storing. In one case it was for JSON-encoding to a debug
log line, but I replaced that with a `Display` impl.

## Testing
- Existing unit tests of the `StateReadExt` and `StateWriteExt` traits
already had almost full coverage. Where coverage was missing, I added
round trip tests.

## Breaking Changelist
- The on-disk format of stored data has changed.

## Related Issues
Closes #1434.

crates/astria-cli/src/commands/bridge/submit.rs

@@ -118,7 +118,7 @@ async fn submit_transaction(
     actions: Vec<Action>,
 ) -> eyre::Result<Response> {
     let from_address = Address::builder()
-        .array(signing_key.verification_key().address_bytes())
+        .array(*signing_key.verification_key().address_bytes())
         .prefix(prefix)
         .try_build()
         .wrap_err("failed constructing a valid from address from the provided prefix")?;

crates/astria-cli/src/commands/sequencer.rs

@@ -463,7 +463,7 @@ async fn submit_transaction(
     let sequencer_key = SigningKey::from(private_key_bytes);
 
     let from_address = Address::builder()
-        .array(sequencer_key.verification_key().address_bytes())
+        .array(*sequencer_key.verification_key().address_bytes())
         .prefix(prefix)
         .try_build()
         .wrap_err("failed constructing a valid from address from the provided prefix")?;

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

@@ -58,7 +58,7 @@ impl Builder {
 
         let sequencer_address = Address::builder()
             .prefix(sequencer_address_prefix)
-            .array(sequencer_key.verification_key().address_bytes())
+            .array(*sequencer_key.verification_key().address_bytes())
             .try_build()
             .wrap_err("failed constructing a sequencer address from private key")?;
 

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

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use astria_core::{
     primitive::v1::RollupId,
     sequencerblock::v1alpha1::{
+        celestia::UncheckedSubmittedMetadata,
         SubmittedMetadata,
         SubmittedRollupData,
     },
@@ -52,14 +53,19 @@ pub(super) fn reconstruct_blocks_from_verified_blobs(
         if let Some(header_blob) =
             remove_header_blob_matching_rollup_blob(&mut header_blobs, &rollup)
         {
+            let UncheckedSubmittedMetadata {
+                block_hash,
+                header,
+                ..
+            } = header_blob.into_unchecked();
             reconstructed_blocks.push(ReconstructedBlock {
                 celestia_height,
-                block_hash: header_blob.block_hash(),
-                header: header_blob.into_unchecked().header,
+                block_hash,
+                header,
                 transactions: rollup.into_unchecked().transactions,
             });
         } else {
-            let reason = if header_blobs.contains_key(&rollup.sequencer_block_hash()) {
+            let reason = if header_blobs.contains_key(rollup.sequencer_block_hash()) {
                 "sequencer header blobs with the same block hash as the rollup blob found, but the \
                  rollup's Merkle proof did not lead any Merkle roots"
             } else {
@@ -77,13 +83,13 @@ pub(super) fn reconstruct_blocks_from_verified_blobs(
     for header_blob in header_blobs.into_values() {
         if header_blob.contains_rollup_id(rollup_id) {
             warn!(
-                block_hash = %base64(&header_blob.block_hash()),
+                block_hash = %base64(header_blob.block_hash()),
                 "sequencer header blob contains the target rollup ID, but no matching rollup blob was found; dropping it",
             );
         } else {
             reconstructed_blocks.push(ReconstructedBlock {
                 celestia_height,
-                block_hash: header_blob.block_hash(),
+                block_hash: *header_blob.block_hash(),
                 header: header_blob.into_unchecked().header,
                 transactions: vec![],
             });
@@ -98,11 +104,11 @@ fn remove_header_blob_matching_rollup_blob(
 ) -> Option<SubmittedMetadata> {
     // chaining methods and returning () to use the ? operator and to not bind the value
     headers
-        .get(&rollup.sequencer_block_hash())
+        .get(rollup.sequencer_block_hash())
         .and_then(|header| {
             verify_rollup_blob_against_sequencer_blob(rollup, header).then_some(())
         })?;
-    headers.remove(&rollup.sequencer_block_hash())
+    headers.remove(rollup.sequencer_block_hash())
 }
 
 fn verify_rollup_blob_against_sequencer_blob(
@@ -112,9 +118,9 @@ fn verify_rollup_blob_against_sequencer_blob(
     rollup_blob
         .proof()
         .audit()
-        .with_root(sequencer_blob.rollup_transactions_root())
+        .with_root(*sequencer_blob.rollup_transactions_root())
         .with_leaf_builder()
-        .write(&rollup_blob.rollup_id().get())
+        .write(rollup_blob.rollup_id().as_bytes())
         .write(&merkle::Tree::from_leaves(rollup_blob.transactions()).root())
         .finish_leaf()
         .perform()

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

@@ -121,7 +121,7 @@ pub(super) async fn verify_metadata(
             verification_tasks.spawn(
                 VerificationTaskKey {
                     index,
-                    block_hash: blob.block_hash(),
+                    block_hash: *blob.block_hash(),
                     sequencer_height: blob.height(),
                 },
                 blob_verifier
@@ -136,10 +136,10 @@ pub(super) async fn verify_metadata(
         match verification_result {
             Ok(Some(verified_blob)) => {
                 if let Some(dropped_entry) =
-                    verified_header_blobs.insert(verified_blob.block_hash(), verified_blob)
+                    verified_header_blobs.insert(*verified_blob.block_hash(), verified_blob)
                 {
                     let accepted_entry = verified_header_blobs
-                        .get(&dropped_entry.block_hash())
+                        .get(dropped_entry.block_hash())
                         .expect("must exist; just inserted an item under the same key");
                     info!(
                         block_hash = %base64(&dropped_entry.block_hash()),
@@ -291,7 +291,7 @@ impl BlobVerifier {
         .and_then(|()| {
             ensure_block_hashes_match(
                 cached.commit_header.commit.block_id.hash.as_bytes(),
-                &metadata.block_hash(),
+                metadata.block_hash(),
             )
         }) {
             info!(reason = %error, "failed to verify metadata retrieved from Celestia; dropping it");

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

@@ -714,19 +714,20 @@ impl ExecutableBlock {
     }
 
     fn from_sequencer(block: FilteredSequencerBlock, id: RollupId) -> Self {
-        let hash = block.block_hash();
-        let height = block.height();
-        let timestamp = convert_tendermint_time_to_protobuf_timestamp(block.header().time());
         let FilteredSequencerBlockParts {
+            block_hash,
+            header,
             mut rollup_transactions,
             ..
         } = block.into_parts();
+        let height = header.height();
+        let timestamp = convert_tendermint_time_to_protobuf_timestamp(header.time());
         let transactions = rollup_transactions
             .swap_remove(&id)
             .map(|txs| txs.transactions().to_vec())
             .unwrap_or_default();
         Self {
-            hash,
+            hash: block_hash,
             height,
             timestamp,
             transactions,

crates/astria-conductor/src/lib.rs

@@ -1,6 +1,7 @@
 //! # Astria Conductor
 //!
 //! The Astria conductor connects the shared sequencer layer and the execution layer.
+//!
 //! When a block is received from the sequencer layer, the conductor pushes it to the execution
 //! layer. There are two ways for a block to be received:
 //!

crates/astria-conductor/tests/blackbox/helpers/mod.rs

@@ -48,7 +48,7 @@ use wiremock::MockServer;
 pub const CELESTIA_BEARER_TOKEN: &str = "ABCDEFGH";
 
 pub const ROLLUP_ID: RollupId = RollupId::new([42; 32]);
-pub static ROLLUP_ID_BYTES: Bytes = Bytes::from_static(&RollupId::get(ROLLUP_ID));
+pub static ROLLUP_ID_BYTES: Bytes = Bytes::from_static(ROLLUP_ID.as_bytes());
 
 pub const SEQUENCER_CHAIN_ID: &str = "test_sequencer-1000";
 pub const CELESTIA_CHAIN_ID: &str = "test_celestia-1000";
@@ -634,7 +634,7 @@ pub fn make_commit(height: u32) -> tendermint::block::Commit {
     let signing_key = signing_key();
     let validator = validator();
 
-    let block_hash = make_sequencer_block(height).block_hash();
+    let block_hash = *make_sequencer_block(height).block_hash();
 
     let timestamp = tendermint::Time::from_unix_timestamp(1, 1).unwrap();
     let canonical_vote = tendermint::vote::CanonicalVote {

crates/astria-core/Cargo.toml

@@ -55,6 +55,10 @@ server = ["dep:tonic"]
 test-utils = []
 base64-serde = ["dep:base64-serde"]
 brotli = ["dep:brotli"]
+# When enabled, this adds constructors for some types that skip the normal constructor validity
+# checks. It supports the case where the inputs are already deemed valid, e.g. having read them from
+# local storage.
+unchecked-constructors = []
 
 [dev-dependencies]
 astria-core = { path = ".", features = ["serde"] }

crates/astria-core/src/celestia.rs

@@ -20,7 +20,7 @@ pub const fn namespace_v0_from_first_10_bytes(bytes: &[u8]) -> Namespace {
 /// 10 bytes of [`crate::primitive::v1::RollupId`].
 #[must_use = "a celestia namespace must be used in order to be useful"]
 pub const fn namespace_v0_from_rollup_id(rollup_id: crate::primitive::v1::RollupId) -> Namespace {
-    namespace_v0_from_first_10_bytes(&rollup_id.get())
+    namespace_v0_from_first_10_bytes(rollup_id.as_bytes())
 }
 
 /// Constructs a [`celestia_types::nmt::Namespace`] from the first 10 bytes of the sha256 hash of

crates/astria-core/src/crypto.rs

@@ -86,7 +86,7 @@ impl SigningKey {
     /// Returns the address bytes of the verification key associated with this signing key.
     #[must_use]
     pub fn address_bytes(&self) -> [u8; ADDRESS_LEN] {
-        self.verification_key().address_bytes()
+        *self.verification_key().address_bytes()
     }
 
     /// Attempts to create an Astria bech32m `[Address]` with the given prefix.
@@ -167,8 +167,8 @@ impl VerificationKey {
     ///
     /// The address is the first 20 bytes of the sha256 hash of the verification key.
     #[must_use]
-    pub fn address_bytes(&self) -> [u8; ADDRESS_LEN] {
-        *self.address_bytes.get_or_init(|| {
+    pub fn address_bytes(&self) -> &[u8; ADDRESS_LEN] {
+        self.address_bytes.get_or_init(|| {
             fn first_20(array: [u8; 32]) -> [u8; ADDRESS_LEN] {
                 [
                     array[0], array[1], array[2], array[3], array[4], array[5], array[6], array[7],

crates/astria-core/src/primitive/v1/asset/denom.rs

@@ -134,6 +134,12 @@ impl<'a> From<&'a Denom> for IbcPrefixed {
     }
 }
 
+impl<'a> From<&'a IbcPrefixed> for IbcPrefixed {
+    fn from(value: &IbcPrefixed) -> Self {
+        *value
+    }
+}
+
 impl FromStr for Denom {
     type Err = ParseDenomError;
 
@@ -282,6 +288,44 @@ impl TracePrefixed {
         }
         len + self.base_denom.len()
     }
+
+    pub fn trace(&self) -> impl Iterator<Item = (&str, &str)> {
+        self.trace
+            .inner
+            .iter()
+            .map(|segment| (segment.port.as_str(), segment.channel.as_str()))
+    }
+
+    #[must_use]
+    pub fn base_denom(&self) -> &str {
+        &self.base_denom
+    }
+
+    /// This should only be used where the inputs have been provided by a trusted entity, e.g. read
+    /// from our own state store.
+    ///
+    /// Note that this function is not considered part of the public API and is subject to breaking
+    /// change at any time.
+    #[cfg(feature = "unchecked-constructors")]
+    #[doc(hidden)]
+    #[must_use]
+    pub fn unchecked_from_parts<I: IntoIterator<Item = (String, String)>>(
+        trace: I,
+        base_denom: String,
+    ) -> Self {
+        Self {
+            trace: TraceSegments {
+                inner: trace
+                    .into_iter()
+                    .map(|(port, channel)| PortAndChannel {
+                        port,
+                        channel,
+                    })
+                    .collect(),
+            },
+            base_denom,
+        }
+    }
 }
 
 #[derive(Debug, Clone, Hash, PartialEq, Eq)]
@@ -507,8 +551,8 @@ impl IbcPrefixed {
     }
 
     #[must_use]
-    pub fn get(&self) -> [u8; 32] {
-        self.id
+    pub fn as_bytes(&self) -> &[u8; 32] {
+        &self.id
     }
 
     #[must_use]