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

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,
@@ -451,12 +451,21 @@ impl SyncTask {
 
             // TODO: 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.
+            // 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.
+
+            // WIP: How do we check that only one follower is left, and that the slot buffer
+            // (overlap) criteria is met?
+            let only_one_chain_follower_left = false;
+            if only_one_chain_follower_left {
+                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,317 @@
+//! 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
+#[allow(unused_variables)]
+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`.
+#[allow(clippy::unused_async)]
+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`.
+#[allow(clippy::unused_async)]
+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(())
+}