stores: remove old subscriber code

api/bus.go

@@ -1,10 +1,7 @@
 package api
 
 import (
-	"time"
-
 	"go.sia.tech/core/types"
-	"go.sia.tech/coreutils/wallet"
 )
 
 type (
@@ -21,56 +18,6 @@ type (
 	}
 )
 
-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"`
-	}
-)
-
-func ConvertToSiacoinElements(sces []wallet.SiacoinElement) []SiacoinElement {
-	elements := make([]SiacoinElement, len(sces))
-	for i, sce := range sces {
-		elements[i] = SiacoinElement{
-			ID: sce.StateElement.ID,
-			SiacoinOutput: types.SiacoinOutput{
-				Value:   sce.SiacoinOutput.Value,
-				Address: sce.SiacoinOutput.Address,
-			},
-			MaturityHeight: sce.MaturityHeight,
-		}
-	}
-	return elements
-}
-
-func ConvertToTransactions(events []wallet.Event) []Transaction {
-	transactions := make([]Transaction, len(events))
-	for i, e := range events {
-		transactions[i] = Transaction{
-			Raw:       e.Transaction,
-			Index:     e.Index,
-			ID:        types.TransactionID(e.ID),
-			Inflow:    e.Inflow,
-			Outflow:   e.Outflow,
-			Timestamp: e.Timestamp,
-		}
-	}
-	return transactions
-}
-
 type (
 	// UploadParams contains the metadata needed by a worker to upload an object.
 	UploadParams struct {

api/wallet.go

@@ -1,15 +1,73 @@
 package api
 
 import (
+	"errors"
 	"fmt"
 	"net/url"
 	"time"
 
 	rhpv2 "go.sia.tech/core/rhp/v2"
 	rhpv3 "go.sia.tech/core/rhp/v3"
 	"go.sia.tech/core/types"
+	"go.sia.tech/coreutils/wallet"
 )
 
+var (
+	// ErrInsufficientBalance is returned when there aren't enough unused outputs to
+	// cover the requested amount.
+	ErrInsufficientBalance = errors.New("insufficient balance")
+)
+
+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"`
+	}
+)
+
+func ConvertToSiacoinElements(sces []wallet.SiacoinElement) []SiacoinElement {
+	elements := make([]SiacoinElement, len(sces))
+	for i, sce := range sces {
+		elements[i] = SiacoinElement{
+			ID: sce.StateElement.ID,
+			SiacoinOutput: types.SiacoinOutput{
+				Value:   sce.SiacoinOutput.Value,
+				Address: sce.SiacoinOutput.Address,
+			},
+			MaturityHeight: sce.MaturityHeight,
+		}
+	}
+	return elements
+}
+
+func ConvertToTransactions(events []wallet.Event) []Transaction {
+	transactions := make([]Transaction, len(events))
+	for i, e := range events {
+		transactions[i] = Transaction{
+			Raw:       e.Transaction,
+			Index:     e.Index,
+			ID:        types.TransactionID(e.ID),
+			Inflow:    e.Inflow,
+			Outflow:   e.Outflow,
+			Timestamp: e.Timestamp,
+		}
+	}
+	return transactions
+}
+
 type (
 	// WalletFundRequest is the request type for the /wallet/fund endpoint.
 	WalletFundRequest struct {

autopilot/contractor.go

@@ -1348,7 +1348,7 @@ func (c *contractor) renewContract(ctx context.Context, w Worker, ci contractInf
 			"renterFunds", renterFunds,
 			"expectedNewStorage", expectedNewStorage,
 		)
-		if isErr(err, wallet.ErrNotEnoughFunds) {
+		if isErr(err, wallet.ErrNotEnoughFunds) || isErr(err, api.ErrInsufficientBalance) {
 			return api.ContractMetadata{}, false, err
 		}
 		return api.ContractMetadata{}, true, err
@@ -1431,7 +1431,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 isErr(err, wallet.ErrNotEnoughFunds) {
+		if isErr(err, wallet.ErrNotEnoughFunds) || isErr(err, api.ErrInsufficientBalance) {
 			return api.ContractMetadata{}, false, err
 		}
 		return api.ContractMetadata{}, true, err
@@ -1495,7 +1495,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 isErr(err, wallet.ErrNotEnoughFunds) {
+		if isErr(err, wallet.ErrNotEnoughFunds) || isErr(err, api.ErrInsufficientBalance) {
 			return api.ContractMetadata{}, false, err
 		}
 		return api.ContractMetadata{}, true, err

bus/bus.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"go.sia.tech/core/consensus"
+	"go.sia.tech/core/gateway"
 	rhpv2 "go.sia.tech/core/rhp/v2"
 	rhpv3 "go.sia.tech/core/rhp/v3"
 	"go.sia.tech/core/types"
@@ -404,10 +405,18 @@ func (b *bus) consensusAcceptBlock(jc jape.Context) {
 
 	// TODO: should we extend the API with a way to accept multiple blocks at once?
 	// TODO: should we deprecate this route in favor of /addblocks
-
 	if jc.Check("failed to accept block", b.cm.AddBlocks([]types.Block{block})) != nil {
 		return
 	}
+
+	if block.V2 == nil {
+		b.s.BroadcastHeader(gateway.BlockHeader{
+			ParentID:   block.ParentID,
+			Nonce:      block.Nonce,
+			Timestamp:  block.Timestamp,
+			MerkleRoot: block.MerkleRoot(),
+		})
+	}
 }
 
 func (b *bus) syncerAddrHandler(jc jape.Context) {
@@ -416,7 +425,11 @@ func (b *bus) syncerAddrHandler(jc jape.Context) {
 }
 
 func (b *bus) syncerPeersHandler(jc jape.Context) {
-	jc.Encode(b.s.Peers())
+	var peers []string
+	for _, p := range b.s.Peers() {
+		peers = append(peers, p.String())
+	}
+	jc.Encode(peers)
 }
 
 func (b *bus) syncerConnectHandler(jc jape.Context) {
@@ -448,11 +461,17 @@ func (b *bus) txpoolTransactionsHandler(jc jape.Context) {
 
 func (b *bus) txpoolBroadcastHandler(jc jape.Context) {
 	var txnSet []types.Transaction
-	if jc.Decode(&txnSet) == nil {
-		// TODO: should we handle 'known' return value
-		_, err := b.cm.AddPoolTransactions(txnSet)
-		jc.Check("couldn't broadcast transaction set", err)
+	if jc.Decode(&txnSet) != nil {
+		return
 	}
+
+	// TODO: should we handle 'known' return value
+	_, err := b.cm.AddPoolTransactions(txnSet)
+	if jc.Check("couldn't broadcast transaction set", err) != nil {
+		return
+	}
+
+	b.s.BroadcastTransactionSet(txnSet)
 }
 
 func (b *bus) bucketsHandlerGET(jc jape.Context) {
@@ -597,6 +616,7 @@ func (b *bus) walletFundHandler(jc jape.Context) {
 		return
 	}
 	txn := wfr.Transaction
+
 	if len(txn.MinerFees) == 0 {
 		// if no fees are specified, we add some
 		fee := b.cm.RecommendedFee().Mul64(b.cm.TipState().TransactionWeight(txn))
@@ -689,6 +709,7 @@ func (b *bus) walletPrepareFormHandler(jc jape.Context) {
 	if jc.Check("couldn't fund transaction", err) != nil {
 		return
 	}
+
 	b.w.SignTransaction(&txn, toSign, ExplicitCoveredFields(txn))
 
 	// TODO: UnconfirmedParents needs a ctx (be sure to release inputs on err)

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/montanaflynn/stats v0.7.1
 	gitlab.com/NebulousLabs/encoding v0.0.0-20200604091946-456c3dc907fe
 	go.sia.tech/core v0.2.1
-	go.sia.tech/coreutils v0.0.2
+	go.sia.tech/coreutils v0.0.3
 	go.sia.tech/gofakes3 v0.0.0-20231109151325-e0d47c10dce2
 	go.sia.tech/hostd v1.0.2-beta.2.0.20240131203318-9d84aad6ef13
 	go.sia.tech/jape v0.11.2-0.20240124024603-93559895d640

go.sum

@@ -13,6 +13,9 @@ github.com/aws/aws-sdk-go v1.50.1 h1:AwnLUM7TcH9vMZqA4TcDKmGfLmDW5VXwT5tPH6kXylo
 github.com/aws/aws-sdk-go v1.50.1/go.mod h1:LF8svs817+Nz+DmiMQKTO3ubZ/6IaTpq3TjupRn3Eqk=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
+github.com/br0xen/boltbrowser v0.0.0-20230531143731-fcc13603daaf h1:NyqdH+vWNYPwQIK9jNv7sdIVbRGclwIdFhQk3+qlNEs=
+github.com/br0xen/boltbrowser v0.0.0-20230531143731-fcc13603daaf/go.mod h1:uhjRwoqgy4g6fCwo7OJHjCxDOmx/YSCz2rnAYb63ZhY=
+github.com/br0xen/termbox-util v0.0.0-20170904143325-de1d4c83380e/go.mod h1:x9wJlgOj74OFTOBwXOuO8pBguW37EgYNx51Dbjkfzo4=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/cloudflare-go v0.86.0 h1:jEKN5VHNYNYtfDL2lUFLTRo+nOVNPFxpXTstVx0rqHI=
@@ -130,6 +133,7 @@ github.com/mattn/go-colorable v0.1.12 h1:jF+Du6AlPIjs2BiUiQlKOX0rt3SujHxPnksPKZb
 github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4=
 github.com/mattn/go-isatty v0.0.17 h1:BTarxUcIeDqL27Mc+vyvdWYSL28zpIhv3RoTdsLMPng=
 github.com/mattn/go-isatty v0.0.17/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/mattn/go-sqlite3 v1.14.18 h1:JL0eqdCOq6DJVNPSvArO/bIV9/P7fbGrV00LZHc+5aI=
 github.com/mattn/go-sqlite3 v1.14.18/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
@@ -149,6 +153,7 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
 github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/nsf/termbox-go v1.1.1/go.mod h1:T0cTdVuOwf7pHQNtfhnEbzHbcNyCEcVU4YPpouCbVxo=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -191,9 +196,14 @@ github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnIn
 github.com/spf13/viper v1.4.0/go.mod h1:PTJ7Z/lr49W6bUbkmS1V3by4uWynFiR9p7+dSq/yZzE=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
@@ -239,12 +249,16 @@ gitlab.com/NebulousLabs/threadgroup v0.0.0-20200608151952-38921fbef213/go.mod h1
 gitlab.com/NebulousLabs/writeaheadlog v0.0.0-20200618142844-c59a90f49130/go.mod h1:SxigdS5Q1ui+OMgGAXt1E/Fg3RB6PvKXMov2O3gvIzs=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
+go.etcd.io/bbolt v1.3.7/go.mod h1:N9Mkw9X8x5fupy0IKsmuqVtoGDyxsaDlbk4Rd05IAQw=
 go.etcd.io/bbolt v1.3.8 h1:xs88BrvEv273UsB79e0hcVrlUWmS0a8upikMFhSyAtA=
 go.etcd.io/bbolt v1.3.8/go.mod h1:N9Mkw9X8x5fupy0IKsmuqVtoGDyxsaDlbk4Rd05IAQw=
+go.etcd.io/gofail v0.1.0/go.mod h1:VZBCXYGZhHAinaBiiqYvuDynvahNsAyLFwB3kEHKz1M=
 go.sia.tech/core v0.2.1 h1:CqmMd+T5rAhC+Py3NxfvGtvsj/GgwIqQHHVrdts/LqY=
 go.sia.tech/core v0.2.1/go.mod h1:3EoY+rR78w1/uGoXXVqcYdwSjSJKuEMI5bL7WROA27Q=
 go.sia.tech/coreutils v0.0.2 h1:vDqMDM6dW6b/R3sO1ycr8fAnJXUiAvrzxehEIq/AsKA=
 go.sia.tech/coreutils v0.0.2/go.mod h1:UBFc77wXiE//eyilO5HLOncIEj7F69j0Nv2OkFujtP0=
+go.sia.tech/coreutils v0.0.3 h1:ZxuzovRpQMvfy/pCOV4om1cPF6sE15GyJyK36kIrF1Y=
+go.sia.tech/coreutils v0.0.3/go.mod h1:UBFc77wXiE//eyilO5HLOncIEj7F69j0Nv2OkFujtP0=
 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 v1.0.2-beta.2.0.20240131203318-9d84aad6ef13 h1:JcyVUtJfzeMh+zJAW20BMVhBYekg+h0T8dMeF7GzAFs=
@@ -332,8 +346,10 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

internal/node/node.go

@@ -33,6 +33,7 @@ import (
 // RHP4 TODOs:
 // - get rid of dbConsensusInfo
 // - get rid of returned chain manager in bus constructor
+// - pass last tip to AddSubscriber
 // - all wallet metrics support
 // - add UPNP support
 
@@ -146,7 +147,7 @@ func NewBus(cfg BusConfig, dir string, seed types.PrivateKey, logger *zap.Logger
 		UniqueID:   gateway.GenerateUniqueID(),
 		NetAddress: syncerAddr,
 	}
-	s := syncer.New(l, cm, sqlStore, header)
+	s := syncer.New(l, cm, sqlStore, header, syncer.WithSyncInterval(100*time.Millisecond), syncer.WithLogger(logger.Named("syncer")))
 
 	b, err := bus.New(alertsMgr, wh, cm, s, w, sqlStore, sqlStore, sqlStore, sqlStore, sqlStore, sqlStore, logger)
 	if err != nil {