diff --git a/Cargo.lock b/Cargo.lock index 39391b257..6b847a18b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3813,7 +3813,7 @@ dependencies = [ [[package]] name = "magicblock-delegation-program-api" version = "3.1.0" -source = "git+https://github.com/magicblock-labs/delegation-program.git?tag=v3.1.0#27f7fd8178630b88e2be45d92b174b8f2ef4661e" +source = "git+https://github.com/magicblock-labs/delegation-program.git?rev=6365d3c2e#6365d3c2e2fa9d9dc0eb2f6d770ab78e855d4064" dependencies = [ "bincode", "borsh 0.10.4", @@ -3839,7 +3839,7 @@ dependencies = [ "solana-system-interface 2.0.0", "static_assertions", "strum 0.28.0", - "thiserror 2.0.18", + "thiserror 1.0.69", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 2ce194650..54ff4188c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -114,7 +114,7 @@ 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-api = { git = "https://github.com/magicblock-labs/delegation-program.git", tag = "v3.1.0" } +magicblock-delegation-program-api = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "6365d3c2e" } magicblock-ledger = { path = "./magicblock-ledger" } magicblock-magic-program-api = { path = "./magicblock-magic-program-api" } magicblock-metrics = { path = "./magicblock-metrics" } diff --git a/magicblock-committor-service/src/intent_executor/error.rs b/magicblock-committor-service/src/intent_executor/error.rs index f428d3b2b..2f2b6e0fb 100644 --- a/magicblock-committor-service/src/intent_executor/error.rs +++ b/magicblock-committor-service/src/intent_executor/error.rs @@ -4,6 +4,7 @@ use magicblock_rpc_client::{ utils::TransactionErrorMapper, MagicBlockRpcClientError, }; use solana_instruction::error::InstructionError; +use solana_pubkey::Pubkey; use solana_rpc_client_api::{ client_error::{Error as RpcClientError, ErrorKind as RpcClientErrorKind}, request::{RpcError, RpcRequest}, @@ -77,6 +78,23 @@ impl InternalError { } } +#[derive(thiserror::Error, Debug)] +pub enum FailedToRecoverError { + #[error("FailedToFetchAccountError: {0}")] + FailedToFetchAccountError(#[source] InternalError), + #[error("MissingCommitStateAccount for delegated account {0}")] + MissingCommitStateAccount(Pubkey), +} + +impl FailedToRecoverError { + pub fn is_transient(&self) -> bool { + match self { + Self::FailedToFetchAccountError(err) => err.is_transient(), + Self::MissingCommitStateAccount(_) => false, + } + } +} + #[derive(thiserror::Error, Debug)] pub enum IntentExecutorError { #[error("EmptyIntentError")] @@ -105,6 +123,8 @@ pub enum IntentExecutorError { FailedCommitPreparationError(#[source] TransactionPreparatorError), #[error("FailedFinalizePreparationError: {0}")] FailedFinalizePreparationError(#[source] TransactionPreparatorError), + #[error("FailedToRecoverError: {0}")] + FailedToRecoverError(#[from] FailedToRecoverError), } impl IntentExecutorError { @@ -149,6 +169,7 @@ impl IntentExecutorError { Self::FailedToFinalizeError { .. } | Self::FailedFinalizePreparationError(_) => false, Self::FailedCommitPreparationError(err) => err.is_transient(), + Self::FailedToRecoverError(err) => err.is_transient(), } } @@ -171,7 +192,8 @@ impl IntentExecutorError { } => commit_signature.map(|el| (el, *finalize_signature)), IntentExecutorError::EmptyIntentError | IntentExecutorError::FailedToFitError - | IntentExecutorError::SignerError(_) => None, + | IntentExecutorError::SignerError(_) + | IntentExecutorError::FailedToRecoverError(_) => None, } } } diff --git a/magicblock-committor-service/src/intent_executor/intent_execution_client.rs b/magicblock-committor-service/src/intent_executor/intent_execution_client.rs index 903fc5564..735752862 100644 --- a/magicblock-committor-service/src/intent_executor/intent_execution_client.rs +++ b/magicblock-committor-service/src/intent_executor/intent_execution_client.rs @@ -6,11 +6,14 @@ use magicblock_rpc_client::{ decide_rpc_error_flow, map_magicblock_client_error, send_transaction_with_retries, SendErrorMapper, TransactionErrorMapper, }, - MagicBlockRpcClientError, MagicBlockSendTransactionConfig, - MagicBlockSendTransactionOutcome, MagicblockRpcClient, + MagicBlockRpcClientError, MagicBlockRpcClientResult, + MagicBlockSendTransactionConfig, MagicBlockSendTransactionOutcome, + MagicblockRpcClient, }; +use solana_account::Account; use solana_keypair::Keypair; use solana_message::VersionedMessage; +use solana_pubkey::Pubkey; use solana_rpc_client_api::config::RpcTransactionConfig; use solana_signature::Signature; use solana_transaction::versioned::VersionedTransaction; @@ -211,4 +214,11 @@ impl IntentExecutionClient { _ => {} } } + + pub(in crate::intent_executor) async fn get_account( + &self, + pubkey: &Pubkey, + ) -> MagicBlockRpcClientResult> { + self.rpc_client.get_account(pubkey).await + } } diff --git a/magicblock-committor-service/src/intent_executor/mod.rs b/magicblock-committor-service/src/intent_executor/mod.rs index a43f3cdda..7f2b1b22c 100644 --- a/magicblock-committor-service/src/intent_executor/mod.rs +++ b/magicblock-committor-service/src/intent_executor/mod.rs @@ -546,6 +546,12 @@ where Err(IntentExecutorError::SignerError(_)) => { Some(CommitStatus::Failed) } + Err(IntentExecutorError::FailedToRecoverError(_)) => { + // Occurs while preparing a finalize retry, before any + // transaction is sent — same treatment as + // FailedFinalizePreparationError. + None + } }; if let Some(update_status) = update_status { @@ -685,6 +691,7 @@ mod tests { fn test_requires_uniqueness_nonce_on_first_commit_only() { let finalize = BaseTaskImpl::Finalize(FinalizeTask { delegated_account: Pubkey::new_unique(), + state_size: 0, }); assert!(requires_uniqueness_nonce(&[commit_task(1)])); diff --git a/magicblock-committor-service/src/intent_executor/single_stage_executor.rs b/magicblock-committor-service/src/intent_executor/single_stage_executor.rs index 6a11ba6a6..91b743182 100644 --- a/magicblock-committor-service/src/intent_executor/single_stage_executor.rs +++ b/magicblock-committor-service/src/intent_executor/single_stage_executor.rs @@ -16,8 +16,9 @@ use crate::{ intent_execution_client::IntentExecutionClient, task_info_fetcher::{CacheTaskInfoFetcher, TaskInfoFetcher}, utils::{ - handle_actions_result, handle_commit_id_error, - handle_undelegation_error, prepare_and_execute_strategy, + fetch_finalize_state_size, handle_actions_result, + handle_commit_id_error, handle_undelegation_error, + prepare_and_execute_strategy, }, IntentExecutionReport, }, @@ -279,8 +280,15 @@ where delegated_account: Pubkey, transaction_preparator: &T, ) -> IntentExecutorResult> { - let finalize_task: BaseTaskImpl = - FinalizeTask { delegated_account }.into(); + // Fetch commit_state to size prealloca + let state_size = + fetch_finalize_state_size(&self.intent_client, delegated_account) + .await?; + let finalize_task: BaseTaskImpl = FinalizeTask { + delegated_account, + state_size, + } + .into(); prepare_and_execute_strategy( &self.intent_client, &self.authority, diff --git a/magicblock-committor-service/src/intent_executor/two_stage_executor.rs b/magicblock-committor-service/src/intent_executor/two_stage_executor.rs index d43499e7a..75844b2f6 100644 --- a/magicblock-committor-service/src/intent_executor/two_stage_executor.rs +++ b/magicblock-committor-service/src/intent_executor/two_stage_executor.rs @@ -17,8 +17,9 @@ use crate::{ task_info_fetcher::{CacheTaskInfoFetcher, TaskInfoFetcher}, two_stage_executor::sealed::Sealed, utils::{ - handle_actions_result, handle_commit_id_error, - handle_undelegation_error, prepare_and_execute_strategy, + fetch_finalize_state_size, handle_actions_result, + handle_commit_id_error, handle_undelegation_error, + prepare_and_execute_strategy, }, IntentExecutionReport, }, @@ -300,8 +301,15 @@ where delegated_account: Pubkey, transaction_preparator: &T, ) -> IntentExecutorResult> { - let finalize_task: BaseTaskImpl = - FinalizeTask { delegated_account }.into(); + let state_size = + fetch_finalize_state_size(&self.intent_client, delegated_account) + .await?; + + let finalize_task: BaseTaskImpl = FinalizeTask { + delegated_account, + state_size, + } + .into(); prepare_and_execute_strategy( &self.intent_client, &self.authority, diff --git a/magicblock-committor-service/src/intent_executor/utils.rs b/magicblock-committor-service/src/intent_executor/utils.rs index 190a738ae..ad8b2e157 100644 --- a/magicblock-committor-service/src/intent_executor/utils.rs +++ b/magicblock-committor-service/src/intent_executor/utils.rs @@ -12,7 +12,10 @@ use tracing::info; use crate::{ intent_executor::{ - error::{IntentExecutorResult, TransactionStrategyExecutionError}, + error::{ + FailedToRecoverError, IntentExecutorResult, + TransactionStrategyExecutionError, + }, intent_execution_client::IntentExecutionClient, single_stage_executor::SingleStageExecutor, task_info_fetcher::{CacheTaskInfoFetcher, ResetType, TaskInfoFetcher}, @@ -433,6 +436,31 @@ where } } +/// Fetches the committed state's size for a delegated account whose commit +/// already landed (i.e. we're only retrying finalize), by reading its +/// commit_state PDA — which is always sized to the full committed state, +/// diff-delivered or not. Used to size finalize's account preallocation +/// correctly instead of guessing. +pub async fn fetch_finalize_state_size( + intent_client: &IntentExecutionClient, + delegated_account: Pubkey, +) -> Result { + let commit_state_pda = + dlp_api::pda::commit_state_pda_from_delegated_account( + &delegated_account, + ); + let commit_state_account = intent_client + .get_account(&commit_state_pda) + .await + .map_err(|err| { + FailedToRecoverError::FailedToFetchAccountError(err.into()) + })? + .ok_or(FailedToRecoverError::MissingCommitStateAccount( + delegated_account, + ))?; + Ok(commit_state_account.data.len()) +} + #[cfg(test)] mod tests { use std::collections::HashMap; diff --git a/magicblock-committor-service/src/tasks/intent_size_validator.rs b/magicblock-committor-service/src/tasks/intent_size_validator.rs index 9cb4a405f..fdd8a4ead 100644 --- a/magicblock-committor-service/src/tasks/intent_size_validator.rs +++ b/magicblock-committor-service/src/tasks/intent_size_validator.rs @@ -84,6 +84,7 @@ impl IntentSizeValidator { fn finalize_task(account: &CommittedAccount) -> BaseTaskImpl { FinalizeTask { delegated_account: account.pubkey, + state_size: account.account.data.len(), } .into() } @@ -96,6 +97,7 @@ impl IntentSizeValidator { rent_reimbursement: Pubkey::default(), // We lack context here so let's assume worst case scenario include_undelegation_request: true, + state_size: account.account.data.len(), } .into() } diff --git a/magicblock-committor-service/src/tasks/mod.rs b/magicblock-committor-service/src/tasks/mod.rs index 5a4be8689..78f74d9fa 100644 --- a/magicblock-committor-service/src/tasks/mod.rs +++ b/magicblock-committor-service/src/tasks/mod.rs @@ -5,9 +5,9 @@ use solana_instruction::{AccountMeta, Instruction}; use solana_pubkey::Pubkey; pub mod commit_finalize_task; -pub mod commit_stage_task; pub mod commit_task; pub mod intent_size_validator; +pub mod preparation_task; pub mod task_builder; pub mod task_strategist; pub mod utils; @@ -174,6 +174,8 @@ pub struct UndelegateTask { pub owner_program: Pubkey, pub rent_reimbursement: Pubkey, pub include_undelegation_request: bool, + /// Size of state to bFe processed + pub state_size: usize, } impl UndelegateTask { @@ -205,6 +207,8 @@ impl From for BaseTaskImpl { #[derive(Clone, Debug)] pub struct FinalizeTask { pub delegated_account: Pubkey, + /// Size of state to be processed + pub state_size: usize, } impl FinalizeTask { @@ -404,8 +408,8 @@ mod serialization_safety_test { use crate::{ tasks::{ - commit_stage_task::PreparationTask, commit_task::{CommitDelivery, CommitTask}, + preparation_task::BufferPreparationTask, *, }, test_utils, @@ -452,6 +456,7 @@ mod serialization_safety_test { // Test Finalize variant let finalize_task: BaseTaskImpl = FinalizeTask { delegated_account: Pubkey::new_unique(), + state_size: 101, } .into(); assert_serializable(&finalize_task.instruction(&validator)); @@ -462,6 +467,7 @@ mod serialization_safety_test { owner_program: Pubkey::new_unique(), rent_reimbursement: Pubkey::new_unique(), include_undelegation_request: false, + state_size: 101, } .into(); assert_serializable(&undelegate_task.instruction(&validator)); @@ -522,6 +528,7 @@ mod serialization_safety_test { owner_program: Pubkey::new_unique(), rent_reimbursement: Pubkey::new_unique(), include_undelegation_request: true, + state_size: 101, } .instruction(&Pubkey::new_unique()); @@ -568,7 +575,7 @@ mod serialization_safety_test { make_buffer_commit_task(789, true, vec![0; 1024], 3000); let Some(preparation_task) = - PreparationTask::from_commit(&mut commit_task) + BufferPreparationTask::from_commit(&mut commit_task) else { panic!("invalid preparation state on creation!"); }; @@ -597,7 +604,7 @@ fn test_close_buffer_limit() { use tracing::info; use crate::{ - tasks::{commit_stage_task::CleanupTask, utils::TransactionUtils}, + tasks::{preparation_task::CleanupTask, utils::TransactionUtils}, test_utils, transactions::{ serialized_transaction_size, MAX_TRANSACTION_WIRE_SIZE, diff --git a/magicblock-committor-service/src/tasks/commit_stage_task.rs b/magicblock-committor-service/src/tasks/preparation_task.rs similarity index 73% rename from magicblock-committor-service/src/tasks/commit_stage_task.rs rename to magicblock-committor-service/src/tasks/preparation_task.rs index 65da85987..dc8b54fdc 100644 --- a/magicblock-committor-service/src/tasks/commit_stage_task.rs +++ b/magicblock-committor-service/src/tasks/preparation_task.rs @@ -1,4 +1,6 @@ -use dlp_api::diff::compute_diff; +use std::marker::PhantomData; + +use dlp_api::{args::PreallocateBufferKind, diff::compute_diff}; use magicblock_committor_program::{ instruction_builder::{ close_buffer::{create_close_ix, CreateCloseIxArgs}, @@ -12,17 +14,19 @@ use magicblock_committor_program::{ }; use magicblock_program::Pubkey; use solana_instruction::Instruction; +use solana_program::account_info::MAX_PERMITTED_DATA_INCREASE; use crate::{ consts::MAX_WRITE_CHUNK_SIZE, tasks::{ commit_finalize_task::CommitFinalizeTask, commit_task::{CommitDelivery, CommitTask}, + FinalizeTask, UndelegateTask, }, }; #[derive(Debug)] -pub struct PreparationTask<'a> { +pub struct BufferPreparationTask<'a> { pub commit_id: u64, pub pubkey: Pubkey, pub chunks: Chunks, @@ -30,7 +34,7 @@ pub struct PreparationTask<'a> { prepared: &'a mut bool, } -impl<'a> PreparationTask<'a> { +impl<'a> BufferPreparationTask<'a> { pub fn from_commit(task: &'a mut CommitTask) -> Option { match &mut task.delivery_details { CommitDelivery::StateInArgs | CommitDelivery::DiffInArgs { .. } => { @@ -294,6 +298,116 @@ impl CleanupTask { } } +pub struct Preallocate { + delegated_account: Pubkey, + current_size: u32, + target_size: u32, + _phantom: PhantomData, +} + +pub trait Preallocatable { + const KIND: PreallocateBufferKind; +} + +impl Preallocatable for CommitTask { + const KIND: PreallocateBufferKind = PreallocateBufferKind::CommitState; +} + +impl Preallocatable for CommitFinalizeTask { + const KIND: PreallocateBufferKind = PreallocateBufferKind::DelegatedAccount; +} + +impl Preallocatable for FinalizeTask { + const KIND: PreallocateBufferKind = PreallocateBufferKind::DelegatedAccount; +} + +impl Preallocatable for UndelegateTask { + const KIND: PreallocateBufferKind = PreallocateBufferKind::UndelegateBuffer; +} + +impl Preallocate { + pub fn new( + delegated_account: Pubkey, + current_size: u32, + target_size: u32, + ) -> Option { + (target_size > current_size + MAX_PERMITTED_DATA_INCREASE as u32) + .then_some(Self { + delegated_account, + current_size, + target_size, + _phantom: PhantomData, + }) + } + + pub fn instructions(&self, validator: &Pubkey) -> Vec { + dlp_api::instruction_builder::preallocate_buffer_chunks( + *validator, + self.delegated_account, + T::KIND, + self.current_size, + self.target_size, + ) + } +} + +impl Preallocate { + pub fn from_commit(task: &CommitTask) -> Option { + let delegated_account = task.committed_account.pubkey; + match &task.delivery_details { + // We assume that tx can't carry over 10kb in arguments + CommitDelivery::StateInArgs | CommitDelivery::DiffInArgs { .. } => { + None + } + CommitDelivery::StateInBuffer { .. } + // commit_state is always written with the full reconstructed + // account state, so the target size is the same regardless of + // whether delivery is diff or state based. see dlp merge_diff_copy + | CommitDelivery::DiffInBuffer { .. } => { + let target_size = task.committed_account.account.data.len() as u32; + Self::new(delegated_account, 0, target_size) + } + } + } +} + +impl Preallocate { + pub fn from_commit_finalize(task: &CommitFinalizeTask) -> Option { + let delegated_account = task.committed_account.pubkey; + match &task.delivery { + // We assume that tx can't carry over 10kb in arguments + CommitDelivery::StateInArgs | CommitDelivery::DiffInArgs { .. } => { + None + } + CommitDelivery::StateInBuffer { .. } => { + let target_size = + task.committed_account.account.data.len() as u32; + Self::new(delegated_account, 0, target_size) + } + CommitDelivery::DiffInBuffer { base_account, .. } => { + // This increases delegated account, so we care about delta and not resulting size + let current_size = base_account.data.len() as u32; + let target_size = + task.committed_account.account.data.len() as u32; + Self::new(delegated_account, current_size, target_size) + } + } + } +} + +impl Preallocate { + pub fn from_finalize_task(task: &FinalizeTask) -> Option { + Self::new(task.delegated_account, 0, task.state_size as u32) + } +} + +// TODO: use once dlp side support large undelegations +impl Preallocate { + pub fn from_undelegate_task(task: &UndelegateTask) -> Option { + Self::new(task.delegated_account, 0, task.state_size as u32) + } +} + #[cfg(test)] mod tests { use solana_hash::Hash; @@ -321,7 +435,7 @@ mod tests { let authority = Keypair::new(); let data = vec![0; MAX_WRITE_CHUNK_SIZE as usize]; let mut prepared = false; - let preparation_task = PreparationTask { + let preparation_task = BufferPreparationTask { commit_id: 1, pubkey: Pubkey::new_unique(), chunks: Chunks::from_data_length(data.len(), MAX_WRITE_CHUNK_SIZE), diff --git a/magicblock-committor-service/src/tasks/task_builder.rs b/magicblock-committor-service/src/tasks/task_builder.rs index f39570c91..89fd68fba 100644 --- a/magicblock-committor-service/src/tasks/task_builder.rs +++ b/magicblock-committor-service/src/tasks/task_builder.rs @@ -6,7 +6,7 @@ use magicblock_core::intent::{ types::CommittedAccount, CommitAndUndelegate, CommitType, UndelegateType, }; use magicblock_program::magic_scheduled_base_intent::ScheduledIntentBundle; -use solana_account::Account; +use solana_account::{Account, ReadableAccount}; use solana_pubkey::Pubkey; use solana_signature::Signature; use tracing::error; @@ -49,6 +49,12 @@ pub struct CommitStageTaskInfo { base_accounts: HashMap, } +/// Necessary info for finalize stage task creation +pub struct FinalizeStageTaskInfo { + /// Delegation metadata for a given address + delegation_metadata: HashMap, +} + /// Task builder pub struct TaskBuilderImpl; @@ -68,6 +74,7 @@ impl TaskBuilderImpl { .await } + /// Returns base state for account that use diff async fn fetch_diffable_accounts( task_info_fetcher: &Arc, accounts: &[CommittedAccount], @@ -147,6 +154,30 @@ impl TaskBuilderImpl { base_accounts, }) } + + async fn fetch_finalize_stage_info( + intent_bundle: &ScheduledIntentBundle, + task_info_fetcher: &Arc, + ) -> TaskBuilderResult { + // Fetch necessary data for BaseTasks creation + let all_committed_accounts = intent_bundle.get_all_committed_accounts(); + let all_committed_pubkeys = + Self::get_finalize_stage_metadata_pubkeys(intent_bundle); + // Get commit nonces and base accounts + let min_context_slot = all_committed_accounts + .iter() + .map(|account| account.remote_slot) + .max() + .unwrap_or(0); + let delegation_metadata = task_info_fetcher + .fetch_delegation_metadata(&all_committed_pubkeys, min_context_slot) + .await + .map_err(TaskBuilderError::FinalizedTasksBuildError)?; + + Ok(FinalizeStageTaskInfo { + delegation_metadata, + }) + } } #[async_trait] @@ -228,6 +259,7 @@ impl TasksBuilder for TaskBuilderImpl { fn finalize_task(account: &CommittedAccount) -> BaseTaskImpl { FinalizeTask { delegated_account: account.pubkey, + state_size: account.account.data().len(), } .into() } @@ -243,6 +275,7 @@ impl TasksBuilder for TaskBuilderImpl { owner_program: account.account.owner, rent_reimbursement: *rent_reimbursement, include_undelegation_request, + state_size: account.account.data().len(), } .into() } @@ -297,23 +330,12 @@ impl TasksBuilder for TaskBuilderImpl { Ok(tasks) } - let mut tasks = Vec::new(); - let finalize_metadata_pubkeys = - Self::get_finalize_stage_metadata_pubkeys(intent_bundle); - let min_context_slot = intent_bundle - .get_all_committed_accounts() - .iter() - .map(|account| account.remote_slot) - .max() - .unwrap_or(0); - let delegation_metadata = info_fetcher - .fetch_delegation_metadata( - &finalize_metadata_pubkeys, - min_context_slot, - ) - .await - .map_err(TaskBuilderError::FinalizedTasksBuildError)?; + let FinalizeStageTaskInfo { + delegation_metadata, + } = Self::fetch_finalize_stage_info(intent_bundle, info_fetcher) + .await?; + let mut tasks = Vec::new(); if let Some(ref value) = intent_bundle.intent_bundle.commit { tasks.extend(create_finalize_tasks(value)); } diff --git a/magicblock-committor-service/src/tasks/task_strategist.rs b/magicblock-committor-service/src/tasks/task_strategist.rs index 883c242ce..a8b237d9e 100644 --- a/magicblock-committor-service/src/tasks/task_strategist.rs +++ b/magicblock-committor-service/src/tasks/task_strategist.rs @@ -638,6 +638,7 @@ mod tests { fn create_test_finalize_task() -> FinalizeTask { FinalizeTask { delegated_account: Pubkey::new_unique(), + state_size: 0, } } @@ -648,6 +649,7 @@ mod tests { owner_program: Pubkey::default(), rent_reimbursement: Pubkey::new_unique(), include_undelegation_request: false, + state_size: 0, } } diff --git a/magicblock-committor-service/src/transaction_preparator/delivery_preparator.rs b/magicblock-committor-service/src/transaction_preparator/delivery_preparator.rs index e5868f627..60c42a5b4 100644 --- a/magicblock-committor-service/src/transaction_preparator/delivery_preparator.rs +++ b/magicblock-committor-service/src/transaction_preparator/delivery_preparator.rs @@ -30,7 +30,7 @@ use tracing::{error, info}; use crate::{ persist::{CommitStatus, IntentPersister}, tasks::{ - commit_stage_task::{CleanupTask, PreparationTask}, + preparation_task::{BufferPreparationTask, CleanupTask, Preallocate}, task_strategist::TransactionStrategy, utils::TransactionUtils, BaseTaskImpl, @@ -45,6 +45,11 @@ pub struct DeliveryPreparator { compute_budget_config: ComputeBudgetConfig, } +/// Preallocate instructions are packed this many per transaction. +/// TX wire ceiling is 36, but due yo MaxInstructionTraceLengthExceeded +/// ceiling is 31 +const PREALLOCATE_CHUNK_SIZE: usize = 31; + const MAX_PARALLEL_BUFFER_SENDS: usize = 8; impl DeliveryPreparator { @@ -68,41 +73,157 @@ impl DeliveryPreparator { persister: &Option

, ) -> DeliveryPreparatorResult> { let uniqueness_nonce = strategy.uniqueness_nonce; - let preparation_futures = - strategy.optimized_tasks.iter_mut().map(|task| async move { - let _timer = - metrics::observe_committor_intent_task_preparation_time( - &*task, - ); - self.prepare_task_handling_errors( - authority, - task, - persister, - uniqueness_nonce, - ) - .await - }); - let task_preparations = join_all(preparation_futures); - let alts_preparations = async { + let tasks_preparation = self.prepare_tasks( + authority, + &mut strategy.optimized_tasks, + persister, + uniqueness_nonce, + ); + let alts_preparation = async { let _timer = metrics::observe_committor_intent_alt_preparation_time(); self.prepare_lookup_tables(authority, &strategy.lookup_tables_keys) .await + .map_err(DeliveryPreparatorError::FailedToCreateALTError) }; - let (res1, res2) = join(task_preparations, alts_preparations).await; + let (res1, res2) = join(tasks_preparation, alts_preparation).await; + res1?; + res2 + } + + /// Prepares buffer content (writes) and account/buffer preallocation for + /// all tasks concurrently, since the two touch independent on-chain + /// state. + async fn prepare_tasks( + &self, + authority: &Keypair, + tasks: &mut [BaseTaskImpl], + persister: &Option

, + uniqueness_nonce: Option, + ) -> DeliveryPreparatorResult<()> { + // Preallocate futures + let mut preallocate_chunks = + Self::chunk_preallocate_instructions(tasks, &authority.pubkey()); + let preallocate_preparations = self.send_preallocate_chunks( + authority, + &mut preallocate_chunks, + uniqueness_nonce, + ); + + // Buffer preparation futures + let buffer_preparation_futs = tasks.iter_mut().map(|task| async move { + let _timer = + metrics::observe_committor_intent_task_preparation_time(&*task); + self.prepare_buffer_handling_errors( + authority, + task, + persister, + uniqueness_nonce, + ) + .await + }); + let task_preparations = join_all(buffer_preparation_futs); + + // Run both in parallel + let (res1, res2) = + join(task_preparations, preallocate_preparations).await; res1.into_iter() .collect::, _>>() .map_err(DeliveryPreparatorError::FailedToPrepareBufferAccounts)?; - let lookup_tables = - res2.map_err(DeliveryPreparatorError::FailedToCreateALTError)?; + res2.map_err(DeliveryPreparatorError::FailedToPreallocateError)?; + + Ok(()) + } - Ok(lookup_tables) + /// Preallocates accounts for tasks which may require it, such as: + /// Commit, CommitFinalize, Finalize, Undelegate + /// IXs can be executed in any order so for fee optimization we group them + pub async fn preallocate( + &self, + authority: &Keypair, + tasks: &[BaseTaskImpl], + uniqueness_nonce: Option, + ) -> DeliveryPreparatorResult<()> { + let mut chunked_ixs = + Self::chunk_preallocate_instructions(tasks, &authority.pubkey()); + self.send_preallocate_chunks( + authority, + &mut chunked_ixs, + uniqueness_nonce, + ) + .await + .map_err(DeliveryPreparatorError::FailedToPreallocateError) + } + + /// Derives the chunked PreallocateBuffer instructions for whichever + /// tasks need it. Synchronous, owned output — only needs a shared + /// borrow of `tasks`, so it can run before anything takes `&mut` + /// access to them. + fn chunk_preallocate_instructions( + tasks: &[BaseTaskImpl], + authority: &Pubkey, + ) -> Vec> { + // Iterator with preallocate instructions + let mut preallocate_instructions = tasks + .iter() + .filter_map(|task| match task { + BaseTaskImpl::Commit(value) => Preallocate::from_commit(value) + .map(|el| el.instructions(authority)), + BaseTaskImpl::CommitFinalize(value) => { + Preallocate::from_commit_finalize(value) + .map(|el| el.instructions(authority)) + } + BaseTaskImpl::Finalize(value) => { + Preallocate::from_finalize_task(value) + .map(|el| el.instructions(authority)) + } + // TODO: not supported by dlp. use Preallocate once enabled + BaseTaskImpl::Undelegate(_) => None, + _ => None, + }) + .flatten(); + + // Get ix chunks that optimally fit into tx to save mOney + let mut chunked_ixs = Vec::new(); + loop { + let chunk: Vec<_> = preallocate_instructions + .by_ref() + .take(PREALLOCATE_CHUNK_SIZE) + .collect(); + if chunk.is_empty() { + break; + } + chunked_ixs.push(chunk); + } + chunked_ixs + } + + /// Sends already-derived preallocate instruction chunks. Takes owned + /// instruction data, not the originating tasks — never conflicts with + /// `&mut` access to those tasks elsewhere. + async fn send_preallocate_chunks( + &self, + authority: &Keypair, + chunked_ixs: &mut [Vec], + uniqueness_nonce: Option, + ) -> Result<(), TransactionSendError> { + let iter = chunked_ixs.iter_mut().map(|el| { + self.send_ixs_with_retry( + el, + authority, + 3, + uniqueness_nonce, + PreallocateErrorMapper, + ) + }); + + join_all(iter).await.into_iter().try_for_each(|el| el) } /// Prepares necessary parts for TX if needed, otherwise returns immediately - pub async fn prepare_task( + pub async fn prepare_buffer( &self, authority: &Keypair, task: &mut BaseTaskImpl, @@ -111,10 +232,12 @@ impl DeliveryPreparator { ) -> DeliveryPreparatorResult<(), InternalError> { let preparation_task = match task { BaseTaskImpl::Commit(commit_task) => { - PreparationTask::from_commit(commit_task) + BufferPreparationTask::from_commit(commit_task) } BaseTaskImpl::CommitFinalize(commit_finalize_task) => { - PreparationTask::from_commit_finalize(commit_finalize_task) + BufferPreparationTask::from_commit_finalize( + commit_finalize_task, + ) } _ => None, }; @@ -170,7 +293,7 @@ impl DeliveryPreparator { /// Runs `prepare_task` and, if the buffer was already initialized, /// performs cleanup and retries once. - pub async fn prepare_task_handling_errors( + pub async fn prepare_buffer_handling_errors( &self, authority: &Keypair, task: &mut BaseTaskImpl, @@ -178,7 +301,7 @@ impl DeliveryPreparator { uniqueness_nonce: Option, ) -> Result<(), InternalError> { let res = self - .prepare_task(authority, task, persister, uniqueness_nonce) + .prepare_buffer(authority, task, persister, uniqueness_nonce) .await; match res { Err(InternalError::BufferExecutionError( @@ -196,10 +319,12 @@ impl DeliveryPreparator { // Preparation failed due to buffer existing - cleanup and retry let preparation_task = match task { BaseTaskImpl::Commit(commit_task) => { - PreparationTask::from_commit(commit_task) + BufferPreparationTask::from_commit(commit_task) } BaseTaskImpl::CommitFinalize(commit_finalize_task) => { - PreparationTask::from_commit_finalize(commit_finalize_task) + BufferPreparationTask::from_commit_finalize( + commit_finalize_task, + ) } _ => None, }; @@ -214,7 +339,7 @@ impl DeliveryPreparator { self.rpc_client.invalidate_cached_blockhash().await; // Retry preparation - self.prepare_task(authority, task, persister, uniqueness_nonce) + self.prepare_buffer(authority, task, persister, uniqueness_nonce) .await } @@ -223,7 +348,7 @@ impl DeliveryPreparator { async fn initialize_buffer_account( &self, authority: &Keypair, - preparation_task: &PreparationTask<'_>, + preparation_task: &BufferPreparationTask<'_>, uniqueness_nonce: Option, ) -> DeliveryPreparatorResult<(), BufferExecutionError> { let authority_pubkey = authority.pubkey(); @@ -262,6 +387,9 @@ impl DeliveryPreparator { authority, 5, uniqueness_nonce, + BufferErrorMapper { + transaction_error_mapper: BufferTransactionErrorMapper, + }, ) .await?; @@ -275,6 +403,9 @@ impl DeliveryPreparator { authority, 5, uniqueness_nonce, + BufferErrorMapper { + transaction_error_mapper: BufferTransactionErrorMapper, + }, ) }); try_join_all(preparation_futs).await?; @@ -287,7 +418,7 @@ impl DeliveryPreparator { async fn write_buffer_with_retries( &self, authority: &Keypair, - preparation_task: &PreparationTask<'_>, + preparation_task: &BufferPreparationTask<'_>, uniqueness_nonce: Option, ) -> DeliveryPreparatorResult<(), InternalError> { let authority_pubkey = authority.pubkey(); @@ -334,6 +465,9 @@ impl DeliveryPreparator { authority, 5, uniqueness_nonce, + BufferErrorMapper { + transaction_error_mapper: BufferTransactionErrorMapper, + }, ) }); try_join_all(fut_iter).await?; @@ -343,82 +477,26 @@ impl DeliveryPreparator { } // CommitProcessor::init_accounts analog - async fn send_ixs_with_retry( + async fn send_ixs_with_retry( &self, instructions: &mut Vec, authority: &Keypair, max_attempts: usize, uniqueness_nonce: Option, - ) -> DeliveryPreparatorResult<(), BufferExecutionError> { + send_error_mapper: impl SendErrorMapper< + TransactionSendError, + ExecutionError = E, + >, + ) -> DeliveryPreparatorResult<(), E> { if let Some(nonce) = uniqueness_nonce { instructions .push(TransactionUtils::uniqueness_noop_instruction(nonce)); } - /// Error mappers - struct IntentTransactionErrorMapper; - impl TransactionErrorMapper for IntentTransactionErrorMapper { - type ExecutionError = BufferExecutionError; - fn try_map( - &self, - error: TransactionError, - signature: Option, - ) -> Result { - match error { - err @ TransactionError::InstructionError( - _, - InstructionError::AccountAlreadyInitialized, - ) => Ok( - BufferExecutionError::AccountAlreadyInitializedError( - err, signature, - ), - ), - err => Err(err), - } - } - } - - struct BufferErrorMapper { - transaction_error_mapper: TxMap, - } - impl SendErrorMapper for BufferErrorMapper - where - TxMap: - TransactionErrorMapper, - { - type ExecutionError = BufferExecutionError; - fn map(&self, error: TransactionSendError) -> Self::ExecutionError { - match error { - TransactionSendError::MagicBlockRpcClientError(err) => { - map_magicblock_client_error( - &self.transaction_error_mapper, - *err, - ) - } - err => BufferExecutionError::TransactionSendError(err), - } - } - - fn decide_flow( - &self, - err: &Self::ExecutionError, - ) -> ControlFlow<(), Duration> { - match err { - BufferExecutionError::TransactionSendError( - TransactionSendError::MagicBlockRpcClientError(err), - ) => decide_rpc_error_flow(err), - _ => ControlFlow::Break(()), - } - } - } - - let default_error_mapper = BufferErrorMapper { - transaction_error_mapper: IntentTransactionErrorMapper, - }; let attempt = || async { self.try_send_ixs(instructions, authority).await }; - send_transaction_with_retries(attempt, default_error_mapper, |i, _| { + send_transaction_with_retries(attempt, send_error_mapper, |i, _| { i >= max_attempts }) .await?; @@ -549,6 +627,9 @@ impl DeliveryPreparator { .iter() .map(|task| task.instruction(&authority.pubkey())), ); + let buffer_error_mapper = BufferErrorMapper { + transaction_error_mapper: BufferTransactionErrorMapper, + }; async move { ( @@ -558,6 +639,7 @@ impl DeliveryPreparator { authority, 1, uniqueness_nonce, + buffer_error_mapper, ) .await, ) @@ -611,6 +693,30 @@ impl TransactionSendError { } } +/// Dummy error mapper for preallocate +/// We introduce it to reuse `decide_rpc_error_flow` mechanics +/// Atm there's no need to map error to anything as we assume idempotent execution of preallocate +pub struct PreallocateErrorMapper; +impl SendErrorMapper for PreallocateErrorMapper { + type ExecutionError = TransactionSendError; + fn map(&self, error: TransactionSendError) -> Self::ExecutionError { + error + } + + fn decide_flow( + &self, + mapped_error: &Self::ExecutionError, + ) -> ControlFlow<(), Duration> { + match mapped_error { + TransactionSendError::CompileError(_) + | TransactionSendError::SignerError(_) => ControlFlow::Break(()), + TransactionSendError::MagicBlockRpcClientError(value) => { + decide_rpc_error_flow(value.as_ref()) + } + } + } +} + #[derive(thiserror::Error, Debug)] pub enum BufferExecutionError { #[error("AccountAlreadyInitializedError: {0}")] @@ -694,8 +800,66 @@ impl InternalError { } } +// Error mappers +struct BufferTransactionErrorMapper; +impl TransactionErrorMapper for BufferTransactionErrorMapper { + type ExecutionError = BufferExecutionError; + fn try_map( + &self, + error: TransactionError, + signature: Option, + ) -> Result { + match error { + err @ TransactionError::InstructionError( + _, + InstructionError::AccountAlreadyInitialized, + ) => Ok(BufferExecutionError::AccountAlreadyInitializedError( + err, signature, + )), + err => Err(err), + } + } +} + +struct BufferErrorMapper { + transaction_error_mapper: TxMap, +} +impl SendErrorMapper for BufferErrorMapper +where + TxMap: TransactionErrorMapper, +{ + type ExecutionError = BufferExecutionError; + fn map(&self, error: TransactionSendError) -> Self::ExecutionError { + match error { + TransactionSendError::MagicBlockRpcClientError(err) => { + map_magicblock_client_error( + &self.transaction_error_mapper, + *err, + ) + } + err => BufferExecutionError::TransactionSendError(err), + } + } + + fn decide_flow( + &self, + err: &Self::ExecutionError, + ) -> ControlFlow<(), Duration> { + match err { + BufferExecutionError::TransactionSendError( + TransactionSendError::MagicBlockRpcClientError(err), + ) => decide_rpc_error_flow(err), + _ => ControlFlow::Break(()), + } + } +} + #[derive(thiserror::Error, Debug)] pub enum DeliveryPreparatorError { + #[error( + "Failed to preallocate delegated account for finalize instruction: {0}" + )] + FailedToPreallocateError(#[source] TransactionSendError), #[error("FailedToPrepareBufferAccounts: {0}")] FailedToPrepareBufferAccounts(#[source] InternalError), #[error("FailedToCreateALTError: {0}")] @@ -707,6 +871,7 @@ impl DeliveryPreparatorError { match self { Self::FailedToCreateALTError(err) | Self::FailedToPrepareBufferAccounts(err) => err.signature(), + Self::FailedToPreallocateError(err) => err.signature(), } } @@ -714,9 +879,74 @@ impl DeliveryPreparatorError { match self { Self::FailedToCreateALTError(err) | Self::FailedToPrepareBufferAccounts(err) => err.is_transient(), + Self::FailedToPreallocateError(err) => err.is_transient(), } } } pub type DeliveryPreparatorResult = Result; + +#[cfg(test)] +mod tests { + use dlp_api::args::PreallocateBufferKind; + use solana_hash::Hash; + use solana_keypair::Keypair; + use solana_message::v0::Message; + use solana_signer::Signer; + use solana_transaction::versioned::VersionedTransaction; + + use super::*; + use crate::transactions::{ + serialized_transaction_size, MAX_TRANSACTION_WIRE_SIZE, + }; + + /// Justifies `PREALLOCATE_CHUNK_SIZE`: packs that many real + /// PreallocateBuffer instructions (plus a uniqueness noop, the worst + /// case send_preallocate_chunks actually produces) into one transaction + /// and asserts it still fits on the wire. All three PreallocateBufferKind + /// variants produce the same fixed-shape instruction (7 accounts + a + /// small enum tag + u32), so the choice of kind here doesn't matter. + #[test] + fn preallocate_chunk_of_max_size_fits_in_one_tx() { + let authority = Keypair::new(); + let delegated_account = Pubkey::new_unique(); + + let mut instructions: Vec = (0..PREALLOCATE_CHUNK_SIZE) + .map(|_| { + dlp_api::instruction_builder::preallocate_buffer( + authority.pubkey(), + delegated_account, + PreallocateBufferKind::CommitState, + u32::MAX, + ) + }) + .collect(); + instructions.push(TransactionUtils::uniqueness_noop_instruction(42)); + + let message = Message::try_compile( + &authority.pubkey(), + &instructions, + &[], + Hash::new_unique(), + ) + .expect("compile preallocate chunk transaction"); + let transaction = VersionedTransaction::try_new( + VersionedMessage::V0(message), + &[&authority], + ) + .expect("sign preallocate chunk transaction"); + + let transaction_size = serialized_transaction_size(&transaction); + info!( + transaction_size, + chunk_size = PREALLOCATE_CHUNK_SIZE, + "Preallocate chunk transaction size" + ); + assert!( + transaction_size <= MAX_TRANSACTION_WIRE_SIZE, + "PREALLOCATE_CHUNK_SIZE={PREALLOCATE_CHUNK_SIZE} no longer fits \ + in one tx ({transaction_size} > {MAX_TRANSACTION_WIRE_SIZE})" + ); + } +} diff --git a/magicblock-committor-service/src/transaction_preparator/mod.rs b/magicblock-committor-service/src/transaction_preparator/mod.rs index a301e27cf..4c9b0f021 100644 --- a/magicblock-committor-service/src/transaction_preparator/mod.rs +++ b/magicblock-committor-service/src/transaction_preparator/mod.rs @@ -8,7 +8,7 @@ use solana_message::VersionedMessage; use crate::{ persist::IntentPersister, tasks::{ - commit_stage_task::CleanupTask, task_strategist::TransactionStrategy, + preparation_task::CleanupTask, task_strategist::TransactionStrategy, utils::TransactionUtils, BaseTaskImpl, }, transaction_preparator::{ diff --git a/magicblock-core/src/intent/types.rs b/magicblock-core/src/intent/types.rs index 7f9a3e622..38b96f5d9 100644 --- a/magicblock-core/src/intent/types.rs +++ b/magicblock-core/src/intent/types.rs @@ -53,4 +53,8 @@ impl CommittedAccount { remote_slot, } } + + pub fn data_len(&self) -> usize { + self.account.data.len() + } } diff --git a/test-integration/Cargo.lock b/test-integration/Cargo.lock index 6fc24d2d6..d60c62429 100644 --- a/test-integration/Cargo.lock +++ b/test-integration/Cargo.lock @@ -3838,7 +3838,7 @@ dependencies = [ [[package]] name = "magicblock-delegation-program-api" version = "3.1.0" -source = "git+https://github.com/magicblock-labs/delegation-program.git?tag=v3.1.0#27f7fd8178630b88e2be45d92b174b8f2ef4661e" +source = "git+https://github.com/magicblock-labs/delegation-program.git?rev=6365d3c2e#6365d3c2e2fa9d9dc0eb2f6d770ab78e855d4064" dependencies = [ "bincode", "borsh 0.10.4", @@ -3864,7 +3864,7 @@ dependencies = [ "solana-system-interface 2.0.0", "static_assertions", "strum", - "thiserror 2.0.18", + "thiserror 1.0.69", ] [[package]] diff --git a/test-integration/Cargo.toml b/test-integration/Cargo.toml index d891eb7cc..fb06094d2 100644 --- a/test-integration/Cargo.toml +++ b/test-integration/Cargo.toml @@ -57,7 +57,7 @@ magicblock-committor-program = { path = "../magicblock-committor-program", featu magicblock-committor-service = { path = "../magicblock-committor-service" } magicblock-config = { path = "../magicblock-config" } magicblock-core = { path = "../magicblock-core" } -magicblock-delegation-program-api = { git = "https://github.com/magicblock-labs/delegation-program.git", tag = "v3.1.0", default-features = false } +magicblock-delegation-program-api = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "6365d3c2e", default-features = false } magicblock-magic-program-api = { path = "../magicblock-magic-program-api" } magicblock-program = { path = "../programs/magicblock" } magicblock-rpc-client = { path = "../magicblock-rpc-client" } diff --git a/test-integration/schedulecommit/elfs/dlp.so b/test-integration/schedulecommit/elfs/dlp.so index b8fa734fe..ea5712653 100755 Binary files a/test-integration/schedulecommit/elfs/dlp.so and b/test-integration/schedulecommit/elfs/dlp.so differ diff --git a/test-integration/test-committor-service/tests/test_delivery_preparator.rs b/test-integration/test-committor-service/tests/test_delivery_preparator.rs index 1cc7c6784..52820db52 100644 --- a/test-integration/test-committor-service/tests/test_delivery_preparator.rs +++ b/test-integration/test-committor-service/tests/test_delivery_preparator.rs @@ -3,20 +3,30 @@ use magicblock_committor_program::Chunks; use magicblock_committor_service::{ persist::IntentPersisterImpl, tasks::{ - commit_stage_task::{CleanupTask, PreparationTask}, commit_task::CommitDelivery, + preparation_task::{BufferPreparationTask, CleanupTask}, task_strategist::{TaskStrategist, TransactionStrategy}, BaseTaskImpl, }, }; -use solana_sdk::signer::Signer; +use solana_sdk::{signature::Keypair, signer::Signer}; -use crate::common::{ - create_buffer_commit_task, create_commit_task, generate_random_bytes, - TestFixture, +use crate::{ + common::{ + create_buffer_commit_task, create_commit_task, generate_random_bytes, + TestFixture, + }, + utils::{ + ensure_validator_authority, + transactions::{ + fund_validator_auth_and_ensure_validator_fees_vault, + init_and_delegate_account_on_chain, + }, + }, }; mod common; +mod utils; #[tokio::test] async fn test_prepare_10kb_buffer() { @@ -81,7 +91,15 @@ async fn test_prepare_10kb_buffer() { #[tokio::test] async fn test_prepare_multiple_buffers() { - let fixture = TestFixture::new().await; + // The 500KB task exceeds MAX_PERMITTED_DATA_INCREASE, so preparation + // also preallocates its commit_state PDA -- that requires a real, + // currently delegated account. + const DELEGATED_ACCOUNT_SIZE: u64 = 100; + + let validator_auth = ensure_validator_authority(); + let fixture = TestFixture::new_with_keypair(validator_auth).await; + fund_validator_auth_and_ensure_validator_fees_vault(&fixture.authority) + .await; let preparator = fixture.create_delivery_preparator(); let datas = [ @@ -89,10 +107,21 @@ async fn test_prepare_multiple_buffers() { generate_random_bytes(10), generate_random_bytes(500 * 1024), ]; - let buffer_tasks: Vec = datas + let mut buffer_tasks: Vec = datas .iter() .map(|data| create_buffer_commit_task(data).into()) .collect(); + + let (large_pubkey, _) = init_and_delegate_account_on_chain( + &Keypair::new(), + DELEGATED_ACCOUNT_SIZE, + None, + ) + .await; + if let BaseTaskImpl::Commit(task) = &mut buffer_tasks[2] { + task.committed_account.pubkey = large_pubkey; + } + let mut strategy = TransactionStrategy { optimized_tasks: buffer_tasks, lookup_tables_keys: vec![], @@ -353,7 +382,8 @@ async fn test_reprepare_closed_buffer_with_distinct_intent_nonce() { let data = generate_random_bytes(112); let mut commit_task = create_buffer_commit_task(&data); commit_task.reset_commit_id(1); - let Some(preparation_task) = PreparationTask::from_commit(&mut commit_task) + let Some(preparation_task) = + BufferPreparationTask::from_commit(&mut commit_task) else { panic!("expected preparation stage"); }; @@ -480,8 +510,13 @@ async fn test_reprepare_closed_buffer_with_distinct_intent_nonce() { #[tokio::test] async fn test_prepare_cleanup_and_reprepare_mixed_tasks() { use borsh::BorshDeserialize; + const DELEGATED_ACCOUNT_SIZE: u64 = 100; + + let validator_auth = ensure_validator_authority(); + let fixture = TestFixture::new_with_keypair(validator_auth).await; + fund_validator_auth_and_ensure_validator_fees_vault(&fixture.authority) + .await; - let fixture = TestFixture::new().await; let preparator = fixture.create_delivery_preparator(); // Data of committed accs @@ -489,10 +524,26 @@ async fn test_prepare_cleanup_and_reprepare_mixed_tasks() { let buf_a_data = generate_random_bytes(12 * 1024); let buf_b_data = generate_random_bytes(64 * 1024 + 3); + // For preallocation buf_a, buf_b needs to be delegated + let (buf_a_pubkey, _) = init_and_delegate_account_on_chain( + &Keypair::new(), + DELEGATED_ACCOUNT_SIZE, + None, + ) + .await; + let (buf_b_pubkey, _) = init_and_delegate_account_on_chain( + &Keypair::new(), + DELEGATED_ACCOUNT_SIZE, + None, + ) + .await; + // Keep these around to modify data later (same commit IDs, different data) let mut commit_args = create_commit_task(&args_data); let mut commit_a = create_buffer_commit_task(&buf_a_data); + commit_a.committed_account.pubkey = buf_a_pubkey; let mut commit_b = create_buffer_commit_task(&buf_b_data); + commit_b.committed_account.pubkey = buf_b_pubkey; let mut strategy = TransactionStrategy { optimized_tasks: vec![ diff --git a/test-integration/test-committor-service/tests/test_ix_commit_local.rs b/test-integration/test-committor-service/tests/test_ix_commit_local.rs index 339d2e5e3..b59d8c071 100644 --- a/test-integration/test-committor-service/tests/test_ix_commit_local.rs +++ b/test-integration/test-committor-service/tests/test_ix_commit_local.rs @@ -1,6 +1,7 @@ use std::{collections::HashMap, sync::Arc}; use borsh::to_vec; +use dlp_api::solana_program::native_token::LAMPORTS_PER_SOL; use magicblock_committor_service::{ committor_processor::CommittorProcessor, config::ChainConfig, @@ -19,7 +20,8 @@ use solana_commitment_config::CommitmentConfig; use solana_pubkey::Pubkey; use solana_rpc_client::nonblocking::rpc_client::RpcClient; use solana_sdk::{ - hash::Hash, signature::Keypair, signer::Signer, transaction::Transaction, + hash::Hash, rent::Rent, signature::Keypair, signer::Signer, + transaction::Transaction, }; use test_kit::init_logger; use tokio::task::JoinSet; @@ -67,6 +69,33 @@ fn expect_strategies( expected_strategies } +/// Wraps `accounts` into the `MagicBaseIntent` shape matching `commit_type`. +fn base_intent_for( + commit_type: CommitIntentKind, + accounts: Vec, +) -> MagicBaseIntent { + match commit_type { + CommitIntentKind::Commit => { + MagicBaseIntent::Commit(CommitType::Standalone(accounts)) + } + CommitIntentKind::CommitAndUndelegate => { + MagicBaseIntent::CommitAndUndelegate(CommitAndUndelegate { + commit_action: CommitType::Standalone(accounts), + undelegate_action: UndelegateType::Standalone, + }) + } + CommitIntentKind::CommitFinalize => { + MagicBaseIntent::CommitFinalize(CommitType::Standalone(accounts)) + } + CommitIntentKind::CommitFinalizeAndUndelegate => { + MagicBaseIntent::CommitFinalizeAndUndelegate(CommitAndUndelegate { + commit_action: CommitType::Standalone(accounts), + undelegate_action: UndelegateType::Standalone, + }) + } + } +} + // ----------------- // +++++ Tests +++++ // ----------------- @@ -270,28 +299,7 @@ async fn commit_single_account( account, remote_slot: Default::default(), }; - let base_intent = match commit_type { - CommitIntentKind::Commit => { - MagicBaseIntent::Commit(CommitType::Standalone(vec![account])) - } - CommitIntentKind::CommitAndUndelegate => { - MagicBaseIntent::CommitAndUndelegate(CommitAndUndelegate { - commit_action: CommitType::Standalone(vec![account]), - undelegate_action: UndelegateType::Standalone, - }) - } - CommitIntentKind::CommitFinalize => { - MagicBaseIntent::CommitFinalize(CommitType::Standalone(vec![ - account, - ])) - } - CommitIntentKind::CommitFinalizeAndUndelegate => { - MagicBaseIntent::CommitFinalizeAndUndelegate(CommitAndUndelegate { - commit_action: CommitType::Standalone(vec![account]), - undelegate_action: UndelegateType::Standalone, - }) - } - }; + let base_intent = base_intent_for(commit_type, vec![account]); let intent = ScheduledIntentBundle { id: 0, @@ -352,28 +360,7 @@ async fn commit_book_order_account( account: order_book_ac, remote_slot: Default::default(), }; - let base_intent = match commit_type { - CommitIntentKind::Commit => { - MagicBaseIntent::Commit(CommitType::Standalone(vec![account])) - } - CommitIntentKind::CommitAndUndelegate => { - MagicBaseIntent::CommitAndUndelegate(CommitAndUndelegate { - commit_action: CommitType::Standalone(vec![account]), - undelegate_action: UndelegateType::Standalone, - }) - } - CommitIntentKind::CommitFinalize => { - MagicBaseIntent::CommitFinalize(CommitType::Standalone(vec![ - account, - ])) - } - CommitIntentKind::CommitFinalizeAndUndelegate => { - MagicBaseIntent::CommitFinalizeAndUndelegate(CommitAndUndelegate { - commit_action: CommitType::Standalone(vec![account]), - undelegate_action: UndelegateType::Standalone, - }) - } - }; + let base_intent = base_intent_for(commit_type, vec![account]); let intent = ScheduledIntentBundle { id: 0, @@ -393,8 +380,303 @@ async fn commit_book_order_account( .await; } -// TODO(thlorenz/snawaz): once delegation program supports larger commits -// add 1MB and 10MB tests +// ----------------- +// Oversized Commits (PreallocateBuffer) +// ----------------- +// +// Accounts whose committed state exceeds MAX_PERMITTED_DATA_INCREASE +// (10_240 bytes) can no longer grow their commit_state PDA (or, on +// finalize, the delegated account itself) in a single instruction, so +// preparation sends PreallocateBuffer instructions ahead of the actual +// commit/finalize to grow those accounts in steps first. + +#[tokio::test] +async fn test_ix_commit_single_account_50kb_buffer() { + commit_large_account( + 50 * 1024, + CommitStrategy::DiffBuffer, + CommitIntentKind::Commit, + ) + .await; +} + +#[tokio::test] +async fn test_ix_commit_finalize_single_account_350kb_buffer() { + // 350KB needs more preallocate steps than fit in one transaction + // (350 * 1024 / MAX_PERMITTED_DATA_INCREASE > PREALLOCATE_CHUNK_SIZE), + // so this also exercises preallocate chunking across multiple txs, for + // both the commit_state PDA (commit) and the delegated account (finalize). + commit_large_account( + 350 * 1024, + CommitStrategy::DiffBuffer, + CommitIntentKind::CommitFinalize, + ) + .await; +} + +#[tokio::test] +#[ignore = "large undelegations are not yet supported by dlp side"] +async fn test_ix_commit_and_undelegate_single_account_50kb_buffer() { + // Two-stage: commit (PreallocateCommitStateTask grows commit_state PDA) + // then finalize + undelegate (PreallocateFinalizeTask grows the + // delegated account itself, PreallocateUndelegateTask grows the + // undelegate buffer) -- exercises all three PreallocateBufferKind paths. + commit_large_account( + 50 * 1024, + CommitStrategy::DiffBuffer, + CommitIntentKind::CommitAndUndelegate, + ) + .await; +} + +#[tokio::test] +#[ignore = "large undelegations are not yet supported by dlp side"] +async fn test_ix_commit_finalize_and_undelegate_single_account_50kb_buffer() { + // Single combined commit+finalize (PreallocateCommitFinalizeTask grows + // the delegated account directly) followed by undelegate + // (PreallocateUndelegateTask grows the undelegate buffer). + commit_large_account( + 50 * 1024, + CommitStrategy::DiffBuffer, + CommitIntentKind::CommitFinalizeAndUndelegate, + ) + .await; +} + +/// Delegates a *small* account, then commits (and optionally finalizes) it +/// with `bytes` of brand new random data -- far more than what's currently +/// delegated. The delegated account starts small deliberately: Solana's own +/// realloc cap makes it impossible to delegate an account that's already +/// bigger than MAX_PERMITTED_DATA_INCREASE in one shot, so PreallocateBuffer +/// only ever needs to grow accounts *after* they're already delegated -- +/// exactly this shape. +/// +/// The random data (vs. only a few changed fields) guarantees a large diff +/// against the small base state, forcing buffer delivery: a small diff, +/// however large the account grows, could still fit inline as +/// [`CommitStrategy::DiffArgs`]. +async fn commit_large_account( + bytes: usize, + expected_strategy: CommitStrategy, + commit_type: CommitIntentKind, +) { + const DELEGATED_ACCOUNT_SIZE: u64 = 100; + + init_logger!(); + + let validator_auth = ensure_validator_authority(); + fund_validator_auth_and_ensure_validator_fees_vault(&validator_auth).await; + + let processor = Arc::new( + CommittorProcessor::try_new( + validator_auth.insecure_clone(), + ":memory:", + ChainConfig::local(ComputeBudgetConfig::new(1_000_000)), + None, + common::MockActionsCallbackExecutor::default(), + ) + .unwrap(), + ); + + let counter_auth = Keypair::new(); + let (pubkey, mut account) = init_and_delegate_account_on_chain( + &counter_auth, + DELEGATED_ACCOUNT_SIZE, + None, + ) + .await; + + account.data = common::generate_random_bytes(bytes); + account.owner = program_flexi_counter::id(); + account.lamports = + LAMPORTS_PER_SOL + Rent::default().minimum_balance(account.data.len()); + + let account = CommittedAccount { + pubkey, + account, + remote_slot: Default::default(), + }; + let base_intent = base_intent_for(commit_type, vec![account]); + + let intent = ScheduledIntentBundle { + id: 0, + slot: 10, + blockhash: Hash::new_unique(), + sent_transaction: Transaction::default(), + payer: counter_auth.pubkey(), + intent_bundle: base_intent.into(), + }; + + ix_commit_local( + processor, + vec![intent], + expect_strategies(&[(expected_strategy, 1)]), + program_flexi_counter::ID, + ) + .await; +} + +async fn commit_large_accounts( + bytes: &[usize], + bundle_size: usize, + commit_type: CommitIntentKind, + expected_strategies: ExpectedStrategies, +) { + const DELEGATED_ACCOUNT_SIZE: usize = 100; + + init_logger!(); + + let validator_auth = ensure_validator_authority(); + fund_validator_auth_and_ensure_validator_fees_vault(&validator_auth).await; + + // Create bundles of prepared to commit accounts of small size + let initial_bytes = bytes + .iter() + .map(|_| DELEGATED_ACCOUNT_SIZE) + .collect::>(); + let mut bundles_of_committees = + create_bundles(bundle_size, &initial_bytes).await; + + // Set data to required size + let mut i = 0; + for bundle in bundles_of_committees.iter_mut() { + for account in bundle { + account.account.data = common::generate_random_bytes(bytes[i]); + account.account.owner = program_flexi_counter::id(); + account.account.lamports = LAMPORTS_PER_SOL + + Rent::default().minimum_balance(account.data_len()); + i += 1; + } + } + + // Create intent for each bundle + let intents = bundles_of_committees + .into_iter() + .map(|committees| base_intent_for(commit_type, committees)) + .enumerate() + .map(|(id, base_intent)| ScheduledIntentBundle { + id: id as u64, + slot: 0, + blockhash: Hash::new_unique(), + sent_transaction: Transaction::default(), + payer: Pubkey::new_unique(), + intent_bundle: base_intent.into(), + }) + .collect::>(); + + let processor = Arc::new( + CommittorProcessor::try_new( + validator_auth.insecure_clone(), + ":memory:", + ChainConfig::local(ComputeBudgetConfig::new(1_000_000)), + None, + common::MockActionsCallbackExecutor::default(), + ) + .unwrap(), + ); + + ix_commit_local( + processor, + intents, + expected_strategies, + program_flexi_counter::ID, + ) + .await; +} + +#[tokio::test] +async fn test_ix_commit_two_large_one_small() { + init_logger!(); + commit_large_accounts( + &[20 * (1 << 10) - 7, 101, 20 * (1 << 10)], + 1, + CommitIntentKind::Commit, + expect_strategies(&[ + (CommitStrategy::DiffBuffer, 2), + (CommitStrategy::StateArgs, 1), + ]), + ) + .await; +} + +#[tokio::test] +async fn test_ix_commit_two_accounts_20kb() { + init_logger!(); + commit_large_accounts( + &[20 * (1 << 10), 20 * (1 << 10)], + 1, + CommitIntentKind::Commit, + expect_strategies(&[(CommitStrategy::DiffBuffer, 2)]), + ) + .await; +} + +#[tokio::test] +async fn test_ix_commit_finalize_3_large_1_small() { + init_logger!(); + commit_large_accounts( + &[20 * (1 << 10), 20 * (1 << 10), 32 * (1 << 10) + 3, 201], + 1, + CommitIntentKind::CommitFinalize, + expect_strategies(&[ + (CommitStrategy::DiffBuffer, 3), + (CommitStrategy::StateArgs, 1), + ]), + ) + .await; +} + +#[tokio::test] +async fn test_ix_commit_7_large_2_small() { + init_logger!(); + let large = 22 * (1 << 10); + commit_large_accounts( + &[ + large, + large, + 32 * (1 << 10) + 3, + large, + 201, + large, + large - 1, + large, + 103, + ], + 1, + CommitIntentKind::Commit, + expect_strategies(&[ + (CommitStrategy::DiffBuffer, 7), + (CommitStrategy::StateArgs, 2), + ]), + ) + .await; +} + +#[tokio::test] +async fn test_ix_commit_finalize_7_large_2_small() { + init_logger!(); + let large = 21 * (1 << 10); + commit_large_accounts( + &[ + large, + large, + 32 * (1 << 10) + 3, + large, + 201, + large, + large - 1, + large, + 103, + ], + 1, + CommitIntentKind::CommitFinalize, + expect_strategies(&[ + (CommitStrategy::DiffBuffer, 7), + (CommitStrategy::StateArgs, 2), + ]), + ) + .await; +} // ----------------- // Multiple Account Commits @@ -814,30 +1096,7 @@ async fn commit_multiple_accounts( // Create intent for each bundle let intents = bundles_of_committees .into_iter() - .map(|committees| match commit_type { - CommitIntentKind::Commit => { - MagicBaseIntent::Commit(CommitType::Standalone(committees)) - } - CommitIntentKind::CommitAndUndelegate => { - MagicBaseIntent::CommitAndUndelegate(CommitAndUndelegate { - commit_action: CommitType::Standalone(committees), - undelegate_action: UndelegateType::Standalone, - }) - } - CommitIntentKind::CommitFinalize => { - MagicBaseIntent::CommitFinalize(CommitType::Standalone( - committees, - )) - } - CommitIntentKind::CommitFinalizeAndUndelegate => { - MagicBaseIntent::CommitFinalizeAndUndelegate( - CommitAndUndelegate { - commit_action: CommitType::Standalone(committees), - undelegate_action: UndelegateType::Standalone, - }, - ) - } - }) + .map(|committees| base_intent_for(commit_type, committees)) .enumerate() .map(|(id, base_intent)| ScheduledIntentBundle { id: id as u64, @@ -948,12 +1207,18 @@ async fn execute_intent_bundle( // ----------------- // Test Executor // ----------------- + async fn ix_commit_local( processor: Arc, intent_bundles: Vec, expected_strategies: ExpectedStrategies, program_id: Pubkey, ) { + let rpc_client = RpcClient::new_with_commitment( + "http://localhost:7799".to_string(), + CommitmentConfig::confirmed(), + ); + let execution_outputs = processor .execute_intent_bundles(intent_bundles.clone()) .await @@ -964,7 +1229,6 @@ async fn ix_commit_local( // Assert that all completed assert_eq!(execution_outputs.len(), intent_bundles.len()); - let rpc_client = RpcClient::new("http://localhost:7799".to_string()); let mut strategies = ExpectedStrategies::new(); for (execution_result, base_intent) in execution_outputs .into_iter() @@ -1055,7 +1319,6 @@ async fn ix_commit_local( dlp_api::id() }; - let lamports = account.account.lamports; get_account!( rpc_client, account.pubkey, @@ -1065,7 +1328,7 @@ async fn ix_commit_local( acc, remaining_tries, &account.account.data, - lamports, + account.account.lamports, expected_owner, account.pubkey, has_undelegate, @@ -1097,8 +1360,9 @@ fn validate_account( account_pubkey: Pubkey, is_undelegate: bool, ) -> bool { - let matches_data = - acc.data() == expected_data && acc.lamports() == expected_lamports; + let matches_data = acc.data() == expected_data; + let matched_balance = acc.lamports() == expected_lamports; + let matches_data = matches_data && matched_balance; let matches_undelegation = acc.owner().eq(&expected_owner); let matches_all = matches_data && matches_undelegation; diff --git a/test-integration/test-committor-service/tests/test_transaction_preparator.rs b/test-integration/test-committor-service/tests/test_transaction_preparator.rs index 059279d68..9b1e75dce 100644 --- a/test-integration/test-committor-service/tests/test_transaction_preparator.rs +++ b/test-integration/test-committor-service/tests/test_transaction_preparator.rs @@ -3,7 +3,7 @@ use magicblock_committor_program::Chunks; use magicblock_committor_service::{ persist::IntentPersisterImpl, tasks::{ - commit_stage_task::CleanupTask, + preparation_task::CleanupTask, task_strategist::{TaskStrategist, TransactionStrategy}, utils::{create_commit_task, TransactionUtils}, BaseActionTask, BaseActionTaskV1, BaseTaskImpl, FinalizeTask, @@ -37,6 +37,7 @@ async fn test_prepare_commit_tx_with_single_account() { create_commit_task(1, true, committed_account.clone(), None).into(), FinalizeTask { delegated_account: committed_account.pubkey, + state_size: committed_account.account.data.len(), } .into(), ]; @@ -97,11 +98,13 @@ async fn test_prepare_commit_tx_with_multiple_accounts() { // finalize account 1 FinalizeTask { delegated_account: committed_account1.pubkey, + state_size: committed_account1.data_len(), } .into(), // finalize account 2 FinalizeTask { delegated_account: committed_account2.pubkey, + state_size: committed_account2.data_len(), } .into(), ]; @@ -189,6 +192,7 @@ async fn test_prepare_commit_tx_with_base_actions() { // finalize account FinalizeTask { delegated_account: committed_account.pubkey, + state_size: committed_account.data_len(), } .into(), // BaseAction @@ -263,6 +267,7 @@ async fn test_prepare_finalize_tx_with_undelegate_with_atls() { // finalize account FinalizeTask { delegated_account: committed_account.pubkey, + state_size: committed_account.data_len(), } .into(), // Undelegate @@ -271,6 +276,7 @@ async fn test_prepare_finalize_tx_with_undelegate_with_atls() { owner_program: Pubkey::new_unique(), rent_reimbursement: Pubkey::new_unique(), include_undelegation_request: false, + state_size: committed_account.data_len(), } .into(), ];