combine pdp watchers for optimistic sequence create and add

Signed-off-by: Jakub Sztandera <[email protected]>

Author: Kubuxu

cmd/curio/tasks/tasks.go

@@ -293,8 +293,7 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 			es := getSenderEth()
 			sdeps.EthSender = es
 
-			pdp.NewWatcherCreate(db, must.One(dependencies.EthClient.Val()), chainSched)
-			pdp.NewWatcherPieceAdd(db, must.One(dependencies.EthClient.Val()), chainSched)
+			pdp.NewProofSetWatch(db, must.One(dependencies.EthClient.Val()), chainSched)
 
 			pdpProveTask := pdp.NewProveTask(chainSched, db, must.One(dependencies.EthClient.Val()), dependencies.Chain, es, dependencies.CachedPieceReader)
 			pdpNextProvingPeriodTask := pdp.NewNextProvingPeriodTask(db, must.One(dependencies.EthClient.Val()), dependencies.Chain, chainSched, es)
@@ -512,7 +511,7 @@ func machineDetails(deps *deps.Deps, activeTasks []harmonytask.TaskInterface, ma
 	})
 	sort.Strings(miners)
 
-	_, err := deps.DB.Exec(context.Background(), `INSERT INTO harmony_machine_details 
+	_, err := deps.DB.Exec(context.Background(), `INSERT INTO harmony_machine_details
 		(tasks, layers, startup_time, miners, machine_id, machine_name) VALUES ($1, $2, $3, $4, $5, $6)
 		ON CONFLICT (machine_id) DO UPDATE SET tasks=$1, layers=$2, startup_time=$3, miners=$4, machine_id=$5, machine_name=$6`,
 		strings.Join(taskNames, ","), strings.Join(deps.Layers, ","),

tasks/pdp/proofset_addroot_watch.go

@@ -12,10 +12,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/curio/harmony/harmonydb"
-	"github.com/filecoin-project/curio/lib/chainsched"
 	"github.com/filecoin-project/curio/pdp/contract"
-
-	chainTypes "github.com/filecoin-project/lotus/chain/types"
 )
 
 // Structures to represent database records
@@ -37,21 +34,8 @@ type PieceAddEntry struct {
 	AddMessageOK    *bool         `db:"add_message_ok"`
 }
 
-// NewWatcherPieceAdd sets up the watcher for data set piece additions
-func NewWatcherPieceAdd(db *harmonydb.DB, ethClient *ethclient.Client, pcs *chainsched.CurioChainSched) {
-	if err := pcs.AddHandler(func(ctx context.Context, revert, apply *chainTypes.TipSet) error {
-		err := processPendingDataSetPieceAdds(ctx, db, ethClient)
-		if err != nil {
-			log.Warnf("Failed to process pending data set piece adds: %v", err)
-		}
-
-		return nil
-	}); err != nil {
-		panic(err)
-	}
-}
-
 // processPendingDataSetPieceAdds processes piece additions that have been confirmed on-chain
+// it is called from proofset_watch.go
 func processPendingDataSetPieceAdds(ctx context.Context, db *harmonydb.DB, ethClient *ethclient.Client) error {
 	// Query for pdp_data_set_piece_adds entries where add_message_ok = TRUE
 	var pieceAdds []DataSetPieceAdd
@@ -131,7 +115,9 @@ func extractAndInsertPiecesFromReceipt(ctx context.Context, db *harmonydb.DB, re
 			return fmt.Errorf("failed to check if data set exists: %w", err)
 		}
 		if !exists {
-			// XXX: maybe return nil instead to avoid warning?
+			// this is a rare case where the transaction is marked as complete between create_watch being called and this function
+			// if that happens, we return an error which will get logged and ignored
+			// piece addition will get picked up in the next run of the watcher
 			return fmt.Errorf("data set %d not found in pdp_data_sets", resolvedDataSetId.Int64)
 		}
 	}

tasks/pdp/proofset_create_watch.go

@@ -12,29 +12,16 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/curio/harmony/harmonydb"
-	"github.com/filecoin-project/curio/lib/chainsched"
 	"github.com/filecoin-project/curio/pdp/contract"
-
-	chainTypes "github.com/filecoin-project/lotus/chain/types"
 )
 
 type DataSetCreate struct {
 	CreateMessageHash string `db:"create_message_hash"`
 	Service           string `db:"service"`
 }
 
-func NewWatcherCreate(db *harmonydb.DB, ethClient *ethclient.Client, pcs *chainsched.CurioChainSched) {
-	if err := pcs.AddHandler(func(ctx context.Context, revert, apply *chainTypes.TipSet) error {
-		err := processPendingDataSetCreates(ctx, db, ethClient)
-		if err != nil {
-			log.Warnf("Failed to process pending data set creates: %v", err)
-		}
-		return nil
-	}); err != nil {
-		panic(err)
-	}
-}
-
+// processPendingDataSetCreates finalises data set creation best on transactions logs
+// it is called from proofset_watch.go
 func processPendingDataSetCreates(ctx context.Context, db *harmonydb.DB, ethClient *ethclient.Client) error {
 	// Query for pdp_data_set_creates entries where ok = TRUE and data_set_created = FALSE
 	var dataSetCreates []DataSetCreate

tasks/pdp/proofset_watch.go

@@ -0,0 +1,32 @@
+package pdp
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/ethclient"
+
+	"github.com/filecoin-project/curio/harmony/harmonydb"
+	"github.com/filecoin-project/curio/lib/chainsched"
+
+	chainTypes "github.com/filecoin-project/lotus/chain/types"
+)
+
+// NewProofSetWatch runes processing steps for proofset creation and piece addtion
+// These two are run in sequence to allow for combined create-and-add flow to first
+// create the proofset, then add the pieces to it.
+func NewProofSetWatch(db *harmonydb.DB, ethClient *ethclient.Client, pcs *chainsched.CurioChainSched) {
+	if err := pcs.AddHandler(func(ctx context.Context, revert, apply *chainTypes.TipSet) error {
+		err := processPendingDataSetCreates(ctx, db, ethClient)
+		if err != nil {
+			log.Warnf("Failed to process pending data set creates: %v", err)
+		}
+
+		err = processPendingDataSetPieceAdds(ctx, db, ethClient)
+		if err != nil {
+			log.Warnf("Failed to process pending data set piece adds: %v", err)
+		}
+		return nil
+	}); err != nil {
+		panic(err)
+	}
+}