fix: store newSwap tx in db, handle pending newSwap on restart (#470)

common/consts.go

@@ -26,13 +26,6 @@ const (
 	TimeFmtNSecs = "2006-01-02-15:04:05.999999999"
 )
 
-// SwapCreator.sol event signatures
-const (
-	ReadyEventSignature    = "Ready(bytes32)"
-	ClaimedEventSignature  = "Claimed(bytes32,bytes32)"
-	RefundedEventSignature = "Refunded(bytes32,bytes32)"
-)
-
 // Ethereum chain IDs
 const (
 	MainnetChainID = 1

common/utils_test.go

@@ -43,11 +43,6 @@ func TestEthereumPrivateKeyToAddress(t *testing.T) {
 	require.Equal(t, ethAddressHex, addr.String())
 }
 
-func TestGetTopic(t *testing.T) {
-	refundedTopic := ethcommon.HexToHash("0x007c875846b687732a7579c19bb1dade66cd14e9f4f809565e2b2b5e76c72b4f")
-	require.Equal(t, GetTopic(RefundedEventSignature), refundedTopic)
-}
-
 func TestMakeDir(t *testing.T) {
 	path := path.Join(t.TempDir(), "mainnet")
 	require.NoError(t, MakeDir(path))

daemon/cancel_or_refund_test.go

@@ -0,0 +1,135 @@
+// Copyright 2023 The AthanorLabs/atomic-swap Authors
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package daemon
+
+import (
+	"context"
+	"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/tests"
+)
+
+// This test starts a swap between Bob and Alice. The nodes are shut down
+// after the key exchange step, while Alice is trying to lock funds,
+// and the newSwap tx is in the mempool but not yet included.
+// Alice's node is then restarted, and depending on whether the newSwap
+// tx was included yet or not, she should be able to cancel or refund the swap.
+// In this case, the tx is always included by the time she restarts,
+// so she refunds the swap.
+// Bob should have aborted the swap in all cases.
+func TestXMRTakerCancelOrRefundAfterKeyExchange(t *testing.T) {
+	minXMR := coins.StrToDecimal("1")
+	maxXMR := minXMR
+	exRate := coins.StrToExchangeRate("300")
+	providesAmt, err := exRate.ToETH(minXMR)
+	require.NoError(t, err)
+
+	ec, _ := tests.NewEthClient(t)
+
+	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)
+
+	// Use an independent context for these clients that will execute across multiple runs of the daemons
+	bc := rpcclient.NewClient(context.Background(), bobConf.RPCPort)
+	ac := rpcclient.NewClient(context.Background(), aliceConf.RPCPort)
+
+	makeResp, bobStatusCh, err := bc.MakeOfferAndSubscribe(minXMR, maxXMR, exRate, types.EthAssetETH, false)
+	require.NoError(t, err)
+
+	aliceStatusCh, err := ac.TakeOfferAndSubscribe(makeResp.PeerID, makeResp.OfferID, providesAmt)
+	require.NoError(t, err)
+
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+		for {
+			count, err := ec.PendingTransactionCount(ctx) //nolint:govet
+			if err != nil {
+				t.Errorf("failed to get pending tx count: %s", err)
+				return
+			}
+
+			if count > 0 {
+				// the newSwap tx is in the mempool, shut down the nodes
+				cancel()
+				t.Log("cancelling context of Alice's and Bob's servers")
+				return
+			}
+
+			time.Sleep(time.Millisecond * 200)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case status := <-bobStatusCh:
+				t.Log("> Bob got status:", status)
+				if !status.IsOngoing() {
+					assert.Equal(t, types.CompletedAbort.String(), status.String())
+					return
+				}
+			case <-ctx.Done():
+				t.Logf("Bob's context cancelled before he completed the swap [expected]")
+				return
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.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
+			}
+		}
+	}()
+
+	wg.Wait()
+	t.Logf("Both swaps cancelled")
+	if t.Failed() {
+		return
+	}
+
+	// Make sure both servers had time to fully shut down
+	time.Sleep(3 * time.Second)
+
+	t.Logf("daemons stopped, now re-launching them")
+	_, _ = LaunchDaemons(t, 3*time.Minute, bobConf, aliceConf)
+
+	pastSwap, err := ac.GetPastSwap(&makeResp.OfferID)
+	require.NoError(t, err)
+	require.NotEmpty(t, pastSwap.Swaps)
+	require.Equal(t, types.CompletedRefund.String(), pastSwap.Swaps[0].Status.String(),
+		"Alice should have refunded the swap")
+
+	pastSwap, err = bc.GetPastSwap(&makeResp.OfferID)
+	require.NoError(t, err)
+	require.NotEmpty(t, pastSwap.Swaps)
+	require.Equal(t, types.CompletedAbort.String(), pastSwap.Swaps[0].Status.String())
+}

db/recovery_db.go

@@ -18,6 +18,7 @@ const (
 	counterpartySwapPrivateKeyPrefix = "cspriv"
 	relayerInfoPrefix                = "relayer"
 	counterpartySwapKeysPrefix       = "cskeys"
+	newSwapTxHashPrefix              = "newswap"
 )
 
 // RecoveryDB contains information about ongoing swaps required for recovery
@@ -218,6 +219,31 @@ func (db *RecoveryDB) GetCounterpartySwapKeys(id types.Hash) (*mcrypto.PublicKey
 	return info.PublicSpendKey, info.PrivateViewKey, nil
 }
 
+// PutNewSwapTxHash stores the newSwap transaction hash for the given swap ID.
+func (db *RecoveryDB) PutNewSwapTxHash(id types.Hash, txHash types.Hash) error {
+	key := getRecoveryDBKey(id, newSwapTxHashPrefix)
+	err := db.db.Put(key, txHash[:])
+	if err != nil {
+		return err
+	}
+
+	return db.db.Flush()
+}
+
+// GetNewSwapTxHash returns the newSwap transaction hash for the given swap ID.
+func (db *RecoveryDB) GetNewSwapTxHash(id types.Hash) (types.Hash, error) {
+	key := getRecoveryDBKey(id, newSwapTxHashPrefix)
+	value, err := db.db.Get(key)
+	if err != nil {
+		return types.Hash{}, err
+	}
+
+	var txHash types.Hash
+	copy(txHash[:], value)
+
+	return txHash, nil
+}
+
 // DeleteSwap deletes all recovery info from the db for the given swap.
 // TODO: this is currently unimplemented
 func (db *RecoveryDB) DeleteSwap(id types.Hash) error {
@@ -232,6 +258,7 @@ func (db *RecoveryDB) deleteSwap(id types.Hash) error {
 		getRecoveryDBKey(id, swapPrivateKeyPrefix),
 		getRecoveryDBKey(id, counterpartySwapPrivateKeyPrefix),
 		getRecoveryDBKey(id, counterpartySwapKeysPrefix),
+		getRecoveryDBKey(id, newSwapTxHashPrefix),
 	}
 
 	for _, key := range keys {

ethereum/extethclient/eth_wallet_client.go

@@ -55,6 +55,9 @@ type EthClient interface {
 	// does not need locking, as it locks internally
 	Transfer(ctx context.Context, to ethcommon.Address, amount *coins.WeiAmount) (ethcommon.Hash, error)
 
+	// attempts to cancel a transaction with the given nonce by sending a zero-value tx to ourselves
+	CancelTxWithNonce(ctx context.Context, nonce uint64, gasPrice *big.Int) (ethcommon.Hash, error)
+
 	WaitForReceipt(ctx context.Context, txHash ethcommon.Hash) (*ethtypes.Receipt, error)
 	WaitForTimestamp(ctx context.Context, ts time.Time) error
 	LatestBlockTimestamp(ctx context.Context) (time.Time, error)
@@ -298,6 +301,30 @@ func (c *ethClient) Raw() *ethclient.Client {
 	return c.ec
 }
 
+func (c *ethClient) CancelTxWithNonce(
+	ctx context.Context,
+	nonce uint64,
+	gasPrice *big.Int,
+) (ethcommon.Hash, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	tx := ethtypes.NewTransaction(nonce, c.ethAddress, big.NewInt(0), 21000, gasPrice, nil)
+
+	signer := ethtypes.LatestSignerForChainID(c.chainID)
+	signedTx, err := ethtypes.SignTx(tx, signer, c.ethPrivKey)
+	if err != nil {
+		return ethcommon.Hash{}, fmt.Errorf("failed to sign tx: %w", err)
+	}
+
+	err = c.ec.SendTransaction(ctx, signedTx)
+	if err != nil {
+		return ethcommon.Hash{}, fmt.Errorf("failed to send tx: %w", err)
+	}
+
+	return signedTx.Hash(), nil
+}
+
 func (c *ethClient) Transfer(
 	ctx context.Context,
 	to ethcommon.Address,
@@ -311,16 +338,6 @@ func (c *ethClient) Transfer(
 		return ethcommon.Hash{}, fmt.Errorf("failed to get nonce: %w", err)
 	}
 
-	// TODO: why does this type not implement ethtypes.TxData? seems like a bug in geth
-	// txData := ethtypes.DynamicFeeTx{
-	// 	ChainID: c.chainID,
-	// 	Nonce:   nonce,
-	// 	Gas:     21000,
-	// 	To:      &to,
-	// 	Value:   amount.BigInt(),
-	// }
-	// tx := ethtypes.NewTx(txData)
-
 	gasPrice, err := c.ec.SuggestGasPrice(ctx)
 	if err != nil {
 		return ethcommon.Hash{}, fmt.Errorf("failed to get gas price: %w", err)

ethereum/utils.go

@@ -36,8 +36,13 @@ var (
 	// which case it will be nil we'll see panics when vetting the binaries.
 	SwapCreatorParsedABI, _ = SwapCreatorMetaData.GetAbi()
 
-	claimedTopic  = common.GetTopic(common.ClaimedEventSignature)
-	refundedTopic = common.GetTopic(common.RefundedEventSignature)
+	claimedTopic  = common.GetTopic(ClaimedEventSignature)
+	refundedTopic = common.GetTopic(RefundedEventSignature)
+
+	NewSwapFunctionSignature = SwapCreatorParsedABI.Methods["newSwap"].Sig //nolint:revive
+	ReadyEventSignature      = SwapCreatorParsedABI.Events["Ready"].Sig    //nolint:revive
+	ClaimedEventSignature    = SwapCreatorParsedABI.Events["Claimed"].Sig  //nolint:revive
+	RefundedEventSignature   = SwapCreatorParsedABI.Events["Refunded"].Sig //nolint:revive
 )
 
 // StageToString converts a contract Stage enum value to a string

ethereum/utils_test.go

@@ -6,6 +6,9 @@ package contracts
 import (
 	"testing"
 
+	"github.com/athanorlabs/atomic-swap/common"
+
+	ethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/stretchr/testify/require"
 )
 
@@ -21,3 +24,8 @@ func TestStage_StageToString(t *testing.T) {
 		require.Equal(t, expectedValues[s], StageToString(s))
 	}
 }
+
+func TestGetTopic(t *testing.T) {
+	refundedTopic := ethcommon.HexToHash("0x007c875846b687732a7579c19bb1dade66cd14e9f4f809565e2b2b5e76c72b4f")
+	require.Equal(t, common.GetTopic(RefundedEventSignature), refundedTopic)
+}

protocol/backend/backend.go

@@ -53,6 +53,8 @@ type RecoveryDB interface {
 	GetSwapRelayerInfo(id types.Hash) (*types.OfferExtra, error)
 	PutCounterpartySwapKeys(id types.Hash, sk *mcrypto.PublicKey, vk *mcrypto.PrivateViewKey) error
 	GetCounterpartySwapKeys(id types.Hash) (*mcrypto.PublicKey, *mcrypto.PrivateViewKey, error)
+	PutNewSwapTxHash(id types.Hash, txHash types.Hash) error
+	GetNewSwapTxHash(id types.Hash) (types.Hash, error)
 	DeleteSwap(id types.Hash) error
 }
 

protocol/backend/mock_recovery_db.go

@@ -98,6 +98,21 @@ func (mr *MockRecoveryDBMockRecorder) GetCounterpartySwapPrivateKey(arg0 interfa
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCounterpartySwapPrivateKey", reflect.TypeOf((*MockRecoveryDB)(nil).GetCounterpartySwapPrivateKey), arg0)
 }
 
+// GetNewSwapTxHash mocks base method.
+func (m *MockRecoveryDB) GetNewSwapTxHash(arg0 common.Hash) (common.Hash, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetNewSwapTxHash", arg0)
+	ret0, _ := ret[0].(common.Hash)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetNewSwapTxHash indicates an expected call of GetNewSwapTxHash.
+func (mr *MockRecoveryDBMockRecorder) GetNewSwapTxHash(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNewSwapTxHash", reflect.TypeOf((*MockRecoveryDB)(nil).GetNewSwapTxHash), arg0)
+}
+
 // GetSwapPrivateKey mocks base method.
 func (m *MockRecoveryDB) GetSwapPrivateKey(arg0 common.Hash) (*mcrypto.PrivateSpendKey, error) {
 	m.ctrl.T.Helper()
@@ -170,6 +185,20 @@ func (mr *MockRecoveryDBMockRecorder) PutCounterpartySwapPrivateKey(arg0, arg1 i
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PutCounterpartySwapPrivateKey", reflect.TypeOf((*MockRecoveryDB)(nil).PutCounterpartySwapPrivateKey), arg0, arg1)
 }
 
+// PutNewSwapTxHash mocks base method.
+func (m *MockRecoveryDB) PutNewSwapTxHash(arg0, arg1 common.Hash) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PutNewSwapTxHash", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// PutNewSwapTxHash indicates an expected call of PutNewSwapTxHash.
+func (mr *MockRecoveryDBMockRecorder) PutNewSwapTxHash(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PutNewSwapTxHash", reflect.TypeOf((*MockRecoveryDB)(nil).PutNewSwapTxHash), arg0, arg1)
+}
+
 // PutSwapPrivateKey mocks base method.
 func (m *MockRecoveryDB) PutSwapPrivateKey(arg0 common.Hash, arg1 *mcrypto.PrivateSpendKey) error {
 	m.ctrl.T.Helper()

protocol/txsender/external_sender.go

@@ -117,16 +117,16 @@ func (s *ExternalSender) NewSwap(
 	timeoutDuration *big.Int,
 	nonce *big.Int,
 	amount coins.EthAssetAmount,
-) (*ethtypes.Receipt, error) {
+) (ethcommon.Hash, error) {
 	// TODO: Add ERC20 token support and approve new_swap for the token transfer
 	if amount.IsToken() {
-		return nil, errors.New("external sender does not support ERC20 token swaps")
+		return ethcommon.Hash{}, errors.New("external sender does not support ERC20 token swaps")
 	}
 
 	input, err := s.abi.Pack("new_swap", pubKeyClaim, pubKeyRefund, claimer, timeoutDuration,
 		amount.TokenAddress(), amount.BigInt(), nonce)
 	if err != nil {
-		return nil, err
+		return ethcommon.Hash{}, err
 	}
 
 	tx := &Transaction{
@@ -142,16 +142,11 @@ func (s *ExternalSender) NewSwap(
 	var txHash ethcommon.Hash
 	select {
 	case <-time.After(transactionTimeout):
-		return nil, errTransactionTimeout
+		return ethcommon.Hash{}, errTransactionTimeout
 	case txHash = <-s.in:
 	}
 
-	receipt, err := block.WaitForReceipt(s.ctx, s.ec, txHash)
-	if err != nil {
-		return nil, err
-	}
-
-	return receipt, nil
+	return txHash, nil
 }
 
 // SetReady prompts the external sender to sign a set_ready transaction

protocol/txsender/sender.go

@@ -39,7 +39,7 @@ type Sender interface {
 		timeoutDuration *big.Int,
 		nonce *big.Int,
 		amount coins.EthAssetAmount,
-	) (*ethtypes.Receipt, error)
+	) (ethcommon.Hash, error)
 	SetReady(swap *contracts.SwapCreatorSwap) (*ethtypes.Receipt, error)
 	Claim(swap *contracts.SwapCreatorSwap, secret [32]byte) (*ethtypes.Receipt, error)
 	Refund(swap *contracts.SwapCreatorSwap, secret [32]byte) (*ethtypes.Receipt, error)
@@ -83,9 +83,8 @@ func (s *privateKeySender) NewSwap(
 	timeoutDuration *big.Int,
 	nonce *big.Int,
 	amount coins.EthAssetAmount,
-) (*ethtypes.Receipt, error) {
-	s.ethClient.Lock()
-	defer s.ethClient.Unlock()
+) (ethcommon.Hash, error) {
+	// note: the caller must lock the ethclient.
 
 	value := amount.BigInt()
 
@@ -96,17 +95,17 @@ func (s *privateKeySender) NewSwap(
 	if amount.IsToken() {
 		txOpts, err := s.ethClient.TxOpts(s.ctx)
 		if err != nil {
-			return nil, err
+			return ethcommon.Hash{}, err
 		}
 
 		tx, err := s.erc20Contract.Approve(txOpts, s.swapCreatorAddr, value)
 		if err != nil {
-			return nil, fmt.Errorf("approve tx creation failed, %w", err)
+			return ethcommon.Hash{}, fmt.Errorf("approve tx creation failed, %w", err)
 		}
 
 		receipt, err := block.WaitForReceipt(s.ctx, s.ethClient.Raw(), tx.Hash())
 		if err != nil {
-			return nil, fmt.Errorf("approve failed, %w", err)
+			return ethcommon.Hash{}, fmt.Errorf("approve failed, %w", err)
 		}
 
 		log.Debugf("approve transaction included %s", common.ReceiptInfo(receipt))
@@ -116,7 +115,7 @@ func (s *privateKeySender) NewSwap(
 
 	txOpts, err := s.ethClient.TxOpts(s.ctx)
 	if err != nil {
-		return nil, err
+		return ethcommon.Hash{}, err
 	}
 
 	// transfer ETH if we're not doing an ERC20 swap
@@ -128,16 +127,10 @@ func (s *privateKeySender) NewSwap(
 		amount.TokenAddress(), value, nonce)
 	if err != nil {
 		err = fmt.Errorf("new_swap tx creation failed, %w", err)
-		return nil, err
+		return ethcommon.Hash{}, err
 	}
 
-	receipt, err := block.WaitForReceipt(s.ctx, s.ethClient.Raw(), tx.Hash())
-	if err != nil {
-		err = fmt.Errorf("new_swap failed, %w", err)
-		return nil, err
-	}
-
-	return receipt, nil
+	return tx.Hash(), nil
 }
 
 func (s *privateKeySender) SetReady(swap *contracts.SwapCreatorSwap) (*ethtypes.Receipt, error) {

protocol/xmrmaker/swap_state.go

@@ -39,9 +39,9 @@ import (
 )
 
 var (
-	readyTopic    = common.GetTopic(common.ReadyEventSignature)
-	claimedTopic  = common.GetTopic(common.ClaimedEventSignature)
-	refundedTopic = common.GetTopic(common.RefundedEventSignature)
+	readyTopic    = common.GetTopic(contracts.ReadyEventSignature)
+	claimedTopic  = common.GetTopic(contracts.ClaimedEventSignature)
+	refundedTopic = common.GetTopic(contracts.RefundedEventSignature)
 )
 
 type swapState struct {

protocol/xmrtaker/event_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/athanorlabs/atomic-swap/common"
 	"github.com/athanorlabs/atomic-swap/common/types"
 	mcrypto "github.com/athanorlabs/atomic-swap/crypto/monero"
+	contracts "github.com/athanorlabs/atomic-swap/ethereum"
 	"github.com/athanorlabs/atomic-swap/ethereum/watcher"
 	"github.com/athanorlabs/atomic-swap/monero"
 	"github.com/athanorlabs/atomic-swap/net/message"
@@ -40,7 +41,7 @@ func lockXMRAndCheckForReadyLog(t *testing.T, s *swapState, xmrAddr *mcrypto.Add
 	require.NoError(t, err)
 	logReadyCh := make(chan ethtypes.Log)
 
-	readyTopic := common.GetTopic(common.ReadyEventSignature)
+	readyTopic := common.GetTopic(contracts.ReadyEventSignature)
 	readyWatcher := watcher.NewEventFilter(
 		s.Backend.Ctx(),
 		s.Backend.ETHClient().Raw(),

protocol/xmrtaker/instance.go

@@ -4,15 +4,25 @@
 package xmrtaker
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"math/big"
 	"sync"
 
+	"github.com/ChainSafe/chaindb"
+	ethereum "github.com/ethereum/go-ethereum"
+	ethcommon "github.com/ethereum/go-ethereum/common"
+	ethtypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
 	logging "github.com/ipfs/go-log"
 
 	"github.com/athanorlabs/atomic-swap/coins"
 	"github.com/athanorlabs/atomic-swap/common"
 	"github.com/athanorlabs/atomic-swap/common/types"
 	mcrypto "github.com/athanorlabs/atomic-swap/crypto/monero"
+	contracts "github.com/athanorlabs/atomic-swap/ethereum"
+	"github.com/athanorlabs/atomic-swap/ethereum/block"
 	pcommon "github.com/athanorlabs/atomic-swap/protocol"
 	"github.com/athanorlabs/atomic-swap/protocol/backend"
 	"github.com/athanorlabs/atomic-swap/protocol/swap"
@@ -79,13 +89,30 @@ func (inst *Instance) checkForOngoingSwaps() error {
 			continue
 		}
 
-		if s.Status == types.KeysExchanged || s.Status == types.ExpectingKeys {
-			// set status to aborted, delete info from recovery db
-			log.Infof("found ongoing swap %s in DB, aborting since no funds were locked", s.OfferID)
-			err = inst.abortOngoingSwap(s)
-			if err != nil {
-				log.Warnf("failed to abort ongoing swap %s: %s", s.OfferID, err)
+		// in this case, we exited either before locking funds, or before the newSwap tx
+		// was included in the chain.
+		if s.Status == types.ExpectingKeys {
+			txHash, err := inst.backend.RecoveryDB().GetNewSwapTxHash(s.OfferID) //nolint:govet
+			if err != nil && errors.Is(err, chaindb.ErrKeyNotFound) {
+				// since there was no newSwap tx hash, it means there was never an attempt to lock funds,
+				// and we can safely abort the swap.
+				log.Infof("found ongoing swap %s in DB, aborting since no funds were locked", s.OfferID)
+				err = inst.abortOngoingSwap(s)
+				if err != nil {
+					return fmt.Errorf("failed to abort ongoing swap %s: %s", s.OfferID, err)
+				}
+
+				continue
+			} else if err != nil {
+				return fmt.Errorf("failed to get newSwap tx hash for ongoing swap %s: %w", s.OfferID, err)
 			}
+
+			// we have a newSwap tx hash, so we need to check if it was included in the chain.
+			err = inst.refundOrCancelNewSwap(s, txHash)
+			if err != nil {
+				return fmt.Errorf("failed to refund or cancel swap %s: %w", s.OfferID, err)
+			}
+
 			continue
 		}
 
@@ -124,6 +151,152 @@ func (inst *Instance) abortOngoingSwap(s *swap.Info) error {
 	return inst.backend.RecoveryDB().DeleteSwap(s.OfferID)
 }
 
+// refundOrCancelNewSwap checks if the newSwap tx was included in the chain.
+// if it was, it attempts to refund the swap.
+// otherwise, it attempts to cancel the swap by sending a zero-value transfer to our own account
+// with the same nonce.
+func (inst *Instance) refundOrCancelNewSwap(s *swap.Info, txHash ethcommon.Hash) error {
+	log.Infof("found ongoing swap %s with status %s in DB, checking to either refund or cancel", s.OfferID, s.Status)
+
+	cancelled, err := inst.maybeCancelNewSwap(txHash)
+	if err != nil {
+		return err
+	}
+
+	if cancelled {
+		return nil
+	}
+
+	receipt, err := block.WaitForReceipt(inst.backend.Ctx(), inst.backend.ETHClient().Raw(), txHash)
+	if err != nil {
+		return fmt.Errorf("failed to get newSwap transaction receipt: %w", err)
+	}
+
+	if len(receipt.Logs) == 0 {
+		return errSwapInstantiationNoLogs
+	}
+
+	var t1 *big.Int
+	var t2 *big.Int
+	for _, log := range receipt.Logs {
+		t1, t2, err = contracts.GetTimeoutsFromLog(log)
+		if err == nil {
+			break
+		}
+	}
+	if err != nil {
+		return fmt.Errorf("timeouts not found in transaction receipt's logs: %w", err)
+	}
+
+	// we have a tx hash, so we can assume that the swap is ongoing
+	params, err := getNewSwapParametersFromTx(inst.backend.Ctx(), inst.backend.ETHClient().Raw(), txHash)
+	if err != nil {
+		return fmt.Errorf("failed to get newSwap parameters from tx %s: %w", txHash, err)
+	}
+
+	swap := contracts.SwapCreatorSwap{
+		Owner:        params.owner,
+		Claimer:      params.claimer,
+		PubKeyClaim:  params.cmtXMRMaker,
+		PubKeyRefund: params.cmtXMRTaker,
+		Timeout1:     t1,
+		Timeout2:     t2,
+		Asset:        params.asset,
+		Value:        params.value,
+		Nonce:        params.nonce,
+	}
+
+	// our secret value
+	secret, err := inst.backend.RecoveryDB().GetSwapPrivateKey(s.OfferID)
+	if err != nil {
+		return fmt.Errorf("failed to get private key for ongoing swap from db with offer id %s: %w",
+			s.OfferID, err)
+	}
+
+	swapCreator, err := contracts.NewSwapCreator(params.swapCreatorAddr, inst.backend.ETHClient().Raw())
+	if err != nil {
+		return fmt.Errorf("failed to instantiate SwapCreator contract: %w", err)
+	}
+
+	stage, err := swapCreator.Swaps(nil, swap.SwapID())
+	if err != nil {
+		return fmt.Errorf("failed to get swap stage: %w", err)
+	}
+
+	// TODO: if this is not the case, then the swap has probably already been claimed.
+	// however, this seems very unlikely to happen, as the swap counterparty will exit
+	// the swap as soon as the network stream is closed, and they will likely not have
+	// locked XMR or claimed.
+	if stage != contracts.StagePending {
+		return fmt.Errorf("swap %s is not in pending stage, aborting", s.OfferID)
+	}
+
+	// TODO: check for t1/t2? if between t1 and t2, we need to wait for t2
+
+	txOpts, err := inst.backend.ETHClient().TxOpts(inst.backend.Ctx())
+	if err != nil {
+		return fmt.Errorf("failed to get tx opts: %w", err)
+	}
+
+	refundTx, err := swapCreator.Refund(txOpts, swap, [32]byte(common.Reverse(secret.Bytes())))
+	if err != nil {
+		return fmt.Errorf("failed to create refund tx: %w", err)
+	}
+
+	log.Infof("submit refund tx %s for swap %s", refundTx.Hash(), s.OfferID)
+	receipt, err = block.WaitForReceipt(inst.backend.Ctx(), inst.backend.ETHClient().Raw(), refundTx.Hash())
+	if err != nil {
+		return fmt.Errorf("failed to get refund transaction receipt: %w", err)
+	}
+
+	log.Infof("refunded swap %s successfully: %s", s.OfferID, common.ReceiptInfo(receipt))
+
+	// set status to refunded
+	s.Status = types.CompletedRefund
+	return inst.backend.SwapManager().CompleteOngoingSwap(s)
+}
+
+func (inst *Instance) maybeCancelNewSwap(txHash ethcommon.Hash) (bool, error) {
+	tx, isPending, err := inst.backend.ETHClient().Raw().TransactionByHash(inst.backend.Ctx(), txHash)
+	if err != nil {
+		return false, fmt.Errorf("failed to get newSwap transaction: %w", err)
+	}
+
+	if !isPending {
+		// tx is already included, so we can't cancel it
+		return false, nil
+	}
+
+	log.Infof("newSwap tx %s is still pending, attempting to cancel", tx.Hash())
+
+	// just double the gas price for now, this is higher than needed for a replacement tx though
+	gasPrice := new(big.Int).Mul(tx.GasPrice(), big.NewInt(2))
+	cancelTx, err := inst.backend.ETHClient().CancelTxWithNonce(inst.backend.Ctx(), tx.Nonce(), gasPrice)
+	if err != nil {
+		return false, fmt.Errorf("failed to create or send cancel tx: %w", err)
+	}
+
+	log.Infof("submit cancel tx %s", cancelTx)
+
+	// TODO: if newSwap is included instead of the cancel tx (unlikely), block.WaitForReceipt will actually
+	// loop for 1 hour and block as there will never be a receipt for the cancel tx.
+	// we should probably poll for our account nonce to increase, and when it does, check for receipts
+	// on both txs to see which one was successful.
+	receipt, err := block.WaitForReceipt(inst.backend.Ctx(), inst.backend.ETHClient().Raw(), cancelTx)
+	if err != nil && !errors.Is(err, ethereum.NotFound) {
+		return false, fmt.Errorf("failed to get cancel transaction receipt: %w", err)
+	}
+
+	if errors.Is(err, ethereum.NotFound) || receipt.Status == ethtypes.ReceiptStatusFailed {
+		// this is okay, it means newSwap was included instead, and we can refund it in the calling function
+		log.Infof("failed to cancel swap, attempting to refund")
+		return false, nil
+	}
+
+	log.Infof("cancelled newSwap tx %s successfully: %s", tx.Hash(), common.ReceiptInfo(receipt))
+	return true, nil
+}
+
 func (inst *Instance) createOngoingSwap(s *swap.Info) error {
 	log.Infof("found ongoing swap %s with status %s in DB, restarting swap", s.OfferID, s.Status)
 
@@ -256,3 +429,72 @@ func (inst *Instance) ExternalSender(offerID types.Hash) (*txsender.ExternalSend
 
 	return es, nil
 }
+
+type newSwapParameters struct {
+	swapCreatorAddr ethcommon.Address
+	owner           ethcommon.Address
+	claimer         ethcommon.Address
+	cmtXMRMaker     [32]byte
+	cmtXMRTaker     [32]byte
+	asset           ethcommon.Address
+	value           *big.Int
+	nonce           *big.Int
+}
+
+func getNewSwapParametersFromTx(
+	ctx context.Context,
+	ec *ethclient.Client,
+	txHash ethcommon.Hash,
+) (*newSwapParameters, error) {
+	var newSwapTopic = common.GetTopic(contracts.NewSwapFunctionSignature)
+
+	tx, _, err := ec.TransactionByHash(ctx, txHash)
+	if err != nil {
+		return nil, err
+	}
+
+	if tx.To() == nil {
+		return nil, fmt.Errorf("invalid transaction: to address is nil")
+	}
+
+	data := tx.Data()
+	if len(data) < 4 {
+		return nil, fmt.Errorf("invalid transaction data: too short")
+	}
+
+	m, err := contracts.SwapCreatorParsedABI.MethodById(newSwapTopic[:4] /*data[:4]*/)
+	if err != nil {
+		return nil, err
+	}
+
+	newSwapInputs := make(map[string]interface{})
+
+	err = m.Inputs.UnpackIntoMap(newSwapInputs, data[4:])
+	if err != nil {
+		return nil, err
+	}
+
+	claimer := newSwapInputs["_claimer"].(ethcommon.Address)
+	cmtXMRMaker := newSwapInputs["_pubKeyClaim"].([32]byte)
+	cmtXMRTaker := newSwapInputs["_pubKeyRefund"].([32]byte)
+	asset := newSwapInputs["_asset"].(ethcommon.Address)
+	value := newSwapInputs["_value"].(*big.Int)
+	nonce := newSwapInputs["_nonce"].(*big.Int)
+
+	signer := ethtypes.LatestSignerForChainID(tx.ChainId())
+	from, err := ethtypes.Sender(signer, tx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &newSwapParameters{
+		swapCreatorAddr: *tx.To(),
+		owner:           from,
+		claimer:         claimer,
+		cmtXMRMaker:     cmtXMRMaker,
+		cmtXMRTaker:     cmtXMRTaker,
+		asset:           asset,
+		value:           value,
+		nonce:           nonce,
+	}, nil
+}

protocol/xmrtaker/instance_test.go

@@ -5,6 +5,7 @@ package xmrtaker
 
 import (
 	"errors"
+	"fmt"
 	"math/big"
 	"testing"
 
@@ -13,6 +14,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/athanorlabs/atomic-swap/coins"
+	"github.com/athanorlabs/atomic-swap/common"
 	"github.com/athanorlabs/atomic-swap/common/types"
 	mcrypto "github.com/athanorlabs/atomic-swap/crypto/monero"
 	"github.com/athanorlabs/atomic-swap/db"
@@ -86,3 +88,9 @@ func TestInstance_createOngoingSwap(t *testing.T) {
 	defer inst.swapMu.Unlock()
 	close(inst.swapStates[s.OfferID].done)
 }
+
+func TestNewSwapFunctionSignatureToTopic(t *testing.T) {
+	expected := "c41e46cf"
+	newSwapTopic := common.GetTopic(contracts.NewSwapFunctionSignature)
+	require.Equal(t, expected, fmt.Sprintf("%x", newSwapTopic[:4]))
+}

protocol/xmrtaker/message_handler.go

@@ -235,7 +235,7 @@ func (s *swapState) handleNotifyXMRLock() error {
 	close(s.xmrLockedCh)
 	log.Info("XMR was locked successfully, setting contract to ready...")
 
-	if err := s.ready(); err != nil {
+	if err := s.setReady(); err != nil {
 		return fmt.Errorf("failed to call Ready: %w", err)
 	}
 

protocol/xmrtaker/swap_state.go

@@ -24,6 +24,7 @@ import (
 	"github.com/athanorlabs/atomic-swap/db"
 	"github.com/athanorlabs/atomic-swap/dleq"
 	contracts "github.com/athanorlabs/atomic-swap/ethereum"
+	"github.com/athanorlabs/atomic-swap/ethereum/block"
 	"github.com/athanorlabs/atomic-swap/ethereum/watcher"
 	"github.com/athanorlabs/atomic-swap/monero"
 	"github.com/athanorlabs/atomic-swap/net/message"
@@ -39,7 +40,7 @@ import (
 
 const revertSwapCompleted = "swap is already completed"
 
-var claimedTopic = common.GetTopic(common.ClaimedEventSignature)
+var claimedTopic = common.GetTopic(contracts.ClaimedEventSignature)
 
 // swapState is an instance of a swap. it holds the info needed for the swap,
 // and its current state.
@@ -601,21 +602,12 @@ func (s *swapState) lockAsset() (*ethtypes.Receipt, error) {
 
 	cmtXMRTaker := s.secp256k1Pub.Keccak256()
 	cmtXMRMaker := s.xmrmakerSecp256k1PublicKey.Keccak256()
-	providedAmt := s.providedAmount
-
-	log.Debugf("locking %s %s in contract", providedAmt.AsStandard(), providedAmt.StandardSymbol())
+	log.Debugf("locking %s %s in contract", s.providedAmount.AsStandard(), s.providedAmount.StandardSymbol())
 
 	nonce := contracts.GenerateNewSwapNonce()
-	receipt, err := s.sender.NewSwap(
-		cmtXMRMaker,
-		cmtXMRTaker,
-		s.xmrmakerAddress,
-		big.NewInt(int64(s.SwapTimeout().Seconds())),
-		nonce,
-		providedAmt,
-	)
+	receipt, err := s.lockAndWaitForReceipt(cmtXMRMaker, cmtXMRTaker, nonce)
 	if err != nil {
-		return nil, fmt.Errorf("failed to instantiate swap on-chain: %w", err)
+		return nil, fmt.Errorf("failed to lock asset: %w", err)
 	}
 
 	log.Infof("instantiated swap on-chain: amount=%s asset=%s %s",
@@ -673,14 +665,46 @@ func (s *swapState) lockAsset() (*ethtypes.Receipt, error) {
 		return nil, err
 	}
 
-	log.Infof("locked %s in swap contract, waiting for XMR to be locked", providedAmt.StandardSymbol())
+	log.Infof("locked %s in swap contract, waiting for XMR to be locked", s.providedAmount.StandardSymbol())
 	return receipt, nil
 }
 
-// ready calls the Ready() method on the Swap contract, indicating to XMRMaker he has until time t_1 to
+func (s *swapState) lockAndWaitForReceipt(
+	cmtXMRMaker, cmtXMRTaker [32]byte,
+	nonce *big.Int,
+) (*ethtypes.Receipt, error) {
+	s.Backend.ETHClient().Lock()
+	defer s.Backend.ETHClient().Unlock()
+
+	txHash, err := s.sender.NewSwap(
+		cmtXMRMaker,
+		cmtXMRTaker,
+		s.xmrmakerAddress,
+		big.NewInt(int64(s.SwapTimeout().Seconds())),
+		nonce,
+		s.providedAmount,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to instantiate swap on-chain: %w", err)
+	}
+
+	err = s.Backend.RecoveryDB().PutNewSwapTxHash(s.OfferID(), txHash)
+	if err != nil {
+		return nil, fmt.Errorf("failed to write newSwap tx hash to db: %w", err)
+	}
+
+	receipt, err := block.WaitForReceipt(s.ctx, s.ETHClient().Raw(), txHash)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get newSwap transaction receipt: %w", err)
+	}
+
+	return receipt, nil
+}
+
+// ready calls the setReady() method on the Swap contract, indicating to XMRMaker he has until time t_1 to
 // call Claim(). Ready() should only be called once XMRTaker sees XMRMaker lock his XMR.
 // If time t_0 has passed, there is no point of calling Ready().
-func (s *swapState) ready() error {
+func (s *swapState) setReady() error {
 	stage, err := s.SwapCreator().Swaps(s.ETHClient().CallOpts(s.ctx), s.contractSwapID)
 	if err != nil {
 		return err

protocol/xmrtaker/swap_state_test.go

@@ -112,6 +112,7 @@ func newBackendAndNet(t *testing.T) (backend.Backend, *mockNet) {
 	rdb.EXPECT().PutSwapPrivateKey(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	rdb.EXPECT().PutCounterpartySwapPrivateKey(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	rdb.EXPECT().PutCounterpartySwapKeys(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	rdb.EXPECT().PutNewSwapTxHash(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	rdb.EXPECT().DeleteSwap(gomock.Any()).Return(nil).AnyTimes()
 
 	net := new(mockNet)