Merge pull request #1871 from subspace/operator-er

Use the ER from the canonical consensus chain when construct bundle

domains/client/domain-operator/src/aux_schema.rs

@@ -24,9 +24,12 @@ const BAD_RECEIPT_MISMATCH_INFO: &[u8] = b"bad_receipt_mismatch_info";
 /// NOTE: Unbounded but the size is not expected to be large.
 const BAD_RECEIPT_NUMBERS: &[u8] = b"bad_receipt_numbers";
 
-/// domain_block_hash => consensus_block_hash
+/// domain_block_hash => Vec<consensus_block_hash>
 ///
 /// Updated only when there is a new domain block produced
+///
+/// NOTE: different consensus blocks may derive the exact same domain block, thus one domain block may
+/// mapping to multiple consensus block.
 const CONSENSUS_HASH: &[u8] = b"consensus_block_hash";
 
 /// domain_block_hash => latest_consensus_block_hash
@@ -128,6 +131,13 @@ where
         }
     }
 
+    let consensus_hashes = {
+        let mut hashes =
+            consensus_block_hash_for::<Backend, Block::Hash, CBlock::Hash>(backend, domain_hash)?;
+        hashes.push(consensus_hash);
+        hashes
+    };
+
     backend.insert_aux(
         &[
             (
@@ -137,7 +147,7 @@ where
             // TODO: prune the stale mappings.
             (
                 (CONSENSUS_HASH, domain_hash).encode().as_slice(),
-                consensus_hash.encode().as_slice(),
+                consensus_hashes.encode().as_slice(),
             ),
             (
                 block_number_key.as_slice(),
@@ -155,25 +165,6 @@ where
     )
 }
 
-/// Load the execution receipt for given domain block hash.
-pub(super) fn load_execution_receipt_by_domain_hash<Backend, Block, CBlock>(
-    backend: &Backend,
-    domain_hash: Block::Hash,
-) -> ClientResult<Option<ExecutionReceiptFor<Block, CBlock>>>
-where
-    Backend: AuxStore,
-    Block: BlockT,
-    CBlock: BlockT,
-{
-    match consensus_block_hash_for::<Backend, Block::Hash, CBlock::Hash>(backend, domain_hash)? {
-        Some(consensus_block_hash) => load_decode(
-            backend,
-            execution_receipt_key(consensus_block_hash).as_slice(),
-        ),
-        None => Ok(None),
-    }
-}
-
 /// Load the execution receipt for given consensus block hash.
 pub(super) fn load_execution_receipt<Backend, Block, CBlock>(
     backend: &Backend,
@@ -254,13 +245,16 @@ where
 pub(super) fn consensus_block_hash_for<Backend, Hash, PHash>(
     backend: &Backend,
     domain_hash: Hash,
-) -> ClientResult<Option<PHash>>
+) -> ClientResult<Vec<PHash>>
 where
     Backend: AuxStore,
     Hash: Encode,
     PHash: Decode,
 {
-    load_decode(backend, (CONSENSUS_HASH, domain_hash).encode().as_slice())
+    Ok(
+        load_decode(backend, (CONSENSUS_HASH, domain_hash).encode().as_slice())?
+            .unwrap_or_default(),
+    )
 }
 
 // TODO: Unlock once domain test infra is workable again.

domains/client/domain-operator/src/domain_block_processor.rs

@@ -341,15 +341,12 @@ where
                 *genesis_header.state_root(),
             )
         } else {
-            crate::aux_schema::load_execution_receipt_by_domain_hash::<_, Block, CBlock>(
+            crate::load_execution_receipt_by_domain_hash::<Block, CBlock, _, _>(
                 &*self.client,
+                &self.consensus_client,
                 parent_hash,
+                parent_number,
             )?
-            .ok_or_else(|| {
-                sp_blockchain::Error::Backend(format!(
-                    "Receipt of domain block #{parent_number},{parent_hash} not found"
-                ))
-            })?
         };
 
         // Get the accumulated transaction fee of all transactions included in the block

domains/client/domain-operator/src/domain_bundle_producer.rs

@@ -149,12 +149,12 @@ where
             global_randomness,
         } = slot_info;
 
-        let best_receipt_is_written = crate::aux_schema::consensus_block_hash_for::<
+        let best_receipt_is_written = !crate::aux_schema::consensus_block_hash_for::<
             _,
             _,
             CBlock::Hash,
         >(&*self.client, self.client.info().best_hash)?
-        .is_some();
+        .is_empty();
 
         // TODO: remove once the receipt generation can be done before the domain block is
         // committed to the database, in other words, only when the receipt of block N+1 has

domains/client/domain-operator/src/domain_bundle_proposer.rs

@@ -188,12 +188,12 @@ where
             "Collecting receipts at {parent_chain_block_hash:?}"
         );
 
-        let header_block_receipt_is_written = crate::aux_schema::consensus_block_hash_for::<
+        let header_block_receipt_is_written = !crate::aux_schema::consensus_block_hash_for::<
             _,
             _,
             CBlock::Hash,
         >(&*self.client, header_hash)?
-        .is_some();
+        .is_empty();
 
         // TODO: remove once the receipt generation can be done before the domain block is
         // committed to the database, in other words, only when the receipt of block N+1 has
@@ -229,14 +229,11 @@ where
             ))
         })?;
 
-        crate::aux_schema::load_execution_receipt_by_domain_hash::<_, Block, CBlock>(
+        crate::load_execution_receipt_by_domain_hash::<Block, CBlock, _, _>(
             &*self.client,
+            &self.consensus_client,
             domain_hash,
-        )?
-        .ok_or_else(|| {
-            sp_blockchain::Error::Backend(format!(
-                "Receipt of domain block #{receipt_number},{domain_hash} not found"
-            ))
-        })
+            receipt_number,
+        )
     }
 }

domains/client/domain-operator/src/lib.rs

@@ -84,7 +84,7 @@ pub use self::utils::{DomainBlockImportNotification, DomainImportNotifications};
 use crate::utils::BlockInfo;
 use futures::channel::mpsc;
 use futures::Stream;
-use sc_client_api::BlockImportNotification;
+use sc_client_api::{AuxStore, BlockImportNotification};
 use sc_utils::mpsc::TracingUnboundedSender;
 use sp_api::ProvideRuntimeApi;
 use sp_blockchain::HeaderBackend;
@@ -239,3 +239,57 @@ where
 
     Ok(leaves.into_iter().rev().take(MAX_ACTIVE_LEAVES).collect())
 }
+
+pub(crate) fn load_execution_receipt_by_domain_hash<Block, CBlock, Client, CClient>(
+    domain_client: &Client,
+    consensus_client: &Arc<CClient>,
+    domain_hash: Block::Hash,
+    domain_number: NumberFor<Block>,
+) -> Result<ExecutionReceiptFor<Block, CBlock>, sp_blockchain::Error>
+where
+    Block: BlockT,
+    CBlock: BlockT,
+    Client: AuxStore,
+    CClient: HeaderBackend<CBlock>,
+{
+    let not_found_error = || {
+        sp_blockchain::Error::Backend(format!(
+            "Receipt for domain block {domain_hash}#{domain_number} not found"
+        ))
+    };
+
+    // Get all the consensus blocks that mapped to `domain_hash`
+    let consensus_block_hashes = crate::aux_schema::consensus_block_hash_for::<_, _, CBlock::Hash>(
+        domain_client,
+        domain_hash,
+    )?;
+
+    // Get the consensus block that is in the current canonical consensus chain
+    let consensus_block_hash = match consensus_block_hashes.len() {
+        0 => return Err(not_found_error()),
+        1 => consensus_block_hashes[0],
+        _ => {
+            let mut canonical_consensus_hash = None;
+            for hash in consensus_block_hashes {
+                // Check if `hash` is in the canonical chain
+                let block_number = consensus_client.number(hash)?.ok_or_else(not_found_error)?;
+                let canonical_block_hash = consensus_client
+                    .hash(block_number)?
+                    .ok_or_else(not_found_error)?;
+
+                if canonical_block_hash == hash {
+                    canonical_consensus_hash.replace(hash);
+                    break;
+                }
+            }
+            canonical_consensus_hash.ok_or_else(not_found_error)?
+        }
+    };
+
+    // Get receipt by consensus block hash
+    crate::aux_schema::load_execution_receipt::<_, Block, CBlock>(
+        domain_client,
+        consensus_block_hash,
+    )?
+    .ok_or_else(not_found_error)
+}

domains/client/domain-operator/src/tests.rs

@@ -20,9 +20,10 @@ use sp_domains::fraud_proof::{ExecutionPhase, FraudProof, InvalidStateTransition
 use sp_domains::transaction::InvalidTransactionCode;
 use sp_domains::{Bundle, DomainId, DomainsApi};
 use sp_runtime::generic::{BlockId, Digest, DigestItem};
-use sp_runtime::traits::{BlakeTwo256, Header as HeaderT};
+use sp_runtime::traits::{BlakeTwo256, Block as BlockT, Header as HeaderT};
 use sp_runtime::OpaqueExtrinsic;
 use subspace_fraud_proof::invalid_state_transition_proof::ExecutionProver;
+use subspace_runtime_primitives::opaque::Block as CBlock;
 use subspace_runtime_primitives::Balance;
 use subspace_test_service::{
     produce_block_with, produce_blocks, produce_blocks_until, MockConsensusNode,
@@ -1885,3 +1886,96 @@ async fn test_domain_transaction_fee_and_operator_reward() {
     // All the transaction fee is collected as operator reward
     assert_eq!(alice_free_balance_changes, receipt.total_rewards);
 }
+
+#[substrate_test_utils::test(flavor = "multi_thread")]
+async fn test_multiple_consensus_blocks_derive_same_domain_block() {
+    let directory = TempDir::new().expect("Must be able to create temporary directory");
+
+    let mut builder = sc_cli::LoggerBuilder::new("");
+    builder.with_colors(false);
+    let _ = builder.init();
+
+    let tokio_handle = tokio::runtime::Handle::current();
+
+    // Start Ferdie
+    let mut ferdie = MockConsensusNode::run(
+        tokio_handle.clone(),
+        Ferdie,
+        BasePath::new(directory.path().join("ferdie")),
+    );
+
+    // Produce 1 consensus block to initialize genesis domain
+    ferdie.produce_block_with_slot(1.into()).await.unwrap();
+
+    // Run Alice (a evm domain authority node)
+    let mut alice = domain_test_service::DomainNodeBuilder::new(
+        tokio_handle.clone(),
+        Alice,
+        BasePath::new(directory.path().join("alice")),
+    )
+    .build_evm_node(Role::Authority, GENESIS_DOMAIN_ID, &mut ferdie)
+    .await;
+
+    produce_blocks!(ferdie, alice, 3).await.unwrap();
+    let common_block_hash = ferdie.client.info().best_hash;
+    let bundle_to_tx = |opaque_bundle| {
+        subspace_test_runtime::UncheckedExtrinsic::new_unsigned(
+            pallet_domains::Call::submit_bundle { opaque_bundle }.into(),
+        )
+        .into()
+    };
+
+    // Fork A
+    let (slot, bundle) = ferdie.produce_slot_and_wait_for_bundle_submission().await;
+    // Include one more extrinsic in fork A such that we can have a different consensus block
+    let remark_tx = subspace_test_runtime::UncheckedExtrinsic::new_unsigned(
+        frame_system::Call::remark { remark: vec![0; 8] }.into(),
+    )
+    .into();
+    let consensus_block_hash_fork_a = ferdie
+        .produce_block_with_slot_at(
+            slot,
+            common_block_hash,
+            Some(vec![bundle_to_tx(bundle.clone().unwrap()), remark_tx]),
+        )
+        .await
+        .unwrap();
+
+    // Fork B
+    let consensus_block_hash_fork_b = ferdie
+        .produce_block_with_slot_at(
+            slot,
+            common_block_hash,
+            Some(vec![bundle_to_tx(bundle.unwrap())]),
+        )
+        .await
+        .unwrap();
+
+    // The same domain block mapped to 2 different consensus blocks
+    let consensus_best_hashes = crate::aux_schema::consensus_block_hash_for::<
+        _,
+        _,
+        <CBlock as BlockT>::Hash,
+    >(&*alice.client, alice.client.info().best_hash)
+    .unwrap();
+    assert_eq!(
+        consensus_best_hashes,
+        vec![consensus_block_hash_fork_a, consensus_block_hash_fork_b]
+    );
+    assert_ne!(consensus_block_hash_fork_a, consensus_block_hash_fork_b);
+
+    // Produce one more block at fork A to make it the canonical chain and the operator
+    // should submit the ER of fork A
+    let (slot, bundle) = ferdie.produce_slot_and_wait_for_bundle_submission().await;
+    ferdie
+        .produce_block_with_slot_at(slot, consensus_block_hash_fork_a, None)
+        .await
+        .unwrap();
+    assert_eq!(
+        bundle.unwrap().into_receipt().consensus_block_hash,
+        consensus_block_hash_fork_a
+    );
+
+    // Simply produce more block
+    produce_blocks!(ferdie, alice, 3).await.unwrap();
+}