v0.1.0-pre.038

This commit is contained in:
2026-07-25 13:23:56 +02:00
parent 5b4a0ae0c5
commit e4cae3584e
3 changed files with 928 additions and 7 deletions

View File

@@ -1,5 +1,5 @@
// file: kb-pipeline/src/lib.rs // file: kb-pipeline/src/lib.rs
// version: 11 // version: 12
#![forbid(unsafe_code)] #![forbid(unsafe_code)]
#![deny(unreachable_pub)] #![deny(unreachable_pub)]
@@ -14,6 +14,7 @@ mod decode_replay;
mod plan; mod plan;
mod solana_elgamal_registry_stateful; mod solana_elgamal_registry_stateful;
mod solana_execution; mod solana_execution;
mod solana_memo_execution;
mod solana_stateful; mod solana_stateful;
mod solana_token2022_correlation; mod solana_token2022_correlation;
mod solana_token2022_crypto_preflight; mod solana_token2022_crypto_preflight;
@@ -24,6 +25,17 @@ mod solana_token2022_proof_orchestration;
mod solana_token2022_stateful; mod solana_token2022_stateful;
mod solana_token2022_validation; mod solana_token2022_validation;
/// Emits one execution progress event for specialized pipeline orchestrators.
pub(crate) use self::solana_execution::emit;
/// Rejects one execution stage when the observer reports cancellation.
pub(crate) use self::solana_execution::ensure_not_cancelled;
/// Loads the persistent wallet configured by one execution profile.
pub(crate) use self::solana_execution::load_profile_wallet;
/// Formats one failed simulation without discarding runtime diagnostics.
pub(crate) use self::solana_execution::simulation_failure_message;
/// Formats safety violations for one denied execution plan.
pub(crate) use self::solana_execution::violation_message;
/// Address category used by one targeted backfill campaign. /// Address category used by one targeted backfill campaign.
pub use self::backfill::BackfillAddressKind; pub use self::backfill::BackfillAddressKind;
/// Chronological direction relative to one anchor signature. /// Chronological direction relative to one anchor signature.
@@ -120,6 +132,12 @@ pub use self::solana_execution::SolanaExecutionProgressEvent;
pub use self::solana_execution::SolanaExecutionProgressLevel; pub use self::solana_execution::SolanaExecutionProgressLevel;
/// Executes one bounded System Program transfer on Devnet. /// Executes one bounded System Program transfer on Devnet.
pub use self::solana_execution::execute_devnet_system_transfer; pub use self::solana_execution::execute_devnet_system_transfer;
/// Complete request for one SPL Memo v4 Devnet execution.
pub use self::solana_memo_execution::DevnetMemoExecutionRequest;
/// Complete result of one SPL Memo v4 Devnet execution and post-validation.
pub use self::solana_memo_execution::DevnetMemoExecutionSummary;
/// Executes one SPL Memo v4 Devnet simulation or explicitly authorized submission.
pub use self::solana_memo_execution::execute_devnet_memo;
/// One machine-readable native Solana stateful readiness check. /// One machine-readable native Solana stateful readiness check.
pub use self::solana_stateful::SolanaCoreStatefulCheck; pub use self::solana_stateful::SolanaCoreStatefulCheck;
/// One contextual fact measured during native Solana stateful readiness. /// One contextual fact measured during native Solana stateful readiness.

View File

@@ -1,5 +1,5 @@
// file: kb-pipeline/src/solana_execution.rs // file: kb-pipeline/src/solana_execution.rs
// version: 5 // version: 6
//! Devnet Solana execution orchestration with canonical post-validation. //! Devnet Solana execution orchestration with canonical post-validation.
@@ -904,7 +904,7 @@ fn validate_recipient_transfer_amount(
return std::result::Result::Ok(()); return std::result::Result::Ok(());
} }
fn simulation_failure_message( pub(crate) fn simulation_failure_message(
simulation: &kb_lib::ExApiExecutionSimulationResult, simulation: &kb_lib::ExApiExecutionSimulationResult,
) -> std::string::String { ) -> std::string::String {
let error = match simulation.error.as_deref() { let error = match simulation.error.as_deref() {
@@ -976,7 +976,7 @@ fn validate_devnet_profile(
return std::result::Result::Ok(()); return std::result::Result::Ok(());
} }
async fn load_profile_wallet( pub(crate) async fn load_profile_wallet(
profile: &kb_config::ProfileConfig, profile: &kb_config::ProfileConfig,
workspace_root: &std::path::Path, workspace_root: &std::path::Path,
) -> kb_core::Result<kb_wallet::TemporaryWallet> { ) -> kb_core::Result<kb_wallet::TemporaryWallet> {
@@ -1176,7 +1176,7 @@ fn build_transfer_plan(
return kb_lib::ExApiTypedInstructionExecutor::build_prepared_plan(&executor, &intent); return kb_lib::ExApiTypedInstructionExecutor::build_prepared_plan(&executor, &intent);
} }
fn violation_message(violations: &[kb_lib::ExSafetyViolation]) -> std::string::String { pub(crate) fn violation_message(violations: &[kb_lib::ExSafetyViolation]) -> std::string::String {
if violations.is_empty() { if violations.is_empty() {
return "execution policy denied the plan without a diagnostic".to_string(); return "execution policy denied the plan without a diagnostic".to_string();
} }
@@ -1187,7 +1187,7 @@ fn violation_message(violations: &[kb_lib::ExSafetyViolation]) -> std::string::S
.join("; "); .join("; ");
} }
fn ensure_not_cancelled<O>(observer: &O, stage: &str) -> kb_core::Result<()> pub(crate) fn ensure_not_cancelled<O>(observer: &O, stage: &str) -> kb_core::Result<()>
where where
O: crate::SolanaExecutionObserver, O: crate::SolanaExecutionObserver,
{ {
@@ -1200,7 +1200,7 @@ where
return std::result::Result::Ok(()); return std::result::Result::Ok(());
} }
fn emit<O>( pub(crate) fn emit<O>(
observer: &O, observer: &O,
level: crate::SolanaExecutionProgressLevel, level: crate::SolanaExecutionProgressLevel,
stage: impl std::convert::Into<std::string::String>, stage: impl std::convert::Into<std::string::String>,

View File

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