no op metadata json requests based on slot

core/src/metadata_json.rs

@@ -1,13 +1,15 @@
 use {
     backon::{ExponentialBuilder, Retryable},
     clap::Parser,
-    digital_asset_types::dao::asset_data,
+    digital_asset_types::dao::{asset_data, sea_orm_active_enums::MetadataJsonFetchResult},
     futures::{future::BoxFuture, stream::FuturesUnordered, StreamExt},
     indicatif::HumanDuration,
     log::{debug, error},
     reqwest::{Client, Url as ReqwestUrl},
     sea_orm::{entity::*, SqlxPostgresConnector},
     serde::{Deserialize, Serialize},
+    sqlx::{Pool, Postgres},
+    std::sync::atomic::{AtomicU8, Ordering},
     tokio::{
         sync::mpsc::{error::SendError, unbounded_channel, UnboundedSender},
         task::JoinHandle,
@@ -65,6 +67,25 @@ pub struct MetadataJsonDownloadWorkerArgs {
     metadata_json_download_worker_request_timeout: u64,
 }
 
+async fn is_asset_data_fetch_req(
+    download_metadata_info: &DownloadMetadataInfo,
+    pool: Pool<Postgres>,
+) -> bool {
+    let DownloadMetadataInfo {
+        asset_data_id,
+        slot: incoming_slot,
+        ..
+    } = download_metadata_info;
+
+    let conn = SqlxPostgresConnector::from_sqlx_postgres_pool(pool);
+
+    asset_data::Entity::find_by_id(asset_data_id.clone())
+        .one(&conn)
+        .await
+        .unwrap_or(None)
+        .is_some_and(|model| *incoming_slot > model.slot_updated)
+}
+
 impl MetadataJsonDownloadWorkerArgs {
     pub fn start(
         &self,
@@ -90,9 +111,10 @@ impl MetadataJsonDownloadWorkerArgs {
                 }
 
                 let pool = pool.clone();
-                let client = client.clone();
 
-                handlers.push(spawn_task(client, pool, download_metadata_info));
+                if is_asset_data_fetch_req(&download_metadata_info, pool.clone()).await {
+                    handlers.push(spawn_task(client.clone(), pool, download_metadata_info));
+                }
             }
 
             while handlers.next().await.is_some() {}
@@ -159,20 +181,39 @@ pub enum StatusCode {
     Code(reqwest::StatusCode),
 }
 
+pub struct MetadataJsonData {
+    value: serde_json::Value,
+    time_elapsed: u64,
+    retries: u8,
+}
+
+pub struct MetadataJsonFetchError {
+    error: FetchMetadataJsonError,
+    time_elapsed: u64,
+    retries: u8,
+}
+
 async fn fetch_metadata_json(
     client: Client,
     metadata_json_url: &str,
-) -> Result<serde_json::Value, FetchMetadataJsonError> {
-    (|| async {
+) -> Result<MetadataJsonData, MetadataJsonFetchError> {
+    let retries = AtomicU8::new(0);
+    let start = Instant::now();
+
+    let res = (|| async {
         let url = ReqwestUrl::parse(metadata_json_url)?;
 
         let response = client.get(url.clone()).send().await?;
 
         match response.error_for_status() {
-            Ok(res) => res
-                .json::<serde_json::Value>()
-                .await
-                .map_err(|source| FetchMetadataJsonError::Parse { source, url }),
+            Ok(res) => {
+                let value = res
+                    .json::<serde_json::Value>()
+                    .await
+                    .map_err(|source| FetchMetadataJsonError::Parse { source, url })?;
+
+                Ok(value)
+            }
             Err(source) => {
                 let status = source
                     .status()
@@ -188,7 +229,25 @@ async fn fetch_metadata_json(
         }
     })
     .retry(&ExponentialBuilder::default())
-    .await
+    .notify(|_e, _d| {
+        retries.fetch_add(1, Ordering::Relaxed);
+    })
+    .await;
+
+    let time_elapsed = start.elapsed().as_secs();
+
+    let retries = retries.load(Ordering::Relaxed);
+
+    res.map(|value| MetadataJsonData {
+        value,
+        time_elapsed,
+        retries,
+    })
+    .map_err(|error| MetadataJsonFetchError {
+        error,
+        time_elapsed,
+        retries,
+    })
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -206,22 +265,37 @@ pub async fn perform_metadata_json_task(
     pool: sqlx::PgPool,
     download_metadata_info: &DownloadMetadataInfo,
 ) -> Result<asset_data::Model, MetadataJsonTaskError> {
+    let conn = SqlxPostgresConnector::from_sqlx_postgres_pool(pool);
     match fetch_metadata_json(client, &download_metadata_info.uri).await {
         Ok(metadata) => {
             let active_model = asset_data::ActiveModel {
                 id: Set(download_metadata_info.asset_data_id.clone()),
-                metadata: Set(metadata),
+                metadata: Set(metadata.value),
                 reindex: Set(Some(false)),
+                last_requested_status_code: Set(Some(MetadataJsonFetchResult::Success)),
+                fetch_duration_in_secs: Set(Some(metadata.time_elapsed)),
+                failed_fetch_attempts: Set(Some(metadata.retries)),
                 ..Default::default()
             };
 
-            let conn = SqlxPostgresConnector::from_sqlx_postgres_pool(pool);
-
             let model = active_model.update(&conn).await?;
 
             Ok(model)
         }
-        Err(e) => Err(MetadataJsonTaskError::Fetch(e)),
+        Err(e) => {
+            let active_model = asset_data::ActiveModel {
+                id: Set(download_metadata_info.asset_data_id.clone()),
+                reindex: Set(Some(true)),
+                last_requested_status_code: Set(Some(MetadataJsonFetchResult::Failure)),
+                failed_fetch_attempts: Set(Some(e.retries)),
+                fetch_duration_in_secs: Set(Some(e.time_elapsed)),
+                ..Default::default()
+            };
+
+            active_model.update(&conn).await?;
+
+            Err(MetadataJsonTaskError::Fetch(e.error))
+        }
     }
 }
 

digital_asset_types/src/dao/generated/asset_data.rs

@@ -1,6 +1,7 @@
 //! SeaORM Entity. Generated by sea-orm-codegen 0.9.3
 
 use super::sea_orm_active_enums::ChainMutability;
+use super::sea_orm_active_enums::MetadataJsonFetchResult;
 use super::sea_orm_active_enums::Mutability;
 use sea_orm::entity::prelude::*;
 use serde::{Deserialize, Serialize};
@@ -27,6 +28,9 @@ pub struct Model {
     pub raw_name: Option<Vec<u8>>,
     pub raw_symbol: Option<Vec<u8>>,
     pub base_info_seq: Option<i64>,
+    pub fetch_duration_in_secs: Option<u64>,
+    pub last_requested_status_code: Option<MetadataJsonFetchResult>,
+    pub failed_fetch_attempts: Option<u8>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
@@ -42,6 +46,9 @@ pub enum Column {
     RawName,
     RawSymbol,
     BaseInfoSeq,
+    FetchDurationInSecs,
+    LastRequestedStatusCode,
+    FailedFetchAttempts,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DerivePrimaryKey)]
@@ -74,6 +81,9 @@ impl ColumnTrait for Column {
             Self::RawName => ColumnType::Binary.def().null(),
             Self::RawSymbol => ColumnType::Binary.def().null(),
             Self::BaseInfoSeq => ColumnType::BigInteger.def().null(),
+            Self::FetchDurationInSecs => ColumnType::Unsigned.def().null(),
+            Self::LastRequestedStatusCode => ColumnType::Unsigned.def().null(),
+            Self::FailedFetchAttempts => ColumnType::Unsigned.def().null(),
         }
     }
 }

digital_asset_types/src/dao/generated/sea_orm_active_enums.rs

@@ -163,3 +163,12 @@ pub enum Instruction {
     #[sea_orm(string_value = "verify_creator")]
     VerifyCreator,
 }
+
+#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "fetch_result")]
+pub enum MetadataJsonFetchResult {
+    #[sea_orm(string_value = "success")]
+    Success,
+    #[sea_orm(string_value = "failure")]
+    Failure,
+}

digital_asset_types/tests/common.rs

@@ -86,6 +86,9 @@ pub fn create_asset_data(
             raw_name: Some(metadata.name.into_bytes().to_vec().clone()),
             raw_symbol: Some(metadata.symbol.into_bytes().to_vec().clone()),
             base_info_seq: Some(0),
+            fetch_duration_in_secs: None,
+            last_requested_status_code: None,
+            failed_fetch_attempts: None,
         },
     )
 }

digital_asset_types/tests/json_parsing.rs

@@ -37,6 +37,9 @@ pub async fn parse_onchain_json(json: serde_json::Value) -> Content {
         raw_name: Some(String::from("Handalf").into_bytes().to_vec()),
         raw_symbol: Some(String::from("").into_bytes().to_vec()),
         base_info_seq: Some(0),
+        fetch_duration_in_secs: None,
+        last_requested_status_code: None,
+        failed_fetch_attempts: None,
     };
 
     v1_content_from_json(&asset_data).unwrap()

program_transformers/src/bubblegum/db.rs

@@ -390,6 +390,7 @@ where
         raw_name: ActiveValue::Set(Some(raw_name)),
         raw_symbol: ActiveValue::Set(Some(raw_symbol)),
         base_info_seq: ActiveValue::Set(Some(seq)),
+        ..Default::default()
     };
 
     let mut query = asset_data::Entity::insert(model)

program_transformers/src/mpl_core_program/v1_asset.rs

@@ -195,6 +195,7 @@ pub async fn save_v1_asset<T: ConnectionTrait + TransactionTrait>(
         raw_name: ActiveValue::Set(Some(name.to_vec())),
         raw_symbol: ActiveValue::Set(None),
         base_info_seq: ActiveValue::Set(Some(0)),
+        ..Default::default()
     };
 
     let mut query = asset_data::Entity::insert(asset_data_model)

program_transformers/src/token_metadata/v1_asset.rs

@@ -233,6 +233,7 @@ pub async fn save_v1_asset<T: ConnectionTrait + TransactionTrait>(
         raw_name: ActiveValue::Set(Some(name.to_vec())),
         raw_symbol: ActiveValue::Set(Some(symbol.to_vec())),
         base_info_seq: ActiveValue::Set(Some(0)),
+        ..Default::default()
     };
     let txn = conn.begin().await?;