[#917] Wait for every peer replication server to forward the ReplicaOfflineMsg

[vharseko]

Fixes #917

The bug

DSRSShutdownSync kept one pending entry per domain and per replica, and the first
ServerWriter to publish the message cleared it for everybody.
ReplicationServerDomain.put() queues a ReplicaOfflineMsg for every connected peer RS,
and each of them is served by its own writer, so the faster one released the wait while the
other still had the message in its queue. ReplicationServer.shutdown() then went on to
stopAllServers(true), which clears that queue (MessageHandler.shutdown()) and closes the
session, and the peer never learned the replica went offline.

The window is not scheduling jitter. The writer of a lagging peer is behind up to
queue-size (10000 by default) messages of its own, behind Session.sendQueue and behind TCP,
and blocks in acquirePermitInSendWindow() when its send window is closed - so the grace period
was systematically granted to the peer which never needed it and denied to the one it exists
for.

What it costs is more than the notification. ChangeNumberIndexer.getOldestLastAliveCSN()
excludes offline replicas from the medium consistency point, and
MultiDomainServerState.getOldestCSNExcluding() computes that point across all ECL enabled
domains - so one replica a peer was never told about pins the change number index and the
external changelog of that peer for every domain it serves, until the replica writes again. The
peer has no other way to learn it: notifyReplicaOffline is only reached from the forwarded
message (ReplicationServerDomain.publishUpdateMsg()), a TopologyMsg does not carry it, and
nothing is recorded in that peer's own changelog state, so its own restart does not help either.

The fix

The pending entry knows which replication servers the message was queued for, and is released
only once all of them have forwarded it.

• Recipients recorded before the message is queued. put() collects the peer RSs which
  pass the generation id filter, hands their ids to
  DSRSShutdownSync.replicaOfflineMsgDispatched() and only then queues the message: a peer can
  forward it as soon as it is in its queue, and a forward reported before the recipients are
  known would end the wait at once. A peer the message is not queued for - none connected, a
  different generation id - is not waited for, and neither is a peer which connected after the
  message was queued, since it was never given it.
• The writer says which peer forwarded it. replicaOfflineMsgForwarded() takes the server
  id of the peer, and the entry is removed when the awaited set empties.
• A peer which will not forward what it was given is dropped from the awaited set, so a
  disconnection does not cost the whole grace period: one which stops during the window
  (stopServer(), where the give-up now sits in a finally of its own so that no exit path -
  an interrupted lock acquisition, an unregistration which throws - can skip it, since
  engageShutdown() is one-shot and no later call will retry); one whose handshake is aborted
  (unregisterFailedHandshake()); one whose teardown had already started when the recipients
  were recorded; and one whose writer drops the message on the way out - the filter in
  ServerWriter.isUpdateMsgFiltered() is wider than the one put() applies when it queues the
  message (it drops anything for a peer whose generation id is unknown), so a peer can be given
  a message which is then dropped with nothing ever reporting a forward for it.
• The two teardowns and put() are ordered on the handler's shutdown flag. Both
  stopServer() and unregisterFailedHandshake() raise it before they give up, and put()
  reads it after it has recorded the recipients. Of the two, at least one therefore always sees
  the other: a teardown whose give-up ran too early to find a recipient recorded has
  necessarily raised the flag put() goes on to read. Without that ordering an aborted
  handshake is invisible to both - its give-up is a no-op and its flag is never raised - and the
  message is queued for a handler whose queue was just cleared, which nothing will ever
  unregister, so the shutdown waits out the whole grace period.
• With no recipient recorded the wait keeps the behaviour it had: the first forward ends it.
  That covers the message of a replica which picked a remote replication server, which the
  collocated one only sees relayed, and the race of A ReplicaOfflineMsg still queued behind an uncommitted change is recorded as sent, and the shutdown then waits its whole grace period for nothing #918 in which the announcement is recorded
  after the message it belongs to was already relayed. The caller side check on
  getConnectedRSs().isEmpty() in ReplicationServer.awaitReplicaOfflineMsgsForwarded() stays
  for exactly that case: it is the only thing which keeps a server with no peer connected from
  spending its grace period on an announcement no recipient can be recorded for.

The hot path is untouched: the recipient list is built only for a ReplicaOfflineMsg, every
other update keeps the single allocation free loop.

Tests

• DSRSShutdownSyncTest (10 new, 22 total) - the forward of one peer does not end the wait of
  the others and the last one does; a message queued for nobody waits for nothing; a peer which
  stops is no longer waited for but does not release the others; the forward of a peer the
  message was not queued for changes nothing; the recipients of an earlier announcement are not
  taken for those of this one; the fallback with no recipient recorded; a peer which stops
  before the message is queued does not end the wait, which is the asymmetry between a
  forward and a give-up when nothing is recorded yet; and the wait is woken up by the peer which
  goes away.
• ReplicationServerShutdownSyncTest (5 new, 13 total). All of them drive put() through a
  real broker, and assert on the barrier itself through a DSRSShutdownSync which records what
  the production code reports to it, rather than on a duration alone:
  • theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline connects two peers through the
    real handshake, one of them advertising a send window of one so that a first change fills it
    and its writer blocks on the permit of the ReplicaOfflineMsg. It now asserts first that
    both writers reported a forward before the shutdown returned, and only then that both peers
    received the message: the first assertion cannot race, the second one can be lost to the
    Session.close() limitation of [#900] Time the ReplicaOfflineMsg grace period per replica, and spend it where the message can still be forwarded #919, and the order tells the two apart.
  • theShutdownStopsWaitingForAPeerWhoseHandshakeWasAborted aborts the handshake of a second
    peer inside the dispatch, in the exact window in which put() has read the peers of the
    domain and nothing knows yet which of them the message is for. It asserts that the aborted
    peer really was recorded among the recipients - so the window is reproduced and not merely
    hoped for - and that the shutdown does not wait for it.
  • theShutdownStopsWaitingForAPeerWhichDisconnected holds one of two peers back with a full
    send window so it cannot have forwarded anything, closes it after the message was queued for
    it, and pins that stopServer() strikes it off.
  • theMessageIsNotQueuedForAPeerWhoseGenerationIdDiffers pins the recipient side filter: a
    connected peer which no longer shares the generation id of the domain is not recorded at all,
    which is what the caller side check on getConnectedRSs() cannot see.
  • theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped now goes through put()
    instead of calling the synchronization object by hand: the peer is held back by a full send
    window, so the message is queued for it and its writer reaches the filter only once the
    window is opened again. It asserts the give-up itself, not a duration.

Every new assertion was watched failing against the code it pins, one mutation at a time:

mutation	fails with
engageShutdown() out of unregisterFailedHandshake()	theShutdownStopsWaitingForAPeerWhoseHandshakeWasAborted - the shutdown waited for a peer whose handshake had been aborted, and which nothing else will ever strike off
the give-up out of the finally of stopServer()	theShutdownStopsWaitingForAPeerWhichDisconnected - the shutdown kept waiting for a forward from a peer which had disconnected
the generation id filter out of the recipient list	theMessageIsNotQueuedForAPeerWhoseGenerationIdDiffers - the message was recorded as queued for a peer which does not share the generation id of the domain
the give-up branch out of ServerWriter	theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped - the writer dropped the message without telling the shutdown to stop waiting for the peer it was queued for
awaited != null out of giveUpOn()	aPeerStoppingBeforeTheMessageIsQueuedDoesNotEndTheWait
forwardedBy() releasing on the first forward	theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline - the wait ended on the first peer forwarding the message, without the writer of the peer which was held back ever reporting one, plus two DSRSShutdownSyncTest cases

Each of those produced exactly the failures listed and no others.

Regression: ReplicationServerTest, HandshakeAbortRegistrationTest,
HandshakeAbortGenerationIdTest, ReplicationDomainTest, MonitorTest, GenerationIdTest,
TopologyViewTest, ReplicationServerFailoverTest, StateMachineTest,
ReplicationServerDynamicConfTest, ChangelogBackendTestCase, FileChangelogDBTest,
ChangeNumberControlPluginTestCase.

Left out

• The fix turns no protection at all for the second peer into up to one shared grace period
  for each of them. A peer wedged for longer than the grace period still loses the message; the
  wait is bounded as before and no shutdown can take longer than one grace period in total.
• The wait still proves the message was handed to the Session, not that it reached the wire -
  the limitation recorded in [#900] Time the ReplicaOfflineMsg grace period per replica, and spend it where the message can still be forwarded #919. Fixing it belongs in Session: drain sendQueue before
  interrupting the publisher on close, or report the forward from the publisher thread.
• A ReplicaOfflineMsg still queued behind an uncommitted change is recorded as sent, and the shutdown then waits its whole grace period for nothing #918 keeps its half of the ordering: publishReplicaOfflineMsg() records the announcement
  after PendingChanges has already published it, so a recipient list computed before that
  record finds nothing to attach itself to. The fallback above makes that case no worse than
  master, and fixing the ordering belongs to that issue.
• The give up is keyed on the server id, like connectedRSs and unregisterServerHandler
  themselves: a peer reconnecting under the same id while an announcement is in flight can have
  the obligation of the new handler cancelled by the teardown of the old one. Narrowing that
  means keying the handler maps on the instance, which is a change of its own.
• DSRSShutdownSync.canShutdown(DN) has had no production caller since the wait moved to a
  blocking one in The ReplicaOfflineMsg grace period is spent on the first message of the process: DSRSShutdownSync latches its timestamp and never resets it #900; it is the only way to read the state without waiting for it, and most of
  the unit tests assert through it. The server tests above go through ReplicationServer.shutdown()
  instead, but retiring the method is a change of its own.
• The interrupted lock acquisition of stopServer() still leaves the handler registered and its
  queue uncleared - only the give-up is now guaranteed. That leak predates this PR.

[maximthomas]

praise: The barrier now has the same granularity as the fan-out it guards, and the details that make
that work are all present.

• Recipients are recorded before anything is queued, so a fast writer cannot report a forward for a
  recipient set that does not exist yet. put() keeps its original allocation-free loop for every other
  message type, so the hot path pays nothing.
• The forward carries the forwarder's id (replicaOfflineMsgForwarded(baseDN, csn, serverId)), and the
  give-up is reported from every place a peer can stop being able to forward — including the
  writer-side filter, which is genuinely wider than the one put() applied
  (isDifferentGenerationId is generationId > 0 && generationId != g; the writer also drops on -1).
• replicaOfflineMsgDispatched gates on pending.csn.equals(offlineCSN), so the recipients of an
  earlier announcement cannot be attached to this one — the same care #919 took on the forward path.
• The awaitedForwarders == null fallback keeps the pre-PR behaviour where no recipient was recorded
  (relay, the #918 ordering half). That is the right trade: "no worse than master" instead of a
  guaranteed 5 s stall on a path the fix does not own.
• An empty recipient list clears the entry immediately, which is strictly sharper than the caller-side
  getConnectedRSs().isEmpty() check it supersedes.
• theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline goes through the real handshake with a
  peer advertising a send window of one, so its writer really blocks on the permit. It asserts on the
  message arriving at both peers, not on a duration. That is the test this bug needed.
• The DS/RS asymmetry is applied consistently at all four new call sites, and the shared shutdown
  deadline from #919 is untouched — the whole shutdown is still bounded by one grace period.

---

issue (blocking): unregisterFailedHandshake reports its give-up before the recipients exist, and
shuttingDown() cannot see that teardown — the recipient stays awaited forever.

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

engageShutdown() has exactly one caller in the tree — stopServer() at :1119. ServerHandler.shutdown()
never sets the flag, so shuttingDown() is false for every aborted handshake. And the give-up on that path
runs while nothing is recorded yet, where giveUpOn is a deliberate no-op:

// unregisterFailedHandshake
connectedServers.remove(sHandler.getServerId(), sHandler);
if (!isDataServer) { noLongerAwaitTheForwardOf(sHandler); } // no-op: awaitedForwarders is still null
sHandler.shutdown();                                        // clears the queue, no writer was started

// pushReplicaOfflineMsgToReplicationServers
localReplicationServer.getDSRSShutdownSync()
    .replicaOfflineMsgDispatched(baseDN, offlineMsg.getCSN(), recipientIds); // records it AFTER the give-up
for (ReplicationServerHandler rsHandler : recipients)
{
  if (rsHandler.shuttingDown()) { noLongerAwaitTheForwardOf(rsHandler); } // false for an aborted handshake
  else { addUpdate(rsHandler, offlineMsg, notAssuredUpdateMsg, assuredServers); }
}

The window is not exotic: ReplicationServer.shutdown() interrupts listenThread at :1189, before
awaitReplicaOfflineMsgsForwarded() at :1203. The handshake runs inline on that thread and blocks in
session.waitForStartup() before reader/writer.start(), so the interrupt becomes
abortStart → unregisterFailedHandshake. put() holds no domain lock, so it runs concurrently.

Result: the id is recorded, addUpdate goes into a queue that was just cleared, and nobody can ever report
that id — the handler is out of connectedRSs, so no stopServer will run for it. The wait burns the full
REPLICA_OFFLINE_GRACE_PERIOD on every shutdown that races an aborted RS handshake with at least one other
peer connected. Cost is the stall, not a lost message — that peer's session was already closed.

The PR body lists unregisterFailedHandshake() among the covered cases, so this is also a claim/code gap.

Suggested fix: call engageShutdown() on that path too, or re-check connectedRSs membership after
replicaOfflineMsgDispatched and strike off whoever left in between.

---

todo (blocking): three of the four give-up call sites have no test.

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

Neither test file mentions shuttingDown, stopServer, unregisterFailedHandshake or abortStart. Only the
ServerWriter filtered branch is exercised at server level. Deleting noLongerAwaitTheForwardOf from
stopServer (:1157) or from unregisterFailedHandshake (:1263), or deleting the shuttingDown() branch
(:429-432), is caught by nothing.

The DSRSShutdownSyncTest cases that look like coverage — aPeerWhichStoppedIsNoLongerWaitedFor,
aPeerWhichStoppedDoesNotEndTheWaitOfTheOthers, theWaitEndsWhenTheLastPeerExpectedToForwardStops — call
replicaOfflineMsgNotForwarded directly. They pin the sync object, never the wiring.

This is why the issue above went unnoticed. One server test that disconnects a peer during the grace period
and one that aborts a handshake would cover both.

---

issue (non-blocking): the two-peer test asserts wire receipt after shutdown() closed the sessions.

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

final long startTime = System.nanoTime();
windowOpener.start();
replicationServer.shutdown();          // stopAllServers -> MessageHandler.shutdown -> session.close()
...
assertThat(receivedWhenHeldBack.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
    .as("... the peer which was held back never learned that the replica went offline ...")
    .isNotNull();

The barrier is released by session.publish(), which enqueues into a 4000-slot sendQueue; Session.close()
discards it without draining — the #919 limitation this PR restates in "Left out". A scheduling window
between publish() returning and the publisher thread draining makes the future null, and the test then
fails with a message pointing at the bug being fixed rather than at the race.

Asserting on the forward being reported (barrier state) rather than on bytes reaching the peer would remove
the ambiguity.

---

issue (non-blocking): the PR body overstates the second server test's evidence.

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

The body says "Both new server tests were run with the guard they name reverted and fail on the message, not
on a duration". That holds for the first test. theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped
has exactly one assertion, and it is a duration whose .as() merely reads like a message:

assertThat(elapsed)
    .as("the shutdown kept waiting for a forward its own writer had already dropped")
    .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);

It also bypasses the production path it is named after — replicaOfflineMsgDispatched is called by hand and
the message is enqueued with rsHandler.add(...), never entering put(). And the mutation margin is thin:
reverting the ServerWriter branch gives elapsed ≈ 5000 − t_setup + t_teardown, killed only because the
post-wait teardown costs more than the two setup calls.

Worth correcting the body and adding a state assertion.

---

todo (non-blocking): the giveUpOn null-branch asymmetry is untested, and the mutation that undoes it
survives.

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

private boolean giveUpOn(int replicationServerId)
{
  final Set<Integer> awaited = awaitedForwarders;
  return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
}

The awaited != null conjunct is load-bearing and deliberate — a disconnect says nothing about a message the
peer was never given, which is exactly the opposite of forwardedBy's fallback. Every test calling
replicaOfflineMsgNotForwarded first calls replicaOfflineMsgDispatched for the same base DN, so this branch
is never entered. Dropping the conjunct turns any peer disconnect into an immediate release of a not-yet-
dispatched message — the master bug in a new shape — and the suite stays green.

A four-line unit test closes it: replicaOfflineMsgSent, then replicaOfflineMsgNotForwarded with no
dispatch, assert the message is still pending.

---

todo (non-blocking): the capability claimed over the superseded guard has no test.

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

The body claims a peer with a different generation id is no longer waited for, "which the caller side check
on getConnectedRSs().isEmpty() did not cover". No test drives a connected peer with a mismatched
generation id through put(). canShutdownWhenTheMessageWasQueuedForNoPeer passes an empty list straight to
the sync object, and shutdownIsNotDelayedWhenNoOtherReplicationServerCanForwardTheMessage is still released
by the unchanged ReplicationServer:1235 guard — so it would pass with the new filter removed.

---

suggestion (non-blocking): move the give-up in stopServer into a finally.

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

if (!sHandler.engageShutdown()) { ... }   // one-shot: no later stopServer re-enters
try { lock(); }                            // lock.lockInterruptibly()
catch (InterruptedException ex) { Thread.currentThread().interrupt(); return; }  // skips :1156-1157

The peer is then never struck off and nothing can retry. The residual is narrow — the early return also skips
unregisterServerHandler, so the queue is not cleared and a live writer can still publish and report; a real
stall needs the caller to be the dying writer/reader itself. No in-tree interrupter of those threads was found
inside the grace window, so this is a hardening item rather than a live defect.

---

nitpick (non-blocking): the new comment promises an ordering the code does not enforce.

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

// after the unregistration, which must not be skipped if this throws
unregisterServerHandler(sHandler, shutdown, false);
noLongerAwaitTheForwardOf(sHandler);

Two plain statements inside a try whose only handler is a method-scope catch (Exception) and whose
finally releases the lock only — a throw at the first line does skip the second. No realistic throw site
exists inside unregisterServerHandler, so either a try/finally or a reworded comment.

---

suggestion (non-blocking): retire what the new mechanism replaced.

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

if (!domain.getConnectedRSs().isEmpty()) { domainsToWaitFor.add(domain.getBaseDN()); }

The body says the recipient set supersedes this check, but it is still there and sampled once before the wait
— a second, weaker source of truth (a peer that connects after the message was queued owes nothing, yet still
makes the domain eligible). Two related leftovers:

• DSRSShutdownSync.java:56 still opens with "The ReplicaOfflineMsg which has not been forwarded yet, per
  domain and per replica"; an entry now legitimately holds a message that has been forwarded, by some of
  its recipients.
• 7 of the 9 new unit tests assert through canShutdown(DN), which has no production caller at HEAD. Not
  introduced here, but it is where most of the new test evidence rests.

[vharseko]

@maximthomas All eight points are addressed in dcda1e6569. The blocking one reproduced exactly as you described, and closing it turned out to be an ordering rule between three call sites rather than a fix at one of them. One suggestion I did not take, with the reason at the end rather than silently.

issue (blocking): unregisterFailedHandshake reports its give-up before the recipients exist

Confirmed, in every step. engageShutdown() has one caller - stopServer() at :1119 - and ServerHandler.shutdown() goes to MessageHandler.shutdown(), which clears the queue and deregisters the monitor and touches no flag, so shuttingDown() really is false for every aborted handshake. And abortStart() says why nothing retries, in its own comment: "the reader and writer threads that normally trigger the cleanup when the session dies were never started, so nothing else will ever remove it". The handler is out of connectedRSs, so no stopServer() can run for it either.

I took the first of your two suggestions, because the flag is what makes the check in put() mean anything at all:

    /*
     * This handler is stopping, and nothing else will ever say so for it: ...
     * The flag has to be set before the give-up below, because that is the
     * order put() reads the two in - it records the recipients of a ReplicaOfflineMsg and only
     * then asks each of them whether it is shutting down.
     */
    sHandler.engageShutdown();
    if (!isDataServer)
    {
      noLongerAwaitTheForwardOf(sHandler);
    }

The rule that closes the window is worth writing down, because it is what stopServer() was already obeying by accident: every teardown raises the flag before it gives up, and put() reads the flag after it has recorded the recipients. Of the two, at least one then always sees the other - a give-up which ran too early to find a recipient recorded has necessarily raised the flag put() goes on to read, and a give-up which runs later finds the recipient. The membership re-check you offered as the alternative closes it the same way (the removal precedes the give-up, the read follows the record) but on a second piece of state; the flag is the one the loop already reads. It is now stated at both ends.

The claim/code gap in the body is gone with it.

todo (blocking): three of the four give-up call sites have no test

Right - and the reason you gave is the reason: a unit test of the nested class cannot catch a bug in how the nested class is called. Four server tests now, all of them driving put() through a real broker, and all of them asserting on the barrier through a DSRSShutdownSync which records what the production code reports to it rather than on a duration alone.

• theShutdownStopsWaitingForAPeerWhoseHandshakeWasAborted aborts the handshake inside the dispatch - replicaOfflineMsgDispatched is overridden to run the abort before it records - which is the window itself rather than an imitation of it: put() has read the peers of the domain and nothing knows yet which of them the message is for. It asserts the aborted peer really was recorded among the recipients, so a future change which stops reproducing the window fails rather than passes quietly.
• theShutdownStopsWaitingForAPeerWhichDisconnected holds one of two peers back with a full send window so it cannot have forwarded anything, closes it once the message is queued for it, and pins stopServer().
• theMessageIsNotQueuedForAPeerWhoseGenerationIdDiffers pins the recipient side filter, which is the item below.
• theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped, reworked, pins the ServerWriter branch.

Watched failing, one mutation at a time:

mutation	fails with
engageShutdown() out of unregisterFailedHandshake()	the shutdown waited for a peer whose handshake had been aborted, and which nothing else will ever strike off
the give-up out of the finally of stopServer()	the shutdown kept waiting for a forward from a peer which had disconnected
the generation id filter out of the recipient list	the message was recorded as queued for a peer which does not share the generation id of the domain
the give-up branch out of ServerWriter	the writer dropped the message without telling the shutdown to stop waiting for the peer it was queued for
awaited != null out of giveUpOn()	aPeerStoppingBeforeTheMessageIsQueuedDoesNotEndTheWait
forwardedBy() releasing on the first forward	the wait ended on the first peer forwarding the message, without the writer of the peer which was held back ever reporting one

Each produced exactly the failures listed and no others.

issue (non-blocking): the two-peer test asserts wire receipt after shutdown() closed the sessions

The barrier assertion is in, and first:

      assertThat(shutdownSync.forwardedBy())
          .as("the wait ended on the first peer forwarding the message, without the writer of "
              + "the peer which was held back ever reporting one")
          .contains(REMOTE_RS_ID, HELD_BACK_RS_ID);

I kept the wire assertion rather than replacing it - it is the outcome #917 is about, and it is what caught the bug in the first place - but it is now last, and its message says that the forward above it was already asserted. A run where only that one fails is the Session.close() window of #919 and nothing else, which is exactly the ambiguity you were pointing at. Verified: with forwardedBy() releasing on the first forward, the test fails on the barrier assertion at line 363, not on the read.

issue (non-blocking): the PR body overstates the second server test's evidence

Both halves right, and both fixed rather than just the sentence.

The test goes through put() now. A peer whose send window is one is filled with a change, so the ReplicaOfflineMsg published by the broker is queued for it by put() and taken off the queue by its writer, which then blocks on the permit - and ServerHandler.take() acquires that permit before isUpdateMsgFiltered() is reached, so the generation id can be changed while the message is already in flight and the drop is deterministic rather than raced. The assertion is the give-up itself, waited for with a bound well short of the grace period, so a writer which never reports is named as such instead of appearing as a shutdown which waited.

The .as()-that-reads-like-a-message is gone with the duration it decorated; the body says what the test now measures.

issue (non-blocking): the giveUpOn null-branch asymmetry is untested

Your four lines, with the reason in the javadoc:

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

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

Dropping the conjunct fails it, and nothing else.

todo (non-blocking): the capability claimed over the superseded guard has no test

theMessageIsNotQueuedForAPeerWhoseGenerationIdDiffers drives a connected peer with a mismatched generation id through put() and asserts the recipients are empty - the claim itself, rather than a duration which the writer side give-up would satisfy anyway a moment later.

suggestion (non-blocking): move the give-up in stopServer into a finally

Done, and covering the interrupted lock acquisition as well:

      finally
      {
        /*
         * On every exit of this block, including the interrupted lock acquisition above and an
         * unregistration which threw: the flag engageShutdown() has just set is one-shot, so no
         * later stopServer() will run for this handler and nothing else would strike it off.
         * ...
         */
        if (sHandler.isReplicationServer())
        {
          noLongerAwaitTheForwardOf(sHandler);
        }
      }

I hesitated over the interrupt path, since giving up on a peer whose writer might still publish is the wrong direction. It is not that case: every caller of stopServer(h, false) closes the session before it gets here - ServerReader and ServerWriter both do it in their own finally - or is stopping the handler on purpose. So by the time this runs the peer cannot be told anything, whatever the method managed to do. What is still left behind on that path is the handler itself, registered with its queue uncleared; that leak predates this PR and is now named in "Left out".

nitpick (non-blocking): the new comment promises an ordering the code does not enforce

The two statements are no longer two statements - the give-up moved to the finally above, so the comment which claimed the ordering is gone and the structure states it.

suggestion (non-blocking): retire what the new mechanism replaced

Here I disagree with the conclusion, though not with the reading, and I corrected the body instead of removing the check.

getConnectedRSs().isEmpty() is not superseded, because it does not overlap with the recipient set on the case it exists for. An announcement which never went through this replication server at all - a replica which picked a remote RS, the #918 ordering - has awaitedForwarders == null, and remainingGracePeriod() reads only sentTime: nothing can clear such an entry except a forward or the grace period itself. With no peer connected, no forward will ever be reported, so dropping the check would make every such shutdown pay the full 5 s. The recipient set is sharper wherever a dispatch did happen, and the check is the only bound wherever one did not.

You are right that it is a second, weaker source of truth for the domains where both apply. That is now written where it is read, in the javadoc of awaitReplicaOfflineMsgsForwarded(), and the body no longer claims the recipient set supersedes it.

The two related leftovers:

• DSRSShutdownSync:56 corrected - an entry is owed a forward rather than not forwarded yet, and it says an entry legitimately holds a message some of its recipients have already sent.
• canShutdown(DN) having no production caller since The ReplicaOfflineMsg grace period is spent on the first message of the process: DSRSShutdownSync latches its timestamp and never resets it #900 is right. The four new server tests go through ReplicationServer.shutdown() rather than through it, so the new evidence does not rest there; retiring the method would touch the seven unit tests which do, and belongs on its own.

Verification

mvn -Pprecommit -pl opendj-server-legacy verify over the classes below, no failures:

• DSRSShutdownSyncTest 22, ReplicationServerShutdownSyncTest 13
• ReplicationServerTest 7, ReplicationDomainTest 12, MonitorTest 1, GenerationIdTest 4, TopologyViewTest 1, ReplicationServerFailoverTest 2, StateMachineTest 5, ReplicationServerDynamicConfTest 9, ChangelogBackendTestCase 30, FileChangelogDBTest 8, ChangeNumberControlPluginTestCase 4
• HandshakeAbortRegistrationTest 3 and HandshakeAbortGenerationIdTest 2 on their own - in a long run they collide on the administration connector port, which is the harness rather than this change.

[vharseko]

CI on dcda1e65 failed one job out of 23 - build-maven (ubuntu-latest, 17) - on one assertion out of 32252:

ReplicationServerShutdownSyncTest.thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns:219
[the peer replication server was never told that the replica went offline,
 its read ended with: java.io.IOException: no more data]
Expecting actual not to be null      (Time elapsed: 0.555 s)

The five other ubuntu JDKs, macOS, Windows and CodeQL were green, and the previous run of this branch (3a0180a4) was green throughout. c3af1ec fixes the test; below is what the failure was, and why it is not the barrier.

What happened

The replication log dumped for that test - RS(8226) is its shutdownSyncDeliveryDb server - has these two lines next to each other:

msgID=206 RS(8226) has accepted a connection from replication server RS(92) ... at 127.0.0.1:44412
msgID=216 RS(8226) session thread to RS(92) ... at 127.0.0.1:44412 was interrupted in the startup phase

ERR_SESSION_STARTUP_INTERRUPTED has a single origin - ServerHandler.finalizeStart() interrupted in session.waitForStartup() (ServerHandler:365-371) - and msgID 206 is logged one statement before that call. So:

1. startFromRemoteRS() registers the handler (ReplicationServerHandler:368) before finalizeStart() (:380), and waitForConnectedReplicationServer() polls getConnectedRSs() every 10 ms: it returned while the handshake was still inside finalizeStart().
2. That call was on session.waitForStartup(), the latch the Session publisher thread counts down on its first line (Session:530); reader.start() and writer.start() come after it (ServerHandler:373-374).
3. The handshake of an accepted connection runs in the listen thread (ReplicationServer.runListen()), and shutdown() interrupts that thread (ReplicationServer:1187-1190) before awaitReplicaOfflineMsgsForwarded() (:1203). The shutdown aborted its own unfinished handshake.
4. abortStart() closed the session - hence no more data on the peer side - and unregistered the handler.
5. 500 ms later the broker published the ReplicaOfflineMsg, put() found no peer replication server, the recipient list was empty and the entry cleared at once: elapsed 0.555 s, so both duration assertions held and only the delivery one failed.

Not a regression of this PR

The window is in code this PR does not touch, and it is not new: it is #821 seen from the shutdown side - "The leaking window is narrow - from register(this) to reader.start()/writer.start() - and today the only way in is an interrupt of the handshake thread inside session.waitForStartup()". #821 closed the leak that window used to leave behind - the unregisterFailedHandshake() in abortStart(), which is exactly what ran here - not the loss of the message.

On master the same race fails the same test, only differently: with the peer unregistered before getConnectedRSs() is sampled the shutdown returns immediately and elapsed >= FORWARD_DELAY fails; otherwise nothing ever reports a forward and the wait ends on the grace period, failing elapsed < GRACE_PERIOD. A peer whose session was closed during its handshake cannot receive the message under any granularity of the barrier - what this PR changes is only which assertion reports it first.

The fix

waitForConnectedReplicationServer() and waitForConnectedDirectoryServer() now wait for the handshake to be over rather than for the registration: finalizeStart() registers the handler as a monitor provider by its last statement, after the reader and the writer are started, so a handler in that registry has left the listen thread for good. The listen thread serves one handshake at a time, so a peer past that window puts every connection accepted before it - the collocated directory server of these tests - past it as well.

shutdownIsNotDelayedByTheWriterServingADirectoryServer waits for its directory server handler too. The writer that test is about does not exist until the handshake finishes, so without the wait the test can pass with nothing that could have held the shutdown back.

Verification

With a 300 ms sleep inserted before session.waitForStartup(), so the window is entered on purpose rather than hoped for:

tree	result
sleep, without c3af1ec	thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns:219 - the peer replication server was never told that the replica went offline ... no more data: the CI failure, and no other
sleep, with c3af1ec	Tests run: 13, Failures: 0
no sleep, with c3af1ec	Tests run: 13, Failures: 0

Left out

The production side of the same window: a peer already in connectedRSs - one put() would record as a recipient - can be killed by the shutdown's own interrupt before it can be told anything. Moving listenThread.interrupt() after awaitReplicaOfflineMsgsForwarded() would close it (close(listenSocket) is what stops accept(); the interrupt only matters for a handshake blocked in session.receive()), but that is a production change with its own trade-offs, and belongs in an issue of its own rather than in this PR - filed as #983.