// file: ks-pipeline-demo-scenarios/src/solana.rs // version: 14 //! Devnet Solana execution orchestration with canonical post-validation. /// Progress severity for one Solana execution orchestration. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum SolanaExecutionProgressLevel { /// Diagnostic detail. Debug, /// Normal execution information. Info, /// Recoverable issue or non-terminal limitation. Warning, /// Terminal execution or validation failure. Error, } impl crate::SolanaExecutionProgressLevel { /// Returns the stable lowercase level code. pub fn code(&self) -> &'static str { return match self { Self::Debug => "debug", Self::Info => "info", Self::Warning => "warn", Self::Error => "error", }; } } /// One operator-visible execution progress event. #[derive(Clone, Debug, Eq, PartialEq)] pub struct SolanaExecutionProgressEvent { /// UTC timestamp rendered in RFC 3339. pub timestamp: std::string::String, /// Severity. pub level: crate::SolanaExecutionProgressLevel, /// Stable execution stage code. pub stage: std::string::String, /// Human-readable message. pub message: std::string::String, /// Transaction signature when available. pub signature: std::option::Option, } impl crate::SolanaExecutionProgressEvent { fn new( level: crate::SolanaExecutionProgressLevel, stage: impl std::convert::Into, message: impl std::convert::Into, signature: std::option::Option, ) -> Self { return Self { timestamp: chrono::Utc::now().to_rfc3339(), level, stage: stage.into(), message: message.into(), signature, }; } } /// Composite observer used by execution, backfill, extraction and decode replay. pub trait SolanaExecutionObserver: ks_pipeline::BackfillObserver + ks_pipeline::CoreExtractionObserver + ks_pipeline::DecodeReplayObserver + Sync { /// Receives one execution-specific progress event. fn on_execution_progress(&self, event: &crate::SolanaExecutionProgressEvent); /// Returns true when execution orchestration should stop before the next stage. fn is_execution_cancelled(&self) -> bool; } /// No-op observer suitable for CLI tools and opt-in integration tests. #[derive(Clone, Copy, Debug, Default)] pub struct NoopSolanaExecutionObserver; impl ks_pipeline::BackfillObserver for crate::NoopSolanaExecutionObserver { fn on_progress(&self, _event: &ks_pipeline::BackfillProgressEvent) { return; } fn is_cancelled(&self) -> bool { return false; } } impl ks_pipeline::CoreExtractionObserver for crate::NoopSolanaExecutionObserver { fn on_progress(&self, _event: &ks_pipeline::CoreExtractionProgressEvent) { return; } fn is_cancelled(&self) -> bool { return false; } } impl ks_pipeline::DecodeReplayObserver for crate::NoopSolanaExecutionObserver { fn on_progress(&self, _event: &ks_pipeline::DecodeReplayProgressEvent) { return; } fn is_cancelled(&self) -> bool { return false; } } impl crate::SolanaExecutionObserver for crate::NoopSolanaExecutionObserver { fn on_execution_progress(&self, _event: &crate::SolanaExecutionProgressEvent) { return; } fn is_execution_cancelled(&self) -> bool { return false; } } /// Complete request for one bounded System Program transfer on Devnet. #[derive(Clone, Debug, Eq, PartialEq)] pub struct DevnetSystemTransferRequest { /// Stable caller-provided identifier used in tracing and plan correlation. pub intent_id: std::string::String, /// Endpoint role used for cluster, blockhash, fee, balance, airdrop and hydration calls. pub query_role: std::string::String, /// Endpoint role used for simulation, submission and signature status polling. pub transaction_role: std::string::String, /// Recipient public key. pub recipient: ks_lib::MdPubkey, /// Lamports transferred by the System instruction. pub lamports: u64, /// Maximum faucet airdrop requested only when the source balance is insufficient. pub airdrop_lamports: u64, /// Explicitly authorizes signing and submission instead of returning after simulation. pub submit: bool, /// Explicit operator confirmation required by the active profile when configured. pub operator_confirmed: bool, /// Number of `getTransaction` retries after the first post-confirmation attempt. pub post_validation_max_retries: u32, /// Replaces existing core/decode outputs for the exact submitted signature. pub force_post_validation_replay: bool, /// Runs compatible materializers after decode replay. pub materialize_after_decode: bool, } impl crate::DevnetSystemTransferRequest { /// Creates a conservative simulation-only request. pub fn new( intent_id: impl std::convert::Into, recipient: ks_lib::MdPubkey, lamports: u64, ) -> Self { return Self { intent_id: intent_id.into(), query_role: "http_queries".to_string(), transaction_role: "http_transactions".to_string(), recipient, lamports, airdrop_lamports: 0, submit: false, operator_confirmed: false, post_validation_max_retries: 10, force_post_validation_replay: false, materialize_after_decode: false, }; } /// Validates request-local values 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 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 execution endpoint roles must not be empty", )); } let recipient_result = ks_onchain_transport::validate_solana_pubkey_text( self.recipient.0.as_str(), "devnet System transfer recipient", ); if let std::result::Result::Err(error) = recipient_result { return std::result::Result::Err(error); } if self.lamports == 0 { return std::result::Result::Err(ks_core::Error::config( "devnet System transfer lamports must be greater than zero", )); } 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 transfer orchestration. #[derive(Clone, Debug, Eq, PartialEq)] pub struct DevnetSystemTransferSummary { /// Profile used by the orchestration. pub profile_name: std::string::String, /// 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, /// Transfer recipient. pub recipient: ks_lib::MdPubkey, /// Whether the recipient account existed before execution. pub recipient_existed_before: bool, /// Recipient balance before execution when the account existed. pub recipient_balance_before_lamports: std::option::Option, /// Minimum lamports required when creating a zero-data recipient account. pub recipient_minimum_balance_lamports: u64, /// Source wallet balance before an optional airdrop. pub balance_before_lamports: u64, /// Source wallet balance after optional funding. pub balance_after_funding_lamports: u64, /// Faucet signature when an airdrop was required. pub airdrop_signature: std::option::Option, /// Airdrop confirmation when an airdrop was required. pub airdrop_confirmation: std::option::Option, /// Exact prepared plan. pub plan: ks_lib::ExApiPreparedExecutionPlan, /// Recent blockhash and expiration height used by the 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 `submit` was enabled. pub send_result: std::option::Option, /// Confirmation result when the transaction was submitted. pub confirmation: std::option::Option, /// Canonical hydration result after an observed transaction reached a terminal status. pub backfill: std::option::Option, /// Core extraction result for the exact signature. pub core_extraction: std::option::Option, /// Contextual decode result for the exact signature. pub decode_replay: std::option::Option, /// Aggregated post-execution validation diagnostic. pub post_execution: std::option::Option, } /// Executes a bounded Devnet System transfer and its canonical replay validation. #[allow(clippy::too_many_arguments)] pub async fn execute_devnet_system_transfer( http_pool: &ks_onchain_transport::HttpEndpointPool, store: &S, profile: &ks_config::ProfileConfig, workspace_root: &std::path::Path, request: &crate::DevnetSystemTransferRequest, 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 request_result = request.validate(); if let std::result::Result::Err(error) = request_result { return std::result::Result::Err(error); } let profile_result = validate_devnet_profile(profile, request); if let std::result::Result::Err(error) = profile_result { return std::result::Result::Err(error); } let cancellation_result = ensure_not_cancelled(observer, "validate"); if let std::result::Result::Err(error) = cancellation_result { return std::result::Result::Err(error); } emit( observer, crate::SolanaExecutionProgressLevel::Info, "cluster_check", format!("checking Devnet cluster for profile {}", profile.name), std::option::Option::None, ); 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 cancellation_result = ensure_not_cancelled(observer, "wallet"); if let std::result::Result::Err(error) = cancellation_result { return std::result::Result::Err(error); } let wallet = match load_profile_wallet(profile, workspace_root).await { std::result::Result::Ok(wallet) => wallet, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let wallet_summary = wallet.summary(); let source_pubkey = ks_lib::MdPubkey(wallet_summary.public_key.clone()); if source_pubkey == request.recipient { return std::result::Result::Err(ks_core::Error::config( "devnet System transfer recipient must differ from the source wallet", )); } let account_info_config = ks_onchain_transport::GetAccountInfoConfig::confirmed(); let recipient_account = match http_pool .get_account_info_for_role( request.query_role.as_str(), &request.recipient, &account_info_config, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let recipient_existed_before = recipient_account.account.is_some(); let recipient_balance_before_lamports = recipient_account.account.as_ref().map(|account| return account.lamports); let recipient_minimum_balance_lamports = if recipient_existed_before { 0 } else { let rent_config = ks_onchain_transport::GetMinimumBalanceForRentExemptionConfig::confirmed(); let rent = match http_pool .get_minimum_balance_for_rent_exemption_for_role( request.query_role.as_str(), 0, &rent_config, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; rent.minimum_balance_lamports }; let recipient_validation = validate_recipient_transfer_amount( recipient_existed_before, request.lamports, recipient_minimum_balance_lamports, ); if let std::result::Result::Err(error) = recipient_validation { return std::result::Result::Err(error); } let balance_config = ks_onchain_transport::GetBalanceConfig::confirmed(); let balance_before = match http_pool .get_balance_for_role(request.query_role.as_str(), &source_pubkey, &balance_config) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let required_balance = match request.lamports.checked_add(profile.execution.max_fee_lamports) { std::option::Option::Some(value) => value, std::option::Option::None => { return std::result::Result::Err(ks_core::Error::config( "devnet transfer amount plus fee ceiling overflows u64", )); }, }; let funding = match ensure_devnet_funding( http_pool, profile, request, observer, &source_pubkey, balance_before.lamports, required_balance, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let cancellation_result = ensure_not_cancelled(observer, "plan"); if let std::result::Result::Err(error) = cancellation_result { return std::result::Result::Err(error); } let plan = match build_transfer_plan(profile, request, source_pubkey.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", violation_message(plan_evaluation.violations.as_slice()), )); } let latest_config = ks_onchain_transport::GetLatestBlockhashConfig::confirmed(); let latest_blockhash = match http_pool .get_latest_blockhash_for_role(request.query_role.as_str(), &latest_config) .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_config = ks_onchain_transport::GetFeeForMessageConfig::new( ks_onchain_transport::RpcCommitmentLevel::Confirmed, std::option::Option::Some(latest_blockhash.context.slot), ); 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(), &fee_config, ) .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), }; emit( observer, crate::SolanaExecutionProgressLevel::Info, "simulation", format!("simulating exact 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 blockhash_age_slots = simulation_rpc.context.slot.saturating_sub(latest_blockhash.context.slot); let simulation = simulation_rpc.to_execution_result( ks_lib::ExApiExecutionCluster::Devnet, ks_lib::ExApiExecutionBlockhashKind::Latest, std::option::Option::Some(blockhash_age_slots), std::option::Option::None, std::option::Option::None, std::option::Option::Some(&fee), ); let evidence = unsigned.bind_simulation(simulation.clone()); let mut summary = crate::DevnetSystemTransferSummary { profile_name: profile.name.clone(), cluster: ks_lib::ExApiExecutionCluster::Devnet, genesis_hash: genesis.genesis_hash, wallet: wallet_summary, recipient: request.recipient.clone(), recipient_existed_before, recipient_balance_before_lamports, recipient_minimum_balance_lamports, balance_before_lamports: balance_before.lamports, balance_after_funding_lamports: funding.balance_after_lamports, airdrop_signature: funding.airdrop_signature, airdrop_confirmation: funding.airdrop_confirmation, 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, post_execution: std::option::Option::None, }; if !request.submit { return std::result::Result::Ok(summary); } if !summary.simulation.success { return std::result::Result::Err(ks_core::Error::new( "execution_simulation_failed", simulation_failure_message(&summary.simulation), )); } let cancellation_result = ensure_not_cancelled(observer, "sign"); if let std::result::Result::Err(error) = cancellation_result { return std::result::Result::Err(error); } let signed = match evidence.message_hash() == unsigned.message_hash() { true => 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), }, false => { return std::result::Result::Err(ks_core::Error::new( "execution_simulation_message_mismatch", "simulation evidence does not match the unsigned transaction", )); }, }; let verify_result = signed.verify_signatures(); if let std::result::Result::Err(error) = verify_result { return std::result::Result::Err(error); } let signature = signed.primary_signature().clone(); let mut post_execution = 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), }; emit( observer, crate::SolanaExecutionProgressLevel::Info, "send", "submitting signed Devnet transaction", std::option::Option::Some(signature.0.clone()), ); let signed_transaction_base64 = signed.transaction_base64(); 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) => { post_execution .diagnostics .push(format!("sendTransaction failed for locally signed transaction: {error}")); summary.post_execution = std::option::Option::Some(post_execution); tracing::error!( target: crate::TRACING_TARGET, action = "execute_devnet_system_transfer", stage = "send", intent_id = %request.intent_id, signature = %signature.0, error_message = %error, "Devnet System transfer submission failed" ); 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) => { post_execution.diagnostics.push(format!( "confirmation policy could not be created after submission: {error}" )); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); }, }; 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) => { post_execution .diagnostics .push(format!("confirmation polling failed after transaction submission: {error}")); summary.post_execution = std::option::Option::Some(post_execution); tracing::error!( target: crate::TRACING_TARGET, action = "execute_devnet_system_transfer", stage = "confirmation", intent_id = %request.intent_id, signature = %signature.0, error_message = %error, "Devnet System transfer confirmation failed" ); return std::result::Result::Ok(summary); }, }; let confirmation_status = confirmation.status; summary.confirmation = std::option::Option::Some(confirmation); if confirmation_status == ks_lib::ExApiExecutionConfirmationStatus::Expired || confirmation_status == ks_lib::ExApiExecutionConfirmationStatus::TimedOut { post_execution.diagnostics.push(format!( "transaction was submitted but post-validation stopped at confirmation status {:?}", confirmation_status )); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); } let cancellation_result = ensure_not_cancelled(observer, "canonical_insert"); if let std::result::Result::Err(error) = cancellation_result { post_execution .diagnostics .push(format!("post-execution validation stopped after submission: {error}")); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); } let backfill = match execute_post_validation_backfill( http_pool, store, profile, request, observer, &signature, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { post_execution .diagnostics .push(format!("canonical hydration failed after submission: {error}")); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); }, }; let canonical_available = canonical_backfill_available(&backfill); post_execution.canonical_inserted = canonical_available; summary.backfill = std::option::Option::Some(backfill); if !canonical_available { post_execution.diagnostics.push( "confirmed transaction was not available for canonical post-validation".to_string(), ); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); } let extraction_request = 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, }; let extraction = match ks_pipeline::execute_core_extraction(store, &extraction_request, observer).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { post_execution .diagnostics .push(format!("core extraction failed after submission: {error}")); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); }, }; let core_extracted = extraction.failed == 0 && !extraction.cancelled && extraction.selected == 1 && extraction.extracted.saturating_add(extraction.skipped) >= 1; post_execution.core_extracted = core_extracted; summary.core_extraction = std::option::Option::Some(extraction); if !core_extracted { post_execution .diagnostics .push("canonical transaction did not complete core extraction".to_string()); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); } let program_ids = summary .plan .instructions .iter() .map(|instruction| return instruction.program_id.0.clone()) .collect::>(); let selection = match ks_store::DecodeSelectionFilter::new( std::vec![signature.0.clone()], std::vec![ ks_store::CoreInstructionProcessingState::Pending, ks_store::CoreInstructionProcessingState::Failed, ks_store::CoreInstructionProcessingState::ReplayRequested, ], std::option::Option::None, std::option::Option::None, program_ids, std::vec::Vec::new(), false, 32, ) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { post_execution.diagnostics.push(format!( "decode selection could not be constructed after submission: {error}" )); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); }, }; let decode_request = 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: request.force_post_validation_replay, force_replay_all_matching: false, materialize_after_decode: request.materialize_after_decode, }; let decode = match ks_pipeline::execute_decode_replay( store, &decode_request, decoders, materializers, observer, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { post_execution .diagnostics .push(format!("decode replay failed after submission: {error}")); summary.post_execution = std::option::Option::Some(post_execution); return std::result::Result::Ok(summary); }, }; let processors_clean = decode.processors.iter().all(|processor| { return processor.failed == 0 && processor.processing_errors == 0 && processor.unsupported == 0; }); let decoded_observations = decode.processors.iter().map(|processor| return processor.decoded).sum::(); let decode_replayed = decode.failed_inputs == 0 && decode.processing_error_inputs == 0 && decode.unmatched == 0 && !decode.cancelled && decode.completed >= 1 && processors_clean && decoded_observations >= 1; let materialized = request.materialize_after_decode && decode_replayed && decode .processors .iter() .all(|processor| return processor.materialization_refused == 0); post_execution.decode_replayed = decode_replayed; post_execution.materialized = materialized; summary.decode_replay = std::option::Option::Some(decode); if confirmation_status == ks_lib::ExApiExecutionConfirmationStatus::Failed { post_execution.diagnostics.push( "transaction was committed with an on-chain error and replayed as a failed intent" .to_string(), ); } else if !decode_replayed { post_execution.diagnostics.push( "transaction reached core storage but decode replay did not complete cleanly" .to_string(), ); } else if request.materialize_after_decode && !materialized { post_execution.diagnostics.push( "decode replay completed but requested materialization was refused or failed" .to_string(), ); } else { post_execution.diagnostics.push( "transaction completed canonical insertion, core extraction and decode replay" .to_string(), ); } let canonical_inserted = post_execution.canonical_inserted; let core_extracted = post_execution.core_extracted; let decode_replayed = post_execution.decode_replayed; let materialized = post_execution.materialized; summary.post_execution = std::option::Option::Some(post_execution); tracing::info!( target: crate::TRACING_TARGET, action = "execute_devnet_system_transfer", intent_id = %request.intent_id, signature = %signature.0, confirmation_status = ?confirmation_status, canonical_inserted, core_extracted, decode_replayed, materialized, "Devnet System transfer orchestration completed" ); return std::result::Result::Ok(summary); } async fn execute_post_validation_backfill( http_pool: &ks_onchain_transport::HttpEndpointPool, store: &S, profile: &ks_config::ProfileConfig, request: &crate::DevnetSystemTransferRequest, observer: &O, signature: &ks_lib::MdSignature, ) -> ks_core::Result where S: ks_store::RawTransactionStore + Sync, O: crate::SolanaExecutionObserver, { let mut retry_index = 0_u32; loop { let backfill_request = ks_pipeline::BackfillRequest { role: request.query_role.clone(), 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: 0, }; let backfill = match ks_pipeline::execute_http_backfill(http_pool, store, &backfill_request, observer) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if canonical_backfill_available(&backfill) || retry_index >= request.post_validation_max_retries || observer.is_execution_cancelled() { return std::result::Result::Ok(backfill); } retry_index = retry_index.saturating_add(1); let delay_ms = std::cmp::max(profile.execution.confirmation_poll_interval_ms, 500); emit( observer, crate::SolanaExecutionProgressLevel::Warning, "canonical_insert", format!( "getTransaction is not available yet for {}; retry {}/{} after {} ms", signature.0, retry_index, request.post_validation_max_retries, delay_ms ), std::option::Option::Some(signature.0.clone()), ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; } } fn canonical_backfill_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; } fn validate_recipient_transfer_amount( recipient_exists: bool, lamports: u64, minimum_balance_lamports: u64, ) -> ks_core::Result<()> { if !recipient_exists && lamports < minimum_balance_lamports { return std::result::Result::Err(ks_core::Error::new( "execution_recipient_rent_exemption_required", format!( "recipient account does not exist: transfer requires at least {minimum_balance_lamports} lamports for a zero-data rent-exempt account, requested {lamports}" ), )); } return std::result::Result::Ok(()); } pub(crate) fn simulation_failure_message( simulation: &ks_lib::ExApiExecutionSimulationResult, ) -> std::string::String { let error = match simulation.error.as_deref() { std::option::Option::Some(value) => value, std::option::Option::None => "runtime returned no structured error", }; let retained_logs = simulation.logs.iter().take(20).cloned().collect::>(); if retained_logs.is_empty() { return format!("exact transaction simulation failed: error={error}; logs=[]"); } return format!( "exact transaction simulation failed: error={error}; logs=[{}]", retained_logs.join(" | ") ); } #[derive(Clone, Debug, Eq, PartialEq)] struct FundingResult { balance_after_lamports: u64, airdrop_signature: std::option::Option, airdrop_confirmation: std::option::Option, } fn validate_devnet_profile( profile: &ks_config::ProfileConfig, request: &crate::DevnetSystemTransferRequest, ) -> ks_core::Result<()> { if profile.wallet.cluster != "devnet" { return std::result::Result::Err(ks_core::Error::config(format!( "profile {} wallet cluster must be devnet for this orchestration", profile.name ))); } if !profile.wallet.temporary_wallet_enabled || !profile.wallet.temporary_wallet_persist { return std::result::Result::Err(ks_core::Error::config( "Devnet execution requires an enabled persistent temporary wallet", )); } if !profile.execution.require_simulation { return std::result::Result::Err(ks_core::Error::config( "Devnet execution requires simulation in the active profile", )); } if request.lamports > profile.execution.devnet_max_spend_lamports { return std::result::Result::Err(ks_core::Error::config(format!( "requested transfer {} exceeds Devnet spend ceiling {}", request.lamports, profile.execution.devnet_max_spend_lamports ))); } if request.airdrop_lamports > profile.execution.devnet_airdrop_max_lamports { return std::result::Result::Err(ks_core::Error::config(format!( "requested airdrop {} exceeds Devnet airdrop ceiling {}", request.airdrop_lamports, profile.execution.devnet_airdrop_max_lamports ))); } if request.submit && !profile.wallet.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( "Devnet submission requires explicit operator confirmation", )); } return std::result::Result::Ok(()); } pub(crate) async fn load_profile_wallet( profile: &ks_config::ProfileConfig, workspace_root: &std::path::Path, ) -> ks_core::Result { let configured = std::path::PathBuf::from(profile.wallet.wallet_dir.as_str()); let directory = if configured.is_absolute() { configured } else { workspace_root.join(configured) }; let store = match ks_wallet::TemporaryWalletStore::new(directory) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let alias = match ks_wallet::WalletAlias::parse(profile.wallet.temporary_wallet_alias.clone()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; return store.load_or_create(alias).await; } async fn ensure_devnet_funding( http_pool: &ks_onchain_transport::HttpEndpointPool, profile: &ks_config::ProfileConfig, request: &crate::DevnetSystemTransferRequest, observer: &O, source_pubkey: &ks_lib::MdPubkey, balance_before_lamports: u64, required_balance_lamports: u64, ) -> ks_core::Result where O: crate::SolanaExecutionObserver, { if balance_before_lamports >= required_balance_lamports { return std::result::Result::Ok(FundingResult { balance_after_lamports: balance_before_lamports, airdrop_signature: std::option::Option::None, airdrop_confirmation: std::option::Option::None, }); } if !request.submit { return std::result::Result::Err(ks_core::Error::new( "execution_balance_insufficient", format!( "wallet balance {} is below required {} and simulation-only mode cannot request a funding transaction", balance_before_lamports, required_balance_lamports ), )); } if request.airdrop_lamports == 0 { return std::result::Result::Err(ks_core::Error::new( "execution_airdrop_required", format!( "wallet balance {} is below required {}; configure a bounded Devnet airdrop", balance_before_lamports, required_balance_lamports ), )); } let projected_balance = match balance_before_lamports.checked_add(request.airdrop_lamports) { std::option::Option::Some(value) => value, std::option::Option::None => { return std::result::Result::Err(ks_core::Error::config( "Devnet balance plus airdrop overflows u64", )); }, }; if projected_balance < required_balance_lamports { return std::result::Result::Err(ks_core::Error::config(format!( "configured airdrop would raise balance only to {}, below required {}", projected_balance, required_balance_lamports ))); } emit( observer, crate::SolanaExecutionProgressLevel::Info, "airdrop", format!("requesting {} Devnet lamports for temporary wallet", request.airdrop_lamports), std::option::Option::None, ); let airdrop = match http_pool .request_airdrop_for_role( request.query_role.as_str(), source_pubkey, request.airdrop_lamports, &ks_onchain_transport::RequestAirdropConfig::confirmed(), ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let confirmation_config = match ks_onchain_transport::ConfirmTransactionConfig::from_execution_config( &profile.execution, std::option::Option::None, std::option::Option::None, ) { 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, &airdrop.signature, &confirmation_config, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if confirmation.status != ks_lib::ExApiExecutionConfirmationStatus::Confirmed && confirmation.status != ks_lib::ExApiExecutionConfirmationStatus::Finalized { return std::result::Result::Err(ks_core::Error::new( "execution_airdrop_not_confirmed", format!("Devnet airdrop ended with confirmation status {:?}", confirmation.status), )); } let balance = match http_pool .get_balance_for_role( request.query_role.as_str(), source_pubkey, &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 < required_balance_lamports { return std::result::Result::Err(ks_core::Error::new( "execution_balance_still_insufficient", format!( "confirmed airdrop left balance {} below required {}", balance.lamports, required_balance_lamports ), )); } return std::result::Result::Ok(FundingResult { balance_after_lamports: balance.lamports, airdrop_signature: std::option::Option::Some(airdrop.signature), airdrop_confirmation: std::option::Option::Some(confirmation), }); } fn build_transfer_plan( profile: &ks_config::ProfileConfig, request: &crate::DevnetSystemTransferRequest, source_pubkey: ks_lib::MdPubkey, ) -> ks_core::Result { let 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: std::vec![source_pubkey.clone()], dry_run: !request.submit, post_execution_validation: ks_lib::ExApiPostExecutionValidationPolicy { canonical_insert_required: request.submit, core_extraction_required: request.submit, decode_replay_required: request.submit, materialization_required: request.submit && request.materialize_after_decode, }, }; let intent = ks_lib::ExSolanaCoreExecutionIntent { intent_id: request.intent_id.clone(), fee_payer: source_pubkey.clone(), policy, operation: ks_lib::ExSolanaCoreOperation::SystemTransfer { from: source_pubkey, to: request.recipient.clone(), lamports: request.lamports, }, }; let executor = ks_lib::ExSolanaCoreExecutor; return ks_lib::ExApiTypedInstructionExecutor::build_prepared_plan(&executor, &intent); } pub(crate) fn violation_message(violations: &[ks_lib::ExSafetyViolation]) -> std::string::String { if violations.is_empty() { return "execution policy denied the plan without a diagnostic".to_string(); } return violations .iter() .map(|violation| return format!("{}: {}", violation.code, violation.message)) .collect::>() .join("; "); } pub(crate) fn ensure_not_cancelled(observer: &O, stage: &str) -> ks_core::Result<()> where O: crate::SolanaExecutionObserver, { if observer.is_execution_cancelled() { return std::result::Result::Err(ks_core::Error::new( "execution_cancelled", format!("execution was cancelled before stage {stage}"), )); } return std::result::Result::Ok(()); } pub(crate) fn emit( observer: &O, level: crate::SolanaExecutionProgressLevel, stage: impl std::convert::Into, message: impl std::convert::Into, signature: std::option::Option, ) where O: crate::SolanaExecutionObserver, { observer.on_execution_progress(&crate::SolanaExecutionProgressEvent::new( level, stage, message, signature, )); } #[cfg(test)] mod tests { fn example_devnet_profile() -> ks_config::ProfileConfig { let config = match ks_config::parse_config_json(include_str!("../../config/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 recipient() -> ks_lib::MdPubkey { return ks_lib::MdPubkey("Vote111111111111111111111111111111111111111".to_string()); } #[test] fn request_and_profile_validation_are_conservative() { let profile = example_devnet_profile(); let mut request = crate::DevnetSystemTransferRequest::new("intent-1", recipient(), 1_000); assert!(request.validate().is_ok()); assert!(super::validate_devnet_profile(&profile, &request).is_ok()); request.submit = true; assert!(super::validate_devnet_profile(&profile, &request).is_err()); request.operator_confirmed = true; assert!(super::validate_devnet_profile(&profile, &request).is_ok()); request.airdrop_lamports = profile.execution.devnet_airdrop_max_lamports.saturating_add(1); assert!(super::validate_devnet_profile(&profile, &request).is_err()); } #[test] fn resolved_devnet_profile_routes_complete_execution_flow() { let profile = example_devnet_profile(); 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}"), }; for method in [ "getGenesisHash", "getAccountInfo", "getMinimumBalanceForRentExemption", "getBalance", "requestAirdrop", "getLatestBlockhash", "getFeeForMessage", "getBlockHeight", "getTransaction", ] { let selected = pool.select_client_for_role_and_method("http_queries", method); assert!(selected.is_ok(), "http_queries must route {method}"); } for method in ["simulateTransaction", "sendTransaction", "getSignatureStatuses"] { let selected = pool.select_client_for_role_and_method("http_transactions", method); assert!(selected.is_ok(), "http_transactions must route {method}"); } } #[test] fn transfer_plan_uses_exact_devnet_policy() { let profile = example_devnet_profile(); let mut request = crate::DevnetSystemTransferRequest::new("intent-2", recipient(), 5_000); request.submit = true; request.operator_confirmed = true; let source = ks_lib::MdPubkey("11111111111111111111111111111111".to_string()); let plan = match super::build_transfer_plan(&profile, &request, source.clone()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("plan build failed: {error}"), }; assert_eq!(plan.fee_payer, source); assert_eq!(plan.requested_spend_lamports, 5_000); assert_eq!(plan.policy.cluster.expected_cluster, ks_lib::ExApiExecutionCluster::Devnet); assert!(!plan.policy.dry_run); assert!(plan.policy.post_execution_validation.canonical_insert_required); assert_eq!(plan.instructions.len(), 1); assert_eq!(plan.operation_code, ks_lib::EX_SOLANA_CORE_SYSTEM_TRANSFER_OPERATION); } #[test] fn new_recipient_requires_zero_data_rent_exemption() { assert!(super::validate_recipient_transfer_amount(false, 1_000, 890_880,).is_err()); assert!(super::validate_recipient_transfer_amount(false, 890_880, 890_880,).is_ok()); assert!(super::validate_recipient_transfer_amount(true, 1, 890_880,).is_ok()); } #[test] fn simulation_failure_message_preserves_runtime_error_and_logs() { let simulation = ks_lib::ExApiExecutionSimulationResult { simulated: true, success: false, cluster: ks_lib::ExApiExecutionCluster::Devnet, blockhash_kind: ks_lib::ExApiExecutionBlockhashKind::Latest, blockhash_age_slots: std::option::Option::Some(1), replacement_blockhash: std::option::Option::None, replacement_last_valid_block_height: std::option::Option::None, nonce_account: std::option::Option::None, nonce_authority: std::option::Option::None, units_consumed: std::option::Option::Some(150), estimated_fee_lamports: std::option::Option::Some(5_000), logs: std::vec!["Program log: insufficient rent".to_string()], return_data: std::option::Option::None, error: std::option::Option::Some("InsufficientFundsForRent".to_string()), }; let message = super::simulation_failure_message(&simulation); assert!(message.contains("InsufficientFundsForRent")); assert!(message.contains("insufficient rent")); } #[tokio::test] async fn optional_devnet_system_transfer_from_env() { if std::env::var("KS_DEVNET_EXECUTION_TEST").ok().as_deref() != std::option::Option::Some("1") { return; } 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: {error}"); }, }; let mut profile = example_devnet_profile(); profile.database.backend = "postgres".to_string(); profile.database.postgres.url = database_url; if let std::result::Result::Ok(directory) = std::env::var("KS_DEVNET_WALLET_DIR") { profile.wallet.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 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 recipient_alias = match ks_wallet::WalletAlias::parse("devnet-recipient") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => panic!("recipient alias failed: {error}"), }; let recipient_wallet = ks_wallet::TemporaryWallet::generate(recipient_alias); let transfer_lamports = std::env::var("KS_DEVNET_TRANSFER_LAMPORTS") .ok() .and_then(|value| return value.parse::().ok()) .unwrap_or(1_000_000); let airdrop_lamports = std::env::var("KS_DEVNET_AIRDROP_LAMPORTS") .ok() .and_then(|value| return value.parse::().ok()) .unwrap_or(100_000_000); let mut request = crate::DevnetSystemTransferRequest::new( format!("devnet-test-{}", uuid::Uuid::new_v4()), ks_lib::MdPubkey(recipient_wallet.public_key()), transfer_lamports, ); request.airdrop_lamports = airdrop_lamports; request.post_validation_max_retries = 20; request.submit = true; request.operator_confirmed = true; let decoders: std::vec::Vec> = std::vec![std::sync::Arc::new(ks_lib::DcSolanaCoreDecoder,)]; let materializers: std::vec::Vec> = std::vec::Vec::new(); 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_system_transfer( &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 execution failed: {error}"), }; let confirmation = match summary.confirmation { std::option::Option::Some(value) => value, std::option::Option::None => panic!("transaction confirmation missing"), }; assert!( confirmation.status == ks_lib::ExApiExecutionConfirmationStatus::Confirmed || confirmation.status == ks_lib::ExApiExecutionConfirmationStatus::Finalized ); let post_execution = match summary.post_execution { std::option::Option::Some(value) => value, std::option::Option::None => panic!("post-execution diagnostic missing"), }; assert!(post_execution.canonical_inserted); assert!(post_execution.core_extracted); assert!(post_execution.decode_replayed); } }