fix xmrtaker key overwrite on restart (#423)

daemon/double_restart_test.go

@@ -0,0 +1,139 @@
+package daemon
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/athanorlabs/atomic-swap/coins"
+	"github.com/athanorlabs/atomic-swap/common/types"
+	"github.com/athanorlabs/atomic-swap/monero"
+	"github.com/athanorlabs/atomic-swap/rpcclient"
+	"github.com/athanorlabs/atomic-swap/rpcclient/wsclient"
+	"github.com/athanorlabs/atomic-swap/tests"
+)
+
+// This test starts a swap between Bob and Alice. The nodes restart *twice* after the xmr is locked
+// but before Bob claims. Then, on second restart, Bob should have claimed and
+// Alice should be able to get the XMR.
+// The test restarts twice specifically to check that the swap keys were not overwritten or
+// deleted from the database.
+func TestAliceDoubleRestartAfterXMRLock(t *testing.T) {
+	minXMR := coins.StrToDecimal("1")
+	maxXMR := minXMR
+	exRate := coins.StrToExchangeRate("300")
+	providesAmt, err := exRate.ToETH(minXMR)
+	require.NoError(t, err)
+
+	bobConf := CreateTestConf(t, tests.GetMakerTestKey(t))
+	monero.MineMinXMRBalance(t, bobConf.MoneroClient, coins.MoneroToPiconero(maxXMR))
+
+	aliceConf := CreateTestConf(t, tests.GetTakerTestKey(t))
+
+	timeout := 7 * time.Minute
+	ctx, cancel := LaunchDaemons(t, timeout, bobConf, aliceConf)
+
+	bws, err := wsclient.NewWsClient(ctx, fmt.Sprintf("ws://127.0.0.1:%d/ws", bobConf.RPCPort))
+	require.NoError(t, err)
+	aws, err := wsclient.NewWsClient(ctx, fmt.Sprintf("ws://127.0.0.1:%d/ws", aliceConf.RPCPort))
+	require.NoError(t, err)
+
+	// Use an independent context for these clients that will execute across 2 runs of the daemons
+	bc := rpcclient.NewClient(context.Background(), fmt.Sprintf("http://127.0.0.1:%d", bobConf.RPCPort))
+	ac := rpcclient.NewClient(context.Background(), fmt.Sprintf("http://127.0.0.1:%d", aliceConf.RPCPort))
+
+	tokenAddr := GetMockTokens(t, aliceConf.EthereumClient)[MockTether]
+	tokenAsset := types.EthAsset(tokenAddr)
+
+	makeResp, bobStatusCh, err := bws.MakeOfferAndSubscribe(minXMR, maxXMR, exRate, tokenAsset, false)
+	require.NoError(t, err)
+
+	aliceStatusCh, err := aws.TakeOfferAndSubscribe(makeResp.PeerID, makeResp.OfferID, providesAmt)
+	require.NoError(t, err)
+
+	var statusWG sync.WaitGroup
+	statusWG.Add(2)
+
+	// Test that Bob completes the swap successfully
+	go func() {
+		defer statusWG.Done()
+		for {
+			select {
+			case status := <-bobStatusCh:
+				t.Log("> Bob got status:", status)
+				if !status.IsOngoing() {
+					assert.Equal(t, types.CompletedSuccess.String(), status.String())
+					return
+				}
+
+				if status == types.XMRLocked {
+					cancel()
+					t.Log("cancelling context of Alice's and Bob's servers")
+					return
+				}
+			case <-ctx.Done():
+				t.Logf("Bob's context cancelled before she completed the swap [expected]")
+				return
+			}
+		}
+	}()
+
+	// In theory, Alice won't complete the swap in the background goroutine
+	// below, because we shut down her server as soon as Bob's end succeeded,
+	// and Bob would normally succeed first, while Alice is still sweeping XMR
+	// funds from the swap wallet back to her primary wallet.
+	go func() {
+		defer statusWG.Done()
+		for {
+			select {
+			case status := <-aliceStatusCh:
+				t.Log("> Alice got status:", status)
+				if !status.IsOngoing() {
+					return
+				}
+			case <-ctx.Done():
+				t.Logf("Alice's context cancelled before she completed the swap [expected]")
+				return
+			}
+		}
+	}()
+
+	statusWG.Wait()
+	t.Logf("Both swaps completed or cancelled")
+	if t.Failed() {
+		return
+	}
+
+	// Make sure both servers had time to fully shut down
+	time.Sleep(3 * time.Second)
+
+	// relaunch the daemons
+	t.Logf("daemons stopped, now re-launching them")
+	_, cancel = LaunchDaemons(t, 3*time.Minute, bobConf, aliceConf)
+
+	t.Logf("daemons relaunched, waiting a few seconds before restarting them")
+	time.Sleep(3 * time.Second)
+	cancel()
+	time.Sleep(5 * time.Second) // wait for daemons to shut down
+
+	// relaunch the daemons
+	t.Logf("daemons stopped, now re-launching them")
+	_, cancel = LaunchDaemons(t, 3*time.Minute, bobConf, aliceConf)
+	t.Logf("daemons relaunched, waiting a few seconds before checking swap status")
+	time.Sleep(10 * time.Second) // give nodes time to complete the swap
+
+	pastSwap, err := ac.GetPastSwap(&makeResp.OfferID)
+	require.NoError(t, err)
+	t.Logf("Alice past status: %s", pastSwap.Swaps[0].Status)
+	require.Equal(t, types.CompletedSuccess, pastSwap.Swaps[0].Status)
+
+	pastSwap, err = bc.GetPastSwap(&makeResp.OfferID)
+	require.NoError(t, err)
+	t.Logf("Bob past status: %s", pastSwap.Swaps[0].Status)
+	require.Equal(t, types.CompletedSuccess, pastSwap.Swaps[0].Status)
+}

daemon/test_support.go

@@ -85,7 +85,6 @@ func LaunchDaemons(t *testing.T, timeout time.Duration, configs ...*SwapdConfig)
 	var bootNodes []string // First daemon to launch has no bootnodes
 
 	for n, conf := range configs {
-
 		conf.EnvConf.Bootnodes = bootNodes
 
 		wg.Add(1)

protocol/xmrtaker/swap_state.go

@@ -150,6 +150,10 @@ func newSwapStateFromStart(
 		return nil, err
 	}
 
+	if err := s.generateAndSetKeys(); err != nil {
+		return nil, err
+	}
+
 	statusCh <- stage
 	return s, nil
 }
@@ -294,11 +298,6 @@ func newSwapState(
 		statusCh:          info.StatusCh(),
 	}
 
-	if err := s.generateAndSetKeys(); err != nil {
-		cancel()
-		return nil, err
-	}
-
 	go s.runHandleEvents()
 	go s.runContractEventWatcher()
 	return s, nil
@@ -378,7 +377,7 @@ func (s *swapState) exit() error {
 		s.clearNextExpectedEvent(types.CompletedAbort)
 		return nil
 	case EventXMRLockedType, EventETHClaimedType:
-		// for EventXMRLocked, we already deployed the contract,
+		// for EventXMRLocked, we already locked our ETH-asset,
 		// so we should call Refund().
 		//
 		// for EventETHClaimed, the XMR has been locked, but the
@@ -636,6 +635,22 @@ func (s *swapState) ready() error {
 	}
 
 	if stage != contracts.StagePending {
+		if stage == contracts.StageReady {
+			log.Warnf("contract already set to ready, ignoring call to ready()")
+			return nil
+		}
+
+		if stage == contracts.StageCompleted {
+			log.Infof("contract aleady set to completed, ignoring call to ready() and sending EventExit")
+			go func() {
+				err = s.Exit()
+				if err != nil {
+					log.Errorf("failed to handle EventExit: %s", err)
+				}
+			}()
+			return nil
+		}
+
 		return fmt.Errorf("cannot set contract to ready when swap stage is %s", contracts.StageToString(stage))
 	}
 

rpc/swap.go

@@ -399,6 +399,8 @@ func estimatedTimeToCompletionForStatus(env common.Environment, status types.Sta
 		return (moneroBlockTime * 10) + (ethBlockTime * 4), nil
 	case types.ContractReady:
 		return (moneroBlockTime * 2) + (ethBlockTime * 2), nil
+	case types.SweepingXMR:
+		return (moneroBlockTime * 2), nil
 	default:
 		return 0, fmt.Errorf("invalid status %s; must be ongoing status type", status)
 	}