Track deals funding for deals that are being negotiated

docs/storageclient.mmd

@@ -57,6 +57,15 @@ stateDiagram-v2
 	7 --> 8 : ClientEventDealExpired
 	7 --> 26 : ClientEventDealCompletionFailed
 	11 --> 26 : ClientEventFailed
+
+	note left of 21 : The following events only record in this state.<br><br>ClientEventFundsReserved
+
+
+	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
+
 	9 --> [*]
 	8 --> [*]
 	26 --> [*]

docs/storageprovider.mmd

@@ -78,6 +78,16 @@ stateDiagram-v2
 	10 --> 26 : ProviderEventRestart
 	14 --> 26 : ProviderEventRestart
 	15 --> 26 : ProviderEventRestart
+	20 --> 11 : ProviderEventTrackFundsFailed
+
+	note left of 20 : The following events only record in this state.<br><br>ProviderEventFundsReserved
+
+
+	note left of 11 : The following events only record in this state.<br><br>ProviderEventFundsReleased
+
+
+	note left of 25 : The following events only record in this state.<br><br>ProviderEventFundsReleased
+
 	26 --> [*]
 	9 --> [*]
 	8 --> [*]

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/filecoin-project/go-data-transfer v0.5.1
 	github.com/filecoin-project/go-multistore v0.0.2
 	github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6
-	github.com/filecoin-project/go-statemachine v0.0.0-20200714194326-a77c3ae20989
+	github.com/filecoin-project/go-statemachine v0.0.0-20200730031800-c3336614d2a7
 	github.com/filecoin-project/go-statestore v0.1.0
 	github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b
 	github.com/filecoin-project/sector-storage v0.0.0-20200615154852-728a47ab99d6

go.sum

@@ -122,6 +122,8 @@ github.com/filecoin-project/go-paramfetch v0.0.1 h1:gV7bs5YaqlgpGFMiLxInGK2L1FyC
 github.com/filecoin-project/go-paramfetch v0.0.1/go.mod h1:fZzmf4tftbwf9S37XRifoJlz7nCjRdIrMGLR07dKLCc=
 github.com/filecoin-project/go-statemachine v0.0.0-20200714194326-a77c3ae20989 h1:1GjCS3xy/CRIw7Tq0HfzX6Al8mklrszQZ3iIFnjPzHk=
 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-20200730031800-c3336614d2a7 h1:KAF3WM/xSnl6G6RHX8vDJthg4+e4PSgBh72//6c6Qvc=
+github.com/filecoin-project/go-statemachine v0.0.0-20200730031800-c3336614d2a7/go.mod h1:FGwQgZAt2Gh5mjlwJUlVB62JeYdo+if0xWxSEfBD9ig=
 github.com/filecoin-project/go-statestore v0.1.0 h1:t56reH59843TwXHkMcwyuayStBIiWBRilQjQ+5IiwdQ=
 github.com/filecoin-project/go-statestore v0.1.0/go.mod h1:LFc9hD+fRxPqiHiaqUEZOinUJB4WARkRfNl10O7kTnI=
 github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b h1:fkRZSPrYpk42PV3/lIXiL0LHetxde7vyYYvSsttQtfg=

retrievalmarket/storage_retrieval_integration_test.go

@@ -40,6 +40,7 @@ import (
 	tut "github.com/filecoin-project/go-fil-markets/shared_testutil"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	stormkt "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
+	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask"
 	stornet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
@@ -262,6 +263,9 @@ func newStorageHarness(ctx context.Context, t *testing.T) *storageHarness {
 
 	peerResolver := discovery.NewLocal(td.Ds1)
 
+	clientDealFunds, err := funds.NewDealFunds(td.Ds1, datastore.NewKey("storage/client/dealfunds"))
+	require.NoError(t, err)
+
 	client, err := stormkt.NewClient(
 		stornet.NewFromLibp2pHost(td.Host1),
 		td.Bs1,
@@ -270,6 +274,7 @@ func newStorageHarness(ctx context.Context, t *testing.T) *storageHarness {
 		peerResolver,
 		td.Ds1,
 		&clientNode,
+		clientDealFunds,
 		stormkt.DealPollingInterval(0),
 	)
 	require.NoError(t, err)
@@ -281,9 +286,11 @@ func newStorageHarness(ctx context.Context, t *testing.T) *storageHarness {
 	require.NoError(t, err)
 	rv2 := requestvalidation.NewUnifiedRequestValidator(statestore.New(td.Ds2), nil)
 	require.NoError(t, dt2.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, rv2))
-
 	storedAsk, err := storedask.NewStoredAsk(td.Ds2, datastore.NewKey("latest-ask"), providerNode, providerAddr)
 	require.NoError(t, err)
+	providerDealFunds, err := funds.NewDealFunds(td.Ds1, datastore.NewKey("storage/provider/dealfunds"))
+	require.NoError(t, err)
+
 	provider, err := stormkt.NewProvider(
 		stornet.NewFromLibp2pHost(td.Host2),
 		td.Ds2,
@@ -295,6 +302,7 @@ func newStorageHarness(ctx context.Context, t *testing.T) *storageHarness {
 		providerAddr,
 		abi.RegisteredSealProof_StackedDrg2KiBV1,
 		storedAsk,
+		providerDealFunds,
 	)
 	require.NoError(t, err)
 

shared_testutil/test_deal_funds.go

@@ -0,0 +1,38 @@
+package shared_testutil
+
+import (
+	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
+
+	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
+)
+
+func NewTestDealFunds() *TestDealFunds {
+	return &TestDealFunds{
+		reserved: big.Zero(),
+	}
+}
+
+type TestDealFunds struct {
+	reserved     abi.TokenAmount
+	ReserveCalls []abi.TokenAmount
+	ReleaseCalls []abi.TokenAmount
+}
+
+func (f *TestDealFunds) Get() abi.TokenAmount {
+	return f.reserved
+}
+
+func (f *TestDealFunds) Reserve(amount abi.TokenAmount) (abi.TokenAmount, error) {
+	f.reserved = big.Add(f.reserved, amount)
+	f.ReserveCalls = append(f.ReserveCalls, amount)
+	return f.reserved, nil
+}
+
+func (f *TestDealFunds) Release(amount abi.TokenAmount) (abi.TokenAmount, error) {
+	f.reserved = big.Sub(f.reserved, amount)
+	f.ReleaseCalls = append(f.ReleaseCalls, amount)
+	return f.reserved, nil
+}
+
+var _ funds.DealFunds = &TestDealFunds{}

storagemarket/events.go

@@ -13,6 +13,12 @@ const (
 	// ClientEventFundingInitiated happens when a client has sent a message adding funds to its balance
 	ClientEventFundingInitiated
 
+	// ClientEventFundsReserved happens when a client reserves funds for a deal (updating our tracked funds)
+	ClientEventFundsReserved
+
+	// ClientEventFundsReleased happens when a client released funds for a deal (updating our tracked funds)
+	ClientEventFundsReleased
+
 	// ClientEventFundsEnsured happens when a client successfully ensures it has funds for a deal
 	ClientEventFundsEnsured
 
@@ -88,6 +94,8 @@ var ClientEvents = map[ClientEvent]string{
 	ClientEventOpen:                       "ClientEventOpen",
 	ClientEventEnsureFundsFailed:          "ClientEventEnsureFundsFailed",
 	ClientEventFundingInitiated:           "ClientEventFundingInitiated",
+	ClientEventFundsReserved:              "ClientEventFundsReserved",
+	ClientEventFundsReleased:              "ClientEventFundsReleased",
 	ClientEventFundsEnsured:               "ClientEventFundsEnsured",
 	ClientEventWriteProposalFailed:        "ClientEventWriteProposalFailed",
 	ClientEventInitiateDataTransfer:       "ClientEventInitiateDataTransfer",
@@ -138,6 +146,12 @@ const (
 	// ProviderEventInsufficientFunds indicates not enough funds available for a deal
 	ProviderEventInsufficientFunds
 
+	// ProviderEventFundsReserved indicates we've reserved funds for a deal, adding to our overall total
+	ProviderEventFundsReserved
+
+	// ProviderEventFundsReleased indicates we've released funds for a deal
+	ProviderEventFundsReleased
+
 	// ProviderEventFundingInitiated indicates provider collateral funding has been initiated
 	ProviderEventFundingInitiated
 
@@ -220,6 +234,9 @@ const (
 	// ProviderEventFailed indicates a deal has failed and should no longer be processed
 	ProviderEventFailed
 
+	// ProviderEventTrackFundsFailed indicates a failure trying to locally track funds needed for deals
+	ProviderEventTrackFundsFailed
+
 	// ProviderEventRestart is used to resume the deal after a state machine shutdown
 	ProviderEventRestart
 )
@@ -233,6 +250,8 @@ var ProviderEvents = map[ProviderEvent]string{
 	ProviderEventDealAccepted:           "ProviderEventDealAccepted",
 	ProviderEventDealDeciding:           "ProviderEventDealDeciding",
 	ProviderEventInsufficientFunds:      "ProviderEventInsufficientFunds",
+	ProviderEventFundsReserved:          "ProviderEventFundsReserved",
+	ProviderEventFundsReleased:          "ProviderEventFundsReleased",
 	ProviderEventFundingInitiated:       "ProviderEventFundingInitiated",
 	ProviderEventFunded:                 "ProviderEventFunded",
 	ProviderEventDataTransferFailed:     "ProviderEventDataTransferFailed",
@@ -260,5 +279,6 @@ var ProviderEvents = map[ProviderEvent]string{
 	ProviderEventDealExpired:            "ProviderEventDealExpired",
 	ProviderEventDealSlashed:            "ProviderEventDealSlashed",
 	ProviderEventFailed:                 "ProviderEventFailed",
+	ProviderEventTrackFundsFailed:       "ProviderEventTrackFundsFailed",
 	ProviderEventRestart:                "ProviderEventRestart",
 }

storagemarket/impl/client.go

@@ -33,6 +33,7 @@ import (
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/clientstates"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/clientutils"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/dtutils"
+	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/network"
 )
@@ -56,6 +57,7 @@ type Client struct {
 	pubSub          *pubsub.PubSub
 	statemachines   fsm.Group
 	pollingInterval time.Duration
+	dealFunds       funds.DealFunds
 }
 
 // StorageClientOption allows custom configuration of a storage client
@@ -78,6 +80,7 @@ func NewClient(
 	discovery *discovery.Local,
 	ds datastore.Batching,
 	scn storagemarket.StorageClientNode,
+	dealFunds funds.DealFunds,
 	options ...StorageClientOption,
 ) (*Client, error) {
 	carIO := cario.NewCarIO()
@@ -91,6 +94,7 @@ func NewClient(
 		pio:             pio,
 		pubSub:          pubsub.New(clientDispatcher),
 		pollingInterval: DefaultPollingInterval,
+		dealFunds:       dealFunds,
 	}
 
 	statemachines, err := newClientStateMachine(
@@ -544,6 +548,10 @@ func (c *clientDealEnvironment) PollingInterval() time.Duration {
 	return c.c.pollingInterval
 }
 
+func (c *clientDealEnvironment) DealFunds() funds.DealFunds {
+	return c.c.dealFunds
+}
+
 type clientStoreGetter struct {
 	c *Client
 }

storagemarket/impl/clientstates/client_fsm.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/filecoin-project/go-statemachine/fsm"
 	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
 
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 )
@@ -26,6 +27,22 @@ var ClientEvents = fsm.Events{
 			deal.Message = xerrors.Errorf("adding market funds failed: %w", err).Error()
 			return nil
 		}),
+	fsm.Event(storagemarket.ClientEventFundsReserved).
+		From(storagemarket.StorageDealEnsureClientFunds).ToJustRecord().
+		Action(func(deal *storagemarket.ClientDeal, fundsReserved abi.TokenAmount) error {
+			if deal.FundsReserved.Nil() {
+				deal.FundsReserved = fundsReserved
+			} else {
+				deal.FundsReserved = big.Add(deal.FundsReserved, fundsReserved)
+			}
+			return nil
+		}),
+	fsm.Event(storagemarket.ClientEventFundsReleased).
+		FromMany(storagemarket.StorageDealProposalAccepted, storagemarket.StorageDealFailing).ToJustRecord().
+		Action(func(deal *storagemarket.ClientDeal, fundsReleased abi.TokenAmount) error {
+			deal.FundsReserved = big.Subtract(deal.FundsReserved, fundsReleased)
+			return nil
+		}),
 	fsm.Event(storagemarket.ClientEventFundsEnsured).
 		FromMany(storagemarket.StorageDealEnsureClientFunds, storagemarket.StorageDealClientFunding).To(storagemarket.StorageDealFundsEnsured),
 	fsm.Event(storagemarket.ClientEventWriteProposalFailed).
@@ -67,9 +84,9 @@ var ClientEvents = fsm.Events{
 	fsm.Event(storagemarket.ClientEventWaitForDealState).
 		From(storagemarket.StorageDealCheckForAcceptance).ToNoChange().
 		Action(func(deal *storagemarket.ClientDeal, pollError bool) error {
-			deal.PollRetryCount += 1
+			deal.PollRetryCount++
 			if pollError {
-				deal.PollErrorCount += 1
+				deal.PollErrorCount++
 			}
 			return nil
 		}),

storagemarket/impl/clientstates/client_states.go

@@ -18,6 +18,7 @@ import (
 	"github.com/filecoin-project/go-fil-markets/shared"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/clientutils"
+	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/network"
 )
@@ -32,6 +33,7 @@ type ClientDealEnvironment interface {
 	StartDataTransfer(ctx context.Context, to peer.ID, voucher datatransfer.Voucher, baseCid cid.Cid, selector ipld.Node) error
 	GetProviderDealState(ctx context.Context, proposalCid cid.Cid) (*storagemarket.ProviderDealState, error)
 	PollingInterval() time.Duration
+	DealFunds() funds.DealFunds
 }
 
 // ClientStateEntryFunc is the type for all state entry functions on a storage client
@@ -46,7 +48,19 @@ func EnsureClientFunds(ctx fsm.Context, environment ClientDealEnvironment, deal
 		return ctx.Trigger(storagemarket.ClientEventEnsureFundsFailed, xerrors.Errorf("acquiring chain head: %w", err))
 	}
 
-	mcid, err := node.EnsureFunds(ctx.Context(), deal.Proposal.Client, deal.Proposal.Client, deal.Proposal.ClientBalanceRequirement(), tok)
+	var requiredFunds abi.TokenAmount
+	if deal.FundsReserved.Nil() || deal.FundsReserved.IsZero() {
+		requiredFunds, err = environment.DealFunds().Reserve(deal.Proposal.ClientBalanceRequirement())
+		if err != nil {
+			return ctx.Trigger(storagemarket.ClientEventEnsureFundsFailed, xerrors.Errorf("tracking deal funds: %w", err))
+		}
+	} else {
+		requiredFunds = environment.DealFunds().Get()
+	}
+
+	_ = ctx.Trigger(storagemarket.ClientEventFundsReserved, deal.Proposal.ClientBalanceRequirement())
+
+	mcid, err := node.EnsureFunds(ctx.Context(), deal.Proposal.Client, deal.Proposal.Client, requiredFunds, tok)
 
 	if err != nil {
 		return ctx.Trigger(storagemarket.ClientEventEnsureFundsFailed, err)
@@ -191,6 +205,15 @@ func ValidateDealPublished(ctx fsm.Context, environment ClientDealEnvironment, d
 		return ctx.Trigger(storagemarket.ClientEventDealPublishFailed, err)
 	}
 
+	if !deal.FundsReserved.Nil() && !deal.FundsReserved.IsZero() {
+		_, err = environment.DealFunds().Release(deal.FundsReserved)
+		if err != nil {
+			// nonfatal error
+			log.Warnf("failed to release funds from local tracker: %s", err)
+		}
+		_ = ctx.Trigger(storagemarket.ClientEventFundsReleased, deal.FundsReserved)
+	}
+
 	return ctx.Trigger(storagemarket.ClientEventDealPublished, dealID)
 }
 
@@ -242,6 +265,15 @@ func WaitForDealCompletion(ctx fsm.Context, environment ClientDealEnvironment, d
 
 // FailDeal cleans up a failing deal
 func FailDeal(ctx fsm.Context, environment ClientDealEnvironment, deal storagemarket.ClientDeal) error {
+	if !deal.FundsReserved.Nil() && !deal.FundsReserved.IsZero() {
+		_, err := environment.DealFunds().Release(deal.FundsReserved)
+		if err != nil {
+			// nonfatal error
+			log.Warnf("failed to release funds from local tracker: %s", err)
+		}
+		_ = ctx.Trigger(storagemarket.ClientEventFundsReleased, deal.FundsReserved)
+	}
+
 	// TODO: store in some sort of audit log
 	log.Errorf("deal %s failed: %s", deal.ProposalCid, deal.Message)