provider/lpseal/poller.go

@@ -15,6 +15,7 @@ var log = logging.Logger("lpseal")
 const (
 	pollerSDR = iota
 	pollerTrees
+	pollerPrecommitMsg
 
 	numPollers
 )
@@ -123,7 +124,7 @@ func (s *SealPoller) poll(ctx context.Context) error {
 		if task.TaskTreeD == nil && task.TaskTreeC == nil && task.TaskTreeR == nil && s.pollers[pollerTrees].IsSet() && task.AfterSDR {
 			s.pollers[pollerTrees].Val(ctx)(func(id harmonytask.TaskID, tx *harmonydb.Tx) (shouldCommit bool, seriousError error) {
 				n, err := tx.Exec(`UPDATE sectors_sdr_pipeline SET task_id_tree_d = $1, task_id_tree_c = $1, task_id_tree_r = $1
-                            WHERE sp_id = $2 AND sector_number = $3 and task_id_tree_d is null and task_id_tree_c is null and task_id_tree_r is null`, id, task.SpID, task.SectorNumber)
+                            WHERE sp_id = $2 AND sector_number = $3 and after_sdr = true and task_id_tree_d is null and task_id_tree_c is null and task_id_tree_r is null`, id, task.SpID, task.SectorNumber)
 				if err != nil {
 					return false, xerrors.Errorf("update sectors_sdr_pipeline: %w", err)
 				}
@@ -136,7 +137,17 @@ func (s *SealPoller) poll(ctx context.Context) error {
 		}
 
 		if task.TaskPrecommitMsg == nil && task.AfterTreeR && task.AfterTreeD {
-			// todo start precommit msg task
+			s.pollers[pollerPrecommitMsg].Val(ctx)(func(id harmonytask.TaskID, tx *harmonydb.Tx) (shouldCommit bool, seriousError error) {
+				n, err := tx.Exec(`UPDATE sectors_sdr_pipeline SET task_id_precommit_msg = $1 WHERE sp_id = $2 AND sector_number = $3 and task_id_precommit_msg is null and after_tree_r = true and after_tree_d = true`, id, task.SpID, task.SectorNumber)
+				if err != nil {
+					return false, xerrors.Errorf("update sectors_sdr_pipeline: %w", err)
+				}
+				if n != 1 {
+					return false, xerrors.Errorf("expected to update 1 row, updated %d", n)
+				}
+
+				return true, nil
+			})
 		}
 
 		if task.TaskPrecommitMsgWait == nil && task.AfterPrecommitMsg {

provider/lpseal/task_submit_precommit.go

@@ -0,0 +1,148 @@
+package lpseal
+
+import (
+	"bytes"
+	"context"
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/big"
+	"github.com/filecoin-project/go-state-types/builtin"
+	miner12 "github.com/filecoin-project/go-state-types/builtin/v12/miner"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
+	"github.com/filecoin-project/lotus/lib/harmony/harmonytask"
+	"github.com/filecoin-project/lotus/lib/harmony/resources"
+	"github.com/filecoin-project/lotus/provider/lpmessage"
+	"github.com/ipfs/go-cid"
+	"golang.org/x/xerrors"
+)
+
+type SubmitPrecommitTaskApi interface {
+	StateMinerPreCommitDepositForPower(context.Context, address.Address, miner.SectorPreCommitInfo, types.TipSetKey) (big.Int, error)
+}
+
+type SubmitPrecommitTask struct {
+	sp     *SealPoller
+	db     *harmonydb.DB
+	api    SubmitPrecommitTaskApi
+	sender *lpmessage.Sender
+
+	maxFee types.FIL
+}
+
+func (s *SubmitPrecommitTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) {
+	ctx := context.Background()
+
+	var sectorParamsArr []struct {
+		SpID         int64                   `db:"sp_id"`
+		SectorNumber int64                   `db:"sector_number"`
+		RegSealProof abi.RegisteredSealProof `db:"reg_seal_proof"`
+		TicketEpoch  abi.ChainEpoch          `db:"ticket_epoch"`
+		SealedCID    string                  `db:"sealed_cid"`
+		UnsealedCID  string                  `db:"unsealed_cid"`
+	}
+
+	err = s.db.Select(ctx, &sectorParamsArr, `
+		SELECT sp_id, sector_number, reg_seal_proof, ticket_epoch, tree_r_cid, tree_d_cid
+		FROM sectors_sdr_pipeline
+		WHERE task_id_precommit_msg = $1`, taskID)
+	if err != nil {
+		return false, xerrors.Errorf("getting sector params: %w", err)
+	}
+
+	if len(sectorParamsArr) != 1 {
+		return false, xerrors.Errorf("expected 1 sector params, got %d", len(sectorParamsArr))
+	}
+	sectorParams := sectorParamsArr[0]
+
+	maddr, err := address.NewIDAddress(uint64(sectorParams.SpID))
+	if err != nil {
+		return false, xerrors.Errorf("getting miner address: %w", err)
+	}
+
+	sealedCID, err := cid.Parse(sectorParams.SealedCID)
+	if err != nil {
+		return false, xerrors.Errorf("parsing sealed CID: %w", err)
+	}
+
+	unsealedCID, err := cid.Parse(sectorParams.UnsealedCID)
+	if err != nil {
+		return false, xerrors.Errorf("parsing unsealed CID: %w", err)
+	}
+	_ = unsealedCID
+
+	params := miner.PreCommitSectorBatchParams2{}
+
+	expiration := sectorParams.TicketEpoch + miner12.MaxSectorExpirationExtension
+
+	params.Sectors = append(params.Sectors, miner.SectorPreCommitInfo{
+		SealProof:     sectorParams.RegSealProof,
+		SectorNumber:  abi.SectorNumber(sectorParams.SectorNumber),
+		SealedCID:     sealedCID,
+		SealRandEpoch: sectorParams.TicketEpoch,
+		DealIDs:       nil, // todo
+		Expiration:    expiration,
+		//UnsealedCid:   unsealedCID, // todo with deals
+	})
+	var pbuf bytes.Buffer
+	if err := params.MarshalCBOR(&pbuf); err != nil {
+		return false, xerrors.Errorf("serializing params: %w", err)
+	}
+
+	collateral, err := s.api.StateMinerPreCommitDepositForPower(ctx, maddr, params.Sectors[0], types.EmptyTSK)
+	if err != nil {
+		return false, xerrors.Errorf("getting precommit deposit: %w", err)
+	}
+
+	msg := &types.Message{
+		To:     maddr,
+		Method: builtin.MethodsMiner.PreCommitSectorBatch2,
+		Params: pbuf.Bytes(),
+		Value:  collateral, // todo config for pulling from miner balance!!
+	}
+
+	mss := &api.MessageSendSpec{
+		MaxFee: abi.TokenAmount(s.maxFee),
+	}
+
+	mcid, err := s.sender.Send(ctx, msg, mss, "precommit")
+	if err != nil {
+		return false, xerrors.Errorf("sending message: %w", err)
+	}
+
+	// set precommit_msg_cid
+	_, err = s.db.Exec(ctx, `UPDATE sectors_sdr_pipeline
+		SET precommit_msg_cid = $1
+		WHERE task_id_precommit_msg = $2`, mcid, taskID)
+	if err != nil {
+		return false, xerrors.Errorf("updating precommit_msg_cid: %w", err)
+	}
+
+	return true, nil
+}
+
+func (s *SubmitPrecommitTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) (*harmonytask.TaskID, error) {
+	id := ids[0]
+	return &id, nil
+}
+
+func (s *SubmitPrecommitTask) TypeDetails() harmonytask.TaskTypeDetails {
+	return harmonytask.TaskTypeDetails{
+		Max:  1024,
+		Name: "PreCommitSubmit",
+		Cost: resources.Resources{
+			Cpu: 0,
+			Gpu: 0,
+			Ram: 1 << 20,
+		},
+		MaxFailures: 16,
+	}
+}
+
+func (s *SubmitPrecommitTask) Adder(taskFunc harmonytask.AddTaskFunc) {
+	s.sp.pollers[pollerPrecommitMsg].Set(taskFunc)
+}
+
+var _ harmonytask.TaskInterface = &SubmitPrecommitTask{}