[CAPPL-735] Engine V2 execution phase (minimal impl)

Author: bolekk

core/services/workflows/engine.go

@@ -2,8 +2,6 @@ package workflows
 
 import (
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
 	"errors"
 	"fmt"
 	"strconv"
@@ -494,21 +492,6 @@ func (e *Engine) stepUpdateLoop(ctx context.Context, executionID string, stepUpd
 	}
 }
 
-func generateExecutionID(workflowID, eventID string) (string, error) {
-	s := sha256.New()
-	_, err := s.Write([]byte(workflowID))
-	if err != nil {
-		return "", err
-	}
-
-	_, err = s.Write([]byte(eventID))
-	if err != nil {
-		return "", err
-	}
-
-	return hex.EncodeToString(s.Sum(nil)), nil
-}
-
 // startExecution kicks off a new workflow execution when a trigger event is received.
 func (e *Engine) startExecution(ctx context.Context, executionID string, triggerID string, event *values.Map) error {
 	e.meterReports.Add(executionID, NewMeteringReport())
@@ -741,7 +724,7 @@ func (e *Engine) worker(ctx context.Context) {
 				continue
 			}
 
-			executionID, err := generateExecutionID(e.workflow.id, te.ID)
+			executionID, err := types.GenerateExecutionID(e.workflow.id, te.ID)
 			if err != nil {
 				e.logger.With(platform.KeyTriggerID, te.ID).Errorf("could not generate execution ID: %v", err)
 				continue

core/services/workflows/types/execution.go

@@ -0,0 +1,23 @@
+package types
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
+
+// hash of (workflowID, triggerEventID)
+// TODO(CAPPL-838): improve for V2
+func GenerateExecutionID(workflowID, triggerEventID string) (string, error) {
+	s := sha256.New()
+	_, err := s.Write([]byte(workflowID))
+	if err != nil {
+		return "", err
+	}
+
+	_, err = s.Write([]byte(triggerEventID))
+	if err != nil {
+		return "", err
+	}
+
+	return hex.EncodeToString(s.Sum(nil)), nil
+}

core/services/workflows/v2/config.go

@@ -42,8 +42,10 @@ const (
 	defaultMaxTriggerSubscriptions             = 10
 	defaultTriggerEventQueueSize               = 1000
 
-	defaultMaxConcurrentWorkflowExecutions = 100
-	defaultMaxConcurrentCapabilityCalls    = 10
+	defaultMaxConcurrentWorkflowExecutions         = 100
+	defaultMaxConcurrentCapabilityCallsPerWorkflow = 10
+	defaultWorkflowExecutionTimeoutMs              = 1000 * 60 * 10 // 10 minutes
+	defaultCapabilityCallTimeoutMs                 = 1000 * 60 * 8  // 8 minutes
 
 	defaultShutdownTimeoutMs = 5000
 )
@@ -57,8 +59,10 @@ type EngineLimits struct {
 	MaxTriggerSubscriptions             uint16
 	TriggerEventQueueSize               uint16
 
-	MaxConcurrentWorkflowExecutions uint16
-	MaxConcurrentCapabilityCalls    uint16
+	MaxConcurrentWorkflowExecutions         uint16
+	MaxConcurrentCapabilityCallsPerWorkflow uint16
+	WorkflowExecutionTimeoutMs              uint32
+	CapabilityCallTimeoutMs                 uint32
 
 	ShutdownTimeoutMs uint32
 }
@@ -133,8 +137,14 @@ func (l *EngineLimits) setDefaultLimits() {
 	if l.MaxConcurrentWorkflowExecutions == 0 {
 		l.MaxConcurrentWorkflowExecutions = defaultMaxConcurrentWorkflowExecutions
 	}
-	if l.MaxConcurrentCapabilityCalls == 0 {
-		l.MaxConcurrentCapabilityCalls = defaultMaxConcurrentCapabilityCalls
+	if l.MaxConcurrentCapabilityCallsPerWorkflow == 0 {
+		l.MaxConcurrentCapabilityCallsPerWorkflow = defaultMaxConcurrentCapabilityCallsPerWorkflow
+	}
+	if l.WorkflowExecutionTimeoutMs == 0 {
+		l.WorkflowExecutionTimeoutMs = defaultWorkflowExecutionTimeoutMs
+	}
+	if l.CapabilityCallTimeoutMs == 0 {
+		l.CapabilityCallTimeoutMs = defaultCapabilityCallTimeoutMs
 	}
 	if l.ShutdownTimeoutMs == 0 {
 		l.ShutdownTimeoutMs = defaultShutdownTimeoutMs

core/services/workflows/v2/engine.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -12,6 +13,9 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 
+	cappb "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb"
+	sdkpb "github.com/smartcontractkit/chainlink-common/pkg/workflows/sdk/v2/pb"
+	"github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host"
 	wasmpb "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/v2/pb"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/internal"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/types"
@@ -31,13 +35,18 @@ type Engine struct {
 
 	allTriggerEventsQueueCh chan enqueuedTriggerEvent
 	executionsSemaphore     chan struct{}
+	capCallsSemaphore       chan struct{}
 }
 
 type enqueuedTriggerEvent struct {
-	event     capabilities.TriggerResponse
-	timestamp time.Time
+	triggerCapID string
+	triggerIndex int
+	timestamp    time.Time
+	event        capabilities.TriggerResponse
 }
 
+var _ host.CapabilityExecutor = (*Engine)(nil)
+
 func NewEngine(ctx context.Context, cfg *EngineConfig) (*Engine, error) {
 	err := cfg.Validate()
 	if err != nil {
@@ -48,6 +57,7 @@ func NewEngine(ctx context.Context, cfg *EngineConfig) (*Engine, error) {
 		triggers:                make(map[string]capabilities.TriggerCapability),
 		allTriggerEventsQueueCh: make(chan enqueuedTriggerEvent, cfg.LocalLimits.TriggerEventQueueSize),
 		executionsSemaphore:     make(chan struct{}, cfg.LocalLimits.MaxConcurrentWorkflowExecutions),
+		capCallsSemaphore:       make(chan struct{}, cfg.LocalLimits.MaxConcurrentCapabilityCallsPerWorkflow),
 	}
 	engine.Service, engine.srvcEng = services.Config{
 		Name:  "WorkflowEngineV2",
@@ -102,7 +112,15 @@ func (e *Engine) init(ctx context.Context) {
 		return
 	}
 
-	err := e.runTriggerSubscriptionPhase(ctx)
+	err := e.cfg.Module.SetCapabilityExecutor(e)
+	if err != nil {
+		e.cfg.Lggr.Errorw("Workflow Engine initialization failed", "err", err)
+		// TODO(CAPPL-736): observability
+		e.cfg.Hooks.OnInitialized(err)
+		return
+	}
+
+	err = e.runTriggerSubscriptionPhase(ctx)
 	if err != nil {
 		e.cfg.Lggr.Errorw("Workflow Engine initialization failed", "err", err)
 		// TODO(CAPPL-736): observability
@@ -185,7 +203,7 @@ func (e *Engine) runTriggerSubscriptionPhase(ctx context.Context) error {
 	}
 
 	// start listening for trigger events only if all registrations succeeded
-	for _, triggerEventCh := range eventChans {
+	for idx, triggerEventCh := range eventChans {
 		e.srvcEng.Go(func(srvcCtx context.Context) {
 			for {
 				select {
@@ -197,8 +215,10 @@ func (e *Engine) runTriggerSubscriptionPhase(ctx context.Context) error {
 					}
 					select {
 					case e.allTriggerEventsQueueCh <- enqueuedTriggerEvent{
-						event:     event,
-						timestamp: e.cfg.Clock.Now(),
+						triggerCapID: subs.Subscriptions[idx].Id,
+						triggerIndex: idx,
+						timestamp:    e.cfg.Clock.Now(),
+						event:        event,
 					}:
 					default: // queue full, drop the event
 						// TODO(CAPPL-736): observability
@@ -216,15 +236,15 @@ func (e *Engine) handleAllTriggerEvents(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case queueElem, isOpen := <-e.allTriggerEventsQueueCh:
+		case queueHead, isOpen := <-e.allTriggerEventsQueueCh:
 			if !isOpen {
 				return
 			}
 			// TODO(CAPPL-737): check if expired
 			select {
 			case e.executionsSemaphore <- struct{}{}: // block if too many concurrent workflow executions
 				e.srvcEng.Go(func(srvcCtx context.Context) {
-					e.startNewWorkflowExecution(srvcCtx, queueElem.event)
+					e.startExecution(srvcCtx, queueHead)
 					<-e.executionsSemaphore
 				})
 			case <-ctx.Done():
@@ -234,8 +254,64 @@ func (e *Engine) handleAllTriggerEvents(ctx context.Context) {
 	}
 }
 
-func (e *Engine) startNewWorkflowExecution(_ context.Context, _ capabilities.TriggerResponse) {
-	// TODO(CAPPL-735): implement execution phase
+// new workflow execution, blocking until completed
+func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueuedTriggerEvent) {
+	triggerEvent := wrappedTriggerEvent.event.Event
+	executionID, err := types.GenerateExecutionID(e.cfg.WorkflowID, triggerEvent.ID)
+	if err != nil {
+		// TODO(CAPPL-736): observability
+		return
+	}
+
+	subCtx, cancel := context.WithTimeout(ctx, time.Millisecond*time.Duration(e.cfg.LocalLimits.WorkflowExecutionTimeoutMs))
+	defer cancel()
+	result, err := e.cfg.Module.Execute(subCtx, &wasmpb.ExecuteRequest{
+		Id: executionID,
+		Request: &wasmpb.ExecuteRequest_Trigger{
+			Trigger: &sdkpb.Trigger{
+				Id:      strconv.FormatInt(int64(wrappedTriggerEvent.triggerIndex), 10), // TODO: change to an integer oncce proto is refactored
+				Payload: triggerEvent.Payload,
+			},
+		},
+		MaxResponseSize: uint64(e.cfg.LocalLimits.ModuleExecuteMaxResponseSizeBytes),
+		// no Config needed
+	})
+	if err != nil {
+		e.cfg.Lggr.Errorw("Workflow execution failed", "err", err)
+		// TODO(CAPPL-736): observability
+		return
+	}
+	// TODO(CAPPL-736): handle execution result
+	e.cfg.Lggr.Debugw("Workflow execution finished", "executionID", executionID, "result", result)
+	e.cfg.Hooks.OnExecutionFinished(executionID)
+}
+
+// blocking
+func (e *Engine) CallCapability(ctx context.Context, request *cappb.CapabilityRequest) (*cappb.CapabilityResponse, error) {
+	select {
+	case e.capCallsSemaphore <- struct{}{}: // block if too many concurrent capability calls
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	defer func() { <-e.capCallsSemaphore }()
+
+	// TODO (CAPPL-735): use request.Metadata.WorkflowExecutionId to associate the call with a specific execution
+	capability, err := e.cfg.CapRegistry.GetExecutable(ctx, request.CapabilityId)
+	if err != nil {
+		return nil, fmt.Errorf("trigger capability not found: %w", err)
+	}
+
+	capReq, err := cappb.CapabilityRequestFromProto(request)
+	if err != nil {
+		return nil, fmt.Errorf("failed to convert capability request: %w", err)
+	}
+
+	// TODO(CAPPL-737): run with a timeout
+	capResp, err := capability.Execute(ctx, capReq)
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute capability: %w", err)
+	}
+	return cappb.CapabilityResponseToProto(capResp), nil
 }
 
 func (e *Engine) close() error {

core/services/workflows/v2/engine_test.go

@@ -42,6 +42,7 @@ func TestEngine_Init(t *testing.T) {
 	require.NoError(t, err)
 
 	module.EXPECT().Start().Once()
+	module.EXPECT().SetCapabilityExecutor(mock.Anything).Return(nil).Once()
 	module.EXPECT().Execute(matches.AnyContext, mock.Anything).Return(newTriggerSubs(0), nil).Once()
 	capreg.EXPECT().LocalNode(matches.AnyContext).Return(capabilities.Node{}, nil).Once()
 	require.NoError(t, engine.Start(t.Context()))
@@ -62,6 +63,7 @@ func TestEngine_Start_RateLimited(t *testing.T) {
 
 	module := modulemocks.NewModuleV2(t)
 	module.EXPECT().Start()
+	module.EXPECT().SetCapabilityExecutor(mock.Anything).Return(nil)
 	module.EXPECT().Execute(matches.AnyContext, mock.Anything).Return(newTriggerSubs(0), nil).Times(2)
 	module.EXPECT().Close()
 	capreg := regmocks.NewCapabilitiesRegistry(t)
@@ -123,6 +125,7 @@ func TestEngine_TriggerSubscriptions(t *testing.T) {
 
 	module := modulemocks.NewModuleV2(t)
 	module.EXPECT().Start()
+	module.EXPECT().SetCapabilityExecutor(mock.Anything).Return(nil)
 	module.EXPECT().Close()
 	capreg := regmocks.NewCapabilitiesRegistry(t)
 	capreg.EXPECT().LocalNode(matches.AnyContext).Return(capabilities.Node{}, nil)
@@ -215,3 +218,61 @@ func newTriggerSubs(n int) *wasmpb.ExecutionResult {
 		},
 	}
 }
+
+func TestEngine_Execution(t *testing.T) {
+	t.Parallel()
+
+	module := modulemocks.NewModuleV2(t)
+	module.EXPECT().Start()
+	module.EXPECT().SetCapabilityExecutor(mock.Anything).Return(nil)
+	module.EXPECT().Close()
+	capreg := regmocks.NewCapabilitiesRegistry(t)
+	capreg.EXPECT().LocalNode(matches.AnyContext).Return(capabilities.Node{}, nil)
+
+	initDoneCh := make(chan error)
+	subscribedToTriggersCh := make(chan []string, 1)
+	executionFinishedCh := make(chan string)
+
+	cfg := defaultTestConfig(t)
+	cfg.Module = module
+	cfg.CapRegistry = capreg
+	cfg.Hooks = v2.LifecycleHooks{
+		OnInitialized: func(err error) {
+			initDoneCh <- err
+		},
+		OnSubscribedToTriggers: func(triggerIDs []string) {
+			subscribedToTriggersCh <- triggerIDs
+		},
+		OnExecutionFinished: func(executionID string) {
+			executionFinishedCh <- executionID
+		},
+	}
+
+	t.Run("successful execution with no capability calls", func(t *testing.T) {
+		engine, err := v2.NewEngine(t.Context(), cfg)
+		require.NoError(t, err)
+		module.EXPECT().Execute(matches.AnyContext, mock.Anything).Return(newTriggerSubs(1), nil).Once()
+		trigger := capmocks.NewTriggerCapability(t)
+		capreg.EXPECT().GetTrigger(matches.AnyContext, "id_0").Return(trigger, nil).Once()
+		eventCh := make(chan capabilities.TriggerResponse)
+		trigger.EXPECT().RegisterTrigger(matches.AnyContext, mock.Anything).Return(eventCh, nil).Once()
+		trigger.EXPECT().UnregisterTrigger(matches.AnyContext, mock.Anything).Return(nil).Once()
+
+		require.NoError(t, engine.Start(t.Context()))
+
+		require.NoError(t, <-initDoneCh) // successful trigger registration
+		require.Equal(t, []string{"id_0"}, <-subscribedToTriggersCh)
+
+		module.EXPECT().Execute(matches.AnyContext, mock.Anything).Return(nil, nil).Once()
+		eventCh <- capabilities.TriggerResponse{
+			Event: capabilities.TriggerEvent{
+				TriggerType: "[email protected]",
+				ID:          "event_012345",
+				Payload:     nil,
+			},
+		}
+		<-executionFinishedCh
+
+		require.NoError(t, engine.Close())
+	})
+}