From 7418d7bbf67a3a345b5b86b610788301c88931b7 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 14 Sep 2026 19:02:18 +0300 Subject: [PATCH] [#1041] Claim the import context for the length of a session restart, and refuse the total update which lands across it The owner read of a session restart and the listener thread's claim of the import context for an InitializeTargetMsg share no lock, and cannot: disableService() waits for the listener thread under serviceStateLock, so the listener cannot take that lock before its claim. A restart which read no owner a few statements before the claim landed stopped the broker the import was about to read: the import ran over it and ended on the nothing which arrived, the suffix replaced by it - reported as a finished total update until issue #1039, recorded as a failed import since - or, the stop landing before the full update status was published, the listener thread died on a NullPointerException in signalStatusChange(). The two now contend on the one reference. disableServiceUnlessImportInProgress() claims importExportContext with SESSION_BEING_STOPPED before it stops anything, and releases it once disableService() returns - the listener thread is gone by then. The listener's claim fails against it and the InitializeTargetMsg is refused: ERR_INIT_REJECTED_SESSION_STOPPING, which names the refusing server, to the exporter and to this server's log. An export in the context is not an owner and is stopped from under, as before; once its claim is made the listener re-reads whether the broker is stopping and refuses the total update if it is, which narrows - it is a read, so it does not close - the roads which stop the session while claiming nothing: a disable(), the shutdown of the domain, and the stop which follows an export. The answer to a total update this replica asked for itself gets the same read, and fails the task which asked for it. The claim is invisible to ieRunning(), importInProgress(), getImportExportContext() and every reader of the context; releaseIEContext() now releases only the context it is given, so an operation which fails to claim one no longer releases the context it failed against. Two test seams: the listener before its claim - or, for a total update this replica asked for, before its read of the context the request claimed - and disableService() before the broker is stopped. ReplayDuringImportTest holds the listener there, drives a restart through the gap with a change whose attempts in place are spent, and releases the listener before the stop: the import must be refused, the exporter told, the data of this replica left alone, and a stale answer to a request of this replica ignored rather than imported into the claim. A second case disables the domain across the gap, a third does the same to a total update this replica asked for, and a fourth has the restart find an export in the context and stop the session that export streams over. Each road out of the claim is pinned: the restart case fails with the read in place of the claim, the two disable cases with the read of the broker removed from their arm, the export case with any owner leaving the session alone, and the assertions around them with the claim visible to ieRunning() or to the arm of a request of this replica, with a refused request releasing it, with it never released, and with the refused import keeping its context. --- .../plugin/LDAPReplicationDomain.java | 17 +- .../service/ReplicationDomain.java | 237 ++++++++- .../opends/messages/replication.properties | 3 + .../plugin/ReplayDuringImportTest.java | 488 +++++++++++++++++- 4 files changed, 718 insertions(+), 27 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 1467c353c5..1238dbd63f 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 @@ -4287,13 +4287,17 @@ private void restartSession(boolean wait) final long wakes; synchronized (serviceStateLock) { - if (sessionHasAnOwner()) + /* + * The domain is going away or is being imported into: the session is not this + * thread's to stop. The total update is claimed against rather than read (issue + * #1041): the listener thread claims one this replica did not ask for under no lock, + * and a read here a few statements before that claim would stop the session the + * import is about to read. + */ + if (ownsItsSession() || !disableServiceUnlessImportInProgress()) { - // The domain is going away or is being imported into: the session is not this - // thread's to stop. return; } - disableService(); stoppedSession = getSessionGeneration(); wakes = sessionRestartBackoffWakes(); } @@ -6213,6 +6217,11 @@ private boolean ownsItsSession() * it carries: the domain itself, when it is shutting down or disabled * ({@link #ownsItsSession()}), or a total update into this replica. *

+ * What this reads, {@link #restartSession(boolean)} claims: a total update the listener + * thread is about to claim is not visible to a read, and the restart must not stop the + * session such a total update reads (issue #1041). This is the early exit of the roads + * which lead to that restart, and the answer for the ones which never restart anything. + *

* 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 diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java index 472f68cf8d..80c6bbf5bc 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java @@ -56,6 +56,7 @@ import org.forgerock.opendj.ldap.ResultCode; import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType; import org.forgerock.opendj.server.config.server.ReplicationDomainCfg; +import org.forgerock.util.annotations.VisibleForTesting; import org.opends.server.api.DirectoryThread; import org.opends.server.api.MonitorData; import org.opends.server.backends.task.Task; @@ -262,6 +263,39 @@ public ECLIncludes addIncludedAttributes(int serverId, * Null when none is being processed. */ private final AtomicReference importExportContext = new AtomicReference<>(); + /** + * Holds {@link #importExportContext} for the length of a session stop which a total update + * into this replica must not be claimed across (issue #1041). + *

+ * A restart of the session reads whether such a total update owns it before it stops + * anything, and the listener thread claims the context for the {@code InitializeTargetMsg} + * it took off the session - and the two share no lock: {@link #disableService()} waits for + * the listener thread under {@link #serviceStateLock}, so the listener can not take that + * lock before its claim. A restart which read no owner a few statements before the claim + * landed stopped the broker the import was about to read, and the import ended on the + * nothing which arrived - recorded as a failed import since issue #1039, over a suffix + * which has been replaced by it all the same. The two contend on the one reference + * instead: the restart claims it with this context, the claim of the listener fails + * against it, and exactly one of them wins - the total update is either the owner the + * restart reads, or refused. + *

+ * It is neither an import nor an export: {@link #ieRunning()}, {@link #importInProgress()} + * and {@link #getImportExportContext()} do not report it. + */ + private static final ImportExportContext SESSION_BEING_STOPPED = new ImportExportContext(false); + /** + * Run by the listener thread between the {@code InitializeTargetMsg} it took off the + * session and its claim of the import context - or, for a total update this replica asked + * for, its read of the context the request claimed. Only there for the tests, which hold + * the listener thread there: nothing else runs in that gap. + */ + private volatile Runnable importClaimHook; + /** + * Run by {@link #disableService()} under its locks, before the broker is stopped. Only + * there for the tests, which hold a stop of the service there: what the claim of a session + * stop is for is the total update which lands between the decision to stop and the stop. + */ + private volatile Runnable serviceStopHook; /** * The Thread waiting for incoming update messages for this domain and pushing @@ -838,7 +872,7 @@ else if (msg instanceof InitializeTargetMsg) else if (msg instanceof ErrorMsg) { ErrorMsg errorMsg = (ErrorMsg)msg; - ImportExportContext ieCtx = importExportContext.get(); + ImportExportContext ieCtx = getImportExportContext(); if (ieCtx != null) { /* @@ -900,7 +934,7 @@ else if (msg instanceof UpdateMsg) } else if (msg instanceof InitializeRcvAckMsg) { - ImportExportContext ieCtx = importExportContext.get(); + ImportExportContext ieCtx = getImportExportContext(); if (ieCtx != null) { InitializeRcvAckMsg ackMsg = (InitializeRcvAckMsg) msg; @@ -1660,7 +1694,7 @@ context is live here (ErrorMsg carries no correlation id) - and // Release the context whatever the outcome, otherwise ieRunning() would // remain true forever (resolves the historical "FIXME should not this // be in a finally?"). - releaseIEContext(); + releaseIEContext(ieCtx); } } @@ -2036,16 +2070,27 @@ private ImportExportContext acquireIEContext(boolean importInProgress) final ImportExportContext ieCtx = new ImportExportContext(importInProgress); if (!importExportContext.compareAndSet(null, ieCtx)) { - // Rejects 2 simultaneous exports - LocalizableMessage message = ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get(); + // Rejects 2 simultaneous exports, and a total update which is claimed while the + // session is being stopped - in either direction: the entries of an export out of this + // server are streamed over that session too (see SESSION_BEING_STOPPED) + final LocalizableMessage message = importExportContext.get() == SESSION_BEING_STOPPED + ? ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId()) + : ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get(); throw new DirectoryException(ResultCode.OTHER, message); } return ieCtx; } - private void releaseIEContext() + /** + * Releases the provided import/export context, and only that one: a road which failed to + * acquire a context of its own must not release the one it failed against - the import or + * export which owns it, or the claim of a session stop ({@code SESSION_BEING_STOPPED}). + * + * @param ieCtx the context to release + */ + private void releaseIEContext(ImportExportContext ieCtx) { - importExportContext.set(null); + importExportContext.compareAndSet(ieCtx, null); } /** @@ -2059,7 +2104,7 @@ private void releaseIEContext() */ private void completeInitializeTask(ImportExportContext ieCtx) { - releaseIEContext(); + releaseIEContext(ieCtx); if (ieCtx.initializeTask instanceof InitializeTask) { // Update the task that initiated the import @@ -2108,7 +2153,7 @@ protected byte[] receiveEntryBytes() ReplicationMsg msg; while (true) { - ImportExportContext ieCtx = importExportContext.get(); + ImportExportContext ieCtx = getImportExportContext(); try { // In the context of the total update, we don't want any automatic @@ -2281,7 +2326,7 @@ void exportLDIFEntry(byte[] lDIFEntry, int pos, int length) } // build the message - ImportExportContext ieCtx = importExportContext.get(); + ImportExportContext ieCtx = getImportExportContext(); EntryMsg entryMessage = new EntryMsg( getServerId(), ieCtx.getExportTarget(), lDIFEntry, pos, length, ++ieCtx.msgCnt); @@ -2425,6 +2470,7 @@ running this method directly coming from initialize() method and did not processed any topology message in between the failure and the new attempt. */ + ImportExportContext ieCtx = null; try { /* @@ -2434,7 +2480,7 @@ The context will be used when we (the listener thread) will receive update the task. */ - final ImportExportContext ieCtx = acquireIEContext(true); + ieCtx = acquireIEContext(true); ieCtx.initializeTask = initTask; ieCtx.attemptCnt = 0; ieCtx.initReqMsgSent = new InitializeRequestMsg( @@ -2474,7 +2520,10 @@ The context will be used when we (the listener thread) will receive { // No need to call here updateTaskCompletionState - will be done // by the caller - releaseIEContext(); + if (ieCtx != null) + { + releaseIEContext(ieCtx); + } throw new DirectoryException(ResultCode.OTHER, errMsg); } } @@ -2494,7 +2543,7 @@ The context will be used when we (the listener thread) will receive */ public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs) { - final ImportExportContext ieCtx = importExportContext.get(); + final ImportExportContext ieCtx = getImportExportContext(); if (ieCtx == null || !ieCtx.importInProgress() || ieCtx.initReqMsgSent == null || !ieCtx.abandonIfStalled(stalledTimeoutMs)) { @@ -2510,6 +2559,23 @@ public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs) return true; } + /** + * Refuses a total update another server started into this replica: the exporter is told + * so that it does not stream to a replica which will discard the entries, and this server + * records why the total update it was the target of did not run - the exporter's task + * reports the failure, and an administrator reading this server's log has to find it here. + * + * @param requesterServerId the server which asked for the total update + * @param reason why it is refused + */ + private void rejectInitializeTarget(int requesterServerId, LocalizableMessage reason) + { + logger.error(reason); + // Silently not sent over a session which is already stopped: the replication server + // then tells the exporter that this replica is not there to stream to. + broker.publish(new ErrorMsg(requesterServerId, reason)); + } + /** * Processes an InitializeTargetMsg received from a remote server * meaning processes an initialization from the entries expected to be @@ -2532,9 +2598,14 @@ private void initialize(InitializeTargetMsg initTargetMsgReceived, int requester InitializeTask initFromTask = null; final int source = initTargetMsgReceived.getSenderID(); final ImportExportContext ieCtx; + final Runnable hook = importClaimHook; + if (hook != null) + { + hook.run(); + } if (initTargetMsgReceived.getInitiatorID() == getServerId()) { - ieCtx = importExportContext.get(); + ieCtx = getImportExportContext(); if (ieCtx == null || !ieCtx.markInitStartReceived()) { /* @@ -2551,6 +2622,20 @@ private void initialize(InitializeTargetMsg initTargetMsgReceived, int requester } return; } + if (broker.shuttingDown()) + { + /* + * The same read as for a total update another server started (see below), with the + * same window: the context is the one the request claimed, and no restart stops the + * session under it - an import owns the session - but the domain going down or being + * disabled does. The task which asked for the total update is failed with the reason; + * the exporter learns of the stop the way it does of any other stop of this session. + */ + ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER, + ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId()))); + completeInitializeTask(ieCtx); + return; + } } else { @@ -2564,11 +2649,32 @@ private void initialize(InitializeTargetMsg initTargetMsgReceived, int requester } catch (DirectoryException e) { - // A concurrent import/export owns the context: reject this - // initialization without touching that operation's context, and let - // the exporter know so that it does not export to a replica that - // will discard the entries - broker.publish(new ErrorMsg(requesterServerId, e.getMessageObject())); + // A concurrent import/export owns the context, or the session is being stopped: + // reject this initialization without touching that operation's context, and let + // the exporter know so that it does not export to a replica that will discard the + // entries + rejectInitializeTarget(requesterServerId, e.getMessageObject()); + return; + } + if (broker.shuttingDown()) + { + /* + * The claim won against no restart, and the session is being stopped all the same: + * the domain is going down or being disabled, or a restart found an export in the + * context and stopped the session it streams over. The import would read that broker + * as the end of its stream, and what runs before it publishes over the session: it + * is refused here, before the backend is taken away. + * + * Read rather than claimed against: none of these roads claims anything - they stop + * the session whatever owns it - so nothing orders this read against the stop. It + * narrows the window, it does not close it: a stop which lands after it still has + * the import run over a session which is going down, and end as a failed import over + * the suffix it has replaced (issue #1039). Every one of these roads had that window + * before this claim, and has it still. + */ + releaseIEContext(ieCtx); + rejectInitializeTarget(requesterServerId, + ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId())); return; } } @@ -2773,7 +2879,37 @@ private void setNewStatus(StatusMachineEvent event) */ public boolean ieRunning() { - return importExportContext.get() != null; + return getImportExportContext() != null; + } + + /** + * Sets what the listener thread runs between the {@code InitializeTargetMsg} it took off + * the session and its claim of the import context - or, for a total update this replica + * asked for, its read of the context the request claimed. + *

+ * Only there for the tests which drive something else through that gap: it is a few + * statements wide, and nothing else can hold the listener thread there. + * + * @param hook what to run there, or {@code null} to run nothing + */ + @VisibleForTesting + public void setImportClaimHook(Runnable hook) + { + importClaimHook = hook; + } + + /** + * Sets what {@link #disableService()} runs, under its locks, before it stops the broker. + *

+ * Only there for the tests which drive something else through that gap: a total update + * which is claimed after the decision to stop the service and before the stop. + * + * @param hook what to run there, or {@code null} to run nothing + */ + @VisibleForTesting + public void setServiceStopHook(Runnable hook) + { + serviceStopHook = hook; } /** @@ -2790,7 +2926,7 @@ public boolean ieRunning() */ protected boolean importInProgress() { - final ImportExportContext ieCtx = importExportContext.get(); + final ImportExportContext ieCtx = getImportExportContext(); return ieCtx != null && ieCtx.importInProgress(); } @@ -3343,6 +3479,11 @@ public final void disableService() { synchronized (sessionLock) { + final Runnable hook = serviceStopHook; + if (hook != null) + { + hook.run(); + } /* * Stop the broker first in order to prevent the listener from reconnecting - see OPENDJ-457. */ @@ -3370,6 +3511,57 @@ public final void disableService() } } + /** + * Stops the Replication Service the way {@link #disableService()} does, unless a total + * update into this replica owns the session. + *

+ * Whether one does is claimed rather than read (issue #1041): the listener thread claims + * the import context for an {@code InitializeTargetMsg} under no lock, so a read of it + * under {@link #serviceStateLock} orders nothing. The claim is + * {@code SESSION_BEING_STOPPED}, held for the length of the stop and released once the + * listener thread is gone - it is the one thread which claims a total update this replica + * did not ask for, and {@link #disableService()} waits for it. An export in the context is + * not an owner: the session is stopped from under it and the exporter reports the cut, as + * it does for every other stop. A total update which lands between the end of that export + * and the stop is refused by the listener when it reads the broker as stopping after its + * claim; a stop which lands after that read still has the import run over a session which + * is going down, and end as a failed import over the suffix it has replaced (issue + * #1039): the read narrows that window, it does not close it. + * + * @return {@code true} when the service was stopped, {@code false} when a total update + * into this replica owns the session and it was left alone + */ + protected final boolean disableServiceUnlessImportInProgress() + { + synchronized (serviceStateLock) + { + while (!importExportContext.compareAndSet(null, SESSION_BEING_STOPPED)) + { + final ImportExportContext owner = importExportContext.get(); + if (owner == null) + { + // Released between the two reads: claim again. + continue; + } + if (owner.importInProgress()) + { + return false; + } + disableService(); + return true; + } + try + { + disableService(); + } + finally + { + importExportContext.compareAndSet(SESSION_BEING_STOPPED, null); + } + return true; + } + } + /** * Returns {@code true} if the listener thread is shutting down or has * shutdown. @@ -3863,7 +4055,8 @@ public void addAdditionalMonitoring(MonitorData monitorData) */ protected ImportExportContext getImportExportContext() { - return importExportContext.get(); + final ImportExportContext ieCtx = importExportContext.get(); + return ieCtx != SESSION_BEING_STOPPED ? ieCtx : null; } /** 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 a4eb51f839..0276c4e193 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -712,3 +712,6 @@ ERR_INIT_SESSION_STOPPED_DURING_IMPORT_329=Domain %s (server id: %s) : the sessi WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_328=Replication server RS(%d) not sending update \ %s for domain "%s" to server %d at %s because the replication protocol version %d \ negotiated with it has no encoding for this message +ERR_INIT_REJECTED_SESSION_STOPPING_330=The total update of domain "%s" was refused by directory \ + server %d: its session to the replication server is being stopped, and the entries would have \ + been streamed over that session. Ask for the total update again once the session is back diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java index 23d4f018d6..538c7e1060 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 @@ -17,6 +17,7 @@ import static java.nio.charset.StandardCharsets.*; import static org.assertj.core.api.Assertions.*; +import static org.opends.messages.CoreMessages.ERR_UNCAUGHT_THREAD_EXCEPTION; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.TestCaseUtils.*; import static org.opends.server.core.DirectoryServer.*; @@ -26,7 +27,11 @@ import java.util.List; import java.util.SortedSet; import java.util.TreeSet; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; import org.forgerock.opendj.ldap.DN; import org.forgerock.opendj.ldap.ResultCode; @@ -40,6 +45,7 @@ import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.protocol.DoneMsg; import org.opends.server.replication.protocol.EntryMsg; +import org.opends.server.replication.protocol.ErrorMsg; import org.opends.server.replication.protocol.InitializeRequestMsg; import org.opends.server.replication.protocol.InitializeTargetMsg; import org.opends.server.replication.protocol.LDAPUpdateMsg; @@ -48,6 +54,7 @@ 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.DirectoryException; import org.opends.server.types.Entry; import org.opends.server.types.OperationType; import org.testng.Assert; @@ -75,6 +82,11 @@ * change is replayed while the import is waiting for them - or, for the request, while the * exporter is holding the answer. *

+ * The claim of a total update this replica did not ask for is made by the listener thread + * under no lock, so a restart of the session which reads no owner a moment before that claim + * would stop the session the import is about to read (issue #1041): the listener is held + * before its claim, and what stops the session is driven through the gap. + *

* The {@code timeOut} each case declares is what it is expected to take at the most; it is * not what bounds it. {@code TestListener} sets the timeout of every test method from the * {@code org.opends.test.timeout} property, ten minutes under Maven and none outside it. @@ -508,6 +520,421 @@ public void aWarningAfterTheImportDoesNotCountTheDeliveriesBefore() throws Excep } } + /** + * A session restart decided after the {@code InitializeTargetMsg} was taken off the session + * and before the import claimed its context must not have the import run over the session + * it stops (issue #1041). + *

+ * The owner read of the restart and the claim of the listener share no lock: the restart + * reads no owner, stops the broker and waits for the listener thread to end - which is the + * thread about to run the import. Run over that broker, the import ends on the nothing + * which arrived - as a failed import since issue #1039, and as a finished one before it - + * over a suffix which has been replaced by it all the same. Here the listener is held + * before its claim, the restart is driven through the gap by a change whose attempts in + * place are spent and held between its decision and the stop, and the listener is released + * in between: the broker it finds is still up, so what refuses the import is the claim of + * the restart, and the refusal reaches the exporter over the session which is about to be + * stopped. + */ + @Test(timeOut = 120_000) + public void aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport() 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 int totalUpdatesStartedBefore = + errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size(); + final int totalUpdatesEndedBefore = + errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size(); + final int listenerDeathsBefore = listenerDeaths().size(); + final int refusalsBefore = errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size(); + + // The listener thread has taken the InitializeTargetMsg off the session and is held + // before it claims the import; the restart is held after its decision, before the stop. + final CountDownLatch listenerHeld = new CountDownLatch(1); + final CountDownLatch releaseListener = new CountDownLatch(1); + final CountDownLatch stopHeld = new CountDownLatch(1); + final CountDownLatch releaseStop = new CountDownLatch(1); + domain.setImportClaimHook(() -> { + listenerHeld.countDown(); + awaitUninterruptibly(releaseListener); + }); + domain.setServiceStopHook(() -> { + stopHeld.countDown(); + awaitUninterruptibly(releaseStop); + }); + try + { + exporter.publish(new InitializeTargetMsg( + baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW)); + assertTrue(listenerHeld.await(30, TimeUnit.SECONDS), + "the listener thread did not reach the claim of the import"); + + /* + * A change whose entryUUID search never runs spends its attempts in place, finds no + * owner and restarts the session. On a thread of its own: the restart is held before + * the stop, and then waits for the listener thread. + */ + final CSN csn = gen.newCSN(); + final AtomicReference replayFailure = new AtomicReference<>(); + final Thread replay = new Thread(() -> { + try + { + replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN), + generatemods("description", "replayed before the import was claimed"), entryUUID)); + } + catch (Throwable t) + { + replayFailure.set(t); + } + }, "replay of " + csn); + ShortCircuitPlugin.registerShortCircuit( + OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue()); + try + { + replay.start(); + assertTrue(stopHeld.await(30, TimeUnit.SECONDS), + "the failed replay did not decide to restart the session"); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse"); + } + assertTrue(domain.isConnected(), "the session was stopped before the stop was held"); + assertFalse(domain.ieRunning(), "the claim of the stop is visible as a running import"); + /* + * A total update asked for here is refused against the claim of the stop, and the + * claim is left where it is: the road which fails to acquire a context of its own + * releases nothing. + */ + assertThatThrownBy(() -> domain.initializeFromRemote(EXPORTER_ID, null)) + .as("a total update asked for while the session is being stopped was not refused") + .isInstanceOf(DirectoryException.class) + .hasMessageContaining(ERR_INIT_REJECTED_SESSION_STOPPING.get(baseDN, DS_ID).toString()); + + /* + * The import is claimed against a restart which is decided and not yet made. Decided + * either way before the stop is released: without the claim the import runs, and the + * exporter is then waited for over a socket which nothing bounds. + */ + releaseListener.countDown(); + waitUntil(() -> errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size() > refusalsBefore + || errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size() > totalUpdatesStartedBefore, + "the listener neither refused nor started the total update"); + assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal())) + .as("a total update claimed against a restart which was decided was started") + .hasSize(totalUpdatesStartedBefore); + final ErrorMsg refusal = waitForSpecificMsg(exporter, ErrorMsg.class); + assertThat(refusal.getDetails().toString()) + .as("the exporter was not told why the total update was refused") + .isEqualTo(ERR_INIT_REJECTED_SESSION_STOPPING.get(baseDN, DS_ID).toString()); + + /* + * An answer to a total update this replica asked for, which no context stands for - the + * request was abandoned as stalled (issue #861) - finds only the claim of the stop, and + * the claim is no context to import into: the answer is ignored. The total update + * another server starts after it is what shows that the listener is past it: refused + * here, against the same claim. + */ + final int refusalsOfTheFirst = + errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size(); + exporter.publish(new InitializeTargetMsg( + baseDN, EXPORTER_ID, DS_ID, DS_ID, exportedEntries().length, INIT_WINDOW)); + exporter.publish(new InitializeTargetMsg( + baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW)); + waitUntil(() -> errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size() > refusalsOfTheFirst + || errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size() > totalUpdatesStartedBefore, + "the listener neither refused nor started the total update after the stale answer"); + assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal())) + .as("an answer no context stands for was imported into the claim of the stop") + .hasSize(totalUpdatesStartedBefore); + + releaseStop.countDown(); + replay.join(60_000); + assertFalse(replay.isAlive(), "the restart did not end: the listener thread it waits for is still there"); + assertNull(replayFailure.get(), "the replay failed: " + replayFailure.get()); + } + finally + { + releaseListener.countDown(); + releaseStop.countDown(); + domain.setImportClaimHook(null); + domain.setServiceStopHook(null); + } + + waitUntil(domain::isConnected, "the session was not started back after the restart"); + assertTrue(entryExists(entry.getName()), "the import ran over the session the restart" + + " stopped: the suffix was replaced by the nothing which arrived"); + // A total update which got past the claim ran over the broker the restart then stopped + // and ended on the nothing which arrived - as a failed import since issue #1039, and as + // a finished one before it; neither is a total update which never ran. + assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal())) + .as("a total update which was refused was run") + .hasSize(totalUpdatesEndedBefore); + assertThat(listenerDeaths()) + .as("the listener thread ended on an uncaught exception") + .hasSize(listenerDeathsBefore); + // Every record is written twice - the error log has two publishers in the tests. + assertThat(errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal())) + .as("the refusal of the total update was not recorded on this server") + .hasSizeGreaterThan(refusalsBefore); + + /* + * The claim of the stop was released with the stop: the next total update into this + * replica is claimed by the listener and runs to its end. Held, it would be invisible + * to every reader of the context and refuse every total update for the life of the + * domain. + */ + startImportInto(exportedEntries().length); + finishImport(exportedEntries()); + assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal())) + .as("the total update after the restart did not run to its end") + .hasSize(totalUpdatesEndedBefore + 2); + } + + /** + * A domain disabled after the {@code InitializeTargetMsg} was taken off the session and + * before the import claimed its context must refuse the import as well. + *

+ * Nothing claims against the listener here - the domain disabling itself stops the session + * whatever owns it - so what refuses the import is the listener reading, once its claim is + * made, that the broker it would stream over is stopping. Without that read the claim wins, + * and what runs next publishes the full update status over a session which is gone. + */ + @Test(timeOut = 120_000) + public void aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport() throws Exception + { + final Entry entry = TestCaseUtils.addEntry( + "dn: cn=survivor," + EXAMPLE_DN, + "objectClass: top", + "objectClass: person", + "cn: survivor", + "sn: survivor"); + final int totalUpdatesStartedBefore = + errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size(); + final int listenerDeathsBefore = listenerDeaths().size(); + final int refusalsBefore = errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size(); + + final CountDownLatch listenerHeld = new CountDownLatch(1); + final CountDownLatch releaseListener = new CountDownLatch(1); + domain.setImportClaimHook(() -> { + listenerHeld.countDown(); + awaitUninterruptibly(releaseListener); + }); + try + { + exporter.publish(new InitializeTargetMsg( + baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW)); + assertTrue(listenerHeld.await(30, TimeUnit.SECONDS), + "the listener thread did not reach the claim of the import"); + + // On a thread of its own: disabling the domain waits for the listener thread. + final Thread disable = new Thread(domain::disable, "disable of " + EXAMPLE_DN); + disable.start(); + waitUntil(() -> !domain.isConnected(), "disabling the domain did not stop the session"); + releaseListener.countDown(); + disable.join(60_000); + assertFalse(disable.isAlive(), "disabling the domain did not end: the listener thread" + + " it waits for is still there"); + } + finally + { + releaseListener.countDown(); + domain.setImportClaimHook(null); + } + domain.enable(); + waitUntil(domain::isConnected, "the session was not started back by enable()"); + assertFalse(domain.ieRunning(), "the refused import left its context claimed"); + + assertTrue(entryExists(entry.getName()), "the import ran over the session the disable" + + " stopped: the suffix was replaced by the nothing which arrived"); + assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal())) + .as("a total update claimed against a session which is being stopped was started") + .hasSize(totalUpdatesStartedBefore); + assertThat(listenerDeaths()) + .as("the listener thread ended on an uncaught exception") + .hasSize(listenerDeathsBefore); + // Every record is written twice - the error log has two publishers in the tests. + assertThat(errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal())) + .as("the refusal of the total update was not recorded on this server") + .hasSizeGreaterThan(refusalsBefore); + } + + /** + * A domain disabled after the answer to a total update this replica asked for was taken off + * the session, and before the import started, must refuse the import too. + *

+ * The context is the one the request claimed, so there is nothing to claim against: what + * refuses the import is the same read of the broker as for a total update another server + * started. Without it the import runs over the session the disable stopped, and replaces the + * suffix with the nothing which arrived. + */ + @Test(timeOut = 120_000) + public void aDomainDisabledBeforeTheImportItAskedForStartsRefusesTheImport() throws Exception + { + final Entry entry = TestCaseUtils.addEntry( + "dn: cn=survivor," + EXAMPLE_DN, + "objectClass: top", + "objectClass: person", + "cn: survivor", + "sn: survivor"); + final int totalUpdatesStartedBefore = + errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size(); + final int listenerDeathsBefore = listenerDeaths().size(); + + domain.initializeFromRemote(EXPORTER_ID, null); + assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class)); + + final CountDownLatch listenerHeld = new CountDownLatch(1); + final CountDownLatch releaseListener = new CountDownLatch(1); + domain.setImportClaimHook(() -> { + listenerHeld.countDown(); + awaitUninterruptibly(releaseListener); + }); + try + { + exporter.publish(new InitializeTargetMsg( + baseDN, EXPORTER_ID, DS_ID, DS_ID, exportedEntries().length, INIT_WINDOW)); + assertTrue(listenerHeld.await(30, TimeUnit.SECONDS), + "the listener thread did not reach the start of the import"); + + // On a thread of its own: disabling the domain waits for the listener thread. + final Thread disable = new Thread(domain::disable, "disable of " + EXAMPLE_DN); + disable.start(); + waitUntil(() -> !domain.isConnected(), "disabling the domain did not stop the session"); + releaseListener.countDown(); + disable.join(60_000); + assertFalse(disable.isAlive(), "disabling the domain did not end: the listener thread" + + " it waits for is still there"); + } + finally + { + releaseListener.countDown(); + domain.setImportClaimHook(null); + } + domain.enable(); + waitUntil(domain::isConnected, "the session was not started back by enable()"); + assertFalse(domain.ieRunning(), "the refused import left the context of its request claimed"); + + assertTrue(entryExists(entry.getName()), "the import ran over the session the disable" + + " stopped: the suffix was replaced by the nothing which arrived"); + assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal())) + .as("a total update answered over a session which is being stopped was started") + .hasSize(totalUpdatesStartedBefore); + assertThat(listenerDeaths()) + .as("the listener thread ended on an uncaught exception") + .hasSize(listenerDeathsBefore); + } + + /** + * A session restart decided while a total update out of this replica is running stops the + * session that export streams over (issue #1041). + *

+ * What the restart must leave alone is a total update into this replica: the data it is + * about to replace is read over the session, and the import is the thread the stop waits + * for. An export is not that: it streams out of a backend nothing is taking away, on a + * thread of its own, and a session stopped under it is the cut it reports to whoever asked + * for the total update - the same cut every other stop of the session is. The claim the + * restart makes for the import is not made here, and the session is stopped as it was + * before the claim. + *

+ * The export holds the context by standing where it waits for its target to report the + * start of the total update: the target is a broker of this test, and reports nothing. + */ + @Test(timeOut = 120_000) + public void aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver() 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()); + waitUntil(() -> domain.getReplicaInfos().containsKey(EXPORTER_ID), + "the exporter is not in the replicas of the domain: nothing to export into"); + + final AtomicReference exportFailure = new AtomicReference<>(); + final Thread export = new Thread(() -> { + try + { + domain.initializeRemote(EXPORTER_ID, null); + } + catch (Throwable t) + { + exportFailure.set(t); + } + }, "export of " + EXAMPLE_DN); + + // The restart is held after its decision, before the stop: what the case reads is the + // decision the export was found by, not the session which is down a moment later. + final CountDownLatch stopHeld = new CountDownLatch(1); + final CountDownLatch releaseStop = new CountDownLatch(1); + domain.setServiceStopHook(() -> { + stopHeld.countDown(); + awaitUninterruptibly(releaseStop); + }); + final CSN csn = gen.newCSN(); + final AtomicReference replayFailure = new AtomicReference<>(); + final Thread replay = new Thread(() -> { + try + { + replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN), + generatemods("description", "replayed while the export was running"), entryUUID)); + } + catch (Throwable t) + { + replayFailure.set(t); + } + }, "replay of " + csn); + try + { + export.start(); + waitUntil(() -> domain.ieRunning() || exportFailure.get() != null, + "the export did not claim the import context"); + assertNull(exportFailure.get(), + "the export failed before it claimed the context: " + exportFailure.get()); + + // A change whose entryUUID search never runs spends its attempts in place and asks + // for the session to be restarted, the way it does in the case above. + ShortCircuitPlugin.registerShortCircuit( + OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue()); + try + { + replay.start(); + assertTrue(stopHeld.await(30, TimeUnit.SECONDS), + "the restart left the session to the export: an export is not the owner a total" + + " update into this replica is"); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse"); + } + assertTrue(domain.ieRunning(), "the export ended before the restart was decided"); + } + finally + { + releaseStop.countDown(); + domain.setServiceStopHook(null); + } + + replay.join(60_000); + assertFalse(replay.isAlive(), "the restart did not end"); + assertNull(replayFailure.get(), "the replay failed: " + replayFailure.get()); + export.join(60_000); + assertFalse(export.isAlive(), "the export did not end once the session it streams over" + + " was stopped"); + assertThat(exportFailure.get()) + .as("the export was not told that the session it streams over was cut") + .isInstanceOf(DirectoryException.class); + waitUntil(domain::isConnected, "the session was not started back after the restart"); + assertFalse(domain.ieRunning(), "the export which was cut left its context claimed"); + } + /** * 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, @@ -587,11 +1014,39 @@ private static DN dnOf(String ldif) /** 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 : errorLogRecordsOf(msgId)) + { + if (record.contains(csn.toString())) + { + records.add(record); + } + } + return records; + } + + /** The records of the error log which carry the provided message id. */ + private static List errorLogRecordsOf(int msgId) { final List records = new ArrayList<>(); for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) { - if (record.contains("msgID=" + msgId) && record.contains(csn.toString())) + if (record.contains("msgID=" + msgId)) + { + records.add(record); + } + } + return records; + } + + /** The records of the error log which report the listener thread of the domain ending abnormally. */ + private static List listenerDeaths() + { + final List records = new ArrayList<>(); + for (String record : errorLogRecordsOf(ERR_UNCAUGHT_THREAD_EXCEPTION.ordinal())) + { + if (record.contains("listener for domain \"" + EXAMPLE_DN + "\"")) { records.add(record); } @@ -599,6 +1054,37 @@ private static List errorLogRecordsOf(int msgId, CSN csn) return records; } + private static void waitUntil(BooleanSupplier condition, String failure) throws InterruptedException + { + final long deadline = System.currentTimeMillis() + 30_000; + while (!condition.getAsBoolean()) + { + assertTrue(System.currentTimeMillis() < deadline, failure); + Thread.sleep(20); + } + } + + private static void awaitUninterruptibly(CountDownLatch latch) + { + boolean interrupted = false; + while (true) + { + try + { + latch.await(); + break; + } + catch (InterruptedException e) + { + interrupted = true; + } + } + if (interrupted) + { + Thread.currentThread().interrupt(); + } + } + private void replayMsg(UpdateMsg updateMsg) throws InterruptedException { domain.processUpdate(updateMsg);