[#710] Fix replication catch-up re-sending updates with the original assured flag

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;
@@ -68,6 +72,15 @@ public abstract class ExpectedAcksInfo
    */
   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)
@@ -99,6 +112,22 @@ protected ExpectedAcksInfo(CSN csn, ServerHandler requesterServerHandler,
     {
       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/ReplicationServerDomain.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;
 
@@ -788,6 +789,20 @@ private boolean isDifferentGenerationId(long generationId)
     return this.generationId > 0 && this.generationId != generationId;
   }
 
+  /**
+   * 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.
    *

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 @@ public boolean processReceivedAck(ServerHandler ackingServer, AckMsg ackMsg)
 
     // 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 @@ public boolean processReceivedAck(ServerHandler ackingServer, AckMsg ackMsg)
   {
     // 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,23 @@ public void shutdown()
    */
   public UpdateMsg take() throws ChangelogException
   {
-    final UpdateMsg msg = getNextMessage();
+    UpdateMsg msg = getNextMessage();
 
     acquirePermitInSendWindow();
 
     if (msg != null)
     {
+      // Decide whether to keep the assured flag as late as possible, right
+      // before the message is handed to the ServerWriter for publishing, so
+      // the ack window check reflects the state at send time and not the
+      // potentially much earlier time the message was dequeued:
+      // acquirePermitInSendWindow() above may block longer than the assured
+      // timeout when the send window is closed.
+      if (msg.isAssured()
+          && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId))
+      {
+        msg = toNotAssuredUpdateMsg(msg);
+      }
       incrementOutCount();
       if (msg.isAssured())
       {
@@ -940,6 +954,40 @@ public UpdateMsg take() throws ChangelogException
     return null;
   }
 
+  /**
+   * Substitutes a not assured version of the provided update message so that a
+   * peer not (or no longer) 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. Doing the
+   * check here in {@link #take()} makes it the single guard covering both
+   * paths - on the queue path the message posted to a non-eligible peer is
+   * already a {@code NotAssuredUpdateMsg}, so {@code isAssured()} is false and
+   * this method is not even reached.
+   */
+  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 @@ private void checkUpdateAssuredParameters(UpdateMsg updateMsg)
      * 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 @@ private void sendNewFakeUpdate() throws TimeoutException
       prepareWaitForAckIfAssuredEnabled(delMsg);
       publish(delMsg);
       waitForAckIfAssuredEnabled(delMsg);
+      return delMsg.getCSN();
     }
 
     private void assertReceivedWrongUpdates(int expectedNbUpdates, int expectedNbWrongUpdates)
@@ -751,9 +753,14 @@ private class FakeReplicationServer extends Thread
     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
@@ -1075,6 +1082,110 @@ public void testSafeDataLevelOnePrecommit() throws Exception
     testSafeDataLevelOne(DEFAULT_GID, false, false, DEFAULT_GID, DEFAULT_GID);
   }
 
+  /**
+   * 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.
    */