From cdd887a0d9b5cd04114999230d71a96b1f89a584 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 8 Sep 2026 20:45:22 +0300 Subject: [PATCH 1/6] [#956] Tell a failed entryUUID search apart from an entry which is not there solveNamingConflict() decides an entry is gone by searching for its entryUUID and getting nothing back, and getFirstResult() answers the same thing for a search which found nothing and for a search which never ran. Every caller read that as "the entry has been deleted", which answers NOTHING_TO_DO - and that branch commits the CSN unconditionally, without the guard #892 gave the FAILED branch next to it. A change which was never applied was recorded as replayed, the replication server never sent it again, and no alert was raised: the #889 failure mode through a branch #892 did not harden. findEntryDN() now reports a search which did not run rather than answering "no entry" out of it, and the replay takes that as the failure of the server it is: the change is retried in place and left out of the ServerState once the attempts are spent, so the replication server delivers it again. The searches which check a replayed Add for a conflict before it runs get the same treatment - reading them as a parent which is gone renamed the entry as a conflicting one, which an administrator has to repair by hand. Fixes #956 --- .../plugin/LDAPReplicationDomain.java | 198 ++++++++++++++---- .../opends/messages/replication.properties | 4 + .../plugin/NamingConflictTest.java | 105 ++++++++++ 3 files changed, 270 insertions(+), 37 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 537e70c0dc..a9530219bd 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -1849,7 +1849,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 +1875,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 +1910,27 @@ 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: 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). + * + * @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) + { + final LocalizableMessage message = WARN_REPLAY_ENTRYUUID_SEARCH_FAILED.get( + csn, getBaseDN(), e.entryUUID, e.getMessage()); + logger.warn(message); + return new SynchronizationProviderResult.StopProcessing(ResultCode.UNAVAILABLE, message); + } + /** * Check that the broker associated to this ReplicationDomain has found * a Replication Server and that this LDAP server is therefore able to @@ -2730,6 +2768,12 @@ private void replayChangeAndTheChangesWaitingForIt( dependency = remotePendingChanges.checkDependencies(op, msg); boolean replayDone = false; boolean firstAttempt = true; + /* + * Whether the last attempt ended on a search conflict resolution could not run: + * 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. + */ + boolean searchFailedResolvingConflict = false; int retryCount = IN_PLACE_REPLAY_ATTEMPTS; while (!dependency && !replayDone && retryCount-- > 0) { @@ -2845,34 +2889,50 @@ else if (isServerFailure(result, serverErrorResultCode)) else { ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO; - 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) + try { - 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. + */ + logger.warn(WARN_REPLAY_ENTRYUUID_SEARCH_FAILED, + csn, getBaseDN(), e.entryUUID, e.getMessage()); + resolution = ConflictResolution.SEARCH_FAILED; } - // else: unknown type of operation ?! there is nothing to replay if (!dependency) { @@ -2886,6 +2946,20 @@ 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). + */ + searchFailedResolvingConflict = true; + Thread.sleep(50); + break; + case FAILED: if (serverErrorResultCode.equals(result)) { @@ -2924,6 +2998,7 @@ else if (op instanceof ModifyDNOperation) * reflecting the new state of the UpdateMsg after conflict resolution * modified it, and dependencies might have been replayed by now. */ + searchFailedResolvingConflict = false; break; } } @@ -2980,11 +3055,15 @@ 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) { /* * The server kept failing to apply the change, so the change is not in the data. @@ -3782,24 +3861,36 @@ 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 { + final InternalSearchOperation search; try { final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, "entryuuid=" + uuid); - InternalSearchOperation search = conn.processSearch(request); - final SearchResultEntry resultEntry = getFirstResult(search); - if (resultEntry != null) - { - return resultEntry.getName(); - } + search = conn.processSearch(request); } catch (DirectoryException e) { - // never happens because the filter is always valid. + /* + * The filter did not parse, so no search ever ran: the entryUUID it is built from + * comes off the wire and nothing validates it as one. + */ + throw new SearchFailedException(uuid, e.getResultCode().getName() + " " + e.getMessageObject()); } - return null; + if (search.getResultCode() != ResultCode.SUCCESS) + { + /* + * 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; } /** Outcome of the conflict resolution attempted after a replayed operation failed. */ @@ -3809,18 +3900,49 @@ 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; + } + } + /** * 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 +4027,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); 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..912a50a6d1 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 +WARN_REPLAY_ENTRYUUID_SEARCH_FAILED_310=Could not read the data to solve a conflict change %s ran \ + into in domain "%s": the search of the entry with entryUUID %s did not run (%s). The change is \ + being attempted again rather than resolved against what a search which read nothing seemed to \ + say about the data diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java index cb74306e97..eb1ffd2401 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java @@ -33,6 +33,7 @@ import org.opends.server.TestCaseUtils; import org.opends.server.core.DirectoryServer; import org.opends.server.core.ModifyDNOperation; +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; @@ -40,8 +41,10 @@ import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.protocol.LDAPUpdateMsg; import org.opends.server.replication.protocol.ModifyDNMsg; +import org.opends.server.replication.protocol.ModifyMsg; import org.opends.server.replication.protocol.UpdateMsg; 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; @@ -244,6 +247,108 @@ public void modifyDnOnAnEntryAndANewSuperiorWhichAreBothGone() throws Exception "a ModifyDN which the delete of its entry has settled must be recorded as replayed"); } + /** + * Test case for [Issue 956]: conflict resolution reads the data with a search of the + * entryUUID, and a search which did not run is no evidence about the data - the entry + * it did not report is not an entry which was deleted. + *

+ * 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)); + } + 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 - a parent it did not report is not a parent which was deleted. + *

+ * Reading them that way renames the entry under the base DN as a conflicting entry: + * a divergence which is left for an administrator to repair by hand, where the entry + * only had to be added again once the storage served the searches. + */ + @Test + public void addIsRetriedWhileTheEntryUUIDSearchesCanNotRun() throws Exception + { + final Entry parent = TestCaseUtils.addEntry( + "dn: ou=addWhoseSearchesCanNotRun," + TEST_ROOT_DN_STRING, + "objectClass: top", + "objectClass: organizationalUnit", + "ou: addWhoseSearchesCanNotRun"); + final String parentUUID = getEntryUUID(parent.getName()); + + final Entry child = TestCaseUtils.makeEntry( + "dn: cn=addedWhileTheSearchesFailed," + parent.getName(), + "objectClass: top", + "objectClass: person", + "cn: addedWhileTheSearchesFailed", + "sn: Amar"); + final CSN csn = gen.newCSN(); + + /* + * Every search this attempt makes fails: the one which checks whether the Add was + * replayed here already, the one which checks that the parent is still the one the + * change was made under, and the one conflict resolution reads the data with once + * the Add failed. The next attempt has them all served again. + */ + ShortCircuitPlugin.registerShortCircuit( + OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 3); + try + { + replayMsg(addMsg(child, csn, parentUUID, "0b2b1b3c-1a4d-4a8e-9f6b-2c4d5e6f7a8b")); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse"); + } + + assertTrue(entryExists(child.getName()), + "the entry was not added under its parent: searches which could not run were read " + + "as a parent which is gone"); + assertTrue(domain.getServerState().cover(csn), + "a change which was applied must be recorded as replayed"); + } + /** * Test that when a previous conflict is resolved because * a delete operation has removed one of the conflicting entries From 48a6c57fcec493a9d9e3fc4f5f33b213b9604868 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 13:53:36 +0300 Subject: [PATCH 2/6] [#956] Take an ordinal in replication.properties nothing else claims --- .../src/messages/org/opends/messages/replication.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 912a50a6d1..0790d68fa7 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -686,7 +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 -WARN_REPLAY_ENTRYUUID_SEARCH_FAILED_310=Could not read the data to solve a conflict change %s ran \ +WARN_REPLAY_ENTRYUUID_SEARCH_FAILED_322=Could not read the data to solve a conflict change %s ran \ into in domain "%s": the search of the entry with entryUUID %s did not run (%s). The change is \ being attempted again rather than resolved against what a search which read nothing seemed to \ say about the data From b82d8fffbd8a51b81eca2ff2df70551abde33560 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 11 Sep 2026 15:40:55 +0300 Subject: [PATCH 3/6] [#956] Let the base entry of an empty replica land, and look the entryUUID up as a value Review round 1 of #968. A backend which serves the base DN and has no base entry answers NO_SUCH_OBJECT to a search under it, on every route - and so does a backend which is not there. findEntryDN() read every non-SUCCESS code as a search which did not run, so the base entry of a domain replayed into an empty replica was retried until the give-up budget skipped it: two empty replicas share the generation ID of an empty backend, and the base entry is the first change. baseEntryIsAbsentFromALiveBackend() asks the backend which of the two it is. The entryUUID is looked up as the value it is rather than read as part of a filter string: it comes off the wire and nothing validates it as one, and a value which does not parse as a filter was a search which never runs, retried as a transient failure for as long as the change was asked for. There is no filter to parse now, and no branch left. Nothing is logged per attempt in place any more, as nothing is for a storage which failed to serve the operation: the attempt keeps the search failure and the exhaustion exit reports it in the ERR_ERROR_REPLAYING_OPERATION line it already logs, in place of the error of the operation, which for this case only named the conflict. The Add hook puts the same text on the operation it stops. The message is ERR_REPLAY_ENTRYUUID_SEARCH_FAILED now that it is never logged on its own; the ordinal stays. Tests: the exhaustion exit is pinned by a case whose search never runs; Delete and ModifyDN get a case each; the Add hook is three cases, one per search, on a ShortCircuitPlugin which can let the first searches through before it applies; every bounded short circuit asserts, before it is deregistered, that its budget was spent and the search after it ran; the base entry of an empty replica and an entryUUID no filter string parses each get a case. --- .../plugin/LDAPReplicationDomain.java | 125 ++++-- .../opends/messages/replication.properties | 8 +- .../server/plugins/ShortCircuitPlugin.java | 39 +- .../plugin/NamingConflictTest.java | 371 ++++++++++++++++-- 4 files changed, 479 insertions(+), 64 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index a9530219bd..5dbf0ff75d 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -1915,9 +1915,12 @@ SynchronizationProviderResult handleConflictResolution( * 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: 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). + * 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 @@ -1925,10 +1928,8 @@ SynchronizationProviderResult handleConflictResolution( */ private SynchronizationProviderResult searchDidNotRun(CSN csn, SearchFailedException e) { - final LocalizableMessage message = WARN_REPLAY_ENTRYUUID_SEARCH_FAILED.get( - csn, getBaseDN(), e.entryUUID, e.getMessage()); - logger.warn(message); - return new SynchronizationProviderResult.StopProcessing(ResultCode.UNAVAILABLE, message); + return new SynchronizationProviderResult.StopProcessing( + ResultCode.UNAVAILABLE, e.report(csn, getBaseDN())); } /** @@ -2769,11 +2770,12 @@ private void replayChangeAndTheChangesWaitingForIt( boolean replayDone = false; boolean firstAttempt = true; /* - * Whether the last attempt ended on a search conflict resolution could not run: - * 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. + * 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. */ - boolean searchFailedResolvingConflict = false; + SearchFailedException searchFailedResolvingConflict = null; int retryCount = IN_PLACE_REPLAY_ATTEMPTS; while (!dependency && !replayDone && retryCount-- > 0) { @@ -2927,10 +2929,12 @@ else if (op instanceof ModifyDNOperation) * 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. + * 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. */ - logger.warn(WARN_REPLAY_ENTRYUUID_SEARCH_FAILED, - csn, getBaseDN(), e.entryUUID, e.getMessage()); + searchFailedResolvingConflict = e; resolution = ConflictResolution.SEARCH_FAILED; } @@ -2956,7 +2960,6 @@ else if (op instanceof ModifyDNOperation) * below the loop reports and acts on. A change which is not in the data * must not advance the ServerState (issue #889). */ - searchFailedResolvingConflict = true; Thread.sleep(50); break; @@ -2998,7 +3001,7 @@ else if (op instanceof ModifyDNOperation) * reflecting the new state of the UpdateMsg after conflict resolution * modified it, and dependencies might have been replayed by now. */ - searchFailedResolvingConflict = false; + searchFailedResolvingConflict = null; break; } } @@ -3063,16 +3066,23 @@ else if (op instanceof ModifyDNOperation) if (isServerFailure(lastResult, serverErrorResultCode) || ResultCode.BUSY.equals(lastResult) || serverErrorResultCode.equals(lastResult) - || searchFailedResolvingConflict) + || 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; @@ -3866,22 +3876,34 @@ private static SearchResultEntry getFirstResult(InternalSearchOperation search) */ private DN findEntryDN(String uuid) throws SearchFailedException { - final InternalSearchOperation search; - try - { - final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, "entryuuid=" + uuid); - search = conn.processSearch(request); - } - catch (DirectoryException e) + if (uuid == null) { - /* - * The filter did not parse, so no search ever ran: the entryUUID it is built from - * comes off the wire and nothing validates it as one. - */ - throw new SearchFailedException(uuid, e.getResultCode().getName() + " " + e.getMessageObject()); + // 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()) + { + /* + * 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 @@ -3893,6 +3915,37 @@ private DN findEntryDN(String uuid) throws SearchFailedException 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) + { + // The storage did not answer this any more than it answered the search. + return false; + } + } + /** Outcome of the conflict resolution attempted after a replayed operation failed. */ private enum ConflictResolution { @@ -3931,6 +3984,18 @@ 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()); + } } /** 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 0790d68fa7..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,7 +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 -WARN_REPLAY_ENTRYUUID_SEARCH_FAILED_322=Could not read the data to solve a conflict change %s ran \ - into in domain "%s": the search of the entry with entryUUID %s did not run (%s). The change is \ - being attempted again rather than resolved against what a search which read nothing seemed to \ - say about the data +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..624ec3aa03 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 @@ -647,13 +647,15 @@ private int shortCircuitInternal(PluginOperation operation, String section) if (resultCode != null) { 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,9 @@ public static List createShortCircuitControlList(int resultCode, String /** How many times a registered short circuit must be applied, when it is limited. */ private static final Map shortCircuitLimits = new ConcurrentHashMap<>(); + /** How many operations a registered short circuit lets through before it applies. */ + private static final Map shortCircuitSkips = new ConcurrentHashMap<>(); + /** * Returns how many times the short circuit registered for the given operation type and * plugin point was reached. A short circuit registered for a limited number of times is @@ -765,10 +770,12 @@ public static int getShortCircuitCount(OperationType operation, String section) public static void registerShortCircuit(OperationType operation, String section, int resultCode) { final String key = keyFor(operation, section); - // This registration applies to every operation, and it counts from zero: a limit or - // a count left behind by a previous registration is not part of it. + // This registration applies to every operation, and it counts from zero: a limit, a + // number of operations let through or a count left behind by a previous registration + // is not part of it. shortCircuitCounts.remove(key); shortCircuitLimits.remove(key); + shortCircuitSkips.remove(key); shortCircuits.put(key, resultCode); } @@ -782,9 +789,30 @@ public static void registerShortCircuit(OperationType operation, String section, * @param maxTimes How many operations must be short circuited. */ public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes) + { + registerShortCircuit(operation, section, resultCode, 0, maxTimes); + } + + /** + * Register a short circuit which lets the given number of operations through before it + * applies, then applies to the given number of operations, the ones which follow being + * let through again: this is how a transient failure which starts part way through a + * sequence of operations is simulated - the second search of an attempt failing while + * the first one ran, say. + * + * @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 operations must be let through before the short + * circuit applies. + * @param maxTimes How many operations must be short circuited after them. + */ + public static void registerShortCircuit(OperationType operation, String section, int resultCode, + int letThroughFirst, int maxTimes) { final String key = keyFor(operation, section); shortCircuitCounts.remove(key); + shortCircuitSkips.put(key, letThroughFirst); shortCircuitLimits.put(key, maxTimes); shortCircuits.put(key, resultCode); } @@ -799,6 +827,7 @@ public static void deregisterShortCircuit(OperationType operation, String sectio final String key = keyFor(operation, section); shortCircuits.remove(key); shortCircuitLimits.remove(key); + shortCircuitSkips.remove(key); // The count belongs to the registration which is being removed: a test which counts // the operations it short circuits must not inherit the count of the previous one. shortCircuitCounts.remove(key); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java index eb1ffd2401..c15e28643b 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java @@ -55,6 +55,10 @@ public class NamingConflictTest extends ReplicationTestCase { private static final AtomicBoolean SHUTDOWN = new AtomicBoolean(false); + /** The monitor attributes which count the naming conflicts a domain solved, and did not. */ + private static final String RESOLVED_NAMING_CONFLICTS = "resolved-naming-conflicts"; + private static final String UNRESOLVED_NAMING_CONFLICTS = "unresolved-naming-conflicts"; + private DN baseDN; private LDAPReplicationDomain domain; private CSNGenerator gen; @@ -285,6 +289,7 @@ public void modifyIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception try { replayMsg(new ModifyMsg(csn, staleDN, generatemods("telephonenumber", phoneNumber), entryUUID)); + assertShortCircuitSpentBy(2); } finally { @@ -301,41 +306,84 @@ public void modifyIsRetriedWhileTheEntryUUIDSearchCanNotRun() throws Exception /** * 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 - a parent it did not report is not a parent which was deleted. + * about it either. The first of them checks whether the Add was replayed here already. *

- * Reading them that way renames the entry under the base DN as a conflicting entry: - * a divergence which is left for an administrator to repair by hand, where the entry - * only had to be added again once the storage served the searches. + * 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 addIsRetriedWhileTheEntryUUIDSearchesCanNotRun() throws Exception + public void addIsNotReplayedTwiceWhileTheEntryUUIDSearchCanNotRun() throws Exception { - final Entry parent = TestCaseUtils.addEntry( - "dn: ou=addWhoseSearchesCanNotRun," + TEST_ROOT_DN_STRING, - "objectClass: top", - "objectClass: organizationalUnit", - "ou: addWhoseSearchesCanNotRun"); - final String parentUUID = getEntryUUID(parent.getName()); + 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(); - final Entry child = TestCaseUtils.makeEntry( - "dn: cn=addedWhileTheSearchesFailed," + parent.getName(), - "objectClass: top", - "objectClass: person", - "cn: addedWhileTheSearchesFailed", - "sn: Amar"); + // 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); /* - * Every search this attempt makes fails: the one which checks whether the Add was - * replayed here already, the one which checks that the parent is still the one the - * change was made under, and the one conflict resolution reads the data with once - * the Add failed. The next attempt has them all served again. + * 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(), 3); + OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 1, 1); try { - replayMsg(addMsg(child, csn, parentUUID, "0b2b1b3c-1a4d-4a8e-9f6b-2c4d5e6f7a8b")); + replayMsg(addMsg(child, csn, parentUUID, "1c3c2c4d-2b5e-4b9f-8a7c-3d5e6f7a8b9c")); + assertShortCircuitSpentBy(2); } finally { @@ -343,12 +391,285 @@ public void addIsRetriedWhileTheEntryUUIDSearchesCanNotRun() throws Exception } assertTrue(entryExists(child.getName()), - "the entry was not added under its parent: searches which could not run were read " - + "as a parent which is gone"); + "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 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, "2d4d3d5e-3c6f-4ca0-9b8d-4e6f7a8b9cad")); + 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"); + assertFalse(entryExists(DN.valueOf(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(); + } + + /** + * Test case for [Issue 956]: the base entry of the domain replayed into a replica + * which has none. + *

+ * 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(); + } + + /** + * Asserts that the searches a short circuit was registered over were made - the ones + * let through before it and the ones it applied to - and that the search after them + * was made as well. + *

+ * 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 From 262036ff5cbfaa039ed123209384d8a949ee389d Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 14 Sep 2026 10:46:43 +0300 Subject: [PATCH 4/6] [#956] Report the attempt which spent the last of the attempts in place Review round 2 of #968. The mark an attempt leaves when the search conflict resolution reads the data with did not run outlived the attempt which set it: it was cleared only when conflict resolution rewrote the operation, so an attempt the server refused before it reached the data - BUSY, UNAVAILABLE from the operation itself, or the configured server-error-result-code out of conflict resolution - left it as it was, and the exhaustion exit reported the search of an earlier attempt next to the result code of the last one: two causes on one line. The CSN was left out of the ServerState either way. The mark is reset at the top of every attempt now, which is what the comment above the verdict promised, and the clear on a rewritten operation has nothing left to do. The assertion of addIsRetriedWhileTheConflictResolutionSearchCanNotRun which guards against the entry being renamed as a conflicting entry names the DN that rename produces - the entryUUID added to the RDN, under the base DN - rather than one the code never makes. --- .../plugin/LDAPReplicationDomain.java | 7 +- .../plugin/NamingConflictTest.java | 77 ++++++++++++++++++- 2 files changed, 81 insertions(+), 3 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 5dbf0ff75d..0547a43376 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -2779,6 +2779,12 @@ private void replayChangeAndTheChangesWaitingForIt( int retryCount = IN_PLACE_REPLAY_ATTEMPTS; while (!dependency && !replayDone && retryCount-- > 0) { + /* + * 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 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 @@ -3001,7 +3007,6 @@ else if (op instanceof ModifyDNOperation) * reflecting the new state of the UpdateMsg after conflict resolution * modified it, and dependencies might have been replayed by now. */ - searchFailedResolvingConflict = null; break; } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java index c15e28643b..fff7fe9f62 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java @@ -18,11 +18,14 @@ package org.opends.server.replication.plugin; 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.opends.server.protocols.internal.InternalClientConnection.*; import static org.testng.Assert.*; +import java.util.ArrayList; +import java.util.List; import java.util.TreeSet; import java.util.concurrent.atomic.AtomicBoolean; @@ -417,6 +420,7 @@ public void addIsRetriedWhileTheConflictResolutionSearchCanNotRun() throws Excep 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"); @@ -437,7 +441,7 @@ public void addIsRetriedWhileTheConflictResolutionSearchCanNotRun() throws Excep OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(), 2, 1); try { - replayMsg(addMsg(child, csn, parentUUID, "2d4d3d5e-3c6f-4ca0-9b8d-4e6f7a8b9cad")); + replayMsg(addMsg(child, csn, parentUUID, entryUUID)); assertShortCircuitSpentBy(3); } finally @@ -448,7 +452,10 @@ public void addIsRetriedWhileTheConflictResolutionSearchCanNotRun() throws Excep 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"); - assertFalse(entryExists(DN.valueOf(child.getName().rdn() + "," + TEST_ROOT_DN_STRING)), + // 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"); @@ -571,6 +578,72 @@ public void modifyIsLeftOutOfTheServerStateWhenTheEntryUUIDSearchNeverRuns() thr .as("the change was applied to the entry the searches never found").isEmpty(); } + /** + * Test case for [Issue 956]: the exhaustion exit reports the attempt which spent the + * last of the attempts in place, not an earlier one which ended on a search conflict + * resolution could not run. + *

+ * 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 record the error logger writes carries the id of the message rather than its text. + final String exhaustionExit = "msgID=" + ERR_ERROR_REPLAYING_OPERATION.ordinal(); + + /* + * 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); + ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", + ResultCode.UNAVAILABLE.intValue(), 1, LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS - 1); + try + { + 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 reports = new ArrayList<>(); + for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) + { + if (record.contains(exhaustionExit) && record.contains(csn.toString())) + { + reports.add(record); + } + } + assertThat(reports).as("the change was not reported once the attempts in place were spent") + .isNotEmpty(); + // The last attempt was refused before it reached the data and made no search: a line + // which names the entryUUID reports the search of an earlier attempt next to its result. + assertThat(reports).as("the exhaustion exit reports an attempt other than the last one") + .allMatch(record -> !record.contains(entryUUID)); + } + /** * Test case for [Issue 956]: the base entry of the domain replayed into a replica * which has none. From 3368a1d626fdaa0228e0d2481956d366c40acb13 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 14 Sep 2026 13:06:02 +0300 Subject: [PATCH 5/6] [#956] Leave the session to the total update into this replica, and keep the replay out of the data it replaces A change queued for replay before the InitializeTargetMsg arrived was replayed into no backend: preBackendImport() takes the backend away without disabling the domain, so the operation got NO_SUCH_OBJECT, the entryUUID search did not run, and once the attempts in place were spent the session was restarted for the change to be delivered again - the session the import was reading. The import ended on the entries which had arrived and was reported as finished, since receiveEntryBytes() records no exception for a broker which is shutting down. importBackend() now holds the replay off the way disable() does - a flag the replay threads read where they read `disabled`, and the drain of the attempts in flight - before the backend is taken away; the restart after a failed or abandoned replay is refused for the whole of the total update (sessionHasAnOwner(): ownsItsSession() || importInProgress()), from the request rather than from the first entry; and once the state is loaded from the imported data, the pending changes, the restart request and the backoff are reset, as disable() resets them - a change given back during the import would otherwise stay listed and uncommitted, and hold the ServerState back for good, since the imported state already covers it. Review round 3, with the flush thread's Modify kept out of the MODIFY short circuit (a predicate on the replayed operation, pinned by the case making that Modify inside the window) and the text of the exhaustion exit asserted positively. --- .../plugin/LDAPReplicationDomain.java | 116 +++++-- .../server/plugins/ShortCircuitPlugin.java | 46 ++- .../plugin/NamingConflictTest.java | 78 ++++- .../plugin/ReplayDuringImportTest.java | 293 ++++++++++++++++++ 4 files changed, 503 insertions(+), 30 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 0547a43376..4639d9701b 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -496,6 +496,23 @@ && getBackend().getBackendID().equals(backend.getBackendID())) { private final InternalClientConnection conn = getRootConnection(); private final AtomicBoolean shutdown = new AtomicBoolean(); private volatile boolean disabled; + /** + * Whether the data of this domain is being replaced by a total update into this replica: + * the import streams over the session of this domain, on its listener thread, and the + * backend the replay would apply a change to is deregistered for the length of it. + *

+ * 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 @@ -2786,11 +2803,11 @@ private void replayChangeAndTheChangesWaitingForIt( */ searchFailedResolvingConflict = null; /* - * 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. + * 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 @@ -2798,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) { /* @@ -2811,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) @@ -3618,13 +3635,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; } @@ -3736,9 +3759,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; } /* @@ -3760,9 +3784,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(); @@ -3788,6 +3813,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; } @@ -4597,11 +4627,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() { @@ -5130,6 +5161,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, @@ -5195,6 +5236,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) @@ -5470,6 +5530,26 @@ 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. + */ + private boolean sessionHasAnOwner() + { + return ownsItsSession() || importInProgress(); + } + @Override protected void restartService() { 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 624ec3aa03..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,7 +644,7 @@ 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); @@ -745,6 +745,20 @@ public static List createShortCircuitControlList(int resultCode, String /** How many operations a registered short circuit lets through before it applies. */ private static final Map shortCircuitSkips = new ConcurrentHashMap<>(); + /** + * Which operations a registered short circuit is for, when it is not for every operation + * of its type: the ones it is not for are neither short circuited nor counted. + */ + private static final Map> shortCircuitFilters = + new ConcurrentHashMap<>(); + + /** Returns whether the short circuit registered under the given key is for the given operation. */ + private static boolean appliesTo(String key, PluginOperation operation) + { + final Predicate filter = shortCircuitFilters.get(key); + return filter == null || filter.test(operation); + } + /** * Returns how many times the short circuit registered for the given operation type and * plugin point was reached. A short circuit registered for a limited number of times is @@ -776,6 +790,7 @@ public static void registerShortCircuit(OperationType operation, String section, shortCircuitCounts.remove(key); shortCircuitLimits.remove(key); shortCircuitSkips.remove(key); + shortCircuitFilters.remove(key); shortCircuits.put(key, resultCode); } @@ -812,11 +827,39 @@ public static void registerShortCircuit(OperationType operation, String section, { final String key = keyFor(operation, section); shortCircuitCounts.remove(key); + shortCircuitFilters.remove(key); shortCircuitSkips.put(key, letThroughFirst); shortCircuitLimits.put(key, maxTimes); shortCircuits.put(key, resultCode); } + /** + * Register a short circuit like + * {@link #registerShortCircuit(OperationType, String, int, int, int)}, for some of the + * operations of the given type only: the ones the predicate does not accept are let + * through without being counted, as if the short circuit were not there. + *

+ * 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 appliesTo) + { + registerShortCircuit(operation, section, resultCode, letThroughFirst, maxTimes); + shortCircuitFilters.put(keyFor(operation, section), appliesTo); + } + /** * Deregister a short circuit for the given operation type and plugin point. * @param operation The type of operation the short circuit applies to. @@ -828,6 +871,7 @@ public static void deregisterShortCircuit(OperationType operation, String sectio shortCircuits.remove(key); shortCircuitLimits.remove(key); shortCircuitSkips.remove(key); + shortCircuitFilters.remove(key); // The count belongs to the registration which is being removed: a test which counts // the operations it short circuits must not inherit the count of the previous one. shortCircuitCounts.remove(key); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java index fff7fe9f62..29659d1a2b 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java @@ -36,6 +36,7 @@ import org.opends.server.TestCaseUtils; import org.opends.server.core.DirectoryServer; import org.opends.server.core.ModifyDNOperation; +import org.opends.server.core.ModifyOperationBasis; import org.opends.server.plugins.ShortCircuitPlugin; import org.opends.server.replication.ReplicationTestCase; import org.opends.server.replication.common.CSN; @@ -45,6 +46,7 @@ import org.opends.server.replication.protocol.LDAPUpdateMsg; import org.opends.server.replication.protocol.ModifyDNMsg; import org.opends.server.replication.protocol.ModifyMsg; +import org.opends.server.replication.protocol.OperationContext; import org.opends.server.replication.protocol.UpdateMsg; import org.opends.server.types.Entry; import org.opends.server.types.OperationType; @@ -576,6 +578,17 @@ public void modifyIsLeftOutOfTheServerStateWhenTheEntryUUIDSearchNeverRuns() thr "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 reports = exhaustionExitRecordsOf(csn); + assertThat(reports).as("the change was not reported once the attempts in place were spent") + .isNotEmpty(); + assertThat(reports) + .as("the exhaustion exit reports the error of the operation, not the search which did not run") + .allMatch(record -> record.contains(entryUUID)); } /** @@ -598,8 +611,6 @@ public void theExhaustionExitReportsTheAttemptWhichSpentTheLastOfThem() throws E final String entryUUID = getEntryUUID(entry.getName()); final DN staleDN = DN.valueOf("cn=movedAway," + TEST_ROOT_DN_STRING); final CSN csn = gen.newCSN(); - // The record the error logger writes carries the id of the message rather than its text. - final String exhaustionExit = "msgID=" + ERR_ERROR_REPLAYING_OPERATION.ordinal(); /* * The first attempt is let through to the data and fails on the conflict of the @@ -609,10 +620,25 @@ public void theExhaustionExitReportsTheAttemptWhichSpentTheLastOfThem() throws E */ 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); + 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"); @@ -628,14 +654,7 @@ public void theExhaustionExitReportsTheAttemptWhichSpentTheLastOfThem() throws E assertFalse(domain.getServerState().cover(csn), "a change the server kept refusing is not in the data and must not advance the ServerState"); - final List reports = new ArrayList<>(); - for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) - { - if (record.contains(exhaustionExit) && record.contains(csn.toString())) - { - reports.add(record); - } - } + final List reports = exhaustionExitRecordsOf(csn); assertThat(reports).as("the change was not reported once the attempts in place were spent") .isNotEmpty(); // The last attempt was refused before it reached the data and made no search: a line @@ -704,6 +723,25 @@ public void anEntryUUIDWhichIsNotOneNamesNoEntry() throws Exception .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 exhaustionExitRecordsOf(CSN csn) + { + final String exhaustionExit = "msgID=" + ERR_ERROR_REPLAYING_OPERATION.ordinal(); + final List reports = new ArrayList<>(); + for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) + { + if (record.contains(exhaustionExit) && record.contains(csn.toString())) + { + reports.add(record); + } + } + return reports; + } + /** * Asserts that the searches a short circuit was registered over were made - the ones * let through before it and the ones it applied to - and that the search after them @@ -816,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..916cfef0e9 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java @@ -0,0 +1,293 @@ +/* + * 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.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy; +import org.opends.server.TestCaseUtils; +import org.opends.server.core.DirectoryServer; +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.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.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 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. + */ +@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())); + + final SortedSet replServers = new TreeSet<>(); + replServers.add("localhost:" + rsPort); + final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID, replServers); + conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES); + queue = new TestSynchronousReplayQueue(); + domain = MultimasterReplication.createNewDomain(conf, queue); + domain.start(); + assertTrue(domain.isConnected(), "the domain did not connect to the replication server"); + + exporter = openReplicationSession(baseDN, EXPORTER_ID, 100, rsPort, 10000); + gen = new CSNGenerator(201, 0); + } + + @AfterMethod + public void tearDown() throws Exception + { + try + { + stop(exporter); + MultimasterReplication.deleteDomain(baseDN); + } + finally + { + remove(replicationServer); + } + } + + /** + * A change replayed while the import streams must leave the session to the import. + *

+ * 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. A failure + * of the server is what that is, and once the attempts in place are spent the change is + * given back and the session restarted for it to be delivered again: that restart stops + * the broker the import is reading, the import ends on the entries which had arrived, and + * nothing reports it - the stream ends the way a finished one does. + */ + @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"); + } + 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"); + } + + /** + * 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 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 errorLogRecordsOf(int msgId, CSN csn) + { + final List records = new ArrayList<>(); + for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) + { + if (record.contains("msgID=" + msgId) && record.contains(csn.toString())) + { + records.add(record); + } + } + return records; + } + + private void replayMsg(UpdateMsg updateMsg) throws InterruptedException + { + domain.processUpdate(updateMsg); + final LDAPUpdateMsg ldapUpdate = queue.take().getUpdateMessage(); + domain.markInProgress(ldapUpdate); + domain.replay(ldapUpdate, SHUTDOWN); + } +} From 00798abf336a0c30f0d542165327a6748cd32506 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 14 Sep 2026 17:16:22 +0300 Subject: [PATCH 6/6] [#956] Pin the hold-off and the owner from the request, and describe the road as it is The importingData hold-off was pinned by no test: with `|| importingData` deleted from both goingDown reads, the change was attempted ten times into no backend and the session was saved by sessionHasAnOwner() alone, which the retry-warning check could not tell from the hold-off - that guard returns before the warning. aReplayDuringTheImportLeavesTheSessionToTheImport now asserts that no ERR_ERROR_REPLAYING_OPERATION record names the change: none at head, one under that mutant. sessionHasAnOwner() in recoverFromReplayFailure() was reached by neither case, since the hold-off gives the change back at the top of its first attempt. aRequestOnItsWayOwnsTheSessionTheAnswerArrivesOver reaches it: a total update this replica asked for, the answer held by the exporter, the backend live, a change whose entryUUID search never runs spent in place - the exhaustion exit reports it, no retry warning follows, and the import then runs to its end over the session the request was made over. With that guard on ownsItsSession() alone, the warning is logged and the case is red on it. Two comments described the road before the previous commit - the change given back is delivered again after the import restarts the session, not now, and the first case of ReplayDuringImportTest never spends an attempt - and the javadoc of sessionHasAnOwner() says that a change left listed holds the ServerState back, in memory and as persisted, until the next restart. Review round 4. --- .../plugin/LDAPReplicationDomain.java | 8 +- .../plugin/ReplayDuringImportTest.java | 108 ++++++++++++++++-- 2 files changed, 106 insertions(+), 10 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 4639d9701b..fb908f810f 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -3325,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; } @@ -5544,6 +5545,9 @@ private boolean ownsItsSession() * 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() { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java index 916cfef0e9..3bedecb489 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java @@ -29,14 +29,17 @@ 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; @@ -45,6 +48,7 @@ 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; @@ -56,10 +60,13 @@ * 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). + * 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. + * 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 @@ -126,12 +133,15 @@ public void tearDown() throws Exception /** * A change replayed while the import streams must leave the session to the import. *

- * 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. A failure - * of the server is what that is, and once the attempts in place are spent the change is - * given back and the session restarted for it to be delivered again: that restart stops - * the broker the import is reading, the import ends on the entries which had arrived, and - * nothing reports it - the stream ends the way a finished one does. + * 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 @@ -159,6 +169,15 @@ public void aReplayDuringTheImportLeavesTheSessionToTheImport() throws Exception 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(); @@ -203,6 +222,68 @@ public void aChangeGivenBackDuringTheImportDoesNotHoldTheServerStateBack() throw + " 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, @@ -221,6 +302,17 @@ private void startImportInto(int entryCount) throws Exception } } + /** + * 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 {