Merge branch 'master' into feat/cid-to-piece-idx

node/builder_chain.go

@@ -121,7 +121,7 @@ var ChainNode = Options(
 	// Markets (retrieval)
 	Override(new(discovery.PeerResolver), modules.RetrievalResolver),
 	Override(new(retrievalmarket.BlockstoreAccessor), modules.RetrievalBlockstoreAccessor),
-	Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient),
+	Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient(false)),
 	Override(new(dtypes.ClientDataTransfer), modules.NewClientGraphsyncDataTransfer),
 
 	// Markets (storage)
@@ -223,6 +223,8 @@ func ConfigFullNode(c interface{}) Option {
 		),
 		Override(new(dtypes.Graphsync), modules.Graphsync(cfg.Client.SimultaneousTransfersForStorage, cfg.Client.SimultaneousTransfersForRetrieval)),
 
+		Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient(cfg.Client.OffChainRetrieval)),
+
 		If(cfg.Wallet.RemoteBackend != "",
 			Override(new(*remotewallet.RemoteWallet), remotewallet.SetupRemoteWallet(cfg.Wallet.RemoteBackend)),
 		),

node/config/def_test.go

@@ -55,5 +55,6 @@ func TestDefaultMinerRoundtrip(t *testing.T) {
 
 func TestDefaultStorageMiner_SetsIndexIngestTopic(t *testing.T) {
 	subject := DefaultStorageMiner()
-	require.Equal(t, "/indexer/ingest/mainnet", subject.IndexProvider.PubSubTopic)
+	require.False(t, subject.IndexProvider.Enable)
+	require.Equal(t, "/indexer/ingest/mainnet", subject.IndexProvider.TopicName)
 }

node/config/doc_gen.go

@@ -105,6 +105,14 @@ and storage providers for storage deals`,
 			Comment: `The maximum number of simultaneous data transfers between the client
 and storage providers for retrieval deals`,
 		},
+		{
+			Name: "OffChainRetrieval",
+			Type: "bool",
+
+			Comment: `Require that retrievals perform no on-chain operations. Paid retrievals
+without existing payment channels with available funds will fail instead
+of automatically performing on-chain operations.`,
+		},
 	},
 	"Common": []DocField{
 		{

node/config/types.go

@@ -417,6 +417,11 @@ type Client struct {
 	// The maximum number of simultaneous data transfers between the client
 	// and storage providers for retrieval deals
 	SimultaneousTransfersForRetrieval uint64
+
+	// Require that retrievals perform no on-chain operations. Paid retrievals
+	// without existing payment channels with available funds will fail instead
+	// of automatically performing on-chain operations.
+	OffChainRetrieval bool
 }
 
 type Wallet struct {

node/impl/paych/paych.go

@@ -22,8 +22,26 @@ type PaychAPI struct {
 	PaychMgr *paychmgr.Manager
 }
 
-func (a *PaychAPI) PaychGet(ctx context.Context, from, to address.Address, amt types.BigInt) (*api.ChannelInfo, error) {
-	ch, mcid, err := a.PaychMgr.GetPaych(ctx, from, to, amt)
+func (a *PaychAPI) PaychGet(ctx context.Context, from, to address.Address, amt types.BigInt, opts api.PaychGetOpts) (*api.ChannelInfo, error) {
+	ch, mcid, err := a.PaychMgr.GetPaych(ctx, from, to, amt, paychmgr.GetOpts{
+		Reserve:  true,
+		OffChain: opts.OffChain,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &api.ChannelInfo{
+		Channel:      ch,
+		WaitSentinel: mcid,
+	}, nil
+}
+
+func (a *PaychAPI) PaychFund(ctx context.Context, from, to address.Address, amt types.BigInt) (*api.ChannelInfo, error) {
+	ch, mcid, err := a.PaychMgr.GetPaych(ctx, from, to, amt, paychmgr.GetOpts{
+		Reserve:  false,
+		OffChain: false,
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +73,7 @@ func (a *PaychAPI) PaychNewPayment(ctx context.Context, from, to address.Address
 
 	// TODO: Fix free fund tracking in PaychGet
 	// TODO: validate voucher spec before locking funds
-	ch, err := a.PaychGet(ctx, from, to, amount)
+	ch, err := a.PaychGet(ctx, from, to, amount, api.PaychGetOpts{OffChain: false})
 	if err != nil {
 		return nil, err
 	}

node/modules/client.go

@@ -202,26 +202,28 @@ func StorageClient(lc fx.Lifecycle, h host.Host, dataTransfer dtypes.ClientDataT
 }
 
 // RetrievalClient creates a new retrieval client attached to the client blockstore
-func RetrievalClient(lc fx.Lifecycle, h host.Host, r repo.LockedRepo, dt dtypes.ClientDataTransfer, payAPI payapi.PaychAPI, resolver discovery.PeerResolver,
+func RetrievalClient(forceOffChain bool) func(lc fx.Lifecycle, h host.Host, r repo.LockedRepo, dt dtypes.ClientDataTransfer, payAPI payapi.PaychAPI, resolver discovery.PeerResolver,
 	ds dtypes.MetadataDS, chainAPI full.ChainAPI, stateAPI full.StateAPI, accessor retrievalmarket.BlockstoreAccessor, j journal.Journal) (retrievalmarket.RetrievalClient, error) {
+	return func(lc fx.Lifecycle, h host.Host, r repo.LockedRepo, dt dtypes.ClientDataTransfer, payAPI payapi.PaychAPI, resolver discovery.PeerResolver,
+		ds dtypes.MetadataDS, chainAPI full.ChainAPI, stateAPI full.StateAPI, accessor retrievalmarket.BlockstoreAccessor, j journal.Journal) (retrievalmarket.RetrievalClient, error) {
+		adapter := retrievaladapter.NewRetrievalClientNode(forceOffChain, payAPI, chainAPI, stateAPI)
+		network := rmnet.NewFromLibp2pHost(h)
+		ds = namespace.Wrap(ds, datastore.NewKey("/retrievals/client"))
+		client, err := retrievalimpl.NewClient(network, dt, adapter, resolver, ds, accessor)
+		if err != nil {
+			return nil, err
+		}
+		client.OnReady(marketevents.ReadyLogger("retrieval client"))
+		lc.Append(fx.Hook{
+			OnStart: func(ctx context.Context) error {
+				client.SubscribeToEvents(marketevents.RetrievalClientLogger)
 
-	adapter := retrievaladapter.NewRetrievalClientNode(payAPI, chainAPI, stateAPI)
-	network := rmnet.NewFromLibp2pHost(h)
-	ds = namespace.Wrap(ds, datastore.NewKey("/retrievals/client"))
-	client, err := retrievalimpl.NewClient(network, dt, adapter, resolver, ds, accessor)
-	if err != nil {
-		return nil, err
+				evtType := j.RegisterEventType("markets/retrieval/client", "state_change")
+				client.SubscribeToEvents(markets.RetrievalClientJournaler(j, evtType))
+
+				return client.Start(ctx)
+			},
+		})
+		return client, nil
 	}
-	client.OnReady(marketevents.ReadyLogger("retrieval client"))
-	lc.Append(fx.Hook{
-		OnStart: func(ctx context.Context) error {
-			client.SubscribeToEvents(marketevents.RetrievalClientLogger)
-
-			evtType := j.RegisterEventType("markets/retrieval/client", "state_change")
-			client.SubscribeToEvents(markets.RetrievalClientJournaler(j, evtType))
-
-			return client.Start(ctx)
-		},
-	})
-	return client, nil
 }

node/modules/storageminer_idxprov.go

@@ -2,14 +2,14 @@ package modules
 
 import (
 	"context"
-	"github.com/ipfs/go-datastore"
-	"go.uber.org/fx"
 
 	"github.com/filecoin-project/go-address"
 	provider "github.com/filecoin-project/index-provider"
 	"github.com/filecoin-project/index-provider/engine"
+	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	"github.com/libp2p/go-libp2p-core/host"
+	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/node/config"