Make kaspawallet ignore outputs that exist in the mempool (#2053)

* Make kaspawallet ignore outputs that exist in the mempool

* Make kaspawallet ignore outputs that exist in the mempool

* Add comment

app/rpc/rpchandlers/get_mempool_entries_by_addresses.go

@@ -3,6 +3,7 @@ package rpchandlers
 import (
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/app/rpc/rpccontext"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/consensushashing"
 	"github.com/kaspanet/kaspad/domain/consensus/utils/txscript"
 	"github.com/kaspanet/kaspad/infrastructure/network/netadapter/router"
 	"github.com/kaspanet/kaspad/util"
@@ -29,7 +30,13 @@ func HandleGetMempoolEntriesByAddresses(context *rpccontext.Context, _ *router.R
 
 		for _, transaction := range transactions {
 
-			for _, input := range transaction.Inputs {
+			for i, input := range transaction.Inputs {
+				// TODO: Fix this
+				if input.UTXOEntry == nil {
+					log.Errorf("Couldn't find UTXO entry for input %d in mempool transaction %s. This is a bug and should be fixed.", i, consensushashing.TransactionID(transaction))
+					continue
+				}
+
 				_, transactionSendingAddress, err := txscript.ExtractScriptPubKeyAddress(
 					input.UTXOEntry.ScriptPublicKey(),
 					context.Config.ActiveNetParams)

cmd/kaspawallet/daemon/server/broadcast.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"context"
-
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/cmd/kaspawallet/daemon/pb"
 	"github.com/kaspanet/kaspad/cmd/kaspawallet/libkaspawallet"
@@ -10,9 +9,13 @@ import (
 	"github.com/kaspanet/kaspad/domain/consensus/model/externalapi"
 	"github.com/kaspanet/kaspad/infrastructure/network/rpcclient"
 	"github.com/pkg/errors"
+	"time"
 )
 
 func (s *server) Broadcast(_ context.Context, request *pb.BroadcastRequest) (*pb.BroadcastResponse, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	txIDs, err := s.broadcast(request.Transactions, request.IsDomain)
 	if err != nil {
 		return nil, err
@@ -45,6 +48,15 @@ func (s *server) broadcast(transactions [][]byte, isDomain bool) ([]string, erro
 		if err != nil {
 			return nil, err
 		}
+
+		for _, input := range tx.Inputs {
+			s.usedOutpoints[input.PreviousOutpoint] = time.Now()
+		}
+	}
+
+	err = s.refreshUTXOs()
+	if err != nil {
+		return nil, err
 	}
 
 	return txIDs, nil

cmd/kaspawallet/daemon/server/create_unsigned_transaction.go

@@ -3,13 +3,13 @@ package server
 import (
 	"context"
 	"fmt"
-	"golang.org/x/exp/slices"
-
 	"github.com/kaspanet/kaspad/cmd/kaspawallet/daemon/pb"
 	"github.com/kaspanet/kaspad/cmd/kaspawallet/libkaspawallet"
 	"github.com/kaspanet/kaspad/domain/consensus/utils/constants"
 	"github.com/kaspanet/kaspad/util"
 	"github.com/pkg/errors"
+	"golang.org/x/exp/slices"
+	"time"
 )
 
 // TODO: Implement a better fee estimation mechanism
@@ -103,6 +103,14 @@ func (s *server) selectUTXOs(spendAmount uint64, feePerInput uint64, fromAddress
 			continue
 		}
 
+		if broadcastTime, ok := s.usedOutpoints[*utxo.Outpoint]; ok {
+			if time.Since(broadcastTime) > time.Minute {
+				delete(s.usedOutpoints, *utxo.Outpoint)
+			} else {
+				continue
+			}
+		}
+
 		selectedUTXOs = append(selectedUTXOs, &libkaspawallet.UTXO{
 			Outpoint:       utxo.Outpoint,
 			UTXOEntry:      utxo.UTXOEntry,

cmd/kaspawallet/daemon/server/server.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"fmt"
+	"github.com/kaspanet/kaspad/domain/consensus/model/externalapi"
 	"net"
 	"os"
 	"sync"
@@ -35,6 +36,7 @@ type server struct {
 	shutdown            chan struct{}
 	addressSet          walletAddressSet
 	txMassCalculator    *txmass.Calculator
+	usedOutpoints       map[externalapi.DomainOutpoint]time.Time
 }
 
 // Start starts the kaspawalletd server
@@ -73,6 +75,7 @@ func Start(params *dagconfig.Params, listen, rpcServer string, keysFilePath stri
 		shutdown:            make(chan struct{}),
 		addressSet:          make(walletAddressSet),
 		txMassCalculator:    txmass.NewCalculator(params.MassPerTxByte, params.MassPerScriptPubKeyByte, params.MassPerSigOp),
+		usedOutpoints:       map[externalapi.DomainOutpoint]time.Time{},
 	}
 
 	spawn("serverInstance.sync", func() {

cmd/kaspawallet/daemon/server/sync.go

@@ -189,10 +189,23 @@ func (s *server) refreshExistingUTXOsWithLock() error {
 }
 
 // updateUTXOSet clears the current UTXO set, and re-fills it with the given entries
-func (s *server) updateUTXOSet(entries []*appmessage.UTXOsByAddressesEntry) error {
+func (s *server) updateUTXOSet(entries []*appmessage.UTXOsByAddressesEntry, mempoolEntries []*appmessage.MempoolEntryByAddress) error {
-	utxos := make([]*walletUTXO, len(entries))
+	utxos := make([]*walletUTXO, 0, len(entries))
+
+	exclude := make(map[appmessage.RPCOutpoint]struct{})
+	for _, entriesByAddress := range mempoolEntries {
+		for _, entry := range entriesByAddress.Sending {
+			for _, input := range entry.Transaction.Inputs {
+				exclude[*input.PreviousOutpoint] = struct{}{}
+			}
+		}
+	}
+
+	for _, entry := range entries {
+		if _, ok := exclude[*entry.Outpoint]; ok {
+			continue
+		}
 
-	for i, entry := range entries {
 		outpoint, err := appmessage.RPCOutpointToDomainOutpoint(entry.Outpoint)
 		if err != nil {
 			return err
@@ -207,11 +220,11 @@ func (s *server) updateUTXOSet(entries []*appmessage.UTXOsByAddressesEntry) erro
 		if !ok {
 			return errors.Errorf("Got result from address %s even though it wasn't requested", entry.Address)
 		}
-		utxos[i] = &walletUTXO{
+		utxos = append(utxos, &walletUTXO{
 			Outpoint:  outpoint,
 			UTXOEntry: utxoEntry,
 			address:   address,
-		}
+		})
 	}
 
 	sort.Slice(utxos, func(i, j int) bool { return utxos[i].UTXOEntry.Amount() > utxos[j].UTXOEntry.Amount() })
@@ -222,12 +235,22 @@ func (s *server) updateUTXOSet(entries []*appmessage.UTXOsByAddressesEntry) erro
 }
 
 func (s *server) refreshUTXOs() error {
+	// It's important to check the mempool before calling `GetUTXOsByAddresses`:
+	// If we would do it the other way around an output can be spent in the mempool
+	// and not in consensus, and between the calls its spending transaction will be
+	// added to consensus and removed from the mempool, so `getUTXOsByAddressesResponse`
+	// will include an obsolete output.
+	mempoolEntriesByAddresses, err := s.rpcClient.GetMempoolEntriesByAddresses(s.addressSet.strings())
+	if err != nil {
+		return err
+	}
+
 	getUTXOsByAddressesResponse, err := s.rpcClient.GetUTXOsByAddresses(s.addressSet.strings())
 	if err != nil {
 		return err
 	}
 
-	return s.updateUTXOSet(getUTXOsByAddressesResponse.Entries)
+	return s.updateUTXOSet(getUTXOsByAddressesResponse.Entries, mempoolEntriesByAddresses.Entries)
 }
 
 func (s *server) isSynced() bool {