feat: batch publish deal messages

node/builder.go

@@ -377,6 +377,7 @@ func Online() Option {
 			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(nil)),
 			Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(nil)),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),
+			Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, nil)),
 			Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(nil)),
 			Override(HandleMigrateProviderFundsKey, modules.HandleMigrateProviderFunds),
 			Override(HandleRetrievalKey, modules.HandleRetrieval),
@@ -519,6 +520,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(dealfilter.CliRetrievalDealFilter(cfg.Dealmaking.RetrievalFilter))),
 		),
 
+		Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(&cfg.Fees, &cfg.PublishMsg)),
 		Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees)),
 
 		Override(new(sectorstorage.SealerConfig), cfg.Storage),
@@ -646,5 +648,6 @@ func Test() Option {
 		Unset(RunPeerMgrKey),
 		Unset(new(*peermgr.PeerMgr)),
 		Override(new(beacon.Schedule), testing.RandomBeacon),
+		Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, nil)),
 	)
 }

node/config/def.go

@@ -33,6 +33,7 @@ type StorageMiner struct {
 	Common
 
 	Dealmaking DealmakingConfig
+	PublishMsg PublishMsgConfig
 	Sealing    SealingConfig
 	Storage    sectorstorage.SealerConfig
 	Fees       MinerFeeConfig
@@ -53,6 +54,15 @@ type DealmakingConfig struct {
 	RetrievalFilter string
 }
 
+type PublishMsgConfig struct {
+	// The amount of time to wait for more deals to arrive before
+	// publishing
+	PublishPeriod Duration
+	// The maximum number of deals to include in a single PublishStorageDeals
+	// message
+	MaxDealsPerMsg uint64
+}
+
 type SealingConfig struct {
 	// 0 = no limit
 	MaxWaitDealsSectors uint64
@@ -209,6 +219,11 @@ func DefaultStorageMiner() *StorageMiner {
 			ExpectedSealDuration: Duration(time.Hour * 24),
 		},
 
+		PublishMsg: PublishMsgConfig{
+			PublishPeriod:  Duration(time.Hour),
+			MaxDealsPerMsg: 8,
+		},
+
 		Fees: MinerFeeConfig{
 			MaxPreCommitGasFee:     types.MustParseFIL("0.025"),
 			MaxCommitGasFee:        types.MustParseFIL("0.05"),

node/impl/full/state.go

@@ -54,8 +54,8 @@ type StateModuleAPI interface {
 	StateMinerProvingDeadline(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*dline.Info, error)
 	StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
 	StateNetworkVersion(ctx context.Context, key types.TipSetKey) (network.Version, error)
-	StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
 	StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error)
+	StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
 	StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error)
 	StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error)
 }

node/node_test.go

@@ -5,14 +5,12 @@ import (
 	"testing"
 	"time"
 
-	builder "github.com/filecoin-project/lotus/node/test"
-
 	"github.com/filecoin-project/go-state-types/abi"
-	"github.com/filecoin-project/lotus/lib/lotuslog"
-	logging "github.com/ipfs/go-log/v2"
-
 	"github.com/filecoin-project/lotus/api/test"
 	"github.com/filecoin-project/lotus/chain/actors/policy"
+	"github.com/filecoin-project/lotus/lib/lotuslog"
+	builder "github.com/filecoin-project/lotus/node/test"
+	logging "github.com/ipfs/go-log/v2"
 )
 
 func init() {
@@ -57,6 +55,9 @@ func TestAPIDealFlow(t *testing.T) {
 	t.Run("TestFastRetrievalDealFlow", func(t *testing.T) {
 		test.TestFastRetrievalDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
 	})
+	t.Run("TestPublishDealsBatching", func(t *testing.T) {
+		test.TestPublishDealsBatching(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
+	})
 }
 
 func TestAPIDealFlowReal(t *testing.T) {

node/test/builder.go

@@ -288,7 +288,11 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
 		genMiner := maddrs[i]
 		wa := genms[i].Worker
 
-		storers[i] = CreateTestStorageNode(ctx, t, wa, genMiner, pk, f, mn, node.Options())
+		opts := def.Opts
+		if opts == nil {
+			opts = node.Options()
+		}
+		storers[i] = CreateTestStorageNode(ctx, t, wa, genMiner, pk, f, mn, opts)
 		if err := storers[i].StorageAddLocal(ctx, presealDirs[i]); err != nil {
 			t.Fatalf("%+v", err)
 		}
@@ -455,12 +459,17 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 			}
 		}
 
+		opts := def.Opts
+		if opts == nil {
+			opts = node.Options()
+		}
 		storers[i] = CreateTestStorageNode(ctx, t, genms[i].Worker, maddrs[i], pidKeys[i], f, mn, node.Options(
 			node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
 				return mock.NewMockSectorMgr(sectors), nil
 			}),
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
 			node.Unset(new(*sectorstorage.Manager)),
+			opts,
 		))
 
 		if rpc {