Updates

Author: hoffmabc

.sqlx/query-02638977689e77d619be18b7361d24006788ddfe96aa974c8d8e4a4eece728ac.json

@@ -25,6 +25,7 @@ redis = { version = "0.23", features = ["tokio-comp"], default-features = false
 reqwest = { version = "0.11", features = ["json", "rustls-tls"], default-features = false }
 http-body-util = "0.1"
 tower-http = { version = "0.5", features = ["cors"] }
+tokio-retry = "0.3"
 
 [lib]
 name = "arch_indexer"

.sqlx/query-20813cb4d560d8cf0249c75937a203e20f6c309debea40fcd0b59a9cd9b60bf4.json

@@ -9,7 +9,7 @@ CREATE TABLE IF NOT EXISTS transactions (
     txid text PRIMARY KEY,
     block_height bigint NOT NULL REFERENCES blocks(height),
     data jsonb NOT NULL,
-    status smallint NOT NULL,
+    status jsonb NOT NULL, -- Change this line
     bitcoin_txids text[],
     created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP
 );

.sqlx/query-3edeaae05a38b76e0bbb075d5a7274d6725dd8f415a2f8ec616c4ec8120f7d69.json

@@ -0,0 +1,2 @@
+ALTER TABLE transactions
+ALTER COLUMN status TYPE JSONB USING to_jsonb(status::text);

.sqlx/query-3f2360a30d1497437a073242c1ca24a216b2f447c5c20d1e7dd998bb17c2392a.json

@@ -1,7 +1,7 @@
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use sqlx::types::JsonValue;
-
+use serde_json::Value;
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Block {
     pub height: i64,
@@ -15,7 +15,7 @@ pub struct Transaction {
     pub txid: String,
     pub block_height: i64,
     pub data: JsonValue,
-    pub status: i32,
+    pub status: Value,
     pub bitcoin_txids: Option<Vec<String>>,
     pub created_at: chrono::NaiveDateTime,
 }

.sqlx/query-87f4e011d12f2e13397657a52c81aac9fe0a9c2ede2130cf7c8f157f0e226720.json

@@ -4,16 +4,12 @@ use chrono::Utc;
 use dashmap::DashMap;
 use futures::stream;
 use futures::StreamExt;
-use sqlx::Postgres;
-use sqlx::QueryBuilder;
 use std::fmt::Write;
 use std::sync::Arc;
 use tracing::error;
-use crate::arch_rpc::Block as ArchBlock;
 use crate::arch_rpc::ArchRpcClient;
 use crate::arch_rpc::Block;
 use crate::db::models::Transaction;
-use chrono::NaiveDateTime;
 use sqlx::PgPool;
 use std::sync::atomic::{AtomicU64, AtomicI64};
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -146,8 +142,6 @@ impl BlockProcessor {
                             .expect("Failed to serialize transaction data");
 
                         let bitcoin_txids: Option<&[String]> = transaction.bitcoin_txids.as_deref();
-
-                        // Convert
 
                         sqlx::query!(
                             r#"
@@ -157,7 +151,7 @@ impl BlockProcessor {
                             transaction.txid,
                             transaction.block_height,
                             serde_json::Value::String(data_json),
-                            transaction.status,
+                            serde_json::Value::String(transaction.status.to_string()),
                             bitcoin_txids,
                             transaction.created_at
                         )
@@ -201,7 +195,7 @@ impl BlockProcessor {
                             txid: txid_clone,
                             block_height: height,
                             data: tx.runtime_transaction,
-                            status: if tx.status == "Processing" { 0 } else { 1 },
+                            status: tx.status,
                             bitcoin_txids: tx.bitcoin_txids,
                             created_at: chrono::Utc::now().naive_utc(),
                         })

.sqlx/query-8e3c032a1f69ca82474c7deee6fa77d7353ef5b5fd79e15129f011b51a855ac5.json

@@ -5,6 +5,9 @@ use std::sync::Arc;
 use tokio::time::{Duration, sleep};
 use tracing::{info, warn, error};
 
+use tokio_retry::strategy::{ExponentialBackoff, jitter};
+use tokio_retry::Retry;
+
 use super::block_processor::BlockProcessor;
 
 pub struct ChainSync {
@@ -42,7 +45,8 @@ impl ChainSync {
             }
 
             if current > target_height {
-                break; // Exit the loop if we're already ahead of the target height
+                // If current is greater than target, continue polling for new blocks
+                continue;
             }
 
             let batch_starts: Vec<_> = (0..self.concurrent_batches)
@@ -56,27 +60,23 @@ impl ChainSync {
                     let end = (start + self.batch_size as i64 - 1).min(target_height);
                     let heights: Vec<_> = (start..=end).collect();
                     let processor = Arc::clone(&self.processor);
-                    
+
                     async move {
-                        match processor.process_blocks_batch(heights).await {
-                            Ok(blocks) => {
-                                // for block in &blocks {
-                                //     info!("Processed block {}", block.height);
-                                // }
-                                Ok(blocks)
-                            }
-                            Err(e) => {
-                                error!("Failed to process batch starting at {}: {:?}", start, e);
-                                Err(e)
-                            }
-                        }
+                        let retry_strategy = ExponentialBackoff::from_millis(10)
+                            .map(jitter) // add jitter to delays
+                            .take(5); // retry up to 5 times
+
+                        Retry::spawn(retry_strategy, || async {
+                            processor.process_blocks_batch(heights.clone()).await
+                        })
+                        .await
                     }
                 })
                 .collect();
 
             // Process batches concurrently
             let results = futures::future::join_all(batch_futures).await;
-            
+
             // Update progress
             for result in results {
                 if let Ok(blocks) = result {
@@ -88,8 +88,6 @@ impl ChainSync {
 
             current = self.current_height.load(Ordering::Relaxed);
         }
-
-        Ok(())
     }
 
     async fn sync_blocks(&self) -> Result<()> {

.sqlx/query-8ec4b70a13e815516ab5e426d75cd9133825f4afbf434fdc241ef70b39b9da49.json

@@ -179,8 +179,6 @@ async fn main() -> Result<()> {
         }
     };
 
-    let current_height = 97935;
-
     // Start the chain sync process
     let sync = ChainSync::new(
         Arc::clone(&processor),