From c57788143560a0d581eb1a1f94faaa3e34f37fbf Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 16 Sep 2026 13:36:19 +0300 Subject: [PATCH] [#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued ServerWriter reported the forward as soon as Session.publish() had handed the message to the send queue of the session, and Session.close() drops that queue without draining it: a session thread busy writing an earlier buffer when the message was queued let the shutdown, released by that report, close the session with the message still queued, and the peer received the StopMsg alone. Session.publish(msg, whenWritten) runs the callback once, on the thread which wrote the message, after the write returned, and never for a message which was not written; it returns false for a message it neither wrote nor queued. ServerWriter reports the forward from that callback, and gives the peer up when the session refuses the message. --- .../server/replication/protocol/Session.java | 84 ++- .../replication/server/ServerWriter.java | 51 +- .../replication/protocol/SessionTest.java | 245 +++++++++ .../ReplicationServerShutdownSyncTest.java | 478 ++++++++++++++++-- 4 files changed, 797 insertions(+), 61 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.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..ae318b30f0 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; @@ -96,7 +98,20 @@ public final class Session extends DirectoryThread implements Closeable */ private BufferedOutputStream output; - private final LinkedBlockingQueue sendQueue = new LinkedBlockingQueue<>(4000); + /** A message queued for the thread of this session, and what to run once it is written. */ + private static final class Outgoing + { + private final byte[] buffer; + private final Runnable whenWritten; + + private Outgoing(byte[] buffer, Runnable whenWritten) + { + this.buffer = buffer; + this.whenWritten = whenWritten; + } + } + + private final LinkedBlockingQueue sendQueue = new LinkedBlockingQueue<>(4000); private AtomicBoolean isRunning = new AtomicBoolean(false); private final CountDownLatch latch = new CountDownLatch(1); @@ -140,6 +155,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. + *

+ * The thread of this session is stopped where it is: whatever is still queued for it is not + * written, and the callbacks of those messages never run - see + * {@link #publish(ReplicationMsg, Runnable)}. */ @Override public void close() @@ -306,23 +325,51 @@ public boolean isEncrypted() * If an IO error occurred. */ public void publish(final ReplicationMsg msg) throws IOException + { + publish(msg, null); + } + + /** + * Sends a replication message to the remote peer, and runs the provided callback once the + * message has been written to the socket. + *

+ * While the thread of this session runs, a message published is queued for it and written + * later, so the return of this method says only that the message is queued. The callback is + * the only word that the message has left this server: it runs once, on the thread which wrote + * the message, after the write returned - and never for a message which was not written, which + * is what becomes of a message the write of which fails, and of everything still queued when + * the session is closed. It must be short and must not block: the session writes nothing else + * until it returns. + * + * @param msg + * The message to be sent. + * @param whenWritten + * What to run once the message has been written, or null. + * @return whether the message was written or queued to be written; false when it was neither, + * because it has no encoding for the protocol version of the peer or because the + * session is being closed - the callback then never runs. + * @throws IOException + * If an IO error occurred. + */ + public boolean publish(final ReplicationMsg msg, final Runnable whenWritten) throws IOException { final byte[] buffer = msg.getBytes(protocolVersion); if (buffer == null) { // skip anything that cannot be encoded for this peer. - return; + return false; } if (isRunning.get()) { + final Outgoing outgoing = new Outgoing(buffer, whenWritten); while (!closeInitiated) { try { // Avoid blocking forever so that we can check for session closure. - if (sendQueue.offer(buffer, 100, TimeUnit.MILLISECONDS)) + if (sendQueue.offer(outgoing, 100, TimeUnit.MILLISECONDS)) { - return; + return true; } } catch (final InterruptedException e) @@ -331,10 +378,27 @@ public void publish(final ReplicationMsg msg) throws IOException throw new IOException(e.getMessage()); } } + return false; } - else + send(buffer); + written(whenWritten); + return true; + } + + /** Runs what was to run once a message is written; a callback which fails takes nothing down. */ + private void written(final Runnable whenWritten) + { + if (whenWritten != null) { - send(buffer); + try + { + whenWritten.run(); + } + catch (final RuntimeException e) + { + logger.error(LocalizableMessage.raw("The callback of a message written to %s failed: %s", + readableRemoteAddress, stackTraceToSingleLineString(e))); + } } } @@ -535,10 +599,10 @@ public void run() boolean needClosing = false; while (!closeInitiated) { - byte[] buffer; + Outgoing outgoing; try { - buffer = sendQueue.take(); + outgoing = sendQueue.take(); } catch (InterruptedException ie) { @@ -546,13 +610,15 @@ public void run() } try { - send(buffer); + send(outgoing.buffer); } catch (IOException e) { setSessionError(e); needClosing = true; + continue; } + written(outgoing.whenWritten); } isRunning.set(false); if (needClosing) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java index 9c91cec200..12ecb570ba 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java @@ -17,9 +17,11 @@ */ package org.opends.server.replication.server; +import java.io.IOException; import java.net.SocketException; import org.forgerock.i18n.LocalizableMessage; +import org.forgerock.opendj.ldap.DN; import org.opends.server.api.DirectoryThread; import org.forgerock.i18n.slf4j.LocalizedLogger; import org.opends.server.replication.common.ServerStatus; @@ -120,24 +122,14 @@ public void run() replicationServerDomain.getBaseDN(), handler.getServerId()); } } + else if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer()) + { + forwardReplicaOfflineMsg((ReplicaOfflineMsg) updateMsg); + } else { // Publish the update to the remote server using a protocol version it supports session.publish(updateMsg); - /* - * Only the forward to a peer RS ends the wait of the shutdown: what the grace period - * buys is the rest of the topology learning that the replica went offline. - * ReplicationServerDomain.put() never queues this message for a directory server - its - * isUpdateMsgFiltered() drops it there - but a directory server which is catching up - * reads its updates from the changelog, where ReplicaCursor synthesizes a - * ReplicaOfflineMsg from the offline CSN of the replica. Publishing that one says - * nothing about the peer RSs the shutdown is waiting for. - */ - if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer()) - { - dsrsShutdownSync.replicaOfflineMsgForwarded( - replicationServerDomain.getBaseDN(), updateMsg.getCSN(), handler.getServerId()); - } } } } @@ -170,6 +162,37 @@ public void run() } } + /** + * Publishes a ReplicaOfflineMsg to the peer replication server, and reports the forward to the + * shutdown which may be waiting for it. + *

+ * Only the forward to a peer RS ends the wait of the shutdown: what the grace period buys is + * the rest of the topology learning that the replica went offline. + * ReplicationServerDomain.put() never queues this message for a directory server - its + * isUpdateMsgFiltered() drops it there - but a directory server which is catching up reads its + * updates from the changelog, where ReplicaCursor synthesizes a ReplicaOfflineMsg from the + * offline CSN of the replica. Publishing that one says nothing about the peer RSs the shutdown + * is waiting for, so it goes the way of every other update. + *

+ * The forward is reported once the message has been written to the peer, not once it is queued + * for the thread of the session: the shutdown closes the session as soon as its wait ends, and + * Session.close() drops whatever is still queued, so a message reported forwarded while it + * was queued behind one the peer had not read yet would never reach the peer. A message the + * session refuses - one the protocol version of the peer cannot carry, or one published while + * the session is being closed - will never be written, and the shutdown must not wait for it. + */ + private void forwardReplicaOfflineMsg(final ReplicaOfflineMsg msg) throws IOException + { + final DN baseDN = replicationServerDomain.getBaseDN(); + final int serverId = handler.getServerId(); + final boolean accepted = session.publish(msg, + () -> dsrsShutdownSync.replicaOfflineMsgForwarded(baseDN, msg.getCSN(), serverId)); + if (!accepted) + { + dsrsShutdownSync.replicaOfflineMsgNotForwarded(baseDN, serverId); + } + } + private boolean isUpdateMsgFiltered(UpdateMsg updateMsg) { if (handler.isDataServer()) diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java new file mode 100644 index 0000000000..7525fdd7a3 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SessionTest.java @@ -0,0 +1,245 @@ +/* + * 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 java.io.Closeable; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import 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.util.StaticUtils; +import org.testng.annotations.Test; + +/** + * A message published to a running session is handed to the session thread, which writes it to + * the socket later. A caller which needs to know when that happened - the replication server + * forwarding a ReplicaOfflineMsg, whose shutdown must not close the session before the message + * is on the wire - attaches a callback to the message, and the session runs it once, from the + * thread which wrote it, only after the write returned. + *

+ * The peer of each test reads nothing until the test lets it, and both ends of the connection + * have socket buffers far smaller than the first message published, so that the session thread + * of the end under test is held inside the write of that message for as long as the test wants + * - the state in which a message published behind it is queued and not written. + */ +@SuppressWarnings("javadoc") +public class SessionTest extends ReplicationTestCase +{ + private static final int SOCKET_TIMEOUT_MS = 30000; + /** + * Socket buffers small enough that {@link #BLOCKING_MESSAGE_SIZE} bytes cannot be written + * through them: the write blocks until the peer reads. Set explicitly on both ends, since the + * buffers the kernel picks on its own grow well beyond it on a loopback link - and the message + * is larger by far than what they hold, since a kernel which does not honour the size asked + * for on the receiving side, as macOS does not, must still be unable to take the whole of it. + */ + private static final int SOCKET_BUFFER_SIZE = 8 * 1024; + private static final int BLOCKING_MESSAGE_SIZE = 4 * 1024 * 1024; + /** Time given to a callback which must not run, to see that it does not. */ + private static final long SETTLE_MS = 500; + private static final int SENDER_ID = 1; + private static final int PEER_ID = 2; + + @Test + public void theCallbackRunsOnceTheMessageIsWrittenAndNotWhenItIsQueued() throws Exception + { + try (SessionPair pair = connectSessionPair()) + { + pair.publisher.start(); + pair.publisher.waitForStartup(); + + // The session thread is inside the write of this message until the peer reads it. + pair.publisher.publish(newBlockingMsg()); + pair.awaitBytesReachedThePeer(); + + final CountDownLatch written = new CountDownLatch(1); + final boolean accepted = pair.publisher.publish(new HeartbeatMsg(), written::countDown); + + assertThat(accepted).as("the message was refused by a running session").isTrue(); + assertThat(written.await(SETTLE_MS, TimeUnit.MILLISECONDS)) + .as("the callback ran while the message was still queued behind a message the peer " + + "had not read") + .isFalse(); + + assertThat(pair.peer.receive()).isInstanceOf(EntryMsg.class); + assertThat(written.await(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS)) + .as("the callback did not run once the message had been written") + .isTrue(); + assertThat(pair.peer.receive()).isInstanceOf(HeartbeatMsg.class); + } + } + + @Test + public void aMessageWhichCannotBeEncodedForThePeerIsRefusedWithoutRunningTheCallback() + throws Exception + { + try (SessionPair pair = connectSessionPair()) + { + pair.publisher.start(); + pair.publisher.waitForStartup(); + // A ReplicaOfflineMsg has no encoding before protocol version 8. + pair.publisher.setProtocolVersion(ProtocolVersion.REPLICATION_PROTOCOL_V7); + + final CountDownLatch written = new CountDownLatch(1); + final boolean accepted = + pair.publisher.publish(new ReplicaOfflineMsg(newCSN()), written::countDown); + + assertThat(accepted) + .as("a message the peer cannot decode was reported as accepted") + .isFalse(); + assertThat(written.await(SETTLE_MS, TimeUnit.MILLISECONDS)) + .as("the callback ran for a message which was never written") + .isFalse(); + } + } + + /** + * Before its thread is started, and once that thread is gone, a session writes on the + * publishing thread itself; the callback then runs on that same thread, after the write. + */ + @Test + public void theCallbackRunsAfterAMessageWrittenOnThePublishingThread() throws Exception + { + try (SessionPair pair = connectSessionPair()) + { + final CountDownLatch written = new CountDownLatch(1); + final boolean accepted = pair.publisher.publish(new HeartbeatMsg(), written::countDown); + + assertThat(accepted).as("the message was refused by a session with no thread").isTrue(); + assertThat(written.getCount()) + .as("the callback had not run when the publish which wrote the message returned") + .isZero(); + assertThat(pair.peer.receive()).isInstanceOf(HeartbeatMsg.class); + } + } + + private static EntryMsg newBlockingMsg() + { + return new EntryMsg(SENDER_ID, PEER_ID, new byte[BLOCKING_MESSAGE_SIZE], 1); + } + + private static CSN newCSN() + { + return new CSNGenerator(SENDER_ID, 0).newCSN(); + } + + /** + * Connects the end under test, in the server role of the replication protocol, with a peer + * which reads only when a test does. Both ends exchange one message under TLS and then drop the + * security layer, as the replication handshake does when encryption is not required, so that + * the socket buffers alone decide when a write blocks: a message read under TLS by each end is + * what consumes the records TLS itself sends after its negotiation, which would otherwise be + * read as the start of a replication message once the layer is gone. + */ + private static SessionPair connectSessionPair() throws Exception + { + final ReplSessionSecurity security = getReplSessionSecurity(); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final Socket peerSocket = new Socket(); + Socket publisherSocket = null; + Session publisher = null; + boolean connected = false; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + peerSocket.setReceiveBufferSize(SOCKET_BUFFER_SIZE); + peerSocket.setTcpNoDelay(true); + peerSocket.connect(new InetSocketAddress("127.0.0.1", listen.getLocalPort()), SOCKET_TIMEOUT_MS); + // The TLS negotiation needs both ends handshaking at the same time. + final Future peerEnd = + executor.submit(() -> security.createClientSession(peerSocket, SOCKET_TIMEOUT_MS)); + + publisherSocket = listen.accept(); + publisherSocket.setSendBufferSize(SOCKET_BUFFER_SIZE); + publisherSocket.setTcpNoDelay(true); + publisher = security.createServerSession(publisherSocket, SOCKET_TIMEOUT_MS); + assertThat(publisher).as("could not create the session under test").isNotNull(); + final Session peer = peerEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + + publisher.publish(new HeartbeatMsg()); + assertThat(peer.receive()).isInstanceOf(HeartbeatMsg.class); + peer.publish(new HeartbeatMsg()); + assertThat(publisher.receive()).isInstanceOf(HeartbeatMsg.class); + publisher.stopEncryption(); + peer.stopEncryption(); + connected = true; + return new SessionPair(publisher, peer, peerSocket); + } + finally + { + executor.shutdownNow(); + if (!connected) + { + if (publisher != null) + { + publisher.close(); + } + StaticUtils.close(publisherSocket, peerSocket); + } + } + } + + private static final class SessionPair implements Closeable + { + private final Session publisher; + private final Session peer; + private final Socket peerSocket; + + private SessionPair(Session publisher, Session peer, Socket peerSocket) + { + this.publisher = publisher; + this.peer = peer; + this.peerSocket = peerSocket; + } + + /** + * Waits for the first bytes of a message to reach the peer: the session thread of the end + * under test is then inside the write of that message, and stays there until the peer + * reads, since the message is larger than the buffers on both sides of the connection. + */ + void awaitBytesReachedThePeer() throws Exception + { + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SOCKET_TIMEOUT_MS); + while (peerSocket.getInputStream().available() == 0) + { + assertThat(System.nanoTime() < deadline) + .as("nothing was written to the peer") + .isTrue(); + Thread.sleep(10); + } + } + + @Override + public void close() + { + // The peer first: a session thread held inside a write is released by the peer going away, + // and close() joins that thread. + peer.close(); + publisher.close(); + } + } +} 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 dbd66e387a..0cdf0b600b 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 @@ -23,6 +23,8 @@ import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.List; import java.util.TreeSet; @@ -38,6 +40,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.forgerock.opendj.ldap.DN; +import org.forgerock.opendj.ldap.ModificationType; import org.opends.server.TestCaseUtils; import org.opends.server.core.DirectoryServer; import org.opends.server.replication.ReplicationTestCase; @@ -46,6 +49,8 @@ import org.opends.server.replication.common.RSInfo; import org.opends.server.replication.common.ServerState; import org.opends.server.replication.protocol.DeleteMsg; +import org.opends.server.replication.protocol.ModifyMsg; +import org.opends.server.replication.protocol.ProtocolVersion; import org.opends.server.replication.protocol.ReplServerStartMsg; import org.opends.server.replication.protocol.ReplSessionSecurity; import org.opends.server.replication.protocol.ReplicaOfflineMsg; @@ -56,6 +61,8 @@ import org.opends.server.replication.protocol.WindowMsg; import org.opends.server.replication.service.DSRSShutdownSync; import org.opends.server.replication.service.ReplicationBroker; +import org.opends.server.types.Attributes; +import org.opends.server.types.Modification; import org.opends.server.util.StaticUtils; import org.opends.server.util.TestTimer; import org.testng.annotations.Test; @@ -96,8 +103,38 @@ public class ReplicationServerShutdownSyncTest extends ReplicationTestCase * published one, so the shutdown spends its whole grace period waiting for it. */ private static final int UNREACHABLE_DS_ID = 98; + /** The peer replication server whose session thread is busy writing an earlier change. */ + private static final int BUSY_RS_ID = 99; + /** The peer replication server whose protocol version predates the ReplicaOfflineMsg. */ + private static final int LEGACY_RS_ID = 100; /** Send window a peer advertises when nothing has to hold its writer back. */ private static final int PEER_WINDOW = 100; + /** + * Socket buffers of the connection to the peer which does not read: small enough that + * {@link #SOCKET_FILLING_CHANGE_SIZE} bytes cannot be written through them, so that the session + * thread writing that change is held inside the write until the peer reads. Set on both ends, + * since the buffers the kernel picks on its own grow well beyond it on a loopback link. + */ + private static final int SMALL_SOCKET_BUFFER_SIZE = 8 * 1024; + /** + * Larger by far than what the socket buffers hold: a kernel which does not honour the size + * asked for on the receiving side - macOS keeps a few hundred kilobytes there - must still + * be unable to take the whole change. + */ + private static final int SOCKET_FILLING_CHANGE_SIZE = 4 * 1024 * 1024; + /** + * What the peer which does not read must have been sent, and not read, for the session thread + * serving it to be inside a write: well below its receive buffer, since the kernel advertises + * less than the whole of it, and well above any of the small messages a replication server + * sends a peer on its own. + */ + private static final int SOCKET_BUFFER_FILL_MARK = SMALL_SOCKET_BUFFER_SIZE / 4; + /** + * Time a shutdown released by the ReplicaOfflineMsg being queued, rather than written, is given + * to close the session of the peer. A shutdown which waits for the write cannot close the + * session before the peer reads, so it spends this time and no more. + */ + private static final long EARLY_CLOSE_TIMEOUT_MS = 1000; /** * Send window of the peer which is held back: one change fills it, and the message which * follows stays with its writer until the peer gives it credit again. @@ -416,6 +453,97 @@ public void call() throws Exception } } + /** + * The forward the shutdown waits for must mean that the message has been written to the peer, + * not that it has been queued for the thread of its session: Session.close() interrupts and + * joins that thread without draining its queue, so a message still queued when the shutdown + * closes the session is lost, and the peer receives the StopMsg alone. The session thread is + * busy with an earlier message when the ReplicaOfflineMsg is queued behind it whenever the peer + * reads slower than the replication server writes. Here the peer does not read at all, and the + * socket buffers on both sides of its connection are far smaller than the change which fills + * them, so the session thread is held inside the write of that change until the test lets the + * peer read. + */ + @Test + public void thePeerStillReadingAnEarlierChangeIsToldTheReplicaWentOfflineBeforeItIsStopped() + throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final RecordingShutdownSync shutdownSync = new RecordingShutdownSync(); + final ExecutorService executor = Executors.newFixedThreadPool(2); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + FakePeerReplicationServer peer = null; + Future shutdown = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncBusySessionDb", 8235, replicationPort); + broker = + openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + + final ConnectedSessions connection = + connectSessionPair(listen, getReplSessionSecurity(), SMALL_SOCKET_BUFFER_SIZE); + final Future served = + serveAsTheListenThreadWould(replicationServer, connection.localEnd, executor); + peer = FakePeerReplicationServer.connected(connection.remoteEnd, connection.remoteSocket, + BUSY_RS_ID, baseDN, EMPTY_DN_GENID, PEER_WINDOW); + served.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + final ReplicationServerDomain domain = + replicationServer.getReplicationServerDomain(baseDN, true); + waitForConnectedReplicationServer(domain, BUSY_RS_ID); + + /* + * One generator for the change and the announcement: a CSN which does not follow the one + * of the change would be dropped by the handler of the peer as already seen. + */ + final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0); + broker.publish(newChangeLargerThanTheSocketBuffers(csns.newCSN())); + peer.awaitReceiveBufferFilled(); + + final CSN offlineCSN = csns.newCSN(); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + broker.publish(new ReplicaOfflineMsg(offlineCSN)); + shutdownSync.awaitDispatch(); + + shutdown = executor.submit(newShutdown(replicationServer)); + awaitCloseInitiated(connection.localEnd, EARLY_CLOSE_TIMEOUT_MS); + final List forwardedBeforeThePeerRead = new ArrayList<>(shutdownSync.forwardedBy()); + + final Future received = peer.receive(ReplicaOfflineMsg.class); + final ReplicaOfflineMsg forwarded = received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + final long elapsed = shutdown.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + + assertThat(forwarded) + .as("the peer was never told that the replica went offline: its session was closed " + + "with the message still queued behind the change it was reading, and its read " + + "ended with: %s (forward reported by %s, the shutdown took %d ms)", + peer.failure(), shutdownSync.forwardedBy(), elapsed) + .isNotNull(); + assertThat(forwardedBeforeThePeerRead) + .as("the writer reported the message forwarded while it was still queued behind a " + + "change the peer had not read") + .doesNotContain(BUSY_RS_ID); + assertThat(shutdownSync.forwardedBy()) + .as("the message was written to the peer and nothing reported the forward") + .contains(BUSY_RS_ID); + assertThat(elapsed) + .as("the shutdown waited out the grace period after the message had been written") + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + // Closing the peer releases a session thread held inside a write, and the shutdown with it. + closeQuietly(peer); + awaitQuietly(shutdown); + stop(broker); + removeQuietly(replicationServer); + executor.shutdownNow(); + } + } + /** * Only a peer replication server learning about the offline replica ends the wait. * ReplicationServerDomain.put() never queues a ReplicaOfflineMsg for a directory server, but @@ -652,6 +780,64 @@ public void theShutdownStopsWaitingForAPeerWhoseMessageTheWriterDropped() throws } } + /** + * A peer whose protocol version has no encoding for the ReplicaOfflineMsg cannot be told that + * the replica went offline - it predates the message, and has nothing to do with it - and + * nothing will ever report a forward to it: the session refuses the message, and the writer + * must strike the peer off rather than let the shutdown wait out the grace period for it. + */ + @Test + public void theShutdownStopsWaitingForAPeerWhoseProtocolCannotCarryTheMessage() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final RecordingShutdownSync shutdownSync = new RecordingShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + FakePeerReplicationServer peer = null; + try + { + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = newReplicationServer( + shutdownSync, "shutdownSyncLegacyProtocolDb", 8236, replicationPort); + broker = + openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + peer = FakePeerReplicationServer.connected(replicationPort, LEGACY_RS_ID, baseDN, EMPTY_DN_GENID, + PEER_WINDOW, ProtocolVersion.REPLICATION_PROTOCOL_V7); + + final ReplicationServerDomain domain = + replicationServer.getReplicationServerDomain(baseDN, true); + waitForConnectedReplicationServer(domain, LEGACY_RS_ID); + + final CSN offlineCSN = newOfflineCSN(); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + broker.publish(new ReplicaOfflineMsg(offlineCSN)); + shutdownSync.awaitDispatch(); + awaitGiveUpOn(shutdownSync, LEGACY_RS_ID, + "the writer let the shutdown wait for a peer whose protocol cannot carry the message"); + + final long startTime = System.nanoTime(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(shutdownSync.dispatchedTo()) + .as("the message was not queued for the peer, so this test never reproduced the " + + "refusal it is about") + .contains(LEGACY_RS_ID); + assertThat(shutdownSync.forwardedBy()) + .as("a message the peer cannot decode was reported forwarded to it") + .doesNotContain(LEGACY_RS_ID); + assertThat(elapsed) + .as("the shutdown waited for a forward to a peer whose protocol cannot carry the message") + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + closeQuietly(peer); + stop(broker); + removeQuietly(replicationServer); + } + } + /** * A peer whose handshake is aborted while the message is being pushed must not be waited for. *

@@ -1265,6 +1451,73 @@ private static CSN newOfflineCSN(int serverId) return new CSNGenerator(serverId, 0).newCSN(); } + /** + * A change of the collocated replica larger than the socket buffers of the connection to the + * peer which does not read, so that the session thread writing it to that peer is held inside + * the write. + */ + private static ModifyMsg newChangeLargerThanTheSocketBuffers(CSN csn) + { + final char[] value = new char[SOCKET_FILLING_CHANGE_SIZE]; + Arrays.fill(value, 'x'); + final List mods = newArrayList( + new Modification(ModificationType.REPLACE, Attributes.create("description", new String(value)))); + return new ModifyMsg(csn, DN.valueOf("uid=busy," + TEST_ROOT_DN_STRING), mods, "busy-entry-uuid"); + } + + /** The shutdown of the replication server, reporting how long it took. */ + private static Callable newShutdown(final ReplicationServer replicationServer) + { + return new Callable() + { + @Override + public Long call() + { + final long startTime = System.nanoTime(); + replicationServer.shutdown(); + return elapsedMillis(startTime); + } + }; + } + + /** + * Serves a connection to the replication server as its listen thread does - see + * ReplicationServer.runListen() - over a session the test established itself, so that the + * sockets underneath are its own to configure: the start message of the peer is read, and the + * handler is created and started from it. The start blocks until the handshake is over, so it + * runs on a thread of its own, as it does on the listen thread. + */ + private static Future serveAsTheListenThreadWould( + final ReplicationServer replicationServer, final Session session, ExecutorService executor) + { + return executor.submit(new Callable() + { + @Override + public ReplicationServerHandler call() throws Exception + { + final ReplServerStartMsg startMsg = (ReplServerStartMsg) session.receive(); + final ReplicationServerHandler rsHandler = + new ReplicationServerHandler(session, 100, replicationServer, 100); + rsHandler.startFromRemoteRS(startMsg); + return rsHandler; + } + }); + } + + /** + * Waits for the close of the session to have been initiated, and gives up quietly once the + * timeout is over: the caller says what a close within the timeout, or none, means. + */ + private static void awaitCloseInitiated(Session session, long timeoutMillis) + throws InterruptedException + { + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis); + while (!session.closeInitiated() && System.nanoTime() < deadline) + { + Thread.sleep(10); + } + } + private static boolean sleepQuietly(long millis) { try @@ -1314,6 +1567,24 @@ private void closeQuietly(FakePeerReplicationServer peer) } } + private static void awaitQuietly(Future future) + { + if (future != null) + { + try + { + future.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + catch (Exception ignored) + { + } + } + } + /** * Establishes a connected session pair over the given listen socket, as a remote server * connecting to the RS would. The TLS negotiation performed by the session factories needs both @@ -1325,8 +1596,20 @@ private void closeQuietly(FakePeerReplicationServer peer) private Session[] connectSessionPair(ServerSocket listenSocket, final ReplSessionSecurity security) throws Exception { - final Socket clientSocket = new Socket("127.0.0.1", listenSocket.getLocalPort()); - clientSocket.setTcpNoDelay(true); + final ConnectedSessions connection = connectSessionPair(listenSocket, security, 0); + return new Session[] { connection.remoteEnd, connection.localEnd }; + } + + /** + * Establishes a connected session pair over the given listen socket, with the send buffer of + * the local end and the receive buffer of the remote end bounded by the given size: a message + * larger than both then holds the thread writing it until the remote end reads. A size of 0 + * leaves the buffers to the kernel. + */ + private ConnectedSessions connectSessionPair(ServerSocket listenSocket, + final ReplSessionSecurity security, int socketBufferSize) throws Exception + { + final Socket clientSocket = new Socket(); final ExecutorService executor = Executors.newSingleThreadExecutor(); Future clientEnd = null; Socket serverSocket = null; @@ -1334,6 +1617,14 @@ private Session[] connectSessionPair(ServerSocket listenSocket, final ReplSessio boolean connected = false; try { + if (socketBufferSize > 0) + { + // Before the connection is made: the window the local end is told is sized from it. + clientSocket.setReceiveBufferSize(socketBufferSize); + } + clientSocket.setTcpNoDelay(true); + clientSocket.connect( + new InetSocketAddress("127.0.0.1", listenSocket.getLocalPort()), SOCKET_TIMEOUT_MS); clientEnd = executor.submit(new Callable() { @Override @@ -1344,14 +1635,18 @@ public Session call() throws Exception }); serverSocket = listenSocket.accept(); + if (socketBufferSize > 0) + { + serverSocket.setSendBufferSize(socketBufferSize); + } serverSocket.setTcpNoDelay(true); serverEnd = security.createServerSession(serverSocket, SOCKET_TIMEOUT_MS); assertThat(serverEnd).as("could not create a session for the handler under test").isNotNull(); - final Session[] sessionPair = - new Session[] { clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), serverEnd }; + final ConnectedSessions connection = new ConnectedSessions( + clientEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS), clientSocket, serverEnd); connected = true; - return sessionPair; + return connection; } finally { @@ -1365,6 +1660,21 @@ public Session call() throws Exception } } + /** The two ends of a connection to the replication server, and the socket of the remote one. */ + private static final class ConnectedSessions + { + private final Session remoteEnd; + private final Socket remoteSocket; + private final Session localEnd; + + private ConnectedSessions(Session remoteEnd, Socket remoteSocket, Session localEnd) + { + this.remoteEnd = remoteEnd; + this.remoteSocket = remoteSocket; + this.localEnd = localEnd; + } + } + private void closeServerEndQuietly(Session serverEnd, Socket serverSocket) { if (serverEnd != null) @@ -1497,6 +1807,9 @@ private static final class FakePeerReplicationServer private final long generationId; private final String serverURL; private final Session session; + /** The socket under the session: what it has received and not read is what the replication + * server has written to this peer. */ + private final Socket socket; private final ExecutorService reader = Executors.newSingleThreadExecutor(); /** * What ended the exchange with the replication server, so that a message which never arrived @@ -1517,8 +1830,40 @@ static FakePeerReplicationServer connected( static FakePeerReplicationServer connected(int replicationPort, int serverId, DN baseDN, long generationId, int windowSize) throws Exception { - final FakePeerReplicationServer peer = new FakePeerReplicationServer( - replicationPort, serverId, baseDN, generationId, windowSize); + return connected(replicationPort, serverId, baseDN, generationId, windowSize, + ProtocolVersion.getCurrentVersion()); + } + + /** A connected peer speaking the given version of the replication protocol. */ + static FakePeerReplicationServer connected(int replicationPort, int serverId, DN baseDN, + long generationId, int windowSize, short protocolVersion) throws Exception + { + return completed(new FakePeerReplicationServer( + replicationPort, serverId, baseDN, generationId, windowSize, protocolVersion)); + } + + /** + * A connected peer over a session the test established itself - one whose sockets it + * configured - which the replication server serves as its listen thread would. + */ + static FakePeerReplicationServer connected(Session newSession, Socket newSocket, int serverId, + DN baseDN, long generationId, int windowSize) throws Exception + { + return completed(new FakePeerReplicationServer(newSession, newSocket, serverId, baseDN, + generationId, windowSize, ProtocolVersion.getCurrentVersion())); + } + + /** A peer whose handshake stops after its first phase, before it sends its TopologyMsg. */ + static FakePeerReplicationServer handshaking( + int replicationPort, int serverId, DN baseDN, long generationId) throws Exception + { + return new FakePeerReplicationServer(replicationPort, serverId, baseDN, generationId, + PEER_WINDOW, ProtocolVersion.getCurrentVersion()); + } + + private static FakePeerReplicationServer completed(FakePeerReplicationServer peer) + throws Exception + { boolean handshaken = false; try { @@ -1536,58 +1881,115 @@ static FakePeerReplicationServer connected(int replicationPort, int serverId, DN return peer; } - /** A peer whose handshake stops after its first phase, before it sends its TopologyMsg. */ - static FakePeerReplicationServer handshaking( - int replicationPort, int serverId, DN baseDN, long generationId) throws Exception - { - return new FakePeerReplicationServer( - replicationPort, serverId, baseDN, generationId, PEER_WINDOW); - } - private FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, - long generationId, int windowSize) throws Exception + long generationId, int windowSize, short protocolVersion) throws Exception { this.serverId = serverId; this.generationId = generationId; - final Socket socket = new Socket(); + final Socket newSocket = new Socket(); Session newSession = null; String newServerURL = null; boolean started = false; try { - socket.setTcpNoDelay(true); - socket.connect(new InetSocketAddress("127.0.0.1", replicationPort), SOCKET_TIMEOUT_MS); - newSession = getReplSessionSecurity().createClientSession(socket, SOCKET_TIMEOUT_MS); - - newServerURL = "127.0.0.1:" + socket.getLocalPort(); - newSession.publish(new ReplServerStartMsg(serverId, newServerURL, baseDN, windowSize, - new ServerState(), generationId, false, GROUP_ID, 5000)); - final ReplServerStartMsg inStartMsg = - waitForSpecificMsg(newSession, ReplServerStartMsg.class); - if (!inStartMsg.getSSLEncryption()) + newSocket.setTcpNoDelay(true); + newSocket.connect(new InetSocketAddress("127.0.0.1", replicationPort), SOCKET_TIMEOUT_MS); + newSession = getReplSessionSecurity().createClientSession(newSocket, SOCKET_TIMEOUT_MS); + newServerURL = start(newSession, newSocket, serverId, baseDN, generationId, windowSize, + protocolVersion); + started = true; + } + finally + { + if (!started) { - newSession.stopEncryption(); + abandon(newSession, newSocket); } + } + serverURL = newServerURL; + session = newSession; + socket = newSocket; + } + + private FakePeerReplicationServer(Session newSession, Socket newSocket, int serverId, + DN baseDN, long generationId, int windowSize, short protocolVersion) throws Exception + { + this.serverId = serverId; + this.generationId = generationId; + String newServerURL = null; + boolean started = false; + try + { + newServerURL = start(newSession, newSocket, serverId, baseDN, generationId, windowSize, + protocolVersion); started = true; } finally { if (!started) { - // The caller has no handle on this peer yet, so nothing else would close it. - reader.shutdownNow(); - if (newSession != null) - { - newSession.close(); - } - else - { - StaticUtils.close(socket); - } + abandon(newSession, newSocket); } } serverURL = newServerURL; session = newSession; + socket = newSocket; + } + + /** + * Runs the first phase of the handshake, the exchange of the start messages, and returns the + * URL this peer announced itself under. + */ + private static String start(Session newSession, Socket newSocket, int serverId, DN baseDN, + long generationId, int windowSize, short protocolVersion) throws Exception + { + // The replication server speaks the older of the two versions from the start message on. + newSession.setProtocolVersion(protocolVersion); + final String newServerURL = "127.0.0.1:" + newSocket.getLocalPort(); + newSession.publish(new ReplServerStartMsg(serverId, newServerURL, baseDN, windowSize, + new ServerState(), generationId, false, GROUP_ID, 5000)); + final ReplServerStartMsg inStartMsg = + waitForSpecificMsg(newSession, ReplServerStartMsg.class); + if (!inStartMsg.getSSLEncryption()) + { + newSession.stopEncryption(); + } + return newServerURL; + } + + /** The caller has no handle on this peer yet, so nothing else would close it. */ + private void abandon(Session newSession, Socket newSocket) + { + reader.shutdownNow(); + if (newSession != null) + { + newSession.close(); + } + else + { + StaticUtils.close(newSocket); + } + } + + /** + * Waits for the replication server to have filled the receive buffer of this peer, which + * reads nothing meanwhile, up to {@link #SOCKET_BUFFER_FILL_MARK}: the session thread + * serving this peer is then inside the write of a message larger than the buffers on both + * sides of the connection, and stays there until this peer reads. + */ + void awaitReceiveBufferFilled() throws Exception + { + newConnectionTimer().repeatUntilSuccess(new Callable() + { + @Override + public Void call() throws Exception + { + assertThat(socket.getInputStream().available()) + .as("the replication server never filled the receive buffer of the peer") + .isGreaterThanOrEqualTo(SOCKET_BUFFER_FILL_MARK); + return null; + } + }); } /**