feat(test-suite): add multiple coprocessors

test-suite/.gitignore

@@ -5,6 +5,8 @@ backup
 # Ignore all local .env files
 fhevm/**/.env.*.local
 fhevm/**/config/**/*.yaml.local
+!fhevm/env/staging/.env.coprocessor
+!fhevm/env/staging/.env.coprocessor-[0-2]
 
 # Ignore hexadecimal filenames imported from s3, generated by kms-core
 fhevm/[a-f0-9][a-f0-9][a-f0-9][a-f0-9][a-f0-9]*

test-suite/e2e/test/instance.ts

@@ -14,6 +14,8 @@ const relayerUrl = process.env.RELAYER_URL!;
 
 export const createInstances = async (accounts: Signers): Promise<FhevmInstances> => {
   // Create instance
+  console.log('relayer url given to create instance', relayerUrl);
+  console.log('network', network.name, network.config.url);
   const instances: FhevmInstances = {} as FhevmInstances;
   await Promise.all(
     Object.keys(accounts).map(async (k) => {
@@ -24,8 +26,6 @@ export const createInstances = async (accounts: Signers): Promise<FhevmInstances
 };
 
 export const createInstance = async () => {
-  console.log('relayer url given to create instance', relayerUrl);
-  console.log('network', network.name, network.config.url);
   const instance = await createFhevmInstance({
     verifyingContractAddressDecryption: verifyingContractAddressDecryption,
     verifyingContractAddressInputVerification: verifyingContractAddressInputVerification,

test-suite/e2e/test/userInput/inputFlow.ts

@@ -4,6 +4,10 @@ import { ethers } from 'hardhat';
 import { createInstances } from '../instance';
 import { getSigners, initSigners } from '../signers';
 
+function sleep(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
 describe('Input Flow', function () {
   before(async function () {
     await initSigners(2);
@@ -19,6 +23,25 @@ describe('Input Flow', function () {
   });
 
   it('test user input uint64 (non-trivial)', async function () {
+    if (!process.env.GATEWAY_NODE_URL) {
+      throw new Error('❌ Environment variable GATEWAY_NODE_URL must be set');
+    }
+
+    if (!process.env.CIPHERTEXT_COMMITS_CONTRACT_ADDRESS) {
+      throw new Error('❌ Environment variable CIPHERTEXT_COMMITS_CONTRACT_ADDRESS must be set');
+    }
+
+    if (!process.env.NUMBER_OF_COPROCESSORS) {
+      throw new Error('❌ Environment variable NUMBER_OF_COPROCESSORS must be set');
+    }
+    const GATEWAY_NODE_URL = process.env.GATEWAY_NODE_URL;
+    const CIPHERTEXT_COMMITS_CONTRACT_ADDRESS = process.env.CIPHERTEXT_COMMITS_CONTRACT_ADDRESS;
+    const NUMBER_OF_COPROCESSORS = parseInt(process.env.NUMBER_OF_COPROCESSORS);
+
+    const CIPHERTEXT_COMMITS_ABI = [
+      'function getAddCiphertextMaterialConsensusTxSenders(bytes32 ctHandle) view returns (address[])',
+    ];
+
     const inputAlice = this.instances.alice.createEncryptedInput(this.contractAddress, this.signers.alice.address);
     inputAlice.add64(18446744073709550042n);
     const encryptedAmount = await inputAlice.encrypt();
@@ -27,8 +50,48 @@ describe('Input Flow', function () {
       console.log(`  Handle ${index}: 0x${Buffer.from(handle).toString('hex')}`);
     });
     console.log('InputProof: 0x' + Buffer.from(encryptedAmount.inputProof).toString('hex'));
+
+    const ctHandleBytes32 = ethers.zeroPadValue(ethers.hexlify(encryptedAmount.handles[0]), 32);
+
     const tx = await this.contract.requestUint64NonTrivial(encryptedAmount.handles[0], encryptedAmount.inputProof);
     const receipt = await tx.wait();
     expect(receipt.status).to.equal(1);
+
+    // We are checking consensus on ciphertext commits, not on ZKPoK verification.
+    // Reason: the ZKPoK verification getter requires a ZKPoK ID, which is harder to obtain here.
+    // The ciphertext commits getter, on the other hand, only requires the handle,
+    // which we already have at this stage.
+    console.log('Fetching consensus senders from Gateway...');
+    console.log('ctHandleBytes32:', ctHandleBytes32);
+    console.log('GATEWAY_NODE_URL:', GATEWAY_NODE_URL);
+    console.log('CIPHERTEXT_COMMITS_CONTRACT_ADDRESS:', CIPHERTEXT_COMMITS_CONTRACT_ADDRESS);
+    console.log('CIPHERTEXT_COMMITS_ABI:', CIPHERTEXT_COMMITS_ABI);
+
+    const extProvider = new ethers.JsonRpcProvider(GATEWAY_NODE_URL); // Gateway node
+    const commitsExternal = new ethers.Contract(
+      CIPHERTEXT_COMMITS_CONTRACT_ADDRESS,
+      CIPHERTEXT_COMMITS_ABI,
+      extProvider,
+    );
+
+    let sleepTime = 15000;
+    console.log(`Sleeping ${sleepTime / 1000} seconds before checking consensus for coprocessor senders`);
+    console.log(`This is needed because the add ciphertext is handled asynchronously by coprocessors`);
+    await sleep(sleepTime);
+
+    const senders: string[] = await commitsExternal.getAddCiphertextMaterialConsensusTxSenders(ctHandleBytes32);
+
+    expect(senders, 'should have exactly 3 consensus senders on external chain').to.have.lengthOf(
+      NUMBER_OF_COPROCESSORS,
+    );
+
+    //  extra safety checks
+    const unique = new Set(senders.map((a) => a.toLowerCase()));
+    expect(unique.size, 'consensus senders must be unique').to.equal(NUMBER_OF_COPROCESSORS);
+    senders.forEach((a, i) =>
+      expect(a, `sender[${i}] must not be zero address`).to.not.equal('0x0000000000000000000000000000000000000000'),
+    );
+
+    console.log('Gateway chain consensus senders on ciphertext commitment:', senders);
   });
 });

test-suite/fhevm/docker-compose/coprocessor-0-docker-compose.yml

@@ -0,0 +1,247 @@
+services:
+  ######################## DATABASE SETUP #######################
+  db-0:
+    container_name: fhevm-0-coprocessor-db
+    image: postgres:15.7
+    restart: always
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    ports:
+      - '5432:5432'
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 10s
+      timeout: 5s
+      retries: 3
+    volumes:
+      - db-0:/var/lib/postgresql/data
+
+  key-downloader-0:
+    container_name: fhevm-0-key-downloader
+    image: busybox
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    volumes:
+      - keys-cache-0:/keys
+    command: >
+      sh -c "echo 'Starting key downloads...' &&
+        wget $KMS_PUBLIC_KEY -O /keys/pks &&
+        echo 'Downloaded public key' &&
+        wget $KMS_SERVER_KEY -O /keys/sns_pk &&
+        echo 'Downloaded server key' &&
+        wget $KMS_CRS_KEY -O /keys/pp &&
+        echo 'Downloaded PP key - All downloads complete'"
+
+  db-migration-0:
+    container_name: fhevm-0-db-migration
+    image: ghcr.io/zama-ai/fhevm/coprocessor/db-migration:${DB_MIGRATION_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/db-migration/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    environment:
+      - KEY_ID=${FHE_KEY_ID}
+    command:
+      - /initialize_db.sh
+    volumes:
+      - keys-cache-0:/fhevm-keys
+    depends_on:
+      db-0:
+        condition: service_healthy
+      key-downloader-0:
+        condition: service_completed_successfully
+
+  ####################### COPROCESSOR SERVICES #######################
+  host-listener-0:
+    container_name: fhevm-0-host-listener
+    image: ghcr.io/zama-ai/fhevm/coprocessor/host-listener:${HOST_LISTENER_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/host-listener/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    command:
+      - host_listener
+      - --database-url=${DATABASE_URL}
+      - --coprocessor-api-key=${TENANT_API_KEY}
+      - --acl-contract-address=${ACL_CONTRACT_ADDRESS}
+      - --tfhe-contract-address=${FHEVM_EXECUTOR_CONTRACT_ADDRESS}
+      - --url=${RPC_URL}
+      - --initial-block-time=1
+    depends_on:
+      db-0:
+        condition: service_healthy
+      db-migration-0:
+        condition: service_completed_successfully
+
+  gw-listener-0:
+    container_name: fhevm-0-gw-listener
+    image: ghcr.io/zama-ai/fhevm/coprocessor/gw-listener:${GW_LISTENER_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/gw-listener/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    healthcheck:
+      test: ["CMD-SHELL", "curl -f http://localhost:8080/liveness || exit 1"]
+      interval: 10s
+      timeout: 5s
+      retries: 3
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    command:
+      - gw_listener
+      - --database-url=${DATABASE_URL}
+      - --database-pool-size=16
+      - --verify-proof-req-database-channel="event_zkpok_new_work"
+      - --gw-url=${GATEWAY_WS_URL}
+      - --input-verification-address=${INPUT_VERIFICATION_ADDRESS}
+      - --error-sleep-initial-secs=1
+      - --error-sleep-max-secs=10
+    depends_on:
+      db-0:
+        condition: service_healthy
+      db-migration-0:
+        condition: service_completed_successfully
+
+  tfhe-worker-0:
+    container_name: fhevm-0-tfhe-worker
+    image: ghcr.io/zama-ai/fhevm/coprocessor/tfhe-worker:${TFHE_WORKER_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/tfhe-worker/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    command:
+      - tfhe_worker 
+      - --run-bg-worker
+      - --database-url=${DATABASE_URL}
+      - --pg-pool-max-connections=10
+      - --worker-polling-interval-ms=1000
+      - --work-items-batch-size=10
+      - --tenant-key-cache-size=32
+      - --coprocessor-fhe-threads=8
+      - --tokio-threads=4
+    depends_on:
+      db-0:
+        condition: service_healthy
+      db-migration-0:
+        condition: service_completed_successfully
+
+  zkproof-worker-0:
+    container_name: fhevm-0-zkproof-worker
+    image: ghcr.io/zama-ai/fhevm/coprocessor/zkproof-worker:${ZKPROOF_WORKER_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/zkproof-worker/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    command:
+      - zkproof_worker
+      - --database-url=${DATABASE_URL}
+      - --pg-listen-channel="event_zkpok_new_work"
+      - --pg-notify-channel="event_zkpok_computed"
+      - --pg-polling-interval=5
+      - --pg-pool-connections=5
+      - --worker-thread-count=4
+    depends_on:
+      db-0:
+        condition: service_healthy
+      db-migration-0:
+        condition: service_completed_successfully
+
+  sns-worker-0:
+    container_name: fhevm-0-sns-worker
+    image: ghcr.io/zama-ai/fhevm/coprocessor/sns-worker:${SNS_WORKER_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/sns-worker/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    command:
+      - sns_worker
+      - --database-url=${DATABASE_URL}
+      - --tenant-api-key=${TENANT_API_KEY}
+      - --pg-listen-channels
+      - event_pbs_computations
+      - event_ciphertext_computed
+      - --pg-notify-channel
+      - event_ciphertext128_computed
+      - --work-items-batch-size=20
+      - --pg-polling-interval=30
+      - --pg-pool-connections=10
+      - --bucket-name-ct64=ct64
+      - --bucket-name-ct128=ct128-0
+      - --s3-max-concurrent-uploads=100
+      - --s3-max-retries-per-upload=100
+      - --s3-max-backoff=10s
+      - --s3-max-retries-timeout=120s
+      - --s3-recheck-duration=2s
+      - --s3-regular-recheck-duration=120s
+      - --enable-compression
+    depends_on:
+      db-0:
+        condition: service_healthy
+      db-migration-0:
+        condition: service_completed_successfully
+
+  transaction-sender-0:
+    container_name: fhevm-0-transaction-sender
+    image: ghcr.io/zama-ai/fhevm/coprocessor/tx-sender:${TX_SENDER_VERSION}
+    build:
+      context: ../../..
+      dockerfile: coprocessor/fhevm-engine/transaction-sender/Dockerfile
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max
+    env_file:
+      - ../env/staging/.env.coprocessor-0.local
+    command:
+      - transaction_sender
+      - --database-url=${DATABASE_URL}
+      - --gateway-url=${GATEWAY_WS_URL}
+      - --private-key=${TX_SENDER_PRIVATE_KEY}
+      - --ciphertext-commits-address=${CIPHERTEXT_COMMITS_ADDRESS}
+      - --input-verification-address=${INPUT_VERIFICATION_ADDRESS}
+      - --multichain-acl-address=${MULTICHAIN_ACL_ADDRESS}
+      - --database-pool-size=10
+      - --database-polling-interval-secs=5
+      - --verify-proof-resp-database-channel="event_zkpok_computed"
+      - --add-ciphertexts-database-channel="event_ciphertexts_uploaded"
+      - --verify-proof-resp-batch-limit=128
+      - --verify-proof-resp-max-retries=15
+      - --verify-proof-remove-after-max-retries
+      - --signer-type=private-key
+    depends_on:
+      db-0:
+        condition: service_healthy
+      db-migration-0:
+        condition: service_completed_successfully
+
+volumes:
+  db-0:
+  keys-cache-0: