[#735] Do not roll back a concurrently adopted generation ID on aborted handshake

opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.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;
 
@@ -365,7 +366,6 @@ public void startFromRemoteDS(ServerStartMsg inServerStartMsg)
     {
       // initializations
       localGenerationId = -1;
-      oldGenerationId = -100;
 
       // processes the ServerStart message received
       boolean sessionInitiatorSSLEncryption =
@@ -395,7 +395,6 @@ public void startFromRemoteDS(ServerStartMsg inServerStartMsg)
       lockDomainNoTimeout();
 
       localGenerationId = replicationServerDomain.getGenerationId();
-      oldGenerationId = localGenerationId;
 
       if (replicationServerDomain.isAlreadyConnectedToDS(this))
       {
@@ -601,7 +600,7 @@ private StartSessionMsg waitAndProcessStartSessionFromRemoteDS()
         // WARNING: Must be done before computing topo message to send
         // to peer server as topo message must embed valid generation id
         // for our server
-        oldGenerationId = replicationServerDomain.changeGenerationId(generationId);
+        setDomainGenerationIdOnStart(generationId);
       }
     }
     return startSessionMsg;

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

@@ -1739,17 +1739,67 @@ public long changeGenerationId(long generationId)
         this.generationIdSavedStatus = false;
 
         // generationId gossip is purely event-driven: it only travels in the
-        // topology messages sent on connect/disconnect/status events. Re-advertise
-        // on every real transition so a peer that missed one converges on the next.
-        if (generationId > 0)
-        {
-          sendTopoInfoToAll();
-        }
+        // topology messages sent on connect/disconnect/status events.
+        // Re-advertise on every real transition — including a reset or
+        // rollback to -1 — so peers that recorded the previous value converge
+        // instead of diverging until the next unrelated topology event.
+        sendTopoInfoToAll();
       }
       return oldGenerationId;
     }
   }
 
+  /**
+   * Sets the provided value as the new in memory generationId, but only if the
+   * current generation id still equals {@code expectedGenerationId}: a
+   * decision made on a stale snapshot must never overwrite a value another
+   * thread legitimately set in the meantime.
+   *
+   * @param expectedGenerationId The generation id the caller based its
+   *        decision on.
+   * @param newGenerationId The new value of generationId.
+   * @return whether the generation id was changed
+   */
+  public boolean changeGenerationIdIfUnchanged(long expectedGenerationId,
+      long newGenerationId)
+  {
+    synchronized (generationIDLock)
+    {
+      if (this.generationId != expectedGenerationId)
+      {
+        return false;
+      }
+      changeGenerationId(newGenerationId);
+      return true;
+    }
+  }
+
+  /**
+   * Rolls the generation id back to {@code oldGenerationId}, but only if it
+   * still has the {@code expectedGenerationId} value the caller previously set.
+   * An aborting handshake must undo its own change without overwriting a value
+   * that another thread legitimately set in the meantime (e.g. adopted from a
+   * peer topology message). The rollback is also skipped once the generation
+   * id has been saved to the changelog or while data servers are connected —
+   * the same invariant {@link #resetGenerationIdIfPossible()} enforces —
+   * because rolling back would clear a changelog that now holds real changes.
+   *
+   * @param expectedGenerationId The generation id the caller set and expects
+   *        to still be in place.
+   * @param oldGenerationId The value to restore.
+   */
+  public void rollbackGenerationIdIfUnchanged(long expectedGenerationId,
+      long oldGenerationId)
+  {
+    synchronized (generationIDLock)
+    {
+      if (!generationIdSavedStatus && connectedDSs.isEmpty())
+      {
+        changeGenerationIdIfUnchanged(expectedGenerationId, oldGenerationId);
+      }
+    }
+  }
+
   /**
    * Resets the generationID.
    *
@@ -2123,9 +2173,12 @@ public void receiveTopoInfoFromRS(TopologyMsg topoMsg,
 
   private void setGenerationIdIfUnset(long generationId)
   {
-    if (this.generationId < 0)
+    synchronized (generationIDLock)
     {
-      this.generationId = generationId;
+      if (this.generationId < 0)
+      {
+        this.generationId = generationId;
+      }
     }
   }
 

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.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;
 
@@ -87,8 +88,6 @@ private boolean processStartFromRemote(
         // Only V2 protocol has the group id in repl server start message
         this.groupId = inReplServerStartMsg.getGroupId();
       }
-
-      oldGenerationId = -100;
     }
     catch(Exception e)
     {
@@ -150,13 +149,17 @@ public void connect(DN baseDN, boolean sslEncryption)
 
     setBaseDNAndDomain(baseDN, false);
 
-    localGenerationId = replicationServerDomain.getGenerationId();
-    oldGenerationId = localGenerationId;
-
     try
     {
       lockDomainNoTimeout();
 
+      // Read under the domain lock so the start message advertises any change
+      // made by a previous handshake (handshakes serialize on this lock). The
+      // field itself is only guarded by generationIDLock and lock-free
+      // adopters can still move it — the arming CAS in
+      // setDomainGenerationIdOnStart handles that residual race.
+      localGenerationId = replicationServerDomain.getGenerationId();
+
       ReplServerStartMsg outReplServerStartMsg = sendStartToRemote();
 
       // Wait answer
@@ -196,8 +199,7 @@ public void connect(DN baseDN, boolean sslEncryption)
       */
       if (localGenerationId < 0 && generationId > 0)
       {
-        oldGenerationId =
-            replicationServerDomain.changeGenerationId(generationId);
+        setDomainGenerationIdOnStart(generationId);
       }
 
       logStartHandshakeSNDandRCV(outReplServerStartMsg,(ReplServerStartMsg)msg);
@@ -283,7 +285,6 @@ TopologyMsg then TopologyMsg (with a RS)
   public void startFromRemoteRS(ReplServerStartMsg inReplServerStartMsg)
   {
     localGenerationId = -1;
-    oldGenerationId = -100;
     try
     {
       // The initiator decides if the session is encrypted
@@ -479,8 +480,7 @@ private void checkGenerationId()
       // The local RS is not initialized - take the one received
       // WARNING: Must be done before computing topo message to send to peer
       // server as topo message must embed valid generation id for our server
-      oldGenerationId =
-          replicationServerDomain.changeGenerationId(generationId);
+      setDomainGenerationIdOnStart(generationId);
       return;
     }
 

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

@@ -148,8 +148,20 @@ public abstract class ServerHandler extends MessageHandler
   protected long generationId = -1;
   /** The generation id of the hosting RS. */
   protected long localGenerationId = -1;
-  /** The generation id before processing a new start handshake. */
-  protected long oldGenerationId = -1;
+  /**
+   * The domain generation id that this handler replaced during the start
+   * handshake, or -100 when this handler has not changed the domain generation
+   * id and there is nothing to roll back on {@link #abortStart(LocalizableMessage)}.
+   */
+  protected long oldGenerationId = -100;
+  /**
+   * The generation id this handler wrote into the domain during the start
+   * handshake, or -100 when it wrote none. Used by
+   * {@link #abortStart(LocalizableMessage)} to roll back only this handler's
+   * own change: a value concurrently set by another thread must not be
+   * overwritten with the stale {@link #oldGenerationId}.
+   */
+  private long generationIdSetOnStart = -100;
   /** Group id of this remote server. */
   protected byte groupId = -1;
   /** The SSL encryption after the negotiation with the peer. */
@@ -219,15 +231,54 @@ protected void abortStart(LocalizableMessage reason)
       localSession.close();
     }
 
+    // If this handler changed the domain generation id during the handshake,
+    // set it back to the old value. Only undo our own change: another thread
+    // may have legitimately changed the generation id in the meantime (e.g.
+    // adopted it from a peer topology message) and that value must not be
+    // overwritten with our stale snapshot. Do it BEFORE releasing the domain
+    // lock, so a handshake queued on the lock cannot read, advertise or arm on
+    // the doomed value in between.
+    if (oldGenerationId != -100)
+    {
+      replicationServerDomain.rollbackGenerationIdIfUnchanged(
+          generationIdSetOnStart, oldGenerationId);
+      oldGenerationId = -100;
+      generationIdSetOnStart = -100;
+    }
+
     releaseDomainLock();
+  }
 
-    // If generation id of domain was changed, set it back to old value
-    // We may have changed it as it was -1 and we received a value >0 from peer
-    // server and the last topo message sent may have failed being sent: in that
-    // case retrieve old value of generation id for replication server domain
-    if (oldGenerationId != -100)
+  /**
+   * Changes the domain generation id during the start handshake, remembering
+   * the replaced value so that {@link #abortStart(LocalizableMessage)} can
+   * undo this handler's own change (and only it) if the handshake
+   * subsequently fails.
+   * <p>
+   * The change is applied only if the domain generation id still equals
+   * {@link #localGenerationId}, the value this handler based its decision on:
+   * a generation id concurrently adopted by a lock-free path (an update
+   * received from an already connected server) must not be stomped. This also
+   * makes a repeated call within one handshake a no-op, preserving the first
+   * arming's rollback point. Wire values below -1 are rejected — they can only
+   * come from a malformed peer and would collide with the -100 sentinel.
+   *
+   * @param generationId the generation id to set on the domain
+   */
+  protected void setDomainGenerationIdOnStart(long generationId)
+  {
+    if (generationId < -1)
     {
-      replicationServerDomain.changeGenerationId(oldGenerationId);
+      return;
+    }
+    if (replicationServerDomain.changeGenerationIdIfUnchanged(
+        localGenerationId, generationId))
+    {
+      if (oldGenerationId == -100)
+      {
+        oldGenerationId = localGenerationId;
+      }
+      generationIdSetOnStart = generationId;
     }
   }
 

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

@@ -555,7 +555,7 @@ public void initializeImport() throws Exception
       if (server2 == null)
       {
         server2 = openReplicationSession(baseDN,
-          server2ID, 100, getReplServerPort(replServer1ID), 1000);
+          server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       // In S1 launch the total update
@@ -608,7 +608,7 @@ public void initializeExport() throws Exception
       if (server2 == null)
       {
         server2 = openReplicationSession(baseDN,
-          server2ID, 100, getReplServerPort(replServer1ID), 1000);
+          server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       InitializeRequestMsg initMsg = new InitializeRequestMsg(baseDN, server2ID, server1ID, 100);
@@ -652,7 +652,7 @@ public void initializeTargetExport() throws Exception
       if (server2 == null)
       {
         server2 = openReplicationSession(baseDN,
-          server2ID, 100, getReplServerPort(replServer1ID), 1000);
+          server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       // Launch in S1 the task that will initialize S2
@@ -705,13 +705,13 @@ public void initializeTargetExportAll() throws Exception
       if (server2 == null)
       {
         server2 = openReplicationSession(baseDN,
-          server2ID, 100, getReplServerPort(replServer1ID), 1000);
+          server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       if (server3 == null)
       {
         server3 = openReplicationSession(baseDN,
-          server3ID, 100, getReplServerPort(replServer1ID), 1000);
+          server3ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       // Launch in S1 the task that will initialize S2
@@ -752,7 +752,7 @@ public void initializeTargetImport() throws Exception
       if (server2==null)
       {
         server2 = openReplicationSession(baseDN,
-          server2ID, 100, getReplServerPort(replServer1ID), 1000);
+          server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       // Creates config to synchronize suffix
@@ -941,11 +941,11 @@ public void testReplServerInfos() throws Exception
 
       // Connects lDAP2 to replServer2
       broker2 = openReplicationSession(baseDN,
-        server2ID, 100, getReplServerPort(replServer2ID), 1000);
+        server2ID, 100, getReplServerPort(replServer2ID), 10000);
 
       // Connects lDAP3 to replServer2
       broker3 = openReplicationSession(baseDN,
-        server3ID, 100, getReplServerPort(replServer2ID), 1000);
+        server3ID, 100, getReplServerPort(replServer2ID), 10000);
 
       // Check that the list of connected LDAP servers is correct in each replication servers
       Assertions.assertThat(getConnectedDSServerIds(replServer1)).containsExactly(server1ID);
@@ -958,7 +958,7 @@ public void testReplServerInfos() throws Exception
       Assertions.assertThat(getConnectedDSServerIds(replServer2)).containsExactly(server2ID);
 
       broker3 = openReplicationSession(baseDN,
-        server3ID, 100, getReplServerPort(replServer2ID), 1000);
+        server3ID, 100, getReplServerPort(replServer2ID), 10000);
       broker2.stop();
       Thread.sleep(1000);
       Assertions.assertThat(getConnectedDSServerIds(replServer2)).containsExactly(server3ID);
@@ -1001,7 +1001,7 @@ public void initializeTargetExportMultiSS() throws Exception
       {
         log(testCase + " Will connect server 2 to " + replServer2ID);
         server2 = openReplicationSession(baseDN,
-            server2ID, 100, getReplServerPort(replServer2ID), 1000);
+            server2ID, 100, getReplServerPort(replServer2ID), 10000);
       }
 
       // Launch in S1 the task that will initialize S2
@@ -1108,7 +1108,7 @@ public void initializeExportMultiSS() throws Exception
         log(testCase + " Will connect server 2 to " + replServer2ID);
         server2 = openReplicationSession(baseDN,
           server2ID, 100, getReplServerPort(replServer2ID),
-          1000, replServer1.getGenerationId(baseDN));
+          10000, replServer1.getGenerationId(baseDN));
       }
 
       // Connect a broker acting as server 3 to Repl Server 3
@@ -1120,7 +1120,7 @@ public void initializeExportMultiSS() throws Exception
         log(testCase + " Will connect server 3 to " + replServer3ID);
         server3 = openReplicationSession(baseDN,
           server3ID, 100, getReplServerPort(replServer3ID),
-          1000, replServer1.getGenerationId(baseDN));
+          10000, replServer1.getGenerationId(baseDN));
       }
 
       // S3 sends init request
@@ -1279,7 +1279,7 @@ public void initializeSimultaneous() throws Exception
       if (server2 == null)
       {
         server2 = openReplicationSession(baseDN,
-          server2ID, 100, getReplServerPort(replServer1ID), 1000);
+          server2ID, 100, getReplServerPort(replServer1ID), 10000);
       }
 
       // Creates config to synchronize suffix