Resume Storage Market Data Transfer (#430)

* Resume Storage Market Client Data Transfer  (#407)

* first checkpoint

* draft 2

* working client restarts

* more fixes

* comment

* document test failure

* Resume StorageMarket Provider Data Transfer (#411)

* first checkpoint

* draft 2

* working client restarts

* more fixes

* comment

* provider restarts

* provider restarts

* fix tests

* v1 with problems

* more fixes

* feat(storagemarket): rerecord channel id on restarted

Co-authored-by: hannahhoward <[email protected]>

docs/storageclient.mmd

@@ -13,6 +13,7 @@ stateDiagram-v2
 	state "StorageDealEnsureClientFunds" as 21
 	state "StorageDealClientFunding" as 23
 	state "StorageDealError" as 26
+	state "StorageDealClientTransferRestart" as 28
 	3 : On entry runs ValidateDealPublished
 	5 : On entry runs VerifyDealActivated
 	7 : On entry runs WaitForDealCompletion
@@ -22,6 +23,7 @@ stateDiagram-v2
 	16 : On entry runs InitiateDataTransfer
 	21 : On entry runs EnsureClientFunds
 	23 : On entry runs WaitForFunding
+	28 : On entry runs RestartDataTransfer
 	[*] --> 0
 	note right of 0
 		The following events are not shown cause they can trigger from any state.
@@ -42,7 +44,10 @@ stateDiagram-v2
 	12 --> 11 : ClientEventUnexpectedDealState
 	16 --> 11 : ClientEventDataTransferFailed
 	17 --> 11 : ClientEventDataTransferFailed
+	28 --> 11 : ClientEventDataTransferRestartFailed
 	16 --> 17 : ClientEventDataTransferInitiated
+	16 --> 17 : ClientEventDataTransferRestarted
+	28 --> 17 : ClientEventDataTransferRestarted
 	16 --> 13 : ClientEventDataTransferComplete
 	17 --> 13 : ClientEventDataTransferComplete
 	13 --> 13 : ClientEventWaitForDealState
@@ -57,14 +62,17 @@ stateDiagram-v2
 	7 --> 8 : ClientEventDealExpired
 	7 --> 26 : ClientEventDealCompletionFailed
 	11 --> 26 : ClientEventFailed
-	17 --> 16 : ClientEventRestart
+	17 --> 28 : ClientEventRestart
 
 	note left of 3 : The following events only record in this state.<br><br>ClientEventFundsReleased
 
 
 	note left of 11 : The following events only record in this state.<br><br>ClientEventFundsReleased
 
 
+	note left of 17 : The following events only record in this state.<br><br>ClientEventDataTransferRestarted
+
+
 	note left of 21 : The following events only record in this state.<br><br>ClientEventFundsReserved
 
 	9 --> [*]

docs/storageprovider.mmd

@@ -18,6 +18,7 @@ stateDiagram-v2
 	state "StorageDealPublish" as 24
 	state "StorageDealPublishing" as 25
 	state "StorageDealError" as 26
+	state "StorageDealProviderTransferRestart" as 27
 	4 : On entry runs HandoffDeal
 	5 : On entry runs VerifyDealActivated
 	6 : On entry runs CleanupDeal
@@ -31,6 +32,7 @@ stateDiagram-v2
 	22 : On entry runs WaitForFunding
 	24 : On entry runs PublishDeal
 	25 : On entry runs WaitForPublish
+	27 : On entry runs RestartDataTransfer
 	[*] --> 0
 	note right of 0
 		The following events are not shown cause they can trigger from any state.
@@ -47,6 +49,9 @@ stateDiagram-v2
 	15 --> 18 : ProviderEventDataRequested
 	17 --> 11 : ProviderEventDataTransferFailed
 	18 --> 17 : ProviderEventDataTransferInitiated
+	27 --> 11 : ProviderEventDataTransferRestartFailed
+	18 --> 17 : ProviderEventDataTransferRestarted
+	27 --> 17 : ProviderEventDataTransferRestarted
 	17 --> 19 : ProviderEventDataTransferCompleted
 	19 --> 11 : ProviderEventDataVerificationFailed
 	18 --> 20 : ProviderEventVerifiedData
@@ -75,6 +80,7 @@ stateDiagram-v2
 	10 --> 26 : ProviderEventRestart
 	14 --> 26 : ProviderEventRestart
 	15 --> 26 : ProviderEventRestart
+	17 --> 27 : ProviderEventRestart
 	20 --> 11 : ProviderEventTrackFundsFailed
 
 	note left of 4 : The following events only record in this state.<br><br>ProviderEventPieceStoreErrored
@@ -83,6 +89,9 @@ stateDiagram-v2
 	note left of 11 : The following events only record in this state.<br><br>ProviderEventFundsReleased
 
 
+	note left of 17 : The following events only record in this state.<br><br>ProviderEventDataTransferRestarted
+
+
 	note left of 20 : The following events only record in this state.<br><br>ProviderEventFundsReserved
 
 

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/filecoin-project/filecoin-ffi v0.30.4-0.20200716204036-cddc56607e1d
 	github.com/filecoin-project/go-address v0.0.3
 	github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2
-	github.com/filecoin-project/go-data-transfer v0.6.7
+	github.com/filecoin-project/go-data-transfer v0.6.8-0.20201012061237-d181ba9cacad
 	github.com/filecoin-project/go-ds-versioning v0.1.0
 	github.com/filecoin-project/go-multistore v0.0.3
 	github.com/filecoin-project/go-padreader v0.0.0-20200903213702-ed5fae088b20

go.sum

@@ -104,8 +104,8 @@ github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2 h1:a
 github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2/go.mod h1:pqTiPHobNkOVM5thSRsHYjyQfq7O5QSCMhvuu9JoDlg=
 github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03 h1:2pMXdBnCiXjfCYx/hLqFxccPoqsSveQFxVLvNxy9bus=
 github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03/go.mod h1:+viYnvGtUTgJRdy6oaeF4MTFKAfatX071MPDPBL11EQ=
-github.com/filecoin-project/go-data-transfer v0.6.7 h1:Kacr5qz2YWtd3sensU6aXFtES7joeapVDeXApeUD35I=
-github.com/filecoin-project/go-data-transfer v0.6.7/go.mod h1:C++k1U6+jMQODOaen5OPDo9XQbth9Yq3ie94vNjBJbk=
+github.com/filecoin-project/go-data-transfer v0.6.8-0.20201012061237-d181ba9cacad h1:yO7pQMXjM1i6IGkYOx4Q3v3hXCGc4Agn0bapsr/twcU=
+github.com/filecoin-project/go-data-transfer v0.6.8-0.20201012061237-d181ba9cacad/go.mod h1:QvXh2HwVSuKAmIkqZGuqW7VBwO1Yj1F03WdrM76mJ/s=
 github.com/filecoin-project/go-ds-versioning v0.1.0 h1:y/X6UksYTsK8TLCI7rttCKEvl8btmWxyFMEeeWGUxIQ=
 github.com/filecoin-project/go-ds-versioning v0.1.0/go.mod h1:mp16rb4i2QPmxBnmanUx8i/XANp+PFCCJWiAb+VW4/s=
 github.com/filecoin-project/go-fil-commcid v0.0.0-20200716160307-8f644712406f h1:GxJzR3oRIMTPtpZ0b7QF8FKPK6/iPAc7trhlL5k/g+s=
@@ -123,7 +123,6 @@ github.com/filecoin-project/go-state-types v0.0.0-20200904021452-1883f36ca2f4 h1
 github.com/filecoin-project/go-state-types v0.0.0-20200904021452-1883f36ca2f4/go.mod h1:IQ0MBPnonv35CJHtWSN3YY1Hz2gkPru1Q9qoaYLxx9I=
 github.com/filecoin-project/go-state-types v0.0.0-20200928172055-2df22083d8ab h1:cEDC5Ei8UuT99hPWhCjA72SM9AuRtnpvdSTIYbnzN8I=
 github.com/filecoin-project/go-state-types v0.0.0-20200928172055-2df22083d8ab/go.mod h1:ezYnPf0bNkTsDibL/psSz5dy4B5awOJ/E7P2Saeep8g=
-github.com/filecoin-project/go-statemachine v0.0.0-20200714194326-a77c3ae20989/go.mod h1:FGwQgZAt2Gh5mjlwJUlVB62JeYdo+if0xWxSEfBD9ig=
 github.com/filecoin-project/go-statemachine v0.0.0-20200925024713-05bd7c71fbfe h1:dF8u+LEWeIcTcfUcCf3WFVlc81Fr2JKg8zPzIbBDKDw=
 github.com/filecoin-project/go-statemachine v0.0.0-20200925024713-05bd7c71fbfe/go.mod h1:FGwQgZAt2Gh5mjlwJUlVB62JeYdo+if0xWxSEfBD9ig=
 github.com/filecoin-project/go-statestore v0.1.0 h1:t56reH59843TwXHkMcwyuayStBIiWBRilQjQ+5IiwdQ=

retrievalmarket/impl/dtutils/dtutils.go

@@ -98,7 +98,7 @@ const noEvent = rm.ClientEvent(math.MaxUint64)
 
 func clientEvent(event datatransfer.Event, channelState datatransfer.ChannelState) (rm.ClientEvent, []interface{}) {
 	switch event.Code {
-	case datatransfer.Progress:
+	case datatransfer.DataReceived:
 		return rm.ClientEventBlocksReceived, []interface{}{channelState.Received()}
 	case datatransfer.FinishTransfer:
 		return rm.ClientEventAllBlocksReceived, nil

retrievalmarket/impl/dtutils/dtutils_test.go

@@ -6,6 +6,7 @@ import (
 	"math/rand"
 	"testing"
 
+	"github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
 	peer "github.com/libp2p/go-libp2p-core/peer"
 	"github.com/stretchr/testify/require"
@@ -147,7 +148,7 @@ func TestClientDataTransferSubscriber(t *testing.T) {
 			ignored: true,
 		},
 		"progress": {
-			code: datatransfer.Progress,
+			code: datatransfer.DataReceived,
 			state: shared_testutil.TestChannelParams{
 				Vouchers: []datatransfer.Voucher{&dealProposal},
 				Status:   datatransfer.Ongoing,
@@ -433,7 +434,8 @@ func (fsg *fakeStoreGetter) Get(otherPeer peer.ID, dealID rm.DealID) (*multistor
 
 type fakeTransport struct{}
 
-func (ft *fakeTransport) OpenChannel(ctx context.Context, dataSender peer.ID, channelID datatransfer.ChannelID, root ipld.Link, stor ipld.Node, msg datatransfer.Message) error {
+func (ft *fakeTransport) OpenChannel(ctx context.Context, dataSender peer.ID, channelID datatransfer.ChannelID, root ipld.Link, stor ipld.Node,
+	doNotSend []cid.Cid, msg datatransfer.Message) error {
 	return nil
 }
 

retrievalmarket/impl/integration_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
+	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
+	"github.com/ipfs/go-graphsync/network"
 	"github.com/ipld/go-ipld-prime"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	basicnode "github.com/ipld/go-ipld-prime/node/basic"
@@ -19,6 +21,7 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	dtimpl "github.com/filecoin-project/go-data-transfer/impl"
+	"github.com/filecoin-project/go-data-transfer/testutil"
 	dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
 	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-state-types/abi"
@@ -97,10 +100,12 @@ func requireSetupTestClientAndProvider(ctx context.Context, t *testing.T, payChA
 		CreatePaychCID: cids[0],
 		AddFundsCID:    cids[1],
 	})
-	dtTransport1 := dtgstransport.NewTransport(testData.Host1.ID(), testData.GraphSync1)
+
+	gs1 := graphsyncimpl.New(ctx, network.NewFromLibp2pHost(testData.Host1), testData.Loader1, testData.Storer1)
+	dtTransport1 := dtgstransport.NewTransport(testData.Host1.ID(), gs1)
 	dt1, err := dtimpl.NewDataTransfer(testData.DTStore1, testData.DTNet1, dtTransport1, testData.DTStoredCounter1)
 	require.NoError(t, err)
-	err = dt1.Start(ctx)
+	testutil.StartAndWaitForReady(ctx, t, dt1)
 	require.NoError(t, err)
 	clientDs := namespace.Wrap(testData.Ds1, datastore.NewKey("/retrievals/client"))
 	client, err := retrievalimpl.NewClient(nw1, testData.MultiStore1, dt1, rcNode1, &tut.TestPeerResolver{}, clientDs, testData.RetrievalStoredCounter1)
@@ -135,10 +140,12 @@ func requireSetupTestClientAndProvider(ctx context.Context, t *testing.T, payChA
 	}
 
 	paymentAddress := address.TestAddress2
-	dtTransport2 := dtgstransport.NewTransport(testData.Host2.ID(), testData.GraphSync2)
+
+	gs2 := graphsyncimpl.New(ctx, network.NewFromLibp2pHost(testData.Host2), testData.Loader2, testData.Storer2)
+	dtTransport2 := dtgstransport.NewTransport(testData.Host2.ID(), gs2)
 	dt2, err := dtimpl.NewDataTransfer(testData.DTStore2, testData.DTNet2, dtTransport2, testData.DTStoredCounter2)
 	require.NoError(t, err)
-	err = dt2.Start(ctx)
+	testutil.StartAndWaitForReady(ctx, t, dt2)
 	require.NoError(t, err)
 	providerDs := namespace.Wrap(testData.Ds2, datastore.NewKey("/retrievals/provider"))
 	provider, err := retrievalimpl.NewProvider(paymentAddress, providerNode, nw2, pieceStore, testData.MultiStore2, dt2, providerDs)
@@ -589,10 +596,12 @@ func setupClient(
 		IntegrationTest:        true,
 		ChannelAvailableFunds:  channelAvailableFunds,
 	})
-	dtTransport1 := dtgstransport.NewTransport(testData.Host1.ID(), testData.GraphSync1)
+
+	gs1 := graphsyncimpl.New(ctx, network.NewFromLibp2pHost(testData.Host1), testData.Loader1, testData.Storer1)
+	dtTransport1 := dtgstransport.NewTransport(testData.Host1.ID(), gs1)
 	dt1, err := dtimpl.NewDataTransfer(testData.DTStore1, testData.DTNet1, dtTransport1, testData.DTStoredCounter1)
 	require.NoError(t, err)
-	err = dt1.Start(ctx)
+	testutil.StartAndWaitForReady(ctx, t, dt1)
 	require.NoError(t, err)
 	clientDs := namespace.Wrap(testData.Ds1, datastore.NewKey("/retrievals/client"))
 
@@ -624,10 +633,12 @@ func setupProvider(
 	}
 	pieceStore.ExpectCID(payloadCID, cidInfo)
 	pieceStore.ExpectPiece(expectedPiece, pieceInfo)
-	dtTransport2 := dtgstransport.NewTransport(testData.Host2.ID(), testData.GraphSync2)
+
+	gs2 := graphsyncimpl.New(ctx, network.NewFromLibp2pHost(testData.Host2), testData.Loader2, testData.Storer2)
+	dtTransport2 := dtgstransport.NewTransport(testData.Host2.ID(), gs2)
 	dt2, err := dtimpl.NewDataTransfer(testData.DTStore2, testData.DTNet2, dtTransport2, testData.DTStoredCounter2)
 	require.NoError(t, err)
-	err = dt2.Start(ctx)
+	testutil.StartAndWaitForReady(ctx, t, dt2)
 	require.NoError(t, err)
 	providerDs := namespace.Wrap(testData.Ds2, datastore.NewKey("/retrievals/provider"))
 

retrievalmarket/storage_retrieval_integration_test.go

@@ -35,7 +35,8 @@ import (
 
 func TestStorageRetrieval(t *testing.T) {
 	bgCtx := context.Background()
-	sh := testharness.NewHarness(t, bgCtx, true, testnodes.DelayFakeCommonNode{}, false)
+	sh := testharness.NewHarness(t, bgCtx, true, testnodes.DelayFakeCommonNode{},
+		testnodes.DelayFakeCommonNode{}, false)
 	shared_testutil.StartAndWaitForReady(bgCtx, t, sh.Client)
 	shared_testutil.StartAndWaitForReady(bgCtx, t, sh.Provider)
 

shared_testutil/mocknet.go

@@ -15,9 +15,6 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	dss "github.com/ipfs/go-datastore/sync"
-	"github.com/ipfs/go-graphsync"
-	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
-	"github.com/ipfs/go-graphsync/network"
 	bstore "github.com/ipfs/go-ipfs-blockstore"
 	chunk "github.com/ipfs/go-ipfs-chunker"
 	offline "github.com/ipfs/go-ipfs-exchange-offline"
@@ -53,8 +50,6 @@ type Libp2pTestData struct {
 	MultiStore2             *multistore.MultiStore
 	DagService1             ipldformat.DAGService
 	DagService2             ipldformat.DAGService
-	GraphSync1              graphsync.GraphExchange
-	GraphSync2              graphsync.GraphExchange
 	DTNet1                  dtnet.DataTransferNetwork
 	DTNet2                  dtnet.DataTransferNetwork
 	DTStore1                datastore.Batching
@@ -66,6 +61,8 @@ type Libp2pTestData struct {
 	Host1                   host.Host
 	Host2                   host.Host
 	OrigBytes               []byte
+
+	MockNet mocknet.Mocknet
 }
 
 func NewLibp2pTestData(ctx context.Context, t *testing.T) *Libp2pTestData {
@@ -146,15 +143,14 @@ func NewLibp2pTestData(ctx context.Context, t *testing.T) *Libp2pTestData {
 	err = mn.LinkAll()
 	require.NoError(t, err)
 
-	testData.GraphSync1 = graphsyncimpl.New(ctx, network.NewFromLibp2pHost(testData.Host1), testData.Loader1, testData.Storer1)
-	testData.GraphSync2 = graphsyncimpl.New(ctx, network.NewFromLibp2pHost(testData.Host2), testData.Loader2, testData.Storer2)
-
 	testData.DTNet1 = dtnet.NewFromLibp2pHost(testData.Host1)
 	testData.DTNet2 = dtnet.NewFromLibp2pHost(testData.Host2)
 
 	testData.DTStore1 = namespace.Wrap(testData.Ds1, datastore.NewKey("DataTransfer1"))
 	testData.DTStore2 = namespace.Wrap(testData.Ds1, datastore.NewKey("DataTransfer2"))
 
+	testData.MockNet = mn
+
 	return testData
 }
 

shared_testutil/test_datatransfer.go

@@ -82,6 +82,10 @@ func (tdt *TestDataTransfer) OpenPushDataChannel(ctx context.Context, to peer.ID
 	return datatransfer.ChannelID{}, nil
 }
 
+func (tdt *TestDataTransfer) RestartDataTransferChannel(ctx context.Context, chId datatransfer.ChannelID) error {
+	return nil
+}
+
 // OpenPullDataChannel does nothing
 func (tdt *TestDataTransfer) OpenPullDataChannel(ctx context.Context, to peer.ID, voucher datatransfer.Voucher, baseCid cid.Cid, selector ipld.Node) (datatransfer.ChannelID, error) {
 	return datatransfer.ChannelID{}, nil
@@ -123,4 +127,8 @@ func (tdt *TestDataTransfer) InProgressChannels(ctx context.Context) (map[datatr
 	return map[datatransfer.ChannelID]datatransfer.ChannelState{}, nil
 }
 
+func (tdt *TestDataTransfer) OnReady(f datatransfer.ReadyFunc) {
+
+}
+
 var _ datatransfer.Manager = new(TestDataTransfer)

shared_testutil/testchannel.go

@@ -16,6 +16,7 @@ type TestChannelParams struct {
 	TransferID     datatransfer.TransferID
 	BaseCID        cid.Cid
 	Selector       ipld.Node
+	SelfPeer       peer.ID
 	Sender         peer.ID
 	Recipient      peer.ID
 	TotalSize      uint64
@@ -26,10 +27,12 @@ type TestChannelParams struct {
 	Status         datatransfer.Status
 	Vouchers       []datatransfer.Voucher
 	VoucherResults []datatransfer.VoucherResult
+	ReceivedCids   []cid.Cid
 }
 
 // TestChannel implements a datatransfer channel with set values
 type TestChannel struct {
+	selfPeer       peer.ID
 	transferID     datatransfer.TransferID
 	baseCID        cid.Cid
 	selector       ipld.Node
@@ -43,6 +46,7 @@ type TestChannel struct {
 	status         datatransfer.Status
 	vouchers       []datatransfer.Voucher
 	voucherResults []datatransfer.VoucherResult
+	receivedCids   []cid.Cid
 }
 
 // FakeDTType is a fake voucher type
@@ -56,6 +60,7 @@ func (f FakeDTType) Type() datatransfer.TypeIdentifier { return "Fake" }
 func NewTestChannel(params TestChannelParams) datatransfer.ChannelState {
 	peers := GeneratePeers(2)
 	tc := &TestChannel{
+		selfPeer:       peers[0],
 		transferID:     datatransfer.TransferID(rand.Uint64()),
 		baseCID:        GenerateCids(1)[0],
 		selector:       shared.AllSelector(),
@@ -69,6 +74,9 @@ func NewTestChannel(params TestChannelParams) datatransfer.ChannelState {
 		vouchers:       []datatransfer.Voucher{FakeDTType{}},
 		voucherResults: []datatransfer.VoucherResult{FakeDTType{}},
 	}
+
+	tc.receivedCids = params.ReceivedCids
+
 	if params.TransferID != 0 {
 		tc.transferID = params.TransferID
 	}
@@ -78,6 +86,10 @@ func NewTestChannel(params TestChannelParams) datatransfer.ChannelState {
 	if params.Selector != nil {
 		tc.selector = params.Selector
 	}
+	if params.SelfPeer != peer.ID("") {
+		tc.selfPeer = params.SelfPeer
+	}
+
 	if params.Sender != peer.ID("") {
 		tc.sender = params.Sender
 	}
@@ -121,6 +133,11 @@ func (tc *TestChannel) Selector() ipld.Node {
 	return tc.selector
 }
 
+// ReceivedCids returns the cids received so far
+func (tc *TestChannel) ReceivedCids() []cid.Cid {
+	return tc.receivedCids
+}
+
 // Voucher returns the voucher for this data transfer
 func (tc *TestChannel) Voucher() datatransfer.Voucher {
 	return tc.vouchers[0]
@@ -155,6 +172,19 @@ func (tc *TestChannel) ChannelID() datatransfer.ChannelID {
 	}
 }
 
+// SelfPeer returns the peer this channel belongs to
+func (tc *TestChannel) SelfPeer() peer.ID {
+	return tc.selfPeer
+}
+
+// OtherPeer returns the channel counter party peer
+func (tc *TestChannel) OtherPeer() peer.ID {
+	if tc.selfPeer == tc.sender {
+		return tc.recipient
+	}
+	return tc.sender
+}
+
 // OtherParty returns the opposite party in the channel to the passed in party
 func (tc *TestChannel) OtherParty(thisParty peer.ID) peer.ID {
 	if tc.sender == thisParty {