logpoller: refactors and persistent store #219
Draft
+5,120
−2,286
Conversation
This file contains hidden or 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
jadepark-dev
changed the title
ogtownsend
reviewed
Nov 7, 2025
|
|
||
| starting_seq_no INTEGER NOT NULL, | ||
|
|
||
| created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), |
There was a problem hiding this comment.
how is the time zone set for TIMESTAMPTZ? Will everything default to UTC?
ogtownsend
reviewed
Nov 7, 2025
Comment on lines
-59
to
-70
| hasFilter, err := a.logPoller.HasFilter(ctx, name) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to check for filter: %w", err) | ||
| } | ||
| // If filter exists, unregister it first to handle address changes | ||
| if hasFilter { | ||
| if err := a.logPoller.UnregisterFilter(ctx, name); err != nil { | ||
| return fmt.Errorf("failed to unregister logpoller filter: %w", err) | ||
| } | ||
| } | ||
|
|
||
| filter := types.Filter{ |
There was a problem hiding this comment.
I don't think I see this logic being handled inside of logPoller.ReigsterFilter or elsewhere, but maybe I just can't find it. Is it not needed anymore?
ogtownsend
reviewed
Nov 7, 2025
Comment on lines
+216
to
+230
| // Filter at database level using byte-level filtering | ||
| // CCIPMessageSent struct layout: | ||
| // - Message: 40 bytes at offset 0 (Message struct) | ||
| // - DestChainSelector: uint64 (8 bytes) at offset 40 | ||
| // - SequenceNumber: uint64 (8 bytes) at offset 48 | ||
| logs, _, _, err := a.logPoller.NewQuery(). | ||
| WithSource(onrampAddr). | ||
| WithEventSig(hash.CRC32(consts.EventNameCCIPMessageSent)). | ||
| SkipBytes(40). // Skip to DestChainSelector | ||
| FilterBytes(8, query.EQ(binary.BigEndian.AppendUint64(nil, uint64(dest)))). | ||
| FilterBytes(8, | ||
| query.GTE(binary.BigEndian.AppendUint64(nil, uint64(seqNumRange.Start()))), | ||
| query.LTE(binary.BigEndian.AppendUint64(nil, uint64(seqNumRange.End()))), | ||
| WithBocBytes( | ||
| query.SkipBytes(40), | ||
| query.MatchBytes(8, query.WithCondition(binary.BigEndian.AppendUint64(nil, uint64(dest)), primitives.Eq)), | ||
| query.MatchBytes(8, | ||
| query.WithCondition(binary.BigEndian.AppendUint64(nil, uint64(seqNumRange.Start())), primitives.Gte), | ||
| query.WithCondition(binary.BigEndian.AppendUint64(nil, uint64(seqNumRange.End())), primitives.Lte), | ||
| ), |
There was a problem hiding this comment.
no need to address in this PR, but I'm wondering if this super exact byte matching query builder logic deserves its own layer in between the accessor and the log poller in the future. It's almost like a separate go-binding layer but for messages/events instead of contract calls. It is product-specific so it makes sense to keep in the accessor for now, but it might be more maintainable as a standalone layer in the long run 🤷
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
This PR adds persistent store in TON logpoller, improves query interface, and log ingestion pipeline
Next Steps