fix(refresh): avoid deadlock by moving connector refresh outside txn and adding per-token mutex

Previously, `updateRefreshToken` executed `refreshWithConnector` inside the
`UpdateRefreshToken` transaction. With SQL backends that enforce strict
connection limits (e.g. SQLite), this blocked the only available connection
while the connector call could indirectly trigger further storage access
(e.g. when using PasswordDB), causing the system to hang.

This patch moves connector refresh calls outside of the storage transaction
and introduces a per-refresh-ID mutex to ensure only one concurrent request
per token hits the external IdP. Other concurrent requests wait for the mutex
and reuse the updated identity.

Signed-off-by: Tommaso Sardelli <[email protected]>

Author: tsardelli-dh

server/refreshhandlers.go

@@ -232,24 +232,50 @@ func (s *Server) updateOfflineSession(ctx context.Context, refresh *storage.Refr
 	return nil
 }
 
-// updateRefreshToken updates refresh token and offline session in the storage
+// updateRefreshToken updates refresh token and offline session in the storage.
+// Connector refresh is guarded by a per-refresh-ID mutex so only one concurrent
+// caller hits the IdP.
 func (s *Server) updateRefreshToken(ctx context.Context, rCtx *refreshContext) (*internal.RefreshToken, connector.Identity, *refreshError) {
-	var rerr *refreshError
-
 	newToken := &internal.RefreshToken{
 		Token:     rCtx.requestToken.Token,
 		RefreshId: rCtx.requestToken.RefreshId,
 	}
-
 	lastUsed := s.now()
 
+	// Base identity
 	ident := connector.Identity{
 		UserID:            rCtx.storageToken.Claims.UserID,
 		Username:          rCtx.storageToken.Claims.Username,
 		PreferredUsername: rCtx.storageToken.Claims.PreferredUsername,
 		Email:             rCtx.storageToken.Claims.Email,
 		EmailVerified:     rCtx.storageToken.Claims.EmailVerified,
 		Groups:            rCtx.storageToken.Claims.Groups,
+		ConnectorData:     rCtx.connectorData,
+	}
+
+	rotationEnabled := s.refreshTokenPolicy.RotationEnabled()
+	reusingAllowed := s.refreshTokenPolicy.AllowedToReuse(rCtx.storageToken.LastUsed)
+	needConnectorRefresh := rotationEnabled && !reusingAllowed
+
+	if needConnectorRefresh {
+		// Serialize concurrent refreshes for the same refresh ID.
+		lock := s.getRefreshLock(rCtx.storageToken.ID)
+		lock.Lock()
+		s.logger.Debug("Acquired refresh lock", "refreshID", rCtx.storageToken.ID)
+		defer func() {
+			lock.Unlock()
+			s.logger.Debug("Released refresh lock", "refreshID", rCtx.storageToken.ID)
+		}()
+
+		// Double-check if another goroutine already refreshed while we waited:
+		if !s.refreshTokenPolicy.AllowedToReuse(rCtx.storageToken.LastUsed) {
+			s.logger.Debug("Calling refreshWithConnector", "ident", ident)
+			var rerr *refreshError
+			ident, rerr = s.refreshWithConnector(ctx, rCtx, ident)
+			if rerr != nil {
+				return nil, ident, rerr
+			}
+		}
 	}
 
 	refreshTokenUpdater := func(old storage.RefreshToken) (storage.RefreshToken, error) {
@@ -258,21 +284,16 @@ func (s *Server) updateRefreshToken(ctx context.Context, rCtx *refreshContext) (
 
 		switch {
 		case !rotationEnabled && reusingAllowed:
-			// If rotation is disabled and the offline session was updated not so long ago - skip further actions.
 			old.ConnectorData = nil
 			return old, nil
 
 		case rotationEnabled && reusingAllowed:
 			if old.Token != rCtx.requestToken.Token && old.ObsoleteToken != rCtx.requestToken.Token {
 				return old, errors.New("refresh token claimed twice")
 			}
-
-			// Return previously generated token for all requests with an obsolete tokens
 			if old.ObsoleteToken == rCtx.requestToken.Token {
 				newToken.Token = old.Token
 			}
-
-			// Do not update last used time for offline session if token is allowed to be reused
 			lastUsed = old.LastUsed
 			old.ConnectorData = nil
 			return old, nil
@@ -281,29 +302,15 @@ func (s *Server) updateRefreshToken(ctx context.Context, rCtx *refreshContext) (
 			if old.Token != rCtx.requestToken.Token {
 				return old, errors.New("refresh token claimed twice")
 			}
-
-			// Issue new refresh token
 			old.ObsoleteToken = old.Token
 			newToken.Token = storage.NewID()
 		}
 
 		old.Token = newToken.Token
 		old.LastUsed = lastUsed
-
-		// ConnectorData has been moved to OfflineSession
 		old.ConnectorData = nil
 
-		// Call  only once if there is a request which is not in the reuse interval.
-		// This is required to avoid multiple calls to the external IdP for concurrent requests.
-		// Dex will call the connector's Refresh method only once if request is not in reuse interval.
-		ident, rerr = s.refreshWithConnector(ctx, rCtx, ident)
-		if rerr != nil {
-			return old, rerr
-		}
-
-		// Update the claims of the refresh token.
-		//
-		// UserID intentionally ignored for now.
+		// Always update claims
 		old.Claims.Username = ident.Username
 		old.Claims.PreferredUsername = ident.PreferredUsername
 		old.Claims.Email = ident.Email
@@ -313,15 +320,12 @@ func (s *Server) updateRefreshToken(ctx context.Context, rCtx *refreshContext) (
 		return old, nil
 	}
 
-	// Update refresh token in the storage.
-	err := s.storage.UpdateRefreshToken(ctx, rCtx.storageToken.ID, refreshTokenUpdater)
-	if err != nil {
+	if err := s.storage.UpdateRefreshToken(ctx, rCtx.storageToken.ID, refreshTokenUpdater); err != nil {
 		s.logger.ErrorContext(ctx, "failed to update refresh token", "err", err)
 		return nil, ident, newInternalServerError()
 	}
 
-	rerr = s.updateOfflineSession(ctx, rCtx.storageToken, ident, lastUsed)
-	if rerr != nil {
+	if rerr := s.updateOfflineSession(ctx, rCtx.storageToken, ident, lastUsed); rerr != nil {
 		return nil, ident, rerr
 	}
 

server/server.go

@@ -198,6 +198,8 @@ type Server struct {
 	deviceRequestsValidFor time.Duration
 
 	refreshTokenPolicy *RefreshTokenPolicy
+	// mutex to refresh the same token only once for concurrent requests
+	refreshLocks sync.Map
 
 	logger *slog.Logger
 }
@@ -758,6 +760,12 @@ func (s *Server) getConnector(ctx context.Context, id string) (Connector, error)
 	return conn, nil
 }
 
+// getRefreshLock returns a per-refresh-ID mutex.
+func (s *Server) getRefreshLock(refreshID string) *sync.Mutex {
+	m, _ := s.refreshLocks.LoadOrStore(refreshID, &sync.Mutex{})
+	return m.(*sync.Mutex)
+}
+
 type logRequestKey string
 
 const (