Merge pull request #213 from filecoin-project/feat/state-replay
add state replay api and command
This commit is contained in:
commit
7073676983
@ -108,6 +108,7 @@ type FullNode interface {
|
||||
StateMinerPeerID(ctx context.Context, m address.Address, ts *types.TipSet) (peer.ID, error)
|
||||
// if tipset is nil, we'll use heaviest
|
||||
StateCall(context.Context, *types.Message, *types.TipSet) (*types.MessageReceipt, error)
|
||||
StateReplay(context.Context, *types.TipSet, cid.Cid) (*ReplayResults, error)
|
||||
StateGetActor(ctx context.Context, actor address.Address, ts *types.TipSet) (*types.Actor, error)
|
||||
StateReadState(ctx context.Context, act *types.Actor, ts *types.TipSet) (*ActorState, error)
|
||||
|
||||
@ -269,3 +270,9 @@ type RetrievalOrder struct {
|
||||
Miner address.Address
|
||||
MinerPeerID peer.ID
|
||||
}
|
||||
|
||||
type ReplayResults struct {
|
||||
Msg *types.Message
|
||||
Receipt *types.MessageReceipt
|
||||
Error string
|
||||
}
|
||||
|
@ -81,6 +81,7 @@ type FullNodeStruct struct {
|
||||
StateMinerWorker func(context.Context, address.Address, *types.TipSet) (address.Address, error) `perm:"read"`
|
||||
StateMinerPeerID func(ctx context.Context, m address.Address, ts *types.TipSet) (peer.ID, error) `perm:"read"`
|
||||
StateCall func(context.Context, *types.Message, *types.TipSet) (*types.MessageReceipt, error) `perm:"read"`
|
||||
StateReplay func(context.Context, *types.TipSet, cid.Cid) (*ReplayResults, error) `perm:"read"`
|
||||
StateGetActor func(context.Context, address.Address, *types.TipSet) (*types.Actor, error) `perm:"read"`
|
||||
StateReadState func(context.Context, *types.Actor, *types.TipSet) (*ActorState, error) `perm:"read"`
|
||||
|
||||
@ -302,6 +303,10 @@ func (c *FullNodeStruct) StateCall(ctx context.Context, msg *types.Message, ts *
|
||||
return c.Internal.StateCall(ctx, msg, ts)
|
||||
}
|
||||
|
||||
func (c *FullNodeStruct) StateReplay(ctx context.Context, ts *types.TipSet, mc cid.Cid) (*ReplayResults, error) {
|
||||
return c.Internal.StateReplay(ctx, ts, mc)
|
||||
}
|
||||
|
||||
func (c *FullNodeStruct) StateGetActor(ctx context.Context, actor address.Address, ts *types.TipSet) (*types.Actor, error) {
|
||||
return c.Internal.StateGetActor(ctx, actor, ts)
|
||||
}
|
||||
|
@ -2,6 +2,7 @@ package stmgr
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/filecoin-project/go-lotus/chain/actors"
|
||||
"github.com/filecoin-project/go-lotus/chain/types"
|
||||
@ -59,3 +60,24 @@ func (sm *StateManager) Call(ctx context.Context, msg *types.Message, ts *types.
|
||||
|
||||
return sm.CallRaw(ctx, msg, state, ts.Height())
|
||||
}
|
||||
|
||||
var errHaltExecution = fmt.Errorf("halt")
|
||||
|
||||
func (sm *StateManager) Replay(ctx context.Context, ts *types.TipSet, mcid cid.Cid) (*types.Message, *vm.ApplyRet, error) {
|
||||
var outm *types.Message
|
||||
var outr *vm.ApplyRet
|
||||
|
||||
_, err := sm.computeTipSetState(ctx, ts.Blocks(), func(c cid.Cid, m *types.Message, ret *vm.ApplyRet) error {
|
||||
if c == mcid {
|
||||
outm = m
|
||||
outr = ret
|
||||
return errHaltExecution
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil && err != errHaltExecution {
|
||||
return nil, nil, xerrors.Errorf("unexpected error during execution: %w", err)
|
||||
}
|
||||
|
||||
return outm, outr, nil
|
||||
}
|
||||
|
@ -41,6 +41,8 @@ func cidsToKey(cids []cid.Cid) string {
|
||||
}
|
||||
|
||||
func (sm *StateManager) TipSetState(cids []cid.Cid) (cid.Cid, error) {
|
||||
ctx := context.TODO()
|
||||
|
||||
ck := cidsToKey(cids)
|
||||
sm.stlk.Lock()
|
||||
cached, ok := sm.stCache[ck]
|
||||
@ -49,7 +51,12 @@ func (sm *StateManager) TipSetState(cids []cid.Cid) (cid.Cid, error) {
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
out, err := sm.computeTipSetState(cids)
|
||||
ts, err := sm.cs.LoadTipSet(cids)
|
||||
if err != nil {
|
||||
return cid.Undef, err
|
||||
}
|
||||
|
||||
out, err := sm.computeTipSetState(ctx, ts.Blocks(), nil)
|
||||
if err != nil {
|
||||
return cid.Undef, err
|
||||
}
|
||||
@ -60,31 +67,23 @@ func (sm *StateManager) TipSetState(cids []cid.Cid) (cid.Cid, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (sm *StateManager) computeTipSetState(cids []cid.Cid) (cid.Cid, error) {
|
||||
ctx := context.TODO()
|
||||
|
||||
ts, err := sm.cs.LoadTipSet(cids)
|
||||
if err != nil {
|
||||
log.Error("failed loading tipset: ", cids)
|
||||
return cid.Undef, err
|
||||
func (sm *StateManager) computeTipSetState(ctx context.Context, blks []*types.BlockHeader, cb func(cid.Cid, *types.Message, *vm.ApplyRet) error) (cid.Cid, error) {
|
||||
if len(blks) == 1 && cb == nil {
|
||||
return blks[0].StateRoot, nil
|
||||
}
|
||||
|
||||
if len(ts.Blocks()) == 1 {
|
||||
return ts.Blocks()[0].StateRoot, nil
|
||||
}
|
||||
|
||||
pstate, err := sm.TipSetState(ts.Parents())
|
||||
pstate, err := sm.TipSetState(blks[0].Parents)
|
||||
if err != nil {
|
||||
return cid.Undef, xerrors.Errorf("recursive TipSetState failed: %w", err)
|
||||
}
|
||||
|
||||
vmi, err := vm.NewVM(pstate, ts.Height(), address.Undef, sm.cs)
|
||||
vmi, err := vm.NewVM(pstate, blks[0].Height, address.Undef, sm.cs)
|
||||
if err != nil {
|
||||
return cid.Undef, xerrors.Errorf("instantiating VM failed: %w", err)
|
||||
}
|
||||
|
||||
applied := make(map[cid.Cid]bool)
|
||||
for _, b := range ts.Blocks() {
|
||||
for _, b := range blks {
|
||||
vmi.SetBlockMiner(b.Miner)
|
||||
|
||||
bms, sms, err := sm.cs.MessagesForBlock(b)
|
||||
@ -98,10 +97,16 @@ func (sm *StateManager) computeTipSetState(cids []cid.Cid) (cid.Cid, error) {
|
||||
}
|
||||
applied[m.Cid()] = true
|
||||
|
||||
_, err := vmi.ApplyMessage(ctx, m)
|
||||
r, err := vmi.ApplyMessage(ctx, m)
|
||||
if err != nil {
|
||||
return cid.Undef, err
|
||||
}
|
||||
|
||||
if cb != nil {
|
||||
if err := cb(m.Cid(), m, r); err != nil {
|
||||
return cid.Undef, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for _, sm := range sms {
|
||||
@ -110,10 +115,16 @@ func (sm *StateManager) computeTipSetState(cids []cid.Cid) (cid.Cid, error) {
|
||||
}
|
||||
applied[sm.Cid()] = true
|
||||
|
||||
_, err := vmi.ApplyMessage(ctx, &sm.Message)
|
||||
r, err := vmi.ApplyMessage(ctx, &sm.Message)
|
||||
if err != nil {
|
||||
return cid.Undef, err
|
||||
}
|
||||
|
||||
if cb != nil {
|
||||
if err := cb(sm.Cid(), &sm.Message, r); err != nil {
|
||||
return cid.Undef, err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
60
cli/state.go
60
cli/state.go
@ -3,9 +3,11 @@ package cli
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"gopkg.in/urfave/cli.v2"
|
||||
|
||||
"github.com/filecoin-project/go-lotus/chain/address"
|
||||
"github.com/filecoin-project/go-lotus/chain/types"
|
||||
|
||||
"github.com/ipfs/go-cid"
|
||||
"gopkg.in/urfave/cli.v2"
|
||||
)
|
||||
|
||||
var stateCmd = &cli.Command{
|
||||
@ -116,3 +118,57 @@ var stateProvingSetCmd = &cli.Command{
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
var stateReplaySetCmd = &cli.Command{
|
||||
Name: "replay",
|
||||
Usage: "Replay a particular message within a tipset",
|
||||
Action: func(cctx *cli.Context) error {
|
||||
if cctx.Args().Len() < 2 {
|
||||
fmt.Println("usage: <tipset> <message cid>")
|
||||
fmt.Println("The last cid passed will be used as the message CID")
|
||||
fmt.Println("All preceding ones will be used as the tipset")
|
||||
return nil
|
||||
}
|
||||
|
||||
args := cctx.Args().Slice()
|
||||
mcid, err := cid.Decode(args[len(args)-1])
|
||||
if err != nil {
|
||||
return fmt.Errorf("message cid was invalid: %s", err)
|
||||
}
|
||||
|
||||
var tscids []cid.Cid
|
||||
for _, s := range args[:len(args)-1] {
|
||||
c, err := cid.Decode(s)
|
||||
if err != nil {
|
||||
return fmt.Errorf("tipset cid was invalid: %s", err)
|
||||
}
|
||||
tscids = append(tscids, c)
|
||||
}
|
||||
|
||||
api, err := GetFullNodeAPI(cctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ctx := ReqContext(cctx)
|
||||
|
||||
var headers []*types.BlockHeader
|
||||
for _, c := range tscids {
|
||||
h, err := api.ChainGetBlock(ctx, c)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
headers = append(headers, h)
|
||||
}
|
||||
|
||||
ts, err := types.NewTipSet(headers)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
api.StateReplay(ctx, ts, mcid)
|
||||
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
|
||||
cid "github.com/ipfs/go-cid"
|
||||
"github.com/ipfs/go-hamt-ipld"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
@ -194,6 +195,24 @@ func (a *StateAPI) StateCall(ctx context.Context, msg *types.Message, ts *types.
|
||||
return a.StateManager.Call(ctx, msg, ts)
|
||||
}
|
||||
|
||||
func (a *StateAPI) StateReplay(ctx context.Context, ts *types.TipSet, mc cid.Cid) (*api.ReplayResults, error) {
|
||||
m, r, err := a.StateManager.Replay(ctx, ts, mc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var errstr string
|
||||
if r.ActorErr != nil {
|
||||
errstr = r.ActorErr.Error()
|
||||
}
|
||||
|
||||
return &api.ReplayResults{
|
||||
Msg: m,
|
||||
Receipt: &r.MessageReceipt,
|
||||
Error: errstr,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *StateAPI) stateForTs(ts *types.TipSet) (*state.StateTree, error) {
|
||||
if ts == nil {
|
||||
ts = a.Chain.GetHeaviestTipSet()
|
||||
|
Loading…
Reference in New Issue
Block a user