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; + } + }); } /**