[CAPPL-257] GetUnfinished should filter by workflowID (#15152)

core/services/workflows/engine.go

@@ -357,7 +357,7 @@ var (
 )
 
 func (e *Engine) resumeInProgressExecutions(ctx context.Context) error {
-	wipExecutions, err := e.executionStates.GetUnfinished(ctx, defaultOffset, defaultLimit)
+	wipExecutions, err := e.executionStates.GetUnfinished(ctx, e.workflow.id, defaultOffset, defaultLimit)
 	if err != nil {
 		return err
 	}

core/services/workflows/engine_test.go

@@ -35,7 +35,6 @@ import (
 	p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types"
 	"github.com/smartcontractkit/chainlink/v2/core/services/registrysyncer"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/store"
-	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer"
 )
 
 const testWorkflowId = "<workflow-id>"
@@ -149,6 +148,12 @@ func newTestEngineWithYAMLSpec(t *testing.T, reg *coreCap.Registry, spec string,
 	return newTestEngine(t, reg, sdkSpec, opts...)
 }
 
+type mockSecretsFetcher struct{}
+
+func (s mockSecretsFetcher) SecretsFor(workflowOwner, workflowName string) (map[string]string, error) {
+	return map[string]string{}, nil
+}
+
 // newTestEngine creates a new engine with some test defaults.
 func newTestEngine(t *testing.T, reg *coreCap.Registry, sdkSpec sdk.WorkflowSpec, opts ...func(c *Config)) (*Engine, *testHooks) {
 	initFailed := make(chan struct{})
@@ -174,7 +179,7 @@ func newTestEngine(t *testing.T, reg *coreCap.Registry, sdkSpec sdk.WorkflowSpec
 		onExecutionFinished: func(weid string) {
 			executionFinished <- weid
 		},
-		SecretsFetcher: syncer.NewWorkflowRegistry(),
+		SecretsFetcher: mockSecretsFetcher{},
 		clock:          clock,
 	}
 	for _, o := range opts {

core/services/workflows/store/store.go

@@ -9,7 +9,7 @@ type Store interface {
 	UpsertStep(ctx context.Context, step *WorkflowExecutionStep) (WorkflowExecution, error)
 	UpdateStatus(ctx context.Context, executionID string, status string) error
 	Get(ctx context.Context, executionID string) (WorkflowExecution, error)
-	GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error)
+	GetUnfinished(ctx context.Context, workflowID string, offset, limit int) ([]WorkflowExecution, error)
 }
 
 var _ Store = (*DBStore)(nil)

core/services/workflows/store/store_db.go

@@ -29,12 +29,12 @@ type DBStore struct {
 // `workflowExecutionRow` describes a row
 // of the `workflow_executions` table
 type workflowExecutionRow struct {
-	ID         string
-	WorkflowID *string
-	Status     string
-	CreatedAt  *time.Time
-	UpdatedAt  *time.Time
-	FinishedAt *time.Time
+	ID         string     `db:"id"`
+	WorkflowID *string    `db:"workflow_id"`
+	Status     string     `db:"status"`
+	CreatedAt  *time.Time `db:"created_at"`
+	UpdatedAt  *time.Time `db:"updated_at"`
+	FinishedAt *time.Time `db:"finished_at"`
 }
 
 // `workflowStepRow` describes a row
@@ -362,7 +362,7 @@ func (d *DBStore) transact(ctx context.Context, fn func(*DBStore) error) error {
 	)
 }
 
-func (d *DBStore) GetUnfinished(ctx context.Context, offset, limit int) ([]WorkflowExecution, error) {
+func (d *DBStore) GetUnfinished(ctx context.Context, workflowID string, offset, limit int) ([]WorkflowExecution, error) {
 	sql := `
 	SELECT
 		workflow_steps.workflow_execution_id AS ws_workflow_execution_id,
@@ -382,12 +382,13 @@ func (d *DBStore) GetUnfinished(ctx context.Context, offset, limit int) ([]Workf
 	JOIN workflow_steps
 	ON  workflow_steps.workflow_execution_id = workflow_executions.id
 	WHERE workflow_executions.status = $1
+	AND workflow_executions.workflow_id = $2
 	ORDER BY workflow_executions.created_at DESC
-	LIMIT $2
-	OFFSET $3
+	LIMIT $3
+	OFFSET $4
 	`
 	var joinRecords []workflowExecutionWithStep
-	err := d.db.SelectContext(ctx, &joinRecords, sql, StatusStarted, limit, offset)
+	err := d.db.SelectContext(ctx, &joinRecords, sql, StatusStarted, workflowID, limit, offset)
 	if err != nil {
 		return []WorkflowExecution{}, err
 	}

core/services/workflows/store/store_db_test.go

@@ -12,13 +12,18 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
+	"github.com/smartcontractkit/chainlink/v2/core/services/job"
 
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 )
 
 func randomID() string {
-	b := make([]byte, 32)
+	return random(32)
+}
+
+func random(length int) string {
+	b := make([]byte, length)
 	_, err := rand.Read(b)
 	if err != nil {
 		panic(err)
@@ -234,10 +239,24 @@ func Test_StoreDB_WorkflowStepStatus(t *testing.T) {
 	}
 }
 
+func createWorkflow(t *testing.T, store *DBStore, id string) {
+	sql := `INSERT INTO workflow_specs (workflow, workflow_id, workflow_owner, workflow_name, created_at, updated_at)
+	VALUES (:workflow, :workflow_id, :workflow_owner, :workflow_name, NOW(), NOW())`
+	var wfSpec job.WorkflowSpec
+	wfSpec.Workflow = ""
+	wfSpec.WorkflowID = id
+	wfSpec.WorkflowOwner = random(20)
+	wfSpec.WorkflowName = random(20)
+	_, err := store.db.NamedExecContext(tests.Context(t), sql, wfSpec)
+	require.NoError(t, err)
+}
+
 func Test_StoreDB_GetUnfinishedSteps(t *testing.T) {
 	store := newTestDBStore(t)
 
 	id := randomID()
+	wid := randomID()
+	createWorkflow(t, store, wid)
 	stepOne := &WorkflowExecutionStep{
 		ExecutionID: id,
 		Ref:         "step1",
@@ -254,12 +273,29 @@ func Test_StoreDB_GetUnfinishedSteps(t *testing.T) {
 			"step2": stepTwo,
 		},
 		ExecutionID: id,
+		WorkflowID:  wid,
 		Status:      StatusStarted,
 	}
 
 	_, err := store.Add(tests.Context(t), &es)
 	require.NoError(t, err)
 
+	id2 := randomID()
+	wid2 := randomID()
+	createWorkflow(t, store, wid2)
+	es2 := WorkflowExecution{
+		Steps: map[string]*WorkflowExecutionStep{
+			"step1": stepOne,
+			"step2": stepTwo,
+		},
+		ExecutionID: id2,
+		WorkflowID:  wid2,
+		Status:      StatusStarted,
+	}
+
+	_, err = store.Add(tests.Context(t), &es2)
+	require.NoError(t, err)
+
 	id = randomID()
 	esTwo := WorkflowExecution{
 		ExecutionID: id,
@@ -269,7 +305,7 @@ func Test_StoreDB_GetUnfinishedSteps(t *testing.T) {
 	_, err = store.Add(tests.Context(t), &esTwo)
 	require.NoError(t, err)
 
-	states, err := store.GetUnfinished(tests.Context(t), 0, 100)
+	states, err := store.GetUnfinished(tests.Context(t), wid, 0, 100)
 	require.NoError(t, err)
 
 	assert.Len(t, states, 1)