feat: add resourceVersion to Freight, Stage, Warehouse list and watch endpoints

api/service/v1alpha1/service.proto

@@ -463,6 +463,9 @@ message ListStagesRequest {
 message ListStagesResponse {
   // stages is the list of Stage resources found in the project.
   repeated github.com.akuity.kargo.api.v1alpha1.Stage stages = 1;
+  // resource_version is the Kubernetes ListMeta.resourceVersion of the
+  // underlying list, suitable for use as a starting point for Watch calls.
+  string resource_version = 2 [json_name = "resourceVersion"];
 }
 
 // ListImagesRequest is the request for listing images and their usage across stages.
@@ -519,6 +522,9 @@ message WatchStagesRequest {
   // When specified, only events for Stages that subscribe to at least one of
   // the named Warehouses are streamed.
   repeated string freight_origins = 3 [json_name = "freightOrigins"];
+  // resource_version, when set, is passed to the underlying Kubernetes Watch
+  // call so that only events newer than this version are received.
+  string resource_version = 4 [json_name = "resourceVersion"];
 }
 
 // WatchStagesResponse contains stage change notifications.
@@ -815,6 +821,9 @@ message WatchFreightRequest {
   // When specified, only events for Freight that originated from at least one
   // of the named Warehouses are streamed.
   repeated string origins = 2;
+  // resource_version, when set, is passed to the underlying Kubernetes Watch
+  // call so that only events newer than this version are received.
+  string resource_version = 3 [json_name = "resourceVersion"];
 }
 
 // WatchFreightResponse contains freight change notifications.
@@ -883,6 +892,10 @@ message QueryFreightRequest {
 message QueryFreightResponse {
   // groups maps group names to their corresponding freight lists.
   map<string, FreightList> groups = 1;
+  // resource_version is the Kubernetes ListMeta.resourceVersion of the
+  // underlying list, populated only when no stage or origin filters are
+  // applied. Suitable for use as a starting point for Watch calls.
+  string resource_version = 2 [json_name = "resourceVersion"];
 }
 
 // FreightList contains a list of freight resources.
@@ -948,6 +961,9 @@ message ListWarehousesRequest {
 message ListWarehousesResponse {
   // warehouses is the list of Warehouse resources found in the project.
   repeated github.com.akuity.kargo.api.v1alpha1.Warehouse warehouses = 1;
+  // resource_version is the Kubernetes ListMeta.resourceVersion of the
+  // underlying list, suitable for use as a starting point for Watch calls.
+  string resource_version = 2 [json_name = "resourceVersion"];
 }
 
 // GetWarehouseRequest is the request for retrieving details of a specific warehouse.
@@ -976,6 +992,9 @@ message WatchWarehousesRequest {
   string project = 1;
   // name is the name of a specific warehouse to watch, if empty all warehouses in the project are watched.
   string name = 2;
+  // resource_version, when set, is passed to the underlying Kubernetes Watch
+  // call so that only events newer than this version are received.
+  string resource_version = 3 [json_name = "resourceVersion"];
 }
 
 // WatchWarehousesResponse contains warehouse change notifications.

docs/docs/90-api-documentation.md

@@ -1193,6 +1193,7 @@ Stability is not guaranteed.
 | Field | Type | Description |
 | ----- | ---- | ----------- |
 | stages | [github.com.akuity.kargo.api.v1alpha1.Stage](#github-com-akuity-kargo-api-v1alpha1-Stage) |  stages is the list of Stage resources found in the project. |
+| resource_version | [string](#string) |  resource_version is the Kubernetes ListMeta.resourceVersion of the underlying list, suitable for use as a starting point for Watch calls. |
 
 <a name="akuity-io-kargo-service-v1alpha1-ListWarehousesRequest"></a>
 
@@ -1209,6 +1210,7 @@ Stability is not guaranteed.
 | Field | Type | Description |
 | ----- | ---- | ----------- |
 | warehouses | [github.com.akuity.kargo.api.v1alpha1.Warehouse](#github-com-akuity-kargo-api-v1alpha1-Warehouse) |  warehouses is the list of Warehouse resources found in the project. |
+| resource_version | [string](#string) |  resource_version is the Kubernetes ListMeta.resourceVersion of the underlying list, suitable for use as a starting point for Watch calls. |
 
 <a name="akuity-io-kargo-service-v1alpha1-OIDCConfig"></a>
 
@@ -1280,6 +1282,7 @@ Stability is not guaranteed.
 | Field | Type | Description |
 | ----- | ---- | ----------- |
 | groups | [QueryFreightResponse.GroupsEntry](#akuity-io-kargo-service-v1alpha1-QueryFreightResponse-GroupsEntry) |  groups maps group names to their corresponding freight lists. |
+| resource_version | [string](#string) |  resource_version is the Kubernetes ListMeta.resourceVersion of the underlying list, populated only when no stage or origin filters are applied. Suitable for use as a starting point for Watch calls. |
 
 <a name="akuity-io-kargo-service-v1alpha1-QueryFreightResponse-GroupsEntry"></a>
 
@@ -1531,6 +1534,7 @@ Stability is not guaranteed.
 | ----- | ---- | ----------- |
 | project | [string](#string) |  project is the name of the project whose freight should be watched. |
 | origins | [string](#string) |  origins is an optional list of Warehouse names to filter Freight by. When specified, only events for Freight that originated from at least one of the named Warehouses are streamed. |
+| resource_version | [string](#string) |  resource_version, when set, is passed to the underlying Kubernetes Watch call so that only events newer than this version are received. |
 
 <a name="akuity-io-kargo-service-v1alpha1-WatchFreightResponse"></a>
 
@@ -1603,6 +1607,7 @@ Stability is not guaranteed.
 | project | [string](#string) |  project is the name of the project whose stages should be watched. |
 | name | [string](#string) |  name is the name of a specific stage to watch, if empty all stages in the project are watched. |
 | freight_origins | [string](#string) |  freight_origins is an optional list of Warehouse names to filter Stages by. When specified, only events for Stages that subscribe to at least one of the named Warehouses are streamed. |
+| resource_version | [string](#string) |  resource_version, when set, is passed to the underlying Kubernetes Watch call so that only events newer than this version are received. |
 
 <a name="akuity-io-kargo-service-v1alpha1-WatchStagesResponse"></a>
 
@@ -1621,6 +1626,7 @@ Stability is not guaranteed.
 | ----- | ---- | ----------- |
 | project | [string](#string) |  project is the name of the project whose warehouses should be watched. |
 | name | [string](#string) |  name is the name of a specific warehouse to watch, if empty all warehouses in the project are watched. |
+| resource_version | [string](#string) |  resource_version, when set, is passed to the underlying Kubernetes Watch call so that only events newer than this version are received. |
 
 <a name="akuity-io-kargo-service-v1alpha1-WatchWarehousesResponse"></a>
 

pkg/server/list_stages_v1alpha1.go

@@ -10,6 +10,7 @@ import (
 
 	"connectrpc.com/connect"
 	"github.com/gin-gonic/gin"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	svcv1alpha1 "github.com/akuity/kargo/api/service/v1alpha1"
@@ -32,21 +33,24 @@ func (s *server) ListStages(
 
 	warehouses := req.Msg.GetFreightOrigins()
 
-	items, err := s.listStagesByWarehouses(ctx, project, warehouses)
+	list, err := s.listStagesByWarehouses(ctx, project, warehouses)
 	if err != nil {
 		return nil, fmt.Errorf("list stages: %w", err)
 	}
 
-	slices.SortFunc(items, func(a, b kargoapi.Stage) int {
+	slices.SortFunc(list.Items, func(a, b kargoapi.Stage) int {
 		return strings.Compare(a.Name, b.Name)
 	})
 
-	stages := make([]*kargoapi.Stage, len(items))
-	for idx := range items {
-		stages[idx] = &items[idx]
+	stages := make([]*kargoapi.Stage, len(list.Items))
+	rvs := make([]string, len(list.Items))
+	for idx := range list.Items {
+		stages[idx] = &list.Items[idx]
+		rvs[idx] = list.Items[idx].ResourceVersion
 	}
 	return connect.NewResponse(&svcv1alpha1.ListStagesResponse{
-		Stages: stages,
+		Stages:          stages,
+		ResourceVersion: effectiveResourceVersion(list.ResourceVersion, rvs),
 	}), nil
 }
 
@@ -67,24 +71,30 @@ 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 := s.listStagesByWarehouses(ctx, project, 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) {
+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))
+	watchOpts := []client.ListOption{client.InNamespace(project)}
+	if resourceVersion != "" {
+		watchOpts = append(watchOpts, &client.ListOptions{
+			Raw: &metav1.ListOptions{ResourceVersion: resourceVersion},
+		})
+	}
+	w, err := s.client.Watch(ctx, &kargoapi.StageList{}, watchOpts...)
 	if err != nil {
 		logger.Error(err, "failed to start watch")
 		_ = c.Error(fmt.Errorf("watch stages: %w", err))
@@ -133,26 +143,28 @@ func (s *server) watchStages(c *gin.Context, project string, warehouses []string
 // 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.
+// all Stages are returned. The returned StageList carries the ResourceVersion
+// from the underlying Kubernetes List call.
 func (s *server) listStagesByWarehouses(
 	ctx context.Context,
 	project string,
 	warehouses []string,
-) ([]kargoapi.Stage, error) {
+) (*kargoapi.StageList, error) {
 	var list kargoapi.StageList
 	if err := s.client.List(ctx, &list, client.InNamespace(project)); err != nil {
 		return nil, err
 	}
 	if len(warehouses) == 0 {
-		return list.Items, nil
+		return &list, nil
 	}
 	var stages []kargoapi.Stage
 	for _, stage := range list.Items {
 		if stageMatchesAnyWarehouse(&stage, warehouses) {
 			stages = append(stages, stage)
 		}
 	}
-	return stages, nil
+	list.Items = stages
+	return &list, nil
 }
 
 // stageMatchesAnyWarehouse returns true if the Stage requests Freight that

pkg/server/list_stages_v1alpha1_test.go

@@ -15,6 +15,7 @@ import (
 	"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"
 
 	svcv1alpha1 "github.com/akuity/kargo/api/service/v1alpha1"
 	kargoapi "github.com/akuity/kargo/api/v1alpha1"
@@ -240,6 +241,124 @@ func TestListStages(t *testing.T) {
 	}
 }
 
+func TestListStages_resourceVersion(t *testing.T) {
+	t.Parallel()
+
+	testNamespace := mustNewObject[corev1.Namespace]("testdata/namespace.yaml")
+
+	ctx := t.Context()
+
+	c, err := kubernetes.NewClient(
+		ctx,
+		&rest.Config{},
+		kubernetes.ClientOptions{
+			SkipAuthorization: true,
+			NewInternalClient: func(
+				context.Context,
+				*rest.Config,
+				*runtime.Scheme,
+			) (client.WithWatch, error) {
+				return fake.NewClientBuilder().
+					WithScheme(mustNewScheme()).
+					WithObjects(testNamespace).
+					WithInterceptorFuncs(interceptor.Funcs{
+						List: func(
+							ctx context.Context,
+							cl client.WithWatch,
+							list client.ObjectList,
+							opts ...client.ListOption,
+						) error {
+							if err := cl.List(ctx, list, opts...); err != nil {
+								return err
+							}
+							if sl, ok := list.(*kargoapi.StageList); ok {
+								sl.ResourceVersion = "42"
+							}
+							return nil
+						},
+					}).
+					Build(), nil
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	svr := &server{client: c}
+	svr.externalValidateProjectFn = validation.ValidateProject
+
+	res, err := svr.ListStages(ctx, connect.NewRequest(&svcv1alpha1.ListStagesRequest{
+		Project: "kargo-demo",
+	}))
+	require.NoError(t, err)
+	require.Equal(t, "42", res.Msg.GetResourceVersion())
+}
+
+func TestListStages_resourceVersionFallback(t *testing.T) {
+	t.Parallel()
+
+	testNamespace := mustNewObject[corev1.Namespace]("testdata/namespace.yaml")
+
+	ctx := t.Context()
+
+	c, err := kubernetes.NewClient(
+		ctx,
+		&rest.Config{},
+		kubernetes.ClientOptions{
+			SkipAuthorization: true,
+			NewInternalClient: func(
+				context.Context,
+				*rest.Config,
+				*runtime.Scheme,
+			) (client.WithWatch, error) {
+				return fake.NewClientBuilder().
+					WithScheme(mustNewScheme()).
+					WithObjects(testNamespace).
+					WithInterceptorFuncs(interceptor.Funcs{
+						List: func(
+							ctx context.Context,
+							cl client.WithWatch,
+							list client.ObjectList,
+							opts ...client.ListOption,
+						) error {
+							if err := cl.List(ctx, list, opts...); err != nil {
+								return err
+							}
+							if sl, ok := list.(*kargoapi.StageList); ok {
+								// Simulate what the controller-runtime cached client
+								// returns: "0" for list-level ResourceVersion.
+								sl.ResourceVersion = "0"
+								for i := range sl.Items {
+									sl.Items[i].ResourceVersion = "100"
+								}
+							}
+							return nil
+						},
+					}).
+					WithObjects(
+						&kargoapi.Stage{
+							ObjectMeta: metav1.ObjectMeta{
+								Name:      "test-stage",
+								Namespace: "kargo-demo",
+							},
+						},
+					).
+					Build(), nil
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	svr := &server{client: c}
+	svr.externalValidateProjectFn = validation.ValidateProject
+
+	res, err := svr.ListStages(ctx, connect.NewRequest(&svcv1alpha1.ListStagesRequest{
+		Project: "kargo-demo",
+	}))
+	require.NoError(t, err)
+	// When list-level ResourceVersion is "0", fallback to max(item.ResourceVersion)
+	require.Equal(t, "100", res.Msg.GetResourceVersion())
+}
+
 func Test_server_listStages(t *testing.T) {
 	testProject := &kargoapi.Project{
 		ObjectMeta: metav1.ObjectMeta{Name: "fake-project"},