diff --git a/core/cmd/shell.go b/core/cmd/shell.go index 4494588bdbd..8733a6f602e 100644 --- a/core/cmd/shell.go +++ b/core/cmd/shell.go @@ -155,6 +155,7 @@ func newBeholderClient( ChipIngressEmitterGRPCEndpoint: cfgTelemetry.ChipIngressEndpoint(), ChipIngressInsecureConnection: cfgTelemetry.ChipIngressInsecureConnection(), ChipIngressBatchEmitterEnabled: cfgTelemetry.ChipIngressBatchEmitterEnabled(), + ChipIngressMaxMessageBufferBytes: cfgTelemetry.ChipIngressMaxMessageBufferBytes(), ChipIngressLogger: lggr, LogStreamingEnabled: cfgTelemetry.LogStreamingEnabled(), LogLevel: cfgTelemetry.LogLevel(), diff --git a/core/config/docs/core.toml b/core/config/docs/core.toml index 8a285a216e6..7a037f0635b 100644 --- a/core/config/docs/core.toml +++ b/core/config/docs/core.toml @@ -889,6 +889,8 @@ ChipIngressInsecureConnection = false # Default # ChipIngressBatchEmitterEnabled enables batching for chip-ingress events. # When false, events are sent individually (legacy behavior). ChipIngressBatchEmitterEnabled = true # Default +# ChipIngressMaxMessageBufferBytes is the max live byte size of messages buffered in the chip-ingress batch client before drop (default 1 GiB in batch client when unset). +ChipIngressMaxMessageBufferBytes = 1073741824 # 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. diff --git a/core/config/telemetry_config.go b/core/config/telemetry_config.go index 1036eac775e..fca934aa6a2 100644 --- a/core/config/telemetry_config.go +++ b/core/config/telemetry_config.go @@ -19,6 +19,7 @@ type Telemetry interface { ChipIngressEndpoint() string ChipIngressInsecureConnection() bool ChipIngressBatchEmitterEnabled() bool + ChipIngressMaxMessageBufferBytes() uint DurableEmitterEnabled() bool HeartbeatInterval() time.Duration LogStreamingEnabled() bool diff --git a/core/config/toml/types.go b/core/config/toml/types.go index 6b3bb050f1e..dc2911b6942 100644 --- a/core/config/toml/types.go +++ b/core/config/toml/types.go @@ -2985,6 +2985,7 @@ type Telemetry struct { ChipIngressEndpoint *string ChipIngressInsecureConnection *bool ChipIngressBatchEmitterEnabled *bool + ChipIngressMaxMessageBufferBytes *uint DurableEmitterEnabled *bool HeartbeatInterval *commonconfig.Duration LogLevel *string @@ -3035,6 +3036,9 @@ func (b *Telemetry) setFrom(f *Telemetry) { if v := f.ChipIngressBatchEmitterEnabled; v != nil { b.ChipIngressBatchEmitterEnabled = v } + if v := f.ChipIngressMaxMessageBufferBytes; v != nil { + b.ChipIngressMaxMessageBufferBytes = v + } if v := f.DurableEmitterEnabled; v != nil { b.DurableEmitterEnabled = v } @@ -3081,6 +3085,9 @@ func (b *Telemetry) ValidateConfig() (err error) { 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.ChipIngressMaxMessageBufferBytes; v != nil && *v == 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxMessageBufferBytes", Value: *v, Msg: "must be greater than 0"}) + } return err } diff --git a/core/services/chainlink/application.go b/core/services/chainlink/application.go index 4fa893ba62f..30b1683c631 100644 --- a/core/services/chainlink/application.go +++ b/core/services/chainlink/application.go @@ -423,6 +423,9 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err MaxConcurrentSends: 8, MessageBufferSize: 50_000, } + if maxBuf := int(cfg.Telemetry().ChipIngressMaxMessageBufferBytes()); maxBuf > 0 { + durableCfg.MaxMessageBufferBytes = maxBuf + } pgStore := durableemitter.NewPgDurableEventStore(opts.DS) durableEmitter, setupErr := durableemitter.Setup(pgStore, durableCfg, globalLogger) if setupErr != nil { diff --git a/core/services/chainlink/config_telemetry.go b/core/services/chainlink/config_telemetry.go index df61e0d3269..2dc8c94737a 100644 --- a/core/services/chainlink/config_telemetry.go +++ b/core/services/chainlink/config_telemetry.go @@ -107,6 +107,13 @@ func (b *telemetryConfig) ChipIngressBatchEmitterEnabled() bool { return *b.s.ChipIngressBatchEmitterEnabled } +func (b *telemetryConfig) ChipIngressMaxMessageBufferBytes() uint { + if b.s.ChipIngressMaxMessageBufferBytes == nil { + return 0 + } + return *b.s.ChipIngressMaxMessageBufferBytes +} + func (b *telemetryConfig) DurableEmitterEnabled() bool { if b.s.DurableEmitterEnabled == nil { return false diff --git a/core/services/chainlink/config_telemetry_test.go b/core/services/chainlink/config_telemetry_test.go index 5baf19de43c..132c4ce15c2 100644 --- a/core/services/chainlink/config_telemetry_test.go +++ b/core/services/chainlink/config_telemetry_test.go @@ -237,6 +237,24 @@ func TestTelemetryConfig_ChipIngressBatchEmitterEnabled(t *testing.T) { } } +func TestTelemetryConfig_ChipIngressMaxMessageBufferBytes(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected uint + }{ + {"ChipIngressMaxMessageBufferBytesSet", toml.Telemetry{ChipIngressMaxMessageBufferBytes: ptr[uint](1073741824)}, 1073741824}, + {"ChipIngressMaxMessageBufferBytesNil", toml.Telemetry{ChipIngressMaxMessageBufferBytes: 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.ChipIngressMaxMessageBufferBytes()) + }) + } +} + func ptrDuration(d time.Duration) *config.Duration { return config.MustNewDuration(d) } diff --git a/docs/CONFIG.md b/docs/CONFIG.md index af944ec92e5..db69ddfd1d6 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -2393,6 +2393,7 @@ AuthHeadersTTL = '0s' # Default ChipIngressEndpoint = '' # Default ChipIngressInsecureConnection = false # Default ChipIngressBatchEmitterEnabled = true # Default +ChipIngressMaxMessageBufferBytes = 1073741824 # Default DurableEmitterEnabled = false # Default HeartbeatInterval = '1s' # Default LogLevel = "info" # Default @@ -2477,6 +2478,12 @@ ChipIngressBatchEmitterEnabled = true # Default ChipIngressBatchEmitterEnabled enables batching for chip-ingress events. When false, events are sent individually (legacy behavior). +### ChipIngressMaxMessageBufferBytes +```toml +ChipIngressMaxMessageBufferBytes = 1073741824 # Default +``` +ChipIngressMaxMessageBufferBytes is the max live byte size of messages buffered in the chip-ingress batch client before drop (default 1 GiB in batch client when unset). + ### DurableEmitterEnabled ```toml DurableEmitterEnabled = false # Default diff --git a/go.mod b/go.mod index 3cae40d9d3d..fa2401f6a94 100644 --- a/go.mod +++ b/go.mod @@ -437,3 +437,7 @@ tool github.com/smartcontractkit/chainlink-common/pkg/loop/cmd/loopinstall tool github.com/smartcontractkit/chainlink-common/script/cmd/dependabot replace github.com/smartcontractkit/chainlink-sui => github.com/smartcontractkit/chainlink-sui v0.0.0-20260707125635-abec997b6eae + +replace github.com/smartcontractkit/chainlink-common => ../../chainlink-common/chipingress-buffer-byte-cap + +replace github.com/smartcontractkit/chainlink-common/pkg/chipingress => ../../chainlink-common/chipingress-buffer-byte-cap/pkg/chipingress diff --git a/go.sum b/go.sum index 234ff314969..f8b8367c87b 100644 --- a/go.sum +++ b/go.sum @@ -1161,12 +1161,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260415165642-49f23e4d76cc/go.mod h1:67YbnoglYD61Pz/jTVCgav9wFq7S35OU8UyQSvPllRw= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h1:uMn1w/n95p+pSxK6hYNqji/sDaab8D0Cxuph9qUkM2g= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260709153841-bbd5dee84731 h1:+2I8Hwb441OVH1KzXMU5mLDT88HLDNYbn+6eOV/0XvU= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260709153841-bbd5dee84731/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= github.com/smartcontractkit/chainlink-common/keystore v1.2.0 h1:1BH/b14CkGjArfzznlioQpIJiynECWVT48JUP9E277U= github.com/smartcontractkit/chainlink-common/keystore v1.2.0/go.mod h1:9R/74vN+bJ5PbkOyM/pUy/AeAZaRwYb/k4XPeXcbDio= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a h1:8bIqv4r7SgDWkXL2Qz/Ijw+YjZY1uroIte3E2v2keVk= github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae h1:VrOaSNVhElURmca9ZVVyKlSiOg9Hz372+oFJD38fYRo= diff --git a/plugins/loop_registry.go b/plugins/loop_registry.go index 3968018cdbc..eec625e8360 100644 --- a/plugins/loop_registry.go +++ b/plugins/loop_registry.go @@ -155,6 +155,7 @@ func (m *LoopRegistry) Register(id string) (*RegisteredLoop, error) { envCfg.ChipIngressEndpoint = m.cfgTelemetry.ChipIngressEndpoint() envCfg.ChipIngressInsecureConnection = m.cfgTelemetry.ChipIngressInsecureConnection() envCfg.ChipIngressBatchEmitterEnabled = m.cfgTelemetry.ChipIngressBatchEmitterEnabled() + envCfg.ChipIngressMaxMessageBufferBytes = m.cfgTelemetry.ChipIngressMaxMessageBufferBytes() envCfg.ChipIngressDurableEmitterEnabled = m.cfgTelemetry.DurableEmitterEnabled() envCfg.TelemetryLogStreamingEnabled = m.cfgTelemetry.LogStreamingEnabled() envCfg.TelemetryLogLevel = m.cfgTelemetry.LogLevel()