feat(copro): PoC implementation

coprocessor/fhevm-engine/Cargo.toml

@@ -2,6 +2,8 @@
 resolver = "2"
 members = [
     "tfhe-worker",
+    "tfhe-dispatcher",
+    "tfhe-compute-node",
     "fhevm-engine-common",
     "host-listener",
     "gw-listener",

coprocessor/fhevm-engine/fhevm-engine-common/src/types.rs

@@ -402,6 +402,32 @@ impl SupportedFheCiphertexts {
         }
     }
 
+    pub fn deserialize(ct_type: i16, bytes: &[u8]) -> Result<Self, FhevmError> {
+        match ct_type {
+            0 => Ok(SupportedFheCiphertexts::FheBool(safe_deserialize(bytes)?)),
+            1 => Ok(SupportedFheCiphertexts::FheUint4(safe_deserialize(bytes)?)),
+            2 => Ok(SupportedFheCiphertexts::FheUint8(safe_deserialize(bytes)?)),
+            3 => Ok(SupportedFheCiphertexts::FheUint16(safe_deserialize(bytes)?)),
+            4 => Ok(SupportedFheCiphertexts::FheUint32(safe_deserialize(bytes)?)),
+            5 => Ok(SupportedFheCiphertexts::FheUint64(safe_deserialize(bytes)?)),
+            6 => Ok(SupportedFheCiphertexts::FheUint128(safe_deserialize(
+                bytes,
+            )?)),
+            7 => Ok(SupportedFheCiphertexts::FheUint160(safe_deserialize(
+                bytes,
+            )?)),
+            8 => Ok(SupportedFheCiphertexts::FheUint256(safe_deserialize(
+                bytes,
+            )?)),
+            9 => Ok(SupportedFheCiphertexts::FheBytes64(safe_deserialize(
+                bytes,
+            )?)),
+            10 => Ok(SupportedFheCiphertexts::FheBytes128(safe_deserialize(bytes)?)),
+            11 => Ok(SupportedFheCiphertexts::FheBytes256(safe_deserialize(bytes)?)),
+            _ => Err(FhevmError::UnknownFheType(ct_type as i32).into()),
+        }
+    }
+
     pub fn to_ciphertext64(self) -> BaseRadixCiphertext<Ciphertext> {
         match self {
             SupportedFheCiphertexts::FheBool(v) => {

coprocessor/fhevm-engine/tfhe-compute-node/Cargo.toml

@@ -0,0 +1,81 @@
+[package]
+name = "tfhe-compute-node"
+version = "0.7.0"
+default-run = "tfhe_compute_node"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+# workspace dependencies
+alloy = { workspace = true }
+bigdecimal = { workspace = true }
+clap = { workspace = true }
+hex = { workspace = true }
+lru = { workspace = true }
+prometheus = { workspace = true }
+prost = { workspace = true }
+rand = { workspace = true }
+rayon = { workspace = true }
+serde_json = { workspace = true }
+sha3 = { workspace = true }
+strum = { workspace = true }
+sqlx = { workspace = true }
+tfhe = { workspace = true }
+tfhe-zk-pok = { workspace = true }
+time = { workspace = true }
+tokio = { workspace = true }
+tokio-util = { workspace = true }
+tonic = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+# opentelemetry support
+opentelemetry = { workspace = true }
+opentelemetry-otlp = { workspace = true }
+opentelemetry_sdk = { workspace = true }
+opentelemetry-semantic-conventions = { workspace = true }
+chrono = { workspace = true }
+serde = { workspace = true }
+
+# crates.io dependencies
+actix-web = "4.9.0"
+itertools = "0.13.0"
+lazy_static = "1.5.0"
+regex = "1.10.6"
+tonic-health = "0.12.3"
+tonic-types = "0.12.3"
+tonic-web = "0.12.3"
+uuid = { version = "1", features = ["v4"] }
+
+# local dependencies
+fhevm-engine-common = { path = "../fhevm-engine-common" }
+scheduler = { path = "../scheduler" }
+lapin = "3.7.2"
+futures-util.workspace = true
+redis = {version = "1.0.3", features = ["tokio-comp", "aio", "connection-manager"] }
+postcard = { version = "1.1.3", features = ["alloc"] }
+daggy.workspace = true
+thiserror.workspace = true
+
+
+[features]
+nightly-avx512 = ["tfhe/nightly-avx512"]
+gpu = ["tfhe/gpu", "scheduler/gpu", "fhevm-engine-common/gpu"]
+bench = []
+latency = ["fhevm-engine-common/latency"]
+throughput = ["fhevm-engine-common/throughput"]
+
+[dev-dependencies]
+criterion = { version = "0.5.1", features = ["async_futures"] }
+testcontainers = { workspace = true }
+test-harness = { path = "../test-harness" }
+serde = { workspace = true }
+serial_test = { workspace = true }
+
+[build-dependencies]
+tonic-build = { workspace = true }
+
+[[bin]]
+name = "tfhe_compute_node"
+path = "src/bin/tfhe_compute_node.rs"
+

coprocessor/fhevm-engine/tfhe-compute-node/src/bin/tfhe_compute_node.rs

@@ -0,0 +1,5 @@
+fn main() {
+    let args = tfhe_compute_node::cli::parse();
+    //TODO: add signal handler and pass close_recv
+    tfhe_compute_node::start_runtime(args, None);
+}

coprocessor/fhevm-engine/tfhe-compute-node/src/cli.rs

@@ -0,0 +1,60 @@
+use clap::Parser;
+use fhevm_engine_common::utils::DatabaseURL;
+use tracing::Level;
+use uuid::Uuid;
+
+#[derive(Parser, Debug, Clone)]
+#[command(version, about, long_about = None)]
+pub struct Args {
+    /// Postgres database url. If unspecified DATABASE_URL environment variable is used
+    #[arg(long)]
+    pub database_url: Option<DatabaseURL>,
+
+    /// Redis URL  
+    #[arg(long)]
+    pub redis_url: String,
+
+    /// RabbitMQ URI
+    #[arg(long, default_value = "amqp://guest:guest@localhost:5672/%2f")]
+    pub rmq_uri: String,
+
+    /// Tenant key cache size
+    #[arg(long, default_value_t = 32)]
+    pub tenant_key_cache_size: i32,
+
+    /// Ciphertext cache size
+    #[arg(long, default_value_t = 1000)]
+    pub ciphertext_cache_size: i32,
+
+    /// tfhe-worker service name in OTLP traces
+    #[arg(long, default_value = "tfhe-compute-node")]
+    pub service_name: String,
+
+    /// Prometheus metrics server address
+    #[arg(long, default_value = "0.0.0.0:9100")]
+    pub metrics_addr: Option<String>,
+
+    /// Tokio processing threads
+    #[arg(long, default_value_t = 32)]
+    pub blocking_fhe_threads: usize,
+
+    /// Tokio Async IO threads
+    #[arg(long, default_value_t = 4)]
+    pub tokio_threads: usize,
+
+    /// Log level for the application
+    #[arg(
+        long,
+        value_parser = clap::value_parser!(Level),
+        default_value_t = Level::INFO)]
+    pub log_level: Level,
+
+    /// Worker/replica ID for this worker instance
+    /// If not provided, a random UUID will be generated
+    #[arg(long, value_parser = clap::value_parser!(Uuid))]
+    pub worker_id: Option<Uuid>,
+}
+
+pub fn parse() -> Args {
+    Args::parse()
+}