avoid interacting with nil mutatedWfs

pkg/controller/handler.go

@@ -248,7 +248,9 @@ func (p *Propeller) Handle(ctx context.Context, namespace, name string) error {
 		if wfClosureCrdFields != nil {
 			// strip data populated from WorkflowClosureReference
 			w.SubWorkflows, w.Tasks, w.WorkflowSpec = nil, nil, nil
-			mutatedWf.SubWorkflows, mutatedWf.Tasks, mutatedWf.WorkflowSpec = nil, nil, nil
+			if mutatedWf != nil {
+				mutatedWf.SubWorkflows, mutatedWf.Tasks, mutatedWf.WorkflowSpec = nil, nil, nil
+			}
 		}
 
 		if err != nil {

pkg/controller/handler_test.go

@@ -894,4 +894,25 @@ func TestPropellerHandler_OffloadedWorkflowClosure(t *testing.T) {
 		err := p.Handle(ctx, namespace, name)
 		assert.Error(t, err)
 	})
+
+	t.Run("TryMutate failure is handled", func(t *testing.T) {
+		scope := promutils.NewTestScope()
+
+		protoStore := &storagemocks.ComposedProtobufStore{}
+		protoStore.OnReadProtobufMatch(mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("foo"))
+		exec.HandleCb = func(ctx context.Context, w *v1alpha1.FlyteWorkflow) error {
+			return fmt.Errorf("foo")
+		}
+		dataStore := storage.NewCompositeDataStore(storage.URLPathConstructor{}, protoStore)
+		p := NewPropellerHandler(ctx, cfg, dataStore, s, exec, scope)
+
+		err := p.Handle(ctx, namespace, name)
+		assert.Error(t, err, "foo")
+
+		r, err := s.Get(ctx, namespace, name)
+		assert.NoError(t, err)
+		assert.Nil(t, r.WorkflowSpec)
+		assert.Nil(t, r.SubWorkflows)
+		assert.Nil(t, r.Tasks)
+	})
 }