[#1250] Report unexpected replication stream EOF as a bad disconnection (#1...

Valery Kharseko

13 hours ago a77994116fdc36bff765e79f85fdb29027259f32

[#1250] Report unexpected replication stream EOF as a bad disconnection (#1254)

Fixes #1250

A replication server reports an unexpected end of a peer's stream at
NOTICE, while a reset is reported as a bad disconnection at ERROR. After
#1248, the reader's NOTICE is the only record when the peer closes
without a `StopMsg` and the read reaches EOF.

`Session.read()` now throws `EOFException`. For protocol V4 and later,
`ServerReader` logs the existing badly-disconnected error (211 for a
directory server, 181 for a replication server), unless the handler is
shutting down or this side has initiated the session close. Those
local-close and shutdown cases retain NOTICE 53. Older peers also retain
the NOTICE, since they do not send `StopMsg`. An actual `StopMsg` still
ends the reader without a disconnection report. Where a log record
includes the exception class, EOF now appears as `EOFException: no more
data` instead of `IOException: no more data`; external log parsers
matching the old text may need updating.

A peer whose session has recorded an error, including an abort by the
stop watchdog or heartbeat monitor, sends no `StopMsg`. EOF from such a
V4+ peer is intentionally reported at ERROR. The convergence filter
continues to reject 181/211 in a graceful restart window; admitting them
would also hide failures that already produced a reset. The reader and
writer comments describe the session-error exception and the
local-close/shutdown conditions.

The reader tests use a real `Session` and controlled input streams.
Eight cases cover EOF at the V4 boundary and at the current protocol
version, EOF during a frame, an older peer, shutdown, local close before
reading, local close while reading is blocked, and `StopMsg`. The
shutdown case requires exactly one NOTICE 53; the `StopMsg` case
requires neither ERROR 211 nor NOTICE 53. The concurrent case releases
EOF only after local close; it uses bounded waits, propagates reader
failures through a future, and checks no ERROR 211 and exactly one
NOTICE 53.

**Validation**

On the current revision, JDK 21:

```sh
mvn -o -Pprecommit verify -pl opendj-server-legacy -am \
  -Dit.test=ServerReaderDisconnectTest,ServerWriterDisconnectTest,SessionTest \
  -Dfailsafe.failIfNoSpecifiedTests=false
```

All 18 selected tests pass: 8 reader, 4 writer and 6 session tests.
Reactor verify succeeds.

The eight reader cases were also run directly with TestNG on the
reactor's test classpath, using separately compiled reader variants. The
unmodified reader passes all eight. Each of these mutations fails
exactly the corresponding test, with the other seven passing:

- Restrict the EOF error guard to protocol V4 (`==` instead of `>=`):
`eofFromCurrentPeerIsReported` fails because ERROR 211 is missing.
- Suppress the EOF record during shutdown:
`eofDuringShutdownIsNotReportedAsBadDisconnection` fails because NOTICE
53 is missing.
- Throw from the `StopMsg` arm:
`stopMessageIsNotReportedAsBadDisconnection` fails on the unexpected
NOTICE 53.

The mutation classes were built outside the reactor output directories;
the final reactor verify used the unmodified production code.

Earlier head `af6ec1301b` passed [CI run
41065150477](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/41065150477),
including all four convergence rows and Docker default/alpine. That
result does not establish coverage of the abort path. Convergence and
Docker replication were not run locally for this revision.

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

@@ -22,6 +22,7 @@
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.Closeable;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -795,7 +796,7 @@
      if (read == -1)
      {
        lastReceiveTime = 0;
        throw new IOException("no more data");
        throw new EOFException("no more data");
      }
      length += read;
    }

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

@@ -13,9 +13,11 @@
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

import java.io.EOFException;
import java.net.SocketException;

import org.forgerock.i18n.LocalizableMessage;
@@ -206,6 +208,22 @@
        logger.error(errMessage);
      }
    }
    catch (EOFException e)
    {
      // V4 and later peers send StopMsg before an orderly close unless their session recorded an error.
      if (!handler.shuttingDown() && !session.closeInitiated()
          && session.getProtocolVersion() >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
      {
        logException(e);
        errMessage = handler.getBadlyDisconnectedErrorMessage();
        logger.error(errMessage);
      }
      else
      {
        errMessage = NOTE_READER_EXCEPTION.get(handler, stackTraceToSingleLineString(e));
        logger.info(errMessage);
      }
    }
    catch (Exception e)
    {
      /*

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

@@ -150,9 +150,9 @@
       * 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).
       * StopMsg before it is reported as a bad disconnection for a peer using protocol V4 or
       * later, unless the handler is shutting down or this side has begun the close; otherwise,
       * and for an older peer, at NOTICE.
       *
       * 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

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

New file
@@ -0,0 +1,246 @@
/*
 * 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.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.opends.messages.ReplicationMessages.ERR_DS_BADLY_DISCONNECTED;
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLSocket;

import org.forgerock.opendj.ldap.DN;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.StopMsg;
import org.testng.annotations.Test;

/** Checks the records emitted when a replication server reader loses its peer. */
@SuppressWarnings("javadoc")
public class ServerReaderDisconnectTest extends ReplicationTestCase
{
  @Test
  public void eofFromModernPeerIsReported() throws Exception
  {
    assertThat(records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V4, false, false))
        .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 "))
        .hasSize(1);
  }

  @Test
  public void eofFromCurrentPeerIsReported() throws Exception
  {
    assertThat(records(new byte[0], ProtocolVersion.getCurrentVersion(), false, false))
        .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 "))
        .hasSize(1);
  }

  @Test
  public void truncatedFrameFromModernPeerIsReported() throws Exception
  {
    assertThat(records("00000004x".getBytes(StandardCharsets.US_ASCII),
        ProtocolVersion.REPLICATION_PROTOCOL_V4, false, false))
        .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 "))
        .hasSize(1);
  }

  @Test
  public void eofFromOldPeerKeepsNotice() throws Exception
  {
    final List<String> logged = records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V3, false, false);
    assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
    assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).hasSize(1);
  }

  @Test
  public void eofDuringShutdownIsNotReportedAsBadDisconnection() throws Exception
  {
    final List<String> logged = records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V4, true, false);
    assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
    assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).hasSize(1);
  }

  @Test
  public void eofAfterLocalCloseIsNotReportedAsBadDisconnection() throws Exception
  {
    assertThat(records(new byte[0], ProtocolVersion.REPLICATION_PROTOCOL_V4, false, true))
        .filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
  }

  @Test
  public void eofAfterConcurrentLocalCloseIsNotReportedAsBadDisconnection() throws Exception
  {
    final CountDownLatch reading = new CountDownLatch(1);
    final CountDownLatch eof = new CountDownLatch(1);
    final InputStream input = new InputStream()
    {
      @Override
      public int read() throws IOException
      {
        reading.countDown();
        try
        {
          assertThat(eof.await(10, TimeUnit.SECONDS)).as("local close released the read").isTrue();
        }
        catch (InterruptedException e)
        {
          Thread.currentThread().interrupt();
          throw new InterruptedIOException("Interrupted while waiting for EOF");
        }
        return -1;
      }
    };
    try (ServerSocket listener = new ServerSocket(0);
         Socket peer = new Socket("127.0.0.1", listener.getLocalPort());
         Socket socket = listener.accept();
         Session session = new Session(socket, new TestSocket(input)))
    {
      session.setProtocolVersion(ProtocolVersion.REPLICATION_PROTOCOL_V4);
      final DataServerHandler handler = handler(session, false);
      final ExecutorService executor = Executors.newSingleThreadExecutor();
      try
      {
        final List<String> logged = errorLogRecordsOf(() -> {
          final Future<?> reader = executor.submit(() -> new ServerReader(session, handler).run());
          try
          {
            assertThat(reading.await(10, TimeUnit.SECONDS)).as("reader entered receive").isTrue();
            // Reading closeInitiated before receive() would miss this close and log a false ERROR.
            session.close();
            eof.countDown();
            reader.get(10, TimeUnit.SECONDS);
          }
          finally
          {
            eof.countDown();
            executor.shutdownNow();
            assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).as("reader stopped").isTrue();
          }
          return null;
        });
        assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
        assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).hasSize(1);
      }
      finally
      {
        executor.shutdownNow();
      }
    }
  }

  @Test
  public void stopMessageIsNotReportedAsBadDisconnection() throws Exception
  {
    final byte[] message = new StopMsg().getBytes(ProtocolVersion.REPLICATION_PROTOCOL_V4);
    final ByteArrayOutputStream frame = new ByteArrayOutputStream();
    frame.write(String.format("%08x", message.length).getBytes(StandardCharsets.US_ASCII));
    frame.write(message);
    final List<String> logged = records(frame.toByteArray(), ProtocolVersion.REPLICATION_PROTOCOL_V4, false, false);
    assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=ERROR msgID=211 ")).isEmpty();
    assertThat(logged).filteredOn(s -> s.contains("category=SYNC severity=NOTICE msgID=53 ")).isEmpty();
  }

  private List<String> records(final byte[] input, final short version, final boolean shuttingDown,
      final boolean closeFirst) throws Exception
  {
    try (ServerSocket listener = new ServerSocket(0);
         Socket peer = new Socket("127.0.0.1", listener.getLocalPort());
         Socket socket = listener.accept())
    {
      final SSLSocket secure = new TestSocket(input);
      final Session session = new Session(socket, secure);
      session.setProtocolVersion(version);
      final DataServerHandler handler = handler(session, shuttingDown);
      if (closeFirst)
      {
        session.close();
      }
      try
      {
        return errorLogRecordsOf(() -> {
          new ServerReader(session, handler).run();
          return null;
        });
      }
      finally
      {
        session.close();
      }
    }
  }

  private static DataServerHandler handler(final Session session, final boolean shuttingDown) throws Exception
  {
    final DataServerHandler handler = mock(DataServerHandler.class);
    final ReplicationServer server = mock(ReplicationServer.class);
    when(server.getServerId()).thenReturn(9301);
    final java.lang.reflect.Field field = MessageHandler.class.getDeclaredField("replicationServer");
    field.setAccessible(true);
    field.set(handler, server);
    when(handler.shuttingDown()).thenReturn(shuttingDown);
    when(handler.getBadlyDisconnectedErrorMessage()).thenReturn(ERR_DS_BADLY_DISCONNECTED.get(
        9301, 9311, session.getReadableRemoteAddress(), DN.valueOf(TEST_ROOT_DN_STRING)));
    return handler;
  }

  private static final class TestSocket extends SSLSocket
  {
    private final InputStream input;
    private final OutputStream output = new ByteArrayOutputStream();

    TestSocket(final byte[] bytes) { this(new ByteArrayInputStream(bytes)); }
    TestSocket(final InputStream input) { this.input = input; }
    @Override public InputStream getInputStream() { return input; }
    @Override public OutputStream getOutputStream() { return output; }
    @Override public String[] getSupportedCipherSuites() { return new String[0]; }
    @Override public String[] getEnabledCipherSuites() { return new String[0]; }
    @Override public void setEnabledCipherSuites(String[] suites) { }
    @Override public String[] getSupportedProtocols() { return new String[0]; }
    @Override public String[] getEnabledProtocols() { return new String[0]; }
    @Override public void setEnabledProtocols(String[] protocols) { }
    @Override public javax.net.ssl.SSLSession getSession() { return null; }
    @Override public void addHandshakeCompletedListener(javax.net.ssl.HandshakeCompletedListener listener) { }
    @Override public void removeHandshakeCompletedListener(javax.net.ssl.HandshakeCompletedListener listener) { }
    @Override public void startHandshake() throws IOException { }
    @Override public void setUseClientMode(boolean mode) { }
    @Override public boolean getUseClientMode() { return false; }
    @Override public void setNeedClientAuth(boolean need) { }
    @Override public boolean getNeedClientAuth() { return false; }
    @Override public void setWantClientAuth(boolean want) { }
    @Override public boolean getWantClientAuth() { return false; }
    @Override public void setEnableSessionCreation(boolean flag) { }
    @Override public boolean getEnableSessionCreation() { return false; }
  }
}