* On a domain whose session has an owner - the domain itself, going away, or a total - * update into it, from the moment it is asked for - they are released and nothing more, - * the way {@code abandonReplay()} hands a change back on that road (see - * {@link #sessionHasAnOwner()}): there is no session of this thread's to restart, and - * the restart it would ask for is refused where it runs. The domain forgets its pending - * changes on its way down, the import forgets them at its end, and a change released - * for a total update which never begins stays listed until the next failed replay of - * this domain restarts the session, which has the replication server send it again. A - * line which says the replication server sends the change again would not hold on any - * of these - a server which is shutting down abandons every change in flight, and none - * of them is delivered again before it is started back. + * update into it, from the moment it is asked for - they are released and asked for, and + * nothing more, the way {@code abandonReplay()} hands a change back on that road (see + * {@link #sessionHasAnOwner()}): there is no session of this thread's to restart, so the + * request is left standing for the state checkpointer, which runs it once the owner is + * gone - or finds it forgotten, with the pending changes it was made for, by the domain + * on its way down or by the import at its end. A line which says the replication server + * sends the change again would not hold on every one of these - a server which is + * shutting down abandons every change in flight, and none of them is delivered again + * before it is started back - so none is logged, and the deliveries are not counted. *
* A replay thread which is stopping calls this through
* {@link #giveBackChangesParkedByStoppingThread()}, for every domain of this server: what
@@ -4307,7 +4312,7 @@ private void runPendingSessionRestart()
* restart asked for them - on a thread which is not stopping, and on a domain whose
* session has no owner - and reports what that restart did for them. Empty when
* this thread had parked none, and empty on a domain whose session has an owner,
- * where nothing is asked for. The list is the one the release allocated: nothing is
+ * where the request is left standing for the state checkpointer. The list is the one the release allocated: nothing is
* allocated for the answer on the road out of a JVM which has run out of memory
*/
private List
* 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.
+ * and the changes which follow it are held back until it is replayed. A change which
+ * this thread did not give back - it is not listed anymore, or another thread owns it -
+ * is asked for by nobody here.
*
* @param csn the CSN of the change this thread was replaying
*/
private void abandonReplay(CSN csn)
{
- remotePendingChanges.replayFailed(csn);
- if (sessionHasAnOwner())
+ if (!remotePendingChanges.replayFailed(csn))
{
- // The domain, or the import into it, owns its session, and the pending changes are
- // forgotten with the ServerState on its way down or at the end of the import.
+ /*
+ * Not listed anymore: the owner which forgot the pending changes - disable() with
+ * enable(), or the import at its end - did it while this thread was on its way here,
+ * and it starts the session again from the ServerState it loaded, which asks for
+ * everything that state does not cover. A request made now would outlive the clear
+ * of that owner and have the state checkpointer stop and start the session it has
+ * just started, for a change which is gone. The other road to false - the change is
+ * owned by another thread, which took it over since - needs no restart either: that
+ * thread is the one which replays it or gives it back.
+ */
return;
}
- /*
- * Logged here rather than where the change is abandoned: a server which is shutting
- * down abandons every change in flight, and none of them is asked for again before it
- * is started back - one line per change would say otherwise.
- */
- logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN());
/*
* 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
@@ -4370,8 +4378,29 @@ private void abandonReplay(CSN csn)
* 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.
+ *
+ * Asked for whether or not the session has an owner, the way every road which releases
+ * a change asks (see sessionHasAnOwner()): the domain on its way down and the import at
+ * its end forget the request with the pending changes, and a total update which never
+ * begins leaves it to the state checkpointer. Asked for once the change is released and
+ * not before: asked for first, it could be taken and run by another thread while this
+ * one still owned the change, and the delivery the new session brought would be turned
+ * down as the duplicate of a change a replay thread owns.
*/
sessionRestarts.request(SessionRestart.NOW);
+ if (sessionHasAnOwner())
+ {
+ // The domain, or the import into it, owns its session: a server which is shutting
+ // down abandons every change in flight, and none of them is delivered again before
+ // it is started back, so the line below would not hold.
+ return;
+ }
+ /*
+ * Logged here rather than where the change is abandoned: a server which is shutting
+ * down abandons every change in flight, and none of them is asked for again before it
+ * is started back - one line per change would say otherwise.
+ */
+ logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN());
}
/**
@@ -4517,10 +4546,10 @@ public int getSessionRestartFailuresLeft()
* 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.
+ * time of their choosing without a change released for it: the one a replay thread makes
+ * on a domain whose session has no owner is made and run in one go, and one made under
+ * an owner stands for a change which is listed, and which the owner forgets or the
+ * checkpointer has delivered again.
*/
@VisibleForTesting
public void requestSessionRestart()
@@ -5492,24 +5521,25 @@ 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.
+ * Cleared here as well as by disable(): a request made after disable()'s own clear -
+ * every road which releases a change asks for the restart, whether or not the domain
+ * owns its session, and a replay which outlasted the drain disable() waits for
+ * releases its change under the flag - 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.
*
- * The deliveries folded into no warning are forgotten here for the same window: a
- * thread which is recording a failed replay reads the flag, then folds the delivery
- * - recoverFromReplayFailure() logs before it asks - then asks, and runs what it
- * asked for itself, which restartSession() turns down on a domain which owns its
- * session. So on that road it is the fold rather than the request which outlives
- * disable()'s clear, and the first warning over the data loaded back would count a
- * delivery of a change which went with the pending changes. What reads the count is
- * a warning, and none is logged between disable() and here short of that thread's
- * own, so a test tells this zeroing and disable()'s apart by nothing: they stand or
- * fall together.
+ * The deliveries folded into no warning are forgotten here for a window of the same
+ * kind: a thread which is recording a failed replay asks, then reads the flag, then
+ * folds the delivery and runs what it asked for, which restartSession() turns down
+ * on a domain which owns its session. A read which found the domain enabled a moment
+ * before disable() set the flag leaves a fold which outlives disable()'s clear, and
+ * the first warning over the data loaded back would count a delivery of a change
+ * which went with the pending changes. What reads the count is a warning, and none
+ * is logged between disable() and here short of that thread's own, so a test tells
+ * this zeroing and disable()'s apart by nothing: they stand or fall together.
*/
sessionRestarts.clear();
foldedReplayRetryWarnings.set(0);
@@ -5532,12 +5562,11 @@ public void enable()
* stopping one: enableService() ends with startListenService(), so the listener it
* starts can list a delivery and hand it to a replay thread while this method is
* still running. A replay thread which reads a flag that still says "disabled"
- * gives the change up at the top of its replay loop, and abandonReplay() does not
- * ask for it again - a domain on its way down owns its session - so the change is
- * left listed, uncommitted and owned by nobody. Nothing would replay it: the
- * replication server only sends it again over a session which is restarted, so this
- * domain's ServerState, and every change which depends on that one, would be held
- * back for as long as the session lives.
+ * gives the change up at the top of its replay loop, and abandonReplay() releases it
+ * with a request standing - made after the clear above, so nothing clears it - which
+ * the state checkpointer runs within its tick: the change would be delivered again,
+ * but over a restart of the session just started rather than to the replay it had
+ * been handed to.
*/
disabled = false;
boolean started = false;
@@ -6021,10 +6050,12 @@ protected void importBackend(InputStream input) throws DirectoryException
* uncommitted change - the state would never move past it, and the replication
* server does not send a change again which the state it is given covers. Nothing
* listed a change meanwhile: the listener thread is the one running this import,
- * and the replay threads gave up every attempt while the flag was set. The restart
- * a replay thread may have asked for before the total update owned the session
- * goes with them, as do the deliveries folded into no warning: the caller starts
- * the session again from the reloaded state.
+ * and the replay threads gave up every attempt while the flag was set. The restarts
+ * they asked for on their way out, while the total update owned the session, and
+ * any asked for before it did, go with them, as do the deliveries folded into no
+ * warning: the caller starts the session again from the reloaded state. A replay
+ * thread which reaches its give-up after this clear finds its change unlisted and
+ * asks for nothing.
*/
remotePendingChanges.clear();
sessionRestarts.clear();
@@ -6324,13 +6355,25 @@ private boolean ownsItsSession()
* arrives over that session, so a restart made while it is on its way loses it, and the
* import which follows reads its entries over the same session - stopping it ends the
* import on the entries which had arrived. The import starts the next session itself,
- * from the state it loaded. A change given back while the total update owned the session
- * is not asked for again by anyone until then; if no import follows - the request was
- * refused, or gave up waiting - it stays listed until the next failed replay restarts
- * the session, which has the replication server send it again with everything after it.
- * Listed, it holds the ServerState back as well: a commit moves the state no further than
- * the oldest uncommitted change, so the state in memory, and the one persisted from it,
- * stop at the change until that restart.
+ * from the state it loaded.
+ *
+ * A change released while the session has an owner is asked for again all the same - every
+ * road which releases one asks, owner or not - and the request is what is left to the
+ * owner: none of the roads runs the restart it asks for under an owner, since
+ * {@link #restartSession(boolean)} refuses it and the request would be spent on the
+ * refusal, and the state checkpointer holds every request for as long as the session has
+ * an owner. Two owners forget the pending changes and the request with them, when the
+ * ServerState they went with is replaced: {@link #disable()} and {@link #enable()}, and
+ * the import at its end - the session started then asks for everything the state it
+ * loaded does not cover. The third does not: a total update which is asked for and never
+ * begins - the request refused, or given up as unanswered - replaces nothing, and the
+ * request left standing under it is run by the checkpointer within its tick of the owner
+ * letting go, which has the replication server send the change again with everything
+ * after it. Left listed and asked for by nobody, the change would hold the ServerState
+ * back for good on a domain which then goes quiet: a commit moves the state no further
+ * than the oldest uncommitted change, so the state in memory, and the one persisted from
+ * it, would stop at the change until the next failed replay of this domain restarted
+ * the session (issue #1061).
*
* A configuration change is refused while a total update runs - by the listener of the
* domain entry and by the one of its external changelog entry - so what reaches the
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
index 5835b12e91..085d14776b 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -332,8 +332,12 @@ public void commit(CSN csn)
* now is the double replay the ownership is there to prevent (issue #922).
*
* @param csn the CSN of the change whose replay failed
+ * @return whether the change was given back: {@code false} when it is not listed as an
+ * uncommitted change anymore - the domain forgot its pending changes while it was
+ * being replayed - or when another thread owns it, in which case nothing was
+ * released here and there is no delivery for a session restart to ask for
*/
- public void replayFailed(CSN csn)
+ public boolean replayFailed(CSN csn)
{
pendingChangesWriteLock.lock();
try
@@ -343,7 +347,9 @@ public void replayFailed(CSN csn)
{
change.setOwner(null);
changeBeingReplayed.remove(Thread.currentThread(), csn);
+ return true;
}
+ return false;
}
finally
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java
new file mode 100644
index 0000000000..1c84d7c7ef
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java
@@ -0,0 +1,74 @@
+/*
+ * 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.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.zip.DataFormatException;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.core.ModifyOperation;
+import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.types.LDAPException;
+import org.opends.server.types.Modification;
+
+/**
+ * A ModifyMsg whose replay waits, before anything is attempted, until the test lets it go.
+ *
+ * The operation is built first thing by a replay, before any lock is taken and before the
+ * change is checked against the ones it may depend on, so a replay which waits here holds
+ * the change as one being replayed - listed, uncommitted and owned by its thread - for as
+ * long as the test wants, while every other replay of the domain runs: a change which
+ * follows this one on the same entry is parked as waiting for it, and nothing has failed,
+ * so no session restart has been asked for. Let go, the replay runs to its end and commits
+ * the change as any other.
+ *
+ * Such a message can not travel the protocol: it is handed to the domain rather than
+ * published, and replayed on a thread of the test.
+ */
+final class ModifyMsgWhoseOperationWaitsToBeBuilt extends ModifyMsg
+{
+ private final CountDownLatch letGo;
+
+ /**
+ * @param letGo the latch the replay waits on before the operation is built
+ */
+ ModifyMsgWhoseOperationWaitsToBeBuilt(
+ CSN csn, DN dn, List
* 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
- * exporter is holding the answer.
+ * exporter is holding the answer, which it may never give. A change which has to be delivered
+ * again is published through the replication server, which is what has it to send again;
+ * the replay queue of the domain is the test's, so every delivery is replayed when, and on
+ * the thread, the test says.
*
* The claim of a total update this replica did not ask for is made by the listener thread
* under no lock, so a restart of the session which reads no owner a moment before that claim
@@ -309,12 +314,13 @@ public void aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver() throws Exceptio
/**
* The changes a replay which is unwound had parked as waiting for another change are
- * released and nothing more while a total update owns the session (issue #954): no
- * session restart is asked for them - the one it would ask for is refused where it runs,
- * and the request would be spent on it - and they are neither reported as changes the
- * replication server sends again, which it does not before the import has replaced the
- * data, nor counted as processed. That is the road a change a stopping replay thread
- * abandons takes on this domain, and the give-back of the parked changes takes it too.
+ * released while a total update owns the session, and the session is left to the owner
+ * (issue #954): the restart asked for them is not run - it is refused where it runs, and
+ * the request would be spent on it - and they are neither reported as changes the
+ * replication server sends again, which it does not before the total update has let go
+ * of the session, nor counted as processed. That is the road a change a stopping replay
+ * thread abandons takes on this domain, and the give-back of the parked changes takes it
+ * too.
*
* Pinned on the import road because it is the one road with an owner which a test holds
* open for as long as it needs: the request is on its way until the exporter answers it,
@@ -329,7 +335,7 @@ public void aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver() throws Exceptio
* back, and the error which ends a replay thread is caught here instead.
*/
@Test(timeOut = 120_000)
- public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayIsNotAskedForAgain() throws Exception
+ public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayLeavesTheSessionToTheOwner() throws Exception
{
final Entry entry = TestCaseUtils.addEntry(
"dn: cn=renamedSince," + EXAMPLE_DN,
@@ -399,10 +405,10 @@ public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayIsNotAskedForAgain()
"the change parked by the replay which was unwound must be given back");
assertEquals(getMonitorAttrValue(baseDN, "replayed-updates"), processed,
"a change released while a total update owns the session must not be counted as"
- + " processed: no session sends it again before the import has replaced the data");
+ + " processed: no session sends it again before the total update lets go of it");
assertThat(errorLogRecordsOf(NOTE_REPLAY_PARKED_CHANGE_GIVEN_BACK.ordinal(), parked))
.as("the change was reported as one the replication server sends again, which it does"
- + " not before the import has replaced the data")
+ + " not before the total update lets go of the session")
.isEmpty();
assertTrue(domain.isConnected(), "the session the answer to the request arrives over was stopped");
@@ -416,14 +422,288 @@ public void aParkedChangeGivenBackWhileTheRequestIsOnItsWayIsNotAskedForAgain()
}
}
+ /**
+ * A change released while a total update which never begins owns the session is asked for
+ * again under the owner, and delivered again over the session the state checkpointer
+ * restarts once the owner is gone (issue #1061).
+ *
+ * A total update this replica asked for owns the session from the request on, and a change
+ * whose replay fails meanwhile is released and left to the owner: the domain forgets its
+ * pending changes on its way down, and the import forgets them at its end - but a request
+ * which is refused, or which gives up waiting for its answer, replaces nothing and forgets
+ * nothing. Released and asked for by nobody, the change would stay listed and uncommitted
+ * until the next failed replay of this domain restarted the session, and the ServerState -
+ * which a commit moves no further than the oldest uncommitted change - would stop at it
+ * with everything behind it. So the restart is asked for under the owner as well, and the
+ * state checkpointer, which holds every request for as long as the total update owns the
+ * session, runs it within its tick of the owner letting go.
+ *
+ * The road pinned here is the one a change whose attempts in place are spent takes: every
+ * attempt ends on an entryUUID search which does not run. The request gives up through
+ * the watchdog of the initialize task, which is the one road out of an unanswered request
+ * a test can take at a time of its choosing.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeReleasedUnderARequestWhichIsNeverAnsweredIsDeliveredAgain() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+
+ // The request is out, and the exporter never answers it.
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CSN csn = gen.newCSN();
+ final LDAPUpdateMsg delivered = publishAndAwaitDelivery(new ModifyMsg(csn,
+ DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "released while the request was on its way"), entryUUID));
+ // A restart run on this thread would be back up before any read of the session: this counts it.
+ domain.failNextSessionRestarts(1);
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replay(delivered);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+ assertFalse(domain.getServerState().cover(csn),
+ "the change whose replay fails must stay listed as one which is not in the data");
+ assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn))
+ .as("the change was warned about as one the replication server sends again, which it"
+ + " does not while the total update owns the session")
+ .isEmpty();
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "a session restart was run under the owner by the replay whose attempts were spent");
+ domain.failNextSessionRestarts(0);
+
+ giveUpTheRequest();
+
+ final LDAPUpdateMsg again = awaitDelivery(csn, 30_000, "the change released under the"
+ + " request was not delivered again once the request gave up: nothing asked for the"
+ + " session restart which has the replication server send it again");
+ replay(again);
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("description"))
+ .as("the change delivered again was not applied").isNotEmpty();
+ assertTrue(domain.getServerState().cover(csn),
+ "the change delivered again was applied and not recorded: it is still listed");
+ }
+
+ /**
+ * A change a stopping replay thread abandons while a total update which never begins owns
+ * the session takes the same road (issue #1061): abandoned at the top of its first attempt
+ * without being counted against its budget, released, asked for again under the owner,
+ * and delivered again once the owner is gone.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeAbandonedUnderARequestWhichIsNeverAnsweredIsDeliveredAgain() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CSN csn = gen.newCSN();
+ final LDAPUpdateMsg delivered = publishAndAwaitDelivery(new ModifyMsg(csn, entry.getName(),
+ generatemods("description", "abandoned while the request was on its way"), entryUUID));
+ // The thread of this test is one which is stopping: the change is abandoned unapplied.
+ assertTrue(domain.markInProgress(delivered), "the delivery was not handed to this thread");
+ domain.failNextSessionRestarts(1);
+ domain.replay(delivered, new AtomicBoolean(true));
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("description"))
+ .as("a change abandoned by a stopping thread was applied").isEmpty();
+ assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), csn))
+ .as("the change was reported as one the replication server sends again, which it"
+ + " does not while the total update owns the session")
+ .isEmpty();
+ assertEquals(getMonitorAttrValue(baseDN, "changes-with-failed-replay"), 0,
+ "the abandoned change was counted against its budget: it was never attempted");
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "a session restart was run under the owner by the thread which abandoned the change");
+ domain.failNextSessionRestarts(0);
+
+ giveUpTheRequest();
+
+ final LDAPUpdateMsg again = awaitDelivery(csn, 30_000, "the change abandoned under the"
+ + " request was not delivered again once the request gave up: nothing asked for the"
+ + " session restart which has the replication server send it again");
+ replay(again);
+ assertTrue(domain.getServerState().cover(csn),
+ "the change delivered again was applied and not recorded: it is still listed");
+ }
+
+ /**
+ * A change a stopping replay thread abandons once the import has forgotten it asks for no
+ * session restart (issue #1061).
+ *
+ * The import forgets the pending changes at its end, and the session it starts asks for
+ * everything the ServerState it loaded does not cover. A thread which read the flag while
+ * the import ran and reaches the give-back only after that finds its change unlisted: a
+ * restart asked for then would outlive the clear, and the state checkpointer would stop
+ * and start the session the import has just started, for a change which is gone.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeAbandonedOnceTheImportForgotItAsksForNoRestart() throws Exception
+ {
+ final String[] exported = exportedEntries();
+
+ // Owned by this thread before the import begins, and still owned when the import ends.
+ final CSN csn = gen.newCSN();
+ domain.processUpdate(new ModifyMsg(csn, DN.valueOf(IMPORTED_ENTRY_DN),
+ generatemods("description", "abandoned once the import forgot it"), IMPORTED_ENTRY_UUID));
+ final LDAPUpdateMsg delivered = queue.take().getUpdateMessage();
+ assertTrue(domain.markInProgress(delivered), "the delivery was not handed to this thread");
+
+ startImportInto(exported.length);
+ finishImport(exported);
+
+ domain.failNextSessionRestarts(1);
+ try
+ {
+ domain.replay(delivered, new AtomicBoolean(true));
+
+ // Two ticks of the checkpointer: a request standing now is run on the first.
+ Thread.sleep(2000);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1, "a change the import forgot"
+ + " asked for a restart of the session the import started at its end");
+ assertThat(errorLogRecordsOf(NOTE_REPLAY_ABANDONED_CHANGE.ordinal(), csn))
+ .as("a change the import forgot was reported as one the replication server sends again")
+ .isEmpty();
+ }
+ finally
+ {
+ domain.failNextSessionRestarts(0);
+ }
+ }
+
+ /**
+ * A change the give-back released while a total update which never begins owns the session
+ * is asked for again under the owner by the give-back itself, and the request is left
+ * standing rather than spent (issue #1061).
+ *
+ * The change it waited for is one another thread is replaying, held before its operation
+ * is built: nothing has failed, so no road but the give-back has asked for anything, and
+ * the request found standing once the owner is gone is the give-back's own. No restart is
+ * run on the thread of the give-back under the owner: one run there would be refused where
+ * it runs, and the case counts the restarts which get that far.
+ *
+ * The replay which is unwound is this thread's, as in the case above: its change is
+ * applied, and the ack of its delivery runs out of memory.
+ */
+ @Test(timeOut = 120_000)
+ public void aParkedChangeGivenBackUnderARequestWhichIsNeverAnsweredIsDeliveredAgain() throws Exception
+ {
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=renamedSince," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: renamedSince",
+ "sn: renamedSince");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final Entry other = TestCaseUtils.addEntry(
+ "dn: cn=unwound," + EXAMPLE_DN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: unwound",
+ "sn: unwound");
+ final String otherUUID = getEntryUUID(other.getName());
+
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ /*
+ * The change the parked one waits for: replayed by a thread of the test which is held
+ * before the operation is built, so the change is being replayed - listed, uncommitted,
+ * owned - for as long as the latch holds, and nothing has failed.
+ */
+ final CountDownLatch letGo = new CountDownLatch(1);
+ final CSN held = gen.newCSN();
+ domain.processUpdate(new ModifyMsgWhoseOperationWaitsToBeBuilt(held, entry.getName(),
+ generatemods("description", "the change being replayed by another thread"), entryUUID,
+ letGo));
+ final LDAPUpdateMsg heldMsg = queue.take().getUpdateMessage();
+ final Thread otherThread = new Thread(() ->
+ {
+ assertTrue(domain.markInProgress(heldMsg), "the held change must be the one listed");
+ domain.replay(heldMsg, SHUTDOWN);
+ }, "ReplayDuringImportTest replay held before its operation is built");
+ otherThread.start();
+ try
+ {
+ // Parked as waiting for the held change by this thread, which owns it from here on.
+ final CSN parked = gen.newCSN();
+ final LDAPUpdateMsg parkedDelivery = publishAndAwaitDelivery(new ModifyMsg(parked,
+ entry.getName(),
+ generatemods("description", "the change which was parked as a dependency"), entryUUID));
+ replay(parkedDelivery);
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 1,
+ "a change which waits for one being replayed by another thread must be parked");
+
+ // The replay which is unwound while this thread still holds the parked change.
+ final CSN unwound = gen.newCSN();
+ domain.failNextSessionRestarts(1);
+ try
+ {
+ replayMsg(new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, other.getName(),
+ generatemods("description", "the replay of this change is unwound once it is applied"),
+ otherUUID));
+ Assert.fail("the replay was not unwound: the ack of the delivery must run out of memory");
+ }
+ catch (OutOfMemoryError unwinding)
+ {
+ // The error is the fixture's own, and this is the thread it would have ended.
+ }
+ assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+ "the change parked by the replay which was unwound must be given back");
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "a session restart was run under the owner by the replay which was unwound");
+ domain.failNextSessionRestarts(0);
+
+ // The held change runs to its end: nothing fails, nothing asks for a restart.
+ letGo.countDown();
+ otherThread.join(30_000);
+ assertFalse(otherThread.isAlive(), "the held replay did not end once let go");
+ assertTrue(domain.getServerState().cover(held), "the held change was not recorded");
+ assertFalse(domain.getServerState().cover(parked),
+ "the parked change must stay listed as one which is not in the data");
+
+ giveUpTheRequest();
+
+ final LDAPUpdateMsg again = awaitDelivery(parked, 30_000, "the parked change given back"
+ + " under the request was not delivered again once the request gave up: the session"
+ + " restart the give-back asked for under the owner was not run, or was spent");
+ replay(again);
+ assertTrue(domain.getServerState().cover(parked),
+ "the parked change delivered again was applied and not recorded: it is still listed");
+ }
+ finally
+ {
+ letGo.countDown();
+ otherThread.join(30_000);
+ }
+ }
+
/**
* 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.
+ * for a change it gave back - before the total update owned the session, or under the
+ * owner - 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 the one a failed replay makes.
*
* 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
@@ -983,8 +1263,53 @@ private static void awaitUninterruptibly(CountDownLatch latch)
private void replayMsg(UpdateMsg updateMsg) throws InterruptedException
{
domain.processUpdate(updateMsg);
- final LDAPUpdateMsg ldapUpdate = queue.take().getUpdateMessage();
- domain.markInProgress(ldapUpdate);
- domain.replay(ldapUpdate, SHUTDOWN);
+ replay(queue.take().getUpdateMessage());
+ }
+
+ /** Replays a delivery on the thread of this test, as a replay thread would. */
+ private void replay(LDAPUpdateMsg delivery)
+ {
+ assertTrue(domain.markInProgress(delivery), "the delivery is not the one listed: " + delivery);
+ domain.replay(delivery, SHUTDOWN);
+ }
+
+ /**
+ * Publishes a change through the replication server, which is what has it to deliver again
+ * once the session is restarted for it, and waits for the delivery to this replica.
+ */
+ private LDAPUpdateMsg publishAndAwaitDelivery(LDAPUpdateMsg msg) throws Exception
+ {
+ exporter.publish(msg);
+ return awaitDelivery(msg.getCSN(), 30_000, "the change published was not delivered");
+ }
+
+ /**
+ * Waits for the replication server to deliver the change to this replica: the listener
+ * thread of the domain puts it in the replay queue of the test, which takes it out.
+ */
+ private LDAPUpdateMsg awaitDelivery(CSN csn, long timeoutMs, String orElse) throws Exception
+ {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ while (queue.peek() == null)
+ {
+ assertTrue(System.currentTimeMillis() < deadline, orElse + " within " + timeoutMs + " ms");
+ Thread.sleep(50);
+ }
+ final LDAPUpdateMsg msg = queue.take().getUpdateMessage();
+ assertEquals(msg.getCSN(), csn, "another change than the one awaited was delivered");
+ return msg;
+ }
+
+ /**
+ * Has the total update this replica asked for give up on its request, the way the
+ * watchdog of the initialize task does once the request has waited two minutes for an
+ * answer: the total update never begins, and its context is released with nothing
+ * replaced, so the session has no owner anymore.
+ */
+ private void giveUpTheRequest()
+ {
+ assertTrue(domain.abortStalledInitializeFromRemote(0),
+ "the request was not the one waiting for an answer");
+ assertFalse(domain.ieRunning(), "the total update was given up and is still being processed");
}
}