From 6cade0fe635156eecbfdc4a4e764e9c3f99e1a20 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Tue, 21 Jul 2026 00:23:25 -0400 Subject: [PATCH 1/2] synchronization: log and throttle partial delivery errors from PublishBatch Jira: INFOPLAT-13901 --- .../chip_ingress_batch_worker.go | 36 ++++- .../chip_ingress_batch_worker_test.go | 151 ++++++++++++++++++ core/services/synchronization/metrics.go | 5 + 3 files changed, 191 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..e75cd4e219a 100644 --- a/core/services/synchronization/chip_ingress_batch_worker.go +++ b/core/services/synchronization/chip_ingress_batch_worker.go @@ -30,6 +30,7 @@ type chipIngressBatchWorker struct { logging bool lggr logger.Logger dropMessageCount atomic.Uint32 + partialDropCount atomic.Uint32 } // NewChipIngressBatchWorker returns a worker for a given contractID that can send @@ -70,13 +71,46 @@ 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 { + // Partial delivery is often caused by a persistent condition (missing/misconfigured + // schema), which would otherwise re-trigger this WARN on every send interval + // (as low as 500ms) for as long as the condition lasts. Throttle with the same + // exponential-then-linear backoff used by logBufferFullWithExpBackoff. + count := cw.partialDropCount.Add(uint32(partialDrops)) + if count > 0 && (count%100 == 0 || count&(count-1) == 0) { + cw.lggr.Warnw("chip ingress partial delivery errors", + "dropped", partialDrops, + "totalDropped", count, + "by_code", byCode, + "telemType", cw.telemType, + "contractID", cw.contractID, + ) + } + } else { + cw.partialDropCount.Store(0) + } + 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 8dc68ad5fbd..3c5b86ee4ca 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,124 @@ func (noopChipIngressPublisher) RegisterSchemas(ctx context.Context, schemas ... return nil, nil } +func TestChipIngressBatchWorker_Send_PartialDelivery(t *testing.T) { + t.Parallel() + // 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_Send_PartialDeliveryThrottled(t *testing.T) { + t.Parallel() + // A persistent partial-delivery condition (e.g. missing schema) must not log on every + // Send call - it should follow the same backoff cadence as logBufferFullWithExpBackoff. + singleFailureResp := &pb.PublishResponse{ + Results: []*pb.PublishResult{ + { + EventId: "evt-1", + Error: &pb.PublishError{ + ErrorCode: pb.PublishErrorCode_PUBLISH_ERROR_CODE_SCHEMA_MISSING, + Reason: "schema not found", + }, + }, + }, + } + publisher := partialChipIngressPublisher{resp: singleFailureResp} + + lggr, observed := logger.TestLoggerObserved(t, zap.WarnLevel) + chTelemetry := make(chan TelemPayload, 1) + worker := NewChipIngressBatchWorker( + 1, + time.Second, + publisher, + chTelemetry, + "0xabc", + OCR, + lggr, + true, + ) + + payload := TelemPayload{ + Telemetry: []byte("payload"), + TelemType: OCR, + ContractID: "0xabc", + Domain: "data-feeds", + Entity: "ocr.v1.telemetry", + ChainSelector: 7700, + Network: "EVM", + ChainID: "1", + } + + // Drop counts after each send: 1, 2, 3. Backoff logs at 1 and 2 (powers of two), not at 3. + for range 3 { + chTelemetry <- payload + require.NotPanics(t, func() { worker.Send(t.Context()) }) + } + + logs := observed.FilterMessage("chip ingress partial delivery errors") + assert.Equal(t, 2, logs.Len(), "expected the third consecutive failure to be throttled") +} + 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"}) ) From 68635a2bd59adcb13a4eda31e8241f3d9bcbe1c8 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Tue, 21 Jul 2026 01:11:53 -0400 Subject: [PATCH 2/2] synchronization: remove partial delivery WARN log, keep metric only Partial delivery is a per-event, often persistent server-side condition (e.g. missing schema), so logging it would WARN on every send interval (as low as 500ms) for as long as it lasts. ChipIngressPartialDeliveryDropped (telemetry_type, error_code) already reports it with bounded cardinality and is the intended signal for alerting/investigation. --- .../chip_ingress_batch_worker.go | 29 ++------ .../chip_ingress_batch_worker_test.go | 67 ++++--------------- 2 files changed, 17 insertions(+), 79 deletions(-) diff --git a/core/services/synchronization/chip_ingress_batch_worker.go b/core/services/synchronization/chip_ingress_batch_worker.go index e75cd4e219a..15ebd21a3df 100644 --- a/core/services/synchronization/chip_ingress_batch_worker.go +++ b/core/services/synchronization/chip_ingress_batch_worker.go @@ -30,7 +30,6 @@ type chipIngressBatchWorker struct { logging bool lggr logger.Logger dropMessageCount atomic.Uint32 - partialDropCount atomic.Uint32 } // NewChipIngressBatchWorker returns a worker for a given contractID that can send @@ -80,36 +79,18 @@ func (cw *chipIngressBatchWorker) Send(ctx context.Context) { // 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{} + // + // 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() - byCode[code]++ - partialDrops++ TelemetryClientMessagesDropped.WithLabelValues(chipIngress, string(cw.telemType)).Inc() ChipIngressPartialDeliveryDropped.WithLabelValues(string(cw.telemType), code).Inc() } } - if partialDrops > 0 { - // Partial delivery is often caused by a persistent condition (missing/misconfigured - // schema), which would otherwise re-trigger this WARN on every send interval - // (as low as 500ms) for as long as the condition lasts. Throttle with the same - // exponential-then-linear backoff used by logBufferFullWithExpBackoff. - count := cw.partialDropCount.Add(uint32(partialDrops)) - if count > 0 && (count%100 == 0 || count&(count-1) == 0) { - cw.lggr.Warnw("chip ingress partial delivery errors", - "dropped", partialDrops, - "totalDropped", count, - "by_code", byCode, - "telemType", cw.telemType, - "contractID", cw.contractID, - ) - } - } else { - cw.partialDropCount.Store(0) - } TelemetryClientMessagesSent.WithLabelValues(chipIngress, string(cw.telemType)).Inc() if cw.logging { diff --git a/core/services/synchronization/chip_ingress_batch_worker_test.go b/core/services/synchronization/chip_ingress_batch_worker_test.go index 3c5b86ee4ca..f3acf2fe642 100644 --- a/core/services/synchronization/chip_ingress_batch_worker_test.go +++ b/core/services/synchronization/chip_ingress_batch_worker_test.go @@ -6,6 +6,7 @@ import ( "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" @@ -80,8 +81,11 @@ func (noopChipIngressPublisher) RegisterSchemas(ctx context.Context, schemas ... func TestChipIngressBatchWorker_Send_PartialDelivery(t *testing.T) { t.Parallel() - // Verify that Send groups per-event errors into a single WARN log line - // (partial delivery, transactionEnabled=false). + // 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{ { @@ -136,64 +140,17 @@ func TestChipIngressBatchWorker_Send_PartialDelivery(t *testing.T) { 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) - // 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_Send_PartialDeliveryThrottled(t *testing.T) { - t.Parallel() - // A persistent partial-delivery condition (e.g. missing schema) must not log on every - // Send call - it should follow the same backoff cadence as logBufferFullWithExpBackoff. - singleFailureResp := &pb.PublishResponse{ - Results: []*pb.PublishResult{ - { - EventId: "evt-1", - Error: &pb.PublishError{ - ErrorCode: pb.PublishErrorCode_PUBLISH_ERROR_CODE_SCHEMA_MISSING, - Reason: "schema not found", - }, - }, - }, - } - publisher := partialChipIngressPublisher{resp: singleFailureResp} - - lggr, observed := logger.TestLoggerObserved(t, zap.WarnLevel) - chTelemetry := make(chan TelemPayload, 1) - worker := NewChipIngressBatchWorker( - 1, - time.Second, - publisher, - chTelemetry, - "0xabc", - OCR, - lggr, - true, - ) - - payload := TelemPayload{ - Telemetry: []byte("payload"), - TelemType: OCR, - ContractID: "0xabc", - Domain: "data-feeds", - Entity: "ocr.v1.telemetry", - ChainSelector: 7700, - Network: "EVM", - ChainID: "1", - } - - // Drop counts after each send: 1, 2, 3. Backoff logs at 1 and 2 (powers of two), not at 3. - for range 3 { - chTelemetry <- payload - require.NotPanics(t, func() { worker.Send(t.Context()) }) - } + 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.Equal(t, 2, logs.Len(), "expected the third consecutive failure to be throttled") + assert.Zero(t, logs.Len(), "partial delivery drops must not be logged") } func TestChipIngressBatchWorker_BuildCloudEventBatch(t *testing.T) {