Make the polling interval configurable.

docs/getting_started/fhevm/coprocessor/contracts.md

@@ -1,4 +1,4 @@
-# Deploy initial contracts
+# Deploy initial contractsL
 
 Following is an example of how to deploy initial contracts on the Ethereum Sepolia testnet. Deploying on Ethereum mainnet should be almost identical and should be possible by just changing the SEPOLIA_RPC_URL to ETHEREUM_MAINNET_RPC_URL and poiting to a correct RPC node.
 
@@ -36,7 +36,7 @@ For the `SEPOLIA_RPC_URL` env variable, you can either get one from a service pr
 - `node_modules/fhevm-core-contracts/addresses/.env.fhepayment` for FHEPayment address
 - `gateway/.env.gateway` for GatewayContract address.
 
-This script is found exactly inside the [`./precompute-addresses.sh` file](https://github.com/zama-ai/fhevm/blob/main/precompute-addresses.sh):
+This script is found exactly inside the [`./precompute-addresses.sh` file](https://github.com/zama-ai/fhevm/blob/v0.6.0-0/precompute-addresses.sh):
 
 ```
 #!/bin/bash

fhevm-engine/coprocessor/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "coprocessor"
-version = "0.1.1"
+version = "0.1.2"
 default-run = "coprocessor"
 authors.workspace = true
 edition.workspace = true

fhevm-engine/coprocessor/README.md

@@ -50,6 +50,7 @@ Reload database and apply schemas from scratch
 make recreate_db
 ```
 Run the server and background fhe worker
+
+```
+cargo run -- --run-server --run-bg-worker --worker-polling-interval-ms 1000
 ```
-cargo run -- --run-server --run-bg-worker
-```

fhevm-engine/coprocessor/src/daemon_cli.rs

@@ -11,6 +11,10 @@ pub struct Args {
     #[arg(long)]
     pub run_bg_worker: bool,
 
+    /// Polling interval for the background worker to fetch jobs
+    #[arg(long, default_value_t = 1000)]
+    pub worker_polling_interval_ms: u64,
+
     /// Generate fhe keys and exit
     #[arg(long)]
     pub generate_fhe_keys: bool,

fhevm-engine/coprocessor/src/tfhe_worker.rs

@@ -1,5 +1,5 @@
 use crate::{db_queries::populate_cache_with_tenant_keys, types::TfheTenantKeys};
-use fhevm_engine_common::types::{FhevmError, Handle, SupportedFheCiphertexts};
+use fhevm_engine_common::types::{Handle, SupportedFheCiphertexts};
 use fhevm_engine_common::{tfhe_ops::current_ciphertext_version, types::SupportedFheOperations};
 use itertools::Itertools;
 use lazy_static::lazy_static;
@@ -85,7 +85,7 @@ async fn tfhe_worker_cycle(
                     WORK_ITEMS_NOTIFICATIONS_COUNTER.inc();
                     info!(target: "tfhe_worker", "Received work_available notification from postgres");
                 },
-                _ = tokio::time::sleep(tokio::time::Duration::from_millis(5000)) => {
+                _ = tokio::time::sleep(tokio::time::Duration::from_millis(args.worker_polling_interval_ms)) => {
                     WORK_ITEMS_POLL_COUNTER.inc();
                     info!(target: "tfhe_worker", "Polling the database for more work on timer");
                 },

fhevm-engine/fhevm-engine-common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fhevm-engine-common"
-version = "0.1.1"
+version = "0.6.0-6"
 authors.workspace = true
 edition.workspace = true
 license.workspace = true

fhevm-engine/fhevm-go-coproc/fhevm/api.go

@@ -18,6 +18,8 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+const DEFAULT_FHEVM_COPROCESSOR_SYNC_INTERVAL = "1s"
+
 type FheUintType uint8
 
 const (
@@ -144,6 +146,7 @@ type ApiImpl struct {
 	store              *SqliteComputationStore
 	address            common.Address
 	aclContractAddress common.Address
+	flushInterval      time.Duration
 }
 
 type SessionImpl struct {
@@ -429,10 +432,21 @@ func InitCoprocessor() (CoprocessorApi, error) {
 		}
 		aclContractAddress := common.HexToAddress(aclContractAddressHex)
 
+		coprocessorSyncIntervalString, isConfigured := os.LookupEnv("FHEVM_COPROCESSOR_SYNC_INTERVAL")
+		if !isConfigured {
+			coprocessorSyncIntervalString = DEFAULT_FHEVM_COPROCESSOR_SYNC_INTERVAL
+		}
+		coprocessorSyncInterval, err := time.ParseDuration(coprocessorSyncIntervalString)
+		if err != nil {
+			return nil, fmt.Errorf("parsing FHEVM_COPROCESSOR_SYNC_INTERVAL (%s): %w", coprocessorSyncIntervalString, err)
+		}
+		fmt.Printf("Using coprocessor sync interval %s\n", coprocessorSyncInterval.String())
+
 		apiImpl := ApiImpl{
 			store:              ciphertextDb,
 			address:            fhevmContractAddress,
 			aclContractAddress: aclContractAddress,
+			flushInterval:      coprocessorSyncInterval,
 		}
 
 		// background job to submit computations to coprocessor
@@ -448,7 +462,7 @@ func scheduleCoprocessorFlushes(impl *ApiImpl) {
 	// timer to send polling for messages every 10 seconds
 	go func() {
 		for {
-			time.Sleep(time.Millisecond * 10000)
+			time.Sleep(impl.flushInterval)
 			select {
 			case impl.store.jobChannel <- true:
 			default: