[#950] Announce a ReplicaOfflineMsg before it is published, not after it may have been forwarded

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

@@ -828,7 +828,8 @@ public void run()
      * The generator time is adjusted to the time of the last CSN received from
      * remote other servers.
      */
-    pendingChanges = new PendingChanges(getGenerator(), this);
+    pendingChanges = new PendingChanges(getGenerator(), this,
+        new ShutdownSyncAnnouncer(dsrsShutdownSync, getBaseDN()));
     remotePendingChanges = new RemotePendingChanges(getServerState());
 
     // listen for changes on the configuration
@@ -2271,17 +2272,13 @@ void doPreOperation(PreOperationAddOperation addOperation)
   public void publishReplicaOfflineMsg()
   {
     final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
-    if (offlineCSN != null)
+    if (offlineCSN == null && logger.isTraceEnabled())
     {
       /*
-       * Only a message which really was published is announced: the shutdown of a collocated
-       * replication server waits for it to be forwarded, and would spend the whole grace
-       * period waiting for one which never reached the wire.
+       * The announcement itself is made where the message is published, so nothing has to be
+       * reported here: a message a change in flight held back was never announced, and the
+       * announcement of one the broker refused has been withdrawn.
        */
-      dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN);
-    }
-    else if (logger.isTraceEnabled())
-    {
       logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
           + " could not announce itself offline: the message was not published - a change which"
           + " is still in flight holds it back, or the broker had no session to write it to,"

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

@@ -53,18 +53,26 @@ class PendingChanges
   /** The ReplicationDomain that will be used to send UpdateMsg. */
   private final ReplicationDomain domain;
 
+  /** Told that the replica of this domain announces itself offline, or takes that back. */
+  private final ReplicaOfflineAnnouncer replicaOfflineAnnouncer;
+
   private boolean recoveringOldChanges;
 
   /**
    * Creates a new PendingChanges using the provided CSNGenerator.
    *
    * @param csnGenerator The CSNGenerator to use to create new unique CSNs.
    * @param domain  The ReplicationDomain that will be used to send UpdateMsg.
+   * @param replicaOfflineAnnouncer Told that the replica of this domain announces itself
+   *                  offline, before the message announcing it is published, and that it takes
+   *                  the announcement back when the broker refused the message.
    */
-  PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain)
+  PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain,
+      ReplicaOfflineAnnouncer replicaOfflineAnnouncer)
   {
     this.csnGenerator = csnGenerator;
     this.domain = domain;
+    this.replicaOfflineAnnouncer = replicaOfflineAnnouncer;
   }
 
   /**
@@ -198,9 +206,22 @@ synchronized CSN pushCommittedChanges()
       }
       else if (msg instanceof ReplicaOfflineMsg)
       {
+        /*
+         * Announce the replica offline before the message reaches the wire, and not after:
+         * a collocated replication server forwards it as soon as it has it, and a forward
+         * which finds nothing announced leaves the shutdown waiting out the whole grace
+         * period of a message the topology already has.
+         */
+        final CSN offlineCSN = msg.getCSN();
+        replicaOfflineAnnouncer.announce(offlineCSN);
         if (domain.publish(msg))
         {
-          publishedOfflineCSN = msg.getCSN();
+          publishedOfflineCSN = offlineCSN;
+        }
+        else
+        {
+          // The broker wrote it to no session, so nobody will forward what was announced.
+          replicaOfflineAnnouncer.withdraw(offlineCSN);
         }
       }
 
@@ -266,4 +287,34 @@ synchronized boolean recoveryUntil(CSN recovered)
     }
     return recoveringOldChanges;
   }
+
+  /**
+   * Told that the replica of this domain announces itself offline, or takes that back.
+   * <p>
+   * A collocated replication server can forward a {@link ReplicaOfflineMsg} as soon as it is on
+   * the wire, and its shutdown waits for that forward, so the announcement has to be in place
+   * before the message is published: one made afterwards is one the forward found nothing to
+   * clear, and the shutdown spends its whole grace period on a message which has already gone
+   * out. The broker may still refuse the message once it is announced, and then the
+   * announcement is withdrawn: what stays announced is what the broker reports as written.
+   */
+  interface ReplicaOfflineAnnouncer
+  {
+    /**
+     * Announces that the replica goes offline at the provided CSN.
+     *
+     * @param offlineCSN
+     *          the CSN of the ReplicaOfflineMsg which is about to be published
+     */
+    void announce(CSN offlineCSN);
+
+    /**
+     * Withdraws the announcement of a message the broker refused: it was written to no session,
+     * so nobody will forward it.
+     *
+     * @param offlineCSN
+     *          the CSN of the ReplicaOfflineMsg which was announced and not published
+     */
+    void withdraw(CSN offlineCSN);
+  }
 }

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

@@ -0,0 +1,62 @@
+/*
+ * 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.plugin;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.plugin.PendingChanges.ReplicaOfflineAnnouncer;
+import org.opends.server.replication.service.DSRSShutdownSync;
+
+/**
+ * Announces the replica of one domain offline to the {@link DSRSShutdownSync} the shutdown of a
+ * collocated replication server waits on, and takes such an announcement back.
+ * <p>
+ * This is the announcer {@link LDAPReplicationDomain} hands its {@link PendingChanges}: the
+ * announcement goes through {@link DSRSShutdownSync#replicaOfflineMsgSent(DN, CSN)} and the
+ * withdrawal through {@link DSRSShutdownSync#replicaOfflineMsgNotSent(DN, CSN)}, for the domain
+ * the announcer was built for.
+ */
+final class ShutdownSyncAnnouncer implements ReplicaOfflineAnnouncer
+{
+  private final DSRSShutdownSync shutdownSync;
+  private final DN baseDN;
+
+  /**
+   * Creates an announcer for the replica of one domain.
+   *
+   * @param shutdownSync
+   *          the synchronization object the collocated replication server's shutdown waits on
+   * @param baseDN
+   *          the domain whose replica announces itself
+   */
+  ShutdownSyncAnnouncer(DSRSShutdownSync shutdownSync, DN baseDN)
+  {
+    this.shutdownSync = shutdownSync;
+    this.baseDN = baseDN;
+  }
+
+  @Override
+  public void announce(CSN offlineCSN)
+  {
+    shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN);
+  }
+
+  @Override
+  public void withdraw(CSN offlineCSN)
+  {
+    shutdownSync.replicaOfflineMsgNotSent(baseDN, offlineCSN);
+  }
+}

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

@@ -46,7 +46,7 @@ public class DSRSShutdownSync
 {
   /**
    * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated
-   * RS, in milliseconds, counted from the moment the message was sent.
+   * RS, in milliseconds, counted from the moment the message was announced.
    */
   public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000;
 
@@ -97,19 +97,90 @@ public DSRSShutdownSync()
   }
 
   /**
-   * Message has been sent.
+   * Message is about to be sent.
+   * <p>
+   * The announcement comes before the message is published rather than after: a collocated
+   * replication server can forward the message as soon as it is on the wire, and a forward which
+   * finds nothing announced has nothing to clear. The announcement of a message the broker then
+   * refuses is taken back by {@link #replicaOfflineMsgNotSent(DN, CSN)}.
+   * <p>
+   * A replica announces itself offline on every disableService(), so this may take the place of
+   * an earlier announcement of the same replica which is still owed its forward. The earlier one
+   * is kept behind the new one: a forward of the newer message, which the replication server
+   * queued behind the earlier one, covers both, and a withdrawal of the newer one gives the
+   * earlier one its wait back. It is kept only while its own grace period runs: past it, the
+   * announcement holds nothing back any more, and keeping it would chain every announcement of
+   * a replica whose message nobody in this process forwards - a directory server without a
+   * collocated replication server, or connected to a remote one - for the life of the process.
    *
    * @param baseDN
-   *          the domain for which the message has been sent
+   *          the domain for which the message is being sent
    * @param offlineCSN
-   *          the CSN of the message, which identifies both the replica which announced itself
+   *          the CSN of the message, which identifies both the replica which announces itself
    *          offline and the announcement being waited for
    */
   public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN)
   {
+    final long announcedAt = System.nanoTime();
     replicaOfflineMsgs
         .computeIfAbsent(baseDN, dn -> new ConcurrentHashMap<Integer, PendingOfflineMsg>())
-        .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime()));
+        .compute(offlineCSN.getServerId(), (serverId, displaced) ->
+            new PendingOfflineMsg(offlineCSN, announcedAt,
+                displaced != null && gracePeriodLeft(displaced, announcedAt) > 0 ? displaced : null));
+  }
+
+  /**
+   * The message which was announced was not sent after all: the broker had no session to write
+   * it to, or was stopped before it could.
+   * <p>
+   * The announcement is made before the message is published, since a collocated replication
+   * server can forward it as soon as it is on the wire, so the announcement of a message the
+   * broker then refused has to be taken back: nobody will forward it, and the shutdown would
+   * spend the whole grace period waiting for that forward. Only the announcement carrying that
+   * CSN is withdrawn, and the announcement it displaced - an earlier message of the same replica
+   * which did go out and is still owed its forward - takes its place again.
+   * <p>
+   * Whatever is reported about that earlier message while the announcement of the refused one
+   * stands in its place is not seen by it. A forward, or the loss of a peer it was queued for,
+   * is lost, and the shutdown then waits out what is left of the earlier message's own grace
+   * period; the peers it is queued for, if they are recorded in that window, are lost too, with
+   * the opposite effect - the first forward ends its wait, as for a message no peer was recorded
+   * for. That window is the one publish the broker refuses: at once on a connection error or a
+   * pending recovery, the broker's retry loop up to the reconnect when it has no session. The
+   * wait it can cost is bounded by a grace period which is already running.
+   *
+   * @param baseDN
+   *          the domain for which the message was announced
+   * @param offlineCSN
+   *          the CSN of the message which was not sent
+   */
+  public void replicaOfflineMsgNotSent(DN baseDN, CSN offlineCSN)
+  {
+    final ConcurrentMap<Integer, PendingOfflineMsg> msgs = replicaOfflineMsgs.get(baseDN);
+    if (msgs != null)
+    {
+      final int serverId = offlineCSN.getServerId();
+      final PendingOfflineMsg pending = msgs.get(serverId);
+      if (pending != null && pending.csn.equals(offlineCSN))
+      {
+        /*
+         * The displaced announcement may owe nothing any more: the forward which released it
+         * can have been reported while this announcement was being made, so that its remove(),
+         * which matches the entry it read, found this one in its place. Given its place back,
+         * such an announcement would hold the shutdown for the rest of its grace period, since
+         * nobody will report that forward again.
+         */
+        if (pending.displaced != null && !pending.displaced.isFullyForwarded())
+        {
+          msgs.replace(serverId, pending, pending.displaced);
+        }
+        else
+        {
+          msgs.remove(serverId, pending);
+        }
+      }
+    }
+    notifyForwarded();
   }
 
   /**
@@ -338,11 +409,20 @@ private long remainingGracePeriod(DN baseDN)
     long remaining = 0;
     for (PendingOfflineMsg pending : msgs.values())
     {
-      remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime));
+      remaining = Math.max(remaining, gracePeriodLeft(pending, now));
     }
     return remaining;
   }
 
+  /**
+   * Returns the time left, in milliseconds, of the grace period of one announcement, zero or
+   * less once it has expired.
+   */
+  private long gracePeriodLeft(PendingOfflineMsg pending, long now)
+  {
+    return gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime);
+  }
+
   /**
    * A ReplicaOfflineMsg a replica announced and which has not been forwarded yet.
    * <p>
@@ -356,16 +436,23 @@ private static final class PendingOfflineMsg
     private final CSN csn;
     /** When the message was announced, on the {@link System#nanoTime()} clock. */
     private final long sentTime;
+    /**
+     * The announcement of the same replica this one took the place of and which is still owed its
+     * forward, null when there was none or when its grace period had already expired. It is
+     * given its place back if this message is withdrawn.
+     */
+    private final PendingOfflineMsg displaced;
     /**
      * The replication servers the message was queued for and which have not forwarded it yet,
      * null as long as it has not been queued for anybody.
      */
     private volatile Set<Integer> awaitedForwarders;
 
-    private PendingOfflineMsg(CSN csn, long sentTime)
+    private PendingOfflineMsg(CSN csn, long sentTime, PendingOfflineMsg displaced)
     {
       this.csn = csn;
       this.sentTime = sentTime;
+      this.displaced = displaced;
     }
 
     /**
@@ -391,9 +478,8 @@ private boolean forwardedBy(int replicationServerId)
       {
         /*
          * The message never went through the collocated RS - a replica which picked a remote one
-         * announcing itself offline, or an announcement recorded after the message it belongs to
-         * was already relayed. Nobody is known to owe a forward, so keep the behaviour the wait
-         * had before the recipients were tracked: the first forward ends it.
+         * is announcing itself offline. Nobody is known to owe a forward, so keep the behaviour
+         * the wait had before the recipients were tracked: the first forward ends it.
          */
         return true;
       }
@@ -412,6 +498,16 @@ private boolean giveUpOn(int replicationServerId)
       return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty();
     }
 
+    /**
+     * Returns whether every replication server the message was queued for has forwarded it, or
+     * has been given up on: nothing is left to wait for. False while no recipient is known.
+     */
+    private boolean isFullyForwarded()
+    {
+      final Set<Integer> awaited = awaitedForwarders;
+      return awaited != null && awaited.isEmpty();
+    }
+
     @Override
     public String toString()
     {