Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: recompute tipset to generate missing events if event indexing is enabled #12463

Merged
merged 4 commits into from
Sep 16, 2024
Merged

feat: recompute tipset to generate missing events if event indexing is enabled #12463

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6501c3a
auto repair events
aarshkshah1992 Sep 14, 2024
bd7a132
make jen
aarshkshah1992 Sep 16, 2024
3070eaf
fix leaky abstraction
aarshkshah1992 Sep 16, 2024
9ada38e
Merge branch 'feat/msg-eth-tx-index' into feat/auto-repair-event-storage
aarshkshah1992 Sep 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion chain/index/events.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
6 changes: 6 additions & 0 deletions chain/index/indexer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Leaky abstraction here, since the two CIDs are always discarded. A couple of options to make this a bit nicer:

  1. Just use an error return and make the wrapper function 1 line longer to discard the others (even though currently it doesn't even need to be a wrapper, just indexer.SetTipsetExecutorFnc(sm.RecomputeTipSetState) would do).
  2. Just call it recomputeTipSetState and admit that this is all we're expecting here.
  3. Make SqliteIndexer depend on StateManager and load it in via fx to avoid jumping through hoops and to be completely honest.

(also, Fnc -> Func?)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rvagg Fixed this by using a recomputeTipSetState function here that simply returns an error and passing in a wrapper to hide the AMT cids during construction.


type preparedStatements struct {
insertEthTxHashStmt *sql.Stmt
Expand Down Expand Up @@ -53,6 +54,7 @@ type SqliteIndexer struct {
cs ChainStore

idToRobustAddrFunc IdToRobustAddrFunc
tipsetExecutorFnc tipsetExecutorFnc

stmts *preparedStatements

Expand Down Expand Up @@ -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()
Expand Down
1 change: 1 addition & 0 deletions chain/index/interface.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
5 changes: 5 additions & 0 deletions node/modules/chainindex.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"path/filepath"

"github.com/ipfs/go-cid"
"go.uber.org/fx"
"golang.org/x/xerrors"

Expand Down Expand Up @@ -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
Expand Down
Loading