feat(coprocessor): host-listener, dependents for dependency_chain

Author: rudy-6-4

coprocessor/fhevm-engine/.sqlx/query-a287f23e7450a0f1f52d25ad518cdd3ef227e24e2bb4e7212dd2f60347349de3.json

@@ -274,7 +274,9 @@ fn grouping_to_chains(ordered_txs: &mut [Transaction]) -> OrderedChains {
                         size: 0,
                         before_size: 0,
                         dependencies: vec![],
+                        dependents: vec![],
                         allowed_handle: tx.allowed_handle.clone(), // needed to publish in cache
+                        new_chain: false,
                     };
                     ordered_chains_hash.push(new_chain.hash);
                     e.insert(new_chain);
@@ -296,7 +298,9 @@ fn grouping_to_chains(ordered_txs: &mut [Transaction]) -> OrderedChains {
                 size: tx.size,
                 before_size,
                 dependencies,
+                dependents: vec![],
                 allowed_handle: tx.allowed_handle.clone(),
+                new_chain: true,
             };
             ordered_chains_hash.push(new_chain.hash);
             chains.insert(new_chain.hash, new_chain);
@@ -305,6 +309,23 @@ fn grouping_to_chains(ordered_txs: &mut [Transaction]) -> OrderedChains {
             used_tx.insert(tx.tx_hash, tx);
         }
     }
+    // compute dependents field
+    for chain_hash in ordered_chains_hash.iter() {
+        let Some(chain) = chains.get(chain_hash) else {
+            continue;
+        };
+        if !chain.new_chain {
+            continue;
+        }
+        for dep in chain.dependencies.clone() {
+            if let Some(dep_chain) = chains.get_mut(&dep) {
+                if !dep_chain.new_chain {
+                    continue;
+                }
+                dep_chain.dependents.push(*chain_hash);
+            }
+        }
+    }
     ordered_chains_hash
         .iter()
         .filter_map(|hash| chains.remove(hash))
@@ -632,15 +653,20 @@ mod tests {
         assert_eq!(chains[0].dependencies.len(), 0);
         assert_eq!(chains[1].dependencies.len(), 0);
         assert_eq!(chains[2].dependencies.len(), 2);
+        assert_eq!(chains[0].dependents, vec![tx3]);
+        assert_eq!(chains[1].dependents, vec![tx3]);
+        assert!(chains[2].dependents.is_empty());
     }
 
     fn past_chain(last_byte: u8) -> Chain {
         Chain {
             hash: TransactionHash::with_last_byte(last_byte),
             dependencies: vec![],
+            dependents: vec![],
             size: 1,
             before_size: 0,
             allowed_handle: vec![],
+            new_chain: false,
         }
     }
 

coprocessor/fhevm-engine/.sqlx/query-d1e5c42539bb901184fe1f4690e62c26850ba790fe0c8d6bdba3adfa18dce89b.json

@@ -160,8 +160,13 @@ pub async fn ingest_block_logs(
 
     db.mark_block_as_valid(&mut tx, &block_logs.summary).await?;
     if at_least_one_insertion {
-        db.update_dependence_chain(&mut tx, chains, block_timestamp)
-            .await?;
+        db.update_dependence_chain(
+            &mut tx,
+            chains,
+            block_timestamp,
+            &block_logs.summary,
+        )
+        .await?;
     }
     tx.commit().await
 }

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

@@ -43,9 +43,11 @@ pub type ChainHash = TransactionHash;
 pub struct Chain {
     pub hash: ChainHash,
     pub dependencies: Vec<ChainHash>,
+    pub dependents: Vec<ChainHash>,
     pub allowed_handle: Vec<Handle>,
     pub size: usize,
     pub before_size: usize,
+    pub new_chain: bool,
 }
 pub type ChainCache = RwLock<lru::LruCache<Handle, ChainHash>>;
 pub type OrderedChains = Vec<Chain>;
@@ -764,40 +766,43 @@ impl Database {
         tx: &mut Transaction<'_>,
         chains: OrderedChains,
         block_timestamp: PrimitiveDateTime,
+        block_summary: &BlockSummary,
     ) -> Result<(), SqlxError> {
-        if chains.is_empty() {
-            return Ok(());
-        }
-        let chains_hash =
-            chains.iter().map(|c| c.hash.to_vec()).collect::<Vec<_>>();
-        let dependency_counts =
-            chains.iter().map(|c| c.dependencies.len() as i64).collect::<Vec<_>>();
-        let timestamps: Vec<PrimitiveDateTime> = chains
-            .iter()
-            .map(|c| {
-                block_timestamp.saturating_add(TimeDuration::microseconds(
-                    c.before_size as i64,
-                ))
-            })
-            .collect();
-        let query = sqlx::query!(
-            r#"
-            INSERT INTO dependence_chain(
-                dependence_chain_id,
-                status,
+        for chain in chains {
+            let last_updated_at = block_timestamp.saturating_add(
+                TimeDuration::microseconds(chain.before_size as i64),
+            );
+            let dependents = chain
+                .dependents
+                .iter()
+                .map(|h| h.to_vec())
+                .collect::<Vec<_>>();
+            sqlx::query!(
+                r#"
+                INSERT INTO dependence_chain(
+                    dependence_chain_id,
+                    status,
+                    last_updated_at,
+                    dependency_count,
+                    dependents,
+                    block_hash,
+                    block_height
+                ) VALUES (
+                  $1, 'updated', $2::timestamp, $3, $4, $5, $6
+                )
+                ON CONFLICT (dependence_chain_id) DO UPDATE
+                SET status = 'updated'
+                "#,
+                chain.hash.to_vec(),
                 last_updated_at,
-                dependency_count
+                chain.dependencies.len() as i64,
+                &dependents,
+                block_summary.hash.to_vec(),
+                block_summary.number as i64,
             )
-            SELECT dcid, 'updated' AS status, ts as last_updated_at, dependency_count
-            FROM unnest($1::bytea[], $2::timestamp[], $3::bigint[]) AS t(dcid, ts, dependency_count)
-            ON CONFLICT (dependence_chain_id) DO UPDATE
-            SET status = 'updated'
-            "#,
-            &chains_hash,
-            &timestamps,
-            &dependency_counts
-        );
-        query.execute(tx.deref_mut()).await?;
+            .execute(tx.deref_mut())
+            .await?;
+        }
         Ok(())
     }
 }