Unique Blockstores: RetrievalMarket

docs/retrievalprovider.mmd

@@ -22,6 +22,7 @@ stateDiagram-v2
 		The following events are not shown cause they can trigger from any state.
 
 		ProviderEventDataTransferError - transitions state to DealStatusErrored
+		ProviderEventMultiStoreError - transitions state to DealStatusErrored
 	end note
 	0 --> 0 : ProviderEventOpen
 	0 --> 1 : ProviderEventDealAccepted

retrievalmarket/client.go

@@ -7,6 +7,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 )
 
@@ -37,6 +38,7 @@ type RetrievalClient interface {
 		miner peer.ID,
 		clientWallet address.Address,
 		minerWallet address.Address,
+		storeID *multistore.StoreID,
 	) (DealID, error)
 
 	// SubscribeToEvents listens for events that happen related to client retrievals

retrievalmarket/events.go

@@ -185,6 +185,9 @@ const (
 
 	// ProviderEventCleanupComplete happens when a deal is finished cleaning up and enters a complete state
 	ProviderEventCleanupComplete
+
+	// ProviderEventMultiStoreError occurs when an error happens attempting to operate on the multistore
+	ProviderEventMultiStoreError
 )
 
 // ProviderEvents is a human readable map of provider event name -> event description
@@ -205,4 +208,5 @@ var ProviderEvents = map[ProviderEvent]string{
 	ProviderEventDataTransferError:      "ProviderEventDataTransferError",
 	ProviderEventCancelComplete:         "ProviderEventCancelComplete",
 	ProviderEventCleanupComplete:        "ProviderEventCleanupComplete",
+	ProviderEventMultiStoreError:        "ProviderEventMultiStoreError",
 }

retrievalmarket/impl/client.go

@@ -7,13 +7,13 @@ import (
 	"github.com/hannahhoward/go-pubsub"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
+	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-statemachine/fsm"
 	"github.com/filecoin-project/go-storedcounter"
 	"github.com/filecoin-project/specs-actors/actors/abi"
@@ -32,7 +32,7 @@ var log = logging.Logger("retrieval")
 type Client struct {
 	network       rmnet.RetrievalMarketNetwork
 	dataTransfer  datatransfer.Manager
-	bs            blockstore.Blockstore
+	multiStore    *multistore.MultiStore
 	node          retrievalmarket.RetrievalClientNode
 	storedCounter *storedcounter.StoredCounter
 
@@ -64,7 +64,7 @@ var _ retrievalmarket.RetrievalClient = &Client{}
 // NewClient creates a new retrieval client
 func NewClient(
 	network rmnet.RetrievalMarketNetwork,
-	bs blockstore.Blockstore,
+	multiStore *multistore.MultiStore,
 	dataTransfer datatransfer.Manager,
 	node retrievalmarket.RetrievalClientNode,
 	resolver retrievalmarket.PeerResolver,
@@ -73,7 +73,7 @@ func NewClient(
 ) (retrievalmarket.RetrievalClient, error) {
 	c := &Client{
 		network:       network,
-		bs:            bs,
+		multiStore:    multiStore,
 		dataTransfer:  dataTransfer,
 		node:          node,
 		resolver:      resolver,
@@ -106,6 +106,10 @@ func NewClient(
 		return nil, err
 	}
 	dataTransfer.SubscribeToEvents(dtutils.ClientDataTransferSubscriber(c.stateMachines))
+	err = dataTransfer.RegisterTransportConfigurer(&retrievalmarket.DealProposal{}, dtutils.TransportConfigurer(network.ID(), &clientStoreGetter{c}))
+	if err != nil {
+		return nil, err
+	}
 	return c, nil
 }
 
@@ -177,14 +181,20 @@ From then on, the statemachine controls the deal flow in the client. Other compo
 
 Documentation of the client state machine can be found at https://godoc.org/github.com/filecoin-project/go-fil-markets/retrievalmarket/impl/clientstates
 */
-func (c *Client) Retrieve(ctx context.Context, payloadCID cid.Cid, params retrievalmarket.Params, totalFunds abi.TokenAmount, miner peer.ID, clientWallet address.Address, minerWallet address.Address) (retrievalmarket.DealID, error) {
+func (c *Client) Retrieve(ctx context.Context, payloadCID cid.Cid, params retrievalmarket.Params, totalFunds abi.TokenAmount, miner peer.ID, clientWallet address.Address, minerWallet address.Address, storeID *multistore.StoreID) (retrievalmarket.DealID, error) {
 	var err error
 	next, err := c.storedCounter.Next()
 	if err != nil {
 		return 0, err
 	}
+	// make sure the store is loadable
+	if storeID != nil {
+		_, err = c.multiStore.Get(*storeID)
+		if err != nil {
+			return 0, err
+		}
+	}
 	dealID := retrievalmarket.DealID(next)
-
 	dealState := retrievalmarket.ClientDealState{
 		DealProposal: retrievalmarket.DealProposal{
 			PayloadCID: payloadCID,
@@ -202,6 +212,7 @@ func (c *Client) Retrieve(ctx context.Context, payloadCID cid.Cid, params retrie
 		Status:           retrievalmarket.DealStatusNew,
 		Sender:           miner,
 		UnsealFundsPaid:  big.Zero(),
+		StoreID:          storeID,
 	}
 
 	// start the deal processing
@@ -286,6 +297,22 @@ func (c *clientDealEnvironment) CloseDataTransfer(ctx context.Context, channelID
 	return c.c.dataTransfer.CloseDataTransferChannel(ctx, channelID)
 }
 
+type clientStoreGetter struct {
+	c *Client
+}
+
+func (csg *clientStoreGetter) Get(otherPeer peer.ID, dealID retrievalmarket.DealID) (*multistore.Store, error) {
+	var deal retrievalmarket.ClientDealState
+	err := csg.c.stateMachines.Get(dealID).Get(&deal)
+	if err != nil {
+		return nil, err
+	}
+	if deal.StoreID == nil {
+		return nil, nil
+	}
+	return csg.c.multiStore.Get(*deal.StoreID)
+}
+
 // ClientFSMParameterSpec is a valid set of parameters for a client deal FSM - used in doc generation
 var ClientFSMParameterSpec = fsm.Parameters{
 	Environment:     &clientDealEnvironment{},

retrievalmarket/impl/client_test.go

@@ -7,12 +7,12 @@ import (
 
 	"github.com/ipfs/go-datastore"
 	dss "github.com/ipfs/go-datastore/sync"
-	bstore "github.com/ipfs/go-ipfs-blockstore"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-storedcounter"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 
@@ -28,12 +28,13 @@ func TestClient_Construction(t *testing.T) {
 
 	ds := dss.MutexWrap(datastore.NewMapDatastore())
 	storedCounter := storedcounter.New(ds, datastore.NewKey("nextDealID"))
-	bs := bstore.NewBlockstore(ds)
+	multiStore, err := multistore.NewMultiDstore(ds)
+	require.NoError(t, err)
 	dt := tut.NewTestDataTransfer()
 	net := tut.NewTestRetrievalMarketNetwork(tut.TestNetworkParams{})
-	_, err := retrievalimpl.NewClient(
+	_, err = retrievalimpl.NewClient(
 		net,
-		bs,
+		multiStore,
 		dt,
 		testnodes.NewTestRetrievalClientNode(testnodes.TestRetrievalClientNodeParams{}),
 		&tut.TestPeerResolver{},
@@ -50,14 +51,18 @@ func TestClient_Construction(t *testing.T) {
 	require.True(t, ok)
 	_, ok = dt.RegisteredVoucherTypes[1].VoucherType.(*retrievalmarket.DealPayment)
 	require.True(t, ok)
+	require.Len(t, dt.RegisteredTransportConfigurers, 1)
+	_, ok = dt.RegisteredTransportConfigurers[0].VoucherType.(*retrievalmarket.DealProposal)
+	require.True(t, ok)
 }
 
 func TestClient_Query(t *testing.T) {
 	ctx := context.Background()
 
 	ds := dss.MutexWrap(datastore.NewMapDatastore())
 	storedCounter := storedcounter.New(ds, datastore.NewKey("nextDealID"))
-	bs := bstore.NewBlockstore(ds)
+	multiStore, err := multistore.NewMultiDstore(ds)
+	require.NoError(t, err)
 	dt := tut.NewTestDataTransfer()
 
 	pcid := tut.GenerateCids(1)[0]
@@ -92,7 +97,7 @@ func TestClient_Query(t *testing.T) {
 		})
 		c, err := retrievalimpl.NewClient(
 			net,
-			bs,
+			multiStore,
 			dt,
 			testnodes.NewTestRetrievalClientNode(testnodes.TestRetrievalClientNodeParams{}),
 			&tut.TestPeerResolver{},
@@ -112,7 +117,7 @@ func TestClient_Query(t *testing.T) {
 		})
 		c, err := retrievalimpl.NewClient(
 			net,
-			bs,
+			multiStore,
 			dt,
 			testnodes.NewTestRetrievalClientNode(testnodes.TestRetrievalClientNodeParams{}),
 			&tut.TestPeerResolver{},
@@ -139,7 +144,7 @@ func TestClient_Query(t *testing.T) {
 		})
 		c, err := retrievalimpl.NewClient(
 			net,
-			bs,
+			multiStore,
 			dt,
 			testnodes.NewTestRetrievalClientNode(testnodes.TestRetrievalClientNodeParams{}),
 			&tut.TestPeerResolver{},
@@ -165,7 +170,7 @@ func TestClient_Query(t *testing.T) {
 		})
 		c, err := retrievalimpl.NewClient(
 			net,
-			bs,
+			multiStore,
 			dt,
 			testnodes.NewTestRetrievalClientNode(testnodes.TestRetrievalClientNodeParams{}),
 			&tut.TestPeerResolver{},
@@ -182,7 +187,8 @@ func TestClient_Query(t *testing.T) {
 func TestClient_FindProviders(t *testing.T) {
 	ds := dss.MutexWrap(datastore.NewMapDatastore())
 	storedCounter := storedcounter.New(ds, datastore.NewKey("nextDealID"))
-	bs := bstore.NewBlockstore(ds)
+	multiStore, err := multistore.NewMultiDstore(ds)
+	require.NoError(t, err)
 	dt := tut.NewTestDataTransfer()
 	expectedPeer := peer.ID("somevalue")
 
@@ -200,7 +206,7 @@ func TestClient_FindProviders(t *testing.T) {
 		peers := tut.RequireGenerateRetrievalPeers(t, 3)
 		testResolver := tut.TestPeerResolver{Peers: peers}
 
-		c, err := retrievalimpl.NewClient(net, bs, dt, &testnodes.TestRetrievalClientNode{}, &testResolver, ds, storedCounter)
+		c, err := retrievalimpl.NewClient(net, multiStore, dt, &testnodes.TestRetrievalClientNode{}, &testResolver, ds, storedCounter)
 		require.NoError(t, err)
 
 		testCid := tut.GenerateCids(1)[0]
@@ -209,7 +215,7 @@ func TestClient_FindProviders(t *testing.T) {
 
 	t.Run("when there is an error, returns empty provider list", func(t *testing.T) {
 		testResolver := tut.TestPeerResolver{Peers: []retrievalmarket.RetrievalPeer{}, ResolverError: errors.New("boom")}
-		c, err := retrievalimpl.NewClient(net, bs, dt, &testnodes.TestRetrievalClientNode{}, &testResolver, ds, storedCounter)
+		c, err := retrievalimpl.NewClient(net, multiStore, dt, &testnodes.TestRetrievalClientNode{}, &testResolver, ds, storedCounter)
 		require.NoError(t, err)
 
 		badCid := tut.GenerateCids(1)[0]
@@ -218,7 +224,7 @@ func TestClient_FindProviders(t *testing.T) {
 
 	t.Run("when there are no providers", func(t *testing.T) {
 		testResolver := tut.TestPeerResolver{Peers: []retrievalmarket.RetrievalPeer{}}
-		c, err := retrievalimpl.NewClient(net, bs, dt, &testnodes.TestRetrievalClientNode{}, &testResolver, ds, storedCounter)
+		c, err := retrievalimpl.NewClient(net, multiStore, dt, &testnodes.TestRetrievalClientNode{}, &testResolver, ds, storedCounter)
 		require.NoError(t, err)
 
 		testCid := tut.GenerateCids(1)[0]

retrievalmarket/impl/dtutils/dtutils.go

@@ -7,8 +7,11 @@ import (
 	"math"
 
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/ipld/go-ipld-prime"
+	peer "github.com/libp2p/go-libp2p-core/peer"
 
 	datatransfer "github.com/filecoin-project/go-data-transfer"
+	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-statemachine/fsm"
 
 	rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
@@ -134,3 +137,41 @@ func ClientDataTransferSubscriber(deals EventReceiver) datatransfer.Subscriber {
 		}
 	}
 }
+
+// StoreGetter retrieves the store for a given proposal cid
+type StoreGetter interface {
+	Get(otherPeer peer.ID, dealID rm.DealID) (*multistore.Store, error)
+}
+
+// StoreConfigurableTransport defines the methods needed to
+// configure a data transfer transport use a unique store for a given request
+type StoreConfigurableTransport interface {
+	UseStore(datatransfer.ChannelID, ipld.Loader, ipld.Storer) error
+}
+
+// TransportConfigurer configurers the graphsync transport to use a custom blockstore per deal
+func TransportConfigurer(thisPeer peer.ID, storeGetter StoreGetter) datatransfer.TransportConfigurer {
+	return func(channelID datatransfer.ChannelID, voucher datatransfer.Voucher, transport datatransfer.Transport) {
+		dealProposal, ok := voucher.(*rm.DealProposal)
+		if !ok {
+			return
+		}
+		gsTransport, ok := transport.(StoreConfigurableTransport)
+		if !ok {
+			return
+		}
+		otherPeer := channelID.OtherParty(thisPeer)
+		store, err := storeGetter.Get(otherPeer, dealProposal.ID)
+		if err != nil {
+			log.Errorf("attempting to configure data store: %w", err)
+			return
+		}
+		if store == nil {
+			return
+		}
+		err = gsTransport.UseStore(channelID, store.Loader, store.Storer)
+		if err != nil {
+			log.Errorf("attempting to configure data store: %w", err)
+		}
+	}
+}