From 69f502abd9219bf113a3b062d30068d699f79403 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 16 Jul 2026 05:16:04 +0000
Subject: [PATCH] [#735] Do not roll back a concurrently adopted generation ID on aborted handshake (#736)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortGenerationIdTest.java | 175 +++++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java | 26 +-
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java | 20 +-
opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java | 5
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java | 69 ++++++++-
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java | 69 ++++++++-
6 files changed, 321 insertions(+), 43 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java
index b6429e9..a68f2d2 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/DataServerHandler.java
+++ b/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;
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
index 81d6e48..336124d 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
+++ b/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;
+ }
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java
index dce9118..f7ca1ba 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerHandler.java
+++ b/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;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
index 6d348dd..ccb3ebd 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
+++ b/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;
}
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java
index 64f1695..c5215b0 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java
+++ b/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
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortGenerationIdTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortGenerationIdTest.java
new file mode 100644
index 0000000..848d379
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortGenerationIdTest.java
@@ -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;
+ }
+}
--
Gitblit v1.10.0