From c015da815f61533f63a0e7138d444293be9d4e9d Mon Sep 17 00:00:00 2001 From: Matthew Pendrey Date: Thu, 14 Nov 2024 21:26:59 +0000 Subject: [PATCH] Action Remote Shims Register/Unregister (#15232) * execute capability shims * shims * common bump * review comment * common bump * step ref * build --- core/capabilities/launcher.go | 58 ++- core/capabilities/launcher_test.go | 7 +- .../remote/{target => executable}/client.go | 147 ++++--- .../remote/executable/client_test.go | 383 ++++++++++++++++++ .../{target => executable}/endtoend_test.go | 263 +++++++++--- .../request/client_request.go | 120 ++++-- .../request/client_request_test.go | 189 ++++++++- .../request/server_request.go | 104 +++-- .../request/server_request_test.go | 155 +++++-- .../remote/{target => executable}/server.go | 40 +- .../{target => executable}/server_test.go | 166 +++++++- .../capabilities/remote/target/client_test.go | 316 --------------- core/capabilities/remote/types/types.go | 10 +- .../local_target_capability_test.go | 8 +- .../capabilities/transmission/transmission.go | 25 +- .../transmission/transmission_test.go | 14 +- core/scripts/go.mod | 2 +- core/scripts/go.sum | 4 +- core/services/workflows/engine.go | 4 +- deployment/go.mod | 2 +- deployment/go.sum | 4 +- go.mod | 2 +- go.sum | 4 +- integration-tests/go.mod | 2 +- integration-tests/go.sum | 4 +- integration-tests/load/go.mod | 2 +- integration-tests/load/go.sum | 4 +- 27 files changed, 1438 insertions(+), 601 deletions(-) rename core/capabilities/remote/{target => executable}/client.go (51%) create mode 100644 core/capabilities/remote/executable/client_test.go rename core/capabilities/remote/{target => executable}/endtoend_test.go (50%) rename core/capabilities/remote/{target => executable}/request/client_request.go (53%) rename core/capabilities/remote/{target => executable}/request/client_request_test.go (61%) rename core/capabilities/remote/{target => executable}/request/server_request.go (64%) rename core/capabilities/remote/{target => executable}/request/server_request_test.go (55%) rename core/capabilities/remote/{target => executable}/server.go (83%) rename core/capabilities/remote/{target => executable}/server_test.go (58%) delete mode 100644 core/capabilities/remote/target/client_test.go diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index 8deb42fdd68..e75f2ebbc8f 100644 --- a/core/capabilities/launcher.go +++ b/core/capabilities/launcher.go @@ -19,7 +19,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/values" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable" remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" "github.com/smartcontractkit/chainlink/v2/core/capabilities/streams" "github.com/smartcontractkit/chainlink/v2/core/logger" @@ -294,12 +294,26 @@ func (w *launcher) addRemoteCapabilities(ctx context.Context, myDON registrysync return fmt.Errorf("failed to add trigger shim: %w", err) } case capabilities.CapabilityTypeAction: - w.lggr.Warn("no remote client configured for capability type action, skipping configuration") + newActionFn := func(info capabilities.CapabilityInfo) (capabilityService, error) { + client := executable.NewClient( + info, + myDON.DON, + w.dispatcher, + defaultTargetRequestTimeout, + w.lggr, + ) + return client, nil + } + + err := w.addToRegistryAndSetDispatcher(ctx, capability, remoteDON, newActionFn) + if err != nil { + return fmt.Errorf("failed to add action shim: %w", err) + } case capabilities.CapabilityTypeConsensus: w.lggr.Warn("no remote client configured for capability type consensus, skipping configuration") case capabilities.CapabilityTypeTarget: newTargetFn := func(info capabilities.CapabilityInfo) (capabilityService, error) { - client := target.NewClient( + client := executable.NewClient( info, myDON.DON, w.dispatcher, @@ -419,7 +433,34 @@ func (w *launcher) exposeCapabilities(ctx context.Context, myPeerID p2ptypes.Pee // continue attempting other capabilities } case capabilities.CapabilityTypeAction: - w.lggr.Warn("no remote client configured for capability type action, skipping configuration") + newActionServer := func(cap capabilities.BaseCapability, info capabilities.CapabilityInfo) (remotetypes.ReceiverService, error) { + actionCapability, ok := (cap).(capabilities.ActionCapability) + if !ok { + return nil, errors.New("capability does not implement ActionCapability") + } + + remoteConfig := &capabilities.RemoteExecutableConfig{} + if capabilityConfig.RemoteTargetConfig != nil { + remoteConfig.RequestHashExcludedAttributes = capabilityConfig.RemoteTargetConfig.RequestHashExcludedAttributes + } + + return executable.NewServer( + capabilityConfig.RemoteExecutableConfig, + myPeerID, + actionCapability, + info, + don.DON, + idsToDONs, + w.dispatcher, + defaultTargetRequestTimeout, + w.lggr, + ), nil + } + + err = w.addReceiver(ctx, capability, don, newActionServer) + if err != nil { + return fmt.Errorf("failed to add action server-side receiver: %w", err) + } case capabilities.CapabilityTypeConsensus: w.lggr.Warn("no remote client configured for capability type consensus, skipping configuration") case capabilities.CapabilityTypeTarget: @@ -429,8 +470,13 @@ func (w *launcher) exposeCapabilities(ctx context.Context, myPeerID p2ptypes.Pee return nil, errors.New("capability does not implement TargetCapability") } - return target.NewServer( - capabilityConfig.RemoteTargetConfig, + remoteConfig := &capabilities.RemoteExecutableConfig{} + if capabilityConfig.RemoteTargetConfig != nil { + remoteConfig.RequestHashExcludedAttributes = capabilityConfig.RemoteTargetConfig.RequestHashExcludedAttributes + } + + return executable.NewServer( + remoteConfig, myPeerID, targetCapability, info, diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index 3ebed639cb0..013463bfdbb 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -199,7 +199,7 @@ func TestLauncher(t *testing.T) { ) dispatcher.On("SetReceiver", fullTriggerCapID, dID, mock.AnythingOfType("*remote.triggerPublisher")).Return(nil) - dispatcher.On("SetReceiver", fullTargetID, dID, mock.AnythingOfType("*target.server")).Return(nil) + dispatcher.On("SetReceiver", fullTargetID, dID, mock.AnythingOfType("*executable.server")).Return(nil) err = launcher.Launch(ctx, state) require.NoError(t, err) @@ -603,7 +603,8 @@ func TestLauncher_RemoteTriggerModeAggregatorShim(t *testing.T) { ) dispatcher.On("SetReceiver", fullTriggerCapID, capDonID, mock.AnythingOfType("*remote.triggerSubscriber")).Return(nil) - dispatcher.On("SetReceiver", fullTargetID, capDonID, mock.AnythingOfType("*target.client")).Return(nil) + dispatcher.On("SetReceiver", fullTargetID, capDonID, mock.AnythingOfType("*executable.client")).Return(nil) + dispatcher.On("Ready").Return(nil).Maybe() awaitRegistrationMessageCh := make(chan struct{}) dispatcher.On("Send", mock.Anything, mock.Anything).Return(nil).Run(func(args mock.Arguments) { select { @@ -919,7 +920,7 @@ func TestLauncher_WiresUpClientsForPublicWorkflowDON(t *testing.T) { ) dispatcher.On("SetReceiver", fullTriggerCapID, capDonID, mock.AnythingOfType("*remote.triggerSubscriber")).Return(nil) - dispatcher.On("SetReceiver", fullTargetID, capDonID, mock.AnythingOfType("*target.client")).Return(nil) + dispatcher.On("SetReceiver", fullTargetID, capDonID, mock.AnythingOfType("*executable.client")).Return(nil) err = launcher.Launch(ctx, state) require.NoError(t, err) diff --git a/core/capabilities/remote/target/client.go b/core/capabilities/remote/executable/client.go similarity index 51% rename from core/capabilities/remote/target/client.go rename to core/capabilities/remote/executable/client.go index 8249c40bb01..08c773cdb86 100644 --- a/core/capabilities/remote/target/client.go +++ b/core/capabilities/remote/executable/client.go @@ -1,4 +1,4 @@ -package target +package executable import ( "context" @@ -8,15 +8,15 @@ import ( "time" commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target/request" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable/request" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/validation" "github.com/smartcontractkit/chainlink/v2/core/logger" ) -// client is a shim for remote target capabilities. +// client is a shim for remote executable capabilities. // It translates between capability API calls and network messages. // Its responsibilities are: // 1. Transmit capability requests to remote nodes according to a transmission schedule @@ -31,25 +31,25 @@ type client struct { dispatcher types.Dispatcher requestTimeout time.Duration - messageIDToCallerRequest map[string]*request.ClientRequest + requestIDToCallerRequest map[string]*request.ClientRequest mutex sync.Mutex stopCh services.StopChan wg sync.WaitGroup } -var _ commoncap.TargetCapability = &client{} +var _ commoncap.ExecutableCapability = &client{} var _ types.Receiver = &client{} var _ services.Service = &client{} func NewClient(remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, dispatcher types.Dispatcher, requestTimeout time.Duration, lggr logger.Logger) *client { return &client{ - lggr: lggr.Named("TargetClient"), + lggr: lggr.Named("ExecutableCapabilityClient"), remoteCapabilityInfo: remoteCapabilityInfo, localDONInfo: localDonInfo, dispatcher: dispatcher, requestTimeout: requestTimeout, - messageIDToCallerRequest: make(map[string]*request.ClientRequest), + requestIDToCallerRequest: make(map[string]*request.ClientRequest), stopCh: make(services.StopChan), } } @@ -61,7 +61,13 @@ func (c *client) Start(ctx context.Context) error { defer c.wg.Done() c.checkForExpiredRequests() }() - c.lggr.Info("TargetClient started") + c.wg.Add(1) + go func() { + defer c.wg.Done() + c.checkDispatcherReady() + }() + + c.lggr.Info("ExecutableCapability Client started") return nil }) } @@ -71,11 +77,26 @@ func (c *client) Close() error { close(c.stopCh) c.cancelAllRequests(errors.New("client closed")) c.wg.Wait() - c.lggr.Info("TargetClient closed") + c.lggr.Info("ExecutableCapability closed") return nil }) } +func (c *client) checkDispatcherReady() { + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + for { + select { + case <-c.stopCh: + return + case <-ticker.C: + if err := c.dispatcher.Ready(); err != nil { + c.cancelAllRequests(fmt.Errorf("dispatcher not ready: %w", err)) + } + } + } +} + func (c *client) checkForExpiredRequests() { ticker := time.NewTicker(c.requestTimeout) defer ticker.Stop() @@ -93,10 +114,15 @@ func (c *client) expireRequests() { c.mutex.Lock() defer c.mutex.Unlock() - for messageID, req := range c.messageIDToCallerRequest { + for messageID, req := range c.requestIDToCallerRequest { if req.Expired() { req.Cancel(errors.New("request expired")) - delete(c.messageIDToCallerRequest, messageID) + delete(c.requestIDToCallerRequest, messageID) + } + + if c.dispatcher.Ready() != nil { + c.cancelAllRequests(errors.New("dispatcher not ready")) + return } } } @@ -104,7 +130,7 @@ func (c *client) expireRequests() { func (c *client) cancelAllRequests(err error) { c.mutex.Lock() defer c.mutex.Unlock() - for _, req := range c.messageIDToCallerRequest { + for _, req := range c.requestIDToCallerRequest { req.Cancel(err) } } @@ -113,49 +139,80 @@ func (c *client) Info(ctx context.Context) (commoncap.CapabilityInfo, error) { return c.remoteCapabilityInfo, nil } -func (c *client) RegisterToWorkflow(ctx context.Context, request commoncap.RegisterToWorkflowRequest) error { - // do nothing - return nil -} +func (c *client) RegisterToWorkflow(ctx context.Context, registerRequest commoncap.RegisterToWorkflowRequest) error { + req, err := request.NewClientRegisterToWorkflowRequest(ctx, c.lggr, registerRequest, c.remoteCapabilityInfo, c.localDONInfo, c.dispatcher, + c.requestTimeout) + + if err != nil { + return fmt.Errorf("failed to create client request: %w", err) + } + + if err = c.sendRequest(req); err != nil { + return fmt.Errorf("failed to send request: %w", err) + } -func (c *client) UnregisterFromWorkflow(ctx context.Context, request commoncap.UnregisterFromWorkflowRequest) error { - // do nothing + resp := <-req.ResponseChan() + if resp.Err != nil { + return fmt.Errorf("error executing request: %w", resp.Err) + } return nil } -func (c *client) Execute(ctx context.Context, capReq commoncap.CapabilityRequest) (commoncap.CapabilityResponse, error) { - req, err := c.executeRequest(ctx, capReq) +func (c *client) UnregisterFromWorkflow(ctx context.Context, unregisterRequest commoncap.UnregisterFromWorkflowRequest) error { + req, err := request.NewClientUnregisterFromWorkflowRequest(ctx, c.lggr, unregisterRequest, c.remoteCapabilityInfo, + c.localDONInfo, c.dispatcher, c.requestTimeout) + if err != nil { - return commoncap.CapabilityResponse{}, fmt.Errorf("failed to execute request: %w", err) + return fmt.Errorf("failed to create client request: %w", err) + } + + if err = c.sendRequest(req); err != nil { + return fmt.Errorf("failed to send request: %w", err) } resp := <-req.ResponseChan() - return resp.CapabilityResponse, resp.Err + if resp.Err != nil { + return fmt.Errorf("error executing request: %w", resp.Err) + } + return nil } -func (c *client) executeRequest(ctx context.Context, capReq commoncap.CapabilityRequest) (*request.ClientRequest, error) { - c.mutex.Lock() - defer c.mutex.Unlock() - - messageID, err := GetMessageIDForRequest(capReq) +func (c *client) Execute(ctx context.Context, capReq commoncap.CapabilityRequest) (commoncap.CapabilityResponse, error) { + req, err := request.NewClientExecuteRequest(ctx, c.lggr, capReq, c.remoteCapabilityInfo, c.localDONInfo, c.dispatcher, + c.requestTimeout) if err != nil { - return nil, fmt.Errorf("failed to get message ID for request: %w", err) + return commoncap.CapabilityResponse{}, fmt.Errorf("failed to create client request: %w", err) } - c.lggr.Debugw("executing remote target", "messageID", messageID) + if err = c.sendRequest(req); err != nil { + return commoncap.CapabilityResponse{}, fmt.Errorf("failed to send request: %w", err) + } - if _, ok := c.messageIDToCallerRequest[messageID]; ok { - return nil, fmt.Errorf("request for message ID %s already exists", messageID) + resp := <-req.ResponseChan() + if resp.Err != nil { + return commoncap.CapabilityResponse{}, fmt.Errorf("error executing request: %w", resp.Err) } - req, err := request.NewClientRequest(ctx, c.lggr, capReq, messageID, c.remoteCapabilityInfo, c.localDONInfo, c.dispatcher, - c.requestTimeout) + capabilityResponse, err := pb.UnmarshalCapabilityResponse(resp.Result) if err != nil { - return nil, fmt.Errorf("failed to create client request: %w", err) + return commoncap.CapabilityResponse{}, fmt.Errorf("failed to unmarshal capability response: %w", err) + } + + return capabilityResponse, nil +} + +func (c *client) sendRequest(req *request.ClientRequest) error { + c.mutex.Lock() + defer c.mutex.Unlock() + + c.lggr.Debugw("executing remote execute capability", "requestID", req.ID()) + + if _, ok := c.requestIDToCallerRequest[req.ID()]; ok { + return fmt.Errorf("request for ID %s already exists", req.ID()) } - c.messageIDToCallerRequest[messageID] = req - return req, nil + c.requestIDToCallerRequest[req.ID()] = req + return nil } func (c *client) Receive(ctx context.Context, msg *types.MessageBody) { @@ -168,9 +225,9 @@ func (c *client) Receive(ctx context.Context, msg *types.MessageBody) { return } - c.lggr.Debugw("Remote client target receiving message", "messageID", messageID) + c.lggr.Debugw("Remote client executable receiving message", "messageID", messageID) - req := c.messageIDToCallerRequest[messageID] + req := c.requestIDToCallerRequest[messageID] if req == nil { c.lggr.Warnw("received response for unknown message ID ", "messageID", messageID) return @@ -181,18 +238,6 @@ func (c *client) Receive(ctx context.Context, msg *types.MessageBody) { } } -func GetMessageIDForRequest(req commoncap.CapabilityRequest) (string, error) { - if err := validation.ValidateWorkflowOrExecutionID(req.Metadata.WorkflowID); err != nil { - return "", fmt.Errorf("workflow ID is invalid: %w", err) - } - - if err := validation.ValidateWorkflowOrExecutionID(req.Metadata.WorkflowExecutionID); err != nil { - return "", fmt.Errorf("workflow execution ID is invalid: %w", err) - } - - return req.Metadata.WorkflowID + req.Metadata.WorkflowExecutionID, nil -} - func (c *client) Ready() error { return nil } diff --git a/core/capabilities/remote/executable/client_test.go b/core/capabilities/remote/executable/client_test.go new file mode 100644 index 00000000000..5c4da350b9e --- /dev/null +++ b/core/capabilities/remote/executable/client_test.go @@ -0,0 +1,383 @@ +package executable_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" + "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" + "github.com/smartcontractkit/chainlink-common/pkg/values" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable" + remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/transmission" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" + "github.com/smartcontractkit/chainlink/v2/core/logger" + p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" +) + +const ( + stepReferenceID1 = "step1" + workflowID1 = "15c631d295ef5e32deb99a10ee6804bc4af13855687559d7ff6552ac6dbb2ce0" + workflowExecutionID1 = "95ef5e32deb99a10ee6804bc4af13855687559d7ff6552ac6dbb2ce0abbadeed" + workflowOwnerID = "0xAA" +) + +func Test_Client_DonTopologies(t *testing.T) { + ctx := testutils.Context(t) + + transmissionSchedule, err := values.NewMap(map[string]any{ + "schedule": transmission.Schedule_OneAtATime, + "deltaStage": "10ms", + }) + require.NoError(t, err) + + responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { + require.NoError(t, responseError) + mp, err := response.Value.Unwrap() + require.NoError(t, err) + assert.Equal(t, "aValue1", mp.(map[string]any)["response"].(string)) + } + + capability := &TestCapability{} + + responseTimeOut := 10 * time.Minute + + var methods []func(caller commoncap.ExecutableCapability) + + methods = append(methods, func(caller commoncap.ExecutableCapability) { + executeInputs, err := values.NewMap(map[string]any{"executeValue1": "aValue1"}) + require.NoError(t, err) + executeMethod(ctx, caller, transmissionSchedule, executeInputs, responseTest, t) + }) + + methods = append(methods, func(caller commoncap.ExecutableCapability) { + registerToWorkflowMethod(ctx, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.NoError(t, responseError) + }, t) + }) + + methods = append(methods, func(caller commoncap.ExecutableCapability) { + unregisterFromWorkflowMethod(ctx, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.NoError(t, responseError) + }, t) + }) + + for _, method := range methods { + testClient(t, 1, responseTimeOut, 1, 0, + capability, method) + + testClient(t, 10, responseTimeOut, 1, 0, + capability, method) + + testClient(t, 1, responseTimeOut, 10, 3, + capability, method) + + testClient(t, 10, responseTimeOut, 10, 3, + capability, method) + + testClient(t, 10, responseTimeOut, 10, 9, + capability, method) + } +} + +func Test_Client_TransmissionSchedules(t *testing.T) { + ctx := testutils.Context(t) + + responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { + require.NoError(t, responseError) + mp, err := response.Value.Unwrap() + require.NoError(t, err) + assert.Equal(t, "aValue1", mp.(map[string]any)["response"].(string)) + } + + capability := &TestCapability{} + + responseTimeOut := 10 * time.Minute + + transmissionSchedule, err := values.NewMap(map[string]any{ + "schedule": transmission.Schedule_OneAtATime, + "deltaStage": "10ms", + }) + require.NoError(t, err) + + testClient(t, 1, responseTimeOut, 1, 0, + capability, func(caller commoncap.ExecutableCapability) { + executeInputs, err2 := values.NewMap(map[string]any{"executeValue1": "aValue1"}) + require.NoError(t, err2) + executeMethod(ctx, caller, transmissionSchedule, executeInputs, responseTest, t) + }) + testClient(t, 10, responseTimeOut, 10, 3, + capability, func(caller commoncap.ExecutableCapability) { + executeInputs, err2 := values.NewMap(map[string]any{"executeValue1": "aValue1"}) + require.NoError(t, err2) + executeMethod(ctx, caller, transmissionSchedule, executeInputs, responseTest, t) + }) + + transmissionSchedule, err = values.NewMap(map[string]any{ + "schedule": transmission.Schedule_AllAtOnce, + "deltaStage": "10ms", + }) + require.NoError(t, err) + + testClient(t, 1, responseTimeOut, 1, 0, + capability, func(caller commoncap.ExecutableCapability) { + executeInputs, err := values.NewMap(map[string]any{"executeValue1": "aValue1"}) + require.NoError(t, err) + executeMethod(ctx, caller, transmissionSchedule, executeInputs, responseTest, t) + }) + testClient(t, 10, responseTimeOut, 10, 3, + capability, func(caller commoncap.ExecutableCapability) { + executeInputs, err := values.NewMap(map[string]any{"executeValue1": "aValue1"}) + require.NoError(t, err) + executeMethod(ctx, caller, transmissionSchedule, executeInputs, responseTest, t) + }) +} + +func Test_Client_TimesOutIfInsufficientCapabilityPeerResponses(t *testing.T) { + ctx := testutils.Context(t) + + responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { + assert.Error(t, responseError) + } + + capability := &TestCapability{} + + transmissionSchedule, err := values.NewMap(map[string]any{ + "schedule": transmission.Schedule_AllAtOnce, + "deltaStage": "10ms", + }) + require.NoError(t, err) + + // number of capability peers is less than F + 1 + + testClient(t, 10, 1*time.Second, 10, 11, + capability, + func(caller commoncap.ExecutableCapability) { + executeInputs, err := values.NewMap(map[string]any{"executeValue1": "aValue1"}) + require.NoError(t, err) + executeMethod(ctx, caller, transmissionSchedule, executeInputs, responseTest, t) + }) +} + +func testClient(t *testing.T, numWorkflowPeers int, workflowNodeResponseTimeout time.Duration, + numCapabilityPeers int, capabilityDonF uint8, underlying commoncap.ExecutableCapability, + method func(caller commoncap.ExecutableCapability)) { + lggr := logger.TestLogger(t) + + capabilityPeers := make([]p2ptypes.PeerID, numCapabilityPeers) + for i := 0; i < numCapabilityPeers; i++ { + capabilityPeers[i] = NewP2PPeerID(t) + } + + capDonInfo := commoncap.DON{ + ID: 1, + Members: capabilityPeers, + F: capabilityDonF, + } + + capInfo := commoncap.CapabilityInfo{ + ID: "cap_id@1.0.0", + CapabilityType: commoncap.CapabilityTypeTrigger, + Description: "Remote Executable Capability", + DON: &capDonInfo, + } + + workflowPeers := make([]p2ptypes.PeerID, numWorkflowPeers) + for i := 0; i < numWorkflowPeers; i++ { + workflowPeers[i] = NewP2PPeerID(t) + } + + workflowDonInfo := commoncap.DON{ + Members: workflowPeers, + ID: 2, + } + + broker := newTestAsyncMessageBroker(t, 100) + + receivers := make([]remotetypes.Receiver, numCapabilityPeers) + for i := 0; i < numCapabilityPeers; i++ { + capabilityDispatcher := broker.NewDispatcherForNode(capabilityPeers[i]) + receiver := newTestServer(capabilityPeers[i], capabilityDispatcher, workflowDonInfo, underlying) + broker.RegisterReceiverNode(capabilityPeers[i], receiver) + receivers[i] = receiver + } + + callers := make([]commoncap.ExecutableCapability, numWorkflowPeers) + + for i := 0; i < numWorkflowPeers; i++ { + workflowPeerDispatcher := broker.NewDispatcherForNode(workflowPeers[i]) + caller := executable.NewClient(capInfo, workflowDonInfo, workflowPeerDispatcher, workflowNodeResponseTimeout, lggr) + servicetest.Run(t, caller) + broker.RegisterReceiverNode(workflowPeers[i], caller) + callers[i] = caller + } + + servicetest.Run(t, broker) + + wg := &sync.WaitGroup{} + wg.Add(len(callers)) + + // Fire off all the requests + for _, caller := range callers { + go func(caller commoncap.ExecutableCapability) { + defer wg.Done() + method(caller) + }(caller) + } + + wg.Wait() +} + +func registerToWorkflowMethod(ctx context.Context, caller commoncap.ExecutableCapability, transmissionSchedule *values.Map, + responseTest func(t *testing.T, responseError error), t *testing.T) { + err := caller.RegisterToWorkflow(ctx, commoncap.RegisterToWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + Config: transmissionSchedule, + }) + + responseTest(t, err) +} + +func unregisterFromWorkflowMethod(ctx context.Context, caller commoncap.ExecutableCapability, transmissionSchedule *values.Map, + responseTest func(t *testing.T, responseError error), t *testing.T) { + err := caller.UnregisterFromWorkflow(ctx, commoncap.UnregisterFromWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + Config: transmissionSchedule, + }) + + responseTest(t, err) +} + +func executeMethod(ctx context.Context, caller commoncap.ExecutableCapability, transmissionSchedule *values.Map, + executeInputs *values.Map, responseTest func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error), t *testing.T) { + responseCh, err := caller.Execute(ctx, + commoncap.CapabilityRequest{ + Metadata: commoncap.RequestMetadata{ + WorkflowID: workflowID1, + WorkflowExecutionID: workflowExecutionID1, + WorkflowOwner: workflowOwnerID, + }, + Config: transmissionSchedule, + Inputs: executeInputs, + }) + + responseTest(t, responseCh, err) +} + +// Simple client that only responds once it has received a message from each workflow peer +type clientTestServer struct { + peerID p2ptypes.PeerID + dispatcher remotetypes.Dispatcher + workflowDonInfo commoncap.DON + messageIDToSenders map[string]map[p2ptypes.PeerID]bool + + executableCapability commoncap.ExecutableCapability + + mux sync.Mutex +} + +func newTestServer(peerID p2ptypes.PeerID, dispatcher remotetypes.Dispatcher, workflowDonInfo commoncap.DON, + executableCapability commoncap.ExecutableCapability) *clientTestServer { + return &clientTestServer{ + dispatcher: dispatcher, + workflowDonInfo: workflowDonInfo, + peerID: peerID, + messageIDToSenders: make(map[string]map[p2ptypes.PeerID]bool), + executableCapability: executableCapability, + } +} + +func (t *clientTestServer) Receive(_ context.Context, msg *remotetypes.MessageBody) { + t.mux.Lock() + defer t.mux.Unlock() + + sender := toPeerID(msg.Sender) + messageID, err := executable.GetMessageID(msg) + if err != nil { + panic(err) + } + + if t.messageIDToSenders[messageID] == nil { + t.messageIDToSenders[messageID] = make(map[p2ptypes.PeerID]bool) + } + + sendersOfMessageID := t.messageIDToSenders[messageID] + if sendersOfMessageID[sender] { + panic("received duplicate message") + } + + sendersOfMessageID[sender] = true + + if len(t.messageIDToSenders[messageID]) == len(t.workflowDonInfo.Members) { + switch msg.Method { + case remotetypes.MethodExecute: + capabilityRequest, err := pb.UnmarshalCapabilityRequest(msg.Payload) + if err != nil { + panic(err) + } + resp, responseErr := t.executableCapability.Execute(context.Background(), capabilityRequest) + payload, marshalErr := pb.MarshalCapabilityResponse(resp) + t.sendResponse(messageID, responseErr, payload, marshalErr) + + case remotetypes.MethodRegisterToWorkflow: + registerRequest, err := pb.UnmarshalRegisterToWorkflowRequest(msg.Payload) + if err != nil { + panic(err) + } + responseErr := t.executableCapability.RegisterToWorkflow(context.Background(), registerRequest) + t.sendResponse(messageID, responseErr, nil, nil) + case remotetypes.MethodUnregisterFromWorkflow: + unregisterRequest, err := pb.UnmarshalUnregisterFromWorkflowRequest(msg.Payload) + if err != nil { + panic(err) + } + responseErr := t.executableCapability.UnregisterFromWorkflow(context.Background(), unregisterRequest) + t.sendResponse(messageID, responseErr, nil, nil) + default: + panic("unknown method") + } + } +} + +func (t *clientTestServer) sendResponse(messageID string, responseErr error, + payload []byte, marshalErr error) { + for receiver := range t.messageIDToSenders[messageID] { + var responseMsg = &remotetypes.MessageBody{ + CapabilityId: "cap_id@1.0.0", + CapabilityDonId: 1, + CallerDonId: t.workflowDonInfo.ID, + Method: remotetypes.MethodExecute, + MessageId: []byte(messageID), + Sender: t.peerID[:], + Receiver: receiver[:], + } + + if responseErr != nil { + responseMsg.Error = remotetypes.Error_INTERNAL_ERROR + } else { + if marshalErr != nil { + panic(marshalErr) + } + responseMsg.Payload = payload + } + + err := t.dispatcher.Send(receiver, responseMsg) + if err != nil { + panic(err) + } + } +} diff --git a/core/capabilities/remote/target/endtoend_test.go b/core/capabilities/remote/executable/endtoend_test.go similarity index 50% rename from core/capabilities/remote/target/endtoend_test.go rename to core/capabilities/remote/executable/endtoend_test.go index 0cade4f7855..29f29ed9ee1 100644 --- a/core/capabilities/remote/target/endtoend_test.go +++ b/core/capabilities/remote/executable/endtoend_test.go @@ -1,4 +1,4 @@ -package target_test +package executable_test import ( "context" @@ -18,7 +18,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable" remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" "github.com/smartcontractkit/chainlink/v2/core/capabilities/transmission" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" @@ -26,7 +26,7 @@ import ( p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" ) -func Test_RemoteTargetCapability_InsufficientCapabilityResponses(t *testing.T) { +func Test_RemoteExecutableCapability_InsufficientCapabilityResponses(t *testing.T) { ctx := testutils.Context(t) responseTest := func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error) { @@ -41,10 +41,30 @@ func Test_RemoteTargetCapability_InsufficientCapabilityResponses(t *testing.T) { }) require.NoError(t, err) - testRemoteTarget(ctx, t, capability, 10, 9, 10*time.Millisecond, 10, 10, 10*time.Minute, transmissionSchedule, responseTest) + var methods []func(ctx context.Context, caller commoncap.ExecutableCapability) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, responseTest) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + registerWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.Error(t, responseError) + }) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + unregisterWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.Error(t, responseError) + }) + }) + + for _, method := range methods { + testRemoteExecutableCapability(ctx, t, capability, 10, 9, 10*time.Millisecond, 10, 10, 10*time.Minute, method) + } } -func Test_RemoteTargetCapability_InsufficientWorkflowRequests(t *testing.T) { +func Test_RemoteExecutableCapability_InsufficientWorkflowRequests(t *testing.T) { ctx := testutils.Context(t) responseTest := func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error) { @@ -61,10 +81,30 @@ func Test_RemoteTargetCapability_InsufficientWorkflowRequests(t *testing.T) { }) require.NoError(t, err) - testRemoteTarget(ctx, t, capability, 10, 10, 10*time.Millisecond, 10, 9, timeOut, transmissionSchedule, responseTest) + var methods []func(ctx context.Context, caller commoncap.ExecutableCapability) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, responseTest) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + registerWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.Error(t, responseError) + }) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + unregisterWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.Error(t, responseError) + }) + }) + + for _, method := range methods { + testRemoteExecutableCapability(ctx, t, capability, 10, 10, 10*time.Millisecond, 10, 9, timeOut, method) + } } -func Test_RemoteTargetCapability_TransmissionSchedules(t *testing.T) { +func Test_RemoteExecutableCapability_TransmissionSchedules(t *testing.T) { ctx := testutils.Context(t) responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { @@ -84,18 +124,24 @@ func Test_RemoteTargetCapability_TransmissionSchedules(t *testing.T) { capability := &TestCapability{} - testRemoteTarget(ctx, t, capability, 10, 9, timeOut, 10, 9, timeOut, transmissionSchedule, responseTest) + method := func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, responseTest) + } + testRemoteExecutableCapability(ctx, t, capability, 10, 9, timeOut, 10, 9, timeOut, method) transmissionSchedule, err = values.NewMap(map[string]any{ "schedule": transmission.Schedule_AllAtOnce, "deltaStage": "10ms", }) require.NoError(t, err) + method = func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, responseTest) + } - testRemoteTarget(ctx, t, capability, 10, 9, timeOut, 10, 9, timeOut, transmissionSchedule, responseTest) + testRemoteExecutableCapability(ctx, t, capability, 10, 9, timeOut, 10, 9, timeOut, method) } -func Test_RemoteTargetCapability_DonTopologies(t *testing.T) { +func Test_RemoteExecutionCapability_DonTopologies(t *testing.T) { ctx := testutils.Context(t) responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { @@ -115,26 +161,42 @@ func Test_RemoteTargetCapability_DonTopologies(t *testing.T) { capability := &TestCapability{} - // Test scenarios where the number of submissions is greater than or equal to F + 1 - testRemoteTarget(ctx, t, capability, 1, 0, timeOut, 1, 0, timeOut, transmissionSchedule, responseTest) - testRemoteTarget(ctx, t, capability, 4, 3, timeOut, 1, 0, timeOut, transmissionSchedule, responseTest) - testRemoteTarget(ctx, t, capability, 10, 3, timeOut, 1, 0, timeOut, transmissionSchedule, responseTest) + var methods []func(ctx context.Context, caller commoncap.ExecutableCapability) - testRemoteTarget(ctx, t, capability, 1, 0, timeOut, 1, 0, timeOut, transmissionSchedule, responseTest) - testRemoteTarget(ctx, t, capability, 1, 0, timeOut, 4, 3, timeOut, transmissionSchedule, responseTest) - testRemoteTarget(ctx, t, capability, 1, 0, timeOut, 10, 3, timeOut, transmissionSchedule, responseTest) + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, responseTest) + }) - testRemoteTarget(ctx, t, capability, 4, 3, timeOut, 4, 3, timeOut, transmissionSchedule, responseTest) - testRemoteTarget(ctx, t, capability, 10, 3, timeOut, 10, 3, timeOut, transmissionSchedule, responseTest) - testRemoteTarget(ctx, t, capability, 10, 9, timeOut, 10, 9, timeOut, transmissionSchedule, responseTest) -} + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + registerWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.NoError(t, responseError) + }) + }) -func Test_RemoteTargetCapability_CapabilityError(t *testing.T) { - ctx := testutils.Context(t) + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + unregisterWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + require.NoError(t, responseError) + }) + }) - responseTest := func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error) { - assert.Equal(t, "failed to execute capability: an error", responseError.Error()) + for _, method := range methods { + // Test scenarios where the number of submissions is greater than or equal to F + 1 + testRemoteExecutableCapability(ctx, t, capability, 1, 0, timeOut, 1, 0, timeOut, method) + testRemoteExecutableCapability(ctx, t, capability, 4, 3, timeOut, 1, 0, timeOut, method) + testRemoteExecutableCapability(ctx, t, capability, 10, 3, timeOut, 1, 0, timeOut, method) + + testRemoteExecutableCapability(ctx, t, capability, 1, 0, timeOut, 1, 0, timeOut, method) + testRemoteExecutableCapability(ctx, t, capability, 1, 0, timeOut, 4, 3, timeOut, method) + testRemoteExecutableCapability(ctx, t, capability, 1, 0, timeOut, 10, 3, timeOut, method) + + testRemoteExecutableCapability(ctx, t, capability, 4, 3, timeOut, 4, 3, timeOut, method) + testRemoteExecutableCapability(ctx, t, capability, 10, 3, timeOut, 10, 3, timeOut, method) + testRemoteExecutableCapability(ctx, t, capability, 10, 9, timeOut, 10, 9, timeOut, method) } +} + +func Test_RemoteExecutionCapability_CapabilityError(t *testing.T) { + ctx := testutils.Context(t) capability := &TestErrorCapability{} @@ -144,15 +206,33 @@ func Test_RemoteTargetCapability_CapabilityError(t *testing.T) { }) require.NoError(t, err) - testRemoteTarget(ctx, t, capability, 10, 9, 10*time.Minute, 10, 9, 10*time.Minute, transmissionSchedule, responseTest) -} + var methods []func(ctx context.Context, caller commoncap.ExecutableCapability) -func Test_RemoteTargetCapability_RandomCapabilityError(t *testing.T) { - ctx := testutils.Context(t) + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error) { + assert.Equal(t, "error executing request: failed to execute capability: an error", responseError.Error()) + }) + }) - responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { - assert.Equal(t, "request expired", responseError.Error()) + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + registerWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + assert.Equal(t, "error executing request: failed to register to workflow: an error", responseError.Error()) + }) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + unregisterWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + assert.Equal(t, "error executing request: failed to unregister from workflow: an error", responseError.Error()) + }) + }) + + for _, method := range methods { + testRemoteExecutableCapability(ctx, t, capability, 10, 9, 10*time.Minute, 10, 9, 10*time.Minute, method) } +} + +func Test_RemoteExecutableCapability_RandomCapabilityError(t *testing.T) { + ctx := testutils.Context(t) capability := &TestRandomErrorCapability{} @@ -162,12 +242,35 @@ func Test_RemoteTargetCapability_RandomCapabilityError(t *testing.T) { }) require.NoError(t, err) - testRemoteTarget(ctx, t, capability, 10, 9, 10*time.Millisecond, 10, 9, 10*time.Minute, transmissionSchedule, responseTest) + var methods []func(ctx context.Context, caller commoncap.ExecutableCapability) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + executeCapability(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error) { + assert.Equal(t, "error executing request: request expired", responseError.Error()) + }) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + registerWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + assert.Equal(t, "error executing request: request expired", responseError.Error()) + }) + }) + + methods = append(methods, func(ctx context.Context, caller commoncap.ExecutableCapability) { + unregisterWorkflow(ctx, t, caller, transmissionSchedule, func(t *testing.T, responseError error) { + assert.Equal(t, "error executing request: request expired", responseError.Error()) + }) + }) + + for _, method := range methods { + testRemoteExecutableCapability(ctx, t, capability, 10, 9, 10*time.Millisecond, 10, 9, 10*time.Minute, + method) + } } -func testRemoteTarget(ctx context.Context, t *testing.T, underlying commoncap.TargetCapability, numWorkflowPeers int, workflowDonF uint8, workflowNodeTimeout time.Duration, - numCapabilityPeers int, capabilityDonF uint8, capabilityNodeResponseTimeout time.Duration, transmissionSchedule *values.Map, - responseTest func(t *testing.T, response commoncap.CapabilityResponse, responseError error)) { +func testRemoteExecutableCapability(ctx context.Context, t *testing.T, underlying commoncap.ExecutableCapability, numWorkflowPeers int, workflowDonF uint8, workflowNodeTimeout time.Duration, + numCapabilityPeers int, capabilityDonF uint8, capabilityNodeResponseTimeout time.Duration, + method func(ctx context.Context, caller commoncap.ExecutableCapability)) { lggr := logger.TestLogger(t) capabilityPeers := make([]p2ptypes.PeerID, numCapabilityPeers) @@ -216,17 +319,17 @@ func testRemoteTarget(ctx context.Context, t *testing.T, underlying commoncap.Ta for i := 0; i < numCapabilityPeers; i++ { capabilityPeer := capabilityPeers[i] capabilityDispatcher := broker.NewDispatcherForNode(capabilityPeer) - capabilityNode := target.NewServer(&commoncap.RemoteTargetConfig{RequestHashExcludedAttributes: []string{}}, capabilityPeer, underlying, capInfo, capDonInfo, workflowDONs, capabilityDispatcher, + capabilityNode := executable.NewServer(&commoncap.RemoteExecutableConfig{RequestHashExcludedAttributes: []string{}}, capabilityPeer, underlying, capInfo, capDonInfo, workflowDONs, capabilityDispatcher, capabilityNodeResponseTimeout, lggr) servicetest.Run(t, capabilityNode) broker.RegisterReceiverNode(capabilityPeer, capabilityNode) capabilityNodes[i] = capabilityNode } - workflowNodes := make([]commoncap.TargetCapability, numWorkflowPeers) + workflowNodes := make([]commoncap.ExecutableCapability, numWorkflowPeers) for i := 0; i < numWorkflowPeers; i++ { workflowPeerDispatcher := broker.NewDispatcherForNode(workflowPeers[i]) - workflowNode := target.NewClient(capInfo, workflowDonInfo, workflowPeerDispatcher, workflowNodeTimeout, lggr) + workflowNode := executable.NewClient(capInfo, workflowDonInfo, workflowPeerDispatcher, workflowNodeTimeout, lggr) servicetest.Run(t, workflowNode) broker.RegisterReceiverNode(workflowPeers[i], workflowNode) workflowNodes[i] = workflowNode @@ -234,31 +337,13 @@ func testRemoteTarget(ctx context.Context, t *testing.T, underlying commoncap.Ta servicetest.Run(t, broker) - executeInputs, err := values.NewMap( - map[string]any{ - "executeValue1": "aValue1", - }, - ) - - require.NoError(t, err) - wg := &sync.WaitGroup{} wg.Add(len(workflowNodes)) for _, caller := range workflowNodes { - go func(caller commoncap.TargetCapability) { + go func(caller commoncap.ExecutableCapability) { defer wg.Done() - response, err := caller.Execute(ctx, - commoncap.CapabilityRequest{ - Metadata: commoncap.RequestMetadata{ - WorkflowID: workflowID1, - WorkflowExecutionID: workflowExecutionID1, - }, - Config: transmissionSchedule, - Inputs: executeInputs, - }) - - responseTest(t, response, err) + method(ctx, caller) }(caller) } @@ -413,6 +498,14 @@ func (t TestErrorCapability) Execute(ctx context.Context, request commoncap.Capa return commoncap.CapabilityResponse{}, errors.New("an error") } +func (t TestErrorCapability) RegisterToWorkflow(ctx context.Context, request commoncap.RegisterToWorkflowRequest) error { + return errors.New("an error") +} + +func (t TestErrorCapability) UnregisterFromWorkflow(ctx context.Context, request commoncap.UnregisterFromWorkflowRequest) error { + return errors.New("an error") +} + type TestRandomErrorCapability struct { abstractTestCapability } @@ -421,6 +514,14 @@ func (t TestRandomErrorCapability) Execute(ctx context.Context, request commonca return commoncap.CapabilityResponse{}, errors.New(uuid.New().String()) } +func (t TestRandomErrorCapability) RegisterToWorkflow(ctx context.Context, request commoncap.RegisterToWorkflowRequest) error { + return errors.New(uuid.New().String()) +} + +func (t TestRandomErrorCapability) UnregisterFromWorkflow(ctx context.Context, request commoncap.UnregisterFromWorkflowRequest) error { + return errors.New(uuid.New().String()) +} + func NewP2PPeerID(t *testing.T) p2ptypes.PeerID { id := p2ptypes.PeerID{} require.NoError(t, id.UnmarshalText([]byte(NewPeerID()))) @@ -442,3 +543,49 @@ func NewPeerID() string { func libp2pMagic() []byte { return []byte{0x00, 0x24, 0x08, 0x01, 0x12, 0x20} } + +func executeCapability(ctx context.Context, t *testing.T, caller commoncap.ExecutableCapability, transmissionSchedule *values.Map, responseTest func(t *testing.T, response commoncap.CapabilityResponse, responseError error)) { + executeInputs, err := values.NewMap( + map[string]any{ + "executeValue1": "aValue1", + }, + ) + require.NoError(t, err) + response, err := caller.Execute(ctx, + commoncap.CapabilityRequest{ + Metadata: commoncap.RequestMetadata{ + WorkflowID: workflowID1, + WorkflowExecutionID: workflowExecutionID1, + }, + Config: transmissionSchedule, + Inputs: executeInputs, + }) + + responseTest(t, response, err) +} + +func registerWorkflow(ctx context.Context, t *testing.T, caller commoncap.ExecutableCapability, transmissionSchedule *values.Map, responseTest func(t *testing.T, responseError error)) { + err := caller.RegisterToWorkflow(ctx, commoncap.RegisterToWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + Config: transmissionSchedule, + }) + + responseTest(t, err) +} + +func unregisterWorkflow(ctx context.Context, t *testing.T, caller commoncap.ExecutableCapability, transmissionSchedule *values.Map, responseTest func(t *testing.T, responseError error)) { + err := caller.UnregisterFromWorkflow(ctx, commoncap.UnregisterFromWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + Config: transmissionSchedule, + }) + + responseTest(t, err) +} diff --git a/core/capabilities/remote/target/request/client_request.go b/core/capabilities/remote/executable/request/client_request.go similarity index 53% rename from core/capabilities/remote/target/request/client_request.go rename to core/capabilities/remote/executable/request/client_request.go index 1a0d707146b..6b4b9e3a0cd 100644 --- a/core/capabilities/remote/target/request/client_request.go +++ b/core/capabilities/remote/executable/request/client_request.go @@ -12,6 +12,8 @@ import ( ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/validation" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" @@ -22,14 +24,15 @@ import ( p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" ) -type asyncCapabilityResponse struct { - capabilities.CapabilityResponse - Err error +type clientResponse struct { + Result []byte + Err error } type ClientRequest struct { + id string cancelFn context.CancelFunc - responseCh chan asyncCapabilityResponse + responseCh chan clientResponse createdAt time.Time responseIDCount map[[32]byte]int errorCount map[string]int @@ -45,26 +48,91 @@ type ClientRequest struct { wg *sync.WaitGroup } -func NewClientRequest(ctx context.Context, lggr logger.Logger, req commoncap.CapabilityRequest, messageID string, +func NewClientRegisterToWorkflowRequest(ctx context.Context, lggr logger.Logger, req commoncap.RegisterToWorkflowRequest, remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo capabilities.DON, dispatcher types.Dispatcher, requestTimeout time.Duration) (*ClientRequest, error) { - remoteCapabilityDonInfo := remoteCapabilityInfo.DON - if remoteCapabilityDonInfo == nil { - return nil, errors.New("remote capability info missing DON") + rawRequest, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb.RegisterToWorkflowRequestToProto(req)) + if err != nil { + return nil, fmt.Errorf("failed to marshal register to workflow request: %w", err) } - rawRequest, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb.CapabilityRequestToProto(req)) + workflowID := req.Metadata.WorkflowID + if err := validation.ValidateWorkflowOrExecutionID(workflowID); err != nil { + return nil, fmt.Errorf("workflow ID is invalid: %w", err) + } + + requestID := types.MethodRegisterToWorkflow + ":" + workflowID + + tc := transmission.TransmissionConfig{ + Schedule: transmission.Schedule_AllAtOnce, + DeltaStage: 0, + } + + return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, + tc, types.MethodRegisterToWorkflow, rawRequest) +} +func NewClientUnregisterFromWorkflowRequest(ctx context.Context, lggr logger.Logger, req commoncap.UnregisterFromWorkflowRequest, + remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo capabilities.DON, dispatcher types.Dispatcher, + requestTimeout time.Duration) (*ClientRequest, error) { + rawRequest, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb.UnregisterFromWorkflowRequestToProto(req)) + if err != nil { + return nil, fmt.Errorf("failed to marshal unregister from workflow request: %w", err) + } + + workflowID := req.Metadata.WorkflowID + if err := validation.ValidateWorkflowOrExecutionID(workflowID); err != nil { + return nil, fmt.Errorf("workflow ID is invalid: %w", err) + } + + requestID := types.MethodUnregisterFromWorkflow + ":" + workflowID + + tc := transmission.TransmissionConfig{ + Schedule: transmission.Schedule_AllAtOnce, + DeltaStage: 0, + } + + return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, + tc, types.MethodUnregisterFromWorkflow, rawRequest) +} + +func NewClientExecuteRequest(ctx context.Context, lggr logger.Logger, req commoncap.CapabilityRequest, + remoteCapabilityInfo commoncap.CapabilityInfo, localDonInfo capabilities.DON, dispatcher types.Dispatcher, + requestTimeout time.Duration) (*ClientRequest, error) { + rawRequest, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb.CapabilityRequestToProto(req)) if err != nil { return nil, fmt.Errorf("failed to marshal capability request: %w", err) } - peerIDToTransmissionDelay, err := transmission.GetPeerIDToTransmissionDelay(remoteCapabilityDonInfo.Members, req) + workflowExecutionID := req.Metadata.WorkflowExecutionID + if err = validation.ValidateWorkflowOrExecutionID(workflowExecutionID); err != nil { + return nil, fmt.Errorf("workflow execution ID is invalid: %w", err) + } + + requestID := types.MethodExecute + ":" + workflowExecutionID + + tc, err := transmission.ExtractTransmissionConfig(req.Config) + if err != nil { + return nil, fmt.Errorf("failed to extract transmission config from request: %w", err) + } + + return newClientRequest(ctx, lggr, requestID, remoteCapabilityInfo, localDonInfo, dispatcher, requestTimeout, tc, types.MethodExecute, rawRequest) +} + +func newClientRequest(ctx context.Context, lggr logger.Logger, requestID string, remoteCapabilityInfo commoncap.CapabilityInfo, + localDonInfo commoncap.DON, dispatcher types.Dispatcher, requestTimeout time.Duration, + tc transmission.TransmissionConfig, methodType string, rawRequest []byte) (*ClientRequest, error) { + remoteCapabilityDonInfo := remoteCapabilityInfo.DON + if remoteCapabilityDonInfo == nil { + return nil, errors.New("remote capability info missing DON") + } + + peerIDToTransmissionDelay, err := transmission.GetPeerIDToTransmissionDelaysForConfig(remoteCapabilityDonInfo.Members, requestID, tc) if err != nil { return nil, fmt.Errorf("failed to get peer ID to transmission delay: %w", err) } - lggr.Debugw("sending request to peers", "execID", req.Metadata.WorkflowExecutionID, "schedule", peerIDToTransmissionDelay) + lggr.Debugw("sending request to peers", "requestID", requestID, "schedule", peerIDToTransmissionDelay) responseReceived := make(map[p2ptypes.PeerID]bool) @@ -79,17 +147,17 @@ func NewClientRequest(ctx context.Context, lggr logger.Logger, req commoncap.Cap CapabilityId: remoteCapabilityInfo.ID, CapabilityDonId: remoteCapabilityDonInfo.ID, CallerDonId: localDonInfo.ID, - Method: types.MethodExecute, + Method: methodType, Payload: rawRequest, - MessageId: []byte(messageID), + MessageId: []byte(requestID), } select { case <-ctxWithCancel.Done(): - lggr.Debugw("context done, not sending request to peer", "execID", req.Metadata.WorkflowExecutionID, "peerID", peerID) + lggr.Debugw("context done, not sending request to peer", "requestID", requestID, "peerID", peerID) return case <-time.After(delay): - lggr.Debugw("sending request to peer", "execID", req.Metadata.WorkflowExecutionID, "peerID", peerID) + lggr.Debugw("sending request to peer", "requestID", requestID, "peerID", peerID) err := dispatcher.Send(peerID, message) if err != nil { lggr.Errorw("failed to send message", "peerID", peerID, "err", err) @@ -99,6 +167,7 @@ func NewClientRequest(ctx context.Context, lggr logger.Logger, req commoncap.Cap } return &ClientRequest{ + id: requestID, cancelFn: cancelFn, createdAt: time.Now(), requestTimeout: requestTimeout, @@ -106,13 +175,17 @@ func NewClientRequest(ctx context.Context, lggr logger.Logger, req commoncap.Cap responseIDCount: make(map[[32]byte]int), errorCount: make(map[string]int), responseReceived: responseReceived, - responseCh: make(chan asyncCapabilityResponse, 1), + responseCh: make(chan clientResponse, 1), wg: wg, lggr: lggr, }, nil } -func (c *ClientRequest) ResponseChan() <-chan asyncCapabilityResponse { +func (c *ClientRequest) ID() string { + return c.id +} + +func (c *ClientRequest) ResponseChan() <-chan clientResponse { return c.responseCh } @@ -126,7 +199,7 @@ func (c *ClientRequest) Cancel(err error) { c.mux.Lock() defer c.mux.Unlock() if !c.respSent { - c.sendResponse(asyncCapabilityResponse{Err: err}) + c.sendResponse(clientResponse{Err: err}) } } @@ -169,24 +242,19 @@ func (c *ClientRequest) OnMessage(_ context.Context, msg *types.MessageBody) err } if c.responseIDCount[responseID] == c.requiredIdenticalResponses { - capabilityResponse, err := pb.UnmarshalCapabilityResponse(msg.Payload) - if err != nil { - c.sendResponse(asyncCapabilityResponse{Err: fmt.Errorf("failed to unmarshal capability response: %w", err)}) - } else { - c.sendResponse(asyncCapabilityResponse{CapabilityResponse: commoncap.CapabilityResponse{Value: capabilityResponse.Value}}) - } + c.sendResponse(clientResponse{Result: msg.Payload}) } } else { c.lggr.Warnw("received error response", "error", remote.SanitizeLogString(msg.ErrorMsg)) c.errorCount[msg.ErrorMsg]++ if c.errorCount[msg.ErrorMsg] == c.requiredIdenticalResponses { - c.sendResponse(asyncCapabilityResponse{Err: errors.New(msg.ErrorMsg)}) + c.sendResponse(clientResponse{Err: errors.New(msg.ErrorMsg)}) } } return nil } -func (c *ClientRequest) sendResponse(response asyncCapabilityResponse) { +func (c *ClientRequest) sendResponse(response clientResponse) { c.responseCh <- response close(c.responseCh) c.respSent = true diff --git a/core/capabilities/remote/target/request/client_request_test.go b/core/capabilities/remote/executable/request/client_request_test.go similarity index 61% rename from core/capabilities/remote/target/request/client_request_test.go rename to core/capabilities/remote/executable/request/client_request_test.go index 095c73e8ad9..c46fd1363a0 100644 --- a/core/capabilities/remote/target/request/client_request_test.go +++ b/core/capabilities/remote/executable/request/client_request_test.go @@ -12,8 +12,8 @@ import ( commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target/request" + + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable/request" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" "github.com/smartcontractkit/chainlink/v2/core/capabilities/transmission" "github.com/smartcontractkit/chainlink/v2/core/logger" @@ -80,6 +80,15 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { Config: transmissionSchedule, } + registerToWorkflowRequest := commoncap.RegisterToWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + WorkflowOwner: "0xaa", + ReferenceID: "refID", + }, + Config: transmissionSchedule, + } + m, err := values.NewMap(map[string]any{"response": "response1"}) require.NoError(t, err) capabilityResponse := commoncap.CapabilityResponse{ @@ -89,9 +98,6 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { rawResponse, err := pb.MarshalCapabilityResponse(capabilityResponse) require.NoError(t, err) - messageID, err := target.GetMessageIDForRequest(capabilityRequest) - require.NoError(t, err) - msg := &types.MessageBody{ CapabilityId: capInfo.ID, CapabilityDonId: capDonInfo.ID, @@ -106,7 +112,7 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { defer cancel() dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} - request, err := request.NewClientRequest(ctx, lggr, capabilityRequest, messageID, capInfo, + request, err := request.NewClientExecuteRequest(ctx, lggr, capabilityRequest, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) defer request.Cancel(errors.New("test end")) @@ -149,7 +155,7 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { defer cancel() dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} - request, err := request.NewClientRequest(ctx, lggr, capabilityRequest, messageID, capInfo, + request, err := request.NewClientExecuteRequest(ctx, lggr, capabilityRequest, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) require.NoError(t, err) defer request.Cancel(errors.New("test end")) @@ -175,7 +181,7 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { defer cancel() dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} - request, err := request.NewClientRequest(ctx, lggr, capabilityRequest, messageID, capInfo, + request, err := request.NewClientExecuteRequest(ctx, lggr, capabilityRequest, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) require.NoError(t, err) defer request.Cancel(errors.New("test end")) @@ -198,7 +204,7 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { defer cancel() dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} - request, err := request.NewClientRequest(ctx, lggr, capabilityRequest, messageID, capInfo, + request, err := request.NewClientExecuteRequest(ctx, lggr, capabilityRequest, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) require.NoError(t, err) defer request.Cancel(errors.New("test end")) @@ -236,7 +242,7 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { defer cancel() dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} - request, err := request.NewClientRequest(ctx, lggr, capabilityRequest, messageID, capInfo, + request, err := request.NewClientExecuteRequest(ctx, lggr, capabilityRequest, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) require.NoError(t, err) defer request.Cancel(errors.New("test end")) @@ -281,12 +287,12 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { } }) - t.Run("Send second valid message", func(t *testing.T) { + t.Run("Execute Request", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} - request, err := request.NewClientRequest(ctx, lggr, capabilityRequest, messageID, capInfo, + request, err := request.NewClientExecuteRequest(ctx, lggr, capabilityRequest, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) require.NoError(t, err) defer request.Cancel(errors.New("test end")) @@ -304,10 +310,167 @@ func Test_ClientRequest_MessageValidation(t *testing.T) { require.NoError(t, err) response := <-request.ResponseChan() - resp := response.Value.Underlying["response"] + capResponse, err := pb.UnmarshalCapabilityResponse(response.Result) + require.NoError(t, err) + + resp := capResponse.Value.Underlying["response"] assert.Equal(t, resp, values.NewString("response1")) }) + + t.Run("Register To Workflow Request", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} + request, err := request.NewClientRegisterToWorkflowRequest(ctx, lggr, registerToWorkflowRequest, capInfo, + workflowDonInfo, dispatcher, 10*time.Minute) + require.NoError(t, err) + defer request.Cancel(errors.New("test end")) + + <-dispatcher.msgs + <-dispatcher.msgs + assert.Empty(t, dispatcher.msgs) + + msg := &types.MessageBody{ + CapabilityId: capInfo.ID, + CapabilityDonId: capDonInfo.ID, + CallerDonId: workflowDonInfo.ID, + Method: types.MethodRegisterToWorkflow, + Payload: nil, + MessageId: []byte("messageID"), + } + + msg.Sender = capabilityPeers[0][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + msg.Sender = capabilityPeers[1][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + response := <-request.ResponseChan() + require.Nil(t, response.Result) + require.NoError(t, response.Err) + }) + + t.Run("Register To Workflow Request with error", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} + request, err := request.NewClientRegisterToWorkflowRequest(ctx, lggr, registerToWorkflowRequest, capInfo, + workflowDonInfo, dispatcher, 10*time.Minute) + require.NoError(t, err) + defer request.Cancel(errors.New("test end")) + + <-dispatcher.msgs + <-dispatcher.msgs + assert.Empty(t, dispatcher.msgs) + + msg := &types.MessageBody{ + CapabilityId: capInfo.ID, + CapabilityDonId: capDonInfo.ID, + CallerDonId: workflowDonInfo.ID, + Method: types.MethodRegisterToWorkflow, + Payload: nil, + MessageId: []byte("messageID"), + Error: types.Error_INTERNAL_ERROR, + ErrorMsg: "an error", + } + + msg.Sender = capabilityPeers[0][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + msg.Sender = capabilityPeers[1][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + response := <-request.ResponseChan() + require.Nil(t, response.Result) + assert.Equal(t, "an error", response.Err.Error()) + }) + + t.Run("Unregister From Workflow Request", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} + request, err := request.NewClientUnregisterFromWorkflowRequest(ctx, lggr, commoncap.UnregisterFromWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + }, + }, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) + require.NoError(t, err) + defer request.Cancel(errors.New("test end")) + + <-dispatcher.msgs + <-dispatcher.msgs + assert.Empty(t, dispatcher.msgs) + + msg := &types.MessageBody{ + CapabilityId: capInfo.ID, + CapabilityDonId: capDonInfo.ID, + CallerDonId: workflowDonInfo.ID, + Method: types.MethodUnregisterFromWorkflow, + Payload: nil, + MessageId: []byte("messageID"), + } + + msg.Sender = capabilityPeers[0][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + msg.Sender = capabilityPeers[1][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + response := <-request.ResponseChan() + require.Nil(t, response.Result) + require.NoError(t, response.Err) + }) + + t.Run("Unregister From Workflow Request with error", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + dispatcher := &clientRequestTestDispatcher{msgs: make(chan *types.MessageBody, 100)} + request, err := request.NewClientUnregisterFromWorkflowRequest(ctx, lggr, commoncap.UnregisterFromWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + }, + }, capInfo, workflowDonInfo, dispatcher, 10*time.Minute) + require.NoError(t, err) + defer request.Cancel(errors.New("test end")) + + <-dispatcher.msgs + <-dispatcher.msgs + assert.Empty(t, dispatcher.msgs) + + msg := &types.MessageBody{ + CapabilityId: capInfo.ID, + CapabilityDonId: capDonInfo.ID, + CallerDonId: workflowDonInfo.ID, + Method: types.MethodUnregisterFromWorkflow, + Payload: nil, + MessageId: []byte("messageID"), + Error: types.Error_INTERNAL_ERROR, + ErrorMsg: "an error", + } + + msg.Sender = capabilityPeers[0][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + msg.Sender = capabilityPeers[1][:] + err = request.OnMessage(ctx, msg) + require.NoError(t, err) + + response := <-request.ResponseChan() + require.Nil(t, response.Result) + assert.Equal(t, "an error", response.Err.Error()) + }) } type clientRequestTestDispatcher struct { diff --git a/core/capabilities/remote/target/request/server_request.go b/core/capabilities/remote/executable/request/server_request.go similarity index 64% rename from core/capabilities/remote/target/request/server_request.go rename to core/capabilities/remote/executable/request/server_request.go index d23ba93a44d..a4662e93987 100644 --- a/core/capabilities/remote/target/request/server_request.go +++ b/core/capabilities/remote/executable/request/server_request.go @@ -23,7 +23,7 @@ type response struct { } type ServerRequest struct { - capability capabilities.TargetCapability + capability capabilities.ExecutableCapability capabilityPeerId p2ptypes.PeerID capabilityID string @@ -41,26 +41,29 @@ type ServerRequest struct { callingDon commoncap.DON requestMessageID string + method string requestTimeout time.Duration mux sync.Mutex lggr logger.Logger } -func NewServerRequest(capability capabilities.TargetCapability, capabilityID string, capabilityDonID uint32, capabilityPeerId p2ptypes.PeerID, - callingDon commoncap.DON, requestMessageID string, +func NewServerRequest(capability capabilities.ExecutableCapability, method string, capabilityID string, capabilityDonID uint32, + capabilityPeerID p2ptypes.PeerID, + callingDon commoncap.DON, requestID string, dispatcher types.Dispatcher, requestTimeout time.Duration, lggr logger.Logger) *ServerRequest { return &ServerRequest{ capability: capability, createdTime: time.Now(), capabilityID: capabilityID, capabilityDonID: capabilityDonID, - capabilityPeerId: capabilityPeerId, + capabilityPeerId: capabilityPeerID, dispatcher: dispatcher, requesters: map[p2ptypes.PeerID]bool{}, responseSentToRequester: map[p2ptypes.PeerID]bool{}, callingDon: callingDon, - requestMessageID: requestMessageID, + requestMessageID: requestID, + method: method, requestTimeout: requestTimeout, lggr: lggr.Named("ServerRequest"), } @@ -85,8 +88,15 @@ func (e *ServerRequest) OnMessage(ctx context.Context, msg *types.MessageBody) e e.lggr.Debugw("OnMessage called for request", "msgId", msg.MessageId, "calls", len(e.requesters), "hasResponse", e.response != nil) if e.minimumRequiredRequestsReceived() && !e.hasResponse() { - if err := e.executeRequest(ctx, msg.Payload); err != nil { - e.setError(types.Error_INTERNAL_ERROR, err.Error()) + switch e.method { + case types.MethodExecute: + e.executeRequest(ctx, msg.Payload, executeCapabilityRequest) + case types.MethodRegisterToWorkflow: + e.executeRequest(ctx, msg.Payload, registerToWorkflow) + case types.MethodUnregisterFromWorkflow: + e.executeRequest(ctx, msg.Payload, unregisterFromWorkflow) + default: + e.setError(types.Error_INTERNAL_ERROR, "unknown method %s"+e.method) } } @@ -115,31 +125,17 @@ func (e *ServerRequest) Cancel(err types.Error, msg string) error { return nil } -func (e *ServerRequest) executeRequest(ctx context.Context, payload []byte) error { +func (e *ServerRequest) executeRequest(ctx context.Context, payload []byte, method func(ctx context.Context, lggr logger.Logger, capability capabilities.ExecutableCapability, + payload []byte) ([]byte, error)) { ctxWithTimeout, cancel := context.WithTimeout(ctx, e.requestTimeout) defer cancel() - capabilityRequest, err := pb.UnmarshalCapabilityRequest(payload) - if err != nil { - return fmt.Errorf("failed to unmarshal capability request: %w", err) - } - - e.lggr.Debugw("executing capability", "metadata", capabilityRequest.Metadata) - capResponse, err := e.capability.Execute(ctxWithTimeout, capabilityRequest) - + responsePayload, err := method(ctxWithTimeout, e.lggr, e.capability, payload) if err != nil { - e.lggr.Debugw("received execution error", "workflowExecutionID", capabilityRequest.Metadata.WorkflowExecutionID, "error", err) - return fmt.Errorf("failed to execute capability: %w", err) - } - - responsePayload, err := pb.MarshalCapabilityResponse(capResponse) - if err != nil { - return fmt.Errorf("failed to marshal capability response: %w", err) + e.setError(types.Error_INTERNAL_ERROR, err.Error()) + } else { + e.setResult(responsePayload) } - - e.lggr.Debugw("received execution results", "workflowExecutionID", capabilityRequest.Metadata.WorkflowExecutionID) - e.setResult(responsePayload) - return nil } func (e *ServerRequest) addRequester(from p2ptypes.PeerID) error { @@ -227,3 +223,57 @@ func (e *ServerRequest) sendResponse(requester p2ptypes.PeerID) error { return nil } + +func executeCapabilityRequest(ctx context.Context, lggr logger.Logger, capability capabilities.ExecutableCapability, + payload []byte) ([]byte, error) { + capabilityRequest, err := pb.UnmarshalCapabilityRequest(payload) + if err != nil { + return nil, fmt.Errorf("failed to unmarshal capability request: %w", err) + } + + lggr.Debugw("executing capability", "metadata", capabilityRequest.Metadata) + capResponse, err := capability.Execute(ctx, capabilityRequest) + + if err != nil { + lggr.Debugw("received execution error", "workflowExecutionID", capabilityRequest.Metadata.WorkflowExecutionID, "error", err) + return nil, fmt.Errorf("failed to execute capability: %w", err) + } + + responsePayload, err := pb.MarshalCapabilityResponse(capResponse) + if err != nil { + return nil, fmt.Errorf("failed to marshal capability response: %w", err) + } + + lggr.Debugw("received execution results", "workflowExecutionID", capabilityRequest.Metadata.WorkflowExecutionID) + return responsePayload, nil +} + +func registerToWorkflow(ctx context.Context, _ logger.Logger, capability capabilities.ExecutableCapability, + payload []byte) ([]byte, error) { + registerRequest, err := pb.UnmarshalRegisterToWorkflowRequest(payload) + if err != nil { + return nil, fmt.Errorf("failed to unmarshal register to workflow request: %w", err) + } + + err = capability.RegisterToWorkflow(ctx, registerRequest) + if err != nil { + return nil, fmt.Errorf("failed to register to workflow: %w", err) + } + + return nil, nil +} + +func unregisterFromWorkflow(ctx context.Context, _ logger.Logger, capability capabilities.ExecutableCapability, + payload []byte) ([]byte, error) { + unregisterRequest, err := pb.UnmarshalUnregisterFromWorkflowRequest(payload) + if err != nil { + return nil, fmt.Errorf("failed to unmarshal unregister from workflow request: %w", err) + } + + err = capability.UnregisterFromWorkflow(ctx, unregisterRequest) + if err != nil { + return nil, fmt.Errorf("failed to unregister from workflow: %w", err) + } + + return nil, nil +} diff --git a/core/capabilities/remote/target/request/server_request_test.go b/core/capabilities/remote/executable/request/server_request_test.go similarity index 55% rename from core/capabilities/remote/target/request/server_request_test.go rename to core/capabilities/remote/executable/request/server_request_test.go index b4bddbc955f..cbeec833a1f 100644 --- a/core/capabilities/remote/target/request/server_request_test.go +++ b/core/capabilities/remote/executable/request/server_request_test.go @@ -14,7 +14,7 @@ import ( commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target/request" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable/request" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" "github.com/smartcontractkit/chainlink/v2/core/logger" p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" @@ -58,17 +58,17 @@ func Test_ServerRequest_MessageValidation(t *testing.T) { require.NoError(t, err) t.Run("Send duplicate message", func(t *testing.T) { - req := request.NewServerRequest(capability, "capabilityID", 2, + req := request.NewServerRequest(capability, types.MethodExecute, "capabilityID", 2, capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) err := sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) require.NoError(t, err) err = sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) - assert.NotNil(t, err) + assert.Error(t, err) }) t.Run("Send message with non calling don peer", func(t *testing.T) { - req := request.NewServerRequest(capability, "capabilityID", 2, + req := request.NewServerRequest(capability, types.MethodExecute, "capabilityID", 2, capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) err := sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) @@ -87,11 +87,11 @@ func Test_ServerRequest_MessageValidation(t *testing.T) { Payload: rawRequest, }) - assert.NotNil(t, err) + assert.Error(t, err) }) t.Run("Send message invalid payload", func(t *testing.T) { - req := request.NewServerRequest(capability, "capabilityID", 2, + req := request.NewServerRequest(capability, types.MethodExecute, "capabilityID", 2, capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) err := sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) @@ -108,15 +108,15 @@ func Test_ServerRequest_MessageValidation(t *testing.T) { Method: types.MethodExecute, Payload: append(rawRequest, []byte("asdf")...), }) - assert.NoError(t, err) - assert.Equal(t, 2, len(dispatcher.msgs)) - assert.Equal(t, dispatcher.msgs[0].Error, types.Error_INTERNAL_ERROR) - assert.Equal(t, dispatcher.msgs[1].Error, types.Error_INTERNAL_ERROR) + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[0].Error) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[1].Error) }) t.Run("Send second valid request when capability errors", func(t *testing.T) { dispatcher := &testDispatcher{} - req := request.NewServerRequest(TestErrorCapability{}, "capabilityID", 2, + req := request.NewServerRequest(TestErrorCapability{}, types.MethodExecute, "capabilityID", 2, capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) err := sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) @@ -133,17 +133,17 @@ func Test_ServerRequest_MessageValidation(t *testing.T) { Method: types.MethodExecute, Payload: rawRequest, }) - assert.NoError(t, err) - assert.Equal(t, 2, len(dispatcher.msgs)) - assert.Equal(t, dispatcher.msgs[0].Error, types.Error_INTERNAL_ERROR) - assert.Equal(t, dispatcher.msgs[0].ErrorMsg, "failed to execute capability: an error") - assert.Equal(t, dispatcher.msgs[1].Error, types.Error_INTERNAL_ERROR) - assert.Equal(t, dispatcher.msgs[1].ErrorMsg, "failed to execute capability: an error") + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[0].Error) + assert.Equal(t, "failed to execute capability: an error", dispatcher.msgs[0].ErrorMsg) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[1].Error) + assert.Equal(t, "failed to execute capability: an error", dispatcher.msgs[1].ErrorMsg) }) - t.Run("Send second valid request", func(t *testing.T) { + t.Run("Execute capability", func(t *testing.T) { dispatcher := &testDispatcher{} - request := request.NewServerRequest(capability, "capabilityID", 2, + request := request.NewServerRequest(capability, types.MethodExecute, "capabilityID", 2, capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) err := sendValidRequest(request, workflowPeers, capabilityPeerID, rawRequest) @@ -160,10 +160,111 @@ func Test_ServerRequest_MessageValidation(t *testing.T) { Method: types.MethodExecute, Payload: rawRequest, }) - assert.NoError(t, err) - assert.Equal(t, 2, len(dispatcher.msgs)) - assert.Equal(t, dispatcher.msgs[0].Error, types.Error_OK) - assert.Equal(t, dispatcher.msgs[1].Error, types.Error_OK) + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_OK, dispatcher.msgs[0].Error) + assert.Equal(t, types.Error_OK, dispatcher.msgs[1].Error) + }) + t.Run("Register to workflow request", func(t *testing.T) { + dispatcher := &testDispatcher{} + request := request.NewServerRequest(capability, types.MethodRegisterToWorkflow, "capabilityID", 2, + capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) + + err := sendValidRequest(request, workflowPeers, capabilityPeerID, rawRequest) + require.NoError(t, err) + + err = request.OnMessage(context.Background(), &types.MessageBody{ + Version: 0, + Sender: workflowPeers[1][:], + Receiver: capabilityPeerID[:], + MessageId: []byte("workflowID" + "workflowExecutionID"), + CapabilityId: "capabilityID", + CapabilityDonId: 2, + CallerDonId: 1, + Method: types.MethodRegisterToWorkflow, + Payload: rawRequest, + }) + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_OK, dispatcher.msgs[0].Error) + assert.Equal(t, types.Error_OK, dispatcher.msgs[1].Error) + }) + t.Run("Register to workflow request errors", func(t *testing.T) { + dispatcher := &testDispatcher{} + req := request.NewServerRequest(TestErrorCapability{}, types.MethodRegisterToWorkflow, "capabilityID", 2, + capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) + + err := sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) + require.NoError(t, err) + + err = req.OnMessage(context.Background(), &types.MessageBody{ + Version: 0, + Sender: workflowPeers[1][:], + Receiver: capabilityPeerID[:], + MessageId: []byte("workflowID" + "workflowExecutionID"), + CapabilityId: "capabilityID", + CapabilityDonId: 2, + CallerDonId: 1, + Method: types.MethodRegisterToWorkflow, + Payload: rawRequest, + }) + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[0].Error) + assert.Equal(t, "failed to register to workflow: an error", dispatcher.msgs[0].ErrorMsg) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[1].Error) + assert.Equal(t, "failed to register to workflow: an error", dispatcher.msgs[1].ErrorMsg) + }) + t.Run("Unregister from workflow request", func(t *testing.T) { + dispatcher := &testDispatcher{} + request := request.NewServerRequest(capability, types.MethodUnregisterFromWorkflow, "capabilityID", 2, + capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) + + err := sendValidRequest(request, workflowPeers, capabilityPeerID, rawRequest) + require.NoError(t, err) + + err = request.OnMessage(context.Background(), &types.MessageBody{ + Version: 0, + Sender: workflowPeers[1][:], + Receiver: capabilityPeerID[:], + MessageId: []byte("workflowID" + "workflowExecutionID"), + CapabilityId: "capabilityID", + CapabilityDonId: 2, + CallerDonId: 1, + Method: types.MethodUnregisterFromWorkflow, + Payload: rawRequest, + }) + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_OK, dispatcher.msgs[0].Error) + assert.Equal(t, types.Error_OK, dispatcher.msgs[1].Error) + }) + + t.Run("Unregister from workflow request errors", func(t *testing.T) { + dispatcher := &testDispatcher{} + req := request.NewServerRequest(TestErrorCapability{}, types.MethodUnregisterFromWorkflow, "capabilityID", 2, + capabilityPeerID, callingDon, "requestMessageID", dispatcher, 10*time.Minute, lggr) + + err := sendValidRequest(req, workflowPeers, capabilityPeerID, rawRequest) + require.NoError(t, err) + + err = req.OnMessage(context.Background(), &types.MessageBody{ + Version: 0, + Sender: workflowPeers[1][:], + Receiver: capabilityPeerID[:], + MessageId: []byte("workflowID" + "workflowExecutionID"), + CapabilityId: "capabilityID", + CapabilityDonId: 2, + CallerDonId: 1, + Method: types.MethodUnregisterFromWorkflow, + Payload: rawRequest, + }) + require.NoError(t, err) + assert.Len(t, dispatcher.msgs, 2) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[0].Error) + assert.Equal(t, "failed to unregister from workflow: an error", dispatcher.msgs[0].ErrorMsg) + assert.Equal(t, types.Error_INTERNAL_ERROR, dispatcher.msgs[1].Error) + assert.Equal(t, "failed to unregister from workflow: an error", dispatcher.msgs[1].ErrorMsg) }) } @@ -261,6 +362,14 @@ func (t TestErrorCapability) Execute(ctx context.Context, request commoncap.Capa return commoncap.CapabilityResponse{}, errors.New("an error") } +func (t TestErrorCapability) RegisterToWorkflow(ctx context.Context, request commoncap.RegisterToWorkflowRequest) error { + return errors.New("an error") +} + +func (t TestErrorCapability) UnregisterFromWorkflow(ctx context.Context, request commoncap.UnregisterFromWorkflowRequest) error { + return errors.New("an error") +} + func NewP2PPeerID(t *testing.T) p2ptypes.PeerID { id := p2ptypes.PeerID{} require.NoError(t, id.UnmarshalText([]byte(NewPeerID()))) diff --git a/core/capabilities/remote/target/server.go b/core/capabilities/remote/executable/server.go similarity index 83% rename from core/capabilities/remote/target/server.go rename to core/capabilities/remote/executable/server.go index 61725a8220c..b767a2d7030 100644 --- a/core/capabilities/remote/target/server.go +++ b/core/capabilities/remote/executable/server.go @@ -1,4 +1,4 @@ -package target +package executable import ( "context" @@ -12,7 +12,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target/request" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable/request" "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" "github.com/smartcontractkit/chainlink/v2/core/capabilities/validation" p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" @@ -20,9 +20,9 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/logger" ) -// server manages all external users of a local target capability. +// server manages all external users of a local executable capability. // Its responsibilities are: -// 1. Manage requests from external nodes executing the target capability once sufficient requests are received. +// 1. Manage requests from external nodes executing the executable capability once sufficient requests are received. // 2. Send out responses produced by an underlying capability to all requesters. // // server communicates with corresponding client on remote nodes. @@ -30,9 +30,9 @@ type server struct { services.StateMachine lggr logger.Logger - config *commoncap.RemoteTargetConfig + config *commoncap.RemoteExecutableConfig peerID p2ptypes.PeerID - underlying commoncap.TargetCapability + underlying commoncap.ExecutableCapability capInfo commoncap.CapabilityInfo localDonInfo commoncap.DON workflowDONs map[uint32]commoncap.DON @@ -57,14 +57,15 @@ type requestAndMsgID struct { messageID string } -func NewServer(config *commoncap.RemoteTargetConfig, peerID p2ptypes.PeerID, underlying commoncap.TargetCapability, capInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, +func NewServer(remoteExecutableConfig *commoncap.RemoteExecutableConfig, peerID p2ptypes.PeerID, underlying commoncap.ExecutableCapability, + capInfo commoncap.CapabilityInfo, localDonInfo commoncap.DON, workflowDONs map[uint32]commoncap.DON, dispatcher types.Dispatcher, requestTimeout time.Duration, lggr logger.Logger) *server { - if config == nil { - lggr.Info("no config provided, using default values") - config = &commoncap.RemoteTargetConfig{} + if remoteExecutableConfig == nil { + lggr.Info("no remote config provided, using default values") + remoteExecutableConfig = &commoncap.RemoteExecutableConfig{} } return &server{ - config: config, + config: remoteExecutableConfig, underlying: underlying, peerID: peerID, capInfo: capInfo, @@ -76,7 +77,7 @@ func NewServer(config *commoncap.RemoteTargetConfig, peerID p2ptypes.PeerID, und messageIDToRequestIDsCount: map[string]map[string]int{}, requestTimeout: requestTimeout, - lggr: lggr.Named("TargetServer"), + lggr: lggr.Named("ExecutableCapabilityServer"), stopCh: make(services.StopChan), } } @@ -88,7 +89,7 @@ func (r *server) Start(ctx context.Context) error { defer r.wg.Done() ticker := time.NewTicker(r.requestTimeout) defer ticker.Stop() - r.lggr.Info("TargetServer started") + r.lggr.Info("executable capability server started") for { select { case <-r.stopCh: @@ -106,7 +107,7 @@ func (r *server) Close() error { return r.StopOnce(r.Name(), func() error { close(r.stopCh) r.wg.Wait() - r.lggr.Info("TargetServer closed") + r.lggr.Info("executable capability server closed") return nil }) } @@ -131,9 +132,10 @@ func (r *server) Receive(ctx context.Context, msg *types.MessageBody) { r.receiveLock.Lock() defer r.receiveLock.Unlock() - if msg.Method != types.MethodExecute { + switch msg.Method { + case types.MethodExecute, types.MethodRegisterToWorkflow, types.MethodUnregisterFromWorkflow: + default: r.lggr.Errorw("received request for unsupported method type", "method", remote.SanitizeLogString(msg.Method)) - return } messageId, err := GetMessageID(msg) @@ -175,7 +177,7 @@ func (r *server) Receive(ctx context.Context, msg *types.MessageBody) { } r.requestIDToRequest[requestID] = requestAndMsgID{ - request: request.NewServerRequest(r.underlying, r.capInfo.ID, r.localDonInfo.ID, r.peerID, + request: request.NewServerRequest(r.underlying, msg.Method, r.capInfo.ID, r.localDonInfo.ID, r.peerID, callingDon, messageId, r.dispatcher, r.requestTimeout, r.lggr), messageID: messageId, } @@ -196,8 +198,8 @@ func (r *server) getMessageHash(msg *types.MessageBody) ([32]byte, error) { } for _, path := range r.config.RequestHashExcludedAttributes { - if !req.Inputs.DeleteAtPath(path) { - return [32]byte{}, fmt.Errorf("failed to delete attribute from map at path: %s", path) + if req.Inputs != nil { + req.Inputs.DeleteAtPath(path) } } diff --git a/core/capabilities/remote/target/server_test.go b/core/capabilities/remote/executable/server_test.go similarity index 58% rename from core/capabilities/remote/target/server_test.go rename to core/capabilities/remote/executable/server_test.go index 505a2dcce5d..1fb5c2dd413 100644 --- a/core/capabilities/remote/target/server_test.go +++ b/core/capabilities/remote/executable/server_test.go @@ -1,4 +1,4 @@ -package target_test +package executable_test import ( "context" @@ -13,7 +13,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/executable" remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" "github.com/smartcontractkit/chainlink/v2/core/logger" @@ -25,7 +25,7 @@ func Test_Server_ExcludesNonDeterministicInputAttributes(t *testing.T) { numCapabilityPeers := 4 - callers, srvcs := testRemoteTargetServer(ctx, t, &commoncap.RemoteTargetConfig{RequestHashExcludedAttributes: []string{"signed_report.Signatures"}}, + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{RequestHashExcludedAttributes: []string{"signed_report.Signatures"}}, &TestCapability{}, 10, 9, numCapabilityPeers, 3, 10*time.Minute) for idx, caller := range callers { @@ -56,12 +56,12 @@ func Test_Server_ExcludesNonDeterministicInputAttributes(t *testing.T) { closeServices(t, srvcs) } -func Test_Server_RespondsAfterSufficientRequests(t *testing.T) { +func Test_Server_Execute_RespondsAfterSufficientRequests(t *testing.T) { ctx := testutils.Context(t) numCapabilityPeers := 4 - callers, srvcs := testRemoteTargetServer(ctx, t, &commoncap.RemoteTargetConfig{}, &TestCapability{}, 10, 9, numCapabilityPeers, 3, 10*time.Minute) + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{}, &TestCapability{}, 10, 9, numCapabilityPeers, 3, 10*time.Minute) for _, caller := range callers { _, err := caller.Execute(context.Background(), @@ -83,12 +83,95 @@ func Test_Server_RespondsAfterSufficientRequests(t *testing.T) { closeServices(t, srvcs) } +func Test_Server_RegisterToWorkflow_RespondsAfterSufficientRequests(t *testing.T) { + ctx := testutils.Context(t) + + numCapabilityPeers := 4 + + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{}, &TestCapability{}, 10, 9, numCapabilityPeers, 3, 10*time.Minute) + + for _, caller := range callers { + err := caller.RegisterToWorkflow(context.Background(), commoncap.RegisterToWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + }) + + require.NoError(t, err) + } + + for _, caller := range callers { + for i := 0; i < numCapabilityPeers; i++ { + msg := <-caller.receivedMessages + assert.Equal(t, remotetypes.Error_OK, msg.Error) + } + } + closeServices(t, srvcs) +} + +func Test_Server_RegisterToWorkflow_Error(t *testing.T) { + ctx := testutils.Context(t) + + numCapabilityPeers := 4 + + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{}, &TestErrorCapability{}, 10, 9, numCapabilityPeers, 3, 10*time.Minute) + + for _, caller := range callers { + err := caller.RegisterToWorkflow(context.Background(), commoncap.RegisterToWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + }) + + require.NoError(t, err) + } + + for _, caller := range callers { + for i := 0; i < numCapabilityPeers; i++ { + msg := <-caller.receivedMessages + assert.Equal(t, remotetypes.Error_INTERNAL_ERROR, msg.Error) + } + } + closeServices(t, srvcs) +} + +func Test_Server_UnregisterFromWorkflow_RespondsAfterSufficientRequests(t *testing.T) { + ctx := testutils.Context(t) + + numCapabilityPeers := 4 + + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{}, &TestCapability{}, 10, 9, numCapabilityPeers, 3, 10*time.Minute) + + for _, caller := range callers { + err := caller.UnregisterFromWorkflow(context.Background(), commoncap.UnregisterFromWorkflowRequest{ + Metadata: commoncap.RegistrationMetadata{ + WorkflowID: workflowID1, + ReferenceID: stepReferenceID1, + WorkflowOwner: workflowOwnerID, + }, + }) + require.NoError(t, err) + } + + for _, caller := range callers { + for i := 0; i < numCapabilityPeers; i++ { + msg := <-caller.receivedMessages + assert.Equal(t, remotetypes.Error_OK, msg.Error) + } + } + closeServices(t, srvcs) +} + func Test_Server_InsufficientCallers(t *testing.T) { ctx := testutils.Context(t) numCapabilityPeers := 4 - callers, srvcs := testRemoteTargetServer(ctx, t, &commoncap.RemoteTargetConfig{}, &TestCapability{}, 10, 10, numCapabilityPeers, 3, 100*time.Millisecond) + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{}, &TestCapability{}, 10, 10, numCapabilityPeers, 3, 100*time.Millisecond) for _, caller := range callers { _, err := caller.Execute(context.Background(), @@ -115,7 +198,7 @@ func Test_Server_CapabilityError(t *testing.T) { numCapabilityPeers := 4 - callers, srvcs := testRemoteTargetServer(ctx, t, &commoncap.RemoteTargetConfig{}, &TestErrorCapability{}, 10, 9, numCapabilityPeers, 3, 100*time.Millisecond) + callers, srvcs := testRemoteExecutableCapabilityServer(ctx, t, &commoncap.RemoteExecutableConfig{}, &TestErrorCapability{}, 10, 9, numCapabilityPeers, 3, 100*time.Millisecond) for _, caller := range callers { _, err := caller.Execute(context.Background(), @@ -137,9 +220,9 @@ func Test_Server_CapabilityError(t *testing.T) { closeServices(t, srvcs) } -func testRemoteTargetServer(ctx context.Context, t *testing.T, - config *commoncap.RemoteTargetConfig, - underlying commoncap.TargetCapability, +func testRemoteExecutableCapabilityServer(ctx context.Context, t *testing.T, + config *commoncap.RemoteExecutableConfig, + underlying commoncap.ExecutableCapability, numWorkflowPeers int, workflowDonF uint8, numCapabilityPeers int, capabilityDonF uint8, capabilityNodeResponseTimeout time.Duration) ([]*serverTestClient, []services.Service) { lggr := logger.TestLogger(t) @@ -189,7 +272,7 @@ func testRemoteTargetServer(ctx context.Context, t *testing.T, for i := 0; i < numCapabilityPeers; i++ { capabilityPeer := capabilityPeers[i] capabilityDispatcher := broker.NewDispatcherForNode(capabilityPeer) - capabilityNode := target.NewServer(config, capabilityPeer, underlying, capInfo, capDonInfo, workflowDONs, capabilityDispatcher, + capabilityNode := executable.NewServer(config, capabilityPeer, underlying, capInfo, capDonInfo, workflowDONs, capabilityDispatcher, capabilityNodeResponseTimeout, lggr) require.NoError(t, capabilityNode.Start(ctx)) broker.RegisterReceiverNode(capabilityPeer, capabilityNode) @@ -236,12 +319,60 @@ func (r *serverTestClient) Info(ctx context.Context) (commoncap.CapabilityInfo, panic("not implemented") } -func (r *serverTestClient) RegisterToWorkflow(ctx context.Context, request commoncap.RegisterToWorkflowRequest) error { - panic("not implemented") +func (r *serverTestClient) RegisterToWorkflow(ctx context.Context, req commoncap.RegisterToWorkflowRequest) error { + rawRequest, err := pb.MarshalRegisterToWorkflowRequest(req) + if err != nil { + return err + } + + messageID := remotetypes.MethodRegisterToWorkflow + ":" + req.Metadata.WorkflowID + + for _, node := range r.capabilityDonInfo.Members { + message := &remotetypes.MessageBody{ + CapabilityId: "capability-id", + CapabilityDonId: 1, + CallerDonId: 2, + Method: remotetypes.MethodRegisterToWorkflow, + Payload: rawRequest, + MessageId: []byte(messageID), + Sender: r.peerID[:], + Receiver: node[:], + } + + if err = r.dispatcher.Send(node, message); err != nil { + return err + } + } + + return nil } -func (r *serverTestClient) UnregisterFromWorkflow(ctx context.Context, request commoncap.UnregisterFromWorkflowRequest) error { - panic("not implemented") +func (r *serverTestClient) UnregisterFromWorkflow(ctx context.Context, req commoncap.UnregisterFromWorkflowRequest) error { + rawRequest, err := pb.MarshalUnregisterFromWorkflowRequest(req) + if err != nil { + return err + } + + messageID := remotetypes.MethodUnregisterFromWorkflow + ":" + req.Metadata.WorkflowID + + for _, node := range r.capabilityDonInfo.Members { + message := &remotetypes.MessageBody{ + CapabilityId: "capability-id", + CapabilityDonId: 1, + CallerDonId: 2, + Method: remotetypes.MethodUnregisterFromWorkflow, + Payload: rawRequest, + MessageId: []byte(messageID), + Sender: r.peerID[:], + Receiver: node[:], + } + + if err = r.dispatcher.Send(node, message); err != nil { + return err + } + } + + return nil } func (r *serverTestClient) Execute(ctx context.Context, req commoncap.CapabilityRequest) (<-chan commoncap.CapabilityResponse, error) { @@ -250,10 +381,7 @@ func (r *serverTestClient) Execute(ctx context.Context, req commoncap.Capability return nil, err } - messageID, err := target.GetMessageIDForRequest(req) - if err != nil { - return nil, err - } + messageID := remotetypes.MethodExecute + ":" + req.Metadata.WorkflowExecutionID for _, node := range r.capabilityDonInfo.Members { message := &remotetypes.MessageBody{ diff --git a/core/capabilities/remote/target/client_test.go b/core/capabilities/remote/target/client_test.go deleted file mode 100644 index 697cb6e383a..00000000000 --- a/core/capabilities/remote/target/client_test.go +++ /dev/null @@ -1,316 +0,0 @@ -package target_test - -import ( - "context" - "sync" - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - - commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" - "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" - "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/target" - remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/transmission" - "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" - "github.com/smartcontractkit/chainlink/v2/core/logger" - p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" -) - -const ( - workflowID1 = "15c631d295ef5e32deb99a10ee6804bc4af13855687559d7ff6552ac6dbb2ce0" - workflowExecutionID1 = "95ef5e32deb99a10ee6804bc4af13855687559d7ff6552ac6dbb2ce0abbadeed" -) - -func Test_Client_DonTopologies(t *testing.T) { - ctx := testutils.Context(t) - - transmissionSchedule, err := values.NewMap(map[string]any{ - "schedule": transmission.Schedule_OneAtATime, - "deltaStage": "10ms", - }) - require.NoError(t, err) - - responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { - require.NoError(t, responseError) - mp, err := response.Value.Unwrap() - require.NoError(t, err) - assert.Equal(t, "aValue1", mp.(map[string]any)["response"].(string)) - } - - capability := &TestCapability{} - - responseTimeOut := 10 * time.Minute - - testClient(ctx, t, 1, responseTimeOut, 1, 0, - capability, transmissionSchedule, responseTest) - - testClient(ctx, t, 10, responseTimeOut, 1, 0, - capability, transmissionSchedule, responseTest) - - testClient(ctx, t, 1, responseTimeOut, 10, 3, - capability, transmissionSchedule, responseTest) - - testClient(ctx, t, 10, responseTimeOut, 10, 3, - capability, transmissionSchedule, responseTest) - - testClient(ctx, t, 10, responseTimeOut, 10, 9, - capability, transmissionSchedule, responseTest) -} - -func Test_Client_TransmissionSchedules(t *testing.T) { - ctx := testutils.Context(t) - - responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { - require.NoError(t, responseError) - mp, err := response.Value.Unwrap() - require.NoError(t, err) - assert.Equal(t, "aValue1", mp.(map[string]any)["response"].(string)) - } - - capability := &TestCapability{} - - responseTimeOut := 10 * time.Minute - - transmissionSchedule, err := values.NewMap(map[string]any{ - "schedule": transmission.Schedule_OneAtATime, - "deltaStage": "10ms", - }) - require.NoError(t, err) - - testClient(ctx, t, 1, responseTimeOut, 1, 0, - capability, transmissionSchedule, responseTest) - testClient(ctx, t, 10, responseTimeOut, 10, 3, - capability, transmissionSchedule, responseTest) - - transmissionSchedule, err = values.NewMap(map[string]any{ - "schedule": transmission.Schedule_AllAtOnce, - "deltaStage": "10ms", - }) - require.NoError(t, err) - - testClient(ctx, t, 1, responseTimeOut, 1, 0, - capability, transmissionSchedule, responseTest) - testClient(ctx, t, 10, responseTimeOut, 10, 3, - capability, transmissionSchedule, responseTest) -} - -func Test_Client_TimesOutIfInsufficientCapabilityPeerResponses(t *testing.T) { - ctx := testutils.Context(t) - - responseTest := func(t *testing.T, response commoncap.CapabilityResponse, responseError error) { - assert.NotNil(t, responseError) - } - - capability := &TestCapability{} - - transmissionSchedule, err := values.NewMap(map[string]any{ - "schedule": transmission.Schedule_AllAtOnce, - "deltaStage": "10ms", - }) - require.NoError(t, err) - - // number of capability peers is less than F + 1 - - testClient(ctx, t, 10, 1*time.Second, 10, 11, - capability, transmissionSchedule, responseTest) -} - -func testClient(ctx context.Context, t *testing.T, numWorkflowPeers int, workflowNodeResponseTimeout time.Duration, - numCapabilityPeers int, capabilityDonF uint8, underlying commoncap.TargetCapability, transmissionSchedule *values.Map, - responseTest func(t *testing.T, responseCh commoncap.CapabilityResponse, responseError error)) { - lggr := logger.TestLogger(t) - - capabilityPeers := make([]p2ptypes.PeerID, numCapabilityPeers) - for i := 0; i < numCapabilityPeers; i++ { - capabilityPeers[i] = NewP2PPeerID(t) - } - - capDonInfo := commoncap.DON{ - ID: 1, - Members: capabilityPeers, - F: capabilityDonF, - } - - capInfo := commoncap.CapabilityInfo{ - ID: "cap_id@1.0.0", - CapabilityType: commoncap.CapabilityTypeTarget, - Description: "Remote Target", - DON: &capDonInfo, - } - - workflowPeers := make([]p2ptypes.PeerID, numWorkflowPeers) - for i := 0; i < numWorkflowPeers; i++ { - workflowPeers[i] = NewP2PPeerID(t) - } - - workflowDonInfo := commoncap.DON{ - Members: workflowPeers, - ID: 2, - } - - broker := newTestAsyncMessageBroker(t, 100) - - receivers := make([]remotetypes.Receiver, numCapabilityPeers) - for i := 0; i < numCapabilityPeers; i++ { - capabilityDispatcher := broker.NewDispatcherForNode(capabilityPeers[i]) - receiver := newTestServer(capabilityPeers[i], capabilityDispatcher, workflowDonInfo, underlying) - broker.RegisterReceiverNode(capabilityPeers[i], receiver) - receivers[i] = receiver - } - - callers := make([]commoncap.TargetCapability, numWorkflowPeers) - - for i := 0; i < numWorkflowPeers; i++ { - workflowPeerDispatcher := broker.NewDispatcherForNode(workflowPeers[i]) - caller := target.NewClient(capInfo, workflowDonInfo, workflowPeerDispatcher, workflowNodeResponseTimeout, lggr) - servicetest.Run(t, caller) - broker.RegisterReceiverNode(workflowPeers[i], caller) - callers[i] = caller - } - - servicetest.Run(t, broker) - - executeInputs, err := values.NewMap( - map[string]any{ - "executeValue1": "aValue1", - }, - ) - - require.NoError(t, err) - - wg := &sync.WaitGroup{} - wg.Add(len(callers)) - - // Fire off all the requests - for _, caller := range callers { - go func(caller commoncap.TargetCapability) { - defer wg.Done() - responseCh, err := caller.Execute(ctx, - commoncap.CapabilityRequest{ - Metadata: commoncap.RequestMetadata{ - WorkflowID: workflowID1, - WorkflowExecutionID: workflowExecutionID1, - }, - Config: transmissionSchedule, - Inputs: executeInputs, - }) - - responseTest(t, responseCh, err) - }(caller) - } - - wg.Wait() -} - -// Simple client that only responds once it has received a message from each workflow peer -type clientTestServer struct { - peerID p2ptypes.PeerID - dispatcher remotetypes.Dispatcher - workflowDonInfo commoncap.DON - messageIDToSenders map[string]map[p2ptypes.PeerID]bool - - targetCapability commoncap.TargetCapability - - mux sync.Mutex -} - -func newTestServer(peerID p2ptypes.PeerID, dispatcher remotetypes.Dispatcher, workflowDonInfo commoncap.DON, - targetCapability commoncap.TargetCapability) *clientTestServer { - return &clientTestServer{ - dispatcher: dispatcher, - workflowDonInfo: workflowDonInfo, - peerID: peerID, - messageIDToSenders: make(map[string]map[p2ptypes.PeerID]bool), - targetCapability: targetCapability, - } -} - -func (t *clientTestServer) Receive(_ context.Context, msg *remotetypes.MessageBody) { - t.mux.Lock() - defer t.mux.Unlock() - - sender := toPeerID(msg.Sender) - messageID, err := target.GetMessageID(msg) - if err != nil { - panic(err) - } - - if t.messageIDToSenders[messageID] == nil { - t.messageIDToSenders[messageID] = make(map[p2ptypes.PeerID]bool) - } - - sendersOfMessageID := t.messageIDToSenders[messageID] - if sendersOfMessageID[sender] { - panic("received duplicate message") - } - - sendersOfMessageID[sender] = true - - if len(t.messageIDToSenders[messageID]) == len(t.workflowDonInfo.Members) { - capabilityRequest, err := pb.UnmarshalCapabilityRequest(msg.Payload) - if err != nil { - panic(err) - } - - resp, responseErr := t.targetCapability.Execute(context.Background(), capabilityRequest) - - for receiver := range t.messageIDToSenders[messageID] { - var responseMsg = &remotetypes.MessageBody{ - CapabilityId: "cap_id@1.0.0", - CapabilityDonId: 1, - CallerDonId: t.workflowDonInfo.ID, - Method: remotetypes.MethodExecute, - MessageId: []byte(messageID), - Sender: t.peerID[:], - Receiver: receiver[:], - } - - if responseErr != nil { - responseMsg.Error = remotetypes.Error_INTERNAL_ERROR - } else { - payload, marshalErr := pb.MarshalCapabilityResponse(resp) - if marshalErr != nil { - panic(marshalErr) - } - responseMsg.Payload = payload - } - - err = t.dispatcher.Send(receiver, responseMsg) - if err != nil { - panic(err) - } - } - } -} - -type TestDispatcher struct { - sentMessagesCh chan *remotetypes.MessageBody - receiver remotetypes.Receiver -} - -func NewTestDispatcher() *TestDispatcher { - return &TestDispatcher{ - sentMessagesCh: make(chan *remotetypes.MessageBody, 1), - } -} - -func (t *TestDispatcher) SendToReceiver(msgBody *remotetypes.MessageBody) { - t.receiver.Receive(context.Background(), msgBody) -} - -func (t *TestDispatcher) SetReceiver(capabilityId string, donId string, receiver remotetypes.Receiver) error { - t.receiver = receiver - return nil -} - -func (t *TestDispatcher) RemoveReceiver(capabilityId string, donId string) {} - -func (t *TestDispatcher) Send(peerID p2ptypes.PeerID, msgBody *remotetypes.MessageBody) error { - t.sentMessagesCh <- msgBody - return nil -} diff --git a/core/capabilities/remote/types/types.go b/core/capabilities/remote/types/types.go index 54ec16f09f1..fefc9a9b5fe 100644 --- a/core/capabilities/remote/types/types.go +++ b/core/capabilities/remote/types/types.go @@ -13,10 +13,12 @@ import ( ) const ( - MethodRegisterTrigger = "RegisterTrigger" - MethodUnRegisterTrigger = "UnregisterTrigger" - MethodTriggerEvent = "TriggerEvent" - MethodExecute = "Execute" + MethodRegisterTrigger = "RegisterTrigger" + MethodUnRegisterTrigger = "UnregisterTrigger" + MethodTriggerEvent = "TriggerEvent" + MethodExecute = "Execute" + MethodRegisterToWorkflow = "RegisterToWorkflow" + MethodUnregisterFromWorkflow = "UnregisterFromWorkflow" ) type Dispatcher interface { diff --git a/core/capabilities/transmission/local_target_capability_test.go b/core/capabilities/transmission/local_target_capability_test.go index 67f22753bda..e1057ed3f8d 100644 --- a/core/capabilities/transmission/local_target_capability_test.go +++ b/core/capabilities/transmission/local_target_capability_test.go @@ -61,15 +61,15 @@ func TestScheduledExecutionStrategy_LocalDON(t *testing.T) { name: "position 1; oneAtATime", position: 1, schedule: "oneAtATime", - low: 100 * time.Millisecond, - high: 300 * time.Millisecond, + low: 300 * time.Millisecond, + high: 400 * time.Millisecond, }, { name: "position 2; oneAtATime", position: 2, schedule: "oneAtATime", - low: 300 * time.Millisecond, - high: 400 * time.Millisecond, + low: 100 * time.Millisecond, + high: 200 * time.Millisecond, }, { name: "position 3; oneAtATime", diff --git a/core/capabilities/transmission/transmission.go b/core/capabilities/transmission/transmission.go index 88ce0fa3edd..8dd90414fb9 100644 --- a/core/capabilities/transmission/transmission.go +++ b/core/capabilities/transmission/transmission.go @@ -27,7 +27,7 @@ type TransmissionConfig struct { DeltaStage time.Duration } -func extractTransmissionConfig(config *values.Map) (TransmissionConfig, error) { +func ExtractTransmissionConfig(config *values.Map) (TransmissionConfig, error) { var tc struct { DeltaStage string Schedule string @@ -37,6 +37,14 @@ func extractTransmissionConfig(config *values.Map) (TransmissionConfig, error) { return TransmissionConfig{}, fmt.Errorf("failed to unwrap tranmission config from value map: %w", err) } + // Default if no schedule and deltaStage is provided + if len(tc.Schedule) == 0 && len(tc.DeltaStage) == 0 { + return TransmissionConfig{ + Schedule: Schedule_AllAtOnce, + DeltaStage: 0, + }, nil + } + duration, err := time.ParseDuration(tc.DeltaStage) if err != nil { return TransmissionConfig{}, fmt.Errorf("failed to parse DeltaStage %s as duration: %w", tc.DeltaStage, err) @@ -51,21 +59,20 @@ func extractTransmissionConfig(config *values.Map) (TransmissionConfig, error) { // GetPeerIDToTransmissionDelay returns a map of PeerID to the time.Duration that the node with that PeerID should wait // before transmitting the capability request. If a node is not in the map, it should not transmit. func GetPeerIDToTransmissionDelay(donPeerIDs []types.PeerID, req capabilities.CapabilityRequest) (map[types.PeerID]time.Duration, error) { - tc, err := extractTransmissionConfig(req.Config) + tc, err := ExtractTransmissionConfig(req.Config) if err != nil { return nil, fmt.Errorf("failed to extract transmission config from request: %w", err) } - if err = validation.ValidateWorkflowOrExecutionID(req.Metadata.WorkflowID); err != nil { - return nil, fmt.Errorf("workflow ID is invalid: %w", err) + workflowExecutionID := req.Metadata.WorkflowExecutionID + if err := validation.ValidateWorkflowOrExecutionID(workflowExecutionID); err != nil { + return nil, fmt.Errorf("workflow or execution ID is invalid: %w", err) } - if err = validation.ValidateWorkflowOrExecutionID(req.Metadata.WorkflowExecutionID); err != nil { - return nil, fmt.Errorf("workflow execution ID is invalid: %w", err) - } - - transmissionID := req.Metadata.WorkflowID + req.Metadata.WorkflowExecutionID + return GetPeerIDToTransmissionDelaysForConfig(donPeerIDs, workflowExecutionID, tc) +} +func GetPeerIDToTransmissionDelaysForConfig(donPeerIDs []types.PeerID, transmissionID string, tc TransmissionConfig) (map[types.PeerID]time.Duration, error) { donMemberCount := len(donPeerIDs) key := transmissionScheduleSeed(transmissionID) schedule, err := createTransmissionSchedule(tc.Schedule, donMemberCount) diff --git a/core/capabilities/transmission/transmission_test.go b/core/capabilities/transmission/transmission_test.go index aaa367e78cf..1cb91c364a5 100644 --- a/core/capabilities/transmission/transmission_test.go +++ b/core/capabilities/transmission/transmission_test.go @@ -38,10 +38,10 @@ func Test_GetPeerIDToTransmissionDelay(t *testing.T) { "100ms", "15c631d295ef5e32deb99a10ee6804bc4af13855687559d7ff6552ac6dbb2ce0", map[string]time.Duration{ - "one": 300 * time.Millisecond, - "two": 0 * time.Millisecond, - "three": 100 * time.Millisecond, - "four": 200 * time.Millisecond, + "one": 100 * time.Millisecond, + "two": 200 * time.Millisecond, + "three": 300 * time.Millisecond, + "four": 0 * time.Millisecond, }, }, @@ -66,10 +66,10 @@ func Test_GetPeerIDToTransmissionDelay(t *testing.T) { "100ms", "16c631d295ef5e32deb99a10ee6804bc4af13855687559d7ff6552ac6dbb2ce1", map[string]time.Duration{ - "one": 300 * time.Millisecond, + "one": 200 * time.Millisecond, "two": 100 * time.Millisecond, - "three": 200 * time.Millisecond, - "four": 0 * time.Millisecond, + "three": 0 * time.Millisecond, + "four": 300 * time.Millisecond, }, }, } diff --git a/core/scripts/go.mod b/core/scripts/go.mod index caf3d5e68e6..028eb9fac9e 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -24,7 +24,7 @@ require ( github.com/prometheus/client_golang v1.20.5 github.com/shopspring/decimal v1.4.0 github.com/smartcontractkit/chainlink-automation v0.8.1 - github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 + github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 github.com/smartcontractkit/chainlink/deployment v0.0.0-00010101000000-000000000000 github.com/smartcontractkit/chainlink/v2 v2.14.0-mercury-20240807.0.20241106193309-5560cd76211a github.com/smartcontractkit/libocr v0.0.0-20241007185508-adbe57025f12 diff --git a/core/scripts/go.sum b/core/scripts/go.sum index 362d28f28c3..bdf3511c482 100644 --- a/core/scripts/go.sum +++ b/core/scripts/go.sum @@ -1094,8 +1094,8 @@ github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgB github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b h1:4kmZtaQ4fXwduHnw9xk5VmiIOW4nHg/Mx6iidlZJt5o= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b/go.mod h1:4adKaHNaxFsRvV/lYfqtbsWyyvIPUMLR0FdOJN/ljis= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 h1:vnNqMaAvheZgR8IDMGw0QIV1Qen3XTh7IChwW40SNfU= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 h1:2llRW4Tn9W/EZp2XvXclQ9IjeTBwwxVPrrqaerX+vCE= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f h1:BwrIaQIx5Iy6eT+DfLhFfK2XqjxRm74mVdlX8gbu4dw= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f/go.mod h1:wHtwSR3F1CQSJJZDQKuqaqFYnvkT+kMyget7dl8Clvo= github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241018134907-a00ba3729b5e h1:JiETqdNM0bktAUGMc62COwXIaw3rR3M77Me6bBLG0Fg= diff --git a/core/services/workflows/engine.go b/core/services/workflows/engine.go index 69c36c1c174..b958e171c0c 100644 --- a/core/services/workflows/engine.go +++ b/core/services/workflows/engine.go @@ -285,6 +285,7 @@ func (e *Engine) initializeCapability(ctx context.Context, step *step) error { Metadata: capabilities.RegistrationMetadata{ WorkflowID: e.workflow.id, WorkflowOwner: e.workflow.owner, + ReferenceID: step.Vertex.Ref, }, Config: stepConfig, } @@ -1112,6 +1113,7 @@ func (e *Engine) Close() error { Metadata: capabilities.RegistrationMetadata{ WorkflowID: e.workflow.id, WorkflowOwner: e.workflow.owner, + ReferenceID: s.Vertex.Ref, }, Config: stepConfig, } @@ -1119,7 +1121,7 @@ func (e *Engine) Close() error { innerErr := s.capability.UnregisterFromWorkflow(ctx, reg) if innerErr != nil { return &workflowError{err: innerErr, - reason: fmt.Sprintf("failed to unregister capability from workflow: %+v", reg), + reason: fmt.Sprintf("failed to unregister capability from workflow: %+v", reg), labels: map[string]string{ platform.KeyWorkflowID: e.workflow.id, platform.KeyStepID: s.ID, diff --git a/deployment/go.mod b/deployment/go.mod index 19720794189..73e23aa83f3 100644 --- a/deployment/go.mod +++ b/deployment/go.mod @@ -23,7 +23,7 @@ require ( github.com/smartcontractkit/ccip-owner-contracts v0.0.0-20240926212305-a6deabdfce86 github.com/smartcontractkit/chain-selectors v1.0.29 github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b - github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 + github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 github.com/smartcontractkit/chainlink-protos/job-distributor v0.4.0 github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.13 github.com/smartcontractkit/chainlink/v2 v2.0.0-00010101000000-000000000000 diff --git a/deployment/go.sum b/deployment/go.sum index ce9bf9e0b7f..6c372de39f2 100644 --- a/deployment/go.sum +++ b/deployment/go.sum @@ -1384,8 +1384,8 @@ github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgB github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b h1:4kmZtaQ4fXwduHnw9xk5VmiIOW4nHg/Mx6iidlZJt5o= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b/go.mod h1:4adKaHNaxFsRvV/lYfqtbsWyyvIPUMLR0FdOJN/ljis= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 h1:vnNqMaAvheZgR8IDMGw0QIV1Qen3XTh7IChwW40SNfU= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 h1:2llRW4Tn9W/EZp2XvXclQ9IjeTBwwxVPrrqaerX+vCE= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f h1:BwrIaQIx5Iy6eT+DfLhFfK2XqjxRm74mVdlX8gbu4dw= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f/go.mod h1:wHtwSR3F1CQSJJZDQKuqaqFYnvkT+kMyget7dl8Clvo= github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241018134907-a00ba3729b5e h1:JiETqdNM0bktAUGMc62COwXIaw3rR3M77Me6bBLG0Fg= diff --git a/go.mod b/go.mod index 2b6f03333c0..d9a65c16063 100644 --- a/go.mod +++ b/go.mod @@ -77,7 +77,7 @@ require ( github.com/smartcontractkit/chain-selectors v1.0.29 github.com/smartcontractkit/chainlink-automation v0.8.1 github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b - github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 + github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241018134907-a00ba3729b5e github.com/smartcontractkit/chainlink-feeds v0.1.1 diff --git a/go.sum b/go.sum index 13217384ff6..885518d69aa 100644 --- a/go.sum +++ b/go.sum @@ -1078,8 +1078,8 @@ github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgB github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b h1:4kmZtaQ4fXwduHnw9xk5VmiIOW4nHg/Mx6iidlZJt5o= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b/go.mod h1:4adKaHNaxFsRvV/lYfqtbsWyyvIPUMLR0FdOJN/ljis= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 h1:vnNqMaAvheZgR8IDMGw0QIV1Qen3XTh7IChwW40SNfU= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 h1:2llRW4Tn9W/EZp2XvXclQ9IjeTBwwxVPrrqaerX+vCE= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f h1:BwrIaQIx5Iy6eT+DfLhFfK2XqjxRm74mVdlX8gbu4dw= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f/go.mod h1:wHtwSR3F1CQSJJZDQKuqaqFYnvkT+kMyget7dl8Clvo= github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241018134907-a00ba3729b5e h1:JiETqdNM0bktAUGMc62COwXIaw3rR3M77Me6bBLG0Fg= diff --git a/integration-tests/go.mod b/integration-tests/go.mod index aba17e10397..b5948028f32 100644 --- a/integration-tests/go.mod +++ b/integration-tests/go.mod @@ -37,7 +37,7 @@ require ( github.com/smartcontractkit/chain-selectors v1.0.29 github.com/smartcontractkit/chainlink-automation v0.8.1 github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b - github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 + github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 github.com/smartcontractkit/chainlink-protos/job-distributor v0.4.0 github.com/smartcontractkit/chainlink-testing-framework/havoc v1.50.2 github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.13 diff --git a/integration-tests/go.sum b/integration-tests/go.sum index 5e6793bbb0f..5a4bcb648cd 100644 --- a/integration-tests/go.sum +++ b/integration-tests/go.sum @@ -1405,8 +1405,8 @@ github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgB github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b h1:4kmZtaQ4fXwduHnw9xk5VmiIOW4nHg/Mx6iidlZJt5o= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b/go.mod h1:4adKaHNaxFsRvV/lYfqtbsWyyvIPUMLR0FdOJN/ljis= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 h1:vnNqMaAvheZgR8IDMGw0QIV1Qen3XTh7IChwW40SNfU= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 h1:2llRW4Tn9W/EZp2XvXclQ9IjeTBwwxVPrrqaerX+vCE= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f h1:BwrIaQIx5Iy6eT+DfLhFfK2XqjxRm74mVdlX8gbu4dw= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f/go.mod h1:wHtwSR3F1CQSJJZDQKuqaqFYnvkT+kMyget7dl8Clvo= github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241018134907-a00ba3729b5e h1:JiETqdNM0bktAUGMc62COwXIaw3rR3M77Me6bBLG0Fg= diff --git a/integration-tests/load/go.mod b/integration-tests/load/go.mod index c89baf21bd9..7398abc5af9 100644 --- a/integration-tests/load/go.mod +++ b/integration-tests/load/go.mod @@ -17,7 +17,7 @@ require ( github.com/pkg/errors v0.9.1 github.com/rs/zerolog v1.33.0 github.com/slack-go/slack v0.15.0 - github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 + github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.13 github.com/smartcontractkit/chainlink-testing-framework/seth v1.50.5 github.com/smartcontractkit/chainlink-testing-framework/wasp v1.50.2 diff --git a/integration-tests/load/go.sum b/integration-tests/load/go.sum index f2c309ea33a..6ca863f3d08 100644 --- a/integration-tests/load/go.sum +++ b/integration-tests/load/go.sum @@ -1394,8 +1394,8 @@ github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgB github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b h1:4kmZtaQ4fXwduHnw9xk5VmiIOW4nHg/Mx6iidlZJt5o= github.com/smartcontractkit/chainlink-ccip v0.0.0-20241112095015-3e85d9f1898b/go.mod h1:4adKaHNaxFsRvV/lYfqtbsWyyvIPUMLR0FdOJN/ljis= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371 h1:vnNqMaAvheZgR8IDMGw0QIV1Qen3XTh7IChwW40SNfU= -github.com/smartcontractkit/chainlink-common v0.3.1-0.20241113142256-8a7a997a0371/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068 h1:2llRW4Tn9W/EZp2XvXclQ9IjeTBwwxVPrrqaerX+vCE= +github.com/smartcontractkit/chainlink-common v0.3.1-0.20241114134822-aadff98ef068/go.mod h1:ny87uTW6hLjCTLiBqBRNFEhETSXhHWevYlPclT5lSco= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f h1:BwrIaQIx5Iy6eT+DfLhFfK2XqjxRm74mVdlX8gbu4dw= github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241017133723-5277829bd53f/go.mod h1:wHtwSR3F1CQSJJZDQKuqaqFYnvkT+kMyget7dl8Clvo= github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241018134907-a00ba3729b5e h1:JiETqdNM0bktAUGMc62COwXIaw3rR3M77Me6bBLG0Fg=