fix: resolve cache race with singleflight

Before this change, cache Get and Add were separate operations,
creating a TOCTOU race: concurrent requests for the same resource
could all miss the cache and each trigger a remote resolution.

Merge Get/Add/GetFromCacheOrResolve into a single
GetCachedOrResolveFromRemote method on resolverCache that wraps
the resolve callback with golang.org/x/sync/singleflight. Only
one in-flight resolution per cache key proceeds; all concurrent
callers share its result.

Other changes in this commit:
- Use strings.Builder in generateCacheKey instead of
  string concatenation
- Remove unused resolverType param from ShouldUse
- Rework e2e tests to verify caching by counting actual
  registry GET requests (via logs and metrics) instead of
  checking resolver log messages
- Add multi-replica resolver test (4 replicas, 200 TaskRuns)
- Allow -run flag to bypass category filtering in TestMain

Issue #9364

Signed-off-by: Stanislav Jakuschevskij <stas@two-giants.com>

Author: twoGiants

pkg/remoteresolution/resolver/bundle/resolver.go

@@ -109,9 +109,8 @@ func (r *Resolver) Resolve(ctx context.Context, req *v1beta1.ResolutionRequestSp
 		return nil, errors.New("no params")
 	}
 
-	if cache.ShouldUse(ctx, r, req.Params, LabelValueBundleResolverType) {
-		return cache.GetFromCacheOrResolve(
-			ctx,
+	if cache.ShouldUse(ctx, r, req.Params) {
+		return cache.Get(ctx).GetCachedOrResolveFromRemote(
 			req.Params,
 			LabelValueBundleResolverType,
 			func() (resolutionframework.ResolvedResource, error) {

pkg/remoteresolution/resolver/cluster/resolver.go

@@ -88,9 +88,8 @@ func (r *Resolver) IsImmutable([]v1.Param) bool {
 
 // Resolve uses the given params to resolve the requested file or resource.
 func (r *Resolver) Resolve(ctx context.Context, req *v1beta1.ResolutionRequestSpec) (resolutionframework.ResolvedResource, error) {
-	if cache.ShouldUse(ctx, r, req.Params, LabelValueClusterResolverType) {
-		return cache.GetFromCacheOrResolve(
-			ctx,
+	if cache.ShouldUse(ctx, r, req.Params) {
+		return cache.Get(ctx).GetCachedOrResolveFromRemote(
 			req.Params,
 			LabelValueClusterResolverType,
 			func() (resolutionframework.ResolvedResource, error) {

pkg/remoteresolution/resolver/cluster/resolver_test.go

@@ -632,8 +632,11 @@ func TestResolveWithCacheHit(t *testing.T) {
 		{Name: "cache", Value: *pipelinev1.NewStructuredValues("always")},
 	}
 
-	// add the mock resource to the cache
-	cache.Get(ctx).Add(cluster.LabelValueClusterResolverType, params, mockResource)
+	// prepopulate the cache using GetCachedOrResolveFromRemote
+	resolveFn := func() (framework.ResolvedResource, error) {
+		return mockResource, nil
+	}
+	cache.Get(ctx).GetCachedOrResolveFromRemote(params, cluster.LabelValueClusterResolverType, resolveFn)
 
 	// create request with same parameters
 	req := &v1beta1.ResolutionRequestSpec{Params: params}

pkg/remoteresolution/resolver/framework/cache/cache.go

@@ -20,25 +20,31 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"sort"
+	"strings"
 	"time"
 
 	pipelinev1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1"
 	resolutionframework "github.com/tektoncd/pipeline/pkg/resolution/resolver/framework"
 	"go.uber.org/zap"
+	"golang.org/x/sync/singleflight"
 	utilcache "k8s.io/apimachinery/pkg/util/cache"
 )
 
+type resolveFn = func() (resolutionframework.ResolvedResource, error)
+
 var _ resolutionframework.ConfigWatcher = (*resolverCache)(nil)
 
 // resolverCache is a wrapper around utilcache.LRUExpireCache that provides
 // type-safe methods for caching resolver results.
 type resolverCache struct {
-	cache   *utilcache.LRUExpireCache
-	logger  *zap.SugaredLogger
-	ttl     time.Duration
-	maxSize int
-	clock   utilcache.Clock
+	cache       *utilcache.LRUExpireCache
+	logger      *zap.SugaredLogger
+	ttl         time.Duration
+	maxSize     int
+	clock       utilcache.Clock
+	flightGroup *singleflight.Group
 }
 
 func newResolverCache(maxSize int, ttl time.Duration) *resolverCache {
@@ -47,10 +53,11 @@ func newResolverCache(maxSize int, ttl time.Duration) *resolverCache {
 
 func newResolverCacheWithClock(maxSize int, ttl time.Duration, clock utilcache.Clock) *resolverCache {
 	return &resolverCache{
-		cache:   utilcache.NewLRUExpireCacheWithClock(maxSize, clock),
-		ttl:     ttl,
-		maxSize: maxSize,
-		clock:   clock,
+		cache:       utilcache.NewLRUExpireCacheWithClock(maxSize, clock),
+		ttl:         ttl,
+		maxSize:     maxSize,
+		clock:       clock,
+		flightGroup: &singleflight.Group{},
 	}
 }
 
@@ -62,7 +69,7 @@ func (c *resolverCache) GetConfigName(_ context.Context) string {
 // withLogger returns a new ResolverCache instance with the provided logger.
 // This prevents state leak by not storing logger in the global singleton.
 func (c *resolverCache) withLogger(logger *zap.SugaredLogger) *resolverCache {
-	return &resolverCache{logger: logger, cache: c.cache, ttl: c.ttl, maxSize: c.maxSize, clock: c.clock}
+	return &resolverCache{logger: logger, cache: c.cache, ttl: c.ttl, maxSize: c.maxSize, clock: c.clock, flightGroup: c.flightGroup}
 }
 
 // TTL returns the time-to-live duration for cache entries.
@@ -75,57 +82,57 @@ func (c *resolverCache) MaxSize() int {
 	return c.maxSize
 }
 
-// Get retrieves a cached resource by resolver type and parameters, returning
-// the resource and whether it was found.
-func (c *resolverCache) Get(resolverType string, params []pipelinev1.Param) (resolutionframework.ResolvedResource, bool) {
+func (c *resolverCache) GetCachedOrResolveFromRemote(
+	params []pipelinev1.Param,
+	resolverType string,
+	resolveFromRemote resolveFn,
+) (resolutionframework.ResolvedResource, error) {
 	key := generateCacheKey(resolverType, params)
-	value, found := c.cache.Get(key)
-	if !found {
-		c.infow("Cache miss", "key", key)
-		return nil, found
-	}
 
-	resource, ok := value.(resolutionframework.ResolvedResource)
-	if !ok {
-		c.infow("Failed casting cached resource", "key", key)
-		return nil, false
-	}
+	if untyped, found := c.cache.Get(key); found {
+		cached, ok := untyped.(resolutionframework.ResolvedResource)
+		if !ok {
+			c.infow("Failed casting cached resource", "key", key)
+			return nil, errors.New("failed casting cached resource")
+		}
 
-	c.infow("Cache hit", "key", key)
-	timestamp := c.clock.Now().Format(time.RFC3339)
-	return newAnnotatedResource(resource, resolverType, cacheOperationRetrieve, timestamp), true
-}
+		c.infow("Cache hit", "key", key)
 
-func (c *resolverCache) infow(msg string, keysAndValues ...any) {
-	if c.logger != nil {
-		c.logger.Infow(msg, keysAndValues...)
+		return c.annotate(cached, resolverType, cacheOperationRetrieve), nil
 	}
-}
 
-// Add stores a resource in the cache with the configured TTL and returns an
-// annotated version of the resource.
-func (c *resolverCache) Add(
-	resolverType string,
-	params []pipelinev1.Param,
-	resource resolutionframework.ResolvedResource,
-) resolutionframework.ResolvedResource {
-	key := generateCacheKey(resolverType, params)
-	c.infow("Adding to cache", "key", key, "expiration", c.ttl)
+	// If cache miss, resolve from remote using singleflight
+	untyped, err, _ := c.flightGroup.Do(key, func() (any, error) {
+		resolved, err := resolveFromRemote()
+		if err != nil {
+			return nil, err
+		}
 
-	timestamp := c.clock.Now().Format(time.RFC3339)
-	annotatedResource := newAnnotatedResource(resource, resolverType, cacheOperationStore, timestamp)
+		annotated := c.annotate(resolved, resolverType, cacheOperationStore)
 
-	c.cache.Add(key, annotatedResource, c.ttl)
+		// Store annotated resource with store operation and return annotated resource
+		// to indicate it was stored in cache
+		c.infow("Adding to cache", "key", key, "expiration", c.ttl)
+		c.cache.Add(key, annotated, c.ttl)
+		return annotated, nil
+	})
+	if err != nil {
+		return nil, err
+	}
 
-	return annotatedResource
+	return untyped.(resolutionframework.ResolvedResource), nil
 }
 
-// Remove deletes a cached resource identified by resolver type and parameters.
-func (c *resolverCache) Remove(resolverType string, params []pipelinev1.Param) {
-	key := generateCacheKey(resolverType, params)
-	c.infow("Removing from cache", "key", key)
+func (c *resolverCache) annotate(resolvedResource resolutionframework.ResolvedResource, resolverType, operation string) *annotatedResource {
+	timestamp := c.clock.Now().Format(time.RFC3339)
+	result := newAnnotatedResource(resolvedResource, resolverType, operation, timestamp)
+	return result
+}
 
-	c.cache.Remove(key)
+func (c *resolverCache) infow(msg string, keysAndValues ...any) {
+	if c.logger != nil {
+		c.logger.Infow(msg, keysAndValues...)
+	}
 }
 
 // Clear removes all entries from the cache.
@@ -137,7 +144,9 @@ func (c *resolverCache) Clear() {
 
 func generateCacheKey(resolverType string, params []pipelinev1.Param) string {
 	// Create a deterministic string representation of the parameters
-	paramStr := resolverType + ":"
+	var sb strings.Builder
+	sb.WriteString(resolverType)
+	sb.WriteByte(':')
 
 	// Filter out the 'cache' parameter and sort remaining params by name for determinism
 	filteredParams := make([]pipelinev1.Param, 0, len(params))
@@ -153,21 +162,22 @@ func generateCacheKey(resolverType string, params []pipelinev1.Param) string {
 	})
 
 	for _, p := range filteredParams {
-		paramStr += p.Name + "="
+		sb.WriteString(p.Name)
+		sb.WriteByte('=')
 
 		switch p.Value.Type {
 		case pipelinev1.ParamTypeString:
-			paramStr += p.Value.StringVal
+			sb.WriteString(p.Value.StringVal)
 		case pipelinev1.ParamTypeArray:
 			// Sort array values for determinism
 			arrayVals := make([]string, len(p.Value.ArrayVal))
 			copy(arrayVals, p.Value.ArrayVal)
 			sort.Strings(arrayVals)
 			for i, val := range arrayVals {
 				if i > 0 {
-					paramStr += ","
+					sb.WriteByte(',')
 				}
-				paramStr += val
+				sb.WriteString(val)
 			}
 		case pipelinev1.ParamTypeObject:
 			// Sort object keys for determinism
@@ -178,18 +188,20 @@ func generateCacheKey(resolverType string, params []pipelinev1.Param) string {
 			sort.Strings(keys)
 			for i, key := range keys {
 				if i > 0 {
-					paramStr += ","
+					sb.WriteByte(',')
 				}
-				paramStr += key + ":" + p.Value.ObjectVal[key]
+				sb.WriteString(key)
+				sb.WriteByte(':')
+				sb.WriteString(p.Value.ObjectVal[key])
 			}
 		default:
 			// For unknown types, use StringVal as fallback
-			paramStr += p.Value.StringVal
+			sb.WriteString(p.Value.StringVal)
 		}
-		paramStr += ";"
+		sb.WriteByte(';')
 	}
 
 	// Generate a SHA-256 hash of the parameter string
-	hash := sha256.Sum256([]byte(paramStr))
+	hash := sha256.Sum256([]byte(sb.String()))
 	return hex.EncodeToString(hash[:])
 }