Merge remote-tracking branch 'origin/next' into fix/sealing-betterering

node/builder.go

@@ -150,6 +150,7 @@ func defaults() []Option {
 		Override(new(helpers.MetricsCtx), context.Background),
 		Override(new(record.Validator), modules.RecordValidator),
 		Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(false)),
+		Override(new(dtypes.ShutdownChan), make(chan struct{})),
 
 		// Filecoin modules
 
@@ -180,6 +181,7 @@ func libp2p() Option {
 		Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
 		Override(AutoNATSvcKey, lp2p.AutoNATService),
 
+		Override(new(*dtypes.ScoreKeeper), lp2p.ScoreKeeper),
 		Override(new(*pubsub.PubSub), lp2p.GossipSub),
 		Override(new(*config.Pubsub), func(bs dtypes.Bootstrapper) *config.Pubsub {
 			return &config.Pubsub{

node/impl/common/common.go

@@ -2,6 +2,8 @@ package common
 
 import (
 	"context"
+	"sort"
+	"strings"
 
 	logging "github.com/ipfs/go-log/v2"
 
@@ -25,9 +27,11 @@ import (
 type CommonAPI struct {
 	fx.In
 
-	APISecret *dtypes.APIAlg
-	Host      host.Host
-	Router    lp2p.BaseIpfsRouting
+	APISecret    *dtypes.APIAlg
+	Host         host.Host
+	Router       lp2p.BaseIpfsRouting
+	Sk           *dtypes.ScoreKeeper
+	ShutdownChan dtypes.ShutdownChan
 }
 
 type jwtPayload struct {
@@ -54,6 +58,21 @@ func (a *CommonAPI) AuthNew(ctx context.Context, perms []auth.Permission) ([]byt
 func (a *CommonAPI) NetConnectedness(ctx context.Context, pid peer.ID) (network.Connectedness, error) {
 	return a.Host.Network().Connectedness(pid), nil
 }
+func (a *CommonAPI) NetPubsubScores(context.Context) ([]api.PubsubScore, error) {
+	scores := a.Sk.Get()
+	out := make([]api.PubsubScore, len(scores))
+	i := 0
+	for k, v := range scores {
+		out[i] = api.PubsubScore{ID: k, Score: v}
+		i++
+	}
+
+	sort.Slice(out, func(i, j int) bool {
+		return strings.Compare(string(out[i].ID), string(out[j].ID)) > 0
+	})
+
+	return out, nil
+}
 
 func (a *CommonAPI) NetPeers(context.Context) ([]peer.AddrInfo, error) {
 	conns := a.Host.Network().Conns()
@@ -115,4 +134,9 @@ func (a *CommonAPI) LogSetLevel(ctx context.Context, subsystem, level string) er
 	return logging.SetLogLevel(subsystem, level)
 }
 
+func (a *CommonAPI) Shutdown(ctx context.Context) error {
+	a.ShutdownChan <- struct{}{}
+	return nil
+}
+
 var _ api.Common = &CommonAPI{}

node/impl/full/state.go

@@ -344,10 +344,8 @@ func (a *StateAPI) MinerCreateBlock(ctx context.Context, bt *api.BlockTemplate)
 	return &out, nil
 }
 
-func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error) {
-	// TODO: consider using event system for this, expose confidence
-
-	ts, recpt, err := a.StateManager.WaitForMessage(ctx, msg)
+func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error) {
+	ts, recpt, err := a.StateManager.WaitForMessage(ctx, msg, confidence)
 	if err != nil {
 		return nil, err
 	}

node/modules/dtypes/scorekeeper.go

@@ -0,0 +1,24 @@
+package dtypes
+
+import (
+	"sync"
+
+	peer "github.com/libp2p/go-libp2p-core/peer"
+)
+
+type ScoreKeeper struct {
+	lk     sync.Mutex
+	scores map[peer.ID]float64
+}
+
+func (sk *ScoreKeeper) Update(scores map[peer.ID]float64) {
+	sk.lk.Lock()
+	sk.scores = scores
+	sk.lk.Unlock()
+}
+
+func (sk *ScoreKeeper) Get() map[peer.ID]float64 {
+	sk.lk.Lock()
+	defer sk.lk.Unlock()
+	return sk.scores
+}

node/modules/dtypes/shutdown.go

@@ -0,0 +1,5 @@
+package dtypes
+
+// ShutdownChan is a channel to which you send a value if you intend to shut
+// down the daemon (or storage miner), including the node and RPC server.
+type ShutdownChan chan struct{}

node/modules/lp2p/pubsub.go

@@ -22,13 +22,18 @@ func init() {
 	// configure larger overlay parameters
 	pubsub.GossipSubD = 8
 	pubsub.GossipSubDscore = 6
+	pubsub.GossipSubDout = 3
 	pubsub.GossipSubDlo = 6
 	pubsub.GossipSubDhi = 12
 	pubsub.GossipSubDlazy = 12
 	pubsub.GossipSubDirectConnectInitialDelay = 30 * time.Second
 }
 
-func GossipSub(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, nn dtypes.NetworkName, bp dtypes.BootstrapPeers, cfg *config.Pubsub) (service *pubsub.PubSub, err error) {
+func ScoreKeeper() *dtypes.ScoreKeeper {
+	return new(dtypes.ScoreKeeper)
+}
+
+func GossipSub(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, nn dtypes.NetworkName, bp dtypes.BootstrapPeers, cfg *config.Pubsub, sk *dtypes.ScoreKeeper) (service *pubsub.PubSub, err error) {
 	bootstrappers := make(map[peer.ID]struct{})
 	for _, pi := range bp {
 		bootstrappers[pi.ID] = struct{}{}
@@ -161,6 +166,7 @@ func GossipSub(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, nn dtyp
 				OpportunisticGraftThreshold: 5,
 			},
 		),
+		pubsub.WithPeerScoreInspect(sk.Update, 10*time.Second),
 	}
 
 	// enable Peer eXchange on bootstrappers