[#983] Interrupt the listen thread after the ReplicaOfflineMsgs are forwarded

[vharseko]

Fixes #983

ReplicationServer.shutdown() interrupted its listen thread before it waited for the
ReplicaOfflineMsgs of its domains, and the handshake of an accepted connection runs in that
thread. A peer replication server whose handshake was in flight at that moment was therefore torn
down by the shutdown itself. The interrupt is latent - session.receive() reads a plain socket
and is not broken by it - so wherever in the handshake it landed, the flag survived until the
first interruptible operation, the latch of the session publisher in finalizeStart(). By then
startFromRemoteRS() had registered the peer in the domain - so it was one of the servers the
message had to be forwarded to - while its reader and its writer were not started yet, and
abortStart() closed its session and unregistered it. It was never told that the collocated
replica went offline, and its ChangeNumberIndexer keeps the medium consistency point pinned to
the last CSN of that replica until the replica comes back.

That latency is what makes the window the whole handshake rather than its tail: a peer still
owing its TopologyMsg when the interrupt is sent is torn down just the same, once its handshake
gets to the startup of its session.

The change

close(listenSocket) stays where it is, since that is what ends the accept() of the listen
thread and its loop already stops on a closed socket, so nothing new is accepted meanwhile. Only
the interrupt moves, to after awaitReplicaOfflineMsgsForwarded() and still before the domains
are stopped: a handshake which has not finished by then is aborted as before, on the startup of
its session and ahead of its reader and its writer.

Nothing waits for the listen thread in this path, so the shutdown does not become any longer:
shutdown() never joins it, and stopServer() takes the domain lock only when it is not
shutting down, so the domains do not block on the lock a handshake in flight holds either. The
same choice is already made a few lines above for the port switch, where stopListenThread()
closes the socket and joins without interrupting, deliberately letting the connection being
served finish.

The tests

thePeerWhoseHandshakeIsInFlightIsStillToldTheReplicaWentOffline pins the consequence rather
than the ordering: the peer whose handshake was in flight receives the ReplicaOfflineMsg.

FakePeerReplicationServer is split into its two handshake phases - handshaking() and
completeHandshake() - so that a peer can stop between them, which leaves the listen thread
blocked in the receive of the TopologyMsg it owes, the peer not yet registered on the domain: the
state in which the old code had already sent the interrupt which was going to abort the handshake
at the startup of its session. The peers of every other test are connected in one call as
before, now through FakePeerReplicationServer.connected(), which runs both phases and keeps the
send window #947 gave them. The gate is exact rather than timed: the DSRSShutdownSync of the
test counts a latch down when the shutdown enters its wait, so the handshake is completed at a
point where the interrupt has already been sent on the old code and has not been sent yet on the
new one. A message nobody can forward holds the shutdown in its wait meanwhile, so the forward of
the already connected peer cannot end the wait before the second peer has been served - the
recipient accounting of #947 is deliberately not relied upon here. Once the forward is in, the
test releases that announcement itself rather than sitting out the rest of its grace period, and
asserts that the forward landed inside the grace period: that is the budget the success road
runs in, and a runner which blows it is reported as such rather than as the regression.

Without the production change the test fails on the peer replication server 97 never connected:
waitForConnectedReplicationServer() - whose javadoc records this very window - never sees the
peer, because the shutdown aborted its handshake and abortStart() unregistered it. The server
log carries the msgID=216 ... was interrupted in the startup phase of that peer, the same line
as the CI failure #983 was found through.

thePeerWhoseHandshakeCompletesAfterTheShutdownIsStoppedNotServed pins the other half of the
order, that the interrupt still precedes the shutdown of the domains: a peer whose handshake
completes only once shutdown() has returned receives a StopMsg and is not left in the domain.
Without the interrupt such a peer registers after the domains were stopped and nothing ever
stops it - its session, reader, writer and heartbeat are started, and its writer parks forever
on a cursor over the changelog the shutdown has closed. With the interrupt block deleted the
test fails on the served road: heartbeats, and never a StopMsg.

Verification

Rebased onto master, which now carries #947 in the same file.

• ReplicationServerShutdownSyncTest - 15/15
• DSRSShutdownSyncTest, HandshakeAbortRegistrationTest, HandshakeAbortGenerationIdTest,
  ReplicationServerDynamicConfTest, ReplicationServerTest - 43/43 in one run
• the first test with the production hunk alone reverted - fails as above; the second with the
  interrupt block alone deleted - fails as above

What this does not close

A handshake which registers its peer after the message was queued still owes nothing: the
recipients are recorded when ReplicationServerDomain.put() dispatches, so such a peer is not
waited for and is served best-effort, through the ReplicaOfflineMsg ReplicaCursor synthesizes
from the offline CSN. That is the remaining half of the race, and it sits with the recipient
accounting #947 added rather than here.

"Forwarded" means handed to the send queue of the session, not written: a session thread busy
with an earlier buffer when the message is queued lets the shutdown close the session with the
message still queued. That is older than this change and is filed as #1055.

[vharseko]

Rebased onto master. #947 landed in the same two files meanwhile, so this took a real merge: ReplicationServer.java merged cleanly - what #947 added there is a javadoc on awaitReplicaOfflineMsgsForwarded() - while ReplicationServerShutdownSyncTest conflicted in six places, all of them where #947 rewrote FakePeerReplicationServer and the wait helpers.

What the resolution decided, since that is easier to read here than out of a force-push:

• FakePeerReplicationServer keeps both refactorings: the two handshake phases of this PR (handshaking() / completeHandshake()) on top of the send window, the generic receive(Class<T>) and the failure() of [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947. Its constructor is private now, so the peers of the existing tests are built through FakePeerReplicationServer.connected(...).
• The server ids of the new test moved to 97 and 98 - 95 and 96 are HELD_BACK_RS_ID and ABORTED_RS_ID in master - and its backend id to 8234, since 8227 is shutdownSyncDataServerForwardDb there.
• The waitForHandshakenReplicationServer() this PR added is dropped for waitForConnectedReplicationServer(domain, serverId) from [#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947: it waits for the same registration and additionally for the end of the handshake, without which the peer has no writer to forward the message to it anyway. That is what changed the failure message the description quotes.
• The javadoc of that helper said the shutdown interrupts the listen thread before it waits for the ReplicaOfflineMsgs - precisely what this PR changes - so it now describes the new order and points at the new test.

Re-run after the rebase: ReplicationServerShutdownSyncTest 14/14, and DSRSShutdownSyncTest, HandshakeAbortRegistrationTest, HandshakeAbortGenerationIdTest, ReplicationServerDynamicConfTest, ReplicationServerTest 43/43 in one run. With the production hunk alone reverted the new test still fails, now on the peer replication server 97 never connected, with the msgID=216 ... was interrupted in the startup phase of that peer in the server log.

[maximthomas]

praise: The move is the right fix, and the case pins the side it names.

• listenThread.interrupt() after awaitReplicaOfflineMsgsForwarded() keeps a handshake in flight alive for the forward, and still precedes domain.shutdown(), so a handshake finishing late is aborted at ServerHandler.waitForStartup() instead of being started on a shut-down domain — for a DS peer as much as for an RS peer (one flag, one consumer).
• Measured: ReplicationServerShutdownSyncTest 14/14 at head; with the hunk reverted exactly thePeerWhoseHandshakeIsInFlightIsStillToldTheReplicaWentOffline fails, deterministically (34.7 s = the 30 s connection timer + the 5 s grace), with the message the description quotes.
• FakePeerReplicationServer.handshaking() / completeHandshake() turn a paused handshake into a fixture, and failure() tells a message which never arrived from an exchange which failed.
• The description's "What this does not close" is accurate: a lone in-flight peer leaves connectedRSs empty and the wait short-circuits.

---

issue (blocking): The "interrupt still precedes the shutdown of the domains" half of the move is pinned by no case — deleting the interrupt block is green 14/14.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java:1213-1216, :1210-1211

The new case completes the handshake inside the wait, so when the interrupt is sent peer 97 already has a reader and a writer and the listen thread has left its loop; no assertion sees whether the interrupt went out. Measured: with if (listenThread != null) { listenThread.interrupt(); } deleted, the class is 14/14 green (the new case 7.09 s vs 7.10 s at head). What the interrupt protects: a peer whose TopologyMsg lands after the wait ends is not in connectedRSs when stopAllServers(true) runs, registers after the domain loop, and without the flag finalizeStart() completes — session, reader, writer, heartbeat started; the writer's cursor over the shut-down changelog is empty, not an exception (FileChangelogDB.getCursorFrom over the emptied domain map), so it parks forever. On the remove() road, with the JVM alive, that is five non-daemon threads and a peer which keeps heartbeating a deleted RS. The comment at :1210-1211 states the invariant; nothing holds it.

Pin — a peer whose handshake completes after shutdown() returned is stopped, not served:

@Test
public void thePeerWhoseHandshakeCompletesAfterTheShutdownIsStoppedNotServed() throws Exception
{
  // same fixture as the case above: replicationServer, connectedPeer, domain
  handshakingPeer = FakePeerReplicationServer.handshaking(
      replicationPort, HANDSHAKING_RS_ID, baseDN, EMPTY_DN_GENID);

  // Nothing pending: the wait returns at once and the interrupt lands on the handshake in flight.
  replicationServer.shutdown();

  // Promptly: the RS-side handshake receive is bounded by connectionTimeoutMS (5 s) from the
  // peer's first message. The TopologyMsg the RS owes was sent before its receive, so this
  // completes on both roads.
  handshakingPeer.completeHandshake();
  final Future<StopMsg> stopped = handshakingPeer.receive(StopMsg.class);
  assertThat(stopped.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
      .as("the peer whose handshake completed after the shutdown was served instead of "
          + "stopped, its read ended with: %s", handshakingPeer.failure())
      .isNotNull();
  // unregisterFailedHandshake() removes the peer before Session.close() sends the StopMsg
  assertThat(domain.getConnectedRSs()).doesNotContainKey(HANDSHAKING_RS_ID);
}

With the block deleted this goes red: the peer is registered, finalizeStart() completes, it receives heartbeats and never a StopMsg (stopped.get times out) and 97 stays in connectedRSs. Order matters: the StopMsg first, then connectedRSs — the registration is transient on the good road.

---

suggestion (non-blocking): The success road of the new case runs inside two unnamed 5 s budgets.

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

The wait is held only by the UNREACHABLE_DS_ID entry, i.e. exactly DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD: completeHandshake(), finalizeStart(), waitForConnectedReplicationServer() (10 ms steps), publish() and the forward must land before the wait ends, else domain.shutdown() closes 97's session and :303-305 reports the regression's own message. A second clock of the same size starts earlier: the RS-side handshake receive (MultimasterReplication.getConnectionTimeoutMS(), 5 s, ReplicationServer.java:305-306) runs from the peer's first message to completeHandshake() at :295. Both are ~50x the loopback cost, so a slow runner fails closed with a message that looks like the regression. The sibling at :231-232 names the first budget with an assertion; do the same here and name the second in the comment at :279-285:

final long startTime = System.currentTimeMillis();   // after shutdownIsWaiting.await(...)
...
final ReplicaOfflineMsg forwarded = received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
assertThat(elapsedMillis(startTime))
    .as("the handshake and the forward must land inside the grace period which holds the wait")
    .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);

Or: hold the wait with an entry the case releases itself, so the success road has no wall-clock budget.

---

nitpick (non-blocking): "Such a peer is already registered in the domain" describes the abort site, not the state the case builds.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java:1206-1209, ReplicationServerShutdownSyncTest.java:1379-1389, the description

ReplicationServerHandler.startFromRemoteRS() registers after the TopologyMsg receive (register() :368-373, wait :318-325); the case pauses the peer before its TopologyMsg — not registered — and publishes the ReplicaOfflineMsg only after the handshake completed (:295-301). So the window the comment and the description describe (registered, message queued, no writer yet) has no case of its own; the fix covers it by the same mechanism, as the description says ("pins the consequence rather than the ordering"). Also: an interrupt during the first receive or lockDomainWithTimeout() aborts at the tryLock (ReplicationServerDomain.java:2458-2460), not "on the startup of its session"; and the class javadoc of FakePeerReplicationServer ("completes the handshake, so that the handler it leaves behind on the domain has a real writer") is no longer true of handshaking().

 * is in flight - one of the very servers the message has to be forwarded to. Whether such a peer
 * is already registered in the domain or still owes its TopologyMsg, the interrupt reaches
 * ServerHandler.waitForStartup() or the domain lock before it, the abort unregisters it, and it
 * is never told that the replica went offline.

---

thought (non-blocking): "Forwarded" means handed to the session's send queue, not written — pre-existing, outside this PR.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:126-138, opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:145-198, :323

ServerWriter records replicaOfflineMsgForwarded() right after Session.publish(), which is sendQueue.offer(); Session.close() does not drain the queue. The tests' road is deterministic (the offer signals the idle session thread before the shutdown thread is notified, and send() is uninterruptible), but a session thread busy writing an earlier buffer — a heartbeat, a TopologyMsg — at offer time lets the shutdown reach close() with the ReplicaOfflineMsg still queued: the peer gets the StopMsg and never the message the shutdown was told was forwarded. Same at base for every connected() case; this hunk narrows the window if anything. A follow-up, if the guarantee should mean "written": record forwarded from the session thread after send(), or drain before closeInitiated.

[vharseko]

Round 2 pushed as 6812cfe, on top of the rebased head.

issue (blocking) - the interrupt itself was pinned by nothing. Added
thePeerWhoseHandshakeCompletesAfterTheShutdownIsStoppedNotServed, on the shape proposed, with
three departures:

• abortStart() closes the session - which is what sends the StopMsg - before it calls
  unregisterFailedHandshake() (ServerHandler.java:231 vs :256), not after, so the peer can
  read its StopMsg while the handler is still in connectedRSs. The registration check waits on
  the connection timer instead of asserting right away.
• stopped.get() on the served road ends in a TimeoutException, which would have swallowed
  the assertion message; it is caught and the assertion names what happened instead.
• The fixture is the bare one: the replication server and the handshaking peer. With nothing
  announced the wait returns at once whether or not another peer is connected, so the broker and
  the connected peer of the sibling add nothing here.

The TopologyMsg the replication server owes is sent after its receive, not before
(startFromRemoteRS() :333 then :343) - but before register() and finalizeStart() on
both roads, so the conclusion holds and completeHandshake() completes either way.

Measured: ReplicationServerShutdownSyncTest 15/15 at head; with the interrupt block deleted
exactly this case fails, on the served road - no StopMsg, heartbeats only - as described.

suggestion - the two 5 s budgets. The success road now asserts that the forward landed
inside REPLICA_OFFLINE_GRACE_PERIOD, measured from the announcement which holds the wait -
that is where the grace clock starts, a little before the wait is entered - and the comment
names the connection timeout as the second clock. The announcement is also released once the
forward is in (replicaOfflineMsgForwarded() on an entry which was never queued ends the wait,
as it did before the recipients were tracked), so the case no longer sits out the rest of the
grace period in its finally: 0.2 s instead of ~7 s. The budget itself does not go away with
that - the grace period expires on its own, and the constructor which could lengthen it is
package-private in another package - hence the assertion.

nitpick - "already registered". Reworded in the three places: the comment in
ReplicationServer.shutdown() now covers both a peer waiting on the startup of its session and
one still owing its TopologyMsg, for which the flag survives the receive and reaches the startup
of the session once the handshake has registered it; the class javadoc of
FakePeerReplicationServer no longer says every peer completes the handshake; and the
description says what state the case builds.

thought - "forwarded" means queued. Agreed, and pre-existing: filed as #1055, with the
blocked-send() window and the two directions.

[maximthomas]

praise: Round 2 pins the half of the move that round 1 could delete with green CI, and the pin is measured on both sides.

• thePeerWhoseHandshakeCompletesAfterTheShutdownIsStoppedNotServed (ReplicationServerShutdownSyncTest.java:358-414): 15/15 at head here; with the listenThread.interrupt() block at ReplicationServer.java:1216-1219 deleted exactly this case fails, at :398 after 30.11 s, with the message it names — the round-1 mutant survived 14/14, now killed 1/1.
• The three departures from the proposed shape are each right at head: the absence poll is reached only after the StopMsg assertion, the TimeoutException catch leaves stopMsg null (the callable returns null on any exception, so no ExecutionException road), and the wait is a no-op with nothing pending (ReplicationServer.java:1259-1268).
• The early release in the round-1 case (:336-338, the never-queued replicaOfflineMsgForwarded() road) takes it from 7.098 s to 0.142 s, and the elapsed < REPLICA_OFFLINE_GRACE_PERIOD clock starts before replicaOfflineMsgSent() stamps its own — conservative against both clocks.