Show transaction status (helius-labs#148)

* Show transaction status

* Bump

Cargo.toml

@@ -7,7 +7,7 @@ name = "photon-indexer"
 publish = true
 readme = "README.md"
 repository = "https://github.com/helius-labs/photon"
-version = "0.36.0"
+version = "0.38.0"
 
 [[bin]]
 name = "photon"
@@ -97,6 +97,7 @@ yellowstone-grpc-proto = "1.14.0"
 cadence = "1.4.0"
 async-stream = "0.3.5"
 rand = "0.8.5"
+bincode = "1.3.3"
 
 [dev-dependencies]
 function_name = "0.3.0"

src/api/api.rs

@@ -38,8 +38,10 @@ use super::method::get_transaction_with_compression_info::{
 use super::method::get_validity_proof::{
     get_validity_proof, GetValidityProofRequest, GetValidityProofResponse,
 };
-use super::method::utils::GetLatestSignaturesRequest;
 use super::method::utils::{AccountBalanceResponse, GetPaginatedSignaturesResponse, HashRequest};
+use super::method::utils::{
+    GetLatestSignaturesRequest, GetNonPaginatedSignaturesResponseWithError,
+};
 use super::{
     error::PhotonApiError,
     method::{
@@ -290,7 +292,7 @@ impl PhotonApi {
     pub async fn get_latest_non_voting_signatures(
         &self,
         request: GetLatestSignaturesRequest,
-    ) -> Result<GetNonPaginatedSignaturesResponse, PhotonApiError> {
+    ) -> Result<GetNonPaginatedSignaturesResponseWithError, PhotonApiError> {
         get_latest_non_voting_signatures(self.db_conn.as_ref(), request).await
     }
 
@@ -394,7 +396,7 @@ impl PhotonApi {
             OpenApiSpec {
                 name: "getLatestNonVotingSignatures".to_string(),
                 request: Some(GetLatestSignaturesRequest::schema().1),
-                response: GetNonPaginatedSignaturesResponse::schema().1,
+                response: GetNonPaginatedSignaturesResponseWithError::schema().1,
             },
             OpenApiSpec {
                 name: "getIndexerHealth".to_string(),

src/api/method/get_compression_signatures_for_account.rs

@@ -40,7 +40,9 @@ pub async fn get_compression_signatures_for_account(
     }
 
     Ok(GetNonPaginatedSignaturesResponse {
-        value: SignatureInfoList { items: signatures },
+        value: SignatureInfoList {
+            items: signatures.into_iter().map(|s| s.into()).collect(),
+        },
         context,
     })
 }

src/api/method/get_compression_signatures_for_address.rs

@@ -36,7 +36,7 @@ pub async fn get_compression_signatures_for_address(
     .await?;
 
     Ok(GetPaginatedSignaturesResponse {
-        value: signatures,
+        value: signatures.into(),
         context,
     })
 }

src/api/method/get_compression_signatures_for_owner.rs

@@ -36,7 +36,7 @@ pub async fn get_compression_signatures_for_owner(
     .await?;
 
     Ok(GetPaginatedSignaturesResponse {
-        value: signatures,
+        value: signatures.into(),
         context,
     })
 }

src/api/method/get_compression_signatures_for_token_owner.rs

@@ -36,7 +36,7 @@ pub async fn get_compression_signatures_for_token_owner(
     .await?;
 
     Ok(GetPaginatedSignaturesResponse {
-        value: signatures,
+        value: signatures.into(),
         context,
     })
 }

src/api/method/get_latest_compression_signatures.rs

@@ -23,7 +23,7 @@ pub async fn get_latest_compression_signatures(
     .await?;
 
     Ok(GetPaginatedSignaturesResponse {
-        value: signatures,
+        value: signatures.into(),
         context,
     })
 }

src/api/method/get_latest_non_voting_signatures.rs

@@ -1,4 +1,7 @@
-use super::utils::{GetLatestSignaturesRequest, GetNonPaginatedSignaturesResponse};
+use super::utils::{
+    GetLatestSignaturesRequest, GetNonPaginatedSignaturesResponseWithError,
+    SignatureInfoListWithError,
+};
 use sea_orm::DatabaseConnection;
 
 use super::{
@@ -9,7 +12,7 @@ use super::{
 pub async fn get_latest_non_voting_signatures(
     conn: &DatabaseConnection,
     request: GetLatestSignaturesRequest,
-) -> Result<GetNonPaginatedSignaturesResponse, PhotonApiError> {
+) -> Result<GetNonPaginatedSignaturesResponseWithError, PhotonApiError> {
     let context = Context::extract(conn).await?;
 
     let signatures = search_for_signatures(
@@ -22,8 +25,8 @@ pub async fn get_latest_non_voting_signatures(
     )
     .await?;
 
-    Ok(GetNonPaginatedSignaturesResponse {
-        value: super::utils::SignatureInfoList {
+    Ok(GetNonPaginatedSignaturesResponseWithError {
+        value: SignatureInfoListWithError {
             items: signatures.items,
         },
         context,

src/api/method/utils.rs

@@ -441,11 +441,21 @@ pub struct SignatureInfo {
     pub block_time: UnixTimestamp,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct SignatureInfoWithError {
+    pub signature: SerializableSignature,
+    pub slot: UnsignedInteger,
+    pub block_time: UnixTimestamp,
+    pub error: Option<String>,
+}
+
 #[derive(FromQueryResult)]
 pub struct SignatureInfoModel {
     pub signature: Vec<u8>,
     pub slot: i64,
     pub block_time: i64,
+    pub error: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, Default)]
@@ -454,13 +464,45 @@ pub struct SignatureInfoList {
     pub items: Vec<SignatureInfo>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct SignatureInfoListWithError {
+    pub items: Vec<SignatureInfoWithError>,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct PaginatedSignatureInfoList {
     pub items: Vec<SignatureInfo>,
     pub cursor: Option<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct PaginatedSignatureInfoListWithError {
+    pub items: Vec<SignatureInfoWithError>,
+    pub cursor: Option<String>,
+}
+
+impl From<PaginatedSignatureInfoListWithError> for PaginatedSignatureInfoList {
+    fn from(paginated: PaginatedSignatureInfoListWithError) -> Self {
+        PaginatedSignatureInfoList {
+            items: paginated.items.into_iter().map(Into::into).collect(),
+            cursor: paginated.cursor,
+        }
+    }
+}
+
+impl From<SignatureInfoWithError> for SignatureInfo {
+    fn from(signature_info: SignatureInfoWithError) -> Self {
+        SignatureInfo {
+            signature: signature_info.signature,
+            slot: signature_info.slot,
+            block_time: signature_info.block_time,
+        }
+    }
+}
+
 pub enum SignatureFilter {
     Account(Hash),
     Address(SerializablePubkey),
@@ -561,7 +603,7 @@ fn compute_raw_sql_query_and_args(
 
             let raw_sql = format!(
                 "
-                SELECT DISTINCT transactions.signature, transactions.slot, blocks.block_time
+                SELECT DISTINCT transactions.signature, transactions.slot, transactions.error, blocks.block_time
                 FROM account_transactions
                 JOIN transactions ON account_transactions.signature = transactions.signature
                 JOIN blocks ON transactions.slot = blocks.slot
@@ -588,7 +630,7 @@ fn compute_raw_sql_query_and_args(
             let (cursor_filter, cursor_args) = compute_cursor_filter(cursor, 0)?;
             let raw_sql = format!(
                 "
-                SELECT transactions.signature, transactions.slot, blocks.block_time
+                SELECT transactions.signature, transactions.slot, transactions.error, blocks.block_time
                 FROM transactions
                 JOIN blocks ON transactions.slot = blocks.slot
                 {cursor_filter}
@@ -609,7 +651,7 @@ pub async fn search_for_signatures(
     only_compressed: bool,
     cursor: Option<String>,
     limit: Option<Limit>,
-) -> Result<PaginatedSignatureInfoList, PhotonApiError> {
+) -> Result<PaginatedSignatureInfoListWithError, PhotonApiError> {
     let limit = limit.unwrap_or_default().0;
     let (raw_sql, args) = compute_raw_sql_query_and_args(
         search_type,
@@ -639,26 +681,29 @@ pub async fn search_for_signatures(
         }),
     };
 
-    Ok(PaginatedSignatureInfoList {
+    Ok(PaginatedSignatureInfoListWithError {
         items: signatures,
         cursor,
     })
 }
 
 pub fn parse_signatures_infos(
     signatures: Vec<SignatureInfoModel>,
-) -> Result<Vec<SignatureInfo>, PhotonApiError> {
+) -> Result<Vec<SignatureInfoWithError>, PhotonApiError> {
     signatures.into_iter().map(parse_signature_info).collect()
 }
 
-fn parse_signature_info(model: SignatureInfoModel) -> Result<SignatureInfo, PhotonApiError> {
-    Ok(SignatureInfo {
+fn parse_signature_info(
+    model: SignatureInfoModel,
+) -> Result<SignatureInfoWithError, PhotonApiError> {
+    Ok(SignatureInfoWithError {
         signature: SerializableSignature(
             Signature::try_from(model.signature)
                 .map_err(|_| PhotonApiError::UnexpectedError("Invalid signature".to_string()))?,
         ),
         slot: UnsignedInteger(model.slot as u64),
         block_time: UnixTimestamp(model.block_time as u64),
+        error: model.error,
     })
 }
 
@@ -692,3 +737,11 @@ pub struct GetNonPaginatedSignaturesResponse {
     pub context: Context,
     pub value: SignatureInfoList,
 }
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+// We do not use generics to simplify documentation generation.
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct GetNonPaginatedSignaturesResponseWithError {
+    pub context: Context,
+    pub value: SignatureInfoListWithError,
+}

src/dao/generated/indexed_trees.rs

@@ -7,6 +7,7 @@ use sea_orm::entity::prelude::*;
 pub struct Model {
     #[sea_orm(primary_key, auto_increment = false)]
     pub tree: Vec<u8>,
+    #[sea_orm(primary_key, auto_increment = false)]
     pub leaf_index: i64,
     pub value: Vec<u8>,
     pub next_index: i64,

src/dao/generated/transactions.rs

@@ -9,6 +9,8 @@ pub struct Model {
     pub signature: Vec<u8>,
     pub slot: i64,
     pub uses_compression: bool,
+    #[sea_orm(column_type = "Text", nullable)]
+    pub error: Option<String>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

src/ingester/fetchers/grpc.rs

@@ -237,8 +237,8 @@ fn parse_block(block: SubscribeUpdateBlock) -> BlockInfo {
 
 fn parse_transaction(transaction: SubscribeUpdateTransactionInfo) -> TransactionInfo {
     let meta = transaction.meta.unwrap();
+    let error = meta.err.map(|e| bincode::deserialize(&e.err).unwrap());
     let signature = Signature::try_from(transaction.signature).unwrap();
-    let success = meta.err.is_none();
     let message = transaction.transaction.unwrap().message.unwrap();
     let outer_intructions = message.instructions;
     let mut accounts = message.account_keys;
@@ -302,6 +302,6 @@ fn parse_transaction(transaction: SubscribeUpdateTransactionInfo) -> Transaction
     TransactionInfo {
         instruction_groups,
         signature,
-        success,
+        error,
     }
 }