diff --git a/openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java b/openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java index cb7d0e60f1..c1f36663a7 100644 --- a/openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java +++ b/openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java @@ -12,6 +12,7 @@ * information: "Portions copyright [year] [name of copyright owner]". * * Copyright 2015-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package com.iplanet.services.ldap.event; @@ -54,6 +55,7 @@ import org.forgerock.opendj.ldap.responses.SearchResultEntry; import org.forgerock.opendj.ldap.responses.SearchResultReference; import org.forgerock.util.annotations.VisibleForTesting; +import org.forgerock.util.promise.ExceptionHandler; import com.sun.identity.common.GeneralTaskRunnable; import com.sun.identity.common.SystemTimerPool; @@ -73,6 +75,7 @@ public abstract class LDAPv3PersistentSearch { private static final boolean CHANGES_ONLY = true; private static final boolean RETURN_CONTROLS = true; private static final boolean IS_CRITICAL = true; + private static final long ERROR_LOG_INTERVAL = 60000; private static final List AD_DEFAULT_ATTRIBUTES = Collections.unmodifiableList(Arrays.asList( AD_IS_DELETED_ATTR, AD_WHEN_CHANGED_ATTR, AD_WHEN_CREATED_ATTR)); @@ -88,6 +91,7 @@ public abstract class LDAPv3PersistentSearch { private LdapPromise futureResult; private PersistentSearchMode mode; private RetryTask retryTask; + private volatile long lastErrorLogged = 0; private enum PersistentSearchMode { STANDARD, AD, NONE @@ -184,6 +188,20 @@ private void logError(Throwable t) { DEBUG.error("An error occurred while trying to initiate persistent search connection", t); } + /** + * Logs a failure of a running or restarting search at most once per {@link #ERROR_LOG_INTERVAL}. A directory that + * keeps refusing the search fails it again on every restart, so every retry interval would log it otherwise. + */ + private void logRestartError(String message, Throwable t) { + long now = currentTimeMillis(); + if (now - lastErrorLogged > ERROR_LOG_INTERVAL) { + lastErrorLogged = now; + DEBUG.error(message, t); + } else { + DEBUG.message(message, t); + } + } + private void startSearch(Connection conn) throws LdapException { if (mode == null) { detectPersistentSearchMode(conn); @@ -224,6 +242,7 @@ private void startSearch(Connection conn) throws LdapException { //cache, while PS was stopped, those gets cleared out and we start with a clean cache. clearCaches(); futureResult = conn.searchAsync(searchRequest, null, new PersistentSearchResultHandler()); + futureResult.thenOnException(new PersistentSearchExceptionHandler()); } /** @@ -365,10 +384,27 @@ public boolean handleReference(SearchResultReference reference) { } + /** + * Restarts the persistent search once the directory terminates it, e.g. because the connection was lost. + */ + private class PersistentSearchExceptionHandler implements ExceptionHandler { + + @Override + public void handleException(LdapException error) { + if (!shutdown) { + logRestartError("An error occurred while executing persistent search", error); + DEBUG.message("Restarting persistent search. Some changes may have been missed in the interim."); + clearCaches(); + restartSearch(); + } else { + DEBUG.message("Persistent search has been cancelled", error); + } + } + } + private class RetryTask extends GeneralTaskRunnable { private long runPeriod; - private long lastLogged = 0; public RetryTask() { runPeriod = retryInterval; @@ -391,18 +427,18 @@ public long getRunPeriod() { } public void run() { + //the failure handler may have scheduled this task while the search was being stopped + if (shutdown) { + runPeriod = -1; + return; + } try { conn = factory.create(); startSearch(conn); - //everything seems to work, let's disable retryTask and reset the debug limit + //everything seems to work, let's disable retryTask runPeriod = -1; - lastLogged = 0; } catch (DataLayerException | LdapException ex) { - long now = currentTimeMillis(); - if (now - lastLogged > 60000) { - DEBUG.error("Unable to start persistent search: " + ex.getMessage()); - lastLogged = now; - } + logRestartError("Unable to start persistent search: " + ex.getMessage(), null); IOUtils.closeIfNotNull(conn); } } diff --git a/openam-core/src/main/java/org/forgerock/openam/session/service/access/persistence/watchers/SessionModificationWatcher.java b/openam-core/src/main/java/org/forgerock/openam/session/service/access/persistence/watchers/SessionModificationWatcher.java index 743bbd8555..29fe6c53b3 100644 --- a/openam-core/src/main/java/org/forgerock/openam/session/service/access/persistence/watchers/SessionModificationWatcher.java +++ b/openam-core/src/main/java/org/forgerock/openam/session/service/access/persistence/watchers/SessionModificationWatcher.java @@ -12,7 +12,7 @@ * information: "Portions copyright [year] [name of copyright owner]". * * Copyright 2016 ForgeRock AS. - * Portions copyright 2025 3A Systems LLC. + * Portions copyright 2025-2026 3A Systems LLC. */ package org.forgerock.openam.session.service.access.persistence.watchers; @@ -109,7 +109,8 @@ public void objectsChanged(Set tokenIds) { @Override public void connectionLost() { - debug.error("Continuous query listener has lost its connection"); + // The persistent search already reports the failure at error level, at most once per minute. + debug.warning("Continuous query listener has lost its connection"); } @Override diff --git a/openam-core/src/main/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearch.java b/openam-core/src/main/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearch.java index 8e97bd1dc0..bc32d6f838 100644 --- a/openam-core/src/main/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearch.java +++ b/openam-core/src/main/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearch.java @@ -12,6 +12,7 @@ * information: "Portions copyright [year] [name of copyright owner]". * * Copyright 2016 ForgeRock AS. +* Portions Copyright 2026 3A Systems, LLC. */ package org.forgerock.openam.sm.datalayer.impl.ldap; @@ -19,6 +20,7 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import org.forgerock.openam.cts.continuous.ChangeType; import org.forgerock.openam.cts.continuous.ContinuousQuery; @@ -44,6 +46,7 @@ public class CTSDJLDAPv3PersistentSearch extends LDAPv3PersistentSearchcreate()); + + //when + pSearch.startQuery(); + + //then + verify(mockListener, never()).connectionLost(); + } + @Test - public void shouldInformListenersConnectionDown() { + public void shouldInformListenersConnectionDownWhenSearchFails() throws Exception { //given ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); ContinuousQueryListener mockListener2 = mock(ContinuousQueryListener.class); pSearch.addContinuousQueryListener(mockListener); pSearch.addContinuousQueryListener(mockListener2); + PromiseImpl searchResult = PromiseImpl.create(); + givenDirectoryAcceptsPersistentSearch(searchResult); + pSearch.startQuery(); //when - pSearch.clearCaches(); + searchResult.handleException(newLdapException(ResultCode.UNAVAILABLE)); //then verify(mockListener, times(1)).connectionLost(); verify(mockListener2, times(1)).connectionLost(); } + @Test + public void shouldRestartSearchWhenSearchFails() throws Exception { + //given + pSearch = new CTSDJLDAPv3PersistentSearch(1000, DN.rootDN(), Filter.alwaysFalse(), + SearchScope.WHOLE_SUBTREE, mockConnectionFactory); + pSearch.addContinuousQueryListener(mock(ContinuousQueryListener.class)); + PromiseImpl searchResult = PromiseImpl.create(); + givenDirectoryAcceptsPersistentSearch(searchResult); + pSearch.startQuery(); + + //when + searchResult.handleException(newLdapException(ResultCode.UNAVAILABLE)); + + //then - every restart gets the same failed search, so the count keeps growing + verify(mockConnectionFactory, timeout(5000).atLeast(2)).create(); + } + + @Test + public void shouldRestartSearchWhenDirectoryRefusesSearch() throws Exception { + //given + pSearch = new CTSDJLDAPv3PersistentSearch(1000, DN.rootDN(), Filter.alwaysFalse(), + SearchScope.WHOLE_SUBTREE, mockConnectionFactory); + pSearch.addContinuousQueryListener(mock(ContinuousQueryListener.class)); + PromiseImpl searchResult = PromiseImpl.create(); + givenDirectoryAcceptsPersistentSearch(searchResult); + pSearch.startQuery(); + + //when + searchResult.handleException(newLdapException(ResultCode.INSUFFICIENT_ACCESS_RIGHTS)); + + //then - the access rights may be fixed without restarting AM, so the search keeps being restarted, + //also after the restarted search is refused again + verify(mockConnectionFactory, timeout(5000).atLeast(3)).create(); + } + + @Test + public void shouldInformListenersWhenSearchIsReestablished() throws Exception { + //given + pSearch = new CTSDJLDAPv3PersistentSearch(1000, DN.rootDN(), Filter.alwaysFalse(), + SearchScope.WHOLE_SUBTREE, mockConnectionFactory); + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + pSearch.addContinuousQueryListener(mockListener); + PromiseImpl failedSearch = PromiseImpl.create(); + givenDirectoryAcceptsPersistentSearch(failedSearch, PromiseImpl.create()); + pSearch.startQuery(); + + //when + failedSearch.handleException(newLdapException(ResultCode.UNAVAILABLE)); + + //then - one call when the search fails, one when it is re-established + verify(mockConnectionFactory, timeout(5000).times(2)).create(); + verify(mockListener, timeout(5000).times(2)).connectionLost(); + } + + @Test + public void shouldNotRestartSearchWhenStopped() throws Exception { + //given + pSearch = new CTSDJLDAPv3PersistentSearch(1000, DN.rootDN(), Filter.alwaysFalse(), + SearchScope.WHOLE_SUBTREE, mockConnectionFactory); + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + pSearch.addContinuousQueryListener(mockListener); + PromiseImpl searchResult = PromiseImpl.create(); + givenDirectoryAcceptsPersistentSearch(searchResult); + pSearch.startQuery(); + pSearch.stopQuery(); + + //when + searchResult.handleException(newLdapException(ResultCode.CLIENT_SIDE_USER_CANCELLED)); + + //then + verify(mockListener, never()).connectionLost(); + verify(mockConnectionFactory, after(2000).times(1)).create(); + } + + private void givenDirectoryAcceptsPersistentSearch(PromiseImpl searchResult) + throws Exception { + givenDirectoryAcceptsPersistentSearch(searchResult, searchResult); + } + + private void givenDirectoryAcceptsPersistentSearch(PromiseImpl firstSearchResult, + PromiseImpl laterSearchResult) throws Exception { + Connection connection = mock(Connection.class); + given(mockConnectionFactory.create()).willReturn(connection); + given(connection.searchSingleEntry(any(SearchRequest.class))).willReturn( + Responses.newSearchResultEntry("").addAttribute("supportedControl", PersistentSearchRequestControl.OID)); + given(connection.searchAsync(any(SearchRequest.class), any(), any(SearchResultHandler.class))) + .willReturn(LdapPromises.asPromise(firstSearchResult), LdapPromises.asPromise(laterSearchResult)); + } + @Test public void shouldStopAllListenersAndStopSearch() { //given