synchronization: log per-event partial delivery errors from PublishBatch

core/services/synchronization/chip_ingress_batch_worker.go

@@ -70,13 +70,36 @@ 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.
+	// Group by error code so a large batch with many failures produces one log line.
+	var partialDrops int
+	byCode := map[string]int{}
+	for _, result := range resp.GetResults() {
+		if e := result.GetError(); e != nil {
+			code := e.GetErrorCode().String()
+			byCode[code]++
+			partialDrops++
+			TelemetryClientMessagesDropped.WithLabelValues(chipIngress, string(cw.telemType)).Inc()
+			ChipIngressPartialDeliveryDropped.WithLabelValues(string(cw.telemType), code).Inc()
+		}
+	}
+	if partialDrops > 0 {
+		cw.lggr.Warnw("chip ingress partial delivery errors",
+			"dropped", partialDrops,
+			"by_code", byCode,
+			"telemType", cw.telemType,
+			"contractID", cw.contractID,
+		)
+	}
+
 	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

@@ -8,6 +8,7 @@
 	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
 	"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 +26,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 +78,72 @@
 	return nil, nil
 }
 
+func TestChipIngressBatchWorker_Send_PartialDelivery(t *testing.T) {
+	// Verify that Send groups per-event errors into a single WARN log line
+	// (partial delivery, transactionEnabled=false).
+	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",
+	}
+
+	require.NotPanics(t, func() { worker.Send(t.Context()) })
+	assert.Empty(t, chTelemetry)
+
+	// Two failed events must produce exactly one grouped WARN log line.
+	logs := observed.FilterMessage("chip ingress partial delivery errors")
+	require.Equal(t, 1, logs.Len(), "expected exactly one grouped log line for 2 failed events")
+	assert.Equal(t, zap.WarnLevel, logs.All()[0].Level)
+}
+
 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"})
 )