feat(chain): review changes

Author: andrussal

nomos-core/chain-defs/src/block/mod.rs

@@ -1,6 +1,6 @@
 use ::serde::{de::DeserializeOwned, Deserialize, Serialize};
 use bytes::Bytes;
-use ed25519_dalek::{ed25519::signature::Signer as _, SigningKey};
+use ed25519_dalek::SigningKey;
 use groth16::Fr;
 
 use crate::{header::Header, mantle::Transaction};
@@ -11,10 +11,12 @@ pub type TxHash = [u8; 32];
 pub type BlockNumber = u64;
 pub type SessionNumber = u64;
 
-#[derive(Debug, thiserror::Error)]
+#[derive(Clone, Debug, thiserror::Error)]
 pub enum Error {
     #[error("Failed to serialize: {0}")]
     Serialisation(#[from] crate::wire::Error),
+    #[error("Signing error: {0}")]
+    Signing(String),
 }
 
 /// A block proposal
@@ -35,8 +37,9 @@ pub struct References {
 #[derive(Clone, Debug)]
 pub struct Block<Tx> {
     header: Header,
+    signature: ed25519_dalek::Signature,
+    service_reward: Option<Fr>,
     transactions: Vec<Tx>,
-    pub service_reward: Option<Fr>,
 }
 
 impl Proposal {
@@ -58,24 +61,17 @@ impl Proposal {
 
 impl<Tx> Block<Tx> {
     #[must_use]
-    pub const fn new(header: Header, transactions: Vec<Tx>) -> Self {
-        Self {
-            header,
-            transactions,
-            service_reward: None,
-        }
-    }
-
-    #[must_use]
-    pub const fn new_with_service_reward(
+    pub const fn new(
         header: Header,
         transactions: Vec<Tx>,
-        service_reward: Fr,
+        service_reward: Option<Fr>,
+        signature: ed25519_dalek::Signature,
     ) -> Self {
         Self {
             header,
+            signature,
+            service_reward,
             transactions,
-            service_reward: Some(service_reward),
         }
     }
 
@@ -94,6 +90,37 @@ impl<Tx> Block<Tx> {
         self.transactions
     }
 
+    #[must_use]
+    pub const fn service_reward(&self) -> Option<Fr> {
+        self.service_reward
+    }
+
+    #[must_use]
+    pub const fn signature(&self) -> &ed25519_dalek::Signature {
+        &self.signature
+    }
+
+    /// Basic validation
+    pub fn validate(&self) -> Result<(), Error>
+    where
+        Tx: Transaction,
+        Tx::Hash: Into<Fr>,
+    {
+        if !self.header.is_valid_bedrock_version() {
+            return Err(Error::Signing("Invalid header version".to_owned()));
+        }
+
+        if self.transactions.len() > 1024 {
+            return Err(Error::Signing(
+                "Too many transactions (max 1024)".to_owned(),
+            ));
+        }
+
+        // TODO: validate signature?
+
+        Ok(())
+    }
+
     pub fn to_proposal(&self, signing_key: &SigningKey) -> Result<Proposal, Error>
     where
         Tx: Transaction,
@@ -110,8 +137,7 @@ impl<Tx> Block<Tx> {
             mempool_transactions,
         };
 
-        let header_bytes = crate::wire::serialize(&self.header)?;
-        let signature = signing_key.sign(&header_bytes);
+        let signature = self.header.sign(signing_key)?;
 
         Ok(Proposal {
             header: self.header.clone(),

nomos-core/chain-defs/src/block/wire.rs

@@ -73,8 +73,9 @@ impl<'de> Deserialize<'de> for References {
 #[derive(serde::Serialize, serde::Deserialize)]
 struct WireBlock<Tx> {
     header: Header,
-    transactions: Vec<Tx>,
+    signature: ed25519_dalek::Signature,
     service_reward: Option<SerializableFr>,
+    transactions: Vec<Tx>,
 }
 
 impl<Tx: Serialize + Clone> Serialize for Block<Tx> {
@@ -86,8 +87,9 @@ impl<Tx: Serialize + Clone> Serialize for Block<Tx> {
 
         let wire = WireBlock {
             header: self.header.clone(),
-            transactions: self.transactions.clone(),
+            signature: self.signature,
             service_reward,
+            transactions: self.transactions.clone(),
         };
 
         wire.serialize(serializer)
@@ -105,14 +107,17 @@ impl<'de, Tx: Deserialize<'de>> Deserialize<'de> for Block<Tx> {
 
         Ok(Self {
             header: wire.header,
-            transactions: wire.transactions,
+            signature: wire.signature,
             service_reward,
+            transactions: wire.transactions,
         })
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use ed25519_dalek::SigningKey;
+
     use super::Fr;
     use crate::block::{Block, References};
 
@@ -186,7 +191,12 @@ mod tests {
         let transactions = vec!["tx1".to_owned(), "tx2".to_owned()];
         let service_reward = Fr::from(123u64);
 
-        let block = Block::new_with_service_reward(header, transactions, service_reward);
+        let signing_key = SigningKey::from_bytes(&[1u8; 32]);
+        let signature = header
+            .sign(&signing_key)
+            .expect("Header signing should work in test");
+
+        let block = Block::new(header, transactions, Some(service_reward), signature);
 
         let wire_bytes = crate::wire::serialize(&block).expect("Failed to serialize block");
         let deserialized: Block<String> =

nomos-core/chain-defs/src/header/mod.rs

@@ -1,5 +1,6 @@
 use blake2::Digest as _;
 use cryptarchia_engine::Slot;
+use ed25519_dalek::Signer as _;
 use serde::{Deserialize, Serialize};
 
 pub const BEDROCK_VERSION: u8 = 1;
@@ -103,6 +104,14 @@ impl Header {
         self.slot
     }
 
+    pub fn sign(
+        &self,
+        signing_key: &ed25519_dalek::SigningKey,
+    ) -> Result<ed25519_dalek::Signature, crate::block::Error> {
+        let header_bytes = crate::wire::serialize(self)?;
+        Ok(signing_key.sign(&header_bytes))
+    }
+
     #[must_use]
     pub fn new(
         parent_block: HeaderId,

nomos-services/chain-service/src/lib.rs

@@ -92,8 +92,10 @@ pub enum Error {
     Ledger(#[from] nomos_ledger::LedgerError<HeaderId>),
     #[error("Consensus error: {0}")]
     Consensus(#[from] cryptarchia_engine::Error<HeaderId>),
-    #[error("Proposal error: {0}")]
-    Proposal(String),
+    #[error("Serialization error: {0}")]
+    Serialisation(#[from] nomos_core::wire::Error),
+    #[error("Invalid block: {0}")]
+    InvalidBlock(String),
     #[error("Storage error: {0}")]
     Storage(String),
 }
@@ -764,32 +766,48 @@ where
                         }
                     }
 
-                    // TODO: we maybe can keep existing orphan downloading logic but convert Block to Proposal
                     Some(block) = orphan_downloader.next(), if orphan_downloader.should_poll() => {
-                        let header_id= block.header().id();
+                        let header_id = block.header().id();
                         info!("Processing block from orphan downloader: {header_id:?}");
 
                         if cryptarchia.has_block(&block.header().id()) {
                             continue;
                         }
 
-                        match Self::process_block_and_update_state(
-                            cryptarchia.clone(),
-                            &leader,
-                            block.clone(),
-                            &storage_blocks_to_remove,
-                            &relays,
-                            &self.block_subscription_sender,
-                            &self.service_resources_handle.state_updater
-                        ).await {
-                            Ok((new_cryptarchia, new_storage_blocks_to_remove)) => {
-                                cryptarchia = new_cryptarchia;
-                                storage_blocks_to_remove = new_storage_blocks_to_remove;
+                        match block.validate() {
+                            Ok(()) => {
+                                Self::log_received_block(&block);
+
+                                match Self::process_block_and_update_state(
+                                    cryptarchia.clone(),
+                                    &leader,
+                                    block,
+                                    &storage_blocks_to_remove,
+                                    &relays,
+                                    &self.block_subscription_sender,
+                                    &self.service_resources_handle.state_updater
+                                ).await {
+                                    Ok((new_cryptarchia, new_storage_blocks_to_remove)) => {
+                                        cryptarchia = new_cryptarchia;
+                                        storage_blocks_to_remove = new_storage_blocks_to_remove;
+
+                                        orphan_downloader.remove_orphan(&header_id);
+                                        info!(counter.consensus_processed_blocks = 1);
+                                    }
+                                    Err(Error::Ledger(nomos_ledger::LedgerError::ParentNotFound(parent))
+                                        | Error::Consensus(cryptarchia_engine::Error::ParentMissing(parent))) => {
 
-                                info!(counter.consensus_processed_blocks = 1);
+                                        orphan_downloader.enqueue_orphan(header_id, cryptarchia.tip(), cryptarchia.lib());
+                                        error!(target: LOG_TARGET, "Orphan block with parent {:?} not in ledger state", parent);
+                                    }
+                                    Err(e) => {
+                                        error!(target: LOG_TARGET, "Error processing orphan block: {e:?}");
+                                        orphan_downloader.cancel_active_download();
+                                    }
+                                }
                             }
                             Err(e) => {
-                                error!(target: LOG_TARGET, "Error processing orphan downloader block: {e:?}");
+                                error!(target: LOG_TARGET, "Failed to validate orphan block: {e:?}");
                                 orphan_downloader.cancel_active_download();
                             }
                         }
@@ -1139,7 +1157,15 @@ where
                     .collect::<Vec<_>>();
                 let content_id = [0; 32].into(); // TODO: calculate the actual content id
                                                  // TODO: this should probably be a proposal or be transformed into a proposal
-                let block = Block::new(Header::new(parent, content_id, slot, proof), txs);
+                                                 // TODO: Use correct derived one time key
+                let dummy_signing_key = SigningKey::from_bytes(&[1u8; 32]);
+                let header = Header::new(parent, content_id, slot, proof);
+
+                let Ok(signature) = header.sign(&dummy_signing_key) else {
+                    error!("Failed to sign header during block proposal");
+                    return None;
+                };
+                let block = Block::new(header, txs, None, signature);
                 debug!("proposed block with id {:?}", block.header().id());
                 output = Some(block);
             }
@@ -1615,10 +1641,10 @@ where
     BlendService::BroadcastSettings: Clone + Sync,
 {
     // TODO: Use correct derived one time key
-    let signing_key = SigningKey::generate(&mut rand::thread_rng());
+    let signing_key = SigningKey::from_bytes(&rand::random::<[u8; 32]>());
     let proposal = block
         .to_proposal(&signing_key)
-        .map_err(|e| Error::Proposal(format!("Failed to create proposal from block: {e}")))?;
+        .map_err(|e| Error::InvalidBlock(format!("Failed to create proposal from block: {e}")))?;
 
     blend_adapter.publish_proposal(proposal).await;
 
@@ -1634,22 +1660,6 @@ where
     Payload: Send,
     Item: Clone + Transaction<Hash = TxHash> + Send,
 {
-    if !proposal.header().is_valid_bedrock_version() {
-        return Err(Error::Proposal(format!(
-            "Invalid header version: expected all fields = 0x01, got {:?}",
-            proposal.header().version()
-        )));
-    }
-
-    if proposal.references().mempool_transactions.len() > 1024 {
-        return Err(Error::Proposal(
-            "Too many transaction references (max 1024)".into(),
-        ));
-    }
-
-    // TODO: the rest of validations should probably go to
-    // process_block_and_update_state
-
     let needed_hashes: Vec<TxHash> = proposal
         .references()
         .mempool_transactions
@@ -1659,7 +1669,7 @@ where
 
     let reconstructed_transactions = get_transactions_by_hashes(mempool, needed_hashes.clone())
         .await
-        .map_err(|e| Error::Proposal(format!("Failed to get transactions by hashes: {e}")))?;
+        .map_err(|e| Error::InvalidBlock(format!("Failed to get transactions by hashes: {e}")))?;
 
     if reconstructed_transactions.len() != needed_hashes.len() {
         let missing_transactions: Vec<TxHash> = needed_hashes
@@ -1671,18 +1681,26 @@ where
             })
             .collect();
 
-        return Err(Error::Proposal(format!(
+        return Err(Error::InvalidBlock(format!(
             "Failed to reconstruct block: {missing_transactions:?} transactions not found in mempool",
         )));
     }
 
     let header = proposal.header().clone();
     let service_reward = proposal.references().service_reward;
-
-    Ok(match service_reward {
-        Some(reward) => Block::new_with_service_reward(header, reconstructed_transactions, reward),
-        None => Block::new(header, reconstructed_transactions),
-    })
+    let signature = *proposal.signature();
+
+    let block = Block::new(
+        header,
+        reconstructed_transactions,
+        service_reward,
+        signature,
+    );
+    block
+        .validate()
+        .map_err(|e| Error::InvalidBlock(format!("Reconstructed block is invalid: {e}")))?;
+
+    Ok(block)
 }
 
 #[cfg(test)]

nomos-services/chain-service/src/sync/block_provider.rs

@@ -750,7 +750,10 @@ mod tests {
 
             for i in 0..count {
                 let slot = Slot::from(slot_offset + i as u64);
-                let block = self.build_block_with_parent(prev_header, slot);
+                let Some(block) = self.build_block_with_parent(prev_header, slot) else {
+                    error!("Failed to build block with parent");
+                    break;
+                };
                 let header_id = block.header().id();
 
                 blocks.push((block, header_id, prev_header, slot));
@@ -813,11 +816,17 @@ mod tests {
             &self,
             prev_header: HeaderId,
             slot: Slot,
-        ) -> Block<SignedMantleTx> {
-            Block::new(
-                Header::new(prev_header, [0; 32].into(), slot, self.proof.clone()),
-                vec![],
-            )
+        ) -> Option<Block<SignedMantleTx>> {
+            // TODO: Use correct derived one time key
+            let dummy_signing_key = ed25519_dalek::SigningKey::from_bytes(&[1u8; 32]);
+            let header = Header::new(prev_header, [0; 32].into(), slot, self.proof.clone());
+
+            let Ok(signature) = header.sign(&dummy_signing_key) else {
+                error!("Failed to sign header in block provider");
+                return None;
+            };
+
+            Some(Block::new(header, vec![], None, signature))
         }
 
         async fn add_block(