diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 42e0a88cd6..c1dec4b6ed 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) @@ -242,7 +247,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: } @@ -276,7 +281,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..19e0c93698 --- /dev/null +++ b/pkg/chipingress/batch/drop_failure.go @@ -0,0 +1,53 @@ +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" +) + +// ErrorCodeFor returns a bounded error code string for a batch send/queue error. +// The returned code is intended for metric dimensions; callers can infer the +// error category from the code: +// - PUBLISH_ERROR_CODE_* -> partial_delivery +// - "buffer_full" -> buffer_full +// - "client_shutdown" -> client_error +// - gRPC code name -> rpc_error +// - "client_error" -> client_error (fallback for unrecognized client-side errors) +func ErrorCodeFor(err error) string { + if err == nil { + return "" + } + + var pubErr *PublishError + if errors.As(err, &pubErr) { + return pubErr.Code.String() + } + + if errors.Is(err, ErrMessageBufferFull) { + return ErrMessageBufferFull.Error() + } + if errors.Is(err, ErrClientShutdown) { + return ErrClientShutdown.Error() + } + if st, ok := status.FromError(err); ok { + return st.Code().String() + } + if errors.Is(err, context.DeadlineExceeded) { + return codes.DeadlineExceeded.String() + } + if errors.Is(err, context.Canceled) { + return codes.Canceled.String() + } + + return ErrorTypeClientError +} diff --git a/pkg/chipingress/batch/drop_failure_test.go b/pkg/chipingress/batch/drop_failure_test.go new file mode 100644 index 0000000000..0d1361b70a --- /dev/null +++ b/pkg/chipingress/batch/drop_failure_test.go @@ -0,0 +1,78 @@ +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 TestErrorCodeFor(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + want string + }{ + { + name: "partial delivery publish error", + err: &batch.PublishError{Code: chipingress.PublishErrorCode(1), Reason: "schema not found"}, + want: chipingress.PublishErrorCode(1).String(), + }, + { + name: "deadline exceeded status", + err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), + want: codes.DeadlineExceeded.String(), + }, + { + name: "deadline exceeded context", + err: context.DeadlineExceeded, + want: codes.DeadlineExceeded.String(), + }, + { + name: "unavailable gateway 502", + err: status.Error(codes.Unavailable, `unexpected HTTP status code received from server: 502 (Bad Gateway)`), + want: codes.Unavailable.String(), + }, + { + name: "internal publish failure", + err: status.Error(codes.Internal, "failed to publish events"), + want: codes.Internal.String(), + }, + { + name: "buffer full", + err: batch.ErrMessageBufferFull, + want: batch.ErrMessageBufferFull.Error(), + }, + { + name: "client shutdown", + err: batch.ErrClientShutdown, + want: batch.ErrClientShutdown.Error(), + }, + { + name: "unknown error", + err: errors.New("something else"), + want: batch.ErrorTypeClientError, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, batch.ErrorCodeFor(tt.err)) + }) + } +} + +func TestErrorCodeFor_nil(t *testing.T) { + t.Parallel() + require.Empty(t, batch.ErrorCodeFor(nil)) +}