beholder: log partial delivery per-event errors with error_code and reason

go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.100
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260713200205-3ec922174fe1
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b

pkg/beholder/batch_emitter_service.go

@@ -157,8 +157,26 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body
 			// for metric recording — OTel Add is non-blocking and tolerates
 			// cancelled contexts.
 			if sendErr != nil {
-				e.metrics.eventsDropped.Add(ctx, 1, metricAttrs)
-				e.eng.Errorw("failed to emit to chip ingress", "error", sendErr, "domain", domain, "entity", entity)
+				drop := batch.ClassifyDropFailure(sendErr)
+				e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, drop))
+				if drop.ErrorType == batch.ErrorTypePartialDelivery {
+					e.eng.Warnw("failed to emit to chip ingress (partial delivery)",
+						"error_type", drop.ErrorType,
+						"error_code", drop.ErrorCode,
+						"error_reason", drop.ErrorReason,
+						"domain", domain,
+						"entity", entity,
+					)
+				} else {
+					e.eng.Errorw("failed to emit to chip ingress",
+						"error", sendErr,
+						"error_type", drop.ErrorType,
+						"error_code", drop.ErrorCode,
+						"error_reason", drop.ErrorReason,
+						"domain", domain,
+						"entity", entity,
+					)
+				}
 			} else {
 				e.metrics.eventsSent.Add(ctx, 1, metricAttrs)
 			}
@@ -167,8 +185,16 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body
 			}
 		})
 		if queueErr != nil {
-			e.metrics.eventsDropped.Add(ctx, 1, metricAttrs)
-			e.eng.Errorw("failed to queue message for chip ingress", "error", queueErr, "domain", domain, "entity", entity)
+			drop := batch.ClassifyDropFailure(queueErr)
+			e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, drop))
+			e.eng.Errorw("failed to queue message for chip ingress",
+				"error", queueErr,
+				"error_type", drop.ErrorType,
+				"error_code", drop.ErrorCode,
+				"error_reason", drop.ErrorReason,
+				"domain", domain,
+				"entity", entity,
+			)
 			if callback != nil {
 				callback(queueErr)
 			}
@@ -191,6 +217,23 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o
 	return v.(otelmetric.MeasurementOption)
 }
 
+// dropMetricAttrsFor returns a measurement option for the eventsDropped counter.
+// Not cached — drop paths are not on the hot path.
+func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity string, drop batch.DropFailure) otelmetric.MeasurementOption {
+	attrs := []attribute.KeyValue{
+		attribute.String("domain", domain),
+		attribute.String("entity", entity),
+		attribute.String("error_type", drop.ErrorType),
+	}
+	if drop.ErrorCode != "" {
+		attrs = append(attrs, attribute.String("error_code", drop.ErrorCode))
+	}
+	if drop.ErrorReason != "" {
+		attrs = append(attrs, attribute.String("error_reason", drop.ErrorReason))
+	}
+	return otelmetric.WithAttributeSet(attribute.NewSet(attrs...))
+}
+
 func newBatchEmitterMetrics(meter otelmetric.Meter) (batchEmitterMetrics, error) {
 	eventsSent, err := meter.Int64Counter("chip_ingress.events_sent",
 		otelmetric.WithDescription("Total events successfully sent via PublishBatch"),

pkg/beholder/batch_emitter_service_test.go

@@ -16,6 +16,8 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest/observer"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
@@ -458,6 +460,169 @@ func TestChipIngressBatchEmitterService_EmitWithCallback(t *testing.T) {
 	})
 }
 
+func TestChipIngressBatchEmitterService_PartialDeliveryError(t *testing.T) {
+	t.Run("logs error_code and reason on per-event PublishError", func(t *testing.T) {
+		lggr, observed := logger.TestObserved(t, zap.InfoLevel)
+
+		clientMock := mocks.NewClient(t)
+		clientMock.EXPECT().Close().Return(nil).Maybe()
+
+		partialResp := &chipingress.PublishResponse{
+			Results: []*chipingress.PublishResult{
+				{
+					Error: &chipingress.PublishError{
+						ErrorCode: chipingress.PublishErrorCode(1), // VALIDATION_FAILED
+						Reason:    "schema not found",
+					},
+				},
+			},
+		}
+		done := make(chan struct{})
+		clientMock.
+			On("PublishBatch", mock.Anything, mock.Anything).
+			Return(partialResp, nil).
+			Run(func(_ mock.Arguments) { close(done) }).
+			Once()
+
+		cfg := newTestConfig()
+		cfg.ChipIngressMaxBatchSize = 1
+		cfg.ChipIngressSendInterval = time.Second
+
+		emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, lggr)
+		require.NoError(t, err)
+		require.NoError(t, emitter.Start(t.Context()))
+
+		err = emitter.Emit(t.Context(), []byte("body"),
+			beholder.AttrKeyDomain, "platform",
+			beholder.AttrKeyEntity, "TestEvent",
+		)
+		require.NoError(t, err)
+
+		select {
+		case <-done:
+		case <-time.After(3 * time.Second):
+			t.Fatal("timeout waiting for publish")
+		}
+		require.NoError(t, emitter.Close())
+
+		logs := observed.FilterMessage("failed to emit to chip ingress (partial delivery)")
+		require.GreaterOrEqual(t, logs.Len(), 1, "expected partial delivery error log")
+		entry := logs.All()[0]
+		assert.Equal(t, zap.WarnLevel, entry.Level)
+		fieldMap := logFieldMap(entry)
+		assert.Equal(t, "platform", fieldMap["domain"])
+		assert.Equal(t, "TestEvent", fieldMap["entity"])
+		assert.Equal(t, "partial_delivery", fieldMap["error_type"])
+		assert.Contains(t, fieldMap["error_code"], "PUBLISH_ERROR_CODE")
+		assert.Equal(t, "schema not found", fieldMap["error_reason"])
+	})
+
+	t.Run("records events_dropped with partial_delivery error_type", func(t *testing.T) {
+		reader, restore := useEmitterTestMeterProvider(t)
+		defer restore()
+
+		clientMock := mocks.NewClient(t)
+		clientMock.EXPECT().Close().Return(nil).Maybe()
+
+		partialResp := &chipingress.PublishResponse{
+			Results: []*chipingress.PublishResult{
+				{
+					Error: &chipingress.PublishError{
+						ErrorCode: chipingress.PublishErrorCode(1),
+						Reason:    "encode error",
+					},
+				},
+			},
+		}
+		done := make(chan struct{})
+		clientMock.
+			On("PublishBatch", mock.Anything, mock.Anything).
+			Return(partialResp, nil).
+			Run(func(_ mock.Arguments) { close(done) }).
+			Once()
+
+		cfg := newTestConfig()
+		cfg.ChipIngressMaxBatchSize = 1
+		cfg.ChipIngressSendInterval = time.Second
+
+		emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, newTestLogger(t))
+		require.NoError(t, err)
+		require.NoError(t, emitter.Start(t.Context()))
+
+		err = emitter.Emit(t.Context(), []byte("body"),
+			beholder.AttrKeyDomain, "platform",
+			beholder.AttrKeyEntity, "PartialEvent",
+		)
+		require.NoError(t, err)
+
+		select {
+		case <-done:
+		case <-time.After(3 * time.Second):
+			t.Fatal("timeout waiting for publish")
+		}
+		require.NoError(t, emitter.Close())
+
+		rm := collectEmitterMetrics(t, reader)
+		metric := mustEmitterMetric(t, rm, "chip_ingress.events_dropped")
+		sum, ok := metric.Data.(metricdata.Sum[int64])
+		require.True(t, ok)
+		dp := mustEmitterInt64SumPoint(t, sum, "error_type", "partial_delivery", "entity", "PartialEvent")
+		assert.GreaterOrEqual(t, dp.Value, int64(1))
+		assert.True(t,
+			hasEmitterStringAttr(dp.Attributes, "error_code", "PUBLISH_ERROR_CODE_VALIDATION_FAILED"),
+			"expected error_code label on partial_delivery drop metric datapoint",
+		)
+	})
+}
+
+func TestChipIngressBatchEmitterService_RPCError(t *testing.T) {
+	t.Run("records events_dropped with error_code on RPC failure", func(t *testing.T) {
+		reader, restore := useEmitterTestMeterProvider(t)
+		defer restore()
+
+		clientMock := mocks.NewClient(t)
+		clientMock.EXPECT().Close().Return(nil).Maybe()
+
+		done := make(chan struct{})
+		clientMock.
+			On("PublishBatch", mock.Anything, mock.Anything).
+			Return(nil, status.Error(codes.Internal, "failed to publish events")).
+			Run(func(_ mock.Arguments) { close(done) }).
+			Once()
+
+		cfg := newTestConfig()
+		cfg.ChipIngressMaxBatchSize = 1
+		cfg.ChipIngressSendInterval = time.Second
+
+		emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, newTestLogger(t))
+		require.NoError(t, err)
+		require.NoError(t, emitter.Start(t.Context()))
+
+		err = emitter.Emit(t.Context(), []byte("body"),
+			beholder.AttrKeyDomain, "platform",
+			beholder.AttrKeyEntity, "RPCDropEvent",
+		)
+		require.NoError(t, err)
+
+		select {
+		case <-done:
+		case <-time.After(3 * time.Second):
+			t.Fatal("timeout waiting for publish")
+		}
+		require.NoError(t, emitter.Close())
+
+		rm := collectEmitterMetrics(t, reader)
+		metric := mustEmitterMetric(t, rm, "chip_ingress.events_dropped")
+		sum, ok := metric.Data.(metricdata.Sum[int64])
+		require.True(t, ok)
+		dp := mustEmitterInt64SumPoint(t, sum, "error_type", "rpc_error", "entity", "RPCDropEvent")
+		assert.GreaterOrEqual(t, dp.Value, int64(1))
+		assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "Internal"))
+		assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_reason", "failed to publish events"))
+	})
+}
+
+
 func TestChipIngressBatchEmitterService_Metrics(t *testing.T) {
 	t.Run("records events_sent on successful publish", func(t *testing.T) {
 		reader, restore := useEmitterTestMeterProvider(t)
@@ -511,7 +676,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) {
 		done := make(chan struct{})
 		clientMock.
 			On("PublishBatch", mock.Anything, mock.Anything).
-			Return(nil, assert.AnError).
+			Return(nil, status.Error(codes.DeadlineExceeded, "context deadline exceeded")).
 			Run(func(_ mock.Arguments) { close(done) }).
 			Once()
 
@@ -539,15 +704,17 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) {
 		metric := mustEmitterMetric(t, rm, "chip_ingress.events_dropped")
 		sum, ok := metric.Data.(metricdata.Sum[int64])
 		require.True(t, ok)
-		dp := mustEmitterInt64SumPoint(t, sum, "domain", "platform", "entity", "MetricDropEvent")
+		dp := mustEmitterInt64SumPoint(t, sum, "error_type", "rpc_error", "entity", "MetricDropEvent")
 		assert.GreaterOrEqual(t, dp.Value, int64(1))
+		assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "DeadlineExceeded"))
 
 		logs := observed.FilterMessage("failed to emit to chip ingress")
 		require.GreaterOrEqual(t, logs.Len(), 1, "expected error log for publish failure")
 		entry := logs.All()[0]
 		assert.Equal(t, zap.ErrorLevel, entry.Level)
 		fieldMap := logFieldMap(entry)
 		assert.Contains(t, fieldMap, "error")
+		assert.Equal(t, "DeadlineExceeded", fieldMap["error_code"])
 		assert.Equal(t, "platform", fieldMap["domain"])
 		assert.Equal(t, "MetricDropEvent", fieldMap["entity"])
 	})

pkg/chipingress/batch/client.go

@@ -19,6 +19,11 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 )
 
+var (
+	ErrMessageBufferFull = errors.New("message buffer is full")
+	ErrClientShutdown    = errors.New("client is shutdown")
+)
+
 type messageWithCallback struct {
 	event    *chipingress.CloudEventPb
 	callback func(error)
@@ -240,7 +245,7 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err
 	// Check shutdown first to avoid race with buffer send
 	select {
 	case <-b.stopCh:
-		return errors.New("client is shutdown")
+		return ErrClientShutdown
 	default:
 	}
 
@@ -274,7 +279,7 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err
 	case b.messageBuffer <- msg:
 		return nil
 	default:
-		return errors.New("message buffer is full")
+		return ErrMessageBufferFull
 	}
 }