[#900] Time the ReplicaOfflineMsg grace period per replica, and spend it where the message can still be forwarded

[vharseko]

Fixes #900

The bug

DSRSShutdownSync.stopInstanceTimestamp was a static AtomicLong written once per JVM with
compareAndSet(0, now). A domain publishes a ReplicaOfflineMsg on every disableService(),
not only at shutdown - an online import or a restore (MultimasterReplication.processImportBegin
/ processRestoreBegin), a backend being disabled
(LDAPReplicationDomain.performBackendPreFinalizationProcessing), a fractional or assured
configuration change - so the first of those latched the timestamp and canShutdown() returned
true at once for the rest of the process lifetime. At a later, real shutdown the grace period
OPENDJ-1453 added was already spent.

The message not being forwarded costs more than the notification: ChangeNumberIndexer
excludes offline replicas from the medium consistency point
(replicaOffline() / getOldestLastAliveCSN()), so a replication server which never learns the
replica went offline keeps that point pinned to its last CSN, and its change number index and
external changelog stop advancing for the domain.

The fix

• Per domain, per replica, per announcement. replicaOfflineMsgs now maps
  baseDN -> serverId -> the announcement still to be forwarded, and that entry carries the CSN
  of the message. Per domain because the message is sent on every disableService(); per replica
  because the collocated RS relays the message of every replica connected to it, so the forward of
  another replica's message says nothing about this one; per announcement because the message of
  an earlier disableService() may still be queued behind a backlog, and forwarding it must not
  spend the grace period of the one the shutdown is waiting for.
  PendingChanges.putReplicaOfflineMsg() hands back the CSN it generated, so the domain records
  what it actually announced. The fields are no longer static: the collocated DS and RS sides
  coordinate through the single instance MultimasterReplication hands to both.

• Wait where the forward is still possible. ServerHandler.shutdown() sets the consumer
  inactive, clears the message queue (MessageHandler.shutdown()) and closes the session
  before it joins the writer, so the loop on canShutdown() in ServerWriter could not
  forward anything by the time it ran - and a ReplicaOfflineMsg is never written to the
  changelog by the RS (publishUpdateMsg() only calls notifyReplicaOffline()), so the late
  queue cannot bring it back either. ReplicationServer.shutdown() now waits before it stops any
  of its domains, and the spin in ServerWriter is gone.

• Only a peer RS ends the wait. ReplicationServerDomain.put() never queues a
  ReplicaOfflineMsg for a directory server - its isUpdateMsgFiltered() drops it there - but a
  directory server which is catching up fills its late queue from the changelog, where
  ReplicaCursor synthesizes one from the offline CSN of the replica. Publishing that says
  nothing about the peer replication servers, so ServerWriter reports the forward only for a
  peer RS.

• Only when it can pay off, and once for the whole shutdown. The wait is skipped when no RS
  is connected: there would be nobody to forward the message to, and a standalone server would
  pay the whole grace period at every shutdown. All the domains of one shutdown wait together and
  share one deadline, so N base DNs cost one grace period rather than N, and the wait of one
  domain does not spend the window of the next. It runs before the assured timer of every domain
  is cancelled, so an assured update still waiting for acks keeps timing out during it. It is
  measured on System.nanoTime(), so a backwards clock step cannot stretch it, and
  abortInitialization() does not wait at all: nothing in an instance which never came up can
  forward a message.

Tests

• DSRSShutdownSyncTest (12) - the grace period is not spent by an earlier message, is counted
  per domain and per replica, neither the forward of another replica's message nor that of an
  older announcement ends the wait, one wait covers every domain of a shutdown, and the deadline
  bounds it.
• ReplicationServerShutdownSyncTest (8) - a peer connected through the real handshake has
  received a genuine ReplicaOfflineMsg by the time shutdown() returns; a message published to
  a directory server does not end the wait; the shutdown is not delayed when no peer RS is
  connected, when only a directory server is, or when no message is pending; and the whole
  shutdown stays within one grace period while two domains keep announcing themselves offline.

Every guard was checked by reverting it and observing that exactly the test which names it fails:
the writer guard (theForwardToADirectoryServerDoesNotEndTheWait, 256 ms instead of the grace
period), the CSN of the announcement (aStaleForwardDoesNotConsumeTheGracePeriodOfANewerMessage),
the shared deadline (theGracePeriodIsSharedByAllTheDomainsOfOneShutdown and
aSharedDeadline.../oneWaitCoversEveryDomainOfTheShutdown), and the wait running before the
handlers are stopped (thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns, which is the
only one of them a duration-only suite would not have caught). The two classes take about a
minute together, three of their tests spending the production grace period on purpose.

Regression: ReplicationServerTest, ReplicationServerDynamicConfTest, ReplicationDomainTest,
ReSyncTest, SchemaReplicationTest, ChangeNumberControlPluginTestCase, MonitorTest,
HandshakeAbortRegistrationTest, HandshakeAbortGenerationIdTest.

Left out

Limitations of the same mechanism, found while fixing this one and left to their own issues:

• The ReplicaOfflineMsg grace period ends at the first forward, so a second replication server can be left unaware of the offline replica #917 - the first forward releases the wait, so a second peer RS can be left unaware.
• A ReplicaOfflineMsg still queued behind an uncommitted change is recorded as sent, and the shutdown then waits its whole grace period for nothing #918 - a message still queued behind an uncommitted change is recorded as sent. The same issue
  covers the window in which publishReplicaOfflineMsg() records the announcement only after
  PendingChanges has already published it: a forward which wins that race finds nothing to
  clear, and the entry recorded afterwards holds the shutdown for its whole grace period.
• The wait proves the message was handed to the Session, not that it reached the wire:
  Session.publish() enqueues the buffer for the session's own publisher thread, and
  Session.close() interrupts that thread without draining the queue. Fixing it belongs in
  Session - drain before interrupting, or report the forward from the publisher thread. It is
  still strictly better than master, which loses the message one step earlier, in msgQueue.

[maximthomas]

praise: The diagnosis is exactly right and the fix goes to the right depth.

• The root cause is named precisely: compareAndSet(0, now) on a static field latches the window at the first disableService() — an import, a restore, a config change — not at the shutdown it exists for.
• Deleting the ServerWriter spin was necessary, not scope creep. ServerHandler.shutdown() sets the consumer inactive, clears msgQueue and closes the session before it joins the writer, and MessageHandler.getNextMessage() is while (activeConsumer) — so the old loop was a hot spin on an empty queue and a dead socket. The obvious per-baseDN-timestamp fix alone would have turned a silent no-op into a real 5 s busy spin at every shutdown.
• Instance fields instead of statics also remove state bleed between several ReplicationServers in one JVM.
• The wait/notifyAll is correct: the remove happens-before notifyAll, the waiter re-checks canShutdown under the same monitor, so no wakeup can be lost; and the deadline clamp bounds the wait at gracePeriod even if the domain keeps re-announcing itself offline.
• Waiting before assuredTimeoutTimer.cancel() is a deliberate, easily-missed detail, and getting it right matters.
• Deferring #917 and #918 explicitly, rather than half-fixing them, is the right call.

---

issue (blocking): the wait is granted for a peer RS but released by any writer, including a directory server's.

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java:1680 gates on !connectedRSs.isEmpty(). The release does not:

// ServerWriter.java:112-117 — runs for DataServerHandler and ReplicationServerHandler alike
if (updateMsg instanceof ReplicaOfflineMsg)
{
  dsrsShutdownSync.replicaOfflineMsgForwarded(
      replicationServerDomain.getBaseDN(), updateMsg.getCSN().getServerId());
}

put() fans a DS-sourced ReplicaOfflineMsg out to connectedDSs as well as connectedRSs, and isUpdateMsgFiltered() only drops it for a DS in BAD_GEN_ID_STATUS/FULL_UPDATE. Peer RS blocked in session.publish on a backlogged link + a local DS with an empty queue ⇒ the DS writer clears the key first, stopAllServers(true) then discards the peer RS's queued message. Zero peer RSs told — the #900 outcome, on a shutdown that paid for the grace period. put()'s RS-before-DS enqueue order buys microseconds; the writers are independent threads.

if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())

Same root cause as #917, different consequence (#917 = a second RS unaware; this = every RS unaware), and a one-liner — worth taking here.

Two comments currently record the opposite and should go with it:

• ReplicationServerShutdownSyncTest.java:153-157 — "a ReplicaOfflineMsg is never sent to a directory server anyway" is false.
• ReplicationServerDomain.java:1669 — "forwarded to the other RSs" promises all-of-N; the code delivers one-of-N.

---

issue (blocking): no test pins the ordering the PR exists for.

opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java:228-247 registers a peer RS by field assignment and never runs finalizeStart() — where the handshake creates the writer, the reader and the sendWindow. The registered peer has a null writer. No test constructs a ReplicaOfflineMsg; remoteEnd is only closed, never read.

Both waiting tests assert elapsed time only:

assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);

Move awaitReplicaOfflineMsgForwarded to the end of ReplicationServerDomain.shutdown(), after stopAllServers(true) — fully reintroducing OPENDJ-1453 — and every assertion stays green. The suite pins duration, never delivery, which is why the two issues above are invisible to CI.

Wanted: one test that registers a peer RS through the real handshake, publishes a genuine ReplicaOfflineMsg, and asserts it reaches the remote end before shutdown() returns. It would fail today (see next). Minimum: assert ordering — the peer handler still registered, its consumer still active, when the wait returns.

note: "Each of them was seen failing on the unfixed code first" doesn't hold as written — neither class compiles against the base (no DSRSShutdownSync(long), no REPLICA_OFFLINE_GRACE_PERIOD, one-arg signatures), and two of the five RS tests pass on base behaviour, since base shutdown() has no wait and these fixtures have no writer to spin. Name the adapted variants or drop the sentence.

---

issue (non-blocking): the wait is released by enqueue, not by transmission.

ServerWriter calls replicaOfflineMsgForwarded right after session.publish(). For a peer-RS handler that never touches the socket — ServerHandler.finalizeStart() runs session.start() + waitForStartup() before writer.start(), so isRunning is always true:

// Session.java:316-326  -> LinkedBlockingQueue<>(4000) at :99
if (isRunning.get()) {
  while (!closeInitiated) {
    if (sendQueue.offer(buffer, 100, MILLISECONDS)) return;   // queued, not sent
  }
}

// Session.java:151-166 — no drain, no flush
closeInitiated = true; interrupt(); join(); /* then close sockets */

So the wait returns, session.close() interrupts the publisher out of sendQueue.take(), and the buffer dies with the socket. It is never in the changelog, so nothing re-delivers it. With a backpressured peer up to 4000 buffers can sit ahead of it — the window is bounded by TCP, not by scheduling. That is precisely the case a grace period is for; a healthy peer never needed one.

Still strictly better than master, which loses the message one step earlier in msgQueue. But the PR text's "spend it where the message can still be forwarded" should be qualified: the wait proves handed to the session, not written to the wire. Fix properly by releasing on drain, or file it beside #917/#918.

---

suggestion (non-blocking): cap the grace period per shutdown, not per domain.

// ReplicationServer.shutdown()
for (ReplicationServerDomain domain : getReplicationServerDomains()) { domain.shutdown(); }

Sequential, and each awaitReplicaOfflineMsgForwarded takes its own deadline = now + gracePeriod. The old static timestamp meant later domains never waited; that accident is gone, so N base DNs now cost up to N × 5 s. Two paths pay it in full because nothing in the JVM can release them:

• the collocated DS picked a remote RS — the local RS sees the message only relayed, and put() skips the connectedRSs push for an RS-sourced message. Reachable because computeBestReplicationServer applies filterServersOnSameHost last, after the group-id / generation-id filters.
• ReplicationServerDomain.java:546 — a ChangelogException calls localReplicationServer.shutdown() on the ServerReader thread; put() already returned at :339 before any handler push, so that message can never be forwarded. Delay only, not a deadlock.

One shared deadline computed once in ReplicationServer.shutdown() restores the 5 s ceiling.

(The wait is the remainder of 5 s from the send, not a fresh 5 s — a slow DS shutdown absorbs it.)

[vharseko]

All four points are addressed in 1b0a3c7, and the description is rewritten to match.

---

issue (blocking): the wait is granted for a peer RS but released by any writer.

Taken - with a correction to the mechanism, because the fix has to be justified by the path that
actually exists.

ReplicationServerDomain.put() does not fan a ReplicaOfflineMsg out to connectedDSs. Its own
filter ends at ReplicationServerDomain.java:436-442 with:

    // Replica offline messages should not get to a connected DS, they are meant to be
    // exchanged only between RSes
    return updateMsg instanceof ReplicaOfflineMsg;

The method quoted in the review is the one in ServerWriter, which only filters on the status of
the consumer. So the interleaving as described - the local DS's writer clearing the key while the
peer RS is blocked in session.publish - is not reachable through put().

The conclusion holds through another door. A directory server which is catching up fills its late
queue from the changelog (MessageHandler.fillLateQueue() -> ReplicationServerDomain.getCursorFrom()
-> FileChangelogDB.getCursorFrom()), and ReplicaCursor.next() synthesizes a ReplicaOfflineMsg
from the offline CSN that notifyReplicaOffline() recorded for the replica. Its writer publishes
that one, and the entry the shutdown is watching is cleared although no peer has been told
anything. The one-liner is in, with that path named in the comment above it.

The two comments went with it, one of them differently than asked: ReplicationServerDomain now
says the message is forwarded to a connected RS and that the first forward ends the wait (#917).
The comment in the test was reworded rather than deleted - "a ReplicaOfflineMsg is never sent to a
directory server" is true of put(), which is what that test is about; what was missing is the
changelog path, and it is now stated where it matters.

---

issue (blocking): no test pins the ordering the PR exists for.

Taken.

thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns registers a peer RS through the real
handshake - ReplServerStartMsg and TopologyMsg exchanged with the RS under test, so
finalizeStart() runs and the handler has a real writer and a real sendWindow - has a broker
publish a genuine ReplicaOfflineMsg while the shutdown is already waiting, and asserts that the
remote end received it before shutdown() returned. Moving
awaitReplicaOfflineMsgsForwarded() after stopAllServers(true) makes exactly this test fail on
the message while the duration assertions stay green; that check was run.

theForwardToADirectoryServerDoesNotEndTheWait goes through the production path too: with a peer
RS registered, the shutdown must spend the whole grace period even though a directory server did
receive the message. Dropping the !handler.isDataServer() guard makes shutdown() return in
256 ms and the test fails.

The "seen failing on the unfixed code first" sentence is gone from the description. What replaces
it is per guard: each one was reverted in turn and exactly the test which names it fails.

---

issue (non-blocking): the wait is released by enqueue, not by transmission.

Correct, and now recorded as a limitation in the description instead of being implied away: the
wait proves the message was handed to the Session, not that it reached the wire. Not fixed here
because the fix belongs in Session - drain sendQueue before interrupting the publisher on
close, or report the forward from the publisher thread after send() - which is a change to every
message the session carries, not to this one. Happy to file it beside #917/#918 if you want it
tracked.

---

suggestion (non-blocking): cap the grace period per shutdown, not per domain.

Taken, and one step further than a shared deadline. With the domains waiting one after the other,
a single deadline starves the later ones: the first domain whose peer is wedged spends it, and
every domain after it finds the deadline passed and stops its handlers immediately - the #900
outcome again, now for every base DN but the first. So ReplicationServer.shutdown() waits once,
over all the domains that have a peer RS connected, before any of them is stopped;
ReplicationServerDomain.shutdown() no longer waits at all.
DSRSShutdownSyncTest.oneWaitCoversEveryDomainOfTheShutdown pins it - forwarding the message of
the first domain must not end the wait of the second.

That also removed the deadline parameter from the public shutdown() signature and the
"expired deadline" sentinel it forced on the abort path.

---

Two more things came out of the same pass, both in the commit:

• the pending entry is keyed by the CSN of the announcement. A replica announces itself offline on
  every disableService(), so forwarding a message that has been queued behind a backlog since an
  earlier import used to clear the entry of the announcement the shutdown was waiting for - the
  grace period was then skipped entirely. PendingChanges.putReplicaOfflineMsg() returns the CSN
  it generates so the domain records what it announced;
• the wait is measured on System.nanoTime() rather than the wall clock, so a clock step
  backwards during a shutdown cannot stretch it.