feat(cat-gateway): Purge volatile data after immutable roll forward (#…

…1188)

* wip(cat-gateway): add select and delete queries for purging volatile data

* wip(cat-gateway): refactor purge queries

* feat(cat-gateway): refactor session execution code for reuse

* feat(cat-gateway): add purge_queries to CassandraSession

* fix(cat-gateway): refactor into queries::purge module

* wip(cat-gateway): add purge methods to cassandra session

* wip(cat-gateway): rust code for txo by stake queries

* wip(cat-gateway): refactor queries for purging volatile data

* wip(cat-gateway): refactor queries for purging volatile data

* fix(cat-gateway): modify types for query result

* fix(cat-gateway): cleanup naming types

* chore(cat-gateway): fix spelling

* chore(cat-gateway): fix spelling

* wip(cat-gateway): rust code for txo assets by stake queries

* fix(cat-gateway): cleanup field names

* fix(cat-gateway): incorporate unstaked txo queries, cleanup cql comments

* fix(cat-gateway): incorporate unstaked txi queries, cleanup cql comments

* fix(cat-gateway): incorporate stake registration queries, cleanup cql comments

* fix(cat-gateway): incorporate cip36 registration queries

* fix(cat-gateway): incorporate cip36 registration for votekey queries

* chore(cat-gateway): fix spelling

* fix(cat-gateway): incorporate cip36 registration for votekey queries

* fix(cat-gateway): incorporate cip36 invalid registration queries

* fix(cat-gateway): incorporate rbac509 registration queries

* fix(cat-gateway): incorporate chain root for role0 registration queries

* chore(cat-gateway): fix CQL comments

* fix(cat-gateway): incorporate chain root for stake addr and txn id registration queries

* fix(cat-gateway): box large futures lint

* feat(cat-gateway): add PURGE_SLOT_BUFFER parameter

* fix(cat-gateway): update scylla type usage

* fix(cat-gateway): fix prepared query variants

* wip(cat-gateway): add purge command when only one chain follower is left

* feat(cat-gateway): purge roll forward volatile data

* fix(cat-gateway): correct names for chain root queries

* feat(cat-gateway): wrap up queries to purge volatile data

* fix(cat-gateway): cleanup

* fix(cat-gateway): update cardano-chain-follower to v0.0.6

* test(rm schema check): temp

* revert schema_version_check

* revert `/health/live` endpoint

* fix purge queris

* remain comment

* fix

---------

Co-authored-by: Steven Johnson <[email protected]>
Co-authored-by: cong-or <[email protected]>
Co-authored-by: Mr-Leshiy <[email protected]>

catalyst-gateway/bin/src/cardano/mod.rs

@@ -12,7 +12,7 @@ use tracing::{debug, error, info, warn};
 
 use crate::{
     db::index::{
-        block::index_block,
+        block::{index_block, roll_forward},
         queries::sync_status::{
             get::{get_sync_status, SyncStatus},
             update::update_sync_status,
@@ -453,14 +453,19 @@ impl SyncTask {
                 },
             }
 
-            // TODO: IF there is only 1 chain follower left in sync_tasks, then all
+            // IF there is only 1 chain follower left in sync_tasks, then all
             // immutable followers have finished.
             // When this happens we need to purge the live index of any records that exist
             // before the current immutable tip.
             // Note: to prevent a data race when multiple nodes are syncing, we probably
             // want to put a gap in this, so that there are X slots of overlap
             // between the live chain and immutable chain.  This gap should be
             // a parameter.
+            if self.sync_tasks.len() == 1 {
+                if let Err(error) = roll_forward::purge_live_index(self.immutable_tip_slot).await {
+                    error!(chain=%self.cfg.chain, error=%error, "BUG: Purging volatile data task failed.");
+                }
+            }
         }
 
         error!(chain=%self.cfg.chain,"BUG: Sync tasks have all stopped.  This is an unexpected error!");

catalyst-gateway/bin/src/db/index/block/mod.rs

@@ -4,6 +4,7 @@
 pub(crate) mod certs;
 pub(crate) mod cip36;
 pub(crate) mod rbac509;
+pub(crate) mod roll_forward;
 pub(crate) mod txi;
 pub(crate) mod txo;
 

catalyst-gateway/bin/src/db/index/block/roll_forward.rs

@@ -0,0 +1,314 @@
+//! Immutable Roll Forward logic.
+
+use std::{collections::HashSet, sync::Arc};
+
+use futures::StreamExt;
+
+use crate::{
+    db::index::{block::CassandraSession, queries::purge},
+    settings::Settings,
+};
+
+/// Purge Data from Live Index
+pub(crate) async fn purge_live_index(purge_slot: u64) -> anyhow::Result<()> {
+    let persistent = false; // get volatile session
+    let Some(session) = CassandraSession::get(persistent) else {
+        anyhow::bail!("Failed to acquire db session");
+    };
+
+    // Purge data up to this slot
+    let purge_to_slot: num_bigint::BigInt = purge_slot
+        .saturating_sub(Settings::purge_slot_buffer())
+        .into();
+
+    let txn_hashes = purge_txi_by_hash(&session, &purge_to_slot).await?;
+    purge_chain_root_for_role0_key(&session, &purge_to_slot).await?;
+    purge_chain_root_for_stake_address(&session, &purge_to_slot).await?;
+    purge_chain_root_for_txn_id(&session, &txn_hashes).await?;
+    purge_cip36_registration(&session, &purge_to_slot).await?;
+    purge_cip36_registration_for_vote_key(&session, &purge_to_slot).await?;
+    purge_cip36_registration_invalid(&session, &purge_to_slot).await?;
+    purge_rbac509_registration(&session, &purge_to_slot).await?;
+    purge_stake_registration(&session, &purge_to_slot).await?;
+    purge_txo_ada(&session, &purge_to_slot).await?;
+    purge_txo_assets(&session, &purge_to_slot).await?;
+    purge_unstaked_txo_ada(&session, &purge_to_slot).await?;
+    purge_unstaked_txo_assets(&session, &purge_to_slot).await?;
+
+    Ok(())
+}
+
+/// Purge data from `chain_root_for_role0_key`.
+async fn purge_chain_root_for_role0_key(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::chain_root_for_role0_key::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `chain_root_for_stake_address`.
+async fn purge_chain_root_for_stake_address(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::chain_root_for_stake_address::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `chain_root_for_txn_id`.
+async fn purge_chain_root_for_txn_id(
+    session: &Arc<CassandraSession>, txn_hashes: &HashSet<Vec<u8>>,
+) -> anyhow::Result<()> {
+    use purge::chain_root_for_txn_id::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if txn_hashes.contains(&params.transaction_id) {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `cip36_registration`.
+async fn purge_cip36_registration(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::cip36_registration::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `cip36_registration_for_vote_key`.
+async fn purge_cip36_registration_for_vote_key(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::cip36_registration_for_vote_key::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `cip36_registration_invalid`.
+async fn purge_cip36_registration_invalid(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::cip36_registration_invalid::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `rbac509_registration`.
+async fn purge_rbac509_registration(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::rbac509_registration::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `stake_registration`.
+async fn purge_stake_registration(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::stake_registration::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `txi_by_hash`.
+async fn purge_txi_by_hash(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<HashSet<Vec<u8>>> {
+    use purge::txi_by_hash::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    let mut txn_hashes: HashSet<Vec<u8>> = HashSet::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        if &primary_key.2 <= purge_to_slot {
+            let params: Params = primary_key.into();
+            txn_hashes.insert(params.txn_hash.clone());
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(txn_hashes)
+}
+
+/// Purge data from `txo_ada`.
+async fn purge_txo_ada(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::txo_ada::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `txo_assets`.
+async fn purge_txo_assets(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::txo_assets::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `unstaked_txo_ada`.
+async fn purge_unstaked_txo_ada(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::unstaked_txo_ada::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        if &primary_key.2 <= purge_to_slot {
+            let params: Params = primary_key.into();
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}
+
+/// Purge data from `unstaked_txo_assets`.
+async fn purge_unstaked_txo_assets(
+    session: &Arc<CassandraSession>, purge_to_slot: &num_bigint::BigInt,
+) -> anyhow::Result<()> {
+    use purge::unstaked_txo_assets::{DeleteQuery, Params, PrimaryKeyQuery};
+
+    // Get all keys
+    let mut primary_keys_stream = PrimaryKeyQuery::execute(session).await?;
+    // Filter
+    let mut delete_params: Vec<Params> = Vec::new();
+    while let Some(Ok(primary_key)) = primary_keys_stream.next().await {
+        let params: Params = primary_key.into();
+        if &params.slot_no <= purge_to_slot {
+            delete_params.push(params);
+        }
+    }
+    // Delete filtered keys
+    DeleteQuery::execute(session, delete_params).await?;
+    Ok(())
+}