From b72eafefd64e249f2feb2bd87c84e565781c9c47 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Sat, 12 Sep 2026 12:34:09 +0300 Subject: [PATCH 1/5] Send what a replication session's publisher left queued when it is closed Session.close() set closeInitiated, interrupted the publisher thread and joined it, and everything still in sendQueue went with it. The StopMsg published afterwards still went out, so the peer read an orderly close with no sign that anything was missing. PR #919 recorded this as a known limitation and named the fix; this takes the first of the two options it listed. The queue is now sent from close(), after the join - the publisher is gone, so the closing thread owns the socket - and before the StopMsg, so that message stays last on the wire. A session which already failed is left alone, for the reason the StopMsg is. The drain is bounded by DRAIN_BUDGET_MS, 5 s, the value DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD already spends waiting for one of these messages to be forwarded: close() is on shutdown paths, and an unbounded drain would hold the thread shutting the server down for as long as a stalled consumer stays stalled. What it gives up on is logged rather than dropped in silence. SessionPublisherDrainTest also pins which end of a session could ever lose a queued message: only ServerHandler starts a session's publisher, so a change a directory server publishes is written to the socket before publish() returns. That rules the send queue out as the explanation of #963. Refs #963 --- .../server/replication/protocol/Session.java | 89 ++++++ .../protocol/SessionPublisherDrainTest.java | 297 ++++++++++++++++++ 2 files changed, 386 insertions(+) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java 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 596521927a..12ead4eaf6 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; @@ -36,6 +37,7 @@ 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 +50,17 @@ public final class Session extends DirectoryThread implements Closeable { private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); + /** + * How long a close spends sending what the publisher thread left queued in {@code sendQueue}, + * in milliseconds. + *

+ * 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. + */ + private static final long DRAIN_BUDGET_MS = 5000; + private final Socket plainSocket; private final SSLSocket secureSocket; private final InputStream plainInput; @@ -140,6 +153,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() @@ -186,6 +203,22 @@ public void close() } } + /* + * The publisher thread has stopped, so this thread is the only one left writing this socket - + * which is what lets the StopMsg below be published on it - and what that thread 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. + * + * Skipped on a session which already failed, for the reason the StopMsg is: writing more to it + * cannot work. That is also the path where close() runs on the publisher thread itself + * (run() calls it when a send threw), where the queue is unsendable by construction. + */ + if (localSessionError == null) + { + sendWhatThePublisherLeftQueued(); + } + // V4 protocol introduces a StopMsg to properly end communications. if (localSessionError == null && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4) @@ -205,6 +238,62 @@ public void close() + /** + * 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, so the socket is this + * thread's alone. 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 budget is what bounds a close of a session whose peer has stopped reading: without one, + * a stalled consumer would hold the thread which is shutting the server down for as long as it + * stays stalled. A peer which is reading pays none of it, and a peer which is gone pays none + * either - the write fails at once. It is checked between messages, so a single write which + * blocks past the budget still runs to completion: bounding that needs a non-blocking socket, + * which this session is not. 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() + { + if (sendQueue.isEmpty()) + { + return; + } + + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS); + byte[] buffer; + while ((buffer = sendQueue.poll()) != null) + { + if (System.nanoTime() - deadline >= 0) + { + reportQueueNotSent(sendQueue.size() + 1, "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. + reportQueueNotSent(sendQueue.size() + 1, stackTraceToSingleLineString(e)); + return; + } + } + } + + /** 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. 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..4234cc164d --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionPublisherDrainTest.java @@ -0,0 +1,297 @@ +/* + * 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.lang.reflect.Field; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.List; +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 the last test here is what holds the close to + * sending that queue instead. + *

+ * The other two 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 outrun what the socket + * buffers of a loopback pair can swallow while nothing reads them, 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 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())); + 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: a close sends a + * FIN, 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. + *

+ * 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(); + + /* + * Nothing reads the peer end while these are published, so the socket buffers fill and + * the publisher thread is left inside its write with the rest of them still queued. + */ + 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 Future closed = executor.submit(new Callable() + { + @Override + public Void call() + { + sender.close(); + return null; + } + }); + + final List received = drain(receiver); + closed.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + + assertThat(received) + .as("the close dropped %d of the %d messages published: the publisher thread stopped " + + "with them still queued and nothing sent them", + MESSAGES_PUBLISHED - received.size(), MESSAGES_PUBLISHED) + .hasSize(MESSAGES_PUBLISHED); + } + finally + { + StaticUtils.close(sender, receiver); + } + } + finally + { + executor.shutdownNow(); + } + } + + /** Reads until the session gives nothing back, and answers the CSNs of the changes it read. */ + private List drain(final Session session) + { + final List received = new CopyOnWriteArrayList<>(); + try + { + while (true) + { + final ReplicationMsg msg = session.receive(); + if (msg instanceof DeleteMsg) + { + received.add(((DeleteMsg) msg).getCSN()); + } + else if (msg instanceof StopMsg) + { + return received; + } + } + } + catch (final Exception ignored) + { + // The close of the far end ends the read, which is the end of the drain. + return received; + } + } + + /** 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 clientEnd = executor.submit(new Callable() + { + @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(); + } + } +} From 983eac9c1be86fcbeccb3f7f57ec3d7dd3fbb448 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Sat, 12 Sep 2026 16:36:08 +0300 Subject: [PATCH 2/5] Keep a reader on the closing side of the drain case, as the server does The case failed on three ubuntu legs of CI and on nothing else: the peer had received 2025, 2134 and 1306 of the 3000 messages. The give-up warning appeared in none of those logs, so the drain had written the whole queue and the loss was under the write. The case had no reader on the closing side, so that side reached close() with inbound bytes nobody had read, and a close in that state ends the connection with a reset rather than a FIN - which discards what the peer has not read yet, the drained queue included. Measured with a bare socket pair, 8 MiB to a peer reading behind the writer, the only difference being whether the closing side drained its own inbound: linux 6.12/jdk11 53.4% against 100%, macos 15.7/jdk26 98.3% against 100%. That spread is why every macos and windows leg passed. The server has no such condition where the drain does anything: ServerHandler.shutdown() closes the session at :946 and joins its ServerReader only at :966, and the paths which close without a live reader are the ones the drain skips anyway - Session.run() after a send threw, and ServerReader's finally on an error, both of which leave sessionError set. So the case keeps a reader too, with its soTimeout lifted, receive() handing a read timeout to setSessionError() being enough to skip the drain and test nothing. The read of the peer end now reports what ended it. Run on linux/jdk11 with that reader commented out, the case 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. Refs #963 --- .../protocol/SessionPublisherDrainTest.java | 113 ++++++++++++++++-- 1 file changed, 100 insertions(+), 13 deletions(-) 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 index 4234cc164d..c8cc22eb65 100644 --- 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 @@ -186,6 +186,36 @@ public void aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed() t 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, so the socket buffers fill and * the publisher thread is left inside its write with the rest of them still queued. @@ -206,13 +236,12 @@ public Void call() } }); - final List received = drain(receiver); + final Drained drained = drain(receiver); closed.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); - assertThat(received) - .as("the close dropped %d of the %d messages published: the publisher thread stopped " - + "with them still queued and nothing sent them", - MESSAGES_PUBLISHED - received.size(), MESSAGES_PUBLISHED) + 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); } finally @@ -226,10 +255,62 @@ public Void call() } } - /** Reads until the session gives nothing back, and answers the CSNs of the changes it read. */ - private List drain(final Session session) + /** + * Consumes whatever arrives on a session until it is closed, as {@code ServerReader} does for a + * server handler. + *

+ * 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 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 List received = new CopyOnWriteArrayList<>(); + final Drained drained = new Drained(); try { while (true) @@ -237,18 +318,24 @@ private List drain(final Session session) final ReplicationMsg msg = session.receive(); if (msg instanceof DeleteMsg) { - received.add(((DeleteMsg) msg).getCSN()); + drained.received.add(((DeleteMsg) msg).getCSN()); } else if (msg instanceof StopMsg) { - return received; + drained.endedBy = "a StopMsg"; + return drained; + } + else + { + drained.endedBy = "an unexpected " + msg.getClass().getSimpleName(); + return drained; } } } - catch (final Exception ignored) + catch (final Exception e) { - // The close of the far end ends the read, which is the end of the drain. - return received; + drained.endedBy = e.getClass().getName() + ": " + e.getMessage(); + return drained; } } From 8d824b809af44bc13f253be421ff2103d8f732eb Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 23 Sep 2026 11:05:46 +0300 Subject: [PATCH 3/5] Write the queue a close drains under publishLock, and pin what it gives up on The drain was not the only writer of the socket it drains: a ServerWriter is joined only after the close which reaches it, a HeartbeatThread is shut down after it, and their publish() takes the synchronous branch once the publisher has stopped - so a newer message could be written between two older drained ones, which a peer replication server answers by dropping the older ones at debug level. The queue now goes out under publishLock, and the comments say that rather than claiming a single writer. The error the drain and the StopMsg are skipped on is re-read after the join, so a publisher whose write failed while it was being joined is not answered with the snapshot taken before it. The give-up on a failed write names the exception instead of tracing it: that arm is reached whenever a peer which announced it is leaving closes first, where the stack trace says nothing. The javadoc of the budget says what holds - a peer which is reading slowly pays it, a single blocked write still runs to completion - and that on the road which does not shut the server down the wait is paid under the lock of the replication domain. A fourth case pins the give-up: a queue which cannot be written reports once and counts the message the failed write took out of it. The drain case now asserts the backlog it needs to pin anything, and that the StopMsg is what ended the stream; its comment says the producer outruns the publisher rather than that the socket buffers fill. --- .../server/replication/protocol/Session.java | 105 +++++++++----- .../protocol/SessionPublisherDrainTest.java | 131 +++++++++++++++++- .../ReplicationServerShutdownSyncTest.java | 9 +- 3 files changed, 209 insertions(+), 36 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 12ead4eaf6..7fc39d59cc 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 @@ -182,6 +182,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()) { @@ -204,11 +216,14 @@ public void close() } /* - * The publisher thread has stopped, so this thread is the only one left writing this socket - - * which is what lets the StopMsg below be published on it - and what that thread 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. + * 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 and its publish() now takes the synchronous branch - + * so the drain holds publishLock across the whole queue - see + * sendWhatThePublisherLeftQueued() below. * * Skipped on a session which already failed, for the reason the StopMsg is: writing more to it * cannot work. That is also the path where close() runs on the publisher thread itself @@ -242,18 +257,31 @@ public void close() * 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, so the socket is this - * thread's alone. 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. + * Called from {@link #close()} once the publisher has been joined. 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 budget is what bounds a close of a session whose peer has stopped reading: without one, - * a stalled consumer would hold the thread which is shutting the server down for as long as it - * stays stalled. A peer which is reading pays none of it, and a peer which is gone pays none - * either - the write fails at once. It is checked between messages, so a single write which - * blocks past the budget still runs to completion: bounding that needs a non-blocking socket, - * which this session is not. What is given up on is reported rather than dropped in silence, - * that being the part of this which cost the most to diagnose. + * 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 publisher has + * stopped, 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. Holding the lock makes + * them wait for the drain and land after it. 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. 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() { @@ -264,25 +292,40 @@ private void sendWhatThePublisherLeftQueued() final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS); byte[] buffer; - while ((buffer = sendQueue.poll()) != null) + publishLock.lock(); + try { - if (System.nanoTime() - deadline >= 0) - { - reportQueueNotSent(sendQueue.size() + 1, "the peer did not read them within " - + DRAIN_BUDGET_MS + " ms"); - return; - } - try + while ((buffer = sendQueue.poll()) != null) { - send(buffer); - } - catch (final IOException e) - { - // send() has recorded the error; the rest of the queue cannot go out either. - reportQueueNotSent(sendQueue.size() + 1, stackTraceToSingleLineString(e)); - return; + if (System.nanoTime() - deadline >= 0) + { + reportQueueNotSent(sendQueue.size() + 1, "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(sendQueue.size() + 1, + "the write failed with " + e.getClass().getName() + ": " + e.getMessage()); + return; + } } } + finally + { + publishLock.unlock(); + } } /** Says which messages a close could not hand to the peer, and why. */ 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 index c8cc22eb65..435e15f2b1 100644 --- 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 @@ -18,10 +18,14 @@ 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; @@ -77,6 +81,15 @@ public class SessionPublisherDrainTest extends ReplicationTestCase */ 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 @@ -217,8 +230,13 @@ public void aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed() t senderReader.start(); /* - * Nothing reads the peer end while these are published, so the socket buffers fill and - * the publisher thread is left inside its write with the rest of them still queued. + * 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++) { @@ -226,6 +244,12 @@ public void aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed() t 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() { @Override @@ -243,6 +267,10 @@ public Void call() .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 { @@ -255,6 +283,82 @@ public Void call() } } + /** + * 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. + *

+ * 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 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 records = errorLogRecordsOf(new Callable() + { + @Override + public Void call() + { + sender.close(); + return null; + } + }); + + /* + * The text from the report on, so that the severity and the timestamp a record carries do + * not make one give-up captured twice look like two - and so that two give-ups, which is + * what a drain which does not stop at the first failed write reports, still do. + */ + final Set reported = new LinkedHashSet<>(); + for (final String record : records) + { + final int start = record.indexOf(NOT_SENT_REPORT); + if (start >= 0) + { + reported.add(record.substring(start)); + } + } + 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); + } + } + } + /** * Consumes whatever arrives on a session until it is closed, as {@code ServerReader} does for a * server handler. @@ -339,6 +443,29 @@ else if (msg instanceof StopMsg) } } + /** The queue a started session's publisher thread takes its buffers from. */ + @SuppressWarnings("unchecked") + private static Queue sendQueueOf(final Session session) throws Exception + { + final Field sendQueue = Session.class.getDeclaredField("sendQueue"); + sendQueue.setAccessible(true); + return (Queue) 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 { 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 c8e503c651..7bf62c29b6 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,12 @@ public void theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline() throws E .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 the budget ran out, and its absence says the message + * left this end and the loss is past it. */ assertThat(receivedWhenHeldBack.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS)) .as("the peer which was held back never learned that the replica went offline, " From 4e60005cd89585a9185fa45a0b5f3517d1f649f7 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 23 Sep 2026 15:09:56 +0300 Subject: [PATCH 4/5] Clear a closed session's publisher flag under publishLock, and report the queue of a failed one The close, not run(), now takes the session off the queueing branch of publish(), under publishLock, and holds the lock across the drain and the StopMsg. run() cleared the flag as it ended, so a ServerWriter or HeartbeatThread publish() could take the synchronous branch and get the lock before the close reached it, writing a newer message ahead of the queue. run() still clears it on a loop which ended without a close, so that publish() does not go on queueing onto a queue nobody sends. A session which had already failed - its publisher failing while the close joined it - now has its queue reported rather than dropped without a word, and is closed without taking publishLock, so that a thread blocked in a write of that socket is released by the close of the sockets. The error is also re-read after the drain, so that a drain whose write failed is not followed by a StopMsg on the same socket. aCloseOfAFailedSessionWritesNothingAndReportsTheQueue pins the failed-session road: nothing is written, and the queue is reported once with its count. The budget javadoc says it is paid per close and a shutdown closes its sessions in turn; the test javadoc and comments no longer describe an earlier shape of the cases. --- .../server/replication/protocol/Session.java | 174 +++++++++++------- .../protocol/SessionPublisherDrainTest.java | 100 +++++++++- .../ReplicationServerShutdownSyncTest.java | 5 +- 3 files changed, 201 insertions(+), 78 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 7fc39d59cc..eb39c422b7 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 @@ -57,7 +57,9 @@ public final class Session extends DirectoryThread implements Closeable * 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. + * 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; @@ -215,38 +217,71 @@ public void close() } } + if (localSessionError != null) + { + /* + * 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. The queue is still reported - the + * publisher may have failed while this thread was joining it, with a backlog behind the + * write which failed - and it is left 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); + if (!sendQueue.isEmpty()) + { + reportQueueNotSent(sendQueue.size(), "the session had already failed: " + localSessionError); + } + 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 and its publish() now takes the synchronous branch - - * so the drain holds publishLock across the whole queue - see - * sendWhatThePublisherLeftQueued() below. - * - * Skipped on a session which already failed, for the reason the StopMsg is: writing more to it - * cannot work. That is also the path where close() runs on the publisher thread itself - * (run() calls it when a send threw), where the queue is unsendable by construction. + * 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, sees closeInitiated and returns, 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. */ - if (localSessionError == null) + publishLock.lock(); + try { + isRunning.set(false); sendWhatThePublisherLeftQueued(); - } - // V4 protocol introduces a StopMsg to properly end communications. - if (localSessionError == null - && protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4) - { - try + /* + * 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) { - publish(new StopMsg()); + localSessionError = sessionError; } - catch (final IOException ignored) + + // 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); } @@ -257,74 +292,67 @@ public void close() * 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. 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. + * 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 publisher has - * stopped, 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. Holding the lock makes - * them wait for the drain and land after it. The wait for the lock itself is not part of the - * budget below, no more than it is for the {@code StopMsg} which follows. + * 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 returns at the door, having seen the close, or waits for the drain + * and lands after it. 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. 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. + * 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() { - if (sendQueue.isEmpty()) - { - return; - } - final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS); byte[] buffer; - publishLock.lock(); - try + while ((buffer = sendQueue.poll()) != null) { - while ((buffer = sendQueue.poll()) != null) + if (System.nanoTime() - deadline >= 0) { - if (System.nanoTime() - deadline >= 0) - { - reportQueueNotSent(sendQueue.size() + 1, "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(sendQueue.size() + 1, - "the write failed with " + e.getClass().getName() + ": " + e.getMessage()); - return; - } + reportQueueNotSent(sendQueue.size() + 1, "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(sendQueue.size() + 1, + "the write failed with " + e.getClass().getName() + ": " + e.getMessage()); + return; } - } - finally - { - publishLock.unlock(); } } @@ -686,7 +714,17 @@ public void run() 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 index 435e15f2b1..91389f9c87 100644 --- 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 @@ -56,10 +56,13 @@ * 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 the last test here is what holds the close to - * sending that queue instead. + * That is the limitation PR #919 recorded, and + * {@code aSessionWithAPublisherThreadSendsWhatIsStillQueuedWhenItIsClosed} is what holds the + * close to sending that queue instead. The two cases after it pin what a close gives up on: a + * queue whose write fails is reported with every message it held, and a session which had + * already failed is written nothing more and still has its queue reported. *

- * The other two pin which end could ever pay it. Only {@code ServerHandler} starts a session's + * 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 - @@ -75,9 +78,11 @@ public class SessionPublisherDrainTest extends ReplicationTestCase private static final int SOCKET_TIMEOUT_MS = 5000; /** - * The number of messages the queued-message test publishes. It has to outrun what the socket - * buffers of a loopback pair can swallow while nothing reads them, and stay under the 4000 the - * send queue holds, past which {@code publish()} would block instead of queueing. + * 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; @@ -148,8 +153,8 @@ public void aSessionWithNoPublisherThreadHasReachedTheWireWhenPublishReturns() t 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: a close sends a - * FIN, it does not unsend the bytes ahead of it. + * 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(); @@ -359,6 +364,85 @@ public Void call() } } + /** + * 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. + *

+ * 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 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 records = errorLogRecordsOf(new Callable() + { + @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 reported = new LinkedHashSet<>(); + for (final String record : records) + { + final int start = record.indexOf(NOT_SENT_REPORT); + if (start >= 0) + { + reported.add(record.substring(start)); + } + } + 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); + } + } + } + /** * Consumes whatever arrives on a session until it is closed, as {@code ServerReader} does for a * server handler. 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 7bf62c29b6..31eabf6874 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 @@ -730,8 +730,9 @@ public void theShutdownWaitsForEveryPeerToBeToldTheReplicaWentOffline() throws E * 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 the budget ran out, and its absence says the message - * left this end and the loss is past it. + * 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, " From b35374dc61ccd8614eda7c7a3d18434eda62d147 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 23 Sep 2026 17:27:17 +0300 Subject: [PATCH 5/5] Keep a session closed before its start off the queueing branch, and report all a close gives up on Round 4 of the review of #1035. run() no longer sets isRunning on a session whose close came before its start: nothing clears the flag after such a close, and every later publish() returned as if queued - a HeartbeatThread started by a handshake whose session a stop of all servers had closed went on publishing into nothing until the server stopped. The check is under stateLock, as the close sets closeInitiated. The close of a failed session reports what the publisher thread took off the queue and failed to write as well as what is left in it: after a failed write that thread goes on taking the queue until the close, so the queue alone undercounted, and said nothing once it was empty. A close which gives up empties the queue rather than counting it. publish() checks the flag once it has queued: a buffer queued after the close took the session off the queueing branch, and still in the queue under publishLock, is one nothing will send, so it is taken back and reported. The javadoc no longer claims a two-way split the code did not guarantee. New cases pin the close before the start, the count of a started session whose writes fail together with its publish() after the close, and the silence of a failed session with nothing to report. --- .../server/replication/protocol/Session.java | 107 +++++++-- .../protocol/SessionPublisherDrainTest.java | 214 ++++++++++++++++-- 2 files changed, 280 insertions(+), 41 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 eb39c422b7..bd867c0220 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 @@ -31,6 +31,7 @@ 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; @@ -115,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. * @@ -221,16 +230,18 @@ public void close() { /* * 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. The queue is still reported - the - * publisher may have failed while this thread was joining it, with a backlog behind the - * write which failed - and it is left 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. + * 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); - if (!sendQueue.isEmpty()) + final int notSent = publisherFailedWrites.get() + takeWhatIsLeftQueued(); + if (notSent > 0) { - reportQueueNotSent(sendQueue.size(), "the session had already failed: " + localSessionError); + reportQueueNotSent(notSent, "the session had already failed: " + localSessionError); } StaticUtils.close(plainSocket, secureSocket); return; @@ -245,9 +256,9 @@ public void close() * 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, sees closeInitiated and returns, 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. + * 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 @@ -306,9 +317,12 @@ public void close() * 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 returns at the door, having seen the close, or waits for the drain - * and lands after it. The wait for the lock itself is not part of the budget below, no more - * than it is for the {@code StopMsg} which follows. + * {@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 @@ -332,7 +346,7 @@ private void sendWhatThePublisherLeftQueued() { if (System.nanoTime() - deadline >= 0) { - reportQueueNotSent(sendQueue.size() + 1, "the peer did not read them within " + reportQueueNotSent(1 + takeWhatIsLeftQueued(), "the peer did not read them within " + DRAIN_BUDGET_MS + " ms"); return; } @@ -349,13 +363,28 @@ private void sendWhatThePublisherLeftQueued() * 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(sendQueue.size() + 1, + 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) { @@ -482,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; } } @@ -498,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 @@ -686,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()) { @@ -711,6 +785,7 @@ public void run() catch (IOException e) { setSessionError(e); + publisherFailedWrites.incrementAndGet(); needClosing = true; } } 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 index 91389f9c87..915a312ee4 100644 --- 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 @@ -58,9 +58,12 @@ * 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 two cases after it pin what a close gives up on: a - * queue whose write fails is reported with every message it held, and a session which had - * already failed is written nothing more and still has its queue reported. + * 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 @@ -333,20 +336,8 @@ public Void call() } }); - /* - * The text from the report on, so that the severity and the timestamp a record carries do - * not make one give-up captured twice look like two - and so that two give-ups, which is - * what a drain which does not stop at the first failed write reports, still do. - */ - final Set reported = new LinkedHashSet<>(); - for (final String record : records) - { - final int start = record.indexOf(NOT_SENT_REPORT); - if (start >= 0) - { - reported.add(record.substring(start)); - } - } + // Two give-ups are what a drain which does not stop at the first failed write reports. + final Set reported = reportsIn(records); assertThat(reported) .as("a close which could not write the queue reports that once, and here it reported: " + reported) @@ -418,23 +409,134 @@ public Void call() .as("the close wrote to a session which had already failed") .isNull(); - final Set reported = new LinkedHashSet<>(); - for (final String record : records) + final Set 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 records = errorLogRecordsOf(new Callable() { - final int start = record.indexOf(NOT_SENT_REPORT); - if (start >= 0) + @Override + public Void call() { - reported.add(record.substring(start)); + 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. + *

+ * 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 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 records = errorLogRecordsOf(new Callable() + { + @Override + public Void call() + { + sender.close(); + return null; + } + }); + + final Set reported = reportsIn(records); assertThat(reported) - .as("the close of a failed session holding a queue reports that queue once, and here " + .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") - .contains("injected"); + .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 { @@ -443,6 +545,68 @@ public Void call() } } + /** + * 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 reportsIn(final List records) + { + final Set 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.