feat(coprocessor): topologic timestamp

Author: rudy-6-4

coprocessor/fhevm-engine/.sqlx/query-04c918d1380072cbacde5fb7937e17c3b50874fe082a6761f25f4a9b1b082bf3.json

@@ -85,6 +85,7 @@ tfhe = { version = "=1.4.0-alpha.4", features = [
 ] }
 tfhe-versionable = "=0.6.2"
 tfhe-zk-pok = "=0.7.4"
+time = "0.3.43"
 tokio = { version = "1.45.0", features = ["full"] }
 tokio-util = "0.7.15"
 tonic = { version = "0.12.3", features = ["server"] }

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

@@ -29,6 +29,7 @@ rustls = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 sqlx = { workspace = true }
+time = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
 tracing = { workspace = true }

coprocessor/fhevm-engine/Cargo.lock

@@ -32,7 +32,7 @@ impl Transaction {
             input_tx: HashSet::with_capacity(3),
             output_tx: HashSet::with_capacity(3),
             linear_chain: tx_hash, //  before coallescing linear tx chains
-            size: 1,
+            size: 0,
         }
     }
 }
@@ -603,6 +603,9 @@ mod tests {
         assert_eq!(logs[3].dependence_chain, tx1);
         assert_eq!(logs[4].dependence_chain, tx3);
         assert_eq!(cache.read().await.len(), 3);
+        assert_eq!(chains[0].before_size, 0);
+        assert_eq!(chains[1].before_size, 0);
+        assert_eq!(chains[2].before_size, 2);
     }
 
     fn past_chain(last_byte: u8) -> Chain {
@@ -765,6 +768,6 @@ mod tests {
         eprintln!("Logs: {:?}", logs);
         let chains = dependence_chains(&mut logs, &cache).await;
         assert_eq!(chains.len(), 6);
-        // assert_eq!(cache.read().await.len(), 66);
+        assert!(chains.iter().all(|c| c.before_size == 0));
     }
 }

coprocessor/fhevm-engine/Cargo.toml

@@ -160,7 +160,8 @@ 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).await?;
+        db.update_dependence_chain(&mut tx, chains, block_timestamp)
+            .await?;
     }
     tx.commit().await
 }

coprocessor/fhevm-engine/host-listener/Cargo.toml

@@ -17,6 +17,7 @@ use sqlx::{PgPool, Postgres};
 use std::ops::DerefMut;
 use std::sync::Arc;
 use std::time::Duration;
+use time::{Duration as TimeDuration, PrimitiveDateTime};
 use tokio::sync::RwLock;
 use tracing::error;
 use tracing::info;
@@ -99,7 +100,7 @@ pub struct LogTfhe {
     pub transaction_hash: Option<TransactionHash>,
     pub is_allowed: bool,
     pub block_number: u64,
-    pub block_timestamp: sqlx::types::time::PrimitiveDateTime,
+    pub block_timestamp: PrimitiveDateTime,
     pub dependence_chain: TransactionHash,
 }
 
@@ -762,24 +763,39 @@ impl Database {
         &self,
         tx: &mut Transaction<'_>,
         chains: OrderedChains,
+        block_timestamp: PrimitiveDateTime,
     ) -> Result<(), SqlxError> {
         if chains.is_empty() {
             return Ok(());
         }
-        let chains = chains.iter().map(|d| d.hash.to_vec()).collect::<Vec<_>>();
+        let chains_hash =
+            chains.iter().map(|c| c.hash.to_vec()).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,
                 last_updated_at
             )
-            VALUES (unnest($1::bytea[]), 'updated', statement_timestamp())
+            SELECT dcid, 'updated' AS status, ts as last_updated_at
+            FROM unnest($1::bytea[], $2::timestamp[]) AS t(dcid, ts)
             ON CONFLICT (dependence_chain_id) DO UPDATE
-            SET status = EXCLUDED.status,
-                last_updated_at = EXCLUDED.last_updated_at
+            SET status = 'updated',
+                last_updated_at = GREATEST(
+                    dependence_chain.last_updated_at,
+                    EXCLUDED.last_updated_at
+                )
             "#,
-            &chains,
+            &chains_hash,
+            &timestamps,
         );
         query.execute(tx.deref_mut()).await?;
         Ok(())