/*
 * 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.assertj.core.api.Assertions.assertThat;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.testng.Assert.*;

import java.io.IOException;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collections;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLSocket;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.DSInfo;
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.ServerStartMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.TopologyMsg;
import org.testng.annotations.Test;

/**
 * Tests for issue #1241: the severity an aborted replication handshake is logged with.
 * <p>
 * {@code abortStart} used to log the reason of every abort at ERROR, so a peer which
 * restarts under load reads as an error of every server it was handshaking with: the
 * listen thread which gives way to a cross connection
 * ({@code WARN_TIMEOUT_WHEN_CROSS_CONNECTION}, a warning by its own name), and the
 * handshake its peer left ({@code *_DISCONNECTED_DURING_HANDSHAKE}, where a peer which gave
 * up on its own timeout logs that timeout at WARNING itself). Both are warnings now, in
 * either phase of the handshake, and so is a peer which stays silent past the connection
 * timeout: this server gives up on it as it does on a peer which went away.
 * <p>
 * What this server rejects, a peer which breaks the protocol, a frame which does not decode
 * and a frame too large to allocate are still errors. Each call site picks the severity of its
 * own abort, so a case pins only the one it reaches: the cases which say so reach every
 * {@code abortStart} of a reason in the handshakes of a replication server and of a data
 * server but two, so that logging an abort as a warning is not a way to pass them. The two are
 * the {@code catch (Exception)} of {@code startFromRemoteRS()}, which a peer reaches only
 * through a failure of the JDK (a topology without the replication server which sent it fails
 * on {@code get(0)}), and the {@code catch (DirectoryException)} of
 * {@code startFromRemoteDS()}, which no peer reaches with a message.
 * <p>
 * Each test drives one handler on a session pair of its own, as the listen thread or the
 * connect thread of the server would, and reads the severity off the record of the reason.
 * The peer which goes away aborts its end of the session rather than closing it:
 * {@code Session.close()} publishes a {@code StopMsg}, which is a peer ending the handshake
 * in order, and is aborted on silently.
 */
@SuppressWarnings("javadoc")
public class HandshakeAbortSeverityTest extends ReplicationTestCase
{
  private static final int SOCKET_TIMEOUT_MS = 30000;
  /** The connection timeout of a handshake whose peer stays silent: nothing comes before it. */
  private static final int SILENT_PEER_TIMEOUT_MS = 500;
  /** Longer than the longest wait of {@code lockDomainWithTimeout()}, which is 8 seconds. */
  private static final long LOCK_TIMEOUT_MS = 30000;
  private static final int REMOTE_DS_ID = 61;
  private static final int REMOTE_RS_ID = 62;
  private static final String REMOTE_RS_URL = "127.0.0.1:8989";
  /** What {@code Session.receive()} reports of the frame {@link #sendTooLargeFrameHeader} announces. */
  private static final LocalizableMessage PACKET_TOO_LARGE = LocalizableMessage.raw(
      "Packet too large, can't allocate " + Integer.MAX_VALUE + " bytes.");

  /**
   * The listen thread waits a random 3 to 8 seconds for the domain lock, so that of two
   * replication servers connecting to each other at the same time one gives way: the
   * connect thread holding the lock here is the one which goes on.
   */
  @Test
  public void aCrossConnectionGivenWayToIsAWarning() throws Exception
  {
    final int rsId = 8261;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityCrossDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplicationServerDomain domain =
          replicationServer.getReplicationServerDomain(baseDN, true);
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      // The lock is held by a thread of its own, as the connect thread holds it: the
      // domain lock is reentrant, so the thread of the handshake would take it.
      final ExecutorService lockHolder = Executors.newSingleThreadExecutor();
      final List<String> records;
      try
      {
        final CountDownLatch locked = new CountDownLatch(1);
        final CountDownLatch handshakeDone = new CountDownLatch(1);
        final Future<Void> holding = lockHolder.submit(() -> {
          domain.lock();
          try
          {
            locked.countDown();
            handshakeDone.await();
          }
          finally
          {
            domain.release();
          }
          return null;
        });
        assertTrue(locked.await(LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS),
            "the domain lock should have been taken");

        try
        {
          records = errorLogRecordsOf(() -> {
            handler.startFromRemoteRS(startMsg);
            return null;
          });
        }
        finally
        {
          handshakeDone.countDown();
        }
        holding.get(LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      }
      finally
      {
        lockHolder.shutdownNow();
      }

      assertThat(recordOf(records, WARN_TIMEOUT_WHEN_CROSS_CONNECTION.get(
              baseDN, REMOTE_RS_ID, session.getReadableRemoteAddress(), rsId)))
          .as("giving way to a cross connection is a warning, as its message says")
          .contains("severity=WARNING");
      assertThat(countRecordsOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId).toString()))
          .as("the handshake ends where this server gives way to the cross connection")
          .isZero();
    });
  }

  @Test
  public void aDataServerGoneDuringTheHandshakeIsAWarning() throws Exception
  {
    final int rsId = 8262;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityDSDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      remoteEnd.abort(new IOException("the directory server gave up on the handshake"));
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_DS_ID, rsId)))
          .as("a directory server which went away during the handshake is a warning")
          .contains("severity=WARNING");
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  /**
   * A peer which stays connected and says nothing past the connection timeout is given up on
   * as one which went away.
   */
  @Test
  public void aDataServerSilentDuringTheHandshakeIsAWarning() throws Exception
  {
    final int rsId = 8274;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeveritySilentDSDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      session.setSoTimeout(SILENT_PEER_TIMEOUT_MS);
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_DS_ID, rsId)))
          .as("a directory server silent past the connection timeout is a warning")
          .contains("severity=WARNING");
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  /**
   * A directory server which sends something else where its start session message is due
   * breaks the protocol, and that is still an error.
   */
  @Test
  public void aDataServerWhichSendsNoStartSessionIsAnError() throws Exception
  {
    final int rsId = 8275;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityNoStartSessionDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      // Published ahead of the start message it answers: the session reads it all the same.
      remoteEnd.publish(serverStartMsg(baseDN));
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertThat(recordOf(records, LocalizableMessage.raw("Protocol error: StartSessionMsg required.")))
          .as("a directory server which breaks the protocol is an error")
          .contains("severity=ERROR");
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  @Test
  public void aReplicationServerGoneDuringTheHandshakeItStartedIsAWarning() throws Exception
  {
    final int rsId = 8263;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      remoteEnd.abort(new IOException("the replication server gave up on the handshake"));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId)))
          .as("a replication server which went away during the handshake it started is a warning")
          .contains("severity=WARNING");
    });
  }

  @Test
  public void aReplicationServerGoneDuringTheHandshakeThisServerStartedIsAWarning() throws Exception
  {
    final int rsId = 8264;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSDb", rsId, (replicationServer, remoteEnd, session) -> {
      remoteEnd.abort(new IOException("the replication server went away"));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer went away, so this is no session");
        return null;
      });

      assertThat(recordOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())))
          .as("a replication server which went away during the handshake this server started"
              + " is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * What the connect thread of a cross connection reads, when the listen thread of its peer
   * waits for the domain lock longer than the connection timeout: up to 8 seconds against 5 by
   * default.
   */
  @Test
  public void aReplicationServerSilentDuringTheHandshakeThisServerStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8276;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeveritySilentToRSDb", rsId, (replicationServer, remoteEnd, session) -> {
      session.setSoTimeout(SILENT_PEER_TIMEOUT_MS);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer did not answer, so this is no session");
        return null;
      });

      assertThat(recordOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())))
          .as("a replication server silent past the connection timeout is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * The second phase of the handshake reads the topology of the peer through a method of its
   * own, which used to report every failure of the session as the error of a raw message.
   */
  @Test
  public void aReplicationServerGoneBeforeItsTopologyInTheHandshakeItStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8266;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        // Gone once it has read the start message of this server: that is sent, and the
        // handshake waits for the topology of the peer.
        goneAfterReading(1, remoteEnd, () -> {
          handler.startFromRemoteRS(startMsg);
          return null;
        });
        return null;
      });

      assertThat(recordOf(records, WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId)))
          .as("a replication server which went away before it sent its topology is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * The case above for a peer which stays silent: the second phase used to log the timeout
   * as the error of the raw "Read timed out".
   */
  @Test
  public void aReplicationServerSilentBeforeItsTopologyInTheHandshakeItStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8277;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeveritySilentFromRSTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      session.setSoTimeout(SILENT_PEER_TIMEOUT_MS);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertThat(recordOf(records, WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId)))
          .as("a replication server silent past the connection timeout before it sent its topology"
              + " is a warning")
          .contains("severity=WARNING");
    });
  }

  @Test
  public void aReplicationServerGoneBeforeItsTopologyInTheHandshakeThisServerStartedIsAWarning()
      throws Exception
  {
    final int rsId = 8267;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      // Published ahead of the start message it answers: the session reads it all the same.
      remoteEnd.publish(replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        // Gone once it has read the start message and the topology of this server: both
        // are sent, and the handshake waits for the topology of the peer.
        assertFalse(goneAfterReading(2, remoteEnd, () -> handler.connect(baseDN, true)),
            "the peer went away, so this is no session");
        return null;
      });

      assertThat(recordOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())))
          .as("a replication server which went away before it sent its topology is a warning")
          .contains("severity=WARNING");
    });
  }

  /**
   * A peer which answers the start message of this server with something else breaks the
   * protocol, and that is still an error: of the aborts, only the peer going away and the
   * cross connection are warnings.
   */
  @Test
  public void aPeerWhichBreaksTheProtocolIsStillAnError() throws Exception
  {
    final int rsId = 8265;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityProtocolDb", rsId, (replicationServer, remoteEnd, session) -> {
      // Published ahead of the start message it answers: the session reads it all the same.
      remoteEnd.publish(new TopologyMsg(
          Collections.<DSInfo> emptyList(), Collections.<RSInfo> emptyList()));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer broke the protocol, so this is no session");
        return null;
      });

      assertThat(recordOf(records, ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
              TopologyMsg.class.getCanonicalName(), "ReplServerStartMsg")))
          .as("a peer which breaks the protocol is an error")
          .contains("severity=ERROR");
    });
  }

  /**
   * The case above in the second phase, which is aborted from the {@code DirectoryException}
   * arm of the handshake: a peer which sends its start message again where its topology is
   * due.
   */
  @Test
  public void aPeerWhichSendsNoTopologyInTheHandshakeItStartedIsAnError() throws Exception
  {
    final int rsId = 8268;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSNoTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      remoteEnd.publish(replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertThat(recordOf(records, ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
              ReplServerStartMsg.class.getCanonicalName(), "TopologyMsg")))
          .as("a peer which breaks the protocol is an error")
          .contains("severity=ERROR");
    });
  }

  @Test
  public void aPeerWhichSendsNoTopologyInTheHandshakeThisServerStartedIsAnError() throws Exception
  {
    final int rsId = 8269;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSNoTopoDb", rsId, (replicationServer, remoteEnd, session) -> {
      // Published ahead of the messages they answer: the session reads them all the same.
      remoteEnd.publish(replServerStartMsg(baseDN));
      remoteEnd.publish(replServerStartMsg(baseDN));
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the peer broke the protocol, so this is no session");
        return null;
      });

      assertThat(recordOf(records, ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
              ReplServerStartMsg.class.getCanonicalName(), "TopologyMsg")))
          .as("a peer which breaks the protocol is an error")
          .contains("severity=ERROR");
    });
  }

  /**
   * A rejection by this server: a peer which names the server id of a replication server the
   * domain already holds a session with, at another address.
   */
  @Test
  public void aServerIdAnotherReplicationServerHoldsIsAnError() throws Exception
  {
    final int rsId = 8270;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityDuplicateDb", rsId, (replicationServer, firstEnd, first) -> {
      // The first peer completes its handshake, so that the domain holds it: its topology is
      // published ahead of the start message of this server it answers.
      final ReplServerStartMsg firstStartMsg =
          (ReplServerStartMsg) sendThrough(firstEnd, first, replServerStartMsg(baseDN));
      firstEnd.publish(new TopologyMsg(Collections.<DSInfo> emptyList(),
          Collections.singletonList(new RSInfo(REMOTE_RS_ID, REMOTE_RS_URL, -1, (byte) 1, 1))));
      final ReplicationServerHandler registered =
          new ReplicationServerHandler(first, 100, replicationServer, 100);
      registered.startFromRemoteRS(firstStartMsg);
      // The precondition of the case rather than an assumption.
      assertThat(replicationServer.getReplicationServerDomain(baseDN).getConnectedRSs().get(REMOTE_RS_ID))
          .as("the first peer should hold its session")
          .isSameAs(registered);

      try (ServerSocket listen = TestCaseUtils.bindFreePort())
      {
        listen.setSoTimeout(SOCKET_TIMEOUT_MS);
        final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity());
        try (Session remoteEnd = sessionPair[0];
            Session session = sessionPair[1])
        {
          final ReplServerStartMsg startMsg = (ReplServerStartMsg) sendThrough(remoteEnd, session,
              replServerStartMsg(baseDN, "127.0.0.1:8990"));
          final ReplicationServerHandler handler =
              new ReplicationServerHandler(session, 100, replicationServer, 100);

          final List<String> records = errorLogRecordsOf(() -> {
            handler.startFromRemoteRS(startMsg);
            return null;
          });

          assertThat(recordOf(records, ERR_DUPLICATE_REPLICATION_SERVER_ID.get(
                  replicationServer.getMonitorInstanceName(), registered.getServerAddressURL(),
                  handler.getServerAddressURL(), REMOTE_RS_ID)))
              .as("a peer this server rejects is an error")
              .contains("severity=ERROR");
        }
      }
    });
  }

  /**
   * A frame too large to allocate is not the peer going away, though the session reports it
   * as an {@code IOException}: either the peer breaks the protocol, a handshake message being
   * nowhere near that size, or this server has no heap to spare it, and both are errors.
   */
  @Test
  public void aFrameTooLargeForTheStartMessageOfTheHandshakeThisServerStartedIsAnError()
      throws Exception
  {
    final int rsId = 8271;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSFrameDb", rsId, (replicationServer, remoteEnd, session) -> {
      sendTooLargeFrameHeader(remoteEnd);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the frame cannot be read, so this is no session");
        return null;
      });

      assertTooLargeFrameIsAnError(records, session,
          WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress()));
    });
  }

  /**
   * A frame which does not decode fails in {@code Session.receive()} with neither an
   * {@code IOException} nor a {@code DirectoryException}, and is aborted on from the
   * {@code catch (Exception)} of the handshake, still as an error.
   */
  @Test
  public void aFrameWhichDoesNotDecodeInTheHandshakeThisServerStartedIsAnError()
      throws Exception
  {
    final int rsId = 8278;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityToRSUndecodableDb", rsId, (replicationServer, remoteEnd, session) -> {
      // A frame of one byte, a message type which no message has.
      sendRaw(remoteEnd, new byte[] { '0', '0', '0', '0', '0', '0', '0', '1', 0 });
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        assertFalse(handler.connect(baseDN, true), "the frame cannot be decoded, so this is no session");
        return null;
      });

      assertThat(recordOf(records, LocalizableMessage.raw("received message with unknown type")))
          .as("a frame which does not decode is an error")
          .contains("severity=ERROR");
      assertThat(countRecordsOf(records,
              WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(rsId, session.getReadableRemoteAddress())
                  .toString()))
          .as("a frame which does not decode is not the peer going away")
          .isZero();
    });
  }

  @Test
  public void aFrameTooLargeForTheTopologyInTheHandshakeItStartedIsAnError() throws Exception
  {
    final int rsId = 8272;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityFromRSFrameDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ReplServerStartMsg startMsg =
          (ReplServerStartMsg) sendThrough(remoteEnd, session, replServerStartMsg(baseDN));
      sendTooLargeFrameHeader(remoteEnd);
      final ReplicationServerHandler handler =
          new ReplicationServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteRS(startMsg);
        return null;
      });

      assertTooLargeFrameIsAnError(records, session,
          WARN_RS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_RS_ID, rsId));
    });
  }

  @Test
  public void aFrameTooLargeForTheStartSessionOfADataServerIsAnError() throws Exception
  {
    final int rsId = 8273;
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    runCase("handshakeAbortSeverityDSFrameDb", rsId, (replicationServer, remoteEnd, session) -> {
      final ServerStartMsg startMsg =
          (ServerStartMsg) sendThrough(remoteEnd, session, serverStartMsg(baseDN));
      sendTooLargeFrameHeader(remoteEnd);
      final DataServerHandler handler =
          new DataServerHandler(session, 100, replicationServer, 100);

      final List<String> records = errorLogRecordsOf(() -> {
        handler.startFromRemoteDS(startMsg);
        return null;
      });

      assertTooLargeFrameIsAnError(records, session,
          WARN_DS_DISCONNECTED_DURING_HANDSHAKE.get(REMOTE_DS_ID, rsId));
      assertAbortedHandshakeWentNoFurther(session);
    });
  }

  /** One case: the replication server under test, and a session pair to it. */
  private interface HandshakeCase
  {
    void run(ReplicationServer replicationServer, Session remoteEnd, Session session)
        throws Exception;
  }

  /**
   * Runs the provided case on a replication server of its own and a session pair to it,
   * which are torn down once it ran.
   */
  private void runCase(String dbName, int rsId, HandshakeCase handshakeCase) throws Exception
  {
    ReplicationServer replicationServer = null;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      replicationServer = newReplicationServer(dbName, rsId);

      final Session[] sessionPair =
          connectSessionPair(listen, getReplSessionSecurity());
      try (Session remoteEnd = sessionPair[0];
          Session session = sessionPair[1])
      {
        handshakeCase.run(replicationServer, remoteEnd, session);
      }
    }
    finally
    {
      removeQuietly(replicationServer);
    }
  }

  /**
   * Runs the provided handshake on a thread of its own, as the listen thread or the connect
   * thread of the server would, while the remote end reads the provided number of messages
   * and goes away: the handshake then fails on what it waits for next, rather than on a
   * message it sends.
   */
  private static <T> T goneAfterReading(int messages, Session remoteEnd, Callable<T> handshake)
      throws Exception
  {
    final ExecutorService handshakeThread = Executors.newSingleThreadExecutor();
    try
    {
      final Future<T> done = handshakeThread.submit(handshake);
      for (int i = 0; i < messages; i++)
      {
        remoteEnd.receive();
      }
      remoteEnd.abort(new IOException("the replication server went away"));
      return done.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    }
    finally
    {
      handshakeThread.shutdownNow();
    }
  }

  /**
   * Writes the length header of a frame of {@link Integer#MAX_VALUE} bytes from the remote
   * end, and nothing after it: more than any array can hold, so the local end fails to
   * allocate it whatever its heap. The session writes no frame it did not encode itself, so
   * the header goes to the socket under it.
   */
  private static void sendTooLargeFrameHeader(Session remoteEnd) throws Exception
  {
    sendRaw(remoteEnd, String.format("%08x", Integer.MAX_VALUE).getBytes());
  }

  /** Writes the provided bytes from the remote end to the socket under it, as they are. */
  private static void sendRaw(Session remoteEnd, byte[] bytes) throws Exception
  {
    final Field secureSocket = Session.class.getDeclaredField("secureSocket");
    secureSocket.setAccessible(true);
    final OutputStream output = ((SSLSocket) secureSocket.get(remoteEnd)).getOutputStream();
    output.write(bytes);
    output.flush();
  }

  private static void assertTooLargeFrameIsAnError(List<String> records, Session session,
      LocalizableMessage disconnected)
  {
    final String record = recordOf(records, PACKET_TOO_LARGE);
    assertThat(record)
        .as("a frame too large to allocate is an error")
        .contains("severity=ERROR");
    assertThat(record)
        .as("the record of a frame too large to allocate names the peer which sent it")
        .contains("Peer: " + session.getReadableRemoteAddress());
    assertThat(countRecordsOf(records, disconnected.toString()))
        .as("a frame too large to allocate is not the peer going away")
        .isZero();
  }

  /**
   * Asserts that the handshake of a data server ended where it was aborted: only
   * {@code finalizeStart()} starts the thread of the session, once the handler is registered
   * with the domain. Read off the thread rather than off the domain, which the reader of a
   * handler started on a closed session may have unregistered it from already.
   */
  private static void assertAbortedHandshakeWentNoFurther(Session session)
  {
    assertThat(session.getState())
        .as("a handshake which was aborted starts no session thread")
        .isEqualTo(Thread.State.NEW);
  }

  private ReplicationServer newReplicationServer(String dbName, int rsId) throws Exception
  {
    return new ReplicationServer(new ReplServerFakeConfiguration(
        TestCaseUtils.findFreePort(), dbName, 0, rsId, 0, 100, new TreeSet<String>()));
  }

  private ServerStartMsg serverStartMsg(DN baseDN)
  {
    return new ServerStartMsg(REMOTE_DS_ID, "127.0.0.1:1389", baseDN, 100, 0,
        new ServerState(), -1, true, (byte) 1);
  }

  private ReplServerStartMsg replServerStartMsg(DN baseDN)
  {
    return replServerStartMsg(baseDN, REMOTE_RS_URL);
  }

  /**
   * Returns the start message of a fake peer replication server.
   * <p>
   * A generation id of -1 leaves the one of the domain alone, and the session stays
   * encrypted, so that neither end has to leave the SSL session for the handshake to go on.
   */
  private ReplServerStartMsg replServerStartMsg(DN baseDN, String serverURL)
  {
    return new ReplServerStartMsg(REMOTE_RS_ID, serverURL, baseDN, 100,
        new ServerState(), -1, true, (byte) 1, 5000);
  }

  /**
   * Publishes the provided start message from the remote end and returns it as the local
   * end reads it, which is the message the listen thread hands to the handler: it carries
   * the protocol version of the session, which a message built here does not.
   */
  private ReplicationMsg sendThrough(Session remoteEnd, Session session, ReplicationMsg startMsg)
      throws Exception
  {
    remoteEnd.publish(startMsg);
    return session.receive();
  }

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