feat: recompute tipset to generate missing events if event indexing i…

…s enabled (#12463)

* auto repair events

* make jen

* fix leaky abstraction

chain/index/events.go

@@ -163,7 +163,21 @@ func (si *SqliteIndexer) loadExecutedMessages(ctx context.Context, msgTs, rctTs
 
 		eventsArr, err := amt4.LoadAMT(ctx, st, *rct.EventsRoot, amt4.UseTreeBitWidth(types.EventAMTBitwidth))
 		if err != nil {
-			return nil, xerrors.Errorf("failed to load events amt: %w", err)
+			if si.recomputeTipSetStateFunc == nil {
+				return nil, xerrors.Errorf("failed to load events amt for message %s: %w", ems[i].msg.Cid(), err)
+			}
+			log.Warnf("failed to load events amt for message %s: %s; recomputing tipset state to regenerate events", ems[i].msg.Cid(), err)
+
+			if err := si.recomputeTipSetStateFunc(ctx, msgTs); err != nil {
+				return nil, xerrors.Errorf("failed to recompute missing events; failed to recompute tipset state: %w", err)
+			}
+
+			eventsArr, err = amt4.LoadAMT(ctx, st, *rct.EventsRoot, amt4.UseTreeBitWidth(types.EventAMTBitwidth))
+			if err != nil {
+				return nil, xerrors.Errorf("failed to load events amt for message %s: %w", ems[i].msg.Cid(), err)
+			}
+
+			log.Infof("successfully recomputed tipset state and loaded events amt for message %s", ems[i].msg.Cid())
 		}
 
 		ems[i].evs = make([]types.Event, eventsArr.Len())

chain/index/indexer.go

@@ -21,6 +21,7 @@ var _ Indexer = (*SqliteIndexer)(nil)
 
 // IdToRobustAddrFunc is a function type that resolves an actor ID to a robust address
 type IdToRobustAddrFunc func(ctx context.Context, emitter abi.ActorID, ts *types.TipSet) (address.Address, bool)
+type recomputeTipSetStateFunc func(ctx context.Context, ts *types.TipSet) error
 
 type preparedStatements struct {
 	insertEthTxHashStmt                   *sql.Stmt
@@ -52,7 +53,8 @@ type SqliteIndexer struct {
 	db *sql.DB
 	cs ChainStore
 
-	idToRobustAddrFunc IdToRobustAddrFunc
+	idToRobustAddrFunc       IdToRobustAddrFunc
+	recomputeTipSetStateFunc recomputeTipSetStateFunc
 
 	stmts *preparedStatements
 
@@ -120,6 +122,10 @@ func (si *SqliteIndexer) SetIdToRobustAddrFunc(idToRobustAddrFunc IdToRobustAddr
 	si.idToRobustAddrFunc = idToRobustAddrFunc
 }
 
+func (si *SqliteIndexer) SetRecomputeTipSetStateFunc(recomputeTipSetStateFunc recomputeTipSetStateFunc) {
+	si.recomputeTipSetStateFunc = recomputeTipSetStateFunc
+}
+
 func (si *SqliteIndexer) Close() error {
 	si.closeLk.Lock()
 	defer si.closeLk.Unlock()

chain/index/interface.go

@@ -56,6 +56,7 @@ type Indexer interface {
 	IndexEthTxHash(ctx context.Context, txHash ethtypes.EthHash, c cid.Cid) error
 
 	SetIdToRobustAddrFunc(idToRobustAddrFunc IdToRobustAddrFunc)
+	SetRecomputeTipSetStateFunc(recomputeTipSetStateFunc recomputeTipSetStateFunc)
 	Apply(ctx context.Context, from, to *types.TipSet) error
 	Revert(ctx context.Context, from, to *types.TipSet) error
 

node/modules/chainindex.go

@@ -70,6 +70,11 @@ func InitChainIndexer(lc fx.Lifecycle, mctx helpers.MetricsCtx, indexer index.In
 				return *actor.DelegatedAddress, true
 			})
 
+			indexer.SetRecomputeTipSetStateFunc(func(ctx context.Context, ts *types.TipSet) error {
+				_, _, err := sm.RecomputeTipSetState(ctx, ts)
+				return err
+			})
+
 			ch, err := mp.Updates(ctx)
 			if err != nil {
 				return err