feat: lotus-provider: WinningPoSt support

cmd/lotus-provider/migrate.go

@@ -6,6 +6,8 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"github.com/filecoin-project/go-address"
+	"github.com/ipfs/go-datastore"
 	"os"
 	"path"
 	"strings"
@@ -33,6 +35,12 @@ var configMigrateCmd = &cli.Command{
 			Value:   "~/.lotusminer",
 			Usage:   fmt.Sprintf("Specify miner repo path. flag(%s) and env(LOTUS_STORAGE_PATH) are DEPRECATION, will REMOVE SOON", FlagMinerRepoDeprecation),
 		},
+		&cli.StringFlag{
+			Name:    "repo",
+			EnvVars: []string{"LOTUS_PATH"},
+			Hidden:  true,
+			Value:   "~/.lotus",
+		},
 		&cli.StringFlag{
 			Name:    "to-layer",
 			Aliases: []string{"t"},
@@ -117,14 +125,20 @@ func fromMiner(cctx *cli.Context) (err error) {
 	}
 
 	// Populate Miner Address
-	sm, cc, err := cliutil.GetStorageMinerAPI(cctx)
+	mmeta, err := lr.Datastore(ctx, "/metadata")
+	if err != nil {
+		return xerrors.Errorf("opening miner metadata datastore: %w", err)
+	}
+	defer mmeta.Close()
+
+	maddrBytes, err := mmeta.Get(ctx, datastore.NewKey("miner-address"))
 	if err != nil {
-		return fmt.Errorf("could not get storageMiner API: %w", err)
+		return xerrors.Errorf("getting miner address datastore entry: %w", err)
 	}
-	defer cc()
-	addr, err := sm.ActorAddress(ctx)
+
+	addr, err := address.NewFromBytes(maddrBytes)
 	if err != nil {
-		return fmt.Errorf("could not read actor address: %w", err)
+		return xerrors.Errorf("parsing miner actor address: %w", err)
 	}
 
 	lpCfg.Addresses.MinerAddresses = []string{addr.String()}
@@ -137,7 +151,7 @@ func fromMiner(cctx *cli.Context) (err error) {
 	if err != nil {
 		return xerrors.Errorf("error getting JWTSecretName: %w", err)
 	}
-	lpCfg.Apis.StorageRPCSecret = base64.RawStdEncoding.EncodeToString(js.PrivateKey)
+	lpCfg.Apis.StorageRPCSecret = base64.StdEncoding.EncodeToString(js.PrivateKey)
 
 	// Populate API Key
 	_, header, err := cliutil.GetRawAPI(cctx, repo.FullNode, "v0")
@@ -161,7 +175,7 @@ environment variable LOTUS_WORKER_WINDOWPOST.
 	}
 
 	if !lo.Contains(titles, "base") {
-		_, err = db.Exec(ctx, "INSERT INTO harmony_config (title, config) VALUES ('base', '')", "base")
+		_, err = db.Exec(ctx, "INSERT INTO harmony_config (title, config) VALUES ('base', '')")
 		if err != nil {
 			return err
 		}

cmd/lotus-provider/run.go

@@ -40,6 +40,7 @@ import (
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/repo"
 	"github.com/filecoin-project/lotus/provider"
+	"github.com/filecoin-project/lotus/provider/lpwinning"
 	"github.com/filecoin-project/lotus/storage/paths"
 	"github.com/filecoin-project/lotus/storage/sealer"
 	"github.com/filecoin-project/lotus/storage/sealer/ffiwrapper"
@@ -251,6 +252,11 @@ var runCmd = &cli.Command{
 				}
 				activeTasks = append(activeTasks, wdPostTask, wdPoStSubmitTask, derlareRecoverTask)
 			}
+
+			if cfg.Subsystems.EnableWinningPost {
+				winPoStTask := lpwinning.NewWinPostTask(cfg.Subsystems.WinningPostMaxTasks, db, lw, verif, full, maddrs)
+				activeTasks = append(activeTasks, winPoStTask)
+			}
 		}
 		taskEngine, err := harmonytask.New(db, activeTasks, listenAddr)
 		if err != nil {

documentation/en/default-lotus-provider-config.toml

@@ -8,6 +8,9 @@
   # type: bool
   #EnableWinningPost = false
 
+  # type: int
+  #WinningPostMaxTasks = 0
+
 
 [Fees]
   # type: types.FIL

lib/harmony/harmonydb/sql/20230823.sql

@@ -37,12 +37,12 @@ create table wdpost_proofs
 create table wdpost_recovery_tasks
 (
     task_id              bigint not null
-        constraint wdpost_partition_tasks_pk
+        constraint wdpost_recovery_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
+    constraint wdpost_recovery_partition_tasks_identity_key
         unique (sp_id, proving_period_start, deadline_index, partition_index)
 );

lib/harmony/harmonydb/sql/20231110.sql

@@ -0,0 +1,39 @@
+create table mining_tasks
+(
+    task_id bigint not null
+        constraint mining_tasks_pk
+            primary key,
+    sp_id   bigint not null,
+    epoch   bigint not null,
+    base_compute_time timestamp not null,
+
+    won bool not null default false,
+    mined_cid text,
+    mined_header jsonb,
+    mined_at timestamp,
+
+    submitted_at timestamp,
+
+    constraint mining_tasks_sp_epoch
+        unique (sp_id, epoch)
+);
+
+create table mining_base_block
+(
+    id        bigserial not null
+        constraint mining_base_block_pk
+            primary key,
+    task_id   bigint    not null
+        constraint mining_base_block_mining_tasks_task_id_fk
+            references mining_tasks
+            on delete cascade,
+    sp_id     bigint,
+    block_cid text      not null,
+
+    no_win    bool      not null default false,
+
+    constraint mining_base_block_pk2
+        unique (sp_id, task_id, block_cid)
+);
+
+CREATE UNIQUE INDEX mining_base_block_cid_k ON mining_base_block (sp_id, block_cid) WHERE no_win = false;

miner/miner.go

@@ -71,7 +71,7 @@ func NewMiner(api v1api.FullNode, epp gen.WinningPoStProver, addr address.Addres
 		api:     api,
 		epp:     epp,
 		address: addr,
-		waitFunc: func(ctx context.Context, baseTime uint64) (func(bool, abi.ChainEpoch, error), abi.ChainEpoch, error) {
+		propagationWaitFunc: func(ctx context.Context, baseTime uint64) (func(bool, abi.ChainEpoch, error), abi.ChainEpoch, error) {
 			// wait around for half the block time in case other parents come in
 			//
 			// if we're mining a block in the past via catch-up/rush mining,
@@ -114,7 +114,7 @@ type Miner struct {
 	stop     chan struct{}
 	stopping chan struct{}
 
-	waitFunc waitFunc
+	propagationWaitFunc waitFunc
 
 	// lastWork holds the last MiningBase we built upon.
 	lastWork *MiningBase
@@ -205,15 +205,21 @@ func (m *Miner) mine(ctx context.Context) {
 	ctx, span := trace.StartSpan(ctx, "/mine")
 	defer span.End()
 
+	// Perform the Winning PoSt warmup in a separate goroutine.
 	go m.doWinPoStWarmup(ctx)
 
 	var lastBase MiningBase
+
+	// Start the main mining loop.
 minerLoop:
 	for {
+		// Prepare a context for a single node operation.
 		ctx := cliutil.OnSingleNode(ctx)
 
+		// Handle stop signals.
 		select {
 		case <-m.stop:
+			// If a stop signal is received, clean up and exit the mining loop.
 			stopping := m.stopping
 			m.stop = nil
 			m.stopping = nil
@@ -223,10 +229,11 @@ minerLoop:
 		default:
 		}
 
-		var base *MiningBase
+		var base *MiningBase // NOTE: This points to m.lastWork; Incrementing nulls here will increment it there.
 		var onDone func(bool, abi.ChainEpoch, error)
 		var injectNulls abi.ChainEpoch
 
+		// Look for the best mining candidate.
 		for {
 			prebase, err := m.GetBestMiningCandidate(ctx)
 			if err != nil {
@@ -237,6 +244,7 @@ minerLoop:
 				continue
 			}
 
+			// Check if we have a new base or if the current base is still valid.
 			if base != nil && base.TipSet.Height() == prebase.TipSet.Height() && base.NullRounds == prebase.NullRounds {
 				base = prebase
 				break
@@ -253,13 +261,13 @@ minerLoop:
 			// best mining candidate at that time.
 
 			// Wait until propagation delay period after block we plan to mine on
-			onDone, injectNulls, err = m.waitFunc(ctx, prebase.TipSet.MinTimestamp())
+			onDone, injectNulls, err = m.propagationWaitFunc(ctx, prebase.TipSet.MinTimestamp())
 			if err != nil {
 				log.Error(err)
 				continue
 			}
 
-			// just wait for the beacon entry to become available before we select our final mining base
+			// Ensure the beacon entry is available before finalizing the mining base.
 			_, err = m.api.StateGetBeaconEntry(ctx, prebase.TipSet.Height()+prebase.NullRounds+1)
 			if err != nil {
 				log.Errorf("failed getting beacon entry: %s", err)
@@ -272,8 +280,9 @@ minerLoop:
 			base = prebase
 		}
 
-		base.NullRounds += injectNulls // testing
+		base.NullRounds += injectNulls // Adjust for testing purposes.
 
+		// Check for repeated mining candidates and handle sleep for the next round.
 		if base.TipSet.Equals(lastBase.TipSet) && lastBase.NullRounds == base.NullRounds {
 			log.Warnf("BestMiningCandidate from the previous round: %s (nulls:%d)", lastBase.TipSet.Cids(), lastBase.NullRounds)
 			if !m.niceSleep(time.Duration(build.BlockDelaySecs) * time.Second) {
@@ -282,6 +291,7 @@ minerLoop:
 			continue
 		}
 
+		// Attempt to mine a block.
 		b, err := m.mineOne(ctx, base)
 		if err != nil {
 			log.Errorf("mining block failed: %+v", err)
@@ -299,9 +309,12 @@ minerLoop:
 		}
 		onDone(b != nil, h, nil)
 
+		// Process the mined block.
 		if b != nil {
+			// Record the event of mining a block.
 			m.journal.RecordEvent(m.evtTypes[evtTypeBlockMined], func() interface{} {
 				return map[string]interface{}{
+					// Data about the mined block.
 					"parents":   base.TipSet.Cids(),
 					"nulls":     base.NullRounds,
 					"epoch":     b.Header.Height,
@@ -312,19 +325,23 @@ minerLoop:
 
 			btime := time.Unix(int64(b.Header.Timestamp), 0)
 			now := build.Clock.Now()
+			// Handle timing for broadcasting the block.
 			switch {
 			case btime == now:
 				// block timestamp is perfectly aligned with time.
 			case btime.After(now):
+				// Wait until it's time to broadcast the block.
 				if !m.niceSleep(build.Clock.Until(btime)) {
 					log.Warnf("received interrupt while waiting to broadcast block, will shutdown after block is sent out")
 					build.Clock.Sleep(build.Clock.Until(btime))
 				}
 			default:
+				// Log if the block was mined in the past.
 				log.Warnw("mined block in the past",
 					"block-time", btime, "time", build.Clock.Now(), "difference", build.Clock.Since(btime))
 			}
 
+			// Check for slash filter conditions.
 			if os.Getenv("LOTUS_MINER_NO_SLASHFILTER") != "_yes_i_know_i_can_and_probably_will_lose_all_my_fil_and_power_" && !build.IsNearUpgrade(base.TipSet.Height(), build.UpgradeWatermelonFixHeight) {
 				witness, fault, err := m.sf.MinedBlock(ctx, b.Header, base.TipSet.Height()+base.NullRounds)
 				if err != nil {
@@ -339,25 +356,27 @@ minerLoop:
 				}
 			}
 
+			// Check for blocks created at the same height.
 			if _, ok := m.minedBlockHeights.Get(b.Header.Height); ok {
 				log.Warnw("Created a block at the same height as another block we've created", "height", b.Header.Height, "miner", b.Header.Miner, "parents", b.Header.Parents)
 				continue
 			}
 
+			// Add the block height to the mined block heights.
 			m.minedBlockHeights.Add(b.Header.Height, true)
 
+			// Submit the newly mined block.
 			if err := m.api.SyncSubmitBlock(ctx, b); err != nil {
 				log.Errorf("failed to submit newly mined block: %+v", err)
 			}
 		} else {
+			// If no block was mined, increase the null rounds and wait for the next epoch.
 			base.NullRounds++
 
-			// Wait until the next epoch, plus the propagation delay, so a new tipset
-			// has enough time to form.
-			//
-			// See:  https://github.com/filecoin-project/lotus/issues/1845
+			// Calculate the time for the next round.
 			nextRound := time.Unix(int64(base.TipSet.MinTimestamp()+build.BlockDelaySecs*uint64(base.NullRounds))+int64(build.PropagationDelaySecs), 0)
 
+			// Wait for the next round or stop signal.
 			select {
 			case <-build.Clock.After(build.Clock.Until(nextRound)):
 			case <-m.stop:

miner/testminer.go

@@ -28,13 +28,13 @@ func NewTestMiner(nextCh <-chan MineReq, addr address.Address) func(v1api.FullNo
 		}
 
 		m := &Miner{
-			api:               api,
-			waitFunc:          chanWaiter(nextCh),
-			epp:               epp,
-			minedBlockHeights: arc,
-			address:           addr,
-			sf:                slashfilter.New(ds.NewMapDatastore()),
-			journal:           journal.NilJournal(),
+			api:                 api,
+			propagationWaitFunc: chanWaiter(nextCh),
+			epp:                 epp,
+			minedBlockHeights:   arc,
+			address:             addr,
+			sf:                  slashfilter.New(ds.NewMapDatastore()),
+			journal:             journal.NilJournal(),
 		}
 
 		if err := m.Start(context.TODO()); err != nil {

node/builder_miner.go

@@ -121,8 +121,12 @@ func ConfigStorageMiner(c interface{}) Option {
 
 			// Mining / proving
 			Override(new(*slashfilter.SlashFilter), modules.NewSlashFilter),
-			Override(new(*miner.Miner), modules.SetupBlockProducer),
-			Override(new(gen.WinningPoStProver), storage.NewWinningPoStProver),
+
+			If(!cfg.Subsystems.DisableWinningPoSt,
+				Override(new(*miner.Miner), modules.SetupBlockProducer),
+				Override(new(gen.WinningPoStProver), storage.NewWinningPoStProver),
+			),
+
 			Override(PreflightChecksKey, modules.PreflightChecks),
 			Override(new(*sealing.Sealing), modules.SealingPipeline(cfg.Fees)),
 

node/config/types.go

@@ -92,9 +92,10 @@ type JournalConfig struct {
 }
 
 type ProviderSubsystemsConfig struct {
-	EnableWindowPost   bool
-	WindowPostMaxTasks int
-	EnableWinningPost  bool
+	EnableWindowPost    bool
+	WindowPostMaxTasks  int
+	EnableWinningPost   bool
+	WinningPostMaxTasks int
 }
 
 type DAGStoreConfig struct {
@@ -161,6 +162,13 @@ type MinerSubsystemConfig struct {
 	//   to window post, including scheduling, submitting proofs, and recovering
 	//   sectors.
 	DisableWindowPoSt bool
+
+	// When winning post is disabled, the miner process will NOT attempt to mine
+	// blocks. This should only be set when there's an external process mining
+	// blocks on behalf of the miner.
+	// When disabled and no external block producers are configured, all potential
+	// block rewards will be missed!
+	DisableWinningPoSt bool
 }
 
 type DealmakingConfig struct {