drk follower implementation

src/darkfi/follower.rs

@@ -27,7 +27,10 @@ impl<C: OtherChainClient> DrkFollower<C> {
 
 impl<C: OtherChainClient + Send + Sync> Follower for DrkFollower<C> {
     // handle the swap initiation by locking funds on chain B
-    fn handle_counterparty_funds_locked(&self) -> Result<(), crate::Error> {
+    fn handle_counterparty_funds_locked(
+        &self,
+        contract_swap_id: [u8; 32],
+    ) -> Result<(), crate::Error> {
         // lock DRK funds to shared swap account
         let shared_swap_public_key = Keypair::new(self.secret).public;
 

src/protocol/error.rs

@@ -2,6 +2,7 @@ use crate::protocol::traits::CounterpartyKeys;
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
+    // initiator errors
     #[error("unexpected received counterparty keys event: {0}")]
     UnexpectedReceivedCounterpartyKeysEvent(CounterpartyKeys),
     #[error("unexpected counterparty funds locked event")]
@@ -12,6 +13,12 @@ pub enum Error {
     UnexpectedAlmostTimeout1Event,
     #[error("unexpected past timeout 2 event")]
     UnexpectedPastTimeout2Event,
-    #[error("failed to receive event")]
-    RecvError(#[from] smol::channel::RecvError),
+
+    // follower errors
+    #[error("unexpected counterparty funds locked event")]
+    UnexpectedCounterpartyFundsLocked,
+    #[error("unexpected ready to claim event")]
+    UnexpectedReadyToClaim,
+    #[error("unexpected counterparty funds refunded event")]
+    UnexpectedCounterpartyFundsRefunded,
 }

src/protocol/follower.rs

@@ -1,38 +1,101 @@
+use super::Error;
 use crate::protocol::traits::Follower;
 use smol::channel;
 
-use log::info;
+use log::{info, warn};
 
 #[allow(dead_code)]
-enum Event {
-    CounterpartyFundsLocked,
+pub(crate) enum Event {
+    // occurs when the counterparty has locked funds in the contract.
+    // contains the swap id within the contract.
+    CounterpartyFundsLocked([u8; 32]),
     ReadyToClaim,
     CounterpartyFundsRefunded([u8; 32]),
 }
 
+#[allow(dead_code)]
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum State {
+    WaitingForCounterpartyFundsLocked,
+    WaitingForContractReady,
+    Completed,
+}
+
 #[allow(dead_code)]
 struct Swap {
-    handler: Box<dyn Follower>,
+    // the chain-specific event handler
+    handler: Box<dyn Follower + Send + Sync>,
+
+    // the event receiver channel for the swap
+    // the [`Watcher`] sends events to this channel
     event_rx: channel::Receiver<Event>,
+
+    // the current state of the swap
+    state_tx: async_watch::Sender<State>,
+    state_rx: async_watch::Receiver<State>,
 }
 
 #[allow(dead_code)]
 impl Swap {
-    fn new(handler: Box<dyn Follower>, event_rx: channel::Receiver<Event>) -> Self {
-        Self { handler, event_rx }
+    fn new(
+        handler: Box<dyn Follower + Send + Sync>,
+        event_rx: channel::Receiver<Event>,
+    ) -> (Self, async_watch::Receiver<State>) {
+        let state = async_watch::channel(State::WaitingForCounterpartyFundsLocked);
+        (Self { handler, event_rx, state_tx: state.0, state_rx: state.1.clone() }, state.1)
     }
 
     async fn run(&mut self) -> Result<(), crate::Error> {
         loop {
             match self.event_rx.recv().await {
-                Ok(Event::CounterpartyFundsLocked) => {
-                    self.handler.handle_counterparty_funds_locked()?;
+                Ok(Event::CounterpartyFundsLocked(contract_swap_id)) => {
+                    info!("counterparty funds locked");
+
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyFundsLocked)
+                    {
+                        warn!(
+                            "unexpected event CounterpartyFundsLocked, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedCounterpartyFundsLocked.into());
+                    }
+
+                    self.handler.handle_counterparty_funds_locked(contract_swap_id)?;
+
+                    self.state_tx
+                        .send(State::WaitingForContractReady)
+                        .expect("state channel should not be dropped");
                 }
                 Ok(Event::ReadyToClaim) => {
+                    info!("ready to claim funds on counterparty chain");
+
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForContractReady) {
+                        warn!(
+                            "unexpected event ReadyToClaim, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedReadyToClaim.into());
+                    }
+
                     self.handler.handle_ready_to_claim()?;
+                    self.state_tx
+                        .send(State::Completed)
+                        .expect("state channel should not be dropped");
                 }
                 Ok(Event::CounterpartyFundsRefunded(counterparty_secret)) => {
+                    info!("counterparty refunded funds");
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForContractReady) {
+                        warn!(
+                            "unexpected event CounterpartyFundsRefunded, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedCounterpartyFundsRefunded.into());
+                    }
+
                     self.handler.handle_counterparty_funds_refunded(counterparty_secret)?;
+                    self.state_tx
+                        .send(State::Completed)
+                        .expect("state channel should not be dropped");
                 }
                 Err(_) => {
                     info!("event channel closed, exiting");

src/protocol/initiator.rs

@@ -206,6 +206,7 @@ impl Swap {
                 }
             }
 
+            // TODO: handle the case where the swap is *not* completed
             if matches!(*self.state_rx.borrow(), State::Completed) {
                 info!("swap completed, exiting");
                 break;

src/protocol/traits.rs

@@ -1,5 +1,9 @@
 use crate::ethereum::swap_creator::Swap; // TODO: shouldn't depend on this
-use crate::{error::Error, ethereum::swap_creator::SwapCreator, protocol::initiator::Event};
+use crate::{
+    error::Error,
+    ethereum::swap_creator::SwapCreator,
+    protocol::{follower, initiator},
+};
 use darkfi_serial::async_trait;
 use ethers::{prelude::*, utils::hex};
 use pasta_curves::pallas;
@@ -107,30 +111,30 @@ pub(crate) trait Initiator {
 #[async_trait]
 pub(crate) trait InitiatorEventWatcher {
     async fn run_received_counterparty_keys_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         counterparty_keys_rx: channel::Receiver<CounterpartyKeys>,
     ) -> Result<(), Error>;
 
     async fn run_counterparty_funds_locked_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
     ) -> Result<(), Error>;
 
     // TODO: make this generic for both chains
     async fn run_counterparty_funds_claimed_watcher<M: Middleware>(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         contract: SwapCreator<M>,
         contract_swap_id: &[u8; 32],
         from_block: u64,
     ) -> Result<(), Error>;
 
     async fn run_timeout_1_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         timeout_1: u64,
         buffer_seconds: u64,
     ) -> Result<(), Error>;
 
     async fn run_timeout_2_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         timeout_2: u64,
     ) -> Result<(), Error>;
 }
@@ -138,7 +142,10 @@ pub(crate) trait InitiatorEventWatcher {
 /// the chain that is the counterparty to the swap; ie. the second-mover
 pub(crate) trait Follower {
     // handle the swap initiation by locking funds on chain B
-    fn handle_counterparty_funds_locked(&self) -> Result<(), crate::Error>;
+    fn handle_counterparty_funds_locked(
+        &self,
+        contract_swap_id: [u8; 32],
+    ) -> Result<(), crate::Error>;
 
     // handle the funds being ready to be claimed by us
     fn handle_ready_to_claim(&self) -> Result<(), crate::Error>;
@@ -149,3 +156,24 @@ pub(crate) trait Follower {
         counterparty_secret: [u8; 32],
     ) -> Result<(), crate::Error>;
 }
+
+#[async_trait]
+pub(crate) trait FollowerEventWatcher {
+    async fn run_counterparty_funds_locked_watcher(
+        event_tx: channel::Sender<follower::Event>,
+    ) -> Result<(), Error>;
+
+    async fn run_ready_to_claim_watcher<M: Middleware>(
+        event_tx: channel::Sender<follower::Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), Error>;
+
+    async fn run_counterparty_funds_refunded_watcher<M: Middleware>(
+        event_tx: channel::Sender<follower::Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), Error>;
+}