[1/2] chipingress: expose drop failure classification for partial delivery metrics

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
 	}
 }
 

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
+}

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))
+}