synchronization: log per-event partial delivery errors from PublishBatch

core/services/synchronization/chip_ingress_batch_worker.go

@@ -70,13 +70,28 @@ func (cw *chipIngressBatchWorker) Send(ctx context.Context) {
 	ctx, cancel := context.WithTimeout(ctx, cw.sendTimeout)
 	defer cancel()
 
-	_, err := cw.chipClient.PublishBatch(ctx, batch)
+	resp, err := cw.chipClient.PublishBatch(ctx, batch)
 	if err != nil {
 		cw.lggr.Warnf("Could not send telemetry via ChIP ingress: %v", err)
 		TelemetryClientMessagesSendErrors.WithLabelValues(chipIngress, string(cw.telemType)).Inc()
 		return
 	}
 
+	// Inspect per-event results for partial delivery errors. When transaction_enabled=false
+	// (the default), the server can accept some events while rejecting others.
+	//
+	// This is intentionally not logged: it's a per-event, often persistent server-side
+	// condition (e.g. missing schema) that would otherwise WARN on every send interval
+	// (as low as 500ms) for as long as it lasts. ChipIngressPartialDeliveryDropped
+	// (telemetry_type, error_code) is the intended signal for alerting/investigation.
+	for _, result := range resp.GetResults() {
+		if e := result.GetError(); e != nil {
+			code := e.GetErrorCode().String()
+			TelemetryClientMessagesDropped.WithLabelValues(chipIngress, string(cw.telemType)).Inc()
+			ChipIngressPartialDeliveryDropped.WithLabelValues(string(cw.telemType), code).Inc()
+		}
+	}
+
 	TelemetryClientMessagesSent.WithLabelValues(chipIngress, string(cw.telemType)).Inc()
 	if cw.logging {
 		cw.lggr.Debugw("Successfully sent telemetry to ChIP ingress", "contractID", cw.contractID, "telemType", cw.telemType, "batchSize", len(batch.Events))

core/services/synchronization/chip_ingress_batch_worker_test.go

@@ -6,8 +6,10 @@
 	"time"
 
 	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
@@ -25,6 +27,38 @@
 	return &pb.PublishResponse{}, nil
 }
 
+type partialChipIngressPublisher struct {
+	resp *pb.PublishResponse
+}
+
+func (p partialChipIngressPublisher) Publish(ctx context.Context, event *cepb.CloudEvent, opts ...grpc.CallOption) (*pb.PublishResponse, error) {
+	return &pb.PublishResponse{}, nil
+}
+
+func (p partialChipIngressPublisher) PublishBatch(ctx context.Context, batch *pb.CloudEventBatch, opts ...grpc.CallOption) (*pb.PublishResponse, error) {
+	return p.resp, nil
+}
+
+func (p partialChipIngressPublisher) Ping(ctx context.Context, req *pb.EmptyRequest, opts ...grpc.CallOption) (*pb.PingResponse, error) {
+	return &pb.PingResponse{}, nil
+}
+
+func (p partialChipIngressPublisher) StreamEvents(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[pb.StreamEventsRequest, pb.StreamEventsResponse], error) {
+	return nil, nil
+}
+
+func (p partialChipIngressPublisher) RegisterSchema(ctx context.Context, req *pb.RegisterSchemaRequest, opts ...grpc.CallOption) (*pb.RegisterSchemaResponse, error) {
+	return &pb.RegisterSchemaResponse{}, nil
+}
+
+func (p partialChipIngressPublisher) Close() error {
+	return nil
+}
+
+func (p partialChipIngressPublisher) RegisterSchemas(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error) {
+	return nil, nil
+}
+
 func (noopChipIngressPublisher) Ping(ctx context.Context, req *pb.EmptyRequest, opts ...grpc.CallOption) (*pb.PingResponse, error) {
 	return &pb.PingResponse{}, nil
 }
@@ -45,6 +79,80 @@
 	return nil, nil
 }
 
+func TestChipIngressBatchWorker_Send_PartialDelivery(t *testing.T) {
+	t.Parallel()
+	// Verify that Send records per-event partial-delivery errors as metrics without
+	// logging: partial delivery is a per-event, often persistent condition (e.g. missing
+	// schema), so logging it would spam at fleet-wide volume. The
+	// ChipIngressPartialDeliveryDropped metric (telemetry_type, error_code) is the
+	// intended signal.
+	partialResp := &pb.PublishResponse{
+		Results: []*pb.PublishResult{
+			{
+				EventId: "evt-1",
+				Error: &pb.PublishError{
+					ErrorCode: pb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED,
+					Reason:    "schema not found",
+				},
+			},
+			{
+				EventId: "evt-2",
+				Error: &pb.PublishError{
+					ErrorCode: pb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED,
+					Reason:    "schema not found",
+				},
+			},
+		},
+	}
+	publisher := partialChipIngressPublisher{resp: partialResp}
+
+	lggr, observed := logger.TestLoggerObserved(t, zap.WarnLevel)
+	chTelemetry := make(chan TelemPayload, 5)
+	worker := NewChipIngressBatchWorker(
+		2,
+		time.Second,
+		publisher,
+		chTelemetry,
+		"0xabc",
+		OCR,
+		lggr,
+		true,
+	)
+
+	chTelemetry <- TelemPayload{
+		Telemetry:     []byte("payload1"),
+		TelemType:     OCR,
+		ContractID:    "0xabc",
+		Domain:        "data-feeds",
+		Entity:        "ocr.v1.telemetry",
+		ChainSelector: 7700,
+		Network:       "EVM",
+		ChainID:       "1",
+	}
+	chTelemetry <- TelemPayload{
+		Telemetry:     []byte("payload2"),
+		TelemType:     OCR,
+		ContractID:    "0xabc",
+		Domain:        "data-feeds",
+		Entity:        "ocr.v1.telemetry",
+		ChainSelector: 7700,
+		Network:       "EVM",
+		ChainID:       "1",
+	}
+
+	code := pb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED.String()
+	before := testutil.ToFloat64(ChipIngressPartialDeliveryDropped.WithLabelValues(string(OCR), code))
+
+	require.NotPanics(t, func() { worker.Send(t.Context()) })
+	assert.Empty(t, chTelemetry)
+
+	after := testutil.ToFloat64(ChipIngressPartialDeliveryDropped.WithLabelValues(string(OCR), code))
+	assert.Equal(t, float64(2), after-before, "expected both failed events to increment the metric")
+
+	logs := observed.FilterMessage("chip ingress partial delivery errors")
+	assert.Zero(t, logs.Len(), "partial delivery drops must not be logged")
+}
+
 func TestChipIngressBatchWorker_BuildCloudEventBatch(t *testing.T) {
 	maxBatchSize := 3
 	chTelemetry := make(chan TelemPayload, 10)

core/services/synchronization/metrics.go

@@ -30,4 +30,9 @@ var (
 		Name: "telemetry_client_workers",
 		Help: "Number of telemetry workers",
 	}, []string{"endpoint", "telemetry_type"})
+
+	ChipIngressPartialDeliveryDropped = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "chip_ingress_partial_delivery_dropped",
+		Help: "Number of chip ingress events dropped due to per-event server rejection (partial delivery)",
+	}, []string{"telemetry_type", "error_code"})
 )