Cap chip-ingress batch buffer at 1 GiB live bytes

go.mod

@@ -163,3 +163,5 @@ require (
 retract v1.3.0 // accidentally published
 
 retract v1.3.1 // exists only to retract v1.3.0
+
+replace github.com/smartcontractkit/chainlink-common/pkg/chipingress => ./pkg/chipingress

pkg/beholder/batch_emitter_service.go

@@ -2,6 +2,7 @@ package beholder
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -63,6 +64,10 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 	if drainTimeout == 0 {
 		drainTimeout = defaults.ChipIngressDrainTimeout
 	}
+	maxMessageBufferBytes := int(cfg.ChipIngressMaxMessageBufferBytes)
+	if maxMessageBufferBytes == 0 {
+		maxMessageBufferBytes = int(defaults.ChipIngressMaxMessageBufferBytes)
+	}
 
 	meter := otel.Meter("beholder/chip_ingress_batch_emitter")
 	metrics, err := newBatchEmitterMetrics(meter)
@@ -73,6 +78,7 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 	batchClient, err := batch.NewBatchClient(client,
 		batch.WithBatchSize(maxBatchSize),
 		batch.WithMessageBuffer(bufferSize),
+		batch.WithMaxMessageBufferBytes(maxMessageBufferBytes),
 		batch.WithBatchInterval(sendInterval),
 		batch.WithMaxPublishTimeout(sendTimeout),
 		batch.WithShutdownTimeout(drainTimeout),
@@ -167,7 +173,7 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body
 			}
 		})
 		if queueErr != nil {
-			e.metrics.eventsDropped.Add(ctx, 1, metricAttrs)
+			e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrs(domain, entity, queueFailureType(queueErr)))
 			e.eng.Errorw("failed to queue message for chip ingress", "error", queueErr, "domain", domain, "entity", entity)
 			if callback != nil {
 				callback(queueErr)
@@ -178,6 +184,31 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body
 	})
 }
 
+func queueFailureType(err error) string {
+	switch {
+	case errors.Is(err, batch.ErrMessageBufferBytesExceeded):
+		return "buffer_bytes_exceeded"
+	case errors.Is(err, batch.ErrMessageBufferFull):
+		return "buffer_full"
+	default:
+		return "queue_error"
+	}
+}
+
+func (e *ChipIngressBatchEmitterService) dropMetricAttrs(domain, entity, failureType string) otelmetric.MeasurementOption {
+	key := domain + "\x00" + entity + "\x00" + failureType
+	if v, ok := e.metricAttrsCache.Load(key); ok {
+		return v.(otelmetric.MeasurementOption)
+	}
+	attrs := otelmetric.WithAttributeSet(attribute.NewSet(
+		attribute.String("domain", domain),
+		attribute.String("entity", entity),
+		attribute.String("failure_type", failureType),
+	))
+	v, _ := e.metricAttrsCache.LoadOrStore(key, attrs)
+	return v.(otelmetric.MeasurementOption)
+}
+
 func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) otelmetric.MeasurementOption {
 	key := domain + "\x00" + entity
 	if v, ok := e.metricAttrsCache.Load(key); ok {

pkg/beholder/config.go

@@ -8,6 +8,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.uber.org/zap/zapcore"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 )
 
@@ -56,8 +57,9 @@ type Config struct {
 	ChipIngressSendInterval        time.Duration // Flush interval (default 100ms)
 	ChipIngressSendTimeout         time.Duration // Timeout per PublishBatch call (default 3s)
 	ChipIngressDrainTimeout        time.Duration // Max time to flush remaining events on shutdown (default 10s)
-	ChipIngressMaxConcurrentSends  int           // Max concurrent PublishBatch calls (default 10)
-	ChipIngressLogger              logger.Logger // Required when ChipIngressBatchEmitterEnabled is true
+	ChipIngressMaxConcurrentSends     int           // Max concurrent PublishBatch calls (default 10)
+	ChipIngressMaxMessageBufferBytes  uint          // Max live byte size of buffered messages (default 1 GiB)
+	ChipIngressLogger                 logger.Logger // Required when ChipIngressBatchEmitterEnabled is true
 
 	// OTel Log
 	LogExportTimeout      time.Duration
@@ -157,7 +159,8 @@ func DefaultConfig() Config {
 		ChipIngressSendInterval:        100 * time.Millisecond,
 		ChipIngressSendTimeout:         3 * time.Second,
 		ChipIngressDrainTimeout:        10 * time.Second,
-		ChipIngressMaxConcurrentSends:  defaultMaxConcurrentSends,
+		ChipIngressMaxConcurrentSends:    defaultMaxConcurrentSends,
+		ChipIngressMaxMessageBufferBytes: uint(batch.DefaultMaxMessageBufferBytes),
 		// Auth (defaults to static auth mode with TTL=0)
 		AuthHeadersTTL: 0,
 	}

pkg/beholder/testdata/config-example.json

@@ -44,6 +44,7 @@
   "ChipIngressSendTimeout": 0,
   "ChipIngressDrainTimeout": 0,
   "ChipIngressMaxConcurrentSends": 0,
+  "ChipIngressMaxMessageBufferBytes": 0,
   "ChipIngressLogger": null,
   "LogExportTimeout": 1000000000,
   "LogExportInterval": 1000000000,

pkg/chipingress/batch/batcher.go

@@ -7,13 +7,19 @@ import (
 
 // batchWithInterval reads from a channel and calls processFn with batches based on size or time interval.
 // When the context is cancelled, it automatically flushes any remaining items in the batch.
+// onDequeue, when non-nil, is invoked immediately after each message is read from input.
 func batchWithInterval[T any](
 	ctx context.Context,
 	input <-chan T,
 	batchSize int,
 	interval time.Duration,
 	processFn func([]T),
+	onDequeue ...func(T),
 ) {
+	var dequeue func(T)
+	if len(onDequeue) > 0 {
+		dequeue = onDequeue[0]
+	}
 	var batch []T
 	timer := time.NewTimer(interval)
 	timer.Stop()
@@ -38,6 +44,10 @@ func batchWithInterval[T any](
 				return
 			}
 
+			if dequeue != nil {
+				dequeue(msg)
+			}
+
 			// Start timer on first message in batch
 			if len(batch) == 0 {
 				timer.Reset(interval)

pkg/chipingress/batch/client.go

@@ -19,9 +19,23 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 )
 
+// DefaultMaxMessageBufferBytes is the default live-byte cap for the message buffer channel.
+const DefaultMaxMessageBufferBytes = 1 << 30 // 1 GiB
+
+// defaultMaxMessageBufferBytes is an alias kept for internal use.
+const defaultMaxMessageBufferBytes = DefaultMaxMessageBufferBytes
+
+var (
+	// ErrMessageBufferFull is returned when the message-count buffer is full.
+	ErrMessageBufferFull = errors.New("message buffer is full")
+	// ErrMessageBufferBytesExceeded is returned when enqueueing would exceed the byte cap.
+	ErrMessageBufferBytesExceeded = errors.New("message buffer bytes exceeded")
+)
+
 type messageWithCallback struct {
 	event    *chipingress.CloudEventPb
 	callback func(error)
+	byteSize int
 }
 
 type seqnumKey struct {
@@ -62,6 +76,9 @@ type Client struct {
 	batchInterval           time.Duration
 	maxPublishTimeout       time.Duration
 	messageBuffer           chan *messageWithCallback
+	maxMessageBufferBytes   int
+	bufferedBytes           atomic.Int64
+	bufferMu                sync.Mutex
 	stopCh                  stopCh
 	log                     *zap.SugaredLogger
 	callbackWg              sync.WaitGroup
@@ -82,6 +99,7 @@ type batchClientMetrics struct {
 	requestSizeBytes     otelmetric.Int64Histogram
 	requestLatencyMS     otelmetric.Float64Histogram
 	configInfo           otelmetric.Int64Gauge
+	bufferLiveBytes      otelmetric.Int64ObservableGauge
 	batchSplitsTotal     otelmetric.Int64Counter
 	resultsMismatchTotal otelmetric.Int64Counter
 	batchSizeAttr        otelmetric.MeasurementOption
@@ -105,6 +123,7 @@ func NewBatchClient(client chipingress.Client, opts ...Opt) (*Client, error) {
 		transactionEnabled:      false,
 		maxConcurrentSends:      make(chan struct{}, 1),
 		messageBuffer:           make(chan *messageWithCallback, 200),
+		maxMessageBufferBytes:   defaultMaxMessageBufferBytes,
 		batchInterval:           100 * time.Millisecond,
 		maxPublishTimeout:       5 * time.Second,
 		stopCh:                  make(chan struct{}),
@@ -132,6 +151,7 @@ func NewBatchClient(client chipingress.Client, opts ...Opt) (*Client, error) {
 // context that is cancelled when Stop is called.
 func (b *Client) Start(ctx context.Context) {
 	b.metrics.recordConfig(ctx, b)
+	b.metrics.registerBufferLiveBytesCallback(b)
 	b.started = true
 
 	// Detach from the caller's cancellation but keep its values (trace IDs, etc.).
@@ -158,6 +178,11 @@ func (b *Client) Start(ctx context.Context) {
 				// sendBatch still enforces maxPublishTimeout for each publish call.
 				b.sendBatch(context.WithoutCancel(batcherCtx), batch)
 			},
+			func(msg *messageWithCallback) {
+				if msg.byteSize > 0 {
+					b.bufferedBytes.Add(-int64(msg.byteSize))
+				}
+			},
 		)
 	}()
 }
@@ -265,16 +290,30 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err
 		},
 	}
 
+	msgSize := proto.Size(eventToQueue)
 	msg := &messageWithCallback{
 		event:    eventToQueue,
 		callback: callback,
+		byteSize: msgSize,
+	}
+
+	b.bufferMu.Lock()
+	if b.maxMessageBufferBytes > 0 {
+		nextBytes := b.bufferedBytes.Load() + int64(msgSize)
+		if nextBytes > int64(b.maxMessageBufferBytes) {
+			b.bufferMu.Unlock()
+			return ErrMessageBufferBytesExceeded
+		}
 	}
 
 	select {
 	case b.messageBuffer <- msg:
+		b.bufferedBytes.Add(int64(msgSize))
+		b.bufferMu.Unlock()
 		return nil
 	default:
-		return errors.New("message buffer is full")
+		b.bufferMu.Unlock()
+		return ErrMessageBufferFull
 	}
 }
 
@@ -497,6 +536,14 @@ func WithMessageBuffer(messageBufferSize int) Opt {
 	}
 }
 
+// WithMaxMessageBufferBytes sets the maximum live byte size of messages buffered in the
+// message channel. Zero disables the byte cap (intended for tests).
+func WithMaxMessageBufferBytes(maxBytes int) Opt {
+	return func(c *Client) {
+		c.maxMessageBufferBytes = maxBytes
+	}
+}
+
 // WithMaxPublishTimeout sets the maximum time to wait for a batch publish operation
 func WithMaxPublishTimeout(maxPublishTimeout time.Duration) Opt {
 	return func(c *Client) {
@@ -573,6 +620,14 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 	if err != nil {
 		return batchClientMetrics{}, err
 	}
+	bufferLiveBytes, err := meter.Int64ObservableGauge(
+		"chip_ingress.batch.buffer_live_bytes",
+		otelmetric.WithDescription("Live byte size of messages currently buffered in the batch client message channel"),
+		otelmetric.WithUnit("By"),
+	)
+	if err != nil {
+		return batchClientMetrics{}, err
+	}
 	batchSplitsTotal, err := meter.Int64Counter(
 		"chip_ingress.batch.batch_splits_total",
 		otelmetric.WithDescription("Total number of times a batch was split due to exceeding the effective gRPC request size limit (max request size minus reserved framing overhead)"),
@@ -596,6 +651,7 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 		requestSizeBytes:     requestSizeBytes,
 		requestLatencyMS:     requestLatencyMS,
 		configInfo:           configInfo,
+		bufferLiveBytes:      bufferLiveBytes,
 		batchSplitsTotal:     batchSplitsTotal,
 		resultsMismatchTotal: resultsMismatchTotal,
 		successStatusAttr: otelmetric.WithAttributeSet(attribute.NewSet(
@@ -607,6 +663,19 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 	}, nil
 }
 
+func (m *batchClientMetrics) registerBufferLiveBytesCallback(c *Client) {
+	if m.bufferLiveBytes == nil {
+		return
+	}
+	_, _ = otel.Meter("chipingress/batch_client").RegisterCallback(
+		func(_ context.Context, o otelmetric.Observer) error {
+			o.ObserveInt64(m.bufferLiveBytes, c.bufferedBytes.Load())
+			return nil
+		},
+		m.bufferLiveBytes,
+	)
+}
+
 func (m *batchClientMetrics) recordConfig(ctx context.Context, c *Client) {
 	m.batchSizeAttr = otelmetric.WithAttributeSet(attribute.NewSet(
 		attribute.Int("max_batch_size", c.batchSize),
@@ -624,6 +693,7 @@ func (m *batchClientMetrics) recordConfig(ctx context.Context, c *Client) {
 		attribute.Bool("clone_event", c.cloneEvent),
 		attribute.Bool("transaction_enabled", c.transactionEnabled),
 		attribute.Int("max_grpc_request_size_bytes", c.maxGRPCRequestSize),
+		attribute.Int("max_message_buffer_bytes", c.maxMessageBufferBytes),
 	))
 }