executor.go

package common

import (
	"context"
	"sync/atomic"

	"github.com/ipfs/go-cid"
	cbg "github.com/whyrusleeping/cbor-gen"
	"go.opencensus.io/stats"
	"go.opencensus.io/trace"
	"golang.org/x/xerrors"

	"github.com/filecoin-project/go-state-types/abi"
	"github.com/filecoin-project/go-state-types/big"
	"github.com/filecoin-project/go-state-types/network"
	blockadt "github.com/filecoin-project/specs-actors/actors/util/adt"

	"github.com/filecoin-project/lotus/chain/consensus/actors/registry"
	"github.com/filecoin-project/lotus/chain/consensus/actors/reward"
	"github.com/filecoin-project/lotus/chain/consensus/common/crossmsg"
	"github.com/filecoin-project/lotus/chain/consensus/hierarchical"
	"github.com/filecoin-project/lotus/chain/consensus/hierarchical/subnet"
	"github.com/filecoin-project/lotus/chain/consensus/hierarchical/subnet/resolver"
	"github.com/filecoin-project/lotus/chain/rand"
	"github.com/filecoin-project/lotus/chain/stmgr"
	"github.com/filecoin-project/lotus/chain/store"
	"github.com/filecoin-project/lotus/chain/types"
	"github.com/filecoin-project/lotus/chain/vm"
	"github.com/filecoin-project/lotus/metrics"
)

func DefaultUpgradeSchedule() stmgr.UpgradeSchedule {
	var us stmgr.UpgradeSchedule

	updates := []stmgr.Upgrade{{
		Height:    -1,
		Network:   network.Version15,
		Migration: nil,
		Expensive: true,
	},
	}

	for _, u := range updates {
		// NOTE: Using default `UpgradeScheduler` for eudico instead of disabling it. This may be revisited if eudico
		// ends up needing different upgrade strategies.
		// if u.Height < 0 {
		// 	// upgrade disabled
		// 	continue
		// }
		us = append(us, u)
	}
	return us
}

type tipSetExecutor struct {
	submgr subnet.Manager
}

func (t *tipSetExecutor) NewActorRegistry() *vm.ActorRegistry {
	return registry.NewActorRegistry()
}

func TipSetExecutor(submgr subnet.Manager) stmgr.Executor {
	return &tipSetExecutor{submgr}
}

func RootTipSetExecutor() stmgr.Executor {
	return &tipSetExecutor{nil}
}

func (t *tipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager, cr *resolver.Resolver, parentEpoch abi.ChainEpoch, pstate cid.Cid, bms []store.BlockMessages, epoch abi.ChainEpoch, r vm.Rand, em stmgr.ExecMonitor, baseFee abi.TokenAmount, ts *types.TipSet) (cid.Cid, cid.Cid, error) {

	done := metrics.Timer(ctx, metrics.VMApplyBlocksTotal)
	defer done()

	partDone := metrics.Timer(ctx, metrics.VMApplyEarly)
	defer func() {
		partDone()
	}()

	makeVmWithBaseStateAndEpoch := func(base cid.Cid, e abi.ChainEpoch) (vm.Interface, error) {
		vmopt := &vm.VMOpts{
			StateBase:      base,
			Epoch:          e,
			Rand:           r,
			Bstore:         sm.ChainStore().StateBlockstore(),
			Actors:         registry.NewActorRegistry(),
			Syscalls:       sm.Syscalls,
			CircSupplyCalc: sm.GetVMCirculatingSupply,
			NetworkVersion: sm.GetNetworkVersion(ctx, e),
			BaseFee:        baseFee,
			LookbackState:  stmgr.LookbackStateGetterForTipset(sm, ts),
		}

		return sm.VMConstructor()(ctx, vmopt)
	}

	vmi, err := makeVmWithBaseStateAndEpoch(pstate, epoch)
	if err != nil {
		return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
	}

	for i := parentEpoch; i < epoch; i++ {
		// handle state forks
		// XXX: The state tree
		newState, err := sm.HandleStateForks(ctx, pstate, i, em, ts)
		if err != nil {
			return cid.Undef, cid.Undef, xerrors.Errorf("error handling state forks: %w", err)
		}

		if pstate != newState {
			vmi, err = makeVmWithBaseStateAndEpoch(newState, i)
			if err != nil {
				return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
			}
		}

		pstate = newState
	}

	partDone()
	partDone = metrics.Timer(ctx, metrics.VMApplyMessages)

	var receipts []cbg.CBORMarshaler
	processedMsgs := make(map[cid.Cid]struct{})
	for _, b := range bms {
		secp, cross := hierarchical.UnbundleCrossAndSecpMsgs(b.SecpkMessages)
		penalty := types.NewInt(0)
		gasReward := big.Zero()

		for _, cm := range append(b.BlsMessages, secp...) {
			m := cm.VMMessage()
			if _, found := processedMsgs[m.Cid()]; found {
				continue
			}
			r, err := vmi.ApplyMessage(ctx, cm)
			if err != nil {
				return cid.Undef, cid.Undef, err
			}

			receipts = append(receipts, &r.MessageReceipt)
			gasReward = big.Add(gasReward, r.GasCosts.MinerTip)
			penalty = big.Add(penalty, r.GasCosts.MinerPenalty)

			if em != nil {
				if err := em.MessageApplied(ctx, ts, cm.Cid(), m, r, false); err != nil {
					return cid.Undef, cid.Undef, err
				}
			}
			processedMsgs[m.Cid()] = struct{}{}
		}

		// FIXME: Here we send rewards for miners.
		// Rewards are only applied in the root, for subnets
		// rewards are disabled at this point and miners only get
		// the gas reward.
		// if t.submgr == nil we are in root net.
		reward := gasReward
		// In root consensus, there is currently a static reward of 1FIL per block.
		// (this is mainly for testing purposes).
		if t.submgr == nil {
			reward = big.Add(reward, types.FromFil(1))
		}
		if err := applyMiningRewards(ctx, vmi, em, b, epoch, ts, reward); err != nil {
			return cid.Undef, cid.Undef, err
		}

		// Sort cross-messages deterministically before applying them
		crossm, err := crossmsg.SortCrossMsgs(ctx, sm, cr, cross, ts)
		if err != nil {
			return cid.Undef, cid.Undef, xerrors.Errorf("error sorting cross-msgs: %w", err)
		}

		processedMsgs = make(map[cid.Cid]struct{})
		for _, m := range crossm {
			// m := crossm.VMMessage()
			// additional sanity-check to avoid processing a message
			// included in a block twice (although this is already checked
			// by SCA, and there are a few more additional checks, so this
			// may not be needed).
			if _, found := processedMsgs[m.Cid()]; found {
				continue
			}
			log.Infof("Executing cross message: %v", crossm)

			if err := crossmsg.ApplyCrossMsg(ctx, vmi, t.submgr, em, m, ts); err != nil {
				return cid.Undef, cid.Undef, xerrors.Errorf("cross messsage application failed: %w", err)
			}
			processedMsgs[m.Cid()] = struct{}{}
		}

	}

	partDone()
	partDone = metrics.Timer(ctx, metrics.VMApplyCron)

	partDone()
	partDone = metrics.Timer(ctx, metrics.VMApplyFlush)

	rectarr := blockadt.MakeEmptyArray(sm.ChainStore().ActorStore(ctx))
	for i, receipt := range receipts {
		if err := rectarr.Set(uint64(i), receipt); err != nil {
			return cid.Undef, cid.Undef, xerrors.Errorf("failed to build receipts amt: %w", err)
		}
	}
	rectroot, err := rectarr.Root()
	if err != nil {
		return cid.Undef, cid.Undef, xerrors.Errorf("failed to build receipts amt: %w", err)
	}

	st, err := vmi.Flush(ctx)
	if err != nil {
		return cid.Undef, cid.Undef, xerrors.Errorf("vm flush failed: %w", err)
	}

	stats.Record(ctx, metrics.VMSends.M(int64(atomic.LoadUint64(&vm.StatSends))),
		metrics.VMApplied.M(int64(atomic.LoadUint64(&vm.StatApplied))))

	return st, rectroot, nil
}

func applyMiningRewards(ctx context.Context, vmi vm.Interface, em stmgr.ExecMonitor, b store.BlockMessages, epoch abi.ChainEpoch, ts *types.TipSet, value abi.TokenAmount) error {
	rwMsg := &types.Message{
		From:       reward.RewardActorAddr,
		To:         b.Miner,
		Nonce:      uint64(epoch),
		Value:      value,
		GasFeeCap:  types.NewInt(0),
		GasPremium: types.NewInt(0),
		GasLimit:   1 << 30,
		Method:     0,
	}
	ret, actErr := vmi.ApplyImplicitMessage(ctx, rwMsg)
	if actErr != nil {
		return xerrors.Errorf("failed to apply reward message for miner %s: %w", b.Miner, actErr)
	}
	if em != nil {
		if err := em.MessageApplied(ctx, ts, rwMsg.Cid(), rwMsg, ret, true); err != nil {
			return xerrors.Errorf("callback failed on reward message: %w", err)
		}
	}

	if ret.ExitCode != 0 {
		return xerrors.Errorf("reward application message failed (exit %d): %s", ret.ExitCode, ret.ActorErr)
	}
	return nil
}

func (t *tipSetExecutor) ExecuteTipSet(ctx context.Context, sm *stmgr.StateManager, cr *resolver.Resolver, ts *types.TipSet, em stmgr.ExecMonitor) (stateroot cid.Cid, rectsroot cid.Cid, err error) {
	ctx, span := trace.StartSpan(ctx, "computeTipSetState")
	defer span.End()

	blks := ts.Blocks()

	for i := 0; i < len(blks); i++ {
		for j := i + 1; j < len(blks); j++ {
			if blks[i].Miner == blks[j].Miner {
				return cid.Undef, cid.Undef,
					xerrors.Errorf("duplicate miner in a tipset (%s %s)",
						blks[i].Miner, blks[j].Miner)
			}
		}
	}

	var parentEpoch abi.ChainEpoch
	pstate := blks[0].ParentStateRoot
	if blks[0].Height > 0 {
		parent, err := sm.ChainStore().GetBlock(ctx, blks[0].Parents[0])
		if err != nil {
			return cid.Undef, cid.Undef, xerrors.Errorf("getting parent block: %w", err)
		}

		parentEpoch = parent.Height
	}

	r := rand.NewStateRand(sm.ChainStore(), ts.Cids(), sm.Beacon(), sm.GetNetworkVersion)

	blkmsgs, err := sm.ChainStore().BlockMsgsForTipset(ctx, ts)
	if err != nil {
		return cid.Undef, cid.Undef, xerrors.Errorf("getting block messages for tipset: %w", err)
	}

	baseFee := blks[0].ParentBaseFee

	return t.ApplyBlocks(ctx, sm, cr, parentEpoch, pstate, blkmsgs, blks[0].Height, r, em, baseFee, ts)
}

var _ stmgr.Executor = &tipSetExecutor{}