[#1195] Publish every change done while a recovery runs, and end the recove...

Valery Kharseko

23 hours ago f0a287fcba94440f81845a04fd0b6a4a70075a2c

 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;
    }


 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;
  }
}

 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;
  }

  /**

 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

 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)

 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;
        }

 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

 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.
   */

 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"));
  }
}