fix(coprocessor): support multichain

Author: rudy-6-4

coprocessor/fhevm-engine/host-listener/src/cmd/mod.rs

@@ -40,6 +40,7 @@ pub const DEFAULT_DEPENDENCE_BY_CONNEXITY: bool = false;
 pub const DEFAULT_DEPENDENCE_CROSS_BLOCK: bool = true;
 
 const TIMEOUT_REQUEST_ON_WEBSOCKET: u64 = 15;
+pub const MINIMUM_DELAY_FOR_FINALIZATION_IN_BLOCKS: u64 = 5;
 
 #[derive(Parser, Debug, Clone)]
 #[command(version, about, long_about = None)]
@@ -459,6 +460,7 @@ impl InfiniteLogIter {
                 logs: std::mem::take(&mut current_logs),
                 summary,
                 catchup: true,
+                finalized: self.catchup_finalization_in_blocks > MINIMUM_DELAY_FOR_FINALIZATION_IN_BLOCKS,
             };
             blocks_logs.push(block_logs);
         }
@@ -731,6 +733,7 @@ impl InfiniteLogIter {
                 logs,
                 summary: missing_block,
                 catchup: true,
+                finalized: false, // let catchups with finality conditions do the finalize later
             });
             self.block_history.add_block(missing_block);
         }
@@ -825,6 +828,7 @@ impl InfiniteLogIter {
             logs: self.get_logs_at_hash(block_header.hash).await?,
             summary: block_header.into(),
             catchup: false,
+            finalized: false,
         })
     }
 

coprocessor/fhevm-engine/host-listener/src/database/ingest.rs

@@ -19,6 +19,7 @@ pub struct BlockLogs<T> {
     pub logs: Vec<T>,
     pub summary: BlockSummary,
     pub catchup: bool,
+    pub finalized: bool,
 }
 
 #[derive(Copy, Clone, Debug)]
@@ -309,8 +310,7 @@ pub async fn ingest_block_logs(
             info!(block_number, catchup_insertion, "Catchup inserted events");
         }
     }
-
-    db.mark_block_as_valid(&mut tx, &block_logs.summary).await?;
+    db.mark_block_as_valid(&mut tx, &block_logs.summary, block_logs.finalized).await?;
     if at_least_one_insertion {
         db.update_dependence_chain(
             &mut tx,

coprocessor/fhevm-engine/host-listener/src/database/tfhe_event_propagate.rs

@@ -495,19 +495,45 @@ impl Database {
         &self,
         tx: &mut Transaction<'_>,
         block_summary: &BlockSummary,
+        finalized: bool,
     ) -> Result<(), SqlxError> {
+        let status = if finalized { "finalized" } else { "pending" };
+        // 1. Insert if not exists (never overwrites existing row)
         sqlx::query!(
             r#"
-            INSERT INTO host_chain_blocks_valid (chain_id, block_hash, block_number)
-            VALUES ($1, $2, $3)
-            ON CONFLICT (chain_id, block_hash) DO NOTHING;
+            INSERT INTO host_chain_blocks_valid (chain_id, block_hash, block_number, block_status, ancestor_hash)
+            VALUES ($1, $2, $3, $4, $5)
+            ON CONFLICT (chain_id, block_hash) DO NOTHING
+            ;
             "#,
             self.chain_id.as_i64(),
             block_summary.hash.to_vec(),
             block_summary.number as i64,
+            status,
+            block_summary.parent_hash.to_vec(),
         )
         .execute(tx.deref_mut())
         .await?;
+
+        // 2. Update to finalized or orphan if needed
+        if finalized {
+            sqlx::query!(
+                r#"
+                UPDATE host_chain_blocks_valid
+                SET block_status = CASE
+                    WHEN block_hash = $2
+                        THEN 'finalized'
+                        ELSE 'orphan'
+                    END
+                WHERE block_status = 'pending' AND block_number = $3 AND chain_id = $1
+                "#,
+                self.chain_id as i64,
+                block_summary.hash.to_vec(),
+                block_summary.number as i64,
+            )
+            .execute(tx.deref_mut())
+            .await?;
+        }
         Ok(())
     }
 

coprocessor/fhevm-engine/host-listener/src/poller/mod.rs

@@ -272,6 +272,7 @@ pub async fn run_poller(config: PollerConfig) -> Result<()> {
                 logs,
                 summary,
                 catchup: true,
+                finalized: config.finality_lag > 5,
             };
 
             let ingest_options = IngestOptions {

coprocessor/fhevm-engine/host-listener/tests/host_listener_integration_tests.rs

@@ -944,6 +944,56 @@ async fn test_listener_restart_and_chain_reorg() -> Result<(), anyhow::Error> {
     test_listener_no_event_loss(true, true).await
 }
 
+async fn check_finalization_status(
+    setup: Setup
+) {
+    let provider = ProviderBuilder::new()
+                .wallet(setup.wallets[0].clone())
+                .connect_ws(WsConnect::new(setup.args.url.to_string()))
+                .await
+                .unwrap();
+    // Verify block finalization status: for each block number, one should be finalized and others orphaned
+    let blocks = sqlx::query!(
+        "SELECT block_number, block_hash, block_status FROM host_chain_blocks_valid"
+    )
+    .fetch_all(&setup.db_pool)
+    .await;
+
+    let blocks = blocks.expect("Failed to fetch blocks from database");
+
+    let mut blocks_by_number: std::collections::HashMap<i64, Vec<(Vec<u8>, String)>> = std::collections::HashMap::new();
+    for block in blocks {
+        blocks_by_number
+            .entry(block.block_number)
+            .or_insert_with(Vec::new)
+            .push((block.block_hash, block.block_status));
+    }
+
+    for (block_number, block_variants) in blocks_by_number.iter() {
+        let finalized_count = block_variants.iter().filter(|(_, status)| status == "finalized").count();
+        let orphan_count = block_variants.iter().filter(|(_, status)| status == "orphan").count();
+        assert_eq!(
+            finalized_count, 1,
+            "Block {} should have exactly one finalized variant, found {}",
+            block_number, finalized_count
+        );
+        let finalized_hash = block_variants.iter().find(|(_, status)| status == "finalized").map(|(hash, _)| hash).unwrap();
+        assert_eq!(
+            orphan_count,
+            block_variants.len() - 1,
+            "Block {} should have remaining variants as orphan",
+            block_number
+        );
+        let expected_hash= provider.get_block_by_number((*block_number as u64).into()).await.unwrap().unwrap().header.hash;
+        assert_eq!(
+            &expected_hash.0,
+            finalized_hash.as_slice(),
+            "Finalized block hash for block {} does not match expected",
+            block_number
+        );
+    }
+}
+
 async fn test_listener_no_event_loss(
     kill: bool,
     reorg: bool,
@@ -1046,6 +1096,8 @@ async fn test_listener_no_event_loss(
     }
     assert_eq!(tfhe_events_count, expected_tfhe_events);
     assert_eq!(acl_events_count, expected_acl_events);
+
+    check_finalization_status(setup).await;
     Ok(())
 }
 

coprocessor/fhevm-engine/transaction-sender/src/ops/delegate_user_decrypt.rs

@@ -257,11 +257,9 @@ impl<P: Provider<Ethereum> + Clone + 'static> DelegateUserDecryptOperation<P> {
             let block_status = if let Some(status) = blocks_status.get(&delegation.block_number) {
                 *status
             } else {
-                let status = match self.get_block_hash(delegation.block_number).await {
-                    Ok(block_hash) if delegation.block_hash == block_hash.to_vec() => {
-                        BlockStatus::Stable
-                    }
-                    Ok(block_hash) => {
+                let status = self.get_block_status(&delegation).await;
+                match status {
+                    BlockStatus::Dismissed => {
                         warn!(
                             delegation_block_hash = ?delegation.block_hash,
                             block_hash = ?block_hash.to_vec(),
@@ -270,16 +268,15 @@ impl<P: Provider<Ethereum> + Clone + 'static> DelegateUserDecryptOperation<P> {
                         );
                         // ignoring delegation due to reorg, will be marked as reorg_out
                         reorg_out_block.push(delegation.block_hash.clone());
-                        BlockStatus::Dismissed
                     }
-                    Err(_) => {
+                    BlockStatus::Unknown => {
                         error!(
                             block_number = delegation.block_number,
                             "Cannot get block hash for delegation, will retry next block"
                         );
                         unsure_block.push(delegation.block_number);
-                        BlockStatus::Unknown
                     }
+                    BlockStatus::Stable => (),
                 };
                 blocks_status.insert(delegation.block_number, status);
                 status
@@ -315,17 +312,32 @@ impl<P: Provider<Ethereum> + Clone + 'static> DelegateUserDecryptOperation<P> {
         Ok((stable_delegations, reorg_out_block))
     }
 
-    async fn get_block_hash(&self, block_number: u64) -> Result<BlockHash> {
-        let search_block = BlockNumberOrTag::Number(block_number);
-        let some_block = self
-            .host_chain_provider
-            .get_block_by_number(search_block)
-            .await?;
-        let Some(block) = some_block else {
-            error!(block_number, "A past block cannot be found by number");
-            anyhow::bail!("Cannot get past block by number, giving up");
-        };
-        Ok(block.header.hash)
+    async fn get_block_status(&self, delegation: &DelegationRow) -> BlockStatus {
+        let status = sqlx::query!(r#"
+            SELECT block_status
+            FROM host_chain_blocks_valid
+            WHERE block_hash = $2 AND chain_id = $1
+            "#,
+            delegation.host_chain_id as i64, // TODO: chain id
+            delegation.block_hash, //.as_bytes(),
+        )
+        .fetch_one(&self.db_pool)
+        .await;
+        match status {
+            Ok(record) => match record.block_status.as_str() {
+                "finalized" => BlockStatus::Stable,
+                "orphaned" => BlockStatus::Dismissed,
+                _ => BlockStatus::Unknown,
+            },
+            Err(e) => {
+                error!(
+                    %e,
+                    ?delegation,
+                    "Error querying block status from database"
+                );
+                BlockStatus::Unknown
+            }
+        }
     }
 
     async fn wait_last_block_number(&self) -> Result<u64> {

test-suite/fhevm/docker-compose/coprocessor-docker-compose.yml

@@ -203,6 +203,7 @@ services:
       - --s3-recheck-duration=2s
       - --s3-regular-recheck-duration=120s
       - --enable-compression
+      - --private-key=${TX_SENDER_PRIVATE_KEY}
     depends_on:
       coprocessor-db-migration:
         condition: service_completed_successfully