diff --git a/Cargo.lock b/Cargo.lock index 6db54a4..89f149b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6385,6 +6385,7 @@ dependencies = [ "solana-account-decoder-client-types", "solana-clock", "solana-commitment-config", + "solana-epoch-schedule", "solana-rpc-client-api", "solana-rpc-client-types", "solana-sdk", diff --git a/crates/superbank-rpc/Cargo.toml b/crates/superbank-rpc/Cargo.toml index 70bfba9..1acec5e 100644 --- a/crates/superbank-rpc/Cargo.toml +++ b/crates/superbank-rpc/Cargo.toml @@ -40,6 +40,7 @@ bincode = "1.3.3" serde_bytes = "0.11" serde-big-array = "0.5" solana-commitment-config = "3.0.0" +solana-epoch-schedule = "3.0.0" solana-rpc-client-api = "4.0.0-rc.0" solana-rpc-client-types = "4.0.0-rc.0" solana-account-decoder-client-types = "4.0.0-rc.0" diff --git a/crates/superbank-rpc/README.md b/crates/superbank-rpc/README.md index debdfe5..20cfef9 100644 --- a/crates/superbank-rpc/README.md +++ b/crates/superbank-rpc/README.md @@ -81,11 +81,17 @@ Required files in the chosen set: Optional: - `gsfa_hot.sql` when using hot-address routing. - `token_owner_activity.sql` when using token-owner filters in `getTransactionsForAddress`. +- `epoch_schedule.sql` for warmup-aware epoch math on non-mainnet clusters (see the note below). Apply `transactions.sql` before the materialized-view schemas (`gsfa*.sql`, `signatures.sql`, and `token_owner_activity.sql`) because those views read from the transactions table. If you use `gsfa_hot.sql`, apply `gsfa_nohot.sql` instead of `gsfa.sql`, then apply `gsfa_hot.sql`. +The RPC reads the cluster's epoch schedule from the `epoch_schedule` table at startup for +warmup-aware epoch math (currently `getInflationReward`). If the table is absent or empty, it +defaults to the mainnet schedule (no warmup). Non-mainnet clusters must run the ingestor with +`RPC_URL` set so the real schedule gets stored. + ## Run ```bash diff --git a/crates/superbank-rpc/src/clickhouse/blocks.rs b/crates/superbank-rpc/src/clickhouse/blocks.rs index 24ef898..6b8ca6e 100644 --- a/crates/superbank-rpc/src/clickhouse/blocks.rs +++ b/crates/superbank-rpc/src/clickhouse/blocks.rs @@ -8,7 +8,7 @@ use std::time::Instant; use serde::Deserialize; use serde_big_array::Array; -use solana_clock::DEFAULT_SLOTS_PER_EPOCH; +use solana_epoch_schedule::EpochSchedule; use crate::processing::{ProcessingError, ProcessingResult}; @@ -36,11 +36,11 @@ use super::types::{ }; use super::util::{annotate_required_query, http_query_with_id}; -fn inflation_epoch_slot_bounds(epoch: u64) -> Option<(u64, u64)> { +fn inflation_epoch_slot_bounds(epoch: u64, schedule: &EpochSchedule) -> Option<(u64, u64)> { let next_epoch = epoch.checked_add(1)?; let following_epoch = next_epoch.checked_add(1)?; - let start_slot = next_epoch.checked_mul(DEFAULT_SLOTS_PER_EPOCH)?; - let end_slot_exclusive = following_epoch.checked_mul(DEFAULT_SLOTS_PER_EPOCH)?; + let start_slot = schedule.get_first_slot_in_epoch(next_epoch); + let end_slot_exclusive = schedule.get_first_slot_in_epoch(following_epoch); Some((start_slot, end_slot_exclusive)) } @@ -827,6 +827,7 @@ impl ClickHouseClient { &self, addresses: &[[u8; 32]], epoch: u64, + schedule: &EpochSchedule, ) -> ProcessingResult<(Vec, QueryTimings)> { const OPERATION: &str = "get_inflation_rewards_for_epoch"; @@ -834,8 +835,8 @@ impl ClickHouseClient { return Ok((Vec::new(), QueryTimings::zero())); } - let (start_slot, end_slot_exclusive) = - inflation_epoch_slot_bounds(epoch).ok_or_else(|| { + let (start_slot, end_slot_exclusive) = inflation_epoch_slot_bounds(epoch, schedule) + .ok_or_else(|| { ProcessingError::deserialization_msg(format!("epoch {epoch} is out of range")) })?; @@ -1749,6 +1750,10 @@ impl ClickHouseClient { #[cfg(test)] mod tests { + use solana_epoch_schedule::EpochSchedule; + + use crate::clickhouse::blocks::inflation_epoch_slot_bounds; + use super::{ BlockTransactionProjection, SLOT_SHARD_DIVISOR, build_block_metadata_query, build_block_transactions_query, build_blockhash_valid_query, build_transaction_count_query, @@ -1947,4 +1952,22 @@ mod tests { assert!(!query.contains("\n block_time,")); assert!(query.contains("meta_reward_pubkey")); } + + #[test] + fn inflation_bounds_mainnet_matches_default_math() { + let sched = EpochSchedule::without_warmup(); + assert_eq!( + inflation_epoch_slot_bounds(2, &sched), + Some((3 * 432_000, 4 * 432_000)) + ); + } + + #[test] + fn inflation_bounds_warmup_uses_real_schedule() { + let sched = EpochSchedule::default(); + let (start, end) = inflation_epoch_slot_bounds(2, &sched).unwrap(); + assert_eq!(start, sched.get_first_slot_in_epoch(3)); + assert_eq!(end, sched.get_first_slot_in_epoch(4)); + assert_ne!(start, 3 * 432_000); + } } diff --git a/crates/superbank-rpc/src/clickhouse/client.rs b/crates/superbank-rpc/src/clickhouse/client.rs index 4da471d..d1d9de0 100644 --- a/crates/superbank-rpc/src/clickhouse/client.rs +++ b/crates/superbank-rpc/src/clickhouse/client.rs @@ -7,7 +7,7 @@ use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; use std::time::Duration; -use clickhouse::Client as HttpClient; +use clickhouse::{Client as HttpClient, Row}; use clickhouse_rs::{ Block as TcpBlock, types::{Complex as TcpComplex, Query as TcpQuery}, @@ -16,6 +16,7 @@ use hyper_util::client::legacy::{Client as HyperClient, connect::HttpConnector}; use hyper_util::rt::TokioExecutor; use reqwest::Url; use serde::Deserialize; +use solana_epoch_schedule::EpochSchedule; use solana_sdk::pubkey::Pubkey; use tokio::sync::Semaphore; @@ -1890,6 +1891,58 @@ impl ClickHouseClient { sample ))) } + + pub async fn load_epoch_schedule(&self) -> EpochSchedule { + #[derive(Deserialize, Row)] + struct EpochScheduleRow { + slots_per_epoch: u64, + leader_schedule_slot_offset: u64, + warmup: u8, + first_normal_epoch: u64, + first_normal_slot: u64, + } + + let query = "SELECT slots_per_epoch, leader_schedule_slot_offset, warmup, \ + first_normal_epoch, first_normal_slot FROM epoch_schedule LIMIT 1"; + match self + .with_timeout("load_epoch_schedule", async { + self.client + .query(query) + .fetch_optional::() + .await + .map_err(|e| ProcessingError::database(e.to_string(), e)) + }) + .await + { + Ok(Some(row)) => { + let schedule = EpochSchedule { + slots_per_epoch: row.slots_per_epoch, + leader_schedule_slot_offset: row.leader_schedule_slot_offset, + warmup: row.warmup != 0, + first_normal_epoch: row.first_normal_epoch, + first_normal_slot: row.first_normal_slot, + }; + tracing::info!( + warmup = row.warmup, + slots_per_epoch = row.slots_per_epoch, + "loaded epoch schedule from ClickHouse" + ); + schedule + } + Ok(None) => { + tracing::info!( + "epoch_schedule table empty; defaulting to mainnet (without_warmup)" + ); + EpochSchedule::without_warmup() + } + Err(e) => { + tracing::warn!( + "failed to read epoch_schedule table: {e}; defaulting to mainnet (without_warmup)" + ); + EpochSchedule::without_warmup() + } + } + } } fn normalize_clickhouse_ddl(value: &str) -> String { diff --git a/crates/superbank-rpc/src/handlers/blocks.rs b/crates/superbank-rpc/src/handlers/blocks.rs index cbdb6a2..840f51c 100644 --- a/crates/superbank-rpc/src/handlers/blocks.rs +++ b/crates/superbank-rpc/src/handlers/blocks.rs @@ -2189,14 +2189,17 @@ pub(crate) async fn handle_get_inflation_reward( None => { let context_slot = context_slot_opt.expect("context slot must be resolved when epoch is omitted"); - (context_slot / DEFAULT_SLOTS_PER_EPOCH).saturating_sub(1) + state + .epoch_schedule + .get_epoch(context_slot) + .saturating_sub(1) } }; route.source_clickhouse(); let (reward_rows, timings) = match state .clickhouse - .get_inflation_rewards_for_epoch(&requested_pubkeys, epoch) + .get_inflation_rewards_for_epoch(&requested_pubkeys, epoch, &state.epoch_schedule) .await { Ok(result) => result, diff --git a/crates/superbank-rpc/src/server.rs b/crates/superbank-rpc/src/server.rs index 88937ef..7d5f2f6 100644 --- a/crates/superbank-rpc/src/server.rs +++ b/crates/superbank-rpc/src/server.rs @@ -176,6 +176,8 @@ pub async fn run_server(args: RpcConfig) -> RpcResult<()> { // Verify ClickHouse connection clickhouse.create_tables().await?; + let epoch_schedule = clickhouse.load_epoch_schedule().await; + if let Err(err) = metrics::force_init() { warn!("Metrics initialization failed; metrics disabled: {err}"); } @@ -287,6 +289,7 @@ pub async fn run_server(args: RpcConfig) -> RpcResult<()> { latest_block_height_cache: LatestBlockHeightCache::new(Duration::from_millis(1000)), rpc_request_timeout: Duration::from_millis(args.rpc_request_timeout_ms), emit_http_errors: args.emit_http_errors, + epoch_schedule, metrics_header_capture: MetricsHeaderCaptureConfig { capture_x_endpoint: args.metrics_capture_x_endpoint(), capture_x_rpc_node: args.metrics_capture_x_rpc_node(), diff --git a/crates/superbank-rpc/src/state.rs b/crates/superbank-rpc/src/state.rs index adb12b7..8e6c530 100644 --- a/crates/superbank-rpc/src/state.rs +++ b/crates/superbank-rpc/src/state.rs @@ -8,6 +8,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use solana_commitment_config::CommitmentLevel; +use solana_epoch_schedule::EpochSchedule; use tokio::sync::futures::OwnedNotified; use tokio::sync::{Mutex, Notify, Semaphore}; @@ -40,6 +41,7 @@ pub(crate) struct AppState { pub(crate) emit_http_errors: bool, pub(crate) metrics_header_capture: MetricsHeaderCaptureConfig, pub(crate) hydration_sem: Arc, + pub(crate) epoch_schedule: EpochSchedule, #[cfg(feature = "grpc-head-cache")] pub(crate) head_cache: Option>, /// Finalized recent-slot cache on local disk; consulted between the head diff --git a/crates/superbank-rpc/src/tests/mod.rs b/crates/superbank-rpc/src/tests/mod.rs index 000184a..3edb8e8 100644 --- a/crates/superbank-rpc/src/tests/mod.rs +++ b/crates/superbank-rpc/src/tests/mod.rs @@ -11,6 +11,7 @@ use axum::{ }; use base64::{Engine, engine::general_purpose::STANDARD}; use serde_json::{Value, json}; +use solana_epoch_schedule::EpochSchedule; use solana_sdk::{ hash::Hash, instruction::InstructionError, @@ -159,6 +160,7 @@ fn test_state_with_token_owner_activity_available(available: bool) -> Arc Arc { rpc_request_timeout: Duration::from_millis(10_000), emit_http_errors: false, metrics_header_capture: Default::default(), + epoch_schedule: EpochSchedule::default(), hydration_sem: Arc::new(Semaphore::new(8)), #[cfg(feature = "grpc-head-cache")] head_cache: None, @@ -294,6 +297,7 @@ async fn test_state_with_clickhouse_cached_signature_slot( rpc_request_timeout: Duration::from_millis(10_000), emit_http_errors: false, metrics_header_capture: Default::default(), + epoch_schedule: EpochSchedule::default(), hydration_sem: Arc::new(Semaphore::new(8)), #[cfg(feature = "grpc-head-cache")] head_cache: None, @@ -341,6 +345,7 @@ fn test_state_with_head_cache(head_cache: Arc) -> Arc { rpc_request_timeout: Duration::from_millis(10_000), emit_http_errors: false, metrics_header_capture: Default::default(), + epoch_schedule: EpochSchedule::default(), hydration_sem: Arc::new(Semaphore::new(8)), head_cache: Some(head_cache), #[cfg(feature = "disk-cache")] @@ -390,6 +395,7 @@ fn test_state_with_head_cache_and_clickhouse_url( rpc_request_timeout: Duration::from_millis(10_000), emit_http_errors: false, metrics_header_capture: Default::default(), + epoch_schedule: EpochSchedule::default(), hydration_sem: Arc::new(Semaphore::new(8)), head_cache: Some(head_cache), #[cfg(feature = "disk-cache")] @@ -446,6 +452,7 @@ async fn test_state_with_head_cache_and_cached_signature_slot( rpc_request_timeout: Duration::from_millis(10_000), emit_http_errors: false, metrics_header_capture: Default::default(), + epoch_schedule: EpochSchedule::default(), hydration_sem: Arc::new(Semaphore::new(8)), head_cache: Some(head_cache), #[cfg(feature = "disk-cache")] diff --git a/crates/superbank/README.md b/crates/superbank/README.md index eef13f9..99bbc0a 100644 --- a/crates/superbank/README.md +++ b/crates/superbank/README.md @@ -238,5 +238,6 @@ cargo run -p superbank -- --config path/to/superbank.yaml - RPC and Bigtable modes do not currently populate the `entries` table because those sources do not expose the per-entry payload Superbank needs. - Bigtable epoch ranges use the RPC epoch schedule to resolve epochs to slots; provide `RPC_URL` when using `1-10` or single-epoch ranges. - Bigtable slot lists do not require `RPC_URL` because slots are explicit. +- When `RPC_URL` is set, the ingestor also fetches the cluster's epoch schedule at startup and stores it in the `epoch_schedule` table, so the RPC can compute epoch fields correctly on non-mainnet (warmup) clusters. Without an RPC URL, no schedule is stored and the RPC falls back to the mainnet schedule. - Superbank forces `async_insert=0` by default for ClickHouse writes; enable `--clickhouse-async-insert` only when your ClickHouse profile and dependent materialized views support it. diff --git a/crates/superbank/src/clickhouse.rs b/crates/superbank/src/clickhouse.rs index 9740a46..367e4cb 100644 --- a/crates/superbank/src/clickhouse.rs +++ b/crates/superbank/src/clickhouse.rs @@ -13,7 +13,9 @@ use tokio::time::sleep; use tracing::{info, warn}; use crate::cli::Args; +use crate::commitment::parse_commitment_config; use crate::metrics; +use crate::rpc_client::build_rpc_client; #[derive(Row, Serialize)] pub(crate) struct TransactionRow { @@ -113,6 +115,15 @@ pub(crate) struct EntryRow { pub(crate) hash: Array, } +#[derive(Row, Serialize)] +struct EpochScheduleRow { + slots_per_epoch: u64, + leader_schedule_slot_offset: u64, + warmup: u8, + first_normal_epoch: u64, + first_normal_slot: u64, +} + #[derive(Clone, Copy)] pub(crate) struct ProgressSnapshot { pub(crate) processed: u64, @@ -141,6 +152,42 @@ impl InsertTables { } } +pub(crate) async fn ingest_epoch_schedule(args: &Args) -> Result<()> { + let Some(rpc_url) = &args.rpc_url else { + tracing::info!( + "RPC_URL not set, skipping epoch schedule ingest (RPC will fall back to the mainnet schedule)" + ); + return Ok(()); + }; + + let commitment = parse_commitment_config(&args.commitment)?; + let rpc = build_rpc_client( + rpc_url, + commitment, + args.rpc_timeout_secs, + args.rpc_max_inflight.max(1), + )?; + let schedule = rpc.get_epoch_schedule().await?; + + let client = build_clickhouse_client(args); + let row = EpochScheduleRow { + slots_per_epoch: schedule.slots_per_epoch, + leader_schedule_slot_offset: schedule.leader_schedule_slot_offset, + warmup: schedule.warmup as u8, + first_normal_epoch: schedule.first_normal_epoch, + first_normal_slot: schedule.first_normal_slot, + }; + insert_rows(&client, "epoch_schedule", &[row]).await?; + + info!( + slots_per_epoch = schedule.slots_per_epoch, + warmup = schedule.warmup, + "ingested epoch schedule" + ); + + Ok(()) +} + pub(crate) fn build_clickhouse_client(args: &Args) -> ClickHouseClient { let mut client = ClickHouseClient::default() .with_url(&args.clickhouse_url) diff --git a/crates/superbank/src/main.rs b/crates/superbank/src/main.rs index a23d760..4307481 100644 --- a/crates/superbank/src/main.rs +++ b/crates/superbank/src/main.rs @@ -55,6 +55,10 @@ async fn main() -> Result<()> { axum::serve(metrics_listener, app).await }); + if let Err(err) = clickhouse::ingest_epoch_schedule(&args).await { + warn!("failed to ingest epoch schedule: {err}"); + } + let ingest_result = match args.source { cli::IngestSource::Fumarole => ingest::fumarole::run_fumarole_ingest(&args).await, cli::IngestSource::Grpc => ingest::grpc::run_grpc_ingest(&args).await, diff --git a/ddl/README.md b/ddl/README.md index ac141e1..cabd4f3 100644 --- a/ddl/README.md +++ b/ddl/README.md @@ -16,6 +16,7 @@ Each folder contains the same file basenames: - `gsfa_hot.sql` - `signatures.sql` - `token_owner_activity.sql` +- `epoch_schedule.sql` Pick one folder and apply the matching schema set consistently. Apply `transactions.sql` before materialized-view files such as `gsfa*.sql`, `signatures.sql`, and diff --git a/ddl/cluster/epoch_schedule.sql b/ddl/cluster/epoch_schedule.sql new file mode 100644 index 0000000..42d8fab --- /dev/null +++ b/ddl/cluster/epoch_schedule.sql @@ -0,0 +1,28 @@ +-- SPDX-License-Identifier: AGPL-3.0-only +-- +-- Copyright 2025-2026 Triton One Limited. All rights reserved. +-- + +-- Cluster epoch schedule table. +-- Singleton: the schedule is immutable per cluster, so keying on slots_per_epoch +-- collapses every re-ingest to one row (ReplacingMergeTree needs a non-empty key). +CREATE TABLE IF NOT EXISTS default.epoch_schedule_local ON CLUSTER '{cluster}' +( + slots_per_epoch UInt64, + leader_schedule_slot_offset UInt64, + warmup UInt8, + first_normal_epoch UInt64, + first_normal_slot UInt64 +) +ENGINE = ReplacingMergeTree +ORDER BY slots_per_epoch; + +CREATE TABLE IF NOT EXISTS default.epoch_schedule ON CLUSTER '{cluster}' +( + slots_per_epoch UInt64, + leader_schedule_slot_offset UInt64, + warmup UInt8, + first_normal_epoch UInt64, + first_normal_slot UInt64 +) +ENGINE = Distributed('{cluster}', 'default', 'epoch_schedule_local', 0); diff --git a/ddl/local/epoch_schedule.sql b/ddl/local/epoch_schedule.sql new file mode 100644 index 0000000..68fd992 --- /dev/null +++ b/ddl/local/epoch_schedule.sql @@ -0,0 +1,18 @@ +-- SPDX-License-Identifier: AGPL-3.0-only +-- +-- Copyright 2025-2026 Triton One Limited. All rights reserved. +-- + +-- Local-only epoch schedule table (single-node deployments). +-- Singleton: the schedule is immutable per cluster, so keying on slots_per_epoch +-- collapses every re-ingest to one row (ReplacingMergeTree needs a non-empty key). +CREATE TABLE IF NOT EXISTS default.epoch_schedule +( + slots_per_epoch UInt64, + leader_schedule_slot_offset UInt64, + warmup UInt8, + first_normal_epoch UInt64, + first_normal_slot UInt64 +) +ENGINE = ReplacingMergeTree +ORDER BY slots_per_epoch; diff --git a/ddl/replicated/epoch_schedule.sql b/ddl/replicated/epoch_schedule.sql new file mode 100644 index 0000000..1849cd5 --- /dev/null +++ b/ddl/replicated/epoch_schedule.sql @@ -0,0 +1,28 @@ +-- SPDX-License-Identifier: AGPL-3.0-only +-- +-- Copyright 2025-2026 Triton One Limited. All rights reserved. +-- + +-- Replicated epoch schedule table. +-- Singleton: the schedule is immutable per cluster, so keying on slots_per_epoch +-- collapses every re-ingest to one row (ReplacingMergeTree needs a non-empty key). +CREATE TABLE IF NOT EXISTS default.epoch_schedule_local ON CLUSTER '{cluster}' +( + slots_per_epoch UInt64, + leader_schedule_slot_offset UInt64, + warmup UInt8, + first_normal_epoch UInt64, + first_normal_slot UInt64 +) +ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{cluster}/{database}/{table}/{shard}', '{replica}') +ORDER BY slots_per_epoch; + +CREATE TABLE IF NOT EXISTS default.epoch_schedule ON CLUSTER '{cluster}' +( + slots_per_epoch UInt64, + leader_schedule_slot_offset UInt64, + warmup UInt8, + first_normal_epoch UInt64, + first_normal_slot UInt64 +) +ENGINE = Distributed('{cluster}', 'default', 'epoch_schedule_local', 0); diff --git a/deploy/k8s/11-clickhouse-ddl-job.yaml b/deploy/k8s/11-clickhouse-ddl-job.yaml index c4d5b64..d5c09ac 100644 --- a/deploy/k8s/11-clickhouse-ddl-job.yaml +++ b/deploy/k8s/11-clickhouse-ddl-job.yaml @@ -50,6 +50,7 @@ spec: clickhouse-client --host clickhouse --user "${CLICKHOUSE_USER}" --password "${CLICKHOUSE_PASSWORD}" --multiquery < /ddl/gsfa_hot.sql clickhouse-client --host clickhouse --user "${CLICKHOUSE_USER}" --password "${CLICKHOUSE_PASSWORD}" --multiquery < /ddl/signatures.sql clickhouse-client --host clickhouse --user "${CLICKHOUSE_USER}" --password "${CLICKHOUSE_PASSWORD}" --multiquery < /ddl/token_owner_activity.sql + clickhouse-client --host clickhouse --user "${CLICKHOUSE_USER}" --password "${CLICKHOUSE_PASSWORD}" --multiquery < /ddl/epoch_schedule.sql echo "DDL applied." volumeMounts: - name: ddl