fix(coprocessor): port DB and TFHE worker fixes from release 0.10.x (#1396)

* fix(coprocessor): fix missing downgrade of scheduler error in tfhe-worker (#1352)

* fix(coprocessor): update tfhe-worker query for work (#1357)

Author: antoniupop

coprocessor/fhevm-engine/.sqlx/query-86869a71d7bd80262a82765e77b819bab29e95c1d60ff31525172fb859c61cee.json renamed to coprocessor/fhevm-engine/.sqlx/query-63f097acd50a5ec09f959668fce297d0c2c1c9851741c2a5bbb3238af2fe5b75.json

@@ -0,0 +1,3 @@
+CREATE INDEX IF NOT EXISTS idx_computations_created_at
+  ON computations USING BTREE (created_at)
+  WHERE is_completed = false;

coprocessor/fhevm-engine/db-migration/migrations/20251126110000_computations_created_at_index.sql

@@ -346,8 +346,8 @@ fn execute_partition(
         for (h, i) in tx_inputs.iter_mut() {
             if i.is_none() {
                 let Some(Ok(ct)) = res.get(h) else {
-                    warn!(target: "scheduler", {transaction_id = ?tid },
-                         "Missing input to compute transaction - skipping");
+                    warn!(target: "scheduler", {transaction_id = ?hex::encode(tid) },
+		       "Missing input to compute transaction - skipping");
                     for nidx in dfg.graph.node_identifiers() {
                         let Some(node) = dfg.graph.node_weight_mut(nidx) else {
                             error!(target: "scheduler", {index = ?nidx.index() }, "Wrong dataflow graph index");
@@ -375,7 +375,7 @@ fn execute_partition(
             if let Err(e) =
                 re_randomise_transaction_inputs(tx_inputs, &tid, cid, gpu_idx, cpk.clone())
             {
-                error!(target: "scheduler", {transaction_id = ?tid, error = ?e },
+                error!(target: "scheduler", {transaction_id = ?hex::encode(tid), error = ?e },
 		       "Error while re-randomising inputs");
                 for nidx in dfg.graph.node_identifiers() {
                     let Some(node) = dfg.graph.node_weight_mut(nidx) else {
@@ -401,7 +401,7 @@ fn execute_partition(
             // If re-randomisation is not available (e.g., on GPU),
             // only decompress ciphertexts
             if let Err(e) = decompress_transaction_inputs(tx_inputs, &tid, gpu_idx, cpk.clone()) {
-                error!(target: "scheduler", {transaction_id = ?tid, error = ?e },
+                error!(target: "scheduler", {transaction_id = ?hex::encode(tid), error = ?e },
 		       "Error while decompressing inputs");
                 for nidx in dfg.graph.node_identifiers() {
                     let Some(node) = dfg.graph.node_weight_mut(nidx) else {
@@ -499,7 +499,7 @@ fn try_execute_node(
             cts.push(i);
         } else {
             // That should not be possible as we called the checker.
-            error!(target: "scheduler", { handle = ?node.result_handle }, "Computation missing inputs");
+            error!(target: "scheduler", { handle = ?hex::encode(&node.result_handle) }, "Computation missing inputs");
             return Err(SchedulerError::MissingInputs.into());
         }
     }

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

@@ -303,7 +303,19 @@ async fn query_for_work<'a>(
 WITH selected_computations AS (
   (
     SELECT DISTINCT
-      c.transaction_id
+      c_creation_order.transaction_id
+    FROM (
+      SELECT transaction_id
+      FROM computations 
+      WHERE is_completed = FALSE
+        AND is_error = FALSE
+        AND is_allowed = TRUE
+      ORDER BY created_at
+      LIMIT $1
+    ) as c_creation_order
+   UNION ALL
+    SELECT DISTINCT
+      c_schedule_order.transaction_id
     FROM (
       SELECT transaction_id
       FROM computations 
@@ -312,7 +324,7 @@ WITH selected_computations AS (
         AND is_allowed = TRUE
       ORDER BY schedule_order
       LIMIT $1
-    ) as c
+    ) as c_schedule_order
   )
 )
 -- Acquire all computations from this transaction set
@@ -548,6 +560,11 @@ async fn upload_transaction_graph_results<'a>(
                         CoprocessorError::SchedulerError(SchedulerError::MissingInputs)
                     ) {
                         uncomputable.push((result.handle.clone(), result.transaction_id.clone()));
+                        // Make sure we don't mark this as an error since this simply means that the
+                        // inputs weren't available when we tried scheduling these operations.
+                        // Setting them as uncomputable will postpone them with an exponential backoff
+                        // and they will be retried later.
+                        continue;
                     }
                 }
                 set_computation_error(