feat: action callbacks

Cargo.toml

@@ -20,6 +20,7 @@ members = [
   "magicblock-metrics",
   "magicblock-processor",
   "magicblock-rpc-client",
+  "magicblock-services",
   "magicblock-table-mania",
   "magicblock-task-scheduler",
   "magicblock-validator",
@@ -109,6 +110,7 @@ magicblock-metrics = { path = "./magicblock-metrics" }
 magicblock-processor = { path = "./magicblock-processor" }
 magicblock-program = { path = "./programs/magicblock" }
 magicblock-rpc-client = { path = "./magicblock-rpc-client" }
+magicblock-services = { path = "./magicblock-services" }
 magicblock-table-mania = { path = "./magicblock-table-mania" }
 magicblock-task-scheduler = { path = "./magicblock-task-scheduler" }
 magicblock-tui-client = { path = "./tools/magicblock-tui-client" }

magicblock-accounts/src/scheduled_commits_processor.rs

@@ -255,6 +255,23 @@ impl ScheduledCommitsProcessorImpl {
             })
             .collect();
 
+        let callbacks_report = result
+            .callbacks_report
+            .iter()
+            .map(|r| match r {
+                Ok(sig) => {
+                    format!("OK: {sig}")
+                }
+                Err(err) => {
+                    error!(
+                        "Callback failed to schedule: {}. error: {}",
+                        intent_id, err
+                    );
+                    format!("ERR: {err}")
+                }
+            })
+            .collect();
+
         SentCommit {
             message_id: intent_id,
             slot: intent_meta.slot,
@@ -266,6 +283,7 @@ impl ScheduledCommitsProcessorImpl {
             requested_undelegation: intent_meta.requested_undelegation,
             error_message,
             patched_errors,
+            callbacks_scheduling_results: callbacks_report,
         }
     }
 }

magicblock-api/Cargo.toml

@@ -28,6 +28,7 @@ magicblock-magic-program-api = { workspace = true }
 magicblock-metrics = { workspace = true }
 magicblock-processor = { workspace = true }
 magicblock-program = { workspace = true }
+magicblock-services = { workspace = true }
 magicblock-task-scheduler = { workspace = true }
 magicblock-validator-admin = { workspace = true }
 

magicblock-api/src/magic_sys_adapter.rs

@@ -1,11 +1,11 @@
-use std::{collections::HashMap, error::Error, sync::Arc, time::Duration};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use magicblock_committor_service::CommittorService;
 use magicblock_core::{intent::CommittedAccount, traits::MagicSys};
 use magicblock_metrics::metrics;
 use solana_instruction::error::InstructionError;
 use solana_pubkey::Pubkey;
-use tracing::{error, trace};
+use tracing::error;
 
 #[derive(Clone)]
 pub struct MagicSysAdapter {
@@ -30,15 +30,6 @@ impl MagicSysAdapter {
 }
 
 impl MagicSys for MagicSysAdapter {
-    fn persist(&self, id: u64, data: Vec<u8>) -> Result<(), Box<dyn Error>> {
-        trace!(id, data_len = data.len(), "Persisting data");
-        Ok(())
-    }
-
-    fn load(&self, _id: u64) -> Result<Option<Vec<u8>>, Box<dyn Error>> {
-        Ok(None)
-    }
-
     fn fetch_current_commit_nonces(
         &self,
         commits: &[CommittedAccount],

magicblock-api/src/magic_validator.rs

@@ -31,7 +31,7 @@ use magicblock_chainlink::{
 };
 use magicblock_committor_service::{
     config::ChainConfig, BaseIntentCommittor, CommittorService,
-    ComputeBudgetConfig,
+    ComputeBudgetConfig, DEFAULT_ACTIONS_TIMEOUT,
 };
 use magicblock_config::{
     config::{
@@ -65,6 +65,7 @@ use magicblock_program::{
     validator::{self, validator_authority},
     TransactionScheduler as ActionTransactionScheduler,
 };
+use magicblock_services::actions_callback_service::ActionsCallbackService;
 use magicblock_task_scheduler::{SchedulerDatabase, TaskSchedulerService};
 use magicblock_validator_admin::claim_fees::ClaimFeesTask;
 use mdp::state::{
@@ -220,7 +221,9 @@ impl MagicValidator {
         let (mut dispatch, validator_channels) = link();
 
         let step_start = Instant::now();
-        let committor_service = Self::init_committor_service(&config).await?;
+        let committor_service =
+            Self::init_committor_service(&config, ledger.latest_block())
+                .await?;
         log_timing("startup", "committor_service_init", step_start);
         init_magic_sys(Arc::new(MagicSysAdapter::new(
             committor_service.clone(),
@@ -391,14 +394,20 @@ impl MagicValidator {
         })
     }
 
-    #[instrument(skip(config))]
+    #[instrument(skip(config, latest_block))]
     async fn init_committor_service(
         config: &ValidatorParams,
+        latest_block: &LatestBlock,
     ) -> ApiResult<Option<Arc<CommittorService>>> {
         let committor_persist_path =
             config.storage.join("committor_service.sqlite");
         debug!(path = %committor_persist_path.display(), "Initializing committor service");
         // TODO(thlorenz): when we support lifecycle modes again, only start it when needed
+        let actions_callback_executor = ActionsCallbackService::new(
+            Arc::new(RpcClient::new(config.aperture.listen.http())),
+            config.validator.keypair.insecure_clone(),
+            latest_block.clone(),
+        );
         let committor_service = Some(Arc::new(CommittorService::try_start(
             config.validator.keypair.insecure_clone(),
             committor_persist_path,
@@ -408,7 +417,9 @@ impl MagicValidator {
                 compute_budget_config: ComputeBudgetConfig::new(
                     config.commit.compute_unit_price,
                 ),
+                actions_timeout: DEFAULT_ACTIONS_TIMEOUT,
             },
+            actions_callback_executor,
         )?));
 
         if let Some(committor_service) = &committor_service {

magicblock-committor-service/src/committor_processor.rs

@@ -4,6 +4,7 @@ use std::{
     sync::Arc,
 };
 
+use magicblock_core::traits::ActionsCallbackScheduler;
 use magicblock_program::magic_scheduled_base_intent::ScheduledIntentBundle;
 use magicblock_rpc_client::MagicblockRpcClient;
 use magicblock_table_mania::{GarbageCollectorConfig, TableMania};
@@ -20,9 +21,12 @@ use crate::{
     intent_execution_manager::{
         db::DummyDB, BroadcastedIntentExecutionResult, IntentExecutionManager,
     },
-    intent_executor::task_info_fetcher::{
-        CacheTaskInfoFetcher, RpcTaskInfoFetcher, TaskInfoFetcher,
-        TaskInfoFetcherResult,
+    intent_executor::{
+        intent_executor_factory::ExecutorConfig,
+        task_info_fetcher::{
+            CacheTaskInfoFetcher, RpcTaskInfoFetcher, TaskInfoFetcher,
+            TaskInfoFetcherResult,
+        },
     },
     persist::{
         CommitStatusRow, IntentPersister, IntentPersisterImpl,
@@ -40,13 +44,15 @@ pub(crate) struct CommittorProcessor {
 }
 
 impl CommittorProcessor {
-    pub fn try_new<P>(
+    pub fn try_new<P, A>(
         authority: Keypair,
         persist_file: P,
         chain_config: ChainConfig,
+        actions_callback_executor: A,
     ) -> CommittorServiceResult<Self>
     where
         P: AsRef<Path>,
+        A: ActionsCallbackScheduler,
     {
         let rpc_client = RpcClient::new_with_commitment(
             chain_config.rpc_uri.to_string(),
@@ -76,7 +82,13 @@ impl CommittorProcessor {
             task_info_fetcher.clone(),
             Some(persister.clone()),
             table_mania.clone(),
-            chain_config.compute_budget_config.clone(),
+            ExecutorConfig {
+                compute_budget_config: chain_config
+                    .compute_budget_config
+                    .clone(),
+                actions_timeout: chain_config.actions_timeout,
+            },
+            actions_callback_executor,
         );
 
         Ok(Self {

magicblock-committor-service/src/config.rs

@@ -1,12 +1,17 @@
+use std::time::Duration;
+
 use solana_commitment_config::CommitmentConfig;
 
 use crate::compute_budget::ComputeBudgetConfig;
 
+pub const DEFAULT_ACTIONS_TIMEOUT: Duration = Duration::from_secs(60);
+
 #[derive(Debug, Clone)]
 pub struct ChainConfig {
     pub rpc_uri: String,
     pub commitment: CommitmentConfig,
     pub compute_budget_config: ComputeBudgetConfig,
+    pub actions_timeout: Duration,
 }
 
 impl ChainConfig {
@@ -15,6 +20,7 @@ impl ChainConfig {
             rpc_uri: "https://api.devnet.solana.com".to_string(),
             commitment: CommitmentConfig::confirmed(),
             compute_budget_config,
+            actions_timeout: DEFAULT_ACTIONS_TIMEOUT,
         }
     }
 
@@ -23,6 +29,7 @@ impl ChainConfig {
             rpc_uri: "https://api.mainnet-beta.solana.com".to_string(),
             commitment: CommitmentConfig::confirmed(),
             compute_budget_config,
+            actions_timeout: DEFAULT_ACTIONS_TIMEOUT,
         }
     }
 
@@ -31,6 +38,7 @@ impl ChainConfig {
             rpc_uri: "http://localhost:7799".to_string(),
             commitment: CommitmentConfig::processed(),
             compute_budget_config,
+            actions_timeout: DEFAULT_ACTIONS_TIMEOUT,
         }
     }
 }

magicblock-committor-service/src/intent_execution_manager.rs

@@ -5,6 +5,7 @@ pub mod intent_scheduler;
 use std::sync::Arc;
 
 pub use intent_execution_engine::BroadcastedIntentExecutionResult;
+use magicblock_core::traits::ActionsCallbackScheduler;
 use magicblock_program::magic_scheduled_base_intent::ScheduledIntentBundle;
 use magicblock_rpc_client::MagicblockRpcClient;
 use magicblock_table_mania::TableMania;
@@ -16,11 +17,10 @@ use crate::{
         intent_execution_engine::{IntentExecutionEngine, ResultSubscriber},
     },
     intent_executor::{
-        intent_executor_factory::IntentExecutorFactoryImpl,
+        intent_executor_factory::{ExecutorConfig, IntentExecutorFactoryImpl},
         task_info_fetcher::{CacheTaskInfoFetcher, RpcTaskInfoFetcher},
     },
     persist::IntentPersister,
-    ComputeBudgetConfig,
 };
 
 pub struct IntentExecutionManager<D: DB> {
@@ -30,21 +30,27 @@ pub struct IntentExecutionManager<D: DB> {
 }
 
 impl<D: DB> IntentExecutionManager<D> {
-    pub fn new<P: IntentPersister>(
+    pub fn new<P, A>(
         rpc_client: MagicblockRpcClient,
         db: D,
         task_info_fetcher: Arc<CacheTaskInfoFetcher<RpcTaskInfoFetcher>>,
         intent_persister: Option<P>,
         table_mania: TableMania,
-        compute_budget_config: ComputeBudgetConfig,
-    ) -> Self {
+        executor_config: ExecutorConfig,
+        actions_callback_executor: A,
+    ) -> Self
+    where
+        A: ActionsCallbackScheduler,
+        P: IntentPersister,
+    {
         let db = Arc::new(db);
 
         let executor_factory = IntentExecutorFactoryImpl {
             rpc_client,
             table_mania,
-            compute_budget_config,
+            executor_config,
             task_info_fetcher,
+            actions_callback_executor,
         };
 
         let (sender, receiver) = mpsc::channel(1000);