Feat/setup progress log (#172)

sapinb · Zk2u · web-flow · commit dd1ce4fee893 · 2026-04-18T19:57:20.000Z
* add progess in rpc status

* add logs on handle_receive_garbling_table failure cases

* fix: create new txn after receiving tables in handle_receive_garbling_table

* feat: add RetryEvalState wrapper to auto-retry evaluator reads on FDB txn timeout

* refactor: rename _with_retry helpers, use RetryEvalState in begin_table_commitment

* refactor: drop RetryEvalState

---------

Co-authored-by: azz &lt;azz@alpenlabs.io&gt;
diff --git a/crates/job/executors/src/evaluator.rs b/crates/job/executors/src/evaluator.rs
@@ -249,11 +249,15 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
 
     let eval_state = match ctx.storage.evaluator_state(peer_id).await {
         Ok(state) => state,
-        Err(_) => return HandlerOutcome::Retry,
+        Err(e) => {
+            warn!(%peer_id, ?e, "failed to load evaluator state for receive_garbling_table");
+            return HandlerOutcome::Retry;
+        }
     };
 
     // Resolve the commitment → circuit index from the evaluator root state.
     let Some(root_state) = eval_state.get_root_state().await.ok().flatten() else {
+        warn!(%peer_id, "root state not available for receive_garbling_table");
         return HandlerOutcome::Retry;
     };
     let (eval_indices, eval_commitments) = match &root_state.step {
@@ -262,14 +266,18 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
             eval_commitments,
             ..
         } => (*eval_indices, eval_commitments.clone()),
-        _ => return HandlerOutcome::Retry,
+        _ => {
+            warn!(%peer_id, step = ?root_state.step, "unexpected step in receive_garbling_table");
+            return HandlerOutcome::Retry;
+        }
     };
 
     let Some(pos) = eval_commitments
         .iter()
         .position(|c| *c == expected_commitment)
     else {
         // Commitment not found — stale action or state mismatch.
+        warn!(%peer_id, "commitment not found in eval_commitments for receive_garbling_table");
         return HandlerOutcome::Retry;
     };
     let index = eval_indices[pos];
@@ -286,6 +294,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
         .await;
 
     let Ok(expectation) = expectation else {
+        warn!(%peer_id, "bulk receiver registration failed for receive_garbling_table");
         return HandlerOutcome::Retry;
     };
 
@@ -298,6 +307,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
 
     // Wait for the garbler to open the stream.
     let Ok(mut stream) = expectation.recv().await else {
+        warn!(%peer_id, "bulk stream receive failed for receive_garbling_table");
         return HandlerOutcome::Retry;
     };
 
@@ -312,6 +322,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
     };
     let writer = ctx.table_store.create(&table_id).await;
     let Ok(mut writer) = writer else {
+        warn!(%peer_id, "table writer creation failed for receive_garbling_table");
         return HandlerOutcome::Retry;
     };
 
@@ -344,6 +355,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
             if !ct_part.is_empty() {
                 ct_hasher.update(ct_part);
                 if writer.write_ciphertext(ct_part).await.is_err() {
+                    error!(%peer_id, "ciphertext write failed (translation overflow) for receive_garbling_table");
                     let _ = ctx.table_store.delete(&table_id).await;
                     return HandlerOutcome::Retry;
                 }
@@ -352,6 +364,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
             // All translation received — remaining data is ciphertext.
             ct_hasher.update(&chunk);
             if writer.write_ciphertext(&chunk).await.is_err() {
+                error!(%peer_id, "ciphertext write failed for receive_garbling_table");
                 let _ = ctx.table_store.delete(&table_id).await;
                 return HandlerOutcome::Retry;
             }
@@ -360,24 +373,29 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
 
     // Verify we received enough translation data.
     if translation_remaining > 0 {
+        error!(%peer_id, translation_remaining, "incomplete translation data for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     }
 
     let translate_hash = translate_hasher.finalize();
     let ct_hash = ct_hasher.finalize();
+
     // Load metadata from evaluator state. These were stored when the garbler's
     // CommitMsgHeader (aes keys, public S) and ChallengeResponseMsgHeader
     // (output label ciphertexts) were processed by the STF.
     let Some(aes_key) = eval_state.get_aes128_key(index).await.ok().flatten() else {
+        warn!(%peer_id, ?index, "aes128_key not available for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     };
     let Some(public_s) = eval_state.get_public_s(index).await.ok().flatten() else {
+        warn!(%peer_id, ?index, "public_s not available for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     };
     let Some(output_label_ct) = eval_state.get_output_label_ct(index).await.ok().flatten() else {
+        warn!(%peer_id, ?index, "output_label_ct not available for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     };
@@ -387,6 +405,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
         .ok()
         .flatten()
     else {
+        warn!(%peer_id, ?index, "constant_zero_label not available for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     };
@@ -396,13 +415,15 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
         .ok()
         .flatten()
     else {
+        warn!(%peer_id, ?index, "constant_one_label not available for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     };
     // Verify the received data matches the expected commitment.
     let params_hash = hash_garbling_params(&aes_key, &public_s, &constant_one, &constant_zero);
     let computed = compute_commitment(&ct_hash, &translate_hash, &output_label_ct, &params_hash);
     if computed != expected_commitment {
+        error!(%peer_id, ?index, "commitment mismatch in receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     }
@@ -413,6 +434,7 @@ pub(crate) async fn handle_receive_garbling_table<SP: StorageProvider, TS: Table
     };
 
     if writer.finish(&translation_buf, metadata).await.is_err() {
+        error!(%peer_id, ?index, "table writer finish failed for receive_garbling_table");
         let _ = ctx.table_store.delete(&table_id).await;
         return HandlerOutcome::Retry;
     }
diff --git a/crates/rpc/service/src/types.rs b/crates/rpc/service/src/types.rs
@@ -5,7 +5,7 @@
 
 use bitcoin::{XOnlyPublicKey, secp256k1::schnorr::Signature as SchnorrSignature};
 use mosaic_cac_types::{
-    DepositId, DepositInputs, SetupInputs, Sighashes, WithdrawalInputs,
+    DepositId, DepositInputs, HeapArray, SetupInputs, Sighashes, WithdrawalInputs,
     state_machine::{
         Role,
         evaluator::{self},
@@ -129,20 +129,62 @@ pub struct EvaluatorWithdrawalData {
 
 // --- From impls: garbler/evaluator state -> service domain types ---
 
+/// Format progress of a `HeapArray<bool, N>` as `"done/total"`.
+fn arr_progress<const N: usize>(arr: &HeapArray<bool, N>) -> String {
+    format!("{}/{}", arr.count_ones(), arr.len())
+}
+
+/// Format progress of a single `bool` flag as `"0/1"` or `"1/1"`.
+fn bool_progress(b: bool) -> &'static str {
+    if b { "1/1" } else { "0/1" }
+}
+
 impl From<&garbler::Step> for TablesetStatus {
     fn from(step: &garbler::Step) -> Self {
         use garbler::Step::*;
         match step {
-            Uninit
-            | GeneratingPolynomialCommitments { .. }
-            | GeneratingShares { .. }
-            | GeneratingTableCommitments { .. }
-            | SendingCommit { .. }
-            | WaitingForChallenge
-            | SendingChallengeResponse { .. }
-            | TransferringGarblingTables { .. } => TablesetStatus::Incomplete {
+            Uninit | WaitingForChallenge => TablesetStatus::Incomplete {
                 details: step.step_name().into(),
             },
+            GeneratingPolynomialCommitments { inputs, output } => TablesetStatus::Incomplete {
+                details: format!(
+                    "{} (inputs: {}, output: {})",
+                    step.step_name(),
+                    arr_progress(inputs),
+                    bool_progress(*output),
+                ),
+            },
+            GeneratingShares { generated } => TablesetStatus::Incomplete {
+                details: format!("{} ({})", step.step_name(), arr_progress(generated)),
+            },
+            GeneratingTableCommitments { generated, .. } => TablesetStatus::Incomplete {
+                details: format!("{} ({})", step.step_name(), arr_progress(generated)),
+            },
+            SendingCommit {
+                header_acked,
+                chunk_acked,
+            } => TablesetStatus::Incomplete {
+                details: format!(
+                    "{} (header: {}, chunks: {})",
+                    step.step_name(),
+                    bool_progress(*header_acked),
+                    arr_progress(chunk_acked),
+                ),
+            },
+            SendingChallengeResponse {
+                header_acked,
+                chunk_acked,
+            } => TablesetStatus::Incomplete {
+                details: format!(
+                    "{} (header: {}, chunks: {})",
+                    step.step_name(),
+                    bool_progress(*header_acked),
+                    arr_progress(chunk_acked),
+                ),
+            },
+            TransferringGarblingTables { transferred, .. } => TablesetStatus::Incomplete {
+                details: format!("{} ({})", step.step_name(), arr_progress(transferred)),
+            },
             SetupComplete => TablesetStatus::SetupComplete,
             CompletingAdaptors { deposit_id } => TablesetStatus::Contest {
                 deposit_id: *deposit_id,
@@ -162,14 +204,34 @@ impl From<&evaluator::Step> for TablesetStatus {
     fn from(step: &evaluator::Step) -> Self {
         use evaluator::Step::*;
         match step {
-            Uninit
-            | WaitingForCommit { .. }
-            | WaitingForChallengeResponse { .. }
-            | VerifyingOpenedInputShares
-            | VerifyingTableCommitments { .. }
-            | ReceivingGarblingTables { .. } => TablesetStatus::Incomplete {
+            Uninit | VerifyingOpenedInputShares => TablesetStatus::Incomplete {
                 details: step.step_name().into(),
             },
+            WaitingForCommit { header, chunks } => TablesetStatus::Incomplete {
+                details: format!(
+                    "{} (header: {}, chunks: {})",
+                    step.step_name(),
+                    bool_progress(*header),
+                    arr_progress(chunks),
+                ),
+            },
+            WaitingForChallengeResponse {
+                header,
+                remaining_chunks,
+            } => TablesetStatus::Incomplete {
+                details: format!(
+                    "{} (header: {}, chunks: {})",
+                    step.step_name(),
+                    bool_progress(*header),
+                    arr_progress(remaining_chunks),
+                ),
+            },
+            VerifyingTableCommitments { verified, .. } => TablesetStatus::Incomplete {
+                details: format!("{} ({})", step.step_name(), arr_progress(verified)),
+            },
+            ReceivingGarblingTables { received, .. } => TablesetStatus::Incomplete {
+                details: format!("{} ({})", step.step_name(), arr_progress(received)),
+            },
             SetupComplete => TablesetStatus::SetupComplete,
             EvaluatingTables { deposit_id, .. } => TablesetStatus::Contest {
                 deposit_id: *deposit_id,
@@ -191,8 +253,10 @@ impl From<&evaluator::Step> for TablesetStatus {
 impl From<garbler::DepositState> for DepositStatus {
     fn from(deposit: garbler::DepositState) -> Self {
         match deposit.step {
-            garbler::DepositStep::WaitingForAdaptors { .. }
-            | garbler::DepositStep::VerifyingAdaptors => DepositStatus::Incomplete {
+            garbler::DepositStep::WaitingForAdaptors { ref chunks } => DepositStatus::Incomplete {
+                details: format!("{} ({})", deposit.step.step_name(), arr_progress(chunks),),
+            },
+            garbler::DepositStep::VerifyingAdaptors => DepositStatus::Incomplete {
                 details: deposit.step.step_name().into(),
             },
             garbler::DepositStep::DepositReady => DepositStatus::Ready,
@@ -205,9 +269,19 @@ impl From<garbler::DepositState> for DepositStatus {
 impl From<evaluator::DepositState> for DepositStatus {
     fn from(deposit: evaluator::DepositState) -> Self {
         match deposit.step {
-            evaluator::DepositStep::GeneratingAdaptors { .. }
-            | evaluator::DepositStep::SendingAdaptors { .. } => DepositStatus::Incomplete {
-                details: deposit.step.step_name().into(),
+            evaluator::DepositStep::GeneratingAdaptors {
+                deposit: ref dep,
+                ref withdrawal_chunks,
+            } => DepositStatus::Incomplete {
+                details: format!(
+                    "{} (deposit: {}, withdrawal: {})",
+                    deposit.step.step_name(),
+                    bool_progress(*dep),
+                    arr_progress(withdrawal_chunks),
+                ),
+            },
+            evaluator::DepositStep::SendingAdaptors { ref acked } => DepositStatus::Incomplete {
+                details: format!("{} ({})", deposit.step.step_name(), arr_progress(acked),),
             },
             evaluator::DepositStep::DepositReady => DepositStatus::Ready,
             evaluator::DepositStep::WithdrawnUndisputed => DepositStatus::UncontestedWithdrawal,
