feat: ingest and store the epoch schedule for warmup-aware epoch math

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"

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

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,15 +827,16 @@ impl ClickHouseClient {
         &self,
         addresses: &[[u8; 32]],
         epoch: u64,
+        schedule: &EpochSchedule,
     ) -> ProcessingResult<(Vec<InflationRewardRecord>, QueryTimings)> {
         const OPERATION: &str = "get_inflation_rewards_for_epoch";
 
         if addresses.is_empty() {
             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);
+    }
 }

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::<EpochScheduleRow>()
+                    .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 {

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,

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(),

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<Semaphore>,
+    pub(crate) epoch_schedule: EpochSchedule,
     #[cfg(feature = "grpc-head-cache")]
     pub(crate) head_cache: Option<Arc<HeadCache>>,
     /// Finalized recent-slot cache on local disk; consulted between the head

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<AppSta
         latest_block_height_cache: height_cache,
         rpc_request_timeout: Duration::from_millis(10_000),
         emit_http_errors: false,
+        epoch_schedule: EpochSchedule::default(),
         metrics_header_capture: Default::default(),
         hydration_sem: Arc::new(Semaphore::new(8)),
         #[cfg(feature = "grpc-head-cache")]
@@ -239,6 +241,7 @@ fn test_state_with_clickhouse_url(clickhouse_url: &str) -> Arc<AppState> {
         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<HeadCache>) -> Arc<AppState> {
         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")]

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.

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<u8, 32>,
 }
 
+#[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)

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,

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