[#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, and Session.close() interrupts and joins the session
thread without draining that queue. 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, reach close() with the message still queued: 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. 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 - and never for a message which was not written: one whose
  write failed, or one still queued when the session is closed. It returns false for a message
  it neither wrote nor queued: one the protocol version of the peer cannot carry, or one
  published while the session is being closed. 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.

Session.close() is unchanged. What it drops is simply no longer reported as forwarded, so 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.

ReplicationServerShutdownSyncTest (17, 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. On master the peer receives the StopMsg alone, with the forward reported by the writer
  and the shutdown over in 39 ms.
• theShutdownStopsWaitingForAPeerWhoseProtocolCannotCarryTheMessage connects a peer speaking
  protocol version 7, for which the message has no encoding, and pins the give-up: the refusal
  of the session must strike the peer off, not leave the shutdown waiting for a forward nobody
  can report.

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 129fc4e346 (#987), which split the handshake of FakePeerReplicationServer
into the two phases its own cases need - handshaking() stops before the TopologyMsg,
completeHandshake() sends it - and turned the constructors into the factories connected(...).
This PR's two ways in ride on that shape: connected(port, id, baseDN, generationId, windowSize, protocolVersion) for the peer speaking version 7, and connected(session, socket, id, baseDN, generationId, windowSize) for the peer over the connection the test established itself. Both run
the first phase in a private constructor over a shared start(), the second through completed(),
and abandon() closes what a failed first phase leaves behind, as before. The class carries the
two cases of #987 with it. On this head: ReplicationServerShutdownSyncTest 17/17, SessionTest 3/3,
DSRSShutdownSyncTest 29/29.

Every assertion was watched failing against the code it pins:

mutation	fails with
the forward reported after publish() returns, as on master	thePeerStillReadingAnEarlierChange... - the peer was never told that the replica went offline ... (forward reported by [99], the shutdown took 39 ms); theShutdownStopsWaitingForAPeerWhoseProtocolCannotCarryTheMessage - the writer let the shutdown wait for a peer whose protocol cannot carry the message
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 writer ignores a message the session refused	theShutdownStopsWaitingForAPeerWhoseProtocolCannotCarryTheMessage alone

Regression: 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.

Left out

• A message whose protocol version the peer cannot carry used to be reported forwarded, which
  released a wait with no recipient recorded - the relay and A ReplicaOfflineMsg still queued behind an uncommitted change is recorded as sent, and the shutdown then waits its whole grace period for nothing #918 cases - at once; it is a
  give-up now, which such a wait does not see, so a shutdown whose only peers predate protocol
  version 8 spends the grace period on those announcements. A peer that old has no
  ChangeNumberIndexer to tell, and the alternative is to keep calling a message nobody wrote
  forwarded.
• 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 synchronous write of the StopMsg, with no bound
  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, and its own issue.

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