review and fix callbacks during undelegation recovery

Report failure callbacks before discarding an undelegation and its following actions. Extract callbacks through the existing authorization checks, retain the failure signature, and leave callbacks on retained actions for the eventual retry result.

Move the sole transaction loop onto TransactionExecutor. Remove the borrowed ExecutionState adapter and closure parameters while preserving pending-signature resolution, outbox-before-send ordering, cancellation state, cleanup ownership, and retry limits.

Remove the unused InternalError transaction-size wrapper and replace four small predicate tests with one table-driven test of the surviving send error mapper. Extend the existing recovery matrix to check discarded and retained callback identities, outcomes, and failure signatures.

Validation: cargo fmt; cargo check -p magicblock-committor-service --tests; cargo test -p magicblock-committor-service recovery_preserves_progress_and_resources -- --nocapture; git diff --check.

Author: snawaz

committor/src/intent_executor/error.rs

@@ -43,13 +43,6 @@ impl InternalError {
             Self::MagicBlockRpcClientError(err) => err.is_transient(),
         }
     }
-
-    pub fn is_transaction_too_large(&self) -> bool {
-        match self {
-            Self::MagicBlockRpcClientError(err) => err.is_transaction_too_large(),
-            Self::SignerError(_) => false,
-        }
-    }
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -143,53 +136,20 @@ mod tests {
     use rpc_client::MagicBlockRpcClientError;
     use solana_rpc_client_api::{
         client_error::{Error as RpcClientError, ErrorKind as RpcClientErrorKind},
-        request::{RpcError, RpcRequest, RpcResponseErrorData},
+        request::RpcRequest,
     };
 
     use super::InternalError;
-    use crate::intent_executor::strategy_executor::error::TransactionStrategyExecutionError;
-
-    const TX_TOO_LARGE_SOLANA: &str = "base64 encoded too large";
-    const TX_TOO_LARGE_MAGICBLOCK: &str = "base64 encoded solana_transaction::versioned::VersionedTransaction too large: 1684 bytes (max: encoded/raw 1644/1232)";
-
-    fn make_send_transaction_error(message: &str) -> InternalError {
-        let rpc_error = RpcClientError {
-            request: Some(RpcRequest::SendTransaction),
-            kind: Box::new(RpcClientErrorKind::RpcError(RpcError::RpcResponseError {
-                code: -32602,
-                message: message.to_string(),
-                data: RpcResponseErrorData::Empty,
-            })),
-        };
-        InternalError::MagicBlockRpcClientError(Box::new(
-            MagicBlockRpcClientError::SendTransaction(Box::new(rpc_error)),
-        ))
-    }
-
-    #[test]
-    fn is_transaction_too_large_matches_solana_format() {
-        let err = make_send_transaction_error(TX_TOO_LARGE_SOLANA);
-        assert!(err.is_transaction_too_large());
-    }
-
-    #[test]
-    fn is_transaction_too_large_matches_magicblock_format() {
-        let err = make_send_transaction_error(TX_TOO_LARGE_MAGICBLOCK);
-        assert!(err.is_transaction_too_large());
-    }
-
-    #[test]
-    fn transaction_too_large_error_is_recoverable_by_splitting() {
-        let inner = make_send_transaction_error(TX_TOO_LARGE_MAGICBLOCK);
-        let err = TransactionStrategyExecutionError::TransactionTooLargeError(inner);
-        assert!(err.is_recoverable_by_splitting());
-    }
+    use crate::intent_executor::{
+        strategy_executor::error::TransactionStrategyExecutionError,
+        test_utils::send_transaction_error,
+    };
 
     #[test]
     fn transport_errors_are_transient() {
-        let err = TransactionStrategyExecutionError::InternalError(make_send_transaction_error(
-            "connection reset",
-        ));
+        let err = TransactionStrategyExecutionError::InternalError(
+            send_transaction_error("connection reset").into(),
+        );
         assert!(err.is_transient());
 
         let err = TransactionStrategyExecutionError::InternalError(
@@ -309,9 +269,9 @@ mod tests {
         use solana_signature::Signature;
 
         let transient_err = || {
-            TransactionStrategyExecutionError::InternalError(make_send_transaction_error(
-                "connection reset",
-            ))
+            TransactionStrategyExecutionError::InternalError(
+                send_transaction_error("connection reset").into(),
+            )
         };
 
         let signature = Signature::new_unique();
@@ -336,16 +296,4 @@ mod tests {
             );
         }
     }
-
-    #[test]
-    fn unrelated_internal_errors_do_not_trigger_splitting() {
-        let err = TransactionStrategyExecutionError::InternalError(
-            InternalError::MagicBlockRpcClientError(Box::new(
-                MagicBlockRpcClientError::LookupTableDeserialize(
-                    solana_instruction::error::InstructionError::GenericError,
-                ),
-            )),
-        );
-        assert!(!err.is_recoverable_by_splitting());
-    }
 }

committor/src/intent_executor/intent_execution_client.rs

@@ -276,3 +276,37 @@ impl SendErrorMapper<MagicBlockRpcClientError> for DefaultGetErrorMapper {
         decide_rpc_error_flow(mapped_error)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::intent_executor::test_utils::send_transaction_error;
+
+    #[test]
+    fn send_errors_preserve_split_recovery() {
+        let mapper = IntentErrorMapper {
+            transaction_error_mapper: IntentTransactionErrorMapper {
+                tasks: &[],
+                task_instruction_offset: 0,
+            },
+        };
+        for (error, splits) in [
+            (send_transaction_error("base64 encoded too large"), true),
+            (
+                send_transaction_error(
+                    "base64 encoded solana_transaction::versioned::VersionedTransaction too large: 1684 bytes (max: encoded/raw 1644/1232)",
+                ),
+                true,
+            ),
+            (send_transaction_error("connection reset"), false),
+            (
+                MagicBlockRpcClientError::LookupTableDeserialize(
+                    solana_instruction::error::InstructionError::GenericError,
+                ),
+                false,
+            ),
+        ] {
+            assert_eq!(mapper.map(error).is_recoverable_by_splitting(), splits);
+        }
+    }
+}

committor/src/intent_executor/strategy_executor/recovery.rs

@@ -96,7 +96,17 @@ impl<F: TaskInfoFetcher, A: ActionsCallbackScheduler> TransactionRecovery<'_, F,
                     .iter()
                     .any(|task| matches!(task, BaseTaskImpl::Undelegate(_))) =>
             {
-                handle_undelegation_error(&self.authority.pubkey(), strategy)
+                let mut removed = handle_undelegation_error(&self.authority.pubkey(), strategy);
+                // Report only discarded actions; retained actions still need to execute.
+                let callbacks = removed.extract_action_callbacks();
+                if !callbacks.is_empty() {
+                    report.add_callback_report(self.callback_scheduler.schedule(
+                        callbacks,
+                        err.signature(),
+                        Err(ActionError::from(err)),
+                    ));
+                }
+                removed
             }
             _ => return Ok(ControlFlow::Break(())),
         };

committor/src/intent_executor/strategy_executor/transaction.rs

@@ -1,27 +1,32 @@
-use std::{mem, time::Duration};
+use std::{mem, ops::ControlFlow, time::Duration};
 
 use mbv_core::traits::{ActionError, ActionsCallbackScheduler};
 use program::outbox::PendingTransaction;
 use solana_pubkey::Pubkey;
+use solana_rpc_client::rpc_client::SerializableTransaction;
 use solana_signature::Signature;
 use solana_signer::Signer;
 use tokio::time::timeout;
-use tracing::info;
+use tracing::{error, info, warn};
 
 use super::{
     error::TransactionStrategyExecutionError,
     recovery::TransactionRecovery,
     utils::{
-        ExecutionState, handle_actions_result, handle_cpi_limit_error, transaction_execution_loop,
+        handle_actions_result, handle_cpi_limit_error, prepare_transaction, recovery_commit_nonces,
+        resolve_pending_signature,
     },
 };
 use crate::{
     intent_executor::{
         ExecutionOutput, IntentExecutionReport, IntentExecutorCtx,
         error::{IntentExecutorError, IntentExecutorResult},
-        progress::ExecutionProgress,
+        progress::{ExecutionProgress, TransactionProgress},
+    },
+    outbox::{
+        OutboxClient, outbox_intent_bundles_reader::OutboxIntentBundlesReader,
+        progress::PendingExecution,
     },
-    outbox::{OutboxClient, progress::PendingExecution},
     tasks::{
         BaseTaskImpl,
         task_info_fetcher::TaskInfoFetcher,
@@ -33,6 +38,8 @@ use crate::{
 #[cfg(test)]
 mod tests;
 
+const TRANSACTION_LOOP_CEILING: u8 = 10;
+
 /// Runs the selected transaction and any remaining actions or undelegations.
 /// A confirmed commit is never included in subsequent recovery attempts.
 pub(in crate::intent_executor) struct TransactionExecutor<'a, T, F, A, O> {
@@ -192,47 +199,149 @@ where
         let ctx = self.ctx;
         let confirmed_commit = self.report.progress.confirmed_commit;
         let has_follow_up = self.follow_up.is_some();
-        let make_outbox_stage = |transaction| {
-            PendingExecution {
-                transaction,
-                confirmed_commit,
-                has_follow_up,
-            }
-            .into_stage()
+        let recovery = TransactionRecovery {
+            authority: &ctx.authority,
+            intent_client: &ctx.intent_client,
+            intent_id: self.intent_id,
+            callback_scheduler: &ctx.actions_callback_executor,
+            task_info_fetcher: &ctx.task_info_fetcher,
+            committed_pubkeys: self.committed_pubkeys,
         };
-        let state = ExecutionState {
-            current_attempt: &mut self.current_attempt,
-            transaction_strategy: &mut self.current,
-            pending_transaction: &mut self.pending_transaction,
-            execution_report: self.report,
-            commit_signature: confirmed_commit,
-        };
-        transaction_execution_loop(
-            &ctx.authority,
-            &ctx.intent_client,
-            ctx.outbox_client.as_ref(),
-            &ctx.transaction_preparator,
-            TransactionRecovery {
-                authority: &ctx.authority,
-                intent_client: &ctx.intent_client,
-                intent_id: self.intent_id,
-                callback_scheduler: &ctx.actions_callback_executor,
-                task_info_fetcher: &ctx.task_info_fetcher,
-                committed_pubkeys: self.committed_pubkeys,
-            },
-            self.intent_id,
-            make_outbox_stage,
-            |status| {
-                PendingExecution::from_status(status)
-                    .filter(|pending| {
-                        pending.has_follow_up == has_follow_up
-                            && pending.confirmed_commit.is_some() == confirmed_commit.is_some()
-                    })
-                    .map(|pending| pending.transaction)
-            },
-            state,
-        )
-        .await
+        let mut reconciled_outbox = false;
+        loop {
+            if self.pending_transaction.is_none() && !reconciled_outbox {
+                reconciled_outbox = true;
+                self.pending_transaction = self.recover_pending_from_outbox().await;
+            }
+
+            if let Some(ref pending) = self.pending_transaction {
+                self.report.progress.transaction =
+                    TransactionProgress::Submitted(pending.signature);
+                if resolve_pending_signature(&ctx.intent_client, pending).await? {
+                    self.report.progress.transaction =
+                        TransactionProgress::Confirmed(pending.signature);
+                    break Ok(Ok(pending.signature));
+                }
+                self.report.progress.transaction = TransactionProgress::Rejected(pending.signature);
+                self.pending_transaction = None;
+            }
+
+            // Resolve a previously sent transaction before skipping an empty follow-up.
+            if let Some(commit_signature) = confirmed_commit
+                && self.current.optimized_tasks.is_empty()
+            {
+                return Ok(Ok(commit_signature));
+            }
+
+            self.current_attempt += 1;
+
+            // Prepare message
+            let prepared_transaction = prepare_transaction(
+                &ctx.intent_client,
+                &ctx.authority,
+                &ctx.transaction_preparator,
+                &mut self.current,
+            )
+            .await
+            .map_err(IntentExecutorError::FailedPreparation)?;
+
+            // Record in outbox before sending
+            let pending = PendingTransaction {
+                signature: *prepared_transaction.get_signature(),
+                blockhash: *prepared_transaction.get_recent_blockhash(),
+            };
+            self.report.progress.transaction = TransactionProgress::Prepared(pending.signature);
+            let recovery_commit_nonces = recovery_commit_nonces(&self.current.optimized_tasks);
+            ctx.outbox_client
+                .set_intent_execution_stage(
+                    self.intent_id,
+                    PendingExecution {
+                        transaction: pending,
+                        confirmed_commit,
+                        has_follow_up,
+                    }
+                    .into_stage(),
+                    recovery_commit_nonces,
+                )
+                .await
+                .map_err(Into::into)?;
+
+            // Now record locally signature of tx we about to send
+            // Precaution for timeout in between
+            self.pending_transaction = Some(pending);
+            self.report.progress.transaction = TransactionProgress::Submitted(pending.signature);
+            let signature = &pending.signature;
+
+            // Send signed tx
+            let execution_result = ctx
+                .intent_client
+                .send_signed_tx_with_retries(&prepared_transaction, &self.current.optimized_tasks)
+                .await;
+
+            // Result returned, cleanup pending signature
+            self.pending_transaction = None;
+
+            let execution_err = match execution_result {
+                Ok(()) => {
+                    self.report.progress.transaction = TransactionProgress::Confirmed(*signature);
+                    return Ok(Ok(*signature));
+                }
+                Err(err) => err,
+            };
+            if execution_err.is_rejected() {
+                self.report.progress.transaction = TransactionProgress::Rejected(*signature);
+            }
+            let flow = recovery.recover(&execution_err, &mut self.current, self.report).await?;
+            let cleanup = match flow {
+                ControlFlow::Continue(value) => value,
+                ControlFlow::Break(()) => return Ok(Err(execution_err)),
+            };
+            ctx.intent_client.invalidate_cached_blockhash().await;
+            self.report.dispose(cleanup);
+
+            if let Some(commit_signature) = confirmed_commit
+                && self.current.optimized_tasks.is_empty()
+            {
+                // Failed actions can leave no work after a confirmed commit.
+                // Keep undelegation failures visible even when patching removes all tasks.
+                if matches!(
+                    execution_err,
+                    TransactionStrategyExecutionError::ActionsError(..)
+                ) {
+                    self.report.add_patched_error(execution_err);
+                    return Ok(Ok(commit_signature));
+                }
+                return Ok(Err(execution_err));
+            }
+
+            if self.current_attempt >= TRANSACTION_LOOP_CEILING {
+                error!("CRITICAL! Recursion ceiling reached");
+                return Ok(Err(execution_err));
+            } else {
+                self.report.add_patched_error(execution_err);
+            }
+        }
+    }
+
+    async fn recover_pending_from_outbox(&self) -> Option<PendingTransaction> {
+        let reader = self.ctx.outbox_client.outbox_reader();
+        match reader.fetch_outbox_intent(self.intent_id).await {
+            Ok(Some(bundle)) => PendingExecution::from_status(bundle.status())
+                .filter(|pending| {
+                    pending.has_follow_up == self.follow_up.is_some()
+                        && pending.confirmed_commit.is_some()
+                            == self.report.progress.confirmed_commit.is_some()
+                })
+                .map(|pending| pending.transaction),
+            Ok(None) => None,
+            Err(_) => {
+                warn!(
+                    intent_id = self.intent_id,
+                    "Failed to reconcile pending transaction from outbox"
+                );
+                None
+            }
+        }
     }
 
     fn execute_callbacks(&mut self, signature: Option<Signature>, result: Result<(), ActionError>) {