ahus1 · Oct 26, 2026 · Oct 26, 2026 · Oct 27, 2026 · Oct 27, 2026 · Oct 27, 2026
@@ -18,17 +18,21 @@
 package org.keycloak.models.sessions.infinispan;
 
 
+import java.lang.invoke.MethodHandles;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import org.keycloak.Config;
 import org.keycloak.common.Profile;
 import org.keycloak.common.util.Time;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
+import org.keycloak.executors.ExecutorsProvider;
 import org.keycloak.infinispan.util.InfinispanUtils;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
@@ -37,14 +41,20 @@
 import org.keycloak.models.session.RevokedTokenPersisterProvider;
 import org.keycloak.models.sessions.infinispan.entities.SingleUseObjectValueEntity;
 import org.keycloak.models.sessions.infinispan.transaction.InfinispanTransactionProvider;
+import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.PostMigrationEvent;
 import org.keycloak.provider.EnvironmentDependentProviderFactory;
 import org.keycloak.provider.Provider;
 import org.keycloak.provider.ProviderConfigProperty;
 import org.keycloak.provider.ProviderConfigurationBuilder;
 import org.keycloak.provider.ServerInfoAwareProviderFactory;
 
-import org.infinispan.commons.api.BasicCache;
+import org.infinispan.Cache;
+import org.infinispan.notifications.Listener;
+import org.infinispan.notifications.cachemanagerlistener.annotation.Merged;
+import org.infinispan.notifications.cachemanagerlistener.annotation.ViewChanged;
+import org.infinispan.notifications.cachemanagerlistener.event.ViewChangedEvent;
+import org.jboss.logging.Logger;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.ACTION_TOKEN_CACHE;
 
@@ -53,18 +63,25 @@
  */
 public class InfinispanSingleUseObjectProviderFactory implements SingleUseObjectProviderFactory<InfinispanSingleUseObjectProvider>, EnvironmentDependentProviderFactory, ServerInfoAwareProviderFactory {
 
+    private static final Logger logger = Logger.getLogger(MethodHandles.lookup().lookupClass());
+    private static final String RELOAD_EXECUTOR_TASK = "revoked-token-reload";
+
     public static final String CONFIG_PERSIST_REVOKED_TOKENS = "persistRevokedTokens";
     public static final boolean DEFAULT_PERSIST_REVOKED_TOKENS = true;
     public static final String LOADED = "loaded" + SingleUseObjectProvider.REVOKED_KEY;
 
-    protected BasicCache<String, SingleUseObjectValueEntity> singleUseObjectCache;
+    protected Cache<String, SingleUseObjectValueEntity> singleUseObjectCache;
 
     private volatile boolean initialized;
     private boolean persistRevokedTokens;
+    private final ViewChangeListener viewChangeListener = new ViewChangeListener();
+    private Executor reloadExecutor;
+    private KeycloakSessionFactory sessionFactory;
+    private volatile boolean closed;
 
     @Override
     public Set<Class<? extends Provider>> dependsOn() {
-        return Set.of(InfinispanConnectionProvider.class, InfinispanTransactionProvider.class);
+        return Set.of(InfinispanConnectionProvider.class, InfinispanTransactionProvider.class, ExecutorsProvider.class);
     }
 
     @Override
@@ -84,14 +101,9 @@ private void initialize(KeycloakSession session) {
                 if (!initialized) {
                     RevokedTokenPersisterProvider provider = session.getProvider(RevokedTokenPersisterProvider.class);
                     if (singleUseObjectCache.get(LOADED) == null) {
-                        // in a cluster, multiple Keycloak instances might load the same data in parallel, but that wouldn't matter
-                        provider.getAllRevokedTokens().forEach(revokedToken -> {
-                            long lifespanSeconds = revokedToken.expiry() - Time.currentTime();
-                            if (lifespanSeconds > 0) {
-                                singleUseObjectCache.put(revokedToken.tokenId() + SingleUseObjectProvider.REVOKED_KEY, new SingleUseObjectValueEntity(Collections.emptyMap()),
-                                         Time.toMillis(lifespanSeconds), TimeUnit.MILLISECONDS);
-                            }
-                        });
+                        // LOADED is cluster-wide, so nodes joining an initialized cluster avoid repeating the database scan.
+                        // Multiple nodes can still race during the first startup, but importing the same entries is harmless.
+                        loadRevokedTokens(provider);
                         singleUseObjectCache.put(LOADED, new SingleUseObjectValueEntity(Collections.emptyMap()));
                     }
                     initialized = true;
@@ -104,19 +116,26 @@ private void initialize(KeycloakSession session) {
     public void postInit(KeycloakSessionFactory factory) {
         // It is necessary to put the cache initialization here, otherwise the cache would be initialized lazily, that
         // means also listeners will start only after first cache initialization - that would be too late
-        if (singleUseObjectCache == null) {
-            try (var session = factory.create()) {
+        try (var session = factory.create()) {
+            if (singleUseObjectCache == null) {
                 InfinispanConnectionProvider connections = session.getProvider(InfinispanConnectionProvider.class);
                 singleUseObjectCache = connections.getCache(ACTION_TOKEN_CACHE);
             }
+            if (persistRevokedTokens) {
+                reloadExecutor = session.getProvider(ExecutorsProvider.class).getExecutor(RELOAD_EXECUTOR_TASK);
+            }
         }
 
         if (persistRevokedTokens) {
+            sessionFactory = factory;
+            // Revocation entries can be lost with departed owners while LOADED survives. Reconcile on membership loss
+            // instead of adding a database lookup to every cache miss on the token-validation path.
+            singleUseObjectCache.getCacheManager().addListener(viewChangeListener);
             factory.register(event -> {
                 if (event instanceof PostMigrationEvent pme) {
                     KeycloakSessionFactory sessionFactory = pme.getFactory();
                     try (KeycloakSession session = sessionFactory.create()) {
-                        // load sessions during startup, not on first request to avoid congestion
+                        // Load persisted revocations during startup, not on the first token check, to avoid request-time congestion.
                         initialize(session);
                     }
                 }
@@ -126,7 +145,11 @@ public void postInit(KeycloakSessionFactory factory) {
 
     @Override
     public void close() {
-
+        closed = true;
+        // The cache is not assigned when initialization failed before postInit(), and the cleanup must not mask that failure.
+        if (persistRevokedTokens && singleUseObjectCache != null) {
+            singleUseObjectCache.getCacheManager().removeListener(viewChangeListener);
+        }
     }
 
     @Override
@@ -172,4 +195,88 @@ private static InfinispanKeycloakTransaction createTransaction(KeycloakSession s
         return tx;
     }
 
+    private void loadRevokedTokens(RevokedTokenPersisterProvider provider) {
+        provider.getAllRevokedTokens().forEach(revokedToken -> {
+            // The scan can run for a long time, so the time is taken per record to skip tokens that expired meanwhile.
+            long lifespanSeconds = revokedToken.expiry() - Time.currentTimeSeconds();
+            if (lifespanSeconds > 0) {
+                String key = revokedToken.tokenId() + SingleUseObjectProvider.REVOKED_KEY;
+                SingleUseObjectValueEntity value = new SingleUseObjectValueEntity(Collections.emptyMap());
+                // Loading only fills gaps. Replacing an entry written by another node would unnecessarily restart its
+                // cache lifespan, both during concurrent startup and membership-change recovery.
+                singleUseObjectCache.putIfAbsent(key, value, lifespanSeconds, TimeUnit.SECONDS);
+            }
+        });
+    }
+
+    private void requestReload() {
+        // Every member observes the view change, so skip submitting a task on nodes that would not scan anyway.
+        // The coordinator is checked again when the task runs, as it can change in between.
+        if (!closed && singleUseObjectCache.getCacheManager().isCoordinator()) {
+            // Keycloak's executor provider owns this shared pool. Offloading avoids blocking the Infinispan notification
+            // thread without retaining a thread owned by this provider between the rare cluster membership changes.
+            try {
+                reloadExecutor.execute(this::reloadRevokedTokens);
+            } catch (RejectedExecutionException e) {
+                // The default executor has an unbounded queue, so a rejection means the executor is shutting down.
+                // There is nothing left to retry, but clearing the marker lets the next node starting up reload.
+                if (!closed) {
+                    logger.warn("Failed to submit the reload of revoked tokens after a cluster membership change", e);
+                }
+                clearLoadedMarker();
+            }
+        }
+    }
+
+    private void clearLoadedMarker() {
+        try {
+            // Without the marker, the next node starting up loads the persisted revocations again. This also applies
+            // to every later node until a reload succeeds, which is intended: the cache may be incomplete, and
+            // claiming otherwise would hide the missing entries. A successful scan sets the marker again.
+            singleUseObjectCache.remove(LOADED);
+        } catch (RuntimeException removeFailure) {
+            if (!closed) {
+                logger.warn("Failed to clear the revoked tokens loaded marker", removeFailure);
+            }
+        }
+    }
+
+    private synchronized void reloadRevokedTokens() {
+        if (closed) {
+            return;
+        }
+
+        // Every member observes the view change. Restrict the database scan to one node; writes to the distributed
+        // cache restore the entries for the whole cluster.
+        if (!singleUseObjectCache.getCacheManager().isCoordinator()) {
+            return;
+        }
+
+        try {
+            logger.debug("Reloading revoked tokens after a cluster membership change");
+            KeycloakModelUtils.runJobInTransaction(sessionFactory, session ->
+                    loadRevokedTokens(session.getProvider(RevokedTokenPersisterProvider.class)));
+        } catch (RuntimeException e) {
+            if (!closed) {
+                logger.warn("Failed to reload revoked tokens after a cluster membership change", e);
+                clearLoadedMarker();
+            }
+        }
+    }
+
+    @Listener
+    public class ViewChangeListener {
+
+        // On a merge, Infinispan notifies only @Merged listeners and not @ViewChanged, so both are needed.
+        @Merged
+        @ViewChanged
+        public void viewChanged(ViewChangedEvent event) {
+            // Joining members receive state from existing owners and must not make every scale-up scan the database.
+            // After a merge, neither partition can assume its surviving entries are complete.
+            if (event.isMergeView() || !event.getNewMembers().containsAll(event.getOldMembers())) {
+                requestReload();
+            }
+        }
+    }
+
 }
@@ -20,7 +20,6 @@
 import java.util.stream.Stream;
 
 import jakarta.persistence.EntityManager;
-import jakarta.persistence.TypedQuery;
 import jakarta.persistence.criteria.CriteriaBuilder;
 import jakarta.persistence.criteria.CriteriaDelete;
 import jakarta.persistence.criteria.CriteriaQuery;
@@ -45,7 +44,7 @@ public JpaRevokedTokensPersisterProvider(EntityManager em) {
     @Override
     public void revokeToken(String tokenId, long lifetime) {
         RevokedTokenEntity revokedTokenEntity = em.find(RevokedTokenEntity.class, tokenId);
-        long expire = Time.currentTime() + lifetime;
+        long expire = Time.currentTimeSeconds() + lifetime;
         if (revokedTokenEntity != null) {
             // The token has already been revoked.
             // There shouldn't be much need to update the expiry of the token, let's do it anyway to be on the safe side.
@@ -64,21 +63,20 @@ public void revokeToken(String tokenId, long lifetime) {
     @Override
     public Stream<RevokedToken> getAllRevokedTokens() {
         CriteriaBuilder cb = em.getCriteriaBuilder();
-        CriteriaQuery<RevokedTokenEntity> cq = cb.createQuery(RevokedTokenEntity.class);
+        CriteriaQuery<RevokedToken> cq = cb.createQuery(RevokedToken.class);
         Root<RevokedTokenEntity> rootEntry = cq.from(RevokedTokenEntity.class);
-        CriteriaQuery<RevokedTokenEntity> all = cq.select(rootEntry)
-                .where(cb.gt(rootEntry.get("expire"), Time.currentTime()));
+        CriteriaQuery<RevokedToken> all = cq.select(cb.construct(RevokedToken.class, rootEntry.get("id"), rootEntry.get("expire")))
+                .where(cb.gt(rootEntry.get("expire"), Time.currentTimeSeconds()));
 
-        TypedQuery<RevokedTokenEntity> allQuery = em.createQuery(all);
-        return allQuery.getResultStream().map(revokedTokenEntity -> new RevokedToken(revokedTokenEntity.getId(), revokedTokenEntity.getExpire()));
+        return em.createQuery(all).getResultStream();
     }
 
     @Override
     public void expireTokens() {
         CriteriaBuilder cb = em.getCriteriaBuilder();
         CriteriaDelete<RevokedTokenEntity> cd = cb.createCriteriaDelete(RevokedTokenEntity.class);
         Root<RevokedTokenEntity> rootEntry = cd.from(RevokedTokenEntity.class);
-        cd.where(cb.lt(rootEntry.get("expire"), Time.currentTime()));
+        cd.where(cb.lt(rootEntry.get("expire"), Time.currentTimeSeconds()));
         em.createQuery(cd).executeUpdate();
     }
 

@@ -1,17 +1,23 @@
 package org.keycloak.testsuite.model.infinispan;
 
+import java.time.Duration;
+import java.util.UUID;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
 
 import org.keycloak.common.Profile;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.Constants;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.RealmModel;
+import org.keycloak.models.SingleUseObjectProvider;
 import org.keycloak.models.cache.CacheRealmProvider;
+import org.keycloak.models.sessions.infinispan.InfinispanSingleUseObjectProviderFactory;
+import org.keycloak.models.sessions.infinispan.entities.SingleUseObjectValueEntity;
 import org.keycloak.testsuite.model.KeycloakModelTest;
 import org.keycloak.testsuite.model.RequireProvider;
 
@@ -20,8 +26,11 @@
 import org.junit.Test;
 import org.junit.rules.TestRule;
 
+import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 
 /**
@@ -102,6 +111,65 @@ public void testLocalCacheClearedOnMergeEvent() throws InterruptedException {
       });
    }
 
+   @Test
+   public void testRevokedTokensReloadedOnMergeEvent() throws InterruptedException {
+      // The actionTokens cache does not exist when STATELESS is enabled
+      Assume.assumeFalse(Profile.isFeatureEnabled(Profile.Feature.STATELESS));
+      var numFactories = 2;
+      var partitionManager = new PartitionManager(numFactories);
+      var factoryIndex = new AtomicInteger(0);
+      var barrier = new CyclicBarrier(numFactories);
+      var tokenId = UUID.randomUUID().toString();
+      closeKeycloakSessionFactory();
+
+      inIndependentFactories(numFactories, 60, () -> {
+         var index = factoryIndex.getAndIncrement();
+
+         inComittedTransaction(session -> {
+            var cacheManager = session.getProvider(InfinispanConnectionProvider.class)
+                  .getCache(InfinispanConnectionProvider.ACTION_TOKEN_CACHE)
+                  .getCacheManager();
+            partitionManager.addManager(index, cacheManager);
+
+            if (index == 0) {
+               session.revokedTokens().put(tokenId, 60);
+            }
+         });
+         awaitBarrier(barrier);
+
+         if (index == 0) {
+            partitionManager.splitCluster(new int[]{0}, new int[]{1});
+         }
+         awaitBarrier(barrier);
+
+         // Simulate both partitions losing the cached entry while the database and cluster-wide startup marker survive.
+         inComittedTransaction(session -> {
+            var cache = session.getProvider(InfinispanConnectionProvider.class)
+                  .<String, SingleUseObjectValueEntity>getCache(InfinispanConnectionProvider.ACTION_TOKEN_CACHE);
+            cache.remove(tokenId + SingleUseObjectProvider.REVOKED_KEY);
+            assertNotNull(cache.get(InfinispanSingleUseObjectProviderFactory.LOADED));
+            assertFalse(session.revokedTokens().contains(tokenId));
+         });
+         awaitBarrier(barrier);
+
+         if (index == 0) {
+            partitionManager.merge(0, 1);
+         }
+         awaitBarrier(barrier);
+
+         // The reload runs asynchronously after the merge view is installed.
+         await("Persisted revocation was not restored after the view merge")
+               .atMost(Duration.ofSeconds(30))
+               .pollInterval(Duration.ofMillis(100))
+               // the session factory is bound to the thread running inIndependentFactories
+               .pollInSameThread()
+               .until(() -> {
+                  Function<KeycloakSession, Boolean> contains = session -> session.revokedTokens().contains(tokenId);
+                  return inComittedTransaction(contains);
+               });
+      });
+   }
+
    private void awaitBarrier(CyclicBarrier barrier) {
       try {
          barrier.await(10, TimeUnit.SECONDS);