[#1202] [#1209] [#1222] Give the replication server a bounded time to give credit and to read once the broker is being stopped, and write nothing after a message given up (#1210)

Fixes #1202
Fixes #1209
Fixes #1222

## Summary

`ReplicationBroker.stop()` could not end while its replication server
kept the session up and took no update from it, and with it hung
`ReplicationDomain.disableService()`, which calls `stop()` holding
`serviceStateLock` and `sessionLock` - the shutdown of the domain, an
import, a restore, a session restart. It hung in two ways: waiting for
window credit the replication server did not give (#1202), and, once the
socket buffers were full, in a write to the session which the
replication server did not read (#1209) - with the default window of
100000 updates, the buffers fill long before the window closes. The stop
now gives the replication server a bounded time to give credit, and as
long again to read what was written, after which it closes the session
as a failed one; once a publish gave its message up for the stop, the
broker writes nothing more. The same write held the session with no stop
at all once the replication server was gone without a reset (#1222): the
heartbeat monitor, which then closes the session, waited for it as well,
until TCP gave up retransmitting. It now closes such a session as a
failed one too.

## Root cause

- The one flag which ends the wait of a publish for window credit is
`shutdown` (`while (!done && !shutdown)` in
`ReplicationBroker.publish()`), and `stop()` set it only after
`domain.publishReplicaOfflineMsg()` returned.
- `ReplicaOfflineMsg` extends `UpdateMsg`: it takes a permit of the send
window like a change. With the window spent, the stop waited for the
credit of its own replica offline message even with no local change in
flight - wider than the case #1202 describes.
- In the case of #1202, a local change waits for credit inside
`PendingChanges.commitAndPushCommittedChanges()`, under the monitor
`PendingChanges.putReplicaOfflineMsg()` needs: the stop waits for that
change, which waits for credit.
- Credit comes only as the replication server takes the updates of the
session, while its heartbeats come from a thread of their own and keep
the session up. Nothing else ended the wait.
- A directory server writes its session on the thread which publishes -
its `Session` runs no publisher thread - and a local change is published
under the lock of the pending changes. A write which the replication
server does not read blocks in the socket once the buffers are full,
holding that lock, `connectPhaseLock` and the `publishLock` of the
session (#1209). A socket write has no timeout, and `Session.close()` of
a session with no error takes `publishLock` first: nothing on the
directory server ended that write.
- The heartbeat monitor, which closes the session once the heartbeats of
the replication server stop, closed it with `Session.close()` too
(#1222). A replication server which is gone without a reset acknowledges
nothing: the writes fill the buffers, the next one blocks, and the close
waited for it until TCP gave up retransmitting - while the listener
stayed in a read which neither data nor a reset ends, and the local
changes waited for the lock of the pending changes.

## Change

`ReplicationBroker`:

- `stop()` records the stop in progress, with a deadline, before it
publishes the replica offline message: a `StopInProgress` in the
`volatile` field `stopping`, which `start()` clears to `null`. The
deadline is the connection timeout of the replication provider
(`MultimasterReplication.getConnectionTimeoutMS()`, 5 seconds by
default): the replication server gets as long to give credit as it gets
to accept a connection. One which is only slow still takes the changes
and the replica offline message, as before. A connection timeout of 0,
which sets no limit to a connection or to an SSL negotiation, sets none
to the stop either: the stop then waits as it did before.
- A stop which overlaps one in progress keeps that one, its deadlines
and what it gave up, rather than replacing it.
- A publish which finds no credit once the deadline is past gives its
message up and marks the stop (`StopInProgress.gaveUp`), which ends the
publish loop for every later message as well, until `start()` clears the
stop. Before it marks the stop it reads it again: if `start()` ran
since, the loop head decides. What is given up:
- a local change, which the next session republishes from the historical
information of its entry - or a local delete, one the recovery publishes
included, from the deletes the domain keeps aside in memory, which a
restart of the server loses;
- or the replica offline message, which
`PendingChanges.putReplicaOfflineMsg()` then reports as not sent, its
announcement withdrawn.
- Nothing is written by `publish()` after a message given up: the
replica offline message carries the newest CSN of the replica, and
written after a change given up it would tell the replication server
that nothing older is to come.
- The give-up is marked on the stop, not on the broker. A publish which
found the stop past its deadline may mark it only after `start()`
cleared it - the read and the write are not atomic, and no lock is taken
for them - and it then marks a stop which is over, rather than the
broker started again, which would refuse every publish until its next
stop.
- `stop()` starts a watchdog for the stop, unless the connection timeout
is 0 (`endWhatStillHoldsTheStop()`). If the stop is still held when it
is past its credit deadline by as long again - twice the connection
timeout from its start - what holds it is no wait for credit, which
gives up by itself at the deadline, but a write the replication server
does not read, or a lock such a write holds. The watchdog then marks the
give-up of the stop, logs a warning which names the replication server,
and closes the session as a failed one (`Session.abort()`): the write
fails, and the publish which made it returns rather than retries. A stop
which waits for credit and gives up still ends its session in order,
with its `StopMsg`.
- The watchdog acts under the monitor of the stop, and `stop()` ends the
stop under `startStopLock` once it is done: the watchdog of a stop never
closes a session of a broker started again.
- A publish reads the give-up again under `connectPhaseLock`, before it
writes a message and before it writes a window probe. A publish which
passed the loop head before the give-up may wait for that lock behind a
connection of the listener - one which the replication server ending the
session in the stop set off, and which takes the connection timeout for
each replication server that answers nothing - past the deadline of the
watchdog, which then finds no session to close. Without that read the
publish wrote to the session the connection opened, and a replication
server which did not read it would hold the stop with nothing left to
end it. The watchdog marks the give-up before it reads the session, and
a connection sets its session before it releases the lock: either the
watchdog closes that session, or the publish finds the mark.
- Once the stop gave up, the broker opens no session again while the
stop runs (`reStart()`): the listener, which the close of the session
wakes, waits for the end of the stop rather than connects. A session
opened then would carry nothing but its change time heartbeats - which
tell the replication server that nothing older is to come from this
replica, while it still has the changes given up - and the stop would
wait for the connection. A connection already under way when the stop
gives up still opens its session: `publish()` writes nothing to it, and
the stop closes it as it ends.
- `stop()` stops the broker in a `finally`: a failure of the replica
offline message no longer leaves a stopping broker running, which would
give up every publish that waits for credit. The watchdog is started
inside the same `try`: a thread which cannot be started leaves no
stopping broker running either.

`Session.abort()` (new) records the error, marks the close begun, closes
the sockets, and then closes the session, which takes the branch of a
failed session: it writes nothing more - neither what the publisher
thread left queued nor the `StopMsg` - and takes no lock. The sockets
closed first end a write blocked in them, whichever thread makes it -
the publisher thread, which any close joins, included - and release a
close which had begun before, with no error, and waits for the lock of
the writes. The close is marked begun before them, so that the listener
whose read their close ends finds it begun, and does not report it as a
failure of the replication server.

`HeartbeatMonitor` closes a session whose heartbeats stopped with
`Session.abort()` (#1222): the blocked write and the read of the
listener fail, and the reconnection proceeds as for any session which
failed; no `StopMsg` is written, which a replication server whose
heartbeats stopped is not expected to read. A monitor stopped with its
broker still closes the session as before.

The javadoc of `ReplicationDomain.publish(UpdateMsg)` and the trace of
`LDAPReplicationDomain.publishReplicaOfflineMsg()` name the new reasons
a message is not published, and the javadoc of
`DSRSShutdownSync.replicaOfflineMsgNotSent()` counts the wait of a
stopping broker in the window between an announcement and its
withdrawal. The synopsis of `connection-timeout`
(`ReplicationSynchronizationProviderConfiguration.xml`) says that it
also bounds the wait of a stopping domain for credit, and its
description says when a domain is stopped, what raising or lowering the
value does to such a stop, that a replication server which does not read
what was written gets as long again before the session is closed, and
that 0 sets no limit.

## Not changed

- A publish held up by an `IOException` still retries until the broker
is connected again or finds no replication server to connect to - unless
the stop gave up.
- A recovery publish given up is refused to the recovery, as one refused
once `shutdown` is set (#1204): the round ends, and the changes from it
- a delete included, which stays aside - are left to the recovery of the
next session.
- An assured change given up for the stop still waits for its
acknowledgment until the assured timeout, and is counted as sent and
timed out. Ending that wait early needs a mark of its own in the assured
machinery: a CSN taken out of the messages which wait for an
acknowledgment reads as acknowledged.
- A change given up comes back with the recovery of the next session,
which runs in a thread of its own, and the change time heartbeats of
that session do not wait for it. When the recovery takes longer than a
heartbeat interval, the replication server takes a heartbeat as "nothing
older is to come" from this replica, and the change gets its change
number after newer changes of other replicas. Nothing is lost; any
change a recovery publishes does the same on master.
- The change time heartbeats, and the window and status messages written
straight to the session, are not held back after a give-up: only
`publish()` reads the mark, until the session is closed. That holds for
the session a connection already under way at the give-up opens too,
until the stop closes it as it ends.
- A replication server which drops the session after a give-up for
credit is reported (`WARN_REPLICATION_SERVER_BADLY_DISCONNECTED`), as
with no stop at all: the give-up closes nothing. Only a close this
server makes - by the watchdog of a stop, or by the heartbeat monitor -
is not, as the close is marked begun before the sockets are closed.
- A total update this domain is sending ends with an error once the stop
gives up, as every message of it is refused - as it does on master as
soon as a stop with the window open sets `shutdown`. Only the stop which
waited for credit for good let it go on.
- A session the watchdog or the heartbeat monitor closes ends with no
`StopMsg`: the replication server sees the connection lost.
- A reconnection of the listener to a better replication server
(`reStart()` from `receive()`) still closes the current session with no
error, and waits for a write blocked in it.
- With the window open, a stop still waits for the end of a round #1204
holds the local changes back for, as #1202 records.

## Tests

`ReplicationBrokerStopTest` (new) connects a domain to a fake
replication server which keeps the session up and reads all of it, but
opens a window of one update - or of the default 100000, in the cases of
a write it does not read - and gives no credit after it unless the test
asks it to. Like a replication server, it reports as its state the
changes it received; it serves each connection on a thread of its own,
and stops reading its sessions when the test asks it to. It sends the
heartbeats the domain asks for, and can go away altogether: no reading,
no heartbeat, no answer to a connection. It ends a session in order or
drops it when the test asks it to. In the case of a connection under
way, more replication servers take each connection and, once told to,
answer nothing on it, as one which hangs does: a connection to each
waits for the connection timeout in its SSL negotiation. Once the domain
closed its session while the server read nothing, the server ends that
session with a heartbeat, which the closed end answers with a reset:
what the closed socket left queued, its end included, comes only once
its TCP finds the window open again, and after a stall of 14 seconds its
next window probe is 13 seconds away on Linux - the CI run of round 2
failed `aDisableEndsAWrite…` waiting for that end. A publish which waits
for credit shows as a `WindowProbeMsg` on that server; "the stop waits
for the lock" is the disabling thread `BLOCKED` in
`PendingChanges.putReplicaOfflineMsg()`; a write it does not read is a
writer which stays in `Session.send()` while the number of its changes
stands still for two seconds in a row - half a second could be a pause
of the collector, as a run after cases which allocate a lot showed;
"reported as a close by the replication server" is a
`WARN_REPLICATION_SERVER_BADLY_DISCONNECTED` in the error log; "the
watchdog acted" is the end of its thread. The class runs with a
connection timeout of 2 seconds instead of the 30 seconds
`config-changes.ldif` gives the suite, and a case which needs another
one sets it for its own run.

| Case | What it pins | master | round 1 |
|---|---|---|---|
| `aDisableEndsWhenTheReplicationServerGivesNoCredit` | Once a change
spent the window, disabling the domain ends, the replica offline message
not written - no sooner than the connection timeout of 6 seconds the
case sets, and within 22 seconds; the announcement of the message is
withdrawn, and the session ends in order, with a `StopMsg`. | red |
green |
| `withNoConnectionTimeoutAStopWaitsForTheCredit` | With a connection
timeout of 0, the replica offline message still waits for credit past
its first wait and past 7 seconds, beyond the shipped default, and goes
out once the credit comes. | green | green |
| `aDisableEndsWhileALocalChangeWaitsForCredit` | Disabling while a
second change waits for credit under the lock ends, and the change ends
with it. | red | green |
| `aStopPublishesWhatTheReplicationServerTakesInTime` | Credit given a
second into the stop gets that change out, and then the replica offline
message, last. | green | green |
| `nothingIsWrittenAfterAChangeTheStopGaveUp` | A change given up is
followed by nothing, even with credit offered right after it; a broker
started again republishes it with the recovery of its session, and then
writes again. | red | green |
| `aSessionRestartRepublishesTheChangeTheStopGaveUp` | A restart of the
session (`restartService()`) gives the change up as a disable does, and
the recovery of the next session republishes it. Unlike `enable()`,
which loads the state again and repairs it from `ds-sync-hist`, a
restart keeps the state in memory: the change is found missing only
because `ReplicationDomain.publish()` put it in the state although the
broker did not write it. | red | green |
| `aStopWritesNothingAfterAGiveUpEvenWithCredit` | A stop which gave a
change up refuses the next one with the window open again, and an
`EntryMsg` and a `DoneMsg`, which need no credit: no race with the 500
ms wait for credit. | red | green |
| `aStopWhichOverlapsOneInProgressKeepsItsGiveUp` | After a change given
up, the first stop writes nothing more, with the window open again,
while a second stop runs beside it. | red | red |
| `aBrokerStartedAgainWaitsForCredit` | A stop with the window open
writes the replica offline message; a broker started again waits for
credit past twice the connection timeout of that stop, and keeps its
session. | green | green |
| `aStopWhichCannotPublishTheOfflineMessageStillStopsTheBroker` | A stop
whose replica offline message fails rethrows the failure and still stops
the broker. | red | green |
| `aDisableEndsAWriteToAReplicationServerWhichReadsNothing` | A local
change blocked in its write under the lock of the pending changes, with
credit to spare: disabling ends, no sooner than twice the connection
timeout of 6 seconds and sooner than two and a half times it; nothing is
written after the change, no session is opened again during the stop,
nothing reports the close as one by the replication server, and the
recovery of the session after `enable()` republishes the change. | red |
red |
| `aStopEndsItsOwnWriteAndOpensNoSessionAgain` | A write of the stop
itself, with no lock of the pending changes in the way, ends the same
way, no sooner than twice the connection timeout; the listener the close
wakes waits for the end of the stop rather than connects, nothing
reports the close as one by the replication server, and nothing is
written after the write given up. | red | red |
| `theHeartbeatMonitorEndsAWriteToAReplicationServerWhichIsGone` | A
local change blocked in its write to a replication server which then
reads nothing, sends no heartbeat and answers no connection: the write
ends, no sooner than two heartbeat intervals, nothing reports the close
as one by the replication server, and once the server is back the domain
connects again and its change reaches it. | red | red |
| `aConnectionUnderWayWhenTheStopGivesUpGetsNothingWritten` | The
replication server ends the session in the stop, and four more which
answer nothing hold the connection of the listener for the connection
timeout each. The watchdog acts at its deadline with the connection
still under way; an entry the stop publishes, which waits for the lock
of that connection, is not written to the session the connection opens,
and the stop ends once the connection does. | red | red |
| `aDropByTheReplicationServerAfterAGiveUpIsStillReported` | After a
change given up for credit, a replication server which drops the session
with no `StopMsg` is reported, and the stop ends. | red | n/a |

The fake replication server drops a session with `Session.abort()`,
which round 1 does not have: `aDropBy…` cannot run there. On master its
stop waits for credit and does not end. Round 2 (`26f7da4439`) fails
`aDropBy…` and `aConnectionUnderWay…`, and round 3 (`4d1130e44e`) fails
`aConnectionUnderWay…`, and only it.

`SessionTest` gains three cases, red on master and on round 1, which
have no `Session.abort()`; round 2 fails the third:

| Case | What it pins |
|---|---|
| `anAbortEndsAWriteThePeerDoesNotReadAndACloseWhichWaitsForIt` | A
write on the publishing thread, which a peer that reads nothing holds,
and a close which waits for the lock of the writes behind it, both end
with an abort; the peer finds the message cut short and the end of the
stream, with no `StopMsg`. |
| `anAbortEndsAWriteOfTheThreadOfTheSession` | A write of the thread of
the session, which any close joins, ends with an abort, and the thread
with it. |
| `anAbortMarksTheCloseBegunBeforeItClosesTheSockets` | The socket of
the session records, as it is closed, whether the close was marked begun
by then: an abort marks it before it closes the sockets, so that the
listener whose read their close ends finds it begun. |

Mutants, each run against the classes of round 4 (`d3a6c33741`):

| Mutant | Caught by |
|---|---|
| the stop not cleared by `start()` |
`aBrokerStartedAgainWaitsForCredit`, `aDisableEndsAWrite…`,
`aSessionRestart…`, `nothingIsWritten…` |
| the stop recorded only after the replica offline message |
`aConnectionUnderWay…`, `aDisableEndsAWrite…`, `aDisableEnds…NoCredit`,
`aDisableEnds…WaitsForCredit`, `aDropBy…`, `aSessionRestart…`,
`aStopEndsItsOwnWrite…`, `aStopWhichOverlaps…`, `aStopWritesNothing…`,
`nothingIsWritten…` |
| no deadline: give up at the first wait without credit |
`aDisableEnds…NoCredit`, `aStopPublishes…`, `withNoConnectionTimeout…` |
| give-up not kept for the later messages | `aStopWhichOverlaps…`,
`aStopWritesNothing…`, `nothingIsWritten…` |
| a stopping broker writes nothing at all |
`aBrokerStartedAgainWaitsForCredit`, `aConnectionUnderWay…`,
`aDisableEnds…NoCredit`, `aStopEndsItsOwnWrite…`, `aStopPublishes…`,
`withNoConnectionTimeout…` |
| only the replica offline message given up | `aDropBy…`,
`aStopWhichOverlaps…`, `aStopWritesNothing…` |
| no `finally` around the replica offline message |
`aStopWhichCannotPublish…` |
| a connection timeout of 0 gives up at the first wait (round 0) |
`withNoConnectionTimeout…` |
| a connection timeout of 0 read as the shipped 5 seconds |
`withNoConnectionTimeout…` |
| a constant deadline of 5 seconds instead of the connection timeout |
`aConnectionUnderWay…`, `aDisableEndsAWrite…`, `aDisableEnds…NoCredit`,
`withNoConnectionTimeout…` |
| a constant deadline of 30 seconds | `aConnectionUnderWay…`,
`aDisableEndsAWrite…`, `aDisableEnds…NoCredit`,
`aDisableEnds…WaitsForCredit`, `aDropBy…`, `aSessionRestart…`,
`aStopEndsItsOwnWrite…`, `aStopWhichOverlaps…`, `aStopWritesNothing…`,
`nothingIsWritten…` |
| `ReplicationDomain.publish()` puts in the state only what the broker
wrote | `aSessionRestart…` |
| an overlapping stop replaces the one in progress |
`aStopWhichOverlaps…` |
| the give-up of a publish reported as a message written |
`aDisableEnds…NoCredit` (the announcement is left) |
| after a give-up, only the updates refused | `aConnectionUnderWay…`,
`aStopEndsItsOwnWrite…`, `aStopWritesNothing…` |
| no watchdog (round 1) | `aConnectionUnderWay…`, `aDisableEndsAWrite…`,
`aStopEndsItsOwnWrite…` |
| the watchdog closes the session but gives nothing up |
`aConnectionUnderWay…`, `aDisableEndsAWrite…`, `aStopEndsItsOwnWrite…`
(a session opened again) |
| the watchdog acts at the credit deadline | `aDisableEndsAWrite…`,
`aDisableEnds…NoCredit` (no `StopMsg`), `aDropBy…`,
`aStopEndsItsOwnWrite…`, `aStopWhichOverlaps…`, `aStopWritesNothing…` |
| the watchdog acts whether the stop ended or not |
`aBrokerStartedAgainWaitsForCredit` |
| a watchdog with a connection timeout of 0 | `withNoConnectionTimeout…`
|
| the write deadline a constant 5 seconds past the credit deadline |
`aConnectionUnderWay…` (the watchdog had not acted),
`aDisableEndsAWrite…` |
| the write deadline three times the connection timeout from the start
of the stop | `aConnectionUnderWay…` (the watchdog had not acted),
`aDisableEndsAWrite…` |
| the write deadline two and a half times it | `aConnectionUnderWay…`
(the watchdog had not acted), `aDisableEndsAWrite…` |
| a session opened again once the stop gave up | `aDisableEndsAWrite…`,
`aDropBy…`, `aStopEndsItsOwnWrite…` |
| no read of the give-up under `connectPhaseLock` before the write, as
round 3 did | `aConnectionUnderWay…` |
| the listener reports no drop by the replication server once the stop
gave up, as round 2 did | `aDropBy…` |
| `Session.abort()` a plain close | `anAbortEndsAWriteThePeer…`,
`anAbortEndsAWriteOfTheThread…`, `aDisableEndsAWrite…`, `aDropBy…`,
`aStopEndsItsOwnWrite…`, `theHeartbeatMonitorEnds…` |
| `Session.abort()` closes the session before the sockets |
`anAbortEndsAWriteOfTheThreadOfTheSession` |
| `Session.abort()` leaves the sockets to the close | both `SessionTest`
cases of the abort |
| `Session.abort()` leaves a close which had begun before to wait |
`anAbortEndsAWriteThePeerDoesNotReadAndACloseWhichWaitsForIt` |
| the heartbeat monitor closes a session whose heartbeats stopped as one
with no error, as round 2 did | `theHeartbeatMonitorEnds…` |
| `Session.abort()` closes the sockets before it marks the close begun,
as round 2 did | `anAbortMarksTheCloseBegun…` |
| no second read of the stop before its give-up is marked | not caught -
see below |
| a window probe written after the give-up | not caught - see below |

Under the mutant which gives up only the replica offline message, a
local change which waits for credit under the lock never gives up, and
the watchdog ends it at twice the connection timeout: that is why that
mutant no longer fails the cases where such a change holds the stop, as
it did in round 1.

The mutant "the state only what the broker wrote" survives a disable
followed by `enable()`: `enable()` loads the state again, and
`PersistentServerState.checkAndUpdateServerState()` repairs the CSNs of
this replica from `ds-sync-hist`. A restart of the session keeps the
state in memory, and only there does the mutant lose the change.

Two races are not pinned, as a test cannot put a thread between the two
steps without a hook: a give-up marked after `start()` cleared the stop
it was read from - the mark lives in the object of that stop, and the
second read of the stop narrows the window to that of master's loop head
- and a watchdog acting after the end of its stop, which it does under
the monitor `StopInProgress.end()` takes, under `startStopLock`.

Nor is the start of the watchdog inside the `try`: a thread which fails
to start needs a hook as well. The read of the give-up before a window
probe is not pinned either: the probe is a few bytes which neither hold
the stop nor tell the replication server anything of the changes, and
the read only keeps `publish()` from writing anything after a give-up.

## Verification

- `ReplicationBrokerStopTest` and `SessionTest` on their own at round 4
(`d3a6c33741`, on `616d5f6c17`), on macOS: 15/15 four times and 6/6
three times. Master fails 12 cases of the first - all but
`aBrokerStartedAgain…`, `aStopPublishes…` and `withNoConnectionTimeout…`
- and the 3 cases of the abort of the second. Round 1 (`8fb5973e30`)
fails the 4 earlier cases the table marks red for it,
`aConnectionUnderWay…`, and `aDropBy…`, which cannot run there, and the
3 cases of the abort. Round 2 (`26f7da4439`) fails
`aConnectionUnderWay…`, `aDropBy…`, `theHeartbeatMonitorEnds…` and
`anAbortMarks…`. Round 3 (`4d1130e44e`) fails `aConnectionUnderWay…`
only. Every mutant above is caught, but the two noted; the runs a port
collision with a test run of another worktree voided were run again.
- On Linux in a container (JDK 21): `ReplicationBrokerStopTest` 15/15
twice, and `aConnectionUnderWay…`, `aDropBy…` and `aDisableEndsAWrite…`
ten times more.
- `ReplicationSynchronizationProviderConfiguration.xml` validates
against `admin.xsd`.
- Rebased onto `d954c4c53f` (#1204, #1211, #1219 and #1218 came in), at
`276317c33a`, whose commit on top of round 4 changes a comment only; on
macOS with JDK 21: `ReplicationBrokerStopTest` 15/15 and `SessionTest`
6/6, twice each, and the neighbours which share `publish()`:
`ReplicationBrokerTest` (#1218) 37/37 and `HistoricalCsnOrderingTest`
(#1204) 54/54.