[#900] Time the ReplicaOfflineMsg grace period per replica, and spend it wh...

Valery Kharseko

12 hours ago 995d54a8963d87335196b86d8d4fc1b252a38ee4

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

@@ -1975,8 +1975,8 @@
  @Override
  public void publishReplicaOfflineMsg()
  {
    pendingChanges.putReplicaOfflineMsg();
    dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN());
    final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
    dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN);
  }

  /**

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

@@ -13,6 +13,7 @@
 *
 * Copyright 2009 Sun Microsystems, Inc.
 * Portions Copyright 2011-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

@@ -123,8 +124,10 @@

  /**
   * Add a replica offline message to the pending list.
   *
   * @return the CSN of the message which was added
   */
  public synchronized void putReplicaOfflineMsg()
  public synchronized CSN putReplicaOfflineMsg()
  {
    final CSN offlineCSN = csnGenerator.newCSN();
    final PendingChange pendingChange =
@@ -133,6 +136,7 @@

    pendingChanges.put(offlineCSN, pendingChange);
    pushCommittedChanges();
    return offlineCSN;
  }

  /**

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

@@ -178,6 +178,11 @@

  /**
   * Creates a new Replication server using the provided configuration entry.
   * <p>
   * The synchronization object this creates is its own, so the resulting server does not
   * synchronize its shutdown with a collocated directory server. A server which has to must be
   * built with {@link #ReplicationServer(ReplicationServerCfg, DSRSShutdownSync)}, passing the
   * instance the directory server side records its ReplicaOfflineMsgs on.
   *
   * @param cfg The configuration of this replication server.
   * @throws ConfigException When Configuration is invalid.
@@ -802,6 +807,9 @@
    // listen port which cannot be bound, would otherwise leave them behind. Shut them down
    // before the changelog they write to, and one unchecked exception at a time: the changelog
    // this one is built on is known to be broken, and what follows still has to run.
    // Nothing in an instance which never finished coming up can forward a pending
    // ReplicaOfflineMsg, so this path does not wait for one: it would only delay the failure
    // which is being reported by a grace period which cannot pay off.
    for (ReplicationServerDomain domain : getReplicationServerDomains())
    {
      try
@@ -1181,7 +1189,19 @@
      listenThread.interrupt();
    }

    // shutdown all the replication domains
    /*
     * Let the ReplicaOfflineMsgs a collocated DS sent be forwarded while every handler is still
     * up, and only then stop the domains: shutting a domain down deactivates the consumer of its
     * handlers, clears their message queue and closes their session - see OPENDJ-1453. All the
     * domains wait together and share one deadline, so the shutdown is bounded by one grace
     * period and the wait of one domain does not spend the grace period of the next.
     * <p>
     * This also runs before the assured timer of any domain is cancelled, so an assured update
     * still waiting for acks keeps timing out during the wait instead of holding its sender
     * until the sessions are closed.
     */
    awaitReplicaOfflineMsgsForwarded();

    for (ReplicationServerDomain domain : getReplicationServerDomains())
    {
      domain.shutdown();
@@ -1203,6 +1223,28 @@
  }

  /**
   * Waits for the ReplicaOfflineMsg of every domain which has a replication server to forward it
   * to. With no such server connected there is nobody to forward the message to, and waiting
   * would only delay the shutdown by the whole grace period.
   */
  private void awaitReplicaOfflineMsgsForwarded()
  {
    final List<DN> domainsToWaitFor = new ArrayList<>();
    for (ReplicationServerDomain domain : getReplicationServerDomains())
    {
      if (!domain.getConnectedRSs().isEmpty())
      {
        domainsToWaitFor.add(domain.getBaseDN());
      }
    }
    if (!domainsToWaitFor.isEmpty())
    {
      dsrsShutdownSync.awaitReplicaOfflineMsgsForwarded(
          domainsToWaitFor, dsrsShutdownSync.newShutdownDeadline());
    }
  }

  /**
   * Retrieves the time after which changes must be deleted from the
   * persistent storage (in milliseconds).
   *

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

@@ -1661,7 +1661,15 @@
    return monitorMsg;
  }

  /** Shutdown this ReplicationServerDomain. */
  /**
   * Shutdown this ReplicationServerDomain.
   * <p>
   * A ReplicaOfflineMsg which a collocated DS sent and which is still to be forwarded must be
   * waited for before this runs: stopping the server handlers deactivates their consumer, clears
   * their message queue and closes their session, after which the message can no longer be sent
   * - see OPENDJ-1453. ReplicationServer.shutdown() waits for the messages of all of its domains
   * before it stops any of them.
   */
  public void shutdown()
  {
    DirectoryServer.deregisterMonitorProvider(this);

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

@@ -13,6 +13,7 @@
 *
 * Copyright 2006-2009 Sun Microsystems, Inc.
 * Portions Copyright 2011-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

@@ -87,9 +88,14 @@
    LocalizableMessage errMessage = null;
    try
    {
      boolean shutdown = false;
      while (!shutdown
          || !dsrsShutdownSync.canShutdown(replicationServerDomain.getBaseDN()))
      /*
       * Looping here to wait for a pending ReplicaOfflineMsg would achieve nothing: this writer
       * only stops once its handler has been shut down, which deactivates the consumer, clears
       * the message queue and closes the session. The shutdown of the domain waits for the
       * message to be forwarded before it stops the handlers - see
       * ReplicationServerDomain.shutdown() and OPENDJ-1453.
       */
      while (true)
      {
        final UpdateMsg updateMsg = this.handler.take();
        if (updateMsg == null)
@@ -97,15 +103,25 @@
          // this connection is closing
          errMessage = LocalizableMessage.raw(
           "Connection closure: null update returned by domain.");
          shutdown = true;
          break;
        }
        else if (!isUpdateMsgFiltered(updateMsg))
        if (!isUpdateMsgFiltered(updateMsg))
        {
          // Publish the update to the remote server using a protocol version it supports
          session.publish(updateMsg);
          if (updateMsg instanceof ReplicaOfflineMsg)
          /*
           * Only the forward to a peer RS ends the wait of the shutdown: what the grace period
           * buys is the rest of the topology learning that the replica went offline.
           * ReplicationServerDomain.put() never queues this message for a directory server - its
           * isUpdateMsgFiltered() drops it there - but a directory server which is catching up
           * reads its updates from the changelog, where ReplicaCursor synthesizes a
           * ReplicaOfflineMsg from the offline CSN of the replica. Publishing that one says
           * nothing about the peer RSs the shutdown is waiting for.
           */
          if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
          {
            dsrsShutdownSync.replicaOfflineMsgForwarded(replicationServerDomain.getBaseDN());
            dsrsShutdownSync.replicaOfflineMsgForwarded(
                replicationServerDomain.getBaseDN(), updateMsg.getCSN());
          }
        }
      }

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

@@ -12,13 +12,19 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.service;

import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.AtomicLong;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.replication.common.CSN;

/**
 * Class useful for the case where DS/RS instances are collocated inside the
@@ -27,24 +33,73 @@
 * More specifically, it ensures a ReplicaOfflineMsg sent by the DS is
 * relayed/forwarded by the collocated RS to the other RSs in the topology
 * before the whole process shuts down.
 * <p>
 * The state is kept per domain and per instance: the collocated DS and RS
 * sides coordinate through the single instance MultimasterReplication hands
 * to both of them.
 *
 * @since OPENDJ-1453
 */
public class DSRSShutdownSync
{
  private static final ConcurrentSkipListSet<DN> replicaOfflineMsgs = new ConcurrentSkipListSet<>();
  private static AtomicLong stopInstanceTimestamp = new AtomicLong();
  /**
   * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated
   * RS, in milliseconds, counted from the moment the message was sent.
   */
  public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000;

  private final long gracePeriod;

  /**
   * The ReplicaOfflineMsg which has not been forwarded yet, per domain and per
   * replica of that domain.
   * <p>
   * It is kept per domain because a domain sends this message whenever its
   * replication service is disabled - an online import, a restore, a
   * configuration change - and not only when the process shuts down. A single
   * entry for the whole process would be the one of the first such message and
   * would leave no grace period at all to the shutdown this class exists for.
   * <p>
   * It is kept per replica because the collocated RS relays the message of
   * every replica connected to it, and the forward of another replica's
   * message says nothing about this one.
   */
  private final ConcurrentMap<DN, ConcurrentMap<Integer, PendingOfflineMsg>> replicaOfflineMsgs =
      new ConcurrentHashMap<>();
  /** Monitor notified whenever a ReplicaOfflineMsg has been forwarded. */
  private final Object forwardedMonitor = new Object();

  /** Creates a synchronization object using the default grace period. */
  public DSRSShutdownSync()
  {
    this(REPLICA_OFFLINE_GRACE_PERIOD);
  }

  /**
   * Creates a synchronization object using the provided grace period.
   *
   * @param gracePeriod
   *          how long a ReplicaOfflineMsg may hold back the shutdown, in milliseconds
   */
  DSRSShutdownSync(long gracePeriod)
  {
    this.gracePeriod = gracePeriod;
  }

  /**
   * Message has been sent.
   *
   * @param baseDN
   *          the domain for which the message has been sent
   * @param offlineCSN
   *          the CSN of the message, which identifies both the replica which announced itself
   *          offline and the announcement being waited for
   */
  public void replicaOfflineMsgSent(DN baseDN)
  public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN)
  {
    stopInstanceTimestamp.compareAndSet(0, System.currentTimeMillis());
    replicaOfflineMsgs.add(baseDN);
    replicaOfflineMsgs
        .computeIfAbsent(baseDN, dn -> new ConcurrentHashMap<Integer, PendingOfflineMsg>())
        .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime()));
  }

  /**
@@ -52,23 +107,179 @@
   *
   * @param baseDN
   *          the domain for which the message has been sent
   * @param forwardedCSN
   *          the CSN of the forwarded message
   */
  public void replicaOfflineMsgForwarded(DN baseDN)
  public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN)
  {
    replicaOfflineMsgs.remove(baseDN);
    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
    if (msgs != null)
    {
      final int serverId = forwardedCSN.getServerId();
      final PendingOfflineMsg pending = msgs.get(serverId);
      /*
       * A replica announces itself offline on every disableService(), so the message which is
       * forwarded now may be an older one - queued behind a backlog since an earlier import, or
       * synthesized from the offline CSN of the changelog for a server which is catching up.
       * Such a forward says nothing about the announcement the shutdown is waiting for, and must
       * not consume its grace period.
       */
      if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN))
      {
        msgs.remove(serverId, pending);
      }
    }
    synchronized (forwardedMonitor)
    {
      forwardedMonitor.notifyAll();
    }
  }

  /**
   * Whether a ReplicationServer ServerReader or ServerWriter can proceed with
   * shutdown.
   * Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg
   * has been forwarded or its grace period has expired.
   * <p>
   * The shutdown itself blocks on {@link #awaitReplicaOfflineMsgsForwarded(Collection, long)}
   * rather than polling this; it is the same state, observable without waiting for it.
   *
   * @param baseDN
   *          the baseDN of the ServerReader or ServerWriter .
   * @return true if the caller can shutdown, false otherwise
   *          the baseDN of the domain being shut down
   * @return true if the shutdown of this domain need not wait any longer, i.e. its message was
   *         forwarded or its grace period has expired, false otherwise
   */
  public boolean canShutdown(DN baseDN)
  {
    return !replicaOfflineMsgs.contains(baseDN)
        || System.currentTimeMillis() - stopInstanceTimestamp.get() > 5000;
    return remainingGracePeriod(baseDN) <= 0;
  }

  /**
   * Returns the time by which every wait of one shutdown must be over.
   * <p>
   * A process shuts its domains down one after the other and each of them may have a message
   * pending, so a deadline computed once and shared by all of them keeps the whole shutdown
   * bounded by one grace period instead of one per domain.
   *
   * @return the point in time, on the {@link System#nanoTime()} clock, by which the waits must
   *         be over
   */
  public long newShutdownDeadline()
  {
    return System.nanoTime() + MILLISECONDS.toNanos(gracePeriod);
  }

  /**
   * Waits for the ReplicaOfflineMsg of every provided domain to be forwarded, or for their grace
   * periods or the provided deadline to expire.
   * <p>
   * This must be called before the server handlers of those domains are stopped: stopping them
   * deactivates their consumer, clears their message queue and closes their session, after which
   * the message can no longer be forwarded.
   * <p>
   * All the domains of one shutdown wait together rather than one after the other, so that the
   * shutdown is bounded by one grace period without the wait of one domain spending the grace
   * period of the next.
   *
   * @param baseDNs
   *          the baseDNs of the domains whose messages must be forwarded
   * @param deadline
   *          the point in time, on the {@link System#nanoTime()} clock, by which this wait must
   *          be over whatever the domains announce in the meantime - see
   *          {@link #newShutdownDeadline()}. A deadline which is not in the future returns
   *          without waiting at all, for a caller which has nothing to wait for.
   */
  public void awaitReplicaOfflineMsgsForwarded(Collection<DN> baseDNs, long deadline)
  {
    if (deadline - System.nanoTime() <= 0)
    {
      return;
    }
    synchronized (forwardedMonitor)
    {
      while (true)
      {
        final long timeout = Math.min(remainingGracePeriod(baseDNs),
            NANOSECONDS.toMillis(deadline - System.nanoTime()));
        if (timeout <= 0)
        {
          return;
        }
        try
        {
          forwardedMonitor.wait(timeout);
        }
        catch (InterruptedException e)
        {
          /*
           * Give up waiting. The interrupt is deliberately not restored: what follows this call is
           * the rest of the shutdown - joining the reader and writer thread of every handler, then
           * closing the changelog DB - and an interrupt flag would make all of it give up too.
           */
          return;
        }
      }
    }
  }

  /**
   * Returns the time left, in milliseconds, to forward the ReplicaOfflineMsg of the replica of
   * the provided domains which has the longest to wait, zero or less if none of them has a
   * message pending.
   */
  private long remainingGracePeriod(Collection<DN> baseDNs)
  {
    long remaining = 0;
    for (DN baseDN : baseDNs)
    {
      remaining = Math.max(remaining, remainingGracePeriod(baseDN));
    }
    return remaining;
  }

  /**
   * Returns the time left, in milliseconds, to forward the ReplicaOfflineMsg of the replica of
   * this domain which has the longest to wait, zero or less if no message of this domain is
   * pending.
   */
  private long remainingGracePeriod(DN baseDN)
  {
    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
    if (msgs == null)
    {
      return 0;
    }
    final long now = System.nanoTime();
    long remaining = 0;
    for (PendingOfflineMsg pending : msgs.values())
    {
      remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime));
    }
    return remaining;
  }

  /**
   * A ReplicaOfflineMsg a replica announced and which has not been forwarded yet.
   * <p>
   * This deliberately does not override {@code equals}: the two-argument
   * {@link ConcurrentMap#remove(Object, Object)} of the forward guard must match the very
   * announcement it read, not another one which happens to carry the same values.
   */
  private static final class PendingOfflineMsg
  {
    /** The CSN of the message, so that the forward of an older one is not taken for this one. */
    private final CSN csn;
    /** When the message was announced, on the {@link System#nanoTime()} clock. */
    private final long sentTime;

    private PendingOfflineMsg(CSN csn, long sentTime)
    {
      this.csn = csn;
      this.sentTime = sentTime;
    }

    @Override
    public String toString()
    {
      return "PendingOfflineMsg(" + csn + ")";
    }
  }
}

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

New file
@@ -0,0 +1,793 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

import static org.assertj.core.api.Assertions.assertThat;
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;

import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.util.StaticUtils;
import org.opends.server.util.TestTimer;
import org.testng.annotations.Test;

/**
 * The shutdown of a replication server must let a ReplicaOfflineMsg sent by a collocated
 * directory server be forwarded to the other replication servers of the topology before the
 * server handlers are stopped - stopping them deactivates their consumer, clears their message
 * queue and closes their session, after which the message can no longer be sent.
 * <p>
 * Most tests drive {@link DSRSShutdownSync} directly rather than through a collocated directory
 * server: the contract they pin is when the shutdown of the replication server waits, and how
 * long. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} pins the outcome
 * those waits exist for, on a peer connected through the real handshake.
 */
@SuppressWarnings("javadoc")
public class ReplicationServerShutdownSyncTest extends ReplicationTestCase
{
  private static final int SOCKET_TIMEOUT_MS = 30000;
  /** A session end nobody owns is discarded, so its cleanup waits far less than a live one. */
  private static final int DISCARDED_SESSION_TIMEOUT_MS = 2000;
  private static final int REMOTE_RS_ID = 92;
  private static final int REMOTE_DS_ID = 93;
  /** The collocated replica whose ReplicaOfflineMsg the shutdown waits for. */
  private static final int LOCAL_DS_ID = 94;
  /** Time given to the forwarding thread before it releases the shutdown. */
  private static final long FORWARD_DELAY = 500;
  /** How often the domains of {@link #theGracePeriodIsSharedByAllTheDomainsOfOneShutdown()}
   * announce themselves offline again while the shutdown is waiting for them. */
  private static final long REANNOUNCE_INTERVAL = 200;

  @Test
  public void shutdownWaitsForTheReplicaOfflineMsgToBeForwarded() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      replicationServer = newReplicationServer(shutdownSync, "shutdownSyncWaitDb", 8221);
      final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        registerConnectedReplicationServer(replicationServer, baseDN, session);

        final long startTime = System.nanoTime();
        shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN());
        replicationServer.shutdown();
        final long elapsed = elapsedMillis(startTime);

        assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      }
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  @Test
  public void shutdownResumesAsSoonAsTheReplicaOfflineMsgIsForwarded() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    Thread forwarder = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      replicationServer = newReplicationServer(shutdownSync, "shutdownSyncForwardDb", 8222);
      final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        registerConnectedReplicationServer(replicationServer, baseDN, session);
        final CSN offlineCSN = newOfflineCSN();
        forwarder = newForwarderThread(shutdownSync, baseDN, offlineCSN);
        shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);

        final long startTime = System.nanoTime();
        forwarder.start();
        replicationServer.shutdown();
        final long elapsed = elapsedMillis(startTime);

        assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY)
            .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      }
    }
    finally
    {
      joinQuietly(forwarder);
      removeQuietly(replicationServer);
    }
  }

  /**
   * The outcome the grace period exists for, end to end: a peer replication server connected
   * through the real handshake has received the ReplicaOfflineMsg of the collocated replica by
   * the time the shutdown returns.
   * <p>
   * The waiting tests above measure durations only, so they stay green if the wait is moved
   * after the handlers are stopped - which reintroduces OPENDJ-1453 and loses the message. This
   * one fails in that case.
   */
  @Test
  public void thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    FakePeerReplicationServer peer = null;
    Thread publisher = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer =
          newReplicationServer(shutdownSync, "shutdownSyncDeliveryDb", 8226, replicationPort);
      broker = openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      peer = new FakePeerReplicationServer(replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID);

      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      waitForConnectedReplicationServer(domain);
      final Future<ReplicaOfflineMsg> received = peer.receiveReplicaOfflineMsg();

      /*
       * The replica announces itself offline once the shutdown of the replication server is
       * already waiting for the message, which is the ordering the grace period exists for.
       */
      final CSN offlineCSN = newOfflineCSN();
      shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
      publisher = newPublisherThread(broker, offlineCSN);

      final long startTime = System.nanoTime();
      publisher.start();
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      final ReplicaOfflineMsg forwarded = received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      assertThat(forwarded)
          .as("the peer replication server was never told that the replica went offline, its "
              + "read ended with: %s", peer.readerFailure())
          .isNotNull();
      assertThat(forwarded.getCSN().getServerId()).isEqualTo(LOCAL_DS_ID);
      assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY)
          .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
    }
    finally
    {
      joinQuietly(publisher);
      closeQuietly(peer);
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * Only a peer replication server learning about the offline replica ends the wait.
   * ReplicationServerDomain.put() never queues a ReplicaOfflineMsg for a directory server, but
   * the changelog cursor of a directory server which is catching up synthesizes one from the
   * offline CSN of the replica, so the writer serving a directory server can publish it - and
   * the peer replication servers would still know nothing.
   */
  @Test
  public void theForwardToADirectoryServerDoesNotEndTheWait() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer =
          newReplicationServer(shutdownSync, "shutdownSyncDataServerForwardDb", 8227, replicationPort);
      broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
      final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        // a peer replication server, so that the shutdown does wait for the message: what this
        // test pins is that the directory server receiving it is not what ends that wait
        registerConnectedReplicationServer(replicationServer, baseDN, session);
        final ReplicationServerDomain domain =
            replicationServer.getReplicationServerDomain(baseDN, true);
        final DataServerHandler dsHandler = waitForConnectedDirectoryServer(domain);

        final CSN offlineCSN = newOfflineCSN();
        final long startTime = System.nanoTime();
        shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
        // the very message the shutdown waits for, so only the guard of the writer can save it
        dsHandler.add(new ReplicaOfflineMsg(offlineCSN));

        // the directory server did receive it, so its writer went through the forwarding code
        assertThat(waitForSpecificMsg(broker, ReplicaOfflineMsg.class).getCSN().getServerId())
            .isEqualTo(LOCAL_DS_ID);
        assertThat(elapsedMillis(startTime))
            .as("the fixture must deliver the message well inside the grace period, otherwise "
                + "the wait asserted below cannot be told apart from a slow delivery")
            .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD / 2);

        replicationServer.shutdown();
        final long elapsed = elapsedMillis(startTime);

        assertThat(elapsed)
            .as("the message published to a directory server ended the wait of the shutdown")
            .isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      }
    }
    finally
    {
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * With no other replication server connected there is nobody to forward the message to, so
   * waiting would only delay the shutdown of a standalone server by the whole grace period.
   */
  @Test
  public void shutdownIsNotDelayedWhenNoOtherReplicationServerCanForwardTheMessage() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer(shutdownSync, "shutdownSyncAloneDb", 8223);
      replicationServer.getReplicationServerDomain(baseDN, true);

      final long startTime = System.nanoTime();
      shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN());
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The writer serving a directory server must not hold back the shutdown either: it used to
   * loop on the pending message until the grace period expired, although its handler had already
   * been shut down - which deactivates its consumer and leaves the loop nothing to take.
   */
  @Test
  public void shutdownIsNotDelayedByTheWriterServingADirectoryServer() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    ReplicationBroker broker = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer =
          newReplicationServer(shutdownSync, "shutdownSyncDataServerDb", 8225, replicationPort);
      broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);

      final long startTime = System.nanoTime();
      shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN());
      replicationServer.shutdown();
      final long elapsed = elapsedMillis(startTime);

      assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
    }
    finally
    {
      stop(broker);
      removeQuietly(replicationServer);
    }
  }

  @Test
  public void shutdownIsNotDelayedWhenNoReplicaOfflineMsgIsPending() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    ReplicationServer replicationServer = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      replicationServer = newReplicationServer(shutdownSync, "shutdownSyncNoMsgDb", 8224);
      final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        registerConnectedReplicationServer(replicationServer, baseDN, session);

        final long startTime = System.nanoTime();
        replicationServer.shutdown();
        final long elapsed = elapsedMillis(startTime);

        assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      }
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The domains of a replication server are shut down one after the other, so the grace period
   * must bound the whole shutdown and not each of its domains: a process with several base DNs
   * would otherwise pay it once per domain.
   * <p>
   * Both domains keep announcing themselves offline while the shutdown is running, so neither
   * wait can be ended by a forward and each of them runs to its bound - one grace period in
   * total if it is shared, one per domain otherwise.
   */
  @Test
  public void theGracePeriodIsSharedByAllTheDomainsOfOneShutdown() throws Exception
  {
    final DN baseDN1 = DN.valueOf(TEST_ROOT_DN_STRING);
    final DN baseDN2 = DN.valueOf("dc=world,dc=company");
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    final AtomicBoolean stopped = new AtomicBoolean();
    ReplicationServer replicationServer = null;
    Thread reAnnouncer = null;
    try (ServerSocket listen1 = TestCaseUtils.bindFreePort();
        ServerSocket listen2 = TestCaseUtils.bindFreePort())
    {
      listen1.setSoTimeout(SOCKET_TIMEOUT_MS);
      listen2.setSoTimeout(SOCKET_TIMEOUT_MS);
      replicationServer = newReplicationServer(shutdownSync, "shutdownSyncSharedDeadlineDb", 8228);
      final Session[] sessionPair1 = connectSessionPair(listen1, getReplSessionSecurity());
      final Session[] sessionPair2 = connectSessionPair(listen2, getReplSessionSecurity());
      try (Session remoteEnd1 = sessionPair1[0];
          Session session1 = sessionPair1[1];
          Session remoteEnd2 = sessionPair2[0];
          Session session2 = sessionPair2[1])
      {
        registerConnectedReplicationServer(replicationServer, baseDN1, session1);
        registerConnectedReplicationServer(replicationServer, baseDN2, session2);
        /*
         * Announce both domains offline here rather than leaving it to the thread below: the
         * wait of the shutdown must be armed whatever that thread has had time to run.
         */
        final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
        shutdownSync.replicaOfflineMsgSent(baseDN1, csns.newCSN());
        shutdownSync.replicaOfflineMsgSent(baseDN2, csns.newCSN());
        reAnnouncer = newReAnnouncerThread(shutdownSync, baseDN1, baseDN2, stopped);
        reAnnouncer.start();

        final long startTime = System.nanoTime();
        replicationServer.shutdown();
        final long elapsed = elapsedMillis(startTime);

        assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
        assertThat(elapsed)
            .as("each domain waited its own grace period instead of sharing one deadline")
            .isLessThan(2 * DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      }
    }
    finally
    {
      stopped.set(true);
      joinQuietly(reAnnouncer);
      removeQuietly(replicationServer);
    }
  }

  private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName,
      int serverId) throws Exception
  {
    return newReplicationServer(shutdownSync, dbDirName, serverId, TestCaseUtils.findFreePort());
  }

  private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName,
      int serverId, int replicationPort) throws Exception
  {
    return new ReplicationServer(new ReplServerFakeConfiguration(
        replicationPort, dbDirName, 0, serverId, 0, 100, new TreeSet<String>()), shutdownSync);
  }

  /**
   * Registers a peer replication server on the domain as the handshake does, but without the
   * protocol exchange: the handler this leaves behind has no writer, which is enough for the
   * tests which only need a domain with a connected peer.
   */
  private void registerConnectedReplicationServer(
      ReplicationServer replicationServer, DN baseDN, Session session) throws Exception
  {
    final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true);
    final ReplicationServerHandler rsHandler =
        new ReplicationServerHandler(session, 100, replicationServer, 100);
    rsHandler.serverId = REMOTE_RS_ID;
    rsHandler.serverURL = "127.0.0.1:1636";
    rsHandler.setBaseDNAndDomain(baseDN, false);
    domain.lock();
    try
    {
      domain.register(rsHandler);
    }
    finally
    {
      domain.release();
    }
  }

  private void waitForConnectedReplicationServer(final ReplicationServerDomain domain)
      throws Exception
  {
    newConnectionTimer().repeatUntilSuccess(new TestTimer.CallableVoid()
    {
      @Override
      public void call() throws Exception
      {
        assertThat(domain.getConnectedRSs())
            .as("the peer replication server never connected").containsKey(REMOTE_RS_ID);
      }
    });
  }

  private DataServerHandler waitForConnectedDirectoryServer(final ReplicationServerDomain domain)
      throws Exception
  {
    return newConnectionTimer().repeatUntilSuccess(new Callable<DataServerHandler>()
    {
      @Override
      public DataServerHandler call() throws Exception
      {
        final DataServerHandler dsHandler = domain.getConnectedDSs().get(REMOTE_DS_ID);
        assertThat(dsHandler).as("the directory server never connected").isNotNull();
        return dsHandler;
      }
    });
  }

  private static TestTimer newConnectionTimer()
  {
    return new TestTimer.Builder()
        .maxSleep(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS)
        .sleepTimes(10, TimeUnit.MILLISECONDS)
        .toTimer();
  }

  private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, final DN baseDN,
      final CSN offlineCSN)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        if (!sleepQuietly(FORWARD_DELAY))
        {
          return;
        }
        shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN);
      }
    });
  }

  private Thread newPublisherThread(final ReplicationBroker broker, final CSN offlineCSN)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        if (!sleepQuietly(FORWARD_DELAY))
        {
          return;
        }
        broker.publish(new ReplicaOfflineMsg(offlineCSN));
      }
    });
  }

  private Thread newReAnnouncerThread(final DSRSShutdownSync shutdownSync, final DN baseDN1,
      final DN baseDN2, final AtomicBoolean stopped)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
        while (!stopped.get())
        {
          shutdownSync.replicaOfflineMsgSent(baseDN1, csns.newCSN());
          shutdownSync.replicaOfflineMsgSent(baseDN2, csns.newCSN());
          if (!sleepQuietly(REANNOUNCE_INTERVAL))
          {
            return;
          }
        }
      }
    });
  }

  /** Milliseconds elapsed since a {@link System#nanoTime()} reading, the clock the waits use. */
  private static long elapsedMillis(long startTime)
  {
    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
  }

  /** The CSN of a message the collocated replica announces, as PendingChanges generates it. */
  private static CSN newOfflineCSN()
  {
    return new CSNGenerator(LOCAL_DS_ID, 0).newCSN();
  }

  private static boolean sleepQuietly(long millis)
  {
    try
    {
      Thread.sleep(millis);
      return true;
    }
    catch (InterruptedException e)
    {
      Thread.currentThread().interrupt();
      return false;
    }
  }

  /** Teardown must never mask the primary assertion failure. */
  private void removeQuietly(ReplicationServer replicationServer)
  {
    try
    {
      remove(replicationServer);
    }
    catch (Exception ignored)
    {
    }
  }

  private void joinQuietly(Thread thread)
  {
    if (thread != null)
    {
      try
      {
        thread.join(SOCKET_TIMEOUT_MS);
      }
      catch (InterruptedException e)
      {
        Thread.currentThread().interrupt();
      }
    }
  }

  private void closeQuietly(FakePeerReplicationServer peer)
  {
    if (peer != null)
    {
      peer.close();
    }
  }

  /**
   * Establishes a connected session pair over the given listen socket, as a remote server
   * connecting to the RS would. The TLS negotiation performed by the session factories needs both
   * ends handshaking at the same time, so the client end runs on its own thread.
   *
   * @return the two sessions: the remote (client) end first, then the local (server) end to hand
   *         to the handler under test
   */
  private Session[] connectSessionPair(ServerSocket listenSocket, final ReplSessionSecurity security)
      throws Exception
  {
    final Socket clientSocket = new Socket("127.0.0.1", listenSocket.getLocalPort());
    clientSocket.setTcpNoDelay(true);
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    Future<Session> clientEnd = null;
    Socket serverSocket = null;
    Session serverEnd = null;
    boolean connected = false;
    try
    {
      clientEnd = executor.submit(new Callable<Session>()
      {
        @Override
        public Session call() throws Exception
        {
          return security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS);
        }
      });

      serverSocket = listenSocket.accept();
      serverSocket.setTcpNoDelay(true);
      serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS);
      assertThat(serverEnd).as("could not create a session for the handler under test").isNotNull();

      final Session[] sessionPair =
          new Session[] { clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
      connected = true;
      return sessionPair;
    }
    finally
    {
      if (!connected)
      {
        // Nobody owns either end yet: close whatever they managed to create.
        closeClientEndQuietly(clientEnd, clientSocket);
        closeServerEndQuietly(serverEnd, serverSocket);
      }
      executor.shutdown();
    }
  }

  private void closeServerEndQuietly(Session serverEnd, Socket serverSocket)
  {
    if (serverEnd != null)
    {
      serverEnd.close();
    }
    else
    {
      StaticUtils.close(serverSocket);
    }
  }

  private void closeClientEndQuietly(Future<Session> clientEnd, Socket clientSocket)
  {
    if (clientEnd != null)
    {
      try
      {
        final Session session = clientEnd.get(DISCARDED_SESSION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
        if (session != null)
        {
          session.close();
        }
      }
      catch (Exception ignored)
      {
        clientEnd.cancel(true);
      }
    }
    StaticUtils.close(clientSocket);
  }

  /**
   * A peer replication server which connects to the replication server under test and completes
   * the handshake, so that the handler it leaves behind on the domain has a real writer and can
   * actually forward what the domain pushes to it.
   */
  private static final class FakePeerReplicationServer
  {
    private final Session session;
    private final ExecutorService reader = Executors.newSingleThreadExecutor();
    /** Why the peer stopped reading, so that a missing message can be told from a failed one. */
    private volatile Exception readerFailure;

    FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId)
        throws Exception
    {
      final Socket socket = new Socket();
      Session newSession = null;
      boolean handshaken = false;
      try
      {
        socket.setTcpNoDelay(true);
        socket.connect(new InetSocketAddress("127.0.0.1", replicationPort), SOCKET_TIMEOUT_MS);
        newSession = getReplSessionSecurity().createClientSession(socket, SOCKET_TIMEOUT_MS);

        final String serverURL = "127.0.0.1:" + socket.getLocalPort();
        final byte groupId = (byte) 1;
        newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, 100,
            new ServerState(), generationId, false, groupId, 5000));
        final ReplServerStartMsg inStartMsg =
            waitForSpecificMsg(newSession, ReplServerStartMsg.class);
        if (!inStartMsg.getSSLEncryption())
        {
          newSession.stopEncryption();
        }
        newSession.publish(new TopologyMsg(null,
            newArrayList(new RSInfo(serverId, serverURL, generationId, groupId, 1))));
        waitForSpecificMsg(newSession, TopologyMsg.class);
        handshaken = true;
      }
      finally
      {
        if (!handshaken)
        {
          // The caller has no handle on this peer yet, so nothing else would close it.
          reader.shutdownNow();
          if (newSession != null)
          {
            newSession.close();
          }
          else
          {
            StaticUtils.close(socket);
          }
        }
      }
      session = newSession;
    }

    /** Returns the first ReplicaOfflineMsg this peer receives, or null if its session ends first. */
    Future<ReplicaOfflineMsg> receiveReplicaOfflineMsg()
    {
      return reader.submit(new Callable<ReplicaOfflineMsg>()
      {
        @Override
        public ReplicaOfflineMsg call()
        {
          try
          {
            while (true)
            {
              final ReplicationMsg msg = session.receive();
              if (msg instanceof ReplicaOfflineMsg)
              {
                return (ReplicaOfflineMsg) msg;
              }
            }
          }
          catch (Exception e)
          {
            // The session is closed when the replication server completes its shutdown: whatever
            // has not arrived by then never will.
            readerFailure = e;
            return null;
          }
        }
      });
    }

    /** Returns what ended the read of this peer, null if nothing did. */
    Exception readerFailure()
    {
      return readerFailure;
    }

    void close()
    {
      reader.shutdownNow();
      session.close();
    }
  }
}

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

New file
@@ -0,0 +1,293 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.service;

import static java.util.Arrays.asList;
import static org.assertj.core.api.Assertions.assertThat;

import java.util.concurrent.TimeUnit;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

/** Test the {@link DSRSShutdownSync} class. */
@SuppressWarnings("javadoc")
public class DSRSShutdownSyncTest extends DirectoryServerTestCase
{
  /** Short grace period, for the contracts a test has to wait out. */
  private static final long GRACE_PERIOD = 500;
  /**
   * Grace period for the contracts which only read the state: long enough that no scheduling
   * pause between announcing a message and reading the state can expire it.
   */
  private static final long LONG_GRACE_PERIOD = 60000;
  /** Time given to the forwarding thread before it forwards the message of one domain. */
  private static final long FORWARD_DELAY = 200;
  private static final int SERVER_ID = 1;
  private static final int OTHER_SERVER_ID = 2;

  private static DN baseDN1;
  private static DN baseDN2;

  @BeforeClass
  public static void classSetup() throws Exception
  {
    baseDN1 = DN.valueOf("dc=example,dc=com");
    baseDN2 = DN.valueOf("dc=world,dc=company");
  }

  @Test
  public void canShutdownWhenNoReplicaOfflineMsgWasSent() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  @Test
  public void cannotShutdownUntilTheReplicaOfflineMsgIsForwarded() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);

    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  @Test
  public void canShutdownOnceTheReplicaOfflineMsgIsForwarded() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  @Test
  public void canShutdownOnceTheGracePeriodExpired() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD);

    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
    Thread.sleep(GRACE_PERIOD + 50);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * A message sent earlier in the life of the process - an online import, a restore, a
   * configuration change - must not consume the grace period of the message sent by the
   * shutdown this class exists for.
   */
  @Test
  public void gracePeriodOfAShutdownIsNotSpentByAnEarlierMessage() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD);

    // an import disables then re-enables the replication service
    final CSN sentByTheImport = newCSN(SERVER_ID, 1);
    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheImport);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport);
    Thread.sleep(GRACE_PERIOD + 50);

    // the shutdown of the process, much later
    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID, 2));

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  /**
   * The message of an earlier announcement may still be queued behind a backlog when the
   * shutdown announces the replica offline again. Forwarding that older message says nothing
   * about the one the shutdown is waiting for, so it must not end the wait.
   */
  @Test
  public void aStaleForwardDoesNotConsumeTheGracePeriodOfANewerMessage() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN queuedByAnEarlierImport = newCSN(SERVER_ID, 1);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, queuedByAnEarlierImport);
    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();

    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  @Test
  public void gracePeriodIsCountedPerDomain() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD);

    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
    Thread.sleep(GRACE_PERIOD + 50);
    shutdownSync.replicaOfflineMsgSent(baseDN2, newCSN(SERVER_ID));

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
    assertThat(shutdownSync.canShutdown(baseDN2)).isFalse();
  }

  /**
   * A replication server relays the ReplicaOfflineMsg of every replica connected to it, so the
   * forward of another replica's message must not release the shutdown of this one.
   */
  @Test
  public void theForwardOfAnotherReplicasMessageDoesNotEndTheWait() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);

    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID));

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  /** The domain waits for the message of every one of its replicas, not for the first of them. */
  @Test
  public void aReplicaWhichIsStillWaitingHoldsBackTheShutdownOfItsDomain() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN ofOneReplica = newCSN(SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(OTHER_SERVER_ID));
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica);

    assertThat(shutdownSync.canShutdown(baseDN1)).isFalse();
  }

  @Test
  public void canShutdownOnceEveryReplicaOfTheDomainIsForwarded() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN ofOneReplica = newCSN(SERVER_ID);
    final CSN ofTheOtherReplica = newCSN(OTHER_SERVER_ID);

    shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheOtherReplica);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica);
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica);

    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * The domains of a shutdown wait together: the wait ends when the message of every one of them
   * has been forwarded, not when the first one has. Waiting for them one after the other would
   * leave the domains which come later without a grace period at all, since the wait of the
   * first one spends the deadline they share.
   */
  @Test
  public void oneWaitCoversEveryDomainOfTheShutdown() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN ofTheFirstDomain = newCSN(SERVER_ID, 1);
    final CSN ofTheSecondDomain = newCSN(SERVER_ID, 2);
    shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheFirstDomain);
    shutdownSync.replicaOfflineMsgSent(baseDN2, ofTheSecondDomain);
    final Thread forwarder =
        newForwarderThread(shutdownSync, ofTheFirstDomain, ofTheSecondDomain);

    final long startTime = System.nanoTime();
    forwarder.start();
    shutdownSync.awaitReplicaOfflineMsgsForwarded(
        asList(baseDN1, baseDN2), shutdownSync.newShutdownDeadline());
    final long elapsed = millisSince(startTime);
    forwarder.join();

    assertThat(elapsed)
        .as("the wait ended on the first domain forwarded, leaving the second one nothing")
        .isGreaterThanOrEqualTo(2 * FORWARD_DELAY);
    assertThat(elapsed).isLessThan(LONG_GRACE_PERIOD);
  }

  /**
   * However long the messages of a shutdown may still hold it back, the deadline the shutdown
   * was given bounds the wait.
   */
  @Test
  public void theWaitIsBoundedByTheDeadlineOfTheShutdown() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID));
    shutdownSync.replicaOfflineMsgSent(baseDN2, newCSN(SERVER_ID));
    final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(GRACE_PERIOD);

    final long startTime = System.nanoTime();
    shutdownSync.awaitReplicaOfflineMsgsForwarded(asList(baseDN1, baseDN2), deadline);
    final long elapsed = millisSince(startTime);

    assertThat(elapsed).isGreaterThanOrEqualTo(GRACE_PERIOD - 50);
    assertThat(elapsed)
        .as("the wait outlived the deadline of the shutdown")
        .isLessThan(2 * GRACE_PERIOD);
  }

  /** Forwards the message of the first domain, then, as long again later, of the second one. */
  private Thread newForwarderThread(final DSRSShutdownSync shutdownSync,
      final CSN ofTheFirstDomain, final CSN ofTheSecondDomain)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        try
        {
          Thread.sleep(FORWARD_DELAY);
          shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain);
          Thread.sleep(FORWARD_DELAY);
          shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain);
        }
        catch (InterruptedException e)
        {
          Thread.currentThread().interrupt();
        }
      }
    });
  }

  private static long millisSince(long startTime)
  {
    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
  }

  /**
   * The CSN of a message a replica announced. They are built by hand rather than with a
   * CSNGenerator: this class has no state to share with the server, and a generator would tie
   * the test to the time service the server starts.
   */
  private static CSN newCSN(int serverId)
  {
    return newCSN(serverId, 1);
  }

  private static CSN newCSN(int serverId, int seqNum)
  {
    return new CSN(1, seqNum, serverId);
  }
}