TMP

Author: tomxey

crates/iota-indexer/src/apis/write_api.rs

@@ -131,6 +131,11 @@ impl WriteApi {
             // index it safely since we may have already pruned optimistic
             // tables from that epoch. It's very likely that it's already
             // indexed anyway, let's just wait for it.
+            println!(
+                "Epoch change from {} to {}",
+                effects.executed_epoch(),
+                latest_epoch.epoch
+            );
         } else if let (Some(input_objects), Some(output_objects)) = (input_objects, output_objects)
         {
             // We have all needed data, let's optimistically index the tx.
@@ -145,6 +150,7 @@ impl WriteApi {
         } else {
             // TODO: input/output objects are missing, let's create some metric
             // for this
+            println!("Missing in/out objs");
         }
 
         let tx_block_response = self
@@ -163,8 +169,10 @@ impl WriteApi {
         tx_digest: TransactionDigest,
         options: Option<IotaTransactionBlockResponseOptions>,
     ) -> Result<IotaTransactionBlockResponse, IndexerError> {
-        let mut backoff = backoff::ExponentialBackoff::default();
-        backoff.max_elapsed_time = Some(Duration::from_secs(30));
+        let backoff = backoff::ExponentialBackoff {
+            max_elapsed_time: Some(Duration::from_secs(30)),
+            ..Default::default()
+        };
 
         backoff::future::retry(backoff, async || {
             let tx_block_response = self
@@ -181,15 +189,11 @@ impl WriteApi {
                 .pop();
 
             match tx_block_response {
-                Some(tx_block_response) => return Ok(tx_block_response),
-                None => {
-                    return Err(backoff::Error::Transient {
-                        err: IndexerError::PostgresRead(
-                            "Transaction not present in DB".to_string(),
-                        ),
-                        retry_after: None,
-                    });
-                }
+                Some(tx_block_response) => Ok(tx_block_response),
+                None => Err(backoff::Error::Transient {
+                    err: IndexerError::PostgresRead("Transaction not present in DB".to_string()),
+                    retry_after: None,
+                }),
             }
         })
         .await

crates/iota-indexer/src/models/objects.rs

@@ -19,7 +19,7 @@ use serde::de::DeserializeOwned;
 
 use crate::{
     errors::IndexerError,
-    schema::{objects, objects_history, objects_snapshot},
+    schema::{objects, objects_history, objects_snapshot, optimistic_deleted_objects_versions},
     types::{IndexedDeletedObject, IndexedObject, ObjectStatus, owner_to_owner_info},
 };
 
@@ -76,6 +76,13 @@ pub struct StoredObjectSnapshot {
     pub df_kind: Option<i16>,
 }
 
+#[derive(Queryable, Insertable, Debug, Identifiable, Clone, QueryableByName)]
+#[diesel(table_name = optimistic_deleted_objects_versions, primary_key(object_id))]
+pub struct OptimisticDeletedObjectVersion {
+    pub object_id: Vec<u8>,
+    pub object_version: i64,
+}
+
 impl From<IndexedObject> for StoredObjectSnapshot {
     fn from(o: IndexedObject) -> Self {
         let IndexedObject {

crates/iota-indexer/tests/rpc-tests/write_api.rs

@@ -354,6 +354,10 @@ fn test_consecutive_modifications_of_owned_object() -> Result<(), anyhow::Error>
 
 #[test]
 fn test_consecutive_wrap_unwrap() -> Result<(), anyhow::Error> {
+    let _guard = telemetry_subscribers::TelemetryConfig::new()
+        .with_env()
+        .init();
+
     let ApiTestSetup {
         runtime,
         store,
@@ -363,7 +367,7 @@ fn test_consecutive_wrap_unwrap() -> Result<(), anyhow::Error> {
     runtime.block_on(async move {
         indexer_wait_for_checkpoint(store, 1).await;
         let (sender, sender_kp): (_, AccountKeyPair) = get_key_pair();
-        let consecutive_updates = 50;
+        let consecutive_updates = 2500;
 
         let gas = cluster
             .fund_address_and_return_gas(
@@ -546,7 +550,7 @@ fn test_repeatedly_update_display() {
     } = ApiTestSetup::get_or_init();
 
     runtime.block_on(async {
-        let consecutive_updates = 50;
+        let consecutive_updates = 150;
         indexer_wait_for_checkpoint(store, 1).await;
 
         let (sender, sender_kp): (_, AccountKeyPair) = get_key_pair();