[#889] Keep a change the replay could not apply out of the ServerState

[vharseko]

Fixes #889.

A change whose replay fails with anything other than NO_OPERATION, BUSY or UNAVAILABLE is recorded as replayed: the ServerState advances past it, the replication server never sends it again, and an assured (SAFE_READ) ack goes back to the originating master as if the change had been applied. The replica silently diverges while reporting itself fully caught up, with unresolved-naming-conflicts at 0 and a single line in the error log. This is storage-agnostic: JE, PersistIt and JDBC all reach it on any StorageRuntimeException.

What changed

The four solveNamingConflict() overloads no longer collapse two outcomes into one return true. They report a ConflictResolution (REPLAY_AGAIN / NOTHING_TO_DO / FAILED), so replay() can tell "the operation became a no-op after conflict resolution" from "the operation failed". The four copies of the ERR_ERROR_REPLAYING_OPERATION log move to the single place which decides what to do with the failure.

A failure of the server itself is no longer recorded as applied. UNAVAILABLE and the server-error-result-code - the code BackendImpl.createDirectoryException() puts on every StorageRuntimeException, 80 by default - are retried in place, and if they keep failing the change is deliberately left out of the ServerState. The replication server still owns it, so the domain restarts its session and the change is delivered and replayed again on a backend which has hopefully recovered. The codes conflict resolution knows how to solve (NO_SUCH_OBJECT, ENTRY_ALREADY_EXISTS, NOT_ALLOWED_ON_RDN, NOT_ALLOWED_ON_NONLEAF) are excluded from that test: server-error-result-code is configurable and is not validated as a result code, and it must never take a change away from solveNamingConflict().

The bookkeeping of the pending changes follows. RemotePendingChanges.clearUncommitted() forgets the changes which were not replayed, so processUpdate() does not discard them as duplicates when the replication server sends them again (the check of OPENDJ-1115). The ones which were replayed stay: they sit behind the failing change, so the ServerState does not cover them yet and they are sent again too - forgetting them would have them applied and acked a second time. Two things follow from keeping them: putRemoteUpdate() no longer overwrites the copy which is listed with the one the new delivery came with (which lost the fact that it had been replayed), and a duplicate delivery no longer has the listener thread push its CSN to the ServerState - the copy which is listed owns the change and records it once it really has been replayed, while the ack and the window credit stay per delivery. A message left in the replay queue by the restarted session is dropped when a replay thread takes it out: markInProgress() only accepts the delivery which is listed as pending.

Every failed replay is now reported. replayErrorMsg is set on all failure paths, so the SAFE_READ ack carries the replay error instead of telling the originating master that the write is durable here. The new replayed-updates-failed monitor attribute counts the changes this replica gave up on - once each, like replayed-updates-ok counts the ones it applied.

A change which can never be applied here does not stop the replica for good. The failures are counted per CSN - a backend which is failing fails every change in flight, and a single counter would be reset by each of them in turn - and on the MAX_REPLAY_ATTEMPTS-th (3rd) failed attempt the change is skipped, but loudly: ERR_REPLAY_SKIPPING_CHANGE plus the new UnreplayedChange alert telling the administrator that this replica has diverged and must be reinitialized. The attempts in between are logged as WARN_REPLAY_RETRYING_CHANGE, and the session is left down for a moment before the change is asked for again, so that a backend which keeps failing is not asked for every change as fast as the replication server can send them.

What deliberately did not change

Failures which are not the server's fault - a schema or constraint violation, or the conflict-resolution loop of #798 - keep being skipped, as they always were: redelivering them cannot help, and stopping the domain on them would turn a one-entry divergence into an outage. They now carry the same honest ack, counter and alert, so the divergence is visible instead of silent. Restarting the session on those failures was tried first and is what UpdateOperationTest.infiniteReplayLoop and namingConflicts rightly rejected.

Tests

test	covers
RemotePendingChangesTest (new)	an uncommitted change holds back the ServerState; clearUncommitted() forgets the changes which were not replayed and lets the same CSN be accepted again, while the ones which were replayed stay and are not accepted again; only the delivery which is listed as pending is replayed
UpdateOperationTest.failedReplayIsNotRecordedAsReplayed (new)	the change is replayed again after a failed replay - impossible if the ServerState had advanced - the entry is untouched, and the replica eventually gives up, counting the change once and raising the UnreplayedChange alert
UpdateOperationTest.everyChangeWhichCanNotBeReplayedIsGivenUpOn (new)	two changes failing at once - what a backend outage looks like - are both given up on, which a counter kept for the last failed change only never reaches
UpdateOperationTest.transientReplayFailureIsRetriedAndTheChangeApplied (new)	an UNAVAILABLE backend which serves the operation again within the retry window has its change replayed exactly once, with no session restart and nothing counted as failed
AssuredReplicationPluginTest.testSafeReadModeReplyWithReplayError (new)	the SAFE_READ ack carries hasReplayError and failedServers=[1]

The AssuredReplicationPluginTest one closes a TODO which has been in the tree since upstream: "make the domain return an error: use a plugin? The resolution code does not generate any error so we need to find a way to have the replay not working to test this...". The short circuit has to be set at the pre-parse plugin point - LocalBackendDeleteOperation and friends skip the pre-operation plugins for synchronization operations.

UpdateOperationTest, NamingConflictTest, AssuredReplicationPluginTest, ModifyConflictTest, DependencyTest, IsolationTest, StateMachineTest and RemotePendingChangesTest: 82/82 green.

[maximthomas]

The diagnosis is right and the direction is right — a replay failure must not advance the ServerState. But in the
exact scenario this targets, a backend outage failing several in-flight changes, the new recovery path restarts the
session without bound and re-applies changes that were already applied. Two blockers, three majors below. Nothing
was executed; all of it is traced from source.

The MAX_REPLAY_ATTEMPTS bound is unreachable (blocker)

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java:2639

The attempt count lives in two scalar fields, not a per-CSN map:

lastFailedCSNAttempts = csn.equals(lastFailedCSN) ? lastFailedCSNAttempts + 1 : 1;
lastFailedCSN = csn;

A failure for a different CSN resets the count to 1, so attempts > MAX_REPLAY_ATTEMPTS (:2644) is only reached
when one CSN fails four times with nothing else failing in between. A backend outage fails every change in flight.
With two failing changes:

recover(c1)  lastFailedCSN=null -> attempts=1 -> disableService/clear/enableService
recover(c2)  c2 != c1           -> attempts=1 -> restart
   RS resumes from a ServerState covering neither, resends both
recover(c1)  c1 != c2           -> attempts=1 -> restart ...

The counter never leaves 1 and there is no sleep or backoff anywhere in :2623-2671. The escape hatch the PR
describes — "a change which can never be applied here does not stop the replica for good" — is dead code, and the
replica restarts its replication session indefinitely.

This holds single-threaded too; two CSNs alternating through one thread reset each other identically.
synchronized (replayFailureLock) at :2644 does not help: it makes the read-modify-write atomic, but one slot is
still one slot.

UpdateOperationTest.failedReplayIsNotRecordedAsReplayed misses it because it publishes exactly one DeleteMsg, so
the counter walks 1,2,3,4 and the skip fires as intended. A test with two concurrent failing CSNs would catch it.

Suggested: key the count on the CSN (a bounded map or small LRU), dropping entries on successful replay and on skip.

RemotePendingChanges.clear() re-arms OPENDJ-1115 (blocker)

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java:212

clear() empties pendingChanges unconditionally, but commit() flushes only the contiguous committed prefix —
it breaks at the first uncommitted entry. This fix deliberately leaves the failing change uncommitted, so every
later change that another replay thread applies successfully stays in the map as committed-but-unflushed, with its
CSN absent from the ServerState. clear() drops those too.

c1 fails, stays uncommitted
c2, c3 applied OK but sit behind c1  -> ServerState covers none of the three
clear()                              -> all three forgotten
enableService() -> RS resends c2, c3 -> putRemoteUpdate() now returns true (map is empty)
                                     -> c2, c3 replayed and acked a second time

That putRemoteUpdate() check at LDAPReplicationDomain.java:4425 is the OPENDJ-1115 guard, and its own comment
says surviving pendingChanges is exactly what makes session failover safe. Correctness now rests entirely on
conflict resolution being idempotent for a replayed Add/Modify/Delete — the reliance OPENDJ-1115 was filed to remove.
The pre-existing session restarts (disable() at :3304, the fractional reconnect at :702/:729) do not call
clear().

Only the uncommitted changes need forgetting. Keeping the committed ones is still correct: once the failing CSN
commits, the prefix flushes and the state advances over them. A selective removal gets the stated benefit without
reopening OPENDJ-1115.

clear() runs against a replay queue nobody drained (major)

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java:234

recoverFromReplayFailure does disableService(); clear(); enableService();. disableService() stops the broker
and joins the listener thread only — it never stops the replay threads, and updateToReplayQueue is a static
10 000-entry queue shared by every domain. Messages for this domain are still in it, pollable, while the map is
empty. Then:

// markInProgress, no null guard; activeAndDependentChanges is a ConcurrentSkipListSet
activeAndDependentChanges.add(pendingChanges.get(msg.getCSN()));

Two outcomes, depending on whether the redelivered copy has re-run putRemoteUpdate() yet:

• not yet — get() returns null, add(null) throws NPE into ReplayThread's catch-all. replay() never runs:
  no replay, no processUpdateDone(), no assured ack, so the originating master waits out its SAFE_READ
  timeout instead of getting the honest error ack this PR adds. One ERR_EXCEPTION_REPLAYING per queued change.
• already — the stale copy and the redelivered copy are both replayed: same CSN, two threads, one map entry. The
  loser hits commit() -> NoSuchElementException -> ERR_OPERATION_NOT_FOUND_IN_PENDING and returns early, so it
  is applied but never recorded.

switchQueueLock does not serialise this — it is released before domain.replay(), and clear() is reached from
inside replay(). A null guard alone would turn the first case into a silent drop; the queue needs draining for
this domain before clear().

The shutdown guard reads the wrong flag (major)

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java:2630

private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean shutdown) {
  if (shutdown.get() || disabled) return true;

That shutdown is the parameter threaded down from replay(LDAPUpdateMsg, AtomicBoolean) — the ReplayThread's
flag — and it shadows the domain field at :352. disabled is set only inside disable(), and shutdown() (:2242)
calls disableService() directly rather than disable(), so it stays false. Neither guard sees domain shutdown.

MultimasterReplication.finalizeSynchronizationProvider() runs domain.shutdown() for every domain at :572 and
stopReplayThreads() only at :576, so the thread flag is false throughout — and that is precisely when replays fail,
because the backend is going offline and returns a code isServerFailure() now routes here. Both guards pass,
enableService() runs, ReplicationBroker.start() clears its own shutdown flag and reconnects: a broker and a fresh
listener thread come back up on a domain whose alert generator, flush thread and RSUpdater are already gone.

Before this PR no replay failure ever called enableService(). Checking the domain's own state (and renaming the
parameter — the shadowing is what hid this) is enough.

isServerFailure() keys off ResultCode.OTHER (major)

opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java:2583

return result == ResultCode.UNAVAILABLE
    || result == getServerContext().getCoreConfigManager().getServerErrorResultCode();

server-error-result-code defaults to ResultCode.OTHER — not a storage-specific code, but what the server returns
whenever nothing more specific applies. ReferentialIntegrityPlugin.java:1155 does
catch (Exception de) { stopProcessing(ResultCode.OTHER, ...) }, reached from doPreOperation(add):1054 and
doPreOperation(modify):985, neither guarded by isSynchronizationOperation(); SaltedSHA1PasswordStorageScheme
and SaltedSHA512PasswordStorageScheme do the same at :447/:451.

So a deterministic referential-integrity failure now costs, per delivery, ten rounds of Thread.sleep(50) at :2375
plus a full session restart, three times over — where before it fell into solveNamingConflict, was logged, and was
stepped over in one pass. It is also a realistic input for the first blocker: one misconfiguration fails many
changes at once.

Separately, the branch sits ahead of conflict resolution (:2375 vs :2386) and the knob is unvalidated —
GlobalConfiguration.xml:181 constrains it only with <adm:integer lower-limit="0"/> and CoreConfigManager does a
bare ResultCode.valueOf(). Setting it to 32, 66, 67 or 68 makes those codes bypass solveNamingConflict entirely.

Detecting the storage failure at its source — a marker exception type, or the exception cause — would be sturdier
than matching a configurable result code. Failing that, move the branch after conflict resolution.

Nits

• replayed-updates-failed counts attempts, not changes: the three numFailedReplayedUpdates.incrementAndGet()
  sites (:2441, :2467, :2512) are all inside replay(), re-entered on every redelivery, while replayed-updates-ok
  is incremented once per committed change (:2018). One unreplayable change contributes up to 4 — and unboundedly
  once the first blocker fires.
• Off-by-one in the skip: attempts > MAX_REPLAY_ATTEMPTS with attempts starting at 1 skips on the fourth
  delivery, and ERR_REPLAY_SKIPPING_CHANGE_308 ("after %d attempts") prints 4 while the constant and the PR
  description both say 3.
• The attempt counter is never reset: lastFailedCSN / lastFailedCSNAttempts are assigned only at :2639-2640
  — not on a successful replay, not after a skip. A CSN skipped once is skipped again with zero retries if
  redelivered; a change that fails, succeeds, then fails much later starts at attempt 2.
• The new alert type is undocumented: org.opends.server.replication.UnresolvedConflict is listed in
  opendj-doc-generated-ref/src/main/asciidoc/admin-guide/chap-monitoring.adoc:2179 and the docbkx copy at
  chap-monitoring.xml:980; UnreplayedChange appears in neither. enabled-alert-type is an allow-list — "if
  there are any values for this attribute, then only alerts with one of the specified types are allowed" — so an
  operator with a non-empty list gets no notification of a diverged replica and no documented string to add.
• A self-healing retry is logged at ERROR: ERR_REPLAY_RETRYING_CHANGE_307 fires via logger.error on every
  attempt of a path the code expects to recover from; the neighbouring retry messages are
  WARN_RETRYING_BIND_CHANGELOG_301 / NOTE_BOUND_CHANGELOG_AFTER_RETRY_302. Worth a WARN_ prefix — and worth
  mentioning in the description, which announces only ERR_REPLAY_SKIPPING_CHANGE.
• The advertised retry-in-place success is untested: both new tests keep the short circuit registered for the
  whole run, so all ten in-loop attempts fail and only the give-up path is observed — nothing shows that a transient
  failure clearing within the window is replayed exactly once and committed. Both inject ResultCode.OTHER, so the
  UNAVAILABLE half of isServerFailure() is unexercised, and the alert assertion is
  assertThat(DummyAlertHandler.getAlertCount()).isGreaterThan(initialAlerts) — any alert emitted during the three
  session restarts satisfies it, including one of a different type.

[vharseko]

Thanks - the review was traced from the source and it holds. Both blockers and all three majors are fixed in 8d3a0cc, together with every nit. Two of the fixes turned out to need more than what was suggested, and one of the majors is answered rather than applied; details below.

The MAX_REPLAY_ATTEMPTS bound (blocker) - fixed as suggested

The two scalar fields are gone. The count is kept per CSN in a ConcurrentSkipListMap<CSN, Integer>, dropped as soon as the change is replayed (synchronize()) or given up on, and bounded at 1000 entries with the oldest evicted - only failing changes are listed, so the bound is never reached in practice. The session is also left down for 1s * attempts before the change is asked for again, so a backend which keeps failing is not asked for every change as fast as the RS can send them.

UpdateOperationTest.everyChangeWhichCanNotBeReplayedIsGivenUpOn is the regression test you asked for: two CSNs failing at once, which is the case the single slot could never carry to the give up - each of them resets the count of the other.

clear() re-arming OPENDJ-1115 (blocker) - fixed, plus two things it uncovered

clearUncommitted() now removes only the changes which were not replayed, exactly as you suggested. Writing the unit test for it turned up two more holes on that path:

• putRemoteUpdate() overwrote the entry it found. pendingChanges.put(csn, new PendingChange(...)) == null reports the duplicate but has already replaced the committed-but-unflushed change with a fresh uncommitted one, so nothing would ever commit it and the ServerState would stay behind it for good. It is putIfAbsent() now.
• A duplicate delivery had the listener push its CSN to the ServerState. processUpdate() returned true for a duplicate, and ReplicationDomain's listener does processUpdateDone(msg, null); state.update(msg.getCSN()) for everything which returns true (ReplicationDomain.java:3246-3256). With the committed changes now kept, the resent copy of c2 would push c2 into the ServerState over c1, which is still failing - the very bug this PR is about. It returns false now and calls processUpdateDone(msg, null) itself, so the ack and the window credit stay per delivery while the recording of the change stays with the copy which is listed.

clear() against an undrained replay queue (major) - fixed, by a different means

I first did what you suggested - MultimasterReplication.dropQueuedUpdates(domain) holding switchQueueLock. It works, but it is unusably slow: instrumenting the recovery showed disable=1-3ms, enable=64-329ms, drain+clear=19527ms / 61786ms / 83782ms. switchQueueLock is non-fair and ten replay threads re-take it in a tryLock(1s) / poll(1s) loop, so the draining thread starves for tens of seconds. That is what made everyChangeWhichCanNotBeReplayedIsGivenUpOn time out at 120s.

The queue is not drained at all now. markInProgress() returns whether this message is the delivery which is listed as pending, and ReplayThread skips the message when it is not:

final PendingChange change = pendingChanges.get(msg.getCSN());
if (change == null || change.getLDAPUpdateMsg() != msg) { return false; }

change == null is the first outcome you described - the NPE - and the identity check is the second one: the stale copy and the redelivered copy are never both replayed, because only the one the domain is listing gets through. No lock, and a stale message costs one poll().

The shutdown guard (major) - fixed, with one correction

The parameter is replayThreadShutdown now and the guard reads the state of the domain (shutdown.get() || disabled), re-read after the wait and before enableService(). One correction to the reasoning: at server shutdown the synchronization providers are finalized before the backends (DirectoryServer.java:4170-4172), so "the backend is going offline" is not the usual trigger - the window is real for any replay failing in flight, and for disable()/delete() from a configuration change.

isListenerShuttingDown() looked like a good extra guard and is not: during a recovery started by another replay thread the listener is already gone, so a change failing at that moment took the "domain is going away" exit and never counted its attempt. The count now happens before the guard.

isServerFailure() keying off ResultCode.OTHER (major) - partly

The premise is right: server-error-result-code defaults to 80 = OTHER, CoreConfigManager does a bare valueOf() on an unvalidated integer, and the code is used all over the server for "internal error".

The two examples do not reach a replay, though. Pre-operation plugins are not invoked for synchronization operations - LocalBackendModifyOperation.java:323-333, LocalBackendAddOperation.java:436-439, LocalBackendDeleteOperation.java:262-265 - and ReferentialIntegrityPlugin registers only pre-operation, post-operation and subordinate types (isConfigurationAcceptable() at :255-280), no pre-parse, so its stopProcessing(ResultCode.OTHER, ...) cannot fire for a replayed change. SaltedSHA* :447/:451 is "the JVM has no SHA-1 MessageDigest", not a per-change failure. So the "ten Thread.sleep(50) plus three session restarts for a deterministic referential-integrity failure" does not happen.

What is real is the misconfiguration you point at second, and that is now closed: the codes conflict resolution owns (NO_SUCH_OBJECT, ENTRY_ALREADY_EXISTS, NOT_ALLOWED_ON_RDN, NOT_ALLOWED_ON_NONLEAF) are excluded from isServerFailure(), so setting the knob to 32/66/67/68 can no longer take a change away from solveNamingConflict().

Detecting the storage failure at its source is the better design and I did not do it: Operation carries neither the exception nor a marker, so it means plumbing a flag from BackendImpl.createDirectoryException() through setResponseData() and the Operation interface (and OperationWrapper, and every implementor) - a change to a core API well outside this fix. Worth its own issue if you want it.

Nits

All applied.

• replayed-updates-failed counts changes, not attempts: the increment moved into skipUnreplayableChange(), and it only counts when the change was really recorded as skipped (updateError() reports whether it committed), so a change skipped while another thread was clearing the pending list is counted when it is skipped for good rather than twice.
• Off-by-one: attempts >= MAX_REPLAY_ATTEMPTS, so the change is skipped on the 3rd attempt and the message prints 3.
• The attempt count is dropped on a successful replay and on a skip.
• org.opends.server.replication.UnreplayedChange is documented in chap-monitoring.adoc and the docbkx chap-monitoring.xml, next to UnresolvedConflict.
• ERR_REPLAY_RETRYING_CHANGE_307 is WARN_REPLAY_RETRYING_CHANGE_307 and is logged with logger.warn.
• Tests: transientReplayFailureIsRetriedAndTheChangeApplied covers the UNAVAILABLE half and the retry in place - the backend serves the operation again after three attempts and the change is applied exactly once, with nothing counted as failed and no alert; failedReplayIsNotRecordedAsReplayed now counts the redeliveries through the short circuit rather than through the monitor attribute, asserts replayed-updates-failed is +1 exactly, and asserts the alert count of that type.

One thing not in the review: the alert. Whatever makes one change unreplayable makes every change in flight unreplayable, and the alert was raised per skipped change. It is now raised at most once a minute per domain, while every skipped change is still logged.

Tests

UpdateOperationTest 11/11, AssuredReplicationPluginTest 14/14, RemotePendingChangesTest 6/6, NamingConflictTest 6/6, ModifyConflictTest 36/36, DependencyTest 3/3, IsolationTest 1/1, StateMachineTest 5/5 - 82/82 green.