Frrist/receipt events schema #1132
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| package messages | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "go.opencensus.io/tag" | ||
|
|
||
| "github.com/filecoin-project/lily/metrics" | ||
| "github.com/filecoin-project/lily/model" | ||
| ) | ||
|
|
||
| type ActorEvent struct { | ||
| tableName struct{} `pg:"actor_events"` | ||
|
|
||
| Height int64 `pg:",pk,notnull,use_zero"` | ||
| StateRoot string `pg:",pk,notnull"` | ||
| MessageCid string `pg:",pk,notnull"` | ||
| EventIndex int64 `pg:",pk,notnull,use_zero"` | ||
|
|
||
| Emitter string `pg:",notnull"` | ||
| Flags []byte `pg:",notnull"` | ||
| Key string `pg:",notnull"` | ||
| Value []byte `pg:",notnull"` | ||
| } | ||
|
|
||
| func (a *ActorEvent) Persist(ctx context.Context, s model.StorageBatch, version model.Version) error { | ||
| ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "actor_events")) | ||
| stop := metrics.Timer(ctx, metrics.PersistDuration) | ||
| defer stop() | ||
|
|
||
| metrics.RecordCount(ctx, metrics.PersistModel, 1) | ||
| return s.PersistModel(ctx, a) | ||
| } | ||
|
|
||
| type ActorEventList []*ActorEvent | ||
|
|
||
| func (al ActorEventList) Persist(ctx context.Context, s model.StorageBatch, version model.Version) error { | ||
| if len(al) == 0 { | ||
| return nil | ||
| } | ||
| ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "actor_events")) | ||
| stop := metrics.Timer(ctx, metrics.PersistDuration) | ||
| defer stop() | ||
|
|
||
| metrics.RecordCount(ctx, metrics.PersistModel, len(al)) | ||
| return s.PersistModel(ctx, al) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| package v1 | ||
|
|
||
| func init() { | ||
| patches.Register( | ||
| 13, | ||
| ` | ||
| CREATE TABLE IF NOT EXISTS {{ .SchemaName | default "public"}}.actor_events ( | ||
| height bigint NOT NULL, | ||
| state_root text NOT NULL, | ||
| event_index bigint NOT NULL, | ||
| message_cid text NOT NULL, | ||
|
|
||
| emitter text NOT NULL, | ||
| flags bytea NOT NULL, | ||
| key text NOT NULL, | ||
| value bytea NOT NULL, | ||
|
|
||
| PRIMARY KEY ("height", "state_root", "event_index", "message_cid") | ||
| ); | ||
| `, | ||
| ) | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| package actorevent | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "math" | ||
|
|
||
| "github.com/filecoin-project/go-address" | ||
| "github.com/filecoin-project/go-amt-ipld/v4" | ||
| "github.com/filecoin-project/lotus/chain/types" | ||
| "github.com/ipfs/go-cid" | ||
| cbg "github.com/whyrusleeping/cbor-gen" | ||
| "go.opentelemetry.io/otel" | ||
| "go.opentelemetry.io/otel/attribute" | ||
| "golang.org/x/xerrors" | ||
|
|
||
| "github.com/filecoin-project/lily/model" | ||
| messagemodel "github.com/filecoin-project/lily/model/messages" | ||
| visormodel "github.com/filecoin-project/lily/model/visor" | ||
| "github.com/filecoin-project/lily/tasks" | ||
| "github.com/filecoin-project/lily/tasks/messages" | ||
| ) | ||
|
|
||
| type Task struct { | ||
| node tasks.DataSource | ||
| } | ||
|
|
||
| func NewTask(node tasks.DataSource) *Task { | ||
| return &Task{ | ||
| node: node, | ||
| } | ||
| } | ||
|
|
||
| func (t *Task) ProcessTipSets(ctx context.Context, current *types.TipSet, executed *types.TipSet) (model.Persistable, *visormodel.ProcessingReport, error) { | ||
| ctx, span := otel.Tracer("").Start(ctx, "ProcessTipSets") | ||
| if span.IsRecording() { | ||
| span.SetAttributes( | ||
| attribute.String("current", current.String()), | ||
| attribute.Int64("current_height", int64(current.Height())), | ||
| attribute.String("executed", executed.String()), | ||
| attribute.Int64("executed_height", int64(executed.Height())), | ||
| attribute.String("processor", "actor_events"), | ||
| ) | ||
| } | ||
| defer span.End() | ||
|
|
||
| report := &visormodel.ProcessingReport{ | ||
| Height: int64(current.Height()), | ||
| StateRoot: current.ParentState().String(), | ||
| } | ||
|
|
||
| blkMsgRect, err := t.node.TipSetMessageReceipts(ctx, current, executed) | ||
| if err != nil { | ||
| report.ErrorsDetected = fmt.Errorf("getting tipset message receipet: %w", err) | ||
| return nil, report, nil | ||
| } | ||
|
|
||
| var ( | ||
| out = make(messagemodel.ActorEventList, 0, len(blkMsgRect)) | ||
| errorsDetected = make([]*messages.MessageError, 0, len(blkMsgRect)) | ||
| msgsSeen = make(map[cid.Cid]bool, len(blkMsgRect)) | ||
| ) | ||
|
|
||
| for _, m := range blkMsgRect { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return nil, nil, fmt.Errorf("context done: %w", ctx.Err()) | ||
| default: | ||
| } | ||
|
|
||
| itr, err := m.Iterator() | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| for itr.HasNext() { | ||
| msg, _, rec := itr.Next() | ||
| if msgsSeen[msg.Cid()] { | ||
| continue | ||
| } | ||
| msgsSeen[msg.Cid()] = true | ||
|
|
||
| if rec.EventsRoot == nil { | ||
| continue | ||
| } | ||
|
|
||
| evtArr, err := amt.LoadAMT(ctx, t.node.Store(), *rec.EventsRoot, amt.UseTreeBitWidth(types.EventAMTBitwidth)) | ||
| if err != nil { | ||
| report.ErrorsDetected = fmt.Errorf("loading actor events amt (%s): %w", *rec.EventsRoot, err) | ||
| return nil, report, nil | ||
| } | ||
|
Comment on lines
+88
to
+92
There was a problem hiding this comment. If this is capable of using Lotus APIs, you can now invoke the There was a problem hiding this comment. Yup this has the full lotus API, in #1137 we call that instead, replacing the amt loading logic in this file with the lotus API method. |
||
| var evt types.Event | ||
|
There was a problem hiding this comment. Ah, if the dependencies are updated, this should already. be using the correct event schema (incl. codec). However, you might want to consider inserting the codec as well into the Lily table, although not a big deal since it's a fixed value 0x55 and we can always backfill once other values become possible. |
||
| err = evtArr.ForEach(ctx, func(evtIdx uint64, deferred *cbg.Deferred) error { | ||
| if evtIdx > math.MaxInt { | ||
| return xerrors.Errorf("too many events") | ||
| } | ||
| if err := evt.UnmarshalCBOR(bytes.NewReader(deferred.Raw)); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| emitter, err := address.NewIDAddress(uint64(evt.Emitter)) | ||
| if err != nil { | ||
| errorsDetected = append(errorsDetected, &messages.MessageError{ | ||
| Cid: msg.Cid(), | ||
| Error: fmt.Sprintf("failed to make ID address from event emitter (%s): %s", evt.Emitter, err), | ||
| }) | ||
| return err | ||
| } | ||
| for _, e := range evt.Entries { | ||
| out = append(out, &messagemodel.ActorEvent{ | ||
| Height: int64(current.Height()), | ||
| StateRoot: current.ParentState().String(), | ||
| MessageCid: msg.Cid().String(), | ||
| EventIndex: int64(evtIdx), | ||
| Emitter: emitter.String(), | ||
| Flags: []byte{e.Flags}, | ||
| Key: e.Key, | ||
| Value: e.Value, | ||
| }) | ||
| } | ||
| return nil | ||
| }) | ||
|
|
||
| if err != nil { | ||
| errorsDetected = append(errorsDetected, &messages.MessageError{ | ||
| Cid: msg.Cid(), | ||
| Error: fmt.Sprintf("loading actor events amt (%s): %s", *rec.EventsRoot, err), | ||
| }) | ||
| continue | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if len(errorsDetected) != 0 { | ||
| report.ErrorsDetected = errorsDetected | ||
| } | ||
|
|
||
| return model.PersistableList{ | ||
| out, | ||
| }, report, nil | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@frrist is this tracking the new event schema which includes the codec? FIP-0049 was updated ~2 weeks ago to add a codec field (value 0x55, RAW) always.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will add the codec value shortly, missed that in my initial implementation. Thanks for the pointer!