move the step output to a value type, fixes nil pointer exception bug…

… in step persistence code (#13416)
smartcontractkit · Jun 4, 2024 · 68e00b3 · 68e00b3
1 parent fc007a9
commit 68e00b3
Show file tree

Hide file tree

Showing 8 changed files with 29 additions and 33 deletions.
diff --git a/.changeset/kind-cobras-hope.md b/.changeset/kind-cobras-hope.md
@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+#internal fix for workflow step persistence
diff --git a/core/services/workflows/engine.go b/core/services/workflows/engine.go
@@ -413,7 +413,7 @@ func (e *Engine) startExecution(ctx context.Context, executionID string, event v
 	ec := &store.WorkflowExecution{
 		Steps: map[string]*store.WorkflowExecutionStep{
 			workflows.KeywordTrigger: {
-				Outputs: &store.StepOutput{
+				Outputs: store.StepOutput{
 					Value: event,
 				},
 				Status:      store.StatusCompleted,
@@ -562,7 +562,7 @@ func (e *Engine) workerForStepRequest(ctx context.Context, msg stepRequest) {
 
 	l.Debugw("executing on a step event")
 	stepState := &store.WorkflowExecutionStep{
-		Outputs:     &store.StepOutput{},
+		Outputs:     store.StepOutput{},
 		ExecutionID: msg.state.ExecutionID,
 		Ref:         msg.stepRef,
 	}

diff --git a/core/services/workflows/engine_test.go b/core/services/workflows/engine_test.go
@@ -544,7 +544,7 @@ func TestEngine_ResumesPendingExecutions(t *testing.T) {
 	ec := &store.WorkflowExecution{
 		Steps: map[string]*store.WorkflowExecutionStep{
 			workflows.KeywordTrigger: {
-				Outputs: &store.StepOutput{
+				Outputs: store.StepOutput{
 					Value: resp,
 				},
 				Status:      store.StatusCompleted,
@@ -599,7 +599,7 @@ func TestEngine_TimesOutOldExecutions(t *testing.T) {
 	ec := &store.WorkflowExecution{
 		Steps: map[string]*store.WorkflowExecutionStep{
 			workflows.KeywordTrigger: {
-				Outputs: &store.StepOutput{
+				Outputs: store.StepOutput{
 					Value: resp,
 				},
 				Status:      store.StatusCompleted,

diff --git a/core/services/workflows/state.go b/core/services/workflows/state.go
@@ -29,7 +29,7 @@ func copyState(es store.WorkflowExecution) store.WorkflowExecution {
 			Ref:         step.Ref,
 			Status:      step.Status,
 
-			Outputs: &store.StepOutput{
+			Outputs: store.StepOutput{
 				Err:   step.Outputs.Err,
 				Value: copiedov,
 			},

diff --git a/core/services/workflows/state_test.go b/core/services/workflows/state_test.go
@@ -38,7 +38,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: values.NewString("<a report>"),
 						},
 					},
@@ -68,7 +68,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: values.NewString("<a report>"),
 						},
 					},
@@ -82,7 +82,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Err: errors.New("catastrophic error"),
 						},
 					},
@@ -96,7 +96,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -110,7 +110,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -124,7 +124,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -138,7 +138,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -152,7 +152,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -166,7 +166,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -180,7 +180,7 @@ func TestInterpolateKey(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: val,
 						},
 					},
@@ -222,7 +222,7 @@ func TestInterpolateInputsFromState(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: values.NewString("<a report>"),
 						},
 					},
@@ -242,7 +242,7 @@ func TestInterpolateInputsFromState(t *testing.T) {
 			state: store.WorkflowExecution{
 				Steps: map[string]*store.WorkflowExecutionStep{
 					"evm_median": {
-						Outputs: &store.StepOutput{
+						Outputs: store.StepOutput{
 							Value: values.NewString("<a report>"),
 						},
 					},

diff --git a/core/services/workflows/store/models.go b/core/services/workflows/store/models.go
@@ -24,7 +24,7 @@ type WorkflowExecutionStep struct {
 	Status      string
 
 	Inputs  *values.Map
-	Outputs *StepOutput
+	Outputs StepOutput
 
 	UpdatedAt *time.Time
 }

diff --git a/core/services/workflows/store/store_db.go b/core/services/workflows/store/store_db.go
@@ -147,20 +147,15 @@ func stepToState(step workflowStepRow) (*WorkflowExecutionStep, error) {
 		outputs = values.FromProto(vProto)
 	}
 
-	var so *StepOutput
-	if outputErr != nil || outputs != nil {
-		so = &StepOutput{
-			Err:   outputErr,
-			Value: outputs,
-		}
-	}
-
 	return &WorkflowExecutionStep{
 		ExecutionID: step.WorkflowExecutionID,
 		Ref:         step.Ref,
 		Status:      step.Status,
 		Inputs:      inputs,
-		Outputs:     so,
+		Outputs: StepOutput{
+			Err:   outputErr,
+			Value: outputs,
+		},
 	}, nil
 }
 
@@ -182,10 +177,6 @@ func stateToStep(state *WorkflowExecutionStep) (workflowStepRow, error) {
 		Inputs:              inpb,
 	}
 
-	if state.Outputs == nil {
-		return wsr, nil
-	}
-
 	if state.Outputs.Value != nil {
 		p := values.Proto(state.Outputs.Value)
 		ob, err := proto.Marshal(p)

diff --git a/core/services/workflows/store/store_db_test.go b/core/services/workflows/store/store_db_test.go
@@ -153,15 +153,15 @@ func Test_StoreDB_UpdateStep(t *testing.T) {
 	require.NoError(t, err)
 
 	stepOne.Inputs = nm
-	stepOne.Outputs = &StepOutput{Err: errors.New("some error")}
+	stepOne.Outputs = StepOutput{Err: errors.New("some error")}
 
 	es, err = store.UpsertStep(tests.Context(t), stepOne)
 	require.NoError(t, err)
 
 	gotStep := es.Steps[stepOne.Ref]
 	assert.Equal(t, stepOne, gotStep)
 
-	stepTwo.Outputs = &StepOutput{Value: nm}
+	stepTwo.Outputs = StepOutput{Value: nm}
 	es, err = store.UpsertStep(tests.Context(t), stepTwo)
 	require.NoError(t, err)