[#1269] Preserve replay order across duplicate replication deliveries (#1270)

Valery Kharseko

98 mins ago 5154f34503981b2f96755b1b6da9b425bb20ca7f

[#1269] Preserve replay order across duplicate replication deliveries (#1270)

After a directory server reconnects, duplicate delivery could replay an
already applied ADD after its DELETE, or replace a queued ADD and move
it behind its DELETE. Either path could leave a deleted entry present on
the replica.

Preserve the accepted queued delivery and reject changes already covered
by replay progress. Remote-origin duplicates use the applied
ServerState. Own-origin catch-up uses a separate watermark initialized
from backend state and advanced by committed replay, so a newer local
publication cannot hide an older own-origin change already delivered
after restore. Reloading backend state resets that watermark.

Keep uncommitted changes in the dependency index from admission through
commit, including failed queue hand-offs. This prevents a DELETE from
overtaking the retry of a cancelled ADD. Each delivery has its own
cancellation token. Enqueue, dequeue and claim failures request
redelivery without reacquiring the pending lock. The consumer retains
the queue head before removing it, covering a failure inside `poll()`
after unlinking as well as a failure before claim. Failed dependency
registration cancels and requests recovery before attempting rollback,
so another allocation failure during cleanup cannot skip that request.

A normal listener shutdown cancels an unqueued delivery without
requesting an extra restart while that thread remains the registered
listener: a completed join already precedes the next session's
redelivery. Cancellation occurs before comparing listener identity. If
an interrupted join has detached the old listener, its late cancellation
requests a fresh restart, including when a newer session has already
refused the still-live delivery. Exceptional hand-off failures always
request recovery. Cancellation affects only its original delivery.

Queued duplicates increment the processed-delivery counter without
returning early window credit or a SAFE_READ ack. The accepted copy
retains its queue position, takes the latest redelivery's assured
settings, and completes after replay with its actual result. Applied,
committed and owned duplicates retain their existing completion
behavior. Assured-field updates affect in-memory ACK metadata; the code
documents that any cached wire encoding remains stale.

Add deterministic backend regressions for both resurrection paths and
cancellation/retry ordering, plus coverage for completion counts,
assured success/error, changing assured flags and modes, own-origin
catch-up, backend state reload, interrupted claims, completed and
interrupted listener joins, and late cancellation. Constructor-supplied
thread-index and dependency-set state inject claim, registration and
rollback failures without replacing private final fields. Retry tests
hold recovery to prevent the checkpointer from restarting the session
during their controlled retry. A real replay consumer test verifies
notification from an empty queue, prompt draining of its backlog, and
committed replay before shutdown. Two naming-conflict fixtures use
different origins for changes deliberately replayed in reverse CSN
order, preserving their conflict-resolution coverage.

Fixes #1269.

## Validation

- Backend regressions reproduced both original resurrection paths.
Further regressions reproduced early duplicate completion, suppression
of own-origin catch-up, lost queue hand-offs, DELETE overtaking a
cancelled ADD's retry, an unnecessary restart after a completed listener
join, and missed cancellation after a repeated OOME during rollback.
- Java 21.0.12.1: the complete replication/changelog selection passed
**4,155 tests**, with zero failures, errors or skips. Reactor build and
Javadoc/doclint passed.
- Focused compilation and execution passed **54/54 tests on Java 11 and
Java 21**, with zero failures or skips.
- The initial **seven targeted mutations** were detected: queued
dependency registration, claim cancellation, cancellation restart, early
completion, assured metadata, own-origin filtering and backend reload.
- A further **eight targeted mutations** were detected: assured-mode
copy, producer notification, unconditional consumer wait, cancellation
after rollback, missing registration rollback, unnecessary
normal-shutdown restart, missing detached-listener restart, and missing
exceptional-handoff restart.
- Independent read-only review of the latest change found a
consumer-test shutdown race; it was corrected before the final full run.
- After the full run, the branch was rebased without conflicts onto the
documentation-only #1268 update to `origin/master`. All six verified
source hashes remained unchanged; the full reactor result precedes this
base-only update. `git diff --check` passed after rebase.

```sh
mvn -o -Pprecommit verify -pl opendj-server-legacy -am \
  '-Dit.test=**/replication/**/*Test*,ChangelogBackendTestCase' \
  -Dfailsafe.failIfNoSpecifiedTests=false -DfailIfNoTests=false
```

## Limits

- Fault injection covers the failure boundaries without exhausting the
JVM heap. The listener-join test explicitly interrupts a controlled stop
thread using the real `disableService()`/`join()` path and a supplied
listener-slot update; it does not establish how an external production
interrupt reaches that path or reproduce a full RS reconnect with a
saturated shared queue.
- The pre-existing early completion of owned duplicates and RS-side
suppression of own changes not yet sent before a newer local publication
are unchanged.
- The Docker convergence scenario from #1244 and a throughput benchmark
were not run locally.

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

@@ -1147,7 +1147,7 @@
     */
    pendingChanges = new PendingChanges(getGenerator(), this,
        new ShutdownSyncAnnouncer(dsrsShutdownSync, getBaseDN()));
    remotePendingChanges = new RemotePendingChanges(getServerState());
    remotePendingChanges = new RemotePendingChanges(getServerState(), getServerId());

    // listen for changes on the configuration
    configuration.addChangeListener(this);
@@ -2986,9 +2986,8 @@
   * Marks the specified message as the one currently processed by a replay thread.
   *
   * @param msg the message being processed
   * @return {@code false} if the change is not pending anymore, which happens when the
   *         session was restarted after a failed replay while this message was waiting
   *         in the replay queue: it is sent again, 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 a retry has replaced this stale delivery
   */
  boolean markInProgress(LDAPUpdateMsg msg)
  {
@@ -4842,13 +4841,9 @@
   * Asks for a session restart the way a replay thread on its way out does, without
   * running it.
   * <p>
   * Only there for the tests, which have no other way to leave a request standing at a
   * time of their choosing without a change released for it: the one a replay thread makes
   * on a domain whose session has no owner is made and run in one go, and one made under
   * an owner stands for a change which is listed, and which the owner forgets or the
   * checkpointer has delivered again.
   * Also used when a queue hand-off fails: the listener or replay thread must not
   * synchronously restart the session from its failure cleanup.
   */
  @VisibleForTesting
  public void requestSessionRestart()
  {
    sessionRestarts.request(SessionRestart.NOW);
@@ -5831,6 +5826,7 @@
    state.clearInMemory();
    state.loadState(strict);
    bringAStateWithNoChangeOfThisReplicaUpToTheData(strict);
    remotePendingChanges.resetReplayState();
    getGenerator().adjust(state.getMaxCSN(getServerId()));
    forgetTheLocalChangesTheDataLacks(strict);

@@ -8189,24 +8185,30 @@
    {
      LDAPUpdateMsg msg = (LDAPUpdateMsg) updateMsg;

      // Put the UpdateMsg in the RemotePendingChanges list.
      if (!remotePendingChanges.putRemoteUpdate(msg))
      final UpdateToReplay updateToReplay = new UpdateToReplay(msg, this);
      final RemotePendingChanges.Admission admission = remotePendingChanges.putRemoteUpdate(updateToReplay);
      if (admission == RemotePendingChanges.Admission.QUEUED_DUPLICATE)
      {
        // The queued copy completes the current session's delivery after replay, including
        // its error. An immediate completion here would grant extra credit and an early ack.
        incProcessedUpdates();
        return false;
      }
      if (admission == RemotePendingChanges.Admission.DUPLICATE)
      {
        /*
         * A replay thread already owns this change, so this delivery is a duplicate:
         * ignore it. This happens when there are uncommitted changes in the queue and
         * session failover occurs causing a recovery of all changes since the current
         * committed server state. See OPENDJ-1115.
         * This change is already applied, owned by a replay thread, or committed
         * behind another pending change. Session failover can deliver any of these again
         * (OPENDJ-1115 and issue #1269).
         *
         * The copy which is already listed owns the change: it is the one which records
         * it in the ServerState once it really has been replayed. Report this delivery
         * The accepted copy is the one which records the change in the ServerState once
         * it really has been replayed, if it has not already done so. Report this delivery
         * as done - the window and the ack are per delivery - but as handled
         * asynchronously, so that the listener does not push the CSN to the ServerState
         * over a change which is still being replayed or is failing (issue #889).
         *
         * A change whose replay failed is not owned by anyone anymore, so this is not the
         * path it takes: putRemoteUpdate() takes this delivery over from the one which
         * failed and has it replayed again.
         * A released change has neither an owner nor a queued delivery, so this is not
         * the path it takes: putRemoteUpdate() accepts its next delivery for replay.
         */
        if (logger.isTraceEnabled())
        {
@@ -8220,22 +8222,45 @@

      // Put update message into the replay queue
      // (block until some place in the queue is available)
      final UpdateToReplay updateToReplay = new UpdateToReplay(msg, this);
      while (!isListenerShuttingDown())
      boolean queued = false;
      boolean handoffFailed = true;
      try
      {
        // loop until we can offer to the queue or shutdown was initiated
        try
        while (!isListenerShuttingDown())
        {
          if (updateToReplayQueue.offer(updateToReplay, 1, TimeUnit.SECONDS))
          // loop until we can offer to the queue or shutdown was initiated
          try
          {
            // successful offer to the queue, let's exit the loop
            break;
            if (updateToReplayQueue.offer(updateToReplay, 1, TimeUnit.SECONDS))
            {
              queued = true;
              synchronized (updateToReplayQueue)
              {
                updateToReplayQueue.notifyAll();
              }
              break;
            }
          }
          catch (InterruptedException e)
          {
            // Thread interrupted: check for shutdown.
            Thread.currentThread().interrupt();
          }
        }
        catch (InterruptedException e)
        handoffFailed = false;
      }
      finally
      {
        if (!queued)
        {
          // Thread interrupted: check for shutdown.
          Thread.currentThread().interrupt();
          // Cancel before comparing listener identity. If join completes, the next session
          // starts after this cancellation and can redeliver without an extra restart. If
          // join was interrupted, a newer session may already have refused the delivery.
          updateToReplay.cancel(false);
          if (handoffFailed || !isCurrentListenerThread())
          {
            requestSessionRestart();
          }
        }
      }


 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java

@@ -38,9 +38,20 @@
   */
  private volatile UpdateMsg msg;
  /**
   * A remote delivery has been accepted and is waiting to be claimed from the replay
   * queue. It has no owner yet, but replacing it would move it behind newer changes.
   * Local pending changes do not use this flag.
   */
  @GuardedBy("RemotePendingChanges.pendingChangesLock")
  private boolean queuedForReplay = true;
  @GuardedBy("RemotePendingChanges.pendingChangesLock")
  private boolean replayStarted;
  @GuardedBy("RemotePendingChanges.pendingChangesLock")
  private UpdateToReplay delivery;
  /**
   * The replay thread which owns this change: it is being replayed, or it waits for the
   * change it depends on. A remote change which no thread owns is one whose replay
   * failed and which the replication server is expected to deliver again.
   * change it depends on. A remote change which no thread owns is either queued for
   * replay or was released and may be taken over by another delivery.
   * <p>
   * The owner is kept rather than the bare fact that there is one, so that a change is
   * given back by the thread it was handed to and by nobody else: a release which arrives
@@ -137,6 +148,43 @@
  public void setMsg(LDAPUpdateMsg msg)
  {
    this.msg = msg;
    queuedForReplay = true;
  }

  /**
   * Returns whether a remote delivery is waiting to be claimed from the replay queue.
   *
   * @return {@code true} if the accepted delivery has not been dequeued or cancelled
   */
  public boolean isQueuedForReplay()
  {
    return queuedForReplay && !isDeliveryCancelled();
  }

  /** Installs a retry without changing this change's ordering or failure history. */
  void setDelivery(UpdateToReplay delivery)
  {
    setMsg(delivery.getUpdateMessage());
    this.delivery = delivery;
  }

  /** A failed queue hand-off can be cancelled without acquiring the pending lock. */
  boolean isDeliveryCancelled()
  {
    return delivery != null && delivery.isCancelled();
  }

  /** Marks a remote delivery as dequeued while its replay claim is being established. */
  public void markDequeued()
  {
    queuedForReplay = false;
    replayStarted = true;
  }

  /** Keeps the in-progress monitor stable while a failed replay waits in the retry queue. */
  boolean hasStartedReplay()
  {
    return replayStarted;
  }

  /**

 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))
      {

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java

@@ -74,6 +74,10 @@
  public void shutdown()
  {
    shutdown.set(true);
    synchronized (updateToReplayQueue)
    {
      updateToReplayQueue.notifyAll();
    }
  }

  /**
@@ -143,29 +147,53 @@
        {
          LDAPReplicationDomain domain;
          LDAPUpdateMsg updateMsg;
          UpdateToReplay delivery = null;
          try
          {
            if (shutdown.get())
            synchronized (updateToReplayQueue)
            {
              break;
              if (shutdown.get())
              {
                break;
              }
              if (updateToReplayQueue.isEmpty())
              {
                updateToReplayQueue.wait(1000L);
              }
              if (shutdown.get())
              {
                break;
              }
              // Retain the head before removing it. LinkedBlockingQueue.poll() can throw
              // OOME after unlinking it (signalNotFull takes another lock on JDK 11/17).
              // Every consumer holds this monitor, so peek and poll refer to the same item.
              delivery = updateToReplayQueue.peek();
              if (delivery == null)
              {
                continue;
              }
              updateToReplayQueue.poll();
              updateMsg = delivery.getUpdateMessage();
              domain = delivery.getReplicationDomain();
              if (!domain.markInProgress(updateMsg))
              {
                /*
                 * This delivery was cancelled, replaced, committed or forgotten. The
                 * domain counted the stale copy; only its accepted retry may replay it.
                 */
                continue;
              }
            }
            UpdateToReplay updateToreplay = updateToReplayQueue.poll(1L, TimeUnit.SECONDS);
            if (updateToreplay == null)
          }
          catch (Throwable t)
          {
            if (delivery != null)
            {
              continue;
              // No pending-changes lock is needed: the failed claim may itself have
              // run out of memory while trying to acquire that lock.
              delivery.cancel();
            }
            // Find replication domain for that update message and mark it as "in progress"
            updateMsg = updateToreplay.getUpdateMessage();
            domain = updateToreplay.getReplicationDomain();
            if (!domain.markInProgress(updateMsg))
            {
              /*
               * The domain restarted its session after a failed replay while this
               * message was waiting here, so it does not know about this change
               * anymore: the replication server sends it again over the new session.
               */
              continue;
            }
            throw t;
          }
          finally
          {

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/UpdateToReplay.java

@@ -13,21 +13,22 @@
 *
 * Copyright 2006-2008 Sun Microsystems, Inc.
 * Portions Copyright 2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import org.opends.server.replication.protocol.LDAPUpdateMsg;

/**
 * This is a bag class to hold an update to replay in the queue of updates to
 * be replayed by the replay threads.
 * It associates an update message to replay with the matching
 * ReplicationDomain.
 * A delivery in the shared replay queue, associating an update with its domain.
 * Cancellation belongs to this delivery, so an enqueue or claim failure can release
 * it without acquiring the pending-changes lock or cancelling a newer retry.
 */
public class UpdateToReplay
{
  private LDAPUpdateMsg updateMessage;
  private LDAPReplicationDomain replicationDomain;
  private final LDAPUpdateMsg updateMessage;
  private final LDAPReplicationDomain replicationDomain;
  private volatile boolean cancelled;

  /**
   * Construct the object associating the update message with the replication
@@ -44,6 +45,30 @@
  }

  /**
   * Releases a failed hand-off without a lock or allocation, even on an OOME road.
   * A stale cancellation affects only this delivery, never the retry which replaced it.
   */
  void cancel()
  {
    cancel(true);
  }

  /** Cancels this delivery, optionally requesting another session. */
  void cancel(boolean requestRestart)
  {
    cancelled = true;
    if (requestRestart && replicationDomain != null)
    {
      replicationDomain.requestSessionRestart();
    }
  }

  boolean isCancelled()
  {
    return cancelled;
  }

  /**
   * Getter for update message.
   * @return The update message
   */

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java

@@ -3599,6 +3599,12 @@
    return tmp == null || tmp.isShutdownInitiated();
  }

  /** Returns whether the calling thread is still the registered listener for this session. */
  protected final boolean isCurrentListenerThread()
  {
    return Thread.currentThread() == listenerThread;
  }

  /**
   * Returns the generation of the session of this domain: bumped by
   * {@link #disableService()} and {@link #enableService()} under

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/DuplicateReplayTest.java

New file
@@ -0,0 +1,746 @@
/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.core.DirectoryServer.*;
import static org.testng.Assert.*;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.opends.server.api.DirectoryThread;
import org.opends.server.core.AddOperation;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.protocol.AckMsg;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.replication.service.ReplicationDomain;
import org.opends.server.types.Entry;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

/** Tests duplicate delivery across reconnects against the real replay and backend (issue #1269). */
@SuppressWarnings("javadoc")
public class DuplicateReplayTest extends ReplicationTestCase
{
  private static final String ENTRY_UUID = "c9cb8c3c-615a-4122-865d-50323aaaed48";

  private DN baseDN;
  private LDAPReplicationDomain domain;
  private TestSynchronousReplayQueue queue;
  private Entry entry;
  private String parentUUID;
  private CSN addCSN;
  private CSN deleteCSN;
  private OutOfMemoryError enqueueFailure;
  private OutOfMemoryError pollFailure;
  private boolean failBeforePoll;
  private Runnable beforeEnqueue;
  private boolean refuseEnqueue;
  private Runnable onEmpty;
  private Runnable afterPoll;

  @BeforeMethod
  public void setUpLocal() throws Exception
  {
    baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    initializeTestBackend(true);
    enqueueFailure = null;
    pollFailure = null;
    failBeforePoll = false;
    beforeEnqueue = null;
    refuseEnqueue = false;
    onEmpty = null;
    afterPoll = null;
    queue = new TestSynchronousReplayQueue()
    {
      @Override
      public boolean offer(UpdateToReplay update, long timeout, TimeUnit unit) throws InterruptedException
      {
        if (beforeEnqueue != null)
        {
          beforeEnqueue.run();
        }
        if (enqueueFailure != null)
        {
          throw enqueueFailure;
        }
        return !refuseEnqueue && super.offer(update, timeout, unit);
      }

      @Override
      public UpdateToReplay poll(long timeout, TimeUnit unit)
      {
        return poll();
      }

      @Override
      public UpdateToReplay poll()
      {
        if (pollFailure != null && failBeforePoll)
        {
          throw pollFailure;
        }
        final UpdateToReplay result = super.poll();
        if (pollFailure != null)
        {
          throw pollFailure;
        }
        if (afterPoll != null)
        {
          afterPoll.run();
        }
        return result;
      }

      @Override
      public boolean isEmpty()
      {
        final boolean empty = super.isEmpty();
        if (empty && onEmpty != null)
        {
          onEmpty.run();
        }
        return empty;
      }
    };
    final DomainFakeCfg conf = new DomainFakeCfg(baseDN, 1, new TreeSet<String>());
    conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES);
    domain = MultimasterReplication.createNewDomain(conf, queue);
    domain.start();

    entry = makeEntry("dn: cn=duplicate," + baseDN,
        "objectClass: top", "objectClass: person", "cn: duplicate", "sn: duplicate",
        "entryUUID: " + ENTRY_UUID);
    parentUUID = getEntryUUID(baseDN);
    final CSNGenerator generator = new CSNGenerator(201, 0);
    addCSN = generator.newCSN();
    deleteCSN = generator.newCSN();
  }

  @AfterMethod
  public void tearDown() throws Exception
  {
    MultimasterReplication.deleteDomain(baseDN);
  }

  @Test
  public void appliedAddDeliveredAgainAfterDeleteDoesNotResurrectEntry() throws Exception
  {
    domain.processUpdate(addMsg());
    replayQueuedMessages();
    assertTrue(entryExists(entry.getName()));
    assertEquals(getEntryUUID(entry.getName()), ENTRY_UUID);
    domain.processUpdate(deleteMsg());
    replayQueuedMessages();
    assertFalse(entryExists(entry.getName()));
    assertTrue(domain.getServerState().cover(deleteCSN));

    // A new session can deliver an old change after it has left the pending list.
    assertFalse(domain.processUpdate(addMsg()));
    replayQueuedMessages();

    assertFalse(entryExists(entry.getName()), "an applied ADD was replayed after its DELETE");
    assertEquals(processedUpdates(), 3, "each delivery must be counted, including the duplicate");
  }

  @Test
  public void queuedDuplicateAddDoesNotMoveTheOriginalBehindDelete() throws Exception
  {
    domain.processUpdate(addMsg());
    domain.processUpdate(deleteMsg());
    final int completions = completedUpdates();
    // Pause replay until the duplicate has arrived: no timing or reconnect race is needed.
    assertFalse(domain.processUpdate(addMsg()));
    assertEquals(completedUpdates(), completions, "a queued duplicate must not return credit before replay");
    assertFalse(domain.getServerState().cover(addCSN), "discarding a duplicate must not advance state");
    assertFalse(domain.getServerState().cover(deleteCSN));

    replayQueuedMessages();

    assertEquals(completedUpdates(), completions + 2, "one completion per replayed change");
    assertFalse(entryExists(entry.getName()), "the queued ADD moved behind its DELETE");
    assertTrue(domain.getServerState().cover(deleteCSN));
    assertEquals(processedUpdates(), 3, "each delivery must be counted, including the duplicate");
  }

  @Test
  public void deliveryAbandonedBeforeEnqueueCanBeRetried() throws Exception
  {
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    recovering.set(true);
    try
    {
      domain.disableService();
      domain.processUpdate(addMsg());
      assertTrue(queue.isEmpty());
      assertFalse(domain.getServerState().cover(addCSN));

      domain.enableService();
      domain.processUpdate(addMsg());
      replayQueuedMessages();

      assertTrue(entryExists(entry.getName()), "a delivery abandoned before enqueue was refused on retry");
      assertTrue(domain.getServerState().cover(addCSN));
    }
    finally
    {
      restarts.clear();
      recovering.set(false);
    }
  }

  @Test
  public void failedEnqueueCanBeRetried() throws Exception
  {
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    recovering.set(true);
    try
    {
      enqueueFailure = new OutOfMemoryError("enqueue allocation failed");
      try
      {
        domain.processUpdate(addMsg());
        fail("the injected allocation failure did not reach enqueue");
      }
      catch (OutOfMemoryError e)
      {
        assertSame(e, enqueueFailure);
      }
      finally
      {
        enqueueFailure = null;
      }
      assertTrue(queue.isEmpty());
      assertFalse(domain.getServerState().cover(addCSN));

      assertTrue(restarts.isPending(), "a failed enqueue must request redelivery");
      domain.processUpdate(addMsg());
      replayQueuedMessages();

      assertTrue(entryExists(entry.getName()), "a failed enqueue was refused on retry");
      assertTrue(domain.getServerState().cover(addCSN));
    }
    finally
    {
      restarts.clear();
      recovering.set(false);
    }
  }

  @Test
  public void ownChangeSentBeforeANewerLocalPublishIsStillReplayed() throws Exception
  {
    // Model the incoming backlog after restoring a backup that predates these changes.
    final CSNGenerator generator = new CSNGenerator(1, 0);
    addCSN = generator.newCSN();
    deleteCSN = generator.newCSN();
    domain.publish(deleteMsg());
    assertTrue(domain.getServerState().cover(addCSN));

    domain.processUpdate(addMsg());
    replayQueuedMessages();
    assertTrue(entryExists(entry.getName()), "local publication must not hide a missing own-origin change");
    domain.processUpdate(deleteMsg());
    replayQueuedMessages();
    assertFalse(entryExists(entry.getName()));
    domain.processUpdate(addMsg());
    replayQueuedMessages();
    assertFalse(entryExists(entry.getName()), "own-origin replay also needs duplicate suppression");
  }

  @Test
  public void backendReloadResetsTheOwnReplayWatermark() throws Exception
  {
    final CSNGenerator generator = new CSNGenerator(1, 0);
    addCSN = generator.newCSN();
    deleteCSN = generator.newCSN();
    domain.processUpdate(addMsg());
    domain.processUpdate(deleteMsg());
    replayQueuedMessages();
    assertFalse(entryExists(entry.getName()));
    domain.disable();
    // Restore the persisted state of a backup taken before the ADD. Its entry is absent,
    // as it is in this backend after DELETE; no history on that entry can cover the ADD.
    ((PersistentServerState) domainField("state")).clear();
    domain.enable();
    assertFalse(domain.getServerState().cover(addCSN));
    domain.processUpdate(addMsg());
    replayQueuedMessages();
    assertTrue(entryExists(entry.getName()), "backend reload retained the pre-restore replay watermark");
  }

  private int completedUpdates() throws Exception
  {
    final Field brokerField = ReplicationDomain.class.getDeclaredField("broker");
    brokerField.setAccessible(true);
    final Object broker = brokerField.get(domain);
    final Field counter = ReplicationBroker.class.getDeclaredField("updateDoneCount");
    counter.setAccessible(true);
    return counter.getInt(broker);
  }

  @DataProvider
  public Object[][] queueFailurePositions()
  {
    return new Object[][] { { false }, { true } };
  }

  @Test(dataProvider = "queueFailurePositions")
  public void failureAroundQueueRemovalDoesNotLoseTheChange(boolean beforeRemoval) throws Exception
  {
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    recovering.set(true); // Keep the checkpointer from consuming the request under test.
    try
    {
      domain.processUpdate(addMsg());
      domain.processUpdate(deleteMsg());
      failBeforePoll = beforeRemoval;
      pollFailure = new OutOfMemoryError("dequeue failed");
      try
      {
        new ReplayThread(queue, new ReentrantLock()).run();
        fail("the injected dequeue failure did not escape");
      }
      catch (OutOfMemoryError e)
      {
        assertSame(e, pollFailure);
      }
      finally
      {
        pollFailure = null;
      }
      assertEquals(queue.size(), beforeRemoval ? 2 : 1);
      assertTrue(restarts.isPending(), "a lost queue delivery must be requested again");
      replayQueuedMessages(); // DELETE must wait for the cancelled ADD, not commit NO_SUCH_OBJECT.
      domain.processUpdate(addMsg());
      replayQueuedMessages();
      assertFalse(entryExists(entry.getName()), "DELETE overtook the retry of the removed ADD");
      assertTrue(domain.getServerState().cover(deleteCSN));
    }
    finally
    {
      restarts.clear();
      recovering.set(false);
    }
  }

  @Test
  public void cancellationAfterANewerDeliveryWasRefusedRequestsAnotherSession() throws Exception
  {
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    recovering.set(true);
    try
    {
      enqueueFailure = new OutOfMemoryError("old listener cannot finish enqueue");
      beforeEnqueue = () ->
      {
        // A newer session redelivers while the old listener is still in offer().
        assertFalse(domain.processUpdate(addMsg()));
        assertTrue(queue.isEmpty());
        beforeEnqueue = null;
        final OutOfMemoryError oldFailure = enqueueFailure;
        enqueueFailure = null;
        try
        {
          domain.processUpdate(deleteMsg());
        }
        finally
        {
          enqueueFailure = oldFailure;
        }
      };
      try
      {
        domain.processUpdate(addMsg());
        fail("enqueue must fail");
      }
      catch (OutOfMemoryError e)
      {
        assertSame(e, enqueueFailure);
      }
      finally
      {
        beforeEnqueue = null;
        enqueueFailure = null;
      }
      assertTrue(restarts.isPending(), "the newer session already missed its chance to retry");
      replayQueuedMessages();
      domain.processUpdate(addMsg());
      replayQueuedMessages();
      assertFalse(entryExists(entry.getName()), "DELETE overtook the cancelled ADD's retry");
      assertTrue(domain.getServerState().cover(deleteCSN));
    }
    finally
    {
      restarts.clear();
      recovering.set(false);
    }
  }

  @DataProvider
  public Object[][] listenerStops()
  {
    return new Object[][] { { false }, { true } };
  }

  @Test(dataProvider = "listenerStops")
  public void shutdownCancellationRestartsOnlyAfterAnIncompleteJoin(boolean interruptJoin) throws Exception
  {
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    final CountDownLatch offering = new CountDownLatch(1);
    final CountDownLatch releaseOffer = new CountDownLatch(1);
    final CountDownLatch stopping = new CountDownLatch(1);
    final AtomicReference<Throwable> listenerFailure = new AtomicReference<>();
    final Field listenerField = ReplicationDomain.class.getDeclaredField("listenerThread");
    listenerField.setAccessible(true);
    recovering.set(true);
    domain.disableService();
    refuseEnqueue = true;
    beforeEnqueue = () ->
    {
      offering.countDown();
      await(releaseOffer);
    };
    // Use the real listener slot and disableService/join, but supply the incoming change
    // directly so the interleaving does not need a replication server or a full 10000-slot queue.
    final DirectoryThread listener = new DirectoryThread(() ->
    {
      try
      {
        domain.processUpdate(addMsg());
      }
      catch (Throwable t)
      {
        listenerFailure.set(t);
      }
    }, "held duplicate-replay listener");
    final FutureTask<Void> stop = new FutureTask<>(() -> { domain.disableService(); return null; });
    final Thread stopper = new Thread(stop, "duplicate-replay service stop");
    try
    {
      listenerField.set(domain, listener);
      listener.start();
      assertTrue(offering.await(10, TimeUnit.SECONDS));
      domain.setServiceStopHook(stopping::countDown);
      stopper.start();
      assertTrue(stopping.await(10, TimeUnit.SECONDS));
      if (interruptJoin)
      {
        stopper.interrupt();
        stop.get(10, TimeUnit.SECONDS);
        domain.setServiceStopHook(null);
        domain.enableService();
        domain.processUpdate(addMsg()); // The new session refuses the still-live old delivery.
        assertTrue(queue.isEmpty());
        domain.disableService();
        restarts.clear(); // A restart before cancellation cannot be enough.
      }
      else
      {
        // Taking sessionLock would deadlock with join; wait on the thread's own flag.
        final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
        while (!listener.isShutdownInitiated() && System.nanoTime() < deadline)
        {
          Thread.yield();
        }
        assertTrue(listener.isShutdownInitiated());
      }
      releaseOffer.countDown();
      stop.get(10, TimeUnit.SECONDS);
      listener.join(10000);
      assertFalse(listener.isAlive());
      assertNull(listenerFailure.get());
      assertEquals(restarts.isPending(), interruptJoin,
          "a completed join already precedes redelivery; a detached listener needs a fresh restart");
      refuseEnqueue = false;
      beforeEnqueue = null;
      domain.enableService();
      domain.processUpdate(addMsg());
      replayQueuedMessages();
      assertTrue(entryExists(entry.getName()));
    }
    finally
    {
      releaseOffer.countDown();
      stopper.join(10000);
      listener.join(10000);
      domain.setServiceStopHook(null);
      restarts.clear();
      recovering.set(false);
    }
  }

  @Test
  public void failedHandoffByRegisteredListenerRequestsRestart() throws Exception
  {
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    final AtomicReference<Throwable> listenerFailure = new AtomicReference<>();
    final Field listenerField = ReplicationDomain.class.getDeclaredField("listenerThread");
    listenerField.setAccessible(true);
    recovering.set(true);
    domain.disableService();
    enqueueFailure = new OutOfMemoryError("registered listener enqueue failed");
    final DirectoryThread listener = new DirectoryThread(() ->
    {
      try
      {
        domain.processUpdate(addMsg());
      }
      catch (Throwable t)
      {
        listenerFailure.set(t);
      }
    }, "failed duplicate-replay listener");
    try
    {
      listenerField.set(domain, listener);
      listener.start();
      listener.join(10000);
      assertFalse(listener.isAlive());
      assertSame(listenerFailure.get(), enqueueFailure);
      assertTrue(restarts.isPending(), "failure needs recovery even while the listener is registered");
      enqueueFailure = null;
      domain.disableService();
      domain.enableService();
      domain.processUpdate(addMsg());
      replayQueuedMessages();
      assertTrue(entryExists(entry.getName()));
    }
    finally
    {
      listener.join(10000);
      enqueueFailure = null;
      restarts.clear();
      recovering.set(false);
    }
  }

  @Test
  public void replayConsumerWakesForAnEmptyQueueAndDrainsItsBacklog() throws Exception
  {
    final CountDownLatch empty = new CountDownLatch(1);
    final CountDownLatch removed = new CountDownLatch(2);
    final CountDownLatch replayed = new CountDownLatch(1);
    final ReplayThread consumer = new ReplayThread(queue, new ReentrantLock());
    onEmpty = () ->
    {
      empty.countDown();
      if (domain.getServerState().cover(deleteCSN))
      {
        replayed.countDown();
      }
    };
    afterPoll = removed::countDown;
    try
    {
      consumer.start();
      assertTrue(empty.await(10, TimeUnit.SECONDS));
      synchronized (queue)
      {
        // The consumer observed an empty queue while holding this monitor. Acquiring it
        // proves it has now entered wait(). Deliver both changes before it can wake up.
        domain.processUpdate(addMsg());
        domain.processUpdate(deleteMsg());
      }
      assertTrue(removed.await(750, TimeUnit.MILLISECONDS),
          "notification must wake the consumer and a non-empty queue must not wait again");
      // poll() runs before replay. Stopping at the second removal could abandon DELETE.
      assertTrue(replayed.await(10, TimeUnit.SECONDS), "the consumer did not commit its backlog");
    }
    finally
    {
      consumer.shutdown();
      consumer.join(10000);
      onEmpty = null;
      afterPoll = null;
    }
    assertFalse(consumer.isAlive());
    assertFalse(entryExists(entry.getName()));
    assertTrue(domain.getServerState().cover(deleteCSN));
  }

  private static void await(CountDownLatch latch)
  {
    try
    {
      assertTrue(latch.await(10, TimeUnit.SECONDS), "controlled hand-off timed out");
    }
    catch (InterruptedException e)
    {
      Thread.currentThread().interrupt();
      throw new AssertionError(e);
    }
  }

  private Object domainField(String name) throws Exception
  {
    final Field field = LDAPReplicationDomain.class.getDeclaredField(name);
    field.setAccessible(true);
    return field.get(domain);
  }

  @DataProvider
  public Object[][] assuredDeliveries()
  {
    return new Object[][] {
      { false, true, false }, { true, true, false }, { true, false, false },
      { false, true, true }, { true, true, true }, { true, false, true }
    };
  }

  @Test(dataProvider = "assuredDeliveries")
  public void queuedRedeliveryAcknowledgesOnlyTheReplayResult(
      boolean firstAssured, boolean lastAssured, boolean failReplay) throws Exception
  {
    final Field brokerField = ReplicationDomain.class.getDeclaredField("broker");
    brokerField.setAccessible(true);
    final ReplicationBroker originalBroker = (ReplicationBroker) brokerField.get(domain);
    final List<AckMsg> acks = new ArrayList<>();
    // A non-started broker retains real completion accounting but records the wire ack.
    // No domain is passed to its constructor, so it registers no second monitor.
    final ReplicationBroker recordingBroker = new ReplicationBroker(null, domain.getServerState(),
        new DomainFakeCfg(baseDN, 1, new TreeSet<String>()), new ReplSessionSecurity())
    {
      @Override
      public byte getRsGroupId()
      {
        return domain.getGroupId();
      }

      @Override
      public void publish(ReplicationMsg msg)
      {
        if (msg instanceof AckMsg)
        {
          acks.add((AckMsg) msg);
        }
      }
    };
    brokerField.set(domain, recordingBroker);
    final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
    final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
    recovering.set(true);
    final OutOfMemoryError failure = new OutOfMemoryError("cannot construct replay operation");
    try
    {
      final AddMsg original = failReplay ? new AddMsg(addCSN, entry.getName(), ENTRY_UUID, parentUUID,
          entry.getObjectClasses(), entry.getUserAttributes(), entry.getOperationalAttributes())
      {
        @Override
        public AddOperation createOperation(InternalClientConnection connection, DN newDN)
        {
          throw failure;
        }
      } : addMsg();
      original.setAssured(firstAssured);
      original.setAssuredMode(AssuredMode.SAFE_DATA_MODE);
      domain.processUpdate(original);
      for (int i = 0; i < 2; i++)
      {
        final AddMsg duplicate = addMsg();
        duplicate.setAssured(lastAssured);
        duplicate.setAssuredMode(AssuredMode.SAFE_READ_MODE);
        domain.processUpdate(duplicate);
        assertTrue(acks.isEmpty(), "queued redelivery must not acknowledge unapplied data");
        assertEquals(completedUpdates(), 0);
      }
      try
      {
        replayQueuedMessages();
        assertFalse(failReplay, "the injected replay failure must escape");
      }
      catch (OutOfMemoryError e)
      {
        assertTrue(failReplay);
        assertSame(e, failure);
      }
      assertEquals(completedUpdates(), 1, "only the accepted replay returns window credit");
      assertEquals(acks.size(), lastAssured ? 1 : 0);
      if (lastAssured)
      {
        assertEquals(acks.get(0).getCSN(), addCSN);
        assertEquals(acks.get(0).hasReplayError(), failReplay);
      }
      assertEquals(entryExists(entry.getName()), !failReplay);
    }
    finally
    {
      brokerField.set(domain, originalBroker);
      restarts.clear();
      recovering.set(false);
    }
  }

  private AddMsg addMsg()
  {
    return new AddMsg(addCSN, entry.getName(), ENTRY_UUID, parentUUID,
        entry.getObjectClasses(), entry.getUserAttributes(), entry.getOperationalAttributes());
  }

  private DeleteMsg deleteMsg()
  {
    return new DeleteMsg(entry.getName(), deleteCSN, ENTRY_UUID);
  }

  private void replayQueuedMessages() throws Exception
  {
    while (!queue.isEmpty())
    {
      final LDAPUpdateMsg msg = queue.take().getUpdateMessage();
      if (domain.markInProgress(msg))
      {
        domain.replay(msg, new AtomicBoolean(false));
      }
    }
  }

  private int processedUpdates() throws Exception
  {
    // The monitor counter is package-private in service; this domain has no RS to register a monitor.
    final Method getter = ReplicationDomain.class.getDeclaredMethod("getNumProcessedUpdates");
    getter.setAccessible(true);
    return (Integer) getter.invoke(domain);
  }
}

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java

@@ -147,11 +147,13 @@
    Entry entry = createAndAddEntry("simultaneousModrdnConflict");
    String entryUUID = getEntryUUID(entry.getName());

    // generate two consecutive CSN that will be used in backward order
    // Changes from different origins can arrive in reverse CSN order. Older changes
    // from the same origin would already be covered by its applied ServerState.
    CSN csn1 = gen.newCSN();
    CSN csn2 = gen.newCSN();
    CSN csn2 = new CSN(csn1.getTime() + 1, 0, 202);

    replayMsg(modDnMsg(entry, entryUUID, parentUUID, csn2, "uid=simultaneous2"));
    assertFalse(domain.getServerState().cover(csn1));

    // This MODIFY DN uses an older DN and should therefore be cancelled at replay time.
    replayMsg(modDnMsg(entry, entryUUID, parentUUID, csn1, "uid=simulatneouswrong"));
@@ -287,10 +289,11 @@
    final String entryUUID = getEntryUUID(entry.getName());
    final int givenUpBefore = failedReplayedUpdates();

    // Two consecutive CSNs, replayed in the reverse order.
    // Reverse order across origins must still reach naming-conflict resolution.
    final CSN older = gen.newCSN();
    final CSN newer = gen.newCSN();
    final CSN newer = new CSN(older.getTime() + 1, 0, 202);
    replayMsg(modDnMsg(entry, entryUUID, parentUUID, newer, "cn=renamedAfter"));
    assertFalse(domain.getServerState().cover(older));
    replayMsg(modDnMsg(entry, entryUUID, parentUUID, older, "cn=renamedBefore"));

    assertTrue(entryExists(DN.valueOf("cn=renamedAfter," + TEST_ROOT_DN_STRING)),

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java

@@ -19,9 +19,14 @@
import static org.testng.Assert.*;

import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
@@ -33,6 +38,7 @@
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

/**
@@ -68,6 +74,264 @@
    assertEquals(pendingChanges.getQueueSize(), 0);
  }

  @Test
  public void appliedChangesAreRejectedAfterLeavingThePendingList() throws Exception
  {
    final ServerState state = new ServerState();
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
    final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
    final CSN first = generator.newCSN();
    final CSN last = generator.newCSN();

    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(first, "uuid-1")));
    pendingChanges.commit(first);
    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(last, "uuid-2")));
    pendingChanges.commit(last);

    assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(first, "uuid-1")));
    assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(last, "uuid-2")));
    assertEquals(pendingChanges.getQueueSize(), 0);

    // The watermark belongs to the originating server, not to all servers at once.
    final CSN anotherOrigin = new CSN(first.getTime(), first.getSeqnum(), SERVER_ID + 1);
    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(anotherOrigin, "uuid-3")));
  }

  @Test
  public void duplicateDoesNotReplaceAQueuedDelivery() throws Exception
  {
    final ServerState state = new ServerState();
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg original = deleteMsg(csn, "uuid-1");
    final DeleteMsg duplicate = deleteMsg(csn, "uuid-1");

    assertTrue(pendingChanges.putRemoteUpdate(original));
    assertFalse(pendingChanges.putRemoteUpdate(duplicate));
    assertFalse(pendingChanges.markInProgress(duplicate));
    assertTrue(pendingChanges.markInProgress(original));
    assertTrue(state.isEmpty(), "discarding a duplicate must not record the queued change as replayed");
    pendingChanges.commit(csn);
    assertTrue(state.cover(csn));
  }

  @Test
  public void duplicateDoesNotReplaceAQueuedRetry() throws Exception
  {
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg original = deleteMsg(csn, "uuid-1");
    final DeleteMsg retry = deleteMsg(csn, "uuid-1");
    final DeleteMsg duplicate = deleteMsg(csn, "uuid-1");

    assertTrue(pendingChanges.putRemoteUpdate(original));
    assertTrue(pendingChanges.markInProgress(original));
    assertTrue(pendingChanges.replayFailed(csn));
    assertTrue(pendingChanges.putRemoteUpdate(retry));
    assertEquals(pendingChanges.changesInProgressSize(), 1, "a queued retry retains its in-progress count");
    assertFalse(pendingChanges.putRemoteUpdate(duplicate));
    assertFalse(pendingChanges.markInProgress(original));
    assertFalse(pendingChanges.markInProgress(duplicate));
    assertTrue(pendingChanges.markInProgress(retry));
  }

  @Test
  public void interruptedClaimDoesNotLeaveTheDeliveryMarkedAsQueued() throws Exception
  {
    final OutOfMemoryError failure = new OutOfMemoryError("claim allocation failed");
    final AtomicBoolean failClaim = new AtomicBoolean(true);
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState(), -1,
        new ConcurrentHashMap<Thread, CSN>()
        {
          @Override
          public CSN put(Thread thread, CSN csn)
          {
            if (failClaim.getAndSet(false))
            {
              throw failure;
            }
            return super.put(thread, csn);
          }
        });
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg original = deleteMsg(csn, "uuid-1");
    assertTrue(pendingChanges.putRemoteUpdate(original));

    try
    {
      pendingChanges.markInProgress(original);
      fail("the injected allocation failure did not reach the claim");
    }
    catch (OutOfMemoryError e)
    {
      assertSame(e, failure);
    }

    final DeleteMsg retry = deleteMsg(csn, "uuid-1");
    assertTrue(pendingChanges.putRemoteUpdate(retry), "the dequeued delivery has no owner to finish it");
    assertTrue(pendingChanges.markInProgress(retry));
  }

  @DataProvider
  public Object[][] registrationFailures()
  {
    return new Object[][] { { false }, { true } };
  }

  @Test(dataProvider = "registrationFailures")
  public void failedDependencyRegistrationCancelsDeliveryBeforeRollback(boolean failRollback) throws Exception
  {
    final OutOfMemoryError registrationFailure = new OutOfMemoryError("dependency registration failed");
    final OutOfMemoryError rollbackFailure = new OutOfMemoryError("dependency rollback failed");
    final AtomicBoolean failRegistration = new AtomicBoolean(true);
    final ServerState state = new ServerState();
    final RemotePendingChanges pending = new RemotePendingChanges(state, -1, new ConcurrentHashMap<>(),
        new ConcurrentSkipListSet<PendingChange>()
        {
          @Override
          public boolean add(PendingChange change)
          {
            if (failRegistration.getAndSet(false))
            {
              // Also cover a set insertion that throws after linking its element.
              super.add(change);
              throw registrationFailure;
            }
            return super.add(change);
          }

          @Override
          public boolean remove(Object change)
          {
            if (failRollback)
            {
              throw rollbackFailure;
            }
            return super.remove(change);
          }
        });
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final UpdateToReplay delivery = new UpdateToReplay(deleteMsg(csn, "uuid-1"), null);
    try
    {
      pending.putRemoteUpdate(delivery);
      fail("registration failure must escape admission");
    }
    catch (OutOfMemoryError e)
    {
      assertSame(e, failRollback ? rollbackFailure : registrationFailure);
    }
    assertTrue(delivery.isCancelled(), "rollback failure must not skip delivery cancellation");
    assertFalse(state.cover(csn));
    if (!failRollback)
    {
      assertEquals(pending.getQueueSize(), 0);
      final DeleteMsg retry = deleteMsg(csn, "uuid-1");
      assertTrue(pending.putRemoteUpdate(retry), "failed registration must not wedge the next delivery");
      assertTrue(pending.markInProgress(retry));
      pending.commit(csn);
      assertTrue(state.cover(csn));
    }
  }

  @Test
  public void lateEnqueueCancellationDoesNotReleaseAnotherDeliveryOrItsOwner() throws Exception
  {
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg original = deleteMsg(csn, "uuid-1");
    final DeleteMsg retry = deleteMsg(csn, "uuid-1");
    final UpdateToReplay oldDelivery = new UpdateToReplay(original, null);
    final UpdateToReplay newDelivery = new UpdateToReplay(retry, null);
    assertEquals(pendingChanges.putRemoteUpdate(oldDelivery), RemotePendingChanges.Admission.ACCEPTED);
    oldDelivery.cancel();
    assertEquals(pendingChanges.putRemoteUpdate(newDelivery), RemotePendingChanges.Admission.ACCEPTED);

    oldDelivery.cancel();
    assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
        "cancelling the previous delivery must not release its queued replacement");
    assertTrue(pendingChanges.markInProgress(retry));

    newDelivery.cancel();
    assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
        "enqueue cancellation must not take a delivery away from its replay owner");
    assertTrue(pendingChanges.replayFailed(csn));
    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
  }

  @Test
  public void resettingStateAllowsPreviouslyAppliedChangesToBeReplayed() throws Exception
  {
    final ServerState state = new ServerState();
    final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
    pendingChanges.commit(csn);

    // Disabling a domain clears pending changes, but retains its applied state.
    pendingChanges.clear();
    assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));

    // A restore/import may reset that state so the same changes are needed again.
    state.clear();
    final DeleteMsg redelivered = deleteMsg(csn, "uuid-1");
    assertTrue(pendingChanges.putRemoteUpdate(redelivered));
    assertTrue(pendingChanges.markInProgress(redelivered));
  }

  @Test
  public void failedClaimBeforeTakingThePendingLockReleasesItsDelivery() throws Exception
  {
    final RemotePendingChanges pending = new RemotePendingChanges(new ServerState());
    final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
    final DeleteMsg original = deleteMsg(csn, "uuid-1");
    final OutOfMemoryError failure = new OutOfMemoryError("pending lock allocation failed");
    // Inject between dequeue and claim, before RemotePendingChanges can acquire its lock.
    final UpdateToReplay delivery = new UpdateToReplay(original, null)
    {
      @Override
      public LDAPReplicationDomain getReplicationDomain()
      {
        throw failure;
      }
    };
    assertEquals(pending.putRemoteUpdate(delivery), RemotePendingChanges.Admission.ACCEPTED);
    final LinkedBlockingQueue<UpdateToReplay> queue = new LinkedBlockingQueue<>();
    queue.add(delivery);
    try
    {
      new ReplayThread(queue, new ReentrantLock()).run();
      fail("the failed claim must end the replay thread");
    }
    catch (OutOfMemoryError e)
    {
      assertSame(e, failure);
    }
    assertTrue(queue.isEmpty());
    assertFalse(pending.markInProgress(original), "a cancelled delivery must not acquire ownership");
    final DeleteMsg retry = deleteMsg(csn, "uuid-1");
    assertTrue(pending.putRemoteUpdate(retry), "the failed claim must not block redelivery");
    assertTrue(pending.markInProgress(retry));
  }

  @Test
  public void ownReplayWatermarkSurvivesClearAndResetsWithBackendState() throws Exception
  {
    final ServerState state = new ServerState();
    final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
    final CSN first = generator.newCSN();
    final CSN later = generator.newCSN();
    final RemotePendingChanges pending = new RemotePendingChanges(state, SERVER_ID);
    state.update(later); // A local publication does not prove that first was replayed.
    assertTrue(pending.putRemoteUpdate(deleteMsg(first, "uuid-1")));
    pending.commit(first);
    pending.clear();
    assertFalse(pending.putRemoteUpdate(deleteMsg(first, "uuid-1")));
    state.clear();
    pending.resetReplayState();
    assertTrue(pending.putRemoteUpdate(deleteMsg(first, "uuid-1")));
  }

  /**
   * A change which is not committed must hold back the ServerState, even when the
   * changes which follow it have been replayed: the replication server resumes from the
@@ -180,9 +444,8 @@
  }

  /**
   * The replication server sends the change again over the new session while the message
   * of the previous delivery may still be waiting in the replay queue: only the delivery
   * which is listed as pending is replayed, or the same change would be applied twice.
   * A failed delivery must not claim the change after the replication server has sent
   * a replacement over the new session: only the replacement may replay it.
   */
  @Test
  public void markInProgressRejectsThePreviousDeliveryOfAChange() throws Exception
@@ -193,7 +456,8 @@
    final DeleteMsg newDelivery = deleteMsg(csn, "uuid-1");

    assertTrue(pendingChanges.putRemoteUpdate(previousDelivery));
    pendingChanges.replayFailed(csn);
    assertTrue(pendingChanges.markInProgress(previousDelivery));
    assertTrue(pendingChanges.replayFailed(csn));
    assertTrue(pendingChanges.putRemoteUpdate(newDelivery), "the next delivery must be replayed");

    assertFalse(pendingChanges.markInProgress(previousDelivery),