[#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued

[vharseko]

Fixes #1055

ServerWriter reported a ReplicaOfflineMsg forwarded as soon as Session.publish() had handed
it to the send queue of the session. A session thread busy writing an earlier buffer when the
message was queued - a heartbeat, a TopologyMsg, an update to a peer whose TCP window is full -
let the shutdown, released by that report, end its wait for a peer which had not been told and
reach close() with the message still queued. Before #1035 close() dropped that queue: the
peer received the StopMsg and never the message the shutdown had been told was forwarded, and
its ChangeNumberIndexer kept the medium consistency point pinned to the last CSN of the
replica until the replica came back - the cost of #917 and #983, on a peer the wait believed it
had served. Since #1035 close() sends what is still queued, but only once the write it joins
has returned and within a budget of its own (DRAIN_BUDGET_MS), not within the grace period the
shutdown is meant to spend on that message, and the shutdown still takes the peer for told
before it is. Recorded as a limitation by #919 and #947, whose wait proved the message was
handed to the session, not written.

The change

• Session.publish(msg, whenWritten) runs the callback once, on the thread which wrote the
  message, after the write returned - the thread of the session, or the close for a message it
  sends out of the queue - and never for a message which was not written: one whose write
  failed, or one the close gives up on (its budget ran out, its write failed, or the session had
  already failed). It returns false for a message it neither wrote nor queued: one the protocol
  version of the peer cannot carry, one published while the session is being closed, or one
  queued after the close drained the queue, which the close takes back. publish(msg) delegates
  to it, so nothing else changes for the messages which have no callback.
• ServerWriter reports the forward from that callback, so "forwarded" now means written to the
  socket - the bytes are with the kernel, which delivers them whatever the process does next,
  since no replication socket sets SO_LINGER - which is what the grace period is spent on. A
  message the session refuses is reported as a give-up, as the message the writer's own filter
  drops already was ([#917] Wait for every peer replication server to forward the ReplicaOfflineMsg #947): nothing will ever forward it, and the shutdown must not wait for it.
  Since [#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it #1019 the filter drops a message the protocol version of the peer cannot carry before the
  session is given it, so the refusal left to this road is the one of a session being closed.
• Session gets a hook, beforeQueueing(Runnable), which publish() runs between its check
  that no close has begun and the offer of the message to the queue. Only the tests set it: a
  publish() descheduled at that spot is the only one which can queue a message after a close
  has drained the queue, and nothing else holds a thread there on cue. Unset, it costs one
  volatile read.

The drain of Session.close() is #1035's; this PR only carries the callback through it, so a
message the close writes on the way out is reported as forwarded rather than written in silence.
The shutdown does not reach close() before the message is written unless the grace period runs
out - and a session thread held inside a write is something close() already waited for, in its
join().

The tests

SessionTest (3, new) pins the contract of the callback on a session whose peer reads only when
the test lets it: the callback does not run while the message is queued behind a write the peer
has not consumed, and runs once it has; a message with no encoding for the peer is refused and
its callback never runs; a session with no thread of its own writes on the publishing thread and
runs the callback before publish() returns.

SessionPublisherDrainTest (#1035's, 10, 2 new):

• aCloseRunsTheCallbackOfEachMessageItSendsOutOfTheQueue queues seven messages with callbacks
  on a session which was never started, closes it, and asserts that each callback ran once, in
  the order of the writes, on the thread which closed the session, and that the peer received
  exactly those messages.
• aMessageQueuedAfterTheCloseDrainedTheQueueIsTakenBackAndReported holds a publish() on a
  started session at the hook above, runs close() to its end, then lets the offer go. It
  asserts that publish() returns false, that the callback never runs, that the queue is
  empty, that the take-back is reported once, and that the peer reads the StopMsg with nothing
  before it.
• aCloseWhichCannotSendTheQueueReportsEveryMessageTheQueueHeld gives its messages callbacks and
  asserts that none runs for a queue whose write failed.
• The cases which fill the queue through its field build the Session$Outgoing element
  publish() queues, rather than a bare buffer.

ReplicationServerShutdownSyncTest (21, 2 new):

• thePeerStillReadingAnEarlierChangeIsToldTheReplicaWentOfflineBeforeItIsStopped holds the
  session thread serving a peer inside the write of a 4 MiB change - the peer reads nothing and
  the socket buffers on both sides of its connection are bounded to 8 KiB - queues the
  ReplicaOfflineMsg behind it, starts the shutdown, and lets the peer read only once the
  shutdown has either closed the session or spent a second not doing so. It asserts that the
  peer receives the message, that no forward was reported before the peer read, that one was
  reported by the time the shutdown returned, and that the shutdown returned inside the grace
  period. Before Send what a replication session's publisher left queued when it is closed #1035 the peer received the StopMsg alone, with the forward reported by the
  writer and the shutdown over in 39 ms; since Send what a replication session's publisher left queued when it is closed #1035 the drain of the close delivers the message,
  and what fails is the forward reported while it was still queued.
• theShutdownStopsWaitingForAPeerWhoseProtocolCannotCarryTheMessage connects a peer speaking
  protocol version 7, for which the message has no encoding, and asserts that the peer is struck
  off and the shutdown ends inside the grace period. It was written against the refusal of the
  session; since [#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it #1019 the writer's filter drops that message first, so the case now runs the
  road of [#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it #1019 - which onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline pins as
  well - and it passes with the forward reported once publish() returns (the first mutant
  below). It stays for the one assertion [#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it #1019's case does not make: the duration of the
  shutdown.

The peer of the first test is served over a connection the test established itself, as
ReplicationServer.runListen() serves an accepted one, because holding a thread inside a write
needs the send buffer of the replication server's own socket bounded: the buffers a kernel picks
on its own on a loopback link absorb hundreds of kilobytes (and macOS keeps a few hundred on the
receiving side whatever it is asked for), so only an explicit SO_SNDBUF on the accepted socket
makes the block deterministic.

Rebased onto master at 0793e27bbc, commits 6ae52a56b6 (the change) and 7c7f055890 (the
take-back case and its hook, from the review of dacc944545). #1035 landed during the review
and made Session.close() send what the publisher left queued; the two met in Session:

• the queue holds Outgoing (the buffer and its callback) and the drain of the close writes
  outgoing.buffer and runs the callback after each write which returned;
• takeBackWhatWasQueuedTooLate() takes the Outgoing back and says whether it did, and
  publish() returns false for a message taken back;
• the comments which said that close() drops the queue - the javadoc of
  forwardReplicaOfflineMsg() and two in ReplicationServerShutdownSyncTest - say what it does
  now, and Send what a replication session's publisher left queued when it is closed #1035's SessionPublisherDrainTest fills the queue with Outgoing (it failed with a
  ClassCastException otherwise).

On this head: SessionPublisherDrainTest 10/10, SessionTest 3/3, Skipped: 0. On
dacc944545, not repeated here (the commits master took since touch none of these files):
ReplicationServerShutdownSyncTest 21/21, DSRSShutdownSyncTest 29/29,
ReplicaOfflineMsgCatchUpTest 2/2, SynchronizationMsgTest 78/78.

Every assertion was watched failing against the code it pins. The take-back row was measured
on this head; the first, third and last on dacc944545; the second on an earlier head of this
PR and not again (Session.run() is unchanged by the rebases):

mutation	fails with
the forward reported after publish() returns, as on master	thePeerStillReadingAnEarlierChange... alone - the writer reported the message forwarded while it was still queued behind a change the peer had not read
the session thread never runs the callback	SessionTest.theCallbackRunsOnceTheMessageIsWrittenAndNotWhenItIsQueued - the callback did not run once the message had been written; thePeerStillReadingAnEarlierChange... - the message was written to the peer and nothing reported the forward; and the four existing cases which need a forward to be reported at all
the drain of the close never runs the callback	aCloseRunsTheCallbackOfEachMessageItSendsOutOfTheQueue - the close sent the queue without running the callback of each message it wrote
publish() answers true for a message queued after the drain (return true; in place of the take-back)	aMessageQueuedAfterTheCloseDrainedTheQueueIsTakenBackAndReported alone - a message queued after the close drained the queue was answered as queued, though nothing is left to send it
the writer ignores a message the session refused	nothing - survives on this head (see below)

Regression, on f287cbdb1c: mvn -Pprecommit -pl opendj-server-legacy verify over SessionTest, ReplicationServerShutdownSyncTest, DSRSShutdownSyncTest, ReplicationServerDynamicConfTest, ReplicationServerTest, ReplicationDomainTest, MonitorTest, GenerationIdTest, TopologyViewTest, ReplicationServerFailoverTest, StateMachineTest, ChangelogBackendTestCase, FileChangelogDBTest, ChangeNumberControlPluginTestCase, ReSyncTest, SchemaReplicationTest, AssuredReplicationServerTest, AssuredReplicationPluginTest - 512 tests, no failures - and HandshakeAbortRegistrationTest, HandshakeAbortGenerationIdTest on their own (they collide on the administration connector port in a long run) - 5, no failures. Not repeated on this head; CI is the gate for it.

Left out

• The !accepted arm of forwardReplicaOfflineMsg() is not pinned any more. It was pinned by
  the peer speaking version 7, whose message [#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it #1019 now drops before the session is given it;
  the refusal left to it is a message published while the session is being closed. The
  session's side of that refusal - the take-back, and the false it answers - is pinned by
  aMessageQueuedAfterTheCloseDrainedTheQueueIsTakenBackAndReported. No case drives it through
  the writer, and the mutant which drops the writer's arm survives.
• A message whose protocol version the peer cannot carry is no longer this PR's: [#1014] Report a ReplicaOfflineMsg as forwarded only to a peer which can decode it #1019 drops it
  in the writer's filter, logs the drop and gives the peer up, on master.
• Session.close() still blocks in its join() for as long as the peer takes to read the buffer
  the session thread is writing, and then in the writes of its drain and of the StopMsg, with no
  bound on a single write but TCP's own: a peer which stops reading altogether holds the shutdown
  of the replication server until the kernel gives the connection up. Older than this change,
  recorded by Send what a replication session's publisher left queued when it is closed #1035 as well.

[vharseko]

@maximthomas rebased onto master at 129fc4e346, now that #987 has landed - the branch was conflicting - and the one [#1055] commit goes over it (c577881435). Nothing new is proposed here; this only says what the conflict was and how it was taken.

The conflict was in ReplicationServerShutdownSyncTest, in three places. The constants and the cases were both sides appending at the same spot - HANDSHAKING_RS_ID 97 and UNREACHABLE_DS_ID 98 are #987's, BUSY_RS_ID 99 and LEGACY_RS_ID 100 this PR's, and the two cases of each sit next to each other, master's first. The third was FakePeerReplicationServer, and it was not textual: #987 split the handshake into the two phases its own cases need - handshaking() stops before the TopologyMsg, completeHandshake() sends it - and turned the constructors into the factories connected(...), where this PR had moved the handshake into a handshake() of its own and added the two ways in it needs. Both ride on #987's shape now: connected(port, id, baseDN, generationId, windowSize, protocolVersion) for the peer speaking version 7, connected(session, socket, id, baseDN, generationId, windowSize) for the peer over the connection the test established itself. Each runs the first phase in a private constructor over a shared start() - the protocol version set on the session before the start message, as before - and the second through completed(), which closes the peer when that phase fails, the way #987's factory does; abandon() closes what a failed first phase leaves behind. Session, ServerWriter and SessionTest merged on their own; #987 is in ReplicationServer.shutdown(), away from the callback.

Re-run after the rebase: ReplicationServerShutdownSyncTest 17/17 - the two cases of #987 included - SessionTest 3/3, DSRSShutdownSyncTest 29/29. The description says so, and counts the class at 17.

Not measured again: the three mutations of the table. The sources under test are those the table was measured on.

[vharseko]

@maximthomas rebased onto master at 26defc28ef - the branch was conflicting again - and the one [#1055] commit goes over it (f287cbdb1c). Nothing new is proposed here beyond what the merge itself called for; this says what the conflict was, how it was taken, and one consequence of #1019 for this PR.

ServerWriter. #1019 put isEncodableFor() in front of isUpdateMsgFiltered(), and #1034 reworded the comment on the forward this PR moves into the callback. This PR's shape is kept: the ReplicaOfflineMsg of a peer RS goes through forwardReplicaOfflineMsg() and the forward is reported from the callback of Session.publish(msg, whenWritten), with #1019's filter in front of it. The javadoc of forwardReplicaOfflineMsg() was brought in line with master: it takes #1029's account of the directory server (never handed the message, rather than handed one synthesized by the cursor) and says that a message the peer cannot carry is dropped by the filter before the session is given it.

ReplicationServerShutdownSyncTest. #1019 added its own connected(..., protocolVersion) to FakePeerReplicationServer; the factories are this PR's, over completed() and the shared start(), with #1019's javadoc on the version overload. Two ids collided: #1019's OLD_PEER_RS_ID is 99, and the replication server ids 8235 and 8236 are taken on master, so this PR's BUSY_RS_ID is 100, LEGACY_RS_ID 101, and its two replication servers 8238 and 8239. The javadoc and one assertion message of the version 7 case say "drop" rather than "refusal", for the reason below.

What #1019 changes for this PR. The peer speaking version 7 no longer reaches the session's refusal: the writer's filter drops its message first, logs it and gives the peer up. Measured again on this head:

mutation	on f287cbdb1c
the forward reported once publish() returns, as before this PR	red on thePeerStillReadingAnEarlierChange... alone - forward reported by [100], the shutdown took 177 ms; the version 7 case is green under it now
the writer ignores a message the session refused (!accepted arm dropped)	survives

The refusal left to the !accepted arm is a message published while the session is being closed, which no case here reproduces. I have kept the arm and recorded it as unpinned in the description rather than build a fixture for that race in a rebase; if you would rather have it pinned in this PR, say so and it goes in the next round. The version 7 case stays for the one thing #1019's case does not assert, the duration of the shutdown.

On this head: ReplicationServerShutdownSyncTest 21/21, SessionTest 3/3, DSRSShutdownSyncTest 29/29, ReplicaOfflineMsgCatchUpTest 2/2, SynchronizationMsgTest 78/78, Skipped: 0. The description says so, with the mutation table and "Left out" updated.

[maximthomas]

praise: The forward is now reported by the one thread that knows the write returned.

• Session.run() calls written(outgoing.whenWritten) only after send() returns, and continues past it on IOException (Session.java:619-621).
• publish(ReplicationMsg) keeps its void signature and delegates with a null callback (Session.java:327), so no caller without a callback changes.
• written() keeps a failing callback from taking the session thread down (Session.java:389-397).

[vharseko]

@maximthomas thanks for the approval. #1035 landed right after it and the branch went conflicting, so it is rebased onto master at dd94891ad6; the one [#1055] commit is now dacc944545. This says what the conflict was, how it was taken, and the one case it added - please take another look, since the approved head is gone.

The conflict. #1035 made Session.close() send what the publisher left in sendQueue, and this PR had turned the elements of that queue into Outgoing (the buffer and its callback). They met in Session:

• the drain of the close writes outgoing.buffer and runs written(outgoing.whenWritten) after each write which returned - so a ReplicaOfflineMsg the close sends on the way out is reported as forwarded, from the thread which closed the session, rather than written in silence; what the close gives up on (budget spent, a failed write, a session which had already failed) never has its callback run;
• takeBackWhatWasQueuedTooLate() takes the Outgoing back and says whether it did, and publish() returns false for a message taken back - it was neither written nor left queued;
• the javadoc of close() and publish(msg, whenWritten) say both of these.

What followed from it.

• The comments which said that close() drops the queue - the javadoc of forwardReplicaOfflineMsg() and two in ReplicationServerShutdownSyncTest - now say what it does since Send what a replication session's publisher left queued when it is closed #1035. The reason for the callback stands: a forward reported on the queue ends the wait of the shutdown before the peer was told, and leaves the message to the budget of the close rather than to the grace period.
• Send what a replication session's publisher left queued when it is closed #1035's SessionPublisherDrainTest fills the queue through its field with bare buffers and failed with a ClassCastException; it now builds the Session$Outgoing element.
• The drain calling the callback was a seam no case pinned (the mutant which drops it survived), so SessionPublisherDrainTest got aCloseRunsTheCallbackOfEachMessageItSendsOutOfTheQueue - seven queued messages with callbacks on a session never started: each callback runs once, in the order of the writes, on the closing thread, and the peer receives exactly those messages - and aCloseWhichCannotSendTheQueueReportsEveryMessageTheQueueHeld now asserts that no callback runs for a queue whose write failed.

mutation	on dacc944545
the forward reported once publish() returns, as on master	red on thePeerStillReadingAnEarlierChange... alone - now at the writer reported the message forwarded while it was still queued, since the drain of #1035 does deliver the message
the drain of the close never runs the callback	red on aCloseRunsTheCallbackOfEachMessageItSendsOutOfTheQueue

The !accepted arm stays as recorded in the previous round - unpinned.

On this head: ReplicationServerShutdownSyncTest 21/21, SessionTest 3/3, SessionPublisherDrainTest 9/9, DSRSShutdownSyncTest 29/29, ReplicaOfflineMsgCatchUpTest 2/2, SynchronizationMsgTest 78/78, Skipped: 0. The description says so, with the account of #1035 and the mutation table updated.

[maximthomas]

praise: The forward is reported from the thread that knows the write returned, and the rebase over #1035 gives both writers one contract.

• Session.run() hits continue after a failed send() (Session.java:854), so written() (:856) never runs for a write that threw.
• #1035's drain calls written(outgoing.whenWritten) only after its own send() has returned (Session.java:370, :385). A ReplicaOfflineMsg that the close sends out of the queue is still reported, and one the close gives up on is not.
• ServerWriter.forwardReplicaOfflineMsg() calls replicaOfflineMsgNotForwarded when the session refuses the message (ServerWriter.java:195), so the shutdown does not wait for a write that will never happen.

---

suggestion (non-blocking): No case pins the take-back refusal of publish(msg, whenWritten), so the mutant return true; at Session.java:560 survives.

opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java:560, :535-538, opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java:193-196

A message whose offer succeeds after close() has cleared isRunning and drained the queue is taken back and reported as refused, as the javadoc and the description both say. No test reaches this arm. SessionPublisherDrainTest publishes after the close (:608, :650), but through publish(msg) after close() has returned, which takes the synchronous branch. Nothing in the replication protocol tests reaches takeBackWhatWasQueuedTooLate(). The only consumer of this false is the !accepted arm of forwardReplicaOfflineMsg(), which "Left out" already lists as unpinned. This road into that arm came with the rebase over #1035 and has its own mutant. Not run.

Pin: only a publish() that is already inside offer() when close() sets closeInitiated can reach the arm, and only if its offer lands after the drain's last poll. SessionTest's fixture cannot schedule that. A package-private seam in Session, between the successful offer and the isRunning read, would let a case run close() to completion, release the offer, and then assert that publish() returns false and the callback never runs. That kills return true; at :560.

---

nitpick (non-blocking): SessionTest's class javadoc says every case holds the session thread inside a write, but only the first case does.

opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java:44-47

newBlockingMsg() is published only at :76. In the second case the message is refused before any write, and the third case never starts the session thread. The 8 KiB buffers are set for every case, so the sentence describes the fixture, not each test.

 * The peer of each test reads nothing until the test lets it, and both ends of the connection
 * have socket buffers far smaller than {@link #BLOCKING_MESSAGE_SIZE}, so that a test which
 * publishes a message of that size holds the session thread of the end under test inside its
 * write for as long as it wants - the state in which a message published behind it is queued
 * and not written.

---

note (non-blocking): CI: the red cell is not this PR's.

• build-maven (ubuntu-latest, 17) fails only on PDBStorageTest.testCanAddLargeValues with OutOfMemoryError: Java heap space. That is the #1071 flake, which still shows up after #1072. In the same cell SessionTest passes 3/3, SessionPublisherDrainTest 9/9 and ReplicationServerShutdownSyncTest 21/21. Re-running the cell clears it.

[vharseko]

@maximthomas thanks. All three points are taken. The branch is rebased onto master at 0793e27bbc without conflicts, and the round is a commit of its own over the [#1055] one (now 6ae52a56b6): 7c7f055890. Please take another look, since the approved head is gone again.

suggestion - the take-back refusal of publish(msg, whenWritten). Agreed, and so is the reason no fixture reaches it without a seam. I looked for a way around one first, and none holds:

• sendQueue is final, so a queue whose offer() blocks could only be swapped in by writing a final field through reflection, which JDK 26 (a CI leg) warns about under JEP 500.
• isRunning is not final, but both of its reads are on the wrong side of the offer: holding the first ends the publish at closeInitiated, and by the second the message is already queued for the drain to send.
• A full queue bounds the wait to one 100 ms offer(), and the drain takes the message before the take-back finds it. That is a race, not a state.

So Session gets the hook you describe: beforeQueueing(Runnable), package-private and @VisibleForTesting. publish() runs it inside its loop, between the closeInitiated check and offer(). Unset, it costs one volatile read.

SessionPublisherDrainTest.aMessageQueuedAfterTheCloseDrainedTheQueueIsTakenBackAndReported holds a publish() on a started session at the hook, runs close() to its end, and then lets the offer go. It asserts that:

• publish() returns false;
• the callback never runs;
• the queue is empty;
• the take-back is reported once (1 message(s) ... it was queued after the publisher of the session had stopped);
• the peer reads the StopMsg with nothing before it.

The sending end keeps a reader across the close, as aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed does, so that the close ends with a FIN rather than a reset which would cut the StopMsg off.

mutation	on 7c7f055890
return true; at the take-back	red on aMessageQueuedAfterTheCloseDrainedTheQueueIsTakenBackAndReported alone - a message queued after the close drained the queue was answered as queued, though nothing is left to send it

This pins the session's side of the refusal, and the take-back #1035 left unpinned on master along with it. The writer's !accepted arm is still not driven by any case, and "Left out" now says exactly that.

nitpick - SessionTest's class javadoc. Replaced with your wording: the buffers are what the fixture sets for every case, and holding the thread is what a test publishing a message of BLOCKING_MESSAGE_SIZE does.

note - CI. Confirmed from the log of build-maven (ubuntu-latest, 17): the one failure is PDBStorageTest.testCanAddLargeValues with OutOfMemoryError: Java heap space. That is #1096, fixed by #1097, which is under review. The push re-runs the whole matrix anyway.

On this head: SessionPublisherDrainTest 10/10, SessionTest 3/3, Skipped: 0. The description lists the hook, the new case, the mutation row and the narrowed "Left out" item.