[Backport release-1.4] fix(api): emit initial-events-end bookmark for core.cozystack.io watches (#2844)

# Description
Backport of #2786 to `release-1.4`.

pkg/registry/apps/application/rest.go

@@ -681,13 +681,19 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 	}
 	helmLabelSelector = labels.NewSelector().Add(labelRequirements...)
 
-	// Handle SendInitialEvents for WatchList feature (Kubernetes 1.27+)
-	// When sendInitialEvents=true, the client expects:
-	// 1. All existing resources as ADDED events
-	// 2. A Bookmark event with "k8s.io/initial-events-end": "true" annotation
-	// controller-runtime cache already sends ADDED events for all cached objects,
-	// so we just need to send the bookmark after those initial events
+	// Honor SendInitialEvents (WatchList): the client expects all existing
+	// objects as ADDED events, then a Bookmark annotated with
+	// metav1.InitialEventsAnnotationKey. controller-runtime's cache already
+	// replays the ADDED events, so we just emit the terminating bookmark.
 	sendInitialEvents := options.SendInitialEvents != nil && *options.SendInitialEvents
+	bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, options.ResourceVersion, func() runtime.Object {
+		app := &appsv1alpha1.Application{}
+		app.TypeMeta = metav1.TypeMeta{
+			APIVersion: appsv1alpha1.SchemeGroupVersion.String(),
+			Kind:       r.kindName,
+		}
+		return app
+	})
 
 	// Create a custom watcher to transform events
 	customW := &customWatcher{
@@ -702,6 +708,10 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 	helmWatcher, err := r.w.Watch(ctx, hrList, &client.ListOptions{
 		Namespace:     namespace,
 		LabelSelector: helmLabelSelector,
+		// Ask the backing watch for bookmarks on a WatchList request; the
+		// apiserver omits them by default, leaving the terminating
+		// initial-events-end bookmark with no reliable trigger.
+		Raw: &metav1.ListOptions{AllowWatchBookmarks: sendInitialEvents},
 	})
 	if err != nil {
 		klog.Errorf("Error setting up watch for HelmReleases: %v", err)
@@ -738,10 +748,6 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 			defer wmWatcherForCleanup.Stop()
 		}
 
-		// Track whether we've sent the initial-events-end bookmark
-		initialEventsEndSent := !sendInitialEvents // If not sendInitialEvents, consider it already sent
-		var lastResourceVersion string
-
 		// Buffer of WorkloadMonitor events that arrived before the initial
 		// snapshot finished. The watch-list contract requires the stream to
 		// deliver all ADDED events followed by the initial-events-end bookmark
@@ -762,48 +768,26 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 			}
 		}
 
-		drainPendingWMEvents := func() {
+		// send forwards an event, returning false if the watch or context ended.
+		send := func(ev watch.Event) bool {
+			select {
+			case customW.resultChan <- ev:
+				return true
+			case <-customW.stopChan:
+				return false
+			case <-ctx.Done():
+				return false
+			}
+		}
+
+		drainPendingWMEvents := func() bool {
 			for _, ev := range pendingWMEvents {
-				select {
-				case customW.resultChan <- ev:
-				case <-customW.stopChan:
-					return
-				case <-ctx.Done():
-					return
+				if !send(ev) {
+					return false
 				}
 			}
 			pendingWMEvents = nil
-		}
-
-		// Helper function to send initial-events-end bookmark
-		sendInitialEventsEndBookmark := func() {
-			if initialEventsEndSent {
-				return
-			}
-			initialEventsEndSent = true
-
-			bookmarkApp := &appsv1alpha1.Application{}
-			bookmarkApp.SetResourceVersion(lastResourceVersion)
-			bookmarkApp.TypeMeta = metav1.TypeMeta{
-				APIVersion: appsv1alpha1.SchemeGroupVersion.String(),
-				Kind:       r.kindName,
-			}
-			bookmarkApp.SetAnnotations(map[string]string{
-				"k8s.io/initial-events-end": "true",
-			})
-			bookmarkEvent := watch.Event{
-				Type:   watch.Bookmark,
-				Object: bookmarkApp,
-			}
-			klog.V(6).Infof("Sending initial-events-end bookmark with RV=%s", lastResourceVersion)
-			select {
-			case customW.resultChan <- bookmarkEvent:
-			case <-customW.stopChan:
-				return
-			case <-ctx.Done():
-				return
-			}
-			drainPendingWMEvents()
+			return true
 		}
 
 		// Process watch events
@@ -814,45 +798,25 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 					// The watcher has been closed
 					klog.Warning("HelmRelease watcher closed")
 					// Send initial-events-end bookmark before closing if not yet sent
-					sendInitialEventsEndBookmark()
+					if bookmark, ok := bookmarker.OnClose(); ok {
+						if send(bookmark) {
+							drainPendingWMEvents()
+						}
+					}
 					return
 				}
 
 				// Handle bookmark events - these are critical for informer sync
 				if event.Type == watch.Bookmark {
 					if hr, ok := event.Object.(*helmv2.HelmRelease); ok {
-						lastResourceVersion = hr.GetResourceVersion()
-
-						// If sendInitialEvents and we haven't sent initial-events-end yet,
-						// add the annotation to this bookmark
-						bookmarkApp := &appsv1alpha1.Application{}
-						bookmarkApp.SetResourceVersion(lastResourceVersion)
-						bookmarkApp.TypeMeta = metav1.TypeMeta{
-							APIVersion: appsv1alpha1.SchemeGroupVersion.String(),
-							Kind:       r.kindName,
-						}
-						justFlipped := false
-						if !initialEventsEndSent {
-							initialEventsEndSent = true
-							justFlipped = true
-							bookmarkApp.SetAnnotations(map[string]string{
-								"k8s.io/initial-events-end": "true",
-							})
-							klog.V(6).Infof("Sending initial-events-end bookmark with RV=%s", lastResourceVersion)
-						}
-						bookmarkEvent := watch.Event{
-							Type:   watch.Bookmark,
-							Object: bookmarkApp,
-						}
-						select {
-						case customW.resultChan <- bookmarkEvent:
-						case <-customW.stopChan:
-							return
-						case <-ctx.Done():
+						// controller-runtime bookmarks after the initial list
+						// replay; the first one becomes the terminating marker.
+						bookmark, initialEventsEnd := bookmarker.OnBackingBookmark(hr.GetResourceVersion())
+						if !send(bookmark) {
 							return
 						}
-						if justFlipped {
-							drainPendingWMEvents()
+						if initialEventsEnd && !drainPendingWMEvents() {
+							return
 						}
 					}
 					continue
@@ -871,8 +835,8 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 					continue
 				}
 
-				// Update lastResourceVersion for bookmark
-				lastResourceVersion = hr.GetResourceVersion()
+				// Track resourceVersion for the eventual bookmark
+				bookmarker.Observe(hr.GetResourceVersion())
 
 				// Apply manual field selector filtering (metadata.name and metadata.namespace)
 				// controller-runtime cache doesn't support field selectors
@@ -909,9 +873,15 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 					}
 				}
 
-				// If this is not an ADDED event and we haven't sent initial-events-end, send it now
-				if event.Type != watch.Added && !initialEventsEndSent {
-					sendInitialEventsEndBookmark()
+				// Emit the terminating bookmark before the first live event, then
+				// replay any buffered WorkloadMonitor events.
+				if bookmark, ok := bookmarker.BeforeLiveEvent(event.Type); ok {
+					if !send(bookmark) {
+						return
+					}
+					if !drainPendingWMEvents() {
+						return
+					}
 				}
 
 				// Skip ADDED events based on resourceVersion comparison
@@ -926,18 +896,8 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 				}
 				// When startingRV == 0, always send ADDED events (client wants full state)
 
-				// Create watch event with Application object
-				appEvent := watch.Event{
-					Type:   event.Type,
-					Object: &app,
-				}
-
 				// Send event to custom watcher
-				select {
-				case customW.resultChan <- appEvent:
-				case <-customW.stopChan:
-					return
-				case <-ctx.Done():
+				if !send(watch.Event{Type: event.Type, Object: &app}) {
 					return
 				}
 
@@ -1026,16 +986,12 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 				// Buffer WM-triggered events that arrive before the
 				// initial-events-end bookmark. They will be replayed in order
 				// immediately after the bookmark is emitted.
-				if !initialEventsEndSent {
+				if !bookmarker.Sent() {
 					pendingWMEvents = append(pendingWMEvents, outEvent)
 					continue
 				}
-				lastResourceVersion = wm.GetResourceVersion()
-				select {
-				case customW.resultChan <- outEvent:
-				case <-customW.stopChan:
-					return
-				case <-ctx.Done():
+				bookmarker.Observe(wm.GetResourceVersion())
+				if !send(outEvent) {
 					return
 				}
 

pkg/registry/core/tenantmodule/rest.go

@@ -366,6 +366,11 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 		}
 	}
 
+	// For a SendInitialEvents (WatchList) request, ask the backing watch for
+	// bookmarks — the apiserver omits them by default, which would leave the
+	// terminating initial-events-end bookmark with no reliable trigger.
+	sendInitialEvents := options.SendInitialEvents != nil && *options.SendInitialEvents
+
 	// Start watch on HelmRelease with label selector only
 	// Field selectors are not supported by controller-runtime cache
 	// See: https://github.com/kubernetes-sigs/controller-runtime/issues/612
@@ -373,12 +378,24 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 	helmWatcher, err := r.w.Watch(ctx, hrList, &client.ListOptions{
 		Namespace:     namespace,
 		LabelSelector: helmLabelSelector,
+		Raw:           &metav1.ListOptions{AllowWatchBookmarks: sendInitialEvents},
 	})
 	if err != nil {
 		klog.Errorf("Error setting up watch for HelmReleases: %v", err)
 		return nil, err
 	}
 
+	// Emit the initial-events-end bookmark after the initial ADDED events so
+	// client-go reflectors reach HasSynced.
+	bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, options.ResourceVersion, func() runtime.Object {
+		module := &corev1alpha1.TenantModule{}
+		module.TypeMeta = metav1.TypeMeta{
+			APIVersion: corev1alpha1.SchemeGroupVersion.String(),
+			Kind:       r.kindName,
+		}
+		return module
+	})
+
 	// Create a custom watcher to transform events
 	customW := &customWatcher{
 		resultChan: make(chan watch.Event),
@@ -390,32 +407,35 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 		defer close(customW.resultChan)
 		defer customW.underlying.Stop()
 
+		// send forwards an event, returning false if the watch or context ended.
+		send := func(ev watch.Event) bool {
+			select {
+			case customW.resultChan <- ev:
+				return true
+			case <-customW.stopChan:
+				return false
+			case <-ctx.Done():
+				return false
+			}
+		}
+
 		for {
 			select {
 			case event, ok := <-customW.underlying.ResultChan():
 				if !ok {
 					klog.Warning("HelmRelease watcher closed")
+					// Flush the terminating bookmark if still pending.
+					if bookmark, ok := bookmarker.OnClose(); ok {
+						send(bookmark)
+					}
 					return
 				}
 
 				// Handle bookmark events
 				if event.Type == watch.Bookmark {
 					if hr, ok := event.Object.(*helmv2.HelmRelease); ok {
-						bookmarkModule := &corev1alpha1.TenantModule{}
-						bookmarkModule.SetResourceVersion(hr.GetResourceVersion())
-						bookmarkModule.TypeMeta = metav1.TypeMeta{
-							APIVersion: corev1alpha1.SchemeGroupVersion.String(),
-							Kind:       r.kindName,
-						}
-						bookmarkEvent := watch.Event{
-							Type:   watch.Bookmark,
-							Object: bookmarkModule,
-						}
-						select {
-						case customW.resultChan <- bookmarkEvent:
-						case <-customW.stopChan:
-							return
-						case <-ctx.Done():
+						bookmark, _ := bookmarker.OnBackingBookmark(hr.GetResourceVersion())
+						if !send(bookmark) {
 							return
 						}
 					}
@@ -434,6 +454,7 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 					klog.V(4).Infof("Expected HelmRelease object, got %T", event.Object)
 					continue
 				}
+				bookmarker.Observe(hr.GetResourceVersion())
 
 				// Apply manual field selector filtering
 				if !fieldFilter.MatchesName(hr.Name) {
@@ -482,18 +503,15 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
 					}
 				}
 
-				// Create watch event with TenantModule object
-				moduleEvent := watch.Event{
-					Type:   event.Type,
-					Object: &module,
+				// Emit the initial-events-end bookmark before the first live event.
+				if bookmark, ok := bookmarker.BeforeLiveEvent(event.Type); ok {
+					if !send(bookmark) {
+						return
+					}
 				}
 
-				// Send event to custom watcher
-				select {
-				case customW.resultChan <- moduleEvent:
-				case <-customW.stopChan:
-					return
-				case <-ctx.Done():
+				// Send the translated TenantModule event.
+				if !send(watch.Event{Type: event.Type, Object: &module}) {
 					return
 				}
 

pkg/registry/core/tenantmodule/rest_watch_test.go

@@ -0,0 +1,118 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package tenantmodule
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	helmv2 "github.com/fluxcd/helm-controller/api/v2"
+	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	corev1alpha1 "github.com/cozystack/cozystack/pkg/apis/core/v1alpha1"
+)
+
+const testNamespace = "tenant-foo"
+
+// collectEvents drains up to n events from the watch, or returns early if the
+// channel closes or the timeout fires.
+func collectEvents(t *testing.T, w watch.Interface, n int, timeout time.Duration) []watch.Event {
+	t.Helper()
+	out := make([]watch.Event, 0, n)
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+	for len(out) < n {
+		select {
+		case ev, ok := <-w.ResultChan():
+			if !ok {
+				return out
+			}
+			out = append(out, ev)
+		case <-deadline.C:
+			return out
+		}
+	}
+	return out
+}
+
+// TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark asserts the
+// WatchList contract for TenantModule: ADDED events, then a bookmark annotated
+// with k8s.io/initial-events-end, then live events.
+//
+// fake.Client.Watch doesn't replay existing objects as ADDED, so the HelmRelease
+// is created after the watch starts and then mutated to drive a live event.
+func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := helmv2.AddToScheme(scheme); err != nil {
+		t.Fatalf("add helmv2 to scheme: %v", err)
+	}
+	fc := fake.NewClientBuilder().WithScheme(scheme).Build()
+	r := NewREST(fc, fc)
+
+	ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), testNamespace))
+	defer cancel()
+
+	sendInitialEvents := true
+	w, err := r.Watch(ctx, &metainternalversion.ListOptions{SendInitialEvents: &sendInitialEvents})
+	if err != nil {
+		t.Fatalf("Watch returned error: %v", err)
+	}
+	defer w.Stop()
+
+	hr := &helmv2.HelmRelease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "module-a",
+			Namespace: testNamespace,
+			Labels: map[string]string{
+				TenantModuleLabelKey: TenantModuleLabelValue,
+			},
+		},
+	}
+	if err := r.c.Create(ctx, hr); err != nil {
+		t.Fatalf("create HelmRelease: %v", err)
+	}
+	hr.Annotations = map[string]string{"touched": "1"}
+	if err := r.c.Update(ctx, hr); err != nil {
+		t.Fatalf("update HelmRelease: %v", err)
+	}
+
+	evs := collectEvents(t, w, 3, 2*time.Second)
+	if len(evs) < 3 {
+		t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs)
+	}
+
+	if evs[0].Type != watch.Added {
+		t.Fatalf("event[0]: expected Added, got %s", evs[0].Type)
+	}
+	added, ok := evs[0].Object.(*corev1alpha1.TenantModule)
+	if !ok {
+		t.Fatalf("event[0]: expected *TenantModule, got %T", evs[0].Object)
+	}
+	if added.Name != "module-a" {
+		t.Fatalf("event[0]: expected name module-a, got %q", added.Name)
+	}
+
+	if evs[1].Type != watch.Bookmark {
+		t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type)
+	}
+	bookmark, ok := evs[1].Object.(*corev1alpha1.TenantModule)
+	if !ok {
+		t.Fatalf("event[1]: expected *TenantModule, got %T", evs[1].Object)
+	}
+	if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" {
+		t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got)
+	}
+	if bookmark.ResourceVersion == "" {
+		t.Fatal("event[1]: expected non-empty resourceVersion on bookmark")
+	}
+
+	if evs[2].Type != watch.Modified {
+		t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type)
+	}
+}

pkg/registry/core/tenantnamespace/rest.go

@@ -170,10 +170,16 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 
 	nsList := &corev1.NamespaceList{}
 
+	// For a SendInitialEvents (WatchList) request, ask the backing watch for
+	// bookmarks — the apiserver omits them by default, which would leave the
+	// terminating initial-events-end bookmark with no reliable trigger.
+	sendInitialEvents := opts.SendInitialEvents != nil && *opts.SendInitialEvents
+
 	// Build upstream watch options with field and label selectors
 	rawOpts := &metav1.ListOptions{
-		Watch:           true,
-		ResourceVersion: opts.ResourceVersion,
+		Watch:               true,
+		ResourceVersion:     opts.ResourceVersion,
+		AllowWatchBookmarks: sendInitialEvents,
 	}
 	if opts.FieldSelector != nil {
 		rawOpts.FieldSelector = opts.FieldSelector.String()
@@ -195,26 +201,44 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 		}
 	}
 
+	// Emit the initial-events-end bookmark after the initial ADDED events so
+	// client-go reflectors reach HasSynced.
+	bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, opts.ResourceVersion, func() runtime.Object {
+		return &corev1alpha1.TenantNamespace{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: corev1alpha1.SchemeGroupVersion.String(),
+				Kind:       "TenantNamespace",
+			},
+		}
+	})
+
 	events := make(chan watch.Event)
 	pw := watch.NewProxyWatcher(events)
 
 	go func() {
 		defer pw.Stop()
+		defer nsWatch.Stop()
+
+		// send forwards an event, returning false if the watch or context ended.
+		send := func(ev watch.Event) bool {
+			select {
+			case events <- ev:
+				return true
+			case <-pw.StopChan():
+				return false
+			case <-ctx.Done():
+				return false
+			}
+		}
 
 		for ev := range nsWatch.ResultChan() {
 			// Handle bookmark events
 			if ev.Type == watch.Bookmark {
 				if ns, ok := ev.Object.(*corev1.Namespace); ok {
-					out := &corev1alpha1.TenantNamespace{
-						TypeMeta: metav1.TypeMeta{
-							APIVersion: corev1alpha1.SchemeGroupVersion.String(),
-							Kind:       "TenantNamespace",
-						},
-						ObjectMeta: metav1.ObjectMeta{
-							ResourceVersion: ns.ResourceVersion,
-						},
+					bookmark, _ := bookmarker.OnBackingBookmark(ns.ResourceVersion)
+					if !send(bookmark) {
+						return
 					}
-					events <- watch.Event{Type: watch.Bookmark, Object: out}
 				}
 				continue
 			}
@@ -223,6 +247,7 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 			if !ok || !strings.HasPrefix(ns.Name, prefix) {
 				continue
 			}
+			bookmarker.Observe(ns.ResourceVersion)
 
 			// Apply defensive filtering for field and label selectors
 			if opts.FieldSelector != nil {
@@ -274,7 +299,21 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 			}
 			// When startingRV == 0, always send ADDED events (client wants full state)
 
-			events <- watch.Event{Type: ev.Type, Object: out}
+			// Emit the initial-events-end bookmark before the first live event.
+			if bookmark, ok := bookmarker.BeforeLiveEvent(ev.Type); ok {
+				if !send(bookmark) {
+					return
+				}
+			}
+
+			if !send(watch.Event{Type: ev.Type, Object: out}) {
+				return
+			}
+		}
+
+		// Backing watcher closed: flush the terminating bookmark if still pending.
+		if bookmark, ok := bookmarker.OnClose(); ok {
+			send(bookmark)
 		}
 	}()
 

pkg/registry/core/tenantnamespace/rest_watch_test.go

@@ -0,0 +1,121 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package tenantnamespace
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	corev1alpha1 "github.com/cozystack/cozystack/pkg/apis/core/v1alpha1"
+)
+
+// collectEvents drains up to n events from the watch, or returns early if the
+// channel closes or the timeout fires.
+func collectEvents(t *testing.T, w watch.Interface, n int, timeout time.Duration) []watch.Event {
+	t.Helper()
+	out := make([]watch.Event, 0, n)
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+	for len(out) < n {
+		select {
+		case ev, ok := <-w.ResultChan():
+			if !ok {
+				return out
+			}
+			out = append(out, ev)
+		case <-deadline.C:
+			return out
+		}
+	}
+	return out
+}
+
+// TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark asserts the
+// WatchList contract for TenantNamespace: ADDED events, then a bookmark
+// annotated with k8s.io/initial-events-end, then live events.
+//
+// The user is in system:masters so the access check passes without RBAC
+// fixtures. fake.Client.Watch doesn't replay existing objects as ADDED, so the
+// namespace is created after the watch starts and then mutated to drive a live
+// event.
+func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add corev1 to scheme: %v", err)
+	}
+	fc := fake.NewClientBuilder().WithScheme(scheme).Build()
+	r := &REST{
+		c: fc,
+		w: fc,
+		gvr: schema.GroupVersionResource{
+			Group:    corev1alpha1.GroupName,
+			Version:  "v1alpha1",
+			Resource: "tenantnamespaces",
+		},
+	}
+
+	u := &user.DefaultInfo{Name: "admin", Groups: []string{"system:masters"}}
+	ctx, cancel := context.WithCancel(request.WithUser(context.Background(), u))
+	defer cancel()
+
+	sendInitialEvents := true
+	w, err := r.Watch(ctx, &metainternal.ListOptions{SendInitialEvents: &sendInitialEvents})
+	if err != nil {
+		t.Fatalf("Watch returned error: %v", err)
+	}
+	defer w.Stop()
+
+	ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-foo"}}
+	if err := r.c.Create(ctx, ns); err != nil {
+		t.Fatalf("create namespace: %v", err)
+	}
+	ns.Labels = map[string]string{"touched": "1"}
+	if err := r.c.Update(ctx, ns); err != nil {
+		t.Fatalf("update namespace: %v", err)
+	}
+
+	evs := collectEvents(t, w, 3, 2*time.Second)
+	if len(evs) < 3 {
+		t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs)
+	}
+
+	if evs[0].Type != watch.Added {
+		t.Fatalf("event[0]: expected Added, got %s", evs[0].Type)
+	}
+	added, ok := evs[0].Object.(*corev1alpha1.TenantNamespace)
+	if !ok {
+		t.Fatalf("event[0]: expected *TenantNamespace, got %T", evs[0].Object)
+	}
+	if added.Name != "tenant-foo" {
+		t.Fatalf("event[0]: expected name tenant-foo, got %q", added.Name)
+	}
+
+	if evs[1].Type != watch.Bookmark {
+		t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type)
+	}
+	bookmark, ok := evs[1].Object.(*corev1alpha1.TenantNamespace)
+	if !ok {
+		t.Fatalf("event[1]: expected *TenantNamespace, got %T", evs[1].Object)
+	}
+	if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" {
+		t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got)
+	}
+	if bookmark.ResourceVersion == "" {
+		t.Fatal("event[1]: expected non-empty resourceVersion on bookmark")
+	}
+
+	if evs[2].Type != watch.Modified {
+		t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type)
+	}
+}

pkg/registry/core/tenantsecret/rest.go

@@ -465,13 +465,19 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 		return watch.NewProxyWatcher(ch), nil
 	}
 
+	// For a SendInitialEvents (WatchList) request, ask the backing watch for
+	// bookmarks — the apiserver omits them by default, which would leave the
+	// terminating initial-events-end bookmark with no reliable trigger.
+	sendInitialEvents := opts.SendInitialEvents != nil && *opts.SendInitialEvents
+
 	secList := &corev1.SecretList{}
 	base, err := r.w.Watch(ctx, secList, &client.ListOptions{
 		Namespace:     ns,
 		LabelSelector: ls,
 		Raw: &metav1.ListOptions{
-			Watch:           true,
-			ResourceVersion: opts.ResourceVersion,
+			Watch:               true,
+			ResourceVersion:     opts.ResourceVersion,
+			AllowWatchBookmarks: sendInitialEvents,
 		},
 	})
 	if err != nil {
@@ -486,26 +492,44 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 		}
 	}
 
+	// Emit the initial-events-end bookmark after the initial ADDED events so
+	// client-go reflectors reach HasSynced.
+	bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, opts.ResourceVersion, func() runtime.Object {
+		return &corev1alpha1.TenantSecret{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: corev1alpha1.SchemeGroupVersion.String(),
+				Kind:       kindTenantSecret,
+			},
+		}
+	})
+
 	ch := make(chan watch.Event)
 	proxy := watch.NewProxyWatcher(ch)
 
 	go func() {
 		defer proxy.Stop()
+		defer base.Stop()
+
+		// send forwards an event, returning false if the watch or context ended.
+		send := func(ev watch.Event) bool {
+			select {
+			case ch <- ev:
+				return true
+			case <-proxy.StopChan():
+				return false
+			case <-ctx.Done():
+				return false
+			}
+		}
 
 		for ev := range base.ResultChan() {
 			// Handle bookmark events
 			if ev.Type == watch.Bookmark {
 				if sec, ok := ev.Object.(*corev1.Secret); ok {
-					out := &corev1alpha1.TenantSecret{
-						TypeMeta: metav1.TypeMeta{
-							APIVersion: corev1alpha1.SchemeGroupVersion.String(),
-							Kind:       kindTenantSecret,
-						},
-						ObjectMeta: metav1.ObjectMeta{
-							ResourceVersion: sec.ResourceVersion,
-						},
+					bookmark, _ := bookmarker.OnBackingBookmark(sec.ResourceVersion)
+					if !send(bookmark) {
+						return
 					}
-					ch <- watch.Event{Type: watch.Bookmark, Object: out}
 				}
 				continue
 			}
@@ -514,6 +538,7 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 			if !ok || sec == nil {
 				continue
 			}
+			bookmarker.Observe(sec.ResourceVersion)
 
 			// Defensive: post-filter against the merged selector. The underlying
 			// watch already filters by label, but this guards against any client
@@ -539,10 +564,21 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
 			}
 			// When startingRV == 0, always send ADDED events (client wants full state)
 
-			ch <- watch.Event{
-				Type:   ev.Type,
-				Object: tenant,
+			// Emit the initial-events-end bookmark before the first live event.
+			if bookmark, ok := bookmarker.BeforeLiveEvent(ev.Type); ok {
+				if !send(bookmark) {
+					return
+				}
 			}
+
+			if !send(watch.Event{Type: ev.Type, Object: tenant}) {
+				return
+			}
+		}
+
+		// Backing watcher closed: flush the terminating bookmark if still pending.
+		if bookmark, ok := bookmarker.OnClose(); ok {
+			send(bookmark)
 		}
 	}()
 

pkg/registry/core/tenantsecret/rest_watch_test.go

@@ -0,0 +1,100 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package tenantsecret
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/apiserver/pkg/endpoints/request"
+
+	corev1alpha1 "github.com/cozystack/cozystack/pkg/apis/core/v1alpha1"
+)
+
+// collectEvents drains up to n events from the watch, or returns early if the
+// channel closes or the timeout fires.
+func collectEvents(t *testing.T, w watch.Interface, n int, timeout time.Duration) []watch.Event {
+	t.Helper()
+	out := make([]watch.Event, 0, n)
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+	for len(out) < n {
+		select {
+		case ev, ok := <-w.ResultChan():
+			if !ok {
+				return out
+			}
+			out = append(out, ev)
+		case <-deadline.C:
+			return out
+		}
+	}
+	return out
+}
+
+// TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark asserts the
+// WatchList contract for TenantSecret: ADDED events, then a bookmark annotated
+// with k8s.io/initial-events-end, then live events.
+//
+// fake.Client.Watch doesn't replay existing objects as ADDED, so the secret is
+// created after the watch starts and then mutated to drive a live event.
+func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) {
+	r := newTestREST(t)
+
+	ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), testNamespace))
+	defer cancel()
+
+	sendInitialEvents := true
+	w, err := r.Watch(ctx, &metainternal.ListOptions{SendInitialEvents: &sendInitialEvents})
+	if err != nil {
+		t.Fatalf("Watch returned error: %v", err)
+	}
+	defer w.Stop()
+
+	sec := makeTenantSecret("harbor-test-credentials", nil)
+	if err := r.c.Create(ctx, sec); err != nil {
+		t.Fatalf("create secret: %v", err)
+	}
+	sec.Annotations = map[string]string{"touched": "1"}
+	if err := r.c.Update(ctx, sec); err != nil {
+		t.Fatalf("update secret: %v", err)
+	}
+
+	evs := collectEvents(t, w, 3, 2*time.Second)
+	if len(evs) < 3 {
+		t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs)
+	}
+
+	if evs[0].Type != watch.Added {
+		t.Fatalf("event[0]: expected Added, got %s", evs[0].Type)
+	}
+	added, ok := evs[0].Object.(*corev1alpha1.TenantSecret)
+	if !ok {
+		t.Fatalf("event[0]: expected *TenantSecret, got %T", evs[0].Object)
+	}
+	if added.Name != "harbor-test-credentials" {
+		t.Fatalf("event[0]: expected name harbor-test-credentials, got %q", added.Name)
+	}
+
+	if evs[1].Type != watch.Bookmark {
+		t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type)
+	}
+	bookmark, ok := evs[1].Object.(*corev1alpha1.TenantSecret)
+	if !ok {
+		t.Fatalf("event[1]: expected *TenantSecret, got %T", evs[1].Object)
+	}
+	if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" {
+		t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got)
+	}
+	if bookmark.ResourceVersion == "" {
+		t.Fatal("event[1]: expected non-empty resourceVersion on bookmark")
+	}
+
+	if evs[2].Type != watch.Modified {
+		t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type)
+	}
+}

pkg/registry/utils.go

@@ -20,7 +20,9 @@ import (
 	"strconv"
 
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 // MaxResourceVersion returns the maximum resourceVersion from all items in a list.
@@ -55,3 +57,89 @@ func MaxResourceVersion(list runtime.Object) (string, error) {
 
 	return strconv.FormatUint(max, 10), nil
 }
+
+// InitialEventsBookmarker implements the WatchList initial-events-end contract
+// for a Watch that translates events from a backing controller-runtime watcher.
+// It annotates the first backing Bookmark as the terminating marker, falling
+// back to emitting one before the first live event or when the watcher closes.
+//
+// Construct with NewInitialEventsBookmarker. Methods are not safe for concurrent
+// use; call them from a single watch goroutine.
+type InitialEventsBookmarker struct {
+	// newBookmark returns an empty, TypeMeta-populated object of the translated
+	// resource type; the bookmarker fills in the resourceVersion and annotation.
+	newBookmark func() runtime.Object
+	sent        bool
+	lastRV      string
+}
+
+// NewInitialEventsBookmarker returns a bookmarker for a watch. When
+// sendInitialEvents is false every method is a no-op. startingRV seeds the
+// resourceVersion (the client's requested ResourceVersion) so the terminating
+// bookmark carries a valid version even if the watcher closes before any
+// backing event is observed.
+func NewInitialEventsBookmarker(sendInitialEvents bool, startingRV string, newBookmark func() runtime.Object) *InitialEventsBookmarker {
+	return &InitialEventsBookmarker{
+		newBookmark: newBookmark,
+		lastRV:      startingRV,
+		sent:        !sendInitialEvents,
+	}
+}
+
+// Sent reports whether the initial-events-end bookmark has been emitted (or was
+// never required). Callers buffering live events use it to decide when to start
+// delivering them.
+func (b *InitialEventsBookmarker) Sent() bool { return b.sent }
+
+// Observe records the resourceVersion of the latest backing event so the
+// bookmark carries an up-to-date version. Empty versions are ignored.
+func (b *InitialEventsBookmarker) Observe(resourceVersion string) {
+	if resourceVersion != "" {
+		b.lastRV = resourceVersion
+	}
+}
+
+// bookmarkEvent builds a Bookmark at the tracked resourceVersion, annotated as
+// the terminating marker when end is true.
+func (b *InitialEventsBookmarker) bookmarkEvent(end bool) watch.Event {
+	obj := b.newBookmark()
+	if accessor, err := meta.Accessor(obj); err == nil {
+		accessor.SetResourceVersion(b.lastRV)
+		if end {
+			accessor.SetAnnotations(map[string]string{metav1.InitialEventsAnnotationKey: "true"})
+		}
+	}
+	return watch.Event{Type: watch.Bookmark, Object: obj}
+}
+
+// OnBackingBookmark forwards the backing watcher's Bookmark at rv. The first one
+// after the initial events becomes the terminating marker; the returned bool
+// reports whether this call emitted it, so callers can flush buffered events.
+func (b *InitialEventsBookmarker) OnBackingBookmark(rv string) (event watch.Event, initialEventsEnd bool) {
+	b.Observe(rv)
+	if b.sent {
+		return b.bookmarkEvent(false), false
+	}
+	b.sent = true
+	return b.bookmarkEvent(true), true
+}
+
+// BeforeLiveEvent returns the terminating bookmark and true when a live
+// (non-ADDED) event is about to be delivered and the marker is still pending.
+func (b *InitialEventsBookmarker) BeforeLiveEvent(eventType watch.EventType) (event watch.Event, needed bool) {
+	if b.sent || eventType == watch.Added {
+		return watch.Event{}, false
+	}
+	b.sent = true
+	return b.bookmarkEvent(true), true
+}
+
+// OnClose returns the terminating bookmark and true if the backing watcher
+// closed before the marker was emitted.
+func (b *InitialEventsBookmarker) OnClose() (event watch.Event, needed bool) {
+	if b.sent {
+		return watch.Event{}, false
+	}
+	b.sent = true
+	return b.bookmarkEvent(true), true
+}

pkg/registry/utils_test.go

@@ -0,0 +1,169 @@
+/*
+Copyright 2024 The Cozystack Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package registry
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+// newTestBookmarkObject returns a minimal object for the bookmarker to stamp.
+func newTestBookmarkObject() runtime.Object {
+	return &metav1.PartialObjectMetadata{
+		TypeMeta: metav1.TypeMeta{APIVersion: "core.cozystack.io/v1alpha1", Kind: "TenantSecret"},
+	}
+}
+
+// assertInitialEventsEnd verifies ev is a Bookmark carrying
+// metav1.InitialEventsAnnotationKey=true and a non-empty resourceVersion.
+func assertInitialEventsEnd(t *testing.T, ev watch.Event, wantRV string) {
+	t.Helper()
+	if ev.Type != watch.Bookmark {
+		t.Fatalf("expected Bookmark event, got %s", ev.Type)
+	}
+	obj, ok := ev.Object.(*metav1.PartialObjectMetadata)
+	if !ok {
+		t.Fatalf("expected *PartialObjectMetadata, got %T", ev.Object)
+	}
+	if got := obj.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" {
+		t.Fatalf("expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got)
+	}
+	if obj.ResourceVersion == "" {
+		t.Fatal("expected non-empty resourceVersion on terminating bookmark")
+	}
+	if wantRV != "" && obj.ResourceVersion != wantRV {
+		t.Fatalf("expected resourceVersion %q, got %q", wantRV, obj.ResourceVersion)
+	}
+}
+
+func TestInitialEventsBookmarker_Disabled(t *testing.T) {
+	b := NewInitialEventsBookmarker(false, "", newTestBookmarkObject)
+
+	if !b.Sent() {
+		t.Fatal("disabled bookmarker should report Sent()==true")
+	}
+	if _, ok := b.BeforeLiveEvent(watch.Modified); ok {
+		t.Fatal("disabled bookmarker should never need a before-live bookmark")
+	}
+	if _, ok := b.OnClose(); ok {
+		t.Fatal("disabled bookmarker should never need an on-close bookmark")
+	}
+	// A backing bookmark is still forwarded, but never carries the marker.
+	ev, end := b.OnBackingBookmark("7")
+	if end {
+		t.Fatal("disabled bookmarker should not mark a backing bookmark as initial-events-end")
+	}
+	if ev.Type != watch.Bookmark {
+		t.Fatalf("expected forwarded Bookmark, got %s", ev.Type)
+	}
+	if obj := ev.Object.(*metav1.PartialObjectMetadata); obj.Annotations[metav1.InitialEventsAnnotationKey] != "" {
+		t.Fatal("disabled bookmarker must not annotate forwarded bookmarks")
+	}
+}
+
+func TestInitialEventsBookmarker_BackingBookmarkTerminatesStream(t *testing.T) {
+	b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject)
+	b.Observe("12")
+
+	ev, end := b.OnBackingBookmark("15")
+	if !end {
+		t.Fatal("first backing bookmark should terminate the initial stream")
+	}
+	assertInitialEventsEnd(t, ev, "15")
+	if !b.Sent() {
+		t.Fatal("bookmarker should report Sent()==true after terminating bookmark")
+	}
+
+	// Subsequent backing bookmarks are forwarded without the marker.
+	ev2, end2 := b.OnBackingBookmark("20")
+	if end2 {
+		t.Fatal("second backing bookmark must not re-terminate the stream")
+	}
+	if obj := ev2.Object.(*metav1.PartialObjectMetadata); obj.Annotations[metav1.InitialEventsAnnotationKey] != "" {
+		t.Fatal("subsequent bookmark must not carry the initial-events-end annotation")
+	}
+}
+
+func TestInitialEventsBookmarker_BeforeLiveEvent(t *testing.T) {
+	b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject)
+	b.Observe("9")
+
+	// ADDED events are part of the initial stream and never trigger the marker.
+	if _, ok := b.BeforeLiveEvent(watch.Added); ok {
+		t.Fatal("ADDED event should not trigger the initial-events-end bookmark")
+	}
+
+	ev, ok := b.BeforeLiveEvent(watch.Modified)
+	if !ok {
+		t.Fatal("first non-ADDED event should trigger the initial-events-end bookmark")
+	}
+	assertInitialEventsEnd(t, ev, "9")
+
+	// Once sent, later live events do not re-trigger it.
+	if _, ok := b.BeforeLiveEvent(watch.Deleted); ok {
+		t.Fatal("initial-events-end bookmark should only be emitted once")
+	}
+}
+
+func TestInitialEventsBookmarker_OnClose(t *testing.T) {
+	b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject)
+	b.Observe("42")
+
+	ev, ok := b.OnClose()
+	if !ok {
+		t.Fatal("closing before the marker was sent should flush a terminating bookmark")
+	}
+	assertInitialEventsEnd(t, ev, "42")
+
+	if _, ok := b.OnClose(); ok {
+		t.Fatal("on-close bookmark must not be emitted twice")
+	}
+}
+
+func TestInitialEventsBookmarker_ObserveIgnoresEmpty(t *testing.T) {
+	b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject)
+	b.Observe("5")
+	b.Observe("") // must not clobber the tracked version
+
+	ev, _ := b.OnClose()
+	assertInitialEventsEnd(t, ev, "5")
+}
+
+func TestInitialEventsBookmarker_SeedsStartingResourceVersion(t *testing.T) {
+	// Watcher closes before any backing event is observed: the terminating
+	// bookmark falls back to the client's requested resourceVersion rather than
+	// an empty one.
+	b := NewInitialEventsBookmarker(true, "100", newTestBookmarkObject)
+
+	ev, ok := b.OnClose()
+	if !ok {
+		t.Fatal("expected a terminating bookmark on close")
+	}
+	assertInitialEventsEnd(t, ev, "100")
+}
+
+func TestInitialEventsBookmarker_ObserveOverridesStartingResourceVersion(t *testing.T) {
+	// Once a backing event is observed, its resourceVersion supersedes the seed.
+	b := NewInitialEventsBookmarker(true, "100", newTestBookmarkObject)
+	b.Observe("150")
+
+	ev, _ := b.OnClose()
+	assertInitialEventsEnd(t, ev, "150")
+}