Owner-route finalize slot account cleanup

README.md

@@ -731,6 +731,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.

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/cleanupClosedWithOwner.sql

@@ -0,0 +1,12 @@
+-- SPDX-License-Identifier: AGPL-3.0-only
+/*
+ * Copyright 2025-2026 Triton One Limited. All rights reserved.
+ */
+
+-- Owner-routed variant of closedAccountscleanup.sql. Inclusive slot bound so the
+-- closed-account mask row inserted at the finalized slot is deleted too.
+DELETE FROM accounts AS a
+USING unnest($2::bytea[], $3::bytea[]) AS k(pubkey, owner)
+WHERE a.owner = k.owner
+  AND a.pubkey = k.pubkey
+  AND a.slot <= $1;  -- $1 is finalized_slot

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 variant of cleanup.sql: the owner in each (pubkey, owner) pair lets
+-- Postgres prune the DELETE to a single hash partition instead of scanning all 64.
+DELETE FROM accounts_table_name AS a  -- placeholder to be replaced with the actual table name
+USING unnest($2::bytea[], $3::bytea[]) AS k(pubkey, owner)
+WHERE a.owner = k.owner
+  AND a.pubkey = k.pubkey
+  AND a.slot < $1;  -- $1 is finalized_slot

crates/index/src/db_queries.rs

@@ -144,36 +144,62 @@ pub fn insert_closed_accounts(
 }
 
 /// Deletes the last special "closed" version inserted for the set of closed accounts for the given slot
+///
+/// When `owners` pairs up with `pubkeys` the delete is owner-routed so Postgres prunes to the
+/// matching hash partitions; an empty `owners` uses the no-owner SQL (owner map disabled).
 pub async fn cleanup_closed_accounts(
     db: &DatabaseConnection,
     pubkeys: Vec<Vec<u8>>,
+    owners: 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,
+    let owner_routed = !owners.is_empty() && owners.len() == pubkeys.len();
+    let cleanup_closed_accounts_sql = if owner_routed {
+        include_str!("db/cleanupClosedWithOwner.sql")
+    } else {
+        include_str!("db/closedAccountscleanup.sql")
+    };
+
+    let pubkeys_value = Value::Array(
+        sea_orm::sea_query::ArrayType::Bytes,
+        Some(Box::new(
+            pubkeys
+                .into_iter()
+                .map(|pubkey| Value::Bytes(Some(Box::new(pubkey))))
+                .collect(),
+        )),
+    );
+    let params = if owner_routed {
         vec![
+            Value::BigInt(Some(slot as i64)),
+            pubkeys_value,
             Value::Array(
                 sea_orm::sea_query::ArrayType::Bytes,
                 Some(Box::new(
-                    pubkeys
+                    owners
                         .into_iter()
-                        .map(|pubkey| Value::Bytes(Some(Box::new(pubkey))))
+                        .map(|owner| Value::Bytes(Some(Box::new(owner))))
                         .collect(),
                 )),
             ),
-            Value::BigInt(Some(slot as i64)),
-        ],
+        ]
+    } else {
+        vec![pubkeys_value, Value::BigInt(Some(slot as i64))]
+    };
+
+    let query = db.execute(Statement::from_sql_and_values(
+        sea_orm::DatabaseBackend::Postgres,
+        cleanup_closed_accounts_sql,
+        params,
     ));
 
     let result = timeout(query_timeout, query)
@@ -210,9 +236,13 @@ pub async fn cleanup_closed_accounts(
 }
 
 /// Cleans up older versions (slot less than the received slot) of the accounts from the database (for the given table)
+///
+/// When `owners` pairs up with `pubkeys` the delete is owner-routed so Postgres prunes to the
+/// matching hash partitions; an empty `owners` uses the no-owner SQL (owner map disabled).
 pub async fn cleanup_accounts(
     db: &DatabaseConnection,
     pubkeys: Vec<Vec<u8>>,
+    owners: Vec<Vec<u8>>,
     slot: u64,
     table_name: &str,
     new_accounts_in_slot: Arc<Mutex<usize>>,
@@ -221,25 +251,43 @@ pub async fn cleanup_accounts(
 ) {
     let start_time = Instant::now();
     let pubkeys_len = pubkeys.len();
-    let cleanup_sql = include_str!("db/cleanup.sql");
+    let owner_routed = !owners.is_empty() && owners.len() == pubkeys.len();
+    let cleanup_sql = if owner_routed {
+        include_str!("db/cleanupWithOwner.sql")
+    } else {
+        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 mut params = 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(),
+            )),
+        ),
+    ];
+    if owner_routed {
+        params.push(Value::Array(
+            sea_orm::sea_query::ArrayType::Bytes,
+            Some(Box::new(
+                owners
+                    .into_iter()
+                    .map(|owner| Value::Bytes(Some(Box::new(owner))))
+                    .collect(),
+            )),
+        ));
+    }
+
     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(),
-                )),
-            ),
-        ],
+        params,
     ));
 
     let result = timeout(query_timeout, query)

crates/index/src/indexer.rs

@@ -314,5 +314,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/modules/finalize_slot.rs

@@ -434,19 +434,36 @@ async fn finalize_slot(
     //  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::<Vec<_>>();
+    // Each batch carries the matching owners chunk so the cleanup DELETE prunes to one hash
+    // partition per key. Owners are empty when the owner map is disabled (no-owner fallback).
+    let owner_routed = updated_accounts.accounts_owners.len() == updated_accounts.accounts.len();
+    let batches = if owner_routed {
+        updated_accounts
+            .accounts
+            .chunks(SLOT_FINALIZE_BATCH_SIZE)
+            .zip(
+                updated_accounts
+                    .accounts_owners
+                    .chunks(SLOT_FINALIZE_BATCH_SIZE),
+            )
+            .map(|(batch, owners)| (batch.to_vec(), owners.to_vec()))
+            .collect::<Vec<_>>()
+    } else {
+        updated_accounts
+            .accounts
+            .chunks(SLOT_FINALIZE_BATCH_SIZE)
+            .map(|batch| (batch.to_vec(), Vec::new()))
+            .collect::<Vec<_>>()
+    };
 
     let mut join_set = JoinSet::new();
 
     updated_accounts_during_startup.cleanup_stored_accounts_once(&db, slot, config);
 
-    for batch in batches {
+    for (batch, owner_batch) in batches {
         let db_clone = db.clone();
         let batch_clone = batch.clone();
+        let owner_batch_clone = owner_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();
@@ -456,6 +473,7 @@ async fn finalize_slot(
             db_queries::cleanup_accounts(
                 &db_clone,
                 batch_clone,
+                owner_batch_clone,
                 slot,
                 "accounts",
                 new_accounts_in_slot_clone,
@@ -483,6 +501,7 @@ async fn finalize_slot(
             db_queries::cleanup_accounts(
                 &db_clone,
                 batch,
+                owner_batch,
                 slot,
                 "snapshot_accounts",
                 dummy_new_accounts_in_slot,
@@ -495,14 +514,36 @@ async fn finalize_slot(
 
     let _ = prune_slot_tx.send(slot);
 
-    let closed_accounts = updated_accounts.closed_accounts.clone();
+    // The captured pairs also carry the old-owner keys for in-slot owner changes, so an
+    // account that moved from owner A to B has its A-partition rows and mask deleted under A.
+    let closed_owner_routed = !updated_accounts.closed_cleanup_owners.is_empty()
+        && updated_accounts.closed_cleanup_owners.len()
+            == updated_accounts.closed_cleanup_pubkeys.len();
+
+    let (closed_pubkeys, closed_owners) = if closed_owner_routed {
+        (
+            updated_accounts.closed_cleanup_pubkeys.clone(),
+            updated_accounts.closed_cleanup_owners.clone(),
+        )
+    } else {
+        (updated_accounts.closed_accounts.clone(), Vec::new())
+    };
+
     let db_clone = db.clone();
     let config_clone = config.clone();
+    let closed_pubkeys_clone = closed_pubkeys.clone();
+    let closed_owners_clone = closed_owners.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;
+        // Deletes the closed-account masks and their old versions, including old-owner rows for owner
+        // changes. Runs concurrently with cleanup_accounts because the owner match keeps the row sets disjoint.
+        db_queries::cleanup_closed_accounts(
+            &db_clone,
+            closed_pubkeys_clone,
+            closed_owners_clone,
+            slot,
+            &config_clone,
+        )
+        .await;
     });
 
     // If we are in startup, we just save the closed accounts to delete them after the snapshot is processed
@@ -515,7 +556,8 @@ async fn finalize_slot(
             // 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,
+                closed_pubkeys,
+                closed_owners,
                 slot,
                 "snapshot_accounts",
                 Arc::new(Mutex::new(0)),
@@ -649,6 +691,7 @@ impl UpdatedAccountsDuringStartup {
                     db_queries::cleanup_accounts(
                         &db,
                         batch,
+                        Vec::new(),
                         slot,
                         "snapshot_accounts",
                         Arc::new(Mutex::new(0)),

crates/index/src/modules/save_block.rs

@@ -63,7 +63,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");
 
@@ -138,6 +142,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),
@@ -171,6 +178,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 = {
@@ -202,7 +217,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,