alert for missing sectors (#486)

Author: LexLuthr

alertmanager/alerts.go

@@ -11,10 +11,12 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/dustin/go-humanize"
+	cbor "github.com/ipfs/go-ipld-cbor"
 	"github.com/samber/lo"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-bitfield"
 	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
@@ -24,6 +26,8 @@ import (
 	"github.com/filecoin-project/curio/deps/config"
 	"github.com/filecoin-project/curio/harmony/harmonydb"
 
+	"github.com/filecoin-project/lotus/blockstore"
+	"github.com/filecoin-project/lotus/chain/actors/adt"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/types"
 	cliutil "github.com/filecoin-project/lotus/cli/util"
@@ -717,3 +721,81 @@ func chainSyncCheck(al *alerts) {
 		}
 	}
 }
+
+func missingSectorCheck(al *alerts) {
+	Name := "MissingSectors"
+	al.alertMap[Name] = &alertOut{}
+
+	var sectors []struct {
+		MinerID  int64 `db:"miner_id"`
+		SectorID int64 `db:"sector_num"`
+	}
+
+	err := al.db.Select(al.ctx, &sectors, `SELECT miner_id, sector_num  FROM sector_location WHERE sector_filetype = 2 GROUP BY miner_id, sector_num ORDER BY miner_id, sector_num`)
+	if err != nil {
+		al.alertMap[Name].err = xerrors.Errorf("getting sealed sectors from database: %w", err)
+		return
+	}
+
+	dbMap := map[address.Address]*bitfield.BitField{}
+	minerMap := map[int64]address.Address{}
+
+	for _, sector := range sectors {
+		m, ok := minerMap[sector.MinerID]
+		if !ok {
+			m, err = address.NewIDAddress(uint64(sector.MinerID))
+			if err != nil {
+				al.alertMap[Name].err = xerrors.Errorf("parsing miner address: %w", err)
+				return
+			}
+			minerMap[sector.MinerID] = m
+		}
+		bf, ok := dbMap[m]
+		if !ok {
+			newbf := bitfield.New()
+			newbf.Set(uint64(sector.SectorID))
+			dbMap[m] = &newbf
+			continue
+		}
+		bf.Set(uint64(sector.SectorID))
+	}
+
+	head, err := al.api.ChainHead(al.ctx)
+	if err != nil {
+		al.alertMap[Name].err = xerrors.Errorf("ChainHead: %w", err)
+		return
+	}
+
+	for _, m := range minerMap {
+		mact, err := al.api.StateGetActor(al.ctx, m, head.Key())
+		if err != nil {
+			al.alertMap[Name].err = xerrors.Errorf("getting miner actor %s: %w", m.String(), err)
+			return
+		}
+		tbs := blockstore.NewTieredBstore(blockstore.NewAPIBlockstore(al.api), blockstore.NewMemory())
+		mas, err := miner.Load(adt.WrapStore(al.ctx, cbor.NewCborStore(tbs)), mact)
+		if err != nil {
+			al.alertMap[Name].err = xerrors.Errorf("loading miner %s: %w", m.String(), err)
+			return
+		}
+
+		liveSectors, err := miner.AllPartSectors(mas, miner.Partition.LiveSectors)
+		if err != nil {
+			al.alertMap[Name].err = xerrors.Errorf("getting live sectors for miner %s: %w", m.String(), err)
+			return
+		}
+
+		diff, err := bitfield.SubtractBitField(liveSectors, *dbMap[m])
+		if err != nil {
+			al.alertMap[Name].err = xerrors.Errorf("subtracting live sectors for miner %s: %w", m.String(), err)
+		}
+		err = diff.ForEach(func(i uint64) error {
+			al.alertMap[Name].alertString += fmt.Sprintf("Missing sector %d in storage for miner %s. ", i, m.String())
+			return nil
+		})
+		if err != nil {
+			al.alertMap[Name].err = xerrors.Errorf("getting missing sectors for miner %s: %w", m.String(), err)
+			return
+		}
+	}
+}

alertmanager/task_alert.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"time"
 
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 
 	"github.com/filecoin-project/go-address"
@@ -32,11 +34,15 @@ var log = logging.Logger("curio/alertmanager")
 
 type AlertAPI interface {
 	ctladdr.NodeApi
+	ChainReadObj(context.Context, cid.Cid) ([]byte, error)
+	ChainHasObj(context.Context, cid.Cid) (bool, error)
+	ChainPutObj(context.Context, blocks.Block) error
 	ChainHead(context.Context) (*types.TipSet, error)
 	ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
 	StateMinerInfo(ctx context.Context, actor address.Address, tsk types.TipSetKey) (api.MinerInfo, error)
 	StateMinerProvingDeadline(context.Context, address.Address, types.TipSetKey) (*dline.Info, error)
 	StateMinerPartitions(context.Context, address.Address, uint64, types.TipSetKey) ([]api.Partition, error)
+	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
 }
 
 type AlertTask struct {
@@ -70,6 +76,7 @@ var AlertFuncs = []AlertFunc{
 	wnPostCheck,
 	NowCheck,
 	chainSyncCheck,
+	missingSectorCheck,
 }
 
 func NewAlertTask(