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

Valery Kharseko

89 mins ago 5154f34503981b2f96755b1b6da9b425bb20ca7f

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