fix(workflows/syncer): check that no engine is running before registering new one

core/services/relay/evm/capabilities/workflows/syncer/workflow_syncer_test.go

@@ -17,7 +17,6 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
-	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/services/servicetest"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows"
@@ -317,15 +316,11 @@ func (m *mockService) Start(context.Context) error { return nil }
 
 func (m *mockService) Close() error { return nil }
 
-func (m *mockService) HealthReport() map[string]error { return map[string]error{"svc": nil} }
-
 func (m *mockService) Ready() error { return nil }
 
-func (m *mockService) Name() string { return "svc" }
-
 type mockEngineFactory struct{}
 
-func (m *mockEngineFactory) new(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (services.Service, error) {
+func (m *mockEngineFactory) new(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (syncer.StartReadyCloser, error) {
 	return &mockService{}, nil
 }
 

core/services/workflows/syncer/engine_registry.go

@@ -1,32 +1,42 @@
 package syncer
 
 import (
+	"context"
 	"errors"
 	"sync"
-
-	"github.com/smartcontractkit/chainlink-common/pkg/services"
 )
 
+// StartReadyCloser is an abstraction for engines that can be checked for readiness and closed.
+type StartReadyCloser interface {
+	Start(context.Context) error
+
+	// Ready returns nil if the engine is ready to be used.
+	Ready() error
+
+	// Close closes the engine.
+	Close() error
+}
+
 type EngineRegistry struct {
-	engines map[string]services.Service
+	engines map[string]StartReadyCloser
 	mu      sync.RWMutex
 }
 
 func NewEngineRegistry() *EngineRegistry {
 	return &EngineRegistry{
-		engines: make(map[string]services.Service),
+		engines: make(map[string]StartReadyCloser),
 	}
 }
 
 // Add adds an engine to the registry.
-func (r *EngineRegistry) Add(id string, engine services.Service) {
+func (r *EngineRegistry) Add(id string, engine StartReadyCloser) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.engines[id] = engine
 }
 
 // Get retrieves an engine from the registry.
-func (r *EngineRegistry) Get(id string) (services.Service, error) {
+func (r *EngineRegistry) Get(id string) (StartReadyCloser, error) {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 	engine, found := r.engines[id]
@@ -49,7 +59,7 @@ func (r *EngineRegistry) IsRunning(id string) bool {
 }
 
 // Pop removes an engine from the registry and returns the engine if found.
-func (r *EngineRegistry) Pop(id string) (services.Service, error) {
+func (r *EngineRegistry) Pop(id string) (StartReadyCloser, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	engine, ok := r.engines[id]

core/services/workflows/syncer/handler.go

@@ -14,7 +14,6 @@ import (
 	"github.com/jonboulle/clockwork"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
-	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
 	pkgworkflows "github.com/smartcontractkit/chainlink-common/pkg/workflows"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows/secrets"
@@ -127,7 +126,7 @@ func newLastFetchedAtMap() *lastFetchedAtMap {
 	}
 }
 
-type engineFactoryFn func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (services.Service, error)
+type engineFactoryFn func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (StartReadyCloser, error)
 
 // eventHandler is a handler for WorkflowRegistryEvent events.  Each event type has a corresponding
 // method that handles the event.
@@ -438,6 +437,11 @@ func (h *eventHandler) workflowRegisteredEvent(
 		return fmt.Errorf("workflowID mismatch: %x != %x", hash, payload.WorkflowID)
 	}
 
+	// Ensure that there is no running workflow engine for the given workflow ID.
+	if h.engineRegistry.IsRunning(hex.EncodeToString(payload.WorkflowID[:])) {
+		return fmt.Errorf("workflow is already running, so not starting it : %s", hex.EncodeToString(payload.WorkflowID[:]))
+	}
+
 	// Save the workflow secrets
 	urlHash, err := h.orm.GetSecretsURLHash(payload.WorkflowOwner, []byte(payload.SecretsURL))
 	if err != nil {
@@ -493,7 +497,7 @@ func (h *eventHandler) workflowRegisteredEvent(
 	return nil
 }
 
-func (h *eventHandler) engineFactoryFn(ctx context.Context, id string, owner string, name string, config []byte, binary []byte) (services.Service, error) {
+func (h *eventHandler) engineFactoryFn(ctx context.Context, id string, owner string, name string, config []byte, binary []byte) (StartReadyCloser, error) {
 	moduleConfig := &host.ModuleConfig{Logger: h.lggr, Labeler: h.emitter}
 	sdkSpec, err := host.GetWorkflowSpec(ctx, moduleConfig, binary, config)
 	if err != nil {

core/services/workflows/syncer/handler_test.go

@@ -47,6 +47,24 @@ func newMockFetcher(m map[string]mockFetchResp) FetcherFunc {
 	return (&mockFetcher{responseMap: m}).Fetch
 }
 
+type mockEngine struct {
+	CloseErr error
+	ReadyErr error
+	StartErr error
+}
+
+func (m *mockEngine) Ready() error {
+	return m.ReadyErr
+}
+
+func (m *mockEngine) Close() error {
+	return m.CloseErr
+}
+
+func (m *mockEngine) Start(_ context.Context) error {
+	return m.StartErr
+}
+
 func Test_Handler(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	emitter := custmsg.NewLabeler()
@@ -181,7 +199,11 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 	var wfOwner = []byte("0xOwner")
 	var binary = wasmtest.CreateTestBinary(binaryCmd, binaryLocation, true, t)
 	var encodedBinary = []byte(base64.StdEncoding.EncodeToString(binary))
-	defaultValidationFn := func(t *testing.T, ctx context.Context, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+
+	defaultValidationFn := func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+		err := h.workflowRegisteredEvent(ctx, event)
+		require.NoError(t, err)
+
 		// Verify the record is updated in the database
 		dbSpec, err := h.orm.GetWorkflowSpec(ctx, hex.EncodeToString(wfOwner), "workflow-name")
 		require.NoError(t, err)
@@ -204,6 +226,9 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 				configURL:  {Body: config, Err: nil},
 				secretsURL: {Body: []byte("secrets"), Err: nil},
 			}),
+			engineFactoryFn: func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (StartReadyCloser, error) {
+				return &mockEngine{}, nil
+			},
 			GiveConfig: config,
 			ConfigURL:  configURL,
 			SecretsURL: secretsURL,
@@ -223,6 +248,71 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 			},
 			validationFn: defaultValidationFn,
 		},
+		{
+			Name: "fails to start engine",
+			fetcher: newMockFetcher(map[string]mockFetchResp{
+				binaryURL:  {Body: encodedBinary, Err: nil},
+				configURL:  {Body: config, Err: nil},
+				secretsURL: {Body: []byte("secrets"), Err: nil},
+			}),
+			engineFactoryFn: func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (StartReadyCloser, error) {
+				return &mockEngine{StartErr: assert.AnError}, nil
+			},
+			GiveConfig: config,
+			ConfigURL:  configURL,
+			SecretsURL: secretsURL,
+			BinaryURL:  binaryURL,
+			GiveBinary: binary,
+			WFOwner:    wfOwner,
+			Event: func(wfID []byte) WorkflowRegistryWorkflowRegisteredV1 {
+				return WorkflowRegistryWorkflowRegisteredV1{
+					Status:        uint8(0),
+					WorkflowID:    [32]byte(wfID),
+					WorkflowOwner: wfOwner,
+					WorkflowName:  "workflow-name",
+					BinaryURL:     binaryURL,
+					ConfigURL:     configURL,
+					SecretsURL:    secretsURL,
+				}
+			},
+			validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+				err := h.workflowRegisteredEvent(ctx, event)
+				require.Error(t, err)
+				require.ErrorIs(t, err, assert.AnError)
+			},
+		},
+		{
+			Name: "fails if running engine exists",
+			fetcher: newMockFetcher(map[string]mockFetchResp{
+				binaryURL:  {Body: encodedBinary, Err: nil},
+				configURL:  {Body: config, Err: nil},
+				secretsURL: {Body: []byte("secrets"), Err: nil},
+			}),
+			GiveConfig: config,
+			ConfigURL:  configURL,
+			SecretsURL: secretsURL,
+			BinaryURL:  binaryURL,
+			GiveBinary: binary,
+			WFOwner:    wfOwner,
+			Event: func(wfID []byte) WorkflowRegistryWorkflowRegisteredV1 {
+				return WorkflowRegistryWorkflowRegisteredV1{
+					Status:        uint8(0),
+					WorkflowID:    [32]byte(wfID),
+					WorkflowOwner: wfOwner,
+					WorkflowName:  "workflow-name",
+					BinaryURL:     binaryURL,
+					ConfigURL:     configURL,
+					SecretsURL:    secretsURL,
+				}
+			},
+			validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+				me := &mockEngine{}
+				h.engineRegistry.Add(wfID, me)
+				err := h.workflowRegisteredEvent(ctx, event)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "workflow is already running")
+			},
+		},
 		{
 			Name: "success with paused workflow registered",
 			fetcher: newMockFetcher(map[string]mockFetchResp{
@@ -247,7 +337,10 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 					SecretsURL:    secretsURL,
 				}
 			},
-			validationFn: func(t *testing.T, ctx context.Context, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+			validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string) {
+				err := h.workflowRegisteredEvent(ctx, event)
+				require.NoError(t, err)
+
 				// Verify the record is updated in the database
 				dbSpec, err := h.orm.GetWorkflowSpec(ctx, hex.EncodeToString(wfOwner), "workflow-name")
 				require.NoError(t, err)
@@ -315,62 +408,58 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 }
 
 type testCase struct {
-	Name         string
-	SecretsURL   string
-	BinaryURL    string
-	GiveBinary   []byte
-	GiveConfig   []byte
-	ConfigURL    string
-	WFOwner      []byte
-	fetcher      FetcherFunc
-	Event        func([]byte) WorkflowRegistryWorkflowRegisteredV1
-	validationFn func(t *testing.T, ctx context.Context, h *eventHandler, wfOwner []byte, wfName string, wfID string)
+	Name            string
+	SecretsURL      string
+	BinaryURL       string
+	GiveBinary      []byte
+	GiveConfig      []byte
+	ConfigURL       string
+	WFOwner         []byte
+	fetcher         FetcherFunc
+	Event           func([]byte) WorkflowRegistryWorkflowRegisteredV1
+	validationFn    func(t *testing.T, ctx context.Context, event WorkflowRegistryWorkflowRegisteredV1, h *eventHandler, wfOwner []byte, wfName string, wfID string)
+	engineFactoryFn func(ctx context.Context, wfid string, owner string, name string, config []byte, binary []byte) (StartReadyCloser, error)
 }
 
-func testRunningWorkflow(t *testing.T, cmd testCase) {
+func testRunningWorkflow(t *testing.T, tc testCase) {
 	t.Helper()
-	t.Run(cmd.Name, func(t *testing.T) {
+	t.Run(tc.Name, func(t *testing.T) {
 		var (
 			ctx     = testutils.Context(t)
 			lggr    = logger.TestLogger(t)
 			db      = pgtest.NewSqlxDB(t)
 			orm     = NewWorkflowRegistryDS(db, lggr)
 			emitter = custmsg.NewLabeler()
 
-			binary     = cmd.GiveBinary
-			config     = cmd.GiveConfig
-			secretsURL = cmd.SecretsURL
-			wfOwner    = cmd.WFOwner
+			binary     = tc.GiveBinary
+			config     = tc.GiveConfig
+			secretsURL = tc.SecretsURL
+			wfOwner    = tc.WFOwner
 
-			fetcher = cmd.fetcher
+			fetcher = tc.fetcher
 		)
 
 		giveWFID, err := pkgworkflows.GenerateWorkflowID(wfOwner, binary, config, secretsURL)
 		require.NoError(t, err)
 
 		wfID := hex.EncodeToString(giveWFID[:])
 
-		event := cmd.Event(giveWFID[:])
+		event := tc.Event(giveWFID[:])
 
 		er := NewEngineRegistry()
+		opts := []func(*eventHandler){
+			WithEngineRegistry(er),
+		}
+		if tc.engineFactoryFn != nil {
+			opts = append(opts, WithEngineFactoryFn(tc.engineFactoryFn))
+		}
 		store := wfstore.NewDBStore(db, lggr, clockwork.NewFakeClock())
 		registry := capabilities.NewRegistry(lggr)
 		registry.SetLocalRegistry(&capabilities.TestMetadataRegistry{})
-		h := NewEventHandler(
-			lggr,
-			orm,
-			fetcher,
-			store,
-			registry,
-			emitter,
-			clockwork.NewFakeClock(),
-			workflowkey.Key{},
-			WithEngineRegistry(er),
-		)
-		err = h.workflowRegisteredEvent(ctx, event)
-		require.NoError(t, err)
+		h := NewEventHandler(lggr, orm, fetcher, store, registry, emitter, clockwork.NewFakeClock(),
+			workflowkey.Key{}, opts...)
 
-		cmd.validationFn(t, ctx, h, wfOwner, "workflow-name", wfID)
+		tc.validationFn(t, ctx, event, h, wfOwner, "workflow-name", wfID)
 	})
 }