lpwinning: adder done

lib/harmony/harmonydb/sql/20231110.sql

@@ -0,0 +1,24 @@
+create table mining_tasks
+(
+    task_id bigint not null
+        constraint mining_tasks_pk
+            primary key,
+    sp_id   bigint,
+    epoch   bigint,
+    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,
+    block_cid text      not null
+        constraint mining_base_block_cid_k
+            unique
+);

provider/lpwinning/winning_task.go

@@ -4,13 +4,17 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/binary"
+	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/build"
 	"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/lib/promise"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	logging "github.com/ipfs/go-log/v2"
-	"os"
+	"golang.org/x/xerrors"
 	"time"
 )
 
@@ -23,6 +27,9 @@ type WinPostTask struct {
 	lastWork *MiningBase
 
 	api    WinPostAPI
+	actors []dtypes.MinerAddress
+
+	mineTF promise.Promise[harmonytask.AddTaskFunc]
 }
 
 type WinPostAPI interface {
@@ -68,8 +75,7 @@ func (t *WinPostTask) TypeDetails() harmonytask.TaskTypeDetails {
 }
 
 func (t *WinPostTask) Adder(taskFunc harmonytask.AddTaskFunc) {
-	//TODO implement me
+	t.mineTF.Set(taskFunc)
-	panic("implement me")
 }
 
 // MiningBase is the tipset on top of which we plan to construct our next block.
@@ -77,13 +83,18 @@ func (t *WinPostTask) Adder(taskFunc harmonytask.AddTaskFunc) {
 type MiningBase struct {
 	TipSet      *types.TipSet
 	ComputeTime time.Time
-	NullRounds  abi.ChainEpoch
+	AddRounds   abi.ChainEpoch
+}
+
+func (mb MiningBase) epoch() abi.ChainEpoch {
+	// return the epoch that will result from mining on this base
+	return mb.TipSet.Height() + mb.AddRounds + 1
 }
 
 func (mb MiningBase) baseTime() time.Time {
 	tsTime := time.Unix(int64(mb.TipSet.MinTimestamp()), 0)
-	nullDelay := build.BlockDelaySecs * uint64(mb.NullRounds)
+	roundDelay := build.BlockDelaySecs * uint64(mb.AddRounds+1)
-	tsTime = tsTime.Add(time.Duration(nullDelay) * time.Second)
+	tsTime = tsTime.Add(time.Duration(roundDelay) * time.Second)
 	return tsTime
 }
 
@@ -108,6 +119,8 @@ func retry1[R any](f func() (R, error)) R {
 func (t *WinPostTask) mineBasic(ctx context.Context) {
 	var workBase MiningBase
 
+	taskFn := t.mineTF.Val(ctx)
+
 	{
 		head := retry1(func() (*types.TipSet, error) {
 			return t.api.ChainHead(ctx)
@@ -115,13 +128,26 @@ func (t *WinPostTask) mineBasic(ctx context.Context) {
 
 		workBase = MiningBase{
 			TipSet:      head,
-			NullRounds:  0,
+			AddRounds:   0,
 			ComputeTime: time.Now(),
 		}
 	}
 
+	/*
+
+		         /- T+0 == workBase.baseTime
+		         |
+		>--------*------*--------[wait until next round]----->
+		                |
+		                |- T+PD == workBase.afterPropDelay+(~1s)
+		                |- Here we acquire the new workBase, and start a new round task
+		                \- Then we loop around, and wait for the next head
+
+		time -->
+	*/
+
 	for {
-		// wait for propagation delay
+		// wait for *NEXT* propagation delay
 		time.Sleep(time.Until(workBase.afterPropDelay()))
 
 		// check current best candidate
@@ -130,7 +156,9 @@ func (t *WinPostTask) mineBasic(ctx context.Context) {
 		})
 
 		if workBase.TipSet.Equals(maybeBase) {
-			workBase.NullRounds++
+			// workbase didn't change in the new round so we have a null round here
+			workBase.AddRounds++
+			log.Debugw("workbase update", "tipset", workBase.TipSet.Cids(), "nulls", workBase.AddRounds, "lastUpdate", time.Since(workBase.ComputeTime), "type", "same-tipset")
 		} else {
 			btsw := retry1(func() (types.BigInt, error) {
 				return t.api.ChainTipSetWeight(ctx, maybeBase.Key())
@@ -141,19 +169,52 @@ func (t *WinPostTask) mineBasic(ctx context.Context) {
 			})
 
 			if types.BigCmp(btsw, ltsw) <= 0 {
-				workBase.NullRounds++
+				// new tipset for some reason has less weight than the old one, assume null round here
+				// NOTE: the backing node may have reorged, or manually changed head
+				workBase.AddRounds++
+				log.Debugw("workbase update", "tipset", workBase.TipSet.Cids(), "nulls", workBase.AddRounds, "lastUpdate", time.Since(workBase.ComputeTime), "type", "prefer-local-weight")
 			} else {
+				// new tipset has more weight, so we should mine on it, no null round here
+				log.Debugw("workbase update", "tipset", workBase.TipSet.Cids(), "nulls", workBase.AddRounds, "lastUpdate", time.Since(workBase.ComputeTime), "type", "prefer-new-tipset")
+
 				workBase = MiningBase{
 					TipSet:      maybeBase,
-					NullRounds:  0,
+					AddRounds:   0,
 					ComputeTime: time.Now(),
 				}
 			}
 		}
 
+		// dispatch mining task
+		// (note equivocation prevention is handled by the mining code)
+
+		for _, act := range t.actors {
+			spID, err := address.IDFromAddress(address.Address(act))
+			if err != nil {
+				log.Errorf("failed to get spID from address %s: %s", act, err)
+				continue
+			}
+
+			taskFn(func(id harmonytask.TaskID, tx *harmonydb.Tx) (shouldCommit bool, seriousError error) {
+				_, err := tx.Exec(`INSERT INTO mining_tasks (task_id, sp_id, epoch) VALUES ($1, $2, $3)`, id, spID, workBase.epoch())
+				if err != nil {
+					return false, xerrors.Errorf("inserting mining_tasks: %w", err)
+				}
+
+				for _, c := range workBase.TipSet.Cids() {
+					_, err = tx.Exec(`INSERT INTO mining_base_block (task_id, block_cid) VALUES ($1, $2)`, id, c)
+					if err != nil {
+						return false, xerrors.Errorf("inserting mining base blocks: %w", err)
 					}
 				}
 
+				return true, nil // no errors, commit the transaction
+			})
+		}
+	}
+}
+
+/*
 	func (t *WinPostTask) mine2(ctx context.Context) {
 		var lastBase MiningBase
 
@@ -173,7 +234,7 @@ func (t *WinPostTask) mine2(ctx context.Context) {
 				}
 
 				// 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 {
+				if base != nil && base.TipSet.Height() == prebase.TipSet.Height() && base.AddRounds == prebase.AddRounds {
 					// We have a valid base.
 					base = prebase
 					break
@@ -203,7 +264,7 @@ func (t *WinPostTask) mine2(ctx context.Context) {
 				}
 
 				// Ensure the beacon entry is available before finalizing the mining base.
-			_, err = t.api.StateGetBeaconEntry(ctx, prebase.TipSet.Height()+prebase.NullRounds+1)
+				_, err = t.api.StateGetBeaconEntry(ctx, prebase.TipSet.Height()+prebase.AddRounds+1)
 				if err != nil {
 					log.Errorf("failed getting beacon entry: %s", err)
 					time.Sleep(time.Second)
@@ -214,8 +275,8 @@ func (t *WinPostTask) mine2(ctx context.Context) {
 			}
 
 			// Check for repeated mining candidates and handle sleep for the next round.
-		if base.TipSet.Equals(lastBase.TipSet) && lastBase.NullRounds == base.NullRounds {
+			if base.TipSet.Equals(lastBase.TipSet) && lastBase.AddRounds == base.AddRounds {
-			log.Warnf("BestMiningCandidate from the previous round: %s (nulls:%d)", lastBase.TipSet.Cids(), lastBase.NullRounds)
+				log.Warnf("BestMiningCandidate from the previous round: %s (nulls:%d)", lastBase.TipSet.Cids(), lastBase.AddRounds)
 				time.Sleep(time.Duration(build.BlockDelaySecs) * time.Second)
 				continue
 			}
@@ -255,7 +316,7 @@ func (t *WinPostTask) mine2(ctx context.Context) {
 
 				// 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)
+					witness, fault, err := m.sf.MinedBlock(ctx, b.Header, base.TipSet.Height()+base.AddRounds)
 					if err != nil {
 						log.Errorf("<!!> SLASH FILTER ERRORED: %s", err)
 						// Continue here, because it's _probably_ wiser to not submit this block
@@ -274,10 +335,10 @@ func (t *WinPostTask) mine2(ctx context.Context) {
 				}
 			} else {
 				// If no block was mined, increase the null rounds and wait for the next epoch.
-			base.NullRounds++
+				base.AddRounds++
 
 				// Calculate the time for the next round.
-			nextRound := time.Unix(int64(base.TipSet.MinTimestamp()+build.BlockDelaySecs*uint64(base.NullRounds))+int64(build.PropagationDelaySecs), 0)
+				nextRound := time.Unix(int64(base.TipSet.MinTimestamp()+build.BlockDelaySecs*uint64(base.AddRounds))+int64(build.PropagationDelaySecs), 0)
 
 				// Wait for the next round.
 				time.Sleep(time.Until(nextRound))
@@ -291,6 +352,7 @@ func (t *WinPostTask) mine2(ctx context.Context) {
 // It obtains the current chain head (HEAD), and compares it to the last tipset
 // we selected as our mining base (LAST). If HEAD's weight is larger than
 // LAST's weight, it selects HEAD to build on. Else, it selects LAST.
+
 	func (t *WinPostTask) GetBestMiningCandidate(ctx context.Context) (*MiningBase, error) {
 		bts, err := t.api.ChainHead(ctx)
 		if err != nil {
@@ -320,7 +382,7 @@ func (t *WinPostTask) GetBestMiningCandidate(ctx context.Context) (*MiningBase,
 		t.lastWork = &MiningBase{TipSet: bts, ComputeTime: time.Now()}
 		return t.lastWork, nil
 	}
-
+*/
 func randTimeOffset(width time.Duration) time.Duration {
 	buf := make([]byte, 8)
 	rand.Reader.Read(buf) //nolint:errcheck