beholder: add per-domain chip ingress emitters (ForDomain, EmitEvent)

go.mod

@@ -41,7 +41,7 @@ require (
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.104
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261026195813-8ed7b172ab8c
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b

pkg/beholder/batch_emitter_service.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
+	"time"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -16,13 +18,21 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 )
 
-// ChipIngressBatchEmitterService batches events and sends them via chipingress.Client.PublishBatch.
-// It implements the Emitter interface.
+// ChipIngressBatchEmitterService batches events and sends them to chip ingress with
+// chipingress.Client.PublishBatch (or PublishBatchNoAck when built with batch.WithNoAck, as domain
+// emitters are). It implements the Emitter interface. A domain emitter also restricts events to
+// its Domain; see DomainEmitter.
 type ChipIngressBatchEmitterService struct {
 	services.Service
 	eng *services.Engine
 
 	batchClient *batch.Client
+	clientName  string
+	// domain, when set, restricts events to this Domain or its subdomains: a client-side guard so
+	// an event is not sent down another domain's route. Unset for the default emitter.
+	domain Domain
+	// errorLogs limits per-event failure logs to one per error code per errorLogInterval.
+	errorLogs *errorLogLimiter
 
 	metricAttrsCache sync.Map // map[string]otelmetric.MeasurementOption
 	metrics          batchEmitterMetrics
@@ -35,6 +45,13 @@ type batchEmitterMetrics struct {
 
 // NewChipIngressBatchEmitterService creates a batch emitter service backed by the given chipingress client.
 func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger) (*ChipIngressBatchEmitterService, error) {
+	return newChipIngressBatchEmitterService(client, cfg, lggr, batch.ClientNameBeholder, "")
+}
+
+// newChipIngressBatchEmitterService is NewChipIngressBatchEmitterService with a client name, a
+// Domain to restrict events to ("" for none), and extra batch options (applied last), used to
+// build domain emitters.
+func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger, clientName string, domain Domain, extraOpts ...batch.Opt) (*ChipIngressBatchEmitterService, error) {
 	if client == nil {
 		return nil, errors.New("chip ingress client is nil")
 	}
@@ -75,7 +92,7 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 		return nil, fmt.Errorf("failed to create batch emitter metrics: %w", err)
 	}
 
-	batchClient, err := batch.NewBatchClient(client,
+	opts := append([]batch.Opt{
 		batch.WithBatchSize(maxBatchSize),
 		batch.WithMessageBuffer(bufferSize),
 		batch.WithBatchInterval(sendInterval),
@@ -84,14 +101,18 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 		batch.WithMaxConcurrentSends(maxConcurrentSends),
 		batch.WithMaxGRPCRequestSize(maxGRPCRequestSize),
 		batch.WithEventClone(false),
-		batch.WithClientName(batch.ClientNameBeholder),
-	)
+		batch.WithClientName(clientName),
+	}, extraOpts...)
+	batchClient, err := batch.NewBatchClient(client, opts...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create batch client: %w", err)
 	}
 
 	e := &ChipIngressBatchEmitterService{
 		batchClient: batchClient,
+		clientName:  clientName,
+		domain:      domain,
+		errorLogs:   newErrorLogLimiter(errorLogInterval),
 		metrics:     metrics,
 	}
 
@@ -113,6 +134,7 @@ func (e *ChipIngressBatchEmitterService) start(_ context.Context) error {
 	return nil
 }
 
+// stop drains the batch client, which then closes the chip ingress client it was built with.
 func (e *ChipIngressBatchEmitterService) stop() error {
 	e.batchClient.Stop()
 	return nil
@@ -135,11 +157,15 @@ func (e *ChipIngressBatchEmitterService) EmitWithCallback(ctx context.Context, b
 }
 
 func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body []byte, callback func(error), attrKVs ...any) error {
-	return e.eng.IfStarted(func() error {
+	var queueErr error
+	if err := e.eng.IfStarted(func() error {
 		domain, entity, err := ExtractSourceAndType(attrKVs...)
 		if err != nil {
 			return err
 		}
+		if err := e.checkDomain(domain); err != nil {
+			return err
+		}
 
 		attributes := newAttributes(attrKVs...)
 
@@ -156,54 +182,122 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body
 			return err
 		}
 
-		metricAttrs := e.metricAttrsFor(domain, entity)
-
-		queueErr := e.batchClient.QueueMessage(eventPb, func(sendErr error) {
-			// The callback fires asynchronously after the batch is sent,
-			// so the caller's ctx may already be cancelled. Use ctx directly
-			// for metric recording — OTel Add is non-blocking and tolerates
-			// cancelled contexts.
-			if sendErr != nil {
-				errorCode := batch.ErrorCodeFor(sendErr)
-				e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode))
-				// Partial delivery is not logged: it is a per-event, often persistent
-				// server-side condition (e.g. missing schema) that would otherwise log on
-				// every dropped event. chip_ingress.events_dropped (error_code) already
-				// captures that it's happening and roughly why; the full reason isn't
-				// needed at fleet-wide log volume.
-				if _, ok := errors.AsType[*batch.PublishError](sendErr); !ok {
+		queueErr = e.queue(ctx, eventPb, domain, entity, callback)
+		return nil
+	}); err != nil {
+		return err
+	}
+	e.failQueued(callback, queueErr)
+	return nil
+}
+
+// EmitEvent queues a fully built CloudEvent for batched delivery without blocking. Unlike Emit,
+// which builds the event from attributes, it sends the event as given (see DomainEmitter.EmitEvent
+// for what that means, including the "seqnum" overwrite and not reusing the event afterwards).
+//
+// If EmitEvent returns a non-nil error, the callback will NOT be invoked.
+// If it returns nil, a non-nil callback is guaranteed to fire exactly once: synchronously before
+// EmitEvent returns if the event cannot be queued, otherwise after the batch is sent (possibly
+// after Close returns).
+func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event chipingress.CloudEvent, callback func(error)) error {
+	var queueErr error
+	if err := e.eng.IfStarted(func() error {
+		if err := event.Validate(); err != nil {
+			return fmt.Errorf("invalid CloudEvent: %w", err)
+		}
+		if err := e.checkDomain(event.Source()); err != nil {
+			return err
+		}
+		eventPb, err := chipingress.EventToProto(event)
+		if err != nil {
+			return fmt.Errorf("failed to convert to proto: %w", err)
+		}
+
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
+		queueErr = e.queue(ctx, eventPb, event.Source(), event.Type(), callback)
+		return nil
+	}); err != nil {
+		return err
+	}
+	e.failQueued(callback, queueErr)
+	return nil
+}
+
+// checkDomain rejects a source outside e.domain: it must be the domain itself or a subdomain of
+// it (e.g. "data-feeds" or "data-feeds.telemetry.ocr2-median" for DomainDataFeeds). It is a
+// client-side guard against misrouting, not server-side enforcement.
+func (e *ChipIngressBatchEmitterService) checkDomain(source string) error {
+	if e.domain == "" || source == string(e.domain) || strings.HasPrefix(source, string(e.domain)+".") {
+		return nil
+	}
+	return fmt.Errorf("beholder: event source %q is not in domain %q", source, e.domain)
+}
+
+// failQueued reports a queueing failure to the caller's callback. It runs after the service's
+// lock is released, so a callback may safely call back into the emitter (e.g. to retry) even
+// while Close is waiting to stop the service.
+func (e *ChipIngressBatchEmitterService) failQueued(callback func(error), queueErr error) {
+	if queueErr != nil && callback != nil {
+		callback(queueErr)
+	}
+}
+
+// queue hands eventPb to the batch client and records its fate. domain and entity are the
+// event's source and type, used for metrics and logs. A queueing failure is recorded and
+// returned, not passed to callback: the caller does that once the service's lock is released.
+func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chipingress.CloudEventPb, domain, entity string, callback func(error)) error {
+	metricAttrs := e.metricAttrsFor(domain, entity)
+
+	queueErr := e.batchClient.QueueMessage(eventPb, func(sendErr error) {
+		// The callback fires asynchronously after the batch is sent,
+		// so the caller's ctx may already be cancelled. Use ctx directly
+		// for metric recording — OTel Add is non-blocking and tolerates
+		// cancelled contexts.
+		if sendErr != nil {
+			errorCode := batch.ErrorCodeFor(sendErr)
+			e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode))
+			// Partial delivery is not logged: it is a per-event, often persistent
+			// server-side condition (e.g. missing schema) that would otherwise log on
+			// every dropped event. chip_ingress.events_dropped (error_code) already
+			// captures that it's happening and roughly why; the full reason isn't
+			// needed at fleet-wide log volume.
+			if _, ok := errors.AsType[*batch.PublishError](sendErr); !ok {
+				if log, suppressed := e.errorLogs.allow("emit:"+errorCode, time.Now()); log {
 					e.eng.Errorw("failed to emit to chip ingress",
 						"error", sendErr,
 						"error_code", errorCode,
 						"error_reason", sendErr.Error(),
 						"domain", domain,
 						"entity", entity,
+						"suppressed_since_last_log", suppressed,
 					)
 				}
-			} else {
-				e.metrics.eventsSent.Add(ctx, 1, metricAttrs)
-			}
-			if callback != nil {
-				callback(sendErr)
 			}
-		})
-		if queueErr != nil {
-			errorCode := batch.ErrorCodeFor(queueErr)
-			e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode))
+		} else {
+			e.metrics.eventsSent.Add(ctx, 1, metricAttrs)
+		}
+		if callback != nil {
+			callback(sendErr)
+		}
+	})
+	if queueErr != nil {
+		errorCode := batch.ErrorCodeFor(queueErr)
+		e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode))
+		if log, suppressed := e.errorLogs.allow("queue:"+errorCode, time.Now()); log {
 			e.eng.Errorw("failed to queue message for chip ingress",
 				"error", queueErr,
 				"error_code", errorCode,
 				"error_reason", queueErr.Error(),
 				"domain", domain,
 				"entity", entity,
+				"suppressed_since_last_log", suppressed,
 			)
-			if callback != nil {
-				callback(queueErr)
-			}
 		}
-
-		return nil
-	})
+	}
+	return queueErr
 }
 
 func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) otelmetric.MeasurementOption {
@@ -214,7 +308,7 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o
 	attrs := otelmetric.WithAttributeSet(attribute.NewSet(
 		attribute.String("domain", domain),
 		attribute.String("entity", entity),
-		attribute.String("client_name", batch.ClientNameBeholder),
+		attribute.String("client_name", e.clientName),
 	))
 	v, _ := e.metricAttrsCache.LoadOrStore(key, attrs)
 	return v.(otelmetric.MeasurementOption)
@@ -232,7 +326,7 @@ func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, erro
 	attrs := []attribute.KeyValue{
 		attribute.String("domain", domain),
 		attribute.String("entity", entity),
-		attribute.String("client_name", batch.ClientNameBeholder),
+		attribute.String("client_name", e.clientName),
 	}
 	if errorCode != "" {
 		attrs = append(attrs, attribute.String("error_code", errorCode))

pkg/beholder/batch_emitter_service_reentrant_test.go

@@ -0,0 +1,79 @@
+package beholder
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+// hangingNoAckClient never answers PublishBatchNoAck until released or the call's context ends.
+type hangingNoAckClient struct {
+	chipingress.NoopClient
+	release chan struct{}
+}
+
+func (c *hangingNoAckClient) PublishBatchNoAck(ctx context.Context, _ *chipingress.CloudEventBatch, _ ...grpc.CallOption) (*chipingress.PublishResponse, error) {
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-c.release:
+		return &chipingress.PublishResponse{}, nil
+	}
+}
+
+// A callback that re-enters EmitEvent (e.g. to retry a buffer-full drop) while Close is waiting
+// to stop the service must not deadlock: the callback must not run under the service's lock.
+func TestChipIngressBatchEmitterService_ReentrantCallbackDuringClose(t *testing.T) {
+	client := &hangingNoAckClient{release: make(chan struct{})}
+	t.Cleanup(func() { close(client.release) })
+
+	svc, err := newChipIngressBatchEmitterService(client, Config{
+		ChipIngressBufferSize:         1,
+		ChipIngressMaxBatchSize:       1,
+		ChipIngressMaxConcurrentSends: 1,
+		ChipIngressSendInterval:       10 * time.Millisecond,
+		ChipIngressSendTimeout:        30 * time.Second,
+		ChipIngressDrainTimeout:       200 * time.Millisecond,
+	}, logger.Test(t), "test", "", batch.WithNoAck())
+	require.NoError(t, err)
+	require.NoError(t, svc.Start(t.Context()))
+
+	event := func(i int) chipingress.CloudEvent {
+		ev, err := chipingress.NewEvent("test-domain", "Entity", []byte(fmt.Sprint(i)), nil)
+		require.NoError(t, err)
+		return ev
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		var retried atomic.Bool
+		for i := 0; i < 100 && !retried.Load(); i++ {
+			_ = svc.EmitEvent(context.Background(), event(i), func(err error) {
+				if !errors.Is(err, batch.ErrMessageBufferFull) || !retried.CompareAndSwap(false, true) {
+					return
+				}
+				go func() { _ = svc.Close() }()
+				time.Sleep(100 * time.Millisecond) // let Close start waiting to stop the service
+				_ = svc.EmitEvent(context.Background(), event(-1), nil)
+			})
+			time.Sleep(5 * time.Millisecond)
+		}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("deadlock: a callback re-entering EmitEvent during Close never returned")
+	}
+}