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

Valery Kharseko

18 hours ago 69f502abd9219bf113a3b062d30068d699f79403

 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 @@
    {
      // initializations
      localGenerationId = -1;
      oldGenerationId = -100;

      // processes the ServerStart message received
      boolean sessionInitiatorSSLEncryption =
@@ -395,7 +395,6 @@
      lockDomainNoTimeout();

      localGenerationId = replicationServerDomain.getGenerationId();
      oldGenerationId = localGenerationId;

      if (replicationServerDomain.isAlreadyConnectedToDS(this))
      {
@@ -601,7 +600,7 @@
        // 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,18 +1739,68 @@
        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.
   *
   * @param senderHandler The handler associated to the server
@@ -2123,9 +2173,12 @@

  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 @@
        // 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 @@

    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 @@
      */
      if (localGenerationId < 0 && generationId > 0)
      {
        oldGenerationId =
            replicationServerDomain.changeGenerationId(generationId);
        setDomainGenerationIdOnStart(generationId);
      }

      logStartHandshakeSNDandRCV(outReplServerStartMsg,(ReplServerStartMsg)msg);
@@ -283,7 +285,6 @@
  public void startFromRemoteRS(ReplServerStartMsg inReplServerStartMsg)
  {
    localGenerationId = -1;
    oldGenerationId = -100;
    try
    {
      // The initiator decides if the session is encrypted
@@ -479,8 +480,7 @@
      // 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 @@
  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 @@
      localSession.close();
    }

    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 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.changeGenerationId(oldGenerationId);
      replicationServerDomain.rollbackGenerationIdIfUnchanged(
          generationIdSetOnStart, oldGenerationId);
      oldGenerationId = -100;
      generationIdSetOnStart = -100;
    }

    releaseDomainLock();
  }

  /**
   * 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)
    {
      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 @@
      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 @@
      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 @@
      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 @@
      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 @@
      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 @@

      // 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 @@
      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 @@
      {
        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 @@
        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 @@
        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 @@
      if (server2 == null)
      {
        server2 = openReplicationSession(baseDN,
          server2ID, 100, getReplServerPort(replServer1ID), 1000);
          server2ID, 100, getReplServerPort(replServer1ID), 10000);
      }

      // Creates config to synchronize suffix

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

New file
@@ -0,0 +1,175 @@
/*
 * 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.server;

import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;

import java.net.ServerSocket;
import java.net.Socket;
import java.util.TreeSet;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.StopMsg;
import org.testng.annotations.Test;

/**
 * Deterministic reproducer for the generation id rollback race fixed in the
 * same change (issue #735): an outgoing RS to RS handshake that is aborted
 * (e.g. rejected by the peer on a simultaneous cross-connect) must not roll
 * the domain generation id back to the value it had when the handshake
 * started, wiping a value that was legitimately adopted in the meantime.
 * <p>
 * The test plays the remote peer itself, so it fully controls the interleaving:
 * the connecting RS snapshots the domain generation id before sending its start
 * message, the test then changes the generation id while the RS is blocked
 * waiting for the answer, and only then rejects the handshake with a StopMsg.
 * The generation id advertised on the next connection attempt is sent strictly
 * after the abort completed on the same connector thread, so it exposes a
 * rollback deterministically, without any timing assumptions.
 */
@SuppressWarnings("javadoc")
public class HandshakeAbortGenerationIdTest extends ReplicationTestCase
{
  private static final long ADOPTED_GEN_ID = 4801;
  private static final int SOCKET_TIMEOUT_MS = 30000;

  @Test
  public void abortedConnectMustNotRollBackConcurrentlyAdoptedGenId() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try (ServerSocket fakePeer = TestCaseUtils.bindFreePort())
    {
      fakePeer.setSoTimeout(SOCKET_TIMEOUT_MS);

      final int rsPort = TestCaseUtils.findFreePort();
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          rsPort, "handshakeAbortGenIdTestDb", 0, 811, 0, 100,
          newTreeSet("127.0.0.1:" + fakePeer.getLocalPort())));
      // Creating the domain makes the RS connect thread dial the fake peer.
      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);

      final ReplSessionSecurity security = getReplSessionSecurity();
      try (Session session = accept(fakePeer, security))
      {
        // The RS is now inside connect(): it snapshotted the domain generation
        // id (-1) before sending its start message.
        final ReplServerStartMsg startMsg =
            waitForSpecificMsg(session, ReplServerStartMsg.class);
        assertEquals(startMsg.getGenerationId(), -1);

        // While connect() is blocked waiting for our answer, the domain adopts
        // a generation id, as happens when gossip arrives from another peer RS.
        domain.changeGenerationId(ADOPTED_GEN_ID);

        // Reject the handshake as a peer does on a simultaneous cross-connect.
        session.publish(new StopMsg());
      }

      // The second connection attempt is made by the same connector thread
      // strictly after abortStart() returned, so the generation id it
      // advertises deterministically shows whether the abort rolled back the
      // concurrently adopted value.
      try (Session session = accept(fakePeer, security))
      {
        final ReplServerStartMsg startMsg =
            waitForSpecificMsg(session, ReplServerStartMsg.class);
        assertEquals(startMsg.getGenerationId(), ADOPTED_GEN_ID,
            "the aborted handshake rolled back a generation id it did not set");
        session.publish(new StopMsg());
      }
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * The guards on the generation id primitives introduced for the abort
   * rollback: a compare-and-set never overwrites a value it did not expect,
   * and a rollback never clears a generation id that has been saved to the
   * changelog.
   */
  @Test
  public void generationIdPrimitivesHonorGuards() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
          TestCaseUtils.findFreePort(), "handshakeAbortGenIdPrimitivesDb", 0,
          812, 0, 100, new TreeSet<String>()));
      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);

      assertFalse(domain.changeGenerationIdIfUnchanged(123, 456),
          "CAS with a stale expected value must not fire");
      assertEquals(domain.getGenerationId(), -1);

      assertTrue(domain.changeGenerationIdIfUnchanged(-1, ADOPTED_GEN_ID));
      assertEquals(domain.getGenerationId(), ADOPTED_GEN_ID);

      domain.rollbackGenerationIdIfUnchanged(999, -1);
      assertEquals(domain.getGenerationId(), ADOPTED_GEN_ID,
          "rollback with a stale expected value must not fire");

      domain.rollbackGenerationIdIfUnchanged(ADOPTED_GEN_ID, -1);
      assertEquals(domain.getGenerationId(), -1,
          "rollback must fire when the value is unchanged and unsaved");

      domain.initGenerationID(ADOPTED_GEN_ID);
      domain.rollbackGenerationIdIfUnchanged(ADOPTED_GEN_ID, -1);
      assertEquals(domain.getGenerationId(), ADOPTED_GEN_ID,
          "rollback must never clear a saved generation id");
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /** Teardown must never mask the primary assertion failure. */
  private void removeQuietly(ReplicationServer replicationServer)
  {
    try
    {
      remove(replicationServer);
    }
    catch (Exception ignored)
    {
    }
  }

  private Session accept(ServerSocket listenSocket, ReplSessionSecurity security)
      throws Exception
  {
    final Socket socket = listenSocket.accept();
    socket.setTcpNoDelay(true);
    final Session session = security.createServerSession(socket, SOCKET_TIMEOUT_MS);
    assertNotNull(session, "could not create a session with the connecting RS");
    return session;
  }
}