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

Merge pull request #2195 from filecoin-project/feat/update-markets-0.3.2

Browse Source 
Markets update 0.3.2
This commit is contained in:
Łukasz Magiera 2020-07-01 12:19:14 +02:00 committed by GitHub
commit f48681a61d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 157 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
chain/events/state/predicates.go
View File

@ -2,7 +2,6 @@ package state
import (
"context"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-amt-ipld/v2"
"github.com/filecoin-project/lotus/api/apibstore"
@ -107,8 +106,8 @@ type ChangedDeals map[abi.DealID]DealStateChange
// DealStateChange is a change in deal state from -> to
type DealStateChange struct {
From market.DealState
To market.DealState
From *market.DealState
To *market.DealState
}
// DealStateChangedForIDs detects changes in the deal state AMT for the given deal IDs
@ -116,17 +115,25 @@ func (sp *StatePredicates) DealStateChangedForIDs(dealIds []abi.DealID) DiffDeal
return func(ctx context.Context, oldDealStateRoot *amt.Root, newDealStateRoot *amt.Root) (changed bool, user UserData, err error) {
changedDeals := make(ChangedDeals)
for _, dealID := range dealIds {
var oldDealPtr, newDealPtr *market.DealState
var oldDeal, newDeal market.DealState
// If the deal has been removed, we just set it to nil
err := oldDealStateRoot.Get(ctx, uint64(dealID), &oldDeal)
if err != nil {
if err == nil {
oldDealPtr = &oldDeal
} else if _, ok := err.(*amt.ErrNotFound); !ok {
return false, nil, err
}
err = newDealStateRoot.Get(ctx, uint64(dealID), &newDeal)
if err != nil {
if err == nil {
newDealPtr = &newDeal
} else if _, ok := err.(*amt.ErrNotFound); !ok {
return false, nil, err
}
if oldDeal != newDeal {
changedDeals[dealID] = DealStateChange{oldDeal, newDeal}
changedDeals[dealID] = DealStateChange{oldDealPtr, newDealPtr}
}
}
if len(changedDeals) > 0 {

19
chain/events/state/predicates_test.go
View File

@ -71,12 +71,10 @@ func TestPredicates(t *testing.T) {
abi.DealID(1): {
SectorStartEpoch: 1,
LastUpdatedEpoch: 2,
SlashEpoch: 0,
},
abi.DealID(2): {
SectorStartEpoch: 4,
LastUpdatedEpoch: 5,
SlashEpoch: 0,
},
}
oldStateC := createMarketState(ctx, t, store, oldDeals)
@ -85,12 +83,6 @@ func TestPredicates(t *testing.T) {
abi.DealID(1): {
SectorStartEpoch: 1,
LastUpdatedEpoch: 3,
SlashEpoch: 0,
},
abi.DealID(2): {
SectorStartEpoch: 4,
LastUpdatedEpoch: 6,
SlashEpoch: 6,
},
}
newStateC := createMarketState(ctx, t, store, newDeals)
@ -131,10 +123,17 @@ func TestPredicates(t *testing.T) {
t.Fatal("Unexpected change to LastUpdatedEpoch")
}
deal2 := changedDeals[abi.DealID(2)]
if deal2.From.SlashEpoch != 0 || deal2.To.SlashEpoch != 6 {
t.Fatal("Unexpected change to SlashEpoch")
if deal2.From.LastUpdatedEpoch != 5 || deal2.To != nil {
t.Fatal("Expected To to be nil")
}
// Diff with non-existent deal.
noDeal := []abi.DealID{3}
diffNoDealFn := preds.OnStorageMarketActorChanged(preds.OnDealStateChanged(preds.DealStateChangedForIDs(noDeal)))
changed, _, err = diffNoDealFn(ctx, oldState, newState)
require.NoError(t, err)
require.False(t, changed)
// Test that OnActorStateChanged does not call the callback if the state has not changed
mockAddr, err := address.NewFromString("t01")
require.NoError(t, err)

2
go.mod
View File

@ -23,7 +23,7 @@ require (
github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03
github.com/filecoin-project/go-data-transfer v0.3.0
github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5
github.com/filecoin-project/go-fil-markets v0.3.1
github.com/filecoin-project/go-fil-markets v0.3.2
github.com/filecoin-project/go-jsonrpc v0.1.1-0.20200602181149-522144ab4e24
github.com/filecoin-project/go-paramfetch v0.0.2-0.20200605171344-fcac609550ca
github.com/filecoin-project/go-statestore v0.1.0

8
go.sum
View File

@ -235,8 +235,8 @@ github.com/filecoin-project/go-data-transfer v0.3.0 h1:BwBrrXu9Unh9JjjX4GAc5FfzU
github.com/filecoin-project/go-data-transfer v0.3.0/go.mod h1:cONglGP4s/d+IUQw5mWZrQK+FQATQxr3AXzi4dRh0l4=
github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5 h1:yvQJCW9mmi9zy+51xA01Ea2X7/dL7r8eKDPuGUjRmbo=
github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5/go.mod h1:JbkIgFF/Z9BDlvrJO1FuKkaWsH673/UdFaiVS6uIHlA=
github.com/filecoin-project/go-fil-markets v0.3.1 h1:YLH4ck4hQrKBpQ3fo0VcA2SXqiAosizxBJ/QHYgR9aE=
github.com/filecoin-project/go-fil-markets v0.3.1/go.mod h1:UY+/zwNXHN73HcrN6HxNDpv6KKM6ehqfCuE9vK9khF8=
github.com/filecoin-project/go-fil-markets v0.3.2 h1:fvNgdTTIVtckBu61wxbKYSMJzedoFFIKYJagiCDFCiM=
github.com/filecoin-project/go-fil-markets v0.3.2/go.mod h1:e/IofcotbwH7ftgeK+TjjdjFsrCDWrh5vvnr7k1OSH8=
github.com/filecoin-project/go-jsonrpc v0.1.1-0.20200602181149-522144ab4e24 h1:Jc7vkplmZYVuaEcSXGHDwefvZIdoyyaoGDLqSr8Svms=
github.com/filecoin-project/go-jsonrpc v0.1.1-0.20200602181149-522144ab4e24/go.mod h1:j6zV//WXIIY5kky873Q3iIKt/ViOE8rcijovmpxrXzM=
github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6 h1:92PET+sx1Hb4W/8CgFwGuxaKbttwY+UNspYZTvXY0vs=
@ -247,8 +247,8 @@ github.com/filecoin-project/go-paramfetch v0.0.2-0.20200605171344-fcac609550ca h
github.com/filecoin-project/go-paramfetch v0.0.2-0.20200605171344-fcac609550ca/go.mod h1:fZzmf4tftbwf9S37XRifoJlz7nCjRdIrMGLR07dKLCc=
github.com/filecoin-project/go-statemachine v0.0.0-20200226041606-2074af6d51d9 h1:k9qVR9ItcziSB2rxtlkN/MDWNlbsI6yzec+zjUatLW0=
github.com/filecoin-project/go-statemachine v0.0.0-20200226041606-2074af6d51d9/go.mod h1:FGwQgZAt2Gh5mjlwJUlVB62JeYdo+if0xWxSEfBD9ig=
github.com/filecoin-project/go-statemachine v0.0.0-20200612181802-4eb3d0c68eba h1:GEWb/6KQyNZt4jm8fgVcIFPH0ElAGXfHM59ZSiqPTvY=
github.com/filecoin-project/go-statemachine v0.0.0-20200612181802-4eb3d0c68eba/go.mod h1:FGwQgZAt2Gh5mjlwJUlVB62JeYdo+if0xWxSEfBD9ig=
github.com/filecoin-project/go-statemachine v0.0.0-20200619205156-c7bf525c06ef h1:qFXGHKLq49qFmvXjvhvQ2eU3jVk2Z0QaKYQpO5S3SF0=
github.com/filecoin-project/go-statemachine v0.0.0-20200619205156-c7bf525c06ef/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=

156
markets/storageadapter/client.go
View File

@ -6,6 +6,9 @@ import (
"bytes"
"context"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/events/state"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
@ -61,13 +64,13 @@ func NewClientNodeAdapter(state full.StateAPI, chain full.ChainAPI, mpool full.M
}
}
func (n *ClientNodeAdapter) ListStorageProviders(ctx context.Context, encodedTs shared.TipSetToken) ([]*storagemarket.StorageProviderInfo, error) {
func (c *ClientNodeAdapter) ListStorageProviders(ctx context.Context, encodedTs shared.TipSetToken) ([]*storagemarket.StorageProviderInfo, error) {
tsk, err := types.TipSetKeyFromBytes(encodedTs)
if err != nil {
return nil, err
}
addresses, err := n.StateListMiners(ctx, tsk)
addresses, err := c.StateListMiners(ctx, tsk)
if err != nil {
return nil, err
}
@ -75,7 +78,7 @@ func (n *ClientNodeAdapter) ListStorageProviders(ctx context.Context, encodedTs
var out []*storagemarket.StorageProviderInfo
for _, addr := range addresses {
mi, err := n.StateMinerInfo(ctx, addr, tsk)
mi, err := c.StateMinerInfo(ctx, addr, tsk)
if err != nil {
return nil, err
}
@ -95,8 +98,8 @@ func (n *ClientNodeAdapter) ListStorageProviders(ctx context.Context, encodedTs
return out, nil
}
func (n *ClientNodeAdapter) VerifySignature(ctx context.Context, sig crypto.Signature, addr address.Address, input []byte, encodedTs shared.TipSetToken) (bool, error) {
addr, err := n.StateAccountKey(ctx, addr, types.EmptyTSK)
func (c *ClientNodeAdapter) VerifySignature(ctx context.Context, sig crypto.Signature, addr address.Address, input []byte, encodedTs shared.TipSetToken) (bool, error) {
addr, err := c.StateAccountKey(ctx, addr, types.EmptyTSK)
if err != nil {
return false, err
}
@ -105,13 +108,13 @@ func (n *ClientNodeAdapter) VerifySignature(ctx context.Context, sig crypto.Sign
return err == nil, err
}
func (n *ClientNodeAdapter) ListClientDeals(ctx context.Context, addr address.Address, encodedTs shared.TipSetToken) ([]storagemarket.StorageDeal, error) {
func (c *ClientNodeAdapter) ListClientDeals(ctx context.Context, addr address.Address, encodedTs shared.TipSetToken) ([]storagemarket.StorageDeal, error) {
tsk, err := types.TipSetKeyFromBytes(encodedTs)
if err != nil {
return nil, err
}
allDeals, err := n.StateMarketDeals(ctx, tsk)
allDeals, err := c.StateMarketDeals(ctx, tsk)
if err != nil {
return nil, err
}
@ -129,9 +132,9 @@ func (n *ClientNodeAdapter) ListClientDeals(ctx context.Context, addr address.Ad
}
// Adds funds with the StorageMinerActor for a storage participant. Used by both providers and clients.
func (n *ClientNodeAdapter) AddFunds(ctx context.Context, addr address.Address, amount abi.TokenAmount) (cid.Cid, error) {
func (c *ClientNodeAdapter) AddFunds(ctx context.Context, addr address.Address, amount abi.TokenAmount) (cid.Cid, error) {
// (Provider Node API)
smsg, err := n.MpoolPushMessage(ctx, &types.Message{
smsg, err := c.MpoolPushMessage(ctx, &types.Message{
To: builtin.StorageMarketActorAddr,
From: addr,
Value: amount,
@ -146,17 +149,17 @@ func (n *ClientNodeAdapter) AddFunds(ctx context.Context, addr address.Address,
return smsg.Cid(), nil
}
func (n *ClientNodeAdapter) EnsureFunds(ctx context.Context, addr, wallet address.Address, amount abi.TokenAmount, ts shared.TipSetToken) (cid.Cid, error) {
return n.fm.EnsureAvailable(ctx, addr, wallet, amount)
func (c *ClientNodeAdapter) EnsureFunds(ctx context.Context, addr, wallet address.Address, amount abi.TokenAmount, ts shared.TipSetToken) (cid.Cid, error) {
return c.fm.EnsureAvailable(ctx, addr, wallet, amount)
}
func (n *ClientNodeAdapter) GetBalance(ctx context.Context, addr address.Address, encodedTs shared.TipSetToken) (storagemarket.Balance, error) {
func (c *ClientNodeAdapter) GetBalance(ctx context.Context, addr address.Address, encodedTs shared.TipSetToken) (storagemarket.Balance, error) {
tsk, err := types.TipSetKeyFromBytes(encodedTs)
if err != nil {
return storagemarket.Balance{}, err
}
bal, err := n.StateMarketBalance(ctx, addr, tsk)
bal, err := c.StateMarketBalance(ctx, addr, tsk)
if err != nil {
return storagemarket.Balance{}, err
}
@ -337,19 +340,103 @@ func (c *ClientNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider
return nil
}
func (n *ClientNodeAdapter) SignProposal(ctx context.Context, signer address.Address, proposal samarket.DealProposal) (*samarket.ClientDealProposal, error) {
func (c *ClientNodeAdapter) OnDealExpiredOrSlashed(ctx context.Context, dealID abi.DealID, onDealExpired storagemarket.DealExpiredCallback, onDealSlashed storagemarket.DealSlashedCallback) error {
var sd *api.MarketDeal
// Called immediately to check if the deal has already expired or been slashed
checkFunc := func(ts *types.TipSet) (done bool, more bool, err error) {
sd, err = stmgr.GetStorageDeal(ctx, c.StateManager, dealID, ts)
if err != nil {
return false, false, xerrors.Errorf("client: failed to look up deal on chain: %w", err)
}
// Check if the deal has already expired
if sd.Proposal.EndEpoch <= ts.Height() {
onDealExpired(nil)
return true, false, nil
}
// If there is no deal assume it's already been slashed
if sd.State.SectorStartEpoch < 0 {
onDealSlashed(ts.Height(), nil)
return true, false, nil
}
// No events have occurred yet, so return
// done: false, more: true (keep listening for events)
return false, true, nil
}
// Called when there was a match against the state change we're looking for
// and the chain has advanced to the confidence height
stateChanged := func(ts *types.TipSet, ts2 *types.TipSet, states events.StateChange, h abi.ChainEpoch) (more bool, err error) {
if states == nil {
log.Error("timed out waiting for deal expiry")
return false, nil
}
// Check if the deal has already expired
if sd.Proposal.EndEpoch <= ts2.Height() {
onDealExpired(nil)
return false, nil
}
changedDeals, ok := states.(state.ChangedDeals)
if !ok {
panic("Expected state.ChangedDeals")
}
deal, ok := changedDeals[dealID]
if !ok {
// No change to deal
return true, nil
}
// Deal was slashed
if deal.To == nil {
onDealSlashed(ts2.Height(), nil)
return false, nil
}
return true, nil
}
// Called when there was a chain reorg and the state change was reverted
revert := func(ctx context.Context, ts *types.TipSet) error {
// TODO: Is it ok to just ignore this?
log.Warn("deal state reverted; TODO: actually handle this!")
return nil
}
// Watch for state changes to the deal
preds := state.NewStatePredicates(c)
dealDiff := preds.OnStorageMarketActorChanged(
preds.OnDealStateChanged(
preds.DealStateChangedForIDs([]abi.DealID{dealID})))
match := func(oldTs, newTs *types.TipSet) (bool, events.StateChange, error) {
return dealDiff(ctx, oldTs, newTs)
}
if err := c.ev.StateChanged(checkFunc, stateChanged, revert, int(build.MessageConfidence)+1, build.SealRandomnessLookbackLimit, match); err != nil {
return xerrors.Errorf("failed to set up state changed handler: %w", err)
}
return nil
}
func (c *ClientNodeAdapter) SignProposal(ctx context.Context, signer address.Address, proposal samarket.DealProposal) (*samarket.ClientDealProposal, error) {
// TODO: output spec signed proposal
buf, err := cborutil.Dump(&proposal)
if err != nil {
return nil, err
}
signer, err = n.StateAccountKey(ctx, signer, types.EmptyTSK)
signer, err = c.StateAccountKey(ctx, signer, types.EmptyTSK)
if err != nil {
return nil, err
}
sig, err := n.Wallet.Sign(ctx, signer, buf)
sig, err := c.Wallet.Sign(ctx, signer, buf)
if err != nil {
return nil, err
}
@ -360,18 +447,18 @@ func (n *ClientNodeAdapter) SignProposal(ctx context.Context, signer address.Add
}, nil
}
func (n *ClientNodeAdapter) GetDefaultWalletAddress(ctx context.Context) (address.Address, error) {
addr, err := n.Wallet.GetDefault()
func (c *ClientNodeAdapter) GetDefaultWalletAddress(ctx context.Context) (address.Address, error) {
addr, err := c.Wallet.GetDefault()
return addr, err
}
func (n *ClientNodeAdapter) ValidateAskSignature(ctx context.Context, ask *storagemarket.SignedStorageAsk, encodedTs shared.TipSetToken) (bool, error) {
func (c *ClientNodeAdapter) ValidateAskSignature(ctx context.Context, ask *storagemarket.SignedStorageAsk, encodedTs shared.TipSetToken) (bool, error) {
tsk, err := types.TipSetKeyFromBytes(encodedTs)
if err != nil {
return false, err
}
mi, err := n.StateMinerInfo(ctx, ask.Ask.Miner, tsk)
mi, err := c.StateMinerInfo(ctx, ask.Ask.Miner, tsk)
if err != nil {
return false, xerrors.Errorf("failed to get worker for miner in ask", err)
}
@ -381,12 +468,12 @@ func (n *ClientNodeAdapter) ValidateAskSignature(ctx context.Context, ask *stora
return false, xerrors.Errorf("failed to re-serialize ask")
}
ts, err := n.ChainGetTipSet(ctx, tsk)
ts, err := c.ChainGetTipSet(ctx, tsk)
if err != nil {
return false, xerrors.Errorf("failed to load tipset")
}
m, err := n.StateManager.ResolveToKeyAddress(ctx, mi.Worker, ts)
m, err := c.StateManager.ResolveToKeyAddress(ctx, mi.Worker, ts)
if err != nil {
return false, xerrors.Errorf("failed to resolve miner to key address")
@ -396,8 +483,8 @@ func (n *ClientNodeAdapter) ValidateAskSignature(ctx context.Context, ask *stora
return err == nil, err
}
func (n *ClientNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetToken, abi.ChainEpoch, error) {
head, err := n.ChainHead(ctx)
func (c *ClientNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetToken, abi.ChainEpoch, error) {
head, err := c.ChainHead(ctx)
if err != nil {
return nil, 0, err
}
@ -405,20 +492,20 @@ func (n *ClientNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetToke
return head.Key().Bytes(), head.Height(), nil
}
func (n *ClientNodeAdapter) WaitForMessage(ctx context.Context, mcid cid.Cid, cb func(code exitcode.ExitCode, bytes []byte, err error) error) error {
receipt, err := n.StateWaitMsg(ctx, mcid, build.MessageConfidence)
func (c *ClientNodeAdapter) WaitForMessage(ctx context.Context, mcid cid.Cid, cb func(code exitcode.ExitCode, bytes []byte, err error) error) error {
receipt, err := c.StateWaitMsg(ctx, mcid, build.MessageConfidence)
if err != nil {
return cb(0, nil, err)
}
return cb(receipt.Receipt.ExitCode, receipt.Receipt.Return, nil)
}
func (n *ClientNodeAdapter) GetMinerInfo(ctx context.Context, addr address.Address, encodedTs shared.TipSetToken) (*storagemarket.StorageProviderInfo, error) {
func (c *ClientNodeAdapter) GetMinerInfo(ctx context.Context, addr address.Address, encodedTs shared.TipSetToken) (*storagemarket.StorageProviderInfo, error) {
tsk, err := types.TipSetKeyFromBytes(encodedTs)
if err != nil {
return nil, err
}
mi, err := n.StateMinerInfo(ctx, addr, tsk)
mi, err := c.StateMinerInfo(ctx, addr, tsk)
if err != nil {
return nil, err
}
@ -434,4 +521,17 @@ func (n *ClientNodeAdapter) GetMinerInfo(ctx context.Context, addr address.Addre
return &out, nil
}
func (c *ClientNodeAdapter) SignBytes(ctx context.Context, signer address.Address, b []byte) (*crypto.Signature, error) {
signer, err := c.StateAccountKey(ctx, signer, types.EmptyTSK)
if err != nil {
return nil, err
}
localSignature, err := c.Wallet.Sign(ctx, signer, b)
if err != nil {
return nil, err
}
return localSignature, nil
}
var _ storagemarket.StorageClientNode = &ClientNodeAdapter{}

2
node/impl/client/client.go
View File

@ -136,6 +136,8 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
params.EpochPrice,
big.Zero(),
rt,
false,
false,
)
if err != nil {