[#950] Announce a ReplicaOfflineMsg before it is published, not after it ma...

Valery Kharseko

15 hours ago 600df926522919b63bfca2816ef9588b6f1c6e34

[#950] Announce a ReplicaOfflineMsg before it is published, not after it may have been forwarded (#978)

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

@@ -828,7 +828,8 @@
     * The generator time is adjusted to the time of the last CSN received from
     * remote other servers.
     */
    pendingChanges = new PendingChanges(getGenerator(), this);
    pendingChanges = new PendingChanges(getGenerator(), this,
        new ShutdownSyncAnnouncer(dsrsShutdownSync, getBaseDN()));
    remotePendingChanges = new RemotePendingChanges(getServerState());

    // listen for changes on the configuration
@@ -2271,17 +2272,13 @@
  public void publishReplicaOfflineMsg()
  {
    final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
    if (offlineCSN != null)
    if (offlineCSN == null && logger.isTraceEnabled())
    {
      /*
       * Only a message which really was published is announced: the shutdown of a collocated
       * replication server waits for it to be forwarded, and would spend the whole grace
       * period waiting for one which never reached the wire.
       * The announcement itself is made where the message is published, so nothing has to be
       * reported here: a message a change in flight held back was never announced, and the
       * announcement of one the broker refused has been withdrawn.
       */
      dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN);
    }
    else if (logger.isTraceEnabled())
    {
      logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
          + " could not announce itself offline: the message was not published - a change which"
          + " is still in flight holds it back, or the broker had no session to write it to,"

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

@@ -53,6 +53,9 @@
  /** The ReplicationDomain that will be used to send UpdateMsg. */
  private final ReplicationDomain domain;

  /** Told that the replica of this domain announces itself offline, or takes that back. */
  private final ReplicaOfflineAnnouncer replicaOfflineAnnouncer;

  private boolean recoveringOldChanges;

  /**
@@ -60,11 +63,16 @@
   *
   * @param csnGenerator The CSNGenerator to use to create new unique CSNs.
   * @param domain  The ReplicationDomain that will be used to send UpdateMsg.
   * @param replicaOfflineAnnouncer Told that the replica of this domain announces itself
   *                  offline, before the message announcing it is published, and that it takes
   *                  the announcement back when the broker refused the message.
   */
  PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain)
  PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain,
      ReplicaOfflineAnnouncer replicaOfflineAnnouncer)
  {
    this.csnGenerator = csnGenerator;
    this.domain = domain;
    this.replicaOfflineAnnouncer = replicaOfflineAnnouncer;
  }

  /**
@@ -198,9 +206,22 @@
      }
      else if (msg instanceof ReplicaOfflineMsg)
      {
        /*
         * Announce the replica offline before the message reaches the wire, and not after:
         * a collocated replication server forwards it as soon as it has it, and a forward
         * which finds nothing announced leaves the shutdown waiting out the whole grace
         * period of a message the topology already has.
         */
        final CSN offlineCSN = msg.getCSN();
        replicaOfflineAnnouncer.announce(offlineCSN);
        if (domain.publish(msg))
        {
          publishedOfflineCSN = msg.getCSN();
          publishedOfflineCSN = offlineCSN;
        }
        else
        {
          // The broker wrote it to no session, so nobody will forward what was announced.
          replicaOfflineAnnouncer.withdraw(offlineCSN);
        }
      }

@@ -266,4 +287,34 @@
    }
    return recoveringOldChanges;
  }

  /**
   * Told that the replica of this domain announces itself offline, or takes that back.
   * <p>
   * A collocated replication server can forward a {@link ReplicaOfflineMsg} as soon as it is on
   * the wire, and its shutdown waits for that forward, so the announcement has to be in place
   * before the message is published: one made afterwards is one the forward found nothing to
   * clear, and the shutdown spends its whole grace period on a message which has already gone
   * out. The broker may still refuse the message once it is announced, and then the
   * announcement is withdrawn: what stays announced is what the broker reports as written.
   */
  interface ReplicaOfflineAnnouncer
  {
    /**
     * Announces that the replica goes offline at the provided CSN.
     *
     * @param offlineCSN
     *          the CSN of the ReplicaOfflineMsg which is about to be published
     */
    void announce(CSN offlineCSN);

    /**
     * Withdraws the announcement of a message the broker refused: it was written to no session,
     * so nobody will forward it.
     *
     * @param offlineCSN
     *          the CSN of the ReplicaOfflineMsg which was announced and not published
     */
    void withdraw(CSN offlineCSN);
  }
}

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

New file
@@ -0,0 +1,62 @@
/*
 * 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.plugin;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.plugin.PendingChanges.ReplicaOfflineAnnouncer;
import org.opends.server.replication.service.DSRSShutdownSync;

/**
 * Announces the replica of one domain offline to the {@link DSRSShutdownSync} the shutdown of a
 * collocated replication server waits on, and takes such an announcement back.
 * <p>
 * This is the announcer {@link LDAPReplicationDomain} hands its {@link PendingChanges}: the
 * announcement goes through {@link DSRSShutdownSync#replicaOfflineMsgSent(DN, CSN)} and the
 * withdrawal through {@link DSRSShutdownSync#replicaOfflineMsgNotSent(DN, CSN)}, for the domain
 * the announcer was built for.
 */
final class ShutdownSyncAnnouncer implements ReplicaOfflineAnnouncer
{
  private final DSRSShutdownSync shutdownSync;
  private final DN baseDN;

  /**
   * Creates an announcer for the replica of one domain.
   *
   * @param shutdownSync
   *          the synchronization object the collocated replication server's shutdown waits on
   * @param baseDN
   *          the domain whose replica announces itself
   */
  ShutdownSyncAnnouncer(DSRSShutdownSync shutdownSync, DN baseDN)
  {
    this.shutdownSync = shutdownSync;
    this.baseDN = baseDN;
  }

  @Override
  public void announce(CSN offlineCSN)
  {
    shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
  }

  @Override
  public void withdraw(CSN offlineCSN)
  {
    shutdownSync.replicaOfflineMsgNotSent(baseDN, offlineCSN);
  }
}

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

@@ -46,7 +46,7 @@
{
  /**
   * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated
   * RS, in milliseconds, counted from the moment the message was sent.
   * RS, in milliseconds, counted from the moment the message was announced.
   */
  public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000;

@@ -97,19 +97,90 @@
  }

  /**
   * Message has been sent.
   * Message is about to be sent.
   * <p>
   * The announcement comes before the message is published rather than after: a collocated
   * replication server can forward the message as soon as it is on the wire, and a forward which
   * finds nothing announced has nothing to clear. The announcement of a message the broker then
   * refuses is taken back by {@link #replicaOfflineMsgNotSent(DN, CSN)}.
   * <p>
   * A replica announces itself offline on every disableService(), so this may take the place of
   * an earlier announcement of the same replica which is still owed its forward. The earlier one
   * is kept behind the new one: a forward of the newer message, which the replication server
   * queued behind the earlier one, covers both, and a withdrawal of the newer one gives the
   * earlier one its wait back. It is kept only while its own grace period runs: past it, the
   * announcement holds nothing back any more, and keeping it would chain every announcement of
   * a replica whose message nobody in this process forwards - a directory server without a
   * collocated replication server, or connected to a remote one - for the life of the process.
   *
   * @param baseDN
   *          the domain for which the message has been sent
   *          the domain for which the message is being sent
   * @param offlineCSN
   *          the CSN of the message, which identifies both the replica which announced itself
   *          the CSN of the message, which identifies both the replica which announces itself
   *          offline and the announcement being waited for
   */
  public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN)
  {
    final long announcedAt = System.nanoTime();
    replicaOfflineMsgs
        .computeIfAbsent(baseDN, dn -> new ConcurrentHashMap<Integer, PendingOfflineMsg>())
        .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime()));
        .compute(offlineCSN.getServerId(), (serverId, displaced) ->
            new PendingOfflineMsg(offlineCSN, announcedAt,
                displaced != null && gracePeriodLeft(displaced, announcedAt) > 0 ? displaced : null));
  }

  /**
   * The message which was announced was not sent after all: the broker had no session to write
   * it to, or was stopped before it could.
   * <p>
   * The announcement is made before the message is published, since a collocated replication
   * server can forward it as soon as it is on the wire, so the announcement of a message the
   * broker then refused has to be taken back: nobody will forward it, and the shutdown would
   * spend the whole grace period waiting for that forward. Only the announcement carrying that
   * CSN is withdrawn, and the announcement it displaced - an earlier message of the same replica
   * which did go out and is still owed its forward - takes its place again.
   * <p>
   * Whatever is reported about that earlier message while the announcement of the refused one
   * stands in its place is not seen by it. A forward, or the loss of a peer it was queued for,
   * is lost, and the shutdown then waits out what is left of the earlier message's own grace
   * period; the peers it is queued for, if they are recorded in that window, are lost too, with
   * the opposite effect - the first forward ends its wait, as for a message no peer was recorded
   * for. That window is the one publish the broker refuses: at once on a connection error or a
   * pending recovery, the broker's retry loop up to the reconnect when it has no session. The
   * wait it can cost is bounded by a grace period which is already running.
   *
   * @param baseDN
   *          the domain for which the message was announced
   * @param offlineCSN
   *          the CSN of the message which was not sent
   */
  public void replicaOfflineMsgNotSent(DN baseDN, CSN offlineCSN)
  {
    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
    if (msgs != null)
    {
      final int serverId = offlineCSN.getServerId();
      final PendingOfflineMsg pending = msgs.get(serverId);
      if (pending != null && pending.csn.equals(offlineCSN))
      {
        /*
         * The displaced announcement may owe nothing any more: the forward which released it
         * can have been reported while this announcement was being made, so that its remove(),
         * which matches the entry it read, found this one in its place. Given its place back,
         * such an announcement would hold the shutdown for the rest of its grace period, since
         * nobody will report that forward again.
         */
        if (pending.displaced != null && !pending.displaced.isFullyForwarded())
        {
          msgs.replace(serverId, pending, pending.displaced);
        }
        else
        {
          msgs.remove(serverId, pending);
        }
      }
    }
    notifyForwarded();
  }

  /**
@@ -338,12 +409,21 @@
    long remaining = 0;
    for (PendingOfflineMsg pending : msgs.values())
    {
      remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime));
      remaining = Math.max(remaining, gracePeriodLeft(pending, now));
    }
    return remaining;
  }

  /**
   * Returns the time left, in milliseconds, of the grace period of one announcement, zero or
   * less once it has expired.
   */
  private long gracePeriodLeft(PendingOfflineMsg pending, long now)
  {
    return gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime);
  }

  /**
   * A ReplicaOfflineMsg a replica announced and which has not been forwarded yet.
   * <p>
   * This deliberately does not override {@code equals}: the two-argument
@@ -357,15 +437,22 @@
    /** When the message was announced, on the {@link System#nanoTime()} clock. */
    private final long sentTime;
    /**
     * The announcement of the same replica this one took the place of and which is still owed its
     * forward, null when there was none or when its grace period had already expired. It is
     * given its place back if this message is withdrawn.
     */
    private final PendingOfflineMsg displaced;
    /**
     * The replication servers the message was queued for and which have not forwarded it yet,
     * null as long as it has not been queued for anybody.
     */
    private volatile Set<Integer> awaitedForwarders;

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

    /**
@@ -391,9 +478,8 @@
      {
        /*
         * The message never went through the collocated RS - a replica which picked a remote one
         * announcing itself offline, or an announcement recorded after the message it belongs to
         * was already relayed. Nobody is known to owe a forward, so keep the behaviour the wait
         * had before the recipients were tracked: the first forward ends it.
         * is announcing itself offline. Nobody is known to owe a forward, so keep the behaviour
         * the wait had before the recipients were tracked: the first forward ends it.
         */
        return true;
      }
@@ -412,6 +498,16 @@
      return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
    }

    /**
     * Returns whether every replication server the message was queued for has forwarded it, or
     * has been given up on: nothing is left to wait for. False while no recipient is known.
     */
    private boolean isFullyForwarded()
    {
      final Set<Integer> awaited = awaitedForwarders;
      return awaited != null && awaited.isEmpty();
    }

    @Override
    public String toString()
    {

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

@@ -16,12 +16,14 @@
package org.opends.server.replication.plugin;

import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;

import org.forgerock.opendj.ldap.DN;
import org.mockito.ArgumentCaptor;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
@@ -29,23 +31,36 @@
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.replication.service.ReplicationDomain;
import org.opends.server.types.operation.PluginOperation;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

/**
 * Tests the bookkeeping a replica does on its own changes: they are published in the order of
 * their CSNs, and the announcement that the replica goes offline is only reported as sent when
 * it really was published.
 * their CSNs, and the announcement that the replica goes offline is made before the message is
 * published, since the shutdown of a collocated replication server waits for that message to be
 * forwarded - and stands only for a message the broker reports as written.
 * <p>
 * These tests need no server: the changes are built by a CSNGenerator, which reads the time
 * service, and the time service is up as soon as its class is loaded.
 * These tests need no server: the changes are built by a CSNGenerator, which needs nothing but
 * a server id.
 */
@SuppressWarnings("javadoc")
@Test(groups = { "precommit", "replication" }, sequential = true)
public class PendingChangesTest extends DirectoryServerTestCase
{
  private static final int SERVER_ID = 42;
  /** A peer replication server the collocated one relays the message to. */
  private static final int RS_ID = 11;

  private static DN baseDN;

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

  @Test
  public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Exception
@@ -62,6 +77,47 @@
  }

  /**
   * The collocated replication server forwards the message as soon as it is on the wire, so an
   * announcement made after the publish is one the forward found nothing to clear: nothing will
   * ever remove it, and the shutdown waits out its whole grace period for a message which the
   * topology already has.
   * <p>
   * The forward is reported from inside publish(), which is where the message reaches the
   * session, so the race is reproduced rather than waited for.
   */
  @Test
  public void theReplicaOfflineMsgIsAnnouncedBeforeItIsPublished() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    forwardWhilePublishing(domain, shutdownSync);
    final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync);

    pendingChanges.putReplicaOfflineMsg();

    assertTrue(shutdownSync.canShutdown(baseDN),
        "the message was forwarded, so nothing must hold the shutdown back any longer");
  }

  /**
   * The announcement of a message the broker took stands until a peer forwards it: the
   * withdrawal is for the message the broker refused, and an announcement taken back after a
   * publish which succeeded would leave the shutdown nothing to wait for.
   */
  @Test
  public void theAnnouncementOfAPublishedMessageStandsUntilItIsForwarded() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    final PendingChanges pendingChanges =
        newPendingChanges(domainWhichPublishes(true), shutdownSync);

    pendingChanges.putReplicaOfflineMsg();

    assertFalse(shutdownSync.canShutdown(baseDN),
        "the message went out and nobody has forwarded it yet, so the shutdown must wait for it");
  }

  /**
   * The broker writes nothing when it has no usable session, when the changes which come before
   * this one still have to be republished by the recovery, or when it is stopped in between - and
   * what was not written must not be reported as sent: the shutdown of a collocated replication
@@ -80,6 +136,27 @@
  }

  /**
   * The announcement is made before the message is published, and the broker may refuse it
   * once it is: an announcement which stayed would be one nobody will ever forward, and the
   * shutdown would wait out its whole grace period for a message which never left. It is
   * therefore withdrawn - and it is a withdrawal, not an announcement which was never made: the
   * shutdown is held back while the broker holds the message.
   */
  @Test
  public void theAnnouncementOfAReplicaOfflineMsgTheBrokerRefusedIsWithdrawn() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    final ReplicationDomain domain = mock(ReplicationDomain.class);
    refuseWhilePublishing(domain, shutdownSync);
    final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync);

    assertNull(pendingChanges.putReplicaOfflineMsg(), "the broker refused the message");

    assertTrue(shutdownSync.canShutdown(baseDN),
        "the message never reached the wire, so nothing must hold the shutdown back");
  }

  /**
   * The message carries the newest CSN of the replica, so a change which is still in flight
   * holds it back, and the broker is never even asked to publish it.
   */
@@ -117,6 +194,32 @@
  }

  /**
   * The announcement follows the publication rather than the queueing, so a message which a
   * change in flight holds back is not announced: neither while it waits, nor when the change
   * which held it back completes and the message is given up on. Announcing it either time
   * would leave the shutdown waiting out its whole grace period for a forward which cannot
   * happen.
   */
  @Test
  public void theReplicaOfflineMsgHeldBackByAChangeInFlightIsNeverAnnounced() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
    final ReplicationDomain domain = domainWhichPublishes(true);
    final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync);
    final CSN inFlight = pendingChanges.putLocalOperation(newLocalOperation());

    pendingChanges.putReplicaOfflineMsg();

    assertTrue(shutdownSync.canShutdown(baseDN),
        "the message is still queued behind a change in flight, and nothing was announced");

    pendingChanges.commitAndPushCommittedChanges(inFlight, mock(LDAPUpdateMsg.class));

    assertTrue(shutdownSync.canShutdown(baseDN),
        "the message was given up on with the change which held it back, and never announced");
  }

  /**
   * A change the broker refused leaves the pending changes all the same: the replica has done
   * it, its ServerState says so, and it is by finding that state ahead of the one its
   * replication server reports that the next session republishes the change from the historical
@@ -137,9 +240,55 @@
    assertEquals(pendingChanges.size(), 0, "and is not queued for a second attempt");
  }

  /**
   * Reports the forward of the message from within the publish which puts it on the wire, after
   * checking from there that the announcement is already in place, and publishes it. The forward
   * is the one of a peer the message was never recorded as queued for, which is what a forward
   * racing the announcement looks like.
   */
  private void forwardWhilePublishing(
      final ReplicationDomain domain, final DSRSShutdownSync shutdownSync)
  {
    doAnswer(invocation -> {
      final UpdateMsg msg = (UpdateMsg) invocation.getArguments()[0];
      if (msg instanceof ReplicaOfflineMsg)
      {
        assertFalse(shutdownSync.canShutdown(baseDN),
            "the message must be announced before it is published");
        shutdownSync.replicaOfflineMsgForwarded(baseDN, msg.getCSN(), RS_ID);
      }
      return true;
    }).when(domain).publish(any(UpdateMsg.class));
  }

  /**
   * Refuses to publish the message, the way a broker with no usable session does, after checking
   * from within the publish that the announcement is already in place.
   */
  private void refuseWhilePublishing(
      final ReplicationDomain domain, final DSRSShutdownSync shutdownSync)
  {
    doAnswer(invocation -> {
      assertFalse(shutdownSync.canShutdown(baseDN),
          "the message must be announced before it is published");
      return false;
    }).when(domain).publish(any(UpdateMsg.class));
  }

  private PendingChanges newPendingChanges(ReplicationDomain domain)
  {
    return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain);
    return newPendingChanges(domain, new DSRSShutdownSync());
  }

  /**
   * The pending changes of a replica whose domain announces itself through the shutdown sync,
   * with the announcer LDAPReplicationDomain hands its own pending changes.
   */
  private PendingChanges newPendingChanges(
      final ReplicationDomain domain, final DSRSShutdownSync shutdownSync)
  {
    return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain,
        new ShutdownSyncAnnouncer(shutdownSync, baseDN));
  }

  /** A domain whose broker accepts, or refuses, whatever it is given to publish. */

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

@@ -86,6 +86,150 @@
    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * The announcement is made before the message is published, so the broker may still refuse
   * it - no usable session, or stopped in between. What was announced and never written must
   * not hold the shutdown back: nobody will forward it.
   */
  @Test
  public void canShutdownOnceTheReplicaOfflineMsgIsWithdrawn() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);

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

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

  /**
   * A withdrawal takes back the very announcement it names, and not whatever the replica
   * announced last: a stale one, of a message the replica has since announced again, is ignored,
   * and the newer announcement is still owed its forward. Ignored, and not taken for the
   * withdrawal of the newer one: that would put the stale announcement back in front, where the
   * forward of its message - which says nothing about the newer one - would end the wait.
   */
  @Test
  public void theWithdrawalOfAnEarlierMessageLeavesANewerOneAlone() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN refusedByTheBroker = newCSN(SERVER_ID, 1);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);

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

    shutdownSync.replicaOfflineMsgForwarded(baseDN1, refusedByTheBroker, RS_ID);
    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("the stale withdrawal was ignored, not turned into a restore")
        .isFalse();
  }

  /**
   * A replica announces itself offline on every disableService(), and a later announcement
   * takes the place of the earlier one. When the broker then refuses the later message, the
   * earlier one - which did go out, and which a peer still has to forward - must get its wait
   * back: withdrawing the later announcement must not take the earlier one with it.
   */
  @Test
  public void theWithdrawalOfALaterMessageGivesTheEarlierOneItsWaitBack() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
    final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
    shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);

    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("the earlier message went out and nobody has forwarded it yet")
        .isFalse();
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("the forward of the earlier message ends the wait")
        .isTrue();
  }

  /**
   * What the withdrawal gives back is the very announcement which was displaced, with the peers
   * its message was queued for: the forward of one of them does not end a wait which is for
   * several, as it would for a message which was announced again from scratch.
   */
  @Test
  public void theRestoredAnnouncementIsStillOwedTheForwardsItWasQueuedFor() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
    final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgDispatched(
        baseDN1, sentByTheShutdown, asList(RS_ID, OTHER_RS_ID));
    shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
    shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);

    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("the restored message is still owed the other peer's forward")
        .isFalse();
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, OTHER_RS_ID);
    assertThat(shutdownSync.canShutdown(baseDN1)).isTrue();
  }

  /**
   * The restored announcement keeps its own clock: the shutdown waits out what is left of the
   * earlier message's grace period, not a new one counted from the withdrawal.
   */
  @Test
  public void theRestoredAnnouncementKeepsWhatIsLeftOfItsOwnGracePeriod() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
    final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    Thread.sleep(GRACE_PERIOD - 200);
    shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
    shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);
    // past the end of the earlier message's grace period, well short of a whole new one
    Thread.sleep(250);

    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("the wait started over at the withdrawal")
        .isTrue();
  }

  /**
   * While the announcement of a refused message stands in the place of the earlier one, what is
   * reported about the earlier message is not seen by it: a forward reported in that window is
   * lost, and the restored announcement waits out what is left of its own grace period. The
   * window is the one refused publish; this pins the trade-off, so that a change to it is made
   * knowingly.
   */
  @Test
  public void aForwardReportedWhileARefusedAnnouncementStoodIsNotSeen() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN sentByTheShutdown = newCSN(SERVER_ID, 1);
    final CSN refusedByTheBroker = newCSN(SERVER_ID, 2);

    shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown);
    shutdownSync.replicaOfflineMsgDispatched(baseDN1, sentByTheShutdown, asList(RS_ID));
    shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker);
    // not seen: the announcement of the refused message stands in front
    shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID);
    shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker);

    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("a forward reported while the refused announcement stood is not seen")
        .isFalse();
  }

  @Test
  public void canShutdownOnceTheGracePeriodExpired() throws Exception
  {
@@ -430,6 +574,55 @@
        .isLessThan(LONG_GRACE_PERIOD);
  }

  /**
   * A withdrawal must wake the shutdown up as a forward does, and not leave it waiting for the
   * forward of a message which never left.
   */
  @Test
  public void theWaitEndsWhenTheMessageIsWithdrawn() throws Exception
  {
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD);
    final CSN offlineCSN = newCSN(SERVER_ID);
    shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN);
    final Thread withdrawer = newWithdrawerThread(shutdownSync, offlineCSN);

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

    assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY);
    assertThat(elapsed)
        .as("the withdrawal did not wake the wait up")
        .isLessThan(LONG_GRACE_PERIOD);
    assertThat(shutdownSync.canShutdown(baseDN1))
        .as("the withdrawn message holds nothing back")
        .isTrue();
  }

  /** Withdraws the announcement, as the broker refusing the message during the wait does. */
  private Thread newWithdrawerThread(final DSRSShutdownSync shutdownSync, final CSN offlineCSN)
  {
    return new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        try
        {
          Thread.sleep(FORWARD_DELAY);
          shutdownSync.replicaOfflineMsgNotSent(baseDN1, offlineCSN);
        }
        catch (InterruptedException e)
        {
          Thread.currentThread().interrupt();
        }
      }
    });
  }

  /** Stops the peer the message was queued for, as a disconnection during the wait does. */
  private Thread newPeerStopperThread(final DSRSShutdownSync shutdownSync, final int peerId)
  {