diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index f3eb61d309f..d71aea8fa08 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). 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 } diff --git a/core/capabilities/remote/executable/client.go b/core/capabilities/remote/executable/client.go index ab9e1ec5904..1d095da9107 100644 --- a/core/capabilities/remote/executable/client.go +++ b/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,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) + +// 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, @@ -75,6 +92,10 @@ 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) + } + 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) } diff --git a/core/capabilities/remote/executable/client_test.go b/core/capabilities/remote/executable/client_test.go index f65207e3d4a..80ee650915a 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,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) +} diff --git a/core/capabilities/remote/executable/request/client_request.go b/core/capabilities/remote/executable/request/client_request.go index 933e05daf46..5986e22fedf 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), @@ -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 } 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 diff --git a/core/services/cre/cre.go b/core/services/cre/cre.go index e8442a9feda..d289960947d 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, + capabilities.WithDurableEmit(opts.DurableEmit), ) if err != nil { return nil, nil, nil, fmt.Errorf("could not create workflow launcher: %w", err)