// file: ks-pipeline-demo-scenarios/src/spl/token/execution.rs // version: 17 //! Devnet classic SPL Token execution with stateful and canonical post-validation. /// Complete request for one Devnet classic SPL Token execution. #[derive(Clone, Debug, Eq, PartialEq)] pub struct DevnetSplTokenExecutionRequest { /// Stable caller-provided execution identifier. pub intent_id: std::string::String, /// Endpoint role used for state, 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 typed classic SPL Token operation. pub operation: ks_lib::ExSplClassicTokenOperation, /// 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 crate::DevnetSplTokenExecutionRequest { /// Creates a conservative simulation-only request. pub fn new( intent_id: impl std::convert::Into, operation: ks_lib::ExSplClassicTokenOperation, ) -> Self { return Self { intent_id: intent_id.into(), query_role: "http_queries".to_string(), transaction_role: "http_transactions".to_string(), operation, 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) -> ks_core::Result<()> { if self.intent_id.trim().is_empty() { return std::result::Result::Err(ks_core::Error::config( "Devnet SPL Token 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(ks_core::Error::config( "Devnet SPL Token execution endpoint roles must not be empty", )); } if self.post_validation_max_retries > 20 { return std::result::Result::Err(ks_core::Error::config( "post-execution getTransaction retries must not exceed 20", )); } return std::result::Result::Ok(()); } } /// Complete result of one Devnet classic SPL Token execution. #[derive(Clone, Debug, PartialEq)] pub struct DevnetSplTokenExecutionSummary { /// Profile used by the orchestration. pub profile_name: std::string::String, /// Exact classified cluster. pub cluster: ks_lib::ExApiExecutionCluster, /// Genesis hash returned by the selected endpoint. pub genesis_hash: std::string::String, /// Non-secret persistent wallet description. pub wallet: ks_wallet::WalletSummary, /// Wallet balance observed before planning. pub balance_lamports: u64, /// Stateful readiness report produced before plan simulation. pub stateful_readiness: ks_pipeline::SplTokenStatefulReadinessReport, /// Exact prepared Token plan. pub plan: ks_lib::ExApiPreparedExecutionPlan, /// Recent blockhash used by the exact transaction. pub latest_blockhash: ks_onchain_transport::LatestBlockhashResult, /// Fee estimate for the exact compiled message. pub fee: ks_onchain_transport::FeeForMessageResult, /// Exact simulation result bound to the compiled message. pub simulation: ks_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 Token decode and materialization replay. pub decode_replay: std::option::Option, /// Second replay proving idempotence for the same decoder version and input. pub idempotence_replay: std::option::Option, /// Exact materialized rows produced for the submitted Token transaction. pub materializations: std::vec::Vec, /// Aggregated post-execution validation diagnostic. pub post_execution: std::option::Option, } struct PreparedTokenExecution { unsigned: ks_lib::ExSolanaUnsignedTransaction, evidence: ks_lib::ExSolanaSimulationEvidence, summary: crate::DevnetSplTokenExecutionSummary, } /// Simulates one Devnet classic SPL Token operation after stateful preflight. pub async fn simulate_devnet_spl_token( http_pool: &ks_onchain_transport::HttpEndpointPool, profile: &ks_config::ProfileConfig, workspace_root: &std::path::Path, request: &crate::DevnetSplTokenExecutionRequest, observer: &O, ) -> ks_core::Result where O: crate::SolanaExecutionObserver, { if request.submit { return std::result::Result::Err(ks_core::Error::config( "simulate_devnet_spl_token requires submit=false", )); } let temporary = match crate::load_profile_temporary_wallet(profile, workspace_root).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let wallet = crate::DevnetExecutionWallet::Temporary(temporary); let prepared = match prepare_execution(http_pool, profile, &wallet, request, observer).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; return std::result::Result::Ok(prepared.summary); } /// Executes one Devnet classic SPL Token simulation or authorized submission. #[allow(clippy::too_many_arguments)] pub async fn execute_devnet_spl_token( http_pool: &ks_onchain_transport::HttpEndpointPool, store: &S, profile: &ks_config::ProfileConfig, workspace_root: &std::path::Path, request: &crate::DevnetSplTokenExecutionRequest, decoders: &[std::sync::Arc], materializers: &[std::sync::Arc], observer: &O, ) -> ks_core::Result where S: ks_store::RawTransactionStore + ks_store::CoreExtractionStore + ks_store::DecodePipelineStore + Sync, O: crate::SolanaExecutionObserver, { let temporary = match crate::load_profile_temporary_wallet(profile, workspace_root).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let wallet = crate::DevnetExecutionWallet::Temporary(temporary); return execute_devnet_spl_token_with_wallet( http_pool, store, profile, &wallet, request, decoders, materializers, observer, ) .await; } /// Executes this Devnet operation with an explicitly resolved wallet capability. #[allow(clippy::too_many_arguments)] pub async fn execute_devnet_spl_token_with_wallet( http_pool: &ks_onchain_transport::HttpEndpointPool, store: &S, profile: &ks_config::ProfileConfig, wallet: &crate::DevnetExecutionWallet, request: &crate::DevnetSplTokenExecutionRequest, decoders: &[std::sync::Arc], materializers: &[std::sync::Arc], observer: &O, ) -> ks_core::Result where S: ks_store::RawTransactionStore + ks_store::CoreExtractionStore + ks_store::DecodePipelineStore + Sync, O: crate::SolanaExecutionObserver, { let prepared = match prepare_execution(http_pool, profile, wallet, request, observer).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if !request.submit { return std::result::Result::Ok(prepared.summary); } if !prepared.summary.simulation.success { return std::result::Result::Err(ks_core::Error::new( "execution_simulation_failed", crate::simulation_failure_message(&prepared.summary.simulation), )); } if let std::result::Result::Err(error) = validate_profile_wallet_signers( prepared.unsigned.required_signer_pubkeys(), prepared.summary.wallet.public_key.as_str(), ) { return std::result::Result::Err(error); } let send_evaluation = match ks_lib::ExSafetyChecker .evaluate_send(&prepared.summary.plan, &prepared.summary.simulation) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if send_evaluation.decision == ks_lib::ExSafetyDecision::Deny { return std::result::Result::Err(ks_core::Error::new( "execution_send_denied", crate::violation_message(send_evaluation.violations.as_slice()), )); } let signed = match prepared .unsigned .sign_after_simulation(&prepared.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 summary = prepared.summary; let mut diagnostic = ks_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 ks_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 sent = match http_pool .send_transaction_for_role( request.transaction_role.as_str(), signed.transaction_base64().as_str(), &signature, &send_config, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { diagnostic.diagnostics.push(format!("SPL Token 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(ks_lib::ExApiExecutionCluster::Devnet)); let confirmation_config = match ks_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(), ks_lib::ExApiExecutionCluster::Devnet, &signature, &confirmation_config, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { diagnostic.diagnostics.push(format!("SPL Token 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, ks_lib::ExApiExecutionConfirmationStatus::Confirmed | ks_lib::ExApiExecutionConfirmationStatus::Finalized ) { diagnostic.diagnostics.push(format!( "SPL Token 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 crate::hydrate_signature( http_pool, store, profile, request.query_role.as_str(), request.post_validation_max_retries, observer, &signature, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { diagnostic.diagnostics.push(format!("SPL Token hydration failed: {error}")); summary.post_execution = std::option::Option::Some(diagnostic); return std::result::Result::Ok(summary); }, }; diagnostic.canonical_inserted = crate::canonical_available(&backfill); summary.backfill = std::option::Option::Some(backfill); if !diagnostic.canonical_inserted { diagnostic.diagnostics.push( "confirmed SPL Token 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 ks_pipeline::execute_core_extraction( store, &ks_pipeline::CoreExtractionRequest { source: ks_pipeline::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!("SPL Token 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("SPL Token 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 crate::replay_program( store, &signature, false, request.force_post_validation_replay, &[ks_program_ids::SPL_TOKEN_PROGRAM_ID], decoders, materializers, observer, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { diagnostic.diagnostics.push(format!("SPL Token decode replay failed: {error}")); summary.post_execution = std::option::Option::Some(diagnostic); return std::result::Result::Ok(summary); }, }; diagnostic.decode_replayed = crate::decode_completed(&first_replay); summary.decode_replay = std::option::Option::Some(first_replay); let filter = match ks_store::MaterializedEventFilter::new( std::option::Option::None, std::option::Option::None, std::option::Option::Some(signature.0.clone()), 64, ) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let rows = match ks_store::DecodePipelineStore::list_materialized_events(store, &filter).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { diagnostic .diagnostics .push(format!("SPL Token materialization query failed: {error}")); summary.post_execution = std::option::Option::Some(diagnostic); return std::result::Result::Ok(summary); }, }; summary.materializations = rows .into_iter() .filter(|row| return row.source_decoder_name == "spl.token") .collect(); diagnostic.materialized = !summary.plan.policy.post_execution_validation.materialization_required || !summary.materializations.is_empty(); let second_replay = match crate::replay_program( store, &signature, true, request.force_post_validation_replay, &[ks_program_ids::SPL_TOKEN_PROGRAM_ID], decoders, materializers, observer, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { diagnostic .diagnostics .push(format!("SPL Token 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.processing_error_inputs == 0 && second_replay.processors.iter().all(|processor| { return processor.failed == 0 && processor.processing_errors == 0 && processor.materialized_outputs == 0 && processor.materialization_refused == 0; }); summary.idempotence_replay = std::option::Option::Some(second_replay); if !idempotent { diagnostic .diagnostics .push("second SPL Token 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( "SPL Token completed canonical hydration, core extraction, decode, materialization and idempotence validation" .to_string(), ); } summary.post_execution = std::option::Option::Some(diagnostic); return std::result::Result::Ok(summary); } async fn prepare_execution( http_pool: &ks_onchain_transport::HttpEndpointPool, profile: &ks_config::ProfileConfig, wallet: &crate::DevnetExecutionWallet, request: &crate::DevnetSplTokenExecutionRequest, observer: &O, ) -> ks_core::Result where 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, "token_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(ks_lib::ExApiExecutionCluster::Devnet) { return std::result::Result::Err(ks_core::Error::new( "execution_cluster_mismatch", format!( "expected Devnet genesis hash but endpoint returned {} classified as {:?}", genesis.genesis_hash, genesis.classified_cluster ), )); } let readiness = match ks_pipeline::inspect_spl_token_stateful_readiness( http_pool, &ks_pipeline::SplTokenStatefulReadinessRequest { query_role: request.query_role.clone(), cluster: ks_lib::ExApiExecutionCluster::Devnet, operation: request.operation.clone(), }, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if readiness.status != ks_pipeline::SplTokenStatefulReadinessStatus::Ready { return std::result::Result::Err(ks_core::Error::new( "execution_spl_token_stateful_blocked", failed_readiness_message(&readiness), )); } let wallet_summary = wallet.identity(); let fee_payer = ks_lib::MdPubkey(wallet_summary.public_key.clone()); let balance = match http_pool .get_balance_for_role( request.query_role.as_str(), &fee_payer, &ks_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(ks_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 ks_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 == ks_lib::ExSafetyDecision::Deny { return std::result::Result::Err(ks_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(), &ks_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 ks_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 fee = match http_pool .get_fee_for_message_for_role( request.query_role.as_str(), unsigned.message_base64().as_str(), &ks_onchain_transport::GetFeeForMessageConfig::new( ks_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(ks_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 ks_onchain_transport::SimulateTransactionConfig::new( ks_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, "spl_token_simulation", format!("simulating exact SPL Token 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( ks_lib::ExApiExecutionCluster::Devnet, ks_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 summary = crate::DevnetSplTokenExecutionSummary { profile_name: profile.name.clone(), cluster: ks_lib::ExApiExecutionCluster::Devnet, genesis_hash: genesis.genesis_hash, wallet: wallet_summary, balance_lamports: balance.lamports, stateful_readiness: readiness, 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, materializations: std::vec::Vec::new(), post_execution: std::option::Option::None, }; return std::result::Result::Ok(PreparedTokenExecution { unsigned, evidence, summary }); } fn validate_profile( profile: &ks_config::ProfileConfig, request: &crate::DevnetSplTokenExecutionRequest, ) -> ks_core::Result<()> { if profile.wallet.cluster != "devnet" { return std::result::Result::Err(ks_core::Error::config( "SPL Token Devnet orchestration requires a Devnet wallet profile", )); } if profile.wallet.wallet_alias.is_none() && (!profile.wallet.temporary_wallet_enabled || !profile.wallet.temporary_wallet_persist) { return std::result::Result::Err(ks_core::Error::config( "SPL Token Devnet orchestration requires an enabled persistent temporary wallet when no native wallet alias is selected", )); } if !profile.execution.require_simulation { return std::result::Result::Err(ks_core::Error::config( "SPL Token Devnet orchestration requires simulation", )); } if request.submit && !profile.execution.devnet_send_enabled { return std::result::Result::Err(ks_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(ks_core::Error::config( "SPL Token Devnet submission requires explicit operator confirmation", )); } return std::result::Result::Ok(()); } fn build_plan( profile: &ks_config::ProfileConfig, request: &crate::DevnetSplTokenExecutionRequest, fee_payer: ks_lib::MdPubkey, ) -> ks_core::Result { let materialization_required = operation_requires_materialization(&request.operation); let mut authorized_signers = std::vec![fee_payer.clone()]; collect_operation_signers(&request.operation, &mut authorized_signers); let intent = ks_lib::ExSplClassicTokenExecutionIntent { intent_id: request.intent_id.clone(), fee_payer, policy: ks_lib::ExApiExecutionPolicy { cluster: ks_lib::ExApiExecutionClusterPolicy { expected_cluster: ks_lib::ExApiExecutionCluster::Devnet, allow_mainnet: false, mainnet_confirmation: false, }, simulation: ks_lib::ExApiExecutionSimulationPolicy::Required, blockhash: ks_lib::ExApiExecutionBlockhashPolicy { kind: ks_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: ks_lib::ExApiExecutionCostLimit { max_spend_lamports: std::option::Option::Some( profile.execution.devnet_max_spend_lamports, ), 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, dry_run: !request.submit, post_execution_validation: ks_lib::ExApiPostExecutionValidationPolicy { canonical_insert_required: true, core_extraction_required: true, decode_replay_required: true, materialization_required, }, }, operation: request.operation.clone(), }; return ks_lib::ExApiTypedInstructionExecutor::build_prepared_plan( &ks_lib::ExSplTokenExecutor, &intent, ); } fn collect_operation_signers( operation: &ks_lib::ExSplClassicTokenOperation, signers: &mut std::vec::Vec, ) { match operation { ks_lib::ExSplClassicTokenOperation::Instruction { value } => { collect_single_signers(value, signers); }, ks_lib::ExSplClassicTokenOperation::Batch { instructions } => { for value in instructions { collect_single_signers(value, signers); } }, } } fn collect_single_signers( operation: &ks_lib::ExSplClassicTokenSingleOperation, signers: &mut std::vec::Vec, ) { let authority = match operation { ks_lib::ExSplClassicTokenSingleOperation::Transfer { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::Approve { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::Revoke { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::MintTo { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::Burn { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::CloseAccount { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::FreezeAccount { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::ThawAccount { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::TransferChecked { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::ApproveChecked { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::MintToChecked { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::BurnChecked { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::WithdrawExcessLamports { authority, .. } | ks_lib::ExSplClassicTokenSingleOperation::UnwrapLamports { authority, .. } => { std::option::Option::Some(authority) }, ks_lib::ExSplClassicTokenSingleOperation::SetAuthority { current_authority, .. } => { std::option::Option::Some(current_authority) }, ks_lib::ExSplClassicTokenSingleOperation::InitializeMint { .. } | ks_lib::ExSplClassicTokenSingleOperation::InitializeAccount { .. } | ks_lib::ExSplClassicTokenSingleOperation::InitializeMultisig { .. } | ks_lib::ExSplClassicTokenSingleOperation::SyncNative { .. } | ks_lib::ExSplClassicTokenSingleOperation::GetAccountDataSize { .. } | ks_lib::ExSplClassicTokenSingleOperation::InitializeImmutableOwner { .. } | ks_lib::ExSplClassicTokenSingleOperation::AmountToUiAmount { .. } | ks_lib::ExSplClassicTokenSingleOperation::UiAmountToAmount { .. } => { std::option::Option::None }, }; if let std::option::Option::Some(authority) = authority { if authority.multisig_signers.is_empty() { add_signer(signers, &authority.authority); } else { for signer in &authority.multisig_signers { add_signer(signers, signer); } } } } fn add_signer(signers: &mut std::vec::Vec, signer: &ks_lib::MdPubkey) { if !signers.iter().any(|value| return value == signer) { signers.push(signer.clone()); } } fn operation_requires_materialization(operation: &ks_lib::ExSplClassicTokenOperation) -> bool { return match operation { ks_lib::ExSplClassicTokenOperation::Instruction { value } => { single_requires_materialization(value) }, ks_lib::ExSplClassicTokenOperation::Batch { instructions } => { instructions.iter().any(|value| { return single_requires_materialization(value); }) }, }; } fn single_requires_materialization(operation: &ks_lib::ExSplClassicTokenSingleOperation) -> bool { return !matches!( operation, ks_lib::ExSplClassicTokenSingleOperation::GetAccountDataSize { .. } | ks_lib::ExSplClassicTokenSingleOperation::InitializeImmutableOwner { .. } | ks_lib::ExSplClassicTokenSingleOperation::AmountToUiAmount { .. } | ks_lib::ExSplClassicTokenSingleOperation::UiAmountToAmount { .. } ); } fn failed_readiness_message( readiness: &ks_pipeline::SplTokenStatefulReadinessReport, ) -> std::string::String { let failures = readiness .checks .iter() .filter(|check| return !check.passed) .take(8) .map(|check| return format!("{}: {}", check.code, check.message)) .collect::>(); if failures.is_empty() { return "SPL Token stateful readiness returned Blocked without diagnostics".to_string(); } return failures.join("; "); } fn validate_profile_wallet_signers( required_signers: &[std::string::String], wallet_pubkey: &str, ) -> ks_core::Result<()> { if required_signers.iter().any(|value| return value.as_str() != wallet_pubkey) { return std::result::Result::Err(ks_core::Error::new( "execution_spl_token_external_signer_unavailable", format!( "the Devnet Token orchestrator can sign only with profile wallet {}; required signers are {}", wallet_pubkey, required_signers.join(",") ), )); } return std::result::Result::Ok(()); } pub(crate) async fn hydrate_signature( http_pool: &ks_onchain_transport::HttpEndpointPool, store: &S, profile: &ks_config::ProfileConfig, query_role: &str, post_validation_max_retries: u32, observer: &O, signature: &ks_lib::MdSignature, ) -> ks_core::Result where S: ks_store::RawTransactionStore + Sync, O: crate::SolanaExecutionObserver, { let mut retry = 0_u32; loop { let result = match ks_pipeline::execute_http_backfill( http_pool, store, &ks_pipeline::BackfillRequest { role: query_role.to_string(), commitment: "confirmed".to_string(), source: ks_pipeline::BackfillSource::ExplicitSignatures(std::vec![ signature.0.clone() ]), page_size: 1, max_pages: 1, max_concurrent_requests: 1, max_retries: 2, }, observer, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if crate::canonical_available(&result) || retry >= 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; } } pub(crate) fn canonical_available(summary: &ks_pipeline::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; } pub(crate) async fn replay_program( store: &S, signature: &ks_lib::MdSignature, include_materialized_state: bool, force_post_validation_replay: bool, program_ids: &[&str], decoders: &[std::sync::Arc], materializers: &[std::sync::Arc], observer: &O, ) -> ks_core::Result where S: ks_store::DecodePipelineStore + Sync, O: crate::SolanaExecutionObserver, { let mut states = std::vec![ ks_store::CoreInstructionProcessingState::Pending, ks_store::CoreInstructionProcessingState::Failed, ks_store::CoreInstructionProcessingState::ReplayRequested, ]; if include_materialized_state { states.push(ks_store::CoreInstructionProcessingState::Materialized); } let selection = match ks_store::DecodeSelectionFilter::new( std::vec![signature.0.clone()], states, std::option::Option::None, std::option::Option::None, program_ids.iter().map(|value| return (*value).to_string()).collect(), std::vec::Vec::new(), false, 64, ) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; return ks_pipeline::execute_decode_replay( store, &ks_pipeline::DecodeReplayRequest { campaign_id: ks_pipeline::new_decode_campaign_id(), selection, decoder_names: std::vec::Vec::new(), dispatch_policy: ks_pipeline::DecodeDispatchPolicy::HighestPriority, max_concurrent_inputs: 1, force_replay: if include_materialized_state { false } else { force_post_validation_replay }, force_replay_all_matching: false, materialize_after_decode: true, }, decoders, materializers, observer, ) .await; } pub(crate) fn decode_completed(summary: &ks_pipeline::DecodeReplaySummary) -> bool { return summary.failed_inputs == 0 && summary.processing_error_inputs == 0 && summary.unmatched == 0 && !summary.cancelled && summary.completed >= 1 && summary.processors.iter().all(|processor| { return processor.failed == 0 && processor.processing_errors == 0 && processor.unsupported == 0 && processor.materialization_refused == 0; }) && summary.processors.iter().map(|processor| return processor.decoded).sum::() >= 1; } #[cfg(test)] mod tests { fn example_devnet_profile() -> ks_config::ProfileConfig { let config = match ks_config::parse_config_json(include_str!( "../../../../test-fixtures/config/resolved.app.config.json" )) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("example config parse failed: {error}"), }; return match crate::resolve_demo_devnet_profile(&config, std::option::Option::None) { std::result::Result::Ok(profile) => profile, std::result::Result::Err(error) => panic!("Devnet profile resolution failed: {error}"), }; } fn checked_transfer(fee_authority: &str) -> ks_lib::ExSplClassicTokenOperation { return ks_lib::ExSplClassicTokenOperation::Instruction { value: ks_lib::ExSplClassicTokenSingleOperation::TransferChecked { source: ks_lib::MdPubkey(ks_program_ids::VOTE_PROGRAM_ID.to_string()), mint: ks_lib::MdPubkey(ks_program_ids::STAKE_PROGRAM_ID.to_string()), destination: ks_lib::MdPubkey(ks_program_ids::CONFIG_PROGRAM_ID.to_string()), authority: ks_lib::ExSplClassicTokenAuthority { authority: ks_lib::MdPubkey(fee_authority.to_string()), multisig_signers: std::vec::Vec::new(), }, amount: ks_lib::ExSplClassicTokenAmount("1".to_string()), decimals: 9, }, }; } #[test] fn request_is_simulation_only_and_bounded_by_default() { let request = crate::DevnetSplTokenExecutionRequest::new( "token-1", checked_transfer(ks_program_ids::SYSTEM_PROGRAM_ID), ); assert!(!request.submit); assert!(request.validate().is_ok()); let mut invalid = request; invalid.post_validation_max_retries = 21; assert!(invalid.validate().is_err()); } #[test] fn checked_transfer_plan_is_simulation_first_and_authorizes_exact_signers() { let profile = example_devnet_profile(); let fee_payer = ks_lib::MdPubkey(ks_program_ids::SYSTEM_PROGRAM_ID.to_string()); let authority = ks_program_ids::VOTE_PROGRAM_ID; let request = crate::DevnetSplTokenExecutionRequest::new("token-2", checked_transfer(authority)); let plan = match super::build_plan(&profile, &request, fee_payer.clone()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("Token plan failed: {error}"), }; assert!(plan.policy.dry_run); assert_eq!(plan.fee_payer, fee_payer); assert!(plan.policy.authorized_signers.iter().any(|value| return value.0 == authority)); assert!(plan.policy.post_execution_validation.materialization_required); assert_eq!(plan.instructions.len(), 1); } #[test] fn conversion_plan_does_not_invent_materialization_requirement() { let profile = example_devnet_profile(); let request = crate::DevnetSplTokenExecutionRequest::new( "token-3", ks_lib::ExSplClassicTokenOperation::Instruction { value: ks_lib::ExSplClassicTokenSingleOperation::AmountToUiAmount { mint: ks_lib::MdPubkey(ks_program_ids::STAKE_PROGRAM_ID.to_string()), amount: ks_lib::ExSplClassicTokenAmount("1".to_string()), }, }, ); let plan = match super::build_plan( &profile, &request, ks_lib::MdPubkey(ks_program_ids::SYSTEM_PROGRAM_ID.to_string()), ) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("conversion plan failed: {error}"), }; assert!(!plan.policy.post_execution_validation.materialization_required); } #[test] fn submission_signers_must_all_resolve_to_the_profile_wallet() { let wallet = ks_program_ids::SYSTEM_PROGRAM_ID.to_string(); assert!( super::validate_profile_wallet_signers(std::slice::from_ref(&wallet), wallet.as_str(),) .is_ok() ); let external = ks_program_ids::VOTE_PROGRAM_ID.to_string(); let error = super::validate_profile_wallet_signers(&[wallet.clone(), external], wallet.as_str()); assert!(error.is_err()); } #[tokio::test] async fn optional_devnet_checked_transfer_simulation_and_submission_from_env() { if std::env::var("KS_DEVNET_SPL_TOKEN_EXECUTION_TEST").ok().as_deref() != std::option::Option::Some("1") { return; } let source = required_pubkey_env("KS_DEVNET_SPL_TOKEN_SOURCE"); let mint = required_pubkey_env("KS_DEVNET_SPL_TOKEN_MINT"); let destination = required_pubkey_env("KS_DEVNET_SPL_TOKEN_DESTINATION"); let authority = required_pubkey_env("KS_DEVNET_SPL_TOKEN_AUTHORITY"); let decimals = match std::env::var("KS_DEVNET_SPL_TOKEN_DECIMALS") { std::result::Result::Ok(value) => match value.parse::() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("invalid Token decimals: {error}"), }, std::result::Result::Err(_) => 9, }; let amount = match std::env::var("KS_DEVNET_SPL_TOKEN_AMOUNT") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => "1".to_string(), }; let mut profile = example_devnet_profile(); if let std::result::Result::Ok(directory) = std::env::var("KS_DEVNET_WALLET_DIR") { profile.wallet.temporary_wallet_dir = directory; } let pool = match ks_onchain_transport::HttpEndpointPool::from_profile(&profile) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("HTTP pool creation failed: {error}"), }; 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 mut request = crate::DevnetSplTokenExecutionRequest::new( format!("devnet-token-test-{}", uuid::Uuid::new_v4()), ks_lib::ExSplClassicTokenOperation::Instruction { value: ks_lib::ExSplClassicTokenSingleOperation::TransferChecked { source, mint, destination, authority: ks_lib::ExSplClassicTokenAuthority { authority, multisig_signers: std::vec::Vec::new(), }, amount: ks_lib::ExSplClassicTokenAmount(amount), decimals, }, }, ); if std::env::var("KS_DEVNET_SPL_TOKEN_SUBMIT").ok().as_deref() != std::option::Option::Some("1") { let summary = match crate::simulate_devnet_spl_token( &pool, &profile, workspace_root, &request, &crate::NoopSolanaExecutionObserver, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { panic!("Devnet Token simulation failed: {error}"); }, }; assert_eq!( summary.stateful_readiness.status, ks_pipeline::SplTokenStatefulReadinessStatus::Ready ); assert!(summary.simulation.success, "{:#?}", summary.simulation); return; } request.submit = true; request.operator_confirmed = true; request.post_validation_max_retries = 20; let database_url = match std::env::var("KS_SECRET_POSTGRES_TEST_URL") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { panic!("KS_SECRET_POSTGRES_TEST_URL is required for submission: {error}"); }, }; profile.database.backend = "postgres".to_string(); profile.database.postgres.url = database_url; let store_options = match ks_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 ks_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 decoders: std::vec::Vec> = std::vec![std::sync::Arc::new(ks_lib::DcSplTokenDecoder)]; let materializers: std::vec::Vec> = std::vec![std::sync::Arc::new(ks_lib::MtTokenAccountsMaterializer,)]; let summary = match crate::execute_devnet_spl_token( &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 Token execution failed: {error}"), }; assert!(summary.simulation.success, "{:#?}", summary.simulation); let send_result = match summary.send_result.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => panic!("Token submission result missing"), }; assert!(send_result.submitted); let confirmation = match summary.confirmation.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => panic!("Token confirmation result missing"), }; assert!(matches!( confirmation.status, ks_lib::ExApiExecutionConfirmationStatus::Confirmed | ks_lib::ExApiExecutionConfirmationStatus::Finalized )); let first_replay = match summary.decode_replay.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => panic!("Token first decode replay missing"), }; assert_eq!(first_replay.failed_inputs, 0); assert!(first_replay.processors.iter().all(|processor| { return processor.failed == 0 && processor.processing_errors == 0 && processor.materialization_refused == 0; })); let idempotence_replay = match summary.idempotence_replay.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => panic!("Token idempotence replay missing"), }; assert_eq!(idempotence_replay.failed_inputs, 0); assert!(idempotence_replay.processors.iter().all(|processor| { return processor.failed == 0 && processor.processing_errors == 0 && processor.materialized_outputs == 0 && processor.materialization_refused == 0; })); assert!(!summary.materializations.is_empty()); let diagnostic = match summary.post_execution.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => panic!("Token post-execution diagnostic missing"), }; assert!(diagnostic.canonical_inserted); assert!(diagnostic.core_extracted); assert!(diagnostic.decode_replayed); assert!(diagnostic.materialized); println!( "SPL Token Devnet signature={} status={:?} materializations={} idempotent=true", send_result.signature.0, confirmation.status, summary.materializations.len(), ); } #[tokio::test] async fn optional_devnet_recent_instruction_probe_from_env() { if std::env::var("KS_DEVNET_SPL_TOKEN_RECENT_PROBE_TEST").ok().as_deref() != std::option::Option::Some("1") { return; } let source = required_pubkey_env("KS_DEVNET_SPL_TOKEN_SOURCE"); let mint = required_pubkey_env("KS_DEVNET_SPL_TOKEN_MINT"); let destination = required_pubkey_env("KS_DEVNET_SPL_TOKEN_DESTINATION"); let authority = required_pubkey_env("KS_DEVNET_SPL_TOKEN_AUTHORITY"); let native_account = required_pubkey_env("KS_DEVNET_SPL_TOKEN_NATIVE_ACCOUNT"); let decimals = match std::env::var("KS_DEVNET_SPL_TOKEN_DECIMALS") { std::result::Result::Ok(value) => match value.parse::() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("invalid Token decimals: {error}"), }, std::result::Result::Err(_) => 9, }; let mut profile = example_devnet_profile(); if let std::result::Result::Ok(directory) = std::env::var("KS_DEVNET_WALLET_DIR") { profile.wallet.temporary_wallet_dir = directory; } let pool = match ks_onchain_transport::HttpEndpointPool::from_profile(&profile) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("HTTP pool creation failed: {error}"), }; 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 token_authority = ks_lib::ExSplClassicTokenAuthority { authority: authority.clone(), multisig_signers: std::vec::Vec::new(), }; let batch_request = crate::DevnetSplTokenExecutionRequest::new( format!("devnet-token-batch-probe-{}", uuid::Uuid::new_v4()), ks_lib::ExSplClassicTokenOperation::Batch { instructions: std::vec![ ks_lib::ExSplClassicTokenSingleOperation::TransferChecked { source, mint, destination, authority: token_authority.clone(), amount: ks_lib::ExSplClassicTokenAmount("0".to_string()), decimals, }, ], }, ); let batch = match crate::simulate_devnet_spl_token( &pool, &profile, workspace_root, &batch_request, &crate::NoopSolanaExecutionObserver, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("Devnet Batch probe failed: {error}"), }; assert!(batch.simulation.simulated); print_recent_probe("batch", &batch.simulation); let unwrap_request = crate::DevnetSplTokenExecutionRequest::new( format!("devnet-token-unwrap-probe-{}", uuid::Uuid::new_v4()), ks_lib::ExSplClassicTokenOperation::Instruction { value: ks_lib::ExSplClassicTokenSingleOperation::UnwrapLamports { account: native_account, destination: authority, authority: token_authority, amount_lamports: std::option::Option::Some(ks_lib::ExSplClassicTokenAmount( "1".to_string(), )), }, }, ); let unwrap = match crate::simulate_devnet_spl_token( &pool, &profile, workspace_root, &unwrap_request, &crate::NoopSolanaExecutionObserver, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { panic!("Devnet UnwrapLamports probe failed: {error}") }, }; assert!(unwrap.simulation.simulated); print_recent_probe("unwrap_lamports", &unwrap.simulation); } fn print_recent_probe(operation: &str, simulation: &ks_lib::ExApiExecutionSimulationResult) { println!( "SPL Token recent Devnet probe operation={operation} success={} error={:?} units={:?} logs={:?}", simulation.success, simulation.error, simulation.units_consumed, simulation.logs, ); } fn required_pubkey_env(name: &str) -> ks_lib::MdPubkey { return match std::env::var(name) { std::result::Result::Ok(value) => ks_lib::MdPubkey(value), std::result::Result::Err(error) => panic!("{name} is required: {error}"), }; } }