+ * It is what {@link #disabled} is for an import or a restore run on this server, on the + * one road which does not set that flag: {@link #preBackendImport(LocalBackend)} takes + * the backend away without disabling the domain, because the domain can not stop the + * session it is importing over. The replay threads read it where they read + * {@link #disabled}: an attempt made meanwhile is made into no backend, and whatever it + * decided is overwritten by the import, so the change is given back instead - without a + * session restart, which {@link #sessionHasAnOwner()} refuses for the whole of the total + * update. It is set for the length of {@link #importBackend(InputStream)}, so it covers + * the reload of the ServerState and the reset of the pending changes which follow the + * import. + */ + private volatile boolean importingData; /** * This list is used to temporary store operations that needs to be replayed @@ -1849,7 +1866,16 @@ SynchronizationProviderResult handleConflictResolution( * this operation has already been replayed in the past. */ String uuid = ctx.getEntryUUID(); - if (findEntryDN(uuid) != null) + final DN replayedEntryDN; + try + { + replayedEntryDN = findEntryDN(uuid); + } + catch (SearchFailedException e) + { + return searchDidNotRun(ctx.getCSN(), e); + } + if (replayedEntryDN != null) { return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_OPERATION, null); @@ -1866,7 +1892,15 @@ SynchronizationProviderResult handleConflictResolution( { // There is a potential of perfs improvement here // if we could avoid the following parent entry retrieval - DN parentDnFromCtx = findEntryDN(ctx.getParentEntryUUID()); + final DN parentDnFromCtx; + try + { + parentDnFromCtx = findEntryDN(ctx.getParentEntryUUID()); + } + catch (SearchFailedException e) + { + return searchDidNotRun(ctx.getCSN(), e); + } if (parentDnFromCtx == null) { // The parent does not exist with the specified unique id @@ -1893,6 +1927,28 @@ SynchronizationProviderResult handleConflictResolution( return new SynchronizationProviderResult.ContinueProcessing(); } + /** + * Turns down an operation being replayed whose conflict resolution could not read the + * data, so that the change is attempted again rather than applied on what a search + * which did not run seemed to say. + *
+ * The operation is reported as unavailable, which is what it is, and the search which + * did not run is its error message: the change is retried in place, and left out of + * the ServerState and asked for again if this server keeps failing to serve the + * searches this phase reads the data with (issue #956). Nothing is logged here: this + * hook runs on every attempt in place, and the error is reported once, with the + * attempt which ended them, where a storage which failed to serve the operation is. + * + * @param csn the CSN of the change being replayed + * @param e the search which did not run + * @return the result which stops the operation + */ + private SynchronizationProviderResult searchDidNotRun(CSN csn, SearchFailedException e) + { + return new SynchronizationProviderResult.StopProcessing( + ResultCode.UNAVAILABLE, e.report(csn, getBaseDN())); + } + /** * Check that the broker associated to this ReplicationDomain has found * a Replication Server and that this LDAP server is therefore able to @@ -2730,15 +2786,28 @@ private void replayChangeAndTheChangesWaitingForIt( dependency = remotePendingChanges.checkDependencies(op, msg); boolean replayDone = false; boolean firstAttempt = true; + /* + * The search conflict resolution could not run on the last attempt, when it ended + * on one: the result code of the operation says nothing of it - it is the conflict + * the operation failed on - so the failure of the server below the loop is told + * here, and reported there once the attempts are spent. + */ + SearchFailedException searchFailedResolvingConflict = null; int retryCount = IN_PLACE_REPLAY_ATTEMPTS; while (!dependency && !replayDone && retryCount-- > 0) { /* - * The flag which says this domain is going down is read before the lock as well - * as under it. The replay threads are a pool shared by every domain of this - * server, so a thread which took a change of a domain which is going down should - * not queue behind the wait for that domain: the changes of every other domain - * are behind it in the same pool. + * What this attempt ends on, not what an earlier one did: an attempt the server + * refuses before it reaches the data makes no search, and the verdict below the + * loop reads the attempt which spent the last of them. + */ + searchFailedResolvingConflict = null; + /* + * The flags which say this domain is going down, or that its data is being + * replaced, are read before the lock as well as under it. The replay threads are + * a pool shared by every domain of this server, so a thread which took a change + * of a domain which is going down should not queue behind the wait for that + * domain: the changes of every other domain are behind it in the same pool. * * The read under the lock is the one which decides; the one above it is a * scheduling optimisation for the common case and nothing more. It cannot keep @@ -2746,7 +2815,7 @@ private void replayChangeAndTheChangesWaitingForIt( * between the two reads, and a reader which arrives behind a queued writer blocks * even on a lock which is not the fair kind. */ - boolean goingDown = replayThreadShutdown.get() || shutdown.get() || disabled; + boolean goingDown = replayThreadShutdown.get() || shutdown.get() || disabled || importingData; if (!goingDown) { /* @@ -2759,7 +2828,7 @@ private void replayChangeAndTheChangesWaitingForIt( replayReadLock.lock(); try { - goingDown = replayThreadShutdown.get() || shutdown.get() || disabled; + goingDown = replayThreadShutdown.get() || shutdown.get() || disabled || importingData; if (!goingDown) { if (!firstAttempt) @@ -2845,34 +2914,52 @@ else if (isServerFailure(result, serverErrorResultCode)) else { ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO; - if (op instanceof ModifyOperation) + try { - ModifyOperation castOp = (ModifyOperation) op; - dependency = remotePendingChanges.checkDependencies(castOp); - ModifyMsg modifyMsg = (ModifyMsg) msg; - resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg); - } - else if (op instanceof DeleteOperation) - { - DeleteOperation castOp = (DeleteOperation) op; - dependency = remotePendingChanges.checkDependencies(castOp); - resolution = dependency ? resolution : solveNamingConflict(castOp, msg); - } - else if (op instanceof AddOperation) - { - AddOperation castOp = (AddOperation) op; - AddMsg addMsg = (AddMsg) msg; - dependency = remotePendingChanges.checkDependencies(castOp); - resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg); + if (op instanceof ModifyOperation) + { + ModifyOperation castOp = (ModifyOperation) op; + dependency = remotePendingChanges.checkDependencies(castOp); + ModifyMsg modifyMsg = (ModifyMsg) msg; + resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg); + } + else if (op instanceof DeleteOperation) + { + DeleteOperation castOp = (DeleteOperation) op; + dependency = remotePendingChanges.checkDependencies(castOp); + resolution = dependency ? resolution : solveNamingConflict(castOp, msg); + } + else if (op instanceof AddOperation) + { + AddOperation castOp = (AddOperation) op; + AddMsg addMsg = (AddMsg) msg; + dependency = remotePendingChanges.checkDependencies(castOp); + resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg); + } + else if (op instanceof ModifyDNOperation) + { + ModifyDNOperation castOp = (ModifyDNOperation) op; + ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg; + dependency = remotePendingChanges.checkDependencies(modifyDNMsg); + resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg); + } + // else: unknown type of operation ?! there is nothing to replay } - else if (op instanceof ModifyDNOperation) + catch (SearchFailedException e) { - ModifyDNOperation castOp = (ModifyDNOperation) op; - ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg; - dependency = remotePendingChanges.checkDependencies(modifyDNMsg); - resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg); + /* + * Conflict resolution reads the data with a search of the entryUUID, and + * that search did not run: nothing was decided here, and whether the entry + * is still in the data is not known. Report the failure of the server it + * is rather than let a caller read "no entry" out of a search which never + * answered. It is logged once the attempts in place are spent rather than + * on every one of them, as a storage which failed to serve the operation + * is: a backend which is down for a while fails every attempt of every + * change delivered meanwhile. + */ + searchFailedResolvingConflict = e; + resolution = ConflictResolution.SEARCH_FAILED; } - // else: unknown type of operation ?! there is nothing to replay if (!dependency) { @@ -2886,6 +2973,19 @@ else if (op instanceof ModifyDNOperation) recordChangeResolved(csn); break; + case SEARCH_FAILED: + /* + * Conflict resolution could not read the data, so the change is not in + * it and nothing was concluded about the entry it targets. Give it the + * in-place attempts a storage which failed gets - a storage busy for a + * moment must not cost a session restart - and leave it out of the + * ServerState once they are spent, which the failure of the server + * below the loop reports and acts on. A change which is not in the data + * must not advance the ServerState (issue #889). + */ + Thread.sleep(50); + break; + case FAILED: if (serverErrorResultCode.equals(result)) { @@ -2980,20 +3080,31 @@ else if (op instanceof ModifyDNOperation) * is the storage failing to serve the operation. The result of that attempt is * what decides, so that this branch reports the failure it is acting on: an * attempt which ended on something conflict resolution kept rewriting is the - * loop below, however the attempts before it ended. + * loop below, however the attempts before it ended. So is an attempt whose + * conflict resolution could not read the data, which the attempt itself says + * rather than its result code: that one is the conflict the operation failed + * on, not the search which did not run (issue #956). */ if (isServerFailure(lastResult, serverErrorResultCode) || ResultCode.BUSY.equals(lastResult) - || serverErrorResultCode.equals(lastResult)) + || serverErrorResultCode.equals(lastResult) + || searchFailedResolvingConflict != null) { /* * The server kept failing to apply the change, so the change is not in the data. * Leave it out of the ServerState, otherwise the replication server would never * send it again and this replica would silently diverge while reporting itself * up to date. + * + * The error reported is the operation's, unless the attempt ended on a search + * conflict resolution could not run: the operation then only says which conflict + * it failed on, and the search is what failed. */ + final Object error = searchFailedResolvingConflict != null + ? searchFailedResolvingConflict.report(csn, getBaseDN()) + : op.getErrorMessage(); final LocalizableMessage message = ERR_ERROR_REPLAYING_OPERATION.get( - op, csn, lastResult, op.getErrorMessage()); + op, csn, lastResult, error); logger.error(message); replayErrorMsg = message.toString(); replayFailed = true; @@ -3214,8 +3325,9 @@ else if (op instanceof ModifyDNOperation) */ if (replayFailed && recoverFromReplayFailure(msg.getCSN(), replayThreadShutdown)) { - // The ack has been published and the change, still owned by the replication - // server, is being delivered again: there is nothing left to replay here. + // 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. return; } @@ -3524,13 +3636,19 @@ private boolean recoverFromReplayFailure( */ remotePendingChanges.replayFailed(csn); - if (shutdown.get() || disabled) + if (sessionHasAnOwner()) { /* * This whole domain is going away or is being imported into: there is no session of * 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. + * 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()). */ return true; } @@ -3642,9 +3760,10 @@ private void runRequestedSessionRestarts(boolean wait) private void abandonReplay(CSN csn) { remotePendingChanges.replayFailed(csn); - if (shutdown.get() || disabled) + if (sessionHasAnOwner()) { - // The domain owns its session, and it forgets its pending changes on its way down. + // 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. return; } /* @@ -3666,9 +3785,10 @@ private void restartSession(boolean wait) final long stoppedSession; synchronized (serviceStateLock) { - if (ownsItsSession()) + if (sessionHasAnOwner()) { - // The domain is going away or is being imported into: it owns its session. + // The domain is going away or is being imported into: the session is not this + // thread's to stop. return; } disableService(); @@ -3694,6 +3814,11 @@ private void restartSession(boolean wait) * import - in the meantime: the session this thread stopped is gone, so it has * nothing left to start. Every stop and every start of a session is counted, so * the generation alone tells one session from another. + * + * An import is not asked about here: the session it would stream over is the one + * this thread stopped, so none is streaming, and a total update which was asked + * for meanwhile needs the session started back to be answered at all - its + * request is what fails, and loudly, when no answer comes. */ return; } @@ -3782,24 +3907,79 @@ private static SearchResultEntry getFirstResult(InternalSearchOperation search) * @param uuid the Entry Unique ID. * @return The current DN of the entry or null if there is no entry with * the specified UUID. + * @throws SearchFailedException if the search did not run, so that whether there is + * such an entry is not known */ - private DN findEntryDN(String uuid) + private DN findEntryDN(String uuid) throws SearchFailedException { - try + if (uuid == null) { - final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, "entryuuid=" + uuid); - InternalSearchOperation search = conn.processSearch(request); - final SearchResultEntry resultEntry = getFirstResult(search); - if (resultEntry != null) + // A change which carries no entryUUID names no entry. + return null; + } + /* + * The entryUUID comes off the wire and nothing validates it as one, so it is looked up + * as the value it is rather than read as part of a filter string: a value which + * carried a wildcard would be read as a substring filter, and one which did not parse + * - a dangling escape - would be a search which never runs, for as long as the change + * is asked for. + */ + final SearchFilter filter = SearchFilter.createEqualityFilter( + getServerContext().getSchema().getAttributeType(ENTRYUUID_ATTRIBUTE_NAME), + ByteString.valueOfUtf8(uuid)); + final InternalSearchOperation search = + conn.processSearch(newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, filter)); + if (search.getResultCode() != ResultCode.SUCCESS) + { + if (search.getResultCode() == ResultCode.NO_SUCH_OBJECT && baseEntryIsAbsentFromALiveBackend()) { - return resultEntry.getName(); + /* + * The backend serves the base DN and has no base entry yet: the search ran, and + * nothing is below a base entry which is not there. This is the empty replica + * about to receive it. + */ + return null; } + /* + * The search did not run - the backend is offline or being rebuilt, or the storage + * failed to serve it - so it read nothing of the data: an entry it did not report + * is not an entry which is not there. + */ + throw new SearchFailedException(uuid, search.getResultCode().getName() + " " + search.getErrorMessage()); + } + final SearchResultEntry resultEntry = getFirstResult(search); + return resultEntry != null ? resultEntry.getName() : null; + } + + /** + * Returns whether the backend which serves the base DN of this domain is there and + * holds no base entry, which is the state of an empty replica. + *
+ * A search under the base DN of such a backend answers NO_SUCH_OBJECT, which is the + * answer of a backend which is offline or being rebuilt as well - and no backend + * answers SUCCESS with no entry for a base which is not there. The backend itself + * tells the two apart: the search ran over a backend which is there and empty, and it + * did not run over one which is gone or fails to answer. + * + * @return {@code true} if the backend is there and holds no base entry + */ + private boolean baseEntryIsAbsentFromALiveBackend() + { + final LocalBackend backend = getBackend(); + if (backend == null) + { + // Nothing serves the base DN: the backend is offline or being rebuilt. + return false; + } + try + { + return !backend.entryExists(getBaseDN()); } catch (DirectoryException e) { - // never happens because the filter is always valid. + // The storage did not answer this any more than it answered the search. + return false; } - return null; } /** Outcome of the conflict resolution attempted after a replayed operation failed. */ @@ -3809,18 +3989,61 @@ private enum ConflictResolution REPLAY_AGAIN, /** The change is already reflected in the data: there is nothing left to replay. */ NOTHING_TO_DO, + /** + * The search conflict resolution reads the data with did not run: nothing was + * decided, and the operation has to be attempted again. + */ + SEARCH_FAILED, /** The operation failed for a reason which is not a naming conflict. */ FAILED } + /** + * Reports a search conflict resolution reads the data with and which did not run: it + * read nothing, so its result is no evidence about the data. + *
+ * A search which failed and a search which found nothing look the same to a caller + * which only reads the entries it returned, and every caller here reads "no entry" as + * "the entry has been deleted": that answers {@link ConflictResolution#NOTHING_TO_DO}, + * which records a change that was never applied as replayed, and the replication server + * never sends a change again which this replica reports itself past (issue #956). + */ + private static final class SearchFailedException extends Exception + { + private static final long serialVersionUID = 1L; + + /** The entryUUID the search which did not run was looking for. */ + private final String entryUUID; + + private SearchFailedException(String entryUUID, String cause) + { + super(cause); + this.entryUUID = entryUUID; + } + + /** + * Describes the search which did not run for the change it was made for. + * + * @param csn the CSN of the change being replayed + * @param baseDN the base DN of the domain the change belongs to + * @return the error to report the change with + */ + private LocalizableMessage report(CSN csn, DN baseDN) + { + return ERR_REPLAY_ENTRYUUID_SEARCH_FAILED.get(csn, baseDN, entryUUID, getMessage()); + } + } + /** * Solve a conflict detected when replaying a modify operation. * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. * @return the outcome of the conflict resolution + * @throws SearchFailedException if the data could not be read */ private ConflictResolution solveNamingConflict(ModifyOperation op, ModifyMsg msg) + throws SearchFailedException { ResultCode result = op.getResultCode(); ModifyContext ctx = (ModifyContext) op.getAttachment(SYNCHROCONTEXT); @@ -3905,8 +4128,10 @@ else if (result == ResultCode.NOT_ALLOWED_ON_RDN) * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. * @return the outcome of the conflict resolution + * @throws SearchFailedException if the data could not be read */ private ConflictResolution solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg) + throws SearchFailedException { ResultCode result = op.getResultCode(); DeleteContext ctx = (DeleteContext) op.getAttachment(SYNCHROCONTEXT); @@ -4403,11 +4628,12 @@ public void disable() * Waits for the replay threads which are applying a change of this domain to be done * with it. *
- * Called once {@link #disabled} or {@link #shutdown} has been set, which is what bounds - * the wait: a replay thread reads those under {@link #replayReadLock}, the lock this - * takes exclusively, so no attempt starts once this returns and what it waits for is the - * attempts which were running already. The lock is released before returning for the - * same reason - what keeps the replay out is the flag, not the lock. + * Called once {@link #disabled}, {@link #shutdown} or {@link #importingData} has been + * set, which is what bounds the wait: a replay thread reads those under + * {@link #replayReadLock}, the lock this takes exclusively, so no attempt starts once + * this returns and what it waits for is the attempts which were running already. The + * lock is released before returning for the same reason - what keeps the replay out is + * the flag, not the lock. */ private void awaitReplayDrained() { @@ -4936,6 +5162,16 @@ protected void importBackend(InputStream input) throws DirectoryException ImportExportContext ieCtx = getImportExportContext(); try { + /* + * The replay of this domain is held off before the backend is taken away, the way + * disable() holds it off before the backend is taken away for an import run on this + * server: the flag keeps the attempts which have not started from starting, and the + * wait is for the ones which had. What this road can not do is stop the session - + * it is the session the entries are about to stream over, on this very thread. + */ + importingData = true; + awaitReplayDrained(); + if (!backend.supports(BackendOperation.LDIF_IMPORT)) { ieCtx.setExceptionIfNoneSet(new DirectoryException(OTHER, @@ -5001,6 +5237,25 @@ protected void importBackend(InputStream input) throws DirectoryException ResultCode.OTHER, ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe)))); } + finally + { + /* + * The ServerState in memory is the one read from the data now, and the changes + * listed as pending must not outlive the one they went with, as they do not when + * the domain is disabled: a change given back while the import ran would stay + * listed, uncommitted and owned by nobody, and a commit stops at the first + * 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: the caller starts the session again from the reloaded state. + */ + remotePendingChanges.clear(); + sessionRestartRequested.set(false); + consecutiveSessionRestarts.set(0); + importingData = false; + } } if (ieCtx.getException() != null) @@ -5276,6 +5531,29 @@ private boolean ownsItsSession() return shutdown.get() || disabled; } + /** + * Whether the session of this domain has an owner other than the replay thread which + * would restart it after a failed replay: the domain itself, when it is shutting down or + * disabled ({@link #ownsItsSession()}), or a total update into this replica. + *
+ * The total update owns the session from the moment it is asked for, not from the
+ * moment its entries stream: the {@code InitializeTargetMsg} which answers the request
+ * 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.
+ */
+ private boolean sessionHasAnOwner()
+ {
+ return ownsItsSession() || importInProgress();
+ }
+
@Override
protected void restartService()
{
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index fe6c496d9d..f83968145f 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -686,3 +686,7 @@ NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED_327=The configuration change was a
the domain is shutting down, or it is disabled for the length of a total update. The change is \
stored and takes effect when the session is started again, which a domain left disabled by a \
failed import or restore never does
+ERR_REPLAY_ENTRYUUID_SEARCH_FAILED_322=Could not read the data to check change %s for a conflict \
+ in domain "%s": the search of the entry with entryUUID %s did not run (%s). The change is not \
+ applied on what a search which read nothing seemed to say about the data, and is not recorded \
+ as replayed
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
index f61b23d1f1..786fce9d50 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/plugins/ShortCircuitPlugin.java
@@ -644,16 +644,18 @@ private int shortCircuitInternal(PluginOperation operation, String section)
// Check for registered short circuits.
final String key = keyFor(operation.getOperationType(), section);
Integer resultCode = shortCircuits.get(key);
- if (resultCode != null)
+ if (resultCode != null && appliesTo(key, operation))
{
final int reached = shortCircuitCounts.computeIfAbsent(key, k -> new AtomicInteger()).incrementAndGet();
+ final int letThroughFirst = shortCircuitSkips.getOrDefault(key, 0);
final Integer maxTimes = shortCircuitLimits.get(key);
- if (maxTimes == null || reached <= maxTimes)
+ if (reached > letThroughFirst && (maxTimes == null || reached <= letThroughFirst + maxTimes))
{
return resultCode;
}
- // The short circuit was applied as many times as it was asked for: from now on the
- // operations are let through, which is how a transient failure is simulated.
+ // The operations before the short circuit are let through, and so are the ones after
+ // it was applied as many times as it was asked for: this is how a transient failure
+ // which starts, or ends, part way through a sequence of operations is simulated.
}
/*
@@ -740,6 +742,23 @@ public static List
+ * The operations of one type which reach a plugin point are not all the test's: the
+ * server makes internal operations of its own on its own schedule - the ServerState flush
+ * thread of a replication domain writes the base entry with a Modify on its tick, say -
+ * and a short circuit which counts them takes a let-through, or a refusal, meant for the
+ * operation the test is driving. A test which counts its operations one by one names them.
+ *
+ * @param operation The type of operation the short circuit applies to.
+ * @param section The plugin point the short circuit applies to.
+ * @param resultCode The result code to be returned for the short circuit.
+ * @param letThroughFirst How many of the operations the predicate accepts must be let
+ * through before the short circuit applies.
+ * @param maxTimes How many of them must be short circuited after those.
+ * @param appliesTo Which operations of that type the short circuit is for.
+ */
+ public static void registerShortCircuit(OperationType operation, String section, int resultCode,
+ int letThroughFirst, int maxTimes, Predicate
+ * The change replayed here was made on the master under a DN this replica does not
+ * have: the entry lives under another one, the way it does after a rename which was
+ * replayed first, and only the entryUUID search finds it. So the change is applied
+ * only if that search is given another chance once the storage serves it again.
+ * Reading its failure as "the entry has been deleted" answers NOTHING_TO_DO, which
+ * records the change as replayed and loses it for good - the replication server never
+ * sends a change this replica reports itself past.
+ */
+ @Test
+ public void modifyIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final String phoneNumber = "01 02 45";
+
+ // The DN the change carries is the one the entry had on the master. Here the entry
+ // is the one which was just added, which only the entryUUID search finds.
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ /*
+ * The storage does not serve the search for the first attempts and serves it after
+ * them: a failure which lasts less than the attempts made in place, the way a
+ * backend which is being rebuilt or a connection which was lost does. The short
+ * circuit is put in force right before the replay and dropped right after it - it
+ * applies to every search of this server while it is registered, and the replay
+ * here runs on this thread.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2);
+ try
+ {
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", phoneNumber), entryUUID));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ final Entry replayedEntry = DirectoryServer.getEntry(entry.getName());
+ assertEquals(replayedEntry.parseAttribute("telephonenumber").asString(), phoneNumber,
+ "the change was not applied: a search which could not run was read as a deleted entry");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: the entryUUID searches which check a replayed Add for a
+ * conflict read the data the same way, and a search which did not run is no evidence
+ * about it either. The first of them checks whether the Add was replayed here already.
+ *
+ * The Add is delivered a second time - the replication server sends again what a
+ * replica does not report itself past - and the entry was renamed here since it was
+ * added: only the entryUUID search finds it. An entry that search did not report is
+ * not an entry which is not there: reading it that way adds the entry a second time,
+ * under its former DN, and the data holds one entryUUID twice.
+ */
+ @Test
+ public void addIsNotReplayedTwiceWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("addWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final RDN renamedRDN = RDN.valueOf("cn=renamedAfterTheAdd");
+ final ModifyDNOperation rename =
+ getRootConnection().processModifyDN(entry.getName(), renamedRDN, true);
+ assertEquals(rename.getResultCode(), ResultCode.SUCCESS);
+ final CSN csn = gen.newCSN();
+
+ // The first attempt fails its first search; the second attempt has it served.
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1);
+ try
+ {
+ replayMsg(addMsg(entry, csn, getEntryUUID(baseDN), entryUUID));
+ assertShortCircuitSpentBy(1);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(entryExists(entry.getName()),
+ "the entry was added a second time: a search which could not run was read as an "
+ + "Add which was not replayed here yet");
+ assertTrue(entryExists(baseDN.child(renamedRDN)), "the renamed entry is gone");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which is in the data must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: the search which checks that the parent of a replayed
+ * Add is still the one the change was made under fails, and the one before it - the
+ * check that the Add was not replayed here already - ran.
+ *
+ * A parent that search did not report is not a parent which was deleted: the Add is
+ * attempted again once the storage serves the search, and no naming conflict is
+ * counted for a search which read nothing. Reading it as a parent which is gone hands
+ * the Add to conflict resolution as the naming conflict it is not - and renames the
+ * entry under the base DN as a conflicting entry, a divergence which is left for an
+ * administrator to repair by hand, when the search conflict resolution makes fails as
+ * well.
+ */
+ @Test
+ public void addIsRetriedWhileTheParentEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry parent = addParentEntry("addWhoseParentSearchCanNotRun");
+ final String parentUUID = getEntryUUID(parent.getName());
+ final Entry child = makeChildEntry("addedWhileTheParentSearchFailed", parent.getName());
+ final CSN csn = gen.newCSN();
+ final long resolvedConflicts = getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS);
+ final long unresolvedConflicts = getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS);
+
+ /*
+ * The first search of the first attempt - the check for an Add replayed already - is
+ * let through, the parent check right after it fails, and every search after that
+ * one is served. A parent search read as a parent which is gone hands the Add to
+ * conflict resolution, whose own search finds the parent where it was and rewrites
+ * the message to the DN the Add already carries: the entry lands where it should
+ * either way, and what tells the two apart is the naming conflict counted for a
+ * search which read nothing.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1, 1);
+ try
+ {
+ replayMsg(addMsg(child, csn, parentUUID, "1c3c2c4d-2b5e-4b9f-8a7c-3d5e6f7a8b9c"));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertTrue(entryExists(child.getName()),
+ "the entry was not added under its parent: a parent search which could not run "
+ + "was read as a parent which is gone");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ assertEquals(getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS), resolvedConflicts,
+ "a search which did not run is not a naming conflict which was resolved");
+ assertEquals(getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS), unresolvedConflicts,
+ "a search which did not run is not a naming conflict which could not be resolved");
+ }
+
+ /**
+ * Test case for [Issue 956]: the search conflict resolution reads the data with once a
+ * replayed Add failed on a genuine conflict fails, and the checks before the Add ran.
+ *
+ * The parent of the entry was renamed here, so the Add carries a DN which is not
+ * where the parent is anymore: a conflict which is solved by adding the entry under
+ * the parent's current DN, once the search which finds that DN runs. A search which
+ * did not run is not a parent which is gone, and the conflict is counted once, when
+ * it is solved - not for the search which read nothing.
+ */
+ @Test
+ public void addIsRetriedWhileTheConflictResolutionSearchCanNotRun() throws Exception
+ {
+ final Entry parent = addParentEntry("addWhoseConflictSearchCanNotRun");
+ final String parentUUID = getEntryUUID(parent.getName());
+ final Entry child = makeChildEntry("addedWhileTheConflictSearchFailed", parent.getName());
+ final String entryUUID = "2d4d3d5e-3c6f-4ca0-9b8d-4e6f7a8b9cad";
+ final CSN csn = gen.newCSN();
+
+ final RDN renamedParentRDN = RDN.valueOf("ou=renamedBeforeTheAdd");
+ final ModifyDNOperation renameParent =
+ getRootConnection().processModifyDN(parent.getName(), renamedParentRDN, true);
+ assertEquals(renameParent.getResultCode(), ResultCode.SUCCESS);
+ final DN expectedDN = baseDN.child(renamedParentRDN).child(child.getName().rdn());
+ final long resolvedConflicts = getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS);
+ final long unresolvedConflicts = getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS);
+
+ /*
+ * The two searches which check the Add before it runs are let through - they find
+ * the parent under its new DN, which is what fails the Add on a conflict - and the
+ * search conflict resolution then reads the data with is the one which fails. The
+ * next attempt has every search served.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2, 1);
+ try
+ {
+ replayMsg(addMsg(child, csn, parentUUID, entryUUID));
+ assertShortCircuitSpentBy(3);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertTrue(entryExists(expectedDN),
+ "the entry was not added under the current DN of its parent: a search which could "
+ + "not run was read as a parent which is gone");
+ // A parent read as gone puts the entry under the base DN as a conflicting entry, with
+ // its entryUUID added to its RDN.
+ assertFalse(entryExists(DN.valueOf(
+ "entryuuid=" + entryUUID + "+" + child.getName().rdn() + "," + TEST_ROOT_DN_STRING)),
+ "the entry was renamed under the base DN as a conflicting entry");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ assertEquals(getMonitorAttrValue(baseDN, RESOLVED_NAMING_CONFLICTS), resolvedConflicts + 1,
+ "the renamed parent is one naming conflict, solved once the search ran");
+ assertEquals(getMonitorAttrValue(baseDN, UNRESOLVED_NAMING_CONFLICTS), unresolvedConflicts,
+ "a search which did not run is not a naming conflict which could not be resolved");
+ }
+
+ /**
+ * Test case for [Issue 956]: a replayed Delete reads the data with the same search
+ * once it failed on a conflict, and rides on the same retry.
+ *
+ * The entry was renamed here, so the Delete carries a DN which is not the entry's
+ * anymore, and only the entryUUID search finds it. Reading the search's failure as an
+ * entry which was deleted already answers NOTHING_TO_DO: the entry stays, and the
+ * change is recorded as replayed.
+ */
+ @Test
+ public void deleteIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("deleteWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2);
+ try
+ {
+ replayMsg(new DeleteMsg(staleDN, csn, entryUUID));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(entryExists(entry.getName()),
+ "the entry was not deleted: a search which could not run was read as an entry "
+ + "which was deleted already");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: a replayed Modify DN reads the data with the same search
+ * once it failed on a conflict, and rides on the same retry.
+ *
+ * {@code solveNamingConflict(ModifyDNOperation)} declares {@code throws Exception}
+ * rather than the search failure alone, so this is the one path where the failure
+ * reaches the replay loop through a declaration which does not name it.
+ */
+ @Test
+ public void modifyDnIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyDnWhoseSearchCanNotRun");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final RDN newRDN = RDN.valueOf("cn=renamedWhileTheSearchFailed");
+ final CSN csn = gen.newCSN();
+
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2);
+ try
+ {
+ replayMsg(new ModifyDNMsg(staleDN, csn, entryUUID, null, false, null, newRDN.toString()));
+ assertShortCircuitSpentBy(2);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertTrue(entryExists(baseDN.child(newRDN)),
+ "the entry was not renamed: a search which could not run was read as an entry "
+ + "which is not in the data anymore");
+ assertFalse(entryExists(entry.getName()), "the entry kept its former DN");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956], the half which closes [Issue 889] for this path: a
+ * change whose entryUUID search never runs is left out of the ServerState once the
+ * attempts in place are spent, so that the replication server sends it again.
+ *
+ * The result code of every attempt is the conflict the operation failed on, which the
+ * exhaustion exit does not read as a failure of the server: without the attempt itself
+ * telling that its search did not run, the exit reads the attempts as conflict
+ * resolution rewriting an operation which keeps failing, and skips the change - the
+ * CSN is committed, and a change which is not in the data is never asked for again.
+ */
+ @Test
+ public void modifyIsLeftOutOfTheServerStateWhenTheEntryUUIDSearchNeverRuns() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseSearchNeverRuns");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ // No number of times: every attempt in place fails its search.
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", "01 02 45"), entryUUID));
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse")
+ >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS,
+ "every attempt in place must have made its search");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(domain.getServerState().cover(csn),
+ "a change whose search never ran is not in the data and must not advance the ServerState");
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("telephonenumber"))
+ .as("the change was applied to the entry the searches never found").isEmpty();
+ /*
+ * The result code of the last attempt is the conflict the operation failed on, which
+ * says nothing of the search: the line which reports the change must carry the search
+ * which did not run, which is the one thing that names the entryUUID it was made for.
+ */
+ final List
+ * The first attempt reaches the data, fails on the conflict, and its search does not
+ * run; the server then refuses every attempt after it before the data is reached. Each
+ * of these is a failure of the server, and the change is left out of the ServerState
+ * either way - what the line which reports it carries is the question. Its result
+ * code is the last attempt's, and so must be the error next to it: a line which reads
+ * the server refusing the operation next to a search which did not run names two
+ * causes, and the operator chases the wrong one.
+ */
+ @Test
+ public void theExhaustionExitReportsTheAttemptWhichSpentTheLastOfThem() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseLastAttemptIsRefused");
+ final String entryUUID = getEntryUUID(entry.getName());
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ /*
+ * The first attempt is let through to the data and fails on the conflict of the
+ * stale DN; its search is the one the short circuit stops. The attempts after it are
+ * refused before they reach the data, the way a backend which went offline after the
+ * first attempt refuses them: no search is made on any of them.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1);
+ // The replayed operation only, named by its CSN: see the flush below.
+ ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse",
+ ResultCode.UNAVAILABLE.intValue(), 1, LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS - 1,
+ op -> csn.equals(OperationContext.getCSN(op)));
+ try
+ {
+ /*
+ * The ServerState flush thread saves the state with a Modify of the base entry on
+ * its tick, which the add above made dirty: a tick which lands among the attempts in
+ * place is a Modify the short circuit meets like any other. Made here rather than
+ * left to the tick, so that the case says what it does about it every time instead
+ * of once in a while: that Modify is not the replayed operation, and the short
+ * circuit must neither let it through in place of the first attempt nor refuse it
+ * in place of a later one.
+ */
+ flushLikeTheStateFlushThread();
+ assertEquals(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse"), 0,
+ "the Modify of the state flush thread was counted against the short circuit");
+
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", "01 02 45"), entryUUID));
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") >= 1,
+ "the first attempt must have made its search");
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse")
+ >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS,
+ "every attempt in place must have been made");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertFalse(domain.getServerState().cover(csn),
+ "a change the server kept refusing is not in the data and must not advance the ServerState");
+ final List
+ * Two empty replicas share the generation ID of an empty backend, so no initialization
+ * is needed and the first change replayed is the base entry itself. The searches which
+ * check that Add for a conflict run under the base DN, which the backend serves and
+ * has no entry for: they answer NO_SUCH_OBJECT, which is the answer of a backend which
+ * is offline as well. Here the search ran and nothing is below a base entry which is
+ * not there, so the Add must go through rather than be retried until the give-up
+ * budget skips it as a change this replica can not apply.
+ */
+ @Test
+ public void baseEntryIsAddedToAnEmptyReplica() throws Exception
+ {
+ // The backend, without its base entry.
+ TestCaseUtils.initializeTestBackend(false);
+ final Entry base = TestCaseUtils.makeEntry(
+ "dn: " + TEST_ROOT_DN_STRING,
+ "objectClass: top",
+ "objectClass: organization",
+ "o: test");
+ final CSN csn = gen.newCSN();
+
+ replayMsg(addMsg(base, csn, null, "7c1a0d2e-4b6f-4c8a-9e1d-3f5b7a9c1e2d"));
+
+ assertTrue(entryExists(base.getName()),
+ "the base entry of an empty replica must land: its searches found nothing, they did not fail");
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was applied must be recorded as replayed");
+ }
+
+ /**
+ * Test case for [Issue 956]: the entryUUID a change carries is looked up as a value,
+ * not read as a filter.
+ *
+ * The entryUUID comes off the wire and nothing validates it as one. Built into a
+ * filter string, a value which does not parse as a filter is a search which never
+ * runs - a permanent condition retried as a transient one, for as long as the change
+ * is asked for, until the give-up budget skips it and raises an alert. Looked up as a
+ * value, such an entryUUID names no entry, which is what a search which ran and found
+ * nothing says: the change is resolved as one on an entry which is not in the data.
+ */
+ @Test
+ public void anEntryUUIDWhichIsNotOneNamesNoEntry() throws Exception
+ {
+ final Entry entry = createAndAddEntry("modifyWhoseEntryUUIDIsNotOne");
+ // A value no filter string parses: the backslash escapes nothing.
+ final String entryUUID = getEntryUUID(entry.getName()) + "\\";
+ final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING);
+ final CSN csn = gen.newCSN();
+
+ replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", "01 02 45"), entryUUID));
+
+ assertTrue(domain.getServerState().cover(csn),
+ "a change on an entry which is not in the data is a conflict which is resolved, and recorded");
+ assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("telephonenumber"))
+ .as("a change on an entryUUID which is not one was applied to an entry").isEmpty();
+ }
+
+ /**
+ * The records of the error log which report the provided change once its attempts in
+ * place were spent. The record the error logger writes carries the id of the message
+ * rather than its text.
+ */
+ private static List
+ * The count includes the searches let through once the short circuit was spent, so a
+ * count past what it was registered over says that the budget was used and that the
+ * search after it ran. Read before the short circuit is deregistered, which drops the
+ * count with it.
+ *
+ * @param searchesRegisteredOver the searches let through before the short circuit plus
+ * the ones it applied to
+ */
+ private void assertShortCircuitSpentBy(int searchesRegisteredOver)
+ {
+ assertTrue(
+ ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") > searchesRegisteredOver,
+ "the short circuit must have been spent by the attempts in place");
+ }
+
+ private Entry addParentEntry(String ou) throws Exception
+ {
+ return TestCaseUtils.addEntry(
+ "dn: ou=" + ou + "," + TEST_ROOT_DN_STRING,
+ "objectClass: top",
+ "objectClass: organizationalUnit",
+ "ou: " + ou);
+ }
+
+ private Entry makeChildEntry(String cn, DN parentDN) throws Exception
+ {
+ return TestCaseUtils.makeEntry(
+ "dn: cn=" + cn + "," + parentDN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: " + cn,
+ "sn: Amar");
+ }
+
/**
* Test that when a previous conflict is resolved because
* a delete operation has removed one of the conflicting entries
@@ -317,6 +854,24 @@ public void conflictCleaningMODDN() throws Exception
assertThat(resultEntry.getAllAttributes(LDAPReplicationDomain.DS_SYNC_CONFLICT)).isEmpty();
}
+ /**
+ * Makes the Modify the ServerState flush thread makes on its tick: an internal
+ * synchronization Modify of the base entry, which is not synchronized itself. The
+ * attribute is a harmless one rather than ds-sync-state, which is the flush thread's to
+ * write.
+ */
+ private void flushLikeTheStateFlushThread()
+ {
+ final ModifyOperationBasis op = new ModifyOperationBasis(getRootConnection(),
+ nextOperationID(), nextMessageID(), null,
+ baseDN, generatemods("description", "written on the tick of the flush thread"));
+ op.setInternalOperation(true);
+ op.setSynchronizationOperation(true);
+ op.setDontSynchronize(true);
+ op.run();
+ assertEquals(op.getResultCode(), ResultCode.SUCCESS, op.getErrorMessage().toString());
+ }
+
private Entry createAndAddEntry(String commonName) throws Exception
{
// @formatter:off
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
new file mode 100644
index 0000000000..3bedecb489
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -0,0 +1,385 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions copyright [year] [name of copyright owner]".
+ *
+ * Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.replication.plugin;
+
+import static java.nio.charset.StandardCharsets.*;
+import static org.assertj.core.api.Assertions.*;
+import static org.opends.messages.ReplicationMessages.*;
+import static org.opends.server.TestCaseUtils.*;
+import static org.opends.server.core.DirectoryServer.*;
+import static org.testng.Assert.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.ResultCode;
+import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.core.DirectoryServer;
+import org.opends.server.plugins.ShortCircuitPlugin;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.protocol.DoneMsg;
+import org.opends.server.replication.protocol.EntryMsg;
+import org.opends.server.replication.protocol.InitializeRequestMsg;
+import org.opends.server.replication.protocol.InitializeTargetMsg;
+import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ModifyMsg;
+import org.opends.server.replication.protocol.UpdateMsg;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.types.Entry;
+import org.opends.server.types.OperationType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Tests the replay of a change while this replica is the target of a total update.
+ *
+ * The import of a total update streams over the session of the domain, on its listener
+ * thread, and the backend it replaces is deregistered for the length of it. A change which
+ * was queued for replay before the {@code InitializeTargetMsg} arrived is replayed into no
+ * backend: whatever such a replay decides is about to be overwritten by the import, and the
+ * 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.
+ *
+ * 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.
+ */
+@SuppressWarnings("javadoc")
+public class ReplayDuringImportTest extends ReplicationTestCase
+{
+ /**
+ * The memory backend of {@code o=test} loses its data when it is disabled and enabled
+ * back, which is what an import does to the backend it replaces: a total update needs a
+ * backend which keeps what was imported into it.
+ */
+ private static final String EXAMPLE_DN = "dc=example,dc=com";
+ private static final int RS_ID = 611;
+ private static final int DS_ID = 1;
+ private static final int EXPORTER_ID = 2;
+ private static final int INIT_WINDOW = 100;
+ private static final AtomicBoolean SHUTDOWN = new AtomicBoolean(false);
+ /** An entry of the exporter's data, and its entryUUID. */
+ private static final String IMPORTED_ENTRY_DN = "cn=imported,ou=People," + EXAMPLE_DN;
+ private static final String IMPORTED_ENTRY_UUID = "21111111-1111-1111-1111-111111111113";
+
+ private DN baseDN;
+ private ReplicationServer replicationServer;
+ private LDAPReplicationDomain domain;
+ private TestSynchronousReplayQueue queue;
+ private ReplicationBroker exporter;
+ private CSNGenerator gen;
+
+ @BeforeMethod
+ public void setUpLocal() throws Exception
+ {
+ baseDN = DN.valueOf(EXAMPLE_DN);
+ TestCaseUtils.clearBackend("userRoot", EXAMPLE_DN);
+
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ rsPort, "replayDuringImportTestDb", 0, RS_ID, 0, 100, new TreeSet
+ * The change is given back at the top of its first attempt: the data it would be applied
+ * to is being replaced, so nothing is attempted into the backend the import took away,
+ * nothing is reported, and the session is left to the import - which streams every entry
+ * to its end. Without the hold-off the operation is refused with NO_SUCH_OBJECT - nothing
+ * serves the base DN - and the entryUUID search conflict resolution reads the data with
+ * can not run either: the attempts in place are spent into no backend and the exit
+ * reports the change; without the owner the total update is, the session is then
+ * restarted for the change to be delivered again, which stops the broker the import is
+ * reading, and the import ends on the entries which had arrived with nothing to say it.
+ */
+ @Test(timeOut = 120_000)
+ public void aReplayDuringTheImportLeavesTheSessionToTheImport() 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 String[] exported = exportedEntries();
+ startImportInto(exported.length);
+
+ // Queued before the InitializeTargetMsg arrived, replayed into no backend.
+ final CSN csn = gen.newCSN();
+ replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "replayed during the import"), entryUUID));
+
+ finishImport(exported);
+
+ for (String ldif : exported)
+ {
+ final DN dn = dnOf(ldif);
+ assertTrue(entryExists(dn), "the import ended before " + dn
+ + " arrived: the session it streams over was stopped from under it");
+ }
+ /*
+ * The two roads which leave the session to the import are told apart here: the
+ * hold-off gives the change back before an attempt is made, the guard on the restart
+ * after the attempts are spent. The exhaustion exit is the one thing the first road
+ * leaves no record of.
+ */
+ assertThat(errorLogRecordsOf(ERR_ERROR_REPLAYING_OPERATION.ordinal(), csn))
+ .as("the change was attempted into no backend instead of being given back at once")
+ .isEmpty();
+ assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn))
+ .as("the change was asked for again, which restarts the session the import streams over")
+ .isEmpty();
+ }
+
+ /**
+ * A change given back while the import ran must not hold the ServerState back once the
+ * import has replaced the data.
+ *
+ * A change which is given back stays listed as pending and uncommitted - that is what
+ * has the replication server send it again - and a commit advances the ServerState no
+ * further than the oldest uncommitted change. The state the import loads is the
+ * exporter's, which covers the change already, so nothing sends it again: left listed,
+ * it would stop the ServerState of this replica for good.
+ */
+ @Test(timeOut = 120_000)
+ public void aChangeGivenBackDuringTheImportDoesNotHoldTheServerStateBack() 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 String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ replayMsg(new ModifyMsg(gen.newCSN(), DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "replayed during the import"), entryUUID));
+ finishImport(exported);
+
+ // A change on an entry the import brought, replayed once the import is over.
+ final DN importedDN = DN.valueOf(IMPORTED_ENTRY_DN);
+ final CSN csn = gen.newCSN();
+ replayMsg(new ModifyMsg(csn, importedDN,
+ generatemods("description", "replayed after the import"), IMPORTED_ENTRY_UUID));
+
+ assertThat(DirectoryServer.getEntry(importedDN).getAllAttributes("description"))
+ .as("a change replayed after the import was not applied").isNotEmpty();
+ assertTrue(domain.getServerState().cover(csn), "a change applied after the import was not"
+ + " recorded: the change given back during the import is still listed and holds the"
+ + " ServerState back");
+ }
+
+ /**
+ * A total update this replica asked for owns the session from the request, not from the
+ * first entry: the {@code InitializeTargetMsg} which answers the request arrives over
+ * that session, and a restart made while the answer is on its way loses it.
+ *
+ * The backend is live for the length of the request - nothing has been taken away yet -
+ * so the change is attempted, every attempt ends on an entryUUID search which does not
+ * run, and the exhaustion exit reports it: what is refused is the restart which would
+ * have followed, and the retry warning which goes with it. The exporter then answers the
+ * request, and its entries stream to their end over the session which was left alone.
+ */
+ @Test(timeOut = 120_000)
+ public void aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver() 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 String[] exported = exportedEntries();
+
+ // The request is out, and the exporter holds it until the change below has been replayed.
+ domain.initializeFromRemote(EXPORTER_ID, null);
+ assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));
+
+ final CSN csn = gen.newCSN();
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
+ try
+ {
+ replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
+ generatemods("description", "replayed while the request was on its way"), entryUUID));
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse")
+ >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS,
+ "every attempt in place must have made its search: the backend is live while the"
+ + " request is on its way, so nothing holds the replay off");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ }
+
+ assertThat(errorLogRecordsOf(ERR_ERROR_REPLAYING_OPERATION.ordinal(), csn))
+ .as("the attempts in place were spent, which the exhaustion exit reports").isNotEmpty();
+ assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn))
+ .as("the change was asked for again, which restarts the session the answer to the"
+ + " request arrives over")
+ .isEmpty();
+ assertTrue(domain.isConnected(), "the session the request was made over was stopped");
+
+ answerImportRequest(exported.length);
+ finishImport(exported);
+ for (String ldif : exported)
+ {
+ final DN dn = dnOf(ldif);
+ assertTrue(entryExists(dn), "the import ended before " + dn
+ + " arrived: the answer to the request was lost with the session it was made over");
+ }
+ }
+
+ /**
+ * Has the exporter start a total update into this replica, and returns once the backend
+ * of the domain is deregistered for it: from then on the import is reading the session,
+ * and a change replayed here is replayed into no backend.
+ */
+ private void startImportInto(int entryCount) throws Exception
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, entryCount, INIT_WINDOW));
+ final long deadline = System.currentTimeMillis() + 30_000;
+ while (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) != null)
+ {
+ assertTrue(System.currentTimeMillis() < deadline,
+ "the import did not deregister the backend of the domain");
+ Thread.sleep(20);
+ }
+ }
+
+ /**
+ * Has the exporter answer the total update this replica asked for: the requestor of the
+ * {@code InitializeTargetMsg} is this replica, so the import runs in the context the
+ * request acquired.
+ */
+ private void answerImportRequest(int entryCount) throws Exception
+ {
+ exporter.publish(new InitializeTargetMsg(
+ baseDN, EXPORTER_ID, DS_ID, DS_ID, entryCount, INIT_WINDOW));
+ }
+
+ /** Has the exporter send the entries of the total update, and waits for the import to end. */
+ private void finishImport(String... ldifEntries) throws Exception
+ {
+ int msgId = 0;
+ for (String ldif : ldifEntries)
+ {
+ exporter.publish(new EntryMsg(EXPORTER_ID, DS_ID, ldif.getBytes(UTF_8), ++msgId));
+ }
+ exporter.publish(new DoneMsg(EXPORTER_ID, DS_ID));
+ final long deadline = System.currentTimeMillis() + 60_000;
+ while (domain.ieRunning())
+ {
+ assertTrue(System.currentTimeMillis() < deadline, "the import did not end");
+ Thread.sleep(50);
+ }
+ }
+
+ /** The data of the exporter: the base entry and two entries below it. */
+ private static String[] exportedEntries()
+ {
+ return new String[] {
+ "dn: " + EXAMPLE_DN + "\n"
+ + "objectClass: top\n"
+ + "objectClass: domain\n"
+ + "dc: example\n"
+ + "entryUUID: 21111111-1111-1111-1111-111111111111\n"
+ + "\n",
+ "dn: ou=People," + EXAMPLE_DN + "\n"
+ + "objectClass: top\n"
+ + "objectClass: organizationalUnit\n"
+ + "ou: People\n"
+ + "entryUUID: 21111111-1111-1111-1111-111111111112\n"
+ + "\n",
+ "dn: " + IMPORTED_ENTRY_DN + "\n"
+ + "objectClass: top\n"
+ + "objectClass: person\n"
+ + "cn: imported\n"
+ + "sn: imported\n"
+ + "entryUUID: " + IMPORTED_ENTRY_UUID + "\n"
+ + "\n",
+ };
+ }
+
+ private static DN dnOf(String ldif)
+ {
+ return DN.valueOf(ldif.substring("dn: ".length(), ldif.indexOf('\n')));
+ }
+
+ /** The records of the error log which carry the provided message id and the provided CSN. */
+ private static List