batch deal input processing test

api/test/deals.go

@@ -23,9 +23,11 @@ import (
 	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
 	"github.com/filecoin-project/lotus/chain/types"
 	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
+	"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
 	"github.com/filecoin-project/lotus/markets/storageadapter"
 	"github.com/filecoin-project/lotus/node"
 	"github.com/filecoin-project/lotus/node/impl"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
 	ipld "github.com/ipfs/go-ipld-format"
 	dag "github.com/ipfs/go-merkledag"
@@ -183,6 +185,71 @@ func TestPublishDealsBatching(t *testing.T, b APIBuilder, blocktime time.Duratio
 	}
 }
 
+func TestBatchDealInput(t *testing.T, b APIBuilder, blocktime time.Duration, startEpoch abi.ChainEpoch) {
+	publishPeriod := 10 * time.Second
+	maxDealsPerMsg := uint64(4)
+
+	// Set max deals per publish deals message to maxDealsPerMsg
+	minerDef := []StorageMiner{{
+		Full: 0,
+		Opts: node.Options(
+			node.Override(
+				new(*storageadapter.DealPublisher),
+				storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{
+					Period:         publishPeriod,
+					MaxDealsPerMsg: maxDealsPerMsg,
+				})),
+			node.Override(new(dtypes.GetSealingConfigFunc), func() (dtypes.GetSealingConfigFunc, error) {
+				return func() (sealiface.Config, error) {
+					return sealiface.Config{
+						MaxWaitDealsSectors:       1,
+						MaxSealingSectors:         2,
+						MaxSealingSectorsForDeals: 3,
+						AlwaysKeepUnsealedCopy:    true,
+					}, nil
+				}, nil
+			}),
+		),
+		Preseal: PresealGenesis,
+	}}
+
+	// Create a connect client and miner node
+	n, sn := b(t, OneFull, minerDef)
+	client := n[0].FullNode.(*impl.FullNodeAPI)
+	miner := sn[0]
+	s := connectAndStartMining(t, b, blocktime, client, miner)
+	defer s.blockMiner.Stop()
+
+	// Starts a deal and waits until it's published
+	runDealTillSeal := func(rseed int) {
+		res, _, err := CreateClientFile(s.ctx, s.client, rseed)
+		require.NoError(t, err)
+
+		dc := startDeal(t, s.ctx, s.miner, s.client, res.Root, false, startEpoch)
+		waitDealSealed(t, s.ctx, s.miner, s.client, dc, false)
+	}
+
+	// Run maxDealsPerMsg+1 deals in parallel
+	done := make(chan struct{}, maxDealsPerMsg+1)
+	for rseed := 1; rseed <= int(maxDealsPerMsg+1); rseed++ {
+		rseed := rseed
+		go func() {
+			runDealTillSeal(rseed)
+			done <- struct{}{}
+		}()
+	}
+
+	// Wait for maxDealsPerMsg of the deals to be published
+	for i := 0; i < int(maxDealsPerMsg); i++ {
+		<-done
+	}
+
+	sl, err := sn[0].SectorsList(s.ctx)
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, len(sl), 4)
+	require.LessOrEqual(t, len(sl), 5)
+}
+
 func TestFastRetrievalDealFlow(t *testing.T, b APIBuilder, blocktime time.Duration, startEpoch abi.ChainEpoch) {
 	s := setupOneClientOneMiner(t, b, blocktime)
 	defer s.blockMiner.Stop()

node/node_test.go

@@ -58,6 +58,9 @@ func TestAPIDealFlow(t *testing.T) {
 	t.Run("TestPublishDealsBatching", func(t *testing.T) {
 		test.TestPublishDealsBatching(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
 	})
+	t.Run("TestBatchDealInput", func(t *testing.T) {
+		test.TestBatchDealInput(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
+	})
 }
 
 func TestAPIDealFlowReal(t *testing.T) {