From 5154f34503981b2f96755b1b6da9b425bb20ca7f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 05 Nov 2026 14:32:12 +0000
Subject: [PATCH] [#1269] Preserve replay order across duplicate replication deliveries (#1270)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 186 +++++++++++++++++++++++++++++++++++-----------
1 files changed, 142 insertions(+), 44 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 085d147..5e801bc 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
@@ -74,10 +74,10 @@
@GuardedBy("dependentChangesLock")
private final SortedSet<PendingChange> dependentChanges = new TreeSet<>();
/**
- * {@code activeAndDependentChanges} also contains changes discovered to be dependent
- * on currently in progress changes.
+ * Uncommitted changes, registered before enqueue. A queued change whose hand-off
+ * fails must remain a predecessor: otherwise its DELETE can run before its ADD retry.
*/
- private final ConcurrentSkipListSet<PendingChange> activeAndDependentChanges = new ConcurrentSkipListSet<>();
+ private final ConcurrentSkipListSet<PendingChange> replayDependencies;
/**
* The change each replay thread is replaying right now, read by the give-back on the way
@@ -106,7 +106,7 @@
* races nothing for it. {@link #clear()} is the one writer of every entry, and it holds
* both locks.
*/
- private final ConcurrentMap<Thread, CSN> changeBeingReplayed = new ConcurrentHashMap<>();
+ private final ConcurrentMap<Thread, CSN> changeBeingReplayed;
private final ReentrantReadWriteLock pendingChangesLock = new ReentrantReadWriteLock(true);
private final ReentrantReadWriteLock.ReadLock pendingChangesReadLock = pendingChangesLock.readLock();
@@ -115,6 +115,16 @@
/** The ServerState that will be updated when LDAPUpdateMsg are fully replayed. */
private final ServerState state;
+ private final int localServerId;
+ /** Own-origin replay progress; local publication must not cover restore catch-up. */
+ @GuardedBy("pendingChangesLock")
+ private CSN localReplayCSN;
+
+ /** Whether a delivery needs replay or how its duplicate should be completed. */
+ enum Admission
+ {
+ ACCEPTED, QUEUED_DUPLICATE, DUPLICATE
+ }
/**
* How many of the changes listed here have a replay failure recorded against them.
@@ -137,7 +147,43 @@
*/
public RemotePendingChanges(ServerState state)
{
+ this(state, -1);
+ }
+
+ RemotePendingChanges(ServerState state, int localServerId)
+ {
+ this(state, localServerId, new ConcurrentHashMap<Thread, CSN>());
+ }
+
+ RemotePendingChanges(ServerState state, int localServerId,
+ ConcurrentMap<Thread, CSN> changeBeingReplayed)
+ {
+ this(state, localServerId, changeBeingReplayed, new ConcurrentSkipListSet<PendingChange>());
+ }
+
+ RemotePendingChanges(ServerState state, int localServerId,
+ ConcurrentMap<Thread, CSN> changeBeingReplayed,
+ ConcurrentSkipListSet<PendingChange> replayDependencies)
+ {
+ this.replayDependencies = replayDependencies;
this.state = state;
+ this.localServerId = localServerId;
+ this.changeBeingReplayed = changeBeingReplayed;
+ resetReplayState();
+ }
+
+ /** Reloads the own-origin watermark only when the backend state is loaded. */
+ void resetReplayState()
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ localReplayCSN = state.getCSN(localServerId);
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
}
/**
@@ -169,7 +215,23 @@
*/
public int changesInProgressSize()
{
- return activeAndDependentChanges.size();
+ pendingChangesReadLock.lock();
+ try
+ {
+ int count = 0;
+ for (PendingChange change : replayDependencies)
+ {
+ if (change.hasStartedReplay())
+ {
+ count++;
+ }
+ }
+ return count;
+ }
+ finally
+ {
+ pendingChangesReadLock.unlock();
+ }
}
/**
@@ -194,47 +256,82 @@
* 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).
+ * Changes covered by the applied ServerState are rejected even after they have left
+ * the pending list. For this replica's own origin, only the state loaded from the backend
+ * and subsequent replays count: a newer local publication does not prove that an older
+ * change sent back after restore is in the data. A listed change is also left alone while its delivery is queued,
+ * owned by a replay thread, or committed and waiting for earlier changes. Replacing a
+ * queued delivery would move it behind newer changes: a DELETE could then be replayed
+ * before its ADD, leaving the entry present once the ADD is replayed (issue #1269).
* <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.
+ * Only a delivery which was released after failed or abandoned replay, or cancelled
+ * before being enqueued, may be replaced. It is deliberately left out of the
+ * ServerState so the replication server sends it again over the restarted session
+ * (issue #889). Neither an owner nor a queued delivery remains to finish it.
* <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 and is owned by a replay thread.
+ * @return {@code false} if the change is already applied or has a delivery which is
+ * still queued, owned, or committed
*/
public boolean putRemoteUpdate(LDAPUpdateMsg update)
{
+ return putRemoteUpdate(new UpdateToReplay(update, null)) == Admission.ACCEPTED;
+ }
+
+ Admission putRemoteUpdate(UpdateToReplay delivery)
+ {
+ final LDAPUpdateMsg update = delivery.getUpdateMessage();
pendingChangesWriteLock.lock();
try
{
final CSN csn = update.getCSN();
+ if (csn.getServerId() == localServerId
+ ? localReplayCSN != null && localReplayCSN.compareTo(csn) >= 0 : state.cover(csn))
+ {
+ return Admission.DUPLICATE;
+ }
final PendingChange listed = pendingChanges.get(csn);
if (listed == null)
{
- pendingChanges.put(csn, new PendingChange(csn, null, update));
- return true;
+ final PendingChange change = new PendingChange(csn, null, update);
+ change.setDelivery(delivery);
+ pendingChanges.put(csn, change);
+ try
+ {
+ replayDependencies.add(change);
+ }
+ catch (RuntimeException | Error e)
+ {
+ // No other admission sees this partial registration while the write lock is held.
+ // Request recovery before rollback: removing a skip-list node can also fail.
+ delivery.cancel();
+ pendingChanges.remove(csn);
+ replayDependencies.remove(change);
+ throw e;
+ }
+ return Admission.ACCEPTED;
}
if (listed.isCommitted() || listed.isOwned())
{
- return false;
+ return Admission.DUPLICATE;
}
- listed.setMsg(update);
- return true;
+ if (listed.isQueuedForReplay())
+ {
+ // Keep the queue position, but complete for the latest delivery's assured mode.
+ // This changes in-memory ack metadata only; the received message's cached wire
+ // encoding is stale and must not be reused to publish this message.
+ final LDAPUpdateMsg accepted = listed.getLDAPUpdateMsg();
+ accepted.setAssured(update.isAssured());
+ accepted.setAssuredMode(update.getAssuredMode());
+ accepted.setSafeDataLevel(update.getSafeDataLevel());
+ return Admission.QUEUED_DUPLICATE;
+ }
+ listed.setDelivery(delivery);
+ return Admission.ACCEPTED;
}
finally
{
@@ -278,7 +375,7 @@
curChange.setCommitted(true);
curChange.setOwner(null);
changeBeingReplayed.remove(Thread.currentThread(), csn);
- activeAndDependentChanges.remove(curChange);
+ replayDependencies.remove(curChange);
final Iterator<PendingChange> it = pendingChanges.values().iterator();
while (it.hasNext())
@@ -291,6 +388,10 @@
if (pendingChange.getMsg().contributesToDomainState())
{
state.update(pendingChange.getCSN());
+ if (pendingChange.getCSN().getServerId() == localServerId)
+ {
+ localReplayCSN = pendingChange.getCSN();
+ }
}
if (pendingChange.getReplayFailures() > 0)
{
@@ -504,7 +605,7 @@
{
pendingChanges.clear();
dependentChanges.clear();
- activeAndDependentChanges.clear();
+ replayDependencies.clear();
changeBeingReplayed.clear();
failingChanges = 0;
}
@@ -520,12 +621,8 @@
*
* @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.
+ * @return {@code false} if the change is already committed, the domain has forgotten
+ * it, the delivery was cancelled, or this is a stale delivery which a retry has replaced
*/
public boolean markInProgress(LDAPUpdateMsg msg)
{
@@ -533,18 +630,19 @@
try
{
final PendingChange change = pendingChanges.get(msg.getCSN());
- if (change == null || change.isCommitted() || change.getLDAPUpdateMsg() != msg)
+ if (change == null || change.isCommitted() || change.isDeliveryCancelled()
+ || change.getLDAPUpdateMsg() != msg)
{
return false;
}
+ // This delivery has left the queue. If an allocation below fails before it gets
+ // an owner, a later delivery must be able to take it over.
+ change.markDequeued();
/*
- * Listed as being replayed before it is owned, and not the other way round: the
- * caller enters the replay - where the give-back on the way out lives - once this
- * returns, so nothing which allocates must run between the owner being stamped and
- * that. A change listed here without an owner is the state a failed replay leaves
- * behind, and the dependency checks which read this set do not read the owner.
+ * The dependency was registered before enqueue, so a failed claim still holds back
+ * its successors. Allocate the thread index before stamping the owner: the caller
+ * reaches replay's give-back only once this method returns.
*/
- activeAndDependentChanges.add(change);
changeBeingReplayed.put(Thread.currentThread(), change.getCSN());
change.setOwner(Thread.currentThread());
return true;
@@ -836,7 +934,7 @@
boolean hasDependencies = false;
final DN targetDN = op.getEntryDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
@@ -910,7 +1008,7 @@
boolean hasDependencies = false;
final DN targetDN = change.getLDAPUpdateMsg().getDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
@@ -977,7 +1075,7 @@
boolean hasDependencies = false;
final DN targetDN = change.getLDAPUpdateMsg().getDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
@@ -1052,7 +1150,7 @@
boolean hasDependencies = false;
final DN targetDN = op.getEntryDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
--
Gitblit v1.10.0