[#1206] Take a late handler of a replication server back to its queue only ...

Valery Kharseko

22 hours ago b8a4d35f03b7fa5cf0d80a9424224f6ddeb3b3f9

[#1206] Take a late handler of a replication server back to its queue only once its consumer has every change the queue lacks (#1219)

Fixes #1206
Fixes #1224
Fixes #1225

## Problem

A replication server stopped sending a directory server the changes of
another replica, and went on minutes further. The changes in between
were in the changelog of every replication server, never reached that
directory server, and `missing-changes` still reached 0.

A handler which is behind reads the changes for its consumer from the
changelog. `MessageHandler.getNextMessage()` takes it back to its
in-memory queue once the queue holds the first change of what it read,
which is the oldest change the consumer lacks. Holding that change does
not mean holding the ones after it:

- The queue is ordered by CSN, and drops its oldest changes when it is
full.
- A change which reaches the queue late with an older CSN goes in front:
one relayed by another replication server, or published again by a
recovery.
- Behind it, the queue can lack newer changes of another replica: the
ones it dropped before that change came, or the ones which reached the
domain before the handler was registered.

That change is then the oldest one the consumer lacks, and it is in the
queue. The handler went back to the queue, the consumer was given the
changes the queue still held of the other replica, and its state moved
past the ones in between. The changelog is read again only for a change
the state of the consumer does not cover (#963), so they were never
sent.

A full queue keeps such a change because it is bounded in bytes as well
and drops whole changes: the room a large change leaves takes a small
one in.

The changelog is read without the lock of the queue, so a change
arriving in between is lost the same way:

- with an older CSN than the first change read, it was dropped along
with the changes before that one, which the consumer has;
- into a full queue as its oldest change, it was dropped at once, and a
handler whose read gave nothing went back to the queue all the same.

This fits what #1206 found:

- The replica skipped was connected to the same replication server as
the directory server. The replicas whose changes came through the other
replication server were not skipped.
- Both jumps of the description landed on the last 10 to 13 s of the
skipped replica, which is what the queue held of it.
- The new handlers of cases 2 and 3 of the comment skipped from the
moment they connected: their queues had nothing from before.

#1225 is the same with a hole of one change, in the `combined` topology
of the convergence check. The replication server of `cv-c2` never sent
its directory server one delete of `cv-c1`, though it sent the changes
of `cv-c1` on both sides of it and every add and delete of the other
replicas. That is test 1 below: the queue dropped the change, and a
change of another replica with an older CSN came in after it.

## Change

The handler records, for each replica, the newest change of it the queue
may lack:

- what the domain held when it handed the queue its first change: the
domain hands a registered handler every change it receives which is
meant for its consumer, and writes a change to the changelog before it
hands it over;
- every change the queue drops when it is full.

It goes back to the queue only once the state of the consumer covers
them, and once the queue holds nothing before the first change read that
the consumer lacks (`MsgQueue.holdsAChangeBefore()`). It reads the
changelog until then. All three ways back to the queue check what the
queue lacks: the one which finds the first change read in the queue, the
one taken when the read gave nothing, and the one taken when both queues
are empty and nothing is found after the state of the consumer.

- Only the changes a read of the changelog can still give count, for
what the queue lacks and for what it holds before the first change read.
A change the changelog no longer holds, as one of a changelog cleared
for a new generation id, is left out once the consumer has every change
of its replica the changelog holds. Otherwise a read which gives nothing
but replica offline messages would be made again and again, with no
wait.
- A replica offline message in the queue is left out as well: no state
of a consumer ever holds its CSN.
- When the queue holds every change after the state of the consumer,
nothing changes: what it lacks is older than the oldest change the
consumer lacks.
- For the handler of a peer replication server, what the domain held
includes changes of replicas connected to other replication servers,
which the domain never hands it. The handler reads those the state of
the peer lacks from the changelog once, up to what the domain held then,
before it goes back to its queue. The peer gets them from the
replication servers they come from as well. Its changelog keeps one copy
of each, and with #1224 fixed below, its cookie based persistent
searches are not told of them twice either.
- Changes the domain does not hand a handler on purpose, to a directory
server with another generation id or being initialized, are not
recorded. Such a directory server ends that status only with a new
session (`StatusMachine` lets neither status go to normal): after its
import (`ReplicationDomain`, `broker.reStart(false)`), or once the
changelog is cleared for its generation id, when the replication server
closes its session (`DataServerHandler.changeStatusForResetGenId()`).

### Cookie based persistent searches (#1224)

`FileChangelogDB.publishUpdateMsg()` notified the cookie based
persistent searches of `cn=changelog` of every change it was given, also
of one its replica DB did not take: `Log.append()` drops a record whose
key is not after the newest one, which is what a change the changelog
already holds is. A change a replication server receives twice reached
those searches twice. `LogFile.append()`, `Log.append()` and
`FileReplicaDB.add()` now tell whether the record was appended, and
`publishUpdateMsg()` notifies the persistent searches only of a change
the replica DB took.

Not addressed here, from the comment on #1206:

- case 4: the span all three streams skipped after the changes sent
again at the reconnection of 23:08:57;
- `RemotePendingChanges.putRemoteUpdate()` taking a change sent again
which the state of the directory server covers.

## Tests

`MessageHandlerCatchUpTest` (new) drives a handler no domain feeds from
one thread, as the writer of a session does, over the changelog of a
real replication server. A change arrives the way
`ReplicationServerDomain.put()` hands it over: written to the changelog,
then given to the queue. For the reads, the domain of the handler can
hand it a change right after it has read the changelog, or once a read
gave nothing.

| # | Test | What it pins |
|---|---|---|
| 1 | `aChangeTheFullQueueDroppedIsSentWhenAnOlderOneCameInAfterIt` |
The queue dropped a change of A, then took in a change of B with an
older CSN, in the room the drop left: the consumer is sent the dropped
change. Red on master. |
| 2 |
`aChangeWhichReachedTheDomainBeforeTheHandlerIsSentWhenAnOlderOneIsQueued`
| The queue never had the changes of A which reached the domain before
it, then got a change of B with an older CSN and a later change of A:
the consumer is sent them. Red on master. |
| 3 |
`aChangeOfAReplicaTheQueueIsNeverHandedIsSentWhenAnOlderOneIsQueued` |
The same with no later change of A. Red on master. |
| 4 | `aChangeWhichArrivedWhileTheChangelogWasReadIsSent` | A change of
B with an older CSN than the first change read arrives once the
changelog is read: the consumer is sent it. Red on master. |
| 5 | `aChangeTheFullQueueDroppedWhileTheChangelogWasReadIsSent` | A
change the full queue drops at once arrives once a read gave nothing:
the consumer is sent it. Red on master. |
| 6 | `aHandlerWhoseQueueHoldsWhatTheConsumerLacksFollowsTheQueueAgain`
| The queue of test 1, for a consumer which has the dropped change: the
handler goes back to the queue. |
| 7 | `aDroppedChangeTheChangelogNoLongerHoldsDoesNotHoldTheHandlerBack`
| The queue dropped a change the changelog no longer holds, of a replica
which writes nothing more: the handler goes back to the queue. |
| 8 | `aReplicaOfflineMessageDoesNotHoldTheHandlerBack` | A replica
offline message in the queue before the first change read does not keep
the handler from it, also once its replica has written a change the
consumer lacks. |
| 9 | `aQueuedChangeTheChangelogNoLongerHoldsDoesNotHoldTheHandlerBack`
| The queue holds a change the changelog no longer holds, and the
changelog gives nothing but a replica offline message: the handler goes
back to the queue instead of reading the changelog again and again. Red
before round 2. |
| 10 | `aDroppedReplicaOfflineMessageDoesNotHoldTheHandlerBack` | The
full queue dropped a replica offline message, for a consumer which has
every change of its replica: the handler goes back to the queue. |
| 11 | `aHandlerWhichCaughtUpFromTheChangelogFollowsTheQueueAgain` | A
handler which has been handed a change catches up from the changelog
until a read gives nothing: the change handed after that read is served
from the queue. |
| 12 |
`aHandlerWhichReadTheChangesOfAReplicaItIsNeverHandedFollowsTheQueueAgain`
| The domain held a change of a replica the queue is never handed when
it handed the queue its first change: the handler reads it from the
changelog, then goes back to the queue once a read gives nothing. Red on
master. |

For #1224:

| Test | What it pins |
|---|---|
|
`ChangelogBackendTestCase.aChangeTheChangelogAlreadyHoldsIsNotNotifiedAgainToACookieBasedPersistentSearch`
| A directory server publishes a change, the same change again, then a
newer one: a changes-only cookie based persistent search of
`cn=changelog` is notified of each once. Red on master: `[csn1, csn1,
csn2]`. |
| `LogTest.appendTellsWhetherTheRecordWasAppended` | A record whose key
is not after the newest one is not appended, whether the head log file
is rotated or not, and `append()` says so. |

Mutants of `MessageHandler` and `MsgQueue`, each run against
`MessageHandlerCatchUpTest`:

| Mutant | Caught by |
|---|---|
| the changes the queue drops are not recorded | 1, 5 |
| what the domain held when it handed the queue its first change is not
recorded | 2, 3, 12 |
| what the domain holds is recorded at every change handed to the queue
| 6, 7, 8, 10, 11, 12 |
| a change the queue dropped counts when the changelog no longer holds
it | 7, 10 |
| a change of the queue before the first change read counts when the
changelog no longer holds it | 9 |
| a change counts when the consumer has every change of its replica the
changelog holds | 10 |
| the queue before the first change read is not checked | 4 |
| a replica offline message counts as a change the consumer lacks | 8 |
| the handler never goes back to the queue on finding there the first
change read | 6, 7, 8, 9, 10 |
| the handler goes back to the queue when the read gave nothing,
whatever the queue lacks | 5 |
| a handler which has been handed a change never goes back to the queue
when a read gives nothing | 11, 12 |
| the third way back to the queue is not checked | none: no state the
handler reaches lets the check decide on that way |

Mutants of the fix of #1224, each run against its two tests:

| Mutant | Caught by |
|---|---|
| the persistent searches are notified whatever the replica DB took |
`ChangelogBackendTestCase` |
| the replica DB tells it added a change the log did not append |
`ChangelogBackendTestCase` |
| the log file tells it appended a record breaking the key ordering |
both |
| the log tells it appended a record when its head log file is not to be
rotated | both |
| the log tells it appended a record when it gave up a rotation |
`LogTest` |

## Verification

Run directly with TestNG, JDK 21, macOS:

- `MessageHandlerCatchUpTest`: 12 tests, 0 failures. On master 6 of them
fail (1-5, 12); at the head of round 1, `ad3e21ad6d`, 1 fails (9).
- The two tests of #1224: 0 failures. The one of
`ChangelogBackendTestCase` fails on master.

In the reactor (`-Pprecommit verify -pl opendj-server-legacy -am`,
javadoc included), JDK 27, macOS, `org/opends/server/replication/**` and
`ChangelogBackendTestCase`: 3879 tests, 0 failures, 0 errors, 0 skipped.
Among them is `MissedUpdateRecoveryTest` (#963), which drives the same
handler through real sessions: 8 tests, 0 failures.

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

@@ -13,7 +13,7 @@
 *
 * Copyright 2009-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

@@ -81,6 +81,19 @@
  /** Specifies whether the consumer is following the producer (is not late). */
  private boolean following;
  /**
   * For each replica, the newest of its changes the queue may lack: the changes the domain held
   * when it handed the queue its first change - the ones before never reached it - and the changes
   * the queue dropped when it was full. Once the handler is registered, the domain hands it every
   * change it receives which is meant for its consumer: a peer replication server is handed only
   * the changes of the directory servers connected to this one, and a directory server nothing
   * while its status is bad generation id or full update, which it leaves only with a new session.
   * The consumer must have those the changelog still holds before it is served from the queue
   * again, see {@link #consumerHasWhatTheQueueLacks(ServerState)}. Guarded by {@link #msgQueue}.
   */
  private final ServerState queueLacksUpTo = new ServerState();
  /** Whether the queue has been handed a change yet. Guarded by {@link #msgQueue}. */
  private boolean queueHandedAChange;
  /**
   * The state of the domain at the previous wait of {@link #getNextMessage()} on an empty queue,
   * or {@code null} when the queue was not empty then. A change it holds has had a whole wait to
   * reach the queue, see {@link #stopFollowingWhenChangesAreMissing()}. Guarded by
@@ -146,17 +159,67 @@
        msgQueue.notifyAll();
      }

      if (!queueHandedAChange)
      {
        // the changes the domain held before this one never reached the queue. The domain writes
        // a change to the changelog before it hands it over, so this one is counted as well
        queueHandedAChange = true;
        queueLacksUpTo.update(replicationServerDomain.getLatestServerState());
      }
      msgQueue.add(update);

      // TODO : size should be configurable and larger than max-receive-queue-size
      while (isMsgQueueAboveThreshold())
      {
        following = false;
        msgQueue.removeFirst();
        queueLacksUpTo.update(msgQueue.removeFirst().getCSN());
      }
    }
  }

  /**
   * Whether the consumer has every change the queue may lack which the changelog still holds - see
   * {@link #queueLacksUpTo} and {@link #consumerLacksFromTheChangelog(CSN, ServerState)}.
   * <p>
   * Must be called while holding the {@link #msgQueue} monitor.
   *
   * @param inTheChangelog
   *          the newest change of each replica the changelog holds
   * @return {@code true} when the queue lacks nothing the consumer may still be sent
   */
  private boolean consumerHasWhatTheQueueLacks(ServerState inTheChangelog)
  {
    for (CSN lacked : queueLacksUpTo)
    {
      if (consumerLacksFromTheChangelog(lacked, inTheChangelog))
      {
        return false;
      }
    }
    return true;
  }

  /**
   * Whether the consumer lacks the given change, and a read of the changelog can still give it the
   * changes of that replica it lacks. A change the changelog no longer holds, such as one of a
   * changelog cleared for a new generation id, cannot be read from it: once the consumer has every
   * change of its replica the changelog holds, no read gives anything more of that replica, and such
   * a change must not keep this handler from its queue.
   *
   * @param csn
   *          the CSN of the change
   * @param inTheChangelog
   *          the newest change of each replica the changelog holds
   * @return {@code true} when the consumer lacks the change and the changelog holds a change of its
   *         replica the consumer lacks
   */
  private boolean consumerLacksFromTheChangelog(CSN csn, ServerState inTheChangelog)
  {
    final CSN newest = inTheChangelog.getCSN(csn.getServerId());
    final CSN sent = serverState.getCSN(csn.getServerId());
    return !serverState.cover(csn) && newest != null && (sent == null || newest.isNewerThan(sent));
  }

  private boolean isMsgQueueAboveThreshold()
  {
    final long count = msgQueue.count();
@@ -339,8 +402,11 @@
            synchronized (msgQueue)
            {
              // Ensure we are below threshold so this server will follow the
              // msgQueue without fearing the msgQueue gets trimmed
              if (isMsgQueueBelowThreshold())
              // msgQueue without fearing the msgQueue gets trimmed.
              // The changelog was read without the lock: a change written since may have been
              // dropped from the queue already
              if (isMsgQueueBelowThreshold()
                  && consumerHasWhatTheQueueLacks(replicationServerDomain.getLatestServerState()))
              {
                following = true;
              }
@@ -350,8 +416,19 @@
          {
            /*
             * if the first change in the lateQueue is also on the regular queue,
             * and the regular queue holds every other change the consumer lacks,
             * we can resume the processing from the regular queue
             * -> set following to true and empty the lateQueue.
             * The regular queue is ordered by CSN, so holding the first change of the lateQueue does
             * not mean holding the changes after it: a change which reaches the queue late with an
             * older CSN - relayed by another replication server, or published again by a recovery -
             * goes in front of the changes of another replica which the queue dropped before it came,
             * or never had as they reached the domain before this handler did. Nor does it mean that
             * the changes before it are ones the consumer has: one such late change may have reached
             * the queue since the changelog was read. Served from the queue then, the consumer would
             * not be sent those changes, and its state would move past them for good - see #1206.
             * Only the changes the changelog can still give count: a read which gives nothing but
             * replica offline messages would otherwise be made again and again, with no wait.
             */
            UpdateMsg msg = lateQueue.first();
            synchronized (msgQueue)
@@ -365,7 +442,10 @@
               }
              }

              if (msgQueue.contains(msg))
              final ServerState inTheChangelog = replicationServerDomain.getLatestServerState();
              if (msgQueue.contains(msg)
                  && consumerHasWhatTheQueueLacks(inTheChangelog)
                  && !msgQueue.holdsAChangeBefore(msg, csn -> consumerLacksFromTheChangelog(csn, inTheChangelog)))
              {
                /* we finally catch up with the regular queue */
                following = true;
@@ -395,8 +475,11 @@

            if (lateQueue.isEmpty() && msgQueue.isEmpty())
            {
              // no change left after the state of the consumer means it has what the queue lacks,
              // unless the changelog could not be read
              CSN nextChange = findOldestCSNFromReplicaDBs();
              if (nextChange == null)
              if (nextChange == null
                  && consumerHasWhatTheQueueLacks(replicationServerDomain.getLatestServerState()))
              {
                following = true;
              }

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

@@ -20,6 +20,7 @@
import static org.opends.messages.ReplicationMessages.*;

import java.util.TreeMap;
import java.util.function.Predicate;

import net.jcip.annotations.ThreadSafe;

@@ -178,6 +179,32 @@
    }
  }

  /**
   * Returns whether this queue holds, before the given UpdateMsg, an UpdateMsg which carries the
   * CSN of a change and matches the given condition: a replica offline message, for one, does not
   * carry the CSN of a change, and no state ever covers its CSN.
   *
   * @param msg
   *          the UpdateMsg the ones to check come before
   * @param condition
   *          the condition the CSN of a change must match
   * @return {@code true} if a change of this queue before the given UpdateMsg matches the condition
   */
  public boolean holdsAChangeBefore(UpdateMsg msg, Predicate<CSN> condition)
  {
    synchronized (lock)
    {
      for (UpdateMsg before : map.headMap(msg.getCSN()).values())
      {
        if (before.contributesToDomainState() && condition.test(before.getCSN()))
        {
          return true;
        }
      }
      return false;
    }
  }

  /** Removes all UpdateMsg form this queue. */
  public void clear()
  {

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java

@@ -923,9 +923,12 @@
    final Pair<FileReplicaDB, Boolean> pair = getOrCreateReplicaDB(baseDN,
        csn.getServerId(), replicationServer);
    final FileReplicaDB replicaDB = pair.getFirst();
    replicaDB.add(updateMsg);

    ChangelogBackend.getInstance().notifyCookieEntryAdded(baseDN, updateMsg);
    if (replicaDB.add(updateMsg))
    {
      // a change the replica DB does not take - one it holds already, sent again - is not in the
      // changelog a second time, and the persistent searches were told of it already
      ChangelogBackend.getInstance().notifyCookieEntryAdded(baseDN, updateMsg);
    }

    final ChangeNumberIndexer indexer = cnIndexer.get();
    if (indexer != null)

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileReplicaDB.java

@@ -155,10 +155,12 @@
   *
   * @param updateMsg
   *          The update message to add.
   * @return {@code true} if the message was added, {@code false} if it was not: this replica DB
   *         holds it already or holds a newer change, see {@link Log#append(Record)}
   * @throws ChangelogException
   *           If an error occurs when trying to add the message.
   */
  void add(final UpdateMsg updateMsg) throws ChangelogException
  boolean add(final UpdateMsg updateMsg) throws ChangelogException
  {
    if (shutdown.get())
    {
@@ -167,7 +169,10 @@
              .toString(), String.valueOf(baseDN), String.valueOf(serverId)));
    }

    log.append(Record.from(updateMsg.getCSN(), updateMsg));
    if (!log.append(Record.from(updateMsg.getCSN(), updateMsg)))
    {
      return false;
    }

    final CSNLimits limits = csnLimits;
    final boolean updateNew = limits.newestCSN == null || limits.newestCSN.isOlderThan(updateMsg.getCSN());
@@ -178,6 +183,7 @@
          updateOld ? updateMsg.getCSN() : limits.oldestCSN,
          updateNew ? updateMsg.getCSN() : limits.newestCSN);
    }
    return true;
  }

  /**

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java

@@ -12,6 +12,7 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server.changelog.file;

@@ -411,10 +412,12 @@
   *
   * @param record
   *          The record to add.
   * @return {@code true} if the record was appended, {@code false} if it was not: its key is not
   *         higher than the key of the last record added, or this log is closed
   * @throws ChangelogException
   *           If an error occurs while adding the record to the log.
   */
  public void append(final Record<K, V> record) throws ChangelogException
  public boolean append(final Record<K, V> record) throws ChangelogException
  {
    // Fast-path - assume that no rotation is needed and use shared lock.
    sharedLock.lock();
@@ -422,13 +425,12 @@
    {
      if (isClosed)
      {
        return;
        return false;
      }
      LogFile<K, V> headLogFile = getHeadLogFile();
      if (!mustRotate(headLogFile))
      {
        headLogFile.append(record);
        return;
        return headLogFile.append(record);
      }
    }
    finally
@@ -442,13 +444,13 @@
    {
      if (isClosed)
      {
        return;
        return false;
      }
      LogFile<K, V> headLogFile = getHeadLogFile();
      if (headLogFile.appendWouldBreakKeyOrdering(record))
      {
        // abort rotation
        return;
        return false;
      }
      if (mustRotate(headLogFile))
      {
@@ -457,7 +459,7 @@
        rotateHeadLogFile();
        headLogFile = getHeadLogFile();
      }
      headLogFile.append(record);
      return headLogFile.append(record);
    }
    finally
    {

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java

@@ -253,10 +253,12 @@
   *
   * @param record
   *          The record to add.
   * @return {@code true} if the record was appended, {@code false} if its key is not higher than
   *         the key of the last record added
   * @throws ChangelogException
   *           If the record can't be added to the log.
   */
  void append(final Record<K, V> record) throws ChangelogException
  boolean append(final Record<K, V> record) throws ChangelogException
  {
    checkLogIsEnabledForWrite();
    exclusiveLock.lock();
@@ -264,10 +266,11 @@
    {
      if (appendWouldBreakKeyOrdering(record))
      {
        return;
        return false;
      }
      writer.write(record);
      newestRecord = record;
      return true;
    }
    finally
    {

 opendj-server-legacy/src/test/java/org/opends/server/backends/ChangelogBackendTestCase.java

@@ -34,6 +34,7 @@

import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -43,6 +44,7 @@
import java.util.Set;
import java.util.SortedSet;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;

import org.assertj.core.api.SoftAssertions;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -58,12 +60,16 @@
import org.opends.server.backends.ChangelogBackend.ChangeNumberRange;
import org.opends.server.controls.EntryChangelogNotificationControl;
import org.opends.server.controls.ExternalChangelogRequestControl;
import org.opends.server.controls.PersistentSearchChangeType;
import org.opends.server.controls.PersistentSearchControl;
import org.opends.server.core.BackendConfigManager;
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.ModifyDNOperation;
import org.opends.server.core.ModifyDNOperationBasis;
import org.opends.server.core.ModifyOperation;
import org.opends.server.core.PersistentSearch;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.internal.InternalSearchListener;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.internal.Requests;
import org.opends.server.protocols.internal.SearchRequest;
@@ -103,6 +109,7 @@
import org.opends.server.types.Operation;
import org.opends.server.types.SearchFilter;
import org.opends.server.types.SearchResultEntry;
import org.opends.server.types.SearchResultReference;
import org.opends.server.util.LDIFWriter;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
@@ -356,6 +363,104 @@
  }

  /**
   * A change the changelog already holds can reach its replication server again: a directory
   * server publishes it again, or a peer replication server sends it from its own changelog. The
   * replica DB does not take it, and a cookie based persistent search must not be notified of it a
   * second time.
   */
  @Test
  public void aChangeTheChangelogAlreadyHoldsIsNotNotifiedAgainToACookieBasedPersistentSearch() throws Exception
  {
    String test = "CookiePersistentSearchNotifiedOnce";
    debugInfo(test, "Starting test");

    final List<CSN> notified = new CopyOnWriteArrayList<>();
    final InternalSearchOperation searchOp =
        startChangesOnlyCookieBasedPersistentSearch("(targetdn=*" + test + "*,o=test)", notified);
    try
    {
      final CSN[] csns = generateCSNs(2, server1);
      final UpdateMsg sentAgain = generateDeleteMsg(server1, csns[0], test, 1);
      publishUpdateMessagesInOTest(test, false, sentAgain, sentAgain, generateDeleteMsg(server1, csns[1], test, 2));

      // the replication server handles the changes of a session in order: once the last one is
      // notified, the one sent again has been handled
      TestTimer timer = new TestTimer.Builder()
        .maxSleep(30, SECONDS)
        .sleepTimes(100, MILLISECONDS)
        .toTimer();
      timer.repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          assertThat(notified).contains(csns[1]);
        }
      });
      assertThat(notified).as("the changes the persistent search was notified of")
          .containsExactly(csns[0], csns[1]);
    }
    finally
    {
      cancelPersistentSearch(searchOp);
    }
    debugInfo(test, "Ending test successfully");
  }

  /**
   * Starts a persistent search of the changes only, in cookie mode, which records the CSN of every
   * change it is notified of.
   */
  private InternalSearchOperation startChangesOnlyCookieBasedPersistentSearch(String filterString,
      final List<CSN> notified) throws Exception
  {
    final SearchRequest request = newSearchRequest(filterString)
        .addControl(createCookieControl(""))
        .addControl(new PersistentSearchControl(EnumSet.allOf(PersistentSearchChangeType.class), true, false));
    final InternalSearchOperation searchOp = connection.processSearch(request, new InternalSearchListener()
    {
      @Override
      public void handleInternalSearchEntry(InternalSearchOperation searchOperation, SearchResultEntry searchEntry)
      {
        notified.add(new CSN(getAttributeValue(searchEntry, "replicationcsn")));
      }

      @Override
      public void handleInternalSearchReference(InternalSearchOperation searchOperation,
          SearchResultReference searchReference)
      {
        // the changelog holds no reference
      }
    });
    assertThat(findPersistentSearch(searchOp)).as(searchOp.getErrorMessage().toString()).isNotNull();
    return searchOp;
  }

  private void cancelPersistentSearch(InternalSearchOperation searchOp)
  {
    final PersistentSearch pSearch = findPersistentSearch(searchOp);
    if (pSearch != null)
    {
      pSearch.cancel();
    }
  }

  private PersistentSearch findPersistentSearch(InternalSearchOperation searchOp)
  {
    for (PersistentSearch pSearch : ChangelogBackend.getInstance().getPersistentSearches())
    {
      // the persistent search holds the operation the backend processed, which wraps the given one
      final Operation op = pSearch.getSearchOperation();
      if (op.getClientConnection() == searchOp.getClientConnection()
          && op.getOperationID() == searchOp.getOperationID())
      {
        return pSearch;
      }
    }
    return null;
  }

  /**
   * This test enables a second suffix. It will break all tests using search on
   * one suffix if run before them, so it is necessary to add them as
   * dependencies.

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

New file
@@ -0,0 +1,868 @@
/*
 * 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.server;

import static org.assertj.core.api.Assertions.assertThat;
import static org.forgerock.opendj.ldap.ModificationType.REPLACE;
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;

import java.util.ArrayList;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.DBCursor;
import org.opends.server.types.Attributes;
import org.opends.server.types.Modification;
import org.opends.server.util.TimeThread;
import org.testng.annotations.Test;

/**
 * A handler which is behind reads the changes for its consumer from the changelog, and goes back
 * to its in-memory queue once the queue holds what the consumer lacks - see issue #1206. The queue
 * is ordered by CSN and drops its oldest changes when it is full, so holding the oldest change the
 * consumer lacks does not mean holding all of them. A change which reaches the queue late with an
 * older CSN - relayed by another replication server, or published again by a recovery - can be
 * the oldest change the consumer lacks while the queue lacks newer changes of another replica: the
 * ones it dropped before that change came, or the ones which reached the domain before the handler
 * did. Served from the queue then, the consumer would be given the changes of that replica which
 * the queue still holds, and its state would move past the ones in between for good.
 * <p>
 * Each test drives a handler which no domain feeds, from a single thread, as the writer of a
 * session does. A change "arrives" the way {@code ReplicationServerDomain.put()} hands it over to
 * a registered handler: written to the changelog, then given to the queue. A change written to the
 * changelog only stands for one which reached the domain before the handler was registered.
 * Replica {@link #A} stands for a directory server connected to the replication server, replica
 * {@link #B} for one whose changes another replication server relays late: the changes of B carry
 * older CSNs than the changes of A which arrive with them.
 */
@SuppressWarnings("javadoc")
public class MessageHandlerCatchUpTest extends ReplicationTestCase
{
  private static final int RS_ID = 120;
  private static final int A = 121;
  private static final int B = 122;
  private static final int C = 123;
  private static final int WINDOW_SIZE = 100;
  /** A queue no test fills. */
  private static final int LARGE_QUEUE_SIZE = 100;
  /** How long the handler is given to come up with a change it has: it reads them from memory. */
  private static final long NEXT_MESSAGE_TIMEOUT_MS = 5000;

  /**
   * The case of the description of #1206, on a handler connected for long: the queue, full, has
   * dropped a change of A the consumer lacks, and a change of B with an older CSN has come in
   * after, into the room the drop left. The change of B is the oldest change the consumer lacks,
   * and it is in the queue - but the change of A is not, and must still be sent.
   */
  @Test
  public void aChangeTheFullQueueDroppedIsSentWhenAnOlderOneCameInAfterIt() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerDroppedDb");
      final DroppingQueue queue = new DroppingQueue(baseDN, TimeThread.getTime() - 60000, A, true);
      final MessageHandler handler = newHandler(replicationServer, baseDN, queue.size, queue.consumerState);

      queue.arriveIn(replicationServer, handler);

      assertThat(csnsOf(nextMessages(handler, queue.changesTheConsumerLacks.size())))
          .as("the consumer was not sent every change it lacks, in order")
          .containsExactlyElementsOf(csnsOf(queue.changesTheConsumerLacks));
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The case of the reconnections of #1206: a new handler, whose queue does not hold the changes
   * of A which reached the domain before it did. A change of B with an older CSN then arrives with
   * a change of A: the change of B is the oldest change the consumer lacks and it is in the queue,
   * but the earlier changes of A are not, and must still be sent.
   */
  @Test
  public void aChangeWhichReachedTheDomainBeforeTheHandlerIsSentWhenAnOlderOneIsQueued() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerBeforeDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
      final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
      final List<UpdateMsg> earlierChangesOfA = changesOfA(baseDN, t, 1, 3);
      final DeleteMsg a4 = newDeleteMsg(baseDN, "a4", new CSN(t + 24, 4, A));
      // the consumer has the changes of B up to b0, and none of A
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(b0));

      publishToChangelogOnly(replicationServer, baseDN, b0);
      publishToChangelogOnly(replicationServer, baseDN, earlierChangesOfA);
      arrive(replicationServer, handler, a4);
      arrive(replicationServer, handler, b1);

      final List<UpdateMsg> expected = new ArrayList<>();
      expected.add(b1);
      expected.addAll(earlierChangesOfA);
      expected.add(a4);
      assertThat(csnsOf(nextMessages(handler, expected.size())))
          .as("the consumer was not sent every change it lacks, in order")
          .containsExactlyElementsOf(csnsOf(expected));
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The same as {@link #aChangeWhichReachedTheDomainBeforeTheHandlerIsSentWhenAnOlderOneIsQueued()}
   * for a replica which writes nothing more once the handler is there: the queue is handed changes
   * of B only, never one of A, and the earlier changes of A must still be sent.
   */
  @Test
  public void aChangeOfAReplicaTheQueueIsNeverHandedIsSentWhenAnOlderOneIsQueued() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerQuietDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
      final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
      final List<UpdateMsg> earlierChangesOfA = changesOfA(baseDN, t, 1, 3);
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(b0));

      publishToChangelogOnly(replicationServer, baseDN, earlierChangesOfA);
      arrive(replicationServer, handler, b0);
      arrive(replicationServer, handler, b1);

      final List<UpdateMsg> expected = new ArrayList<>();
      expected.add(b1);
      expected.addAll(earlierChangesOfA);
      assertThat(csnsOf(nextMessages(handler, expected.size())))
          .as("the consumer was not sent every change it lacks, in order")
          .containsExactlyElementsOf(csnsOf(expected));
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * A change of B with an older CSN arrives while the handler reads the changelog, too late for the
   * read: the first change read is a change of A, and it is in the queue - but the change of B
   * before it is not one the consumer has, and must still be sent.
   */
  @Test
  public void aChangeWhichArrivedWhileTheChangelogWasReadIsSent() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ChangeAfterTheRead domain = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerWhileReadDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
      final DeleteMsg a0 = newDeleteMsg(baseDN, "a0", new CSN(t + 10, 0, A));
      final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 2);
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(b0, a0));
      domain = new ChangeAfterTheRead(replicationServer, handler, b1);

      arrive(replicationServer, handler, b0);
      arrive(replicationServer, handler, a0);
      for (UpdateMsg msg : changesOfA)
      {
        arrive(replicationServer, handler, msg);
      }

      final List<UpdateMsg> expected = new ArrayList<>(changesOfA);
      expected.add(b1);
      assertThat(csnsOf(nextMessages(handler, expected.size())))
          .as("the consumer was not sent every change it lacks")
          .containsExactlyElementsOf(csnsOf(expected));
    }
    finally
    {
      deregister(domain);
      removeQuietly(replicationServer);
    }
  }

  /**
   * The same as {@link #aChangeWhichArrivedWhileTheChangelogWasReadIsSent()} for a handler whose
   * consumer has everything the changelog held: the read gives nothing, and the change of B, which
   * arrives after it, is the oldest change of the full queue, which drops it at once. The queue
   * holds nothing the consumer lacks, but the change of B must still be sent.
   */
  @Test
  public void aChangeTheFullQueueDroppedWhileTheChangelogWasReadIsSent() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ChangeAfterTheRead domain = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerDroppedWhileReadDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
      final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 5);
      final ServerState consumerState = stateOf(b0, changesOfA.get(changesOfA.size() - 1));
      // full with the changes of A: one more change makes it drop its oldest one
      final MessageHandler handler = newHandler(replicationServer, baseDN, changesOfA.size(), consumerState);
      domain = new ChangeAfterTheRead(replicationServer, handler, b1);

      for (UpdateMsg msg : changesOfA)
      {
        arrive(replicationServer, handler, msg);
      }

      assertThat(csnsOf(nextMessages(handler, 1)))
          .as("the consumer was not sent the change the queue dropped")
          .containsExactly(b1.getCSN());
    }
    finally
    {
      deregister(domain);
      removeQuietly(replicationServer);
    }
  }

  /**
   * The same queue as {@link #aChangeTheFullQueueDroppedIsSentWhenAnOlderOneCameInAfterIt()}, for
   * a consumer which already has the change of A the queue dropped: the queue holds everything the
   * consumer lacks, and the handler serves it from the queue again, as soon as it finds there the
   * oldest change the consumer lacks.
   */
  @Test
  public void aHandlerWhoseQueueHoldsWhatTheConsumerLacksFollowsTheQueueAgain() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerFollowsDb");
      final DroppingQueue queue = new DroppingQueue(baseDN, TimeThread.getTime() - 60000, A, true);
      final ServerState consumerState = queue.consumerState.duplicate();
      consumerState.update(queue.dropped.getCSN());
      final MessageHandler handler = newHandler(replicationServer, baseDN, queue.size, consumerState);

      queue.arriveIn(replicationServer, handler);

      final List<UpdateMsg> expected = new ArrayList<>(queue.changesTheConsumerLacks);
      expected.remove(queue.dropped);
      assertServedFromTheQueueAfterTheFirst(handler, expected);
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The queue dropped a change the changelog no longer holds, as after the changelog was cleared
   * for a new generation id, of a replica which writes nothing more: nothing can send it to the
   * consumer, and it must not keep the handler from its queue, which holds everything else the
   * consumer lacks.
   */
  @Test
  public void aDroppedChangeTheChangelogNoLongerHoldsDoesNotHoldTheHandlerBack() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerNoLongerHeldDb");
      final DroppingQueue queue = new DroppingQueue(baseDN, TimeThread.getTime() - 60000, C, false);
      final MessageHandler handler = newHandler(replicationServer, baseDN, queue.size, queue.consumerState);

      queue.arriveIn(replicationServer, handler);

      final List<UpdateMsg> expected = new ArrayList<>(queue.changesTheConsumerLacks);
      expected.remove(queue.dropped);
      assertServedFromTheQueueAfterTheFirst(handler, expected);
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * A replica offline message carries the CSN of no change, and no state of a consumer ever holds
   * it: one in the queue before the first change the consumer lacks must not keep the handler from
   * its queue, even once its replica, back online, has written a change the consumer lacks.
   */
  @Test
  public void aReplicaOfflineMessageDoesNotHoldTheHandlerBack() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerOfflineDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg c0 = newDeleteMsg(baseDN, "c0", new CSN(t, 0, C));
      final DeleteMsg a1 = newDeleteMsg(baseDN, "a1", new CSN(t + 1, 1, A));
      final DeleteMsg a2 = newDeleteMsg(baseDN, "a2", new CSN(t + 3, 2, A));
      final DeleteMsg c3 = newDeleteMsg(baseDN, "c3", new CSN(t + 4, 3, C));
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(c0, a1));

      arrive(replicationServer, handler, c0);
      arrive(replicationServer, handler, a1);
      // a replica offline message is not written to the changelog, its CSN goes to the state of
      // the replica instead: ReplicationServerDomain.publishUpdateMsg()
      handler.add(new ReplicaOfflineMsg(new CSN(t + 2, 0, C)));
      arrive(replicationServer, handler, a2);
      arrive(replicationServer, handler, c3);

      final List<UpdateMsg> expected = new ArrayList<>();
      expected.add(a2);
      expected.add(c3);
      assertServedFromTheQueueAfterTheFirst(handler, expected);
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The queue still holds a change the changelog no longer holds, as after the changelog was
   * cleared for a new generation id, and the changelog gives the handler nothing but the replica
   * offline message of a replica which wrote since: no read of the changelog can give the consumer
   * that change, and the handler must not read the changelog again and again for it. It goes back
   * to its queue, leaving out what the queue holds before the offline message, that change too.
   */
  @Test
  public void aQueuedChangeTheChangelogNoLongerHoldsDoesNotHoldTheHandlerBack() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerQueuedNoLongerHeldDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg c0 = newDeleteMsg(baseDN, "c0", new CSN(t, 0, C));
      final DeleteMsg a1 = newDeleteMsg(baseDN, "a1", new CSN(t + 1, 1, A));
      final CSN offlineC = new CSN(t + 2, 0, C);
      final DeleteMsg a2 = newDeleteMsg(baseDN, "a2", new CSN(t + 3, 2, A));
      // the consumer has every change the changelog holds
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, stateOf(c0));

      publishToChangelogOnly(replicationServer, baseDN, c0);
      // the change of a changelog cleared since
      handler.add(a1);
      // every read of the changelog gives it, as no state of a consumer ever holds its CSN
      replicationServer.getChangelogDB().getReplicationDomainDB().notifyReplicaOffline(baseDN, offlineC);

      assertThat(csnsOf(nextMessages(handler, 1)))
          .as("the handler kept reading the changelog for a change no read can give")
          .containsExactly(offlineC);
      assertThat(handler.isFollowing()).isTrue();

      arrive(replicationServer, handler, a2);
      final List<Sent> sent = nextMessages(handler, 1);
      assertThat(csnsOf(sent)).containsExactly(a2.getCSN());
      assertThat(sent.get(0).fromTheChangelog).isFalse();
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The full queue of a handler drops a replica offline message, which the domain hands to the
   * handler of a peer replication server: the consumer has every change of that replica the
   * changelog holds, and no state of a consumer ever holds the CSN of the message, so it must not
   * keep the handler from its queue, which holds every change the consumer lacks.
   */
  @Test
  public void aDroppedReplicaOfflineMessageDoesNotHoldTheHandlerBack() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerDroppedOfflineDb");
      final long t = TimeThread.getTime() - 60000;
      final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 5);
      // full with the changes of A: they make it drop the offline message, its oldest one
      final MessageHandler handler = newHandler(replicationServer, baseDN, changesOfA.size(), stateOf(b0));

      publishToChangelogOnly(replicationServer, baseDN, b0);
      handler.add(new ReplicaOfflineMsg(new CSN(t + 5, 0, B)));
      for (UpdateMsg msg : changesOfA)
      {
        arrive(replicationServer, handler, msg);
      }

      assertServedFromTheQueueAfterTheFirst(handler, changesOfA);
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * A handler which has been handed a change catches up from the changelog until a read gives
   * nothing, and goes back to its queue: the change handed to it after that read is served from
   * the queue.
   */
  @Test
  public void aHandlerWhichCaughtUpFromTheChangelogFollowsTheQueueAgain() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ChangeAfterTheRead domain = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerCaughtUpDb");
      final long t = TimeThread.getTime() - 60000;
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 3);
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, new ServerState());
      domain = new ChangeAfterTheRead(replicationServer, handler, changesOfA.get(2), true);

      publishToChangelogOnly(replicationServer, baseDN, changesOfA.get(0));
      arrive(replicationServer, handler, changesOfA.get(1));

      assertServedFromTheQueueAfterAReadGaveNothing(handler, changesOfA);
    }
    finally
    {
      deregister(domain);
      removeQuietly(replicationServer);
    }
  }

  /**
   * The same as {@link #aHandlerWhichCaughtUpFromTheChangelogFollowsTheQueueAgain()} for a handler
   * which is never handed the changes of a replica, as the handler of a peer replication server is
   * never handed the changes of the replicas connected to other replication servers: the domain held
   * a change of B when it handed the queue its first change, so the handler reads it from the
   * changelog, although it finds in its queue the first change it reads, then goes back to its queue.
   */
  @Test
  public void aHandlerWhichReadTheChangesOfAReplicaItIsNeverHandedFollowsTheQueueAgain() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ChangeAfterTheRead domain = null;
    try
    {
      replicationServer = newReplicationServer("messageHandlerNeverHandedDb");
      final long t = TimeThread.getTime() - 60000;
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, t, 1, 2);
      final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 30, 1, B));
      final MessageHandler handler = newHandler(replicationServer, baseDN, LARGE_QUEUE_SIZE, new ServerState());
      domain = new ChangeAfterTheRead(replicationServer, handler, changesOfA.get(1), true);

      publishToChangelogOnly(replicationServer, baseDN, b1);
      arrive(replicationServer, handler, changesOfA.get(0));

      final List<UpdateMsg> expected = new ArrayList<>();
      expected.add(changesOfA.get(0));
      expected.add(b1);
      expected.add(changesOfA.get(1));
      assertServedFromTheQueueAfterAReadGaveNothing(handler, expected);
    }
    finally
    {
      deregister(domain);
      removeQuietly(replicationServer);
    }
  }

  /**
   * A queue of {@link #size} changes which drops a change the consumer lacks, then takes a change
   * of B with an older CSN in the room the drop left - which only a queue bounded by its size in
   * bytes leaves, as it drops whole changes: one large change, dropped, leaves room for a small one.
   * <p>
   * The queue is not full until it holds 5 changes. Changes {@code b0} and {@code a0}, which the
   * consumer has, show that the handler has been handed changes for long; the large change pushes
   * the queue over its size with the fifth change, so the queue drops {@code b0}, then {@code a0},
   * then the large change. The changes after it are small, so {@code b1} comes into the room the
   * large change left: the queue ends up holding {@code b1} and {@code a2} .. {@code a6}.
   */
  private final class DroppingQueue
  {
    private final List<UpdateMsg> arrivals = new ArrayList<>();
    /** The changes the consumer lacks, in the order it must be sent them. */
    private final List<UpdateMsg> changesTheConsumerLacks = new ArrayList<>();
    private final ServerState consumerState;
    private final UpdateMsg dropped;
    private final boolean droppedInTheChangelog;
    private final int size;

    /**
     * @param droppedReplica
     *          the replica of the change the queue drops
     * @param droppedInTheChangelog
     *          whether the changelog holds the change the queue drops
     */
    private DroppingQueue(DN baseDN, long t, int droppedReplica, boolean droppedInTheChangelog)
        throws Exception
    {
      final DeleteMsg b0 = newDeleteMsg(baseDN, "b0", new CSN(t, 0, B));
      final DeleteMsg a0 = newDeleteMsg(baseDN, "a0", new CSN(t + 10, 0, A));
      final DeleteMsg b1 = newDeleteMsg(baseDN, "b1", new CSN(t + 15, 1, B));
      // a replace of a large value: the size of a delete does not count its DN
      dropped = new ModifyMsg(new CSN(t + 21, 1, droppedReplica), DN.valueOf("cn=large," + baseDN),
          newArrayList(new Modification(REPLACE, Attributes.create("description", "a1".repeat(1000)))),
          "uuid");
      this.droppedInTheChangelog = droppedInTheChangelog;
      final List<UpdateMsg> keptChangesOfA = changesOfA(baseDN, t, 2, 6);

      arrivals.add(b0);
      arrivals.add(a0);
      arrivals.add(dropped);
      arrivals.addAll(keptChangesOfA);
      arrivals.add(b1);
      changesTheConsumerLacks.add(b1);
      changesTheConsumerLacks.add(dropped);
      changesTheConsumerLacks.addAll(keptChangesOfA);
      consumerState = stateOf(b0, a0);

      // room for b1 and a2 .. a6 in bytes, as many changes as the queue ever holds, and not for the
      // large change
      int keptBytes = b1.size();
      for (UpdateMsg msg : keptChangesOfA)
      {
        keptBytes += msg.size();
      }
      size = Math.max(arrivals.size(), (keptBytes + 99) / 100);
      assertThat(dropped.size())
          .as("the large change must push the queue over its size in bytes on its own")
          .isGreaterThan(size * 100);
    }

    private void arriveIn(ReplicationServer replicationServer, MessageHandler handler) throws Exception
    {
      for (UpdateMsg msg : arrivals)
      {
        if (msg != dropped || droppedInTheChangelog)
        {
          arrive(replicationServer, handler, msg);
        }
        else
        {
          handler.add(msg);
        }
      }
    }
  }

  /**
   * The domain of a handler, which hands it a change once it has read the changelog for the first
   * time - or once a read of the changelog gave nothing - before it looks at its queue: the domain
   * writes the changes it receives and hands them over on threads of their own, and nothing keeps
   * one from coming in between. The change comes the way {@code ReplicationServerDomain.put()}
   * hands it over - see {@link #arrive}.
   */
  private final class ChangeAfterTheRead extends ReplicationServerDomain
  {
    private final ReplicationServer replicationServer;
    private final MessageHandler handler;
    private final boolean onceAReadGivesNothing;
    private UpdateMsg change;

    private ChangeAfterTheRead(ReplicationServer replicationServer, MessageHandler handler, UpdateMsg change)
    {
      this(replicationServer, handler, change, false);
    }

    /**
     * @param onceAReadGivesNothing
     *          whether the change comes once a read of the changelog gave nothing, rather than once
     *          the changelog was read for the first time
     */
    private ChangeAfterTheRead(ReplicationServer replicationServer, MessageHandler handler, UpdateMsg change,
        boolean onceAReadGivesNothing)
    {
      super(handler.getBaseDN(), replicationServer);
      this.replicationServer = replicationServer;
      this.handler = handler;
      this.change = change;
      this.onceAReadGivesNothing = onceAReadGivesNothing;
      handler.replicationServerDomain = this;
    }

    @Override
    public String getMonitorInstanceName()
    {
      // apart from the one of the domain of the replication server, which it would replace
      return super.getMonitorInstanceName() + " for " + MessageHandlerCatchUpTest.class.getSimpleName();
    }

    @Override
    public DBCursor<UpdateMsg> getCursorFrom(ServerState startAfterServerState) throws ChangelogException
    {
      final DBCursor<UpdateMsg> read = super.getCursorFrom(startAfterServerState);
      return new DBCursor<UpdateMsg>()
      {
        private boolean gaveSomething;

        @Override
        public UpdateMsg getRecord()
        {
          return read.getRecord();
        }

        @Override
        public boolean next() throws ChangelogException
        {
          final boolean hasNext = read.next();
          gaveSomething |= hasNext;
          return hasNext;
        }

        @Override
        public void close()
        {
          read.close();
          if (change != null && (!onceAReadGivesNothing || !gaveSomething))
          {
            final UpdateMsg msg = change;
            change = null;
            try
            {
              arrive(replicationServer, handler, msg);
            }
            catch (Exception e)
            {
              throw new RuntimeException(e);
            }
          }
        }
      };
    }
  }

  private static void deregister(ReplicationServerDomain domain)
  {
    if (domain != null)
    {
      DirectoryServer.deregisterMonitorProvider(domain);
    }
  }

  /** A message the handler came up with, and whether it read it from the changelog. */
  private static final class Sent
  {
    private final UpdateMsg msg;
    private final boolean fromTheChangelog;

    private Sent(UpdateMsg msg, boolean fromTheChangelog)
    {
      this.msg = msg;
      this.fromTheChangelog = fromTheChangelog;
    }
  }

  /**
   * The handler sends the given changes, in order, and serves them from its queue from the second
   * one on: it finds the first one in its queue and goes back to it.
   */
  private void assertServedFromTheQueueAfterTheFirst(MessageHandler handler, List<UpdateMsg> expected)
      throws Exception
  {
    final List<Sent> sent = nextMessages(handler, expected.size());
    assertThat(csnsOf(sent)).containsExactlyElementsOf(csnsOf(expected));
    assertThat(sent.get(1).fromTheChangelog)
        .as("the handler did not go back to its queue, which holds every change the consumer lacks")
        .isFalse();
    assertThat(handler.isFollowing()).isTrue();
  }

  /**
   * The handler sends the given changes, in order, and serves the last one from its queue: it is
   * handed to the queue once a read of the changelog gave nothing, see {@link ChangeAfterTheRead}.
   */
  private void assertServedFromTheQueueAfterAReadGaveNothing(MessageHandler handler, List<UpdateMsg> expected)
      throws Exception
  {
    final List<Sent> sent = nextMessages(handler, expected.size());
    assertThat(csnsOf(sent)).containsExactlyElementsOf(csnsOf(expected));
    assertThat(sent.get(sent.size() - 1).fromTheChangelog)
        .as("the handler did not go back to its queue once the changelog had nothing more for the consumer")
        .isFalse();
    assertThat(handler.isFollowing()).isTrue();
  }

  /**
   * The next messages the handler comes up with for its consumer, at most {@code count}: fewer
   * when it has none to come up with. It is asked from one thread, as the writer of a session asks.
   */
  private List<Sent> nextMessages(final MessageHandler handler, int count) throws Exception
  {
    final ExecutorService writer = Executors.newSingleThreadExecutor();
    try
    {
      final List<Sent> sent = new ArrayList<>();
      for (int i = 0; i < count; i++)
      {
        final Future<Sent> next = writer.submit(new Callable<Sent>()
        {
          @Override
          public Sent call() throws Exception
          {
            final UpdateMsg msg = handler.getNextMessage();
            return msg != null ? new Sent(msg, handler.isLastMessageFromLateQueue()) : null;
          }
        });
        final Sent nextSent;
        try
        {
          nextSent = next.get(NEXT_MESSAGE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
        }
        catch (TimeoutException e)
        {
          // it waits for a change to reach its queue: it has nothing more for the consumer
          handler.setConsumerActive(false);
          break;
        }
        if (nextSent == null)
        {
          break;
        }
        sent.add(nextSent);
      }
      return sent;
    }
    finally
    {
      writer.shutdownNow();
    }
  }

  private MessageHandler newHandler(ReplicationServer replicationServer, DN baseDN, int queueSize,
      ServerState consumerState) throws Exception
  {
    final MessageHandler handler = new MessageHandler(queueSize, replicationServer);
    handler.setBaseDNAndDomain(baseDN, false);
    handler.setInitialServerState(consumerState);
    return handler;
  }

  /** Hands a change over the way {@code ReplicationServerDomain.put()} does. */
  private void arrive(ReplicationServer replicationServer, MessageHandler handler, UpdateMsg msg)
      throws Exception
  {
    publishToChangelogOnly(replicationServer, handler.getBaseDN(), msg);
    handler.add(msg);
  }

  private void publishToChangelogOnly(ReplicationServer replicationServer, DN baseDN, List<UpdateMsg> msgs)
      throws Exception
  {
    for (UpdateMsg msg : msgs)
    {
      publishToChangelogOnly(replicationServer, baseDN, msg);
    }
  }

  private void publishToChangelogOnly(ReplicationServer replicationServer, DN baseDN, UpdateMsg msg)
      throws Exception
  {
    replicationServer.getChangelogDB().getReplicationDomainDB().publishUpdateMsg(baseDN, msg);
  }

  /** Changes {@code a<first>} .. {@code a<last>} of A, a second apart, the first 20 s after B's first one. */
  private List<UpdateMsg> changesOfA(DN baseDN, long t, int first, int last) throws Exception
  {
    final List<UpdateMsg> changes = new ArrayList<>();
    for (int i = first; i <= last; i++)
    {
      changes.add(newDeleteMsg(baseDN, "a" + i, new CSN(t + 20 + i, i, A)));
    }
    return changes;
  }

  private static ServerState stateOf(UpdateMsg... msgs)
  {
    final ServerState state = new ServerState();
    for (UpdateMsg msg : msgs)
    {
      state.update(msg.getCSN());
    }
    return state;
  }

  private static List<CSN> csnsOf(List<?> msgs)
  {
    final List<CSN> csns = new ArrayList<>();
    for (Object msg : msgs)
    {
      csns.add(msg instanceof Sent ? ((Sent) msg).msg.getCSN() : ((UpdateMsg) msg).getCSN());
    }
    return csns;
  }

  private DeleteMsg newDeleteMsg(DN baseDN, String rdnValue, CSN csn) throws Exception
  {
    return new DeleteMsg(DN.valueOf("cn=" + rdnValue + "," + baseDN), csn, "uuid");
  }

  private ReplicationServer newReplicationServer(String dbDirName) throws Exception
  {
    return new ReplicationServer(new ReplServerFakeConfiguration(
        TestCaseUtils.findFreePort(), dbDirName, 0, RS_ID, 0, WINDOW_SIZE, new TreeSet<String>()));
  }

  private void removeQuietly(ReplicationServer replicationServer)
  {
    try
    {
      if (replicationServer != null)
      {
        remove(replicationServer);
      }
    }
    catch (Exception ignored)
    {
      // the test has already reported what matters
    }
  }
}

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java

@@ -12,6 +12,7 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server.changelog.file;

@@ -82,6 +83,29 @@
    return Log.openLog(replicationEnv, LOG_DIRECTORY, parser, rotationParams);
  }

  /**
   * A record whose key is not after the newest one is not appended, whether the head log file is
   * to be rotated or not, and the log tells so.
   */
  @Test
  public void appendTellsWhetherTheRecordWasAppended() throws Exception
  {
    try (Log<String, String> log = openLog(RECORD_PARSER))
    {
      // the head log file holds 2 records: the next append rotates it
      assertThat(log.append(Record.from("key010", "value10 again"))).as("the newest key again").isFalse();
      assertThat(log.append(Record.from("key005", "value5 again"))).as("an older key").isFalse();
      assertThat(log.append(Record.from("key011", "value11"))).as("a newer key").isTrue();
      // the new head log file holds 1 record: no rotation
      assertThat(log.append(Record.from("key011", "value11 again"))).as("the newest key again").isFalse();

      try (DBCursor<Record<String, String>> cursor = log.getCursor())
      {
        assertThatCursorCanBeFullyReadFromStart(cursor, 1, 11);
      }
    }
  }

  @Test
  public void testCursor() throws Exception
  {