[#1198] Take the own state of a replication server again after each answer ...

Valery Kharseko

yesterday 49bde76364c753aa8e1ddbca0581216b1939add2

[#1198] Take the own state of a replication server again after each answer of its peers when it gathers the monitoring data (#1205)

Fixes #1198

## Summary

Under a steady write load a directory server left its replication server
for a peer which only had its changes from it (#1198) - the switch which
closed the session of #1195 with changes still on their way. The states
it compares were taken at different moments by the replication server
that sends the monitoring message: its own state first, before it asked
its peers, the state of each peer later, in the peer's answer. The
replication server now takes its own state again after each answer of a
peer is stored, so it is never shown behind a peer on the changes of its
directly connected directory servers.

## Root cause

- A directory server checks on every second monitoring message whether
its replication server still holds its newest change, and leaves it when
another replication server holds the change and it does not
(`ReplicationBroker.receive()`, `filterServersWithAllLocalDSChanges()`).
- The states of the replication servers in that message come from
`ReplicationDomainMonitor.recomputeMonitorData()`. It took the state of
the replication server itself - together with the states of its directly
connected directory servers - in `initializePendingMonitorData()`,
before it sent the monitor requests. The state of each peer is the one
the peer put in its answer, after it got the request.
- The changes of a directory server reach the peers only through the
replication server it is connected to, and that one stores a change
before it forwards it (`ReplicationServerDomain.put()`:
`publishUpdateMsg()`, then `addUpdate()` for each peer). So at any
moment a peer holds no change of that directory server which its
replication server does not hold. A peer can only look ahead when its
state was taken later - which under load it always was.
- In the run of #1195: RS(3637) on `cv-c0` at or past seqnum 84376 of
DS(20750), next to RS(13750) at 84374, while RS(13750) is the only way
the changes of DS(20750) reach RS(3637).

## Change

`ReplicationDomainMonitor`:

- `initializePendingMonitorData()` runs once when the computing starts,
as before, and again each time the answer of a peer is stored
(`receiveMonitorDataResponse()`, under the lock the answers are stored
under). The state of the replication server itself is then never older
than an answer, and never later than the last one.
- It is not taken once the wait for the answers is over. A peer which
does not answer would put the whole wait - up to 5 seconds - between the
own state and the answers of the others. Those peers would then be shown
missing the changes which came in meanwhile, and the local directory
servers missing the changes of each other.
- The block which publishes the computed data also takes it out of the
reach of the answers (`pendingMonitorData = null`). Before, an answer
coming late, between that block and the `finally` which cleared the
pending data, went into the published data with a state taken after the
own one.

A peer which really holds changes this replication server lacks -
published to it by the directory server on an earlier session - is still
reported ahead.

## Why the directory server side is left as it is

The directions in #1198 were on the directory server side: judging the
replication server by the changes published when the states were taken,
or only after several checks. Once the states are taken consistently,
comparing them with the live state of the directory server no longer
makes it leave its replication server. When the changes still on their
way are missing on its replication server, they are missing on every
peer too. The criterion then finds no replication server with the newest
change and keeps them all (`RSEvaluations.keepBest()` keeps the list
when nothing was accepted). The states the directory server keeps per
replication server are merged with a max and survive topology messages
(`ReplicationServerInfo.update()`, `Topology.computeRSInfos()`), so that
holds across messages as well.

## Not covered

- When a directory server connects, the phase-one handshake asks the
replication servers for their states one after the other
(`collectReplicationServersInfo()`). On the tail of a closed session, a
later answer could in theory be ahead in the same way. This was not seen
in the runs and this change does not touch it.
- An older replication server in a mixed topology still sends the states
as before.

## Tests

`ReplicationDomainMonitorTest` (new) gathers the monitor data of a
mocked domain whose peers answer from reader threads of their own:

| Case | What it pins |
|---|---|
| `aPeerDoesNotLookAheadOnTheChangesItGetsFromThisReplicationServer` | A
change comes in once the requests are sent and the requester waits, and
the peer answers with it: the own state is not older than the peer's.
Red on master, green with the change. |
| `aPeerWhichHoldsChangesThisReplicationServerLacksIsReportedAhead` | A
peer really ahead is still reported ahead. |
| `aPeerWhichAnswersAfterTheWaitLeavesTheOthersUpToDate` | One peer
answers at once and another only after the wait, with changes coming in
meanwhile: the first one shows no missing changes, and the late answer
stays out of the published data. Takes the 5 seconds of the wait. |

Mutants of `ReplicationDomainMonitor`, each run against the class:

| Mutant | Caught by |
|---|---|
| master | `aPeerDoesNotLookAhead…` |
| no new snapshot after an answer | `aPeerDoesNotLookAhead…` |
| one new snapshot right after the requests are sent |
`aPeerDoesNotLookAhead…` |
| one new snapshot once the wait is over |
`aPeerWhichAnswersAfterTheWait…` |
| own state reported as the max of its and of the peers' |
`aPeerWhichHoldsChanges…` |
| published data not taken out of the reach of the answers | not caught:
the late answer cannot be put between the two blocks without a hook, so
the change closes the gap by construction |

## Verification

- `org/opends/server/replication/**` in the reactor (`-pl
opendj-server-legacy -am`): 3835 tests, 0 failures, 0 errors, 0 skipped.

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

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

@@ -201,6 +202,8 @@
            pendingMonitorData.completeComputing();
            monitorData = pendingMonitorData;
            monitorDataLastBuildDate = TimeThread.getTime();
            // A late answer must not go into the published data: it would come after our own state.
            pendingMonitorData = null;
          }
        }
        finally
@@ -219,7 +222,17 @@
    return monitorData;
  }

  /** Start collecting global monitoring information for the replication domain. */
  /**
   * Takes the states of this RS and of its directly connected DSs into the pending monitor data:
   * once when the computing starts, and again after each answer of another RS is stored.
   * <p>
   * The changes of our directly connected DSs reach the other RSs through us only, so a state of
   * ours taken before theirs would show us behind them on those changes, and a DS reading that in
   * the monitoring message would leave us for an RS which only has its changes from us (issue
   * #1198). Taken again at each answer rather than once the wait for the answers is over, our
   * state is no later than the last answer either: an RS which does not answer would otherwise put
   * the whole wait between our state and the answers of the others, and they would look behind.
   */
  private void initializePendingMonitorData()
  {
    // Let's process our directly connected DS
@@ -318,6 +331,9 @@
            }
          }
        }

        // Our own state must not be older than the one this RS just answered with (issue #1198)
        initializePendingMonitorData();
      }
      catch (RuntimeException e)
      {

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

New file
@@ -0,0 +1,246 @@
/*
 * 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 java.util.concurrent.TimeUnit.*;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.*;
import static org.testng.Assert.*;

import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.MonitorMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

/**
 * Tests the states {@link ReplicationDomainMonitor} gathers for the replication servers of a
 * domain, which the monitoring messages hand to the directory servers.
 * <p>
 * A directory server connected to a replication server checks on those messages whether its
 * replication server still holds its newest change, and leaves it when another one holds it and
 * it does not (issue #1198). The changes of a directory server reach the other replication
 * servers through the one it is connected to only, so none of them can hold one of its changes
 * which that one does not - unless the states compared were not taken at the same moment.
 */
@SuppressWarnings("javadoc")
public class ReplicationDomainMonitorTest extends DirectoryServerTestCase
{
  private static final DN BASE_DN = DN.valueOf("dc=example,dc=com");
  private static final int LOCAL_RS_ID = 1;
  private static final int PEER_RS_ID = 2;
  private static final int SILENT_RS_ID = 3;
  private static final int DS_ID = 10;
  private static final CSN FIRST = new CSN(1000, 1, DS_ID);
  private static final CSN FORWARDED = new CSN(1000, 2, DS_ID);
  private static final CSN AHEAD = new CSN(1000, 5, DS_ID);
  private static final long STEP_TIMEOUT_MS = 10000;

  /** What the reader of the session to a peer does once the monitor request has been sent to it. */
  private interface PeerReader
  {
    void run(ReplicationDomainMonitor monitor, Thread requester) throws Exception;
  }

  /** The state of the changelog of this replication server, which the mocked domain reports. */
  private AtomicReference<ServerState> changelog;
  /** Counted down once the monitor data is published. */
  private CountDownLatch published;
  private Map<Integer, Thread> readers;
  private AtomicReference<Throwable> readerFailure;

  @BeforeMethod
  public void setUp()
  {
    changelog = new AtomicReference<>(stateWith(FIRST));
    published = new CountDownLatch(1);
    readers = new ConcurrentHashMap<>();
    readerFailure = new AtomicReference<>();
  }

  /**
   * A change of a directory server connected to this replication server comes in while the
   * answers are waited for, and this replication server forwards it to its peer before the peer
   * answers: the state reported for this replication server must not be older than the one of its
   * peer on the changes of that directory server.
   */
  @Test
  public void aPeerDoesNotLookAheadOnTheChangesItGetsFromThisReplicationServer() throws Exception
  {
    final Map<Integer, PeerReader> peers = new LinkedHashMap<>();
    peers.put(PEER_RS_ID, (monitor, requester) ->
    {
      awaitTheWaitForTheAnswers(requester);
      changelog.set(stateWith(FORWARDED));
      monitor.receiveMonitorDataResponse(answer(PEER_RS_ID, FORWARDED), PEER_RS_ID);
    });

    final ReplicationDomainMonitorData data = recompute(peers);

    final CSN local = data.getRSStates(LOCAL_RS_ID).getCSN(DS_ID);
    final CSN peer = stateOf(data, PEER_RS_ID).getCSN(DS_ID);
    assertFalse(local.isOlderThan(peer), "RS(" + LOCAL_RS_ID + ") is reported at " + local.toStringUI()
        + ", behind RS(" + PEER_RS_ID + ") at " + peer.toStringUI()
        + ", which got the changes of DS(" + DS_ID + ") from it");
  }

  /**
   * A peer can hold changes of a directory server which this replication server does not, when
   * the directory server published them to that peer on an earlier session: the peer must still
   * be reported ahead then.
   */
  @Test
  public void aPeerWhichHoldsChangesThisReplicationServerLacksIsReportedAhead() throws Exception
  {
    final Map<Integer, PeerReader> peers = new LinkedHashMap<>();
    peers.put(PEER_RS_ID,
        (monitor, requester) -> monitor.receiveMonitorDataResponse(answer(PEER_RS_ID, AHEAD), PEER_RS_ID));

    final ReplicationDomainMonitorData data = recompute(peers);

    assertEquals(data.getRSStates(LOCAL_RS_ID).getCSN(DS_ID), FIRST);
    assertEquals(stateOf(data, PEER_RS_ID).getCSN(DS_ID), AHEAD);
  }

  /**
   * One peer answers at once, another one only once the wait for the answers is over, and changes
   * come in meanwhile: the peer which answered is not shown missing the changes which came in
   * while the other one was waited for, and the late answer does not go into the published data.
   */
  @Test
  public void aPeerWhichAnswersAfterTheWaitLeavesTheOthersUpToDate() throws Exception
  {
    final Map<Integer, PeerReader> peers = new LinkedHashMap<>();
    peers.put(PEER_RS_ID,
        (monitor, requester) -> monitor.receiveMonitorDataResponse(answer(PEER_RS_ID, FIRST), PEER_RS_ID));
    peers.put(SILENT_RS_ID, (monitor, requester) ->
    {
      awaitTheWaitForTheAnswers(requester);
      readers.get(PEER_RS_ID).join(STEP_TIMEOUT_MS);
      changelog.set(stateWith(FORWARDED));
      assertTrue(published.await(STEP_TIMEOUT_MS, MILLISECONDS), "the monitor data was not published");
      monitor.receiveMonitorDataResponse(answer(SILENT_RS_ID, FORWARDED), SILENT_RS_ID);
    });

    final ReplicationDomainMonitorData data = recompute(peers);

    assertEquals(data.getMissingChangesRS(PEER_RS_ID), 0, "RS(" + PEER_RS_ID + ") answered with every "
        + "change there was then, and is shown missing the ones which came in while RS(" + SILENT_RS_ID
        + ") was waited for");
    assertNull(data.getRSStates(SILENT_RS_ID),
        "the answer of RS(" + SILENT_RS_ID + "), which came after the wait, went into the published data");
  }

  /**
   * Gathers the monitor data of a domain with the provided peers, in this order: the reader of
   * the session to each of them runs once the monitor request is sent to it. The readers are done
   * when this returns.
   */
  private ReplicationDomainMonitorData recompute(Map<Integer, PeerReader> peerReaders) throws Exception
  {
    final ReplicationServerDomain domain = mock(ReplicationServerDomain.class);
    when(domain.getBaseDN()).thenReturn(BASE_DN);
    when(domain.getLocalRSServerId()).thenReturn(LOCAL_RS_ID);
    when(domain.getConnectedDSs()).thenReturn(Collections.<Integer, DataServerHandler> emptyMap());
    when(domain.getLatestServerState()).thenAnswer(invocation -> changelog.get().duplicate());
    final ReplicationDomainMonitor monitor = new ReplicationDomainMonitor(domain);

    final Map<Integer, ReplicationServerHandler> peers = new LinkedHashMap<>();
    for (Map.Entry<Integer, PeerReader> entry : peerReaders.entrySet())
    {
      final int peerId = entry.getKey();
      final PeerReader peerReader = entry.getValue();
      final ReplicationServerHandler peer = mock(ReplicationServerHandler.class);
      when(peer.getServerId()).thenReturn(peerId);
      doAnswer(invocation ->
      {
        final Thread requester = Thread.currentThread();
        final Thread reader = new Thread(() ->
        {
          try
          {
            peerReader.run(monitor, requester);
          }
          catch (Throwable e)
          {
            readerFailure.compareAndSet(null, e);
          }
        }, "reader of the session to RS(" + peerId + ")");
        reader.setDaemon(true);
        readers.put(peerId, reader);
        reader.start();
        return null;
      }).when(peer).send(any(ReplicationMsg.class));
      peers.put(peerId, peer);
    }
    when(domain.getConnectedRSs()).thenReturn(peers);

    final ReplicationDomainMonitorData data = monitor.recomputeMonitorData();
    published.countDown();
    for (Map.Entry<Integer, Thread> reader : readers.entrySet())
    {
      reader.getValue().join(STEP_TIMEOUT_MS);
      assertFalse(reader.getValue().isAlive(), "the reader of the session to RS(" + reader.getKey() + ") is stuck");
    }
    if (readerFailure.get() != null)
    {
      throw new AssertionError("a reader failed", readerFailure.get());
    }
    return data;
  }

  /** Returns once the requests are all sent, and the requester waits for the answers. */
  private static void awaitTheWaitForTheAnswers(Thread requester) throws InterruptedException
  {
    final long deadline = System.currentTimeMillis() + STEP_TIMEOUT_MS;
    while (requester.getState() != Thread.State.TIMED_WAITING)
    {
      assertTrue(System.currentTimeMillis() < deadline, "the requester did not wait for the answers");
      Thread.sleep(1);
    }
  }

  private static ServerState stateOf(ReplicationDomainMonitorData data, int rsId)
  {
    final ServerState state = data.getRSStates(rsId);
    assertNotNull(state, "the answer of RS(" + rsId + ") did not come within the wait");
    return state;
  }

  private static MonitorMsg answer(int rsId, CSN newest)
  {
    final MonitorMsg answer = new MonitorMsg(rsId, LOCAL_RS_ID);
    answer.setReplServerDbState(stateWith(newest));
    return answer;
  }

  private static ServerState stateWith(CSN csn)
  {
    final ServerState state = new ServerState();
    state.update(csn);
    return state;
  }
}