fix(workflows/syncer): skip handling new registration if engine running

core/services/workflows/syncer/engine_registry.go

@@ -3,30 +3,36 @@ package syncer
 import (
 	"errors"
 	"sync"
-
-	"github.com/smartcontractkit/chainlink/v2/core/services/workflows"
 )
 
+// ReadyCloser is an abstraction for engines that can be checked for readiness and closed.
+type ReadyCloser interface {
+	// 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]*workflows.Engine
+	engines map[string]ReadyCloser
 	mu      sync.RWMutex
 }
 
 func newEngineRegistry() *engineRegistry {
 	return &engineRegistry{
-		engines: make(map[string]*workflows.Engine),
+		engines: make(map[string]ReadyCloser),
 	}
 }
 
 // Add adds an engine to the registry.
-func (r *engineRegistry) Add(id string, engine *workflows.Engine) {
+func (r *engineRegistry) Add(id string, engine ReadyCloser) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.engines[id] = engine
 }
 
 // Get retrieves an engine from the registry.
-func (r *engineRegistry) Get(id string) (*workflows.Engine, error) {
+func (r *engineRegistry) Get(id string) (ReadyCloser, error) {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 	engine, found := r.engines[id]
@@ -49,7 +55,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) (*workflows.Engine, error) {
+func (r *engineRegistry) Pop(id string) (ReadyCloser, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	engine, ok := r.engines[id]

core/services/workflows/syncer/handler.go

@@ -410,6 +410,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.Owner, []byte(payload.SecretsURL))
 	if err != nil {

core/services/workflows/syncer/handler_test.go

@@ -46,6 +46,30 @@ func newMockFetcher(m map[string]mockFetchResp) FetcherFunc {
 	return (&mockFetcher{responseMap: m}).Fetch
 }
 
+type mockEngine struct {
+	CloseErr error
+	ReadyErr error
+}
+
+func newMockEngine(errs ...error) *mockEngine {
+	e := &mockEngine{}
+	if len(errs) > 0 {
+		e.ReadyErr = errs[0]
+	}
+	if len(errs) > 1 {
+		e.CloseErr = errs[1]
+	}
+	return e
+}
+
+func (m *mockEngine) Ready() error {
+	return m.ReadyErr
+}
+
+func (m *mockEngine) Close() error {
+	return m.CloseErr
+}
+
 func Test_Handler(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	emitter := custmsg.NewLabeler()
@@ -179,7 +203,10 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 	var config = []byte("")
 	var wfOwner = []byte("0xOwner")
 	var binary = wasmtest.CreateTestBinary(binaryCmd, binaryLocation, true, t)
-	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)
@@ -221,6 +248,38 @@ func Test_workflowRegisteredHandler(t *testing.T) {
 			},
 			validationFn: defaultValidationFn,
 		},
+		{
+			Name: "fails if running engine exists",
+			fetcher: newMockFetcher(map[string]mockFetchResp{
+				binaryURL:  {Body: binary, 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),
+					Owner:        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 := newMockEngine()
+				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{
@@ -245,7 +304,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)
@@ -322,7 +384,7 @@ type testCase struct {
 	WFOwner      []byte
 	fetcher      FetcherFunc
 	Event        func([]byte) WorkflowRegistryWorkflowRegisteredV1
-	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)
 }
 
 func testRunningWorkflow(t *testing.T, cmd testCase) {
@@ -363,10 +425,8 @@ func testRunningWorkflow(t *testing.T, cmd testCase) {
 			capRegistry:    registry,
 			workflowStore:  store,
 		}
-		err = h.workflowRegisteredEvent(ctx, event)
-		require.NoError(t, err)
 
-		cmd.validationFn(t, ctx, h, wfOwner, "workflow-name", wfID)
+		cmd.validationFn(t, ctx, event, h, wfOwner, "workflow-name", wfID)
 	})
 }
 

core/services/workflows/syncer/workflow_registry.go

@@ -272,7 +272,7 @@ func (w *workflowRegistry) handlerLoop(ctx context.Context) {
 			event := resp.Event
 			w.lggr.Debugf("handling event: %+v", event)
 			if err := w.handler.Handle(ctx, *event); err != nil {
-				w.lggr.Errorf("failed to handle event: %+v", event)
+				w.lggr.Errorf("failed to handle event: %s : %+v", err, event)
 				continue
 			}
 		}