Add miner_sector_posts tracking of window posts

model/actors/miner/sectorposts.go

@@ -0,0 +1,67 @@
+package miner
+
+import (
+	"context"
+
+	"github.com/go-pg/pg/v10"
+	"github.com/ipfs/go-cid"
+	"go.opentelemetry.io/otel/api/global"
+	"go.opentelemetry.io/otel/api/trace"
+	"go.opentelemetry.io/otel/label"
+	"golang.org/x/xerrors"
+)
+
+type MinerSectorPost struct {
+	MinerID  string `pg:",pk,notnull"`
+	SectorID uint64 `pg:",pk,notnull,use_zero"`
+	Epoch    int64  `pg:",pk,notnull"`
+
+	PostMessageCID string
+}
+
+type MinerSectorPosts []*MinerSectorPost
+
+func (msp *MinerSectorPost) PersistWithTx(ctx context.Context, tx *pg.Tx) error {
+	if _, err := tx.ModelContext(ctx, msp).
+		OnConflict("do nothing").
+		Insert(); err != nil {
+		return xerrors.Errorf("persisting miner sector window post: %w", err)
+	}
+	return nil
+}
+
+func NewMinerSectorPost(task *MinerTaskResult) MinerSectorPosts {
+	out := make([]*MinerSectorPost, len(task.Posts))
+	for s, c := range task.Posts {
+		mid := ""
+		if c != cid.Undef {
+			mid = c.String()
+		}
+		post := &MinerSectorPost{
+			MinerID:        task.Addr.String(),
+			SectorID:       s,
+			Epoch:          task.Height,
+			PostMessageCID: mid,
+		}
+		out = append(out, post)
+	}
+
+	return out
+}
+
+func (msps MinerSectorPosts) Persist(ctx context.Context, db *pg.DB) error {
+	return db.RunInTransaction(ctx, func(tx *pg.Tx) error {
+		return msps.PersistWithTx(ctx, tx)
+	})
+}
+
+func (msps MinerSectorPosts) PersistWithTx(ctx context.Context, tx *pg.Tx) error {
+	ctx, span := global.Tracer("").Start(ctx, "MinerSectorPosts.PersistWithTx", trace.WithAttributes(label.Int("count", len(msps))))
+	defer span.End()
+	for _, msp := range msps {
+		if err := msp.PersistWithTx(ctx, tx); err != nil {
+			return err
+		}
+	}
+	return nil
+}

model/actors/miner/task.go

@@ -28,6 +28,7 @@ type MinerTaskResult struct {
 	Ts        types.TipSetKey
 	Pts       types.TipSetKey
 	StateRoot cid.Cid
+	Height    int64
 
 	Addr  address.Address
 	Actor *types.Actor
@@ -38,6 +39,7 @@ type MinerTaskResult struct {
 	PreCommitChanges *miner.PreCommitChanges
 	SectorChanges    *miner.SectorChanges
 	PartitionDiff    map[uint64]*PartitionStatus
+	Posts            map[uint64]cid.Cid
 }
 
 func (mtr *MinerTaskResult) Persist(ctx context.Context, db *pg.DB) error {

storage/migrations/8_windowpost.go

@@ -0,0 +1,24 @@
+package migrations
+
+import (
+	"github.com/go-pg/migrations/v8"
+)
+
+// Schema version 8 provides additional tracking of the miner->deadline->partition->sector relation
+
+func init() {
+
+	up := batch(`
+	CREATE TABLE IF NOT EXISTS "miner_sector_posts" (
+		"miner_id" text not null,
+		"sector_id" bigserial not null,
+		"epoch" bigint not null,
+		"post_message_cid" text,
+		PRIMARY KEY ("miner_id", "sector_id", "epoch")
+	);
+`)
+	down := batch(`
+DROP TABLE IF EXISTS "miner_sector_posts";
+`)
+	migrations.MustRegisterTx(up, down)
+}

storage/sql.go

@@ -37,6 +37,7 @@ var models = []interface{}{
 	(*miner.MinerState)(nil),
 	(*miner.MinerDealSector)(nil),
 	(*miner.MinerSectorInfo)(nil),
+	(*miner.MinerSectorPost)(nil),
 	(*miner.MinerPreCommitInfo)(nil),
 
 	(*market.MarketDealProposal)(nil),

tasks/actorstate/miner.go

@@ -1,13 +1,18 @@
 package actorstate
 
 import (
+	"bytes"
 	"context"
 
 	"go.opentelemetry.io/otel/api/global"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-bitfield"
+	"github.com/filecoin-project/go-state-types/abi"
 	miner "github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
+	"github.com/ipfs/go-cid"
 
 	"github.com/filecoin-project/sentinel-visor/lens"
 	"github.com/filecoin-project/sentinel-visor/metrics"
@@ -40,6 +45,11 @@ func (m StorageMinerExtractor) Extract(ctx context.Context, a ActorInfo, node le
 		return nil, xerrors.Errorf("loading current miner actor: %w", err)
 	}
 
+	curTipset, err := node.ChainGetTipSet(ctx, a.TipSet)
+	if err != nil {
+		return nil, xerrors.Errorf("loading current tipset: %w", err)
+	}
+
 	curState, err := miner.Load(node.Store(), curActor)
 	if err != nil {
 		return nil, xerrors.Errorf("loading current miner state: %w", err)
@@ -84,12 +94,19 @@ func (m StorageMinerExtractor) Extract(ctx context.Context, a ActorInfo, node le
 		return nil, xerrors.Errorf("diffing miner partitions: %v", err)
 	}
 
+	// miner posts
+	posts, err := m.minerPosts(ctx, &a, curTipset.Height(), curState, node)
+	if err != nil {
+		return nil, xerrors.Errorf("constructing miner posts: %v", err)
+	}
+
 	// TODO we still need to do a little bit more processing here around sectors to get all the info we need, but this is okay for spike.
 
 	return &minermodel.MinerTaskResult{
 		Ts:               a.TipSet,
 		Pts:              a.ParentTipSet,
 		Addr:             a.Address,
+		Height:           int64(curTipset.Height()),
 		StateRoot:        a.ParentStateRoot,
 		Actor:            curActor,
 		State:            curState,
@@ -98,9 +115,98 @@ func (m StorageMinerExtractor) Extract(ctx context.Context, a ActorInfo, node le
 		PreCommitChanges: preCommitChanges,
 		SectorChanges:    sectorChanges,
 		PartitionDiff:    partitionsDiff,
+		Posts:            posts,
 	}, nil
 }
 
 func (m StorageMinerExtractor) minerPartitionsDiff(ctx context.Context, prevState, curState miner.State) (map[uint64]*minermodel.PartitionStatus, error) {
 	return nil, nil
 }
+
+func (m StorageMinerExtractor) minerPosts(ctx context.Context, actor *ActorInfo, epoch abi.ChainEpoch, curState miner.State, node lens.API) (map[uint64]cid.Cid, error) {
+	posts := make(map[uint64]cid.Cid)
+	block := actor.TipSet.Cids()[0]
+	msgs, err := node.ChainGetBlockMessages(ctx, block)
+	if err != nil {
+		return nil, xerrors.Errorf("diffing miner posts: %v", err)
+	}
+
+	var partitions map[uint64]miner.Partition
+	loadPartitions := func(state miner.State, epoch abi.ChainEpoch) (map[uint64]miner.Partition, error) {
+		info, err := state.DeadlineInfo(epoch)
+		if err != nil {
+			return nil, err
+		}
+		dline, err := state.LoadDeadline(info.Index)
+		if err != nil {
+			return nil, err
+		}
+		pmap := make(map[uint64]miner.Partition)
+		if err := dline.ForEachPartition(func(idx uint64, p miner.Partition) error {
+			pmap[idx] = p
+			return nil
+		}); err != nil {
+			return nil, err
+		}
+		return pmap, nil
+	}
+
+	processPostMsg := func(msg *types.Message) error {
+		sectors := make([]uint64, 0)
+		rcpt, err := node.StateGetReceipt(ctx, msg.Cid(), actor.TipSet)
+		if err != nil {
+			return err
+		}
+		if rcpt.ExitCode.IsError() {
+			return nil
+		}
+		params := miner.SubmitWindowedPoStParams{}
+		if err := params.UnmarshalCBOR(bytes.NewBuffer(msg.Params)); err != nil {
+			return err
+		}
+
+		if partitions == nil {
+			partitions, err = loadPartitions(curState, epoch)
+			if err != nil {
+				return err
+			}
+		}
+
+		for _, p := range params.Partitions {
+			all, err := partitions[p.Index].AllSectors()
+			if err != nil {
+				return err
+			}
+			proven, err := bitfield.SubtractBitField(all, p.Skipped)
+			if err != nil {
+				return err
+			}
+
+			proven.ForEach(func(sector uint64) error {
+				sectors = append(sectors, sector)
+				return nil
+			})
+		}
+
+		for _, s := range sectors {
+			posts[s] = msg.Cid()
+		}
+		return nil
+	}
+
+	for _, msg := range msgs.BlsMessages {
+		if msg.To == actor.Address && msg.Method == 5 /* miner.SubmitWindowedPoSt */ {
+			if err := processPostMsg(msg); err != nil {
+				return nil, err
+			}
+		}
+	}
+	for _, msg := range msgs.SecpkMessages {
+		if msg.Message.To == actor.Address && msg.Message.Method == 5 /* miner.SubmitWindowedPoSt */ {
+			if err := processPostMsg(&msg.Message); err != nil {
+				return nil, err
+			}
+		}
+	}
+	return posts, nil
+}