Use new usable hosts endpoint

internal/host/host.go

@@ -11,10 +11,14 @@ import (
 )
 
 type (
+	Downloader interface {
+		DownloadSector(ctx context.Context, w io.Writer, root types.Hash256, offset, length uint32, overpay bool) error
+		PublicKey() types.PublicKey
+	}
+
 	Host interface {
 		PublicKey() types.PublicKey
 
-		DownloadSector(ctx context.Context, w io.Writer, root types.Hash256, offset, length uint32, overpay bool) error
 		UploadSector(ctx context.Context, sectorRoot types.Hash256, sector *[rhpv2.SectorSize]byte, rev types.FileContractRevision) error
 
 		PriceTable(ctx context.Context, rev *types.FileContractRevision) (api.HostPriceTable, types.Currency, error)
@@ -25,6 +29,7 @@ type (
 	}
 
 	HostManager interface {
+		Downloader(hk types.PublicKey, siamuxAddr string) Downloader
 		Host(hk types.PublicKey, fcid types.FileContractID, siamuxAddr string) Host
 	}
 )

internal/test/e2e/cluster_test.go

@@ -2622,13 +2622,9 @@ func TestDownloadAllHosts(t *testing.T) {
 		t.SkipNow()
 	}
 
-	// get rid of redundancy
-	rs := test.RedundancySettings
-	rs.MinShards = rs.TotalShards
-
 	// create a test cluster
 	cluster := newTestCluster(t, testClusterOptions{
-		hosts:         rs.TotalShards,
+		hosts:         test.RedundancySettings.TotalShards,
 		uploadPacking: false, // make sure data is uploaded
 	})
 	defer cluster.Shutdown()
@@ -2637,12 +2633,6 @@ func TestDownloadAllHosts(t *testing.T) {
 	w := cluster.Worker
 	tt := cluster.tt
 
-	// update redundancy settings
-	us, err := b.UploadSettings(context.Background())
-	tt.OK(err)
-	us.Redundancy = rs
-	tt.OK(b.UpdateUploadSettings(context.Background(), us))
-
 	// prepare a file
 	data := make([]byte, 128)
 	tt.OKAll(frand.Read(data))
@@ -2664,7 +2654,7 @@ func TestDownloadAllHosts(t *testing.T) {
 			}
 		}
 	}
-	if len(usedHosts) != rs.TotalShards {
+	if len(usedHosts) != test.RedundancySettings.TotalShards {
 		t.Fatalf("unexpected number of used hosts %d", len(usedHosts))
 	}
 

internal/test/mocks/hoststore.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net"
 	"sync"
 	"time"
 
@@ -46,6 +47,25 @@ func (hs *HostStore) RecordContractSpending(ctx context.Context, records []api.C
 	return nil
 }
 
+func (hs *HostStore) UsableHosts(ctx context.Context) (hosts []api.HostInfo, _ error) {
+	hs.mu.Lock()
+	defer hs.mu.Unlock()
+
+	for _, h := range hs.hosts {
+		host, _, err := net.SplitHostPort(h.hi.NetAddress)
+		if err != nil || host == "" {
+			continue
+		}
+
+		hosts = append(hosts, api.HostInfo{
+			PublicKey:  h.hk,
+			SiamuxAddr: net.JoinHostPort(host, h.hi.Settings.SiaMuxPort),
+		})
+	}
+
+	return
+}
+
 func (hs *HostStore) AddHost() *Host {
 	hs.mu.Lock()
 	defer hs.mu.Unlock()
@@ -66,6 +86,10 @@ func NewHostManager() *HostManager {
 	}
 }
 
+func (hm *HostManager) Downloader(hk types.PublicKey, siamuxAddr string) host.Downloader {
+	return NewHost(hk)
+}
+
 func (hm *HostManager) Host(hk types.PublicKey, fcid types.FileContractID, siamuxAddr string) host.Host {
 	return NewHost(hk)
 }

internal/worker/accounts.go

@@ -45,12 +45,12 @@ type (
 		UpdateAccounts(context.Context, []api.Account) error
 	}
 
-	ConsensusState interface {
+	ConsensusStateStore interface {
 		ConsensusState(ctx context.Context) (api.ConsensusState, error)
 	}
 
-	DownloadContracts interface {
-		DownloadContracts(ctx context.Context) ([]api.ContractMetadata, error)
+	ContractStore interface {
+		Contracts(ctx context.Context, opts api.ContractsOpts) ([]api.ContractMetadata, error)
 	}
 )
 
@@ -59,8 +59,8 @@ type (
 		alerts                   alerts.Alerter
 		funder                   AccountFunder
 		syncer                   AccountSyncer
-		dc                       DownloadContracts
-		cs                       ConsensusState
+		cs                       ContractStore
+		css                      ConsensusStateStore
 		s                        AccountStore
 		key                      utils.AccountsKey
 		logger                   *zap.SugaredLogger
@@ -92,7 +92,7 @@ type (
 // NewAccountManager creates a new account manager. It will load all accounts
 // from the given store and mark the shutdown as unclean. When Shutdown is
 // called it will save all accounts.
-func NewAccountManager(key utils.AccountsKey, owner string, alerter alerts.Alerter, funder AccountFunder, syncer AccountSyncer, cs ConsensusState, dc DownloadContracts, s AccountStore, refillInterval time.Duration, l *zap.Logger) (*AccountMgr, error) {
+func NewAccountManager(key utils.AccountsKey, owner string, alerter alerts.Alerter, funder AccountFunder, syncer AccountSyncer, css ConsensusStateStore, cs ContractStore, s AccountStore, refillInterval time.Duration, l *zap.Logger) (*AccountMgr, error) {
 	logger := l.Named("accounts").Sugar()
 
 	shutdownCtx, shutdownCancel := context.WithCancel(context.Background())
@@ -101,7 +101,7 @@ func NewAccountManager(key utils.AccountsKey, owner string, alerter alerts.Alert
 		funder: funder,
 		syncer: syncer,
 		cs:     cs,
-		dc:     dc,
+		css:    css,
 		s:      s,
 		key:    key,
 		logger: logger,
@@ -310,14 +310,14 @@ func (a *AccountMgr) markRefillDone(hk types.PublicKey) {
 // until the previously launched goroutine returns.
 func (a *AccountMgr) refillAccounts() {
 	// fetch config
-	cs, err := a.cs.ConsensusState(a.shutdownCtx)
+	cs, err := a.css.ConsensusState(a.shutdownCtx)
 	if err != nil {
 		a.logger.Errorw(fmt.Sprintf("failed to fetch consensus state for refill: %v", err))
 		return
 	}
 
 	// fetch all contracts
-	contracts, err := a.dc.DownloadContracts(a.shutdownCtx)
+	contracts, err := a.cs.Contracts(a.shutdownCtx, api.ContractsOpts{})
 	if err != nil {
 		a.logger.Errorw(fmt.Sprintf("failed to fetch contracts for refill: %v", err))
 		return

internal/worker/accounts_test.go

@@ -45,7 +45,7 @@ func (b *mockAccountMgrBackend) UpdateAccounts(context.Context, []api.Account) e
 func (b *mockAccountMgrBackend) ConsensusState(ctx context.Context) (api.ConsensusState, error) {
 	return api.ConsensusState{}, nil
 }
-func (b *mockAccountMgrBackend) DownloadContracts(ctx context.Context) ([]api.ContractMetadata, error) {
+func (b *mockAccountMgrBackend) Contracts(ctx context.Context, opts api.ContractsOpts) ([]api.ContractMetadata, error) {
 	return nil, nil
 }
 

internal/worker/cache.go

@@ -18,8 +18,8 @@ import (
 )
 
 const (
-	cacheKeyDownloadContracts = "downloadcontracts"
-	cacheKeyGougingParams     = "gougingparams"
+	cacheKeyGougingParams = "gougingparams"
+	cacheKeyUsableHosts   = "usablehosts"
 
 	cacheEntryExpiry = 5 * time.Minute
 )
@@ -83,12 +83,12 @@ func (c *memoryCache) Invalidate(key string) {
 
 type (
 	Bus interface {
-		Contracts(ctx context.Context, opts api.ContractsOpts) ([]api.ContractMetadata, error)
+		UsableHosts(ctx context.Context) ([]api.HostInfo, error)
 		GougingParams(ctx context.Context) (api.GougingParams, error)
 	}
 
 	WorkerCache interface {
-		DownloadContracts(ctx context.Context) ([]api.ContractMetadata, error)
+		UsableHosts(ctx context.Context) ([]api.HostInfo, error)
 		GougingParams(ctx context.Context) (api.GougingParams, error)
 		HandleEvent(event webhooks.Event) error
 		Subscribe(e EventSubscriber) error
@@ -115,28 +115,27 @@ func NewCache(b Bus, logger *zap.Logger) WorkerCache {
 	}
 }
 
-func (c *cache) DownloadContracts(ctx context.Context) (contracts []api.ContractMetadata, err error) {
+func (c *cache) UsableHosts(ctx context.Context) (hosts []api.HostInfo, err error) {
 	// fetch directly from bus if the cache is not ready
 	if !c.isReady() {
 		c.logger.Warn(errCacheNotReady)
-		contracts, err = c.b.Contracts(ctx, api.ContractsOpts{})
+		hosts, err = c.b.UsableHosts(ctx)
 		return
 	}
 
 	// fetch from bus if it's not cached or expired
-	value, found, expired := c.cache.Get(cacheKeyDownloadContracts)
+	value, found, expired := c.cache.Get(cacheKeyUsableHosts)
 	if !found || expired {
-		contracts, err = c.b.Contracts(ctx, api.ContractsOpts{})
+		hosts, err = c.b.UsableHosts(ctx)
 		if err == nil {
-			c.cache.Set(cacheKeyDownloadContracts, contracts)
+			c.cache.Set(cacheKeyUsableHosts, hosts)
 		}
-		if expired && !contractsEqual(value.([]api.ContractMetadata), contracts) {
-			c.logger.Warn(fmt.Errorf("%w: key %v", errCacheOutdated, cacheKeyDownloadContracts))
+		if expired && !hostsEqual(value.([]api.HostInfo), hosts) {
+			c.logger.Warn(fmt.Errorf("%w: key %v", errCacheOutdated, cacheKeyUsableHosts))
 		}
 		return
 	}
-
-	return value.([]api.ContractMetadata), nil
+	return value.([]api.HostInfo), nil
 }
 
 func (c *cache) GougingParams(ctx context.Context) (gp api.GougingParams, err error) {
@@ -178,21 +177,14 @@ func (c *cache) HandleEvent(event webhooks.Event) (err error) {
 	case api.EventConsensusUpdate:
 		log = log.With("bh", e.BlockHeight, "ts", e.Timestamp)
 		c.handleConsensusUpdate(e)
+	case api.EventSettingUpdate:
+		log = log.With("gouging", e.GougingSettings != nil, "pinned", e.PinnedSettings != nil, "upload", e.UploadSettings != nil, "ts", e.Timestamp)
+		c.handleSettingUpdate(e)
 	case api.EventContractAdd:
-		log = log.With("fcid", e.Added.ID, "ts", e.Timestamp)
-		c.handleContractAdd(e)
 	case api.EventContractArchive:
-		log = log.With("fcid", e.ContractID, "ts", e.Timestamp)
-		c.handleContractArchive(e)
 	case api.EventContractRenew:
-		log = log.With("fcid", e.Renewal.ID, "renewedFrom", e.Renewal.RenewedFrom, "ts", e.Timestamp)
-		c.handleContractRenew(e)
 	case api.EventHostUpdate:
-		log = log.With("hk", e.HostKey, "ts", e.Timestamp)
-		c.handleHostUpdate(e)
-	case api.EventSettingUpdate:
-		log = log.With("gouging", e.GougingSettings != nil, "pinned", e.PinnedSettings != nil, "upload", e.UploadSettings != nil, "ts", e.Timestamp)
-		c.handleSettingUpdate(e)
+		c.cache.Invalidate(cacheKeyUsableHosts)
 	default:
 		log.Info("unhandled event", e)
 		return
@@ -246,79 +238,6 @@ func (c *cache) handleConsensusUpdate(event api.EventConsensusUpdate) {
 	c.cache.Set(cacheKeyGougingParams, gp)
 }
 
-func (c *cache) handleContractAdd(event api.EventContractAdd) {
-	// return early if the cache doesn't have contracts
-	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)
-	if !found {
-		return
-	}
-	contracts := value.([]api.ContractMetadata)
-
-	// add the contract to the cache
-	for _, contract := range contracts {
-		if contract.ID == event.Added.ID {
-			return
-		}
-	}
-	contracts = append(contracts, event.Added)
-	c.cache.Set(cacheKeyDownloadContracts, contracts)
-}
-
-func (c *cache) handleContractArchive(event api.EventContractArchive) {
-	// return early if the cache doesn't have contracts
-	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)
-	if !found {
-		return
-	}
-	contracts := value.([]api.ContractMetadata)
-
-	// remove the contract from the cache
-	for i, contract := range contracts {
-		if contract.ID == event.ContractID {
-			contracts = append(contracts[:i], contracts[i+1:]...)
-			break
-		}
-	}
-	c.cache.Set(cacheKeyDownloadContracts, contracts)
-}
-
-func (c *cache) handleContractRenew(event api.EventContractRenew) {
-	// return early if the cache doesn't have contracts
-	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)
-	if !found {
-		return
-	}
-	contracts := value.([]api.ContractMetadata)
-
-	// update the renewed contract in the cache
-	for i, contract := range contracts {
-		if contract.ID == event.Renewal.RenewedFrom {
-			contracts[i] = event.Renewal
-			break
-		}
-	}
-
-	c.cache.Set(cacheKeyDownloadContracts, contracts)
-}
-
-func (c *cache) handleHostUpdate(e api.EventHostUpdate) {
-	// return early if the cache doesn't have contracts
-	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)
-	if !found {
-		return
-	}
-	contracts := value.([]api.ContractMetadata)
-
-	// update the host's IP in the cache
-	for i, contract := range contracts {
-		if contract.HostKey == e.HostKey {
-			contracts[i].HostIP = e.NetAddr
-		}
-	}
-
-	c.cache.Set(cacheKeyDownloadContracts, contracts)
-}
-
 func (c *cache) handleSettingUpdate(e api.EventSettingUpdate) {
 	// return early if the cache doesn't have gouging params to update
 	value, found, _ := c.cache.Get(cacheKeyGougingParams)
@@ -337,14 +256,14 @@ func (c *cache) handleSettingUpdate(e api.EventSettingUpdate) {
 	c.cache.Set(cacheKeyGougingParams, gp)
 }
 
-func contractsEqual(x, y []api.ContractMetadata) bool {
+func hostsEqual(x, y []api.HostInfo) bool {
 	if len(x) != len(y) {
 		return false
 	}
-	sort.Slice(x, func(i, j int) bool { return x[i].ID.String() < x[j].ID.String() })
-	sort.Slice(y, func(i, j int) bool { return y[i].ID.String() < y[j].ID.String() })
+	sort.Slice(x, func(i, j int) bool { return x[i].PublicKey.String() < x[j].PublicKey.String() })
+	sort.Slice(y, func(i, j int) bool { return y[i].PublicKey.String() < y[j].PublicKey.String() })
 	for i, c := range x {
-		if c.ID.String() != y[i].ID.String() {
+		if c.PublicKey.String() != y[i].PublicKey.String() || c.SiamuxAddr != y[i].SiamuxAddr {
 			return false
 		}
 	}