Implement singleflight

Author: imogenkraak

gateway/middleware.go

@@ -371,7 +371,6 @@ func (t *BaseMiddleware) SetOrgExpiry(orgid string, expiry int64) {
 func (t *BaseMiddleware) OrgSessionExpiry(orgid string) int64 {
 	t.Logger().Debug("Checking: ", orgid)
 
-	// Emergency Mode: Return default immediately
 	if rpc.IsEmergencyMode() {
 		return DEFAULT_ORG_SESSION_EXPIRATION
 	}
@@ -382,22 +381,23 @@ func (t *BaseMiddleware) OrgSessionExpiry(orgid string) int64 {
 		return cachedVal.(int64)
 	}
 
-	// Background Refresh: Start async refresh in background
+	// Start async refresh in background
 	go t.refreshOrgSessionExpiry(orgid)
 
-	// Return default immediately
 	return DEFAULT_ORG_SESSION_EXPIRATION
 }
 
 func (t *BaseMiddleware) refreshOrgSessionExpiry(orgid string) {
-	// This RPC call now happens safely in the background
-	s, found := t.OrgSession(orgid)
-	if found && t.Spec.GlobalConfig.EnforceOrgDataAge {
-		t.SetOrgExpiry(orgid, s.DataExpires)
-	} else {
+	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
+	})
 }
 
 func (t *BaseMiddleware) UpdateRequestSession(r *http.Request) bool {

gateway/mw_organisation_activity.go

@@ -6,6 +6,8 @@ import (
 	"sync"
 	"time"
 
+	"golang.org/x/sync/singleflight"
+
 	"github.com/TykTechnologies/tyk/ctx"
 	"github.com/TykTechnologies/tyk/request"
 	"github.com/TykTechnologies/tyk/user"
@@ -18,6 +20,7 @@ type orgChanMapMu struct {
 
 var orgChanMap = orgChanMapMu{channels: map[string](chan bool){}}
 var orgActiveMap sync.Map
+var orgSessionFetchGroup singleflight.Group
 
 // RateLimitAndQuotaCheck will check the incoming request and key whether it is within it's quota and
 // within it's rate limit, it makes use of the SessionLimiter object to do this
@@ -101,13 +104,18 @@ func (k *OrganizationMonitor) ProcessRequest(w http.ResponseWriter, r *http.Requ
 }
 
 func (k *OrganizationMonitor) refreshOrgSession(orgID string) {
-	session, found := k.OrgSession(orgID)
-	if found && !k.Spec.GlobalConfig.LocalSessionCache.DisableCacheSessionState {
-		k.Gw.SessionCache.Set(orgID, session.Clone(), int64(time.Hour))
-		k.Logger().Debug("Background org session fetch completed for: ", orgID)
-	} else if !found {
-		k.setOrgHasNoSession(true)
-	}
+	orgSessionFetchGroup.Do(orgID, func() (interface{}, error) {
+		session, found := k.OrgSession(orgID)
+		if found && !k.Spec.GlobalConfig.LocalSessionCache.DisableCacheSessionState {
+			k.Gw.SessionCache.Set(orgID, session.Clone(), int64(time.Hour))
+			k.Logger().Debug("Background org session fetch completed for: ", orgID)
+			return session, nil
+		}
+		if !found {
+			k.setOrgHasNoSession(true)
+		}
+		return nil, nil
+	})
 }
 
 // ProcessRequest will run any checks on the request on the way through the system, return an error to have the chain fail