resolve conflicts

api/api_full.go

@@ -340,6 +340,9 @@ type FullNode interface {
 	// which are stuck due to insufficient funds
 	ClientRetrieveTryRestartInsufficientFunds(ctx context.Context, paymentChannel address.Address) error //perm:write
 
+	// ClientCancelRetrievalDeal cancels an ongoing retrieval deal based on DealID
+	ClientCancelRetrievalDeal(ctx context.Context, dealid retrievalmarket.DealID) error //perm:write
+
 	// ClientUnimport removes references to the specified file from filestore
 	//ClientUnimport(path string)
 

api/apistruct/struct.go

@@ -155,6 +155,8 @@ type FullNodeStruct struct {
 
 		ClientCancelDataTransfer func(p0 context.Context, p1 datatransfer.TransferID, p2 peer.ID, p3 bool) error `perm:"write"`
 
+		ClientCancelRetrievalDeal func(p0 context.Context, p1 retrievalmarket.DealID) error `perm:"write"`
+
 		ClientDataTransferUpdates func(p0 context.Context) (<-chan api.DataTransferChannel, error) `perm:"write"`
 
 		ClientDealPieceCID func(p0 context.Context, p1 cid.Cid) (api.DataCIDSize, error) `perm:"read"`
@@ -981,6 +983,10 @@ func (s *FullNodeStruct) ClientCancelDataTransfer(p0 context.Context, p1 datatra
 	return s.Internal.ClientCancelDataTransfer(p0, p1, p2, p3)
 }
 
+func (s *FullNodeStruct) ClientCancelRetrievalDeal(p0 context.Context, p1 retrievalmarket.DealID) error {
+	return s.Internal.ClientCancelRetrievalDeal(p0, p1)
+}
+
 func (s *FullNodeStruct) ClientDataTransferUpdates(p0 context.Context) (<-chan api.DataTransferChannel, error) {
 	return s.Internal.ClientDataTransferUpdates(p0)
 }

api/mocks/mock_full.go

@@ -9,6 +9,7 @@ import (
 	address "github.com/filecoin-project/go-address"
 	bitfield "github.com/filecoin-project/go-bitfield"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
+	retrievalmarket "github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	storagemarket "github.com/filecoin-project/go-fil-markets/storagemarket"
 	auth "github.com/filecoin-project/go-jsonrpc/auth"
 	multistore "github.com/filecoin-project/go-multistore"
@@ -445,6 +446,20 @@ func (mr *MockFullNodeMockRecorder) ClientCancelDataTransfer(arg0, arg1, arg2, a
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClientCancelDataTransfer", reflect.TypeOf((*MockFullNode)(nil).ClientCancelDataTransfer), arg0, arg1, arg2, arg3)
 }
 
+// ClientCancelRetrievalDeal mocks base method
+func (m *MockFullNode) ClientCancelRetrievalDeal(arg0 context.Context, arg1 retrievalmarket.DealID) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ClientCancelRetrievalDeal", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// ClientCancelRetrievalDeal indicates an expected call of ClientCancelRetrievalDeal
+func (mr *MockFullNodeMockRecorder) ClientCancelRetrievalDeal(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClientCancelRetrievalDeal", reflect.TypeOf((*MockFullNode)(nil).ClientCancelRetrievalDeal), arg0, arg1)
+}
+
 // ClientDataTransferUpdates mocks base method
 func (m *MockFullNode) ClientDataTransferUpdates(arg0 context.Context) (<-chan api.DataTransferChannel, error) {
 	m.ctrl.T.Helper()

cli/client.go

@@ -91,6 +91,7 @@ var clientCmd = &cli.Command{
 		WithCategory("data", clientStat),
 		WithCategory("retrieval", clientFindCmd),
 		WithCategory("retrieval", clientRetrieveCmd),
+		WithCategory("retrieval", clientCancelRetrievalDealCmd),
 		WithCategory("util", clientCommPCmd),
 		WithCategory("util", clientCarGenCmd),
 		WithCategory("util", clientBalancesCmd),
@@ -2054,6 +2055,33 @@ var clientCancelTransfer = &cli.Command{
 	},
 }
 
+var clientCancelRetrievalDealCmd = &cli.Command{
+	Name:  "cancel-retrieval",
+	Usage: "Cancel a retrieval deal by deal ID; this also cancels the associated transfer",
+	Flags: []cli.Flag{
+		&cli.Int64Flag{
+			Name:     "deal-id",
+			Usage:    "specify retrieval deal by deal ID",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := GetFullNodeAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := ReqContext(cctx)
+
+		id := cctx.Int64("deal-id")
+		if id < 0 {
+			return errors.New("deal id cannot be negative")
+		}
+
+		return api.ClientCancelRetrievalDeal(ctx, retrievalmarket.DealID(id))
+	},
+}
+
 var clientListTransfers = &cli.Command{
 	Name:  "list-transfers",
 	Usage: "List ongoing data transfers for deals",

documentation/en/api-methods.md

@@ -35,6 +35,7 @@
 * [Client](#Client)
   * [ClientCalcCommP](#ClientCalcCommP)
   * [ClientCancelDataTransfer](#ClientCancelDataTransfer)
+  * [ClientCancelRetrievalDeal](#ClientCancelRetrievalDeal)
   * [ClientDataTransferUpdates](#ClientDataTransferUpdates)
   * [ClientDealPieceCID](#ClientDealPieceCID)
   * [ClientDealSize](#ClientDealSize)
@@ -921,6 +922,21 @@ Inputs:
 
 Response: `{}`
 
+### ClientCancelRetrievalDeal
+ClientCancelRetrievalDeal cancels an ongoing retrieval deal based on DealID
+
+
+Perms: write
+
+Inputs:
+```json
+[
+  5
+]
+```
+
+Response: `{}`
+
 ### ClientDataTransferUpdates
 
 

extern/sector-storage/stores/local.go

@@ -392,8 +392,10 @@ func (st *Local) Reserve(ctx context.Context, sid storage.SectorRef, ft storifac
 		}
 
 		p.reserved += overhead
+		p.reservations[sid.ID] |= fileType
 
 		prevDone := done
+		saveFileType := fileType
 		done = func() {
 			prevDone()
 
@@ -401,6 +403,10 @@ func (st *Local) Reserve(ctx context.Context, sid storage.SectorRef, ft storifac
 			defer st.localLk.Unlock()
 
 			p.reserved -= overhead
+			p.reservations[sid.ID] ^= saveFileType
+			if p.reservations[sid.ID] == storiface.FTNone {
+				delete(p.reservations, sid.ID)
+			}
 		}
 	}
 

go.sum

@@ -255,7 +255,6 @@ github.com/filecoin-project/go-amt-ipld/v3 v3.0.0 h1:Ou/q82QeHGOhpkedvaxxzpBYuqT
 github.com/filecoin-project/go-amt-ipld/v3 v3.0.0/go.mod h1:Qa95YNAbtoVCTSVtX38aAC1ptBnJfPma1R/zZsKmx4o=
 github.com/filecoin-project/go-bitfield v0.2.0/go.mod h1:CNl9WG8hgR5mttCnUErjcQjGvuiZjRqK9rHVBsQF4oM=
 github.com/filecoin-project/go-bitfield v0.2.3/go.mod h1:CNl9WG8hgR5mttCnUErjcQjGvuiZjRqK9rHVBsQF4oM=
-github.com/filecoin-project/go-bitfield v0.2.3/go.mod h1:CNl9WG8hgR5mttCnUErjcQjGvuiZjRqK9rHVBsQF4oM=
 github.com/filecoin-project/go-bitfield v0.2.4 h1:uZ7MeE+XfM5lqrHJZ93OnhQKc/rveW8p9au0C68JPgk=
 github.com/filecoin-project/go-bitfield v0.2.4/go.mod h1:CNl9WG8hgR5mttCnUErjcQjGvuiZjRqK9rHVBsQF4oM=
 github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2 h1:av5fw6wmm58FYMgJeoB/lK9XXrgdugYiTqkdxjTy9k8=
@@ -1993,7 +1992,6 @@ modernc.org/cc v1.0.0 h1:nPibNuDEx6tvYrUAtvDTTw98rx5juGsa5zuDnKwEEQQ=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/fileutil v1.0.0/go.mod h1:JHsWpkrk/CnVV1H/eGlFf85BEpfkrp56ro8nojIq9Q8=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
-modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/golex v1.0.1 h1:EYKY1a3wStt0RzHaH8mdSRNg78Ub0OHxYfCRWw35YtM=
 modernc.org/golex v1.0.1/go.mod h1:QCA53QtsT1NdGkaZZkF5ezFwk4IXh4BGNafAARTC254=
 modernc.org/lex v1.0.0/go.mod h1:G6rxMTy3cH2iA0iXL/HRRv4Znu8MK4higxph/lE7ypk=

node/impl/client/client.go

@@ -476,6 +476,29 @@ func (a *API) ClientListImports(ctx context.Context) ([]api.Import, error) {
 	return out, nil
 }
 
+func (a *API) ClientCancelRetrievalDeal(ctx context.Context, dealID retrievalmarket.DealID) error {
+	cerr := make(chan error)
+	go func() {
+		err := a.Retrieval.CancelDeal(dealID)
+
+		select {
+		case cerr <- err:
+		case <-ctx.Done():
+		}
+	}()
+
+	select {
+	case err := <-cerr:
+		if err != nil {
+			return xerrors.Errorf("failed to cancel retrieval deal: %w", err)
+		}
+
+		return nil
+	case <-ctx.Done():
+		return xerrors.Errorf("context timeout while canceling retrieval deal: %w", ctx.Err())
+	}
+}
+
 func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error {
 	events := make(chan marketevents.RetrievalEvent)
 	go a.clientRetrieve(ctx, order, ref, events)

node/modules/client.go

@@ -7,11 +7,8 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/filecoin-project/go-multistore"
-	"github.com/filecoin-project/go-state-types/abi"
-	"golang.org/x/xerrors"
-
 	"go.uber.org/fx"
+	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-data-transfer/channelmonitor"
 	dtimpl "github.com/filecoin-project/go-data-transfer/impl"
@@ -26,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"
@@ -134,17 +133,30 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 		return nil, err
 	}
 
-	// data-transfer push channel restart configuration
+	// data-transfer push / pull channel restart configuration:
-	dtRestartConf := dtimpl.ChannelRestartConfig(channelmonitor.Config{
+	dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
-		AcceptTimeout:          30 * time.Second,
+		// For now only monitor push channels (for storage deals)
-		Interval:               1 * time.Minute,
+		MonitorPushChannels: true,
-		MinBytesTransferred:    1024,
+		// TODO: Enable pull channel monitoring (for retrievals) when the
-		ChecksPerInterval:      10,
+		//  following issue has been fixed:
-		RestartBackoff:         10 * time.Minute,
+		// 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,
-		CompleteTimeout:        30 * time.Second,
+		// 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, dtRestartConf)
+	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
 	if err != nil {
 		return nil, err
 	}