For persistent sessions, don't remove user session if there is no session in the remote store (#31756)

Closes #31115

Signed-off-by: Alexander Schwartz <aschwart@redhat.com>

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/InfinispanUserLoginFailureProviderFactory.java

@@ -157,7 +157,7 @@ public class InfinispanUserLoginFailureProviderFactory implements UserLoginFailu
 
             remoteCacheInvoker.addRemoteCache(ispnCache.getName(), remoteCache, maxIdleLoader);
 
-            RemoteCacheSessionListener hotrodListener = RemoteCacheSessionListener.createListener(session, ispnCache, remoteCache, lifespanMsLoader, maxIdleTimeMsLoader);
+            RemoteCacheSessionListener hotrodListener = RemoteCacheSessionListener.createListener(session, ispnCache, remoteCache, lifespanMsLoader, maxIdleTimeMsLoader, null);
             remoteCache.addClientListener(hotrodListener);
             return remoteCache;
         }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/InfinispanUserSessionProviderFactory.java

@@ -366,7 +366,15 @@ public class InfinispanUserSessionProviderFactory implements UserSessionProvider
 
             remoteCacheInvoker.addRemoteCache(ispnCache.getName(), remoteCache, maxIdleLoader);
 
-            RemoteCacheSessionListener hotrodListener = RemoteCacheSessionListener.createListener(session, ispnCache, remoteCache, lifespanMsLoader, maxIdleTimeMsLoader);
+            Runnable onFailover = null;
+            if (useCaches && Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
+                // If persistent sessions are enabled, we want to clear the local caches when a failover of the listener on the remote store changes as we might have missed some of the remote store events
+                // which might have been triggered by another Keycloak site connected to the same remote Infinispan cluster.
+                // Due to this, we can be sure that we never have outdated information in our local cache. All entries will be re-loaded from the remote cache or the database as necessary lazily.
+                onFailover = ispnCache::clear;
+            }
+
+            RemoteCacheSessionListener hotrodListener = RemoteCacheSessionListener.createListener(session, ispnCache, remoteCache, lifespanMsLoader, maxIdleTimeMsLoader, onFailover);
             remoteCache.addClientListener(hotrodListener);
             return remoteCache;
         }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/PersistentUserSessionProvider.java

@@ -42,7 +42,6 @@ import org.infinispan.client.hotrod.exceptions.HotRodClientException;
 import org.infinispan.commons.api.BasicCache;
 import org.infinispan.commons.util.ByRef;
 import org.infinispan.commons.util.concurrent.CompletionStages;
-import org.infinispan.context.Flag;
 import org.infinispan.factories.ComponentRegistry;
 import org.infinispan.persistence.manager.PersistenceManager;
 import org.jboss.logging.Logger;
@@ -423,55 +422,9 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
             return userSession;
         }
 
-        // Try lookup userSession from remoteCache
+        // The logic to remove the local entry if there is no entry in the remote cache that is present in the InfinispanUserSessionProvider is removed here,
-        Cache<String, SessionEntityWrapper<UserSessionEntity>> cache = getCache(offline);
+        // as with persistent sessions we will have only a subset of all sessions in memory (both locally and in the remote store).
-        if (cache == null) {
+        return null;
-            return null;
-        }
-
-        RemoteCache remoteCache = InfinispanUtil.getRemoteCache(cache);
-
-        if (remoteCache != null) {
-            SessionEntityWrapper<UserSessionEntity> remoteSessionEntityWrapper = (SessionEntityWrapper<UserSessionEntity>) remoteCache.get(id);
-            if (remoteSessionEntityWrapper != null) {
-                UserSessionEntity remoteSessionEntity = remoteSessionEntityWrapper.getEntity();
-                remoteSessionEntity.setOffline(offline);
-                log.debugf("getUserSessionWithPredicate(%s): remote cache contains session entity %s", id, remoteSessionEntity);
-
-                UserSessionModel remoteSessionAdapter = wrap(realm, remoteSessionEntity, offline);
-                if (predicate.test(remoteSessionAdapter)) {
-
-                    // Remote entity contains our predicate. Update local cache with the remote entity
-                    SessionEntityWrapper<UserSessionEntity> sessionWrapper = remoteSessionEntity.mergeRemoteEntityWithLocalEntity(sessionTx.get(id, offline));
-
-                    // Replace entity just in ispn cache. Skip remoteStore
-                    cache.getAdvancedCache().withFlags(Flag.SKIP_CACHE_STORE, Flag.SKIP_CACHE_LOAD, Flag.IGNORE_RETURN_VALUES)
-                            .replace(id, sessionWrapper);
-
-                    sessionTx.reloadEntityInCurrentTransaction(realm, id, sessionWrapper);
-
-                    // Recursion. We should have it locally now
-                    return getUserSessionWithPredicate(realm, id, offline, predicate);
-                } else {
-                    log.debugf("getUserSessionWithPredicate(%s): found, but predicate doesn't pass", id);
-
-                    return null;
-                }
-            } else {
-                log.debugf("getUserSessionWithPredicate(%s): not found", id);
-
-                // Session not available on remoteCache. Was already removed there. So removing locally too.
-                // TODO: Can be optimized to skip calling remoteCache.remove
-                removeUserSession(realm, userSession);
-
-                return null;
-            }
-        } else {
-
-            log.debugf("getUserSessionWithPredicate(%s): remote cache not available", id);
-
-            return null;
-        }
     }
 
 

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remotestore/ClientListenerExecutorDecorator.java

@@ -26,6 +26,7 @@ import java.util.concurrent.RejectedExecutionException;
 import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
 import org.infinispan.client.hotrod.event.ClientCacheEntryModifiedEvent;
 import org.infinispan.client.hotrod.event.ClientCacheEntryRemovedEvent;
+import org.infinispan.client.hotrod.event.ClientCacheFailoverEvent;
 import org.infinispan.client.hotrod.event.ClientEvent;
 import org.jboss.logging.Logger;
 import org.keycloak.common.util.MultivaluedHashMap;
@@ -59,7 +60,11 @@ public class ClientListenerExecutorDecorator<K> {
     }
 
 
-    // Explicitly use 3 submit methods to ensure that different type of ClientEvent is not used
+    // Use explicit submit methods to ensure that different type of ClientEvent is not used
+
+    public void submit(ClientCacheFailoverEvent clientCacheFailoverEvent, Runnable r) {
+        decorated.submit(r);
+    }
 
     public void submit(ClientCacheEntryCreatedEvent<K> cacheEntryCreatedEvent, Runnable r) {
         MyClientEvent event = convertIspnClientEvent(cacheEntryCreatedEvent);

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remotestore/RemoteCacheSessionListener.java

@@ -28,10 +28,12 @@ import org.infinispan.client.hotrod.VersionedValue;
 import org.infinispan.client.hotrod.annotation.ClientCacheEntryCreated;
 import org.infinispan.client.hotrod.annotation.ClientCacheEntryModified;
 import org.infinispan.client.hotrod.annotation.ClientCacheEntryRemoved;
+import org.infinispan.client.hotrod.annotation.ClientCacheFailover;
 import org.infinispan.client.hotrod.annotation.ClientListener;
 import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
 import org.infinispan.client.hotrod.event.ClientCacheEntryModifiedEvent;
 import org.infinispan.client.hotrod.event.ClientCacheEntryRemovedEvent;
+import org.infinispan.client.hotrod.event.ClientCacheFailoverEvent;
 import org.infinispan.client.hotrod.event.ClientEvent;
 import org.infinispan.context.Flag;
 import org.jboss.logging.Logger;
@@ -57,6 +59,7 @@ public class RemoteCacheSessionListener<K, V extends SessionEntity>  {
     protected static final Logger logger = Logger.getLogger(RemoteCacheSessionListener.class);
 
     private static final int MAXIMUM_REPLACE_RETRIES = 10;
+    private final Runnable onFailover;
 
     private Cache<K, SessionEntityWrapper<V>> cache;
     private RemoteCache<K, SessionEntityWrapper<V>> remoteCache;
@@ -67,7 +70,8 @@ public class RemoteCacheSessionListener<K, V extends SessionEntity>  {
     private KeycloakSessionFactory sessionFactory;
 
 
-    protected RemoteCacheSessionListener() {
+    protected RemoteCacheSessionListener(Runnable onFailover) {
+        this.onFailover = onFailover;
     }
 
 
@@ -103,6 +107,13 @@ public class RemoteCacheSessionListener<K, V extends SessionEntity>  {
         }
     }
 
+    @ClientCacheFailover
+    public void cacheFailover(ClientCacheFailoverEvent event) {
+        if (onFailover != null) {
+            this.executor.submit(event, onFailover);
+        }
+    }
+
 
     @ClientCacheEntryModified
     public void updated(ClientCacheEntryModifiedEvent event) {
@@ -250,7 +261,7 @@ public class RemoteCacheSessionListener<K, V extends SessionEntity>  {
     }
 
     public static <K, V extends SessionEntity> RemoteCacheSessionListener createListener(KeycloakSession session, Cache<K, SessionEntityWrapper<V>> cache, RemoteCache<K, SessionEntityWrapper<V>> remoteCache,
-                                                                                         SessionFunction<V> lifespanMsLoader, SessionFunction<V> maxIdleTimeMsLoader) {
+                                                                                         SessionFunction<V> lifespanMsLoader, SessionFunction<V> maxIdleTimeMsLoader, Runnable onFailover) {
         /*boolean isCoordinator = InfinispanUtil.isCoordinator(cache);
 
         // Just cluster coordinator will fetch userSessions from remote cache.
@@ -264,7 +275,7 @@ public class RemoteCacheSessionListener<K, V extends SessionEntity>  {
             listener = new DontFetchInitialStateCacheListener();
         }*/
 
-        RemoteCacheSessionListener<K, V> listener = new RemoteCacheSessionListener<>();
+        RemoteCacheSessionListener<K, V> listener = new RemoteCacheSessionListener<>(onFailover);
         listener.init(session, cache, remoteCache, lifespanMsLoader, maxIdleTimeMsLoader);
 
         return listener;

services/src/main/java/org/keycloak/services/managers/UserSessionCrossDCManager.java

@@ -17,10 +17,6 @@
 
 package org.keycloak.services.managers;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-
 import org.jboss.logging.Logger;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.RealmModel;
@@ -87,6 +83,7 @@ public class UserSessionCrossDCManager {
         AuthSessionId authSessionId = asm.decodeAuthSessionId(oldEncodedId);
         String sessionId = authSessionId.getDecodedId();
 
+        // TODO: remove this code once InfinispanUserSessionProvider is removed or no longer using any remote caches, as other implementations don't need this call.
         // This will remove userSession "locally" if it doesn't exist on remoteCache
         kcSession.sessions().getUserSessionWithPredicate(realm, sessionId, false, (UserSessionModel userSession2) -> userSession2 == null);