introduce MaxStagingDealsGiB - reject new deals if our staging deals area is "full"

node/builder_miner.go

@@ -165,7 +165,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			// Markets (storage)
 			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
-			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(nil)),
+			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(cfg.Dealmaking, nil)),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),
 			Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
 			Override(HandleMigrateProviderFundsKey, modules.HandleMigrateProviderFunds),
@@ -192,7 +192,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(dtypes.GetMaxDealStartDelayFunc), modules.NewGetMaxDealStartDelayFunc),
 
 			If(cfg.Dealmaking.Filter != "",
-				Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(dealfilter.CliStorageDealFilter(cfg.Dealmaking.Filter))),
+				Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(cfg.Dealmaking, dealfilter.CliStorageDealFilter(cfg.Dealmaking.Filter))),
 			),
 
 			If(cfg.Dealmaking.RetrievalFilter != "",

node/config/doc_gen.go

@@ -251,6 +251,13 @@ message`,
 
 			Comment: `The maximum collateral that the provider will put up against a deal,
 as a multiplier of the minimum collateral bound`,
+		},
+		{
+			Name: "MaxStagingDealsGiB",
+			Type: "int64",
+
+			Comment: `The maximum size of staging deals not yet passed to the sealing node,
+that the markets service can accept`,
 		},
 		{
 			Name: "SimultaneousTransfers",

node/config/types.go

@@ -126,7 +126,9 @@ type DealmakingConfig struct {
 	// The maximum collateral that the provider will put up against a deal,
 	// as a multiplier of the minimum collateral bound
 	MaxProviderCollateralMultiplier uint64
-
+	// The maximum size of staging deals not yet passed to the sealing node,
+	// that the markets service can accept
+	MaxStagingDealsGiB int64
 	// The maximum number of parallel online data transfers (storage+retrieval)
 	SimultaneousTransfers uint64
 

node/modules/storageminer.go

@@ -442,14 +442,16 @@ func NewStorageAsk(ctx helpers.MetricsCtx, fapi v1api.FullNode, ds dtypes.Metada
 		storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize)))
 }
 
-func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
+func BasicDealFilter(cfg config.DealmakingConfig, user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 	offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 	verifiedOk dtypes.ConsiderVerifiedStorageDealsConfigFunc,
 	unverifiedOk dtypes.ConsiderUnverifiedStorageDealsConfigFunc,
 	blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
 	expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
 	startDelay dtypes.GetMaxDealStartDelayFunc,
-	spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
+	spn storagemarket.StorageProviderNode,
+	r repo.LockedRepo,
+) dtypes.StorageDealFilter {
 	return func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 		offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 		verifiedOk dtypes.ConsiderVerifiedStorageDealsConfigFunc,
@@ -457,7 +459,9 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 		blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
 		expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
 		startDelay dtypes.GetMaxDealStartDelayFunc,
-		spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
+		spn storagemarket.StorageProviderNode,
+		r repo.LockedRepo,
+	) dtypes.StorageDealFilter {
 
 		return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
 			b, err := onlineOk()
@@ -533,6 +537,17 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 				return false, "miner error", err
 			}
 
+			diskUsageBytes, err := r.DiskUsage(r.Path() + "/deal-staging")
+			if err != nil {
+				return false, "miner error", err
+			}
+
+			diskUsageGiB := diskUsageBytes / 1024 / 1024 / 1024
+			if cfg.MaxStagingDealsGiB != 0 && diskUsageGiB >= cfg.MaxStagingDealsGiB {
+				log.Errorw("proposed deal rejected because there are too many deals in the staging area at the moment", "MaxStagingDealsGiB", cfg.MaxStagingDealsGiB, "DiskUsageGiB", diskUsageGiB)
+				return false, "cannot accept deal as miner is overloaded at the moment - there are too many staging deals being processed", nil
+			}
+
 			// Reject if it's more than 7 days in the future
 			// TODO: read from cfg
 			maxStartEpoch := earliest + abi.ChainEpoch(uint64(sd.Seconds())/build.BlockDelaySecs)