use msgindex as epoch=>tipsetkey index to speed up eth_getBlockByNumber queries

chain/index/interface.go

@@ -33,6 +33,14 @@ type MsgIndex interface {
 	Close() error
 }
 
+// TipsetIndex is the interface to the tipset index
+type TipsetIndex interface {
+	// GetTipsetCID returns the tipset cid for the given epoch
+	GetTipsetCID(ctx context.Context, epoch abi.ChainEpoch) (*cid.Cid, error)
+	// Close closes the index
+	Close() error
+}
+
 type dummyMsgIndex struct{}
 
 func (dummyMsgIndex) GetMsgInfo(ctx context.Context, mCid cid.Cid) (MsgInfo, cid.Cid, error) {
@@ -44,3 +52,15 @@ func (dummyMsgIndex) Close() error {
 }
 
 var DummyMsgIndex MsgIndex = dummyMsgIndex{}
+
+type dummyTipsetIndex struct{}
+
+func (dummyTipsetIndex) GetTipsetCID(ctx context.Context, epoch abi.ChainEpoch) (*cid.Cid, error) {
+	return nil, ErrNotFound
+}
+
+func (dummyTipsetIndex) Close() error {
+	return nil
+}
+
+var DummyTipsetIndex TipsetIndex = dummyTipsetIndex{}

chain/index/msgindex.go

@@ -45,7 +45,6 @@ const (
 	dbqInsertMessage        = "INSERT INTO messages VALUES (?, ?, ?)"
 	dbqDeleteTipsetMessages = "DELETE FROM messages WHERE tipset_cid = ?"
 	dbqGetTipsetByEpoch     = "SELECT tipset_cid FROM messages WHERE epoch = ? LIMIT 1"
-
 	// reconciliation
 	dbqCountMessages         = "SELECT COUNT(*) FROM messages"
 	dbqMinEpoch              = "SELECT MIN(epoch) FROM messages"
@@ -546,6 +545,33 @@ func (x *msgIndex) GetMsgInfo(ctx context.Context, mCid cid.Cid) (MsgInfo, cid.C
 	return msgInfo, xtsCid, nil
 }
 
+func (x *msgIndex) GetTipsetCID(ctx context.Context, epoch abi.ChainEpoch) (*cid.Cid, error) {
+	x.closeLk.RLock()
+	defer x.closeLk.RUnlock()
+
+	if x.closed {
+		return nil, ErrClosed
+	}
+
+	var tipset string
+
+	row := x.selectTipsetStmt.QueryRow(epoch)
+	err := row.Scan(&tipset)
+	switch {
+	case err == sql.ErrNoRows:
+		return nil, ErrNotFound
+
+	case err != nil:
+		return nil, xerrors.Errorf("error querying msgindex database: %w", err)
+	}
+
+	tipsetCid, err := cid.Decode(tipset)
+	if err != nil {
+		return nil, xerrors.Errorf("error decoding tipset cid: %w", err)
+	}
+	return &tipsetCid, nil
+}
+
 func (x *msgIndex) Close() error {
 	x.closeLk.Lock()
 	defer x.closeLk.Unlock()

chain/store/index.go

@@ -6,10 +6,11 @@ import (
 	"strconv"
 	"sync"
 
+	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-state-types/abi"
-
+	"github.com/filecoin-project/lotus/chain/index"
 	"github.com/filecoin-project/lotus/chain/types"
 )
 
@@ -30,16 +31,20 @@ type ChainIndex struct {
 	indexCacheLk sync.Mutex
 	indexCache   map[types.TipSetKey]*lbEntry
 
-	loadTipSet loadTipSetFunc
+	loadTipSet   loadTipSetFunc
+	lookupTipSet lookupTipSetCIDFunc
 
 	skipLength abi.ChainEpoch
 }
 type loadTipSetFunc func(context.Context, types.TipSetKey) (*types.TipSet, error)
 
-func NewChainIndex(lts loadTipSetFunc) *ChainIndex {
+type lookupTipSetCIDFunc func(ctx context.Context, epoch abi.ChainEpoch) (*cid.Cid, error)
+
+func NewChainIndex(lts loadTipSetFunc, luts lookupTipSetCIDFunc) *ChainIndex {
 	return &ChainIndex{
 		indexCache: make(map[types.TipSetKey]*lbEntry, DefaultChainIndexCacheSize),
 		loadTipSet: lts,
+		lookupTipSet:  luts,
 		skipLength: 20,
 	}
 }
@@ -50,6 +55,25 @@ type lbEntry struct {
 }
 
 func (ci *ChainIndex) GetTipsetByHeight(ctx context.Context, from *types.TipSet, to abi.ChainEpoch) (*types.TipSet, error) {
+	if ci.lookupTipSet != nil {
+		tsc, err := ci.lookupTipSet(ctx, to)
+		switch {
+		case err == index.ErrNotFound:
+			// fall through
+		case err != nil:
+			return nil, xerrors.Errorf("failed to load tipset cid: %w", err)
+		default:
+			ts, err := ci.loadTipSet(ctx, types.NewTipSetKey(*tsc))
+			if err != nil {
+				return nil, xerrors.Errorf("failed to load tipset: %w", err)
+			}
+			// make sure that the tipset is correct!
+			if ts.Height() == to {
+				return ts, nil
+			}
+			// otherwise, fall through
+		}
+	}
 	if from.Height()-to <= ci.skipLength {
 		return ci.walkBack(ctx, from, to)
 	}

chain/store/store.go

@@ -161,7 +161,8 @@ func NewChainStore(chainBs bstore.Blockstore, stateBs bstore.Blockstore, ds dsto
 		evtTypeHeadChange: j.RegisterEventType("sync", "head_change"),
 	}
 
-	ci := NewChainIndex(cs.LoadTipSet)
+	// TODO: what's the best way to get the message (tipset) index loaded into here?
+	ci := NewChainIndex(cs.LoadTipSet, nil)
 
 	cs.cindex = ci