Prevent response cache mutex from serializing concurrent HTTP action requests (#21880)

* Add singleflight and short lock for concurrent

* Implement feedback

* Use atomic integer for fetch

* Add test for singleflight leader behavior and refine cache fetch logic

* Adjust response cache test to use seconds for delays and refactor assertions

Author: george-dorin

core/services/gateway/handlers/capabilities/v2/http_handler.go

@@ -59,12 +59,12 @@ type gatewayHandler struct {
 
 type ResponseCache interface {
 	// Set caches a response if it is cacheable (2xx or 4xx status codes) and the cache is empty or expired for the given request.
-	Set(workflowID string, req gateway_common.OutboundHTTPRequest, response gateway_common.OutboundHTTPResponse)
+	Set(req gateway_common.OutboundHTTPRequest, response gateway_common.OutboundHTTPResponse)
 
 	// Fetch retrieves a response from the cache if it exists and the age of cached response is less than the max age of the request.
 	// If the cached response is expired or not cached, it fetches a new response from the fetchFn.
 	// The response is cached if it is cacheable and storeOnFetch is true.
-	Fetch(ctx context.Context, workflowID string, req gateway_common.OutboundHTTPRequest, fetchFn func() gateway_common.OutboundHTTPResponse, storeOnFetch bool) gateway_common.OutboundHTTPResponse
+	Fetch(ctx context.Context, req gateway_common.OutboundHTTPRequest, fetchFn func() gateway_common.OutboundHTTPResponse, storeOnFetch bool) gateway_common.OutboundHTTPResponse
 
 	// DeleteExpired removes all cached responses that have exceeded their TTL (Time To Live).
 	DeleteExpired(ctx context.Context) int
@@ -306,16 +306,6 @@ func (h *gatewayHandler) createHTTPRequestCallback(ctx context.Context, requestI
 	}
 }
 
-// extractWorkflowIDFromRequestPath extracts the workflowID from an outgoing request path string.
-// The workflowID is expected to be the first element after splitting the string by "/".
-func extractWorkflowIDFromRequestPath(path string) string {
-	parts := strings.Split(path, "/")
-	if len(parts) > 1 {
-		return parts[1]
-	}
-	return ""
-}
-
 func (h *gatewayHandler) HandleLegacyUserMessage(context.Context, *api.Message, handlers.Callback) error {
 	return errors.New("HTTP capability gateway handler does not support legacy messages")
 }
@@ -340,7 +330,6 @@ func (h *gatewayHandler) makeOutgoingRequest(ctx context.Context, resp *jsonrpc.
 	if err != nil {
 		return fmt.Errorf("failed to unmarshal HTTP request from node %s: %w", nodeAddr, err)
 	}
-	workflowID := extractWorkflowIDFromRequestPath(requestID)
 	timeout := time.Duration(req.TimeoutMs) * time.Millisecond
 	httpReq := network.HTTPRequest{
 		Method:           req.Method,
@@ -367,11 +356,11 @@ func (h *gatewayHandler) makeOutgoingRequest(ctx context.Context, resp *jsonrpc.
 		callback := h.createHTTPRequestCallback(httpCtx, requestID, httpReq, req)
 		if req.CacheSettings.MaxAgeMs > 0 {
 			h.metrics.IncrementCacheReadCount(ctx, h.lggr)
-			outboundResp = h.responseCache.Fetch(httpCtx, workflowID, req, callback, req.CacheSettings.Store)
+			outboundResp = h.responseCache.Fetch(httpCtx, req, callback, req.CacheSettings.Store)
 		} else {
 			outboundResp = callback()
 			if req.CacheSettings.Store {
-				h.responseCache.Set(workflowID, req, outboundResp)
+				h.responseCache.Set(req, outboundResp)
 			}
 		}
 		h.metrics.IncrementActionCapabilityRequestCount(ctx, nodeAddr, h.lggr)

core/services/gateway/handlers/capabilities/v2/http_handler_test.go

@@ -402,11 +402,11 @@ func newMockResponseCache() *mockResponseCache {
 	}
 }
 
-func (m *mockResponseCache) Set(workflowID string, req gateway_common.OutboundHTTPRequest, response gateway_common.OutboundHTTPResponse) {
+func (m *mockResponseCache) Set(req gateway_common.OutboundHTTPRequest, response gateway_common.OutboundHTTPResponse) {
 	m.setCallCount++
 }
 
-func (m *mockResponseCache) Fetch(ctx context.Context, workflowID string, req gateway_common.OutboundHTTPRequest, fetchFn func() gateway_common.OutboundHTTPResponse, storeOnFetch bool) gateway_common.OutboundHTTPResponse {
+func (m *mockResponseCache) Fetch(ctx context.Context, req gateway_common.OutboundHTTPRequest, fetchFn func() gateway_common.OutboundHTTPResponse, storeOnFetch bool) gateway_common.OutboundHTTPResponse {
 	m.fetchCallCount++
 	return fetchFn()
 }

core/services/gateway/handlers/capabilities/v2/response_cache.go

@@ -5,17 +5,19 @@ import (
 	"sync"
 	"time"
 
+	"golang.org/x/sync/singleflight"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/gateway"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities/v2/metrics"
 )
 
 // responseCache is a thread-safe cache for storing HTTP responses.
-// It uses a map to store responses keyed by a unique identifier generated from the request
-// cache key is prefixed by workflowID to avoid collisions between different workflows.
+// It uses a map to store responses keyed by a hash of the request (method, URL, headers, body, workflowOwner).
 type responseCache struct {
-	cacheMu sync.Mutex
+	cacheMu sync.RWMutex
 	cache   map[string]*cachedResponse
+	flight  singleflight.Group
 	lggr    logger.Logger
 	ttl     time.Duration
 	metrics *metrics.Metrics
@@ -42,8 +44,8 @@ func isCacheableStatusCode(statusCode int) bool {
 }
 
 // isExpiredOrNotCached returns true if the cached response is expired or not cached.
-// IMPORTANT: this method does not lock the cache map. MUST be called with the cacheMu locked.
-func (rc *responseCache) isExpiredOrNotCached(_ string, req gateway.OutboundHTTPRequest) bool {
+// IMPORTANT: this method does not lock the cache map. MUST be called with cacheMu write-locked.
+func (rc *responseCache) isExpiredOrNotCached(req gateway.OutboundHTTPRequest) bool {
 	cachedResp, exists := rc.cache[req.Hash()]
 	if !exists || time.Now().After(cachedResp.storedAt.Add(rc.ttl)) {
 		return true
@@ -53,32 +55,63 @@ func (rc *responseCache) isExpiredOrNotCached(_ string, req gateway.OutboundHTTP
 
 // Fetch fetches a response from the cache if it exists and
 // the age of cached response is less than the max age of the request.
-// If the cached response is expired or not cached, it fetches a new response from the fetchFn.
+// If the cached response is expired or not cached, it fetches a new response from the fetchFn
 // and caches the response if it is cacheable and storeOnFetch is true.
-func (rc *responseCache) Fetch(ctx context.Context, workflowID string, req gateway.OutboundHTTPRequest, fetchFn func() gateway.OutboundHTTPResponse, storeOnFetch bool) gateway.OutboundHTTPResponse {
-	rc.cacheMu.Lock()
-	defer rc.cacheMu.Unlock()
+//
+// The mutex is only held during cache map access (microseconds), not during fetchFn execution.
+// Singleflight deduplicates concurrent requests to the same cache key so only one fetchFn
+// runs per key, while requests to different keys execute in parallel.
+// Cache read and write happen inside the singleflight callback to ensure the key remains
+// in-flight until the result is stored, preventing duplicate fetches.
+func (rc *responseCache) Fetch(ctx context.Context, req gateway.OutboundHTTPRequest, fetchFn func() gateway.OutboundHTTPResponse, storeOnFetch bool) gateway.OutboundHTTPResponse {
+	cacheKey := req.Hash()
 	cacheMaxAge := time.Duration(req.CacheSettings.MaxAgeMs) * time.Millisecond
-	cachedResp, exists := rc.cache[req.Hash()]
+
+	// Fast path: check cache without singleflight overhead.
+	rc.cacheMu.RLock()
+	cachedResp, exists := rc.cache[cacheKey]
+	rc.cacheMu.RUnlock()
 	if exists && cachedResp.storedAt.Add(cacheMaxAge).After(time.Now()) {
 		rc.metrics.IncrementCacheHitCount(ctx, rc.lggr)
 		return cachedResp.response
 	}
-	response := fetchFn()
-	if storeOnFetch && isCacheableStatusCode(response.StatusCode) && rc.isExpiredOrNotCached(workflowID, req) {
-		rc.cache[req.Hash()] = &cachedResponse{
-			response: response,
-			storedAt: time.Now(),
+
+	// Slow path: singleflight deduplicates concurrent fetches per key.
+	// Cache check + store happen inside the flight so the key isn't released
+	// until the result is cached, closing the race window between singleflight
+	// completion and cache write.
+	result, _, _ := rc.flight.Do(cacheKey, func() (interface{}, error) {
+		// Re-check cache: a previous flight may have just stored the result.
+		rc.cacheMu.RLock()
+		cachedResp, exists := rc.cache[cacheKey]
+		rc.cacheMu.RUnlock()
+		if exists && cachedResp.storedAt.Add(cacheMaxAge).After(time.Now()) {
+			rc.metrics.IncrementCacheHitCount(ctx, rc.lggr)
+			return cachedResp.response, nil
 		}
-	}
-	return response
+
+		response := fetchFn()
+
+		if storeOnFetch && isCacheableStatusCode(response.StatusCode) {
+			rc.cacheMu.Lock()
+			rc.cache[cacheKey] = &cachedResponse{
+				response: response,
+				storedAt: time.Now(),
+			}
+			rc.cacheMu.Unlock()
+		}
+
+		return response, nil
+	})
+
+	return result.(gateway.OutboundHTTPResponse)
 }
 
 // Set caches a response if it is cacheable (2xx or 4xx and cache is empty or expired for the given request)
-func (rc *responseCache) Set(workflowID string, req gateway.OutboundHTTPRequest, response gateway.OutboundHTTPResponse) {
+func (rc *responseCache) Set(req gateway.OutboundHTTPRequest, response gateway.OutboundHTTPResponse) {
 	rc.cacheMu.Lock()
 	defer rc.cacheMu.Unlock()
-	if isCacheableStatusCode(response.StatusCode) && rc.isExpiredOrNotCached(workflowID, req) {
+	if isCacheableStatusCode(response.StatusCode) && rc.isExpiredOrNotCached(req) {
 		rc.cache[req.Hash()] = &cachedResponse{
 			response: response,
 			storedAt: time.Now(),