fix restore transactions after interaption (#336)

near · Aug 9, 2024 · 9c190cd · 9c190cd
1 parent 5aeb8f8
commit 9c190cd
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 31 deletions.
diff --git a/cache-storage/src/lib.rs b/cache-storage/src/lib.rs
@@ -269,17 +269,21 @@ impl TxIndexerCache {
         transaction_key: &readnode_primitives::TransactionKey,
         indexer_execution_outcome_with_receipt: near_indexer_primitives::IndexerExecutionOutcomeWithReceipt,
     ) -> anyhow::Result<()> {
-        let execution_outcome_with_receipt = readnode_primitives::ExecutionOutcomeWithReceipt {
-            execution_outcome: indexer_execution_outcome_with_receipt
-                .execution_outcome
-                .clone(),
-            receipt: indexer_execution_outcome_with_receipt.receipt,
+        // We should not store outcomes for not indexed transactions
+        if let Ok(_) = self.get_tx(transaction_key).await {
+            let execution_outcome_with_receipt = readnode_primitives::ExecutionOutcomeWithReceipt {
+                execution_outcome: indexer_execution_outcome_with_receipt
+                    .execution_outcome
+                    .clone(),
+                receipt: indexer_execution_outcome_with_receipt.receipt,
+            };
+            self.cache_storage
+                .rpush(
+                    format!("outcomes_{}", transaction_key),
+                    borsh::to_vec(&execution_outcome_with_receipt)?,
+                )
+                .await?
         };
-        self.cache_storage
-            .rpush(
-                format!("outcomes_{}", transaction_key),
-                borsh::to_vec(&execution_outcome_with_receipt)?,
-            )
-            .await
+        Ok(())
     }
 }
diff --git a/tx-indexer/src/storage.rs b/tx-indexer/src/storage.rs
@@ -172,26 +172,27 @@ impl CacheStorage {
         transaction_key: &readnode_primitives::TransactionKey,
         indexer_execution_outcome_with_receipt: near_indexer_primitives::IndexerExecutionOutcomeWithReceipt,
     ) -> anyhow::Result<()> {
-        let mut transaction_details = self.get_tx(transaction_key).await?;
-        self.remove_receipt_from_watching_list(
-            &indexer_execution_outcome_with_receipt
-                .receipt
-                .receipt_id
-                .to_string(),
-        )
-        .await?;
-        transaction_details
-            .receipts
-            .push(indexer_execution_outcome_with_receipt.receipt);
-        transaction_details
-            .execution_outcomes
-            .push(indexer_execution_outcome_with_receipt.execution_outcome);
-        let transaction_receipts_watching_count =
-            self.receipts_transaction_count(transaction_key).await?;
-        if transaction_receipts_watching_count == 0 {
-            self.move_tx_to_save(transaction_details.clone()).await?;
-        } else {
-            self.update_tx(transaction_details.clone()).await?;
+        if let Ok(mut transaction_details) = self.get_tx(transaction_key).await {
+            self.remove_receipt_from_watching_list(
+                &indexer_execution_outcome_with_receipt
+                    .receipt
+                    .receipt_id
+                    .to_string(),
+            )
+            .await?;
+            transaction_details
+                .receipts
+                .push(indexer_execution_outcome_with_receipt.receipt);
+            transaction_details
+                .execution_outcomes
+                .push(indexer_execution_outcome_with_receipt.execution_outcome);
+            let transaction_receipts_watching_count =
+                self.receipts_transaction_count(transaction_key).await?;
+            if transaction_receipts_watching_count == 0 {
+                self.move_tx_to_save(transaction_details.clone()).await?;
+            } else {
+                self.update_tx(transaction_details.clone()).await?;
+            }
         }
         Ok(())
     }