✨ POC for a read-your-own-write client

pkg/cache/cache.go

@@ -70,6 +70,37 @@ type Cache interface {
 	Informers
 }
 
+// internalCache is implemented by all cache implementations in controller-runtime. It
+// adds the requirement methods to be able to work for a consistent client.
+type internalCache interface {
+	Cache
+	// SetMinimumRVForGVKAndKey causes subsequent Get requests for the given
+	// GVK and key to block until the informer for that GVK has observed a
+	// resource version >= rv (or the context times out). For List requests
+	// on the given GVK, it blocks until the highest minimum RV across all
+	// keys for that GVK has been observed.
+	//
+	// TODO: This shouldn't be part of the public interface
+	SetMinimumRVForGVKAndKey(gvk schema.GroupVersionKind, key client.ObjectKey, rv int64)
+
+	// AddRequiredDeleteForGVKKeyAndUID causes subsequent Get requests for the
+	// given GVK and key to block until the UID has been observed as deleted.
+	// For List requests on the given GVK, it blocks until all pending delete
+	// UIDs across all keys for that GVK have been observed.
+	//
+	//
+	// An informer for the given object must have been created before the delete
+	// added here was executed, otherwise this will cause a deadlock.
+	//
+	// TODO: This shouldn't be part of the public interface
+	AddRequiredDeleteForObject(obj client.Object) error
+
+	// RemoveRequiredDeleteForObject removes a previously added pending delete.
+	//
+	// TODO: This shouldn't be part of the public interface
+	RemoveRequiredDeleteForObject(obj client.Object) error
+}
+
 // Informers knows how to create or fetch informers for different
 // group-version-kinds, and add indices to those informers.  It's safe to call
 // GetInformer from multiple threads.
@@ -133,6 +164,11 @@ type Informer interface {
 
 	// IsStopped returns true if the informer has been stopped.
 	IsStopped() bool
+
+	// LastSyncResourceVersion is the resource version observed when last synced with the underlying
+	// store. The value returned is not synchronized with access to the underlying store and is not
+	// thread-safe.
+	LastSyncResourceVersion() string
 }
 
 // AllNamespaces should be used as the map key to deliminate namespace settings
@@ -432,7 +468,7 @@ func New(cfg *rest.Config, opts Options) (Cache, error) {
 
 	newCacheFunc := newCache(cfg, opts)
 
-	var defaultCache Cache
+	var defaultCache internalCache
 	if len(opts.DefaultNamespaces) > 0 {
 		defaultConfig := optionDefaultsToConfig(&opts)
 		defaultCache = newMultiNamespaceCache(newCacheFunc, opts.Scheme, opts.Mapper, opts.DefaultNamespaces, &defaultConfig)
@@ -446,7 +482,7 @@ func New(cfg *rest.Config, opts Options) (Cache, error) {
 
 	delegating := &delegatingByGVKCache{
 		scheme:       opts.Scheme,
-		caches:       make(map[schema.GroupVersionKind]Cache, len(opts.ByObject)),
+		caches:       make(map[schema.GroupVersionKind]internalCache, len(opts.ByObject)),
 		defaultCache: defaultCache,
 	}
 
@@ -455,7 +491,7 @@ func New(cfg *rest.Config, opts Options) (Cache, error) {
 		if err != nil {
 			return nil, fmt.Errorf("failed to get GVK for type %T: %w", obj, err)
 		}
-		var cache Cache
+		var cache internalCache
 		if len(config.Namespaces) > 0 {
 			cache = newMultiNamespaceCache(newCacheFunc, opts.Scheme, opts.Mapper, config.Namespaces, nil)
 		} else {
@@ -503,10 +539,10 @@ func byObjectToConfig(byObject ByObject) Config {
 	}
 }
 
-type newCacheFunc func(config Config, namespace string) Cache
+type newCacheFunc func(config Config, namespace string) internalCache
 
 func newCache(restConfig *rest.Config, opts Options) newCacheFunc {
-	return func(config Config, namespace string) Cache {
+	return func(config Config, namespace string) internalCache {
 		return &informerCache{
 			scheme: opts.Scheme,
 			Informers: internal.NewInformers(restConfig, &internal.InformersOpts{

pkg/cache/delegating_by_gvk_cache.go

@@ -18,6 +18,7 @@ package cache
 
 import (
 	"context"
+	"fmt"
 	"maps"
 	"slices"
 	"strings"
@@ -33,8 +34,8 @@ import (
 // and uses the defaultCache otherwise.
 type delegatingByGVKCache struct {
 	scheme       *runtime.Scheme
-	caches       map[schema.GroupVersionKind]Cache
-	defaultCache Cache
+	caches       map[schema.GroupVersionKind]internalCache
+	defaultCache internalCache
 }
 
 func (dbt *delegatingByGVKCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
@@ -69,6 +70,26 @@ func (dbt *delegatingByGVKCache) GetInformer(ctx context.Context, obj client.Obj
 	return cache.GetInformer(ctx, obj, opts...)
 }
 
+func (dbt *delegatingByGVKCache) SetMinimumRVForGVKAndKey(gvk schema.GroupVersionKind, key client.ObjectKey, rv int64) {
+	dbt.cacheForGVK(gvk).SetMinimumRVForGVKAndKey(gvk, key, rv)
+}
+
+func (dbt *delegatingByGVKCache) AddRequiredDeleteForObject(obj client.Object) error {
+	cache, err := dbt.cacheForObject(obj)
+	if err != nil {
+		return err
+	}
+	return cache.AddRequiredDeleteForObject(obj)
+}
+
+func (dbt *delegatingByGVKCache) RemoveRequiredDeleteForObject(obj client.Object) error {
+	cache, err := dbt.cacheForObject(obj)
+	if err != nil {
+		return fmt.Errorf("getting cache for object: %w", err)
+	}
+	return cache.RemoveRequiredDeleteForObject(obj)
+}
+
 func (dbt *delegatingByGVKCache) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind, opts ...InformerGetOption) (Informer, error) {
 	return dbt.cacheForGVK(gvk).GetInformerForKind(ctx, gvk, opts...)
 }
@@ -116,7 +137,7 @@ func (dbt *delegatingByGVKCache) IndexField(ctx context.Context, obj client.Obje
 	return cache.IndexField(ctx, obj, field, extractValue)
 }
 
-func (dbt *delegatingByGVKCache) cacheForObject(o runtime.Object) (Cache, error) {
+func (dbt *delegatingByGVKCache) cacheForObject(o runtime.Object) (internalCache, error) {
 	gvk, err := apiutil.GVKForObject(o, dbt.scheme)
 	if err != nil {
 		return nil, err
@@ -125,7 +146,7 @@ func (dbt *delegatingByGVKCache) cacheForObject(o runtime.Object) (Cache, error)
 	return dbt.cacheForGVK(gvk), nil
 }
 
-func (dbt *delegatingByGVKCache) cacheForGVK(gvk schema.GroupVersionKind) Cache {
+func (dbt *delegatingByGVKCache) cacheForGVK(gvk schema.GroupVersionKind) internalCache {
 	if specific, hasSpecific := dbt.caches[gvk]; hasSpecific {
 		return specific
 	}

pkg/cache/informer_cache.go

@@ -76,7 +76,8 @@ func (ic *informerCache) Get(ctx context.Context, key client.ObjectKey, out clie
 	if !started {
 		return &ErrCacheNotStarted{}
 	}
-	return cache.Reader.Get(ctx, key, out, opts...)
+
+	return cache.Reader.Get(ctx, key, out, ic.Informers.MinimumRVs.GetForKey(gvk, key), opts...)
 }
 
 // List implements Reader.
@@ -95,7 +96,7 @@ func (ic *informerCache) List(ctx context.Context, out client.ObjectList, opts .
 		return &ErrCacheNotStarted{}
 	}
 
-	return cache.Reader.List(ctx, out, opts...)
+	return cache.Reader.List(ctx, out, ic.Informers.MinimumRVs.GetMaxForGVK(*gvk), opts...)
 }
 
 // objectTypeForListObject tries to find the runtime.Object and associated GVK
@@ -177,6 +178,50 @@ func (ic *informerCache) getInformerForKind(ctx context.Context, gvk schema.Grou
 	return started, cache, nil
 }
 
+func (ic *informerCache) SetMinimumRVForGVKAndKey(gvk schema.GroupVersionKind, key client.ObjectKey, rv int64) {
+	ic.MinimumRVs.Set(gvk, key, rv)
+}
+
+func (ic *informerCache) AddRequiredDeleteForObject(obj client.Object) error {
+	gvk, err := apiutil.GVKForObject(obj, ic.scheme)
+	if err != nil {
+		return err
+	}
+	cache, started, ok := ic.Peek(gvk, obj)
+	if !ok {
+		return fmt.Errorf("informer for GVK %v not found in cache", gvk)
+	}
+	if !started {
+		return &ErrCacheNotStarted{}
+	}
+	if !cache.Informer.HasSynced() {
+		return fmt.Errorf("informer for GVK %v is not synced", gvk)
+	}
+
+	cache.Reader.ConsistencyHandler.AddPendingDelete(
+		client.ObjectKey{Namespace: obj.GetNamespace(), Name: obj.GetName()},
+		obj.GetUID(),
+	)
+	return nil
+}
+
+func (ic *informerCache) RemoveRequiredDeleteForObject(obj client.Object) error {
+	gvk, err := apiutil.GVKForObject(obj, ic.scheme)
+	if err != nil {
+		return fmt.Errorf("failed to get GVK for object: %w", err)
+	}
+	cache, _, ok := ic.Peek(gvk, obj)
+	if !ok {
+		return fmt.Errorf("informer for GVK %v not found in cache", gvk)
+	}
+	cache.Reader.ConsistencyHandler.RemovePendingDelete(
+		client.ObjectKey{Namespace: obj.GetNamespace(), Name: obj.GetName()},
+		obj.GetUID(),
+	)
+
+	return nil
+}
+
 // RemoveInformer deactivates and removes the informer from the cache.
 func (ic *informerCache) RemoveInformer(_ context.Context, obj client.Object) error {
 	gvk, err := apiutil.GVKForObject(obj, ic.scheme)

pkg/cache/internal/cache_reader.go

@@ -30,13 +30,11 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
 
+	"sigs.k8s.io/controller-runtime/pkg/cache/internal/readerconsistency"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/internal/field/selector"
 )
 
-// CacheReader is a client.Reader.
-var _ client.Reader = &CacheReader{}
-
 // CacheReader wraps a cache.Index to implement the client.Reader interface for a single type.
 type CacheReader struct {
 	// indexer is the underlying indexer wrapped by this cache.
@@ -52,10 +50,12 @@ type CacheReader struct {
 	// Be very careful with this, when enabled you must DeepCopy any object before mutating it,
 	// otherwise you will mutate the object in the cache.
 	disableDeepCopy bool
+
+	*readerconsistency.ConsistencyHandler
 }
 
 // Get checks the indexer for the object and writes a copy of it if found.
-func (c *CacheReader) Get(_ context.Context, key client.ObjectKey, out client.Object, opts ...client.GetOption) error {
+func (c *CacheReader) Get(ctx context.Context, key client.ObjectKey, out client.Object, minRV int64, opts ...client.GetOption) error {
 	getOpts := client.GetOptions{}
 	getOpts.ApplyOptions(opts)
 
@@ -64,6 +64,10 @@ func (c *CacheReader) Get(_ context.Context, key client.ObjectKey, out client.Ob
 	}
 	storeKey := objectKeyToStoreKey(key)
 
+	if err := c.ConsistencyHandler.WaitForGet(ctx, key, minRV); err != nil {
+		return err
+	}
+
 	// Lookup the object from the indexer cache
 	obj, exists, err := c.indexer.GetByKey(storeKey)
 	if err != nil {
@@ -109,7 +113,7 @@ func (c *CacheReader) Get(_ context.Context, key client.ObjectKey, out client.Ob
 }
 
 // List lists items out of the indexer and writes them to out.
-func (c *CacheReader) List(_ context.Context, out client.ObjectList, opts ...client.ListOption) error {
+func (c *CacheReader) List(ctx context.Context, out client.ObjectList, minRV int64, opts ...client.ListOption) error {
 	var objs []any
 	var err error
 
@@ -120,6 +124,10 @@ func (c *CacheReader) List(_ context.Context, out client.ObjectList, opts ...cli
 		return fmt.Errorf("continue list option is not supported by the cache")
 	}
 
+	if err := c.ConsistencyHandler.WaitForList(ctx, minRV); err != nil {
+		return err
+	}
+
 	switch {
 	case listOpts.FieldSelector != nil:
 		requiresExact := selector.RequiresExactMatch(listOpts.FieldSelector)