[#1175] Restart CTS persistent search after connection loss

[vharseko]

Fixes #1175

Problem

• Since the OpenDJ SDK 3 migration (AME-11134, 536ce7533c), LDAPv3PersistentSearch lost its handleErrorResult() and nothing replaced it: no handler is attached to the promise returned by searchAsync. When the directory terminates a running persistent search (CTS/DS restart or failover, an idle timeout on a load balancer or firewall), the search is never restarted. restartSearch() was only reachable when the initial start failed. For the CTS this means cross-server session invalidation (SessionModificationWatcher) and CTS notifications (CTSNotificationBroker) silently stop until AM is restarted. The same base class backs the EventService (config store) and DJLDAPv3PersistentSearch (IdRepo) searches.
• startSearch() always calls clearCaches(), and the CTS implementation turns that into ContinuousQueryListener.connectionLost(). As a result, every normal start of a continuous query is logged at ERROR as Continuous query listener has lost its connection, which is what the reporter of AM goes down with CTs exception often #1175 sees.

Change

• LDAPv3PersistentSearch: attach a PersistentSearchExceptionHandler to the search promise. Unless the search is being shut down, the handler clears the caches and schedules the existing RetryTask. This restores the pre-AME-11134 behaviour.
• LDAPv3PersistentSearch: the search keeps being restarted whatever the result code, because a refusal such as INSUFFICIENT_ACCESS_RIGHTS can be fixed on the directory without restarting AM. The 60 s error-log throttle moved from RetryTask to the search itself and is shared by the handler and RetryTask, so a directory that keeps refusing the search logs one ERROR per minute instead of one per retry interval.
• LDAPv3PersistentSearch: RetryTask.run() does nothing once the search is shut down. The handler runs on the thread that completes the promise and can schedule a retry while stopSearch() runs.
• CTSDJLDAPv3PersistentSearch: the first clearCaches() call (the initial start) no longer notifies connectionLost(). A real loss and the later re-establishment of the search still notify the listeners.

Tests

• CTSDJLDAPv3PersistentSearchTest covers six cases: no connectionLost() on the initial start; connectionLost() for every listener when the search fails; a reconnect after a failure; a reconnect after the directory refuses the search with INSUFFICIENT_ACCESS_RIGHTS; connectionLost() again when the search is re-established; no restart after stopQuery().
• Mutation checks: removing the thenOnException line fails the failure and restart tests; skipping clearCaches() when the search is re-established fails the re-establishment test.
• Not covered by a test: the race between the handler and stopSearch(), and the log throttle.
• openam-core (2142 tests) passes locally.

[maximthomas]

praise: The fix goes where the bug is, and it removes the false ERROR that #1175 reports.

• futureResult.thenOnException(new PersistentSearchExceptionHandler()) (LDAPv3PersistentSearch.java:229) attaches a handler to the promise that has had none since AME-11134, so the CTS, EventService and IdRepo searches all get the restart from this one line.
• restartSearch() creates a new RetryTask for every restart. If a failure completes synchronously inside RetryTask.run(), the old task's runPeriod = -1 therefore cannot cancel the new task.
• searchStarted.compareAndSet(false, true) (CTSDJLDAPv3PersistentSearch.java:70) stops the "Continuous query listener has lost its connection" ERROR from being logged on every normal start.

---

question (non-blocking): Should the handler stop restarting, or back off, when the directory refuses the persistent search for a reason that a retry cannot fix?

openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java:374-387, :254-266

handleException restarts on every LdapException and never checks the result code. Each restart creates a new RetryTask, so the 60 s lastLogged throttle in RetryTask never applies on this path. Some directories accept the connection and the root DSE read but reject the psearch itself, for example with INSUFFICIENT_ACCESS_RIGHTS or UNAVAILABLE_CRITICAL_EXTENSION on the critical control. Against such a directory, the code repeats the following every ~3 s (EventService.RETRY_INTERVAL) until AM restarts: an ERROR with a stack trace, and clearCaches() twice (startSearch and the handler). For EventService that is a full SMS cache flush, for IdRepo an IdRepo cache flush, and for CTS two "lost its connection" lines per listener. Before this PR the search died once and stayed dead. shouldRestartSearchWhenSearchFails already runs this loop, because its mock returns the same failed promise on every attempt. Severity: minor; major if AM is expected to run against directories that refuse the psearch control. If the endless retry is intended (pre-AME-11134 parity), the pin is a test that fails the promise with INSUFFICIENT_ACCESS_RIGHTS and verifies the restart.

// import org.forgerock.opendj.ldap.ResultCode;
@Override
public void handleException(LdapException error) {
    if (shutdown) {
        DEBUG.message("Persistent search has been cancelled", error);
        return;
    }
    ResultCode resultCode = error.getResult().getResultCode();
    if (ResultCode.INSUFFICIENT_ACCESS_RIGHTS.equals(resultCode)
            || ResultCode.UNAVAILABLE_CRITICAL_EXTENSION.equals(resultCode)) {
        DEBUG.error("The directory refused the persistent search, it will not be restarted", error);
        return;
    }
    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();
}

Pin: fail the promise with ResultCode.INSUFFICIENT_ACCESS_RIGHTS and check verify(mockConnectionFactory, after(2000).times(1)).create(). On the head this goes red.
Or: keep restarting, but log through the same 60 s throttle as RetryTask and back off the interval.

---

suggestion (non-blocking): No test checks the connectionLost() call made when the search is re-established, which the description promises.

openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java:80-113, openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java:227

shouldInformListenersConnectionDownWhenSearchFails checks times(1) before the retry fires. shouldRestartSearchWhenSearchFails only counts create(). The mutant clearCaches(); → if (futureResult == null) { clearCaches(); } at :227 stops the re-established search from notifying anything, including the EventService/IdRepo cache flush after the outage, and it survives 9/9. Counting calls in the existing restart test would not kill it: that test returns the same failed promise on every attempt, so the handler supplies the second call by itself.

@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<Result, LdapException> failedSearch = PromiseImpl.create();
    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(failedSearch),
                    LdapPromises.asPromise(PromiseImpl.<Result, LdapException>create()));
    pSearch.startQuery();

    //when
    failedSearch.handleException(newLdapException(ResultCode.UNAVAILABLE));

    //then - one call from the handler, one from the re-established search
    verify(mockListener, timeout(5000).times(2)).connectionLost();
}

Pin: green on the head; on the mutant above the count stays at 1.

---

issue (non-blocking): A restart triggered by the new handler can be scheduled after stopSearch() has run, and the resulting RetryTask is never cancelled.

openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java:254-266, :236-252, :414-428

Before this PR, restartSearch() was called only from startQuery() on the caller thread. Now the handler calls it from whichever thread completes the promise. That thread can pass the !shutdown check at :258. If stopSearch() then sets shutdown, cancels the old (or null) retryTask and closes the factory, the handler still assigns and schedules a new task at :260-263, and nothing cancels it. RetryTask.run() never checks shutdown. For the CTS pool the stray task fails once on the closed pool and dies. What the EventService and IdRepo factories do after close() was not traced. Not run: it needs a harness that stops the search between :258 and :263. TimerPool holds the head lock across run(), so one check at the start of run() covers every factory.

public void run() {
    if (shutdown) {
        runPeriod = -1;
        return;
    }
    try {

[vharseko]

All three points are addressed in 36e7fe3. The branch is also rebased onto the current master.

question: restarting a search the directory refuses. The search keeps restarting, but the error is now throttled. A stop on INSUFFICIENT_ACCESS_RIGHTS / UNAVAILABLE_CRITICAL_EXTENSION would turn a fixable ACI problem on the DS side into the silent failure #1175 is about: the access rights can be fixed without restarting AM, and the search would never come back. The handler and the restart before AME-11134 were identical to this PR, so the endless retry is the old behaviour.

• The 60 s throttle moved from RetryTask to LDAPv3PersistentSearch (logRestartError), so it survives the new RetryTask that every restart creates. The handler and RetryTask share it: one ERROR per 60 s, the rest at message level.
• The throttle is no longer reset when startSearch() returns: against a refusing directory startSearch() succeeds and the refusal arrives asynchronously, so the reset would re-enable the ERROR on every cycle.
• No back-off: the cost of a refusing directory is one cache flush per retry interval, which an operator can tune through the existing retry interval settings.
• Pin: shouldRestartSearchWhenDirectoryRefusesSearch fails the promise with INSUFFICIENT_ACCESS_RIGHTS and checks the restart.

suggestion: no test for connectionLost() on re-establishment. Added shouldInformListenersWhenSearchIsReestablished as you proposed: the first search fails, the second stays open, connectionLost() is expected twice. The mutant if (futureResult == null) { clearCaches(); } now fails exactly this test.

issue: RetryTask scheduled after stopSearch(). RetryTask.run() now returns with runPeriod = -1 when shutdown is set. One correction to the description: the stray task would not necessarily fail once and die. catch keeps runPeriod at the retry interval, so a closed factory that throws DataLayerException reschedules it on every interval. The guard covers that case as well. The race itself is not covered by a test, it needs the search to be stopped between the shutdown check and the schedule call.

[maximthomas]

issue (non-blocking): The 60 s throttle does not cover the CTS listeners: a directory that keeps refusing the search still logs Continuous query listener has lost its connection at ERROR twice per retry interval.

openam-core/src/main/java/com/iplanet/services/ldap/event/LDAPv3PersistentSearch.java:397, :243, openam-core/src/main/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearch.java:68-76

Each refusal cycle calls clearCaches() twice — in PersistentSearchExceptionHandler.handleException (:397) and again in startSearch() (:243) from RetryTask.run() — and logRestartError wraps neither. After the first start, CTSDJLDAPv3PersistentSearch.clearCaches() turns every call into connectionLost(), and SessionModificationWatcher.connectionLost() logs at ERROR (SessionModificationWatcher.java:111-112). With the default 3000 ms interval (EventService.RETRY_INTERVAL, used for the CTS at LdapQueryBuilder.java:119), a CTS that refuses the search with INSUFFICIENT_ACCESS_RIGHTS logs about 40 of the lines #1175 complains about per minute, plus as many CTSNotificationBroker WARNs, until the ACI is fixed. The description says "one ERROR per minute". For the config store and IdRepo the same loop flushes the caches twice per retry interval, not once. Traced at the head, not run.

@Override
public void connectionLost() {
    debug.warning("Continuous query listener has lost its connection");
}

Or: reword the description to "the search's own ERROR is logged once per minute". The search's throttled An error occurred while executing persistent search already reports the outage at ERROR.

---

suggestion (non-blocking): shouldRestartSearchWhenSearchFails and shouldRestartSearchWhenDirectoryRefusesSearch verify an exact times(2) on a create() count that keeps growing.

openam-core/src/test/java/org/forgerock/openam/sm/datalayer/impl/ldap/CTSDJLDAPv3PersistentSearchTest.java:112, :129

Both tests use the single-promise givenDirectoryAcceptsPersistentSearch(searchResult), so every restart's searchAsync returns the already-failed promise. thenOnException then runs the handler inline and schedules the next restart one retryInterval later, and create() counts 1, 2, 3 … until tearDown. timeout(5000).times(2) passes only on a poll that lands while the count is exactly 2, which is a window of about 1 s. If the verifying thread is starved for that long on a loaded runner, it sees 3 and fails at 5 s with no product defect behind it. shouldInformListenersWhenSearchIsReestablished already avoids this with a second, pending promise.

verify(mockConnectionFactory, timeout(5000).atLeast(2)).create();