Merge pull request #47 from Concordium/forward-comp/COR-1780-compatibility

Upward Unknown will be handled to raise an error

Author: rbaconcordium

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased changes
 
+- Update the Rust SDK for better forwards compatibility with future node versions and revised error handling or reporting for unknown transaction and event types. If within the database hook callback has unknown data variants in the block processing, the process would complain and cease, raising `IndexingError` to alert.
+
 ## [0.14.0] - 2025-08-07
 
 ### Changed

deps/concordium-rust-sdk

@@ -168,13 +168,13 @@ pub trait DatabaseHooks<D, P> {
     async fn insert_into_db(
         db_conn: &mut DBConn<P>,
         data: &D,
-    ) -> Result<BlockInsertSuccess, DatabaseError>;
+    ) -> Result<BlockInsertSuccess, IndexingError>;
 
     /// Invoked by the database thread to request the latest recorded height in
     /// the database.
     async fn on_request_max_height(
         db: &DatabaseClient,
-    ) -> Result<Option<AbsoluteBlockHeight>, DatabaseError>;
+    ) -> Result<Option<AbsoluteBlockHeight>, IndexingError>;
 }
 
 /// A collection of possible errors that can happen while using the node to
@@ -198,6 +198,17 @@ pub enum NodeError {
     OtherError(#[from] anyhow::Error),
 }
 
+#[derive(Debug, thiserror::Error)]
+/// Possible errors while indexing blocks into the database
+pub enum IndexingError {
+    /// Database error.
+    #[error("Error using the database {0}.")]
+    PostgresError(#[from] postgres::Error),
+    /// Unknown Data type encountered error
+    #[error("Please update the rust SDK. Reason for this could be due to {0}.")]
+    UnknownData(String),
+}
+
 /// Defines a set of necessary callbacks used while interacting with a node.
 #[async_trait]
 pub trait NodeHooks<D> {
@@ -289,6 +300,11 @@ where
                     );
                 }
                 Err(e) => {
+                    if let IndexingError::UnknownData(reasons) = &e {
+                        log::error!("Encountered unknown data variant. Please update the rust SDK. Reason(s): {}", reasons);
+                        return Err(anyhow::anyhow!("Encountered unknown data variant. Please update the rust SDK. Reason(s): {}", reasons));
+                    }
+
                     successive_errors += 1;
                     // wait for 2^(min(successive_errors - 1, 7)) seconds before attempting.
                     // The reason for the min is that we bound the time between reconnects.

src/lib.rs

@@ -8,22 +8,26 @@ use concordium_rust_sdk::{
         hashes::BlockHash, queries::BlockInfo, AbsoluteBlockHeight, BlockItemSummary,
         ContractAddress, SpecialTransactionOutcome,
     },
-    v2::{self, FinalizedBlockInfo},
+    v2::{self, FinalizedBlockInfo, Upward},
 };
-use futures::TryStreamExt;
+use futures::{StreamExt, TryStreamExt};
 use std::{collections::HashSet, convert::TryFrom, hash::Hash};
 use structopt::StructOpt;
 use tokio_postgres::{
     types::{Json, ToSql},
     Transaction as DBTransaction,
 };
 use tonic::async_trait;
+use tonic::Status;
 use transaction_logger::{
     postgres::{self, DatabaseClient},
-    run_service, set_shutdown, BlockInsertSuccess, DatabaseError, DatabaseHooks, NodeError,
+    run_service, set_shutdown, BlockInsertSuccess, DatabaseHooks, IndexingError, NodeError,
     NodeHooks, PrepareStatements, SharedIndexerArgs,
 };
 
+type ContractEvents = Vec<cis2::Event>;
+type ContractEffects = Vec<(ContractAddress, ContractEvents)>;
+
 type DBConn = transaction_logger::DBConn<PreparedStatements>;
 
 const MAX_CONNECT_ATTEMPTS: u32 = 6;
@@ -258,7 +262,7 @@ impl PreparedStatements {
         block_time: Timestamp,
         block_height: AbsoluteBlockHeight,
         ts: &BlockItemSummaryWithCanonicalAddresses,
-    ) -> Result<(), postgres::Error> {
+    ) -> Result<(), IndexingError> {
         let affected_addresses = &ts.addresses;
         let summary_row = SummaryRow {
             block_hash,
@@ -275,15 +279,27 @@ impl PreparedStatements {
             ];
             tx.query_opt(&self.insert_ati, &values).await?;
         }
+
         // insert contracts
-        for affected in ts.summary.affected_contracts() {
+        let affected_contracts = ts.summary.affected_contracts().known_or_else(|| {
+            log::error!("Could not determine affected contracts for a BlockItemSummary of unknown type. {:?}", ts.summary);
+            IndexingError::UnknownData("Could not determine affected contracts for a BlockItemSummary of unknown type.".to_string())
+        })?; // if Unknown, throw an error
+
+        for contract_address in affected_contracts {
+            let affected = contract_address.known_or_else(|| {
+                log::error!("Could not determine affected contracts of an unknown type of AccountTransactionEffects. {:?}", ts.summary);
+                IndexingError::UnknownData("Could not determine affected contracts of an unknown type of AccountTransactionEffects.".to_string())
+            })?; // encountered unknown contract_address, throw an error, this will stop the insert into db process and alert the user to update the rust SDK
+
             let index = affected.index;
             let subindex = affected.subindex;
             let values = [
                 &(index as i64) as &(dyn ToSql + Sync),
                 &(subindex as i64),
                 &id,
             ];
+
             tx.query_opt(&self.insert_cti, &values).await?;
         }
 
@@ -398,41 +414,41 @@ impl PreparedStatements {
         &self,
         tx: &DBTransaction<'_>,
         ts: &BlockItemSummary,
-    ) -> Result<(), postgres::Error> {
-        if let Some(effects) = get_cis2_events(ts) {
-            for (ca, events) in effects {
-                for event in events {
-                    match event {
-                        cis2::Event::Transfer { .. } => {
-                            // do nothing, tokens are not created here.
-                        }
-                        cis2::Event::Mint {
-                            ref token_id,
-                            ref amount,
-                            ..
-                        } => {
-                            self.cis2_increase_total_supply(tx, ca, token_id, amount)
-                                .await?;
-                        }
-                        cis2::Event::Burn {
-                            ref token_id,
-                            ref amount,
-                            ..
-                        } => {
-                            self.cis2_decrease_total_supply(tx, ca, token_id, amount)
-                                .await?;
-                        }
-                        cis2::Event::UpdateOperator { .. } => {
-                            // do nothing, updating operators does not change
-                            // token suply
-                        }
-                        cis2::Event::TokenMetadata { .. } => {
-                            // do nothing, updating token metadata does not
-                            // change token supply.
-                        }
-                        cis2::Event::Unknown => {
-                            // do nothing, not a CIS2 event
-                        }
+    ) -> Result<(), IndexingError> {
+        let effects = get_cis2_events(ts)?;
+
+        for (ca, events) in effects {
+            for event in events {
+                match event {
+                    cis2::Event::Transfer { .. } => {
+                        // do nothing, tokens are not created here.
+                    }
+                    cis2::Event::Mint {
+                        ref token_id,
+                        ref amount,
+                        ..
+                    } => {
+                        self.cis2_increase_total_supply(tx, ca, token_id, amount)
+                            .await?;
+                    }
+                    cis2::Event::Burn {
+                        ref token_id,
+                        ref amount,
+                        ..
+                    } => {
+                        self.cis2_decrease_total_supply(tx, ca, token_id, amount)
+                            .await?;
+                    }
+                    cis2::Event::UpdateOperator { .. } => {
+                        // do nothing, updating operators does not change
+                        // token suply
+                    }
+                    cis2::Event::TokenMetadata { .. } => {
+                        // do nothing, updating token metadata does not
+                        // change token supply.
+                    }
+                    cis2::Event::Unknown => {
+                        // do nothing, not a CIS2 event
                     }
                 }
             }
@@ -450,7 +466,7 @@ async fn insert_block(
     block_height: AbsoluteBlockHeight,
     item_summaries: &[BlockItemSummaryWithCanonicalAddresses],
     special_events: &[SpecialTransactionOutcome],
-) -> Result<chrono::Duration, postgres::Error> {
+) -> Result<chrono::Duration, IndexingError> {
     let start = chrono::Utc::now();
     let db_tx = db.client.as_mut().transaction().await?;
     let prepared = &db.prepared;
@@ -492,33 +508,55 @@ async fn get_last_block_height(
 /// parsing then the logs for that section of execution are ignored, since it
 /// indicates an error in the contract.
 ///
-/// The return value of [`None`] means there are no understandable CIS2 logs
+/// The return value of empty vec means there are no understandable CIS2 logs
 /// produced.
-fn get_cis2_events(bi: &BlockItemSummary) -> Option<Vec<(ContractAddress, Vec<cis2::Event>)>> {
+fn get_cis2_events(bi: &BlockItemSummary) -> Result<ContractEffects, IndexingError> {
     match bi.contract_update_logs() {
-        Some(log_iter) => Some(
-            log_iter
-                .flat_map(|(ca, logs)| {
-                    match logs
-                        .iter()
-                        .map(cis2::Event::try_from)
-                        .collect::<Result<Vec<cis2::Event>, _>>()
-                    {
-                        Ok(events) => Some((ca, events)),
-                        Err(_) => None,
-                    }
-                })
-                .collect(),
-        ),
+        Some(log_iter) => {
+            // Map each log into a Result
+            let mut events: ContractEffects = Vec::new();
+            for contract_log in log_iter {
+                let (ca, logs) = contract_log.known_or_else(|| {
+                    log::error!(
+                        "Could not determine contract log, unknown type. {:?}",
+                        contract_log
+                    );
+                    IndexingError::UnknownData(
+                        "Could not determine contract log, unknown type.".to_string(),
+                    )
+                })?;
+
+                let evs = logs
+                    .iter()
+                    .map(cis2::Event::try_from)
+                    .collect::<Result<Vec<cis2::Event>, _>>()
+                    .ok();
+
+                if let Some(evs) = evs {
+                    events.push((ca, evs));
+                }
+            }
+            //if no events were parsed due to non cis2 logs, the vector will be empty
+            //so just return the events vector, empty or not empty.
+            Ok(events)
+        }
         None => {
-            let init = bi.contract_init()?;
-            let cis2 = init
+            let init = match bi.contract_init() {
+                Some(init) => init,
+                None => return Ok(vec![]),
+            };
+
+            let cis2 = match init
                 .events
                 .iter()
                 .map(cis2::Event::try_from)
                 .collect::<Result<Vec<cis2::Event>, _>>()
-                .ok()?;
-            Some(vec![(init.address, cis2)])
+            {
+                Ok(vec) => vec,
+                Err(_) => return Ok(vec![]),
+            };
+
+            Ok(vec![(init.address, cis2)])
         }
     }
 }
@@ -531,7 +569,7 @@ impl DatabaseHooks<TransactionLogData, PreparedStatements> for DatabaseState {
     async fn insert_into_db(
         db_conn: &mut DBConn,
         (bi, item_summaries, special_events): &TransactionLogData,
-    ) -> Result<BlockInsertSuccess, DatabaseError> {
+    ) -> Result<BlockInsertSuccess, IndexingError> {
         let duration = insert_block(
             db_conn,
             bi.block_hash,
@@ -551,7 +589,7 @@ impl DatabaseHooks<TransactionLogData, PreparedStatements> for DatabaseState {
 
     async fn on_request_max_height(
         db: &DatabaseClient,
-    ) -> Result<Option<AbsoluteBlockHeight>, DatabaseError> {
+    ) -> Result<Option<AbsoluteBlockHeight>, IndexingError> {
         let height = get_last_block_height(db).await?;
         Ok(height)
     }
@@ -602,18 +640,30 @@ impl NodeHooks<TransactionLogData> for CanonicalAddressCache {
                 .try_collect()
                 .await?
         };
+
         let special_events = node
             .get_block_special_events(finalized_block_info.height)
             .await?
             .response
+            .map(|upward_res| {
+                upward_res.and_then(|upward| match upward {
+                    Upward::Known(special_transaction_outcome) => Ok(special_transaction_outcome),
+                    Upward::Unknown => {
+                        Err(Status::unknown("Unknown SpecialTransactionOutcome type"))
+                    }
+                })
+            })
             .try_collect()
             .await?;
 
         // Map account addresses affected by each summary to their respective canonical
         // address
         let mut with_addresses = Vec::with_capacity(transaction_summaries.len());
         for summary in transaction_summaries {
-            let affected_addresses = summary.affected_addresses();
+            let affected_addresses = summary.affected_addresses().known_or_else(|| {
+                Status::unknown("Could not determine affected addresses for BlockItem")
+            })?; // if unknown, throw Err Status::unknown
+
             let mut addresses = Vec::with_capacity(affected_addresses.len());
             // resolve canonical addresses. This part is only needed because the index
             // is currently expected by "canonical address",