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

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

@@ -887,6 +887,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 = 1000 # Default
+# ChipIngressMaxBatchSize is the max events per PublishBatch RPC.
+ChipIngressMaxBatchSize = 500 # Default
+# ChipIngressMaxConcurrentSends limits parallel PublishBatch calls.
+ChipIngressMaxConcurrentSends = 10 # Default
+# ChipIngressSendInterval is the max wait before flushing an incomplete batch.
+ChipIngressSendInterval = '100ms' # Default
+# ChipIngressSendTimeout is the per-RPC timeout for PublishBatch.
+ChipIngressSendTimeout = '3s' # Default
+# ChipIngressDrainTimeout is the max shutdown wait to flush queued events.
+ChipIngressDrainTimeout = '10s' # 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 = false # Default
 # HeartbeatInterval is the interval at which a the application heartbeat is sent to telemetry backends.

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() uint
 	DurableEmitterEnabled() bool
 	HeartbeatInterval() time.Duration
 	LogStreamingEnabled() bool

core/config/toml/types.go

@@ -2968,6 +2968,13 @@
 	ChipIngressEndpoint            *string
 	ChipIngressInsecureConnection  *bool
 	ChipIngressBatchEmitterEnabled *bool
+	ChipIngressBufferSize          *uint
+	ChipIngressMaxBatchSize        *uint
+	ChipIngressMaxConcurrentSends  *int
+	ChipIngressSendInterval        *commonconfig.Duration
+	ChipIngressSendTimeout         *commonconfig.Duration
+	ChipIngressDrainTimeout        *commonconfig.Duration
+	ChipIngressMaxGRPCRequestSize  *uint
 	DurableEmitterEnabled          *bool
 	HeartbeatInterval              *commonconfig.Duration
 	LogLevel                       *string
@@ -3018,6 +3025,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
 	}
@@ -3064,6 +3092,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/services/chainlink/config_telemetry.go

@@ -107,6 +107,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() uint {
+	if b.s.ChipIngressMaxGRPCRequestSize == nil {
+		return 0
+	}
+	return *b.s.ChipIngressMaxGRPCRequestSize
+}
+
 func (b *telemetryConfig) DurableEmitterEnabled() bool {
 	if b.s.DurableEmitterEnabled == nil {
 		return false

core/services/chainlink/config_telemetry_test.go

@@ -245,6 +245,129 @@
 	return &f
 }
 
+func ptrUint(u uint) *uint {
+	return &u
+}
+
+func TestTelemetryConfig_ChipIngressBufferSize(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  uint
+	}{
+		{"ChipIngressBufferSizeSet", toml.Telemetry{ChipIngressBufferSize: ptrUint(1000)}, 1000},
+		{"ChipIngressBufferSizeNil", toml.Telemetry{ChipIngressBufferSize: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressBufferSize())
+		})
+	}
+}
+
+func TestTelemetryConfig_ChipIngressMaxBatchSize(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  uint
+	}{
+		{"ChipIngressMaxBatchSizeSet", toml.Telemetry{ChipIngressMaxBatchSize: ptrUint(500)}, 500},
+		{"ChipIngressMaxBatchSizeNil", toml.Telemetry{ChipIngressMaxBatchSize: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressMaxBatchSize())
+		})
+	}
+}
+
+func TestTelemetryConfig_ChipIngressMaxConcurrentSends(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  int
+	}{
+		{"ChipIngressMaxConcurrentSendsSet", toml.Telemetry{ChipIngressMaxConcurrentSends: ptrInt(10)}, 10},
+		{"ChipIngressMaxConcurrentSendsNil", toml.Telemetry{ChipIngressMaxConcurrentSends: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressMaxConcurrentSends())
+		})
+	}
+}
+
+func TestTelemetryConfig_ChipIngressSendInterval(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  time.Duration
+	}{
+		{"ChipIngressSendIntervalSet", toml.Telemetry{ChipIngressSendInterval: ptrDuration(100 * time.Millisecond)}, 100 * time.Millisecond},
+		{"ChipIngressSendIntervalNil", toml.Telemetry{ChipIngressSendInterval: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressSendInterval())
+		})
+	}
+}
+
+func TestTelemetryConfig_ChipIngressSendTimeout(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  time.Duration
+	}{
+		{"ChipIngressSendTimeoutSet", toml.Telemetry{ChipIngressSendTimeout: ptrDuration(3 * time.Second)}, 3 * time.Second},
+		{"ChipIngressSendTimeoutNil", toml.Telemetry{ChipIngressSendTimeout: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressSendTimeout())
+		})
+	}
+}
+
+func TestTelemetryConfig_ChipIngressDrainTimeout(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  time.Duration
+	}{
+		{"ChipIngressDrainTimeoutSet", toml.Telemetry{ChipIngressDrainTimeout: ptrDuration(10 * time.Second)}, 10 * time.Second},
+		{"ChipIngressDrainTimeoutNil", toml.Telemetry{ChipIngressDrainTimeout: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressDrainTimeout())
+		})
+	}
+}
+
+func TestTelemetryConfig_ChipIngressMaxGRPCRequestSize(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  uint
+	}{
+		{"ChipIngressMaxGRPCRequestSizeSet", toml.Telemetry{ChipIngressMaxGRPCRequestSize: ptrUint(10485760)}, 10485760},
+		{"ChipIngressMaxGRPCRequestSizeNil", toml.Telemetry{ChipIngressMaxGRPCRequestSize: nil}, 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.ChipIngressMaxGRPCRequestSize())
+		})
+	}
+}
+
 func TestTelemetryConfig_HeartbeatInterval(t *testing.T) {
 	tests := []struct {
 		name      string

core/services/chainlink/config_test.go

@@ -562,6 +562,13 @@
 		ChipIngressEndpoint:            ptr("example.com/chip-ingress"),
 		ChipIngressInsecureConnection:  ptr(false),
 		ChipIngressBatchEmitterEnabled: ptr(true),
+		ChipIngressBufferSize:          ptr[uint](1000),
+		ChipIngressMaxBatchSize:        ptr[uint](500),
+		ChipIngressMaxConcurrentSends:  ptr[int](10),
+		ChipIngressSendInterval:        commoncfg.MustNewDuration(100 * time.Millisecond),
+		ChipIngressSendTimeout:         commoncfg.MustNewDuration(3 * time.Second),
+		ChipIngressDrainTimeout:        commoncfg.MustNewDuration(10 * time.Second),
+		ChipIngressMaxGRPCRequestSize:  ptr[uint](10485760),
 		DurableEmitterEnabled:          ptr(false),
 		HeartbeatInterval:              commoncfg.MustNewDuration(1 * time.Second),
 		LogStreamingEnabled:            ptr(false),

core/services/chainlink/testdata/config-empty-effective.toml

@@ -349,6 +349,13 @@ AuthHeadersTTL = '0s'
 ChipIngressEndpoint = ''
 ChipIngressInsecureConnection = false
 ChipIngressBatchEmitterEnabled = true
+ChipIngressBufferSize = 1000
+ChipIngressMaxBatchSize = 500
+ChipIngressMaxConcurrentSends = 10
+ChipIngressSendInterval = '100ms'
+ChipIngressSendTimeout = '3s'
+ChipIngressDrainTimeout = '10s'
+ChipIngressMaxGRPCRequestSize = 10485760
 DurableEmitterEnabled = false
 HeartbeatInterval = '1s'
 LogLevel = 'info'

core/services/chainlink/testdata/config-full.toml

@@ -384,6 +384,13 @@ AuthHeadersTTL = '0s'
 ChipIngressEndpoint = 'example.com/chip-ingress'
 ChipIngressInsecureConnection = false
 ChipIngressBatchEmitterEnabled = true
+ChipIngressBufferSize = 1000
+ChipIngressMaxBatchSize = 500
+ChipIngressMaxConcurrentSends = 10
+ChipIngressSendInterval = '100ms'
+ChipIngressSendTimeout = '3s'
+ChipIngressDrainTimeout = '10s'
+ChipIngressMaxGRPCRequestSize = 10485760
 DurableEmitterEnabled = false
 HeartbeatInterval = '1s'
 LogLevel = 'info'

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -349,6 +349,13 @@ AuthHeadersTTL = '0s'
 ChipIngressEndpoint = ''
 ChipIngressInsecureConnection = false
 ChipIngressBatchEmitterEnabled = true
+ChipIngressBufferSize = 1000
+ChipIngressMaxBatchSize = 500
+ChipIngressMaxConcurrentSends = 10
+ChipIngressSendInterval = '100ms'
+ChipIngressSendTimeout = '3s'
+ChipIngressDrainTimeout = '10s'
+ChipIngressMaxGRPCRequestSize = 10485760
 DurableEmitterEnabled = false
 HeartbeatInterval = '1s'
 LogLevel = 'info'