[#710] Fix replication catch-up re-sending updates with the original assure...

Valery Kharseko

16 hours ago e8c0d1b865426420b4bd9c2e482b27be0e4df57f

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

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

@@ -254,5 +255,11 @@
      "concerned server ids: " + idList;
  }

  /** {@inheritDoc} */
  @Override
  public String toString()
  {
    return getClass().getSimpleName() + " csn: " + csn + ", " + errorsToString();
  }
}


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

@@ -13,13 +13,17 @@
 *
 * Copyright 2008-2009 Sun Microsystems, Inc.
 * Portions Copyright 2013-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC
 */

package org.opends.server.replication.server;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
@@ -69,6 +73,15 @@
  protected Map<Integer,Boolean> expectedServersAckStatus = new HashMap<>();

  /**
   * Immutable snapshot of the ids of the servers we expect an ack from, taken
   * at construction time. Used for lock-free membership checks: the key set of
   * {@link #expectedServersAckStatus} never changes after construction, but
   * that map has its values mutated under lock by {@code processReceivedAck()},
   * so it must not be read concurrently without synchronization.
   */
  private final Set<Integer> expectedServerIds;

  /**
   * Facility for monitoring:
   * If the timeout occurs for the original update, we call createAck(true)
   * in the timeout code for sending back an error ack to the original server.
@@ -99,6 +112,22 @@
    {
      expectedServersAckStatus.put(serverId, false);
    }
    this.expectedServerIds =
        Collections.unmodifiableSet(new HashSet<>(expectedServers));
  }

  /**
   * Indicates whether the provided server is one of the servers an ack is
   * expected from for the matching update message. Reads an immutable snapshot
   * taken at construction time, so it is safe to call without holding the lock
   * on this object.
   *
   * @param serverId The serverId of the server.
   * @return true if an ack is expected from the provided server.
   */
  public boolean isExpectedServer(int serverId)
  {
    return expectedServerIds.contains(serverId);
  }

  /**

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

@@ -80,6 +80,13 @@
  private final int maxQueueBytesSize;
  /** Specifies whether the consumer is following the producer (is not late). */
  private boolean following;
  /**
   * Specifies whether the last update message returned by
   * {@link #getNextMessage()} was re-read from the changelog DB (catch-up
   * path) rather than taken from the in-memory {@link #msgQueue}. Only ever
   * accessed by the single consumer thread calling {@link #getNextMessage()}.
   */
  private boolean lastMessageFromLateQueue;
  /** Specifies the current serverState of this handler. */
  private ServerState serverState;
  /** Specifies the baseDN of the domain. */
@@ -226,6 +233,21 @@
  }

  /**
   * Indicates whether the last update message returned by
   * {@link #getNextMessage()} was re-read from the changelog DB (catch-up
   * path) rather than taken from the in-memory queue.
   * <p>
   * Must only be called from the consumer thread calling
   * {@link #getNextMessage()}.
   *
   * @return true if the last returned update message came from the late queue
   */
  protected boolean isLastMessageFromLateQueue()
  {
    return lastMessageFromLateQueue;
  }

  /**
   * Retrieves the name of this monitor provider.  It should be unique among all
   * monitor providers, including all instances of the same monitor provider.
   *
@@ -319,6 +341,9 @@
                msgQueue.consumeUpTo(msg);
                if (updateServerState(msg))
                {
                  // the returned instance is the one re-read from the
                  // changelog DB, not its msgQueue equivalent
                  lastMessageFromLateQueue = true;
                  return msg;
                }
              }
@@ -347,6 +372,7 @@
          }
          if (updateServerState(msg))
          {
            lastMessageFromLateQueue = true;
            return msg;
          }
          continue;
@@ -379,6 +405,7 @@
             * by the other server.
             * Otherwise just loop to select the next message.
             */
            lastMessageFromLateQueue = false;
            return msg;
          }
        }

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

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

@@ -790,6 +790,20 @@
  }

  /**
   * Indicates whether the ack window for the provided CSN is still open and
   * the provided server is one of the servers an ack is expected from.
   *
   * @param csn The CSN of the update message.
   * @param serverId The serverId of the candidate acknowledging server.
   * @return true if an ack from the provided server would be accounted for.
   */
  boolean isExpectedAck(CSN csn, int serverId)
  {
    final ExpectedAcksInfo expectedAcksInfo = waitingAcks.get(csn);
    return expectedAcksInfo != null && expectedAcksInfo.isExpectedServer(serverId);
  }

  /**
   * Process an ack received from a given server.
   *
   * @param ack The ack message received.

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

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

@@ -84,7 +85,19 @@

    // Get the ack status for the matching server
    int ackingServerId = ackingServer.getServerId();
    boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
    Boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
    if (ackReceived == null)
    {
      // Ack from a server we were not expecting an ack from, for instance
      // because the update was delivered to it with the assured flag of the
      // original sender through the changelog catch-up path: ignore it.
      if (logger.isTraceEnabled())
      {
        logger.trace("Received ack from not expected server id: " +
          ackingServerId + " ack message: " + ackMsg);
      }
      return false;
    }
    if (ackReceived)
    {
      // Sanity check: this should never happen

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

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

@@ -146,7 +147,19 @@
  {
    // Get the ack status for the matching server
    int ackingServerId = ackingServer.getServerId();
    boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
    Boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
    if (ackReceived == null)
    {
      // Ack from a server we were not expecting an ack from, for instance
      // because the update was delivered to it with the assured flag of the
      // original sender through the changelog catch-up path: ignore it.
      if (logger.isTraceEnabled())
      {
        logger.trace("Received ack from not expected server id: "
          + ackingServerId + " ack message: " + ackMsg);
      }
      return false;
    }
    if (ackReceived)
    {
      // Sanity check: this should never happen

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

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

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.util.StaticUtils.*;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -924,12 +927,32 @@
   */
  public UpdateMsg take() throws ChangelogException
  {
    final UpdateMsg msg = getNextMessage();
    UpdateMsg msg = getNextMessage();
    final boolean fromLateQueue = isLastMessageFromLateQueue();

    acquirePermitInSendWindow();

    if (msg != null)
    {
      // Updates re-read from the changelog DB (catch-up path) carry the
      // assured flag, mode and safe data level of their original sender:
      // the NotAssuredUpdateMsg substitution performed at publish time by
      // ReplicationServerDomain.addUpdate() only exists on the in-memory
      // queue path. Normalize them here: keep the assured flag only while
      // the ack window is still open and this server is expected to ack.
      // Messages taken from the in-memory queue already carry the
      // publish-time decision and must NOT be revisited: the ack window may
      // legitimately close (timeout, or enough acks already received)
      // before a slow peer gets here - e.g. when acquirePermitInSendWindow()
      // above blocks on a closed send window - and such a peer must still
      // receive the assured flag it was deemed eligible for. Its late ack is
      // then safely ignored by ReplicationServerDomain.processAck() and
      // ExpectedAcksInfo.processReceivedAck().
      if (fromLateQueue && msg.isAssured()
          && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId))
      {
        msg = toNotAssuredUpdateMsg(msg);
      }
      incrementOutCount();
      if (msg.isAssured())
      {
@@ -940,6 +963,37 @@
    return null;
  }

  /**
   * Substitutes a not assured version of the provided update message so that a
   * peer not expected to acknowledge it does not receive it with the assured
   * flag.
   * <p>
   * This is the counterpart, for the changelog catch-up path, of the
   * {@link NotAssuredUpdateMsg} substitution performed on the in-memory queue
   * path by ReplicationServerDomain.addUpdate(): updates re-read from the
   * changelog DB keep the assured flag of their original sender and must be
   * normalized in {@link #take()} before being handed to the ServerWriter.
   */
  private UpdateMsg toNotAssuredUpdateMsg(UpdateMsg msg)
  {
    try
    {
      return new NotAssuredUpdateMsg(msg);
    }
    catch (UnsupportedEncodingException e)
    {
      // Could not build the not assured form (unexpected message encoding).
      // Deliver the original message rather than dropping it: losing the
      // update would break replication consistency, which is worse than a
      // spurious ack - and such an ack is now safely ignored by
      // ExpectedAcksInfo.processReceivedAck().
      logger.error(LocalizableMessage.raw(
          "Could not substitute a not assured version of update message %s: %s",
          msg, stackTraceToSingleLineString(e)));
      return msg;
    }
  }

  private void acquirePermitInSendWindow()
  {
    boolean acquired = false;

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

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

@@ -51,6 +51,7 @@
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.DSInfo;
import org.opends.server.replication.common.RSInfo;
@@ -581,7 +582,7 @@
     * Sends a new update from this DS.
     * @throws TimeoutException If timeout waiting for an assured ack
     */
    private void sendNewFakeUpdate() throws TimeoutException
    private CSN sendNewFakeUpdate() throws TimeoutException
    {
      // Create a new delete update message (the simplest to create)
      DeleteMsg delMsg = new DeleteMsg(getBaseDN(), gen.newCSN(), UUID.randomUUID().toString());
@@ -590,6 +591,7 @@
      prepareWaitForAckIfAssuredEnabled(delMsg);
      publish(delMsg);
      waitForAckIfAssuredEnabled(delMsg);
      return delMsg.getCSN();
    }

    private void assertReceivedWrongUpdates(int expectedNbUpdates, int expectedNbWrongUpdates)
@@ -751,9 +753,14 @@
    private CSNGenerator gen;

    /** False if a received update had assured parameters not as expected. */
    private boolean everyUpdatesAreOk = true;
    /** Number of received updates. */
    private int nReceivedUpdates;
    private volatile boolean everyUpdatesAreOk = true;
    /**
     * Number of received updates. Volatile as it is incremented by the listen
     * thread and polled from the test thread (waitForReceivedUpdates()); a read
     * observing the incremented value also publishes everyUpdatesAreOk, which
     * is written before the increment.
     */
    private volatile int nReceivedUpdates;

    /**
     * True if an ack has been replied to a received assured update (in assured
@@ -1076,6 +1083,110 @@
  }

  /**
   * Parameters for {@link #testCatchUpClearsAssuredFlag}: assured mode and
   * safe data level of the update a peer catches up from the changelog.
   */
  @DataProvider(name = "catchUpAssuredModeProvider")
  private Object[][] catchUpAssuredModeProvider()
  {
    return new Object[][]
    {
      { AssuredMode.SAFE_DATA_MODE, 1 },
      { AssuredMode.SAFE_DATA_MODE, 2 },
      { AssuredMode.SAFE_READ_MODE, 1 },
    };
  }

  /**
   * Regression test for issue #710: an update served to a peer through the
   * changelog catch-up path must not keep the assured flag of its original
   * sender.
   * <p>
   * A fake RS connects with an empty server state after an assured update has
   * already been received and persisted by the real RS. As the fake RS was not
   * connected when the update was received, the update is re-read from the
   * changelog DB (catch-up path) rather than served from the in-memory queue,
   * and the fake RS is not an expected ack server for it. The update must
   * therefore be forwarded with its assured flag cleared (while its assured
   * mode and safe data level are preserved). This used to be delivered with
   * assured=true, making the fake RS reply a spurious ack. Exercised across
   * safe data (level 1 and level > 1) and safe read modes.
   */
  @Test(dataProvider = "catchUpAssuredModeProvider", enabled = true)
  public void testCatchUpClearsAssuredFlag(AssuredMode assuredMode,
      int safeDataLevel) throws Exception
  {
    String testCase = "testCatchUpClearsAssuredFlag";
    debugInfo("Starting " + testCase + " mode=" + assuredMode + " sdl=" + safeDataLevel);
    initTest();
    try
    {
      // Real RS to be tested
      rs1 = createReplicationServer(RS1_ID, DEFAULT_GID, SMALL_TIMEOUT, testCase, 0);

      // Main DS sends an assured update, acknowledged by the RS. No other peer
      // is connected yet, so the update only lands in the changelog DB, not in
      // any peer message queue.
      fakeRDs[1] = createFakeReplicationDomain(FDS1_ID, DEFAULT_GID, RS1_ID,
          DEFAULT_GENID, assuredMode, safeDataLevel, LONG_TIMEOUT, TIMEOUT_DS_SCENARIO);
      CSN csn = fakeRDs[1].sendNewFakeUpdate();

      // Ensure the update is persisted before the peer connects, so the peer is
      // served through the changelog catch-up path (the RS acks the DS before
      // persisting, so returning from sendNewFakeUpdate() does not guarantee it).
      waitForChangePersisted(rs1, csn);

      // A fake RS connects with an empty state: it must catch up the historical
      // update from the changelog DB. It expects a non-assured forward and
      // never replies an ack (TIMEOUT_RS_SCENARIO).
      fakeRs1 = createFakeReplicationServer(FRS1_ID, DEFAULT_GID, DEFAULT_GENID,
          false, assuredMode, safeDataLevel, TIMEOUT_RS_SCENARIO);

      // The historical update must have been received, with assured flag off
      waitForReceivedUpdates(fakeRs1, 1);
      fakeRs1.assertReceivedUpdates(1);
    }
    finally
    {
      endTest();
    }
  }

  /**
   * Waits until the provided change has been persisted in the changelog of the
   * provided replication server, i.e. is retrievable through the catch-up path.
   */
  private void waitForChangePersisted(ReplicationServer rs, CSN csn) throws Exception
  {
    ReplicationServerDomain domain =
        rs.getReplicationServerDomain(DN.valueOf(TEST_ROOT_DN_STRING));
    assertNotNull(domain);
    int i = 0;
    while (!domain.getLatestServerState().cover(csn))
    {
      if (i++ > 50)
      {
        Assert.fail("Change " + csn + " was not persisted in the changelog in time.");
      }
      Thread.sleep(100);
    }
  }

  /**
   * Waits until the provided fake RS has received at least the expected number
   * of updates, so the assertion on the updates does not race with delivery.
   */
  private void waitForReceivedUpdates(FakeReplicationServer fakeRs, int expected)
      throws Exception
  {
    int i = 0;
    while (fakeRs.nReceivedUpdates < expected && i++ <= 50)
    {
      Thread.sleep(100);
    }
  }

  /**
   * Returns possible combinations of parameters for testSafeDataLevelOne test.
   */
  @DataProvider(name = "testSafeDataLevelOneProvider")

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

New file
@@ -0,0 +1,85 @@
/*
 * 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]".
 *
 * Portions Copyright 2026 3A Systems, LLC
 */
package org.opends.server.replication.server;

import static org.mockito.Mockito.*;
import static org.testng.Assert.*;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.protocol.AckMsg;
import org.testng.annotations.Test;

/**
 * Test the handling of acks received from servers that were not expected to
 * acknowledge an assured update.
 * <p>
 * A peer served through the changelog catch-up path receives the update with
 * the assured flag of the original sender (see issue #710), so it may send
 * back an ack for a server id that is not part of the expected servers of the
 * matching {@link ExpectedAcksInfo}. That ack must be ignored instead of
 * raising a {@link NullPointerException} while auto-unboxing the missing map
 * entry, which would close the connection to the acking peer.
 */
@SuppressWarnings("javadoc")
public class ExpectedAcksInfoTest extends DirectoryServerTestCase
{
  private static final CSN CSN1 = new CSN(1, 1, 10);

  private static ServerHandler mockServerHandler(int serverId, boolean isDataServer)
  {
    ServerHandler handler = mock(ServerHandler.class);
    when(handler.getServerId()).thenReturn(serverId);
    when(handler.isDataServer()).thenReturn(isDataServer);
    return handler;
  }

  @Test
  public void safeReadAckFromNotExpectedServerIsIgnored()
  {
    ServerHandler requester = mockServerHandler(1, true);
    List<Integer> expectedServers = Arrays.asList(2, 3);
    SafeReadExpectedAcksInfo acksInfo = new SafeReadExpectedAcksInfo(
        CSN1, requester, expectedServers, Collections.<Integer> emptyList());

    // Server id 99 is not in the expected servers list
    ServerHandler notExpected = mockServerHandler(99, false);
    assertFalse(acksInfo.processReceivedAck(notExpected, new AckMsg(CSN1)),
        "An ack from a not expected server must not complete the ack info");
    assertFalse(acksInfo.isExpectedServer(99));
    assertTrue(acksInfo.isExpectedServer(2));
  }

  @Test
  public void safeDataAckFromNotExpectedServerIsIgnored()
  {
    ServerHandler requester = mockServerHandler(1, true);
    List<Integer> expectedServers = Arrays.asList(2, 3);
    SafeDataExpectedAcksInfo acksInfo = new SafeDataExpectedAcksInfo(
        CSN1, requester, (byte) 3, expectedServers);

    // Server id 99 is a RS not in the expected servers list
    ServerHandler notExpected = mockServerHandler(99, false);
    assertFalse(acksInfo.processReceivedAck(notExpected, new AckMsg(CSN1)),
        "An ack from a not expected server must not complete the ack info");
    assertFalse(acksInfo.isExpectedServer(99));
    assertTrue(acksInfo.isExpectedServer(3));
  }
}