diff --git a/crates/superbank-rpc/README.md b/crates/superbank-rpc/README.md index 52a0f21..6272431 100644 --- a/crates/superbank-rpc/README.md +++ b/crates/superbank-rpc/README.md @@ -11,6 +11,7 @@ writer that matches the same schemas). - `getBlock` - `getBlockHeight` - `getSlot` +- `getEpochInfo` - `getTransactionCount` - `getLatestBlockhash` - `isBlockhashValid` @@ -32,7 +33,7 @@ Notes: - `processed` commitment is supported for a subset of methods when compiled with `--features grpc-head-cache` and enabled at runtime with `HEAD_CACHE_ENABLED=true` (see "Optional gRPC head cache" below). -- `getBlockHeight`, `getSlot`, and `getTransactionCount` accept an optional single config object as the sole param: +- `getBlockHeight`, `getSlot`, `getEpochInfo`, and `getTransactionCount` accept an optional single config object as the sole param: - `commitment`: `processed|confirmed|finalized` (defaults to `finalized`; `processed` requires the head cache) - `minContextSlot`: optional `u64`; if the server's current context slot is below this value, the call fails with JSON-RPC error `-32016` ("Minimum context slot has not been reached") and diff --git a/crates/superbank-rpc/src/handlers/blocks.rs b/crates/superbank-rpc/src/handlers/blocks.rs index 11e4632..9a7affd 100644 --- a/crates/superbank-rpc/src/handlers/blocks.rs +++ b/crates/superbank-rpc/src/handlers/blocks.rs @@ -24,9 +24,10 @@ use crate::clickhouse::{QueryTimings, StoredBlockPayload, StoredBlockRecord}; use crate::handlers::{ RouteMetric, types::{ - GetBlockHeightConfig, GetBlocksConfig, GetInflationRewardConfig, GetLatestBlockhashConfig, - GetLatestBlockhashResult, GetLatestBlockhashValue, GetSlotConfig, InflationRewardInfo, - IsBlockhashValidResult, MAX_GET_BLOCKS_RANGE, RpcContextSlot, reject_unknown_fields, + EpochInfo, GetBlockHeightConfig, GetBlocksConfig, GetInflationRewardConfig, + GetLatestBlockhashConfig, GetLatestBlockhashResult, GetLatestBlockhashValue, GetSlotConfig, + InflationRewardInfo, IsBlockhashValidResult, MAX_GET_BLOCKS_RANGE, RpcContextSlot, + reject_unknown_fields, }, }; use crate::hydration::{BlockHydrationError, hydrate_block_payload}; @@ -627,6 +628,251 @@ pub(crate) async fn handle_get_slot( Ok(json_rpc_success_response(id, json!(slot))) } +pub(crate) async fn handle_get_epoch_info( + state: Arc, + id: Value, + params: Option>, +) -> Result { + let mut route = RouteMetric::for_state("getEpochInfo", state.as_ref()); + + // getEpochInfo accepts the same optional config as getSlot: an object with + // an optional commitment and minContextSlot. + let config = match params.filter(|v| !v.is_empty()) { + None => GetSlotConfig::default(), + Some(mut params) => { + if params.len() != 1 { + route.invalid_params(); + return Ok(json_rpc_error_response( + id, + -32602, + "Invalid params: expected a single config object", + None, + )); + } + + let value = params.remove(0); + if value.is_null() { + GetSlotConfig::default() + } else if value.is_object() { + match serde_json::from_value::(value) { + Ok(config) => config, + Err(e) => { + route.invalid_params(); + return Ok(json_rpc_error_response( + id, + -32602, + format!("Invalid params: failed to parse config ({e})"), + None, + )); + } + } + } else { + route.invalid_params(); + return Ok(json_rpc_error_response( + id, + -32602, + "Invalid params: config must be an object", + None, + )); + } + } + }; + + let commitment = config.commitment.unwrap_or_default(); + + if commitment.is_processed() { + #[cfg(feature = "grpc-head-cache")] + { + if state.head_cache.is_none() { + route.invalid_params(); + return Ok(json_rpc_error_response( + id, + -32602, + "Only confirmed or finalized commitments are supported", + Some(json!({ "requestedCommitment": commitment.commitment })), + )); + } + } + #[cfg(not(feature = "grpc-head-cache"))] + { + route.invalid_params(); + return Ok(json_rpc_error_response( + id, + -32602, + "Only confirmed or finalized commitments are supported", + Some(json!({ "requestedCommitment": commitment.commitment })), + )); + } + } + + // Resolve the context slot honoring the requested commitment. Start from the + // latest ClickHouse-ingested slot, then, when the head cache holds a newer + // slot that meets the commitment (e.g. a confirmed slot not yet in + // ClickHouse), derive the snapshot from it instead. This keeps getEpochInfo + // consistent with getSlot/getBlockHeight/getTransactionCount, which all use + // the head cache for confirmed reads. With the head cache off, every field + // resolves to the finalized ClickHouse slot. Reuses TransactionCountPlan + // since the slot/count resolution is identical. Epoch math assumes the + // default schedule with no warmup, matching getInflationReward. + let clickhouse_slot = match state + .latest_slot_cache + .get_or_refresh(&state.clickhouse) + .await + { + Ok(slot) => slot, + Err(e) => { + metrics::backend_error("get_latest_finalized_slot"); + error!("Failed to fetch latest slot for getEpochInfo: {}", e); + return Ok(json_rpc_internal_error_response(id)); + } + }; + + #[cfg(feature = "grpc-head-cache")] + let plan = { + let mut plan = TransactionCountPlan::ClickHouseThroughSlot { + context_slot: clickhouse_slot, + }; + if let Some(cache) = state.head_cache.as_ref() + && let Some(overlay) = + cache.transaction_count_overlay_at_least(commitment.commitment, clickhouse_slot) + && overlay.context_slot > clickhouse_slot + { + plan = TransactionCountPlan::ClickHouseBeforeSlotPlusHead { + context_slot: overlay.context_slot, + clickhouse_before_slot: overlay.start_slot, + head_transaction_count: overlay.transaction_count, + }; + } + plan + }; + #[cfg(not(feature = "grpc-head-cache"))] + let plan = TransactionCountPlan::ClickHouseThroughSlot { + context_slot: clickhouse_slot, + }; + + let context_slot = plan.context_slot(); + match &plan { + TransactionCountPlan::ClickHouseThroughSlot { .. } => route.source_clickhouse(), + #[cfg(feature = "grpc-head-cache")] + TransactionCountPlan::ClickHouseBeforeSlotPlusHead { .. } => route.source_head_cache(), + } + + if let Some(min_context_slot) = config.min_context_slot + && context_slot < min_context_slot + { + route.rpc_error(); + return Ok(json_rpc_error_response( + id, + JSON_RPC_SERVER_ERROR_MIN_CONTEXT_SLOT_NOT_REACHED as i32, + "Minimum context slot has not been reached", + Some(json!({ "contextSlot": context_slot })), + )); + } + + // Block height for the resolved slot: the ClickHouse path reads it from the + // block-height cache; the head path reads it from the head cache, since a + // confirmed slot isn't ingested into ClickHouse yet (getBlockHeight pattern). + let block_height = match &plan { + TransactionCountPlan::ClickHouseThroughSlot { .. } => match state + .latest_block_height_cache + .get_or_refresh(context_slot, &state.clickhouse) + .await + { + Ok(Some(height)) => height, + Ok(None) => { + route.source_none(); + error!( + slot = context_slot, + "Block height unavailable for getEpochInfo" + ); + return Ok(json_rpc_internal_error_response(id)); + } + Err(e) => { + metrics::backend_error("get_block_height_by_slot"); + error!( + "Failed to query ClickHouse block height at slot {} for getEpochInfo: {}", + context_slot, e + ); + return Ok(json_rpc_internal_error_response(id)); + } + }, + #[cfg(feature = "grpc-head-cache")] + TransactionCountPlan::ClickHouseBeforeSlotPlusHead { .. } => match state + .head_cache + .as_ref() + .and_then(|cache| cache.latest_block_height_at_least(commitment.commitment)) + { + Some(height) => height, + None => { + route.source_none(); + error!( + slot = context_slot, + "Head block height unavailable for getEpochInfo" + ); + return Ok(json_rpc_internal_error_response(id)); + } + }, + }; + + // Transaction count through the resolved slot, following getTransactionCount. + let (transaction_count, timings) = match plan { + TransactionCountPlan::ClickHouseThroughSlot { context_slot } => { + match state + .clickhouse + .get_transaction_count_by_slot(context_slot) + .await + { + Ok(result) => result, + Err(e) => { + metrics::backend_error("get_transaction_count_by_slot"); + error!( + "Failed to query ClickHouse transaction count at slot {} for getEpochInfo: {}", + context_slot, e + ); + return Ok(json_rpc_internal_error_response(id)); + } + } + } + #[cfg(feature = "grpc-head-cache")] + TransactionCountPlan::ClickHouseBeforeSlotPlusHead { + context_slot, + clickhouse_before_slot, + head_transaction_count, + } => match state + .clickhouse + .get_transaction_count_before_slot(clickhouse_before_slot) + .await + { + Ok((base_transaction_count, timings)) => ( + base_transaction_count.saturating_add(head_transaction_count), + timings, + ), + Err(e) => { + metrics::backend_error("get_transaction_count_before_slot"); + error!( + "Failed to query ClickHouse transaction count before slot {} for context slot {} (getEpochInfo): {}", + clickhouse_before_slot, context_slot, e + ); + return Ok(json_rpc_internal_error_response(id)); + } + }, + }; + + let epoch_info = EpochInfo { + absolute_slot: context_slot, + block_height, + epoch: context_slot / DEFAULT_SLOTS_PER_EPOCH, + slot_index: context_slot % DEFAULT_SLOTS_PER_EPOCH, + slots_in_epoch: DEFAULT_SLOTS_PER_EPOCH, + transaction_count: Some(transaction_count), + }; + + route.success(); + let mut resp = json_rpc_success_response(id, json!(epoch_info)); + add_downstream_header(&mut resp, &timings); + Ok(resp) +} + enum TransactionCountPlan { ClickHouseThroughSlot { context_slot: u64, diff --git a/crates/superbank-rpc/src/handlers/mod.rs b/crates/superbank-rpc/src/handlers/mod.rs index 2be4244..f323f74 100644 --- a/crates/superbank-rpc/src/handlers/mod.rs +++ b/crates/superbank-rpc/src/handlers/mod.rs @@ -535,6 +535,7 @@ fn metrics_method_label(method: &str) -> &'static str { "getBlock" => "getBlock", "getBlockHeight" => "getBlockHeight", "getSlot" => "getSlot", + "getEpochInfo" => "getEpochInfo", "getTransactionCount" => "getTransactionCount", "getLatestBlockhash" => "getLatestBlockhash", "isBlockhashValid" => "isBlockhashValid", @@ -862,6 +863,7 @@ async fn dispatch_json_rpc_request( blocks::handle_get_block_height(state, id_for_dispatch, params).await } "getSlot" => blocks::handle_get_slot(state, id_for_dispatch, params).await, + "getEpochInfo" => blocks::handle_get_epoch_info(state, id_for_dispatch, params).await, "getTransactionCount" => { blocks::handle_get_transaction_count(state, id_for_dispatch, params).await } diff --git a/crates/superbank-rpc/src/handlers/types.rs b/crates/superbank-rpc/src/handlers/types.rs index d5a3613..f2dd631 100644 --- a/crates/superbank-rpc/src/handlers/types.rs +++ b/crates/superbank-rpc/src/handlers/types.rs @@ -256,6 +256,17 @@ pub(crate) struct InflationRewardInfo { pub(crate) commission: Option, } +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct EpochInfo { + pub(crate) absolute_slot: u64, + pub(crate) block_height: u64, + pub(crate) epoch: u64, + pub(crate) slot_index: u64, + pub(crate) slots_in_epoch: u64, + pub(crate) transaction_count: Option, +} + #[derive(Debug, Serialize)] pub(crate) struct GetLatestBlockhashResult { pub(crate) context: RpcContextSlot, diff --git a/crates/superbank-rpc/src/tests/mod.rs b/crates/superbank-rpc/src/tests/mod.rs index 3009fc6..6123e31 100644 --- a/crates/superbank-rpc/src/tests/mod.rs +++ b/crates/superbank-rpc/src/tests/mod.rs @@ -36,8 +36,8 @@ use crate::clickhouse::{ }; use crate::handlers::blocks::{ handle_get_block, handle_get_block_height, handle_get_block_time, handle_get_blocks, - handle_get_blocks_with_limit, handle_get_first_available_block, handle_get_health, - handle_get_inflation_reward, handle_get_latest_blockhash, handle_get_slot, + handle_get_blocks_with_limit, handle_get_epoch_info, handle_get_first_available_block, + handle_get_health, handle_get_inflation_reward, handle_get_latest_blockhash, handle_get_slot, handle_get_transaction_count, handle_is_blockhash_valid, handle_minimum_ledger_slot, }; use crate::handlers::handle_json_rpc_with_headers; @@ -45,7 +45,7 @@ use crate::handlers::signatures::{ handle_get_signature_statuses, handle_get_signatures_for_address, }; use crate::handlers::transactions::handle_get_transactions_for_address; -use crate::handlers::types::MAX_GET_BLOCKS_RANGE; +use crate::handlers::types::{EpochInfo, MAX_GET_BLOCKS_RANGE}; use crate::hydration::BlockHydrationError; use crate::hydration::build_transaction_status_meta; use crate::hydration::{ @@ -3702,6 +3702,151 @@ async fn emit_http_errors_keeps_success_http_200() { assert!(parsed.error.is_none()); } +#[tokio::test] +async fn get_epoch_info_rejects_non_object_config() { + let state = test_state(); + let request = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "getEpochInfo", + "params": ["finalized"] + }); + + let response = handle_json_rpc_value(state, &request).await; + assert_eq!(response.status(), StatusCode::OK); + let parsed = parse_json_rpc_response(response).await; + // -32602 rather than -32601 also confirms the method is wired to the handler. + assert_eq!(parsed.error.expect("invalid params").code, -32602); + assert!(parsed.result.is_none()); +} + +#[tokio::test] +async fn get_epoch_info_rejects_unknown_config_field() { + let state = test_state(); + let request = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "getEpochInfo", + "params": [{ "notARealField": true }] + }); + + let response = handle_json_rpc_value(state, &request).await; + assert_eq!(response.status(), StatusCode::OK); + let parsed = parse_json_rpc_response(response).await; + assert_eq!(parsed.error.expect("invalid params").code, -32602); +} + +#[tokio::test] +async fn get_epoch_info_rejects_more_than_one_param() { + let state = test_state(); + let response = handle_get_epoch_info(state, json!(1), Some(vec![json!({}), json!({})])) + .await + .expect("response"); + assert_eq!(response.status(), StatusCode::OK); + let parsed = parse_json_rpc_response(response).await; + assert_eq!(parsed.error.expect("invalid params").code, -32602); +} + +#[tokio::test] +async fn get_epoch_info_rejects_processed_commitment() { + // test_state has no head cache, so processed commitment is unsupported. + let state = test_state(); + let request = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "getEpochInfo", + "params": [{ "commitment": "processed" }] + }); + + let response = handle_json_rpc_value(state, &request).await; + assert_eq!(response.status(), StatusCode::OK); + let parsed = parse_json_rpc_response(response).await; + let err = parsed.error.expect("commitment rejected"); + assert_eq!(err.code, -32602); + assert_eq!( + err.message, + "Only confirmed or finalized commitments are supported" + ); +} + +#[test] +fn epoch_info_serializes_with_solana_field_names() { + let value = serde_json::to_value(EpochInfo { + absolute_slot: 500, + block_height: 480, + epoch: 1, + slot_index: 68, + slots_in_epoch: 432_000, + transaction_count: Some(1_234), + }) + .expect("serialize"); + + assert_eq!( + value, + json!({ + "absoluteSlot": 500, + "blockHeight": 480, + "epoch": 1, + "slotIndex": 68, + "slotsInEpoch": 432_000, + "transactionCount": 1_234 + }) + ); + + // transactionCount is nullable, matching the Solana spec. + let null_count = serde_json::to_value(EpochInfo { + absolute_slot: 1, + block_height: 1, + epoch: 0, + slot_index: 1, + slots_in_epoch: 432_000, + transaction_count: None, + }) + .expect("serialize"); + assert_eq!(null_count.get("transactionCount"), Some(&Value::Null)); +} + +#[cfg(feature = "grpc-head-cache")] +#[tokio::test] +async fn get_epoch_info_confirmed_uses_head_overlay_context() { + // ClickHouse is at slot 95; the head cache has a newer confirmed slot 105. + // A confirmed getEpochInfo must resolve the context slot to 105 (head), not + // 95 (ClickHouse) — proven here via the minContextSlot boundary. Mirrors + // get_transaction_count_min_context_uses_head_overlay_context. + let cache = Arc::new(HeadCache::new(32, 1024)); + cache.note_block_metadata(head_cache_metadata(100, 95, 3)); + cache.note_block_metadata(head_cache_metadata(105, 100, 5)); + cache.note_slot_commitment(105, CommitmentLevel::Confirmed); + + let state = test_state_with_head_cache_and_clickhouse_url(cache, "http://127.0.0.1:1"); + state.latest_slot_cache.value.store(95, Ordering::Relaxed); + state + .latest_slot_cache + .last_updated_ms + .store(current_time_millis(), Ordering::Relaxed); + + let response = handle_get_epoch_info( + state, + json!(1), + Some(vec![ + json!({ "commitment": "confirmed", "minContextSlot": 106 }), + ]), + ) + .await + .expect("response"); + + let parsed = parse_json_rpc_response(response).await; + let err = parsed.error.expect("error present"); + assert_eq!( + err.code, + JSON_RPC_SERVER_ERROR_MIN_CONTEXT_SLOT_NOT_REACHED as i32 + ); + assert_eq!( + err.data.and_then(|d| d.get("contextSlot").cloned()), + Some(json!(105u64)) + ); +} + #[tokio::test] async fn emit_http_errors_promotes_mixed_batch_with_server_error() { let state =