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

Valery Kharseko

11 hours ago c6650aedac5c1ade8aee9e7a76eb0dd100a3e179

[#1246] Keep the messages of the last local changes, and have a recovery publish the changes done before it began from them, as they were done (#1251)

Fixes #1246

## Problem and behavior

When a directory server connects to a replication server that is behind
it, rebuilding missing changes from entry history cannot reproduce every
original operation. A later modify can replace the historical record of
an earlier value, and deleting an entry removes the history of its add.
Once the replication server accepts a newer CSN, its ordered changelog
cannot accept those missing older changes.

The directory server now keeps a bounded buffer of recent local change
messages. Recovery publishes covered changes as they were originally
made, including changes made before recovery began. Older ranges still
fall back to historical reconstruction.

## Implementation

- `PendingChanges` retains local messages whether the broker accepted
them, refused them, or left them to recovery. Repeated attempts to push
the same change do not count it twice.
- Each written domain keeps at most 100,000 messages and the smaller of
64 MiB or one sixteenth of the maximum JVM heap. The memory estimate is
`2 * message.size() + 560`; it is an estimate, not an exact heap
measurement. A recovery snapshot can temporarily retain approximately
another buffer's worth of messages. Multiple written domains each have
their own limit.
- A recovery uses recent messages where coverage is known and
reconstructs older ranges from entry history and retained deletes.
WARNING 335 is emitted once per recovery when a reconstructed round
finds changes or retained deletes. It is not emitted solely because the
recovery starts before the buffer's coverage boundary.
- The boundary before a message with sequence number zero is the
preceding millisecond with `Integer.MAX_VALUE`, preserving the
historical index's ordering.
- Assured messages are republished as non-assured copies. The original
messages remain unchanged for operations still awaiting acknowledgment.

## Data replacement and failed import cleanup

Reloading data state forgets recent messages **before** reading the new
state. It also marks every pending local operation, including operations
not yet marked committed. When such an operation reaches the head of the
queue, its message is not added to the recent buffer; the gap
invalidates coverage and sends that range back to historical
reconstruction. Aborted operations remove their markers.

A total update forgets recent messages before cleanup and again when
loading state after enabling the backend. The second forget can also
mark operations on the new data, conservatively reducing coverage. This
behavior is documented at the call site.

A failed cleanup must not leave a live replication domain using the
replaced data's old state:

- Before replacement, the import waits for replay and an in-flight
server-state save to finish, with a bounded wait. A timeout abandons the
import with `UNAVAILABLE` / error 337 and logs that diagnostic at ERROR
level. An interrupted preparation logs a distinct error 339 and
preserves the interrupt flag. Neither is wrapped in a generic stack
trace or reported as a shutdown-only replay warning. Saves resume on an
abandoned preparation; after replacement, they resume only after
successful cleanup and state loading.
- Cleanup tracks whether this import actually acquired the backend lock.
A failed acquisition does not release another owner's lock.
- State and generation-ID loading after total update treat read failures
as errors rather than silently using a stale fallback. Failure to save a
recomputed generation ID is also reported, except when the suffix has no
base entry.
- A cleanup failure before data replacement reports the failure without
quarantining intact data. Availability is checked by the import's exact
backend ID, so a nested suffix cannot mistake its parent backend for its
own. If the original backend remains available, the domain stays usable.
Otherwise, the domain stops reversibly by aborting the session without
an offline-CSN announcement or a wait for send credit, including with
`connection-timeout: 0`.
- The domain retains the ID of the backend it is waiting for. After
import completion, the state-checkpointer thread rechecks availability
and recovers an enable event ignored during cleanup. Backend callbacks
and this recheck serialize through the service lock; callbacks filter
foreign backends before taking that lock and cannot treat a parent
backend as the missing child. An explicit domain disable clears the
pending recovery. The old listener is joined before a replacement
session starts.
- Pre-finalization callbacks use the same backend identity check as
initialization callbacks. An unavailable top-level backend does not
cause an unrelated backend's removal to throw, and disabling then
enabling a parent backend preserves the awaited child ID. Unexpected
automatic-restoration failures are reported as ERROR 340 with the
domain, backend ID and cause.
- Error 338 names the stopped domain and backend, explains that the
original data was not replaced, and tells the operator to correct the
backend failure and disable then enable the backend, or restart the
server process. The disable/enable sequence also covers an apply-phase
failure that left the configuration marked enabled while the backend was
not opened.
- A cleanup failure after replacement stops the broker permanently,
without an offline-CSN announcement or automatic import retry. It
records an operator-facing error and blocks writes overlapping the
affected suffix until the server process is restarted. An in-core
restart, including a restart task or `stop-ds --restart` with connection
options, does not clear this barrier. Neither does re-enabling the
backend, recreating the replication domain, or removing the
synchronization provider. Rename checks both source and destination.
- Import completion consumes replay restart requests already satisfied
by its reconnect, including when preparation was abandoned. Pending
remote changes are preserved when the original backend remains usable;
they are cleared after replacement or when an unavailable backend
requires stopping the domain.

After a cleanup failure following replacement, the operator must correct
the backend failure before restarting the server process. A partial
import may require another full initialization. Successful imports
resume normal state saving and replication.

The exporter also bounds its end-of-initialization wait for disconnected
targets. Each target gets its own ten-second grace, measured with a
monotonic clock and reset when it reconnects. A target that stays absent
is recorded as failed, allowing the exporter to release its
import/export context. Other targets still in full update do not prevent
tracking absent peers. The ten seconds remain an explicit absence bound,
not a guarantee covering every successful reconnect: the importer's
connection timeout and replication-server list are not available in the
topology and can differ from the exporter's. A sufficiently slow
successful reconnect can therefore still be reported as failed.

## Limits

The recent-message buffer does not provide durable history across
directory-server restarts, and cannot guarantee original operations
outside its coverage. WARNING 335 can fire for a retained delete even
when no earlier operation was lost; an overwritten value that leaves no
trace may still produce no warning. The pending-at-reload markers
protect the new recent-message buffer; they do not redesign the existing
live-publication or recovery queues.

The reversible stop does not force a final ServerState save. Re-enabling
can redeliver changes since the last checkpoint; the unchanged data and
its historical metadata absorb those repeats.

The reversible stop after aborted import preparation deliberately sends
no `ReplicaOfflineMsg`: the listener must not wait for send credit or a
blocked socket write. If the replica participates in the external
changelog, its last known CSN can hold back new change-number assignment
in `cn=changelog` until it reconnects and advances that CSN. The
consistency point is shared across the replication server's ECL-enabled
domains, so other suffixes can also be affected. This has the same
effect on change numbering as an abruptly disconnected directory server;
it does not stop ordinary replication among the remaining servers. A
free send permit alone would not make offline publication non-blocking.

This PR does not change replication-server selection or load balancing.

## Validation

JDK 21 offline Maven reactor: `mvn -o -Pprecommit verify -pl
opendj-server-legacy -am` with the nine integration-test classes below.
The final run passed **194/194 selected cases**, with zero failures,
errors or skips; the reactor, including its upstream-module tests and
Javadoc generation, completed successfully. `git diff --check` passed.

| Test class | Passed / run |
| --- | --- |
| ConfigChangeDuringImportTest | 4 / 4 |
| DisabledDomainServerStateTest | 2 / 2 |
| HistoricalCsnOrderingTest | 70 / 70 |
| PendingChangesTest | 19 / 19 |
| PersistentServerStateTest | 12 / 12 |
| ReplayDuringImportTest | 28 / 28 |
| ReplicationBrokerStopTest | 15 / 15 |
| ServerStateFlushTest | 4 / 4 |
| ReplicationDomainTest | 40 / 40 |

- Before the callback fix, the two extended failed-preparation cases
failed as expected: a foreign backend's pre-finalization threw an NPE
when the top-level backend was absent, and the parent's pre/post
callbacks revived a domain awaiting its nested child. Both now pass and
verify a subsequent successful import.
- Interruption tests cover both replay drain and an in-flight
ServerState save. Each checks fresh ERROR 339 records and completes
another import afterward. A compiled mutant deleting the save-wait
catch's re-interrupt failed the save row; the replay row still passed.
The mutation was compiled outside the working tree and was not included
in the final reactor run.
- All fifteen broker-stop cases passed in the final run, including the
heartbeat case that failed in the previous round. This run does not
establish the cause of that earlier timing failure.
- The selected suites also cover pending-at-reload markers, recovery
coverage and warning frequency, memory limits, sequence-number zero,
save suspension, backend lock ownership, failed import cleanup,
permanent write quarantine, ignored backend enables, empty send credit,
independent export absence timers and reset on reconnect.
- An independent read-only review of this round found no additional
issues.

The convergence scenario `rs534, restart-rs` from #1208 was not rerun in
this round. The completion-loop tests still use real elapsed time;
injecting both a test clock and controlled sleeps is deferred. The
optional assertion of WARNING 335's exact message arguments has not been
added; its occurrence and once-per-recovery behavior are tested.

 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
  {