feat: add chip-ingress batch config keys to TOML and wire through

.changeset/chip-ingress-send-timeout.md

@@ -0,0 +1,8 @@
+---
+"chainlink": patch
+---
+
+#changed Chip-ingress batch emitter defaults tuned from staging/prod capacity analysis:
+`ChipIngressBufferSize` 1000 → 10000, `ChipIngressMaxBatchSize` 500 → 1000,
+`ChipIngressSendInterval` 100ms → 500ms, `ChipIngressSendTimeout` 3s → 10s,
+`ChipIngressDrainTimeout` 10s → 30s.

core/cmd/shell.go

@@ -155,6 +155,13 @@ func newBeholderClient(
 		ChipIngressEmitterGRPCEndpoint: cfgTelemetry.ChipIngressEndpoint(),
 		ChipIngressInsecureConnection:  cfgTelemetry.ChipIngressInsecureConnection(),
 		ChipIngressBatchEmitterEnabled: cfgTelemetry.ChipIngressBatchEmitterEnabled(),
+		ChipIngressBufferSize:          cfgTelemetry.ChipIngressBufferSize(),
+		ChipIngressMaxBatchSize:        cfgTelemetry.ChipIngressMaxBatchSize(),
+		ChipIngressMaxConcurrentSends:  cfgTelemetry.ChipIngressMaxConcurrentSends(),
+		ChipIngressSendInterval:        cfgTelemetry.ChipIngressSendInterval(),
+		ChipIngressSendTimeout:         cfgTelemetry.ChipIngressSendTimeout(),
+		ChipIngressDrainTimeout:        cfgTelemetry.ChipIngressDrainTimeout(),
+		ChipIngressMaxGRPCRequestSize:  cfgTelemetry.ChipIngressMaxGRPCRequestSize(),
 		ChipIngressLogger:              lggr,
 		LogStreamingEnabled:            cfgTelemetry.LogStreamingEnabled(),
 		LogLevel:                       cfgTelemetry.LogLevel(),

core/config/docs/core.toml

@@ -889,6 +889,20 @@ ChipIngressInsecureConnection = false # Default
 # ChipIngressBatchEmitterEnabled enables batching for chip-ingress events.
 # When false, events are sent individually (legacy behavior).
 ChipIngressBatchEmitterEnabled = true # Default
+# ChipIngressBufferSize is the in-memory queue size for chip-ingress events.
+ChipIngressBufferSize = 10000 # Default
+# ChipIngressMaxBatchSize is the max events per PublishBatch RPC.
+ChipIngressMaxBatchSize = 1000 # Default
+# ChipIngressMaxConcurrentSends limits parallel PublishBatch calls.
+ChipIngressMaxConcurrentSends = 10 # Default
+# ChipIngressSendInterval is the max wait before flushing an incomplete batch.
+ChipIngressSendInterval = '500ms' # Default
+# ChipIngressSendTimeout is the per-RPC timeout for PublishBatch.
+ChipIngressSendTimeout = '10s' # Default
+# ChipIngressDrainTimeout is the max shutdown wait to flush queued events.
+ChipIngressDrainTimeout = '30s' # Default
+# ChipIngressMaxGRPCRequestSize is the max serialized PublishBatch request size in bytes. Batches exceeding this are split before send (min 1 MiB enforced by batch client).
+ChipIngressMaxGRPCRequestSize = 10485760 # Default
 # DurableEmitterEnabled enables persisting outbound CHIP events to Postgres for at-least-once delivery.
 DurableEmitterEnabled = true # Default
 # DurableEmitterRetransmitBatchSize is the number of pending events the durable emitter replays per retransmit tick.

core/config/telemetry_config.go

@@ -19,6 +19,13 @@ type Telemetry interface {
 	ChipIngressEndpoint() string
 	ChipIngressInsecureConnection() bool
 	ChipIngressBatchEmitterEnabled() bool
+	ChipIngressBufferSize() uint
+	ChipIngressMaxBatchSize() uint
+	ChipIngressMaxConcurrentSends() int
+	ChipIngressSendInterval() time.Duration
+	ChipIngressSendTimeout() time.Duration
+	ChipIngressDrainTimeout() time.Duration
+	ChipIngressMaxGRPCRequestSize() int
 	DurableEmitterEnabled() bool
 	DurableEmitterRetransmitBatchSize() int
 	DurableEmitterEventTTL() time.Duration

core/config/toml/types.go

@@ -2983,6 +2983,13 @@
 	ChipIngressEndpoint                *string
 	ChipIngressInsecureConnection      *bool
 	ChipIngressBatchEmitterEnabled     *bool
+	ChipIngressBufferSize              *uint
+	ChipIngressMaxBatchSize            *uint
+	ChipIngressMaxConcurrentSends      *int
+	ChipIngressSendInterval            *commonconfig.Duration
+	ChipIngressSendTimeout             *commonconfig.Duration
+	ChipIngressDrainTimeout            *commonconfig.Duration
+	ChipIngressMaxGRPCRequestSize      *int
 	DurableEmitterEnabled              *bool
 	DurableEmitterRetransmitBatchSize  *int
 	DurableEmitterEventTTL             *commonconfig.Duration
@@ -3001,7 +3008,7 @@
 	PrometheusBridge PrometheusBridge `toml:",omitempty"`
 }
 
 func (b *Telemetry) setFrom(f *Telemetry) {
 	if v := f.Enabled; v != nil {
 		b.Enabled = v
 	}
@@ -3038,6 +3045,27 @@
 	if v := f.ChipIngressBatchEmitterEnabled; v != nil {
 		b.ChipIngressBatchEmitterEnabled = v
 	}
+	if v := f.ChipIngressBufferSize; v != nil {
+		b.ChipIngressBufferSize = v
+	}
+	if v := f.ChipIngressMaxBatchSize; v != nil {
+		b.ChipIngressMaxBatchSize = v
+	}
+	if v := f.ChipIngressMaxConcurrentSends; v != nil {
+		b.ChipIngressMaxConcurrentSends = v
+	}
+	if v := f.ChipIngressSendInterval; v != nil {
+		b.ChipIngressSendInterval = v
+	}
+	if v := f.ChipIngressSendTimeout; v != nil {
+		b.ChipIngressSendTimeout = v
+	}
+	if v := f.ChipIngressDrainTimeout; v != nil {
+		b.ChipIngressDrainTimeout = v
+	}
+	if v := f.ChipIngressMaxGRPCRequestSize; v != nil {
+		b.ChipIngressMaxGRPCRequestSize = v
+	}
 	if v := f.DurableEmitterEnabled; v != nil {
 		b.DurableEmitterEnabled = v
 	}
@@ -3096,6 +3124,27 @@
 	if ratio := b.TraceSampleRatio; ratio != nil && (*ratio < 0 || *ratio > 1) {
 		err = errors.Join(err, configutils.ErrInvalid{Name: "TraceSampleRatio", Value: *ratio, Msg: "must be between 0 and 1"})
 	}
+	if v := b.ChipIngressBufferSize; v != nil && *v == 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressBufferSize", Value: *v, Msg: "must be greater than 0"})
+	}
+	if v := b.ChipIngressMaxBatchSize; v != nil && *v == 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxBatchSize", Value: *v, Msg: "must be greater than 0"})
+	}
+	if v := b.ChipIngressMaxConcurrentSends; v != nil && *v <= 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxConcurrentSends", Value: *v, Msg: "must be greater than 0"})
+	}
+	if v := b.ChipIngressSendInterval; v != nil && v.Duration() <= 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressSendInterval", Value: v.Duration(), Msg: "must be greater than 0"})
+	}
+	if v := b.ChipIngressSendTimeout; v != nil && v.Duration() <= 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressSendTimeout", Value: v.Duration(), Msg: "must be greater than 0"})
+	}
+	if v := b.ChipIngressDrainTimeout; v != nil && v.Duration() <= 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressDrainTimeout", Value: v.Duration(), Msg: "must be greater than 0"})
+	}
+	if v := b.ChipIngressMaxGRPCRequestSize; v != nil && *v <= 0 {
+		err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxGRPCRequestSize", Value: *v, Msg: "must be greater than 0"})
+	}
 	return err
 }
 

core/scripts/cre/environment/examples/workflows/proof-of-reserve/cron-based/go.mod

@@ -4,7 +4,7 @@ go 1.26.4
 
 require (
 	github.com/ethereum/go-ethereum v1.17.1
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260625162847-bdf9e82b2f75
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260722120418-c1a1e0e75034
 	github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20251222115927-36a18321243c
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b
 	github.com/smartcontractkit/cre-sdk-go v1.5.0
@@ -74,7 +74,7 @@ require (
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260528204832-58c7145c53f8 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260716165322-7f2edff6e954 // indirect
 	github.com/smartcontractkit/libocr v0.0.0-20260403184524-b6409238958d // indirect
 	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/supranational/blst v0.3.16 // indirect

core/scripts/go.mod

@@ -47,7 +47,7 @@ require (
 	github.com/smartcontractkit/chain-selectors v1.0.106
 	github.com/smartcontractkit/chainlink-automation v0.8.1
 	github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260721154648-3e1c1fb5d8dc
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260722120418-c1a1e0e75034
 	github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6
 	github.com/smartcontractkit/chainlink-data-streams v1.0.0
 	github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
@@ -166,7 +166,7 @@ require (
 	github.com/buger/goterm v1.0.4 // indirect
 	github.com/buger/jsonparser v1.2.0 // indirect
 	github.com/buraksezer/consistent v0.10.0 // indirect
-	github.com/bytecodealliance/wasmtime-go/v28 v28.0.0 // indirect
+	github.com/bytecodealliance/wasmtime-go/v47 v47.0.0 // indirect
 	github.com/bytedance/sonic v1.15.0 // indirect
 	github.com/bytedance/sonic/loader v0.5.0 // indirect
 	github.com/cdk8s-team/cdk8s-core-go/cdk8s/v2 v2.70.2 // indirect

core/services/chainlink/config_telemetry.go

@@ -118,6 +118,55 @@ func (b *telemetryConfig) ChipIngressBatchEmitterEnabled() bool {
 	return *b.s.ChipIngressBatchEmitterEnabled
 }
 
+func (b *telemetryConfig) ChipIngressBufferSize() uint {
+	if b.s.ChipIngressBufferSize == nil {
+		return 0
+	}
+	return *b.s.ChipIngressBufferSize
+}
+
+func (b *telemetryConfig) ChipIngressMaxBatchSize() uint {
+	if b.s.ChipIngressMaxBatchSize == nil {
+		return 0
+	}
+	return *b.s.ChipIngressMaxBatchSize
+}
+
+func (b *telemetryConfig) ChipIngressMaxConcurrentSends() int {
+	if b.s.ChipIngressMaxConcurrentSends == nil {
+		return 0
+	}
+	return *b.s.ChipIngressMaxConcurrentSends
+}
+
+func (b *telemetryConfig) ChipIngressSendInterval() time.Duration {
+	if b.s.ChipIngressSendInterval == nil {
+		return 0
+	}
+	return b.s.ChipIngressSendInterval.Duration()
+}
+
+func (b *telemetryConfig) ChipIngressSendTimeout() time.Duration {
+	if b.s.ChipIngressSendTimeout == nil {
+		return 0
+	}
+	return b.s.ChipIngressSendTimeout.Duration()
+}
+
+func (b *telemetryConfig) ChipIngressDrainTimeout() time.Duration {
+	if b.s.ChipIngressDrainTimeout == nil {
+		return 0
+	}
+	return b.s.ChipIngressDrainTimeout.Duration()
+}
+
+func (b *telemetryConfig) ChipIngressMaxGRPCRequestSize() int {
+	if b.s.ChipIngressMaxGRPCRequestSize == nil {
+		return 0
+	}
+	return *b.s.ChipIngressMaxGRPCRequestSize
+}
+
 func (b *telemetryConfig) DurableEmitterEnabled() bool {
 	if b.s.DurableEmitterEnabled == nil {
 		return true