feat: add ChipIngress batch emitter support

[thomaska]

Ticket: https://smartcontract-it.atlassian.net/browse/INFOPLAT-3436

Summary

• Add ChipIngressBatchEmitterEnabled telemetry config flag (default false)
  to toggle batch mode per-node without a code change
• Implement PublishBatch on the chip-testsink gRPC server so CRE system tests
  work when batch mode is enabled
• Bump chainlink-common to include batch-emitter feature-flag support

Detail

Config flag – new ChipIngressBatchEmitterEnabled boolean 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 UNIMPLEMENTED for PublishBatch. Now delegates each batch event to
the configured PublishFunc and forwards the full batch upstream in one RPC.

Why

9 CRE system tests depend on chip-testsink. Without this change they get gRPC
UNIMPLEMENTED errors once batch mode is the default.

Requires

smartcontractkit/chainlink-common#1862

[github-actions]

👋 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!

[github-actions]

✅ No conflicts with other open PRs targeting develop

[Copilot]

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 PublishBatch on the chip-testsink ChipIngressServer.
• Delegate batch handling to the existing Publish flow (including configured PublishFunc and optional upstream forwarding).

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

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.

Suggested change

	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 {

[Copilot]

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.

Suggested change

	// 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.

[trunk-io]

     

View Full Report ↗︎ ⋅ Docs

[github-actions]

I see you updated files related to core. Please run make gocs in the root directory to add a changeset as well as in the text include at least one of the following tags:

• #added For any new functionality added.
• #breaking_change For any functionality that requires manual action for the node to boot.
• #bugfix For bug fixes.
• #changed For any change to the existing functionality.
• #db_update For any feature that introduces updates to database schema.
• #deprecation_notice For any upcoming deprecation functionality.
• #internal For changesets that need to be excluded from the final changelog.
• #nops For any feature that is NOP facing and needs to be in the official Release Notes for the release.
• #removed For any functionality/config that is removed.
• #updated For any functionality that is updated.
• #wip For any change that is not ready yet and external communication about it should be held off till it is feature complete.

[pkcll]

Based on my thorough review of both PRs, here is a concrete implementation plan for registering ChipIngressBatchEmitter as a managed service in NewApplication.

---

Implementation Plan

Problem

The ChipIngressBatchEmitter is created and started inside beholder.NewGRPCClient() (called from initGlobals in shell.go), but it's not registered in the chainlink application's service list. This means:

• No health check visibility (/health)
• No ordered shutdown participation
• No service lifecycle management

Step 1: Expose the batch emitter from beholder.Client (chainlink-common PR #1862)

Currently in pkg/beholder/client.go, the batchEmitterService is a local variable inside NewGRPCClient:

var batchEmitterService *ChipIngressBatchEmitter

Changes needed:

A. Add a field to the Client struct:

type Client struct {
    // ... existing fields ...
    BatchEmitter *ChipIngressBatchEmitter // nil when batch mode is disabled
}

B. Store the emitter when constructing the client (line ~261):

Change the return statement to include the batch emitter:

return &Client{cfg, logger, tracer, meter, emitter, chipIngressClient, loggerProvider, tracerProvider, meterProvider, messageLoggerProvider, signer, onClose, batchEmitterService}, nil

C. Add a public getter:

// GetChipIngressBatchEmitter returns the batch emitter service, or nil if batch mode is disabled.
func (c *Client) GetChipIngressBatchEmitter() *ChipIngressBatchEmitter {
    if c == nil {
        return nil
    }
    return c.BatchEmitter
}

Step 2: Register the batch emitter in NewApplication (chainlink PR #21327)

In core/services/chainlink/application.go, after telemetryManager is appended to srvcs (around line 447–448), add:

telemetryManager := telemetry.NewManager(cfg.TelemetryIngress(), csaKeystore, globalLogger)
srvcs = append(srvcs, telemetryManager)

// Register ChipIngressBatchEmitter for health checks and ordered shutdown.
// The emitter is already started by beholder.NewGRPCClient during initGlobals;
// appending it here gives us health visibility and ensures Close() runs on shutdown.
if beholderClient := beholder.GetClient(); beholderClient != nil {
    if batchEmitter := beholderClient.GetChipIngressBatchEmitter(); batchEmitter != nil {
        srvcs = append(srvcs, batchEmitter)
    }
}

Step 3: No changes to initGlobals in shell.go

The current PR #21327 changes to shell.go are already correct — they pass ChipIngressBatchEmitterEnabled and ChipIngressLogger through to beholder.Config. Since beholder creates/starts the emitter internally and we just retrieve a reference afterward, initGlobals doesn't need any additional changes.

---

Why this works

Concern	How it's addressed
Initialization order	initGlobals → beholder.NewGRPCClient runs in beforeNode() before NewApplication, so beholder.GetClient() is already populated
Double-start	services.Engine's Start() is idempotent — the service framework calling Start() again is a no-op since it's already running
Nil safety	When ChipIngressBatchEmitterEnabled = false (default), batchEmitterService remains nil, and both nil checks protect against it
Shutdown order	Services in srvcs are stopped in reverse order. The emitter is closed before beholder's own Client.Close() (which closes the gRPC connection), matching the drain-before-disconnect requirement already implemented in PR #1862's reordered Client.Close()
Health checks	ChipIngressBatchEmitter implements services.Service via services.Engine, exposing Name(), Ready(), HealthReport() — all needed for /health

---

Files summary

File	Repo	Change
pkg/beholder/client.go	chainlink-common (PR #1862)	Add BatchEmitter field to Client, store it in constructor, add GetChipIngressBatchEmitter() getter
core/services/chainlink/application.go	chainlink (PR #21327)	After telemetryManager, retrieve and append batch emitter to srvcs

Note: The PR #21327 file list may be incomplete (API results limited to 30 files). You can view the full file list here.

[cl-sonarqube-production]

Quality Gate passed

Issues
1 New issue
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube