diff --git a/README.md b/README.md index 7daacde..b4d85af 100644 --- a/README.md +++ b/README.md @@ -332,6 +332,7 @@ If you want to run a tracker outside Compose, point `endpoint` at it. | ---------------------------- | ------- | ------- | --------------------------------------------------------------------------------------------------------------------- | | `finalize-slot-buffer-size` | `usize` | `1000` | Buffer size for queuing finalize-slot events. | | `accounts-owner-map-enabled` | `bool` | `false` | Enables in-memory account-to-owner map for closed-account handling and owner-change tracking. Increases memory usage. | +| `cleanup-interval-slots` | `u64` | `1` | Finalized slots between cleanup drains. `1` drains every slot. Higher values coalesce repeat touches of an account into one delete, at the cost of extra account versions every latest-version read scans past. | #### `[supply]` (optional) @@ -581,6 +582,7 @@ the previous behavior. | `excluded-programs` | `Vec` | `[]` | Never create indexes for these program pubkeys. | | `indexer-metrics` | `String` | **required** | `host:port` of the indexer's Prometheus metrics endpoint. Used to defer DDL (create/drop) when the indexer is behind. | | `indexer-metrics-threshold` | `u64` | `5` | `cloudbreak_finalize_slot_handler_queue_size` threshold above which DDL is deferred. | +| `indexer-cleanup-lag-threshold` | `u64` | `32` | `cloudbreak_cleanup_lag_slots` threshold above which DDL is deferred. An indexer that does not publish the gauge reads as no pressure. | | `max-auto-indexes` | `usize?` | (none) | Optional cap on total indexes on the `snapshot_accounts` table. | | `index-eviction-enabled` | `bool` | `false` | Enable usage-based eviction of idle auto-indexes. | | `use-supply-for-eviction` | `bool` | `false` | When `true`, also require supply-idle (`last_seen_used` / no `idx_scan` growth) for eviction eligibility. Off = demand-idle + age-grace only. Enabling is more conservative but the eviction-pass supply refresh can collapse the candidates list for up to `index-min-idle` afterward on high-rotation DBs (mitigate with more frequent eviction passes). | @@ -644,7 +646,7 @@ The query tracker enables automatic creation of database indexes based on observ 2. When a pattern's cumulative demand reaches `index-generation-threshold`, it becomes a creation candidate. Demand is cumulative and persisted — there is no reset interval. 3. The creation loop wakes every `index-creation-delay`, ranks the candidates on read according to `priority-mode` (`frequency`, `cost`, `cost-per-hit`, or `weighted`), and builds the highest-priority one. Eviction uses the **same** ranking in reverse (drops the lowest), so "most worth building" and "least worth keeping" are two ends of one order. Indexes are always created as a **pair** on the `accounts` and `snapshot_accounts` tables. 4. Only one index pair is created per tick, to avoid overloading the database. -5. Before creating (or dropping) an index, the query tracker checks the indexer's `cloudbreak_finalize_slot_handler_queue_size` metric. If it exceeds `indexer-metrics-threshold`, DDL is deferred until the indexer catches up. +5. Before creating (or dropping) an index, the query tracker checks two indexer gauges: `cloudbreak_finalize_slot_handler_queue_size` against `indexer-metrics-threshold`, and `cloudbreak_cleanup_lag_slots` against `indexer-cleanup-lag-threshold`. Either one over its threshold defers DDL until the indexer catches up. 6. If `max-auto-indexes` is set, creation is bounded by two sizes: the **fill target** `floor(eviction-fill-threshold × max-auto-indexes)` and the hard **cap** `max-auto-indexes`. Below the target the top candidate is built freely; in the buffer band `(target, cap]` a **creation-time value guard** (see below) only builds a candidate that out-values the index it would displace — unless nothing is reclaimable to displace, in which case it builds anyway up to the cap and lets eviction reclaim later; at the cap creation is **paused** until eviction reclaims the buffer. Candidates are kept queued in `index_patterns` with no loss of data. 7. Use `included-programs` and `excluded-programs` to control which programs are eligible for automatic indexing. @@ -763,6 +765,8 @@ When the `[snapshot]` section (with `[snapshot.tracker_endpoint]`) is present in Set `accounts-owner-map-enabled = true` in the indexer config to maintain an in-memory map of account pubkey to owner + slot. This is used for tracking owner changes and closed-account handling. It increases memory usage but improves correctness for owner-change scenarios. +The map also speeds up the finalize-slot cleanup. The `accounts` and `snapshot_accounts` tables are hash-partitioned by owner, so a cleanup DELETE that only knows the pubkey has to probe all 64 partitions. When the map is enabled, the indexer captures each account's owner at save time and the cleanup carries (pubkey, owner) pairs, letting Postgres prune each delete to a single partition. For an account whose owner changed, the old-owner pair recorded at the change routes the deletion of its old rows and closed-account mask to the old owner's partition. When the map is disabled the cleanup runs the no-owner SQL, with no owner routing. + ### Slot Synchronizer The API server's `[slot-syncronizer]` section controls periodic slot fetching from the database. Enabled by default (200ms interval). Disable with `enabled = false` if not needed. @@ -859,7 +863,7 @@ All metrics are emitted in the Prometheus text exposition format on each service | `cloudbreak_chunk_processing` | Histogram | `origin` | Per-chunk DB insert latency in **seconds**. `origin` = `block`. Buckets: 10 ms – 10 s. | | `cloudbreak_block_size` | Histogram | `origin` | Block size in bytes. `origin` = `block`. Buckets: 100 KB – 30 MB. | | `cloudbreak_chunk_size` | Histogram | `origin` | Chunk size in bytes. `origin` = `chunk`. Buckets: 50 KB – 2 MB. | -| `cloudbreak_finalize_slot` | Histogram | `origin` | Finalize-slot stage latency in **seconds**. `origin` values: `total` (full finalize cycle including all sub-tasks), `cleanup_closed_accounts`, `cleanup_accounts_batch`, `cleanup_snapshot_accounts_batch`, `cleanup_snapshot_closed_accounts`, `cleanup_startup_snapshot_accounts_batch`. Buckets: 5 ms – 10 s. | +| `cloudbreak_finalize_slot` | Histogram | `origin` | Finalize-slot stage latency in **seconds**. `origin` values: `total` (the finalize cycle, which writes the finalized marker and queues the slot's cleanup keys), `cleanup_accounts`, `cleanup_snapshot_accounts`, `cleanup_startup_snapshot_accounts` (the cleanup statements, run by the drainer off this worker). Buckets: 5 ms – 10 s. | | `cloudbreak_finalize_slot_deleted_accounts` | Histogram | — | Number of older account versions deleted per `cleanup_accounts` call. Useful to spot slots that churn an unusually large number of rows. | | `cloudbreak_new_accounts_in_slot` | Histogram | `origin` | Accounts seen in a slot. `origin` = `new_accounts_in_slot` (accounts net-new to the database after finalize-time deduplication) or `block_accounts_total` (raw count of account updates in the block, including overwrites). The ratio gives a sense of write churn per slot. | | `cloudbreak_grpc_buffer_channel_size` | Histogram | `origin` | gRPC ingestion buffer depth sampled per push. `origin` = `grpc_buffer_channel_size`. Buckets: 1 – 10 000. Rising values indicate the indexer is falling behind the stream. | @@ -871,8 +875,9 @@ All metrics are emitted in the Prometheus text exposition format on each service | `cloudbreak_db_errors` | Counter | — | All database errors (insert / cleanup / read) hit by the indexer. When the cumulative count exceeds `database.max-db-errors-threshold` (default `100`), the indexer process exits. | | `cloudbreak_closed_accounts_per_slot` | Histogram | — | Number of accounts marked closed per slot. Buckets: 1 – 10 000. | | `cloudbreak_insert_closed_accounts_per_slot_ms` | Histogram | — | Latency in **milliseconds** of inserting closed-account markers for one slot. Buckets: 0.1 ms – 10 s. | -| `cloudbreak_current_tokio_tasks` | IntGauge | `task_type` | Current count of spawned Tokio tasks per category. `task_type` values: `grpc`, `finalize_slot_internal`, `self_healing`, `self_healing_fill_gaps`, `snapshot_processing`, `insert_accounts_chunk`, `insert_closed_accounts`, `startup_snapshot_accounts_cleanup`, `metrics_server`, `epoch_stakes_recomputer`, `largest_accounts_pruner`, `supply_cache_sweeper`. Counter is guarded by `TokioTaskCounterGuard` so panicking tasks still decrement. | +| `cloudbreak_current_tokio_tasks` | IntGauge | `task_type` | Current count of spawned Tokio tasks per category. `task_type` values: `grpc`, `slot_finalizer`, `self_healing_fill_gaps`, `snapshot_processing`, `insert_accounts_chunk`, `insert_closed_accounts`, `startup_snapshot_accounts_cleanup`, `finalize_cleanup_drainer`, `metrics_server`, `epoch_stakes_recomputer`, `largest_accounts_pruner`, `supply_cache_sweeper`. Counter is guarded by `TokioTaskCounterGuard` so panicking tasks still decrement. | | `cloudbreak_finalize_slot_handler_queue_size` | IntGauge | — | Pending items in the finalize-slot handler channel. The query tracker reads this via the configured `indexer-metrics` endpoint to pause `CREATE INDEX` when the indexer is behind (`indexer-metrics-threshold`). | +| `cloudbreak_cleanup_lag_slots` | IntGauge | — | Finalized slots since the oldest cleanup key still waiting. Cleanup runs off the finalize worker, so this is the signal that it is falling behind, and the query tracker reads it as the second DDL gate (`indexer-cleanup-lag-threshold`). | | `cloudbreak_largest_accounts_db_errors` | Counter | — | Largest-accounts DB write/prune errors (record persist, stale/cleared-mint deletes, and the pruner). Only emitted when a largest-accounts section (`[largest-accounts]` / `[token-largest-accounts]`) is enabled. | | `cloudbreak_largest_accounts_stale_mints` | IntGauge | — | Number of tracked mints currently marked stale in the largest-accounts tracker (reservoir exhausted or a persist failure dropped the record). A non-zero value means those mints have no persisted record and `getTokenLargestAccounts` fails fast for them. Only emitted when a largest-accounts section (`[largest-accounts]` / `[token-largest-accounts]`) is enabled. | | `cloudbreak_supply_total_lamports` | IntGauge | — | Running total supply in lamports, updated on every committed block while the supply tracker is live. Only emitted when the `[supply]` section is enabled. | diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index 631583b..9d07622 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -416,6 +416,12 @@ pub struct IndexConfig { /// section is present with `enabled = true`. #[serde(rename = "largest-accounts")] pub largest_accounts: Option, + /// Finalized slots between cleanup drains. 1 drains every slot. + #[serde( + rename = "cleanup-interval-slots", + default = "IndexConfig::default_cleanup_interval_slots" + )] + pub cleanup_interval_slots: u64, /// The indexer maintains per-mint `getTokenLargestAccounts` tops when this /// section is present with `enabled = true`. #[serde(rename = "token-largest-accounts")] @@ -523,6 +529,10 @@ impl IndexConfig { 1000 } + const fn default_cleanup_interval_slots() -> u64 { + 1 + } + /// Smallest configured prune interval among the enabled largest-accounts /// sections; `None` when neither section is enabled. pub fn largest_accounts_prune_interval_slots(&self) -> Option { @@ -1024,6 +1034,15 @@ pub struct QueryTrackerConfig { default = "QueryTrackerConfig::default_indexer_metrics_threshold" )] pub indexer_metrics_threshold: u64, + /// If `cloudbreak_cleanup_lag_slots` exceeds this, CREATE and DROP INDEX are deferred. + /// The indexer's finalize queue stops reflecting database pressure once cleanup is + /// decoupled from the finalize worker, so this gauge is the second signal. An indexer + /// that does not publish it reads as no pressure. + #[serde( + rename = "indexer-cleanup-lag-threshold", + default = "QueryTrackerConfig::default_indexer_cleanup_lag_threshold" + )] + pub indexer_cleanup_lag_threshold: u64, /// Optional cap on the total number of indexes on the target table. #[serde(rename = "max-auto-indexes", default)] pub max_auto_indexes: Option, @@ -1321,6 +1340,10 @@ impl QueryTrackerConfig { 5 } + const fn default_indexer_cleanup_lag_threshold() -> u64 { + 32 + } + pub fn deserialize_indexer_metrics<'de, D>(deserializer: D) -> Result where D: Deserializer<'de>, @@ -1347,6 +1370,7 @@ impl Default for QueryTrackerConfig { excluded_programs: Vec::new(), indexer_metrics: String::default(), indexer_metrics_threshold: Self::default_indexer_metrics_threshold(), + indexer_cleanup_lag_threshold: Self::default_indexer_cleanup_lag_threshold(), max_auto_indexes: None, index_eviction_enabled: Self::default_index_eviction_enabled(), mark_unhealthy_for_eviction: Self::default_mark_unhealthy_for_eviction(), diff --git a/crates/core/src/modules/account_owner_map.rs b/crates/core/src/modules/account_owner_map.rs index 9e09027..59b19ff 100644 --- a/crates/core/src/modules/account_owner_map.rs +++ b/crates/core/src/modules/account_owner_map.rs @@ -125,6 +125,41 @@ impl AccountOwnerMap { } } + /// Returns the (pubkeys, owners) pairs that owner-route the closed-account cleanup at + /// finalization: closed accounts present in the map plus the slot's changed-owner pairs. + /// + /// Read-only: [`Self::save_closed_accounts`] later drains the same entries for the mask insert. + pub fn closed_cleanup_pairs( + &self, + closed_accounts: &[Vec], + slot: u64, + ) -> (Vec>, Vec>) { + let mut pubkeys = Vec::new(); + let mut owners = Vec::new(); + if let Some(accounts) = &self.accounts { + let map = accounts.read().expect("Failed to read accounts"); + for pubkey_bytes in closed_accounts { + let pubkey = Pubkey::try_from(pubkey_bytes.as_slice()).unwrap(); + if let Some(item) = map.get(&pubkey) { + pubkeys.push(pubkey_bytes.clone()); + owners.push(item.owner.to_bytes().to_vec()); + } + } + } + if let Some(changed) = self + .changed_owners + .lock() + .expect("Failed to lock changed_owners") + .get(&slot) + { + for ChangedOwner { pubkey, owner } in changed { + pubkeys.push(pubkey.to_bytes().to_vec()); + owners.push(owner.to_bytes().to_vec()); + } + } + (pubkeys, owners) + } + /// For accounts present in the map, saves the mock "closed account" mask into the DB using /// the previous owner and the new slot. /// diff --git a/crates/index/src/db/cleanup.sql b/crates/index/src/db/cleanup.sql index e31e758..6996ad0 100644 --- a/crates/index/src/db/cleanup.sql +++ b/crates/index/src/db/cleanup.sql @@ -3,17 +3,10 @@ * Copyright 2025-2026 Triton One Limited. All rights reserved. */ --- EXPLAIN ANALYZE -DELETE FROM accounts_table_name -- placeholder to be replaced with the actual table name -WHERE - pubkey = ANY($2) -- $2 is array of pubkeys - AND slot < $1 -- $1 is finalized_slot - -- tecnically this is not needed, because if we keep track of the updated accounts, if there is - -- any older version we can safely delete it (so this is just an extra check) - -- AND EXISTS ( - -- SELECT 1 FROM accounts AS a2 - -- WHERE - -- a2.pubkey = accounts.pubkey - -- AND a2.slot > accounts.slot - -- AND a2.slot < $1 - -- ); +-- Unrouted cleanup, for nodes with the owner map off and for keys the map did not know. +-- Same per-key exclusive cutoff as cleanupWithOwner.sql without the owner terms. Every read +-- path dedupes on pubkey alone, so "newest version of this pubkey regardless of owner" is the +-- visibility rule this form matches. On a partitioned table it fans out to every partition. +DELETE FROM accounts_table_name a -- placeholder to be replaced with the actual table name +USING unnest($1::bytea[], $2::bigint[]) AS k(pubkey, cutoff) +WHERE a.pubkey = k.pubkey AND a.slot < k.cutoff diff --git a/crates/index/src/db/cleanupWithOwner.sql b/crates/index/src/db/cleanupWithOwner.sql new file mode 100644 index 0000000..da16e2a --- /dev/null +++ b/crates/index/src/db/cleanupWithOwner.sql @@ -0,0 +1,12 @@ +-- SPDX-License-Identifier: AGPL-3.0-only +/* + * Copyright 2025-2026 Triton One Limited. All rights reserved. + */ + +-- Owner-routed cleanup. Each key carries its own exclusive cutoff, so one form expresses both +-- an open key (cutoff = the finalized slot it was written at) and a closed or owner-moved key +-- (cutoff = one past the slot of its mask). The owner lets Postgres prune the DELETE to a single +-- hash partition per key, and `slot < k.cutoff` sits inside the index condition. +DELETE FROM accounts_table_name a -- placeholder to be replaced with the actual table name +USING unnest($1::bytea[], $2::bytea[], $3::bigint[]) AS k(pubkey, owner, cutoff) +WHERE a.owner = k.owner AND a.pubkey = k.pubkey AND a.slot < k.cutoff diff --git a/crates/index/src/db/closedAccountscleanup.sql b/crates/index/src/db/closedAccountscleanup.sql deleted file mode 100644 index 0fba2fa..0000000 --- a/crates/index/src/db/closedAccountscleanup.sql +++ /dev/null @@ -1,9 +0,0 @@ --- SPDX-License-Identifier: AGPL-3.0-only -/* - * Copyright 2025-2026 Triton One Limited. All rights reserved. - */ - -DELETE FROM accounts -WHERE - pubkey = ANY($1) -- $1 is array of pubkeys - AND slot <= $2; -- $2 is finalized_slot diff --git a/crates/index/src/db_queries.rs b/crates/index/src/db_queries.rs index 1f2099c..34979be 100644 --- a/crates/index/src/db_queries.rs +++ b/crates/index/src/db_queries.rs @@ -3,10 +3,7 @@ * Copyright 2025-2026 Triton One Limited. All rights reserved. */ -use std::{ - sync::{Arc, Mutex}, - time::Duration, -}; +use std::time::Duration; use cloudbreak_core::{IndexConfig, modules::account_owner_map::AccountOwnerMap}; use cloudbreak_entity::{accounts, slots}; @@ -148,153 +145,6 @@ pub fn insert_closed_accounts( Some(handle) } -/// Deletes the last special "closed" version inserted for the set of closed accounts for the given slot -pub async fn cleanup_closed_accounts( - db: &DatabaseConnection, - pubkeys: Vec>, - slot: u64, - config: &IndexConfig, -) { - let query_timeout = Duration::from_secs(config.database.finalize_slot_queries_timeout); - - let start_time = Instant::now(); - let cleanup_closed_accounts_sql = include_str!("db/closedAccountscleanup.sql"); - - if pubkeys.is_empty() { - return; - } - - let query = db.execute(Statement::from_sql_and_values( - sea_orm::DatabaseBackend::Postgres, - cleanup_closed_accounts_sql, - vec![ - Value::Array( - sea_orm::sea_query::ArrayType::Bytes, - Some(Box::new( - pubkeys - .into_iter() - .map(|pubkey| Value::Bytes(Some(Box::new(pubkey)))) - .collect(), - )), - ), - Value::BigInt(Some(slot as i64)), - ], - )); - - let result = timeout(query_timeout, query) - .await - .unwrap_or_else(|elapsed| { - tracing::error!("cleanup_closed_accounts timeout ERROR: {}", elapsed); - metrics::increment_db_errors(); - Err(sea_orm::DbErr::RecordNotInserted) - }); - - match result { - Ok(res) => { - tracing::debug!( - target: "cleanup_closed_accounts", - "cleaned up {} closed accounts for slot {}", - res.rows_affected(), - slot - ); - } - Err(e) => { - tracing::error!( - "cleanup_closed_accounts: failed to cleanup closed accounts for slot {}: {}", - slot, - e - ); - metrics::increment_db_errors(); - } - } - - metrics::record_finalize_slot( - start_time.elapsed().as_secs_f64(), - "cleanup_closed_accounts", - ); -} - -/// Cleans up older versions (slot less than the received slot) of the accounts from the database (for the given table) -pub async fn cleanup_accounts( - db: &DatabaseConnection, - pubkeys: Vec>, - slot: u64, - table_name: &str, - new_accounts_in_slot: Arc>, - metrics_tag: &str, - config: &IndexConfig, -) { - let start_time = Instant::now(); - let pubkeys_len = pubkeys.len(); - let cleanup_sql = include_str!("db/cleanup.sql"); - let cleanup_sql = cleanup_sql.replace("accounts_table_name", table_name); - let query_timeout = Duration::from_secs(config.database.finalize_slot_queries_timeout); - - let query = db.execute(Statement::from_sql_and_values( - sea_orm::DatabaseBackend::Postgres, - cleanup_sql, - vec![ - Value::BigInt(Some(slot as i64)), - Value::Array( - sea_orm::sea_query::ArrayType::Bytes, - Some(Box::new( - pubkeys - .into_iter() - .map(|pubkey| Value::Bytes(Some(Box::new(pubkey)))) - .collect(), - )), - ), - ], - )); - - let result = timeout(query_timeout, query) - .await - .unwrap_or_else(|elapsed| { - tracing::error!("cleanup_accounts timeout ERROR: {}", elapsed); - metrics::increment_db_errors(); - Err(sea_orm::DbErr::RecordNotInserted) - }); - - match result { - Ok(res) => { - tracing::debug!( - "finalize_slot: deleted {} old account versions for slot {} - accounts_in_batch: {}", - res.rows_affected(), - slot, - pubkeys_len - ); - - let mut new_accounts_in_slot = new_accounts_in_slot - .lock() - .expect("Failed to lock new_accounts_in_slot"); - let deleted = res.rows_affected() as usize; - - metrics::FINALIZE_SLOT_DELETED_ACCOUNTS.observe(deleted as f64); - - if table_name != "snapshot_accounts" { - if pubkeys_len >= deleted { - *new_accounts_in_slot += pubkeys_len - deleted; - } else { - tracing::debug!( - target: "finalize_slot_debug", - "finalize_slot: deleted more accounts than the batch size({}) for {} - slot {}: {}", - pubkeys_len, - table_name, - slot, - deleted - ); - } - } - } - Err(e) => { - tracing::error!("finalize_slot: failed to finalize slot {}: {}", slot, e); - metrics::increment_db_errors(); - } - } - - metrics::record_finalize_slot(start_time.elapsed().as_secs_f64(), metrics_tag); -} - pub async fn insert_slot( slot: u64, block_time: Option, diff --git a/crates/index/src/indexer.rs b/crates/index/src/indexer.rs index d1a0c19..0fd265d 100644 --- a/crates/index/src/indexer.rs +++ b/crates/index/src/indexer.rs @@ -131,12 +131,22 @@ pub async fn run(config: &str) -> CloudbreakResult<()> { prune_slot_tx.subscribe(), ); + // The finalize worker hands each slot's cleanup keys here and returns. + let cleanup = modules::cleanup::CleanupHandle::new(config.cleanup_interval_slots); + modules::cleanup::spawn_cleanup_drainer( + cleanup.clone(), + Arc::new(db.clone()), + updated_accounts_during_startup.clone(), + Duration::from_secs(config.database.finalize_slot_queries_timeout), + ); + let slot_finalizer = SlotFinalizer::spawn( db.clone(), config.clone(), updated_accounts_during_startup.clone(), health.clone(), prune_slot_tx, + cleanup, ); let indexer_state = IndexerState { @@ -329,5 +339,12 @@ pub async fn process_update( pub struct AccountsReceivedPerBlock { pub block_time: Option, pub accounts: Vec>, + /// Owner of each entry in `accounts` (same order). Empty when the owner map is disabled, + /// which makes the finalize cleanup fall back to the no-owner SQL. + pub accounts_owners: Vec>, pub closed_accounts: Vec>, + /// (pubkey, owner) pairs that owner-route the closed-account cleanup, captured at save time + /// from the owner map. Includes the old-owner pairs for in-slot owner changes. + pub closed_cleanup_pubkeys: Vec>, + pub closed_cleanup_owners: Vec>, } diff --git a/crates/index/src/metrics.rs b/crates/index/src/metrics.rs index 0c475e0..b0f7c93 100644 --- a/crates/index/src/metrics.rs +++ b/crates/index/src/metrics.rs @@ -277,6 +277,11 @@ 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 CLEANUP_LAG_SLOTS: IntGauge = IntGauge::new( + "cloudbreak_cleanup_lag_slots", "Finalized slots since the oldest unprocessed cleanup enqueue" + ) + .expect("Failed to create cleanup lag slots gauge"); } pub fn record_block_processing(elapsed: f64, origin: &str) { @@ -387,6 +392,7 @@ pub fn register_collectors() { register!(FINALIZE_SLOT_DELETED_ACCOUNTS); register!(LARGEST_ACCOUNTS_DB_ERRORS); register!(LARGEST_ACCOUNTS_STALE_MINTS); + register!(CLEANUP_LAG_SLOTS); register!(SUPPLY_CACHE_ENTRIES); register!(SUPPLY_CACHE_HITS_TOTAL); register!(SUPPLY_CACHE_MISSES_TOTAL); diff --git a/crates/index/src/modules/cleanup/drain.rs b/crates/index/src/modules/cleanup/drain.rs new file mode 100644 index 0000000..0c63dd9 --- /dev/null +++ b/crates/index/src/modules/cleanup/drain.rs @@ -0,0 +1,149 @@ +// SPDX-License-Identifier: AGPL-3.0-only +/* + * Copyright 2025-2026 Triton One Limited. All rights reserved. + */ + +//! The spawned drainer. +//! +//! One task, one batch at a time. A failed drain is queued again and reattempted on the next +//! round, never inside this one: every failed statement counts against the process-wide +//! `max-db-errors-threshold`, so a batch that fails forever must cost one error per slot, which +//! is what the inline cleanup it replaces cost. +//! +//! Not health gated. It publishes no state and deletes only rows the read paths already hide, +//! and the unhealthy window is a gap fill, which is when the queue holds the most work. This +//! deviates from `docs/feature-guideline.md` rule 8 deliberately. + +use std::sync::Arc; +use std::time::Duration; + +use super::CleanupHandle; +use super::persist::{self, CleanupExecutor}; +use crate::metrics; +use crate::modules::finalize_slot::UpdatedAccountsDuringStartup; + +pub fn spawn_cleanup_drainer( + handle: CleanupHandle, + executor: Arc, + startup: UpdatedAccountsDuringStartup, + query_timeout: Duration, +) -> tokio::task::JoinHandle<()> +where + E: CleanupExecutor, +{ + tokio::spawn(async move { + let _guard = metrics::TokioTaskCounterGuard::new("finalize_cleanup_drainer"); + run(handle, executor, startup, query_timeout).await; + }) +} + +async fn run( + handle: CleanupHandle, + executor: Arc, + startup: UpdatedAccountsDuringStartup, + query_timeout: Duration, +) { + loop { + handle.wait_for_work().await; + handle.note_drain(); + + let taken = handle.take_all(); + if taken.is_empty() { + handle.finish(); + continue; + } + + let keys = taken.len(); + match persist::drain_all( + executor.as_ref(), + &taken, + query_timeout, + startup.is_startup(), + ) + .await + { + Ok(new_accounts) => { + handle.finish(); + metrics::record_new_accounts_in_slot(new_accounts, "new_accounts_in_slot"); + } + Err(error) => { + handle.reinsert(taken); + tracing::warn!( + target: "finalize_cleanup", + "cleanup of {} keys failed, queued for the next round: {}", + keys, + error + ); + } + } + + handle.publish_lag(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::modules::cleanup::pending::tests::routed; + use crate::modules::cleanup::persist::tests::{RecordingExecutor, allow_db_errors}; + use crate::modules::snapshot::SnapshotProcessingState; + use crate::modules::{finalize_slot::UpdatedAccountsDuringStartup, health::ServiceHealth}; + use std::sync::Mutex; + + fn ready_startup() -> UpdatedAccountsDuringStartup { + UpdatedAccountsDuringStartup::new( + Arc::new(Mutex::new(SnapshotProcessingState::FinishedAndCleanedUp)), + ServiceHealth::new(sea_orm::DatabaseConnection::Disconnected), + ) + } + + async fn drain_once(handle: &CleanupHandle, executor: Arc) { + let task = spawn_cleanup_drainer( + handle.clone(), + executor, + ready_startup(), + Duration::from_secs(5), + ); + for _ in 0..200 { + if handle.is_quiescent() { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + task.abort(); + } + + #[tokio::test] + async fn the_run_loop_drains_the_queue_to_empty() { + let handle = CleanupHandle::new(1); + let executor = Arc::new(RecordingExecutor::default()); + handle.enqueue(100, &[(routed(1, 1), 100), (routed(1, 2), 100)]); + + drain_once(&handle, executor.clone()).await; + + assert!(handle.is_quiescent(), "the drainer left work behind"); + assert_eq!( + executor.issued(), + vec!["snapshot_accounts:routed", "accounts:routed"] + ); + } + + #[tokio::test] + async fn a_failed_drain_leaves_its_keys_queued() { + allow_db_errors(); + let handle = CleanupHandle::new(1); + let executor = Arc::new(RecordingExecutor::failing("snapshot_accounts")); + handle.enqueue(100, &[(routed(1, 1), 100)]); + + let task = spawn_cleanup_drainer( + handle.clone(), + executor, + ready_startup(), + Duration::from_secs(5), + ); + tokio::time::sleep(Duration::from_millis(50)).await; + task.abort(); + + assert!(!handle.is_quiescent(), "the key must not be lost"); + } +} diff --git a/crates/index/src/modules/cleanup/mod.rs b/crates/index/src/modules/cleanup/mod.rs new file mode 100644 index 0000000..0498bf3 --- /dev/null +++ b/crates/index/src/modules/cleanup/mod.rs @@ -0,0 +1,140 @@ +// SPDX-License-Identifier: AGPL-3.0-only +/* + * Copyright 2025-2026 Triton One Limited. All rights reserved. + */ + +//! Deferred finalize-slot cleanup. +//! +//! The finalize worker writes the finalized marker, hands the slot's cleanup keys to this queue, +//! and returns. A background drainer runs the DELETEs, so no slot's marker waits on the previous +//! slot's deletes. +//! +//! | File | Role | +//! |---|---| +//! | `mod.rs` | The handle. | +//! | `pending.rs` | The pending map, the merge rule, the keys a slot contributes. | +//! | `drain.rs` | The spawned drainer. | +//! | `persist.rs` | The statements and the two-table ordering rule. | +//! +//! # The cutoff rule +//! +//! Each key carries one exclusive cutoff, merged by max, and the drain deletes its rows below +//! that slot. An open key at finalized slot `s` takes `s`, so the row written at `s` survives. A +//! closed or owner-moved key takes `s + 1`, so its mask at `s` goes with the rows it shadows. +//! Every cutoff comes from a slot at which the key was written or masked, which is why max is +//! safe in any arrival order, including a repaired slot finalizing below the frontier. +//! +//! # Always on +//! +//! There is no enable flag. A node without cleanup fills its disk and slows every latest-version +//! read. `cleanup-interval-slots` is the only knob. This deviates from +//! `docs/feature-guideline.md` rules 2 and 5 deliberately. +//! +//! # Where this lives +//! +//! Rule 1 puts a feature in `crates/core` so the API can share a read path. This one has no +//! reader and no API surface, and its dependencies are all in `crates/index`. Deliberate. +//! +//! # Restart +//! +//! The queue is in memory and a restart drops it. Nothing incorrect becomes visible: every lost +//! key describes superseded rows or a mask, which the read paths already hide, and the rows clear +//! on the key's next touch under a higher cutoff. + +pub mod drain; +pub mod pending; +pub mod persist; + +use std::sync::{Arc, Mutex}; + +use tokio::sync::Notify; + +pub use drain::spawn_cleanup_drainer; +pub use pending::{CleanupKey, keys_for_slot}; + +use crate::metrics; +use pending::{Pending, Taken}; + +struct Shared { + pending: Mutex, + /// Wakes the drainer. The finalize worker never blocks on it. + work_available: Notify, +} + +/// Cheap-clone handle over the pending map, held by the finalize worker and the drainer. +#[derive(Clone)] +pub struct CleanupHandle { + shared: Arc, +} + +impl CleanupHandle { + /// `interval_slots` is clamped to at least 1, so a zero in config drains every slot. + pub fn new(interval_slots: u64) -> Self { + Self { + shared: Arc::new(Shared { + pending: Mutex::new(Pending::new(interval_slots.max(1))), + work_available: Notify::new(), + }), + } + } + + /// Hands one finalized slot's keys to the queue. Never blocks. + pub fn enqueue(&self, slot: u64, items: &[(CleanupKey, u64)]) { + let wake = { + let mut pending = self.lock(); + pending.enqueue(slot, items); + pending.should_drain() + }; + if wake { + self.shared.work_available.notify_one(); + } + self.publish_lag(); + } + + pub async fn wait_for_work(&self) { + self.shared.work_available.notified().await; + } + + pub fn take_all(&self) -> Taken { + self.lock().take_all() + } + + pub fn finish(&self) { + self.lock().finish(); + } + + /// Returns a failed drain's keys to the queue. + pub fn reinsert(&self, taken: Taken) { + self.lock().reinsert(taken); + } + + pub fn note_drain(&self) { + self.lock().note_drain(); + } + + pub fn is_quiescent(&self) -> bool { + self.lock().is_quiescent() + } + + pub fn publish_lag(&self) { + let lag = self.lock().lag_slots(); + metrics::CLEANUP_LAG_SLOTS.set(lag as i64); + } + + fn lock(&self) -> std::sync::MutexGuard<'_, Pending> { + self.shared.pending.lock().expect("Failed to lock cleanup") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pending::tests::routed; + + #[test] + fn a_zero_interval_drains_every_slot() { + let handle = CleanupHandle::new(0); + handle.enqueue(100, &[(routed(1, 1), 100)]); + assert!(!handle.is_quiescent()); + } +} diff --git a/crates/index/src/modules/cleanup/pending.rs b/crates/index/src/modules/cleanup/pending.rs new file mode 100644 index 0000000..56d2c7e --- /dev/null +++ b/crates/index/src/modules/cleanup/pending.rs @@ -0,0 +1,551 @@ +// SPDX-License-Identifier: AGPL-3.0-only +/* + * Copyright 2025-2026 Triton One Limited. All rights reserved. + */ + +//! The pending map and the keys one finalized slot contributes to it. + +use std::collections::{HashMap, HashSet}; + +use crate::indexer::AccountsReceivedPerBlock; + +/// A cleanup target. `owner` is `Some` when the owner map knew the account, which lets the +/// DELETE prune to one hash partition. `None` runs the unrouted form. +#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] +pub struct CleanupKey { + pub owner: Option<[u8; 32]>, + pub pubkey: [u8; 32], +} + +/// Which SQL form a key takes. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum KeyForm { + Routed, + Unrouted, +} + +impl KeyForm { + pub fn as_str(self) -> &'static str { + match self { + Self::Routed => "routed", + Self::Unrouted => "unrouted", + } + } +} + +/// One drain's worth of keys, split by the SQL form they take. +#[derive(Debug, Default)] +pub struct Taken { + pub routed: Vec<(CleanupKey, u64)>, + pub unrouted: Vec<(CleanupKey, u64)>, + /// The frontier when the oldest of these keys was queued. Drives the lag gauge. + pub oldest_stamp: u64, +} + +impl Taken { + pub fn is_empty(&self) -> bool { + self.routed.is_empty() && self.unrouted.is_empty() + } + + pub fn len(&self) -> usize { + self.routed.len() + self.unrouted.len() + } +} + +/// Keys derived from one finalized slot, plus the closes the owner map did not cover. +pub struct DerivedKeys { + pub items: Vec<(CleanupKey, u64)>, + pub uncovered_closes: usize, +} + +/// Turns one finalized slot's block record into keys with cutoffs. +/// +/// Open keys take the finalized slot as their cutoff. Closed and owner-moved keys take one past +/// it, so the mask written at the slot goes with the rows it shadows. +/// +/// On a node with the owner map on, only closes the map knew are queued. `save_block` collects +/// every zero-lamport account in the block before the program filter runs, so the rest are +/// accounts this node does not index and they have no rows to delete. +pub fn keys_for_slot( + accounts: &AccountsReceivedPerBlock, + slot: u64, + owner_map_enabled: bool, +) -> DerivedKeys { + let closed_cutoff = slot.saturating_add(1); + let mut items = + Vec::with_capacity(accounts.accounts.len() + accounts.closed_cleanup_pubkeys.len()); + + let open_routed = + owner_map_enabled && accounts.accounts_owners.len() == accounts.accounts.len(); + for (index, pubkey) in accounts.accounts.iter().enumerate() { + let Some(pubkey) = to_key_bytes(pubkey) else { + continue; + }; + let owner = if open_routed { + to_key_bytes(&accounts.accounts_owners[index]) + } else { + None + }; + items.push((CleanupKey { owner, pubkey }, slot)); + } + + let mut uncovered_closes = 0; + if owner_map_enabled { + let paired = accounts.closed_cleanup_owners.len() == accounts.closed_cleanup_pubkeys.len(); + let mut covered: HashSet<&[u8]> = HashSet::new(); + if paired { + for (index, pubkey) in accounts.closed_cleanup_pubkeys.iter().enumerate() { + covered.insert(pubkey.as_slice()); + let Some(pubkey) = to_key_bytes(pubkey) else { + continue; + }; + let owner = to_key_bytes(&accounts.closed_cleanup_owners[index]); + items.push((CleanupKey { owner, pubkey }, closed_cutoff)); + } + } + uncovered_closes = accounts + .closed_accounts + .iter() + .filter(|pubkey| !covered.contains(pubkey.as_slice())) + .count(); + } else { + for pubkey in &accounts.closed_accounts { + let Some(pubkey) = to_key_bytes(pubkey) else { + continue; + }; + items.push(( + CleanupKey { + owner: None, + pubkey, + }, + closed_cutoff, + )); + } + } + + DerivedKeys { + items, + uncovered_closes, + } +} + +fn to_key_bytes(bytes: &[u8]) -> Option<[u8; 32]> { + bytes.try_into().ok() +} + +/// Keys waiting to be cleaned, one exclusive cutoff each, merged by max. +/// +/// One drainer takes the whole map at once, so a key is never in two statements and no in-flight +/// bookkeeping per key is needed. A key touched again while a drain is running lands back in the +/// map with its newer cutoff and drains on the next round. +pub(crate) struct Pending { + keys: HashMap, + /// The frontier when the oldest key still queued was first inserted. + oldest_stamp: Option, + /// Same, for the batch a drain is running right now. + in_flight_oldest: Option, + high_water: u64, + enqueued_since_drain: u64, + last_drain_slot: Option, + interval_slots: u64, +} + +impl Pending { + pub(crate) fn new(interval_slots: u64) -> Self { + Self { + keys: HashMap::new(), + oldest_stamp: None, + in_flight_oldest: None, + high_water: 0, + enqueued_since_drain: 0, + last_drain_slot: None, + interval_slots, + } + } + + /// Merges one finalized slot's keys in and returns how many coalesced into a queued key. + pub(crate) fn enqueue(&mut self, slot: u64, items: &[(CleanupKey, u64)]) -> usize { + self.high_water = self.high_water.max(slot); + self.enqueued_since_drain = self.enqueued_since_drain.saturating_add(1); + + let stamp = self.high_water; + let mut coalesced = 0; + for (key, cutoff) in items { + match self.keys.get_mut(key) { + Some(existing) => { + *existing = (*existing).max(*cutoff); + coalesced += 1; + } + None => { + self.keys.insert(*key, *cutoff); + self.oldest_stamp = Some(self.oldest_stamp.unwrap_or(stamp).min(stamp)); + } + } + } + coalesced + } + + /// Takes every queued key, split by SQL form. + pub(crate) fn take_all(&mut self) -> Taken { + let oldest_stamp = self.oldest_stamp.unwrap_or(self.high_water); + let mut taken = Taken { + oldest_stamp, + ..Default::default() + }; + for (key, cutoff) in self.keys.drain() { + if key.owner.is_some() { + taken.routed.push((key, cutoff)); + } else { + taken.unrouted.push((key, cutoff)); + } + } + self.oldest_stamp = None; + self.in_flight_oldest = (!taken.is_empty()).then_some(oldest_stamp); + taken + } + + /// Releases a drain that succeeded. + pub(crate) fn finish(&mut self) { + self.in_flight_oldest = None; + } + + /// Returns a failed drain's keys, keeping the older of the two cutoffs' stamps. + pub(crate) fn reinsert(&mut self, taken: Taken) { + let stamp = taken.oldest_stamp; + for (key, cutoff) in taken.routed.into_iter().chain(taken.unrouted) { + let entry = self.keys.entry(key).or_insert(cutoff); + *entry = (*entry).max(cutoff); + } + if !self.keys.is_empty() { + self.oldest_stamp = Some(self.oldest_stamp.unwrap_or(stamp).min(stamp)); + } + self.in_flight_oldest = None; + } + + /// `true` when the drainer should wake. + /// + /// The count term catches the ordinary case. The span term catches an ancestor walk + /// finalizing several slots in one call and a gap fill jumping the frontier. + pub(crate) fn should_drain(&self) -> bool { + let span = self + .high_water + .saturating_sub(self.last_drain_slot.unwrap_or(self.high_water)); + self.enqueued_since_drain >= self.interval_slots || span >= self.interval_slots + } + + pub(crate) fn note_drain(&mut self) { + self.enqueued_since_drain = 0; + self.last_drain_slot = Some(self.high_water); + } + + /// Finalized slots since the oldest key still waiting, counting a drain in flight. + pub(crate) fn lag_slots(&self) -> u64 { + let oldest = match (self.oldest_stamp, self.in_flight_oldest) { + (Some(a), Some(b)) => Some(a.min(b)), + (Some(a), None) => Some(a), + (None, Some(b)) => Some(b), + (None, None) => None, + }; + oldest.map_or(0, |stamp| self.high_water.saturating_sub(stamp)) + } + + pub(crate) fn is_quiescent(&self) -> bool { + self.keys.is_empty() && self.in_flight_oldest.is_none() + } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + + pub(crate) fn pubkey(byte: u8) -> [u8; 32] { + [byte; 32] + } + + pub(crate) fn routed(owner: u8, pk: u8) -> CleanupKey { + CleanupKey { + owner: Some(pubkey(owner)), + pubkey: pubkey(pk), + } + } + + pub(crate) fn unrouted(pk: u8) -> CleanupKey { + CleanupKey { + owner: None, + pubkey: pubkey(pk), + } + } + + fn cutoff_of(pending: &Pending, key: &CleanupKey) -> Option { + pending.keys.get(key).copied() + } + + fn block( + accounts: Vec>, + owners: Vec>, + closed: Vec>, + closed_pubkeys: Vec>, + closed_owners: Vec>, + ) -> AccountsReceivedPerBlock { + AccountsReceivedPerBlock { + block_time: None, + accounts, + accounts_owners: owners, + closed_accounts: closed, + closed_cleanup_pubkeys: closed_pubkeys, + closed_cleanup_owners: closed_owners, + } + } + + #[test] + fn enqueue_twice_merges_cutoff_by_max() { + let mut pending = Pending::new(1); + let key = routed(1, 1); + pending.enqueue(100, &[(key, 100)]); + pending.enqueue(120, &[(key, 120)]); + assert_eq!(cutoff_of(&pending, &key), Some(120)); + } + + #[test] + fn a_lower_cutoff_after_a_higher_one_is_absorbed() { + let mut pending = Pending::new(1); + let key = routed(1, 1); + pending.enqueue(120, &[(key, 120)]); + // A repaired slot finalizing below the frontier must not lower the bound. + pending.enqueue(90, &[(key, 90)]); + assert_eq!(cutoff_of(&pending, &key), Some(120)); + } + + #[test] + fn open_key_gets_cutoff_s_and_closed_key_gets_s_plus_one() { + let derived = keys_for_slot( + &block( + vec![pubkey(1).to_vec()], + vec![pubkey(9).to_vec()], + vec![pubkey(2).to_vec()], + vec![pubkey(2).to_vec()], + vec![pubkey(9).to_vec()], + ), + 100, + true, + ); + let open = derived + .items + .iter() + .find(|(key, _)| key.pubkey == pubkey(1)) + .expect("open key"); + let closed = derived + .items + .iter() + .find(|(key, _)| key.pubkey == pubkey(2)) + .expect("closed key"); + assert_eq!(open.1, 100); + assert_eq!(closed.1, 101); + } + + #[test] + fn close_then_reopen_merges_to_the_reopen_cutoff() { + let mut pending = Pending::new(1); + let key = routed(1, 1); + pending.enqueue(100, &[(key, 101)]); + pending.enqueue(120, &[(key, 120)]); + assert_eq!(cutoff_of(&pending, &key), Some(120)); + } + + #[test] + fn owner_change_yields_two_keys() { + // The account moved from owner 8 to owner 9 in the slot. + let derived = keys_for_slot( + &block( + vec![pubkey(1).to_vec()], + vec![pubkey(9).to_vec()], + vec![], + vec![pubkey(1).to_vec()], + vec![pubkey(8).to_vec()], + ), + 100, + true, + ); + assert_eq!(derived.items.len(), 2); + let new_owner = derived + .items + .iter() + .find(|(key, _)| key.owner == Some(pubkey(9))) + .expect("new owner key"); + let old_owner = derived + .items + .iter() + .find(|(key, _)| key.owner == Some(pubkey(8))) + .expect("old owner key"); + assert_eq!(new_owner.1, 100, "the live row at the slot must survive"); + assert_eq!(old_owner.1, 101, "the old owner's mask must go"); + } + + #[test] + fn owner_map_off_builds_pubkey_only_keys() { + let derived = keys_for_slot( + &block( + vec![pubkey(1).to_vec()], + vec![], + vec![pubkey(2).to_vec()], + vec![], + vec![], + ), + 100, + false, + ); + assert_eq!(derived.items.len(), 2); + assert!(derived.items.iter().all(|(key, _)| key.owner.is_none())); + assert_eq!(derived.uncovered_closes, 0); + } + + #[test] + fn uncovered_closes_are_counted_and_dropped_on_a_map_on_node() { + // Two closes reached the block, the map knew one. The other is not indexed here. + let derived = keys_for_slot( + &block( + vec![], + vec![], + vec![pubkey(2).to_vec(), pubkey(3).to_vec()], + vec![pubkey(2).to_vec()], + vec![pubkey(9).to_vec()], + ), + 100, + true, + ); + assert_eq!(derived.items.len(), 1); + assert_eq!(derived.uncovered_closes, 1); + } + + #[test] + fn mismatched_owner_lengths_fall_back_to_unrouted_open_keys() { + let derived = keys_for_slot( + &block( + vec![pubkey(1).to_vec(), pubkey(2).to_vec()], + vec![pubkey(9).to_vec()], + vec![], + vec![], + vec![], + ), + 100, + true, + ); + assert_eq!(derived.items.len(), 2); + assert!(derived.items.iter().all(|(key, _)| key.owner.is_none())); + } + + #[test] + fn take_all_splits_by_form_and_empties_the_map() { + let mut pending = Pending::new(1); + pending.enqueue(100, &[(routed(1, 1), 100), (unrouted(2), 100)]); + + let taken = pending.take_all(); + assert_eq!(taken.routed.len(), 1); + assert_eq!(taken.unrouted.len(), 1); + assert!(pending.keys.is_empty()); + assert!(!pending.is_quiescent(), "the drain is still in flight"); + + pending.finish(); + assert!(pending.is_quiescent()); + } + + #[test] + fn a_key_retouched_during_a_drain_stays_queued() { + let mut pending = Pending::new(1); + let key = routed(1, 1); + pending.enqueue(100, &[(key, 100)]); + let taken = pending.take_all(); + + pending.enqueue(101, &[(key, 101)]); + pending.finish(); + + // The drain that finished covered cutoff 100. The newer cutoff is still owed. + assert_eq!(taken.routed[0].1, 100); + assert_eq!(cutoff_of(&pending, &key), Some(101)); + } + + #[test] + fn reinsert_after_failure_loses_no_key_and_merges_a_retouched_key() { + let mut pending = Pending::new(1); + let key = routed(1, 1); + let other = routed(1, 2); + pending.enqueue(100, &[(key, 100), (other, 100)]); + let taken = pending.take_all(); + + pending.enqueue(101, &[(key, 101)]); + pending.reinsert(taken); + + assert_eq!( + cutoff_of(&pending, &key), + Some(101), + "the newer cutoff wins" + ); + assert_eq!(cutoff_of(&pending, &other), Some(100)); + assert!(!pending.is_quiescent()); + } + + #[test] + fn lag_is_zero_when_empty_and_counts_a_drain_in_flight() { + let mut pending = Pending::new(1); + assert_eq!(pending.lag_slots(), 0); + + pending.enqueue(100, &[(routed(1, 1), 100)]); + pending.enqueue(105, &[]); + assert_eq!(pending.lag_slots(), 5); + + let _taken = pending.take_all(); + assert_eq!(pending.lag_slots(), 5, "a drain in flight still counts"); + + pending.finish(); + assert_eq!(pending.lag_slots(), 0); + } + + #[test] + fn lag_does_not_spike_on_a_repaired_slot_below_the_frontier() { + let mut pending = Pending::new(1); + pending.enqueue(1000, &[(routed(1, 1), 1000)]); + // A repaired slot far below the frontier is stamped with the frontier, not its own slot. + pending.enqueue(60, &[(routed(1, 2), 60)]); + assert_eq!(pending.lag_slots(), 0); + } + + #[test] + fn interval_trigger_fires_on_enqueued_count() { + let mut pending = Pending::new(3); + pending.enqueue(100, &[]); + assert!(!pending.should_drain()); + pending.enqueue(101, &[]); + assert!(!pending.should_drain()); + pending.enqueue(102, &[]); + assert!(pending.should_drain()); + } + + #[test] + fn interval_trigger_fires_on_slot_span_after_a_gap_fill() { + let mut pending = Pending::new(10); + pending.enqueue(100, &[]); + pending.note_drain(); + // One call, but the frontier jumped a whole gap. + pending.enqueue(200, &[]); + assert!(pending.should_drain()); + } + + #[test] + fn a_wide_interval_is_not_preempted_by_volume() { + let mut pending = Pending::new(1_000); + for slot in 100..200u64 { + pending.enqueue(slot, &[(routed(1, (slot % 250) as u8), slot)]); + } + pending.note_drain(); + assert!(!pending.should_drain(), "only the interval governs"); + } + + #[test] + fn zero_key_enqueue_raises_high_water_and_counts_a_slot() { + let mut pending = Pending::new(1); + pending.enqueue(100, &[]); + assert!(pending.should_drain()); + assert_eq!(pending.lag_slots(), 0); + assert!(pending.is_quiescent()); + } +} diff --git a/crates/index/src/modules/cleanup/persist.rs b/crates/index/src/modules/cleanup/persist.rs new file mode 100644 index 0000000..1f6f29c --- /dev/null +++ b/crates/index/src/modules/cleanup/persist.rs @@ -0,0 +1,422 @@ +// SPDX-License-Identifier: AGPL-3.0-only +/* + * Copyright 2025-2026 Triton One Limited. All rights reserved. + */ + +//! The cleanup statements. +//! +//! # Ordering across the two tables +//! +//! A closed account's mask lives in `accounts` and the rows it shadows can live in either table. +//! Every read unions both tables in one statement and takes the newest version per pubkey, then +//! drops the ones with no lamports. So a reader sees one Postgres snapshot, and the only state +//! that reads wrong is "mask gone, older positive row still there". +//! +//! [`drain_all`] therefore issues every `snapshot_accounts` statement first and the `accounts` +//! statements only after they succeed. If the first half fails the second never runs, which +//! leaves the mask in place, and a mask in place reads as closed. That is why this needs +//! ordering and not a transaction. +//! +//! This leans on every serving read being a single statement over both tables. That invariant +//! lives in `crates/api/src/db/*.sql` and the indexer's own readers, and it must be rechecked if +//! a reader ever splits into two statements. + +use std::time::Duration; + +use sea_orm::sea_query::ArrayType; +use sea_orm::{ConnectionTrait, DatabaseConnection, DbErr, Statement, Value}; +use tokio::time::{Instant, timeout}; + +use super::pending::{CleanupKey, KeyForm, Taken}; +use crate::metrics; + +const ACCOUNTS_TABLE: &str = "accounts"; +const SNAPSHOT_ACCOUNTS_TABLE: &str = "snapshot_accounts"; + +/// Keys per statement in the startup one-shot, which drains a whole snapshot load at once. +const STARTUP_BATCH_SIZE: usize = 500; + +/// Runs one cleanup statement. Implemented for [`DatabaseConnection`], and for a recording fake +/// in the tests so the ordering rule can be pinned without a database. +pub trait CleanupExecutor: Send + Sync + 'static { + fn execute_cleanup( + &self, + statement: Statement, + ) -> impl std::future::Future> + Send; +} + +impl CleanupExecutor for DatabaseConnection { + async fn execute_cleanup(&self, statement: Statement) -> Result { + self.execute(statement) + .await + .map(|result| result.rows_affected()) + } +} + +/// Runs one drain against both tables, `snapshot_accounts` first. +/// +/// `skip_snapshot` is the startup rule: while the snapshot load is running that table has no +/// indexes and is still taking rows, so its keys go to the one-shot instead. +/// +/// Returns the keys that had no older version in `accounts`. +pub async fn drain_all( + executor: &E, + taken: &Taken, + query_timeout: Duration, + skip_snapshot: bool, +) -> Result { + if !skip_snapshot { + for (form, items) in forms(taken) { + run_statement( + executor, + SNAPSHOT_ACCOUNTS_TABLE, + "cleanup_snapshot_accounts", + form, + items, + query_timeout, + ) + .await?; + } + } + + let mut new_accounts = 0; + for (form, items) in forms(taken) { + let deleted = run_statement( + executor, + ACCOUNTS_TABLE, + "cleanup_accounts", + form, + items, + query_timeout, + ) + .await?; + new_accounts += items.len().saturating_sub(deleted as usize); + } + Ok(new_accounts) +} + +fn forms(taken: &Taken) -> [(KeyForm, &[(CleanupKey, u64)]); 2] { + [ + (KeyForm::Routed, taken.routed.as_slice()), + (KeyForm::Unrouted, taken.unrouted.as_slice()), + ] +} + +/// Deletes every `snapshot_accounts` row of the given pubkeys below one shared cutoff. +/// +/// This is the startup one-shot for the accounts touched while the snapshot was loading. A +/// uniform cutoff is safe on `snapshot_accounts` alone: every row it deletes is either below the +/// first live slot, and so superseded, or has a twin in `accounts` at the same slot. No such twin +/// rule holds for `accounts`, so this cutoff must never be applied there. +/// +/// The error is returned so the caller can leave the node unhealthy rather than declare startup +/// complete over a batch that never ran. +pub async fn delete_below_uniform_cutoff( + executor: &E, + pubkeys: Vec>, + cutoff: u64, + query_timeout: Duration, +) -> Result { + let mut deleted_total = 0; + for chunk in pubkeys.chunks(STARTUP_BATCH_SIZE) { + let items: Vec<(CleanupKey, u64)> = chunk + .iter() + .filter_map(|pubkey| { + Some(( + CleanupKey { + owner: None, + pubkey: pubkey.as_slice().try_into().ok()?, + }, + cutoff, + )) + }) + .collect(); + + deleted_total += run_statement( + executor, + SNAPSHOT_ACCOUNTS_TABLE, + "cleanup_startup_snapshot_accounts", + KeyForm::Unrouted, + &items, + query_timeout, + ) + .await?; + } + Ok(deleted_total) +} + +async fn run_statement( + executor: &E, + table: &str, + origin: &str, + form: KeyForm, + items: &[(CleanupKey, u64)], + query_timeout: Duration, +) -> Result { + if items.is_empty() { + return Ok(0); + } + + let start_time = Instant::now(); + let statement = build_statement(table, form, items); + + let result = timeout(query_timeout, executor.execute_cleanup(statement)) + .await + .unwrap_or_else(|elapsed| { + tracing::error!(target: "finalize_cleanup", "cleanup timeout on {}: {}", table, elapsed); + Err(DbErr::RecordNotInserted) + }); + + metrics::record_finalize_slot(start_time.elapsed().as_secs_f64(), origin); + + match result { + Ok(deleted) => { + metrics::FINALIZE_SLOT_DELETED_ACCOUNTS.observe(deleted as f64); + Ok(deleted) + } + Err(error) => { + tracing::error!( + target: "finalize_cleanup", + "cleanup failed on {} for {} keys: {}", + table, + items.len(), + error + ); + metrics::increment_db_errors(); + Err(error) + } + } +} + +fn build_statement(table: &str, form: KeyForm, items: &[(CleanupKey, u64)]) -> Statement { + let pubkeys = Value::Array( + ArrayType::Bytes, + Some(Box::new( + items + .iter() + .map(|(key, _)| Value::Bytes(Some(Box::new(key.pubkey.to_vec())))) + .collect(), + )), + ); + let cutoffs = Value::Array( + ArrayType::BigInt, + Some(Box::new( + items + .iter() + .map(|(_, cutoff)| Value::BigInt(Some(*cutoff as i64))) + .collect(), + )), + ); + + match form { + KeyForm::Routed => { + let sql = + include_str!("../../db/cleanupWithOwner.sql").replace("accounts_table_name", table); + let owners = Value::Array( + ArrayType::Bytes, + Some(Box::new( + items + .iter() + .map(|(key, _)| { + Value::Bytes(Some(Box::new(key.owner.unwrap_or_default().to_vec()))) + }) + .collect(), + )), + ); + Statement::from_sql_and_values( + sea_orm::DatabaseBackend::Postgres, + sql, + vec![pubkeys, owners, cutoffs], + ) + } + KeyForm::Unrouted => { + let sql = include_str!("../../db/cleanup.sql").replace("accounts_table_name", table); + Statement::from_sql_and_values( + sea_orm::DatabaseBackend::Postgres, + sql, + vec![pubkeys, cutoffs], + ) + } + } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use crate::modules::cleanup::pending::tests::{routed, unrouted}; + use std::sync::Mutex; + + /// Records every statement it is handed, in order, and fails the tables it is told to fail. + #[derive(Default)] + pub(crate) struct RecordingExecutor { + pub(crate) issued: Mutex>, + pub(crate) values: Mutex>, + fail_on: Mutex>, + } + + impl RecordingExecutor { + pub(crate) fn failing(table: &str) -> Self { + Self { + fail_on: Mutex::new(vec![table.to_string()]), + ..Default::default() + } + } + + pub(crate) fn issued(&self) -> Vec { + self.issued.lock().expect("lock").clone() + } + + pub(crate) fn last_values(&self) -> Vec { + self.values.lock().expect("lock").clone() + } + } + + impl CleanupExecutor for RecordingExecutor { + async fn execute_cleanup(&self, statement: Statement) -> Result { + let sql = statement.sql.clone(); + let table = if sql.contains("DELETE FROM snapshot_accounts") { + "snapshot_accounts" + } else { + "accounts" + }; + let form = if sql.contains("k.owner") { + "routed" + } else { + "unrouted" + }; + self.issued + .lock() + .expect("lock") + .push(format!("{table}:{form}")); + self.values + .lock() + .expect("lock") + .push(format!("{:?}", statement.values)); + + if self + .fail_on + .lock() + .expect("lock") + .iter() + .any(|t| t == table) + { + return Err(DbErr::RecordNotInserted); + } + Ok(1) + } + } + + /// Raises the DB error threshold so a test that drives a failing statement does not trip the + /// process exit in `increment_db_errors`. + pub(crate) fn allow_db_errors() { + let _ = crate::metrics::DB_ERRORS_THRESHOLD.set(f64::MAX); + } + + fn taken(routed_keys: Vec<(CleanupKey, u64)>, unrouted_keys: Vec<(CleanupKey, u64)>) -> Taken { + Taken { + routed: routed_keys, + unrouted: unrouted_keys, + oldest_stamp: 0, + } + } + + #[tokio::test] + async fn every_snapshot_statement_runs_before_any_accounts_statement() { + let executor = RecordingExecutor::default(); + let batch = taken(vec![(routed(1, 1), 100)], vec![(unrouted(2), 100)]); + + drain_all(&executor, &batch, Duration::from_secs(5), false) + .await + .expect("succeeds"); + + let issued = executor.issued(); + assert_eq!(issued.len(), 4); + assert!( + issued[..2] + .iter() + .all(|s| s.starts_with("snapshot_accounts")) + ); + assert!(issued[2..].iter().all(|s| s.starts_with("accounts"))); + } + + #[tokio::test] + async fn a_failed_snapshot_statement_stops_the_accounts_statement() { + allow_db_errors(); + let executor = RecordingExecutor::failing("snapshot_accounts"); + let batch = taken(vec![(routed(1, 1), 100)], vec![]); + + let result = drain_all(&executor, &batch, Duration::from_secs(5), false).await; + + assert!(result.is_err()); + assert_eq!( + executor.issued(), + vec!["snapshot_accounts:routed".to_string()], + "the mask must not be deleted while the rows it shadows survive" + ); + } + + #[tokio::test] + async fn the_snapshot_half_is_skipped_during_startup() { + let executor = RecordingExecutor::default(); + let batch = taken(vec![(routed(1, 1), 100)], vec![]); + + drain_all(&executor, &batch, Duration::from_secs(5), true) + .await + .expect("succeeds"); + + assert_eq!(executor.issued(), vec!["accounts:routed".to_string()]); + } + + #[tokio::test] + async fn the_routed_form_binds_owners_and_the_unrouted_form_does_not() { + let executor = RecordingExecutor::default(); + let batch = taken(vec![(routed(1, 1), 100)], vec![(unrouted(2), 200)]); + + drain_all(&executor, &batch, Duration::from_secs(5), true) + .await + .expect("succeeds"); + + let values = executor.last_values(); + assert_eq!( + values[0].matches("Array(").count(), + 3, + "routed binds pubkeys, owners and cutoffs" + ); + assert!(values[0].contains("100")); + assert_eq!( + values[1].matches("Array(").count(), + 2, + "unrouted binds pubkeys and cutoffs" + ); + assert!(values[1].contains("200")); + } + + #[tokio::test] + async fn the_uniform_cutoff_binds_the_same_slot_for_every_key_and_never_touches_accounts() { + let executor = RecordingExecutor::default(); + let pubkeys = vec![vec![1u8; 32], vec![2u8; 32]]; + + delete_below_uniform_cutoff(&executor, pubkeys, 900, Duration::from_secs(5)) + .await + .expect("succeeds"); + + assert_eq!(executor.issued(), vec!["snapshot_accounts:unrouted"]); + assert!(executor.last_values()[0].contains("900")); + } + + #[tokio::test] + async fn the_uniform_cutoff_returns_the_error_so_startup_stays_unhealthy() { + allow_db_errors(); + let executor = RecordingExecutor::failing("snapshot_accounts"); + + let result = delete_below_uniform_cutoff( + &executor, + vec![vec![1u8; 32]], + 900, + Duration::from_secs(5), + ) + .await; + + assert!(result.is_err()); + } +} diff --git a/crates/index/src/modules/finalize_slot.rs b/crates/index/src/modules/finalize_slot.rs index ab04b1a..d5ba1fd 100644 --- a/crates/index/src/modules/finalize_slot.rs +++ b/crates/index/src/modules/finalize_slot.rs @@ -9,16 +9,15 @@ use std::collections::{BTreeSet, HashMap, HashSet}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{Notify, watch}; -use tokio::{task::JoinSet, time::Instant}; +use tokio::time::Instant; use yellowstone_grpc_proto::geyser::CommitmentLevel; use crate::indexer::AccountsReceivedPerBlock; +use crate::modules::cleanup::{self, CleanupHandle}; use crate::modules::health::{HealthReason, ServiceHealth}; use crate::modules::snapshot::SnapshotProcessingState; use crate::{db_queries, metrics}; -const SLOT_FINALIZE_BATCH_SIZE: usize = 500; - /// Emits a warning when the in-memory blocks map grows beyond this size, as an alert for /// further debugging (e.g. finalization stalled or a fork is leaving orphaned entries behind). const BLOCKS_MAP_WARN_THRESHOLD: usize = 500; @@ -92,6 +91,8 @@ pub struct SlotFinalizer { health: ServiceHealth, /// Latest finalized slot, watched by the largest-accounts pruner task. prune_slot_tx: watch::Sender, + /// Queue the slot's cleanup keys go to. The worker never waits on its drainer. + cleanup: CleanupHandle, /// Max number of pending live slots before `note_finalized` blocks (back-pressure bound). /// Bypassed by `enqueue_unbounded` and `enqueue_gap_boundary`(gap fill). pub bound: usize, @@ -105,6 +106,7 @@ impl SlotFinalizer { updated_accounts_during_startup: UpdatedAccountsDuringStartup, health: ServiceHealth, prune_slot_tx: watch::Sender, + cleanup: CleanupHandle, ) -> Self { let bound = config.finalize_slot_buffer_size; let finalizer = Self { @@ -116,6 +118,7 @@ impl SlotFinalizer { updated_accounts_during_startup, health, prune_slot_tx, + cleanup, bound, }; @@ -319,6 +322,7 @@ impl SlotFinalizer { entry.accounts, self.updated_accounts_during_startup.clone(), &self.prune_slot_tx, + &self.cleanup, ) .await; } @@ -413,128 +417,43 @@ async fn finalize_slot( updated_accounts: AccountsReceivedPerBlock, updated_accounts_during_startup: UpdatedAccountsDuringStartup, prune_slot_tx: &watch::Sender, + cleanup: &CleanupHandle, ) { let start_time = Instant::now(); - let db_clone = db.clone(); - let config_clone = config.clone(); - - // Mark the slot as finalized before starting the cleanup tasks for API queries consistency + // Mark the slot finalized before the cleanup keys are queued, for API query consistency. db_queries::insert_slot( slot, updated_accounts.block_time, CommitmentLevel::Finalized, updated_accounts_during_startup.health.is_healthy(), - &db_clone, - &config_clone, + &db, + config, ) .await; - // These are accounts that were in the slot but did not have an older version (which means - // they are completely new to our db) - let new_accounts_in_slot = Arc::new(Mutex::new(0)); - - let batches = updated_accounts - .accounts - .chunks(SLOT_FINALIZE_BATCH_SIZE) - .map(|batch| batch.to_vec()) - .collect::>(); - - let mut join_set = JoinSet::new(); - updated_accounts_during_startup.cleanup_stored_accounts_once(&db, slot, config); - for batch in batches { - let db_clone = db.clone(); - let batch_clone = batch.clone(); - let new_accounts_in_slot_clone = new_accounts_in_slot.clone(); - let updated_accounts_during_startup = updated_accounts_during_startup.clone(); - let config_clone = config.clone(); - join_set.spawn(async move { - let _guard = metrics::TokioTaskCounterGuard::new("finalize_slot_internal"); - - db_queries::cleanup_accounts( - &db_clone, - batch_clone, - slot, - "accounts", - new_accounts_in_slot_clone, - "cleanup_accounts_batch", - &config_clone, - ) - .await; - }); - - // If we are in startup, we just save the updated accounts to delete them after the snapshot is processed - if updated_accounts_during_startup.is_startup() { - updated_accounts_during_startup.add_batch_to_cache_during_startup(batch); - continue; - } - - let db_clone = db.clone(); - let config_clone = config.clone(); - join_set.spawn(async move { - let _guard = metrics::TokioTaskCounterGuard::new("finalize_slot_internal"); - - // with the latest changes it doesn't make sense any more to try to measure this on the snapshot accounts table - // but this will asintotically become more accurate as the snapshot accounts table is deleted/cleaned up - let dummy_new_accounts_in_slot = Arc::new(Mutex::new(0)); - - db_queries::cleanup_accounts( - &db_clone, - batch, - slot, - "snapshot_accounts", - dummy_new_accounts_in_slot, - "cleanup_snapshot_accounts_batch", - &config_clone, - ) - .await; - }); - } - - let _ = prune_slot_tx.send(slot); + let derived = + cleanup::keys_for_slot(&updated_accounts, slot, config.accounts_owner_map_enabled); - let closed_accounts = updated_accounts.closed_accounts.clone(); - let db_clone = db.clone(); - let config_clone = config.clone(); - join_set.spawn(async move { - // Updated accounts doesn't include the closed accounts, instead this query will delete the closed accounts inserted - // and any previous version of the accounts, so it's safe to execute concurrently with the cleanup_accounts tasks - // because there is not overlap between the accounts sets - db_queries::cleanup_closed_accounts(&db_clone, closed_accounts, slot, &config_clone).await; - }); - - // If we are in startup, we just save the closed accounts to delete them after the snapshot is processed + // During startup `snapshot_accounts` is still loading and the owner map is still seeding, so + // a close the map has not learned yet can still have rows there. Cache every closed pubkey, + // not just the routed ones, and let the one-shot clear them once the load finishes. if updated_accounts_during_startup.is_startup() { - updated_accounts_during_startup - .add_batch_to_cache_during_startup(updated_accounts.closed_accounts); - } else { - let config_clone = config.clone(); - join_set.spawn(async move { - // Closed accounts are not included in the updated accounts, so we need to cleanup them separately - db_queries::cleanup_accounts( - &db, - updated_accounts.closed_accounts, - slot, - "snapshot_accounts", - Arc::new(Mutex::new(0)), - "cleanup_snapshot_closed_accounts", - &config_clone, - ) - .await; - }); + updated_accounts_during_startup.add_batch_to_cache_during_startup( + derived + .items + .iter() + .map(|(key, _)| key.pubkey.to_vec()) + .chain(updated_accounts.closed_accounts.iter().cloned()), + ); } - join_set.join_all().await; + let _ = prune_slot_tx.send(slot); + cleanup.enqueue(slot, &derived.items); metrics::record_finalize_slot(start_time.elapsed().as_secs_f64(), "total"); - metrics::record_new_accounts_in_slot( - *new_accounts_in_slot - .lock() - .expect("Failed to lock new_accounts_in_slot"), - "new_accounts_in_slot", - ); } ///Used to store all accounts that are updated/closed while loading the snapshot, and delete them after the snapshot is processed @@ -586,7 +505,10 @@ impl UpdatedAccountsDuringStartup { self.health.remove_reason(HealthReason::Startup).await; } - pub fn add_batch_to_cache_during_startup(&self, batch: Vec>) { + pub fn add_batch_to_cache_during_startup(&self, batch: I) + where + I: IntoIterator>, + { let mut accounts = self.accounts.lock().expect("Failed to lock accounts"); accounts.extend(batch); } @@ -616,7 +538,7 @@ impl UpdatedAccountsDuringStartup { .collect::>(); let db = db.clone(); - let config = config.clone(); + let query_timeout = Duration::from_secs(config.database.finalize_slot_queries_timeout); let snapshot_processing_state = self.snapshot_processing_state.clone(); let health = self.health.clone(); @@ -624,52 +546,38 @@ impl UpdatedAccountsDuringStartup { let _guard = metrics::TokioTaskCounterGuard::new("startup_snapshot_accounts_cleanup"); let start_time = Instant::now(); + let total = accounts.len(); + tracing::info!(target: "cleanup_stored_accounts", "Cleaning up stored accounts from snapshot_accounts - accounts: {}", total); - tracing::info!(target: "cleanup_stored_accounts", "Cleaning up stored accounts from snapshot_accounts - accounts: {}", accounts.len()); - - let batches = accounts - .chunks(SLOT_FINALIZE_BATCH_SIZE) - .map(|batch| batch.to_vec()) - .collect::>(); - - let mut join_set = JoinSet::new(); - const MAX_CONCURRENT_CLEANUP_TASKS: usize = 10; - - for batch in batches { - while join_set.len() >= MAX_CONCURRENT_CLEANUP_TASKS { - join_set.join_next().await; - } - - let db = db.clone(); - let config_clone = config.clone(); - join_set.spawn(async move { - let _guard = - metrics::TokioTaskCounterGuard::new("startup_snapshot_accounts_cleanup"); - - db_queries::cleanup_accounts( - &db, - batch, - slot, - "snapshot_accounts", - Arc::new(Mutex::new(0)), - "cleanup_startup_snapshot_accounts_batch", - &config_clone, - ) + // A uniform cutoff is safe on this table only. See delete_below_uniform_cutoff. + let result = + cleanup::persist::delete_below_uniform_cutoff(&db, accounts, slot, query_timeout) .await; - }); - } - - join_set.join_all().await; let elapsed = start_time.elapsed().as_secs_f64(); - tracing::info!(target: "cleanup_stored_accounts", "Cleaned up stored accounts from snapshot_accounts in {} seconds", elapsed); - - // Startup snapshot processing is complete: clear the startup unhealthy reason. - *snapshot_processing_state - .lock() - .expect("Failed to lock snapshot_processing_state") = - SnapshotProcessingState::FinishedAndCleanedUp; - health.remove_reason(HealthReason::Startup).await; + match result { + Ok(deleted) => { + tracing::info!(target: "cleanup_stored_accounts", "Cleaned up {} rows from snapshot_accounts in {} seconds", deleted, elapsed); + + // Startup is complete only when every batch landed. A batch that never ran + // leaves rows a closed account still needs removed, so the node stays + // unhealthy instead of declaring itself ready over them. + *snapshot_processing_state + .lock() + .expect("Failed to lock snapshot_processing_state") = + SnapshotProcessingState::FinishedAndCleanedUp; + health.remove_reason(HealthReason::Startup).await; + } + Err(error) => { + tracing::error!( + target: "cleanup_stored_accounts", + "startup snapshot_accounts cleanup failed after {} seconds, leaving the node unhealthy: {}", + elapsed, + error + ); + metrics::increment_db_errors(); + } + } }); } } diff --git a/crates/index/src/modules/mod.rs b/crates/index/src/modules/mod.rs index 16ba93a..20af2ac 100644 --- a/crates/index/src/modules/mod.rs +++ b/crates/index/src/modules/mod.rs @@ -3,6 +3,7 @@ * Copyright 2025-2026 Triton One Limited. All rights reserved. */ +pub mod cleanup; pub mod epoch_stakes; pub mod finalize_slot; pub mod grpc; diff --git a/crates/index/src/modules/save_block.rs b/crates/index/src/modules/save_block.rs index 5645d58..a811de6 100644 --- a/crates/index/src/modules/save_block.rs +++ b/crates/index/src/modules/save_block.rs @@ -64,7 +64,11 @@ pub async fn save_block( let mut current_chunk_bytes = 0; let mut updated_accounts_for_slot = Vec::new(); + let mut updated_accounts_owners_for_slot = Vec::new(); let mut closed_accounts_for_slot = Vec::new(); + // Owners are only captured when the owner map is enabled; empty owners make the + // finalize cleanup fall back to the no-owner SQL. + let capture_owners = accounts_owner_map.is_enabled(); metrics::record_new_accounts_in_slot(block.accounts.len(), "block_accounts_total"); @@ -145,6 +149,9 @@ pub async fn save_block( current_chunk_bytes += account.data.len(); updated_accounts_for_slot.push(account.pubkey.clone()); + if capture_owners { + updated_accounts_owners_for_slot.push(account.owner.clone()); + } current_chunk.push(accounts::ActiveModel { pubkey: Set(account.pubkey), @@ -178,6 +185,14 @@ pub async fn save_block( let closed_account_for_slot_len = closed_accounts_for_slot.len(); + // Captured before `save_closed_accounts` drains the map entries and changed-owner + // record for this slot; at finalize time the old owners are unrecoverable. + let (closed_cleanup_pubkeys, closed_cleanup_owners) = if capture_owners { + accounts_owner_map.closed_cleanup_pairs(&closed_accounts_for_slot, slot) + } else { + (Vec::new(), Vec::new()) + }; + // We delay the closed accounts insertion until the snapshot is processed to avoid reads while // the `snapshot_accounts` table still doesn't have indexes let snapshot_processing_state: SnapshotProcessingState = { @@ -198,7 +213,10 @@ pub async fn save_block( AccountsReceivedPerBlock { block_time: block.block_time, accounts: updated_accounts_for_slot, + accounts_owners: updated_accounts_owners_for_slot, closed_accounts: closed_accounts_for_slot, + closed_cleanup_pubkeys, + closed_cleanup_owners, }, block.blockhash.clone(), block.parent_slot, diff --git a/crates/query-tracker/src/modules/creation.rs b/crates/query-tracker/src/modules/creation.rs index bac1632..ac940e5 100644 --- a/crates/query-tracker/src/modules/creation.rs +++ b/crates/query-tracker/src/modules/creation.rs @@ -143,6 +143,7 @@ pub async fn run(store: Store, config: QueryTrackerConfig) { if indexer_backpressure::is_under_pressure( &config.indexer_metrics, config.indexer_metrics_threshold, + config.indexer_cleanup_lag_threshold, ) .await { diff --git a/crates/query-tracker/src/modules/eviction.rs b/crates/query-tracker/src/modules/eviction.rs index ee95925..259c6f5 100644 --- a/crates/query-tracker/src/modules/eviction.rs +++ b/crates/query-tracker/src/modules/eviction.rs @@ -79,6 +79,7 @@ async fn wait_out_backpressure(config: &QueryTrackerConfig) -> bool { if !indexer_backpressure::is_under_pressure( &config.indexer_metrics, config.indexer_metrics_threshold, + config.indexer_cleanup_lag_threshold, ) .await { @@ -95,6 +96,7 @@ async fn wait_out_backpressure(config: &QueryTrackerConfig) -> bool { if !indexer_backpressure::is_under_pressure( &config.indexer_metrics, config.indexer_metrics_threshold, + config.indexer_cleanup_lag_threshold, ) .await { diff --git a/crates/query-tracker/src/modules/indexer_backpressure.rs b/crates/query-tracker/src/modules/indexer_backpressure.rs index 191765a..ab899c9 100644 --- a/crates/query-tracker/src/modules/indexer_backpressure.rs +++ b/crates/query-tracker/src/modules/indexer_backpressure.rs @@ -6,17 +6,35 @@ //! Indexer backpressure — use the indexer metrics to decide how busy is the DB. //! //! CREATE INDEX and DROP INDEX both take heavy locks on the hot `accounts` / -//! `snapshot_accounts` tables. When the indexer is behind (its -//! `cloudbreak_finalize_slot_handler_queue_size` is high) we defer DDL so we do +//! `snapshot_accounts` tables. When the indexer is behind we defer DDL so we do //! not make ingest lag worse. Both the creation loop and the eviction pass gate //! on this. +//! +//! Two gauges say the indexer is behind, and either one defers. +//! `cloudbreak_finalize_slot_handler_queue_size` is the finalize backlog, which a gap-fill pause +//! still builds. `cloudbreak_cleanup_lag_slots` is the cleanup backlog, which is the signal that +//! database pressure shows up in once cleanup runs off the finalize worker. Reading only the +//! first would report "safe" while the cleanup drainer is drowning. +//! +//! The cleanup gauge is optional: an indexer that does not publish it reads as no pressure, so +//! this is inert against a build without a cleanup drainer. The finalize gauge is not optional. +//! A body that does not carry it is not a healthy indexer answering, it is the wrong endpoint or +//! one that has not registered its collectors, and that defers. An endpoint that cannot be read +//! at all defers too. use tracing::{debug, error}; -/// Scrape the indexer's Prometheus endpoint for the finalize-slot queue size. +/// The indexer gauges that gate DDL. `None` means the indexer does not publish that gauge. +#[derive(Debug, Default, Clone, Copy)] +pub struct IndexerPressure { + pub finalize_queue: Option, + pub cleanup_lag: Option, +} + +/// Scrape the indexer's Prometheus endpoint for both backpressure gauges. /// Returns `None` on any transport/parse failure (caller treats that as /// "cannot confirm safe" and defers). -pub async fn read_indexer_queue_size(metrics_url: &str) -> Option { +pub async fn read_indexer_pressure(metrics_url: &str) -> Option { let client = reqwest::Client::new(); let body = client .get(metrics_url) @@ -27,37 +45,119 @@ pub async fn read_indexer_queue_size(metrics_url: &str) -> Option { .await .ok()?; + Some(parse_indexer_pressure(&body)) +} + +/// Reads both gauges out of a Prometheus exposition body. +fn parse_indexer_pressure(body: &str) -> IndexerPressure { + let mut pressure = IndexerPressure::default(); for line in body.lines() { - if line.starts_with("cloudbreak_finalize_slot_handler_queue_size") { - let value = line - .split_whitespace() - .last() - .and_then(|v| v.parse::().ok()); - debug!(target: "query_tracker_backpressure", "indexer queue size: {value:?}"); - return value; + if line.starts_with('#') { + continue; + } + if let Some(value) = gauge_value(line, "cloudbreak_finalize_slot_handler_queue_size") { + pressure.finalize_queue = Some(value); + } else if let Some(value) = gauge_value(line, "cloudbreak_cleanup_lag_slots") { + pressure.cleanup_lag = Some(value); } } - None + debug!(target: "query_tracker_backpressure", "indexer pressure: {pressure:?}"); + pressure } -/// `true` when DDL(CREATE/DROP INDEX) should be deferred: the indexer queue is above `threshold`, -/// or we could not read it at all. -pub async fn is_under_pressure(metrics_url: &str, threshold: u64) -> bool { - match read_indexer_queue_size(metrics_url).await { - Some(size) if size > threshold => { - debug!( - target: "query_tracker_backpressure", - "indexer queue {size} > threshold {threshold}; deferring DDL" - ); - true - } - Some(_) => false, - None => { - error!( - target: "query_tracker_backpressure", - "failed to read indexer metrics at {metrics_url}; deferring DDL" - ); - true - } +/// Parses ` `, taking a negative gauge as zero. +fn gauge_value(line: &str, name: &str) -> Option { + let rest = line.strip_prefix(name)?; + if !rest.starts_with(' ') { + return None; + } + let value: f64 = rest.split_whitespace().last()?.parse().ok()?; + Some(if value < 0.0 { 0 } else { value as u64 }) +} + +/// `true` when DDL(CREATE/DROP INDEX) should be deferred: either indexer gauge is above its +/// threshold, or the endpoint could not be read at all. +/// +/// A gauge the indexer does not publish is not pressure. That keeps this inert against an +/// indexer build that has no cleanup drainer. +pub async fn is_under_pressure( + metrics_url: &str, + queue_threshold: u64, + cleanup_lag_threshold: u64, +) -> bool { + let Some(pressure) = read_indexer_pressure(metrics_url).await else { + error!( + target: "query_tracker_backpressure", + "failed to read indexer metrics at {metrics_url}; deferring DDL" + ); + return true; + }; + + let Some(finalize_queue) = pressure.finalize_queue else { + error!( + target: "query_tracker_backpressure", + "indexer metrics at {metrics_url} carry no finalize queue gauge; deferring DDL" + ); + return true; + }; + + if finalize_queue > queue_threshold { + debug!( + target: "query_tracker_backpressure", + "indexer finalize queue {finalize_queue} > threshold {queue_threshold}; deferring DDL" + ); + return true; + } + if pressure + .cleanup_lag + .is_some_and(|lag| lag > cleanup_lag_threshold) + { + debug!( + target: "query_tracker_backpressure", + "indexer cleanup lag {:?} > threshold {cleanup_lag_threshold}; deferring DDL", + pressure.cleanup_lag + ); + return true; + } + false +} + +#[cfg(test)] +mod tests { + use super::*; + + const BODY: &str = "# HELP cloudbreak_finalize_slot_handler_queue_size q\ncloudbreak_finalize_slot_handler_queue_size 3\ncloudbreak_cleanup_lag_slots 7\n"; + + #[test] + fn reads_both_gauges() { + let pressure = parse_indexer_pressure(BODY); + assert_eq!(pressure.finalize_queue, Some(3)); + assert_eq!(pressure.cleanup_lag, Some(7)); + } + + #[test] + fn an_absent_cleanup_gauge_is_not_pressure() { + let pressure = parse_indexer_pressure("cloudbreak_finalize_slot_handler_queue_size 1\n"); + assert_eq!(pressure.finalize_queue, Some(1)); + assert_eq!(pressure.cleanup_lag, None); + } + + #[test] + fn an_absent_finalize_gauge_is_pressure() { + // The finalize gauge is not optional: a body without it is not a healthy indexer. + let pressure = parse_indexer_pressure("cloudbreak_cleanup_lag_slots 0\n"); + assert_eq!(pressure.finalize_queue, None); + } + + #[test] + fn a_comment_line_is_not_a_sample() { + let pressure = parse_indexer_pressure("# TYPE cloudbreak_cleanup_lag_slots gauge\n"); + assert_eq!(pressure.cleanup_lag, None); + } + + #[test] + fn a_negative_gauge_reads_as_zero() { + let pressure = parse_indexer_pressure("cloudbreak_cleanup_lag_slots -1\n"); + assert_eq!(pressure.cleanup_lag, Some(0)); } } diff --git a/example.cloudbreak.index.toml b/example.cloudbreak.index.toml index c0ab9d0..bd726c6 100644 --- a/example.cloudbreak.index.toml +++ b/example.cloudbreak.index.toml @@ -1,6 +1,9 @@ # The buffer size for queuing finalize slot events finalize-slot-buffer-size = 1000 accounts-owner-map-enabled = false +# Finalized slots between cleanup drains. 1 drains every slot. Higher values coalesce repeat +# touches into one delete and retain more account versions for reads to scan past. +cleanup-interval-slots = 1 # Maintains the running total-supply figure served by getSupply from a bounded # hot-accounts cache. Requires an empty [programs] filter, diff --git a/example.cloudbreak.query-tracker.toml b/example.cloudbreak.query-tracker.toml index f5ff753..360b8f7 100644 --- a/example.cloudbreak.query-tracker.toml +++ b/example.cloudbreak.query-tracker.toml @@ -49,6 +49,10 @@ excluded-programs = [ # --- backpressure --- indexer-metrics = "localhost:8875" indexer-metrics-threshold = 5 +# Defers CREATE and DROP INDEX when the indexer's cleanup backlog exceeds this many slots. +# The finalize queue stops reflecting database pressure once cleanup runs off the finalize +# worker, so this is the second signal. An indexer that does not publish it reads as no pressure. +indexer-cleanup-lag-threshold = 32 max-auto-indexes = 100 # --- eviction (usage-based; off by default) ---