Audit p2p sync errors

crates/p2p/src/client/peer_agnostic.rs

@@ -779,9 +779,9 @@ mod transaction_stream {
         tracing::trace!(?start, ?stop, "Streaming Transactions");
 
         make_stream::from_future(move |tx| async move {
-            let mut counts_and_commitments_stream = Box::pin(counts_stream);
+            let mut expected_transaction_counts_stream = Box::pin(counts_stream);
 
-            let cnt = match try_next(&mut counts_and_commitments_stream).await {
+            let cnt = match try_next(&mut expected_transaction_counts_stream).await {
                 Ok(x) => x,
                 Err(e) => {
                     _ = tx.send(Err(e)).await;
@@ -826,7 +826,7 @@ mod transaction_stream {
                         if yield_block(
                             peer,
                             &mut progress,
-                            &mut counts_and_commitments_stream,
+                            &mut expected_transaction_counts_stream,
                             transactions,
                             &mut start,
                             stop,
@@ -1557,14 +1557,15 @@ mod event_stream {
 
 async fn try_next<T>(
     count_stream: &mut (impl Stream<Item = anyhow::Result<T>> + Unpin + Send + 'static),
-) -> Result<T, PeerData<anyhow::Error>> {
+) -> Result<T, anyhow::Error> {
     match count_stream.next().await {
         Some(Ok(cnt)) => Ok(cnt),
-        Some(Err(e)) => Err(PeerData::new(PeerId::random(), e)),
-        None => Err(PeerData::new(
-            PeerId::random(),
-            anyhow::anyhow!("Count stream terminated prematurely"),
-        )),
+        // This is a non-recoverable error, because "Counter" streams fail only if the underlying
+        // database fails.
+        Some(Err(e)) => Err(e),
+        // This is a non-recoverable error, because we expect all the necessary headers that are the
+        // source of the stream to be in the database.
+        None => Err(anyhow::anyhow!("Count stream terminated prematurely")),
     }
 }
 

crates/p2p/src/client/peer_agnostic/traits.rs

@@ -16,7 +16,7 @@ use crate::client::types::{
 };
 use crate::PeerData;
 
-pub type StreamItem<T> = Result<PeerData<T>, PeerData<anyhow::Error>>;
+pub type StreamItem<T> = Result<PeerData<T>, anyhow::Error>;
 
 pub trait HeaderStream {
     fn header_stream(

crates/p2p/src/peer_data.rs

@@ -13,12 +13,6 @@ impl<T> PeerData<T> {
         Self { peer, data }
     }
 
-    pub fn from_result<E>(peer: PeerId, result: Result<T, E>) -> Result<PeerData<T>, PeerData<E>> {
-        result
-            .map(|x| Self::new(peer, x))
-            .map_err(|e| PeerData::<E>::new(peer, e))
-    }
-
     pub fn for_tests(data: T) -> Self {
         Self {
             peer: PeerId::random(),

crates/pathfinder/examples/compute_pre0132_hashes.rs

@@ -126,7 +126,7 @@ fn main() -> anyhow::Result<()> {
 
         // Compute the block hash in the 0.13.2 style
         let header_data = get_header_data(&header);
-        let new_block_hash = compute_final_hash(&header_data).context("Computing block hash")?;
+        let new_block_hash = compute_final_hash(&header_data);
 
         // Write to the CSV file
         writeln!(csv_file, "{},{}", block_number, new_block_hash)?;

crates/pathfinder/src/state/block_hash.rs

@@ -253,7 +253,7 @@ pub fn verify_block_hash(
             false
         }
     } else {
-        let computed_hash = compute_final_hash(&header)?;
+        let computed_hash = compute_final_hash(&header);
         computed_hash == header.hash
     };
 
@@ -405,7 +405,7 @@ fn compute_final_hash_pre_0_13_2(header: &BlockHeaderData) -> BlockHash {
     BlockHash(chain.finalize())
 }
 
-fn compute_final_hash_v0(header: &BlockHeaderData) -> Result<BlockHash> {
+fn compute_final_hash_v0(header: &BlockHeaderData) -> BlockHash {
     // Hash the block header.
     let mut hasher = PoseidonHasher::new();
     hasher.write(felt_bytes!(b"STARKNET_BLOCK_HASH0").into());
@@ -429,12 +429,12 @@ fn compute_final_hash_v0(header: &BlockHeaderData) -> Result<BlockHash> {
     );
     hasher.write(MontFelt::ZERO);
     hasher.write(header.parent_hash.0.into());
-    Ok(BlockHash(hasher.finish().into()))
+    BlockHash(hasher.finish().into())
 }
 
 // Bumps the initial STARKNET_BLOCK_HASH0 to STARKNET_BLOCK_HASH1,
 // replaces gas price elements with gas_prices_hash.
-fn compute_final_hash_v1(header: &BlockHeaderData) -> Result<BlockHash> {
+fn compute_final_hash_v1(header: &BlockHeaderData) -> BlockHash {
     // Hash the block header.
     let mut hasher = PoseidonHasher::new();
     hasher.write(felt_bytes!(b"STARKNET_BLOCK_HASH1").into());
@@ -455,10 +455,10 @@ fn compute_final_hash_v1(header: &BlockHeaderData) -> Result<BlockHash> {
     );
     hasher.write(MontFelt::ZERO);
     hasher.write(header.parent_hash.0.into());
-    Ok(BlockHash(hasher.finish().into()))
+    BlockHash(hasher.finish().into())
 }
 
-pub fn compute_final_hash(header: &BlockHeaderData) -> Result<BlockHash> {
+pub fn compute_final_hash(header: &BlockHeaderData) -> BlockHash {
     if header.starknet_version < StarknetVersion::V_0_13_4 {
         compute_final_hash_v0(header)
     } else {
@@ -927,7 +927,7 @@ mod tests {
             }
         });
 
-        assert_eq!(compute_final_hash(&block_header).unwrap(), expected_hash);
+        assert_eq!(compute_final_hash(&block_header), expected_hash);
     }
 
     #[test]
@@ -1235,7 +1235,7 @@ mod tests {
         let expected_hash = BlockHash(felt!(
             "0x061e4998d51a248f1d0288d7e17f6287757b0e5e6c5e1e58ddf740616e312134"
         ));
-        assert_eq!(compute_final_hash(&header).unwrap(), expected_hash);
+        assert_eq!(compute_final_hash(&header), expected_hash);
     }
 
     // Source
@@ -1271,9 +1271,6 @@ mod tests {
         )
         .unwrap();
 
-        assert_eq!(
-            compute_final_hash(&block_header_data).unwrap(),
-            expected_hash
-        );
+        assert_eq!(compute_final_hash(&block_header_data), expected_hash);
     }
 }

crates/pathfinder/src/sync.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context;
-use error::{SyncError, SyncError2};
+use error::SyncError;
 use futures::{pin_mut, Stream, StreamExt};
 use p2p::client::peer_agnostic::Client as P2PClient;
 use p2p::PeerData;
@@ -63,9 +63,9 @@ impl Sync {
         self.track_sync(next, parent_hash).await
     }
 
-    async fn handle_recoverable_error(&self, err: &PeerData<error::SyncError2>) {
+    async fn handle_recoverable_error(&self, err: &error::SyncError) {
         // TODO
-        tracing::debug!(?err, "Log and punish as appropriate");
+        tracing::debug!(%err, "Log and punish as appropriate");
     }
 
     /// Retry forever until a valid L1 checkpoint is retrieved
@@ -126,13 +126,13 @@ impl Sync {
                     tracing::debug!(?continue_from, "Checkpoint sync complete");
                     continue_from
                 }
-                Err(SyncError::Other(error)) => {
-                    tracing::error!(?error, "Stopping checkpoint sync");
-                    return Err(error);
+                Err(SyncError::Fatal(mut error)) => {
+                    tracing::error!(%error, "Stopping checkpoint sync");
+                    return Err(error.take_or_deep_clone());
                 }
                 Err(error) => {
-                    tracing::debug!(?error, "Restarting checkpoint sync");
-                    self.handle_recoverable_error(&error.into_v2()).await;
+                    tracing::debug!(%error, "Restarting checkpoint sync");
+                    self.handle_recoverable_error(&error).await;
                     continue;
                 }
             };
@@ -180,19 +180,16 @@ impl Sync {
             .run(next, parent_hash, self.fgw_client.clone())
             .await;
 
-            match &mut result {
+            match result {
                 Ok(_) => tracing::debug!("Restarting track sync: unexpected end of Block stream"),
-                Err(PeerData {
-                    data: SyncError2::Other(error),
-                    ..
-                }) => {
-                    tracing::error!(?error, "Stopping track sync");
+                Err(SyncError::Fatal(mut error)) => {
+                    tracing::error!(%error, "Stopping track sync");
                     use pathfinder_common::error::AnyhowExt;
                     return Err(error.take_or_deep_clone());
                 }
                 Err(error) => {
-                    tracing::debug!(error=?error.data, "Restarting track sync");
-                    self.handle_recoverable_error(error).await;
+                    tracing::debug!(%error, "Restarting track sync");
+                    self.handle_recoverable_error(&error).await;
                 }
             }
         }