fix(executor): prevent invalid TaskActions from staying nonterminal (#7967)

fix(executor): retry failed validation status writes

Signed-off-by: 1fanwang <1fannnw@gmail.com>

Author: 1fanwang

executor/pkg/controller/taskaction_controller.go

@@ -374,9 +374,9 @@ func (r *TaskActionReconciler) reconcileTask(
 		setCondition(taskAction, flyteorgv1.ConditionTypeFailed, metav1.ConditionTrue, reason, err.Error())
 		setCondition(taskAction, flyteorgv1.ConditionTypeProgressing, metav1.ConditionFalse, reason, err.Error())
 		start := time.Now()
-		updErr := r.Status().Update(ctx, taskAction) // error intentionally ignored: terminal either way
+		updErr := r.Status().Update(ctx, taskAction)
 		r.metrics.recordK8sOp(ctx, opStatusUpdate, start, updErr)
-		return ctrl.Result{}, nil // terminal — do not requeue
+		return ctrl.Result{}, updErr
 	}
 
 	// Ensure finalizer is present (once validation passes)

executor/pkg/controller/taskaction_validation_test.go

@@ -2,10 +2,16 @@ package controller
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
 
+	"k8s.io/client-go/tools/events"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
 	flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1"
 	pluginsCore "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/core"
 )
@@ -109,3 +115,38 @@ func TestValidateTaskAction_PluginNotFound(t *testing.T) {
 		t.Errorf("expected reason %q, got %q", flyteorgv1.ConditionReasonPluginNotFound, reason)
 	}
 }
+
+func TestReconcileTask_ValidationStatusUpdate(t *testing.T) {
+	statusErr := errors.New("status update failed")
+	for _, tc := range []struct {
+		name      string
+		updateErr error
+	}{
+		{name: "success"},
+		{name: "failure", updateErr: statusErr},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			k8sClient := fake.NewClientBuilder().
+				WithInterceptorFuncs(interceptor.Funcs{
+					SubResourceUpdate: func(context.Context, client.Client, string, client.Object, ...client.SubResourceUpdateOption) error {
+						return tc.updateErr
+					},
+				}).
+				Build()
+			reconciler := &TaskActionReconciler{
+				Client:   k8sClient,
+				Recorder: events.NewFakeRecorder(1),
+			}
+			taskAction := validTaskAction()
+			taskAction.Spec.RunName = ""
+
+			_, err := reconciler.reconcileTask(context.Background(), taskAction, taskAction.DeepCopy())
+			if !errors.Is(err, tc.updateErr) {
+				t.Fatalf("expected %v, got %v", tc.updateErr, err)
+			}
+			if !isTerminal(taskAction) {
+				t.Fatal("expected validation failure to remain terminal")
+			}
+		})
+	}
+}