[ic-1047] feat: data pipeline - #371
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a protobuf-defined MQ event stream with Kafka-backed server and client binaries, event serialization utilities, and a detector that monitors topic heights and requests missing blocks. It also adds reusable encoding configuration construction and MQ build targets. ChangesMQ event pipeline
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant StreamClient
participant StreamServer
participant Kafka
StreamClient->>StreamServer: EventStream request
StreamServer->>Kafka: Poll requested topic
Kafka-->>StreamServer: EventStreamResponse record
StreamServer-->>StreamClient: Stream decoded response
StreamClient->>StreamClient: Persist and print event
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
|
Review the following changes in direct dependencies. Learn more about Socket for GitHub.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (4)
proto/injective/mq/v1beta1/query.proto (1)
32-35: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueResolve or remove the outstanding TODO on a wire-level enum.
AdditionalDataTypesships with a singleEVM = 0value and an unresolved// todo: still needed?comment. Since this is a public protobuf contract consumed by generated client/server code, it's worth confirming intent before merge.Do you want me to help evaluate whether
AdditionalDataTypes/AdditionalDataEntryis still needed, or should this ship as-is with the TODO tracked separately?🤖 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 `@proto/injective/mq/v1beta1/query.proto` around lines 32 - 35, Resolve the TODO around the public protobuf enum AdditionalDataTypes before merging: confirm whether AdditionalDataTypes and its associated AdditionalDataEntry are required by the wire contract; if they are unused, remove the enum and related definitions safely, otherwise retain them and remove the TODO while documenting or preserving their intended contract.go.mod (1)
41-41: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick win
golang.org/x/crypto v0.41.0has several documented CVEs — bump the version.OSV flags multiple critical advisories against this version (e.g. GHSA-j5w8-q4qc-rx2x / CVE-2025-58181, GHSA-f6x5-jh6r-wrfv / CVE-2025-47914), with fixes starting in v0.45.0 and further hardening in later releases.
📦 Suggested dependency bump
- golang.org/x/crypto v0.41.0 + golang.org/x/crypto v0.52.0Please confirm the latest stable patched version and run
go get -u golang.org/x/crypto && go mod tidybefore merging.🤖 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 `@go.mod` at line 41, Update the golang.org/x/crypto dependency from v0.41.0 to the latest stable patched version, at least v0.45.0, then run go get -u golang.org/x/crypto and go mod tidy to refresh the module metadata and checksums.client/mq/stream/server/main.go (1)
84-97: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider
GracefulStop()instead ofStop()on shutdown.On
ctx.Done(), the server callsgrpcServer.Stop(), which immediately terminates all active streams. For a long-lived event-streaming RPC,GracefulStop()(optionally with a bounded timeout via a secondary hardStop()) would let in-flight subscribers drain more cleanly during rolling restarts/SIGTERM.♻️ Suggested pattern
case <-ctx.Done(): logger.Info("stopping event stream server") - grpcServer.Stop() + stopped := make(chan struct{}) + go func() { + grpcServer.GracefulStop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(shutdownGracePeriod): + grpcServer.Stop() + } if err := context.Cause(ctx); err != nil && !errors.Is(err, context.Canceled) {🤖 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 `@client/mq/stream/server/main.go` around lines 84 - 97, Update the ctx.Done() shutdown branch in the event stream server to use grpcServer.GracefulStop() so active event-streaming RPCs can drain before termination. If bounded shutdown is required by the existing lifecycle, add a timeout that invokes the existing hard Stop() as a fallback, while preserving the current context-cause handling and serveErrCh behavior.client/mq/detector/config.go (1)
49-55: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winTrim split broker/node entries.
strings.Splitdoesn't trim whitespace;"broker-1:9092, broker-2:9092"yields" broker-2:9092", which passesValidate()'s non-empty check (Line 70-72) but breaks the actual Kafka/HTTP connection since it's an untrimmed value with a leading space. Same issue applies toFullNodes(Line 91-95).Proposed fix
if *kafkaBrokers != "" { - cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") + for _, b := range strings.Split(*kafkaBrokers, ",") { + cfg.KafkaBrokers = append(cfg.KafkaBrokers, strings.TrimSpace(b)) + } } if *fullNodes != "" { - cfg.FullNodes = strings.Split(*fullNodes, ",") + for _, n := range strings.Split(*fullNodes, ",") { + cfg.FullNodes = append(cfg.FullNodes, strings.TrimSpace(n)) + } }🤖 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 `@client/mq/detector/config.go` around lines 49 - 55, Trim whitespace from each entry produced when parsing the kafkaBrokers and fullNodes flags in the configuration-loading flow. Update the logic around the KafkaBrokers and FullNodes assignments to apply per-entry trimming after splitting, while preserving the existing empty-flag behavior and resulting slice structure.
🤖 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 `@client/mq/detector/main_test.go`:
- Line 32: Fix both requestControlPlaneBlocks calls in main_test.go by matching
the function signature: pass the start height value before server.URL, followed
by the control token. Keep the existing context and client arguments unchanged.
In `@client/mq/detector/main.go`:
- Around line 117-126: Update callControlPlane and its call sites
noLatestForAWhile and the gap-check path to prevent overlapping duplicate
recovery requests. Track in-flight or recently requested state keyed by the
relevant height, and only issue another request after the existing request
completes or an appropriate cooldown expires while preserving recovery for new
heights.
- Around line 195-204: Update the gap-detection logic after building
sortedHeights so it scans adjacent sorted heights for any missing value, rather
than checking only sortedHeights[0] against latestHeight. When an internal gap
is found, call callControlPlane for the first missing height; preserve the
existing latestHeight-based detection for gaps before the smallest observed
height.
- Around line 118-125: Update the callControlPlane closure’s node loop so
logger.Info("requested blocks from node", ...) executes only after
requestControlPlaneBlocks succeeds; when it returns an error, log the error and
skip the success message for that node.
- Around line 128-181: Add a timer-based select case to the loop containing
noLatestForAWhile so timeout recovery is evaluated while both topic channels are
silent. Use a timer or ticker derived from config.MessageTimeout, invoke
callControlPlane with latestHeight + 1 when the timeout elapses, and reset or
stop the timer correctly when latestTopicCh receives a message to preserve the
existing last-seen timeout semantics.
In `@client/mq/stream/client/util.go`:
- Around line 86-106: Update transformEventSet to handle decodePublishEvent
failures like printPublishEvent: log the malformed or unregistered event and
continue processing the remaining published events instead of returning an
error. Adjust writeEventsFile and the related tests, including
TestTransformEventSetReturnsPublishDecodeError and
TestWriteEventsFileReturnsTransformErrors, to preserve successful streaming when
an individual event cannot be decoded.
In `@client/mq/stream/server/config.go`:
- Around line 66-90: Trim each Kafka broker entry when populating
cfg.KafkaBrokers in the configuration-loading flow, rather than only trimming
temporarily in mqStreamConfig.Validate. Ensure values such as " broker2:9092"
are stored as "broker2:9092" before they are passed to downstream consumers,
while preserving validation of empty entries.
---
Nitpick comments:
In `@client/mq/detector/config.go`:
- Around line 49-55: Trim whitespace from each entry produced when parsing the
kafkaBrokers and fullNodes flags in the configuration-loading flow. Update the
logic around the KafkaBrokers and FullNodes assignments to apply per-entry
trimming after splitting, while preserving the existing empty-flag behavior and
resulting slice structure.
In `@client/mq/stream/server/main.go`:
- Around line 84-97: Update the ctx.Done() shutdown branch in the event stream
server to use grpcServer.GracefulStop() so active event-streaming RPCs can drain
before termination. If bounded shutdown is required by the existing lifecycle,
add a timeout that invokes the existing hard Stop() as a fallback, while
preserving the current context-cause handling and serveErrCh behavior.
In `@go.mod`:
- Line 41: Update the golang.org/x/crypto dependency from v0.41.0 to the latest
stable patched version, at least v0.45.0, then run go get -u golang.org/x/crypto
and go mod tidy to refresh the module metadata and checksums.
In `@proto/injective/mq/v1beta1/query.proto`:
- Around line 32-35: Resolve the TODO around the public protobuf enum
AdditionalDataTypes before merging: confirm whether AdditionalDataTypes and its
associated AdditionalDataEntry are required by the wire contract; if they are
unused, remove the enum and related definitions safely, otherwise retain them
and remove the TODO while documenting or preserving their intended contract.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 4b47e3d3-d600-4311-9018-e5a847632ef8
⛔ Files ignored due to path filters (3)
chain/mq/types/query.pb.gois excluded by!**/*.pb.gogo.sumis excluded by!**/*.sumproto/buf.lockis excluded by!**/*.lock
📒 Files selected for processing (14)
Makefileclient/mq/detector/config.goclient/mq/detector/config_test.goclient/mq/detector/main.goclient/mq/detector/main_test.goclient/mq/stream/client/config.goclient/mq/stream/client/main.goclient/mq/stream/client/util.goclient/mq/stream/client/util_test.goclient/mq/stream/server/config.goclient/mq/stream/server/main.gogo.modproto/buf.yamlproto/injective/mq/v1beta1/query.proto
| sortedHeights := make([]int64, 0, len(rawHeights)) | ||
| for h := range rawHeights { | ||
| sortedHeights = append(sortedHeights, h) | ||
| } | ||
|
|
||
| slices.Sort(sortedHeights) | ||
|
|
||
| if thereIsAGap := latestHeight+1 < sortedHeights[0]; thereIsAGap { | ||
| go callControlPlane(uint64(latestHeight) + 1) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Gap check only inspects the smallest raw height, missing internal gaps.
thereIsAGap only compares latestHeight+1 against sortedHeights[0]. If rawHeights contains e.g. {101, 103} while latestHeight == 100, height 102 is genuinely missing but no gap is detected until latestHeight itself later advances to 101 (pruning 101 out and exposing the 102/103 gap). Detection is thus delayed by however long it takes the "latest" pointer to catch up, rather than firing as soon as the internal gap is observable in rawHeights.
Proposed fix
slices.Sort(sortedHeights)
- if thereIsAGap := latestHeight+1 < sortedHeights[0]; thereIsAGap {
- go callControlPlane(uint64(latestHeight) + 1)
- }
+ if thereIsAGap := latestHeight+1 < sortedHeights[0]; thereIsAGap {
+ go callControlPlane(uint64(latestHeight) + 1)
+ } else {
+ for i := 1; i < len(sortedHeights); i++ {
+ if sortedHeights[i] > sortedHeights[i-1]+1 {
+ go callControlPlane(uint64(sortedHeights[i-1]) + 1)
+ break
+ }
+ }
+ }📝 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.
| sortedHeights := make([]int64, 0, len(rawHeights)) | |
| for h := range rawHeights { | |
| sortedHeights = append(sortedHeights, h) | |
| } | |
| slices.Sort(sortedHeights) | |
| if thereIsAGap := latestHeight+1 < sortedHeights[0]; thereIsAGap { | |
| go callControlPlane(uint64(latestHeight) + 1) | |
| } | |
| sortedHeights := make([]int64, 0, len(rawHeights)) | |
| for h := range rawHeights { | |
| sortedHeights = append(sortedHeights, h) | |
| } | |
| slices.Sort(sortedHeights) | |
| if thereIsAGap := latestHeight+1 < sortedHeights[0]; thereIsAGap { | |
| go callControlPlane(uint64(latestHeight) + 1) | |
| } else { | |
| for i := 1; i < len(sortedHeights); i++ { | |
| if sortedHeights[i] > sortedHeights[i-1]+1 { | |
| go callControlPlane(uint64(sortedHeights[i-1]) + 1) | |
| break | |
| } | |
| } | |
| } |
🤖 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 `@client/mq/detector/main.go` around lines 195 - 204, Update the gap-detection
logic after building sortedHeights so it scans adjacent sorted heights for any
missing value, rather than checking only sortedHeights[0] against latestHeight.
When an internal gap is found, call callControlPlane for the first missing
height; preserve the existing latestHeight-based detection for gaps before the
smallest observed height.
| if *kafkaBrokers != "" { | ||
| cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") | ||
| } | ||
|
|
||
| if err := cfg.Validate(); err != nil { | ||
| return mqStreamConfig{}, err | ||
| } | ||
|
|
||
| return cfg, nil | ||
| } | ||
|
|
||
| func (cfg mqStreamConfig) Validate() error { | ||
| if strings.TrimSpace(cfg.ListenAddress) == "" { | ||
| return errors.New("invalid MQ stream config: listen address cannot be empty") | ||
| } | ||
|
|
||
| if len(cfg.KafkaBrokers) == 0 { | ||
| return errors.New("invalid MQ stream config: no Kafka brokers specified") | ||
| } | ||
|
|
||
| for i, broker := range cfg.KafkaBrokers { | ||
| if strings.TrimSpace(broker) == "" { | ||
| return fmt.Errorf("invalid MQ stream config: Kafka broker #%d is empty", i+1) | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Kafka broker entries aren't trimmed before use, only during validation.
cfg.KafkaBrokers is populated by a raw strings.Split without trimming. Validate() checks strings.TrimSpace(broker) == "" but never writes the trimmed value back, so a broker list like "broker1:9092, broker2:9092" passes validation yet stores " broker2:9092" (leading space) — this will be handed to kgo.SeedBrokers in main.go and likely fail to connect.
🔧 Proposed fix
if *kafkaBrokers != "" {
- cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",")
+ for _, broker := range strings.Split(*kafkaBrokers, ",") {
+ cfg.KafkaBrokers = append(cfg.KafkaBrokers, strings.TrimSpace(broker))
+ }
}📝 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.
| if *kafkaBrokers != "" { | |
| cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") | |
| } | |
| if err := cfg.Validate(); err != nil { | |
| return mqStreamConfig{}, err | |
| } | |
| return cfg, nil | |
| } | |
| func (cfg mqStreamConfig) Validate() error { | |
| if strings.TrimSpace(cfg.ListenAddress) == "" { | |
| return errors.New("invalid MQ stream config: listen address cannot be empty") | |
| } | |
| if len(cfg.KafkaBrokers) == 0 { | |
| return errors.New("invalid MQ stream config: no Kafka brokers specified") | |
| } | |
| for i, broker := range cfg.KafkaBrokers { | |
| if strings.TrimSpace(broker) == "" { | |
| return fmt.Errorf("invalid MQ stream config: Kafka broker #%d is empty", i+1) | |
| } | |
| } | |
| if *kafkaBrokers != "" { | |
| for _, broker := range strings.Split(*kafkaBrokers, ",") { | |
| cfg.KafkaBrokers = append(cfg.KafkaBrokers, strings.TrimSpace(broker)) | |
| } | |
| } |
🤖 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 `@client/mq/stream/server/config.go` around lines 66 - 90, Trim each Kafka
broker entry when populating cfg.KafkaBrokers in the configuration-loading flow,
rather than only trimming temporarily in mqStreamConfig.Validate. Ensure values
such as " broker2:9092" are stored as "broker2:9092" before they are passed to
downstream consumers, while preserving validation of empty entries.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
client/mq/stream/client/util_test.go (1)
123-133: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winUpdate these tests if the malformed-event drop behavior changes.
TestTransformEventSetSkipsMalformedPublishEventandTestWriteEventsFileSkipsMalformedPublishEventsassert that malformed publish events are fully omitted from output. IftransformEventSetis changed to retain a placeholder entry for undecodable events (see the corresponding comment inclient/mq/stream/client/util.goat lines 86-100), these assertions need to change fromrequire.Emptyto check for the placeholder entry instead.Also applies to: 343-378
🤖 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 `@client/mq/stream/client/util_test.go` around lines 123 - 133, Update TestTransformEventSetSkipsMalformedPublishEvent and TestWriteEventsFileSkipsMalformedPublishEvents to match the revised transformEventSet behavior: assert that malformed publish events produce the expected placeholder entry rather than requiring empty output. Keep the assertions focused on the placeholder’s presence and relevant contents.
🤖 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 `@client/mq/stream/client/util.go`:
- Around line 86-100: Update transformEventSet so decodePublishEvent failures
retain a placeholder publishEventFile containing the malformed event’s type URL
and error detail instead of continuing without appending; preserve the existing
stderr logging and successful decoding behavior. Update
TestTransformEventSetSkipsMalformedPublishEvent and
TestWriteEventsFileSkipsMalformedPublishEvents to assert the placeholder is
persisted in PublishEvents and the written JSON.
---
Nitpick comments:
In `@client/mq/stream/client/util_test.go`:
- Around line 123-133: Update TestTransformEventSetSkipsMalformedPublishEvent
and TestWriteEventsFileSkipsMalformedPublishEvents to match the revised
transformEventSet behavior: assert that malformed publish events produce the
expected placeholder entry rather than requiring empty output. Keep the
assertions focused on the placeholder’s presence and relevant contents.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 9bb356ad-9477-490e-afde-2f2d1cf5550b
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (6)
Makefileclient/chain/context.goclient/mq/stream/client/main.goclient/mq/stream/client/util.goclient/mq/stream/client/util_test.gogo.mod
🚧 Files skipped from review as they are similar to previous changes (2)
- go.mod
- Makefile
| func transformEventSet(events mqtypes.EventSet, decoder *publishEventDecoder) eventSetEventsFile { | ||
| trueOrders := make([]string, 0, len(events.TrueOrders)) | ||
| for _, order := range events.TrueOrders { | ||
| trueOrders = append(trueOrders, order.String()) | ||
| } | ||
|
|
||
| publishEvents := make([]publishEventFile, 0, len(events.PublishedEvents)) | ||
| for idx, event := range events.PublishedEvents { | ||
| publishEvent, err := decoder.decodePublishEvent(event) | ||
| if err != nil { | ||
| _, _ = fmt.Fprintf(os.Stderr, "error decoding publish event %d: %v\n", idx, err) | ||
| continue | ||
| } | ||
| publishEvents = append(publishEvents, publishEvent) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Malformed publish events are silently dropped from the persisted JSON, not just logged.
When decoder.decodePublishEvent fails, the event is logged to stderr and skipped with continue. It is never added to publishEvents, so the resulting eventSetEventsFile.PublishEvents (and the JSON file written by writeEventsFile) contains no trace that an event was dropped. A downstream consumer reading only the persisted block-*.json file has no way to detect data loss; the stderr log is the only signal, and it is not part of the artifact used for replay/analytics.
Preserve a placeholder entry (type URL plus error detail) in the output instead of dropping the record, so the persisted file stays complete and auditable.
🔧 Suggested fix: keep a placeholder for undecodable events
publishEvents := make([]publishEventFile, 0, len(events.PublishedEvents))
for idx, event := range events.PublishedEvents {
publishEvent, err := decoder.decodePublishEvent(event)
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "error decoding publish event %d: %v\n", idx, err)
- continue
+ publishEvent = publishEventFile{
+ TypeURL: "unknown",
+ Value: json.RawMessage(fmt.Sprintf(`{"error":%q}`, err.Error())),
+ }
}
publishEvents = append(publishEvents, publishEvent)
}Note: this requires updating TestTransformEventSetSkipsMalformedPublishEvent and TestWriteEventsFileSkipsMalformedPublishEvents in util_test.go, which currently assert the drop behavior.
📝 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.
| func transformEventSet(events mqtypes.EventSet, decoder *publishEventDecoder) eventSetEventsFile { | |
| trueOrders := make([]string, 0, len(events.TrueOrders)) | |
| for _, order := range events.TrueOrders { | |
| trueOrders = append(trueOrders, order.String()) | |
| } | |
| publishEvents := make([]publishEventFile, 0, len(events.PublishedEvents)) | |
| for idx, event := range events.PublishedEvents { | |
| publishEvent, err := decoder.decodePublishEvent(event) | |
| if err != nil { | |
| _, _ = fmt.Fprintf(os.Stderr, "error decoding publish event %d: %v\n", idx, err) | |
| continue | |
| } | |
| publishEvents = append(publishEvents, publishEvent) | |
| } | |
| func transformEventSet(events mqtypes.EventSet, decoder *publishEventDecoder) eventSetEventsFile { | |
| trueOrders := make([]string, 0, len(events.TrueOrders)) | |
| for _, order := range events.TrueOrders { | |
| trueOrders = append(trueOrders, order.String()) | |
| } | |
| publishEvents := make([]publishEventFile, 0, len(events.PublishedEvents)) | |
| for idx, event := range events.PublishedEvents { | |
| publishEvent, err := decoder.decodePublishEvent(event) | |
| if err != nil { | |
| _, _ = fmt.Fprintf(os.Stderr, "error decoding publish event %d: %v\n", idx, err) | |
| publishEvent = publishEventFile{ | |
| TypeURL: "unknown", | |
| Value: json.RawMessage(fmt.Sprintf(`{"error":%q}`, err.Error())), | |
| } | |
| } | |
| publishEvents = append(publishEvents, publishEvent) | |
| } |
🤖 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 `@client/mq/stream/client/util.go` around lines 86 - 100, Update
transformEventSet so decodePublishEvent failures retain a placeholder
publishEventFile containing the malformed event’s type URL and error detail
instead of continuing without appending; preserve the existing stderr logging
and successful decoding behavior. Update
TestTransformEventSetSkipsMalformedPublishEvent and
TestWriteEventsFileSkipsMalformedPublishEvents to assert the placeholder is
persisted in PublishEvents and the written JSON.
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
client/mq/stream/client/config.go (1)
33-52: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winAllow file-backed consumer IDs.
startMQStreamClientcallscfg.Validate()beforeresolveConsumerID. The current non-emptyConsumerIDrequirement rejects the configuration beforeConsumerIDFilecan supply or persist an ID. This disables the intended fallback path.Require a consumer ID only when both
ConsumerIDandConsumerIDFileare blank, or resolve the ID before validation.Proposed fix
- if strings.TrimSpace(cfg.ConsumerID) == "" { - return errors.New("invalid MQ stream client config: consumer id cannot be empty") + if strings.TrimSpace(cfg.ConsumerID) == "" && strings.TrimSpace(cfg.ConsumerIDFile) == "" { + return errors.New("invalid MQ stream client config: consumer id and consumer id file cannot both be empty") }🤖 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 `@client/mq/stream/client/config.go` around lines 33 - 52, Update mqStreamClientConfig.Validate to permit configurations with a blank ConsumerID when ConsumerIDFile is non-blank; return the missing-consumer-ID error only when both fields are empty. Preserve the existing validation for Address, Topic, and Format, allowing startMQStreamClient to resolve the ID from ConsumerIDFile afterward.
🧹 Nitpick comments (1)
.golangci.yml (1)
45-54: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeep complexity thresholds scoped and enforceable.
These values allow cognitive and cyclomatic complexity of
100, plus functions with100statements or200lines. This effectively disables the checks for most functions. Revive definesfunction-lengthas(maximum statements, maximum lines)and documents much lower defaults. (github.com)Because CI and the Makefile consume
.golangci.yml, this relaxation applies repository-wide. Keep stricter project thresholds and add narrow exclusions for exceptional MQ code, or refactor the affected functions.🤖 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 @.golangci.yml around lines 45 - 54, Update the complexity thresholds in the cyclomatic, cognitive, and function-length entries of .golangci.yml to stricter, enforceable project-appropriate values instead of 100/100/200. Preserve narrow exclusions for exceptional MQ code where needed, or refactor those functions, while keeping CI and Makefile linting effective repository-wide.
🤖 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 `@client/mq/detector/config.go`:
- Line 26: Update resolveConsumerID to read and return the existing consumer ID
from ConsumerIDFile when --consumer-id is omitted; only generate a UUID when the
file has no usable ID, then persist that generated ID to ConsumerIDFile before
returning it. Preserve explicitly supplied consumer IDs without replacing them.
In `@client/mq/stream/server/main.go`:
- Around line 47-49: Normalize each broker address when populating
cfg.KafkaBrokers in the kafkaBrokers handling block: trim surrounding whitespace
from every result of strings.Split before passing the list to NewStreamServer,
while preserving the existing empty-input behavior.
---
Outside diff comments:
In `@client/mq/stream/client/config.go`:
- Around line 33-52: Update mqStreamClientConfig.Validate to permit
configurations with a blank ConsumerID when ConsumerIDFile is non-blank; return
the missing-consumer-ID error only when both fields are empty. Preserve the
existing validation for Address, Topic, and Format, allowing startMQStreamClient
to resolve the ID from ConsumerIDFile afterward.
---
Nitpick comments:
In @.golangci.yml:
- Around line 45-54: Update the complexity thresholds in the cyclomatic,
cognitive, and function-length entries of .golangci.yml to stricter, enforceable
project-appropriate values instead of 100/100/200. Preserve narrow exclusions
for exceptional MQ code where needed, or refactor those functions, while keeping
CI and Makefile linting effective repository-wide.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 06736343-ad0a-4624-9c64-a77a0e5cabb8
📒 Files selected for processing (7)
.golangci.ymlclient/mq/detector/config.goclient/mq/detector/main.goclient/mq/stream/client/config.goclient/mq/stream/client/main.goclient/mq/stream/server/config.goclient/mq/stream/server/main.go
💤 Files with no reviewable changes (1)
- client/mq/stream/server/config.go
🚧 Files skipped from review as they are similar to previous changes (2)
- client/mq/stream/client/main.go
- client/mq/detector/main.go
| type mqDetectorConfig struct { | ||
| KafkaBrokers []string | ||
| ConsumerID string | ||
| ConsumerIDFile string |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
matches="$(
rg -l --glob '*.go' \
'ConsumerIDFile|resolveConsumerID|ConsumerGroup|ConsumeResetOffset' \
client/mq/detector || true
)"
if [[ -z "$matches" ]]; then
echo "No detector consumer-ID flow found" >&2
exit 1
fi
while IFS= read -r file; do
printf '\n===== %s =====\n' "$file"
rg -n -C 8 \
'ConsumerIDFile|resolveConsumerID|ConsumerGroup|ConsumeResetOffset|ReadFile|WriteFile|Rename' \
"$file"
done <<< "$matches"Repository: InjectiveLabs/sdk-go
Length of output: 2788
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' '===== detector files ====='
git ls-files 'client/mq/detector/*.go'
printf '%s\n' '===== all ConsumerIDFile references ====='
rg -n --glob '*.go' 'ConsumerIDFile' . || true
printf '%s\n' '===== detector configuration and startup ====='
sed -n '1,130p' client/mq/detector/config.go
sed -n '1,130p' client/mq/detector/main.go
printf '%s\n' '===== detector tests ====='
sed -n '1,180p' client/mq/detector/config_test.goRepository: InjectiveLabs/sdk-go
Length of output: 9738
Persist and reuse the generated consumer ID.
When --consumer-id is omitted, resolveConsumerID generates a new UUID on every start. ConsumerIDFile is never read or written, so each restart creates a new Kafka consumer group and may replay retained events from AtStart.
Read an existing ID before generating one, and persist generated IDs to ConsumerIDFile.
🤖 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 `@client/mq/detector/config.go` at line 26, Update resolveConsumerID to read
and return the existing consumer ID from ConsumerIDFile when --consumer-id is
omitted; only generate a UUID when the file has no usable ID, then persist that
generated ID to ConsumerIDFile before returning it. Preserve explicitly supplied
consumer IDs without replacing them.
| if *kafkaBrokers != "" { | ||
| cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Normalize each Kafka broker address.
strings.Split retains whitespace. mqStreamConfig.Validate accepts " broker:9092" because it trims only for validation, but NewStreamServer receives the untrimmed value. A normal comma-separated value with spaces can then fail to connect.
Proposed fix
if *kafkaBrokers != "" {
cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",")
+ for i := range cfg.KafkaBrokers {
+ cfg.KafkaBrokers[i] = strings.TrimSpace(cfg.KafkaBrokers[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.
| if *kafkaBrokers != "" { | |
| cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") | |
| } | |
| if *kafkaBrokers != "" { | |
| cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") | |
| for i := range cfg.KafkaBrokers { | |
| cfg.KafkaBrokers[i] = strings.TrimSpace(cfg.KafkaBrokers[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 `@client/mq/stream/server/main.go` around lines 47 - 49, Normalize each broker
address when populating cfg.KafkaBrokers in the kafkaBrokers handling block:
trim surrounding whitespace from every result of strings.Split before passing
the list to NewStreamServer, while preserving the existing empty-input behavior.
1 participant
Ports
detectorand gRPC stream server/client impls from https://github.com/InjectiveLabs/injective-core/pull/2823Summary by CodeRabbit