diff --git a/Cargo.lock b/Cargo.lock index 04e8d89..4537940 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3234,6 +3234,7 @@ dependencies = [ "kobe-core", "log", "mongodb", + "serde_json", "solana-client", "solana-metrics", "solana-sdk", diff --git a/core/src/rpc_utils.rs b/core/src/rpc_utils.rs index d9acac8..fd34bcb 100644 --- a/core/src/rpc_utils.rs +++ b/core/src/rpc_utils.rs @@ -11,6 +11,14 @@ use tokio_retry::strategy::{jitter, FibonacciBackoff}; use tokio_retry::Retry; pub const MAX_RPC_RETRIES: usize = 10; + +/// Highest transaction version we ask the RPC to return. +/// +/// `getTransaction` fails outright with `UnsupportedTransactionVersion` when it +/// is asked for a transaction newer than this, so it has to keep up with what +/// the cluster can produce. v1 (SIMD-0296 / SIMD-0385) ships in Agave v4.2. +pub const MAX_SUPPORTED_TRANSACTION_VERSION: u8 = 1; + type RetryStrategy = Take Duration>>; pub fn retry() -> RetryStrategy { @@ -21,10 +29,16 @@ pub fn retry() -> RetryStrategy { .take(MAX_RPC_RETRIES) } +/// Fetches each signature's transaction, paired with the signature it was +/// requested for. +/// +/// Returning the signature lets callers line results up with their own +/// bookkeeping without decoding the transaction, which fails for any version +/// the pinned SDK predates. pub async fn retry_get_transactions( rpc_client: &RpcClient, transaction_signatures: &[Signature], -) -> Result, Box> { +) -> Result, Box> { let txes = Retry::spawn(retry(), || { get_signatures_internal(rpc_client, transaction_signatures) }) @@ -36,11 +50,11 @@ pub async fn retry_get_transactions( async fn get_signatures_internal( rpc_client: &RpcClient, transaction_signatures: &[Signature], -) -> Result, Box> { +) -> Result, Box> { let config = RpcTransactionConfig { commitment: CommitmentConfig::finalized().into(), - encoding: UiTransactionEncoding::Base64.into(), - max_supported_transaction_version: Some(0), + encoding: UiTransactionEncoding::Json.into(), + max_supported_transaction_version: Some(MAX_SUPPORTED_TRANSACTION_VERSION), }; let mut temp_txs = vec![]; @@ -48,7 +62,7 @@ async fn get_signatures_internal( let tx = rpc_client .get_transaction_with_config(signature, config) .await?; - temp_txs.push(tx); + temp_txs.push((*signature, tx)); } Ok(temp_txs) } diff --git a/steward-writer-service/Cargo.toml b/steward-writer-service/Cargo.toml index 0b6e1a7..1e9cf3f 100644 --- a/steward-writer-service/Cargo.toml +++ b/steward-writer-service/Cargo.toml @@ -22,3 +22,10 @@ solana-metrics = { workspace = true } solana-sdk = { workspace = true } solana-transaction-status = { workspace = true } tokio = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } + +[[example]] +name = "verify_transaction_format" +path = "examples/verify_transaction_format.rs" diff --git a/steward-writer-service/examples/verify_transaction_format.rs b/steward-writer-service/examples/verify_transaction_format.rs new file mode 100644 index 0000000..75e0432 --- /dev/null +++ b/steward-writer-service/examples/verify_transaction_format.rs @@ -0,0 +1,304 @@ +//! Checks that the steward writer can still read the transaction format the +//! cluster is producing right now. +//! +//! The writer asks the RPC for `Json`-encoded transactions with +//! `maxSupportedTransactionVersion: 1` and reads the fee payer out of the +//! node's parsed message, rather than deserializing the message itself. That +//! change is what makes v1 (SIMD-0296 / SIMD-0385) transactions readable on the +//! pinned 2.3.x SDK, but it also has to keep working for the legacy and v0 +//! transactions that are all the cluster produces today. Unit tests cover both +//! against synthetic payloads; this samples real ones. +//! +//! ```text +//! cargo run -p kobe-steward-writer-service --example verify_transaction_format -- \ +//! --rpc-url https://api.mainnet-beta.solana.com +//! ``` +//! +//! Exits non-zero if any sampled transaction could not be read, so it doubles +//! as a smoke check to re-run once v1 activates on a cluster. + +use std::{collections::BTreeMap, process::ExitCode, str::FromStr}; + +use clap::Parser; +use kobe_core::rpc_utils::retry_get_transactions; +use kobe_steward_writer_service::{describe_version, fee_payer, get_epoch_from_slot, parse_log}; +use solana_client::{ + nonblocking::rpc_client::RpcClient, rpc_client::GetConfirmedSignaturesForAddress2Config, +}; +use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Signature}; +use solana_transaction_status::{ + option_serializer::OptionSerializer, EncodedConfirmedTransactionWithStatusMeta, +}; + +#[derive(Parser)] +#[command(about = "Check the steward writer can read the cluster's current transaction format")] +struct Args { + /// RPC URL to check against. + #[arg(long)] + rpc_url: String, + + /// Program whose recent transactions to sample. Defaults to the steward program. + #[arg(long)] + program_id: Option, + + /// How many recent transactions to sample. + #[arg(long, default_value_t = 20)] + limit: usize, + + /// Check these signatures instead of sampling. Repeatable. + #[arg(long = "signature")] + signatures: Vec, + + /// Page back from this signature rather than from the most recent. + #[arg(long)] + before: Option, +} + +#[tokio::main] +async fn main() -> ExitCode { + env_logger::init(); + let args = Args::parse(); + + let program_id = args.program_id.unwrap_or_else(jito_steward::id); + let rpc_client = RpcClient::new_with_commitment(args.rpc_url, CommitmentConfig::confirmed()); + + let signatures = if args.signatures.is_empty() { + println!( + "Sampling up to {} transactions for {program_id}\n", + args.limit + ); + + let statuses = match rpc_client + .get_signatures_for_address_with_config( + &program_id, + GetConfirmedSignaturesForAddress2Config { + before: args.before, + until: None, + limit: Some(args.limit), + commitment: Some(CommitmentConfig::confirmed()), + }, + ) + .await + { + Ok(statuses) => statuses, + Err(e) => { + eprintln!("Failed to fetch signatures: {e}"); + return ExitCode::FAILURE; + } + }; + + if statuses.is_empty() { + println!("No recent transactions for {program_id}; nothing to check."); + return ExitCode::SUCCESS; + } + + statuses + .iter() + .map(|status| { + Signature::from_str(&status.signature).expect("RPC returned a valid signature") + }) + .collect() + } else { + println!("Checking {} given signatures\n", args.signatures.len()); + args.signatures.clone() + }; + + // The writer's own fetch path, so whatever encoding and + // `maxSupportedTransactionVersion` it requests, this requests too. + let transactions = match retry_get_transactions(&rpc_client, &signatures).await { + Ok(transactions) => transactions, + Err(e) => { + eprintln!("Failed to fetch transactions: {e}"); + eprintln!( + "An UnsupportedTransactionVersion error here means the cluster produces a \ + version above MAX_SUPPORTED_TRANSACTION_VERSION in kobe_core::rpc_utils." + ); + return ExitCode::FAILURE; + } + }; + + let mut readable_by_version: BTreeMap = BTreeMap::new(); + let mut decoded_by_type: BTreeMap = BTreeMap::new(); + let mut unreadable = Vec::new(); + let mut total_event_logs = 0; + let mut undecodable_event_logs = 0; + + for (signature, tx) in &transactions { + let version = describe_version(tx.transaction.version.as_ref()); + let (logs, event_logs) = log_counts(tx); + let instruction = instruction_name(tx); + total_event_logs += event_logs; + + let events = decoded_events(signature, tx).await; + for event in &events { + match event { + Some(name) => *decoded_by_type.entry(name.clone()).or_default() += 1, + None => undecodable_event_logs += 1, + } + } + + match fee_payer(&tx.transaction.transaction) { + Some(payer) => { + *readable_by_version.entry(version.clone()).or_default() += 1; + println!( + " ok {version:>7} {instruction:<18} logs {logs:>3} events {event_logs:>3} payer {payer}" + ); + } + None => { + println!( + " FAIL {version:>7} {instruction:<18} no fee payer readable {signature}" + ); + unreadable.push((*signature, version)); + } + } + + if !events.is_empty() { + let names: Vec<&str> = events + .iter() + .map(|e| e.as_deref().unwrap_or("UNDECODABLE")) + .collect(); + println!(" decoded: {}", names.join(", ")); + } + } + + println!("\nReadable by version:"); + for (version, count) in &readable_by_version { + println!(" {version:>7}: {count}"); + } + + println!("Anchor event log lines seen: {total_event_logs}"); + if total_event_logs == 0 { + println!( + " (0 is expected for `Idle` cranks, which emit no events. Pass --signature with a \ + `Rebalance` or `ComputeScore` transaction to exercise the log path.)" + ); + } else { + println!("Decoded steward events:"); + for (name, count) in &decoded_by_type { + println!(" {name:<28}: {count}"); + } + if undecodable_event_logs > 0 { + println!( + " {:<28}: {undecodable_event_logs} <-- writer would store nothing for these", + "UNDECODABLE" + ); + } + } + + if unreadable.is_empty() && undecodable_event_logs == 0 { + println!( + "\nAll {} sampled transactions readable, all {total_event_logs} event logs decoded. \ + The writer can read this cluster's format.", + transactions.len() + ); + return ExitCode::SUCCESS; + } + + if !unreadable.is_empty() { + println!( + "\n{} of {} transactions unreadable:", + unreadable.len(), + transactions.len() + ); + for (signature, version) in &unreadable { + println!(" {version} {signature}"); + } + println!( + "The writer skips these and emits steward_writer_service-unreadable_transaction \ + for each, so their events are lost. Check the encoding requested in \ + kobe_core::rpc_utils::transaction_config." + ); + } + + if undecodable_event_logs > 0 { + println!( + "\n{undecodable_event_logs} event log(s) matched no event type the writer knows. \ + That is silent loss: parse_log returns Ok(None) and nothing is stored or logged. \ + Likely an on-chain event type newer than the pinned jito-steward rev." + ); + } + + ExitCode::FAILURE +} + +/// The instruction Anchor logged, so a zero event count can be read in context +/// — `Idle` emitting nothing is correct, `Rebalance` emitting nothing is not. +fn instruction_name(tx: &EncodedConfirmedTransactionWithStatusMeta) -> String { + const PREFIX: &str = "Program log: Instruction: "; + + match tx.transaction.meta.as_ref().map(|meta| &meta.log_messages) { + Some(OptionSerializer::Some(logs)) => logs + .iter() + .find_map(|log| log.strip_prefix(PREFIX)) + .unwrap_or("?") + .to_string(), + _ => "?".to_string(), + } +} + +/// Total log lines and the `Program data:` subset Anchor emits events as. +/// +/// The writer's `parse_log` consumes these. They come from `meta`, not the +/// encoded message, so they are unaffected by the encoding — reporting them +/// just confirms the event pipeline still has its input. +fn log_counts(tx: &EncodedConfirmedTransactionWithStatusMeta) -> (usize, usize) { + match tx.transaction.meta.as_ref().map(|meta| &meta.log_messages) { + Some(OptionSerializer::Some(logs)) => ( + logs.len(), + logs.iter() + .filter(|log| log.starts_with("Program data:")) + .count(), + ), + _ => (0, 0), + } +} + +/// Every `Program data:` line paired with the `event_type` the writer's own +/// `parse_log` decodes it to, or `None` when it recognises nothing. +/// +/// This calls the production decoder rather than reimplementing its dispatch, +/// so the names printed are exactly what would land in Mongo. A `None` is worth +/// surfacing: `parse_log` returns `Ok(None)` and the writer stores nothing and +/// logs nothing, which is silent loss rather than a visible failure. +async fn decoded_events( + signature: &Signature, + tx: &EncodedConfirmedTransactionWithStatusMeta, +) -> Vec> { + let Some(meta) = tx.transaction.meta.as_ref() else { + return vec![]; + }; + let OptionSerializer::Some(logs) = &meta.log_messages else { + return vec![]; + }; + + // `parse_log` only uses the stake pool to label the event, and this checks + // decodability rather than event contents. + let stake_pool = Pubkey::default(); + let signer = fee_payer(&tx.transaction.transaction).unwrap_or_default(); + + let mut decoded = Vec::new(); + for log in logs.iter().filter(|log| log.starts_with("Program data:")) { + let event = parse_log( + log.clone(), + signature, + 0, + &signer, + &stake_pool, + tx.block_time, + meta.err.clone(), + get_epoch_from_slot(tx.slot), + tx.slot, + ) + .await; + + decoded.push(match event { + Ok(Some(event)) => Some(event.event_type), + Ok(None) => None, + Err(e) => { + eprintln!(" parse_log errored on {signature}: {e}"); + None + } + }); + } + decoded +} diff --git a/steward-writer-service/src/lib.rs b/steward-writer-service/src/lib.rs new file mode 100644 index 0000000..f8e9205 --- /dev/null +++ b/steward-writer-service/src/lib.rs @@ -0,0 +1,441 @@ +//! Transaction reading and event parsing for the steward writer. +//! +//! These live in a library rather than the binary so the reading path can be +//! driven directly — see `examples/verify_transaction_format.rs`, which runs it +//! against a live RPC. + +use std::str::FromStr; + +use anchor_client::handle_program_log; +use jito_steward::{ + events::{ + AutoAddValidatorEvent, AutoRemoveValidatorEvent, DecreaseComponents, + DirectedRebalanceEvent, EpochMaintenanceEvent, InstantUnstakeComponents, RebalanceEvent, + ScoreComponents, StateTransition, + }, + score::{InstantUnstakeComponentsV3, ScoreComponentsV5}, +}; +use kobe_core::db_models::steward_events::StewardEvent; +use log::error; +use solana_client::rpc_response::RpcConfirmedTransactionStatusWithSignature; +use solana_sdk::{ + pubkey::Pubkey, + signature::Signature, + transaction::{TransactionError, TransactionVersion}, +}; +use solana_transaction_status::{ + EncodedConfirmedTransactionWithStatusMeta, EncodedTransaction, UiMessage, +}; + +/// Pairs fetched transactions back up with the signature statuses they came from. +/// +/// `retry_get_transactions` returns one entry per requested signature in request +/// order, so this is a positional zip. The equality check guards against +/// attributing a transaction's events to the wrong signature if that ever stops +/// holding. +pub fn pair_with_statuses( + signatures: &[RpcConfirmedTransactionStatusWithSignature], + transactions: Vec<(Signature, EncodedConfirmedTransactionWithStatusMeta)>, +) -> Vec<( + RpcConfirmedTransactionStatusWithSignature, + EncodedConfirmedTransactionWithStatusMeta, +)> { + signatures + .iter() + .zip(transactions) + .filter_map(|(status, (signature, tx))| { + if status.signature != signature.to_string() { + error!( + "Signature mismatch: requested {}, RPC returned {signature}", + status.signature + ); + return None; + } + Some((status.clone(), tx)) + }) + .collect() +} + +/// Human-readable transaction version, for diagnostics. +pub fn describe_version(version: Option<&TransactionVersion>) -> String { + match version { + Some(TransactionVersion::Number(n)) => format!("v{n}"), + Some(TransactionVersion::Legacy(_)) => "legacy".to_string(), + None => "unknown-version".to_string(), + } +} + +/// Reads the fee payer out of an RPC-parsed transaction message. +/// +/// The first account key is the fee payer in every transaction version. Taking +/// it from the node's parsed output rather than deserializing the message +/// ourselves is what lets this keep working for versions the pinned SDK has no +/// decoder for — see the encoding choice in `kobe_core::rpc_utils`. +pub fn fee_payer(transaction: &EncodedTransaction) -> Option { + let first_key = match transaction { + EncodedTransaction::Json(ui_transaction) => match &ui_transaction.message { + UiMessage::Raw(message) => message.account_keys.first().cloned(), + UiMessage::Parsed(message) => { + message.account_keys.first().map(|key| key.pubkey.clone()) + } + }, + // A binary encoding was requested somewhere; we can't parse it here + // without a version-aware decoder. + EncodedTransaction::LegacyBinary(_) + | EncodedTransaction::Binary(_, _) + | EncodedTransaction::Accounts(_) => None, + }?; + + Pubkey::from_str(&first_key).ok() +} + +#[allow(clippy::too_many_arguments)] +pub async fn parse_log( + log: String, + signature: &Signature, + instruction_idx: u32, + signer: &Pubkey, + stake_pool: &Pubkey, + timestamp: Option, + transaction_err: Option, + epoch: u64, + slot: u64, +) -> Result, Box> { + // Parse the log + let program = jito_steward::id().to_string(); + let tx_error = transaction_err.map(|e| e.to_string()); + + // DecreaseComponents + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { + let steward_event = StewardEvent::from_decrease_components( + event, + signature, + instruction_idx, + tx_error, + epoch, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // InstantUnstakeComponents + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) + { + let steward_event = StewardEvent::from_instant_unstake_components( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // InstantUnstakeComponentsV3 + if let Ok((Some(event), _, _)) = + handle_program_log::(&program, &log) + { + let steward_event = StewardEvent::from_instant_unstake_components_v3( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // RebalanceEvent + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { + let steward_event = StewardEvent::from_rebalance_event( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // DirectedRebalanceEvent + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { + let steward_event = StewardEvent::from_directed_rebalance_event( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // ScoreComponents + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { + let steward_event = StewardEvent::from_score_components( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // ScoreComponentsV5 + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { + let steward_event = StewardEvent::from_score_components_v5( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // StateTransition + if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { + let steward_event = StewardEvent::from_state_transition( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + slot, + ); + return Ok(Some(steward_event)); + } + + // AutoRemoveValidatorEvent + if let Ok((Some(event), _, _)) = + handle_program_log::(&program.to_string(), &log) + { + let steward_event = StewardEvent::from_auto_remove_validator_event( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + epoch, + slot, + ); + return Ok(Some(steward_event)); + } + + // AutoAddValidatorEvent + if let Ok((Some(event), _, _)) = + handle_program_log::(&program.to_string(), &log) + { + let steward_event = StewardEvent::from_auto_add_validator_event( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + epoch, + slot, + ); + return Ok(Some(steward_event)); + } + + // EpochMaintenanceEvent + if let Ok((Some(event), _, _)) = + handle_program_log::(&program.to_string(), &log) + { + let steward_event = StewardEvent::from_epoch_maintenance_event( + event, + signature, + instruction_idx, + tx_error, + signer, + stake_pool, + timestamp, + epoch, + slot, + ); + return Ok(Some(steward_event)); + } + + Ok(None) +} + +pub fn get_epoch_from_slot(slot: u64) -> u64 { + // Calculate the epoch from the slot + + slot / 432_000 +} + +#[cfg(test)] +mod tests { + use super::*; + use solana_transaction_status::EncodedTransactionWithStatusMeta; + + const FEE_PAYER: &str = "6WS1UtWtyeJHrsGbcARDPuLoXQeQTuFJHnsL2h1yc9CB"; + const OTHER_KEY: &str = "SysvarC1ock11111111111111111111111111111111"; + + /// A `getTransaction` response as an Agave v4.2 node renders a v1 + /// transaction: `version: 1`, and a `transactionConfig` field on the + /// message that this SDK's `UiRawMessage` has no field for. + fn v1_transaction_json() -> String { + format!( + r#"{{ + "transaction": {{ + "signatures": ["4HVYbFHkGwPjsHo3jNTfBnhJKzHrMg1S5g8vCsWpMHQFmDGKcNEqzP2M8j1YFwGZJnVw8UzKRZKtQPk2ZTgNYQwr"], + "message": {{ + "header": {{ + "numRequiredSignatures": 1, + "numReadonlySignedAccounts": 0, + "numReadonlyUnsignedAccounts": 1 + }}, + "accountKeys": ["{FEE_PAYER}", "{OTHER_KEY}"], + "recentBlockhash": "EkSnNWid2cvwEVnVx9aBqawnmiCNiDgp3gUdkDPTKN1N", + "instructions": [ + {{"programIdIndex": 1, "accounts": [0], "data": "3Bxs"}} + ], + "transactionConfig": {{ + "computeUnitLimit": 200000, + "loadedAccountsDataSizeLimit": 65536, + "feeLamports": 5000 + }} + }} + }}, + "meta": {{ + "err": null, + "status": {{"Ok": null}}, + "fee": 5000, + "preBalances": [1000000, 0], + "postBalances": [995000, 0], + "logMessages": ["Program log: hello"] + }}, + "version": 1 + }}"# + ) + } + + /// The whole v1 mitigation rests on this: an unknown `transactionConfig` + /// field must not fail deserialization, or every v1 transaction becomes an + /// RPC error rather than a readable record. + #[test] + fn v1_response_deserializes_under_pinned_sdk() { + let encoded: EncodedTransactionWithStatusMeta = + serde_json::from_str(&v1_transaction_json()).expect("v1 response should deserialize"); + + assert_eq!(encoded.version, Some(TransactionVersion::Number(1))); + assert_eq!(describe_version(encoded.version.as_ref()), "v1"); + } + + #[test] + fn fee_payer_reads_first_account_key_of_v1_transaction() { + let encoded: EncodedTransactionWithStatusMeta = + serde_json::from_str(&v1_transaction_json()).unwrap(); + + assert_eq!( + fee_payer(&encoded.transaction), + Some(Pubkey::from_str(FEE_PAYER).unwrap()) + ); + } + + /// A binary encoding yields no fee payer here by design: parsing it needs a + /// version-aware `bincode` decoder, which is the thing this SDK lacks. The + /// matching guard on the request side lives in `kobe_core::rpc_utils`. + #[test] + fn fee_payer_rejects_binary_encoding() { + let binary = EncodedTransaction::LegacyBinary("not parseable here".to_string()); + + assert_eq!(fee_payer(&binary), None); + } + + /// The five Anchor event logs emitted by a real `ComputeScore` transaction, + /// mainnet signature + /// `3AaEZ6PUYUmNH8tFbx3tHUs6buJhY3KhhNtri6U1Rm4xGo17ct2M6nVyCEdXqn7DoVjakgYBwoaAAg9EDAY5PmjB` + /// at slot 440863635. Verbatim, so this covers the real on-chain event + /// layout rather than one we constructed to match the decoder. + const COMPUTE_SCORE_EVENT_LOGS: [&str; 5] = [ + "Program data: HqAPeIvXzYIAAAAAAAAAAK//PBgAoIxfBegDDAAAAK//PAABAQEAAAEBAQaIjdQdpdfhgZtLvBcd/40lJ0Xw87GYrNLoZWOCWwnL/APoA/AD//8AAAAAAAAAAN4DBfADBfADAAD//wEB", + "Program data: HqAPeIvXzYIAAAAAAAAAAI4APRgAoIxfBegDDAAAAI4APQABAQEAAQEBAfY3xpialume/Kul7cD7zTtHMU62qGkHLR3yAm8ydlrv/APoA/AD//8AAAAAAAAAAN4DBfADBfADAAD//wEB", + "Program data: HqAPeIvXzYIAAAAAAAAAAF6lKBAAcJRfBfQBCAAAAF6lKAABAQEAAAEBAeQpUoVdDp1KFUpvB84QbrDAG6KlLw/zMKtIqUEytqoy/AP0AfQD//8AAAAAAAAAAN4DBfQDBfQDAAD//wEB", + "Program data: HqAPeIvXzYIAAAAAAAAAALxiGQoAQJxfBQAABQAAALxiGQABAQEAAQEBAb7uYNbiR9EumO+5zbWeuSxnHpx9d/vAP8EO42iowV/1/AMAAPwD//8AAAAAAAAAAN4DBfcDBfcDAAD//wEB", + "Program data: agl496lqzun8AwAAAAAAAJMLRxoAAAAADQAAAENvbXB1dGVTY29yZXMSAAAAQ29tcHV0ZURlbGVnYXRpb25z", + ]; + + /// Answers the question the encoding change raises end to end: a real + /// `ComputeScore` transaction still decodes into the events the writer + /// stores. `parse_log` reads `meta.log_messages`, which the `Json` encoding + /// leaves untouched — this pins that. + #[tokio::test] + async fn decodes_events_from_real_compute_score_transaction() { + let signature = Signature::from_str( + "3AaEZ6PUYUmNH8tFbx3tHUs6buJhY3KhhNtri6U1Rm4xGo17ct2M6nVyCEdXqn7DoVjakgYBwoaAAg9EDAY5PmjB", + ) + .unwrap(); + let signer = Pubkey::from_str("CRnkKQTxctQ7LHVN3yssdgJyEksBJeBrDdwZAxBtsJoZ").unwrap(); + let stake_pool = Pubkey::new_unique(); + let slot = 440_863_635; + + let mut events = Vec::new(); + for log in COMPUTE_SCORE_EVENT_LOGS { + let parsed = parse_log( + log.to_string(), + &signature, + 0, + &signer, + &stake_pool, + Some(1_787_380_325), + None, + get_epoch_from_slot(slot), + slot, + ) + .await + .expect("parse_log should not error"); + + if let Some(event) = parsed { + events.push(event); + } + } + + let event_types: Vec<&str> = events.iter().map(|e| e.event_type.as_str()).collect(); + assert_eq!( + event_types, + vec![ + "ScoreComponentsV5", + "ScoreComponentsV5", + "ScoreComponentsV5", + "ScoreComponentsV5", + "StateTransition", + ], + ); + + // The score events carry the validator they scored; the state + // transition is pool-wide and carries none. + assert!(events[..4].iter().all(|e| e.vote_account.is_some()),); + assert!(events.iter().all(|e| e.signer == signer.to_string())); + assert!(events.iter().all(|e| e.slot == slot)); + } +} diff --git a/steward-writer-service/src/main.rs b/steward-writer-service/src/main.rs index 1c8f3dc..c264f81 100644 --- a/steward-writer-service/src/main.rs +++ b/steward-writer-service/src/main.rs @@ -1,28 +1,19 @@ use std::{str::FromStr, time::Duration}; -use anchor_client::handle_program_log; use clap::{Parser, Subcommand}; -use jito_steward::{ - events::{ - AutoAddValidatorEvent, AutoRemoveValidatorEvent, DecreaseComponents, - DirectedRebalanceEvent, EpochMaintenanceEvent, InstantUnstakeComponents, RebalanceEvent, - ScoreComponents, StateTransition, - }, - score::{InstantUnstakeComponentsV3, ScoreComponentsV5}, -}; use kobe_core::db_models::steward_events::{StewardEvent, StewardEventsStore}; use kobe_core::rpc_utils::{retry_get_slot, retry_get_transactions}; +use kobe_steward_writer_service::{ + describe_version, fee_payer, get_epoch_from_slot, pair_with_statuses, parse_log, +}; use log::{debug, error, info}; use mongodb::{Client, Collection}; use solana_client::{ nonblocking::rpc_client::RpcClient, rpc_client::GetConfirmedSignaturesForAddress2Config, rpc_response::RpcConfirmedTransactionStatusWithSignature, }; -use solana_metrics::datapoint_info; -use solana_sdk::{ - commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Signature, - transaction::TransactionError, -}; +use solana_metrics::{datapoint_error, datapoint_info}; +use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Signature}; use solana_transaction_status::{ option_serializer::OptionSerializer, EncodedConfirmedTransactionWithStatusMeta, }; @@ -143,6 +134,7 @@ async fn main() { start_slot, end_slot, args.dry_run, + &args.cluster_name, ) .await { @@ -218,8 +210,15 @@ async fn listen( .first() .map(|status| Signature::from_str(&status.signature).unwrap()); - fetch_and_process_transactions(rpc_client, &rpc_signatures, stake_pool, store, dry_run) - .await?; + fetch_and_process_transactions( + rpc_client, + &rpc_signatures, + stake_pool, + store, + dry_run, + cluster_name, + ) + .await?; } } } @@ -230,13 +229,8 @@ async fn fetch_and_process_transactions( stake_pool: &Pubkey, store: &StewardEventsStore, dry_run: bool, -) -> Result< - Vec<( - RpcConfirmedTransactionStatusWithSignature, - EncodedConfirmedTransactionWithStatusMeta, - )>, - Box, -> { + cluster_name: &str, +) -> Result<(), Box> { let transaction_signatures: Vec = signatures .iter() .map(|status| Signature::from_str(&status.signature).unwrap()) @@ -246,23 +240,9 @@ async fn fetch_and_process_transactions( info!("Fetched {} transactions from rpc", transactions.len()); - let mut transaction_data = vec![]; - for tx in transactions.into_iter() { - let target_signature = if let Some(tx) = tx.transaction.transaction.decode() { - *tx.signatures.first().unwrap() - } else { - continue; - }; - let transaction_status = signatures - .iter() - .find(|status| Signature::from_str(&status.signature).unwrap() == target_signature); - if let Some(status) = transaction_status { - transaction_data.push((status.clone(), tx)); - } - } - process_transactions(&transaction_data, stake_pool, store, dry_run).await?; + let transaction_data = pair_with_statuses(signatures, transactions); - Ok(transaction_data) + process_transactions(&transaction_data, stake_pool, store, dry_run, cluster_name).await } async fn process_transactions( @@ -273,6 +253,7 @@ async fn process_transactions( stake_pool: &Pubkey, store: &StewardEventsStore, dry_run: bool, + cluster_name: &str, ) -> Result<(), Box> { // If the slot from `signatures` doesn't match the slot in `transactions`, print it out for (status, tx) in transactions.iter() { @@ -299,16 +280,18 @@ async fn process_transactions( let EncodedConfirmedTransactionWithStatusMeta { transaction, .. } = encoded_tx_with_meta; - let signer: Pubkey = match transaction.transaction.decode() { - Some(tx) => match tx.message.static_account_keys().first() { - Some(signer) => *signer, - None => { - error!("No signer found in transaction {signature}"); - continue; - } - }, + let signer: Pubkey = match fee_payer(&transaction.transaction) { + Some(signer) => signer, None => { - error!("No transaction found in encoded transaction {signature}"); + let version = describe_version(transaction.version.as_ref()); + error!("No fee payer in {version} transaction {signature}, skipping its events"); + datapoint_error!( + "steward_writer_service-unreadable_transaction", + ("signature", signature.to_string(), String), + ("version", version, String), + ("slot", *slot as i64, i64), + "cluster" => cluster_name, + ); continue; } }; @@ -362,6 +345,7 @@ async fn process_transactions( const NUM_TRANSACTIONS: usize = 1000; +#[allow(clippy::too_many_arguments)] async fn fetch_historical_program_transactions( program_id: &Pubkey, rpc_client: &RpcClient, @@ -370,6 +354,7 @@ async fn fetch_historical_program_transactions( start_slot: u64, end_slot: u64, dry_run: bool, + cluster_name: &str, ) -> Result<(), Box> { info!("Backfilling transactions between slots {start_slot} and {end_slot}"); let mut before = None; @@ -436,23 +421,9 @@ async fn fetch_historical_program_transactions( let transactions = retry_get_transactions(rpc_client, &transaction_signatures).await?; - // Align transactions with signatures - let mut transaction_data = vec![]; - for tx in transactions.into_iter() { - let target_signature = if let Some(tx) = tx.transaction.transaction.decode() { - *tx.signatures.first().unwrap() - } else { - continue; - }; - let transaction_status = valid_signatures - .iter() - .find(|status| Signature::from_str(&status.signature).unwrap() == target_signature); - if let Some(status) = transaction_status { - transaction_data.push((status.clone(), tx)); - } - } + let transaction_data = pair_with_statuses(&valid_signatures, transactions); - process_transactions(&transaction_data, stake_pool, store, dry_run).await?; + process_transactions(&transaction_data, stake_pool, store, dry_run, cluster_name).await?; if should_break { break; @@ -461,205 +432,3 @@ async fn fetch_historical_program_transactions( Ok(()) } -#[allow(clippy::too_many_arguments)] -async fn parse_log( - log: String, - signature: &Signature, - instruction_idx: u32, - signer: &Pubkey, - stake_pool: &Pubkey, - timestamp: Option, - transaction_err: Option, - epoch: u64, - slot: u64, -) -> Result, Box> { - // Parse the log - let program = jito_steward::id().to_string(); - let tx_error = transaction_err.map(|e| e.to_string()); - - // DecreaseComponents - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { - let steward_event = StewardEvent::from_decrease_components( - event, - signature, - instruction_idx, - tx_error, - epoch, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // InstantUnstakeComponents - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) - { - let steward_event = StewardEvent::from_instant_unstake_components( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // InstantUnstakeComponentsV3 - if let Ok((Some(event), _, _)) = - handle_program_log::(&program, &log) - { - let steward_event = StewardEvent::from_instant_unstake_components_v3( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // RebalanceEvent - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { - let steward_event = StewardEvent::from_rebalance_event( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // DirectedRebalanceEvent - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { - let steward_event = StewardEvent::from_directed_rebalance_event( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // ScoreComponents - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { - let steward_event = StewardEvent::from_score_components( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // ScoreComponentsV5 - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { - let steward_event = StewardEvent::from_score_components_v5( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // StateTransition - if let Ok((Some(event), _, _)) = handle_program_log::(&program, &log) { - let steward_event = StewardEvent::from_state_transition( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - slot, - ); - return Ok(Some(steward_event)); - } - - // AutoRemoveValidatorEvent - if let Ok((Some(event), _, _)) = - handle_program_log::(&program.to_string(), &log) - { - let steward_event = StewardEvent::from_auto_remove_validator_event( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - epoch, - slot, - ); - return Ok(Some(steward_event)); - } - - // AutoAddValidatorEvent - if let Ok((Some(event), _, _)) = - handle_program_log::(&program.to_string(), &log) - { - let steward_event = StewardEvent::from_auto_add_validator_event( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - epoch, - slot, - ); - return Ok(Some(steward_event)); - } - - // EpochMaintenanceEvent - if let Ok((Some(event), _, _)) = - handle_program_log::(&program.to_string(), &log) - { - let steward_event = StewardEvent::from_epoch_maintenance_event( - event, - signature, - instruction_idx, - tx_error, - signer, - stake_pool, - timestamp, - epoch, - slot, - ); - return Ok(Some(steward_event)); - } - - Ok(None) -} - -fn get_epoch_from_slot(slot: u64) -> u64 { - // Calculate the epoch from the slot - - slot / 432_000 -}