From 1887e0bdbe3b0d5fdd084eedae798cf18c88c8d5 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 09:16:51 +0300 Subject: [PATCH 1/5] [#925] Keep asking for a session restart until it has run The recovery from a failed replay takes the request before it runs the restart, and restartSession() stops the session in its first synchronized block and starts it again in the second. Anything thrown in between left the request cleared and the session stopped, and nothing asked for it again: no change is delivered over a session which is down, so no replay fails and no thread comes back to the recovery. The domain sat out of the topology - the change it could not replay still owned by the replication server, its ServerState stopped behind it - until the server was restarted or a configuration change happened to restart the service. The throw is narrow but not theoretical: enableService() is the one call on that road nothing guards, while the reconnect loop of the broker wraps the same connectAsDataServer() in a catch, so every failure that path is written to survive is fatal to this one. The request is a SessionRestartRequests now. It is still taken before the restart runs, as it must be - a change released while a restart is under way is not one that restart asks for - and it is given back when the restart could not run, with the backoff whatever it was asked for with, since a session which can not be started is what that wait exists for. The wait belongs to the request rather than to the thread which runs it: a request made by a replay thread on its way out is owed none, one made by a failing replay is owed one, and either thread can end up running the other's request. Nothing else runs a request which is left standing, so the state checkpointer of the domain does, once a second: it is the one thread a domain has for as long as it is up, whatever its session is doing. That also lets abandonReplay() ask for the restart rather than run it - the threads of the pool are stopped one after the other and joined, so a live num-update-replay-threads change cost one stop and reconnect per thread which was replaying a change, none of them waiting, while the configuration change waited for all of them - and it runs the bare hand-back of issue #922, which asked for a restart nothing was going to run. The backoff is waited on a monitor rather than slept through, and a domain which is going away or is being disabled wakes it: the thread which holds it is a replay thread the shutdown of the pool joins, or the checkpointer the shutdown of the domain waits for. --- .../plugin/LDAPReplicationDomain.java | 271 +++++++++++++----- .../plugin/SessionRestartRequests.java | 116 ++++++++ .../opends/messages/replication.properties | 4 + .../replication/UpdateOperationTest.java | 74 +++++ .../plugin/SessionRestartRequestsTest.java | 111 +++++++ 5 files changed, 508 insertions(+), 68 deletions(-) create mode 100644 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index f8c6fcc0a7..6f2a134ba6 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -117,6 +117,7 @@ import org.opends.server.replication.common.ServerState; import org.opends.server.replication.common.ServerStatus; import org.opends.server.replication.common.StatusMachineEvent; +import org.opends.server.replication.plugin.SessionRestartRequests.SessionRestart; import org.opends.server.replication.protocol.AddContext; import org.opends.server.replication.protocol.AddMsg; import org.opends.server.replication.protocol.DeleteContext; @@ -354,12 +355,23 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { /** Set while a replay thread is restarting the session after a failed replay. */ private final AtomicBoolean replayFailureRecovery = new AtomicBoolean(); /** - * Set when a change whose replay failed has to be delivered again, and cleared by the - * replay thread which restarts the session for it. A change released while the session - * was being restarted has to be asked for over yet another session: the delivery this - * one makes is turned down as a duplicate while a replay thread still owns it. + * The session restart a change whose replay failed is waiting for, asked for by the + * thread which released the change and taken by the thread which runs it. A change + * released while the session was being restarted has to be asked for over yet another + * session: the delivery this one makes is turned down as a duplicate while a replay + * thread still owns it. + *

+ * The request outlives the thread which made it and the restart which could not run it, + * so that no change is left waiting for a delivery nobody asks for: the restart is run + * by whichever thread takes the request, and by {@link ServerStateFlush} when no thread + * comes back for it. + */ + private final SessionRestartRequests sessionRestarts = new SessionRestartRequests(); + /** + * Woken when a session restart which is sitting through its backoff has nothing left to + * wait for: the domain it would start a session for is going away or is being disabled. */ - private final AtomicBoolean sessionRestartRequested = new AtomicBoolean(); + private final Object sessionRestartBackoff = new Object(); /** * How many times in a row the session was restarted without a change being replayed in * between. The backoff is computed from this rather than from the failures of the @@ -368,6 +380,13 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { * shortest for as long as the outage lasts. */ private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger(); + /** + * How many of the next session restarts must fail rather than start the session again. + *

+ * Only there for the tests, which have no other way to fail a restart: see {@link + * #failNextSessionRestarts(int)}. It is zero in production, where nothing ever sets it. + */ + private final AtomicInteger sessionRestartFailuresToInject = new AtomicInteger(); /** * Set by {@link #restartService()} when it left the session of this domain alone, so * that the configuration change which asked for the restart can say so. @@ -632,7 +651,8 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { /** * The thread that periodically saves the ServerState of this - * LDAPReplicationDomain in the database. + * LDAPReplicationDomain in the database, and runs the session restart this domain has + * been asked for when the threads which asked for it are gone or could not run it. */ private class ServerStateFlush extends DirectoryThread { @@ -656,6 +676,12 @@ public void run() { saveState(); } + /* + * Run outside the monitor above, as the save is: a session restart holds the + * backoff a failing backend is owed, and shutdown() takes that monitor to wake + * this thread up. + */ + runPendingSessionRestart(); } catch (InterruptedException e) { @@ -2572,6 +2598,13 @@ public void shutdown() { flushThread.notifyAll(); } + /* + * A session restart which is sitting through its backoff has nothing left to wait + * for either, and this thread waits for the checkpointer below: a restart left to + * wait out a backend which is not what is going away would hold this shutdown for + * as long as the backoff it had climbed to. + */ + wakeSessionRestartBackoff(); DirectoryServer.deregisterAlertGenerator(this); getServerContext().getBackendConfigManager() @@ -2732,14 +2765,14 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * of memory - would leave the change owned by this thread after all. Hand it back * bare, without the failure count that road did not reach, and restart the session * so that it is delivered again: this is the last resort, the throwable is rethrown - * whatever happens here, and the thread this runs on may well be ending on it - so - * a request left for the next recovery of this domain to pick up is a request which - * may never be run. + * whatever happens here, and the thread this runs on may well be ending on it. A + * restart which can not run here leaves its request standing, and the state + * checkpointer of this domain runs it. */ if (owned != null) { remotePendingChanges.replayFailed(owned); - sessionRestartRequested.set(true); + sessionRestarts.request(SessionRestart.NOW); /* * Reported and restarted under guards of their own, and in that order: the report * is the line an operator acts on, and the restart is what has the change @@ -2757,14 +2790,15 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) } try { - runRequestedSessionRestarts(false); + runRequestedSessionRestarts(); } catch (Throwable restartFailure) { /* - * Nothing is left to try: the change is listed, uncommitted and unowned, so any - * later session restart of this domain delivers it again. This goes with the - * throwable which is rethrown below rather than being reported on its own. + * Nothing is left to try here: the change is listed, uncommitted and unowned, + * and the restart which threw has asked for one again, so the session restart + * the state checkpointer runs delivers it again. This goes with the throwable + * which is rethrown below rather than being reported on its own. */ suppress(recoveryFailure, restartFailure); } @@ -3760,74 +3794,58 @@ private boolean recoverFromReplayFailure( * run it: a restart which is already under way may have started before this change * was released, and the delivery it asked for would then have been turned down as a * duplicate of a change a replay thread still owned. + * + * A replay thread which is stopping - the number of them is being changed - asks for + * the restart all the same: nothing else would ask for the change it just released, + * and the ServerState would stay behind it for good. What it asks for is not owed the + * backoff, though: the backend is not what is going away. Neither is what the thread + * an OutOfMemoryError is ending asks for, for the same reason. The wait belongs to the + * request rather than to the thread which runs it, or a hand-back which is owed none + * would spend the wait another request is owed - and the other way around. */ - sessionRestartRequested.set(true); - /* - * A replay thread which is stopping - the number of them is being changed - restarts - * the session all the same: nothing else would ask for the change it just released, - * and the ServerState would stay behind it for good. It does not sit through the - * backoff on its way out, though: the backend is not what is going away. Neither does - * the thread an OutOfMemoryError is ending, for the same reason - and the restart is - * run rather than left to be asked for again, because that thread will not be there to - * run it, and a change nobody asks for again holds this domain's ServerState back. - */ - runRequestedSessionRestarts(!replayThreadShutdown.get() && !outOfMemory); + sessionRestarts.request(replayThreadShutdown.get() || outOfMemory + ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF); + runRequestedSessionRestarts(); return true; } /** * Restarts the session as long as changes which could not be replayed are waiting to be * delivered again. - * - * @param wait whether to leave the backend some time to recover between two restarts */ - private void runRequestedSessionRestarts(boolean wait) + private void runRequestedSessionRestarts() { /* * The outer loop is what makes a request which was made while this thread was giving * up the recovery its own: the thread which made it found the recovery taken and left * it to this one. */ - while (sessionRestartRequested.get() && replayFailureRecovery.compareAndSet(false, true)) + while (sessionRestarts.isPending() && replayFailureRecovery.compareAndSet(false, true)) { try { - while (sessionRestartRequested.getAndSet(false)) + for (SessionRestart restart = sessionRestarts.take(); + restart != SessionRestart.NONE; + restart = sessionRestarts.take()) { - boolean restarted = false; try { - restartSession(wait); - restarted = true; + restartSession(restart == SessionRestart.AFTER_BACKOFF); } - finally + catch (Throwable t) { - if (!restarted) - { - /* - * The request is put back where it was taken from. The flag is read and - * cleared before the restart runs, so a restart which ends abruptly - the - * session is stopped first, and starting it again creates a listener thread, - * which the operating system can refuse - would otherwise leave this domain - * with no session and with nothing left to ask for one. - * - * What a request left standing buys is bounded, and the bound is worth - * stating. Its two readers are the roads out of a failed and of an abandoned - * replay of this domain, and with no listener thread nothing is delivered - * anymore: the replays left to run are the changes already taken off the - * session - the ones waiting in the replay queue, and the ones parked as - * dependencies. One of those failing finds the request standing and runs the - * restart, which starts from a clean state, since disableService() drops the - * listener thread which was never started. Once they are spent, the domain - * stays down until it is disabled and enabled back, or the server is - * restarted. That is said where it can be heard: a refused thread is an - * OutOfMemoryError, and one which leaves recoverFromReplayFailure() or - * abandonReplay() ends the replay thread it is met on, so the uncaught - * exception handler of DirectoryThread writes the line and raises the alert, - * with the start of the listener thread in the trace. - */ - sessionRestartRequested.set(true); - } + /* + * The request is taken before the restart runs, so a restart which could not + * run - the session is stopped first, and starting it again creates a listener + * thread, which the operating system can refuse - has to ask for one again: + * the session has been stopped and was not started back, and nothing else + * would ask - no change is delivered over a session which is down, so no + * replay fails and no thread comes back here. The request is asked for with + * the backoff whatever it was made with, since a session which can not be + * started is the very thing that wait is for. + */ + sessionRestarts.giveBack(SessionRestart.AFTER_BACKOFF); + throw t; } } } @@ -3838,13 +3856,48 @@ private void runRequestedSessionRestarts(boolean wait) } } + /** + * Runs the session restart this domain was asked for and which no thread of its own + * ran. + *

+ * A restart is asked for by the thread which released the change it could not replay, + * and that thread usually runs it. It does not always: a replay thread on its way out + * hands its change back and leaves, and a restart which threw where it starts the + * session again asks for one anew rather than take the request away with it. Nothing + * would run what is left standing - a session which is down delivers no change, so no + * replay fails and no thread comes back for it - and this domain would sit out of the + * topology, with the changes it did not replay owned by the replication server and its + * ServerState stopped behind them. + */ + private void runPendingSessionRestart() + { + if (shutdown.get() || disabled || ieRunning() || !sessionRestarts.isPending()) + { + return; + } + try + { + runRequestedSessionRestarts(); + } + catch (Throwable t) + { + /* + * The restart which threw has asked for one again, so this thread runs it again + * once the backoff has been waited out. It must not end on it: nothing would save + * the ServerState of this domain anymore, and shutdown() waits for this thread. + */ + logger.error(ERR_REPLAY_SESSION_RESTART_FAILED, + getBaseDN(), stackTraceToSingleLineString(t)); + } + } + /** * Gives a change back to the replication server when this replay thread stops before it * could apply it. *

- * The change is not owned by anyone anymore and it was never applied, so the session is - * restarted for it to be delivered again: it is left out of the ServerState, and the - * changes which follow it are held back until it is replayed. + * The change is not owned by anyone anymore and it was never applied, so a session + * restart is asked for to have it delivered again: it is left out of the ServerState, + * and the changes which follow it are held back until it is replayed. * * @param csn the CSN of the change this thread was replaying */ @@ -3863,8 +3916,15 @@ private void abandonReplay(CSN csn) * is started back - one line per change would say otherwise. */ logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN()); - sessionRestartRequested.set(true); - runRequestedSessionRestarts(false); + /* + * Asked for rather than run here. The threads of the pool are stopped one after the + * other and joined, so every one of them which was replaying a change would stop and + * start the session on its way out, one restart per change abandoned and none of them + * waiting - while the configuration change which is stopping them waits for all of + * them. The state checkpointer runs one restart for the lot a moment later, which is + * all the replication server needs to send every change which was handed back. + */ + sessionRestarts.request(SessionRestart.NOW); } /** @@ -3913,6 +3973,7 @@ private void restartSession(boolean wait) */ return; } + failSessionRestartIfATestAskedFor(); enableService(); } } @@ -3926,9 +3987,26 @@ private void restartSession(boolean wait) */ private void waitBeforeSessionRestart(int restarts) { + final long until = monotonicNowInMs() + + Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS); try { - Thread.sleep(Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS)); + /* + * Waited on a monitor rather than slept through, so that a domain which is going + * away or is being disabled is not waited out: the thread which holds this wait is + * a replay thread the shutdown of the pool joins, or the state checkpointer the + * shutdown of the domain waits for. The session it would start back is one the + * domain is stopping anyway. + */ + synchronized (sessionRestartBackoff) + { + for (long left = until - monotonicNowInMs(); + left > 0 && !shutdown.get() && !disabled; + left = until - monotonicNowInMs()) + { + sessionRestartBackoff.wait(left); + } + } } catch (InterruptedException e) { @@ -3942,6 +4020,51 @@ private void waitBeforeSessionRestart(int restarts) } } + /** + * Has the next session restarts of this domain fail where they would start the session + * again. + *

+ * Only there for the tests: nothing asks a restart to fail in production, and what this + * stands in for - anything thrown out of {@link #enableService()}, which stops at no + * failure the callers of {@link ReplicationBroker#start()} report - can not be provoked + * from the outside. + * + * @param failures how many session restarts must fail before one is allowed to run + */ + @VisibleForTesting + public void failNextSessionRestarts(int failures) + { + sessionRestartFailuresToInject.set(failures); + } + + /** + * Returns how many of the session restart failures {@link #failNextSessionRestarts(int)} + * asked for have not been spent yet. + *

+ * Only there for the tests, which read it to tell a restart which failed from one which + * was never run. + * + * @return how many session restarts are still to fail + */ + @VisibleForTesting + public int getSessionRestartFailuresLeft() + { + return sessionRestartFailuresToInject.get(); + } + + /** + * Fails this session restart when a test asked for it, and does nothing at all + * otherwise. + */ + private void failSessionRestartIfATestAskedFor() + { + if (sessionRestartFailuresToInject.getAndUpdate(left -> Math.max(left - 1, 0)) > 0) + { + throw new IllegalStateException("the session of domain " + getBaseDN() + + " could not be started again, as a test asked"); + } + } + /** * Generate a new CSN and insert it in the pending list. * @@ -4710,9 +4833,21 @@ public void disable() * is gone with the pending changes, so a leftover request would have a replay thread * stop and start the session once for a delivery which can not come. */ - sessionRestartRequested.set(false); + sessionRestarts.clear(); consecutiveSessionRestarts.set(0); } + // Woken outside the lock it does not hold: a restart which is waiting has nothing + // left to wait for, the session it would start being one this domain is not serving. + wakeSessionRestartBackoff(); + } + + /** Has a session restart which is sitting through its backoff stop waiting. */ + private void wakeSessionRestartBackoff() + { + synchronized (sessionRestartBackoff) + { + sessionRestartBackoff.notifyAll(); + } } /** @@ -5343,7 +5478,7 @@ protected void importBackend(InputStream input) throws DirectoryException * goes with them: the caller starts the session again from the reloaded state. */ remotePendingChanges.clear(); - sessionRestartRequested.set(false); + sessionRestarts.clear(); consecutiveSessionRestarts.set(0); importingData = false; } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java new file mode 100644 index 0000000000..b91f7c6104 --- /dev/null +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java @@ -0,0 +1,116 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.plugin; + +import java.util.concurrent.atomic.AtomicReference; + +/** + * The session restart a replication domain has been asked for and has not run yet. + *

+ * A change which was released without being replayed is delivered again only over a new + * session, so the thread which released it asks this domain for one. The request is what + * the restart is run from, and it outlives the thread which made it: several threads + * releasing changes at once are answered by one restart, and a restart which could not + * run gives the request back rather than take it away with it - a change nobody asks for + * again holds the ServerState of this domain back for as long as the server is up. + *

+ * The wait a failing backend is owed belongs to the request rather than to the thread + * which runs it: a thread which is not owed one - a replay thread on its way out, where + * the backend is not what is going away - must not spend the wait another thread's + * request was made with. + */ +class SessionRestartRequests +{ + /** + * What a domain has been asked to do with its session, from what is not being asked for + * to what is asked for most insistently: {@link #merge(SessionRestart)} keeps the + * furthest down this list, so that a request is never answered by less than it asked + * for. + */ + enum SessionRestart + { + /** Nothing is being asked for: every request made has been run. */ + NONE, + /** The session is to be restarted as soon as a thread can run it. */ + NOW, + /** + * The session is to be restarted once the backend has been left the time to recover + * which the restarts made in a row have climbed to. + */ + AFTER_BACKOFF; + } + + private final AtomicReference requested = + new AtomicReference<>(SessionRestart.NONE); + + /** + * Asks this domain to restart its session. + * + * @param restart what is being asked for, {@link SessionRestart#NONE} asking for + * nothing + */ + void request(SessionRestart restart) + { + merge(restart); + } + + /** + * Takes the request which is standing, so that the caller runs it. + *

+ * It is taken before the restart is run rather than once it has run: a change released + * while the restart was under way is not one that restart asks for - its delivery would + * have been turned down as a duplicate of a change a replay thread still owned - so the + * request it makes must outlive the restart which was already running. + * + * @return what is being asked for, {@link SessionRestart#NONE} when nothing is + */ + SessionRestart take() + { + return requested.getAndSet(SessionRestart.NONE); + } + + /** + * Asks again for a restart which was taken and could not be run. + * + * @param taken what {@link #take()} returned to the caller which could not run it + */ + void giveBack(SessionRestart taken) + { + merge(taken); + } + + /** Forgets what this domain was asked for, its pending changes being gone with it. */ + void clear() + { + requested.set(SessionRestart.NONE); + } + + /** + * Returns whether a restart is being asked for. + * + * @return {@code true} when a restart has been asked for and not run yet + */ + boolean isPending() + { + return requested.get() != SessionRestart.NONE; + } + + private void merge(SessionRestart restart) + { + requested.accumulateAndGet(restart, + (standing, asked) -> standing.compareTo(asked) >= 0 ? standing : asked); + } +} diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index d28000e54a..8f9e3deeb6 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -679,6 +679,10 @@ WARN_REPLAY_DRAIN_INTERRUPTED_320=Domain "%s" is going down and was interrupted WARN_CHANGELOG_READ_AGAIN_FOR_MISSING_CHANGES_321=Replication server RS(%d) holds changes for domain \ "%s" which never reached the message queue of %s: its state %s is behind the state %s of the \ domain although it is being served from that queue. Reading the changelog again to send them +ERR_REPLAY_SESSION_RESTART_FAILED_316=Could not restart the replication session of domain \ + "%s" for the changes which could not be replayed: %s. The session is left stopped and is \ + started again a moment later; the changes it is being restarted for are not recorded as \ + replayed and are still owned by the replication server ERR_REPLICATION_DOMAIN_CONFIG_CHANGE_FAILED_326=Could not apply a configuration change to the \ replication domain on "%s": %s NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED_327=The configuration change was applied to the \ diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java index 77d0acfdd3..c2a0980fdf 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java @@ -2354,6 +2354,80 @@ public void call() throws Exception } } + /** + * Test case for [Issue 925]: a session restart which could not run is run again, so + * that the change it was asked for is delivered again rather than left waiting for a + * delivery which can not come. + *

+ * The request is taken by the thread which runs the restart before the restart runs - + * a change released while a restart is under way is not one that restart asks for - so + * a restart which throws where it starts the session again used to take the request + * away with it. Nothing asked for it a second time: the session had been stopped and + * was not started back, and the domain stayed out of the topology, with the change + * still owned by the replication server and the ServerState of this replica stopped + * behind it, until the server was restarted. + */ + @Test + public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception + { + testSetUp("aSessionRestartWhichCouldNotRunIsRunAgain"); + logger.error(LocalizableMessage.raw( + "Starting replication test : aSessionRestartWhichCouldNotRunIsRunAgain")); + + final int serverId = 24; + ReplicationBroker broker = + openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); + try + { + CSNGenerator gen = new CSNGenerator(serverId, 0); + + Entry tmp = TestCaseUtils.addEntry( + "dn: uid=user.925," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.925", + "cn: Aaccf Amar", + "sn: Amar"); + String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + final long initialReplayed = getMonitorAttrValue(baseDN, "replayed-updates-ok"); + try + { + /* + * The backend is unavailable for longer than the replay is retried in place, so + * the change is only applied if the session is restarted and the replication + * server delivers it again - and the first restart the domain runs for it fails + * the way a broken enableService() does, leaving the session stopped. + */ + ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse", + ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2); + domain.failNextSessionRestarts(1); + + final CSN csn = gen.newCSN(); + broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + + assertNull(getEntry(tmp.getName(), 60000, false), + "the change was not delivered again after the session restart which failed"); + assertEquals(domain.getSessionRestartFailuresLeft(), 0, + "the restart which was asked to fail never ran, so this test proves nothing"); + assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1, + "the change must be recorded as replayed"); + } + finally + { + domain.failNextSessionRestarts(0); + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + } + } + finally + { + broker.stop(); + } + } + /** * Test case for [Issue 889]: the result code the server puts on an internal error is * configurable and only has to report a failure, so it can be set to one conflict diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java new file mode 100644 index 0000000000..3ea76ad74b --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java @@ -0,0 +1,111 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.plugin; + +import static org.opends.server.replication.plugin.SessionRestartRequests.SessionRestart.*; +import static org.testng.Assert.*; + +import org.opends.server.DirectoryServerTestCase; +import org.testng.annotations.Test; + +/** + * Tests the session restarts a replication domain has been asked for and has not run yet: + * a request is answered once, it is not lost by the thread which took it and could not + * run it, and the backoff a failing backend is owed is not dropped by a request which is + * not owed one. + */ +@SuppressWarnings("javadoc") +@Test(groups = { "precommit", "replication" }, sequential = true) +public class SessionRestartRequestsTest extends DirectoryServerTestCase +{ + @Test + public void nothingIsAskedForBeforeAnythingAsks() + { + final SessionRestartRequests requests = new SessionRestartRequests(); + + assertFalse(requests.isPending(), "a domain nobody asked anything of has nothing to run"); + assertEquals(requests.take(), NONE); + } + + @Test + public void aRequestIsAnsweredByOneRestart() + { + final SessionRestartRequests requests = new SessionRestartRequests(); + + requests.request(NOW); + + assertTrue(requests.isPending()); + assertEquals(requests.take(), NOW); + assertFalse(requests.isPending(), "the request has been taken by a thread which runs it"); + assertEquals(requests.take(), NONE); + } + + @Test + public void theRestartWhichIsOwedTheBackoffWinsWhicheverOrderTheyComeIn() + { + final SessionRestartRequests backoffFirst = new SessionRestartRequests(); + backoffFirst.request(AFTER_BACKOFF); + backoffFirst.request(NOW); + + final SessionRestartRequests backoffLast = new SessionRestartRequests(); + backoffLast.request(NOW); + backoffLast.request(AFTER_BACKOFF); + + assertEquals(backoffFirst.take(), AFTER_BACKOFF, + "a thread which is not owed the backoff must not spend the one another thread is owed"); + assertEquals(backoffLast.take(), AFTER_BACKOFF); + } + + @Test + public void aRestartWhichCouldNotRunIsAskedForAgain() + { + final SessionRestartRequests requests = new SessionRestartRequests(); + requests.request(AFTER_BACKOFF); + + final SessionRestartRequests.SessionRestart taken = requests.take(); + requests.giveBack(taken); + + assertTrue(requests.isPending(), "a restart which did not run is still being asked for"); + assertEquals(requests.take(), AFTER_BACKOFF); + } + + @Test + public void aRestartGivenBackDoesNotUndoTheOneAskedForMeanwhile() + { + final SessionRestartRequests requests = new SessionRestartRequests(); + requests.request(NOW); + + final SessionRestartRequests.SessionRestart taken = requests.take(); + requests.request(AFTER_BACKOFF); + requests.giveBack(taken); + + assertEquals(requests.take(), AFTER_BACKOFF, + "the request which arrived while the restart was running keeps its backoff"); + } + + @Test + public void aDomainWhichIsDisabledForgetsWhatItWasAskedFor() + { + final SessionRestartRequests requests = new SessionRestartRequests(); + requests.request(AFTER_BACKOFF); + + requests.clear(); + + assertFalse(requests.isPending(), + "the change the restart was asked for is gone with the pending changes"); + assertEquals(requests.take(), NONE); + } +} From 618361ef7b5645002cfd39254c861abbe6f0a4e0 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 13:53:45 +0300 Subject: [PATCH 2/5] [#925] Take an ordinal in replication.properties nothing else claims --- .../src/messages/org/opends/messages/replication.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index 8f9e3deeb6..075de4cb4b 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -679,7 +679,7 @@ WARN_REPLAY_DRAIN_INTERRUPTED_320=Domain "%s" is going down and was interrupted WARN_CHANGELOG_READ_AGAIN_FOR_MISSING_CHANGES_321=Replication server RS(%d) holds changes for domain \ "%s" which never reached the message queue of %s: its state %s is behind the state %s of the \ domain although it is being served from that queue. Reading the changelog again to send them -ERR_REPLAY_SESSION_RESTART_FAILED_316=Could not restart the replication session of domain \ +ERR_REPLAY_SESSION_RESTART_FAILED_325=Could not restart the replication session of domain \ "%s" for the changes which could not be replayed: %s. The session is left stopped and is \ started again a moment later; the changes it is being restarted for are not recorded as \ replayed and are still owned by the replication server From 1f25a96d41b9d241a390ca8d1ca6330640615755 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 15 Sep 2026 10:29:11 +0300 Subject: [PATCH 3/5] [#925] Pin the checkpointer's own failure, say why its restart waits out a total update, and wake its backoff Review round 1 of #981. The catch of the state checkpointer was entered by no test: the one injected failure was spent by the replay thread's restart, and the checkpointer's ran. aSessionRestartWhichCouldNotRunIsRunAgain now fails two restarts, and asserts that the checkpointer reports the one which threw on it exactly once before the change is delivered; a catch which rethrows is red. The report inside that catch is guarded the way the give-back's report in replay() is: on the road out of a JVM which has refused an allocation, building the line is a throw of its own, and it would end the thread shutdown() waits for. The ieRunning() gate of runPendingSessionRestart() is said out loud. A restart stops the session a total update runs over, in either direction: an import into this replica ends on the entries which had arrived - and ownsItsSession() does not see an online total update at all, since preBackendImport() keeps this domain's own backend events from disabling it - and an export is given up by exportLDIFEntry() on the broker stop. The checkpointer can wait: the request stands until the context is released. enable() clears the holder as disable() does. A request made between a replay thread's read of the flag and disable()'s clear survived the disabled span, and the checkpointer restarted the session enable() had just started, within the second, for a change gone with the pending changes. giveBack() takes what the caller asks for again, not what take() returned, and says so; a NOW take given back as AFTER_BACKOFF is pinned. SessionRestartBackoffTest pins the monitor wait and both its wakes on a domain of its own: shutdown() during the backoff returns in a fraction of it, and after disable() / enable() the checkpointer saves a change within its next tick. Each wake is caught by its own case. --- .../plugin/LDAPReplicationDomain.java | 40 ++- .../plugin/SessionRestartRequests.java | 20 +- .../replication/UpdateOperationTest.java | 36 ++- .../plugin/SessionRestartBackoffTest.java | 286 ++++++++++++++++++ .../plugin/SessionRestartRequestsTest.java | 14 + 5 files changed, 381 insertions(+), 15 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 6f2a134ba6..dcfe4804a1 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -3868,6 +3868,19 @@ private void runRequestedSessionRestarts() * replay fails and no thread comes back for it - and this domain would sit out of the * topology, with the changes it did not replay owned by the replication server and its * ServerState stopped behind them. + *

+ * Not run while a total update is being processed, in either direction: a restart stops + * the session the total update runs over. An import into this replica reads its entries + * from that session and would end on the ones which had arrived - and {@code disabled} + * does not say an import is running, since {@code preBackendImport()} keeps the backend + * events this domain is the cause of from disabling it. An export from this replica + * publishes its entries over it, and {@code exportLDIFEntry()} gives the export up as + * {@code ERR_INIT_RS_DISCONNECTION_DURING_EXPORT} once the broker has been stopped + * under it, which leaves the replica it was initializing to be initialized again. This + * thread is the one which can afford to wait: the request stays standing, and it comes + * back here once a second, so the restart is run as soon as the total update is over. + * The change the restart was asked for waits for as long as the total update takes, and + * the ServerState with it; the replay of this domain keeps running in the meantime. */ private void runPendingSessionRestart() { @@ -3885,9 +3898,22 @@ private void runPendingSessionRestart() * The restart which threw has asked for one again, so this thread runs it again * once the backoff has been waited out. It must not end on it: nothing would save * the ServerState of this domain anymore, and shutdown() waits for this thread. + * + * The report is guarded on its own, the way the report of a give-back which failed + * is: building the line walks the stack of the throwable, and on the road out of a + * JVM which has just refused an allocation that is a throw of its own, which would + * end this thread all the same. The restart is asked for again already, and the next + * run of it reports again if it fails again. */ - logger.error(ERR_REPLAY_SESSION_RESTART_FAILED, - getBaseDN(), stackTraceToSingleLineString(t)); + try + { + logger.error(ERR_REPLAY_SESSION_RESTART_FAILED, + getBaseDN(), stackTraceToSingleLineString(t)); + } + catch (Throwable reportFailure) + { + // Nothing is left to say it with, and the report must not end this thread. + } } } @@ -4965,6 +4991,16 @@ public void enable() { synchronized (serviceStateLock) { + /* + * Cleared here as well as by disable(): a request made in the window between a + * replay thread's read of the flag and disable()'s own clear - abandonReplay() reads + * it, then logs, then asks - survives the whole of the disabled span, and the state + * checkpointer would restart the session started below within the second for a + * change which is gone with the pending changes. Every request standing here is that + * one: the domain has been disabled since anything could ask, and the session started + * below asks for everything the ServerState loaded below does not cover. + */ + sessionRestarts.clear(); try { loadDataState(); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java index b91f7c6104..a854b5aab4 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java @@ -28,9 +28,11 @@ * again holds the ServerState of this domain back for as long as the server is up. *

* The wait a failing backend is owed belongs to the request rather than to the thread - * which runs it: a thread which is not owed one - a replay thread on its way out, where - * the backend is not what is going away - must not spend the wait another thread's - * request was made with. + * which runs it. A replay which failed asks for the restart with the wait, and a replay + * thread on its way out asks for it without - the backend is not what is going away - and + * either thread can end up running what the other asked for: the one which is owed no + * wait must not spend the wait the other's request was made with, and the one which is + * owed the wait must not have its request run without it, whichever of the two runs it. */ class SessionRestartRequests { @@ -84,12 +86,18 @@ SessionRestart take() /** * Asks again for a restart which was taken and could not be run. + *

+ * What is asked for again is not what {@link #take()} returned: the caller asks for the + * restart with the backoff whether or not the request it took was made with one, since + * a session which could not be started is the very thing that wait is for. A request + * made while the restart was running is not undone by it - the two are merged, and the + * one which asks for more wins. * - * @param taken what {@link #take()} returned to the caller which could not run it + * @param restart what the caller which could not run the restart asks for again */ - void giveBack(SessionRestart taken) + void giveBack(SessionRestart restart) { - merge(taken); + merge(restart); } /** Forgets what this domain was asked for, its pending changes being gone with it. */ diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java index c2a0980fdf..410544b010 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java @@ -2366,6 +2366,13 @@ public void call() throws Exception * was not started back, and the domain stayed out of the topology, with the change * still owned by the replication server and the ServerState of this replica stopped * behind it, until the server was restarted. + *

+ * Two restarts fail rather than one, so that both threads which run one meet a failure: + * the first is spent by the replay thread which released the change, and the request it + * gives back is run by the state checkpointer, whose restart is the second to fail. The + * checkpointer reports that one and runs the request again a moment later - which is + * what delivers the change - so a checkpointer which ended on the failure instead would + * leave the change where the replay thread left it. */ @Test public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception @@ -2399,20 +2406,35 @@ public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception /* * The backend is unavailable for longer than the replay is retried in place, so * the change is only applied if the session is restarted and the replication - * server delivers it again - and the first restart the domain runs for it fails - * the way a broken enableService() does, leaving the session stopped. + * server delivers it again - and the first two restarts the domain runs for it + * fail the way a broken enableService() does, leaving the session stopped: the one + * the replay thread runs, and the one the state checkpointer runs for it. */ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse", ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2); - domain.failNextSessionRestarts(1); + domain.failNextSessionRestarts(2); final CSN csn = gen.newCSN(); - broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + final List records = errorLogRecordsOf(() -> { + broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); - assertNull(getEntry(tmp.getName(), 60000, false), - "the change was not delivered again after the session restart which failed"); + assertNull(getEntry(tmp.getName(), 60000, false), + "the change was not delivered again after the session restarts which failed"); + return null; + }); assertEquals(domain.getSessionRestartFailuresLeft(), 0, - "the restart which was asked to fail never ran, so this test proves nothing"); + "the restarts which were asked to fail never ran, so this test proves nothing"); + /* + * The failure the replay thread meets is reported by the replay thread's own catch, + * as an exception replaying a message; only the checkpointer says this, once per + * restart which threw on it, and one did. The replay thread's request is its own to + * run unless the checkpointer's tick lands in the instants between the request + * being made and being taken, which is what a count of two here would say. + */ + assertEquals(countRecordsOf(records, + "Could not restart the replication session of domain \"" + baseDN + "\""), 1, + "the state checkpointer reports the restart which threw on it, once, and runs" + + " it again: " + records); assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1, "the change must be recorded as replayed"); } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java new file mode 100644 index 0000000000..2122a5bd46 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java @@ -0,0 +1,286 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.plugin; + +import static java.util.concurrent.TimeUnit.*; +import static org.forgerock.opendj.ldap.ModificationType.*; +import static org.opends.server.TestCaseUtils.*; +import static org.opends.server.protocols.internal.InternalClientConnection.*; +import static org.opends.server.replication.plugin.LDAPReplicationDomain.*; +import static org.testng.Assert.*; + +import java.util.SortedSet; +import java.util.TreeSet; + +import org.forgerock.opendj.ldap.DN; +import org.forgerock.opendj.ldap.ResultCode; +import org.opends.server.TestCaseUtils; +import org.opends.server.core.ModifyOperation; +import org.opends.server.plugins.ShortCircuitPlugin; +import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.common.CSNGenerator; +import org.opends.server.replication.common.ServerState; +import org.opends.server.replication.protocol.DeleteMsg; +import org.opends.server.replication.protocol.OperationContext; +import org.opends.server.replication.server.ReplServerFakeConfiguration; +import org.opends.server.replication.server.ReplicationServer; +import org.opends.server.replication.service.ReplicationBroker; +import org.opends.server.types.Entry; +import org.opends.server.types.OperationType; +import org.testng.annotations.Test; + +/** + * Tests that a domain which is going down, or is being disabled, does not wait out the + * backoff of the session restart its state checkpointer is sitting through: the session + * that restart would start back is one the domain is stopping anyway, so the wait is on a + * monitor which {@code shutdown()} and {@code disable()} wake, rather than slept through. + *

+ * The checkpointer is the thread put through the wait here because it is the one whose + * wait can be seen from a test: {@code shutdown()} waits for it, and it is what saves the + * ServerState once the domain is enabled back. A replay thread sitting through the same + * wait is one thread of a shared pool, and its absence shows nowhere. + */ +@SuppressWarnings("javadoc") +public class SessionRestartBackoffTest extends ReplicationTestCase +{ + private static final int RS_ID = 602; + private static final int DS_ID = 1; + private static final int BROKER_ID = 2; + private static final int GROUP_ID = 1; + + /** + * How long after the second restart has failed this test acts on the domain. + *

+ * The checkpointer reports that failure and goes back to its wait of one second, takes + * the request back once that second is out, and then holds the backoff of a third + * restart in a row: three seconds. Acting a fifth of a second into those three lands + * well inside them, and a wait which is not woken has most of them left to hold whoever + * is waiting for the checkpointer. A machine slow enough to push the checkpointer's tick + * past this delay has this test act before the backoff begins, which a wait that is + * slept through survives - the test then proves less, but reports nothing false. + */ + private static final long INTO_THE_BACKOFF_IN_MS = 1200; + + /** + * How long {@code shutdown()} may take when the backoff is woken: it drains a replay + * which is not running, stops a session which is already stopped, and waits for the + * checkpointer to save the ServerState once more. A checkpointer which sleeps the + * backoff through holds it for the seconds the backoff has left instead. + */ + private static final long SHUTDOWN_BOUND_IN_MS = 1500; + + /** + * How long the checkpointer may take to save a change made once the domain is enabled + * back: its next tick, at most a second away, plus the save. One which was left asleep + * in the backoff {@code disable()} cut saves nothing until the backoff is out. + */ + private static final long CHECKPOINT_BOUND_IN_MS = 2500; + + @Test + public void aDomainWhichIsShutDownDoesNotWaitOutTheBackoffOfItsSessionRestart() + throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + ReplicationServer replicationServer = null; + LDAPReplicationDomain domain = null; + ReplicationBroker broker = null; + try + { + final int rsPort = TestCaseUtils.findFreePort(); + replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestShutdownDb"); + domain = startDomain(baseDN, rsPort); + broker = openReplicationSession(baseDN, BROKER_ID, 100, rsPort, 1000); + + leaveTheCheckpointerInTheBackoff(domain, broker, "user.925.shutdown"); + + final long started = System.nanoTime(); + domain.shutdown(); + final long tookMs = NANOSECONDS.toMillis(System.nanoTime() - started); + + assertTrue(tookMs < SHUTDOWN_BOUND_IN_MS, "shutdown() waited " + tookMs + + " ms: it sat out the backoff of the session restart the state checkpointer" + + " was waiting through, for a session the domain was stopping anyway"); + } + finally + { + release(domain, broker); + if (domain != null) + { + MultimasterReplication.deleteDomain(baseDN); + } + remove(replicationServer); + } + } + + @Test + public void aDomainWhichIsDisabledDoesNotWaitOutTheBackoffOfItsSessionRestart() + throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + ReplicationServer replicationServer = null; + LDAPReplicationDomain domain = null; + ReplicationBroker broker = null; + try + { + final int rsPort = TestCaseUtils.findFreePort(); + replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestDisableDb"); + domain = startDomain(baseDN, rsPort); + broker = openReplicationSession(baseDN, BROKER_ID, 100, rsPort, 1000); + + final DN entryDN = leaveTheCheckpointerInTheBackoff(domain, broker, "user.925.disable"); + + // The data of this domain is about to be replaced, then has been. + domain.disable(); + domain.enable(); + assertTrue(domain.isConnected(), "the domain did not come back up once enabled"); + + /* + * A change of this replica's own, for the checkpointer to save: nothing but the + * checkpointer writes the ServerState while the domain is up, so the change reaching + * the saved state says that the checkpointer is ticking rather than still asleep in + * the backoff disable() cut. It is read as the CSN the change was given, since the + * change the restart was asked for is delivered again over the new session and is + * recorded in that state as well. + */ + final ModifyOperation modify = getRootConnection().processModify( + modifyRequest(baseDN, REPLACE, "description", "the checkpointer is ticking")); + assertEquals(modify.getResultCode(), ResultCode.SUCCESS, + modify.getAdditionalLogItems().toString()); + final CSN csn = OperationContext.getCSN(modify); + assertNotNull(csn, "the change of this replica's own was given no CSN"); + + final long deadline = System.nanoTime() + MILLISECONDS.toNanos(CHECKPOINT_BOUND_IN_MS); + while (!persistedServerState(baseDN).cover(csn)) + { + assertTrue(System.nanoTime() < deadline, "the state checkpointer did not save the" + + " ServerState within " + CHECKPOINT_BOUND_IN_MS + " ms of the domain being" + + " enabled back: it was left asleep in the backoff of the session restart" + + " disable() cut"); + Thread.sleep(50); + } + + // The change the restart was asked for is delivered over the session enable() started. + assertNull(getEntry(entryDN, 30000, false), + "the change was not delivered again over the session the domain was enabled with"); + } + finally + { + release(domain, broker); + if (domain != null) + { + MultimasterReplication.deleteDomain(baseDN); + } + remove(replicationServer); + } + } + + /** + * Leaves the state checkpointer of the provided domain sitting through the backoff of + * a session restart, with the session stopped. + *

+ * A change whose replay fails for longer than it is retried in place has the session + * restarted for it to be delivered again, and the first two restarts fail where they + * would start the session back: the one the replay thread runs, and the one the + * checkpointer runs for the request it gave back. The checkpointer runs the third, and + * it is the checkpointer whichever thread the second failure fell to, since the replay + * thread leaves on the failure it meets and no delivery brings one back over a session + * which is down. That third restart is owed the backoff of three restarts in a row, and + * this returns once the checkpointer is inside it. + * + * @return the DN of the entry the change which is waiting to be delivered again deletes + */ + private DN leaveTheCheckpointerInTheBackoff( + LDAPReplicationDomain domain, ReplicationBroker broker, String uid) throws Exception + { + final Entry entry = TestCaseUtils.addEntry( + "dn: uid=" + uid + "," + domain.getBaseDN(), + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: " + uid, + "cn: Aaccf Amar", + "sn: Amar"); + final String uuid = getEntry(entry.getName(), 1, true).parseAttribute("entryuuid").asString(); + + /* + * Unavailable for the attempts in place of the first delivery and two more, so that + * the delivery a session enabled back brings ends in the change being applied. + */ + ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse", + ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2); + domain.failNextSessionRestarts(2); + + final CSNGenerator gen = new CSNGenerator(BROKER_ID, 0); + broker.publish(new DeleteMsg(entry.getName(), gen.newCSN(), uuid)); + + final long deadline = System.nanoTime() + SECONDS.toNanos(30); + while (domain.getSessionRestartFailuresLeft() > 0) + { + assertTrue(System.nanoTime() < deadline, + "the two session restarts which were asked to fail did not both run within 30 s"); + Thread.sleep(50); + } + Thread.sleep(INTO_THE_BACKOFF_IN_MS); + assertFalse(domain.isConnected(), + "the session was started back before this test could act on the domain"); + return entry.getName(); + } + + /** Lets go of what {@link #leaveTheCheckpointerInTheBackoff} set up. */ + private void release(LDAPReplicationDomain domain, ReplicationBroker broker) + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + if (domain != null) + { + domain.failNextSessionRestarts(0); + } + if (broker != null) + { + broker.stop(); + } + } + + private ServerState persistedServerState(DN baseDN) throws Exception + { + final ServerState persisted = new ServerState(); + for (String value : getEntry(baseDN, 1, true).parseAttribute("ds-sync-state").asSetOfString()) + { + persisted.update(new CSN(value)); + } + return persisted; + } + + private LDAPReplicationDomain startDomain(DN baseDN, int rsPort) throws Exception + { + final SortedSet replServers = new TreeSet<>(); + replServers.add("localhost:" + rsPort); + final DomainFakeCfg domainCfg = new DomainFakeCfg(baseDN, DS_ID, replServers, GROUP_ID); + domainCfg.setHeartbeatInterval(100000); + final LDAPReplicationDomain domain = MultimasterReplication.createNewDomain(domainCfg); + domain.start(); + assertTrue(domain.isConnected(), "the domain did not connect to its replication server"); + return domain; + } + + private ReplicationServer createReplicationServer(int rsPort, String dbDir) throws Exception + { + final ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration( + rsPort, dbDir, 0, RS_ID, 0, 100, new TreeSet(), GROUP_ID, 1000, 5000); + return new ReplicationServer(conf); + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java index 3ea76ad74b..fde84a8fe0 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java @@ -82,6 +82,20 @@ public void aRestartWhichCouldNotRunIsAskedForAgain() assertEquals(requests.take(), AFTER_BACKOFF); } + @Test + public void aRestartGivenBackAfterItThrewIsOwedTheBackoff() + { + final SessionRestartRequests requests = new SessionRestartRequests(); + requests.request(NOW); + assertEquals(requests.take(), NOW); + + // What the domain gives back after a restart threw: the backoff, whatever was taken. + requests.giveBack(AFTER_BACKOFF); + + assertEquals(requests.take(), AFTER_BACKOFF, + "a session which could not be started is what the wait exists for"); + } + @Test public void aRestartGivenBackDoesNotUndoTheOneAskedForMeanwhile() { From a7e01d5ab93b2b9216673c39b7baf4528c87522d Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 15 Sep 2026 20:59:47 +0300 Subject: [PATCH 4/5] [#925] Time disable() against the woken backoff, count the wakes, and pin the clears and the give-back level Review round 2. - SessionRestartBackoffTest times disable() as it times shutdown(), against the one bound both wakes share: a checkpointer which sleeps the backoff through while it holds the monitor holds disable() for the rest of it, and the disable case was green under that shape of the sleep mutant. Both shapes are in the mutant table now. - The wake of the backoff is counted as well as flagged, and the count is read where the session is stopped, under serviceStateLock: `disabled` does not stay set, so a checkpointer descheduled between disable()'s notifyAll() and its own re-check for as long as enable() takes would wait out the rest of the backoff for a session it has nothing left to start. - aSessionRestartWhichCouldNotRunIsRunAgain accepts the checkpointer's report twice, on the road its own comment names: its tick landing between the replay thread's request and the CAS. - requestSessionRestart() (@VisibleForTesting) leaves a request standing, and two cases pin the clears in enable() and importBackend() on it - the second in #968's ReplayDuringImportTest, which has the import fixture. - aRestartAskedForWithoutTheBackoffIsGivenBackWithIt pins the level a restart which could not run is given back with, on the abandonReplay() road #941's fixture reaches: a NOW request given back as taken has the session back a second early. - startDomain() asserts nothing, so that a case which fails on its way up still holds the domain to delete, and the cleanup chains its three steps. - The comment on the OOME road of replay() and ERR_REPLAY_GIVE_BACK_FAILED say what happens: the restart is asked for, and the last resort re-runs a given-back request on the same thread, backoff and all. --- .../plugin/LDAPReplicationDomain.java | 74 +++++++- .../opends/messages/replication.properties | 3 +- .../replication/UpdateOperationTest.java | 158 +++++++++++++++++- .../plugin/ReplayDuringImportTest.java | 41 +++++ .../plugin/SessionRestartBackoffTest.java | 137 ++++++++++++--- 5 files changed, 374 insertions(+), 39 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index dcfe4804a1..9b9ae6c276 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -372,6 +372,19 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { * wait for: the domain it would start a session for is going away or is being disabled. */ private final Object sessionRestartBackoff = new Object(); + /** + * How many times {@link #sessionRestartBackoff} has been woken, so that a wake is never + * missed. + *

+ * The wake is given after the flag it stands for is set, and the wait checks the flag + * again once woken - but {@code disabled} does not stay set: an {@link #enable()} which + * runs between the wake and that check puts it back, and the wait would go on for the + * rest of the backoff, with the session {@code enable()} started left to be found gone + * at its end. The count only grows, so a restart which took it where it stopped the + * session sees every wake given since, whether it was waiting yet or not. + */ + @GuardedBy("sessionRestartBackoff") + private long sessionRestartBackoffWakes; /** * How many times in a row the session was restarted without a change being replayed in * between. The backoff is computed from this rather than from the failures of the @@ -2745,8 +2758,12 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * its own road: the change is given back counted, the way every other unwound * replay gives it back, but the line which says it is being asked for again is * not built - that asks the JVM for the memory it has just refused - and the - * session is restarted without sitting through the backoff, since the thread - * which is doing it is on its way out. + * restart is asked for without the backoff, since the thread which runs it is + * on its way out. That holds for the restart as first run: one which throws is + * given back with the backoff, as any restart which could not run is, and the + * last resort below runs what is standing - the given-back request, merged with + * its own - on this same thread, backoff and all. Only a restart which throws + * there as well is left to the state checkpointer. * * A change whose budget is spent is still reported and still raises its alert * on this road: it is the one line which says this replica has diverged, and an @@ -3960,6 +3977,7 @@ private void abandonReplay(CSN csn) private void restartSession(boolean wait) { final long stoppedSession; + final long wakes; synchronized (serviceStateLock) { if (sessionHasAnOwner()) @@ -3970,6 +3988,7 @@ private void restartSession(boolean wait) } disableService(); stoppedSession = getSessionGeneration(); + wakes = sessionRestartBackoffWakes(); } if (wait) { @@ -3979,7 +3998,7 @@ private void restartSession(boolean wait) * is not held under the lock, or a domain being disabled for an import would wait * it out. */ - waitBeforeSessionRestart(consecutiveSessionRestarts.incrementAndGet()); + waitBeforeSessionRestart(consecutiveSessionRestarts.incrementAndGet(), wakes); } synchronized (serviceStateLock) { @@ -4010,8 +4029,11 @@ private void restartSession(boolean wait) * fast as the replication server can send them. * * @param restarts how many times in a row the session was restarted already + * @param wakes how many times the backoff had been woken when the session was stopped, + * so that a wake given since ends the wait whether it found the wait under way + * or not */ - private void waitBeforeSessionRestart(int restarts) + private void waitBeforeSessionRestart(int restarts, long wakes) { final long until = monotonicNowInMs() + Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS); @@ -4023,11 +4045,15 @@ private void waitBeforeSessionRestart(int restarts) * a replay thread the shutdown of the pool joins, or the state checkpointer the * shutdown of the domain waits for. The session it would start back is one the * domain is stopping anyway. + * + * The wake is counted rather than only checked for by its flag: a domain which was + * disabled and enabled back has its flag cleared again, and a wait which read the + * flag only would go on for a session the restart has nothing left to start. */ synchronized (sessionRestartBackoff) { for (long left = until - monotonicNowInMs(); - left > 0 && !shutdown.get() && !disabled; + left > 0 && !shutdown.get() && !disabled && wakes == sessionRestartBackoffWakes; left = until - monotonicNowInMs()) { sessionRestartBackoff.wait(left); @@ -4078,6 +4104,22 @@ public int getSessionRestartFailuresLeft() return sessionRestartFailuresToInject.get(); } + /** + * Asks for a session restart the way a replay thread on its way out does, without + * running it. + *

+ * Only there for the tests, which have no other way to leave a request standing at a + * time of their choosing: the one a replay thread makes is made and run in one go, and + * the requests which stand across a span nothing runs them in - the domain disabled, or + * being imported into - are made in a window between a thread's read of the flag and the + * flag being set, which no test can hit on purpose. + */ + @VisibleForTesting + public void requestSessionRestart() + { + sessionRestarts.request(SessionRestart.NOW); + } + /** * Fails this session restart when a test asked for it, and does nothing at all * otherwise. @@ -4867,15 +4909,35 @@ public void disable() wakeSessionRestartBackoff(); } - /** Has a session restart which is sitting through its backoff stop waiting. */ + /** + * Has a session restart which is sitting through its backoff stop waiting, and one which + * is about to sit through it not wait at all: the wake is counted (see + * {@link #sessionRestartBackoffWakes}). + */ private void wakeSessionRestartBackoff() { synchronized (sessionRestartBackoff) { + sessionRestartBackoffWakes++; sessionRestartBackoff.notifyAll(); } } + /** + * Returns how many times the backoff has been woken so far, for a session restart to + * take under the lock it stops the session under: the wake of a {@link #disable()} which + * has yet to take that lock is then one the restart sees, whether it is waiting by then + * or not, and one which took it already has left the restart nothing to stop. The wake + * of {@link #shutdown()} needs no counting - the flag it stands for never comes back. + */ + private long sessionRestartBackoffWakes() + { + synchronized (sessionRestartBackoff) + { + return sessionRestartBackoffWakes; + } + } + /** * Waits for the replay threads which are applying a change of this domain to be done * with it. diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index 075de4cb4b..c6b8fc1718 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -667,7 +667,8 @@ ERR_ACK_NOT_PUBLISHED_316=Could not complete the delivery of change %s in domain timeout instead ERR_REPLAY_GIVE_BACK_FAILED_317=Could not give change %s of domain "%s" back to the replication \ server after the replay which owned it was unwound: %s. The change has been released without its \ - failure being counted, and the session is being restarted so that the change is delivered again + failure being counted, and a restart of the session is asked for so that the change is delivered \ + again WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \ the replay of one of its changes to finish. A change which reaches the backend from now on \ is not recorded in the ServerState being saved, so the replication server sends it again \ diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java index 410544b010..3d71563e1e 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java @@ -156,6 +156,20 @@ public class UpdateOperationTest extends ReplicationTestCase */ private static final long TEST_REPLAY_DRAIN_TIMEOUT_IN_MS = 2000; + /** + * How long after a session restart threw on the state checkpointer the test which reads + * the level the request was given back with looks at the session. + *

+ * The checkpointer takes the request back on its next tick, a second later, and holds + * the backoff it was given back with - a second, the first time - before it starts the + * session: two seconds after the throw, then. A request given back without the backoff + * has the session up a second after the throw instead. Halfway between the two, so that + * either outcome has half a second to be told apart in. A machine slow enough to push the + * tick past this delay has the test look before the session could be back under either + * level, which proves less and reports nothing false. + */ + private static final long INTO_THE_BACKOFF_IN_MS = 1500; + /** An entry with a entryUUID. */ private Entry personWithUUIDEntry; private Entry personWithSecondUniqueID; @@ -2429,12 +2443,16 @@ public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception * as an exception replaying a message; only the checkpointer says this, once per * restart which threw on it, and one did. The replay thread's request is its own to * run unless the checkpointer's tick lands in the instants between the request - * being made and being taken, which is what a count of two here would say. + * being made and being taken: the checkpointer then spends both failures and says + * this twice, which is the same report on a rarer road rather than a double one. A + * count of none has no road - the replay thread's own failure is never reported as + * this - so the count still says that the checkpointer reports what threw on it. */ - assertEquals(countRecordsOf(records, - "Could not restart the replication session of domain \"" + baseDN + "\""), 1, - "the state checkpointer reports the restart which threw on it, once, and runs" - + " it again: " + records); + final int reported = countRecordsOf(records, + "Could not restart the replication session of domain \"" + baseDN + "\""); + assertTrue(reported == 1 || reported == 2, "the state checkpointer reports the" + + " restart which threw on it once, or twice when its tick took the replay" + + " thread's request as well, and runs it again: " + reported + " in " + records); assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1, "the change must be recorded as replayed"); } @@ -2450,6 +2468,136 @@ public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception } } + /** + * Test case for [Issue 925]: a session restart which was asked for without the backoff + * and could not run is asked for again with it. A replay thread on its way out - the + * number of them is being changed - hands its change back and asks for the restart + * without the wait, since the backend is not what is going away; the state checkpointer + * runs what it asked for. A session which can not be started is what the wait exists + * for, though, and a request given back as it was made would have the checkpointer stop + * and fail to start the session once a second, for as long as the failure lasts. + *

+ * The restart which fails is the checkpointer's, so the two levels are told apart by + * when the session is back: on the checkpointer's next tick, a second after the throw, + * when the request was given back as it was made, and one backoff later when it was + * given back with the wait it is owed. Halfway between the two the session is still down + * under the second and up under the first. + */ + @Test + public void aRestartAskedForWithoutTheBackoffIsGivenBackWithIt() throws Exception + { + testSetUp("aRestartAskedForWithoutTheBackoffIsGivenBackWithIt"); + logger.error(LocalizableMessage.raw( + "Starting replication test : aRestartAskedForWithoutTheBackoffIsGivenBackWithIt")); + + final int serverId = 25; + ReplicationBroker broker = + openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); + try + { + CSNGenerator gen = new CSNGenerator(serverId, 0); + + Entry tmp = TestCaseUtils.addEntry( + "dn: uid=user.925.backoff," + baseDN, + "objectClass: top", + "objectClass: person", + "objectClass: organizationalPerson", + "objectClass: inetOrgPerson", + "uid: user.925.backoff", + "cn: Aaccf Amar", + "sn: Amar"); + String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString(); + + final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); + + /* + * The replayed delete is held where it is, so that the replay thread is stopped + * while it owns the change: see aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain + * for the fixture. + */ + final CSN csn = gen.newCSN(); + final ParkedReplay parked = ShortCircuitPlugin.parkReplayedOperations( + OperationType.DELETE, "PreParse", op -> csn.equals(OperationContext.getCSN(op))); + final AtomicReference reconfigurationFailure = new AtomicReference<>(); + Thread reconfiguration = null; + boolean reconfigurationFinished = true; + try + { + broker.publish(new DeleteMsg(tmp.getName(), csn, uuid)); + parked.awaitParked(60, SECONDS); + + reconfiguration = startReplayThreadReconfiguration(2, reconfigurationFailure); + awaitStoppingTheReplayThreads(reconfiguration, reconfigurationFailure); + + /* + * The thread which holds the change has been asked to stop: let go of it, and it + * hands the change back and asks for the restart on its way out. The restart is + * the checkpointer's to run, and it fails where it would start the session again. + */ + domain.failNextSessionRestarts(1); + parked.release(ResultCode.UNAVAILABLE.intValue()); + + final long deadline = System.nanoTime() + SECONDS.toNanos(30); + while (domain.getSessionRestartFailuresLeft() > 0) + { + assertTrue(System.nanoTime() < deadline, + "the session restart which was asked to fail did not run within 30 s"); + Thread.sleep(50); + } + Thread.sleep(INTO_THE_BACKOFF_IN_MS); + assertFalse(domain.isConnected(), "the session was started back within " + + INTO_THE_BACKOFF_IN_MS + " ms of the restart which threw: the request a" + + " replay thread on its way out made without the backoff was given back" + + " without it, and the checkpointer ran it again on its next tick rather" + + " than one backoff later"); + + /* + * Stop parking before the session is back: the delivery it brings is the change + * being delivered again, and it is to be applied. The replication server owns the + * change the thread handed back, so the session started once the backoff is out + * is what has it delivered. + */ + parked.deregister(); + reconfiguration.join(SECONDS.toMillis(60)); + assertFalse(reconfiguration.isAlive(), + "the replay threads were reconfigured, but applyConfigurationChange never returned"); + if (reconfigurationFailure.get() != null) + { + throw new AssertionError("the replay threads could not be reconfigured", + reconfigurationFailure.get()); + } + assertNull(getEntry(tmp.getName(), 60000, false), + "the change was not delivered again once the backoff was out"); + } + finally + { + /* + * Whatever happened above: no failure may be left to inject, no replay thread of + * this server may be left parked, and the reconfiguration has to be over before + * the pool is restored - see aChangeAStoppedReplayThreadHeldIsGivenBackAndDeliveredAgain + * for why the pool is otherwise left alone. + */ + domain.failNextSessionRestarts(0); + parked.deregister(); + if (reconfiguration != null) + { + reconfiguration.join(SECONDS.toMillis(60)); + reconfigurationFinished = !reconfiguration.isAlive(); + } + if (reconfigurationFinished) + { + setNumberOfReplayThreads(null); + } + } + assertTrue(reconfigurationFinished, + "the replay thread reconfiguration never finished; the pool was left alone"); + } + finally + { + broker.stop(); + } + } + /** * Test case for [Issue 889]: the result code the server puts on an internal error is * configurable and only has to report a failure, so it can be set to one conflict diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java index 3bedecb489..3dc590f719 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java @@ -63,6 +63,9 @@ * one thing it must not do is stop the session the import is reading (issue #956). The same * holds from the moment the total update is asked for: the answer to the request arrives * over that session, so a replay which fails while it is on its way must not restart it. + * A restart asked for before the total update took the session, and left standing for the + * length of it, is not run once it is over either: the change it was asked for is gone with + * the ServerState the import replaced. *

* The exporter is a broker of this test, so that the test says when the entries arrive: the * change is replayed while the import is waiting for them - or, for the request, while the @@ -284,6 +287,44 @@ public void aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver() throws Exceptio } } + /** + * A session restart which stood while the import ran was asked for by a replay thread + * for a change given back before the total update owned the session, and that change is + * forgotten with the pending changes when the imported data replaces the ServerState: + * the session started back at the end of the import asks for everything the imported + * state does not cover. Run, the request would stop that session once for a delivery + * which can not come. The request is made here by hand, in the place of one made + * between a replay thread's read of the owner and the import claiming the session. + *

+ * The restart is the state checkpointer's to run, within its first tick after the total + * update has released the session, so the pin is that the failure it would meet is never + * spent: a restart which ran would have spent it, and would have left the session it + * stopped down. + */ + @Test(timeOut = 120_000) + public void aRequestWhichStoodWhileTheImportRanIsNotRunOnceItIsOver() throws Exception + { + final String[] exported = exportedEntries(); + startImportInto(exported.length); + domain.requestSessionRestart(); + domain.failNextSessionRestarts(1); + try + { + finishImport(exported); + + // Two ticks of the checkpointer: a request standing when the import ends is run on the first. + Thread.sleep(2000); + assertEquals(domain.getSessionRestartFailuresLeft(), 1, "the request which stood while" + + " the import ran was run against the session started back at its end"); + assertTrue(domain.isConnected(), "the session started back at the end of the import" + + " was stopped for a request made before it"); + } + finally + { + domain.failNextSessionRestarts(0); + } + } + /** * Has the exporter start a total update into this replica, and returns once the backend * of the domain is deregistered for it: from then on the import is reading the session, diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java index 2122a5bd46..225e2a76db 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java @@ -48,6 +48,8 @@ * backoff of the session restart its state checkpointer is sitting through: the session * that restart would start back is one the domain is stopping anyway, so the wait is on a * monitor which {@code shutdown()} and {@code disable()} wake, rather than slept through. + * And that a request which stood while the domain was disabled is not run once it is + * enabled back: the change it was made for is gone with the pending changes. *

* The checkpointer is the thread put through the wait here because it is the one whose * wait can be seen from a test: {@code shutdown()} waits for it, and it is what saves the @@ -76,12 +78,13 @@ public class SessionRestartBackoffTest extends ReplicationTestCase private static final long INTO_THE_BACKOFF_IN_MS = 1200; /** - * How long {@code shutdown()} may take when the backoff is woken: it drains a replay - * which is not running, stops a session which is already stopped, and waits for the - * checkpointer to save the ServerState once more. A checkpointer which sleeps the - * backoff through holds it for the seconds the backoff has left instead. + * How long {@code shutdown()} or {@code disable()} may take when the backoff is woken: + * either drains a replay which is not running, stops a session which is already stopped, + * and saves the ServerState - {@code shutdown()} by waiting for the checkpointer to do it + * once more. A checkpointer which sleeps the backoff through while it holds the monitor + * the wake is given on holds either of them for the seconds the backoff has left instead. */ - private static final long SHUTDOWN_BOUND_IN_MS = 1500; + private static final long WAKE_BOUND_IN_MS = 1500; /** * How long the checkpointer may take to save a change made once the domain is enabled @@ -90,6 +93,12 @@ public class SessionRestartBackoffTest extends ReplicationTestCase */ private static final long CHECKPOINT_BOUND_IN_MS = 2500; + /** + * How long a request left standing is given to be run by the checkpointer: two of its + * ticks, where one which is standing when the domain is enabled back is run on the first. + */ + private static final long LEFTOVER_REQUEST_BOUND_IN_MS = 2000; + @Test public void aDomainWhichIsShutDownDoesNotWaitOutTheBackoffOfItsSessionRestart() throws Exception @@ -103,6 +112,7 @@ public void aDomainWhichIsShutDownDoesNotWaitOutTheBackoffOfItsSessionRestart() final int rsPort = TestCaseUtils.findFreePort(); replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestShutdownDb"); domain = startDomain(baseDN, rsPort); + assertTrue(domain.isConnected(), "the domain did not connect to its replication server"); broker = openReplicationSession(baseDN, BROKER_ID, 100, rsPort, 1000); leaveTheCheckpointerInTheBackoff(domain, broker, "user.925.shutdown"); @@ -111,18 +121,13 @@ public void aDomainWhichIsShutDownDoesNotWaitOutTheBackoffOfItsSessionRestart() domain.shutdown(); final long tookMs = NANOSECONDS.toMillis(System.nanoTime() - started); - assertTrue(tookMs < SHUTDOWN_BOUND_IN_MS, "shutdown() waited " + tookMs + assertTrue(tookMs < WAKE_BOUND_IN_MS, "shutdown() waited " + tookMs + " ms: it sat out the backoff of the session restart the state checkpointer" + " was waiting through, for a session the domain was stopping anyway"); } finally { - release(domain, broker); - if (domain != null) - { - MultimasterReplication.deleteDomain(baseDN); - } - remove(replicationServer); + release(domain, broker, replicationServer); } } @@ -139,12 +144,24 @@ public void aDomainWhichIsDisabledDoesNotWaitOutTheBackoffOfItsSessionRestart() final int rsPort = TestCaseUtils.findFreePort(); replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestDisableDb"); domain = startDomain(baseDN, rsPort); + assertTrue(domain.isConnected(), "the domain did not connect to its replication server"); broker = openReplicationSession(baseDN, BROKER_ID, 100, rsPort, 1000); final DN entryDN = leaveTheCheckpointerInTheBackoff(domain, broker, "user.925.disable"); - // The data of this domain is about to be replaced, then has been. + /* + * The data of this domain is about to be replaced, then has been. disable() is timed + * as shutdown() is: its wake takes the monitor the checkpointer waits on, so a + * checkpointer which sleeps the backoff through while holding that monitor holds + * disable() for the rest of it - and everything below would then start late enough + * for the save to land inside its bound all the same. + */ + final long started = System.nanoTime(); domain.disable(); + final long tookMs = NANOSECONDS.toMillis(System.nanoTime() - started); + assertTrue(tookMs < WAKE_BOUND_IN_MS, "disable() waited " + tookMs + + " ms: it sat out the backoff of the session restart the state checkpointer" + + " was waiting through, for a session disable() was cutting anyway"); domain.enable(); assertTrue(domain.isConnected(), "the domain did not come back up once enabled"); @@ -179,12 +196,51 @@ public void aDomainWhichIsDisabledDoesNotWaitOutTheBackoffOfItsSessionRestart() } finally { - release(domain, broker); - if (domain != null) - { - MultimasterReplication.deleteDomain(baseDN); - } - remove(replicationServer); + release(domain, broker, replicationServer); + } + } + + /** + * A request which stood while the domain was disabled was made for a change which is + * gone with the pending changes, and the session {@code enable()} starts asks for + * everything the ServerState it loads does not cover: run, the request would stop and + * start that session once for a delivery which can not come. The request is made here by + * hand, in the place of one made between a replay thread's read of the flag and the + * clear {@code disable()} does after it. + *

+ * The restart is the checkpointer's to run, within its first tick after the domain is + * enabled back, so the pin is that the failure it would meet is never spent: a restart + * which ran would have spent it, and would have left the session it stopped down. + */ + @Test + public void aRequestWhichStoodWhileTheDomainWasDisabledIsNotRunOnceItIsEnabledBack() + throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + ReplicationServer replicationServer = null; + LDAPReplicationDomain domain = null; + try + { + final int rsPort = TestCaseUtils.findFreePort(); + replicationServer = createReplicationServer(rsPort, "sessionRestartBackoffTestEnableDb"); + domain = startDomain(baseDN, rsPort); + assertTrue(domain.isConnected(), "the domain did not connect to its replication server"); + + domain.disable(); + domain.requestSessionRestart(); + domain.failNextSessionRestarts(1); + domain.enable(); + assertTrue(domain.isConnected(), "the domain did not come back up once enabled"); + + Thread.sleep(LEFTOVER_REQUEST_BOUND_IN_MS); + assertEquals(domain.getSessionRestartFailuresLeft(), 1, "the request which stood while" + + " the domain was disabled was run against the session enable() started"); + assertTrue(domain.isConnected(), + "the session enable() started was stopped for a request made before it"); + } + finally + { + release(domain, null, replicationServer); } } @@ -241,17 +297,40 @@ private DN leaveTheCheckpointerInTheBackoff( return entry.getName(); } - /** Lets go of what {@link #leaveTheCheckpointerInTheBackoff} set up. */ - private void release(LDAPReplicationDomain domain, ReplicationBroker broker) + /** + * Lets go of what {@link #leaveTheCheckpointerInTheBackoff} set up, and of the domain and + * the replication server of the case - each step whatever the one before it threw, or a + * case which failed on its way up would leave its domain registered for the next one to + * replace silently. + */ + private void release(LDAPReplicationDomain domain, ReplicationBroker broker, + ReplicationServer replicationServer) throws Exception { - ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); - if (domain != null) + try { - domain.failNextSessionRestarts(0); + ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); + if (domain != null) + { + domain.failNextSessionRestarts(0); + } + if (broker != null) + { + broker.stop(); + } } - if (broker != null) + finally { - broker.stop(); + try + { + if (domain != null) + { + MultimasterReplication.deleteDomain(domain.getBaseDN()); + } + } + finally + { + remove(replicationServer); + } } } @@ -265,6 +344,11 @@ private ServerState persistedServerState(DN baseDN) throws Exception return persisted; } + /** + * Creates and starts a domain on the provided base DN, and asserts nothing about it: the + * caller holds the domain before it looks at it, so that a case which fails there still + * has it to delete. + */ private LDAPReplicationDomain startDomain(DN baseDN, int rsPort) throws Exception { final SortedSet replServers = new TreeSet<>(); @@ -273,7 +357,6 @@ private LDAPReplicationDomain startDomain(DN baseDN, int rsPort) throws Exceptio domainCfg.setHeartbeatInterval(100000); final LDAPReplicationDomain domain = MultimasterReplication.createNewDomain(domainCfg); domain.start(); - assertTrue(domain.isConnected(), "the domain did not connect to its replication server"); return domain; } From b31a29794b9b1ecdabd0c9d8d64aa1a3f291aa37 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 16 Sep 2026 10:00:53 +0300 Subject: [PATCH 5/5] [#925] Read which thread ran the restart off the error log, and see the backoff entered rather than sleep into it Review round 3. The give-back case of UpdateOperationTest pinned the level a request is given back with by the clock, and claimed to pin "the thread asks rather than runs, and the checkpointer runs it" the same way - which it did on about 45 % of the runs: under the base's abandonReplay(), which runs the restart it asks for, the thread spends the injected failure at once and the checkpointer's tick, at a phase nothing sets, brings the session back inside the 1.5 s the case looks at or not. Which thread ran the restart is now read off the error log over the run: ERR_REPLAY_SESSION_RESTART_FAILED is reported by the checkpointer's own catch and by nothing else, a replay thread reports the failure it meets as an exception replaying a message, so the case asserts that report exactly once and the base's abandonReplay() is red on the count on every run - the same observable its sibling case already stands on, and no hook for it. SessionRestartBackoffTest slept 1.2 s into a backoff which begins about 1.0-1.05 s after the second failure, and a checkpointer tick pushed past that had the test act on a request standing unwaited - green, nothing false, and the run proving nothing about the wake. The fixture now polls getConsecutiveSessionRestarts(), the fourth test hook of the class, until the third restart in a row has bumped it on its way into the wait, and acts with the whole backoff still to be waited. The javadoc of the wake count, and the wake-count bullet of the description, now say what a disable() and an enable() which both took the lock before the first block of restartSession() leave: the count holding the wake already, no owner to see, and the session enable() started stopped and waited out a backoff for, as the request taken before either clear is owed. --- .../plugin/LDAPReplicationDomain.java | 28 +++++++++- .../replication/UpdateOperationTest.java | 51 ++++++++++++++----- .../plugin/SessionRestartBackoffTest.java | 33 ++++++++---- 3 files changed, 85 insertions(+), 27 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 9b9ae6c276..409083ed20 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -4120,6 +4120,24 @@ public void requestSessionRestart() sessionRestarts.request(SessionRestart.NOW); } + /** + * Returns how many times in a row the session was restarted without a change being + * replayed in between: the count the backoff of the next restart is computed from. + *

+ * Only there for the tests, which read it to see a restart reach its backoff rather than + * wait a delay out and hope it began: the count is bumped on the way into the wait, so + * the restart of {@code n} restarts in a row is at its backoff, or a few instructions + * short of it, once this returns {@code n} - and a wake given in those instructions is + * counted, so the wait sees it all the same. + * + * @return how many session restarts in a row the domain has run + */ + @VisibleForTesting + public int getConsecutiveSessionRestarts() + { + return consecutiveSessionRestarts.get(); + } + /** * Fails this session restart when a test asked for it, and does nothing at all * otherwise. @@ -4927,8 +4945,14 @@ private void wakeSessionRestartBackoff() * Returns how many times the backoff has been woken so far, for a session restart to * take under the lock it stops the session under: the wake of a {@link #disable()} which * has yet to take that lock is then one the restart sees, whether it is waiting by then - * or not, and one which took it already has left the restart nothing to stop. The wake - * of {@link #shutdown()} needs no counting - the flag it stands for never comes back. + * or not, and one which took it already has left the restart nothing to stop - unless an + * {@link #enable()} took it since as well. The restart then stops the session + * {@code enable()} started and sits the whole backoff for it, as it is owed: its request + * was taken before either clear could reach it, and the wake it would have ended on is + * spent. It costs one restart of a session which delivers the change anyway, in a window + * between the take and the block a {@code disable()} and an {@code enable()} would both + * have to fit in. The wake of {@link #shutdown()} needs no counting - the flag it stands + * for never comes back. */ private long sessionRestartBackoffWakes() { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java index 3d71563e1e..d57e0f99ca 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java @@ -2482,6 +2482,13 @@ public void aSessionRestartWhichCouldNotRunIsRunAgain() throws Exception * when the request was given back as it was made, and one backoff later when it was * given back with the wait it is owed. Halfway between the two the session is still down * under the second and up under the first. + *

+ * That the restart which fails is the checkpointer's is read off the error log rather + * than assumed: only the checkpointer reports a restart which threw on it as + * {@code ERR_REPLAY_SESSION_RESTART_FAILED}, so one such report says the thread on its + * way out asked for the restart rather than ran it, on every run and not by the clock - + * a thread which ran it itself would spend the failure at once, and the checkpointer's + * own restart would then be one which runs. */ @Test public void aRestartAskedForWithoutTheBackoffIsGivenBackWithIt() throws Exception @@ -2535,21 +2542,37 @@ public void aRestartAskedForWithoutTheBackoffIsGivenBackWithIt() throws Exceptio * the checkpointer's to run, and it fails where it would start the session again. */ domain.failNextSessionRestarts(1); - parked.release(ResultCode.UNAVAILABLE.intValue()); + final List records = errorLogRecordsOf(() -> { + parked.release(ResultCode.UNAVAILABLE.intValue()); - final long deadline = System.nanoTime() + SECONDS.toNanos(30); - while (domain.getSessionRestartFailuresLeft() > 0) - { - assertTrue(System.nanoTime() < deadline, - "the session restart which was asked to fail did not run within 30 s"); - Thread.sleep(50); - } - Thread.sleep(INTO_THE_BACKOFF_IN_MS); - assertFalse(domain.isConnected(), "the session was started back within " - + INTO_THE_BACKOFF_IN_MS + " ms of the restart which threw: the request a" - + " replay thread on its way out made without the backoff was given back" - + " without it, and the checkpointer ran it again on its next tick rather" - + " than one backoff later"); + final long deadline = System.nanoTime() + SECONDS.toNanos(30); + while (domain.getSessionRestartFailuresLeft() > 0) + { + assertTrue(System.nanoTime() < deadline, + "the session restart which was asked to fail did not run within 30 s"); + Thread.sleep(50); + } + Thread.sleep(INTO_THE_BACKOFF_IN_MS); + assertFalse(domain.isConnected(), "the session was started back within " + + INTO_THE_BACKOFF_IN_MS + " ms of the restart which threw: the request a" + + " replay thread on its way out made without the backoff was given back" + + " without it, and the checkpointer ran it again on its next tick rather" + + " than one backoff later"); + return null; + }); + /* + * Which thread ran the restart is read off the error log rather than off the clock: + * only the state checkpointer reports a restart which threw on it as this, where a + * replay thread reports the failure it meets as an exception replaying a message. + * One report says the checkpointer ran the restart the thread asked for; none would + * say the thread ran it itself on its way out, as it did before, and a second has no + * road - one failure was left to spend. + */ + final int reported = countRecordsOf(records, + "Could not restart the replication session of domain \"" + baseDN + "\""); + assertEquals(reported, 1, "the restart a replay thread on its way out asked for was" + + " run by that thread rather than left to the state checkpointer: " + reported + + " in " + records); /* * Stop parking before the session is back: the delivery it brings is the change diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java index 225e2a76db..632f0d9adb 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java @@ -65,17 +65,18 @@ public class SessionRestartBackoffTest extends ReplicationTestCase private static final int GROUP_ID = 1; /** - * How long after the second restart has failed this test acts on the domain. + * How many session restarts in a row the checkpointer is left at the backoff of. *

- * The checkpointer reports that failure and goes back to its wait of one second, takes - * the request back once that second is out, and then holds the backoff of a third - * restart in a row: three seconds. Acting a fifth of a second into those three lands - * well inside them, and a wait which is not woken has most of them left to hold whoever - * is waiting for the checkpointer. A machine slow enough to push the checkpointer's tick - * past this delay has this test act before the backoff begins, which a wait that is - * slept through survives - the test then proves less, but reports nothing false. + * The first two fail, and the third is the one this test acts on: it is owed a backoff + * of three seconds, which a wait that is not woken holds whoever is waiting for the + * checkpointer through most of. The test waits to see the restart reach that backoff + * rather than act a delay after the second failure and hope the backoff has begun by + * then: the checkpointer takes the request back on its next tick, up to a second after + * it reported that failure, and a tick pushed past a fixed delay would have the test act + * on a request which stands unwaited - which a wait that is slept through survives, so + * the run would prove nothing about the wake. */ - private static final long INTO_THE_BACKOFF_IN_MS = 1200; + private static final int RESTARTS_IN_A_ROW = 3; /** * How long {@code shutdown()} or {@code disable()} may take when the backoff is woken: @@ -255,7 +256,7 @@ public void aRequestWhichStoodWhileTheDomainWasDisabledIsNotRunOnceItIsEnabledBa * it is the checkpointer whichever thread the second failure fell to, since the replay * thread leaves on the failure it meets and no delivery brings one back over a session * which is down. That third restart is owed the backoff of three restarts in a row, and - * this returns once the checkpointer is inside it. + * this returns once the checkpointer has reached it, with the whole of it still to wait. * * @return the DN of the entry the change which is waiting to be delivered again deletes */ @@ -291,7 +292,17 @@ private DN leaveTheCheckpointerInTheBackoff( "the two session restarts which were asked to fail did not both run within 30 s"); Thread.sleep(50); } - Thread.sleep(INTO_THE_BACKOFF_IN_MS); + /* + * The third restart bumps the count on its way into the wait, so the checkpointer is + * at the backoff, or a few instructions short of it, once the count says three - and a + * wake given in those instructions is counted, so it ends the wait all the same. + */ + while (domain.getConsecutiveSessionRestarts() < RESTARTS_IN_A_ROW) + { + assertTrue(System.nanoTime() < deadline, "the session restart of " + RESTARTS_IN_A_ROW + + " in a row did not reach its backoff within 30 s"); + Thread.sleep(50); + } assertFalse(domain.isConnected(), "the session was started back before this test could act on the domain"); return entry.getName();