Send what a replication session's publisher left queued when it is closed

[vharseko]

Refs #963. Two things which came out of investigating that issue, and which belong together
because the second is what the first ruled out.

The tests: which end of a session can lose a queued message

Session.publish() has two branches, and which one a message takes decides whether a close can
lose it:

if (isRunning.get())                                  // the session's publisher thread is running
{ ... sendQueue.offer(buffer, 100, MILLISECONDS) ... } // an enqueue
else
{ send(buffer); }                                     // a synchronous write of the socket

isRunning is set inside run(), so it is true only where something called Session.start() -
and in the whole server that is one place, ServerHandler.java:362. ReplicationBroker never
starts its own. So the replication-server end of a session has a publisher thread and the
directory-server end does not.

SessionPublisherDrainTest pins that, because it is what decides where a close can lose anything:

• theSessionOfADirectoryServerBrokerHasNoPublisherThread - the session a real broker publishes
  its changes on is still Thread.State.NEW after the handshake;
• aSessionWithNoPublisherThreadHasReachedTheWireWhenPublishReturns - a change published and the
  session closed at once still reaches the peer, and the session's own thread never ran, so
  publish() is what wrote it. Without that second assertion the case would pass on either
  branch: a publisher thread usually outruns a close for a single message, which is measured
  rather than assumed - starting the publisher in that case is what makes it fail;
• aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed - the case for the change
  below. drain() stops at the StopMsg, so the size it asserts pins the order too.

What this says about #963. The candidate raised there - that the AddMsg of
ReSyncTest.testResyncAfterRestore was dropped by Session.close() without being drained -
cannot be what happened. That message is published by a directory server, on a session with no
publisher thread, so it was written to the socket before publish() returned. The conclusion of
that candidate, that the change never reached the replication server, stands on other evidence and
this PR does not touch it; only the mechanism is ruled out. Hence Refs, not Fixes: #963 stays
open.

The change: a close sends the queue instead of dropping it

Where a publisher thread does exist, close() set closeInitiated, interrupted that thread and
joined it, and everything still in sendQueue went with it. The StopMsg published afterwards
still went out - isRunning is false by then, so it takes the direct branch - so the peer read an
orderly close with no sign that anything was missing.

PR #919 recorded this as a known limitation and named the fix: "Fixing it belongs in Session -
drain before interrupting, or report the forward from the publisher thread."
ReplicationServerShutdownSyncTest carries a comment pointing at it as the reason one of its
assertions may fail. This takes the first of the two.

close() now sends that queue:

• after the join, because the publisher is gone by then and the closing thread owns the socket -
  which is what already lets the StopMsg be published on it;
• before the StopMsg, so that message stays last on the wire, which is what it means;
• not at all on a session which already failed, for the reason the StopMsg is skipped there:
  writing more to it cannot work. That is also the path where close() runs on the publisher
  thread itself - run() calls it when a send() threw - where the queue is unsendable by
  construction.

The budget. close() is called from shutdown paths, so an unbounded drain would hold the
thread shutting the server down for as long as a stalled consumer stays stalled, which is the
hazard #952 and #983 are about. DRAIN_BUDGET_MS is 5 s, the same value as
DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD: a close has no reason to wait longer for one of
these messages than the shutdown which is waiting on the close. A peer which is reading pays none
of it; a peer which is gone pays none either, its write failing at once. Only a peer which is alive
and not reading pays, and that is the case the old code answered by dropping the messages.

What is given up on is reported. A logger.warn names the session, how many messages it could
not hand over and why. The silence is what cost the most to diagnose in #963.
LocalizableMessage.raw is the idiom this package already uses for a log line, so this needs no
new ordinal in replication.properties.

What the first CI run found, and why the case needed a reader

The first run of this branch failed on three ubuntu legs - and on nothing else: 32560 tests, one
failure, this suite's own. The peer had received 2025, 2134 and 1306 of the 3000 messages. Every
macos and windows leg passed, and so did ubuntu 21 and 26.

The give-up warning this change adds appeared in none of those logs, which is what made it
diagnosable: the drain had not given up on anything, so the whole queue had been written to the
socket. The loss was under the write, in the teardown.

close() ends with StaticUtils.close(plainSocket, secureSocket) straight after the last write,
and the case had no reader on the closing side - so that side reached the close with inbound
bytes nobody had ever read. A close in that state ends the connection with a reset instead of a
FIN, and a reset discards whatever the peer has not read yet. Measured with a bare socket pair,
8 MiB written to a peer reading behind the writer, the only difference between runs being whether
the closing side drained its own inbound:

	no reader	reader
linux 6.12 / jdk 11	53.4% arrived, Connection reset	100% arrived, end of stream
macos 15.7 / jdk 26	98.3% arrived, Connection reset	100% arrived, end of stream

That is the whole of it: the platform spread explains why the suite passed on every macos leg and
on this machine while losing half the queue on ubuntu.

The server does not have that condition where the drain does anything.
ServerHandler.shutdown() closes the session at ServerHandler.java:946 and joins its
ServerReader only at :966, so the reader is still consuming inbound across the close. The paths
that close without a live reader are the ones that skip the drain anyway: Session.run() calls
close() after a send() threw, and ServerReader's finally (:228) is reached either on an
error - both leave sessionError set - or on a StopMsg, where the peer has announced it is
leaving and there is nothing unread inbound. The directory-server side has no publisher thread at
all, so its queue is empty and the drain is a no-op there.

So the case now keeps a reader on the sending end, as the server does, and its soTimeout is
lifted because receive() hands a read timeout to setSessionError() - which would skip the
drain and test nothing. The read of the peer end reports what ended it, so a short read names its
cause instead of leaving the next reader to find this out again.

Limits, stated rather than left to be found

• The budget is checked between messages, so one write which blocks past it still runs to
  completion. Bounding a single write needs a non-blocking socket, which this session is not.
• The give-up path therefore has no test: driving it needs a peer which never reads, and such a
  case would itself hang on that blocked write. The drain is covered; the give-up is not.
• A message published concurrently with a close is still dropped in silence: with isRunning
  true and closeInitiated true, the while (!closeInitiated) loop of publish() does not run and
  the call returns having done nothing. That is a different silent drop - at the door rather than in
  the queue - and it is not touched here.
• The drain hands the queue to the socket; it does not make the teardown orderly. As the table
  above shows, a close whose side has unread inbound resets the connection and the peer loses what
  it has not read - which would undo a drain. Every server path where the drain does something has
  a reader consuming inbound across the close, so this does not bite today, but it is a property of
  the callers rather than of close() itself. Making the close orderly regardless - reading the
  inbound to its end before closing the socket - is a change to every session teardown and belongs
  in its own PR, not bundled here.

Testing

mvn -o -pl opendj-server-legacy -Pprecommit verify -Dfailsafe.failIfNoSpecifiedTests=false \
    -Dit.test='SessionPublisherDrainTest,ReplSessionSecurityTest,ProtocolCompatibilityTest,
      ReplicationServerShutdownSyncTest,DSRSShutdownSyncTest,HandshakeAbortRegistrationTest,
      HandshakeAbortGenerationIdTest,ReplicationServerTest,ReplicationBrokerTest,
      ReplicationDomainTest,MonitorTest,ReplicationServerDynamicConfTest,ReSyncTest,
      GenerationIdTest,InitOnLineTest,ProtocolWindowTest,UpdateOperationTest,DependencyTest,
      SchemaReplicationTest'

190 tests, 0 failures, 0 errors, 0 skips, on the branch as it stands on master. The set is wider than the change because the drain adds
time to a close() whose peer is not reading, and close() is on every shutdown and handshake
path of this package - the shutdown and handshake classes are in there to catch a timing shift
rather than a logic one.

Every assertion was built with its defect put back:

variant	result
the drain removed	...SendsWhatIsStillQueuedWhenItIsClosed fails with the close dropped 2820 of the 3000 messages published - 180 reached the peer through the socket buffers
the publisher thread started in the no-publisher case	...HasReachedTheWireWhenPublishReturns fails on the thread state, which is what stops that case from passing on both branches

[vharseko]

@maximthomas this was in draft while I chased its own CI failure. It is back for review, and the
description is rewritten around what that failure turned out to be - worth reading before the diff,
because the interesting part is not the change but what it took to trust the case.

The first run failed on three ubuntu legs and on nothing else - 32560 tests, one failure, this
suite's own. The peer had received 2025, 2134 and 1306 of the 3000 messages. Every macos and windows
leg passed, and so did ubuntu 21 and 26.

The warning this change adds for a queue it cannot send appeared in none of those logs, which is
what made it diagnosable: the drain had not given up on anything, so it had written the whole queue
to the socket and the loss was under the write. The case had no reader on the closing side, so that
side reached close() with inbound bytes nobody had read - and a close in that state ends the
connection with a reset rather than a FIN, which discards whatever the peer has not read yet, the
drained queue included.

Measured with a bare socket pair, 8 MiB written to a peer reading behind the writer, the only
difference between runs being whether the closing side drained its own inbound:

	no reader	reader
linux 6.12 / jdk 11	53.4% arrived, Connection reset	100% arrived, end of stream
macos 15.7 / jdk 26	98.3% arrived, Connection reset	100% arrived, end of stream

That spread is the whole reason the case passed on every macos and windows leg, and on my machine,
while losing half the queue on ubuntu. I would have got this wrong from macOS alone, so the real test
was run on linux/jdk11 in a container, both ways:

• reader in - 3/3 green;
• reader commented out - fails with the peer received 2873 of the 3000 messages published; the read ended by java.net.SocketException: Connection reset.

The server does not have that condition where the drain does anything, which is why the change to
Session is untouched by all this - the second commit is test-only. ServerHandler.shutdown() closes
the session at :946 and joins its ServerReader only at :966, so a reader is consuming inbound
across the close; the paths that close without a live reader are the ones the drain skips anyway
(Session.run() after a send() threw, and ServerReader's finally on an error - both leave
sessionError set). The directory-server side has no publisher thread at all, so its queue is empty
and the drain is a no-op there. The case now keeps a reader for the same reason, with its soTimeout
lifted because receive() hands a read timeout to setSessionError(), which would skip the drain and
leave the case testing nothing.

Two things I would rather you heard from me than found:

• The drain hands the queue to the socket; it does not make the teardown orderly. As the table
  above shows, a close whose side has unread inbound resets and undoes a drain. It does not bite today
  because of what the callers do, not because of anything close() guarantees. Making the teardown
  orderly for every session is a change to every close path and I have kept it out of this PR - say if
  you would rather have it here.
• The give-up path has no test. Driving it needs a peer that never reads, and such a case would
  hang on the blocked write it is trying to observe. The drain is covered; the give-up is not.

On the part of this which is yours: the send-queue candidate you raised on #963 is ruled out as the
mechanism there - the AddMsg of ReSyncTest is published by a directory server, on a session with
no publisher thread, so publish() wrote it to the socket before returning, and the first two cases
here pin that. Your conclusion stands, and not on my say-so: the absence of
WARN_CHANGELOG_READ_AGAIN_FOR_MISSING_CHANGES on the head of #964 means the changelog never held the
change, so it did not reach RS(104). #963 stays open and this PR says Refs, not Fixes.