+ * 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; @@ -100,6 +116,14 @@ public final class Session extends DirectoryThread implements Closeable private AtomicBoolean isRunning = new AtomicBoolean(false); 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. * @@ -140,6 +164,10 @@ public Session(final Socket socket, /** * 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. + *
+ * 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 @@ public void close() 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,25 +226,174 @@ public void close() } } - // 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) + { + localSessionError = sessionError; + } + + // V4 protocol introduces a StopMsg to properly end communications. + if (localSessionError == null + && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4) { - // Ignore errors on close. + try + { + publish(new StopMsg()); + } + catch (final IOException ignored) + { + // Ignore errors on close. + } } } + finally + { + publishLock.unlock(); + } StaticUtils.close(plainSocket, secureSocket); } + /** + * Sends the buffers the publisher thread had not sent when it stopped, so that a close of the + * session does not drop them. + *
+ * 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. + *
+ * 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. + *
+ * 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 @@ public void publish(final ReplicationMsg msg) throws IOException // 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 @@ public void publish(final ReplicationMsg msg) throws IOException } } + /** + * Takes a buffer back out of the queue if nothing is going to send it, and reports it. + *
+ * {@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 @@ private void setSessionError(final Exception e) @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 @@ public void run() 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 0000000000..915a312ee4 --- /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. + *
+ * {@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. + *
+ * 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 issue #963
+ */
+@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
+ * 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
+ * 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.
+ *
+ * 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
+ * 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
+ * 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
+ * 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.
+ *
+ * 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