sealing pipeline: Fix failing ProveCommit3 aggregate (#11710)

* itests: Repro failing ProveCommit3 aggregate

* commit batch: Correctly sort sectors in processBatchV2

* fix imports

* ci: Bigger instance for sector_pledge test

* itests: Use Must-Post mining in TestPledgeBatching

.circleci/config.yml

@@ -933,6 +933,7 @@ workflows:
             - build
           suite: itest-sector_pledge
           target: "./itests/sector_pledge_test.go"
+          resource_class: 2xlarge
           get-params: true
 
       - test:

.circleci/template.yml

@@ -545,7 +545,7 @@ workflows:
             - build
           suite: itest-[[ $name ]]
           target: "./itests/[[ $file ]]"
-          [[- if or (eq $name "worker") (eq $name "deals_concurrent") (eq $name "wdpost_worker_config")]]
+          [[- if or (eq $name "worker") (eq $name "deals_concurrent") (eq $name "wdpost_worker_config") (eq $name "sector_pledge")]]
           resource_class: 2xlarge
           [[- end]]
           [[- if or (eq $name "wdpost") (eq $name "sector_pledge")]]

itests/kit/ensemble_opts.go

@@ -35,7 +35,13 @@ var DefaultEnsembleOpts = ensembleOpts{
 }
 
 // MockProofs activates mock proofs for the entire ensemble.
-func MockProofs() EnsembleOpt {
+func MockProofs(e ...bool) EnsembleOpt {
+	if len(e) > 0 && !e[0] {
+		return func(opts *ensembleOpts) error {
+			return nil
+		}
+	}
+
 	return func(opts *ensembleOpts) error {
 		opts.mockProofs = true
 		// since we're using mock proofs, we don't need to download

itests/sector_pledge_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/big"
 	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
 
 	"github.com/filecoin-project/lotus/api"
@@ -39,7 +40,7 @@ func TestPledgeSectors(t *testing.T) {
 		defer cancel()
 
 		_, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs())
-		ens.InterconnectAll().BeginMining(blockTime)
+		ens.InterconnectAll().BeginMiningMustPost(blockTime)
 
 		miner.PledgeSectors(ctx, nSectors, 0, nil)
 	}
@@ -65,12 +66,18 @@ func TestPledgeBatching(t *testing.T) {
 	//stm: @SECTOR_PRE_COMMIT_FLUSH_001, @SECTOR_COMMIT_FLUSH_001
 	blockTime := 50 * time.Millisecond
 
-	runTest := func(t *testing.T, nSectors int) {
+	runTest := func(t *testing.T, nSectors int, aggregate bool) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
-		client, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs())
-		ens.InterconnectAll().BeginMining(blockTime)
+		kit.QuietMiningLogs()
+
+		client, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs(!aggregate), kit.MutateSealingConfig(func(sc *config.SealingConfig) {
+			if aggregate {
+				sc.AggregateAboveBaseFee = types.FIL(big.Zero())
+			}
+		}))
+		ens.InterconnectAll().BeginMiningMustPost(blockTime)
 
 		client.WaitTillChain(ctx, kit.HeightAtLeast(10))
 
@@ -114,7 +121,10 @@ func TestPledgeBatching(t *testing.T) {
 	}
 
 	t.Run("100", func(t *testing.T) {
-		runTest(t, 100)
+		runTest(t, 100, false)
+	})
+	t.Run("10-agg", func(t *testing.T) {
+		runTest(t, 10, true)
 	})
 }
 

storage/pipeline/commit_batch.go

@@ -331,6 +331,9 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 		return nil, err
 	}
 
+	// sort sectors by number
+	sort.Slice(sectors, func(i, j int) bool { return sectors[i] < sectors[j] })
+
 	total := len(sectors)
 
 	res := sealiface.CommitBatchRes{
@@ -371,10 +374,6 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 		return nil, nil
 	}
 
-	sort.Slice(infos, func(i, j int) bool {
-		return infos[i].Number < infos[j].Number
-	})
-
 	proofs := make([][]byte, 0, total)
 	for _, info := range infos {
 		proofs = append(proofs, b.todo[info.Number].Proof)
@@ -450,7 +449,7 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 	_, err = simulateMsgGas(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.ProveCommitSectors3, needFunds, maxFee, enc.Bytes())
 
 	if err != nil && (!api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) || len(sectors) < miner.MinAggregatedSectors*2) {
-		log.Errorf("simulating CommitBatch message failed: %s", err)
+		log.Errorf("simulating CommitBatch message failed (%x): %s", enc.Bytes(), err)
 		res.Error = err.Error()
 		return []sealiface.CommitBatchRes{res}, xerrors.Errorf("simulating CommitBatch message failed: %w", err)
 	}
@@ -474,7 +473,7 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 
 	res.Msg = &mcid
 
-	log.Infow("Sent ProveCommitSectors2 message", "cid", mcid, "from", from, "todo", total, "sectors", len(infos))
+	log.Infow("Sent ProveCommitSectors3 message", "cid", mcid, "from", from, "todo", total, "sectors", len(infos))
 
 	return []sealiface.CommitBatchRes{res}, nil
 }
@@ -591,7 +590,7 @@ func (b *CommitBatcher) processBatchV1(cfg sealiface.Config, sectors []abi.Secto
 	_, err = simulateMsgGas(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.ProveCommitAggregate, needFunds, maxFee, enc.Bytes())
 
 	if err != nil && (!api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) || len(sectors) < miner.MinAggregatedSectors*2) {
-		log.Errorf("simulating CommitBatch message failed: %s", err)
+		log.Errorf("simulating CommitBatch message failed (%x): %s", enc.Bytes(), err)
 		res.Error = err.Error()
 		return []sealiface.CommitBatchRes{res}, xerrors.Errorf("simulating CommitBatch message failed: %w", err)
 	}