feat(markets): record data transfer errors

docs/storageclient.mmd

@@ -70,7 +70,7 @@ stateDiagram-v2
 	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 17 : The following events only record in this state.<br><br>ClientEventDataTransferRestarted<br>ClientEventDataTransferStalled
 
 
 	note left of 21 : The following events only record in this state.<br><br>ClientEventFundsReserved

docs/storageprovider.mmd

@@ -89,7 +89,7 @@ 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 17 : The following events only record in this state.<br><br>ProviderEventDataTransferRestarted<br>ProviderEventDataTransferStalled
 
 
 	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.8-0.20201012061237-d181ba9cacad
+	github.com/filecoin-project/go-data-transfer v0.6.8-0.20201013092017-eee8d1cde5d5
 	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
@@ -24,7 +24,7 @@ require (
 	github.com/ipfs/go-blockservice v0.1.4-0.20200624145336-a978cec6e834
 	github.com/ipfs/go-cid v0.0.7
 	github.com/ipfs/go-datastore v0.4.5
-	github.com/ipfs/go-graphsync v0.2.1
+	github.com/ipfs/go-graphsync v0.2.1-0.20201013053840-5d8ea8076a2c
 	github.com/ipfs/go-ipfs-blockstore v1.0.1
 	github.com/ipfs/go-ipfs-blocksutil v0.0.1
 	github.com/ipfs/go-ipfs-chunker v0.0.5

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.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-data-transfer v0.6.8-0.20201013092017-eee8d1cde5d5 h1:kgmVHDP+rqcPqytpC215MzRRXrW9FqTfn0VFWhB2oog=
+github.com/filecoin-project/go-data-transfer v0.6.8-0.20201013092017-eee8d1cde5d5/go.mod h1:1zaap881YZLMeIxhvLM0zA9qfUkh3bjOLSRaOoO9WRM=
 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=
@@ -278,8 +278,8 @@ github.com/ipfs/go-filestore v1.0.0 h1:QR7ekKH+q2AGiWDc7W2Q0qHuYSRZGUJqUn0GsegEP
 github.com/ipfs/go-filestore v1.0.0/go.mod h1:/XOCuNtIe2f1YPbiXdYvD0BKLA0JR1MgPiFOdcuu9SM=
 github.com/ipfs/go-graphsync v0.1.0 h1:RjLk7ha1tJtDXktqoxOjhvx4lDuzzIU+xQ+PEi74r3s=
 github.com/ipfs/go-graphsync v0.1.0/go.mod h1:jMXfqIEDFukLPZHqDPp8tJMbHO9Rmeb9CEGevngQbmE=
-github.com/ipfs/go-graphsync v0.2.1 h1:MdehhqBSuTI2LARfKLkpYnt0mUrqHs/mtuDnESXHBfU=
-github.com/ipfs/go-graphsync v0.2.1/go.mod h1:gEBvJUNelzMkaRPJTpg/jaKN4AQW/7wDWu0K92D8o10=
+github.com/ipfs/go-graphsync v0.2.1-0.20201013053840-5d8ea8076a2c h1:De/AZGvRa3WMyw5zdMMhcvRcho46BVo+C0NRud+T4io=
+github.com/ipfs/go-graphsync v0.2.1-0.20201013053840-5d8ea8076a2c/go.mod h1:gEBvJUNelzMkaRPJTpg/jaKN4AQW/7wDWu0K92D8o10=
 github.com/ipfs/go-hamt-ipld v0.1.1 h1:0IQdvwnAAUKmDE+PMJa5y1QiwOPHpI9+eAbQEEEYthk=
 github.com/ipfs/go-hamt-ipld v0.1.1/go.mod h1:1EZCr2v0jlCnhpa+aZ0JZYp8Tt2w16+JJOAVz17YcDk=
 github.com/ipfs/go-ipfs-blockstore v0.0.1/go.mod h1:d3WClOmRQKFnJ0Jz/jj/zmksX0ma1gROTlovZKBmN08=

retrievalmarket/impl/dtutils/dtutils.go

@@ -31,6 +31,8 @@ func providerEvent(event datatransfer.Event, channelState datatransfer.ChannelSt
 	switch event.Code {
 	case datatransfer.Accept:
 		return rm.ProviderEventDealAccepted, []interface{}{channelState.ChannelID()}
+	case datatransfer.Disconnected:
+		return rm.ProviderEventDataTransferError, []interface{}{fmt.Errorf("deal data transfer stalled (peer hungup)")}
 	case datatransfer.Error:
 		return rm.ProviderEventDataTransferError, []interface{}{fmt.Errorf("deal data transfer failed: %s", event.Message)}
 	case datatransfer.Cancel:
@@ -112,6 +114,8 @@ func clientEvent(event datatransfer.Event, channelState datatransfer.ChannelStat
 		}
 
 		return clientEventForResponse(response)
+	case datatransfer.Disconnected:
+		return rm.ClientEventDataTransferError, []interface{}{fmt.Errorf("deal data transfer stalled (peer hungup)")}
 	case datatransfer.Error:
 		if channelState.Message() == datatransfer.ErrRejected.Error() {
 			return rm.ClientEventDealRejected, []interface{}{"rejected for unknown reasons"}

retrievalmarket/impl/dtutils/dtutils_test.go

@@ -90,6 +90,20 @@ func TestProviderDataTransferSubscriber(t *testing.T) {
 			expectedEvent: rm.ProviderEventDataTransferError,
 			expectedArgs:  []interface{}{errors.New("deal data transfer failed: something went wrong")},
 		},
+		"disconnected": {
+			code:    datatransfer.Disconnected,
+			message: "something went wrong",
+			state: shared_testutil.TestChannelParams{
+				IsPull:     true,
+				TransferID: transferID,
+				Sender:     testPeers[0],
+				Recipient:  testPeers[1],
+				Vouchers:   []datatransfer.Voucher{&dealProposal},
+				Status:     datatransfer.Ongoing},
+			expectedID:    rm.ProviderDealIdentifier{DealID: dealProposal.ID, Receiver: testPeers[1]},
+			expectedEvent: rm.ProviderEventDataTransferError,
+			expectedArgs:  []interface{}{errors.New("deal data transfer stalled (peer hungup)")},
+		},
 		"completed": {
 			code: datatransfer.ResumeResponder,
 			state: shared_testutil.TestChannelParams{
@@ -287,6 +301,16 @@ func TestClientDataTransferSubscriber(t *testing.T) {
 			expectedEvent: rm.ClientEventDataTransferError,
 			expectedArgs:  []interface{}{errors.New("deal data transfer failed: something went wrong")},
 		},
+		"disconnected": {
+			code:    datatransfer.Disconnected,
+			message: "something went wrong",
+			state: shared_testutil.TestChannelParams{
+				Vouchers: []datatransfer.Voucher{&dealProposal},
+				Status:   datatransfer.Ongoing},
+			expectedID:    dealProposal.ID,
+			expectedEvent: rm.ClientEventDataTransferError,
+			expectedArgs:  []interface{}{errors.New("deal data transfer stalled (peer hungup)")},
+		},
 		"error, response rejected": {
 			code:    datatransfer.Error,
 			message: datatransfer.ErrRejected.Error(),

storagemarket/events.go

@@ -93,6 +93,9 @@ const (
 
 	// ClientEventRestart is used to resume the deal after a state machine shutdown
 	ClientEventRestart
+
+	// ClientEventDataTransferStalled happens when the clients data transfer experiences a disconnect
+	ClientEventDataTransferStalled
 )
 
 // ClientEvents maps client event codes to string names
@@ -127,6 +130,7 @@ var ClientEvents = map[ClientEvent]string{
 	ClientEventRestart:                    "ClientEventRestart",
 	ClientEventDataTransferRestarted:      "ClientEventDataTransferRestarted",
 	ClientEventDataTransferRestartFailed:  "ClientEventDataTransferRestartFailed",
+	ClientEventDataTransferStalled:        "ClientEventDataTransferStalled",
 }
 
 // ProviderEvent is an event that happens in the provider's deal state machine
@@ -253,6 +257,9 @@ const (
 
 	// ProviderEventDataTransferRestartFailed means a data transfer that was restarted by the provider failed
 	ProviderEventDataTransferRestartFailed
+
+	// ProviderEventDataTransferStalled happens when the providers data transfer experiences a disconnect
+	ProviderEventDataTransferStalled
 )
 
 // ProviderEvents maps provider event codes to string names
@@ -295,4 +302,5 @@ var ProviderEvents = map[ProviderEvent]string{
 	ProviderEventRestart:                   "ProviderEventRestart",
 	ProviderEventDataTransferRestarted:     "ProviderEventDataTransferRestarted",
 	ProviderEventDataTransferRestartFailed: "ProviderEventDataTransferRestartFailed",
+	ProviderEventDataTransferStalled:       "ProviderEventDataTransferStalled",
 }

storagemarket/impl/clientstates/client_fsm.go

@@ -100,9 +100,16 @@ var ClientEvents = fsm.Events{
 		From(storagemarket.StorageDealTransferring).ToJustRecord().
 		Action(func(deal *storagemarket.ClientDeal, channelId datatransfer.ChannelID) error {
 			deal.TransferChannelID = &channelId
+			deal.Message = ""
 			return nil
 		}),
 
+	fsm.Event(storagemarket.ClientEventDataTransferStalled).
+		From(storagemarket.StorageDealTransferring).ToJustRecord().Action(func(deal *storagemarket.ClientDeal) error {
+		deal.Message = "data transfer appears to be stalled. attempt restart"
+		return nil
+	}),
+
 	fsm.Event(storagemarket.ClientEventDataTransferComplete).
 		FromMany(storagemarket.StorageDealTransferring, storagemarket.StorageDealStartDataTransfer).To(storagemarket.StorageDealCheckForAcceptance),
 	fsm.Event(storagemarket.ClientEventWaitForDealState).

storagemarket/impl/dtutils/dtutils.go

@@ -51,6 +51,11 @@ func ProviderDataTransferSubscriber(deals EventReceiver) datatransfer.Subscriber
 			if err != nil {
 				log.Errorf("processing dt event: %w", err)
 			}
+		case datatransfer.Disconnected:
+			err := deals.Send(voucher.Proposal, storagemarket.ProviderEventDataTransferStalled)
+			if err != nil {
+				log.Errorf("processing dt event: %w", err)
+			}
 		case datatransfer.Open:
 			err := deals.Send(voucher.Proposal, storagemarket.ProviderEventDataTransferInitiated, channelState.ChannelID())
 			if err != nil {
@@ -92,6 +97,11 @@ func ClientDataTransferSubscriber(deals EventReceiver) datatransfer.Subscriber {
 			if err != nil {
 				log.Errorf("processing dt event: %w", err)
 			}
+		case datatransfer.Disconnected:
+			err := deals.Send(voucher.Proposal, storagemarket.ClientEventDataTransferStalled)
+			if err != nil {
+				log.Errorf("processing dt event: %w", err)
+			}
 		case datatransfer.Accept:
 			err := deals.Send(voucher.Proposal, storagemarket.ClientEventDataTransferInitiated, channelState.ChannelID())
 			if err != nil {

storagemarket/impl/dtutils/dtutils_test.go

@@ -62,6 +62,16 @@ func TestProviderDataTransferSubscriber(t *testing.T) {
 			expectedEvent: storagemarket.ProviderEventDataTransferRestarted,
 			expectedArgs:  []interface{}{datatransfer.ChannelID{Initiator: init, Responder: resp, ID: tid}},
 		},
+		"disconnected event": {
+			code:   datatransfer.Disconnected,
+			status: datatransfer.Ongoing,
+			called: true,
+			voucher: &requestvalidation.StorageDataTransferVoucher{
+				Proposal: expectedProposalCID,
+			},
+			expectedID:    expectedProposalCID,
+			expectedEvent: storagemarket.ProviderEventDataTransferStalled,
+		},
 		"completion status": {
 			code:   datatransfer.Complete,
 			status: datatransfer.Completed,
@@ -164,6 +174,16 @@ func TestClientDataTransferSubscriber(t *testing.T) {
 			expectedEvent: storagemarket.ClientEventDataTransferRestarted,
 			expectedArgs:  []interface{}{datatransfer.ChannelID{Initiator: init, Responder: resp, ID: tid}},
 		},
+		"disconnected event": {
+			code:   datatransfer.Disconnected,
+			status: datatransfer.Ongoing,
+			called: true,
+			voucher: &requestvalidation.StorageDataTransferVoucher{
+				Proposal: expectedProposalCID,
+			},
+			expectedID:    expectedProposalCID,
+			expectedEvent: storagemarket.ClientEventDataTransferStalled,
+		},
 		"accept event": {
 			code:   datatransfer.Accept,
 			status: datatransfer.Requested,

storagemarket/impl/providerstates/provider_fsm.go

@@ -58,9 +58,16 @@ var ProviderEvents = fsm.Events{
 		From(storagemarket.StorageDealTransferring).ToJustRecord().
 		Action(func(deal *storagemarket.MinerDeal, channelId datatransfer.ChannelID) error {
 			deal.TransferChannelId = &channelId
+			deal.Message = ""
 			return nil
 		}),
 
+	fsm.Event(storagemarket.ProviderEventDataTransferStalled).
+		From(storagemarket.StorageDealTransferring).ToJustRecord().Action(func(deal *storagemarket.MinerDeal) error {
+		deal.Message = "data transfer appears to be stalled. attempt restart"
+		return nil
+	}),
+
 	fsm.Event(storagemarket.ProviderEventDataTransferCompleted).
 		From(storagemarket.StorageDealTransferring).To(storagemarket.StorageDealVerifyData),
 	fsm.Event(storagemarket.ProviderEventDataVerificationFailed).