Defrag wallet implicitly and reduce wallet maintenance to 10 outputs

README.md

@@ -177,9 +177,6 @@ formed.
 
 ```json
 {
-	"wallet": {
-		"defragThreshold": 1000
-	},
 	"hosts": {
 		"allowRedundantIPs": false,
 		"maxDowntimeHours": 1440,

api/autopilot.go

@@ -39,7 +39,6 @@ type (
 	AutopilotConfig struct {
 		Contracts ContractsConfig `json:"contracts"`
 		Hosts     HostsConfig     `json:"hosts"`
-		Wallet    WalletConfig    `json:"wallet"`
 	}
 
 	// ContractsConfig contains all contract settings used in the autopilot.
@@ -62,11 +61,6 @@ type (
 		MinRecentScanFailures uint64                      `json:"minRecentScanFailures"`
 		ScoreOverrides        map[types.PublicKey]float64 `json:"scoreOverrides"`
 	}
-
-	// WalletConfig contains all wallet settings used in the autopilot.
-	WalletConfig struct {
-		DefragThreshold uint64 `json:"defragThreshold"`
-	}
 )
 
 type (

autopilot/contractor.go

@@ -578,31 +578,26 @@ func (c *contractor) performWalletMaintenance(ctx context.Context) error {
 		}
 	}
 
+	wantedNumOutputs := 10
+
 	// enough outputs - nothing to do
 	available, err := b.WalletOutputs(ctx)
 	if err != nil {
 		return err
 	}
-	if uint64(len(available)) >= cfg.Contracts.Amount {
-		l.Debugf("no wallet maintenance needed, plenty of outputs available (%v>=%v)", len(available), cfg.Contracts.Amount)
+	if uint64(len(available)) >= uint64(wantedNumOutputs) {
+		l.Debugf("no wallet maintenance needed, plenty of outputs available (%v>=%v)", len(available), uint64(wantedNumOutputs))
 		return nil
 	}
+	wantedNumOutputs -= len(available)
 
-	// not enough balance to redistribute outputs - nothing to do
-	amount := cfg.Contracts.Allowance.Div64(cfg.Contracts.Amount)
-	outputs := balance.Div(amount).Big().Uint64()
-	if outputs < 2 {
-		l.Warnf("wallet maintenance skipped, wallet has insufficient balance %v", balance)
-		return err
-	}
-	if outputs > cfg.Contracts.Amount {
-		outputs = cfg.Contracts.Amount
-	}
+	// figure out the amount per output
+	amount := cfg.Contracts.Allowance.Div64(uint64(wantedNumOutputs))
 
 	// redistribute outputs
-	ids, err := b.WalletRedistribute(ctx, int(outputs), amount)
+	ids, err := b.WalletRedistribute(ctx, wantedNumOutputs, amount)
 	if err != nil {
-		return fmt.Errorf("failed to redistribute wallet into %d outputs of amount %v, balance %v, err %v", outputs, amount, balance, err)
+		return fmt.Errorf("failed to redistribute wallet into %d outputs of amount %v, balance %v, err %v", wantedNumOutputs, amount, balance, err)
 	}
 
 	l.Debugf("wallet maintenance succeeded, txns %v", ids)

autopilot/hostscore_test.go

@@ -29,9 +29,6 @@ var cfg = api.AutopilotConfig{
 		MaxDowntimeHours:      24 * 7 * 2,
 		MinRecentScanFailures: 10,
 	},
-	Wallet: api.WalletConfig{
-		DefragThreshold: 1000,
-	},
 }
 
 func TestHostScore(t *testing.T) {

go.mod

@@ -13,7 +13,7 @@ require (
 	gitlab.com/NebulousLabs/encoding v0.0.0-20200604091946-456c3dc907fe
 	go.sia.tech/core v0.1.12-0.20231211182757-77190f04f90b
 	go.sia.tech/gofakes3 v0.0.0-20231109151325-e0d47c10dce2
-	go.sia.tech/hostd v0.2.2
+	go.sia.tech/hostd v0.3.0-beta.1
 	go.sia.tech/jape v0.11.1
 	go.sia.tech/mux v1.2.0
 	go.sia.tech/siad v1.5.10-0.20230228235644-3059c0b930ca

go.sum

@@ -235,8 +235,8 @@ go.sia.tech/core v0.1.12-0.20231211182757-77190f04f90b h1:xJSxYN2kZD3NAijHIwjXhG
 go.sia.tech/core v0.1.12-0.20231211182757-77190f04f90b/go.mod h1:3EoY+rR78w1/uGoXXVqcYdwSjSJKuEMI5bL7WROA27Q=
 go.sia.tech/gofakes3 v0.0.0-20231109151325-e0d47c10dce2 h1:ulzfJNjxN5DjXHClkW2pTiDk+eJ+0NQhX87lFDZ03t0=
 go.sia.tech/gofakes3 v0.0.0-20231109151325-e0d47c10dce2/go.mod h1:PlsiVCn6+wssrR7bsOIlZm0DahsVrDydrlbjY4F14sg=
-go.sia.tech/hostd v0.2.2 h1:HbXB4WripvVFUSpTrckEhC8DteL+QpXKZeonJHUpq3M=
-go.sia.tech/hostd v0.2.2/go.mod h1:t1hzcQUFyNnP1mW1AxdFiAya2uyUz5lGLuCVehSkXkg=
+go.sia.tech/hostd v0.3.0-beta.1 h1:A2RL4wkW18eb28+fJtdyK9OYNiiwpCDO8FO3cyT9r7A=
+go.sia.tech/hostd v0.3.0-beta.1/go.mod h1:gVtU631RkbtOEHJKb8qghudhWcYIL8w3phjvV2/bz0A=
 go.sia.tech/jape v0.11.1 h1:M7IP+byXL7xOqzxcHUQuXW+q3sYMkYzmMlMw+q8ZZw0=
 go.sia.tech/jape v0.11.1/go.mod h1:4QqmBB+t3W7cNplXPj++ZqpoUb2PeiS66RLpXmEGap4=
 go.sia.tech/mux v1.2.0 h1:ofa1Us9mdymBbGMY2XH/lSpY8itFsKIo/Aq8zwe+GHU=

internal/testing/host.go

@@ -22,6 +22,7 @@ import (
 	rhpv2 "go.sia.tech/hostd/rhp/v2"
 	rhpv3 "go.sia.tech/hostd/rhp/v3"
 	"go.sia.tech/hostd/wallet"
+	"go.sia.tech/hostd/webhooks"
 	"go.sia.tech/renterd/bus"
 	"go.sia.tech/renterd/internal/node"
 	"go.sia.tech/siad/modules"
@@ -194,11 +195,17 @@ func NewHost(privKey types.PrivateKey, dir string, network *consensus.Network, d
 		return nil, fmt.Errorf("failed to create wallet: %w", err)
 	}
 
-	am := alerts.NewManager()
+	wr, err := webhooks.NewManager(db, log.Named("webhooks"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create webhook reporter: %w", err)
+	}
+
+	am := alerts.NewManager(wr, log.Named("alerts"))
 	storage, err := storage.NewVolumeManager(db, am, cm, log.Named("storage"), 0)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create storage manager: %w", err)
 	}
+
 	contracts, err := contracts.NewManager(db, am, storage, cm, tp, wallet, log.Named("contracts"))
 	if err != nil {
 		return nil, fmt.Errorf("failed to create contract manager: %w", err)

stores/autopilot_test.go

@@ -41,9 +41,6 @@ func TestAutopilotStore(t *testing.T) {
 			MinRecentScanFailures: 10,
 			AllowRedundantIPs:     true, // allow for integration tests by default
 		},
-		Wallet: api.WalletConfig{
-			DefragThreshold: 1234,
-		},
 	}
 
 	// add an autopilot with that config

wallet/wallet.go

@@ -25,6 +25,16 @@ const (
 	// redistributeBatchSize is the number of outputs to redistribute per txn to
 	// avoid creating a txn that is too large.
 	redistributeBatchSize = 10
+
+	// transactionDefragThreshold is the number of utxos at which the wallet
+	// will attempt to defrag itself by including small utxos in transactions.
+	transactionDefragThreshold = 30
+	// maxInputsForDefrag is the maximum number of inputs a transaction can
+	// have before the wallet will stop adding inputs
+	maxInputsForDefrag = 30
+	// maxDefragUTXOs is the maximum number of utxos that will be added to a
+	// transaction when defragging
+	maxDefragUTXOs = 10
 )
 
 // ErrInsufficientBalance is returned when there aren't enough unused outputs to
@@ -215,11 +225,11 @@ func (w *SingleAddressWallet) Transactions(before, since time.Time, offset, limi
 // inputs will not be available to future calls to FundTransaction unless
 // ReleaseInputs is called or enough time has passed.
 func (w *SingleAddressWallet) FundTransaction(cs consensus.State, txn *types.Transaction, amount types.Currency, useUnconfirmedTxns bool) ([]types.Hash256, error) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
 	if amount.IsZero() {
 		return nil, nil
 	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
 
 	// fetch all unspent siacoin elements
 	utxos, err := w.store.UnspentSiacoinElements(false)
@@ -245,34 +255,67 @@ func (w *SingleAddressWallet) FundTransaction(cs consensus.State, txn *types.Tra
 		utxos = append(utxos, tpoolUtxos...)
 	}
 
-	var outputSum types.Currency
-	var fundingElements []SiacoinElement
+	// remove locked and spent outputs
+	usableUTXOs := utxos[:0]
 	for _, sce := range utxos {
-		if w.isOutputUsed(sce.ID) || w.tpoolSpent[types.SiacoinOutputID(sce.ID)] || cs.Index.Height < sce.MaturityHeight {
+		if w.isOutputUsed(sce.ID) {
 			continue
 		}
-		fundingElements = append(fundingElements, sce)
-		outputSum = outputSum.Add(sce.Value)
-		if outputSum.Cmp(amount) >= 0 {
+		usableUTXOs = append(usableUTXOs, sce)
+	}
+
+	// fund the transaction using the largest utxos first
+	var selected []SiacoinElement
+	var inputSum types.Currency
+	for i, sce := range usableUTXOs {
+		if inputSum.Cmp(amount) >= 0 {
+			usableUTXOs = usableUTXOs[i:]
 			break
 		}
+		selected = append(selected, sce)
+		inputSum = inputSum.Add(sce.Value)
+	}
+
+	// if the transaction can't be funded, return an error
+	if inputSum.Cmp(amount) < 0 {
+		return nil, fmt.Errorf("%w: inputSum: %v, amount: %v", ErrInsufficientBalance, inputSum.String(), amount.String())
 	}
-	if outputSum.Cmp(amount) < 0 {
-		return nil, fmt.Errorf("%w: outputSum: %v, amount: %v", ErrInsufficientBalance, outputSum.String(), amount.String())
-	} else if outputSum.Cmp(amount) > 0 {
+
+	// check if remaining utxos should be defragged
+	txnInputs := len(txn.SiacoinInputs) + len(selected)
+	if len(usableUTXOs) > transactionDefragThreshold && txnInputs < maxInputsForDefrag {
+		// add the smallest utxos to the transaction
+		defraggable := usableUTXOs
+		if len(defraggable) > maxDefragUTXOs {
+			defraggable = defraggable[len(defraggable)-maxDefragUTXOs:]
+		}
+		for i := len(defraggable) - 1; i >= 0; i-- {
+			if txnInputs >= maxInputsForDefrag {
+				break
+			}
+
+			sce := defraggable[i]
+			selected = append(selected, sce)
+			inputSum = inputSum.Add(sce.Value)
+			txnInputs++
+		}
+	}
+
+	// add a change output if necessary
+	if inputSum.Cmp(amount) > 0 {
 		txn.SiacoinOutputs = append(txn.SiacoinOutputs, types.SiacoinOutput{
-			Value:   outputSum.Sub(amount),
+			Value:   inputSum.Sub(amount),
 			Address: w.addr,
 		})
 	}
 
-	toSign := make([]types.Hash256, len(fundingElements))
-	for i, sce := range fundingElements {
+	toSign := make([]types.Hash256, len(selected))
+	for i, sce := range selected {
 		txn.SiacoinInputs = append(txn.SiacoinInputs, types.SiacoinInput{
 			ParentID:         types.SiacoinOutputID(sce.ID),
-			UnlockConditions: StandardUnlockConditions(w.priv.PublicKey()),
+			UnlockConditions: types.StandardUnlockConditions(w.priv.PublicKey()),
 		})
-		toSign[i] = sce.ID
+		toSign[i] = types.Hash256(sce.ID)
 		w.lastUsed[sce.ID] = time.Now()
 	}