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/LDAPReplicationDomain.java | 1173 ++++++++++++++++++++++++++++++++++++++++++++++++++--------
1 files changed, 1,003 insertions(+), 170 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index 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;
}
/**
--
Gitblit v1.10.0