Merge remote-tracking branch 'upstream/master' into bloxico/system-test-matrix

node/builder.go

@@ -15,6 +15,7 @@ import (
 	logging "github.com/ipfs/go-log/v2"
 	ci "github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/routing"
@@ -68,6 +69,7 @@ var (
 	BandwidthReporterKey = special{11} // Libp2p option
 	ConnGaterKey         = special{12} // libp2p option
 	DAGStoreKey          = special{13} // constructor returns multiple values
+	ResourceManagerKey   = special{14} // Libp2p option
 )
 
 type invoke int
@@ -215,6 +217,10 @@ var LibP2P = Options(
 	Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
 	Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
 	Override(ConnGaterKey, lp2p.ConnGaterOption),
+
+	// Services (resource management)
+	Override(new(network.ResourceManager), lp2p.ResourceManager),
+	Override(ResourceManagerKey, lp2p.ResourceManagerOption),
 )
 
 func IsType(t repo.RepoType) func(s *Settings) bool {

node/builder_miner.go

@@ -164,7 +164,9 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(HandleRetrievalKey, modules.HandleRetrieval),
 
 			// Markets (storage)
-			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
+			Override(new(dtypes.ProviderTransferNetwork), modules.NewProviderTransferNetwork),
+			Override(new(dtypes.ProviderTransport), modules.NewProviderTransport),
+			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDataTransfer),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
 			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(cfg.Dealmaking, nil)),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),

node/config/def.go

@@ -109,10 +109,11 @@ func DefaultStorageMiner() *StorageMiner {
 			AvailableBalanceBuffer:     types.FIL(big.Zero()),
 			DisableCollateralFallback:  false,
 
-			BatchPreCommits:     true,
-			MaxPreCommitBatch:   miner5.PreCommitSectorBatchMaxSize, // up to 256 sectors
-			PreCommitBatchWait:  Duration(24 * time.Hour),           // this should be less than 31.5 hours, which is the expiration of a precommit ticket
-			PreCommitBatchSlack: Duration(3 * time.Hour),            // time buffer for forceful batch submission before sectors/deals in batch would start expiring, higher value will lower the chances for message fail due to expiration
+			BatchPreCommits:    true,
+			MaxPreCommitBatch:  miner5.PreCommitSectorBatchMaxSize, // up to 256 sectors
+			PreCommitBatchWait: Duration(24 * time.Hour),           // this should be less than 31.5 hours, which is the expiration of a precommit ticket
+			// XXX snap deals wait deals slack if first
+			PreCommitBatchSlack: Duration(3 * time.Hour), // time buffer for forceful batch submission before sectors/deals in batch would start expiring, higher value will lower the chances for message fail due to expiration
 
 			CommittedCapacitySectorLifetime: Duration(builtin.EpochDurationSeconds * uint64(policy.GetMaxSectorExpirationExtension()) * uint64(time.Second)),
 
@@ -131,11 +132,13 @@ func DefaultStorageMiner() *StorageMiner {
 		},
 
 		Storage: sectorstorage.SealerConfig{
-			AllowAddPiece:   true,
-			AllowPreCommit1: true,
-			AllowPreCommit2: true,
-			AllowCommit:     true,
-			AllowUnseal:     true,
+			AllowAddPiece:            true,
+			AllowPreCommit1:          true,
+			AllowPreCommit2:          true,
+			AllowCommit:              true,
+			AllowUnseal:              true,
+			AllowReplicaUpdate:       true,
+			AllowProveReplicaUpdate2: true,
 
 			// Default to 10 - tcp should still be able to figure this out, and
 			// it's the ratio between 10gbit / 1gbit
@@ -213,6 +216,7 @@ func DefaultStorageMiner() *StorageMiner {
 		DAGStore: DAGStoreConfig{
 			MaxConcurrentIndex:         5,
 			MaxConcurrencyStorageCalls: 100,
+			MaxConcurrentUnseals:       5,
 			GCInterval:                 Duration(1 * time.Minute),
 		},
 	}

node/config/doc_gen.go

@@ -162,6 +162,14 @@ Default value: 5.`,
 
 			Comment: `The maximum amount of unsealed deals that can be fetched simultaneously
 from the storage subsystem. 0 means unlimited.
+Default value: 0 (unlimited).`,
+		},
+		{
+			Name: "MaxConcurrentUnseals",
+			Type: "int",
+
+			Comment: `The maximum amount of unseals that can be processed simultaneously
+from the storage subsystem. 0 means unlimited.
 Default value: 0 (unlimited).`,
 		},
 		{

node/config/types.go

@@ -75,6 +75,11 @@ type DAGStoreConfig struct {
 	// Default value: 0 (unlimited).
 	MaxConcurrentReadyFetches int
 
+	// The maximum amount of unseals that can be processed simultaneously
+	// from the storage subsystem. 0 means unlimited.
+	// Default value: 0 (unlimited).
+	MaxConcurrentUnseals int
+
 	// The maximum number of simultaneous inflight API calls to the storage
 	// subsystem.
 	// Default value: 100.

node/impl/client/client.go

@@ -490,6 +490,7 @@ func (a *API) makeRetrievalQuery(ctx context.Context, rp rm.RetrievalPeer, paylo
 		Size:                    queryResponse.Size,
 		MinPrice:                queryResponse.PieceRetrievalPrice(),
 		UnsealPrice:             queryResponse.UnsealPrice,
+		PricePerByte:            queryResponse.MinPricePerByte,
 		PaymentInterval:         queryResponse.MaxPaymentInterval,
 		PaymentIntervalIncrease: queryResponse.MaxPaymentIntervalIncrease,
 		Miner:                   queryResponse.PaymentAddress, // TODO: check

node/impl/full/state.go

@@ -1365,7 +1365,7 @@ func (m *StateModule) StateDealProviderCollateralBounds(ctx context.Context, siz
 		powClaim.QualityAdjPower,
 		rewPow,
 		circ.FilCirculating,
-		m.StateManager.GetNtwkVersion(ctx, ts.Height()))
+		m.StateManager.GetNetworkVersion(ctx, ts.Height()))
 	if err != nil {
 		return api.DealCollateralBounds{}, xerrors.Errorf("getting deal provider coll bounds: %w", err)
 	}
@@ -1418,7 +1418,7 @@ func (m *StateModule) StateNetworkVersion(ctx context.Context, tsk types.TipSetK
 
 	// TODO: Height-1 to be consistent with the rest of the APIs?
 	// But that's likely going to break a bunch of stuff.
-	return m.StateManager.GetNtwkVersion(ctx, ts.Height()), nil
+	return m.StateManager.GetNetworkVersion(ctx, ts.Height()), nil
 }
 
 func (a *StateAPI) StateGetRandomnessFromTickets(ctx context.Context, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte, tsk types.TipSetKey) (abi.Randomness, error) {

node/impl/net/net.go

@@ -25,12 +25,13 @@ import (
 type NetAPI struct {
 	fx.In
 
-	RawHost   lp2p.RawHost
-	Host      host.Host
-	Router    lp2p.BaseIpfsRouting
-	ConnGater *conngater.BasicConnectionGater
-	Reporter  metrics.Reporter
-	Sk        *dtypes.ScoreKeeper
+	RawHost         lp2p.RawHost
+	Host            host.Host
+	Router          lp2p.BaseIpfsRouting
+	ConnGater       *conngater.BasicConnectionGater
+	ResourceManager network.ResourceManager
+	Reporter        metrics.Reporter
+	Sk              *dtypes.ScoreKeeper
 }
 
 func (a *NetAPI) ID(context.Context) (peer.ID, error) {

node/impl/net/rcmgr.go

@@ -0,0 +1,271 @@
+package net
+
+import (
+	"context"
+	"strings"
+
+	"golang.org/x/xerrors"
+
+	"github.com/libp2p/go-libp2p-core/network"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/protocol"
+	rcmgr "github.com/libp2p/go-libp2p-resource-manager"
+
+	"github.com/filecoin-project/lotus/api"
+)
+
+func (a *NetAPI) NetStat(ctx context.Context, scope string) (result api.NetStat, err error) {
+	switch {
+	case scope == "all":
+		rapi, ok := a.ResourceManager.(rcmgr.ResourceManagerState)
+		if !ok {
+			return result, xerrors.Errorf("rexource manager does not support ResourceManagerState API")
+		}
+
+		stat := rapi.Stat()
+		result.System = &stat.System
+		result.Transient = &stat.Transient
+		if len(stat.Services) > 0 {
+			result.Services = stat.Services
+		}
+		if len(stat.Protocols) > 0 {
+			result.Protocols = make(map[string]network.ScopeStat, len(stat.Protocols))
+			for proto, stat := range stat.Protocols {
+				result.Protocols[string(proto)] = stat
+			}
+		}
+		if len(stat.Peers) > 0 {
+			result.Peers = make(map[string]network.ScopeStat, len(stat.Peers))
+			for p, stat := range stat.Peers {
+				result.Peers[p.Pretty()] = stat
+			}
+		}
+
+		return result, nil
+
+	case scope == "system":
+		err = a.ResourceManager.ViewSystem(func(s network.ResourceScope) error {
+			stat := s.Stat()
+			result.System = &stat
+			return nil
+		})
+		return result, err
+
+	case scope == "transient":
+		err = a.ResourceManager.ViewTransient(func(s network.ResourceScope) error {
+			stat := s.Stat()
+			result.Transient = &stat
+			return nil
+		})
+		return result, err
+
+	case strings.HasPrefix(scope, "svc:"):
+		svc := scope[4:]
+		err = a.ResourceManager.ViewService(svc, func(s network.ServiceScope) error {
+			stat := s.Stat()
+			result.Services = map[string]network.ScopeStat{
+				svc: stat,
+			}
+			return nil
+		})
+		return result, err
+
+	case strings.HasPrefix(scope, "proto:"):
+		proto := scope[6:]
+		err = a.ResourceManager.ViewProtocol(protocol.ID(proto), func(s network.ProtocolScope) error {
+			stat := s.Stat()
+			result.Protocols = map[string]network.ScopeStat{
+				proto: stat,
+			}
+			return nil
+		})
+		return result, err
+
+	case strings.HasPrefix(scope, "peer:"):
+		p := scope[5:]
+		pid, err := peer.IDFromString(p)
+		if err != nil {
+			return result, xerrors.Errorf("invalid peer ID: %s: %w", p, err)
+		}
+		err = a.ResourceManager.ViewPeer(pid, func(s network.PeerScope) error {
+			stat := s.Stat()
+			result.Peers = map[string]network.ScopeStat{
+				p: stat,
+			}
+			return nil
+		})
+		return result, err
+
+	default:
+		return result, xerrors.Errorf("invalid scope %s", scope)
+	}
+}
+
+func (a *NetAPI) NetLimit(ctx context.Context, scope string) (result api.NetLimit, err error) {
+	getLimit := func(s network.ResourceScope) error {
+		limiter, ok := s.(rcmgr.ResourceScopeLimiter)
+		if !ok {
+			return xerrors.Errorf("resource scope doesn't implement ResourceScopeLimiter interface")
+		}
+
+		limit := limiter.Limit()
+		switch l := limit.(type) {
+		case *rcmgr.StaticLimit:
+			result.Memory = l.Memory
+			result.Streams = l.BaseLimit.Streams
+			result.StreamsInbound = l.BaseLimit.StreamsInbound
+			result.StreamsOutbound = l.BaseLimit.StreamsOutbound
+			result.Conns = l.BaseLimit.Conns
+			result.ConnsInbound = l.BaseLimit.ConnsInbound
+			result.ConnsOutbound = l.BaseLimit.ConnsOutbound
+			result.FD = l.BaseLimit.FD
+
+		case *rcmgr.DynamicLimit:
+			result.Dynamic = true
+			result.MemoryFraction = l.MemoryLimit.MemoryFraction
+			result.MinMemory = l.MemoryLimit.MinMemory
+			result.MaxMemory = l.MemoryLimit.MaxMemory
+			result.Streams = l.BaseLimit.Streams
+			result.StreamsInbound = l.BaseLimit.StreamsInbound
+			result.StreamsOutbound = l.BaseLimit.StreamsOutbound
+			result.Conns = l.BaseLimit.Conns
+			result.ConnsInbound = l.BaseLimit.ConnsInbound
+			result.ConnsOutbound = l.BaseLimit.ConnsOutbound
+			result.FD = l.BaseLimit.FD
+
+		default:
+			return xerrors.Errorf("unknown limit type %T", limit)
+		}
+
+		return nil
+	}
+
+	switch {
+	case scope == "system":
+		err = a.ResourceManager.ViewSystem(func(s network.ResourceScope) error {
+			return getLimit(s)
+		})
+		return result, err
+
+	case scope == "transient":
+		err = a.ResourceManager.ViewTransient(func(s network.ResourceScope) error {
+			return getLimit(s)
+		})
+		return result, err
+
+	case strings.HasPrefix(scope, "svc:"):
+		svc := scope[4:]
+		err = a.ResourceManager.ViewService(svc, func(s network.ServiceScope) error {
+			return getLimit(s)
+		})
+		return result, err
+
+	case strings.HasPrefix(scope, "proto:"):
+		proto := scope[6:]
+		err = a.ResourceManager.ViewProtocol(protocol.ID(proto), func(s network.ProtocolScope) error {
+			return getLimit(s)
+		})
+		return result, err
+
+	case strings.HasPrefix(scope, "peer:"):
+		p := scope[5:]
+		pid, err := peer.IDFromString(p)
+		if err != nil {
+			return result, xerrors.Errorf("invalid peer ID: %s: %w", p, err)
+		}
+		err = a.ResourceManager.ViewPeer(pid, func(s network.PeerScope) error {
+			return getLimit(s)
+		})
+		return result, err
+
+	default:
+		return result, xerrors.Errorf("invalid scope %s", scope)
+	}
+}
+
+func (a *NetAPI) NetSetLimit(ctx context.Context, scope string, limit api.NetLimit) error {
+	setLimit := func(s network.ResourceScope) error {
+		limiter, ok := s.(rcmgr.ResourceScopeLimiter)
+		if !ok {
+			return xerrors.Errorf("resource scope doesn't implement ResourceScopeLimiter interface")
+		}
+
+		var newLimit rcmgr.Limit
+		if limit.Dynamic {
+			newLimit = &rcmgr.DynamicLimit{
+				MemoryLimit: rcmgr.MemoryLimit{
+					MemoryFraction: limit.MemoryFraction,
+					MinMemory:      limit.MinMemory,
+					MaxMemory:      limit.MaxMemory,
+				},
+				BaseLimit: rcmgr.BaseLimit{
+					Streams:         limit.Streams,
+					StreamsInbound:  limit.StreamsInbound,
+					StreamsOutbound: limit.StreamsOutbound,
+					Conns:           limit.Conns,
+					ConnsInbound:    limit.ConnsInbound,
+					ConnsOutbound:   limit.ConnsOutbound,
+					FD:              limit.FD,
+				},
+			}
+		} else {
+			newLimit = &rcmgr.StaticLimit{
+				Memory: limit.Memory,
+				BaseLimit: rcmgr.BaseLimit{
+					Streams:         limit.Streams,
+					StreamsInbound:  limit.StreamsInbound,
+					StreamsOutbound: limit.StreamsOutbound,
+					Conns:           limit.Conns,
+					ConnsInbound:    limit.ConnsInbound,
+					ConnsOutbound:   limit.ConnsOutbound,
+					FD:              limit.FD,
+				},
+			}
+		}
+
+		limiter.SetLimit(newLimit)
+		return nil
+	}
+
+	switch {
+	case scope == "system":
+		err := a.ResourceManager.ViewSystem(func(s network.ResourceScope) error {
+			return setLimit(s)
+		})
+		return err
+
+	case scope == "transient":
+		err := a.ResourceManager.ViewTransient(func(s network.ResourceScope) error {
+			return setLimit(s)
+		})
+		return err
+
+	case strings.HasPrefix(scope, "svc:"):
+		svc := scope[4:]
+		err := a.ResourceManager.ViewService(svc, func(s network.ServiceScope) error {
+			return setLimit(s)
+		})
+		return err
+
+	case strings.HasPrefix(scope, "proto:"):
+		proto := scope[6:]
+		err := a.ResourceManager.ViewProtocol(protocol.ID(proto), func(s network.ProtocolScope) error {
+			return setLimit(s)
+		})
+		return err
+
+	case strings.HasPrefix(scope, "peer:"):
+		p := scope[5:]
+		pid, err := peer.IDFromString(p)
+		if err != nil {
+			return xerrors.Errorf("invalid peer ID: %s: %w", p, err)
+		}
+		err = a.ResourceManager.ViewPeer(pid, func(s network.PeerScope) error {
+			return setLimit(s)
+		})
+		return err
+
+	default:
+		return xerrors.Errorf("invalid scope %s", scope)
+	}
+}

node/impl/storminer.go

@@ -3,6 +3,7 @@ package impl
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -19,6 +20,9 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-graphsync"
+	gsimpl "github.com/ipfs/go-graphsync/impl"
+	"github.com/ipfs/go-graphsync/peerstate"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"go.uber.org/fx"
@@ -28,10 +32,12 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
+	gst "github.com/filecoin-project/go-data-transfer/transport/graphsync"
 	"github.com/filecoin-project/go-fil-markets/piecestore"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/network"
 
 	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
@@ -70,6 +76,8 @@ type StorageMinerAPI struct {
 	RetrievalProvider retrievalmarket.RetrievalProvider `optional:"true"`
 	SectorAccessor    retrievalmarket.SectorAccessor    `optional:"true"`
 	DataTransfer      dtypes.ProviderDataTransfer       `optional:"true"`
+	StagingGraphsync  dtypes.StagingGraphsync           `optional:"true"`
+	Transport         dtypes.ProviderTransport          `optional:"true"`
 	DealPublisher     *storageadapter.DealPublisher     `optional:"true"`
 	SectorBlocks      *sectorblocks.SectorBlocks        `optional:"true"`
 	Host              host.Host                         `optional:"true"`
@@ -379,8 +387,8 @@ func (sm *StorageMinerAPI) SectorPreCommitPending(ctx context.Context) ([]abi.Se
 	return sm.Miner.SectorPreCommitPending(ctx)
 }
 
-func (sm *StorageMinerAPI) SectorMarkForUpgrade(ctx context.Context, id abi.SectorNumber) error {
-	return sm.Miner.MarkForUpgrade(id)
+func (sm *StorageMinerAPI) SectorMarkForUpgrade(ctx context.Context, id abi.SectorNumber, snap bool) error {
+	return sm.Miner.MarkForUpgrade(ctx, id, snap)
 }
 
 func (sm *StorageMinerAPI) SectorCommitFlush(ctx context.Context) ([]sealiface.CommitBatchRes, error) {
@@ -391,6 +399,10 @@ func (sm *StorageMinerAPI) SectorCommitPending(ctx context.Context) ([]abi.Secto
 	return sm.Miner.CommitPending(ctx)
 }
 
+func (sm *StorageMinerAPI) SectorMatchPendingPiecesToOpenSectors(ctx context.Context) error {
+	return sm.Miner.SectorMatchPendingPiecesToOpenSectors(ctx)
+}
+
 func (sm *StorageMinerAPI) WorkerConnect(ctx context.Context, url string) error {
 	w, err := connectRemoteWorker(ctx, sm, url)
 	if err != nil {
@@ -553,6 +565,166 @@ func (sm *StorageMinerAPI) MarketDataTransferUpdates(ctx context.Context) (<-cha
 	return channels, nil
 }
 
+func (sm *StorageMinerAPI) MarketDataTransferDiagnostics(ctx context.Context, mpid peer.ID) (*api.TransferDiagnostics, error) {
+	gsTransport, ok := sm.Transport.(*gst.Transport)
+	if !ok {
+		return nil, errors.New("api only works for graphsync as transport")
+	}
+	graphsyncConcrete, ok := sm.StagingGraphsync.(*gsimpl.GraphSync)
+	if !ok {
+		return nil, errors.New("api only works for non-mock graphsync implementation")
+	}
+
+	inProgressChannels, err := sm.DataTransfer.InProgressChannels(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	allReceivingChannels := make(map[datatransfer.ChannelID]datatransfer.ChannelState)
+	allSendingChannels := make(map[datatransfer.ChannelID]datatransfer.ChannelState)
+	for channelID, channel := range inProgressChannels {
+		if channel.OtherPeer() != mpid {
+			continue
+		}
+		if channel.Status() == datatransfer.Completed {
+			continue
+		}
+		if channel.Status() == datatransfer.Failed || channel.Status() == datatransfer.Cancelled {
+			continue
+		}
+		if channel.SelfPeer() == channel.Sender() {
+			allSendingChannels[channelID] = channel
+		} else {
+			allReceivingChannels[channelID] = channel
+		}
+	}
+
+	// gather information about active transport channels
+	transportChannels := gsTransport.ChannelsForPeer(mpid)
+	// gather information about graphsync state for peer
+	gsPeerState := graphsyncConcrete.PeerState(mpid)
+
+	sendingTransfers := sm.generateTransfers(ctx, transportChannels.SendingChannels, gsPeerState.IncomingState, allSendingChannels)
+	receivingTransfers := sm.generateTransfers(ctx, transportChannels.ReceivingChannels, gsPeerState.OutgoingState, allReceivingChannels)
+
+	return &api.TransferDiagnostics{
+		SendingTransfers:   sendingTransfers,
+		ReceivingTransfers: receivingTransfers,
+	}, nil
+}
+
+// generate transfers matches graphsync state and data transfer state for a given peer
+// to produce detailed output on what's happening with a transfer
+func (sm *StorageMinerAPI) generateTransfers(ctx context.Context,
+	transportChannels map[datatransfer.ChannelID]gst.ChannelGraphsyncRequests,
+	gsPeerState peerstate.PeerState,
+	allChannels map[datatransfer.ChannelID]datatransfer.ChannelState) []*api.GraphSyncDataTransfer {
+	tc := &transferConverter{
+		matchedChannelIds: make(map[datatransfer.ChannelID]struct{}),
+		matchedRequests:   make(map[graphsync.RequestID]*api.GraphSyncDataTransfer),
+		gsDiagnostics:     gsPeerState.Diagnostics(),
+		requestStates:     gsPeerState.RequestStates,
+		allChannels:       allChannels,
+	}
+
+	// iterate through all operating data transfer transport channels
+	for channelID, channelRequests := range transportChannels {
+		originalState, err := sm.DataTransfer.ChannelState(ctx, channelID)
+		var baseDiagnostics []string
+		var channelState *api.DataTransferChannel
+		if err != nil {
+			baseDiagnostics = append(baseDiagnostics, fmt.Sprintf("Unable to lookup channel state: %s", err))
+		} else {
+			cs := api.NewDataTransferChannel(sm.Host.ID(), originalState)
+			channelState = &cs
+		}
+		// add the current request for this channel
+		tc.convertTransfer(channelID, true, channelState, baseDiagnostics, channelRequests.Current, true)
+		for _, requestID := range channelRequests.Previous {
+			// add any previous requests that were cancelled for a restart
+			tc.convertTransfer(channelID, true, channelState, baseDiagnostics, requestID, false)
+		}
+	}
+
+	// collect any graphsync data for channels we don't have any data transfer data for
+	tc.collectRemainingTransfers()
+
+	return tc.transfers
+}
+
+type transferConverter struct {
+	matchedChannelIds map[datatransfer.ChannelID]struct{}
+	matchedRequests   map[graphsync.RequestID]*api.GraphSyncDataTransfer
+	transfers         []*api.GraphSyncDataTransfer
+	gsDiagnostics     map[graphsync.RequestID][]string
+	requestStates     graphsync.RequestStates
+	allChannels       map[datatransfer.ChannelID]datatransfer.ChannelState
+}
+
+// convert transfer assembles transfer and diagnostic data for a given graphsync/data-transfer request
+func (tc *transferConverter) convertTransfer(channelID datatransfer.ChannelID, hasChannelID bool, channelState *api.DataTransferChannel, baseDiagnostics []string,
+	requestID graphsync.RequestID, isCurrentChannelRequest bool) {
+	diagnostics := baseDiagnostics
+	state, hasState := tc.requestStates[requestID]
+	stateString := state.String()
+	if !hasState {
+		stateString = "no graphsync state found"
+	}
+	var channelIDPtr *datatransfer.ChannelID
+	if !hasChannelID {
+		diagnostics = append(diagnostics, fmt.Sprintf("No data transfer channel id for GraphSync request ID %d", requestID))
+	} else {
+		channelIDPtr = &channelID
+		if isCurrentChannelRequest && !hasState {
+			diagnostics = append(diagnostics, fmt.Sprintf("No current request state for data transfer channel id %s", channelID))
+		} else if !isCurrentChannelRequest && hasState {
+			diagnostics = append(diagnostics, fmt.Sprintf("Graphsync request %d is a previous request on data transfer channel id %s that was restarted, but it is still running", requestID, channelID))
+		}
+	}
+	diagnostics = append(diagnostics, tc.gsDiagnostics[requestID]...)
+	transfer := &api.GraphSyncDataTransfer{
+		RequestID:               requestID,
+		RequestState:            stateString,
+		IsCurrentChannelRequest: isCurrentChannelRequest,
+		ChannelID:               channelIDPtr,
+		ChannelState:            channelState,
+		Diagnostics:             diagnostics,
+	}
+	tc.transfers = append(tc.transfers, transfer)
+	tc.matchedRequests[requestID] = transfer
+	if hasChannelID {
+		tc.matchedChannelIds[channelID] = struct{}{}
+	}
+}
+
+func (tc *transferConverter) collectRemainingTransfers() {
+	for requestID := range tc.requestStates {
+		if _, ok := tc.matchedRequests[requestID]; !ok {
+			tc.convertTransfer(datatransfer.ChannelID{}, false, nil, nil, requestID, false)
+		}
+	}
+	for requestID := range tc.gsDiagnostics {
+		if _, ok := tc.matchedRequests[requestID]; !ok {
+			tc.convertTransfer(datatransfer.ChannelID{}, false, nil, nil, requestID, false)
+		}
+	}
+	for channelID, channelState := range tc.allChannels {
+		if _, ok := tc.matchedChannelIds[channelID]; !ok {
+			channelID := channelID
+			cs := api.NewDataTransferChannel(channelState.SelfPeer(), channelState)
+			transfer := &api.GraphSyncDataTransfer{
+				RequestID:               graphsync.RequestID(-1),
+				RequestState:            "graphsync state unknown",
+				IsCurrentChannelRequest: false,
+				ChannelID:               &channelID,
+				ChannelState:            &cs,
+				Diagnostics:             []string{"data transfer with no open transport channel, cannot determine linked graphsync request"},
+			}
+			tc.transfers = append(tc.transfers, transfer)
+		}
+	}
+}
+
 func (sm *StorageMinerAPI) MarketPendingDeals(ctx context.Context) (api.PendingDealInfo, error) {
 	return sm.DealPublisher.PendingDeals(), nil
 }
@@ -988,8 +1160,8 @@ func (sm *StorageMinerAPI) Discover(ctx context.Context) (apitypes.OpenRPCDocume
 	return build.OpenRPCDiscoverJSON_Miner(), nil
 }
 
-func (sm *StorageMinerAPI) ComputeProof(ctx context.Context, ssi []builtin.SectorInfo, rand abi.PoStRandomness) ([]builtin.PoStProof, error) {
-	return sm.Epp.ComputeProof(ctx, ssi, rand)
+func (sm *StorageMinerAPI) ComputeProof(ctx context.Context, ssi []builtin.ExtendedSectorInfo, rand abi.PoStRandomness, poStEpoch abi.ChainEpoch, nv network.Version) ([]builtin.PoStProof, error) {
+	return sm.Epp.ComputeProof(ctx, ssi, rand, poStEpoch, nv)
 }
 
 func (sm *StorageMinerAPI) RuntimeSubsystems(context.Context) (res api.MinerSubsystems, err error) {

node/modules/client.go

@@ -114,11 +114,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 	net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
 
 	dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/client/transfers"))
-	transport := dtgstransport.NewTransport(h.ID(), gs, net)
-	err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
-	if err != nil && !os.IsExist(err) {
-		return nil, err
-	}
+	transport := dtgstransport.NewTransport(h.ID(), gs)
 
 	// data-transfer push / pull channel restart configuration:
 	dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
@@ -138,7 +134,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 		// After trying to restart 3 times, give up and fail the transfer
 		MaxConsecutiveRestarts: 3,
 	})
-	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
+	dt, err := dtimpl.NewDataTransfer(dtDs, net, transport, dtRestartConfig)
 	if err != nil {
 		return nil, err
 	}

node/modules/dtypes/storage.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ipfs/go-graphsync"
 	exchange "github.com/ipfs/go-ipfs-exchange-interface"
 
+	dtnet "github.com/filecoin-project/go-data-transfer/network"
 	"github.com/filecoin-project/go-fil-markets/piecestore"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 
@@ -85,6 +86,7 @@ type ProviderRequestValidator *requestvalidation.UnifiedRequestValidator
 
 // ProviderDataTransfer is a data transfer manager for the provider
 type ProviderDataTransfer datatransfer.Manager
-
+type ProviderTransferNetwork dtnet.DataTransferNetwork
+type ProviderTransport datatransfer.Transport
 type StagingBlockstore blockstore.BasicBlockstore
 type StagingGraphsync graphsync.GraphExchange

node/modules/lp2p/libp2p.go

@@ -10,10 +10,10 @@ import (
 
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p"
-	connmgr "github.com/libp2p/go-libp2p-connmgr"
 	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/peerstore"
+	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
 	"go.uber.org/fx"
 )
 

node/modules/lp2p/rcmgr.go

@@ -0,0 +1,72 @@
+package lp2p
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"go.uber.org/fx"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p-core/network"
+	rcmgr "github.com/libp2p/go-libp2p-resource-manager"
+
+	"github.com/filecoin-project/lotus/node/repo"
+)
+
+func ResourceManager(lc fx.Lifecycle, repo repo.LockedRepo) (network.ResourceManager, error) {
+	var limiter *rcmgr.BasicLimiter
+	var opts []rcmgr.Option
+
+	repoPath := repo.Path()
+
+	// create limiter -- parse $repo/limits.json if exists
+	limitsFile := filepath.Join(repoPath, "limits.json")
+	limitsIn, err := os.Open(limitsFile)
+	switch {
+	case err == nil:
+		defer limitsIn.Close() //nolint:errcheck
+		limiter, err = rcmgr.NewDefaultLimiterFromJSON(limitsIn)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing limit file: %w", err)
+		}
+
+	case errors.Is(err, os.ErrNotExist):
+		limiter = rcmgr.NewDefaultLimiter()
+
+	default:
+		return nil, err
+	}
+
+	// TODO: also set appropriate default limits for lotus protocols
+	libp2p.SetDefaultServiceLimits(limiter)
+
+	if os.Getenv("LOTUS_DEBUG_RCMGR") != "" {
+		debugPath := filepath.Join(repoPath, "debug")
+		if err := os.MkdirAll(debugPath, 0755); err != nil {
+			return nil, fmt.Errorf("error creating debug directory: %w", err)
+		}
+		traceFile := filepath.Join(debugPath, "rcmgr.json.gz")
+		opts = append(opts, rcmgr.WithTrace(traceFile))
+	}
+
+	mgr, err := rcmgr.NewResourceManager(limiter, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("error creating resource manager: %w", err)
+	}
+
+	lc.Append(fx.Hook{
+		OnStop: func(_ context.Context) error {
+			return mgr.Close()
+		}})
+
+	return mgr, nil
+}
+
+func ResourceManagerOption(mgr network.ResourceManager) Libp2pOpts {
+	return Libp2pOpts{
+		Opts: []libp2p.Option{libp2p.ResourceManager(mgr)},
+	}
+}

node/modules/storageminer.go

@@ -335,19 +335,21 @@ func HandleMigrateProviderFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, node api.
 	})
 }
 
-// NewProviderDAGServiceDataTransfer returns a data transfer manager that just
-// uses the provider's Staging DAG service for transfers
-func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.StagingGraphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ProviderDataTransfer, error) {
-	net := dtnet.NewFromLibp2pHost(h)
+// NewProviderTransferNetwork sets up the libp2p2 protocol networking for data transfer
+func NewProviderTransferNetwork(h host.Host) dtypes.ProviderTransferNetwork {
+	return dtnet.NewFromLibp2pHost(h)
+}
 
+// NewProviderTransport sets up a data transfer transport over graphsync
+func NewProviderTransport(h host.Host, gs dtypes.StagingGraphsync) dtypes.ProviderTransport {
+	return dtgstransport.NewTransport(h.ID(), gs)
+}
+
+// NewProviderDataTransfer returns a data transfer manager
+func NewProviderDataTransfer(lc fx.Lifecycle, net dtypes.ProviderTransferNetwork, transport dtypes.ProviderTransport, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ProviderDataTransfer, error) {
 	dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/provider/transfers"))
-	transport := dtgstransport.NewTransport(h.ID(), gs, net)
-	err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
-	if err != nil && !os.IsExist(err) {
-		return nil, err
-	}
 
-	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport)
+	dt, err := dtimpl.NewDataTransfer(dtDs, net, transport)
 	if err != nil {
 		return nil, err
 	}

node/modules/storageminer_dagstore.go

@@ -38,7 +38,7 @@ func NewMinerAPI(lc fx.Lifecycle, r repo.LockedRepo, pieceStore dtypes.ProviderP
 		}
 	}
 
-	mountApi := mdagstore.NewMinerAPI(pieceStore, sa, cfg.MaxConcurrencyStorageCalls)
+	mountApi := mdagstore.NewMinerAPI(pieceStore, sa, cfg.MaxConcurrencyStorageCalls, cfg.MaxConcurrentUnseals)
 	ready := make(chan error, 1)
 	pieceStore.OnReady(func(err error) {
 		ready <- err