[#1029] Send a directory server only the updates it gives send-window credi...

Valery Kharseko

22 hours ago 7606bd26f14a4b9755577bc7dad8ea57c36e7ba1

[#1029] Send a directory server only the updates it gives send-window credit for (#1034)

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

@@ -49,6 +49,7 @@
import org.opends.server.replication.protocol.StartSessionMsg;
import org.opends.server.replication.protocol.StopMsg;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.types.DirectoryException;

/**
@@ -303,6 +304,25 @@
  }

  /**
   * A directory server is sent only the updates which contribute to the domain state: those are
   * the ones it replays, and the only ones it gives credit for on the send window of the session
   * - see the listener of ReplicationDomain, which calls processUpdateDone() for nothing else.
   * ServerHandler.take() takes a permit of that window for every message it hands to the writer,
   * so any other message sent to a directory server would cost the session a permit for good.
   * <p>
   * Today that is the ReplicaOfflineMsg, which is exchanged between replication servers only.
   * ReplicationServerDomain.put() never queues one for a directory server, but the catch-up path
   * reads the changelog, where the cursor of a replica which went offline synthesizes one from
   * its offline CSN, and since the state of this handler never moves past that CSN every
   * catch-up round would read the same message again, one permit each (issue #1029).
   */
  @Override
  boolean updateServerState(UpdateMsg msg)
  {
    return msg.contributesToDomainState() && super.updateServerState(msg);
  }

  /**
   * Process message of a remote server changing his status.
   * @param csMsg The message containing the new status
   * @return The new server status of the DS

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

@@ -759,6 +759,10 @@

  /**
   * Update the serverState with the last message sent.
   * <p>
   * What this returns decides whether {@code getNextMessage()} hands the message to the writer
   * at all: a message the state of the consumer already covers is not sent to it. A handler can
   * narrow that further for its kind of consumer - {@link DataServerHandler} does.
   *
   * @param msg the last update sent.
   * @return boolean indicating if the update was meaningful.

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

@@ -126,12 +126,12 @@
          session.publish(updateMsg);
          /*
           * Only the forward to a peer RS ends the wait of the shutdown: what the grace period
           * buys is the rest of the topology learning that the replica went offline.
           * ReplicationServerDomain.put() never queues this message for a directory server - its
           * isUpdateMsgFiltered() drops it there - but a directory server which is catching up
           * reads its updates from the changelog, where ReplicaCursor synthesizes a
           * ReplicaOfflineMsg from the offline CSN of the replica. Publishing that one says
           * nothing about the peer RSs the shutdown is waiting for.
           * buys is the rest of the topology learning that the replica went offline. A directory
           * server is never handed this message - ReplicationServerDomain.put() does not queue
           * it for one, and DataServerHandler.updateServerState() drops the one the changelog
           * cursor of a directory server which is catching up synthesizes from the offline CSN
           * of the replica (issue #1029) - so the guard says whose forward counts rather than
           * telling two deliveries apart.
           */
          if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
          {

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

@@ -29,6 +29,7 @@
import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;

import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -65,6 +66,7 @@
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.internal.SearchRequest;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.DomainFakeCfg;
import org.opends.server.replication.plugin.DummyReplicationDomain;
@@ -74,6 +76,7 @@
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.changelog.file.FileChangelogDB;
import org.opends.server.replication.service.ReplicationBroker;
@@ -1407,6 +1410,40 @@
  }

  /**
   * Receives from the broker until the update with the given CSN arrives, returning everything
   * received before it - so that what the broker was not sent can be asserted on without
   * waiting out a timeout.
   *
   * @param broker Broker from which the update is expected.
   * @param csn CSN of the update to receive up to.
   * @return the messages received before that update, in order
   * @throws AssertionError if the broker is stopped or times out before the update arrives
   */
  protected static List<ReplicationMsg> receiveUntil(ReplicationBroker broker, CSN csn) throws Exception
  {
    final List<ReplicationMsg> received = new ArrayList<>();
    try
    {
      while (true)
      {
        final ReplicationMsg msg = broker.receive();
        assertNotNull(msg, "The broker was stopped before the update " + csn + " reached it."
            + " Received the following messages before that: " + received);
        if (msg instanceof UpdateMsg && csn.equals(((UpdateMsg) msg).getCSN()))
        {
          return received;
        }
        received.add(msg);
      }
    }
    catch (SocketTimeoutException e)
    {
      throw new AssertionError("Failed to receive the update " + csn + " before the socket timeout."
          + " Received the following messages during wait time: " + received, e);
    }
  }

  /**
   * Performs an internal search, waiting for at most 3 seconds for expected result code and expected
   * number of entries.
   */

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

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

import java.util.ArrayList;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.DomainFakeCfg;
import org.opends.server.replication.plugin.DummyReplicationDomain;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.Attribute;
import org.opends.server.util.TestTimer;
import org.testng.annotations.Test;

/**
 * A directory server is never sent a ReplicaOfflineMsg. ReplicationServerDomain.put() does not
 * queue one for a directory server, but a directory server which is catching up reads its
 * updates from the changelog, where the cursor of a replica which went offline synthesizes one
 * from the offline CSN of that replica, and the writer used to publish it (issue #1029).
 * <p>
 * The message costs the session a permit of its send window for good: the replication server
 * takes one for every message it hands to the writer, and a directory server gives credit only
 * for the updates it replays - a ReplicaOfflineMsg is not one of them. The state of the handler
 * does not move past an offline CSN either, so every catch-up round read the same message
 * again, one permit each. A session which lost more than half its window that way was never
 * sent anything again.
 * <p>
 * The tests read the send window from the monitor entry of the handler once the directory
 * server holds a change published after the offline message: nothing the replication server
 * sends after that is left to account for, and the broker of the tests never gives credit, so
 * the window is the size the directory server announced less the messages it was sent.
 */
@SuppressWarnings("javadoc")
public class ReplicaOfflineMsgCatchUpTest extends ReplicationTestCase
{
  private static final int SOCKET_TIMEOUT_MS = 30000;
  private static final int WINDOW_SIZE = 100;
  /** The replica which goes offline. */
  private static final int OFFLINE_DS_ID = 81;
  /** The directory server whose catch-up meets the offline CSN of {@link #OFFLINE_DS_ID}. */
  private static final int CATCHING_UP_DS_ID = 82;
  /** The replica whose change tells the tests the catch-up is over. */
  private static final int LATER_DS_ID = 83;

  /**
   * The catch-up round of a directory server which is behind the last change of the offline
   * replica holds that change and the offline message which follows it. The change is sent, the
   * message is not.
   */
  @Test
  public void aDirectoryServerBehindTheOfflineReplicaIsSentItsChangesButNotItsOfflineMessage()
      throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ReplicationBroker offlineBroker = null;
    ReplicationBroker broker = null;
    ReplicationBroker laterBroker = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer = newReplicationServer("replicaOfflineCatchUpBehindDb", 8301, replicationPort);
      final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true);

      offlineBroker = openReplicationSession(baseDN, OFFLINE_DS_ID, WINDOW_SIZE, replicationPort,
          5000, EMPTY_DN_GENID);
      final CSNGenerator csns = new CSNGenerator(OFFLINE_DS_ID, 0);
      final DeleteMsg lastChange = newDeleteMsg(csns.newCSN());
      offlineBroker.publish(lastChange);
      offlineBroker.publish(new ReplicaOfflineMsg(csns.newCSN()));
      offlineBroker.stop();
      waitForDisconnectedDirectoryServer(domain, OFFLINE_DS_ID);

      // an empty state: the catch-up starts before the change of the offline replica
      broker = openReplicationSession(baseDN, CATCHING_UP_DS_ID, WINDOW_SIZE, replicationPort,
          5000, EMPTY_DN_GENID);
      laterBroker = openReplicationSession(baseDN, LATER_DS_ID, WINDOW_SIZE, replicationPort,
          5000, EMPTY_DN_GENID);
      final DeleteMsg laterChange = newDeleteMsg(new CSNGenerator(LATER_DS_ID, 0).newCSN());
      laterBroker.publish(laterChange);

      final List<ReplicationMsg> received = receiveUntil(broker, laterChange.getCSN());
      assertThat(currentSendWindow(domain.getConnectedDSs().get(CATCHING_UP_DS_ID)))
          .as("the send window of the session is short of the two changes the directory "
              + "server was sent, and of nothing else - it received: %s", received)
          .isEqualTo(WINDOW_SIZE - 2);
      assertThat(received)
          .as("the directory server was sent the offline message of the replica it caught up past")
          .noneMatch(ReplicaOfflineMsg.class::isInstance);
      assertThat(csnsOf(received)).contains(lastChange.getCSN());
    }
    finally
    {
      stop(laterBroker, broker, offlineBroker);
      removeQuietly(replicationServer);
    }
  }

  /**
   * The catch-up round of a directory server which already holds the last change of the
   * offline replica holds nothing but the offline message. It is not sent, and the directory
   * server is left following the queue of the domain: the change published next comes from
   * there.
   */
  @Test
  public void aDirectoryServerUpToDateWithTheOfflineReplicaIsNotSentItsOfflineMessage()
      throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    ReplicationBroker offlineBroker = null;
    ReplicationBroker broker = null;
    ReplicationBroker laterBroker = null;
    try
    {
      final int replicationPort = TestCaseUtils.findFreePort();
      replicationServer = newReplicationServer("replicaOfflineCatchUpUpToDateDb", 8302, replicationPort);
      final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true);

      offlineBroker = openReplicationSession(baseDN, OFFLINE_DS_ID, WINDOW_SIZE, replicationPort,
          5000, EMPTY_DN_GENID);
      final CSNGenerator csns = new CSNGenerator(OFFLINE_DS_ID, 0);
      final DeleteMsg lastChange = newDeleteMsg(csns.newCSN());
      offlineBroker.publish(lastChange);
      offlineBroker.publish(new ReplicaOfflineMsg(csns.newCSN()));
      offlineBroker.stop();
      waitForDisconnectedDirectoryServer(domain, OFFLINE_DS_ID);

      // a state which holds the change: the catch-up starts between it and the offline CSN
      final ServerState state = new ServerState();
      state.update(lastChange.getCSN());
      broker = openReplicationSession(baseDN, CATCHING_UP_DS_ID, replicationPort, state);
      laterBroker = openReplicationSession(baseDN, LATER_DS_ID, WINDOW_SIZE, replicationPort,
          5000, EMPTY_DN_GENID);
      final DeleteMsg laterChange = newDeleteMsg(new CSNGenerator(LATER_DS_ID, 0).newCSN());
      laterBroker.publish(laterChange);

      final List<ReplicationMsg> received = receiveUntil(broker, laterChange.getCSN());
      assertThat(currentSendWindow(domain.getConnectedDSs().get(CATCHING_UP_DS_ID)))
          .as("the send window of the session is short of the one change the directory "
              + "server was sent, and of nothing else - it received: %s", received)
          .isEqualTo(WINDOW_SIZE - 1);
      assertThat(received)
          .as("the directory server was sent the offline message of a replica it was up to date with")
          .noneMatch(ReplicaOfflineMsg.class::isInstance);
      assertThat(csnsOf(received)).doesNotContain(lastChange.getCSN());
    }
    finally
    {
      stop(laterBroker, broker, offlineBroker);
      removeQuietly(replicationServer);
    }
  }

  private ReplicationServer newReplicationServer(String dbDirName, int serverId, int replicationPort)
      throws Exception
  {
    return new ReplicationServer(new ReplServerFakeConfiguration(
        replicationPort, dbDirName, 0, serverId, 0, WINDOW_SIZE, new TreeSet<String>()));
  }

  /** Opens a session announcing the given state rather than an empty one. */
  private ReplicationBroker openReplicationSession(DN baseDN, int serverId, int replicationPort,
      ServerState state) throws Exception
  {
    final DomainFakeCfg config = newFakeCfg(baseDN, serverId, replicationPort);
    config.setWindowSize(WINDOW_SIZE);
    final ReplicationBroker broker = new ReplicationBroker(
        new DummyReplicationDomain(EMPTY_DN_GENID), state, config, getReplSessionSecurity());
    connect(broker, 5000);
    return broker;
  }

  private static DeleteMsg newDeleteMsg(CSN csn)
  {
    return new DeleteMsg(DN.valueOf("uid=" + csn.getServerId() + "," + TEST_ROOT_DN_STRING), csn,
        "entry-uuid-" + csn.getServerId());
  }

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

  /** The {@code current-send-window} attribute of the monitor entry of the handler. */
  private static int currentSendWindow(DataServerHandler dsHandler)
  {
    assertThat(dsHandler).as("the directory server is not connected anymore").isNotNull();
    for (Attribute attribute : dsHandler.getMonitorData())
    {
      if ("current-send-window".equals(attribute.getAttributeDescription().getNameOrOID()))
      {
        return Integer.parseInt(attribute.iterator().next().toString());
      }
    }
    throw new AssertionError("no current-send-window on the monitor entry of " + dsHandler);
  }

  /**
   * Waits for the reader of the directory server to be done: it processes what the session
   * received in order and stops the handler last, so a replica which is gone from the domain
   * has had its ReplicaOfflineMsg recorded in the changelog.
   */
  private static void waitForDisconnectedDirectoryServer(final ReplicationServerDomain domain,
      final int serverId) throws Exception
  {
    new TestTimer.Builder()
        .maxSleep(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS)
        .sleepTimes(10, TimeUnit.MILLISECONDS)
        .toTimer()
        .repeatUntilSuccess(new Callable<Void>()
        {
          @Override
          public Void call()
          {
            assertThat(domain.getConnectedDSs())
                .as("the replica which went offline never left the domain")
                .doesNotContainKey(serverId);
            return null;
          }
        });
  }

  /** Teardown must never mask the primary assertion failure. */
  private void removeQuietly(ReplicationServer replicationServer)
  {
    try
    {
      remove(replicationServer);
    }
    catch (Exception ignored)
    {
    }
  }
}

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

@@ -417,14 +417,16 @@
  }

  /**
   * Only a peer replication server learning about the offline replica ends the wait.
   * ReplicationServerDomain.put() never queues a ReplicaOfflineMsg for a directory server, but
   * the changelog cursor of a directory server which is catching up synthesizes one from the
   * offline CSN of the replica, so the writer serving a directory server can publish it - and
   * the peer replication servers would still know nothing.
   * Only a peer replication server learning about the offline replica ends the wait, and a
   * directory server is never told. ReplicationServerDomain.put() never queues a
   * ReplicaOfflineMsg for a directory server, and one which reaches the queue of its handler
   * all the same - the changelog cursor of a directory server which is catching up synthesizes
   * one from the offline CSN of the replica - is dropped by the handler before its writer is
   * given it (issue #1029): the directory server is not sent it, nothing reports a forward for
   * it, and the peer replication servers still know nothing.
   */
  @Test
  public void theForwardToADirectoryServerDoesNotEndTheWait() throws Exception
  public void theDirectoryServerIsNeitherSentTheMessageNorEndsTheWait() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
@@ -448,17 +450,24 @@
            replicationServer.getReplicationServerDomain(baseDN, true);
        final DataServerHandler dsHandler = waitForConnectedDirectoryServer(domain);

        final CSN offlineCSN = newOfflineCSN();
        final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0);
        final CSN offlineCSN = csns.newCSN();
        final long startTime = System.nanoTime();
        shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
        // the very message the shutdown waits for, so only the guard of the writer can save it
        // the very message the shutdown waits for, queued the only way it can reach a directory
        // server: put() never does this
        dsHandler.add(new ReplicaOfflineMsg(offlineCSN));
        // a change queued behind it: once the directory server holds this one, its handler is
        // past the message
        final DeleteMsg change = new DeleteMsg(DN.valueOf("uid=offline," + TEST_ROOT_DN_STRING),
            csns.newCSN(), "offline-entry-uuid");
        dsHandler.add(change);

        // the directory server did receive it, so its writer went through the forwarding code
        assertThat(waitForSpecificMsg(broker, ReplicaOfflineMsg.class).getCSN().getServerId())
            .isEqualTo(LOCAL_DS_ID);
        assertThat(receiveUntil(broker, change.getCSN()))
            .as("the directory server was sent the ReplicaOfflineMsg queued for it")
            .noneMatch(ReplicaOfflineMsg.class::isInstance);
        assertThat(elapsedMillis(startTime))
            .as("the fixture must deliver the message well inside the grace period, otherwise "
            .as("the fixture must deliver the change well inside the grace period, otherwise "
                + "the wait asserted below cannot be told apart from a slow delivery")
            .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD / 2);

@@ -466,7 +475,7 @@
        final long elapsed = elapsedMillis(startTime);

        assertThat(elapsed)
            .as("the message published to a directory server ended the wait of the shutdown")
            .as("the handler of a directory server ended the wait of the shutdown")
            .isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
      }
    }