From f0a287fcba94440f81845a04fd0b6a4a70075a2c Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 19 Oct 2026 08:28:51 +0000
Subject: [PATCH] [#1195] Publish every change done while a recovery runs, and end the recovery only once they are out (#1204)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/FakeDelOperation.java | 30
opendj-server-legacy/src/test/java/org/opends/server/replication/common/CSNGeneratorTest.java | 41
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java | 311 ++++
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 6
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java | 2852 +++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/common/CSNGenerator.java | 11
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 1173 +++++++++++++--
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/LDAPUpdateMsg.java | 15
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java | 56
9 files changed, 4,297 insertions(+), 198 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/common/CSNGenerator.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/common/CSNGenerator.java
index a93a7e4..c6d99d7 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/common/CSNGenerator.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/common/CSNGenerator.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2009 Sun Microsystems, Inc.
* Portions Copyright 2011-2014 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.common;
@@ -115,17 +116,15 @@
* received {@link CSN}
*
* @param number
- * the {@link CSN} to adjust with
+ * the {@link CSN} to adjust with, or {@code null} for none: the generator is then left
+ * as it is. Moved back to the clock, it numbered the next CSNs as, or before, ones it had
+ * generated or seen already - a domain reads back a ServerState which holds no change of
+ * its own after a restore or a total update (issue #1223).
*/
public void adjust(CSN number)
{
if (number==null)
{
- synchronized(this)
- {
- lastTime = TimeThread.getTime();
- seqnum = 0;
- }
return;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/FakeDelOperation.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/FakeDelOperation.java
index 6e11f82..af49740 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/FakeDelOperation.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/FakeDelOperation.java
@@ -13,6 +13,7 @@
*
* Copyright 2009 Sun Microsystems, Inc.
* Portions copyright 2012-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -32,6 +33,8 @@
{
private final DN dn;
private final String entryUUID;
+ /** Whether a replication server has taken the delete: set by the threads which publish it. */
+ private volatile boolean published;
/**
* Creates a new FakeDelOperation from the provided information.
@@ -64,4 +67,31 @@
{
return entryUUID;
}
+
+ /**
+ * Retrieves the DN of the entry that was deleted with this operation.
+ *
+ * @return The DN of the entry that was deleted with this operation.
+ */
+ DN getDN()
+ {
+ return dn;
+ }
+
+ /** Records that a replication server has taken the delete: the broker wrote it to a session. */
+ void published()
+ {
+ published = true;
+ }
+
+ /**
+ * Tells whether a replication server has taken the delete: one which none has is published by
+ * nobody once it is dropped from the deletes the domain keeps aside.
+ *
+ * @return {@code true} if the broker wrote the delete to a session
+ */
+ boolean isPublished()
+ {
+ return published;
+ }
}
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 5126a65..5cd3bac 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
@@ -46,9 +46,9 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.NavigableMap;
import java.util.NoSuchElementException;
import java.util.Set;
-import java.util.SortedMap;
import java.util.StringTokenizer;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
@@ -57,8 +57,8 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Consumer;
import net.jcip.annotations.GuardedBy;
@@ -129,6 +129,7 @@
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.OperationContext;
import org.opends.server.replication.protocol.RoutableMsg;
+import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationBroker;
@@ -221,11 +222,14 @@
{
private final CSN startCSN;
private final CSN endCSN;
+ /** What the search found, for the round which runs it alone: see {@link #replayOperations}. */
+ private final NavigableMap<CSN, FakeOperation> found;
- public ScanSearchListener(CSN startCSN, CSN endCSN)
+ public ScanSearchListener(CSN startCSN, CSN endCSN, NavigableMap<CSN, FakeOperation> found)
{
this.startCSN = startCSN;
this.endCSN = endCSN;
+ this.found = found;
}
@Override
@@ -234,19 +238,22 @@
throws DirectoryException
{
// Build the list of Operations that happened on this entry after startCSN
- // and before endCSN and add them to the replayOperations list
+ // and up to endCSN and add them to the list of the round.
+ // CSNs are ordered by time first, so the changes of the other replicas which this one
+ // replayed fall in the range as well: they are left out, it is not for this replica to
+ // publish them.
+ // The end is part of the range, as it is of the search: the next range starts after it, so
+ // a change numbered exactly the end would be in neither.
Iterable<FakeOperation> updates =
EntryHistorical.generateFakeOperations(searchEntry);
for (FakeOperation op : updates)
{
CSN csn = op.getCSN();
- if (csn.isNewerThan(startCSN) && csn.isOlderThan(endCSN))
+ if (csn.getServerId() == getServerId()
+ && csn.isNewerThan(startCSN) && !csn.isNewerThan(endCSN))
{
- synchronized (replayOperations)
- {
- replayOperations.put(csn, op);
- }
+ found.put(csn, op);
}
}
}
@@ -564,7 +571,25 @@
* replication server and that the ServerState is not updated too early.
*/
private final PendingChanges pendingChanges;
- private final AtomicReference<RSUpdater> rsUpdater = new AtomicReference<>(null);
+ /**
+ * Guards {@link #rsUpdater} and the start a session hands over to it, and is held as a recovery
+ * starts, ends, and as its publisher leaves: a session which needs a recovery either hands its
+ * start to the publisher which runs - which then neither ends the recovery nor leaves before it
+ * has run from it - or finds none and starts one. Nothing is done under it which takes another
+ * lock: a session starts under the lock of the connect phase of the broker, and a recovery ends
+ * under the lock of the pending changes.
+ */
+ private final Object recoveryLock = new Object();
+ /** The missing change publisher which runs, if any: under {@link #recoveryLock}. */
+ private RSUpdater rsUpdater;
+ /**
+ * How many missing change publishers in a row failed - their search did not go through, or they
+ * threw - since a recovery last ended: the wait before the session restart each of them asks for
+ * is computed from it. It is a count of its own rather than the one of the session restarts a
+ * failed replay asks for, which a change replayed resets: the replication of the other replicas
+ * into this one goes on while a recovery keeps failing.
+ */
+ private final AtomicInteger consecutiveRecoveryFailures = new AtomicInteger();
/**
* It contain the updates that were done on other servers, transmitted by the
@@ -600,10 +625,43 @@
private volatile boolean importingData;
/**
- * This list is used to temporary store operations that needs to be replayed
- * at session establishment time.
+ * The local deletes, kept aside for the recoveries: a deleted entry leaves no historical
+ * information to publish the delete again from.
+ * <p>
+ * A delete stays here once it is published, and once a recovery is over: the replication server
+ * of a later session may miss it. The list holds the last {@link #MAX_DELETES_KEPT_ASIDE} deletes,
+ * and a delete which no replication server has taken yet is not dropped without a trace. What the
+ * search of a round of a recovery finds is the round's own, so that it neither takes room here
+ * nor stays after the round.
*/
- private final SortedMap<CSN, FakeOperation> replayOperations = new TreeMap<>();
+ private final NavigableMap<CSN, FakeDelOperation> replayOperations = new TreeMap<>();
+ /** How many local deletes {@link #replayOperations} keeps at most. */
+ private static final int MAX_DELETES_KEPT_ASIDE = 10000;
+ /** How many local deletes {@link #replayOperations} keeps at most: fewer in the tests which fill it. */
+ private volatile int maxDeletesKeptAside = MAX_DELETES_KEPT_ASIDE;
+
+ /**
+ * Run by {@link #buildAndPublishMissingChanges(CSN, ReplicationBroker)} after the search of
+ * each of its rounds, before it chooses what the round publishes. Only there for the tests,
+ * which do local changes there - a change done while a round searches is one the search may or
+ * may not have seen, and nothing else puts a change in that gap on cue - or fail the search,
+ * which a local backend which works does not do on cue either.
+ */
+ private volatile Consumer<InternalSearchOperation> missingChangesSearchHook;
+ /**
+ * Run by the missing change publisher after each of its runs, before it takes the start a session
+ * handed over or leaves. Only there for the tests which start a session at that point - once a run
+ * has ended the recovery and before its publisher has left - where nothing else holds it on cue.
+ */
+ private volatile Runnable missingChangesPublisherRunHook;
+ /**
+ * Run by a round of a recovery which publishes the local changes the recovery kept, once it has
+ * published them and before it ends the recovery, handed whether the round holds the local changes
+ * back. Only there for the tests which do local changes at that point, or start a session there -
+ * one which hands its start over while the round holds the local changes back - where nothing else
+ * holds the round on cue: it searches nothing.
+ */
+ private volatile Consumer<Boolean> keptChangesPublishedHook;
/** Published by a configuration change, read by the changelog threads without a lock. */
private volatile ExternalChangelogDomain eclDomain;
@@ -790,18 +848,56 @@
}
}
+ /** Where a missing change publisher starts, and the session it publishes on. */
+ private static final class RecoveryStart
+ {
+ /** The newest change of this replica the replication server of the session holds, or less. */
+ private final CSN csn;
+ /** The session the publisher publishes on: its broker refuses the messages for another one. */
+ private final Session session;
+
+ private RecoveryStart(CSN csn, Session session)
+ {
+ this.csn = csn;
+ this.session = session;
+ }
+ }
+
/**
* The thread that is responsible to update the RS to which this domain is
* connected in case it is late and there is no RS which is up to date.
+ * <p>
+ * One runs at a time. A session which starts while it runs hands it its start, and the publisher
+ * goes back there and publishes on that session from it: its broker refuses what it publishes for
+ * the session before, and the replication server of the new one - which may hold less than the one
+ * before - gets each change it misses, in order. The recovery is not ended while a start is handed
+ * over, and the publisher does not leave before it has run from it.
*/
private class RSUpdater extends DirectoryThread
{
- private final CSN startCSN;
+ /** Where the next run of this publisher starts: under {@link #recoveryLock}, {@code null} once taken. */
+ private RecoveryStart nextStart;
- protected RSUpdater(CSN replServerMaxCSN)
+ protected RSUpdater(RecoveryStart start)
{
super("Replica DS(" + getServerId() + ") missing change publisher for domain \"" + getBaseDN() + "\"");
- this.startCSN = replServerMaxCSN;
+ this.nextStart = start;
+ }
+
+ /**
+ * Hands over the start of a session which started while this publisher runs, and ends the wait of
+ * a failed run for the session restart: this session is the next one. Under {@link #recoveryLock}.
+ */
+ private void handOver(RecoveryStart start)
+ {
+ nextStart = start;
+ recoveryLock.notifyAll();
+ }
+
+ /** Whether a session which started while this publisher runs handed it its start: under {@link #recoveryLock}. */
+ private boolean hasAStartHandedOver()
+ {
+ return nextStart != null;
}
@Override
@@ -811,49 +907,145 @@
// let's send them to him.
logger.trace(DEBUG_GOING_TO_SEARCH_FOR_CHANGES);
- /*
- * Get all the changes that have not been seen by this
- * replication server and publish them.
- */
+ boolean startTheSessionAgain = false;
try
{
- if (buildAndPublishMissingChanges(startCSN, broker))
+ for (RecoveryStart start = takeStart(); start != null; start = takeStart())
{
- logger.trace(DEBUG_CHANGES_SENT);
- synchronized(replayOperations)
+ startTheSessionAgain = false;
+ try
{
- replayOperations.clear();
+ /*
+ * Get all the changes that have not been seen by this
+ * replication server and publish them.
+ * The deletes kept aside are not dropped once the recovery is over: a later session may
+ * reach a replication server which misses some of them, and they have nothing else to be
+ * published from.
+ */
+ if (buildAndPublishMissingChanges(start.csn, broker, start.session, this))
+ {
+ logger.trace(DEBUG_CHANGES_SENT);
+ consecutiveRecoveryFailures.set(0);
+ }
+ // Otherwise the broker refused a change - it reaches no replication server, its session
+ // was replaced, or it was stopped - and the publisher runs from the start the session
+ // which replaced it handed over, or the recovery of the session it opens next publishes
+ // the change; or this domain is going away.
+ }
+ catch (Exception e)
+ {
+ /*
+ * The search of a round did not go through, or the round threw. Nothing it did not
+ * publish is lost - a round whose search failed publishes nothing, and the deletes kept
+ * aside and the local changes the recovery keeps stay where they are - and the recovery
+ * is not over: the local changes are still left to it. No recovery of this session would
+ * publish them, as a session which stays up starts no other, so the session is started
+ * again, and the recovery of the next one publishes them. A domain which is going away
+ * needs none, and one which is disabled starts a session of its own when it is enabled
+ * back.
+ *
+ * The backend is left some time to recover first, with the session up meanwhile, so that
+ * the replication of the other replicas into this one goes on: the restart is asked for
+ * once the wait is over, and not with the backoff of the restart a failed replay asks
+ * for, which stops the session before it waits. This publisher waits registered: a
+ * session which comes meanwhile hands it its start, which ends the wait, and is the next
+ * session the publisher runs from, with no restart.
+ */
+ if (!shutdown.get() && !disabled)
+ {
+ final int failures = consecutiveRecoveryFailures.incrementAndGet();
+ logger.error(ERR_MISSING_CHANGES_NOT_PUBLISHED, getBaseDN(),
+ e instanceof MissingChangesSearchFailedException ? e.getMessage() : stackTraceToSingleLineString(e),
+ sessionRestartBackoffInMs(failures));
+ waitForAStartOrTheBackoff(failures);
+ startTheSessionAgain = true;
+ }
+ }
+ final Runnable runHook = missingChangesPublisherRunHook;
+ if (runHook != null)
+ {
+ runHook.run();
}
}
- else
- {
- /*
- * An error happened trying to search for the updates
- * This server will start accepting again new updates but
- * some inconsistencies will stay between servers.
- * Log an error for the repair tool
- * that will need to re-synchronize the servers.
- */
- logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
- }
- }
- catch (Exception e)
- {
- /*
- * An error happened trying to search for the updates
- * This server will start accepting again new updates but
- * some inconsistencies will stay between servers.
- * Log an error for the repair tool
- * that will need to re-synchronize the servers.
- */
- logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
}
finally
{
- broker.setRecoveryRequired(false);
- // RSUpdater thread has finished its work, let's remove it from memory
- // so another RSUpdater thread can be started if needed.
- rsUpdater.compareAndSet(this, null);
+ /*
+ * Left without taking a start only on an error, which goes on up from here: the recovery is
+ * not over, and no session would start another - the start a session handed over meanwhile
+ * included - so the session is started again, as after a run which failed.
+ */
+ synchronized (recoveryLock)
+ {
+ if (rsUpdater == this)
+ {
+ rsUpdater = null;
+ broker.setRecoveryRequired(false);
+ startTheSessionAgain = true;
+ }
+ }
+ if (startTheSessionAgain && !shutdown.get() && !disabled)
+ {
+ /*
+ * Asked for once this publisher is gone, so that the session the restart brings up starts a
+ * publisher of its own. The state checkpointer runs it within a second, or once a total
+ * update is over. The domain is checked again, as it may have been disabled or shut down
+ * during the wait.
+ */
+ sessionRestarts.request(SessionRestart.NOW);
+ }
+ }
+ }
+
+ /**
+ * Waits before the session restart a failed run asks for: one second more for each failure in a
+ * row, up to ten. A session which hands its start over meanwhile ends the wait, as it is the next
+ * session; so does the domain as it is shut down or disabled.
+ */
+ private void waitForAStartOrTheBackoff(int failures)
+ {
+ final long until = monotonicNowInMs() + sessionRestartBackoffInMs(failures);
+ synchronized (recoveryLock)
+ {
+ for (long left = until - monotonicNowInMs();
+ left > 0 && !hasAStartHandedOver() && !shutdown.get() && !disabled;
+ left = until - monotonicNowInMs())
+ {
+ try
+ {
+ recoveryLock.wait(left);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+ }
+
+ /**
+ * Takes the start of the next run of this publisher, or has it leave if there is none - under
+ * the lock a session hands its start over under, so that a session either finds this publisher
+ * still to run from its start, or finds none and starts one.
+ *
+ * @return the start of the next run, or {@code null} once this publisher has left
+ */
+ private RecoveryStart takeStart()
+ {
+ synchronized (recoveryLock)
+ {
+ final RecoveryStart start = nextStart;
+ nextStart = null;
+ if (start == null || shutdown.get())
+ {
+ // RSUpdater thread has finished its work, let's remove it from memory
+ // so another RSUpdater thread can be started if needed.
+ rsUpdater = null;
+ broker.setRecoveryRequired(false);
+ return null;
+ }
+ return start;
}
}
}
@@ -1926,16 +2118,28 @@
ctx = new DeleteContext(csn, modifiedEntryUUID);
deleteOperation.setAttachment(SYNCHROCONTEXT, ctx);
+ final int maxDeletes = maxDeletesKeptAside;
+ FakeDelOperation dropped = null;
synchronized (replayOperations)
{
- int size = replayOperations.size();
- if (size >= 10000)
+ if (maxDeletes > 0)
{
- replayOperations.remove(replayOperations.firstKey());
+ if (replayOperations.size() >= maxDeletes)
+ {
+ dropped = replayOperations.pollFirstEntry().getValue();
+ }
+ replayOperations.put(csn, new FakeDelOperation(deleteOperation.getEntryDN(), csn, modifiedEntryUUID));
}
- FakeOperation op = new FakeDelOperation(
- deleteOperation.getEntryDN(), csn, modifiedEntryUUID);
- replayOperations.put(csn, op);
+ }
+ /*
+ * The oldest delete goes first: it is the least likely one to be missed by a replication server.
+ * One which no replication server has taken - done while none took the changes, or while a
+ * recovery which kept too many of them runs - is published by nobody from now on, and the
+ * replicas which miss it keep the entry.
+ */
+ if (dropped != null && !dropped.isPublished())
+ {
+ logger.error(ERR_UNPUBLISHED_DELETE_DROPPED, getBaseDN(), dropped.getDN(), dropped.getCSN(), maxDeletes);
}
}
@@ -2376,6 +2580,36 @@
}
/**
+ * {@inheritDoc}
+ * <p>
+ * A local delete the broker wrote is marked as published among the deletes kept aside: dropping
+ * it from there later loses nothing a replication server has not taken.
+ */
+ @Override
+ public boolean publish(UpdateMsg msg)
+ {
+ final boolean published = super.publish(msg);
+ if (published && msg instanceof DeleteMsg)
+ {
+ deleteKeptAsidePublished(msg.getCSN());
+ }
+ return published;
+ }
+
+ /** Marks the delete kept aside with the provided CSN, if there is one, as taken by a replication server. */
+ private void deleteKeptAsidePublished(CSN csn)
+ {
+ synchronized (replayOperations)
+ {
+ final FakeDelOperation delete = replayOperations.get(csn);
+ if (delete != null)
+ {
+ delete.published();
+ }
+ }
+ }
+
+ /**
* Check if an operation must be synchronized.
* Also update the list of pending changes and the server RUV
* @param op the operation
@@ -2473,6 +2707,19 @@
}
else if (!op.isSynchronizationOperation() && curCSN != null)
{
+ /*
+ * A delete is kept aside as its conflicts are resolved, before the access control, the plugins
+ * and the backend have had their say: one which did not go through is no change, and a
+ * recovery would publish it as one - the replicas it reached would delete an entry this one
+ * keeps.
+ */
+ if (op.getOperationType() == OperationType.DELETE)
+ {
+ synchronized (replayOperations)
+ {
+ replayOperations.remove(curCSN);
+ }
+ }
// Remove an unsuccessful non-replication operation from the pending
// changes list.
pendingChanges.remove(curCSN);
@@ -2634,10 +2881,14 @@
*/
awaitReplayDrained();
- final RSUpdater rsUpdater = this.rsUpdater.get();
- if (rsUpdater != null)
+ synchronized (recoveryLock)
{
- rsUpdater.initiateShutdown();
+ if (rsUpdater != null)
+ {
+ rsUpdater.initiateShutdown();
+ }
+ // A publisher which waits to have the session started again has nothing left to wait for.
+ recoveryLock.notifyAll();
}
// stop the thread in charge of flushing the ServerState.
@@ -4472,16 +4723,15 @@
*/
private void waitBeforeSessionRestart(int restarts, long wakes)
{
- final long until = monotonicNowInMs()
- + Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS);
+ final long until = monotonicNowInMs() + sessionRestartBackoffInMs(restarts);
try
{
/*
* Waited on a monitor rather than slept through, so that a domain which is going
* away or is being disabled is not waited out: the thread which holds this wait is
* a replay thread the shutdown of the pool joins, or the state checkpointer the
- * shutdown of the domain waits for. The session it would start back is one the
- * domain is stopping anyway.
+ * shutdown of the domain waits for - or a missing change publisher whose recovery
+ * failed. The session it would start back is one the domain is stopping anyway.
*
* The wake is counted rather than only checked for by its flag: a domain which was
* disabled and enabled back has its flag cleared again, and a wait which read the
@@ -4510,6 +4760,18 @@
}
/**
+ * Returns how long the session is left before it is started again, after the provided number
+ * of restarts in a row.
+ *
+ * @param restarts how many times in a row the session was restarted already
+ * @return the wait, in milliseconds
+ */
+ private static long sessionRestartBackoffInMs(int restarts)
+ {
+ return Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS);
+ }
+
+ /**
* Has the next session restarts of this domain fail where they would start the session
* again.
* <p>
@@ -5339,6 +5601,12 @@
* BackendConfigManager's write lock when a backend is being deregistered.
*/
disabled = true;
+ // A missing change publisher which waits to have the session started again has nothing left
+ // to wait for: the domain starts a session of its own when it is enabled back.
+ synchronized (recoveryLock)
+ {
+ recoveryLock.notifyAll();
+ }
disableService(); // This will cut the session and wake up the listener
awaitReplayDrained();
state.save();
@@ -5502,13 +5770,159 @@
{
state.clearInMemory();
state.loadState();
+ bringAStateWithNoChangeOfThisReplicaUpToTheData();
getGenerator().adjust(state.getMaxCSN(getServerId()));
+ forgetTheLocalChangesTheDataLacks();
// Retrieves the generation ID associated with the data imported
generationId = loadGenerationId();
}
/**
+ * 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.
+ * <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()})
+ * - 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()
+ {
+ final CSN newest = state.getMaxCSN(getServerId());
+
+ // Looked up without the locks of the lists: a local operation holds the lock of its entry, which
+ // the lookup may need, while it waits for the lock of the pending changes.
+ final List<FakeDelOperation> deletes;
+ synchronized (replayOperations)
+ {
+ deletes = new ArrayList<>((newest != null ? replayOperations.tailMap(newest, false) : replayOperations).values());
+ }
+ final List<CSN> deletesTheDataLacks = new ArrayList<>();
+ for (FakeDelOperation delete : deletes)
+ {
+ if (holdsTheEntry(delete.getEntryUUID()))
+ {
+ deletesTheDataLacks.add(delete.getCSN());
+ }
+ }
+ synchronized (replayOperations)
+ {
+ replayOperations.keySet().removeAll(deletesTheDataLacks);
+ }
+
+ final List<CSN> keptChangesTheDataLacks = new ArrayList<>();
+ for (LDAPUpdateMsg kept : pendingChanges.keptChangesNewerThan(newest))
+ {
+ if (!(kept instanceof DeleteMsg) || holdsTheEntry(kept.getEntryUUID()))
+ {
+ keptChangesTheDataLacks.add(kept.getCSN());
+ }
+ }
+ pendingChanges.dropKeptChanges(keptChangesTheDataLacks);
+ }
+
+ /**
+ * Brings a ServerState just read back which holds no change of this replica up to the changes of
+ * this replica the data holds, from the first local change the recovery which runs kept.
+ * <p>
+ * The state read back may miss the last changes of this replica: the save which was to write them
+ * did not - a save gives its turn up to the one in flight, and a write may fail. The state is
+ * brought up to the changes the historical information of the entries holds past the newest change
+ * of this replica it read as it is loaded, but one which read none is left as it is, as that search
+ * would go through every entry this replica ever changed. Such a state is read back once the base
+ * entry, which holds the saved state, was deleted and added back, and no save wrote the state into it
+ * since: the local changes a recovery kept meanwhile would all be taken for changes the data does not
+ * 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()
+ {
+ if (state.getMaxCSN(getServerId()) != null)
+ {
+ return;
+ }
+ final List<LDAPUpdateMsg> kept = pendingChanges.keptChangesNewerThan(null);
+ if (kept.isEmpty())
+ {
+ return;
+ }
+ final CSN newest = newestChangeOfThisReplicaInTheDataFrom(kept.get(0).getCSN());
+ if (newest != null)
+ {
+ state.update(newest);
+ }
+ }
+
+ /**
+ * Returns the newest change of this replica the historical information of the entries holds, among
+ * the changes from the provided one on.
+ *
+ * @return the change, or {@code null} if the data holds none, or the search did not go through: the
+ * 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)
+ {
+ final InternalSearchOperation op;
+ try
+ {
+ op = searchForChangedEntries(getBaseDN(), from, null);
+ }
+ catch (Exception e)
+ {
+ logger.traceException(e);
+ logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
+ return null;
+ }
+ if (op.getResultCode() != ResultCode.SUCCESS)
+ {
+ // A suffix whose base entry is gone holds no change.
+ if (op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
+ {
+ logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
+ }
+ return null;
+ }
+ CSN newest = null;
+ for (SearchResultEntry entry : op.getSearchEntries())
+ {
+ for (FakeOperation change : EntryHistorical.generateFakeOperations(entry))
+ {
+ final CSN csn = change.getCSN();
+ if (csn.getServerId() == getServerId() && (newest == null || csn.isNewerThan(newest)))
+ {
+ newest = csn;
+ }
+ }
+ }
+ return newest;
+ }
+
+ /**
+ * Tells whether the data of this domain holds the entry with the provided entryUUID. A lookup
+ * 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)
+ {
+ try
+ {
+ return findEntryDN(entryUUID) != null;
+ }
+ catch (SearchFailedException e)
+ {
+ return true;
+ }
+ }
+
+ /**
* Enable back the domain after a previous disable.
* The domain will connect back to a replication Server and
* will recreate threads to listen for messages from the Synchronization
@@ -6769,6 +7183,7 @@
* changes.
*/
CSN replServerMaxCSN = rsState.getCSN(getServerId());
+ final CSN ourMaxCSN = state.getMaxCSN(getServerId());
// we don't want to update from here (a DS) an empty RS because
// normally the RS should have been updated by other RSes except for
@@ -6776,20 +7191,32 @@
// ... hence the RS we are connected to should not be empty
// ... or if it is empty, it is due to a voluntary reset
// and we don't want to update it with our changes that could be huge.
- if (replServerMaxCSN != null && replServerMaxCSN.getSeqnum() != 0)
+ final boolean replServerHoldsChanges = replServerMaxCSN != null && replServerMaxCSN.getSeqnum() != 0;
+ if (replServerHoldsChanges
+ && ourMaxCSN != null
+ && !ourMaxCSN.isOlderThanOrEqualTo(replServerMaxCSN))
{
- CSN ourMaxCSN = state.getMaxCSN(getServerId());
- if (ourMaxCSN != null
- && !ourMaxCSN.isOlderThanOrEqualTo(replServerMaxCSN))
- {
- pendingChanges.setRecovering(true);
- broker.setRecoveryRequired(true);
- final RSUpdater rsUpdater = new RSUpdater(replServerMaxCSN);
- if (this.rsUpdater.compareAndSet(null, rsUpdater))
- {
- rsUpdater.start();
- }
- }
+ startMissingChangesPublisher(replServerMaxCSN, true);
+ }
+ else if (pendingChanges.isRecovering())
+ {
+ /*
+ * A missing change publisher which failed - its broker reached no replication server, or
+ * was stopped, or its search failed and it had the session started again - left the
+ * pending changes in recovery, and no recovery of this session would end it, as this
+ * replication server misses no change of this replica: the local changes would be
+ * recorded in the state and published by nobody. A publisher is started all the same,
+ * from what the replication server holds: it finds nothing to publish, or the changes
+ * done since this check, and ends the recovery under the lock the local changes are
+ * pushed under. The recovery is not ended here, as this runs under the lock of the
+ * connect phase of the broker, which a change being pushed waits for holding the lock of
+ * the pending changes.
+ *
+ * A publisher of the previous session which still runs is handed this start: it publishes
+ * on this session from there, and ends the recovery then.
+ */
+ startMissingChangesPublisher(replServerHoldsChanges ? replServerMaxCSN
+ : ourMaxCSN != null ? ourMaxCSN : new CSN(TimeThread.getTime(), 0, getServerId()), false);
}
} catch (Exception e)
{
@@ -6798,133 +7225,539 @@
}
/**
+ * Has the changes of this replica which come after the provided CSN published on the session
+ * which starts: starts the thread which publishes them, or hands the start over to the one which
+ * runs, which then publishes on this session from there - a publisher which kept going from where
+ * it had got to on the session before would leave the replication server of this one without the
+ * changes it misses before that point, and it would take none of them once it had a later one.
+ * <p>
+ * Decided under the lock a recovery ends under, and its publisher leaves under: a publisher which
+ * was about to end the recovery or to leave runs from this start instead.
+ *
+ * @param startCSN
+ * the newest change of this replica the replication server of the session holds, or the
+ * CSN after which a recovery left by an earlier session is to be ended
+ * @param replServerMisses
+ * whether the replication server misses changes of this replica: the recovery starts,
+ * rather than only being ended if a failed publisher left one
+ */
+ private void startMissingChangesPublisher(CSN startCSN, boolean replServerMisses)
+ {
+ final RecoveryStart start = new RecoveryStart(startCSN, broker.getSession());
+ synchronized (recoveryLock)
+ {
+ if (replServerMisses)
+ {
+ pendingChanges.setRecovering(true);
+ broker.setRecoveryRequired(true);
+ }
+ else if (!pendingChanges.isRecovering())
+ {
+ // The publisher of an earlier session ended it since it was checked.
+ return;
+ }
+ if (rsUpdater == null)
+ {
+ rsUpdater = new RSUpdater(start);
+ rsUpdater.start();
+ }
+ else
+ {
+ rsUpdater.handOver(start);
+ }
+ }
+ }
+
+ /**
* 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
+ * 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
+ * ran, the round after it holds them back, so that the recovery ends with it. The local changes
+ * wait for as long as publishing the ones pushed during the round before takes, and for no search.
+ * <p>
+ * A recovery which kept too many of the local changes keeps none, and searches for them as well,
+ * from where it has got to: a round then publishes up to the last change done when it began, or up
+ * to the end of its search window, and a change done while it runs is left to the round after it.
+ * The search may have missed it, and publishing a later change in its place would have the next
+ * round search past it - while a replication server takes no change older than the newest one it
+ * holds of this replica. Once such a round has caught up with the local changes and some were done
+ * while it ran, the rounds after it hold them back up to the last one.
+ * <p>
+ * A round counts as published only what the broker did not refuse: when it refuses a change - it
+ * reaches no replication server, its session is not the one the recovery publishes on anymore, or
+ * it was stopped - the recovery fails, and the changes are left to the start the session which
+ * replaced it hands over, or to the next session, which starts a publisher whether its
+ * replication server misses changes of this replica or not, so that the recovery ends.
+ * <p>
+ * A round whose search does not go through publishes nothing, and the recovery fails with it: what
+ * the search found may be some of the changes only, and a replication server which took them would
+ * take none of the ones it missed, which are older. The caller has the session started again, for
+ * the recovery of the next one to publish them.
+ * <p>
+ * The deletes kept aside are left where they are, the ones a round publishes included: the
+ * replication server of a later session may miss them.
*
* @param startCSN
* The CSN where we need to start the search
* @param session
- * The session to use to publish the changes
- * @return A boolean indicating he success of the operation.
+ * The broker to publish the changes with
+ * @param recoverySession
+ * The session to publish the changes on: the broker refuses them for another one
+ * @param publisher
+ * The missing change publisher which runs the recovery, which a session which starts
+ * meanwhile hands its start over to, or {@code null} if none does
+ * @return {@code true} once the recovery is over, {@code false} if the broker refused a change, a
+ * session which started meanwhile handed its start over, or the domain was shut down
* @throws Exception
- * if an Exception happens during the search.
+ * if the search of a round did not go through, or a round threw: the changes the round
+ * did not publish are left to the next recovery
*/
- boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session)
- throws Exception
+ private boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session,
+ Session recoverySession, RSUpdater publisher) throws Exception
{
- // Trim the changes in replayOperations that are older than the startCSN.
- synchronized (replayOperations)
- {
- Iterator<CSN> it = replayOperations.keySet().iterator();
- while (it.hasNext())
- {
- if (shutdown.get())
- {
- return false;
- }
- if (it.next().isNewerThan(startCSN))
- {
- break;
- }
- it.remove();
- }
- }
-
- CSN lastRetrievedChange;
- InternalSearchOperation op;
+ final MissingChanges missingChanges = new MissingChanges(session, recoverySession, publisher);
CSN currentStartCSN = startCSN;
- do
+ boolean holdBackLocalChanges = false;
+ while (true)
{
- if (shutdown.get())
+ if (shutdown.get() || missingChanges.startHandedOver())
{
return false;
}
-
- lastRetrievedChange = null;
- // We can't do the search in one go because we need to store the results
- // so that we are sure we send the operations in order and because the
- // list might be large.
- // So we search by interval of 10 seconds and store the results in the
- // replayOperations list so that they are sorted before sending them.
- long missingChangesDelta = currentStartCSN.getTime() + 10000;
- CSN endCSN = new CSN(missingChangesDelta, 0xffffffff, getServerId());
-
- ScanSearchListener listener =
- new ScanSearchListener(currentStartCSN, endCSN);
- op = searchForChangedEntries(getBaseDN(), currentStartCSN, endCSN,
- listener);
-
- // Publish and remove all the changes from the replayOperations list
- // that are older than the endCSN.
- final List<FakeOperation> opsToSend = new LinkedList<>();
- synchronized (replayOperations)
+ final CSN fromCSN = currentStartCSN;
+ final MissingChangesRound round = holdBackLocalChanges
+ ? pendingChanges.holdBackLocalChangesWhile(() -> missingChanges.publishUntilCaughtUp(fromCSN))
+ : missingChanges.publish(fromCSN, false);
+ if (round == null || !round.succeeded)
{
- Iterator<FakeOperation> itOp = replayOperations.values().iterator();
- while (itOp.hasNext())
- {
- if (shutdown.get())
- {
- return false;
- }
- FakeOperation fakeOp = itOp.next();
- if (fakeOp.getCSN().isNewerThan(endCSN) // sanity check
- || !state.cover(fakeOp.getCSN())
- // do not look for replay operations in the future
- || currentStartCSN.isNewerThan(now()))
- {
- break;
- }
+ return false;
+ }
+ if (round.endedRecovery)
+ {
+ return true;
+ }
+ holdBackLocalChanges = round.caughtUp;
+ currentStartCSN = round.publishedUpTo;
+ }
+ }
- lastRetrievedChange = fakeOp.getCSN();
- opsToSend.add(fakeOp);
- itOp.remove();
+ /**
+ * Publishes the changes of this replica which come after the provided CSN, as
+ * {@link #buildAndPublishMissingChanges(CSN, ReplicationBroker, Session, RSUpdater)} does, on the
+ * session the provided broker publishes on, and with no missing change publisher a session could
+ * hand a start over to.
+ *
+ * @param startCSN
+ * The CSN where we need to start the search
+ * @param session
+ * The broker to publish the changes with
+ * @return {@code true} once the recovery is over, {@code false} if the broker refused a change or
+ * the domain was shut down
+ * @throws Exception
+ * if the search of a round did not go through, or a round threw
+ */
+ @VisibleForTesting
+ boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session) throws Exception
+ {
+ return buildAndPublishMissingChanges(startCSN, session, session.getSession(), null);
+ }
+
+ /** What one round of {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker)} did. */
+ private static final class MissingChangesRound
+ {
+ /** Every change of this replica up to this CSN is published, or has nothing left to publish it from. */
+ private final CSN publishedUpTo;
+ /** Whether the round went as far as the last local change: neither its search window nor the search of the recovery ended before. */
+ private final boolean caughtUp;
+ /** Whether the broker refused none of the changes the round published: a failed search throws. */
+ private final boolean succeeded;
+ /** Whether the round ended the recovery: it caught up, and no change was done while it ran. */
+ private final boolean endedRecovery;
+
+ private MissingChangesRound(CSN publishedUpTo, boolean caughtUp, boolean succeeded,
+ boolean endedRecovery)
+ {
+ this.publishedUpTo = publishedUpTo;
+ this.caughtUp = caughtUp;
+ this.succeeded = succeeded;
+ this.endedRecovery = endedRecovery;
+ }
+ }
+
+ /** One run of {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker, Session, RSUpdater)}. */
+ private final class MissingChanges
+ {
+ private final ReplicationBroker recoveryBroker;
+ private final Session recoverySession;
+ private final RSUpdater publisher;
+
+ private MissingChanges(ReplicationBroker recoveryBroker, Session recoverySession, RSUpdater publisher)
+ {
+ this.recoveryBroker = recoveryBroker;
+ this.recoverySession = recoverySession;
+ this.publisher = publisher;
+ }
+
+ /** Whether a session which started meanwhile handed its start over to the publisher of this run. */
+ private boolean startHandedOver()
+ {
+ if (publisher == null)
+ {
+ return false;
+ }
+ synchronized (recoveryLock)
+ {
+ return publisher.hasAStartHandedOver();
+ }
+ }
+
+ /**
+ * Ends the recovery, under the lock a session starts one under, unless a session which started
+ * meanwhile handed its start over: its replication server may miss changes this run published to
+ * the one before, and the recovery goes on from that start.
+ */
+ private void endRecovery(Runnable recovered)
+ {
+ synchronized (recoveryLock)
+ {
+ if (publisher == null || !publisher.hasAStartHandedOver())
+ {
+ recovered.run();
+ broker.setRecoveryRequired(false);
}
}
+ }
- for (FakeOperation opToSend : opsToSend)
+ /**
+ * Runs the rounds of a recovery which holds the local changes back, up to the last local change.
+ * <p>
+ * The changes done while the round before ran may go further than one search window: a round
+ * which stopped at the end of its window and let the local changes go again would leave the next
+ * one as far behind. Nothing is done while these rounds run, so the last of them ends the
+ * recovery.
+ *
+ * @return the last round, or {@code null} if the domain was shut down meanwhile
+ */
+ private MissingChangesRound publishUntilCaughtUp(CSN fromCSN) throws Exception
+ {
+ MissingChangesRound round = publish(fromCSN, true);
+ while (round != null && round.succeeded && !round.caughtUp)
{
if (shutdown.get())
{
- return false;
+ return null;
}
- session.publishRecovery(opToSend.generateMessage());
+ round = publish(round.publishedUpTo, true);
+ }
+ return round;
+ }
+
+ /**
+ * 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.
+ *
+ * @param heldBack
+ * whether the round runs while the local changes are held 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 publish(CSN fromCSN, boolean heldBack) throws Exception
+ {
+ if (importingData)
+ {
+ return refusedForTheImport(fromCSN);
+ }
+ final PendingChanges.KeptChanges keeping = pendingChanges.keptChanges();
+ if (keeping == null)
+ {
+ return publishMissingChanges(fromCSN, null);
+ }
+ if (keeping.searchUpTo != null && fromCSN.isOlderThan(keeping.searchUpTo))
+ {
+ return publishMissingChanges(fromCSN, keeping.searchUpTo);
+ }
+ final List<LDAPUpdateMsg> kept = pendingChanges.keptChangesAfter(fromCSN);
+ if (kept == null)
+ {
+ // It kept too many since.
+ return publishMissingChanges(fromCSN, null);
+ }
+ return publishKeptChanges(fromCSN, kept, heldBack);
+ }
+
+ /**
+ * 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
+ * import has a recovery of its own, from the state the new data holds.
+ */
+ private MissingChangesRound refusedForTheImport(CSN fromCSN)
+ {
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("publish loop from " + fromCSN
+ + " stopped: a total update replaces the data of this replica"));
+ }
+ return new MissingChangesRound(fromCSN, false, false, false);
+ }
+
+ /**
+ * Runs a round which publishes the changes of this replica which come after {@code fromCSN},
+ * rebuilt from the historical information: up to {@code searchUpTo}, where the search of the
+ * recovery ends, or - if it is {@code null} - up to the last change done when the round begins,
+ * and in either case up to the end of the search window of the round if that comes first. A
+ * round which went up to the last change done ends the recovery if no change was done since.
+ *
+ * @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 publishMissingChanges(CSN fromCSN, CSN searchUpTo) throws Exception
+ {
+ // Read before the search: every change up to it has its entry written, so the search sees it.
+ final CSN lastChange = searchUpTo != null ? searchUpTo : getLastLocalChange();
+ // We can't do the search in one go because we need to store the results
+ // so that we are sure we send the operations in order and because the
+ // list might be large.
+ // So we search by interval of 10 seconds and store the results in a
+ // list of the round so that they are sorted before sending them.
+ // The greatest sequence number reads as the greatest both as a CSN and as the text the index
+ // of the historical information compares, where 0xffffffff would read as -1 and as the
+ // greatest: the changes of the last millisecond fell then out of this window and of the next.
+ final CSN endCSN = new CSN(fromCSN.getTime() + 10000, Integer.MAX_VALUE, getServerId());
+ final boolean reachesLastChange = lastChange == null || !lastChange.isNewerThan(endCSN);
+ final CSN bound = !reachesLastChange ? endCSN : lastChange != null ? lastChange : fromCSN;
+ // Where the search of the recovery ends, the local changes it kept come next: only a round which
+ // went up to the last local change catches up with them.
+ final boolean caughtUp = reachesLastChange && searchUpTo == null;
+
+ final NavigableMap<CSN, FakeOperation> found = new TreeMap<>();
+ final InternalSearchOperation op =
+ searchForChangedEntries(getBaseDN(), fromCSN, endCSN, new ScanSearchListener(fromCSN, endCSN, found));
+ final Consumer<InternalSearchOperation> searchHook = missingChangesSearchHook;
+ if (searchHook != null)
+ {
+ searchHook.accept(op);
+ }
+ /*
+ * A search which did not go through may have found some of the changes only, and a replication
+ * server which took them would take none of the ones it missed, which are older: the round
+ * publishes nothing. A search which found no base entry went through: the base entry of the
+ * domain is gone, and no entry is left below it to hold a change, so the deletes kept aside are
+ * all there is - and they go out as in any other round, or the deletes which removed the suffix
+ * would wait for it to come back. The same result comes for a base DN with no backend - one an
+ * import has taken away - and that search found nothing of what is there.
+ */
+ final ResultCode resultCode = op.getResultCode();
+ if (!resultCode.equals(ResultCode.SUCCESS)
+ && !(resultCode.equals(ResultCode.NO_SUCH_OBJECT) && baseEntryIsAbsentFromALiveBackend()))
+ {
+ throw new MissingChangesSearchFailedException(fromCSN, op);
}
- if (lastRetrievedChange != null)
+ /*
+ * Publish what the search found and the deletes kept aside which come after fromCSN, up to the
+ * bound, in order. The deletes stay kept aside, the ones before fromCSN as well: the replication
+ * server of this session has them, but the one of a later session may not, and a delete has
+ * nothing else to be published from.
+ */
+ final NavigableMap<CSN, FakeOperation> opsToSend = new TreeMap<>(found.headMap(bound, true));
+ if (bound.isNewerThan(fromCSN))
{
- if (logger.isDebugEnabled())
+ synchronized (replayOperations)
{
- logger.debug(LocalizableMessage.raw("publish loop"
- + " >=" + currentStartCSN + " <=" + endCSN
- + " nentries=" + op.getEntriesSent()
- + " result=" + op.getResultCode()
- + " lastRetrievedChange=" + lastRetrievedChange));
+ opsToSend.putAll(replayOperations.subMap(fromCSN, false, bound, true));
}
- currentStartCSN = lastRetrievedChange;
}
- else
- {
- if (logger.isDebugEnabled())
- {
- logger.debug(LocalizableMessage.raw("publish loop"
- + " >=" + currentStartCSN + " <=" + endCSN
- + " nentries=" + op.getEntriesSent()
- + " result=" + op.getResultCode()
- + " no changes"));
- }
- currentStartCSN = endCSN;
- }
- } while (pendingChanges.recoveryUntil(currentStartCSN)
- && op.getResultCode().equals(ResultCode.SUCCESS));
- return op.getResultCode().equals(ResultCode.SUCCESS);
+ for (FakeOperation opToSend : opsToSend.values())
+ {
+ if (shutdown.get())
+ {
+ return null;
+ }
+ if (importingData)
+ {
+ return refusedForTheImport(fromCSN);
+ }
+ if (!recoveryBroker.publishRecovery(opToSend.generateMessage(), recoverySession))
+ {
+ return refused(fromCSN, opToSend.getCSN());
+ }
+ if (opToSend instanceof FakeDelOperation)
+ {
+ ((FakeDelOperation) opToSend).published();
+ }
+ }
+
+ final CSN publishedUpTo = bound.isNewerThan(fromCSN) ? bound : fromCSN;
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("publish loop"
+ + " >=" + fromCSN + " <=" + endCSN
+ + " nentries=" + op.getEntriesSent()
+ + " result=" + op.getResultCode()
+ + " published=" + opsToSend.size()
+ + " upTo=" + publishedUpTo));
+ }
+ final boolean endedRecovery = caughtUp && !pendingChanges.recoveryUntil(publishedUpTo, this::endRecovery);
+ return new MissingChangesRound(publishedUpTo, caughtUp, true, endedRecovery);
+ }
+
+ /**
+ * Runs a round which publishes the local changes the recovery kept after {@code fromCSN}, in
+ * order, and ends the recovery if no change was pushed since it took them.
+ *
+ * @param heldBack
+ * whether the round runs while the local changes are held back
+ * @return the round, or {@code null} if the domain was shut down meanwhile
+ */
+ private MissingChangesRound publishKeptChanges(CSN fromCSN, List<LDAPUpdateMsg> kept, boolean heldBack)
+ {
+ CSN publishedUpTo = fromCSN;
+ for (LDAPUpdateMsg msg : kept)
+ {
+ if (shutdown.get())
+ {
+ return null;
+ }
+ if (importingData)
+ {
+ return refusedForTheImport(fromCSN);
+ }
+ if (!recoveryBroker.publishRecovery(msg, recoverySession))
+ {
+ return refused(fromCSN, msg.getCSN());
+ }
+ publishedUpTo = msg.getCSN();
+ if (msg instanceof DeleteMsg)
+ {
+ deleteKeptAsidePublished(msg.getCSN());
+ }
+ }
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("publish loop of the kept changes >" + fromCSN
+ + " published=" + kept.size() + " upTo=" + publishedUpTo));
+ }
+ final Consumer<Boolean> publishedHook = keptChangesPublishedHook;
+ if (publishedHook != null)
+ {
+ publishedHook.accept(heldBack);
+ }
+ final boolean endedRecovery = !pendingChanges.recoveryUntil(publishedUpTo, this::endRecovery);
+ if (!endedRecovery && heldBack && !startHandedOver())
+ {
+ /*
+ * Nothing is pushed while the local changes are held back, so the changes kept reach the last
+ * local change - unless one which should have been kept is not there. Rather than go round
+ * for ever under the lock every local change waits for, the recovery keeps none from here on,
+ * and searches for them in the historical information.
+ */
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("the changes kept up to " + publishedUpTo
+ + " do not reach the last local change: the recovery searches for the ones after them"));
+ }
+ pendingChanges.keepNoMoreChanges();
+ return new MissingChangesRound(publishedUpTo, false, true, false);
+ }
+ return new MissingChangesRound(publishedUpTo, true, true, endedRecovery);
+ }
+
+ /**
+ * The broker refused a change - it reaches no replication server, its session is not the one
+ * of this run anymore, or it was stopped: the changes from it are left where they are, to the
+ * start a session which replaced it hands over, or to the recovery of the next session.
+ */
+ private MissingChangesRound refused(CSN fromCSN, CSN csn)
+ {
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("publish loop refused " + csn
+ + ": the changes from it are left to the start a session hands over, or to the recovery of the next one"));
+ }
+ return new MissingChangesRound(fromCSN, false, false, false);
+ }
}
- private static CSN now()
+ /**
+ * Thrown by a round of {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker)}
+ * whose search did not go through.
+ */
+ private static final class MissingChangesSearchFailedException extends Exception
{
- // ensure now() will always come last with isNewerThan() test,
- // even though the timestamp, or the timestamp and seqnum would be the same
- return new CSN(TimeThread.getTime(), Integer.MAX_VALUE, Integer.MAX_VALUE);
+ private static final long serialVersionUID = 1L;
+
+ private MissingChangesSearchFailedException(CSN fromCSN, InternalSearchOperation search)
+ {
+ super("the search of the changes after " + fromCSN + " ended with "
+ + search.getResultCode().getName() + " " + search.getErrorMessage());
+ }
+ }
+
+ /**
+ * Sets what {@link #buildAndPublishMissingChanges(CSN, ReplicationBroker)} runs after the search
+ * of each of its rounds, before it chooses what the round publishes.
+ * <p>
+ * Only there for the tests which do local changes in that gap - the changes a round may publish
+ * are bounded by what was done when its search began, and a change done in between is the one
+ * a bound taken any later lets through without the search having found it - or which fail the
+ * search, by setting its result code or by throwing.
+ *
+ * @param hook what to run there, handed the search of the round, or {@code null} to run nothing
+ */
+ @VisibleForTesting
+ void setMissingChangesSearchHook(Consumer<InternalSearchOperation> hook)
+ {
+ missingChangesSearchHook = hook;
+ }
+
+ /**
+ * Sets how many local deletes this domain keeps aside for its recoveries at most.
+ *
+ * @param maxDeletes
+ * how many local deletes are kept aside at most
+ */
+ @VisibleForTesting
+ void setMaxDeletesKeptAside(int maxDeletes)
+ {
+ maxDeletesKeptAside = maxDeletes;
+ }
+
+ /**
+ * Sets what the missing change publisher runs after each of its runs, before it takes the start a
+ * session handed over or leaves.
+ *
+ * @param hook what to run there, or {@code null} to run nothing
+ */
+ @VisibleForTesting
+ void setMissingChangesPublisherRunHook(Runnable hook)
+ {
+ missingChangesPublisherRunHook = hook;
+ }
+
+ /**
+ * Sets what a round of a recovery which publishes the local changes the recovery kept runs once it
+ * has published them, before it ends the recovery.
+ *
+ * @param hook what to run there, handed whether the round holds the local changes back, or
+ * {@code null} to run nothing
+ */
+ @VisibleForTesting
+ void setKeptChangesPublishedHook(Consumer<Boolean> hook)
+ {
+ keptChangesPublishedHook = hook;
}
/**
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 231298f..378cc37 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
@@ -17,10 +17,17 @@
*/
package org.opends.server.replication.plugin;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import org.forgerock.i18n.LocalizableMessage;
+import org.forgerock.i18n.slf4j.LocalizedLogger;
+import org.forgerock.util.annotations.VisibleForTesting;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
@@ -41,6 +48,15 @@
*/
class PendingChanges
{
+ private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+
+ /**
+ * How many local changes a recovery keeps at most, and how many bytes their messages take at
+ * most: past either, it keeps none, and searches for them as for the changes done before it began.
+ */
+ static final int MAX_KEPT_CHANGES = 10000;
+ static final long MAX_KEPT_BYTES = 32L * 1024 * 1024;
+
/** A map used to store the pending changes. */
private final TreeMap<CSN, PendingChange> pendingChanges = new TreeMap<>();
@@ -56,7 +72,15 @@
/** Told that the replica of this domain announces itself offline, or takes that back. */
private final ReplicaOfflineAnnouncer replicaOfflineAnnouncer;
- private boolean recoveringOldChanges;
+ /**
+ * Changed under the lock the domain starts and ends a recovery under - and ended under the lock of
+ * this list as well - and read without either by {@link #isRecovering()}.
+ */
+ private volatile boolean recoveringOldChanges;
+ /** What the recovery which runs keeps of the local changes: {@code null} until it first needs it. */
+ 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;
/**
* Creates a new PendingChanges using the provided CSNGenerator.
@@ -199,11 +223,28 @@
}
else
{
- // do not push updates until the RS catches up.
+ // do not push updates until the RS catches up: the recovery publishes them.
+ // Kept before the state takes it: the first change kept fixes where the search of the
+ // recovery ends, at the last change pushed before it.
// @see #setRecovering(boolean)
+ recovery().keep((LDAPUpdateMsg) msg);
domain.getServerState().update(msg.getCSN());
}
}
+ else if (msg instanceof ReplicaOfflineMsg && recoveringOldChanges)
+ {
+ /*
+ * The announcement carries the newest CSN of this replica, past the changes the recovery
+ * has still to publish: the replication server would take this replica for offline from
+ * there - its changelog index past the changes which reach it afterwards. It is given up
+ * on, as one a change in flight holds back is, whatever the broker would refuse.
+ */
+ if (logger.isTraceEnabled())
+ {
+ logger.trace("The announcement that the replica goes offline at " + msg.getCSN()
+ + " is given up on: a recovery has changes before it to publish");
+ }
+ }
else if (msg instanceof ReplicaOfflineMsg)
{
/*
@@ -256,6 +297,15 @@
* changes from the pendingChanges because there are older changes that need
* to be published before.
*
+ * <p>
+ * The local changes pushed meanwhile are kept for the recovery, which publishes them once it has
+ * published the ones done before it began. A recovery which is already running goes on, with what
+ * it kept: a session which starts while it runs starts no other.
+ * <p>
+ * It is set without the lock of this list, which the start of a session must not take, but under
+ * the one the domain ends a recovery under as well: a recovery the session needs is not ended by
+ * one which was about to end.
+ *
* @param recovering
* The recovering status that must be set.
*/
@@ -265,30 +315,281 @@
}
/**
+ * Tells whether the local changes are left to a recovery, which publishes them.
+ * <p>
+ * It is read without the lock of this list, which the start of a session must not take: the
+ * session only learns from it that it needs a missing change publisher to end the recovery,
+ * and that one ends it under the lock.
+ *
+ * @return {@code true} if the local changes are left to a recovery
+ */
+ boolean isRecovering()
+ {
+ return recoveringOldChanges;
+ }
+
+ /**
+ * Tells whether the recovery which runs keeps the local changes, and where its search of the
+ * historical information ends then.
+ *
+ * @return the recovery which keeps the local changes, or {@code null} if its search has to go up
+ * to the last local change: no recovery runs, or it kept too many changes to keep any
+ */
+ synchronized KeptChanges keptChanges()
+ {
+ if (!recoveringOldChanges)
+ {
+ return null;
+ }
+ final Recovery running = recovery();
+ return running.kept != null ? new KeptChanges(running.searchUpTo) : null;
+ }
+
+ /**
+ * Returns the local changes the recovery which runs kept which come after the provided CSN, in
+ * order.
+ *
+ * @param csn
+ * where the recovery has got to
+ * @return the changes, or {@code null} if no recovery runs, or it kept too many changes to keep any
+ */
+ synchronized List<LDAPUpdateMsg> keptChangesAfter(CSN csn)
+ {
+ final KeptChanges keeping = keptChanges();
+ return keeping != null ? new ArrayList<>(recovery.kept.tailMap(csn, false).values()) : null;
+ }
+
+ /**
+ * Returns the local changes the recovery which runs kept which come after the provided CSN, in
+ * order, whatever it searches for: the data the domain read its state from again may not reflect
+ * them.
+ *
+ * @param newest
+ * the newest change of this replica the data holds, or {@code null} if it holds none
+ * @return the changes, none if no recovery keeps any
+ */
+ synchronized List<LDAPUpdateMsg> keptChangesNewerThan(CSN newest)
+ {
+ if (recovery == null || recovery.kept == null)
+ {
+ return new ArrayList<>();
+ }
+ return new ArrayList<>((newest != null ? recovery.kept.tailMap(newest, false) : recovery.kept).values());
+ }
+
+ /**
* Allows to update the recovery situation by comparing the CSN of the last
* change that was sent to the ReplicationServer with the CSN of the last
* operation that was taken out of the PendingChanges list. If the two match
* then the recovery is completed and normal procedure can restart. Otherwise
* the RSUpdater thread must continue to look for older changes and no changes
* can be committed from the pendingChanges list.
+ * <p>
+ * The recovery ends at once for every change pushed after it: a change which comes here once
+ * this list publishes again must find the broker taking it as well, or it is refused, while
+ * the domain state records it as done and no recovery is left to publish it. The recovery is
+ * therefore ended here, under the lock {@link #pushCommittedChanges()} takes, before any change
+ * can be pushed. What it kept is dropped with it.
*
* @param recovered
* The CSN of the last change that was published to the
* ReplicationServer.
+ * @param recoveryEnd
+ * Ends the recovery, unless a session which started meanwhile needs it, and has the
+ * broker take the changes again.
* @return A boolean indicating if the recovery is completed (false) or must
* continue (true).
*/
- synchronized boolean recoveryUntil(CSN recovered)
+ synchronized boolean recoveryUntil(CSN recovered, RecoveryEnd recoveryEnd)
{
final CSN lastLocalChange = domain.getLastLocalChange();
- if (recovered != null && recovered.isNewerThanOrEqualTo(lastLocalChange))
+ if (recoveringOldChanges && recovered != null && recovered.isNewerThanOrEqualTo(lastLocalChange))
{
- recoveringOldChanges = false;
+ recoveryEnd.end(() ->
+ {
+ recoveringOldChanges = false;
+ recovery = null;
+ });
}
return recoveringOldChanges;
}
/**
+ * Drops the provided local changes the recovery which runs kept: the data of the domain, read
+ * again, does not hold them - it was restored or imported since they were done - and the replicas
+ * would get changes this one does not hold.
+ *
+ * @param csns
+ * the CSNs of the changes to drop
+ */
+ synchronized void dropKeptChanges(Collection<CSN> csns)
+ {
+ if (recovery == null || recovery.kept == null || csns.isEmpty())
+ {
+ return;
+ }
+ recovery.kept.keySet().removeAll(csns);
+ recovery.keptBytes = 0;
+ for (LDAPUpdateMsg msg : recovery.kept.values())
+ {
+ recovery.keptBytes += msg.size();
+ }
+ }
+
+ /**
+ * Has the recovery which runs keep no more local changes, and drop the ones it kept: it searches
+ * for them in the historical information instead, as one which kept too many does.
+ */
+ synchronized void keepNoMoreChanges()
+ {
+ if (recovery != null)
+ {
+ recovery.kept = null;
+ }
+ }
+
+ /**
+ * Returns what the recovery which runs keeps of the local changes, under the lock of this list.
+ * It is made as it is first needed - as the recovery keeps its first change, or as its publisher
+ * asks where its search ends - so the search ends at the last change pushed before any was kept.
+ */
+ private Recovery recovery()
+ {
+ if (recovery == null)
+ {
+ recovery = new Recovery(domain.getLastLocalChange());
+ }
+ return recovery;
+ }
+
+ /**
+ * Sets how many local changes a recovery which starts from now on keeps at most: with none, it
+ * searches for all of them in the historical information.
+ *
+ * @param maxKeptChanges
+ * how many local changes a recovery keeps at most
+ */
+ @VisibleForTesting
+ synchronized void setMaxKeptChanges(int maxKeptChanges)
+ {
+ this.maxKeptChanges = maxKeptChanges;
+ }
+
+ /**
+ * What a recovery keeps of the local changes pushed while it runs.
+ * <p>
+ * A recovery republishes what the replication server misses from the historical information of
+ * the entries, and its search of it races the commits: a change committed while it searches is in
+ * the domain state, and not in what it found. The changes pushed once it began are kept instead,
+ * in order and with their messages, so that its search only has to cover the ones pushed before:
+ * up to the last of them, which no commit moves. A change kept is published as it was done, where
+ * a later change of the same values leaves nothing of it in the historical information, and a
+ * delete kept does not depend on the deletes the domain keeps aside, which are capped.
+ * <p>
+ * What it keeps stays until it ends: a missing change publisher whose broker refused a change,
+ * or whose session was replaced by one whose replication server holds less, leaves them to the
+ * one which goes on, which may start from further back. Past {@link #maxKeptChanges} changes or
+ * {@link #MAX_KEPT_BYTES} bytes it keeps none, and the recovery searches for them from where it
+ * has got to, holding the local changes back at the end, as it does when nothing is kept.
+ */
+ private final class Recovery
+ {
+ /** The last local change pushed before this recovery kept any: its search ends there. */
+ private final CSN searchUpTo;
+ /** The local changes pushed since, in order; {@code null} once there were too many. */
+ private TreeMap<CSN, LDAPUpdateMsg> kept;
+ private long keptBytes;
+
+ private Recovery(CSN searchUpTo)
+ {
+ this.searchUpTo = searchUpTo;
+ this.kept = maxKeptChanges > 0 ? new TreeMap<CSN, LDAPUpdateMsg>() : null;
+ }
+
+ private void keep(LDAPUpdateMsg msg)
+ {
+ if (kept == null)
+ {
+ return;
+ }
+ /*
+ * Published as a recovery publishes what it rebuilds: not assured. Its operation may still be
+ * waiting for the acknowledgement when a round publishes it - nothing orders the two - and it
+ * then gets none and times out, as it does when the round comes later: it waits as the domain
+ * is configured to, on the CSN of the message, and not as the message says.
+ */
+ if (msg.isAssured())
+ {
+ msg.setNotAssured();
+ }
+ kept.put(msg.getCSN(), msg);
+ keptBytes += msg.size();
+ if (kept.size() > maxKeptChanges || keptBytes > MAX_KEPT_BYTES)
+ {
+ kept = null;
+ if (logger.isDebugEnabled())
+ {
+ logger.debug(LocalizableMessage.raw("the recovery keeps no more local changes after " + msg.getCSN()
+ + ": it searches for them in the historical information"));
+ }
+ }
+ }
+ }
+
+ /** A recovery which keeps the local changes pushed while it runs. */
+ static final class KeptChanges
+ {
+ /**
+ * The last local change pushed before the recovery kept any: its search ends there, and the
+ * changes it kept come next. {@code null} if the domain state held no change of this replica.
+ */
+ final CSN searchUpTo;
+
+ private KeptChanges(CSN searchUpTo)
+ {
+ this.searchUpTo = searchUpTo;
+ }
+ }
+
+ /** Ends a recovery the pending changes have caught up with. */
+ interface RecoveryEnd
+ {
+ /**
+ * Ends the recovery, unless a session which started meanwhile has given it changes of its own
+ * to recover: runs {@code recovered} under the lock a recovery is started under, and has the
+ * broker take the changes again. Called under the lock of the pending changes.
+ *
+ * @param recovered
+ * lets the pending changes publish the local changes again
+ */
+ void end(Runnable recovered);
+ }
+
+ /**
+ * Runs the provided rounds of a recovery while no local change can be numbered or pushed.
+ * <p>
+ * A round publishes the changes done when it began, and a change done while it runs is left
+ * to a round after it, so a recovery could chase the local changes for as long as they go on.
+ * The rounds run here are the last ones: what they publish is all there is, and nothing can be
+ * pushed between their end and the end of the recovery they decide. The local operations wait
+ * for them - for their CSN and for their push - for as long as publishing the changes done while
+ * the round before ran takes, where a publish in normal operation holds the same lock for one
+ * change.
+ *
+ * @param <T>
+ * what the rounds return
+ * @param round
+ * the rounds of the recovery
+ * @return what the round returned
+ * @throws Exception
+ * if the round failed
+ */
+ synchronized <T> T holdBackLocalChangesWhile(Callable<T> round) throws Exception
+ {
+ return round.call();
+ }
+
+ /**
* Told that the replica of this domain announces itself offline, or takes that back.
* <p>
* A collocated replication server can forward a {@link ReplicaOfflineMsg} as soon as it is on
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/LDAPUpdateMsg.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/LDAPUpdateMsg.java
index d2c30a5..c6a6585 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/LDAPUpdateMsg.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/LDAPUpdateMsg.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2009 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.protocol;
@@ -202,6 +203,20 @@
bytes = getBytes(ProtocolVersion.getCurrentVersion());
}
+ /**
+ * Has this message published as one no acknowledgement is waited for: it is no longer assured,
+ * and is encoded again. A recovery publishes the local changes it kept so, as it publishes the
+ * changes it rebuilds from the historical information: an operation which still waits for the
+ * acknowledgement of one gets none, and times out.
+ *
+ * This method is not synchronized and therefore not MT safe.
+ */
+ public void setNotAssured()
+ {
+ setAssured(false);
+ bytes = null;
+ }
+
/** {@inheritDoc} */
@Override
public ByteArrayBuilder encodeHeader(byte msgType, short protocolVersion)
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 91c68af..af26f3e 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
@@ -2259,7 +2259,7 @@
*/
public void publish(ReplicationMsg msg)
{
- publish(msg, false, true);
+ publish(msg, false, true, null);
}
/**
@@ -2270,16 +2270,40 @@
*/
boolean publish(ReplicationMsg msg, boolean retryOnFailure)
{
- return publish(msg, false, retryOnFailure);
+ return publish(msg, false, retryOnFailure, null);
}
/**
- * Publish a recovery message to the other servers.
- * @param msg the message to publish
+ * Returns the session this broker publishes on, which a recovery started for it publishes on:
+ * see {@link #publishRecovery(ReplicationMsg, Session)}.
+ *
+ * @return the session this broker publishes on, or {@code null} if it has none
*/
- public void publishRecovery(ReplicationMsg msg)
+ public Session getSession()
{
- publish(msg, true, true);
+ return connectedRS.get().session;
+ }
+
+ /**
+ * Publish a recovery message to the other servers, on the provided session only.
+ * <p>
+ * A recovery publishes what the replication server of its session misses, and that one only: the
+ * replication server of a session which replaced it may miss older changes, and would take none
+ * of them once it had a newer one. The message is refused rather than published on another
+ * session.
+ *
+ * @param msg the message to publish
+ * @param session the session the recovery publishes on, as {@link #getSession()} returned it
+ * when the recovery was started for it
+ * @return {@code false} if the broker refused the message: no replication
+ * server was reachable, the session is not the one the broker publishes
+ * on anymore, or the broker was stopped; {@code true} once the session
+ * took it - which still drops it if it is closing, as it does any
+ * message the broker publishes.
+ */
+ public boolean publishRecovery(ReplicationMsg msg, Session session)
+ {
+ return publish(msg, true, true, session);
}
/**
@@ -2287,10 +2311,11 @@
* @param msg the message to publish
* @param recoveryMsg the message is a recovery LocalizableMessage
* @param retryOnFailure whether retry should be done on failure
+ * @param recoverySession the only session a recovery message may be published on
* @return whether the message was successfully sent.
*/
private boolean publish(ReplicationMsg msg, boolean recoveryMsg,
- boolean retryOnFailure)
+ boolean retryOnFailure, Session recoverySession)
{
boolean done = false;
@@ -2338,8 +2363,23 @@
recover some changes then it is not allowed to send this
change but it will be the responsibility of the recovery thread to
do it.
+ The messages which are no change of this replica - acknowledgements,
+ total updates, monitoring requests, the reset of the generation ID -
+ do not wait for the recovery (issue #1220): only a change, or the
+ announcement that the replica goes offline, which carries its newest
+ CSN, would get to the replication server ahead of the changes it misses.
*/
- if (!recoveryMsg & connectRequiresRecovery)
+ if (!recoveryMsg && connectRequiresRecovery && msg instanceof UpdateMsg)
+ {
+ return false;
+ }
+ /*
+ A recovery message is for the session the recovery was started for:
+ the one which replaced it has a recovery of its own, from what its
+ replication server holds. The write below takes the message only on
+ the session read here.
+ */
+ if (recoveryMsg && (recoverySession == null || currentSession != recoverySession))
{
return false;
}
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 2842041..5b62db3 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -719,3 +719,9 @@
in domain "%s" is held: a total update is being processed over that session, and stopping it \
would end the total update. The restart runs once the total update is over, and the change is \
sent again then; until then it is not recorded as replayed
+ERR_MISSING_CHANGES_NOT_PUBLISHED_333=Domain "%s" could not publish the changes its replication \
+ server misses: %s. It tries again within %d ms - on the session which starts meanwhile, or on one it \
+ starts then - and the changes done meanwhile are left to that recovery as well
+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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/common/CSNGeneratorTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/common/CSNGeneratorTest.java
index bd5b08b..981e218 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/common/CSNGeneratorTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/common/CSNGeneratorTest.java
@@ -13,6 +13,7 @@
*
* Copyright 2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.common;
@@ -57,6 +58,46 @@
}
/**
+ * An adjust with no CSN - a domain reads back a ServerState which holds no change of its own -
+ * leaves the generator where it is: moved back to the clock, it numbered the next change as one it
+ * had numbered already, or before it.
+ */
+ @Test
+ public void adjustWithNoCSNDoesNotNumberBeforeTheCSNsGenerated()
+ {
+ // Ahead of the clock, as a generator is once it numbered changes within the current tick of the
+ // clock, or followed the CSNs of a server whose clock runs ahead.
+ CSNGenerator generator = new CSNGenerator(5, TimeThread.getTime() + 3600 * 1000);
+ CSN csn1 = generator.newCSN();
+
+ generator.adjust((CSN) null);
+
+ CSN csn2 = generator.newCSN();
+ assertTrue(csn2.compareTo(csn1) > 0,
+ "CSNGenerator generated an earlier CSN after an adjust with no CSN.");
+ }
+
+ /**
+ * An adjust with no CSN does not take the generator back past the CSNs received from the other
+ * servers either: a local change numbered before a change of another server it follows would lose
+ * the conflict resolution to it on the other servers.
+ */
+ @Test
+ public void adjustWithNoCSNDoesNotNumberBeforeTheCSNsReceived()
+ {
+ CSNGenerator generator = new CSNGenerator(5, TimeThread.getTime());
+ // From a server whose clock runs an hour ahead.
+ CSN received = new CSN(TimeThread.getTime() + 3600 * 1000, 0, 6);
+ generator.adjust(received);
+
+ generator.adjust((CSN) null);
+
+ CSN csn = generator.newCSN();
+ assertTrue(csn.compareTo(received) > 0,
+ "CSNGenerator generated a CSN earlier than one it received after an adjust with no CSN.");
+ }
+
+ /**
* Test the correct behavior of the CSNGenerator when
* the seqnum is rolling over its limit.
*/
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 dbc0bab..9a44067 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
@@ -13,13 +13,26 @@
*
* Copyright 2008-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
@@ -29,24 +42,42 @@
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.ConditionResult;
import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.ModificationType;
+import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.ldap.schema.CoreSchema;
import org.forgerock.opendj.ldap.schema.MatchingRule;
import org.forgerock.opendj.ldap.schema.SchemaBuilder;
import org.opends.server.TestCaseUtils;
+import org.opends.server.backends.MemoryBackend;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType;
import org.opends.server.core.DirectoryServer;
+import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.protocol.*;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.replication.server.changelog.api.DBCursor;
+import org.opends.server.replication.server.changelog.api.DBCursor.CursorOptions;
import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.replication.service.ReplicationDomain;
import org.forgerock.opendj.ldap.schema.AttributeType;
import org.opends.server.types.*;
+import org.opends.server.util.StaticUtils;
+import org.opends.server.util.TestTimer;
import org.opends.server.util.TimeThread;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static java.util.concurrent.TimeUnit.*;
+import static org.assertj.core.api.Assertions.assertThat;
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.*;
+import static org.opends.server.util.CollectionUtils.newArrayList;
import static org.testng.Assert.*;
/**
@@ -60,27 +91,70 @@
private final int serverId = 123;
+ /** Another replica, whose changes the replica under test replays. */
+ private static final int OTHER_SERVER_ID = 124;
+ /**
+ * How many rounds the missing change publisher is given to end a recovery while local changes
+ * keep being done: one round which sees changes done while it ran, and the last one, which holds
+ * them back.
+ */
+ private static final int MAX_ROUNDS_UNDER_LOAD = 20;
+ /** How much time of changes a round of the missing change publisher searches at most. */
+ private static final long SEARCH_WINDOW_MS = 10000;
+ /** The server ID of the replication server of the tests, and of the one some of them add. */
+ private static final int FIRST_RS_ID = 1;
+ private static final int SECOND_RS_ID = 2;
private final SortedSet<String> replServers = new TreeSet<>();
+ private int rsPort;
private static class TestBroker extends ReplicationBroker
{
private final List<ReplicationMsg> list;
+ /** Whether this broker writes the messages to a session, or refuses them as one without any. */
+ private final boolean writes;
private TestBroker(List<ReplicationMsg> list)
{
+ this(list, true);
+ }
+
+ private TestBroker(List<ReplicationMsg> list, boolean writes)
+ {
super(new DummyReplicationDomain(0), null,
new DomainFakeCfg(null, 0, new TreeSet<String>()), null);
this.list = list;
+ this.writes = writes;
}
@Override
- public void publishRecovery(ReplicationMsg msg)
+ public boolean publishRecovery(ReplicationMsg msg, Session session)
{
- list.add(msg);
+ if (writes)
+ {
+ list.add(msg);
+ }
+ return writes;
}
}
+ /** Runs a test as the recovery keeps the local changes done while it runs, and as it keeps none. */
+ @DataProvider
+ public Object[][] keepingTheLocalChanges()
+ {
+ return new Object[][] { { true }, { 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.
+ */
+ @AfterMethod
+ public void forgetReplicationServers()
+ {
+ replServers.clear();
+ }
+
private MatchingRule getRule()
{
String oid = "1.3.6.1.4.1.26027.1.4.4";
@@ -321,13 +395,2555 @@
}
}
+ /**
+ * A change done while a round of the missing change publisher searches is one its search did not
+ * see. The round may not publish a later change instead: the next round searches from the last
+ * change published, so the change in between would never be published by anyone. Only a
+ * delete, which is kept aside as it is done, escaped this. A recovery which keeps the local
+ * changes publishes it as it kept it, and one which keeps none finds it in a later round.
+ */
+ @Test(dataProvider = "keepingTheLocalChanges")
+ public void aChangeDoneWhileARoundSearchesIsPublished(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 DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "before the recovery");
+ done.add(rd1.getLastLocalChange());
+ // Started after the change, which the recovery then rebuilds from the historical information.
+ startRecovery(rd1, keeps);
+ rd1.setMissingChangesSearchHook(once(() ->
+ {
+ replace(dnB, "description", "while the first round searches");
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(dnC);
+ done.add(rd1.getLastLocalChange());
+ }));
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A round which has found nothing to publish may not end the recovery past the last local
+ * change: a change done while it searched is left to the recovery all the same, and nothing else
+ * would publish it. Only a recovery which searches for the local changes done while it runs - one
+ * which kept too many of them - has such a round.
+ */
+ @Test
+ public void aRoundWhichFoundNothingDoesNotEndTheRecoveryPastAChange() 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();
+
+ startRecovery(rd1, false);
+ final List<CSN> done = new ArrayList<>();
+ rd1.setMissingChangesSearchHook(once(() ->
+ {
+ replace(dnA, "description", "while the first round searches");
+ done.add(rd1.getLastLocalChange());
+ }));
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * Once the missing change publisher has ended the recovery, a local change is published to the
+ * replication server again: the broker, which refuses the changes left to the recovery, takes
+ * them back as the pending changes stop leaving them to it, and not some time later.
+ */
+ @Test
+ public void aChangeDoneOnceTheRecoveryEndedIsPublished() 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();
+ startRecovery(rd1);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(published).isEmpty();
+
+ replace(dnA, "description", "after the recovery");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The broker takes the local changes again under the lock they are pushed under, as the recovery
+ * ends. Taken after it, a change pushed in between - which the pending changes publish again -
+ * is refused, while the domain state records it as done and no recovery is left to publish it.
+ */
+ @Test
+ public void theBrokerTakesChangesAgainUnderTheLockTheyArePushedUnder() 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);
+
+ final PendingChanges pendingChanges = pendingChangesOf(rd1);
+ final ReplicationBroker realBroker = brokerOf(rd1);
+ final AtomicBoolean clearedUnderTheLock = new AtomicBoolean();
+ setBroker(rd1, new TestBroker(new ArrayList<ReplicationMsg>())
+ {
+ @Override
+ public void setRecoveryRequired(boolean required)
+ {
+ if (!required)
+ {
+ clearedUnderTheLock.set(Thread.holdsLock(pendingChanges));
+ }
+ }
+ });
+ try
+ {
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(new ArrayList<ReplicationMsg>())));
+ }
+ finally
+ {
+ setBroker(rd1, realBroker);
+ realBroker.setRecoveryRequired(false);
+ }
+ assertTrue(clearedUnderTheLock.get(), "the broker takes the changes again under the lock they are pushed under");
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The missing change publisher publishes the changes of its own replica only. A change of
+ * another replica, which this one replayed, is in the historical information of an entry the
+ * search finds all the same; the replication servers have it from that replica, and one they had
+ * from here first would make them drop the older changes of that replica which reach them after
+ * it.
+ */
+ @Test
+ public void onlyTheChangesOfThisReplicaArePublished() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ ReplicationBroker other = null;
+ try
+ {
+ final DN dnA = addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ other = openReplicationSession(baseDN, OTHER_SERVER_ID, 100, rsPort, 1000);
+ final CSN otherCSN = new CSNGenerator(OTHER_SERVER_ID, TimeThread.getTime()).newCSN();
+ publishOtherReplicaChange(other, otherCSN, dnA, "from the other replica");
+ waitUntilReplayed(rd1, otherCSN);
+
+ replace(dnA, "displayName", "from this replica");
+ final CSN localCSN = rd1.getLastLocalChange();
+ assertTrue(otherCSN.isOlderThan(localCSN), "the change of the other replica comes first");
+ // Started after the change, which the recovery then rebuilds from the historical information.
+ startRecovery(rd1);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactly(localCSN);
+ }
+ finally
+ {
+ stop(other);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * Local changes go on while a recovery which searches for them runs - one which kept too many of
+ * them. The recovery still ends - its last round holds them back - and each change is published
+ * either by the recovery or, once it has ended, by the broker: none is left to a recovery which is
+ * over.
+ */
+ @Test
+ public void aRecoveryEndsWhileLocalChangesGoOnAndLosesNone() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ final ExecutorService writer = Executors.newSingleThreadExecutor();
+ try
+ {
+ final List<DN> dns = new ArrayList<>();
+ for (int i = 0; i <= MAX_ROUNDS_UNDER_LOAD; i++)
+ {
+ dns.add(addPerson("load" + i));
+ }
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ startRecovery(rd1, false);
+ final List<CSN> done = new CopyOnWriteArrayList<>();
+ final List<Future<?>> changes = new CopyOnWriteArrayList<>();
+ final AtomicInteger rounds = new AtomicInteger();
+ rd1.setMissingChangesSearchHook(search ->
+ {
+ final int round = rounds.incrementAndGet();
+ if (round > MAX_ROUNDS_UNDER_LOAD)
+ {
+ throw new AssertionError("the recovery is still running after " + MAX_ROUNDS_UNDER_LOAD + " rounds");
+ }
+ doLocalChange(writer, changes, done, rd1, dns.get(round), round);
+ });
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ rd1.setMissingChangesSearchHook(null);
+ assertEachChangePublishedOnce(rs, baseDN, rsMaxCSN, published, changes, done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ writer.shutdownNow();
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A change done in the last millisecond of a search window is published. The end of a window is
+ * compared with the historical information twice: by the index of ds-sync-hist, as text, and by
+ * the round, as a CSN. A sequence number which reads as the greatest in one and as the least in
+ * the other left such a change out of its window, and out of the next one, which starts after it.
+ */
+ @Test
+ public void aChangeInTheLastMillisecondOfASearchWindowIsPublished() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ ReplicationBroker other = null;
+ try
+ {
+ // Opened first: the change below has to be done less than a window after the ones before.
+ other = openReplicationSession(baseDN, OTHER_SERVER_ID, 100, rsPort, 1000);
+ final DN dnA = addPerson("a");
+ final DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ final long endOfFirstWindow = rsMaxCSN.getTime() + SEARCH_WINDOW_MS;
+
+ // A replica numbers its changes after the newest one it replayed: a change done right after
+ // one of the other replica from the millisecond before falls in the last one of the window.
+ final CSN otherCSN = new CSN(endOfFirstWindow - 1, 1, OTHER_SERVER_ID);
+ publishOtherReplicaChange(other, otherCSN, dnB, "a millisecond before the end of the window");
+ waitUntilReplayed(rd1, otherCSN);
+
+ replace(dnA, "description", "in the last millisecond of the window");
+ final CSN localCSN = rd1.getLastLocalChange();
+ assertEquals(localCSN.getTime(), endOfFirstWindow, "the change is in the last millisecond of the window");
+ // Started after the change, which the recovery then rebuilds from the historical information.
+ startRecovery(rd1);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactly(localCSN);
+ }
+ finally
+ {
+ stop(other);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * Once the recovery holds the local changes back, it ends there, however far the changes done
+ * meanwhile ran ahead. A held-back round which stopped at the end of its search window and let
+ * the local changes go again left the next round as far behind as the one before it, and the
+ * recovery went on for as long as the local changes did. Only a recovery which searches for the
+ * local changes done while it runs - one which kept too many of them - holds them back for rounds.
+ */
+ @Test
+ public void aRecoveryEndsOnceItHoldsTheLocalChangesBackHoweverFarTheyRanAhead() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ ReplicationBroker other = null;
+ final ExecutorService writer = Executors.newSingleThreadExecutor();
+ try
+ {
+ final DN dnOther = addPerson("other");
+ final List<DN> dns = new ArrayList<>();
+ for (int i = 0; i <= MAX_ROUNDS_UNDER_LOAD; i++)
+ {
+ dns.add(addPerson("load" + i));
+ }
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ final ReplicationBroker otherReplica = other =
+ openReplicationSession(baseDN, OTHER_SERVER_ID, 100, rsPort, 1000);
+
+ startRecovery(rd1, false);
+ final List<CSN> done = new CopyOnWriteArrayList<>();
+ final List<Future<?>> changes = new CopyOnWriteArrayList<>();
+ final AtomicInteger rounds = new AtomicInteger();
+ rd1.setMissingChangesSearchHook(search ->
+ {
+ final int round = rounds.incrementAndGet();
+ if (round > MAX_ROUNDS_UNDER_LOAD)
+ {
+ throw new AssertionError("the recovery is still running after " + MAX_ROUNDS_UNDER_LOAD + " rounds");
+ }
+ // The local change done after this change of the other replica is numbered after it, further
+ // ahead of the last local change than a search window reaches.
+ final CSN otherCSN = new CSN(
+ rd1.getLastLocalChange().getTime() + SEARCH_WINDOW_MS + 1000, round, OTHER_SERVER_ID);
+ unchecked(() ->
+ {
+ publishOtherReplicaChange(otherReplica, otherCSN, dnOther, "round " + round);
+ waitUntilReplayed(rd1, otherCSN);
+ }).run();
+ doLocalChange(writer, changes, done, rd1, dns.get(round), round);
+ });
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ rd1.setMissingChangesSearchHook(null);
+ assertEachChangePublishedOnce(rs, baseDN, rsMaxCSN, published, changes, done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ writer.shutdownNow();
+ stop(other);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A round counts as published only what the broker did not refuse. A round which counted the
+ * changes refused by a broker which reached no replication server could end the recovery without
+ * 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.
+ */
+ @Test(dataProvider = "keepingTheLocalChanges")
+ public void changesTheBrokerRefusedAreLeftToTheNextRecovery(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 DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ startRecovery(rd1, keeps);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnB);
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "while no session takes it");
+ done.add(rd1.getLastLocalChange());
+
+ assertFalse(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(new ArrayList<ReplicationMsg>(), false)),
+ "a recovery whose changes the broker refused has failed");
+
+ 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 session whose replication server misses no change of this replica ends all the same the
+ * recovery a refused publisher left. Here the replication server had the change the broker
+ * refused already - back on the one which had everything, for one - and no recovery of the
+ * session ended the one left: every local change was recorded in the domain state, and
+ * published by nobody.
+ */
+ @Test
+ public void aSessionWhichNeedsNoRecoveryEndsTheOneARefusedPublisherLeft() 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 beforeB = rd1.getLastLocalChange();
+ addPerson("b");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+
+ startRecovery(rd1);
+ assertFalse(rd1.buildAndPublishMissingChanges(beforeB, new TestBroker(new ArrayList<ReplicationMsg>(), false)),
+ "a recovery whose change the broker refused has failed");
+ // as the missing change publisher does once it failed
+ brokerOf(rd1).setRecoveryRequired(false);
+
+ rd1.disable();
+ rd1.enable();
+ // The broker starts its session before enable() returns: a change done before would have
+ // been ahead of the replication server, and the session would have recovered it.
+ assertTrue(rd1.isConnected(), "the session is up");
+
+ replace(dnA, "description", "in a session which needs no recovery");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /** A stopped broker refuses a recovery message: the recovery has to learn it wrote nothing. */
+ @Test
+ public void aStoppedBrokerRefusesARecoveryMessage() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ ReplicationBroker broker = null;
+ try
+ {
+ broker = openReplicationSession(baseDN, OTHER_SERVER_ID, 100, rsPort, 1000);
+ final Session session = broker.getSession();
+ broker.stop();
+ assertFalse(broker.publishRecovery(new WindowProbeMsg(), session), "a stopped broker writes nothing");
+ }
+ finally
+ {
+ stop(broker);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A broker refuses a recovery message for a session it no longer publishes on. The recovery of
+ * that session publishes what its replication server misses, and the replication server of the
+ * session which replaced it may miss older changes: one which got a later change first would take
+ * none of them.
+ */
+ @Test
+ public void aBrokerRefusesARecoveryMessageForASessionItReplaced() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ ReplicationBroker broker = null;
+ try
+ {
+ broker = openReplicationSession(baseDN, OTHER_SERVER_ID, 100, rsPort, 1000);
+ final Session replaced = broker.getSession();
+ broker.stop();
+ broker.start();
+ final Session session = broker.getSession();
+ assertNotNull(session, "the broker has a session again");
+ assertNotSame(session, replaced);
+ assertFalse(broker.publishRecovery(new WindowProbeMsg(), replaced), "a recovery message for a replaced session");
+ assertTrue(broker.publishRecovery(new WindowProbeMsg(), session), "a recovery message for the session");
+ }
+ finally
+ {
+ stop(broker);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A round whose search did not go through publishes nothing. What the search found may be some
+ * of the changes only: a replication server which took them would take none of the ones it
+ * missed, which are older, and no later recovery could publish those. The next recovery publishes
+ * them all, in order, and the delete kept aside with them.
+ */
+ @Test
+ public void aRoundWhoseSearchFailedPublishesNothing() 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 DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "before the search fails");
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(dnB);
+ done.add(rd1.getLastLocalChange());
+ replace(dnC, "description", "before the search fails");
+ done.add(rd1.getLastLocalChange());
+ // Started after the changes, which the recovery then rebuilds from the historical information.
+ startRecovery(rd1);
+
+ rd1.setMissingChangesSearchHook(onTheFirstSearch(search -> search.setResultCode(ResultCode.UNAVAILABLE)));
+ final List<ReplicationMsg> published = new ArrayList<>();
+ Assertions.assertThatThrownBy(() -> rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)))
+ .as("a recovery whose search failed")
+ .hasMessageContaining(ResultCode.UNAVAILABLE.getName().toString());
+ assertThat(published).as("what a round whose search failed published").isEmpty();
+
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The deletes a round was to publish and did not, as it threw on the way, stay where they were:
+ * they have nothing else to be published from, and the recovery of the session started again
+ * publishes them.
+ */
+ @Test(dataProvider = "keepingTheLocalChanges")
+ public void theDeletesARoundThrewBeforePublishingAreLeftToTheNextRecovery(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 DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ startRecovery(rd1, keeps);
+ TestCaseUtils.deleteEntry(dnA);
+ final CSN first = rd1.getLastLocalChange();
+ TestCaseUtils.deleteEntry(dnB);
+ final CSN second = rd1.getLastLocalChange();
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ final ReplicationBroker throwingOnTheSecond = new TestBroker(published)
+ {
+ @Override
+ public boolean publishRecovery(ReplicationMsg msg, Session session)
+ {
+ if (!published.isEmpty())
+ {
+ throw new IllegalStateException("the publish threw, as a test asked");
+ }
+ return super.publishRecovery(msg, session);
+ }
+ };
+ Assertions.assertThatThrownBy(() -> rd1.buildAndPublishMissingChanges(rsMaxCSN, throwingOnTheSecond))
+ .isInstanceOf(IllegalStateException.class);
+ assertThat(csns(published)).containsExactly(first);
+
+ // The replication server of the session started again holds the delete which was published.
+ final List<ReplicationMsg> republished = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(first, new TestBroker(republished)));
+ assertThat(csns(republished)).containsExactly(second);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A recovery whose search fails on a session which stays up has the session started again, and
+ * the recovery of the next session publishes the changes, in order. No next session came: the
+ * local changes were left to a recovery nobody ran - recorded in the domain state and published by
+ * nobody - until the session went down for some other reason.
+ */
+ @Test
+ public void aRecoveryWhoseSearchFailedHasTheSessionStartedAgain() throws Exception
+ {
+ assertARecoveryWhichFailedHasTheSessionStartedAgain(search -> search.setResultCode(ResultCode.UNAVAILABLE));
+ }
+
+ /** A recovery which threw on a session which stays up has the session started again as well. */
+ @Test
+ public void aRecoveryWhichThrewHasTheSessionStartedAgain() throws Exception
+ {
+ assertARecoveryWhichFailedHasTheSessionStartedAgain(search ->
+ {
+ throw new IllegalStateException("the round threw, as a test asked");
+ });
+ }
+
+ private void assertARecoveryWhichFailedHasTheSessionStartedAgain(Consumer<InternalSearchOperation> failure)
+ 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 DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+
+ // The replication server misses the changes done from here on.
+ refuseTheLocalChanges(rd1);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "missed by the replication server");
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(dnB);
+ done.add(rd1.getLastLocalChange());
+
+ // The session started again recovers them, and the first round of its recovery fails, with a
+ // change done meanwhile: the recovery keeps it, and the next one publishes it.
+ rd1.setMissingChangesSearchHook(onTheFirstSearch(search ->
+ {
+ unchecked(() ->
+ {
+ replace(dnA, "displayName", "while the failing round searches");
+ done.add(rd1.getLastLocalChange());
+ }).run();
+ failure.accept(search);
+ }));
+ rd1.disable();
+ rd1.enable();
+
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(rs, baseDN, rsMaxCSN))
+ .as("the changes the replication server holds after the ones it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+ replace(dnA, "description", "once the recovery is over");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The search of a round which finds no base entry went through: the suffix is gone, and no entry
+ * is left below it to hold a change. The deletes which removed it go out, the recovery ends, and
+ * the base entry added back is published as any other change. A recovery which took that search
+ * for a failed one would wait for the suffix to come back.
+ */
+ @Test
+ public void theDeletesOfASuffixRemovedDuringARecoveryArePublished() 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<>();
+ TestCaseUtils.deleteEntry(dnA);
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(baseDN);
+ done.add(rd1.getLastLocalChange());
+ // Started after the deletes, which the recovery then searches for: the deletes a recovery keeps
+ // as they are done need no search.
+ startRecovery(rd1);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)),
+ "a recovery whose search found no base entry is over");
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+
+ TestCaseUtils.addEntry(StaticUtils.createEntry(baseDN));
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The deletes kept aside outlive a session whose replication server holds no change of this
+ * replica. Its missing change publisher starts from the newest change of this replica and has
+ * nothing older to publish to it, but a later session may reach a replication server which misses
+ * some of those deletes, and they have nothing else to be published from. They were dropped -
+ * trimmed as not newer than where that publisher started, and cleared as its recovery ended - and
+ * the replication server of the later session never had the delete.
+ */
+ @Test
+ public void theDeletesKeptAsideOutliveASessionToAReplicationServerWithNoChangeOfThisReplica()
+ throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ final SortedSet<String> towardsRs = new TreeSet<>(replServers);
+ ReplicationServer empty = null;
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ addPerson("a");
+ final DN dnB = addPerson("b");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+
+ // As a publisher whose broker refused a change leaves it.
+ startRecovery(rd1);
+ brokerOf(rd1).setRecoveryRequired(false);
+ TestCaseUtils.deleteEntry(dnB);
+ final CSN delete = rd1.getLastLocalChange();
+
+ // A session to a replication server which holds no change of this replica, and knows of no
+ // other replication server: the publisher of the session ends the recovery.
+ final int emptyPort = TestCaseUtils.findFreePort();
+ empty = createIsolatedReplicationServer(emptyPort, "HistoricalCsnOrderingEmpty", SECOND_RS_ID);
+ connectTo(rd1, towards(emptyPort), SECOND_RS_ID);
+ waitUntilTheRecoveryIsOver(rd1);
+ assertNull(empty.getReplicationServerDomain(baseDN, true).getLatestServerState().getCSN(serverId),
+ "the replication server holding no change of this replica was sent none");
+
+ // A session to the replication server which misses the delete.
+ assertEquals(rd1.applyConfigurationChange(domainConfiguration(serverId, towardsRs)).getResultCode(),
+ ResultCode.SUCCESS);
+ assertInTheChangelog(rs, baseDN, delete);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, empty);
+ }
+ }
+
+ /**
+ * A session whose replication server holds no change of this replica ends the recovery a failed
+ * publisher left, on a replica whose own domain state holds none of its changes either: its
+ * publisher starts from the current time.
+ */
+ @Test
+ public void aRecoveryLeftOnAReplicaWithNoChangeOfItsOwnEnds() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ assertNull(rd1.getServerState().getCSN(serverId), "the domain state holds no change of this replica");
+ // As a publisher whose broker refused a change leaves it.
+ startRecovery(rd1);
+ brokerOf(rd1).setRecoveryRequired(false);
+
+ rd1.disable();
+ rd1.enable();
+ waitUntilTheRecoveryIsOver(rd1);
+
+ addPerson("a");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A change numbered exactly the end of a search window is published. The search finds it - its
+ * filter takes the end - but a round which took only the changes older than the end left it out,
+ * and the next window starts after it.
+ */
+ @Test
+ public void aChangeNumberedTheEndOfASearchWindowIsPublished() 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 CSN endOfFirstWindow =
+ new CSN(rsMaxCSN.getTime() + SEARCH_WINDOW_MS, Integer.MAX_VALUE, serverId);
+
+ // The next change is numbered right after this one, in the same millisecond: the clock has not
+ // reached it yet, the change being done less than a window after the one before.
+ rd1.getGenerator().adjust(new CSN(endOfFirstWindow.getTime() - 1, Integer.MAX_VALUE - 1, serverId));
+ replace(dnA, "description", "numbered the end of the window");
+ assertEquals(rd1.getLastLocalChange(), endOfFirstWindow, "the change is numbered the end of the window");
+ // Started after the change, which the recovery then rebuilds from the historical information.
+ startRecovery(rd1);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactly(endOfFirstWindow);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A local change done during a recovery is published as it was done, even once a later change has
+ * replaced its values: the historical information of the entry keeps the later one only, so a
+ * recovery which rebuilt the local changes from it published the later change alone, and the other
+ * replicas never had the first - the changelog had a gap in the sequence numbers of this replica
+ * where it was.
+ */
+ @Test
+ public void aChangeWhoseValuesALaterOneReplacedIsPublished() 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();
+
+ startRecovery(rd1);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ 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
+ * waited for it.
+ */
+ @Test
+ public void noSearchRunsWhileTheLocalChangesAreHeldBack() 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 DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "before the recovery");
+ done.add(rd1.getLastLocalChange());
+ // Started after the change, which the recovery then rebuilds from the historical information.
+ startRecovery(rd1);
+ final PendingChanges pendingChanges = pendingChangesOf(rd1);
+ final AtomicInteger searches = new AtomicInteger();
+ final AtomicInteger heldSearches = new AtomicInteger();
+ rd1.setMissingChangesSearchHook(search ->
+ {
+ if (Thread.holdsLock(pendingChanges))
+ {
+ heldSearches.incrementAndGet();
+ }
+ if (searches.incrementAndGet() == 1)
+ {
+ unchecked(() ->
+ {
+ replace(dnB, "description", "while the first round searches");
+ done.add(rd1.getLastLocalChange());
+ }).run();
+ }
+ });
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ assertThat(heldSearches.get()).as("the searches run while the local changes are held back").isZero();
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ @Test
+ public void aRecoveryWhichKeptTooManyChangesSearchesForThemAndLosesNone() 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 DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ pendingChangesOf(rd1).setMaxKeptChanges(2);
+ startRecovery(rd1);
+ // Each change of other values than the ones before, which the historical information keeps.
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "kept");
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(dnB);
+ done.add(rd1.getLastLocalChange());
+ replace(dnC, "description", "one too many to keep");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "displayName", "searched for");
+ done.add(rd1.getLastLocalChange());
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ @Test
+ public void aRecoveryWhichKeepsTheLocalChangesEndsWhileTheyGoOnAndLosesNone() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ final ExecutorService writer = Executors.newSingleThreadExecutor();
+ try
+ {
+ final List<DN> dns = new ArrayList<>();
+ for (int i = 0; i < 4 * MAX_ROUNDS_UNDER_LOAD; i++)
+ {
+ dns.add(addPerson("load" + i));
+ }
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ startRecovery(rd1);
+ final List<CSN> done = new CopyOnWriteArrayList<>();
+ final Future<?> changes = writer.submit(unchecked(() ->
+ {
+ for (int i = 0; i < dns.size(); i++)
+ {
+ replace(dns.get(i), "description", "change " + i);
+ done.add(rd1.getLastLocalChange());
+ }
+ }));
+ // Some changes are kept before the recovery begins to publish, and the others are done while it
+ // runs: it publishes slowly enough for them to.
+ new TestTimer.Builder()
+ .maxSleep(10, SECONDS)
+ .sleepTimes(10, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(done.size()).isGreaterThanOrEqualTo(MAX_ROUNDS_UNDER_LOAD / 4));
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)
+ {
+ @Override
+ public boolean publishRecovery(ReplicationMsg msg, Session session)
+ {
+ unchecked(() -> Thread.sleep(5)).run();
+ return super.publishRecovery(msg, session);
+ }
+ }));
+ assertEachChangePublishedOnce(rs, baseDN, rsMaxCSN, published, Collections.<Future<?>> singletonList(changes), done);
+ }
+ finally
+ {
+ writer.shutdownNow();
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A session which starts while the missing change publisher of the previous one runs has its
+ * replication server get every change of this replica it misses, in order. The publisher went on
+ * from where it had got to, on the new session, and the replication server of the new one - which
+ * held less than the one before - got the later changes first, and took none of the older ones it
+ * missed afterwards (issue #1199).
+ */
+ @Test
+ public void aSessionWhichStartsWhileARecoveryRunsGetsEveryChangeItsReplicationServerMisses()
+ 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);
+ final CountDownLatch held = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ try
+ {
+ second = createIsolatedReplicationServer(secondPort, "HistoricalCsnOrderingSecond", SECOND_RS_ID);
+ final DN dnA = addPerson("a");
+ final DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+ final CSN onBoth = aChangeBothReplicationServersHold(rd1, dnA, rs, towardsRs, second, towardsSecond);
+
+ // The replication servers miss the changes done from here on: the first one in the first search
+ // window after the change they hold, the second one in the next. The first one is a delete,
+ // which the first recovery publishes and the second one has to find kept aside still.
+ refuseTheLocalChanges(rd1);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnC);
+ done.add(rd1.getLastLocalChange());
+ assertTrue(done.get(0).getTime() <= onBoth.getTime() + SEARCH_WINDOW_MS, "the first change is in the first window");
+ rd1.getGenerator().adjust(new CSN(onBoth.getTime() + SEARCH_WINDOW_MS + 1000, 0, serverId));
+ replace(dnB, "description", "in the second window");
+ done.add(rd1.getLastLocalChange());
+
+ // The session started again recovers them, and its publisher is held on the search of its second
+ // round, once the first round has published the first change.
+ rd1.setMissingChangesSearchHook(onTheSearch(2, search -> holdUntil(held, release).run()));
+ rd1.disable();
+ rd1.enable();
+ assertTrue(held.await(30, SECONDS), "the second round of the recovery searches");
+ assertInTheChangelog(rs, baseDN, done.get(0));
+
+ // A session to the second replication server, which misses both, starts while the publisher runs.
+ connectTo(rd1, towardsSecond, SECOND_RS_ID);
+ release.countDown();
+ final ReplicationServer secondRs = second;
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(secondRs, baseDN, onBoth))
+ .as("the changes the second replication server holds after the one it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+ replace(dnA, "description", "once the recovery is over");
+ assertInTheChangelog(second, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ release.countDown();
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, second);
+ }
+ }
+
+ /**
+ * A session which starts once the missing change publisher of the previous one has ended its
+ * recovery, but before the publisher has left, gets a recovery of its own. It found the publisher
+ * still there and started none, the publisher then left, and the recovery the session needed was
+ * run by nobody: its replication server never got the change it missed, nor any local change done
+ * afterwards (issue #1199).
+ */
+ @Test
+ public void aSessionWhichStartsAsThePublisherLeavesGetsARecoveryOfItsOwn() 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);
+ final CountDownLatch held = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ try
+ {
+ second = createIsolatedReplicationServer(secondPort, "HistoricalCsnOrderingSecond", SECOND_RS_ID);
+ final DN dnA = addPerson("a");
+
+ // A change on the second replication server, which holds none of this replica before.
+ connectTo(rd1, towardsSecond, SECOND_RS_ID);
+ replace(dnA, "description", "on the second replication server");
+ final CSN onBoth = rd1.getLastLocalChange();
+ assertInTheChangelog(second, baseDN, onBoth);
+
+ // The session back to the first replication server recovers it, and its publisher is held once
+ // its run has ended the recovery.
+ final AtomicBoolean holdOnce = new AtomicBoolean(true);
+ rd1.setMissingChangesPublisherRunHook(() ->
+ {
+ if (holdOnce.compareAndSet(true, false))
+ {
+ holdUntil(held, release).run();
+ }
+ });
+ connectTo(rd1, towardsRs, FIRST_RS_ID);
+ assertTrue(held.await(30, SECONDS), "the publisher has run");
+ assertInTheChangelog(rs, baseDN, onBoth);
+ assertFalse(pendingChangesOf(rd1).isRecovering(), "the recovery is over");
+ replace(dnA, "description", "once the recovery is over");
+ final CSN missed = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, missed);
+
+ // A session to the second replication server, which misses that change, starts before the
+ // publisher has left.
+ connectTo(rd1, towardsSecond, SECOND_RS_ID);
+ release.countDown();
+ assertInTheChangelog(second, baseDN, missed);
+ waitUntilTheRecoveryIsOver(rd1);
+ replace(dnA, "description", "once the second recovery is over");
+ assertInTheChangelog(second, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ release.countDown();
+ rd1.setMissingChangesPublisherRunHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, second);
+ }
+ }
+
+ /**
+ * A recovery is not ended while a session which started meanwhile needs it. The publisher had
+ * nothing left to publish on the session before, ended the recovery, and the local changes went
+ * out to the replication server of the new session ahead of the ones it missed, which it then
+ * took none of (issue #1199).
+ */
+ @Test
+ public void aRecoveryASessionWhichStartedMeanwhileNeedsIsNotEnded() 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);
+ final CountDownLatch held = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ try
+ {
+ second = createIsolatedReplicationServer(secondPort, "HistoricalCsnOrderingSecond", SECOND_RS_ID);
+ final DN dnA = addPerson("a");
+ final CSN onBoth = aChangeBothReplicationServersHold(rd1, dnA, rs, towardsRs, second, towardsSecond);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "on the first replication server only");
+ done.add(rd1.getLastLocalChange());
+ assertInTheChangelog(rs, baseDN, done.get(0));
+
+ // As a publisher whose broker refused a change leaves it - one which searches for the local
+ // changes. The session started again needs no recovery, and its publisher, which is to end this
+ // one, is held on its search, which finds nothing.
+ startRecovery(rd1, false);
+ brokerOf(rd1).setRecoveryRequired(false);
+ rd1.setMissingChangesSearchHook(onTheSearch(1, search -> holdUntil(held, release).run()));
+ rd1.disable();
+ rd1.enable();
+ assertTrue(held.await(30, SECONDS), "the recovery searches");
+
+ // A session to the second replication server, which misses the change, starts meanwhile; a local
+ // change is done once the publisher has run on the session before.
+ connectTo(rd1, towardsSecond, SECOND_RS_ID);
+ final AtomicBoolean changeOnce = new AtomicBoolean(true);
+ rd1.setMissingChangesPublisherRunHook(() ->
+ {
+ if (changeOnce.compareAndSet(true, false))
+ {
+ unchecked(() ->
+ {
+ replace(dnA, "displayName", "once the publisher ran on the session before");
+ done.add(rd1.getLastLocalChange());
+ }).run();
+ }
+ });
+ release.countDown();
+
+ final ReplicationServer secondRs = second;
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(secondRs, baseDN, onBoth))
+ .as("the changes the second replication server holds after the one it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+ }
+ finally
+ {
+ release.countDown();
+ rd1.setMissingChangesSearchHook(null);
+ rd1.setMissingChangesPublisherRunHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, second);
+ }
+ }
+
+ /**
+ * A session which starts while a round of a recovery holds the local changes back to publish the ones
+ * the recovery kept has its replication server get them all: the recovery goes on from its start with
+ * what it kept. Taking the changes it kept for short of the last local change, as they are when one of
+ * them was not kept, the round had the recovery keep none from then on, and the run for the new session
+ * searched the historical information for them, which holds nothing of a change whose values a later
+ * one replaced.
+ */
+ @Test
+ public void aSessionWhichStartsWhileTheLocalChangesAreHeldBackGetsTheChangesTheRecoveryKept() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ final int secondPort = TestCaseUtils.findFreePort();
+ final SortedSet<String> towardsSecond = towards(secondPort);
+ // Created before the first one, the second replication server is the one which registers the
+ // changelog backend of the server: the first one, which goes away during the test, would take that
+ // backend with it, and the second one would fail on every change it is sent from then on. It is in
+ // another group than the domain, which goes to the first one, knowing of both, for as long as it is
+ // there.
+ ReplicationServer second =
+ createIsolatedReplicationServer(secondPort, "HistoricalCsnOrderingSecond", SECOND_RS_ID, 2);
+ ReplicationServer rs = createReplicationServer();
+ final SortedSet<String> towardsRs = new TreeSet<>(replServers);
+ final SortedSet<String> towardsBoth = new TreeSet<>(towardsRs);
+ towardsBoth.addAll(towardsSecond);
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ final DN dnA = addPerson("a");
+ final DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+ final CSN onBoth = aChangeBothReplicationServersHold(rd1, dnA, rs, towardsRs, second, towardsSecond);
+ connectTo(rd1, towardsBoth, FIRST_RS_ID);
+
+ // Both replication servers miss the changes done from here on. The recovery of the session started
+ // again rebuilds the first one from the historical information, and keeps the others: the first
+ // round keeps one done while it searches, the round which publishes it keeps two more, of the same
+ // values, and the round after it holds the local changes back to publish them.
+ refuseTheLocalChanges(rd1);
+ final List<CSN> done = new CopyOnWriteArrayList<>();
+ replace(dnB, "description", "missed by both");
+ done.add(rd1.getLastLocalChange());
+ rd1.setMissingChangesSearchHook(once(() ->
+ {
+ replace(dnC, "description", "kept while the first round searches");
+ done.add(rd1.getLastLocalChange());
+ }));
+ final AtomicBoolean changeOnce = new AtomicBoolean(true);
+ final AtomicBoolean startOnce = new AtomicBoolean(true);
+ rd1.setKeptChangesPublishedHook(heldBack -> unchecked(() ->
+ {
+ if (!heldBack && changeOnce.compareAndSet(true, false))
+ {
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+ }
+ else if (heldBack && startOnce.compareAndSet(true, false))
+ {
+ // The first replication server goes away once the held round has published the changes to it.
+ // The broker starts a session to the second one, which misses them, without the lock the round
+ // holds, and the session hands its start over to the publisher before the round ends.
+ rs.remove();
+ waitUntilAStartIsHandedOver(rd1);
+ }
+ }).run());
+ rd1.disable();
+ rd1.enable();
+
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(second, baseDN, onBoth))
+ .as("the changes the second replication server holds after the one it had")
+ .isEqualTo(done));
+ assertFalse(startOnce.get(), "the round which held the local changes back saw the start handed over");
+ waitUntilTheRecoveryIsOver(rd1);
+ replace(dnA, "description", "once the recovery is over");
+ assertInTheChangelog(second, baseDN, rd1.getLastLocalChange());
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ rd1.setKeptChangesPublishedHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, second);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ @Test
+ public void theLocalChangesARecoveryKeptGoWithTheDataTheyWereDoneOn() 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();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+ waitUntilTheStateIsSaved(baseDN, rsMaxCSN);
+ final List<Entry> backup = newArrayList(
+ DirectoryServer.getEntry(baseDN).duplicate(false), DirectoryServer.getEntry(dnA).duplicate(false));
+
+ // 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);
+ replace(dnA, "description", "on the data which is replaced");
+
+ // The data is restored while the domain is disabled, as it was before the change.
+ rd1.disable();
+ restore(backup);
+ rd1.enable();
+ waitUntilTheRecoveryIsOver(rd1);
+
+ assertFalse(DirectoryServer.getEntry(dnA).hasAttribute(
+ getServerContext().getSchema().getAttributeType("description")), "the change is not in the data");
+ // A change published once the recovery is over comes after whatever it published, on the session.
+ replace(dnA, "displayName", "once the data was restored");
+ final CSN afterTheRestore = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, afterTheRestore);
+ assertThat(changelogAfter(rs, baseDN, rsMaxCSN)).as("the changes published once the data was restored")
+ .containsExactly(afterTheRestore);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * What the search of a round finds takes no room among the deletes kept aside. It was put in the
+ * same capped list, and a local delete done while the round ran dropped the oldest entry of it - a
+ * change the search had just found, or a delete the replication server lacked - which nothing
+ * published afterwards (issue #1200).
+ */
+ @Test
+ public void theChangesARoundFindsTakeNoRoomAmongTheDeletesKeptAside() 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 DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+ final DN dnD = addPerson("d");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ rd1.setMaxDeletesKeptAside(2);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnA);
+ done.add(rd1.getLastLocalChange());
+ replace(dnB, "description", "found by the search");
+ done.add(rd1.getLastLocalChange());
+ replace(dnC, "description", "found by the search");
+ done.add(rd1.getLastLocalChange());
+ // Started after the changes, which the recovery then searches for.
+ startRecovery(rd1);
+ rd1.setMissingChangesSearchHook(once(() ->
+ {
+ TestCaseUtils.deleteEntry(dnD);
+ done.add(rd1.getLastLocalChange());
+ }));
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A delete no replication server has taken is not dropped from the deletes kept aside without a
+ * trace: nothing publishes it afterwards, and the replicas which miss it keep the entry. One a
+ * replication server has taken is dropped quietly (issue #1200).
+ */
+ @Test
+ public void aDeleteNoReplicationServerTookIsNotDroppedWithoutATrace() 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 DN dnB = addPerson("b");
+ final DN dnC = addPerson("c");
+
+ rd1.setMaxDeletesKeptAside(1);
+ TestCaseUtils.deleteEntry(dnA);
+ final CSN taken = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, taken);
+
+ TestCaseUtils.ERROR_TEXT_WRITER.clear();
+ refuseTheLocalChanges(rd1);
+ TestCaseUtils.deleteEntry(dnB);
+ final CSN notTaken = rd1.getLastLocalChange();
+ TestCaseUtils.deleteEntry(dnC);
+
+ final List<String> errors = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
+ assertThat(errors).as("a delete no replication server took, dropped")
+ .anyMatch(record -> record.contains(notTaken.toString()) && record.contains(dnB.toString()));
+ assertThat(errors).as("a delete a replication server took, dropped")
+ .noneMatch(record -> record.contains(taken.toString()));
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A local delete which did not go through is not published by a recovery. It was kept aside as its
+ * conflicts were resolved, before the backend refused it, and a recovery published it as a delete
+ * which was done: the replicas it reached deleted an entry this one keeps.
+ */
+ @Test
+ public void aDeleteWhichFailedIsNotPublishedByARecovery() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ final DN parent = DN.valueOf("ou=parent," + TEST_ROOT_DN_STRING);
+ TestCaseUtils.addEntry(
+ "dn: " + parent,
+ "objectClass: top",
+ "objectClass: organizationalUnit",
+ "ou: parent");
+ final DN child = DN.valueOf("cn=child," + parent);
+ TestCaseUtils.addEntry(
+ "dn: " + child,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: child",
+ "sn: test");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ assertEquals(InternalClientConnection.getRootConnection().processDelete(parent).getResultCode(),
+ ResultCode.NOT_ALLOWED_ON_NONLEAF, "the delete of an entry which has one below it");
+ replace(child, "description", "after the delete which failed");
+ final CSN change = rd1.getLastLocalChange();
+ // Started after the changes, which the recovery then searches for.
+ startRecovery(rd1);
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactly(change);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A broker which has a recovery to run refuses the changes of its replica only, which the recovery
+ * publishes in order. It refused every other message as well - the acknowledgements of the changes
+ * replayed here, the messages of a total update, its monitoring requests - for as long as the
+ * recovery ran (issue #1220).
+ */
+ @Test
+ public void aBrokerWithARecoveryToRunRefusesTheChangesOnly() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ // It sends the topology to the domain once, as it starts, and then only when it is asked to.
+ ReplicationServer rs = createReplicationServer(MINUTES.toMillis(10));
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ try
+ {
+ final DN dnA = addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+ refuseTheLocalChanges(rd1);
+
+ // A message which is no change goes through: the monitoring request is answered at once, where
+ // the broker waits 10 seconds for an answer before it gives up on it.
+ final long asked = System.nanoTime();
+ brokerOf(rd1).getReplicaStates();
+ assertThat(NANOSECONDS.toMillis(System.nanoTime() - asked)).as("the time the monitoring request took")
+ .isLessThan(5000);
+
+ // A change does not, up to the end of the recovery.
+ replace(dnA, "description", "while a recovery is required");
+ brokerOf(rd1).setRecoveryRequired(false);
+ replace(dnA, "description", "once no recovery is required");
+ final CSN taken = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, taken);
+ assertThat(changelogAfter(rs, baseDN, rsMaxCSN)).as("the changes the replication server holds after the ones it had")
+ .containsExactly(taken);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ @Test
+ public void aRecoveryWhoseKeptChangesMissOneSearchesForIt() 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 DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ startRecovery(rd1);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "kept");
+ done.add(rd1.getLastLocalChange());
+ replace(dnB, "description", "not kept");
+ done.add(rd1.getLastLocalChange());
+ pendingChangesOf(rd1).dropKeptChanges(Collections.singletonList(done.get(1)));
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The local changes a recovery kept outlive a domain disabled and enabled back on its own data: the
+ * state it saved on its way down covers them, and the change whose values a later one replaced is
+ * published still.
+ */
+ @Test
+ public void theLocalChangesARecoveryKeptOutliveADisable() 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();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+
+ // The broker refuses the announcement that the replica goes offline as well, which would have
+ // the replication server hold a CSN past the changes.
+ startRecovery(rd1);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+
+ rd1.disable();
+ rd1.enable();
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(rs, baseDN, rsMaxCSN))
+ .as("the changes the replication server holds after the ones it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The local changes a recovery kept outlive a domain disabled and enabled back on its own data whose
+ * save on its way down did not write the state - a save gives its turn up to the one in flight, and a
+ * write may fail: the state read back is brought up to the changes the historical information of the
+ * entries holds, so it covers them, and they are not taken for changes the data does not reflect.
+ */
+ @Test
+ public void theLocalChangesARecoveryKeptOutliveADisableWhoseSaveDidNotWrite() 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();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+
+ // The broker refuses the announcement that the replica goes offline as well, which would have
+ // the replication server hold a CSN past the changes.
+ startRecovery(rd1);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "first");
+ done.add(rd1.getLastLocalChange());
+ replace(dnA, "description", "second");
+ done.add(rd1.getLastLocalChange());
+
+ // The base entry holds the state saved before the changes.
+ rd1.disable();
+ leaveTheSavedState(baseDN, rsMaxCSN);
+ rd1.enable();
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(rs, baseDN, rsMaxCSN))
+ .as("the changes the replication server holds after the ones it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A suffix deleted and added back while a recovery runs outlives a domain disabled and enabled back
+ * before the state could be saved into the base entry added back. The state read back held no change
+ * of this replica, and the state is brought up to the changes of this replica the historical
+ * information holds only past the newest one it read: the changes the recovery kept were all taken
+ * for changes the data does not reflect, the add of the base entry was dropped, and the recovery
+ * published the deletes which removed the suffix alone - the other replicas lost it.
+ */
+ @Test
+ public void aSuffixAddedBackDuringARecoveryOutlivesAStateWhichCouldNotBeSavedInIt() 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();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+
+ startRecovery(rd1);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnA);
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(baseDN);
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.addEntry(StaticUtils.createEntry(baseDN));
+ done.add(rd1.getLastLocalChange());
+
+ // The base entry added back holds no state: none was saved into it.
+ rd1.disable();
+ leaveTheSavedState(baseDN);
+ rd1.enable();
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(rs, baseDN, rsMaxCSN))
+ .as("the changes the replication server holds after the ones it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A delete kept aside goes with the data it was done on: once the data is restored from a backup
+ * taken before it, the entry is back, and a recovery which published the delete would have the
+ * other replicas delete an entry this one holds (issue #1221).
+ */
+ @Test
+ public void theDeletesKeptAsideGoWithTheDataTheyWereDoneOn() 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 DN dnB = addPerson("b");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+ waitUntilTheStateIsSaved(baseDN, rsMaxCSN);
+ final List<Entry> backup = newArrayList(DirectoryServer.getEntry(baseDN).duplicate(false),
+ DirectoryServer.getEntry(dnA).duplicate(false), DirectoryServer.getEntry(dnB).duplicate(false));
+ TestCaseUtils.deleteEntry(dnB);
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+
+ // The data is restored while the domain is disabled, as it was before the delete.
+ rd1.disable();
+ restore(backup);
+ rd1.enable();
+ assertNotNull(DirectoryServer.getEntry(dnB), "the entry is back");
+
+ // A replication server which misses the changes after the backup: its recovery publishes the
+ // change done since, and not the delete.
+ replace(dnA, "description", "once the data was restored");
+ final CSN afterTheRestore = rd1.getLastLocalChange();
+ startRecovery(rd1);
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactly(afterTheRestore);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A round whose search answers that the base entry is not there, while it is, has not gone
+ * through: the base DN had no backend for the search - one an import took away - and the round
+ * found nothing of what is there. Taken for a search which found everything, it moved the recovery
+ * past the changes of its window, which no later session published.
+ */
+ @Test
+ public void aSearchWhichFindsNoBackendFailsItsRound() 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", "while the backend is away");
+ done.add(rd1.getLastLocalChange());
+ // Started after the change, which the recovery then searches for.
+ startRecovery(rd1);
+
+ rd1.setMissingChangesSearchHook(onTheFirstSearch(search -> search.setResultCode(ResultCode.NO_SUCH_OBJECT)));
+ final List<ReplicationMsg> published = new ArrayList<>();
+ Assertions.assertThatThrownBy(() -> rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)))
+ .as("a recovery whose search found no backend")
+ .hasMessageContaining(ResultCode.NO_SUCH_OBJECT.getName().toString());
+ assertThat(published).as("what a round whose search found no backend published").isEmpty();
+
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A recovery which keeps failing waits longer before each session restart - one second more for
+ * each failure in a row - and forgets its failures once it succeeds. A publisher which restarted
+ * the session at once would keep the session restarting for as long as the backend fails, and one
+ * which kept its count would start the next failure with the longest wait.
+ */
+ @Test
+ public void aRecoveryWhichKeepsFailingWaitsLongerEachTimeAndForgetsItOnceItSucceeds() 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();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+
+ // The replication server misses the change done from here on.
+ refuseTheLocalChanges(rd1);
+ final List<CSN> done = new ArrayList<>();
+ replace(dnA, "description", "missed by the replication server");
+ done.add(rd1.getLastLocalChange());
+
+ // The search of the recovery fails twice in a row, each time on a session of its own.
+ TestCaseUtils.ERROR_TEXT_WRITER.clear();
+ final List<Long> searches = new CopyOnWriteArrayList<>();
+ rd1.setMissingChangesSearchHook(search ->
+ {
+ searches.add(System.nanoTime());
+ if (searches.size() <= 2)
+ {
+ search.setResultCode(ResultCode.UNAVAILABLE);
+ }
+ });
+ rd1.disable();
+ rd1.enable();
+
+ new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(rs, baseDN, rsMaxCSN))
+ .as("the changes the replication server holds after the ones it had")
+ .isEqualTo(done));
+ waitUntilTheRecoveryIsOver(rd1);
+
+ final List<String> errors = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
+ assertThat(errors).as("the first failure")
+ .anyMatch(record -> record.contains("tries again within 1000 ms"));
+ assertThat(errors).as("the second failure in a row")
+ .anyMatch(record -> record.contains("tries again within 2000 ms"));
+ assertThat(searches.size()).as("the searches of the recovery").isGreaterThanOrEqualTo(3);
+ assertThat(NANOSECONDS.toMillis(searches.get(2) - searches.get(1)))
+ .as("the time from the second failure to the next search").isGreaterThanOrEqualTo(2000);
+ assertEquals(consecutiveRecoveryFailuresOf(rd1), 0, "the failures in a row, once the recovery succeeded");
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A recovery which threw an error - which no catch of an exception takes - has the session started
+ * again as well. Its publisher left with the recovery on and asked for nothing: the local changes
+ * were left to a recovery nobody ran.
+ */
+ @Test
+ public void aRecoveryWhichThrewAnErrorHasTheSessionStartedAgain() throws Exception
+ {
+ assertARecoveryWhichFailedHasTheSessionStartedAgain(search ->
+ {
+ throw new AssertionError("the round threw an error, as a test asked");
+ });
+ }
+
+ /**
+ * The replica is not announced offline while a recovery has changes to publish. The announcement
+ * carries the newest CSN of this replica, and the replication server takes the replica for offline
+ * from there: its changelog index would have gone past the changes which reach it afterwards.
+ */
+ @Test
+ public void theReplicaIsNotAnnouncedOfflineWhileARecoveryHasChangesToPublish() 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");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+
+ // As a publisher whose broker refused a change leaves it: the broker would take the announcement.
+ startRecovery(rd1);
+ brokerOf(rd1).setRecoveryRequired(false);
+ replace(dnA, "description", "left to the recovery");
+
+ assertNull(pendingChangesOf(rd1).putReplicaOfflineMsg(), "the announcement that the replica goes offline");
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A session which starts while a failed recovery waits to have the session started again has its
+ * changes recovered at once: it is the next session the recovery waited for. Its start was handed
+ * over and left there until the wait was over - up to ten seconds - while the local changes piled
+ * up for the recovery.
+ */
+ @Test
+ public void aSessionWhichStartsWhileAFailedRecoveryWaitsIsRecoveredAtOnce() 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);
+
+ // The replication servers miss the change done from here on.
+ refuseTheLocalChanges(rd1);
+ replace(dnA, "description", "missed by both replication servers");
+ final CSN missed = rd1.getLastLocalChange();
+
+ // The recovery of the session started again fails, after nine failures in a row: it waits ten
+ // seconds to have the session started again.
+ setConsecutiveRecoveryFailures(rd1, 9);
+ final CountDownLatch failed = new CountDownLatch(1);
+ rd1.setMissingChangesSearchHook(onTheFirstSearch(search ->
+ {
+ search.setResultCode(ResultCode.UNAVAILABLE);
+ failed.countDown();
+ }));
+ rd1.disable();
+ rd1.enable();
+ assertTrue(failed.await(30, SECONDS), "the recovery failed");
+
+ // A session to the second replication server starts while the recovery waits.
+ final long started = System.nanoTime();
+ connectTo(rd1, towardsSecond, SECOND_RS_ID);
+ assertInTheChangelog(second, baseDN, missed);
+ assertThat(NANOSECONDS.toMillis(System.nanoTime() - started))
+ .as("the time the session took to get the change it missed").isLessThan(5000);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs, second);
+ }
+ }
+
+ /**
+ * The deletes of a suffix outlive a ServerState which could not be saved: the delete of the base
+ * entry took the place the state is saved in with it, and the state read back is older than the
+ * 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.
+ */
+ @Test
+ public void theDeletesOfASuffixOutliveAStateWhichCouldNotBeSaved() 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();
+ assertInTheChangelog(rs, baseDN, rsMaxCSN);
+
+ // The replication server misses the deletes.
+ refuseTheLocalChanges(rd1);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnA);
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(baseDN);
+ done.add(rd1.getLastLocalChange());
+
+ rd1.disable();
+ rd1.enable();
+ final CSN readBack = rd1.getLastLocalChange();
+ assertTrue(readBack == null || readBack.isOlderThan(done.get(0)), "the state read back is older than the deletes");
+
+ TestCaseUtils.addEntry(StaticUtils.createEntry(baseDN));
+ done.add(rd1.getLastLocalChange());
+ // Started after the changes, which the recovery then searches for.
+ startRecovery(rd1);
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * The deletes of a suffix a recovery kept outlive a ServerState which could not be saved as well:
+ * their entries are gone, so the data reflects them.
+ */
+ @Test
+ public void theDeletesOfASuffixARecoveryKeptOutliveAStateWhichCouldNotBeSaved() 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");
+ startRecovery(rd1);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnA);
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(baseDN);
+ done.add(rd1.getLastLocalChange());
+ // No recovery runs as the domain is enabled back: what it kept is read as the domain leaves it.
+ pendingChangesOf(rd1).setRecovering(false);
+ brokerOf(rd1).setRecoveryRequired(false);
+
+ rd1.disable();
+ rd1.enable();
+ final CSN readBack = rd1.getLastLocalChange();
+ assertTrue(readBack == null || readBack.isOlderThan(done.get(0)), "the state read back is older than the deletes");
+
+ assertThat(csns(new ArrayList<ReplicationMsg>(pendingChangesOf(rd1).keptChangesNewerThan(null))))
+ .as("the deletes the recovery kept").containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A replica which reads back a ServerState with no change of its own numbers its next change after
+ * the ones it numbered before (issue #1223). The CSN generator went back to the clock, and the
+ * session started next took it up to what its replication server holds only: past a change of a
+ * replica whose clock runs ahead, which this one replayed, and past the last change of this one the
+ * replication server took. The deletes it missed came after that change, and the next change was
+ * numbered as the first of them - a recovery would publish one of the two.
+ */
+ @Test
+ public void aReplicaWhichReadsBackAStateWithNoChangeOfItsOwnNumbersItsNextChangeAfterItsLastOne()
+ throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ TestCaseUtils.initializeTestBackend(true);
+ ReplicationServer rs = createReplicationServer();
+ LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
+ ReplicationBroker other = null;
+ try
+ {
+ final DN dnA = addPerson("a");
+ assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
+
+ // The changes of a replica whose clock runs an hour ahead have the generator follow it.
+ other = openReplicationSession(baseDN, OTHER_SERVER_ID, 100, rsPort, 1000);
+ final CSN otherCSN = new CSNGenerator(OTHER_SERVER_ID, TimeThread.getTime() + HOURS.toMillis(1)).newCSN();
+ publishOtherReplicaChange(other, otherCSN, dnA, "from a replica whose clock runs ahead");
+ waitUntilReplayed(rd1, otherCSN);
+
+ // The replication server misses the deletes, and the delete of the base entry takes the saved
+ // state with it.
+ refuseTheLocalChanges(rd1);
+ final List<CSN> done = new ArrayList<>();
+ TestCaseUtils.deleteEntry(dnA);
+ done.add(rd1.getLastLocalChange());
+ TestCaseUtils.deleteEntry(baseDN);
+ done.add(rd1.getLastLocalChange());
+
+ rd1.disable();
+ rd1.enable();
+ final CSN readBack = rd1.getLastLocalChange();
+ assertTrue(readBack == null || readBack.isOlderThan(done.get(0)), "the state read back is older than the deletes");
+
+ TestCaseUtils.addEntry(StaticUtils.createEntry(baseDN));
+ final CSN next = rd1.getLastLocalChange();
+ assertTrue(next.isNewerThan(done.get(1)), "the next change " + next + " is numbered after the deletes " + done);
+ }
+ finally
+ {
+ stop(other);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A local change an assured domain does during a recovery is published by it as not assured, as a
+ * recovery publishes what it rebuilds: its operation, which may still be waiting for the
+ * acknowledgement, gets none and times out.
+ */
+ @Test
+ public void aKeptChangeIsPublishedNotAssured() 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, 500, null));
+ rd1.start();
+ try
+ {
+ final DN dnA = addPerson("a");
+ final CSN rsMaxCSN = rd1.getLastLocalChange();
+
+ startRecovery(rd1);
+ replace(dnA, "description", "assured");
+
+ final List<ReplicationMsg> published = new ArrayList<>();
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(published).hasSize(1);
+ final UpdateMsg kept = (UpdateMsg) published.get(0);
+ assertFalse(kept.isAssured(), "the kept change is assured");
+ final short version = ProtocolVersion.getCurrentVersion();
+ assertFalse(((UpdateMsg) ReplicationMsg.generateMsg(kept.getBytes(version), version)).isAssured(),
+ "the kept change is assured as it is encoded");
+ }
+ finally
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * A recovery publishes nothing while a total update replaces the data of this replica: what it
+ * would publish was done on the data the import replaces. The session started again after the
+ * import recovers from the state the new data holds.
+ */
+ @Test
+ public void aRecoveryPublishesNothingWhileATotalUpdateReplacesTheData() 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", "searched for");
+ done.add(rd1.getLastLocalChange());
+ // Started after the change, which the recovery then searches for, and before the next one, which
+ // it keeps.
+ startRecovery(rd1);
+ replace(dnA, "displayName", "kept");
+ done.add(rd1.getLastLocalChange());
+ final List<ReplicationMsg> published = new ArrayList<>();
+
+ // A total update which runs as the recovery starts: the recovery searches nothing, and publishes
+ // nothing.
+ final AtomicInteger searches = new AtomicInteger();
+ rd1.setMissingChangesSearchHook(search -> searches.incrementAndGet());
+ setImportingData(rd1, true);
+ try
+ {
+ assertFalse(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)),
+ "a recovery while a total update replaces the data");
+ assertThat(published).as("what a recovery published while a total update replaces the data").isEmpty();
+ assertThat(searches.get()).as("the searches of a recovery while a total update replaces the data").isZero();
+ }
+ finally
+ {
+ setImportingData(rd1, false);
+ }
+
+ // One which starts while a round searches: the round publishes nothing of what it found.
+ rd1.setMissingChangesSearchHook(onTheFirstSearch(search -> unchecked(() -> setImportingData(rd1, true)).run()));
+ try
+ {
+ assertFalse(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)),
+ "a recovery during which a total update began");
+ assertThat(published).as("what a round published once a total update began").isEmpty();
+ }
+ finally
+ {
+ setImportingData(rd1, false);
+ rd1.setMissingChangesSearchHook(null);
+ }
+
+ assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
+ assertThat(csns(published)).containsExactlyElementsOf(done);
+ }
+ finally
+ {
+ rd1.setMissingChangesSearchHook(null);
+ MultimasterReplication.deleteDomain(baseDN);
+ remove(rs);
+ }
+ }
+
+ /**
+ * Does a local change on the writer, and waits for it for a while: a change the last round holds
+ * back is done once the recovery is over.
+ */
+ private static void doLocalChange(ExecutorService writer, List<Future<?>> changes, List<CSN> done,
+ LDAPReplicationDomain domain, DN dn, int round)
+ {
+ final Future<?> change = writer.submit(unchecked(() ->
+ {
+ replace(dn, "description", "done while round " + round + " searches");
+ done.add(domain.getLastLocalChange());
+ }));
+ changes.add(change);
+ try
+ {
+ change.get(500, MILLISECONDS);
+ }
+ catch (TimeoutException heldBackByTheLastRound)
+ {
+ // it is done once the recovery is over
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Asserts that each local change was published once: the first ones by the recovery, in order,
+ * and the others by the broker once the recovery was over - and none of the first ones by the
+ * broker as well.
+ * <p>
+ * The changelog of the replication server holds the changes the broker published, the recovery
+ * publishing to a broker of the test: after the last change it had, it has to hold the others,
+ * and nothing else. One session delivers in order, so a change of the recovery the broker
+ * published as well reaches it before the others and keeps the two from ever matching.
+ */
+ private static void assertEachChangePublishedOnce(ReplicationServer rs, DN baseDN, CSN rsMaxCSN,
+ List<ReplicationMsg> published, List<Future<?>> changes, List<CSN> done) throws Exception
+ {
+ for (Future<?> change : changes)
+ {
+ change.get(10, SECONDS);
+ }
+ final List<CSN> recovered = csns(published);
+ assertThat(recovered.size()).isLessThanOrEqualTo(done.size());
+ assertThat(done.subList(0, recovered.size())).isEqualTo(recovered);
+ final List<CSN> afterTheRecovery = done.subList(recovered.size(), done.size());
+ new TestTimer.Builder()
+ .maxSleep(10, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertThat(changelogAfter(rs, baseDN, rsMaxCSN))
+ .as("the changes the broker published after the recovery")
+ .isEqualTo(afterTheRecovery));
+ }
+
+ /**
+ * Returns the changes of the replica of the provided CSN which the replication server holds after
+ * it. The announcements that the replica went offline, which each stop of its broker publishes,
+ * are no changes, and are left out.
+ */
+ private static List<CSN> changelogAfter(ReplicationServer rs, DN baseDN, CSN csn) throws Exception
+ {
+ final List<CSN> csns = new ArrayList<>();
+ try (DBCursor<UpdateMsg> cursor = rs.getChangelogDB().getReplicationDomainDB().getCursorFrom(
+ baseDN, csn.getServerId(), csn, new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, AFTER_MATCHING_KEY)))
+ {
+ while (cursor.next())
+ {
+ if (!(cursor.getRecord() instanceof ReplicaOfflineMsg))
+ {
+ csns.add(cursor.getRecord().getCSN());
+ }
+ }
+ }
+ return csns;
+ }
+
+ private static DN addPerson(String cn) throws Exception
+ {
+ final DN dn = DN.valueOf("cn=" + cn + "," + TEST_ROOT_DN_STRING);
+ TestCaseUtils.addEntry(
+ "dn: " + dn,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "cn: " + cn,
+ "sn: test");
+ return dn;
+ }
+
+ private static void replace(DN dn, String attribute, String value) throws Exception
+ {
+ assertEquals(TestCaseUtils.applyModifications(false,
+ "dn: " + dn,
+ "changetype: modify",
+ "replace: " + attribute,
+ attribute + ": " + value), 0);
+ }
+
+ /** Has the other replica change the description of the entry, with the provided CSN. */
+ private void publishOtherReplicaChange(ReplicationBroker other, CSN csn, DN dn, String description)
+ throws Exception
+ {
+ final AttributeType type = getServerContext().getSchema().getAttributeType("description");
+ other.publish(new ModifyMsg(csn, dn,
+ newArrayList(new Modification(ModificationType.REPLACE, Attributes.create(type, description))),
+ getEntryUUID(dn)));
+ }
+
+ /**
+ * Puts the domain where a session start leaves it when its replication server misses some of
+ * its changes, without the thread which would publish them: the tests call
+ * {@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.
+ */
+ private static void startRecovery(LDAPReplicationDomain domain) throws Exception
+ {
+ startRecovery(domain, true);
+ }
+
+ /**
+ * Puts the domain where a session start leaves it when its replication server misses some of
+ * its changes, as {@link #startRecovery(LDAPReplicationDomain)} does, with a recovery which keeps
+ * the local changes done while it runs, or - as one which kept too many of them - keeps none, and
+ * searches for them in the historical information as well.
+ */
+ private static void startRecovery(LDAPReplicationDomain domain, boolean keepsTheLocalChanges) throws Exception
+ {
+ if (!keepsTheLocalChanges)
+ {
+ pendingChangesOf(domain).setMaxKeptChanges(0);
+ }
+ pendingChangesOf(domain).setRecovering(true);
+ brokerOf(domain).setRecoveryRequired(true);
+ }
+
+ /**
+ * 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.
+ */
+ private static void refuseTheLocalChanges(LDAPReplicationDomain domain) throws Exception
+ {
+ brokerOf(domain).setRecoveryRequired(true);
+ }
+
+ private static PendingChanges pendingChangesOf(LDAPReplicationDomain domain) throws Exception
+ {
+ final Field pendingChanges = LDAPReplicationDomain.class.getDeclaredField("pendingChanges");
+ pendingChanges.setAccessible(true);
+ return (PendingChanges) pendingChanges.get(domain);
+ }
+
+ private static ReplicationBroker brokerOf(ReplicationDomain domain) throws Exception
+ {
+ return (ReplicationBroker) brokerField().get(domain);
+ }
+
+ private static void setBroker(ReplicationDomain domain, ReplicationBroker broker) throws Exception
+ {
+ brokerField().set(domain, broker);
+ }
+
+ private static Field brokerField() throws Exception
+ {
+ final Field broker = ReplicationDomain.class.getDeclaredField("broker");
+ broker.setAccessible(true);
+ return broker;
+ }
+
+ private static List<CSN> csns(List<ReplicationMsg> msgs)
+ {
+ final List<CSN> csns = new ArrayList<>();
+ for (ReplicationMsg msg : msgs)
+ {
+ csns.add(((UpdateMsg) msg).getCSN());
+ }
+ return csns;
+ }
+
+ private static void waitUntilReplayed(final LDAPReplicationDomain domain, final CSN csn) throws Exception
+ {
+ new TestTimer.Builder()
+ .maxSleep(10, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertTrue(domain.getServerState().cover(csn), csn + " replayed"));
+ }
+
+ /**
+ * Asserts that the change is in the changelog of the replication server. Its state would not
+ * do: it holds the newest change of each replica, which covers a change the replication server
+ * never received as soon as a later one reached it.
+ */
+ private static void assertInTheChangelog(final ReplicationServer rs, final DN baseDN, final CSN csn)
+ throws Exception
+ {
+ new TestTimer.Builder()
+ .maxSleep(10, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() ->
+ {
+ try (DBCursor<UpdateMsg> cursor = rs.getChangelogDB().getReplicationDomainDB().getCursorFrom(
+ baseDN, csn.getServerId(), csn, new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY)))
+ {
+ assertTrue(cursor.next() && csn.equals(cursor.getRecord().getCSN()),
+ csn + " is in the changelog of the replication server");
+ }
+ });
+ }
+
+ /** A step of a test which may throw, for where only a {@link Runnable} is taken. */
+ private interface Step
+ {
+ void run() throws Exception;
+ }
+
+ private static Runnable unchecked(final Step step)
+ {
+ return () ->
+ {
+ try
+ {
+ step.run();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ };
+ }
+
+ /** Runs the step the first time only: the rounds after the first are left alone. */
+ private static Consumer<InternalSearchOperation> once(final Step step)
+ {
+ final Runnable run = unchecked(step);
+ return onTheFirstSearch(search -> run.run());
+ }
+
+ /** Runs the step on the search of the first round only: the rounds after it are left alone. */
+ private static Consumer<InternalSearchOperation> onTheFirstSearch(final Consumer<InternalSearchOperation> step)
+ {
+ final AtomicBoolean ran = new AtomicBoolean();
+ return search ->
+ {
+ if (ran.compareAndSet(false, true))
+ {
+ step.accept(search);
+ }
+ };
+ }
+
private ReplicationServer createReplicationServer() throws Exception
{
- int rsPort = TestCaseUtils.findFreePort();
+ return createReplicationServer(null);
+ }
+
+ /**
+ * Creates the replication server of the tests, whose monitoring publisher has the provided period,
+ * or the default one if it is {@code null}.
+ */
+ private ReplicationServer createReplicationServer(Long monitoringPeriod) throws Exception
+ {
+ rsPort = TestCaseUtils.findFreePort();
replServers.add("localhost:" + rsPort);
- ReplicationServer replicationServer = new ReplicationServer(
- new ReplServerFakeConfiguration(rsPort, "HistoricalCsnOrdering", 0, 1, 0, 100, replServers, 1, 1000, 5000));
+ final ReplServerFakeConfiguration configuration =
+ new ReplServerFakeConfiguration(rsPort, "HistoricalCsnOrdering", 0, FIRST_RS_ID, 0, 100, replServers, 1, 1000, 5000);
+ if (monitoringPeriod != null)
+ {
+ configuration.setMonitoringPeriod(monitoringPeriod);
+ }
+ ReplicationServer replicationServer = new ReplicationServer(configuration);
clearChangelogDB(replicationServer);
return replicationServer;
}
@@ -335,11 +2951,229 @@
private LDAPReplicationDomain createReplicationDomain(int dsId)
throws DirectoryException, ConfigException
{
- final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
- final DomainFakeCfg domainConf = new DomainFakeCfg(
- baseDN, dsId, replServers, AssuredType.NOT_ASSURED, 2, 1, 0, null);
- LDAPReplicationDomain replicationDomain = MultimasterReplication.createNewDomain(domainConf);
+ LDAPReplicationDomain replicationDomain =
+ MultimasterReplication.createNewDomain(domainConfiguration(dsId, replServers));
replicationDomain.start();
return replicationDomain;
}
+
+ /** Returns the configuration of the domain of the tests, connected to the provided replication servers. */
+ private static DomainFakeCfg domainConfiguration(int dsId, SortedSet<String> servers)
+ {
+ return new DomainFakeCfg(
+ DN.valueOf(TEST_ROOT_DN_STRING), dsId, servers, AssuredType.NOT_ASSURED, 2, 1, 0, null);
+ }
+
+ /** Returns the configured replication servers of a domain connected to the one listening on the port only. */
+ private static SortedSet<String> towards(int port)
+ {
+ return new TreeSet<>(Collections.singleton("localhost:" + port));
+ }
+
+ /**
+ * Creates a replication server which knows of no other: what it holds of the domain of the tests,
+ * it holds from the domain alone.
+ */
+ private ReplicationServer createIsolatedReplicationServer(int port, String dirName, int rsServerId)
+ throws Exception
+ {
+ return createIsolatedReplicationServer(port, dirName, rsServerId, 1);
+ }
+
+ /** Creates a replication server which knows of no other, in the provided group. */
+ private ReplicationServer createIsolatedReplicationServer(int port, String dirName, int rsServerId, int groupId)
+ throws Exception
+ {
+ final ReplicationServer replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ port, dirName, 0, rsServerId, 0, 100, towards(port), groupId, 1000, 5000));
+ clearChangelogDB(replicationServer);
+ return replicationServer;
+ }
+
+ /** Has the domain connect to the provided replication servers, and waits for its session to the one expected. */
+ private void connectTo(final LDAPReplicationDomain domain, SortedSet<String> servers, final int rsServerId)
+ throws Exception
+ {
+ assertEquals(domain.applyConfigurationChange(domainConfiguration(serverId, servers)).getResultCode(),
+ ResultCode.SUCCESS);
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() ->
+ {
+ assertTrue(domain.isConnected(), "the domain has a session");
+ assertEquals(domain.getRsServerId(), rsServerId, "the replication server of the session");
+ });
+ }
+
+ /**
+ * Has the domain do a change on the second replication server, which holds no change of this
+ * replica before, and connect back to the first one, which the recovery of that session publishes
+ * the change to. Both hold it then, as the newest change of this replica.
+ *
+ * @return the change
+ */
+ private CSN aChangeBothReplicationServersHold(LDAPReplicationDomain domain, DN dn, ReplicationServer first,
+ SortedSet<String> towardsFirst, ReplicationServer second, SortedSet<String> towardsSecond) throws Exception
+ {
+ final DN baseDN = domain.getBaseDN();
+ connectTo(domain, towardsSecond, SECOND_RS_ID);
+ replace(dn, "description", "on both replication servers");
+ final CSN csn = domain.getLastLocalChange();
+ assertInTheChangelog(second, baseDN, csn);
+ connectTo(domain, towardsFirst, FIRST_RS_ID);
+ assertInTheChangelog(first, baseDN, csn);
+ waitUntilTheRecoveryIsOver(domain);
+ return csn;
+ }
+
+ /** Runs the step on the search of the provided round only: the rounds before and after it are left alone. */
+ private static Consumer<InternalSearchOperation> onTheSearch(final int round,
+ final Consumer<InternalSearchOperation> step)
+ {
+ final AtomicInteger searches = new AtomicInteger();
+ return search ->
+ {
+ if (searches.incrementAndGet() == round)
+ {
+ step.accept(search);
+ }
+ };
+ }
+
+ /** Returns what tells that a thread is held, and holds it until it is released, for a while at most. */
+ private static Runnable holdUntil(final CountDownLatch held, final CountDownLatch release)
+ {
+ return unchecked(() ->
+ {
+ held.countDown();
+ assertTrue(release.await(30, SECONDS), "released");
+ });
+ }
+
+ /**
+ * Waits for the state checkpointer to have saved, in the base entry, a state which covers the
+ * provided change: a backup taken before holds no state of this replica, where the tests restore one
+ * which tells the changes the data reflects from the ones it does not.
+ */
+ private static void waitUntilTheStateIsSaved(final DN baseDN, final CSN csn) throws Exception
+ {
+ final AttributeType stateType = getServerContext().getSchema().getAttributeType("ds-sync-state");
+ new TestTimer.Builder()
+ .maxSleep(10, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() ->
+ {
+ boolean saved = false;
+ for (Attribute attribute : DirectoryServer.getEntry(baseDN).getAllAttributes(stateType))
+ {
+ for (ByteString value : attribute)
+ {
+ final CSN savedCSN = new CSN(value.toString());
+ saved |= savedCSN.getServerId() == csn.getServerId() && savedCSN.isNewerThanOrEqualTo(csn);
+ }
+ }
+ assertTrue(saved, csn + " is in the state saved in the base entry");
+ });
+ }
+
+ /**
+ * Has the base entry hold the provided ServerState - none if no CSN is provided - as a save of the
+ * state which did not write leaves an older one there.
+ */
+ private static void leaveTheSavedState(DN baseDN, CSN... csns) throws Exception
+ {
+ 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 left = saved.duplicate(false);
+ left.removeAttribute(stateType);
+ if (csns.length > 0)
+ {
+ final AttributeBuilder state = new AttributeBuilder(stateType);
+ for (CSN csn : csns)
+ {
+ state.add(csn.toString());
+ }
+ left.putAttribute(stateType, newArrayList(state.toAttribute()));
+ }
+ backend.replaceEntry(saved, left, null);
+ }
+
+ /** Replaces the data of the test backend with the provided entries, as a restore does. */
+ private static void restore(List<Entry> backup) throws Exception
+ {
+ final MemoryBackend backend =
+ (MemoryBackend) getServerContext().getBackendConfigManager().getLocalBackendById(TEST_BACKEND_ID);
+ backend.clearMemoryBackend();
+ for (Entry entry : backup)
+ {
+ backend.addEntry(entry, null);
+ }
+ }
+
+ private static int consecutiveRecoveryFailuresOf(LDAPReplicationDomain domain) throws Exception
+ {
+ final Field failures = LDAPReplicationDomain.class.getDeclaredField("consecutiveRecoveryFailures");
+ failures.setAccessible(true);
+ return ((AtomicInteger) failures.get(domain)).get();
+ }
+
+ private static void setConsecutiveRecoveryFailures(LDAPReplicationDomain domain, int failures) throws Exception
+ {
+ final Field field = LDAPReplicationDomain.class.getDeclaredField("consecutiveRecoveryFailures");
+ field.setAccessible(true);
+ ((AtomicInteger) field.get(domain)).set(failures);
+ }
+
+ /** Has the domain behave as while a total update replaces its data, or no longer. */
+ private static void setImportingData(LDAPReplicationDomain domain, boolean importing) throws Exception
+ {
+ final Field field = LDAPReplicationDomain.class.getDeclaredField("importingData");
+ field.setAccessible(true);
+ field.setBoolean(domain, importing);
+ }
+
+ /**
+ * Waits for a session which started while the missing change publisher of the domain runs to hand it
+ * its start - under the lock the session hands it over under, which a thread holding the lock of the
+ * pending changes may take.
+ */
+ private static void waitUntilAStartIsHandedOver(final LDAPReplicationDomain domain) throws Exception
+ {
+ final Field recoveryLockField = LDAPReplicationDomain.class.getDeclaredField("recoveryLock");
+ recoveryLockField.setAccessible(true);
+ final Object recoveryLock = recoveryLockField.get(domain);
+ final Field publisherField = LDAPReplicationDomain.class.getDeclaredField("rsUpdater");
+ publisherField.setAccessible(true);
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() ->
+ {
+ synchronized (recoveryLock)
+ {
+ final Object publisher = publisherField.get(domain);
+ assertNotNull(publisher, "a missing change publisher runs");
+ final Field nextStart = publisher.getClass().getDeclaredField("nextStart");
+ nextStart.setAccessible(true);
+ assertNotNull(nextStart.get(publisher), "a session handed its start over");
+ }
+ });
+ }
+
+ /** Waits for a missing change publisher of the domain to end the recovery. */
+ private static void waitUntilTheRecoveryIsOver(final LDAPReplicationDomain domain) throws Exception
+ {
+ final PendingChanges pendingChanges = pendingChangesOf(domain);
+ new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer()
+ .repeatUntilSuccess(() -> assertFalse(pendingChanges.isRecovering(), "the recovery is over"));
+ }
}
--
Gitblit v1.10.0