Merge remote-tracking branch 'upstream/main' into refactor/temp-safety-refactor-2

Author: h4sh3d

src/event.rs

@@ -28,6 +28,7 @@ pub trait StateMachineExecutor<
         let event = Event::with(endpoints, runtime.identity(), source, request);
         let sm_display = sm.to_string();
         let sm_name = sm.name();
+        let sm_log_level = sm.log_level();
         if let Some(new_sm) = sm.next(event, runtime)? {
             let new_sm_display = new_sm.to_string();
             // relegate state transitions staying the same to debug
@@ -38,7 +39,8 @@ pub trait StateMachineExecutor<
                     new_sm.bright_green_bold()
                 );
             } else {
-                info!(
+                log!(
+                    new_sm.log_level(),
                     "{} state transition {} -> {}",
                     new_sm.name(),
                     sm_display.red_bold(),
@@ -47,7 +49,8 @@ pub trait StateMachineExecutor<
             }
             Ok(Some(new_sm))
         } else {
-            info!(
+            log!(
+                sm_log_level,
                 "{} state machine ended {} -> {}",
                 sm_name,
                 sm_display.red_bold(),
@@ -70,7 +73,13 @@ where
     where
         Self: Sized;
 
+    /// Return the display name of the state machine
     fn name(&self) -> String;
+
+    /// Return the log level to use for state transitions. Info by default
+    fn log_level(&self) -> log::Level {
+        log::Level::Info
+    }
 }
 
 /// Event changing state machine state, consisting of a certain P2P or RPC `request` sent from some

src/farcasterd/syncer_state_machine.rs

@@ -90,6 +90,10 @@ impl StateMachine<Runtime, Error> for SyncerStateMachine {
     fn name(&self) -> String {
         "Syncer".to_string()
     }
+
+    fn log_level(&self) -> log::Level {
+        log::Level::Debug
+    }
 }
 
 pub struct SyncerStateMachineExecutor {}

src/swapd/runtime.rs

@@ -10,7 +10,6 @@ use super::{
     temporal_safety::TemporalSafety,
     StateReport,
 };
-use crate::service::{Endpoints, Reporter};
 use crate::swapd::temporal_safety::SWEEP_MONERO_THRESHOLD;
 use crate::swapd::Opts;
 use crate::syncerd::bitcoin_syncer::p2wpkh_signed_tx_fee;
@@ -23,6 +22,10 @@ use crate::{
     bus::{BusMsg, Outcome, ServiceBus},
     syncerd::{HeightChanged, TransactionRetrieved, XmrAddressAddendum},
 };
+use crate::{
+    service::{Endpoints, Reporter},
+    syncerd::AddressTransaction,
+};
 use crate::{CtlServer, Error, LogStyle, Service, ServiceConfig, ServiceId};
 
 use std::time::{Duration, SystemTime};
@@ -107,9 +110,7 @@ pub fn run(config: ServiceConfig, opts: Opts) -> Result<(), Error> {
         monero_height: 0,
         bitcoin_height: 0,
         confirmation_bound: 50000,
-        lock_tx_confs: None,
-        cancel_tx_confs: None,
-        buy_tx_confs: None,
+        last_tx_event: none!(),
         network,
         bitcoin_syncer: ServiceId::Syncer(Blockchain::Bitcoin, network),
         monero_syncer: ServiceId::Syncer(Blockchain::Monero, network),
@@ -542,9 +543,23 @@ impl Runtime {
                             self.temporal_safety.acc_finality,
                             endpoints,
                         );
+
+                        // saving requests of interest for later replaying latest event
+                        if let Some(txlabel) = self.syncer_state.tasks.watched_txs.get(id) {
+                            self.syncer_state
+                                .last_tx_event
+                                .insert(*txlabel, request.clone());
+                        }
                     }
 
-                    Event::AddressTransaction(_) => {}
+                    Event::AddressTransaction(AddressTransaction { id, .. }) => {
+                        // saving requests of interest for later replaying latest event
+                        if let Some(txlabel) = self.syncer_state.tasks.watched_addrs.get(id) {
+                            self.syncer_state
+                                .last_tx_event
+                                .insert(*txlabel, request.clone());
+                        }
+                    }
 
                     Event::SweepSuccess(_) => {}
 
@@ -596,19 +611,11 @@ impl Runtime {
                             self.temporal_safety.arb_finality,
                             endpoints,
                         );
-                        let txlabel = self.syncer_state.tasks.watched_txs.get(id);
                         // saving requests of interest for later replaying latest event
-                        match txlabel {
-                            Some(&TxLabel::Lock) => {
-                                self.syncer_state.lock_tx_confs = Some(request.clone());
-                            }
-                            Some(&TxLabel::Cancel) => {
-                                self.syncer_state.cancel_tx_confs = Some(request.clone());
-                            }
-                            Some(&TxLabel::Buy) => {
-                                self.syncer_state.buy_tx_confs = Some(request.clone())
-                            }
-                            _ => {}
+                        if let Some(txlabel) = self.syncer_state.tasks.watched_txs.get(id) {
+                            self.syncer_state
+                                .last_tx_event
+                                .insert(*txlabel, request.clone());
                         }
                     }
 
@@ -624,13 +631,27 @@ impl Runtime {
                             self.temporal_safety.arb_finality,
                             endpoints,
                         );
+                        // saving requests of interest for later replaying latest event
+                        if let Some(txlabel) = self.syncer_state.tasks.watched_txs.get(id) {
+                            self.syncer_state
+                                .last_tx_event
+                                .insert(*txlabel, request.clone());
+                        }
                     }
 
                     Event::TransactionBroadcasted(event) => {
                         self.syncer_state.transaction_broadcasted(event);
                     }
 
-                    Event::AddressTransaction(_) => {}
+                    Event::AddressTransaction(AddressTransaction { id, .. }) => {
+                        // saving requests of interest for later replaying latest event
+                        if let Some(txlabel) = self.syncer_state.tasks.watched_addrs.get(id) {
+                            self.syncer_state
+                                .last_tx_event
+                                .insert(*txlabel, request.clone());
+                        }
+                        self.log_debug(event);
+                    }
 
                     Event::TaskAborted(event) => {
                         self.log_debug(event);
@@ -705,15 +726,9 @@ impl Runtime {
             if let Some(peer_msg) = self.unhandled_peer_message.clone() {
                 self.handle_msg(endpoints, source.clone(), peer_msg)?;
             }
-            // Replay confirmation events to ensure we immediately advance through states that can be skipped
-            if let Some(buy_tx_confs_req) = self.syncer_state.buy_tx_confs.clone() {
-                self.handle_sync(endpoints, source.clone(), buy_tx_confs_req)?;
-            }
-            if let Some(lock_tx_confs_req) = self.syncer_state.lock_tx_confs.clone() {
-                self.handle_sync(endpoints, source.clone(), lock_tx_confs_req)?;
-            }
-            if let Some(cancel_tx_confs_req) = self.syncer_state.cancel_tx_confs.clone() {
-                self.handle_sync(endpoints, source, cancel_tx_confs_req)?;
+            // Replay syncer events to ensure we immediately advance through states that can be skipped
+            for event in self.syncer_state.last_tx_event.clone().values() {
+                self.handle_sync(endpoints, source.clone(), event.clone())?;
             }
         } else if let BusMsg::P2p(peer_msg) = msg {
             self.unhandled_peer_message = Some(peer_msg);