From c6650aedac5c1ade8aee9e7a76eb0dd100a3e179 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 02 Nov 2026 15:39:56 +0000
Subject: [PATCH] [#1246] Keep the messages of the last local changes, and have a recovery publish the changes done before it began from them, as they were done (#1251)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java | 191 ++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java | 88 +
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java | 55 +
opendj-server-legacy/src/main/java/org/opends/server/api/LocalBackend.java | 35
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 22
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java | 466 ++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java | 10
opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java | 2
opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java | 6
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java | 250 +++++
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java | 81 +
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 621 ++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java | 225 +++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 470 +++++++++-
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 73 +
15 files changed, 2,465 insertions(+), 130 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/api/LocalBackend.java b/opendj-server-legacy/src/main/java/org/opends/server/api/LocalBackend.java
index ea42838..88c8de3 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/api/LocalBackend.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/api/LocalBackend.java
@@ -20,6 +20,8 @@
import static org.opends.messages.BackendMessages.*;
import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -76,6 +78,39 @@
{
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+ /** Process-lifetime write barriers for data which cannot safely be used after an online replacement. */
+ private static final Map<DN, LocalizableMessage> writeFailures = new ConcurrentHashMap<>();
+
+ /**
+ * Prevents writes overlapping a failed data replacement until the server process is restarted.
+ * An in-core restart keeps the barrier, as it keeps the JVM running.
+ * The barrier survives backend and synchronization-provider reconfiguration.
+ * @param baseDN the suffix whose data cannot safely be written
+ * @param reason the reason to report to the operator and to rejected updates
+ */
+ public static void quarantineWrites(DN baseDN, LocalizableMessage reason)
+ {
+ writeFailures.putIfAbsent(baseDN, reason);
+ }
+
+ /**
+ * Returns the reason writes overlapping a suffix are forbidden, or null if none are forbidden.
+ * Ancestors are included because deleting or moving a subtree affects the suffix below it.
+ * @param dn the DN of the entry or subtree an operation changes
+ * @return the recorded failure, or null
+ */
+ public static LocalizableMessage getWriteFailure(DN dn)
+ {
+ for (Map.Entry<DN, LocalizableMessage> failure : writeFailures.entrySet())
+ {
+ if (dn.isSubordinateOrEqualTo(failure.getKey()) || dn.isSuperiorOrEqualTo(failure.getKey()))
+ {
+ return failure.getValue();
+ }
+ }
+ return null;
+ }
+
/** Indicates whether this is a private backend or one that holds user data. */
private boolean isPrivateBackend;
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 9215e9f..f14c25e 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
@@ -128,6 +128,8 @@
import org.opends.server.replication.protocol.ModifyDnContext;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.OperationContext;
+import org.opends.server.replication.protocol.ProtocolVersion;
+import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.RoutableMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.UpdateMsg;
@@ -196,7 +198,11 @@
* This flag prevents the Replication Domain to disable/enable itself when
* it is the event initiator
*/
- private boolean ignoreBackendInitializationEvent;
+ private volatile boolean ignoreBackendInitializationEvent;
+ /** Exact backend whose enable may have been ignored while an aborted import was stopping. */
+ private volatile String backendAwaitingInitialization;
+ /** This domain must not resume against data whose post-import state could not be loaded. */
+ private volatile boolean failedImportCleanup;
private volatile boolean serverShutdownRequested;
@@ -279,19 +285,40 @@
@Override
public void performBackendPostInitializationProcessing(LocalBackend<?> backend) {
- if (!ignoreBackendInitializationEvent
- && getBackend().getBackendID().equals(backend.getBackendID())) {
- enable();
+ // The import invokes this callback on its listener. A concurrent disable may hold
+ // serviceStateLock while joining that listener, so the outside guard is essential.
+ // Backend configuration also holds a global lock: skip foreign domains before locking.
+ if (!ignoreBackendInitializationEvent && isBackendForInitialization(backend))
+ {
+ synchronized (serviceStateLock)
+ {
+ if (!ignoreBackendInitializationEvent && disabled && isBackendForInitialization(backend))
+ {
+ enable();
+ }
+ }
}
}
+ private boolean isBackendForInitialization(LocalBackend<?> backend)
+ {
+ final String awaiting = backendAwaitingInitialization;
+ if (awaiting != null)
+ {
+ // A disabled nested suffix can resolve to its parent until its own backend reopens.
+ return awaiting.equals(backend.getBackendID());
+ }
+ final LocalBackend<?> current = getBackend();
+ return current != null && current.getBackendID().equals(backend.getBackendID());
+ }
+
@Override
public void performBackendPreFinalizationProcessing(LocalBackend<?> backend) {
// Do not disable itself during a shutdown
// And ignore the event if this replica is the event trigger (e.g. importing).
if (!ignoreBackendInitializationEvent
&& !serverShutdownRequested
- && getBackend().getBackendID().equals(backend.getBackendID())) {
+ && isBackendForInitialization(backend)) {
disable();
}
}
@@ -612,7 +639,7 @@
* backend the replay would apply a change to is deregistered for the length of it.
* <p>
* 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
+ * one road which does not set that flag: {@link #preBackendImport(LocalBackend, BackendImportState)} 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
@@ -778,6 +805,7 @@
{
wait(1000);
}
+ restoreBackendAfterAbortedImport();
if (!disabled && !ieRunning())
{
saveState();
@@ -1063,6 +1091,11 @@
DSRSShutdownSync dsrsShutdownSync) throws ConfigException
{
super(configuration, -1);
+ final LocalizableMessage failure = LocalBackend.getWriteFailure(getBaseDN());
+ if (failure != null)
+ {
+ throw new ConfigException(failure);
+ }
this.updateToReplayQueue = updateToReplayQueue;
this.dsrsShutdownSync = dsrsShutdownSync;
@@ -5622,6 +5655,8 @@
* it lists after this would be the one left behind.
*/
remotePendingChanges.clear();
+ // An explicit disable wins over a recovery requested by the listener we just joined.
+ backendAwaitingInitialization = null;
/*
* The recovery from a failed replay is over as well: the change it was asking for
* is gone with the pending changes, so a leftover request would have a replay thread
@@ -5686,7 +5721,13 @@
* lock is released before returning for the same reason - what keeps the replay out is
* the flag, not the lock.
*/
- private void awaitReplayDrained()
+ private boolean awaitReplayDrained()
+ {
+ return awaitReplayDrained(true);
+ }
+
+ /** Waits for replay, optionally reporting the consequences of taking the domain down. */
+ private boolean awaitReplayDrained(boolean goingDown)
{
boolean drained = false;
boolean interrupted = false;
@@ -5710,7 +5751,7 @@
if (drained)
{
replayWriteLock.unlock();
- return;
+ return true;
}
/*
* The change which is being applied may reach the backend without being recorded in the
@@ -5719,18 +5760,20 @@
* restore, a backend being taken offline - for as long as a backend which stopped
* answering takes to answer.
*
- * An interrupted wait is reported as what it is: it says nothing about how long the
+ * On shutdown an interrupted wait is reported as what it is; an import reports its own
+ * preparation failure instead. An interruption says nothing about how long the
* replay of this domain takes, and the timeout it never spent would have an operator
* reading a backend which is slow into it.
*/
- if (interrupted)
+ if (goingDown && interrupted)
{
logger.warn(WARN_REPLAY_DRAIN_INTERRUPTED, getBaseDN());
}
- else
+ else if (goingDown)
{
logger.warn(WARN_REPLAY_NOT_DRAINED, getBaseDN(), replayDrainTimeoutInMs);
}
+ return false;
}
/**
@@ -5766,36 +5809,52 @@
* If there is no such information check if there is a
* ReplicaUpdateVector entry and translate it into a state
* and generationId.
+ * <p>
+ * The messages kept of the last local changes are forgotten first, whether the state is read or not:
+ * the data was replaced, by a restore, an import or a total update, and a later recovery would
+ * publish changes done on the data it replaced - see {@link #forgetTheLocalChangesTheDataLacks()}.
* @exception DirectoryException Thrown when an error occurs.
*/
private void loadDataState() throws DirectoryException
{
+ loadDataState(false);
+ }
+
+ /** Loads state after a replacement, optionally refusing every failed read. */
+ private void loadDataState(boolean strict) throws DirectoryException
+ {
+ // Which of them the data lacks is not looked up: the read of the state may fail, and this needs none.
+ // A total update also forgets before enabling the backend. This second forget can mark operations
+ // on the new data, sending their range back to the historical rebuild. It also covers operations
+ // still pending as the state is reloaded.
+ pendingChanges.forgetRecentChanges();
state.clearInMemory();
- state.loadState();
- bringAStateWithNoChangeOfThisReplicaUpToTheData();
+ state.loadState(strict);
+ bringAStateWithNoChangeOfThisReplicaUpToTheData(strict);
getGenerator().adjust(state.getMaxCSN(getServerId()));
- forgetTheLocalChangesTheDataLacks();
+ forgetTheLocalChangesTheDataLacks(strict);
// Retrieves the generation ID associated with the data imported
- generationId = loadGenerationId();
+ generationId = loadGenerationId(strict);
}
/**
* Drops the deletes kept aside, and the local changes a recovery kept, which the data the
* ServerState was just read from does not reflect (issue #1221): it was restored from a backup, or
* imported, since they were done, and a recovery which published them would have the other
- * replicas delete or change entries which this one holds as they were.
+ * replicas delete or change entries which this one holds as they were. The messages kept of the
+ * last local changes are forgotten for the same reason, by {@link #loadDataState()}.
* <p>
* What comes after the newest change of this replica the state holds is not in the data: the state
* read back covers the changes the historical information of the entries holds, whether the save
- * which was to write them wrote or not (see {@link #bringAStateWithNoChangeOfThisReplicaUpToTheData()})
+ * which was to write them wrote or not (see {@link #bringAStateWithNoChangeOfThisReplicaUpToTheData(boolean)})
* - except for a delete, which leaves no historical information, and which the state misses too when
* it could not be saved: a delete of the base entry takes the saved state away with it. A delete is
* therefore dropped only if its entry is in the data. A domain enabled back on its own data loses
* none of them, unless the search which brings its state up to the data does not go through, which
* {@code ERR_CANNOT_RECOVER_CHANGES} tells.
*/
- private void forgetTheLocalChangesTheDataLacks()
+ private void forgetTheLocalChangesTheDataLacks(boolean strict) throws DirectoryException
{
final CSN newest = state.getMaxCSN(getServerId());
@@ -5809,7 +5868,7 @@
final List<CSN> deletesTheDataLacks = new ArrayList<>();
for (FakeDelOperation delete : deletes)
{
- if (holdsTheEntry(delete.getEntryUUID()))
+ if (holdsTheEntry(delete.getEntryUUID(), strict))
{
deletesTheDataLacks.add(delete.getCSN());
}
@@ -5822,7 +5881,7 @@
final List<CSN> keptChangesTheDataLacks = new ArrayList<>();
for (LDAPUpdateMsg kept : pendingChanges.keptChangesNewerThan(newest))
{
- if (!(kept instanceof DeleteMsg) || holdsTheEntry(kept.getEntryUUID()))
+ if (!(kept instanceof DeleteMsg) || holdsTheEntry(kept.getEntryUUID(), strict))
{
keptChangesTheDataLacks.add(kept.getCSN());
}
@@ -5844,7 +5903,7 @@
* reflect, and the add of the base entry dropped, while the deletes which removed the suffix are
* published. The search here starts at the first change the recovery kept, which bounds it.
*/
- private void bringAStateWithNoChangeOfThisReplicaUpToTheData()
+ private void bringAStateWithNoChangeOfThisReplicaUpToTheData(boolean strict) throws DirectoryException
{
if (state.getMaxCSN(getServerId()) != null)
{
@@ -5855,7 +5914,7 @@
{
return;
}
- final CSN newest = newestChangeOfThisReplicaInTheDataFrom(kept.get(0).getCSN());
+ final CSN newest = newestChangeOfThisReplicaInTheDataFrom(kept.get(0).getCSN(), strict);
if (newest != null)
{
state.update(newest);
@@ -5870,7 +5929,7 @@
* changes are then taken for changes the data does not hold, as they are when the search which
* brings the state up to them as it is loaded does not go through
*/
- private CSN newestChangeOfThisReplicaInTheDataFrom(CSN from)
+ private CSN newestChangeOfThisReplicaInTheDataFrom(CSN from, boolean strict) throws DirectoryException
{
final InternalSearchOperation op;
try
@@ -5879,6 +5938,10 @@
}
catch (Exception e)
{
+ if (strict)
+ {
+ throw new DirectoryException(ResultCode.OTHER, ERR_CANNOT_RECOVER_CHANGES.get(getBaseDN()), e);
+ }
logger.traceException(e);
logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
return null;
@@ -5888,6 +5951,10 @@
// A suffix whose base entry is gone holds no change.
if (op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
{
+ if (strict)
+ {
+ throw new DirectoryException(op.getResultCode(), ERR_CANNOT_RECOVER_CHANGES.get(getBaseDN()));
+ }
logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
}
return null;
@@ -5912,7 +5979,7 @@
* which did not go through counts as one which found it: a delete then is not published, rather
* than published over an entry which may be there.
*/
- private boolean holdsTheEntry(String entryUUID)
+ private boolean holdsTheEntry(String entryUUID, boolean strict) throws DirectoryException
{
try
{
@@ -5920,10 +5987,41 @@
}
catch (SearchFailedException e)
{
+ if (strict)
+ {
+ throw new DirectoryException(ResultCode.OTHER, ERR_CANNOT_RECOVER_CHANGES.get(getBaseDN()), e);
+ }
return true;
}
}
+ /** Rechecks an enable which cleanup may have ignored, from a thread that can join the listener. */
+ private void restoreBackendAfterAbortedImport()
+ {
+ if (backendAwaitingInitialization == null || ignoreBackendInitializationEvent || ieRunning())
+ {
+ return;
+ }
+ synchronized (serviceStateLock)
+ {
+ final String backendID = backendAwaitingInitialization;
+ if (backendID != null && disabled && !shutdown.get() && !serverShutdownRequested
+ && !failedImportCleanup && !ignoreBackendInitializationEvent && !ieRunning()
+ && getServerContext().getBackendConfigManager().hasLocalBackend(backendID))
+ {
+ try
+ {
+ enable();
+ }
+ catch (Exception e)
+ {
+ // An unsuccessful restart must not terminate the state checkpointer.
+ logger.error(ERR_INIT_DOMAIN_RESTORE_FAILED, getBaseDN(), backendID, stackTraceToSingleLineString(e));
+ }
+ }
+ }
+ }
+
/**
* Enable back the domain after a previous disable.
* The domain will connect back to a replication Server and
@@ -5936,6 +6034,12 @@
{
synchronized (serviceStateLock)
{
+ if (failedImportCleanup)
+ {
+ return;
+ }
+ // Consume the missed event before a new listener can request another recovery.
+ backendAwaitingInitialization = null;
/*
* Cleared here as well as by disable(): a request made after disable()'s own clear -
* every road which releases a change asks for the restart, whether or not the domain
@@ -6082,6 +6186,16 @@
return result;
}
+ /** Saves a recomputed ID, making a failed write fatal after an import. */
+ private void saveGenerationId(long id, boolean strict) throws DirectoryException
+ {
+ final ResultCode result = saveGenerationId(id);
+ if (strict && result != ResultCode.SUCCESS && result != ResultCode.NO_SUCH_OBJECT)
+ {
+ throw new DirectoryException(result, ERR_UPDATING_GENERATION_ID.get(getBaseDN(), result.getName()));
+ }
+ }
+
/**
* Load the GenerationId from the root entry of the domain
* from the REPLICATION_GENERATION_ID attribute in database
@@ -6092,6 +6206,12 @@
*/
private long loadGenerationId() throws DirectoryException
{
+ return loadGenerationId(false);
+ }
+
+ /** Loads the generation ID, refusing read or persistence failures after an online import. */
+ private long loadGenerationId(boolean strict) throws DirectoryException
+ {
if (logger.isTraceEnabled())
{
logger.trace("Attempt to read generation ID from DB " + getBaseDN());
@@ -6116,6 +6236,10 @@
if (search.getResultCode() != ResultCode.NO_SUCH_OBJECT)
{
String errorMsg = search.getResultCode().getName() + " " + search.getErrorMessage();
+ if (strict)
+ {
+ throw new DirectoryException(search.getResultCode(), ERR_SEARCHING_GENERATION_ID.get(getBaseDN(), errorMsg));
+ }
logger.error(ERR_SEARCHING_GENERATION_ID, getBaseDN(), errorMsg);
}
}
@@ -6132,6 +6256,10 @@
if (attr.size()>1)
{
String errorMsg = "#Values=" + attr.size() + " Must be exactly 1 in entry " + resultEntry.toLDIFString();
+ if (strict)
+ {
+ throw new DirectoryException(ResultCode.OTHER, ERR_LOADING_GENERATION_ID.get(getBaseDN(), errorMsg));
+ }
logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), errorMsg);
}
else if (attr.size() == 1)
@@ -6143,6 +6271,11 @@
}
catch(Exception e)
{
+ if (strict)
+ {
+ throw new DirectoryException(ResultCode.OTHER,
+ ERR_LOADING_GENERATION_ID.get(getBaseDN(), stackTraceToSingleLineString(e)), e);
+ }
logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e));
}
}
@@ -6153,7 +6286,7 @@
if (!found)
{
aGenerationId = computeGenerationId();
- saveGenerationId(aGenerationId);
+ saveGenerationId(aGenerationId, strict);
if (logger.isTraceEnabled())
{
@@ -6347,12 +6480,13 @@
* @throws DirectoryException
* If the backend could not be disabled or locked exclusively.
*/
- private void preBackendImport(LocalBackend<?> backend) throws DirectoryException
+ private void preBackendImport(LocalBackend<?> backend, BackendImportState progress) throws DirectoryException
{
// Prevent the processing of the backend finalisation event as the import will disable the attached backend
ignoreBackendInitializationEvent = true;
// FIXME setBackendEnabled should be part of TaskUtils ?
+ progress.backendDisabled = true;
TaskUtils.disableBackend(backend.getBackendID());
// Acquire an exclusive lock for the backend.
@@ -6364,6 +6498,7 @@
logger.error(message);
throw new DirectoryException(ResultCode.OTHER, message);
}
+ progress.lockHeld = true;
}
/**
@@ -6378,6 +6513,7 @@
LocalBackend<?> backend = getBackend();
LDIFImportConfig importConfig = null;
+ final BackendImportState progress = new BackendImportState();
ImportExportContext ieCtx = getImportExportContext();
try
{
@@ -6389,7 +6525,25 @@
* it is the session the entries are about to stream over, on this very thread.
*/
importingData = true;
- awaitReplayDrained();
+ boolean drained;
+ try
+ {
+ drained = awaitReplayDrained(false) && state.suspendSaves(replayDrainTimeoutInMs);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ drained = false;
+ }
+ if (!drained)
+ {
+ final LocalizableMessage reason = Thread.currentThread().isInterrupted()
+ ? ERR_INIT_PREPARATION_INTERRUPTED.get(getBaseDN())
+ : ERR_INIT_OPERATIONS_NOT_DRAINED.get(getBaseDN());
+ logger.error(reason);
+ ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.UNAVAILABLE, reason));
+ return;
+ }
if (!backend.supports(BackendOperation.LDIF_IMPORT))
{
@@ -6414,70 +6568,126 @@
ExistingFileBehavior.OVERWRITE);
// Process import
- preBackendImport(backend);
+ preBackendImport(backend, progress);
+ progress.replacingData = true;
backend.importLDIF(importConfig, getServerContext());
}
catch(Exception e)
{
+ if (e instanceof InterruptedException)
+ {
+ Thread.currentThread().interrupt();
+ }
ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(e))));
}
finally
{
+ if (progress.replacingData)
+ {
+ // No message of the replaced data may survive a failed cleanup or state reload.
+ pendingChanges.forgetRecentChanges();
+ }
try
{
// Cleanup
if (importConfig != null)
{
importConfig.close();
- closeBackendImport(backend); // Re-enable backend
- backend = getBackend();
+ }
+ if (progress.backendDisabled)
+ {
+ closeBackendImport(backend, progress);
+ }
+ else
+ {
+ ignoreBackendInitializationEvent = false;
}
- loadDataState();
+ if (progress.replacingData)
+ {
+ loadDataState(true);
+ }
- if (ieCtx.getException() != null)
+ if (progress.replacingData && ieCtx.getException() != null)
{
// When an error occurred during an import, most of times
// the generationId coming in the root entry of the imported data,
// is not valid anymore (partial data in the backend).
generationId = computeGenerationId();
- saveGenerationId(generationId);
+ saveGenerationId(generationId, true);
}
}
- catch (DirectoryException fe)
+ catch (Exception fe)
{
- // If we already catch an Exception it's quite possible
- // that the loadDataState() and setGenerationId() fail
- // so we don't bother about the new Exception.
- // However if there was no Exception before we want
- // to return this Exception to the task creator.
- ieCtx.setExceptionIfNoneSet(new DirectoryException(
- ResultCode.OTHER,
- ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe))));
+ if (!progress.replacingData)
+ {
+ // Preparation changed no data: neither a refused disable nor a refused lock warrants
+ // a process-lifetime quarantine. Report cleanup too, preserving the original failure.
+ final LocalizableMessage reason = ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe));
+ logger.error(reason);
+ ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER, reason, fe));
+ // A DN lookup could find the parent backend of a disabled nested suffix.
+ if (!getServerContext().getBackendConfigManager().hasLocalBackend(backend.getBackendID()))
+ {
+ // The backend could not be enabled back. Keep the domain stopped until a later backend
+ // enable reloads its state. This runs on the listener: it cannot join itself, nor take
+ // serviceStateLock while a concurrent disable may hold it and wait for this listener.
+ disabled = true;
+ stopServiceAfterAbortedImport();
+ remotePendingChanges.clear();
+ // A backend enable between the availability check and the final flag clear is
+ // deliberately ignored by the callback. Recheck it after import completion.
+ backendAwaitingInitialization = backend.getBackendID();
+ logger.error(ERR_INIT_DOMAIN_AWAITING_BACKEND, getBaseDN(), backend.getBackendID());
+ }
+ }
+ else
+ {
+ // A successful import and a failed import both need trustworthy cleanup before reuse.
+ // Preserve an earlier import error in the terminal reason, rather than hiding this failure.
+ final LocalizableMessage reason = ERR_INIT_DOMAIN_STOPPED.get(getBaseDN(),
+ ieCtx.getException() == null ? "" : ieCtx.getException().getLocalizedMessage(),
+ stackTraceToSingleLineString(fe));
+ failedImportCleanup = true;
+ disabled = true;
+ LocalBackend.quarantineWrites(getBaseDN(), reason);
+ logger.error(reason);
+ ieCtx.setException(new DirectoryException(ResultCode.OTHER, reason));
+ stopServicePermanently();
+ }
}
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
+ * Once replacement has begun, the changes listed as pending must not outlive their data,
+ * 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 restarts
- * they asked for on their way out, while the total update owned the session, and
- * any asked for before it did, go with them, as do the deliveries folded into no
- * warning: the caller starts the session again from the reloaded state. A replay
- * thread which reaches its give-up after this clear finds its change unlisted and
+ * and the replay threads gave up every attempt while the flag was set. A replay thread
+ * which reaches its give-up after this clear finds its change unlisted and
* asks for nothing.
*/
- remotePendingChanges.clear();
+ if (progress.replacingData)
+ {
+ remotePendingChanges.clear();
+ }
+ // The caller reconnects even after abandoning replacement, unless the service was stopped
+ // above. That reconnect satisfies requests already made during the drain; if the backend
+ // is unavailable, its later enable does so instead. A request made later is a new request.
sessionRestarts.clear();
consecutiveSessionRestarts.set(0);
foldedReplayRetryWarnings.set(0);
+ if (!failedImportCleanup)
+ {
+ state.resumeSaves();
+ }
importingData = false;
+ // In particular, an enable refused during preparation must not suppress the later event
+ // which restores a temporarily stopped domain. All import bookkeeping is settled first.
+ ignoreBackendInitializationEvent = false;
}
}
@@ -6487,18 +6697,25 @@
}
}
+ /** Tracks the resources actually acquired by this import, including failed preparation. */
+ private static final class BackendImportState
+ {
+ private boolean backendDisabled;
+ private boolean lockHeld;
+ private boolean replacingData;
+ }
+
/**
- * Make post import operations.
- * @param backend The backend implied in the import.
- * @exception DirectoryException Thrown when an error occurs.
+ * Releases the import's lock and enables its backend.
+ * @param backend the backend being imported
+ * @param progress resources acquired by this import
+ * @throws DirectoryException if the lock cannot be released or the backend cannot be enabled
*/
- private void closeBackendImport(LocalBackend<?> backend) throws DirectoryException
+ private void closeBackendImport(LocalBackend<?> backend, BackendImportState progress) throws DirectoryException
{
String lockFile = LockFileManager.getBackendLockFileName(backend);
StringBuilder failureReason = new StringBuilder();
-
- // Release lock
- if (!LockFileManager.releaseLock(lockFile, failureReason))
+ if (progress.lockHeld && !LockFileManager.releaseLock(lockFile, failureReason))
{
LocalizableMessage message =
WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.get(backend.getBackendID(), failureReason);
@@ -6506,11 +6723,10 @@
throw new DirectoryException(ResultCode.OTHER, message);
}
+ progress.lockHeld = false;
TaskUtils.enableBackend(backend.getBackendID());
-
- // Restore the processing of backend finalization events.
+ progress.backendDisabled = false;
ignoreBackendInitializationEvent = false;
-
}
/**
@@ -6625,6 +6841,12 @@
{
// Check that there is not already a domain with the same DN
final DN dn = configuration.getBaseDN();
+ final LocalizableMessage failure = LocalBackend.getWriteFailure(dn);
+ if (failure != null)
+ {
+ unacceptableReasons.add(failure);
+ return false;
+ }
LDAPReplicationDomain domain = MultimasterReplication.findDomain(dn, null);
if (domain != null && domain.getBaseDN().equals(dn))
{
@@ -7274,10 +7496,12 @@
* Build the list of changes that have been processed by this server after the
* CSN given as a parameter and publish them using the given session.
* <p>
- * The changes done before the recovery began are rebuilt from the historical information of
- * their entries, in rounds, each one from where the round before it stopped, up to the last of
- * them: each of them has its entry written by then, so the search of a round finds it, or a later
- * change of the same values, which supersedes it. The local changes pushed since the recovery began
+ * The changes done before the recovery began are published from the messages the domain kept of its
+ * last local changes, as they were done; the ones done before the first of those are rebuilt from
+ * the historical information of their entries, in rounds, each one from where the round before it
+ * stopped, up to there: each of them has its entry written by then, so the search of a round finds
+ * it, or a later change of the same values, which supersedes it - and the replication server never
+ * holds the change superseded (issue #1246). The local changes pushed since the recovery began
* are kept by the pending changes, with their messages, and they are published next, in order. The
* recovery ends with the round which publishes up to the last local change, no change having been
* pushed meanwhile; once a round has caught up with the local changes and some were pushed while it
@@ -7399,6 +7623,8 @@
private final ReplicationBroker recoveryBroker;
private final Session recoverySession;
private final RSUpdater publisher;
+ /** Whether this run said it rebuilds changes from the historical information: it says so once. */
+ private boolean rebuildingReported;
private MissingChanges(ReplicationBroker recoveryBroker, Session recoverySession, RSUpdater publisher)
{
@@ -7463,9 +7689,8 @@
/**
* Runs one round: publishes the changes done before the recovery began which come after
- * {@code fromCSN}, rebuilt from the historical information, up to the end of a search window;
- * past them, the local changes the recovery kept; or, if it kept too many to keep any, the
- * changes rebuilt from the historical information up to the last local change.
+ * {@code fromCSN} - see {@link #publishChangesDone(CSN, CSN)}; past them, the local changes the
+ * recovery kept; or, if it kept too many to keep any, the changes done up to the last local change.
*
* @param heldBack
* whether the round runs while the local changes are held back
@@ -7482,22 +7707,114 @@
final PendingChanges.KeptChanges keeping = pendingChanges.keptChanges();
if (keeping == null)
{
- return publishMissingChanges(fromCSN, null);
+ return publishChangesDone(fromCSN, null);
}
if (keeping.searchUpTo != null && fromCSN.isOlderThan(keeping.searchUpTo))
{
- return publishMissingChanges(fromCSN, keeping.searchUpTo);
+ return publishChangesDone(fromCSN, keeping.searchUpTo);
}
final List<LDAPUpdateMsg> kept = pendingChanges.keptChangesAfter(fromCSN);
if (kept == null)
{
// It kept too many since.
- return publishMissingChanges(fromCSN, null);
+ return publishChangesDone(fromCSN, null);
}
return publishKeptChanges(fromCSN, kept, heldBack);
}
/**
+ * Runs a round which publishes the changes of this replica which come after {@code fromCSN}, up to
+ * {@code upTo} - where the search of the recovery ends - or, if it is {@code null}, up to the last
+ * local change: the messages the domain kept of them, as they were done, or - for the changes done
+ * before the first of those - the changes rebuilt from the historical information, up to there.
+ * <p>
+ * Rebuilt, a change whose values a later one replaced is published as the later one, and the add of
+ * an entry deleted since not at all, and the replication server of the session, which takes no change
+ * older than the newest one it holds of this replica, never holds them (issue #1246). The domain says
+ * so, once per run, in the first round which finds a change or a kept delete - see
+ * {@link #publishMissingChanges(CSN, CSN)}: its replication server is further behind than the messages
+ * it keeps go back.
+ *
+ * @return the round, or {@code null} if the domain was shut down meanwhile
+ * @throws MissingChangesSearchFailedException
+ * if the search did not go through: the round published nothing
+ */
+ private MissingChangesRound publishChangesDone(CSN fromCSN, CSN upTo) throws Exception
+ {
+ final PendingChanges.RecentChanges recent = pendingChanges.recentChangesAfter(fromCSN, upTo);
+ if (recent.messages != null)
+ {
+ return publishRecentChanges(fromCSN, recent, upTo == null);
+ }
+ // The messages kept go on from where the ones the domain kept begin, if they begin before the end.
+ final CSN rebuiltUpTo = recent.keptAfter != null && (upTo == null || recent.keptAfter.isOlderThan(upTo))
+ ? recent.keptAfter : upTo;
+ return publishMissingChanges(fromCSN, rebuiltUpTo);
+ }
+
+ /**
+ * Runs a round which publishes the messages the domain kept of its local changes after
+ * {@code fromCSN}, in order and as they were done, up to the end the domain read them up to; a round
+ * which went up to the last local change ends the recovery if no change was pushed since.
+ *
+ * @param reachesLastChange
+ * whether the messages go up to the last local change, rather than to where the search of
+ * the recovery ends
+ * @return the round, or {@code null} if the domain was shut down meanwhile
+ */
+ private MissingChangesRound publishRecentChanges(CSN fromCSN, PendingChanges.RecentChanges recent,
+ boolean reachesLastChange) throws Exception
+ {
+ for (LDAPUpdateMsg msg : recent.messages)
+ {
+ if (shutdown.get())
+ {
+ return null;
+ }
+ if (importingData)
+ {
+ return refusedForTheImport(fromCSN);
+ }
+ if (!recoveryBroker.publishRecovery(notAssured(msg), recoverySession))
+ {
+ return refused(fromCSN, msg.getCSN());
+ }
+ if (msg instanceof DeleteMsg)
+ {
+ deleteKeptAsidePublished(msg.getCSN());
+ }
+ }
+ final CSN publishedUpTo = recent.upTo != null && recent.upTo.isNewerThan(fromCSN) ? recent.upTo : fromCSN;
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("publish loop of the recent changes >" + fromCSN
+ + " published=" + recent.messages.size() + " upTo=" + publishedUpTo));
+ }
+ final boolean endedRecovery =
+ reachesLastChange && !pendingChanges.recoveryUntil(publishedUpTo, this::endRecovery);
+ return new MissingChangesRound(publishedUpTo, reachesLastChange, true, endedRecovery);
+ }
+
+ /**
+ * Returns the message to publish for a local change the domain kept: itself, or - if it was
+ * published assured - a copy which is not, as a recovery publishes the changes it rebuilds and the
+ * ones it kept: an operation which still waits for the acknowledgement of one gets none, and times
+ * out. The message kept was handed to the broker already, so it is not changed in its place: the
+ * operation which published it may still wait for its acknowledgement.
+ */
+ private LDAPUpdateMsg notAssured(LDAPUpdateMsg msg) throws Exception
+ {
+ if (!msg.isAssured())
+ {
+ return msg;
+ }
+ final short version = ProtocolVersion.getCurrentVersion();
+ final LDAPUpdateMsg copy = (LDAPUpdateMsg) ReplicationMsg.generateMsg(msg.getBytes(version), version);
+ copy.setNotAssured();
+ return copy;
+ }
+
+ /**
* A total update replaces the data of this replica: what a recovery would publish meanwhile was
* done on the data it replaces, and the changes it kept which the new data does not reflect are
* dropped once the import is over. The round is refused, and the session started again after the
@@ -7581,6 +7898,17 @@
opsToSend.putAll(replayOperations.subMap(fromCSN, false, bound, true));
}
}
+ /*
+ * A delete is published whole, but its entry may have held earlier changes which the search
+ * cannot find. A value replaced by a later kept message may leave no trace here at all. An
+ * empty window alone does not warrant a warning: after a restart it can precede the first
+ * message kept without containing any change.
+ */
+ if (!rebuildingReported && !opsToSend.isEmpty())
+ {
+ rebuildingReported = true;
+ logger.warn(WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION, getBaseDN(), fromCSN, lastChange);
+ }
for (FakeOperation opToSend : opsToSend.values())
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java
index 49181cf..1b7a39b 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java
@@ -424,6 +424,16 @@
try
{
LDAPReplicationDomain rd = createNewDomain(configuration);
+ if (rd == null)
+ {
+ ccr.setResultCode(ResultCode.UNAVAILABLE);
+ final LocalizableMessage reason = LocalBackend.getWriteFailure(configuration.getBaseDN());
+ if (reason != null)
+ {
+ ccr.addMessage(reason);
+ }
+ return ccr;
+ }
if (State.RUNNING.equals(state.get()))
{
rd.start();
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
index 378cc37..ef2c4f7 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
@@ -19,9 +19,11 @@
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Callable;
@@ -56,9 +58,33 @@
*/
static final int MAX_KEPT_CHANGES = 10000;
static final long MAX_KEPT_BYTES = 32L * 1024 * 1024;
+ /**
+ * How many of the last local changes the domain keeps the messages of at most, and how much of the
+ * heap those take at most, as {@link #footprint(LDAPUpdateMsg)} reckons it: a recovery publishes the
+ * changes its replication server misses from them, and rebuilds the ones before them from the
+ * historical information - see {@link LastLocalChanges}. Under the load of the replication convergence
+ * run, a few hundred changes a second, a replication server more than two minutes behind is covered.
+ * <p>
+ * Each domain written to keeps that much, whether a recovery runs or not, so its bytes are a sixteenth
+ * of the heap at most; multiple written domains can keep that share each. A server given a small heap
+ * covers a shorter lag. A recovery round holds the messages it publishes besides: while it runs, as much
+ * again can stay on the heap.
+ */
+ static final int MAX_RECENT_CHANGES = 100000;
+ static final long MAX_RECENT_BYTES = Math.min(64L * 1024 * 1024, Runtime.getRuntime().maxMemory() / 16);
+ /**
+ * What a message kept takes besides twice its {@link LDAPUpdateMsg#size()}, which counts its encoded
+ * changes once, and its encoding holds them again: its DN, entry UUID and CSN, the objects of its
+ * changes, and its place in the map. Measured on the messages of the load of the convergence run,
+ * kept as they are here: 880 bytes of heap for a modify whose size() is 168, 1300 for an add whose
+ * size() is 370.
+ */
+ static final int MESSAGE_OVERHEAD = 560;
/** A map used to store the pending changes. */
private final TreeMap<CSN, PendingChange> pendingChanges = new TreeMap<>();
+ /** Local operations pending at a reload, which may have written to the replaced data. */
+ private final Set<CSN> pendingAtDataReload = new HashSet<>();
/**
* The {@link CSNGenerator} to use to create new unique CSNs
@@ -81,6 +107,10 @@
private Recovery recovery;
/** How many local changes a recovery keeps at most: fewer in the tests which make it run over. */
private int maxKeptChanges = MAX_KEPT_CHANGES;
+ /** The messages of the last local changes, which a recovery publishes as they were done. */
+ private final LastLocalChanges lastLocalChanges = new LastLocalChanges();
+ /** How many of the last local changes the domain keeps the messages of at most: fewer in tests. */
+ private int maxRecentChanges = MAX_RECENT_CHANGES;
/**
* Creates a new PendingChanges using the provided CSNGenerator.
@@ -108,6 +138,7 @@
synchronized void remove(CSN csn)
{
pendingChanges.remove(csn);
+ pendingAtDataReload.remove(csn);
}
/**
@@ -193,8 +224,8 @@
* {@code null} if none was pushed or the broker refused it. The announcement that a
* replica goes offline is the one message whose delivery the caller must know about:
* it is stored nowhere, so nothing publishes it again, while a change the broker
- * refuses is republished from the historical information of its entry on the next
- * session.
+ * refuses is republished on the next session, from the message kept of it or, once that
+ * is dropped, from the historical information of its entry.
*/
synchronized CSN pushCommittedChanges()
{
@@ -217,6 +248,16 @@
&& op != null
&& !op.isSynchronizationOperation())
{
+ // Kept before the domain state takes it, published or not: a recovery publishes it as it was done.
+ if (pendingAtDataReload.contains(firstChange.getCSN()))
+ {
+ // There is a gap in the messages kept after the reload; earlier ones cannot cover it.
+ lastLocalChanges.forget();
+ }
+ else
+ {
+ lastLocalChanges.add(firstChange.getCSN(), (LDAPUpdateMsg) msg);
+ }
if (!recoveringOldChanges)
{
domain.publish(msg);
@@ -268,6 +309,7 @@
// false warning: firstEntry will not be null if firstChange is not null
pendingChanges.remove(firstEntry.getKey());
+ pendingAtDataReload.remove(firstChange.getCSN());
// peek the oldest change
firstEntry = pendingChanges.firstEntry();
@@ -536,6 +578,185 @@
}
}
+ /**
+ * Returns the messages the domain kept of its local changes which come after the provided CSN, up
+ * to the provided one, in order, or tells where the messages it kept begin if they do not go back
+ * that far.
+ *
+ * @param after
+ * where the recovery has got to
+ * @param upTo
+ * where the changes to publish end, or {@code null} for the last local change
+ * @return the messages, or where the ones kept begin
+ */
+ synchronized RecentChanges recentChangesAfter(CSN after, CSN upTo)
+ {
+ return lastLocalChanges.after(after, upTo != null ? upTo : domain.getLastLocalChange());
+ }
+
+ /**
+ * Forgets the messages kept of the last local changes: the domain read its state from data which
+ * may lack their changes - it was restored or imported since - and a recovery which published them
+ * would have the other replicas do changes this one does not hold. Local operations still pending
+ * may have written to the replaced data even if they have not yet committed, so their messages are
+ * not kept when they are pushed. Later messages are kept, and a recovery rebuilds the changes before
+ * them from the historical information.
+ */
+ synchronized void forgetRecentChanges()
+ {
+ lastLocalChanges.forget();
+ for (PendingChange change : pendingChanges.values())
+ {
+ if (change.getOp() != null && !change.getOp().isSynchronizationOperation())
+ {
+ pendingAtDataReload.add(change.getCSN());
+ }
+ }
+ }
+
+ /**
+ * Sets how many of the last local changes the domain keeps the messages of at most: with none, a
+ * recovery rebuilds every change from the historical information.
+ *
+ * @param maxRecentChanges
+ * how many of the last local changes the domain keeps the messages of at most
+ */
+ @VisibleForTesting
+ synchronized void setMaxRecentChanges(int maxRecentChanges)
+ {
+ this.maxRecentChanges = maxRecentChanges;
+ lastLocalChanges.trim();
+ }
+
+ /**
+ * The messages of the last local changes, in order, which a recovery publishes as they were done.
+ * <p>
+ * A recovery publishes the changes its replication server misses of this replica, and rebuilt from
+ * the historical information of the entries, those are not the changes: it holds nothing of a change
+ * whose values a later one replaced, nor of the add of an entry deleted since. A replication server
+ * which took the later changes takes none of the older ones afterwards - from the peer which got them
+ * from this replica either, whose changes it takes in order, as a log ordered by CSN - so a directory
+ * server which failed over to a replication server behind it, or which the load balancing moved to
+ * one, left gaps in its changelog for good (issue #1246). The message of every local change pushed -
+ * published, refused by the broker, or left to a recovery - is kept here instead, for as many changes
+ * as {@link #maxRecentChanges} and {@link #MAX_RECENT_BYTES} allow, the oldest dropped first.
+ * <p>
+ * Every local change after {@link #keptAfter} has its message kept: the CSN right before the first one
+ * kept, and the newest one dropped once some were. A recovery from further back rebuilds the changes up
+ * to there from the historical information. It is not the last local change the domain state holds:
+ * the state read back once the data was restored, or once the domain was enabled again, can be older
+ * than changes done since the domain started - a delete leaves no historical information to bring it
+ * up to - and taking the messages kept for the changes after it skipped those changes.
+ */
+ private final class LastLocalChanges
+ {
+ private final TreeMap<CSN, LDAPUpdateMsg> messages = new TreeMap<>();
+ private long bytes;
+ /**
+ * Every local change after this one has its message kept; {@code null} if none was kept since the
+ * domain started or forgot them.
+ */
+ private CSN keptAfter;
+
+ private void add(CSN csn, LDAPUpdateMsg msg)
+ {
+ if (maxRecentChanges <= 0)
+ {
+ return;
+ }
+ if (keptAfter == null)
+ {
+ keptAfter = justBefore(csn);
+ }
+ // A change is pushed again when the publish of the push before threw: it is counted once.
+ final LDAPUpdateMsg replaced = messages.put(csn, msg);
+ bytes += footprint(msg) - (replaced != null ? footprint(replaced) : 0);
+ trim();
+ }
+
+ private void trim()
+ {
+ if (maxRecentChanges <= 0)
+ {
+ forget();
+ return;
+ }
+ while (!messages.isEmpty() && (messages.size() > maxRecentChanges || bytes > MAX_RECENT_BYTES))
+ {
+ final Entry<CSN, LDAPUpdateMsg> oldest = messages.pollFirstEntry();
+ bytes -= footprint(oldest.getValue());
+ keptAfter = oldest.getKey();
+ }
+ }
+
+ private void forget()
+ {
+ messages.clear();
+ bytes = 0;
+ keptAfter = null;
+ }
+
+ private RecentChanges after(CSN after, CSN upTo)
+ {
+ if (keptAfter == null)
+ {
+ return new RecentChanges(null, upTo, null);
+ }
+ if (after.isOlderThan(keptAfter))
+ {
+ return new RecentChanges(null, upTo, keptAfter);
+ }
+ final List<LDAPUpdateMsg> found = upTo != null && after.isOlderThan(upTo)
+ ? new ArrayList<>(messages.subMap(after, false, upTo, true).values())
+ : new ArrayList<LDAPUpdateMsg>();
+ return new RecentChanges(found, upTo, keptAfter);
+ }
+ }
+
+ /**
+ * Returns the CSN right before the provided one among the CSNs of its server: no CSN of that server
+ * comes in between. For a sequence number of 0, it is the greatest one of the millisecond before, as
+ * {@code Integer.MAX_VALUE} rather than -1: a search window of a recovery may start there, and the
+ * index of the historical information reads -1 as the greatest sequence number of its millisecond.
+ */
+ private static CSN justBefore(CSN csn)
+ {
+ return csn.getSeqnum() > 0
+ ? new CSN(csn.getTime(), csn.getSeqnum() - 1, csn.getServerId())
+ : new CSN(csn.getTime() - 1, Integer.MAX_VALUE, csn.getServerId());
+ }
+
+ /** Returns how much of the heap a message kept takes, as reckoned for {@link #MAX_RECENT_BYTES}. */
+ static long footprint(LDAPUpdateMsg msg)
+ {
+ return 2L * msg.size() + MESSAGE_OVERHEAD;
+ }
+
+ /** What the domain kept of the messages of its local changes from a CSN on, up to another one. */
+ static final class RecentChanges
+ {
+ /**
+ * The messages of the local changes after the CSN asked from, up to {@link #upTo}, in order;
+ * {@code null} if the messages kept do not go back that far, or the domain keeps none.
+ */
+ final List<LDAPUpdateMsg> messages;
+ /** Where the changes asked for end: the CSN asked for, or the last local change. */
+ final CSN upTo;
+ /**
+ * Every local change after this one has its message kept: where the changes to rebuild from the
+ * historical information end when {@link #messages} is {@code null}. {@code null} if no message is
+ * kept, and the changes are all to rebuild.
+ */
+ final CSN keptAfter;
+
+ private RecentChanges(List<LDAPUpdateMsg> messages, CSN upTo, CSN keptAfter)
+ {
+ this.messages = messages;
+ this.upTo = upTo;
+ this.keptAfter = keptAfter;
+ }
+ }
+
/** A recovery which keeps the local changes pushed while it runs. */
static final class KeptChanges
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java
index 7b7bf70..fce3e8d 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java
@@ -60,9 +60,11 @@
private final ServerState state;
/**
* Held by the save which is writing the state to the backend. It is taken
- * with {@link Lock#tryLock()} and never waited for: see {@link #save()}.
+ * without waiting by save(); the total-update preflight waits for it before replacing data.
*/
private final Lock saveLock = new ReentrantLock();
+ /** A total update must not let a snapshot of the old state reach the replacement data. */
+ private volatile boolean savesSuspended;
/**
* The attribute name used to store the state in the backend.
@@ -131,7 +133,7 @@
*/
public void save()
{
- if (state.isSaved())
+ if (savesSuspended || state.isSaved())
{
// Nothing to write: stay out of the way of whoever is writing.
return;
@@ -143,7 +145,7 @@
}
try
{
- if (state.isSaved())
+ if (savesSuspended || state.isSaved())
{
// The save which just completed carried what this one came to write.
return;
@@ -180,10 +182,51 @@
}
/**
+ * Stops new saves and waits for the one already writing, before replacing the data.
+ * A timeout leaves saves suspended: the caller must abandon the replacement before resuming them.
+ * @param timeoutMillis maximum time to wait for the previous write
+ * @return whether no previous write can still reach the backend
+ * @throws InterruptedException if the wait is interrupted
+ */
+ boolean suspendSaves(long timeoutMillis) throws InterruptedException
+ {
+ savesSuspended = true;
+ if (!saveLock.tryLock(timeoutMillis, java.util.concurrent.TimeUnit.MILLISECONDS))
+ {
+ return false;
+ }
+ saveLock.unlock();
+ return true;
+ }
+
+ /** Resumes saves after an abandoned replacement or after its new state has been loaded. */
+ void resumeSaves()
+ {
+ savesSuspended = false;
+ }
+
+ /**
* Load the ServerState from the backing entry in database to memory.
*/
public void loadState()
{
+ try
+ {
+ loadState(false);
+ }
+ catch (DirectoryException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ /**
+ * Loads state, refusing read failures after an online replacement rather than using stale fallbacks.
+ * @param strict whether a failed read must prevent replication from restarting
+ * @throws DirectoryException if a required read fails in strict mode
+ */
+ void loadState(boolean strict) throws DirectoryException
+ {
/*
* Whatever the state holds on the way in has, as far as this object knows,
* never been written: what follows only merges in what the backend holds.
@@ -195,14 +238,14 @@
final boolean hadCSNs = !state.isEmpty();
// try to load the state from the base entry.
- SearchResultEntry stateEntry = searchBaseEntry();
+ SearchResultEntry stateEntry = searchBaseEntry(strict);
if (stateEntry == null)
{
/*
The base entry does not exist yet in the database or was deleted.
Try to read the ServerState from the configuration instead.
*/
- stateEntry = searchConfigEntry();
+ stateEntry = searchConfigEntry(strict);
}
if (stateEntry != null)
@@ -216,7 +259,7 @@
* updated after the last write of the ServerState.
* Inconsistencies may append after a crash.
*/
- checkAndUpdateServerState();
+ checkAndUpdateServerState(strict);
if (hadCSNs)
{
@@ -230,7 +273,7 @@
*
* @return The base entry or null if no entry was found;
*/
- private SearchResultEntry searchBaseEntry()
+ private SearchResultEntry searchBaseEntry(boolean strict) throws DirectoryException
{
// Search the database entry that is used to periodically save the ServerState
final SearchRequest request = newSearchRequest(baseDN, SearchScope.BASE_OBJECT).addAttribute(REPLICATION_STATE);
@@ -239,7 +282,12 @@
if (resultCode != ResultCode.SUCCESS
&& resultCode != ResultCode.NO_SUCH_OBJECT)
{
- logger.error(ERR_ERROR_SEARCHING_RUV, search.getResultCode().getName(), search, search.getErrorMessage(), baseDN);
+ if (strict)
+ {
+ throw new DirectoryException(resultCode,
+ ERR_ERROR_SEARCHING_RUV.get(resultCode.getName(), search, search.getErrorMessage(), baseDN));
+ }
+ logger.error(ERR_ERROR_SEARCHING_RUV, resultCode.getName(), search, search.getErrorMessage(), baseDN);
return null;
}
return getFirstResult(search);
@@ -251,7 +299,7 @@
*
* @return The configuration Entry or null if no entry was found;
*/
- private SearchResultEntry searchConfigEntry()
+ private SearchResultEntry searchConfigEntry(boolean strict) throws DirectoryException
{
try
{
@@ -260,11 +308,19 @@
.setSizeLimit(1)
.addAttribute(REPLICATION_STATE);
final InternalSearchOperation op = getRootConnection().processSearch(request);
+ if (strict && op.getResultCode() != ResultCode.SUCCESS && op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
+ {
+ throw new DirectoryException(op.getResultCode(),
+ ERR_ERROR_SEARCHING_RUV.get(op.getResultCode().getName(), op, op.getErrorMessage(), baseDN));
+ }
return getFirstResult(op);
}
catch (DirectoryException e)
{
- // can not happen
+ if (strict)
+ {
+ throw e;
+ }
return null;
}
}
@@ -410,7 +466,7 @@
* This is done by using the HistoricalCsnOrderingMatchingRule
* and an ordering index for historical attribute
*/
- private final void checkAndUpdateServerState()
+ private void checkAndUpdateServerState(boolean strict) throws DirectoryException
{
// Retrieves the entries that have changed since the
// maxCsn stored in the serverState
@@ -428,13 +484,21 @@
op = LDAPReplicationDomain.searchForChangedEntries(baseDN,
serverStateMaxCSN, null);
}
- catch (Exception e)
+ catch (Exception e)
{
+ if (strict)
+ {
+ throw new DirectoryException(ResultCode.OTHER, ERR_CANNOT_RECOVER_CHANGES.get(baseDN), e);
+ }
return;
}
if (op.getResultCode() != ResultCode.SUCCESS)
{
+ if (strict && op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
+ {
+ throw new DirectoryException(op.getResultCode(), ERR_CANNOT_RECOVER_CHANGES.get(baseDN));
+ }
// An error happened trying to search for the updates
// Log an error
logger.error(ERR_CANNOT_RECOVER_CHANGES, baseDN);
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
index fc2fb21..ee63266 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
@@ -13,7 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyright 2023-2026 3A Systems LLC.
+ * Portions Copyright 2023-2026 3A Systems, LLC.
* Portions Copyright 2025 Wren Security.
*/
package org.opends.server.replication.service;
@@ -236,6 +236,8 @@
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
private volatile boolean shutdown;
+ /** A backend whose state could not be restored must never reconnect through this broker. */
+ private volatile boolean permanentlyStopped;
/**
* The stop in progress: set by {@link #stop()} before anything else it does, and cleared to
* {@code null} by {@link #start()}. While it is set, a publish which finds no window credit by
@@ -366,7 +368,7 @@
{
synchronized (startStopLock)
{
- if (!shutdown)
+ if (!shutdown || permanentlyStopped)
{
return;
}
@@ -2300,7 +2302,7 @@
// Synchronize inside the loop in order to allow shutdown.
synchronized (startStopLock)
{
- if (rs.isConnected() || shutdown)
+ if (rs.isConnected() || shutdown || permanentlyStopped)
{
break;
}
@@ -2416,7 +2418,7 @@
{
boolean done = false;
- while (!done && !shutdown && !gaveUpForTheStop())
+ while (!done && !shutdown && !permanentlyStopped && !gaveUpForTheStop())
{
if (connectionError)
{
@@ -2508,7 +2510,7 @@
Check the session. If it has changed, some disconnection or
reconnection happened and we need to restart from scratch.
*/
- if (gaveUpForTheStop())
+ if (permanentlyStopped || gaveUpForTheStop())
{
/*
* The stop gave up since the loop head, while this publish waited for this lock - for
@@ -2559,7 +2561,7 @@
Not once the stop gave up, which it may have done while this waited for the lock.
*/
Session session = connectedRS.get().session;
- if (session != null && !gaveUpForTheStop())
+ if (session != null && !permanentlyStopped && !gaveUpForTheStop())
{
session.publish(new WindowProbeMsg());
}
@@ -2659,7 +2661,7 @@
boolean reconnectOnFailure, boolean returnOnTopoChange)
throws SocketTimeoutException
{
- while (!shutdown)
+ while (!shutdown && !permanentlyStopped)
{
ConnectedRS rs = connectedRS.get();
if (!rs.isConnected())
@@ -2896,9 +2898,48 @@
}
}
+ /**
+ * Stops a domain with untrusted data state. No offline CSN is announced and buffered messages
+ * are discarded rather than drained to the peer. A later configuration change cannot restart it.
+ */
+ void stopPermanently()
+ {
+ permanentlyStopped = true;
+ final Session session = connectedRS.get().session;
+ if (session != null)
+ {
+ session.abort(new IOException("The total update could not restore its backend state"));
+ }
+ stop();
+ }
+
+ /** Returns whether this broker has been permanently stopped following an import failure. */
+ boolean isPermanentlyStopped()
+ {
+ return permanentlyStopped;
+ }
+
/** Stop the server. */
public void stop()
{
+ stop(true);
+ }
+
+ /**
+ * Stops an aborted import on its listener without waiting for credit that only that listener
+ * can receive. The next enable may start a new session; no offline CSN is announced.
+ * An ECL-enabled replica therefore leaves its last known CSN in the replication server's
+ * consistency point, as on an abrupt disconnect: new change-number assignment can stall across
+ * its ECL-enabled domains until this replica reconnects and advances that CSN. Having send credit
+ * alone would not make an offline announcement safe here: its locks or socket write can still block.
+ */
+ void stopAfterAbortedImport()
+ {
+ stop(false);
+ }
+
+ private void stop(boolean announceOffline)
+ {
if (logger.isTraceEnabled() && !shutdown)
{
debugInfo("is stopping and will close the connection to RS(" + getRsServerId() + ")");
@@ -2921,13 +2962,33 @@
try
{
// In the try: a watchdog which cannot be started must not leave the broker stopping either.
- if (newStop && stop.bounded)
+ if (newStop && stop.bounded && announceOffline)
{
new DirectoryThread(() -> endWhatStillHoldsTheStop(stop),
"Replica DS(" + getServerId() + ") watchdog of the stop for domain \"" + getBaseDN() + "\"")
.start();
}
- domain.publishReplicaOfflineMsg();
+ if (!announceOffline)
+ {
+ synchronized (stop)
+ {
+ if (!stop.ended)
+ {
+ // End blocked writes too: closing in order could wait for the peer to read forever.
+ // stop.end() takes this monitor, so this close cannot reach a later session.
+ stop.gaveUp = true;
+ final Session session = connectedRS.get().session;
+ if (session != null)
+ {
+ session.abort(new IOException("The total update preparation left its backend unavailable"));
+ }
+ }
+ }
+ }
+ else if (!permanentlyStopped)
+ {
+ domain.publishReplicaOfflineMsg();
+ }
}
finally
{
@@ -3674,7 +3735,7 @@
*/
boolean shuttingDown()
{
- return shutdown;
+ return shutdown || permanentlyStopped;
}
/**
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 f72a4fe..f5fef33 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
@@ -1982,12 +1982,14 @@
*/
replicasWeAreWaitingFor.addAll(getReplicaInfos().keySet());
+ final Map<Integer, Long> disconnectedSince = new HashMap<>();
+ // An absence bound, not a bound on every possible reconnect: the importer's connection
+ // timeout and RS list are not carried in the topology and may differ from this server's.
+ final long reconnectGraceNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(10);
boolean done;
do
{
done = true;
- int reconnectMaxDelayInSec = 10;
- int reconnectWait = 0;
Iterator<Integer> it = replicasWeAreWaitingFor.iterator();
while (it.hasNext())
{
@@ -2006,21 +2008,32 @@
may be for a long time if it crashed or had been stopped
may be just the time to reconnect after import : should be short
*/
- if (++reconnectWait<reconnectMaxDelayInSec)
+ // Each absent target gets its own grace, across passes. A target stopped after a failed
+ // import may never reconnect; it must not retain the exporter's import/export context.
+ final long now = System.nanoTime();
+ final long since = disconnectedSince.computeIfAbsent(serverId, id -> now);
+ if (now - since < reconnectGraceNanos)
{
// let's still wait to give a chance to this server to reconnect
done = false;
}
- // Else we left enough time to the servers to reconnect
+ else
+ {
+ ieCtx.failureList.add(serverId);
+ it.remove();
+ disconnectedSince.remove(serverId);
+ }
}
else
{
+ disconnectedSince.remove(serverId);
// this server is connected
if (dsInfo.getStatus() == ServerStatus.FULL_UPDATE_STATUS)
{
// this one is still doing the Full Update ... retry later
done = false;
- break;
+ // Still visit the other targets, so their absence is timed while this one imports.
+ continue;
}
if (dsInfo.getGenerationId() == getGenerationID())
@@ -2722,9 +2735,12 @@
right here, right now
we never want retryOnFailure if we fails reconnecting in the restart.
*/
- broker.reStart(false);
+ if (!isServicePermanentlyStopped())
+ {
+ broker.reStart(false);
+ }
- if (ieCtx.getException() != null
+ if (!isServicePermanentlyStopped() && ieCtx.getException() != null
&& broker.isConnected()
&& initFromTask != null
&& ++ieCtx.attemptCnt < 2)
@@ -2785,7 +2801,7 @@
try
{
- if (broker.isConnected() && ieCtx.getException() != null)
+ if (!isServicePermanentlyStopped() && broker.isConnected() && ieCtx.getException() != null)
{
// Let's notify the exporter
ErrorMsg errorMsg = new ErrorMsg(requesterServerId,
@@ -3410,6 +3426,10 @@
*/
public void startListenService()
{
+ if (isServicePermanentlyStopped())
+ {
+ return;
+ }
synchronized (sessionLock)
{
if (listenerThread != null)
@@ -3597,6 +3617,33 @@
return sessionGeneration;
}
+ /** Stops the broker without joining the listener which may be running this import. */
+ protected final void stopServicePermanently()
+ {
+ broker.stopPermanently();
+ }
+
+ /**
+ * Stops an import's session reversibly when preparation left its backend unavailable.
+ * Called by the listener itself: joining it, or taking locks held by a thread joining it,
+ * would deadlock. A later enable joins this listener before starting another session.
+ */
+ protected final void stopServiceAfterAbortedImport()
+ {
+ final DirectoryThread listener = listenerThread;
+ if (listener != null)
+ {
+ listener.initiateShutdown();
+ }
+ broker.stopAfterAbortedImport();
+ }
+
+ /** Returns whether a failed import has permanently stopped this service. */
+ protected final boolean isServicePermanentlyStopped()
+ {
+ return broker != null && broker.isPermanentlyStopped();
+ }
+
/**
* Restart the Replication service after a {@link #disableService()}.
* <p>
@@ -3613,8 +3660,18 @@
{
synchronized (serviceStateLock)
{
+ if (isServicePermanentlyStopped())
+ {
+ return;
+ }
synchronized (sessionLock)
{
+ if (listenerThread != null && listenerThread.isShutdownInitiated())
+ {
+ // An aborted import can stop its own session without joining its listener. Join it
+ // while the broker is still stopped, so it cannot consume the new session's messages.
+ disableService();
+ }
broker.start();
startListenService();
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java b/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java
index c9867cb..d2a4eab 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java
@@ -427,6 +427,8 @@
LocalBackendWorkflowElement.checkIfBackendIsWritable(currentBackend,
this, entryDN, ERR_MODDN_SERVER_READONLY, ERR_MODDN_BACKEND_READONLY);
+ LocalBackendWorkflowElement.checkIfBackendIsWritable(currentBackend,
+ this, newDN, ERR_MODDN_SERVER_READONLY, ERR_MODDN_BACKEND_READONLY);
if (noOp)
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java b/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java
index bde925e..d612322 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java
@@ -723,6 +723,12 @@
LocalizableMessageDescriptor.Arg1<Object> backendMsg)
throws DirectoryException
{
+ final LocalizableMessage failure = LocalBackend.getWriteFailure(entryDN);
+ if (failure != null)
+ {
+ // This also applies to internal/repair operations and private backends such as cn=schema.
+ throw new DirectoryException(ResultCode.UNAVAILABLE, failure);
+ }
if (!backend.isPrivateBackend())
{
checkIfWritable(DirectoryServer.getCoreConfigManager().getWritabilityMode(), op, serverMsg, entryDN);
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 a5244ee..0e2873a 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -730,3 +730,25 @@
ERR_UNPUBLISHED_DELETE_DROPPED_334=Domain "%s" dropped the delete of "%s" (%s), which no \
replication server has taken, from the %d deletes it keeps aside for its recoveries: no recovery \
publishes it anymore, and the replicas which miss it keep the entry
+WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION_335=Domain "%s" keeps no message of its \
+ local changes after %s up to %s, which its replication server misses: it publishes what it can \
+ rebuild from historical information and the deletes kept aside. The historical information holds \
+ nothing of a change whose values a later one replaced, nor of the add of an entry deleted since, \
+ so that replication server may never hold those changes
+
+ERR_INIT_DOMAIN_STOPPED_336=Replication and writes for domain "%s" are stopped because the total \
+ update could not restore a trustworthy backend state. Earlier import error: %s. Cleanup error: %s. \
+ Correct the backend failure and restart the server process (an in-core restart, such as a restart \
+ task or stop-ds --restart with connection options, does not clear this); a partial import may \
+ require another full initialization. Enabling or recreating the replication domain does not clear this failure
+ERR_INIT_OPERATIONS_NOT_DRAINED_337=The total update of domain "%s" was abandoned before replacing \
+ its data because replay or a server-state write did not finish. Retry after the outstanding operation completes
+ERR_INIT_DOMAIN_AWAITING_BACKEND_338=Replication for domain "%s" is stopped because backend "%s" \
+ could not be enabled after preparing a total update. The original data was not replaced. \
+ Replication resumes when the backend is available again. Correct the backend failure and disable \
+ then enable the backend, or restart the server process
+ERR_INIT_PREPARATION_INTERRUPTED_339=The total update of domain "%s" was abandoned before replacing \
+ its data because the wait for replay or a server-state write was interrupted
+ERR_INIT_DOMAIN_RESTORE_FAILED_340=Automatic restoration of replication for domain "%s" after \
+ an aborted total update failed for backend "%s": %s. Correct the failure and disable \
+ then enable the backend, or restart the server process
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java
index 95b2df2..a89fe96 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java
@@ -74,6 +74,7 @@
import static java.util.concurrent.TimeUnit.*;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.opends.messages.ReplicationMessages.WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.replication.server.changelog.api.DBCursor.KeyMatchingStrategy.*;
import static org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy.*;
@@ -146,6 +147,23 @@
}
/**
+ * Runs a test as the domain keeps the messages of its last local changes, which a recovery publishes,
+ * and as it keeps none, and the recovery rebuilds every change from the historical information.
+ */
+ @DataProvider
+ public Object[][] keepingTheRecentChanges()
+ {
+ return new Object[][] { { true }, { false } };
+ }
+
+ /** Runs a test with each of {@link #keepingTheLocalChanges()} and {@link #keepingTheRecentChanges()}. */
+ @DataProvider
+ public Object[][] keepingTheLocalAndTheRecentChanges()
+ {
+ return new Object[][] { { true, true }, { true, false }, { false, true }, { false, false } };
+ }
+
+ /**
* Forgets the replication server of the test: the next one, and the domains of the next tests,
* would be configured with its port, which nothing listens on anymore.
*/
@@ -209,6 +227,8 @@
ReplicationServer rs = createReplicationServer();
// Create Replication Server and Domain
LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ // These rebuild the changes from the historical information: the domain keeps no message of them.
+ pendingChangesOf(rd1).setMaxRecentChanges(0);
try
{
@@ -308,6 +328,8 @@
ReplicationServer rs = createReplicationServer();
// Create Replication Server and Domain
LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ // These rebuild the changes from the historical information: the domain keeps no message of them.
+ pendingChangesOf(rd1).setMaxRecentChanges(0);
long startTime = TimeThread.getTime();
try
@@ -774,10 +796,12 @@
* them, and the deletes kept aside for the recovery were dropped as it ended: the next recovery
* has to find all of them still. The first change refused is a delete, which only the list it is
* kept aside in, or the local changes the recovery keeps, can publish again: a modify, the search
- * of the next recovery finds again.
+ * of the next recovery finds again - or the messages of the last local changes, if the domain keeps
+ * them.
*/
- @Test(dataProvider = "keepingTheLocalChanges")
- public void changesTheBrokerRefusedAreLeftToTheNextRecovery(boolean keeps) throws Exception
+ @Test(dataProvider = "keepingTheLocalAndTheRecentChanges")
+ public void changesTheBrokerRefusedAreLeftToTheNextRecovery(boolean keeps, boolean keepsTheRecentChanges)
+ throws Exception
{
final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
@@ -789,7 +813,7 @@
final DN dnB = addPerson("b");
final CSN rsMaxCSN = rd1.getLastLocalChange();
- startRecovery(rd1, keeps);
+ startRecovery(rd1, keeps, keepsTheRecentChanges);
final List<CSN> done = new ArrayList<>();
TestCaseUtils.deleteEntry(dnB);
done.add(rd1.getLastLocalChange());
@@ -837,6 +861,7 @@
// as the missing change publisher does once it failed
brokerOf(rd1).setRecoveryRequired(false);
+ final int logged = TestCaseUtils.ERROR_TEXT_WRITER.getMessages().size();
rd1.disable();
rd1.enable();
// The broker starts its session before enable() returns: a change done before would have
@@ -845,6 +870,13 @@
replace(dnA, "description", "in a session which needs no recovery");
assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+ // The publisher which ended the recovery had no change to rebuild, and says none (issue #1246).
+ final List<String> records = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
+ assertThat(records.subList(logged, records.size()))
+ .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
+ + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
+ .as("the publisher which ended the recovery says it rebuilds changes")
+ .isEmpty();
}
finally
{
@@ -1287,6 +1319,193 @@
}
/**
+ * The changes done before a recovery began are published as they were done, from the messages the
+ * domain keeps of its last local changes: a change whose values a later one replaced, and the add of
+ * an entry deleted since, of which the historical information holds nothing. A recovery which
+ * rebuilt them from it published neither, and the replication server it was for took neither once it
+ * held the later changes: a directory server which failed over to a replication server behind it
+ * left gaps in its changelog for good (issue #1246).
+ */
+ @Test(dataProvider = "keepingTheLocalChanges")
+ public void theChangesDoneBeforeTheRecoveryBeganArePublishedAsTheyWereDone(boolean keeps) throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ final DN dnA = addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+ final DN dnB = addPerson("b");
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(dnB);
+ done.add(rd1.getLastLocalChange());
+ startRecovery(rd1, keeps, true);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A directory server which fails over to a replication server behind it has it hold every change it
+ * misses, as it was done: a change whose values a later one replaced, and the add of an entry deleted
+ * since, included. Rebuilt from the historical information, neither reached it, and it took neither
+ * afterwards, as it held later changes of this replica - the gaps the replication convergence run
+ * found in the changelogs (issue #1246).
+ */
+ @Test
+ public void aReplicationServerBehindGetsEveryChangeItMissesAsItWasDone() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ final SortedSet<String> towardsRs = new TreeSet<>(replServers);
+ final int secondPort = TestCaseUtils.findFreePort();
+ final SortedSet<String> towardsSecond = towards(secondPort);
+ ReplicationServer second = null;
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ second = createIsolatedReplicationServer(secondPort, "HistoricalCsnOrderingSecond", SECOND_RS_ID);
+ final DN dnA = addPerson("a");
+ aChangeBothReplicationServersHold(rd1, dnA, rs, towardsRs, second, towardsSecond);
+
+ // Done on the first replication server only.
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+ final DN dnB = addPerson("b");
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(dnB);
+ done.add(rd1.getLastLocalChange());
+ for (CSN csn : done)
+ {
+ assertInTheChangelog(rs, baseDN, csn);
+ }
+
+ connectTo(rd1, towardsSecond, SECOND_RS_ID);
+ for (CSN csn : done)
+ {
+ assertInTheChangelog(second, baseDN, csn);
+ }
+ waitUntilTheRecoveryIsOver(rd1);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, second);
+ }
+ }
+
+ /**
+ * A replication server further behind than the messages the domain kept of its last local changes go
+ * back gets the changes before the first of them rebuilt from the historical information, and the
+ * others as they were done - a change a later one replaced included - and the domain says so, once
+ * however many rounds rebuild them: that replication server may never hold some of the changes
+ * rebuilt. The recovery starts three search windows back, so that rounds of their own rebuild the
+ * changes, each from where the one before stopped.
+ */
+ @Test
+ public void aRecoveryFromBeforeTheMessagesKeptRebuildsTheChangesBeforeThemAndSaysSo() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ final List<CSN> done = new ArrayList<>();
+ final DN dnA = addPerson("a");
+ done.add(rd1.getLastLocalChange());
+ final DN dnB = addPerson("b");
+ done.add(rd1.getLastLocalChange());
+ final CSN fromCSN = new CSN(rd1.getLastLocalChange().getTime() - 3 * SEARCH_WINDOW_MS, 0, serverId);
+
+ pendingChangesOf(rd1).setMaxRecentChanges(2);
+ // Put the last rebuilt change in a different search window from the adds.
+ rd1.getGenerator().adjust(new CSN(rd1.getLastLocalChange().getTime() + SEARCH_WINDOW_MS, 0, serverId));
+ replace(dnB, "description", "rebuilt");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+ startRecovery(rd1, true, true);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ // A publisher of its own, which has each record once: the error log of the tests has two, whose
+ // copies of one record differ when the second of their time stamps ticks in between.
+ final List<String> records = errorLogRecordsOf(() -> {
+ assertTrue(rd1.buildAndPublishMissingChanges(fromCSN, new TestBroker(published)));
+ return null;
+ });
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ assertThat(records)
+ .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
+ + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
+ .as("the domain says it rebuilds changes from the historical information, once")
+ .hasSize(1);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A change published assured is published by a recovery as one no acknowledgement is waited for, as
+ * a recovery publishes the changes it rebuilds and the ones it kept: a copy, not the message the domain
+ * kept, which it handed to the broker already and whose acknowledgement the operation may still wait for.
+ */
+ @Test
+ public void aChangePublishedAssuredIsPublishedByARecoveryAsOneNoAcknowledgementIsWaitedFor() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = MultimasterReplication.createNewDomain(new DomainFakeCfg(
+ baseDN, serverId, replServers, AssuredType.SAFE_DATA, 1, 1, 1000, null));
+ rd1.start();
+ try
+ {
+ final DN dnA = addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ replace(dnA, "description", "assured");
+ final CSN done = rd1.getLastLocalChange();
+ startRecovery(rd1, true, true);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactly(done);
+ assertFalse(((UpdateMsg) published.get(0)).isAssured(), "the recovery publishes it not assured");
+ assertTrue(pendingChangesOf(rd1).recentChangesAfter(rsMaxCSN, done).messages.get(0).isAssured(),
+ "the message kept is left as it was published");
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
* The local changes are held back at the end of a recovery for as long as publishing the ones done
* during the round before takes, and not for a search: the recovery keeps the changes done since it
* began, and needs none for them. One which searched for them did so while every local write
@@ -1345,10 +1564,11 @@
/**
* A recovery which kept too many of the local changes done while it runs keeps none, and searches
* for them in the historical information instead, from where it has got to: none is lost, and the
- * recovery ends.
+ * recovery ends. With the messages of the last local changes kept, it publishes them from those.
*/
- @Test
- public void aRecoveryWhichKeptTooManyChangesSearchesForThemAndLosesNone() throws Exception
+ @Test(dataProvider = "keepingTheRecentChanges")
+ public void aRecoveryWhichKeptTooManyChangesSearchesForThemAndLosesNone(boolean keepsTheRecentChanges)
+ throws Exception
{
final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
@@ -1362,7 +1582,7 @@
final CSN rsMaxCSN = rd1.getLastLocalChange();
pendingChangesOf(rd1).setMaxKeptChanges(2);
- startRecovery(rd1);
+ startRecovery(rd1, true, keepsTheRecentChanges);
// Each change of other values than the ones before, which the historical information keeps.
final List<CSN> done = new ArrayList<>();
replace(dnA, "description", "kept");
@@ -1388,10 +1608,13 @@
/**
* Local changes go on while a recovery which keeps them runs. The recovery publishes the ones it
* kept, holds the local changes back only to publish the ones pushed while it published the others,
- * and ends; each change is published once, by the recovery or, once it has ended, by the broker.
+ * and ends; each change is published once, by the recovery or, once it has ended, by the broker -
+ * whether the changes done before it began are published from the messages the domain kept of them
+ * or rebuilt from the historical information.
*/
- @Test
- public void aRecoveryWhichKeepsTheLocalChangesEndsWhileTheyGoOnAndLosesNone() throws Exception
+ @Test(dataProvider = "keepingTheRecentChanges")
+ public void aRecoveryWhichKeepsTheLocalChangesEndsWhileTheyGoOnAndLosesNone(boolean keepsTheRecentChanges)
+ throws Exception
{
final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
@@ -1407,7 +1630,7 @@
}
final CSN rsMaxCSN = rd1.getLastLocalChange();
- startRecovery(rd1);
+ startRecovery(rd1, true, keepsTheRecentChanges);
final List<CSN> done = new CopyOnWriteArrayList<>();
final Future<?> changes = writer.submit(unchecked(() ->
{
@@ -1756,10 +1979,11 @@
/**
* The local changes a recovery kept are not published once the data they were done on has been
* replaced, as a restore or an import does while the domain is disabled: the replicas would get
- * changes this one does not hold. The recovery searches the data it finds then instead.
+ * changes this one does not hold. The recovery searches the data it finds then instead. The messages
+ * the domain kept of its last local changes are forgotten with them (issue #1246).
*/
- @Test
- public void theLocalChangesARecoveryKeptGoWithTheDataTheyWereDoneOn() throws Exception
+ @Test(dataProvider = "keepingTheRecentChanges")
+ public void theLocalChangesARecoveryKeptGoWithTheDataTheyWereDoneOn(boolean keepsTheRecentChanges) throws Exception
{
final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
@@ -1776,7 +2000,7 @@
// The change done then is kept. The broker refuses the announcement that the replica goes
// offline as well, which would have the replication server hold a CSN past the change.
- startRecovery(rd1);
+ startRecovery(rd1, true, keepsTheRecentChanges);
replace(dnA, "description", "on the data which is replaced");
// The data is restored while the domain is disabled, as it was before the change.
@@ -1793,6 +2017,159 @@
assertInTheChangelog(rs, baseDN, afterTheRestore);
assertThat(changelogAfter(rs, baseDN, rsMaxCSN)).as("the changes published once the data was restored")
.containsExactly(afterTheRestore);
+
+ // A recovery from where the replication server stood before the change - the one of a session to
+ // a replication server behind, say - publishes what the data holds, and not the change either.
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).as("the changes a later recovery publishes").containsExactly(afterTheRestore);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A recovery which has no change to rebuild - its replication server holds the last local change - says
+ * none is rebuilt from the historical information: the publisher which ends the recovery a failed one
+ * left runs so, and the domain kept no message for it to publish (issue #1246).
+ */
+ @Test
+ public void aRecoveryWithNoChangeToRebuildSaysNoneIsRebuilt() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ startRecovery(rd1, false);
+
+ final int logged = TestCaseUtils.ERROR_TEXT_WRITER.getMessages().size();
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(published).isEmpty();
+ final List<String> records = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
+ assertThat(records.subList(logged, records.size()))
+ .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
+ + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
+ .as("the recovery says it rebuilds changes")
+ .isEmpty();
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ @Test
+ public void aRecoveryWarnsWhenOnlyTheDeleteOfAnEntryCanBePublished() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ addPerson("before");
+ final CSN from = rd1.getLastLocalChange();
+ final DN deleted = addPerson("deleted");
+ TestCaseUtils.deleteEntry(deleted);
+ final CSN delete = rd1.getLastLocalChange();
+ startRecovery(rd1, false);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ final List<String> records = errorLogRecordsOf(() -> {
+ assertTrue(rd1.buildAndPublishMissingChanges(from, new TestBroker(published)));
+ return null;
+ });
+ assertThat(csns(published)).containsExactly(delete);
+ assertThat(records).filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
+ + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " ")).hasSize(1);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ @Test
+ public void messagesAreForgottenEvenWhenTheStateCannotBeRead() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ addPerson("a");
+ final CSN change = rd1.getLastLocalChange();
+ final CSN before = new CSN(change.getTime(), change.getSeqnum() - 1, serverId);
+ assertThat(pendingChangesOf(rd1).recentChangesAfter(before, change).messages).hasSize(1);
+ rd1.disable();
+
+ final MemoryBackend backend =
+ (MemoryBackend) getServerContext().getBackendConfigManager().getLocalBackendById(TEST_BACKEND_ID);
+ final AttributeType stateType = getServerContext().getSchema().getAttributeType("ds-sync-state");
+ final Entry saved = backend.getEntry(baseDN);
+ final Entry invalid = saved.duplicate(false);
+ final AttributeBuilder state = new AttributeBuilder(stateType);
+ state.add("not-a-csn");
+ invalid.putAttribute(stateType, newArrayList(state.toAttribute()));
+ backend.replaceEntry(saved, invalid, null);
+
+ rd1.enable();
+
+ assertNull(pendingChangesOf(rd1).recentChangesAfter(before, change).messages);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A recovery from the last change done before a restart, whose replication server misses the changes
+ * done since, says none is rebuilt from the historical information: the messages kept begin right before
+ * the first change done since, later than the last one done before, and the window in between holds no
+ * change of this replica to rebuild (issue #1246).
+ */
+ @Test
+ public void aRecoveryFromTheChangeBeforeTheMessagesKeptSaysNoneIsRebuilt() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ final DN dnA = addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ // As after a restart: nothing kept of the changes before, and the time of the generator moved on.
+ pendingChangesOf(rd1).forgetRecentChanges();
+ rd1.getGenerator().adjust(new CSN(rsMaxCSN.getTime() + 1000, 0, serverId));
+ replace(dnA, "description", "since");
+ final CSN since = rd1.getLastLocalChange();
+ startRecovery(rd1, true, true);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ final List<String> records = errorLogRecordsOf(() -> {
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ return null;
+ });
+ assertThat(csns(published)).containsExactly(since);
+ assertThat(records)
+ .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
+ + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
+ .as("the recovery says it rebuilds changes")
+ .isEmpty();
}
finally
{
@@ -1986,10 +2363,11 @@
/**
* A recovery whose kept changes do not reach the last local change - one which should have been
* kept is not there - searches for the changes after them, and ends. It went round for ever while
- * it held the local changes back, every local write waiting for it.
+ * it held the local changes back, every local write waiting for it. With the messages of the last
+ * local changes kept, the change the recovery missed is published from its message.
*/
- @Test
- public void aRecoveryWhoseKeptChangesMissOneSearchesForIt() throws Exception
+ @Test(dataProvider = "keepingTheRecentChanges")
+ public void aRecoveryWhoseKeptChangesMissOneSearchesForIt(boolean keepsTheRecentChanges) throws Exception
{
final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
@@ -2001,7 +2379,7 @@
final DN dnB = addPerson("b");
final CSN rsMaxCSN = rd1.getLastLocalChange();
- startRecovery(rd1);
+ startRecovery(rd1, true, keepsTheRecentChanges);
final List<CSN> done = new ArrayList<>();
replace(dnA, "description", "kept");
done.add(rd1.getLastLocalChange());
@@ -2423,9 +2801,13 @@
* deletes. They are in the data all the same - their entries are gone - and a recovery publishes
* them, where taking them for changes the data does not reflect dropped them, and the replicas kept
* the suffix.
+ * <p>
+ * With the messages of the last local changes kept, the messages of the deletes are forgotten as the
+ * state is read back, and the add of the base entry is the first one kept from there: the deletes,
+ * which come after the change the state read back holds, are not taken for kept with it (issue #1246).
*/
- @Test
- public void theDeletesOfASuffixOutliveAStateWhichCouldNotBeSaved() throws Exception
+ @Test(dataProvider = "keepingTheRecentChanges")
+ public void theDeletesOfASuffixOutliveAStateWhichCouldNotBeSaved(boolean keepsTheRecentChanges) throws Exception
{
final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
TestCaseUtils.initializeTestBackend(true);
@@ -2438,7 +2820,7 @@
assertInTheChangelog(rs, baseDN, rsMaxCSN);
// The replication server misses the deletes.
- refuseTheLocalChanges(rd1);
+ refuseTheLocalChanges(rd1, keepsTheRecentChanges);
final List<CSN> done = new ArrayList<>();
TestCaseUtils.deleteEntry(dnA);
done.add(rd1.getLastLocalChange());
@@ -2453,7 +2835,7 @@
TestCaseUtils.addEntry(StaticUtils.createEntry(baseDN));
done.add(rd1.getLastLocalChange());
// Started after the changes, which the recovery then searches for.
- startRecovery(rd1);
+ startRecovery(rd1, true, keepsTheRecentChanges);
final List<ReplicationMsg> published = new ArrayList<>();
assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
assertThat(csns(published)).containsExactlyElementsOf(done);
@@ -2782,7 +3164,7 @@
* {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker)} in its
* place. The pending changes leave the local changes to the recovery, which keeps them, and the
* broker refuses them. The changes done before are the ones the recovery rebuilds from the
- * historical information.
+ * historical information: the domain forgets the messages it kept of them.
*/
private static void startRecovery(LDAPReplicationDomain domain) throws Exception
{
@@ -2797,10 +3179,27 @@
*/
private static void startRecovery(LDAPReplicationDomain domain, boolean keepsTheLocalChanges) throws Exception
{
+ startRecovery(domain, keepsTheLocalChanges, false);
+ }
+
+ /**
+ * Puts the domain where a session start leaves it when its replication server misses some of its
+ * changes, as {@link #startRecovery(LDAPReplicationDomain, boolean)} does, with the messages the
+ * domain kept of the changes done before, which the recovery publishes as they were done (issue
+ * #1246), or - as the tests of the search of a recovery have it - none, and the recovery rebuilds them
+ * from the historical information.
+ */
+ private static void startRecovery(LDAPReplicationDomain domain, boolean keepsTheLocalChanges,
+ boolean keepsTheRecentChanges) throws Exception
+ {
if (!keepsTheLocalChanges)
{
pendingChangesOf(domain).setMaxKeptChanges(0);
}
+ if (!keepsTheRecentChanges)
+ {
+ pendingChangesOf(domain).setMaxRecentChanges(0);
+ }
pendingChangesOf(domain).setRecovering(true);
brokerOf(domain).setRecoveryRequired(true);
}
@@ -2809,7 +3208,7 @@
* Has the broker refuse the local changes, as it does from the start of a session whose replication
* server misses some, up to the pending changes leaving them to the recovery: they are recorded in
* the domain state only, and the replication server misses them. The recovery the next session
- * starts rebuilds them from the historical information.
+ * starts rebuilds them from the historical information: the domain keeps no message of them.
* <p>
* Once the missing change publisher of an earlier session has left, under the lock it leaves under:
* it has the broker take the local changes again as it leaves, after the end of its recovery - which
@@ -2818,6 +3217,21 @@
*/
private static void refuseTheLocalChanges(final LDAPReplicationDomain domain) throws Exception
{
+ refuseTheLocalChanges(domain, false);
+ }
+
+ /**
+ * Has the broker refuse the local changes, as {@link #refuseTheLocalChanges(LDAPReplicationDomain)}
+ * does, the domain keeping the messages of the changes it refuses, which the recovery the next session
+ * starts publishes as they were done (issue #1246), or none.
+ */
+ private static void refuseTheLocalChanges(final LDAPReplicationDomain domain, boolean keepsTheRecentChanges)
+ throws Exception
+ {
+ if (!keepsTheRecentChanges)
+ {
+ pendingChangesOf(domain).setMaxRecentChanges(0);
+ }
final Field recoveryLockField = LDAPReplicationDomain.class.getDeclaredField("recoveryLock");
recoveryLockField.setAccessible(true);
final Object recoveryLock = recoveryLockField.get(domain);
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
index 1842e63..c2bca5c 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
@@ -12,6 +12,7 @@
* information: "Portions copyright [year] [name of copyright owner]".
*
* Copyright 2026 3A Systems, LLC.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -23,11 +24,14 @@
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;
+import java.util.Arrays;
+
import org.forgerock.opendj.ldap.DN;
import org.mockito.ArgumentCaptor;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.UpdateMsg;
@@ -241,6 +245,252 @@
}
/**
+ * The domain keeps the messages of its last local changes, in order, so that a recovery publishes
+ * them as they were done rather than rebuild them from the historical information, which holds
+ * nothing of a change a later one replaced (issue #1246).
+ */
+ @Test
+ public void theMessagesOfTheLastLocalChangesAreKeptInOrder() throws Exception
+ {
+ final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
+ final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
+ final LDAPUpdateMsg second = pushLocalChange(pendingChanges);
+ final LDAPUpdateMsg third = pushLocalChange(pendingChanges);
+
+ assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), third.getCSN()).messages,
+ Arrays.asList(second, third));
+ assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), second.getCSN()).messages,
+ Arrays.asList(second));
+ }
+
+ /**
+ * The messages kept go back to the CSN right before the first one: a recovery from there - from the
+ * change before it, done in the same millisecond - gets the first change as it was done, rather than
+ * rebuilt from the historical information.
+ */
+ @Test
+ public void theMessagesKeptGoBackToRightBeforeTheFirstOne() throws Exception
+ {
+ final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
+ final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
+
+ assertEquals(pendingChanges.recentChangesAfter(rightBefore(first), first.getCSN()).messages,
+ Arrays.asList(first));
+ }
+
+ /**
+ * The messages kept of a first change numbered 0 go back to the greatest sequence number of the
+ * millisecond before, which no CSN of the server comes after before that change: a recovery from there
+ * gets the change as it was done. A CSN which went back to the same millisecond would come after the
+ * change, and the recovery would rebuild it from the historical information instead.
+ */
+ @Test
+ public void theMessagesKeptOfAChangeNumberedZeroGoBackToTheMillisecondBefore() throws Exception
+ {
+ final ServerState state = new ServerState();
+ // The generator goes on from the greatest sequence number, so its first change wraps to 0.
+ state.update(new CSN(0, Integer.MAX_VALUE, SERVER_ID));
+ final PendingChanges pendingChanges = new PendingChanges(new CSNGenerator(SERVER_ID, state),
+ domainWhichPublishes(true), new ShutdownSyncAnnouncer(new DSRSShutdownSync(), baseDN));
+ final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
+ assertEquals(first.getCSN().getSeqnum(), 0);
+
+ final CSN justBefore = new CSN(first.getCSN().getTime() - 1, Integer.MAX_VALUE, SERVER_ID);
+ assertEquals(pendingChanges.recentChangesAfter(new CSN(0, 0, SERVER_ID), first.getCSN()).keptAfter,
+ justBefore);
+ assertEquals(pendingChanges.recentChangesAfter(justBefore, first.getCSN()).messages,
+ Arrays.asList(first));
+ }
+
+ /**
+ * Past as many changes as it keeps at most, the domain drops the oldest messages, and tells where
+ * the ones it kept begin: a recovery rebuilds the changes before that from the historical
+ * information.
+ */
+ @Test
+ public void pastTheMostChangesKeptTheOldestAreDropped() throws Exception
+ {
+ final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
+ pendingChanges.setMaxRecentChanges(2);
+ final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
+ final LDAPUpdateMsg second = pushLocalChange(pendingChanges);
+ final LDAPUpdateMsg third = pushLocalChange(pendingChanges);
+
+ assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), third.getCSN()).messages,
+ Arrays.asList(second, third));
+ final PendingChanges.RecentChanges fromBefore =
+ pendingChanges.recentChangesAfter(new CSN(0, 0, SERVER_ID), third.getCSN());
+ assertNull(fromBefore.messages, "the messages kept do not go back that far");
+ assertEquals(fromBefore.keptAfter, first.getCSN());
+ }
+
+ /** Past as much of the heap as their messages take at most, the domain drops the oldest messages. */
+ @Test
+ public void pastTheMostBytesKeptTheOldestAreDropped() throws Exception
+ {
+ assertTrue(PendingChanges.MAX_RECENT_BYTES <= Runtime.getRuntime().maxMemory() / 16,
+ "a domain must not reserve more than a sixteenth of the heap for recent messages");
+ final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
+ final int half = sizeTaking(PendingChanges.MAX_RECENT_BYTES / 2);
+ final LDAPUpdateMsg first = pushLocalChange(pendingChanges, half);
+ final LDAPUpdateMsg second = pushLocalChange(pendingChanges, half);
+ assertEquals(pendingChanges.recentChangesAfter(rightBefore(first), second.getCSN()).messages,
+ Arrays.asList(first, second));
+
+ final LDAPUpdateMsg third = pushLocalChange(pendingChanges, 0);
+
+ assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), third.getCSN()).messages,
+ Arrays.asList(second, third));
+ assertEquals(pendingChanges.recentChangesAfter(rightBefore(first), third.getCSN()).keptAfter, first.getCSN());
+ }
+
+ /**
+ * A change pushed again - the publish of the push before threw, and left it pending - is counted
+ * once: counted twice, it took more of the heap than it does, and was dropped though it was the only
+ * message kept.
+ */
+ @Test
+ public void aChangePushedAgainIsCountedOnce() throws Exception
+ {
+ final ReplicationDomain domain = mock(ReplicationDomain.class);
+ when(domain.publish(any(UpdateMsg.class))).thenThrow(new IllegalStateException("the publish failed"))
+ .thenReturn(true);
+ final PendingChanges pendingChanges = newPendingChanges(domain);
+ final CSN csn = pendingChanges.putLocalOperation(newLocalOperation());
+ final LDAPUpdateMsg large = newMsg(csn, sizeTaking(PendingChanges.MAX_RECENT_BYTES * 3 / 5));
+ try
+ {
+ pendingChanges.commitAndPushCommittedChanges(csn, large);
+ fail("the publish threw");
+ }
+ catch (IllegalStateException expected)
+ {
+ // The change is still pending, and the next push pushes it again.
+ }
+
+ final LDAPUpdateMsg next = pushLocalChange(pendingChanges, 0);
+
+ assertEquals(pendingChanges.recentChangesAfter(rightBefore(large), next.getCSN()).messages,
+ Arrays.asList(large, next));
+ }
+
+ /**
+ * A change the pending changes leave to a recovery is kept as well: the recovery of a later
+ * session, whose replication server misses it, publishes it as it was done.
+ */
+ @Test
+ public void aChangeLeftToARecoveryIsKeptAsWell() throws Exception
+ {
+ final ReplicationDomain domain = domainWhichPublishes(true);
+ when(domain.getServerState()).thenReturn(new ServerState());
+ final PendingChanges pendingChanges = newPendingChanges(domain);
+ final LDAPUpdateMsg published = pushLocalChange(pendingChanges);
+ pendingChanges.setRecovering(true);
+
+ final LDAPUpdateMsg kept = pushLocalChange(pendingChanges);
+
+ assertEquals(onlyMsgPublishedBy(domain), published);
+ assertEquals(pendingChanges.recentChangesAfter(published.getCSN(), kept.getCSN()).messages,
+ Arrays.asList(kept));
+ }
+
+ /**
+ * The messages kept are forgotten once the domain reads its state from data which may lack their
+ * changes, and none of the changes before is taken for kept: the state read back can be older than
+ * changes done since the domain started. The messages of the changes pushed from there are kept.
+ */
+ @Test
+ public void theMessagesKeptAreForgotten() throws Exception
+ {
+ final ReplicationDomain domain = domainWhichPublishes(true);
+ final PendingChanges pendingChanges = newPendingChanges(domain);
+ final CSN first = pushLocalChange(pendingChanges).getCSN();
+ final CSN last = pushLocalChange(pendingChanges).getCSN();
+ // The state read back is older than the last change.
+ when(domain.getLastLocalChange()).thenReturn(first);
+
+ pendingChanges.forgetRecentChanges();
+
+ final PendingChanges.RecentChanges fromFirst = pendingChanges.recentChangesAfter(first, last);
+ assertNull(fromFirst.messages, "the messages were forgotten");
+ assertNull(fromFirst.keptAfter, "and the change after the last one of the state is not taken for kept");
+
+ final LDAPUpdateMsg next = pushLocalChange(pendingChanges);
+ final LDAPUpdateMsg afterNext = pushLocalChange(pendingChanges);
+ assertNull(pendingChanges.recentChangesAfter(first, afterNext.getCSN()).messages,
+ "the change forgotten is not taken for kept");
+ assertEquals(pendingChanges.recentChangesAfter(next.getCSN(), afterNext.getCSN()).messages,
+ Arrays.asList(afterNext));
+ }
+
+ @Test
+ public void aCommittedChangeHeldBehindAnotherIsNotKeptAfterTheDataWasReplaced() throws Exception
+ {
+ final PendingChanges pending = newPendingChanges(domainWhichPublishes(false));
+ final CSN first = pending.putLocalOperation(newLocalOperation());
+ final CSN second = pending.putLocalOperation(newLocalOperation());
+ final LDAPUpdateMsg secondMsg = newMsg(second, 0);
+ pending.commitAndPushCommittedChanges(second, secondMsg);
+
+ pending.forgetRecentChanges();
+ pending.commitAndPushCommittedChanges(first, newMsg(first, 0));
+ final LDAPUpdateMsg after = pushLocalChange(pending);
+
+ assertNull(pending.recentChangesAfter(first, after.getCSN()).messages,
+ "the change made on the replaced data must not be treated as kept");
+ assertEquals(pending.recentChangesAfter(rightBefore(after), after.getCSN()).messages, Arrays.asList(after));
+ }
+
+ @Test
+ public void anUncommittedChangeAtDataReloadIsNotKeptAfterItCommits() throws Exception
+ {
+ final PendingChanges pending = newPendingChanges(domainWhichPublishes(false));
+ final CSN old = pending.putLocalOperation(newLocalOperation());
+ pending.forgetRecentChanges();
+ pending.commitAndPushCommittedChanges(old, newMsg(old, 0));
+ final LDAPUpdateMsg after = pushLocalChange(pending);
+
+ assertNull(pending.recentChangesAfter(rightBefore(newMsg(old, 0)), after.getCSN()).messages,
+ "a change pending at reload must not be published from the recent messages");
+ assertEquals(pending.recentChangesAfter(rightBefore(after), after.getCSN()).messages, Arrays.asList(after));
+ }
+
+ /** Pushes a local change whose message takes no bytes, and returns its message. */
+ private LDAPUpdateMsg pushLocalChange(PendingChanges pendingChanges)
+ {
+ return pushLocalChange(pendingChanges, 0);
+ }
+
+ /** Pushes a local change whose message has the provided size, and returns its message. */
+ private LDAPUpdateMsg pushLocalChange(PendingChanges pendingChanges, int size)
+ {
+ final CSN csn = pendingChanges.putLocalOperation(newLocalOperation());
+ final LDAPUpdateMsg msg = newMsg(csn, size);
+ pendingChanges.commitAndPushCommittedChanges(csn, msg);
+ return msg;
+ }
+
+ private static LDAPUpdateMsg newMsg(CSN csn, int size)
+ {
+ final LDAPUpdateMsg msg = mock(LDAPUpdateMsg.class);
+ when(msg.getCSN()).thenReturn(csn);
+ when(msg.size()).thenReturn(size);
+ return msg;
+ }
+
+ /** Returns the size of a message which takes the provided part of the heap once kept. */
+ private static int sizeTaking(long footprint)
+ {
+ return (int) ((footprint - PendingChanges.MESSAGE_OVERHEAD) / 2);
+ }
+
+ /** Returns the CSN right before the one of the message: the change before it, in the same millisecond. */
+ private static CSN rightBefore(LDAPUpdateMsg msg)
+ {
+ return new CSN(msg.getCSN().getTime(), msg.getCSN().getSeqnum() - 1, SERVER_ID);
+ }
+
+ /**
* Reports the forward of the message from within the publish which puts it on the wire, after
* checking from there that the announcement is already in place, and publishes it. The forward
* is the one of a peer the message was never recorded as queued for, which is what a forward
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java
index 8ea150e..344e491 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java
@@ -403,6 +403,61 @@
return new PersistentServerState(baseDn, serverId, new ServerState()).getMaxCSN(serverId);
}
+ @Test
+ public void suspendedSavesDrainBeforeDataReplacementAndStaySuspended() throws Exception
+ {
+ final DN dn = DN.valueOf(TEST_ROOT_DN_STRING);
+ final ServerState memory = new ServerState();
+ final CountDownLatch writing = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ final AtomicInteger writes = new AtomicInteger();
+ final PersistentServerState state = new HookedWrite(dn, 1, memory, null, entryDN -> {
+ if (writes.incrementAndGet() == 1)
+ {
+ writing.countDown();
+ try
+ {
+ assertTrue(release.await(30, SECONDS));
+ }
+ catch (InterruptedException e)
+ {
+ throw new AssertionError(e);
+ }
+ }
+ });
+ final CSN first = new CSNGenerator(1, memory).newCSN();
+ state.update(first);
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ final Thread writer = new Thread(state::save, "suspended state save");
+ writer.setDaemon(true);
+ writer.setUncaughtExceptionHandler((thread, error) -> failure.set(error));
+ try
+ {
+ writer.start();
+ assertTrue(writing.await(30, SECONDS));
+ assertFalse(state.suspendSaves(10), "data must not be replaced while the old write is pending");
+ release.countDown();
+ writer.join(30_000);
+ assertFalse(writer.isAlive());
+ assertNull(failure.get());
+ assertTrue(state.suspendSaves(1000));
+ final CSN newer = new CSNGenerator(2, memory).newCSN();
+ state.update(newer);
+ state.save();
+ assertEquals(writes.get(), 1, "a suspended save must not write stale state over imported data");
+ assertNull(loadMaxCSN(dn, 2));
+ state.resumeSaves();
+ state.save();
+ assertEquals(loadMaxCSN(dn, 2), newer);
+ }
+ finally
+ {
+ release.countDown();
+ writer.join(30_000);
+ state.resumeSaves();
+ }
+ }
+
/**
* A PersistentServerState whose write can be steered from the test, at the
* point where the snapshot that goes to the backend has already been taken.
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 0919df6..8f024b5 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
@@ -12,6 +12,7 @@
* information: "Portions copyright [year] [name of copyright owner]".
*
* Copyright 2026 3A Systems, LLC.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -24,7 +25,9 @@
import static org.testng.Assert.*;
import java.util.ArrayList;
+import java.lang.reflect.Field;
import java.util.List;
+import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
@@ -38,6 +41,7 @@
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
+import org.opends.server.core.LockFileManager;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
@@ -57,6 +61,7 @@
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
+import org.opends.server.tasks.TaskUtils;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -121,6 +126,620 @@
private ReplicationBroker exporter;
private CSNGenerator gen;
+ @Test(timeOut = 120_000)
+ public void aFailedImportUnlockForgetsMessagesOfTheReplacedData() throws Exception
+ {
+ TestCaseUtils.addEntry("dn: cn=before," + EXAMPLE_DN, "objectClass: top", "objectClass: person",
+ "cn: before", "sn: before");
+ final CSN change = domain.getLastLocalChange();
+ final CSN before = new CSN(change.getTime(), change.getSeqnum() - 1, DS_ID);
+ final Field pendingField = LDAPReplicationDomain.class.getDeclaredField("pendingChanges");
+ pendingField.setAccessible(true);
+ final PendingChanges pending = (PendingChanges) pendingField.get(domain);
+ assertThat(pending.recentChangesAfter(before, change).messages).hasSize(1);
+
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ final Field locksField = LockFileManager.class.getDeclaredField("exclusiveLocks");
+ final Field mapLockField = LockFileManager.class.getDeclaredField("mapLock");
+ locksField.setAccessible(true);
+ mapLockField.setAccessible(true);
+ final Map<?, ?> locks = (Map<?, ?>) locksField.get(null);
+ final Object mapLock = mapLockField.get(null);
+ try
+ {
+ final String[] exported = exportedEntries();
+ final CSN importedState = gen.newCSN();
+ exported[0] = exported[0].replace("dc: example\n", "dc: example\nds-sync-state: " + importedState + "\n");
+ startImportInto(exported.length);
+ waitUntil(() -> {
+ synchronized (mapLock)
+ {
+ return locks.containsKey(lockFile);
+ }
+ }, "the import did not acquire its exclusive lock");
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ finishImport(exported);
+ assertNull(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN),
+ "the failed unlock must leave the backend disabled");
+ assertThat(errorLogRecordsOf(org.opends.messages.ToolMessages.WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.ordinal()))
+ .isNotEmpty();
+ assertNull(pending.recentChangesAfter(before, change).messages,
+ "the import replaced the data even though its lock could not be released");
+ assertFalse(domain.isConnected(), "a failed cleanup must stop the replication session");
+ TaskUtils.enableBackend("userRoot");
+ domain.enable();
+ domain.enableService();
+ assertFalse(domain.isConnected(), "enabling the backend or service must not revive a failed import");
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection()
+ .processDelete(IMPORTED_ENTRY_DN).getResultCode(), ResultCode.UNAVAILABLE,
+ "writes must be rejected even when the backend is enabled again");
+ domain.backupStart();
+ domain.disable();
+ assertEquals(new PersistentServerState(baseDN, DS_ID,
+ new org.opends.server.replication.common.ServerState()).getMaxCSN(importedState.getServerId()),
+ importedState, "backup or disable must not overwrite the imported server state");
+ MultimasterReplication.deleteDomain(baseDN);
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection()
+ .processDelete(IMPORTED_ENTRY_DN).getResultCode(), ResultCode.UNAVAILABLE,
+ "deleting the replication domain must not permit writes to quarantined data");
+ final DomainFakeCfg replacement = new DomainFakeCfg(baseDN, DS_ID, new TreeSet<String>());
+ assertNull(MultimasterReplication.createNewDomain(replacement),
+ "recreating the domain must not clear the failed-import quarantine");
+ assertEquals(new MultimasterReplication().applyConfigurationAdd(replacement).getResultCode(),
+ ResultCode.UNAVAILABLE, "configuration add must report failure without dereferencing a missing domain");
+ }
+ finally
+ {
+ // A process restart clears the production quarantine; this test shares its JVM with other cases.
+ clearWriteFailure();
+ final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
+ ignored.setAccessible(true);
+ ignored.setBoolean(domain, false);
+ if (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) == null)
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ }
+ }
+
+ @Test(timeOut = 120_000)
+ public void anUnavailableStateReadAfterImportStopsTheDomain() throws Exception
+ {
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ ShortCircuitPlugin.registerShortCircuit(OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(),
+ 0, 1, operation -> {
+ final org.opends.server.types.operation.PreParseSearchOperation search =
+ (org.opends.server.types.operation.PreParseSearchOperation) operation;
+ return search.getRawBaseDN().toString().equals(EXAMPLE_DN)
+ && search.getAttributes().contains("ds-sync-state");
+ });
+ try
+ {
+ finishImport(exported);
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") > 0,
+ "the state-read failure was not injected");
+ assertFalse(domain.isConnected(), "an unavailable state read must not fall back to stale configuration state");
+ assertNotNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ clearWriteFailure();
+ }
+ }
+
+ @Test(timeOut = 120_000)
+ public void anImportWhichCannotAcquireItsLockDoesNotReleaseAnotherOwnersLock() throws Exception
+ {
+ TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: person", "cn: original", "sn: original");
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
+ try
+ {
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the import did not report its failed exclusive lock");
+ assertNotNull(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ assertTrue(entryExists(DN.valueOf("cn=original," + EXAMPLE_DN)));
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ final Field references = LockFileManager.class.getDeclaredField("sharedLockReferences");
+ final Field mapLock = LockFileManager.class.getDeclaredField("mapLock");
+ references.setAccessible(true);
+ mapLock.setAccessible(true);
+ synchronized (mapLock.get(null))
+ {
+ assertEquals(((Map<?, ?>) references.get(null)).get(lockFile), Integer.valueOf(2),
+ "the backend and the other owner must both retain their shared locks");
+ }
+ }
+ finally
+ {
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ }
+ }
+
+ @org.testng.annotations.DataProvider(name = "failedPreparation")
+ public Object[][] failedPreparation()
+ {
+ return new Object[][] { { false, false }, { true, false }, { true, true } };
+ }
+
+ /** A failed preparation must not quarantine data which the import never replaced. */
+ @Test(dataProvider = "failedPreparation", timeOut = 120_000)
+ public void aFailedPreparationCleanupLeavesTheOriginalDataRecoverable(boolean disabledBackend, boolean nestedBackend) throws Exception
+ {
+ if (nestedBackend)
+ {
+ TestCaseUtils.initializeMemoryBackend("importParent", "dc=com", true);
+ }
+ final Entry original = TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: person", "cn: original", "sn: original");
+ if (disabledBackend)
+ {
+ // Isolate the abandoned listing from legitimate redelivery of uncheckpointed local adds.
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ ((PersistentServerState) stateField.get(domain)).save();
+ // A delivery taken from the test queue has no replay thread to finish it. The next
+ // session normally redelivers it; this fixture isolates the cleanup of the old listing.
+ domain.processUpdate(new ModifyMsg(gen.newCSN(), original.getName(),
+ generatemods("description", "before failed preparation"), getEntryUUID(original.getName())));
+ queue.take();
+ }
+ final int stoppedRecords = errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).size();
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ boolean lockHeld = disabledBackend;
+ if (disabledBackend)
+ {
+ assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
+ }
+ ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", ResultCode.UNAVAILABLE.intValue(),
+ disabledBackend ? 1 : 0, 2, operation -> {
+ final org.opends.server.types.operation.PreParseModifyOperation modify =
+ (org.opends.server.types.operation.PreParseModifyOperation) operation;
+ return DN.valueOf(modify.getRawEntryDN().toString())
+ .equals(DN.valueOf("ds-cfg-backend-id=userRoot,cn=Backends,cn=config"));
+ });
+ try
+ {
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the failed preparation did not complete");
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse") > 0);
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN),
+ "an import which replaced no data must not quarantine the original suffix");
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ if (disabledBackend)
+ {
+ assertFalse(domain.isConnected(), "a domain with an unavailable backend must be stopped reversibly");
+ assertThat(errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).size()).isGreaterThan(stoppedRecords);
+ assertThat(errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).get(stoppedRecords))
+ .contains("severity=ERROR", baseDN.toString(), "userRoot", "disable", "enable");
+ if (nestedBackend)
+ {
+ final org.opends.server.api.LocalBackend<?> parent =
+ getServerContext().getBackendConfigManager().getLocalBackendById("importParent");
+ domain.performBackendPreFinalizationProcessing(parent);
+ domain.performBackendPostInitializationProcessing(parent);
+ assertFalse(domain.isConnected(), "enabling the parent must not revive the missing child backend");
+ }
+ else
+ {
+ // A missing top-level backend must not break another backend's removal.
+ TestCaseUtils.initializeMemoryBackend("foreignImportFinalization", "o=foreignImportFinalization", true);
+ final org.opends.server.api.LocalBackend<?> foreign =
+ getServerContext().getBackendConfigManager().getLocalBackendById("foreignImportFinalization");
+ try
+ {
+ domain.performBackendPreFinalizationProcessing(foreign);
+ assertFalse(domain.isConnected(), "a foreign backend must not restore this domain");
+ }
+ finally
+ {
+ foreign.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(foreign);
+ }
+ }
+ TaskUtils.enableBackend("userRoot");
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ lockHeld = false;
+ }
+ waitUntil(domain::isConnected, "the backend enable must restore replication without a process restart");
+ assertTrue(entryExists(original.getName()));
+ final ModifyMsg after = new ModifyMsg(gen.newCSN(), original.getName(),
+ generatemods("description", "after failed preparation"), getEntryUUID(original.getName()));
+ replayMsg(after);
+ assertTrue(domain.getServerState().cover(after.getCSN()),
+ "a delivery abandoned by the old session must not hold back the new session's state");
+ TestCaseUtils.addEntry("dn: cn=after," + EXAMPLE_DN,
+ "objectClass: person", "cn: after", "sn: after");
+ // The restored listener must consume new traffic, not just keep a connected broker.
+ startImportInto(exportedEntries().length);
+ finishImport(exportedEntries());
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ clearWriteFailure();
+ final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
+ ignored.setAccessible(true);
+ ignored.setBoolean(domain, false);
+ if (lockHeld)
+ {
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ }
+ if (!getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ if (nestedBackend)
+ {
+ final org.opends.server.api.LocalBackend<?> parent =
+ getServerContext().getBackendConfigManager().getLocalBackendById("importParent");
+ parent.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(parent);
+ }
+ }
+ }
+
+ /** Backend callbacks run under the configuration lock and must skip foreign domain locks. */
+ @Test
+ public void aForeignBackendCallbackDoesNotWaitForThisDomainsServiceLock() throws Exception
+ {
+ TestCaseUtils.initializeMemoryBackend("foreignImportCallback", "o=foreignImportCallback", true);
+ final org.opends.server.api.LocalBackend<?> foreign =
+ getServerContext().getBackendConfigManager().getLocalBackendById("foreignImportCallback");
+ final Field serviceLock = org.opends.server.replication.service.ReplicationDomain.class
+ .getDeclaredField("serviceStateLock");
+ serviceLock.setAccessible(true);
+ final CountDownLatch completed = new CountDownLatch(1);
+ final Thread callback = new Thread(() -> {
+ domain.performBackendPostInitializationProcessing(foreign);
+ completed.countDown();
+ }, "foreign backend initialization callback");
+ try
+ {
+ synchronized (serviceLock.get(domain))
+ {
+ callback.start();
+ assertTrue(completed.await(5, TimeUnit.SECONDS), "a foreign callback waited for the domain's service lock");
+ }
+ }
+ finally
+ {
+ callback.join(10000);
+ assertFalse(callback.isAlive());
+ foreign.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(foreign);
+ }
+ }
+
+ @org.testng.annotations.DataProvider(name = "abortedImportStop")
+ public Object[][] abortedImportStop()
+ {
+ return new Object[][] { { true }, { false } };
+ }
+
+ /** An enable during cleanup is retained, and stopping on the listener never needs window credit. */
+ @Test(dataProvider = "abortedImportStop", timeOut = 120_000)
+ public void anAbortedImportStopsWithoutCreditAndRetainsABackendEnable(boolean enableDuringStop) throws Exception
+ {
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().getLocalBackendById("userRoot"));
+ assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
+ boolean lockHeld = true;
+ final Field brokerField = org.opends.server.replication.service.ReplicationDomain.class.getDeclaredField("broker");
+ brokerField.setAccessible(true);
+ final ReplicationBroker broker = (ReplicationBroker) brokerField.get(domain);
+ final Field stopLockField = ReplicationBroker.class.getDeclaredField("startStopLock");
+ stopLockField.setAccessible(true);
+ final Field windowField = ReplicationBroker.class.getDeclaredField("sendWindow");
+ windowField.setAccessible(true);
+ final java.util.concurrent.Semaphore window = (java.util.concurrent.Semaphore) windowField.get(broker);
+ final Field disabledField = LDAPReplicationDomain.class.getDeclaredField("disabled");
+ disabledField.setAccessible(true);
+ final long timeout = MultimasterReplication.getConnectionTimeoutMS();
+ final DN provider = DN.valueOf("cn=Multimaster Synchronization,cn=Synchronization Providers,cn=config");
+ ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", ResultCode.UNAVAILABLE.intValue(),
+ 1, 2, operation -> DN.valueOf(((org.opends.server.types.operation.PreParseModifyOperation) operation)
+ .getRawEntryDN().toString()).equals(DN.valueOf("ds-cfg-backend-id=userRoot,cn=Backends,cn=config")));
+ try
+ {
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection().processModify(
+ modifyRequest(provider,
+ org.forgerock.opendj.ldap.ModificationType.REPLACE, "ds-cfg-connection-timeout", "0 ms"))
+ .getResultCode(), ResultCode.SUCCESS);
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ synchronized (stopLockField.get(broker))
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(disabledField.getBoolean(domain)));
+ if (enableDuringStop)
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ TaskUtils.enableBackend("userRoot");
+ }
+ else
+ {
+ // No update is needed: hold the actual window shut at the instant stop starts.
+ window.drainPermits();
+ }
+ }
+ new org.opends.server.util.TestTimer.Builder().maxSleep(5, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(!domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the listener must finish import cleanup without receiving window credit"));
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ if (!enableDuringStop)
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "the backend enable must restore the domain"));
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ lockHeld = false;
+ startImportInto(exportedEntries().length);
+ finishImport(exportedEntries());
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
+ }
+ finally
+ {
+ // Release a broken implementation too, so the negative run cannot wedge the test JVM.
+ window.release(100000);
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ waitUntil(() -> !domain.ieRunning(), "import cleanup did not finish after credit was restored");
+ final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
+ ignored.setAccessible(true);
+ ignored.setBoolean(domain, false);
+ if (!getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ if (lockHeld)
+ {
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ }
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection().processModify(
+ modifyRequest(provider,
+ org.forgerock.opendj.ldap.ModificationType.REPLACE, "ds-cfg-connection-timeout", timeout + " ms"))
+ .getResultCode(), ResultCode.SUCCESS);
+ }
+ }
+
+ /** The reconnect after an abandoned import already satisfies replay's outstanding restart. */
+ @Test(timeOut = 120_000)
+ public void anAbandonedImportConsumesTheReplayRestart() throws Exception
+ {
+ final Entry original = TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: person", "cn: original", "sn: original");
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ final Object state = stateField.get(domain);
+ final Field saveLockField = PersistentServerState.class.getDeclaredField("saveLock");
+ saveLockField.setAccessible(true);
+ final java.util.concurrent.locks.Lock saveLock = (java.util.concurrent.locks.Lock) saveLockField.get(state);
+ final Field importing = LDAPReplicationDomain.class.getDeclaredField("importingData");
+ importing.setAccessible(true);
+ final CSN csn = gen.newCSN();
+ final ModifyMsg msg = new ModifyMsg(csn, original.getName(), generatemods("description", "replayed"),
+ getEntryUUID(original.getName()));
+ final ModifyMsg queued = new ModifyMsg(gen.newCSN(), original.getName(),
+ generatemods("description", "queued before the import"), getEntryUUID(original.getName()));
+ domain.processUpdate(queued);
+ final LDAPUpdateMsg queuedDelivery = queue.take().getUpdateMessage();
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ domain.setReplayDrainTimeout(3000);
+ saveLock.lock();
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ new org.opends.server.util.TestTimer.Builder().maxSleep(30, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(importing.getBoolean(domain)));
+ replayMsg(msg);
+ assertFalse(domain.getServerState().cover(csn), "the replay must be deferred by the import");
+ domain.failNextSessionRestarts(1);
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the import did not abandon its wait for the state save");
+ Thread.sleep(2000);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "the checkpointer restarted the session which import completion already restarted");
+ assertTrue(domain.isConnected());
+ replayMsg(msg);
+ // This delivery was already queued before the import: it must still be listed, without
+ // processUpdate inserting it again. Clearing all pending changes would discard its replay.
+ replay(queuedDelivery);
+ assertTrue(domain.getServerState().cover(queued.getCSN()));
+ assertThat(DirectoryServer.getEntry(original.getName()).getAllAttributes("description")).isNotEmpty();
+ }
+ finally
+ {
+ saveLock.unlock();
+ domain.failNextSessionRestarts(0);
+ domain.setReplayDrainTimeout(30_000);
+ }
+ }
+
+ @org.testng.annotations.DataProvider(name = "importDrains")
+ public Object[][] importDrains()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "importDrains", timeOut = 120_000)
+ public void aTimedOutDrainLeavesTheOriginalDataUsable(boolean stateSave) throws Exception
+ {
+ final Entry original = TestCaseUtils.makeEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: top", "objectClass: person", "cn: original", "sn: original");
+ TestCaseUtils.addEntry(original);
+ final Object owner;
+ final Field lockField;
+ if (stateSave)
+ {
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ owner = stateField.get(domain);
+ lockField = PersistentServerState.class.getDeclaredField("saveLock");
+ }
+ else
+ {
+ owner = domain;
+ lockField = LDAPReplicationDomain.class.getDeclaredField("replayReadLock");
+ }
+ lockField.setAccessible(true);
+ final java.util.concurrent.locks.Lock lock = (java.util.concurrent.locks.Lock) lockField.get(owner);
+ final Object backend = getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN);
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ final int shutdownWarnings = errorLogRecordsOf(WARN_REPLAY_NOT_DRAINED.ordinal()).size();
+ final int drainErrors = errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).size();
+ domain.setReplayDrainTimeout(100);
+ lock.lock();
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the import did not abandon the drain before replacing data");
+ assertThat(errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).size()).isGreaterThan(drainErrors);
+ assertThat(errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).get(drainErrors))
+ .contains("severity=ERROR");
+ assertSame(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN), backend);
+ assertTrue(entryExists(original.getName()), "a rejected preflight must leave the original data intact");
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ assertThat(errorLogRecordsOf(WARN_REPLAY_NOT_DRAINED.ordinal())).hasSize(shutdownWarnings);
+ final ErrorMsg failure = waitForSpecificMsg(exporter, ErrorMsg.class);
+ assertEquals(failure.getDetails().toString(), ERR_INIT_OPERATIONS_NOT_DRAINED.get(baseDN).toString());
+ }
+ finally
+ {
+ lock.unlock();
+ domain.setReplayDrainTimeout(30_000);
+ }
+ waitUntil(domain::isConnected, "a preflight failure must leave the domain recoverable");
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ finishImport(exported);
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)), "a later import must still work");
+ }
+
+ @org.testng.annotations.DataProvider(name = "interruptedPreparation")
+ public Object[][] interruptedPreparation()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "interruptedPreparation", timeOut = 120_000)
+ public void anInterruptedImportDrainReportsTheInterrupt(boolean waitingForSave) throws Exception
+ {
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ final Object state = stateField.get(domain);
+ final Field suspended = PersistentServerState.class.getDeclaredField("savesSuspended");
+ suspended.setAccessible(true);
+ final Field lockField = waitingForSave
+ ? PersistentServerState.class.getDeclaredField("saveLock")
+ : LDAPReplicationDomain.class.getDeclaredField("replayReadLock");
+ lockField.setAccessible(true);
+ final java.util.concurrent.locks.Lock lock =
+ (java.util.concurrent.locks.Lock) lockField.get(waitingForSave ? state : domain);
+ final Field listenerField = org.opends.server.replication.service.ReplicationDomain.class
+ .getDeclaredField("listenerThread");
+ listenerField.setAccessible(true);
+ final Thread listener = (Thread) listenerField.get(domain);
+ final Field importing = LDAPReplicationDomain.class.getDeclaredField("importingData");
+ importing.setAccessible(true);
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ final int interrupts = errorLogRecordsOf(ERR_INIT_PREPARATION_INTERRUPTED.ordinal()).size();
+ lock.lock();
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> {
+ assertTrue(importing.getBoolean(domain));
+ if (waitingForSave)
+ {
+ assertTrue(suspended.getBoolean(state), "the replay drain must finish before interruption");
+ assertEquals(listener.getState(), Thread.State.TIMED_WAITING);
+ }
+ });
+ listener.interrupt();
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the interrupted import did not finish");
+ final List<String> records = errorLogRecordsOf(ERR_INIT_PREPARATION_INTERRUPTED.ordinal());
+ assertThat(records.subList(interrupts, records.size()))
+ .isNotEmpty().allSatisfy(record -> assertThat(record).contains("severity=ERROR", "interrupted"));
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ waitUntil(domain::isConnected, "an interrupted preparation must leave the domain recoverable");
+ startImportInto(exportedEntries().length);
+ finishImport(exportedEntries());
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
+ }
+
+ @Test
+ public void writeQuarantineSurvivesProviderRemovalAndRejectsMovesIntoTheSuffix() throws Exception
+ {
+ TestCaseUtils.addEntry("dn: ou=People," + EXAMPLE_DN, "objectClass: organizationalUnit", "ou: People");
+ TestCaseUtils.addEntry("dn: cn=outside," + EXAMPLE_DN,
+ "objectClass: person", "cn: outside", "sn: outside");
+ final DN quarantined = DN.valueOf("ou=People," + EXAMPLE_DN);
+ final java.util.List<org.opends.server.api.SynchronizationProvider<
+ org.forgerock.opendj.server.config.server.SynchronizationProviderCfg>> providers =
+ new ArrayList<>(getSynchronizationProviders());
+ org.opends.server.api.LocalBackend.quarantineWrites(quarantined,
+ org.forgerock.i18n.LocalizableMessage.raw("failed import test"));
+ try
+ {
+ for (org.opends.server.api.SynchronizationProvider<?> provider : providers)
+ {
+ deregisterSynchronizationProvider(provider);
+ }
+ final org.opends.server.protocols.internal.InternalClientConnection connection =
+ org.opends.server.protocols.internal.InternalClientConnection.getRootConnection();
+ assertEquals(connection.processDelete(quarantined).getResultCode(), ResultCode.UNAVAILABLE);
+ assertEquals(connection.processModifyDN("cn=outside," + EXAMPLE_DN, "cn=outside", true,
+ quarantined.toString()).getResultCode(), ResultCode.UNAVAILABLE,
+ "a move into quarantined data must be refused even when its source is healthy");
+ assertTrue(entryExists(DN.valueOf("cn=outside," + EXAMPLE_DN)));
+ }
+ finally
+ {
+ getSynchronizationProviders().addAll(providers);
+ final Field failures = org.opends.server.api.LocalBackend.class.getDeclaredField("writeFailures");
+ failures.setAccessible(true);
+ ((Map<?, ?>) failures.get(null)).remove(quarantined);
+ }
+ }
+
+ private void clearWriteFailure() throws Exception
+ {
+ final Field failures = org.opends.server.api.LocalBackend.class.getDeclaredField("writeFailures");
+ failures.setAccessible(true);
+ ((Map<?, ?>) failures.get(null)).remove(baseDN);
+ }
+
@BeforeMethod
public void setUpLocal() throws Exception
{
@@ -1120,7 +1739,7 @@
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)
+ while (getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
{
assertTrue(System.currentTimeMillis() < deadline,
"the import did not deregister the backend of the domain");
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
index 16b07f6..f460e63 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java
@@ -25,6 +25,9 @@
import static org.testng.Assert.*;
import java.net.SocketTimeoutException;
+import java.io.InputStream;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.ExecutionException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -1646,6 +1649,194 @@
}
}
+ /** A failed target must not keep the exporter occupied until an operator restarts its process. */
+ @Test
+ public void permanentlyStoppedImportersReleaseTheExporter() throws Exception
+ {
+ final DN baseDN = DN.valueOf("o=test");
+ final int port = TestCaseUtils.findFreePort();
+ final ReplicationServer rs = createReplicationServer(15, port, "stoppedImportersDb", 100);
+ final SortedSet<String> servers = newTreeSet("localhost:" + port);
+ FakeReplicationDomain exporter = null;
+ FakeReplicationDomain first = null;
+ FakeReplicationDomain second = null;
+ FakeReplicationDomain healthy = null;
+ FutureTask<Void> export = null;
+ Thread exportThread = null;
+ final CountDownLatch targetsStopped = new CountDownLatch(2);
+ try
+ {
+ exporter = new FakeReplicationDomain(baseDN, 1, servers, 0, buildExportedData(3), null, 3);
+ first = stoppingImporter(baseDN, 2, servers, targetsStopped);
+ second = stoppingImporter(baseDN, 3, servers, targetsStopped);
+ final FakeReplicationDomain source = exporter;
+ newTimer().repeatUntilSuccess(() -> assertEquals(source.getReplicaInfos().size(), 2));
+ export = new FutureTask<>(() -> {
+ source.initializeRemote(org.opends.server.replication.protocol.RoutableMsg.ALL_SERVERS, NO_INIT_TASK);
+ return null;
+ });
+ exportThread = new Thread(export, "export to permanently stopped targets");
+ exportThread.start();
+ assertTrue(targetsStopped.await(30, SECONDS), "the importers did not finish receiving their data");
+ final long stoppedAt = System.nanoTime();
+ try
+ {
+ export.get(20, SECONDS);
+ fail("an export to disconnected targets must report failure");
+ }
+ catch (ExecutionException e)
+ {
+ assertTrue(e.getCause() instanceof DirectoryException, e.toString());
+ assertEquals(((DirectoryException) e.getCause()).getMessageObject().toString(),
+ ERR_INIT_NO_SUCCESS_END_FROM_SERVERS.get(source.getGenerationID(), newTreeSet(2, 3)).toString());
+ }
+ assertTrue(System.nanoTime() - stoppedAt >= SECONDS.toNanos(9),
+ "two absent targets must not halve the reconnection grace");
+ assertFalse(source.ieRunning(), "failed export retained its context");
+
+ final StringBuffer imported = new StringBuffer();
+ healthy = new FakeReplicationDomain(baseDN, 4, servers, 0, null, imported, 0);
+ newTimer().repeatUntilSuccess(() -> assertTrue(source.getReplicaInfos().containsKey(4)));
+ source.initializeRemote(4, NO_INIT_TASK);
+ assertEquals(imported.toString(), buildExportedData(3));
+ assertFalse(source.ieRunning());
+ }
+ finally
+ {
+ disable(exporter, first, second, healthy);
+ if (exportThread != null)
+ {
+ exportThread.join(30000);
+ }
+ remove(rs);
+ }
+ }
+
+ @DataProvider(name = "importingTarget")
+ public Object[][] importingTarget()
+ {
+ // Each target takes a turn importing, so no iteration order of HashSet is assumed.
+ return new Object[][] { { 2 }, { 3 }, { 4 } };
+ }
+
+ /** Missing peers get independent grace periods even while another peer is still importing. */
+ @Test(dataProvider = "importingTarget")
+ public void absentTargetsAreTimedIndependentlyWhileAnotherImports(int importing) throws Exception
+ {
+ final int absent = importing == 2 ? 3 : 2;
+ final int later = 9 - importing - absent;
+ final EndOfInitWait wait = new EndOfInitWait(2, 3, 4);
+ final java.util.concurrent.atomic.AtomicBoolean expiredDuringImport = new java.util.concurrent.atomic.AtomicBoolean();
+ wait.topology = elapsed -> {
+ if (elapsed < 13000 && wait.failures().contains(absent))
+ {
+ expiredDuringImport.set(true);
+ }
+ final Map<Integer, DSInfo> peers = new java.util.HashMap<>();
+ peers.put(importing, initTarget(importing,
+ elapsed < 13000 ? ServerStatus.FULL_UPDATE_STATUS : ServerStatus.NORMAL_STATUS));
+ if (elapsed < 6000 || elapsed >= 12000)
+ {
+ peers.put(later, initTarget(later,
+ elapsed < 6000 ? ServerStatus.FULL_UPDATE_STATUS : ServerStatus.NORMAL_STATUS));
+ }
+ return peers;
+ };
+ wait.run();
+ assertEquals(wait.failures(), newTreeSet(absent), "the later absence must have its own full grace");
+ assertTrue(expiredDuringImport.get(), "a FULL_UPDATE peer prevented timing an absent peer");
+ }
+
+ /** Two short absences separated by a reconnect must not be added into one long absence. */
+ @Test
+ public void aReconnectDuringFullUpdateResetsTheAbsenceTimer() throws Exception
+ {
+ final EndOfInitWait wait = new EndOfInitWait(2);
+ wait.topology = elapsed -> elapsed >= 14000
+ ? Collections.singletonMap(2, initTarget(2, ServerStatus.NORMAL_STATUS))
+ : elapsed >= 6000 && elapsed < 8000
+ ? Collections.singletonMap(2, initTarget(2, ServerStatus.FULL_UPDATE_STATUS))
+ : Collections.emptyMap();
+ wait.run();
+ assertTrue(wait.failures().isEmpty(), "the reconnect did not reset the target's absence timer");
+ }
+
+ private static DSInfo initTarget(int id, ServerStatus status)
+ {
+ return new DSInfo(id, "localhost", 1, 1, status, false,
+ org.opends.server.replication.common.AssuredMode.SAFE_READ_MODE, (byte) 1, (byte) 1,
+ Collections.emptyList(), Collections.emptySet(), Collections.emptySet(),
+ ProtocolVersion.getCurrentVersion());
+ }
+
+ /** Runs the real completion loop against a controlled topology, without network timing races. */
+ private static final class EndOfInitWait extends DummyReplicationDomain
+ {
+ private final ImportExportContext context;
+ private final java.util.Set<Integer> failures;
+ private java.util.function.LongFunction<Map<Integer, DSInfo>> topology;
+ private long started;
+
+ @SuppressWarnings("unchecked")
+ EndOfInitWait(Integer... targets) throws Exception
+ {
+ super(1);
+ final java.lang.reflect.Constructor<ImportExportContext> constructor =
+ ImportExportContext.class.getDeclaredConstructor(boolean.class);
+ constructor.setAccessible(true);
+ context = constructor.newInstance(false);
+ final java.lang.reflect.Field starts = ImportExportContext.class.getDeclaredField("startList");
+ starts.setAccessible(true);
+ ((java.util.Set<Integer>) starts.get(context)).addAll(java.util.Arrays.asList(targets));
+ final java.lang.reflect.Field failed = ImportExportContext.class.getDeclaredField("failureList");
+ failed.setAccessible(true);
+ failures = (java.util.Set<Integer>) failed.get(context);
+ final java.lang.reflect.Field broker = ReplicationDomain.class.getDeclaredField("broker");
+ broker.setAccessible(true);
+ // Only supplies the external shutdown signal; the production wait and its timers are real.
+ broker.set(this, org.mockito.Mockito.mock(ReplicationBroker.class));
+ }
+
+ java.util.Set<Integer> failures()
+ {
+ return failures;
+ }
+
+ @Override
+ public Map<Integer, DSInfo> getReplicaInfos()
+ {
+ final long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ assertTrue(elapsed < 35000, "the completion loop did not release absent targets");
+ return topology.apply(elapsed);
+ }
+
+ void run() throws Exception
+ {
+ final java.lang.reflect.Method method = ReplicationDomain.class
+ .getDeclaredMethod("waitForRemoteEndOfInit", ImportExportContext.class);
+ method.setAccessible(true);
+ started = System.nanoTime();
+ method.invoke(this, context);
+ }
+ }
+
+ private FakeReplicationDomain stoppingImporter(DN baseDN, int id, SortedSet<String> servers,
+ CountDownLatch stopped) throws Exception
+ {
+ return new FakeReplicationDomain(baseDN, id, servers, 0, null, new StringBuffer(), 0)
+ {
+ @Override
+ protected void importBackend(InputStream input) throws DirectoryException
+ {
+ super.importBackend(input);
+ stopServicePermanently();
+ stopped.countDown();
+ throw new DirectoryException(org.forgerock.opendj.ldap.ResultCode.OTHER,
+ LocalizableMessage.raw("injected cleanup failure"));
+ }
+ };
+ }
+
/** InitializeTask double recording the completion reported by the domain. */
private static final class RecordingInitializeTask extends InitializeTask
{
--
Gitblit v1.10.0