[#1041] Claim the import context for the length of a session restart, and r...

Valery Kharseko

23 hours ago 45794c50d0b48bd59d0b318f781a7b653f134752

[#1041] Claim the import context for the length of a session restart, and refuse the total update which lands across it (#1045)

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

@@ -4287,13 +4287,17 @@
    final long wakes;
    synchronized (serviceStateLock)
    {
      if (sessionHasAnOwner())
      /*
       * The domain is going away or is being imported into: the session is not this
       * thread's to stop. The total update is claimed against rather than read (issue
       * #1041): the listener thread claims one this replica did not ask for under no lock,
       * and a read here a few statements before that claim would stop the session the
       * import is about to read.
       */
      if (ownsItsSession() || !disableServiceUnlessImportInProgress())
      {
        // The domain is going away or is being imported into: the session is not this
        // thread's to stop.
        return;
      }
      disableService();
      stoppedSession = getSessionGeneration();
      wakes = sessionRestartBackoffWakes();
    }
@@ -6213,6 +6217,11 @@
   * it carries: the domain itself, when it is shutting down or disabled
   * ({@link #ownsItsSession()}), or a total update into this replica.
   * <p>
   * What this reads, {@link #restartSession(boolean)} claims: a total update the listener
   * thread is about to claim is not visible to a read, and the restart must not stop the
   * session such a total update reads (issue #1041). This is the early exit of the roads
   * which lead to that restart, and the answer for the ones which never restart anything.
   * <p>
   * The total update owns the session from the moment it is asked for, not from the
   * moment its entries stream: the {@code InitializeTargetMsg} which answers the request
   * arrives over that session, so a restart made while it is on its way loses it, and the

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

@@ -56,6 +56,7 @@
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType;
import org.forgerock.opendj.server.config.server.ReplicationDomainCfg;
import org.forgerock.util.annotations.VisibleForTesting;
import org.opends.server.api.DirectoryThread;
import org.opends.server.api.MonitorData;
import org.opends.server.backends.task.Task;
@@ -262,6 +263,39 @@
   * Null when none is being processed.
   */
  private final AtomicReference<ImportExportContext> importExportContext = new AtomicReference<>();
  /**
   * Holds {@link #importExportContext} for the length of a session stop which a total update
   * into this replica must not be claimed across (issue #1041).
   * <p>
   * A restart of the session reads whether such a total update owns it before it stops
   * anything, and the listener thread claims the context for the {@code InitializeTargetMsg}
   * it took off the session - and the two share no lock: {@link #disableService()} waits for
   * the listener thread under {@link #serviceStateLock}, so the listener can not take that
   * lock before its claim. A restart which read no owner a few statements before the claim
   * landed stopped the broker the import was about to read, and the import ended on the
   * nothing which arrived - recorded as a failed import since issue #1039, over a suffix
   * which has been replaced by it all the same. The two contend on the one reference
   * instead: the restart claims it with this context, the claim of the listener fails
   * against it, and exactly one of them wins - the total update is either the owner the
   * restart reads, or refused.
   * <p>
   * It is neither an import nor an export: {@link #ieRunning()}, {@link #importInProgress()}
   * and {@link #getImportExportContext()} do not report it.
   */
  private static final ImportExportContext SESSION_BEING_STOPPED = new ImportExportContext(false);
  /**
   * Run by the listener thread between the {@code InitializeTargetMsg} it took off the
   * session and its claim of the import context - or, for a total update this replica asked
   * for, its read of the context the request claimed. Only there for the tests, which hold
   * the listener thread there: nothing else runs in that gap.
   */
  private volatile Runnable importClaimHook;
  /**
   * Run by {@link #disableService()} under its locks, before the broker is stopped. Only
   * there for the tests, which hold a stop of the service there: what the claim of a session
   * stop is for is the total update which lands between the decision to stop and the stop.
   */
  private volatile Runnable serviceStopHook;

  /**
   * The Thread waiting for incoming update messages for this domain and pushing
@@ -838,7 +872,7 @@
        else if (msg instanceof ErrorMsg)
        {
          ErrorMsg errorMsg = (ErrorMsg)msg;
          ImportExportContext ieCtx = importExportContext.get();
          ImportExportContext ieCtx = getImportExportContext();
          if (ieCtx != null)
          {
            /*
@@ -900,7 +934,7 @@
        }
        else if (msg instanceof InitializeRcvAckMsg)
        {
          ImportExportContext ieCtx = importExportContext.get();
          ImportExportContext ieCtx = getImportExportContext();
          if (ieCtx != null)
          {
            InitializeRcvAckMsg ackMsg = (InitializeRcvAckMsg) msg;
@@ -1660,7 +1694,7 @@
      // Release the context whatever the outcome, otherwise ieRunning() would
      // remain true forever (resolves the historical "FIXME should not this
      // be in a finally?").
      releaseIEContext();
      releaseIEContext(ieCtx);
    }
  }

@@ -2036,16 +2070,27 @@
    final ImportExportContext ieCtx = new ImportExportContext(importInProgress);
    if (!importExportContext.compareAndSet(null, ieCtx))
    {
      // Rejects 2 simultaneous exports
      LocalizableMessage message = ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get();
      // Rejects 2 simultaneous exports, and a total update which is claimed while the
      // session is being stopped - in either direction: the entries of an export out of this
      // server are streamed over that session too (see SESSION_BEING_STOPPED)
      final LocalizableMessage message = importExportContext.get() == SESSION_BEING_STOPPED
          ? ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId())
          : ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get();
      throw new DirectoryException(ResultCode.OTHER, message);
    }
    return ieCtx;
  }

  private void releaseIEContext()
  /**
   * Releases the provided import/export context, and only that one: a road which failed to
   * acquire a context of its own must not release the one it failed against - the import or
   * export which owns it, or the claim of a session stop ({@code SESSION_BEING_STOPPED}).
   *
   * @param ieCtx the context to release
   */
  private void releaseIEContext(ImportExportContext ieCtx)
  {
    importExportContext.set(null);
    importExportContext.compareAndSet(ieCtx, null);
  }

  /**
@@ -2059,7 +2104,7 @@
   */
  private void completeInitializeTask(ImportExportContext ieCtx)
  {
    releaseIEContext();
    releaseIEContext(ieCtx);
    if (ieCtx.initializeTask instanceof InitializeTask)
    {
      // Update the task that initiated the import
@@ -2108,7 +2153,7 @@
    ReplicationMsg msg;
    while (true)
    {
      ImportExportContext ieCtx = importExportContext.get();
      ImportExportContext ieCtx = getImportExportContext();
      try
      {
        // In the context of the total update, we don't want any automatic
@@ -2281,7 +2326,7 @@
    }

    // build the message
    ImportExportContext ieCtx = importExportContext.get();
    ImportExportContext ieCtx = getImportExportContext();
    EntryMsg entryMessage = new EntryMsg(
        getServerId(), ieCtx.getExportTarget(), lDIFEntry, pos, length,
        ++ieCtx.msgCnt);
@@ -2425,6 +2470,7 @@
    not processed any topology message in between the failure and the
    new attempt.
    */
    ImportExportContext ieCtx = null;
    try
    {
      /*
@@ -2434,7 +2480,7 @@
      update the task.
      */

      final ImportExportContext ieCtx = acquireIEContext(true);
      ieCtx = acquireIEContext(true);
      ieCtx.initializeTask = initTask;
      ieCtx.attemptCnt = 0;
      ieCtx.initReqMsgSent = new InitializeRequestMsg(
@@ -2474,7 +2520,10 @@
    {
      // No need to call here updateTaskCompletionState - will be done
      // by the caller
      releaseIEContext();
      if (ieCtx != null)
      {
        releaseIEContext(ieCtx);
      }
      throw new DirectoryException(ResultCode.OTHER, errMsg);
    }
  }
@@ -2494,7 +2543,7 @@
   */
  public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs)
  {
    final ImportExportContext ieCtx = importExportContext.get();
    final ImportExportContext ieCtx = getImportExportContext();
    if (ieCtx == null || !ieCtx.importInProgress() || ieCtx.initReqMsgSent == null
        || !ieCtx.abandonIfStalled(stalledTimeoutMs))
    {
@@ -2511,6 +2560,23 @@
  }

  /**
   * Refuses a total update another server started into this replica: the exporter is told
   * so that it does not stream to a replica which will discard the entries, and this server
   * records why the total update it was the target of did not run - the exporter's task
   * reports the failure, and an administrator reading this server's log has to find it here.
   *
   * @param requesterServerId the server which asked for the total update
   * @param reason why it is refused
   */
  private void rejectInitializeTarget(int requesterServerId, LocalizableMessage reason)
  {
    logger.error(reason);
    // Silently not sent over a session which is already stopped: the replication server
    // then tells the exporter that this replica is not there to stream to.
    broker.publish(new ErrorMsg(requesterServerId, reason));
  }

  /**
   * Processes an InitializeTargetMsg received from a remote server
   * meaning processes an initialization from the entries expected to be
   * received now.
@@ -2532,9 +2598,14 @@
    InitializeTask initFromTask = null;
    final int source = initTargetMsgReceived.getSenderID();
    final ImportExportContext ieCtx;
    final Runnable hook = importClaimHook;
    if (hook != null)
    {
      hook.run();
    }
    if (initTargetMsgReceived.getInitiatorID() == getServerId())
    {
      ieCtx = importExportContext.get();
      ieCtx = getImportExportContext();
      if (ieCtx == null || !ieCtx.markInitStartReceived())
      {
        /*
@@ -2551,6 +2622,20 @@
        }
        return;
      }
      if (broker.shuttingDown())
      {
        /*
         * The same read as for a total update another server started (see below), with the
         * same window: the context is the one the request claimed, and no restart stops the
         * session under it - an import owns the session - but the domain going down or being
         * disabled does. The task which asked for the total update is failed with the reason;
         * the exporter learns of the stop the way it does of any other stop of this session.
         */
        ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
            ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId())));
        completeInitializeTask(ieCtx);
        return;
      }
    }
    else
    {
@@ -2564,11 +2649,32 @@
      }
      catch (DirectoryException e)
      {
        // A concurrent import/export owns the context: reject this
        // initialization without touching that operation's context, and let
        // the exporter know so that it does not export to a replica that
        // will discard the entries
        broker.publish(new ErrorMsg(requesterServerId, e.getMessageObject()));
        // A concurrent import/export owns the context, or the session is being stopped:
        // reject this initialization without touching that operation's context, and let
        // the exporter know so that it does not export to a replica that will discard the
        // entries
        rejectInitializeTarget(requesterServerId, e.getMessageObject());
        return;
      }
      if (broker.shuttingDown())
      {
        /*
         * The claim won against no restart, and the session is being stopped all the same:
         * the domain is going down or being disabled, or a restart found an export in the
         * context and stopped the session it streams over. The import would read that broker
         * as the end of its stream, and what runs before it publishes over the session: it
         * is refused here, before the backend is taken away.
         *
         * Read rather than claimed against: none of these roads claims anything - they stop
         * the session whatever owns it - so nothing orders this read against the stop. It
         * narrows the window, it does not close it: a stop which lands after it still has
         * the import run over a session which is going down, and end as a failed import over
         * the suffix it has replaced (issue #1039). Every one of these roads had that window
         * before this claim, and has it still.
         */
        releaseIEContext(ieCtx);
        rejectInitializeTarget(requesterServerId,
            ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN(), getServerId()));
        return;
      }
    }
@@ -2773,7 +2879,37 @@
   */
  public boolean ieRunning()
  {
    return importExportContext.get() != null;
    return getImportExportContext() != null;
  }

  /**
   * Sets what the listener thread runs between the {@code InitializeTargetMsg} it took off
   * the session and its claim of the import context - or, for a total update this replica
   * asked for, its read of the context the request claimed.
   * <p>
   * Only there for the tests which drive something else through that gap: it is a few
   * statements wide, and nothing else can hold the listener thread there.
   *
   * @param hook what to run there, or {@code null} to run nothing
   */
  @VisibleForTesting
  public void setImportClaimHook(Runnable hook)
  {
    importClaimHook = hook;
  }

  /**
   * Sets what {@link #disableService()} runs, under its locks, before it stops the broker.
   * <p>
   * Only there for the tests which drive something else through that gap: a total update
   * which is claimed after the decision to stop the service and before the stop.
   *
   * @param hook what to run there, or {@code null} to run nothing
   */
  @VisibleForTesting
  public void setServiceStopHook(Runnable hook)
  {
    serviceStopHook = hook;
  }

  /**
@@ -2790,7 +2926,7 @@
   */
  protected boolean importInProgress()
  {
    final ImportExportContext ieCtx = importExportContext.get();
    final ImportExportContext ieCtx = getImportExportContext();
    return ieCtx != null && ieCtx.importInProgress();
  }

@@ -3343,6 +3479,11 @@
    {
      synchronized (sessionLock)
      {
        final Runnable hook = serviceStopHook;
        if (hook != null)
        {
          hook.run();
        }
        /*
         * Stop the broker first in order to prevent the listener from reconnecting - see OPENDJ-457.
         */
@@ -3371,6 +3512,57 @@
  }

  /**
   * Stops the Replication Service the way {@link #disableService()} does, unless a total
   * update into this replica owns the session.
   * <p>
   * Whether one does is claimed rather than read (issue #1041): the listener thread claims
   * the import context for an {@code InitializeTargetMsg} under no lock, so a read of it
   * under {@link #serviceStateLock} orders nothing. The claim is
   * {@code SESSION_BEING_STOPPED}, held for the length of the stop and released once the
   * listener thread is gone - it is the one thread which claims a total update this replica
   * did not ask for, and {@link #disableService()} waits for it. An export in the context is
   * not an owner: the session is stopped from under it and the exporter reports the cut, as
   * it does for every other stop. A total update which lands between the end of that export
   * and the stop is refused by the listener when it reads the broker as stopping after its
   * claim; a stop which lands after that read still has the import run over a session which
   * is going down, and end as a failed import over the suffix it has replaced (issue
   * #1039): the read narrows that window, it does not close it.
   *
   * @return {@code true} when the service was stopped, {@code false} when a total update
   *         into this replica owns the session and it was left alone
   */
  protected final boolean disableServiceUnlessImportInProgress()
  {
    synchronized (serviceStateLock)
    {
      while (!importExportContext.compareAndSet(null, SESSION_BEING_STOPPED))
      {
        final ImportExportContext owner = importExportContext.get();
        if (owner == null)
        {
          // Released between the two reads: claim again.
          continue;
        }
        if (owner.importInProgress())
        {
          return false;
        }
        disableService();
        return true;
      }
      try
      {
        disableService();
      }
      finally
      {
        importExportContext.compareAndSet(SESSION_BEING_STOPPED, null);
      }
      return true;
    }
  }

  /**
   * Returns {@code true} if the listener thread is shutting down or has
   * shutdown.
   *
@@ -3863,7 +4055,8 @@
   */
  protected ImportExportContext getImportExportContext()
  {
    return importExportContext.get();
    final ImportExportContext ieCtx = importExportContext.get();
    return ieCtx != SESSION_BEING_STOPPED ? ieCtx : null;
  }

  /**

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

@@ -712,3 +712,6 @@
WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_328=Replication server RS(%d) not sending update \
 %s for domain "%s" to server %d at %s because the replication protocol version %d \
 negotiated with it has no encoding for this message
ERR_INIT_REJECTED_SESSION_STOPPING_330=The total update of domain "%s" was refused by directory \
 server %d: its session to the replication server is being stopped, and the entries would have \
 been streamed over that session. Ask for the total update again once the session is back

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

@@ -17,6 +17,7 @@

import static java.nio.charset.StandardCharsets.*;
import static org.assertj.core.api.Assertions.*;
import static org.opends.messages.CoreMessages.ERR_UNCAUGHT_THREAD_EXCEPTION;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.core.DirectoryServer.*;
@@ -26,7 +27,11 @@
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;

import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
@@ -40,6 +45,7 @@
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.EntryMsg;
import org.opends.server.replication.protocol.ErrorMsg;
import org.opends.server.replication.protocol.InitializeRequestMsg;
import org.opends.server.replication.protocol.InitializeTargetMsg;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
@@ -48,6 +54,7 @@
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
import org.testng.Assert;
@@ -75,6 +82,11 @@
 * change is replayed while the import is waiting for them - or, for the request, while the
 * exporter is holding the answer.
 * <p>
 * The claim of a total update this replica did not ask for is made by the listener thread
 * under no lock, so a restart of the session which reads no owner a moment before that claim
 * would stop the session the import is about to read (issue #1041): the listener is held
 * before its claim, and what stops the session is driven through the gap.
 * <p>
 * The {@code timeOut} each case declares is what it is expected to take at the most; it is
 * not what bounds it. {@code TestListener} sets the timeout of every test method from the
 * {@code org.opends.test.timeout} property, ten minutes under Maven and none outside it.
@@ -509,6 +521,421 @@
  }

  /**
   * A session restart decided after the {@code InitializeTargetMsg} was taken off the session
   * and before the import claimed its context must not have the import run over the session
   * it stops (issue #1041).
   * <p>
   * The owner read of the restart and the claim of the listener share no lock: the restart
   * reads no owner, stops the broker and waits for the listener thread to end - which is the
   * thread about to run the import. Run over that broker, the import ends on the nothing
   * which arrived - as a failed import since issue #1039, and as a finished one before it -
   * over a suffix which has been replaced by it all the same. Here the listener is held
   * before its claim, the restart is driven through the gap by a change whose attempts in
   * place are spent and held between its decision and the stop, and the listener is released
   * in between: the broker it finds is still up, so what refuses the import is the claim of
   * the restart, and the refusal reaches the exporter over the session which is about to be
   * stopped.
   */
  @Test(timeOut = 120_000)
  public void aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport() throws Exception
  {
    final Entry entry = TestCaseUtils.addEntry(
        "dn: cn=renamedSince," + EXAMPLE_DN,
        "objectClass: top",
        "objectClass: person",
        "cn: renamedSince",
        "sn: renamedSince");
    final String entryUUID = getEntryUUID(entry.getName());
    final int totalUpdatesStartedBefore =
        errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size();
    final int totalUpdatesEndedBefore =
        errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
    final int listenerDeathsBefore = listenerDeaths().size();
    final int refusalsBefore = errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size();

    // The listener thread has taken the InitializeTargetMsg off the session and is held
    // before it claims the import; the restart is held after its decision, before the stop.
    final CountDownLatch listenerHeld = new CountDownLatch(1);
    final CountDownLatch releaseListener = new CountDownLatch(1);
    final CountDownLatch stopHeld = new CountDownLatch(1);
    final CountDownLatch releaseStop = new CountDownLatch(1);
    domain.setImportClaimHook(() -> {
      listenerHeld.countDown();
      awaitUninterruptibly(releaseListener);
    });
    domain.setServiceStopHook(() -> {
      stopHeld.countDown();
      awaitUninterruptibly(releaseStop);
    });
    try
    {
      exporter.publish(new InitializeTargetMsg(
          baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW));
      assertTrue(listenerHeld.await(30, TimeUnit.SECONDS),
          "the listener thread did not reach the claim of the import");

      /*
       * A change whose entryUUID search never runs spends its attempts in place, finds no
       * owner and restarts the session. On a thread of its own: the restart is held before
       * the stop, and then waits for the listener thread.
       */
      final CSN csn = gen.newCSN();
      final AtomicReference<Throwable> replayFailure = new AtomicReference<>();
      final Thread replay = new Thread(() -> {
        try
        {
          replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
              generatemods("description", "replayed before the import was claimed"), entryUUID));
        }
        catch (Throwable t)
        {
          replayFailure.set(t);
        }
      }, "replay of " + csn);
      ShortCircuitPlugin.registerShortCircuit(
          OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
      try
      {
        replay.start();
        assertTrue(stopHeld.await(30, TimeUnit.SECONDS),
            "the failed replay did not decide to restart the session");
      }
      finally
      {
        ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
      }
      assertTrue(domain.isConnected(), "the session was stopped before the stop was held");
      assertFalse(domain.ieRunning(), "the claim of the stop is visible as a running import");
      /*
       * A total update asked for here is refused against the claim of the stop, and the
       * claim is left where it is: the road which fails to acquire a context of its own
       * releases nothing.
       */
      assertThatThrownBy(() -> domain.initializeFromRemote(EXPORTER_ID, null))
          .as("a total update asked for while the session is being stopped was not refused")
          .isInstanceOf(DirectoryException.class)
          .hasMessageContaining(ERR_INIT_REJECTED_SESSION_STOPPING.get(baseDN, DS_ID).toString());

      /*
       * The import is claimed against a restart which is decided and not yet made. Decided
       * either way before the stop is released: without the claim the import runs, and the
       * exporter is then waited for over a socket which nothing bounds.
       */
      releaseListener.countDown();
      waitUntil(() -> errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size() > refusalsBefore
          || errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size() > totalUpdatesStartedBefore,
          "the listener neither refused nor started the total update");
      assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()))
          .as("a total update claimed against a restart which was decided was started")
          .hasSize(totalUpdatesStartedBefore);
      final ErrorMsg refusal = waitForSpecificMsg(exporter, ErrorMsg.class);
      assertThat(refusal.getDetails().toString())
          .as("the exporter was not told why the total update was refused")
          .isEqualTo(ERR_INIT_REJECTED_SESSION_STOPPING.get(baseDN, DS_ID).toString());

      /*
       * An answer to a total update this replica asked for, which no context stands for - the
       * request was abandoned as stalled (issue #861) - finds only the claim of the stop, and
       * the claim is no context to import into: the answer is ignored. The total update
       * another server starts after it is what shows that the listener is past it: refused
       * here, against the same claim.
       */
      final int refusalsOfTheFirst =
          errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size();
      exporter.publish(new InitializeTargetMsg(
          baseDN, EXPORTER_ID, DS_ID, DS_ID, exportedEntries().length, INIT_WINDOW));
      exporter.publish(new InitializeTargetMsg(
          baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW));
      waitUntil(() -> errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size() > refusalsOfTheFirst
          || errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size() > totalUpdatesStartedBefore,
          "the listener neither refused nor started the total update after the stale answer");
      assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()))
          .as("an answer no context stands for was imported into the claim of the stop")
          .hasSize(totalUpdatesStartedBefore);

      releaseStop.countDown();
      replay.join(60_000);
      assertFalse(replay.isAlive(), "the restart did not end: the listener thread it waits for is still there");
      assertNull(replayFailure.get(), "the replay failed: " + replayFailure.get());
    }
    finally
    {
      releaseListener.countDown();
      releaseStop.countDown();
      domain.setImportClaimHook(null);
      domain.setServiceStopHook(null);
    }

    waitUntil(domain::isConnected, "the session was not started back after the restart");
    assertTrue(entryExists(entry.getName()), "the import ran over the session the restart"
        + " stopped: the suffix was replaced by the nothing which arrived");
    // A total update which got past the claim ran over the broker the restart then stopped
    // and ended on the nothing which arrived - as a failed import since issue #1039, and as
    // a finished one before it; neither is a total update which never ran.
    assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()))
        .as("a total update which was refused was run")
        .hasSize(totalUpdatesEndedBefore);
    assertThat(listenerDeaths())
        .as("the listener thread ended on an uncaught exception")
        .hasSize(listenerDeathsBefore);
    // Every record is written twice - the error log has two publishers in the tests.
    assertThat(errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()))
        .as("the refusal of the total update was not recorded on this server")
        .hasSizeGreaterThan(refusalsBefore);

    /*
     * The claim of the stop was released with the stop: the next total update into this
     * replica is claimed by the listener and runs to its end. Held, it would be invisible
     * to every reader of the context and refuse every total update for the life of the
     * domain.
     */
    startImportInto(exportedEntries().length);
    finishImport(exportedEntries());
    assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()))
        .as("the total update after the restart did not run to its end")
        .hasSize(totalUpdatesEndedBefore + 2);
  }

  /**
   * A domain disabled after the {@code InitializeTargetMsg} was taken off the session and
   * before the import claimed its context must refuse the import as well.
   * <p>
   * Nothing claims against the listener here - the domain disabling itself stops the session
   * whatever owns it - so what refuses the import is the listener reading, once its claim is
   * made, that the broker it would stream over is stopping. Without that read the claim wins,
   * and what runs next publishes the full update status over a session which is gone.
   */
  @Test(timeOut = 120_000)
  public void aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport() throws Exception
  {
    final Entry entry = TestCaseUtils.addEntry(
        "dn: cn=survivor," + EXAMPLE_DN,
        "objectClass: top",
        "objectClass: person",
        "cn: survivor",
        "sn: survivor");
    final int totalUpdatesStartedBefore =
        errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size();
    final int listenerDeathsBefore = listenerDeaths().size();
    final int refusalsBefore = errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()).size();

    final CountDownLatch listenerHeld = new CountDownLatch(1);
    final CountDownLatch releaseListener = new CountDownLatch(1);
    domain.setImportClaimHook(() -> {
      listenerHeld.countDown();
      awaitUninterruptibly(releaseListener);
    });
    try
    {
      exporter.publish(new InitializeTargetMsg(
          baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, exportedEntries().length, INIT_WINDOW));
      assertTrue(listenerHeld.await(30, TimeUnit.SECONDS),
          "the listener thread did not reach the claim of the import");

      // On a thread of its own: disabling the domain waits for the listener thread.
      final Thread disable = new Thread(domain::disable, "disable of " + EXAMPLE_DN);
      disable.start();
      waitUntil(() -> !domain.isConnected(), "disabling the domain did not stop the session");
      releaseListener.countDown();
      disable.join(60_000);
      assertFalse(disable.isAlive(), "disabling the domain did not end: the listener thread"
          + " it waits for is still there");
    }
    finally
    {
      releaseListener.countDown();
      domain.setImportClaimHook(null);
    }
    domain.enable();
    waitUntil(domain::isConnected, "the session was not started back by enable()");
    assertFalse(domain.ieRunning(), "the refused import left its context claimed");

    assertTrue(entryExists(entry.getName()), "the import ran over the session the disable"
        + " stopped: the suffix was replaced by the nothing which arrived");
    assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()))
        .as("a total update claimed against a session which is being stopped was started")
        .hasSize(totalUpdatesStartedBefore);
    assertThat(listenerDeaths())
        .as("the listener thread ended on an uncaught exception")
        .hasSize(listenerDeathsBefore);
    // Every record is written twice - the error log has two publishers in the tests.
    assertThat(errorLogRecordsOf(ERR_INIT_REJECTED_SESSION_STOPPING.ordinal()))
        .as("the refusal of the total update was not recorded on this server")
        .hasSizeGreaterThan(refusalsBefore);
  }

  /**
   * A domain disabled after the answer to a total update this replica asked for was taken off
   * the session, and before the import started, must refuse the import too.
   * <p>
   * The context is the one the request claimed, so there is nothing to claim against: what
   * refuses the import is the same read of the broker as for a total update another server
   * started. Without it the import runs over the session the disable stopped, and replaces the
   * suffix with the nothing which arrived.
   */
  @Test(timeOut = 120_000)
  public void aDomainDisabledBeforeTheImportItAskedForStartsRefusesTheImport() throws Exception
  {
    final Entry entry = TestCaseUtils.addEntry(
        "dn: cn=survivor," + EXAMPLE_DN,
        "objectClass: top",
        "objectClass: person",
        "cn: survivor",
        "sn: survivor");
    final int totalUpdatesStartedBefore =
        errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()).size();
    final int listenerDeathsBefore = listenerDeaths().size();

    domain.initializeFromRemote(EXPORTER_ID, null);
    assertNotNull(waitForSpecificMsg(exporter, InitializeRequestMsg.class));

    final CountDownLatch listenerHeld = new CountDownLatch(1);
    final CountDownLatch releaseListener = new CountDownLatch(1);
    domain.setImportClaimHook(() -> {
      listenerHeld.countDown();
      awaitUninterruptibly(releaseListener);
    });
    try
    {
      exporter.publish(new InitializeTargetMsg(
          baseDN, EXPORTER_ID, DS_ID, DS_ID, exportedEntries().length, INIT_WINDOW));
      assertTrue(listenerHeld.await(30, TimeUnit.SECONDS),
          "the listener thread did not reach the start of the import");

      // On a thread of its own: disabling the domain waits for the listener thread.
      final Thread disable = new Thread(domain::disable, "disable of " + EXAMPLE_DN);
      disable.start();
      waitUntil(() -> !domain.isConnected(), "disabling the domain did not stop the session");
      releaseListener.countDown();
      disable.join(60_000);
      assertFalse(disable.isAlive(), "disabling the domain did not end: the listener thread"
          + " it waits for is still there");
    }
    finally
    {
      releaseListener.countDown();
      domain.setImportClaimHook(null);
    }
    domain.enable();
    waitUntil(domain::isConnected, "the session was not started back by enable()");
    assertFalse(domain.ieRunning(), "the refused import left the context of its request claimed");

    assertTrue(entryExists(entry.getName()), "the import ran over the session the disable"
        + " stopped: the suffix was replaced by the nothing which arrived");
    assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START.ordinal()))
        .as("a total update answered over a session which is being stopped was started")
        .hasSize(totalUpdatesStartedBefore);
    assertThat(listenerDeaths())
        .as("the listener thread ended on an uncaught exception")
        .hasSize(listenerDeathsBefore);
  }

  /**
   * A session restart decided while a total update out of this replica is running stops the
   * session that export streams over (issue #1041).
   * <p>
   * What the restart must leave alone is a total update into this replica: the data it is
   * about to replace is read over the session, and the import is the thread the stop waits
   * for. An export is not that: it streams out of a backend nothing is taking away, on a
   * thread of its own, and a session stopped under it is the cut it reports to whoever asked
   * for the total update - the same cut every other stop of the session is. The claim the
   * restart makes for the import is not made here, and the session is stopped as it was
   * before the claim.
   * <p>
   * The export holds the context by standing where it waits for its target to report the
   * start of the total update: the target is a broker of this test, and reports nothing.
   */
  @Test(timeOut = 120_000)
  public void aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver() throws Exception
  {
    final Entry entry = TestCaseUtils.addEntry(
        "dn: cn=renamedSince," + EXAMPLE_DN,
        "objectClass: top",
        "objectClass: person",
        "cn: renamedSince",
        "sn: renamedSince");
    final String entryUUID = getEntryUUID(entry.getName());
    waitUntil(() -> domain.getReplicaInfos().containsKey(EXPORTER_ID),
        "the exporter is not in the replicas of the domain: nothing to export into");

    final AtomicReference<Throwable> exportFailure = new AtomicReference<>();
    final Thread export = new Thread(() -> {
      try
      {
        domain.initializeRemote(EXPORTER_ID, null);
      }
      catch (Throwable t)
      {
        exportFailure.set(t);
      }
    }, "export of " + EXAMPLE_DN);

    // The restart is held after its decision, before the stop: what the case reads is the
    // decision the export was found by, not the session which is down a moment later.
    final CountDownLatch stopHeld = new CountDownLatch(1);
    final CountDownLatch releaseStop = new CountDownLatch(1);
    domain.setServiceStopHook(() -> {
      stopHeld.countDown();
      awaitUninterruptibly(releaseStop);
    });
    final CSN csn = gen.newCSN();
    final AtomicReference<Throwable> replayFailure = new AtomicReference<>();
    final Thread replay = new Thread(() -> {
      try
      {
        replayMsg(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN),
            generatemods("description", "replayed while the export was running"), entryUUID));
      }
      catch (Throwable t)
      {
        replayFailure.set(t);
      }
    }, "replay of " + csn);
    try
    {
      export.start();
      waitUntil(() -> domain.ieRunning() || exportFailure.get() != null,
          "the export did not claim the import context");
      assertNull(exportFailure.get(),
          "the export failed before it claimed the context: " + exportFailure.get());

      // A change whose entryUUID search never runs spends its attempts in place and asks
      // for the session to be restarted, the way it does in the case above.
      ShortCircuitPlugin.registerShortCircuit(
          OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue());
      try
      {
        replay.start();
        assertTrue(stopHeld.await(30, TimeUnit.SECONDS),
            "the restart left the session to the export: an export is not the owner a total"
                + " update into this replica is");
      }
      finally
      {
        ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
      }
      assertTrue(domain.ieRunning(), "the export ended before the restart was decided");
    }
    finally
    {
      releaseStop.countDown();
      domain.setServiceStopHook(null);
    }

    replay.join(60_000);
    assertFalse(replay.isAlive(), "the restart did not end");
    assertNull(replayFailure.get(), "the replay failed: " + replayFailure.get());
    export.join(60_000);
    assertFalse(export.isAlive(), "the export did not end once the session it streams over"
        + " was stopped");
    assertThat(exportFailure.get())
        .as("the export was not told that the session it streams over was cut")
        .isInstanceOf(DirectoryException.class);
    waitUntil(domain::isConnected, "the session was not started back after the restart");
    assertFalse(domain.ieRunning(), "the export which was cut left its context claimed");
  }

  /**
   * Has the exporter start a total update into this replica, and returns once the backend
   * of the domain is deregistered for it: from then on the import is reading the session,
   * and a change replayed here is replayed into no backend.
@@ -589,9 +1016,9 @@
  private static List<String> errorLogRecordsOf(int msgId, CSN csn)
  {
    final List<String> records = new ArrayList<>();
    for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
    for (String record : errorLogRecordsOf(msgId))
    {
      if (record.contains("msgID=" + msgId) && record.contains(csn.toString()))
      if (record.contains(csn.toString()))
      {
        records.add(record);
      }
@@ -599,6 +1026,65 @@
    return records;
  }

  /** The records of the error log which carry the provided message id. */
  private static List<String> errorLogRecordsOf(int msgId)
  {
    final List<String> records = new ArrayList<>();
    for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
    {
      if (record.contains("msgID=" + msgId))
      {
        records.add(record);
      }
    }
    return records;
  }

  /** The records of the error log which report the listener thread of the domain ending abnormally. */
  private static List<String> listenerDeaths()
  {
    final List<String> records = new ArrayList<>();
    for (String record : errorLogRecordsOf(ERR_UNCAUGHT_THREAD_EXCEPTION.ordinal()))
    {
      if (record.contains("listener for domain \"" + EXAMPLE_DN + "\""))
      {
        records.add(record);
      }
    }
    return records;
  }

  private static void waitUntil(BooleanSupplier condition, String failure) throws InterruptedException
  {
    final long deadline = System.currentTimeMillis() + 30_000;
    while (!condition.getAsBoolean())
    {
      assertTrue(System.currentTimeMillis() < deadline, failure);
      Thread.sleep(20);
    }
  }

  private static void awaitUninterruptibly(CountDownLatch latch)
  {
    boolean interrupted = false;
    while (true)
    {
      try
      {
        latch.await();
        break;
      }
      catch (InterruptedException e)
      {
        interrupted = true;
      }
    }
    if (interrupted)
    {
      Thread.currentThread().interrupt();
    }
  }

  private void replayMsg(UpdateMsg updateMsg) throws InterruptedException
  {
    domain.processUpdate(updateMsg);