fix: perform best effort cleanup for concurrently created refresh tokens

Signed-off-by: Danil Grigorev <daniil.grigorev.dev@gmail.com>

Author: Danil-Grigorev

server/handlers.go

@@ -1769,11 +1769,13 @@ func (s *Server) handlePasswordGrant(w http.ResponseWriter, r *http.Request, cli
 				return
 			}
 		} else {
+			inconsitentDB := false
 			if oldTokenRef, ok := session.Refresh[tokenRef.ClientID]; ok {
 				// Delete old refresh token from storage.
 				if err := s.storage.DeleteRefresh(ctx, oldTokenRef.ID); err != nil {
 					if err == storage.ErrNotFound {
 						s.logger.Warn("database inconsistent, refresh token missing", "token_id", oldTokenRef.ID)
+						inconsitentDB = true
 					} else {
 						s.logger.ErrorContext(r.Context(), "failed to delete refresh token", "err", err)
 						s.tokenErrHelper(w, errServerError, "", http.StatusInternalServerError)
@@ -1785,6 +1787,12 @@ func (s *Server) handlePasswordGrant(w http.ResponseWriter, r *http.Request, cli
 
 			// Update existing OfflineSession obj with new RefreshTokenRef.
 			if err := s.storage.UpdateOfflineSessions(ctx, session.UserID, session.ConnID, func(old storage.OfflineSessions) (storage.OfflineSessions, error) {
+				// Perform removal of the old concurrent/uncached refresh token from a freshly fetched instance
+				if inconsitentDB && old.Refresh[tokenRef.ClientID] != nil {
+					if err := s.storage.DeleteRefresh(ctx, old.Refresh[tokenRef.ClientID].ID); err != nil {
+						s.logger.Warn("failover cleanup failed, database inconsistent, refresh token missing", "token_id", old.Refresh[tokenRef.ClientID].ID, "err", err)
+					}
+				}
 				old.Refresh[tokenRef.ClientID] = &tokenRef
 				old.ConnectorData = identity.ConnectorData
 				return old, nil

storage/kubernetes/storage.go

@@ -395,8 +395,17 @@ func (cli *client) ListClients(ctx context.Context) ([]storage.Client, error) {
 	return nil, errors.New("not implemented")
 }
 
-func (cli *client) ListRefreshTokens(ctx context.Context) ([]storage.RefreshToken, error) {
-	return nil, errors.New("not implemented")
+func (cli *client) ListRefreshTokens(ctx context.Context) (tokens []storage.RefreshToken, err error) {
+	var refreshTokenList RefreshList
+	if err = cli.list(resourceRefreshToken, &refreshTokenList); err != nil {
+		return tokens, fmt.Errorf("failed to list passwords: %v", err)
+	}
+
+	for _, token := range refreshTokenList.RefreshTokens {
+		tokens = append(tokens, toStorageRefreshToken(token))
+	}
+
+	return
 }
 
 func (cli *client) ListPasswords(ctx context.Context) (passwords []storage.Password, err error) {
@@ -679,6 +688,25 @@ func (cli *client) GarbageCollect(ctx context.Context, now time.Time) (result st
 		}
 	}
 
+	refreshTokens, err := cli.ListRefreshTokens(ctx)
+	if err != nil {
+		return result, err
+	}
+
+	for _, refreshToken := range refreshTokens {
+		matches, err := cli.refreshTokenMatchesOfflineSession(ctx, refreshToken)
+		if err != nil {
+			return result, err
+		}
+		if matches {
+			continue
+		}
+		if err := cli.DeleteRefresh(ctx, refreshToken.ID); err != nil && err != storage.ErrNotFound {
+			cli.logger.Error("failed to delete orphan refresh token", "err", err)
+			delErr = fmt.Errorf("failed to delete orphan refresh token: %v", err)
+		}
+	}
+
 	var deviceTokens DeviceTokenList
 	if err := cli.listN(resourceDeviceToken, &deviceTokens, gcResultLimit); err != nil {
 		return result, fmt.Errorf("failed to list device tokens: %v", err)
@@ -716,6 +744,19 @@ func (cli *client) GarbageCollect(ctx context.Context, now time.Time) (result st
 	return result, delErr
 }
 
+func (cli *client) refreshTokenMatchesOfflineSession(ctx context.Context, refreshToken storage.RefreshToken) (bool, error) {
+	offlineSessions, err := cli.GetOfflineSessions(ctx, refreshToken.Claims.UserID, refreshToken.ConnectorID)
+	if err != nil {
+		if err == storage.ErrNotFound {
+			return false, nil
+		}
+		return false, err
+	}
+
+	ref := offlineSessions.Refresh[refreshToken.ClientID]
+	return ref != nil && ref.ID == refreshToken.ID, nil
+}
+
 func (cli *client) CreateDeviceRequest(ctx context.Context, d storage.DeviceRequest) error {
 	return cli.post(resourceDeviceRequest, cli.fromStorageDeviceRequest(d))
 }