From 25f103498105dc6551f2fa1e202730128674067c Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 17 Sep 2026 18:56:15 +0300 Subject: [PATCH] [#1061] Ask for the session restart under an owner as well, and leave it to the state checkpointer A change released while the session of its domain had an owner - the domain shutting down or disabled, or a total update into this replica from the moment it is asked for - was released and nothing more: recoverFromReplayFailure(), abandonReplay() and, with #954, giveBackParkedChanges() all returned before asking for the session restart. Two of the three owners forget the pending changes and every request with them, disable() with enable() and the import at its end, so nothing was lost there. The third does not: a total update which is asked for and never begins - initializeFromRemote() refused at publish, an ErrorMsg from the exporter, the watchdog of InitializeTask, or the second attempt initialize() makes after a failed import - ends in releaseIEContext() alone, and a change released under it stayed listed, uncommitted and owned by nobody until the next failed replay of this domain restarted the session. On a domain which then went quiet that was for good: a commit moves the ServerState no further than the oldest uncommitted change, so the state in memory, and the one persisted from it, stopped at the change - the shape of #954, one road over. #968 left it that way because at the time the request was a flag only the thread of a failed replay ran: a flag set under an owner was served by the next failed replay, which would have set it itself. Since #981 the state checkpointer runs the request nobody ran, and holds it for as long as the session has an owner - which the last resort of replay() relies on already: its request is made under an owner too, and only its run is gated. The three roads now ask on every road, as one policy: the request is made once the change is released, before the owner is read, and the owner arm returns with it standing. The owners which forget the pending changes forget the request with them; the one which does not leaves it to the checkpointer, which runs it within its tick of the owner letting go and has the replication server send the change again. The lines which say the change is being asked for again, and the count of the deliveries given back, stay off the owner roads: a server which is shutting down abandons every change in flight, and none of them is delivered again before it is started back. The gates on the two runs of replay() - the parked road and the last resort - are load-bearing on every owner road with this: a run there would spend the request on restartSession()'s refusal, where the request left standing is what the checkpointer runs. Pinned in ReplayDuringImportTest on the one owner a test can hold open and then take away: the total update is asked for, the exporter holds the request, and in the place of the answer the request is given up through abortStalledInitializeFromRemote(0). One case per road: a change whose attempts in place are spent under the request, one a stopping thread abandons under it, and one a replay which is unwound had parked - behind a change another thread of the test holds before its operation is built, so that nothing has failed and the give-back's request is the only one standing. Each change travels the replication server, and each is delivered again over the session the checkpointer restarts once the request is given up, replayed and covered. The parked case also pins the gate of the parked road: with the import forgotten by the gate, the run under the owner spends the give-back's request and nothing comes back. TestSynchronousReplayQueue is backed by a ConcurrentLinkedDeque, as the redelivery is offered by the listener thread of the domain and taken by the thread of the test. --- .../plugin/LDAPReplicationDomain.java | 204 ++++++----- ...ModifyMsgWhoseOperationWaitsToBeBuilt.java | 74 ++++ .../plugin/ReplayDuringImportTest.java | 323 ++++++++++++++++-- .../plugin/TestSynchronousReplayQueue.java | 12 +- 4 files changed, 498 insertions(+), 115 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index f68df24167..78074bb1ab 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 @@ -2838,9 +2838,10 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * the state checkpointer runs one restart for every change the threads of the * pool hand back on their way out, rather than each of them running one while * the configuration change which is stopping them waits. A domain whose session - * has an owner is left alone the way the give-back left it: nothing was asked - * for on that road, and a request another thread left standing is not this - * one's to spend on a restart which is refused where it runs. + * has an owner is left alone as well: the restart is refused where it runs, and + * a run here would spend the request on the refusal - the give-back's own, made + * under the owner for the state checkpointer to run once the owner is gone, and + * any other thread's standing next to it (see sessionHasAnOwner()). */ runRequestedSessionRestarts(); } @@ -3540,8 +3541,9 @@ else if (op instanceof ModifyDNOperation) if (replayFailed && recoverFromReplayFailure(msg.getCSN(), replayThreadShutdown)) { // The ack has been published and the change is given back: the replication server - // delivers it again, now or - while a total update owns the session - after the - // import restarts it. There is nothing left to replay here. + // delivers it again, now or - while a total update owns the session - over the + // session the import starts, or the one the state checkpointer restarts once the + // owner is gone. There is nothing left to replay here. return; } @@ -3984,6 +3986,24 @@ private boolean recoverFromReplayFailure( */ remotePendingChanges.replayFailed(csn); + /* + * This change is not owned by anyone anymore, so the session has to be restarted for + * the replication server to deliver it again. Ask for the restart before trying to + * 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. + */ + sessionRestarts.request(replayThreadShutdown.get() || outOfMemory + ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF); + if (sessionHasAnOwner()) { /* @@ -3991,12 +4011,15 @@ private boolean recoverFromReplayFailure( * this thread's to restart. Restarting the one which is being stopped would leave a * broker and a listener thread behind on a domain whose alert generator, flush * thread and RSUpdater are already gone; restarting the one an import streams over - * would end the import on the entries which had arrived. The change given back here - * is forgotten with the rest of the pending changes when the ServerState is loaded - * again, from the backend or from the imported data, and the session started then - * asks for everything that state does not cover - or, when the total update it was - * given back for never begins, it is asked for by the next restart - * (see sessionHasAnOwner()). + * would end the import on the entries which had arrived. The request made above is + * left standing, the way it is on every road which releases a change under an owner + * (see sessionHasAnOwner()): the owners which forget the pending changes forget the + * request with them, and the one which does not - a total update which never begins + * - leaves it to the state checkpointer, which runs it once the owner is gone and + * has the replication server send the change again. The line which says the change + * is being asked for again is not logged here: a server which is shutting down + * abandons every change in flight, and none of them is asked for again before it is + * started back. */ return true; } @@ -4014,23 +4037,6 @@ private boolean recoverFromReplayFailure( */ logReplayRetryWarning(csn, failure); } - /* - * This change is not owned by anyone anymore, so the session has to be restarted for - * the replication server to deliver it again. Ask for the restart before trying to - * 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. - */ - sessionRestarts.request(replayThreadShutdown.get() || outOfMemory - ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF); runRequestedSessionRestarts(); return true; } @@ -4188,16 +4194,15 @@ private void runPendingSessionRestart() * to be restarted, and a session which starts is given its receive window anew. *

* 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 @@ -4220,18 +4225,18 @@ private boolean giveBackParkedChanges(SessionRestart restart) { return false; } - if (sessionHasAnOwner()) - { - // The domain owns its session, or a total update does: both forget the pending - // changes, and neither leaves a session for this thread to restart. - return true; - } /* * Asked for before the changes are reported: a throw out of the report - the JVM which * unwound this replay is out of memory - must not lose the restart which is what brings * them back. */ sessionRestarts.request(restart); + if (sessionHasAnOwner()) + { + // The domain owns its session, or a total update does: neither leaves a session for + // this thread to restart, and the request stands for the state checkpointer. + return true; + } for (CSN csn : parked) { incProcessedUpdates(); @@ -4253,10 +4258,28 @@ private boolean giveBackParkedChanges(SessionRestart restart) private void abandonReplay(CSN csn) { remotePendingChanges.replayFailed(csn); + /* + * 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. + * + * 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, and the pending changes are - // forgotten with the ServerState on its way down or at the end of the import. + // 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; } /* @@ -4265,15 +4288,6 @@ private void abandonReplay(CSN csn) * 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 - * 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); } /** @@ -4415,10 +4429,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() @@ -5390,24 +5404,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); @@ -5430,12 +5445,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; @@ -6215,13 +6229,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). */ private boolean sessionHasAnOwner() { 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 mods, String entryUUID, CountDownLatch letGo) + { + super(csn, dn, mods, entryUUID); + this.letGo = letGo; + } + + @Override + public ModifyOperation createOperation(InternalClientConnection connection, DN newDN) + throws LDAPException, IOException, DataFormatException + { + try + { + letGo.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + throw new IOException("interrupted while waiting to be let go", e); + } + return super.createOperation(connection, newDN); + } +} 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 23d4f018d6..be86d8bac0 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 @@ -26,6 +26,7 @@ import java.util.List; import java.util.SortedSet; import java.util.TreeSet; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; import org.forgerock.opendj.ldap.DN; @@ -65,15 +66,20 @@ * one thing it must not do is stop the session the import is reading (issue #956). The same * holds from the moment the total update is asked for: the answer to the request arrives * over that session, so a replay which fails while it is on its way must not restart it. - * A restart asked for before the total update took the session, and left standing for the - * length of it, is not run once it is over either: the change it was asked for is gone with - * the ServerState the import replaced. The changes a replay which is unwound had parked as - * waiting for another one are released on the same terms, and nothing more is done for them - * (issue #954). + * A restart asked for while the total update owns the session is left standing for the + * length of it, and is not run once it is over: the change it was asked for is gone with the + * ServerState the import replaced. The changes a replay which is unwound had parked as + * waiting for another one are released on the same terms (issue #954). A total update which + * is asked for and never begins - the request is refused, or gives up waiting for its answer + * - replaces nothing, and the request left standing under it is what has the changes + * released under it delivered again (issue #1061). *

* 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 {@code timeOut} each case declares is what it is expected to take at the most; it is * not what bounds it. {@code TestListener} sets the timeout of every test method from the @@ -297,12 +303,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, @@ -317,7 +324,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, @@ -387,10 +394,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"); @@ -404,14 +411,237 @@ 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)); + 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(); + assertTrue(domain.isConnected(), + "the session the answer to the request would arrive over was stopped under the owner"); + + 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. + domain.markInProgress(delivered); + 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(); + assertTrue(domain.isConnected(), + "the session the answer to the request would arrive over was stopped under the owner"); + + 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 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. The parked + * road of {@code replay()} runs what is requested when the session has no owner; run under + * the owner, the restart would be refused where it runs and the request spent on the + * refusal, with nothing left for the state checkpointer to run - so the same case pins + * that the run is held back under the owner the give-back asked under. + *

+ * 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(); + 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"); + assertTrue(domain.isConnected(), + "the session the answer to the request would arrive over was stopped under the owner"); + + // 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 @@ -602,8 +832,53 @@ private static List errorLogRecordsOf(int msgId, CSN csn) 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"); } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java index dd908c0623..a921f38b0d 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java @@ -13,13 +13,15 @@ * * Copyright 2009 Sun Microsystems, Inc. * Portions copyright 2013-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; import java.util.Collection; +import java.util.Deque; import java.util.Iterator; -import java.util.LinkedList; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.TimeUnit; /** @@ -30,7 +32,13 @@ */ public class TestSynchronousReplayQueue implements BlockingQueue { - private LinkedList list = new LinkedList<>(); + /** + * Written by the listener thread of the domain - a change the replication server delivers + * is offered here - and read by the thread of the test, which replays it: the two share no + * lock, so the deque has to be safe for that hand-off. {@code take()} is still synchronous - + * it throws when nothing was offered - which is what makes the queue a test one. + */ + private final Deque list = new ConcurrentLinkedDeque<>(); @Override public boolean add(UpdateToReplay e)