0.7.31

kb_lib/src/local_pipeline_replay.rs

@@ -16,6 +16,8 @@ pub struct LocalPipelineReplayConfig {
     pub refresh_missing_token_metadata: bool,
     /// Maximum number of missing token metadata rows to resolve.
     pub token_metadata_limit: std::option::Option<i64>,
+    /// Whether locally replayed market materialization tables are reset before replay.
+    pub reset_market_materialization_before_replay: bool,
 }
 
 impl Default for LocalPipelineReplayConfig {
@@ -24,6 +26,7 @@ impl Default for LocalPipelineReplayConfig {
             limit: Some(10_000),
             refresh_missing_token_metadata: false,
             token_metadata_limit: Some(250),
+            reset_market_materialization_before_replay: true,
         };
     }
 }
@@ -71,6 +74,8 @@ pub struct LocalPipelineReplayResult {
     pub token_metadata_updated_count: usize,
     /// Number of pair symbols updated after replay.
     pub pair_symbol_updated_count: usize,
+    /// Number of derived market materialization rows deleted before replay.
+    pub reset_market_materialization_deleted_count: u64,
     /// Number of errors outside per-signature replay.
     pub global_error_count: usize,
 }
@@ -120,6 +125,19 @@ impl LocalPipelineReplayService {
             Ok(signatures) => signatures,
             Err(error) => return Err(error),
         };
+        let mut reset_market_materialization_deleted_count = 0_u64;
+        if config.reset_market_materialization_before_replay {
+            let reset_result =
+                crate::query_trade_market_materialization_delete_all(self.database.as_ref()).await;
+            reset_market_materialization_deleted_count = match reset_result {
+                Ok(deleted_count) => deleted_count,
+                Err(error) => return Err(error),
+            };
+            tracing::debug!(
+                deleted_count = reset_market_materialization_deleted_count,
+                "local pipeline replay reset market materialization tables"
+            );
+        }
         let dex_decode = crate::DexDecodeService::new(self.database.clone());
         let dex_detect = crate::DexDetectService::new(self.database.clone());
         let trade_aggregation = crate::TradeAggregationService::new(self.database.clone());
@@ -130,6 +148,7 @@ impl LocalPipelineReplayService {
             crate::TransactionClassificationService::new(self.database.clone());
         let mut result = LocalPipelineReplayResult {
             selected_transaction_count: signatures.len(),
+            reset_market_materialization_deleted_count,
             ..Default::default()
         };
         for signature in signatures {