* Package private for testing. * * @param nowNanos * The value of {@link System#nanoTime()} at which the handshake failed. * @return A number greater than or equal to zero if this failure is to be logged as a - * warning, which is then the number of failures logged at debug level since - * the previous warning, or {@code -count - 1} if this failure is itself to be - * logged at debug level, where {@code count} is the number of failures logged - * at debug level since the previous warning, this one included. + * warning, which is then the number of failures suppressed since the previous + * warning, or {@code -count - 1} if this failure is itself to be suppressed, + * where {@code count} is the number of failures suppressed since the previous + * warning, this one included. */ long recordHandshakeFailure(final long nowNanos) { - final long lastWarn = lastHandshakeFailureWarnNanos.get(); - if (nowNanos - lastWarn >= HANDSHAKE_FAILURE_WARN_INTERVAL_NANOS - && lastHandshakeFailureWarnNanos.compareAndSet(lastWarn, nowNanos)) - { - return suppressedHandshakeFailures.getAndSet(0); - } - return -suppressedHandshakeFailures.incrementAndGet() - 1; + return handshakeFailures.record(nowNanos); } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index 29a557206c..c97a6472b7 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java @@ -37,12 +37,16 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.i18n.LocalizableMessageBuilder; +import org.forgerock.i18n.LocalizableMessageDescriptor; import org.forgerock.i18n.slf4j.LocalizedLogger; import org.forgerock.opendj.config.server.ConfigChangeResult; import org.forgerock.opendj.config.server.ConfigException; @@ -81,6 +85,7 @@ import org.opends.server.types.HostPort; import org.opends.server.types.SearchFilter; import org.opends.server.types.VirtualAttributeRule; +import org.opends.server.util.FailureLogThrottle; /** * ReplicationServer Listener. This singleton is the main object of the @@ -135,6 +140,31 @@ public class ReplicationServer private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); + /** + * Minimum interval, in minutes, between two warnings about the same failure of the + * listen thread. Every connection reaching the replication port goes through that + * thread, so the rate has to be bounded. Each message reports the interval it was + * logged under, so this one is free to differ from the one a failed handshake is + * reported with, which happens to be the same five minutes. + */ + private static final long FAILURE_WARN_INTERVAL_MINUTES = 5; + + /** + * Time, in milliseconds, the listen thread waits before accepting again after + * {@link ServerSocket#accept()} failed. + *
+ * Package private for testing: a test which makes {@code accept()} fail asserts that + * the thread waited instead of spinning, so the wait has to be a value it can name. + */ + static final long ACCEPT_FAILURE_BACKOFF_MS = 100; + + /** {@link #ACCEPT_FAILURE_BACKOFF_MS}, which is also how close two failures have to be to count as repeating. */ + private static final long ACCEPT_FAILURE_BACKOFF_NANOS = + TimeUnit.MILLISECONDS.toNanos(ACCEPT_FAILURE_BACKOFF_MS); + + /** Reports a peer this replication server cannot connect to once per outage. */ + private final ConnectFailureReporter connectFailures = new ConnectFailureReporter(); + /** To know whether a domain is enabled for the external changelog. */ private final ECLEnabledDomainPredicate domainPredicate; @@ -288,18 +318,75 @@ void runListen(ServerSocket socket) socket.getInetAddress().getHostAddress(), socket.getLocalPort()); + /* + * When the last failure of accept() on this socket was done being handled, confined to + * this thread: a listen port change runs a second listen thread, and each of them times + * the socket it accepts on. + * + * A failure which follows one closely is what the wait below is for. It starts a whole + * interval in the past, so an isolated failure does not wait, and a successful accept + * puts it back there: a loop which is serving connections is doing work rather than + * spinning, and what paced it before is not what the next failure repeats. Without that + * reset, a stream of connections aborted between the handshake and accept() -- a health + * check or a port scan -- would charge the whole listen port a wait per probe, and the + * peers queued behind them would pay it. + * + * What the reset gives up is the failure which alternates with a connection: a process + * out of file descriptors frees one now and then, the accept it lets through resets the + * clock, and the failure after it is timed as isolated and not waited on. The loop then + * turns as fast as the connections arrive. That is the trade -- a probe must not cost + * the peers behind it a wait, and a loop which is accepting is not the silent spin the + * wait was added for -- and the warning is throttled either way, so the error log holds + * one line per five minutes of it whichever side of the trade the failure falls on. + * + * Read when the previous failure was handled rather than when it happened, so that the + * wait it was granted is not what makes the next failure look isolated: timing from the + * failure would leave every second one of a continuous run unwaited, and bound the spin + * to twice the rate the wait is chosen for. + */ + long handledAcceptFailureNanos = System.nanoTime() - ACCEPT_FAILURE_BACKOFF_NANOS; + + /* + * Bound how often a failure of accept() on this socket, and a connection accepted on it + * which cannot be turned into a session, are warned about. Both are confined to this + * thread for the same reason the clock above is: a listen port change runs a second + * listen thread -- switchListenPort() starts it before it stops this one -- and a five + * minute window opened on the port which was left would otherwise suppress the first + * failure on the port which replaced it, silence right after the administrator changed + * the port to get out of trouble. + */ + final FailureLogThrottle acceptFailures = + new FailureLogThrottle(FAILURE_WARN_INTERVAL_MINUTES, TimeUnit.MINUTES); + final FailureLogThrottle sessionSetupFailures = + new FailureLogThrottle(FAILURE_WARN_INTERVAL_MINUTES, TimeUnit.MINUTES); + while (!shutdown.get() && !socket.isClosed()) { // Wait on the replicationServer port. // Read incoming messages and create LDAP or ReplicationServer listener // and Publisher. + Session session = null; try { - Session session; - Socket newSocket = null; + final Socket newSocket; try { newSocket = socket.accept(); + } + catch (Exception e) + { + final boolean repeated = + System.nanoTime() - handledAcceptFailureNanos < ACCEPT_FAILURE_BACKOFF_NANOS; + handleAcceptFailure(acceptFailures, socket, e, repeated); + handledAcceptFailureNanos = System.nanoTime(); + continue; + } + // A connection served is not a spin: the failures before it stop pacing the loop, + // and a failure alternating with a connection is therefore never waited on. + handledAcceptFailureNanos = System.nanoTime() - ACCEPT_FAILURE_BACKOFF_NANOS; + + try + { newSocket.setTcpNoDelay(true); newSocket.setKeepAlive(true); int timeoutMS = MultimasterReplication.getConnectionTimeoutMS(); @@ -311,12 +398,13 @@ void runListen(ServerSocket socket) } catch (Exception e) { - // If problems happen during the SSL handshake, it is necessary - // to close the socket to free the associated resources. - if (newSocket != null) - { - newSocket.close(); - } + logSessionSetupFailure(sessionSetupFailures, newSocket, e); + // createServerSession() closes the socket itself when it does not return a + // session, so this closes the one whose options could not be set, and closes a + // second time, harmlessly, the one it already released. Through the helper: a + // close which throws here would escape this catch into the one of the loop, + // which reports it as a failure to listen, without a throttle. + close(newSocket); continue; } @@ -349,6 +437,16 @@ else if (msg instanceof ReplServerStartMsg) } catch (Exception e) { + /* + * A session which reaches here is owned by nothing else: both handlers started + * above abort a handshake they cannot complete themselves, closing the session + * and returning, so what lands here is the session of a peer which failed after + * its handshake -- receive() on a peer which was killed, or which stopped + * answering past the connection timeout. Leaving it open leaked the socket and + * its file descriptor for the life of the process, one per such peer, towards the + * very exhaustion the accept loop above now has to survive. + */ + close(session); // The socket has probably been closed as part of the // shutdown or changing the port number process. // Just log debug information and loop. @@ -362,6 +460,128 @@ else if (msg instanceof ReplServerStartMsg) } } + /** + * Reports a failure of {@link ServerSocket#accept()} and, when it is not the first one + * in a row, waits before accepting again. + *
+ * The listen socket stays open across such a failure, so the loop would come straight + * back to {@code accept()} and, when the cause is the process running out of file + * descriptors, spin on it without ever logging anything. The wait is what bounds that + * spin; the throttle is what bounds the log. + *
+ * Only a failure which repeats is waited on: a connection reset between the handshake + * and {@code accept()} fails it once, and making the listen thread pause for a + * connection nobody is waiting for any more would slow down the ones which follow it. + * + * @param throttle + * The throttle of the calling listen thread, which bounds how often this + * failure is warned about. + * @param socket + * The socket connections are accepted on. + * @param e + * The failure. + * @param repeated + * Whether the previous failure of {@code accept()} on that socket was handled + * less than {@link #ACCEPT_FAILURE_BACKOFF_MS} ago. + */ + private void handleAcceptFailure(final FailureLogThrottle throttle, final ServerSocket socket, + final Exception e, final boolean repeated) + { + // Read before the socket is tested rather than after it is reported: a socket closed + // in between is one this method returns on, while a closed socket names no address at + // all, and the warning would report the port the administrator needs as "null". + final Object listenAddress = socket.getLocalSocketAddress(); + if (shutdown.get() || socket.isClosed()) + { + // The socket was closed to stop this thread or to change the listen port: the loop + // is about to end, and this failure is how it is told to. + logger.traceException(e); + return; + } + + logThrottledFailure(throttle, WARN_REPLICATION_SERVER_ACCEPT_ERROR, listenAddress, e); + if (repeated) + { + // The wait is not interruptible on purpose: the flag is left alone, as restoring it + // would make every wait which follows return at once and bring the spin back. + // Nothing is lost by that -- the only interrupt the listen thread gets, from + // abortInitialization(), comes after the shutdown flag is set and the socket + // closed, which is what ends this loop. + sleep(ACCEPT_FAILURE_BACKOFF_MS); + } + } + + /** + * Reports a connection which was accepted but on which no replication session could be + * started, and which is therefore about to be closed. + *
+ * A failed SSL handshake is reported by {@link ReplSessionSecurity#createServerSession} + * itself, so what reaches here is everything else: a trust store which cannot be read, + * which makes every inbound connection fail this way, and a peer which stops + * responding during the handshake. Every connection reaching the replication port goes + * through this path, so the log has to be throttled the way the handshake failure next + * door is. + * + * @param throttle + * The throttle of the listen thread which accepted the connection, which is + * confined to it: see where it is built. + * @param socket + * The accepted socket, which {@code createServerSession} has usually closed + * already, in its own {@code finally}: a closed socket still names the peer it + * was connected to, which is what this reads from it. + * @param e + * The failure. + */ + private void logSessionSetupFailure(final FailureLogThrottle throttle, final Socket socket, + final Exception e) + { + logThrottledFailure(throttle, WARN_REPLICATION_SERVER_SESSION_SETUP_ERROR, + socket.getRemoteSocketAddress(), e); + } + + /** + * Logs a failure of the listen thread as a warning when the provided throttle lets it + * through, reporting how many failures that line stands for. + *
+ * A failure the throttle suppresses is still recorded, with the information severity:
+ * {@code logger.debug} of a localized message publishes to the error log and not to the
+ * debug log. What reads that severity is the replication log, whose shipped publisher
+ * overrides it on for the {@code SYNC} category; the error log does not publish it by
+ * default. So the throttle bounds what the error log holds, and what the replication log
+ * holds is one record per failure, which is what the messages say.
+ *
+ * @param throttle
+ * The throttle bounding how often this failure is warned about.
+ * @param message
+ * The message reporting it, which takes the server id, the address, the cause,
+ * the interval and the number of failures suppressed since the previous warning.
+ * @param address
+ * The address the failure happened on.
+ * @param e
+ * The failure.
+ */
+ private void logThrottledFailure(final FailureLogThrottle throttle,
+ final LocalizableMessageDescriptor.Arg5
+ * Package private for testing: what a peer costs the error log is decided here, and a
+ * test of the bookkeeping alone cannot see how it is called.
*
* @param remoteServerAddress
* The address and port for the server
* @param baseDN
* The baseDN of the connection
+ * @return {@code true} if the peer is connected to, {@code false} if it could not be
+ * reached at all or if the handshake offered to it did not complete. The
+ * caller leaves a peer it gets {@code false} for alone for a few passes, which
+ * a peer answering with an abort every second is as much in need of as one
+ * which does not answer.
*/
- private boolean connect(HostPort remoteServerAddress, DN baseDN)
+ boolean connect(HostPort remoteServerAddress, DN baseDN)
{
boolean sslEncryption = replSessionSecurity.isSslEncryption();
@@ -466,6 +709,7 @@ private boolean connect(HostPort remoteServerAddress, DN baseDN)
Socket socket = new Socket();
Session session = null;
+ final boolean handshakeCompleted;
try
{
socket.setReuseAddress(true);
@@ -490,16 +734,250 @@ private boolean connect(HostPort remoteServerAddress, DN baseDN)
ReplicationServerHandler rsHandler = new ReplicationServerHandler(
session, config.getQueueSize(), this, config.getWindowSize());
- rsHandler.connect(baseDN, sslEncryption);
+ handshakeCompleted = rsHandler.connect(baseDN, sslEncryption);
}
catch (Exception e)
{
logger.traceException(e);
+ if (connectFailures.recordFailure(remoteServerAddress, baseDN))
+ {
+ // The failure used to be traced and nothing else, so a replication server whose
+ // outgoing handshake failed logged nothing at all: the only log naming the problem
+ // was the one of the peer, which is the machine whose configuration is right.
+ logger.warn(WARN_REPLICATION_SERVER_CONNECT_ERROR, getServerId(), remoteServerAddress,
+ baseDN, getExceptionMessage(e));
+ }
close(session);
close(socket);
return false;
}
- return true;
+ /*
+ * The outage closed here is a failure to connect, and reaching this line is the peer
+ * answering on its replication port: everything WARN_REPLICATION_SERVER_CONNECT_ERROR
+ * is reported for is above it -- the socket and the session built on it, the handshake
+ * throwing nothing of its own. So the outage is closed whatever the handshake did next,
+ * and what the handshake did next decides which recovery is reported rather than
+ * whether one is.
+ *
+ * Closing it on the connection alone is what keeps the peers this server never sees
+ * connected under the address it dialled reportable, and there is one of those for each
+ * narrower reading:
+ *
+ * the registration with the domain misses a peer which negotiates protocol version 1.
+ * ReplicationServerHandler.connect() registers only above V1, the FIXME there being
+ * older than this, so such a peer is connected and never registered.
+ *
+ * the address, and the session left open with it, miss a peer which dials out from an
+ * address other than the one it is configured under -- multi homing, NAT. Its inbound
+ * handler is registered under the source address of its own connection,
+ * ServerHandler.toServerAddressURL() reading the host from the session, so the already
+ * connected branch of runConnect() compares the configured address against one it never
+ * matches, and the handshake this server offers that same peer aborts on a duplicate
+ * server id: abortStart() closes the session, and an open session is never seen here
+ * again.
+ *
+ * An outage left open is not a line too few but a peer gone silent: recordFailure()
+ * returns false from then on, so the next real outage of it is not reported at all.
+ *
+ * What the handshake did next is reported all the same, because it is not the log which
+ * can be left to it. ReplicationServerHandler.connect() aborts at seven places, and the
+ * three which pass no message log nothing at all, abortStart() logging nothing without
+ * one: a StopMsg read where the peer's ReplServerStartMsg was due, a cross connect this
+ * server resolves against a peer it is already connected to, and a phase two the peer
+ * leaves unanswered. Those would otherwise leave "connected" as the last thing said
+ * about a domain which has no session for it. The remaining four have their reason
+ * logged here by abortStart(), and they are not all of one kind: this server rejecting
+ * the peer, the peer going away while the handshake ran, and whatever else fails on
+ * this side of it. So the message reported for an answer without a session tells the
+ * two apart by where a reason was logged rather than by naming a side: the operator is
+ * sent to the log of the peer only where nothing was logged here, and that is also the
+ * abort the peer may have logged nothing about either -- it resolves a cross connect
+ * with the same silent abortStart(null), one line up in startFromRemoteRS().
+ *
+ * The cross connect this server resolves is the one abort of the three where a session
+ * for the domain does exist: it is the connection the peer made, which the already
+ * connected branch of runConnect() reports on its next pass. Reaching this line with
+ * one open needs that registration to land between the snapshot that branch reads and
+ * the dial below it, so what it costs is one warning, and the pass after it says what
+ * is true.
+ */
+ reportConnectionRestored(remoteServerAddress, baseDN, handshakeCompleted);
+ return handshakeCompleted;
+ }
+
+ /**
+ * Reports that a peer this replication server had reported it could not connect to can be
+ * reached again, and does nothing when what is now true of that peer is what was last
+ * reported about it.
+ *
+ * A peer which answers is no longer the peer the outage was reported for, whether the
+ * handshake completed or not: leaving the outage recorded for one which answers and aborts
+ * every handshake would silence its next real outage. But the two are not the same
+ * recovery, and the difference is one the operator has to be able to read -- a peer which
+ * stops the handshake is one this server can reach and has no session with, which is not
+ * what "connected" says -- so each moves the peer to a state of its own rather than
+ * clearing what is known about it. That is what leaves the session established after an
+ * abort still reportable: it is a recovery from the answer without a session, which the
+ * clearing form had already consumed.
+ *
+ * @param remoteServerAddress
+ * The address of the peer which answered.
+ * @param baseDN
+ * The base DN of the domain the attempt was for.
+ * @param connected
+ * Whether the handshake completed, so that this server is connected to the peer,
+ * rather than aborted after it had answered.
+ */
+ private void reportConnectionRestored(final HostPort remoteServerAddress, final DN baseDN,
+ final boolean connected)
+ {
+ if (connected)
+ {
+ if (connectFailures.recordConnected(remoteServerAddress, baseDN))
+ {
+ logger.info(NOTE_REPLICATION_SERVER_CONNECT_RESTORED, getServerId(), remoteServerAddress, baseDN);
+ }
+ }
+ else if (connectFailures.recordReachableWithoutSession(remoteServerAddress, baseDN))
+ {
+ logger.warn(WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION, getServerId(), remoteServerAddress, baseDN);
+ }
+ }
+
+ /**
+ * Remembers which replication servers this replication server has already reported it
+ * cannot connect to, so that a peer which stays unreachable is reported once instead of
+ * on every attempt to reach it.
+ *
+ * The connect thread retries a failed peer every few seconds, for as long as it is down,
+ * and a peer being down is a normal state: one stopped for maintenance would otherwise
+ * fill the error log for the duration. The failure is reported when it starts and, through
+ * {@link #recordConnected} and {@link #recordReachableWithoutSession}, when it ends, so
+ * that neither end of it has to be inferred from a silence.
+ *
+ * What is held per peer and domain is the last state which was reported, not
+ * whether an outage is open. A record which is merely present or absent cannot carry the
+ * middle state -- a peer which answers on its replication port and stops the handshake --
+ * because reporting it would consume the record, and the session which is established
+ * seconds later would then find nothing left to close and go unreported. A peer restarting
+ * takes exactly that path: its port answers before its domains are up.
+ *
+ * Package private for testing.
+ */
+ static final class ConnectFailureReporter
+ {
+ /** What the last message about a peer and a domain said about them. */
+ private enum Reported
+ {
+ /** The peer could not be reached at all: {@code WARN_REPLICATION_SERVER_CONNECT_ERROR}. */
+ DOWN,
+ /**
+ * The peer answered and the handshake did not complete:
+ * {@code WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION}.
+ */
+ REACHABLE_NO_SESSION;
+ }
+
+ /**
+ * What was last reported about each domain of each peer, holding no entry for the peers
+ * and domains nothing was reported about, which a connection to them is therefore not to
+ * be reported for.
+ *
+ * Keyed by {@link HostPort} rather than by its string form, which is the raw host
+ * while equality is on the normalized one: two spellings of the same peer would
+ * otherwise be two keys, and a failure recorded under one would never be cleared by
+ * the connection recorded under the other.
+ *
+ * A HostPort holds the resolution its host had when it was built, and is documented as
+ * not meant to be cached. {@link #retainAll} is what makes holding one here bounded: it
+ * runs on every pass of the connect thread against freshly built addresses, so a key
+ * whose resolution has drifted is dropped within one pass, and the outage under it is
+ * reported a second time rather than never reported again.
+ *
+ * The connect thread is the only one which connects to peers, but a configuration
+ * change replaces it, so this is a map two threads may hand over.
+ */
+ private final ConcurrentMap
+ * Reported only where an outage was: an abort which carries a reason logs that reason on
+ * this server, and one which does not is the peer's to log. What this closes is the
+ * message which said the peer could not be reached, which is no longer what is wrong
+ * with it.
+ *
+ * @param peer
+ * The address of the replication server which answered.
+ * @param baseDN
+ * The base DN of the domain the attempt was for.
+ * @return {@code true} if this is to be reported, {@code false} if nothing was reported
+ * about that peer or if it was already reported as answering without a session.
+ */
+ boolean recordReachableWithoutSession(final HostPort peer, final DN baseDN)
+ {
+ final ConcurrentMap
+ * A handshake which does not complete is aborted rather than thrown out of here, and
+ * three of its seven aborts carry no message at all: a peer which stops the handshake
+ * answers with a {@link StopMsg}, which {@code Session.close()} publishes for every
+ * abort of its own, and {@link #abortStart} logs nothing when the reason is null. The
+ * caller is what is left to tell a connection from an attempt which only reached the
+ * replication port of a peer.
+ *
* @param baseDN The baseDN
* @param sslEncryption The sslEncryption requested to the remote RS.
+ * @return {@code true} when the handshake completed and this handler is started,
+ * {@code false} when it was aborted.
* @throws DirectoryException when an error occurs.
*/
- public void connect(DN baseDN, boolean sslEncryption)
+ public boolean connect(DN baseDN, boolean sslEncryption)
throws DirectoryException
{
// we are the initiator and decides of the encryption
@@ -180,7 +190,7 @@ public void connect(DN baseDN, boolean sslEncryption)
.getClass().getCanonicalName(), "ReplServerStartMsg");
abortStart(message);
}
- return;
+ return false;
}
processStartFromRemote((ReplServerStartMsg) msg);
@@ -189,7 +199,7 @@ public void connect(DN baseDN, boolean sslEncryption)
{
// Simultaneous cross connect.
abortStart(null);
- return;
+ return false;
}
/*
@@ -230,7 +240,7 @@ TopologyMsg then TopologyMsg (with a RS)
{
// Simultaneous cross connect.
abortStart(null);
- return;
+ return false;
}
logTopoHandshakeSNDandRCV(outTopoMsg, inTopoMsg);
@@ -252,6 +262,7 @@ TopologyMsg then TopologyMsg (with a RS)
replicationServerDomain.getBaseDN(), session.getReadableRemoteAddress());
super.finalizeStart();
+ return true;
}
catch (IOException e)
{
@@ -260,16 +271,19 @@ TopologyMsg then TopologyMsg (with a RS)
getReplicationServerId(),
session.getReadableRemoteAddress());
abortStart(errMessage);
+ return false;
}
catch (DirectoryException e)
{
logger.traceException(e);
abortStart(e.getMessageObject());
+ return false;
}
catch (Exception e)
{
logger.traceException(e);
abortStart(LocalizableMessage.raw(e.getLocalizedMessage()));
+ return false;
}
finally
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
index adf6c7ea6c..b99c5bbd9e 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
@@ -1161,8 +1161,18 @@ private ConnectedRS performPhaseOneHandshake(String serverURL, boolean keepSessi
catch (Exception e)
{
logger.traceException(e);
- errorMessage = WARN_EXCEPTION_STARTING_SESSION_PHASE.get(
- getServerId(), serverURL, getBaseDN(), stackTraceToSingleLineString(e));
+ /*
+ * The whole trace, on one line, for the server this broker is electing, which is
+ * what this message carried before it was reported for the others as well; the
+ * message alone for a server which was only contacted, whose report is not
+ * throttled. An SSLException lands here, is not paced by a connect timeout the way
+ * a refused connection is, and collectReplicationServersInfo() reruns over every
+ * URL on each reconnection: one bad certificate among several replication servers
+ * would otherwise write a full stack trace per flap cycle. The trace of those is in
+ * the trace log, which traceException() above wrote it to.
+ */
+ errorMessage = WARN_EXCEPTION_STARTING_SESSION_PHASE.get(getServerId(), serverURL, getBaseDN(),
+ keepSession ? stackTraceToSingleLineString(e) : getExceptionMessage(e));
}
finally
{
@@ -1172,16 +1182,51 @@ private ConnectedRS performPhaseOneHandshake(String serverURL, boolean keepSessi
close(socket);
}
- if (!hasConnected && errorMessage != null && !connectionError)
+ if (!hasConnected && errorMessage != null)
{
- // There was no server waiting on this host:port
- // Log a notice and will try the next replicationServer in the list
- if (keepSession) // Log error message only for final connection
+ if (!connectionError)
{
- // log the error message only once to avoid overflowing the error log
- logger.error(errorMessage);
+ /*
+ * Report the cause for every replication server contacted, and not only for the
+ * elected one: none is ever elected when none of them answers, and this is then
+ * the only place naming why -- a refused connection, a rejected certificate, a
+ * wrong port -- next to the "unable to connect to any replication servers"
+ * summary which names none of them.
+ *
+ * connectionError is what bounds the volume, and it bounds it to one line per
+ * replication server and per attempt to connect this broker makes. It is set
+ * when an attempt reaches no replication server at all, and stays set until a
+ * session is established, so the 500 ms loop which retries a total outage
+ * reports its first pass only.
+ *
+ * It is not set while this broker is connected, so the unreachable servers of a
+ * topology which still serves this broker are reported again on each
+ * reconnection -- one line each, so a reconnection costs as many lines as there
+ * are servers it could not reach, where it used to cost none. That is the volume
+ * this reporting is worth: a broker reconnects when its session is lost, not on
+ * a schedule, and a server which cannot be reached over several reconnections is
+ * a server whose configuration or certificate needs looking at.
+ *
+ * The severity says what the failure cost this broker, not what the message is
+ * named: the elected server keeps the error it was reported with, and a server
+ * which was only contacted is a warning, so that a broker which does find a
+ * server to work with does not raise an error over the one it did not need.
+ * ERR_DS_DN_DOES_NOT_MATCH is the one message which reaches the second branch
+ * under an ERR_ name -- it is set without setting hasConnected -- and it is a
+ * permanent misconfiguration rather than a transient. It still goes out as an
+ * error for the server this broker is electing, which is the one it cannot work
+ * without, and where it is only contacted the broker has another server to work
+ * with. Before this, that path logged nothing above trace either way.
+ */
+ if (keepSession)
+ {
+ logger.error(errorMessage);
+ }
+ else
+ {
+ logger.warn(errorMessage);
+ }
}
-
logger.trace(errorMessage);
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/util/FailureLogThrottle.java b/opendj-server-legacy/src/main/java/org/opends/server/util/FailureLogThrottle.java
new file mode 100644
index 0000000000..7b1c185719
--- /dev/null
+++ b/opendj-server-legacy/src/main/java/org/opends/server/util/FailureLogThrottle.java
@@ -0,0 +1,88 @@
+/*
+ * 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.util;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Bounds how often a repeating failure is reported, so that a failure which a retry loop
+ * or a stream of unwanted connections reproduces many times a second is still visible in
+ * the error log without filling it.
+ *
+ * One failure is logged per interval and the failures in between are counted, so that the
+ * next one logged reports how many it stands for and a single line cannot be mistaken for
+ * a single failure. That count looks backwards only: the failures following the last one
+ * logged of a burst are counted but never reported, as nothing flushes the count when the
+ * failures stop.
+ *
+ * The first failure is always logged, however long the throttle has existed before it.
+ *
+ * Safe to record from several threads, and exact for one: a failure recorded while
+ * another thread is opening a new interval may be counted in the interval which is
+ * closing or in the one which opens, so the count a caller reports can be off by the
+ * failures which raced with it. It is a count of failures, not a ledger.
+ */
+public final class FailureLogThrottle
+{
+ /** Minimum interval, in nanoseconds, between two failures logged. */
+ private final long intervalNanos;
+
+ /**
+ * Value of {@link System#nanoTime()} at which the last failure was logged. It starts one
+ * interval in the past so that the first failure is logged.
+ */
+ private final AtomicLong lastLogNanos;
+
+ /** Number of failures suppressed since the last one logged. */
+ private final AtomicLong suppressed = new AtomicLong();
+
+ /**
+ * Creates a throttle logging at most one failure per provided interval.
+ *
+ * @param interval
+ * The interval between two failures logged.
+ * @param unit
+ * The unit the interval is expressed in.
+ */
+ public FailureLogThrottle(final long interval, final TimeUnit unit)
+ {
+ this.intervalNanos = unit.toNanos(interval);
+ this.lastLogNanos = new AtomicLong(System.nanoTime() - intervalNanos);
+ }
+
+ /**
+ * Records a failure which happened at the provided time and tells how it must be logged,
+ * together with the number of failures suppressed since the previous one logged.
+ *
+ * @param nowNanos
+ * The value of {@link System#nanoTime()} at which the failure happened.
+ * @return A number greater than or equal to zero if this failure is to be logged, which is
+ * then the number of failures suppressed since the previous one logged, or
+ * {@code -count - 1} if this failure is itself to be suppressed, where
+ * {@code count} is the number of failures suppressed since the previous one
+ * logged, this one included.
+ */
+ public long record(final long nowNanos)
+ {
+ final long lastLog = lastLogNanos.get();
+ if (nowNanos - lastLog >= intervalNanos && lastLogNanos.compareAndSet(lastLog, nowNanos))
+ {
+ return suppressed.getAndSet(0);
+ }
+ return -suppressed.incrementAndGet() - 1;
+ }
+}
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index a66b5928b2..475f6d0886 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -447,8 +447,9 @@ WARN_SSL_SERVER_CON_ATTEMPT_ERROR_105=Replication server accepted a connection \
misconfigured: check that the certificate nickname configured in the crypto \
manager exists in the ads-truststore, and that every server of the topology \
trusts the certificates of the other servers. At most one such failure is \
- logged as a warning every %d minutes; the others go to the debug log, which is \
- disabled by default (%d since the previous warning). The error was: %s
+ logged as a warning every %d minutes; the others are recorded with the \
+ information severity, which the replication log publishes and the error log \
+ does not (%d since the previous warning). The error was: %s
WARN_MISSING_REMOTE_MONITOR_DATA_106=Timed out waiting for monitor data \
for the domain "%s" from replication server RS(%d)
NOTE_LOAD_BALANCE_REPLICATION_SERVER_213=Directory Server DS(%d) is disconnecting \
@@ -626,6 +627,36 @@ ERR_REPLAY_SKIPPING_CHANGE_308=Could not replay change %s in domain "%s": its re
NOTE_REPLAY_ABANDONED_CHANGE_309=Could not replay change %s in domain "%s": the replay thread \
it was given to is stopping. The change has not been recorded as replayed and is given back to \
the replication server, which still owns it
+WARN_REPLICATION_SERVER_ACCEPT_ERROR_310=Replication server RS(%d) could not accept a connection on %s: %s. \
+ A failure which repeats at once, the process running out of file descriptors for instance, makes the \
+ listen thread wait before accepting again, so that it does not spin. At most one such failure is logged \
+ as a warning every %d minutes; the others are recorded with the information severity, which the \
+ replication log publishes and the error log does not (%d since the previous warning)
+WARN_REPLICATION_SERVER_SESSION_SETUP_ERROR_311=Replication server RS(%d) accepted a connection from %s \
+ but could not start a replication session on it, and closed it: %s. A failed SSL handshake is reported \
+ on its own, so this is something else. It may be benign, a network probe which connects and closes \
+ without saying anything, but it also occurs when the ads-truststore cannot be read, which fails every \
+ inbound connection this way: where the connection came from a replication peer, check that the \
+ ads-truststore still exists and is readable by the server. At most one such failure is logged as a \
+ warning every %d minutes; the others are recorded with the information severity, which the replication \
+ log publishes and the error log does not (%d since the previous warning)
+WARN_REPLICATION_SERVER_CONNECT_ERROR_312=Replication server RS(%d) could not connect to replication \
+ server %s for domain "%s": %s. This is reported once: the next message about this replication server \
+ comes when it can be reached again, either because this server reached it or because it connected to \
+ this server from the address it is configured under. The domain named is the one this attempt was \
+ for, and a replication server which cannot be reached is not tried for the other domains it serves \
+ while it cannot be, so this names one of them rather than all of them
+NOTE_REPLICATION_SERVER_CONNECT_RESTORED_313=Replication server RS(%d) connected to replication server \
+ %s for domain "%s", which it had reported it had no replication session with
+WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION_314=Replication server RS(%d) reached replication server \
+ %s for domain "%s", which it had reported it could not connect to, but the handshake with it did not \
+ complete: this attempt established no replication session. Where this server ended the handshake, the \
+ reason is logged next to this message, whether it rejected the peer or the peer went away while the \
+ handshake was running. Where nothing is logged next to it, the peer ended the handshake, and it may \
+ have logged no reason of its own either: a shutdown under way, or a connection the two servers made \
+ to each other at the same time, of which one is dropped while the other one serves the domain. Until \
+ a handshake completes, no change is replicated over this connection, and the connection which \
+ completes one is reported in its turn
WARN_REPLAY_NOT_DRAINED_319=Domain "%s" is going down and gave up on waiting up to %d ms for \
the replay of one of its changes to finish. A change which reaches the backend from now on \
is not recorded in the ServerState being saved, so the replication server sends it again \
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationBrokerConnectFailureTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationBrokerConnectFailureTest.java
new file mode 100644
index 0000000000..a3d3377236
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/ReplicationBrokerConnectFailureTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.opends.messages.ReplicationMessages.*;
+
+import java.util.List;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.replication.common.ServerState;
+import org.opends.server.replication.plugin.DomainFakeCfg;
+import org.opends.server.replication.plugin.DummyReplicationDomain;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.testng.annotations.Test;
+
+/**
+ * Tests what a directory server logs when it cannot connect to any replication server.
+ */
+@SuppressWarnings("javadoc")
+public class ReplicationBrokerConnectFailureTest extends ReplicationTestCase
+{
+ /**
+ * Tests that a directory server which reaches no replication server at all names the
+ * reason it reached none.
+ *
+ * The cause is built for every replication server contacted, but it used to be logged
+ * only for the elected one, and no server is ever elected when none of them answers: a
+ * rejected certificate, a refused connection and a wrong port then all read as "unable
+ * to connect to any replication servers".
+ */
+ @Test
+ public void aBrokerWhichReachesNoReplicationServerNamesTheCause() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ final DN baseDN = DN.valueOf(TestCaseUtils.TEST_ROOT_DN_STRING);
+ final int serverId = 4021;
+ // Free, and left free: nothing must listen on it for this test to be about a failure.
+ final int deadPort = TestCaseUtils.findFreePorts(1)[0];
+ final String deadServer = "127.0.0.1:" + deadPort;
+
+ final DomainFakeCfg config = newFakeCfg(baseDN, serverId, deadPort);
+ final ReplicationBroker broker = new ReplicationBroker(
+ new DummyReplicationDomain(0), new ServerState(), config, getReplSessionSecurity());
+ try
+ {
+ final List
+ * A publisher of its own is registered for the duration rather than reading the one the
+ * test harness installs, whose contents span the whole test JVM.
+ *
+ * It publishes every severity, so that a record a throttle kept out of the warnings is
+ * captured too. That is server wide while the action runs, so the records of other
+ * threads are captured as well and a caller has to pick out its own.
+ *
+ * @param action
+ * The action to run.
+ * @return The error log records written while the action ran, in order.
+ * @throws Exception
+ * Whatever the action throws.
+ */
+ protected static List
+ * The list handed to the action is the live one, so an action can wait for a server
+ * thread to log something instead of waiting for a duration. It is synchronized, and
+ * iterating it is not: a reader has to copy it, or hold its monitor.
+ *
+ * @param action
+ * The action to run, taking the records written so far.
+ * @return The error log records written while the action ran, in order.
+ * @throws Exception
+ * Whatever the action throws.
+ */
+ // The publisher is built raw and handed to a parameterized addLogPublisher: the
+ // conversion is unchecked, and it is the one the test harness makes as well.
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ protected static List
+ * A record holds the severity it was published with next to the message, so this is how
+ * a test reads the severity of a message it expects, rather than only its text.
+ *
+ * @param records
+ * The error log records to look in, as {@link #errorLogRecordsOf} returned them.
+ * @param message
+ * The message the record is expected to report.
+ * @return The record reporting the provided message.
+ */
+ protected static String recordOf(List
+ * For what a server thread logs on its own schedule: the connect thread of a replication
+ * server retries a peer every second, so what it reports is waited for rather than
+ * expected to be there already.
+ *
+ * Only the records which arrived since the previous poll are read: the capture publishes
+ * every severity, which is every {@code logger.debug} of every thread of the server for
+ * as long as it is installed, so rereading the whole list on each poll would cost the
+ * square of what a slow wait captures.
+ *
+ * @param records
+ * The live error log records, as {@link ErrorLogAction} received them.
+ * @param contained
+ * The text the awaited record holds.
+ * @param timeoutMs
+ * How long to wait for it, in milliseconds.
+ * @throws InterruptedException
+ * If the wait is interrupted.
+ */
+ protected static void waitForErrorLogRecord(List
+ * What is reported is decided in two places -- {@code connect()} and the already connected
+ * branch of {@code runConnect()} -- and the decision each of them makes is only correct
+ * with respect to the other: a test of {@link ReplicationServer.ConnectFailureReporter} on
+ * its own passes with either of them deleted, or with any of the conditions an outage used
+ * to be closed on put back in front of it.
+ *
+ * The tests which drive a peer of their own drive one the already connected branch of
+ * {@code runConnect()} cannot report anything about, so that {@code connect()} is the only
+ * place a recovery can come from: a peer registered with no domain at all, and one
+ * registered under an address which is not the one it is configured under. Both shapes end
+ * on a session {@code abortStart} closed and on a peer which is connected for no domain,
+ * which is what makes them the outages this server has to close without reading either --
+ * and, being aborts, the recoveries it must not report as connections.
+ *
+ * The one case whose peer does end up registered is
+ * {@link #aSessionEstablishedAfterAnAnswerWithoutOneIsReported}, where the session is the
+ * point: what it pins is that the answer without a session left something for the session
+ * to close, which holds wherever the session is reported from.
+ *
+ * Every test configures the peer it drives, and creates the domain it drives it for, before
+ * anything is reported. The connect thread runs {@code retainAll} against the configured
+ * peers and the domains of the pass on every pass, so a peer or a domain it does not see
+ * has whatever was recorded for it cleared about once a second -- which is a record the
+ * test is not allowed to rely on. Where a record is made from the test thread rather than
+ * by the connect thread, one whole pass is waited for as well, so that no pass which
+ * started before the domain existed is still holding a snapshot without it.
+ */
+@SuppressWarnings("javadoc")
+public class ReplicationServerConnectFailureTest extends ReplicationTestCase
+{
+ private static final int SOCKET_TIMEOUT_MS = 30000;
+ /** How long a peer which is down, or which came back, is waited to be reported within. */
+ private static final long REPORT_TIMEOUT_MS = 30000;
+
+ private static final int RS_ID = 8241;
+ private static final int PEER_RS_ID = 8242;
+
+ /**
+ * Tests that a replication server reports a peer it cannot connect to once, and reports
+ * the connection which ends that outage.
+ *
+ * Both servers run their own connect thread, so this goes through {@code runConnect()}:
+ * the peer is reported by the connect thread of the server under test, and the
+ * connection which ends the outage is reported by whichever of {@code connect()} and the
+ * already connected branch wins -- the peer dials back, and either end may connect first.
+ */
+ @Test
+ public void aPeerWhichIsDownIsReportedOnceAndItsReturnIsReported() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final int[] ports = TestCaseUtils.findFreePorts(2);
+ final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
+ final String restored = NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();
+
+ // Held rather than returned: whatever the capture below throws, these are what the
+ // listen ports, the threads and the changelogs of both servers hang on.
+ final ReplicationServer[] servers = new ReplicationServer[2];
+ try
+ {
+ final List
+ * The outage is a failure to connect, so what closes it is the peer answering:
+ * {@code WARN_REPLICATION_SERVER_CONNECT_ERROR} is reported for the socket and for the
+ * session built on it, the handshake throwing nothing of its own. Holding the outage open
+ * across an abort silences the peer this server never sees connected under the address it
+ * dialled -- the multi homed peer of
+ * {@link #aPeerRegisteredUnderAnotherAddressStillClosesItsOutage}, and one protocol
+ * version down a peer which negotiates V1, connected and never registered.
+ *
+ * Reporting it connected instead is the other half of the same line.
+ * {@code Session.close()} publishes a {@code StopMsg} for every abort its own end makes
+ * and {@code abortStart(null)} logs nothing, so a peer which rejects this server -- its
+ * own duplicate server id, a cross connect it resolves against this server, a shutdown
+ * under way -- would have "connected" as the last thing this server ever says about it.
+ */
+ @Test
+ public void aPeerWhichStopsTheHandshakeStillClosesItsOutage() throws Exception
+ {
+ aPeerWhichAnswersClosesItsOutage("replicationServerHandshakeAbortDb",
+ (session, received) -> {
+ session.publish(new StopMsg());
+ // Read until the peer has read the StopMsg and closed its own end: closing this
+ // one first would race that read, and turn the abort into a failed handshake.
+ session.receive();
+ });
+ }
+
+ /**
+ * Tests that the session a peer establishes after it answered without one is reported.
+ *
+ * This is the peer restarting, which is the ordinary way into the middle state: its
+ * replication port answers before its domains are up, so it stops the handshake, and the
+ * handshake which completes seconds later is what an operator is waiting for. The only
+ * other line that session writes is
+ * {@code logger.debug(INFO_REPLICATION_SERVER_CONNECTION_TO_RS)}, which is the
+ * {@code information} severity: the replication log holds it and the error log does not,
+ * so a recovery not reported here is a warning left standing over a connection which
+ * replicates.
+ *
+ * What that needs is a record which holds the last state reported rather than
+ * whether an outage is open, and it is the call site rather than the bookkeeping which
+ * has to hold it: reporting the answer without a session through
+ * {@code recordConnected()} passes every test of
+ * {@link ReplicationServer.ConnectFailureReporter} and consumes the record all the same.
+ *
+ * The peer answers on one socket throughout, and the outage is never reopened in
+ * between: a peer which stops answering is a new outage, and the session would then close
+ * that one instead of the answer without a session, which is what this case is about.
+ */
+ @Test
+ public void aSessionEstablishedAfterAnAnswerWithoutOneIsReported() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final int[] ports = TestCaseUtils.findFreePorts(2);
+ final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
+ final String reachable =
+ WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION.get(RS_ID, peerAddress, baseDN).toString();
+ final String connected = NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();
+
+ final ReplicationServer[] servers = new ReplicationServer[1];
+ final ExecutorService peerThread = Executors.newSingleThreadExecutor();
+ try
+ {
+ final ReplSessionSecurity security = getReplSessionSecurity();
+ // What the peer answers with, flipped once the answer without a session has been
+ // reported: the domains of a peer whose port answers come up while it is answering.
+ final AtomicBoolean domainsAreUp = new AtomicBoolean();
+ final List
+ * This is the multi homed peer, and the reason the recovery can be read neither from the
+ * address nor from the session. {@code ServerHandler.toServerAddressURL()} takes the host
+ * of a handler from {@code session.getRemoteAddress()} and its port from the start message
+ * that handler received, so a peer which dials this server from an address it is not
+ * configured under is registered under that other address. Two things follow, and this
+ * test drives both: the already connected branch of {@code runConnect()} compares the
+ * configured address against one which never matches it, so it can close nothing; and the
+ * handshake this server offers that same peer runs into a handler holding its server id
+ * under another address URL, which is {@code ERR_DUPLICATE_REPLICATION_SERVER_ID}, an
+ * abort of this server rather than of the peer, and a session closed at the end of
+ * {@code connect()} for as long as the peer stays where it is.
+ *
+ * Gating the recovery on either leaves the record of such a peer uncleared for good, and
+ * {@code recordFailure()} returns false from then on: the next real outage of it -- the
+ * second one counted below -- is not reported at all.
+ */
+ @Test
+ public void aPeerRegisteredUnderAnotherAddressStillClosesItsOutage() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ /*
+ * Three ports: the server under test, the address the peer is configured under and
+ * answers on, and the address it registers itself under. The last is never bound --
+ * what a multi homed peer costs is that the two addresses are not compared equal, and
+ * a port nothing listens on is that, without a second address to bind.
+ */
+ final int[] ports = TestCaseUtils.findFreePorts(3);
+ final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
+ final HostPort registeredAs = HostPort.valueOf("127.0.0.1:" + ports[2]);
+ final String reachable =
+ WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION.get(RS_ID, peerAddress, baseDN).toString();
+ final String connected =
+ NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();
+
+ final ReplicationServer[] servers = new ReplicationServer[1];
+ // Held rather than returned: the registration lasts as long as the session does, and
+ // whatever the capture below throws, this is what closes it.
+ final Session[] inbound = new Session[1];
+ final ExecutorService peerThread = Executors.newSingleThreadExecutor();
+ try
+ {
+ final ReplSessionSecurity security = getReplSessionSecurity();
+ final List
+ * The registration is what this is for, and the handshake has to reach its end to get it:
+ * {@code ReplicationServerHandler.startFromRemoteRS()} registers the handler after the
+ * second phase, so the topology message of that phase has to be sent and read back.
+ *
+ * @param security
+ * The session security to build the client session with.
+ * @param port
+ * The replication port of the server under test.
+ * @param registeredAs
+ * The address this peer names in its start message, and is therefore registered
+ * under, whatever address it dialled from.
+ * @param baseDN
+ * The base DN of the domain to register with.
+ * @return The session the registration hangs on, closed by the caller.
+ */
+ private Session registerPeerFrom(ReplSessionSecurity security, int port, HostPort registeredAs, DN baseDN)
+ throws Exception
+ {
+ final Socket socket = new Socket();
+ Session session = null;
+ try
+ {
+ socket.setTcpNoDelay(true);
+ socket.connect(new InetSocketAddress("127.0.0.1", port), SOCKET_TIMEOUT_MS);
+ session = security.createClientSession(socket, SOCKET_TIMEOUT_MS);
+ session.publish(peerStartMsg(registeredAs, baseDN));
+ session.receive();
+ // The initiator of a session decides whether it is encrypted, and the start message
+ // above asked for it not to be: both ends leave the SSL session together, right after
+ // the start messages have been exchanged.
+ session.stopEncryption();
+ /*
+ * The second phase: the server reads this one before it sends its own, and registers
+ * the handler once it has sent it. The list holds this peer and nothing else --
+ * waitAndProcessTopoFromRemoteRS() reads rsInfos.get(0) above protocol version 4, so
+ * an empty one ends the handshake on an IndexOutOfBoundsException instead, which is
+ * an abort like any other and would leave the peer unregistered.
+ */
+ final RSInfo peerInfo = new RSInfo(PEER_RS_ID, registeredAs.toString(), -1, (byte) 1, 1);
+ session.publish(new TopologyMsg(Collections.
+ * The mirror image of {@link #registerPeerFrom}: the initiator of a handshake sends the
+ * first start message and the first topology message, so a peer which answers one reads
+ * what that one publishes and publishes what it reads. The topology message holds this
+ * peer and nothing else, for the reason it holds one there.
+ *
+ * @param session
+ * The session the handshake is running on.
+ * @param received
+ * The {@code ReplServerStartMsg} the server under test sent.
+ * @param answersOn
+ * The address this peer answers on, which is what its start message names and
+ * what it is therefore registered under.
+ * @param baseDN
+ * The base DN of the domain the handshake is for.
+ */
+ private void answerWholeHandshake(Session session, ReplicationMsg received, HostPort answersOn, DN baseDN)
+ throws Exception
+ {
+ session.publish(peerStartMsg(answersOn, baseDN));
+ stopEncryptionWith(session, received);
+ // The second phase: the server under test sends its topology message before it waits
+ // for one, so this is read before the answer to it is published.
+ session.receive();
+ final RSInfo peerInfo = new RSInfo(PEER_RS_ID, answersOn.toString(), -1, (byte) 1, 1);
+ session.publish(new TopologyMsg(Collections.
+ * That handler is what makes the outbound handshake of the case reach
+ * {@code ERR_DUPLICATE_REPLICATION_SERVER_ID}: the server ids match and the address URLs
+ * do not. Without it the handshake ends on the second phase instead, which is an abort as
+ * well and reports the same message -- the case would pass while pinning the wrong path.
+ *
+ * @param rs
+ * The server under test.
+ * @param baseDN
+ * The base DN of the domain the peer registered with.
+ * @param registeredAs
+ * The address URL the peer is expected to be registered under.
+ * @param timeoutMs
+ * How long to wait for the registration, in milliseconds.
+ */
+ private void waitForRegistrationUnder(ReplicationServer rs, DN baseDN, HostPort registeredAs, long timeoutMs)
+ throws Exception
+ {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ ReplicationServerHandler registered;
+ while (true)
+ {
+ registered = rs.getReplicationServerDomain(baseDN).getConnectedRSs().get(PEER_RS_ID);
+ if (registered != null || System.currentTimeMillis() > deadline)
+ {
+ break;
+ }
+ Thread.sleep(50);
+ }
+ assertThat(registered).as("the peer should have registered with the domain").isNotNull();
+ assertThat(registered.getServerAddressURL())
+ .as("the peer should be registered under the address its start message named, which is"
+ + " not the one it is configured under")
+ .isEqualTo(registeredAs.toString());
+ }
+
+ /**
+ * Returns the start message of a fake peer, naming the provided address.
+ *
+ * A generation id of -1 leaves the one of the domain alone: a positive one would be
+ * adopted by the handshake, which is a change none of these tests is about.
+ */
+ private ReplServerStartMsg peerStartMsg(HostPort address, DN baseDN)
+ {
+ return new ReplServerStartMsg(PEER_RS_ID, address.toString(), baseDN, 100,
+ new ServerState(), -1, false, (byte) 1, 5000);
+ }
+
+ /**
+ * Leaves the SSL session when the sender of the provided start message does.
+ *
+ * Both ends leave it together, and what a start message asks for is what its sender does
+ * itself: reading the next message on the other stream would be reading a stream nothing
+ * is written to.
+ */
+ private void stopEncryptionWith(Session session, ReplicationMsg startMsg) throws Exception
+ {
+ if (!((ReplServerStartMsg) startMsg).getSSLEncryption())
+ {
+ session.stopEncryption();
+ }
+ }
+
+ /**
+ * Reports a peer which is down, has every handshake it is offered answered by the
+ * provided answer, and asserts that the outage was reported once, closed once, and
+ * reported again once the peer was gone -- the last of which is only reachable because
+ * the first was closed.
+ *
+ * {@code connect()} is driven from the test thread so that each pass is one the test
+ * names, but the connect thread of the server drives it too, for the same peer. What
+ * makes the counts exact is not that the other thread is kept out of them: it is that
+ * {@link ReplicationServer.ConnectFailureReporter} is idempotent in both directions, so
+ * while the record is held every failure the connect thread adds is one the record
+ * already holds, and while it is cleared every recovery it reports is one already
+ * reported. The pass waited for below is what the counts do need, so that the
+ * {@code retainAll} at the end of a pass which started without the domain cannot erase
+ * the record they are about. The blacklist {@code runConnect()} keeps is local to it and
+ * written only from its own failures, so the calls made here do not feed it.
+ *
+ * @param dbName
+ * The changelog directory of the server under test, which is its own.
+ * @param answer
+ * How the peer answers the handshake this server offers it.
+ */
+ private void aPeerWhichAnswersClosesItsOutage(String dbName, PeerHandshake answer) throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final int[] ports = TestCaseUtils.findFreePorts(2);
+ final HostPort peerAddress = HostPort.valueOf("127.0.0.1:" + ports[1]);
+ final String reachable =
+ WARN_REPLICATION_SERVER_REACHABLE_NO_SESSION.get(RS_ID, peerAddress, baseDN).toString();
+ final String connected = NOTE_REPLICATION_SERVER_CONNECT_RESTORED.get(RS_ID, peerAddress, baseDN).toString();
+
+ final ReplicationServer[] servers = new ReplicationServer[1];
+ final ExecutorService peerThread = Executors.newSingleThreadExecutor();
+ try
+ {
+ final ReplSessionSecurity security = getReplSessionSecurity();
+ final List
+ * Every connection is answered rather than only the one a test drives: the connect thread
+ * of the server dials the same peer, and a handshake left unanswered would end on the
+ * connection timeout rather than on the answer, which is a different failure and a far
+ * slower one.
+ */
+ private void answerEveryHandshake(ServerSocket peerSocket, ReplSessionSecurity security,
+ AtomicBoolean serving, PeerHandshake answer)
+ {
+ while (serving.get() && !peerSocket.isClosed())
+ {
+ Socket accepted = null;
+ Session session = null;
+ try
+ {
+ accepted = peerSocket.accept();
+ accepted.setTcpNoDelay(true);
+ session = security.createServerSession(accepted, SOCKET_TIMEOUT_MS);
+ answer.answer(session, session.receive());
+ }
+ catch (Exception ignored)
+ {
+ // Every answer ends by reading what the peer sends next, or does not send, and what
+ // ends that read is the peer closing its own end. What ends this loop is the socket
+ // being closed under the accept() above.
+ }
+ finally
+ {
+ close(session);
+ close(accepted);
+ }
+ }
+ }
+
+ /** How a fake peer answers the {@code ReplServerStartMsg} a handshake starts with. */
+ @FunctionalInterface
+ private interface PeerHandshake
+ {
+ /**
+ * Answers the provided start message on the provided session, and returns when the
+ * handshake is over: the session is closed under it afterwards.
+ *
+ * @param session
+ * The session the handshake is running on.
+ * @param received
+ * The {@code ReplServerStartMsg} the server under test sent.
+ * @throws Exception
+ * When the session ends, which every answer here ends by waiting for.
+ */
+ void answer(Session session, ReplicationMsg received) throws Exception;
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java
index c15e5379e5..08a8eeb87c 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerDynamicConfTest.java
@@ -22,6 +22,7 @@
import static org.testng.Assert.*;
import java.io.File;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
@@ -34,6 +35,8 @@
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigChangeResult;
@@ -521,6 +524,423 @@ public void replServerKeepsListeningWhenAPortChangeIsInterrupted() throws Except
}
}
+ /**
+ * Tests that the listen thread reports a failure of {@code accept()} and waits before
+ * accepting again when it keeps failing on a socket which stays open, instead of
+ * spinning on the failure in silence.
+ *
+ * A process which ran out of file descriptors fails every {@code accept()} without ever
+ * closing the listen socket, so the loop comes straight back to it. The wait is what
+ * bounds that spin, and nothing but the time the loop takes tells a thread which waits
+ * from one which does not.
+ */
+ @Test
+ public void listenThreadReportsAcceptFailuresAndWaitsBeforeAcceptingAgain() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ ReplicationServer replicationServer = null;
+ try
+ {
+ final int[] ports = TestCaseUtils.findFreePorts(1);
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ ports[0], "listenThreadWaitsBeforeAcceptingAgainDb", 0, 1, 0, 0, null));
+ final ReplicationServer listeningServer = replicationServer;
+
+ // Four, so that two waits in a row are measured: a backoff timed from the previous
+ // failure rather than from the previous wait grants one, and the failure which
+ // follows it then looks isolated -- half of a continuous run goes unwaited, and only
+ // a second interval in a row tells that apart.
+ final int failures = 4;
+ final String acceptFailure = "accept() fails the way it fails without a file descriptor left";
+ final AtomicInteger accepts = new AtomicInteger();
+ // When each accept() was entered: the wait is between two of them, and measuring
+ // the whole loop instead would put the cost of everything else in the same budget.
+ final long[] acceptNanos = new long[failures];
+ /*
+ * A socket which fails every accept() and closes itself once it has failed enough
+ * of them: the listen loop ends on a closed socket, which is what returns
+ * runListen(). It closes itself for real rather than overriding isClosed(), which
+ * ServerSocket.close() consults before closing anything up to Java 17: the port and
+ * its file descriptor would be held for the rest of the test JVM.
+ */
+ final ServerSocket failingSocket = new ServerSocket(0)
+ {
+ @Override
+ public Socket accept() throws IOException
+ {
+ final int attempt = accepts.incrementAndGet();
+ acceptNanos[attempt - 1] = System.nanoTime();
+ if (attempt >= failures)
+ {
+ super.close();
+ }
+ throw new IOException(acceptFailure);
+ }
+ };
+
+ final List
+ * The wait bounds a listen loop which is spinning on a failure in silence. A loop which
+ * accepted a connection in between is doing work instead, and charging it a wait per
+ * failure would make a stream of connections aborted between the handshake and
+ * {@code accept()} -- a health check, a port scan -- pace the whole listen port. What
+ * that gives up is the failure which alternates with a connection, under a process which
+ * frees a file descriptor now and then: the accept it lets through resets the clock, and
+ * the failure behind it is timed as isolated. Both halves are this line, so both are
+ * pinned here rather than left to the comment above it.
+ *
+ * The connection served is one no session can be built on, which is the other half of
+ * what the listen loop reports: the accepted socket is closed, so setting its options
+ * fails at once where a socket connected to nothing would spend the whole connection
+ * timeout inside the SSL handshake. It was connected before it was closed, so it still
+ * names the peer the report is about, which is what the report is read for here.
+ */
+ @Test
+ public void listenThreadDoesNotWaitAfterAFailureWhichFollowedAConnection() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ ReplicationServer replicationServer = null;
+ ServerSocket connectedTo = null;
+ try
+ {
+ final int[] ports = TestCaseUtils.findFreePorts(1);
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ ports[0], "listenThreadResetsBackoffOnAConnectionDb", 0, 2, 0, 0, null));
+ final ReplicationServer listeningServer = replicationServer;
+
+ connectedTo = new ServerSocket(0);
+ final Socket served = new Socket();
+ served.connect(new InetSocketAddress("127.0.0.1", connectedTo.getLocalPort()), 10000);
+ final String servedAddress = served.getRemoteSocketAddress().toString();
+ served.close();
+
+ final int attempts = 4;
+ final String acceptFailure = "accept() fails the way it fails without a file descriptor left";
+ final AtomicInteger accepts = new AtomicInteger();
+ // When each accept() was entered: what is measured is between two of them.
+ final long[] acceptNanos = new long[attempts];
+ /*
+ * Fails, serves one connection, fails again, and closes itself on the fourth attempt:
+ * the listen loop ends on a closed socket, which is what returns runListen(). It
+ * closes itself for real rather than overriding isClosed(), which
+ * ServerSocket.close() consults before closing anything up to Java 17: the port and
+ * its file descriptor would be held for the rest of the test JVM.
+ */
+ final ServerSocket failingSocket = new ServerSocket(0)
+ {
+ @Override
+ public Socket accept() throws IOException
+ {
+ final int attempt = accepts.incrementAndGet();
+ acceptNanos[attempt - 1] = System.nanoTime();
+ if (attempt == 2)
+ {
+ return served;
+ }
+ if (attempt >= attempts)
+ {
+ super.close();
+ }
+ throw new IOException(acceptFailure);
+ }
+ };
+
+ final List
+ * A listen port change runs a second listen thread -- {@code switchListenPort()} starts it
+ * before it stops the one it replaces -- so the two overlap, and a five minute window
+ * opened on the port which was left would suppress the first failure on the port which
+ * replaced it: silence in {@code logs/errors} right after the administrator changed the
+ * port to get out of trouble. What that asks of the code is that neither throttle be a
+ * field of the server, and two passes of {@code runListen()} on one server is that
+ * handover without the timing of it: with either throttle kept in a field, the second
+ * pass reports nothing at all.
+ */
+ @Test
+ public void eachListenPassBoundsItsOwnFailures() throws Exception
+ {
+ TestCaseUtils.startServer();
+
+ ReplicationServer replicationServer = null;
+ ServerSocket connectedTo = null;
+ try
+ {
+ final int[] ports = TestCaseUtils.findFreePorts(1);
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ ports[0], "eachListenPassBoundsItsOwnFailuresDb", 0, 2, 0, 0, null));
+ final ReplicationServer listeningServer = replicationServer;
+
+ connectedTo = new ServerSocket(0);
+ final int connectedToPort = connectedTo.getLocalPort();
+ final String acceptFailure = "accept() fails the way it fails without a file descriptor left";
+ // The peers of the two connections, read before they are closed: a closed socket still
+ // names the peer it was connected to, but only the one which was connected.
+ final List