Merging to release-5.8.9: [TT-15954]: Make org session fetch non-blocking (#7531)

[probelabs]

User description

TT-15954: Make org session fetch non-blocking (#7531)

Description

Fixes request pipeline blocking when MDCB is unavailable by making org
session fetches non-blocking in RPC mode.

Related Issue

TT-15954

Motivation and Context

When MDCB is unavailable, synchronous RPC calls to fetch org sessionsin
OrganizationMonitor were blocking the request pipeline for 90-120
seconds

How This Has Been Tested

Screenshots (if appropriate)

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing
  functionality to change)
• Refactoring or add test (improvements in base code or adds test
  coverage to functionality)

Checklist

• I ensured that the documentation is up to date
• I explained why this PR updates go.mod in detail with reasoning
  why it's required
• I would like a code coverage CI quality gate exception and have
  explained why

Ticket Details

TT-15954

Status	In Code Review
Summary	Request pipeline blocked by synchronous RPC calls every 10
minutes when MDCB unavailable

Generated at: 2025-11-20 13:06:06

---

Co-authored-by: andrei-tyk 97896463+andrei-tyk@users.noreply.github.com

---

PR Type

Bug fix

---

Description

• Make org session fetch non-blocking in RPC

• Add singleflight to dedupe session fetches

• Async refresh for org expiry cache misses

• Add tests for async, non-blocking behavior

---

Diagram Walkthrough

flowchart LR
  A["ProcessRequest (OrganizationMonitor)"] -- "RPC mode & no cache" --> B["refreshOrgSession (async)"]
  A -- "Non-RPC & no cache" --> C["OrgSession (sync)"]
  D["OrgSessionExpiry (BaseMiddleware)"] -- "cache miss" --> E["refreshOrgSessionExpiry (async)"]
  F["singleflight.Group"] -- "dedupe fetches" --> B

Loading

File Walkthrough

	Relevant files
Bug fix	mw_organisation_activity.go Async org session fetch with singleflight                                gateway/mw_organisation_activity.go Introduce singleflight for org session fetches. In RPC mode, fetch org session asynchronously. Add background cache population via refresh function. Minor comment fix and flow adjustments. +29/-2    middleware.go Async org expiry refresh and non-blocking path                      gateway/middleware.go Make OrgSessionExpiry return default on miss. Trigger async refresh on cache miss. Avoid blocking and handle emergency mode. Cache default when session not found. +21/-15
Tests	mw_organisation_activity_test.go Tests for async org session and RPC behavior                          gateway/mw_organisation_activity_test.go Add tests for refreshOrgSession behavior. Add RPC-mode async non-blocking request tests. Implement mock RPC server simulating delays. Verify cache population and no-session flagging. +227/-0  middleware_test.go Tests for async org expiry refresh                                              gateway/middleware_test.go Add tests for async expiry refresh flow. Validate cached value, default on miss. Ensure default retained for missing org. +35/-12
Miscellaneous	coverage.out Add coverage output artifact                                                          gateway/coverage.out Add coverage report artifact to repo. +9299/-0

---

[github-actions]

🚨 Jira Linter Failed

Commit: 68e9b75
Failed at: 2025-12-02 10:54:36 UTC

The Jira linter failed to validate your PR. Please check the error details below:

🔍 Click to view error details

failed to validate Jira issue: jira ticket TT-15954 has status 'In Test' but must be one of: In Dev, In Code Review, Ready For Dev, Dod Check

Next Steps

• Ensure your branch name contains a valid Jira ticket ID (e.g., ABC-123)
• Verify your PR title matches the branch's Jira ticket ID
• Check that the Jira ticket exists and is accessible

---

This comment will be automatically deleted once the linter passes.

[github-actions]

API Changes

no api changes detected

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🎫 Ticket compliance analysis ✅ 7531 - PR Code Verified Compliant requirements: Make org session fetch non-blocking in RPC mode to avoid blocking request pipeline when MDCB is unavailable Ensure requests proceed quickly even if org session is not in cache or MDCB is slow/unavailable Maintain correct behavior for non-RPC mode (no regression) Add tests validating non-blocking behavior and correct cache/flags updates Requires further human verification: Validate behavior under real MDCB outages and network failures in staging (timeouts, retries, logs) Confirm no unintended increase in RPC/backend load due to background fetches in high-concurrency scenarios

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Concurrency The use of singleflight.Group for org session fetches is appropriate, but the returned value from Do is ignored and errors are swallowed. Confirm no race on Spec.OrgHasNoSession and that background updates don't thrash the cache when DisableCacheSessionState is true. // try to get from Redis if !found { // not found in in-app cache, let's read from Redis // RPC mode: start background fetch and allow request to proceed if k.Spec.GlobalConfig.SlaveOptions.UseRPC { go k.refreshOrgSession(k.Spec.OrgID) return nil, http.StatusOK } // Non-RPC mode: synchronous fetch orgSession, found = k.OrgSession(k.Spec.OrgID) if !found { // prevent reads from in-app cache and from Redis for next runs k.setOrgHasNoSession(true) return nil, http.StatusOK } } clone := orgSession.Clone() if k.Spec.GlobalConfig.ExperimentalProcessOrgOffThread { // Make a copy of request before sending to goroutine r2 := r.WithContext(r.Context()) return k.ProcessRequestOffThread(r2, &clone) } return k.ProcessRequestLive(r, &clone) } func (k *OrganizationMonitor) refreshOrgSession(orgID string) { orgSessionFetchGroup.Do(orgID, func() (interface{}, error) { session, found := k.OrgSession(orgID) if found && !k.Spec.GlobalConfig.LocalSessionCache.DisableCacheSessionState { sessionLifeTime := session.Lifetime(k.Spec.GetSessionLifetimeRespectsKeyExpiration(), k.Spec.SessionLifetime, k.Gw.GetConfig().ForceGlobalSessionLifetime, k.Gw.GetConfig().GlobalSessionLifetime) k.Gw.SessionCache.Set(orgID, session.Clone(), sessionLifeTime) k.Logger().Debug("Background org session fetch completed for: ", orgID) return session, nil } if !found { k.setOrgHasNoSession(true) } return nil, nil }) } Behavior Change OrgSessionExpiry now always returns DEFAULT_ORG_SESSION_EXPIRATION on cache miss and fetches asynchronously. Validate downstream logic tolerates default expiry and that EnforceOrgDataAge semantics remain intact. func (t *BaseMiddleware) OrgSessionExpiry(orgid string) int64 { t.Logger().Debug("Checking: ", orgid) if rpc.IsEmergencyMode() { return DEFAULT_ORG_SESSION_EXPIRATION } // Try to get from cache first cachedVal, found := t.Gw.ExpiryCache.Get(orgid) if found { return cachedVal.(int64) } // Start async refresh in background go t.refreshOrgSessionExpiry(orgid) return DEFAULT_ORG_SESSION_EXPIRATION } func (t *BaseMiddleware) refreshOrgSessionExpiry(orgid string) { orgSessionExpiryCache.Do(orgid, func() (interface{}, error) { s, found := t.OrgSession(orgid) if found && t.Spec.GlobalConfig.EnforceOrgDataAge { t.SetOrgExpiry(orgid, s.DataExpires) return s.DataExpires, nil } // On failure or if not found, cache the default value t.SetOrgExpiry(orgid, DEFAULT_ORG_SESSION_EXPIRATION) return DEFAULT_ORG_SESSION_EXPIRATION, nil }) } Test Flakiness Tests rely on time.Sleep with small durations (50-100ms). On slow CI these may flake. Consider synchronization via channels or waiting on cache conditions instead of fixed sleeps. func TestOrganizationMonitor_RefreshOrgSession(t *testing.T) { conf := func(globalConf *config.Config) { globalConf.EnforceOrgQuotas = true globalConf.LocalSessionCache.DisableCacheSessionState = false } ts := StartTest(conf) defer ts.Close() orgID := "test-org-refresh-" + uuid.New() // Build API ts.Gw.BuildAndLoadAPI(func(spec *APISpec) { spec.UseKeylessAccess = true spec.OrgID = orgID spec.Proxy.ListenPath = "/" }) t.Run("refreshOrgSession populates cache when session found", func(t *testing.T) { // Create org session ts.Run(t, test.TestCase{ Path: "/tyk/org/keys/" + orgID, AdminAuth: true, Method: http.MethodPost, Code: http.StatusOK, Data: map[string]interface{}{ "quota_max": 10, "quota_remaining": 10, "quota_renewal_rate": 60, }, }) ts.Gw.SessionCache.Flush() // Verify cache is empty _, found := ts.Gw.SessionCache.Get(orgID) if found { t.Error("Cache should be empty") } spec := ts.Gw.apisByID[ts.Gw.apiSpecs[0].APIID] monitor := &OrganizationMonitor{ BaseMiddleware: &BaseMiddleware{ Spec: spec, Gw: ts.Gw, }, } // Call refreshOrgSession monitor.refreshOrgSession(orgID) // Wait a bit for async operation time.Sleep(50 * time.Millisecond) // Verify cache is now populated _, found = ts.Gw.SessionCache.Get(orgID) if !found { t.Error("Cache should be populated after refreshOrgSession") } }) t.Run("refreshOrgSession sets OrgHasNoSession when session not found", func(t *testing.T) { nonExistentOrgID := "test-org-nonexistent-" + uuid.New() // Build API with non-existent org spec := ts.Gw.BuildAndLoadAPI(func(spec *APISpec) { spec.UseKeylessAccess = true spec.OrgID = nonExistentOrgID spec.Proxy.ListenPath = "/nonexistent/" })[0] // Verify OrgHasNoSession is initially false if spec.OrgHasNoSession { t.Error("OrgHasNoSession should initially be false") } monitor := &OrganizationMonitor{ BaseMiddleware: &BaseMiddleware{ Spec: spec, Gw: ts.Gw, }, } monitor.refreshOrgSession(nonExistentOrgID) // Wait for async operation time.Sleep(50 * time.Millisecond) // Verify OrgHasNoSession is now true if !monitor.getOrgHasNoSession() { t.Error("OrgHasNoSession should be true after refreshOrgSession for non-existent org") } }) } // setupMockRPCServer creates a mock RPC server for testing async RPC behavior. // It simulates a slow MDCB backend with configurable delay. func setupMockRPCServer(orgID string, rpcDelay time.Duration) (*gorpc.Server, string) { dispatcher := gorpc.NewDispatcher() // Simulate slow GetKey response to test async behavior dispatcher.AddFunc("GetKey", func(_clientAddr, _key string) (string, error) { time.Sleep(rpcDelay) return `{"rate": 1000, "per": 1, "quota_max": -1}`, nil }) dispatcher.AddFunc("Login", func(_clientAddr, _userKey string) bool { return true }) dispatcher.AddFunc("GetApiDefinitions", func(_clientAddr string, _dr interface{}) (string, error) { return jsonMarshalString(BuildAPI(func(spec *APISpec) { spec.UseKeylessAccess = true spec.OrgID = orgID spec.Proxy.ListenPath = "/" })), nil }) dispatcher.AddFunc("GetPolicies", func(_clientAddr, _orgId string) (string, error) { return "[]", nil }) return startRPCMock(dispatcher) } func TestOrganizationMonitor_AsyncRPCMode(t *testing.T) { test.Flaky(t) maxRequestTime := 100 * time.Millisecond maxConcurrentTime := 200 * time.Millisecond orgID := "test-org-async-rpc-" + uuid.New() // Create a mock RPC server that simulates slow MDCB rpcMock, connectionString := setupMockRPCServer(orgID, 500*time.Millisecond) defer stopRPCMock(rpcMock) // Configure gateway with RPC mode enabled conf := func(globalConf *config.Config) { globalConf.EnforceOrgQuotas = true globalConf.LocalSessionCache.DisableCacheSessionState = true globalConf.SlaveOptions.UseRPC = true globalConf.SlaveOptions.RPCKey = "test_org" globalConf.SlaveOptions.APIKey = "test" globalConf.SlaveOptions.ConnectionString = connectionString globalConf.SlaveOptions.CallTimeout = 1 globalConf.SlaveOptions.RPCPoolSize = 2 globalConf.Policies.PolicySource = "rpc" } ts := StartTest(conf) defer ts.Close() // Wait for RPC connection and API load time.Sleep(100 * time.Millisecond) t.Run("Request does not block in RPC mode when org session not in cache", func(t *testing.T) { ts.Gw.SessionCache.Flush() _, found := ts.Gw.SessionCache.Get(orgID) if found { t.Error("Cache should be empty before test") } // Make a request, this should not block even though org session is not in cache start := time.Now() resp, _ := ts.Run(t, test.TestCase{ Path: "/", }) elapsed := time.Since(start) if resp != nil { resp.Body.Close() } // Verify request completed quickly if elapsed > maxRequestTime { t.Errorf("Request took too long (%v), suggesting it blocked waiting for RPC. Expected < %v", elapsed, maxRequestTime) } t.Logf("Request completed in %v (expected < %v), async behavior confirmed", elapsed, maxRequestTime) }) t.Run("Multiple concurrent requests do not block in RPC mode", func(t *testing.T) { ts.Gw.SessionCache.Flush() // Make multiple concurrent requests numRequests := 10 results := make(chan time.Duration, numRequests) start := time.Now() for i := 0; i < numRequests; i++ { go func() { reqStart := time.Now() resp, _ := ts.Run(t, test.TestCase{ Path: "/", }) if resp != nil { resp.Body.Close() } results <- time.Since(reqStart) }() } // Wait for all requests to complete for i := 0; i < numRequests; i++ { reqDuration := <-results if reqDuration > maxRequestTime { t.Errorf("Request %d took too long (%v)", i, reqDuration) } } totalElapsed := time.Since(start) // All requests should complete quickly if totalElapsed > maxConcurrentTime { t.Errorf("Requests took too long (%v), suggesting blocking behavior", totalElapsed) } t.Logf("All %d requests completed in %v, no blocking detected", numRequests, totalElapsed) }) }

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Avoid default overwriting fresh value The Do result is ignored, so repeated refresh calls may race without visibility on errors and can overwrite fresher values with defaults. Add a guard to only set default when EnforceOrgDataAge is enabled or session truly missing, and avoid overwriting a newer cached value with default if another goroutine already updated it. This prevents stale default clobbering. gateway/middleware.go [390-401] func (t *BaseMiddleware) refreshOrgSessionExpiry(orgid string) { orgSessionExpiryCache.Do(orgid, func() (interface{}, error) { s, found := t.OrgSession(orgid) if found && t.Spec.GlobalConfig.EnforceOrgDataAge { t.SetOrgExpiry(orgid, s.DataExpires) return s.DataExpires, nil } - // On failure or if not found, cache the default value - t.SetOrgExpiry(orgid, DEFAULT_ORG_SESSION_EXPIRATION) + // Only set default if nothing newer is present + if _, ok := t.Gw.ExpiryCache.Get(orgid); !ok { + t.SetOrgExpiry(orgid, DEFAULT_ORG_SESSION_EXPIRATION) + } return DEFAULT_ORG_SESSION_EXPIRATION, nil }) } Suggestion importance[1-10]: 7 __ Why: Matches the new hunk and proposes a safe guard to avoid clobbering fresher cache entries with the default during concurrent refreshes. It’s a precise and low-risk improvement to concurrency behavior, though not critical.	Medium
	Remove committed coverage artifact Avoid committing coverage artifacts like coverage.out to the repository; they are build/test outputs and can bloat the repo and create noisy diffs. Add this path to .gitignore (or CI artifacts) and remove the file from version control. gateway/coverage.out [1-3] -mode: set -github.com/TykTechnologies/tyk/gateway/analytics.go:28.83,33.78 4 1 -... +(Remove the `gateway/coverage.out` file from the repository and add an ignore rule like: +`gateway/coverage.out` +or a pattern such as: +`**/coverage.out` +in the project's ignore configuration.) Suggestion importance[1-10]: 7 __ Why: Committing coverage.out is undesirable as it is a generated artifact that clutters diffs and bloats the repo; removing it and adding it to ignore rules is correct and beneficial. The 'existing_code' snippet matches the 'new hunk' beginning lines (e.g., mode: set), and the recommendation accurately reflects the appropriate action.	Medium
	Prevent repeated background fetches When returning early in RPC mode, OrgHasNoSession may remain false even if no session exists, causing repeated background fetches per request. Set the no-session marker before returning if the fetch reveals absence, or pessimistically mark before fetch and clear on success. This reduces unnecessary work and avoids potential thundering-herd on cache misses. gateway/mw_organisation_activity.go [78-95] if !found { // not found in in-app cache, let's read from Redis - // RPC mode: start background fetch and allow request to proceed if k.Spec.GlobalConfig.SlaveOptions.UseRPC { + // Optimistically mark no-session; refresh will clear by caching on success + k.setOrgHasNoSession(true) go k.refreshOrgSession(k.Spec.OrgID) - return nil, http.StatusOK } // Non-RPC mode: synchronous fetch orgSession, found = k.OrgSession(k.Spec.OrgID) if !found { - // prevent reads from in-app cache and from Redis for next runs k.setOrgHasNoSession(true) return nil, http.StatusOK } } Suggestion importance[1-10]: 5 __ Why: The concern about repeated background fetches in RPC mode is valid; marking OrgHasNoSession can reduce churn. However, pessimistically setting it may incorrectly gate behavior if a session exists and the fetch succeeds; the suggestion lacks the corresponding “clear on success” logic, reducing accuracy.	Low
Possible issue	Scope singleflight to instance Avoid using a global singleflight.Group for all gateways; it can cause unintended cross-Gateway coupling in tests or multi-gateway processes. Scope the singleflight.Group to the OrganizationMonitor (or Gateway) instance to prevent contention and stale deduplication across unrelated org fetches. This reduces risk of missed refreshes in concurrent tests. gateway/mw_organisation_activity.go [105-120] +type OrganizationMonitor struct { + *BaseMiddleware + fetchGroup singleflight.Group +} + func (k *OrganizationMonitor) refreshOrgSession(orgID string) { - orgSessionFetchGroup.Do(orgID, func() (interface{}, error) { + k.fetchGroup.Do(orgID, func() (interface{}, error) { session, found := k.OrgSession(orgID) if found && !k.Spec.GlobalConfig.LocalSessionCache.DisableCacheSessionState { sessionLifeTime := session.Lifetime(k.Spec.GetSessionLifetimeRespectsKeyExpiration(), k.Spec.SessionLifetime, k.Gw.GetConfig().ForceGlobalSessionLifetime, k.Gw.GetConfig().GlobalSessionLifetime) - k.Gw.SessionCache.Set(orgID, session.Clone(), sessionLifeTime) k.Logger().Debug("Background org session fetch completed for: ", orgID) return session, nil } if !found { k.setOrgHasNoSession(true) } return nil, nil }) } Suggestion importance[1-10]: 6 __ Why: Correctly identifies that using a package-level singleflight.Group (orgSessionFetchGroup) can cause cross-instance contention and proposes a reasonable refactor. Impact is moderate and correct, but it requires structural changes (adding field to OrganizationMonitor) beyond the shown diff.	Low

[sonarqubecloud]

Quality Gate failed

Failed conditions
1 Security Hotspot

See analysis details on SonarQube Cloud