From ef384b2c1b3bd741a29a768e6cd24f28117b381c Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 13 Oct 2026 11:58:22 +0300 Subject: [PATCH 1/3] [#1175] Restart CTS persistent search after connection loss Since the OpenDJ SDK 3 migration (AME-11134) the error handler of the persistent search was dropped, so a psearch that the directory terminated (server restart, failover, idle timeout) was never restarted and cross-server session invalidation / notifications silently stopped. Re-attach an exception handler to the search promise that clears caches and schedules the retry task. Also stop reporting the initial start of a CTS continuous query as "Continuous query listener has lost its connection". --- .../ldap/event/LDAPv3PersistentSearch.java | 21 +++++ .../ldap/CTSDJLDAPv3PersistentSearch.java | 7 ++ .../ldap/CTSDJLDAPv3PersistentSearchTest.java | 88 ++++++++++++++++++- 3 files changed, 114 insertions(+), 2 deletions(-) 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..609522a576 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; @@ -224,6 +226,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,6 +368,24 @@ 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) { + DEBUG.error("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; 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 + verify(mockConnectionFactory, timeout(5000).times(2)).create(); + } + + @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 { + 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(searchResult)); + } + @Test public void shouldStopAllListenersAndStopSearch() { //given From 0d5a2209ab463630ed683fcf796925506527ccb5 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 14 Oct 2026 11:07:28 +0300 Subject: [PATCH 2/3] [#1175] Throttle psearch restart errors and stop stray retry tasks A directory that refuses the persistent search fails it again on every restart, and each restart creates a new RetryTask, so its 60 s log throttle never applied. Move the throttle to the search itself and share it between the failure handler and RetryTask. RetryTask now does nothing once the search is shut down: the failure handler can schedule it from another thread while stopSearch() runs. Add tests for restarting a refused search and for notifying listeners when the search is re-established. --- .../ldap/event/LDAPv3PersistentSearch.java | 33 ++++++++++---- .../ldap/CTSDJLDAPv3PersistentSearchTest.java | 43 ++++++++++++++++++- 2 files changed, 66 insertions(+), 10 deletions(-) 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 609522a576..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 @@ -75,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)); @@ -90,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 @@ -186,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); @@ -376,7 +392,7 @@ private class PersistentSearchExceptionHandler implements ExceptionHandler 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/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java b/openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java index 5195050359..fff9c25815 100644 --- a/openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java +++ b/openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java @@ -112,6 +112,42 @@ public void shouldRestartSearchWhenSearchFails() throws Exception { verify(mockConnectionFactory, timeout(5000).times(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 + verify(mockConnectionFactory, timeout(5000).times(2)).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 @@ -134,12 +170,17 @@ public void shouldNotRestartSearchWhenStopped() throws Exception { 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(searchResult)); + .willReturn(LdapPromises.asPromise(firstSearchResult), LdapPromises.asPromise(laterSearchResult)); } @Test From 2a64becfbe4efca3270c4f3480ea97c729e5e81a Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 16 Oct 2026 13:20:43 +0300 Subject: [PATCH 3/3] [#1175] Log lost CTS query connection at warning level A directory that keeps refusing the persistent search makes the CTS search notify connectionLost() twice per retry interval, outside the 60 s throttle, and SessionModificationWatcher logged every call at ERROR. Log it at warning level, as CTSNotificationBroker already does: the throttled error of the search already reports the failure. The restart tests verified an exact create() count that keeps growing while every restart gets the same failed search. Verify a lower bound, and require a second restart for the refused search. --- .../persistence/watchers/SessionModificationWatcher.java | 5 +++-- .../impl/ldap/CTSDJLDAPv3PersistentSearchTest.java | 9 +++++---- 2 files changed, 8 insertions(+), 6 deletions(-) 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/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java b/openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java index fff9c25815..044879862b 100644 --- a/openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java +++ b/openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java @@ -108,8 +108,8 @@ public void shouldRestartSearchWhenSearchFails() throws Exception { //when searchResult.handleException(newLdapException(ResultCode.UNAVAILABLE)); - //then - verify(mockConnectionFactory, timeout(5000).times(2)).create(); + //then - every restart gets the same failed search, so the count keeps growing + verify(mockConnectionFactory, timeout(5000).atLeast(2)).create(); } @Test @@ -125,8 +125,9 @@ public void shouldRestartSearchWhenDirectoryRefusesSearch() throws Exception { //when searchResult.handleException(newLdapException(ResultCode.INSUFFICIENT_ACCESS_RIGHTS)); - //then - the access rights may be fixed without restarting AM, so the search keeps being restarted - verify(mockConnectionFactory, timeout(5000).times(2)).create(); + //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