Repository: farcaster-project/farcaster-node
Pull request #244 (2021-11-24 22:17:16)

Syncer: Make abort response dependent on a flag


Patch:
diff --git a/src/swapd/runtime.rs b/src/swapd/runtime.rs
index eedd752d..7a1ecacc 100644
--- a/src/swapd/runtime.rs
+++ b/src/swapd/runtime.rs
@@ -1559,6 +1559,7 @@ impl Runtime {
                         self.state_update(senders, State::Bob(BobState::FinishB))?;
                         let abort_all = Task::Abort(Abort {
                             task_target: TaskTarget::AllTasks,
+                            respond: Boolean::False,
                         });
                         senders.send_to(
                             ServiceBus::Ctl,
@@ -1796,6 +1797,7 @@ impl Runtime {
                                 self.state_update(senders, State::Alice(AliceState::FinishA))?;
                                 let abort_all = Task::Abort(Abort {
                                     task_target: TaskTarget::AllTasks,
+                                    respond: Boolean::False,
                                 });
                                 senders.send_to(
                                     ServiceBus::Ctl,
diff --git a/src/syncerd/bitcoin_syncer.rs b/src/syncerd/bitcoin_syncer.rs
index a972887e..7bb6675e 100644
--- a/src/syncerd/bitcoin_syncer.rs
+++ b/src/syncerd/bitcoin_syncer.rs
@@ -8,7 +8,7 @@ use crate::syncerd::runtime::SyncerdTask;
 use crate::syncerd::syncer_state::AddressTx;
 use crate::syncerd::syncer_state::SyncerState;
 use crate::syncerd::syncer_state::WatchedTransaction;
-use crate::syncerd::types::{AddressAddendum, Task};
+use crate::syncerd::types::{AddressAddendum, Boolean, Task};
 use crate::syncerd::BtcAddressAddendum;
 use crate::syncerd::Event;
 use crate::syncerd::TaskTarget;
@@ -432,8 +432,12 @@ async fn run_syncerd_task_receiver(
                         }
                         Task::Abort(task) => {
                             let mut state_guard = state.lock().await;
+                            let respond = match task.respond {
+                                Boolean::True => true,
+                                Boolean::False => false,
+                            };
                             state_guard
-                                .abort(task.task_target, syncerd_task.source)
+                                .abort(task.task_target, syncerd_task.source, respond)
                                 .await;
                         }
                         Task::BroadcastTransaction(task) => {
@@ -491,7 +495,7 @@ async fn run_syncerd_task_receiver(
                                 error!("Aborting watch address task - unable to decode address addendum");
                                 let mut state_guard = state.lock().await;
                                 state_guard
-                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source)
+                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source, true)
                                     .await;
                             }
                         },
diff --git a/src/syncerd/monero_syncer.rs b/src/syncerd/monero_syncer.rs
index 08c4b933..3092a52d 100644
--- a/src/syncerd/monero_syncer.rs
+++ b/src/syncerd/monero_syncer.rs
@@ -9,7 +9,7 @@ use crate::syncerd::runtime::SyncerdTask;
 use crate::syncerd::syncer_state::create_set;
 use crate::syncerd::syncer_state::AddressTx;
 use crate::syncerd::syncer_state::SyncerState;
-use crate::syncerd::types::{AddressAddendum, SweepAddressAddendum, Task};
+use crate::syncerd::types::{AddressAddendum, Boolean, SweepAddressAddendum, Task};
 use crate::syncerd::Event;
 use crate::syncerd::SyncerServers;
 use crate::syncerd::TaskTarget;
@@ -334,14 +334,18 @@ async fn run_syncerd_task_receiver(
                                 error!("Aborting sweep address task - unable to decode sweep address addendum");
                                 let mut state_guard = state.lock().await;
                                 state_guard
-                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source)
+                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source, true)
                                     .await;
                             }
                         },
                         Task::Abort(task) => {
                             let mut state_guard = state.lock().await;
+                            let respond = match task.respond {
+                                Boolean::True => true,
+                                Boolean::False => false,
+                            };
                             state_guard
-                                .abort(task.task_target, syncerd_task.source)
+                                .abort(task.task_target, syncerd_task.source, respond)
                                 .await;
                         }
                         Task::BroadcastTransaction(task) => {
@@ -366,7 +370,7 @@ async fn run_syncerd_task_receiver(
                                 error!("Aborting watch address task - unable to decode address addendum");
                                 let mut state_guard = state.lock().await;
                                 state_guard
-                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source)
+                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source, true)
                                     .await;
                             }
                         },
diff --git a/src/syncerd/syncer_state.rs b/src/syncerd/syncer_state.rs
index e6939fa4..66b0f459 100644
--- a/src/syncerd/syncer_state.rs
+++ b/src/syncerd/syncer_state.rs
@@ -95,7 +95,12 @@ impl SyncerState {
         self.block_height
     }
 
-    pub async fn abort(&mut self, task_task_id_or_all_tasks: TaskTarget, source: ServiceId) {
+    pub async fn abort(
+        &mut self,
+        task_task_id_or_all_tasks: TaskTarget,
+        source: ServiceId,
+        respond: bool,
+    ) {
         let task_id = match task_task_id_or_all_tasks {
             TaskTarget::AllTasks => None,
             TaskTarget::TaskId(task_id) => Some(task_id),
@@ -229,17 +234,19 @@ impl SyncerState {
             return;
         }
 
-        send_event(
-            &self.tx_event,
-            &mut vec![(
-                Event::TaskAborted(TaskAborted {
-                    id: aborted_ids,
-                    error: None,
-                }),
-                source.clone(),
-            )],
-        )
-        .await;
+        if respond {
+            send_event(
+                &self.tx_event,
+                &mut vec![(
+                    Event::TaskAborted(TaskAborted {
+                        id: aborted_ids,
+                        error: None,
+                    }),
+                    source.clone(),
+                )],
+            )
+            .await;
+        }
     }
 
     pub async fn watch_height(&mut self, task: WatchHeight, source: ServiceId) {
@@ -665,7 +672,9 @@ async fn syncer_state_transaction() {
     let source1 = ServiceId::Syncer(Coin::Bitcoin, Network::Mainnet);
 
     state.watch_transaction(transaction_task_one.clone(), source1.clone());
-    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source1.clone(), true)
+        .await;
     assert!(event_rx.try_recv().is_ok());
 
     state.watch_transaction(transaction_task_one.clone(), source1.clone());
@@ -725,7 +734,9 @@ async fn syncer_state_transaction() {
 
     let source2 = ServiceId::Syncer(Coin::Monero, Network::Mainnet);
     state.watch_transaction(transaction_task_two.clone(), source2.clone());
-    state.abort(TaskTarget::TaskId(0), source2.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source2.clone(), true)
+        .await;
     assert_eq!(state.lifetimes.len(), 3);
     assert_eq!(state.transactions.len(), 2);
     assert_eq!(state.tasks_sources.len(), 3);
@@ -772,7 +783,9 @@ async fn syncer_state_addresses() {
     state
         .watch_address(address_task_two.clone(), source1.clone())
         .unwrap();
-    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source1.clone(), true)
+        .await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.addresses.len(), 0);
@@ -868,7 +881,9 @@ async fn syncer_state_addresses() {
     state
         .watch_address(address_task_two.clone(), source2.clone())
         .unwrap();
-    state.abort(TaskTarget::TaskId(0), source2.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source2.clone(), true)
+        .await;
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.addresses.len(), 1);
@@ -927,7 +942,9 @@ async fn syncer_state_sweep_addresses() {
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.sweep_addresses.len(), 1);
-    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source1.clone(), true)
+        .await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.sweep_addresses.len(), 0);
@@ -959,7 +976,9 @@ async fn syncer_state_height() {
     state
         .watch_height(height_task.clone(), source1.clone())
         .await;
-    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source1.clone(), true)
+        .await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.watch_height.len(), 0);
@@ -998,7 +1017,9 @@ async fn syncer_state_height() {
     state
         .watch_height(another_height_task.clone(), source2.clone())
         .await;
-    state.abort(TaskTarget::TaskId(0), source2.clone()).await;
+    state
+        .abort(TaskTarget::TaskId(0), source2.clone(), true)
+        .await;
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.watch_height.len(), 1);
diff --git a/src/syncerd/types.rs b/src/syncerd/types.rs
index 03a42a50..49406b05 100644
--- a/src/syncerd/types.rs
+++ b/src/syncerd/types.rs
@@ -132,6 +132,7 @@ pub struct SweepBitcoinAddress {
 #[display(Debug)]
 pub struct Abort {
     pub task_target: TaskTarget,
+    pub respond: Boolean,
 }
 
 #[derive(Clone, Debug, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]


Test patch:
diff --git a/tests/functional.rs b/tests/functional.rs
index 714660c8..8d5062d6 100644
--- a/tests/functional.rs
+++ b/tests/functional.rs
@@ -608,6 +608,7 @@ fn bitcoin_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::TaskId(0),
+            respond: Boolean::True,
         }),
         source: SOURCE1.clone(),
     };
@@ -621,6 +622,7 @@ fn bitcoin_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::TaskId(0),
+            respond: Boolean::True,
         }),
         source: SOURCE1.clone(),
     };
@@ -668,6 +670,7 @@ fn bitcoin_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::AllTasks,
+            respond: Boolean::True,
         }),
         source: SOURCE1.clone(),
     };
@@ -681,6 +684,7 @@ fn bitcoin_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::TaskId(0),
+            respond: Boolean::True,
         }),
         source: SOURCE1.clone(),
     };
@@ -1379,6 +1383,7 @@ async fn monero_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::TaskId(0),
+            respond: Boolean::True,
         }),
         source: SOURCE2.clone(),
     };
@@ -1392,6 +1397,7 @@ async fn monero_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::TaskId(0),
+            respond: Boolean::True,
         }),
         source: SOURCE2.clone(),
     };
@@ -1435,6 +1441,7 @@ async fn monero_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::AllTasks,
+            respond: Boolean::True,
         }),
         source: SOURCE2.clone(),
     };
@@ -1448,6 +1455,7 @@ async fn monero_syncer_abort_test() {
     let task = SyncerdTask {
         task: Task::Abort(Abort {
             task_target: TaskTarget::TaskId(0),
+            respond: Boolean::True,
         }),
         source: SOURCE2.clone(),
     };