feat: Full UTXO integration test against db-sync/snapshot_tool data (#…

…365) * fix CardanoStakeAddress error handling * refactor, add sync_state_get endpoint * refactor types * refactor * add block_hash validation * wip * wip * wip * wip * add check_network fn * fix * fix schematisis test * try * wip * try * try * try * try * wip * try * try * fix * update Network * add test_utxo test * try * fix * try * fix * wip * fix * fix docket-compose.yml file * try * try * fix * try * try * try * try * wip * fix * wip * try * try * wip * try * try * revert * wip * wip * wip * fix * fix * fix * remove mithril_snapshot loader * wip * wip * wip * wip * wip * wip * wip * add stake addr bech32 encode utils function * wip * wip * update indexing of the utxo data * fix spelling * wip * wip * finish utxo test * fix deny * fix check * fix * fix * update earthly builder versions * wip * ignore test_utxo.py in CI * dont ignore tests * update logging * test(gateway): Add perf metrics when syncing blockchain and also timeout faster when it fails. * feat(backend): Show interval slots/sec as well as cumulative * update preprod version, add additional logging, disable registration indexing * copy snapshot_tool-56364174.json in earthly builder * cleanup --------- Co-authored-by: Steven Johnson <[email protected]>
input-output-hk · Apr 8, 2024 · 2d959bb · 2d959bb
1 parent 41e8781
commit 2d959bb
Show file tree

Hide file tree

Showing 35 changed files with 5,740 additions and 286 deletions.
diff --git a/.config/dictionaries/project.dic b/.config/dictionaries/project.dic
@@ -148,7 +148,7 @@ Traceback
 TXNZD
 Typer
 unmanaged
-UTXO
+utxo
 vitss
 vkey
 voteplan

diff --git a/Earthfile b/Earthfile
@@ -7,17 +7,17 @@ FROM debian:stable-slim
 
 # check-markdown markdown check using catalyst-ci.
 check-markdown:
-    DO github.com/input-output-hk/catalyst-ci/earthly/mdlint:v2.0.10+CHECK
+    DO github.com/input-output-hk/catalyst-ci/earthly/mdlint:v2.11.0+CHECK
 
 # markdown-check-fix markdown check and fix using catalyst-ci.
 markdown-check-fix:
     LOCALLY
 
-    DO github.com/input-output-hk/catalyst-ci/earthly/mdlint:v2.0.10+MDLINT_LOCALLY --src=$(echo ${PWD}) --fix=--fix
+    DO github.com/input-output-hk/catalyst-ci/earthly/mdlint:v2.11.0+MDLINT_LOCALLY --src=$(echo ${PWD}) --fix=--fix
 
 # check-spelling Check spelling in this repo inside a container.
 check-spelling:
-    DO github.com/input-output-hk/catalyst-ci/earthly/cspell:v2.0.10+CHECK
+    DO github.com/input-output-hk/catalyst-ci/earthly/cspell:v2.11.0+CHECK
 
 repo-docs:
     # Create artifacts of extra files we embed inside the documentation when its built.

diff --git a/catalyst-gateway/Earthfile b/catalyst-gateway/Earthfile
@@ -4,9 +4,9 @@ VERSION --try --global-cache 0.7
 
 # Set up our target toolchains, and copy our files.
 builder:
-   DO github.com/input-output-hk/catalyst-ci/earthly/rust:v2.10.3+SETUP
+    DO github.com/input-output-hk/catalyst-ci/earthly/rust:v2.11.0+SETUP
 
-    COPY --dir .cargo .config Cargo.* clippy.toml deny.toml rustfmt.toml bin crates tests .
+    COPY --dir .cargo .config Cargo.* clippy.toml deny.toml rustfmt.toml bin crates .
     COPY --dir ./event-db/queries ./event-db/queries
 
 ## -----------------------------------------------------------------------------
@@ -49,12 +49,16 @@ all-hosts-build:
 
 package-cat-gateway:
     FROM alpine:3.19
+
     ARG tag="latest"
     ARG address
     ARG db_url
     ARG log_level="error"
-    RUN apk upgrade --no-cache && apk add --no-cache gcc
+
+    RUN apk add --no-cache gcc
+
     COPY +build/cat-gateway .
+
     ENTRYPOINT ./cat-gateway run --address $address --database-url $db_url --log-level $log_level
     SAVE IMAGE cat-gateway:$tag
 

diff --git a/catalyst-gateway/bin/src/event_db/error.rs b/catalyst-gateway/bin/src/event_db/error.rs
@@ -42,6 +42,7 @@ pub(crate) enum Error {
     #[error("Unable parse assets: {0}")]
     AssetParsingIssue(String),
     /// Unable to extract hashed witnesses
+    #[allow(dead_code)]
     #[error("Unable to extract hashed witnesses: {0}")]
     HashedWitnessExtraction(String),
 }

diff --git a/catalyst-gateway/bin/src/event_db/utxo.rs b/catalyst-gateway/bin/src/event_db/utxo.rs
@@ -1,18 +1,12 @@
 //! Utxo Queries
 
 use cardano_chain_follower::Network;
-use pallas::ledger::traverse::MultiEraTx;
+use pallas::ledger::{addresses::Address, traverse::MultiEraTx};
 
-use super::{
-    follower::{BlockTime, SlotNumber},
-    voter_registration::StakeCredential,
-};
+use super::{follower::SlotNumber, voter_registration::StakeCredential};
 use crate::{
     event_db::{Error, EventDB},
-    util::{
-        extract_hashed_witnesses, extract_stake_credentials_from_certs,
-        find_matching_stake_credential, parse_policy_assets,
-    },
+    util::parse_policy_assets,
 };
 
 /// Stake amount.
@@ -25,57 +19,58 @@ impl EventDB {
     ) -> Result<(), Error> {
         let conn = self.pool.get().await?;
 
-        for (index, tx) in txs.iter().enumerate() {
-            self.index_txn_data(tx.hash().as_slice(), slot_no, network)
+        for tx in txs {
+            let tx_hash = tx.hash();
+            self.index_txn_data(tx_hash.as_slice(), slot_no, network)
                 .await?;
 
-            let stake_credentials = extract_stake_credentials_from_certs(&tx.certs());
-
-            // Don't index if there is no staking
-            if stake_credentials.is_empty() {
-                return Ok(());
-            }
-
-            let witnesses = match extract_hashed_witnesses(tx.vkey_witnesses()) {
-                Ok(w) => w,
-                Err(err) => return Err(Error::HashedWitnessExtraction(err.to_string())),
-            };
-
-            let (_stake_credential, stake_credential_hash) =
-                match find_matching_stake_credential(&witnesses, &stake_credentials) {
-                    Ok(s) => s,
-                    Err(_err) => {
-                        // Most TXs will not have abided by staking rules, hence logging is too
-                        // noisy. We will not index these TXs and ignore
-                        // them.
-                        return Ok(());
-                    },
-                };
-
             // index outputs
-            for tx_out in tx.outputs() {
+            for (index, tx_out) in tx.outputs().iter().enumerate() {
                 // extract assets
                 let assets = serde_json::to_value(parse_policy_assets(&tx_out.non_ada_assets()))
                     .map_err(|e| Error::AssetParsingIssue(format!("Asset parsing issue {e}")))?;
 
+                let stake_address = match tx_out
+                    .address()
+                    .map_err(|e| Error::Unknown(format!("Address issue {e}")))?
+                {
+                    Address::Shelley(address) => address.try_into().ok(),
+                    Address::Stake(stake_address) => Some(stake_address),
+                    Address::Byron(_) => None,
+                };
+                let stake_credential = stake_address.map(|val| val.payload().as_hash().to_vec());
+
                 let _rows = conn
                     .query(
                         include_str!("../../../event-db/queries/utxo/insert_utxo.sql"),
                         &[
-                            &i32::try_from(index).map_err(|_| Error::NotFound)?,
-                            &tx.hash().as_slice(),
+                            &i32::try_from(index).map_err(|e| Error::Unknown(e.to_string()))?,
+                            &tx_hash.as_slice(),
                             &i64::try_from(tx_out.lovelace_amount())
-                                .map_err(|_| Error::NotFound)?,
-                            &hex::decode(&stake_credential_hash).map_err(|e| {
-                                Error::DecodeHex(format!(
-                                    "Unable to decode stake credential hash {e}"
-                                ))
-                            })?,
+                                .map_err(|e| Error::Unknown(e.to_string()))?,
+                            &stake_credential,
                             &assets,
                         ],
                     )
                     .await?;
             }
+            // update outputs with inputs
+            for tx_in in tx.inputs() {
+                let output = tx_in.output_ref();
+                let output_tx_hash = output.hash();
+                let out_index = output.index();
+
+                let _rows = conn
+                    .query(
+                        include_str!("../../../event-db/queries/utxo/update_utxo.sql"),
+                        &[
+                            &tx_hash.as_slice(),
+                            &output_tx_hash.as_slice(),
+                            &i32::try_from(out_index).map_err(|e| Error::Unknown(e.to_string()))?,
+                        ],
+                    )
+                    .await?;
+            }
         }
 
         Ok(())
@@ -105,10 +100,9 @@ impl EventDB {
     }
 
     /// Get total utxo amount
-    #[allow(dead_code)]
     pub(crate) async fn total_utxo_amount(
-        &self, stake_credential: StakeCredential<'_>, network: Network, date_time: BlockTime,
-    ) -> Result<(StakeAmount, SlotNumber, BlockTime), Error> {
+        &self, stake_credential: StakeCredential<'_>, network: Network, slot_num: SlotNumber,
+    ) -> Result<(StakeAmount, SlotNumber), Error> {
         let conn = self.pool.get().await?;
 
         let network = match network {
@@ -121,7 +115,7 @@ impl EventDB {
         let row = conn
             .query_one(
                 include_str!("../../../event-db/queries/utxo/select_total_utxo_amount.sql"),
-                &[&stake_credential, &network, &date_time],
+                &[&stake_credential, &network, &slot_num],
             )
             .await?;
 
@@ -130,9 +124,8 @@ impl EventDB {
         // https://www.postgresql.org/docs/8.2/functions-aggregate.html
         if let Some(amount) = row.try_get("total_utxo_amount")? {
             let slot_number = row.try_get("slot_no")?;
-            let block_time = row.try_get("block_time")?;
 
-            Ok((amount, slot_number, block_time))
+            Ok((amount, slot_number))
         } else {
             Err(Error::NotFound)
         }

diff --git a/catalyst-gateway/bin/src/event_db/voter_registration.rs b/catalyst-gateway/bin/src/event_db/voter_registration.rs
@@ -56,6 +56,7 @@ impl EventDB {
     }
 
     /// Index registration data
+    #[allow(dead_code)]
     pub async fn index_registration_data(
         &self, txs: Vec<MultiEraTx<'_>>, slot_no: SlotNumber, network: Network,
     ) -> Result<(), Error> {

diff --git a/catalyst-gateway/bin/src/follower.rs b/catalyst-gateway/bin/src/follower.rs
@@ -199,7 +199,6 @@ async fn init_follower(
                             continue;
                         },
                     };
-
                     // Parse block
                     let epoch = match block.epoch(&genesis_values).0.try_into() {
                         Ok(epoch) => epoch,
@@ -242,28 +241,28 @@ async fn init_follower(
                         },
                     }
 
+                    // index utxo
+                    match db.index_utxo_data(block.txs(), slot, network).await {
+                        Ok(()) => (),
+                        Err(err) => {
+                            error!("Unable to index utxo data for block {:?} - skip..", err);
+                            continue;
+                        },
+                    }
+
                     // Block processing for Eras before staking are ignored.
                     if valid_era(block.era()) {
-                        // Utxo
-                        match db.index_utxo_data(block.txs(), slot, network).await {
-                            Ok(()) => (),
-                            Err(err) => {
-                                error!("Unable to index utxo data for block {:?} - skip..", err);
-                                continue;
-                            },
-                        }
-
-                        // Registration
-                        match db.index_registration_data(block.txs(), slot, network).await {
-                            Ok(()) => (),
-                            Err(err) => {
-                                error!(
-                                    "Unable to index registration data for block {:?} - skip..",
-                                    err
-                                );
-                                continue;
-                            },
-                        }
+                        // index catalyst registrations
+                        // match db.index_registration_data(block.txs(), slot,
+                        // network).await {     Ok(()) => (),
+                        //     Err(err) => {
+                        //         error!(
+                        //             "Unable to index registration data for block {:?} -
+                        // skip..",             err
+                        //         );
+                        //         continue;
+                        //     },
+                        // }
 
                         // Rewards
                     }
@@ -319,6 +318,7 @@ async fn follower_connection(
     let mut follower_cfg = if start_from.0.is_none() || start_from.1.is_none() {
         // start from genesis, no previous followers, hence no starting points.
         FollowerConfigBuilder::default()
+            .follow_from(Point::Origin)
             .mithril_snapshot_path(PathBuf::from(snapshot))
             .build()
     } else {

diff --git a/catalyst-gateway/bin/src/service/api/cardano/mod.rs b/catalyst-gateway/bin/src/service/api/cardano/mod.rs
@@ -2,14 +2,14 @@
 
 use std::sync::Arc;
 
-use chrono::{DateTime, Utc};
 use poem::web::Data;
 use poem_openapi::{
     param::{Path, Query},
     OpenApi,
 };
 
 use crate::{
+    event_db::follower::SlotNumber,
     service::{
         common::{
             objects::cardano::{network::Network, stake_address::StakeAddress},
@@ -59,11 +59,13 @@ impl CardanoApi {
         /// `testnet`, to specify `preprod` or `preview` network type use this
         /// query parameter.
         network: Query<Option<Network>>,
-        /// Date time at which the staked ada amount should be calculated.
-        /// If omitted current date time is used.
-        date_time: Query<Option<DateTime<Utc>>>,
+        /// Slot number at which the staked ada amount should be calculated.
+        /// If omitted latest slot number is used.
+        // TODO(bkioshn): https://github.com/input-output-hk/catalyst-voices/issues/239
+        #[oai(validator(minimum(value = "0"), maximum(value = "9223372036854775807")))]
+        slot_number: Query<Option<SlotNumber>>,
     ) -> staked_ada_get::AllResponses {
-        staked_ada_get::endpoint(&data, stake_address.0, network.0, date_time.0).await
+        staked_ada_get::endpoint(&data, stake_address.0, network.0, slot_number.0).await
     }
 
     #[oai(

diff --git a/catalyst-gateway/bin/src/service/api/cardano/staked_ada_get.rs b/catalyst-gateway/bin/src/service/api/cardano/staked_ada_get.rs
@@ -1,6 +1,5 @@
 //! Implementation of the GET `/utxo/staked_ada` endpoint
 
-use chrono::{DateTime, Utc};
 use poem_extensions::{
     response,
     UniResponse::{T200, T400, T404, T503},
@@ -9,11 +8,9 @@ use poem_openapi::{payload::Json, types::ToJSON};
 
 use crate::{
     cli::Error,
-    event_db::error::Error as DBError,
+    event_db::{error::Error as DBError, follower::SlotNumber},
     service::common::{
-        objects::cardano::{
-            network::Network, stake_address::StakeAddress, stake_amount::StakeInfo,
-        },
+        objects::cardano::{network::Network, stake_address::StakeAddress, stake_info::StakeInfo},
         responses::{
             resp_2xx::OK,
             resp_4xx::{ApiValidationError, NotFound},
@@ -76,7 +73,7 @@ fn check_network(
 #[allow(clippy::unused_async)]
 pub(crate) async fn endpoint(
     state: &State, stake_address: StakeAddress, provided_network: Option<Network>,
-    date_time: Option<DateTime<Utc>>,
+    slot_num: Option<SlotNumber>,
 ) -> AllResponses {
     let event_db = match state.event_db() {
         Ok(event_db) => event_db,
@@ -92,7 +89,7 @@ pub(crate) async fn endpoint(
         Err(err) => return server_error_response!("{err}"),
     };
 
-    let date_time = date_time.unwrap_or_else(Utc::now);
+    let date_time = slot_num.unwrap_or(SlotNumber::MAX);
     let stake_credential = stake_address.payload().as_hash().as_ref();
 
     let network = match check_network(stake_address.network(), provided_network) {
@@ -105,11 +102,10 @@ pub(crate) async fn endpoint(
         .total_utxo_amount(stake_credential, network.into(), date_time)
         .await
     {
-        Ok((amount, slot_number, block_time)) => {
+        Ok((amount, slot_number)) => {
             T200(OK(Json(StakeInfo {
                 amount,
                 slot_number,
-                block_time,
             })))
         },
         Err(DBError::NotFound) => T404(NotFound),

diff --git a/catalyst-gateway/bin/src/service/common/objects/cardano/mod.rs b/catalyst-gateway/bin/src/service/common/objects/cardano/mod.rs
@@ -2,5 +2,5 @@
 
 pub(crate) mod network;
 pub(crate) mod stake_address;
-pub(crate) mod stake_amount;
+pub(crate) mod stake_info;
 pub(crate) mod sync_state;