feat(executor): support immediate GC cleanup when MaxTTL <= 0

executor/pkg/config/config.go

@@ -84,7 +84,8 @@ type GCConfig struct {
 	Interval stdconfig.Duration `json:"interval" pflag:",How often the garbage collector runs. 0 disables GC."`
 
 	// MaxTTL is the time-to-live for terminal TaskActions before deletion.
-	MaxTTL stdconfig.Duration `json:"maxTTL" pflag:",Time-to-live for terminal TaskActions before deletion."`
+	// A value <= 0 means terminal TaskActions are deleted immediately on the next GC cycle.
+	MaxTTL stdconfig.Duration `json:"maxTTL" pflag:",Time-to-live for terminal TaskActions before deletion. Use 0 or negative for immediate deletion."`
 }
 
 // GetConfig returns the parsed executor configuration

executor/pkg/controller/garbage_collector.go

@@ -55,7 +55,14 @@ const gcPageSize = 500
 func (gc *GarbageCollector) collect(ctx context.Context) error {
 	logger := log.FromContext(ctx).WithName("gc")
 
-	cutoff := time.Now().UTC().Add(-gc.maxTTL).Format(labelTimeFormat)
+	var cutoff string
+	if gc.maxTTL <= 0 {
+		// MaxTTL <= 0 means immediate deletion: use a far-future cutoff so all terminal
+		// TaskActions are eligible regardless of their completed time.
+		cutoff = "9999-12-31.23-59"
+	} else {
+		cutoff = time.Now().UTC().Add(-gc.maxTTL).Format(labelTimeFormat)
+	}
 	deleted := 0
 	total := 0
 	continueToken := ""

executor/pkg/controller/garbage_collector_test.go

@@ -102,6 +102,38 @@ var _ = Describe("GarbageCollector", func() {
 		gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour)
 		Expect(gc.collect(ctx)).To(Succeed())
 	})
+
+	It("should delete all terminal TaskActions immediately when maxTTL is zero", func() {
+		recentTime := time.Now().UTC().Format(labelTimeFormat)
+		createTaskAction(ctx, "gc-zero-ttl", map[string]string{
+			LabelTerminationStatus: LabelValueTerminated,
+			LabelCompletedTime:     recentTime,
+		})
+
+		gc := NewGarbageCollector(k8sClient, 1*time.Minute, 0)
+		Expect(gc.collect(ctx)).To(Succeed())
+
+		ta := &flyteorgv1.TaskAction{}
+		err := k8sClient.Get(ctx, types.NamespacedName{Name: "gc-zero-ttl", Namespace: "default"}, ta)
+		Expect(err).To(HaveOccurred())
+		Expect(client.IgnoreNotFound(err)).To(Succeed())
+	})
+
+	It("should delete all terminal TaskActions immediately when maxTTL is negative", func() {
+		recentTime := time.Now().UTC().Format(labelTimeFormat)
+		createTaskAction(ctx, "gc-negative-ttl", map[string]string{
+			LabelTerminationStatus: LabelValueTerminated,
+			LabelCompletedTime:     recentTime,
+		})
+
+		gc := NewGarbageCollector(k8sClient, 1*time.Minute, -1*time.Hour)
+		Expect(gc.collect(ctx)).To(Succeed())
+
+		ta := &flyteorgv1.TaskAction{}
+		err := k8sClient.Get(ctx, types.NamespacedName{Name: "gc-negative-ttl", Namespace: "default"}, ta)
+		Expect(err).To(HaveOccurred())
+		Expect(client.IgnoreNotFound(err)).To(Succeed())
+	})
 })
 
 var _ = Describe("ensureTerminalLabels", func() {

executor/setup.go

@@ -122,9 +122,6 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 		}
 
 		if cfg.GC.Interval.Duration > 0 {
-			if cfg.GC.MaxTTL.Duration <= 0 {
-				return fmt.Errorf("executor: gc.maxTTL must be positive when gc is enabled, got %v", cfg.GC.MaxTTL.Duration)
-			}
 			gc := controller.NewGarbageCollector(mgr.GetClient(), cfg.GC.Interval.Duration, cfg.GC.MaxTTL.Duration)
 			if err := mgr.Add(gc); err != nil {
 				return fmt.Errorf("executor: failed to add garbage collector: %w", err)