From 92d88ca699cd8090a26b92cbe46789d2b848195f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 07 Sep 2026 09:30:36 +0000
Subject: [PATCH] [#889] Keep a change the replay could not apply out of the ServerState (#892)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 263 +++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 256 insertions(+), 7 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
index 06d3797..4eef2d8 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -13,6 +13,7 @@
*
* Copyright 2007-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -81,6 +82,19 @@
private final ServerState state;
/**
+ * How many of the changes listed here have a replay failure recorded against them.
+ * <p>
+ * A change is counted from its first failed replay until it leaves this map, and it is
+ * what tells a change which can not be applied from a backend which is serving again:
+ * an outage fails everything in flight, while a change which can never be applied here
+ * fails alone, among changes which replay perfectly well. The session restart backoff
+ * reads it, so that the successful replays which surround a failing change do not keep
+ * resetting the wait this domain has reached on it (issue #889).
+ */
+ @GuardedBy("pendingChangesLock")
+ private int failingChanges;
+
+ /**
* Creates a new RemotePendingChanges using the provided ServerState.
*
* @param state The ServerState that will be updated when LDAPUpdateMsg
@@ -111,6 +125,10 @@
/**
* Returns the number of changes actively being replayed.
+ * <p>
+ * A change whose replay failed counts here until it is applied or given up on: it stays
+ * a dependency of the changes which follow it, since a change which is not in the data
+ * is exactly what they must wait for, and the delivery which comes next takes it over.
*
* @return the number of changes actively being replayed.
*/
@@ -140,19 +158,48 @@
/**
* Add a new LDAPUpdateMsg that was received from the replication server
* to the pendingList.
+ * <p>
+ * A change which is already listed and which a replay thread owns - it is being
+ * replayed, it waits for the change it depends on, or it has been replayed and waits
+ * for the changes before it - is left alone: that copy knows what this one does not,
+ * and replaying it again is exactly what the duplicate check is there to prevent
+ * (OPENDJ-1115).
+ * <p>
+ * A change which is listed but which no replay thread owns is taken over by this
+ * delivery: it is the one to replay, and the copy which may still wait in the replay
+ * queue is dropped by {@link #markInProgress(LDAPUpdateMsg)}. That is a change whose
+ * replay failed - it is deliberately left out of the ServerState, so the replication
+ * server sends it again over the session which was restarted (issue #889) - and it is
+ * also, harmlessly, a change which was listed a moment ago by a delivery no replay
+ * thread has picked up yet: the two deliveries carry the same change, and the last one
+ * listed is the one replayed.
+ * <p>
+ * The failures the change went through are kept, whichever delivery replays it: they
+ * are what has this replica eventually give up on a change it can not apply.
*
* @param update The LDAPUpdateMsg that was received from the replication
* server and that will be added to the pending list.
* @return {@code false} if the update was already registered in the pending
- * changes.
+ * changes and is owned by a replay thread.
*/
public boolean putRemoteUpdate(LDAPUpdateMsg update)
{
pendingChangesWriteLock.lock();
try
{
- CSN csn = update.getCSN();
- return pendingChanges.put(csn, new PendingChange(csn, null, update)) == null;
+ final CSN csn = update.getCSN();
+ final PendingChange listed = pendingChanges.get(csn);
+ if (listed == null)
+ {
+ pendingChanges.put(csn, new PendingChange(csn, null, update));
+ return true;
+ }
+ if (listed.isCommitted() || listed.isOwned())
+ {
+ return false;
+ }
+ listed.setMsg(update);
+ return true;
}
finally
{
@@ -177,6 +224,7 @@
throw new NoSuchElementException();
}
curChange.setCommitted(true);
+ curChange.setOwned(false);
activeAndDependentChanges.remove(curChange);
final Iterator<PendingChange> it = pendingChanges.values().iterator();
@@ -191,6 +239,11 @@
{
state.update(pendingChange.getCSN());
}
+ if (pendingChange.getReplayFailures() > 0)
+ {
+ // The change is leaving this map, so it is not one of the failing ones anymore.
+ failingChanges--;
+ }
it.remove();
}
}
@@ -200,18 +253,203 @@
}
}
- public void markInProgress(LDAPUpdateMsg msg)
+ /**
+ * Gives up the ownership a replay thread had on the change with the provided CSN,
+ * whose replay failed.
+ * <p>
+ * The change stays listed here and stays uncommitted: it is the barrier which keeps
+ * the ServerState - and every change which follows it - from moving past a change
+ * which is not in the data (issue #889), and it is what has the replication server
+ * send it again over the restarted session. Only the mark which says a replay thread
+ * owns it is dropped, so that {@link #putRemoteUpdate(LDAPUpdateMsg)} takes the next
+ * delivery instead of discarding it as a duplicate.
+ * <p>
+ * It also stays listed among the changes the newer ones are checked against: a change
+ * which is not in the data yet is exactly what the changes which follow it must depend
+ * on, whether or not a replay thread owns it right now.
+ * <p>
+ * The changes another replay thread is applying right now are left alone: they are
+ * about to commit, and forgetting them would have their {@code commit()} fail, the
+ * ServerState stay behind them and the replication server replay them a second time.
+ *
+ * @param csn the CSN of the change whose replay failed
+ */
+ public void replayFailed(CSN csn)
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ final PendingChange change = pendingChanges.get(csn);
+ if (change != null && !change.isCommitted())
+ {
+ change.setOwned(false);
+ }
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /** How long, and how many times, the replay of one change has been failing. */
+ static final class ReplayFailure
+ {
+ private final int attempts;
+ private final long failingForMs;
+
+ private ReplayFailure(int attempts, long failingForMs)
+ {
+ this.attempts = attempts;
+ this.failingForMs = failingForMs;
+ }
+
+ /**
+ * Returns how many deliveries of the change failed to replay it in a row. A delivery
+ * is attempted several times in place before it is counted here, so this is a count
+ * of deliveries rather than of attempts made on the backend.
+ *
+ * @return the number of failed deliveries, at least 1
+ */
+ int getAttempts()
+ {
+ return attempts;
+ }
+
+ /**
+ * Returns how long the replay of the change has been failing, that is the time
+ * between its first failure and the one which was just recorded.
+ *
+ * @return the duration in milliseconds, 0 for a first failure
+ */
+ long getFailingForMs()
+ {
+ return failingForMs;
+ }
+ }
+
+ /**
+ * Records that the replay of the change with the provided CSN failed once more.
+ * <p>
+ * The failures are kept on the change itself, which stays listed for as long as it has
+ * not been applied, so a change which keeps failing keeps its give-up budget across the
+ * deliveries which take over from one another, and a change which is replayed or given
+ * up on takes its failures away with it (issue #889).
+ *
+ * @param csn
+ * the CSN of the change whose replay failed
+ * @param nowMs
+ * when it failed, on a clock which only moves forward
+ * @return the failures of the change, or {@code null} when it is not listed as an
+ * uncommitted change anymore, which happens when the domain was disabled while
+ * it was being replayed: there is no change left here to give up on
+ */
+ public ReplayFailure recordReplayFailure(CSN csn, long nowMs)
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ final PendingChange change = pendingChanges.get(csn);
+ if (change == null || change.isCommitted())
+ {
+ return null;
+ }
+ if (change.getReplayFailures() == 0)
+ {
+ // Its first failure: this change joins the ones which are failing right now.
+ failingChanges++;
+ }
+ change.recordReplayFailure(nowMs);
+ return new ReplayFailure(change.getReplayFailures(), change.getReplayFailingForMs(nowMs));
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /**
+ * Returns whether the replay of any change listed here is failing right now.
+ * <p>
+ * A change is failing from its first failed replay until it leaves this map, whether
+ * it leaves it applied or given up on. The session restart backoff reads this: a
+ * change which was replayed only says that this backend is serving again when it is
+ * the last one which was failing, and the successful replays which surround a change
+ * this replica can not apply must not keep resetting the wait it has reached on it.
+ *
+ * @return {@code true} while at least one listed change has a failed replay recorded
+ * against it
+ */
+ boolean hasFailingChanges()
{
pendingChangesReadLock.lock();
try
{
- activeAndDependentChanges.add(pendingChanges.get(msg.getCSN()));
+ return failingChanges > 0;
}
finally
{
pendingChangesReadLock.unlock();
}
}
+
+ /**
+ * Forgets every change listed here, without updating the ServerState.
+ * <p>
+ * Called when the domain is disabled: its ServerState is saved and cleared from
+ * memory, and it is loaded again from the backend when the domain is enabled back, so
+ * the bookkeeping which goes with it must not outlive it. A change which stayed here
+ * would be discarded as a duplicate when the replication server sends it again, and
+ * nothing would ever replay it or record it in the ServerState.
+ */
+ public void clear()
+ {
+ pendingChangesWriteLock.lock();
+ dependentChangesLock.lock();
+ try
+ {
+ pendingChanges.clear();
+ dependentChanges.clear();
+ activeAndDependentChanges.clear();
+ failingChanges = 0;
+ }
+ finally
+ {
+ dependentChangesLock.unlock();
+ pendingChangesWriteLock.unlock();
+ }
+ }
+
+ /**
+ * Marks the change of the provided message as being replayed.
+ *
+ * @param msg
+ * the message whose change is being replayed
+ * @return {@code false} if this message is not the delivery which is listed as
+ * pending, which happens when the session was restarted after a failed replay
+ * while this message was still waiting in the replay queue: the replication
+ * server delivered the change again and that delivery took over from this one,
+ * or the domain was disabled and forgot the change, so this copy must not be
+ * replayed.
+ */
+ public boolean markInProgress(LDAPUpdateMsg msg)
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ final PendingChange change = pendingChanges.get(msg.getCSN());
+ if (change == null || change.isCommitted() || change.getLDAPUpdateMsg() != msg)
+ {
+ return false;
+ }
+ change.setOwned(true);
+ activeAndDependentChanges.add(change);
+ return true;
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
+ }
/**
* Get the first update in the list that have some dependencies cleared.
*
@@ -223,7 +461,7 @@
dependentChangesLock.lock();
try
{
- if (!dependentChanges.isEmpty())
+ if (!dependentChanges.isEmpty() && !pendingChanges.isEmpty())
{
PendingChange firstDependentChange = dependentChanges.first();
if (pendingChanges.firstKey().isNewerThanOrEqualTo(firstDependentChange.getCSN()))
@@ -248,14 +486,25 @@
*/
private void addDependency(PendingChange dependentChange)
{
+ pendingChangesReadLock.lock();
dependentChangesLock.lock();
try
{
- dependentChanges.add(dependentChange);
+ /*
+ * A change which is not listed as pending anymore - the domain was disabled while
+ * its dependencies were computed - must not be listed as dependent either:
+ * getNextUpdate() reads both and would hand out a change nothing owns. The
+ * replication server sends it again when the domain is enabled back.
+ */
+ if (pendingChanges.containsKey(dependentChange.getCSN()))
+ {
+ dependentChanges.add(dependentChange);
+ }
}
finally
{
dependentChangesLock.unlock();
+ pendingChangesReadLock.unlock();
}
}
--
Gitblit v1.10.0