feat: seed UI collection watches from the list resourceVersion

pkg/server/kubernetes/client.go

@@ -93,9 +93,6 @@ func setOptionsDefaults(opts ClientOptions) (ClientOptions, error) {
 			return opts, fmt.Errorf("error adding Kargo API to scheme: %w", err)
 		}
 	}
-	if opts.NewInternalClient == nil {
-		opts.NewInternalClient = newDefaultInternalClient
-	}
 	if opts.KargoNamespace == "" {
 		opts.KargoNamespace = "kargo"
 	}
@@ -124,11 +121,20 @@ type Client interface {
 	// client. This is useful for cases where the API server needs to bypass
 	// the extra authorization checks performed by this client.
 	InternalClient() libClient.WithWatch
+
+	// APIReader returns an uncached reader that reads straight from the
+	// Kubernetes API server, bypassing the internal client's cache. It is the
+	// same reader the cache delegates to (the cluster's GetAPIReader), and is
+	// useful where a cache-served list ResourceVersion would be too old to seed
+	// a follow-up watch. Like InternalClient, it does NOT enforce Kargo's RBAC,
+	// so callers must authorize the user before using it.
+	APIReader() libClient.Reader
 }
 
 // client implements Client.
 type client struct {
 	internalClient libClient.WithWatch
+	apiReader      libClient.Reader
 	opts           ClientOptions
 
 	getAuthorizedClientFn func(
@@ -162,17 +168,44 @@ func NewClient(
 	if opts, err = setOptionsDefaults(opts); err != nil {
 		return nil, fmt.Errorf("error setting client options defaults: %w", err)
 	}
-	internalClient, err := opts.NewInternalClient(
-		ctx,
-		restCfg,
-		opts.Scheme,
-		opts.KargoNamespace,
+	var (
+		internalClient libClient.WithWatch
+		apiReader      libClient.Reader
 	)
-	if err != nil {
-		return nil, fmt.Errorf("error building internal client: %w", err)
+	if opts.NewInternalClient != nil {
+		// A custom internal client (typically a fake in tests) has no separate
+		// cache to bypass, so it doubles as the uncached API reader.
+		if internalClient, err = opts.NewInternalClient(
+			ctx,
+			restCfg,
+			opts.Scheme,
+			opts.KargoNamespace,
+		); err != nil {
+			return nil, fmt.Errorf("error building internal client: %w", err)
+		}
+		apiReader = internalClient
+	} else {
+		// The cluster provides both the cache-backed client and the uncached
+		// reader the cache delegates to (GetAPIReader) -- the same direct-read
+		// pattern the controllers and external webhooks server use.
+		var cluster libCluster.Cluster
+		if cluster, err = newDefaultCluster(
+			ctx,
+			restCfg,
+			opts.Scheme,
+			opts.KargoNamespace,
+		); err != nil {
+			return nil, fmt.Errorf("error building internal cluster: %w", err)
+		}
+		var ok bool
+		if internalClient, ok = cluster.GetClient().(libClient.WithWatch); !ok {
+			return nil, errors.New("internal client does not implement WithWatch")
+		}
+		apiReader = cluster.GetAPIReader()
 	}
 	c := &client{
 		internalClient: internalClient,
+		apiReader:      apiReader,
 		opts:           opts,
 	}
 	if opts.SkipAuthorization {
@@ -195,12 +228,12 @@ func NewClient(
 	return c, nil
 }
 
-func newDefaultInternalClient(
+func newDefaultCluster(
 	ctx context.Context,
 	restCfg *rest.Config,
 	scheme *runtime.Scheme,
 	kargoNamespace string,
-) (libClient.WithWatch, error) {
+) (libCluster.Cluster, error) {
 	cluster, err := libCluster.New(
 		restCfg,
 		func(clusterOptions *libCluster.Options) {
@@ -306,11 +339,7 @@ func newDefaultInternalClient(
 		return nil, fmt.Errorf("error starting cluster: %w", err)
 	}
 
-	cl, ok := cluster.GetClient().(libClient.WithWatch)
-	if !ok {
-		return nil, errors.New("internal client does not implement WithWatch")
-	}
-	return cl, nil
+	return cluster, nil
 }
 
 func (c *client) Get(
@@ -677,6 +706,10 @@ func (c *client) InternalClient() libClient.WithWatch {
 	return c.internalClient
 }
 
+func (c *client) APIReader() libClient.Reader {
+	return c.apiReader
+}
+
 func (c *client) Watch(
 	ctx context.Context,
 	list libClient.ObjectList,

pkg/server/kubernetes/client_test.go

@@ -22,7 +22,9 @@ import (
 func TestSetOptionsDefaults(t *testing.T) {
 	opts, err := setOptionsDefaults(ClientOptions{})
 	require.NoError(t, err)
-	require.NotNil(t, opts.NewInternalClient)
+	// NewInternalClient is intentionally left nil; NewClient builds a real
+	// cluster on the default path and only calls this hook when a test sets it.
+	require.Nil(t, opts.NewInternalClient)
 	require.NotNil(t, opts.Scheme)
 }
 

pkg/server/list_for_watch_seed.go

@@ -0,0 +1,57 @@
+package server
+
+import (
+	"context"
+	"fmt"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kargoapi "github.com/akuity/kargo/api/v1alpha1"
+)
+
+// listForWatchSeed lists Kargo resources straight from the Kubernetes API,
+// bypassing the API server's controller-runtime read cache via the client's
+// uncached APIReader.
+//
+// The cached client can return a list ResourceVersion of "0" or one that is
+// older than the apiserver's compacted floor, which makes follow-up watches
+// either replay the full set or fail with a "too old" error and force the
+// client into a refetch loop. listForWatchSeed avoids that by reading through
+// the uncached APIReader for list+watch seed endpoints, where the returned
+// resourceVersion is used to start a follow-up watch.
+//
+// The APIReader does not enforce Kargo's RBAC, so we authorize the caller first
+// with the same list SubjectAccessReview the cached client would perform.
+// resource is the lowercase resource name in the Kargo API group (e.g.
+// "stages", "warehouses", "promotions", "freights").
+func (s *server) listForWatchSeed(
+	ctx context.Context,
+	resource string,
+	list client.ObjectList,
+	opts ...client.ListOption,
+) error {
+	if s.client == nil {
+		return fmt.Errorf("kubernetes client is not configured")
+	}
+	if s.authorizeFn == nil {
+		return fmt.Errorf("authorize function is not configured")
+	}
+
+	var listOpts client.ListOptions
+	listOpts.ApplyOptions(opts)
+
+	// Authorize the user before bypassing the cache. The APIReader runs with the
+	// API server's own credentials, so without this check a caller could read
+	// data they would otherwise be denied.
+	if err := s.authorizeFn(
+		ctx,
+		"list",
+		kargoapi.GroupVersion.WithResource(resource),
+		"",
+		client.ObjectKey{Namespace: listOpts.Namespace},
+	); err != nil {
+		return err
+	}
+
+	return s.client.APIReader().List(ctx, list, opts...)
+}

pkg/server/list_for_watch_seed_test.go

@@ -0,0 +1,174 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	kargoapi "github.com/akuity/kargo/api/v1alpha1"
+	"github.com/akuity/kargo/pkg/server/kubernetes"
+)
+
+// newSeedKubeClient builds a kubernetes.Client whose internal client (and thus
+// its uncached APIReader) is the supplied fake.
+func newSeedKubeClient(t *testing.T, internalClient client.WithWatch) kubernetes.Client {
+	t.Helper()
+	kubeClient, err := kubernetes.NewClient(
+		t.Context(),
+		&rest.Config{},
+		kubernetes.ClientOptions{
+			SkipAuthorization: true,
+			Scheme:            mustNewScheme(),
+			NewInternalClient: func(
+				context.Context,
+				*rest.Config,
+				*runtime.Scheme,
+				string,
+			) (client.WithWatch, error) {
+				return internalClient, nil
+			},
+		},
+	)
+	require.NoError(t, err)
+	return kubeClient
+}
+
+func TestServer_listForWatchSeed(t *testing.T) {
+	t.Parallel()
+
+	scheme := mustNewScheme()
+
+	t.Run("authorizes and lists through the API reader", func(t *testing.T) {
+		t.Parallel()
+
+		var (
+			authorized     bool
+			authorizedVerb string
+			authorizedGVR  schema.GroupVersionResource
+			authorizedSub  string
+			authorizedKey  client.ObjectKey
+			listCalled     bool
+		)
+		internalClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithInterceptorFuncs(interceptor.Funcs{
+				List: func(
+					ctx context.Context,
+					cl client.WithWatch,
+					list client.ObjectList,
+					opts ...client.ListOption,
+				) error {
+					listCalled = true
+					if err := cl.List(ctx, list, opts...); err != nil {
+						return err
+					}
+					if promotions, ok := list.(*kargoapi.PromotionList); ok {
+						promotions.ResourceVersion = "42"
+					}
+					return nil
+				},
+			}).
+			Build()
+
+		s := &server{
+			client: newSeedKubeClient(t, internalClient),
+			authorizeFn: func(
+				_ context.Context,
+				verb string,
+				gvr schema.GroupVersionResource,
+				subresource string,
+				key client.ObjectKey,
+			) error {
+				authorized = true
+				authorizedVerb = verb
+				authorizedGVR = gvr
+				authorizedSub = subresource
+				authorizedKey = key
+				return nil
+			},
+		}
+
+		promotions := &kargoapi.PromotionList{}
+		err := s.listForWatchSeed(
+			t.Context(),
+			"promotions",
+			promotions,
+			client.InNamespace("fake-project"),
+		)
+		require.NoError(t, err)
+		require.True(t, authorized)
+		require.True(t, listCalled)
+		require.Equal(t, "list", authorizedVerb)
+		require.Empty(t, authorizedSub)
+		require.Equal(t, kargoapi.GroupVersion.WithResource("promotions"), authorizedGVR)
+		require.Equal(t, client.ObjectKey{Namespace: "fake-project"}, authorizedKey)
+		require.Equal(t, "42", promotions.ResourceVersion)
+	})
+
+	t.Run("authorization failure short-circuits the list", func(t *testing.T) {
+		t.Parallel()
+
+		authErr := errors.New("not authorized")
+		var listCalled bool
+		internalClient := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithInterceptorFuncs(interceptor.Funcs{
+				List: func(
+					context.Context,
+					client.WithWatch,
+					client.ObjectList,
+					...client.ListOption,
+				) error {
+					listCalled = true
+					return nil
+				},
+			}).
+			Build()
+
+		s := &server{
+			client: newSeedKubeClient(t, internalClient),
+			authorizeFn: func(
+				context.Context,
+				string,
+				schema.GroupVersionResource,
+				string,
+				client.ObjectKey,
+			) error {
+				return authErr
+			},
+		}
+
+		err := s.listForWatchSeed(
+			t.Context(),
+			"promotions",
+			&kargoapi.PromotionList{},
+			client.InNamespace("fake-project"),
+		)
+		require.ErrorIs(t, err, authErr)
+		require.False(t, listCalled)
+	})
+
+	t.Run("errors when authorize function is not configured", func(t *testing.T) {
+		t.Parallel()
+
+		s := &server{
+			client: newSeedKubeClient(t, fake.NewClientBuilder().WithScheme(scheme).Build()),
+		}
+
+		err := s.listForWatchSeed(
+			t.Context(),
+			"promotions",
+			&kargoapi.PromotionList{},
+			client.InNamespace("fake-project"),
+		)
+		require.ErrorContains(t, err, "authorize function is not configured")
+	})
+}