fix: fallback to live GET when kstatus watcher misses deletion events

During `helm uninstall --wait --cascade foreground`, the kstatus watcher
can intermittently miss deletion events for resources, leaving them
stuck in Terminating status until the timeout expires. This is a race
between the Kubernetes garbage collector removing the foregroundDeletion
finalizer and the informer delivering the corresponding watch event.

Add a fallback verification step in waitForDelete: before reporting a
resource as still existing, issue a live GET to the API server. If the
resource returns 404 (NotFound), treat it as successfully deleted. When
every resource the watcher missed is confirmed deleted this way and none
is in Unknown status, the watcher timeout is no longer returned as an
error.

The caller's deadline has usually passed by then, so the GETs keep the
caller's context values but share one 10-second grace period. The
fallback is skipped if the caller cancelled. GET errors other than
NotFound are reported as "unable to verify deletion" rather than being
treated as the resource still existing.

This only affects the error path; the happy path (watcher observes
deletions normally) is unchanged.

Signed-off-by: Jathavedhan M <jathavedhan.m@ibm.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"
@@ -62,6 +64,10 @@ type statusWaiter struct {
 // when they don't set a timeout.
 var DefaultStatusWatcherTimeout = 30 * time.Second
 
+// deleteVerificationTimeout bounds the live GETs used to confirm deletions that
+// the status watcher did not observe before WaitForDelete gave up.
+const deleteVerificationTimeout = 10 * time.Second
+
 func alwaysReady(_ *unstructured.Unstructured) (*status.Result, error) {
 	return &status.Result{
 		Status:  status.CurrentStatus,
@@ -162,16 +168,50 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
 	}
 
 	errs := []error{}
+	// A cancelled caller context signals an external interruption, so only
+	// fall back to live GETs when the watcher stopped on its own or timed out.
+	verifyWithGet := !errors.Is(ctx.Err(), context.Canceled)
+	// ctx has usually expired by now, so keep its values but not its deadline,
+	// and bound all fallback GETs together by a single grace period.
+	getCtx, cancelGet := context.WithTimeout(context.WithoutCancel(ctx), deleteVerificationTimeout)
+	defer cancelGet()
+	unknown, confirmedGone := 0, 0
 	for _, id := range resources {
 		rs := statusCollector.ResourceStatuses[id]
-		if rs.Status == status.NotFoundStatus || rs.Status == status.UnknownStatus {
+		if rs.Status == status.NotFoundStatus {
+			continue
+		}
+		if rs.Status == status.UnknownStatus {
+			unknown++
 			continue
 		}
+		// The watcher may have missed the deletion event (e.g. due to a
+		// connection drop or informer lag). Verify with a live GET before
+		// reporting the resource as still existing.
+		if verifyWithGet {
+			gone, err := w.isResourceGone(getCtx, id)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("resource %s/%s/%s may still exist. status: %s, message: %s: unable to verify deletion: %w",
+					rs.Identifier.GroupKind.Kind, rs.Identifier.Namespace, rs.Identifier.Name, rs.Status, rs.Message, err))
+				continue
+			}
+			if gone {
+				w.Logger().Debug("watcher reported resource as existing but live GET confirms deletion",
+					"kind", id.GroupKind.Kind, "namespace", id.Namespace, "name", id.Name)
+				confirmedGone++
+				continue
+			}
+		}
 		errs = append(errs, fmt.Errorf("resource %s/%s/%s still exists. status: %s, message: %s",
 			rs.Identifier.GroupKind.Kind, rs.Identifier.Namespace, rs.Identifier.Name, rs.Status, rs.Message))
 	}
 	if err := ctx.Err(); err != nil {
-		errs = append(errs, err)
+		// The watcher timing out is not a failure when every resource it did not
+		// see deleted was confirmed gone by a live GET.
+		missedDeletesOnly := errors.Is(err, context.DeadlineExceeded) && confirmedGone > 0 && unknown == 0 && len(errs) == 0
+		if !missedDeletesOnly {
+			errs = append(errs, err)
+		}
 	}
 	if len(errs) > 0 {
 		return errors.Join(errs...)
@@ -179,6 +219,21 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
 	return nil
 }
 
+// isResourceGone reports whether a live GET returns NotFound for the resource.
+// Any other GET error is returned so that callers do not mistake an inability to
+// verify deletion for the resource still existing.
+func (w *statusWaiter) isResourceGone(ctx context.Context, id object.ObjMetadata) (bool, error) {
+	mapping, err := w.restMapper.RESTMapping(id.GroupKind)
+	if err != nil {
+		return false, err
+	}
+	_, err = w.client.Resource(mapping.Resource).Namespace(id.Namespace).Get(ctx, id.Name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		return true, nil
+	}
+	return false, err
+}
+
 func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw watcher.StatusWatcher) error {
 	cancelCtx, cancel := context.WithCancel(ctx)
 	defer cancel()

pkg/kube/statuswait_test.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"log/slog"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -375,6 +376,158 @@ func TestStatusWaitForDeleteNonExistentObject(t *testing.T) {
 	assert.NoError(t, statusWaiter.WaitForDelete(resourceList, timeout))
 }
 
+func TestWaitForDeleteWithMissedWatchEvent(t *testing.T) {
+	t.Parallel()
+	c := newTestClient(t)
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+	fakeMapper := testutil.NewFakeRESTMapper(
+		v1.SchemeGroupVersion.WithKind("Pod"),
+	)
+	// Return a watcher with no events to simulate a missed deletion notification.
+	// The watch starts after the initial list, so signal that the watcher has
+	// seen the resource before deleting it.
+	watchStarted := make(chan struct{})
+	var once sync.Once
+	fakeClient.PrependWatchReactor("pods", func(_ clienttesting.Action) (bool, watch.Interface, error) {
+		once.Do(func() { close(watchStarted) })
+		return true, watch.NewFake(), nil
+	})
+	sw := statusWaiter{
+		restMapper: fakeMapper,
+		client:     fakeClient,
+	}
+	sw.SetLogger(slog.Default().Handler())
+	objs := getRuntimeObjFromManifests(t, []string{podCurrentManifest})
+	gvrs := make([]schema.GroupVersionResource, 0, len(objs))
+	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)
+		gvrs = append(gvrs, gvr)
+	}
+	// Delete the resource once the watch has started. The watcher will miss
+	// this deletion because watch events are suppressed, but a live GET should
+	// confirm the resource is gone.
+	deleteErrs := make(chan error, 1)
+	go func() {
+		<-watchStarted
+		var errs []error
+		for i, obj := range objs {
+			u := obj.(*unstructured.Unstructured)
+			errs = append(errs, fakeClient.Tracker().Delete(gvrs[i], u.GetNamespace(), u.GetName()))
+		}
+		deleteErrs <- errors.Join(errs...)
+	}()
+	resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+	err := sw.WaitForDelete(resourceList, 500*time.Millisecond)
+	require.NoError(t, <-deleteErrs)
+	assert.NoError(t, err)
+}
+
+func TestWaitForDeleteWithMissedWatchEventGetError(t *testing.T) {
+	t.Parallel()
+	c := newTestClient(t)
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+	fakeMapper := testutil.NewFakeRESTMapper(
+		v1.SchemeGroupVersion.WithKind("Pod"),
+	)
+	fakeClient.PrependWatchReactor("pods", func(_ clienttesting.Action) (bool, watch.Interface, error) {
+		return true, watch.NewFake(), nil
+	})
+	// The live GET cannot verify the deletion, so the resource must not be
+	// reported as simply still existing, nor as deleted.
+	fakeClient.PrependReactor("get", "pods", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+		return true, nil, apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "current-pod", errors.New("get not allowed"))
+	})
+	sw := statusWaiter{
+		restMapper: fakeMapper,
+		client:     fakeClient,
+	}
+	sw.SetLogger(slog.Default().Handler())
+	objs := getRuntimeObjFromManifests(t, []string{podCurrentManifest})
+	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)
+	err := sw.WaitForDelete(resourceList, 500*time.Millisecond)
+	require.Error(t, err)
+	assert.True(t, apierrors.IsForbidden(err), "expected the GET error to be wrapped, got: %v", err)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Contains(t, err.Error(), "resource Pod/ns/current-pod may still exist. status: Current")
+	assert.Contains(t, err.Error(), "unable to verify deletion")
+}
+
+func TestIsResourceGone(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		getErr    error
+		exists    bool
+		expected  bool
+		expectErr func(error) bool
+	}{
+		{
+			name:     "not found",
+			expected: true,
+		},
+		{
+			name:   "still exists",
+			exists: true,
+		},
+		{
+			name:      "forbidden",
+			getErr:    apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "current-pod", errors.New("get not allowed")),
+			expectErr: apierrors.IsForbidden,
+		},
+		{
+			name:      "timeout",
+			getErr:    context.DeadlineExceeded,
+			expectErr: func(err error) bool { return errors.Is(err, context.DeadlineExceeded) },
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+			fakeMapper := testutil.NewFakeRESTMapper(
+				v1.SchemeGroupVersion.WithKind("Pod"),
+			)
+			if tt.getErr != nil {
+				fakeClient.PrependReactor("get", "pods", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+					return true, nil, tt.getErr
+				})
+			}
+			sw := statusWaiter{
+				restMapper: fakeMapper,
+				client:     fakeClient,
+			}
+			sw.SetLogger(slog.Default().Handler())
+			if tt.exists {
+				u := getRuntimeObjFromManifests(t, []string{podCurrentManifest})[0].(*unstructured.Unstructured)
+				err := fakeClient.Tracker().Create(getGVR(t, fakeMapper, u), u, u.GetNamespace())
+				require.NoError(t, err)
+			}
+			id := object.ObjMetadata{
+				GroupKind: v1.SchemeGroupVersion.WithKind("Pod").GroupKind(),
+				Namespace: "ns",
+				Name:      "current-pod",
+			}
+			gone, err := sw.isResourceGone(t.Context(), id)
+			if tt.expectErr != nil {
+				require.Error(t, err)
+				assert.True(t, tt.expectErr(err), "unexpected error: %v", err)
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Equal(t, tt.expected, gone)
+		})
+	}
+}
+
 func TestStatusWait(t *testing.T) {
 	t.Parallel()
 	tests := []struct {