From 899ee68973ffda6cc1fa9ba50363bb663d67dc2c Mon Sep 17 00:00:00 2001 From: Jacob Boykin Date: Fri, 12 Jun 2026 14:49:19 -0400 Subject: [PATCH 1/3] feat(server,ui): seed UI collection watches with list resourceVersion On large projects the UI's Stage/Warehouse/Promotion/Freight collection streams open their follow-up watch with no resourceVersion, so the API server replays every listed object as an individual ADDED event. On a big project that is thousands of redundant events the browser decodes one at a time, freezing the page for the better part of 20s after content appears. Implement standard Kubernetes list-then-watch seeding on the REST/SSE path. Server: - Return the list-level resourceVersion from the list/query endpoints (Stages, Warehouses, Promotions natively; Freight via a new resourceVersion field on the QueryFreights response). - Accept ?resourceVersion= on the SSE watch endpoints and start the watch from it via buildWatchListOptions. - Read seed lists through an uncached direct reader (listForWatchSeed), gated by an explicit SubjectAccessReview that mirrors the cached client, falling back to the cached client when no direct reader is available. - Surface an expired resourceVersion as an SSE error event so clients can relist, and emit synthetic DELETEs for client-side filtered watches. UI: - Centralize seeding in runSeededWatch: seed each watch from the matching list query's resourceVersion (kept out of effect deps to avoid reconnect churn), relist and reopen on an expired-resourceVersion error, and skip replayed ADDED events. On a ~1k-object project this drops ~2,261 replayed ADDED events per page load to 0 and reduces main-thread blocking during load by ~85%. Signed-off-by: Jacob Boykin --- .../pkg_server_query_freights_response.go | 7 + pkg/server/list_for_watch_seed.go | 63 ++++++ pkg/server/list_for_watch_seed_test.go | 185 ++++++++++++++++++ pkg/server/list_promotions_v1alpha1.go | 64 ++++-- pkg/server/list_stages_v1alpha1.go | 79 ++++++-- pkg/server/list_warehouses_v1alpha1.go | 31 ++- pkg/server/query_freights_v1alpha1.go | 117 +++++++---- pkg/server/resource_version.go | 75 +++++++ pkg/server/resource_version_test.go | 71 +++++++ pkg/server/server.go | 36 ++++ pkg/server/watch_helpers.go | 122 +++++++++++- pkg/server/watch_helpers_test.go | 136 +++++++++++++ swagger.json | 4 + .../features/project/pipelines/pipelines.tsx | 2 +- .../promotion/use-watch-promotion.ts | 3 + .../promotion/use-watch-promotions.ts | 67 ++++--- .../project/pipelines/use-watch-freight.ts | 102 ++++++---- .../project/pipelines/use-watch-stages.ts | 96 +++++---- .../project/pipelines/use-watch-warehouses.ts | 131 +++++++------ .../features/project/pipelines/watch-utils.ts | 104 ++++++++-- ui/src/features/stage/promotions.tsx | 2 +- .../models/pkgServerQueryFreightsResponse.ts | 6 + 22 files changed, 1260 insertions(+), 243 deletions(-) create mode 100644 pkg/server/list_for_watch_seed.go create mode 100644 pkg/server/list_for_watch_seed_test.go create mode 100644 pkg/server/resource_version.go create mode 100644 pkg/server/resource_version_test.go create mode 100644 pkg/server/watch_helpers_test.go diff --git a/pkg/client/generated/models/pkg_server_query_freights_response.go b/pkg/client/generated/models/pkg_server_query_freights_response.go index 1a929962c2..e0a2d1ad90 100644 --- a/pkg/client/generated/models/pkg_server_query_freights_response.go +++ b/pkg/client/generated/models/pkg_server_query_freights_response.go @@ -19,6 +19,13 @@ type PkgServerQueryFreightsResponse struct { // groups Groups map[string]PkgServerFreightGroupList `json:"groups,omitempty"` + + // ResourceVersion is the Kubernetes list resourceVersion clients use to + // seed a follow-up Freight watch so the API server does not replay every + // existing Freight as an ADDED event. It is empty for the stage-scoped + // query, whose result is assembled from multiple sources rather than a + // single watchable namespace list. + ResourceVersion string `json:"resourceVersion,omitempty"` } // Validate validates this pkg server query freights response diff --git a/pkg/server/list_for_watch_seed.go b/pkg/server/list_for_watch_seed.go new file mode 100644 index 0000000000..bf3e74f050 --- /dev/null +++ b/pkg/server/list_for_watch_seed.go @@ -0,0 +1,63 @@ +package server + +import ( + "context" + "fmt" + + "sigs.k8s.io/controller-runtime/pkg/client" + + kargoapi "github.com/akuity/kargo/api/v1alpha1" +) + +// listForWatchSeed lists Kargo resources directly from the Kubernetes API, +// bypassing the API server's controller-runtime read cache. +// +// 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 going straight +// to the API for list+watch seed endpoints where the returned resourceVersion +// is used to start a follow-up watch. +// +// The direct reader does not enforce Kargo's RBAC, so we authorize the caller +// first via the authorizing client. resource is the lowercase resource name in +// the Kargo API group (e.g. "stages", "warehouses", "promotions", "freights"). +// When no direct reader is available (tests, or no rest.Config at construction +// time), listForWatchSeed falls back to the standard authorizing client, which +// performs its own SubjectAccessReview as part of List. +func (s *server) listForWatchSeed( + ctx context.Context, + resource string, + list client.ObjectList, + opts ...client.ListOption, +) error { + if s.directReader == nil { + // Fallback: the authorizing client performs its own SAR per call. + if s.client == nil { + return fmt.Errorf("kubernetes client is not configured") + } + return s.client.List(ctx, list, opts...) + } + + 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 direct reader 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.directReader.List(ctx, list, opts...) +} diff --git a/pkg/server/list_for_watch_seed_test.go b/pkg/server/list_for_watch_seed_test.go new file mode 100644 index 0000000000..5e84aae6fd --- /dev/null +++ b/pkg/server/list_for_watch_seed_test.go @@ -0,0 +1,185 @@ +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" +) + +func TestServer_listForWatchSeed(t *testing.T) { + t.Parallel() + + scheme := mustNewScheme() + internalClient := fake.NewClientBuilder().WithScheme(scheme).Build() + kubeClient, err := kubernetes.NewClient( + t.Context(), + &rest.Config{}, + kubernetes.ClientOptions{ + SkipAuthorization: true, + Scheme: scheme, + NewInternalClient: func( + context.Context, + *rest.Config, + *runtime.Scheme, + string, + ) (client.WithWatch, error) { + return internalClient, nil + }, + }, + ) + require.NoError(t, err) + + t.Run("authorizes and lists through direct reader", func(t *testing.T) { + t.Parallel() + + var ( + authorized bool + authorizedVerb string + authorizedGVR schema.GroupVersionResource + authorizedSub string + authorizedKey client.ObjectKey + directListCalled bool + ) + directReader := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func( + ctx context.Context, + cl client.WithWatch, + list client.ObjectList, + opts ...client.ListOption, + ) error { + directListCalled = 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: kubeClient, + directReader: directReader, + 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, directListCalled) + 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 direct list", func(t *testing.T) { + t.Parallel() + + authErr := errors.New("not authorized") + var directListCalled bool + directReader := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func( + context.Context, + client.WithWatch, + client.ObjectList, + ...client.ListOption, + ) error { + directListCalled = true + return nil + }, + }). + Build() + + s := &server{ + client: kubeClient, + directReader: directReader, + 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, directListCalled) + }) + + t.Run("falls back to cached client when no direct reader is wired", func(t *testing.T) { + t.Parallel() + + var authorizeCalled bool + s := &server{ + client: kubeClient, + authorizeFn: func( + context.Context, + string, + schema.GroupVersionResource, + string, + client.ObjectKey, + ) error { + authorizeCalled = true + return nil + }, + } + + // Should not error and should not invoke authorizeFn directly; + // the cached authorizing client performs its own SAR per call, + // which is bypassed here via SkipAuthorization on kubeClient. + err := s.listForWatchSeed( + t.Context(), + "promotions", + &kargoapi.PromotionList{}, + client.InNamespace("fake-project"), + ) + require.NoError(t, err) + require.False(t, authorizeCalled) + }) +} diff --git a/pkg/server/list_promotions_v1alpha1.go b/pkg/server/list_promotions_v1alpha1.go index a28a96bddd..13cbada865 100644 --- a/pkg/server/list_promotions_v1alpha1.go +++ b/pkg/server/list_promotions_v1alpha1.go @@ -74,21 +74,24 @@ func (s *server) listPromotions(c *gin.Context) { stage := c.Query("stage") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchPromotions(c, project, stage) + s.watchPromotions(c, project, stage, c.Query("resourceVersion")) return } list := &kargoapi.PromotionList{} - opts := []client.ListOption{ - client.InNamespace(project), - } - if stage != "" { - opts = append(opts, client.MatchingFields{indexer.PromotionsByStageField: stage}) - } - if err := s.client.List(ctx, list, opts...); err != nil { + if err := s.listForWatchSeed(ctx, "promotions", list, client.InNamespace(project)); err != nil { _ = c.Error(err) return } + if stage != "" { + list.Items = filterPromotionsByStage(list.Items, stage) + } + + list.ResourceVersion = effectiveResourceVersionForList( + list.ResourceVersion, + list.Items, + func(p kargoapi.Promotion) string { return p.ResourceVersion }, + ) // Sort ascending by name slices.SortFunc(list.Items, func(lhs, rhs kargoapi.Promotion) int { @@ -98,14 +101,39 @@ func (s *server) listPromotions(c *gin.Context) { c.JSON(http.StatusOK, list) } -func (s *server) watchPromotions(c *gin.Context, project, stage string) { +// filterPromotionsByStage returns Promotions that target the specified Stage. +// +// Stage filtering is done in-process rather than via the PromotionsByStage +// field index because the watch-seed list goes through listForWatchSeed's +// uncached reader, which cannot serve controller-runtime field indexes. This +// over-fetches the whole namespace, but it matches the (also unfiltered) +// follow-up watch and keeps the returned ResourceVersion watchable. +func filterPromotionsByStage(promotions []kargoapi.Promotion, stage string) []kargoapi.Promotion { + filtered := make([]kargoapi.Promotion, 0, len(promotions)) + for _, promotion := range promotions { + if promotion.Spec.Stage == stage { + filtered = append(filtered, promotion) + } + } + return filtered +} + +// watchPromotions streams Promotion changes through the REST SSE endpoint. +func (s *server) watchPromotions(c *gin.Context, project, stage, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) // Note: We can't filter by stage using field selector in the watch API. // The indexer is for List operations only. We filter events client-side. - w, err := s.client.Watch(ctx, &kargoapi.PromotionList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.PromotionList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch promotions: %w", err)) return @@ -133,18 +161,26 @@ func (s *server) watchPromotions(c *gin.Context, project, stage string) { logger.Debug("watch channel closed") return } + if watchErr := errorFromWatchEvent(e); watchErr != nil { + sendSSEWatchError(c, watchErr) + return + } promotion, ok := convertWatchEventObject(c, e, (*kargoapi.Promotion)(nil)) if !ok { continue } - // Filter by stage if specified (client-side filtering) - if stage != "" && promotion.Spec.Stage != stage { - continue + eventType := e.Type + if stage != "" { + var send bool + eventType, send = filteredWatchEventType(e.Type, promotion.Spec.Stage == stage) + if !send { + continue + } } - if !sendSSEWatchEvent(c, e.Type, promotion) { + if !sendSSEWatchEvent(c, eventType, promotion) { return } } diff --git a/pkg/server/list_stages_v1alpha1.go b/pkg/server/list_stages_v1alpha1.go index 0ee937477b..fd141072f8 100644 --- a/pkg/server/list_stages_v1alpha1.go +++ b/pkg/server/list_stages_v1alpha1.go @@ -73,30 +73,33 @@ func (s *server) listStages(c *gin.Context) { warehouses := c.QueryArray("freightOrigins") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchStages(c, project, warehouses) + s.watchStages(c, project, warehouses, c.Query("resourceVersion")) return } - items, err := api.ListStagesByWarehouses( - ctx, - s.client, - project, - &api.ListStagesOptions{Warehouses: warehouses}, - ) + list, err := s.listStagesByWarehouses(ctx, project, warehouses) if err != nil { _ = c.Error(err) return } - c.JSON(http.StatusOK, &kargoapi.StageList{Items: items}) + c.JSON(http.StatusOK, list) } -func (s *server) watchStages(c *gin.Context, project string, warehouses []string) { +// watchStages streams Stage changes through the REST SSE endpoint. +func (s *server) watchStages(c *gin.Context, project string, warehouses []string, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) - w, err := s.client.Watch(ctx, &kargoapi.StageList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.StageList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch stages: %w", err)) return @@ -124,19 +127,69 @@ func (s *server) watchStages(c *gin.Context, project string, warehouses []string logger.Debug("watch channel closed") return } + if watchErr := errorFromWatchEvent(e); watchErr != nil { + sendSSEWatchError(c, watchErr) + return + } stage, ok := convertWatchEventObject(c, e, (*kargoapi.Stage)(nil)) if !ok { continue } - if len(warehouses) > 0 && !api.StageMatchesAnyWarehouse(stage, warehouses) { - continue + eventType := e.Type + if len(warehouses) > 0 { + var send bool + eventType, send = filteredWatchEventType( + e.Type, + api.StageMatchesAnyWarehouse(stage, warehouses), + ) + if !send { + continue + } } - if !sendSSEWatchEvent(c, e.Type, stage) { + if !sendSSEWatchEvent(c, eventType, stage) { return } } } } + +// listStagesByWarehouses lists Stages in the given project, optionally +// filtered to those that request Freight from at least one of the specified +// warehouses (directly or through upstream stages). When warehouses is empty, +// all Stages are returned. The returned StageList carries an effective +// ResourceVersion derived from the list response or listed items. +func (s *server) listStagesByWarehouses( + ctx context.Context, + project string, + warehouses []string, +) (*kargoapi.StageList, error) { + var list kargoapi.StageList + if err := s.listForWatchSeed(ctx, "stages", &list, client.InNamespace(project)); err != nil { + return nil, err + } + list.ResourceVersion = resourceVersionForStageList(&list) + if len(warehouses) == 0 { + return &list, nil + } + var stages []kargoapi.Stage + for _, stage := range list.Items { + if api.StageMatchesAnyWarehouse(&stage, warehouses) { + stages = append(stages, stage) + } + } + list.Items = stages + return &list, nil +} + +// resourceVersionForStageList returns the list ResourceVersion when useful, +// otherwise it falls back to the maximum Stage item ResourceVersion. +func resourceVersionForStageList(list *kargoapi.StageList) string { + return effectiveResourceVersionForList( + list.ResourceVersion, + list.Items, + func(s kargoapi.Stage) string { return s.ResourceVersion }, + ) +} diff --git a/pkg/server/list_warehouses_v1alpha1.go b/pkg/server/list_warehouses_v1alpha1.go index bcc0a7a697..8c09ffb4c6 100644 --- a/pkg/server/list_warehouses_v1alpha1.go +++ b/pkg/server/list_warehouses_v1alpha1.go @@ -69,25 +69,34 @@ func (s *server) listWarehouses(c *gin.Context) { project := c.Param("project") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchWarehouses(c, project) + s.watchWarehouses(c, project, c.Query("resourceVersion")) return } list := &kargoapi.WarehouseList{} - if err := s.client.List(ctx, list, client.InNamespace(project)); err != nil { + if err := s.listForWatchSeed(ctx, "warehouses", list, client.InNamespace(project)); err != nil { _ = c.Error(err) return } + list.ResourceVersion = resourceVersionForWarehouseList(list) c.JSON(http.StatusOK, list) } -func (s *server) watchWarehouses(c *gin.Context, project string) { +// watchWarehouses streams Warehouse changes through the REST SSE endpoint. +func (s *server) watchWarehouses(c *gin.Context, project string, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) - w, err := s.client.Watch(ctx, &kargoapi.WarehouseList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.WarehouseList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch warehouses: %w", err)) return @@ -115,9 +124,23 @@ func (s *server) watchWarehouses(c *gin.Context, project string) { logger.Debug("watch channel closed") return } + // convertAndSendWatchEvent surfaces watch.Error events itself, so + // no separate errorFromWatchEvent check is needed here (unlike the + // filtered Stage/Promotion handlers, which inspect the event before + // applying their event-type filter). if !convertAndSendWatchEvent(c, e, (*kargoapi.Warehouse)(nil)) { return } } } } + +// resourceVersionForWarehouseList returns the list ResourceVersion when useful, +// otherwise it falls back to the maximum Warehouse item ResourceVersion. +func resourceVersionForWarehouseList(list *kargoapi.WarehouseList) string { + return effectiveResourceVersionForList( + list.ResourceVersion, + list.Items, + func(w kargoapi.Warehouse) string { return w.ResourceVersion }, + ) +} diff --git a/pkg/server/query_freights_v1alpha1.go b/pkg/server/query_freights_v1alpha1.go index d070090acc..5d749e50b5 100644 --- a/pkg/server/query_freights_v1alpha1.go +++ b/pkg/server/query_freights_v1alpha1.go @@ -31,6 +31,12 @@ type freightGroupList struct { type queryFreightsResponse struct { Groups map[string]freightGroupList `json:"groups"` + // ResourceVersion is the Kubernetes list resourceVersion clients use to + // seed a follow-up Freight watch so the API server does not replay every + // existing Freight as an ADDED event. It is empty for the stage-scoped + // query, whose result is assembled from multiple sources rather than a + // single watchable namespace list. + ResourceVersion string `json:"resourceVersion,omitempty"` } const ( @@ -395,7 +401,7 @@ func (s *server) queryFreight(c *gin.Context) { origins := c.QueryArray("origins") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchFreight(c, project, origins) + s.watchFreight(c, project, origins, c.Query("resourceVersion")) return } @@ -406,6 +412,7 @@ func (s *server) queryFreight(c *gin.Context) { } var freight []kargoapi.Freight + var resourceVersion string var err error switch { @@ -428,23 +435,32 @@ func (s *server) queryFreight(c *gin.Context) { _ = c.Error(fmt.Errorf("get available freight for stage: %w", err)) return } + // The stage-availability list is assembled from multiple sources rather + // than a single watchable namespace list, so ResourceVersion is left + // empty and clients fall back to an unseeded watch for this view. case len(origins) > 0: - // Get freight from specific warehouses - freight, err = s.getFreightFromWarehousesREST(ctx, project, origins) - if err != nil { - _ = c.Error(fmt.Errorf("get freight from warehouses: %w", err)) + // Get freight from specific warehouses. The seed list goes through the + // uncached reader for a watchable ResourceVersion, then origins are + // filtered in-process (see filterFreightByOrigins) to match the + // unfiltered follow-up watch. + freightList := &kargoapi.FreightList{} + if err = s.listForWatchSeed(ctx, "freights", freightList, client.InNamespace(project)); err != nil { + _ = c.Error(fmt.Errorf("list freight: %w", err)) return } + freight = filterFreightByOrigins(freightList.Items, origins) + resourceVersion = resourceVersionForFreightList(freightList) default: // Get all freight in the project freightList := &kargoapi.FreightList{} - if err := s.client.List(ctx, freightList, client.InNamespace(project)); err != nil { + if err = s.listForWatchSeed(ctx, "freights", freightList, client.InNamespace(project)); err != nil { _ = c.Error(fmt.Errorf("list freight: %w", err)) return } freight = freightList.Items + resourceVersion = resourceVersionForFreightList(freightList) } // Split the Freight into groups using the generic functions @@ -467,15 +483,28 @@ func (s *server) queryFreight(c *gin.Context) { result[k] = freightGroupList{Items: v} } - c.JSON(http.StatusOK, queryFreightsResponse{Groups: result}) + c.JSON(http.StatusOK, queryFreightsResponse{ + Groups: result, + ResourceVersion: resourceVersion, + }) } -func (s *server) watchFreight(c *gin.Context, project string, origins []string) { +// watchFreight streams Freight changes through the REST SSE endpoint, seeding +// the watch from the provided resourceVersion so the API server does not replay +// every existing Freight as an ADDED event. +func (s *server) watchFreight(c *gin.Context, project string, origins []string, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) - w, err := s.client.Watch(ctx, &kargoapi.FreightList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.FreightList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch freight: %w", err)) return @@ -503,51 +532,59 @@ func (s *server) watchFreight(c *gin.Context, project string, origins []string) logger.Debug("watch channel closed") return } + if watchErr := errorFromWatchEvent(e); watchErr != nil { + sendSSEWatchError(c, watchErr) + return + } freight, ok := convertWatchEventObject(c, e, (*kargoapi.Freight)(nil)) if !ok { continue } - if len(origins) > 0 && !slices.Contains(origins, freight.Origin.Name) { - continue + eventType := e.Type + if len(origins) > 0 { + var send bool + eventType, send = filteredWatchEventType( + e.Type, + slices.Contains(origins, freight.Origin.Name), + ) + if !send { + continue + } } - if !sendSSEWatchEvent(c, e.Type, freight) { + if !sendSSEWatchEvent(c, eventType, freight) { return } } } } -// getFreightFromWarehousesREST is a helper for the REST endpoint that gets freight from warehouses -func (s *server) getFreightFromWarehousesREST( - ctx context.Context, - project string, - warehouses []string, -) ([]kargoapi.Freight, error) { - var allFreight []kargoapi.Freight - for _, warehouse := range warehouses { - var freight kargoapi.FreightList - if err := s.client.List( - ctx, - &freight, - &client.ListOptions{ - Namespace: project, - FieldSelector: fields.OneTermEqualSelector( - indexer.FreightByWarehouseField, - warehouse, - ), - }, - ); err != nil { - return nil, fmt.Errorf( - "error listing Freight for Warehouse %q in namespace %q: %w", - warehouse, - project, - err, - ) +// filterFreightByOrigins returns Freight whose origin Warehouse is one of the +// provided origins. +// +// Origin filtering is done in-process rather than via the FreightByWarehouse +// field index because the watch-seed list goes through listForWatchSeed's +// uncached reader, which cannot serve controller-runtime field indexes. This +// over-fetches the whole namespace, but it matches the (also unfiltered) +// follow-up watch and keeps the returned ResourceVersion watchable. +func filterFreightByOrigins(freight []kargoapi.Freight, origins []string) []kargoapi.Freight { + filtered := make([]kargoapi.Freight, 0, len(freight)) + for _, f := range freight { + if slices.Contains(origins, f.Origin.Name) { + filtered = append(filtered, f) } - allFreight = append(allFreight, freight.Items...) } - return allFreight, nil + return filtered +} + +// resourceVersionForFreightList returns the list ResourceVersion when useful, +// otherwise it falls back to the maximum Freight item ResourceVersion. +func resourceVersionForFreightList(list *kargoapi.FreightList) string { + return effectiveResourceVersionForList( + list.ResourceVersion, + list.Items, + func(f kargoapi.Freight) string { return f.ResourceVersion }, + ) } diff --git a/pkg/server/resource_version.go b/pkg/server/resource_version.go new file mode 100644 index 0000000000..c2dda069e2 --- /dev/null +++ b/pkg/server/resource_version.go @@ -0,0 +1,75 @@ +package server + +import ( + "strconv" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// effectiveResourceVersion returns rv if it is a real, non-default Kubernetes +// resource version (non-empty and non-"0"). Otherwise it computes the maximum +// numeric resource version from the provided item versions as a best-effort +// starting point for a subsequent Watch. +// +// Background: the controller-runtime cached client returns "0" for the +// list-level ResourceVersion, which causes a Kubernetes Watch to replay all +// existing objects as ADDED events. Kubernetes resource versions are officially +// opaque, so this fallback is deliberately conservative: it is only used when +// item versions parse as positive integers, which matches the resource versions +// returned by the API server today. Non-numeric item versions are ignored. If +// no usable item version exists, we return an empty string and preserve the +// previous watch behavior. +// +// In production the max-item fallback is effectively unreachable: the list+watch +// seed endpoints read through listForWatchSeed's uncached reader, which always +// returns a real list-level resource version, so the early return below wins. +// The fallback only runs on the degraded path where no direct reader is wired +// (tests, or no rest.Config) and the cached client reports "0"/"". There it is a +// best effort: the max item version may be older than the apiserver's watch +// window, in which case the follow-up watch simply restarts with a fresh list, +// i.e. the pre-change behavior. +func effectiveResourceVersion(rv string, itemVersions []string) string { + if rv != "" && rv != "0" { + return rv + } + var maxRV int64 + for _, v := range itemVersions { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil || n <= 0 { + continue + } + if n > maxRV { + maxRV = n + } + } + if maxRV == 0 { + return "" + } + return strconv.FormatInt(maxRV, 10) +} + +// effectiveResourceVersionForList returns an effective list resourceVersion for +// a typed list: the list-level resourceVersion when usable, otherwise the +// maximum item resourceVersion (see effectiveResourceVersion). itemRV extracts +// the resourceVersion from each (value-typed) list item, which a plain generic +// constraint cannot do because metav1.Object's methods have pointer receivers. +func effectiveResourceVersionForList[T any](rv string, items []T, itemRV func(T) string) string { + itemVersions := make([]string, len(items)) + for i := range items { + itemVersions[i] = itemRV(items[i]) + } + return effectiveResourceVersion(rv, itemVersions) +} + +// buildWatchListOptions returns namespace-scoped list options for watch calls, +// including ResourceVersion when the caller has one. +func buildWatchListOptions(namespace string, resourceVersion string) []client.ListOption { + watchOpts := []client.ListOption{client.InNamespace(namespace)} + if resourceVersion != "" { + watchOpts = append(watchOpts, &client.ListOptions{ + Raw: &metav1.ListOptions{ResourceVersion: resourceVersion}, + }) + } + return watchOpts +} diff --git a/pkg/server/resource_version_test.go b/pkg/server/resource_version_test.go new file mode 100644 index 0000000000..cc5a040792 --- /dev/null +++ b/pkg/server/resource_version_test.go @@ -0,0 +1,71 @@ +package server + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_effectiveResourceVersion(t *testing.T) { + t.Parallel() + testCases := []struct { + name string + rv string + itemVersions []string + expected string + }{ + { + name: "real list-level rv passthrough", + rv: "12345", + expected: "12345", + }, + { + name: "zero list-level rv falls back to max item rv", + rv: "0", + itemVersions: []string{"100", "300", "200"}, + expected: "300", + }, + { + name: "empty list-level rv falls back to max item rv", + rv: "", + itemVersions: []string{"100", "300", "200"}, + expected: "300", + }, + { + name: "zero list-level rv with no items returns empty", + rv: "0", + itemVersions: []string{}, + expected: "", + }, + { + name: "items with non-numeric rvs are skipped", + rv: "0", + itemVersions: []string{"abc", "100", "def"}, + expected: "100", + }, + { + name: "all invalid item rvs returns empty", + rv: "", + itemVersions: []string{"", "abc"}, + expected: "", + }, + { + name: "real list-level rv wins over items", + rv: "999", + itemVersions: []string{"100", "200"}, + expected: "999", + }, + { + name: "zero item rvs are skipped", + rv: "0", + itemVersions: []string{"0", "0", "100"}, + expected: "100", + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, tc.expected, effectiveResourceVersion(tc.rv, tc.itemVersions)) + }) + } +} diff --git a/pkg/server/server.go b/pkg/server/server.go index 8dfb73e7aa..085e940ecf 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -50,6 +50,17 @@ type server struct { rolesDB rbac.RolesDatabase sender event.Sender + // directReader is a singleton, uncached controller-runtime client used by + // listForWatchSeed to bypass the API server's read cache on list+watch seed + // endpoints whose returned ResourceVersion is reused by a follow-up watch + // (currently Stages, Warehouses, Promotions, and the Freight list served by + // QueryFreight). It is wired once in NewServer from cfg.RestConfig and is + // nil in tests or when no rest.Config is available, in which case + // listForWatchSeed falls back to the standard authorizing client. The + // direct reader does NOT enforce Kargo's RBAC on its own; listForWatchSeed + // authorizes the caller before using it. + directReader client.Reader + // The following behaviors are overridable for testing purposes: // Common validations: @@ -192,6 +203,31 @@ func NewServer( s.getClusterAnalysisTemplateFn = rollouts.GetClusterAnalysisTemplate s.getAnalysisRunFn = rollouts.GetAnalysisRun + // Wire the singleton uncached direct reader once. listForWatchSeed uses it + // to bypass the cache on list+watch endpoints whose cached ResourceVersion + // can otherwise be stale enough to trigger client refetch loops. + if cfg.RestConfig != nil { + directReader, err := client.New( + cfg.RestConfig, + client.Options{ + Scheme: kubeClient.Scheme(), + Mapper: kubeClient.RESTMapper(), + }, + ) + if err != nil { + // Construction failure is unexpected with a valid rest.Config; log + // and continue without the optimization (listForWatchSeed falls back + // to the cached authorizing client). + logging.LoggerFromContext(context.Background()).Error( + err, + "error initializing direct Kubernetes reader; "+ + "falling back to cached client for watch seed endpoints", + ) + } else { + s.directReader = directReader + } + } + return s } diff --git a/pkg/server/watch_helpers.go b/pkg/server/watch_helpers.go index 916581a02b..aeffd3c095 100644 --- a/pkg/server/watch_helpers.go +++ b/pkg/server/watch_helpers.go @@ -2,9 +2,14 @@ package server import ( "encoding/json" + "errors" "fmt" + "net/http" + "connectrpc.com/connect" "github.com/gin-gonic/gin" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/watch" @@ -59,12 +64,105 @@ func convertWatchEventObject[T any](c *gin.Context, e watch.Event, _ T) (T, bool return obj, true } +// errorFromWatchEvent maps Kubernetes watch.Error events into client-visible +// errors that watch handlers can return or stream. +func errorFromWatchEvent(e watch.Event) error { + if e.Type != watch.Error { + return nil + } + + status, ok := e.Object.(*metav1.Status) + if !ok { + return connect.NewError( + connect.CodeUnknown, + fmt.Errorf("watch error: unexpected object type %T", e.Object), + ) + } + + message := status.Message + if message == "" { + message = string(status.Reason) + } + if status.Code == http.StatusGone || status.Reason == metav1.StatusReasonExpired { + return connect.NewError( + connect.CodeOutOfRange, + fmt.Errorf("watch resource version expired: %s", message), + ) + } + return connect.NewError( + connect.CodeUnknown, + fmt.Errorf("watch error: %s", message), + ) +} + +// errorFromWatchStartError maps startup failures from Kubernetes Watch calls +// into the same Connect errors used for watch.Error events. +func errorFromWatchStartError(err error) error { + if err == nil { + return nil + } + if !apierrors.IsResourceExpired(err) { + return err + } + return connect.NewError( + connect.CodeOutOfRange, + fmt.Errorf("watch resource version expired: %s", err.Error()), + ) +} + +// sendSSEWatchError sends a watch error as an SSE error event. Watch endpoints +// may have already sent response headers, so HTTP status codes are no longer a +// reliable way to report expired resource versions after streaming begins. +func sendSSEWatchError(c *gin.Context, err error) { + logger := logging.LoggerFromContext(c.Request.Context()) + + message := err.Error() + var connectErr *connect.Error + if errors.As(err, &connectErr) { + message = connectErr.Message() + } + + event := struct { + Code string `json:"code"` + Message string `json:"message"` + }{ + Code: connect.CodeOf(err).String(), + Message: message, + } + data, marshalErr := json.Marshal(event) + if marshalErr != nil { + logger.Error(marshalErr, "failed to marshal watch error event") + return + } + if _, writeErr := fmt.Fprintf(c.Writer, "event: error\ndata: %s\n\n", data); writeErr != nil { + logger.Debug("failed to write watch error event", "error", writeErr) + return + } + c.Writer.Flush() +} + +// sendSSEWatchStartError sends startup watch errors that are part of the watch +// protocol as SSE error events. It returns true when the error was handled. +func sendSSEWatchStartError(c *gin.Context, err error) bool { + watchErr := errorFromWatchStartError(err) + if connect.CodeOf(watchErr) != connect.CodeOutOfRange { + return false + } + setSSEHeaders(c) + sendSSEWatchError(c, watchErr) + return true +} + // convertAndSendWatchEvent converts a watch event's object to the target type // and sends it as an SSE event. It handles both unstructured objects (from real // clients) and typed objects (from fake clients in tests). Returns true if the // caller should continue processing (including on conversion errors), false if -// the watch should be terminated (write failure). +// the watch should be terminated (write failure or a watch.Error event). func convertAndSendWatchEvent[T any](c *gin.Context, e watch.Event, target T) bool { + if err := errorFromWatchEvent(e); err != nil { + sendSSEWatchError(c, err) + return false + } obj, ok := convertWatchEventObject(c, e, target) if !ok { return true // continue processing, don't terminate watch @@ -72,6 +170,28 @@ func convertAndSendWatchEvent[T any](c *gin.Context, e watch.Event, target T) bo return sendSSEWatchEvent(c, e.Type, obj) } +// filteredWatchEventType returns the event type to send for a client-side +// filtered watch event. Kubernetes server-side selectors send a DELETED event +// when a previously matching object is modified so it no longer matches; this +// helper mirrors that behavior for filters we must evaluate in-process. +// +// Because we hold no per-client matched set, we cannot tell whether a +// non-matching MODIFIED object previously matched, so we emit a synthetic +// DELETED for every non-matching MODIFIED event — including objects the client +// never received an ADDED for. This over-emits DELETEs relative to a real +// server-side selector, but a DELETE for an object the client is not tracking +// is a harmless no-op. Achieving exact fidelity would require tracking sent +// object identities per client, which is not worth the complexity here. +func filteredWatchEventType(eventType watch.EventType, matches bool) (watch.EventType, bool) { + if matches { + return eventType, true + } + if eventType == watch.Modified { + return watch.Deleted, true + } + return "", false +} + // sendSSEWatchEvent sends an object as an SSE watch event. Returns true if // the caller should continue processing, false if the watch should be // terminated (write failure). diff --git a/pkg/server/watch_helpers_test.go b/pkg/server/watch_helpers_test.go new file mode 100644 index 0000000000..8499faae49 --- /dev/null +++ b/pkg/server/watch_helpers_test.go @@ -0,0 +1,136 @@ +package server + +import ( + "errors" + "net/http" + "net/http/httptest" + "testing" + + "connectrpc.com/connect" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" +) + +func Test_errorFromWatchStartError(t *testing.T) { + t.Parallel() + + t.Run("passes through ordinary errors", func(t *testing.T) { + t.Parallel() + err := errors.New("network unavailable") + + require.ErrorIs(t, errorFromWatchStartError(err), err) + }) + + t.Run("maps expired resource versions to out of range", func(t *testing.T) { + t.Parallel() + err := apierrors.NewResourceExpired("too old resource version") + + mapped := errorFromWatchStartError(err) + + require.Error(t, mapped) + require.Equal(t, connect.CodeOutOfRange, connect.CodeOf(mapped)) + require.ErrorContains(t, mapped, "watch resource version expired") + }) + + t.Run("returns nil for nil", func(t *testing.T) { + t.Parallel() + require.NoError(t, errorFromWatchStartError(nil)) + }) +} + +func Test_errorFromWatchStartError_withStatusError(t *testing.T) { + t.Parallel() + + err := apierrors.NewNotFound( + schema.GroupResource{Group: "kargo.akuity.io", Resource: "promotions"}, + "promotion-1", + ) + + mapped := errorFromWatchStartError(err) + + require.ErrorIs(t, mapped, err) + require.Equal(t, connect.CodeUnknown, connect.CodeOf(mapped)) +} + +func Test_sendSSEWatchError(t *testing.T) { + t.Parallel() + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/", nil) + + err := connect.NewError(connect.CodeOutOfRange, errors.New("resource version expired")) + + sendSSEWatchError(c, err) + + require.Contains(t, recorder.Body.String(), "event: error") + require.Contains(t, recorder.Body.String(), `"code":"out_of_range"`) + require.Contains(t, recorder.Body.String(), `"message":"resource version expired"`) + require.NotContains(t, recorder.Body.String(), "out_of_range: resource version expired") +} + +func Test_convertAndSendWatchEvent_errorEvent(t *testing.T) { + t.Parallel() + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/", nil) + + keepGoing := convertAndSendWatchEvent(c, watch.Event{ + Type: watch.Error, + Object: &metav1.Status{ + Status: metav1.StatusFailure, + Message: "too old resource version", + Reason: metav1.StatusReasonExpired, + Code: http.StatusGone, + }, + }, (*metav1.PartialObjectMetadata)(nil)) + + require.False(t, keepGoing) + require.Contains(t, recorder.Body.String(), "event: error") + require.Contains(t, recorder.Body.String(), `"code":"out_of_range"`) +} + +func Test_filteredWatchEventType(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + eventType watch.EventType + matches bool + wantType watch.EventType + wantSend bool + }{ + { + name: "matching event passes through", + eventType: watch.Modified, + matches: true, + wantType: watch.Modified, + wantSend: true, + }, + { + name: "non-matching modified becomes delete", + eventType: watch.Modified, + wantType: watch.Deleted, + wantSend: true, + }, + { + name: "non-matching added is skipped", + eventType: watch.Added, + }, + } + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + eventType, send := filteredWatchEventType(testCase.eventType, testCase.matches) + + require.Equal(t, testCase.wantSend, send) + require.Equal(t, testCase.wantType, eventType) + }) + } +} diff --git a/swagger.json b/swagger.json index 81d0587bc7..f8d82d2ae9 100644 --- a/swagger.json +++ b/swagger.json @@ -8668,6 +8668,10 @@ "additionalProperties": { "$ref": "#/definitions/pkg_server.freightGroupList" } + }, + "resourceVersion": { + "description": "ResourceVersion is the Kubernetes list resourceVersion clients use to\nseed a follow-up Freight watch so the API server does not replay every\nexisting Freight as an ADDED event. It is empty for the stage-scoped\nquery, whose result is assembled from multiple sources rather than a\nsingle watchable namespace list.", + "type": "string" } } }, diff --git a/ui/src/features/project/pipelines/pipelines.tsx b/ui/src/features/project/pipelines/pipelines.tsx index e1b3a3fbdd..abf6ea6290 100644 --- a/ui/src/features/project/pipelines/pipelines.tsx +++ b/ui/src/features/project/pipelines/pipelines.tsx @@ -148,7 +148,7 @@ export const Pipelines = (props: { creatingStage?: boolean; creatingWarehouse?: usePersistPreferredFilter(projectName || '', preferredFilter); - useWatchFreight(projectName || ''); + useWatchFreight(projectName || '', preferredFilter.warehouses, !getFreightQuery.isLoading); if (loading) { return ; diff --git a/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts b/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts index 7d2647a0b7..d2b121c842 100644 --- a/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts +++ b/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts @@ -20,6 +20,9 @@ export const useWatchPromotion = (project: string, promotion: string) => { (async () => { for await (const event of readSSEStream(url, abort.signal)) { + if ('error' in event) { + continue; + } const p = event.object; if (event.type === 'DELETED') { diff --git a/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts b/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts index 6537733ba2..43e92caf9d 100644 --- a/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts +++ b/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts @@ -4,40 +4,63 @@ import { useEffect } from 'react'; import { getListPromotionsQueryKey, listPromotionsResponse } from '@ui/gen/api/v2/core/core'; import { Promotion } from '@ui/gen/api/v2/models'; -import { readSSEStream, upsertOrDelete } from '../watch-utils'; +import { runSeededWatch, upsertOrDelete } from '../watch-utils'; -export const useWatchPromotions = (project: string, stage: string) => { +// enabled gates the watch on the initial list having loaded, so it never opens +// before a seed resourceVersion is available — otherwise it would start an +// unseeded watch that replays every Promotion. (Unlike the pipeline page, this +// page has no loading gate above the watch, so the caller must pass it.) +export const useWatchPromotions = (project: string, stage: string, enabled = true) => { const client = useQueryClient(); useEffect(() => { - if (!project || !stage) { + if (!project || !stage || !enabled) { return; } const abort = new AbortController(); - const params = new URLSearchParams({ watch: 'true', stage }); - const url = `/v1beta1/projects/${encodeURIComponent(project)}/promotions?${params}`; const listKey = getListPromotionsQueryKey(project, { stage }); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const promotion = event.object; + const seedResourceVersion = () => + (client.getQueryData(listKey) as listPromotionsResponse | undefined)?.data?.metadata + ?.resourceVersion; - client.setQueryData(listKey, (old: listPromotionsResponse | undefined) => { - if (!old?.data) { - return old; - } - return { - ...old, - data: { - ...old.data, - items: upsertOrDelete(old.data.items ?? [], promotion, event.type) - } - }; - }); + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true', stage }); + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); } - })(); + return `/v1beta1/projects/${encodeURIComponent(project)}/promotions?${params}`; + }; + + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; + + const onEvent = (type: string, promotion: Promotion) => { + client.setQueryData(listKey, (old: listPromotionsResponse | undefined) => { + if (!old?.data) { + return old; + } + return { + ...old, + data: { + ...old.data, + items: upsertOrDelete(old.data.items ?? [], promotion, type) + } + }; + }); + }; + + runSeededWatch({ + signal: abort.signal, + buildUrl, + seedResourceVersion, + relist, + onEvent + }); return () => abort.abort(); - }, [project, stage, client]); + }, [project, stage, client, enabled]); }; diff --git a/ui/src/features/project/pipelines/use-watch-freight.ts b/ui/src/features/project/pipelines/use-watch-freight.ts index 8278b2e957..affefd5048 100644 --- a/ui/src/features/project/pipelines/use-watch-freight.ts +++ b/ui/src/features/project/pipelines/use-watch-freight.ts @@ -9,59 +9,83 @@ import { } from '@ui/gen/api/v2/core/core'; import { Freight } from '@ui/gen/api/v2/models'; -import { readSSEStream, upsertOrDelete } from './watch-utils'; +import { runSeededWatch, upsertOrDelete } from './watch-utils'; -export const useWatchFreight = (project: string) => { +// origins must match the params the page's useQueryFreightsRest uses, so the +// seed resourceVersion is read from the same query cache entry. enabled gates +// the watch on the initial list having loaded, so it never opens before a seed +// resourceVersion is available (which would start an unseeded, replaying watch). +export const useWatchFreight = (project: string, origins?: string[], enabled = true) => { const client = useQueryClient(); useEffect(() => { - if (!project) { + if (!project || !enabled) { return; } const abort = new AbortController(); - const url = `/v1beta1/projects/${encodeURIComponent(project)}/freight?watch=true`; - const listKey = getQueryFreightsRestQueryKey(project, undefined); + const listKey = getQueryFreightsRestQueryKey(project, { origins }); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const freight = event.object; + const seedResourceVersion = () => + (client.getQueryData(listKey) as queryFreightsRestResponse | undefined)?.data + ?.resourceVersion; - // Update all queryFreight caches for this project, including - // warehouse-filtered variants, which use a different cache key. - // Using setQueriesData (rather than setQueryData) ensures we only - // touch caches that already have an active query backing them, - // avoiding orphaned entries with no queryFn that would crash on - // refetch. - client.setQueriesData( - { queryKey: listKey, exact: false }, - (old) => { - if (!old?.data?.groups) { - return old; - } - const updatedGroups = Object.fromEntries( - Object.entries(old.data.groups).map(([key, group]) => [ - key, - { ...group, items: upsertOrDelete(group.items ?? [], freight, event.type) } - ]) - ); - return { ...old, data: { ...old.data, groups: updatedGroups } }; - } - ); + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true' }); + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); + } + return `/v1beta1/projects/${encodeURIComponent(project)}/freight?${params}`; + }; - const freightKey = getGetFreightQueryKey(project, freight.metadata?.name); + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; - if (event.type === 'DELETED') { - client.removeQueries({ queryKey: freightKey }); - } else { - client.setQueryData(freightKey, (old: getFreightResponse | undefined) => ({ - ...old, - data: freight - })); + const onEvent = (type: string, freight: Freight) => { + // Update all queryFreight caches for this project, including + // warehouse-filtered variants, which use a different cache key. + // Using setQueriesData (rather than setQueryData) ensures we only + // touch caches that already have an active query backing them, + // avoiding orphaned entries with no queryFn that would crash on + // refetch. + client.setQueriesData( + { queryKey: getQueryFreightsRestQueryKey(project, undefined), exact: false }, + (old) => { + if (!old?.data?.groups) { + return old; + } + const updatedGroups = Object.fromEntries( + Object.entries(old.data.groups).map(([key, group]) => [ + key, + { ...group, items: upsertOrDelete(group.items ?? [], freight, type) } + ]) + ); + return { ...old, data: { ...old.data, groups: updatedGroups } }; } + ); + + const freightKey = getGetFreightQueryKey(project, freight.metadata?.name); + + if (type === 'DELETED') { + client.removeQueries({ queryKey: freightKey }); + } else { + client.setQueryData(freightKey, (old: getFreightResponse | undefined) => ({ + ...old, + data: freight + })); } - })(); + }; + + runSeededWatch({ + signal: abort.signal, + buildUrl, + seedResourceVersion, + relist, + onEvent + }); return () => abort.abort(); - }, [project]); + }, [project, client, enabled, (origins || []).join(',')]); }; diff --git a/ui/src/features/project/pipelines/use-watch-stages.ts b/ui/src/features/project/pipelines/use-watch-stages.ts index a533d8dc2e..adb79f57cd 100644 --- a/ui/src/features/project/pipelines/use-watch-stages.ts +++ b/ui/src/features/project/pipelines/use-watch-stages.ts @@ -9,7 +9,7 @@ import { } from '@ui/gen/api/v2/core/core'; import { Stage } from '@ui/gen/api/v2/models'; -import { readSSEStream, upsertOrDelete } from './watch-utils'; +import { runSeededWatch, upsertOrDelete } from './watch-utils'; export const useWatchStages = ( project: string, @@ -24,54 +24,66 @@ export const useWatchStages = ( } const abort = new AbortController(); - - const params = new URLSearchParams({ watch: 'true' }); - for (const wh of warehouses || []) { - params.append('freightOrigins', wh); - } - const url = `/v1beta1/projects/${encodeURIComponent(project)}/stages?${params}`; const listKey = getListStagesQueryKey(project, { freightOrigins: warehouses || [] }); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const stage = event.object; + const seedResourceVersion = () => + (client.getQueryData(listKey) as listStagesResponse | undefined)?.data?.metadata + ?.resourceVersion; - client.setQueriesData( - { exact: false, queryKey: listKey }, - (old: listStagesResponse | undefined) => { - if (!old?.data) { - return old; - } - return { - ...old, - data: { ...old.data, items: upsertOrDelete(old.data.items ?? [], stage, event.type) } - }; + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true' }); + for (const wh of warehouses || []) { + params.append('freightOrigins', wh); + } + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); + } + return `/v1beta1/projects/${encodeURIComponent(project)}/stages?${params}`; + }; + + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; + + const onEvent = (type: string, stage: Stage) => { + client.setQueriesData( + { exact: false, queryKey: listKey }, + (old: listStagesResponse | undefined) => { + if (!old?.data) { + return old; } - ); + return { + ...old, + data: { ...old.data, items: upsertOrDelete(old.data.items ?? [], stage, type) } + }; + } + ); - const stageKey = getGetStageQueryKey(project, stage.metadata?.name); - if (event.type === 'DELETED') { - client.removeQueries({ queryKey: stageKey }); - } else { - client.setQueriesData( - { exact: false, queryKey: stageKey }, - (old: getStageResponse | undefined) => - old - ? { - ...old, - data: { - // WATCH ENDPOINT STAGE COMES WITHOUT kind, apiVersion etc.. - // SO WE NEED TO PRESERVE IT FROM INITIAL DATA - ...old?.data, - ...stage - } + const stageKey = getGetStageQueryKey(project, stage.metadata?.name); + if (type === 'DELETED') { + client.removeQueries({ queryKey: stageKey }); + } else { + client.setQueriesData( + { exact: false, queryKey: stageKey }, + (old: getStageResponse | undefined) => + old + ? { + ...old, + data: { + // WATCH ENDPOINT STAGE COMES WITHOUT kind, apiVersion etc.. + // SO WE NEED TO PRESERVE IT FROM INITIAL DATA + ...old?.data, + ...stage } - : old - ); - onStageEvent?.(stage); - } + } + : old + ); + onStageEvent?.(stage); } - })(); + }; + + runSeededWatch({ signal: abort.signal, buildUrl, seedResourceVersion, relist, onEvent }); return () => abort.abort(); }, [project, (warehouses || []).join(','), client]); diff --git a/ui/src/features/project/pipelines/use-watch-warehouses.ts b/ui/src/features/project/pipelines/use-watch-warehouses.ts index d01aeee138..a05fbae2b4 100644 --- a/ui/src/features/project/pipelines/use-watch-warehouses.ts +++ b/ui/src/features/project/pipelines/use-watch-warehouses.ts @@ -9,7 +9,7 @@ import { } from '@ui/gen/api/v2/core/core'; import { Warehouse } from '@ui/gen/api/v2/models'; -import { readSSEStream, upsertOrDelete } from './watch-utils'; +import { runSeededWatch, upsertOrDelete } from './watch-utils'; export const useWatchWarehouses = ( project: string, @@ -26,70 +26,91 @@ export const useWatchWarehouses = ( } const abort = new AbortController(); - const url = `/v1beta1/projects/${encodeURIComponent(project)}/warehouses?watch=true`; const listKey = getListWarehousesQueryKey(project); const pendingRefresh: Record = {}; - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const warehouse = event.object; - const name = warehouse.metadata?.name || ''; + const seedResourceVersion = () => + (client.getQueryData(listKey) as listWarehousesResponse | undefined)?.data?.metadata + ?.resourceVersion; - client.setQueriesData( - { exact: false, queryKey: listKey }, - (old: listWarehousesResponse | undefined) => { - if (!old?.data) { - return old; - } - return { - ...old, - data: { - ...old.data, - items: upsertOrDelete(old.data.items ?? [], warehouse, event.type) - } - }; + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true' }); + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); + } + return `/v1beta1/projects/${encodeURIComponent(project)}/warehouses?${params}`; + }; + + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; + + const onEvent = (type: string, warehouse: Warehouse) => { + const name = warehouse.metadata?.name || ''; + + client.setQueriesData( + { exact: false, queryKey: listKey }, + (old: listWarehousesResponse | undefined) => { + if (!old?.data) { + return old; } - ); + return { + ...old, + data: { + ...old.data, + items: upsertOrDelete(old.data.items ?? [], warehouse, type) + } + }; + } + ); + + const warehouseKey = getGetWarehouseQueryKey(project, name); - const warehouseKey = getGetWarehouseQueryKey(project, name); - - if (event.type === 'DELETED') { - client.removeQueries({ queryKey: warehouseKey }); - } else { - client.setQueriesData( - { - exact: false, - queryKey: warehouseKey - }, - (old: getWarehouseResponse | undefined) => - old - ? { - ...old, - data: { - // WATCH ENDPOINT WAREHOUSE COMES WITHOUT kind, apiVersion etc.. - // SO WE NEED TO PRESERVE IT FROM INITIAL DATA - ...old?.data, - ...warehouse - } + if (type === 'DELETED') { + client.removeQueries({ queryKey: warehouseKey }); + } else { + client.setQueriesData( + { + exact: false, + queryKey: warehouseKey + }, + (old: getWarehouseResponse | undefined) => + old + ? { + ...old, + data: { + // WATCH ENDPOINT WAREHOUSE COMES WITHOUT kind, apiVersion etc.. + // SO WE NEED TO PRESERVE IT FROM INITIAL DATA + ...old?.data, + ...warehouse } - : old - ); - - const refreshRequest = warehouse.metadata?.annotations?.['kargo.akuity.io/refresh']; - const refreshStatus = warehouse.status?.lastHandledRefresh; - const isRefreshing = refreshRequest !== undefined && refreshRequest !== refreshStatus; - - if (isRefreshing) { - pendingRefresh[name] = true; - } else if (pendingRefresh[name]) { - delete pendingRefresh[name]; - opts?.refreshHook?.(); - } + } + : old + ); + + const refreshRequest = warehouse.metadata?.annotations?.['kargo.akuity.io/refresh']; + const refreshStatus = warehouse.status?.lastHandledRefresh; + const isRefreshing = refreshRequest !== undefined && refreshRequest !== refreshStatus; - opts?.onWarehouseEvent?.(warehouse); + if (isRefreshing) { + pendingRefresh[name] = true; + } else if (pendingRefresh[name]) { + delete pendingRefresh[name]; + opts?.refreshHook?.(); } + + opts?.onWarehouseEvent?.(warehouse); } - })(); + }; + + runSeededWatch({ + signal: abort.signal, + buildUrl, + seedResourceVersion, + relist, + onEvent + }); return () => abort.abort(); }, [project, client]); diff --git a/ui/src/features/project/pipelines/watch-utils.ts b/ui/src/features/project/pipelines/watch-utils.ts index 7d5391e70c..675530884f 100644 --- a/ui/src/features/project/pipelines/watch-utils.ts +++ b/ui/src/features/project/pipelines/watch-utils.ts @@ -2,7 +2,14 @@ import { authTokenKey } from '@ui/config/auth'; export const getBaseUrl = () => (import.meta.env.VITE_API_URL as string | undefined) || ''; -export type SSEWatchEvent = { type: string; object: T }; +// Code the server emits (as `event: error`) when the resourceVersion used to +// seed a watch is older than the API server's watch window. Clients respond by +// relisting to obtain a fresh, watchable resourceVersion. +export const WATCH_ERROR_EXPIRED = 'out_of_range'; + +export type SSEWatchError = { code: string; message: string }; + +export type SSEWatchEvent = { type: string; object: T } | { error: SSEWatchError }; export async function* readSSEStream( url: string, @@ -33,12 +40,17 @@ export async function* readSSEStream( buffer = parts.pop() ?? ''; for (const part of parts) { - const dataLine = part.split('\n').find((l) => l.startsWith('data: ')); + const lines = part.split('\n'); + const dataLine = lines.find((l) => l.startsWith('data: ')); if (!dataLine) { continue; } + const isError = lines.some((l) => l.startsWith('event: error')); try { - yield JSON.parse(dataLine.slice(6)) as SSEWatchEvent; + const parsed = JSON.parse(dataLine.slice(6)); + yield isError + ? { error: parsed as SSEWatchError } + : (parsed as { type: string; object: T }); } catch (_) { // skip malformed events } @@ -49,6 +61,67 @@ export async function* readSSEStream( } } +// runSeededWatch opens a Kubernetes-style list-then-watch SSE stream seeded from +// the list's resourceVersion, so the API server does not replay every existing +// object as an ADDED event. The seed resourceVersion is read once per connection +// from `seedResourceVersion()` (kept out of any React effect deps by the caller +// so a cache-advancing event never tears the stream down). If the seed is too +// old, the server reports `WATCH_ERROR_EXPIRED`; we then `relist()` for a fresh +// resourceVersion and reopen. A clean close or generic error stops the stream +// (matching the pre-seeding behavior — no auto-reconnect). +export async function runSeededWatch(params: { + signal: AbortSignal; + buildUrl: (resourceVersion: string) => string; + seedResourceVersion: () => string | undefined; + relist: () => Promise; + onEvent: (type: string, object: T) => void; +}): Promise { + let resourceVersion = params.seedResourceVersion() || ''; + + for (;;) { + let expired = false; + try { + for await (const event of readSSEStream(params.buildUrl(resourceVersion), params.signal)) { + if ('error' in event) { + if (event.error.code === WATCH_ERROR_EXPIRED) { + expired = true; + break; + } + continue; + } + params.onEvent(event.type, event.object); + } + } catch (_) { + // Aborted unmount or a network failure: stop. + return; + } + + if (params.signal.aborted || !expired) { + return; + } + + resourceVersion = (await params.relist()) || ''; + } +} + +// isNewerResourceVersion reports whether incoming is a strictly newer Kubernetes +// resourceVersion than existing. resourceVersions are officially opaque, so we +// only compare numerically (their actual encoding today) and otherwise fail open +// (treat as newer) to avoid suppressing a real update. +function isNewerResourceVersion(incoming?: string, existing?: string): boolean { + if (!incoming) { + return false; + } + if (!existing) { + return true; + } + try { + return BigInt(incoming) > BigInt(existing); + } catch (_) { + return true; + } +} + // Reads an SSE stream of raw text chunks (e.g. AnalysisRun logs). The server // writes each line of a chunk as its own `data:` line, so rejoining the data // lines with `\n` reconstructs the original chunk verbatim. Unlike @@ -102,16 +175,25 @@ export async function* readSSETextStream(url: string, signal: AbortSignal): Asyn } } -export function upsertOrDelete( - items: T[], - item: T, - eventType: string -): T[] { +export function upsertOrDelete< + T extends { metadata?: { name?: string; resourceVersion?: string } } +>(items: T[], item: T, eventType: string): T[] { const index = items.findIndex((i) => i.metadata?.name === item.metadata?.name); if (eventType === 'DELETED') { return index !== -1 ? [...items.slice(0, index), ...items.slice(index + 1)] : items; } - return index !== -1 - ? [...items.slice(0, index), item, ...items.slice(index + 1)] - : [...items, item]; + if (index === -1) { + return [...items, item]; + } + // Skip a replayed ADDED for an object we already hold at the same or newer + // resourceVersion. Seeding should prevent these server-side; this is a + // belt-and-suspenders guard for an unseeded or just-relisted watch. MODIFIED + // events always carry a newer resourceVersion and are applied unconditionally. + if ( + eventType === 'ADDED' && + !isNewerResourceVersion(item.metadata?.resourceVersion, items[index].metadata?.resourceVersion) + ) { + return items; + } + return [...items.slice(0, index), item, ...items.slice(index + 1)]; } diff --git a/ui/src/features/stage/promotions.tsx b/ui/src/features/stage/promotions.tsx index 57068062f9..81fac0af00 100644 --- a/ui/src/features/stage/promotions.tsx +++ b/ui/src/features/stage/promotions.tsx @@ -52,7 +52,7 @@ export const Promotions = ({ argocdShard }: { argocdShard?: ArgoCDShard }) => { // modal kept in the same component for live view const [selectedPromotion, setSelectedPromotion] = useState(); - useWatchPromotions(projectName || '', stageName || ''); + useWatchPromotions(projectName || '', stageName || '', !listPromotionsQuery.isLoading); const promotions = React.useMemo(() => { // Immutable sorting diff --git a/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts b/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts index ccc87d29a1..3fe472cf69 100644 --- a/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts +++ b/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts @@ -9,4 +9,10 @@ import type { PkgServerQueryFreightsResponseGroups } from './pkgServerQueryFreig export interface PkgServerQueryFreightsResponse { groups?: PkgServerQueryFreightsResponseGroups; + /** ResourceVersion is the Kubernetes list resourceVersion clients use to +seed a follow-up Freight watch so the API server does not replay every +existing Freight as an ADDED event. It is empty for the stage-scoped +query, whose result is assembled from multiple sources rather than a +single watchable namespace list. */ + resourceVersion?: string; } From 553b85489748f0656c7405956638f5f3ab774231 Mon Sep 17 00:00:00 2001 From: Jacob Boykin Date: Tue, 16 Jun 2026 19:08:13 -0400 Subject: [PATCH 2/3] refactor(server): drop unreachable resourceVersion fallback The list+watch seed endpoints read through listForWatchSeed's uncached directReader, which always returns a real list-level resourceVersion. The effectiveResourceVersion max-item fallback only ran on the degraded no-directReader path (tests, or no rest.Config), where its best-effort RV is no better than opening the watch unseeded -- the documented pre-change behavior. Replace it with a small normalizeListResourceVersion that maps the cached client's "0" sentinel to an empty string, and collapse the three near-duplicate resourceVersionFor*List helpers into inline calls. The upsertOrDelete ADDED dedup guard is retained: it makes the degraded unseeded path (which now replays) cheap. Net -120 lines. No behavior change on the production directReader path. Signed-off-by: Jacob Boykin --- pkg/server/list_promotions_v1alpha1.go | 6 +-- pkg/server/list_stages_v1alpha1.go | 12 +---- pkg/server/list_warehouses_v1alpha1.go | 12 +---- pkg/server/query_freights_v1alpha1.go | 14 +----- pkg/server/resource_version.go | 63 +++++--------------------- pkg/server/resource_version_test.go | 61 ++++--------------------- 6 files changed, 24 insertions(+), 144 deletions(-) diff --git a/pkg/server/list_promotions_v1alpha1.go b/pkg/server/list_promotions_v1alpha1.go index 13cbada865..519797eab4 100644 --- a/pkg/server/list_promotions_v1alpha1.go +++ b/pkg/server/list_promotions_v1alpha1.go @@ -87,11 +87,7 @@ func (s *server) listPromotions(c *gin.Context) { list.Items = filterPromotionsByStage(list.Items, stage) } - list.ResourceVersion = effectiveResourceVersionForList( - list.ResourceVersion, - list.Items, - func(p kargoapi.Promotion) string { return p.ResourceVersion }, - ) + list.ResourceVersion = normalizeListResourceVersion(list.ResourceVersion) // Sort ascending by name slices.SortFunc(list.Items, func(lhs, rhs kargoapi.Promotion) int { diff --git a/pkg/server/list_stages_v1alpha1.go b/pkg/server/list_stages_v1alpha1.go index fd141072f8..447e05fa4b 100644 --- a/pkg/server/list_stages_v1alpha1.go +++ b/pkg/server/list_stages_v1alpha1.go @@ -170,7 +170,7 @@ func (s *server) listStagesByWarehouses( if err := s.listForWatchSeed(ctx, "stages", &list, client.InNamespace(project)); err != nil { return nil, err } - list.ResourceVersion = resourceVersionForStageList(&list) + list.ResourceVersion = normalizeListResourceVersion(list.ResourceVersion) if len(warehouses) == 0 { return &list, nil } @@ -183,13 +183,3 @@ func (s *server) listStagesByWarehouses( list.Items = stages return &list, nil } - -// resourceVersionForStageList returns the list ResourceVersion when useful, -// otherwise it falls back to the maximum Stage item ResourceVersion. -func resourceVersionForStageList(list *kargoapi.StageList) string { - return effectiveResourceVersionForList( - list.ResourceVersion, - list.Items, - func(s kargoapi.Stage) string { return s.ResourceVersion }, - ) -} diff --git a/pkg/server/list_warehouses_v1alpha1.go b/pkg/server/list_warehouses_v1alpha1.go index 8c09ffb4c6..fcdf57b260 100644 --- a/pkg/server/list_warehouses_v1alpha1.go +++ b/pkg/server/list_warehouses_v1alpha1.go @@ -78,7 +78,7 @@ func (s *server) listWarehouses(c *gin.Context) { _ = c.Error(err) return } - list.ResourceVersion = resourceVersionForWarehouseList(list) + list.ResourceVersion = normalizeListResourceVersion(list.ResourceVersion) c.JSON(http.StatusOK, list) } @@ -134,13 +134,3 @@ func (s *server) watchWarehouses(c *gin.Context, project string, resourceVersion } } } - -// resourceVersionForWarehouseList returns the list ResourceVersion when useful, -// otherwise it falls back to the maximum Warehouse item ResourceVersion. -func resourceVersionForWarehouseList(list *kargoapi.WarehouseList) string { - return effectiveResourceVersionForList( - list.ResourceVersion, - list.Items, - func(w kargoapi.Warehouse) string { return w.ResourceVersion }, - ) -} diff --git a/pkg/server/query_freights_v1alpha1.go b/pkg/server/query_freights_v1alpha1.go index 5d749e50b5..d7dce47a3b 100644 --- a/pkg/server/query_freights_v1alpha1.go +++ b/pkg/server/query_freights_v1alpha1.go @@ -450,7 +450,7 @@ func (s *server) queryFreight(c *gin.Context) { return } freight = filterFreightByOrigins(freightList.Items, origins) - resourceVersion = resourceVersionForFreightList(freightList) + resourceVersion = normalizeListResourceVersion(freightList.ResourceVersion) default: // Get all freight in the project @@ -460,7 +460,7 @@ func (s *server) queryFreight(c *gin.Context) { return } freight = freightList.Items - resourceVersion = resourceVersionForFreightList(freightList) + resourceVersion = normalizeListResourceVersion(freightList.ResourceVersion) } // Split the Freight into groups using the generic functions @@ -578,13 +578,3 @@ func filterFreightByOrigins(freight []kargoapi.Freight, origins []string) []karg } return filtered } - -// resourceVersionForFreightList returns the list ResourceVersion when useful, -// otherwise it falls back to the maximum Freight item ResourceVersion. -func resourceVersionForFreightList(list *kargoapi.FreightList) string { - return effectiveResourceVersionForList( - list.ResourceVersion, - list.Items, - func(f kargoapi.Freight) string { return f.ResourceVersion }, - ) -} diff --git a/pkg/server/resource_version.go b/pkg/server/resource_version.go index c2dda069e2..8c82f22c29 100644 --- a/pkg/server/resource_version.go +++ b/pkg/server/resource_version.go @@ -1,65 +1,24 @@ package server import ( - "strconv" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" ) -// effectiveResourceVersion returns rv if it is a real, non-default Kubernetes -// resource version (non-empty and non-"0"). Otherwise it computes the maximum -// numeric resource version from the provided item versions as a best-effort -// starting point for a subsequent Watch. -// -// Background: the controller-runtime cached client returns "0" for the -// list-level ResourceVersion, which causes a Kubernetes Watch to replay all -// existing objects as ADDED events. Kubernetes resource versions are officially -// opaque, so this fallback is deliberately conservative: it is only used when -// item versions parse as positive integers, which matches the resource versions -// returned by the API server today. Non-numeric item versions are ignored. If -// no usable item version exists, we return an empty string and preserve the -// previous watch behavior. +// normalizeListResourceVersion returns a list ResourceVersion safe to seed a +// follow-up Watch with: the value as-is when it is a real version, or "" when +// it is the cached client's "0" sentinel. // -// In production the max-item fallback is effectively unreachable: the list+watch -// seed endpoints read through listForWatchSeed's uncached reader, which always -// returns a real list-level resource version, so the early return below wins. -// The fallback only runs on the degraded path where no direct reader is wired -// (tests, or no rest.Config) and the cached client reports "0"/"". There it is a -// best effort: the max item version may be older than the apiserver's watch -// window, in which case the follow-up watch simply restarts with a fresh list, -// i.e. the pre-change behavior. -func effectiveResourceVersion(rv string, itemVersions []string) string { - if rv != "" && rv != "0" { - return rv - } - var maxRV int64 - for _, v := range itemVersions { - n, err := strconv.ParseInt(v, 10, 64) - if err != nil || n <= 0 { - continue - } - if n > maxRV { - maxRV = n - } - } - if maxRV == 0 { +// The list+watch seed endpoints read through listForWatchSeed's uncached +// reader, which always returns a real list-level resource version. The "0" +// sentinel only appears on the degraded path where no direct reader is wired +// (tests, or no rest.Config) and the cached client is used; there we return "" +// so the client opens an unseeded watch — the pre-change behavior. +func normalizeListResourceVersion(rv string) string { + if rv == "0" { return "" } - return strconv.FormatInt(maxRV, 10) -} - -// effectiveResourceVersionForList returns an effective list resourceVersion for -// a typed list: the list-level resourceVersion when usable, otherwise the -// maximum item resourceVersion (see effectiveResourceVersion). itemRV extracts -// the resourceVersion from each (value-typed) list item, which a plain generic -// constraint cannot do because metav1.Object's methods have pointer receivers. -func effectiveResourceVersionForList[T any](rv string, items []T, itemRV func(T) string) string { - itemVersions := make([]string, len(items)) - for i := range items { - itemVersions[i] = itemRV(items[i]) - } - return effectiveResourceVersion(rv, itemVersions) + return rv } // buildWatchListOptions returns namespace-scoped list options for watch calls, diff --git a/pkg/server/resource_version_test.go b/pkg/server/resource_version_test.go index cc5a040792..66871efcbb 100644 --- a/pkg/server/resource_version_test.go +++ b/pkg/server/resource_version_test.go @@ -6,66 +6,21 @@ import ( "github.com/stretchr/testify/require" ) -func Test_effectiveResourceVersion(t *testing.T) { +func Test_normalizeListResourceVersion(t *testing.T) { t.Parallel() testCases := []struct { - name string - rv string - itemVersions []string - expected string + name string + rv string + expected string }{ - { - name: "real list-level rv passthrough", - rv: "12345", - expected: "12345", - }, - { - name: "zero list-level rv falls back to max item rv", - rv: "0", - itemVersions: []string{"100", "300", "200"}, - expected: "300", - }, - { - name: "empty list-level rv falls back to max item rv", - rv: "", - itemVersions: []string{"100", "300", "200"}, - expected: "300", - }, - { - name: "zero list-level rv with no items returns empty", - rv: "0", - itemVersions: []string{}, - expected: "", - }, - { - name: "items with non-numeric rvs are skipped", - rv: "0", - itemVersions: []string{"abc", "100", "def"}, - expected: "100", - }, - { - name: "all invalid item rvs returns empty", - rv: "", - itemVersions: []string{"", "abc"}, - expected: "", - }, - { - name: "real list-level rv wins over items", - rv: "999", - itemVersions: []string{"100", "200"}, - expected: "999", - }, - { - name: "zero item rvs are skipped", - rv: "0", - itemVersions: []string{"0", "0", "100"}, - expected: "100", - }, + {name: "real rv passthrough", rv: "12345", expected: "12345"}, + {name: "zero sentinel becomes empty", rv: "0", expected: ""}, + {name: "empty stays empty", rv: "", expected: ""}, } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { t.Parallel() - require.Equal(t, tc.expected, effectiveResourceVersion(tc.rv, tc.itemVersions)) + require.Equal(t, tc.expected, normalizeListResourceVersion(tc.rv)) }) } } From f5fa835afaca5758fcd08cdb172911cd76eb5d1c Mon Sep 17 00:00:00 2001 From: Jacob Boykin Date: Thu, 18 Jun 2026 11:12:43 -0400 Subject: [PATCH 3/3] refactor(server): source watch-seed reader from cluster APIReader Replace the bespoke uncached client built in the server package with the cluster's existing GetAPIReader, exposed via kubernetes.Client.APIReader. This is the same direct-read pattern the controllers and external webhooks server use, and it drops the parallel client.New, its RESTMapper/connection, and the RestConfig-into-server wiring. listForWatchSeed now reads through client.APIReader and authorizes the caller first, single-mode. Signed-off-by: Jacob Boykin --- pkg/server/kubernetes/client.go | 67 +++++++++++++++++------ pkg/server/kubernetes/client_test.go | 4 +- pkg/server/list_for_watch_seed.go | 38 ++++++------- pkg/server/list_for_watch_seed_test.go | 75 +++++++++++--------------- pkg/server/rest_test.go | 2 + pkg/server/server.go | 36 ------------- 6 files changed, 103 insertions(+), 119 deletions(-) diff --git a/pkg/server/kubernetes/client.go b/pkg/server/kubernetes/client.go index 23c00dc3e9..3a9f0262ad 100644 --- a/pkg/server/kubernetes/client.go +++ b/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, diff --git a/pkg/server/kubernetes/client_test.go b/pkg/server/kubernetes/client_test.go index efeb1726b4..697756be39 100644 --- a/pkg/server/kubernetes/client_test.go +++ b/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) } diff --git a/pkg/server/list_for_watch_seed.go b/pkg/server/list_for_watch_seed.go index bf3e74f050..a78a5499df 100644 --- a/pkg/server/list_for_watch_seed.go +++ b/pkg/server/list_for_watch_seed.go @@ -9,36 +9,30 @@ import ( kargoapi "github.com/akuity/kargo/api/v1alpha1" ) -// listForWatchSeed lists Kargo resources directly from the Kubernetes API, -// bypassing the API server's controller-runtime read cache. +// 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 going straight -// to the API for list+watch seed endpoints where the returned resourceVersion -// is used to start a follow-up watch. +// 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 direct reader does not enforce Kargo's RBAC, so we authorize the caller -// first via the authorizing client. resource is the lowercase resource name in -// the Kargo API group (e.g. "stages", "warehouses", "promotions", "freights"). -// When no direct reader is available (tests, or no rest.Config at construction -// time), listForWatchSeed falls back to the standard authorizing client, which -// performs its own SubjectAccessReview as part of List. +// 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.directReader == nil { - // Fallback: the authorizing client performs its own SAR per call. - if s.client == nil { - return fmt.Errorf("kubernetes client is not configured") - } - return s.client.List(ctx, list, opts...) + if s.client == nil { + return fmt.Errorf("kubernetes client is not configured") } - if s.authorizeFn == nil { return fmt.Errorf("authorize function is not configured") } @@ -46,9 +40,9 @@ func (s *server) listForWatchSeed( var listOpts client.ListOptions listOpts.ApplyOptions(opts) - // Authorize the user before bypassing the cache. The direct reader runs - // with the API server's own credentials, so without this check a caller - // could read data they would otherwise be denied. + // 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", @@ -59,5 +53,5 @@ func (s *server) listForWatchSeed( return err } - return s.directReader.List(ctx, list, opts...) + return s.client.APIReader().List(ctx, list, opts...) } diff --git a/pkg/server/list_for_watch_seed_test.go b/pkg/server/list_for_watch_seed_test.go index 5e84aae6fd..ae61356d87 100644 --- a/pkg/server/list_for_watch_seed_test.go +++ b/pkg/server/list_for_watch_seed_test.go @@ -17,17 +17,16 @@ import ( "github.com/akuity/kargo/pkg/server/kubernetes" ) -func TestServer_listForWatchSeed(t *testing.T) { - t.Parallel() - - scheme := mustNewScheme() - internalClient := fake.NewClientBuilder().WithScheme(scheme).Build() +// 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: scheme, + Scheme: mustNewScheme(), NewInternalClient: func( context.Context, *rest.Config, @@ -39,19 +38,26 @@ func TestServer_listForWatchSeed(t *testing.T) { }, ) require.NoError(t, err) + return kubeClient +} + +func TestServer_listForWatchSeed(t *testing.T) { + t.Parallel() + + scheme := mustNewScheme() - t.Run("authorizes and lists through direct reader", func(t *testing.T) { + 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 - directListCalled bool + authorized bool + authorizedVerb string + authorizedGVR schema.GroupVersionResource + authorizedSub string + authorizedKey client.ObjectKey + listCalled bool ) - directReader := fake.NewClientBuilder(). + internalClient := fake.NewClientBuilder(). WithScheme(scheme). WithInterceptorFuncs(interceptor.Funcs{ List: func( @@ -60,7 +66,7 @@ func TestServer_listForWatchSeed(t *testing.T) { list client.ObjectList, opts ...client.ListOption, ) error { - directListCalled = true + listCalled = true if err := cl.List(ctx, list, opts...); err != nil { return err } @@ -73,8 +79,7 @@ func TestServer_listForWatchSeed(t *testing.T) { Build() s := &server{ - client: kubeClient, - directReader: directReader, + client: newSeedKubeClient(t, internalClient), authorizeFn: func( _ context.Context, verb string, @@ -100,7 +105,7 @@ func TestServer_listForWatchSeed(t *testing.T) { ) require.NoError(t, err) require.True(t, authorized) - require.True(t, directListCalled) + require.True(t, listCalled) require.Equal(t, "list", authorizedVerb) require.Empty(t, authorizedSub) require.Equal(t, kargoapi.GroupVersion.WithResource("promotions"), authorizedGVR) @@ -108,12 +113,12 @@ func TestServer_listForWatchSeed(t *testing.T) { require.Equal(t, "42", promotions.ResourceVersion) }) - t.Run("authorization failure short-circuits direct list", func(t *testing.T) { + t.Run("authorization failure short-circuits the list", func(t *testing.T) { t.Parallel() authErr := errors.New("not authorized") - var directListCalled bool - directReader := fake.NewClientBuilder(). + var listCalled bool + internalClient := fake.NewClientBuilder(). WithScheme(scheme). WithInterceptorFuncs(interceptor.Funcs{ List: func( @@ -122,15 +127,14 @@ func TestServer_listForWatchSeed(t *testing.T) { client.ObjectList, ...client.ListOption, ) error { - directListCalled = true + listCalled = true return nil }, }). Build() s := &server{ - client: kubeClient, - directReader: directReader, + client: newSeedKubeClient(t, internalClient), authorizeFn: func( context.Context, string, @@ -149,37 +153,22 @@ func TestServer_listForWatchSeed(t *testing.T) { client.InNamespace("fake-project"), ) require.ErrorIs(t, err, authErr) - require.False(t, directListCalled) + require.False(t, listCalled) }) - t.Run("falls back to cached client when no direct reader is wired", func(t *testing.T) { + t.Run("errors when authorize function is not configured", func(t *testing.T) { t.Parallel() - var authorizeCalled bool s := &server{ - client: kubeClient, - authorizeFn: func( - context.Context, - string, - schema.GroupVersionResource, - string, - client.ObjectKey, - ) error { - authorizeCalled = true - return nil - }, + client: newSeedKubeClient(t, fake.NewClientBuilder().WithScheme(scheme).Build()), } - // Should not error and should not invoke authorizeFn directly; - // the cached authorizing client performs its own SAR per call, - // which is bypassed here via SkipAuthorization on kubeClient. err := s.listForWatchSeed( t.Context(), "promotions", &kargoapi.PromotionList{}, client.InNamespace("fake-project"), ) - require.NoError(t, err) - require.False(t, authorizeCalled) + require.ErrorContains(t, err, "authorize function is not configured") }) } diff --git a/pkg/server/rest_test.go b/pkg/server/rest_test.go index 4248dc3216..a2664d6e3b 100644 --- a/pkg/server/rest_test.go +++ b/pkg/server/rest_test.go @@ -113,6 +113,7 @@ func testRESTEndpoint( }, ) require.NoError(t, err) + s.authorizeFn = s.client.Authorize s.rolesDB = rbac.NewKubernetesRolesDatabase( s.client, s.client, @@ -226,6 +227,7 @@ func testRESTWatchEndpoint( }, ) require.NoError(t, err) + s.authorizeFn = s.client.Authorize s.rolesDB = rbac.NewKubernetesRolesDatabase( s.client, s.client, diff --git a/pkg/server/server.go b/pkg/server/server.go index 085e940ecf..8dfb73e7aa 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -50,17 +50,6 @@ type server struct { rolesDB rbac.RolesDatabase sender event.Sender - // directReader is a singleton, uncached controller-runtime client used by - // listForWatchSeed to bypass the API server's read cache on list+watch seed - // endpoints whose returned ResourceVersion is reused by a follow-up watch - // (currently Stages, Warehouses, Promotions, and the Freight list served by - // QueryFreight). It is wired once in NewServer from cfg.RestConfig and is - // nil in tests or when no rest.Config is available, in which case - // listForWatchSeed falls back to the standard authorizing client. The - // direct reader does NOT enforce Kargo's RBAC on its own; listForWatchSeed - // authorizes the caller before using it. - directReader client.Reader - // The following behaviors are overridable for testing purposes: // Common validations: @@ -203,31 +192,6 @@ func NewServer( s.getClusterAnalysisTemplateFn = rollouts.GetClusterAnalysisTemplate s.getAnalysisRunFn = rollouts.GetAnalysisRun - // Wire the singleton uncached direct reader once. listForWatchSeed uses it - // to bypass the cache on list+watch endpoints whose cached ResourceVersion - // can otherwise be stale enough to trigger client refetch loops. - if cfg.RestConfig != nil { - directReader, err := client.New( - cfg.RestConfig, - client.Options{ - Scheme: kubeClient.Scheme(), - Mapper: kubeClient.RESTMapper(), - }, - ) - if err != nil { - // Construction failure is unexpected with a valid rest.Config; log - // and continue without the optimization (listForWatchSeed falls back - // to the cached authorizing client). - logging.LoggerFromContext(context.Background()).Error( - err, - "error initializing direct Kubernetes reader; "+ - "falling back to cached client for watch seed endpoints", - ) - } else { - s.directReader = directReader - } - } - return s }