Merge remote-tracking branch 'origin/master' into feat/v1-api

node/modules/client.go

@@ -7,12 +7,10 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/filecoin-project/go-multistore"
-	"github.com/filecoin-project/go-state-types/abi"
+	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
-	"go.uber.org/fx"
-
+	"github.com/filecoin-project/go-data-transfer/channelmonitor"
 	dtimpl "github.com/filecoin-project/go-data-transfer/impl"
 	dtnet "github.com/filecoin-project/go-data-transfer/network"
 	dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
@@ -25,6 +23,8 @@ import (
 	storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 	smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
+	"github.com/filecoin-project/go-multistore"
+	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-storedcounter"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
@@ -121,8 +121,6 @@ func RegisterClientValidator(crv dtypes.ClientRequestValidator, dtm dtypes.Clien
 // NewClientGraphsyncDataTransfer returns a data transfer manager that just
 // uses the clients's Client DAG service for transfers
 func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Graphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ClientDataTransfer, error) {
-	sc := storedcounter.New(ds, datastore.NewKey("/datatransfer/client/counter"))
-
 	// go-data-transfer protocol retries:
 	// 1s, 5s, 25s, 2m5s, 5m x 11 ~= 1 hour
 	dtRetryParams := dtnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
@@ -135,9 +133,30 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 		return nil, err
 	}
 
-	// data-transfer push channel restart configuration
-	dtRestartConfig := dtimpl.PushChannelRestartConfig(time.Minute, 10, 1024, 10*time.Minute, 3)
-	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc, dtRestartConfig)
+	// data-transfer push / pull channel restart configuration:
+	dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
+		// For now only monitor push channels (for storage deals)
+		MonitorPushChannels: true,
+		// TODO: Enable pull channel monitoring (for retrievals) when the
+		//  following issue has been fixed:
+		// https://github.com/filecoin-project/go-data-transfer/issues/172
+		MonitorPullChannels: false,
+		// Wait up to 30s for the other side to respond to an Open channel message
+		AcceptTimeout: 30 * time.Second,
+		// Send a restart message if the data rate falls below 1024 bytes / minute
+		Interval:            time.Minute,
+		MinBytesTransferred: 1024,
+		// Perform check 10 times / minute
+		ChecksPerInterval: 10,
+		// After sending a restart, wait for at least 1 minute before sending another
+		RestartBackoff: time.Minute,
+		// After trying to restart 3 times, give up and fail the transfer
+		MaxConsecutiveRestarts: 3,
+		// Wait up to 30s for the other side to send a Complete message once all
+		// data has been sent / received
+		CompleteTimeout: 30 * time.Second,
+	})
+	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
 	if err != nil {
 		return nil, err
 	}

node/modules/storageminer.go

@@ -330,7 +330,6 @@ 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) {
-	sc := storedcounter.New(ds, datastore.NewKey("/datatransfer/provider/counter"))
 	net := dtnet.NewFromLibp2pHost(h)
 
 	dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/provider/transfers"))
@@ -340,7 +339,7 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
 		return nil, err
 	}
 
-	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc)
+	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport)
 	if err != nil {
 		return nil, err
 	}