feat: wdpost: Declare-recovered task

cmd/lotus-provider/run.go

@@ -244,12 +244,12 @@ var runCmd = &cli.Command{
 		{
 
 			if cfg.Subsystems.EnableWindowPost {
-				wdPostTask, wdPoStSubmitTask, err := provider.WindowPostScheduler(ctx, cfg.Fees, cfg.Proving, full, verif, lw,
+				wdPostTask, wdPoStSubmitTask, derlareRecoverTask, err := provider.WindowPostScheduler(ctx, cfg.Fees, cfg.Proving, full, verif, lw,
 					as, maddrs, db, stor, si, cfg.Subsystems.WindowPostMaxTasks)
 				if err != nil {
 					return err
 				}
-				activeTasks = append(activeTasks, wdPostTask, wdPoStSubmitTask)
+				activeTasks = append(activeTasks, wdPostTask, wdPoStSubmitTask, derlareRecoverTask)
 			}
 		}
 		taskEngine, err := harmonytask.New(db, activeTasks, listenAddr)

lib/harmony/harmonydb/sql/20230823.sql

@@ -33,3 +33,16 @@ create table wdpost_proofs
     constraint wdpost_proofs_identity_key
         unique (sp_id, proving_period_start, deadline, partition)
 );
+
+create table wdpost_recovery_tasks
+(
+    task_id              bigint not null
+        constraint wdpost_partition_tasks_pk
+            primary key,
+    sp_id                bigint not null,
+    proving_period_start bigint not null,
+    deadline_index       bigint not null,
+    partition_index      bigint not null,
+    constraint wdpost_partition_tasks_identity_key
+        unique (sp_id, proving_period_start, deadline_index, partition_index)
+);

provider/builder.go

@@ -23,7 +23,8 @@ var log = logging.Logger("provider")
 
 func WindowPostScheduler(ctx context.Context, fc config.LotusProviderFees, pc config.ProvingConfig,
 	api api.FullNode, verif storiface.Verifier, lw *sealer.LocalWorker,
-	as *ctladdr.AddressSelector, maddr []dtypes.MinerAddress, db *harmonydb.DB, stor paths.Store, idx paths.SectorIndex, max int) (*lpwindow.WdPostTask, *lpwindow.WdPostSubmitTask, error) {
+	as *ctladdr.AddressSelector, addresses []dtypes.MinerAddress, db *harmonydb.DB,
+	stor paths.Store, idx paths.SectorIndex, max int) (*lpwindow.WdPostTask, *lpwindow.WdPostSubmitTask, *lpwindow.WdPostRecoverDeclareTask, error) {
 
 	chainSched := chainsched.New(api)
 
@@ -32,17 +33,22 @@ func WindowPostScheduler(ctx context.Context, fc config.LotusProviderFees, pc co
 
 	sender := lpmessage.NewSender(api, api, db)
 
-	computeTask, err := lpwindow.NewWdPostTask(db, api, ft, lw, verif, chainSched, maddr, max)
+	computeTask, err := lpwindow.NewWdPostTask(db, api, ft, lw, verif, chainSched, addresses, max)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
 	submitTask, err := lpwindow.NewWdPostSubmitTask(chainSched, sender, db, api, fc.MaxWindowPoStGasFee, as)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
+	}
+
+	recoverTask, err := lpwindow.NewWdPostRecoverDeclareTask(sender, db, api, ft, as, chainSched, fc.MaxWindowPoStGasFee, addresses)
+	if err != nil {
+		return nil, nil, nil, err
 	}
 
 	go chainSched.Run(ctx)
 
-	return computeTask, submitTask, nil
+	return computeTask, submitTask, recoverTask, nil
 }

provider/lpwindow/compute_do.go

@@ -3,6 +3,8 @@ package lpwindow
 import (
 	"bytes"
 	"context"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/storage/sealer"
 	"sort"
 	"sync"
 	"time"
@@ -92,7 +94,7 @@ func (t *WdPostTask) doPartition(ctx context.Context, ts *types.TipSet, maddr ad
 			return nil, xerrors.Errorf("copy toProve: %w", err)
 		}
 		if !disablePreChecks {
-			good, err = t.checkSectors(ctx, maddr, toProve, ts.Key())
+			good, err = checkSectors(ctx, t.api, t.faultTracker, maddr, toProve, ts.Key())
 			if err != nil {
 				return nil, xerrors.Errorf("checking sectors to skip: %w", err)
 			}
@@ -222,13 +224,18 @@ func (t *WdPostTask) doPartition(ctx context.Context, ts *types.TipSet, maddr ad
 	return nil, xerrors.Errorf("failed to generate window post")
 }
 
-func (t *WdPostTask) checkSectors(ctx context.Context, maddr address.Address, check bitfield.BitField, tsk types.TipSetKey) (bitfield.BitField, error) {
+type CheckSectorsAPI interface {
+	StateMinerSectors(ctx context.Context, addr address.Address, bf *bitfield.BitField, tsk types.TipSetKey) ([]*miner.SectorOnChainInfo, error)
+}
+
+func checkSectors(ctx context.Context, api CheckSectorsAPI, ft sealer.FaultTracker,
+	maddr address.Address, check bitfield.BitField, tsk types.TipSetKey) (bitfield.BitField, error) {
 	mid, err := address.IDFromAddress(maddr)
 	if err != nil {
 		return bitfield.BitField{}, xerrors.Errorf("failed to convert to ID addr: %w", err)
 	}
 
-	sectorInfos, err := t.api.StateMinerSectors(ctx, maddr, &check, tsk)
+	sectorInfos, err := api.StateMinerSectors(ctx, maddr, &check, tsk)
 	if err != nil {
 		return bitfield.BitField{}, xerrors.Errorf("failed to get sector infos: %w", err)
 	}
@@ -267,7 +274,7 @@ func (t *WdPostTask) checkSectors(ctx context.Context, maddr address.Address, ch
 		return bitfield.BitField{}, xerrors.Errorf("failed to convert to v1_1 post proof: %w", err)
 	}
 
-	bad, err := t.faultTracker.CheckProvable(ctx, pp, tocheck, func(ctx context.Context, id abi.SectorID) (cid.Cid, bool, error) {
+	bad, err := ft.CheckProvable(ctx, pp, tocheck, func(ctx context.Context, id abi.SectorID) (cid.Cid, bool, error) {
 		s, ok := sectors[id.Number]
 		if !ok {
 			return cid.Undef, false, xerrors.Errorf("sealed CID not found")