wallet: return unconfirmed balance in /wallet endpoint

api/bus.go

@@ -388,10 +388,11 @@ type GougingSettings struct {
 }
 
 type WalletResponse struct {
-	ScanHeight uint64         `json:"scanHeight"`
-	Address    types.Address  `json:"address"`
-	Spendable  types.Currency `json:"spendable"`
-	Confirmed  types.Currency `json:"confirmed"`
+	ScanHeight  uint64         `json:"scanHeight"`
+	Address     types.Address  `json:"address"`
+	Spendable   types.Currency `json:"spendable"`
+	Confirmed   types.Currency `json:"confirmed"`
+	Unconfirmed types.Currency `json:"unconfirmed"`
 }
 
 // Validate returns an error if the gouging settings are not considered valid.

bus/bus.go

@@ -53,7 +53,7 @@ type (
 	// A Wallet can spend and receive siacoins.
 	Wallet interface {
 		Address() types.Address
-		Balance() (spendable, confirmed types.Currency, _ error)
+		Balance() (spendable, confirmed, unconfirmed types.Currency, _ error)
 		FundTransaction(cs consensus.State, txn *types.Transaction, amount types.Currency, pool []types.Transaction) ([]types.Hash256, error)
 		Height() uint64
 		Redistribute(cs consensus.State, outputs int, amount, feePerByte types.Currency, pool []types.Transaction) (types.Transaction, []types.Hash256, error)
@@ -214,20 +214,21 @@ func (b *bus) txpoolBroadcastHandler(jc jape.Context) {
 
 func (b *bus) walletHandler(jc jape.Context) {
 	address := b.w.Address()
-	spendable, confirmed, err := b.w.Balance()
+	spendable, confirmed, unconfirmed, err := b.w.Balance()
 	if jc.Check("couldn't fetch wallet balance", err) != nil {
 		return
 	}
 	jc.Encode(api.WalletResponse{
-		ScanHeight: b.w.Height(),
-		Address:    address,
-		Confirmed:  confirmed,
-		Spendable:  spendable,
+		ScanHeight:  b.w.Height(),
+		Address:     address,
+		Confirmed:   confirmed,
+		Spendable:   spendable,
+		Unconfirmed: unconfirmed,
 	})
 }
 
 func (b *bus) walletBalanceHandler(jc jape.Context) {
-	_, balance, err := b.w.Balance()
+	_, balance, _, err := b.w.Balance()
 	if jc.Check("couldn't fetch wallet balance", err) != nil {
 		return
 	}
@@ -267,8 +268,10 @@ func (b *bus) walletFundHandler(jc jape.Context) {
 		return
 	}
 	txn := wfr.Transaction
-	fee := b.tp.RecommendedFee().Mul64(uint64(types.EncodedLen(txn)))
-	txn.MinerFees = []types.Currency{fee}
+	if len(txn.MinerFees) == 0 {
+		fee := b.tp.RecommendedFee().Mul64(uint64(types.EncodedLen(txn)))
+		txn.MinerFees = []types.Currency{fee}
+	}
 	toSign, err := b.w.FundTransaction(b.cm.TipState(jc.Request.Context()), &txn, wfr.Amount.Add(txn.MinerFees[0]), b.tp.Transactions())
 	if jc.Check("couldn't fund transaction", err) != nil {
 		return

internal/node/node.go

@@ -271,7 +271,8 @@ func NewBus(cfg BusConfig, dir string, seed types.PrivateKey, l *zap.Logger) (ht
 		}
 	}()
 
-	w := wallet.NewSingleAddressWallet(seed, sqlStore, cfg.UsedUTXOExpiry)
+	w := wallet.NewSingleAddressWallet(seed, sqlStore, cfg.UsedUTXOExpiry, zap.NewNop().Sugar())
+	tp.TransactionPoolSubscribe(w)
 
 	if m := cfg.Miner; m != nil {
 		if err := cs.ConsensusSetSubscribe(m, ccid, nil); err != nil {

internal/testing/cluster_test.go

@@ -48,24 +48,6 @@ func TestNewTestCluster(t *testing.T) {
 	b := cluster.Bus
 	w := cluster.Worker
 
-	// Check wallet info is sane after startup.
-	wi, err := b.Wallet(context.Background())
-	if err != nil {
-		t.Fatal(err)
-	}
-	if wi.ScanHeight == 0 {
-		t.Fatal("wallet scan height should not be 0")
-	}
-	if wi.Confirmed.IsZero() {
-		t.Fatal("wallet confirmed balance should not be zero")
-	}
-	if !wi.Spendable.Equals(wi.Confirmed) {
-		t.Fatal("wallet spendable balance should match confirmed")
-	}
-	if wi.Address == (types.Address{}) {
-		t.Fatal("wallet address should be set")
-	}
-
 	// Try talking to the bus API by adding an object.
 	err = b.AddObject(context.Background(), "foo", testAutopilotConfig.Contracts.Set, object.Object{
 		Key: object.GenerateEncryptionKey(),
@@ -1766,3 +1748,89 @@ func TestWalletTransactions(t *testing.T) {
 		}
 	}
 }
+
+func TestWallet(t *testing.T) {
+	if testing.Short() {
+		t.SkipNow()
+	}
+
+	cluster, err := newTestCluster(t.TempDir(), newTestLoggerCustom(zapcore.DebugLevel))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := cluster.Shutdown(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+	}()
+	b := cluster.Bus
+
+	// Check wallet info is sane after startup.
+	initialInfo, err := b.Wallet(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if initialInfo.ScanHeight == 0 {
+		t.Fatal("wallet scan height should not be 0")
+	}
+	if initialInfo.Confirmed.IsZero() {
+		t.Fatal("wallet confirmed balance should not be zero")
+	}
+	if !initialInfo.Spendable.Equals(initialInfo.Confirmed) {
+		t.Fatal("wallet spendable balance should match confirmed")
+	}
+	if !initialInfo.Unconfirmed.IsZero() {
+		t.Fatal("wallet unconfirmed balance should be zero")
+	}
+	if initialInfo.Address == (types.Address{}) {
+		t.Fatal("wallet address should be set")
+	}
+
+	// Send 1 SC to an address outside our wallet. We manually do this to be in
+	// control of the miner fees.
+	sendAmt := types.HastingsPerSiacoin
+	minerFee := types.NewCurrency64(1)
+	txn := types.Transaction{
+		SiacoinOutputs: []types.SiacoinOutput{
+			{Value: sendAmt, Address: types.VoidAddress},
+		},
+		MinerFees: []types.Currency{minerFee},
+	}
+	toSign, parents, err := b.WalletFund(context.Background(), &txn, txn.SiacoinOutputs[0].Value)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = b.WalletSign(context.Background(), &txn, toSign, types.CoveredFields{WholeTransaction: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := b.BroadcastTransaction(context.Background(), append(parents, txn)); err != nil {
+		t.Fatal(err)
+	}
+
+	// The wallet should still have the same confirmed balance, a lower
+	// spendable balance and a greater unconfirmed balance.
+	info, err := b.Wallet(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !info.Confirmed.Equals(initialInfo.Confirmed) {
+		t.Fatal("wallet confirmed balance should not be zero", info.Confirmed, initialInfo.Confirmed)
+	}
+	if info.Spendable.Cmp(initialInfo.Spendable) >= 0 {
+		t.Fatal("wallet spendable balance should be lower than before", info.Spendable, initialInfo.Spendable)
+	}
+	if info.Unconfirmed.Cmp(initialInfo.Unconfirmed) < 0 {
+		t.Fatal("wallet unconfirmed balance should be greater than before", info.Unconfirmed, initialInfo.Unconfirmed)
+	}
+
+	// The diffs of the spendable balance and unconfirmed balance should add up
+	// to the amount of money sent as well as the miner fees used.
+	spendableDiff := initialInfo.Spendable.Sub(info.Spendable)
+	unconfirmedDiff := info.Unconfirmed
+	withdrawnAmt := spendableDiff.Sub(unconfirmedDiff)
+	expectedWithdrawnAmt := sendAmt.Add(minerFee)
+	if !withdrawnAmt.Equals(expectedWithdrawnAmt) {
+		t.Fatal("withdrawn amt doesn't match expectation", withdrawnAmt.ExactString(), expectedWithdrawnAmt.ExactString())
+	}
+}

wallet/wallet.go

@@ -1,6 +1,7 @@
 package wallet
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"reflect"
@@ -11,7 +12,9 @@ import (
 	"gitlab.com/NebulousLabs/encoding"
 	"go.sia.tech/core/consensus"
 	"go.sia.tech/core/types"
+	"go.sia.tech/siad/modules"
 	stypes "go.sia.tech/siad/types"
+	"go.uber.org/zap"
 	"lukechampine.com/frand"
 )
 
@@ -121,6 +124,7 @@ type TransactionPool interface {
 // A SingleAddressWallet is a hot wallet that manages the outputs controlled by
 // a single address.
 type SingleAddressWallet struct {
+	log            *zap.SugaredLogger
 	priv           types.PrivateKey
 	addr           types.Address
 	store          SingleAddressStore
@@ -129,6 +133,15 @@ type SingleAddressWallet struct {
 	// for building transactions
 	mu       sync.Mutex
 	lastUsed map[types.Hash256]time.Time
+	// tpoolTxns maps a transaction set ID to the transactions in that set
+	tpoolTxns map[types.Hash256][]Transaction
+	// tpoolUtxos maps a siacoin output ID to its corresponding siacoin
+	// element. It is used to track siacoin outputs that are currently in
+	// the transaction pool.
+	tpoolUtxos map[types.SiacoinOutputID]SiacoinElement
+	// tpoolSpent is a set of siacoin output IDs that are currently in the
+	// transaction pool.
+	tpoolSpent map[types.SiacoinOutputID]bool
 }
 
 // PrivateKey returns the private key of the wallet.
@@ -142,10 +155,10 @@ func (w *SingleAddressWallet) Address() types.Address {
 }
 
 // Balance returns the balance of the wallet.
-func (w *SingleAddressWallet) Balance() (spendable, confirmed types.Currency, _ error) {
+func (w *SingleAddressWallet) Balance() (spendable, confirmed, unconfirmed types.Currency, _ error) {
 	sces, err := w.store.UnspentSiacoinElements(true)
 	if err != nil {
-		return types.Currency{}, types.Currency{}, err
+		return types.Currency{}, types.Currency{}, types.Currency{}, err
 	}
 	w.mu.Lock()
 	defer w.mu.Unlock()
@@ -155,6 +168,9 @@ func (w *SingleAddressWallet) Balance() (spendable, confirmed types.Currency, _
 		}
 		confirmed = confirmed.Add(sce.Value)
 	}
+	for _, sco := range w.tpoolUtxos {
+		unconfirmed = unconfirmed.Add(sco.Value)
+	}
 	return
 }
 
@@ -383,6 +399,101 @@ func (w *SingleAddressWallet) isOutputUsed(id types.Hash256) bool {
 	return time.Since(lastUsed) <= w.usedUTXOExpiry
 }
 
+// ReceiveUpdatedUnconfirmedTransactions implements modules.TransactionPoolSubscriber.
+func (w *SingleAddressWallet) ReceiveUpdatedUnconfirmedTransactions(diff *modules.TransactionPoolDiff) {
+	siacoinOutputs := make(map[types.SiacoinOutputID]SiacoinElement)
+	utxos, err := w.store.UnspentSiacoinElements(false)
+	if err != nil {
+		return
+	}
+	for _, output := range utxos {
+		siacoinOutputs[types.SiacoinOutputID(output.ID)] = output
+	}
+
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	for id, output := range w.tpoolUtxos {
+		siacoinOutputs[id] = output
+	}
+
+	for _, txnsetID := range diff.RevertedTransactions {
+		txns, ok := w.tpoolTxns[types.Hash256(txnsetID)]
+		if !ok {
+			continue
+		}
+		for _, txn := range txns {
+			for _, sci := range txn.Raw.SiacoinInputs {
+				delete(w.tpoolSpent, sci.ParentID)
+			}
+			for i := range txn.Raw.SiacoinOutputs {
+				delete(w.tpoolUtxos, txn.Raw.SiacoinOutputID(i))
+			}
+		}
+		delete(w.tpoolTxns, types.Hash256(txnsetID))
+	}
+
+	currentHeight := w.store.Height()
+
+	for _, txnset := range diff.AppliedTransactions {
+		var relevantTxns []Transaction
+
+	txnLoop:
+		for _, stxn := range txnset.Transactions {
+			var relevant bool
+			var txn types.Transaction
+			convertToCore(stxn, &txn)
+			processed := Transaction{
+				ID: txn.ID(),
+				Index: types.ChainIndex{
+					Height: currentHeight + 1,
+				},
+				Raw:       txn,
+				Timestamp: time.Now(),
+			}
+			for _, sci := range txn.SiacoinInputs {
+				if sci.UnlockConditions.UnlockHash() != w.addr {
+					continue
+				}
+				relevant = true
+				w.tpoolSpent[sci.ParentID] = true
+
+				output, ok := siacoinOutputs[sci.ParentID]
+				if !ok {
+					// note: happens during deep reorgs. Possibly a race
+					// condition in siad. Log and skip.
+					w.log.Debug("tpool transaction unknown utxo", zap.Stringer("outputID", sci.ParentID), zap.Stringer("txnID", txn.ID()))
+					continue txnLoop
+				}
+				processed.Outflow = processed.Outflow.Add(output.Value)
+			}
+
+			for i, sco := range txn.SiacoinOutputs {
+				if sco.Address != w.addr {
+					continue
+				}
+				relevant = true
+				outputID := txn.SiacoinOutputID(i)
+				processed.Inflow = processed.Inflow.Add(sco.Value)
+				sce := SiacoinElement{
+					ID:            types.Hash256(outputID),
+					SiacoinOutput: sco,
+				}
+				siacoinOutputs[outputID] = sce
+				w.tpoolUtxos[outputID] = sce
+			}
+
+			if relevant {
+				relevantTxns = append(relevantTxns, processed)
+			}
+		}
+
+		if len(relevantTxns) != 0 {
+			w.tpoolTxns[types.Hash256(txnset.ID)] = relevantTxns
+		}
+	}
+}
+
 // SumOutputs returns the total value of the supplied outputs.
 func SumOutputs(outputs []SiacoinElement) (sum types.Currency) {
 	for _, o := range outputs {
@@ -392,12 +503,27 @@ func SumOutputs(outputs []SiacoinElement) (sum types.Currency) {
 }
 
 // NewSingleAddressWallet returns a new SingleAddressWallet using the provided private key and store.
-func NewSingleAddressWallet(priv types.PrivateKey, store SingleAddressStore, usedUTXOExpiry time.Duration) *SingleAddressWallet {
+func NewSingleAddressWallet(priv types.PrivateKey, store SingleAddressStore, usedUTXOExpiry time.Duration, log *zap.SugaredLogger) *SingleAddressWallet {
 	return &SingleAddressWallet{
 		priv:           priv,
 		addr:           StandardAddress(priv.PublicKey()),
 		store:          store,
 		lastUsed:       make(map[types.Hash256]time.Time),
 		usedUTXOExpiry: usedUTXOExpiry,
+		tpoolTxns:      make(map[types.Hash256][]Transaction),
+		tpoolUtxos:     make(map[types.SiacoinOutputID]SiacoinElement),
+		tpoolSpent:     make(map[types.SiacoinOutputID]bool),
+		log:            log.Named("wallet"),
+	}
+}
+
+// convertToCore converts a siad type to an equivalent core type.
+func convertToCore(siad encoding.SiaMarshaler, core types.DecoderFrom) {
+	var buf bytes.Buffer
+	siad.MarshalSia(&buf)
+	d := types.NewBufDecoder(buf.Bytes())
+	core.DecodeFrom(d)
+	if d.Err() != nil {
+		panic(d.Err())
 	}
 }