feat: Handle post delegation actions: detect actions, parse and execute on the ER

Cargo.toml

@@ -102,8 +102,8 @@ magicblock-committor-program = { path = "./magicblock-committor-program", featur
 magicblock-committor-service = { path = "./magicblock-committor-service" }
 magicblock-config = { path = "./magicblock-config" }
 magicblock-core = { path = "./magicblock-core" }
-magicblock-delegation-program = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "2cb491032f372", features = [
-  "no-entrypoint",
+magicblock-delegation-program = { path = "../delegation-program", default-features = false, features = [
+  "sdk",
 ] }
 magicblock-ledger = { path = "./magicblock-ledger" }
 magicblock-magic-program-api = { path = "./magicblock-magic-program-api" }

magicblock-account-cloner/src/lib.rs

@@ -122,7 +122,7 @@ impl ChainlinkCloner {
             message,
         );
         // Defined positive commit frequency means commits should be scheduled
-        let ixs = match request.commit_frequency_ms {
+        let mut ixs = match request.commit_frequency_ms {
             // TODO(GabrielePicco): Hotfix. Do not schedule frequency commits until we impose limits.
             // 1. Allow configuring a higher minimum.
             // 2. Stop committing accounts if they have been committed more than X times,
@@ -162,6 +162,9 @@ impl ChainlinkCloner {
             }
             _ => vec![modify_ix],
         };
+        if !request.delegation_actions.is_empty() {
+            ixs.extend(request.delegation_actions.clone());
+        }
 
         let mut tx =
             Transaction::new_with_payer(&ixs, Some(&validator_authority_id()));

magicblock-accounts/src/scheduled_commits_processor.rs

@@ -238,7 +238,7 @@ impl ScheduledCommitsProcessorImpl {
                             .map(|finalize| vec![commit, finalize])
                             .unwrap_or(vec![commit])
                     })
-                    .unwrap_or(vec![])
+                    .unwrap_or_default()
             }
         };
         let patched_errors = result

magicblock-chainlink/src/chainlink/errors.rs

@@ -20,6 +20,8 @@ pub enum ChainlinkError {
 
     #[error("Delegation record could not be decoded: {0} ({1:?})")]
     InvalidDelegationRecord(Pubkey, ProgramError),
+    #[error("Delegation actions could not be decoded: {0} ({1})")]
+    InvalidDelegationActions(Pubkey, String),
 
     #[error("Failed to resolve one or more accounts {0} when getting delegation records")]
     DelegatedAccountResolutionsFailed(String),

magicblock-chainlink/src/chainlink/fetch_cloner/ata_projection.rs

@@ -12,7 +12,7 @@ use tracing::*;
 
 use super::{delegation, types::AccountWithCompanion, FetchCloner};
 use crate::{
-    cloner::{AccountCloneRequest, Cloner},
+    cloner::{AccountCloneRequest, Cloner, DelegationActions},
     remote_account_provider::{ChainPubsubClient, ChainRpcClient},
 };
 
@@ -127,15 +127,16 @@ where
             )
             .await
             {
+                let (deleg_record, _delegation_actions) = deleg;
                 delegated_to_other =
-                    delegation::get_delegated_to_other(this, &deleg);
-                commit_frequency_ms = Some(deleg.commit_frequency_ms);
+                    delegation::get_delegated_to_other(this, &deleg_record);
+                commit_frequency_ms = Some(deleg_record.commit_frequency_ms);
 
                 if let Some(projected_ata) = this
                     .maybe_project_delegated_ata_from_eata(
                         ata_account.account_shared_data(),
                         &eata_shared,
-                        &deleg,
+                        &deleg_record,
                     )
                 {
                     account_to_clone = projected_ata;
@@ -147,6 +148,7 @@ where
             pubkey: ata_pubkey,
             account: account_to_clone,
             commit_frequency_ms,
+            delegation_actions: DelegationActions::default(),
             delegated_to_other,
         });
     }

magicblock-chainlink/src/chainlink/fetch_cloner/delegation.rs

@@ -5,31 +5,60 @@ use magicblock_accounts_db::traits::AccountsBank;
 use magicblock_core::token_programs::EATA_PROGRAM_ID;
 use magicblock_metrics::metrics;
 use solana_account::ReadableAccount;
+use solana_instruction::Instruction;
+use solana_program::program_error::ProgramError;
 use solana_pubkey::Pubkey;
 use tracing::*;
 
 use super::FetchCloner;
 use crate::{
     chainlink::errors::{ChainlinkError, ChainlinkResult},
-    cloner::Cloner,
+    cloner::{Cloner, DelegationActions},
     remote_account_provider::{
         ChainPubsubClient, ChainRpcClient, MatchSlotsConfig,
         ResolvedAccountSharedData,
     },
 };
 
+/// Parses a delegation record from account data bytes.
+/// Returns the parsed DelegationRecord, or InvalidDelegationRecord error
+/// if parsing fails.
 pub(crate) fn parse_delegation_record(
     data: &[u8],
     delegation_record_pubkey: Pubkey,
-) -> ChainlinkResult<DelegationRecord> {
-    DelegationRecord::try_from_bytes_with_discriminator(data)
+) -> ChainlinkResult<(DelegationRecord, Option<DelegationActions>)> {
+    let delegation_record_size = DelegationRecord::size_with_discriminator();
+    if data.len() < delegation_record_size {
+        return Err(ChainlinkError::InvalidDelegationRecord(
+            delegation_record_pubkey,
+            ProgramError::InvalidAccountData,
+        ));
+    }
+    let record =
+        DelegationRecord::try_from_bytes_with_discriminator(
+            &data[..delegation_record_size],
+        )
         .copied()
         .map_err(|err| {
             ChainlinkError::InvalidDelegationRecord(
                 delegation_record_pubkey,
                 err,
             )
-        })
+        })?;
+
+    if data.len() <= delegation_record_size {
+        Ok((record, None))
+    } else {
+        let actions_data = &data[delegation_record_size..];
+        let actions = bincode::deserialize::<Vec<Instruction>>(actions_data)
+            .map_err(|err| {
+                ChainlinkError::InvalidDelegationActions(
+                    delegation_record_pubkey,
+                    err.to_string(),
+                )
+            })?;
+        Ok((record, Some(actions.into())))
+    }
 }
 
 pub(crate) fn apply_delegation_record_to_account<T, U, V, C>(
@@ -87,7 +116,7 @@ pub(crate) async fn fetch_and_parse_delegation_record<T, U, V, C>(
     account_pubkey: Pubkey,
     min_context_slot: u64,
     fetch_origin: metrics::AccountFetchOrigin,
-) -> Option<DelegationRecord>
+) -> Option<(DelegationRecord, Option<DelegationActions>)>
 where
     T: ChainRpcClient,
     U: ChainPubsubClient,

magicblock-chainlink/src/chainlink/fetch_cloner/mod.rs

@@ -50,7 +50,10 @@ use crate::{
         account_still_undelegating_on_chain::account_still_undelegating_on_chain,
         blacklisted_accounts::blacklisted_accounts,
     },
-    cloner::{errors::ClonerResult, AccountCloneRequest, Cloner},
+    cloner::{
+        errors::ClonerResult, AccountCloneRequest, Cloner,
+        DelegationActions,
+    },
     remote_account_provider::{
         program_account::get_loaderv3_get_program_data_address,
         ChainPubsubClient, ChainRpcClient, ForwardedSubscriptionUpdate,
@@ -167,7 +170,7 @@ where
                 // record fetches. This allows multiple updates to be processed in parallel.
                 let this = Arc::clone(&self);
                 tokio::spawn(async move {
-                    let (resolved_account, deleg_record) =
+                    let (resolved_account, deleg_record, delegation_actions) =
                     this.resolve_account_to_clone_from_forwarded_sub_with_unsubscribe(update)
                     .await;
                     if let Some(account) = resolved_account {
@@ -277,6 +280,7 @@ where
                                 pubkey,
                                 account,
                                 commit_frequency_ms: None,
+                                delegation_actions,
                                 delegated_to_other,
                             })
                             .await
@@ -332,7 +336,11 @@ where
     async fn resolve_account_to_clone_from_forwarded_sub_with_unsubscribe(
         &self,
         update: ForwardedSubscriptionUpdate,
-    ) -> (Option<AccountSharedData>, Option<DelegationRecord>) {
+    ) -> (
+        Option<AccountSharedData>,
+        Option<DelegationRecord>,
+        DelegationActions,
+    ) {
         let ForwardedSubscriptionUpdate { pubkey, account } = update;
         let owned_by_delegation_program =
             account.is_owned_by_delegation_program();
@@ -376,8 +384,8 @@ where
                         let account = if let Some(delegation_record) =
                             delegation_record
                         {
-                            let delegation_record =
-                                match Self::parse_delegation_record(
+                            let delegation_record_with_actions =
+                                match delegation::parse_delegation_record(
                                     delegation_record.data(),
                                     delegation_record_pubkey,
                                 ) {
@@ -394,7 +402,11 @@ where
 
                             // If the delegation record is valid we set the owner and delegation
                             // status on the account
-                            if let Some(delegation_record) = delegation_record {
+                            if let Some((
+                                delegation_record,
+                                delegation_actions,
+                            )) = delegation_record_with_actions
+                            {
                                 if tracing::enabled!(tracing::Level::TRACE) {
                                     let delegation_record_display =
                                         format!("{:?}", delegation_record);
@@ -438,17 +450,26 @@ where
                                 (
                                     Some(account.into_account_shared_data()),
                                     Some(delegation_record),
+                                    delegation_actions.unwrap_or_default(),
                                 )
                             } else {
                                 // If the delegation record is invalid we cannot clone the account
                                 // since something is corrupt and we wouldn't know what owner to
                                 // use, etc.
-                                (None, None)
+                                (
+                                    None,
+                                    None,
+                                    DelegationActions::default(),
+                                )
                             }
                         } else {
                             // If no delegation record exists we must assume the account itself is
                             // a delegation record or metadata
-                            (Some(account.into_account_shared_data()), None)
+                            (
+                                Some(account.into_account_shared_data()),
+                                None,
+                                DelegationActions::default(),
+                            )
                         };
 
                         if !subs_to_remove.is_empty() {
@@ -467,28 +488,32 @@ where
                             error = %err,
                             "Failed to fetch delegation record"
                         );
-                        (None, None)
+                        (None, None, DelegationActions::default())
                     }
                     Err(err) => {
                         error!(
                             pubkey = %pubkey,
                             error = %err,
                             "Failed to fetch delegation record"
                         );
-                        (None, None)
+                        (None, None, DelegationActions::default())
                     }
                 }
             } else {
                 let (account, deleg_record) = self
                     .maybe_project_ata_from_subscription_update(pubkey, account)
                     .await;
-                (Some(account), deleg_record)
+                (
+                    Some(account),
+                    deleg_record,
+                    DelegationActions::default(),
+                )
             }
         } else {
             // This should not happen since we call this method with sub updates which always hold
             // a fresh remote account
             error!(pubkey = %pubkey, account = ?account, "BUG: Received subscription update without fresh account");
-            (None, None)
+            (None, None, DelegationActions::default())
         }
     }
 
@@ -547,6 +572,7 @@ where
             pubkey: ata_pubkey,
             account: projected_ata,
             commit_frequency_ms: None,
+            delegation_actions: DelegationActions::default(),
             delegated_to_other: None,
         })
     }
@@ -614,6 +640,7 @@ where
         let Some(deleg_record) = deleg_record else {
             return (ata_account, None);
         };
+        let (deleg_record, _delegation_actions) = deleg_record;
 
         if let Some(projected_ata) = self.maybe_project_delegated_ata_from_eata(
             &ata_account,
@@ -655,7 +682,7 @@ where
     fn parse_delegation_record(
         data: &[u8],
         delegation_record_pubkey: Pubkey,
-    ) -> ChainlinkResult<DelegationRecord> {
+    ) -> ChainlinkResult<(DelegationRecord, Option<DelegationActions>)> {
         delegation::parse_delegation_record(data, delegation_record_pubkey)
     }
 
@@ -691,7 +718,7 @@ where
         account_pubkey: Pubkey,
         min_context_slot: u64,
         fetch_origin: metrics::AccountFetchOrigin,
-    ) -> Option<DelegationRecord> {
+    ) -> Option<(DelegationRecord, Option<DelegationActions>)> {
         delegation::fetch_and_parse_delegation_record(
             self,
             account_pubkey,
@@ -936,9 +963,10 @@ where
             }
 
             let delegated_on_chain = deleg_record.as_ref().is_some_and(|dr| {
-                dr.authority.eq(&self.validator_pubkey)
-                    || dr.authority.eq(&Pubkey::default())
+                dr.0.authority.eq(&self.validator_pubkey)
+                    || dr.0.authority.eq(&Pubkey::default())
             });
+            let deleg_record = deleg_record.map(|el| el.0);
             if !account_still_undelegating_on_chain(
                 pubkey,
                 delegated_on_chain,
@@ -1367,6 +1395,7 @@ where
                 pubkey,
                 account,
                 commit_frequency_ms: None,
+                delegation_actions: DelegationActions::default(),
                 delegated_to_other: None,
             })
             .await?;