magic-program & magic-program-api updated to use bincode 2

magicblock-accounts/src/scheduled_commits_processor.rs

@@ -5,7 +5,7 @@ use std::{
 
 use async_trait::async_trait;
 use magicblock_account_cloner::ChainlinkCloner;
-use magicblock_accounts_db::{traits::AccountsBank, AccountsDb};
+use magicblock_accounts_db::AccountsDb;
 use magicblock_chainlink::{
     remote_account_provider::{
         chain_rpc_client::ChainRpcClientImpl,
@@ -16,13 +16,11 @@ use magicblock_chainlink::{
 };
 use magicblock_committor_service::{
     intent_execution_manager::BroadcastedIntentExecutionResult,
-    intent_executor::ExecutionOutput,
-    types::{ScheduledBaseIntentWrapper, TriggerType},
-    BaseIntentCommittor, CommittorService,
+    intent_executor::ExecutionOutput, BaseIntentCommittor, CommittorService,
 };
 use magicblock_core::link::transactions::TransactionSchedulerHandle;
 use magicblock_program::{
-    magic_scheduled_base_intent::ScheduledBaseIntent,
+    magic_scheduled_base_intent::ScheduledIntentBundle,
     register_scheduled_commit_sent, SentCommit, TransactionScheduler,
 };
 use solana_hash::Hash;
@@ -50,7 +48,6 @@ pub type ChainlinkImpl = Chainlink<
 >;
 
 pub struct ScheduledCommitsProcessorImpl {
-    accounts_bank: Arc<AccountsDb>,
     committor: Arc<CommittorService>,
     chainlink: Arc<ChainlinkImpl>,
     cancellation_token: CancellationToken,
@@ -60,7 +57,6 @@ pub struct ScheduledCommitsProcessorImpl {
 
 impl ScheduledCommitsProcessorImpl {
     pub fn new(
-        accounts_bank: Arc<AccountsDb>,
         committor: Arc<CommittorService>,
         chainlink: Arc<ChainlinkImpl>,
         internal_transaction_scheduler: TransactionSchedulerHandle,
@@ -76,7 +72,6 @@ impl ScheduledCommitsProcessorImpl {
         ));
 
         Self {
-            accounts_bank,
             committor,
             chainlink,
             cancellation_token,
@@ -85,55 +80,6 @@ impl ScheduledCommitsProcessorImpl {
         }
     }
 
-    fn preprocess_intent(
-        &self,
-        mut base_intent: ScheduledBaseIntent,
-    ) -> (ScheduledBaseIntentWrapper, Vec<Pubkey>) {
-        let is_undelegate = base_intent.is_undelegate();
-        let Some(committed_accounts) = base_intent.get_committed_accounts_mut()
-        else {
-            let intent = ScheduledBaseIntentWrapper {
-                inner: base_intent,
-                trigger_type: TriggerType::OnChain,
-            };
-            return (intent, vec![]);
-        };
-
-        // Filter duplicate accounts
-        let mut seen = HashSet::with_capacity(committed_accounts.len());
-        committed_accounts.retain(|account| seen.insert(account.pubkey));
-
-        // dump undelegated pubkeys
-        let pubkeys_being_undelegated: Vec<_> = committed_accounts
-            .iter()
-            .inspect(|account| {
-                let pubkey = account.pubkey;
-                if self.accounts_bank.get_account(&pubkey).is_none() {
-                    // This doesn't affect intent validity
-                    // We assume that intent is correct at the moment of scheduling
-                    // All the checks are performed by runtime & magic-program at the moment of scheduling
-                    // This log could be a sign of eviction or a bug in implementation
-                    info!("Account got evicted from AccountsDB after intent was scheduled!");
-                }
-            })
-            .filter_map(|account| {
-                if is_undelegate {
-                    Some(account.pubkey)
-                } else {
-                    None
-                }
-            })
-            .collect();
-
-        let intent = ScheduledBaseIntentWrapper {
-            inner: base_intent,
-            trigger_type: TriggerType::OnChain,
-        };
-
-        (intent, pubkeys_being_undelegated)
-    }
-
-    #[instrument(skip(self))]
     async fn process_undelegation_requests(&self, pubkeys: Vec<Pubkey>) {
         let mut join_set = task::JoinSet::new();
         for pubkey in pubkeys.into_iter() {
@@ -192,7 +138,7 @@ impl ScheduledCommitsProcessorImpl {
             let execution_result = tokio::select! {
                 biased;
                 _ = cancellation_token.cancelled() => {
-                    info!("Shutting down result processor");
+                    info!("Shutting down");
                     return;
                 }
                 execution_result = result_receiver.recv() => {
@@ -213,14 +159,6 @@ impl ScheduledCommitsProcessorImpl {
             };
 
             let intent_id = execution_result.id;
-            let trigger_type = execution_result.trigger_type;
-            // Here we handle on OnChain triggered intent
-            // TODO: should be removed once crank supported
-            if matches!(trigger_type, TriggerType::OffChain) {
-                info!(intent_id, trigger_type = ?trigger_type, "intent executed");
-                continue;
-            }
-
             // Remove intent from metas
             let intent_meta = if let Some(intent_meta) = intents_meta_map
                 .lock()
@@ -331,44 +269,36 @@ impl ScheduledCommitsProcessorImpl {
 impl ScheduledCommitsProcessor for ScheduledCommitsProcessorImpl {
     #[instrument(skip(self))]
     async fn process(&self) -> ScheduledCommitsProcessorResult<()> {
-        let scheduled_base_intents =
-            self.transaction_scheduler.take_scheduled_actions();
+        let intent_bundles =
+            self.transaction_scheduler.take_scheduled_intent_bundles();
 
-        if scheduled_base_intents.is_empty() {
+        if intent_bundles.is_empty() {
             return Ok(());
         }
 
-        let intents = scheduled_base_intents
-            .into_iter()
-            .map(|intent| self.preprocess_intent(intent));
-
         // Add metas for intent we schedule
-        let (intents, pubkeys_being_undelegated) = {
+        let pubkeys_being_undelegated = {
             let mut intent_metas =
                 self.intents_meta_map.lock().expect(POISONED_MUTEX_MSG);
-            let mut pubkeys_being_undelegated = HashSet::new();
-
-            let intents = intents
-                .map(|(intent, undelegated)| {
-                    intent_metas.insert(
-                        intent.id,
-                        ScheduledBaseIntentMeta::new(&intent),
-                    );
-                    pubkeys_being_undelegated.extend(undelegated);
-
-                    intent
-                })
-                .collect::<Vec<_>>();
+            let mut pubkeys_being_undelegated = HashSet::<Pubkey>::new();
+
+            intent_bundles.iter().for_each(|intent| {
+                intent_metas
+                    .insert(intent.id, ScheduledBaseIntentMeta::new(intent));
+                if let Some(undelegate) = intent.get_undelegate_intent_pubkeys()
+                {
+                    pubkeys_being_undelegated.extend(undelegate);
+                }
+            });
 
-            (
-                intents,
-                pubkeys_being_undelegated.into_iter().collect::<Vec<_>>(),
-            )
+            pubkeys_being_undelegated.into_iter().collect::<Vec<_>>()
         };
 
         self.process_undelegation_requests(pubkeys_being_undelegated)
             .await;
-        self.committor.schedule_base_intent(intents).await??;
+        self.committor
+            .schedule_intent_bundles(intent_bundles)
+            .await??;
         Ok(())
     }
 
@@ -395,16 +325,14 @@ struct ScheduledBaseIntentMeta {
 }
 
 impl ScheduledBaseIntentMeta {
-    fn new(intent: &ScheduledBaseIntent) -> Self {
+    fn new(intent: &ScheduledIntentBundle) -> Self {
         Self {
             slot: intent.slot,
             blockhash: intent.blockhash,
             payer: intent.payer,
-            included_pubkeys: intent
-                .get_committed_pubkeys()
-                .unwrap_or_default(),
-            intent_sent_transaction: intent.action_sent_transaction.clone(),
-            requested_undelegation: intent.is_undelegate(),
+            included_pubkeys: intent.get_all_committed_pubkeys(),
+            intent_sent_transaction: intent.sent_transaction.clone(),
+            requested_undelegation: intent.has_undelegate_intent(),
         }
     }
 }

magicblock-api/src/magic_validator.rs

@@ -209,7 +209,6 @@ impl MagicValidator {
         let scheduled_commits_processor =
             committor_service.as_ref().map(|committor_service| {
                 Arc::new(ScheduledCommitsProcessorImpl::new(
-                    accountsdb.clone(),
                     committor_service.clone(),
                     chainlink.clone(),
                     dispatch.transaction_scheduler.clone(),

magicblock-committor-service/src/committor_processor.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashSet, path::Path, sync::Arc};
 
+use magicblock_program::magic_scheduled_base_intent::ScheduledIntentBundle;
 use magicblock_rpc_client::MagicblockRpcClient;
 use magicblock_table_mania::{GarbageCollectorConfig, TableMania};
 use solana_keypair::Keypair;
@@ -19,7 +20,6 @@ use crate::{
         CommitStatusRow, IntentPersister, IntentPersisterImpl,
         MessageSignatures,
     },
-    types::ScheduledBaseIntentWrapper,
 };
 
 pub(crate) struct CommittorProcessor {
@@ -121,24 +121,20 @@ impl CommittorProcessor {
         Ok(signatures)
     }
 
-    #[instrument(skip(self, base_intents))]
-    pub async fn schedule_base_intents(
+    #[instrument(skip(self, intent_bundles))]
+    pub async fn schedule_intent_bundle(
         &self,
-        base_intents: Vec<ScheduledBaseIntentWrapper>,
+        intent_bundles: Vec<ScheduledIntentBundle>,
     ) -> CommittorServiceResult<()> {
-        let intents = base_intents
-            .iter()
-            .map(|base_intent| base_intent.inner.clone())
-            .collect::<Vec<_>>();
-        if let Err(err) = self.persister.start_base_intents(&intents) {
+        if let Err(err) = self.persister.start_base_intents(&intent_bundles) {
             // We will still try to perform the commits, but the fact that we cannot
             // persist the intent is very serious and we should probably restart the
             // valiator
             error!(error = ?err, "DB EXCEPTION: Failed to persist changeset");
         };
 
         self.commits_scheduler
-            .schedule(base_intents)
+            .schedule(intent_bundles)
             .await
             .inspect_err(|err| {
                 error!(error = ?err, "Failed to schedule intent");

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_program::magic_scheduled_base_intent::ScheduledIntentBundle;
 use magicblock_rpc_client::MagicblockRpcClient;
 use magicblock_table_mania::TableMania;
 use tokio::sync::{broadcast, mpsc, mpsc::error::TrySendError};
@@ -19,14 +20,13 @@ use crate::{
         task_info_fetcher::CacheTaskInfoFetcher,
     },
     persist::IntentPersister,
-    types::ScheduledBaseIntentWrapper,
     ComputeBudgetConfig,
 };
 
 pub struct IntentExecutionManager<D: DB> {
     db: Arc<D>,
     result_subscriber: ResultSubscriber,
-    intent_sender: mpsc::Sender<ScheduledBaseIntentWrapper>,
+    intent_sender: mpsc::Sender<ScheduledIntentBundle>,
 }
 
 impl<D: DB> IntentExecutionManager<D> {
@@ -69,18 +69,18 @@ impl<D: DB> IntentExecutionManager<D> {
     /// Intents will be extracted and handled in the [`IntentExecutionEngine`]
     pub async fn schedule(
         &self,
-        base_intents: Vec<ScheduledBaseIntentWrapper>,
+        intent_bundles: Vec<ScheduledIntentBundle>,
     ) -> Result<(), IntentExecutionManagerError> {
         // If db not empty push el-t there
         // This means that at some point channel got full
         // Worker first will clean-up channel, and then DB.
         // Pushing into channel would break order of commits
         if !self.db.is_empty() {
-            self.db.store_base_intents(base_intents).await?;
+            self.db.store_intent_bundles(intent_bundles).await?;
             return Ok(());
         }
 
-        for el in base_intents {
+        for el in intent_bundles {
             let err = if let Err(err) = self.intent_sender.try_send(el) {
                 err
             } else {
@@ -93,7 +93,7 @@ impl<D: DB> IntentExecutionManager<D> {
                 }
                 TrySendError::Full(el) => self
                     .db
-                    .store_base_intent(el)
+                    .store_intent_bundle(el)
                     .await
                     .map_err(IntentExecutionManagerError::from),
             }?;