Merge pull request #5795 from filecoin-project/feat/compute-state

feat: add compute state api

app/submodule/chain/chaininfo_api.go

@@ -869,3 +869,19 @@ func (cia *chainInfoAPI) ChainGetEvents(ctx context.Context, root cid.Cid) ([]ty
 
 	return ret, err
 }
+
+func (cia *chainInfoAPI) StateCompute(ctx context.Context, height abi.ChainEpoch, msgs []*types.Message, tsk types.TipSetKey) (*types.ComputeStateOutput, error) {
+	ts, err := cia.ChainGetTipSet(ctx, tsk)
+	if err != nil {
+		return nil, fmt.Errorf("loading tipset %s: %w", tsk, err)
+	}
+	st, t, err := statemanger.ComputeState(ctx, cia.chain.Stmgr, height, msgs, ts)
+	if err != nil {
+		return nil, err
+	}
+
+	return &types.ComputeStateOutput{
+		Root:  st,
+		Trace: t,
+	}, nil
+}

cmd/state.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
 	"strconv"
 
 	"github.com/filecoin-project/venus/cmd/tablewriter"
@@ -53,6 +54,7 @@ var stateCmd = &cmds.Command{
 		"list-actor":      stateListActorCmd,
 		"actor-cids":      stateSysActorCIDsCmd,
 		"replay":          stateReplayCmd,
+		"compute-state":   StateComputeStateCmd,
 	},
 }
 
@@ -709,3 +711,86 @@ func makeActorView(act *types.Actor, addr address.Address) *ActorView {
 		Head:    act.Head,
 	}
 }
+
+var StateComputeStateCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Perform state computations",
+	},
+	Options: []cmds.Option{
+		cmds.Uint64Option("vm-height", "set the height that the vm will see"),
+		cmds.BoolOption("apply-mpool-messages", "apply messages from the mempool to the computed state"),
+		cmds.BoolOption("show-trace", "print out full execution trace for given tipset"),
+		cmds.BoolOption("json", "generate json output"),
+		cmds.StringOption("compute-state-output", "a json file containing pre-existing compute-state output, to generate html reports without rerunning state changes"),
+		cmds.BoolOption("no-timing", "don't show timing information in html traces"),
+		cmds.StringOption("tipset", ""),
+	},
+	Run: func(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment) error {
+		ctx := req.Context
+
+		ts, err := LoadTipSet(ctx, req, getEnv(env).ChainAPI)
+		if err != nil {
+			return err
+		}
+		h, _ := req.Options["vm-height"].(uint64)
+		if h == 0 {
+			h = uint64(ts.Height())
+		}
+
+		var msgs []*types.Message
+		if applyMsg, _ := req.Options["apply-mpool-messages"].(bool); applyMsg {
+			pmsgs, err := getEnv(env).MessagePoolAPI.MpoolSelect(ctx, ts.Key(), 1)
+			if err != nil {
+				return err
+			}
+
+			for _, sm := range pmsgs {
+				msgs = append(msgs, &sm.Message)
+			}
+		}
+
+		var stout *types.ComputeStateOutput
+		if csofile, _ := req.Options["compute-state-output"].(string); len(csofile) != 0 {
+			data, err := os.ReadFile(csofile)
+			if err != nil {
+				return err
+			}
+
+			var o types.ComputeStateOutput
+			if err := json.Unmarshal(data, &o); err != nil {
+				return err
+			}
+
+			stout = &o
+		} else {
+			o, err := getEnv(env).ChainAPI.StateCompute(ctx, abi.ChainEpoch(h), msgs, ts.Key())
+			if err != nil {
+				return err
+			}
+
+			stout = o
+		}
+
+		buf := &bytes.Buffer{}
+		writer := NewSilentWriter(buf)
+
+		if ok, _ := req.Options["json"].(bool); ok {
+			out, err := json.Marshal(stout)
+			if err != nil {
+				return err
+			}
+			writer.Println(string(out))
+			return re.Emit(buf)
+		}
+
+		writer.Println("computed state cid: ", stout.Root)
+		if showTrace, _ := req.Options["show-trace"].(bool); showTrace {
+			for _, ir := range stout.Trace {
+				writer.Printf("%s\t%s\t%s\t%d\t%x\t%d\t%x\n", ir.Msg.From, ir.Msg.To, ir.Msg.Value, ir.Msg.Method, ir.Msg.Params, ir.MsgRct.ExitCode, ir.MsgRct.Return)
+				printInternalExecutions(writer, "\t", ir.ExecutionTrace.Subcalls)
+			}
+		}
+
+		return re.Emit(buf)
+	},
+}

pkg/statemanger/state_manger.go

@@ -14,12 +14,15 @@ import (
 	"github.com/filecoin-project/venus/pkg/chain"
 	"github.com/filecoin-project/venus/pkg/consensus"
 	"github.com/filecoin-project/venus/pkg/fork"
+	"github.com/filecoin-project/venus/pkg/fvm"
 	appstate "github.com/filecoin-project/venus/pkg/state"
 	"github.com/filecoin-project/venus/pkg/state/tree"
 	"github.com/filecoin-project/venus/pkg/vm"
 	"github.com/filecoin-project/venus/pkg/vm/gas"
+	"github.com/filecoin-project/venus/pkg/vm/vmcontext"
 	"github.com/filecoin-project/venus/venus-shared/actors/builtin/market"
 	"github.com/filecoin-project/venus/venus-shared/actors/builtin/paych"
+	blockstoreutil "github.com/filecoin-project/venus/venus-shared/blockstore"
 	"github.com/filecoin-project/venus/venus-shared/types"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
@@ -443,7 +446,39 @@ func (s *Stmgr) Replay(ctx context.Context, ts *types.TipSet, msgCID cid.Cid) (*
 	return outm, outr, nil
 }
 
+func (s *Stmgr) ExecutionTrace(ctx context.Context, ts *types.TipSet) (cid.Cid, []*types.InvocResult, error) {
+	var invocTrace []*types.InvocResult
+
+	cb := func(mcid cid.Cid, msg *types.Message, ret *vm.Ret) error {
+		ir := &types.InvocResult{
+			MsgCid:         mcid,
+			Msg:            msg,
+			MsgRct:         &ret.Receipt,
+			ExecutionTrace: ret.GasTracker.ExecutionTrace,
+			Duration:       ret.Duration,
+		}
+		if ret.ActorErr != nil {
+			ir.Error = ret.ActorErr.Error()
+		}
+		if !ret.OutPuts.Refund.Nil() {
+			ir.GasCost = MakeMsgGasCost(msg, ret)
+		}
+
+		invocTrace = append(invocTrace, ir)
+
+		return nil
+	}
+
+	st, _, err := s.cp.RunStateTransition(ctx, ts, cb, true)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	return st, invocTrace, nil
+}
+
 func MakeMsgGasCost(msg *types.Message, ret *vm.Ret) types.MsgGasCost {
+	fmt.Println(msg.RequiredFunds(), ret.OutPuts.Refund)
 	return types.MsgGasCost{
 		Message:            msg.Cid(),
 		GasUsed:            big.NewInt(ret.Receipt.GasUsed),
@@ -456,6 +491,76 @@ func MakeMsgGasCost(msg *types.Message, ret *vm.Ret) types.MsgGasCost {
 	}
 }
 
+func ComputeState(ctx context.Context, s *Stmgr, height abi.ChainEpoch, msgs []*types.Message, ts *types.TipSet) (cid.Cid, []*types.InvocResult, error) {
+	if ts == nil {
+		ts = s.cs.GetHead()
+	}
+
+	base, trace, err := s.ExecutionTrace(ctx, ts)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	for i := ts.Height(); i < height; i++ {
+		// Technically, the tipset we're passing in here should be ts+1, but that may not exist.
+		base, err = s.fork.HandleStateForks(ctx, base, i, ts)
+		if err != nil {
+			return cid.Undef, nil, fmt.Errorf("error handling state forks: %w", err)
+		}
+
+		// We intentionally don't run cron here, as we may be trying to look into the
+		// future. It's not guaranteed to be accurate... but that's fine.
+	}
+
+	buffStore := blockstoreutil.NewTieredBstore(s.cs.Blockstore(), blockstoreutil.NewTemporarySync())
+	vmopt := vm.VmOption{
+		CircSupplyCalculator: func(ctx context.Context, epoch abi.ChainEpoch, tree tree.Tree) (abi.TokenAmount, error) {
+			cs, err := s.cs.GetCirculatingSupplyDetailed(ctx, epoch, tree)
+			if err != nil {
+				return abi.TokenAmount{}, err
+			}
+			return cs.FilCirculating, nil
+		},
+		PRoot:               base,
+		Epoch:               ts.Height(),
+		Timestamp:           ts.MinTimestamp(),
+		Rnd:                 consensus.NewHeadRandomness(s.rnd, ts.Key()),
+		Bsstore:             buffStore,
+		SysCallsImpl:        s.syscallsImpl,
+		GasPriceSchedule:    s.gasSchedule,
+		NetworkVersion:      s.GetNetworkVersion(ctx, height),
+		BaseFee:             ts.Blocks()[0].ParentBaseFee,
+		Fork:                s.fork,
+		LookbackStateGetter: vmcontext.LookbackStateGetterForTipset(ctx, s.cs, s.fork, ts),
+		TipSetGetter:        vmcontext.TipSetGetterForTipset(s.cs.GetTipSetByHeight, ts),
+		Tracing:             true,
+		ActorDebugging:      s.actorDebugging,
+	}
+
+	vmi, err := fvm.NewVM(ctx, vmopt)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	for i, msg := range msgs {
+		// TODO: Use the signed message length for secp messages
+		ret, err := vmi.ApplyMessage(ctx, msg)
+		if err != nil {
+			return cid.Undef, nil, fmt.Errorf("applying message %s: %w", msg.Cid(), err)
+		}
+		if ret.Receipt.ExitCode != 0 {
+			s.log.Infof("compute state apply message %d failed (exit: %d): %s", i, ret.Receipt.ExitCode, ret.ActorErr)
+		}
+	}
+
+	root, err := vmi.Flush(ctx)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	return root, trace, nil
+}
+
 func (s *Stmgr) FlushChainHead() (*types.TipSet, error) {
 	head := s.cs.GetHead()
 	_, _, err := s.RunStateTransition(context.TODO(), head, nil, false)

venus-shared/api/chain/v0/chain.go

@@ -83,6 +83,39 @@ type IChainInfo interface {
 	StateActorManifestCID(context.Context, network.Version) (cid.Cid, error)                            //perm:read
 	StateCall(ctx context.Context, msg *types.Message, tsk types.TipSetKey) (*types.InvocResult, error) //perm:read
 	StateReplay(context.Context, types.TipSetKey, cid.Cid) (*types.InvocResult, error)                  //perm:read
+	// StateCompute is a flexible command that applies the given messages on the given tipset.
+	// The messages are run as though the VM were at the provided height.
+	//
+	// When called, StateCompute will:
+	// - Load the provided tipset, or use the current chain head if not provided
+	// - Compute the tipset state of the provided tipset on top of the parent state
+	//   - (note that this step runs before vmheight is applied to the execution)
+	//   - Execute state upgrade if any were scheduled at the epoch, or in null
+	//     blocks preceding the tipset
+	//   - Call the cron actor on null blocks preceding the tipset
+	//   - For each block in the tipset
+	//     - Apply messages in blocks in the specified
+	//     - Award block reward by calling the reward actor
+	//   - Call the cron actor for the current epoch
+	// - If the specified vmheight is higher than the current epoch, apply any
+	//   needed state upgrades to the state
+	// - Apply the specified messages to the state
+	//
+	// The vmheight parameter sets VM execution epoch, and can be used to simulate
+	// message execution in different network versions. If the specified vmheight
+	// epoch is higher than the epoch of the specified tipset, any state upgrades
+	// until the vmheight will be executed on the state before applying messages
+	// specified by the user.
+	//
+	// Note that the initial tipset state computation is not affected by the
+	// vmheight parameter - only the messages in the `apply` set are
+	//
+	// If the caller wants to simply compute the state, vmheight should be set to
+	// the epoch of the specified tipset.
+	//
+	// Messages in the `apply` parameter must have the correct nonces, and gas
+	// values set.
+	StateCompute(context.Context, abi.ChainEpoch, []*types.Message, types.TipSetKey) (*types.ComputeStateOutput, error) //perm:read
 }
 
 type IMinerState interface {