auto repair events

filecoin-project · Sep 14, 2024 · 6501c3a · 6501c3a
1 parent efe90f8
commit 6501c3a
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 1 deletion.
diff --git a/chain/index/events.go b/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("error loading events amt: %w", err)
+			if si.tipsetExecutorFnc == 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)
+
+			_, _, err = si.tipsetExecutorFnc(ctx, msgTs)
+			if 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())

diff --git a/chain/index/indexer.go b/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 tipsetExecutorFnc func(ctx context.Context, ts *types.TipSet) (cid.Cid, cid.Cid, error)
 
 type preparedStatements struct {
 	insertEthTxHashStmt                   *sql.Stmt
@@ -53,6 +54,7 @@ type SqliteIndexer struct {
 	cs ChainStore
 
 	idToRobustAddrFunc IdToRobustAddrFunc
+	tipsetExecutorFnc  tipsetExecutorFnc
 
 	stmts *preparedStatements
 
@@ -120,6 +122,10 @@ func (si *SqliteIndexer) SetIdToRobustAddrFunc(idToRobustAddrFunc IdToRobustAddr
 	si.idToRobustAddrFunc = idToRobustAddrFunc
 }
 
+func (si *SqliteIndexer) SetTipsetExecutorFnc(tipsetExecutorFnc tipsetExecutorFnc) {
+	si.tipsetExecutorFnc = tipsetExecutorFnc
+}
+
 func (si *SqliteIndexer) Close() error {
 	si.closeLk.Lock()
 	defer si.closeLk.Unlock()

diff --git a/chain/index/interface.go b/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)
+	SetTipsetExecutorFnc(tipsetExecutorFnc tipsetExecutorFnc)
 	Apply(ctx context.Context, from, to *types.TipSet) error
 	Revert(ctx context.Context, from, to *types.TipSet) error
 

diff --git a/node/modules/chainindex.go b/node/modules/chainindex.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/ipfs/go-cid"
 
 	"github.com/filecoin-project/lotus/chain/events"
 	"github.com/filecoin-project/lotus/chain/index"
@@ -70,6 +71,10 @@ func InitChainIndexer(lc fx.Lifecycle, mctx helpers.MetricsCtx, indexer index.In
 				return *actor.DelegatedAddress, true
 			})
 
+			indexer.SetTipsetExecutorFnc(func(ctx context.Context, ts *types.TipSet) (cid.Cid, cid.Cid, error) {
+				return sm.RecomputeTipSetState(ctx, ts)
+			})
+
 			ch, err := mp.Updates(ctx)
 			if err != nil {
 				return err