beholder: inject emitter in remote executable ClientRequest (Phase 3)

core/capabilities/launcher.go

@@ -59,6 +59,10 @@ type launcher struct {
 	metrics             *launcherMetrics
 	localCapMgr         localcapmgr.LocalCapabilityManager
 
+	// durableEmit persists events emitted by remote executable clients (e.g. the
+	// transmission schedule). Nil falls back to durableemitter.GlobalEmit.
+	durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error
+
 	// shardingEnabled and shardIndex carry this node's local shard identity,
 	// taken from the [Sharding] config section. shardIndex is only meaningful
 	// when shardingEnabled is true; it is 0 for unsharded deployments.
@@ -113,6 +117,15 @@ type shimID struct {
 
 // TODO: add metric handler and instrument all the internal log.Error calls
 
+// LauncherOption configures a launcher.
+type LauncherOption func(*launcher)
+
+// WithDurableEmit sets the durable emit func remote executable clients use for
+// transmission schedule events. Nil keeps the default, durableemitter.GlobalEmit.
+func WithDurableEmit(fn func(ctx context.Context, body []byte, attrKVs ...any) error) LauncherOption {
+	return func(l *launcher) { l.durableEmit = fn }
+}
+
 // NewLauncher creates a new capabilities launcher.
 // don2donSharedPeer is required; it provides the node's peer ID and manages DON-to-DON connections.
 func NewLauncher(
@@ -125,6 +138,7 @@ func NewLauncher(
 	limitsFactory limits.Factory,
 	shardingEnabled bool,
 	shardIndex uint16,
+	opts ...LauncherOption,
 ) (*launcher, error) {
 	if don2donSharedPeer == nil {
 		return nil, errors.New("don2donSharedPeer is required")
@@ -159,7 +173,7 @@ func NewLauncher(
 	if err != nil {
 		return nil, fmt.Errorf("failed to create propagate org ID gate limiter: %w", err)
 	}
-	return &launcher{
+	l := &launcher{
 		lggr:       logger.Sugared(lggr).Named("CapabilitiesLauncher"),
 		dispatcher: dispatcher,
 		cachedShims: cachedShims{
@@ -180,7 +194,11 @@ func NewLauncher(
 		shardingEnabled:        shardingEnabled,
 		shardIndex:             shardIndex,
 		stopCh:                 make(services.StopChan),
-	}, nil
+	}
+	for _, opt := range opts {
+		opt(l)
+	}
+	return l, nil
 }
 
 func (w *launcher) publicDONs(
@@ -736,7 +754,7 @@ func (w *launcher) addRemoteCapabilityV2(ctx context.Context, capID string, meth
 		} else { // executable
 			client, alreadyExists := w.cachedShims.executableClients[key]
 			if !alreadyExists {
-				client = executable.NewClient(info.ID, method, w.dispatcher, w.lggr)
+				client = executable.NewClient(info.ID, method, w.dispatcher, w.lggr, executable.WithDurableEmit(w.durableEmit))
 				cc.SetExecutableClient(method, client)
 				// add to cachedShims later, only after startNewShim succeeds
 			}

core/capabilities/remote/executable/client.go

@@ -32,6 +32,11 @@ type client struct {
 	cfg           atomic.Pointer[dynamicConfig]
 	lggr          logger.Logger
 
+	// durableEmit persists transmission schedule events emitted by requests
+	// created through this client. Nil falls back to durableemitter.GlobalEmit
+	// in the request constructor.
+	durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error
+
 	requestIDToCallerRequest map[string]*request.ClientRequest
 	mutex                    sync.Mutex
 	stopCh                   services.StopChan
@@ -55,9 +60,11 @@ type Client interface {
 	SetConfig(remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, requestTimeout time.Duration, signers [][]byte, minResponsesToAggregate uint32) error
 }
 
-var _ Client = &client{}
-var _ types.Receiver = &client{}
-var _ services.Service = &client{}
+var (
+	_ Client           = &client{}
+	_ types.Receiver   = &client{}
+	_ services.Service = &client{}
+)
 
 const defaultExpiryCheckInterval = 30 * time.Second
 
@@ -66,15 +73,29 @@ var (
 	ErrContextDoneBeforeResponseQuorum = errors.New("context done before remote client received a quorum of responses")
 )
 
-func NewClient(capabilityID string, capMethodName string, dispatcher types.Dispatcher, lggr logger.Logger) *client {
-	return &client{
+// ClientOption configures a client.
+type ClientOption func(*client)
+
+// WithDurableEmit sets the durable emit func for transmission schedule events
+// emitted by requests created through this client. Nil keeps the default,
+// durableemitter.GlobalEmit.
+func WithDurableEmit(fn func(ctx context.Context, body []byte, attrKVs ...any) error) ClientOption {
+	return func(c *client) { c.durableEmit = fn }
+}
+
+func NewClient(capabilityID, capMethodName string, dispatcher types.Dispatcher, lggr logger.Logger, opts ...ClientOption) *client {
+	c := &client{
 		capabilityID:             capabilityID,
 		capMethodName:            capMethodName,
 		dispatcher:               dispatcher,
 		lggr:                     logger.With(logger.Named(lggr, "ExecutableCapabilityClient"), "capabilityID", capabilityID, "capMethodName", capMethodName),
 		requestIDToCallerRequest: make(map[string]*request.ClientRequest),
 		stopCh:                   make(services.StopChan),
 	}
+	for _, opt := range opts {
+		opt(c)
+	}
+	return c
 }
 
 // SetConfig sets the remote capability configuration dynamically
@@ -235,7 +256,8 @@ func (c *client) Execute(ctx context.Context, capReq commoncap.CapabilityRequest
 	}
 
 	req, err := request.NewClientExecuteRequest(ctx, c.lggr, capReq, cfg.remoteCapabilityInfo, cfg.localDONInfo, c.dispatcher,
-		cfg.requestTimeout, c.capMethodName, cfg.signers, cfg.minResponsesToAggregate)
+		cfg.requestTimeout, c.capMethodName, cfg.signers, cfg.minResponsesToAggregate,
+		request.WithDurableEmit(c.durableEmit))
 	if err != nil {
 		return commoncap.CapabilityResponse{}, fmt.Errorf("failed to create client request: %w", err)
 	}

core/capabilities/remote/executable/client_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 	"github.com/smartcontractkit/chainlink-protos/cre/go/values"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable"
+	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable/request"
 	remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types"
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/synctest"
 	p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types"
@@ -537,3 +538,48 @@ func TestClient_SetConfig_StartClose(t *testing.T) {
 		require.NoError(t, fixture.Client.Close())
 	})
 }
+
+func Test_Client_Execute_InjectsDurableEmit(t *testing.T) {
+	t.Parallel()
+
+	capabilityID := "test_capability@1.0.0"
+	donInfo := commoncap.DON{ID: 1, Members: []p2ptypes.PeerID{NewP2PPeerID(t)}, F: 0}
+	capInfo := commoncap.CapabilityInfo{ID: capabilityID, CapabilityType: commoncap.CapabilityTypeAction, DON: &donInfo}
+
+	inputs, err := values.NewMap(map[string]any{"input": "value"})
+	require.NoError(t, err)
+
+	var (
+		mu       sync.Mutex
+		gotAttrs []any
+	)
+	client := executable.NewClient(capabilityID, "execute", &noopDispatcher{}, logger.Test(t),
+		executable.WithDurableEmit(func(_ context.Context, _ []byte, attrKVs ...any) error {
+			mu.Lock()
+			defer mu.Unlock()
+			gotAttrs = attrKVs
+			return nil
+		}))
+	require.NoError(t, client.SetConfig(capInfo, donInfo, time.Minute, nil, 0))
+	servicetest.Run(t, client)
+
+	// Cancel up front: Execute emits the transmission schedule before it waits
+	// for a response quorum, so the durable sink must have been called by then.
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	_, err = client.Execute(ctx, commoncap.CapabilityRequest{
+		Metadata: commoncap.RequestMetadata{
+			WorkflowID:          workflowID1,
+			WorkflowExecutionID: workflowExecutionID1,
+			ReferenceID:         stepReferenceID1,
+		},
+		Inputs: inputs,
+	})
+	require.ErrorIs(t, err, executable.ErrContextDoneBeforeResponseQuorum)
+
+	entity := request.TransmissionEventProtoPkg + "." + request.TransmissionEventEntity
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, []any{"source", "platform", "type", entity}, gotAttrs)
+}

core/capabilities/remote/executable/request/client_request.go

@@ -46,7 +46,7 @@ func newRemoteCapabilityExecuteError(transportErr types.Error, errMsg string) er
 	}
 }
 
-func newRemoteCapabilityExecuteErrorWithMessage(display string, errMsg string) error {
+func newRemoteCapabilityExecuteErrorWithMessage(display, errMsg string) error {
 	return &errRemoteCapabilityExecuteError{
 		s:    display,
 		wrap: caperrors.DeserializeErrorFromString(errMsg),
@@ -99,12 +99,35 @@ type ClientRequest struct {
 	wg       *sync.WaitGroup
 }
 
+// clientRequestConfig holds the sinks for the transmission schedule event.
+// Zero values resolve to the late-bound globals.
+type clientRequestConfig struct {
+	emitter beholder.Emitter
+	durable func(ctx context.Context, body []byte, attrKVs ...any) error
+}
+
+// ClientRequestOption configures the sinks used for a ClientRequest's
+// transmission schedule event.
+type ClientRequestOption func(*clientRequestConfig)
+
+// WithEmitter sets the beholder emitter used for the transmission schedule
+// event. A nil emitter defaults to beholder.GlobalEmitter(), which resolves
+// the global client at emit time.
+func WithEmitter(e beholder.Emitter) ClientRequestOption {
+	return func(c *clientRequestConfig) { c.emitter = e }
+}
+
+// WithDurableEmit overrides the durable emit func (default durableemitter.GlobalEmit).
+func WithDurableEmit(fn func(ctx context.Context, body []byte, attrKVs ...any) error) ClientRequestOption {
+	return func(c *clientRequestConfig) { c.durable = fn }
+}
+
 // NewClientExecuteRequest creates a new client request for a remote executable capability.
 // V2 capabilities always use the AllAtOnce transmission schedule.
 func NewClientExecuteRequest(ctx context.Context, lggr logger.Logger, req commoncap.CapabilityRequest,
 	remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, dispatcher types.Dispatcher,
 	requestTimeout time.Duration, capMethodName string,
-	signers [][]byte, minResponsesToAggregate uint32,
+	signers [][]byte, minResponsesToAggregate uint32, opts ...ClientRequestOption,
 ) (*ClientRequest, error) {
 	rawRequest, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb.CapabilityRequestToProto(req))
 	if err != nil {
@@ -121,7 +144,22 @@ func NewClientExecuteRequest(ctx context.Context, lggr logger.Logger, req common
 	requestID := types.MethodExecute + ":" + workflowExecutionID + ":" + req.Metadata.ReferenceID
 
 	lggr = logger.With(lggr, "requestId", requestID) // cap ID and method name included in the parent logger
-	return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, types.MethodExecute, rawRequest, workflowExecutionID, req.Metadata.ReferenceID, capMethodName, signers, minResponsesToAggregate)
+
+	// Apply options and default the sinks before the transmission schedule
+	// event is emitted inside newClientRequest, so overrides apply to it.
+	// A nil emitter defaults to beholder.GlobalEmitter(), which resolves the
+	// global client at emit time, matching the previous direct global read.
+	cfg := clientRequestConfig{}
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+	if cfg.emitter == nil {
+		cfg.emitter = beholder.GlobalEmitter()
+	}
+	if cfg.durable == nil {
+		cfg.durable = durableemitter.GlobalEmit
+	}
+	return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, types.MethodExecute, rawRequest, workflowExecutionID, req.Metadata.ReferenceID, capMethodName, signers, minResponsesToAggregate, cfg)
 }
 
 var (
@@ -143,8 +181,9 @@ func requiredConfirmations(f uint8, minResponsesToAggregate uint32) int {
 
 func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string, remoteCapabilityInfo commoncap.CapabilityInfo,
 	localDonInfo commoncap.DON, dispatcher types.Dispatcher, requestTimeout time.Duration,
-	methodType string, rawRequest []byte, workflowExecutionID string, stepRef string, capMethodName string,
+	methodType string, rawRequest []byte, workflowExecutionID, stepRef, capMethodName string,
 	signers [][]byte, minResponsesToAggregate uint32,
+	cfg clientRequestConfig,
 ) (*ClientRequest, error) {
 	remoteCapabilityDonInfo := remoteCapabilityInfo.DON
 	if remoteCapabilityDonInfo == nil {
@@ -154,7 +193,7 @@ func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string,
 	peerIDToTransmissionDelay := getPeerIDToTransmissionDelays(remoteCapabilityDonInfo.Members)
 
 	// send schedule through beholder for single execution performance tracking
-	if err := emitTransmissionScheduleEvent(ctx,
+	if err := emitTransmissionScheduleEvent(ctx, cfg,
 		workflowExecutionID,
 		requestID,
 		remoteCapabilityInfo.ID,
@@ -267,7 +306,9 @@ func getPeerIDToTransmissionDelays(donPeerIDs []p2ptypes.PeerID) map[p2ptypes.Pe
 	return peerIDToTransmissionDelay
 }
 
-func emitTransmissionScheduleEvent(ctx context.Context, workflowExecutionID, transmissionID, capabilityID, stepRef string, peerIDToTransmissionDelay map[p2ptypes.PeerID]time.Duration) error {
+func emitTransmissionScheduleEvent(ctx context.Context, cfg clientRequestConfig,
+	workflowExecutionID, transmissionID, capabilityID, stepRef string, peerIDToTransmissionDelay map[p2ptypes.PeerID]time.Duration,
+) error {
 	// map of peer IDs to their transmission delays in milliseconds
 	peerDelaysMap := make(map[string]int64, len(peerIDToTransmissionDelay))
 	for peerID, delay := range peerIDToTransmissionDelay {
@@ -291,14 +332,14 @@ func emitTransmissionScheduleEvent(ctx context.Context, workflowExecutionID, tra
 
 	// emit transmission schedule event to track which nodes are successful when called to emit
 	entity := fmt.Sprintf("%s.%s", TransmissionEventProtoPkg, TransmissionEventEntity)
-	if err = beholder.GetEmitter().Emit(ctx, b,
+	if err = cfg.emitter.Emit(ctx, b,
 		"beholder_data_schema", TransmissionEventSchema, // required
 		"beholder_domain", "platform", // required
 		"beholder_entity", entity); err != nil { // required
 		return err
 	}
 
-	err = durableemitter.GlobalEmit(ctx, b, "source", "platform", "type", entity)
+	err = cfg.durable(ctx, b, "source", "platform", "type", entity)
 	if err != nil && !errors.Is(err, durableemitter.ErrNotInitialized) {
 		return err
 	}

core/capabilities/remote/executable/request/client_request_test.go

@@ -627,8 +627,9 @@ func Test_ClientRequest_MessageValidation(t *testing.T) {
 		})
 	})
 
-	t.Run("Executes full schedule", func(t *testing.T) { //nolint:paralleltest // beholdertest.NewObserver is not thread-safe
-		beholderTester := beholdertest.NewObserver(t)
+	t.Run("Executes full schedule", func(t *testing.T) {
+		t.Parallel()
+		rec := beholdertest.NewRecorder()
 		lggr, obs := logger.TestObserved(t, zapcore.DebugLevel)
 
 		capPeers, capDonInfo, capInfo := capabilityDon(t, 3, 1)
@@ -652,6 +653,7 @@ func Test_ClientRequest_MessageValidation(t *testing.T) {
 			"",
 			nil,
 			0,
+			request.WithEmitter(rec),
 		)
 		require.NoError(t, err)
 		defer req.Cancel(errors.New("test end"))
@@ -699,10 +701,10 @@ func Test_ClientRequest_MessageValidation(t *testing.T) {
 		}
 
 		// Verify the TransmissionsScheduledEvent data
-		assert.Equal(t, 1, beholderTester.Len(t, "beholder_entity", fmt.Sprintf("%v.%v", request.TransmissionEventProtoPkg, request.TransmissionEventEntity)))
+		assert.Equal(t, 1, rec.Len(t, "beholder_entity", fmt.Sprintf("%v.%v", request.TransmissionEventProtoPkg, request.TransmissionEventEntity)))
 
 		// Get the messages for the transmission event
-		messages := beholderTester.Messages(t, "beholder_entity", fmt.Sprintf("%v.%v", request.TransmissionEventProtoPkg, request.TransmissionEventEntity))
+		messages := rec.Messages(t, "beholder_entity", fmt.Sprintf("%v.%v", request.TransmissionEventProtoPkg, request.TransmissionEventEntity))
 		assert.Len(t, messages, 1)
 
 		// Unmarshal the message to verify its contents

core/services/cre/cre.go

@@ -469,6 +469,7 @@ func (s *Services) newRegistrySyncer(
 		opts.LimitsFactory,
 		shardingEnabled,
 		shardIndex,
+		capabilities.WithDurableEmit(opts.DurableEmit),
 	)
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("could not create workflow launcher: %w", err)