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.

[maximthomas]

praise: The drain sits exactly where a close can lose something, and the description says what it does not fix.

• Session.close():217-219 drains between the publisher join() and the StopMsg, so the stop stays last on the wire, and the localSessionError == null guard reuses the StopMsg's own skip.
• theSessionOfADirectoryServerBrokerHasNoPublisherThread pins the one session.start() in main (ServerHandler.java:362), which is what rules the #963 mechanism out for a broker.
• The drain-removed mutant is red here too: aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed fails with "the peer received 176 of the 3000 messages published" (local failsafe run at 3dc6939, drain call deleted from close(), reader in place).

---

issue (blocking): The drain is not the only writer of the socket after the join: a ServerWriter's direct send can land between two drained messages, and an RS peer then discards the older ones.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:263-278, :219, :243-244

sendWhatThePublisherLeftQueued() calls send() per buffer and send() takes publishLock per message (:441); nothing is held across the loop. ServerHandler.shutdown() (ServerHandler.java:940-966) sets shutdownWriter, closes the session at :946, and joins the ServerWriter only at :966; take() (:989-994) returns the UpdateMsg it already holds as soon as acquirePermitInSendWindow()'s 500 ms loop (:1072) sees shutdownWriter, and ServerWriter.run() (ServerWriter.java:99-126) calls session.publish() with no shutdown check — isRunning is false after the join, so that is a direct send(). HeartbeatThread.shutdown() (:950) also runs after the close. So the comment at :219 ("this thread is the only one left writing this socket") and the javadoc at :243-244 ("the socket is this thread's alone") state an invariant the code does not have.

The road is the one the drain exists for: a peer whose send window is exhausted has the publisher blocked in output.write() with a backlog queued and the writer waiting for a permit with a newer UpdateMsg in hand. When the peer resumes reading at close, join() returns, the drain writes the backlog at the peer's pace (up to 5 s), and within 500 ms the writer's newer CSN is written between two older drained ones. On an RS peer LogFile.appendWouldBreakKeyOrdering() (LogFile.java:281, key <= newest) then drops every later-arriving older record at debug level, the peer's ServerState is already past them so no reconnect re-serves them, and reportQueueNotSent() never fires. Not a regression — BASE dropped the whole queue and let the same direct send move the peer's state past the loss — but the premise and the guarantee this PR states do not hold there.

Holding publishLock across the loop restores the invariant the comment claims: the writer's and the heartbeat's direct sends wait for the drain and land after it, in order; the IOException arm and the budget return release it in the finally.

    final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS);
    byte[] buffer;
    publishLock.lock();
    try
    {
      while ((buffer = sendQueue.poll()) != null)
      {
        // ... unchanged: deadline check, send(buffer), IOException arm ...
      }
    }
    finally
    {
      publishLock.unlock();
    }

Then :219 and :243 can say that publishLock is held for the whole drain, so the ServerWriter and heartbeat sends which can still reach publish() wait for it — instead of claiming a single writer.

---

issue (non-blocking): localSessionError is a snapshot from before the join, so a publisher whose write fails during the join leaves the drain running on a failed socket.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:170-174, :217

close() copies sessionError under stateLock before interrupt()/join(). A publisher inside send() which gets the reset records it (run(), :633), its own close() returns on closeInitiated, join() returns, and :217 tests the stale null: the drain's first write fails at once on the errored socket and reportQueueNotSent() names the drain's own EPIPE rather than the reset the publisher recorded; the StopMsg then goes to the same socket (the BASE shape). One misleading WARN, no extra hang — this road needs a failed write, not a blocked one. The comment at :207-210 says this road is skipped; re-reading the error after the join makes it so.

    try { interrupt(); join(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    synchronized (stateLock)
    {
      localSessionError = sessionError;
    }

---

suggestion (non-blocking): The between-messages budget runs under the domain lock in stopServer(handler, false), and its javadoc says two things the code does not do.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:250-255, :269, opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java:1135, :1214

stopServer(h, false) takes lock() at :1135 and holds it across unregisterServerHandler() → sHandler.shutdown() (:1214) → ServerHandler.shutdown():946 session.close() → the drain (callers: :925/:1000 after a failed ack, :1060 stopReplicationServers, ServerHandler.doStop():1254, ServerWriter's finally :165, :1644, :1679-1680, :2814). A slowly reading peer with a backlog at close now holds the domain lock for up to 5 s plus one write, where BASE paid the publisher's in-flight write plus the StopMsg; a handshake on that domain meanwhile fails lockDomainWithTimeout() (ServerHandler.java:804-810, tryLock(3000 + rand*1000)) with WARN_TIMEOUT_WHEN_CROSS_CONNECTION and the broker retries. The 5 s is the PR's choice and I would not change it here; it is worth one sentence in the javadoc that the hold is under the domain lock on the non-shutdown road.

The javadoc: :250 "The budget is what bounds a close of a session whose peer has stopped reading" is contradicted by :253-255 of the same paragraph — a peer which has stopped reading is exactly the one whose single write blocks past the budget (until the peer reads or TCP gives up, ~15 min at Linux defaults for a black-holed host; a state which at BASE wrote only the StopMsg and returned, though BASE hung the same way whenever the publisher itself was mid-write). And :251-252 "a peer which is gone pays none either - the write fails at once" holds only for a peer which answers RST. Say what holds:

   * The budget bounds how many messages a close spends on a peer which is reading slowly. A
   * peer which is reading pays none of it; a peer which answers with a reset pays one failed
   * write. It is checked between messages, so a single write which blocks past the budget still
   * runs to completion - for a peer which has stopped reading, or vanished without a reset, that
   * is for as long as TCP keeps the connection alive: bounding it needs a non-blocking socket,
   * which this session is not.

---

question (non-blocking): Was WARN with a stack trace intended on the peer-closed road too, or only on the budget road?

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:279-284, :286-296

A DS which stops while the RS's publisher for it has a backlog: its StopMsg reaches ServerReader, the finally (ServerReader.java:228) calls session.close() with sessionError null, the drain's first writes are absorbed until the DS's close answers with a reset, and the IOException arm reports "N message(s) ... the peer was not told about them: " at WARN. For a DS peer those updates are re-read from the changelog on reconnect (its ServerState never covered them), so the line reads as a loss the protocol recovers from, once per DS per RS on a rolling restart under load; a real loss (a ReplicaOfflineMsg forward to a peer RS) reads the same. Not run — the road is by read. If only the budget road was meant to alarm: on the IOException arm log the exception's class and message, and say a peer which reconnects re-reads them — or INFO there and WARN kept for the budget arm, the one nobody recovers from.

---

suggestion (non-blocking): The IOException arm of the drain and the count it reports are pinned by no test, though that arm is drivable.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:279-284, opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:174-245

The description says the give-up path has no test because it needs a peer which never reads — that is the budget arm. The IOException arm needs a peer which has closed, which the fixture can do. None of the three cases closes the receiver before a sender with a non-empty queue, and none reads the log: deleting the return at :283 (one failed write and one WARN per remaining buffer on every close of a dead peer) or the + 1 at :282 survives green, by construction.

Pin: a fourth case — publish N messages on a sender whose peer reads nothing, close the receiver's socket, then close the sender; capture the error log (ReplicationTestCase's helper at :948) and assert exactly one line carrying "was closed with" whose count equals the messages the peer never received (N minus the frames read off the peer's socket before it closed; count the writer's records once — it holds every line twice).

---

suggestion (non-blocking): The drain case pins the drain through a producer/publisher backlog, not the full socket buffer its comment claims, and asserts neither that precondition nor the order it says it pins.

opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:218-245, :171-173, :74-78

The mutant is red (176 of 3000 received), so the pin holds — but not for the reason at :220-221. Measured on this box, a TCP_NODELAY loopback pair with nobody reading absorbs ~545 KiB; 3000 TLS-wrapped DeleteMsg frames (~150 B each, ~440 KB) sit in the kernel and the publisher never blocks in its write. What leaves ~2800 messages in sendQueue at close is publish() (encode + offer) outrunning the publisher thread (TLS write + flush per frame) — a race the case neither asserts nor bounds, so a faster publisher or a slower producer shrinks it toward zero with no red to say the case stopped pinning anything. drained.endedBy is carried into the assertion text ("the read ended by a StopMsg") and asserted nowhere, so 3000 deletes followed by an exception in place of the StopMsg still passes.

      // right before submitting close(): the case pins nothing unless the publisher is behind
      final int queuedAtClose = ((java.util.Queue<?>) sendQueueField.get(sender)).size();
      assertThat(queuedAtClose).as("messages still queued when close() ran").isGreaterThan(0);
      // ...
      assertThat(drained.endedBy).as("the StopMsg stays last on the wire").isEqualTo("a StopMsg");

Or make the precondition by construction: a peer which reads nothing until close() is submitted, so every message is in the queue. Either way, rewrite :220-221 to say the producer outruns the publisher rather than that the socket buffers fill.

---

nitpick (non-blocking): The comment the description names as the pointer to this limitation still describes the old close().

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

"close() discards whatever is still in its send queue without draining it, which is the limitation issue #919 recorded. If this is the only assertion which fails, that window is the explanation" — the file is not in the diff. Say the close now drains within DRAIN_BUDGET_MS, and that a red there means the budget ran out (the WARN this PR adds is the thing to look for) or the interleave of the blocking issue above.