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. The error that decides this is read after the join rather
  than before it, so a publisher whose write failed while it was being joined is not answered
  with a stale null. What such a session gives up on is still reported rather than dropped
  without a word: what is left in the queue and what the publisher thread took off it and failed
  to write. After a failed write that thread goes on taking the queue until the close, every
  write after the first failing too, so the queue alone would undercount, and say nothing at all
  once the thread had emptied it. That close takes no publishLock, so a thread blocked in a write
  of the socket is released by the close of the sockets rather than waited for. The error is read
  once more after the drain, so a drain whose write failed is not followed by a StopMsg on the
  same socket;
• with publishLock held across the whole queue, because the publisher is not the only thread
  which writes this socket. A ServerWriter is joined only after this close - ServerHandler
  closes the session at :946 and joins the writer at :966, and ServerReader's finally
  closes it before handler.doStop() - and a HeartbeatThread is shut down after it too; their
  publish() takes the synchronous branch once the session is off the queueing one. Without the
  lock a newer message could be written between two older drained ones, and a peer replication
  server answers that by dropping the older ones at debug level
  (LogFile.appendWouldBreakKeyOrdering()). So it is the close, not run(), which takes the
  session off the queueing branch - under that lock, which it keeps across the queue and the
  StopMsg. Until then a concurrent publish() takes the queueing branch - it queues ahead of
  the drain, or returns at the door once it has seen the close - and from then on it waits for
  the lock and lands after the drain. publish() holds no lock between reading the close and
  queueing, though, so one descheduled there can queue its buffer after the drain's last poll;
  it checks the flag once it has queued, and a buffer it then still finds in the queue, under
  publishLock, is one nothing will send: it takes it back and reports it. A close which gives
  up empties the queue rather than counting it, so no buffer is reported twice. run() clears
  the flag itself only when its loop ended without a close, so that publish() does not go on
  queueing onto a queue nobody sends, and it does not set the flag at all on a session which was
  closed before it was started - nothing would clear it then, and every later publish() would
  return as if queued. The wait for the lock itself is not inside the budget, no more than it
  is for the StopMsg.

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. The
budget is per close, not per shutdown: ReplicationServerDomain.stopAllServers() stops its
handlers one after another on one thread, so a domain with several such peers pays it once for
each of them.

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. On the arm where the write failed the exception is named
rather than traced: that arm is reached whenever a peer which has announced it is leaving closes
before the drain gets to it, where a stack trace of an EPIPE says nothing. The level stays WARN
on both arms - a directory server re-reads these from the changelog when it reconnects, a peer
replication server does not, and nothing at that point tells the two apart.

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 Session.run(), which calls close() after a send() threw,
and ServerReader's finally (:228). The first - which in practice follows a close already in
progress, run()'s loop ending only on closeInitiated or an interrupt - and the error road of the
second leave sessionError set and skip the drain outright; the StopMsg road of the second does
drain - the reader is the thread closing, so
nothing is consuming inbound - but the peer has announced it is leaving and there is nothing unread
inbound to reset the connection over. 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 budget arm therefore has no test: driving it needs a peer which never reads, and such a
  case would itself hang on that blocked write. The arm which gives up on a failed write is
  covered - aCloseWhichCannotSendTheQueueReportsEveryMessageTheQueueHeld closes the sockets under
  a session holding a queue, so the first send() throws and the count the report names is exact.
  The road of a session which had already failed is covered too -
  aCloseOfAFailedSessionWritesNothingAndReportsTheQueue sets the error with the sockets intact,
  so a close which wrote to it regardless would be read by the peer;
  aCloseOfAFailedSessionWithNothingLeftToSendReportsNothing holds it to silence when there is
  nothing to report; and
  aCloseOfAStartedSessionWhoseWritesFailedReportsWhatThePublisherTookAsWell starts the
  publisher on a session whose sockets are closed under it, waits for it to empty the queue, and
  asserts one report naming all 7 messages - and that a publish() after the close fails
  rather than returns. aSessionClosedBeforeItIsStartedKeepsFailingPublishes closes a session,
  then starts it, and asserts the same of its publish().
• The ordering publishLock restores is pinned by no test, and neither is the flag being
  cleared by the close rather than by run(): making a ServerWriter's direct send land between
  two drained messages, or ahead of them, needs a race this fixture cannot hold open - with run()
  clearing the flag again the class stays green. What the cases pin is the drain and its give-up;
  the lock is argued from the code, not measured.
• Two re-reads of the error are pinned by no test either: the one after the join needs a publisher
  whose write fails while it is being joined, and the one before the StopMsg guards a write which
  would fail and be swallowed anyway, so nothing outside the session can see it - with it removed
  the class stays green.
• The budget is per close: a shutdown closes its sessions one after another, so it pays the
  budget once per peer which is alive and not reading. A deadline shared across a whole shutdown
  would be a follow-up.
• A message published concurrently with a close, while the close is still joining the
  publisher, 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 buffer which lands in the queue after the drain is taken back and reported,
  but that check is pinned by no test either: it needs a thread descheduled between two lines of
  publish() across a whole close, and with the check removed the class stays green.
• 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'

232 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 peer received 263 of the 3000 messages published; the read ended by a StopMsg - what the publisher had sent before the close, the rest dropped behind an orderly stop; ...ReportsEveryMessageTheQueueHeld fails too, with no report at all
a failed session's queue drained regardless	...WritesNothingAndReportsTheQueue fails on the peer reading the first queued DeleteMsg, and ...ReportsWhatThePublisherTookAsWell fails with no report at all
a failed session's queue dropped without a report	the same case fails on the report: none instead of one
what the publisher took and failed to write left out of the report	...ReportsWhatThePublisherTookAsWell fails with no report: the thread had emptied the queue
the flag not cleared by the close of a failed session	the same case fails on the publish() after the close returning as if queued
the report written even for nothing	...WithNothingLeftToSendReportsNothing fails on a report of 0 messages
run() setting the flag although the close came first	...ClosedBeforeItIsStartedKeepsFailingPublishes fails on the publish() returning as if queued
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
the return of the failed-write arm removed	green, and rightly so: that arm now empties the queue as it reports it, so the loop finds nothing more and ends after the one report of 7. Before this round it failed on seven reports counting 7, 6, ... 1
the 1 + of that arm's count removed	the same case fails on the count: the message the failed write took out of the queue is not accounted for
the queue drained before the close is submitted	...SendsWhatIsStillQueuedWhenItIsClosed fails on its new precondition, which is what stops it from passing while draining nothing
the StopMsg not sent	the same case fails on endedBy - expected "a StopMsg" but was "java.io.IOException: no more data" - while its count of 3000 stays green, which is the case passing on a stream nobody ended

[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.

[vharseko]

@maximthomas Round 2 is pushed as 971b9b666f, the branch rebased onto current master first. Every
point is taken; the question I answered by keeping the level and dropping the trace. The description
is updated, including one thing your review made me re-read - see the last section here.

The blocking one - the drain is not the only writer of that socket. Taken as described: the
queue now goes out with publishLock held across the loop, so the ServerWriter's and the
heartbeat's direct sends wait for the drain and land after it in order. What I checked before
taking it, because the fix is only worth the invariant it restores: ServerWriter.run() publishes
with no shutdown check, take() hands back the UpdateMsg it already holds as soon as
acquirePermitInSendWindow() sees shutdownWriter, and the writer is live across the close on
both roads - ServerHandler.shutdown() closes at :946 and joins at :966, and ServerReader's
finally closes the session before handler.doStop(). LogFile.appendWouldBreakKeyOrdering()
drops the later-arriving older record at debug, as you say, and ServerState never asks for it
again. The comment at :219 and the javadoc no longer claim a single writer: they name the threads
which can still write and say the lock is what orders them. One thing your sketch left implicit and
the javadoc now states - the wait for the lock is not inside the budget, no more than it is for the
StopMsg which follows.

localSessionError is a snapshot from before the join. Taken, re-read under stateLock after
the join. It also stops the StopMsg from being written to a socket the publisher has already
recorded as failed, which BASE did and swallowed.

The budget under the domain lock, and the javadoc which says two things the code does not do.
Taken, the paragraph is yours. stopServer(h, false) does hold lock() across
unregisterServerHandler() → shutdown() → session.close(), and a handshake waiting on it fails
lockDomainWithTimeout(); the javadoc now says the hold is under that lock on the non-shutdown
road. The 5 s is unchanged.

The question: WARN with a stack trace on the peer-closed road. WARN was intended, the stack
trace was not. The arm now names the exception - class and message - instead of tracing it. I kept
the level rather than splitting it: a directory server re-reads those updates from the changelog on
reconnect, a peer replication server given a ReplicaOfflineMsg forward does not, and at that point
nothing tells the two apart - INFO would bury the case nobody recovers from.

The IOException arm has no test. Taken, with a different fixture than you sketched, for a
reason worth stating: a peer closed from the outside gives up somewhere inside the TCP buffers, so
the count is whatever the kernel had absorbed and the assertion cannot pin a number. The new case
closes the sockets under the sending session instead - same failed write, and the first send()
of the drain throws, so the report has to name all 7. The queue is filled through the field for the
same reason: what a publisher thread leaves behind is a race, and this case is about the count.
Both mutants are red: the return removed gives seven reports counting 7, 6, ... 1 against the
hasSize(1), and the + 1 removed fails on the count.

The drain case pins a producer/publisher backlog, not a full socket buffer. Agreed, and your
measurement matches what the case does: ~440 KB of frames fit in a loopback pair's buffers, so the
publisher is not blocked in a write and what leaves the backlog is publish() outrunning it. The
comment says that now, queuedAtClose > 0 is asserted right before the close is submitted, and
drained.endedBy is asserted rather than only carried into the failure text. The mutant for the
precondition - the queue drained before the close is submitted - is red on it.

The nitpick. Rewritten, and the file is in the diff now: the comment says the close drains
within DRAIN_BUDGET_MS and under publishLock, and that a red there means the budget ran out
(the warning being the thing to look for) rather than the granularity of the barrier.

Two things I would rather you heard from me.

• The ordering the lock restores is pinned by no test. Making a ServerWriter's direct send
  land between two drained messages needs a race this fixture cannot hold open. The drain and its
  give-up are pinned; the lock is argued from the code, not measured.
• Your StopMsg road made me re-read my own description. I had written that the paths which
  close without a live reader "skip the drain anyway". That is true of Session.run() and of
  ServerReader's error road, but not of the StopMsg road you describe: sessionError is null
  there, so the drain runs - it simply has nothing unread inbound to reset the connection over. The
  description is corrected rather than left for the next reader to trip on.

Runs. The wide set of the description on the rebased branch: 224 tests, 0 failures, 0 errors,
0 skips. Three of the four mutants above are red as stated. The fourth - the StopMsg not sent, so
endedBy names an exception - is not yet run: one attempt died in setUp on an embedded server
which could not bind its administration port, the next was stopped for want of memory on the box.
It is the one assertion here whose defect has not been put back, and I will report it rather than
leave the table implying otherwise.

[vharseko]

The mutant I reported as missing is in: with the StopMsg not sent,
aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed fails on
the StopMsg is what ended the stream ... expected "a StopMsg" but was "java.io.IOException: no more data" - while the count of 3000 stays green, which is exactly the case passing on a stream
nobody ended, the reason that assertion was added. The table in the description now carries it, and
nothing else in the round changed.

[maximthomas]

praise: Every round-1 item was taken, and the give-up arm now has an exact pin.

• sendWhatThePublisherLeftQueued() holds publishLock across the whole queue (Session.java:295-327, unlocked in finally), so a direct send that reaches the lock waits for the drain.
• aCloseWhichCannotSendTheQueueReportsEveryMessageTheQueueHeld pins the failed-write arm and its + 1, and the description's mutant table says which case goes red for each one.
• The javadoc now states the real costs: a single blocking write, and the domain lock on the stopServer(h, false) road.

---

issue (non-blocking): publishLock is taken only after the publisher has exited, so the direct-send interleave is narrowed, not closed.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:295, :689, :448-467

run() clears isRunning at :689 before the thread ends. Between that point and close() reaching publishLock.lock() (the join wake-up, the stateLock re-read, the trace block, isEmpty()), a ServerWriter or HeartbeatThread publish() takes the synchronous branch and gets the lock without contention. On the ServerReader-finally road (the session closes before handler.doStop()), a newer UpdateMsg can reach the wire ahead of older queued ones. The window is microseconds wide, and BASE lost the whole queue, so this is not a regression. But "Holding the lock makes them wait for the drain and land after it" says more than the code does. Taking the lock before join() does not fix it: a publisher that has already taken a buffer blocks in send() on that lock, and the join never returns. The one fix that works in place is for the closer, not run(), to clear isRunning under the lock after the join. That turns this interleave into the door drop your Limits section already describes. Your call whether to do that here. At minimum, the javadoc should name the window.

---

issue (non-blocking): When the publisher fails during the join, the post-join re-read of sessionError skips the drain, and the queue is dropped without a warning. At 3dc6939 this road was reported.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:192-195, :232

The publisher's send() fails while close() is joining it. run()'s own close() returns because closeInitiated is set, the closer re-reads the error, and it skips sendWhatThePublisherLeftQueued(). That method is the only caller of reportQueueNotSent(), so the backlog disappears with nothing above TRACE. Nothing could have been delivered on that socket; the only loss is the report, which the javadoc promises ("What is given up on is reported rather than dropped in silence"). The re-read was my round-1 suggestion, and I missed this side of it.

    if (localSessionError == null)
    {
      sendWhatThePublisherLeftQueued();
    }
    else if (!sendQueue.isEmpty())
    {
      reportQueueNotSent(sendQueue.size(), "the session had already failed: " + localSessionError);
    }

---

issue (non-blocking): When the drain's own write fails, the StopMsg is still written to that socket.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:237-247, :186-190

The StopMsg guard tests the localSessionError read before the drain. send() records the drain's failure only in sessionError, so the guard never sees it. The extra write fails and is swallowed, so this is harmless. But the comment at :186-190 names the StopMsg as something that "must not be written to a socket which has already failed".

    synchronized (stateLock)
    {
      localSessionError = sessionError;   // the drain may have failed
    }
    if (localSessionError == null
        && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)

---

suggestion (non-blocking): DRAIN_BUDGET_MS applies to each close, but a shutdown closes sessions one after another.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:53-62

ReplicationServerDomain.stopAllServers(true) (:1070-1081) stops every RS handler, then every DS handler, one at a time on one thread. Each session.close() can spend 5 s plus one blocking write. With N slow-reading peers that each hold a backlog, a domain's shutdown takes N x 5 s, which is longer than the single grace period the javadoc sizes the constant against. One sentence in the javadoc and in Limits would cover it. A deadline shared across a whole shutdown belongs in a follow-up.

---

suggestion (non-blocking): Nothing pins the localSessionError == null guard around the drain or the post-join re-read. Removing either one leaves all four cases green.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:192-195, :232; opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:302

No case reaches close() with sessionError set and a non-empty queue. The drain case lifts soTimeout so that sessionError stays null. The give-up case's sessionError is null until the drain's own send() fails. The no-publisher cases close with an empty queue. So "drain unconditionally" and "delete :192-195" both survive by construction.

Pin: use the give-up case's shape, but set the error instead of closing the sockets. With the drain made unconditional, the receiver reads a DeleteMsg and the case goes red. If the else above is taken, also assert its one report.

        // instead of closeTheSocketsUnder(sender):
        final Field error = Session.class.getDeclaredField("sessionError");
        error.setAccessible(true);
        error.set(sender, new java.io.IOException("injected"));
        sender.close();
        ReplicationMsg read = null;
        try
        {
          read = receiver.receive();
        }
        catch (final java.io.IOException expected)
        {
          // the socket closed with nothing written
        }
        assertThat(read).as("a failed session's queue must not be written").isNull();

The re-read needs a publisher whose send() fails while close() joins it. Whether that is worth a case is up to you.

---

nitpick (non-blocking): The new diagnostic comment says the warning "says the budget ran out" and that "its absence says the message left this end".

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

The failed-write arm's warning says "the write failed with ...", not that the budget ran out. And there are roads that lose the message at this end with no warning: the failed-session skip above, and a publish() concurrent with the close, which your Limits section describes. Suggest: "the warning says why the queue was given up; its absence does not prove the message left this end".

---

nitpick (non-blocking): The MESSAGES_PUBLISHED javadoc still says 3000 "has to outrun what the socket buffers of a loopback pair can swallow".

opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:77-81

The round-2 comment inside the case says the opposite: ~440 KB is swallowed, and the backlog comes from publish() outrunning the publisher. The commit message says this was corrected.

---

nitpick (non-blocking): The class javadoc still calls the drain case "the last test here" and the others "the other two".

opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:59-62

There are four cases now, and the last one is the give-up case at :302. Its contract is not mentioned.

---

nitpick (non-blocking): The description's mutant table quotes a failure message the head cannot print.

PR description, row "the drain removed"; opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:264

"the close dropped 2820 of the 3000 messages published - 180 reached the peer through the socket buffers" is the text of an earlier assertion. The head prints "the peer received %d of the %d messages published; the read ended by %s" (a run at 3dc6939 gave 176 of 3000). "through the socket buffers" also contradicts the mechanism the case comment now states.

---

nitpick (non-blocking): The no-publisher case says "a close sends a FIN", but that end closes with no inbound reader, which is the shape the drain case says ends in a reset.

opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:136-176, :207-209

The case passes because loopback queues the one frame before any RST arrives. Its assertions pin the synchronous branch either way. Not run: this needs a probe on Linux and on Darwin (a second receive() after the DeleteMsg). Saying "closes" rather than "sends a FIN" would be exact.

[vharseko]

@maximthomas Round 3 is pushed as 7e55e66686, the branch rebased onto current master first. All nine points
are taken; the first one with an addition to the fix you proposed, which is below. The description
is updated to match.

The window between isRunning.set(false) and publishLock.lock(). Taken the way you put it:
the close, not run(), now clears the flag, after the join and under publishLock, and keeps the
lock across the drain and the StopMsg. A publish() concurrent with the close now either
returns at the door or waits for the drain and lands after it. The addition: run() still clears
the flag itself when its loop ended without a close. If it never did, then after an interrupt
from anywhere but close() the flag would stay true with nothing sending, and a publish() on
a full queue would retry offer(100 ms) forever, because nothing would ever set closeInitiated.
Nothing but close() interrupts that thread today, but I would rather not leave that trap in. The
javadoc no longer says more than the code does. The ordering is still not pinned: with run()
clearing the flag again, the class stays green, and the description says so.

The queue of a session which failed during the join. Taken, with your else. That close now
also takes no publishLock. Before this round a failed session never took it, and taking it there
would make a close wait on a thread blocked in a write of that socket, when closing the sockets is
what releases that thread.

The StopMsg after a failed drain. Taken: the error is read once more after the drain.

The budget per close. Taken: one sentence on the constant, on the method and in the
description. The shared deadline is left for a follow-up, as you suggest.

The unpinned guard. Taken, with your fixture: aCloseOfAFailedSessionWritesNothingAndReportsTheQueue
sets sessionError with the sockets intact, then asserts that the peer reads nothing and that
there is exactly one report naming all 7 messages. Draining regardless is red on the peer reading
uid=failed0. Dropping the else report is red on "reported: []". The re-read after the join
stays unpinned, for the reason you give. So does the new re-read before the StopMsg: the write it
prevents would fail and be swallowed, so nothing outside the session can observe it. Both are
listed in Limits.

The nitpicks. All five taken:

• the ReplicationServerShutdownSyncTest comment uses your wording;
• the MESSAGES_PUBLISHED javadoc describes the backlog the case really relies on;
• the class javadoc names all four cases;
• the no-publisher case says "closes" instead of "sends a FIN";
• the mutant table has the drain-removed row re-measured at this head: the peer received 285 of the 3000 messages published; the read ended by a StopMsg.

One more thing, which your review led me to: the description said run()'s own close() is the
road where the queue cannot be sent. In practice that call follows a close which is already in
progress. The loop ends only on closeInitiated or on an interrupt, so that close() usually
returns at once. The code comment which said this is gone, and the description is corrected.

Runs. SessionPublisherDrainTest 5/5. The description's wide set: 229 tests, 0 failures,
0 errors, 0 skips. The javadoc gate passes. Mutants: failed session drained regardless — red;
its report dropped — red; drain removed — red in two cases; no re-read before the StopMsg, and
run() clearing the flag — green, as stated above.

[maximthomas]

praise: Every round-2 point was taken, and the failed-session road now has a pin.

• close() clears isRunning itself, after the join and under publishLock, and holds the lock across the drain and the StopMsg (Session.java:252-255), so a publish() that sees the flag cleared waits for the drain.
• The failed-session arm reports the queue without taking publishLock (Session.java:220-236), so a thread blocked writing that socket is released when the socket closes instead of being waited on.
• aCloseOfAFailedSessionWritesNothingAndReportsTheQueue covers that arm: the peer reads nothing, and exactly one report names all 7 messages.

---

issue (non-blocking): If a session is closed before start(), isRunning stays true for good, so every later publish() returns silently.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:689, :724-727, :476-478

On an unstarted session, close() takes the normal arm, clears the flag under publishLock and returns. A start() after that runs run(): it sets the flag (:689), skips the loop because closeInitiated is set, and the new if (!closeInitiated) guard skips the clear. publish() then takes the queueing branch, never enters while (!closeInitiated), and returns. This can happen on a group-id change: ReplicationServer.applyConfigurationChange calls stopAllServers(true) (:1851-1856) without interrupting the listen thread and without taking the domain lock. A DS handshake between DataServerHandler register(this) (:483) and finalizeStart() → session.start() (ServerHandler.java:362) then has its session closed while heartbeatThread is still null (ServerHandler.java:946-950). After that it starts the HeartbeatThread (:382-384). At BASE, that thread's first publish() threw on the closed socket and the catch outside its loop (HeartbeatThread.java:137) ended it. At this head it keeps publishing into nothing every interval/3 until the RS stops. The window is narrow, and I found it by reading, not by a run. This road mirrors the case you kept run()'s clear for.

  @Override
  public void run()
  {
    synchronized (stateLock)
    {
      // A close which came first has already cleared the flag: publish() stays on the
      // synchronous branch, which fails on the closed socket.
      if (!closeInitiated)
      {
        isRunning.set(true);
      }
    }
    latch.countDown();

---

issue (non-blocking): The failed-session report counts only what is left in sendQueue. It misses the buffer the publisher's failed write took, and on a started session it misses the backlog run() keeps consuming after that failure.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:231-234, :696-716

When a send() fails, run()'s catch sets needClosing and goes back to take() + send() until closeInitiated (unchanged since BASE). After a failed flush, every later send() fails at once. The comment at :224-226 describes the publisher's write failing during the join. On that road the failed buffer has already left the queue, so the report says N for N+1 lost, and says nothing when N=0. The drain arm counts its own failed write with + 1 (:335, :352). If a started RS-side session fails before any close, the ServerWriter keeps publishing and the publisher keeps taking and fast-failing until the ServerReader closes the session. The report then shows only what was left at that moment. I checked this by reading; I did not measure how much a real reset consumes.

  /** Whether the publisher took a buffer off the queue and failed to write it. */
  private volatile boolean publisherLostABuffer;

      // run()
      catch (IOException e)
      {
        setSessionError(e);
        publisherLostABuffer = true;
        needClosing = true;
        break; // nothing after a failed write can go out
      }

      // close(), failed arm
      final int notSent = sendQueue.size() + (publisherLostABuffer ? 1 : 0);
      if (notSent > 0)
      {
        reportQueueNotSent(notSent, "the session had already failed: " + localSessionError);
      }

Or, if you would rather not touch run()'s loop in this PR, have the report text and the :224-226 comment say that the count is a lower bound.

---

issue (non-blocking): A concurrent publish() does not have to either return at the door or wait for the drain. Its message can also land in the queue after the drain's last poll().

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:476-483, :307-310, :245-250

publish() holds no lock between reading closeInitiated (:478) and offer() (:483). Suppose a ServerWriter or HeartbeatThread is descheduled between those two lines while the close runs the join, the clear, the drain and the StopMsg. Its buffer is then enqueued after the drain's last poll() (:331). Nothing sends or reports it, and publish() returns normally. The window is one thread's preemption across the close, and BASE dropped the whole queue anyway, so this is not a regression. But the javadoc, the close comment and the Limits bullet ("at the door rather than in the queue") describe a two-way split the code does not guarantee.

   * The close takes the session off the queueing branch under the same lock it holds here, so
   * such a {@code publish()} either returns at the door, having seen the close, or waits for the
   * drain and lands after it - unless it read the close as not yet begun and is descheduled before
   * its offer() enqueues: its buffer then lands in the queue after the drain's last poll, where
   * nothing sends or reports it, and the call returns as if it had queued it.

A code fix, if you want one, is your call: publish() could re-check after its offer(), or the close could report whatever the queue still holds once the sockets are closed. The second catches most of these late buffers, not all.

---

suggestion (non-blocking): No case pins the failed arm's own isRunning.set(false) (:230) or its if (!sendQueue.isEmpty()) guard.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:230-234, opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java:377

run() no longer clears the flag after a close (:724), so on a started session :230 is the only place the close clears it. aCloseOfAFailedSessionWritesNothingAndReportsTheQueue never starts the sender, so the flag is already false. It also always queues MESSAGES_LEFT_UNSENT messages, so the empty-queue branch is never taken. Two mutants survive, by reading: deleting :230 makes publish() on a closed, failed, started session return at the door instead of throwing, and the ServerWriter (:144/:153) and the HeartbeatThread (:137) end only on that IOException. Making the guard unconditional writes a "0 message(s)" report. run()'s own guard is already listed in Limits.

        final Field isRunning = Session.class.getDeclaredField("isRunning");
        isRunning.setAccessible(true);
        ((java.util.concurrent.atomic.AtomicBoolean) isRunning.get(sender)).set(true);
        // ... sessionError set and sender.close() as today, then:
        try
        {
          sender.publish(new DeleteMsg(DN.valueOf("uid=late," + TEST_ROOT_DN_STRING),
              csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
          org.assertj.core.api.Assertions.fail("a publish() on a closed session returned as if queued");
        }
        catch (final java.io.IOException expected)
        {
          // The close put the session on the synchronous branch, which fails on the closed socket.
        }

Pin: the publish() above goes red without :230. A second case with an empty queue that asserts reported is empty goes red on the unconditional report.

[vharseko]

@maximthomas Round 5 is pushed as b35374dc61. I rebased the branch onto current master first. All four points are taken; for two of them I chose a different fix than the one you proposed, and the reasons are below. The description is updated to match.

A session closed before start(). Taken with your fix, and it is my regression: the if (!closeInitiated) I added to run() in round 3 is exactly what left the flag set. run() now sets the flag under stateLock, and only if no close came first. aSessionClosedBeforeItIsStartedKeepsFailingPublishes closes a session, starts it, and asserts that publish() throws. It goes red when run() sets the flag unconditionally.

The failed-session report as a lower bound. Taken, but not with break. run() counts every buffer it took and failed to write in publisherFailedWrites, and the failed arm reports that count plus what is left in the queue. The join before it makes the count complete. I left the loop alone: with break, a failed write would end the thread, and run() would then close the session itself, from its own thread. That changes when a failed session is torn down, and it does more than this report needs. The count is exact either way. aCloseOfAStartedSessionWhoseWritesFailedReportsWhatThePublisherTookAsWell starts the publisher on a session whose sockets are closed under it, publishes 7 messages, waits until the thread has emptied the queue, and asserts exactly one report naming all 7. Without the counter, it is red with no report at all.

A buffer queued after the drain. Taken with a code fix, and the javadoc now describes three outcomes instead of two. Once its offer() succeeds, publish() checks isRunning. If the flag is cleared, it takes publishLock and calls sendQueue.remove(buffer); a buffer still found there is one nothing will send, and it is reported. A buffer queued before the flag was cleared is left to the drain. Such a buffer cannot see the flag cleared, and the drain polls only after clearing it. A close which gives up now empties the queue (takeWhatIsLeftQueued()) instead of counting it, so a buffer cannot be reported both by the close and by publish(). remove() compares the arrays by identity. This check is not pinned: removing it leaves the class green. Pinning it needs a thread descheduled between two lines of publish() across a whole close, and Limits says so. The drop at the door, while the close is still joining the publisher, is unchanged and still listed there.

The unpinned isRunning.set(false) and empty-queue guard. Taken. I put the publish() you proposed into the started-session case above rather than using reflection, so the flag is actually set by run(); without the clear, it is red on the publish() after the close. aCloseOfAFailedSessionWithNothingLeftToSendReportsNothing covers the guard, which is now notSent > 0. With the guard unconditional, it is red on a report of 0 messages.

One consequence for the mutant table: with the queue emptied by the give-up, removing the return of the failed-write arm no longer produces seven reports. The loop finds the queue empty and ends after one report. That return now guards against nothing. I kept it because it says what the arm means, and the table row now records it as green, which is correct. Removing the 1 + of that count is still red.

Runs. SessionPublisherDrainTest 8/8. The description's wide set: 232 tests, 0 failures, 0 errors, 0 skips. The javadoc gate passes. Mutants: run() setting the flag unconditionally, the counter left out, the flag not cleared by the failed arm, and the report written for nothing are all red; so are the failed session drained regardless and the drain removed (now 263 of the 3000). The take-back after offer() is green, as stated above.