Migrate renterd subscriber to coreutils types (#1098)

This PR updates `core` and `coreutils` and splits the subscriber logic
off into its own package called `chain`. This PR is the last step in
migrating to the `coreutils` types and the new subscriber pattern.

.github/workflows/test.yml

@@ -49,7 +49,7 @@ jobs:
           host port: 3800
           mysql version: '8'
           mysql root password: test
-      - name: Test
+      - name: Test Stores
         uses: n8maninger/action-golang-test@v1
         with:
           args: "-race;-short"

autopilot/contractor/contractor.go

@@ -89,6 +89,7 @@ type Bus interface {
 	AncestorContracts(ctx context.Context, id types.FileContractID, minStartHeight uint64) ([]api.ArchivedContract, error)
 	ArchiveContracts(ctx context.Context, toArchive map[types.FileContractID]string) error
 	ConsensusState(ctx context.Context) (api.ConsensusState, error)
+	Contract(ctx context.Context, id types.FileContractID) (api.ContractMetadata, error)
 	Contracts(ctx context.Context, opts api.ContractsOpts) (contracts []api.ContractMetadata, err error)
 	FileContractTax(ctx context.Context, payout types.Currency) (types.Currency, error)
 	Host(ctx context.Context, hostKey types.PublicKey) (api.Host, error)
@@ -615,6 +616,7 @@ func (c *Contractor) runContractChecks(ctx *mCtx, hostChecks map[types.PublicKey
 			"toArchive", len(toArchive),
 			"toRefresh", len(toRefresh),
 			"toRenew", len(toRenew),
+			"bh", bh,
 		)
 	}()
 
@@ -650,7 +652,7 @@ LOOP:
 			toArchive[fcid] = errContractMaxRevisionNumber.Error()
 		} else if contract.RevisionNumber == math.MaxUint64 {
 			toArchive[fcid] = errContractMaxRevisionNumber.Error()
-		} else if contract.State == api.ContractStatePending && bh-contract.StartHeight > contractConfirmationDeadline {
+		} else if contract.State == api.ContractStatePending && bh-contract.StartHeight > ContractConfirmationDeadline {
 			toArchive[fcid] = errContractNotConfirmed.Error()
 		}
 		if _, archived := toArchive[fcid]; archived {

autopilot/contractor/hostfilter.go

@@ -14,6 +14,10 @@ import (
 )
 
 const (
+	// ContractConfirmationDeadline is the number of blocks since its start
+	// height we wait for a contract to appear on chain.
+	ContractConfirmationDeadline = 18
+
 	// minContractFundUploadThreshold is the percentage of contract funds
 	// remaining at which the contract gets marked as not good for upload
 	minContractFundUploadThreshold = float64(0.05) // 5%
@@ -23,10 +27,6 @@ const (
 	// acquirable storage below which the contract is considered to be
 	// out-of-collateral.
 	minContractCollateralDenominator = 20 // 5%
-
-	// contractConfirmationDeadline is the number of blocks since its start
-	// height we wait for a contract to appear on chain.
-	contractConfirmationDeadline = 18
 )
 
 var (

bus/bus.go

@@ -26,12 +26,17 @@ import (
 	"go.sia.tech/renterd/api"
 	"go.sia.tech/renterd/build"
 	"go.sia.tech/renterd/bus/client"
+	"go.sia.tech/renterd/chain"
 	"go.sia.tech/renterd/object"
 	"go.sia.tech/renterd/webhooks"
 	"go.sia.tech/siad/modules"
 	"go.uber.org/zap"
 )
 
+// maxSyncTime is the maximum time since the last block before we consider
+// ourselves unsynced.
+const maxSyncTime = time.Hour
+
 // Client re-exports the client from the client package.
 type Client struct {
 	*client.Client
@@ -205,7 +210,7 @@ type (
 		Redistribute(outputs int, amount, feePerByte types.Currency) (txns []types.Transaction, toSign []types.Hash256, err error)
 		ReleaseInputs(txns ...types.Transaction)
 		SignTransaction(txn *types.Transaction, toSign []types.Hash256, cf types.CoveredFields)
-		SpendableOutputs() ([]wallet.SiacoinElement, error)
+		SpendableOutputs() ([]types.SiacoinElement, error)
 		Tip() (types.ChainIndex, error)
 		UnconfirmedTransactions() ([]wallet.Event, error)
 		Events(offset, limit int) ([]wallet.Event, error)
@@ -232,6 +237,7 @@ type bus struct {
 	w  Wallet
 
 	as    AutopilotStore
+	cs    chain.ChainStore
 	eas   EphemeralAccountStore
 	hdb   HostDB
 	ms    MetadataStore
@@ -450,7 +456,11 @@ func (b *bus) syncerConnectHandler(jc jape.Context) {
 }
 
 func (b *bus) consensusStateHandler(jc jape.Context) {
-	jc.Encode(b.consensusState())
+	cs, err := b.consensusState(jc.Request.Context())
+	if jc.Check("couldn't fetch consensus state", err) != nil {
+		return
+	}
+	jc.Encode(cs)
 }
 
 func (b *bus) consensusNetworkHandler(jc jape.Context) {
@@ -1771,19 +1781,23 @@ func (b *bus) paramsHandlerUploadGET(jc jape.Context) {
 	})
 }
 
-func (b *bus) consensusState() api.ConsensusState {
-	cs := b.cm.TipState()
+func (b *bus) consensusState(ctx context.Context) (api.ConsensusState, error) {
+	index, err := b.cs.ChainIndex(ctx)
+	if err != nil {
+		return api.ConsensusState{}, err
+	}
 
 	var synced bool
-	if block, ok := b.cm.Block(cs.Index.ID); ok && time.Since(block.Timestamp) < 2*cs.BlockInterval() {
+	block, found := b.cm.Block(index.ID)
+	if found && time.Since(block.Timestamp) <= maxSyncTime {
 		synced = true
 	}
 
 	return api.ConsensusState{
-		BlockHeight:   cs.Index.Height,
-		LastBlockTime: api.TimeRFC3339(cs.PrevTimestamps[0]),
+		BlockHeight:   index.Height,
+		LastBlockTime: api.TimeRFC3339(block.Timestamp),
 		Synced:        synced,
-	}
+	}, nil
 }
 
 func (b *bus) paramsHandlerGougingGET(jc jape.Context) {
@@ -1809,7 +1823,10 @@ func (b *bus) gougingParams(ctx context.Context) (api.GougingParams, error) {
 		b.logger.Panicf("failed to unmarshal redundancy settings '%s': %v", rss, err)
 	}
 
-	cs := b.consensusState()
+	cs, err := b.consensusState(ctx)
+	if err != nil {
+		return api.GougingParams{}, err
+	}
 
 	return api.GougingParams{
 		ConsensusState:     cs,
@@ -2422,12 +2439,13 @@ func (b *bus) multipartHandlerListPartsPOST(jc jape.Context) {
 }
 
 // New returns a new Bus.
-func New(am *alerts.Manager, hm WebhookManager, cm ChainManager, s Syncer, w Wallet, hdb HostDB, as AutopilotStore, ms MetadataStore, ss SettingStore, eas EphemeralAccountStore, mtrcs MetricsStore, l *zap.Logger) (*bus, error) {
+func New(am *alerts.Manager, hm WebhookManager, cm ChainManager, cs chain.ChainStore, s Syncer, w Wallet, hdb HostDB, as AutopilotStore, ms MetadataStore, ss SettingStore, eas EphemeralAccountStore, mtrcs MetricsStore, l *zap.Logger) (*bus, error) {
 	b := &bus{
 		alerts:           alerts.WithOrigin(am, "bus"),
 		alertMgr:         am,
 		webhooks:         hm,
 		cm:               cm,
+		cs:               cs,
 		s:                s,
 		w:                w,
 		hdb:              hdb,

bus/client/client_test.go

@@ -70,7 +70,7 @@ func newTestClient(dir string) (*client.Client, func() error, func(context.Conte
 
 	// create bus
 	network, genesis := build.Network()
-	b, shutdown, _, err := node.NewBus(node.BusConfig{
+	b, shutdown, _, _, err := node.NewBus(node.BusConfig{
 		Bus: config.Bus{
 			AnnouncementMaxAgeHours:       24 * 7 * 52, // 1 year
 			Bootstrap:                     false,

chain/chain.go

@@ -0,0 +1,26 @@
+package chain
+
+import (
+	"go.sia.tech/core/consensus"
+	"go.sia.tech/core/types"
+	"go.sia.tech/coreutils/chain"
+)
+
+type (
+	Manager          = chain.Manager
+	HostAnnouncement = chain.HostAnnouncement
+	ApplyUpdate      = chain.ApplyUpdate
+	RevertUpdate     = chain.RevertUpdate
+)
+
+func TestnetZen() (*consensus.Network, types.Block) {
+	return chain.TestnetZen()
+}
+
+func NewDBStore(db chain.DB, n *consensus.Network, genesisBlock types.Block) (_ *chain.DBStore, _ consensus.State, err error) {
+	return chain.NewDBStore(db, n, genesisBlock)
+}
+
+func NewManager(store chain.Store, cs consensus.State) *Manager {
+	return chain.NewManager(store, cs)
+}