Repository: farcaster-project/farcaster-node
Pull request #198 (2021-11-18 15:31:38)

Syncer abort all tasks from a certain source


Patch:
diff --git a/src/syncerd/bitcoin_syncer.rs b/src/syncerd/bitcoin_syncer.rs
index 51e609d3..a972887e 100644
--- a/src/syncerd/bitcoin_syncer.rs
+++ b/src/syncerd/bitcoin_syncer.rs
@@ -11,6 +11,7 @@ use crate::syncerd::syncer_state::WatchedTransaction;
 use crate::syncerd::types::{AddressAddendum, Task};
 use crate::syncerd::BtcAddressAddendum;
 use crate::syncerd::Event;
+use crate::syncerd::TaskTarget;
 use crate::syncerd::TransactionBroadcasted;
 use crate::ServiceId;
 use crate::{error::Error, syncerd::syncer_state::create_set};
@@ -431,7 +432,9 @@ async fn run_syncerd_task_receiver(
                         }
                         Task::Abort(task) => {
                             let mut state_guard = state.lock().await;
-                            state_guard.abort(task.id, syncerd_task.source).await;
+                            state_guard
+                                .abort(task.task_target, syncerd_task.source)
+                                .await;
                         }
                         Task::BroadcastTransaction(task) => {
                             // TODO: match error and emit event with fail code
@@ -471,7 +474,7 @@ async fn run_syncerd_task_receiver(
                                             source: syncerd_task.source,
                                         })
                                         .await
-                                        .expect("error sending transactio broadcast event");
+                                        .expect("error sending transaction broadcast event");
                                     error!("failed to broadcast tx: {}", e.err());
                                 }
                             }
@@ -487,7 +490,9 @@ 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(task.id, syncerd_task.source).await;
+                                state_guard
+                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source)
+                                    .await;
                             }
                         },
                         Task::WatchHeight(task) => {
diff --git a/src/syncerd/monero_syncer.rs b/src/syncerd/monero_syncer.rs
index 2850f29e..1634b6e2 100644
--- a/src/syncerd/monero_syncer.rs
+++ b/src/syncerd/monero_syncer.rs
@@ -12,6 +12,7 @@ use crate::syncerd::syncer_state::SyncerState;
 use crate::syncerd::types::{AddressAddendum, SweepAddressAddendum, Task};
 use crate::syncerd::Event;
 use crate::syncerd::SyncerServers;
+use crate::syncerd::TaskTarget;
 use crate::syncerd::TransactionBroadcasted;
 use crate::syncerd::XmrAddressAddendum;
 use internet2::zmqsocket::{Connection, ZmqType};
@@ -340,12 +341,16 @@ 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(task.id, syncerd_task.source).await;
+                                state_guard
+                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source)
+                                    .await;
                             }
                         },
                         Task::Abort(task) => {
                             let mut state_guard = state.lock().await;
-                            state_guard.abort(task.id, syncerd_task.source).await;
+                            state_guard
+                                .abort(task.task_target, syncerd_task.source)
+                                .await;
                         }
                         Task::BroadcastTransaction(task) => {
                             error!("broadcast transaction not available for Monero");
@@ -368,7 +373,9 @@ 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(task.id, syncerd_task.source).await;
+                                state_guard
+                                    .abort(TaskTarget::TaskId(task.id), syncerd_task.source)
+                                    .await;
                             }
                         },
                         Task::WatchHeight(task) => {
diff --git a/src/syncerd/syncer_state.rs b/src/syncerd/syncer_state.rs
index 024b3314..0e6a4669 100644
--- a/src/syncerd/syncer_state.rs
+++ b/src/syncerd/syncer_state.rs
@@ -2,6 +2,7 @@ use crate::farcaster_core::{blockchain::Network, consensus::Decodable};
 use crate::rpc::request::SyncerdBridgeEvent;
 use crate::syncerd::opts::Coin;
 use crate::syncerd::runtime::SyncerdTask;
+use crate::syncerd::TaskTarget;
 use crate::Error;
 use crate::ServiceId;
 use bitcoin::{hashes::Hash, Txid};
@@ -16,18 +17,38 @@ use crate::serde::Deserialize;
 use crate::syncerd::*;
 use hex;
 
+#[derive(Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Hash, Display)]
+#[display(Debug)]
+pub struct InternalId(u32);
+
+impl From<TaskCounter> for InternalId {
+    fn from(counter: TaskCounter) -> Self {
+        Self(counter.0)
+    }
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Display, Debug)]
+#[display(Debug)]
+struct TaskCounter(pub u32);
+
+impl TaskCounter {
+    fn increment(&mut self) {
+        self.0 += 1;
+    }
+}
+
 pub struct SyncerState {
     block_height: u64,
     block_hash: Vec<u8>,
-    tasks_sources: HashMap<u32, ServiceId>,
-    watch_height: HashMap<u32, WatchHeight>,
-    lifetimes: HashMap<u64, HashSet<u32>>,
-    pub addresses: HashMap<u32, AddressTransactions>,
-    pub transactions: HashMap<u32, WatchedTransaction>,
-    pub unseen_transactions: HashSet<u32>,
-    pub sweep_addresses: HashMap<u32, SweepAddress>,
+    tasks_sources: HashMap<InternalId, ServiceId>,
+    watch_height: HashMap<InternalId, WatchHeight>,
+    lifetimes: HashMap<u64, HashSet<InternalId>>,
+    pub addresses: HashMap<InternalId, AddressTransactions>,
+    pub transactions: HashMap<InternalId, WatchedTransaction>,
+    pub unseen_transactions: HashSet<InternalId>,
+    pub sweep_addresses: HashMap<InternalId, SweepAddress>,
     tx_event: TokioSender<SyncerdBridgeEvent>,
-    task_count: u32,
+    task_count: TaskCounter,
 }
 
 #[derive(Clone, Debug)]
@@ -66,7 +87,7 @@ impl SyncerState {
             unseen_transactions: HashSet::new(),
             sweep_addresses: HashMap::new(),
             tx_event,
-            task_count: 0,
+            task_count: TaskCounter(0),
         }
     }
 
@@ -74,144 +95,146 @@ impl SyncerState {
         self.block_height
     }
 
-    pub async fn abort(&mut self, task_id: u32, source: ServiceId) {
+    pub async fn abort(&mut self, task_task_id_or_all_tasks: TaskTarget, source: ServiceId) {
+        let task_id = match task_task_id_or_all_tasks {
+            TaskTarget::AllTasks => None,
+            TaskTarget::TaskId(task_id) => Some(task_id),
+        };
+
+        let mut aborted_ids: Vec<u32> = vec![];
         // check addresses tasks
-        let ids: Vec<u32> = self
+        let ids: Vec<(InternalId, u32)> = self
             .addresses
             .iter()
             .filter_map(|(id, address_transaction)| {
-                if address_transaction.task.id == task_id {
-                    Some(*id)
+                if task_id.is_none() || address_transaction.task.id == task_id.unwrap() {
+                    Some((*id, address_transaction.task.id))
                 } else {
                     None
                 }
             })
             .collect();
-        for id in ids.iter() {
-            if let Some(source_id) = self.tasks_sources.get(id) {
-                if *source_id == source {
-                    self.remove_address(id);
-                    send_event(
-                        &self.tx_event,
-                        &mut vec![(
-                            Event::TaskAborted(TaskAborted {
-                                id: task_id,
-                                error: None,
-                            }),
-                            source.clone(),
-                        )],
-                    )
-                    .await;
-                    return;
-                }
-            }
-        }
+
+        aborted_ids.append(
+            &mut ids
+                .iter()
+                .filter_map(|(internal_id, found_task_id)| {
+                    if let Some(source_id) = self.tasks_sources.get(internal_id) {
+                        if *source_id == source {
+                            self.remove_address(internal_id);
+                            return Some(*found_task_id);
+                        }
+                    }
+                    None
+                })
+                .collect(),
+        );
 
         // check transactions tasks
-        let ids: Vec<u32> = self
+        let ids: Vec<(InternalId, u32)> = self
             .transactions
             .iter()
             .filter_map(|(id, watched_transaction)| {
-                if watched_transaction.task.id == task_id {
-                    Some(*id)
+                if task_id.is_none() || watched_transaction.task.id == task_id.unwrap() {
+                    Some((*id, watched_transaction.task.id))
                 } else {
                     None
                 }
             })
             .collect();
-        for id in ids.iter() {
-            if let Some(source_id) = self.tasks_sources.get(id) {
-                if *source_id == source {
-                    self.remove_transaction(id);
-                    send_event(
-                        &self.tx_event,
-                        &mut vec![(
-                            Event::TaskAborted(TaskAborted {
-                                id: task_id,
-                                error: None,
-                            }),
-                            source.clone(),
-                        )],
-                    )
-                    .await;
-                    return;
-                }
-            }
-        }
+        aborted_ids.append(
+            &mut ids
+                .iter()
+                .filter_map(|(internal_id, found_task_id)| {
+                    if let Some(source_id) = self.tasks_sources.get(internal_id) {
+                        if *source_id == source {
+                            self.remove_transaction(internal_id);
+                            return Some(*found_task_id);
+                        }
+                    }
+                    None
+                })
+                .collect(),
+        );
 
         // check height tasks
-        let ids: Vec<u32> = self
+        let ids: Vec<(InternalId, u32)> = self
             .watch_height
             .iter()
             .filter_map(|(id, watch_height)| {
-                if watch_height.id == task_id {
-                    Some(*id)
+                if task_id.is_none() || watch_height.id == task_id.unwrap() {
+                    Some((*id, watch_height.id))
                 } else {
                     None
                 }
             })
             .collect();
-        for id in ids.iter() {
-            if let Some(source_id) = self.tasks_sources.get(id) {
-                if *source_id == source {
-                    self.remove_height(id);
-                    send_event(
-                        &self.tx_event,
-                        &mut vec![(
-                            Event::TaskAborted(TaskAborted {
-                                id: task_id,
-                                error: None,
-                            }),
-                            source.clone(),
-                        )],
-                    )
-                    .await;
-                    return;
-                }
-            }
-        }
+        aborted_ids.append(
+            &mut ids
+                .iter()
+                .filter_map(|(internal_id, found_task_id)| {
+                    if let Some(source_id) = self.tasks_sources.get(internal_id) {
+                        if *source_id == source {
+                            self.remove_height(internal_id);
+                            return Some(*found_task_id);
+                        }
+                    }
+                    None
+                })
+                .collect(),
+        );
 
         // check sweep address tasks
-        let ids: Vec<u32> = self
+        let ids: Vec<(InternalId, u32)> = self
             .sweep_addresses
             .iter()
             .filter_map(|(id, sweep_address)| {
-                if sweep_address.id == task_id {
-                    Some(*id)
+                if task_id.is_none() || sweep_address.id == task_id.unwrap() {
+                    Some((*id, sweep_address.id))
                 } else {
                     None
                 }
             })
             .collect();
-        for id in ids.iter() {
-            if let Some(source_id) = self.tasks_sources.get(id) {
-                if *source_id == source {
-                    self.remove_sweep_address(id);
-                    send_event(
-                        &self.tx_event,
-                        &mut vec![(
-                            Event::TaskAborted(TaskAborted {
-                                id: task_id,
-                                error: None,
-                            }),
-                            source.clone(),
-                        )],
-                    )
-                    .await;
-                    return;
-                }
-            }
+        aborted_ids.append(
+            &mut ids
+                .iter()
+                .filter_map(|(internal_id, found_task_id)| {
+                    if let Some(source_id) = self.tasks_sources.get(internal_id) {
+                        if *source_id == source {
+                            self.remove_sweep_address(internal_id);
+                            return Some(*found_task_id);
+                        }
+                    }
+                    None
+                })
+                .collect(),
+        );
+
+        if aborted_ids.is_empty() {
+            send_event(
+                &self.tx_event,
+                &mut vec![(
+                    Event::TaskAborted(TaskAborted {
+                        id: vec![],
+                        error: Some(format!(
+                            "abort failed, task from source {} not found",
+                            source
+                        )),
+                    }),
+                    source.clone(),
+                )],
+            )
+            .await;
+            return;
         }
 
         send_event(
             &self.tx_event,
             &mut vec![(
                 Event::TaskAborted(TaskAborted {
-                    id: task_id,
-                    error: Some(format!(
-                        "abort failed, task with id {} from source {} not found",
-                        task_id, source
-                    )),
+                    id: aborted_ids,
+                    error: None,
                 }),
                 source.clone(),
             )],
@@ -221,15 +244,17 @@ impl SyncerState {
 
     pub async fn watch_height(&mut self, task: WatchHeight, source: ServiceId) {
         // increment the count to use it as a unique internal id
-        self.task_count += 1;
+        self.task_count.increment();
         // This is technically valid behavior; immediately prune the task for being past
         // its lifetime by never inserting it
-        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count) {
+        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count.into()) {
             error!("{}", e);
             return;
         }
-        self.watch_height.insert(self.task_count, task.clone());
-        self.tasks_sources.insert(self.task_count, source.clone());
+        self.watch_height
+            .insert(self.task_count.into(), task.clone());
+        self.tasks_sources
+            .insert(self.task_count.into(), source.clone());
 
         if self.block_height != 0 {
             send_event(
@@ -249,28 +274,28 @@ impl SyncerState {
 
     pub fn watch_address(&mut self, task: WatchAddress, source: ServiceId) -> Result<(), Error> {
         // increment the count to use it as a unique internal id
-        self.task_count += 1;
-        self.add_lifetime(task.lifetime, self.task_count)?;
-        self.tasks_sources.insert(self.task_count, source);
+        self.task_count.increment();
+        self.add_lifetime(task.lifetime, self.task_count.into())?;
+        self.tasks_sources.insert(self.task_count.into(), source);
         let address_txs = AddressTransactions {
             task,
             known_txs: none!(),
         };
-        self.addresses.insert(self.task_count, address_txs);
+        self.addresses.insert(self.task_count.into(), address_txs);
         Ok(())
     }
 
     pub fn watch_transaction(&mut self, task: WatchTransaction, source: ServiceId) {
         // increment the count to use it as a unique internal id
-        self.task_count += 1;
+        self.task_count.increment();
 
-        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count) {
+        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count.into()) {
             error!("{}", e);
             return;
         };
-        self.tasks_sources.insert(self.task_count, source);
+        self.tasks_sources.insert(self.task_count.into(), source);
         self.transactions.insert(
-            self.task_count,
+            self.task_count.into(),
             WatchedTransaction {
                 task: task.clone(),
                 transaction_confirmations: TransactionConfirmations {
@@ -280,19 +305,19 @@ impl SyncerState {
                 },
             },
         );
-        self.unseen_transactions.insert(self.task_count);
+        self.unseen_transactions.insert(self.task_count.into());
     }
 
     pub fn sweep_address(&mut self, task: SweepAddress, source: ServiceId) {
-        self.task_count += 1;
+        self.task_count.increment();
         // This is technically valid behavior; immediately prune the task for being past
         // its lifetime by never inserting it
-        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count) {
+        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count.into()) {
             error!("{}", e);
             return;
         }
-        self.sweep_addresses.insert(self.task_count, task);
-        self.tasks_sources.insert(self.task_count, source);
+        self.sweep_addresses.insert(self.task_count.into(), task);
+        self.tasks_sources.insert(self.task_count.into(), source);
     }
 
     pub async fn change_height(&mut self, height: u64, block: Vec<u8>) {
@@ -336,9 +361,9 @@ impl SyncerState {
         );
 
         fn inner(
-            addresses: &mut HashMap<u32, AddressTransactions>,
+            addresses: &mut HashMap<InternalId, AddressTransactions>,
             events: &mut Vec<(Event, ServiceId)>,
-            tasks_sources: &mut HashMap<u32, ServiceId>,
+            tasks_sources: &mut HashMap<InternalId, ServiceId>,
             address_addendum: AddressAddendum,
             txs: HashSet<AddressTx>,
         ) {
@@ -421,10 +446,10 @@ impl SyncerState {
         );
 
         fn inner(
-            transactions: &mut HashMap<u32, WatchedTransaction>,
-            unseen_transactions: &mut HashSet<u32>,
+            transactions: &mut HashMap<InternalId, WatchedTransaction>,
+            unseen_transactions: &mut HashSet<InternalId>,
             events: &mut Vec<(Event, ServiceId)>,
-            tasks_sources: &mut HashMap<u32, ServiceId>,
+            tasks_sources: &mut HashMap<InternalId, ServiceId>,
             tx_id: Vec<u8>,
             block_hash: Option<Vec<u8>>,
             confirmations: Option<u32>,
@@ -482,7 +507,7 @@ impl SyncerState {
         send_event(&self.tx_event, &mut events).await;
     }
 
-    pub async fn success_sweep(&mut self, id: &u32, txids: Vec<Vec<u8>>) {
+    pub async fn success_sweep(&mut self, id: &InternalId, txids: Vec<Vec<u8>>) {
         if let Some(sweep_address) = self.sweep_addresses.get(id) {
             send_event(
                 &self.tx_event,
@@ -519,7 +544,7 @@ impl SyncerState {
         }
     }
 
-    fn add_lifetime(&mut self, lifetime: u64, id: u32) -> Result<(), Error> {
+    fn add_lifetime(&mut self, lifetime: u64, id: InternalId) -> Result<(), Error> {
         if lifetime < self.block_height {
             return Err(Error::Farcaster("task lifetime expired".to_string()));
         }
@@ -549,7 +574,7 @@ impl SyncerState {
         }
     }
 
-    fn remove_transaction(&mut self, id: &u32) {
+    fn remove_transaction(&mut self, id: &InternalId) {
         if let Some(watch_transactions) = self.transactions.get(id) {
             if let Some(ids) = self.lifetimes.get_mut(&watch_transactions.task.lifetime) {
                 ids.remove(id);
@@ -563,7 +588,7 @@ impl SyncerState {
         self.tasks_sources.remove(id);
     }
 
-    fn remove_address(&mut self, id: &u32) {
+    fn remove_address(&mut self, id: &InternalId) {
         if let Some(address_transactions) = self.addresses.get(id) {
             if let Some(ids) = self.lifetimes.get_mut(&address_transactions.task.lifetime) {
                 ids.remove(id);
@@ -576,7 +601,7 @@ impl SyncerState {
         self.tasks_sources.remove(id);
     }
 
-    fn remove_height(&mut self, id: &u32) {
+    fn remove_height(&mut self, id: &InternalId) {
         if let Some(watch_height) = self.watch_height.get(id) {
             if let Some(ids) = self.lifetimes.get_mut(&watch_height.lifetime) {
                 ids.remove(id);
@@ -589,7 +614,7 @@ impl SyncerState {
         self.tasks_sources.remove(id);
     }
 
-    fn remove_sweep_address(&mut self, id: &u32) {
+    fn remove_sweep_address(&mut self, id: &InternalId) {
         if let Some(sweep_address) = self.sweep_addresses.get(id) {
             if let Some(ids) = self.lifetimes.get_mut(&sweep_address.lifetime) {
                 ids.remove(id);
@@ -640,7 +665,7 @@ async fn syncer_state_transaction() {
     let source1 = ServiceId::Syncer(Coin::Bitcoin, Network::Mainnet);
 
     state.watch_transaction(transaction_task_one.clone(), source1.clone());
-    state.abort(0, source1.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
     assert!(event_rx.try_recv().is_ok());
 
     state.watch_transaction(transaction_task_one.clone(), source1.clone());
@@ -700,7 +725,7 @@ async fn syncer_state_transaction() {
 
     let source2 = ServiceId::Syncer(Coin::Monero, Network::Mainnet);
     state.watch_transaction(transaction_task_two.clone(), source2.clone());
-    state.abort(0, source2.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source2.clone()).await;
     assert_eq!(state.lifetimes.len(), 3);
     assert_eq!(state.transactions.len(), 2);
     assert_eq!(state.tasks_sources.len(), 3);
@@ -747,7 +772,7 @@ async fn syncer_state_addresses() {
     state
         .watch_address(address_task_two.clone(), source1.clone())
         .unwrap();
-    state.abort(0, source1.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.addresses.len(), 0);
@@ -843,7 +868,7 @@ async fn syncer_state_addresses() {
     state
         .watch_address(address_task_two.clone(), source2.clone())
         .unwrap();
-    state.abort(0, source2.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source2.clone()).await;
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.addresses.len(), 1);
@@ -899,7 +924,7 @@ 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(0, source1.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.sweep_addresses.len(), 0);
@@ -909,7 +934,7 @@ 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.success_sweep(&2, vec![vec![0]]).await;
+    state.success_sweep(&InternalId(2), vec![vec![0]]).await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.sweep_addresses.len(), 0);
@@ -931,7 +956,7 @@ async fn syncer_state_height() {
     state
         .watch_height(height_task.clone(), source1.clone())
         .await;
-    state.abort(0, source1.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.watch_height.len(), 0);
@@ -970,7 +995,7 @@ async fn syncer_state_height() {
     state
         .watch_height(another_height_task.clone(), source2.clone())
         .await;
-    state.abort(0, source2.clone()).await;
+    state.abort(TaskTarget::TaskId(0), source2.clone()).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 a47997d7..a4c4d043 100644
--- a/src/syncerd/types.rs
+++ b/src/syncerd/types.rs
@@ -60,7 +60,13 @@ pub struct SweepBitcoinAddress {
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
 #[display(Debug)]
 pub struct Abort {
-    pub id: u32,
+    pub task_target: TaskTarget,
+}
+
+#[derive(Clone, Debug, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+pub enum TaskTarget {
+    TaskId(u32),
+    AllTasks,
 }
 
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
@@ -127,7 +133,7 @@ pub enum Task {
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
 #[display(Debug)]
 pub struct TaskAborted {
-    pub id: u32,
+    pub id: Vec<u32>,
     pub error: Option<String>,
 }
 


Test patch:
diff --git a/tests/functional.rs b/tests/functional.rs
index bee8d1b1..9c68e0ac 100644
--- a/tests/functional.rs
+++ b/tests/functional.rs
@@ -11,6 +11,7 @@ use farcaster_node::syncerd::SweepAddress;
 use farcaster_node::syncerd::SweepAddressAddendum;
 use farcaster_node::syncerd::SweepXmrAddress;
 use farcaster_node::syncerd::SyncerServers;
+use farcaster_node::syncerd::TaskTarget;
 use farcaster_node::syncerd::XmrAddressAddendum;
 use farcaster_node::ServiceId;
 use internet2::transport::MAX_FRAME_SIZE;
@@ -569,6 +570,17 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
     assert_transaction_confirmations(request, Some(0), vec![0]);
 }
 
+/*
+We test for the following scenarios in the abort tests:
+
+- Submit a WatchTransaction task for a non-existing transaction, receive a confirmation events
+
+- Submit an Abort task for the transaction, receive success
+
+- Submit an Abort task for the transaction, receive an error
+
+- Submit two WatchTransaction tasks, abort them both and receive both their aborted id's.
+*/
 #[test]
 #[ignore]
 fn bitcoin_syncer_abort_test() {
@@ -593,7 +605,9 @@ fn bitcoin_syncer_abort_test() {
     assert_transaction_confirmations(request, None, vec![0]);
 
     let task = SyncerdTask {
-        task: Task::Abort(Abort { id: 0 }),
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::TaskId(0),
+        }),
         source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
@@ -601,10 +615,72 @@ fn bitcoin_syncer_abort_test() {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("task aborted");
     let request = get_request_from_message(message);
-    assert_task_aborted(request, None, 0);
+    assert_task_aborted(request, None, vec![0]);
 
     let task = SyncerdTask {
-        task: Task::Abort(Abort { id: 0 }),
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::TaskId(0),
+        }),
+        source: SOURCE1.clone(),
+    };
+    tx.send(task).unwrap();
+    println!("waiting for task aborted message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("task aborted");
+    let request = get_request_from_message(message);
+    assert_task_aborted(
+        request,
+        Some("abort failed, task from source syncer<Bitcoin> not found".to_string()),
+        vec![],
+    );
+
+    let task = SyncerdTask {
+        task: Task::WatchTransaction(WatchTransaction {
+            id: 0,
+            lifetime: blocks + 10,
+            hash: vec![0; 32],
+            confirmation_bound: 2,
+        }),
+        source: SOURCE1.clone(),
+    };
+    tx.send(task).unwrap();
+    println!("waiting for confirmation");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("received confirmation");
+    let request = get_request_from_message(message);
+    assert_transaction_confirmations(request, None, vec![0]);
+    let task = SyncerdTask {
+        task: Task::WatchTransaction(WatchTransaction {
+            id: 1,
+            lifetime: blocks + 10,
+            hash: vec![0; 32],
+            confirmation_bound: 2,
+        }),
+        source: SOURCE1.clone(),
+    };
+    tx.send(task).unwrap();
+    println!("waiting for confirmation");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("received confirmation");
+    let request = get_request_from_message(message);
+    assert_transaction_confirmations(request, None, vec![0]);
+    let task = SyncerdTask {
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::AllTasks,
+        }),
+        source: SOURCE1.clone(),
+    };
+    tx.send(task).unwrap();
+    println!("waiting for task aborted message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("task aborted");
+    let request = get_request_from_message(message);
+    assert_task_aborted(request, None, vec![0, 1]);
+
+    let task = SyncerdTask {
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::TaskId(0),
+        }),
         source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
@@ -614,8 +690,8 @@ fn bitcoin_syncer_abort_test() {
     let request = get_request_from_message(message);
     assert_task_aborted(
         request,
-        Some("abort failed, task with id 0 from source syncer<Bitcoin> not found".to_string()),
-        0,
+        Some("abort failed, task from source syncer<Bitcoin> not found".to_string()),
+        vec![],
     );
 }
 
@@ -1266,6 +1342,17 @@ async fn monero_syncer_transaction_test() {
     assert_transaction_confirmations(request, Some(1), block_hash);
 }
 
+/*
+We test for the following scenarios in the abort tests:
+
+- Submit a WatchTransaction task for a non-existing transaction, receive a confirmation events
+
+- Submit an Abort task for the transaction, receive success
+
+- Submit an Abort task for the transaction, receive an error
+
+- Submit two WatchTransaction tasks, abort them both and receive both their aborted id's.
+*/
 #[tokio::test]
 #[ignore]
 async fn monero_syncer_abort_test() {
@@ -1289,7 +1376,9 @@ async fn monero_syncer_abort_test() {
     assert_transaction_confirmations(request, None, vec![0]);
 
     let task = SyncerdTask {
-        task: Task::Abort(Abort { id: 0 }),
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::TaskId(0),
+        }),
         source: SOURCE2.clone(),
     };
     tx.send(task).unwrap();
@@ -1297,10 +1386,12 @@ async fn monero_syncer_abort_test() {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("task aborted");
     let request = get_request_from_message(message);
-    assert_task_aborted(request, None, 0);
+    assert_task_aborted(request, None, vec![0]);
 
     let task = SyncerdTask {
-        task: Task::Abort(Abort { id: 0 }),
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::TaskId(0),
+        }),
         source: SOURCE2.clone(),
     };
     tx.send(task).unwrap();
@@ -1310,8 +1401,64 @@ async fn monero_syncer_abort_test() {
     let request = get_request_from_message(message);
     assert_task_aborted(
         request,
-        Some("abort failed, task with id 0 from source syncer<Monero> not found".to_string()),
-        0,
+        Some("abort failed, task from source syncer<Monero> not found".to_string()),
+        vec![],
+    );
+
+    let task = SyncerdTask {
+        task: Task::WatchTransaction(WatchTransaction {
+            id: 0,
+            lifetime: blocks + 10,
+            hash: vec![0; 32],
+            confirmation_bound: 2,
+        }),
+        source: SOURCE2.clone(),
+    };
+    tx.send(task).unwrap();
+    let message = rx_event.recv_multipart(0).unwrap();
+    let request = get_request_from_message(message);
+    assert_transaction_confirmations(request, None, vec![0]);
+    let task = SyncerdTask {
+        task: Task::WatchTransaction(WatchTransaction {
+            id: 1,
+            lifetime: blocks + 10,
+            hash: vec![0; 32],
+            confirmation_bound: 2,
+        }),
+        source: SOURCE2.clone(),
+    };
+    tx.send(task).unwrap();
+    let message = rx_event.recv_multipart(0).unwrap();
+    let request = get_request_from_message(message);
+    assert_transaction_confirmations(request, None, vec![0]);
+    let task = SyncerdTask {
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::AllTasks,
+        }),
+        source: SOURCE2.clone(),
+    };
+    tx.send(task).unwrap();
+    println!("waiting for task aborted message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("task aborted");
+    let request = get_request_from_message(message);
+    assert_task_aborted(request, None, vec![0, 1]);
+
+    let task = SyncerdTask {
+        task: Task::Abort(Abort {
+            task_target: TaskTarget::TaskId(0),
+        }),
+        source: SOURCE2.clone(),
+    };
+    tx.send(task).unwrap();
+    println!("waiting for task aborted message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("task aborted");
+    let request = get_request_from_message(message);
+    assert_task_aborted(
+        request,
+        Some("abort failed, task from source syncer<Monero> not found".to_string()),
+        vec![],
     );
 }
 
@@ -1517,11 +1664,18 @@ fn assert_transaction_confirmations(
     }
 }
 
-fn assert_task_aborted(request: Request, expected_error: Option<String>, expected_id: u32) {
+fn assert_task_aborted(
+    request: Request,
+    expected_error: Option<String>,
+    mut expected_id: Vec<u32>,
+) {
     match request {
         Request::SyncerdBridgeEvent(event) => match event.event {
-            Event::TaskAborted(task_aborted) => {
-                assert_eq!(task_aborted.id, expected_id);
+            Event::TaskAborted(mut task_aborted) => {
+                assert_eq!(
+                    &task_aborted.id.sort_unstable(),
+                    &expected_id.sort_unstable()
+                );
                 assert_eq!(task_aborted.error, expected_error);
             }
             _ => {