cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #4263 from filecoin-project/feat/lite-market-client

Browse Source 
lite-mode - market storage and retrieval clients
This commit is contained in:
Łukasz Magiera
2020-10-23 17:23:06 +02:00
committed by GitHub
parent 91fae36b86 36816fb4f3
commit dbcba6477e
26 changed files with 763 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/lotus-gateway/api.go
+92 -23
View File
@@ -24,7 +24,7 @@ import (
const (
LookbackCap = time.Hour * 24
stateWaitLookbackLimit = abi.ChainEpoch(20)
StateWaitLookbackLimit = abi.ChainEpoch(20)
)
var (
@@ -35,20 +35,28 @@ var (
// (to make it easy to mock for tests)
type gatewayDepsAPI interface {
Version(context.Context) (api.Version, error)
ChainHasObj(context.Context, cid.Cid) (bool, error)
ChainHead(ctx context.Context) (*types.TipSet, error)
ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error)
ChainGetNode(ctx context.Context, p string) (*api.IpldObject, error)
ChainGetTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error)
ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error)
ChainHasObj(context.Context, cid.Cid) (bool, error)
ChainHead(ctx context.Context) (*types.TipSet, error)
ChainNotify(context.Context) (<-chan []*api.HeadChange, error)
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
ChainGetNode(ctx context.Context, p string) (*api.IpldObject, error)
GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error)
MpoolPushUntrusted(ctx context.Context, sm *types.SignedMessage) (cid.Cid, error)
MsigGetAvailableBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (types.BigInt, error)
MsigGetVested(ctx context.Context, addr address.Address, start types.TipSetKey, end types.TipSetKey) (types.BigInt, error)
StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
StateDealProviderCollateralBounds(ctx context.Context, size abi.PaddedPieceSize, verified bool, tsk types.TipSetKey) (api.DealCollateralBounds, error)
StateGetActor(ctx context.Context, actor address.Address, ts types.TipSetKey) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error)
StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
StateListMiners(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error)
StateMarketBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MarketBalance, error)
StateMarketStorageDeal(ctx context.Context, dealId abi.DealID, tsk types.TipSetKey) (*api.MarketDeal, error)
StateNetworkVersion(context.Context, types.TipSetKey) (network.Version, error)
StateWaitMsgLimited(ctx context.Context, msg cid.Cid, confidence uint64, h abi.ChainEpoch) (*api.MsgLookup, error)
StateReadState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*api.ActorState, error)
StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
@@ -59,23 +67,24 @@ type gatewayDepsAPI interface {
StateMinerAvailableBalance(context.Context, address.Address, types.TipSetKey) (types.BigInt, error)
StateMinerProvingDeadline(context.Context, address.Address, types.TipSetKey) (*dline.Info, error)
StateCirculatingSupply(context.Context, types.TipSetKey) (abi.TokenAmount, error)
StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error)
StateVMCirculatingSupplyInternal(context.Context, types.TipSetKey) (api.CirculatingSupply, error)
StateNetworkVersion(context.Context, types.TipSetKey) (network.Version, error)
}
type GatewayAPI struct {
api gatewayDepsAPI
lookbackCap time.Duration
api gatewayDepsAPI
lookbackCap time.Duration
stateWaitLookbackLimit abi.ChainEpoch
}
// NewGatewayAPI creates a new GatewayAPI with the default lookback cap
func NewGatewayAPI(api gatewayDepsAPI) *GatewayAPI {
return newGatewayAPI(api, LookbackCap)
return newGatewayAPI(api, LookbackCap, StateWaitLookbackLimit)
}
// used by the tests
func newGatewayAPI(api gatewayDepsAPI, lookbackCap time.Duration) *GatewayAPI {
return &GatewayAPI{api: api, lookbackCap: lookbackCap}
func newGatewayAPI(api gatewayDepsAPI, lookbackCap time.Duration, stateWaitLookbackLimit abi.ChainEpoch) *GatewayAPI {
return &GatewayAPI{api: api, lookbackCap: lookbackCap, stateWaitLookbackLimit: stateWaitLookbackLimit}
}
func (a *GatewayAPI) checkTipsetKey(ctx context.Context, tsk types.TipSetKey) error {
@@ -122,6 +131,10 @@ func (a *GatewayAPI) Version(ctx context.Context) (api.Version, error) {
return a.api.Version(ctx)
}
func (a *GatewayAPI) ChainGetBlockMessages(ctx context.Context, c cid.Cid) (*api.BlockMessages, error) {
return a.api.ChainGetBlockMessages(ctx, c)
}
func (a *GatewayAPI) ChainHasObj(ctx context.Context, c cid.Cid) (bool, error) {
return a.api.ChainHasObj(ctx, c)
}
@@ -132,6 +145,10 @@ func (a *GatewayAPI) ChainHead(ctx context.Context) (*types.TipSet, error) {
return a.api.ChainHead(ctx)
}
func (a *GatewayAPI) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error) {
return a.api.ChainGetMessage(ctx, mc)
}
func (a *GatewayAPI) ChainGetTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error) {
return a.api.ChainGetTipSet(ctx, tsk)
}
@@ -165,14 +182,18 @@ func (a *GatewayAPI) ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoc
return a.api.ChainGetTipSetByHeight(ctx, h, tsk)
}
func (a *GatewayAPI) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
return a.api.ChainReadObj(ctx, c)
}
func (a *GatewayAPI) ChainGetNode(ctx context.Context, p string) (*api.IpldObject, error) {
return a.api.ChainGetNode(ctx, p)
}
func (a *GatewayAPI) ChainNotify(ctx context.Context) (<-chan []*api.HeadChange, error) {
return a.api.ChainNotify(ctx)
}
func (a *GatewayAPI) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
return a.api.ChainReadObj(ctx, c)
}
func (a *GatewayAPI) GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
@@ -213,6 +234,14 @@ func (a *GatewayAPI) StateAccountKey(ctx context.Context, addr address.Address,
return a.api.StateAccountKey(ctx, addr, tsk)
}
func (a *GatewayAPI) StateDealProviderCollateralBounds(ctx context.Context, size abi.PaddedPieceSize, verified bool, tsk types.TipSetKey) (api.DealCollateralBounds, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return api.DealCollateralBounds{}, err
}
return a.api.StateDealProviderCollateralBounds(ctx, size, verified, tsk)
}
func (a *GatewayAPI) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
@@ -221,6 +250,22 @@ func (a *GatewayAPI) StateGetActor(ctx context.Context, actor address.Address, t
return a.api.StateGetActor(ctx, actor, tsk)
}
func (a *GatewayAPI) StateGetReceipt(ctx context.Context, c cid.Cid, tsk types.TipSetKey) (*types.MessageReceipt, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return a.api.StateGetReceipt(ctx, c, tsk)
}
func (a *GatewayAPI) StateListMiners(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return a.api.StateListMiners(ctx, tsk)
}
func (a *GatewayAPI) StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return address.Undef, err
@@ -229,8 +274,32 @@ func (a *GatewayAPI) StateLookupID(ctx context.Context, addr address.Address, ts
return a.api.StateLookupID(ctx, addr, tsk)
}
func (a *GatewayAPI) StateMarketBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MarketBalance, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return api.MarketBalance{}, err
}
return a.api.StateMarketBalance(ctx, addr, tsk)
}
func (a *GatewayAPI) StateMarketStorageDeal(ctx context.Context, dealId abi.DealID, tsk types.TipSetKey) (*api.MarketDeal, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return a.api.StateMarketStorageDeal(ctx, dealId, tsk)
}
func (a *GatewayAPI) StateNetworkVersion(ctx context.Context, tsk types.TipSetKey) (network.Version, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return network.VersionMax, err
}
return a.api.StateNetworkVersion(ctx, tsk)
}
func (a *GatewayAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error) {
return a.api.StateWaitMsgLimited(ctx, msg, confidence, stateWaitLookbackLimit)
return a.api.StateWaitMsgLimited(ctx, msg, confidence, a.stateWaitLookbackLimit)
}
func (a *GatewayAPI) StateReadState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*api.ActorState, error) {
@@ -296,6 +365,13 @@ func (a *GatewayAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSe
}
func (a *GatewayAPI) StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return a.api.StateVerifiedClientStatus(ctx, addr, tsk)
}
func (a *GatewayAPI) StateVMCirculatingSupplyInternal(ctx context.Context, tsk types.TipSetKey) (api.CirculatingSupply, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return api.CirculatingSupply{}, err
@@ -303,13 +379,6 @@ func (a *GatewayAPI) StateVMCirculatingSupplyInternal(ctx context.Context, tsk t
return a.api.StateVMCirculatingSupplyInternal(ctx, tsk)
}
func (a *GatewayAPI) StateNetworkVersion(ctx context.Context, tsk types.TipSetKey) (network.Version, error) {
if err := a.checkTipsetKey(ctx, tsk); err != nil {
return 0, err
}
return a.api.StateNetworkVersion(ctx, tsk)
}
func (a *GatewayAPI) WalletVerify(ctx context.Context, k address.Address, msg []byte, sig *crypto.Signature) (bool, error) {
return sigs.Verify(sig, k, msg) == nil, nil
}
cmd/lotus-gateway/api_test.go
+35 -4
View File
@@ -6,6 +6,9 @@ import (
"testing"
"time"
"github.com/filecoin-project/go-state-types/network"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"github.com/filecoin-project/lotus/build"
"github.com/stretchr/testify/require"
@@ -116,6 +119,38 @@ func (m *mockGatewayDepsAPI) ChainHasObj(context.Context, cid.Cid) (bool, error)
panic("implement me")
}
func (m *mockGatewayDepsAPI) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) StateDealProviderCollateralBounds(ctx context.Context, size abi.PaddedPieceSize, verified bool, tsk types.TipSetKey) (api.DealCollateralBounds, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) StateListMiners(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) StateMarketBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MarketBalance, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) StateMarketStorageDeal(ctx context.Context, dealId abi.DealID, tsk types.TipSetKey) (*api.MarketDeal, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) StateMinerInfo(ctx context.Context, actor address.Address, tsk types.TipSetKey) (miner.MinerInfo, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) StateNetworkVersion(ctx context.Context, key types.TipSetKey) (network.Version, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) ChainHead(ctx context.Context) (*types.TipSet, error) {
m.lk.RLock()
defer m.lk.RUnlock()
@@ -165,10 +200,6 @@ func (m *mockGatewayDepsAPI) ChainGetTipSetByHeight(ctx context.Context, h abi.C
return m.tipsets[h], nil
}
func (m *mockGatewayDepsAPI) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
panic("implement me")
}
func (m *mockGatewayDepsAPI) GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error) {
panic("implement me")
}
cmd/lotus-gateway/endtoend_test.go
+107 -43
View File
@@ -25,12 +25,14 @@ import (
"github.com/filecoin-project/lotus/api/client"
"github.com/filecoin-project/lotus/api/test"
"github.com/filecoin-project/lotus/chain/actors/policy"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node"
builder "github.com/filecoin-project/lotus/node/test"
)
const maxLookbackCap = time.Duration(math.MaxInt64)
const maxStateWaitLookbackLimit = stmgr.LookbackNoLimit
func init() {
policy.SetSupportedProofTypes(abi.RegisteredSealProof_StackedDrg2KiBV1)
@@ -38,15 +40,19 @@ func init() {
policy.SetMinVerifiedDealSize(abi.NewStoragePower(256))
}
// TestEndToEndWalletMsig tests that wallet and msig API calls can be made
// on a lite node that is connected through a gateway to a full API node
func TestEndToEndWalletMsig(t *testing.T) {
// TestWalletMsig tests that API calls to wallet and msig can be made on a lite
// node that is connected through a gateway to a full API node
func TestWalletMsig(t *testing.T) {
_ = os.Setenv("BELLMAN_NO_GPU", "1")
clitest.QuietMiningLogs()
blocktime := 5 * time.Millisecond
ctx := context.Background()
full, lite, closer := startNodes(ctx, t, blocktime, maxLookbackCap)
defer closer()
nodes := startNodes(ctx, t, blocktime, maxLookbackCap, maxStateWaitLookbackLimit)
defer nodes.closer()
lite := nodes.lite
full := nodes.full
// The full node starts with a wallet
fullWalletAddr, err := full.WalletDefaultAddress(ctx)
@@ -141,56 +147,83 @@ func TestEndToEndWalletMsig(t *testing.T) {
require.True(t, approveReturn.Applied)
}
// TestEndToEndMsigCLI tests that msig CLI calls can be made
// TestMsigCLI tests that msig CLI calls can be made
// on a lite node that is connected through a gateway to a full API node
func TestEndToEndMsigCLI(t *testing.T) {
func TestMsigCLI(t *testing.T) {
_ = os.Setenv("BELLMAN_NO_GPU", "1")
clitest.QuietMiningLogs()
blocktime := 5 * time.Millisecond
ctx := context.Background()
full, lite, closer := startNodes(ctx, t, blocktime, maxLookbackCap)
defer closer()
// The full node starts with a wallet
fullWalletAddr, err := full.WalletDefaultAddress(ctx)
require.NoError(t, err)
// Create a wallet on the lite node
liteWalletAddr, err := lite.WalletNew(ctx, types.KTSecp256k1)
require.NoError(t, err)
// Send some funds from the full node to the lite node
err = sendFunds(ctx, full, fullWalletAddr, liteWalletAddr, types.NewInt(1e18))
require.NoError(t, err)
nodes := startNodesWithFunds(ctx, t, blocktime, maxLookbackCap, maxStateWaitLookbackLimit)
defer nodes.closer()
lite := nodes.lite
clitest.RunMultisigTest(t, cli.Commands, lite)
}
func sendFunds(ctx context.Context, fromNode test.TestNode, fromAddr address.Address, toAddr address.Address, amt types.BigInt) error {
msg := &types.Message{
From: fromAddr,
To: toAddr,
Value: amt,
}
func TestDealFlow(t *testing.T) {
_ = os.Setenv("BELLMAN_NO_GPU", "1")
clitest.QuietMiningLogs()
sm, err := fromNode.MpoolPushMessage(ctx, msg, nil)
if err != nil {
return err
}
blocktime := 5 * time.Millisecond
ctx := context.Background()
nodes := startNodesWithFunds(ctx, t, blocktime, maxLookbackCap, maxStateWaitLookbackLimit)
defer nodes.closer()
res, err := fromNode.StateWaitMsg(ctx, sm.Cid(), 1)
if err != nil {
return err
}
if res.Receipt.ExitCode != 0 {
return xerrors.Errorf("send funds failed with exit code %d", res.Receipt.ExitCode)
}
return nil
test.MakeDeal(t, ctx, 6, nodes.lite, nodes.miner, false, false)
}
func startNodes(ctx context.Context, t *testing.T, blocktime time.Duration, lookbackCap time.Duration) (test.TestNode, test.TestNode, jsonrpc.ClientCloser) {
func TestCLIDealFlow(t *testing.T) {
_ = os.Setenv("BELLMAN_NO_GPU", "1")
clitest.QuietMiningLogs()
blocktime := 5 * time.Millisecond
ctx := context.Background()
nodes := startNodesWithFunds(ctx, t, blocktime, maxLookbackCap, maxStateWaitLookbackLimit)
defer nodes.closer()
clitest.RunClientTest(t, cli.Commands, nodes.lite)
}
type testNodes struct {
lite test.TestNode
full test.TestNode
miner test.TestStorageNode
closer jsonrpc.ClientCloser
}
func startNodesWithFunds(
ctx context.Context,
t *testing.T,
blocktime time.Duration,
lookbackCap time.Duration,
stateWaitLookbackLimit abi.ChainEpoch,
) *testNodes {
nodes := startNodes(ctx, t, blocktime, lookbackCap, stateWaitLookbackLimit)
// The full node starts with a wallet
fullWalletAddr, err := nodes.full.WalletDefaultAddress(ctx)
require.NoError(t, err)
// Create a wallet on the lite node
liteWalletAddr, err := nodes.lite.WalletNew(ctx, types.KTSecp256k1)
require.NoError(t, err)
// Send some funds from the full node to the lite node
err = sendFunds(ctx, nodes.full, fullWalletAddr, liteWalletAddr, types.NewInt(1e18))
require.NoError(t, err)
return nodes
}
func startNodes(
ctx context.Context,
t *testing.T,
blocktime time.Duration,
lookbackCap time.Duration,
stateWaitLookbackLimit abi.ChainEpoch,
) *testNodes {
var closer jsonrpc.ClientCloser
// Create one miner and two full nodes.
@@ -207,7 +240,8 @@ func startNodes(ctx context.Context, t *testing.T, blocktime time.Duration, look
fullNode := nodes[0]
// Create a gateway server in front of the full node
_, addr, err := builder.CreateRPCServer(newGatewayAPI(fullNode, lookbackCap))
gapiImpl := newGatewayAPI(fullNode, lookbackCap, stateWaitLookbackLimit)
_, addr, err := builder.CreateRPCServer(gapiImpl)
require.NoError(t, err)
// Create a gateway client API that connects to the gateway server
@@ -234,9 +268,39 @@ func startNodes(ctx context.Context, t *testing.T, blocktime time.Duration, look
err = miner.NetConnect(ctx, fullAddr)
require.NoError(t, err)
// Connect the miner and the lite node (so that the lite node can send
// data to the miner)
liteAddr, err := lite.NetAddrsListen(ctx)
require.NoError(t, err)
err = miner.NetConnect(ctx, liteAddr)
require.NoError(t, err)
// Start mining blocks
bm := test.NewBlockMiner(ctx, t, miner, blocktime)
bm.MineBlocks()
return full, lite, closer
return &testNodes{lite: lite, full: full, miner: miner, closer: closer}
}
func sendFunds(ctx context.Context, fromNode test.TestNode, fromAddr address.Address, toAddr address.Address, amt types.BigInt) error {
msg := &types.Message{
From: fromAddr,
To: toAddr,
Value: amt,
}
sm, err := fromNode.MpoolPushMessage(ctx, msg, nil)
if err != nil {
return err
}
res, err := fromNode.StateWaitMsg(ctx, sm.Cid(), 1)
if err != nil {
return err
}
if res.Receipt.ExitCode != 0 {
return xerrors.Errorf("send funds failed with exit code %d", res.Receipt.ExitCode)
}
return nil
}