diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java index 2653da06fc..0cd82bb102 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java @@ -1975,8 +1975,8 @@ void doPreOperation(PreOperationAddOperation addOperation) @Override public void publishReplicaOfflineMsg() { - pendingChanges.putReplicaOfflineMsg(); - dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN()); + final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg(); + dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN); } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java index 5d2f6fdbe9..8ebba8cbd6 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java @@ -13,6 +13,7 @@ * * Copyright 2009 Sun Microsystems, Inc. * Portions Copyright 2011-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; @@ -123,8 +124,10 @@ synchronized CSN putLocalOperation(PluginOperation operation) /** * Add a replica offline message to the pending list. + * + * @return the CSN of the message which was added */ - public synchronized void putReplicaOfflineMsg() + public synchronized CSN putReplicaOfflineMsg() { final CSN offlineCSN = csnGenerator.newCSN(); final PendingChange pendingChange = @@ -133,6 +136,7 @@ public synchronized void putReplicaOfflineMsg() pendingChanges.put(offlineCSN, pendingChange); pushCommittedChanges(); + return offlineCSN; } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java index 4bea16769d..ef1ba8b22c 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java @@ -178,6 +178,11 @@ public class ReplicationServer /** * Creates a new Replication server using the provided configuration entry. + *

+ * The synchronization object this creates is its own, so the resulting server does not + * synchronize its shutdown with a collocated directory server. A server which has to must be + * built with {@link #ReplicationServer(ReplicationServerCfg, DSRSShutdownSync)}, passing the + * instance the directory server side records its ReplicaOfflineMsgs on. * * @param cfg The configuration of this replication server. * @throws ConfigException When Configuration is invalid. @@ -802,6 +807,9 @@ private void abortInitialization() // listen port which cannot be bound, would otherwise leave them behind. Shut them down // before the changelog they write to, and one unchecked exception at a time: the changelog // this one is built on is known to be broken, and what follows still has to run. + // Nothing in an instance which never finished coming up can forward a pending + // ReplicaOfflineMsg, so this path does not wait for one: it would only delay the failure + // which is being reported by a grace period which cannot pay off. for (ReplicationServerDomain domain : getReplicationServerDomains()) { try @@ -1181,7 +1189,19 @@ public void shutdown() listenThread.interrupt(); } - // shutdown all the replication domains + /* + * Let the ReplicaOfflineMsgs a collocated DS sent be forwarded while every handler is still + * up, and only then stop the domains: shutting a domain down deactivates the consumer of its + * handlers, clears their message queue and closes their session - see OPENDJ-1453. All the + * domains wait together and share one deadline, so the shutdown is bounded by one grace + * period and the wait of one domain does not spend the grace period of the next. + *

+ * This also runs before the assured timer of any domain is cancelled, so an assured update + * still waiting for acks keeps timing out during the wait instead of holding its sender + * until the sessions are closed. + */ + awaitReplicaOfflineMsgsForwarded(); + for (ReplicationServerDomain domain : getReplicationServerDomains()) { domain.shutdown(); @@ -1202,6 +1222,28 @@ public void shutdown() allInstances.remove(this); } + /** + * Waits for the ReplicaOfflineMsg of every domain which has a replication server to forward it + * to. With no such server connected there is nobody to forward the message to, and waiting + * would only delay the shutdown by the whole grace period. + */ + private void awaitReplicaOfflineMsgsForwarded() + { + final List domainsToWaitFor = new ArrayList<>(); + for (ReplicationServerDomain domain : getReplicationServerDomains()) + { + if (!domain.getConnectedRSs().isEmpty()) + { + domainsToWaitFor.add(domain.getBaseDN()); + } + } + if (!domainsToWaitFor.isEmpty()) + { + dsrsShutdownSync.awaitReplicaOfflineMsgsForwarded( + domainsToWaitFor, dsrsShutdownSync.newShutdownDeadline()); + } + } + /** * Retrieves the time after which changes must be deleted from the * persistent storage (in milliseconds). diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java index d8d24cef1c..1674f4c5c8 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java @@ -1661,7 +1661,15 @@ private MonitorMsg createLocalTopologyMonitorMsg(int sender, int destination) return monitorMsg; } - /** Shutdown this ReplicationServerDomain. */ + /** + * Shutdown this ReplicationServerDomain. + *

+ * A ReplicaOfflineMsg which a collocated DS sent and which is still to be forwarded must be + * waited for before this runs: stopping the server handlers deactivates their consumer, clears + * their message queue and closes their session, after which the message can no longer be sent + * - see OPENDJ-1453. ReplicationServer.shutdown() waits for the messages of all of its domains + * before it stops any of them. + */ public void shutdown() { DirectoryServer.deregisterMonitorProvider(this); 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 2b7f685d02..aa9dc21805 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 @@ -13,6 +13,7 @@ * * Copyright 2006-2009 Sun Microsystems, Inc. * Portions Copyright 2011-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server; @@ -87,9 +88,14 @@ public void run() LocalizableMessage errMessage = null; try { - boolean shutdown = false; - while (!shutdown - || !dsrsShutdownSync.canShutdown(replicationServerDomain.getBaseDN())) + /* + * Looping here to wait for a pending ReplicaOfflineMsg would achieve nothing: this writer + * only stops once its handler has been shut down, which deactivates the consumer, clears + * the message queue and closes the session. The shutdown of the domain waits for the + * message to be forwarded before it stops the handlers - see + * ReplicationServerDomain.shutdown() and OPENDJ-1453. + */ + while (true) { final UpdateMsg updateMsg = this.handler.take(); if (updateMsg == null) @@ -97,15 +103,25 @@ public void run() // this connection is closing errMessage = LocalizableMessage.raw( "Connection closure: null update returned by domain."); - shutdown = true; + break; } - else if (!isUpdateMsgFiltered(updateMsg)) + if (!isUpdateMsgFiltered(updateMsg)) { // Publish the update to the remote server using a protocol version it supports session.publish(updateMsg); - if (updateMsg instanceof ReplicaOfflineMsg) + /* + * 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()); + dsrsShutdownSync.replicaOfflineMsgForwarded( + replicationServerDomain.getBaseDN(), updateMsg.getCSN()); } } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java index 1b04bd0f52..3d962b0e08 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/DSRSShutdownSync.java @@ -12,13 +12,19 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.service; -import java.util.concurrent.ConcurrentSkipListSet; -import java.util.concurrent.atomic.AtomicLong; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +import java.util.Collection; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import org.forgerock.opendj.ldap.DN; +import org.opends.server.replication.common.CSN; /** * Class useful for the case where DS/RS instances are collocated inside the @@ -27,24 +33,73 @@ * More specifically, it ensures a ReplicaOfflineMsg sent by the DS is * relayed/forwarded by the collocated RS to the other RSs in the topology * before the whole process shuts down. + *

+ * The state is kept per domain and per instance: the collocated DS and RS + * sides coordinate through the single instance MultimasterReplication hands + * to both of them. * * @since OPENDJ-1453 */ public class DSRSShutdownSync { - private static final ConcurrentSkipListSet replicaOfflineMsgs = new ConcurrentSkipListSet<>(); - private static AtomicLong stopInstanceTimestamp = new AtomicLong(); + /** + * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated + * RS, in milliseconds, counted from the moment the message was sent. + */ + public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000; + + private final long gracePeriod; + + /** + * The ReplicaOfflineMsg which has not been forwarded yet, per domain and per + * replica of that domain. + *

+ * It is kept per domain because a domain sends this message whenever its + * replication service is disabled - an online import, a restore, a + * configuration change - and not only when the process shuts down. A single + * entry for the whole process would be the one of the first such message and + * would leave no grace period at all to the shutdown this class exists for. + *

+ * It is kept per replica because the collocated RS relays the message of + * every replica connected to it, and the forward of another replica's + * message says nothing about this one. + */ + private final ConcurrentMap> replicaOfflineMsgs = + new ConcurrentHashMap<>(); + /** Monitor notified whenever a ReplicaOfflineMsg has been forwarded. */ + private final Object forwardedMonitor = new Object(); + + /** Creates a synchronization object using the default grace period. */ + public DSRSShutdownSync() + { + this(REPLICA_OFFLINE_GRACE_PERIOD); + } + + /** + * Creates a synchronization object using the provided grace period. + * + * @param gracePeriod + * how long a ReplicaOfflineMsg may hold back the shutdown, in milliseconds + */ + DSRSShutdownSync(long gracePeriod) + { + this.gracePeriod = gracePeriod; + } /** * Message has been sent. * * @param baseDN * the domain for which the message has been sent + * @param offlineCSN + * the CSN of the message, which identifies both the replica which announced itself + * offline and the announcement being waited for */ - public void replicaOfflineMsgSent(DN baseDN) + public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN) { - stopInstanceTimestamp.compareAndSet(0, System.currentTimeMillis()); - replicaOfflineMsgs.add(baseDN); + replicaOfflineMsgs + .computeIfAbsent(baseDN, dn -> new ConcurrentHashMap()) + .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime())); } /** @@ -52,23 +107,179 @@ public void replicaOfflineMsgSent(DN baseDN) * * @param baseDN * the domain for which the message has been sent + * @param forwardedCSN + * the CSN of the forwarded message */ - public void replicaOfflineMsgForwarded(DN baseDN) + public void replicaOfflineMsgForwarded(DN baseDN, CSN forwardedCSN) { - replicaOfflineMsgs.remove(baseDN); + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + if (msgs != null) + { + final int serverId = forwardedCSN.getServerId(); + final PendingOfflineMsg pending = msgs.get(serverId); + /* + * A replica announces itself offline on every disableService(), so the message which is + * forwarded now may be an older one - queued behind a backlog since an earlier import, or + * synthesized from the offline CSN of the changelog for a server which is catching up. + * Such a forward says nothing about the announcement the shutdown is waiting for, and must + * not consume its grace period. + */ + if (pending != null && pending.csn.isOlderThanOrEqualTo(forwardedCSN)) + { + msgs.remove(serverId, pending); + } + } + synchronized (forwardedMonitor) + { + forwardedMonitor.notifyAll(); + } } /** - * Whether a ReplicationServer ServerReader or ServerWriter can proceed with - * shutdown. + * Whether the shutdown of a domain can proceed, i.e. its ReplicaOfflineMsg + * has been forwarded or its grace period has expired. + *

+ * The shutdown itself blocks on {@link #awaitReplicaOfflineMsgsForwarded(Collection, long)} + * rather than polling this; it is the same state, observable without waiting for it. * * @param baseDN - * the baseDN of the ServerReader or ServerWriter . - * @return true if the caller can shutdown, false otherwise + * the baseDN of the domain being shut down + * @return true if the shutdown of this domain need not wait any longer, i.e. its message was + * forwarded or its grace period has expired, false otherwise */ public boolean canShutdown(DN baseDN) { - return !replicaOfflineMsgs.contains(baseDN) - || System.currentTimeMillis() - stopInstanceTimestamp.get() > 5000; + return remainingGracePeriod(baseDN) <= 0; + } + + /** + * Returns the time by which every wait of one shutdown must be over. + *

+ * A process shuts its domains down one after the other and each of them may have a message + * pending, so a deadline computed once and shared by all of them keeps the whole shutdown + * bounded by one grace period instead of one per domain. + * + * @return the point in time, on the {@link System#nanoTime()} clock, by which the waits must + * be over + */ + public long newShutdownDeadline() + { + return System.nanoTime() + MILLISECONDS.toNanos(gracePeriod); + } + + /** + * Waits for the ReplicaOfflineMsg of every provided domain to be forwarded, or for their grace + * periods or the provided deadline to expire. + *

+ * This must be called before the server handlers of those domains are stopped: stopping them + * deactivates their consumer, clears their message queue and closes their session, after which + * the message can no longer be forwarded. + *

+ * All the domains of one shutdown wait together rather than one after the other, so that the + * shutdown is bounded by one grace period without the wait of one domain spending the grace + * period of the next. + * + * @param baseDNs + * the baseDNs of the domains whose messages must be forwarded + * @param deadline + * the point in time, on the {@link System#nanoTime()} clock, by which this wait must + * be over whatever the domains announce in the meantime - see + * {@link #newShutdownDeadline()}. A deadline which is not in the future returns + * without waiting at all, for a caller which has nothing to wait for. + */ + public void awaitReplicaOfflineMsgsForwarded(Collection baseDNs, long deadline) + { + if (deadline - System.nanoTime() <= 0) + { + return; + } + synchronized (forwardedMonitor) + { + while (true) + { + final long timeout = Math.min(remainingGracePeriod(baseDNs), + NANOSECONDS.toMillis(deadline - System.nanoTime())); + if (timeout <= 0) + { + return; + } + try + { + forwardedMonitor.wait(timeout); + } + catch (InterruptedException e) + { + /* + * Give up waiting. The interrupt is deliberately not restored: what follows this call is + * the rest of the shutdown - joining the reader and writer thread of every handler, then + * closing the changelog DB - and an interrupt flag would make all of it give up too. + */ + return; + } + } + } + } + + /** + * Returns the time left, in milliseconds, to forward the ReplicaOfflineMsg of the replica of + * the provided domains which has the longest to wait, zero or less if none of them has a + * message pending. + */ + private long remainingGracePeriod(Collection baseDNs) + { + long remaining = 0; + for (DN baseDN : baseDNs) + { + remaining = Math.max(remaining, remainingGracePeriod(baseDN)); + } + return remaining; + } + + /** + * Returns the time left, in milliseconds, to forward the ReplicaOfflineMsg of the replica of + * this domain which has the longest to wait, zero or less if no message of this domain is + * pending. + */ + private long remainingGracePeriod(DN baseDN) + { + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + if (msgs == null) + { + return 0; + } + final long now = System.nanoTime(); + long remaining = 0; + for (PendingOfflineMsg pending : msgs.values()) + { + remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime)); + } + return remaining; + } + + /** + * A ReplicaOfflineMsg a replica announced and which has not been forwarded yet. + *

+ * This deliberately does not override {@code equals}: the two-argument + * {@link ConcurrentMap#remove(Object, Object)} of the forward guard must match the very + * announcement it read, not another one which happens to carry the same values. + */ + private static final class PendingOfflineMsg + { + /** The CSN of the message, so that the forward of an older one is not taken for this one. */ + private final CSN csn; + /** When the message was announced, on the {@link System#nanoTime()} clock. */ + private final long sentTime; + + private PendingOfflineMsg(CSN csn, long sentTime) + { + this.csn = csn; + this.sentTime = sentTime; + } + + @Override + public String toString() + { + return "PendingOfflineMsg(" + csn + ")"; + } } } 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 new file mode 100644 index 0000000000..18c7002f85 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java @@ -0,0 +1,793 @@ +/* + * 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.server; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING; +import static org.opends.server.util.CollectionUtils.newArrayList; + +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.TreeSet; +import java.util.concurrent.Callable; +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.AtomicBoolean; + +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.common.RSInfo; +import org.opends.server.replication.common.ServerState; +import org.opends.server.replication.protocol.ReplServerStartMsg; +import org.opends.server.replication.protocol.ReplSessionSecurity; +import org.opends.server.replication.protocol.ReplicaOfflineMsg; +import org.opends.server.replication.protocol.ReplicationMsg; +import org.opends.server.replication.protocol.Session; +import org.opends.server.replication.protocol.TopologyMsg; +import org.opends.server.replication.service.DSRSShutdownSync; +import org.opends.server.replication.service.ReplicationBroker; +import org.opends.server.util.StaticUtils; +import org.opends.server.util.TestTimer; +import org.testng.annotations.Test; + +/** + * The shutdown of a replication server must let a ReplicaOfflineMsg sent by a collocated + * directory server be forwarded to the other replication servers of the topology before the + * server handlers are stopped - stopping them deactivates their consumer, clears their message + * queue and closes their session, after which the message can no longer be sent. + *

+ * Most tests drive {@link DSRSShutdownSync} directly rather than through a collocated directory + * server: the contract they pin is when the shutdown of the replication server waits, and how + * long. {@link #thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns()} pins the outcome + * those waits exist for, on a peer connected through the real handshake. + */ +@SuppressWarnings("javadoc") +public class ReplicationServerShutdownSyncTest extends ReplicationTestCase +{ + private static final int SOCKET_TIMEOUT_MS = 30000; + /** A session end nobody owns is discarded, so its cleanup waits far less than a live one. */ + private static final int DISCARDED_SESSION_TIMEOUT_MS = 2000; + private static final int REMOTE_RS_ID = 92; + private static final int REMOTE_DS_ID = 93; + /** The collocated replica whose ReplicaOfflineMsg the shutdown waits for. */ + private static final int LOCAL_DS_ID = 94; + /** Time given to the forwarding thread before it releases the shutdown. */ + private static final long FORWARD_DELAY = 500; + /** How often the domains of {@link #theGracePeriodIsSharedByAllTheDomainsOfOneShutdown()} + * announce themselves offline again while the shutdown is waiting for them. */ + private static final long REANNOUNCE_INTERVAL = 200; + + @Test + public void shutdownWaitsForTheReplicaOfflineMsgToBeForwarded() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncWaitDb", 8221); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN, session); + + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN()); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + removeQuietly(replicationServer); + } + } + + @Test + public void shutdownResumesAsSoonAsTheReplicaOfflineMsgIsForwarded() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + Thread forwarder = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncForwardDb", 8222); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN, session); + final CSN offlineCSN = newOfflineCSN(); + forwarder = newForwarderThread(shutdownSync, baseDN, offlineCSN); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + + final long startTime = System.nanoTime(); + forwarder.start(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY) + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + joinQuietly(forwarder); + removeQuietly(replicationServer); + } + } + + /** + * The outcome the grace period exists for, end to end: a peer replication server connected + * through the real handshake has received the ReplicaOfflineMsg of the collocated replica by + * the time the shutdown returns. + *

+ * The waiting tests above measure durations only, so they stay green if the wait is moved + * after the handlers are stopped - which reintroduces OPENDJ-1453 and loses the message. This + * one fails in that case. + */ + @Test + public void thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + FakePeerReplicationServer peer = null; + Thread publisher = null; + try + { + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncDeliveryDb", 8226, replicationPort); + broker = openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + peer = new FakePeerReplicationServer(replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID); + + final ReplicationServerDomain domain = + replicationServer.getReplicationServerDomain(baseDN, true); + waitForConnectedReplicationServer(domain); + final Future received = peer.receiveReplicaOfflineMsg(); + + /* + * The replica announces itself offline once the shutdown of the replication server is + * already waiting for the message, which is the ordering the grace period exists for. + */ + final CSN offlineCSN = newOfflineCSN(); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + publisher = newPublisherThread(broker, offlineCSN); + + final long startTime = System.nanoTime(); + publisher.start(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + final ReplicaOfflineMsg forwarded = received.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS); + assertThat(forwarded) + .as("the peer replication server was never told that the replica went offline, its " + + "read ended with: %s", peer.readerFailure()) + .isNotNull(); + assertThat(forwarded.getCSN().getServerId()).isEqualTo(LOCAL_DS_ID); + assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY) + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + joinQuietly(publisher); + closeQuietly(peer); + stop(broker); + removeQuietly(replicationServer); + } + } + + /** + * Only a peer replication server learning about the offline replica ends the wait. + * ReplicationServerDomain.put() never queues a ReplicaOfflineMsg for a directory server, but + * the changelog cursor of a directory server which is catching up synthesizes one from the + * offline CSN of the replica, so the writer serving a directory server can publish it - and + * the peer replication servers would still know nothing. + */ + @Test + public void theForwardToADirectoryServerDoesNotEndTheWait() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncDataServerForwardDb", 8227, replicationPort); + broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + // a peer replication server, so that the shutdown does wait for the message: what this + // test pins is that the directory server receiving it is not what ends that wait + registerConnectedReplicationServer(replicationServer, baseDN, session); + final ReplicationServerDomain domain = + replicationServer.getReplicationServerDomain(baseDN, true); + final DataServerHandler dsHandler = waitForConnectedDirectoryServer(domain); + + final CSN offlineCSN = newOfflineCSN(); + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + // the very message the shutdown waits for, so only the guard of the writer can save it + dsHandler.add(new ReplicaOfflineMsg(offlineCSN)); + + // the directory server did receive it, so its writer went through the forwarding code + assertThat(waitForSpecificMsg(broker, ReplicaOfflineMsg.class).getCSN().getServerId()) + .isEqualTo(LOCAL_DS_ID); + assertThat(elapsedMillis(startTime)) + .as("the fixture must deliver the message well inside the grace period, otherwise " + + "the wait asserted below cannot be told apart from a slow delivery") + .isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD / 2); + + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed) + .as("the message published to a directory server ended the wait of the shutdown") + .isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + stop(broker); + removeQuietly(replicationServer); + } + } + + /** + * With no other replication server connected there is nobody to forward the message to, so + * waiting would only delay the shutdown of a standalone server by the whole grace period. + */ + @Test + public void shutdownIsNotDelayedWhenNoOtherReplicationServerCanForwardTheMessage() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + try + { + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncAloneDb", 8223); + replicationServer.getReplicationServerDomain(baseDN, true); + + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN()); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + removeQuietly(replicationServer); + } + } + + /** + * The writer serving a directory server must not hold back the shutdown either: it used to + * loop on the pending message until the grace period expired, although its handler had already + * been shut down - which deactivates its consumer and leaves the loop nothing to take. + */ + @Test + public void shutdownIsNotDelayedByTheWriterServingADirectoryServer() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + ReplicationBroker broker = null; + try + { + final int replicationPort = TestCaseUtils.findFreePort(); + replicationServer = + newReplicationServer(shutdownSync, "shutdownSyncDataServerDb", 8225, replicationPort); + broker = openReplicationSession(baseDN, REMOTE_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID); + + final long startTime = System.nanoTime(); + shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN()); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + finally + { + stop(broker); + removeQuietly(replicationServer); + } + } + + @Test + public void shutdownIsNotDelayedWhenNoReplicaOfflineMsgIsPending() throws Exception + { + final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + ReplicationServer replicationServer = null; + try (ServerSocket listen = TestCaseUtils.bindFreePort()) + { + listen.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncNoMsgDb", 8224); + final Session[] sessionPair = connectSessionPair(listen, getReplSessionSecurity()); + try (Session remoteEnd = sessionPair[0]; + Session session = sessionPair[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN, session); + + final long startTime = System.nanoTime(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isLessThan(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + removeQuietly(replicationServer); + } + } + + /** + * The domains of a replication server are shut down one after the other, so the grace period + * must bound the whole shutdown and not each of its domains: a process with several base DNs + * would otherwise pay it once per domain. + *

+ * Both domains keep announcing themselves offline while the shutdown is running, so neither + * wait can be ended by a forward and each of them runs to its bound - one grace period in + * total if it is shared, one per domain otherwise. + */ + @Test + public void theGracePeriodIsSharedByAllTheDomainsOfOneShutdown() throws Exception + { + final DN baseDN1 = DN.valueOf(TEST_ROOT_DN_STRING); + final DN baseDN2 = DN.valueOf("dc=world,dc=company"); + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final AtomicBoolean stopped = new AtomicBoolean(); + ReplicationServer replicationServer = null; + Thread reAnnouncer = null; + try (ServerSocket listen1 = TestCaseUtils.bindFreePort(); + ServerSocket listen2 = TestCaseUtils.bindFreePort()) + { + listen1.setSoTimeout(SOCKET_TIMEOUT_MS); + listen2.setSoTimeout(SOCKET_TIMEOUT_MS); + replicationServer = newReplicationServer(shutdownSync, "shutdownSyncSharedDeadlineDb", 8228); + final Session[] sessionPair1 = connectSessionPair(listen1, getReplSessionSecurity()); + final Session[] sessionPair2 = connectSessionPair(listen2, getReplSessionSecurity()); + try (Session remoteEnd1 = sessionPair1[0]; + Session session1 = sessionPair1[1]; + Session remoteEnd2 = sessionPair2[0]; + Session session2 = sessionPair2[1]) + { + registerConnectedReplicationServer(replicationServer, baseDN1, session1); + registerConnectedReplicationServer(replicationServer, baseDN2, session2); + /* + * Announce both domains offline here rather than leaving it to the thread below: the + * wait of the shutdown must be armed whatever that thread has had time to run. + */ + final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0); + shutdownSync.replicaOfflineMsgSent(baseDN1, csns.newCSN()); + shutdownSync.replicaOfflineMsgSent(baseDN2, csns.newCSN()); + reAnnouncer = newReAnnouncerThread(shutdownSync, baseDN1, baseDN2, stopped); + reAnnouncer.start(); + + final long startTime = System.nanoTime(); + replicationServer.shutdown(); + final long elapsed = elapsedMillis(startTime); + + assertThat(elapsed).isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + assertThat(elapsed) + .as("each domain waited its own grace period instead of sharing one deadline") + .isLessThan(2 * DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD); + } + } + finally + { + stopped.set(true); + joinQuietly(reAnnouncer); + removeQuietly(replicationServer); + } + } + + private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName, + int serverId) throws Exception + { + return newReplicationServer(shutdownSync, dbDirName, serverId, TestCaseUtils.findFreePort()); + } + + private ReplicationServer newReplicationServer(DSRSShutdownSync shutdownSync, String dbDirName, + int serverId, int replicationPort) throws Exception + { + return new ReplicationServer(new ReplServerFakeConfiguration( + replicationPort, dbDirName, 0, serverId, 0, 100, new TreeSet()), shutdownSync); + } + + /** + * Registers a peer replication server on the domain as the handshake does, but without the + * protocol exchange: the handler this leaves behind has no writer, which is enough for the + * tests which only need a domain with a connected peer. + */ + private void registerConnectedReplicationServer( + ReplicationServer replicationServer, DN baseDN, Session session) throws Exception + { + final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(baseDN, true); + final ReplicationServerHandler rsHandler = + new ReplicationServerHandler(session, 100, replicationServer, 100); + rsHandler.serverId = REMOTE_RS_ID; + rsHandler.serverURL = "127.0.0.1:1636"; + rsHandler.setBaseDNAndDomain(baseDN, false); + domain.lock(); + try + { + domain.register(rsHandler); + } + finally + { + domain.release(); + } + } + + private void waitForConnectedReplicationServer(final ReplicationServerDomain domain) + throws Exception + { + newConnectionTimer().repeatUntilSuccess(new TestTimer.CallableVoid() + { + @Override + public void call() throws Exception + { + assertThat(domain.getConnectedRSs()) + .as("the peer replication server never connected").containsKey(REMOTE_RS_ID); + } + }); + } + + private DataServerHandler waitForConnectedDirectoryServer(final ReplicationServerDomain domain) + throws Exception + { + return newConnectionTimer().repeatUntilSuccess(new Callable() + { + @Override + public DataServerHandler call() throws Exception + { + final DataServerHandler dsHandler = domain.getConnectedDSs().get(REMOTE_DS_ID); + assertThat(dsHandler).as("the directory server never connected").isNotNull(); + return dsHandler; + } + }); + } + + private static TestTimer newConnectionTimer() + { + return new TestTimer.Builder() + .maxSleep(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS) + .sleepTimes(10, TimeUnit.MILLISECONDS) + .toTimer(); + } + + private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, final DN baseDN, + final CSN offlineCSN) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + if (!sleepQuietly(FORWARD_DELAY)) + { + return; + } + shutdownSync.replicaOfflineMsgForwarded(baseDN, offlineCSN); + } + }); + } + + private Thread newPublisherThread(final ReplicationBroker broker, final CSN offlineCSN) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + if (!sleepQuietly(FORWARD_DELAY)) + { + return; + } + broker.publish(new ReplicaOfflineMsg(offlineCSN)); + } + }); + } + + private Thread newReAnnouncerThread(final DSRSShutdownSync shutdownSync, final DN baseDN1, + final DN baseDN2, final AtomicBoolean stopped) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + final CSNGenerator csns = new CSNGenerator(LOCAL_DS_ID, 0); + while (!stopped.get()) + { + shutdownSync.replicaOfflineMsgSent(baseDN1, csns.newCSN()); + shutdownSync.replicaOfflineMsgSent(baseDN2, csns.newCSN()); + if (!sleepQuietly(REANNOUNCE_INTERVAL)) + { + return; + } + } + } + }); + } + + /** Milliseconds elapsed since a {@link System#nanoTime()} reading, the clock the waits use. */ + private static long elapsedMillis(long startTime) + { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); + } + + /** The CSN of a message the collocated replica announces, as PendingChanges generates it. */ + private static CSN newOfflineCSN() + { + return new CSNGenerator(LOCAL_DS_ID, 0).newCSN(); + } + + private static boolean sleepQuietly(long millis) + { + try + { + Thread.sleep(millis); + return true; + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + return false; + } + } + + /** Teardown must never mask the primary assertion failure. */ + private void removeQuietly(ReplicationServer replicationServer) + { + try + { + remove(replicationServer); + } + catch (Exception ignored) + { + } + } + + private void joinQuietly(Thread thread) + { + if (thread != null) + { + try + { + thread.join(SOCKET_TIMEOUT_MS); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + } + + private void closeQuietly(FakePeerReplicationServer peer) + { + if (peer != null) + { + peer.close(); + } + } + + /** + * 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 + * ends handshaking at the same time, so the client end runs on its own thread. + * + * @return the two sessions: the remote (client) end first, then the local (server) end to hand + * to the handler under test + */ + 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 ExecutorService executor = Executors.newSingleThreadExecutor(); + Future clientEnd = null; + Socket serverSocket = null; + Session serverEnd = null; + boolean connected = false; + try + { + clientEnd = executor.submit(new Callable() + { + @Override + public Session call() throws Exception + { + return security.createClientSession(clientSocket, SOCKET_TIMEOUT_MS); + } + }); + + serverSocket = listenSocket.accept(); + 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 }; + connected = true; + return sessionPair; + } + finally + { + if (!connected) + { + // Nobody owns either end yet: close whatever they managed to create. + closeClientEndQuietly(clientEnd, clientSocket); + closeServerEndQuietly(serverEnd, serverSocket); + } + executor.shutdown(); + } + } + + private void closeServerEndQuietly(Session serverEnd, Socket serverSocket) + { + if (serverEnd != null) + { + serverEnd.close(); + } + else + { + StaticUtils.close(serverSocket); + } + } + + private void closeClientEndQuietly(Future clientEnd, Socket clientSocket) + { + if (clientEnd != null) + { + try + { + final Session session = clientEnd.get(DISCARDED_SESSION_TIMEOUT_MS, TimeUnit.MILLISECONDS); + if (session != null) + { + session.close(); + } + } + catch (Exception ignored) + { + clientEnd.cancel(true); + } + } + StaticUtils.close(clientSocket); + } + + /** + * A peer replication server which connects to the replication server under test and completes + * the handshake, so that the handler it leaves behind on the domain has a real writer and can + * actually forward what the domain pushes to it. + */ + private static final class FakePeerReplicationServer + { + private final Session session; + private final ExecutorService reader = Executors.newSingleThreadExecutor(); + /** Why the peer stopped reading, so that a missing message can be told from a failed one. */ + private volatile Exception readerFailure; + + FakePeerReplicationServer(int replicationPort, int serverId, DN baseDN, long generationId) + throws Exception + { + final Socket socket = new Socket(); + Session newSession = null; + boolean handshaken = false; + try + { + socket.setTcpNoDelay(true); + socket.connect(new InetSocketAddress("127.0.0.1", replicationPort), SOCKET_TIMEOUT_MS); + newSession = getReplSessionSecurity().createClientSession(socket, SOCKET_TIMEOUT_MS); + + final String serverURL = "127.0.0.1:" + socket.getLocalPort(); + final byte groupId = (byte) 1; + newSession.publish(new ReplServerStartMsg(serverId, serverURL, baseDN, 100, + new ServerState(), generationId, false, groupId, 5000)); + final ReplServerStartMsg inStartMsg = + waitForSpecificMsg(newSession, ReplServerStartMsg.class); + if (!inStartMsg.getSSLEncryption()) + { + newSession.stopEncryption(); + } + newSession.publish(new TopologyMsg(null, + newArrayList(new RSInfo(serverId, serverURL, generationId, groupId, 1)))); + waitForSpecificMsg(newSession, TopologyMsg.class); + handshaken = true; + } + finally + { + if (!handshaken) + { + // 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); + } + } + } + session = newSession; + } + + /** Returns the first ReplicaOfflineMsg this peer receives, or null if its session ends first. */ + Future receiveReplicaOfflineMsg() + { + return reader.submit(new Callable() + { + @Override + public ReplicaOfflineMsg call() + { + try + { + while (true) + { + final ReplicationMsg msg = session.receive(); + if (msg instanceof ReplicaOfflineMsg) + { + return (ReplicaOfflineMsg) msg; + } + } + } + catch (Exception e) + { + // The session is closed when the replication server completes its shutdown: whatever + // has not arrived by then never will. + readerFailure = e; + return null; + } + } + }); + } + + /** Returns what ended the read of this peer, null if nothing did. */ + Exception readerFailure() + { + return readerFailure; + } + + void close() + { + reader.shutdownNow(); + session.close(); + } + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java new file mode 100644 index 0000000000..81408a0a87 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/service/DSRSShutdownSyncTest.java @@ -0,0 +1,293 @@ +/* + * 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.service; + +import static java.util.Arrays.asList; +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.TimeUnit; + +import org.forgerock.opendj.ldap.DN; +import org.opends.server.DirectoryServerTestCase; +import org.opends.server.replication.common.CSN; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** Test the {@link DSRSShutdownSync} class. */ +@SuppressWarnings("javadoc") +public class DSRSShutdownSyncTest extends DirectoryServerTestCase +{ + /** Short grace period, for the contracts a test has to wait out. */ + private static final long GRACE_PERIOD = 500; + /** + * Grace period for the contracts which only read the state: long enough that no scheduling + * pause between announcing a message and reading the state can expire it. + */ + private static final long LONG_GRACE_PERIOD = 60000; + /** Time given to the forwarding thread before it forwards the message of one domain. */ + private static final long FORWARD_DELAY = 200; + private static final int SERVER_ID = 1; + private static final int OTHER_SERVER_ID = 2; + + private static DN baseDN1; + private static DN baseDN2; + + @BeforeClass + public static void classSetup() throws Exception + { + baseDN1 = DN.valueOf("dc=example,dc=com"); + baseDN2 = DN.valueOf("dc=world,dc=company"); + } + + @Test + public void canShutdownWhenNoReplicaOfflineMsgWasSent() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + @Test + public void cannotShutdownUntilTheReplicaOfflineMsgIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + @Test + public void canShutdownOnceTheReplicaOfflineMsgIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN offlineCSN = newCSN(SERVER_ID); + + shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, offlineCSN); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + @Test + public void canShutdownOnceTheGracePeriodExpired() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + Thread.sleep(GRACE_PERIOD + 50); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + /** + * A message sent earlier in the life of the process - an online import, a restore, a + * configuration change - must not consume the grace period of the message sent by the + * shutdown this class exists for. + */ + @Test + public void gracePeriodOfAShutdownIsNotSpentByAnEarlierMessage() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + // an import disables then re-enables the replication service + final CSN sentByTheImport = newCSN(SERVER_ID, 1); + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheImport); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheImport); + Thread.sleep(GRACE_PERIOD + 50); + + // the shutdown of the process, much later + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID, 2)); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + /** + * The message of an earlier announcement may still be queued behind a backlog when the + * shutdown announces the replica offline again. Forwarding that older message says nothing + * about the one the shutdown is waiting for, so it must not end the wait. + */ + @Test + public void aStaleForwardDoesNotConsumeTheGracePeriodOfANewerMessage() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN queuedByAnEarlierImport = newCSN(SERVER_ID, 1); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, queuedByAnEarlierImport); + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, queuedByAnEarlierImport); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + @Test + public void gracePeriodIsCountedPerDomain() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + Thread.sleep(GRACE_PERIOD + 50); + shutdownSync.replicaOfflineMsgSent(baseDN2, newCSN(SERVER_ID)); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + assertThat(shutdownSync.canShutdown(baseDN2)).isFalse(); + } + + /** + * A replication server relays the ReplicaOfflineMsg of every replica connected to it, so the + * forward of another replica's message must not release the shutdown of this one. + */ + @Test + public void theForwardOfAnotherReplicasMessageDoesNotEndTheWait() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, newCSN(OTHER_SERVER_ID)); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + /** The domain waits for the message of every one of its replicas, not for the first of them. */ + @Test + public void aReplicaWhichIsStillWaitingHoldsBackTheShutdownOfItsDomain() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN ofOneReplica = newCSN(SERVER_ID); + + shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(OTHER_SERVER_ID)); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + } + + @Test + public void canShutdownOnceEveryReplicaOfTheDomainIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN ofOneReplica = newCSN(SERVER_ID); + final CSN ofTheOtherReplica = newCSN(OTHER_SERVER_ID); + + shutdownSync.replicaOfflineMsgSent(baseDN1, ofOneReplica); + shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheOtherReplica); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofOneReplica); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheOtherReplica); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + /** + * The domains of a shutdown wait together: the wait ends when the message of every one of them + * has been forwarded, not when the first one has. Waiting for them one after the other would + * leave the domains which come later without a grace period at all, since the wait of the + * first one spends the deadline they share. + */ + @Test + public void oneWaitCoversEveryDomainOfTheShutdown() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN ofTheFirstDomain = newCSN(SERVER_ID, 1); + final CSN ofTheSecondDomain = newCSN(SERVER_ID, 2); + shutdownSync.replicaOfflineMsgSent(baseDN1, ofTheFirstDomain); + shutdownSync.replicaOfflineMsgSent(baseDN2, ofTheSecondDomain); + final Thread forwarder = + newForwarderThread(shutdownSync, ofTheFirstDomain, ofTheSecondDomain); + + final long startTime = System.nanoTime(); + forwarder.start(); + shutdownSync.awaitReplicaOfflineMsgsForwarded( + asList(baseDN1, baseDN2), shutdownSync.newShutdownDeadline()); + final long elapsed = millisSince(startTime); + forwarder.join(); + + assertThat(elapsed) + .as("the wait ended on the first domain forwarded, leaving the second one nothing") + .isGreaterThanOrEqualTo(2 * FORWARD_DELAY); + assertThat(elapsed).isLessThan(LONG_GRACE_PERIOD); + } + + /** + * However long the messages of a shutdown may still hold it back, the deadline the shutdown + * was given bounds the wait. + */ + @Test + public void theWaitIsBoundedByTheDeadlineOfTheShutdown() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + shutdownSync.replicaOfflineMsgSent(baseDN1, newCSN(SERVER_ID)); + shutdownSync.replicaOfflineMsgSent(baseDN2, newCSN(SERVER_ID)); + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(GRACE_PERIOD); + + final long startTime = System.nanoTime(); + shutdownSync.awaitReplicaOfflineMsgsForwarded(asList(baseDN1, baseDN2), deadline); + final long elapsed = millisSince(startTime); + + assertThat(elapsed).isGreaterThanOrEqualTo(GRACE_PERIOD - 50); + assertThat(elapsed) + .as("the wait outlived the deadline of the shutdown") + .isLessThan(2 * GRACE_PERIOD); + } + + /** Forwards the message of the first domain, then, as long again later, of the second one. */ + private Thread newForwarderThread(final DSRSShutdownSync shutdownSync, + final CSN ofTheFirstDomain, final CSN ofTheSecondDomain) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + try + { + Thread.sleep(FORWARD_DELAY); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, ofTheFirstDomain); + Thread.sleep(FORWARD_DELAY); + shutdownSync.replicaOfflineMsgForwarded(baseDN2, ofTheSecondDomain); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + }); + } + + private static long millisSince(long startTime) + { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); + } + + /** + * The CSN of a message a replica announced. They are built by hand rather than with a + * CSNGenerator: this class has no state to share with the server, and a generator would tie + * the test to the time service the server starts. + */ + private static CSN newCSN(int serverId) + { + return newCSN(serverId, 1); + } + + private static CSN newCSN(int serverId, int seqNum) + { + return new CSN(1, seqNum, serverId); + } +}