use SingleAddressWallet from coreutils

api/wallet.go

@@ -10,6 +10,26 @@ import (
 	"go.sia.tech/core/types"
 )
 
+type (
+	// A SiacoinElement is a SiacoinOutput along with its ID.
+	SiacoinElement struct {
+		types.SiacoinOutput
+		ID             types.Hash256 `json:"id"`
+		MaturityHeight uint64        `json:"maturityHeight"`
+	}
+
+	// A Transaction is an on-chain transaction relevant to a particular wallet,
+	// paired with useful metadata.
+	Transaction struct {
+		Raw       types.Transaction   `json:"raw,omitempty"`
+		Index     types.ChainIndex    `json:"index"`
+		ID        types.TransactionID `json:"id"`
+		Inflow    types.Currency      `json:"inflow"`
+		Outflow   types.Currency      `json:"outflow"`
+		Timestamp time.Time           `json:"timestamp"`
+	}
+)
+
 type (
 	// WalletFundRequest is the request type for the /wallet/fund endpoint.
 	WalletFundRequest struct {
@@ -73,6 +93,7 @@ type (
 		Spendable   types.Currency `json:"spendable"`
 		Confirmed   types.Currency `json:"confirmed"`
 		Unconfirmed types.Currency `json:"unconfirmed"`
+		Immature    types.Currency `json:"immature"`
 	}
 
 	// WalletSignRequest is the request type for the /wallet/sign endpoint.

autopilot/autopilot.go

@@ -19,7 +19,6 @@ import (
 	"go.sia.tech/renterd/build"
 	"go.sia.tech/renterd/hostdb"
 	"go.sia.tech/renterd/object"
-	"go.sia.tech/renterd/wallet"
 	"go.sia.tech/renterd/webhooks"
 	"go.sia.tech/renterd/worker"
 	"go.uber.org/zap"
@@ -83,7 +82,7 @@ type Bus interface {
 	// wallet
 	Wallet(ctx context.Context) (api.WalletResponse, error)
 	WalletDiscard(ctx context.Context, txn types.Transaction) error
-	WalletOutputs(ctx context.Context) (resp []wallet.SiacoinElement, err error)
+	WalletOutputs(ctx context.Context) (resp []api.SiacoinElement, err error)
 	WalletPending(ctx context.Context) (resp []types.Transaction, err error)
 	WalletRedistribute(ctx context.Context, outputs int, amount types.Currency) (ids []types.TransactionID, err error)
 }
@@ -240,12 +239,18 @@ func (ap *Autopilot) Run() error {
 		ap.workers.withWorker(func(w Worker) {
 			defer ap.logger.Info("autopilot iteration ended")
 
+			// log worker id chosen for this maintenance iteration.
+			workerID, err := w.ID(ap.shutdownCtx)
+			if err != nil {
+				ap.logger.Warn("failed to reach worker, err: %v", err)
+			} else {
+				ap.logger.Infof("using worker %s for iteration", workerID)
+			}
+
 			// initiate a host scan - no need to be synced or configured for scanning
 			ap.s.tryUpdateTimeout()
 			ap.s.tryPerformHostScan(ap.shutdownCtx, w, forceScan)
-
-			// reset forceScan
-			forceScan = false
+			forceScan = false // reset forceScan
 
 			// block until consensus is synced
 			if synced, blocked, interrupted := ap.blockUntilSynced(ap.ticker.C); !synced {
@@ -271,14 +276,6 @@ func (ap *Autopilot) Run() error {
 				return
 			}
 
-			// Log worker id chosen for this maintenance iteration.
-			workerID, err := w.ID(ap.shutdownCtx)
-			if err != nil {
-				ap.logger.Errorf("aborting maintenance, failed to fetch worker id, err: %v", err)
-				return
-			}
-			ap.logger.Infof("using worker %s for iteration", workerID)
-
 			// update the loop state
 			//
 			// NOTE: it is important this is the first action we perform in this

autopilot/contractor.go

@@ -14,9 +14,9 @@ import (
 	rhpv2 "go.sia.tech/core/rhp/v2"
 	rhpv3 "go.sia.tech/core/rhp/v3"
 	"go.sia.tech/core/types"
+	"go.sia.tech/coreutils/wallet"
 	"go.sia.tech/renterd/api"
 	"go.sia.tech/renterd/hostdb"
-	"go.sia.tech/renterd/wallet"
 	"go.sia.tech/renterd/worker"
 	"go.uber.org/zap"
 )
@@ -1425,7 +1425,7 @@ func (c *contractor) renewContract(ctx context.Context, w Worker, ci contractInf
 			"renterFunds", renterFunds,
 			"expectedNewStorage", expectedNewStorage,
 		)
-		if strings.Contains(err.Error(), wallet.ErrInsufficientBalance.Error()) {
+		if isErr(err, wallet.ErrNotEnoughFunds) {
 			return api.ContractMetadata{}, false, err
 		}
 		return api.ContractMetadata{}, true, err
@@ -1508,7 +1508,7 @@ func (c *contractor) refreshContract(ctx context.Context, w Worker, ci contractI
 			return api.ContractMetadata{}, true, err
 		}
 		c.logger.Errorw("refresh failed", zap.Error(err), "hk", hk, "fcid", fcid)
-		if strings.Contains(err.Error(), wallet.ErrInsufficientBalance.Error()) {
+		if isErr(err, wallet.ErrNotEnoughFunds) {
 			return api.ContractMetadata{}, false, err
 		}
 		return api.ContractMetadata{}, true, err
@@ -1572,7 +1572,7 @@ func (c *contractor) formContract(ctx context.Context, w Worker, host hostdb.Hos
 	if err != nil {
 		// TODO: keep track of consecutive failures and break at some point
 		c.logger.Errorw(fmt.Sprintf("contract formation failed, err: %v", err), "hk", hk)
-		if strings.Contains(err.Error(), wallet.ErrInsufficientBalance.Error()) {
+		if isErr(err, wallet.ErrNotEnoughFunds) {
 			return api.ContractMetadata{}, false, err
 		}
 		return api.ContractMetadata{}, true, err

autopilot/migrator.go

@@ -144,7 +144,7 @@ func (m *migrator) performMigrations(p *workerPool) {
 					// fetch worker id once
 					id, err := w.ID(ctx)
 					if err != nil {
-						m.logger.Errorf("failed to fetch worker id: %v", err)
+						m.logger.Errorf("failed to reach worker, err: %v", err)
 						return
 					}
 

autopilot/scanner.go

@@ -193,13 +193,13 @@ func (s *scanner) tryPerformHostScan(ctx context.Context, w scanWorker, force bo
 	s.logger.Infof("%s started", scanType)
 
 	s.wg.Add(1)
+	s.ap.wg.Add(1)
 	go func(st string) {
 		defer s.wg.Done()
+		defer s.ap.wg.Done()
 
-		var interrupted bool
 		for resp := range s.launchScanWorkers(ctx, w, s.launchHostScans()) {
 			if s.isInterrupted() || s.ap.isStopped() {
-				interrupted = true
 				break
 			}
 			if resp.err != nil && !strings.Contains(resp.err.Error(), "connection refused") {
@@ -212,8 +212,7 @@ func (s *scanner) tryPerformHostScan(ctx context.Context, w scanWorker, force bo
 		hostCfg := s.ap.State().cfg.Hosts
 		maxDowntime := time.Duration(hostCfg.MaxDowntimeHours) * time.Hour
 		minRecentScanFailures := hostCfg.MinRecentScanFailures
-
-		if !interrupted && maxDowntime > 0 {
+		if !s.ap.isStopped() && maxDowntime > 0 {
 			s.logger.Debugf("removing hosts that have been offline for more than %v and have failed at least %d scans", maxDowntime, minRecentScanFailures)
 			removed, err := s.bus.RemoveOfflineHosts(ctx, minRecentScanFailures, maxDowntime)
 			if err != nil {
@@ -253,10 +252,7 @@ func (s *scanner) tryUpdateTimeout() {
 
 func (s *scanner) launchHostScans() chan scanReq {
 	reqChan := make(chan scanReq, s.scanBatchSize)
-
-	s.ap.wg.Add(1)
 	go func() {
-		defer s.ap.wg.Done()
 		defer close(reqChan)
 
 		var offset int

autopilot/scanner_test.go

@@ -87,7 +87,7 @@ func TestScanner(t *testing.T) {
 	// init new scanner
 	b := &mockBus{hosts: hosts}
 	w := &mockWorker{blockChan: make(chan struct{})}
-	s := newTestScanner(b, w)
+	s := newTestScanner(b)
 
 	// assert it started a host scan
 	s.tryPerformHostScan(context.Background(), w, false)
@@ -139,7 +139,7 @@ func (s *scanner) isScanning() bool {
 	return s.scanning
 }
 
-func newTestScanner(b *mockBus, w *mockWorker) *scanner {
+func newTestScanner(b *mockBus) *scanner {
 	ap := &Autopilot{}
 	ap.shutdownCtx, ap.shutdownCtxCancel = context.WithCancel(context.Background())
 	return &scanner{