Flush out & fix retrieval bugs

go.mod

@@ -6,7 +6,7 @@ require (
 	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-commp-utils v0.0.0-20201119054358-b88f7a96a434
-	github.com/filecoin-project/go-data-transfer v1.4.1
+	github.com/filecoin-project/go-data-transfer v1.4.2
 	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
@@ -43,7 +43,7 @@ require (
 	github.com/stretchr/testify v1.6.1
 	github.com/whyrusleeping/cbor-gen v0.0.0-20210219115102-f37d292932f2
 	golang.org/x/exp v0.0.0-20200207192155-f17229e696bd
-	golang.org/x/net v0.0.0-20200625001655-4c5254603344
+	golang.org/x/net v0.0.0-20201021035429-f5854403a974
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1
 )
 

retrievalmarket/impl/client_test.go

@@ -341,7 +341,7 @@ func TestClient_DuplicateRetrieve(t *testing.T) {
 
 				select {
 				case <-done:
-				case <-time.After(100 * time.Millisecond):
+				case <-time.After(500 * time.Millisecond):
 				}
 			}
 

retrievalmarket/impl/clientstates/client_fsm.go

@@ -264,13 +264,17 @@ var ClientEvents = fsm.Events{
 			deal.FundsSpent = big.Add(deal.FundsSpent, deal.PaymentRequested)
 
 			paymentForUnsealing := big.Min(deal.PaymentRequested, big.Sub(deal.UnsealPrice, deal.UnsealFundsPaid))
+			deal.UnsealFundsPaid = big.Add(deal.UnsealFundsPaid, paymentForUnsealing)
 
-			bytesPaidFor := big.Div(big.Sub(deal.PaymentRequested, paymentForUnsealing), deal.PricePerByte).Uint64()
-			if bytesPaidFor >= deal.CurrentInterval {
-				deal.CurrentInterval += deal.DealProposal.PaymentIntervalIncrease
+			// If the price per bytes is zero, we ONLY need to account for the Unsealing payments here.
+			if !deal.PricePerByte.IsZero() {
+				bytesPaidFor := big.Div(big.Sub(deal.PaymentRequested, paymentForUnsealing), deal.PricePerByte).Uint64()
+				if bytesPaidFor >= deal.CurrentInterval {
+					deal.CurrentInterval += deal.DealProposal.PaymentIntervalIncrease
+				}
+				deal.BytesPaidFor += bytesPaidFor
 			}
-			deal.BytesPaidFor += bytesPaidFor
-			deal.UnsealFundsPaid = big.Add(deal.UnsealFundsPaid, paymentForUnsealing)
+
 			deal.PaymentRequested = abi.NewTokenAmount(0)
 			return nil
 		}),

retrievalmarket/impl/clientstates/client_states_test.go

@@ -373,6 +373,27 @@ func TestSendFunds(t *testing.T) {
 		require.Equal(t, dealState.Status, retrievalmarket.DealStatusFinalizing)
 	})
 
+	t.Run("only unsealing payment is accounted for when price per bytes is zero", func(t *testing.T) {
+		dealState := makeDealState(retrievalmarket.DealStatusSendFundsLastPayment)
+		unsealPrice := abi.NewTokenAmount(100)
+
+		dealState.UnsealPrice = unsealPrice
+		dealState.PaymentRequested = unsealPrice
+		dealState.PricePerByte = abi.NewTokenAmount(0)
+
+		var sendVoucherError error = nil
+		nodeParams := testnodes.TestRetrievalClientNodeParams{
+			Voucher: testVoucher,
+		}
+		runSendFunds(t, sendVoucherError, nodeParams, dealState)
+		require.Empty(t, dealState.Message)
+		require.Equal(t, dealState.PaymentRequested, abi.NewTokenAmount(0))
+		require.Equal(t, dealState.FundsSpent, big.Add(defaultFundsSpent, unsealPrice))
+		require.Equal(t, dealState.BytesPaidFor, defaultBytesPaidFor)
+		require.Equal(t, dealState.CurrentInterval, defaultCurrentInterval)
+		require.Equal(t, dealState.Status, retrievalmarket.DealStatusFinalizing)
+	})
+
 	t.Run("more bytes since last payment than interval works, can charge more", func(t *testing.T) {
 		dealState := makeDealState(retrievalmarket.DealStatusSendFunds)
 		dealState.BytesPaidFor = defaultBytesPaidFor - 500

retrievalmarket/impl/providerstates/provider_fsm.go

@@ -81,7 +81,11 @@ var ProviderEvents = fsm.Events{
 		From(rm.DealStatusFundsNeededUnseal).To(rm.DealStatusUnsealing).
 		Action(func(deal *rm.ProviderDealState, fundsReceived abi.TokenAmount) error {
 			deal.FundsReceived = big.Add(deal.FundsReceived, fundsReceived)
-			deal.CurrentInterval += deal.PaymentIntervalIncrease
+
+			// only update interval if the payment is for bytes and not for unsealing.
+			if deal.Status != rm.DealStatusFundsNeededUnseal {
+				deal.CurrentInterval += deal.PaymentIntervalIncrease
+			}
 			return nil
 		}),
 

retrievalmarket/impl/requestvalidation/requestvalidation.go

@@ -106,9 +106,10 @@ func (rv *ProviderRequestValidator) validatePull(receiver peer.ID, proposal *ret
 	}
 
 	pds := retrievalmarket.ProviderDealState{
-		DealProposal:   *proposal,
-		Receiver:       receiver,
-		LegacyProtocol: legacyProtocol,
+		DealProposal:    *proposal,
+		Receiver:        receiver,
+		LegacyProtocol:  legacyProtocol,
+		CurrentInterval: proposal.PaymentInterval,
 	}
 
 	status, err := rv.acceptDeal(&pds)

retrievalmarket/impl/requestvalidation/revalidator.go

@@ -131,7 +131,7 @@ func (pr *ProviderRevalidator) Revalidate(channelID datatransfer.ChannelID, vouc
 	}
 
 	response, err := pr.processPayment(channel.dealID, payment)
-	if err == nil {
+	if err == nil || err == datatransfer.ErrResume {
 		channel.reload = true
 	}
 	return finalResponse(response, legacyProtocol), err
@@ -178,13 +178,21 @@ func (pr *ProviderRevalidator) processPayment(dealID rm.ProviderDealIdentifier,
 
 	// resume deal
 	_ = pr.env.SendEvent(dealID, rm.ProviderEventPaymentReceived, received)
+
 	if deal.Status == rm.DealStatusFundsNeededLastPayment {
 		return &rm.DealResponse{
 			ID:     deal.ID,
 			Status: rm.DealStatusCompleted,
-		}, nil
+		}, datatransfer.ErrResume
 	}
-	return nil, nil
+
+	// We shouldn't resume the data transfer if we haven't finished unsealing/reading the unsealed data into the
+	// local block-store.
+	if deal.Status == rm.DealStatusUnsealing || deal.Status == rm.DealStatusFundsNeededUnseal {
+		return nil, nil
+	}
+
+	return nil, datatransfer.ErrResume
 }
 
 func errorDealResponse(dealID rm.ProviderDealIdentifier, err error) *rm.DealResponse {

retrievalmarket/impl/requestvalidation/revalidator_test.go

@@ -446,6 +446,7 @@ func TestRevalidate(t *testing.T) {
 			expectedID:    deal.Identifier(),
 			expectedEvent: rm.ProviderEventPaymentReceived,
 			expectedArgs:  []interface{}{defaultPaymentPerInterval},
+			expectedError: datatransfer.ErrResume,
 		},
 
 		"it completes": {
@@ -458,6 +459,7 @@ func TestRevalidate(t *testing.T) {
 			expectedID:    deal.Identifier(),
 			expectedEvent: rm.ProviderEventPaymentReceived,
 			expectedArgs:  []interface{}{defaultPaymentPerInterval},
+			expectedError: datatransfer.ErrResume,
 			expectedResult: &rm.DealResponse{
 				ID:     deal.ID,
 				Status: rm.DealStatusCompleted,
@@ -471,6 +473,7 @@ func TestRevalidate(t *testing.T) {
 			channelID:     channelID,
 			voucher:       legacyPayment,
 			expectedID:    deal.Identifier(),
+			expectedError: datatransfer.ErrResume,
 			expectedEvent: rm.ProviderEventPaymentReceived,
 			expectedArgs:  []interface{}{defaultPaymentPerInterval},
 			expectedResult: &migrations.DealResponse0{
@@ -486,6 +489,7 @@ func TestRevalidate(t *testing.T) {
 			channelID:     channelID,
 			voucher:       payment,
 			expectedID:    deal.Identifier(),
+			expectedError: datatransfer.ErrResume,
 			expectedEvent: rm.ProviderEventPaymentReceived,
 			expectedArgs:  []interface{}{defaultPaymentPerInterval},
 		},