Merge branch 'vyzo/feat/exec-lanes' into HEAD

Fixes: https://github.com/filecoin-project/lotus/issues/10537

chain/index/msgindex.go

@@ -169,6 +169,89 @@ func NewMsgIndex(lctx context.Context, basePath string, cs ChainStore) (MsgIndex
 	return msgIndex, nil
 }
 
+func PopulateAfterSnapshot(lctx context.Context, basePath string, cs ChainStore) error {
+	err := os.MkdirAll(basePath, 0755)
+	if err != nil {
+		return xerrors.Errorf("error creating msgindex base directory: %w", err)
+	}
+
+	dbPath := path.Join(basePath, dbName)
+
+	// if a database already exists, we try to delete it and create a new one
+	if _, err := os.Stat(dbPath); err == nil {
+		if err = os.Remove(dbPath); err != nil {
+			return xerrors.Errorf("msgindex already exists at %s and can't be deleted", dbPath)
+		}
+	}
+
+	db, err := sql.Open("sqlite3", dbPath)
+	if err != nil {
+		return xerrors.Errorf("error opening msgindex database: %w", err)
+	}
+	defer db.Close()
+
+	if err := prepareDB(db); err != nil {
+		return xerrors.Errorf("error creating msgindex database: %w", err)
+	}
+
+	tx, err := db.Begin()
+	if err != nil {
+		return xerrors.Errorf("error when starting transaction: %w", err)
+	}
+
+	rollback := func() {
+		if err := tx.Rollback(); err != nil {
+			log.Errorf("error in rollback: %s", err)
+		}
+	}
+
+	insertStmt, err := tx.Prepare(dbqInsertMessage)
+	if err != nil {
+		return xerrors.Errorf("error preparing insertMsgStmt: %w", err)
+	}
+	defer insertStmt.Close()
+
+	curTs := cs.GetHeaviestTipSet()
+	startHeight := curTs.Height()
+	for curTs != nil {
+		tscid, err := curTs.Key().Cid()
+		if err != nil {
+			rollback()
+			return xerrors.Errorf("error computing tipset cid: %w", err)
+		}
+
+		tskey := tscid.String()
+		epoch := int64(curTs.Height())
+
+		msgs, err := cs.MessagesForTipset(lctx, curTs)
+		if err != nil {
+			log.Infof("stopping import after %d tipsets", startHeight-curTs.Height())
+			break
+		}
+
+		for _, msg := range msgs {
+			key := msg.Cid().String()
+			if _, err := insertStmt.Exec(key, tskey, epoch); err != nil {
+				rollback()
+				return xerrors.Errorf("error inserting message: %w", err)
+			}
+		}
+
+		curTs, err = cs.GetTipSetFromKey(lctx, curTs.Parents())
+		if err != nil {
+			rollback()
+			return xerrors.Errorf("error walking chain: %w", err)
+		}
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return xerrors.Errorf("error commiting transaction: %w", err)
+	}
+
+	return nil
+}
+
 // init utilities
 func prepareDB(db *sql.DB) error {
 	for _, stmt := range dbDefs {

chain/stmgr/call.go

@@ -52,8 +52,8 @@ func (sm *StateManager) Call(ctx context.Context, msg *types.Message, ts *types.
 }
 
 // CallWithGas calculates the state for a given tipset, and then applies the given message on top of that state.
-func (sm *StateManager) CallWithGas(ctx context.Context, msg *types.Message, priorMsgs []types.ChainMsg, ts *types.TipSet) (*api.InvocResult, error) {
-	return sm.callInternal(ctx, msg, priorMsgs, ts, cid.Undef, sm.GetNetworkVersion, true, true)
+func (sm *StateManager) CallWithGas(ctx context.Context, msg *types.Message, priorMsgs []types.ChainMsg, ts *types.TipSet, applyTsMessages bool) (*api.InvocResult, error) {
+	return sm.callInternal(ctx, msg, priorMsgs, ts, cid.Undef, sm.GetNetworkVersion, true, applyTsMessages)
 }
 
 // CallAtStateAndVersion allows you to specify a message to execute on the given stateCid and network version.
@@ -117,12 +117,22 @@ func (sm *StateManager) callInternal(ctx context.Context, msg *types.Message, pr
 	if stateCid == cid.Undef {
 		stateCid = ts.ParentState()
 	}
+	tsMsgs, err := sm.cs.MessagesForTipset(ctx, ts)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to lookup messages for parent tipset: %w", err)
+	}
+
 	if applyTsMessages {
-		tsMsgs, err := sm.cs.MessagesForTipset(ctx, ts)
-		if err != nil {
-			return nil, xerrors.Errorf("failed to lookup messages for parent tipset: %w", err)
-		}
 		priorMsgs = append(tsMsgs, priorMsgs...)
+	} else {
+		var filteredTsMsgs []types.ChainMsg
+		for _, tsMsg := range tsMsgs {
+			//TODO we should technically be normalizing the filecoin address of from when we compare here
+			if tsMsg.VMMessage().From == msg.VMMessage().From {
+				filteredTsMsgs = append(filteredTsMsgs, tsMsg)
+			}
+		}
+		priorMsgs = append(filteredTsMsgs, priorMsgs...)
 	}
 
 	// Technically, the tipset we're passing in here should be ts+1, but that may not exist.

chain/stmgr/execute.go

@@ -128,10 +128,38 @@ func (sm *StateManager) ExecutionTraceWithMonitor(ctx context.Context, ts *types
 }
 
 func (sm *StateManager) ExecutionTrace(ctx context.Context, ts *types.TipSet) (cid.Cid, []*api.InvocResult, error) {
+	tsKey := ts.Key()
+
+	{
+		// check if we have the trace for this tipset in the cache
+		sm.execTraceCacheLock.Lock()
+		defer sm.execTraceCacheLock.Unlock()
+		if entry, ok := sm.execTraceCache.Get(tsKey); ok {
+			// we have to make a deep copy since caller can modify the invocTrace
+			return entry.postStateRoot, makeDeepCopy(entry.invocTrace), nil
+		}
+	}
+
 	var invocTrace []*api.InvocResult
 	st, err := sm.ExecutionTraceWithMonitor(ctx, ts, &InvocationTracer{trace: &invocTrace})
 	if err != nil {
 		return cid.Undef, nil, err
 	}
+
+	sm.execTraceCache.Add(tsKey, tipSetCacheEntry{st, makeDeepCopy(invocTrace)})
+
 	return st, invocTrace, nil
 }
+
+func makeDeepCopy(invocTrace []*api.InvocResult) []*api.InvocResult {
+	c := make([]*api.InvocResult, len(invocTrace))
+	for i, ir := range invocTrace {
+		if ir == nil {
+			continue
+		}
+		tmp := *ir
+		c[i] = &tmp
+	}
+
+	return c
+}

chain/stmgr/forks_test.go

@@ -347,7 +347,7 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
 		currentHeight := ts.TipSet.TipSet().Height()
 
 		// CallWithGas calls on top of the given tipset.
-		ret, err := sm.CallWithGas(ctx, m, nil, ts.TipSet.TipSet())
+		ret, err := sm.CallWithGas(ctx, m, nil, ts.TipSet.TipSet(), true)
 		if parentHeight <= testForkHeight && currentHeight >= testForkHeight {
 			// If I had a fork, or I _will_ have a fork, it should fail.
 			require.Equal(t, ErrExpensiveFork, err)
@@ -368,7 +368,7 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
 		// Calls without a tipset should walk back to the last non-fork tipset.
 		// We _verify_ that the migration wasn't run multiple times at the end of the
 		// test.
-		ret, err = sm.CallWithGas(ctx, m, nil, nil)
+		ret, err = sm.CallWithGas(ctx, m, nil, nil, true)
 		require.NoError(t, err)
 		require.True(t, ret.MsgRct.ExitCode.IsSuccess())
 

chain/stmgr/stmgr.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sync"
 
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/ipfs/go-cid"
 	dstore "github.com/ipfs/go-datastore"
 	cbor "github.com/ipfs/go-ipld-cbor"
@@ -39,6 +40,8 @@ import (
 const LookbackNoLimit = api.LookbackNoLimit
 const ReceiptAmtBitwidth = 3
 
+const execTraceCacheSize = 16
+
 var log = logging.Logger("statemgr")
 
 type StateManagerAPI interface {
@@ -131,13 +134,17 @@ type StateManager struct {
 	postIgnitionVesting []msig0.State
 	postCalicoVesting   []msig0.State
 
-	genesisPledge      abi.TokenAmount
-	genesisMarketFunds abi.TokenAmount
+	genesisPledge abi.TokenAmount
 
 	tsExec        Executor
 	tsExecMonitor ExecMonitor
 	beacon        beacon.Schedule
 
+	// We keep a small cache for calls to ExecutionTrace which helps improve
+	// performance for node operators like exchanges and block explorers
+	execTraceCache     *lru.ARCCache[types.TipSetKey, tipSetCacheEntry]
+	execTraceCacheLock sync.Mutex
+
 	msgIndex index.MsgIndex
 }
 
@@ -147,6 +154,11 @@ type treeCache struct {
 	tree *state.StateTree
 }
 
+type tipSetCacheEntry struct {
+	postStateRoot cid.Cid
+	invocTrace    []*api.InvocResult
+}
+
 func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, beacon beacon.Schedule, metadataDs dstore.Batching, msgIndex index.MsgIndex) (*StateManager, error) {
 	// If we have upgrades, make sure they're in-order and make sense.
 	if err := us.Validate(); err != nil {
@@ -186,6 +198,11 @@ func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder,
 		}
 	}
 
+	execTraceCache, err := lru.NewARC[types.TipSetKey, tipSetCacheEntry](execTraceCacheSize)
+	if err != nil {
+		return nil, err
+	}
+
 	return &StateManager{
 		networkVersions:   networkVersions,
 		latestVersion:     lastVersion,
@@ -201,7 +218,8 @@ func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder,
 			root: cid.Undef,
 			tree: nil,
 		},
-		compWait: make(map[string]chan struct{}),
+		compWait:       make(map[string]chan struct{}),
+		execTraceCache: execTraceCache,
 		msgIndex: msgIndex,
 	}, nil
 }

chain/stmgr/supply.go

@@ -51,17 +51,13 @@ func (sm *StateManager) setupGenesisVestingSchedule(ctx context.Context) error {
 		return xerrors.Errorf("loading state tree: %w", err)
 	}
 
-	gmf, err := getFilMarketLocked(ctx, sTree)
-	if err != nil {
-		return xerrors.Errorf("setting up genesis market funds: %w", err)
-	}
-
 	gp, err := getFilPowerLocked(ctx, sTree)
 	if err != nil {
 		return xerrors.Errorf("setting up genesis pledge: %w", err)
 	}
 
-	sm.genesisMarketFunds = gmf
+	sm.genesisMsigLk.Lock()
+	defer sm.genesisMsigLk.Unlock()
 	sm.genesisPledge = gp
 
 	totalsByEpoch := make(map[abi.ChainEpoch]abi.TokenAmount)
@@ -128,6 +124,8 @@ func (sm *StateManager) setupPostIgnitionVesting(ctx context.Context) error {
 	totalsByEpoch[sixYears] = big.NewInt(100_000_000)
 	totalsByEpoch[sixYears] = big.Add(totalsByEpoch[sixYears], big.NewInt(300_000_000))
 
+	sm.genesisMsigLk.Lock()
+	defer sm.genesisMsigLk.Unlock()
 	sm.postIgnitionVesting = make([]msig0.State, 0, len(totalsByEpoch))
 	for k, v := range totalsByEpoch {
 		ns := msig0.State{
@@ -178,6 +176,9 @@ func (sm *StateManager) setupPostCalicoVesting(ctx context.Context) error {
 	totalsByEpoch[sixYears] = big.Add(totalsByEpoch[sixYears], big.NewInt(300_000_000))
 	totalsByEpoch[sixYears] = big.Add(totalsByEpoch[sixYears], big.NewInt(9_805_053))
 
+	sm.genesisMsigLk.Lock()
+	defer sm.genesisMsigLk.Unlock()
+
 	sm.postCalicoVesting = make([]msig0.State, 0, len(totalsByEpoch))
 	for k, v := range totalsByEpoch {
 		ns := msig0.State{
@@ -198,21 +199,20 @@ func (sm *StateManager) setupPostCalicoVesting(ctx context.Context) error {
 func (sm *StateManager) GetFilVested(ctx context.Context, height abi.ChainEpoch) (abi.TokenAmount, error) {
 	vf := big.Zero()
 
-	sm.genesisMsigLk.Lock()
-	defer sm.genesisMsigLk.Unlock()
-
 	// TODO: combine all this?
-	if sm.preIgnitionVesting == nil || sm.genesisPledge.IsZero() || sm.genesisMarketFunds.IsZero() {
+	if sm.preIgnitionVesting == nil || sm.genesisPledge.IsZero() {
 		err := sm.setupGenesisVestingSchedule(ctx)
 		if err != nil {
 			return vf, xerrors.Errorf("failed to setup pre-ignition vesting schedule: %w", err)
 		}
+
 	}
 	if sm.postIgnitionVesting == nil {
 		err := sm.setupPostIgnitionVesting(ctx)
 		if err != nil {
 			return vf, xerrors.Errorf("failed to setup post-ignition vesting schedule: %w", err)
 		}
+
 	}
 	if sm.postCalicoVesting == nil {
 		err := sm.setupPostCalicoVesting(ctx)
@@ -246,8 +246,6 @@ func (sm *StateManager) GetFilVested(ctx context.Context, height abi.ChainEpoch)
 	if height <= build.UpgradeAssemblyHeight {
 		// continue to use preIgnitionGenInfos, nothing changed at the Ignition epoch
 		vf = big.Add(vf, sm.genesisPledge)
-		// continue to use preIgnitionGenInfos, nothing changed at the Ignition epoch
-		vf = big.Add(vf, sm.genesisMarketFunds)
 	}
 
 	return vf, nil

cmd/lotus-shed/chainwatch.go

@@ -0,0 +1,383 @@
+package main
+
+import (
+	"container/list"
+	"context"
+	"database/sql"
+	"fmt"
+	"hash/crc32"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/ipfs/go-cid"
+	"github.com/urfave/cli/v2"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/store"
+	"github.com/filecoin-project/lotus/chain/types"
+	cliutil "github.com/filecoin-project/lotus/cli/util"
+)
+
+var chainwatchCmd = &cli.Command{
+	Name:  "chainwatch",
+	Usage: "lotus chainwatch",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:    "db",
+			EnvVars: []string{"CHAINWATCH_DB"},
+			Value:   "./chainwatch.db",
+		},
+	},
+	Subcommands: []*cli.Command{
+		chainwatchRunCmd,
+		chainwatchDotCmd,
+	},
+}
+
+var chainwatchDotCmd = &cli.Command{
+	Name:      "dot",
+	Usage:     "generate dot graphs",
+	ArgsUsage: "<minHeight> <toseeHeight>",
+	Action: func(cctx *cli.Context) error {
+		st, err := cwOpenStorage(cctx.String("db"))
+		if err != nil {
+			return err
+		}
+
+		minH, err := strconv.ParseInt(cctx.Args().Get(0), 10, 32)
+		if err != nil {
+			return err
+		}
+		tosee, err := strconv.ParseInt(cctx.Args().Get(1), 10, 32)
+		if err != nil {
+			return err
+		}
+		maxH := minH + tosee
+
+		res, err := st.db.Query(`select block, parent, b.miner, b.height, p.height from block_parents
+    inner join blocks b on block_parents.block = b.cid
+    inner join blocks p on block_parents.parent = p.cid
+where b.height > ? and b.height < ?`, minH, maxH)
+
+		if err != nil {
+			return err
+		}
+
+		fmt.Println("digraph D {")
+
+		for res.Next() {
+			var block, parent, miner string
+			var height, ph uint64
+			if err := res.Scan(&block, &parent, &miner, &height, &ph); err != nil {
+				return err
+			}
+
+			bc, err := cid.Parse(block)
+			if err != nil {
+				return err
+			}
+
+			has := st.hasBlock(bc)
+
+			col := crc32.Checksum([]byte(miner), crc32.MakeTable(crc32.Castagnoli))&0xc0c0c0c0 + 0x30303030
+
+			hasstr := ""
+			if !has {
+				//col = 0xffffffff
+				hasstr = " UNSYNCED"
+			}
+
+			nulls := height - ph - 1
+			for i := uint64(0); i < nulls; i++ {
+				name := block + "NP" + fmt.Sprint(i)
+
+				fmt.Printf("%s [label = \"NULL:%d\", fillcolor = \"#ffddff\", style=filled, forcelabels=true]\n%s -> %s\n",
+					name, height-nulls+i, name, parent)
+
+				parent = name
+			}
+
+			fmt.Printf("%s [label = \"%s:%d%s\", fillcolor = \"#%06x\", style=filled, forcelabels=true]\n%s -> %s\n", block, miner, height, hasstr, col, block, parent)
+		}
+		if res.Err() != nil {
+			return res.Err()
+		}
+
+		fmt.Println("}")
+
+		return nil
+	},
+}
+
+var chainwatchRunCmd = &cli.Command{
+	Name:  "run",
+	Usage: "Start lotus chainwatch",
+
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := cliutil.GetFullNodeAPIV1(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := cliutil.ReqContext(cctx)
+
+		v, err := api.Version(ctx)
+		if err != nil {
+			return err
+		}
+
+		log.Infof("Remote version: %s", v.Version)
+
+		st, err := cwOpenStorage(cctx.String("db")) // todo flag
+		if err != nil {
+			return err
+		}
+		defer st.close() // nolint:errcheck
+
+		cwRunSyncer(ctx, api, st)
+		go cwSubBlocks(ctx, api, st)
+
+		<-ctx.Done()
+		return nil
+	},
+}
+
+func cwSubBlocks(ctx context.Context, api api.FullNode, st *cwStorage) {
+	sub, err := api.SyncIncomingBlocks(ctx)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+
+	for bh := range sub {
+		err := st.storeHeaders(map[cid.Cid]*types.BlockHeader{
+			bh.Cid(): bh,
+		}, false)
+		if err != nil {
+			log.Error(err)
+		}
+	}
+}
+
+func cwRunSyncer(ctx context.Context, api api.FullNode, st *cwStorage) {
+	notifs, err := api.ChainNotify(ctx)
+	if err != nil {
+		panic(err)
+	}
+	go func() {
+		for notif := range notifs {
+			for _, change := range notif {
+				switch change.Type {
+				case store.HCCurrent:
+					fallthrough
+				case store.HCApply:
+					syncHead(ctx, api, st, change.Val)
+				case store.HCRevert:
+					log.Warnf("revert todo")
+				}
+			}
+		}
+	}()
+}
+
+func syncHead(ctx context.Context, api api.FullNode, st *cwStorage, ts *types.TipSet) {
+	log.Infof("Getting headers / actors")
+
+	toSync := map[cid.Cid]*types.BlockHeader{}
+	toVisit := list.New()
+
+	for _, header := range ts.Blocks() {
+		toVisit.PushBack(header)
+	}
+
+	for toVisit.Len() > 0 {
+		bh := toVisit.Remove(toVisit.Back()).(*types.BlockHeader)
+
+		if _, seen := toSync[bh.Cid()]; seen || st.hasBlock(bh.Cid()) {
+			continue
+		}
+
+		toSync[bh.Cid()] = bh
+
+		if len(toSync)%500 == 10 {
+			log.Infof("todo: (%d) %s", len(toSync), bh.Cid())
+		}
+
+		if len(bh.Parents) == 0 {
+			continue
+		}
+
+		if bh.Height <= 530000 {
+			continue
+		}
+
+		pts, err := api.ChainGetTipSet(ctx, types.NewTipSetKey(bh.Parents...))
+		if err != nil {
+			log.Error(err)
+			continue
+		}
+
+		for _, header := range pts.Blocks() {
+			toVisit.PushBack(header)
+		}
+	}
+
+	log.Infof("Syncing %d blocks", len(toSync))
+
+	log.Infof("Persisting headers")
+	if err := st.storeHeaders(toSync, true); err != nil {
+		log.Error(err)
+		return
+	}
+
+	log.Infof("Sync done")
+}
+
+type cwStorage struct {
+	db *sql.DB
+
+	headerLk sync.Mutex
+}
+
+func cwOpenStorage(dbSource string) (*cwStorage, error) {
+	db, err := sql.Open("sqlite3", dbSource)
+	if err != nil {
+		return nil, err
+	}
+
+	st := &cwStorage{db: db}
+
+	return st, st.setup()
+}
+
+func (st *cwStorage) setup() error {
+	tx, err := st.db.Begin()
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(`
+create table if not exists blocks
+(
+	cid text not null
+		constraint blocks_pk
+			primary key,
+	parentWeight numeric not null,
+	parentStateRoot text not null,
+	height int not null,
+	miner text not null
+		constraint blocks_id_address_map_miner_fk
+			references id_address_map (address),
+	timestamp int not null,
+	vrfproof blob
+);
+
+create unique index if not exists block_cid_uindex
+	on blocks (cid);
+
+create table if not exists blocks_synced
+(
+	cid text not null
+		constraint blocks_synced_pk
+			primary key
+		constraint blocks_synced_blocks_cid_fk
+			references blocks,
+	add_ts int not null
+);
+
+create unique index if not exists blocks_synced_cid_uindex
+	on blocks_synced (cid);
+
+create table if not exists block_parents
+(
+	block text not null
+		constraint block_parents_blocks_cid_fk
+			references blocks,
+	parent text not null
+		constraint block_parents_blocks_cid_fk_2
+			references blocks
+);
+
+create unique index if not exists block_parents_block_parent_uindex
+	on block_parents (block, parent);
+
+create unique index if not exists blocks_cid_uindex
+	on blocks (cid);
+`)
+	if err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+func (st *cwStorage) hasBlock(bh cid.Cid) bool {
+	var exitsts bool
+	err := st.db.QueryRow(`select exists (select 1 FROM blocks_synced where cid=?)`, bh.String()).Scan(&exitsts)
+	if err != nil {
+		log.Error(err)
+		return false
+	}
+	return exitsts
+}
+
+func (st *cwStorage) storeHeaders(bhs map[cid.Cid]*types.BlockHeader, sync bool) error {
+	st.headerLk.Lock()
+	defer st.headerLk.Unlock()
+
+	tx, err := st.db.Begin()
+	if err != nil {
+		return err
+	}
+
+	stmt, err := tx.Prepare(`insert into blocks (cid, parentWeight, parentStateRoot, height, miner, "timestamp", vrfproof) values (?, ?, ?, ?, ?, ?, ?) on conflict do nothing`)
+	if err != nil {
+		return err
+	}
+	defer stmt.Close() // nolint:errcheck
+	for _, bh := range bhs {
+		if _, err := stmt.Exec(bh.Cid().String(),
+			bh.ParentWeight.String(),
+			bh.ParentStateRoot.String(),
+			bh.Height,
+			bh.Miner.String(),
+			bh.Timestamp,
+			bh.Ticket.VRFProof,
+		); err != nil {
+			return err
+		}
+	}
+
+	stmt2, err := tx.Prepare(`insert into block_parents (block, parent) values (?, ?) on conflict do nothing`)
+	if err != nil {
+		return err
+	}
+	defer stmt2.Close() // nolint:errcheck
+	for _, bh := range bhs {
+		for _, parent := range bh.Parents {
+			if _, err := stmt2.Exec(bh.Cid().String(), parent.String()); err != nil {
+				return err
+			}
+		}
+	}
+
+	if sync {
+		stmt, err := tx.Prepare(`insert into blocks_synced (cid, add_ts) values (?, ?) on conflict do nothing`)
+		if err != nil {
+			return err
+		}
+		defer stmt.Close() // nolint:errcheck
+		now := time.Now().Unix()
+
+		for _, bh := range bhs {
+			if _, err := stmt.Exec(bh.Cid().String(), now); err != nil {
+				return err
+			}
+		}
+	}
+
+	return tx.Commit()
+}
+
+func (st *cwStorage) close() error {
+	return st.db.Close()
+}

cmd/lotus-shed/gas-estimation.go

@@ -241,7 +241,7 @@ var replayOfflineCmd = &cli.Command{
 		}
 
 		tw := tabwriter.NewWriter(os.Stdout, 8, 2, 2, ' ', tabwriter.AlignRight)
-		res, err := sm.CallWithGas(ctx, msg, []types.ChainMsg{}, executionTs)
+		res, err := sm.CallWithGas(ctx, msg, []types.ChainMsg{}, executionTs, true)
 		if err != nil {
 			return err
 		}

cmd/lotus-shed/main.go

@@ -26,6 +26,7 @@ func main() {
 		base32Cmd,
 		base16Cmd,
 		bitFieldCmd,
+		chainwatchCmd,
 		cronWcCmd,
 		frozenMinersCmd,
 		dealLabelCmd,

cmd/lotus/daemon.go

@@ -13,6 +13,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"os"
+	"path"
 	"runtime/pprof"
 	"strings"
 
@@ -558,5 +559,11 @@ func ImportChain(ctx context.Context, r repo.Repo, fname string, snapshot bool)
 		return err
 	}
 
+	log.Info("populating message index...")
+	if err := index.PopulateAfterSnapshot(ctx, path.Join(lr.Path(), "sqlite"), cst); err != nil {
+		return err
+	}
+	log.Info("populating message index done")
+
 	return nil
 }

node/impl/full/eth.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
 	"sort"
 	"strconv"
 	"sync"
@@ -888,9 +889,14 @@ func (a *EthModule) applyMessage(ctx context.Context, msg *types.Message, tsk ty
 		return nil, xerrors.Errorf("cannot get tipset: %w", err)
 	}
 
+	applyTsMessages := true
+	if os.Getenv("LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS") == "1" {
+		applyTsMessages = false
+	}
+
 	// Try calling until we find a height with no migration.
 	for {
-		res, err = a.StateManager.CallWithGas(ctx, msg, []types.ChainMsg{}, ts)
+		res, err = a.StateManager.CallWithGas(ctx, msg, []types.ChainMsg{}, ts, applyTsMessages)
 		if err != stmgr.ErrExpensiveFork {
 			break
 		}
@@ -959,10 +965,15 @@ func gasSearch(
 	high := msg.GasLimit
 	low := msg.GasLimit
 
+	applyTsMessages := true
+	if os.Getenv("LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS") == "1" {
+		applyTsMessages = false
+	}
+
 	canSucceed := func(limit int64) (bool, error) {
 		msg.GasLimit = limit
 
-		res, err := smgr.CallWithGas(ctx, &msg, priorMsgs, ts)
+		res, err := smgr.CallWithGas(ctx, &msg, priorMsgs, ts, applyTsMessages)
 		if err != nil {
 			return false, xerrors.Errorf("CallWithGas failed: %w", err)
 		}

node/impl/full/gas.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"math"
 	"math/rand"
+	"os"
 	"sort"
 
 	lru "github.com/hashicorp/golang-lru/v2"
@@ -276,10 +277,15 @@ func gasEstimateCallWithGas(
 		priorMsgs = append(priorMsgs, m)
 	}
 
+	applyTsMessages := true
+	if os.Getenv("LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS") == "1" {
+		applyTsMessages = false
+	}
+
 	// Try calling until we find a height with no migration.
 	var res *api.InvocResult
 	for {
-		res, err = smgr.CallWithGas(ctx, &msg, priorMsgs, ts)
+		res, err = smgr.CallWithGas(ctx, &msg, priorMsgs, ts, applyTsMessages)
 		if err != stmgr.ErrExpensiveFork {
 			break
 		}
@@ -372,18 +378,6 @@ func gasEstimateGasLimit(
 	}
 	ret = (ret * int64(transitionalMulti*1024)) >> 10
 
-	// Special case for PaymentChannel collect, which is deleting actor
-	// We ignore errors in this special case since they CAN occur,
-	// and we just want to detect existing payment channel actors
-	st, err := smgr.ParentState(ts)
-	if err == nil {
-		act, err := st.GetActor(msg.To)
-		if err == nil && lbuiltin.IsPaymentChannelActor(act.Code) && msgIn.Method == builtin.MethodsPaych.Collect {
-			// add the refunded gas for DestroyActor back into the gas used
-			ret += 76e3
-		}
-	}
-
 	return ret, nil
 }