fix: hook upgrade failure returns immediately instead of waiting for timeout

When a pre/post hook Job fails (e.g. BackoffLimitExceeded), the status
observer was skipping the FailedStatus resource instead of cancelling the
watch. This caused WatchUntilReady to block until the full --timeout value
expired, then return both the hook failure AND context deadline exceeded.

Helm v3 failed immediately via ready.go jobReady() which checked
job.Status.Failed > BackoffLimit. The v4 status-watcher framework reintroduced
this regression because statusObserver continued waiting after detecting failure.

Fix: call cancel() as soon as FailedStatus is observed so the watch stops
immediately, returning only the hook failure error.

Fixes #31690

pkg/kube/statuswait.go

@@ -243,10 +243,12 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl
 			if rs.Status == status.UnknownStatus && desired == status.NotFoundStatus {
 				continue
 			}
-			// Failed is a terminal state. This check ensures we don't wait forever for a resource
-			// that has already failed, as intervention is required to resolve the failure.
+			// Failed is a terminal state that cannot recover. Cancel the watch immediately
+			// so the error is returned without waiting for the context timeout to expire.
 			if rs.Status == status.FailedStatus && desired == status.CurrentStatus {
-				continue
+				logger.Debug("resource failed, cancelling wait", "namespace", rs.Identifier.Namespace, "name", rs.Identifier.Name, "kind", rs.Identifier.GroupKind.Kind)
+				cancel()
+				return
 			}
 			rss = append(rss, rs)
 			if rs.Status != desired {

pkg/kube/statuswait_test.go

@@ -1709,6 +1709,46 @@ func TestMethodContextOverridesGeneralContext(t *testing.T) {
 	})
 }
 
+func TestWatchUntilReadyFailsImmediatelyOnHookJobFailure(t *testing.T) {
+	t.Parallel()
+	// Regression test for https://github.com/helm/helm/issues/31690
+	// When a pre/post hook job fails, WatchUntilReady must return immediately
+	// with a clear error and must NOT wait for the full timeout to expire.
+	// In v3 the upgrade failed instantly with "BackoffLimitExceeded"; in v4 it
+	// incorrectly waited until context deadline exceeded before surfacing the error.
+	c := newTestClient(t)
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+	fakeMapper := testutil.NewFakeRESTMapper(
+		batchv1.SchemeGroupVersion.WithKind("Job"),
+	)
+	sw := statusWaiter{
+		client:     fakeClient,
+		restMapper: fakeMapper,
+	}
+	sw.SetLogger(slog.Default().Handler())
+
+	objs := getRuntimeObjFromManifests(t, []string{jobFailedManifest})
+	for _, obj := range objs {
+		u := obj.(*unstructured.Unstructured)
+		gvr := getGVR(t, fakeMapper, u)
+		err := fakeClient.Tracker().Create(gvr, u, u.GetNamespace())
+		require.NoError(t, err)
+	}
+	resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+
+	timeout := 30 * time.Second
+	start := time.Now()
+	err := sw.WatchUntilReady(resourceList, timeout)
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "resource Job/default/failed-job not ready. status: Failed")
+	assert.NotContains(t, err.Error(), "context deadline exceeded",
+		"WatchUntilReady should return immediately on hook failure, not wait for timeout")
+	assert.Less(t, elapsed, 5*time.Second,
+		"WatchUntilReady should return quickly on hook failure, not after the full %s timeout", timeout)
+}
+
 func TestWatchUntilReadyWithCustomReaders(t *testing.T) {
 	t.Parallel()
 	tests := []struct {