feat: a new ChainIndexer to index tipsets, messages and events

chain/events/observer.go

@@ -244,6 +244,24 @@
 	return nil
 }
 
+// ObserveAndBlock registers the observer and returns the current tipset along with a handle function.
+// The observer is guaranteed to observe events starting at this tipset.
+// The returned handle function should be called by the client when it's ready to receive updates.
+//
+// This function should only be called by the client after the observer has been started.
+// Note that the Observer will block all clients from recieving tipset updates until the handle is called.
+func (o *observer) ObserveAndBlock(obs TipSetObserver) (*types.TipSet, func()) {
+	o.lk.Lock()
+	o.observers = append(o.observers, obs)
+	currentHead := o.head
+
+	unlockHandle := func() {
+		o.lk.Unlock()
+	}
+
+	return currentHead, unlockHandle
+}
+
 // Observe registers the observer, and returns the current tipset. The observer is guaranteed to
 // observe events starting at this tipset.
 //

chain/gen/gen.go

@@ -44,6 +44,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"github.com/filecoin-project/lotus/chain/wallet"
+	"github.com/filecoin-project/lotus/chainindex"
 	"github.com/filecoin-project/lotus/cmd/lotus-seed/seed"
 	"github.com/filecoin-project/lotus/genesis"
 	"github.com/filecoin-project/lotus/journal"
@@ -258,7 +259,8 @@ func NewGeneratorWithSectorsAndUpgradeSchedule(numSectors int, us stmgr.UpgradeS
 	//return nil, xerrors.Errorf("creating drand beacon: %w", err)
 	//}
 
-	sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), sys, us, beac, ds, index.DummyMsgIndex)
+	sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), sys, us, beac, ds,
+		index.DummyMsgIndex, chainindex.DummyIndexer)
 	if err != nil {
 		return nil, xerrors.Errorf("initing stmgr: %w", err)
 	}

chain/index/interface.go

@@ -6,36 +6,26 @@ import (
 
 	"github.com/ipfs/go-cid"
 
-	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/lotus/chainindex"
 )
 
 var ErrNotFound = errors.New("message not found")
 var ErrClosed = errors.New("index closed")
 
-// MsgInfo is the Message metadata the index tracks.
-type MsgInfo struct {
-	// the message this record refers to
-	Message cid.Cid
-	// the tipset where this message was included
-	TipSet cid.Cid
-	// the epoch where this message was included
-	Epoch abi.ChainEpoch
-}
-
 // MsgIndex is the interface to the message index
 type MsgIndex interface {
 	// GetMsgInfo retrieves the message metadata through the index.
 	// The lookup is done using the onchain message Cid; that is the signed message Cid
 	// for SECP messages and unsigned message Cid for BLS messages.
-	GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error)
+	GetMsgInfo(ctx context.Context, m cid.Cid) (*chainindex.MsgInfo, error)
 	// Close closes the index
 	Close() error
 }
 
 type dummyMsgIndex struct{}
 
-func (dummyMsgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error) {
-	return MsgInfo{}, ErrNotFound
+func (dummyMsgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (*chainindex.MsgInfo, error) {
+	return nil, ErrNotFound
 }
 
 func (dummyMsgIndex) Close() error {

chain/index/msgindex.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/chainindex"
 	"github.com/filecoin-project/lotus/lib/sqlite"
 )
 
@@ -88,7 +89,7 @@ type headChange struct {
 	app []*types.TipSet
 }
 
-func NewMsgIndex(lctx context.Context, path string, cs ChainStore) (MsgIndex, error) {
+func NewMsgIndex(lctx context.Context, path string, cs ChainStore, enableWrites bool) (MsgIndex, error) {
 	db, exists, err := sqlite.Open(path)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to setup message index db: %w", err)
@@ -124,13 +125,15 @@ func NewMsgIndex(lctx context.Context, path string, cs ChainStore) (MsgIndex, er
 		return nil, xerrors.Errorf("error preparing msgindex database statements: %w", err)
 	}
 
-	rnf := store.WrapHeadChangeCoalescer(
-		msgIndex.onHeadChange,
-		CoalesceMinDelay,
-		CoalesceMaxDelay,
-		CoalesceMergeInterval,
-	)
-	cs.SubscribeHeadChanges(rnf)
+	if enableWrites {
+		rnf := store.WrapHeadChangeCoalescer(
+			msgIndex.onHeadChange,
+			CoalesceMinDelay,
+			CoalesceMaxDelay,
+			CoalesceMergeInterval,
+		)
+		cs.SubscribeHeadChanges(rnf)
+	}
 
 	msgIndex.workers.Add(1)
 	go msgIndex.background(ctx)
@@ -431,12 +434,12 @@ func (x *msgIndex) doApply(ctx context.Context, tx *sql.Tx, ts *types.TipSet) er
 }
 
 // interface
-func (x *msgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error) {
+func (x *msgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (*chainindex.MsgInfo, error) {
 	x.closeLk.RLock()
 	defer x.closeLk.RUnlock()
 
 	if x.closed {
-		return MsgInfo{}, ErrClosed
+		return nil, ErrClosed
 	}
 
 	var (
@@ -449,18 +452,18 @@ func (x *msgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error) {
 	err := row.Scan(&tipset, &epoch)
 	switch {
 	case err == sql.ErrNoRows:
-		return MsgInfo{}, ErrNotFound
+		return nil, ErrNotFound
 
 	case err != nil:
-		return MsgInfo{}, xerrors.Errorf("error querying msgindex database: %w", err)
+		return nil, xerrors.Errorf("error querying msgindex database: %w", err)
 	}
 
 	tipsetCid, err := cid.Decode(tipset)
 	if err != nil {
-		return MsgInfo{}, xerrors.Errorf("error decoding tipset cid: %w", err)
+		return nil, xerrors.Errorf("error decoding tipset cid: %w", err)
 	}
 
-	return MsgInfo{
+	return &chainindex.MsgInfo{
 		Message: m,
 		TipSet:  tipsetCid,
 		Epoch:   abi.ChainEpoch(epoch),

chain/index/msgindex_test.go

@@ -30,7 +30,7 @@ func TestBasicMsgIndex(t *testing.T) {
 	tmp := t.TempDir()
 	t.Cleanup(func() { _ = os.RemoveAll(tmp) })
 
-	msgIndex, err := NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs)
+	msgIndex, err := NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs, true)
 	require.NoError(t, err)
 
 	defer msgIndex.Close() //nolint
@@ -58,7 +58,7 @@ func TestReorgMsgIndex(t *testing.T) {
 	tmp := t.TempDir()
 	t.Cleanup(func() { _ = os.RemoveAll(tmp) })
 
-	msgIndex, err := NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs)
+	msgIndex, err := NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs, true)
 	require.NoError(t, err)
 
 	defer msgIndex.Close() //nolint
@@ -103,7 +103,7 @@ func TestReconcileMsgIndex(t *testing.T) {
 	tmp := t.TempDir()
 	t.Cleanup(func() { _ = os.RemoveAll(tmp) })
 
-	msgIndex, err := NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs)
+	msgIndex, err := NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs, true)
 	require.NoError(t, err)
 
 	for i := 0; i < 10; i++ {
@@ -130,7 +130,7 @@ func TestReconcileMsgIndex(t *testing.T) {
 	require.NoError(t, err)
 
 	// reopen to reconcile
-	msgIndex, err = NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs)
+	msgIndex, err = NewMsgIndex(context.Background(), tmp+"/msgindex.db", cs, true)
 	require.NoError(t, err)
 
 	defer msgIndex.Close() //nolint

chain/stmgr/forks_test.go

@@ -41,6 +41,7 @@ import (
 	. "github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/chainindex"
 	_ "github.com/filecoin-project/lotus/lib/sigs/bls"
 	_ "github.com/filecoin-project/lotus/lib/sigs/secp"
 )
@@ -169,7 +170,7 @@ func TestForkHeightTriggers(t *testing.T) {
 				}
 
 				return st.Flush(ctx)
-			}}}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex)
+			}}}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex, chainindex.DummyIndexer)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -287,7 +288,7 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
 				root cid.Cid, height abi.ChainEpoch, ts *types.TipSet) (cid.Cid, error) {
 				migrationCount++
 				return root, nil
-			}}}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex)
+			}}}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex, chainindex.DummyIndexer)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -519,7 +520,7 @@ func TestForkPreMigration(t *testing.T) {
 					return nil
 				},
 			}}},
-		}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex)
+		}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex, chainindex.DummyIndexer)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -595,6 +596,7 @@ func TestDisablePreMigration(t *testing.T) {
 		cg.BeaconSchedule(),
 		datastore.NewMapDatastore(),
 		index.DummyMsgIndex,
+		chainindex.DummyIndexer,
 	)
 	require.NoError(t, err)
 	require.NoError(t, sm.Start(context.Background()))
@@ -650,6 +652,7 @@ func TestMigrtionCache(t *testing.T) {
 		cg.BeaconSchedule(),
 		metadataDs,
 		index.DummyMsgIndex,
+		chainindex.DummyIndexer,
 	)
 	require.NoError(t, err)
 	require.NoError(t, sm.Start(context.Background()))
@@ -703,6 +706,7 @@ func TestMigrtionCache(t *testing.T) {
 			cg.BeaconSchedule(),
 			metadataDs,
 			index.DummyMsgIndex,
+			chainindex.DummyIndexer,
 		)
 		require.NoError(t, err)
 		sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {

chain/stmgr/searchwait.go

@@ -190,9 +190,13 @@ func (sm *StateManager) SearchForMessage(ctx context.Context, head *types.TipSet
 }
 
 func (sm *StateManager) searchForIndexedMsg(ctx context.Context, mcid cid.Cid, m types.ChainMsg) (*types.TipSet, *types.MessageReceipt, cid.Cid, error) {
-	minfo, err := sm.msgIndex.GetMsgInfo(ctx, mcid)
+	minfo, err := sm.chainIndexer.GetMsgInfo(ctx, mcid)
 	if err != nil {
-		return nil, nil, cid.Undef, xerrors.Errorf("error looking up message in index: %w", err)
+		// If chainIndexer fails, fallback to msgIndex
+		minfo, err = sm.msgIndex.GetMsgInfo(ctx, mcid)
+		if err != nil {
+			return nil, nil, cid.Undef, xerrors.Errorf("error looking up message in indexes: %w", err)
+		}
 	}
 
 	// check the height against the current tipset; minimum execution confidence requires that the

chain/stmgr/stmgr.go

@@ -34,6 +34,7 @@ import (
 	"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/chainindex"
 
 	// Used for genesis.
 	msig0 "github.com/filecoin-project/specs-actors/actors/builtin/multisig"
@@ -156,7 +157,8 @@ type StateManager struct {
 	tsExecMonitor ExecMonitor
 	beacon        beacon.Schedule
 
-	msgIndex index.MsgIndex
+	msgIndex     index.MsgIndex
+	chainIndexer chainindex.Indexer
 
 	// We keep a small cache for calls to ExecutionTrace which helps improve
 	// performance for node operators like exchanges and block explorers
@@ -177,7 +179,8 @@ type tipSetCacheEntry struct {
 	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) {
+func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, beacon beacon.Schedule,
+	metadataDs dstore.Batching, msgIndex index.MsgIndex, chainIndexer chainindex.Indexer) (*StateManager, error) {
 	// If we have upgrades, make sure they're in-order and make sense.
 	if err := us.Validate(); err != nil {
 		return nil, err
@@ -243,12 +246,13 @@ func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder,
 		},
 		compWait:       make(map[string]chan struct{}),
 		msgIndex:       msgIndex,
+		chainIndexer:   chainIndexer,
 		execTraceCache: execTraceCache,
 	}, nil
 }
 
-func NewStateManagerWithUpgradeScheduleAndMonitor(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, b beacon.Schedule, em ExecMonitor, metadataDs dstore.Batching, msgIndex index.MsgIndex) (*StateManager, error) {
-	sm, err := NewStateManager(cs, exec, sys, us, b, metadataDs, msgIndex)
+func NewStateManagerWithUpgradeScheduleAndMonitor(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, b beacon.Schedule, em ExecMonitor, metadataDs dstore.Batching, msgIndex index.MsgIndex, chainIndexer chainindex.Indexer) (*StateManager, error) {
+	sm, err := NewStateManager(cs, exec, sys, us, b, metadataDs, msgIndex, chainIndexer)
 	if err != nil {
 		return nil, err
 	}

chain/store/store_test.go

@@ -23,6 +23,7 @@ import (
 	"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/chainindex"
 	"github.com/filecoin-project/lotus/node/repo"
 )
 
@@ -216,7 +217,8 @@ func TestChainExportImportFull(t *testing.T) {
 		t.Fatal("imported chain differed from exported chain")
 	}
 
-	sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), nil, filcns.DefaultUpgradeSchedule(), cg.BeaconSchedule(), ds, index.DummyMsgIndex)
+	sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), nil, filcns.DefaultUpgradeSchedule(), cg.BeaconSchedule(),
+		ds, index.DummyMsgIndex, chainindex.DummyIndexer)
 	if err != nil {
 		t.Fatal(err)
 	}

chainindex/ddls.go

@@ -0,0 +1,55 @@
+package chainindex
+
+const DefaultDbFilename = "chainindex.db"
+
+const (
+	stmtGetNonRevertedMessageInfo = "SELECT tipset_key_cid, height FROM tipset_message WHERE message_cid = ? AND reverted = 0"
+	stmtGetMsgCidFromEthHash      = "SELECT message_cid FROM eth_tx_hash WHERE tx_hash = ?"
+	stmtInsertEthTxHash           = "INSERT INTO eth_tx_hash (tx_hash, message_cid) VALUES (?, ?) ON CONFLICT (tx_hash) DO UPDATE SET inserted_at = CURRENT_TIMESTAMP"
+
+	stmtInsertTipsetMessage = "INSERT INTO tipset_message (tipset_key_cid, height, reverted, message_cid, message_index) VALUES (?, ?, ?, ?, ?) ON CONFLICT (tipset_key_cid, message_cid) DO UPDATE SET reverted = 0"
+
+	stmtTipsetExists   = "SELECT EXISTS(SELECT 1 FROM tipset_message WHERE tipset_key_cid = ?)"
+	stmtTipsetUnRevert = "UPDATE tipset_message SET reverted = 0 WHERE tipset_key_cid = ?"
+
+	stmtRevertTipset = "UPDATE tipset_message SET reverted = 1 WHERE tipset_key_cid = ?"
+
+	stmtGetMaxNonRevertedTipset = "SELECT tipset_key_cid FROM tipset_message WHERE reverted = 0 ORDER BY height DESC LIMIT 1"
+
+	stmtRemoveRevertedTipsetsBeforeHeight = "DELETE FROM tipset_message WHERE height < ? AND reverted = 1"
+	stmtRemoveTipsetsBeforeHeight         = "DELETE FROM tipset_message WHERE height < ?"
+
+	stmtDeleteEthHashesOlderThan = `DELETE FROM eth_tx_hash WHERE inserted_at < datetime('now', ?);`
+
+	stmtRevertTipsetsFromHeight = "UPDATE tipset_message SET reverted = 1 WHERE height >= ?"
+
+	stmtCountMessages = "SELECT COUNT(*) FROM tipset_message"
+
+	stmtMinNonRevertedHeight = `SELECT MIN(height) FROM tipset_message WHERE reverted = 0`
+
+	stmtTipsetExistsNotReverted = `SELECT EXISTS(SELECT 1 FROM tipset_message WHERE tipset_key_cid = ? AND reverted = 0)`
+)
+
+var ddls = []string{
+	`CREATE TABLE IF NOT EXISTS tipset_message (
+		message_id INTEGER PRIMARY KEY,
+		tipset_key_cid BLOB NOT NULL,
+		height INTEGER NOT NULL,
+		reverted INTEGER NOT NULL,
+		message_cid BLOB,
+		message_index INTEGER,
+		UNIQUE (tipset_key_cid, message_cid)
+	)`,
+
+	`CREATE TABLE IF NOT EXISTS eth_tx_hash (
+		tx_hash TEXT PRIMARY KEY,
+		message_cid BLOB NOT NULL,
+		inserted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+	)`,
+
+	`CREATE INDEX IF NOT EXISTS insertion_time_index ON eth_tx_hash (inserted_at)`,
+
+	`CREATE INDEX IF NOT EXISTS idx_message_cid ON tipset_message (message_cid)`,
+
+	`CREATE INDEX IF NOT EXISTS idx_tipset_key_cid ON tipset_message (tipset_key_cid)`,
+}