[#1245] Do not report a peer as badly disconnected when a ServerWriter fail...

Valery Kharseko

3 hours ago ba575e4f45f790aea97a4d1bd3fae7f954bfed6c

[#1245] Do not report a peer as badly disconnected when a ServerWriter fails to write to a session this side has closed (#1248)

Fixes #1245

## What

`ServerWriter.run()` now logs the badly-disconnected message (msgID 211
for a directory server, 181 for a replication server) on a
`SocketException` only when the handler is not shutting down and the
close of the session was not initiated on this side:

```java
if (!handler.shuttingDown() && !session.closeInitiated())
{
  errMessage = handler.getBadlyDisconnectedErrorMessage();
  logger.error(errMessage);
}
```

## Why

On a replication server the session runs the publisher thread that
`ServerHandler` starts. While it runs, `publish()` only queues, and when
a write of that thread fails, the failure never reaches the writer. The
writer gets a `SocketException` once a close has taken the session off
that thread, and that close is made on this side:

- `ServerReader`, once the peer has sent its `StopMsg`;
- `ServerReader`, once its read has failed;
- a shutdown of the handler.

So a peer that left in order under load was reported as badly
disconnected, at ERROR, whenever the writer had an update for it at that
moment. That covers a directory server that was stopped or moved by the
load balancing, and a replication server that was stopped. The issue
lists the records from the convergence run, on master and on PR #1244.

The two checks are the ones the other ends already make:

- `ServerReader` checks `handler.shuttingDown()` before the same
message. That flag alone would not cover the writer: the `finally` of
the reader closes the session before `handler.doStop()` sets it.
- The directory server end checks `Session.closeInitiated()` in
`ReplicationBroker.receive()` before
`WARN_REPLICATION_SERVER_BADLY_DISCONNECTED`.

How a connection ended is reported by the reader, as before. A read that
fails on the socket is reported as a bad disconnection, at ERROR. The
writer reports a failing write only on a session whose publisher thread
is not running, and a replication server session runs it from the start.

A peer that goes away without a `StopMsg` leaves an end of the stream,
and the reader reports that only at NOTICE (`NOTE_READER_EXCEPTION`,
msgID 53). Such a peer is older than protocol V4, was killed with
nothing left unread, or gave up a session that had failed. Before this
change the writer added a 211 or 181 at ERROR whenever it had an update
for that peer, which is the same race. Now the NOTICE is all there is.
Whether the reader should report an end of the stream as a bad
disconnection is #1250. It is left out of this PR because it changes
what is logged for an idle peer as well.

## Testing

`ServerWriterDisconnectTest` (new) runs a writer on the thread of the
test. It uses a session over the loopback and a mocked handler that
hands the writer one update, then checks what the writer logged when the
write of that update failed:

| case | the session | expected |
|---|---|---|
| `aWriteFailingOnASessionThisSideClosedIsNotReported` | publisher
thread started, then closed, as `ServerReader` closes it on a `StopMsg`
| no 211 |
| `aWriteFailingOnASessionClosedAfterItsReadFailedIsNotReported` |
publisher thread started, the sockets of the peer closed without a
`StopMsg`, `receive()` failing on it, then closed, as the `finally` of
`ServerReader` closes it | no 211 |
| `aWriteFailingOnASessionNothingClosedIsReported` | publisher thread
not started, its sockets closed under it | one 211 |
| `aWriteFailingWhileTheHandlerShutsDownIsNotReported` | publisher
thread not started, its sockets closed under it, the handler shutting
down | no 211 |

The third case is in a state a replication server session never reaches,
because its publisher thread runs from the start. It keeps the guard
from being one that never reports.

Each case also checks that the writer got as far as the write
(`countSentUpdate`) and that the write failed (`take()` is called once).
It also checks that the writer stopped the handler, and that the write
failed on the socket and not on anything else (no
`ERR_WRITER_UNEXPECTED_EXCEPTION`). The records are read through a
publisher of the test's own (`errorLogRecordsOf`), so each record is one
line. None is deduplicated, so a 211 logged twice fails the third case.
Only the records of the `SYNC` category at ERROR are counted, because
the messages of other categories share the ordinals.

Locally, JDK 21, `mvn -o -Pprecommit verify -pl opendj-server-legacy -am
-Dit.test=ServerWriterDisconnectTest`. On `b76fa3619b`, with the test as
it is now:

- with the fix, 4 of 4 pass, the class in 12 s;
- with the 211 logged twice,
`aWriteFailingOnASessionNothingClosedIsReported` fails;
- with `ERR_WRITER_UNEXPECTED_EXCEPTION` also logged on the
`SocketException`, all four fail.

On `5055b6312f`, with the test before its last round (records collected
into a set, any category):

- with the `ServerWriter` of `origin/master`, 3 of 4 fail: the three
"not reported" cases get the 211;
- with `!session.closeInitiated()` dropped,
`aWriteFailingOnASessionThisSideClosedIsNotReported` and
`aWriteFailingOnASessionClosedAfterItsReadFailedIsNotReported` fail;
- with `!handler.shuttingDown()` dropped,
`aWriteFailingWhileTheHandlerShutsDownIsNotReported` fails.

The convergence run in CI is the end-to-end check. The `restart-ds` rows
fail on these records now, `combined` on master and `rs534` on PR #1244.

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

@@ -145,11 +145,31 @@
    catch (SocketException e)
    {
      /*
       * The remote host has disconnected and this particular Tree is going to
       * be removed, just ignore the exception and let the thread die as well
       * A write which fails on a session this side has closed is no disconnection for this writer
       * to report: ServerReader closes the session once its read has ended, and a shutdown of the
       * handler closes it too - how the connection ended is what the reader saw, not this writer.
       * The reader logs nothing for the StopMsg of a peer leaving in order, and reports a read
       * which failed on the socket as a bad disconnection, at ERROR. An end of the stream with no
       * StopMsg before it - a peer older than protocol V4, a peer killed with nothing left unread,
       * a peer giving up a session which had failed - it reports only at NOTICE, as
       * NOTE_READER_EXCEPTION (issue #1250).
       *
       * And a session closed on this side is the only one this writer meets a failing write on
       * while the session runs the publisher thread ServerHandler starts: publish() then queues, a
       * write of that thread which fails never comes back here, and publish() writes the socket
       * itself once a close has taken the session off the queueing branch. So what is reported
       * below is a write on a session whose publisher thread is not running. Reported here, a peer
       * leaving in order under load - a directory server stopped or moving to another replication
       * server, a replication server stopped - read as a badly disconnected one whenever this
       * writer had an update for it at that moment, and the end of the stream of a peer which went
       * away without a StopMsg read at ERROR or at NOTICE as that same race went. The directory
       * server end makes the same check, in ReplicationBroker.receive().
       */
      errMessage = handler.getBadlyDisconnectedErrorMessage();
      logger.error(errMessage);
      if (!handler.shuttingDown() && !session.closeInitiated())
      {
        errMessage = handler.getBadlyDisconnectedErrorMessage();
        logger.error(errMessage);
      }
    }
    catch (Exception e)
    {

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/ServerWriterDisconnectTest.java

New file
@@ -0,0 +1,351 @@
/*
 * 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.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.opends.messages.ReplicationMessages.ERR_DS_BADLY_DISCONNECTED;
import static org.opends.messages.ReplicationMessages.ERR_WRITER_UNEXPECTED_EXCEPTION;
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;

import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Field;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.util.StaticUtils;
import org.testng.annotations.Test;

/**
 * A {@code ServerWriter} reports a write which fails on a session nothing on this side has closed
 * as a peer badly disconnected - msgID 211 for a directory server, 181 for a replication server,
 * both at ERROR - but not a write which fails on a session this side has closed itself:
 * {@code ServerReader} closes the session once its read has ended, and a writer which had an
 * update for that peer at that moment then fails on the closed socket. Under load that is every
 * peer which leaves in order: a directory server stopped, or moving to another replication
 * server, a replication server stopped - the restart rows of the replication convergence run
 * (#1208) logged it for each of them. How the connection ended is the reader's to report.
 * <p>
 * The writer runs on the thread of the test, against a session over the loopback and a handler
 * which hands it one update: what is under test is only what the writer logs when the write of
 * that update fails.
 */
@SuppressWarnings("javadoc")
public class ServerWriterDisconnectTest extends ReplicationTestCase
{
  private static final int RS_ID = 9301;
  private static final int SOCKET_TIMEOUT_MS = 5000;

  /**
   * The session runs its publisher thread, as {@code ServerHandler} starts it, and is closed as
   * {@code ServerReader} closes it on the {@code StopMsg} of the peer. The update the writer takes
   * next is written by {@code publish()} itself - the close took the session off the queueing
   * branch - and fails on the closed socket. That is no disconnection of the peer.
   */
  @Test
  public void aWriteFailingOnASessionThisSideClosedIsNotReported() throws Exception
  {
    final int dsId = 9311;
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session session = pair[1];
      try
      {
        session.start();
        session.waitForStartup();
        session.close();

        final List<String> logged = runWriterWithOneUpdate(session, dsId, false);

        assertThat(badlyDisconnected(logged, dsId))
            .as("a write which failed on a session closed on this side was reported as a peer "
                + "badly disconnected")
            .isEmpty();
      }
      finally
      {
        StaticUtils.close(pair);
      }
    }
  }

  /**
   * The sockets of the session are closed under it, nothing on this side having closed the
   * session, and its publisher thread is not running: the write fails on a connection which broke,
   * and the writer reports it. A replication server session does not get here - see
   * {@link #closeTheSocketsUnder(Session)} - this is what keeps the guard of the writer from being
   * one that never reports.
   */
  @Test
  public void aWriteFailingOnASessionNothingClosedIsReported() throws Exception
  {
    final int dsId = 9312;
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session session = pair[1];
      try
      {
        closeTheSocketsUnder(session);

        final List<String> logged = runWriterWithOneUpdate(session, dsId, false);

        assertThat(badlyDisconnected(logged, dsId))
            .as("a write which failed on a connection which broke under the session was not "
                + "reported")
            .hasSize(1);
      }
      finally
      {
        StaticUtils.close(pair);
      }
    }
  }

  /**
   * The same broken connection while the handler is being shut down: what fails then is the end
   * of a session the shutdown is taking down anyway, which {@code ServerReader} does not report
   * either.
   */
  @Test
  public void aWriteFailingWhileTheHandlerShutsDownIsNotReported() throws Exception
  {
    final int dsId = 9313;
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session session = pair[1];
      try
      {
        closeTheSocketsUnder(session);

        final List<String> logged = runWriterWithOneUpdate(session, dsId, true);

        assertThat(badlyDisconnected(logged, dsId))
            .as("a write which failed while the handler was shutting down was reported as a peer "
                + "badly disconnected")
            .isEmpty();
      }
      finally
      {
        StaticUtils.close(pair);
      }
    }
  }

  /**
   * The peer goes away without a {@code StopMsg} - killed, older than protocol V4, or giving up a
   * session which had failed - and the read of the session fails on it. {@code ServerReader} then
   * reports the end of the connection itself and closes the session in its {@code finally}, as
   * here: the close takes the branch of a session which failed and writes nothing. The update the
   * writer takes next fails on the closed socket, and the writer reports nothing.
   */
  @Test
  public void aWriteFailingOnASessionClosedAfterItsReadFailedIsNotReported() throws Exception
  {
    final int dsId = 9314;
    try (ServerSocket listen = new ServerSocket(0))
    {
      final Session[] pair = connectSessionPair(listen);
      final Session session = pair[1];
      try
      {
        session.start();
        session.waitForStartup();
        closeTheSocketsUnder(pair[0]);
        // an end of the stream, or a reset when the peer had left something unread
        assertThatThrownBy(session::receive)
            .as("the read of a session whose peer went away")
            .isInstanceOf(IOException.class)
            .isNotInstanceOf(SocketTimeoutException.class);
        session.close();

        final List<String> logged = runWriterWithOneUpdate(session, dsId, false);

        assertThat(badlyDisconnected(logged, dsId))
            .as("a write which failed on a session closed after its read failed was reported as "
                + "a peer badly disconnected")
            .isEmpty();
      }
      finally
      {
        StaticUtils.close(pair);
      }
    }
  }

  /**
   * Runs a writer for a directory server on the thread of the test, its handler handing it one
   * update, and answers what was logged meanwhile. It checks that the writer got as far as the
   * write, and that the write failed - a write which returned would have the writer take the next
   * update - on the socket rather than on anything else.
   */
  private List<String> runWriterWithOneUpdate(final Session session, final int dsId,
      final boolean shuttingDown) throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    final UpdateMsg update = new DeleteMsg(DN.valueOf("uid=writer," + TEST_ROOT_DN_STRING),
        new CSNGenerator(dsId, 0).newCSN(), "00000000-0000-0000-0000-000000000000");

    final DataServerHandler handler = mock(DataServerHandler.class);
    /*
     * The handler reaches getReplicationServerId() of the package-private MessageHandler through a
     * bridge method, which the mock does not intercept: the replication server it reads is set.
     */
    final ReplicationServer replicationServer = mock(ReplicationServer.class);
    when(replicationServer.getServerId()).thenReturn(RS_ID);
    final Field replicationServerField = MessageHandler.class.getDeclaredField("replicationServer");
    replicationServerField.setAccessible(true);
    replicationServerField.set(handler, replicationServer);
    when(handler.take()).thenReturn(update, (UpdateMsg) null);
    when(handler.isDataServer()).thenReturn(true);
    when(handler.getServerId()).thenReturn(dsId);
    when(handler.getProtocolVersion()).thenReturn(ProtocolVersion.getCurrentVersion());
    when(handler.shuttingDown()).thenReturn(shuttingDown);
    when(handler.getBadlyDisconnectedErrorMessage()).thenReturn(ERR_DS_BADLY_DISCONNECTED.get(
        RS_ID, dsId, session.getReadableRemoteAddress(), baseDN));
    final ReplicationServerDomain domain = mock(ReplicationServerDomain.class);
    when(domain.getBaseDN()).thenReturn(baseDN);

    /*
     * A publisher of its own rather than the error log of the tests, which holds every record
     * twice, each copy with a timestamp of its own: one record is then one line.
     */
    final List<String> logged = errorLogRecordsOf(() ->
    {
      new ServerWriter(session, handler, domain, new DSRSShutdownSync()).run();
      return null;
    });

    verify(handler).countSentUpdate(update);
    verify(handler).take();
    verify(domain).stopServer(handler, false);
    assertThat(withMsgID(logged, ERR_WRITER_UNEXPECTED_EXCEPTION.ordinal()))
        .as("the write of the update failed on something else than the socket")
        .isEmpty();
    return logged;
  }

  private static List<String> badlyDisconnected(final List<String> logged, final int dsId)
  {
    final List<String> records = withMsgID(logged, ERR_DS_BADLY_DISCONNECTED.ordinal());
    records.removeIf(record -> !record.contains("DS(" + dsId + ")"));
    return records;
  }

  /**
   * Answers the records logged at ERROR with the provided ordinal of the replication messages. The
   * category is matched as well: the messages of other categories share the ordinals, and another
   * thread of the server may log one of them meanwhile. Every record is kept, none deduplicated:
   * the timestamp resolves only to the second, so a message logged twice in a row is the same line
   * twice.
   */
  private static List<String> withMsgID(final List<String> logged, final int msgID)
  {
    final List<String> records = new ArrayList<>();
    for (final String record : logged)
    {
      if (record.contains("category=SYNC severity=ERROR msgID=" + msgID + " "))
      {
        records.add(record);
      }
    }
    return records;
  }

  /**
   * Closes the sockets a session writes through while leaving the session unaware of it, so that
   * its next write fails. A session whose publisher thread is not running is then in the state a
   * connection which broke leaves it in. A replication server session runs it - ServerHandler
   * starts it - and is never in that state: its publish() queues, and the publisher records a
   * write which failed without handing it to the writer, until a close takes the session off the
   * queueing branch.
   * <p>
   * The plain socket is closed first, as Session.close() does: the close of the SSL socket layered
   * on it would otherwise wait out the timeout of the socket for a close_notify of the peer.
   */
  private static void closeTheSocketsUnder(final Session session) throws Exception
  {
    for (final String name : new String[] { "plainSocket", "secureSocket" })
    {
      final Field socket = Session.class.getDeclaredField(name);
      socket.setAccessible(true);
      StaticUtils.close((Closeable) socket.get(session));
    }
  }

  /**
   * A connected pair of sessions over the loopback, the client end first, neither started. The
   * replication server end is the second.
   */
  private Session[] connectSessionPair(final ServerSocket listen) throws Exception
  {
    final ReplSessionSecurity security = getReplSessionSecurity();
    final Socket clientSocket = new Socket("127.0.0.1", listen.getLocalPort());
    clientSocket.setTcpNoDelay(true);
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    try
    {
      final Future<Session> clientEnd = executor.submit(new Callable<Session>()
      {
        @Override
        public Session call() throws Exception
        {
          return security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS);
        }
      });
      final Socket serverSocket = listen.accept();
      serverSocket.setTcpNoDelay(true);
      final Session serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS);
      try
      {
        return new Session[] { clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
      }
      catch (Exception e)
      {
        // the factories close what they opened when they fail, but not the end already built
        StaticUtils.close(serverEnd);
        throw e;
      }
    }
    finally
    {
      executor.shutdown();
    }
  }
}