From 2b20c60a8fe4d5d62f04d0f80cd45267adf0478f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 04 Aug 2026 08:26:25 +0000
Subject: [PATCH] [#821] Unregister the handler when a handshake aborts after registration (#838)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortRegistrationTest.java | 301 +++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java | 10 +
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java | 53 +++++++
3 files changed, 364 insertions(+), 0 deletions(-)
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 01dd007..6d0a2fd 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
@@ -1165,6 +1165,59 @@
}
/**
+ * Removes a handler registered by a handshake that subsequently failed and
+ * performs the same cleanup as {@link #stopServer(ServerHandler, boolean)}
+ * would have done. Without this cleanup the handler stays registered
+ * forever: the reader and writer threads that normally trigger
+ * {@link #stopServer(ServerHandler, boolean)} when the session dies were
+ * never started, so the dead server keeps being advertised to the whole
+ * topology and its server id can never reconnect.
+ * <p>
+ * Unlike {@link #stopServer(ServerHandler, boolean)} this method never
+ * acquires the domain lock: it is called from the handshake thread, which
+ * already holds the lock whenever the handler is registered, and acquiring
+ * it interruptibly would silently skip the cleanup when the handshake was
+ * aborted by an interrupt — the very trigger being cleaned up after.
+ * <p>
+ * The removal only fires when the domain still holds this very handler
+ * instance, so calling it for a handshake aborted before registration never
+ * evicts a legitimately connected server with the same server id.
+ *
+ * @param sHandler the handler whose handshake failed after registration
+ */
+ void unregisterFailedHandshake(ServerHandler sHandler)
+ {
+ final boolean isDataServer = !sHandler.isReplicationServer();
+ final Map<Integer, ? extends ServerHandler> connectedServers =
+ isDataServer ? connectedDSs : connectedRSs;
+ if (!connectedServers.remove(sHandler.getServerId(), sHandler))
+ {
+ return;
+ }
+
+ if (connectedDSs.isEmpty() && connectedRSs.isEmpty())
+ {
+ stopMonitoringPublisher();
+ }
+ sHandler.shutdown();
+
+ resetGenerationIdIfPossible();
+ synchronized (pendingStatusMessagesLock)
+ {
+ if (isDataServer)
+ {
+ // Update the remote replication servers with our list
+ // of connected LDAP servers
+ pendingStatusMessages.enqueueTopoInfoToAllRSs();
+ }
+ // Warn our DSs that a RS or DS has quit (does not use this
+ // handler as already removed from list)
+ pendingStatusMessages.enqueueTopoInfoToAllDSsExcept(null);
+ }
+ statusAnalyzer.notifyPendingStatusMessage();
+ }
+
+ /**
* This method resets the generationId for this domain if there is no LDAP
* server currently connected in the whole topology on this domain and if the
* generationId has never been saved.
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 8037b16..6943c03 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
@@ -246,6 +246,16 @@
generationIdSetOnStart = -100;
}
+ // A handshake that got as far as registering this handler in the domain
+ // must unregister it on abort: the reader and writer threads that
+ // normally trigger the cleanup when the session dies were never started,
+ // so nothing else will ever remove it. Do it BEFORE releasing the domain
+ // lock, so a handshake queued on the lock never sees the dead handler.
+ if (replicationServerDomain != null)
+ {
+ replicationServerDomain.unregisterFailedHandshake(this);
+ }
+
releaseDomainLock();
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortRegistrationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortRegistrationTest.java
new file mode 100644
index 0000000..0dd524a
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/HandshakeAbortRegistrationTest.java
@@ -0,0 +1,301 @@
+/*
+ * 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.testng.Assert.*;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.TreeSet;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.protocol.ReplSessionSecurity;
+import org.opends.server.replication.protocol.Session;
+import org.testng.annotations.Test;
+
+/**
+ * Reproducer for issue #821: a handshake that registered its handler in the
+ * replication server domain and failed afterwards (the only failure left in
+ * that window is {@code finalizeStart()}, before the reader and writer threads
+ * are running) used to leave the dead handler registered forever. Nothing else
+ * ever removes it: the normal cleanup is the reader or writer noticing the
+ * dead session and calling {@code stopServer()}, and neither thread was
+ * started. The stale entry inflates the connected DS count advertised to every
+ * connecting DS, is published to the whole topology, keeps the generation id
+ * from ever being reset, and permanently refuses reconnection of the same
+ * server id with ERR_DUPLICATE_SERVER_ID.
+ * <p>
+ * The tests drive {@code abortStart()} directly on handlers registered exactly
+ * as the handshake code registers them, so they pin the contract without any
+ * timing assumptions: an abort after registration must unregister the handler
+ * and run the same cleanup as {@code stopServer()}, while an abort before
+ * registration (e.g. the duplicate server id rejection) must leave the already
+ * connected handler untouched.
+ */
+@SuppressWarnings("javadoc")
+public class HandshakeAbortRegistrationTest extends ReplicationTestCase
+{
+ private static final int SOCKET_TIMEOUT_MS = 30000;
+ private static final long ADOPTED_GEN_ID = 4801;
+ private static final int REMOTE_DS_ID = 41;
+ private static final int REMOTE_RS_ID = 42;
+
+ @Test
+ public void abortAfterRegisterMustUnregisterDataServer() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try (ServerSocket listen = TestCaseUtils.bindFreePort())
+ {
+ listen.setSoTimeout(SOCKET_TIMEOUT_MS);
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ TestCaseUtils.findFreePort(), "handshakeAbortRegistrationDSDb", 0,
+ 8211, 0, 100, new TreeSet<String>()));
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ final ReplSessionSecurity security = getReplSessionSecurity();
+
+ final Session[] sessionPair = connectSessionPair(listen, security);
+ try (Session remoteEnd = sessionPair[0];
+ Session session = sessionPair[1])
+ {
+ final DataServerHandler dsHandler =
+ new DataServerHandler(session, 100, replicationServer, 100);
+ initializeFromHandshake(dsHandler, baseDN, REMOTE_DS_ID, true);
+
+ try
+ {
+ // The handshake registers the handler while holding the domain lock.
+ domain.lock();
+ domain.register(dsHandler);
+ assertSame(domain.getConnectedDSs().get(REMOTE_DS_ID), dsHandler);
+
+ // A generation id adopted while the doomed DS was the only connected
+ // one must be reset by the abort, as a regular disconnection would.
+ domain.changeGenerationId(ADOPTED_GEN_ID);
+
+ dsHandler.abortStart(null);
+
+ assertFalse(domain.getConnectedDSs().containsKey(REMOTE_DS_ID),
+ "the aborted handshake left a dead DataServerHandler registered");
+ assertFalse(domain.hasLock(),
+ "abortStart must release the domain lock");
+ assertEquals(domain.getGenerationId(), -1,
+ "unregistering the last DS must reset the unsaved generation id");
+
+ final DataServerHandler reconnecting =
+ new DataServerHandler(session, 100, replicationServer, 100);
+ reconnecting.serverId = REMOTE_DS_ID;
+ assertFalse(domain.isAlreadyConnectedToDS(reconnecting),
+ "the dead handler still refuses reconnection of its server id");
+ }
+ finally
+ {
+ if (domain.hasLock())
+ {
+ domain.release();
+ }
+ }
+ }
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ @Test
+ public void abortAfterRegisterMustUnregisterReplicationServer() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try (ServerSocket listen = TestCaseUtils.bindFreePort())
+ {
+ listen.setSoTimeout(SOCKET_TIMEOUT_MS);
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ TestCaseUtils.findFreePort(), "handshakeAbortRegistrationRSDb", 0,
+ 8212, 0, 100, new TreeSet<String>()));
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ final ReplSessionSecurity security = getReplSessionSecurity();
+
+ final Session[] sessionPair = connectSessionPair(listen, security);
+ try (Session remoteEnd = sessionPair[0];
+ Session session = sessionPair[1])
+ {
+ final ReplicationServerHandler rsHandler =
+ new ReplicationServerHandler(session, 100, replicationServer, 100);
+ initializeFromHandshake(rsHandler, baseDN, REMOTE_RS_ID, false);
+
+ try
+ {
+ domain.lock();
+ domain.register(rsHandler);
+ assertSame(domain.getConnectedRSs().get(REMOTE_RS_ID), rsHandler);
+
+ rsHandler.abortStart(null);
+
+ assertFalse(domain.getConnectedRSs().containsKey(REMOTE_RS_ID),
+ "the aborted handshake left a dead ReplicationServerHandler registered");
+ assertFalse(domain.hasLock(),
+ "abortStart must release the domain lock");
+ }
+ finally
+ {
+ if (domain.hasLock())
+ {
+ domain.release();
+ }
+ }
+ }
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /**
+ * The handshake also aborts before registering, e.g. when rejecting a second
+ * connection with an already connected server id. Such an abort must never
+ * evict the legitimately connected handler owning that server id.
+ */
+ @Test
+ public void abortBeforeRegisterMustNotEvictConnectedDataServer() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ try (ServerSocket listen = TestCaseUtils.bindFreePort())
+ {
+ listen.setSoTimeout(SOCKET_TIMEOUT_MS);
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ TestCaseUtils.findFreePort(), "handshakeAbortRegistrationDupDb", 0,
+ 8213, 0, 100, new TreeSet<String>()));
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ final ReplSessionSecurity security = getReplSessionSecurity();
+
+ final Session[] connectedPair = connectSessionPair(listen, security);
+ final Session[] duplicatePair = connectSessionPair(listen, security);
+ try (Session connectedRemoteEnd = connectedPair[0];
+ Session connectedSession = connectedPair[1];
+ Session duplicateRemoteEnd = duplicatePair[0];
+ Session duplicateSession = duplicatePair[1])
+ {
+ final DataServerHandler connected =
+ new DataServerHandler(connectedSession, 100, replicationServer, 100);
+ initializeFromHandshake(connected, baseDN, REMOTE_DS_ID, true);
+ try
+ {
+ domain.lock();
+ domain.register(connected);
+ }
+ finally
+ {
+ domain.release();
+ }
+
+ // Second handshake with the same server id: rejected and aborted
+ // before it ever registered.
+ final DataServerHandler duplicate =
+ new DataServerHandler(duplicateSession, 100, replicationServer, 100);
+ initializeFromHandshake(duplicate, baseDN, REMOTE_DS_ID, true);
+ assertTrue(domain.isAlreadyConnectedToDS(duplicate));
+
+ duplicate.abortStart(null);
+
+ assertSame(domain.getConnectedDSs().get(REMOTE_DS_ID), connected,
+ "aborting an unregistered handshake evicted the connected DS");
+ }
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
+ /** Puts the handler in the state it has when the handshake registers it. */
+ private void initializeFromHandshake(ServerHandler handler, DN baseDN,
+ int remoteServerId, boolean isDataServer) throws Exception
+ {
+ handler.serverId = remoteServerId;
+ handler.serverURL = "127.0.0.1:1636";
+ handler.setBaseDNAndDomain(baseDN, isDataServer);
+ }
+
+ /** Teardown must never mask the primary assertion failure. */
+ private void removeQuietly(ReplicationServer replicationServer)
+ {
+ try
+ {
+ remove(replicationServer);
+ }
+ catch (Exception ignored)
+ {
+ }
+ }
+
+ /**
+ * Establishes a connected session pair over the given listen socket, as a
+ * remote server connecting to the RS would. The TLS negotiation performed by
+ * the session factories needs both ends handshaking at the same time, so the
+ * client end runs on its own thread.
+ *
+ * @return the two sessions: the remote (client) end first, then the local
+ * (server) end to hand to the handler under test
+ */
+ private Session[] connectSessionPair(ServerSocket listenSocket,
+ final ReplSessionSecurity security) throws Exception
+ {
+ final Socket clientSocket =
+ new Socket("127.0.0.1", listenSocket.getLocalPort());
+ clientSocket.setTcpNoDelay(true);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try
+ {
+ final Future<Session> clientEnd = executor.submit(new Callable<Session>()
+ {
+ @Override
+ public Session call() throws Exception
+ {
+ return security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS);
+ }
+ });
+
+ final Socket serverSocket = listenSocket.accept();
+ serverSocket.setTcpNoDelay(true);
+ final Session serverEnd =
+ security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS);
+ assertNotNull(serverEnd,
+ "could not create a session for the handler under test");
+
+ return new Session[] {
+ clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
+ }
+ finally
+ {
+ executor.shutdown();
+ }
+ }
+}
--
Gitblit v1.10.0