fix: opt-in async watcher status computation

The ObjectStatusReporter's informer event handler was calling
readStatusFromObject synchronously, which makes API server calls
(e.g., fetching ReplicaSets/Pods for Deployments). Since client-go's
SharedIndexInformer processes notifications serially, this blocked the
entire notification pipeline for all resources of the same GroupKind.

When many resources are upgraded simultaneously (e.g., ~20 Deployments
via Helm), events queue up and each resource's status is reported with
a growing delay of 1-3+ minutes behind the API server's actual state.

This change dispatches status computation to bounded async goroutines
so the informer notification pipeline is never blocked. A sync.Map
tracks the latest resource version per object; goroutines that finish
after a newer event has arrived detect staleness and drop their result.

By default, StatusComputeWorkers <= 1 preserves the original
synchronous behavior for backward compatibility. When set to a value
> 1, the async goroutine path is used.

Key changes:
- Split eventHandler into syncEventHandler (original serial behavior)
  and asyncEventHandler (bounded concurrent goroutines), selected at
  informer startup based on StatusComputeWorkers
- AddFunc/UpdateFunc dispatch readStatusFromObject to goroutines
- Bounded concurrency via semaphore (configurable StatusComputeWorkers)
- Stale event detection via resource version tracking
- sync.WaitGroup ensures clean shutdown before closing event channels
- DeleteFunc invalidates in-flight async computations

Fixes: helm/helm#31824
Signed-off-by: maplemiao <maplemiao@tencent.com>
Made-with: Cursor

Author: mapleeit

pkg/kstatus/watcher/default_status_watcher.go

@@ -43,6 +43,14 @@ type DefaultStatusWatcher struct {
 	// required for computing parent object status, to compensate for
 	// controllers that aren't following status conventions.
 	ClusterReader engine.ClusterReader
+
+	// StatusComputeWorkers is the maximum number of concurrent goroutines
+	// used to compute object status per informer. When set to a value > 1,
+	// status computation is dispatched to async goroutines, preventing the
+	// informer notification pipeline from being blocked by slow API calls.
+	// Defaults to 0 (synchronous, same as original behavior).
+	// Set to a higher value (e.g., 8) to enable concurrent status computation.
+	StatusComputeWorkers int
 }
 
 var _ StatusWatcher = &DefaultStatusWatcher{}
@@ -88,13 +96,14 @@ func (w *DefaultStatusWatcher) Watch(ctx context.Context, ids object.ObjMetadata
 	}
 
 	informer := &ObjectStatusReporter{
-		InformerFactory: NewDynamicInformerFactory(w.DynamicClient, w.ResyncPeriod),
-		Mapper:          w.Mapper,
-		StatusReader:    w.StatusReader,
-		ClusterReader:   w.ClusterReader,
-		Targets:         targets,
-		ObjectFilter:    &AllowListObjectFilter{AllowList: ids},
-		RESTScope:       scope,
+		InformerFactory:      NewDynamicInformerFactory(w.DynamicClient, w.ResyncPeriod),
+		Mapper:               w.Mapper,
+		StatusReader:         w.StatusReader,
+		ClusterReader:        w.ClusterReader,
+		StatusComputeWorkers: w.StatusComputeWorkers,
+		Targets:              targets,
+		ObjectFilter:         &AllowListObjectFilter{AllowList: ids},
+		RESTScope:            scope,
 	}
 	return informer.Start(ctx)
 }

pkg/kstatus/watcher/default_status_watcher_test.go

@@ -5,13 +5,16 @@ package watcher
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
+	"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"
 	"github.com/fluxcd/cli-utils/pkg/object"
 	"github.com/fluxcd/cli-utils/pkg/testutil"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
@@ -888,3 +891,125 @@ func getGVR(t *testing.T, mapper meta.RESTMapper, obj *unstructured.Unstructured
 	require.NoError(t, err)
 	return mapping.Resource
 }
+
+// slowStatusReader simulates a StatusReader that takes a fixed amount of time
+// to compute status, mimicking the synchronous API calls made by the real
+// deployment/replicaset status readers when fetching generated resources.
+type slowStatusReader struct {
+	delay time.Duration
+}
+
+func (s *slowStatusReader) Supports(schema.GroupKind) bool {
+	return true
+}
+
+func (s *slowStatusReader) ReadStatus(ctx context.Context, _ engine.ClusterReader, id object.ObjMetadata) (*event.ResourceStatus, error) {
+	select {
+	case <-time.After(s.delay):
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	return &event.ResourceStatus{
+		Identifier: id,
+		Status:     status.CurrentStatus,
+		Message:    "Current",
+	}, nil
+}
+
+func (s *slowStatusReader) ReadStatusForObject(ctx context.Context, _ engine.ClusterReader, obj *unstructured.Unstructured) (*event.ResourceStatus, error) {
+	select {
+	case <-time.After(s.delay):
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	id := object.UnstructuredToObjMetadata(obj)
+	return &event.ResourceStatus{
+		Identifier: id,
+		Status:     status.CurrentStatus,
+		Resource:   obj,
+		Message:    "Current",
+	}, nil
+}
+
+// TestEventHandlerConcurrency verifies that the informer event handler does not
+// block the entire notification pipeline when status computation is slow.
+//
+// When multiple objects share the same informer (same GroupKind + namespace),
+// events should be processed concurrently rather than serially. With serial
+// processing, N objects each taking D time would result in ~N*D total latency.
+// With concurrent processing, total latency should be close to D.
+//
+// This test will FAIL before the fix (serial handler) and PASS after (async handler).
+func TestEventHandlerConcurrency(t *testing.T) {
+	const numPods = 10
+	const statusDelay = 100 * time.Millisecond
+
+	fakeMapper := testutil.NewFakeRESTMapper(
+		v1.SchemeGroupVersion.WithKind("Pod"),
+	)
+
+	pods := make([]*unstructured.Unstructured, numPods)
+	ids := make(object.ObjMetadataSet, numPods)
+	for i := 0; i < numPods; i++ {
+		pod := &unstructured.Unstructured{}
+		pod.SetGroupVersionKind(v1.SchemeGroupVersion.WithKind("Pod"))
+		pod.SetNamespace("default")
+		pod.SetName(fmt.Sprintf("pod-%d", i))
+		pods[i] = pod
+		ids[i] = object.UnstructuredToObjMetadata(pod)
+	}
+
+	podGVR := getGVR(t, fakeMapper, pods[0])
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+
+	statusWatcher := NewDefaultStatusWatcher(fakeClient, fakeMapper)
+	statusWatcher.StatusReader = &slowStatusReader{delay: statusDelay}
+	statusWatcher.StatusComputeWorkers = 8
+
+	eventCh := statusWatcher.Watch(ctx, ids, Options{})
+
+	// Wait for the SyncEvent indicating the informer has started watching.
+	syncReceived := false
+	for e := range eventCh {
+		if e.Type == event.SyncEvent {
+			syncReceived = true
+			break
+		}
+	}
+	require.True(t, syncReceived, "expected SyncEvent before any resource events")
+
+	// Create all pods at once to queue up events in the informer.
+	for _, pod := range pods {
+		require.NoError(t, fakeClient.Tracker().Create(podGVR, pod.DeepCopy(), pod.GetNamespace()))
+	}
+
+	// Collect all ResourceUpdateEvents and measure total elapsed time.
+	start := time.Now()
+	received := 0
+	for received < numPods {
+		select {
+		case e, ok := <-eventCh:
+			if !ok {
+				t.Fatalf("event channel closed after receiving %d/%d events", received, numPods)
+			}
+			if e.Type == event.ResourceUpdateEvent {
+				received++
+			}
+		case <-ctx.Done():
+			t.Fatalf("timed out waiting for events: received %d/%d", received, numPods)
+		}
+	}
+	elapsed := time.Since(start)
+	cancel()
+
+	serialTime := time.Duration(numPods) * statusDelay
+	t.Logf("Received %d events in %v (serial estimate: %v)", numPods, elapsed, serialTime)
+
+	assert.Less(t, elapsed, serialTime/2,
+		"Event handler appears to be blocking serially: took %v, expected well under %v for concurrent processing",
+		elapsed, serialTime/2)
+}