diff --git a/README.md b/README.md index 81c7b2c..fec1edb 100644 --- a/README.md +++ b/README.md @@ -16,11 +16,15 @@ The API server exposes the following JSON-RPC methods: | `getMultipleAccounts` | Batched `getAccountInfo` for up to `[server].max-multiple-accounts` pubkeys per request (default `100`). Returns `null` per position for missing or indexer-filter-excluded accounts. | | `getBalance` | Returns the lamport balance of an account. Returns `0` for missing or closed accounts (Agave-compatible). | | `getTokenAccountBalance` | Returns the `UiTokenAmount` of an SPL Token / Token-2022 account, including mint-aware decimals and UI amounts. WSOL native mint is recognised explicitly. | +| `getTokenSupply` | Returns the total supply of an SPL Token / Token-2022 mint as a `UiTokenAmount`, with mint-aware decimals and UI amounts. | +| `getTokenLargestAccounts` | Returns the 20 largest accounts holding a given mint, sorted by amount descending, each with its address and `UiTokenAmount`. | | `getSlot` | Returns the current slot at the requested commitment level. | | `getHealth` | Returns the health status of the service. | | `getVersion` | Returns the cluster version, Agave-compatible (`{"solana-core": ""}`). 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 +707,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 — including program, lookup-table, sysvar, and feature-gate accounts — 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 per-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 +919,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 ` | Main command. Load test with optional dual-endpoint comparison. Types: `gpa`, `gtabo`, `gtabd`, `gpa-token-owner`, `gpa-token-mint`. | +| `benchmark ` | 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. | diff --git a/crates/api/src/http/mod.rs b/crates/api/src/http/mod.rs index ad15522..009bda0 100644 --- a/crates/api/src/http/mod.rs +++ b/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>>, } @@ -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)), } } diff --git a/crates/api/src/http/rpc.rs b/crates/api/src/http/rpc.rs index 852a754..fda37f3 100644 --- a/crates/api/src/http/rpc.rs +++ b/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 = + 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(); diff --git a/crates/api/src/lib.rs b/crates/api/src/lib.rs index cdb7b7e..299c0a2 100644 --- a/crates/api/src/lib.rs +++ b/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..."); diff --git a/crates/api/src/methods/get_supply.rs b/crates/api/src/methods/get_supply.rs new file mode 100644 index 0000000..93aff7b --- /dev/null +++ b/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, +) -> Result, 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, + }, + }) +} diff --git a/crates/api/src/methods/mod.rs b/crates/api/src/methods/mod.rs index 104c6af..161fe35 100644 --- a/crates/api/src/methods/mod.rs +++ b/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; diff --git a/crates/api/src/methods/simulate_transaction.rs b/crates/api/src/methods/simulate_transaction.rs index e7d8484..03b1f36 100644 --- a/crates/api/src/methods/simulate_transaction.rs +++ b/crates/api/src/methods/simulate_transaction.rs @@ -921,7 +921,7 @@ fn sysvar_account_ids() -> [Pubkey; 9] { fn programdata_addresses(accounts: &HashMap) -> Vec { 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; } diff --git a/crates/api/src/modules/mod.rs b/crates/api/src/modules/mod.rs index 9186607..07d0ee5 100644 --- a/crates/api/src/modules/mod.rs +++ b/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; diff --git a/crates/api/src/modules/supply_cache.rs b/crates/api/src/modules/supply_cache.rs new file mode 100644 index 0000000..73589e3 --- /dev/null +++ b/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, + pub non_circulating_accounts: Option>, +} + +#[derive(Debug, Clone, Copy)] +pub struct SupplyRow { + pub slot: u64, + pub total: u64, + pub non_circulating: Option, +} + +pub type SharedSupplySnapshot = Arc>>; + +const SUPPLY_POLL_INTERVAL: Duration = Duration::from_secs(5); + +fn decimal_to_u64(value: Decimal, column: &str) -> Result { + 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, 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 = 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> = 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); + } + } + } + }) +} diff --git a/crates/command/src/main.rs b/crates/command/src/main.rs index 8842eb5..68ae22b 100644 --- a/crates/command/src/main.rs +++ b/crates/command/src/main.rs @@ -6,7 +6,8 @@ use clap::{Parser, Subcommand}; use cloudbreak_api::run as run_api; use cloudbreak_core::{ - Result, SnapshotConfig, TryLoadConfig, modules::account_owner_map::AccountOwnerMap, + Result, SnapshotConfig, TryLoadConfig, + modules::{account_owner_map::AccountOwnerMap, supply_tracker::SupplyTracker}, }; use cloudbreak_index::indexer::run as run_index; use cloudbreak_query_tracker::run as run_query_tracker; @@ -63,6 +64,7 @@ async fn main() -> Result<()> { None, None, AccountOwnerMap::default(), + SupplyTracker::default(), ) .await } diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index eaefe3d..b16b09b 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -408,6 +408,9 @@ pub struct IndexConfig { #[serde(default)] #[serde(rename = "accounts-owner-map-enabled")] pub accounts_owner_map_enabled: bool, + #[serde(default)] + #[serde(rename = "supply-tracker-enabled")] + pub supply_tracker_enabled: bool, } #[derive(Deserialize, Debug, Clone)] diff --git a/crates/core/src/modules/account_owner_map.rs b/crates/core/src/modules/account_owner_map.rs index 5e555f5..21deaad 100644 --- a/crates/core/src/modules/account_owner_map.rs +++ b/crates/core/src/modules/account_owner_map.rs @@ -134,6 +134,7 @@ impl AccountOwnerMap { &self, closed_accounts: Vec>, slot: u64, + defer_removals: bool, ) -> Result { let accounts = self.accounts.as_ref().expect("AccountOwnerMap not enabled"); @@ -146,7 +147,12 @@ impl AccountOwnerMap { let pubkey = Pubkey::try_from(pubkey_bytes.as_slice()).unwrap(); // Only insert closed accounts that are present in the map - if let Some(item) = map.remove(&pubkey) { + let item = if defer_removals { + map.get(&pubkey).cloned() + } else { + map.remove(&pubkey) + }; + if let Some(item) = item { pubkeys.push(pubkey_bytes.clone()); owners.push(item.owner.to_bytes().to_vec()); } @@ -211,6 +217,12 @@ impl AccountOwnerMap { }) } + pub fn get_owner(&self, pubkey: &Pubkey) -> Option { + let accounts = self.accounts.as_ref()?; + let guard = accounts.read().expect("Failed to read accounts"); + guard.get(pubkey).map(|item| item.owner) + } + /// Checks if (in case this was a tracked account) the owner has changed since the last time /// it was seen pub fn check_updated_account_owner(&self, pubkey: Pubkey, owner: Pubkey, slot: u64) -> bool { diff --git a/crates/core/src/modules/mod.rs b/crates/core/src/modules/mod.rs index 3791798..0dde9d5 100644 --- a/crates/core/src/modules/mod.rs +++ b/crates/core/src/modules/mod.rs @@ -8,3 +8,4 @@ pub mod index_identity; pub mod query_tracker_api; pub mod rpc_filter_type; pub mod service_health; +pub mod supply_tracker; diff --git a/crates/core/src/modules/supply_tracker.rs b/crates/core/src/modules/supply_tracker.rs new file mode 100644 index 0000000..365b1c2 --- /dev/null +++ b/crates/core/src/modules/supply_tracker.rs @@ -0,0 +1,375 @@ +use solana_pubkey::Pubkey; +use std::{ + collections::{HashMap, HashSet}, + sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard}, +}; + +pub const SUPPLY_RING_SLOTS: u64 = 128; + +#[derive(Clone, Copy, Debug)] +struct MemberBalance { + slot: u64, + lamports: u64, +} + +#[derive(Clone, Copy, Debug)] +struct TouchedAccount { + slot: u64, + write_version: u64, + lamports: u64, +} + +#[derive(Clone, Debug)] +pub struct NonCirculatingBalance { + pub pubkey: Pubkey, + pub slot: u64, + pub lamports: u64, +} + +#[derive(Clone, Default)] +pub struct SupplyTracker(Option>); + +struct Inner { + state: Mutex, + non_circulating: RwLock, + block_writes: tokio::sync::Mutex<()>, +} + +#[derive(Clone, Copy, Default, PartialEq)] +enum SupplyStatus { + #[default] + Bootstrapping, + Live, + GapFilling, + Stale, +} + +#[derive(Default)] +struct SupplyState { + status: SupplyStatus, + bootstrap_failed: bool, + total: u64, + slot: u64, + startup_slot: u64, + startup_touched: HashMap, + startup_zero_prev: HashSet, + gap_closes: HashMap, +} + +#[derive(Default)] +struct NonCirculatingState { + members: Option>, + balances: HashMap, +} + +impl NonCirculatingState { + fn is_member(&self, pubkey: &Pubkey) -> bool { + self.members + .as_ref() + .is_some_and(|members| members.contains(pubkey)) + } +} + +#[derive(Debug, Clone)] +pub struct SupplyCommit { + pub slot: u64, + pub total: u64, + pub non_circulating: Option, +} + +impl SupplyTracker { + pub fn new() -> Self { + Self(Some(Arc::new(Inner { + state: Mutex::new(SupplyState::default()), + non_circulating: RwLock::new(NonCirculatingState::default()), + block_writes: tokio::sync::Mutex::new(()), + }))) + } + + pub fn is_enabled(&self) -> bool { + self.0.is_some() + } + + pub fn is_tracking_deltas(&self) -> bool { + let Some(inner) = &self.0 else { return false }; + matches!( + inner.state().status, + SupplyStatus::Live | SupplyStatus::GapFilling + ) + } + + pub async fn lock_block_writes(&self) -> Option> { + let inner = self.0.as_deref()?; + Some(inner.block_writes.lock().await) + } + + pub fn observe_account(&self, pubkey: &[u8], slot: u64, lamports: u64) { + let Some(inner) = &self.0 else { return }; + let pubkey = Pubkey::try_from(pubkey).unwrap(); + if !inner.non_circulating_read().is_member(&pubkey) { + return; + } + + let mut non_circulating = inner.non_circulating_write(); + if !non_circulating.is_member(&pubkey) { + return; + } + if non_circulating + .balances + .get(&pubkey) + .is_none_or(|entry| entry.slot <= slot) + { + non_circulating + .balances + .insert(pubkey, MemberBalance { slot, lamports }); + } + } + + pub fn set_non_circulating_accounts( + &self, + accounts: Vec, + balances: Vec, + ) { + let Some(inner) = &self.0 else { return }; + let members: HashSet = accounts.into_iter().collect(); + let mut non_circulating = inner.non_circulating_write(); + non_circulating + .balances + .retain(|pubkey, _| members.contains(pubkey)); + for balance in balances { + if !members.contains(&balance.pubkey) { + continue; + } + if non_circulating + .balances + .get(&balance.pubkey) + .is_none_or(|entry| entry.slot < balance.slot) + { + non_circulating.balances.insert( + balance.pubkey, + MemberBalance { + slot: balance.slot, + lamports: balance.lamports, + }, + ); + } + } + non_circulating.members = Some(members); + } + + pub fn set_startup_total(&self, slot: u64, capitalization: u64) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state(); + if state.status == SupplyStatus::Bootstrapping && slot > state.startup_slot { + state.startup_slot = slot; + state.total = capitalization; + } + } + + pub fn startup_slot(&self) -> Option { + let inner = self.0.as_deref()?; + let state = inner.state(); + (state.status == SupplyStatus::Bootstrapping && state.startup_slot > 0) + .then_some(state.startup_slot) + } + + pub fn record_startup_touches( + &self, + slot: u64, + touches: impl IntoIterator, + ) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state(); + if state.status != SupplyStatus::Bootstrapping { + return; + } + for (pubkey, lamports, write_version) in touches { + let account = TouchedAccount { + slot, + write_version, + lamports, + }; + let entry = state.startup_touched.entry(pubkey).or_insert(account); + if (slot, write_version) > (entry.slot, entry.write_version) { + *entry = account; + } + } + } + + pub fn startup_touched_pubkeys(&self) -> Vec { + let Some(inner) = &self.0 else { + return Vec::new(); + }; + inner.state().startup_touched.keys().copied().collect() + } + + pub fn mark_bootstrap_failed(&self) -> bool { + let Some(inner) = &self.0 else { + return false; + }; + let mut state = inner.state(); + if state.status != SupplyStatus::Bootstrapping || state.bootstrap_failed { + return false; + } + state.bootstrap_failed = true; + true + } + + pub fn bootstrap_failed(&self) -> bool { + let Some(inner) = &self.0 else { + return false; + }; + inner.state().bootstrap_failed + } + + pub fn finish_bootstrap( + &self, + startup_balances: &HashMap, + ) -> Option { + let inner = self.0.as_deref()?; + let mut state = inner.state(); + if state.status != SupplyStatus::Bootstrapping + || state.startup_slot == 0 + || state.bootstrap_failed + { + return None; + } + let startup_slot = state.startup_slot; + let mut window_delta: i128 = 0; + let mut max_slot = startup_slot; + let mut zero_prev = HashSet::new(); + for (pubkey, account) in &state.startup_touched { + if account.lamports == 0 { + zero_prev.insert(*pubkey); + } + if account.slot <= startup_slot { + continue; + } + let balance = *startup_balances.get(pubkey)?; + window_delta += account.lamports as i128 - balance as i128; + max_slot = max_slot.max(account.slot); + } + state.total = (state.total as i128 + window_delta) as u64; + state.slot = state.slot.max(max_slot); + state.startup_touched = HashMap::new(); + state.startup_zero_prev = zero_prev; + state.status = SupplyStatus::Live; + Some(inner.commit(&state)) + } + + pub fn take_zero_prev(&self, pubkey: &Pubkey) -> bool { + let Some(inner) = &self.0 else { + return false; + }; + inner.state().startup_zero_prev.remove(pubkey) + } + + pub fn is_gap_filling(&self) -> bool { + let Some(inner) = &self.0 else { return false }; + inner.state().status == SupplyStatus::GapFilling + } + + pub fn record_gap_closes(&self, slot: u64, closed_accounts: &[Vec]) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state(); + if state.status != SupplyStatus::GapFilling { + return; + } + for pubkey in closed_accounts { + let pubkey = Pubkey::try_from(pubkey.as_slice()).unwrap(); + let entry = state.gap_closes.entry(pubkey).or_insert(slot); + *entry = (*entry).max(slot); + } + } + + pub fn gap_close_floor(&self, pubkey: &Pubkey) -> Option { + let inner = self.0.as_deref()?; + let state = inner.state(); + state.gap_closes.get(pubkey).copied() + } + + pub fn mark_gap(&self) -> bool { + let Some(inner) = &self.0 else { + return false; + }; + let mut state = inner.state(); + if state.status != SupplyStatus::Live { + return false; + } + state.status = SupplyStatus::GapFilling; + true + } + + pub fn finish_gap(&self) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state(); + if state.status == SupplyStatus::GapFilling { + state.status = SupplyStatus::Live; + } + } + + pub fn mark_stale(&self) -> bool { + let Some(inner) = &self.0 else { + return false; + }; + let mut state = inner.state(); + if !matches!(state.status, SupplyStatus::Live | SupplyStatus::GapFilling) { + return false; + } + state.status = SupplyStatus::Stale; + true + } + + pub fn commit_block(&self, slot: u64, block_delta: i128) -> Option { + let inner = self.0.as_deref()?; + let mut state = inner.state(); + state.slot = state.slot.max(slot); + if slot <= state.startup_slot { + return None; + } + match state.status { + SupplyStatus::Bootstrapping | SupplyStatus::Stale => None, + SupplyStatus::GapFilling | SupplyStatus::Live => { + state.total = (state.total as i128 + block_delta) as u64; + (state.status == SupplyStatus::Live).then(|| inner.commit(&state)) + } + } + } +} + +impl Inner { + fn state(&self) -> MutexGuard<'_, SupplyState> { + self.state.lock().expect("Failed to lock supply state") + } + + fn non_circulating_read(&self) -> RwLockReadGuard<'_, NonCirculatingState> { + self.non_circulating + .read() + .expect("Failed to read non-circulating state") + } + + fn non_circulating_write(&self) -> RwLockWriteGuard<'_, NonCirculatingState> { + self.non_circulating + .write() + .expect("Failed to write non-circulating state") + } + + fn commit(&self, state: &SupplyState) -> SupplyCommit { + SupplyCommit { + slot: state.slot, + total: state.total, + non_circulating: self.sum_non_circulating(), + } + } + + fn sum_non_circulating(&self) -> Option { + let non_circulating = self.non_circulating_read(); + non_circulating.members.as_ref()?; + let lamports: u128 = non_circulating + .balances + .values() + .map(|balance| balance.lamports as u128) + .sum(); + Some(lamports as u64) + } +} diff --git a/crates/index/src/db_queries.rs b/crates/index/src/db_queries.rs index 10451cd..c677270 100644 --- a/crates/index/src/db_queries.rs +++ b/crates/index/src/db_queries.rs @@ -8,20 +8,29 @@ use std::{ time::Duration, }; -use cloudbreak_core::{IndexConfig, modules::account_owner_map::AccountOwnerMap}; -use cloudbreak_entity::{accounts, slots}; +use rust_decimal::{Decimal, prelude::ToPrimitive}; +use solana_pubkey::Pubkey; use sea_orm::{ ActiveValue::Set, ColumnTrait, Condition, ConnectionTrait, DatabaseBackend, DatabaseConnection, EntityTrait, QueryFilter, Statement, Value, prelude::Expr, - sea_query::{Alias, OnConflict}, + sea_query::{Alias, ArrayType, OnConflict}, }; use tokio::{ task::JoinHandle, time::{Instant, timeout}, }; use yellowstone_grpc_proto::{geyser::CommitmentLevel, prelude::UnixTimestamp}; +use cloudbreak_core::{ + IndexConfig, + modules::{ + account_owner_map::AccountOwnerMap, + supply_tracker::{NonCirculatingBalance, SUPPLY_RING_SLOTS, SupplyCommit}, + }, +}; +use cloudbreak_entity::{accounts, slots}; +use cloudbreak_snapshot::{bytea_array, owner_pubkey_arrays, parse_pubkey, pubkey_bytea_array}; use crate::metrics; @@ -63,16 +72,20 @@ pub fn insert_closed_accounts( slot: u64, config: &IndexConfig, accounts_owner_map: AccountOwnerMap, -) -> Option> { + defer_map_removals: bool, +) -> Option> { let query_timeout = Duration::from_secs(config.database.save_block_queries_timeout); let handle = tokio::spawn(async move { let _guard = metrics::TokioTaskCounterGuard::new("insert_closed_accounts"); let start_time = Instant::now(); + let mut inserted = true; if accounts_owner_map.is_enabled() { - let result = accounts_owner_map.save_closed_accounts(pubkeys, slot).await; + let result = accounts_owner_map + .save_closed_accounts(pubkeys, slot, defer_map_removals) + .await; match result { Ok(res) => { tracing::debug!("saved {} closed accounts", res.rows_affected()); @@ -80,6 +93,7 @@ pub fn insert_closed_accounts( Err(e) => { tracing::error!(target: "save_closed_accounts_with_map", "failed to save closed accounts with map: {}", e); metrics::increment_db_errors(); + inserted = false; } } } else { @@ -131,6 +145,7 @@ pub fn insert_closed_accounts( e ); metrics::increment_db_errors(); + inserted = false; } } } @@ -138,6 +153,8 @@ pub fn insert_closed_accounts( metrics::INSERT_CLOSED_ACCOUNTS_PER_SLOT_HISTOGRAM .observe(start_time.elapsed().as_micros() as f64 / 1000.0); + + inserted }); Some(handle) @@ -354,6 +371,10 @@ pub async fn insert_recent_blockhash( config: &IndexConfig, ) { if blockhash.is_empty() { + tracing::warn!( + "insert_recent_blockhash: empty blockhash for slot {}. Expected only for snapshot-repaired self-healing blocks", + slot + ); return; } @@ -394,6 +415,184 @@ pub async fn insert_recent_blockhash( } } +pub async fn upsert_supply_row( + db: &DatabaseConnection, + commit: &SupplyCommit, + config: &IndexConfig, +) { + let query_timeout = Duration::from_secs(config.database.save_block_queries_timeout); + let supply = Statement::from_sql_and_values( + DatabaseBackend::Postgres, + "WITH upsert AS ( \ + INSERT INTO supply (slot, total, non_circulating_lamports) VALUES ($1, $2, $3) \ + ON CONFLICT (slot) DO UPDATE SET \ + total = EXCLUDED.total, \ + non_circulating_lamports = EXCLUDED.non_circulating_lamports, \ + updated_at = now() \ + ) \ + DELETE FROM supply WHERE slot < $4", + [ + Value::from(commit.slot as i64), + Value::from(Decimal::from(commit.total)), + Value::from(commit.non_circulating.map(Decimal::from)), + Value::from(commit.slot.saturating_sub(SUPPLY_RING_SLOTS) as i64), + ], + ); + let result = timeout(query_timeout, db.execute(supply)) + .await + .unwrap_or_else(|elapsed| { + tracing::error!("upsert_supply_row timeout ERROR: {}", elapsed); + Err(sea_orm::DbErr::RecordNotInserted) + }); + + if let Err(e) = result { + tracing::error!("upsert_supply_row failed for slot {}: {}", commit.slot, e); + metrics::SUPPLY_QUERY_ERRORS.inc(); + } +} + +pub async fn upsert_non_circulating_accounts( + db: &DatabaseConnection, + slot: u64, + accounts: &[Pubkey], +) { + let result = db + .execute(Statement::from_sql_and_values( + DatabaseBackend::Postgres, + "INSERT INTO non_circulating_accounts (id, slot, accounts, updated_at) \ + VALUES (1, $1, $2, now()) \ + ON CONFLICT (id) DO UPDATE SET \ + slot = EXCLUDED.slot, \ + accounts = EXCLUDED.accounts, \ + updated_at = now()", + [Value::from(slot as i64), pubkey_bytea_array(accounts)], + )) + .await; + if let Err(e) = result { + tracing::error!("upsert_non_circulating_accounts failed for slot {}: {}", slot, e); + metrics::SUPPLY_QUERY_ERRORS.inc(); + } +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct BlockSupplyDelta { + pub block_delta: i128, + pub routed_misses: u64, +} + +const LATEST_ACCOUNT_ROW_SQL: &str = r#" + SELECT lamports, slot FROM ( + SELECT lamports, slot FROM accounts + WHERE owner = v.owner AND pubkey = v.pubkey + UNION ALL + SELECT lamports, slot FROM snapshot_accounts + WHERE owner = v.owner AND pubkey = v.pubkey + ) u + ORDER BY slot DESC + LIMIT 1 +"#; + +fn numeric_array(items: Vec) -> Value { + Value::Array( + ArrayType::Decimal, + Some(Box::new( + items + .into_iter() + .map(|value| Value::Decimal(Some(Box::new(Decimal::from(value))))) + .collect(), + )), + ) +} + +pub async fn fetch_block_supply_delta( + db: &DatabaseConnection, + owners: Vec>, + pubkeys: Vec>, + new_lamports: Vec, + slot: u64, + config: &IndexConfig, +) -> Result { + let query_timeout = Duration::from_secs(config.database.save_block_queries_timeout); + + let query = db.query_one(Statement::from_sql_and_values( + DatabaseBackend::Postgres, + format!( + r#" + SELECT + COALESCE(SUM(v.new_lamports - COALESCE(prev.lamports, 0)) + FILTER (WHERE prev.slot IS NULL OR prev.slot < $4), 0) AS block_delta, + COUNT(*) FILTER (WHERE prev.slot IS NULL) AS routed_misses + FROM unnest($1::bytea[], $2::bytea[], $3::numeric[]) AS v(owner, pubkey, new_lamports) + LEFT JOIN LATERAL ({LATEST_ACCOUNT_ROW_SQL}) prev ON true + "# + ), + [ + bytea_array(owners), + bytea_array(pubkeys), + numeric_array(new_lamports), + Value::BigInt(Some(slot as i64)), + ], + )); + + let row = timeout(query_timeout, query) + .await + .map_err(|elapsed| { + sea_orm::DbErr::Custom(format!("fetch_block_supply_delta timeout: {}", elapsed)) + })?? + .ok_or_else(|| { + sea_orm::DbErr::Custom("fetch_block_supply_delta returned no row".to_string()) + })?; + + let block_delta: Decimal = row.try_get("", "block_delta")?; + let routed_misses: i64 = row.try_get("", "routed_misses")?; + let block_delta = block_delta.to_i128().ok_or_else(|| { + sea_orm::DbErr::Custom(format!("block_delta {} does not fit in i128", block_delta)) + })?; + + Ok(BlockSupplyDelta { + block_delta, + routed_misses: routed_misses as u64, + }) +} + +pub async fn fetch_non_circulating_balances( + db: &DatabaseConnection, + members: &[(Pubkey, Pubkey)], +) -> Result, sea_orm::DbErr> { + if members.is_empty() { + return Ok(Vec::new()); + } + + let (owners, pubkeys) = owner_pubkey_arrays(members); + + let rows = db + .query_all(Statement::from_sql_and_values( + DatabaseBackend::Postgres, + format!( + r#" + SELECT v.pubkey, latest.lamports, latest.slot + FROM unnest($1::bytea[], $2::bytea[]) AS v(owner, pubkey) + JOIN LATERAL ({LATEST_ACCOUNT_ROW_SQL}) latest ON true + "# + ), + [owners, pubkeys], + )) + .await?; + + rows.into_iter() + .map(|row| { + let pubkey = parse_pubkey(row.try_get("", "pubkey")?)?; + let lamports: i64 = row.try_get("", "lamports")?; + let slot: i64 = row.try_get("", "slot")?; + Ok(NonCirculatingBalance { + pubkey, + slot: slot as u64, + lamports: lamports as u64, + }) + }) + .collect() +} + /// The latest persisted slot for each commitment level, plus the finalized→confirmed lag. /// /// The `slots` table holds exactly one row per commitment (its primary key), updated to the @@ -444,7 +643,7 @@ pub async fn insert_accounts_chunk( chunk: Vec, byte_size: usize, config: &IndexConfig, -) { +) -> bool { let query_timeout = Duration::from_secs(config.database.save_block_queries_timeout); let start_time = Instant::now(); @@ -461,17 +660,23 @@ pub async fn insert_accounts_chunk( Err(sea_orm::DbErr::RecordNotInserted) }); - match result { - Ok(res) => tracing::debug!("upsert_accounts_batched: {}", res), + let inserted = match result { + Ok(res) => { + tracing::debug!("upsert_accounts_batched: {}", res); + true + } Err(e) => { tracing::error!("upsert_accounts_batched ERROR: {}", e); metrics::increment_db_errors(); + false } - } + }; let elapsed = start_time.elapsed().as_secs_f64(); if elapsed > 0.250 { tracing::debug!(target: "slow_chunk", "slow chunk: len: {}, size: {}", chunk_len, byte_size); } metrics::record_chunk_processing(elapsed, "block"); + + inserted } diff --git a/crates/index/src/indexer.rs b/crates/index/src/indexer.rs index 27ddb84..ad1f68b 100644 --- a/crates/index/src/indexer.rs +++ b/crates/index/src/indexer.rs @@ -5,7 +5,7 @@ use cloudbreak_core::{ EnvironmentInfo, IndexConfig, Result as CloudbreakResult, TryLoadConfig, - modules::account_owner_map::AccountOwnerMap, + modules::{account_owner_map::AccountOwnerMap, supply_tracker::SupplyTracker}, }; use sea_orm::{ConnectOptions, Database, DatabaseConnection}; use std::{ @@ -49,6 +49,7 @@ pub struct IndexerState { pub finalize_slot_buffer_size: Arc>, /// Used to track the accounts owner pub accounts_owner_map: AccountOwnerMap, + pub supply_tracker: SupplyTracker, } pub async fn run(config: &str) -> CloudbreakResult<()> { @@ -90,6 +91,18 @@ pub async fn run(config: &str) -> CloudbreakResult<()> { AccountOwnerMap::default() }; + let supply_tracker = if config.supply_tracker_enabled { + if !config.programs.supports_simulation() { + panic!("supply-tracker-enabled requires an empty [programs] filter"); + } + if !config.accounts_owner_map_enabled { + panic!("supply-tracker-enabled requires accounts-owner-map-enabled"); + } + SupplyTracker::new() + } else { + SupplyTracker::default() + }; + // Service health is tracked as a set of reasons (Startup is set until the startup snapshot is // processed; GapFill is set while a gap fill is in progress). let health = ServiceHealth::new(db.clone()); @@ -107,11 +120,16 @@ pub async fn run(config: &str) -> CloudbreakResult<()> { let indexer_state = IndexerState { buffer_channel_rx_len: Arc::new(Mutex::new(buffer_channel_rx.len())), snapshot_processing_state: snapshot_processing_state.clone(), - self_healing_state: SelfHealingState::new(&config, slot_finalizer.clone()), + self_healing_state: SelfHealingState::new( + &config, + slot_finalizer.clone(), + supply_tracker.clone(), + ), slot_finalizer, updated_accounts_during_startup, finalize_slot_buffer_size: finalize_slot_buffer_size.clone(), accounts_owner_map, + supply_tracker, }; // Used for the hash-checker to signal the main loop to stop @@ -151,6 +169,12 @@ pub async fn run(config: &str) -> CloudbreakResult<()> { let _epoch_stakes_handle = modules::epoch_stakes::spawn_epoch_stakes_recomputer(db.clone(), config.clone()); + let _non_circulating_handle = modules::non_circulating::spawn_non_circulating_recomputer( + db.clone(), + indexer_state.supply_tracker.clone(), + indexer_state.accounts_owner_map.clone(), + ); + operational_endpoints::self_healing::SELF_HEALING .set(indexer_state.self_healing_state.clone()) .ok() diff --git a/crates/index/src/metrics.rs b/crates/index/src/metrics.rs index 8aa0c64..3a287d5 100644 --- a/crates/index/src/metrics.rs +++ b/crates/index/src/metrics.rs @@ -7,7 +7,8 @@ use std::sync::{Once, OnceLock}; use cloudbreak_core::IndexConfig; use prometheus::{ - Counter, Histogram, HistogramOpts, HistogramVec, IntGauge, IntGaugeVec, Opts, Registry, + Counter, Histogram, HistogramOpts, HistogramVec, IntCounter, IntGauge, IntGaugeVec, Opts, + Registry, }; use tracing::error; @@ -278,6 +279,31 @@ lazy_static::lazy_static! { "cloudbreak_finalize_slot_handler_queue_size", "Size of the finalize slot handler queue" ) .expect("Failed to create finalize slot handler queue size gauge"); + + pub static ref SUPPLY_TOTAL_LAMPORTS: IntGauge = IntGauge::new( + "cloudbreak_supply_total_lamports", "Running total supply in lamports" + ) + .expect("Failed to create supply total lamports gauge"); + + pub static ref SUPPLY_SLOT: IntGauge = IntGauge::new( + "cloudbreak_supply_slot", "Slot of the last committed supply total" + ) + .expect("Failed to create supply slot gauge"); + + pub static ref SUPPLY_STALE: IntGauge = IntGauge::new( + "cloudbreak_supply_stale", "Whether the supply tracker is stale (1) or fresh (0)" + ) + .expect("Failed to create supply stale gauge"); + + pub static ref SUPPLY_ROUTED_MISSES: IntCounter = IntCounter::new( + "cloudbreak_supply_routed_misses", "Owner-routed supply prev-reads that found no previous row" + ) + .expect("Failed to create supply routed misses counter"); + + pub static ref SUPPLY_QUERY_ERRORS: IntCounter = IntCounter::new( + "cloudbreak_supply_query_errors", "Supply prev-read delta queries that failed or timed out" + ) + .expect("Failed to create supply query errors counter"); } /// We use a guard to increment the current tokio tasks metric when a task is created and @@ -414,5 +440,10 @@ pub fn register_collectors() { register!(GRPC_TOTAL_UPDATES_RECEIVED); register!(GRPC_BUFFER_CHANNEL_SIZE_SENDER); register!(FINALIZE_SLOT_DELETED_ACCOUNTS); + register!(SUPPLY_TOTAL_LAMPORTS); + register!(SUPPLY_SLOT); + register!(SUPPLY_STALE); + register!(SUPPLY_ROUTED_MISSES); + register!(SUPPLY_QUERY_ERRORS); }); } diff --git a/crates/index/src/modules/epoch_stakes.rs b/crates/index/src/modules/epoch_stakes.rs index 1055575..2781112 100644 --- a/crates/index/src/modules/epoch_stakes.rs +++ b/crates/index/src/modules/epoch_stakes.rs @@ -42,7 +42,7 @@ const EPOCH_REWARDS_SYSVAR_ID: Pubkey = Pubkey::from_str_const("SysvarEpochRewards1111111111111111111111111"); /// Latest live state per account for a given owner, across the live and snapshot tables. -const LATEST_BY_OWNER_SQL: &str = r#" +pub(crate) const LATEST_BY_OWNER_SQL: &str = r#" WITH latest AS ( SELECT DISTINCT ON (pubkey) pubkey, data, lamports FROM ( @@ -173,7 +173,7 @@ pub fn spawn_epoch_stakes_recomputer( }) } -async fn is_healthy(db: &DatabaseConnection) -> bool { +pub(crate) async fn is_healthy(db: &DatabaseConnection) -> bool { db.query_one(Statement::from_string( DatabaseBackend::Postgres, "SELECT healthy FROM service_health WHERE id = 1".to_string(), diff --git a/crates/index/src/modules/mod.rs b/crates/index/src/modules/mod.rs index 16ba93a..b55469a 100644 --- a/crates/index/src/modules/mod.rs +++ b/crates/index/src/modules/mod.rs @@ -9,6 +9,8 @@ pub mod grpc; pub mod hash_checker; pub mod health; pub mod lt_hash; +pub mod non_circulating; +pub mod non_circulating_lists; pub mod panic_handler; pub mod save_block; pub mod self_healing; diff --git a/crates/index/src/modules/non_circulating.rs b/crates/index/src/modules/non_circulating.rs new file mode 100644 index 0000000..1c83621 --- /dev/null +++ b/crates/index/src/modules/non_circulating.rs @@ -0,0 +1,180 @@ +use std::collections::HashSet; +use std::time::Duration; + +use cloudbreak_core::STAKE_PROGRAM_ID; +use cloudbreak_core::modules::account_owner_map::AccountOwnerMap; +use cloudbreak_core::modules::supply_tracker::SupplyTracker; +use futures::TryStreamExt; +use sea_orm::{ConnectionTrait, DatabaseBackend, DatabaseConnection, Statement, StreamTrait}; +use solana_program::clock::Clock; +use solana_pubkey::Pubkey; +use solana_stake_interface::state::StakeStateV2; +use tokio::task::JoinHandle; +use tokio::time::Instant; + +use crate::modules::epoch_stakes::{LATEST_BY_OWNER_SQL, is_healthy}; +use crate::{db_queries, metrics}; +use crate::modules::non_circulating_lists::{NON_CIRCULATING_ACCOUNTS, WITHDRAW_AUTHORITY}; + +const POLL_INTERVAL: Duration = Duration::from_secs(60); +const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(600); + +const SYSVAR_OWNER_ID: Pubkey = + Pubkey::from_str_const("Sysvar1111111111111111111111111111111111111"); +const CLOCK_SYSVAR_ID: Pubkey = + Pubkey::from_str_const("SysvarC1ock11111111111111111111111111111111"); + +pub fn spawn_non_circulating_recomputer( + db: DatabaseConnection, + supply_tracker: SupplyTracker, + accounts_owner_map: AccountOwnerMap, +) -> JoinHandle<()> { + tokio::spawn(async move { + let _guard = metrics::TokioTaskCounterGuard::new("non_circulating_recomputer"); + + if !supply_tracker.is_enabled() { + return; + } + + let mut last_recompute: Option = None; + let mut next_lockup_expiry: Option = None; + loop { + tokio::time::sleep(POLL_INTERVAL).await; + + if !is_healthy(&db).await { + continue; + } + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let expiry_due = next_lockup_expiry.is_some_and(|ts| now >= ts); + + if !expiry_due && last_recompute.is_some_and(|last| last.elapsed() < HEARTBEAT_INTERVAL) + { + continue; + } + + let (slot, accounts, next_expiry) = match recompute(&db).await { + Ok(result) => result, + Err(e) => { + tracing::error!( + target: "non_circulating_recomputer", + "failed to recompute non-circulating membership: {:?}", + e + ); + continue; + } + }; + let members: Vec<(Pubkey, Pubkey)> = accounts + .iter() + .filter_map(|pubkey| { + accounts_owner_map + .get_owner(pubkey) + .map(|owner| (owner, *pubkey)) + }) + .collect(); + let balances = match db_queries::fetch_non_circulating_balances(&db, &members).await { + Ok(balances) => balances, + Err(e) => { + tracing::error!( + target: "non_circulating_recomputer", + "failed to fetch non-circulating balances: {:?}", + e + ); + continue; + } + }; + + last_recompute = Some(Instant::now()); + next_lockup_expiry = next_expiry; + db_queries::upsert_non_circulating_accounts(&db, slot, &accounts).await; + supply_tracker.set_non_circulating_accounts(accounts, balances); + } + }) +} + +async fn recompute( + db: &DatabaseConnection, +) -> Result<(u64, Vec, Option), anyhow::Error> { + let start_time = Instant::now(); + let clock = read_clock(db) + .await + .ok_or_else(|| anyhow::anyhow!("Clock sysvar not found in index"))?; + + let withdraw_authorities: HashSet = WITHDRAW_AUTHORITY.iter().copied().collect(); + let mut set: HashSet = NON_CIRCULATING_ACCOUNTS.iter().copied().collect(); + let mut next_expiry: Option = None; + + let mut stream = db + .stream(Statement::from_sql_and_values( + DatabaseBackend::Postgres, + LATEST_BY_OWNER_SQL, + [STAKE_PROGRAM_ID.to_bytes().to_vec().into()], + )) + .await?; + while let Some(row) = stream.try_next().await? { + let pubkey_bytes: Vec = row.try_get("", "pubkey")?; + let data: Vec = row.try_get("", "data")?; + let Ok(pubkey) = Pubkey::try_from(pubkey_bytes.as_slice()) else { + continue; + }; + let Ok(state) = bincode::deserialize::(&data) else { + continue; + }; + let meta = match state { + StakeStateV2::Initialized(meta) => meta, + StakeStateV2::Stake(meta, _stake, _flags) => meta, + _ => continue, + }; + let in_force = meta.lockup.is_in_force(&clock, None); + if in_force + && meta.lockup.epoch <= clock.epoch + && meta.lockup.unix_timestamp > clock.unix_timestamp + { + let ts = meta.lockup.unix_timestamp; + next_expiry = Some(next_expiry.map_or(ts, |current| current.min(ts))); + } + if in_force || withdraw_authorities.contains(&meta.authorized.withdrawer) { + set.insert(pubkey); + } + } + + tracing::debug!( + target: "non_circulating_recomputer", + "recomputed membership ({} accounts) in {:.3}s", + set.len(), + start_time.elapsed().as_secs_f64() + ); + Ok((clock.slot, set.into_iter().collect(), next_expiry)) +} + +async fn read_clock(db: &DatabaseConnection) -> Option { + let row = db + .query_one(Statement::from_sql_and_values( + DatabaseBackend::Postgres, + r#" + SELECT data FROM ( + SELECT slot, data, lamports FROM accounts + WHERE owner = $1 AND pubkey = $2 + UNION ALL + SELECT slot, data, lamports FROM snapshot_accounts + WHERE owner = $1 AND pubkey = $2 + ) AS u + WHERE lamports > 0 + ORDER BY slot DESC + LIMIT 1 + "#, + [ + SYSVAR_OWNER_ID.to_bytes().to_vec().into(), + CLOCK_SYSVAR_ID.to_bytes().to_vec().into(), + ], + )) + .await + .ok() + .flatten()?; + + let data: Vec = row.try_get("", "data").ok()?; + bincode::deserialize::(&data).ok() +} diff --git a/crates/index/src/modules/non_circulating_lists.rs b/crates/index/src/modules/non_circulating_lists.rs new file mode 100644 index 0000000..43bb08c --- /dev/null +++ b/crates/index/src/modules/non_circulating_lists.rs @@ -0,0 +1,349 @@ +use solana_pubkey::Pubkey; + +pub const AGAVE_SOURCE_COMMIT: &str = "e8320a0b376d77a90396b04ee07aaba4e0918f15"; +pub const AGAVE_SOURCE_DATE: &str = "2026-07-16"; + +pub static NON_CIRCULATING_ACCOUNTS: &[Pubkey] = &[ + Pubkey::from_str_const("9huDUZfxoJ7wGMTffUE7vh1xePqef7gyrLJu9NApncqA"), + Pubkey::from_str_const("GK2zqSsXLA2rwVZk347RYhh6jJpRsCA69FjLW93ZGi3B"), + Pubkey::from_str_const("CWeRmXme7LmbaUWTZWFLt6FMnpzLCHaQLuR2TdgFn4Lq"), + Pubkey::from_str_const("HCV5dGFJXRrJ3jhDYA4DCeb9TEDTwGGYXtT3wHksu2Zr"), + Pubkey::from_str_const("14FUT96s9swbmH7ZjpDvfEDywnAYy9zaNhv4xvezySGu"), + Pubkey::from_str_const("HbZ5FfmKWNHC7uwk6TF1hVi6TCs7dtYfdjEcuPGgzFAg"), + Pubkey::from_str_const("C7C8odR8oashR5Feyrq2tJKaXL18id1dSj2zbkDGL2C2"), + Pubkey::from_str_const("Eyr9P5XsjK2NUKNCnfu39eqpGoiLFgVAv1LSQgMZCwiQ"), + Pubkey::from_str_const("DE1bawNcRJB9rVm3buyMVfr8mBEoyyu73NBovf2oXJsJ"), + Pubkey::from_str_const("CakcnaRDHka2gXyfbEd2d3xsvkJkqsLw2akB3zsN1D2S"), + Pubkey::from_str_const("7Np41oeYqPefeNQEHSv1UDhYrehxin3NStELsSKCT4K2"), + Pubkey::from_str_const("GdnSyH3YtwcxFvQrVVJMm1JhTS4QVX7MFsX56uJLUfiZ"), + Pubkey::from_str_const("Mc5XB47H3DKJHym5RLa9mPzWv5snERsF3KNv5AauXK8"), + Pubkey::from_str_const("7cvkjYAkUYs4W8XcXsca7cBrEGFeSUjeZmKoNBvEwyri"), + Pubkey::from_str_const("AG3m2bAibcY8raMt4oXEGqRHwX4FWKPPJVjZxn1LySDX"), + Pubkey::from_str_const("5XdtyEDREHJXXW1CTtCsVjJRjBapAwK78ZquzvnNVRrV"), + Pubkey::from_str_const("6yKHERk8rsbmJxvMpPuwPs1ct3hRiP7xaJF2tvnGU6nK"), + Pubkey::from_str_const("CHmdL15akDcJgBkY6BP3hzs98Dqr6wbdDC5p8odvtSbq"), + Pubkey::from_str_const("FR84wZQy3Y3j2gWz6pgETUiUoJtreMEuWfbg6573UCj9"), + Pubkey::from_str_const("5q54XjQ7vDx4y6KphPeE97LUNiYGtP55spjvXAWPGBuf"), + Pubkey::from_str_const("3o6xgkJ9sTmDeQWyfj3sxwon18fXJB9PV5LDc8sfgR4a"), + Pubkey::from_str_const("GumSE5HsMV5HCwBTv2D2D81yy9x17aDkvobkqAfTRgmo"), + Pubkey::from_str_const("AzVV9ZZDxTgW4wWfJmsG6ytaHpQGSe1yz76Nyy84VbQF"), + Pubkey::from_str_const("8CUUMKYNGxdgYio5CLHRHyzMEhhVRMcqefgE6dLqnVRK"), + Pubkey::from_str_const("CQDYc4ET2mbFhVpgj41gXahL6Exn5ZoPcGAzSHuYxwmE"), + Pubkey::from_str_const("5PLJZLJiRR9vf7d1JCCg7UuWjtyN9nkab9uok6TqSyuP"), + Pubkey::from_str_const("7xJ9CLtEAcEShw9kW2gSoZkRWL566Dg12cvgzANJwbTr"), + Pubkey::from_str_const("BuCEvc9ze8UoAQwwsQLy8d447C8sA4zeVtVpc6m5wQeS"), + Pubkey::from_str_const("8ndGYFjav6NDXvzYcxs449Aub3AxYv4vYpk89zRDwgj7"), + Pubkey::from_str_const("8W58E8JVJjH1jCy5CeHJQgvwFXTyAVyesuXRZGbcSUGG"), + Pubkey::from_str_const("GNiz4Mq886bTNDT3pijGsu2gbw6it7sqrwncro45USeB"), + Pubkey::from_str_const("GhsotwFMH6XUrRLJCxcx62h7748N2Uq8mf87hUGkmPhg"), + Pubkey::from_str_const("Fgyh8EeYGZtbW8sS33YmNQnzx54WXPrJ5KWNPkCfWPot"), + Pubkey::from_str_const("8UVjvYyoqP6sqcctTso3xpCdCfgTMiv3VRh7vraC2eJk"), + Pubkey::from_str_const("BhvLngiqqKeZ8rpxch2uGjeCiC88zzewoWPRuoxpp1aS"), + Pubkey::from_str_const("63DtkW7zuARcd185EmHAkfF44bDcC2SiTSEj2spLP3iA"), + Pubkey::from_str_const("GvpCiTgq9dmEeojCDBivoLoZqc4AkbUDACpqPMwYLWKh"), + Pubkey::from_str_const("7Y8smnoUrYKGGuDq2uaFKVxJYhojgg7DVixHyAtGTYEV"), + Pubkey::from_str_const("DUS1KxwUhUyDKB4A81E8vdnTe3hSahd92Abtn9CXsEcj"), + Pubkey::from_str_const("F9MWFw8cnYVwsRq8Am1PGfFL3cQUZV37mbGoxZftzLjN"), + Pubkey::from_str_const("8vqrX3H2BYLaXVintse3gorPEM4TgTwTFZNN1Fm9TdYs"), + Pubkey::from_str_const("CUageMFi49kzoDqtdU8NvQ4Bq3sbtJygjKDAXJ45nmAi"), + Pubkey::from_str_const("5smrYwb1Hr2T8XMnvsqccTgXxuqQs14iuE8RbHFYf2Cf"), + Pubkey::from_str_const("xQadXQiUTCCFhfHjvQx1hyJK6KVWr1w2fD6DT3cdwj7"), + Pubkey::from_str_const("8DE8fqPfv1fp9DHyGyDFFaMjpopMgDeXspzoi9jpBJjC"), + Pubkey::from_str_const("3itU5ME8L6FDqtMiRoUiT1F7PwbkTtHBbW51YWD5jtjm"), + Pubkey::from_str_const("AsrYX4FeLXnZcrjcZmrASY2Eq1jvEeQfwxtNTxS5zojA"), + Pubkey::from_str_const("8rT45mqpuDBR1vcnDc9kwP9DrZAXDR4ZeuKWw3u1gTGa"), + Pubkey::from_str_const("nGME7HgBT6tAJN1f6YuCCngpqT5cvSTndZUVLjQ4jwA"), + Pubkey::from_str_const("CzAHrrrHKx9Lxf6wdCMrsZkLvk74c7J2vGv8VYPUmY6v"), + Pubkey::from_str_const("AzHQ8Bia1grVVbcGyci7wzueSWkgvu7YZVZ4B9rkL5P6"), + Pubkey::from_str_const("FiWYY85b58zEEcPtxe3PuqzWPjqBJXqdwgZeqSBmT9Cn"), + Pubkey::from_str_const("GpxpMVhrBBBEYbEJxdR62w3daWz444V7m6dxYDZKH77D"), + Pubkey::from_str_const("3bTGcGB9F98XxnrBNftmmm48JGfPgi5sYxDEKiCjQYk3"), + Pubkey::from_str_const("8pNBEppa1VcFAsx4Hzq9CpdXUXZjUXbvQwLX2K7QsCwb"), + Pubkey::from_str_const("HKJgYGTTYYR2ZkfJKHbn58w676fKueQXmvbtpyvrSM3N"), + Pubkey::from_str_const("3jnknRabs7G2V9dKhxd2KP85pNWXKXiedYnYxtySnQMs"), + Pubkey::from_str_const("4sxwau4mdqZ8zEJsfryXq4QFYnMJSCp3HWuZQod8WU5k"), + Pubkey::from_str_const("Fg12tB1tz8w6zJSQ4ZAGotWoCztdMJF9hqK8R11pakog"), + Pubkey::from_str_const("GEWSkfWgHkpiLbeKaAnwvqnECGdRNf49at5nFccVey7c"), + Pubkey::from_str_const("CND6ZjRTzaCFVdX7pSSWgjTfHZuhxqFDoUBqWBJguNoA"), + Pubkey::from_str_const("2WWb1gRzuXDd5viZLQF7pNRR6Y7UiyeaPpaL35X6j3ve"), + Pubkey::from_str_const("BUnRE27mYXN9p8H1Ay24GXhJC88q2CuwLoNU2v2CrW4W"), + Pubkey::from_str_const("CsUqV42gVQLJwQsKyjWHqGkfHarxn9hcY4YeSjgaaeTd"), + Pubkey::from_str_const("5khMKAcvmsFaAhoKkdg3u5abvKsmjUQNmhTNP624WB1F"), + Pubkey::from_str_const("GpYnVDgB7dzvwSgsjQFeHznjG6Kt1DLBFYrKxjGU1LuD"), + Pubkey::from_str_const("DQQGPtj7pphPHCLzzBuEyDDQByUcKGrsJdsH7SP3hAug"), + Pubkey::from_str_const("FwfaykN7ACnsEUDHANzGHqTGQZMcGnUSsahAHUqbdPrz"), + Pubkey::from_str_const("JCwT5Ygmq3VeBEbDjL8s8E82Ra2rP9bq45QfZE7Xyaq7"), + Pubkey::from_str_const("H3Ni7vG1CsmJZdTvxF7RkAf9UM5qk4RsohJsmPvtZNnu"), + Pubkey::from_str_const("CVgyXrbEd1ctEuvq11QdpnCQVnPit8NLdhyqXQHLprM2"), + Pubkey::from_str_const("EAJJD6nDqtXcZ4DnQb19F9XEz8y8bRDHxbWbahatZNbL"), + Pubkey::from_str_const("6o5v1HC7WhBnLfRHp8mQTtCP2khdXXjhuyGyYEoy2Suy"), + Pubkey::from_str_const("3ZrsTmNM6AkMcqFfv3ryfhQ2jMfqP64RQbqVyAaxqhrQ"), + Pubkey::from_str_const("6zw7em7uQdmMpuS9fGz8Nq9TLHa5YQhEKKwPjo5PwDK4"), + Pubkey::from_str_const("CuatS6njAcfkFHnvai7zXCs7syA9bykXWsDCJEWfhjHG"), + Pubkey::from_str_const("Hz9nydgN1k15wnwffKX7CSmZp4VFTnTwLXAEdomFGNXy"), + Pubkey::from_str_const("Ep5Y58PaSyALPrdFxDVAdfKtVdP55vApvsWjb3jSmXsG"), + Pubkey::from_str_const("EziVYi3Sv5kJWxmU77PnbrT8jmkVuqwdiFLLzZpLVEn7"), + Pubkey::from_str_const("H1rt8KvXkNhQExTRfkY8r9wjZbZ8yCih6J4wQ5Fz9HGP"), + Pubkey::from_str_const("6nN69B4uZuESZYxr9nrLDjmKRtjDZQXrehwkfQTKw62U"), + Pubkey::from_str_const("Hm9JW7of5i9dnrboS8pCUCSeoQUPh7JsP1rkbJnW7An4"), + Pubkey::from_str_const("5D5NxsNVTgXHyVziwV7mDFwVDS6voaBsyyGxUbhQrhNW"), + Pubkey::from_str_const("EMAY24PrS6rWfvpqffFCsTsFJypeeYYmtUc26wdh3Wup"), + Pubkey::from_str_const("Br3aeVGapRb2xTq17RU2pYZCoJpWA7bq6TKBCcYtMSmt"), + Pubkey::from_str_const("BUjkdqUuH5Lz9XzcMcR4DdEMnFG6r8QzUMBm16Rfau96"), + Pubkey::from_str_const("Es13uD2p64UVPFpEWfDtd6SERdoNR2XVgqBQBZcZSLqW"), + Pubkey::from_str_const("AVYpwVou2BhdLivAwLxKPALZQsY7aZNkNmGbP2fZw7RU"), + Pubkey::from_str_const("DrKzW5koKSZp4mg4BdHLwr72MMXscd2kTiWgckCvvPXz"), + Pubkey::from_str_const("9hknftBZAQL4f48tWfk3bUEV5YSLcYYtDRqNmpNnhCWG"), + Pubkey::from_str_const("GLUmCeJpXB8veNcchPwibkRYwCwvQbKodex5mEjrgToi"), + Pubkey::from_str_const("9S2M3UYPpnPZTBtbcUvehYmiWFK3kBhwfzV2iWuwvaVy"), + Pubkey::from_str_const("HUAkU5psJXZuw54Lrg1ksbXzHv2fzczQ9sNbmisVMeJU"), + Pubkey::from_str_const("GK8R4uUmrawcREZ5xJy5dAzVV5V7aFvYg77id37pVTK"), + Pubkey::from_str_const("4vuWt1oHRqLMhf8Nv1zyEXZsYaeK7dipwrfKLoYU9Riq"), + Pubkey::from_str_const("EMhn1U3TMimW3bvWYbPUvN2eZnCfsuBN4LGWhzzYhiWR"), + Pubkey::from_str_const("BsKsunvENxAraBrL77UfAn1Gi7unVEmQAdCbhsjUN6tU"), + Pubkey::from_str_const("CTvhdUVf8KNyMbyEdnvRrBCHJjBKtQwkbj6zwoqcEssG"), + Pubkey::from_str_const("3fV2GaDKa3pZxyDcpMh5Vrh2FVAMUiWUKbYmnBFv8As3"), + Pubkey::from_str_const("4pV47TiPzZ7SSBPHmgUvSLmH9mMSe8tjyPhQZGbi1zPC"), + Pubkey::from_str_const("P8aKfWQPeRnsZtpBrwWTYzyAoRk74KMz56xc6NEpC4J"), + Pubkey::from_str_const("HuqDWJodFhAEWh6aWdsDVUqsjRket5DYXMYyDYtD8hdN"), + Pubkey::from_str_const("Ab1UcdsFXZVnkSt1Z3vcYU65GQk5MvCbs54SviaiaqHb"), + Pubkey::from_str_const("Dc2oHxFXQaC2QfLStuU7txtD3U5HZ82MrCSGDooWjbsv"), + Pubkey::from_str_const("3iPvAS4xdhYr6SkhVDHCLr7tJjMAFK4wvvHWJxFQVg15"), + Pubkey::from_str_const("GmyW1nqYcrw7P7JqrcyP9ivU9hYNbrgZ1r5SYJJH41Fs"), + Pubkey::from_str_const("E8jcgWvrvV7rwYHJThwfiBeQ8VAH4FgNEEMG9aAuCMAq"), + Pubkey::from_str_const("CY7X5o3Wi2eQhTocLmUS6JSWyx1NinBfW7AXRrkRCpi8"), + Pubkey::from_str_const("HQJtLqvEGGxgNYfRXUurfxV8E1swvCnsbC3456ik27HY"), + Pubkey::from_str_const("9xbcBZoGYFnfJZe81EDuDYKUm8xGkjzW8z4EgnVhNvsv"), + Pubkey::from_str_const("4sNBQyPbJCQyUimBueZkGWnLVqds4rWkm7eXyi9WskGU"), + Pubkey::from_str_const("AZWdNvnZxJnbcT8ZzonpN19AZJadxPdUxSiCEDTJzu8L"), + Pubkey::from_str_const("5kFTzLuM2VgFdb6x16smnY3JWoVdPxNZVFAqeVgjSTUP"), + Pubkey::from_str_const("Gbz6wkNFus8SNEkWGNNENLv9NFwVvF1pWVDpaVKUWcMh"), + Pubkey::from_str_const("AksPzoA9DKCipgdhHjhUzQJe4iEniCvBoEfvayuFA3BN"), + Pubkey::from_str_const("DVhs8YHWrvhhGxoefDNY9KotqtEEnjnSAK8MYGL2Q7X"), + Pubkey::from_str_const("2bvGnYAPSV8pa1H3vRYr5tPAXktP4DkFACHfAgqyyfhd"), + Pubkey::from_str_const("2xP8YQ3sVmfNPtGM17tZi7Lr4vsPUiN6mHLx42roazG5"), + Pubkey::from_str_const("3p9ZxnrSFTkXVrT3KnYg2tT6asnysDApEFB5DRkdeAhB"), + Pubkey::from_str_const("wRVP5MYuqP8HJ1Q8RCJ5NzUraL3DxCKPGMKSBd5iQH1"), + Pubkey::from_str_const("513qFSVgmAQFBDsnyFCM1MrVKBrWiDgb4nXrdGsqa7Z4"), + Pubkey::from_str_const("619qLS85ieR4qh7MGNLLZyLefN7hMi2FDVb5cmX1nisb"), + Pubkey::from_str_const("768NcPfBJpFBtjDAbYZLFkej6ca1W5jeArsW5MtdF8S9"), + Pubkey::from_str_const("EWAjC8a9VPbALSM3D6tGsbRfgDV48kRuHZPu8qtYSNDv"), + Pubkey::from_str_const("H8wurbnaaXsgtrjqkNH1HhncUPUhTLAmKUHkwMeyqmfN"), + Pubkey::from_str_const("HBfi37TwD4kMa1WrAWwXp3ZaFbZQ1g3XxWwNZs8QsCpY"), + Pubkey::from_str_const("6B7mXMM6BixHvDpPAPLSKweLyCXcbtprkfsw3HfMUSjZ"), + Pubkey::from_str_const("5TuV9WpmESXNfTNqasXVehoXpQy65WUHBbzgKPXPLwWx"), + Pubkey::from_str_const("FB6VmiYFnVGp1uKXA3WbNsqg9neGYVpwYYiB9q8bFRrQ"), + Pubkey::from_str_const("7Anoa4ZRiq8qaaiEnhmdpXyTEmBjZASXGfnQqVDynNie"), + Pubkey::from_str_const("6dXeE5hS8bQKeqZsc18ewCyqHimnhCBAvVJBusnRqa2F"), + Pubkey::from_str_const("Ds69ZQPb3D3aVPXdN5REyzALBrzJLdruJZ1cwyfyoEEx"), + Pubkey::from_str_const("5HA8QV7tp59iNpfjs8f84LGUGX4imaynMFSKWHUcTrMT"), + Pubkey::from_str_const("EPfiDzgbdgXdyqqwbYFMqU6Qvfx2J9Zf2J9noXBYYbbx"), + Pubkey::from_str_const("DEGyTHFXmYuyANRDYRoEcShBXLonWNdBRpNzbFZBuzhY"), + Pubkey::from_str_const("9xz1vZSWgY6TFPLZgPM2WJjDM1KiPXcALnjYFHRTkYiK"), + Pubkey::from_str_const("6n1mSmsdGFCkEyyHe4wtgEigbwhiwYRszrerMW9YRyYF"), + Pubkey::from_str_const("8rp9vcJG1Uwk25g9XfQNHysfGBYcHizVEJvavqiffB6P"), + Pubkey::from_str_const("BVsJuJbhXmxPB9XPZrYMXs1SfuZuo998u1jQ9emVp8YD"), + Pubkey::from_str_const("Fas9FrnngyVjdMZ6r8p8SsfyFDtHLMmrw1dyBxRWWVTF"), + Pubkey::from_str_const("6Lfk4e6yxXGcFNMHue5u8KQM9hWYKjbdMFZ2n7SHHd6L"), + Pubkey::from_str_const("6WuU6DWoM7ZpgUVL5TR97Fgose7vLgrv4Qvg1Yhqv8Xw"), + Pubkey::from_str_const("6mXiGJbQbX726ZhrMSmbLCF81E8rkSfo1pvBWyn1asG5"), + Pubkey::from_str_const("2UdDthRdtZKMjEx9oQYnzuLdWy3sa4XKeaqihdXkXFza"), + Pubkey::from_str_const("33XwSCq2Ft5tTCsGK7kTyMCJ4hqufDZT2REFAWXmsia8"), + Pubkey::from_str_const("B5tRZWRCTCLANkY5zLnxEXAn8S2exJD9EkH3cacEYqmE"), + Pubkey::from_str_const("CfCvwXZk1sLSqzuoFx1K7KG1puMofgwHA3tSpcArVysX"), + Pubkey::from_str_const("C4fATCN93YL5iYS3c2MrHeDmkMD6waHuBXLqsri5q5AF"), + Pubkey::from_str_const("HbJUXNHXLzpucDr9wgjuoqAwK6XNKEcqi5h29MkkuSBQ"), + Pubkey::from_str_const("G8EBx1Qo5W8731nmaBSGYyvU96onpEP5adrVSjt4vnLF"), + Pubkey::from_str_const("4dc5ty7TNod2EQfx4bPnDofPbQk5cELcn3it47qnLizL"), + Pubkey::from_str_const("4VCET97mE8ixxAPRYabM7S7dvkiAFZ8ZzEoH6kbmfYus"), + Pubkey::from_str_const("DwqBsPgtAHqQUmRN3WF5YTuLKhK7dqzrb9eRJ4UVRNLf"), + Pubkey::from_str_const("26FqcAbTLvr6cVXk2ktPu9hpeMYWukY7uUwNCtTwu7Wb"), + Pubkey::from_str_const("B22zqCsZL6PDsi4fftmcjj5k7hFjhH1BM8Dcz4fXfFXK"), + Pubkey::from_str_const("9ahLw5LFpeTuBykK3A5aiFZ1CidTY7z4YxG2kmQdSGon"), + Pubkey::from_str_const("8M731ZthMdeVoMFGRaPnUBzR7Ze6qY4zEcQ7WALEAn5i"), + Pubkey::from_str_const("A3aaVsuX7BfNCUQbuepQSaLs9KNB7c6djG6u4UuJzwwP"), + Pubkey::from_str_const("DeXE7LTUqsC3B9kSACfFdW5TPG1eT9n1Z6gVB9dJNxqz"), + Pubkey::from_str_const("BtvSeDfZNfA3VEH6baFprfx6JcS6JX8jdt6NffCxoJGV"), + Pubkey::from_str_const("8o8trB6XdG8o4qo29De2mcooY5E3pW2rvMMkeBwJHkUD"), + Pubkey::from_str_const("CSSxN8vQYqSjg5GV6ZUGSM65dXqhytLzANLpMrBZjLKP"), + Pubkey::from_str_const("5A2yoxdGdC3B3QC39pkXpjx2ZPH4WGgie9exHQnhNz4J"), + Pubkey::from_str_const("DFaHYDcuLmxx3vrJGCgw8sPehzBNjvbvZTjz7PgvYLcb"), + Pubkey::from_str_const("43Cwv866ActL11JqVQXoFPiytUYA3tFXUaNg1zPsarn9"), + Pubkey::from_str_const("3wzw3L4wd3B7DHocbQxHosQXEfqEz5AnkQHKBisGKLzY"), + Pubkey::from_str_const("EnX9MvycxMGKFpMPJHgNo6kD2proVaorKQcSaMmguMEZ"), + Pubkey::from_str_const("GLz4zxemfHJy3BnK7PjZ4uUgJoHtycpT3JrmTDYduVpg"), + Pubkey::from_str_const("362rDzQwChHJ9ToZgLdv8wzFHTTsxsxzFLTZPwKY3CYg"), + Pubkey::from_str_const("3FKwgoQWCz4i43Pbo3jdfsjzkgrkjnk44rDWh7A9J9Ee"), + Pubkey::from_str_const("Cj53e3n1WX9dAUNX5PZpsc9xG9e7hXGnuJkKLJkNaf6v"), + Pubkey::from_str_const("Amn9KQZY5kJj932b49pxwqZuuKfNWewrmNQtdMrRfdvf"), + Pubkey::from_str_const("6spvvQJzFCcw5fVVdF1zKxCeq8Gj2yRFGgFuG16fYSMo"), + Pubkey::from_str_const("H5PutDz8EgAoznaX7Bc8qSB6sPxD2Xho145H6MH6AcbV"), + Pubkey::from_str_const("HNCooAiRiVeqGK9VSjxsvRsfr4weTcx14XhkvM7zDMgx"), + Pubkey::from_str_const("AUqLkd6bR54NtXY9Zt4J2zZ7CJbhVHuDR95YFk5qahaX"), + Pubkey::from_str_const("H35YRuMGE6WDPBLu3fDYLxV7tEcL2yJ7C97jDkw6NrBf"), + Pubkey::from_str_const("EhN8bv8LCXnNAVju9ERGsCccmdBNmNTAESAHXE8PyBLR"), + Pubkey::from_str_const("Ht5roME7Fjzt3dcEyf7UxYpPPAAdyTrnJ6rgEwQrNAFg"), + Pubkey::from_str_const("5zHRch6mowPiobRQHYXZ1siRcthw4dYkFPuaNDQkK3EE"), + Pubkey::from_str_const("AZkdkkHzSEbaZXpr7xBmt71AHj9c8iWE2oWYELm3nns5"), + Pubkey::from_str_const("FThHNjNQYi7FLvP4KjeYf3LXjcN2LhyYHmza35ucqtDT"), + Pubkey::from_str_const("6vHscAGjgbTfi1ooRVWGjSiUhHcUnLBceJyuKjpKrCYy"), + Pubkey::from_str_const("AnbszMCtjEVZTquG3FQ68x1f1WK5B1Z6E7gtN2J6vD5a"), + Pubkey::from_str_const("5aUknZidNKmAPxLMHoUeWsEdmXX8sGKvj8s71pHTDC7H"), + Pubkey::from_str_const("7Xgby7EgwoRkhbQSdNJL9sXwpo3jddUhqB4zv3WuJB4m"), + Pubkey::from_str_const("43KdfhYK6LUXX4LmyxfZ1KLaQrfpRe8vNwopNS7wXsUC"), + Pubkey::from_str_const("FG6UGgT4VpCQxjwQsVy6woHeemwb5jP5u9kR5ry3twcK"), + Pubkey::from_str_const("BDvLyqobZFubd7ESEJde6KqjbNoChkdj5aEiBdCqcN5r"), + Pubkey::from_str_const("BtF4msVoQYJPrQUMwNtEXw58LVjux18ddVhC4XjbsdwX"), + Pubkey::from_str_const("GbPK6cgUKso2ZGNV6GRwbp3hkdUzTVjsybYQMGDBvMFa"), + Pubkey::from_str_const("4bWjzNek2m6P3Q8rKkq11C47TaoAwpj3ykk5yDCYLWr1"), + Pubkey::from_str_const("FPzPT7YV5BHSTjYjqycXbwBsD4E5WsrZNPcCgcHhPkUx"), + Pubkey::from_str_const("EbitXxcwjC1dzrGJ7sa3Y6RBzSM3wxvrxMaBncNBKEf2"), + Pubkey::from_str_const("HMmHHmCmC3oo4G4QCjKvNxctSBQzRFDp2WQHZoXyrW6Q"), + Pubkey::from_str_const("145FLxWrDG9gKifyJKSnkKNAUPhG6hxUMEiZWw6XeSMW"), + Pubkey::from_str_const("FnXgwJZT7RMhpYEJFCXGLHp63bhVLfVn5d4aSVQDe19n"), + Pubkey::from_str_const("5tDptzMGG79DLp9XV4fdcmiRNPQGCkfBJeo8EVCSYKw7"), + Pubkey::from_str_const("9oBNYSARUMuH2swnxzbxCi827YfUnYWyWSAkrHz93h5y"), + Pubkey::from_str_const("DWfXRxmUXKdSTji4S1nXncviTmrhWQwxHAQCcvjRLeWm"), + Pubkey::from_str_const("EHUv6Hc9XGAb57C8iNCUuoE9xnJnRTFr9LaP7zbu8LNB"), + Pubkey::from_str_const("CTqnGTKHrcg8AVK5pSwVzpA3mmSxz7Js9y7rEqCbjryH"), + Pubkey::from_str_const("7PzsZ3ApLcfQosSr2B1C9YV1GNNgriZVFyqzDpeHoAD"), + Pubkey::from_str_const("AKXpgmVhedTT49sekGbh6MXfx2kJKBcaFQLpc83NCp6N"), + Pubkey::from_str_const("3MP45RTcFdLX3HpPD18asXCCxFjsgrZM9V6tEeQxFXd4"), + Pubkey::from_str_const("88YC29WwpugANLcN7DYkuL3benW79WsdxkQ3KX8zER9s"), + Pubkey::from_str_const("98j2yMfxrLzQ5YQJU1FMW3CTkoYgZStJirpm4HLAkA3o"), + Pubkey::from_str_const("56AidDLicyQaLMqbMQeZ7itUmJyd7TAvehVDi8MXDYbx"), + Pubkey::from_str_const("3yeCpRbBU5gzBaweVRqUXFcZUp7gSXBeknKcbGmjwmjJ"), + Pubkey::from_str_const("5McU7LcsRw3RHyrjj98qHefx6ehy7MdjF1uKKi6Kp8hV"), + Pubkey::from_str_const("3CYxk8uTXtANvbU1HW3NHNhKcBEfYesi2BdBFizy3f6s"), + Pubkey::from_str_const("4bAowZuPRG8jzNwMceqjYjZnrGXQNR5ZCgQJM8s4hF4i"), + Pubkey::from_str_const("ATWFW8xBGPe6xB8JnjyjZqG5FuEXnfyuhEocmqAqfeJG"), + Pubkey::from_str_const("73Aa7yCd6AMPqsaLisfgrpjESVFtDrLnHEjfDsXBK3HV"), +]; + +pub static WITHDRAW_AUTHORITY: &[Pubkey] = &[ + Pubkey::from_str_const("8CUUMKYNGxdgYio5CLHRHyzMEhhVRMcqefgE6dLqnVRK"), + Pubkey::from_str_const("3FFaheyqtyAXZSYxDzsr5CVKvJuvZD1WE1VEsBtDbRqB"), + Pubkey::from_str_const("FdGYQdiRky8NZzN9wZtczTBcWLYYRXrJ3LMDhqDPn5rM"), + Pubkey::from_str_const("4e6KwQpyzGQPfgVr5Jn3g5jLjbXB4pKPa2jRLohEb1QA"), + Pubkey::from_str_const("FjiEiVKyMGzSLpqoB27QypukUfyWHrwzPcGNtopzZVdh"), + Pubkey::from_str_const("DwbVjia1mYeSGoJipzhaf4L5hfer2DJ1Ys681VzQm5YY"), + Pubkey::from_str_const("GeMGyvsTEsANVvcT5cme65Xq5MVU8fVVzMQ13KAZFNS2"), + Pubkey::from_str_const("Bj3aQ2oFnZYfNR1njzRjmWizzuhvfcYLckh76cqsbuBM"), + Pubkey::from_str_const("4ZJhPQAgUseCsWhKvJLTmmRRUV74fdoTpQLNfKoekbPY"), + Pubkey::from_str_const("HXdYQ5gixrY2H6Y9gqsD8kPM2JQKSaRiohDQtLbZkRWE"), + Pubkey::from_str_const("4sNBQyPbJCQyUimBueZkGWnLVqds4rWkm7eXyi9WskGU"), + Pubkey::from_str_const("AZWdNvnZxJnbcT8ZzonpN19AZJadxPdUxSiCEDTJzu8L"), + Pubkey::from_str_const("5kFTzLuM2VgFdb6x16smnY3JWoVdPxNZVFAqeVgjSTUP"), + Pubkey::from_str_const("Gbz6wkNFus8SNEkWGNNENLv9NFwVvF1pWVDpaVKUWcMh"), + Pubkey::from_str_const("AksPzoA9DKCipgdhHjhUzQJe4iEniCvBoEfvayuFA3BN"), + Pubkey::from_str_const("DVhs8YHWrvhhGxoefDNY9KotqtEEnjnSAK8MYGL2Q7X"), + Pubkey::from_str_const("2bvGnYAPSV8pa1H3vRYr5tPAXktP4DkFACHfAgqyyfhd"), + Pubkey::from_str_const("2xP8YQ3sVmfNPtGM17tZi7Lr4vsPUiN6mHLx42roazG5"), + Pubkey::from_str_const("3p9ZxnrSFTkXVrT3KnYg2tT6asnysDApEFB5DRkdeAhB"), + Pubkey::from_str_const("wRVP5MYuqP8HJ1Q8RCJ5NzUraL3DxCKPGMKSBd5iQH1"), + Pubkey::from_str_const("513qFSVgmAQFBDsnyFCM1MrVKBrWiDgb4nXrdGsqa7Z4"), + Pubkey::from_str_const("619qLS85ieR4qh7MGNLLZyLefN7hMi2FDVb5cmX1nisb"), + Pubkey::from_str_const("768NcPfBJpFBtjDAbYZLFkej6ca1W5jeArsW5MtdF8S9"), + Pubkey::from_str_const("EWAjC8a9VPbALSM3D6tGsbRfgDV48kRuHZPu8qtYSNDv"), + Pubkey::from_str_const("H8wurbnaaXsgtrjqkNH1HhncUPUhTLAmKUHkwMeyqmfN"), + Pubkey::from_str_const("HBfi37TwD4kMa1WrAWwXp3ZaFbZQ1g3XxWwNZs8QsCpY"), + Pubkey::from_str_const("6B7mXMM6BixHvDpPAPLSKweLyCXcbtprkfsw3HfMUSjZ"), + Pubkey::from_str_const("5TuV9WpmESXNfTNqasXVehoXpQy65WUHBbzgKPXPLwWx"), + Pubkey::from_str_const("FB6VmiYFnVGp1uKXA3WbNsqg9neGYVpwYYiB9q8bFRrQ"), + Pubkey::from_str_const("7Anoa4ZRiq8qaaiEnhmdpXyTEmBjZASXGfnQqVDynNie"), + Pubkey::from_str_const("6dXeE5hS8bQKeqZsc18ewCyqHimnhCBAvVJBusnRqa2F"), + Pubkey::from_str_const("Ds69ZQPb3D3aVPXdN5REyzALBrzJLdruJZ1cwyfyoEEx"), + Pubkey::from_str_const("5HA8QV7tp59iNpfjs8f84LGUGX4imaynMFSKWHUcTrMT"), + Pubkey::from_str_const("EPfiDzgbdgXdyqqwbYFMqU6Qvfx2J9Zf2J9noXBYYbbx"), + Pubkey::from_str_const("DEGyTHFXmYuyANRDYRoEcShBXLonWNdBRpNzbFZBuzhY"), + Pubkey::from_str_const("9xz1vZSWgY6TFPLZgPM2WJjDM1KiPXcALnjYFHRTkYiK"), + Pubkey::from_str_const("6n1mSmsdGFCkEyyHe4wtgEigbwhiwYRszrerMW9YRyYF"), + Pubkey::from_str_const("8rp9vcJG1Uwk25g9XfQNHysfGBYcHizVEJvavqiffB6P"), + Pubkey::from_str_const("BVsJuJbhXmxPB9XPZrYMXs1SfuZuo998u1jQ9emVp8YD"), + Pubkey::from_str_const("Fas9FrnngyVjdMZ6r8p8SsfyFDtHLMmrw1dyBxRWWVTF"), + Pubkey::from_str_const("6Lfk4e6yxXGcFNMHue5u8KQM9hWYKjbdMFZ2n7SHHd6L"), + Pubkey::from_str_const("6WuU6DWoM7ZpgUVL5TR97Fgose7vLgrv4Qvg1Yhqv8Xw"), + Pubkey::from_str_const("6mXiGJbQbX726ZhrMSmbLCF81E8rkSfo1pvBWyn1asG5"), + Pubkey::from_str_const("2UdDthRdtZKMjEx9oQYnzuLdWy3sa4XKeaqihdXkXFza"), + Pubkey::from_str_const("33XwSCq2Ft5tTCsGK7kTyMCJ4hqufDZT2REFAWXmsia8"), + Pubkey::from_str_const("B5tRZWRCTCLANkY5zLnxEXAn8S2exJD9EkH3cacEYqmE"), + Pubkey::from_str_const("CfCvwXZk1sLSqzuoFx1K7KG1puMofgwHA3tSpcArVysX"), + Pubkey::from_str_const("C4fATCN93YL5iYS3c2MrHeDmkMD6waHuBXLqsri5q5AF"), + Pubkey::from_str_const("HbJUXNHXLzpucDr9wgjuoqAwK6XNKEcqi5h29MkkuSBQ"), + Pubkey::from_str_const("G8EBx1Qo5W8731nmaBSGYyvU96onpEP5adrVSjt4vnLF"), + Pubkey::from_str_const("4dc5ty7TNod2EQfx4bPnDofPbQk5cELcn3it47qnLizL"), + Pubkey::from_str_const("4VCET97mE8ixxAPRYabM7S7dvkiAFZ8ZzEoH6kbmfYus"), + Pubkey::from_str_const("DwqBsPgtAHqQUmRN3WF5YTuLKhK7dqzrb9eRJ4UVRNLf"), + Pubkey::from_str_const("26FqcAbTLvr6cVXk2ktPu9hpeMYWukY7uUwNCtTwu7Wb"), + Pubkey::from_str_const("B22zqCsZL6PDsi4fftmcjj5k7hFjhH1BM8Dcz4fXfFXK"), + Pubkey::from_str_const("9ahLw5LFpeTuBykK3A5aiFZ1CidTY7z4YxG2kmQdSGon"), + Pubkey::from_str_const("8M731ZthMdeVoMFGRaPnUBzR7Ze6qY4zEcQ7WALEAn5i"), + Pubkey::from_str_const("A3aaVsuX7BfNCUQbuepQSaLs9KNB7c6djG6u4UuJzwwP"), + Pubkey::from_str_const("DeXE7LTUqsC3B9kSACfFdW5TPG1eT9n1Z6gVB9dJNxqz"), + Pubkey::from_str_const("BtvSeDfZNfA3VEH6baFprfx6JcS6JX8jdt6NffCxoJGV"), + Pubkey::from_str_const("8o8trB6XdG8o4qo29De2mcooY5E3pW2rvMMkeBwJHkUD"), + Pubkey::from_str_const("CSSxN8vQYqSjg5GV6ZUGSM65dXqhytLzANLpMrBZjLKP"), + Pubkey::from_str_const("5A2yoxdGdC3B3QC39pkXpjx2ZPH4WGgie9exHQnhNz4J"), + Pubkey::from_str_const("DFaHYDcuLmxx3vrJGCgw8sPehzBNjvbvZTjz7PgvYLcb"), + Pubkey::from_str_const("43Cwv866ActL11JqVQXoFPiytUYA3tFXUaNg1zPsarn9"), + Pubkey::from_str_const("3wzw3L4wd3B7DHocbQxHosQXEfqEz5AnkQHKBisGKLzY"), + Pubkey::from_str_const("EnX9MvycxMGKFpMPJHgNo6kD2proVaorKQcSaMmguMEZ"), + Pubkey::from_str_const("GLz4zxemfHJy3BnK7PjZ4uUgJoHtycpT3JrmTDYduVpg"), + Pubkey::from_str_const("362rDzQwChHJ9ToZgLdv8wzFHTTsxsxzFLTZPwKY3CYg"), + Pubkey::from_str_const("3FKwgoQWCz4i43Pbo3jdfsjzkgrkjnk44rDWh7A9J9Ee"), + Pubkey::from_str_const("Cj53e3n1WX9dAUNX5PZpsc9xG9e7hXGnuJkKLJkNaf6v"), + Pubkey::from_str_const("Amn9KQZY5kJj932b49pxwqZuuKfNWewrmNQtdMrRfdvf"), + Pubkey::from_str_const("6spvvQJzFCcw5fVVdF1zKxCeq8Gj2yRFGgFuG16fYSMo"), + Pubkey::from_str_const("H5PutDz8EgAoznaX7Bc8qSB6sPxD2Xho145H6MH6AcbV"), + Pubkey::from_str_const("HNCooAiRiVeqGK9VSjxsvRsfr4weTcx14XhkvM7zDMgx"), + Pubkey::from_str_const("AUqLkd6bR54NtXY9Zt4J2zZ7CJbhVHuDR95YFk5qahaX"), + Pubkey::from_str_const("H35YRuMGE6WDPBLu3fDYLxV7tEcL2yJ7C97jDkw6NrBf"), + Pubkey::from_str_const("EhN8bv8LCXnNAVju9ERGsCccmdBNmNTAESAHXE8PyBLR"), + Pubkey::from_str_const("Ht5roME7Fjzt3dcEyf7UxYpPPAAdyTrnJ6rgEwQrNAFg"), + Pubkey::from_str_const("5zHRch6mowPiobRQHYXZ1siRcthw4dYkFPuaNDQkK3EE"), + Pubkey::from_str_const("AZkdkkHzSEbaZXpr7xBmt71AHj9c8iWE2oWYELm3nns5"), + Pubkey::from_str_const("FThHNjNQYi7FLvP4KjeYf3LXjcN2LhyYHmza35ucqtDT"), + Pubkey::from_str_const("6vHscAGjgbTfi1ooRVWGjSiUhHcUnLBceJyuKjpKrCYy"), + Pubkey::from_str_const("AnbszMCtjEVZTquG3FQ68x1f1WK5B1Z6E7gtN2J6vD5a"), + Pubkey::from_str_const("5aUknZidNKmAPxLMHoUeWsEdmXX8sGKvj8s71pHTDC7H"), + Pubkey::from_str_const("7Xgby7EgwoRkhbQSdNJL9sXwpo3jddUhqB4zv3WuJB4m"), + Pubkey::from_str_const("43KdfhYK6LUXX4LmyxfZ1KLaQrfpRe8vNwopNS7wXsUC"), + Pubkey::from_str_const("FG6UGgT4VpCQxjwQsVy6woHeemwb5jP5u9kR5ry3twcK"), + Pubkey::from_str_const("BDvLyqobZFubd7ESEJde6KqjbNoChkdj5aEiBdCqcN5r"), + Pubkey::from_str_const("BtF4msVoQYJPrQUMwNtEXw58LVjux18ddVhC4XjbsdwX"), + Pubkey::from_str_const("GbPK6cgUKso2ZGNV6GRwbp3hkdUzTVjsybYQMGDBvMFa"), + Pubkey::from_str_const("4bWjzNek2m6P3Q8rKkq11C47TaoAwpj3ykk5yDCYLWr1"), + Pubkey::from_str_const("FPzPT7YV5BHSTjYjqycXbwBsD4E5WsrZNPcCgcHhPkUx"), + Pubkey::from_str_const("EbitXxcwjC1dzrGJ7sa3Y6RBzSM3wxvrxMaBncNBKEf2"), + Pubkey::from_str_const("HMmHHmCmC3oo4G4QCjKvNxctSBQzRFDp2WQHZoXyrW6Q"), + Pubkey::from_str_const("145FLxWrDG9gKifyJKSnkKNAUPhG6hxUMEiZWw6XeSMW"), + Pubkey::from_str_const("FnXgwJZT7RMhpYEJFCXGLHp63bhVLfVn5d4aSVQDe19n"), + Pubkey::from_str_const("5tDptzMGG79DLp9XV4fdcmiRNPQGCkfBJeo8EVCSYKw7"), + Pubkey::from_str_const("9oBNYSARUMuH2swnxzbxCi827YfUnYWyWSAkrHz93h5y"), + Pubkey::from_str_const("DWfXRxmUXKdSTji4S1nXncviTmrhWQwxHAQCcvjRLeWm"), + Pubkey::from_str_const("EHUv6Hc9XGAb57C8iNCUuoE9xnJnRTFr9LaP7zbu8LNB"), + Pubkey::from_str_const("CTqnGTKHrcg8AVK5pSwVzpA3mmSxz7Js9y7rEqCbjryH"), + Pubkey::from_str_const("7PzsZ3ApLcfQosSr2B1C9YV1GNNgriZVFyqzDpeHoAD"), + Pubkey::from_str_const("AKXpgmVhedTT49sekGbh6MXfx2kJKBcaFQLpc83NCp6N"), + Pubkey::from_str_const("3MP45RTcFdLX3HpPD18asXCCxFjsgrZM9V6tEeQxFXd4"), + Pubkey::from_str_const("88YC29WwpugANLcN7DYkuL3benW79WsdxkQ3KX8zER9s"), + Pubkey::from_str_const("98j2yMfxrLzQ5YQJU1FMW3CTkoYgZStJirpm4HLAkA3o"), + Pubkey::from_str_const("56AidDLicyQaLMqbMQeZ7itUmJyd7TAvehVDi8MXDYbx"), + Pubkey::from_str_const("3yeCpRbBU5gzBaweVRqUXFcZUp7gSXBeknKcbGmjwmjJ"), + Pubkey::from_str_const("5McU7LcsRw3RHyrjj98qHefx6ehy7MdjF1uKKi6Kp8hV"), + Pubkey::from_str_const("3CYxk8uTXtANvbU1HW3NHNhKcBEfYesi2BdBFizy3f6s"), + Pubkey::from_str_const("4bAowZuPRG8jzNwMceqjYjZnrGXQNR5ZCgQJM8s4hF4i"), + Pubkey::from_str_const("ATWFW8xBGPe6xB8JnjyjZqG5FuEXnfyuhEocmqAqfeJG"), + Pubkey::from_str_const("73Aa7yCd6AMPqsaLisfgrpjESVFtDrLnHEjfDsXBK3HV"), +]; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pinned_list_sizes() { + assert_eq!(NON_CIRCULATING_ACCOUNTS.len(), 214); + assert_eq!(WITHDRAW_AUTHORITY.len(), 114); + } +} diff --git a/crates/index/src/modules/save_block.rs b/crates/index/src/modules/save_block.rs index effe3c2..5c1c17c 100644 --- a/crates/index/src/modules/save_block.rs +++ b/crates/index/src/modules/save_block.rs @@ -10,17 +10,25 @@ use sea_orm::{ DatabaseConnection, }; use solana_pubkey::Pubkey; +use std::collections::HashMap; use tokio::{ task::{JoinHandle, JoinSet}, time::Instant, }; use yellowstone_grpc_proto::geyser::CommitmentLevel; use yellowstone_grpc_proto::geyser::SubscribeUpdateBlock; +use cloudbreak_core::modules::supply_tracker::SupplyTracker; use crate::indexer::{AccountsReceivedPerBlock, IndexerState}; use crate::modules::snapshot::SnapshotProcessingState; use crate::{db_queries, metrics, modules}; +struct PendingSupplyAccount { + owner: Option, + lamports: u64, + write_version: u64, +} + /// Splits the block into chunks and saves them into the "accounts" table /// Also updates the HashMap with the accounts pubkeys that were updated in the slot pub async fn save_block( @@ -37,6 +45,7 @@ pub async fn save_block( buffer_channel_rx_len: _, finalize_slot_buffer_size, accounts_owner_map, + supply_tracker, } = indexer_state; let start_time = Instant::now(); @@ -44,6 +53,7 @@ pub async fn save_block( let max_chunk_bytes_data = config.grpc.max_chunk_bytes_data; let slot = block.slot; + let is_repaired = block.blockhash.is_empty(); modules::snapshot::process_snapshot_if_needed( config.clone(), @@ -51,6 +61,7 @@ pub async fn save_block( &updated_accounts_during_startup, finalize_slot_buffer_size.clone(), accounts_owner_map.clone(), + supply_tracker.clone(), ) .await; @@ -77,9 +88,29 @@ pub async fn save_block( .map(|pubkey| pubkey.0.to_bytes().to_vec()) .collect::>(); + let supply_enabled = supply_tracker.is_enabled(); + let mut pending_supply_accounts: HashMap = HashMap::new(); + // Create the chunks for updating the "accounts" table let system_program_id = [0u8; 32].to_vec(); for account in block.accounts { + supply_tracker.observe_account(&account.pubkey, slot, account.lamports); + + if supply_enabled { + let pubkey = Pubkey::try_from(account.pubkey.as_slice()).unwrap(); + let pending = pending_supply_accounts + .entry(pubkey) + .or_insert_with(|| PendingSupplyAccount { + owner: accounts_owner_map.get_owner(&pubkey), + lamports: account.lamports, + write_version: account.write_version, + }); + if account.write_version > pending.write_version { + pending.lamports = account.lamports; + pending.write_version = account.write_version; + } + } + // If the account is being closed we still add it to the hashmap for cleanup // but we don't add it to the "accounts" table in a normal fashion, instead we added using [`db_queries::insert_closed_accounts`] if account.lamports == 0 { @@ -123,7 +154,13 @@ pub async fn save_block( continue; } - accounts_owner_map.upsert_account(&account.pubkey, &account.owner, slot); + let resurrects_gap_closed_account = is_repaired + && supply_tracker + .gap_close_floor(&Pubkey::try_from(account.pubkey.as_slice()).unwrap()) + .is_some_and(|closed_slot| closed_slot >= slot); + if !resurrects_gap_closed_account { + accounts_owner_map.upsert_account(&account.pubkey, &account.owner, slot); + } block_bytes_data += account.data.len(); current_chunk_bytes += account.data.len(); @@ -160,6 +197,31 @@ pub async fn save_block( chunks.push((current_chunk, current_chunk_bytes)); } + let supply_write_guard = supply_tracker.lock_block_writes().await; + if !is_repaired { + supply_tracker.record_gap_closes(slot, &closed_accounts_for_slot); + } + let defer_map_removals = supply_tracker.is_gap_filling(); + let block_supply_delta = if supply_tracker.is_tracking_deltas() { + compute_block_supply_delta( + db, + &config, + &supply_tracker, + pending_supply_accounts, + slot, + is_repaired, + ) + .await + } else { + supply_tracker.record_startup_touches( + slot, + pending_supply_accounts + .into_iter() + .map(|(pubkey, pending)| (pubkey, pending.lamports, pending.write_version)), + ); + None + }; + let closed_account_for_slot_len = closed_accounts_for_slot.len(); // We delay the closed accounts insertion until the snapshot is processed to avoid reads while @@ -170,7 +232,7 @@ pub async fn save_block( .expect("Failed to lock snapshot_processing_state") }; - let closed_accounts_insert_handle: Option> = if snapshot_processing_state + let closed_accounts_insert_handle: Option> = if snapshot_processing_state == SnapshotProcessingState::Finished || snapshot_processing_state == SnapshotProcessingState::FinishedAndCleanedUp { @@ -180,6 +242,7 @@ pub async fn save_block( slot, &config, accounts_owner_map, + defer_map_removals, ) } else { None @@ -215,18 +278,32 @@ pub async fn save_block( tasks.spawn(async move { let _guard = metrics::TokioTaskCounterGuard::new("insert_accounts_chunk"); - db_queries::insert_accounts_chunk(&db, chunk, byte_size, &config_clone).await; + db_queries::insert_accounts_chunk(&db, chunk, byte_size, &config_clone).await }); } - tasks.join_all().await; + let mut block_writes_ok = tasks.join_all().await.into_iter().all(|inserted| inserted); - if let Some(handle) = closed_accounts_insert_handle - && let Err(e) = handle.await - { - tracing::error!(target: "save_block_closed_accounts_insert", "failed to insert closed accounts: {:?}", e); + if let Some(handle) = closed_accounts_insert_handle { + match handle.await { + Ok(inserted) => block_writes_ok &= inserted, + Err(e) => { + tracing::error!(target: "save_block_closed_accounts_insert", "failed to insert closed accounts: {:?}", e); + block_writes_ok = false; + } + } + } + + if block_supply_delta.is_none() && !block_writes_ok && supply_tracker.mark_bootstrap_failed() { + tracing::error!( + target: "supply_tracker", + "account writes failed for slot {} during supply bootstrap, marking bootstrap failed", + slot + ); } + drop(supply_write_guard); + // Wait until the chunk processing is finished to insert the slot (this ensures that gPA calls can only read from completed slots) db_queries::insert_slot( slot, @@ -247,6 +324,91 @@ pub async fn save_block( ) .await; + if let Some(block_delta) = block_supply_delta { + if !block_writes_ok { + tracing::error!( + target: "supply_tracker", + "account writes failed for slot {}, marking supply stale", + slot + ); + if supply_tracker.mark_stale() { + metrics::SUPPLY_STALE.set(1); + } + } else if let Some(commit) = supply_tracker.commit_block(slot, block_delta) { + db_queries::upsert_supply_row(db, &commit, &config).await; + metrics::SUPPLY_TOTAL_LAMPORTS.set(commit.total as i64); + metrics::SUPPLY_SLOT.set(commit.slot as i64); + metrics::SUPPLY_STALE.set(0); + } + } + let elapsed = start_time.elapsed().as_secs_f64(); metrics::record_block_processing(elapsed, "block"); } + +async fn compute_block_supply_delta( + db: &DatabaseConnection, + config: &IndexConfig, + supply_tracker: &SupplyTracker, + pending_accounts: HashMap, + slot: u64, + is_repaired: bool, +) -> Option { + let mut block_delta: i128 = 0; + let mut owners = Vec::new(); + let mut pubkeys = Vec::new(); + let mut new_lamports = Vec::new(); + for (pubkey, pending) in pending_accounts { + let zero_prev = supply_tracker.take_zero_prev(&pubkey); + if is_repaired + && !zero_prev + && supply_tracker + .gap_close_floor(&pubkey) + .is_some_and(|closed_slot| closed_slot >= slot) + { + continue; + } + match pending.owner { + Some(owner) if !zero_prev => { + owners.push(owner.to_bytes().to_vec()); + pubkeys.push(pubkey.to_bytes().to_vec()); + new_lamports.push(pending.lamports); + } + _ => block_delta += pending.lamports as i128, + } + } + + if pubkeys.is_empty() { + return Some(block_delta); + } + + match db_queries::fetch_block_supply_delta(db, owners, pubkeys, new_lamports, slot, config) + .await + { + Ok(result) => { + if result.routed_misses > 0 { + metrics::SUPPLY_ROUTED_MISSES.inc_by(result.routed_misses); + tracing::warn!( + target: "supply_tracker", + "owner-routed prev-read found no previous row for {} accounts in slot {}", + result.routed_misses, + slot + ); + } + Some(block_delta + result.block_delta) + } + Err(e) => { + metrics::SUPPLY_QUERY_ERRORS.inc(); + tracing::error!( + target: "supply_tracker", + "block supply delta query failed for slot {}, marking supply stale: {}", + slot, + e + ); + if supply_tracker.mark_stale() { + metrics::SUPPLY_STALE.set(1); + } + None + } + } +} diff --git a/crates/index/src/modules/self_healing.rs b/crates/index/src/modules/self_healing.rs index eeb378a..d706c4c 100644 --- a/crates/index/src/modules/self_healing.rs +++ b/crates/index/src/modules/self_healing.rs @@ -3,7 +3,7 @@ * Copyright 2025-2026 Triton One Limited. All rights reserved. */ -use cloudbreak_core::{IndexConfig, SnapshotConfig, SnapshotConfigOnIndexer}; +use cloudbreak_core::{IndexConfig, SnapshotConfig, SnapshotConfigOnIndexer, modules::supply_tracker::SupplyTracker}; use cloudbreak_snapshot::sidecar::{SnapshotPair, SnapshotType}; use sea_orm::DatabaseConnection; use std::{ @@ -38,15 +38,21 @@ pub struct SelfHealingState { /// [`SlotFinalizer::enqueue_gap_boundary`]). pub gap_boundaries: Arc>>, pub finalizer: SlotFinalizer, + pub supply_tracker: SupplyTracker, } impl SelfHealingState { - pub fn new(_config: &IndexConfig, finalizer: SlotFinalizer) -> Self { + pub fn new( + _config: &IndexConfig, + finalizer: SlotFinalizer, + supply_tracker: SupplyTracker, + ) -> Self { Self { last_slot_received: Arc::new(Mutex::new(0)), gaps_list: Arc::new(Mutex::new(Vec::new())), gap_boundaries: Arc::new(Mutex::new(BTreeSet::new())), finalizer, + supply_tracker, } } @@ -130,6 +136,10 @@ impl SelfHealingState { // startup pauses the very worker that completes startup; that is intentionally // unsupported and `fill_gaps` fails fast in that case. self.finalizer.pause().await; + + if self.supply_tracker.mark_gap() { + metrics::SUPPLY_STALE.set(1); + } } } @@ -407,6 +417,7 @@ impl SelfHealingState { .expect("Failed to lock gaps_list") .is_empty(); if !gaps_remaining { + self.supply_tracker.finish_gap(); self.finalizer.resume().await; } } diff --git a/crates/index/src/modules/snapshot.rs b/crates/index/src/modules/snapshot.rs index 0e72f89..e919a12 100644 --- a/crates/index/src/modules/snapshot.rs +++ b/crates/index/src/modules/snapshot.rs @@ -5,6 +5,7 @@ use cloudbreak_core::{IndexConfig, SnapshotConfig, modules::account_owner_map::AccountOwnerMap}; use std::sync::{Arc, Mutex}; +use cloudbreak_core::modules::supply_tracker::SupplyTracker; use crate::metrics; use crate::modules::finalize_slot::UpdatedAccountsDuringStartup; @@ -33,6 +34,7 @@ pub async fn process_snapshot_if_needed( updated_accounts_during_startup: &UpdatedAccountsDuringStartup, finalize_slot_buffer_size: Arc>, accounts_owner_map: AccountOwnerMap, + supply_tracker: SupplyTracker, ) { let snapshot_config = match config.snapshot { Some(snapshot_config) => snapshot_config, @@ -79,6 +81,7 @@ pub async fn process_snapshot_if_needed( Some(metrics::METRICS_REGISTRY.clone()), Some(finalize_slot_buffer_size.clone()), accounts_owner_map, + supply_tracker, ) .await; diff --git a/crates/integration_tests/README.md b/crates/integration_tests/README.md index ef3edce..bee363a 100644 --- a/crates/integration_tests/README.md +++ b/crates/integration_tests/README.md @@ -57,6 +57,7 @@ cargo run --bin integration_tests -- benchmark get-account-info cargo run --bin integration_tests -- benchmark get-multiple-accounts cargo run --bin integration_tests -- benchmark get-balance cargo run --bin integration_tests -- benchmark get-token-account-balance +cargo run --bin integration_tests -- benchmark simulate-transaction cargo run --bin integration_tests -- benchmark -c custom.toml gpa ``` @@ -64,7 +65,7 @@ cargo run --bin integration_tests -- benchmark -c custom.toml gpa | Argument | Description | | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `` | Required. One of: `gpa`, `gtabo`, `gtabd`, `gpa-token-owner`, `gpa-token-mint`, `get-account-info`, `get-multiple-accounts`, `get-balance`, `get-token-account-balance` | +| `` | Required. One of: `gpa`, `gtabo`, `gtabd`, `gpa-token-owner`, `gpa-token-mint`, `get-account-info`, `get-multiple-accounts`, `get-balance`, `get-token-account-balance`, `simulate-transaction` | | `-c, --config ` | Path to TOML config file (default: `cloudbreak.integration_tests.toml`) | **Request types:** diff --git a/crates/integration_tests/src/benchmark.rs b/crates/integration_tests/src/benchmark.rs index 3da844a..759b280 100644 --- a/crates/integration_tests/src/benchmark.rs +++ b/crates/integration_tests/src/benchmark.rs @@ -32,7 +32,10 @@ pub enum RequestType { GetMultipleAccounts, GetBalance, GetTokenAccountBalance, + GetTokenSupply, + GetTokenLargestAccounts, SimulateTransaction, + GetSupply, } pub async fn run(args: &BenchmarkArgs) -> Result<()> { diff --git a/crates/integration_tests/src/response_comparison.rs b/crates/integration_tests/src/response_comparison.rs index b06a463..80ff4d1 100644 --- a/crates/integration_tests/src/response_comparison.rs +++ b/crates/integration_tests/src/response_comparison.rs @@ -6,7 +6,7 @@ use anyhow::Result; use base64::Engine as _; use serde_json::Value as JsonValue; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::time::{Duration, SystemTime}; use crate::benchmark::RequestType; @@ -145,6 +145,12 @@ fn accounts_equal(account1: &JsonValue, account2: &JsonValue, encoding: &str) -> /// - `GetBalance` — `result.value` is a `u64`, compared directly. /// - `GetTokenAccountBalance` — `result.value` is a `UiTokenAmount` object, /// compared directly. +/// - `GetTokenSupply` — `result.value` is a `UiTokenAmount` object, +/// compared directly. +/// - `GetTokenLargestAccounts` — `result.value` is an array of +/// `RpcTokenAccountBalance` objects ordered deterministically by both +/// implementations (amount descending, pubkey-descending tie-break), +/// compared directly. /// /// For zstd-compressed encodings, account data is decompressed before /// comparison to handle non-deterministic compression across implementations. @@ -178,17 +184,22 @@ pub fn compare_responses( compare_multiple_accounts_responses(response1, response2, encoding) } RequestType::GetAccountInfo => compare_value_with_account(response1, response2, encoding), - RequestType::GetBalance | RequestType::GetTokenAccountBalance => { - compare_value_direct(response1, response2) - } + RequestType::GetBalance + | RequestType::GetTokenAccountBalance + | RequestType::GetTokenSupply + | RequestType::GetTokenLargestAccounts => compare_value_direct(response1, response2), RequestType::SimulateTransaction => compare_simulate_responses(response1, response2), + RequestType::GetSupply => compare_supply_responses(response1, response2), }; CompareResponsesResult::new_with_matching_context(matches, compare_context_result) } -/// Compares two `result.value: u64 | object` responses by direct JSON equality. -/// Used for `getBalance` (u64) and `getTokenAccountBalance` (UiTokenAmount). +/// Compares two `result.value: u64 | object | array` responses by direct JSON +/// equality. Used for `getBalance` (u64), `getTokenAccountBalance` / +/// `getTokenSupply` (UiTokenAmount), and `getTokenLargestAccounts` (array of +/// `RpcTokenAccountBalance`, deterministically ordered by amount descending +/// with a pubkey-descending tie-break in both implementations). fn compare_value_direct(response1: &JsonValue, response2: &JsonValue) -> bool { let v1 = response1.get("result").and_then(|r| r.get("value")); let v2 = response2.get("result").and_then(|r| r.get("value")); @@ -197,6 +208,35 @@ fn compare_value_direct(response1: &JsonValue, response2: &JsonValue) -> bool { static NULL_VALUE: JsonValue = JsonValue::Null; +fn compare_supply_responses(response1: &JsonValue, response2: &JsonValue) -> bool { + let v1 = response1.get("result").and_then(|r| r.get("value")); + let v2 = response2.get("result").and_then(|r| r.get("value")); + + let (v1, v2) = match (v1, v2) { + (Some(v1), Some(v2)) => (v1, v2), + _ => return v1 == v2, + }; + + if ["total", "circulating", "nonCirculating"] + .iter() + .any(|key| field(v1, key) != field(v2, key)) + { + return false; + } + + let accounts_set = |value: &JsonValue| -> Option> { + field(value, "nonCirculatingAccounts") + .as_array() + .map(|array| { + array + .iter() + .filter_map(|a| a.as_str().map(String::from)) + .collect() + }) + }; + accounts_set(v1) == accounts_set(v2) +} + fn field<'a>(value: &'a JsonValue, key: &str) -> &'a JsonValue { value.get(key).unwrap_or(&NULL_VALUE) } diff --git a/crates/integration_tests/src/sources/victoria_logs.rs b/crates/integration_tests/src/sources/victoria_logs.rs index 5af2bb4..23eaca6 100644 --- a/crates/integration_tests/src/sources/victoria_logs.rs +++ b/crates/integration_tests/src/sources/victoria_logs.rs @@ -77,9 +77,18 @@ pub fn get_body_query( RequestType::GetTokenAccountBalance => format!( "query={time_filter}rpc_call:=\"getTokenAccountBalance\"{pool_filter} | limit {limit}" ), + RequestType::GetTokenSupply => format!( + "query={time_filter}rpc_call:=\"getTokenSupply\" AND pool_dedicated:~\"liquid\" | limit {limit}" + ), + RequestType::GetTokenLargestAccounts => format!( + "query={time_filter}rpc_call:=\"getTokenLargestAccounts\" AND pool_dedicated:~\"liquid\" | limit {limit}" + ), RequestType::SimulateTransaction => format!( "query={time_filter}rpc_call:=\"simulateTransaction\"{simulate_pool_filter} AND body:* | limit {limit}" ), + RequestType::GetSupply => format!( + "query={time_filter}rpc_call:=\"getSupply\" AND body:* | limit {limit}" + ), } } @@ -185,9 +194,10 @@ pub async fn get_requests( // positives when "encoding" appears in a memcmp filter rather than as // the response encoding parameter. Verify using the actual parsed field. // - // Methods without an encoding field (`getBalance`, `getTokenAccountBalance`) - // report `"none"` from `extract_encoding_from_request` — for those the - // encoding filter is treated as a no-op (the request passes through). + // Methods without an encoding field (`getBalance`, `getTokenAccountBalance`, + // `getTokenSupply`, `getTokenLargestAccounts`) report `"none"` from + // `extract_encoding_from_request` — for those the encoding filter is + // treated as a no-op (the request passes through). if let Some(target_encoding) = encoding { let actual_encoding = utils::extract_encoding_from_request(&body, request_type); if actual_encoding != "none" { diff --git a/crates/integration_tests/src/utils.rs b/crates/integration_tests/src/utils.rs index a899cbc..f270c52 100644 --- a/crates/integration_tests/src/utils.rs +++ b/crates/integration_tests/src/utils.rs @@ -111,8 +111,11 @@ pub fn extract_commitment_from_request(request: &JsonValue, request_type: Reques | RequestType::GetMultipleAccounts | RequestType::GetBalance | RequestType::GetTokenAccountBalance + | RequestType::GetTokenSupply + | RequestType::GetTokenLargestAccounts | RequestType::SimulateTransaction => 1, RequestType::Gtabo | RequestType::Gtabd => 2, + RequestType::GetSupply => 0, }; request .get("params") @@ -131,7 +134,8 @@ pub fn extract_commitment_from_request(request: &JsonValue, request_type: Reques /// `getMultipleAccounts`): the request's `encoding` if present, otherwise the /// per-method Agave default. /// - For methods that don't carry an encoding at all (`getBalance`, -/// `getTokenAccountBalance`): the sentinel string `"none"`. +/// `getTokenAccountBalance`, `getTokenSupply`, `getTokenLargestAccounts`): +/// the sentinel string `"none"`. /// /// Agave defaults followed here: /// - `getProgramAccounts` → `base58` @@ -139,10 +143,15 @@ pub fn extract_commitment_from_request(request: &JsonValue, request_type: Reques /// - `getAccountInfo` → `binary` (deprecated base58 plain-string) /// - `getMultipleAccounts` → `base64` pub fn extract_encoding_from_request(request: &JsonValue, request_type: RequestType) -> String { - // getBalance / getTokenAccountBalance have no encoding concept. + // getBalance / getTokenAccountBalance / getTokenSupply / getTokenLargestAccounts + // have no encoding concept. if matches!( request_type, - RequestType::GetBalance | RequestType::GetTokenAccountBalance + RequestType::GetBalance + | RequestType::GetTokenAccountBalance + | RequestType::GetTokenSupply + | RequestType::GetTokenLargestAccounts + | RequestType::GetSupply ) { return "none".to_string(); } @@ -155,7 +164,13 @@ pub fn extract_encoding_from_request(request: &JsonValue, request_type: RequestT | RequestType::GetMultipleAccounts | RequestType::SimulateTransaction => 1, RequestType::Gtabo | RequestType::Gtabd => 2, - RequestType::GetBalance | RequestType::GetTokenAccountBalance => unreachable!(), + RequestType::GetBalance + | RequestType::GetTokenAccountBalance + | RequestType::GetTokenSupply + | RequestType::GetTokenLargestAccounts + | RequestType::GetSupply => { + unreachable!() + } }; let encoding = request .get("params") @@ -174,7 +189,11 @@ pub fn extract_encoding_from_request(request: &JsonValue, request_type: RequestT RequestType::GetAccountInfo => "binary".to_string(), RequestType::GetMultipleAccounts => "base64".to_string(), RequestType::SimulateTransaction => "base58".to_string(), - RequestType::GetBalance | RequestType::GetTokenAccountBalance => unreachable!(), + RequestType::GetBalance + | RequestType::GetTokenAccountBalance + | RequestType::GetTokenSupply + | RequestType::GetTokenLargestAccounts + | RequestType::GetSupply => unreachable!(), }, } } diff --git a/crates/migration/src/lib.rs b/crates/migration/src/lib.rs index 0a119a7..37dc785 100644 --- a/crates/migration/src/lib.rs +++ b/crates/migration/src/lib.rs @@ -21,6 +21,7 @@ mod m20260528_000000_create_epoch_stakes_table; mod m20260703_000000_create_recent_blockhashes_table; mod m20260709_000000_add_block_height_to_recent_blockhashes; mod m20260711_000000_create_index_patterns_table; +mod m20260717_000000_create_supply_tables; pub struct Migrator; @@ -42,6 +43,7 @@ impl MigratorTrait for Migrator { Box::new(m20260703_000000_create_recent_blockhashes_table::Migration), Box::new(m20260709_000000_add_block_height_to_recent_blockhashes::Migration), Box::new(m20260711_000000_create_index_patterns_table::Migration), + Box::new(m20260717_000000_create_supply_tables::Migration), ] } } diff --git a/crates/migration/src/m20260717_000000_create_supply_tables.rs b/crates/migration/src/m20260717_000000_create_supply_tables.rs new file mode 100644 index 0000000..bb81e07 --- /dev/null +++ b/crates/migration/src/m20260717_000000_create_supply_tables.rs @@ -0,0 +1,43 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let connection = manager.get_connection(); + + connection + .execute_unprepared( + r#" + CREATE TABLE IF NOT EXISTS supply ( + slot BIGINT PRIMARY KEY, + total NUMERIC(20, 0) NOT NULL, + non_circulating_lamports NUMERIC(20, 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE TABLE IF NOT EXISTS non_circulating_accounts ( + id BIGINT PRIMARY KEY DEFAULT 1, + slot BIGINT NOT NULL, + accounts BYTEA[] NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + "#, + ) + .await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + "DROP TABLE IF EXISTS supply; DROP TABLE IF EXISTS non_circulating_accounts;", + ) + .await?; + + Ok(()) + } +} diff --git a/crates/snapshot/src/db_queries.rs b/crates/snapshot/src/db_queries.rs index ab53049..a1add70 100644 --- a/crates/snapshot/src/db_queries.rs +++ b/crates/snapshot/src/db_queries.rs @@ -14,10 +14,12 @@ use cloudbreak_entity::snapshot_accounts::{self}; use rust_decimal::Decimal; use sea_orm::{ ActiveValue::Set, ConnectionTrait, DatabaseConnection, EntityTrait, Statement, - TransactionTrait, Value, + TransactionTrait, Value, sea_query::ArrayType, }; -use tokio::time::Instant; +use solana_pubkey::Pubkey; +use tokio::time::{Instant, timeout}; use yellowstone_grpc_proto::geyser::SubscribeUpdateAccount; +use cloudbreak_core::modules::supply_tracker::{SUPPLY_RING_SLOTS, SupplyCommit}; use crate::metrics; use crate::stake_data::SnapshotStakeData; @@ -553,3 +555,195 @@ pub async fn persist_epoch_stakes( Ok(()) } + +pub async fn persist_supply_seed( + db: &DatabaseConnection, + bank_info: &crate::sidecar::SnapshotBankInfo, +) -> Result<(), anyhow::Error> { + db.execute(Statement::from_sql_and_values( + sea_orm::DatabaseBackend::Postgres, + "WITH upsert AS ( \ + INSERT INTO supply (slot, total) VALUES ($1, $2) \ + ON CONFLICT (slot) DO UPDATE SET total = EXCLUDED.total, updated_at = now() \ + ) \ + DELETE FROM supply WHERE slot < $3", + [ + Value::from(bank_info.slot as i64), + Value::from(Decimal::from(bank_info.capitalization)), + Value::from(bank_info.slot.saturating_sub(SUPPLY_RING_SLOTS) as i64), + ], + )) + .await?; + + tracing::info!( + target: "persist_supply_seed", + "seeded supply from snapshot: slot {} capitalization {}", + bank_info.slot, + bank_info.capitalization + ); + + Ok(()) +} + +const STARTUP_BALANCES_QUERY_TIMEOUT: Duration = Duration::from_secs(60); + +pub fn bytea_array(items: Vec>) -> Value { + Value::Array( + ArrayType::Bytes, + Some(Box::new( + items + .into_iter() + .map(|bytes| Value::Bytes(Some(Box::new(bytes)))) + .collect(), + )), + ) +} + +pub fn pubkey_bytea_array(pubkeys: &[Pubkey]) -> Value { + bytea_array( + pubkeys + .iter() + .map(|pubkey| pubkey.to_bytes().to_vec()) + .collect(), + ) +} + +pub fn parse_pubkey(bytes: Vec) -> Result { + Pubkey::try_from(bytes.as_slice()) + .map_err(|_| sea_orm::DbErr::Custom("invalid pubkey bytes in query result".to_string())) +} + +async fn query_all_with_timeout( + name: &str, + query: impl Future, sea_orm::DbErr>>, +) -> Result, sea_orm::DbErr> { + timeout(STARTUP_BALANCES_QUERY_TIMEOUT, query) + .await + .map_err(|elapsed| sea_orm::DbErr::Custom(format!("{name} timeout: {elapsed}")))? +} + +pub fn owner_pubkey_arrays(pairs: &[(Pubkey, Pubkey)]) -> (Value, Value) { + let owners = pairs + .iter() + .map(|(owner, _)| owner.to_bytes().to_vec()) + .collect(); + let pubkeys = pairs + .iter() + .map(|(_, pubkey)| pubkey.to_bytes().to_vec()) + .collect(); + (bytea_array(owners), bytea_array(pubkeys)) +} + +pub struct StartupBalanceProbe { + pub resolved: Vec<(Pubkey, u64)>, + pub misses: Vec, +} + +pub async fn fetch_startup_balances_by_owner( + db: &DatabaseConnection, + entries: &[(Pubkey, Pubkey)], + startup_slot: u64, +) -> Result { + let mut probe = StartupBalanceProbe { + resolved: Vec::new(), + misses: Vec::new(), + }; + if entries.is_empty() { + return Ok(probe); + } + + let (owners, pubkeys) = owner_pubkey_arrays(entries); + + let query = db.query_all(Statement::from_sql_and_values( + sea_orm::DatabaseBackend::Postgres, + r#" + SELECT v.pubkey, prev.lamports + FROM unnest($1::bytea[], $2::bytea[]) AS v(owner, pubkey) + LEFT JOIN LATERAL ( + SELECT lamports FROM snapshot_accounts + WHERE owner = v.owner AND pubkey = v.pubkey AND slot <= $3 + ORDER BY slot DESC + LIMIT 1 + ) prev ON true + "#, + [owners, pubkeys, Value::BigInt(Some(startup_slot as i64))], + )); + + let rows = query_all_with_timeout("fetch_startup_balances_by_owner", query).await?; + + for row in rows { + let pubkey = parse_pubkey(row.try_get("", "pubkey")?)?; + let lamports: Option = row.try_get("", "lamports")?; + match lamports { + Some(lamports) => probe.resolved.push((pubkey, lamports as u64)), + None => probe.misses.push(pubkey), + } + } + + Ok(probe) +} + +pub async fn fetch_startup_balances_from_versions( + db: &DatabaseConnection, + pubkeys: &[Pubkey], + startup_slot: u64, +) -> Result, sea_orm::DbErr> { + if pubkeys.is_empty() { + return Ok(Vec::new()); + } + + let query = db.query_all(Statement::from_sql_and_values( + sea_orm::DatabaseBackend::Postgres, + r#" + SELECT v.pubkey, COALESCE(prev.lamports, 0) AS lamports + FROM unnest($1::bytea[]) AS v(pubkey) + LEFT JOIN LATERAL ( + SELECT owner FROM temp_snapshot_account_versions + WHERE pubkey = v.pubkey AND slot <= $2 + ORDER BY slot DESC + LIMIT 1 + ) version ON true + LEFT JOIN LATERAL ( + SELECT lamports FROM snapshot_accounts + WHERE owner = version.owner AND pubkey = v.pubkey AND slot <= $2 + ORDER BY slot DESC + LIMIT 1 + ) prev ON true + "#, + [ + pubkey_bytea_array(pubkeys), + Value::BigInt(Some(startup_slot as i64)), + ], + )); + + let rows = query_all_with_timeout("fetch_startup_balances_from_versions", query).await?; + + rows.into_iter() + .map(|row| { + let pubkey = parse_pubkey(row.try_get("", "pubkey")?)?; + let lamports: i64 = row.try_get("", "lamports")?; + Ok((pubkey, lamports as u64)) + }) + .collect() +} + +pub async fn persist_supply_total( + db: &DatabaseConnection, + commit: &SupplyCommit, +) -> Result<(), anyhow::Error> { + db.execute(Statement::from_sql_and_values( + sea_orm::DatabaseBackend::Postgres, + "INSERT INTO supply (slot, total, non_circulating_lamports) VALUES ($1, $2, $3) \ + ON CONFLICT (slot) DO UPDATE SET \ + total = EXCLUDED.total, \ + non_circulating_lamports = EXCLUDED.non_circulating_lamports, \ + updated_at = now()", + [ + Value::from(commit.slot as i64), + Value::from(Decimal::from(commit.total)), + Value::from(commit.non_circulating.map(Decimal::from)), + ], + )) + .await?; + Ok(()) +} diff --git a/crates/snapshot/src/lib.rs b/crates/snapshot/src/lib.rs index 32b3c3d..508ffa8 100644 --- a/crates/snapshot/src/lib.rs +++ b/crates/snapshot/src/lib.rs @@ -6,7 +6,9 @@ use sea_orm::{ConnectOptions, Database, DatabaseConnection}; use agave_fs::FileInfo; use solana_accounts_db::accounts_file::AccountsFile; +use solana_pubkey::Pubkey; use std::{ + collections::HashMap, path::PathBuf, sync::{Arc, Mutex}, }; @@ -16,7 +18,8 @@ use yellowstone_grpc_proto::geyser::{ SubscribeUpdateAccount, SubscribeUpdateAccountInfo, SubscribeUpdateBlock, }; use cloudbreak_core::{ - Result, SnapshotConfig, modules::account_owner_map::AccountOwnerMap, + Result, SnapshotConfig, + modules::{account_owner_map::AccountOwnerMap, supply_tracker::SupplyTracker}, }; use crate::{ @@ -31,7 +34,9 @@ pub mod metrics; pub mod sidecar; pub mod stake_data; -pub use db_queries::persist_epoch_stakes; +pub use db_queries::{ + bytea_array, owner_pubkey_arrays, parse_pubkey, persist_epoch_stakes, pubkey_bytea_array, +}; const DB_ACCOUNTS_BATCH_SIZE: usize = 200; @@ -46,6 +51,7 @@ pub async fn run( metrics_registry: Option, buffer_size: Option>>, accounts_owner_map: AccountOwnerMap, + supply_tracker: SupplyTracker, ) -> Result<()> { let start_time = Instant::now(); @@ -73,6 +79,7 @@ pub async fn run( &database, config.clone(), accounts_owner_map.clone(), + supply_tracker.clone(), ); // Process incremental snapshot only if needed @@ -84,6 +91,7 @@ pub async fn run( &database, config.clone(), accounts_owner_map.clone(), + supply_tracker.clone(), ) .await??; @@ -107,6 +115,8 @@ pub async fn run( db_queries::clean_up_closed_accounts(&database).await?; db_queries::create_database_indexes(&database, &config.pg_indexes).await?; + finish_supply_bootstrap(&database, &accounts_owner_map, &supply_tracker).await; + tracing::info!( "Total snapshot processing time after cleanup: {} secs", start_time.elapsed().as_secs_f64() @@ -115,6 +125,119 @@ pub async fn run( Ok(()) } +const STARTUP_BALANCES_CHUNK_SIZE: usize = 5_000; + +async fn finish_supply_bootstrap( + database: &DatabaseConnection, + accounts_owner_map: &AccountOwnerMap, + supply_tracker: &SupplyTracker, +) { + let Some(startup_slot) = supply_tracker.startup_slot() else { + return; + }; + + if supply_tracker.bootstrap_failed() { + tracing::error!( + "Supply bootstrap poisoned by failed startup account writes, supply stays bootstrapping" + ); + return; + } + + let start_time = Instant::now(); + + let touched = supply_tracker.startup_touched_pubkeys(); + let touched_count = touched.len(); + let Ok(mut balances) = + resolve_startup_balances(database, accounts_owner_map, touched, startup_slot) + .await + .map_err(log_startup_balances_error) + else { + return; + }; + + let _write_guard = supply_tracker.lock_block_writes().await; + + let late_touches: Vec = supply_tracker + .startup_touched_pubkeys() + .into_iter() + .filter(|pubkey| !balances.contains_key(pubkey)) + .collect(); + let Ok(late_balances) = + resolve_startup_balances(database, accounts_owner_map, late_touches, startup_slot) + .await + .map_err(log_startup_balances_error) + else { + return; + }; + balances.extend(late_balances); + + let Some(commit) = supply_tracker.finish_bootstrap(&balances) else { + if supply_tracker.bootstrap_failed() { + tracing::error!( + "Supply bootstrap poisoned by failed startup account writes, supply stays bootstrapping" + ); + } else { + tracing::error!( + "Supply bootstrap reconciliation left unresolved accounts, supply stays bootstrapping" + ); + } + return; + }; + + tracing::info!( + target: "supply_bootstrap", + "Supply bootstrap reconciled {} touched accounts against startup slot {} in {} secs - slot: {}, total: {}", + touched_count, + startup_slot, + start_time.elapsed().as_secs_f64(), + commit.slot, + commit.total + ); + + if let Err(e) = db_queries::persist_supply_total(database, &commit).await { + tracing::error!("Failed to persist bootstrapped supply total: {:?}", e); + } +} + +fn log_startup_balances_error(e: impl std::fmt::Debug) { + tracing::error!( + "Failed to resolve startup balances for supply bootstrap, supply stays bootstrapping: {:?}", + e + ); +} + +async fn resolve_startup_balances( + database: &DatabaseConnection, + accounts_owner_map: &AccountOwnerMap, + pubkeys: Vec, + startup_slot: u64, +) -> Result> { + let mut routed = Vec::new(); + let mut unrouted = Vec::new(); + for pubkey in pubkeys { + match accounts_owner_map.get_owner(&pubkey) { + Some(owner) => routed.push((owner, pubkey)), + None => unrouted.push(pubkey), + } + } + + let mut balances = HashMap::new(); + for chunk in routed.chunks(STARTUP_BALANCES_CHUNK_SIZE) { + let probe = + db_queries::fetch_startup_balances_by_owner(database, chunk, startup_slot).await?; + balances.extend(probe.resolved); + unrouted.extend(probe.misses); + } + + for chunk in unrouted.chunks(STARTUP_BALANCES_CHUNK_SIZE) { + let resolved = + db_queries::fetch_startup_balances_from_versions(database, chunk, startup_slot).await?; + balances.extend(resolved); + } + + Ok(balances) +} + fn download_and_process_snapshot( sidecar_endpoint: String, snapshot_data: SnapshotData, @@ -122,6 +245,7 @@ fn download_and_process_snapshot( database: &DatabaseConnection, config: SnapshotConfig, accounts_owner_map: AccountOwnerMap, + supply_tracker: SupplyTracker, ) -> JoinHandle> { let db_clone = database.clone(); @@ -138,7 +262,14 @@ fn download_and_process_snapshot( tracing::error!("Failed to download snapshot: {:?} ({:?})", e, snapshot_type); })?; - process_downloaded_snapshot(&db_clone, snapshot_data, config, accounts_owner_map).await?; + process_downloaded_snapshot( + &db_clone, + snapshot_data, + config, + accounts_owner_map, + supply_tracker, + ) + .await?; Ok(()) }) @@ -150,6 +281,7 @@ async fn process_downloaded_snapshot( snapshot_data: SnapshotData, config: SnapshotConfig, accounts_owner_map: AccountOwnerMap, + supply_tracker: SupplyTracker, ) -> Result<()> { let start_time = Instant::now(); let total_accounts_files_opening_time_micros = Arc::new(Mutex::new(0)); @@ -159,12 +291,20 @@ async fn process_downloaded_snapshot( let sidecar::UnpackedSnapshot { account_files: solana_snapshot, stake_data, + bank_info, } = sidecar::unpack_compressed_snapshot(path, &base_dir, snapshot_data.slot)?; if let Err(e) = db_queries::persist_epoch_stakes(database, &stake_data).await { tracing::error!("Failed to persist epoch stakes from snapshot: {:?}", e); } + if supply_tracker.is_enabled() { + supply_tracker.set_startup_total(bank_info.slot, bank_info.capitalization); + if let Err(e) = db_queries::persist_supply_seed(database, &bank_info).await { + tracing::error!("Failed to persist supply seed from snapshot: {:?}", e); + } + } + let mut account_file_workers: JoinSet> = JoinSet::new(); let accounts_file_concurency = config.accounts_file_concurency.unwrap_or(32); let programs_include = config @@ -432,7 +572,7 @@ pub async fn process_downloaded_snapshot_with_gap_filling( let path = base_dir.join(&incremental_snapshot_file_name); let sidecar::UnpackedSnapshot { account_files: solana_snapshot, - stake_data: _, + .. } = sidecar::unpack_compressed_snapshot(path, &base_dir, snapshot_slot)?; let mut account_file_workers: JoinSet> = JoinSet::new(); let accounts_file_concurency = config.accounts_file_concurency.unwrap_or(32); diff --git a/crates/snapshot/src/sidecar.rs b/crates/snapshot/src/sidecar.rs index 9feb0d2..368098d 100644 --- a/crates/snapshot/src/sidecar.rs +++ b/crates/snapshot/src/sidecar.rs @@ -374,9 +374,16 @@ pub async fn download_snapshot_file( Ok(()) } +#[derive(Debug, Clone, Copy)] +pub struct SnapshotBankInfo { + pub capitalization: u64, + pub slot: u64, +} + pub struct UnpackedSnapshot { pub account_files: Vec, pub stake_data: crate::stake_data::SnapshotStakeData, + pub bank_info: SnapshotBankInfo, } pub fn unpack_compressed_snapshot>( @@ -436,6 +443,10 @@ pub fn unpack_compressed_snapshot>( let stake_data = crate::stake_data::extract_stake_data(&bank_fields, &extra_fields.versioned_epoch_stakes); + let bank_info = SnapshotBankInfo { + capitalization: bank_fields.capitalization, + slot: bank_fields.slot, + }; tracing::info!( target: "unpack_compressed_snapshot", "Extracted stake data: epoch={}, voters={}, in_epoch_set={}", @@ -522,6 +533,7 @@ pub fn unpack_compressed_snapshot>( Ok(UnpackedSnapshot { account_files: account_file_data, stake_data, + bank_info, }) } diff --git a/example.cloudbreak.index.toml b/example.cloudbreak.index.toml index 50b149d..bbd2d6c 100644 --- a/example.cloudbreak.index.toml +++ b/example.cloudbreak.index.toml @@ -1,6 +1,7 @@ # The buffer size for queuing finalize slot events finalize-slot-buffer-size = 1000 accounts-owner-map-enabled = false +supply-tracker-enabled = false [snapshot] accounts-file-concurency = 32 diff --git a/example.cloudbreak.integration_tests.toml b/example.cloudbreak.integration_tests.toml index 5ecd5aa..8c59305 100644 --- a/example.cloudbreak.integration_tests.toml +++ b/example.cloudbreak.integration_tests.toml @@ -11,6 +11,7 @@ target_rps = 5.0 max_in_flight = 100 duration_secs = 100 timeout_secs = 60 +# start_on_first_request = true # Start the countdown when the first request is caught (use for simulateTransaction) # Optional bandwidth cap in Gbit/s, enforced against actual received rpc1 bytes. # Works together with target_rps: if throughput hits this limit below target_rps, # effective RPS is throttled so the average stays under the cap. Omit to disable. @@ -33,6 +34,9 @@ path = "crates/integration_tests/gpa_benchmark_requests.json" # encoding = "jsonParsed" # pool_dedicated = "liquid" # Value for the `pool_dedicated` VictoriaLogs filter; omit for no constraint # inject_context = true # On no-context mismatch, re-sends with withContext:true and slot compensation before deciding +# window_seconds = 30 # Trailing window in seconds; takes precedence over `minutes` (use for simulateTransaction: keeps replayed blockhashes fresh) +# poll_interval_secs = 5 # How often the background fetcher re-queries VictoriaLogs +# replay_once = true # Send each req_id exactly once instead of cycling the pool (recommended for simulateTransaction) # Example of mismatch_dir source (re-run previously mismatched requests) # [source]