diff --git a/chain/index/interface.go b/chain/index/interface.go index 239abf8fa..2056da848 100644 --- a/chain/index/interface.go +++ b/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{} diff --git a/chain/index/msgindex.go b/chain/index/msgindex.go index 5d6c5cd88..1febb0713 100644 --- a/chain/index/msgindex.go +++ b/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() diff --git a/chain/store/index.go b/chain/store/index.go index 620cb2dee..e4036762e 100644 --- a/chain/store/index.go +++ b/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) } diff --git a/chain/store/store.go b/chain/store/store.go index d7188a7bf..d29555e6f 100644 --- a/chain/store/store.go +++ b/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