From 094609f5e51ac75c7558f998ea2df03c248d751c Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Thu, 2 Jul 2026 18:17:55 -0400 Subject: [PATCH] synchronization: log and meter partial delivery errors from PublishBatch - chip_ingress_batch_worker.Send was discarding the PublishBatch response. - Now inspects resp.GetResults(): per-event errors are grouped by error code and emitted as one WARN per batch ("dropped", "by_code"). - Adds chip_ingress_partial_delivery_dropped Prometheus counter with telemetry_type and error_code labels so dashboards and alerts can track rejection types without relying on logs. Jira: INFOPLAT-13901 --- .../chip_ingress_batch_worker.go | 25 ++++- .../chip_ingress_batch_worker_test.go | 99 +++++++++++++++++++ core/services/synchronization/metrics.go | 5 + 3 files changed, 128 insertions(+), 1 deletion(-) diff --git a/core/services/synchronization/chip_ingress_batch_worker.go b/core/services/synchronization/chip_ingress_batch_worker.go index c59252a79d6..3fd40da566d 100644 --- a/core/services/synchronization/chip_ingress_batch_worker.go +++ b/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)) diff --git a/core/services/synchronization/chip_ingress_batch_worker_test.go b/core/services/synchronization/chip_ingress_batch_worker_test.go index 2825877cb8e..45ea7da014c 100644 --- a/core/services/synchronization/chip_ingress_batch_worker_test.go +++ b/core/services/synchronization/chip_ingress_batch_worker_test.go @@ -8,6 +8,7 @@ import ( 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 @@ func (noopChipIngressPublisher) PublishBatch(ctx context.Context, batch *pb.Clou 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 @@ func (noopChipIngressPublisher) RegisterSchemas(ctx context.Context, schemas ... 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) diff --git a/core/services/synchronization/metrics.go b/core/services/synchronization/metrics.go index d6bafb7b0d6..329bcf542cb 100644 --- a/core/services/synchronization/metrics.go +++ b/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"}) )