feat(coprocessor): implement a retry mechanism for pbs computations

coprocessor/fhevm-engine/db-migration/migrations/20251031144714_pbs_computations_retries.sql

@@ -0,0 +1,23 @@
+-- Add migration script here
+BEGIN;
+-- Lock the table to ensure the backfill of schedule_order is safe
+LOCK TABLE pbs_computations IN ACCESS EXCLUSIVE MODE;
+
+ALTER TABLE pbs_computations
+    ADD COLUMN IF NOT EXISTS error TEXT,
+    ADD COLUMN IF NOT EXISTS schedule_order TIMESTAMPTZ DEFAULT NOW();
+
+-- Backfill existing rows
+UPDATE pbs_computations
+SET schedule_order = created_at
+WHERE schedule_order IS NULL;
+
+-- enforce not-null after backfill
+ALTER TABLE pbs_computations
+    ALTER COLUMN schedule_order SET NOT NULL;
+COMMIT;
+
+-- Partial index for unfinished rows ordered/filtered by created_at
+CREATE INDEX IF NOT EXISTS idx_pbs_comp_unfinished_created_at
+ON pbs_computations (schedule_order, created_at)
+WHERE is_completed = FALSE;

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

@@ -110,3 +110,23 @@ impl Default for HeartBeat {
         Self::new()
     }
 }
+
+#[macro_export]
+macro_rules! with_panic_guard {
+    ($body:expr) => {{
+        use std::panic::{catch_unwind, AssertUnwindSafe};
+        match catch_unwind(AssertUnwindSafe(|| $body)) {
+            Ok(v) => Ok(v),
+            Err(payload) => {
+                let msg = if let Some(s) = (&*payload).downcast_ref::<&'static str>() {
+                    s.to_string()
+                } else if let Some(s) = (&*payload).downcast_ref::<String>() {
+                    s.clone()
+                } else {
+                    "panic payload: non-string".to_string()
+                };
+                Err(msg)
+            }
+        }
+    }};
+}

coprocessor/fhevm-engine/sns-worker/src/aws_upload.rs

@@ -1,5 +1,6 @@
 use crate::{
-    BigCiphertext, Ciphertext128Format, Config, ExecutionError, HandleItem, S3Config, UploadJob,
+    BigCiphertext, Ciphertext128Format, Config, ExecutionError, HandleItem, S3Config, TaskStatus,
+    UploadJob,
 };
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::operation::head_bucket::HeadBucketError;
@@ -486,6 +487,7 @@ async fn fetch_pending_uploads(
                 ct128: Arc::new(ct128),
                 otel: telemetry::tracer_with_handle("recovery_task", handle, &transaction_id),
                 transaction_id,
+                status: TaskStatus::default(),
             };
 
             // Instruct the uploader to acquire DB lock when processing the item