[#1246] Keep the messages of the last local changes, and have a recovery pu...

Valery Kharseko

11 hours ago c6650aedac5c1ade8aee9e7a76eb0dd100a3e179

 opendj-server-legacy/src/main/java/org/opends/server/api/LocalBackend.java

@@ -20,6 +20,8 @@
import static org.opends.messages.BackendMessages.*;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -76,6 +78,39 @@
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** Process-lifetime write barriers for data which cannot safely be used after an online replacement. */
  private static final Map<DN, LocalizableMessage> writeFailures = new ConcurrentHashMap<>();

  /**
   * Prevents writes overlapping a failed data replacement until the server process is restarted.
   * An in-core restart keeps the barrier, as it keeps the JVM running.
   * The barrier survives backend and synchronization-provider reconfiguration.
   * @param baseDN the suffix whose data cannot safely be written
   * @param reason the reason to report to the operator and to rejected updates
   */
  public static void quarantineWrites(DN baseDN, LocalizableMessage reason)
  {
    writeFailures.putIfAbsent(baseDN, reason);
  }

  /**
   * Returns the reason writes overlapping a suffix are forbidden, or null if none are forbidden.
   * Ancestors are included because deleting or moving a subtree affects the suffix below it.
   * @param dn the DN of the entry or subtree an operation changes
   * @return the recorded failure, or null
   */
  public static LocalizableMessage getWriteFailure(DN dn)
  {
    for (Map.Entry<DN, LocalizableMessage> failure : writeFailures.entrySet())
    {
      if (dn.isSubordinateOrEqualTo(failure.getKey()) || dn.isSuperiorOrEqualTo(failure.getKey()))
      {
        return failure.getValue();
      }
    }
    return null;
  }

  /** Indicates whether this is a private backend or one that holds user data. */
  private boolean isPrivateBackend;


 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -128,6 +128,8 @@
import org.opends.server.replication.protocol.ModifyDnContext;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.OperationContext;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.RoutableMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.UpdateMsg;
@@ -196,7 +198,11 @@
   * This flag prevents the Replication Domain to disable/enable itself when
   * it is the event initiator
   */
  private boolean ignoreBackendInitializationEvent;
  private volatile boolean ignoreBackendInitializationEvent;
  /** Exact backend whose enable may have been ignored while an aborted import was stopping. */
  private volatile String backendAwaitingInitialization;
  /** This domain must not resume against data whose post-import state could not be loaded. */
  private volatile boolean failedImportCleanup;

  private volatile boolean  serverShutdownRequested;

@@ -279,19 +285,40 @@

  @Override
  public void performBackendPostInitializationProcessing(LocalBackend<?> backend) {
    if (!ignoreBackendInitializationEvent
            && getBackend().getBackendID().equals(backend.getBackendID())) {
      enable();
    // The import invokes this callback on its listener. A concurrent disable may hold
    // serviceStateLock while joining that listener, so the outside guard is essential.
    // Backend configuration also holds a global lock: skip foreign domains before locking.
    if (!ignoreBackendInitializationEvent && isBackendForInitialization(backend))
    {
      synchronized (serviceStateLock)
      {
        if (!ignoreBackendInitializationEvent && disabled && isBackendForInitialization(backend))
        {
          enable();
        }
      }
    }
  }

  private boolean isBackendForInitialization(LocalBackend<?> backend)
  {
    final String awaiting = backendAwaitingInitialization;
    if (awaiting != null)
    {
      // A disabled nested suffix can resolve to its parent until its own backend reopens.
      return awaiting.equals(backend.getBackendID());
    }
    final LocalBackend<?> current = getBackend();
    return current != null && current.getBackendID().equals(backend.getBackendID());
  }

  @Override
  public void performBackendPreFinalizationProcessing(LocalBackend<?> backend) {
    // Do not disable itself during a shutdown
    // And ignore the event if this replica is the event trigger (e.g. importing).
    if (!ignoreBackendInitializationEvent
            && !serverShutdownRequested
            && getBackend().getBackendID().equals(backend.getBackendID())) {
            && isBackendForInitialization(backend)) {
      disable();
    }
  }
@@ -612,7 +639,7 @@
   * backend the replay would apply a change to is deregistered for the length of it.
   * <p>
   * It is what {@link #disabled} is for an import or a restore run on this server, on the
   * one road which does not set that flag: {@link #preBackendImport(LocalBackend)} takes
   * one road which does not set that flag: {@link #preBackendImport(LocalBackend, BackendImportState)} takes
   * the backend away without disabling the domain, because the domain can not stop the
   * session it is importing over. The replay threads read it where they read
   * {@link #disabled}: an attempt made meanwhile is made into no backend, and whatever it
@@ -778,6 +805,7 @@
          {
            wait(1000);
          }
          restoreBackendAfterAbortedImport();
          if (!disabled && !ieRunning())
          {
            saveState();
@@ -1063,6 +1091,11 @@
      DSRSShutdownSync dsrsShutdownSync) throws ConfigException
  {
    super(configuration, -1);
    final LocalizableMessage failure = LocalBackend.getWriteFailure(getBaseDN());
    if (failure != null)
    {
      throw new ConfigException(failure);
    }

    this.updateToReplayQueue = updateToReplayQueue;
    this.dsrsShutdownSync = dsrsShutdownSync;
@@ -5622,6 +5655,8 @@
       * it lists after this would be the one left behind.
       */
      remotePendingChanges.clear();
      // An explicit disable wins over a recovery requested by the listener we just joined.
      backendAwaitingInitialization = null;
      /*
       * The recovery from a failed replay is over as well: the change it was asking for
       * is gone with the pending changes, so a leftover request would have a replay thread
@@ -5686,7 +5721,13 @@
   * lock is released before returning for the same reason - what keeps the replay out is
   * the flag, not the lock.
   */
  private void awaitReplayDrained()
  private boolean awaitReplayDrained()
  {
    return awaitReplayDrained(true);
  }

  /** Waits for replay, optionally reporting the consequences of taking the domain down. */
  private boolean awaitReplayDrained(boolean goingDown)
  {
    boolean drained = false;
    boolean interrupted = false;
@@ -5710,7 +5751,7 @@
    if (drained)
    {
      replayWriteLock.unlock();
      return;
      return true;
    }
    /*
     * The change which is being applied may reach the backend without being recorded in the
@@ -5719,18 +5760,20 @@
     * restore, a backend being taken offline - for as long as a backend which stopped
     * answering takes to answer.
     *
     * An interrupted wait is reported as what it is: it says nothing about how long the
     * On shutdown an interrupted wait is reported as what it is; an import reports its own
     * preparation failure instead. An interruption says nothing about how long the
     * replay of this domain takes, and the timeout it never spent would have an operator
     * reading a backend which is slow into it.
     */
    if (interrupted)
    if (goingDown && interrupted)
    {
      logger.warn(WARN_REPLAY_DRAIN_INTERRUPTED, getBaseDN());
    }
    else
    else if (goingDown)
    {
      logger.warn(WARN_REPLAY_NOT_DRAINED, getBaseDN(), replayDrainTimeoutInMs);
    }
    return false;
  }

  /**
@@ -5766,36 +5809,52 @@
   * If there is no such information check if there is a
   * ReplicaUpdateVector entry and translate it into a state
   * and generationId.
   * <p>
   * The messages kept of the last local changes are forgotten first, whether the state is read or not:
   * the data was replaced, by a restore, an import or a total update, and a later recovery would
   * publish changes done on the data it replaced - see {@link #forgetTheLocalChangesTheDataLacks()}.
   * @exception DirectoryException Thrown when an error occurs.
   */
  private void loadDataState() throws DirectoryException
  {
    loadDataState(false);
  }

  /** Loads state after a replacement, optionally refusing every failed read. */
  private void loadDataState(boolean strict) throws DirectoryException
  {
    // Which of them the data lacks is not looked up: the read of the state may fail, and this needs none.
    // A total update also forgets before enabling the backend. This second forget can mark operations
    // on the new data, sending their range back to the historical rebuild. It also covers operations
    // still pending as the state is reloaded.
    pendingChanges.forgetRecentChanges();
    state.clearInMemory();
    state.loadState();
    bringAStateWithNoChangeOfThisReplicaUpToTheData();
    state.loadState(strict);
    bringAStateWithNoChangeOfThisReplicaUpToTheData(strict);
    getGenerator().adjust(state.getMaxCSN(getServerId()));
    forgetTheLocalChangesTheDataLacks();
    forgetTheLocalChangesTheDataLacks(strict);

    // Retrieves the generation ID associated with the data imported
    generationId = loadGenerationId();
    generationId = loadGenerationId(strict);
  }

  /**
   * Drops the deletes kept aside, and the local changes a recovery kept, which the data the
   * ServerState was just read from does not reflect (issue #1221): it was restored from a backup, or
   * imported, since they were done, and a recovery which published them would have the other
   * replicas delete or change entries which this one holds as they were.
   * replicas delete or change entries which this one holds as they were. The messages kept of the
   * last local changes are forgotten for the same reason, by {@link #loadDataState()}.
   * <p>
   * What comes after the newest change of this replica the state holds is not in the data: the state
   * read back covers the changes the historical information of the entries holds, whether the save
   * which was to write them wrote or not (see {@link #bringAStateWithNoChangeOfThisReplicaUpToTheData()})
   * which was to write them wrote or not (see {@link #bringAStateWithNoChangeOfThisReplicaUpToTheData(boolean)})
   * - except for a delete, which leaves no historical information, and which the state misses too when
   * it could not be saved: a delete of the base entry takes the saved state away with it. A delete is
   * therefore dropped only if its entry is in the data. A domain enabled back on its own data loses
   * none of them, unless the search which brings its state up to the data does not go through, which
   * {@code ERR_CANNOT_RECOVER_CHANGES} tells.
   */
  private void forgetTheLocalChangesTheDataLacks()
  private void forgetTheLocalChangesTheDataLacks(boolean strict) throws DirectoryException
  {
    final CSN newest = state.getMaxCSN(getServerId());

@@ -5809,7 +5868,7 @@
    final List<CSN> deletesTheDataLacks = new ArrayList<>();
    for (FakeDelOperation delete : deletes)
    {
      if (holdsTheEntry(delete.getEntryUUID()))
      if (holdsTheEntry(delete.getEntryUUID(), strict))
      {
        deletesTheDataLacks.add(delete.getCSN());
      }
@@ -5822,7 +5881,7 @@
    final List<CSN> keptChangesTheDataLacks = new ArrayList<>();
    for (LDAPUpdateMsg kept : pendingChanges.keptChangesNewerThan(newest))
    {
      if (!(kept instanceof DeleteMsg) || holdsTheEntry(kept.getEntryUUID()))
      if (!(kept instanceof DeleteMsg) || holdsTheEntry(kept.getEntryUUID(), strict))
      {
        keptChangesTheDataLacks.add(kept.getCSN());
      }
@@ -5844,7 +5903,7 @@
   * reflect, and the add of the base entry dropped, while the deletes which removed the suffix are
   * published. The search here starts at the first change the recovery kept, which bounds it.
   */
  private void bringAStateWithNoChangeOfThisReplicaUpToTheData()
  private void bringAStateWithNoChangeOfThisReplicaUpToTheData(boolean strict) throws DirectoryException
  {
    if (state.getMaxCSN(getServerId()) != null)
    {
@@ -5855,7 +5914,7 @@
    {
      return;
    }
    final CSN newest = newestChangeOfThisReplicaInTheDataFrom(kept.get(0).getCSN());
    final CSN newest = newestChangeOfThisReplicaInTheDataFrom(kept.get(0).getCSN(), strict);
    if (newest != null)
    {
      state.update(newest);
@@ -5870,7 +5929,7 @@
   *         changes are then taken for changes the data does not hold, as they are when the search which
   *         brings the state up to them as it is loaded does not go through
   */
  private CSN newestChangeOfThisReplicaInTheDataFrom(CSN from)
  private CSN newestChangeOfThisReplicaInTheDataFrom(CSN from, boolean strict) throws DirectoryException
  {
    final InternalSearchOperation op;
    try
@@ -5879,6 +5938,10 @@
    }
    catch (Exception e)
    {
      if (strict)
      {
        throw new DirectoryException(ResultCode.OTHER, ERR_CANNOT_RECOVER_CHANGES.get(getBaseDN()), e);
      }
      logger.traceException(e);
      logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
      return null;
@@ -5888,6 +5951,10 @@
      // A suffix whose base entry is gone holds no change.
      if (op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
      {
        if (strict)
        {
          throw new DirectoryException(op.getResultCode(), ERR_CANNOT_RECOVER_CHANGES.get(getBaseDN()));
        }
        logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
      }
      return null;
@@ -5912,7 +5979,7 @@
   * which did not go through counts as one which found it: a delete then is not published, rather
   * than published over an entry which may be there.
   */
  private boolean holdsTheEntry(String entryUUID)
  private boolean holdsTheEntry(String entryUUID, boolean strict) throws DirectoryException
  {
    try
    {
@@ -5920,10 +5987,41 @@
    }
    catch (SearchFailedException e)
    {
      if (strict)
      {
        throw new DirectoryException(ResultCode.OTHER, ERR_CANNOT_RECOVER_CHANGES.get(getBaseDN()), e);
      }
      return true;
    }
  }

  /** Rechecks an enable which cleanup may have ignored, from a thread that can join the listener. */
  private void restoreBackendAfterAbortedImport()
  {
    if (backendAwaitingInitialization == null || ignoreBackendInitializationEvent || ieRunning())
    {
      return;
    }
    synchronized (serviceStateLock)
    {
      final String backendID = backendAwaitingInitialization;
      if (backendID != null && disabled && !shutdown.get() && !serverShutdownRequested
          && !failedImportCleanup && !ignoreBackendInitializationEvent && !ieRunning()
          && getServerContext().getBackendConfigManager().hasLocalBackend(backendID))
      {
        try
        {
          enable();
        }
        catch (Exception e)
        {
          // An unsuccessful restart must not terminate the state checkpointer.
          logger.error(ERR_INIT_DOMAIN_RESTORE_FAILED, getBaseDN(), backendID, stackTraceToSingleLineString(e));
        }
      }
    }
  }

  /**
   * Enable back the domain after a previous disable.
   * The domain will connect back to a replication Server and
@@ -5936,6 +6034,12 @@
  {
    synchronized (serviceStateLock)
    {
      if (failedImportCleanup)
      {
        return;
      }
      // Consume the missed event before a new listener can request another recovery.
      backendAwaitingInitialization = null;
      /*
       * Cleared here as well as by disable(): a request made after disable()'s own clear -
       * every road which releases a change asks for the restart, whether or not the domain
@@ -6082,6 +6186,16 @@
    return result;
  }

  /** Saves a recomputed ID, making a failed write fatal after an import. */
  private void saveGenerationId(long id, boolean strict) throws DirectoryException
  {
    final ResultCode result = saveGenerationId(id);
    if (strict && result != ResultCode.SUCCESS && result != ResultCode.NO_SUCH_OBJECT)
    {
      throw new DirectoryException(result, ERR_UPDATING_GENERATION_ID.get(getBaseDN(), result.getName()));
    }
  }

  /**
   * Load the GenerationId from the root entry of the domain
   * from the REPLICATION_GENERATION_ID attribute in database
@@ -6092,6 +6206,12 @@
   */
  private long loadGenerationId() throws DirectoryException
  {
    return loadGenerationId(false);
  }

  /** Loads the generation ID, refusing read or persistence failures after an online import. */
  private long loadGenerationId(boolean strict) throws DirectoryException
  {
    if (logger.isTraceEnabled())
    {
      logger.trace("Attempt to read generation ID from DB " + getBaseDN());
@@ -6116,6 +6236,10 @@
      if (search.getResultCode() != ResultCode.NO_SUCH_OBJECT)
      {
        String errorMsg = search.getResultCode().getName() + " " + search.getErrorMessage();
        if (strict)
        {
          throw new DirectoryException(search.getResultCode(), ERR_SEARCHING_GENERATION_ID.get(getBaseDN(), errorMsg));
        }
        logger.error(ERR_SEARCHING_GENERATION_ID, getBaseDN(), errorMsg);
      }
    }
@@ -6132,6 +6256,10 @@
          if (attr.size()>1)
          {
            String errorMsg = "#Values=" + attr.size() + " Must be exactly 1 in entry " + resultEntry.toLDIFString();
            if (strict)
            {
              throw new DirectoryException(ResultCode.OTHER, ERR_LOADING_GENERATION_ID.get(getBaseDN(), errorMsg));
            }
            logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), errorMsg);
          }
          else if (attr.size() == 1)
@@ -6143,6 +6271,11 @@
            }
            catch(Exception e)
            {
              if (strict)
              {
                throw new DirectoryException(ResultCode.OTHER,
                    ERR_LOADING_GENERATION_ID.get(getBaseDN(), stackTraceToSingleLineString(e)), e);
              }
              logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e));
            }
          }
@@ -6153,7 +6286,7 @@
    if (!found)
    {
      aGenerationId = computeGenerationId();
      saveGenerationId(aGenerationId);
      saveGenerationId(aGenerationId, strict);

      if (logger.isTraceEnabled())
      {
@@ -6347,12 +6480,13 @@
   * @throws DirectoryException
   *           If the backend could not be disabled or locked exclusively.
   */
  private void preBackendImport(LocalBackend<?> backend) throws DirectoryException
  private void preBackendImport(LocalBackend<?> backend, BackendImportState progress) throws DirectoryException
  {
    // Prevent the processing of the backend finalisation event as the import will disable the attached backend
    ignoreBackendInitializationEvent = true;

    // FIXME setBackendEnabled should be part of TaskUtils ?
    progress.backendDisabled = true;
    TaskUtils.disableBackend(backend.getBackendID());

    // Acquire an exclusive lock for the backend.
@@ -6364,6 +6498,7 @@
      logger.error(message);
      throw new DirectoryException(ResultCode.OTHER, message);
    }
    progress.lockHeld = true;
  }

  /**
@@ -6378,6 +6513,7 @@
    LocalBackend<?> backend = getBackend();

    LDIFImportConfig importConfig = null;
    final BackendImportState progress = new BackendImportState();
    ImportExportContext ieCtx = getImportExportContext();
    try
    {
@@ -6389,7 +6525,25 @@
       * it is the session the entries are about to stream over, on this very thread.
       */
      importingData = true;
      awaitReplayDrained();
      boolean drained;
      try
      {
        drained = awaitReplayDrained(false) && state.suspendSaves(replayDrainTimeoutInMs);
      }
      catch (InterruptedException e)
      {
        Thread.currentThread().interrupt();
        drained = false;
      }
      if (!drained)
      {
        final LocalizableMessage reason = Thread.currentThread().isInterrupted()
            ? ERR_INIT_PREPARATION_INTERRUPTED.get(getBaseDN())
            : ERR_INIT_OPERATIONS_NOT_DRAINED.get(getBaseDN());
        logger.error(reason);
        ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.UNAVAILABLE, reason));
        return;
      }

      if (!backend.supports(BackendOperation.LDIF_IMPORT))
      {
@@ -6414,70 +6568,126 @@
          ExistingFileBehavior.OVERWRITE);

      // Process import
      preBackendImport(backend);
      preBackendImport(backend, progress);
      progress.replacingData = true;
      backend.importLDIF(importConfig, getServerContext());
    }
    catch(Exception e)
    {
      if (e instanceof InterruptedException)
      {
        Thread.currentThread().interrupt();
      }
      ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
          ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(e))));
    }
    finally
    {
      if (progress.replacingData)
      {
        // No message of the replaced data may survive a failed cleanup or state reload.
        pendingChanges.forgetRecentChanges();
      }
      try
      {
        // Cleanup
        if (importConfig != null)
        {
          importConfig.close();
          closeBackendImport(backend); // Re-enable backend
          backend = getBackend();
        }
        if (progress.backendDisabled)
        {
          closeBackendImport(backend, progress);
        }
        else
        {
          ignoreBackendInitializationEvent = false;
        }

        loadDataState();
        if (progress.replacingData)
        {
          loadDataState(true);
        }

        if (ieCtx.getException() != null)
        if (progress.replacingData && ieCtx.getException() != null)
        {
          // When an error occurred during an import, most of times
          // the generationId coming in the root entry of the imported data,
          // is not valid anymore (partial data in the backend).
          generationId = computeGenerationId();
          saveGenerationId(generationId);
          saveGenerationId(generationId, true);
        }
      }
      catch (DirectoryException fe)
      catch (Exception fe)
      {
        // If we already catch an Exception it's quite possible
        // that the loadDataState() and setGenerationId() fail
        // so we don't bother about the new Exception.
        // However if there was no Exception before we want
        // to return this Exception to the task creator.
        ieCtx.setExceptionIfNoneSet(new DirectoryException(
            ResultCode.OTHER,
            ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe))));
        if (!progress.replacingData)
        {
          // Preparation changed no data: neither a refused disable nor a refused lock warrants
          // a process-lifetime quarantine. Report cleanup too, preserving the original failure.
          final LocalizableMessage reason = ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe));
          logger.error(reason);
          ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER, reason, fe));
          // A DN lookup could find the parent backend of a disabled nested suffix.
          if (!getServerContext().getBackendConfigManager().hasLocalBackend(backend.getBackendID()))
          {
            // The backend could not be enabled back. Keep the domain stopped until a later backend
            // enable reloads its state. This runs on the listener: it cannot join itself, nor take
            // serviceStateLock while a concurrent disable may hold it and wait for this listener.
            disabled = true;
            stopServiceAfterAbortedImport();
            remotePendingChanges.clear();
            // A backend enable between the availability check and the final flag clear is
            // deliberately ignored by the callback. Recheck it after import completion.
            backendAwaitingInitialization = backend.getBackendID();
            logger.error(ERR_INIT_DOMAIN_AWAITING_BACKEND, getBaseDN(), backend.getBackendID());
          }
        }
        else
        {
          // A successful import and a failed import both need trustworthy cleanup before reuse.
          // Preserve an earlier import error in the terminal reason, rather than hiding this failure.
          final LocalizableMessage reason = ERR_INIT_DOMAIN_STOPPED.get(getBaseDN(),
              ieCtx.getException() == null ? "" : ieCtx.getException().getLocalizedMessage(),
              stackTraceToSingleLineString(fe));
          failedImportCleanup = true;
          disabled = true;
          LocalBackend.quarantineWrites(getBaseDN(), reason);
          logger.error(reason);
          ieCtx.setException(new DirectoryException(ResultCode.OTHER, reason));
          stopServicePermanently();
        }
      }
      finally
      {
        /*
         * The ServerState in memory is the one read from the data now, and the changes
         * listed as pending must not outlive the one they went with, as they do not when
         * the domain is disabled: a change given back while the import ran would stay
         * Once replacement has begun, the changes listed as pending must not outlive their data,
         * as they do not when the domain is disabled: a change given back while the import ran would stay
         * listed, uncommitted and owned by nobody, and a commit stops at the first
         * uncommitted change - the state would never move past it, and the replication
         * server does not send a change again which the state it is given covers. Nothing
         * listed a change meanwhile: the listener thread is the one running this import,
         * and the replay threads gave up every attempt while the flag was set. The restarts
         * they asked for on their way out, while the total update owned the session, and
         * any asked for before it did, go with them, as do the deliveries folded into no
         * warning: the caller starts the session again from the reloaded state. A replay
         * thread which reaches its give-up after this clear finds its change unlisted and
         * and the replay threads gave up every attempt while the flag was set. A replay thread
         * which reaches its give-up after this clear finds its change unlisted and
         * asks for nothing.
         */
        remotePendingChanges.clear();
        if (progress.replacingData)
        {
          remotePendingChanges.clear();
        }
        // The caller reconnects even after abandoning replacement, unless the service was stopped
        // above. That reconnect satisfies requests already made during the drain; if the backend
        // is unavailable, its later enable does so instead. A request made later is a new request.
        sessionRestarts.clear();
        consecutiveSessionRestarts.set(0);
        foldedReplayRetryWarnings.set(0);
        if (!failedImportCleanup)
        {
          state.resumeSaves();
        }
        importingData = false;
        // In particular, an enable refused during preparation must not suppress the later event
        // which restores a temporarily stopped domain. All import bookkeeping is settled first.
        ignoreBackendInitializationEvent = false;
      }
    }

@@ -6487,18 +6697,25 @@
    }
  }

  /** Tracks the resources actually acquired by this import, including failed preparation. */
  private static final class BackendImportState
  {
    private boolean backendDisabled;
    private boolean lockHeld;
    private boolean replacingData;
  }

  /**
   * Make post import operations.
   * @param backend The backend implied in the import.
   * @exception DirectoryException Thrown when an error occurs.
   * Releases the import's lock and enables its backend.
   * @param backend the backend being imported
   * @param progress resources acquired by this import
   * @throws DirectoryException if the lock cannot be released or the backend cannot be enabled
   */
  private void closeBackendImport(LocalBackend<?> backend) throws DirectoryException
  private void closeBackendImport(LocalBackend<?> backend, BackendImportState progress) throws DirectoryException
  {
    String lockFile = LockFileManager.getBackendLockFileName(backend);
    StringBuilder failureReason = new StringBuilder();

    // Release lock
    if (!LockFileManager.releaseLock(lockFile, failureReason))
    if (progress.lockHeld && !LockFileManager.releaseLock(lockFile, failureReason))
    {
      LocalizableMessage message =
          WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.get(backend.getBackendID(), failureReason);
@@ -6506,11 +6723,10 @@
      throw new DirectoryException(ResultCode.OTHER, message);
    }

    progress.lockHeld = false;
    TaskUtils.enableBackend(backend.getBackendID());

    // Restore the processing of backend finalization events.
    progress.backendDisabled = false;
    ignoreBackendInitializationEvent = false;

  }

  /**
@@ -6625,6 +6841,12 @@
  {
    // Check that there is not already a domain with the same DN
    final DN dn = configuration.getBaseDN();
    final LocalizableMessage failure = LocalBackend.getWriteFailure(dn);
    if (failure != null)
    {
      unacceptableReasons.add(failure);
      return false;
    }
    LDAPReplicationDomain domain = MultimasterReplication.findDomain(dn, null);
    if (domain != null && domain.getBaseDN().equals(dn))
    {
@@ -7274,10 +7496,12 @@
   * Build the list of changes that have been processed by this server after the
   * CSN given as a parameter and publish them using the given session.
   * <p>
   * The changes done before the recovery began are rebuilt from the historical information of
   * their entries, in rounds, each one from where the round before it stopped, up to the last of
   * them: each of them has its entry written by then, so the search of a round finds it, or a later
   * change of the same values, which supersedes it. The local changes pushed since the recovery began
   * The changes done before the recovery began are published from the messages the domain kept of its
   * last local changes, as they were done; the ones done before the first of those are rebuilt from
   * the historical information of their entries, in rounds, each one from where the round before it
   * stopped, up to there: each of them has its entry written by then, so the search of a round finds
   * it, or a later change of the same values, which supersedes it - and the replication server never
   * holds the change superseded (issue #1246). The local changes pushed since the recovery began
   * are kept by the pending changes, with their messages, and they are published next, in order. The
   * recovery ends with the round which publishes up to the last local change, no change having been
   * pushed meanwhile; once a round has caught up with the local changes and some were pushed while it
@@ -7399,6 +7623,8 @@
    private final ReplicationBroker recoveryBroker;
    private final Session recoverySession;
    private final RSUpdater publisher;
    /** Whether this run said it rebuilds changes from the historical information: it says so once. */
    private boolean rebuildingReported;

    private MissingChanges(ReplicationBroker recoveryBroker, Session recoverySession, RSUpdater publisher)
    {
@@ -7463,9 +7689,8 @@

    /**
     * Runs one round: publishes the changes done before the recovery began which come after
     * {@code fromCSN}, rebuilt from the historical information, up to the end of a search window;
     * past them, the local changes the recovery kept; or, if it kept too many to keep any, the
     * changes rebuilt from the historical information up to the last local change.
     * {@code fromCSN} - see {@link #publishChangesDone(CSN, CSN)}; past them, the local changes the
     * recovery kept; or, if it kept too many to keep any, the changes done up to the last local change.
     *
     * @param heldBack
     *          whether the round runs while the local changes are held back
@@ -7482,22 +7707,114 @@
      final PendingChanges.KeptChanges keeping = pendingChanges.keptChanges();
      if (keeping == null)
      {
        return publishMissingChanges(fromCSN, null);
        return publishChangesDone(fromCSN, null);
      }
      if (keeping.searchUpTo != null && fromCSN.isOlderThan(keeping.searchUpTo))
      {
        return publishMissingChanges(fromCSN, keeping.searchUpTo);
        return publishChangesDone(fromCSN, keeping.searchUpTo);
      }
      final List<LDAPUpdateMsg> kept = pendingChanges.keptChangesAfter(fromCSN);
      if (kept == null)
      {
        // It kept too many since.
        return publishMissingChanges(fromCSN, null);
        return publishChangesDone(fromCSN, null);
      }
      return publishKeptChanges(fromCSN, kept, heldBack);
    }

    /**
     * Runs a round which publishes the changes of this replica which come after {@code fromCSN}, up to
     * {@code upTo} - where the search of the recovery ends - or, if it is {@code null}, up to the last
     * local change: the messages the domain kept of them, as they were done, or - for the changes done
     * before the first of those - the changes rebuilt from the historical information, up to there.
     * <p>
     * Rebuilt, a change whose values a later one replaced is published as the later one, and the add of
     * an entry deleted since not at all, and the replication server of the session, which takes no change
     * older than the newest one it holds of this replica, never holds them (issue #1246). The domain says
     * so, once per run, in the first round which finds a change or a kept delete - see
     * {@link #publishMissingChanges(CSN, CSN)}: its replication server is further behind than the messages
     * it keeps go back.
     *
     * @return the round, or {@code null} if the domain was shut down meanwhile
     * @throws MissingChangesSearchFailedException
     *           if the search did not go through: the round published nothing
     */
    private MissingChangesRound publishChangesDone(CSN fromCSN, CSN upTo) throws Exception
    {
      final PendingChanges.RecentChanges recent = pendingChanges.recentChangesAfter(fromCSN, upTo);
      if (recent.messages != null)
      {
        return publishRecentChanges(fromCSN, recent, upTo == null);
      }
      // The messages kept go on from where the ones the domain kept begin, if they begin before the end.
      final CSN rebuiltUpTo = recent.keptAfter != null && (upTo == null || recent.keptAfter.isOlderThan(upTo))
          ? recent.keptAfter : upTo;
      return publishMissingChanges(fromCSN, rebuiltUpTo);
    }

    /**
     * Runs a round which publishes the messages the domain kept of its local changes after
     * {@code fromCSN}, in order and as they were done, up to the end the domain read them up to; a round
     * which went up to the last local change ends the recovery if no change was pushed since.
     *
     * @param reachesLastChange
     *          whether the messages go up to the last local change, rather than to where the search of
     *          the recovery ends
     * @return the round, or {@code null} if the domain was shut down meanwhile
     */
    private MissingChangesRound publishRecentChanges(CSN fromCSN, PendingChanges.RecentChanges recent,
        boolean reachesLastChange) throws Exception
    {
      for (LDAPUpdateMsg msg : recent.messages)
      {
        if (shutdown.get())
        {
          return null;
        }
        if (importingData)
        {
          return refusedForTheImport(fromCSN);
        }
        if (!recoveryBroker.publishRecovery(notAssured(msg), recoverySession))
        {
          return refused(fromCSN, msg.getCSN());
        }
        if (msg instanceof DeleteMsg)
        {
          deleteKeptAsidePublished(msg.getCSN());
        }
      }
      final CSN publishedUpTo = recent.upTo != null && recent.upTo.isNewerThan(fromCSN) ? recent.upTo : fromCSN;
      if (logger.isDebugEnabled())
      {
        logger.debug(LocalizableMessage.raw("publish loop of the recent changes >" + fromCSN
            + " published=" + recent.messages.size() + " upTo=" + publishedUpTo));
      }
      final boolean endedRecovery =
          reachesLastChange && !pendingChanges.recoveryUntil(publishedUpTo, this::endRecovery);
      return new MissingChangesRound(publishedUpTo, reachesLastChange, true, endedRecovery);
    }

    /**
     * Returns the message to publish for a local change the domain kept: itself, or - if it was
     * published assured - a copy which is not, as a recovery publishes the changes it rebuilds and the
     * ones it kept: an operation which still waits for the acknowledgement of one gets none, and times
     * out. The message kept was handed to the broker already, so it is not changed in its place: the
     * operation which published it may still wait for its acknowledgement.
     */
    private LDAPUpdateMsg notAssured(LDAPUpdateMsg msg) throws Exception
    {
      if (!msg.isAssured())
      {
        return msg;
      }
      final short version = ProtocolVersion.getCurrentVersion();
      final LDAPUpdateMsg copy = (LDAPUpdateMsg) ReplicationMsg.generateMsg(msg.getBytes(version), version);
      copy.setNotAssured();
      return copy;
    }

    /**
     * A total update replaces the data of this replica: what a recovery would publish meanwhile was
     * done on the data it replaces, and the changes it kept which the new data does not reflect are
     * dropped once the import is over. The round is refused, and the session started again after the
@@ -7581,6 +7898,17 @@
          opsToSend.putAll(replayOperations.subMap(fromCSN, false, bound, true));
        }
      }
      /*
       * A delete is published whole, but its entry may have held earlier changes which the search
       * cannot find. A value replaced by a later kept message may leave no trace here at all. An
       * empty window alone does not warrant a warning: after a restart it can precede the first
       * message kept without containing any change.
       */
      if (!rebuildingReported && !opsToSend.isEmpty())
      {
        rebuildingReported = true;
        logger.warn(WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION, getBaseDN(), fromCSN, lastChange);
      }

      for (FakeOperation opToSend : opsToSend.values())
      {

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java

@@ -424,6 +424,16 @@
    try
    {
      LDAPReplicationDomain rd = createNewDomain(configuration);
      if (rd == null)
      {
        ccr.setResultCode(ResultCode.UNAVAILABLE);
        final LocalizableMessage reason = LocalBackend.getWriteFailure(configuration.getBaseDN());
        if (reason != null)
        {
          ccr.addMessage(reason);
        }
        return ccr;
      }
      if (State.RUNNING.equals(state.get()))
      {
        rd.start();

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java

@@ -19,9 +19,11 @@

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Callable;

@@ -56,9 +58,33 @@
   */
  static final int MAX_KEPT_CHANGES = 10000;
  static final long MAX_KEPT_BYTES = 32L * 1024 * 1024;
  /**
   * How many of the last local changes the domain keeps the messages of at most, and how much of the
   * heap those take at most, as {@link #footprint(LDAPUpdateMsg)} reckons it: a recovery publishes the
   * changes its replication server misses from them, and rebuilds the ones before them from the
   * historical information - see {@link LastLocalChanges}. Under the load of the replication convergence
   * run, a few hundred changes a second, a replication server more than two minutes behind is covered.
   * <p>
   * Each domain written to keeps that much, whether a recovery runs or not, so its bytes are a sixteenth
   * of the heap at most; multiple written domains can keep that share each. A server given a small heap
   * covers a shorter lag. A recovery round holds the messages it publishes besides: while it runs, as much
   * again can stay on the heap.
   */
  static final int MAX_RECENT_CHANGES = 100000;
  static final long MAX_RECENT_BYTES = Math.min(64L * 1024 * 1024, Runtime.getRuntime().maxMemory() / 16);
  /**
   * What a message kept takes besides twice its {@link LDAPUpdateMsg#size()}, which counts its encoded
   * changes once, and its encoding holds them again: its DN, entry UUID and CSN, the objects of its
   * changes, and its place in the map. Measured on the messages of the load of the convergence run,
   * kept as they are here: 880 bytes of heap for a modify whose size() is 168, 1300 for an add whose
   * size() is 370.
   */
  static final int MESSAGE_OVERHEAD = 560;

  /** A map used to store the pending changes. */
  private final TreeMap<CSN, PendingChange> pendingChanges = new TreeMap<>();
  /** Local operations pending at a reload, which may have written to the replaced data. */
  private final Set<CSN> pendingAtDataReload = new HashSet<>();

  /**
   * The {@link CSNGenerator} to use to create new unique CSNs
@@ -81,6 +107,10 @@
  private Recovery recovery;
  /** How many local changes a recovery keeps at most: fewer in the tests which make it run over. */
  private int maxKeptChanges = MAX_KEPT_CHANGES;
  /** The messages of the last local changes, which a recovery publishes as they were done. */
  private final LastLocalChanges lastLocalChanges = new LastLocalChanges();
  /** How many of the last local changes the domain keeps the messages of at most: fewer in tests. */
  private int maxRecentChanges = MAX_RECENT_CHANGES;

  /**
   * Creates a new PendingChanges using the provided CSNGenerator.
@@ -108,6 +138,7 @@
  synchronized void remove(CSN csn)
  {
    pendingChanges.remove(csn);
    pendingAtDataReload.remove(csn);
  }

  /**
@@ -193,8 +224,8 @@
   *         {@code null} if none was pushed or the broker refused it. The announcement that a
   *         replica goes offline is the one message whose delivery the caller must know about:
   *         it is stored nowhere, so nothing publishes it again, while a change the broker
   *         refuses is republished from the historical information of its entry on the next
   *         session.
   *         refuses is republished on the next session, from the message kept of it or, once that
   *         is dropped, from the historical information of its entry.
   */
  synchronized CSN pushCommittedChanges()
  {
@@ -217,6 +248,16 @@
          && op != null
          && !op.isSynchronizationOperation())
      {
        // Kept before the domain state takes it, published or not: a recovery publishes it as it was done.
        if (pendingAtDataReload.contains(firstChange.getCSN()))
        {
          // There is a gap in the messages kept after the reload; earlier ones cannot cover it.
          lastLocalChanges.forget();
        }
        else
        {
          lastLocalChanges.add(firstChange.getCSN(), (LDAPUpdateMsg) msg);
        }
        if (!recoveringOldChanges)
        {
          domain.publish(msg);
@@ -268,6 +309,7 @@

      // false warning: firstEntry will not be null if firstChange is not null
      pendingChanges.remove(firstEntry.getKey());
      pendingAtDataReload.remove(firstChange.getCSN());

      // peek the oldest change
      firstEntry = pendingChanges.firstEntry();
@@ -536,6 +578,185 @@
    }
  }

  /**
   * Returns the messages the domain kept of its local changes which come after the provided CSN, up
   * to the provided one, in order, or tells where the messages it kept begin if they do not go back
   * that far.
   *
   * @param after
   *          where the recovery has got to
   * @param upTo
   *          where the changes to publish end, or {@code null} for the last local change
   * @return the messages, or where the ones kept begin
   */
  synchronized RecentChanges recentChangesAfter(CSN after, CSN upTo)
  {
    return lastLocalChanges.after(after, upTo != null ? upTo : domain.getLastLocalChange());
  }

  /**
   * Forgets the messages kept of the last local changes: the domain read its state from data which
   * may lack their changes - it was restored or imported since - and a recovery which published them
   * would have the other replicas do changes this one does not hold. Local operations still pending
   * may have written to the replaced data even if they have not yet committed, so their messages are
   * not kept when they are pushed. Later messages are kept, and a recovery rebuilds the changes before
   * them from the historical information.
   */
  synchronized void forgetRecentChanges()
  {
    lastLocalChanges.forget();
    for (PendingChange change : pendingChanges.values())
    {
      if (change.getOp() != null && !change.getOp().isSynchronizationOperation())
      {
        pendingAtDataReload.add(change.getCSN());
      }
    }
  }

  /**
   * Sets how many of the last local changes the domain keeps the messages of at most: with none, a
   * recovery rebuilds every change from the historical information.
   *
   * @param maxRecentChanges
   *          how many of the last local changes the domain keeps the messages of at most
   */
  @VisibleForTesting
  synchronized void setMaxRecentChanges(int maxRecentChanges)
  {
    this.maxRecentChanges = maxRecentChanges;
    lastLocalChanges.trim();
  }

  /**
   * The messages of the last local changes, in order, which a recovery publishes as they were done.
   * <p>
   * A recovery publishes the changes its replication server misses of this replica, and rebuilt from
   * the historical information of the entries, those are not the changes: it holds nothing of a change
   * whose values a later one replaced, nor of the add of an entry deleted since. A replication server
   * which took the later changes takes none of the older ones afterwards - from the peer which got them
   * from this replica either, whose changes it takes in order, as a log ordered by CSN - so a directory
   * server which failed over to a replication server behind it, or which the load balancing moved to
   * one, left gaps in its changelog for good (issue #1246). The message of every local change pushed -
   * published, refused by the broker, or left to a recovery - is kept here instead, for as many changes
   * as {@link #maxRecentChanges} and {@link #MAX_RECENT_BYTES} allow, the oldest dropped first.
   * <p>
   * Every local change after {@link #keptAfter} has its message kept: the CSN right before the first one
   * kept, and the newest one dropped once some were. A recovery from further back rebuilds the changes up
   * to there from the historical information. It is not the last local change the domain state holds:
   * the state read back once the data was restored, or once the domain was enabled again, can be older
   * than changes done since the domain started - a delete leaves no historical information to bring it
   * up to - and taking the messages kept for the changes after it skipped those changes.
   */
  private final class LastLocalChanges
  {
    private final TreeMap<CSN, LDAPUpdateMsg> messages = new TreeMap<>();
    private long bytes;
    /**
     * Every local change after this one has its message kept; {@code null} if none was kept since the
     * domain started or forgot them.
     */
    private CSN keptAfter;

    private void add(CSN csn, LDAPUpdateMsg msg)
    {
      if (maxRecentChanges <= 0)
      {
        return;
      }
      if (keptAfter == null)
      {
        keptAfter = justBefore(csn);
      }
      // A change is pushed again when the publish of the push before threw: it is counted once.
      final LDAPUpdateMsg replaced = messages.put(csn, msg);
      bytes += footprint(msg) - (replaced != null ? footprint(replaced) : 0);
      trim();
    }

    private void trim()
    {
      if (maxRecentChanges <= 0)
      {
        forget();
        return;
      }
      while (!messages.isEmpty() && (messages.size() > maxRecentChanges || bytes > MAX_RECENT_BYTES))
      {
        final Entry<CSN, LDAPUpdateMsg> oldest = messages.pollFirstEntry();
        bytes -= footprint(oldest.getValue());
        keptAfter = oldest.getKey();
      }
    }

    private void forget()
    {
      messages.clear();
      bytes = 0;
      keptAfter = null;
    }

    private RecentChanges after(CSN after, CSN upTo)
    {
      if (keptAfter == null)
      {
        return new RecentChanges(null, upTo, null);
      }
      if (after.isOlderThan(keptAfter))
      {
        return new RecentChanges(null, upTo, keptAfter);
      }
      final List<LDAPUpdateMsg> found = upTo != null && after.isOlderThan(upTo)
          ? new ArrayList<>(messages.subMap(after, false, upTo, true).values())
          : new ArrayList<LDAPUpdateMsg>();
      return new RecentChanges(found, upTo, keptAfter);
    }
  }

  /**
   * Returns the CSN right before the provided one among the CSNs of its server: no CSN of that server
   * comes in between. For a sequence number of 0, it is the greatest one of the millisecond before, as
   * {@code Integer.MAX_VALUE} rather than -1: a search window of a recovery may start there, and the
   * index of the historical information reads -1 as the greatest sequence number of its millisecond.
   */
  private static CSN justBefore(CSN csn)
  {
    return csn.getSeqnum() > 0
        ? new CSN(csn.getTime(), csn.getSeqnum() - 1, csn.getServerId())
        : new CSN(csn.getTime() - 1, Integer.MAX_VALUE, csn.getServerId());
  }

  /** Returns how much of the heap a message kept takes, as reckoned for {@link #MAX_RECENT_BYTES}. */
  static long footprint(LDAPUpdateMsg msg)
  {
    return 2L * msg.size() + MESSAGE_OVERHEAD;
  }

  /** What the domain kept of the messages of its local changes from a CSN on, up to another one. */
  static final class RecentChanges
  {
    /**
     * The messages of the local changes after the CSN asked from, up to {@link #upTo}, in order;
     * {@code null} if the messages kept do not go back that far, or the domain keeps none.
     */
    final List<LDAPUpdateMsg> messages;
    /** Where the changes asked for end: the CSN asked for, or the last local change. */
    final CSN upTo;
    /**
     * Every local change after this one has its message kept: where the changes to rebuild from the
     * historical information end when {@link #messages} is {@code null}. {@code null} if no message is
     * kept, and the changes are all to rebuild.
     */
    final CSN keptAfter;

    private RecentChanges(List<LDAPUpdateMsg> messages, CSN upTo, CSN keptAfter)
    {
      this.messages = messages;
      this.upTo = upTo;
      this.keptAfter = keptAfter;
    }
  }

  /** A recovery which keeps the local changes pushed while it runs. */
  static final class KeptChanges
  {

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java

@@ -60,9 +60,11 @@
   private final ServerState state;
   /**
    * Held by the save which is writing the state to the backend. It is taken
    * with {@link Lock#tryLock()} and never waited for: see {@link #save()}.
    * without waiting by save(); the total-update preflight waits for it before replacing data.
    */
   private final Lock saveLock = new ReentrantLock();
   /** A total update must not let a snapshot of the old state reach the replacement data. */
   private volatile boolean savesSuspended;

   /**
    * The attribute name used to store the state in the backend.
@@ -131,7 +133,7 @@
   */
  public void save()
  {
    if (state.isSaved())
    if (savesSuspended || state.isSaved())
    {
      // Nothing to write: stay out of the way of whoever is writing.
      return;
@@ -143,7 +145,7 @@
    }
    try
    {
      if (state.isSaved())
      if (savesSuspended || state.isSaved())
      {
        // The save which just completed carried what this one came to write.
        return;
@@ -180,10 +182,51 @@
  }

  /**
   * Stops new saves and waits for the one already writing, before replacing the data.
   * A timeout leaves saves suspended: the caller must abandon the replacement before resuming them.
   * @param timeoutMillis maximum time to wait for the previous write
   * @return whether no previous write can still reach the backend
   * @throws InterruptedException if the wait is interrupted
   */
  boolean suspendSaves(long timeoutMillis) throws InterruptedException
  {
    savesSuspended = true;
    if (!saveLock.tryLock(timeoutMillis, java.util.concurrent.TimeUnit.MILLISECONDS))
    {
      return false;
    }
    saveLock.unlock();
    return true;
  }

  /** Resumes saves after an abandoned replacement or after its new state has been loaded. */
  void resumeSaves()
  {
    savesSuspended = false;
  }

  /**
   * Load the ServerState from the backing entry in database to memory.
   */
  public void loadState()
  {
    try
    {
      loadState(false);
    }
    catch (DirectoryException e)
    {
      throw new IllegalStateException(e);
    }
  }

  /**
   * Loads state, refusing read failures after an online replacement rather than using stale fallbacks.
   * @param strict whether a failed read must prevent replication from restarting
   * @throws DirectoryException if a required read fails in strict mode
   */
  void loadState(boolean strict) throws DirectoryException
  {
    /*
     * Whatever the state holds on the way in has, as far as this object knows,
     * never been written: what follows only merges in what the backend holds.
@@ -195,14 +238,14 @@
    final boolean hadCSNs = !state.isEmpty();

    // try to load the state from the base entry.
    SearchResultEntry stateEntry = searchBaseEntry();
    SearchResultEntry stateEntry = searchBaseEntry(strict);
    if (stateEntry == null)
    {
      /*
      The base entry does not exist yet in the database or was deleted.
      Try to read the ServerState from the configuration instead.
      */
      stateEntry = searchConfigEntry();
      stateEntry = searchConfigEntry(strict);
    }

    if (stateEntry != null)
@@ -216,7 +259,7 @@
     * updated after the last write of the ServerState.
     * Inconsistencies may append after a crash.
     */
    checkAndUpdateServerState();
    checkAndUpdateServerState(strict);

    if (hadCSNs)
    {
@@ -230,7 +273,7 @@
   *
   * @return The base entry or null if no entry was found;
   */
  private SearchResultEntry searchBaseEntry()
  private SearchResultEntry searchBaseEntry(boolean strict) throws DirectoryException
  {
    // Search the database entry that is used to periodically save the ServerState
    final SearchRequest request = newSearchRequest(baseDN, SearchScope.BASE_OBJECT).addAttribute(REPLICATION_STATE);
@@ -239,7 +282,12 @@
    if (resultCode != ResultCode.SUCCESS
        && resultCode != ResultCode.NO_SUCH_OBJECT)
    {
      logger.error(ERR_ERROR_SEARCHING_RUV, search.getResultCode().getName(), search, search.getErrorMessage(), baseDN);
      if (strict)
      {
        throw new DirectoryException(resultCode,
            ERR_ERROR_SEARCHING_RUV.get(resultCode.getName(), search, search.getErrorMessage(), baseDN));
      }
      logger.error(ERR_ERROR_SEARCHING_RUV, resultCode.getName(), search, search.getErrorMessage(), baseDN);
      return null;
    }
    return getFirstResult(search);
@@ -251,7 +299,7 @@
   *
   * @return The configuration Entry or null if no entry was found;
   */
  private SearchResultEntry searchConfigEntry()
  private SearchResultEntry searchConfigEntry(boolean strict) throws DirectoryException
  {
    try
    {
@@ -260,11 +308,19 @@
          .setSizeLimit(1)
          .addAttribute(REPLICATION_STATE);
      final InternalSearchOperation op = getRootConnection().processSearch(request);
      if (strict && op.getResultCode() != ResultCode.SUCCESS && op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
      {
        throw new DirectoryException(op.getResultCode(),
            ERR_ERROR_SEARCHING_RUV.get(op.getResultCode().getName(), op, op.getErrorMessage(), baseDN));
      }
      return getFirstResult(op);
    }
    catch (DirectoryException e)
    {
      // can not happen
      if (strict)
      {
        throw e;
      }
      return null;
    }
  }
@@ -410,7 +466,7 @@
   * This is done by using the HistoricalCsnOrderingMatchingRule
   * and an ordering index for historical attribute
   */
  private final void checkAndUpdateServerState()
  private void checkAndUpdateServerState(boolean strict) throws DirectoryException
  {
    // Retrieves the entries that have changed since the
    // maxCsn stored in the serverState
@@ -428,13 +484,21 @@
        op = LDAPReplicationDomain.searchForChangedEntries(baseDN,
                serverStateMaxCSN, null);
      }
      catch (Exception  e)
      catch (Exception e)
      {
        if (strict)
        {
          throw new DirectoryException(ResultCode.OTHER, ERR_CANNOT_RECOVER_CHANGES.get(baseDN), e);
        }
        return;
      }

      if (op.getResultCode() != ResultCode.SUCCESS)
      {
        if (strict && op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
        {
          throw new DirectoryException(op.getResultCode(), ERR_CANNOT_RECOVER_CHANGES.get(baseDN));
        }
        // An error happened trying to search for the updates
        // Log an error
        logger.error(ERR_CANNOT_RECOVER_CHANGES, baseDN);

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java

@@ -13,7 +13,7 @@
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2023-2026 3A Systems LLC.
 * Portions Copyright 2023-2026 3A Systems, LLC.
 * Portions Copyright 2025 Wren Security.
 */
package org.opends.server.replication.service;
@@ -236,6 +236,8 @@

  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
  private volatile boolean shutdown;
  /** A backend whose state could not be restored must never reconnect through this broker. */
  private volatile boolean permanentlyStopped;
  /**
   * The stop in progress: set by {@link #stop()} before anything else it does, and cleared to
   * {@code null} by {@link #start()}. While it is set, a publish which finds no window credit by
@@ -366,7 +368,7 @@
  {
    synchronized (startStopLock)
    {
      if (!shutdown)
      if (!shutdown || permanentlyStopped)
      {
        return;
      }
@@ -2300,7 +2302,7 @@
      // Synchronize inside the loop in order to allow shutdown.
      synchronized (startStopLock)
      {
        if (rs.isConnected() || shutdown)
        if (rs.isConnected() || shutdown || permanentlyStopped)
        {
          break;
        }
@@ -2416,7 +2418,7 @@
  {
    boolean done = false;

    while (!done && !shutdown && !gaveUpForTheStop())
    while (!done && !shutdown && !permanentlyStopped && !gaveUpForTheStop())
    {
      if (connectionError)
      {
@@ -2508,7 +2510,7 @@
            Check the session. If it has changed, some disconnection or
            reconnection happened and we need to restart from scratch.
            */
            if (gaveUpForTheStop())
            if (permanentlyStopped || gaveUpForTheStop())
            {
              /*
               * The stop gave up since the loop head, while this publish waited for this lock - for
@@ -2559,7 +2561,7 @@
            Not once the stop gave up, which it may have done while this waited for the lock.
            */
            Session session = connectedRS.get().session;
            if (session != null && !gaveUpForTheStop())
            if (session != null && !permanentlyStopped && !gaveUpForTheStop())
            {
              session.publish(new WindowProbeMsg());
            }
@@ -2659,7 +2661,7 @@
      boolean reconnectOnFailure, boolean returnOnTopoChange)
    throws SocketTimeoutException
  {
    while (!shutdown)
    while (!shutdown && !permanentlyStopped)
    {
      ConnectedRS rs = connectedRS.get();
      if (!rs.isConnected())
@@ -2896,9 +2898,48 @@
    }
  }

  /**
   * Stops a domain with untrusted data state. No offline CSN is announced and buffered messages
   * are discarded rather than drained to the peer. A later configuration change cannot restart it.
   */
  void stopPermanently()
  {
    permanentlyStopped = true;
    final Session session = connectedRS.get().session;
    if (session != null)
    {
      session.abort(new IOException("The total update could not restore its backend state"));
    }
    stop();
  }

  /** Returns whether this broker has been permanently stopped following an import failure. */
  boolean isPermanentlyStopped()
  {
    return permanentlyStopped;
  }

  /** Stop the server. */
  public void stop()
  {
    stop(true);
  }

  /**
   * Stops an aborted import on its listener without waiting for credit that only that listener
   * can receive. The next enable may start a new session; no offline CSN is announced.
   * An ECL-enabled replica therefore leaves its last known CSN in the replication server's
   * consistency point, as on an abrupt disconnect: new change-number assignment can stall across
   * its ECL-enabled domains until this replica reconnects and advances that CSN. Having send credit
   * alone would not make an offline announcement safe here: its locks or socket write can still block.
   */
  void stopAfterAbortedImport()
  {
    stop(false);
  }

  private void stop(boolean announceOffline)
  {
    if (logger.isTraceEnabled() && !shutdown)
    {
      debugInfo("is stopping and will close the connection to RS(" + getRsServerId() + ")");
@@ -2921,13 +2962,33 @@
    try
    {
      // In the try: a watchdog which cannot be started must not leave the broker stopping either.
      if (newStop && stop.bounded)
      if (newStop && stop.bounded && announceOffline)
      {
        new DirectoryThread(() -> endWhatStillHoldsTheStop(stop),
            "Replica DS(" + getServerId() + ") watchdog of the stop for domain \"" + getBaseDN() + "\"")
            .start();
      }
      domain.publishReplicaOfflineMsg();
      if (!announceOffline)
      {
        synchronized (stop)
        {
          if (!stop.ended)
          {
            // End blocked writes too: closing in order could wait for the peer to read forever.
            // stop.end() takes this monitor, so this close cannot reach a later session.
            stop.gaveUp = true;
            final Session session = connectedRS.get().session;
            if (session != null)
            {
              session.abort(new IOException("The total update preparation left its backend unavailable"));
            }
          }
        }
      }
      else if (!permanentlyStopped)
      {
        domain.publishReplicaOfflineMsg();
      }
    }
    finally
    {
@@ -3674,7 +3735,7 @@
   */
  boolean shuttingDown()
  {
    return shutdown;
    return shutdown || permanentlyStopped;
  }

  /**

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java

@@ -1982,12 +1982,14 @@
    */
    replicasWeAreWaitingFor.addAll(getReplicaInfos().keySet());

    final Map<Integer, Long> disconnectedSince = new HashMap<>();
    // An absence bound, not a bound on every possible reconnect: the importer's connection
    // timeout and RS list are not carried in the topology and may differ from this server's.
    final long reconnectGraceNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(10);
    boolean done;
    do
    {
      done = true;
      int reconnectMaxDelayInSec = 10;
      int reconnectWait = 0;
      Iterator<Integer> it = replicasWeAreWaitingFor.iterator();
      while (it.hasNext())
      {
@@ -2006,21 +2008,32 @@
          may be for a long time if it crashed or had been stopped
          may be just the time to reconnect after import : should be short
          */
          if (++reconnectWait<reconnectMaxDelayInSec)
          // Each absent target gets its own grace, across passes. A target stopped after a failed
          // import may never reconnect; it must not retain the exporter's import/export context.
          final long now = System.nanoTime();
          final long since = disconnectedSince.computeIfAbsent(serverId, id -> now);
          if (now - since < reconnectGraceNanos)
          {
            // let's still wait to give a chance to this server to reconnect
            done = false;
          }
          // Else we left enough time to the servers to reconnect
          else
          {
            ieCtx.failureList.add(serverId);
            it.remove();
            disconnectedSince.remove(serverId);
          }
        }
        else
        {
          disconnectedSince.remove(serverId);
          // this server is connected
          if (dsInfo.getStatus() == ServerStatus.FULL_UPDATE_STATUS)
          {
            // this one is still doing the Full Update ... retry later
            done = false;
            break;
            // Still visit the other targets, so their absence is timed while this one imports.
            continue;
          }

          if (dsInfo.getGenerationId() == getGenerationID())
@@ -2722,9 +2735,12 @@
      right here, right now
      we never want retryOnFailure if we fails reconnecting in the restart.
      */
      broker.reStart(false);
      if (!isServicePermanentlyStopped())
      {
        broker.reStart(false);
      }

      if (ieCtx.getException() != null
      if (!isServicePermanentlyStopped() && ieCtx.getException() != null
          && broker.isConnected()
          && initFromTask != null
          && ++ieCtx.attemptCnt < 2)
@@ -2785,7 +2801,7 @@

      try
      {
        if (broker.isConnected() && ieCtx.getException() != null)
        if (!isServicePermanentlyStopped() && broker.isConnected() && ieCtx.getException() != null)
        {
          // Let's notify the exporter
          ErrorMsg errorMsg = new ErrorMsg(requesterServerId,
@@ -3410,6 +3426,10 @@
   */
  public void startListenService()
  {
    if (isServicePermanentlyStopped())
    {
      return;
    }
    synchronized (sessionLock)
    {
      if (listenerThread != null)
@@ -3597,6 +3617,33 @@
    return sessionGeneration;
  }

  /** Stops the broker without joining the listener which may be running this import. */
  protected final void stopServicePermanently()
  {
    broker.stopPermanently();
  }

  /**
   * Stops an import's session reversibly when preparation left its backend unavailable.
   * Called by the listener itself: joining it, or taking locks held by a thread joining it,
   * would deadlock. A later enable joins this listener before starting another session.
   */
  protected final void stopServiceAfterAbortedImport()
  {
    final DirectoryThread listener = listenerThread;
    if (listener != null)
    {
      listener.initiateShutdown();
    }
    broker.stopAfterAbortedImport();
  }

  /** Returns whether a failed import has permanently stopped this service. */
  protected final boolean isServicePermanentlyStopped()
  {
    return broker != null && broker.isPermanentlyStopped();
  }

  /**
   * Restart the Replication service after a {@link #disableService()}.
   * <p>
@@ -3613,8 +3660,18 @@
  {
    synchronized (serviceStateLock)
    {
      if (isServicePermanentlyStopped())
      {
        return;
      }
      synchronized (sessionLock)
      {
        if (listenerThread != null && listenerThread.isShutdownInitiated())
        {
          // An aborted import can stop its own session without joining its listener. Join it
          // while the broker is still stopped, so it cannot consume the new session's messages.
          disableService();
        }
        broker.start();
        startListenService();
      }

 opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java

@@ -427,6 +427,8 @@

      LocalBackendWorkflowElement.checkIfBackendIsWritable(currentBackend,
          this, entryDN, ERR_MODDN_SERVER_READONLY, ERR_MODDN_BACKEND_READONLY);
      LocalBackendWorkflowElement.checkIfBackendIsWritable(currentBackend,
          this, newDN, ERR_MODDN_SERVER_READONLY, ERR_MODDN_BACKEND_READONLY);

      if (noOp)
      {

 opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java

@@ -723,6 +723,12 @@
      LocalizableMessageDescriptor.Arg1<Object> backendMsg)
      throws DirectoryException
  {
    final LocalizableMessage failure = LocalBackend.getWriteFailure(entryDN);
    if (failure != null)
    {
      // This also applies to internal/repair operations and private backends such as cn=schema.
      throw new DirectoryException(ResultCode.UNAVAILABLE, failure);
    }
    if (!backend.isPrivateBackend())
    {
      checkIfWritable(DirectoryServer.getCoreConfigManager().getWritabilityMode(), op, serverMsg, entryDN);

 opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -730,3 +730,25 @@
ERR_UNPUBLISHED_DELETE_DROPPED_334=Domain "%s" dropped the delete of "%s" (%s), which no \
 replication server has taken, from the %d deletes it keeps aside for its recoveries: no recovery \
 publishes it anymore, and the replicas which miss it keep the entry
WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION_335=Domain "%s" keeps no message of its \
 local changes after %s up to %s, which its replication server misses: it publishes what it can \
 rebuild from historical information and the deletes kept aside. The historical information holds \
 nothing of a change whose values a later one replaced, nor of the add of an entry deleted since, \
 so that replication server may never hold those changes

ERR_INIT_DOMAIN_STOPPED_336=Replication and writes for domain "%s" are stopped because the total \
 update could not restore a trustworthy backend state. Earlier import error: %s. Cleanup error: %s. \
 Correct the backend failure and restart the server process (an in-core restart, such as a restart \
 task or stop-ds --restart with connection options, does not clear this); a partial import may \
 require another full initialization. Enabling or recreating the replication domain does not clear this failure
ERR_INIT_OPERATIONS_NOT_DRAINED_337=The total update of domain "%s" was abandoned before replacing \
 its data because replay or a server-state write did not finish. Retry after the outstanding operation completes
ERR_INIT_DOMAIN_AWAITING_BACKEND_338=Replication for domain "%s" is stopped because backend "%s" \
 could not be enabled after preparing a total update. The original data was not replaced. \
 Replication resumes when the backend is available again. Correct the backend failure and disable \
 then enable the backend, or restart the server process
ERR_INIT_PREPARATION_INTERRUPTED_339=The total update of domain "%s" was abandoned before replacing \
 its data because the wait for replay or a server-state write was interrupted
ERR_INIT_DOMAIN_RESTORE_FAILED_340=Automatic restoration of replication for domain "%s" after \
 an aborted total update failed for backend "%s": %s. Correct the failure and disable \
 then enable the backend, or restart the server process

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java

@@ -74,6 +74,7 @@

import static java.util.concurrent.TimeUnit.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.opends.messages.ReplicationMessages.WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.replication.server.changelog.api.DBCursor.KeyMatchingStrategy.*;
import static org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy.*;
@@ -146,6 +147,23 @@
  }

  /**
   * Runs a test as the domain keeps the messages of its last local changes, which a recovery publishes,
   * and as it keeps none, and the recovery rebuilds every change from the historical information.
   */
  @DataProvider
  public Object[][] keepingTheRecentChanges()
  {
    return new Object[][] { { true }, { false } };
  }

  /** Runs a test with each of {@link #keepingTheLocalChanges()} and {@link #keepingTheRecentChanges()}. */
  @DataProvider
  public Object[][] keepingTheLocalAndTheRecentChanges()
  {
    return new Object[][] { { true, true }, { true, false }, { false, true }, { false, false } };
  }

  /**
   * Forgets the replication server of the test: the next one, and the domains of the next tests,
   * would be configured with its port, which nothing listens on anymore.
   */
@@ -209,6 +227,8 @@
    ReplicationServer rs = createReplicationServer();
    // Create Replication Server and Domain
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    // These rebuild the changes from the historical information: the domain keeps no message of them.
    pendingChangesOf(rd1).setMaxRecentChanges(0);

    try
    {
@@ -308,6 +328,8 @@
    ReplicationServer rs = createReplicationServer();
    // Create Replication Server and Domain
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    // These rebuild the changes from the historical information: the domain keeps no message of them.
    pendingChangesOf(rd1).setMaxRecentChanges(0);
    long startTime = TimeThread.getTime();

    try
@@ -774,10 +796,12 @@
   * them, and the deletes kept aside for the recovery were dropped as it ended: the next recovery
   * has to find all of them still. The first change refused is a delete, which only the list it is
   * kept aside in, or the local changes the recovery keeps, can publish again: a modify, the search
   * of the next recovery finds again.
   * of the next recovery finds again - or the messages of the last local changes, if the domain keeps
   * them.
   */
  @Test(dataProvider = "keepingTheLocalChanges")
  public void changesTheBrokerRefusedAreLeftToTheNextRecovery(boolean keeps) throws Exception
  @Test(dataProvider = "keepingTheLocalAndTheRecentChanges")
  public void changesTheBrokerRefusedAreLeftToTheNextRecovery(boolean keeps, boolean keepsTheRecentChanges)
      throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
@@ -789,7 +813,7 @@
      final DN dnB = addPerson("b");
      final CSN rsMaxCSN = rd1.getLastLocalChange();

      startRecovery(rd1, keeps);
      startRecovery(rd1, keeps, keepsTheRecentChanges);
      final List<CSN> done = new ArrayList<>();
      TestCaseUtils.deleteEntry(dnB);
      done.add(rd1.getLastLocalChange());
@@ -837,6 +861,7 @@
      // as the missing change publisher does once it failed
      brokerOf(rd1).setRecoveryRequired(false);

      final int logged = TestCaseUtils.ERROR_TEXT_WRITER.getMessages().size();
      rd1.disable();
      rd1.enable();
      // The broker starts its session before enable() returns: a change done before would have
@@ -845,6 +870,13 @@

      replace(dnA, "description", "in a session which needs no recovery");
      assertInTheChangelog(rs, baseDN, rd1.getLastLocalChange());
      // The publisher which ended the recovery had no change to rebuild, and says none (issue #1246).
      final List<String> records = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
      assertThat(records.subList(logged, records.size()))
          .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
              + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
          .as("the publisher which ended the recovery says it rebuilds changes")
          .isEmpty();
    }
    finally
    {
@@ -1287,6 +1319,193 @@
  }

  /**
   * The changes done before a recovery began are published as they were done, from the messages the
   * domain keeps of its last local changes: a change whose values a later one replaced, and the add of
   * an entry deleted since, of which the historical information holds nothing. A recovery which
   * rebuilt them from it published neither, and the replication server it was for took neither once it
   * held the later changes: a directory server which failed over to a replication server behind it
   * left gaps in its changelog for good (issue #1246).
   */
  @Test(dataProvider = "keepingTheLocalChanges")
  public void theChangesDoneBeforeTheRecoveryBeganArePublishedAsTheyWereDone(boolean keeps) throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      final DN dnA = addPerson("a");
      final CSN rsMaxCSN = rd1.getLastLocalChange();

      final List<CSN> done = new ArrayList<>();
      replace(dnA, "description", "first");
      done.add(rd1.getLastLocalChange());
      replace(dnA, "description", "second");
      done.add(rd1.getLastLocalChange());
      final DN dnB = addPerson("b");
      done.add(rd1.getLastLocalChange());
      TestCaseUtils.deleteEntry(dnB);
      done.add(rd1.getLastLocalChange());
      startRecovery(rd1, keeps, true);

      final List<ReplicationMsg> published = new ArrayList<>();
      assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
      assertThat(csns(published)).containsExactlyElementsOf(done);
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  /**
   * A directory server which fails over to a replication server behind it has it hold every change it
   * misses, as it was done: a change whose values a later one replaced, and the add of an entry deleted
   * since, included. Rebuilt from the historical information, neither reached it, and it took neither
   * afterwards, as it held later changes of this replica - the gaps the replication convergence run
   * found in the changelogs (issue #1246).
   */
  @Test
  public void aReplicationServerBehindGetsEveryChangeItMissesAsItWasDone() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    final SortedSet<String> towardsRs = new TreeSet<>(replServers);
    final int secondPort = TestCaseUtils.findFreePort();
    final SortedSet<String> towardsSecond = towards(secondPort);
    ReplicationServer second = null;
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      second = createIsolatedReplicationServer(secondPort, "HistoricalCsnOrderingSecond", SECOND_RS_ID);
      final DN dnA = addPerson("a");
      aChangeBothReplicationServersHold(rd1, dnA, rs, towardsRs, second, towardsSecond);

      // Done on the first replication server only.
      final List<CSN> done = new ArrayList<>();
      replace(dnA, "description", "first");
      done.add(rd1.getLastLocalChange());
      replace(dnA, "description", "second");
      done.add(rd1.getLastLocalChange());
      final DN dnB = addPerson("b");
      done.add(rd1.getLastLocalChange());
      TestCaseUtils.deleteEntry(dnB);
      done.add(rd1.getLastLocalChange());
      for (CSN csn : done)
      {
        assertInTheChangelog(rs, baseDN, csn);
      }

      connectTo(rd1, towardsSecond, SECOND_RS_ID);
      for (CSN csn : done)
      {
        assertInTheChangelog(second, baseDN, csn);
      }
      waitUntilTheRecoveryIsOver(rd1);
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs, second);
    }
  }

  /**
   * A replication server further behind than the messages the domain kept of its last local changes go
   * back gets the changes before the first of them rebuilt from the historical information, and the
   * others as they were done - a change a later one replaced included - and the domain says so, once
   * however many rounds rebuild them: that replication server may never hold some of the changes
   * rebuilt. The recovery starts three search windows back, so that rounds of their own rebuild the
   * changes, each from where the one before stopped.
   */
  @Test
  public void aRecoveryFromBeforeTheMessagesKeptRebuildsTheChangesBeforeThemAndSaysSo() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      final List<CSN> done = new ArrayList<>();
      final DN dnA = addPerson("a");
      done.add(rd1.getLastLocalChange());
      final DN dnB = addPerson("b");
      done.add(rd1.getLastLocalChange());
      final CSN fromCSN = new CSN(rd1.getLastLocalChange().getTime() - 3 * SEARCH_WINDOW_MS, 0, serverId);

      pendingChangesOf(rd1).setMaxRecentChanges(2);
      // Put the last rebuilt change in a different search window from the adds.
      rd1.getGenerator().adjust(new CSN(rd1.getLastLocalChange().getTime() + SEARCH_WINDOW_MS, 0, serverId));
      replace(dnB, "description", "rebuilt");
      done.add(rd1.getLastLocalChange());
      replace(dnA, "description", "first");
      done.add(rd1.getLastLocalChange());
      replace(dnA, "description", "second");
      done.add(rd1.getLastLocalChange());
      startRecovery(rd1, true, true);

      final List<ReplicationMsg> published = new ArrayList<>();
      // A publisher of its own, which has each record once: the error log of the tests has two, whose
      // copies of one record differ when the second of their time stamps ticks in between.
      final List<String> records = errorLogRecordsOf(() -> {
        assertTrue(rd1.buildAndPublishMissingChanges(fromCSN, new TestBroker(published)));
        return null;
      });
      assertThat(csns(published)).containsExactlyElementsOf(done);
      assertThat(records)
          .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
              + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
          .as("the domain says it rebuilds changes from the historical information, once")
          .hasSize(1);
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  /**
   * A change published assured is published by a recovery as one no acknowledgement is waited for, as
   * a recovery publishes the changes it rebuilds and the ones it kept: a copy, not the message the domain
   * kept, which it handed to the broker already and whose acknowledgement the operation may still wait for.
   */
  @Test
  public void aChangePublishedAssuredIsPublishedByARecoveryAsOneNoAcknowledgementIsWaitedFor() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = MultimasterReplication.createNewDomain(new DomainFakeCfg(
        baseDN, serverId, replServers, AssuredType.SAFE_DATA, 1, 1, 1000, null));
    rd1.start();
    try
    {
      final DN dnA = addPerson("a");
      final CSN rsMaxCSN = rd1.getLastLocalChange();
      replace(dnA, "description", "assured");
      final CSN done = rd1.getLastLocalChange();
      startRecovery(rd1, true, true);

      final List<ReplicationMsg> published = new ArrayList<>();
      assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
      assertThat(csns(published)).containsExactly(done);
      assertFalse(((UpdateMsg) published.get(0)).isAssured(), "the recovery publishes it not assured");
      assertTrue(pendingChangesOf(rd1).recentChangesAfter(rsMaxCSN, done).messages.get(0).isAssured(),
          "the message kept is left as it was published");
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  /**
   * The local changes are held back at the end of a recovery for as long as publishing the ones done
   * during the round before takes, and not for a search: the recovery keeps the changes done since it
   * began, and needs none for them. One which searched for them did so while every local write
@@ -1345,10 +1564,11 @@
  /**
   * A recovery which kept too many of the local changes done while it runs keeps none, and searches
   * for them in the historical information instead, from where it has got to: none is lost, and the
   * recovery ends.
   * recovery ends. With the messages of the last local changes kept, it publishes them from those.
   */
  @Test
  public void aRecoveryWhichKeptTooManyChangesSearchesForThemAndLosesNone() throws Exception
  @Test(dataProvider = "keepingTheRecentChanges")
  public void aRecoveryWhichKeptTooManyChangesSearchesForThemAndLosesNone(boolean keepsTheRecentChanges)
      throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
@@ -1362,7 +1582,7 @@
      final CSN rsMaxCSN = rd1.getLastLocalChange();

      pendingChangesOf(rd1).setMaxKeptChanges(2);
      startRecovery(rd1);
      startRecovery(rd1, true, keepsTheRecentChanges);
      // Each change of other values than the ones before, which the historical information keeps.
      final List<CSN> done = new ArrayList<>();
      replace(dnA, "description", "kept");
@@ -1388,10 +1608,13 @@
  /**
   * Local changes go on while a recovery which keeps them runs. The recovery publishes the ones it
   * kept, holds the local changes back only to publish the ones pushed while it published the others,
   * and ends; each change is published once, by the recovery or, once it has ended, by the broker.
   * and ends; each change is published once, by the recovery or, once it has ended, by the broker -
   * whether the changes done before it began are published from the messages the domain kept of them
   * or rebuilt from the historical information.
   */
  @Test
  public void aRecoveryWhichKeepsTheLocalChangesEndsWhileTheyGoOnAndLosesNone() throws Exception
  @Test(dataProvider = "keepingTheRecentChanges")
  public void aRecoveryWhichKeepsTheLocalChangesEndsWhileTheyGoOnAndLosesNone(boolean keepsTheRecentChanges)
      throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
@@ -1407,7 +1630,7 @@
      }
      final CSN rsMaxCSN = rd1.getLastLocalChange();

      startRecovery(rd1);
      startRecovery(rd1, true, keepsTheRecentChanges);
      final List<CSN> done = new CopyOnWriteArrayList<>();
      final Future<?> changes = writer.submit(unchecked(() ->
      {
@@ -1756,10 +1979,11 @@
  /**
   * The local changes a recovery kept are not published once the data they were done on has been
   * replaced, as a restore or an import does while the domain is disabled: the replicas would get
   * changes this one does not hold. The recovery searches the data it finds then instead.
   * changes this one does not hold. The recovery searches the data it finds then instead. The messages
   * the domain kept of its last local changes are forgotten with them (issue #1246).
   */
  @Test
  public void theLocalChangesARecoveryKeptGoWithTheDataTheyWereDoneOn() throws Exception
  @Test(dataProvider = "keepingTheRecentChanges")
  public void theLocalChangesARecoveryKeptGoWithTheDataTheyWereDoneOn(boolean keepsTheRecentChanges) throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
@@ -1776,7 +2000,7 @@

      // The change done then is kept. The broker refuses the announcement that the replica goes
      // offline as well, which would have the replication server hold a CSN past the change.
      startRecovery(rd1);
      startRecovery(rd1, true, keepsTheRecentChanges);
      replace(dnA, "description", "on the data which is replaced");

      // The data is restored while the domain is disabled, as it was before the change.
@@ -1793,6 +2017,159 @@
      assertInTheChangelog(rs, baseDN, afterTheRestore);
      assertThat(changelogAfter(rs, baseDN, rsMaxCSN)).as("the changes published once the data was restored")
          .containsExactly(afterTheRestore);

      // A recovery from where the replication server stood before the change - the one of a session to
      // a replication server behind, say - publishes what the data holds, and not the change either.
      final List<ReplicationMsg> published = new ArrayList<>();
      assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
      assertThat(csns(published)).as("the changes a later recovery publishes").containsExactly(afterTheRestore);
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  /**
   * A recovery which has no change to rebuild - its replication server holds the last local change - says
   * none is rebuilt from the historical information: the publisher which ends the recovery a failed one
   * left runs so, and the domain kept no message for it to publish (issue #1246).
   */
  @Test
  public void aRecoveryWithNoChangeToRebuildSaysNoneIsRebuilt() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      addPerson("a");
      final CSN rsMaxCSN = rd1.getLastLocalChange();
      startRecovery(rd1, false);

      final int logged = TestCaseUtils.ERROR_TEXT_WRITER.getMessages().size();
      final List<ReplicationMsg> published = new ArrayList<>();
      assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
      assertThat(published).isEmpty();
      final List<String> records = TestCaseUtils.ERROR_TEXT_WRITER.getMessages();
      assertThat(records.subList(logged, records.size()))
          .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
              + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
          .as("the recovery says it rebuilds changes")
          .isEmpty();
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  @Test
  public void aRecoveryWarnsWhenOnlyTheDeleteOfAnEntryCanBePublished() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      addPerson("before");
      final CSN from = rd1.getLastLocalChange();
      final DN deleted = addPerson("deleted");
      TestCaseUtils.deleteEntry(deleted);
      final CSN delete = rd1.getLastLocalChange();
      startRecovery(rd1, false);

      final List<ReplicationMsg> published = new ArrayList<>();
      final List<String> records = errorLogRecordsOf(() -> {
        assertTrue(rd1.buildAndPublishMissingChanges(from, new TestBroker(published)));
        return null;
      });
      assertThat(csns(published)).containsExactly(delete);
      assertThat(records).filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
          + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " ")).hasSize(1);
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  @Test
  public void messagesAreForgottenEvenWhenTheStateCannotBeRead() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      addPerson("a");
      final CSN change = rd1.getLastLocalChange();
      final CSN before = new CSN(change.getTime(), change.getSeqnum() - 1, serverId);
      assertThat(pendingChangesOf(rd1).recentChangesAfter(before, change).messages).hasSize(1);
      rd1.disable();

      final MemoryBackend backend =
          (MemoryBackend) getServerContext().getBackendConfigManager().getLocalBackendById(TEST_BACKEND_ID);
      final AttributeType stateType = getServerContext().getSchema().getAttributeType("ds-sync-state");
      final Entry saved = backend.getEntry(baseDN);
      final Entry invalid = saved.duplicate(false);
      final AttributeBuilder state = new AttributeBuilder(stateType);
      state.add("not-a-csn");
      invalid.putAttribute(stateType, newArrayList(state.toAttribute()));
      backend.replaceEntry(saved, invalid, null);

      rd1.enable();

      assertNull(pendingChangesOf(rd1).recentChangesAfter(before, change).messages);
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
      remove(rs);
    }
  }

  /**
   * A recovery from the last change done before a restart, whose replication server misses the changes
   * done since, says none is rebuilt from the historical information: the messages kept begin right before
   * the first change done since, later than the last one done before, and the window in between holds no
   * change of this replica to rebuild (issue #1246).
   */
  @Test
  public void aRecoveryFromTheChangeBeforeTheMessagesKeptSaysNoneIsRebuilt() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
    ReplicationServer rs = createReplicationServer();
    LDAPReplicationDomain rd1 = createReplicationDomain(serverId);
    try
    {
      final DN dnA = addPerson("a");
      final CSN rsMaxCSN = rd1.getLastLocalChange();
      // As after a restart: nothing kept of the changes before, and the time of the generator moved on.
      pendingChangesOf(rd1).forgetRecentChanges();
      rd1.getGenerator().adjust(new CSN(rsMaxCSN.getTime() + 1000, 0, serverId));
      replace(dnA, "description", "since");
      final CSN since = rd1.getLastLocalChange();
      startRecovery(rd1, true, true);

      final List<ReplicationMsg> published = new ArrayList<>();
      final List<String> records = errorLogRecordsOf(() -> {
        assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
        return null;
      });
      assertThat(csns(published)).containsExactly(since);
      assertThat(records)
          .filteredOn(record -> record.contains("category=SYNC severity=WARNING msgID="
              + WARN_LOCAL_CHANGES_REBUILT_FROM_HISTORICAL_INFORMATION.ordinal() + " "))
          .as("the recovery says it rebuilds changes")
          .isEmpty();
    }
    finally
    {
@@ -1986,10 +2363,11 @@
  /**
   * A recovery whose kept changes do not reach the last local change - one which should have been
   * kept is not there - searches for the changes after them, and ends. It went round for ever while
   * it held the local changes back, every local write waiting for it.
   * it held the local changes back, every local write waiting for it. With the messages of the last
   * local changes kept, the change the recovery missed is published from its message.
   */
  @Test
  public void aRecoveryWhoseKeptChangesMissOneSearchesForIt() throws Exception
  @Test(dataProvider = "keepingTheRecentChanges")
  public void aRecoveryWhoseKeptChangesMissOneSearchesForIt(boolean keepsTheRecentChanges) throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
@@ -2001,7 +2379,7 @@
      final DN dnB = addPerson("b");
      final CSN rsMaxCSN = rd1.getLastLocalChange();

      startRecovery(rd1);
      startRecovery(rd1, true, keepsTheRecentChanges);
      final List<CSN> done = new ArrayList<>();
      replace(dnA, "description", "kept");
      done.add(rd1.getLastLocalChange());
@@ -2423,9 +2801,13 @@
   * deletes. They are in the data all the same - their entries are gone - and a recovery publishes
   * them, where taking them for changes the data does not reflect dropped them, and the replicas kept
   * the suffix.
   * <p>
   * With the messages of the last local changes kept, the messages of the deletes are forgotten as the
   * state is read back, and the add of the base entry is the first one kept from there: the deletes,
   * which come after the change the state read back holds, are not taken for kept with it (issue #1246).
   */
  @Test
  public void theDeletesOfASuffixOutliveAStateWhichCouldNotBeSaved() throws Exception
  @Test(dataProvider = "keepingTheRecentChanges")
  public void theDeletesOfASuffixOutliveAStateWhichCouldNotBeSaved(boolean keepsTheRecentChanges) throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    TestCaseUtils.initializeTestBackend(true);
@@ -2438,7 +2820,7 @@
      assertInTheChangelog(rs, baseDN, rsMaxCSN);

      // The replication server misses the deletes.
      refuseTheLocalChanges(rd1);
      refuseTheLocalChanges(rd1, keepsTheRecentChanges);
      final List<CSN> done = new ArrayList<>();
      TestCaseUtils.deleteEntry(dnA);
      done.add(rd1.getLastLocalChange());
@@ -2453,7 +2835,7 @@
      TestCaseUtils.addEntry(StaticUtils.createEntry(baseDN));
      done.add(rd1.getLastLocalChange());
      // Started after the changes, which the recovery then searches for.
      startRecovery(rd1);
      startRecovery(rd1, true, keepsTheRecentChanges);
      final List<ReplicationMsg> published = new ArrayList<>();
      assertTrue(rd1.buildAndPublishMissingChanges(rsMaxCSN, new TestBroker(published)));
      assertThat(csns(published)).containsExactlyElementsOf(done);
@@ -2782,7 +3164,7 @@
   * {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker)} in its
   * place. The pending changes leave the local changes to the recovery, which keeps them, and the
   * broker refuses them. The changes done before are the ones the recovery rebuilds from the
   * historical information.
   * historical information: the domain forgets the messages it kept of them.
   */
  private static void startRecovery(LDAPReplicationDomain domain) throws Exception
  {
@@ -2797,10 +3179,27 @@
   */
  private static void startRecovery(LDAPReplicationDomain domain, boolean keepsTheLocalChanges) throws Exception
  {
    startRecovery(domain, keepsTheLocalChanges, false);
  }

  /**
   * Puts the domain where a session start leaves it when its replication server misses some of its
   * changes, as {@link #startRecovery(LDAPReplicationDomain, boolean)} does, with the messages the
   * domain kept of the changes done before, which the recovery publishes as they were done (issue
   * #1246), or - as the tests of the search of a recovery have it - none, and the recovery rebuilds them
   * from the historical information.
   */
  private static void startRecovery(LDAPReplicationDomain domain, boolean keepsTheLocalChanges,
      boolean keepsTheRecentChanges) throws Exception
  {
    if (!keepsTheLocalChanges)
    {
      pendingChangesOf(domain).setMaxKeptChanges(0);
    }
    if (!keepsTheRecentChanges)
    {
      pendingChangesOf(domain).setMaxRecentChanges(0);
    }
    pendingChangesOf(domain).setRecovering(true);
    brokerOf(domain).setRecoveryRequired(true);
  }
@@ -2809,7 +3208,7 @@
   * Has the broker refuse the local changes, as it does from the start of a session whose replication
   * server misses some, up to the pending changes leaving them to the recovery: they are recorded in
   * the domain state only, and the replication server misses them. The recovery the next session
   * starts rebuilds them from the historical information.
   * starts rebuilds them from the historical information: the domain keeps no message of them.
   * <p>
   * Once the missing change publisher of an earlier session has left, under the lock it leaves under:
   * it has the broker take the local changes again as it leaves, after the end of its recovery - which
@@ -2818,6 +3217,21 @@
   */
  private static void refuseTheLocalChanges(final LDAPReplicationDomain domain) throws Exception
  {
    refuseTheLocalChanges(domain, false);
  }

  /**
   * Has the broker refuse the local changes, as {@link #refuseTheLocalChanges(LDAPReplicationDomain)}
   * does, the domain keeping the messages of the changes it refuses, which the recovery the next session
   * starts publishes as they were done (issue #1246), or none.
   */
  private static void refuseTheLocalChanges(final LDAPReplicationDomain domain, boolean keepsTheRecentChanges)
      throws Exception
  {
    if (!keepsTheRecentChanges)
    {
      pendingChangesOf(domain).setMaxRecentChanges(0);
    }
    final Field recoveryLockField = LDAPReplicationDomain.class.getDeclaredField("recoveryLock");
    recoveryLockField.setAccessible(true);
    final Object recoveryLock = recoveryLockField.get(domain);

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java

@@ -12,6 +12,7 @@
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

@@ -23,11 +24,14 @@
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;

import java.util.Arrays;

import org.forgerock.opendj.ldap.DN;
import org.mockito.ArgumentCaptor;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.UpdateMsg;
@@ -241,6 +245,252 @@
  }

  /**
   * The domain keeps the messages of its last local changes, in order, so that a recovery publishes
   * them as they were done rather than rebuild them from the historical information, which holds
   * nothing of a change a later one replaced (issue #1246).
   */
  @Test
  public void theMessagesOfTheLastLocalChangesAreKeptInOrder() throws Exception
  {
    final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
    final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
    final LDAPUpdateMsg second = pushLocalChange(pendingChanges);
    final LDAPUpdateMsg third = pushLocalChange(pendingChanges);

    assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), third.getCSN()).messages,
        Arrays.asList(second, third));
    assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), second.getCSN()).messages,
        Arrays.asList(second));
  }

  /**
   * The messages kept go back to the CSN right before the first one: a recovery from there - from the
   * change before it, done in the same millisecond - gets the first change as it was done, rather than
   * rebuilt from the historical information.
   */
  @Test
  public void theMessagesKeptGoBackToRightBeforeTheFirstOne() throws Exception
  {
    final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
    final LDAPUpdateMsg first = pushLocalChange(pendingChanges);

    assertEquals(pendingChanges.recentChangesAfter(rightBefore(first), first.getCSN()).messages,
        Arrays.asList(first));
  }

  /**
   * The messages kept of a first change numbered 0 go back to the greatest sequence number of the
   * millisecond before, which no CSN of the server comes after before that change: a recovery from there
   * gets the change as it was done. A CSN which went back to the same millisecond would come after the
   * change, and the recovery would rebuild it from the historical information instead.
   */
  @Test
  public void theMessagesKeptOfAChangeNumberedZeroGoBackToTheMillisecondBefore() throws Exception
  {
    final ServerState state = new ServerState();
    // The generator goes on from the greatest sequence number, so its first change wraps to 0.
    state.update(new CSN(0, Integer.MAX_VALUE, SERVER_ID));
    final PendingChanges pendingChanges = new PendingChanges(new CSNGenerator(SERVER_ID, state),
        domainWhichPublishes(true), new ShutdownSyncAnnouncer(new DSRSShutdownSync(), baseDN));
    final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
    assertEquals(first.getCSN().getSeqnum(), 0);

    final CSN justBefore = new CSN(first.getCSN().getTime() - 1, Integer.MAX_VALUE, SERVER_ID);
    assertEquals(pendingChanges.recentChangesAfter(new CSN(0, 0, SERVER_ID), first.getCSN()).keptAfter,
        justBefore);
    assertEquals(pendingChanges.recentChangesAfter(justBefore, first.getCSN()).messages,
        Arrays.asList(first));
  }

  /**
   * Past as many changes as it keeps at most, the domain drops the oldest messages, and tells where
   * the ones it kept begin: a recovery rebuilds the changes before that from the historical
   * information.
   */
  @Test
  public void pastTheMostChangesKeptTheOldestAreDropped() throws Exception
  {
    final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
    pendingChanges.setMaxRecentChanges(2);
    final LDAPUpdateMsg first = pushLocalChange(pendingChanges);
    final LDAPUpdateMsg second = pushLocalChange(pendingChanges);
    final LDAPUpdateMsg third = pushLocalChange(pendingChanges);

    assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), third.getCSN()).messages,
        Arrays.asList(second, third));
    final PendingChanges.RecentChanges fromBefore =
        pendingChanges.recentChangesAfter(new CSN(0, 0, SERVER_ID), third.getCSN());
    assertNull(fromBefore.messages, "the messages kept do not go back that far");
    assertEquals(fromBefore.keptAfter, first.getCSN());
  }

  /** Past as much of the heap as their messages take at most, the domain drops the oldest messages. */
  @Test
  public void pastTheMostBytesKeptTheOldestAreDropped() throws Exception
  {
    assertTrue(PendingChanges.MAX_RECENT_BYTES <= Runtime.getRuntime().maxMemory() / 16,
        "a domain must not reserve more than a sixteenth of the heap for recent messages");
    final PendingChanges pendingChanges = newPendingChanges(domainWhichPublishes(true));
    final int half = sizeTaking(PendingChanges.MAX_RECENT_BYTES / 2);
    final LDAPUpdateMsg first = pushLocalChange(pendingChanges, half);
    final LDAPUpdateMsg second = pushLocalChange(pendingChanges, half);
    assertEquals(pendingChanges.recentChangesAfter(rightBefore(first), second.getCSN()).messages,
        Arrays.asList(first, second));

    final LDAPUpdateMsg third = pushLocalChange(pendingChanges, 0);

    assertEquals(pendingChanges.recentChangesAfter(first.getCSN(), third.getCSN()).messages,
        Arrays.asList(second, third));
    assertEquals(pendingChanges.recentChangesAfter(rightBefore(first), third.getCSN()).keptAfter, first.getCSN());
  }

  /**
   * A change pushed again - the publish of the push before threw, and left it pending - is counted
   * once: counted twice, it took more of the heap than it does, and was dropped though it was the only
   * message kept.
   */
  @Test
  public void aChangePushedAgainIsCountedOnce() throws Exception
  {
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    when(domain.publish(any(UpdateMsg.class))).thenThrow(new IllegalStateException("the publish failed"))
        .thenReturn(true);
    final PendingChanges pendingChanges = newPendingChanges(domain);
    final CSN csn = pendingChanges.putLocalOperation(newLocalOperation());
    final LDAPUpdateMsg large = newMsg(csn, sizeTaking(PendingChanges.MAX_RECENT_BYTES * 3 / 5));
    try
    {
      pendingChanges.commitAndPushCommittedChanges(csn, large);
      fail("the publish threw");
    }
    catch (IllegalStateException expected)
    {
      // The change is still pending, and the next push pushes it again.
    }

    final LDAPUpdateMsg next = pushLocalChange(pendingChanges, 0);

    assertEquals(pendingChanges.recentChangesAfter(rightBefore(large), next.getCSN()).messages,
        Arrays.asList(large, next));
  }

  /**
   * A change the pending changes leave to a recovery is kept as well: the recovery of a later
   * session, whose replication server misses it, publishes it as it was done.
   */
  @Test
  public void aChangeLeftToARecoveryIsKeptAsWell() throws Exception
  {
    final ReplicationDomain domain = domainWhichPublishes(true);
    when(domain.getServerState()).thenReturn(new ServerState());
    final PendingChanges pendingChanges = newPendingChanges(domain);
    final LDAPUpdateMsg published = pushLocalChange(pendingChanges);
    pendingChanges.setRecovering(true);

    final LDAPUpdateMsg kept = pushLocalChange(pendingChanges);

    assertEquals(onlyMsgPublishedBy(domain), published);
    assertEquals(pendingChanges.recentChangesAfter(published.getCSN(), kept.getCSN()).messages,
        Arrays.asList(kept));
  }

  /**
   * The messages kept are forgotten once the domain reads its state from data which may lack their
   * changes, and none of the changes before is taken for kept: the state read back can be older than
   * changes done since the domain started. The messages of the changes pushed from there are kept.
   */
  @Test
  public void theMessagesKeptAreForgotten() throws Exception
  {
    final ReplicationDomain domain = domainWhichPublishes(true);
    final PendingChanges pendingChanges = newPendingChanges(domain);
    final CSN first = pushLocalChange(pendingChanges).getCSN();
    final CSN last = pushLocalChange(pendingChanges).getCSN();
    // The state read back is older than the last change.
    when(domain.getLastLocalChange()).thenReturn(first);

    pendingChanges.forgetRecentChanges();

    final PendingChanges.RecentChanges fromFirst = pendingChanges.recentChangesAfter(first, last);
    assertNull(fromFirst.messages, "the messages were forgotten");
    assertNull(fromFirst.keptAfter, "and the change after the last one of the state is not taken for kept");

    final LDAPUpdateMsg next = pushLocalChange(pendingChanges);
    final LDAPUpdateMsg afterNext = pushLocalChange(pendingChanges);
    assertNull(pendingChanges.recentChangesAfter(first, afterNext.getCSN()).messages,
        "the change forgotten is not taken for kept");
    assertEquals(pendingChanges.recentChangesAfter(next.getCSN(), afterNext.getCSN()).messages,
        Arrays.asList(afterNext));
  }

  @Test
  public void aCommittedChangeHeldBehindAnotherIsNotKeptAfterTheDataWasReplaced() throws Exception
  {
    final PendingChanges pending = newPendingChanges(domainWhichPublishes(false));
    final CSN first = pending.putLocalOperation(newLocalOperation());
    final CSN second = pending.putLocalOperation(newLocalOperation());
    final LDAPUpdateMsg secondMsg = newMsg(second, 0);
    pending.commitAndPushCommittedChanges(second, secondMsg);

    pending.forgetRecentChanges();
    pending.commitAndPushCommittedChanges(first, newMsg(first, 0));
    final LDAPUpdateMsg after = pushLocalChange(pending);

    assertNull(pending.recentChangesAfter(first, after.getCSN()).messages,
        "the change made on the replaced data must not be treated as kept");
    assertEquals(pending.recentChangesAfter(rightBefore(after), after.getCSN()).messages, Arrays.asList(after));
  }

  @Test
  public void anUncommittedChangeAtDataReloadIsNotKeptAfterItCommits() throws Exception
  {
    final PendingChanges pending = newPendingChanges(domainWhichPublishes(false));
    final CSN old = pending.putLocalOperation(newLocalOperation());
    pending.forgetRecentChanges();
    pending.commitAndPushCommittedChanges(old, newMsg(old, 0));
    final LDAPUpdateMsg after = pushLocalChange(pending);

    assertNull(pending.recentChangesAfter(rightBefore(newMsg(old, 0)), after.getCSN()).messages,
        "a change pending at reload must not be published from the recent messages");
    assertEquals(pending.recentChangesAfter(rightBefore(after), after.getCSN()).messages, Arrays.asList(after));
  }

  /** Pushes a local change whose message takes no bytes, and returns its message. */
  private LDAPUpdateMsg pushLocalChange(PendingChanges pendingChanges)
  {
    return pushLocalChange(pendingChanges, 0);
  }

  /** Pushes a local change whose message has the provided size, and returns its message. */
  private LDAPUpdateMsg pushLocalChange(PendingChanges pendingChanges, int size)
  {
    final CSN csn = pendingChanges.putLocalOperation(newLocalOperation());
    final LDAPUpdateMsg msg = newMsg(csn, size);
    pendingChanges.commitAndPushCommittedChanges(csn, msg);
    return msg;
  }

  private static LDAPUpdateMsg newMsg(CSN csn, int size)
  {
    final LDAPUpdateMsg msg = mock(LDAPUpdateMsg.class);
    when(msg.getCSN()).thenReturn(csn);
    when(msg.size()).thenReturn(size);
    return msg;
  }

  /** Returns the size of a message which takes the provided part of the heap once kept. */
  private static int sizeTaking(long footprint)
  {
    return (int) ((footprint - PendingChanges.MESSAGE_OVERHEAD) / 2);
  }

  /** Returns the CSN right before the one of the message: the change before it, in the same millisecond. */
  private static CSN rightBefore(LDAPUpdateMsg msg)
  {
    return new CSN(msg.getCSN().getTime(), msg.getCSN().getSeqnum() - 1, SERVER_ID);
  }

  /**
   * Reports the forward of the message from within the publish which puts it on the wire, after
   * checking from there that the announcement is already in place, and publishes it. The forward
   * is the one of a peer the message was never recorded as queued for, which is what a forward

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java

@@ -403,6 +403,61 @@
    return new PersistentServerState(baseDn, serverId, new ServerState()).getMaxCSN(serverId);
  }

  @Test
  public void suspendedSavesDrainBeforeDataReplacementAndStaySuspended() throws Exception
  {
    final DN dn = DN.valueOf(TEST_ROOT_DN_STRING);
    final ServerState memory = new ServerState();
    final CountDownLatch writing = new CountDownLatch(1);
    final CountDownLatch release = new CountDownLatch(1);
    final AtomicInteger writes = new AtomicInteger();
    final PersistentServerState state = new HookedWrite(dn, 1, memory, null, entryDN -> {
      if (writes.incrementAndGet() == 1)
      {
        writing.countDown();
        try
        {
          assertTrue(release.await(30, SECONDS));
        }
        catch (InterruptedException e)
        {
          throw new AssertionError(e);
        }
      }
    });
    final CSN first = new CSNGenerator(1, memory).newCSN();
    state.update(first);
    final AtomicReference<Throwable> failure = new AtomicReference<>();
    final Thread writer = new Thread(state::save, "suspended state save");
    writer.setDaemon(true);
    writer.setUncaughtExceptionHandler((thread, error) -> failure.set(error));
    try
    {
      writer.start();
      assertTrue(writing.await(30, SECONDS));
      assertFalse(state.suspendSaves(10), "data must not be replaced while the old write is pending");
      release.countDown();
      writer.join(30_000);
      assertFalse(writer.isAlive());
      assertNull(failure.get());
      assertTrue(state.suspendSaves(1000));
      final CSN newer = new CSNGenerator(2, memory).newCSN();
      state.update(newer);
      state.save();
      assertEquals(writes.get(), 1, "a suspended save must not write stale state over imported data");
      assertNull(loadMaxCSN(dn, 2));
      state.resumeSaves();
      state.save();
      assertEquals(loadMaxCSN(dn, 2), newer);
    }
    finally
    {
      release.countDown();
      writer.join(30_000);
      state.resumeSaves();
    }
  }

  /**
   * A PersistentServerState whose write can be steered from the test, at the
   * point where the snapshot that goes to the backend has already been taken.

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java

@@ -12,6 +12,7 @@
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

@@ -24,7 +25,9 @@
import static org.testng.Assert.*;

import java.util.ArrayList;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
@@ -38,6 +41,7 @@
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.LockFileManager;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
@@ -57,6 +61,7 @@
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
import org.opends.server.tasks.TaskUtils;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -121,6 +126,620 @@
  private ReplicationBroker exporter;
  private CSNGenerator gen;

  @Test(timeOut = 120_000)
  public void aFailedImportUnlockForgetsMessagesOfTheReplacedData() throws Exception
  {
    TestCaseUtils.addEntry("dn: cn=before," + EXAMPLE_DN, "objectClass: top", "objectClass: person",
        "cn: before", "sn: before");
    final CSN change = domain.getLastLocalChange();
    final CSN before = new CSN(change.getTime(), change.getSeqnum() - 1, DS_ID);
    final Field pendingField = LDAPReplicationDomain.class.getDeclaredField("pendingChanges");
    pendingField.setAccessible(true);
    final PendingChanges pending = (PendingChanges) pendingField.get(domain);
    assertThat(pending.recentChangesAfter(before, change).messages).hasSize(1);

    final String lockFile = LockFileManager.getBackendLockFileName(
        getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
    final Field locksField = LockFileManager.class.getDeclaredField("exclusiveLocks");
    final Field mapLockField = LockFileManager.class.getDeclaredField("mapLock");
    locksField.setAccessible(true);
    mapLockField.setAccessible(true);
    final Map<?, ?> locks = (Map<?, ?>) locksField.get(null);
    final Object mapLock = mapLockField.get(null);
    try
    {
      final String[] exported = exportedEntries();
      final CSN importedState = gen.newCSN();
      exported[0] = exported[0].replace("dc: example\n", "dc: example\nds-sync-state: " + importedState + "\n");
      startImportInto(exported.length);
      waitUntil(() -> {
        synchronized (mapLock)
        {
          return locks.containsKey(lockFile);
        }
      }, "the import did not acquire its exclusive lock");
      assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
      finishImport(exported);
      assertNull(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN),
          "the failed unlock must leave the backend disabled");
      assertThat(errorLogRecordsOf(org.opends.messages.ToolMessages.WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.ordinal()))
          .isNotEmpty();
      assertNull(pending.recentChangesAfter(before, change).messages,
          "the import replaced the data even though its lock could not be released");
      assertFalse(domain.isConnected(), "a failed cleanup must stop the replication session");
      TaskUtils.enableBackend("userRoot");
      domain.enable();
      domain.enableService();
      assertFalse(domain.isConnected(), "enabling the backend or service must not revive a failed import");
      assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection()
          .processDelete(IMPORTED_ENTRY_DN).getResultCode(), ResultCode.UNAVAILABLE,
          "writes must be rejected even when the backend is enabled again");
      domain.backupStart();
      domain.disable();
      assertEquals(new PersistentServerState(baseDN, DS_ID,
          new org.opends.server.replication.common.ServerState()).getMaxCSN(importedState.getServerId()),
          importedState, "backup or disable must not overwrite the imported server state");
      MultimasterReplication.deleteDomain(baseDN);
      assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection()
          .processDelete(IMPORTED_ENTRY_DN).getResultCode(), ResultCode.UNAVAILABLE,
          "deleting the replication domain must not permit writes to quarantined data");
      final DomainFakeCfg replacement = new DomainFakeCfg(baseDN, DS_ID, new TreeSet<String>());
      assertNull(MultimasterReplication.createNewDomain(replacement),
          "recreating the domain must not clear the failed-import quarantine");
      assertEquals(new MultimasterReplication().applyConfigurationAdd(replacement).getResultCode(),
          ResultCode.UNAVAILABLE, "configuration add must report failure without dereferencing a missing domain");
    }
    finally
    {
      // A process restart clears the production quarantine; this test shares its JVM with other cases.
      clearWriteFailure();
      final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
      ignored.setAccessible(true);
      ignored.setBoolean(domain, false);
      if (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) == null)
      {
        TaskUtils.enableBackend("userRoot");
      }
    }
  }

  @Test(timeOut = 120_000)
  public void anUnavailableStateReadAfterImportStopsTheDomain() throws Exception
  {
    final String[] exported = exportedEntries();
    startImportInto(exported.length);
    ShortCircuitPlugin.registerShortCircuit(OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(),
        0, 1, operation -> {
          final org.opends.server.types.operation.PreParseSearchOperation search =
              (org.opends.server.types.operation.PreParseSearchOperation) operation;
          return search.getRawBaseDN().toString().equals(EXAMPLE_DN)
              && search.getAttributes().contains("ds-sync-state");
        });
    try
    {
      finishImport(exported);
      assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") > 0,
          "the state-read failure was not injected");
      assertFalse(domain.isConnected(), "an unavailable state read must not fall back to stale configuration state");
      assertNotNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
    }
    finally
    {
      ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
      clearWriteFailure();
    }
  }

  @Test(timeOut = 120_000)
  public void anImportWhichCannotAcquireItsLockDoesNotReleaseAnotherOwnersLock() throws Exception
  {
    TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
        "objectClass: person", "cn: original", "sn: original");
    final String lockFile = LockFileManager.getBackendLockFileName(
        getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
    assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
    try
    {
      final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
      exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
      waitUntil(() -> !domain.ieRunning()
          && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
          "the import did not report its failed exclusive lock");
      assertNotNull(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
      assertTrue(entryExists(DN.valueOf("cn=original," + EXAMPLE_DN)));
      assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
      final Field references = LockFileManager.class.getDeclaredField("sharedLockReferences");
      final Field mapLock = LockFileManager.class.getDeclaredField("mapLock");
      references.setAccessible(true);
      mapLock.setAccessible(true);
      synchronized (mapLock.get(null))
      {
        assertEquals(((Map<?, ?>) references.get(null)).get(lockFile), Integer.valueOf(2),
            "the backend and the other owner must both retain their shared locks");
      }
    }
    finally
    {
      assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
    }
  }

  @org.testng.annotations.DataProvider(name = "failedPreparation")
  public Object[][] failedPreparation()
  {
    return new Object[][] { { false, false }, { true, false }, { true, true } };
  }

  /** A failed preparation must not quarantine data which the import never replaced. */
  @Test(dataProvider = "failedPreparation", timeOut = 120_000)
  public void aFailedPreparationCleanupLeavesTheOriginalDataRecoverable(boolean disabledBackend, boolean nestedBackend) throws Exception
  {
    if (nestedBackend)
    {
      TestCaseUtils.initializeMemoryBackend("importParent", "dc=com", true);
    }
    final Entry original = TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
        "objectClass: person", "cn: original", "sn: original");
    if (disabledBackend)
    {
      // Isolate the abandoned listing from legitimate redelivery of uncheckpointed local adds.
      final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
      stateField.setAccessible(true);
      ((PersistentServerState) stateField.get(domain)).save();
      // A delivery taken from the test queue has no replay thread to finish it. The next
      // session normally redelivers it; this fixture isolates the cleanup of the old listing.
      domain.processUpdate(new ModifyMsg(gen.newCSN(), original.getName(),
          generatemods("description", "before failed preparation"), getEntryUUID(original.getName())));
      queue.take();
    }
    final int stoppedRecords = errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).size();
    final String lockFile = LockFileManager.getBackendLockFileName(
        getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
    boolean lockHeld = disabledBackend;
    if (disabledBackend)
    {
      assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
    }
    ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", ResultCode.UNAVAILABLE.intValue(),
        disabledBackend ? 1 : 0, 2, operation -> {
          final org.opends.server.types.operation.PreParseModifyOperation modify =
              (org.opends.server.types.operation.PreParseModifyOperation) operation;
          return DN.valueOf(modify.getRawEntryDN().toString())
              .equals(DN.valueOf("ds-cfg-backend-id=userRoot,cn=Backends,cn=config"));
        });
    try
    {
      final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
      exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
      waitUntil(() -> !domain.ieRunning()
          && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
          "the failed preparation did not complete");
      assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse") > 0);
      assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN),
          "an import which replaced no data must not quarantine the original suffix");
      ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
      if (disabledBackend)
      {
        assertFalse(domain.isConnected(), "a domain with an unavailable backend must be stopped reversibly");
        assertThat(errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).size()).isGreaterThan(stoppedRecords);
        assertThat(errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).get(stoppedRecords))
            .contains("severity=ERROR", baseDN.toString(), "userRoot", "disable", "enable");
        if (nestedBackend)
        {
          final org.opends.server.api.LocalBackend<?> parent =
              getServerContext().getBackendConfigManager().getLocalBackendById("importParent");
          domain.performBackendPreFinalizationProcessing(parent);
          domain.performBackendPostInitializationProcessing(parent);
          assertFalse(domain.isConnected(), "enabling the parent must not revive the missing child backend");
        }
        else
        {
          // A missing top-level backend must not break another backend's removal.
          TestCaseUtils.initializeMemoryBackend("foreignImportFinalization", "o=foreignImportFinalization", true);
          final org.opends.server.api.LocalBackend<?> foreign =
              getServerContext().getBackendConfigManager().getLocalBackendById("foreignImportFinalization");
          try
          {
            domain.performBackendPreFinalizationProcessing(foreign);
            assertFalse(domain.isConnected(), "a foreign backend must not restore this domain");
          }
          finally
          {
            foreign.finalizeBackend();
            getServerContext().getBackendConfigManager().deregisterLocalBackend(foreign);
          }
        }
        TaskUtils.enableBackend("userRoot");
        assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
        lockHeld = false;
      }
      waitUntil(domain::isConnected, "the backend enable must restore replication without a process restart");
      assertTrue(entryExists(original.getName()));
      final ModifyMsg after = new ModifyMsg(gen.newCSN(), original.getName(),
          generatemods("description", "after failed preparation"), getEntryUUID(original.getName()));
      replayMsg(after);
      assertTrue(domain.getServerState().cover(after.getCSN()),
          "a delivery abandoned by the old session must not hold back the new session's state");
      TestCaseUtils.addEntry("dn: cn=after," + EXAMPLE_DN,
          "objectClass: person", "cn: after", "sn: after");
      // The restored listener must consume new traffic, not just keep a connected broker.
      startImportInto(exportedEntries().length);
      finishImport(exportedEntries());
      assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
    }
    finally
    {
      ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
      clearWriteFailure();
      final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
      ignored.setAccessible(true);
      ignored.setBoolean(domain, false);
      if (lockHeld)
      {
        assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
      }
      if (!getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
      {
        TaskUtils.enableBackend("userRoot");
      }
      if (nestedBackend)
      {
        final org.opends.server.api.LocalBackend<?> parent =
            getServerContext().getBackendConfigManager().getLocalBackendById("importParent");
        parent.finalizeBackend();
        getServerContext().getBackendConfigManager().deregisterLocalBackend(parent);
      }
    }
  }

  /** Backend callbacks run under the configuration lock and must skip foreign domain locks. */
  @Test
  public void aForeignBackendCallbackDoesNotWaitForThisDomainsServiceLock() throws Exception
  {
    TestCaseUtils.initializeMemoryBackend("foreignImportCallback", "o=foreignImportCallback", true);
    final org.opends.server.api.LocalBackend<?> foreign =
        getServerContext().getBackendConfigManager().getLocalBackendById("foreignImportCallback");
    final Field serviceLock = org.opends.server.replication.service.ReplicationDomain.class
        .getDeclaredField("serviceStateLock");
    serviceLock.setAccessible(true);
    final CountDownLatch completed = new CountDownLatch(1);
    final Thread callback = new Thread(() -> {
      domain.performBackendPostInitializationProcessing(foreign);
      completed.countDown();
    }, "foreign backend initialization callback");
    try
    {
      synchronized (serviceLock.get(domain))
      {
        callback.start();
        assertTrue(completed.await(5, TimeUnit.SECONDS), "a foreign callback waited for the domain's service lock");
      }
    }
    finally
    {
      callback.join(10000);
      assertFalse(callback.isAlive());
      foreign.finalizeBackend();
      getServerContext().getBackendConfigManager().deregisterLocalBackend(foreign);
    }
  }

  @org.testng.annotations.DataProvider(name = "abortedImportStop")
  public Object[][] abortedImportStop()
  {
    return new Object[][] { { true }, { false } };
  }

  /** An enable during cleanup is retained, and stopping on the listener never needs window credit. */
  @Test(dataProvider = "abortedImportStop", timeOut = 120_000)
  public void anAbortedImportStopsWithoutCreditAndRetainsABackendEnable(boolean enableDuringStop) throws Exception
  {
    final String lockFile = LockFileManager.getBackendLockFileName(
        getServerContext().getBackendConfigManager().getLocalBackendById("userRoot"));
    assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
    boolean lockHeld = true;
    final Field brokerField = org.opends.server.replication.service.ReplicationDomain.class.getDeclaredField("broker");
    brokerField.setAccessible(true);
    final ReplicationBroker broker = (ReplicationBroker) brokerField.get(domain);
    final Field stopLockField = ReplicationBroker.class.getDeclaredField("startStopLock");
    stopLockField.setAccessible(true);
    final Field windowField = ReplicationBroker.class.getDeclaredField("sendWindow");
    windowField.setAccessible(true);
    final java.util.concurrent.Semaphore window = (java.util.concurrent.Semaphore) windowField.get(broker);
    final Field disabledField = LDAPReplicationDomain.class.getDeclaredField("disabled");
    disabledField.setAccessible(true);
    final long timeout = MultimasterReplication.getConnectionTimeoutMS();
    final DN provider = DN.valueOf("cn=Multimaster Synchronization,cn=Synchronization Providers,cn=config");
    ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", ResultCode.UNAVAILABLE.intValue(),
        1, 2, operation -> DN.valueOf(((org.opends.server.types.operation.PreParseModifyOperation) operation)
            .getRawEntryDN().toString()).equals(DN.valueOf("ds-cfg-backend-id=userRoot,cn=Backends,cn=config")));
    try
    {
      assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection().processModify(
          modifyRequest(provider,
              org.forgerock.opendj.ldap.ModificationType.REPLACE, "ds-cfg-connection-timeout", "0 ms"))
          .getResultCode(), ResultCode.SUCCESS);
      final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
      synchronized (stopLockField.get(broker))
      {
        exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
        new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
            .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
            .repeatUntilSuccess(() -> assertTrue(disabledField.getBoolean(domain)));
        if (enableDuringStop)
        {
          ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
          TaskUtils.enableBackend("userRoot");
        }
        else
        {
          // No update is needed: hold the actual window shut at the instant stop starts.
          window.drainPermits();
        }
      }
      new org.opends.server.util.TestTimer.Builder().maxSleep(5, TimeUnit.SECONDS)
          .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
          .repeatUntilSuccess(() -> assertTrue(!domain.ieRunning()
              && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
              "the listener must finish import cleanup without receiving window credit"));
      ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
      if (!enableDuringStop)
      {
        TaskUtils.enableBackend("userRoot");
      }
      new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
            .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
          .repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "the backend enable must restore the domain"));
      assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
      lockHeld = false;
      startImportInto(exportedEntries().length);
      finishImport(exportedEntries());
      assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
    }
    finally
    {
      // Release a broken implementation too, so the negative run cannot wedge the test JVM.
      window.release(100000);
      ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
      waitUntil(() -> !domain.ieRunning(), "import cleanup did not finish after credit was restored");
      final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
      ignored.setAccessible(true);
      ignored.setBoolean(domain, false);
      if (!getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
      {
        TaskUtils.enableBackend("userRoot");
      }
      if (lockHeld)
      {
        assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
      }
      assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection().processModify(
          modifyRequest(provider,
              org.forgerock.opendj.ldap.ModificationType.REPLACE, "ds-cfg-connection-timeout", timeout + " ms"))
          .getResultCode(), ResultCode.SUCCESS);
    }
  }

  /** The reconnect after an abandoned import already satisfies replay's outstanding restart. */
  @Test(timeOut = 120_000)
  public void anAbandonedImportConsumesTheReplayRestart() throws Exception
  {
    final Entry original = TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
        "objectClass: person", "cn: original", "sn: original");
    final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
    stateField.setAccessible(true);
    final Object state = stateField.get(domain);
    final Field saveLockField = PersistentServerState.class.getDeclaredField("saveLock");
    saveLockField.setAccessible(true);
    final java.util.concurrent.locks.Lock saveLock = (java.util.concurrent.locks.Lock) saveLockField.get(state);
    final Field importing = LDAPReplicationDomain.class.getDeclaredField("importingData");
    importing.setAccessible(true);
    final CSN csn = gen.newCSN();
    final ModifyMsg msg = new ModifyMsg(csn, original.getName(), generatemods("description", "replayed"),
        getEntryUUID(original.getName()));
    final ModifyMsg queued = new ModifyMsg(gen.newCSN(), original.getName(),
        generatemods("description", "queued before the import"), getEntryUUID(original.getName()));
    domain.processUpdate(queued);
    final LDAPUpdateMsg queuedDelivery = queue.take().getUpdateMessage();
    final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
    domain.setReplayDrainTimeout(3000);
    saveLock.lock();
    try
    {
      exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
      new org.opends.server.util.TestTimer.Builder().maxSleep(30, TimeUnit.SECONDS)
          .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
          .repeatUntilSuccess(() -> assertTrue(importing.getBoolean(domain)));
      replayMsg(msg);
      assertFalse(domain.getServerState().cover(csn), "the replay must be deferred by the import");
      domain.failNextSessionRestarts(1);
      waitUntil(() -> !domain.ieRunning()
          && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
          "the import did not abandon its wait for the state save");
      Thread.sleep(2000);
      assertEquals(domain.getSessionRestartFailuresLeft(), 1,
          "the checkpointer restarted the session which import completion already restarted");
      assertTrue(domain.isConnected());
      replayMsg(msg);
      // This delivery was already queued before the import: it must still be listed, without
      // processUpdate inserting it again. Clearing all pending changes would discard its replay.
      replay(queuedDelivery);
      assertTrue(domain.getServerState().cover(queued.getCSN()));
      assertThat(DirectoryServer.getEntry(original.getName()).getAllAttributes("description")).isNotEmpty();
    }
    finally
    {
      saveLock.unlock();
      domain.failNextSessionRestarts(0);
      domain.setReplayDrainTimeout(30_000);
    }
  }

  @org.testng.annotations.DataProvider(name = "importDrains")
  public Object[][] importDrains()
  {
    return new Object[][] { { false }, { true } };
  }

  @Test(dataProvider = "importDrains", timeOut = 120_000)
  public void aTimedOutDrainLeavesTheOriginalDataUsable(boolean stateSave) throws Exception
  {
    final Entry original = TestCaseUtils.makeEntry("dn: cn=original," + EXAMPLE_DN,
        "objectClass: top", "objectClass: person", "cn: original", "sn: original");
    TestCaseUtils.addEntry(original);
    final Object owner;
    final Field lockField;
    if (stateSave)
    {
      final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
      stateField.setAccessible(true);
      owner = stateField.get(domain);
      lockField = PersistentServerState.class.getDeclaredField("saveLock");
    }
    else
    {
      owner = domain;
      lockField = LDAPReplicationDomain.class.getDeclaredField("replayReadLock");
    }
    lockField.setAccessible(true);
    final java.util.concurrent.locks.Lock lock = (java.util.concurrent.locks.Lock) lockField.get(owner);
    final Object backend = getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN);
    final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
    final int shutdownWarnings = errorLogRecordsOf(WARN_REPLAY_NOT_DRAINED.ordinal()).size();
    final int drainErrors = errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).size();
    domain.setReplayDrainTimeout(100);
    lock.lock();
    try
    {
      exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
      waitUntil(() -> !domain.ieRunning()
          && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
          "the import did not abandon the drain before replacing data");
      assertThat(errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).size()).isGreaterThan(drainErrors);
      assertThat(errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).get(drainErrors))
          .contains("severity=ERROR");
      assertSame(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN), backend);
      assertTrue(entryExists(original.getName()), "a rejected preflight must leave the original data intact");
      assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
      assertThat(errorLogRecordsOf(WARN_REPLAY_NOT_DRAINED.ordinal())).hasSize(shutdownWarnings);
      final ErrorMsg failure = waitForSpecificMsg(exporter, ErrorMsg.class);
      assertEquals(failure.getDetails().toString(), ERR_INIT_OPERATIONS_NOT_DRAINED.get(baseDN).toString());
    }
    finally
    {
      lock.unlock();
      domain.setReplayDrainTimeout(30_000);
    }
    waitUntil(domain::isConnected, "a preflight failure must leave the domain recoverable");
    final String[] exported = exportedEntries();
    startImportInto(exported.length);
    finishImport(exported);
    assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)), "a later import must still work");
  }

  @org.testng.annotations.DataProvider(name = "interruptedPreparation")
  public Object[][] interruptedPreparation()
  {
    return new Object[][] { { false }, { true } };
  }

  @Test(dataProvider = "interruptedPreparation", timeOut = 120_000)
  public void anInterruptedImportDrainReportsTheInterrupt(boolean waitingForSave) throws Exception
  {
    final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
    stateField.setAccessible(true);
    final Object state = stateField.get(domain);
    final Field suspended = PersistentServerState.class.getDeclaredField("savesSuspended");
    suspended.setAccessible(true);
    final Field lockField = waitingForSave
        ? PersistentServerState.class.getDeclaredField("saveLock")
        : LDAPReplicationDomain.class.getDeclaredField("replayReadLock");
    lockField.setAccessible(true);
    final java.util.concurrent.locks.Lock lock =
        (java.util.concurrent.locks.Lock) lockField.get(waitingForSave ? state : domain);
    final Field listenerField = org.opends.server.replication.service.ReplicationDomain.class
        .getDeclaredField("listenerThread");
    listenerField.setAccessible(true);
    final Thread listener = (Thread) listenerField.get(domain);
    final Field importing = LDAPReplicationDomain.class.getDeclaredField("importingData");
    importing.setAccessible(true);
    final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
    final int interrupts = errorLogRecordsOf(ERR_INIT_PREPARATION_INTERRUPTED.ordinal()).size();
    lock.lock();
    try
    {
      exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
      new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
            .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
          .repeatUntilSuccess(() -> {
            assertTrue(importing.getBoolean(domain));
            if (waitingForSave)
            {
              assertTrue(suspended.getBoolean(state), "the replay drain must finish before interruption");
              assertEquals(listener.getState(), Thread.State.TIMED_WAITING);
            }
          });
      listener.interrupt();
      waitUntil(() -> !domain.ieRunning()
          && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
          "the interrupted import did not finish");
      final List<String> records = errorLogRecordsOf(ERR_INIT_PREPARATION_INTERRUPTED.ordinal());
      assertThat(records.subList(interrupts, records.size()))
          .isNotEmpty().allSatisfy(record -> assertThat(record).contains("severity=ERROR", "interrupted"));
      assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
    }
    finally
    {
      lock.unlock();
    }
    waitUntil(domain::isConnected, "an interrupted preparation must leave the domain recoverable");
    startImportInto(exportedEntries().length);
    finishImport(exportedEntries());
    assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
  }

  @Test
  public void writeQuarantineSurvivesProviderRemovalAndRejectsMovesIntoTheSuffix() throws Exception
  {
    TestCaseUtils.addEntry("dn: ou=People," + EXAMPLE_DN, "objectClass: organizationalUnit", "ou: People");
    TestCaseUtils.addEntry("dn: cn=outside," + EXAMPLE_DN,
        "objectClass: person", "cn: outside", "sn: outside");
    final DN quarantined = DN.valueOf("ou=People," + EXAMPLE_DN);
    final java.util.List<org.opends.server.api.SynchronizationProvider<
        org.forgerock.opendj.server.config.server.SynchronizationProviderCfg>> providers =
        new ArrayList<>(getSynchronizationProviders());
    org.opends.server.api.LocalBackend.quarantineWrites(quarantined,
        org.forgerock.i18n.LocalizableMessage.raw("failed import test"));
    try
    {
      for (org.opends.server.api.SynchronizationProvider<?> provider : providers)
      {
        deregisterSynchronizationProvider(provider);
      }
      final org.opends.server.protocols.internal.InternalClientConnection connection =
          org.opends.server.protocols.internal.InternalClientConnection.getRootConnection();
      assertEquals(connection.processDelete(quarantined).getResultCode(), ResultCode.UNAVAILABLE);
      assertEquals(connection.processModifyDN("cn=outside," + EXAMPLE_DN, "cn=outside", true,
          quarantined.toString()).getResultCode(), ResultCode.UNAVAILABLE,
          "a move into quarantined data must be refused even when its source is healthy");
      assertTrue(entryExists(DN.valueOf("cn=outside," + EXAMPLE_DN)));
    }
    finally
    {
      getSynchronizationProviders().addAll(providers);
      final Field failures = org.opends.server.api.LocalBackend.class.getDeclaredField("writeFailures");
      failures.setAccessible(true);
      ((Map<?, ?>) failures.get(null)).remove(quarantined);
    }
  }

  private void clearWriteFailure() throws Exception
  {
    final Field failures = org.opends.server.api.LocalBackend.class.getDeclaredField("writeFailures");
    failures.setAccessible(true);
    ((Map<?, ?>) failures.get(null)).remove(baseDN);
  }

  @BeforeMethod
  public void setUpLocal() throws Exception
  {
@@ -1120,7 +1739,7 @@
    exporter.publish(new InitializeTargetMsg(
        baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, entryCount, INIT_WINDOW));
    final long deadline = System.currentTimeMillis() + 30_000;
    while (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) != null)
    while (getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
    {
      assertTrue(System.currentTimeMillis() < deadline,
          "the import did not deregister the backend of the domain");

 opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java

@@ -25,6 +25,9 @@
import static org.testng.Assert.*;

import java.net.SocketTimeoutException;
import java.io.InputStream;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ExecutionException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -1646,6 +1649,194 @@
    }
  }

  /** A failed target must not keep the exporter occupied until an operator restarts its process. */
  @Test
  public void permanentlyStoppedImportersReleaseTheExporter() throws Exception
  {
    final DN baseDN = DN.valueOf("o=test");
    final int port = TestCaseUtils.findFreePort();
    final ReplicationServer rs = createReplicationServer(15, port, "stoppedImportersDb", 100);
    final SortedSet<String> servers = newTreeSet("localhost:" + port);
    FakeReplicationDomain exporter = null;
    FakeReplicationDomain first = null;
    FakeReplicationDomain second = null;
    FakeReplicationDomain healthy = null;
    FutureTask<Void> export = null;
    Thread exportThread = null;
    final CountDownLatch targetsStopped = new CountDownLatch(2);
    try
    {
      exporter = new FakeReplicationDomain(baseDN, 1, servers, 0, buildExportedData(3), null, 3);
      first = stoppingImporter(baseDN, 2, servers, targetsStopped);
      second = stoppingImporter(baseDN, 3, servers, targetsStopped);
      final FakeReplicationDomain source = exporter;
      newTimer().repeatUntilSuccess(() -> assertEquals(source.getReplicaInfos().size(), 2));
      export = new FutureTask<>(() -> {
        source.initializeRemote(org.opends.server.replication.protocol.RoutableMsg.ALL_SERVERS, NO_INIT_TASK);
        return null;
      });
      exportThread = new Thread(export, "export to permanently stopped targets");
      exportThread.start();
      assertTrue(targetsStopped.await(30, SECONDS), "the importers did not finish receiving their data");
      final long stoppedAt = System.nanoTime();
      try
      {
        export.get(20, SECONDS);
        fail("an export to disconnected targets must report failure");
      }
      catch (ExecutionException e)
      {
        assertTrue(e.getCause() instanceof DirectoryException, e.toString());
        assertEquals(((DirectoryException) e.getCause()).getMessageObject().toString(),
            ERR_INIT_NO_SUCCESS_END_FROM_SERVERS.get(source.getGenerationID(), newTreeSet(2, 3)).toString());
      }
      assertTrue(System.nanoTime() - stoppedAt >= SECONDS.toNanos(9),
          "two absent targets must not halve the reconnection grace");
      assertFalse(source.ieRunning(), "failed export retained its context");

      final StringBuffer imported = new StringBuffer();
      healthy = new FakeReplicationDomain(baseDN, 4, servers, 0, null, imported, 0);
      newTimer().repeatUntilSuccess(() -> assertTrue(source.getReplicaInfos().containsKey(4)));
      source.initializeRemote(4, NO_INIT_TASK);
      assertEquals(imported.toString(), buildExportedData(3));
      assertFalse(source.ieRunning());
    }
    finally
    {
      disable(exporter, first, second, healthy);
      if (exportThread != null)
      {
        exportThread.join(30000);
      }
      remove(rs);
    }
  }

  @DataProvider(name = "importingTarget")
  public Object[][] importingTarget()
  {
    // Each target takes a turn importing, so no iteration order of HashSet is assumed.
    return new Object[][] { { 2 }, { 3 }, { 4 } };
  }

  /** Missing peers get independent grace periods even while another peer is still importing. */
  @Test(dataProvider = "importingTarget")
  public void absentTargetsAreTimedIndependentlyWhileAnotherImports(int importing) throws Exception
  {
    final int absent = importing == 2 ? 3 : 2;
    final int later = 9 - importing - absent;
    final EndOfInitWait wait = new EndOfInitWait(2, 3, 4);
    final java.util.concurrent.atomic.AtomicBoolean expiredDuringImport = new java.util.concurrent.atomic.AtomicBoolean();
    wait.topology = elapsed -> {
      if (elapsed < 13000 && wait.failures().contains(absent))
      {
        expiredDuringImport.set(true);
      }
      final Map<Integer, DSInfo> peers = new java.util.HashMap<>();
      peers.put(importing, initTarget(importing,
          elapsed < 13000 ? ServerStatus.FULL_UPDATE_STATUS : ServerStatus.NORMAL_STATUS));
      if (elapsed < 6000 || elapsed >= 12000)
      {
        peers.put(later, initTarget(later,
            elapsed < 6000 ? ServerStatus.FULL_UPDATE_STATUS : ServerStatus.NORMAL_STATUS));
      }
      return peers;
    };
    wait.run();
    assertEquals(wait.failures(), newTreeSet(absent), "the later absence must have its own full grace");
    assertTrue(expiredDuringImport.get(), "a FULL_UPDATE peer prevented timing an absent peer");
  }

  /** Two short absences separated by a reconnect must not be added into one long absence. */
  @Test
  public void aReconnectDuringFullUpdateResetsTheAbsenceTimer() throws Exception
  {
    final EndOfInitWait wait = new EndOfInitWait(2);
    wait.topology = elapsed -> elapsed >= 14000
        ? Collections.singletonMap(2, initTarget(2, ServerStatus.NORMAL_STATUS))
        : elapsed >= 6000 && elapsed < 8000
            ? Collections.singletonMap(2, initTarget(2, ServerStatus.FULL_UPDATE_STATUS))
            : Collections.emptyMap();
    wait.run();
    assertTrue(wait.failures().isEmpty(), "the reconnect did not reset the target's absence timer");
  }

  private static DSInfo initTarget(int id, ServerStatus status)
  {
    return new DSInfo(id, "localhost", 1, 1, status, false,
        org.opends.server.replication.common.AssuredMode.SAFE_READ_MODE, (byte) 1, (byte) 1,
        Collections.emptyList(), Collections.emptySet(), Collections.emptySet(),
        ProtocolVersion.getCurrentVersion());
  }

  /** Runs the real completion loop against a controlled topology, without network timing races. */
  private static final class EndOfInitWait extends DummyReplicationDomain
  {
    private final ImportExportContext context;
    private final java.util.Set<Integer> failures;
    private java.util.function.LongFunction<Map<Integer, DSInfo>> topology;
    private long started;

    @SuppressWarnings("unchecked")
    EndOfInitWait(Integer... targets) throws Exception
    {
      super(1);
      final java.lang.reflect.Constructor<ImportExportContext> constructor =
          ImportExportContext.class.getDeclaredConstructor(boolean.class);
      constructor.setAccessible(true);
      context = constructor.newInstance(false);
      final java.lang.reflect.Field starts = ImportExportContext.class.getDeclaredField("startList");
      starts.setAccessible(true);
      ((java.util.Set<Integer>) starts.get(context)).addAll(java.util.Arrays.asList(targets));
      final java.lang.reflect.Field failed = ImportExportContext.class.getDeclaredField("failureList");
      failed.setAccessible(true);
      failures = (java.util.Set<Integer>) failed.get(context);
      final java.lang.reflect.Field broker = ReplicationDomain.class.getDeclaredField("broker");
      broker.setAccessible(true);
      // Only supplies the external shutdown signal; the production wait and its timers are real.
      broker.set(this, org.mockito.Mockito.mock(ReplicationBroker.class));
    }

    java.util.Set<Integer> failures()
    {
      return failures;
    }

    @Override
    public Map<Integer, DSInfo> getReplicaInfos()
    {
      final long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
      assertTrue(elapsed < 35000, "the completion loop did not release absent targets");
      return topology.apply(elapsed);
    }

    void run() throws Exception
    {
      final java.lang.reflect.Method method = ReplicationDomain.class
          .getDeclaredMethod("waitForRemoteEndOfInit", ImportExportContext.class);
      method.setAccessible(true);
      started = System.nanoTime();
      method.invoke(this, context);
    }
  }

  private FakeReplicationDomain stoppingImporter(DN baseDN, int id, SortedSet<String> servers,
      CountDownLatch stopped) throws Exception
  {
    return new FakeReplicationDomain(baseDN, id, servers, 0, null, new StringBuffer(), 0)
    {
      @Override
      protected void importBackend(InputStream input) throws DirectoryException
      {
        super.importBackend(input);
        stopServicePermanently();
        stopped.countDown();
        throw new DirectoryException(org.forgerock.opendj.ldap.ResultCode.OTHER,
            LocalizableMessage.raw("injected cleanup failure"));
      }
    };
  }

  /** InitializeTask double recording the completion reported by the domain. */
  private static final class RecordingInitializeTask extends InitializeTask
  {