fix: naming mismatch (#1349)

* fix: naming mismatch Signed-off-by: bkioshn <[email protected]> * fix(cat-gateway): better error logging when a query fails. --------- Signed-off-by: bkioshn <[email protected]> Co-authored-by: Steven Johnson <[email protected]>
input-output-hk · Dec 6, 2024 · 9810f16 · 9810f16
1 parent 294268d
commit 9810f16
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 18 deletions.
diff --git a/catalyst-gateway/bin/src/db/index/block/mod.rs b/catalyst-gateway/bin/src/db/index/block/mod.rs
@@ -81,13 +81,10 @@ pub(crate) async fn index_block(block: &MultiEraBlock) -> anyhow::Result<()> {
         }
         match handle.await {
             Ok(join_res) => {
-                match join_res {
-                    Ok(_res) => {}, // debug!(res=?res,"Query OK")
-                    Err(error) => {
-                        // IF a query fails, assume everything else is broken.
-                        error!(error=%error,"Query Failed");
-                        result = Err(error);
-                    },
+                if let Err(error) = join_res {
+                    // IF a query fails, assume everything else is broken.
+                    error!(error=%error,"Query Failed");
+                    result = Err(error);
                 }
             },
             Err(error) => {

diff --git a/catalyst-gateway/bin/src/db/index/block/rbac509/insert_chain_root_for_stake_address.rs b/catalyst-gateway/bin/src/db/index/block/rbac509/insert_chain_root_for_stake_address.rs
@@ -17,7 +17,7 @@ const INSERT_CHAIN_ROOT_FOR_STAKE_ADDRESS_QUERY: &str =
 #[derive(SerializeRow)]
 pub(super) struct Params {
     /// Stake Address Hash. 32 bytes.
-    stake_address: Vec<u8>,
+    stake_addr: Vec<u8>,
     /// Block Slot Number
     slot_no: num_bigint::BigInt,
     /// Transaction Offset inside the block.
@@ -29,7 +29,7 @@ pub(super) struct Params {
 impl Debug for Params {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Params")
-            .field("stake_address", &self.stake_address)
+            .field("stake_addr", &self.stake_addr)
             .field("slot_no", &self.slot_no)
             .field("txn", &self.txn)
             .field("chain_root", &self.chain_root)
@@ -39,9 +39,9 @@ impl Debug for Params {
 
 impl Params {
     /// Create a new record for this transaction.
-    pub(super) fn new(stake_address: &[u8], chain_root: &[u8], slot_no: u64, txn: i16) -> Self {
+    pub(super) fn new(stake_addr: &[u8], chain_root: &[u8], slot_no: u64, txn: i16) -> Self {
         Params {
-            stake_address: stake_address.to_vec(),
+            stake_addr: stake_addr.to_vec(),
             slot_no: num_bigint::BigInt::from(slot_no),
             txn,
             chain_root: chain_root.to_vec(),

diff --git a/catalyst-gateway/bin/src/db/index/queries/mod.rs b/catalyst-gateway/bin/src/db/index/queries/mod.rs
@@ -9,7 +9,7 @@ pub(crate) mod sync_status;
 
 use std::{fmt::Debug, sync::Arc};
 
-use anyhow::{bail, Context};
+use anyhow::bail;
 use crossbeam_skiplist::SkipMap;
 use rbac::{
     get_chain_root::GetChainRootQuery, get_registrations::GetRegistrationsByChainRootQuery,
@@ -29,6 +29,7 @@ use staked_ada::{
     get_txo_by_stake_address::GetTxoByStakeAddressQuery, update_txo_spent::UpdateTxoSpentQuery,
 };
 use sync_status::update::SyncStatusInsertQuery;
+use tracing::error;
 
 use super::block::{
     certs::CertInsertQuery, cip36::Cip36InsertQuery, rbac509::Rbac509InsertQuery,
@@ -379,6 +380,8 @@ impl PreparedQueries {
         let mut results: Vec<QueryResult> = Vec::new();
 
         let chunks = values.chunks(cfg.max_batch_size.try_into().unwrap_or(1));
+        let mut query_failed = false;
+        let query_str = format!("{query}");
 
         for chunk in chunks {
             let chunk_size: u16 = chunk.len().try_into()?;
@@ -387,12 +390,19 @@ 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
-                    .context(format!("query={query}, chunk={chunk:?}"))?,
-            );
+            match session.batch(&batch_query_statements, chunk).await {
+                Ok(result) => results.push(result),
+                Err(err) => {
+                    let chunk_str = format!("{chunk:?}");
+                    error!(error=%err, query=query_str, chunk=chunk_str, "Query Execution Failed");
+                    query_failed = true;
+                    // Defer failure until all batches have been processed.
+                },
+            }
+        }
+
+        if query_failed {
+            bail!("Query Failed: {query_str}!");
         }
 
         Ok(results)