Owner routed deferred cleanup

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<LargestAccountsConfig>,
+    /// 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<u64> {
@@ -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<usize>,
@@ -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<String, D::Error>
     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(),

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<u8>],
+        slot: u64,
+    ) -> (Vec<Vec<u8>>, Vec<Vec<u8>>) {
+        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.
     ///

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

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

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<Vec<u8>>,
-    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<Vec<u8>>,
-    slot: u64,
-    table_name: &str,
-    new_accounts_in_slot: Arc<Mutex<usize>>,
-    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<UnixTimestamp>,

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<UnixTimestamp>,
     pub accounts: Vec<Vec<u8>>,
+    /// 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<Vec<u8>>,
     pub closed_accounts: Vec<Vec<u8>>,
+    /// (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<Vec<u8>>,
+    pub closed_cleanup_owners: Vec<Vec<u8>>,
 }

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);