diff --git a/kb-pipeline/src/lib.rs b/kb-pipeline/src/lib.rs index a549d7c..849263d 100644 --- a/kb-pipeline/src/lib.rs +++ b/kb-pipeline/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-pipeline/src/lib.rs -// version: 11 +// version: 12 #![forbid(unsafe_code)] #![deny(unreachable_pub)] @@ -14,6 +14,7 @@ mod decode_replay; mod plan; mod solana_elgamal_registry_stateful; mod solana_execution; +mod solana_memo_execution; mod solana_stateful; mod solana_token2022_correlation; mod solana_token2022_crypto_preflight; @@ -24,6 +25,17 @@ mod solana_token2022_proof_orchestration; mod solana_token2022_stateful; mod solana_token2022_validation; +/// Emits one execution progress event for specialized pipeline orchestrators. +pub(crate) use self::solana_execution::emit; +/// Rejects one execution stage when the observer reports cancellation. +pub(crate) use self::solana_execution::ensure_not_cancelled; +/// Loads the persistent wallet configured by one execution profile. +pub(crate) use self::solana_execution::load_profile_wallet; +/// Formats one failed simulation without discarding runtime diagnostics. +pub(crate) use self::solana_execution::simulation_failure_message; +/// Formats safety violations for one denied execution plan. +pub(crate) use self::solana_execution::violation_message; + /// Address category used by one targeted backfill campaign. pub use self::backfill::BackfillAddressKind; /// Chronological direction relative to one anchor signature. @@ -120,6 +132,12 @@ pub use self::solana_execution::SolanaExecutionProgressEvent; pub use self::solana_execution::SolanaExecutionProgressLevel; /// Executes one bounded System Program transfer on Devnet. pub use self::solana_execution::execute_devnet_system_transfer; +/// Complete request for one SPL Memo v4 Devnet execution. +pub use self::solana_memo_execution::DevnetMemoExecutionRequest; +/// Complete result of one SPL Memo v4 Devnet execution and post-validation. +pub use self::solana_memo_execution::DevnetMemoExecutionSummary; +/// Executes one SPL Memo v4 Devnet simulation or explicitly authorized submission. +pub use self::solana_memo_execution::execute_devnet_memo; /// One machine-readable native Solana stateful readiness check. pub use self::solana_stateful::SolanaCoreStatefulCheck; /// One contextual fact measured during native Solana stateful readiness. diff --git a/kb-pipeline/src/solana_execution.rs b/kb-pipeline/src/solana_execution.rs index 5d99bb1..1236d75 100644 --- a/kb-pipeline/src/solana_execution.rs +++ b/kb-pipeline/src/solana_execution.rs @@ -1,5 +1,5 @@ // file: kb-pipeline/src/solana_execution.rs -// version: 5 +// version: 6 //! Devnet Solana execution orchestration with canonical post-validation. @@ -904,7 +904,7 @@ fn validate_recipient_transfer_amount( return std::result::Result::Ok(()); } -fn simulation_failure_message( +pub(crate) fn simulation_failure_message( simulation: &kb_lib::ExApiExecutionSimulationResult, ) -> std::string::String { let error = match simulation.error.as_deref() { @@ -976,7 +976,7 @@ fn validate_devnet_profile( return std::result::Result::Ok(()); } -async fn load_profile_wallet( +pub(crate) async fn load_profile_wallet( profile: &kb_config::ProfileConfig, workspace_root: &std::path::Path, ) -> kb_core::Result { @@ -1176,7 +1176,7 @@ fn build_transfer_plan( return kb_lib::ExApiTypedInstructionExecutor::build_prepared_plan(&executor, &intent); } -fn violation_message(violations: &[kb_lib::ExSafetyViolation]) -> std::string::String { +pub(crate) fn violation_message(violations: &[kb_lib::ExSafetyViolation]) -> std::string::String { if violations.is_empty() { return "execution policy denied the plan without a diagnostic".to_string(); } @@ -1187,7 +1187,7 @@ fn violation_message(violations: &[kb_lib::ExSafetyViolation]) -> std::string::S .join("; "); } -fn ensure_not_cancelled(observer: &O, stage: &str) -> kb_core::Result<()> +pub(crate) fn ensure_not_cancelled(observer: &O, stage: &str) -> kb_core::Result<()> where O: crate::SolanaExecutionObserver, { @@ -1200,7 +1200,7 @@ where return std::result::Result::Ok(()); } -fn emit( +pub(crate) fn emit( observer: &O, level: crate::SolanaExecutionProgressLevel, stage: impl std::convert::Into, diff --git a/kb-pipeline/src/solana_memo_execution.rs b/kb-pipeline/src/solana_memo_execution.rs new file mode 100644 index 0000000..0bdba10 --- /dev/null +++ b/kb-pipeline/src/solana_memo_execution.rs @@ -0,0 +1,903 @@ +// file: kb-pipeline/src/solana_memo_execution.rs +// version: 2 + +//! Devnet SPL Memo v4 execution with canonical post-validation. + +/// Complete request for one SPL Memo v4 Devnet execution. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DevnetMemoExecutionRequest { + /// Stable caller-provided execution identifier. + pub intent_id: std::string::String, + /// Endpoint role used for cluster, balance, blockhash, fee and hydration calls. + pub query_role: std::string::String, + /// Endpoint role used for simulation, submission and confirmation polling. + pub transaction_role: std::string::String, + /// Exact UTF-8 Memo payload. + pub message: std::string::String, + /// Supplies the persistent Devnet fee payer as a Memo signer account. + pub include_wallet_as_memo_signer: bool, + /// Explicitly authorizes signing and submission after successful simulation. + pub submit: bool, + /// Explicit operator confirmation required by the active profile. + pub operator_confirmed: bool, + /// Number of `getTransaction` retries after the first hydration attempt. + pub post_validation_max_retries: u32, + /// Replaces existing core and decode outputs for the submitted signature. + pub force_post_validation_replay: bool, +} + +impl DevnetMemoExecutionRequest { + /// Creates a conservative simulation-only Memo v4 request. + pub fn new( + intent_id: impl std::convert::Into, + message: impl std::convert::Into, + ) -> Self { + return Self { + intent_id: intent_id.into(), + query_role: "http_queries".to_string(), + transaction_role: "http_transactions".to_string(), + message: message.into(), + include_wallet_as_memo_signer: true, + submit: false, + operator_confirmed: false, + post_validation_max_retries: 10, + force_post_validation_replay: false, + }; + } + + /// Validates request-local bounds independently from one profile. + pub fn validate(&self) -> kb_core::Result<()> { + if self.intent_id.trim().is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "Devnet Memo execution intent id must not be empty", + )); + } + if self.query_role.trim().is_empty() || self.transaction_role.trim().is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "Devnet Memo execution endpoint roles must not be empty", + )); + } + if self.message.len() > kb_lib::EX_SPL_MEMO_MAX_MESSAGE_BYTES { + return std::result::Result::Err(kb_core::Error::config(format!( + "Devnet Memo payload length {} exceeds the executor limit {}", + self.message.len(), + kb_lib::EX_SPL_MEMO_MAX_MESSAGE_BYTES + ))); + } + if self.post_validation_max_retries > 20 { + return std::result::Result::Err(kb_core::Error::config( + "post-execution getTransaction retries must not exceed 20", + )); + } + return std::result::Result::Ok(()); + } +} + +/// Complete result of one SPL Memo v4 Devnet execution. +#[derive(Clone, Debug, PartialEq)] +pub struct DevnetMemoExecutionSummary { + /// Profile used by the orchestration. + pub profile_name: std::string::String, + /// Exact classified cluster. + pub cluster: kb_lib::ExApiExecutionCluster, + /// Genesis hash returned by the selected endpoint. + pub genesis_hash: std::string::String, + /// Non-secret persistent wallet description. + pub wallet: kb_wallet::WalletSummary, + /// Wallet balance observed before planning. + pub balance_lamports: u64, + /// Exact prepared Memo plan. + pub plan: kb_lib::ExApiPreparedExecutionPlan, + /// Recent blockhash used by the exact transaction. + pub latest_blockhash: kb_onchain_transport::LatestBlockhashResult, + /// Fee estimate for the exact compiled message. + pub fee: kb_onchain_transport::FeeForMessageResult, + /// Exact simulation result bound to the compiled message. + pub simulation: kb_lib::ExApiExecutionSimulationResult, + /// Submission result when explicitly authorized. + pub send_result: std::option::Option, + /// Confirmation result when submitted. + pub confirmation: std::option::Option, + /// Canonical hydration result for the exact signature. + pub backfill: std::option::Option, + /// Core extraction result for the exact signature. + pub core_extraction: std::option::Option, + /// First Memo decode and materialization replay. + pub decode_replay: std::option::Option, + /// Second replay proving that the same decoder version and input are idempotent. + pub idempotence_replay: std::option::Option, + /// Exact persisted transaction annotation rows for the submitted signature. + pub annotations: std::vec::Vec, + /// Aggregated post-execution validation diagnostic. + pub post_execution: std::option::Option, +} + +/// Executes one Memo v4 Devnet simulation or explicitly authorized submission. +#[allow(clippy::too_many_arguments)] +pub async fn execute_devnet_memo( + http_pool: &kb_onchain_transport::HttpEndpointPool, + store: &S, + profile: &kb_config::ProfileConfig, + workspace_root: &std::path::Path, + request: &crate::DevnetMemoExecutionRequest, + decoders: &[std::sync::Arc], + materializers: &[std::sync::Arc], + observer: &O, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + + kb_store::CoreExtractionStore + + kb_store::DecodePipelineStore + + Sync, + O: crate::SolanaExecutionObserver, +{ + if let std::result::Result::Err(error) = request.validate() { + return std::result::Result::Err(error); + } + if let std::result::Result::Err(error) = validate_profile(profile, request) { + return std::result::Result::Err(error); + } + if let std::result::Result::Err(error) = crate::ensure_not_cancelled(observer, "validate") { + return std::result::Result::Err(error); + } + let genesis = match http_pool.get_genesis_hash_for_role(request.query_role.as_str()).await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if genesis.classified_cluster + != std::option::Option::Some(kb_lib::ExApiExecutionCluster::Devnet) + { + return std::result::Result::Err(kb_core::Error::new( + "execution_cluster_mismatch", + format!( + "expected Devnet genesis hash but endpoint returned {} classified as {:?}", + genesis.genesis_hash, genesis.classified_cluster + ), + )); + } + let wallet = match crate::load_profile_wallet(profile, workspace_root).await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let wallet_summary = wallet.summary(); + let fee_payer = kb_lib::MdPubkey(wallet_summary.public_key.clone()); + let balance = match http_pool + .get_balance_for_role( + request.query_role.as_str(), + &fee_payer, + &kb_onchain_transport::GetBalanceConfig::confirmed(), + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if balance.lamports < profile.execution.max_fee_lamports { + return std::result::Result::Err(kb_core::Error::new( + "execution_balance_insufficient", + format!( + "Devnet wallet balance {} is below the configured fee ceiling {}", + balance.lamports, profile.execution.max_fee_lamports + ), + )); + } + let plan = match build_plan(profile, request, fee_payer.clone()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let plan_evaluation = match kb_lib::ExSafetyChecker.evaluate_prepared_plan(&plan) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if plan_evaluation.decision == kb_lib::ExSafetyDecision::Deny { + return std::result::Result::Err(kb_core::Error::new( + "execution_plan_denied", + crate::violation_message(plan_evaluation.violations.as_slice()), + )); + } + let latest_blockhash = match http_pool + .get_latest_blockhash_for_role( + request.query_role.as_str(), + &kb_onchain_transport::GetLatestBlockhashConfig::confirmed(), + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let unsigned = match kb_lib::executor_solana_build_legacy_transaction( + &plan, + latest_blockhash.blockhash.as_str(), + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let message_base64 = unsigned.message_base64(); + let fee = match http_pool + .get_fee_for_message_for_role( + request.query_role.as_str(), + message_base64.as_str(), + &kb_onchain_transport::GetFeeForMessageConfig::new( + kb_onchain_transport::RpcCommitmentLevel::Confirmed, + std::option::Option::Some(latest_blockhash.context.slot), + ), + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if fee.fee_lamports.is_none() { + return std::result::Result::Err(kb_core::Error::new( + "execution_fee_unavailable", + "getFeeForMessage returned null for the selected recent blockhash", + )); + } + let unsigned_base64 = match unsigned.transaction_base64() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let simulation_config = match kb_onchain_transport::SimulateTransactionConfig::new( + kb_onchain_transport::RpcCommitmentLevel::Confirmed, + false, + false, + std::option::Option::Some(latest_blockhash.context.slot), + true, + std::option::Option::None, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + crate::emit( + observer, + crate::SolanaExecutionProgressLevel::Info, + "memo_simulation", + format!("simulating exact Memo message {}", unsigned.message_hash()), + std::option::Option::None, + ); + let simulation_rpc = match http_pool + .simulate_transaction_for_role( + request.transaction_role.as_str(), + unsigned_base64.as_str(), + &simulation_config, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let simulation = simulation_rpc.to_execution_result( + kb_lib::ExApiExecutionCluster::Devnet, + kb_lib::ExApiExecutionBlockhashKind::Latest, + std::option::Option::Some( + simulation_rpc.context.slot.saturating_sub(latest_blockhash.context.slot), + ), + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(&fee), + ); + let evidence = unsigned.bind_simulation(simulation.clone()); + let mut summary = crate::DevnetMemoExecutionSummary { + profile_name: profile.name.clone(), + cluster: kb_lib::ExApiExecutionCluster::Devnet, + genesis_hash: genesis.genesis_hash, + wallet: wallet_summary, + balance_lamports: balance.lamports, + plan, + latest_blockhash, + fee, + simulation, + send_result: std::option::Option::None, + confirmation: std::option::Option::None, + backfill: std::option::Option::None, + core_extraction: std::option::Option::None, + decode_replay: std::option::Option::None, + idempotence_replay: std::option::Option::None, + annotations: std::vec::Vec::new(), + post_execution: std::option::Option::None, + }; + if !request.submit { + return std::result::Result::Ok(summary); + } + if !summary.simulation.success { + return std::result::Result::Err(kb_core::Error::new( + "execution_simulation_failed", + crate::simulation_failure_message(&summary.simulation), + )); + } + let send_evaluation = + match kb_lib::ExSafetyChecker.evaluate_send(&summary.plan, &summary.simulation) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if send_evaluation.decision == kb_lib::ExSafetyDecision::Deny { + return std::result::Result::Err(kb_core::Error::new( + "execution_send_denied", + crate::violation_message(send_evaluation.violations.as_slice()), + )); + } + let signed = match unsigned.sign_after_simulation(&evidence, &[wallet.as_signer()]) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::result::Result::Err(error) = signed.verify_signatures() { + return std::result::Result::Err(error); + } + let signature = signed.primary_signature().clone(); + let mut diagnostic = kb_lib::ExApiPostExecutionDiagnostic { + signature: signature.clone(), + canonical_inserted: false, + core_extracted: false, + decode_replayed: false, + materialized: false, + diagnostics: std::vec::Vec::new(), + }; + let send_config = match kb_onchain_transport::SendTransactionConfig::from_execution_config( + &profile.execution, + std::option::Option::Some(summary.latest_blockhash.context.slot), + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let signed_base64 = signed.transaction_base64(); + let sent = match http_pool + .send_transaction_for_role( + request.transaction_role.as_str(), + signed_base64.as_str(), + &signature, + &send_config, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo submission failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + summary.send_result = + std::option::Option::Some(sent.to_execution_result(kb_lib::ExApiExecutionCluster::Devnet)); + let confirmation_config = + match kb_onchain_transport::ConfirmTransactionConfig::from_execution_config( + &profile.execution, + std::option::Option::Some(summary.latest_blockhash.last_valid_block_height), + std::option::Option::Some(summary.latest_blockhash.context.slot), + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let confirmation = match http_pool + .confirm_transaction_for_roles( + request.transaction_role.as_str(), + request.query_role.as_str(), + kb_lib::ExApiExecutionCluster::Devnet, + &signature, + &confirmation_config, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo confirmation failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + let confirmation_status = confirmation.status; + summary.confirmation = std::option::Option::Some(confirmation); + if !matches!( + confirmation_status, + kb_lib::ExApiExecutionConfirmationStatus::Confirmed + | kb_lib::ExApiExecutionConfirmationStatus::Finalized + ) { + diagnostic.diagnostics.push(format!( + "Memo post-validation stopped at confirmation status {confirmation_status:?}" + )); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + } + let backfill = + match hydrate_signature(http_pool, store, profile, request, observer, &signature).await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo hydration failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + diagnostic.canonical_inserted = canonical_available(&backfill); + summary.backfill = std::option::Option::Some(backfill); + if !diagnostic.canonical_inserted { + diagnostic + .diagnostics + .push("confirmed Memo transaction was unavailable for canonical hydration".to_string()); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + } + let extraction = match crate::execute_core_extraction( + store, + &crate::CoreExtractionRequest { + source: crate::CoreExtractionSource::Signatures(std::vec![signature.0.clone()]), + limit: 1, + max_concurrent_extractions: 1, + force_replay: request.force_post_validation_replay, + }, + observer, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo core extraction failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + diagnostic.core_extracted = extraction.failed == 0 + && !extraction.cancelled + && extraction.selected == 1 + && extraction.extracted.saturating_add(extraction.skipped) >= 1; + summary.core_extraction = std::option::Option::Some(extraction); + if !diagnostic.core_extracted { + diagnostic.diagnostics.push("Memo core extraction did not complete".to_string()); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + } + let first_replay = + match replay_memo(store, request, &signature, false, decoders, materializers, observer) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo decode replay failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + diagnostic.decode_replayed = decode_completed(&first_replay); + summary.decode_replay = std::option::Option::Some(first_replay); + let filter = match kb_store::MaterializedEventFilter::new( + std::option::Option::Some("transaction_annotations".to_string()), + std::option::Option::Some("transaction_annotation".to_string()), + std::option::Option::Some(signature.0.clone()), + 8, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + summary.annotations = + match kb_store::DecodePipelineStore::list_materialized_events(store, &filter).await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo annotation query failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + diagnostic.materialized = diagnostic.decode_replayed + && summary + .annotations + .iter() + .any(|row| return row.signature.as_str() == signature.0.as_str()); + let second_replay = match replay_memo( + store, + request, + &signature, + true, + decoders, + materializers, + observer, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + diagnostic.diagnostics.push(format!("Memo idempotence replay failed: {error}")); + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); + }, + }; + let idempotent = second_replay.failed_inputs == 0 + && second_replay.processors.iter().all(|processor| { + return processor.failed == 0 + && processor.materialized_outputs == 0 + && processor.materialization_refused == 0; + }); + summary.idempotence_replay = std::option::Option::Some(second_replay); + if !idempotent { + diagnostic + .diagnostics + .push("second Memo replay did not prove a clean idempotent skip".to_string()); + } else if diagnostic.canonical_inserted + && diagnostic.core_extracted + && diagnostic.decode_replayed + && diagnostic.materialized + { + diagnostic.diagnostics.push( + "Memo completed canonical hydration, core extraction, decode, annotation projection and idempotence validation" + .to_string(), + ); + } + summary.post_execution = std::option::Option::Some(diagnostic); + return std::result::Result::Ok(summary); +} + +fn validate_profile( + profile: &kb_config::ProfileConfig, + request: &crate::DevnetMemoExecutionRequest, +) -> kb_core::Result<()> { + if profile.wallet.cluster != "devnet" { + return std::result::Result::Err(kb_core::Error::config( + "Memo Devnet orchestration requires a Devnet wallet profile", + )); + } + if !profile.wallet.temporary_wallet_enabled || !profile.wallet.temporary_wallet_persist { + return std::result::Result::Err(kb_core::Error::config( + "Memo Devnet orchestration requires an enabled persistent temporary wallet", + )); + } + if !profile.execution.require_simulation { + return std::result::Result::Err(kb_core::Error::config( + "Memo Devnet orchestration requires simulation", + )); + } + if request.submit && !profile.wallet.devnet_send_enabled { + return std::result::Result::Err(kb_core::Error::config( + "Devnet transaction submission is disabled by the wallet profile", + )); + } + if request.submit + && profile.execution.require_operator_confirmation + && !request.operator_confirmed + { + return std::result::Result::Err(kb_core::Error::config( + "Memo Devnet submission requires explicit operator confirmation", + )); + } + return std::result::Result::Ok(()); +} + +fn build_plan( + profile: &kb_config::ProfileConfig, + request: &crate::DevnetMemoExecutionRequest, + fee_payer: kb_lib::MdPubkey, +) -> kb_core::Result { + let signers = if request.include_wallet_as_memo_signer { + std::vec![kb_lib::ExSplMemoSigner { pubkey: fee_payer.clone() }] + } else { + std::vec::Vec::new() + }; + let intent = kb_lib::ExSplMemoExecutionIntent { + intent_id: request.intent_id.clone(), + fee_payer: fee_payer.clone(), + policy: kb_lib::ExApiExecutionPolicy { + cluster: kb_lib::ExApiExecutionClusterPolicy { + expected_cluster: kb_lib::ExApiExecutionCluster::Devnet, + allow_mainnet: false, + mainnet_confirmation: false, + }, + simulation: kb_lib::ExApiExecutionSimulationPolicy::Required, + blockhash: kb_lib::ExApiExecutionBlockhashPolicy { + kind: kb_lib::ExApiExecutionBlockhashKind::Latest, + max_age_slots: std::option::Option::Some( + profile.execution.recent_blockhash_max_age_slots, + ), + nonce_account: std::option::Option::None, + nonce_authority: std::option::Option::None, + }, + cost_limit: kb_lib::ExApiExecutionCostLimit { + max_spend_lamports: std::option::Option::Some(0), + max_fee_lamports: std::option::Option::Some(profile.execution.max_fee_lamports), + max_compute_unit_price_micro_lamports: std::option::Option::Some( + profile.execution.max_compute_unit_price_micro_lamports, + ), + }, + authorized_signers: std::vec![fee_payer.clone()], + dry_run: !request.submit, + post_execution_validation: kb_lib::ExApiPostExecutionValidationPolicy { + canonical_insert_required: true, + core_extraction_required: true, + decode_replay_required: true, + materialization_required: true, + }, + }, + operation: kb_lib::ExSplMemoOperation::AddMemo { + generation: kb_lib::ExSplMemoGeneration::V4, + message: request.message.clone(), + signers, + }, + }; + return kb_lib::ExApiTypedInstructionExecutor::build_prepared_plan( + &kb_lib::ExSplMemoExecutor, + &intent, + ); +} + +async fn hydrate_signature( + http_pool: &kb_onchain_transport::HttpEndpointPool, + store: &S, + profile: &kb_config::ProfileConfig, + request: &crate::DevnetMemoExecutionRequest, + observer: &O, + signature: &kb_lib::MdSignature, +) -> kb_core::Result +where + S: kb_store::RawTransactionStore + Sync, + O: crate::SolanaExecutionObserver, +{ + let mut retry = 0_u32; + loop { + let result = match crate::execute_http_backfill( + http_pool, + store, + &crate::BackfillRequest { + role: request.query_role.clone(), + commitment: "confirmed".to_string(), + source: crate::BackfillSource::ExplicitSignatures(std::vec![signature.0.clone()]), + page_size: 1, + max_pages: 1, + max_concurrent_requests: 1, + max_retries: 0, + }, + observer, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if canonical_available(&result) + || retry >= request.post_validation_max_retries + || observer.is_execution_cancelled() + { + return std::result::Result::Ok(result); + } + retry = retry.saturating_add(1); + tokio::time::sleep(std::time::Duration::from_millis(std::cmp::max( + profile.execution.confirmation_poll_interval_ms, + 500, + ))) + .await; + } +} + +fn canonical_available(summary: &crate::BackfillSummary) -> bool { + return summary.failed == 0 + && summary.missing == 0 + && summary.candidates_completed == 1 + && summary.candidates_cancelled == 0 + && summary.candidates_not_started == 0 + && summary + .canonical_inserted + .saturating_add(summary.canonical_skipped) + .saturating_add(summary.existing_skipped) + >= 1; +} + +async fn replay_memo( + store: &S, + request: &crate::DevnetMemoExecutionRequest, + signature: &kb_lib::MdSignature, + include_materialized_state: bool, + decoders: &[std::sync::Arc], + materializers: &[std::sync::Arc], + observer: &O, +) -> kb_core::Result +where + S: kb_store::DecodePipelineStore + Sync, + O: crate::SolanaExecutionObserver, +{ + let mut states = std::vec![ + kb_store::CoreInstructionProcessingState::Pending, + kb_store::CoreInstructionProcessingState::Failed, + kb_store::CoreInstructionProcessingState::ReplayRequested, + ]; + if include_materialized_state { + states.push(kb_store::CoreInstructionProcessingState::Materialized); + } + let selection = match kb_store::DecodeSelectionFilter::new( + std::vec![signature.0.clone()], + states, + std::option::Option::None, + std::option::Option::None, + std::vec![kb_program_ids::SPL_MEMO_V4_PROGRAM_ID.to_string()], + std::vec::Vec::new(), + false, + 8, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return crate::execute_decode_replay( + store, + &crate::DecodeReplayRequest { + campaign_id: crate::new_decode_campaign_id(), + selection, + decoder_names: std::vec::Vec::new(), + dispatch_policy: crate::DecodeDispatchPolicy::HighestPriority, + max_concurrent_inputs: 1, + force_replay: if include_materialized_state { + false + } else { + request.force_post_validation_replay + }, + force_replay_all_matching: false, + materialize_after_decode: true, + }, + decoders, + materializers, + observer, + ) + .await; +} + +fn decode_completed(summary: &crate::DecodeReplaySummary) -> bool { + return summary.failed_inputs == 0 + && summary.unmatched == 0 + && !summary.cancelled + && summary.completed >= 1 + && summary.processors.iter().all(|processor| { + return processor.failed == 0 + && processor.unsupported == 0 + && processor.materialization_refused == 0; + }) + && summary.processors.iter().map(|processor| return processor.decoded).sum::() >= 1; +} + +#[cfg(test)] +mod tests { + fn local_devnet_profile() -> kb_config::ProfileConfig { + let config = + match kb_config::parse_config_json(include_str!("../../config/example.config.json")) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("example config parse failed: {error}"), + }; + for profile in config.profiles { + if profile.name == "local_devnet" { + return profile; + } + } + panic!("local_devnet profile missing"); + } + + #[test] + fn request_is_simulation_only_and_bounded_by_default() { + let request = crate::DevnetMemoExecutionRequest::new("memo-1", "audit annotation"); + assert!(!request.submit); + assert!(request.include_wallet_as_memo_signer); + assert!(request.validate().is_ok()); + let oversized = "x".repeat(kb_lib::EX_SPL_MEMO_MAX_MESSAGE_BYTES + 1); + assert!(crate::DevnetMemoExecutionRequest::new("memo-2", oversized).validate().is_err()); + } + + #[test] + fn exact_v4_plan_has_zero_spend_and_wallet_signer() { + let profile = local_devnet_profile(); + let fee_payer = kb_lib::MdPubkey(kb_program_ids::SYSTEM_PROGRAM_ID.to_string()); + let mut request = crate::DevnetMemoExecutionRequest::new("memo-3", "hello"); + request.submit = true; + request.operator_confirmed = true; + let plan = match super::build_plan(&profile, &request, fee_payer.clone()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("Memo plan failed: {error}"), + }; + assert_eq!(plan.requested_spend_lamports, 0); + assert_eq!(plan.fee_payer, fee_payer); + assert_eq!(plan.instructions.len(), 1); + assert_eq!(plan.instructions[0].program_id.0, kb_program_ids::SPL_MEMO_V4_PROGRAM_ID); + assert_eq!(plan.instructions[0].accounts.len(), 1); + assert!(plan.instructions[0].accounts[0].is_signer); + assert!(!plan.instructions[0].accounts[0].is_writable); + assert!(kb_lib::ExSafetyChecker.evaluate_prepared_plan(&plan).is_ok()); + } + + #[test] + fn submission_requires_profile_enablement_and_confirmation() { + let profile = local_devnet_profile(); + let mut request = crate::DevnetMemoExecutionRequest::new("memo-4", "hello"); + assert!(super::validate_profile(&profile, &request).is_ok()); + request.submit = true; + assert!(super::validate_profile(&profile, &request).is_err()); + request.operator_confirmed = true; + assert!(super::validate_profile(&profile, &request).is_ok()); + } + + #[tokio::test] + async fn optional_devnet_memo_execution_from_env() { + if std::env::var("KB_DEVNET_MEMO_EXECUTION_TEST").ok().as_deref() + != std::option::Option::Some("1") + { + return; + } + let database_url = match std::env::var("KB_POSTGRES_TEST_URL") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + panic!("KB_POSTGRES_TEST_URL is required: {error}"); + }, + }; + let mut profile = local_devnet_profile(); + profile.database.backend = "postgres".to_string(); + profile.database.postgres.url = database_url; + if let std::result::Result::Ok(directory) = std::env::var("KB_DEVNET_WALLET_DIR") { + profile.wallet.wallet_dir = directory; + } + let pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&profile) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("HTTP pool creation failed: {error}"), + }; + let store_options = match kb_store::PostgresStoreOptions::new( + profile.database.postgres.url.clone(), + profile.database.postgres.max_connections, + profile.database.postgres.connect_timeout_ms, + false, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("PostgreSQL options failed: {error}"), + }; + let store = match kb_store::PostgresStore::connect(store_options).await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("PostgreSQL connection failed: {error}"), + }; + if let std::result::Result::Err(error) = store.initialize_store_schema().await { + panic!("PostgreSQL schema initialization failed: {error}"); + } + let mut request = crate::DevnetMemoExecutionRequest::new( + format!("devnet-memo-test-{}", uuid::Uuid::new_v4()), + format!("khadhroony-bot3 memo validation {}", uuid::Uuid::new_v4()), + ); + request.post_validation_max_retries = 20; + if std::env::var("KB_DEVNET_MEMO_SUBMIT").ok().as_deref() == std::option::Option::Some("1") + { + request.submit = true; + request.operator_confirmed = true; + } + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(kb_lib::DcSplMemoDecoder)]; + let materializers: std::vec::Vec> = + std::vec![std::sync::Arc::new(kb_lib::MtTransactionAnnotationMaterializer,)]; + let workspace_root = match std::path::Path::new(env!("CARGO_MANIFEST_DIR")).parent() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("workspace root cannot be resolved"), + }; + let summary = match crate::execute_devnet_memo( + &pool, + &store, + &profile, + workspace_root, + &request, + decoders.as_slice(), + materializers.as_slice(), + &crate::NoopSolanaExecutionObserver, + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("Devnet Memo execution failed: {error}"), + }; + assert!(summary.simulation.success); + if request.submit { + let post_execution = match summary.post_execution { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("Memo post-execution diagnostic missing"), + }; + assert!(post_execution.canonical_inserted); + assert!(post_execution.core_extracted); + assert!(post_execution.decode_replayed); + assert!(post_execution.materialized); + assert!(!summary.annotations.is_empty()); + let idempotence = match summary.idempotence_replay { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("Memo idempotence replay missing"), + }; + assert_eq!( + idempotence + .processors + .iter() + .map(|processor| return processor.materialized_outputs) + .sum::(), + 0 + ); + } + } +}