Disconnect from Client after sending intent to accept

- Remove some tagging support from the deal environment
- Remove StartDataTransfer since we use client push now

storagemarket/impl/provider.go

@@ -16,7 +16,6 @@ import (
 	"github.com/ipfs/go-datastore"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/ipld/go-ipld-prime"
-	"github.com/libp2p/go-libp2p-core/peer"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-fil-markets/filestore"
@@ -603,11 +602,6 @@ func (p *providerDealEnvironment) Ask() storagemarket.StorageAsk {
 	return *sask.Ask
 }
 
-func (p *providerDealEnvironment) StartDataTransfer(ctx context.Context, to peer.ID, voucher datatransfer.Voucher, baseCid cid.Cid, selector ipld.Node) error {
-	_, err := p.p.dataTransfer.OpenPullDataChannel(ctx, to, voucher, baseCid, selector)
-	return err
-}
-
 func (p *providerDealEnvironment) GeneratePieceCommitmentToFile(payloadCid cid.Cid, selector ipld.Node) (cid.Cid, filestore.Path, filestore.Path, error) {
 	if p.p.universalRetrievalEnabled {
 		return providerutils.GeneratePieceCommitmentWithMetadata(p.p.fs, p.p.pio.GeneratePieceCommitmentToFile, p.p.proofType, payloadCid, selector)
@@ -648,15 +642,6 @@ func (p *providerDealEnvironment) SendSignedResponse(ctx context.Context, resp *
 	return err
 }
 
-func (p *providerDealEnvironment) TagConnection(proposalCid cid.Cid) error {
-	s, err := p.p.conns.DealStream(proposalCid)
-	if err != nil {
-		return err
-	}
-	s.TagProtectedConnection(proposalCid.String())
-	return nil
-}
-
 func (p *providerDealEnvironment) Disconnect(proposalCid cid.Cid) error {
 	s, err := p.p.conns.DealStream(proposalCid)
 	if err != nil {

storagemarket/impl/providerstates/provider_states.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	"github.com/filecoin-project/go-address"
-	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/filecoin-project/go-padreader"
 	"github.com/filecoin-project/go-statemachine/fsm"
 	"github.com/filecoin-project/specs-actors/actors/abi"
@@ -16,7 +15,6 @@ import (
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipld/go-ipld-prime"
-	"github.com/libp2p/go-libp2p-core/peer"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-fil-markets/filestore"
@@ -35,10 +33,8 @@ type ProviderDealEnvironment interface {
 	Address() address.Address
 	Node() storagemarket.StorageProviderNode
 	Ask() storagemarket.StorageAsk
-	StartDataTransfer(ctx context.Context, to peer.ID, voucher datatransfer.Voucher, baseCid cid.Cid, selector ipld.Node) error
 	GeneratePieceCommitmentToFile(payloadCid cid.Cid, selector ipld.Node) (cid.Cid, filestore.Path, filestore.Path, error)
 	SendSignedResponse(ctx context.Context, response *network.Response) error
-	TagConnection(proposalCid cid.Cid) error
 	Disconnect(proposalCid cid.Cid) error
 	FileStore() filestore.FileStore
 	PieceStore() piecestore.PieceStore
@@ -103,11 +99,6 @@ func ValidateDealProposal(ctx fsm.Context, environment ProviderDealEnvironment,
 		return ctx.Trigger(storagemarket.ProviderEventDealRejected, xerrors.New("clientMarketBalance.Available too small"))
 	}
 
-	if err := environment.TagConnection(deal.ProposalCid); err != nil {
-		// some conns may not support tagging, just log
-		log.Warnf("Error tagging deal connection: %w", err)
-	}
-	// TODO: Send intent to accept
 	return ctx.Trigger(storagemarket.ProviderEventDealDeciding)
 }
 
@@ -133,6 +124,10 @@ func DecideOnProposal(ctx fsm.Context, environment ProviderDealEnvironment, deal
 		return ctx.Trigger(storagemarket.ProviderEventSendResponseFailed, err)
 	}
 
+	if err := environment.Disconnect(deal.ProposalCid); err != nil {
+		log.Warnf("closing client connection: %+v", err)
+	}
+
 	return ctx.Trigger(storagemarket.ProviderEventDataRequested)
 }
 

storagemarket/impl/providerstates/provider_states_test.go

@@ -9,15 +9,13 @@ import (
 	"testing"
 
 	"github.com/filecoin-project/go-address"
-	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/filecoin-project/go-statemachine/fsm"
 	fsmtest "github.com/filecoin-project/go-statemachine/fsm/testutil"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 	"github.com/filecoin-project/specs-actors/actors/builtin/market"
 	"github.com/filecoin-project/specs-actors/actors/runtime/exitcode"
 	"github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
-	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/go-fil-markets/filestore"
@@ -47,9 +45,6 @@ func TestValidateDealProposal(t *testing.T) {
 		dealInspector     func(t *testing.T, deal storagemarket.MinerDeal, env *fakeEnvironment)
 	}{
 		"succeeds": {
-			environmentParams: environmentParams{
-				TagsProposal: true,
-			},
 			dealInspector: func(t *testing.T, deal storagemarket.MinerDeal, env *fakeEnvironment) {
 				tut.AssertDealState(t, storagemarket.StorageDealAcceptWait, deal.State)
 			},
@@ -82,7 +77,7 @@ func TestValidateDealProposal(t *testing.T) {
 			},
 		},
 		"CurrentHeight <= StartEpoch - DealAcceptanceBuffer() succeeds": {
-			environmentParams: environmentParams{DealAcceptanceBuffer: 10, TagsProposal: true},
+			environmentParams: environmentParams{DealAcceptanceBuffer: 10},
 			dealParams:        dealParams{StartEpoch: 200},
 			nodeParams:        nodeParams{Height: 190},
 			dealInspector: func(t *testing.T, deal storagemarket.MinerDeal, env *fakeEnvironment) {
@@ -953,10 +948,6 @@ func (fe *fakeEnvironment) Ask() storagemarket.StorageAsk {
 	return fe.ask
 }
 
-func (fe *fakeEnvironment) StartDataTransfer(ctx context.Context, to peer.ID, voucher datatransfer.Voucher, baseCid cid.Cid, selector ipld.Node) error {
-	return fe.dataTransferError
-}
-
 func (fe *fakeEnvironment) GeneratePieceCommitmentToFile(payloadCid cid.Cid, selector ipld.Node) (cid.Cid, filestore.Path, filestore.Path, error) {
 	return fe.pieceCid, fe.path, fe.metadataPath, fe.generateCommPError
 }
@@ -965,11 +956,6 @@ func (fe *fakeEnvironment) SendSignedResponse(ctx context.Context, response *net
 	return fe.sendSignedResponseError
 }
 
-func (fe *fakeEnvironment) TagConnection(proposalCid cid.Cid) error {
-	fe.receivedTags[proposalCid.String()] = struct{}{}
-	return nil
-}
-
 func (fe *fakeEnvironment) VerifyExpectations(t *testing.T) {
 	require.Equal(t, fe.expectedTags, fe.receivedTags)
 }