[#821] Unregister the handler when a handshake aborts after registration

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

@@ -1164,6 +1164,59 @@ private void unregisterServerHandler(ServerHandler sHandler)
     }
   }
 
+  /**
+   * 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

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

@@ -246,6 +246,16 @@ protected void abortStart(LocalizableMessage reason)
       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();
   }
 

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();
+    }
+  }
+}