Feat/get supply

README.md

@@ -21,6 +21,8 @@ The API server exposes the following JSON-RPC methods:
 | `getVersion`                 | Returns the cluster version, Agave-compatible (`{"solana-core": "<string>"}`). See note below for the composite-string format Cloudbreak uses.              |
 | `getGenesisHash`             | Returns the cluster genesis hash as a base58 string.                                                                                                       |
 | `getVoteAccounts`            | Returns the cluster's `current` and `delinquent` vote accounts with per-voter activated stake, commission, last vote, and recent epoch credits. Optional; only available when the Vote and Stake programs are indexed. See [Vote Accounts](#vote-accounts-getvoteaccounts). |
+| `simulateTransaction`        | Simulates a transaction against indexed account state at the requested slot and returns logs, compute units, return data, requested account state, and balance changes. Supports `sigVerify`, `replaceRecentBlockhash`, `accounts`, `innerInstructions`, and `minContextSlot`. Optional; only available on a full unfiltered index. See [Simulate Transaction](#simulate-transaction-simulatetransaction). |
+| `getSupply`                  | Returns the total and circulating supply in lamports plus the non-circulating account list. Optional; only available on a full unfiltered index with `supply-tracker-enabled = true` on the indexer. See [Supply](#supply-getsupply). |
 
 Only **confirmed** and **finalized** commitment levels are fully supported. By default, requests with `processed` commitment return an error. This can be overridden via the `processed-commitment` configuration option (see [API Configuration](#api-server-cloudbreakapitoml)).
 
@@ -703,6 +705,18 @@ A snapshot captures stake at a single point in time, but activated stake drifts
 
 The recomputer detects drift by comparing the total activated stake across runs. It recomputes every 60 s and treats an epoch as converged once the total is unchanged for three consecutive polls and the indexed `EpochRewards` sysvar reports that reward distribution has finished. If that sysvar is not indexed, it converges on stability alone after roughly 30 minutes. After converging it keeps recomputing on a slower 600 s heartbeat, returning to the 60 s cadence if a late reward write or a healed ingestion gap moves the total again.
 
+### Simulate Transaction (`simulateTransaction`)
+
+`simulateTransaction` is optional and only served on a **full, unfiltered index** (empty `[programs]` include and exclude lists). Simulation must be able to load any account a transaction touches program, lookup-table, sysvar, and feature-gate accounts included so a filtered index cannot serve it. Cloudbreak checks this at startup, if the index is filtered, the method returns a `simulateTransaction is not supported on this node` error.
+
+The transaction is executed read-only against the indexed account state at the requested slot, nothing is committed. Cloudbreak reconstructs the cluster's actually-activated feature set at that slot from the on-chain feature accounts (rather than enabling all features), so compute-unit accounting and execution behaviour match mainnet. `replaceRecentBlockhash` substitutes the latest recorded blockhash before execution and reports it with its `lastValidBlockHeight`, `sigVerify` verifies signatures, `accounts` returns post-simulation state for the requested addresses, `innerInstructions` includes decoded inner instructions.
+
+### Supply (`getSupply`)
+
+`getSupply` is optional and only served on a **full, unfiltered index** (empty `[programs]` include and exclude lists), with `supply-tracker-enabled = true` set on the indexer. The indexer seeds the total supply from the snapshot bank's `capitalization` on every (re)start. While the snapshot is ingesting, accounts touched by live blocks are recorded in a small in-memory touched-account map; once the snapshot pass completes, those touches are reconciled against the startup-slot balances read back from the database and the tracker goes live. From then on it advances the total per confirmed block by reading each updated account's previous lamports from the database (owner-routed via the account owner map) and summing the resulting deltas — no full account→lamports map is kept in memory. One `supply` row carrying the total and the non-circulating lamports is persisted per confirmed block (pruned to the last 128 slots), and a background recomputer derives the non-circulating set from Agave's pinned non-circulating account and withdraw-authority lists plus locked stake accounts, refreshing on the same converge-then-heartbeat cadence as the epoch-stakes recomputer.
+
+The API serves from a polled cache. Requests return a node-unhealthy error until the bootstrap pass has completed and the non-circulating set has been computed. On a detected slot gap the tracker keeps applying block deltas but stops publishing totals until every gap slot has been repaired (requests fail as node-unhealthy in the meantime); a failed account write or a failed delta query marks it stale, and it stays fail-closed until the indexer restarts. For `finalized` commitment the response picks the newest persisted total at or below the finalized slot; `confirmed` returns the newest total. `context.slot` is the slot of the served row; the non-circulating account list may lag it by up to the recomputer heartbeat.
+
 ### Snapshot on Indexer Startup
 
 When the `[snapshot]` section (with `[snapshot.tracker_endpoint]`) is present in the indexer config, the indexer queries the cluster tracker, downloads the latest covering snapshot pair from the source the tracker reports, and processes the archives before beginning gRPC streaming. This provides fast bootstrapping of account state. Omit the entire `[snapshot]` section to skip this.
@@ -903,13 +917,14 @@ cp example.cloudbreak.integration_tests.toml cloudbreak.integration_tests.toml
 cargo run --bin integration_tests -- benchmark gpa
 cargo run --bin integration_tests -- benchmark gtabo
 cargo run --bin integration_tests -- benchmark gtabd
+cargo run --bin integration_tests -- benchmark simulate-transaction
 ```
 
 ### Commands
 
 | Command            | Description                                                                                                                          |
 | ------------------ | ------------------------------------------------------------------------------------------------------------------------------------ |
-| `benchmark <type>` | Main command. Load test with optional dual-endpoint comparison. Types: `gpa`, `gtabo`, `gtabd`, `gpa-token-owner`, `gpa-token-mint`. |
+| `benchmark <type>` | Main command. Load test with optional dual-endpoint comparison. Types: `gpa`, `gtabo`, `gtabd`, `gpa-token-owner`, `gpa-token-mint`, `simulate-transaction`. |
 | `compare`          | (Legacy) Full pubkey set comparison between two endpoints with transaction history checks.                                           |
 | `get-slot`         | (Legacy) Polls `getSlot` on rpc1 every 100ms.                                                                                        |
 

crates/api/src/http/mod.rs

@@ -7,6 +7,7 @@ use crate::http::server::HttpHandlerResponse;
 use crate::http::server::ResponseBody;
 use crate::modules::bandwidth;
 use crate::modules::cache::GpaProcessor;
+use crate::modules::supply_cache::SharedSupplySnapshot;
 use crate::modules::vote_accounts_cache::SharedStakesSnapshot;
 use crate::error::RpcError;
 use crate::query_tracker_client::QueryTrackerClient;
@@ -108,6 +109,8 @@ pub struct CloudbreakRpcState {
     pub stakes_cache: SharedStakesSnapshot,
     pub max_multiple_accounts: usize,
     pub simulation_supported: bool,
+    pub supply_supported: bool,
+    pub supply_cache: SharedSupplySnapshot,
     pub feature_set_cache: Arc<RwLock<Option<CachedFeatureSet>>>,
 }
 
@@ -130,6 +133,8 @@ impl CloudbreakRpcState {
         stakes_cache: SharedStakesSnapshot,
         max_multiple_accounts: usize,
         simulation_supported: bool,
+        supply_supported: bool,
+        supply_cache: SharedSupplySnapshot,
     ) -> Self {
         Self {
             database,
@@ -148,6 +153,8 @@ impl CloudbreakRpcState {
             stakes_cache,
             max_multiple_accounts,
             simulation_supported,
+            supply_supported,
+            supply_cache,
             feature_set_cache: Arc::new(RwLock::new(None)),
         }
     }

crates/api/src/http/rpc.rs

@@ -10,7 +10,9 @@ use hyper::body::Incoming;
 use hyper::{Request, StatusCode};
 use serde::Serialize;
 use solana_commitment_config::CommitmentConfig;
-use solana_rpc_client_api::config::{RpcAccountInfoConfig, RpcContextConfig, RpcSimulateTransactionConfig};
+use solana_rpc_client_api::config::{
+    RpcAccountInfoConfig, RpcContextConfig, RpcSimulateTransactionConfig, RpcSupplyConfig,
+};
 use std::convert::Infallible;
 use std::sync::Arc;
 use tokio::time::Instant;
@@ -169,6 +171,12 @@ async fn process_single_request(
                     .await;
             json_serialize_response(id, result, ctx).await
         }
+        "getSupply" => {
+            let config: Option<RpcSupplyConfig> =
+                extract_param(&rpc_request.params, 0).ok().flatten();
+            let result = methods::get_supply::get_supply(state, config).await;
+            json_serialize_response(id, result, ctx).await
+        }
         "getAccountInfo" => {
             let start_time = Instant::now();
 

crates/api/src/lib.rs

@@ -12,7 +12,7 @@ use cloudbreak_core::{ApiConfig, EnvironmentInfo, TryLoadConfig};
 use crate::{
     http::{CloudbreakRpcState, HeaderKeys},
     metrics::setup_metrics,
-    modules::{cache::GpaProcessor, vote_accounts_cache},
+    modules::{cache::GpaProcessor, supply_cache, vote_accounts_cache},
     query_tracker_client::QueryTrackerClient,
 };
 use std::sync::RwLock;
@@ -127,6 +127,33 @@ pub async fn run(config: &str) -> cloudbreak_core::Result<()> {
         );
     }
 
+    let supply_supported = indexer_filter.supports_simulation();
+    let supply_cache: supply_cache::SharedSupplySnapshot = Arc::default();
+    if supply_supported {
+        match supply_cache::load_latest_supply(&database).await {
+            Ok(Some(snapshot)) => {
+                info!(
+                    "Loaded initial supply snapshot ({} rows)",
+                    snapshot.rows.len()
+                );
+                *supply_cache.write().unwrap() = Arc::new(snapshot);
+            }
+            Ok(None) => {
+                tracing::warn!(
+                    "supply table is empty at startup; getSupply will fail until the \
+                     indexer processes a snapshot"
+                );
+            }
+            Err(e) => {
+                tracing::error!("Failed to load initial supply snapshot: {:?}", e);
+            }
+        }
+        supply_cache::spawn_poll_task(database.clone(), supply_cache.clone());
+        info!("getSupply: supported=true (full unfiltered index)");
+    } else {
+        info!("getSupply: supported=false (indexer filter is not a full unfiltered index)");
+    }
+
     let state = CloudbreakRpcState::new(
         database,
         queries_timeout,
@@ -144,6 +171,8 @@ pub async fn run(config: &str) -> cloudbreak_core::Result<()> {
         stakes_cache,
         max_multiple_accounts,
         simulation_supported,
+        supply_supported,
+        supply_cache,
     );
 
     info!("Server is starting...");

crates/api/src/methods/get_supply.rs

@@ -0,0 +1,104 @@
+use solana_commitment_config::CommitmentLevel;
+use solana_rpc_client_api::{
+    config::RpcSupplyConfig,
+    response::{Response as RpcResponse, RpcResponseContext, RpcSupply},
+};
+
+use crate::{error::RpcError, http::CloudbreakRpcState, methods::resolve_commitment};
+
+const MAX_SUPPLY_STALENESS_SLOTS: u64 = 150;
+
+pub async fn get_supply(
+    state: &CloudbreakRpcState,
+    config: Option<RpcSupplyConfig>,
+) -> Result<RpcResponse<RpcSupply>, RpcError> {
+    if !state.supply_supported {
+        return Err(RpcError::InvalidParamsWithMessage(
+            "getSupply is not supported on this node".to_string(),
+        ));
+    }
+
+    let config = config.unwrap_or_default();
+    let commitment = resolve_commitment(
+        config
+            .commitment
+            .map(|c| c.commitment)
+            .unwrap_or(CommitmentLevel::Finalized),
+        state.processed_commitment,
+    )?;
+
+    let snapshot = state.supply_cache.read().unwrap().clone();
+    let Some(accounts_list) = snapshot.non_circulating_accounts.as_ref() else {
+        tracing::warn!(
+            target: "get_supply",
+            "non-circulating accounts not computed yet; returning node unhealthy"
+        );
+        return Err(state.node_unhealthy());
+    };
+
+    let finalized_slot = match state
+        .slot_syncronizer_data
+        .as_ref()
+        .map(|d| d.read().unwrap().finalized_slot.slot)
+    {
+        Some(finalized) if finalized > 0 => finalized,
+        _ => crate::db_query::get_slot_data(&state.database)
+            .await
+            .map(|d| d.finalized_slot.slot)
+            .unwrap_or(0),
+    };
+
+    let row = match commitment {
+        CommitmentLevel::Finalized => snapshot
+            .rows
+            .iter()
+            .rev()
+            .find(|row| row.slot <= finalized_slot),
+        _ => snapshot.rows.last(),
+    };
+    let Some(row) = row.copied() else {
+        tracing::warn!(
+            target: "get_supply",
+            "no cached supply row for {:?} commitment; returning node unhealthy",
+            commitment
+        );
+        return Err(state.node_unhealthy());
+    };
+    let Some(non_circulating) = row.non_circulating else {
+        tracing::warn!(
+            target: "get_supply",
+            "supply row at slot {} has no non-circulating lamports yet; returning node unhealthy",
+            row.slot
+        );
+        return Err(state.node_unhealthy());
+    };
+    if finalized_slot.saturating_sub(row.slot) > MAX_SUPPLY_STALENESS_SLOTS {
+        tracing::warn!(
+            target: "get_supply",
+            "supply slot {} is more than {} slots behind finalized slot {}; returning node unhealthy",
+            row.slot,
+            MAX_SUPPLY_STALENESS_SLOTS,
+            finalized_slot
+        );
+        return Err(state.node_unhealthy());
+    }
+
+    let non_circulating_accounts = if config.exclude_non_circulating_accounts_list {
+        Vec::new()
+    } else {
+        accounts_list.clone()
+    };
+
+    Ok(RpcResponse {
+        context: RpcResponseContext {
+            slot: row.slot,
+            api_version: None,
+        },
+        value: RpcSupply {
+            total: row.total,
+            circulating: row.total.saturating_sub(non_circulating),
+            non_circulating,
+            non_circulating_accounts,
+        },
+    })
+}

crates/api/src/methods/mod.rs

@@ -14,6 +14,7 @@ pub mod genesis;
 pub mod get_account_info;
 pub mod get_balance;
 pub mod get_multiple_accounts;
+pub mod get_supply;
 pub mod get_token_account_balance;
 pub mod get_token_largest_accounts;
 pub mod get_token_supply;

crates/api/src/methods/simulate_transaction.rs

@@ -921,7 +921,7 @@ fn sysvar_account_ids() -> [Pubkey; 9] {
 fn programdata_addresses(accounts: &HashMap<Pubkey, (AccountSharedData, Slot)>) -> Vec<Pubkey> {
     let loader = solana_sdk_ids::bpf_loader_upgradeable::id();
     let mut out = Vec::new();
-    for (_key, (account, _)) in accounts {
+    for (account, _) in accounts.values() {
         if account.owner() != &loader {
             continue;
         }

crates/api/src/modules/mod.rs

@@ -5,4 +5,5 @@
 
 pub mod bandwidth;
 pub mod cache;
+pub mod supply_cache;
 pub mod vote_accounts_cache;

crates/api/src/modules/supply_cache.rs

@@ -0,0 +1,107 @@
+use std::{
+    sync::{Arc, RwLock},
+    time::Duration,
+};
+
+use cloudbreak_core::modules::supply_tracker::SUPPLY_RING_SLOTS;
+use rust_decimal::{Decimal, prelude::ToPrimitive};
+use sea_orm::{ConnectionTrait, DatabaseBackend, DatabaseConnection, Statement};
+use solana_pubkey::Pubkey;
+use tokio::task::JoinHandle;
+
+#[derive(Debug, Clone, Default)]
+pub struct SupplySnapshot {
+    pub rows: Vec<SupplyRow>,
+    pub non_circulating_accounts: Option<Vec<String>>,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct SupplyRow {
+    pub slot: u64,
+    pub total: u64,
+    pub non_circulating: Option<u64>,
+}
+
+pub type SharedSupplySnapshot = Arc<RwLock<Arc<SupplySnapshot>>>;
+
+const SUPPLY_POLL_INTERVAL: Duration = Duration::from_secs(5);
+
+fn decimal_to_u64(value: Decimal, column: &str) -> Result<u64, anyhow::Error> {
+    value
+        .to_u64()
+        .ok_or_else(|| anyhow::anyhow!("supply.{} {} does not fit in u64", column, value))
+}
+
+pub async fn load_latest_supply(
+    db: &DatabaseConnection,
+) -> Result<Option<SupplySnapshot>, anyhow::Error> {
+    let supply_rows = db
+        .query_all(Statement::from_sql_and_values(
+            DatabaseBackend::Postgres,
+            "SELECT slot, total, non_circulating_lamports FROM supply ORDER BY slot DESC LIMIT $1",
+            [(SUPPLY_RING_SLOTS as i64).into()],
+        ))
+        .await?;
+    if supply_rows.is_empty() {
+        return Ok(None);
+    }
+    let mut rows = Vec::with_capacity(supply_rows.len());
+    for row in supply_rows {
+        let slot: i64 = row.try_get("", "slot")?;
+        let total = decimal_to_u64(row.try_get("", "total")?, "total")?;
+        let non_circulating: Option<Decimal> = row.try_get("", "non_circulating_lamports")?;
+        let non_circulating = non_circulating
+            .map(|lamports| decimal_to_u64(lamports, "non_circulating_lamports"))
+            .transpose()?;
+        rows.push(SupplyRow {
+            slot: slot as u64,
+            total,
+            non_circulating,
+        });
+    }
+    rows.reverse();
+
+    let nc_row = db
+        .query_one(Statement::from_string(
+            DatabaseBackend::Postgres,
+            "SELECT accounts FROM non_circulating_accounts WHERE id = 1".to_string(),
+        ))
+        .await?;
+    let non_circulating_accounts = match nc_row {
+        Some(row) => {
+            let accounts: Vec<Vec<u8>> = row.try_get("", "accounts")?;
+            Some(
+                accounts
+                    .into_iter()
+                    .filter_map(|bytes| Pubkey::try_from(bytes.as_slice()).ok())
+                    .map(|p| p.to_string())
+                    .collect(),
+            )
+        }
+        None => None,
+    };
+
+    Ok(Some(SupplySnapshot {
+        rows,
+        non_circulating_accounts,
+    }))
+}
+
+pub fn spawn_poll_task(db: DatabaseConnection, cache: SharedSupplySnapshot) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(SUPPLY_POLL_INTERVAL).await;
+            match load_latest_supply(&db).await {
+                Ok(Some(snapshot)) => {
+                    *cache.write().unwrap() = Arc::new(snapshot);
+                }
+                Ok(None) => {
+                    tracing::debug!(target: "supply_cache", "supply table is empty; will retry");
+                }
+                Err(e) => {
+                    tracing::error!(target: "supply_cache", "failed to load supply: {:?}", e);
+                }
+            }
+        }
+    })
+}