feat: add ChipIngress batch emitter support#21327
Conversation
|
👋 thomaska, thanks for creating this pull request! To help reviewers, please consider creating future PRs as drafts first. This allows you to self-review and make any final changes before notifying the team. Once you're ready, you can mark it as "Ready for review" to request feedback. Thanks! |
|
✅ No conflicts with other open PRs targeting |
There was a problem hiding this comment.
Pull request overview
Adds PublishBatch support to the chip-testsink gRPC server so CRE/system tests don’t fail with UNIMPLEMENTED when nodes emit batched ChIP ingress events.
Changes:
- Implement
PublishBatchon the chip-testsinkChipIngressServer. - Delegate batch handling to the existing
Publishflow (including configuredPublishFuncand optional upstream forwarding).
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| } | ||
|
|
||
| for _, event := range batch.Events { | ||
| if _, err := s.Publish(ctx, event); err != nil { |
There was a problem hiding this comment.
Calling s.Publish() inside the batch loop triggers the per-event async upstream forwarding goroutine in Publish(). For large batches this can create a burst of goroutines and N upstream RPCs. Consider handling upstream forwarding in PublishBatch with a single PublishBatch call (or at least a bounded worker/pool), and calling the configured PublishFunc directly for local handling to avoid unbounded goroutine/RPC fan-out per batch.
| if _, err := s.Publish(ctx, event); err != nil { | |
| // Forward upstream synchronously to avoid spawning a goroutine per event. | |
| if s.cfg.UpstreamEndpoint != "" { | |
| forwardCtx, cancelFn := context.WithTimeout(context.Background(), 10*time.Second) | |
| _, err := s.upstream.Publish(forwardCtx, event) | |
| cancelFn() | |
| if err != nil { | |
| log.Printf("failed to forward to upstream: %v", err) | |
| } | |
| } | |
| if _, err := s.cfg.PublishFunc(ctx, event); err != nil { |
| // It delegates each event in the batch to the configured PublishFunc, | ||
| // mirroring how the real ChIP Ingress processes batches atomically. |
There was a problem hiding this comment.
The doc comment claims this mirrors how the real ChIP ingress processes batches "atomically", but this implementation is not atomic: it publishes events one-by-one and can return an error after earlier events have already been accepted/forwarded. Please either adjust the comment to reflect best-effort sequential processing, or change the implementation to provide the atomicity guarantees being documented.
| // It delegates each event in the batch to the configured PublishFunc, | |
| // mirroring how the real ChIP Ingress processes batches atomically. | |
| // It delegates each event in the batch to the configured PublishFunc | |
| // sequentially, returning an error on the first failure. Earlier events | |
| // in the batch may already have been published or forwarded when an error | |
| // is returned, so processing is best-effort rather than atomic. |
|
I see you updated files related to
|
|
thomaska
changed the title
Ticket: https://smartcontract-it.atlassian.net/browse/INFOPLAT-3436
Summary
ChipIngressBatchEmitterEnabledtelemetry config flag (defaultfalse)to toggle batch mode per-node without a code change
PublishBatchon the chip-testsink gRPC server so CRE system testswork when batch mode is enabled
chainlink-commonto include batch-emitter feature-flag supportDetail
Config flag – new
ChipIngressBatchEmitterEnabledboolean in[Telemetry].Wired through the config interface, TOML types, beholder globals, docs, and
test fixtures.
chip-testsink – the test-helper server only had single-event
Publish,inheriting
UNIMPLEMENTEDforPublishBatch. Now delegates each batch event tothe configured
PublishFuncand forwards the full batch upstream in one RPC.Why
9 CRE system tests depend on chip-testsink. Without this change they get gRPC
UNIMPLEMENTEDerrors once batch mode is the default.Requires
smartcontractkit/chainlink-common#1862