fix(kube): check resource existence before waiting for deletion

WaitForDelete treated a resource as already gone whenever the kstatus
watcher reported it as Unknown. But the watcher reports Unknown in two
cases: a resource that is genuinely gone, which never emits a delete
event, and a resource that still exists but whose status has not synced
yet. Treating both as deleted let the watch cancel before any real
status arrived, which is the TestStatusWaitForDelete flake in #32261.

The earlier fix for that flake (#32081) deferred that decision, which
made already-gone resources wait the full timeout (the hang in #32214).
It was reverted in v4.2.2, so the flake returned.

Check the cluster directly instead. Before watching, a resource that
returns NotFound from a live GET is already gone and is skipped, and the
observer no longer treats an Unknown status as deleted. Already-gone and
never-created resources, such as before-hook-creation hooks, return
immediately, while a resource that still exists is waited on until it is
actually deleted or the timeout is reached.

closes #32261
refs #32214
refs #32216

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

pkg/kube/statuswait.go

@@ -33,7 +33,9 @@ import (
 	"github.com/fluxcd/cli-utils/pkg/kstatus/watcher"
 	"github.com/fluxcd/cli-utils/pkg/object"
 	appsv1 "k8s.io/api/apps/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/dynamic"
 	watchtools "k8s.io/client-go/tools/watch"
@@ -144,8 +146,15 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
 		if err != nil {
 			return err
 		}
+		if w.isResourceGone(ctx, obj) {
+			w.Logger().Debug("resource already deleted", "kind", obj.GroupKind.Kind, "namespace", obj.Namespace, "name", obj.Name)
+			continue
+		}
 		resources = append(resources, obj)
 	}
+	if len(resources) == 0 {
+		return nil
+	}
 	eventCh := sw.Watch(cancelCtx, resources, watcher.Options{
 		RESTScopeStrategy: watcher.RESTScopeNamespace,
 	})
@@ -175,6 +184,21 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
 	return nil
 }
 
+// isResourceGone reports whether the resource is already absent from the cluster.
+// The watcher only reports NotFound from an observed delete event, so an
+// already-deleted resource would otherwise be waited on until the timeout.
+func (w *statusWaiter) isResourceGone(ctx context.Context, id object.ObjMetadata) bool {
+	mapping, err := w.restMapper.RESTMapping(id.GroupKind)
+	if err != nil {
+		return false
+	}
+	_, err = w.client.Resource(mapping.Resource).Namespace(id.Namespace).Get(ctx, id.Name, metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		w.Logger().Debug("could not check whether resource is deleted, will wait on it", "kind", id.GroupKind.Kind, "namespace", id.Namespace, "name", id.Name, "error", err)
+	}
+	return apierrors.IsNotFound(err)
+}
+
 func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw watcher.StatusWatcher) error {
 	cancelCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -244,11 +268,6 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl
 			if rs == nil {
 				continue
 			}
-			// If a resource is already deleted before waiting has started, it will show as unknown.
-			// This check ensures we don't wait forever for a resource that is already deleted.
-			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.
 			if rs.Status == status.FailedStatus && desired == status.CurrentStatus {

pkg/kube/statuswait_test.go

@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/collector"
 	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/engine"
 	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/event"
 	"github.com/fluxcd/cli-utils/pkg/kstatus/status"
@@ -362,7 +363,8 @@ func TestStatusWaitForDelete(t *testing.T) {
 func TestStatusWaitForDeleteNonExistentObject(t *testing.T) {
 	t.Parallel()
 	c := newTestClient(t)
-	timeout := time.Second
+	// Generous timeout so the "returns promptly" assertion below has wide headroom.
+	timeout := 2 * time.Second
 	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
 	fakeMapper := testutil.NewFakeRESTMapper(
 		v1.SchemeGroupVersion.WithKind("Pod"),
@@ -372,11 +374,108 @@ func TestStatusWaitForDeleteNonExistentObject(t *testing.T) {
 		client:     fakeClient,
 	}
 	statusWaiter.SetLogger(slog.Default().Handler())
-	// Don't create the object to test that the wait for delete works when the object doesn't exist
+	// Regression guard for #32214: a never-created resource must return promptly,
+	// not wait the full timeout.
 	objManifest := getRuntimeObjFromManifests(t, []string{podCurrentManifest})
 	resourceList := getResourceListFromRuntimeObjs(t, c, objManifest)
+	start := time.Now()
 	err := statusWaiter.WaitForDelete(resourceList, timeout)
 	assert.NoError(t, err)
+	assert.Less(t, time.Since(start), timeout/2, "a never-created resource should return promptly, not wait the full timeout")
+}
+
+func TestStatusObserverDoesNotCancelOnUnknown(t *testing.T) {
+	t.Parallel()
+	// While the informer is still syncing, resources briefly show as Unknown. A delete
+	// wait must not read that transient state as "already deleted" and cancel the watch
+	// before any real status arrives. See https://github.com/helm/helm/issues/32261.
+	ids := object.ObjMetadataSet{
+		{GroupKind: schema.GroupKind{Kind: "Pod"}, Namespace: "ns", Name: "first"},
+		{GroupKind: schema.GroupKind{Kind: "Pod"}, Namespace: "ns", Name: "second"},
+	}
+	statusCollector := collector.NewResourceStatusCollector(ids)
+	var cancelled atomic.Bool
+	observer := statusObserver(func() { cancelled.Store(true) }, status.NotFoundStatus, slog.Default())
+	observer.Notify(statusCollector, event.Event{})
+	assert.False(t, cancelled.Load(), "watch was cancelled while resources were still Unknown")
+}
+
+func TestStatusWaitForDeleteAlreadyDeleted(t *testing.T) {
+	t.Parallel()
+	c := newTestClient(t)
+	timeout := 2 * time.Second
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+	fakeMapper := testutil.NewFakeRESTMapper(
+		v1.SchemeGroupVersion.WithKind("Pod"),
+	)
+	statusWaiter := statusWaiter{
+		restMapper: fakeMapper,
+		client:     fakeClient,
+	}
+	statusWaiter.SetLogger(slog.Default().Handler())
+	// A before-hook-creation hook is deleted by Helm and only then waited on, so it is
+	// already gone here; it must return promptly rather than hang (#32214).
+	objs := getRuntimeObjFromManifests(t, []string{podCurrentManifest})
+	for _, obj := range objs {
+		u := obj.(*unstructured.Unstructured)
+		gvr := getGVR(t, fakeMapper, u)
+		require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+		require.NoError(t, fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName()))
+	}
+	resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+	start := time.Now()
+	err := statusWaiter.WaitForDelete(resourceList, timeout)
+	assert.NoError(t, err)
+	assert.Less(t, time.Since(start), timeout/2, "an already-deleted resource should not be waited on")
+}
+
+func TestIsResourceGone(t *testing.T) {
+	t.Parallel()
+	podGVK := v1.SchemeGroupVersion.WithKind("Pod")
+	id := object.ObjMetadata{GroupKind: podGVK.GroupKind(), Namespace: "ns", Name: "current-pod"}
+	tests := []struct {
+		name     string
+		setup    func(*dynamicfake.FakeDynamicClient)
+		wantGone bool
+	}{
+		{
+			name:     "missing object is gone",
+			setup:    func(*dynamicfake.FakeDynamicClient) {},
+			wantGone: true,
+		},
+		{
+			name: "existing object is not gone",
+			setup: func(c *dynamicfake.FakeDynamicClient) {
+				pod := getRuntimeObjFromManifests(t, []string{podCurrentManifest})[0].(*unstructured.Unstructured)
+				require.NoError(t, c.Tracker().Create(v1.SchemeGroupVersion.WithResource("pods"), pod, "ns"))
+			},
+			wantGone: false,
+		},
+		{
+			// A GET that fails for any reason other than NotFound must not be read as gone,
+			// otherwise a transient API error would skip a resource that still exists.
+			name: "get error is not treated as gone",
+			setup: func(c *dynamicfake.FakeDynamicClient) {
+				c.PrependReactor("get", "pods", func(clienttesting.Action) (bool, runtime.Object, error) {
+					return true, nil, errors.New("connection refused")
+				})
+			},
+			wantGone: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+			tt.setup(fakeClient)
+			sw := statusWaiter{
+				restMapper: testutil.NewFakeRESTMapper(podGVK),
+				client:     fakeClient,
+			}
+			sw.SetLogger(slog.Default().Handler())
+			assert.Equal(t, tt.wantGone, sw.isResourceGone(context.Background(), id))
+		})
+	}
 }
 
 func TestStatusWait(t *testing.T) {