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

Author: vharseko

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 @@ public String errorsToString()
       "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;
@@ -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/MessageHandler.java

@@ -80,6 +80,13 @@ class MessageHandler extends MonitorProvider<MonitorProviderCfg>
   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. */
@@ -225,6 +232,21 @@ boolean isFollowing()
     }
   }
 
+  /**
+   * 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 @@ protected UpdateMsg getNextMessage() throws ChangelogException
                 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 @@ protected UpdateMsg getNextMessage() throws ChangelogException
           }
           if (updateServerState(msg))
           {
+            lastMessageFromLateQueue = true;
             return msg;
           }
           continue;
@@ -379,6 +405,7 @@ protected UpdateMsg getNextMessage() throws ChangelogException
              * 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;
 
@@ -789,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,32 @@ public void shutdown()
    */
   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 @@ public UpdateMsg take() throws ChangelogException
     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;