Enhancement: Add a ticker to check job completion as a fallback

pkg/clients/kube/kube.go

@@ -104,6 +104,9 @@ func (c *Client) GetJobLogs(ctx context.Context, jobName string) (string, error)
 	return string(logs), nil
 }
 
+// errJobIncomplete is a sentinel error to indicate a job is still in progress
+var errJobIncomplete = fmt.Errorf("job incomplete")
+
 // WaitForJobCompletion waits for a job to complete (succeed or fail). It's context-aware,
 // so set ctx to something like `context.WithTimeout(ctx, 10*time.Second)` to set a timeout.
 func (c *Client) WaitForJobCompletion(ctx context.Context, jobName string) error {
@@ -113,22 +116,49 @@ func (c *Client) WaitForJobCompletion(ctx context.Context, jobName string) error
 	}
 	defer watcher.Stop()
 
+	// Add a ticker to periodically check job status directly as a fallback
+	// This handles cases where watch events are delayed, missing, or incomplete
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	checkJobCompletion := func(job *batchv1.Job) error {
+		// Rely on Kubernetes' own logic via job conditions
+		jobStatus := getJobStatus(job)
+		if jobStatus == batchv1.JobComplete || jobStatus == batchv1.JobSuccessCriteriaMet {
+			return nil
+		}
+		if jobStatus == batchv1.JobFailed || jobStatus == batchv1.JobFailureTarget {
+			return fmt.Errorf("job %s failed", jobName)
+		}
+
+		// Job not yet complete
+		return errJobIncomplete
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
 			return fmt.Errorf("context cancelled while waiting for job %s to complete", jobName)
-		case event := <-watcher.ResultChan():
-			job, ok := event.Object.(*batchv1.Job)
+		case <-ticker.C:
+			// Polling fallback: directly query job status
+			currentJob, err := c.GetJob(ctx, jobName)
+			if err != nil {
+				continue
+			}
+			if err := checkJobCompletion(currentJob); err != errJobIncomplete {
+				return err
+			}
+		case event, ok := <-watcher.ResultChan():
 			if !ok {
+				// Watch channel closed, fall back to polling
 				continue
 			}
-
-			jobStatus := getJobStatus(job)
-			if jobStatus == batchv1.JobComplete || jobStatus == batchv1.JobSuccessCriteriaMet {
-				return nil
+			job, isJob := event.Object.(*batchv1.Job)
+			if !isJob {
+				continue
 			}
-			if jobStatus == batchv1.JobFailed || jobStatus == batchv1.JobFailureTarget {
-				return fmt.Errorf("job %s failed", jobName)
+			if err := checkJobCompletion(job); err != errJobIncomplete {
+				return err
 			}
 		}
 	}

pkg/clients/kube/kube_test.go

@@ -3,7 +3,9 @@ package kube
 import (
 	"context"
 	"reflect"
+	"strings"
 	"testing"
+	"time"
 
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -309,3 +311,121 @@ func getMockedJob(name string) *batchv1.Job {
 		},
 	}
 }
+
+func TestClient_WaitForJobCompletion(t *testing.T) {
+	tests := []struct {
+		name           string
+		jobName        string
+		setupJob       func(*fak8s.Clientset) error
+		wantErr        bool
+		errContains    string
+		timeoutSeconds int
+	}{
+		{
+			name:    "job completes with JobComplete condition",
+			jobName: "test-job-condition-complete",
+			setupJob: func(clientset *fak8s.Clientset) error {
+				job := getMockedJob("test-job-condition-complete")
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   batchv1.JobComplete,
+						Status: corev1.ConditionTrue,
+					},
+				}
+				_, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{})
+				return err
+			},
+			wantErr:        false,
+			timeoutSeconds: 15,
+		},
+		{
+			name:    "job fails with JobFailed condition",
+			jobName: "test-job-condition-failed",
+			setupJob: func(clientset *fak8s.Clientset) error {
+				job := getMockedJob("test-job-condition-failed")
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   batchv1.JobFailed,
+						Status: corev1.ConditionTrue,
+					},
+				}
+				_, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{})
+				return err
+			},
+			wantErr:        true,
+			errContains:    "failed",
+			timeoutSeconds: 15,
+		},
+		{
+			name:    "job completes with JobSuccessCriteriaMet condition",
+			jobName: "test-job-success-criteria-met",
+			setupJob: func(clientset *fak8s.Clientset) error {
+				job := getMockedJob("test-job-success-criteria-met")
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   batchv1.JobSuccessCriteriaMet,
+						Status: corev1.ConditionTrue,
+					},
+				}
+				_, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{})
+				return err
+			},
+			wantErr:        false,
+			timeoutSeconds: 15,
+		},
+		{
+			name:    "job fails with JobFailureTarget condition",
+			jobName: "test-job-failure-target",
+			setupJob: func(clientset *fak8s.Clientset) error {
+				job := getMockedJob("test-job-failure-target")
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   batchv1.JobFailureTarget,
+						Status: corev1.ConditionTrue,
+					},
+				}
+				_, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{})
+				return err
+			},
+			wantErr:        true,
+			errContains:    "failed",
+			timeoutSeconds: 15,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a new mock clientset for each test
+			mockedClientset := fak8s.NewClientset()
+			setupJobMockingReactors(mockedClientset)
+
+			// Setup the job
+			if err := tt.setupJob(mockedClientset); err != nil {
+				t.Fatalf("Failed to setup job: %v", err)
+			}
+
+			c := &Client{
+				clientset: mockedClientset,
+				namespace: testNamespace,
+			}
+
+			// Create context with timeout
+			ctx, cancel := context.WithTimeout(context.Background(), time.Duration(tt.timeoutSeconds)*time.Second)
+			defer cancel()
+
+			err := c.WaitForJobCompletion(ctx, tt.jobName)
+
+			if tt.wantErr {
+				if err == nil {
+					t.Errorf("WaitForJobCompletion() expected error but got nil")
+				} else if tt.errContains != "" && !strings.Contains(err.Error(), tt.errContains) {
+					t.Errorf("WaitForJobCompletion() error = %v, want error containing %v", err, tt.errContains)
+				}
+			} else {
+				if err != nil {
+					t.Errorf("WaitForJobCompletion() unexpected error = %v", err)
+				}
+			}
+		})
+	}
+}