[#900] Time the ReplicaOfflineMsg grace period per replica, and spend it where the message can still be forwarded

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

@@ -1975,8 +1975,8 @@ void doPreOperation(PreOperationAddOperation addOperation)
   @Override
   public void publishReplicaOfflineMsg()
   {
-    pendingChanges.putReplicaOfflineMsg();
-    dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN());
+    final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
+    dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN);
   }
 
   /**

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

@@ -13,6 +13,7 @@
  *
  * Copyright 2009 Sun Microsystems, Inc.
  * Portions Copyright 2011-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.replication.plugin;
 
@@ -123,8 +124,10 @@ synchronized CSN putLocalOperation(PluginOperation operation)
 
   /**
    * Add a replica offline message to the pending list.
+   *
+   * @return the CSN of the message which was added
    */
-  public synchronized void putReplicaOfflineMsg()
+  public synchronized CSN putReplicaOfflineMsg()
   {
     final CSN offlineCSN = csnGenerator.newCSN();
     final PendingChange pendingChange =
@@ -133,6 +136,7 @@ public synchronized void putReplicaOfflineMsg()
 
     pendingChanges.put(offlineCSN, pendingChange);
     pushCommittedChanges();
+    return offlineCSN;
   }
 
   /**

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

@@ -178,6 +178,11 @@ public class ReplicationServer
 
   /**
    * Creates a new Replication server using the provided configuration entry.
+   * <p>
+   * The synchronization object this creates is its own, so the resulting server does not
+   * synchronize its shutdown with a collocated directory server. A server which has to must be
+   * built with {@link #ReplicationServer(ReplicationServerCfg, DSRSShutdownSync)}, passing the
+   * instance the directory server side records its ReplicaOfflineMsgs on.
    *
    * @param cfg The configuration of this replication server.
    * @throws ConfigException When Configuration is invalid.
@@ -802,6 +807,9 @@ private void abortInitialization()
     // listen port which cannot be bound, would otherwise leave them behind. Shut them down
     // before the changelog they write to, and one unchecked exception at a time: the changelog
     // this one is built on is known to be broken, and what follows still has to run.
+    // Nothing in an instance which never finished coming up can forward a pending
+    // ReplicaOfflineMsg, so this path does not wait for one: it would only delay the failure
+    // which is being reported by a grace period which cannot pay off.
     for (ReplicationServerDomain domain : getReplicationServerDomains())
     {
       try
@@ -1181,7 +1189,19 @@ public void shutdown()
       listenThread.interrupt();
     }
 
-    // shutdown all the replication domains
+    /*
+     * Let the ReplicaOfflineMsgs a collocated DS sent be forwarded while every handler is still
+     * up, and only then stop the domains: shutting a domain down deactivates the consumer of its
+     * handlers, clears their message queue and closes their session - see OPENDJ-1453. All the
+     * domains wait together and share one deadline, so the shutdown is bounded by one grace
+     * period and the wait of one domain does not spend the grace period of the next.
+     * <p>
+     * This also runs before the assured timer of any domain is cancelled, so an assured update
+     * still waiting for acks keeps timing out during the wait instead of holding its sender
+     * until the sessions are closed.
+     */
+    awaitReplicaOfflineMsgsForwarded();
+
     for (ReplicationServerDomain domain : getReplicationServerDomains())
     {
       domain.shutdown();
@@ -1202,6 +1222,28 @@ public void shutdown()
     allInstances.remove(this);
   }
 
+  /**
+   * Waits for the ReplicaOfflineMsg of every domain which has a replication server to forward it
+   * to. With no such server connected there is nobody to forward the message to, and waiting
+   * would only delay the shutdown by the whole grace period.
+   */
+  private void awaitReplicaOfflineMsgsForwarded()
+  {
+    final List<DN> domainsToWaitFor = new ArrayList<>();
+    for (ReplicationServerDomain domain : getReplicationServerDomains())
+    {
+      if (!domain.getConnectedRSs().isEmpty())
+      {
+        domainsToWaitFor.add(domain.getBaseDN());
+      }
+    }
+    if (!domainsToWaitFor.isEmpty())
+    {
+      dsrsShutdownSync.awaitReplicaOfflineMsgsForwarded(
+          domainsToWaitFor, dsrsShutdownSync.newShutdownDeadline());
+    }
+  }
+
   /**
    * Retrieves the time after which changes must be deleted from the
    * persistent storage (in milliseconds).

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

@@ -1661,7 +1661,15 @@ private MonitorMsg createLocalTopologyMonitorMsg(int sender, int destination)
     return monitorMsg;
   }
 
-  /** Shutdown this ReplicationServerDomain. */
+  /**
+   * Shutdown this ReplicationServerDomain.
+   * <p>
+   * A ReplicaOfflineMsg which a collocated DS sent and which is still to be forwarded must be
+   * waited for before this runs: stopping the server handlers deactivates their consumer, clears
+   * their message queue and closes their session, after which the message can no longer be sent
+   * - see OPENDJ-1453. ReplicationServer.shutdown() waits for the messages of all of its domains
+   * before it stops any of them.
+   */
   public void shutdown()
   {
     DirectoryServer.deregisterMonitorProvider(this);

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

@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2009 Sun Microsystems, Inc.
  * Portions Copyright 2011-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.replication.server;
 
@@ -87,25 +88,40 @@ public void run()
     LocalizableMessage errMessage = null;
     try
     {
-      boolean shutdown = false;
-      while (!shutdown
-          || !dsrsShutdownSync.canShutdown(replicationServerDomain.getBaseDN()))
+      /*
+       * Looping here to wait for a pending ReplicaOfflineMsg would achieve nothing: this writer
+       * only stops once its handler has been shut down, which deactivates the consumer, clears
+       * the message queue and closes the session. The shutdown of the domain waits for the
+       * message to be forwarded before it stops the handlers - see
+       * ReplicationServerDomain.shutdown() and OPENDJ-1453.
+       */
+      while (true)
       {
         final UpdateMsg updateMsg = this.handler.take();
         if (updateMsg == null)
         {
           // this connection is closing
           errMessage = LocalizableMessage.raw(
            "Connection closure: null update returned by domain.");
-          shutdown = true;
+          break;
         }
-        else if (!isUpdateMsgFiltered(updateMsg))
+        if (!isUpdateMsgFiltered(updateMsg))
         {
           // Publish the update to the remote server using a protocol version it supports
           session.publish(updateMsg);
-          if (updateMsg instanceof ReplicaOfflineMsg)
+          /*
+           * 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.
+           */
+          if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
           {
-            dsrsShutdownSync.replicaOfflineMsgForwarded(replicationServerDomain.getBaseDN());
+            dsrsShutdownSync.replicaOfflineMsgForwarded(
+                replicationServerDomain.getBaseDN(), updateMsg.getCSN());
           }
         }
       }