From d16297cd57de68b4efee2d5de8b253980b1ff253 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Thu, 2 Jul 2026 18:17:45 -0400 Subject: [PATCH 1/3] beholder: log and meter partial delivery errors per-event with error_code label - Type-assert send callback error to *batch.PublishError; log at WARN with error_code, reason, domain, entity for per-event server rejections. - Add error_code label to chip_ingress.events_dropped metric so dashboards and alerts can distinguish rejection types without relying on logs. - Whole-batch RPC failures keep the existing ERROR log; their drop metric omits error_code (no structured server code available). - Also adds failure_type label (partial_delivery / rpc_error) to the dropped counter to distinguish the two failure modes. Jira: INFOPLAT-13901 --- pkg/beholder/batch_emitter_service.go | 35 ++++++- pkg/beholder/batch_emitter_service_test.go | 114 +++++++++++++++++++++ 2 files changed, 147 insertions(+), 2 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index a81fd82bc9..9831af4be6 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -2,6 +2,7 @@ package beholder import ( "context" + "errors" "fmt" "sync" @@ -157,8 +158,22 @@ 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) + var pubErr *batch.PublishError + if errors.As(sendErr, &pubErr) { + // Per-event partial delivery failure: server accepted the batch but + // rejected this individual event (e.g. schema validation, encode error). + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, "partial_delivery", pubErr.Code.String())) + e.eng.Warnw("failed to emit to chip ingress (partial delivery)", + "error_code", pubErr.Code.String(), + "reason", pubErr.Reason, + "domain", domain, + "entity", entity, + ) + } else { + // Whole-batch RPC failure (network error, timeout, auth, etc.). + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, "rpc_error", "")) + e.eng.Errorw("failed to emit to chip ingress", "error", sendErr, "domain", domain, "entity", entity) + } } else { e.metrics.eventsSent.Add(ctx, 1, metricAttrs) } @@ -191,6 +206,22 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o return v.(otelmetric.MeasurementOption) } +// dropMetricAttrsFor returns a measurement option for the eventsDropped counter. +// errorCode is the structured server error code for partial_delivery failures; +// pass "" for rpc_error (where no structured code is available). +// Not cached — drop paths are not on the hot path. +func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, failureType, errorCode string) otelmetric.MeasurementOption { + attrs := []attribute.KeyValue{ + attribute.String("domain", domain), + attribute.String("entity", entity), + attribute.String("failure_type", failureType), + } + if errorCode != "" { + attrs = append(attrs, attribute.String("error_code", errorCode)) + } + 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"), diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index 4137d450d5..cff2791b96 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/pkg/beholder/batch_emitter_service_test.go @@ -458,6 +458,120 @@ 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.Contains(t, fieldMap, "error_code") + assert.Equal(t, "schema not found", fieldMap["reason"]) + }) + + t.Run("records events_dropped with partial_delivery failure_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, "failure_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_Metrics(t *testing.T) { t.Run("records events_sent on successful publish", func(t *testing.T) { reader, restore := useEmitterTestMeterProvider(t) From 3ec922174fe164a8d23898998b45062d8afb3efe Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:02:05 -0400 Subject: [PATCH 2/3] beholder: unify drop metrics with error_type, error_code, error_reason - Add batch.ClassifyDropFailure for partial_delivery, rpc_error, buffer_full, and client_error paths with shared metric dimensions. - Replace failure_type/grpc_code labels with error_type and error_code (PUBLISH_ERROR_CODE_* or gRPC code name) plus error_reason. - Add ErrMessageBufferFull and ErrClientShutdown sentinels for queue drops. Jira: INFOPLAT-13901 --- pkg/beholder/batch_emitter_service.go | 50 ++++++----- pkg/beholder/batch_emitter_service_test.go | 65 ++++++++++++-- pkg/chipingress/batch/client.go | 9 +- pkg/chipingress/batch/drop_failure.go | 79 +++++++++++++++++ pkg/chipingress/batch/drop_failure_test.go | 98 ++++++++++++++++++++++ 5 files changed, 274 insertions(+), 27 deletions(-) create mode 100644 pkg/chipingress/batch/drop_failure.go create mode 100644 pkg/chipingress/batch/drop_failure_test.go diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 9831af4be6..c26d5ad148 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -2,7 +2,6 @@ package beholder import ( "context" - "errors" "fmt" "sync" @@ -158,21 +157,25 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body // for metric recording — OTel Add is non-blocking and tolerates // cancelled contexts. if sendErr != nil { - var pubErr *batch.PublishError - if errors.As(sendErr, &pubErr) { - // Per-event partial delivery failure: server accepted the batch but - // rejected this individual event (e.g. schema validation, encode error). - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, "partial_delivery", pubErr.Code.String())) + 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_code", pubErr.Code.String(), - "reason", pubErr.Reason, + "error_type", drop.ErrorType, + "error_code", drop.ErrorCode, + "error_reason", drop.ErrorReason, "domain", domain, "entity", entity, ) } else { - // Whole-batch RPC failure (network error, timeout, auth, etc.). - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, "rpc_error", "")) - e.eng.Errorw("failed to emit to chip ingress", "error", sendErr, "domain", domain, "entity", entity) + 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) @@ -182,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) } @@ -207,17 +218,18 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o } // dropMetricAttrsFor returns a measurement option for the eventsDropped counter. -// errorCode is the structured server error code for partial_delivery failures; -// pass "" for rpc_error (where no structured code is available). // Not cached — drop paths are not on the hot path. -func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, failureType, errorCode string) otelmetric.MeasurementOption { +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("failure_type", failureType), + attribute.String("error_type", drop.ErrorType), + } + if drop.ErrorCode != "" { + attrs = append(attrs, attribute.String("error_code", drop.ErrorCode)) } - if errorCode != "" { - attrs = append(attrs, attribute.String("error_code", errorCode)) + if drop.ErrorReason != "" { + attrs = append(attrs, attribute.String("error_reason", drop.ErrorReason)) } return otelmetric.WithAttributeSet(attribute.NewSet(attrs...)) } diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index cff2791b96..5ba1817268 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/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" @@ -510,11 +512,12 @@ func TestChipIngressBatchEmitterService_PartialDeliveryError(t *testing.T) { fieldMap := logFieldMap(entry) assert.Equal(t, "platform", fieldMap["domain"]) assert.Equal(t, "TestEvent", fieldMap["entity"]) - assert.Contains(t, fieldMap, "error_code") - assert.Equal(t, "schema not found", fieldMap["reason"]) + 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 failure_type", func(t *testing.T) { + t.Run("records events_dropped with partial_delivery error_type", func(t *testing.T) { reader, restore := useEmitterTestMeterProvider(t) defer restore() @@ -563,7 +566,7 @@ func TestChipIngressBatchEmitterService_PartialDeliveryError(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, "failure_type", "partial_delivery", "entity", "PartialEvent") + 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"), @@ -572,6 +575,54 @@ func TestChipIngressBatchEmitterService_PartialDeliveryError(t *testing.T) { }) } +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) @@ -625,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() @@ -653,8 +704,9 @@ 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") @@ -662,6 +714,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { 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"]) }) diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 90c3bca980..22c249bf9e 100644 --- a/pkg/chipingress/batch/client.go +++ b/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 } } diff --git a/pkg/chipingress/batch/drop_failure.go b/pkg/chipingress/batch/drop_failure.go new file mode 100644 index 0000000000..07586ac94d --- /dev/null +++ b/pkg/chipingress/batch/drop_failure.go @@ -0,0 +1,79 @@ +package batch + +import ( + "context" + "errors" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + ErrorTypePartialDelivery = "partial_delivery" + ErrorTypeRPCError = "rpc_error" + ErrorTypeBufferFull = "buffer_full" + ErrorTypeClientError = "client_error" +) + +// DropFailure describes a dropped event for metrics and logging. +type DropFailure struct { + ErrorType string // partial_delivery | rpc_error | buffer_full | client_error + ErrorCode string // PUBLISH_ERROR_CODE_* or gRPC code name (DeadlineExceeded, …) + ErrorReason string // server reason, status message, or client error text +} + +// ClassifyDropFailure maps batch send/queue errors to metric dimensions. +func ClassifyDropFailure(err error) DropFailure { + if err == nil { + return DropFailure{} + } + + var pubErr *PublishError + if errors.As(err, &pubErr) { + return DropFailure{ + ErrorType: ErrorTypePartialDelivery, + ErrorCode: pubErr.Code.String(), + ErrorReason: pubErr.Reason, + } + } + + if errors.Is(err, ErrMessageBufferFull) { + return DropFailure{ + ErrorType: ErrorTypeBufferFull, + ErrorReason: ErrMessageBufferFull.Error(), + } + } + if errors.Is(err, ErrClientShutdown) { + return DropFailure{ + ErrorType: ErrorTypeClientError, + ErrorReason: ErrClientShutdown.Error(), + } + } + if errors.Is(err, context.DeadlineExceeded) { + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: codes.DeadlineExceeded.String(), + ErrorReason: context.DeadlineExceeded.Error(), + } + } + if errors.Is(err, context.Canceled) { + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: codes.Canceled.String(), + ErrorReason: context.Canceled.Error(), + } + } + if st, ok := status.FromError(err); ok { + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: st.Code().String(), + ErrorReason: st.Message(), + } + } + + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: codes.Unknown.String(), + ErrorReason: err.Error(), + } +} diff --git a/pkg/chipingress/batch/drop_failure_test.go b/pkg/chipingress/batch/drop_failure_test.go new file mode 100644 index 0000000000..70dfb75ad9 --- /dev/null +++ b/pkg/chipingress/batch/drop_failure_test.go @@ -0,0 +1,98 @@ +package batch_test + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" +) + +func TestClassifyDropFailure(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + wantType string + wantCode string + wantReason string + }{ + { + name: "partial delivery publish error", + err: &batch.PublishError{Code: chipingress.PublishErrorCode(1), Reason: "schema not found"}, + wantType: batch.ErrorTypePartialDelivery, + wantCode: chipingress.PublishErrorCode(1).String(), + wantReason: "schema not found", + }, + { + name: "deadline exceeded status", + err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.DeadlineExceeded.String(), + wantReason: "context deadline exceeded", + }, + { + name: "deadline exceeded context", + err: context.DeadlineExceeded, + wantType: batch.ErrorTypeRPCError, + wantCode: codes.DeadlineExceeded.String(), + wantReason: context.DeadlineExceeded.Error(), + }, + { + name: "unavailable gateway 502", + err: status.Error(codes.Unavailable, `unexpected HTTP status code received from server: 502 (Bad Gateway)`), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Unavailable.String(), + wantReason: `unexpected HTTP status code received from server: 502 (Bad Gateway)`, + }, + { + name: "internal publish failure", + err: status.Error(codes.Internal, "failed to publish events"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Internal.String(), + wantReason: "failed to publish events", + }, + { + name: "buffer full", + err: batch.ErrMessageBufferFull, + wantType: batch.ErrorTypeBufferFull, + wantReason: "message buffer is full", + }, + { + name: "client shutdown", + err: batch.ErrClientShutdown, + wantType: batch.ErrorTypeClientError, + wantReason: "client is shutdown", + }, + { + name: "unknown error", + err: errors.New("something else"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Unknown.String(), + wantReason: "something else", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := batch.ClassifyDropFailure(tt.err) + assert.Equal(t, tt.wantType, got.ErrorType) + assert.Equal(t, tt.wantCode, got.ErrorCode) + assert.Equal(t, tt.wantReason, got.ErrorReason) + }) + } +} + +func TestClassifyDropFailure_nil(t *testing.T) { + t.Parallel() + got := batch.ClassifyDropFailure(nil) + require.Empty(t, got.ErrorType) +} From b2ed99a6a6051c480f7df2a59e3010b683178b87 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:03:33 -0400 Subject: [PATCH 3/3] chore: bump pkg/chipingress for drop failure classification --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 1299440695..0254d2ef1b 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3441393f7f..699a274548 100644 --- a/go.sum +++ b/go.sum @@ -258,8 +258,8 @@ github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/smartcontractkit/chain-selectors v1.0.100 h1:wpiSpmI/eFjY+wx/nPr5VuNF4hki0prIBMKEaQWn3g4= github.com/smartcontractkit/chain-selectors v1.0.100/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= -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-common/pkg/chipingress v0.0.11-0.20260713200205-3ec922174fe1 h1:pBlv8M7iERb44h6cDyLWIzmh1MgLo4Z/0cHpfPmPs38= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260713200205-3ec922174fe1/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo=