[ic-1047] feat: data pipeline for events#91
Open
dbrajovic wants to merge 8 commits into
Open
Conversation
Signed-off-by: 0x0f0d0 <goodbumsu@gmail.com>
Signed-off-by: Nikola Marcetic <n.marcetic86@gmail.com>
Signed-off-by: Nikola Marcetic <n.marcetic86@gmail.com>
…' into ic-1047/data-pipeline
|
@dbrajovic your pull request is missing a changelog! |
📝 WalkthroughWalkthroughThis PR introduces a publish-event pipeline to Cosmos SDK's ChangesPublish Event Pipeline
Sequence Diagram(s)sequenceDiagram
participant internalFinalizeBlock
participant PublishEventManager
participant executeTxs
participant filterOutPublishEvents
participant flushData
internalFinalizeBlock->>PublishEventManager: read beginBlock publish events
internalFinalizeBlock->>publishEvents: accumulate beginBlock events
internalFinalizeBlock->>executeTxs: execute txs, collect per-tx EventSet
executeTxs->>PublishEventManager: reset before each tx
executeTxs->>filterOutPublishEvents: partition tx events
filterOutPublishEvents-->>executeTxs: ABCI events + TrueOrder + PublishEvents
internalFinalizeBlock->>PublishEventManager: read endBlock publish events
internalFinalizeBlock->>filterOutPublishEvents: filter block-level placeholders
internalFinalizeBlock->>flushData: populate BlockEvents, TxEvents, TrueOrder
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
♻️ Duplicate comments (1)
baseapp/baseapp.go (1)
248-248:⚠️ Potential issue | 🔴 CriticalChannel initialization contributes to potential deadlock.
This line initializes
PublishEventsas an unbuffered channel. As flagged in thepublishEvents()method review, sending on this unbuffered channel (line 33 in publish_event.go) can block Commit indefinitely if no receiver is present.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@baseapp/baseapp.go` at line 248, The PublishEvents channel is currently unbuffered causing sends in publishEvents() to potentially block Commit; change the initialization of PublishEvents (the struct field set where PublishEvents is created) from an unbuffered channel to a buffered one (e.g., make(chan PublishEventFlush, N)) with a reasonable capacity (e.g., 100) so producers in Commit won't block if no receiver is immediately ready; update any related comments to note the chosen buffer size and rationale.
🧹 Nitpick comments (1)
types/events.go (1)
393-402: 💤 Low valueConsider simplifying the loop syntax.
The loop at line 397 uses blank identifiers
for _, _ = range eventswhich is unconventional. Since the loop index and value are not used, prefer the cleanerfor range eventssyntax.♻️ Proposed simplification
func (e *EventPlaceholderManager) EmitEvents(events []proto.Message) { e.publishEventManager.EmitEvents(events) placeholders := make(Events, 0, len(events)) - for _, _ = range events { + for range events { placeholders = append(placeholders, NewEvent(PlaceholderEventType)) } e.eventManager.EmitEvents(placeholders) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@types/events.go` around lines 393 - 402, In EventPlaceholderManager.EmitEvents, simplify the loop that builds placeholders: replace the unconventional "for _, _ = range events" with the cleaner "for range events" in the block that appends NewEvent(PlaceholderEventType) to the placeholders slice before calling e.eventManager.EmitEvents; keep the existing placeholders allocation and calls to e.publishEventManager.EmitEvents and e.eventManager.EmitEvents unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@baseapp/publish_event_test.go`:
- Around line 26-28: The Reset method on StringPublishEvent currently uses a
value receiver so assigning to e resets only the local copy; change the receiver
to a pointer (func (e *StringPublishEvent) Reset()) and reset the underlying
value (e.g. assign an empty StringPublishEvent to *e or zero its fields) so the
original struct instance is actually cleared.
- Around line 36-50: The helper getStreamEventFlushChan spawns an unbounded
goroutine forwarding from app.PublishEvents to publishEventChan with no
cancellation, causing leaks; modify getStreamEventFlushChan to accept (or create
and return) a cancel signal (e.g., a done channel or context.Context and a
cleanup function) and use it in the goroutine select alongside app.PublishEvents
and when closed/ctx.Done() triggers, return/exit the goroutine and close
publishEventChan; update callers/tests to call the returned cleanup (or defer
cancel) after getting the chan so the forwarder is stopped when the test
finishes.
- Line 199: Remove the leftover debug print by deleting the
fmt.Println(pevts.TxEvents) call in the test (publish_event_test.go) — locate
the statement that prints pevts.TxEvents (in the test function where pevts is
used) and remove it so the test contains no debug output before merging.
In `@baseapp/publish_event.go`:
- Around line 29-35: publishEvents is blocking on an unbuffered send to
app.PublishEvents (called from Commit()), which can deadlock consensus if no
receiver is present; change the send to a non-blocking send using select with a
default case (or alternatively ensure app.PublishEvents is created buffered),
e.g. in BaseApp.publishEvents replace the direct send of app.flushData with a
select that attempts to send to app.PublishEvents and otherwise drops (and
optionally logs) the event so Commit() cannot block; reference functions/fields:
BaseApp.publishEvents, app.PublishEvents, app.flushData, EnablePublish and the
Commit() caller.
In `@types/context.go`:
- Around line 414-426: The type assertion on pem in writeCache is unsafe and can
panic; change it to the two-value form (e.g., epm, ok :=
pem.(*EventPlaceholderManager)) and only reassign pem to epm.publishEventManager
when ok is true and epm is non-nil, otherwise leave pem as-is (then call
pem.EmitEvents as before). Update the logic around PublishEventManager and
EventPlaceholderManager in writeCache to use this safe check so EmitEvents is
invoked on the correct publishEventManager only when the assertion succeeded.
---
Duplicate comments:
In `@baseapp/baseapp.go`:
- Line 248: The PublishEvents channel is currently unbuffered causing sends in
publishEvents() to potentially block Commit; change the initialization of
PublishEvents (the struct field set where PublishEvents is created) from an
unbuffered channel to a buffered one (e.g., make(chan PublishEventFlush, N))
with a reasonable capacity (e.g., 100) so producers in Commit won't block if no
receiver is immediately ready; update any related comments to note the chosen
buffer size and rationale.
---
Nitpick comments:
In `@types/events.go`:
- Around line 393-402: In EventPlaceholderManager.EmitEvents, simplify the loop
that builds placeholders: replace the unconventional "for _, _ = range events"
with the cleaner "for range events" in the block that appends
NewEvent(PlaceholderEventType) to the placeholders slice before calling
e.eventManager.EmitEvents; keep the existing placeholders allocation and calls
to e.publishEventManager.EmitEvents and e.eventManager.EmitEvents unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 7cfa7d7b-a44b-42c0-99a0-27e546f86216
📒 Files selected for processing (13)
baseapp/abci.gobaseapp/abci_test.gobaseapp/baseapp.gobaseapp/baseapp_test.gobaseapp/publish_event.gobaseapp/publish_event_test.gobaseapp/streaming_test.gobaseapp/utils_test.goserver/config/config.goserver/config/config_test.goserver/config/toml.gotypes/context.gotypes/events.go
Comment on lines
+26
to
+28
| func (e StringPublishEvent) Reset() { | ||
| e = StringPublishEvent{} | ||
| } |
There was a problem hiding this comment.
Reset() method won't actually reset the receiver.
The Reset() method uses a value receiver and assigns to the local copy e, which has no effect on the original struct. This should use a pointer receiver to properly reset the value.
🔧 Proposed fix
-func (e StringPublishEvent) Reset() {
- e = StringPublishEvent{}
+func (e *StringPublishEvent) Reset() {
+ *e = StringPublishEvent{}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func (e StringPublishEvent) Reset() { | |
| e = StringPublishEvent{} | |
| } | |
| func (e *StringPublishEvent) Reset() { | |
| *e = StringPublishEvent{} | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@baseapp/publish_event_test.go` around lines 26 - 28, The Reset method on
StringPublishEvent currently uses a value receiver so assigning to e resets only
the local copy; change the receiver to a pointer (func (e *StringPublishEvent)
Reset()) and reset the underlying value (e.g. assign an empty StringPublishEvent
to *e or zero its fields) so the original struct instance is actually cleared.
Comment on lines
+36
to
+50
| func getStreamEventFlushChan(app *baseapp.BaseApp) chan baseapp.PublishEventFlush { | ||
| app.EnablePublish = true | ||
| publishEventChan := make(chan baseapp.PublishEventFlush) | ||
|
|
||
| go func() { | ||
| for { | ||
| select { | ||
| case e := <-app.PublishEvents: | ||
| publishEventChan <- e | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| return publishEventChan | ||
| } |
There was a problem hiding this comment.
Goroutine leak: no cleanup mechanism for the event forwarder.
The goroutine spawned in getStreamEventFlushChan runs indefinitely with no cancellation mechanism. When tests complete, this goroutine will continue running, causing a leak.
🧹 Recommended fix to add cleanup
Consider using a context or done channel to allow cleanup:
-func getStreamEventFlushChan(app *baseapp.BaseApp) chan baseapp.PublishEventFlush {
+func getStreamEventFlushChan(app *baseapp.BaseApp) (chan baseapp.PublishEventFlush, func()) {
app.EnablePublish = true
publishEventChan := make(chan baseapp.PublishEventFlush)
+ done := make(chan struct{})
go func() {
for {
select {
+ case <-done:
+ return
case e := <-app.PublishEvents:
publishEventChan <- e
}
}
}()
- return publishEventChan
+ cleanup := func() { close(done) }
+ return publishEventChan, cleanup
}Then call defer cleanup() in each test that uses this helper.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func getStreamEventFlushChan(app *baseapp.BaseApp) chan baseapp.PublishEventFlush { | |
| app.EnablePublish = true | |
| publishEventChan := make(chan baseapp.PublishEventFlush) | |
| go func() { | |
| for { | |
| select { | |
| case e := <-app.PublishEvents: | |
| publishEventChan <- e | |
| } | |
| } | |
| }() | |
| return publishEventChan | |
| } | |
| func getStreamEventFlushChan(app *baseapp.BaseApp) (chan baseapp.PublishEventFlush, func()) { | |
| app.EnablePublish = true | |
| publishEventChan := make(chan baseapp.PublishEventFlush) | |
| done := make(chan struct{}) | |
| go func() { | |
| for { | |
| select { | |
| case <-done: | |
| return | |
| case e := <-app.PublishEvents: | |
| publishEventChan <- e | |
| } | |
| } | |
| }() | |
| cleanup := func() { close(done) } | |
| return publishEventChan, cleanup | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@baseapp/publish_event_test.go` around lines 36 - 50, The helper
getStreamEventFlushChan spawns an unbounded goroutine forwarding from
app.PublishEvents to publishEventChan with no cancellation, causing leaks;
modify getStreamEventFlushChan to accept (or create and return) a cancel signal
(e.g., a done channel or context.Context and a cleanup function) and use it in
the goroutine select alongside app.PublishEvents and when closed/ctx.Done()
triggers, return/exit the goroutine and close publishEventChan; update
callers/tests to call the returned cleanup (or defer cancel) after getting the
chan so the forwarder is stopped when the test finishes.
| } | ||
| require.Equal(t, pevts.NewAppHash, res.AppHash) | ||
| require.Len(t, pevts.TxEvents, txPerHeight) | ||
| fmt.Println(pevts.TxEvents) |
There was a problem hiding this comment.
Remove debug output before merging.
The fmt.Println statement on line 199 appears to be debug output left in the code and should be removed.
🗑️ Proposed fix
require.Len(t, pevts.TxEvents, txPerHeight)
- fmt.Println(pevts.TxEvents)
for i := 0; i < txPerHeight; i++ {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| fmt.Println(pevts.TxEvents) | |
| require.Len(t, pevts.TxEvents, txPerHeight) | |
| for i := 0; i < txPerHeight; i++ { |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@baseapp/publish_event_test.go` at line 199, Remove the leftover debug print
by deleting the fmt.Println(pevts.TxEvents) call in the test
(publish_event_test.go) — locate the statement that prints pevts.TxEvents (in
the test function where pevts is used) and remove it so the test contains no
debug output before merging.
Comment on lines
+29
to
+35
| func (app *BaseApp) publishEvents(commitHash []byte) { | ||
| app.flushData.NewAppHash = commitHash | ||
|
|
||
| if app.EnablePublish { | ||
| app.PublishEvents <- app.flushData | ||
| } | ||
| } |
There was a problem hiding this comment.
Channel send can block Commit indefinitely.
At line 33, the code sends on an unbuffered channel (app.PublishEvents <- app.flushData). If EnablePublish is true but no receiver is consuming from PublishEvents, this will deadlock and block the consensus commit flow indefinitely. Since publishEvents is called from Commit() (see baseapp/abci.go line 1033), this could halt the chain.
Consider one of these solutions:
- Use a buffered channel with sufficient capacity
- Use a non-blocking send with
selectanddefault - Document that consumers MUST read from the channel when
EnablePublishis true
🔧 Proposed fix with non-blocking send
func (app *BaseApp) publishEvents(commitHash []byte) {
app.flushData.NewAppHash = commitHash
if app.EnablePublish {
- app.PublishEvents <- app.flushData
+ select {
+ case app.PublishEvents <- app.flushData:
+ // Event published successfully
+ default:
+ app.logger.Warn("PublishEvents channel full, dropping flush data", "height", app.flushData.Height)
+ }
}
}Alternatively, use a buffered channel in baseapp.go:
- PublishEvents: make(chan PublishEventFlush),
+ PublishEvents: make(chan PublishEventFlush, 100),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func (app *BaseApp) publishEvents(commitHash []byte) { | |
| app.flushData.NewAppHash = commitHash | |
| if app.EnablePublish { | |
| app.PublishEvents <- app.flushData | |
| } | |
| } | |
| func (app *BaseApp) publishEvents(commitHash []byte) { | |
| app.flushData.NewAppHash = commitHash | |
| if app.EnablePublish { | |
| select { | |
| case app.PublishEvents <- app.flushData: | |
| // Event published successfully | |
| default: | |
| app.logger.Warn("PublishEvents channel full, dropping flush data", "height", app.flushData.Height) | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@baseapp/publish_event.go` around lines 29 - 35, publishEvents is blocking on
an unbuffered send to app.PublishEvents (called from Commit()), which can
deadlock consensus if no receiver is present; change the send to a non-blocking
send using select with a default case (or alternatively ensure app.PublishEvents
is created buffered), e.g. in BaseApp.publishEvents replace the direct send of
app.flushData with a select that attempts to send to app.PublishEvents and
otherwise drops (and optionally logs) the event so Commit() cannot block;
reference functions/fields: BaseApp.publishEvents, app.PublishEvents,
app.flushData, EnablePublish and the Commit() caller.
Comment on lines
414
to
+426
| writeCache = func() { | ||
| c.EventManager().EmitEvents(cc.EventManager().Events()) | ||
|
|
||
| pem := c.PublishEventManager() | ||
| // pem contains EventPlaceholderManager within the interface and by the time this callback is called | ||
| // it already put placeholder events in the regular EventManager whenever a published event was emitted. | ||
| // We just need to forward the published events to the original context (see PublishEventManager method of sdk Context) | ||
| if pem.(*EventPlaceholderManager) != nil { | ||
| pem = pem.(*EventPlaceholderManager).publishEventManager | ||
| } | ||
|
|
||
| pem.EmitEvents(cc.PublishEventManager().Events()) | ||
|
|
There was a problem hiding this comment.
Type assertion pattern is incorrect and could panic.
At line 421, the code performs a type assertion pem.(*EventPlaceholderManager) without the two-value form and then checks != nil. This pattern is incorrect in Go:
- If
pemis not an*EventPlaceholderManager, the assertion will panic rather than returning nil - The
!= nilcheck is meaningless after a successful assertion—it would have panicked already if the type didn't match
Use the two-value form of type assertion to avoid panics:
🛡️ Proposed fix using safe type assertion
writeCache = func() {
c.EventManager().EmitEvents(cc.EventManager().Events())
pem := c.PublishEventManager()
// pem contains EventPlaceholderManager within the interface and by the time this callback is called
// it already put placeholder events in the regular EventManager whenever a published event was emitted.
// We just need to forward the published events to the original context (see PublishEventManager method of sdk Context)
- if pem.(*EventPlaceholderManager) != nil {
+ if pm, ok := pem.(*EventPlaceholderManager); ok {
- pem = pem.(*EventPlaceholderManager).publishEventManager
+ pem = pm.publishEventManager
}
pem.EmitEvents(cc.PublishEventManager().Events())
cms.Write()
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| writeCache = func() { | |
| c.EventManager().EmitEvents(cc.EventManager().Events()) | |
| pem := c.PublishEventManager() | |
| // pem contains EventPlaceholderManager within the interface and by the time this callback is called | |
| // it already put placeholder events in the regular EventManager whenever a published event was emitted. | |
| // We just need to forward the published events to the original context (see PublishEventManager method of sdk Context) | |
| if pem.(*EventPlaceholderManager) != nil { | |
| pem = pem.(*EventPlaceholderManager).publishEventManager | |
| } | |
| pem.EmitEvents(cc.PublishEventManager().Events()) | |
| writeCache = func() { | |
| c.EventManager().EmitEvents(cc.EventManager().Events()) | |
| pem := c.PublishEventManager() | |
| // pem contains EventPlaceholderManager within the interface and by the time this callback is called | |
| // it already put placeholder events in the regular EventManager whenever a published event was emitted. | |
| // We just need to forward the published events to the original context (see PublishEventManager method of sdk Context) | |
| if pm, ok := pem.(*EventPlaceholderManager); ok { | |
| pem = pm.publishEventManager | |
| } | |
| pem.EmitEvents(cc.PublishEventManager().Events()) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@types/context.go` around lines 414 - 426, The type assertion on pem in
writeCache is unsafe and can panic; change it to the two-value form (e.g., epm,
ok := pem.(*EventPlaceholderManager)) and only reassign pem to
epm.publishEventManager when ok is true and epm is non-nil, otherwise leave pem
as-is (then call pem.EmitEvents as before). Update the logic around
PublishEventManager and EventPlaceholderManager in writeCache to use this safe
check so EmitEvents is invoked on the correct publishEventManager only when the
assertion succeeded.
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 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.
Adds a custom event manager for publishing events to Message Queue consumer (Kafka). Published events are not part of the abci block space and are flushed to the consumer at the end of every block.
Summary by CodeRabbit
Release Notes
New Features
MQPub) for event streaming.Tests