lpseal: Implement SDR for CC sectors

cmd/lotus-provider/pipeline.go

@@ -3,7 +3,6 @@ package main
 import (
 	"fmt"
 	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/go-commp-utils/zerocomm"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -11,7 +10,6 @@ import (
 	"github.com/filecoin-project/lotus/cmd/lotus-provider/deps"
 	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
 	"github.com/filecoin-project/lotus/provider/lpseal"
-	"github.com/ipfs/go-cid"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
 )
@@ -64,6 +62,9 @@ var pipelineStartCmd = &cli.Command{
 		if !cctx.IsSet("actor") {
 			return cli.ShowCommandHelp(cctx, "start")
 		}
+		if !cctx.Bool("cc") {
+			return xerrors.Errorf("only CC sectors supported for now")
+		}
 
 		act, err := address.NewFromString(cctx.String("actor"))
 		if err != nil {
@@ -110,21 +111,9 @@ var pipelineStartCmd = &cli.Command{
 			return xerrors.Errorf("getting seal proof type: %w", err)
 		}
 
-		ssize, err := spt.SectorSize()
-		if err != nil {
-			return xerrors.Errorf("getting sector size: %w", err)
-		}
-
-		var commd cid.Cid
-		if cctx.Bool("cc") {
-			commd = zerocomm.ZeroPieceCommitment(abi.PaddedPieceSize(ssize).Unpadded())
-		} else {
-			return xerrors.Errorf("only CC sectors supported for now")
-		}
-
 		num, err := lpseal.AllocateSectorNumbers(ctx, dep.Full, dep.DB, act, cctx.Int("count"), func(tx *harmonydb.Tx, numbers []abi.SectorNumber) (bool, error) {
 			for _, n := range numbers {
-				_, err := tx.Exec("insert into sectors_sdr_pipeline (sp_id, sector_number, reg_seal_proof, comm_d_cid) values ($1, $2, $3, $4)", mid, n, spt, commd.String())
+				_, err := tx.Exec("insert into sectors_sdr_pipeline (sp_id, sector_number, reg_seal_proof) values ($1, $2, $3)", mid, n, spt)
 				if err != nil {
 					return false, xerrors.Errorf("inserting into sectors_sdr_pipeline: %w", err)
 				}

lib/harmony/harmonydb/sql/20231217-sdr-pipeline.sql

@@ -8,7 +8,6 @@ create table sectors_sdr_pipeline (
     -- at request time
     create_time timestamp not null default current_timestamp,
     reg_seal_proof int not null,
-    comm_d_cid text not null,
 
     -- sdr
     ticket_epoch bigint,
@@ -83,6 +82,9 @@ create table sectors_sdr_initial_pieces (
     piece_cid text not null,
     piece_size bigint not null,
 
+    -- foreign key
+    foreign key (sp_id, sector_number) references sectors_sdr_pipeline (sp_id, sector_number) on delete cascade,
+
     primary key (sp_id, sector_number, piece_index)
 );
 

provider/lpffi/sdr_funcs.go

@@ -0,0 +1,56 @@
+package lpffi
+
+import (
+	"context"
+	ffi "github.com/filecoin-project/filecoin-ffi"
+	commcid "github.com/filecoin-project/go-fil-commcid"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/lotus/storage/sealer/ffiwrapper"
+	"github.com/filecoin-project/lotus/storage/sealer/storiface"
+	"github.com/ipfs/go-cid"
+	"golang.org/x/xerrors"
+)
+
+type ExternPrecommit2 func(ctx context.Context, sector storiface.SectorRef, cache, sealed string, pc1out storiface.PreCommit1Out) (sealedCID cid.Cid, unsealedCID cid.Cid, err error)
+
+type ExternalSealer struct {
+	PreCommit2 ExternPrecommit2
+}
+
+type SealCalls struct {
+	sectors ffiwrapper.SectorProvider
+
+	// externCalls cointain overrides for calling alternative sealing logic
+	externCalls ExternalSealer
+}
+
+func (sb *SealCalls) GenerateSDR(ctx context.Context, sector storiface.SectorRef, ticket abi.SealRandomness, commKcid cid.Cid) error {
+	paths, releaseSector, err := sb.sectors.AcquireSector(ctx, sector, storiface.FTCache, storiface.FTNone, storiface.PathSealing)
+	if err != nil {
+		return xerrors.Errorf("acquiring sector paths: %w", err)
+	}
+	defer releaseSector()
+
+	// prepare SDR params
+	commp, err := commcid.CIDToDataCommitmentV1(commKcid)
+	if err != nil {
+		return xerrors.Errorf("computing commK: %w", err)
+	}
+
+	replicaID, err := sector.ProofType.ReplicaId(sector.ID.Miner, sector.ID.Number, ticket, commp)
+	if err != nil {
+		return xerrors.Errorf("computing replica id: %w", err)
+	}
+
+	// generate new sector key
+	err = ffi.GenerateSDR(
+		sector.ProofType,
+		paths.Cache,
+		replicaID,
+	)
+	if err != nil {
+		return xerrors.Errorf("generating SDR %d (%s): %w", sector.ID.Number, paths.Unsealed, err)
+	}
+
+	return nil
+}

provider/lpseal/poller.go

@@ -0,0 +1,129 @@
+package lpseal
+
+import (
+	"context"
+	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
+	"github.com/filecoin-project/lotus/lib/harmony/harmonytask"
+	"github.com/filecoin-project/lotus/lib/promise"
+	logging "github.com/ipfs/go-log/v2"
+	"golang.org/x/xerrors"
+	"time"
+)
+
+var log = logging.Logger("lpseal")
+
+const (
+	pollerSDR = iota
+
+	numPollers
+)
+
+const sealPollerInterval = 10 * time.Second
+
+type SealPoller struct {
+	db *harmonydb.DB
+
+	pollers [numPollers]promise.Promise[harmonytask.AddTaskFunc]
+}
+
+func (s *SealPoller) RunPoller(ctx context.Context) error {
+	ticker := time.NewTicker(sealPollerInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			s.poll(ctx)
+		}
+	}
+}
+
+func (s *SealPoller) poll(ctx context.Context) error {
+	var tasks []struct {
+		SpID         int64 `db:"sp_id"`
+		SectorNumber int64 `db:"sector_number"`
+
+		TaskSDR  *int64 `db:"task_id_sdr"`
+		AfterSDR bool   `db:"after_sdr"`
+
+		TaskTreeD  *int64 `db:"task_id_tree_d"`
+		AfterTreeD bool   `db:"after_tree_d"`
+
+		TaskTreeC  *int64 `db:"task_id_tree_c"`
+		AfterTreeC bool   `db:"after_tree_c"`
+
+		TaskTreeR  *int64 `db:"task_id_tree_r"`
+		AfterTreeR bool   `db:"after_tree_r"`
+
+		TaskPrecommitMsg  *int64 `db:"task_id_precommit_msg"`
+		AfterPrecommitMsg bool   `db:"after_precommit_msg"`
+
+		TaskPrecommitMsgWait     *int64 `db:"task_id_precommit_msg_wait"`
+		AfterPrecommitMsgSuccess bool   `db:"after_precommit_msg_success"`
+
+		TaskPoRep  *int64 `db:"task_id_porep"`
+		PoRepProof []byte `db:"porep_proof"`
+
+		TaskCommitMsg  *int64 `db:"task_id_commit_msg"`
+		AfterCommitMsg bool   `db:"after_commit_msg"`
+
+		TaskCommitMsgWait     *int64 `db:"task_id_commit_msg_wait"`
+		AfterCommitMsgSuccess bool   `db:"after_commit_msg_success"`
+
+		Failed       bool   `db:"failed"`
+		FailedReason string `db:"failed_reason"`
+	}
+
+	err := s.db.Select(ctx, &tasks, `SELECT * FROM sectors_sdr_pipeline WHERE after_commit_msg_success != true`)
+	if err != nil {
+		return err
+	}
+
+	for _, task := range tasks {
+		if task.Failed {
+			continue
+		}
+
+		if task.TaskSDR == nil {
+			s.pollers[pollerSDR].Val(ctx)(func(id harmonytask.TaskID, tx *harmonydb.Tx) (shouldCommit bool, seriousError error) {
+				n, err := tx.Exec(`UPDATE sectors_sdr_pipeline SET task_id_sdr = $1 WHERE sp_id = $2 AND sector_number = $3`, 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.TaskTreeD == nil {
+			// todo start tree d task
+		}
+
+		// todo those two are really one pc2
+		if task.TaskTreeC == nil && task.AfterSDR {
+			// todo start tree c task
+		}
+		if task.TaskTreeR == nil && task.AfterTreeC {
+			// todo start tree r task
+		}
+
+		if task.TaskPrecommitMsg == nil && task.AfterTreeR && task.AfterTreeD {
+			// todo start precommit msg task
+		}
+
+		if task.TaskPrecommitMsgWait == nil && task.AfterPrecommitMsg {
+			// todo start precommit msg wait task
+		}
+
+		todoWaitSeed := false
+		if task.TaskPoRep == nil && todoWaitSeed {
+			// todo start porep task
+		}
+	}
+
+	return nil
+}

provider/lpseal/task_sdr.go

@@ -1,44 +1,160 @@
 package lpseal
 
 import (
+	"bytes"
+	"context"
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-commp-utils/zerocomm"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/crypto"
+	"github.com/filecoin-project/lotus/chain/actors/policy"
+	"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/lpffi"
+	"github.com/filecoin-project/lotus/storage/sealer/storiface"
+	"golang.org/x/xerrors"
 )
 
 type SDRAPI interface {
+	ChainHead(context.Context) (*types.TipSet, error)
+	StateGetRandomnessFromTickets(context.Context, crypto.DomainSeparationTag, abi.ChainEpoch, []byte, types.TipSetKey) (abi.Randomness, error)
 }
 
 type SDRTask struct {
 	api SDRAPI
 	db  *harmonydb.DB
+	sp  *SealPoller
+
+	sc *lpffi.SealCalls
 
 	maxSDR int
 }
 
 func (s *SDRTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) {
-	//TODO implement me
-	panic("implement me")
+	ctx := context.Background()
+
+	var sectorParams struct {
+		SpID         int64                   `db:"sp_id"`
+		SectorNumber int64                   `db:"sector_number"`
+		RegSealProof abi.RegisteredSealProof `db:"reg_seal_proof"`
+	}
+
+	err = s.db.Select(ctx, &sectorParams, `
+		SELECT sp_id, sector_number, reg_seal_proof
+		FROM sectors_sdr_pipeline`)
+	if err != nil {
+		return false, xerrors.Errorf("getting sector params: %w", err)
+	}
+
+	var pieces []struct {
+		PieceIndex int64  `db:"piece_index"`
+		PieceCID   string `db:"piece_cid"`
+		PieceSize  int64  `db:"piece_size"`
+	}
+
+	err = s.db.Select(ctx, &pieces, `
+		SELECT piece_index, piece_cid, piece_size
+		FROM sectors_sdr_initial_pieces
+		WHERE sp_id = $1 AND sector_number = $2`, sectorParams.SpID, sectorParams.SectorNumber)
+	if err != nil {
+		return false, xerrors.Errorf("getting pieces: %w", err)
+	}
+
+	if len(pieces) > 0 {
+		// todo sdr with data
+		return false, xerrors.Errorf("todo sdr with data")
+	}
+
+	ssize, err := sectorParams.RegSealProof.SectorSize()
+	if err != nil {
+		return false, xerrors.Errorf("getting sector size: %w", err)
+	}
+
+	commd := zerocomm.ZeroPieceCommitment(abi.PaddedPieceSize(ssize).Unpadded())
+
+	sref := storiface.SectorRef{
+		ID: abi.SectorID{
+			Miner:  abi.ActorID(sectorParams.SpID),
+			Number: abi.SectorNumber(sectorParams.SectorNumber),
+		},
+		ProofType: sectorParams.RegSealProof,
+	}
+
+	// get ticket
+	maddr, err := address.NewIDAddress(uint64(sectorParams.SpID))
+	if err != nil {
+		return false, xerrors.Errorf("getting miner address: %w", err)
+	}
+
+	ticket, err := s.getTicket(ctx, maddr)
+	if err != nil {
+		return false, xerrors.Errorf("getting ticket: %w", err)
+	}
+
+	// do the SDR!!
+
+	err = s.sc.GenerateSDR(ctx, sref, ticket, commd)
+	if err != nil {
+		return false, xerrors.Errorf("generating sdr: %w", err)
+	}
+
+	// store success!
+	n, err := s.db.Exec(ctx, `UPDATE sectors_sdr_pipeline SET after_sdr = true WHERE sp_id = $1 AND sector_number = $2`, sectorParams.SpID, sectorParams.SectorNumber)
+	if err != nil {
+		return false, xerrors.Errorf("store sdr success: updating pipeline: %w", err)
+	}
+	if n != 1 {
+		return false, xerrors.Errorf("store sdr success: updated %d rows", n)
+	}
+
+	return true, nil
+}
+
+func (s *SDRTask) getTicket(ctx context.Context, maddr address.Address) (abi.SealRandomness, error) {
+	ts, err := s.api.ChainHead(ctx)
+	if err != nil {
+		return nil, xerrors.Errorf("getting chain head: %w", err)
+	}
+
+	ticketEpoch := ts.Height() - policy.SealRandomnessLookback
+	buf := new(bytes.Buffer)
+	if err := maddr.MarshalCBOR(buf); err != nil {
+		return nil, xerrors.Errorf("marshaling miner address: %w", err)
+	}
+
+	rand, err := s.api.StateGetRandomnessFromTickets(ctx, crypto.DomainSeparationTag_SealRandomness, ticketEpoch, buf.Bytes(), ts.Key())
+	if err != nil {
+		return nil, xerrors.Errorf("getting randomness from tickets: %w", err)
+	}
+
+	return abi.SealRandomness(rand), nil
 }
 
 func (s *SDRTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) (*harmonytask.TaskID, error) {
-	//TODO implement me
-	panic("implement me")
+	// todo check storage (reserve too?)
+
+	id := ids[0]
+	return &id, nil
 }
 
 func (s *SDRTask) TypeDetails() harmonytask.TaskTypeDetails {
 	return harmonytask.TaskTypeDetails{
-		Max:         s.maxSDR,
-		Name:        "SDR",
-		Cost:        resources.Resources{},
+		Max:  s.maxSDR,
+		Name: "SDR",
+		Cost: resources.Resources{ // todo offset for prefetch?
+			Cpu: 4, // todo multicore sdr
+			Gpu: 0,
+			Ram: 54 << 30, // todo measure; lower on 2k devnetn
+		},
 		MaxFailures: 0,
 		Follows:     nil,
 	}
 }
 
 func (s *SDRTask) Adder(taskFunc harmonytask.AddTaskFunc) {
-	//TODO implement me
-	panic("implement me")
+	s.sp.pollers[pollerSDR].Set(taskFunc)
 }
 
 var _ harmonytask.TaskInterface = &SDRTask{}

storage/sealer/ffiwrapper/sealer_cgo.go

@@ -520,7 +520,7 @@ func (sb *Sealer) regenerateSectorKey(ctx context.Context, sector storiface.Sect
 	// prepare SDR params
 	commp, err := commcid.CIDToDataCommitmentV1(keyDataCid)
 	if err != nil {
-		return xerrors.Errorf("computing commP: %w", err)
+		return xerrors.Errorf("computing commK: %w", err)
 	}
 
 	replicaID, err := sector.ProofType.ReplicaId(sector.ID.Miner, sector.ID.Number, ticket, commp)