Merge pull request #7427 from filecoin-project/rvagg/traversal-budget

Limit maximum number of DAG links to traverse

node/config/def.go

@@ -2,6 +2,8 @@ package config
 
 import (
 	"encoding"
+	"os"
+	"strconv"
 	"time"
 
 	"github.com/ipfs/go-cid"
@@ -24,6 +26,16 @@ const (
 	RetrievalPricingExternalMode = "external"
 )
 
+// MaxTraversalLinks configures the maximum number of links to traverse in a DAG while calculating
+// CommP and traversing a DAG with graphsync; invokes a budget on DAG depth and density.
+var MaxTraversalLinks uint64 = 32 * (1 << 20)
+
+func init() {
+	if envMaxTraversal, err := strconv.ParseUint(os.Getenv("LOTUS_MAX_TRAVERSAL_LINKS"), 10, 64); err == nil {
+		MaxTraversalLinks = envMaxTraversal
+	}
+}
+
 func (b *BatchFeeConfig) FeeForSectors(nSectors int) abi.TokenAmount {
 	return big.Add(big.Int(b.Base), big.Mul(big.NewInt(int64(nSectors)), big.Int(b.PerSector)))
 }

node/impl/client/client.go

@@ -54,6 +54,7 @@ import (
 	"github.com/filecoin-project/specs-actors/v3/actors/builtin/market"
 
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
+	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/repo/imports"
 
 	"github.com/filecoin-project/lotus/api"
@@ -981,6 +982,7 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 				Root:     order.Root,
 				Selector: sel,
 			}},
+			car.MaxTraversalLinks(config.MaxTraversalLinks),
 		).Write(f)
 		if err != nil {
 			finish(err)
@@ -1232,7 +1234,11 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
 	allSelector := ssb.ExploreRecursive(
 		selector.RecursionLimitNone(),
 		ssb.ExploreAll(ssb.ExploreRecursiveEdge())).Node()
-	sc := car.NewSelectiveCar(ctx, fs, []car.Dag{{Root: root, Selector: allSelector}})
+	sc := car.NewSelectiveCar(ctx,
+		fs,
+		[]car.Dag{{Root: root, Selector: allSelector}},
+		car.MaxTraversalLinks(config.MaxTraversalLinks),
+	)
 	f, err := os.Create(outputPath)
 	if err != nil {
 		return err

node/modules/client.go

@@ -36,6 +36,7 @@ import (
 	"github.com/filecoin-project/lotus/markets"
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
 	"github.com/filecoin-project/lotus/markets/retrievaladapter"
+	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/impl/full"
 	payapi "github.com/filecoin-project/lotus/node/impl/paych"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -182,7 +183,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, dataTransfer dtypes.ClientDataT
 	marketsRetryParams := smnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
 	net := smnet.NewFromLibp2pHost(h, marketsRetryParams)
 
-	c, err := storageimpl.NewClient(net, dataTransfer, discovery, deals, scn, accessor, storageimpl.DealPollingInterval(time.Second))
+	c, err := storageimpl.NewClient(net, dataTransfer, discovery, deals, scn, accessor, storageimpl.DealPollingInterval(time.Second), storageimpl.MaxTraversalLinks(config.MaxTraversalLinks))
 	if err != nil {
 		return nil, err
 	}

node/modules/graphsync.go

@@ -9,6 +9,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 	"go.uber.org/fx"
 
+	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
 	"github.com/filecoin-project/lotus/node/repo"
@@ -20,7 +21,14 @@ func Graphsync(parallelTransfersForStorage uint64, parallelTransfersForRetrieval
 		graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
 		lsys := storeutil.LinkSystemForBlockstore(clientBs)
 
-		gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, lsys, graphsyncimpl.RejectAllRequestsByDefault(), graphsyncimpl.MaxInProgressIncomingRequests(parallelTransfersForStorage), graphsyncimpl.MaxInProgressOutgoingRequests(parallelTransfersForRetrieval))
+		gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc),
+			graphsyncNetwork,
+			lsys,
+			graphsyncimpl.RejectAllRequestsByDefault(),
+			graphsyncimpl.MaxInProgressIncomingRequests(parallelTransfersForStorage),
+			graphsyncimpl.MaxInProgressOutgoingRequests(parallelTransfersForRetrieval),
+			graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
+			graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
 		chainLinkSystem := storeutil.LinkSystemForBlockstore(chainBs)
 		err := gs.RegisterPersistenceOption("chainstore", chainLinkSystem)
 		if err != nil {

node/modules/storageminer.go

@@ -38,6 +38,7 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	graphsync "github.com/ipfs/go-graphsync/impl"
+	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
 	gsnet "github.com/ipfs/go-graphsync/network"
 	"github.com/ipfs/go-graphsync/storeutil"
 	"github.com/libp2p/go-libp2p-core/host"
@@ -398,7 +399,14 @@ func StagingGraphsync(parallelTransfersForStorage uint64, parallelTransfersForRe
 	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
 		graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
 		lsys := storeutil.LinkSystemForBlockstore(ibs)
-		gs := graphsync.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, lsys, graphsync.RejectAllRequestsByDefault(), graphsync.MaxInProgressOutgoingRequests(parallelTransfersForStorage), graphsync.MaxInProgressIncomingRequests(parallelTransfersForRetrieval))
+		gs := graphsync.New(helpers.LifecycleCtx(mctx, lc),
+			graphsyncNetwork,
+			lsys,
+			graphsync.RejectAllRequestsByDefault(),
+			graphsync.MaxInProgressIncomingRequests(parallelTransfersForRetrieval),
+			graphsync.MaxInProgressOutgoingRequests(parallelTransfersForStorage),
+			graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
+			graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
 
 		return gs
 	}