fix(coprocessor): host-listener, block time in computations.created_at & .schedule_order (#1409)

use block time instead of the default now() == transaction time

Author: rudy-6-4

coprocessor/fhevm-engine/.sqlx/query-0fe024db8ce4cb065633e9f02bd96e67b62460c4d725030ce6fb3b488d0541a7.json renamed to coprocessor/fhevm-engine/.sqlx/query-596c0373ac6af8afaf8893e772a175ab88dc83c8788d8ca57c72fb1e42a00cd2.json

@@ -5,24 +5,22 @@ use alloy::pubsub::SubscriptionStream;
 use alloy::rpc::types::{Block, BlockNumberOrTag, Filter, Header, Log};
 use alloy::sol_types::SolEventInterface;
 use anyhow::{anyhow, Result};
-use fhevm_engine_common::telemetry;
+use clap::Parser;
 use futures_util::stream::StreamExt;
+use rustls;
+use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime};
 use sqlx::types::Uuid;
+use tokio::sync::RwLock;
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info, warn, Level};
 
 use std::collections::{HashSet, VecDeque};
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::RwLock;
-use tracing::{error, info, warn, Level};
-
-use clap::Parser;
-
-use rustls;
-
-use tokio_util::sync::CancellationToken;
 
 use fhevm_engine_common::healthz_server::HttpServer as HealthHttpServer;
+use fhevm_engine_common::telemetry;
 use fhevm_engine_common::types::{BlockchainProvider, Handle};
 use fhevm_engine_common::utils::{DatabaseURL, HeartBeat};
 
@@ -878,6 +876,18 @@ async fn db_insert_block_no_retry(
     let block_hash = block_logs.summary.hash;
     let block_number = block_logs.summary.number;
     let mut catchup_insertion = 0;
+    let block_timestamp = OffsetDateTime::from_unix_timestamp(
+        block_logs.summary.timestamp as i64,
+    )
+    .unwrap_or_else(|_| {
+        error!(
+            timestamp = block_logs.summary.timestamp,
+            "Invalid block timestamp, using now",
+        );
+        OffsetDateTime::now_utc()
+    });
+    let block_timestamp =
+        PrimitiveDateTime::new(block_timestamp.date(), block_timestamp.time());
     for log in &block_logs.logs {
         let current_address = Some(log.inner.address);
         let is_acl_address = &current_address == acl_contract_address;
@@ -919,6 +929,7 @@ async fn db_insert_block_no_retry(
                     transaction_hash: log.transaction_hash,
                     is_allowed: false, // updated in the next loop
                     block_number,
+                    block_timestamp,
                 };
                 tfhe_event_log.push(log);
                 continue;

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

@@ -87,6 +87,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 type Transaction<'l> = sqlx::Transaction<'l, Postgres>;
@@ -289,9 +290,11 @@ impl Database {
                 is_scalar,
                 dependence_chain_id,
                 transaction_id,
-                is_allowed
+                is_allowed,
+                created_at,
+                schedule_order
             )
-            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9)
             ON CONFLICT (tenant_id, output_handle, transaction_id) DO NOTHING
             "#,
             tenant_id as i32,
@@ -302,6 +305,7 @@ impl Database {
             bucket.to_vec(),
             log.transaction_hash.map(|txh| txh.to_vec()),
             log.is_allowed,
+            log.block_timestamp,
         );
         query
             .execute(tx.deref_mut())

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

@@ -6,6 +6,7 @@ use host_listener::database::tfhe_event_propagate::{
     ClearConst, Database as ListenerDatabase, Handle, LogTfhe, TransactionHash,
 };
 use rand::Rng;
+use sqlx::types::time::PrimitiveDateTime;
 use sqlx::Postgres;
 use std::sync::Arc;
 use tracing::info;
@@ -231,6 +232,7 @@ pub async fn generate_trivial_encrypt(
         transaction_hash: Some(transaction_hash),
         is_allowed,
         block_number: 1,
+        block_timestamp: PrimitiveDateTime::MAX,
     };
     let mut tx = listener_event_to_db.new_transaction().await?;
     listener_event_to_db
@@ -412,6 +414,7 @@ pub async fn insert_tfhe_event(
         transaction_hash: Some(transaction_hash),
         is_allowed,
         block_number: 1,
+        block_timestamp: PrimitiveDateTime::MAX,
     };
     listener_event_to_db
         .insert_tfhe_event(&mut tx, &log)

coprocessor/fhevm-engine/stress-test-generator/src/utils.rs

@@ -1,5 +1,6 @@
 use alloy::primitives::{FixedBytes, Log};
 use bigdecimal::num_bigint::BigInt;
+use sqlx::types::time::PrimitiveDateTime;
 
 use fhevm_engine_common::types::AllowEvents;
 use host_listener::contracts::TfheContract;
@@ -42,6 +43,7 @@ async fn insert_tfhe_event(
         transaction_hash: log.transaction_hash,
         is_allowed,
         block_number: log.block_number.unwrap_or(0),
+        block_timestamp: PrimitiveDateTime::MAX,
     };
     db.insert_tfhe_event(tx, &event).await
 }