[#1195] Publish every change done while a recovery runs, and end the recovery only once they are out (#1204)

Fixes #1195, fixes #1197, fixes #1199, fixes #1200, fixes #1201, fixes
#1220, fixes #1221, fixes #1223, closes #1203.

## The defect

When a directory server connects to a replication server which lacks
some of its changes, `LDAPReplicationDomain.RSUpdater` republishes them
from the historical information of the entries
(`buildAndPublishMissingChanges()`), searching by windows of 10 seconds,
while the local changes done meanwhile are only recorded in the domain
state (`PendingChanges.pushCommittedChanges()`) and left to it. Under
load it loses changes three ways.

In the run of #1195 the one switch of replication server lost 81 changes
of cv-c3, not the two deletes the convergence check saw: 21 adds, 2
deletes and 58 modifications. The sequence numbers of that replica in
the changelog have 12 gaps, all of them in the window of that recovery
and none elsewhere in the run. Later deletes healed the adds, and a
check which compares DNs does not see the modifications.

1. **Search against state (67 of the 81).** A round chose what to
publish by the domain state read after its search (`state.cover()`). A
change committed while the search ran was not in what it found but was
covered by that state, so a later change went out in its place and the
next round started past it. Deletes escaped only because they are kept
aside as they are done.
2. **The "no changes" jump (14).** A round which found nothing it could
publish moved ten seconds ahead, which ended the recovery whatever had
been done meanwhile: what the pending changes had left to it was never
searched for, and the deletes kept aside were cleared.
3. **The hand-over.** The pending changes went back to publishing before
the broker took them: its recovery flag was cleared only when the thread
ended, and a change pushed in between was refused while the domain state
recorded it as done.

The replication server does not notice: it takes no change of a replica
older than the newest one it holds
(`LogFile.appendWouldBreakKeyOrdering()`), and its server state holds
the newest CSN of each replica, so `missing-changes` reached 0.

A publisher which failed also left the pending changes in recovery, so
no later local change of the session was published (#1197), and a round
whose search failed part-way still published what it had found, which
made the changes it missed undeliverable (#1201).

A session which started while the publisher of the previous one ran got
no recovery of its own (#1199): the running publisher went on from where
it had got to, on the new session, whose replication server may hold
less than the one before and then took none of the older changes it
missed; or it had already ended its recovery and left without running
the one the new session needed. The list of the deletes kept aside,
capped at 10000, took what the searches found as well, and dropped its
oldest entry - a change a round had just found, or a delete no
replication server had - without a trace (#1200); it also kept a local
delete which did not go through, and the deletes the data no longer
reflected once it was restored or imported (#1221). Rebuilding the local
changes of a recovery from the historical information races the commits,
cannot publish a change whose values a later change replaced, and holds
the local changes back while it searches (#1203). And for as long as a
recovery ran, the broker refused every message which was not a recovery
one - the acknowledgements of the changes replayed here, the messages of
a total update, the monitoring requests - not only the changes of this
replica (#1220). A domain which read back a ServerState with no change
of its own - the delete of the base entry takes the saved state with it
- had its CSN generator go back to the clock, and numbered its next
change as, or before, a change of this replica the replication server
missed: one of the two was lost (#1223).

## The change

- **The local changes done during a recovery are kept (#1203).** While a
recovery runs, the pending changes keep the messages of the local
changes they push, in order, rather than drop them. The search of the
recovery only covers the changes pushed before the first one it kept: up
to a CSN fixed then, which no commit moves - every change up to it has
its entry written, so the search sees it. The kept messages go out next,
as they were done, a change whose values a later change replaced
included. Once a round has caught up with the local changes and some
were pushed while it ran, the round after it holds them back
(`PendingChanges.holdBackLocalChangesWhile()`, under the lock the
pending changes are pushed under) to publish those, and the recovery
ends with it: the local operations wait for that publish alone, and no
search runs under the hold. `PendingChanges.recoveryUntil()` switches
the pending changes and the broker back together, under that lock.
- What a recovery keeps stays until it ends: a publisher whose broker
refused a change, or whose session was replaced, leaves it to the one
which goes on, which may start from further back. Past 10000 changes or
32 MiB, the recovery keeps none and searches for them from where it has
got to, as below. So does a recovery whose round, holding the local
changes back, does not reach the last local change with what it kept - a
change which should have been kept is not there - rather than go round
for ever under the lock every local write waits for.
- A kept change is published as not assured, as the changes a recovery
rebuilds are. Nothing orders it with the wait of its operation for the
acknowledgement: an operation which still waits gets none and times out,
as it does when a round publishes the change later.
`LDAPUpdateMsg.setNotAssured()` drops the encoding the message cached.
- The announcement that the replica goes offline is given up on while a
recovery runs, as one a change in flight holds back is: it carries the
newest CSN of this replica, and the replication server would take the
replica for offline from there - its change number index past the
changes which reach it afterwards.
- **The search.** A round publishes up to where the search of the
recovery ends, or - for a recovery which kept too many changes - up to
the last local change read **before** its search, or up to the end of
its window if that comes first, and the next round starts from there. A
round never goes past that bound. A recovery which searches for the
local changes ends as before: the rounds after one which caught up hold
the local changes back until they reach the last one.
- The end of a window carried the sequence number `0xffffffff`: -1 in a
CSN, and the greatest in the text the index of `ds-sync-hist` compares,
so a change in the last millisecond of a window was in neither that
window nor the next one. The end carries `Integer.MAX_VALUE`, which
reads the same in both, and the search listener takes a change numbered
exactly the end, as the search filter does.
- **One session per recovery run (#1199).**
`ReplicationBroker.publishRecovery(msg, session)` takes the session the
recovery was started for (`getSession()`, read as the session starts),
and refuses the message for any other. It returns `false` when the
broker refuses the message: it reaches no replication server, its
session was replaced, or it was stopped. The round then fails, and
nothing it had not published is lost: the deletes kept aside and the
local changes the recovery keeps stay where they are. A refusal logs a
debug line, not `ERR_CANNOT_RECOVER_CHANGES`.
- A session which starts while the missing change publisher runs hands
it its start - the newest change of this replica its replication server
holds - and its session, and the publisher goes back there and publishes
on that session. The start of a recovery, its end, the hand-over and the
exit of the publisher are decided under one lock (`recoveryLock`), which
is taken inside the connect phase lock of the broker and inside the lock
of the pending changes, and under which neither is taken: a recovery is
not ended while a start is handed over, and the publisher does not leave
before it has run from it.
- **What the broker refuses during a recovery (#1220).** While a
recovery is required, the broker refuses the `UpdateMsg`s only: the
changes of this replica, and the announcement that it goes offline. The
acknowledgements, the messages of a total update, the monitoring
requests and the reset of the generation ID go through.
- The next session ends a recovery a refused publisher left even when
its replication server misses no change of this replica:
`sessionInitiated()` then starts a missing change publisher all the
same, or hands its start to the one which runs, which finds nothing to
publish, or the changes done since, and ends the recovery under the lock
of the pending changes. The recovery is not ended in
`sessionInitiated()` itself, which runs under the connect phase lock of
the broker: a change being pushed waits for that lock holding the lock
of the pending changes.
- A round reads the result of its search before it picks anything: a
search which did not go through publishes nothing. A search which finds
no base entry went through - the suffix is gone, and the deletes kept
aside are all there is - if the backend of the domain is there: a base
DN with no backend, which an import has taken away, answers the same,
and that search found nothing of what is there.
- A recovery whose search failed, or which threw, does not end: the
publisher logs `ERR_MISSING_CHANGES_NOT_PUBLISHED`, waits with the
session up (one second per failure in a row, up to ten), lets go of the
recovery, and asks for a session restart, which the state checkpointer
runs. The recovery of the next session publishes the changes; a session
which starts during the wait hands its start over, which ends the wait,
and the publisher runs from it with no restart. A publisher which throws
an error - which no catch of an exception takes - has the session
started again as well. Ending the recovery instead would let the next
local change out over the ones it did not publish, which the replication
server would then never take.
- A recovery publishes nothing while a total update replaces the data of
this replica: a round refuses, and so does each publish of a round in
flight, as what it would publish was done on the data the import
replaces. The session the import starts again recovers from the state
the new data holds.
- **The deletes kept aside (#1200, #1221).** What the search of a round
finds is the round's own, and the list of deletes kept aside holds
deletes only. The deletes a round publishes stay in the list, and so do
the ones at or before where it starts, and the list is no longer cleared
when a recovery ends: the replication server of a later session may miss
them, and they have nothing else to be published from. The cap of 10000
drops the oldest first; one which no replication server has taken logs
`ERR_UNPUBLISHED_DELETE_DROPPED` with its DN and CSN.
- A local delete which did not go through - its conflicts were resolved,
and then the access control, a plugin or the backend refused it - is
taken out of the deletes kept aside, before the pending changes let the
later changes out: a recovery published it as a delete which was done,
and the replicas it reached deleted an entry this one keeps.
- Once the ServerState is read from the data again - as the domain is
enabled, or after a total update into it - the deletes kept aside and
the local changes a recovery kept which the data does not reflect are
dropped: the data was restored or imported since. What comes after the
newest change of this replica that state holds is not in the data, as
the state covers the historical information of the entries - except for
a delete, which leaves none, and which the state misses too when it
could not be saved: a delete of the base entry takes the saved state
with it. A delete is dropped only if its entry is in the data.
- The state read back covers the historical information whether or not
the last save wrote it: as it is loaded, `PersistentServerState` brings
it up to the changes of this replica the entries hold past the newest
one it read. A state which holds none of them is left as it is - that
search would go through every entry the replica ever changed - and such
a state comes back once the base entry, which holds the saved state, was
deleted and added back, and no save wrote into it since: the add of the
base entry was then dropped, and the recovery published the deletes of
the suffix alone. `loadDataState()` brings such a state up from the
first change the recovery kept. A domain enabled back on its own data
loses none of the changes a recovery kept, unless that search does not
go through, which `ERR_CANNOT_RECOVER_CHANGES` tells.
- **The CSN generator does not go back (#1223).**
`CSNGenerator.adjust(null)` - `loadDataState()` passes the newest CSN of
this replica in the state it reads, which is `null` when that state
holds none - set the generator to the clock and its sequence number to
0. The session which starts next takes it forward to what its
replication server holds only, and the changes that replication server
misses are the ones a recovery publishes: the next local change was
numbered as one of them, or before, and as a change of another replica
it had followed, by as much as that replica's clock runs ahead. An
adjust with no CSN leaves the generator where it is.
- The search listener keeps to the changes of this replica, as the
search was meant to. The changes of the other replicas found in the same
entries were handed over too, as CSNs are ordered by time first; a
replication server which had one of them from here first would drop the
older changes of that replica reaching it afterwards.
- Test hooks (`@VisibleForTesting`): `setMissingChangesSearchHook()`
hands the search of each round to a test, which makes a change while a
round searches, or fails the search;
`setMissingChangesPublisherRunHook()` runs after each run of the
publisher, before it takes a start handed over or leaves;
`setKeptChangesPublishedHook()` runs after a round has published the
local changes the recovery kept, before it ends the recovery;
`setMaxDeletesKeptAside()` and `PendingChanges.setMaxKeptChanges()`
lower the caps.

## Not fixed here

Found on the way and filed:

- #1198 - under load a directory server leaves its replication server
for lacking only the changes still on their way to it (the trigger of
#1195); fixed by #1205.
- #1202 - `ReplicationBroker.stop()` needs the lock of the pending
changes before it sets the flag a publish waiting for window credit
checks; the rounds which hold the local changes back hold that lock
longer.
- #1209 - a write to the replication server which blocks holds the lock
of the pending changes, and with it the local operations and the replays
of their entries.

## Tests

`HistoricalCsnOrderingTest`: the recovery publishes to a broker which
records what it is handed, or runs in the missing change publisher of a
real session, and the changes done after it are looked up in the
changelog of a real replication server - which, where local changes go
on during the recovery, has to hold them and nothing else. The tests of
the search rounds start the recovery after the changes they search for,
or run with a recovery which keeps no local change, as one which kept
too many; three of them run both ways.

- a change done while a round searches is published;
- a round which found nothing does not end the recovery past a change;
- a change done once the recovery has ended is published (the
hand-over);
- only the changes of this replica are published;
- a recovery ends while local changes go on, and loses none;
- a change in the last millisecond of a search window is published;
- a recovery ends once it holds the local changes back, however far they
ran ahead of one window;
- the changes the broker refused are left to the next recovery;
- a session which needs no recovery ends the one a refused publisher
left;
- the broker takes the changes again under the lock they are pushed
under;
- a stopped broker refuses a recovery message;
- a round whose search failed publishes nothing;
- the deletes a round threw before publishing are left to the next
recovery;
- a recovery whose search failed, and one which threw, have the session
started again, and the next session publishes every change, in order,
the one done during the failed round included;
- a recovery which threw an error has the session started again;
- a recovery which keeps failing waits longer before each restart, and
forgets its failures once it succeeds;
- a session which starts while a failed recovery waits is recovered at
once;
- the deletes of a suffix removed during a recovery are published;
- a search which finds no backend fails its round;
- the deletes kept aside outlive a session to a replication server which
holds no change of this replica;
- a recovery left on a replica with no change of its own ends;
- a change numbered exactly the end of a search window is published;
- a change whose values a later one replaced is published;
- no search runs while the local changes are held back;
- a recovery which kept too many changes searches for them and loses
none;
- a recovery which keeps the local changes ends while they go on and
loses none;
- a recovery whose kept changes miss one searches for it;
- the local changes a recovery kept outlive a disable, one whose save
did not write the state too, and go with the data they were done on;
- a suffix deleted and added back during a recovery outlives a state
which could not be saved into the base entry added back;
- the deletes kept aside go with the data they were done on;
- a broker refuses a recovery message for a session it replaced;
- a broker with a recovery to run refuses the changes only;
- a session which starts while a recovery runs gets every change its
replication server misses;
- a session which starts as the publisher leaves gets a recovery of its
own;
- a recovery a session which started meanwhile needs is not ended;
- a session which starts while a round holds the local changes back to
publish the ones the recovery kept gets them all - it comes up as the
broker reconnects, which takes no lock of the pending changes;
- the changes a round finds take no room among the deletes kept aside;
- a delete no replication server took is not dropped without a trace;
- a delete which failed is not published by a recovery;
- the deletes of a suffix outlive a state which could not be saved, the
ones a recovery kept as well;
- a kept change is published as not assured;
- the replica is not announced offline while a recovery has changes to
publish;
- a recovery publishes nothing while a total update replaces the data;
- a replica which reads back a state with no change of its own numbers
its next change after its last one, and `CSNGeneratorTest`: an adjust
with no CSN numbers before neither the CSNs the generator generated nor
the ones it received.

On f06769eda7, with the test hooks only, 17 of the 25 new tests of
#1195-#1221 are red, each on the part of the change it covers, and two
more need a method that head does not have. Six are green there: the
fallback past the cap, the load on a recovery which keeps the local
changes, the kept changes dropped on a restore, the deletes of a suffix
which outlive a state not saved, a kept change published as not assured
- parts that head had no way to get wrong, as it kept nothing and
dropped no delete - and the wait of round 3. 33 mutants each undo one
part of the change, and all of them turn their tests red - two by going
round for ever, until the test JVM runs out of heap. The three tests of
#1223 are red with the generator of master, which f06769eda7 has: the
add of the base entry gets the CSN of the first delete. Three more
mutants come with round 5, each red on its test: without the bring-up of
a state read back with no change of this replica (f7fcf27bb9 for that
test: the replication server gets the deletes of the suffix only),
without the check of a start handed over before a held round keeps no
more changes, and - in `PersistentServerState` - without the bring-up of
a state read back to the changes past its newest one. The class: 54
tests, all green; `CSNGeneratorTest`: 6, all green.
`org/opends/server/replication/**`: not run again locally for this
round: the run was cut short as the machine ran out of memory, after
other test JVMs on it had taken the ports of its servers; the CI runs
it.