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

Event index should be unique for tipsets #11952

Merged
merged 4 commits into from
May 1, 2024
Merged

Event index should be unique for tipsets #11952

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
569e42b
event index should be unique for tipsets
aarshkshah1992 May 1, 2024
4018b8e
Merge branch 'master' into fix/eth-log-index
aarshkshah1992 May 1, 2024
762a94a
fix formatting
aarshkshah1992 May 1, 2024
cd9215a
migrate to version 5
aarshkshah1992 May 1, 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
10 changes: 7 additions & 3 deletions chain/events/filter/event.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ var _ Filter = (*eventFilter)(nil)
type CollectedEvent struct {
Entries []types.EventEntry
EmitterAddr address.Address // address of emitter
EventIdx int // index of the event within the list of emitted events
EventIdx int // index of the event within the list of emitted events in a given tipset
Reverted bool
Height abi.ChainEpoch
TipSetKey types.TipSetKey // tipset that contained the message
Expand Down Expand Up @@ -94,8 +94,11 @@ func (f *eventFilter) CollectEvents(ctx context.Context, te *TipSetEvents, rever
if err != nil {
return xerrors.Errorf("load executed messages: %w", err)
}

eventCount := 0

for msgIdx, em := range ems {
for evIdx, ev := range em.Events() {
for _, ev := range em.Events() {
// lookup address corresponding to the actor id
addr, found := addressLookups[ev.Emitter]
if !found {
Expand All @@ -119,7 +122,7 @@ func (f *eventFilter) CollectEvents(ctx context.Context, te *TipSetEvents, rever
cev := &CollectedEvent{
Entries: ev.Entries,
EmitterAddr: addr,
EventIdx: evIdx,
EventIdx: eventCount,
Reverted: revert,
Height: te.msgTs.Height(),
TipSetKey: te.msgTs.Key(),
Expand All @@ -141,6 +144,7 @@ func (f *eventFilter) CollectEvents(ctx context.Context, te *TipSetEvents, rever
}
f.collected = append(f.collected, cev)
f.mu.Unlock()
eventCount++
}
}

Expand Down
20 changes: 16 additions & 4 deletions chain/events/filter/index.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,6 +356,16 @@ func (ei *EventIndex) migrateToVersion4(ctx context.Context) error {
}
}

stmtEventIndexUpdate, err := tx.PrepareContext(ctx, "UPDATE event SET event_index = (SELECT COUNT(*) FROM event e2 WHERE e2.tipset_key_cid = event.tipset_key_cid AND e2.id <= event.id) - 1")
rvagg marked this conversation as resolved.
Show resolved Hide resolved
if err != nil {
return xerrors.Errorf("prepare stmtEventIndexUpdate: %w", err)
}

_, err = stmtEventIndexUpdate.ExecContext(ctx)
if err != nil {
return xerrors.Errorf("update event index: %w", err)
}

if _, err = tx.Exec("INSERT OR IGNORE INTO _meta (version) VALUES (4)"); err != nil {
return xerrors.Errorf("increment _meta version: %w", err)
}
Expand Down Expand Up @@ -505,10 +515,11 @@ func (ei *EventIndex) CollectEvents(ctx context.Context, te *TipSetEvents, rever
return xerrors.Errorf("load executed messages: %w", err)
}

eventCount := 0
// iterate over all executed messages in this tipset and insert them into the database if they
// don't exist, otherwise mark them as not reverted
for msgIdx, em := range ems {
for evIdx, ev := range em.Events() {
for _, ev := range em.Events() {
addr, found := addressLookups[ev.Emitter]
if !found {
var ok bool
Expand All @@ -532,7 +543,7 @@ func (ei *EventIndex) CollectEvents(ctx context.Context, te *TipSetEvents, rever
te.msgTs.Key().Bytes(), // tipset_key
tsKeyCid.Bytes(), // tipset_key_cid
addr.Bytes(), // emitter_addr
evIdx, // event_index
eventCount, // event_index
em.Message().Cid().Bytes(), // message_cid
msgIdx, // message_index
).Scan(&entryID)
Expand All @@ -547,7 +558,7 @@ func (ei *EventIndex) CollectEvents(ctx context.Context, te *TipSetEvents, rever
te.msgTs.Key().Bytes(), // tipset_key
tsKeyCid.Bytes(), // tipset_key_cid
addr.Bytes(), // emitter_addr
evIdx, // event_index
eventCount, // event_index
em.Message().Cid().Bytes(), // message_cid
msgIdx, // message_index
false, // reverted
Expand Down Expand Up @@ -582,7 +593,7 @@ func (ei *EventIndex) CollectEvents(ctx context.Context, te *TipSetEvents, rever
te.msgTs.Key().Bytes(), // tipset_key
tsKeyCid.Bytes(), // tipset_key_cid
addr.Bytes(), // emitter_addr
evIdx, // event_index
eventCount, // event_index
em.Message().Cid().Bytes(), // message_cid
msgIdx, // message_index
)
Expand All @@ -600,6 +611,7 @@ func (ei *EventIndex) CollectEvents(ctx context.Context, te *TipSetEvents, rever
log.Warnf("restored %d events but expected only one to exist", rowsAffected)
}
}
eventCount++
}
}

Expand Down