itests: MaxStagingDealsBytes integration test

.circleci/config.yml

@@ -820,6 +820,11 @@ workflows:
           suite: itest-deals_concurrent
           target: "./itests/deals_concurrent_test.go"
       
+      - test:
+          name: test-itest-deals_max_staging_deals
+          suite: itest-deals_max_staging_deals
+          target: "./itests/deals_max_staging_deals_test.go"
+      
       - test:
           name: test-itest-deals_offline
           suite: itest-deals_offline

itests/deals_max_staging_deals_test.go

@@ -0,0 +1,65 @@
+package itests
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/stretchr/testify/require"
+
+	"github.com/filecoin-project/lotus/itests/kit"
+)
+
+func TestMaxStagingDeals(t *testing.T) {
+	ctx := context.Background()
+
+	// enable 512MiB proofs so we can conduct larger transfers.
+	kit.EnableLargeSectors(t)
+	kit.QuietMiningLogs()
+
+	client, miner, ens := kit.EnsembleMinimal(t,
+		kit.MockProofs(),
+		kit.WithMaxStagingDealsBytes(8192), // max 8KB staging deals
+		kit.SectorSize(512<<20),            // 512MiB sectors.
+	)
+	ens.InterconnectAll().BeginMining(200 * time.Millisecond)
+
+	dh := kit.NewDealHarness(t, client, miner, miner)
+
+	client.WaitTillChain(ctx, kit.HeightAtLeast(5))
+
+	res, _ := client.CreateImportFile(ctx, 0, 8192) // 8KB file
+	list, err := client.ClientListImports(ctx)
+	require.NoError(t, err)
+	require.Len(t, list, 1)
+	require.Equal(t, res.Root, *list[0].Root)
+
+	res2, _ := client.CreateImportFile(ctx, 0, 4096)
+	list, err = client.ClientListImports(ctx)
+	require.NoError(t, err)
+	require.Len(t, list, 2)
+	require.Equal(t, res2.Root, *list[1].Root)
+
+	// first deal stays in staging area, and is not yet passed to the sealing subsystem
+	dp := dh.DefaultStartDealParams()
+	dp.Data.Root = res.Root
+	dp.FastRetrieval = true
+	dp.EpochPrice = abi.NewTokenAmount(62500000) // minimum asking price.
+	deal := dh.StartDeal(ctx, dp)
+
+	time.Sleep(1 * time.Second)
+
+	// expecting second deal to fail since staging area is full
+	dp.Data.Root = res2.Root
+	dp.FastRetrieval = true
+	dp.EpochPrice = abi.NewTokenAmount(62500000) // minimum asking price.
+	deal2 := dh.StartDeal(ctx, dp)
+
+	_ = deal
+
+	err = dh.ExpectDealFailure(ctx, deal2, "cannot accept deal as miner is overloaded at the moment")
+	if err != nil {
+		t.Fatal(err)
+	}
+}

itests/kit/deals.go

@@ -2,9 +2,11 @@ package kit
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -175,6 +177,49 @@ loop:
 	}
 }
 
+func (dh *DealHarness) ExpectDealFailure(ctx context.Context, deal *cid.Cid, errs string) error {
+	for {
+		di, err := dh.client.ClientGetDealInfo(ctx, *deal)
+		require.NoError(dh.t, err)
+
+		switch di.State {
+		case storagemarket.StorageDealAwaitingPreCommit, storagemarket.StorageDealSealing:
+			return fmt.Errorf("deal is sealing, and we expected an error: %s", errs)
+		case storagemarket.StorageDealProposalRejected:
+			if strings.Contains(di.Message, errs) {
+				return nil
+			}
+			return fmt.Errorf("unexpected error: %s ; expected: %s", di.Message, errs)
+		case storagemarket.StorageDealFailing:
+			if strings.Contains(di.Message, errs) {
+				return nil
+			}
+			return fmt.Errorf("unexpected error: %s ; expected: %s", di.Message, errs)
+		case storagemarket.StorageDealError:
+			if strings.Contains(di.Message, errs) {
+				return nil
+			}
+			return fmt.Errorf("unexpected error: %s ; expected: %s", di.Message, errs)
+		case storagemarket.StorageDealActive:
+			return errors.New("expected to get an error, but didn't get one")
+		}
+
+		mds, err := dh.market.MarketListIncompleteDeals(ctx)
+		require.NoError(dh.t, err)
+
+		var minerState storagemarket.StorageDealStatus
+		for _, md := range mds {
+			if md.DealID == di.DealID {
+				minerState = md.State
+				break
+			}
+		}
+
+		dh.t.Logf("Deal %d state: client:%s provider:%s\n", di.DealID, storagemarket.DealStates[di.State], storagemarket.DealStates[minerState])
+		time.Sleep(time.Second / 2)
+	}
+}
+
 // WaitDealPublished waits until the deal is published.
 func (dh *DealHarness) WaitDealPublished(ctx context.Context, deal *cid.Cid) {
 	subCtx, cancel := context.WithCancel(ctx)

itests/kit/ensemble.go

@@ -453,6 +453,7 @@ func (n *Ensemble) Start() *Ensemble {
 		cfg.Subsystems.EnableMining = m.options.subsystems.Has(SMining)
 		cfg.Subsystems.EnableSealing = m.options.subsystems.Has(SSealing)
 		cfg.Subsystems.EnableSectorStorage = m.options.subsystems.Has(SSectorStorage)
+		cfg.Dealmaking.MaxStagingDealsBytes = m.options.maxStagingDealsBytes
 
 		if m.options.mainMiner != nil {
 			token, err := m.options.mainMiner.FullNode.AuthNew(ctx, api.AllPermissions)

itests/kit/node_opts.go

@@ -27,11 +27,12 @@ type nodeOpts struct {
 	ownerKey      *wallet.Key
 	extraNodeOpts []node.Option
 
-	subsystems    MinerSubsystem
-	mainMiner     *TestMiner
-	disableLibp2p bool
-	optBuilders   []OptBuilder
-	sectorSize    abi.SectorSize
+	subsystems           MinerSubsystem
+	mainMiner            *TestMiner
+	disableLibp2p        bool
+	optBuilders          []OptBuilder
+	sectorSize           abi.SectorSize
+	maxStagingDealsBytes int64
 }
 
 // DefaultNodeOpts are the default options that will be applied to test nodes.
@@ -68,6 +69,13 @@ func WithSubsystems(systems ...MinerSubsystem) NodeOpt {
 	}
 }
 
+func WithMaxStagingDealsBytes(size int64) NodeOpt {
+	return func(opts *nodeOpts) error {
+		opts.maxStagingDealsBytes = size
+		return nil
+	}
+}
+
 func DisableLibp2p() NodeOpt {
 	return func(opts *nodeOpts) error {
 		opts.disableLibp2p = true