From bf0f4f294f477e393b59ef7f160795e00aa02cc9 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 24 Sep 2026 07:57:37 +0000
Subject: [PATCH] Send what a replication session's publisher left queued when it is closed (#1035)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java | 263 ++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java | 10
opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java | 759 ++++++++++++++++++++++++++++++++++++++++++
3 files changed, 1,020 insertions(+), 12 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
index 5965219..bd867c0 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2009 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.protocol;
@@ -30,12 +31,14 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.zip.DataFormatException;
import javax.net.ssl.SSLSocket;
+import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.api.DirectoryThread;
import org.opends.server.types.HostPort;
@@ -48,6 +51,19 @@
{
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+ /**
+ * How long a close spends sending what the publisher thread left queued in {@code sendQueue},
+ * in milliseconds.
+ * <p>
+ * The same 5 s as {@code DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD}, which is how long a
+ * shutdown is already willing to wait for one of these messages - the announcement that a
+ * replica went offline - to be forwarded. A close has no reason to wait longer for it than the
+ * shutdown which is waiting on the close. It is a budget per close, though, not per shutdown:
+ * a shutdown closes its sessions one after another, so it can pay this once for each peer which
+ * is alive and not reading.
+ */
+ private static final long DRAIN_BUDGET_MS = 5000;
+
private final Socket plainSocket;
private final SSLSocket secureSocket;
private final InputStream plainInput;
@@ -101,6 +117,14 @@
private final CountDownLatch latch = new CountDownLatch(1);
/**
+ * How many buffers the publisher thread took off {@code sendQueue} and failed to write. After a
+ * failed write that thread goes on taking the queue until the session is closed, and every
+ * write after the first fails as well, so these are messages the peer was not told about just
+ * as much as what is still queued - see {@link #close()}.
+ */
+ private final AtomicInteger publisherFailedWrites = new AtomicInteger();
+
+ /**
* Creates a new Session.
*
* @param socket
@@ -140,6 +164,10 @@
/**
* This method is called when the session with the remote must be closed.
* This object won't be used anymore after this method is called.
+ * <p>
+ * A message which was published on this session but which its publisher thread had not sent yet
+ * is sent here rather than dropped, within the budget of {@link #DRAIN_BUDGET_MS}. See {@link
+ * #sendWhatThePublisherLeftQueued()}.
*/
@Override
public void close()
@@ -165,6 +193,18 @@
Thread.currentThread().interrupt();
}
+ /*
+ * Re-read the error rather than answer with the snapshot taken before the join: a publisher
+ * whose send() failed while this thread was joining it recorded the error there, and what
+ * follows - the drain and the StopMsg - is what must not be written to a socket which has
+ * already failed. Reading it before the join left both writing to one, the drain naming its
+ * own failure rather than the one the publisher had recorded.
+ */
+ synchronized (stateLock)
+ {
+ localSessionError = sessionError;
+ }
+
// Perform close outside of critical section.
if (logger.isTraceEnabled())
{
@@ -186,18 +226,72 @@
}
}
- // V4 protocol introduces a StopMsg to properly end communications.
- if (localSessionError == null
- && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
+ if (localSessionError != null)
{
- try
+ /*
+ * Nothing more is written to a session which already failed, neither what the publisher left
+ * queued nor the StopMsg: writing more to it cannot work. What it gives up on is still
+ * reported: what is left in the queue, and what the publisher took off it and failed to
+ * write - the write which failed first, and every one it went on failing until this close
+ * stopped it, which the join above makes complete. It is done without taking publishLock,
+ * so that a thread blocked in a write of this socket is released by the close of the
+ * sockets below rather than waited for.
+ */
+ isRunning.set(false);
+ final int notSent = publisherFailedWrites.get() + takeWhatIsLeftQueued();
+ if (notSent > 0)
{
- publish(new StopMsg());
+ reportQueueNotSent(notSent, "the session had already failed: " + localSessionError);
}
- catch (final IOException ignored)
+ StaticUtils.close(plainSocket, secureSocket);
+ return;
+ }
+
+ /*
+ * The publisher thread has stopped and what it had not sent is still in the queue. Send it,
+ * rather than let the close drop it: nothing publishes these again, and the StopMsg which
+ * follows leaves the peer reading an orderly close with no sign that anything was missing.
+ *
+ * This thread is not the only one which can write the socket here - a ServerWriter or a
+ * HeartbeatThread outlives this close. It is this thread, not run(), which takes the session
+ * off the queueing branch of publish(), and it does so under publishLock and keeps the lock
+ * across the queue and the StopMsg: until then a publish() concurrent with the close takes
+ * the queueing branch, and from then on it takes the synchronous branch and waits for the
+ * lock. Either way nothing newer is written between two of the drained messages - see
+ * sendWhatThePublisherLeftQueued() below.
+ */
+ publishLock.lock();
+ try
+ {
+ isRunning.set(false);
+ sendWhatThePublisherLeftQueued();
+
+ /*
+ * Re-read again: a write of the drain which failed recorded its error, and the StopMsg must
+ * not be written to a socket which has already failed either.
+ */
+ synchronized (stateLock)
{
- // Ignore errors on close.
+ localSessionError = sessionError;
}
+
+ // V4 protocol introduces a StopMsg to properly end communications.
+ if (localSessionError == null
+ && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
+ {
+ try
+ {
+ publish(new StopMsg());
+ }
+ catch (final IOException ignored)
+ {
+ // Ignore errors on close.
+ }
+ }
+ }
+ finally
+ {
+ publishLock.unlock();
}
StaticUtils.close(plainSocket, secureSocket);
@@ -206,6 +300,101 @@
/**
+ * Sends the buffers the publisher thread had not sent when it stopped, so that a close of the
+ * session does not drop them.
+ * <p>
+ * Called from {@link #close()} once the publisher has been joined, with {@code publishLock}
+ * held. A queued message is already encoded for this peer's protocol version - {@link
+ * #publish(ReplicationMsg)} did that before queueing it - so there is nothing to decide here
+ * beyond how long to keep trying.
+ * <p>
+ * The whole queue goes out under {@code publishLock}, because the publisher is not the only
+ * thread which writes this socket: a {@code ServerWriter} is joined only after the close which
+ * gets here (ServerHandler.shutdown() closes the session before joining it, and ServerReader's
+ * finally closes it before stopping the handler), and a {@code HeartbeatThread} is shut down
+ * after it too. Their {@code publish()} takes the synchronous branch once the session is off
+ * the queueing one, so without the lock a newer message could be written between two of these
+ * older ones - which a peer replication server answers by dropping the older ones at debug
+ * level, its log file refusing a record which would break its key ordering. The close takes
+ * the session off the queueing branch under the same lock it holds here, so such a
+ * {@code publish()} either queues ahead of that and is drained here, or waits for the drain and
+ * lands after it, or returns at the door, having seen the close. One which read the close as
+ * not yet begun can still be descheduled before its buffer is queued and queue it after the
+ * last poll below; it checks for that once it has, and takes the buffer back and reports it -
+ * see {@link #publish(ReplicationMsg)}. The wait for the lock itself is not part of the budget
+ * below, no more than it is for the {@code StopMsg} which follows.
+ * <p>
+ * The budget bounds how many messages a close spends on a peer which is reading slowly. A peer
+ * which is reading pays none of it; a peer which answers with a reset pays one failed write. It
+ * is checked between messages, so a single write which blocks past the budget still runs to
+ * completion - for a peer which has stopped reading, or which vanished without a reset, that is
+ * for as long as TCP keeps the connection alive: bounding it needs a non-blocking socket, which
+ * this session is not. The budget is per close, and a shutdown closes its sessions one after
+ * another - ReplicationServerDomain.stopAllServers() stops each handler in turn on one thread -
+ * so a domain with several peers which are alive and not reading pays it once per such peer.
+ * On the road which does not shut the whole server down the wait is paid under the lock of the
+ * replication domain - ReplicationServerDomain.stopServer() holds it across the handler
+ * shutdown which closes this session - where a handshake meanwhile waiting on that lock times
+ * out and the broker retries. What is given up on is reported rather than dropped in silence,
+ * that being the part of this which cost the most to diagnose.
+ */
+ private void sendWhatThePublisherLeftQueued()
+ {
+ final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS);
+ byte[] buffer;
+ while ((buffer = sendQueue.poll()) != null)
+ {
+ if (System.nanoTime() - deadline >= 0)
+ {
+ reportQueueNotSent(1 + takeWhatIsLeftQueued(), "the peer did not read them within "
+ + DRAIN_BUDGET_MS + " ms");
+ return;
+ }
+ try
+ {
+ send(buffer);
+ }
+ catch (final IOException e)
+ {
+ /*
+ * send() has recorded the error; the rest of the queue cannot go out either. The
+ * exception is named rather than traced: this is reached whenever a peer which has
+ * announced it is leaving closes before the drain reaches it, where what the write
+ * failed with is the whole of what a reader of the log needs - a directory server
+ * re-reads these from the changelog when it reconnects, a replication server does not.
+ */
+ reportQueueNotSent(1 + takeWhatIsLeftQueued(),
+ "the write failed with " + e.getClass().getName() + ": " + e.getMessage());
+ return;
+ }
+ }
+ }
+
+ /**
+ * Empties the queue a close gives up on, and answers how many buffers it held. Taking them
+ * rather than counting them is what keeps a {@code publish()} which queued a buffer late from
+ * reporting it a second time: it reports only a buffer it still finds in the queue.
+ */
+ private int takeWhatIsLeftQueued()
+ {
+ int taken = 0;
+ while (sendQueue.poll() != null)
+ {
+ taken++;
+ }
+ return taken;
+ }
+
+ /** Says which messages a close could not hand to the peer, and why. */
+ private void reportQueueNotSent(final int count, final String reason)
+ {
+ logger.warn(LocalizableMessage.raw(
+ "The replication session %s was closed with %d message(s) which had been published on it "
+ + "but not yet sent, and the peer was not told about them: %s",
+ getName(), count, reason));
+ }
+
+ /**
* This methods allows to determine if the session close was initiated
* on this Session.
*
@@ -322,6 +511,10 @@
// Avoid blocking forever so that we can check for session closure.
if (sendQueue.offer(buffer, 100, TimeUnit.MILLISECONDS))
{
+ if (!isRunning.get())
+ {
+ takeBackWhatWasQueuedTooLate(buffer);
+ }
return;
}
}
@@ -338,6 +531,34 @@
}
}
+ /**
+ * Takes a buffer back out of the queue if nothing is going to send it, and reports it.
+ * <p>
+ * {@code publish()} reads the close as not yet begun and queues the buffer after that, with no
+ * lock in between: descheduled there, it can queue the buffer after a close has stopped the
+ * publisher thread and drained the queue, and nothing would then send it or say so. Once the
+ * session is off the queueing branch, the lock here is the one the close holds while it takes
+ * the session off that branch and drains the queue, so what is still in the queue after it is
+ * what nothing sends. A buffer queued before the session came off the queueing branch is left
+ * to the drain - it cannot have seen the flag cleared - and a buffer the drain or the close
+ * already took is not found here, so nothing is reported twice.
+ */
+ private void takeBackWhatWasQueuedTooLate(final byte[] buffer)
+ {
+ publishLock.lock();
+ try
+ {
+ if (sendQueue.remove(buffer))
+ {
+ reportQueueNotSent(1, "it was queued after the publisher of the session had stopped");
+ }
+ }
+ finally
+ {
+ publishLock.unlock();
+ }
+ }
+
/** Sends a replication message already encoded to the socket.
*
* @param buffer
@@ -526,7 +747,20 @@
@Override
public void run()
{
- isRunning.set(true);
+ synchronized (stateLock)
+ {
+ /*
+ * A close which came before the start has already run, and it is not coming back to clear
+ * the flag - nor is the end of this method, which leaves that to the close. Set, the flag
+ * would keep every later publish() on the queueing branch, which returns at once on a
+ * closed session as if the message had been queued; unset, publish() stays on the
+ * synchronous branch, which fails on the closed socket.
+ */
+ if (!closeInitiated)
+ {
+ isRunning.set(true);
+ }
+ }
latch.countDown();
if (logger.isTraceEnabled())
{
@@ -551,10 +785,21 @@
catch (IOException e)
{
setSessionError(e);
+ publisherFailedWrites.incrementAndGet();
needClosing = true;
}
}
- isRunning.set(false);
+ /*
+ * A close clears the flag itself, under publishLock, once it has joined this thread - see
+ * close(). Clearing it here would open a window between the end of this thread and the drain
+ * of the close, in which a publish() takes the synchronous branch and writes a newer message
+ * ahead of the queue. Only a loop which ended without a close - an interrupt from elsewhere -
+ * clears it here, so that publish() does not go on queueing onto a queue nobody sends.
+ */
+ if (!closeInitiated)
+ {
+ isRunning.set(false);
+ }
if (needClosing)
{
close();
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java
new file mode 100644
index 0000000..915a312
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java
@@ -0,0 +1,759 @@
+/*
+ * 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.protocol;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
+
+import java.io.Closeable;
+import java.lang.reflect.Field;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Queue;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.forgerock.opendj.ldap.DN;
+import org.opends.server.TestCaseUtils;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.util.StaticUtils;
+import org.testng.annotations.Test;
+
+/**
+ * Which end of a replication session has a publisher thread, and what a close of the session
+ * does to the messages that thread has not sent yet.
+ * <p>
+ * {@link Session#publish(ReplicationMsg)} has two branches, and which one a message takes used
+ * to decide whether a close could lose it. With a publisher thread running the call is an enqueue
+ * onto {@code sendQueue}; without one it is a synchronous write of the socket. {@link
+ * Session#close()} used to drain neither: it set the flag the publisher loops on, interrupted it
+ * and joined, so anything still queued was dropped while the {@code StopMsg} published afterwards
+ * still went out - leaving the peer with an orderly close and no sign that something was lost.
+ * That is the limitation PR #919 recorded, and
+ * {@code aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed} is what holds the
+ * close to sending that queue instead. The cases after it pin what a close gives up on: a
+ * queue whose write fails is reported with every message it held; a session which had already
+ * failed is written nothing more and still has its queue reported, together with what its
+ * publisher took and failed to write, and nothing when there is nothing; and a closed session -
+ * one closed before it was started included - fails a later {@code publish()} rather than take
+ * it into a queue nothing sends.
+ * <p>
+ * The first two cases pin which end could ever pay it. Only {@code ServerHandler} starts a session's
+ * publisher, so it is the replication-server end of a session which has one; the broker of a
+ * directory server never starts its own. A change a directory server publishes is therefore on
+ * the wire by the time {@code publish()} returns, and no close of that session could drop it -
+ * which is what rules the send queue out as the explanation of #963.
+ *
+ * @see <a href="https://github.com/OpenIdentityPlatform/OpenDJ/issues/963">issue #963</a>
+ */
+@SuppressWarnings("javadoc")
+public class SessionPublisherDrainTest extends ReplicationTestCase
+{
+ private static final int DS_ID = 123;
+ private static final int RS_ID = 104;
+ private static final int SOCKET_TIMEOUT_MS = 5000;
+
+ /**
+ * The number of messages the queued-message test publishes. It has to be enough for
+ * {@code publish()} to outrun the publisher thread and leave a backlog behind it - the socket
+ * buffers of a loopback pair swallow all of it, so it is not a full buffer which leaves one -
+ * and stay under the 4000 the send queue holds, past which {@code publish()} would block
+ * instead of queueing.
+ */
+ private static final int MESSAGES_PUBLISHED = 3000;
+
+ /**
+ * The number of messages the case of a queue which cannot be sent leaves on the sender. It is
+ * small and exact: what that case is about is the count the close reports, not a backlog.
+ */
+ private static final int MESSAGES_LEFT_UNSENT = 7;
+
+ /** What the line a close writes about a queue it could not send is recognised by. */
+ private static final String NOT_SENT_REPORT = "was closed with";
+
+ /**
+ * The session a directory server publishes its changes on has no publisher thread: nothing
+ * calls {@link Session#start()} on it, so the thread is still {@code NEW} once the broker is
+ * connected and has completed its handshake.
+ */
+ @Test
+ public void theSessionOfADirectoryServerBrokerHasNoPublisherThread() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ try
+ {
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
+ replicationPort, "sessionPublisherDrainDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
+ broker = openReplicationSession(
+ baseDN, DS_ID, 100, replicationPort, SOCKET_TIMEOUT_MS, EMPTY_DN_GENID);
+
+ final Session session = sessionOf(broker);
+ assertThat(session)
+ .as("the broker reported itself connected without a session")
+ .isNotNull();
+ assertThat(session.isAlive())
+ .as("the publisher thread of the session a directory server publishes on is running, "
+ + "so publish() enqueues and a close of that session can drop what is queued")
+ .isFalse();
+ assertThat(session.getState())
+ .as("the publisher thread of a broker session was started at some point")
+ .isEqualTo(Thread.State.NEW);
+ }
+ finally
+ {
+ stop(broker);
+ remove(replicationServer);
+ }
+ }
+
+ /**
+ * With no publisher thread, {@code publish()} has written the message to the socket by the
+ * time it returns: an immediate close cannot drop it, and the peer reads it after the close.
+ */
+ @Test
+ public void aSessionWithNoPublisherThreadHasReachedTheWireWhenPublishReturns() throws Exception
+ {
+ final CSNGenerator csns = new CSNGenerator(DS_ID, 0);
+ final CSN csn = csns.newCSN();
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ sender.publish(new DeleteMsg(DN.valueOf("uid=onthewire," + TEST_ROOT_DN_STRING),
+ csn, "00000000-0000-0000-0000-000000000000"));
+ /*
+ * No drain, no flush and no wait in between - the close of the restore path of #963 is
+ * this close. What publish() already wrote stays readable by the peer: the session
+ * closes, it does not unsend the bytes ahead of it.
+ */
+ sender.close();
+
+ final ReplicationMsg received = receiver.receive();
+ assertThat(received)
+ .as("the peer did not receive the change published just before the session was closed")
+ .isInstanceOf(DeleteMsg.class);
+ assertThat(((DeleteMsg) received).getCSN()).isEqualTo(csn);
+ /*
+ * What makes this the synchronous branch rather than a race won by a publisher thread:
+ * there was no publisher thread to win it. The message reached the peer and the session's
+ * own thread never ran, so publish() is what wrote it.
+ */
+ assertThat(sender.getState())
+ .as("the session had a publisher thread after all, so the delivery above says only "
+ + "that it outran the close, not that publish() wrote the message itself")
+ .isEqualTo(Thread.State.NEW);
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ }
+
+ /**
+ * With a publisher thread running, a close sends what that thread had not sent yet rather than
+ * dropping it. This is the replication-server end of a session - the end {@code ServerHandler}
+ * starts - and the limitation PR #919 recorded.
+ * <p>
+ * The peer reads the changes and then the {@code StopMsg}, which is what {@link #drain(Session)}
+ * stops on: a queue sent after that message would not be counted, so the size below pins the
+ * order as well as the delivery.
+ */
+ @Test
+ public void aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed() throws Exception
+ {
+ final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ sender.start();
+ sender.waitForStartup();
+
+ /*
+ * A reader on the *sending* end, which is what the server keeps running across a close:
+ * ServerHandler.shutdown() closes the session (ServerHandler.java:946) and only joins its
+ * ServerReader afterwards (:966). It is not decoration here.
+ *
+ * Without it this end reaches close() with inbound bytes nobody ever read, and a close in
+ * that state ends the connection with a reset instead of a FIN - which discards what the
+ * peer has not read yet, the whole of what the drain just wrote included. Measured with a
+ * bare socket pair, 8 MiB written to a peer reading behind the writer, the only difference
+ * between the runs being whether the closing side drained its own inbound:
+ *
+ * linux 6.12 / jdk 11 no reader -> 53.4% arrived, "Connection reset"
+ * reader -> 100% arrived, end of stream
+ * macos 15.7 / jdk 26 no reader -> 98.3% arrived, "Connection reset"
+ * reader -> 100% arrived, end of stream
+ *
+ * Which is why a case without it measures the teardown rather than the drain, and measures
+ * it differently per platform: three ubuntu legs of CI lost between 866 and 1694 of these
+ * messages while every macos and windows leg passed. This case says so itself - with the
+ * start() below commented out and the class run on linux/jdk11, it fails with "the peer
+ * received 2873 of the 3000 messages published; the read ended by java.net.SocketException:
+ * Connection reset", and passes with it in.
+ *
+ * Its soTimeout has to go, too: receive() hands a read timeout to setSessionError(), and a
+ * session carrying an error skips the drain exactly as it skips the StopMsg.
+ */
+ sender.setSoTimeout(0);
+ final Thread senderReader = newInboundReader(sender);
+ senderReader.start();
+
+ /*
+ * Nothing reads the peer end while these are published, but that is not what leaves the
+ * backlog: 3000 frames of ~150 B are ~440 KB, which the socket buffers of a loopback pair
+ * swallow, so the publisher thread is not blocked inside a write. What leaves the backlog
+ * is publish() - encode and offer - outrunning the publisher, which writes and flushes one
+ * frame at a time through the TLS layer. That is a race rather than a state, so the case
+ * asserts below that it was still won when the close ran: with an empty queue there is
+ * nothing for the close to drain and this pins nothing.
+ */
+ for (int i = 0; i < MESSAGES_PUBLISHED; i++)
+ {
+ sender.publish(new DeleteMsg(DN.valueOf("uid=queued" + i + "," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
+ }
+
+ final int queuedAtClose = sendQueueOf(sender).size();
+ assertThat(queuedAtClose)
+ .as("the publisher had sent everything before the close ran, so this case drained "
+ + "nothing: what it measures is the socket rather than close()")
+ .isGreaterThan(0);
+
+ final Future<?> closed = executor.submit(new Callable<Void>()
+ {
+ @Override
+ public Void call()
+ {
+ sender.close();
+ return null;
+ }
+ });
+
+ final Drained drained = drain(receiver);
+ closed.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+
+ assertThat(drained.received)
+ .as("the peer received %d of the %d messages published; the read ended by %s",
+ drained.received.size(), MESSAGES_PUBLISHED, drained.endedBy)
+ .hasSize(MESSAGES_PUBLISHED);
+ assertThat(drained.endedBy)
+ .as("the StopMsg is what ended the stream, which is what puts the drained queue "
+ + "ahead of it rather than after it")
+ .isEqualTo("a StopMsg");
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ finally
+ {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * A close which cannot write the queue reports what the peer was not told, once, and counts the
+ * message it was writing when the write failed.
+ * <p>
+ * The road is a peer which is gone by the time the close reaches the queue - a directory server
+ * whose {@code StopMsg} brought the {@code ServerReader} of its handler to the {@code close()}
+ * of its finally, with the publisher of that session still holding a backlog. Here the sockets
+ * of the sender are closed under it instead, which is the same failed write with an exact
+ * count: the first {@code send()} of the drain throws, so the report has to name every message
+ * the queue held. A peer closed from the outside gives up somewhere inside the TCP buffers
+ * instead, and would pin no number at all.
+ * <p>
+ * The queue is filled through the field rather than by publishing on a started session for the
+ * same reason: what a publisher thread has left behind is a race, and this case is the count.
+ */
+ @Test
+ public void aCloseWhichCannotSendTheQueueReportsEveryMessageTheQueueHeld() throws Exception
+ {
+ final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ final Queue<byte[]> sendQueue = sendQueueOf(sender);
+ for (int i = 0; i < MESSAGES_LEFT_UNSENT; i++)
+ {
+ sendQueue.add(new DeleteMsg(DN.valueOf("uid=unsent" + i + "," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "00000000-0000-0000-0000-000000000000")
+ .getBytes(sender.getProtocolVersion()));
+ }
+ closeTheSocketsUnder(sender);
+
+ final List<String> records = errorLogRecordsOf(new Callable<Void>()
+ {
+ @Override
+ public Void call()
+ {
+ sender.close();
+ return null;
+ }
+ });
+
+ // Two give-ups are what a drain which does not stop at the first failed write reports.
+ final Set<String> reported = reportsIn(records);
+ assertThat(reported)
+ .as("a close which could not write the queue reports that once, and here it reported: "
+ + reported)
+ .hasSize(1);
+ assertThat(reported.iterator().next())
+ .as("the report has to account for the message the failed write took out of the queue "
+ + "as well as for the ones left in it")
+ .contains(MESSAGES_LEFT_UNSENT + " message(s)")
+ .contains("the write failed with");
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ }
+
+ /**
+ * A close of a session which has already failed writes nothing more to it - neither the queue
+ * nor the {@code StopMsg} - and still reports the queue it gives up on, once.
+ * <p>
+ * The error is set through the field, with the sockets left intact: a close which wrote the
+ * queue regardless would then get it through, and the peer reading a change is what says so.
+ * With the sockets closed under it, as in the case above, such a close would fail at its first
+ * write and look the same from the peer.
+ */
+ @Test
+ public void aCloseOfAFailedSessionWritesNothingAndReportsTheQueue() throws Exception
+ {
+ final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ final Queue<byte[]> sendQueue = sendQueueOf(sender);
+ for (int i = 0; i < MESSAGES_LEFT_UNSENT; i++)
+ {
+ sendQueue.add(new DeleteMsg(DN.valueOf("uid=failed" + i + "," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "00000000-0000-0000-0000-000000000000")
+ .getBytes(sender.getProtocolVersion()));
+ }
+ final Field sessionError = Session.class.getDeclaredField("sessionError");
+ sessionError.setAccessible(true);
+ sessionError.set(sender, new java.io.IOException("injected"));
+
+ final List<String> records = errorLogRecordsOf(new Callable<Void>()
+ {
+ @Override
+ public Void call()
+ {
+ sender.close();
+ return null;
+ }
+ });
+
+ ReplicationMsg read = null;
+ try
+ {
+ read = receiver.receive();
+ }
+ catch (final java.io.IOException expected)
+ {
+ // The socket was closed with nothing written to it, which is what this case expects.
+ }
+ assertThat(read)
+ .as("the close wrote to a session which had already failed")
+ .isNull();
+
+ final Set<String> reported = reportsIn(records);
+ assertThat(reported)
+ .as("the close of a failed session holding a queue reports that queue once, and here "
+ + "it reported: " + reported)
+ .hasSize(1);
+ assertThat(reported.iterator().next())
+ .contains(MESSAGES_LEFT_UNSENT + " message(s)")
+ .contains("the session had already failed")
+ .contains("injected");
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ }
+
+ /**
+ * A close of a failed session which has nothing it could not send says nothing: the report is
+ * for messages the peer was not told about, and there are none.
+ */
+ @Test
+ public void aCloseOfAFailedSessionWithNothingLeftToSendReportsNothing() throws Exception
+ {
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ final Field sessionError = Session.class.getDeclaredField("sessionError");
+ sessionError.setAccessible(true);
+ sessionError.set(sender, new java.io.IOException("injected"));
+
+ final List<String> records = errorLogRecordsOf(new Callable<Void>()
+ {
+ @Override
+ public Void call()
+ {
+ sender.close();
+ return null;
+ }
+ });
+
+ assertThat(reportsIn(records))
+ .as("the close of a failed session with an empty queue reported a loss")
+ .isEmpty();
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ }
+
+ /**
+ * On a started session whose writes fail, the publisher thread goes on taking the queue and
+ * failing each write until the close: what it took is lost as much as what it left, and the
+ * close reports both. Once closed, the session is off the queueing branch, so a later
+ * {@code publish()} - a {@code ServerWriter} or a {@code HeartbeatThread}, which end only on
+ * that exception - fails instead of returning as if the message had been queued.
+ * <p>
+ * The count is exact whatever the thread got through before the close: every message published
+ * is either still in the queue or was taken and failed. The case waits for the queue to empty so
+ * that it is the publisher's own count the report stands on - the queue alone would report
+ * nothing.
+ */
+ @Test
+ public void aCloseOfAStartedSessionWhoseWritesFailedReportsWhatThePublisherTookAsWell()
+ throws Exception
+ {
+ final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ sender.start();
+ sender.waitForStartup();
+ closeTheSocketsUnder(sender);
+ for (int i = 0; i < MESSAGES_LEFT_UNSENT; i++)
+ {
+ sender.publish(new DeleteMsg(DN.valueOf("uid=takenandlost" + i + ","
+ + TEST_ROOT_DN_STRING), csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
+ }
+ final Queue<byte[]> sendQueue = sendQueueOf(sender);
+ final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SOCKET_TIMEOUT_MS);
+ while (!sendQueue.isEmpty() && System.nanoTime() - deadline < 0)
+ {
+ Thread.sleep(10);
+ }
+ assertThat(sendQueue)
+ .as("the publisher thread did not take the queue off a session whose writes fail")
+ .isEmpty();
+
+ final List<String> records = errorLogRecordsOf(new Callable<Void>()
+ {
+ @Override
+ public Void call()
+ {
+ sender.close();
+ return null;
+ }
+ });
+
+ final Set<String> reported = reportsIn(records);
+ assertThat(reported)
+ .as("the close reports once what the publisher took and could not write, and here "
+ + "it reported: " + reported)
+ .hasSize(1);
+ assertThat(reported.iterator().next())
+ .contains(MESSAGES_LEFT_UNSENT + " message(s)")
+ .contains("the session had already failed");
+
+ try
+ {
+ sender.publish(new DeleteMsg(DN.valueOf("uid=afterclose," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
+ org.assertj.core.api.Assertions.fail(
+ "a publish() on a closed session returned as if it had queued the message");
+ }
+ catch (final java.io.IOException expected)
+ {
+ // The close put the session on the synchronous branch, which fails on the closed socket.
+ }
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ }
+
+ /**
+ * A session closed before its publisher thread is started - which a handler whose session is
+ * closed by a stop of all servers during its handshake does, before it goes on to start it -
+ * stays on the synchronous branch of {@code publish()}. Put on the queueing one, a publish
+ * would return at once, as if queued, onto a queue nothing sends, and a {@code HeartbeatThread}
+ * started after it would go on publishing into it until the server stops.
+ */
+ @Test
+ public void aSessionClosedBeforeItIsStartedKeepsFailingPublishes() throws Exception
+ {
+ final CSNGenerator csns = new CSNGenerator(RS_ID, 0);
+ try (ServerSocket listen = new ServerSocket(0))
+ {
+ final Session[] pair = connectSessionPair(listen);
+ final Session sender = pair[0];
+ final Session receiver = pair[1];
+ try
+ {
+ sender.close();
+ sender.start();
+ sender.waitForStartup();
+ sender.join(SOCKET_TIMEOUT_MS);
+
+ try
+ {
+ sender.publish(new DeleteMsg(DN.valueOf("uid=afterclose," + TEST_ROOT_DN_STRING),
+ csns.newCSN(), "00000000-0000-0000-0000-000000000000"));
+ org.assertj.core.api.Assertions.fail(
+ "a publish() on a session closed before it was started returned as if it had "
+ + "queued the message");
+ }
+ catch (final java.io.IOException expected)
+ {
+ // The synchronous branch, failing on the closed socket.
+ }
+ }
+ finally
+ {
+ StaticUtils.close(sender, receiver);
+ }
+ }
+ }
+
+ /**
+ * The reports of a queue not sent among the records, from the text of the report on, so that
+ * the severity and the timestamp a record carries do not make one report captured twice look
+ * like two - and so that two reports still do.
+ */
+ private static Set<String> reportsIn(final List<String> records)
+ {
+ final Set<String> reported = new LinkedHashSet<>();
+ for (final String record : records)
+ {
+ final int start = record.indexOf(NOT_SENT_REPORT);
+ if (start >= 0)
+ {
+ reported.add(record.substring(start));
+ }
+ }
+ return reported;
+ }
+
+ /**
+ * Consumes whatever arrives on a session until it is closed, as {@code ServerReader} does for a
+ * server handler.
+ * <p>
+ * It is what it reads at the socket rather than what it returns that matters: the peer of these
+ * cases publishes nothing, so this returns no message at all, while the read it sits in is what
+ * keeps the receive queue of this end empty - which is the condition a close needs to end the
+ * connection in an orderly way.
+ */
+ private Thread newInboundReader(final Session session)
+ {
+ final Thread reader = new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ while (true)
+ {
+ session.receive();
+ }
+ }
+ catch (final Exception ignored)
+ {
+ // The close of the session ends the read, which is the end of this thread.
+ }
+ }
+ }, "inbound reader of " + session.getName());
+ reader.setDaemon(true);
+ return reader;
+ }
+
+ /**
+ * Reads until the session gives nothing back, and answers the CSNs of the changes it read
+ * together with what ended the read.
+ * <p>
+ * Why the reason is carried rather than swallowed: a short read is exactly the failure this
+ * suite is about, and "the peer received fewer than were sent" does not say whether the stream
+ * ended orderly at a {@code StopMsg} or was cut off - which are different defects.
+ */
+ private static final class Drained
+ {
+ private final List<CSN> received = new CopyOnWriteArrayList<>();
+ private String endedBy;
+
+ @Override
+ public String toString()
+ {
+ return received.size() + " change(s), ended by " + endedBy;
+ }
+ }
+
+ private Drained drain(final Session session)
+ {
+ final Drained drained = new Drained();
+ try
+ {
+ while (true)
+ {
+ final ReplicationMsg msg = session.receive();
+ if (msg instanceof DeleteMsg)
+ {
+ drained.received.add(((DeleteMsg) msg).getCSN());
+ }
+ else if (msg instanceof StopMsg)
+ {
+ drained.endedBy = "a StopMsg";
+ return drained;
+ }
+ else
+ {
+ drained.endedBy = "an unexpected " + msg.getClass().getSimpleName();
+ return drained;
+ }
+ }
+ }
+ catch (final Exception e)
+ {
+ drained.endedBy = e.getClass().getName() + ": " + e.getMessage();
+ return drained;
+ }
+ }
+
+ /** The queue a started session's publisher thread takes its buffers from. */
+ @SuppressWarnings("unchecked")
+ private static Queue<byte[]> sendQueueOf(final Session session) throws Exception
+ {
+ final Field sendQueue = Session.class.getDeclaredField("sendQueue");
+ sendQueue.setAccessible(true);
+ return (Queue<byte[]>) sendQueue.get(session);
+ }
+
+ /**
+ * Closes the sockets a session writes through while leaving the session unaware of it, so that
+ * its next write fails - the state a peer which has gone away leaves it in.
+ */
+ private static void closeTheSocketsUnder(final Session session) throws Exception
+ {
+ for (final String name : new String[] { "secureSocket", "plainSocket" })
+ {
+ final Field socket = Session.class.getDeclaredField(name);
+ socket.setAccessible(true);
+ StaticUtils.close((Closeable) socket.get(session));
+ }
+ }
+
+ /** The session a broker publishes on, which it keeps to itself. */
+ private Session sessionOf(final ReplicationBroker broker) throws Exception
+ {
+ final Field connectedRSField = ReplicationBroker.class.getDeclaredField("connectedRS");
+ connectedRSField.setAccessible(true);
+ final Object connectedRS = ((AtomicReference<?>) connectedRSField.get(broker)).get();
+ final Field sessionField = connectedRS.getClass().getDeclaredField("session");
+ sessionField.setAccessible(true);
+ return (Session) sessionField.get(connectedRS);
+ }
+
+ /**
+ * A connected pair of sessions over the loopback, the client end first. Neither end is
+ * started: a session which publishes synchronously is what a broker has, and the test which
+ * needs a publisher thread starts the end it needs.
+ */
+ 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);
+ return new Session[] { clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd };
+ }
+ finally
+ {
+ executor.shutdown();
+ }
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java
index c8e503c..31eabf6 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java
@@ -726,9 +726,13 @@
.isNotNull();
/*
* The forward asserted above proves the message reached the Session, not the wire: close()
- * discards whatever is still in its send queue without draining it, which is the
- * limitation issue #919 recorded. If this is the only assertion which fails, that window
- * is the explanation rather than the granularity of the barrier.
+ * now sends what its publisher left queued, but only within its own budget, and it writes
+ * that queue under publishLock so that a message published meanwhile lands after it rather
+ * than between two of its own. If this is the only assertion which fails, the close is
+ * where to look before the granularity of the barrier: the warning close() writes for a
+ * queue it could not hand over says why it gave that queue up, and its absence does not
+ * prove the message left this end - a publish() concurrent with the close is still dropped
+ * at the door without one.
*/
assertThat(receivedWhenHeldBack.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
.as("the peer which was held back never learned that the replica went offline, "
--
Gitblit v1.10.0