feat(cat-gateway): Chain sync V2 - part 2

catalyst-gateway/Justfile

@@ -22,19 +22,19 @@ code-format:
 
 # Lint the rust code
 code-lint:
-    cargo lintfix
-    cargo lint
+    cargo lintfix -r
+    cargo lint -r
 
 # Synchronize Rust Configs
 sync-cfg:
-    earthly +sync-cfg
+    cd .. && earthly ./catalyst-gateway+sync-cfg
 
 # Pre Push Checks
 pre-push: sync-cfg code-format code-lint license-check
     # Make sure we can actually build inside Earthly which needs to happen in CI.
-    earthly +check
-    earthly +build
-    earthly +package-cat-gateway
+    cd .. && earthly ./catalyst-gateway+check
+    cd .. && earthly ./catalyst-gateway+build
+    cd .. && earthly ./catalyst-gateway+package-cat-gateway
 
 # Build Local release build of catalyst gateway
 build-cat-gateway: code-format code-lint
@@ -46,10 +46,10 @@ run-cat-gateway: build-cat-gateway
     CHAIN_FOLLOWER_SYNC_TASKS="16" \
     RUST_LOG="error,cat-gateway=debug,cardano_chain_follower=debug,mithril-client=debug" \
     CHAIN_NETWORK="Preprod" \
-    ./catalyst-gateway/target/release/cat-gateway run --log-level debug
+    ./target/release/cat-gateway run --log-level debug
 
 # Run cat-gateway natively on mainnet
 run-cat-gateway-mainnet: build-cat-gateway
     CHAIN_FOLLOWER_SYNC_TASKS="1" \
     RUST_LOG="error,cat-gateway=debug,cardano_chain_follower=debug,mithril-client=debug" \
-    ./catalyst-gateway/target/release/cat-gateway run --log-level debug
+    ./target/release/cat-gateway run --log-level debug

catalyst-gateway/bin/Cargo.toml

@@ -58,7 +58,7 @@ build-info = "0.0.38"
 ed25519-dalek = "2.1.1"
 scylla = { version = "0.14.0", features = ["cloud", "full-serialization"] }
 strum = { version = "0.26.3", features = ["derive"] }
-# strum_macros = "0.26.4"
+strum_macros = "0.26.4"
 openssl = { version = "0.10.66", features = ["vendored"] }
 num-bigint = "0.4.6"
 futures = "0.3.30"

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

@@ -1,6 +1,6 @@
 //! Index certs found in a transaction.
 
-use std::sync::Arc;
+use std::{fmt::Debug, sync::Arc};
 
 use cardano_chain_follower::MultiEraBlock;
 use pallas::ledger::primitives::{alonzo, conway};
@@ -12,7 +12,7 @@ use crate::{
         queries::{FallibleQueryTasks, PreparedQueries, PreparedQuery, SizedBatch},
         session::CassandraSession,
     },
-    service::utilities::convert::u16_from_saturating,
+    service::utilities::convert::from_saturating,
     settings::CassandraEnvVars,
 };
 
@@ -37,6 +37,38 @@ pub(crate) struct StakeRegistrationInsertQuery {
     pool_delegation: MaybeUnset<Vec<u8>>,
 }
 
+impl Debug for StakeRegistrationInsertQuery {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::result::Result<(), std::fmt::Error> {
+        let stake_address = match self.stake_address {
+            MaybeUnset::Unset => "UNSET",
+            MaybeUnset::Set(ref v) => &hex::encode(v),
+        };
+        let register = match self.register {
+            MaybeUnset::Unset => "UNSET",
+            MaybeUnset::Set(v) => &format!("{v:?}"),
+        };
+        let deregister = match self.deregister {
+            MaybeUnset::Unset => "UNSET",
+            MaybeUnset::Set(v) => &format!("{v:?}"),
+        };
+        let pool_delegation = match self.pool_delegation {
+            MaybeUnset::Unset => "UNSET",
+            MaybeUnset::Set(ref v) => &hex::encode(v),
+        };
+
+        f.debug_struct("StakeRegistrationInsertQuery")
+            .field("stake_hash", &hex::encode(hex::encode(&self.stake_hash)))
+            .field("slot_no", &self.slot_no)
+            .field("txn", &self.txn)
+            .field("stake_address", &stake_address)
+            .field("script", &self.script)
+            .field("register", &register)
+            .field("deregister", &deregister)
+            .field("pool_delegation", &pool_delegation)
+            .finish()
+    }
+}
+
 /// TXI by Txn hash Index
 const INSERT_STAKE_REGISTRATION_QUERY: &str = include_str!("./cql/insert_stake_registration.cql");
 
@@ -130,7 +162,7 @@ impl CertInsertQuery {
         let (key_hash, pubkey, script) = match cred {
             pallas::ledger::primitives::conway::StakeCredential::AddrKeyhash(cred) => {
                 let addr = block
-                    .witness_for_tx(cred, u16_from_saturating(txn))
+                    .witness_for_tx(cred, from_saturating(txn))
                     .unwrap_or(default_addr);
                 // Note: it is totally possible for the Registration Certificate to not be
                 // witnessed.

catalyst-gateway/bin/src/db/index/block/cip36/cql/insert_cip36.cql

@@ -8,7 +8,7 @@ INSERT INTO cip36_registration (
     payment_address,
     is_payable,
     raw_nonce,
-    cip36,
+    cip36
 ) VALUES (
     :stake_address,
     :nonce,
@@ -18,5 +18,5 @@ INSERT INTO cip36_registration (
     :payment_address,
     :is_payable,
     :raw_nonce,
-    :cip36,
+    :cip36
 );

catalyst-gateway/bin/src/db/index/block/cip36/cql/insert_cip36_for_vote_key.cql

@@ -1,14 +1,14 @@
--- Index CIP-36 Registration (Valid)
-INSERT INTO cip36_registration_for_stake_addr (
+-- Index CIP-36 Registration (For each Vote Key)
+INSERT INTO cip36_registration_for_vote_key (
     vote_key,
     stake_address,
     slot_no,
     txn,
-    valid,
+    valid
 ) VALUES (
     :vote_key,
     :stake_address,
     :slot_no,
     :txn,
-    :valid,
+    :valid
 );

catalyst-gateway/bin/src/db/index/block/cip36/cql/insert_cip36_invalid.cql

@@ -10,7 +10,7 @@ INSERT INTO cip36_registration_invalid (
     nonce,
     cip36,
     signed,
-    error_report,
+    error_report
 ) VALUES (
     :stake_address,
     :slot_no,

catalyst-gateway/bin/src/db/index/block/cip36/insert_cip36.rs

@@ -1,6 +1,6 @@
 //! Insert CIP36 Registration Query
 
-use std::sync::Arc;
+use std::{fmt::Debug, sync::Arc};
 
 use cardano_chain_follower::Metadata::cip36::{Cip36, VotingPubKey};
 use scylla::{frame::value::MaybeUnset, SerializeRow, Session};
@@ -37,6 +37,26 @@ pub(super) struct Params {
     cip36: bool,
 }
 
+impl Debug for Params {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let payment_address = match self.payment_address {
+            MaybeUnset::Unset => "UNSET",
+            MaybeUnset::Set(ref v) => &hex::encode(v),
+        };
+        f.debug_struct("Params")
+            .field("stake_address", &self.stake_address)
+            .field("nonce", &self.nonce)
+            .field("slot_no", &self.slot_no)
+            .field("txn", &self.txn)
+            .field("vote_key", &self.vote_key)
+            .field("payment_address", &payment_address)
+            .field("is_payable", &self.is_payable)
+            .field("raw_nonce", &self.raw_nonce)
+            .field("cip36", &self.cip36)
+            .finish()
+    }
+}
+
 impl Params {
     /// Create a new Insert Query.
     pub fn new(vote_key: &VotingPubKey, slot_no: u64, txn: i16, cip36: &Cip36) -> Self {

catalyst-gateway/bin/src/db/index/block/cip36/insert_cip36_for_vote_key.rs

@@ -16,7 +16,7 @@ const INSERT_CIP36_REGISTRATION_FOR_VOTE_KEY_QUERY: &str =
     include_str!("./cql/insert_cip36_for_vote_key.cql");
 
 /// Insert CIP-36 Registration Invalid Query Parameters
-#[derive(SerializeRow, Clone)]
+#[derive(SerializeRow, Debug)]
 pub(super) struct Params {
     /// Voting Public Key
     vote_key: Vec<u8>,

catalyst-gateway/bin/src/db/index/block/cip36/insert_cip36_invalid.rs

@@ -1,6 +1,6 @@
 //! Insert CIP36 Registration Query (Invalid Records)
 
-use std::sync::Arc;
+use std::{fmt::Debug, sync::Arc};
 
 use cardano_chain_follower::Metadata::cip36::{Cip36, VotingPubKey};
 use scylla::{frame::value::MaybeUnset, SerializeRow, Session};
@@ -42,6 +42,28 @@ pub(super) struct Params {
     error_report: Vec<String>,
 }
 
+impl Debug for Params {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let cip36 = match self.cip36 {
+            MaybeUnset::Unset => "UNSET",
+            MaybeUnset::Set(v) => &format!("{v:?}"),
+        };
+        f.debug_struct("Params")
+            .field("stake_address", &self.stake_address)
+            .field("slot_no", &self.slot_no)
+            .field("txn", &self.txn)
+            .field("vote_key", &self.vote_key)
+            .field("payment_address", &self.payment_address)
+            .field("is_payable", &self.is_payable)
+            .field("raw_nonce", &self.raw_nonce)
+            .field("nonce", &self.nonce)
+            .field("cip36", &cip36)
+            .field("signed", &self.signed)
+            .field("error_report", &self.error_report)
+            .finish()
+    }
+}
+
 impl Params {
     /// Create a new Insert Query.
     pub fn new(

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

@@ -75,7 +75,8 @@ impl Cip36InsertQuery {
                                 vote_key, slot_no, txn_index, cip36, true,
                             ));
                     }
-                } else {
+                } else if cip36.stake_pk.is_some() {
+                    // We can't index an error, if there is no stake public key.
                     if cip36.voting_keys.is_empty() {
                         self.invalid.push(insert_cip36_invalid::Params::new(
                             None,

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

@@ -14,7 +14,7 @@ use txi::TxiInsertQuery;
 use txo::TxoInsertQuery;
 
 use super::{queries::FallibleQueryTasks, session::CassandraSession};
-use crate::service::utilities::convert::i16_from_saturating;
+use crate::service::utilities::convert::from_saturating;
 
 /// Add all data needed from the block into the indexes.
 pub(crate) async fn index_block(block: &MultiEraBlock) -> anyhow::Result<()> {
@@ -34,7 +34,7 @@ pub(crate) async fn index_block(block: &MultiEraBlock) -> anyhow::Result<()> {
 
     // We add all transactions in the block to their respective index data sets.
     for (txn_index, txs) in block_data.txs().iter().enumerate() {
-        let txn = i16_from_saturating(txn_index);
+        let txn = from_saturating(txn_index);
 
         let txn_hash = txs.hash().to_vec();
 

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

@@ -14,7 +14,7 @@ use crate::{
 };
 
 /// Insert TXI Query and Parameters
-#[derive(SerializeRow)]
+#[derive(SerializeRow, Debug)]
 pub(crate) struct TxiInsertParams {
     /// Spent Transactions Hash
     txn_hash: Vec<u8>,

catalyst-gateway/bin/src/db/index/block/txo/insert_txo.rs

@@ -17,7 +17,7 @@ const INSERT_TXO_QUERY: &str = include_str!("./cql/insert_txo.cql");
 
 /// Insert TXO Query Parameters
 /// (Superset of data to support both Staked and Unstaked TXO records.)
-#[derive(SerializeRow)]
+#[derive(SerializeRow, Debug)]
 pub(super) struct Params {
     /// Stake Address - Binary 28 bytes. 0 bytes = not staked.
     stake_address: Vec<u8>,

catalyst-gateway/bin/src/db/index/block/txo/insert_txo_asset.rs

@@ -15,7 +15,7 @@ const INSERT_TXO_ASSET_QUERY: &str = include_str!("./cql/insert_txo_asset.cql");
 
 /// Insert TXO Asset Query Parameters
 /// (Superset of data to support both Staked and Unstaked TXO records.)
-#[derive(SerializeRow)]
+#[derive(SerializeRow, Debug)]
 pub(super) struct Params {
     /// Stake Address - Binary 28 bytes. 0 bytes = not staked.
     stake_address: Vec<u8>,

catalyst-gateway/bin/src/db/index/block/txo/insert_unstaked_txo.rs

@@ -14,7 +14,7 @@ const INSERT_UNSTAKED_TXO_QUERY: &str = include_str!("./cql/insert_unstaked_txo.
 
 /// Insert TXO Unstaked Query Parameters
 /// (Superset of data to support both Staked and Unstaked TXO records.)
-#[derive(SerializeRow)]
+#[derive(SerializeRow, Debug)]
 pub(super) struct Params {
     /// Transactions hash.
     txn_hash: Vec<u8>,

catalyst-gateway/bin/src/db/index/block/txo/insert_unstaked_txo_asset.rs

@@ -15,7 +15,7 @@ const INSERT_UNSTAKED_TXO_ASSET_QUERY: &str = include_str!("./cql/insert_unstake
 
 /// Insert TXO Asset Query Parameters
 /// (Superset of data to support both Staked and Unstaked TXO records.)
-#[derive(SerializeRow)]
+#[derive(SerializeRow, Debug)]
 pub(super) struct Params {
     /// Transactions hash.
     txn_hash: Vec<u8>,

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

@@ -17,7 +17,7 @@ use crate::{
         queries::{FallibleQueryTasks, PreparedQuery, SizedBatch},
         session::CassandraSession,
     },
-    service::utilities::convert::i16_from_saturating,
+    service::utilities::convert::from_saturating,
     settings::CassandraEnvVars,
 };
 
@@ -150,7 +150,7 @@ impl TxoInsertQuery {
             };
 
             let staked = stake_address != NO_STAKE_ADDRESS;
-            let txo_index = i16_from_saturating(txo_index);
+            let txo_index = from_saturating(txo_index);
 
             if staked {
                 let params = insert_txo::Params::new(

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

@@ -4,9 +4,9 @@
 
 pub(crate) mod staked_ada;
 
-use std::sync::Arc;
+use std::{fmt::Debug, sync::Arc};
 
-use anyhow::bail;
+use anyhow::{bail, Context};
 use crossbeam_skiplist::SkipMap;
 use scylla::{
     batch::Batch, prepared_statement::PreparedStatement, serialize::row::SerializeRow,
@@ -26,7 +26,8 @@ use crate::settings::{CassandraEnvVars, CASSANDRA_MIN_BATCH_SIZE};
 pub(crate) type SizedBatch = SkipMap<u16, Arc<Batch>>;
 
 /// All Prepared Queries that we know about.
-#[allow(clippy::enum_variant_names, dead_code)]
+#[derive(strum_macros::Display)]
+#[allow(clippy::enum_variant_names)]
 pub(crate) enum PreparedQuery {
     /// TXO Insert query.
     TxoAdaInsertQuery,
@@ -206,7 +207,7 @@ impl PreparedQueries {
     ///
     /// This will divide the batch into optimal sized chunks and execute them until all
     /// values have been executed or the first error is encountered.
-    pub(crate) async fn execute_batch<T: SerializeRow>(
+    pub(crate) async fn execute_batch<T: SerializeRow + Debug>(
         &self, session: Arc<Session>, cfg: Arc<CassandraEnvVars>, query: PreparedQuery,
         values: Vec<T>,
     ) -> FallibleQueryResults {
@@ -238,7 +239,12 @@ impl PreparedQueries {
                 bail!("No batch query found for size {}", chunk_size);
             };
             let batch_query_statements = batch_query.value().clone();
-            results.push(session.batch(&batch_query_statements, chunk).await?);
+            results.push(
+                session
+                    .batch(&batch_query_statements, chunk)
+                    .await
+                    .context(format!("query={query}, chunk={chunk:?}"))?,
+            );
         }
 
         Ok(results)

catalyst-gateway/bin/src/db/index/queries/staked_ada/update_txo_spent.rs

@@ -17,7 +17,7 @@ use crate::{
 const UPDATE_TXO_SPENT_QUERY: &str = include_str!("../cql/update_txo_spent.cql");
 
 /// Update TXO spent query params.
-#[derive(SerializeRow)]
+#[derive(SerializeRow, Debug)]
 pub(crate) struct UpdateTxoSpentQueryParams {
     /// TXO stake address.
     pub stake_address: Vec<u8>,

catalyst-gateway/bin/src/db/index/schema/cql/cip36_registration.cql

@@ -15,4 +15,4 @@ CREATE TABLE IF NOT EXISTS cip36_registration (
 
     PRIMARY KEY (stake_address, nonce, slot_no, txn)
 )
-WITH CLUSTERING ORDER BY (nonce, DESC, slot_no DESC, txn DESC);
+WITH CLUSTERING ORDER BY (nonce DESC, slot_no DESC, txn DESC);

catalyst-gateway/bin/src/db/index/schema/cql/cip36_registration_for_vote_key.cql

@@ -1,7 +1,7 @@
 -- Index of CIP-36 registrations searchable by Stake Address.
 -- Full registration data needs to be queried from the man cip36 registration tables.
 -- Includes both Valid and Invalid registrations.
-CREATE TABLE IF NOT EXISTS cip36_registration_for_stake_addr (
+CREATE TABLE IF NOT EXISTS cip36_registration_for_vote_key (
     -- Primary Key Data
     vote_key            blob,       -- 32 Bytes of Vote Key.
     stake_address       blob,       -- 32 Bytes of Stake Address.