mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -2847,8 +2847,9 @@ * 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 - it asked for nothing * there, which is why it handed back no change to run a restart for - and a * has an owner is left alone the way the give-back left it - its request there * stands for the state checkpointer to run once the owner is gone, which is why it * handed back no change to run a restart for (see sessionHasAnOwner()) - and a * request another thread left standing is not this one's to spend on a restart * which is refused where it runs. */ @@ -3574,8 +3575,9 @@ // The ack has been published and the change is given back: the replication server // delivers it again, now or - while a total update is being processed over the // session - once it is over, when the restart which was held runs or, on the import // direction, when the session is started from the reloaded state. There is nothing // left to replay here. // direction, when the session is started from the reloaded state, or the one the // state checkpointer restarts once a total update which never began lets go of it. // There is nothing left to replay here. return; } @@ -4023,6 +4025,24 @@ */ 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()) { /* @@ -4030,12 +4050,15 @@ * 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; } @@ -4054,23 +4077,6 @@ */ warned = 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); if (!runRequestedSessionRestarts() && warned) { /* @@ -4281,16 +4287,15 @@ * to be restarted, and a session which starts is given its receive window anew. * <p> * 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. * <p> * A replay thread which is stopping calls this through * {@link #giveBackChangesParkedByStoppingThread()}, for every domain of this server: what @@ -4307,7 +4312,7 @@ * 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<CSN> giveBackParkedChanges(SessionRestart restart) @@ -4317,19 +4322,19 @@ { return parked; } 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. Nothing was asked // for here, so the caller has nothing of this road's to run or to report. return Collections.emptyList(); } /* * 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. The // caller has nothing of this road's to run or to report. return Collections.emptyList(); } for (CSN csn : parked) { incProcessedUpdates(); @@ -4344,17 +4349,50 @@ * <p> * 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 (!remotePendingChanges.replayFailed(csn)) { /* * 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; } /* * 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; } /* @@ -4363,15 +4401,6 @@ * 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); } /** @@ -4517,10 +4546,10 @@ * 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. * 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 @@ 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 @@ * 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 @@ * 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 @@ * 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. * <p> * 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). * <p> * 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 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -332,8 +332,12 @@ * 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 @@ { change.setOwner(null); changeBeingReplayed.remove(Thread.currentThread(), csn); return true; } return false; } finally { opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ModifyMsgWhoseOperationWaitsToBeBuilt.java
New file @@ -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. * <p> * 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. * <p> * 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<Modification> 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); } } opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -120,7 +120,7 @@ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), "a change a replay thread owns must not be taken over (OPENDJ-1115)"); pendingChanges.replayFailed(csn); assertTrue(pendingChanges.replayFailed(csn), "the change its owner gave back was not released"); assertEquals(pendingChanges.getQueueSize(), 1, "the change must stay listed as pending"); assertTrue(state.isEmpty(), "a change which was not replayed must not be recorded as replayed"); @@ -499,15 +499,18 @@ assertTrue(pendingChanges.putRemoteUpdate(delivery)); assertTrue(pendingChanges.markInProgress(delivery)); final AtomicReference<Boolean> releasedByTheOtherThread = new AtomicReference<>(); runAndJoin(new Runnable() { @Override public void run() { pendingChanges.replayFailed(csn); releasedByTheOtherThread.set(pendingChanges.replayFailed(csn)); } }); assertEquals(releasedByTheOtherThread.get(), Boolean.FALSE, "a release which was ignored must say so: nothing was given back to ask for again"); assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), "a change another thread is replaying must not be taken over"); assertEquals(pendingChanges.getQueueSize(), 1); @@ -515,7 +518,8 @@ "a release which was ignored must leave the change with the thread which owns it"); // The thread which owns the change is still the one which decides its fate. pendingChanges.replayFailed(csn); assertTrue(pendingChanges.replayFailed(csn), "the change its owner gave back was not released"); assertFalse(pendingChanges.replayFailed(csn), "a change given back once was released again"); assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")), "the change its owner gave back must be taken over by the next delivery"); /* @@ -529,6 +533,38 @@ } /** * A release of a change which is not listed as an uncommitted one anymore gives nothing * back, and says so: a replay thread which reaches its give-up after the pending changes * were forgotten - by the domain disabled and enabled again, or by the import at its end - * must not ask for a session restart for a change the ServerState loaded then accounts for * (issue #1061). */ @Test public void replayFailedReleasesNothingOnceTheChangeIsForgottenOrCommitted() throws Exception { final ServerState state = new ServerState(); final RemotePendingChanges pendingChanges = new RemotePendingChanges(state); final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0); final CSN forgotten = generator.newCSN(); final DeleteMsg forgottenDelivery = deleteMsg(forgotten, "uuid-1"); assertTrue(pendingChanges.putRemoteUpdate(forgottenDelivery)); assertTrue(pendingChanges.markInProgress(forgottenDelivery)); pendingChanges.clear(); assertFalse(pendingChanges.replayFailed(forgotten), "a change the pending changes forgot was reported as given back"); final CSN committed = generator.newCSN(); final DeleteMsg committedDelivery = deleteMsg(committed, "uuid-2"); assertTrue(pendingChanges.putRemoteUpdate(committedDelivery)); assertTrue(pendingChanges.markInProgress(committedDelivery)); pendingChanges.commit(committed); assertFalse(pendingChanges.replayFailed(committed), "a change which was committed was reported as given back"); assertTrue(state.cover(committed)); } /** * A change which was parked because it depends on another one is handed out by * {@code getNextUpdate()} to whichever replay thread calls it first once the change it * was waiting for is gone, rather than replayed by the thread which parked it as a opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -72,15 +72,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). * <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 * 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. * <p> * 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 @@ /** * 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. * <p> * 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 @@ * 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 @@ "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"); @@ -417,13 +423,287 @@ } /** * 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). * <p> * 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. * <p> * 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). * <p> * 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). * <p> * 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. * <p> * 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. * <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 @@ -983,8 +1263,53 @@ 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"); } }
