Merge of #2170

Author: mergify[bot]

coprocessor/fhevm-engine/.sqlx/query-1e3fb703be1a08661ebe7dc0b7517a96f9fa8b5c9c9388fb5fdb06fbcad1a944.json

@@ -458,13 +458,24 @@ fn try_execute_node(
         RERAND_LATENCY_BATCH_HISTOGRAM.observe(elapsed.as_secs_f64());
     }
     let opcode = node.opcode;
-    Ok(run_computation(
-        opcode,
-        cts,
-        node_index,
-        gpu_idx,
-        transaction_id,
-    ))
+    let result = std::panic::catch_unwind(|| {
+        run_computation(opcode, cts, node_index, gpu_idx, transaction_id)
+    });
+    match result {
+        Err(e) => {
+            let msg = e
+                .downcast_ref::<&str>()
+                .map(|s| s.to_string())
+                .or_else(|| e.downcast_ref::<String>().cloned())
+                .unwrap_or_else(|| "unknown panic payload".to_string());
+            eprintln!("Panic while executing operation: {msg}");
+            error!(target: "scheduler", { handle = ?hex::encode(&node.result_handle), msg },
+               "Panic while executing operation");
+            telemetry::set_current_span_error(&msg);
+            Err(SchedulerError::ExecutionPanic(msg).into())
+        }
+        Ok(r) => Ok(r),
+    }
 }
 
 type OpResult = Result<CompressedCiphertext>;

coprocessor/fhevm-engine/scheduler/src/dfg/scheduler.rs

@@ -62,14 +62,15 @@ impl std::fmt::Debug for DFGTaskInput {
     }
 }
 
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug, Clone)]
 pub enum SchedulerError {
     CyclicDependence,
     DataflowGraphError,
     MissingInputs,
     DecompressionError,
     ReRandomisationError,
     SchedulerError,
+    ExecutionPanic(String),
 }
 
 impl std::error::Error for SchedulerError {}
@@ -95,6 +96,9 @@ impl std::fmt::Display for SchedulerError {
             Self::SchedulerError => {
                 write!(f, "Generic scheduler error")
             }
+            Self::ExecutionPanic(s) => {
+                write!(f, "Panic during execution of operation: {}", s)
+            }
         }
     }
 }

coprocessor/fhevm-engine/scheduler/src/dfg/types.rs

@@ -10,10 +10,12 @@ use crate::tests::event_helpers::{
     zero_address, EventHarness,
 };
 use crate::tests::test_cases::{
-    generate_binary_test_cases, generate_unary_test_cases, BinaryOperatorTestCase,
-    UnaryOperatorTestCase,
+    generate_binary_test_cases, generate_panic_binary_test_cases, generate_unary_test_cases,
+    BinaryOperatorTestCase, UnaryOperatorTestCase,
+};
+use crate::tests::utils::{
+    decrypt_ciphertexts, errors_on_allowed_handles, wait_until_all_allowed_handles_computed,
 };
-use crate::tests::utils::{decrypt_ciphertexts, wait_until_all_allowed_handles_computed};
 
 const LOCAL_SUPPORTED_TYPES: &[i32] = &[
     0, // bool
@@ -319,6 +321,93 @@ async fn test_fhe_binary_operands_events() -> Result<(), Box<dyn std::error::Err
     Ok(())
 }
 
+#[tokio::test]
+#[serial(db)]
+async fn test_fhe_binary_operands_events_panic() -> Result<(), Box<dyn std::error::Error>> {
+    let EventHarness {
+        app,
+        pool: _,
+        listener_db,
+    } = setup_event_harness().await?;
+
+    let mut cases = vec![];
+    for op in generate_panic_binary_test_cases() {
+        if !supported_types().contains(&op.input_types) {
+            continue;
+        }
+        // TrivialEncrypt test setup uses ClearConst (up to 256-bit payloads).
+        if op.bits > 256 {
+            continue;
+        }
+        let lhs_handle = next_handle();
+        let rhs_handle = next_handle();
+        let output_handle = next_handle();
+        let transaction_id = next_handle();
+
+        let lhs_bytes = as_scalar_uint(&op.lhs);
+        let rhs_bytes = as_scalar_uint(&op.rhs);
+
+        println!(
+            "Operations for binary test bits:{} op:{} is_scalar:{} lhs:{} rhs:{}",
+            op.bits, op.operator, op.is_scalar, op.lhs, op.rhs
+        );
+        let caller = zero_address();
+
+        let mut tx = listener_db.new_transaction().await?;
+        insert_event(
+            &listener_db,
+            &mut tx,
+            transaction_id,
+            TfheContractEvents::TrivialEncrypt(TfheContract::TrivialEncrypt {
+                caller,
+                pt: lhs_bytes,
+                toType: to_ty(op.input_types),
+                result: lhs_handle,
+            }),
+            true,
+        )
+        .await?;
+        allow_handle(&listener_db, &mut tx, &lhs_handle).await?;
+        if !op.is_scalar {
+            insert_event(
+                &listener_db,
+                &mut tx,
+                transaction_id,
+                TfheContractEvents::TrivialEncrypt(TfheContract::TrivialEncrypt {
+                    caller,
+                    pt: rhs_bytes,
+                    toType: to_ty(op.input_types),
+                    result: rhs_handle,
+                }),
+                true,
+            )
+            .await?;
+            allow_handle(&listener_db, &mut tx, &rhs_handle).await?;
+        }
+        let op_event = binary_op_to_event(&op, &lhs_handle, &rhs_handle, &op.rhs, &output_handle);
+        insert_event(&listener_db, &mut tx, transaction_id, op_event, true).await?;
+        allow_handle(&listener_db, &mut tx, &output_handle).await?;
+        tx.commit().await?;
+
+        cases.push((op, output_handle));
+    }
+
+    wait_until_all_allowed_handles_computed(&app).await?;
+    let errors = errors_on_allowed_handles(&app).await?;
+    for msg in &errors {
+        assert!(
+            msg.contains("ExecutionPanic"),
+            "Expected error message to contain 'ExecutionPanic', but got: {msg}"
+        );
+    }
+    assert_eq!(
+        errors.len(),
+        cases.len(),
+        "Expected all computations to have errors, but some succeeded"
+    );
+    Ok(())
+}
+
 fn unary_op_to_event(
     op: &UnaryOperatorTestCase,
     input: &Handle,

coprocessor/fhevm-engine/tfhe-worker/src/tests/operators_from_events.rs

@@ -6,6 +6,7 @@ use fhevm_engine_common::types::{is_ebytes_type, FheOperationType, SupportedFheO
 use std::ops::Not;
 use strum::IntoEnumIterator;
 
+#[derive(Clone, Debug)]
 pub struct BinaryOperatorTestCase {
     pub bits: i32,
     pub operator: i32,
@@ -47,6 +48,32 @@ fn supported_bits_to_bit_type_in_db(inp: i32) -> i32 {
     }
 }
 
+pub fn generate_panic_binary_test_cases() -> Vec<BinaryOperatorTestCase> {
+    let mut new_cases = vec![];
+    for mut op in generate_binary_test_cases() {
+        if op.bits >= 256 {
+            // avoid non supported scalar in test
+            continue;
+        }
+        if !op.is_scalar {
+            continue;
+        }
+        let operator = SupportedFheOperations::try_from(op.operator).unwrap();
+        if op.operator != SupportedFheOperations::FheDiv as i32
+            && op.operator != SupportedFheOperations::FheRem as i32
+        {
+            continue;
+        }
+        let modulo = BigInt::from(1) << op.bits;
+        op.lhs = BigInt::from(1) << (op.bits - 1); // max positive value for signed integer of op.bits bits
+        op.rhs = modulo;
+        op.expected_output = compute_expected_binary_output(&op.lhs, &op.rhs, operator)
+            % (BigInt::from(1) << op.bits); // expected result
+        new_cases.push(op);
+    }
+    new_cases
+}
+
 pub fn generate_binary_test_cases() -> Vec<BinaryOperatorTestCase> {
     let mut cases = Vec::new();
     let bit_shift_ops = [

coprocessor/fhevm-engine/tfhe-worker/src/tests/test_cases.rs

@@ -150,6 +150,30 @@ async fn setup_test_app_custom_docker() -> Result<TestInstance, Box<dyn std::err
     })
 }
 
+pub async fn errors_on_allowed_handles(
+    test_instance: &TestInstance,
+) -> Result<Vec<String>, Box<dyn std::error::Error>> {
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(2)
+        .connect(test_instance.db_url())
+        .await?;
+    let records = sqlx::query!(
+        "SELECT error_message FROM computations WHERE is_allowed = TRUE AND is_error = TRUE",
+    )
+    .fetch_all(&pool)
+    .await?;
+    let mut errors: Vec<String> = vec![];
+    for error in &records {
+        errors.push(
+            error
+                .error_message
+                .clone()
+                .unwrap_or_else(|| "No error message".to_string()),
+        );
+    }
+    Ok(errors)
+}
+
 pub async fn wait_until_all_allowed_handles_computed(
     test_instance: &TestInstance,
 ) -> Result<(), Box<dyn std::error::Error>> {

coprocessor/fhevm-engine/tfhe-worker/src/tests/utils.rs

@@ -557,8 +557,9 @@ async fn upload_transaction_graph_results<'a>(
                         CoprocessorError::FhevmError(swap_val).into()
                     } else {
                         CoprocessorError::SchedulerError(
-                            *err.downcast_ref::<SchedulerError>()
-                                .unwrap_or(&SchedulerError::SchedulerError),
+                            err.downcast_ref::<SchedulerError>()
+                                .cloned()
+                                .unwrap_or(SchedulerError::SchedulerError),
                         )
                         .into()
                     };