refactor config

cmd/curio/tasks/tasks.go

@@ -250,8 +250,12 @@ func addSealingTasks(
 
 	var sp *seal.SealPoller
 	var slr *ffi.SealCalls
+	var err error
 	if hasAnySealingTask {
-		sp = seal.NewPoller(db, full, cfg)
+		sp, err = seal.NewPoller(db, full, cfg)
+		if err != nil {
+			return nil, xerrors.Errorf("creating seal poller: %w", err)
+		}
 		go sp.RunPoller(ctx)
 
 		slr = must.One(slrLazy.Val())

deps/config/types.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
-	"github.com/filecoin-project/go-state-types/builtin/v15/miner"
 
 	"github.com/filecoin-project/lotus/chain/types"
 )
@@ -76,14 +75,14 @@ func DefaultCurioConfig() *CurioConfig {
 		},
 		Batching: CurioBatchingConfig{
 			PreCommit: PreCommitBatchingConfig{
-				MaxPreCommitBatch:   miner.PreCommitSectorBatchMaxSize,
-				PreCommitBatchSlack: Duration(6 * time.Hour),
-				BaseFeeThreshold:    types.MustParseFIL("0.005"),
+				BaseFeeThreshold: types.MustParseFIL("0.005"),
+				Timeout:          Duration(4 * time.Hour),
+				Slack:            Duration(6 * time.Hour),
 			},
 			Commit: CommitBatchingConfig{
-				MaxCommitBatch:   miner.MaxAggregatedSectors,
-				CommitBatchSlack: Duration(1 * time.Hour),
 				BaseFeeThreshold: types.MustParseFIL("0.005"),
+				Timeout:          Duration(1 * time.Hour),
+				Slack:            Duration(1 * time.Hour),
 			},
 		},
 	}
@@ -531,31 +530,23 @@ type CurioBatchingConfig struct {
 }
 
 type PreCommitBatchingConfig struct {
-	// Enable / Disable Precommit aggregation
-	AggregatePreCommits bool
+	// Base fee value below which we should try to send Precommit messages immediately
+	BaseFeeThreshold types.FIL
 
-	// Maximum precommit batch size - batches will be sent immediately above this size if BaseFeeThreshold is higher
-	// than the current base fee. If not then we will wait batch if forced due to PreCommitBatchSlack
-	MaxPreCommitBatch int
+	// Maximum amount of time any given sector in the batch can wait for the batch to accumulate
+	Timeout Duration
 
 	// Time buffer for forceful batch submission before sectors/deal in batch would start expiring
-	PreCommitBatchSlack Duration
-
-	// Base fee value below which we should try to send Precommit message. This will be ignored if PreCommitBatchSlack has reached
-	BaseFeeThreshold types.FIL
+	Slack Duration
 }
 
 type CommitBatchingConfig struct {
-	// Enable / Disable commit aggregation
-	AggregateCommits bool
+	// Base fee value below which we should try to send Commit messages immediately
+	BaseFeeThreshold types.FIL
 
-	// Maximum batched commit size - batches will be sent immediately above this size if BaseFeeThreshold is higher
-	//	// than the current base fee. If not then we will wait batch if forced due to CommitBatchSlack
-	MaxCommitBatch int
+	// Maximum amount of time any given sector in the batch can wait for the batch to accumulate
+	Timeout Duration
 
 	// Time buffer for forceful batch submission before sectors/deals in batch would start expiring
-	CommitBatchSlack Duration
-
-	// Base fee value below which we should try to send Commit message. This will be ignored if CommitBatchSlack has reached
-	BaseFeeThreshold types.FIL
+	Slack Duration
 }

deps/deps.go

@@ -227,9 +227,6 @@ func (deps *Deps) PopulateRemainingDeps(ctx context.Context, cctx *cli.Context,
 		if err != nil {
 			return xerrors.Errorf("populate config: %w", err)
 		}
-		if deps.Cfg.Batching.Commit.AggregateCommits && deps.Cfg.Batching.Commit.MaxCommitBatch < miner.MinAggregatedSectors {
-			return xerrors.Errorf("commit batch size less than minimum required for aggregation")
-		}
 	}
 
 	log.Debugw("config", "config", deps.Cfg)

documentation/en/configuration/default-curio-configuration.md

@@ -505,47 +505,35 @@ description: The default curio configuration
 
 [Batching]
   [Batching.PreCommit]
-    # Enable / Disable Precommit aggregation
+    # Base fee value below which we should try to send Precommit messages immediately
     #
-    # type: bool
-    #AggregatePreCommits = false
+    # type: types.FIL
+    #BaseFeeThreshold = "0.005 FIL"
 
-    # Maximum precommit batch size - batches will be sent immediately above this size if BaseFeeThreshold is higher
-    # than the current base fee. If not then we will wait batch if forced due to PreCommitBatchSlack
+    # Maximum amount of time any given sector in the batch can wait for the batch to accumulate
     #
-    # type: int
-    #MaxPreCommitBatch = 256
+    # type: Duration
+    #Timeout = "4h0m0s"
 
     # Time buffer for forceful batch submission before sectors/deal in batch would start expiring
     #
     # type: Duration
-    #PreCommitBatchSlack = "6h0m0s"
+    #Slack = "6h0m0s"
 
-    # Base fee value below which we should try to send Precommit message. This will be ignored if PreCommitBatchSlack has reached
+  [Batching.Commit]
+    # Base fee value below which we should try to send Commit messages immediately
     #
     # type: types.FIL
     #BaseFeeThreshold = "0.005 FIL"
 
-  [Batching.Commit]
-    # Enable / Disable commit aggregation
-    #
-    # type: bool
-    #AggregateCommits = false
-
-    # Maximum batched commit size - batches will be sent immediately above this size if BaseFeeThreshold is higher
-    # // than the current base fee. If not then we will wait batch if forced due to CommitBatchSlack
+    # Maximum amount of time any given sector in the batch can wait for the batch to accumulate
     #
-    # type: int
-    #MaxCommitBatch = 819
+    # type: Duration
+    #Timeout = "1h0m0s"
 
     # Time buffer for forceful batch submission before sectors/deals in batch would start expiring
     #
     # type: Duration
-    #CommitBatchSlack = "1h0m0s"
-
-    # Base fee value below which we should try to send Commit message. This will be ignored if CommitBatchSlack has reached
-    #
-    # type: types.FIL
-    #BaseFeeThreshold = "0.005 FIL"
+    #Slack = "1h0m0s"
 
 ```

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

@@ -36,8 +36,10 @@ create table sectors_sdr_pipeline (
     after_tree_r bool not null default false,
 
     -- synthetic proofs (Added in 20240617-synthetic-proofs.sql)
---     task_id_synth bigint,
---     after_synth bool not null default false,
+    -- task_id_synth bigint,
+    -- after_synth bool not null default false,
+
+    -- precommit_ready_at (Added in 20241210-sdr-batching.sql)
 
     -- precommit message sending
     precommit_msg_cid text,
@@ -66,6 +68,8 @@ create table sectors_sdr_pipeline (
     task_id_move_storage bigint,
     after_move_storage bool not null default false,
 
+    -- commit_ready_at (Added in 20241210-sdr-batching.sql)
+
     -- Commit message sending
     commit_msg_cid text,
 

harmony/harmonydb/sql/20241210-sdr-batching.sql

@@ -0,0 +1,2 @@
+ALTER TABLE sectors_sdr_pipeline ADD COLUMN precommit_ready_at TIMESTAMPTZ;
+ALTER TABLE sectors_sdr_pipeline ADD COLUMN commit_ready_at TIMESTAMPTZ;

itests/curio_test.go

@@ -158,6 +158,11 @@ func TestCurioHappyPath(t *testing.T) {
 
 	require.Contains(t, baseCfg.Addresses[0].MinerAddresses, maddr.String())
 
+	//baseCfg.Batching.PreCommit.BaseFeeThreshold = types.MustParseFIL("100")
+	baseCfg.Batching.PreCommit.Timeout = config.Duration(5 * time.Minute)
+	//baseCfg.Batching.Commit.BaseFeeThreshold = types.MustParseFIL("100")
+	baseCfg.Batching.Commit.Timeout = config.Duration(5 * time.Minute)
+
 	temp := os.TempDir()
 	dir, err := os.MkdirTemp(temp, "curio")
 	require.NoError(t, err)
@@ -182,22 +187,7 @@ func TestCurioHappyPath(t *testing.T) {
 	spt, err := miner2.PreferredSealProofTypeFromWindowPoStType(nv, wpt, false)
 	require.NoError(t, err)
 
-	num, err := seal.AllocateSectorNumbers(ctx, full, db, maddr, 1, 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) values ($1, $2, $3)", mid, n, spt)
-			if err != nil {
-				return false, xerrors.Errorf("inserting into sectors_sdr_pipeline: %w", err)
-			}
-		}
-		return true, nil
-	})
-	require.NoError(t, err)
-	require.Len(t, num, 1)
-
-	spt, err = miner2.PreferredSealProofTypeFromWindowPoStType(nv, wpt, true)
-	require.NoError(t, err)
-
-	num, err = seal.AllocateSectorNumbers(ctx, full, db, maddr, 1, func(tx *harmonydb.Tx, numbers []abi.SectorNumber) (bool, error) {
+	num, err := seal.AllocateSectorNumbers(ctx, full, db, maddr, 5, 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) values ($1, $2, $3)", mid, n, spt)
 			if err != nil {
@@ -207,7 +197,22 @@ func TestCurioHappyPath(t *testing.T) {
 		return true, nil
 	})
 	require.NoError(t, err)
-	require.Len(t, num, 1)
+	require.Len(t, num, 5)
+
+	//spt, err = miner2.PreferredSealProofTypeFromWindowPoStType(nv, wpt, true)
+	//require.NoError(t, err)
+	//
+	//num, err = seal.AllocateSectorNumbers(ctx, full, db, maddr, 0, 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) values ($1, $2, $3)", mid, n, spt)
+	//		if err != nil {
+	//			return false, xerrors.Errorf("inserting into sectors_sdr_pipeline: %w", err)
+	//		}
+	//	}
+	//	return true, nil
+	//})
+	//require.NoError(t, err)
+	//require.Len(t, num, 0)
 	// TODO: add DDO deal, f05 deal 2 MiB each in the sector
 
 	var sectorParamsArr []struct {
@@ -225,13 +230,13 @@ func TestCurioHappyPath(t *testing.T) {
 		FROM sectors_sdr_pipeline
 		WHERE after_commit_msg_success = True`)
 		require.NoError(t, err)
-		return len(sectorParamsArr) == 2
+		return len(sectorParamsArr) == 6
 	}, 10*time.Minute, 1*time.Second, "sector did not finish sealing in 5 minutes")
 
-	require.Equal(t, sectorParamsArr[0].SectorNumber, int64(0))
-	require.Equal(t, sectorParamsArr[0].SpID, int64(mid))
-	require.Equal(t, sectorParamsArr[1].SectorNumber, int64(1))
-	require.Equal(t, sectorParamsArr[1].SpID, int64(mid))
+	//require.Equal(t, sectorParamsArr[0].SectorNumber, int64(0))
+	//require.Equal(t, sectorParamsArr[0].SpID, int64(mid))
+	//require.Equal(t, sectorParamsArr[1].SectorNumber, int64(1))
+	//require.Equal(t, sectorParamsArr[1].SpID, int64(mid))
 
 	_ = capi.Shutdown(ctx)