feat: Add initial support for cleaning orphaned jobs

compute/kubernetes/backend.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"time"
 
 	"dario.cat/mergo"
@@ -259,7 +260,7 @@ func (b *Backend) createResources(ctx context.Context, task *tes.Task, config *c
 		b.log.Debug("creating Worker PV", "taskID", task.Id)
 
 		// Check to make sure required configs are present
-		if config.GenericS3 == nil || len(config.GenericS3) == 0 ||
+		if len(config.GenericS3) == 0 ||
 			config.GenericS3[0].Bucket == "" || config.GenericS3[0].Region == "" {
 			return fmt.Errorf("Bucket or Region not found in GenericS3 config when attempting to create resources for task: %#v", task)
 		}
@@ -569,6 +570,26 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea
 	}
 }
 
+// extractTaskIDFromExecutorJobName parses the taskID from an executor job name.
+// Executor jobs are named "{taskID}-{index}" where index is a non-negative integer.
+// Returns an empty string if the name does not match the expected pattern.
+func extractTaskIDFromExecutorJobName(name string) string {
+	idx := strings.LastIndex(name, "-")
+	if idx <= 0 {
+		return ""
+	}
+	suffix := name[idx+1:]
+	for _, c := range suffix {
+		if c < '0' || c > '9' {
+			return ""
+		}
+	}
+	if len(suffix) == 0 {
+		return ""
+	}
+	return name[:idx]
+}
+
 // isResourceCleanupNeeded returns true when the task is confirmed gone (NotFound)
 // or in a terminal state.
 func (b *Backend) isResourceCleanupNeeded(ctx context.Context, taskID string) (bool, error) {
@@ -626,6 +647,19 @@ func (b *Backend) CleanOrphanedResources(ctx context.Context) {
 
 	// TODO: Add Executor Jobs here beacause orphaned tasks can result in orphaned jobs
 
+	// Executor Jobs (label app=funnel-executor; named {taskID}-{index}).
+	// These are not owned by the worker Job, so they must be discovered and cleaned explicitly.
+	executorJobs, err := b.client.BatchV1().Jobs(namespace).List(ctx, metav1.ListOptions{LabelSelector: "app=funnel-executor"})
+	if err != nil {
+		b.log.Error("backlog cleanup: listing executor jobs", err)
+	} else {
+		for _, j := range executorJobs.Items {
+			if taskID := extractTaskIDFromExecutorJobName(j.Name); taskID != "" {
+				taskIDs[taskID] = struct{}{}
+			}
+		}
+	}
+
 	for taskID := range taskIDs {
 		clean, err := b.isResourceCleanupNeeded(ctx, taskID)
 		if err != nil {

compute/kubernetes/backend_test.go

@@ -19,6 +19,25 @@ import (
 	k8stesting "k8s.io/client-go/testing"
 )
 
+// mockDatabase implements tes.ReadOnlyServer for testing CleanOrphanedResources.
+type mockDatabase struct {
+	tasks map[string]*tes.Task
+}
+
+func (m *mockDatabase) GetTask(_ context.Context, req *tes.GetTaskRequest) (*tes.Task, error) {
+	t, ok := m.tasks[req.Id]
+	if !ok {
+		return nil, fmt.Errorf("not found")
+	}
+	return t, nil
+}
+
+func (m *mockDatabase) ListTasks(_ context.Context, _ *tes.ListTasksRequest) (*tes.ListTasksResponse, error) {
+	return &tes.ListTasksResponse{}, nil
+}
+
+func (m *mockDatabase) Close() {}
+
 // noopEventWriter implements events.Writer for testing.
 type noopEventWriter struct{}
 
@@ -397,3 +416,114 @@ func TestCancel_SAInUse(t *testing.T) {
 		t.Errorf("expected SA to remain while pod is still running, got: %v", err)
 	}
 }
+
+func TestExtractTaskIDFromExecutorJobName(t *testing.T) {
+	tests := []struct {
+		name string
+		want string
+	}{
+		{"abc123-0", "abc123"},
+		{"abc123-42", "abc123"},
+		{"task-id-with-dashes-0", "task-id-with-dashes"},
+		{"abc123", ""},     // no index suffix
+		{"abc123-", ""},    // empty suffix
+		{"abc123-foo", ""}, // non-numeric suffix
+		{"-0", ""},         // empty task ID portion
+		{"", ""},
+	}
+	for _, tt := range tests {
+		got := extractTaskIDFromExecutorJobName(tt.name)
+		if got != tt.want {
+			t.Errorf("extractTaskIDFromExecutorJobName(%q) = %q, want %q", tt.name, got, tt.want)
+		}
+	}
+}
+
+// TestCleanOrphanedResources_ExecutorJobs verifies that CleanOrphanedResources
+// discovers and deletes executor jobs whose parent tasks are in a terminal state.
+func TestCleanOrphanedResources_ExecutorJobs(t *testing.T) {
+	const ns = "test-namespace"
+	const taskID = "orphaned-task"
+	ctx := context.Background()
+
+	// Executor job left behind after the worker job was already removed.
+	executorJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      taskID + "-0",
+			Namespace: ns,
+			Labels:    map[string]string{"app": "funnel-executor"},
+		},
+	}
+
+	fakeClient := fake.NewSimpleClientset(executorJob)
+
+	db := &mockDatabase{
+		tasks: map[string]*tes.Task{
+			taskID: {Id: taskID, State: tes.State_COMPLETE},
+		},
+	}
+
+	conf := config.DefaultConfig()
+	conf.Kubernetes.Namespace = ns
+	conf.Kubernetes.JobsNamespace = ns
+
+	backend := &Backend{
+		client:   fakeClient,
+		log:      logger.NewLogger("test", logger.DefaultConfig()),
+		conf:     conf,
+		event:    &noopEventWriter{},
+		database: db,
+	}
+
+	backend.CleanOrphanedResources(ctx)
+
+	// Executor job must be gone.
+	_, err := fakeClient.BatchV1().Jobs(ns).Get(ctx, taskID+"-0", metav1.GetOptions{})
+	if err == nil {
+		t.Error("expected executor job to be deleted, but it still exists")
+	}
+}
+
+// TestCleanOrphanedResources_ExecutorJobs_ActiveTask verifies that executor jobs
+// for tasks that are still active are NOT deleted.
+func TestCleanOrphanedResources_ExecutorJobs_ActiveTask(t *testing.T) {
+	const ns = "test-namespace"
+	const taskID = "active-task"
+	ctx := context.Background()
+
+	executorJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      taskID + "-0",
+			Namespace: ns,
+			Labels:    map[string]string{"app": "funnel-executor"},
+		},
+	}
+
+	fakeClient := fake.NewSimpleClientset(executorJob)
+
+	db := &mockDatabase{
+		tasks: map[string]*tes.Task{
+			taskID: {Id: taskID, State: tes.State_RUNNING},
+		},
+	}
+
+	conf := config.DefaultConfig()
+	conf.Kubernetes.Namespace = ns
+	conf.Kubernetes.JobsNamespace = ns
+
+	backend := &Backend{
+		client:   fakeClient,
+		log:      logger.NewLogger("test", logger.DefaultConfig()),
+		conf:     conf,
+		event:    &noopEventWriter{},
+		database: db,
+	}
+
+	backend.CleanOrphanedResources(ctx)
+
+	// Executor job must still be present — task is running.
+	_, err := fakeClient.BatchV1().Jobs(ns).Get(ctx, taskID+"-0", metav1.GetOptions{})
+	if err != nil {
+		t.Errorf("expected executor job to remain for running task, got: %v", err)
+	}
+}