From c6b483127f13fc7d96dbc90c5157ab113ac12973 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 16 Sep 2026 10:07:08 +0000
Subject: [PATCH] [#925] Keep asking for a session restart until it has run (#981)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 41 +
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java | 124 +++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 7
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 267 +++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java | 125 +++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 401 ++++++++++++++---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java | 380 ++++++++++++++++
7 files changed, 1,272 insertions(+), 73 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 f8c6fcc..409083e 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,36 @@
/** 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.
+ * <p>
+ * 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 AtomicBoolean sessionRestartRequested = new AtomicBoolean();
+ 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 Object sessionRestartBackoff = new Object();
+ /**
+ * How many times {@link #sessionRestartBackoff} has been woken, so that a wake is never
+ * missed.
+ * <p>
+ * 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
@@ -369,6 +394,13 @@
*/
private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger();
/**
+ * How many of the next session restarts must fail rather than start the session again.
+ * <p>
+ * 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.
* <p>
@@ -632,7 +664,8 @@
/**
* 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 +689,12 @@
{
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 +2611,13 @@
{
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()
@@ -2712,8 +2758,12 @@
* 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
@@ -2732,14 +2782,14 @@
* 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 +2807,15 @@
}
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 +3811,58 @@
* 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;
}
}
}
@@ -3839,12 +3874,73 @@
}
/**
+ * Runs the session restart this domain was asked for and which no thread of its own
+ * ran.
+ * <p>
+ * 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.
+ * <p>
+ * 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()
+ {
+ 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.
+ *
+ * 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.
+ */
+ 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.
+ }
+ }
+ }
+
+ /**
* Gives a change back to the replication server when this replay thread stops before it
* could apply it.
* <p>
- * 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 +3959,15 @@
* 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);
}
/**
@@ -3874,6 +3977,7 @@
private void restartSession(boolean wait)
{
final long stoppedSession;
+ final long wakes;
synchronized (serviceStateLock)
{
if (sessionHasAnOwner())
@@ -3884,6 +3988,7 @@
}
disableService();
stoppedSession = getSessionGeneration();
+ wakes = sessionRestartBackoffWakes();
}
if (wait)
{
@@ -3893,7 +3998,7 @@
* 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)
{
@@ -3913,6 +4018,7 @@
*/
return;
}
+ failSessionRestartIfATestAskedFor();
enableService();
}
}
@@ -3923,12 +4029,36 @@
* 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);
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.
+ *
+ * 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 && wakes == sessionRestartBackoffWakes;
+ left = until - monotonicNowInMs())
+ {
+ sessionRestartBackoff.wait(left);
+ }
+ }
}
catch (InterruptedException e)
{
@@ -3943,6 +4073,85 @@
}
/**
+ * Has the next session restarts of this domain fail where they would start the session
+ * again.
+ * <p>
+ * 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.
+ * <p>
+ * 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();
+ }
+
+ /**
+ * Asks for a session restart the way a replay thread on its way out does, without
+ * running it.
+ * <p>
+ * 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);
+ }
+
+ /**
+ * 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.
+ * <p>
+ * 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.
+ */
+ 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.
*
* @param operation
@@ -4710,9 +4919,47 @@
* 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, 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 - 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()
+ {
+ synchronized (sessionRestartBackoff)
+ {
+ return sessionRestartBackoffWakes;
+ }
}
/**
@@ -4830,6 +5077,16 @@
{
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();
@@ -5343,7 +5600,7 @@
* 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 0000000..a854b5a
--- /dev/null
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/SessionRestartRequests.java
@@ -0,0 +1,124 @@
+/*
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * The wait a failing backend is owed belongs to the request rather than to the thread
+ * 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
+{
+ /**
+ * 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<SessionRestart> 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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 restart what the caller which could not run the restart asks for again
+ */
+ void giveBack(SessionRestart restart)
+ {
+ merge(restart);
+ }
+
+ /** 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 d28000e..c6b8fc1 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 @@
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 \
@@ -679,6 +680,10 @@
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_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
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 77d0acf..d57e0f9 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 @@
*/
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.
+ * <p>
+ * 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;
@@ -2355,6 +2369,259 @@
}
/**
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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
+ {
+ 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 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(2);
+
+ final CSN csn = gen.newCSN();
+ final List<String> 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 restarts which failed");
+ return null;
+ });
+ assertEquals(domain.getSessionRestartFailuresLeft(), 0,
+ "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: 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.
+ */
+ 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");
+ }
+ finally
+ {
+ domain.failNextSessionRestarts(0);
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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
+ {
+ 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<Throwable> 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);
+ final List<String> 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");
+ 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
+ * 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
* resolution knows how to solve. Such a change is left to conflict resolution, and when
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 3bedecb..3dc590f 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.
* <p>
* 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
@@ -285,6 +288,44 @@
}
/**
+ * 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.
+ * <p>
+ * 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,
* and a change replayed here is replayed into no backend.
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 0000000..632f0d9
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartBackoffTest.java
@@ -0,0 +1,380 @@
+/*
+ * 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.
+ * 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.
+ * <p>
+ * 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 many session restarts in a row the checkpointer is left at the backoff of.
+ * <p>
+ * 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 int RESTARTS_IN_A_ROW = 3;
+
+ /**
+ * 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 WAKE_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;
+
+ /**
+ * 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
+ {
+ 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);
+ 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");
+
+ final long started = System.nanoTime();
+ domain.shutdown();
+ final long tookMs = NANOSECONDS.toMillis(System.nanoTime() - started);
+
+ 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, 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);
+ 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. 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");
+
+ /*
+ * 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, 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.
+ * <p>
+ * 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);
+ }
+ }
+
+ /**
+ * Leaves the state checkpointer of the provided domain sitting through the backoff of
+ * a session restart, with the session stopped.
+ * <p>
+ * 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 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
+ */
+ 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);
+ }
+ /*
+ * 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();
+ }
+
+ /**
+ * 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
+ {
+ try
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ if (domain != null)
+ {
+ domain.failNextSessionRestarts(0);
+ }
+ if (broker != null)
+ {
+ broker.stop();
+ }
+ }
+ finally
+ {
+ try
+ {
+ if (domain != null)
+ {
+ MultimasterReplication.deleteDomain(domain.getBaseDN());
+ }
+ }
+ finally
+ {
+ remove(replicationServer);
+ }
+ }
+ }
+
+ 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;
+ }
+
+ /**
+ * 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<String> 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();
+ 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<String>(), 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
new file mode 100644
index 0000000..fde84a8
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartRequestsTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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 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()
+ {
+ 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);
+ }
+}
--
Gitblit v1.10.0