From 59a27a453c217748447a310463ff2868d17e6d59 Mon Sep 17 00:00:00 2001 From: Adam Hamrick Date: Mon, 12 Oct 2026 13:26:14 -0400 Subject: [PATCH 1/3] beholder: inject emitter in remote executable ClientRequest (Phase 3) - ClientRequest: emitter + durable fields; ClientRequestOption with WithEmitter/WithDurableEmit (symmetry with workflows/events); NewClientExecuteRequest takes options, defaults nil emitters to the late-bound globals before the transmission schedule event is emitted - client: emitter field + ClientOption/WithEmitter on NewClient (default global); Execute passes its emitter to each request; launcher unchanged - tests: 'Executes full schedule' uses a per-test recorder and t.Parallel; last NewObserver in core/ removed --- core/capabilities/remote/executable/client.go | 36 ++++++++++-- .../executable/request/client_request.go | 57 ++++++++++++++++--- .../executable/request/client_request_test.go | 10 ++-- 3 files changed, 85 insertions(+), 18 deletions(-) diff --git a/core/capabilities/remote/executable/client.go b/core/capabilities/remote/executable/client.go index ab9e1ec5904..6ecb48758de 100644 --- a/core/capabilities/remote/executable/client.go +++ b/core/capabilities/remote/executable/client.go @@ -8,6 +8,7 @@ import ( "sync/atomic" "time" + "github.com/smartcontractkit/chainlink-common/pkg/beholder" commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/logger" @@ -32,6 +33,10 @@ type client struct { cfg atomic.Pointer[dynamicConfig] lggr logger.Logger + // emitter receives transmission schedule events emitted by requests + // created through this client. Defaults to beholder.GlobalEmitter(). + emitter beholder.Emitter + 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,8 +73,18 @@ 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) + +// WithEmitter sets the beholder emitter used for transmission schedule events +// emitted by requests created through this client. The default is +// beholder.GlobalEmitter(), which resolves the global client at emit time. +func WithEmitter(e beholder.Emitter) ClientOption { + return func(c *client) { c.emitter = e } +} + +func NewClient(capabilityID, capMethodName string, dispatcher types.Dispatcher, lggr logger.Logger, opts ...ClientOption) *client { + c := &client{ capabilityID: capabilityID, capMethodName: capMethodName, dispatcher: dispatcher, @@ -75,6 +92,13 @@ func NewClient(capabilityID string, capMethodName string, dispatcher types.Dispa requestIDToCallerRequest: make(map[string]*request.ClientRequest), stopCh: make(services.StopChan), } + for _, opt := range opts { + opt(c) + } + if c.emitter == nil { + c.emitter = beholder.GlobalEmitter() + } + return c } // SetConfig sets the remote capability configuration dynamically @@ -235,7 +259,7 @@ 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.WithEmitter(c.emitter)) if err != nil { return commoncap.CapabilityResponse{}, fmt.Errorf("failed to create client request: %w", err) } diff --git a/core/capabilities/remote/executable/request/client_request.go b/core/capabilities/remote/executable/request/client_request.go index 933e05daf46..75343cdf64a 100644 --- a/core/capabilities/remote/executable/request/client_request.go +++ b/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), @@ -97,6 +97,26 @@ type ClientRequest struct { respSent bool mux sync.Mutex wg *sync.WaitGroup + + // emitter receives the transmission schedule event; durable receives the + // durable copy. Both default to the late-bound globals. + emitter beholder.Emitter + durable func(ctx context.Context, body []byte, attrKVs ...any) error +} + +// ClientRequestOption configures a ClientRequest. +type ClientRequestOption func(*ClientRequest) + +// 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 *ClientRequest) { 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 *ClientRequest) { c.durable = fn } } // NewClientExecuteRequest creates a new client request for a remote executable capability. @@ -104,7 +124,7 @@ type ClientRequest struct { 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 +141,23 @@ 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 emitters 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. + optsReq := &ClientRequest{} + for _, opt := range opts { + opt(optsReq) + } + emitter, durableEmit := optsReq.emitter, optsReq.durable + if emitter == nil { + emitter = beholder.GlobalEmitter() + } + if durableEmit == nil { + durableEmit = durableemitter.GlobalEmit + } + return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, types.MethodExecute, rawRequest, workflowExecutionID, req.Metadata.ReferenceID, capMethodName, signers, minResponsesToAggregate, emitter, durableEmit) } var ( @@ -143,8 +179,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, + emitter beholder.Emitter, durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error, ) (*ClientRequest, error) { remoteCapabilityDonInfo := remoteCapabilityInfo.DON if remoteCapabilityDonInfo == nil { @@ -154,7 +191,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, emitter, durableEmit, workflowExecutionID, requestID, remoteCapabilityInfo.ID, @@ -252,6 +289,8 @@ func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string, signers: signers, workflowExecutionID: workflowExecutionID, referenceID: stepRef, + emitter: emitter, + durable: durableEmit, }, nil } @@ -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, emitter beholder.Emitter, durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error, + 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 = 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 = durableEmit(ctx, b, "source", "platform", "type", entity) if err != nil && !errors.Is(err, durableemitter.ErrNotInitialized) { return err } diff --git a/core/capabilities/remote/executable/request/client_request_test.go b/core/capabilities/remote/executable/request/client_request_test.go index 17442fdba6c..f07b076ec61 100644 --- a/core/capabilities/remote/executable/request/client_request_test.go +++ b/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 From 50f2f688a208996e238c8eef2ec3fa01292cc6e5 Mon Sep 17 00:00:00 2001 From: Adam Hamrick Date: Wed, 21 Oct 2026 19:54:32 -0400 Subject: [PATCH 2/3] beholder: inject durable emitter in remote executable client Transmission schedule events still went through durableemitter.GlobalEmit because the client never accepted a durable func. - executable.ClientOption gains WithDurableEmit; Execute forwards it to request.WithDurableEmit - launcher.NewLauncher takes the func and gives it to every executable client it creates; cre passes the same Opts.DurableEmit used by the workflow event emitter - test asserts the schedule reaches the injected durable sink --- core/capabilities/launcher.go | 8 +++- .../launcher_offchain_method_configs_test.go | 10 ++--- core/capabilities/launcher_test.go | 37 +++++++-------- core/capabilities/remote/executable/client.go | 14 +++++- .../remote/executable/client_test.go | 45 +++++++++++++++++++ core/services/cre/cre.go | 1 + 6 files changed, 90 insertions(+), 25 deletions(-) diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index f3eb61d309f..5a8181e6b49 100644 --- a/core/capabilities/launcher.go +++ b/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). Empty 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. @@ -125,6 +129,7 @@ func NewLauncher( limitsFactory limits.Factory, shardingEnabled bool, shardIndex uint16, + durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error, ) (*launcher, error) { if don2donSharedPeer == nil { return nil, errors.New("don2donSharedPeer is required") @@ -179,6 +184,7 @@ func NewLauncher( propagateOrgIDGate: propagateOrgIDGate, shardingEnabled: shardingEnabled, shardIndex: shardIndex, + durableEmit: durableEmit, stopCh: make(services.StopChan), }, nil } @@ -736,7 +742,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 } diff --git a/core/capabilities/launcher_offchain_method_configs_test.go b/core/capabilities/launcher_offchain_method_configs_test.go index 1109c2d0e0a..1949efed262 100644 --- a/core/capabilities/launcher_offchain_method_configs_test.go +++ b/core/capabilities/launcher_offchain_method_configs_test.go @@ -142,7 +142,7 @@ func TestOffchainMethodConfigs_GateOff_Unchanged(t *testing.T) { }, }) - launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) launcher.SetOffchainRegistry(gc, false) // gate OFF require.NoError(t, launcher.Start(t.Context())) @@ -215,7 +215,7 @@ func TestOffchainMethodConfigs_GateOn_Overrides(t *testing.T) { }, }) - launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) launcher.SetOffchainRegistry(gc, true) // gate ON require.NoError(t, launcher.Start(t.Context())) @@ -302,7 +302,7 @@ func TestOffchainMethodConfigs_NoThrashAcrossPasses(t *testing.T) { }, }) - launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) launcher.SetOffchainRegistry(gc, true) // gate ON require.NoError(t, launcher.Start(t.Context())) @@ -369,7 +369,7 @@ func TestOffchainMethodConfigs_ServedMethodsReplaceAndRestore(t *testing.T) { dispatcher := remoteMocks.NewDispatcher(t) peer := mocks.NewSharedPeer(t) peer.On("ID").Return(capNodes[0]) - launcher, err := NewLauncher(lggr, peer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(lggr, peer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) t.Cleanup(func() { require.NoError(t, launcher.Close()) }) @@ -408,7 +408,7 @@ func TestOffchainMethodConfigs_LocalOnlySkipsRemoteShims(t *testing.T) { lggr := logger.Test(t) dispatcher := remoteMocks.NewDispatcher(t) peer := mocks.NewSharedPeer(t) - launcher, err := NewLauncher(lggr, peer, nil, dispatcher, regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(lggr, peer, nil, dispatcher, regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) method := &capabilitiespb.CapabilityMethodConfig{RemoteConfig: &capabilitiespb.CapabilityMethodConfig_RemoteTriggerConfig{RemoteTriggerConfig: &capabilitiespb.RemoteTriggerConfig{}}} don := regpkg.DON{ID: 1, Name: "don-1", CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{capID: {Config: onchainMethodCfg(t, map[string]*capabilitiespb.CapabilityMethodConfig{"Trigger": method})}}} diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index acc433cc527..fb0a480f5aa 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -111,7 +111,7 @@ func TestLauncher(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -148,7 +148,7 @@ func TestSyncer_IgnoresCapabilitiesForPrivateDON(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -194,7 +194,7 @@ func TestLauncher_DonPairsToUpdate(t *testing.T) { addCapabilityToDON(localRegistry, uint32(capDONID), fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil) addDON(localRegistry, uint32(mixedDONID), uint32(0), uint8(1), true, true, capabilityDonNodes[2:3], nil, 1, [][32]byte{triggerCapID}) addCapabilityToDON(localRegistry, uint32(mixedDONID), fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Times(3) @@ -277,7 +277,7 @@ func TestLauncher_DonPairsToUpdate_SkipsDifferentFamilies(t *testing.T) { addDON(localRegistry, capDONZoneBID, uint32(0), uint8(1), true, false, capabilityDonNodesZoneB, []string{"zone-b"}, 1, [][32]byte{triggerCapID}) addCapabilityToDON(localRegistry, capDONZoneBID, fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Once() @@ -350,7 +350,7 @@ func TestLauncher_ShardedCapabilityRoutingByFamily(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -406,7 +406,7 @@ func TestLauncher_DonPairsToUpdate_ShardedFamilies(t *testing.T) { addDON(localRegistry, sharedCapDONID, uint32(0), uint8(1), true, false, sharedCapNodes, []string{"zone-a"}, 1, [][32]byte{capIDHash}) addCapabilityToDON(localRegistry, sharedCapDONID, fullTargetID, capabilities.CapabilityTypeTarget, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Once() @@ -438,7 +438,7 @@ func TestLauncher_DonPairsToUpdate_CapShardPairsOnlyWithWorkflowShard(t *testing addDON(localRegistry, capShard0ID, uint32(0), uint8(1), true, false, capShard0Nodes, []string{"zone-a_shard-0"}, 1, [][32]byte{RandomUTF8BytesWord()}) addCapabilityToDON(localRegistry, capShard0ID, "write-chain_evm_1@1.0.0", capabilities.CapabilityTypeTarget, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Once() @@ -477,7 +477,7 @@ func TestLauncher_DonPairsToUpdate_BootstrapConnectsIsolatedCapShard(t *testing. addDON(localRegistry, capShard1ID, uint32(0), uint8(1), true, false, capShard1Nodes, []string{"zone-a_shard-1"}, 1, [][32]byte{RandomUTF8BytesWord()}) addCapabilityToDON(localRegistry, capShard1ID, "write-chain_evm_1@1.0.0", capabilities.CapabilityTypeTarget, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) // bootstrapPID is a member of no DON and therefore of no shard family. @@ -589,7 +589,7 @@ func TestLauncher_V2CapabilitiesAddViaCombinedClient(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) launcher.p2pStreamConfig = customStreamConfig @@ -692,7 +692,7 @@ func TestLauncher_OnNewRegistry_PrunesRemovedRemoteCapability(t *testing.T) { dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) servicetest.Run(t, launcher) @@ -775,7 +775,7 @@ func TestLauncher_OnNewRegistry_HandlesCapabilityDONReassignment(t *testing.T) { dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) servicetest.Run(t, launcher) @@ -863,7 +863,7 @@ func TestLauncher_OnNewRegistry_PrunesRemovedServedCapability(t *testing.T) { dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -988,7 +988,7 @@ func TestLauncher_V2CapabilitiesExposeRemotely(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1116,7 +1116,7 @@ func TestLauncher_OnNewRegistry_CallsLocalCapabilityManagerReconcile(t *testing. dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) launcher.SetLocalCapabilityManager(mockLCM) @@ -1157,7 +1157,7 @@ func TestLauncher_OnNewRegistry_NilLocalCapabilityManager(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1192,7 +1192,7 @@ func TestLauncher_OffchainRegistryUpdate(t *testing.T) { return nil }, } - l, err := NewLauncher(lggr, sharedPeer, nil, remoteMocks.NewDispatcher(t), regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0) + l, err := NewLauncher(lggr, sharedPeer, nil, remoteMocks.NewDispatcher(t), regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0, nil) require.NoError(t, err) gc := globalconfig.New() l.SetLocalCapabilityManager(lcm) @@ -1295,7 +1295,7 @@ func TestLauncher_OnNewRegistry_FiltersRemoteCapabilityDONsByFamily(t *testing.T dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1430,7 +1430,7 @@ func TestLauncher_OnNewRegistry_FiltersRemoteWorkflowDONsByFamily(t *testing.T) dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, + false, 0, nil, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1494,6 +1494,7 @@ func TestLauncher_ShardIdentityFromConfig(t *testing.T) { limits.Factory{}, shardingEnabled, shardIndex, + nil, ) require.NoError(t, err) return l diff --git a/core/capabilities/remote/executable/client.go b/core/capabilities/remote/executable/client.go index 6ecb48758de..7c0e928dbf6 100644 --- a/core/capabilities/remote/executable/client.go +++ b/core/capabilities/remote/executable/client.go @@ -37,6 +37,10 @@ type client struct { // created through this client. Defaults to beholder.GlobalEmitter(). emitter beholder.Emitter + // durableEmit persists the same transmission schedule events. Empty 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 @@ -83,6 +87,13 @@ func WithEmitter(e beholder.Emitter) ClientOption { return func(c *client) { c.emitter = e } } +// WithDurableEmit sets the durable emit func for transmission schedule events +// emitted by requests created through this client. Empty 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, @@ -259,7 +270,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, request.WithEmitter(c.emitter)) + cfg.requestTimeout, c.capMethodName, cfg.signers, cfg.minResponsesToAggregate, + request.WithEmitter(c.emitter), request.WithDurableEmit(c.durableEmit)) if err != nil { return commoncap.CapabilityResponse{}, fmt.Errorf("failed to create client request: %w", err) } diff --git a/core/capabilities/remote/executable/client_test.go b/core/capabilities/remote/executable/client_test.go index f65207e3d4a..344866c31a7 100644 --- a/core/capabilities/remote/executable/client_test.go +++ b/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,47 @@ 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)) + + // 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) +} diff --git a/core/services/cre/cre.go b/core/services/cre/cre.go index e8442a9feda..505dfa2f35f 100644 --- a/core/services/cre/cre.go +++ b/core/services/cre/cre.go @@ -469,6 +469,7 @@ func (s *Services) newRegistrySyncer( opts.LimitsFactory, shardingEnabled, shardIndex, + opts.DurableEmit, ) if err != nil { return nil, nil, nil, fmt.Errorf("could not create workflow launcher: %w", err) From 4cd3fb7c9e258b1f412a2a7b3cd997743260ec40 Mon Sep 17 00:00:00 2001 From: Adam Hamrick Date: Thu, 22 Oct 2026 13:12:29 -0400 Subject: [PATCH 3/3] fix(emitter): simplify --- core/capabilities/launcher.go | 22 +++++++--- .../launcher_offchain_method_configs_test.go | 10 ++--- core/capabilities/launcher_test.go | 37 ++++++++-------- core/capabilities/remote/executable/client.go | 24 +++------- .../remote/executable/client_test.go | 1 + .../executable/request/client_request.go | 44 +++++++++---------- core/services/cre/cre.go | 2 +- 7 files changed, 69 insertions(+), 71 deletions(-) diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index 5a8181e6b49..d71aea8fa08 100644 --- a/core/capabilities/launcher.go +++ b/core/capabilities/launcher.go @@ -60,7 +60,7 @@ type launcher struct { localCapMgr localcapmgr.LocalCapabilityManager // durableEmit persists events emitted by remote executable clients (e.g. the - // transmission schedule). Empty falls back to durableemitter.GlobalEmit. + // 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, @@ -117,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( @@ -129,7 +138,7 @@ func NewLauncher( limitsFactory limits.Factory, shardingEnabled bool, shardIndex uint16, - durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error, + opts ...LauncherOption, ) (*launcher, error) { if don2donSharedPeer == nil { return nil, errors.New("don2donSharedPeer is required") @@ -164,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{ @@ -184,9 +193,12 @@ func NewLauncher( propagateOrgIDGate: propagateOrgIDGate, shardingEnabled: shardingEnabled, shardIndex: shardIndex, - durableEmit: durableEmit, stopCh: make(services.StopChan), - }, nil + } + for _, opt := range opts { + opt(l) + } + return l, nil } func (w *launcher) publicDONs( diff --git a/core/capabilities/launcher_offchain_method_configs_test.go b/core/capabilities/launcher_offchain_method_configs_test.go index 1949efed262..1109c2d0e0a 100644 --- a/core/capabilities/launcher_offchain_method_configs_test.go +++ b/core/capabilities/launcher_offchain_method_configs_test.go @@ -142,7 +142,7 @@ func TestOffchainMethodConfigs_GateOff_Unchanged(t *testing.T) { }, }) - launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) launcher.SetOffchainRegistry(gc, false) // gate OFF require.NoError(t, launcher.Start(t.Context())) @@ -215,7 +215,7 @@ func TestOffchainMethodConfigs_GateOn_Overrides(t *testing.T) { }, }) - launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) launcher.SetOffchainRegistry(gc, true) // gate ON require.NoError(t, launcher.Start(t.Context())) @@ -302,7 +302,7 @@ func TestOffchainMethodConfigs_NoThrashAcrossPasses(t *testing.T) { }, }) - launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(lggr, sharedPeer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) launcher.SetOffchainRegistry(gc, true) // gate ON require.NoError(t, launcher.Start(t.Context())) @@ -369,7 +369,7 @@ func TestOffchainMethodConfigs_ServedMethodsReplaceAndRestore(t *testing.T) { dispatcher := remoteMocks.NewDispatcher(t) peer := mocks.NewSharedPeer(t) peer.On("ID").Return(capNodes[0]) - launcher, err := NewLauncher(lggr, peer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(lggr, peer, nil, dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) t.Cleanup(func() { require.NoError(t, launcher.Close()) }) @@ -408,7 +408,7 @@ func TestOffchainMethodConfigs_LocalOnlySkipsRemoteShims(t *testing.T) { lggr := logger.Test(t) dispatcher := remoteMocks.NewDispatcher(t) peer := mocks.NewSharedPeer(t) - launcher, err := NewLauncher(lggr, peer, nil, dispatcher, regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(lggr, peer, nil, dispatcher, regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) method := &capabilitiespb.CapabilityMethodConfig{RemoteConfig: &capabilitiespb.CapabilityMethodConfig_RemoteTriggerConfig{RemoteTriggerConfig: &capabilitiespb.RemoteTriggerConfig{}}} don := regpkg.DON{ID: 1, Name: "don-1", CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{capID: {Config: onchainMethodCfg(t, map[string]*capabilitiespb.CapabilityMethodConfig{"Trigger": method})}}} diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index fb0a480f5aa..acc433cc527 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -111,7 +111,7 @@ func TestLauncher(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -148,7 +148,7 @@ func TestSyncer_IgnoresCapabilitiesForPrivateDON(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -194,7 +194,7 @@ func TestLauncher_DonPairsToUpdate(t *testing.T) { addCapabilityToDON(localRegistry, uint32(capDONID), fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil) addDON(localRegistry, uint32(mixedDONID), uint32(0), uint8(1), true, true, capabilityDonNodes[2:3], nil, 1, [][32]byte{triggerCapID}) addCapabilityToDON(localRegistry, uint32(mixedDONID), fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Times(3) @@ -277,7 +277,7 @@ func TestLauncher_DonPairsToUpdate_SkipsDifferentFamilies(t *testing.T) { addDON(localRegistry, capDONZoneBID, uint32(0), uint8(1), true, false, capabilityDonNodesZoneB, []string{"zone-b"}, 1, [][32]byte{triggerCapID}) addCapabilityToDON(localRegistry, capDONZoneBID, fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Once() @@ -350,7 +350,7 @@ func TestLauncher_ShardedCapabilityRoutingByFamily(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -406,7 +406,7 @@ func TestLauncher_DonPairsToUpdate_ShardedFamilies(t *testing.T) { addDON(localRegistry, sharedCapDONID, uint32(0), uint8(1), true, false, sharedCapNodes, []string{"zone-a"}, 1, [][32]byte{capIDHash}) addCapabilityToDON(localRegistry, sharedCapDONID, fullTargetID, capabilities.CapabilityTypeTarget, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Once() @@ -438,7 +438,7 @@ func TestLauncher_DonPairsToUpdate_CapShardPairsOnlyWithWorkflowShard(t *testing addDON(localRegistry, capShard0ID, uint32(0), uint8(1), true, false, capShard0Nodes, []string{"zone-a_shard-0"}, 1, [][32]byte{RandomUTF8BytesWord()}) addCapabilityToDON(localRegistry, capShard0ID, "write-chain_evm_1@1.0.0", capabilities.CapabilityTypeTarget, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) sharedPeer.On("IsBootstrap").Return(false).Once() @@ -477,7 +477,7 @@ func TestLauncher_DonPairsToUpdate_BootstrapConnectsIsolatedCapShard(t *testing. addDON(localRegistry, capShard1ID, uint32(0), uint8(1), true, false, capShard1Nodes, []string{"zone-a_shard-1"}, 1, [][32]byte{RandomUTF8BytesWord()}) addCapabilityToDON(localRegistry, capShard1ID, "write-chain_evm_1@1.0.0", capabilities.CapabilityTypeTarget, nil) - launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + launcher, err := NewLauncher(logger.Test(t), sharedPeer, nil, dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) // bootstrapPID is a member of no DON and therefore of no shard family. @@ -589,7 +589,7 @@ func TestLauncher_V2CapabilitiesAddViaCombinedClient(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) launcher.p2pStreamConfig = customStreamConfig @@ -692,7 +692,7 @@ func TestLauncher_OnNewRegistry_PrunesRemovedRemoteCapability(t *testing.T) { dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) servicetest.Run(t, launcher) @@ -775,7 +775,7 @@ func TestLauncher_OnNewRegistry_HandlesCapabilityDONReassignment(t *testing.T) { dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) servicetest.Run(t, launcher) @@ -863,7 +863,7 @@ func TestLauncher_OnNewRegistry_PrunesRemovedServedCapability(t *testing.T) { dispatcher, reg, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -988,7 +988,7 @@ func TestLauncher_V2CapabilitiesExposeRemotely(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1116,7 +1116,7 @@ func TestLauncher_OnNewRegistry_CallsLocalCapabilityManagerReconcile(t *testing. dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) launcher.SetLocalCapabilityManager(mockLCM) @@ -1157,7 +1157,7 @@ func TestLauncher_OnNewRegistry_NilLocalCapabilityManager(t *testing.T) { dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1192,7 +1192,7 @@ func TestLauncher_OffchainRegistryUpdate(t *testing.T) { return nil }, } - l, err := NewLauncher(lggr, sharedPeer, nil, remoteMocks.NewDispatcher(t), regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0, nil) + l, err := NewLauncher(lggr, sharedPeer, nil, remoteMocks.NewDispatcher(t), regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0) require.NoError(t, err) gc := globalconfig.New() l.SetLocalCapabilityManager(lcm) @@ -1295,7 +1295,7 @@ func TestLauncher_OnNewRegistry_FiltersRemoteCapabilityDONsByFamily(t *testing.T dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1430,7 +1430,7 @@ func TestLauncher_OnNewRegistry_FiltersRemoteWorkflowDONsByFamily(t *testing.T) dispatcher, registry, &mockDonNotifier{}, limits.Factory{}, - false, 0, nil, + false, 0, ) require.NoError(t, err) require.NoError(t, launcher.Start(t.Context())) @@ -1494,7 +1494,6 @@ func TestLauncher_ShardIdentityFromConfig(t *testing.T) { limits.Factory{}, shardingEnabled, shardIndex, - nil, ) require.NoError(t, err) return l diff --git a/core/capabilities/remote/executable/client.go b/core/capabilities/remote/executable/client.go index 7c0e928dbf6..1d095da9107 100644 --- a/core/capabilities/remote/executable/client.go +++ b/core/capabilities/remote/executable/client.go @@ -8,7 +8,6 @@ import ( "sync/atomic" "time" - "github.com/smartcontractkit/chainlink-common/pkg/beholder" commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/logger" @@ -33,12 +32,9 @@ type client struct { cfg atomic.Pointer[dynamicConfig] lggr logger.Logger - // emitter receives transmission schedule events emitted by requests - // created through this client. Defaults to beholder.GlobalEmitter(). - emitter beholder.Emitter - - // durableEmit persists the same transmission schedule events. Empty falls - // back to durableemitter.GlobalEmit in the request constructor. + // 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 @@ -80,15 +76,8 @@ var ( // ClientOption configures a client. type ClientOption func(*client) -// WithEmitter sets the beholder emitter used for transmission schedule events -// emitted by requests created through this client. The default is -// beholder.GlobalEmitter(), which resolves the global client at emit time. -func WithEmitter(e beholder.Emitter) ClientOption { - return func(c *client) { c.emitter = e } -} - // WithDurableEmit sets the durable emit func for transmission schedule events -// emitted by requests created through this client. Empty keeps the default, +// 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 } @@ -106,9 +95,6 @@ func NewClient(capabilityID, capMethodName string, dispatcher types.Dispatcher, for _, opt := range opts { opt(c) } - if c.emitter == nil { - c.emitter = beholder.GlobalEmitter() - } return c } @@ -271,7 +257,7 @@ 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, - request.WithEmitter(c.emitter), request.WithDurableEmit(c.durableEmit)) + request.WithDurableEmit(c.durableEmit)) if err != nil { return commoncap.CapabilityResponse{}, fmt.Errorf("failed to create client request: %w", err) } diff --git a/core/capabilities/remote/executable/client_test.go b/core/capabilities/remote/executable/client_test.go index 344866c31a7..80ee650915a 100644 --- a/core/capabilities/remote/executable/client_test.go +++ b/core/capabilities/remote/executable/client_test.go @@ -561,6 +561,7 @@ func Test_Client_Execute_InjectsDurableEmit(t *testing.T) { 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. diff --git a/core/capabilities/remote/executable/request/client_request.go b/core/capabilities/remote/executable/request/client_request.go index 75343cdf64a..5986e22fedf 100644 --- a/core/capabilities/remote/executable/request/client_request.go +++ b/core/capabilities/remote/executable/request/client_request.go @@ -97,26 +97,29 @@ type ClientRequest struct { respSent bool mux sync.Mutex wg *sync.WaitGroup +} - // emitter receives the transmission schedule event; durable receives the - // durable copy. Both default to the late-bound globals. +// 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 a ClientRequest. -type ClientRequestOption func(*ClientRequest) +// 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 *ClientRequest) { c.emitter = e } + 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 *ClientRequest) { c.durable = fn } + return func(c *clientRequestConfig) { c.durable = fn } } // NewClientExecuteRequest creates a new client request for a remote executable capability. @@ -142,22 +145,21 @@ func NewClientExecuteRequest(ctx context.Context, lggr logger.Logger, req common lggr = logger.With(lggr, "requestId", requestID) // cap ID and method name included in the parent logger - // Apply options and default the emitters before the transmission schedule + // 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. - optsReq := &ClientRequest{} + cfg := clientRequestConfig{} for _, opt := range opts { - opt(optsReq) + opt(&cfg) } - emitter, durableEmit := optsReq.emitter, optsReq.durable - if emitter == nil { - emitter = beholder.GlobalEmitter() + if cfg.emitter == nil { + cfg.emitter = beholder.GlobalEmitter() } - if durableEmit == nil { - durableEmit = durableemitter.GlobalEmit + 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, emitter, durableEmit) + return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, types.MethodExecute, rawRequest, workflowExecutionID, req.Metadata.ReferenceID, capMethodName, signers, minResponsesToAggregate, cfg) } var ( @@ -181,7 +183,7 @@ func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string, localDonInfo commoncap.DON, dispatcher types.Dispatcher, requestTimeout time.Duration, methodType string, rawRequest []byte, workflowExecutionID, stepRef, capMethodName string, signers [][]byte, minResponsesToAggregate uint32, - emitter beholder.Emitter, durableEmit func(ctx context.Context, body []byte, attrKVs ...any) error, + cfg clientRequestConfig, ) (*ClientRequest, error) { remoteCapabilityDonInfo := remoteCapabilityInfo.DON if remoteCapabilityDonInfo == nil { @@ -191,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, emitter, durableEmit, + if err := emitTransmissionScheduleEvent(ctx, cfg, workflowExecutionID, requestID, remoteCapabilityInfo.ID, @@ -289,8 +291,6 @@ func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string, signers: signers, workflowExecutionID: workflowExecutionID, referenceID: stepRef, - emitter: emitter, - durable: durableEmit, }, nil } @@ -306,7 +306,7 @@ func getPeerIDToTransmissionDelays(donPeerIDs []p2ptypes.PeerID) map[p2ptypes.Pe return peerIDToTransmissionDelay } -func emitTransmissionScheduleEvent(ctx context.Context, emitter beholder.Emitter, durableEmit func(ctx context.Context, body []byte, attrKVs ...any) 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 @@ -332,14 +332,14 @@ func emitTransmissionScheduleEvent(ctx context.Context, emitter beholder.Emitter // emit transmission schedule event to track which nodes are successful when called to emit entity := fmt.Sprintf("%s.%s", TransmissionEventProtoPkg, TransmissionEventEntity) - if err = emitter.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 = durableEmit(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 } diff --git a/core/services/cre/cre.go b/core/services/cre/cre.go index 505dfa2f35f..d289960947d 100644 --- a/core/services/cre/cre.go +++ b/core/services/cre/cre.go @@ -469,7 +469,7 @@ func (s *Services) newRegistrySyncer( opts.LimitsFactory, shardingEnabled, shardIndex, - opts.DurableEmit, + capabilities.WithDurableEmit(opts.DurableEmit), ) if err != nil { return nil, nil, nil, fmt.Errorf("could not create workflow launcher: %w", err)