From 43f3d15faf8222dc60cc25d830beda7c16586c9a Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 08:45:54 +0300 Subject: [PATCH] [#950] Announce a ReplicaOfflineMsg before it is published, not after it may have been forwarded The announcement the shutdown of a collocated replication server waits on was recorded after PendingChanges.putReplicaOfflineMsg() had already put the message on the wire. A forward which won that race found nothing to clear, and the announcement which followed it was one nothing would ever remove: ReplicationServer.shutdown() then spent the whole REPLICA_OFFLINE_GRACE_PERIOD waiting for the forward of a message the topology already had. The announcement now sits where the message is published - the ReplicaOfflineMsg branch of pushCommittedChanges() - so it is in place before session.publish() is reached and the forward cannot precede it. It goes through ShutdownSyncAnnouncer, the announcer of one domain and one DSRSShutdownSync, which the domain hands its PendingChanges. Announcing at the publish site also makes it follow the publication rather than the queueing, which leaves the guard #918 put around the announcement nothing to do: a message a change in flight holds back is not published, and is therefore not announced either. The trace which reports such a message stays. The broker may still refuse the message once it is announced - no usable session, a recovery pending, or stopped in between - which #949 made domain.publish() report. Such an announcement is one nobody will ever forward, so it is withdrawn through the new DSRSShutdownSync.replicaOfflineMsgNotSent(), which takes back only the entry carrying that CSN and wakes the shutdown up as a forward does: what stays announced is what the broker reports as written. A withdrawal gives back what the announcement displaced. A replica announces itself offline on every disableService(), and each announcement replaces the entry of the replica, so the refused message of a disable which follows a failed re-enable within the grace period had taken the place of the message the earlier disable did send - and withdrawing it emptied the slot that message was still waiting in. PendingOfflineMsg now keeps the announcement it displaced, and the withdrawal puts it back, with its own clock and the peers it was queued for. It keeps it only while its grace period runs, or every announcement of a replica whose message nobody in this process forwards would be chained for the life of the process; and it gives back only an announcement which is still owed a forward, since the forward which released the displaced one may have been reported while the new announcement was being made, and its remove() then found the new entry in its place. New PendingChangesTest cases, all through the production announcer: the forward reported from within publish() finds the announcement in place and clears it, the announcement of a message the broker took stands until a peer forwards it, the announcement of a message the broker refused is withdrawn, and a message held back by a change in flight is never announced - neither while it waits, nor when the change which held it back completes and the message is given up on. DSRSShutdownSyncTest covers the withdrawal: it ends the wait and leaves nothing behind, it wakes a waiting shutdown up, it leaves a newer announcement of the same replica alone - and does not restore the stale one -, it gives an earlier one its wait back with the peers it was queued for and what is left of its own grace period, and a forward reported while the refused announcement stood in front is not seen. --- .../plugin/LDAPReplicationDomain.java | 15 +- .../replication/plugin/PendingChanges.java | 55 ++++- .../plugin/ShutdownSyncAnnouncer.java | 62 ++++++ .../replication/service/DSRSShutdownSync.java | 116 ++++++++++- .../plugin/PendingChangesTest.java | 159 ++++++++++++++- .../service/DSRSShutdownSyncTest.java | 193 ++++++++++++++++++ 6 files changed, 574 insertions(+), 26 deletions(-) create mode 100644 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java 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 989d0e5e87..0b874094d2 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 @@ -828,7 +828,8 @@ public void run() * The generator time is adjusted to the time of the last CSN received from * remote other servers. */ - pendingChanges = new PendingChanges(getGenerator(), this); + pendingChanges = new PendingChanges(getGenerator(), this, + new ShutdownSyncAnnouncer(dsrsShutdownSync, getBaseDN())); remotePendingChanges = new RemotePendingChanges(getServerState()); // listen for changes on the configuration @@ -2271,17 +2272,13 @@ void doPreOperation(PreOperationAddOperation addOperation) public void publishReplicaOfflineMsg() { final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg(); - if (offlineCSN != null) + if (offlineCSN == null && logger.isTraceEnabled()) { /* - * Only a message which really was published is announced: the shutdown of a collocated - * replication server waits for it to be forwarded, and would spend the whole grace - * period waiting for one which never reached the wire. + * The announcement itself is made where the message is published, so nothing has to be + * reported here: a message a change in flight held back was never announced, and the + * announcement of one the broker refused has been withdrawn. */ - dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN); - } - else if (logger.isTraceEnabled()) - { logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN() + " could not announce itself offline: the message was not published - a change which" + " is still in flight holds it back, or the broker had no session to write it to," 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 4f88b62622..231298fbd2 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 @@ -53,6 +53,9 @@ class PendingChanges /** The ReplicationDomain that will be used to send UpdateMsg. */ private final ReplicationDomain domain; + /** Told that the replica of this domain announces itself offline, or takes that back. */ + private final ReplicaOfflineAnnouncer replicaOfflineAnnouncer; + private boolean recoveringOldChanges; /** @@ -60,11 +63,16 @@ class PendingChanges * * @param csnGenerator The CSNGenerator to use to create new unique CSNs. * @param domain The ReplicationDomain that will be used to send UpdateMsg. + * @param replicaOfflineAnnouncer Told that the replica of this domain announces itself + * offline, before the message announcing it is published, and that it takes + * the announcement back when the broker refused the message. */ - PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain) + PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain, + ReplicaOfflineAnnouncer replicaOfflineAnnouncer) { this.csnGenerator = csnGenerator; this.domain = domain; + this.replicaOfflineAnnouncer = replicaOfflineAnnouncer; } /** @@ -198,9 +206,22 @@ synchronized CSN pushCommittedChanges() } else if (msg instanceof ReplicaOfflineMsg) { + /* + * Announce the replica offline before the message reaches the wire, and not after: + * a collocated replication server forwards it as soon as it has it, and a forward + * which finds nothing announced leaves the shutdown waiting out the whole grace + * period of a message the topology already has. + */ + final CSN offlineCSN = msg.getCSN(); + replicaOfflineAnnouncer.announce(offlineCSN); if (domain.publish(msg)) { - publishedOfflineCSN = msg.getCSN(); + publishedOfflineCSN = offlineCSN; + } + else + { + // The broker wrote it to no session, so nobody will forward what was announced. + replicaOfflineAnnouncer.withdraw(offlineCSN); } } @@ -266,4 +287,34 @@ synchronized boolean recoveryUntil(CSN recovered) } return recoveringOldChanges; } + + /** + * Told that the replica of this domain announces itself offline, or takes that back. + *

+ * A collocated replication server can forward a {@link ReplicaOfflineMsg} as soon as it is on + * the wire, and its shutdown waits for that forward, so the announcement has to be in place + * before the message is published: one made afterwards is one the forward found nothing to + * clear, and the shutdown spends its whole grace period on a message which has already gone + * out. The broker may still refuse the message once it is announced, and then the + * announcement is withdrawn: what stays announced is what the broker reports as written. + */ + interface ReplicaOfflineAnnouncer + { + /** + * Announces that the replica goes offline at the provided CSN. + * + * @param offlineCSN + * the CSN of the ReplicaOfflineMsg which is about to be published + */ + void announce(CSN offlineCSN); + + /** + * Withdraws the announcement of a message the broker refused: it was written to no session, + * so nobody will forward it. + * + * @param offlineCSN + * the CSN of the ReplicaOfflineMsg which was announced and not published + */ + void withdraw(CSN offlineCSN); + } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java new file mode 100644 index 0000000000..bb6a162416 --- /dev/null +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ShutdownSyncAnnouncer.java @@ -0,0 +1,62 @@ +/* + * 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.plugin; + +import org.forgerock.opendj.ldap.DN; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.plugin.PendingChanges.ReplicaOfflineAnnouncer; +import org.opends.server.replication.service.DSRSShutdownSync; + +/** + * Announces the replica of one domain offline to the {@link DSRSShutdownSync} the shutdown of a + * collocated replication server waits on, and takes such an announcement back. + *

+ * This is the announcer {@link LDAPReplicationDomain} hands its {@link PendingChanges}: the + * announcement goes through {@link DSRSShutdownSync#replicaOfflineMsgSent(DN, CSN)} and the + * withdrawal through {@link DSRSShutdownSync#replicaOfflineMsgNotSent(DN, CSN)}, for the domain + * the announcer was built for. + */ +final class ShutdownSyncAnnouncer implements ReplicaOfflineAnnouncer +{ + private final DSRSShutdownSync shutdownSync; + private final DN baseDN; + + /** + * Creates an announcer for the replica of one domain. + * + * @param shutdownSync + * the synchronization object the collocated replication server's shutdown waits on + * @param baseDN + * the domain whose replica announces itself + */ + ShutdownSyncAnnouncer(DSRSShutdownSync shutdownSync, DN baseDN) + { + this.shutdownSync = shutdownSync; + this.baseDN = baseDN; + } + + @Override + public void announce(CSN offlineCSN) + { + shutdownSync.replicaOfflineMsgSent(baseDN, offlineCSN); + } + + @Override + public void withdraw(CSN offlineCSN) + { + shutdownSync.replicaOfflineMsgNotSent(baseDN, offlineCSN); + } +} 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 ba432c4b3c..21e5aee36d 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 @@ -46,7 +46,7 @@ public class DSRSShutdownSync { /** * How long a ReplicaOfflineMsg may hold back the shutdown of the collocated - * RS, in milliseconds, counted from the moment the message was sent. + * RS, in milliseconds, counted from the moment the message was announced. */ public static final long REPLICA_OFFLINE_GRACE_PERIOD = 5000; @@ -97,19 +97,90 @@ public DSRSShutdownSync() } /** - * Message has been sent. + * Message is about to be sent. + *

+ * The announcement comes before the message is published rather than after: a collocated + * replication server can forward the message as soon as it is on the wire, and a forward which + * finds nothing announced has nothing to clear. The announcement of a message the broker then + * refuses is taken back by {@link #replicaOfflineMsgNotSent(DN, CSN)}. + *

+ * A replica announces itself offline on every disableService(), so this may take the place of + * an earlier announcement of the same replica which is still owed its forward. The earlier one + * is kept behind the new one: a forward of the newer message, which the replication server + * queued behind the earlier one, covers both, and a withdrawal of the newer one gives the + * earlier one its wait back. It is kept only while its own grace period runs: past it, the + * announcement holds nothing back any more, and keeping it would chain every announcement of + * a replica whose message nobody in this process forwards - a directory server without a + * collocated replication server, or connected to a remote one - for the life of the process. * * @param baseDN - * the domain for which the message has been sent + * the domain for which the message is being sent * @param offlineCSN - * the CSN of the message, which identifies both the replica which announced itself + * the CSN of the message, which identifies both the replica which announces itself * offline and the announcement being waited for */ public void replicaOfflineMsgSent(DN baseDN, CSN offlineCSN) { + final long announcedAt = System.nanoTime(); replicaOfflineMsgs .computeIfAbsent(baseDN, dn -> new ConcurrentHashMap()) - .put(offlineCSN.getServerId(), new PendingOfflineMsg(offlineCSN, System.nanoTime())); + .compute(offlineCSN.getServerId(), (serverId, displaced) -> + new PendingOfflineMsg(offlineCSN, announcedAt, + displaced != null && gracePeriodLeft(displaced, announcedAt) > 0 ? displaced : null)); + } + + /** + * The message which was announced was not sent after all: the broker had no session to write + * it to, or was stopped before it could. + *

+ * The announcement is made before the message is published, since a collocated replication + * server can forward it as soon as it is on the wire, so the announcement of a message the + * broker then refused has to be taken back: nobody will forward it, and the shutdown would + * spend the whole grace period waiting for that forward. Only the announcement carrying that + * CSN is withdrawn, and the announcement it displaced - an earlier message of the same replica + * which did go out and is still owed its forward - takes its place again. + *

+ * Whatever is reported about that earlier message while the announcement of the refused one + * stands in its place is not seen by it. A forward, or the loss of a peer it was queued for, + * is lost, and the shutdown then waits out what is left of the earlier message's own grace + * period; the peers it is queued for, if they are recorded in that window, are lost too, with + * the opposite effect - the first forward ends its wait, as for a message no peer was recorded + * for. That window is the one publish the broker refuses: at once on a connection error or a + * pending recovery, the broker's retry loop up to the reconnect when it has no session. The + * wait it can cost is bounded by a grace period which is already running. + * + * @param baseDN + * the domain for which the message was announced + * @param offlineCSN + * the CSN of the message which was not sent + */ + public void replicaOfflineMsgNotSent(DN baseDN, CSN offlineCSN) + { + final ConcurrentMap msgs = replicaOfflineMsgs.get(baseDN); + if (msgs != null) + { + final int serverId = offlineCSN.getServerId(); + final PendingOfflineMsg pending = msgs.get(serverId); + if (pending != null && pending.csn.equals(offlineCSN)) + { + /* + * The displaced announcement may owe nothing any more: the forward which released it + * can have been reported while this announcement was being made, so that its remove(), + * which matches the entry it read, found this one in its place. Given its place back, + * such an announcement would hold the shutdown for the rest of its grace period, since + * nobody will report that forward again. + */ + if (pending.displaced != null && !pending.displaced.isFullyForwarded()) + { + msgs.replace(serverId, pending, pending.displaced); + } + else + { + msgs.remove(serverId, pending); + } + } + } + notifyForwarded(); } /** @@ -338,11 +409,20 @@ private long remainingGracePeriod(DN baseDN) long remaining = 0; for (PendingOfflineMsg pending : msgs.values()) { - remaining = Math.max(remaining, gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime)); + remaining = Math.max(remaining, gracePeriodLeft(pending, now)); } return remaining; } + /** + * Returns the time left, in milliseconds, of the grace period of one announcement, zero or + * less once it has expired. + */ + private long gracePeriodLeft(PendingOfflineMsg pending, long now) + { + return gracePeriod - NANOSECONDS.toMillis(now - pending.sentTime); + } + /** * A ReplicaOfflineMsg a replica announced and which has not been forwarded yet. *

@@ -356,16 +436,23 @@ private static final class PendingOfflineMsg private final CSN csn; /** When the message was announced, on the {@link System#nanoTime()} clock. */ private final long sentTime; + /** + * The announcement of the same replica this one took the place of and which is still owed its + * forward, null when there was none or when its grace period had already expired. It is + * given its place back if this message is withdrawn. + */ + private final PendingOfflineMsg displaced; /** * The replication servers the message was queued for and which have not forwarded it yet, * null as long as it has not been queued for anybody. */ private volatile Set awaitedForwarders; - private PendingOfflineMsg(CSN csn, long sentTime) + private PendingOfflineMsg(CSN csn, long sentTime, PendingOfflineMsg displaced) { this.csn = csn; this.sentTime = sentTime; + this.displaced = displaced; } /** @@ -391,9 +478,8 @@ private boolean forwardedBy(int replicationServerId) { /* * The message never went through the collocated RS - a replica which picked a remote one - * announcing itself offline, or an announcement recorded after the message it belongs to - * was already relayed. Nobody is known to owe a forward, so keep the behaviour the wait - * had before the recipients were tracked: the first forward ends it. + * is announcing itself offline. Nobody is known to owe a forward, so keep the behaviour + * the wait had before the recipients were tracked: the first forward ends it. */ return true; } @@ -412,6 +498,16 @@ private boolean giveUpOn(int replicationServerId) return awaited != null && awaited.remove(replicationServerId) && awaited.isEmpty(); } + /** + * Returns whether every replication server the message was queued for has forwarded it, or + * has been given up on: nothing is left to wait for. False while no recipient is known. + */ + private boolean isFullyForwarded() + { + final Set awaited = awaitedForwarders; + return awaited != null && awaited.isEmpty(); + } + @Override public String toString() { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java index 23c026facb..1842e633a1 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java @@ -16,12 +16,14 @@ package org.opends.server.replication.plugin; import static org.mockito.Matchers.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.*; +import org.forgerock.opendj.ldap.DN; import org.mockito.ArgumentCaptor; import org.opends.server.DirectoryServerTestCase; import org.opends.server.replication.common.CSN; @@ -29,23 +31,36 @@ import org.opends.server.replication.protocol.LDAPUpdateMsg; import org.opends.server.replication.protocol.ReplicaOfflineMsg; import org.opends.server.replication.protocol.UpdateMsg; +import org.opends.server.replication.service.DSRSShutdownSync; import org.opends.server.replication.service.ReplicationDomain; import org.opends.server.types.operation.PluginOperation; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; /** * Tests the bookkeeping a replica does on its own changes: they are published in the order of - * their CSNs, and the announcement that the replica goes offline is only reported as sent when - * it really was published. + * their CSNs, and the announcement that the replica goes offline is made before the message is + * published, since the shutdown of a collocated replication server waits for that message to be + * forwarded - and stands only for a message the broker reports as written. *

- * These tests need no server: the changes are built by a CSNGenerator, which reads the time - * service, and the time service is up as soon as its class is loaded. + * These tests need no server: the changes are built by a CSNGenerator, which needs nothing but + * a server id. */ @SuppressWarnings("javadoc") @Test(groups = { "precommit", "replication" }, sequential = true) public class PendingChangesTest extends DirectoryServerTestCase { private static final int SERVER_ID = 42; + /** A peer replication server the collocated one relays the message to. */ + private static final int RS_ID = 11; + + private static DN baseDN; + + @BeforeClass + public static void classSetup() throws Exception + { + baseDN = DN.valueOf("dc=example,dc=com"); + } @Test public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Exception @@ -61,6 +76,47 @@ public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Excepti assertEquals(published.getCSN(), offlineCSN); } + /** + * The collocated replication server forwards the message as soon as it is on the wire, so an + * announcement made after the publish is one the forward found nothing to clear: nothing will + * ever remove it, and the shutdown waits out its whole grace period for a message which the + * topology already has. + *

+ * The forward is reported from inside publish(), which is where the message reaches the + * session, so the race is reproduced rather than waited for. + */ + @Test + public void theReplicaOfflineMsgIsAnnouncedBeforeItIsPublished() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final ReplicationDomain domain = mock(ReplicationDomain.class); + forwardWhilePublishing(domain, shutdownSync); + final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync); + + pendingChanges.putReplicaOfflineMsg(); + + assertTrue(shutdownSync.canShutdown(baseDN), + "the message was forwarded, so nothing must hold the shutdown back any longer"); + } + + /** + * The announcement of a message the broker took stands until a peer forwards it: the + * withdrawal is for the message the broker refused, and an announcement taken back after a + * publish which succeeded would leave the shutdown nothing to wait for. + */ + @Test + public void theAnnouncementOfAPublishedMessageStandsUntilItIsForwarded() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final PendingChanges pendingChanges = + newPendingChanges(domainWhichPublishes(true), shutdownSync); + + pendingChanges.putReplicaOfflineMsg(); + + assertFalse(shutdownSync.canShutdown(baseDN), + "the message went out and nobody has forwarded it yet, so the shutdown must wait for it"); + } + /** * The broker writes nothing when it has no usable session, when the changes which come before * this one still have to be republished by the recovery, or when it is stopped in between - and @@ -79,6 +135,27 @@ public void replicaOfflineMsgTheBrokerRefusedIsNotReportedAsSent() throws Except assertTrue(onlyMsgPublishedBy(domain) instanceof ReplicaOfflineMsg, "it was attempted"); } + /** + * The announcement is made before the message is published, and the broker may refuse it + * once it is: an announcement which stayed would be one nobody will ever forward, and the + * shutdown would wait out its whole grace period for a message which never left. It is + * therefore withdrawn - and it is a withdrawal, not an announcement which was never made: the + * shutdown is held back while the broker holds the message. + */ + @Test + public void theAnnouncementOfAReplicaOfflineMsgTheBrokerRefusedIsWithdrawn() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final ReplicationDomain domain = mock(ReplicationDomain.class); + refuseWhilePublishing(domain, shutdownSync); + final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync); + + assertNull(pendingChanges.putReplicaOfflineMsg(), "the broker refused the message"); + + assertTrue(shutdownSync.canShutdown(baseDN), + "the message never reached the wire, so nothing must hold the shutdown back"); + } + /** * The message carries the newest CSN of the replica, so a change which is still in flight * holds it back, and the broker is never even asked to publish it. @@ -116,6 +193,32 @@ public void replicaOfflineMsgWhichCouldNotBeSentIsNotPublishedLater() throws Exc assertTrue(published instanceof LDAPUpdateMsg, "published " + published); } + /** + * The announcement follows the publication rather than the queueing, so a message which a + * change in flight holds back is not announced: neither while it waits, nor when the change + * which held it back completes and the message is given up on. Announcing it either time + * would leave the shutdown waiting out its whole grace period for a forward which cannot + * happen. + */ + @Test + public void theReplicaOfflineMsgHeldBackByAChangeInFlightIsNeverAnnounced() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(); + final ReplicationDomain domain = domainWhichPublishes(true); + final PendingChanges pendingChanges = newPendingChanges(domain, shutdownSync); + final CSN inFlight = pendingChanges.putLocalOperation(newLocalOperation()); + + pendingChanges.putReplicaOfflineMsg(); + + assertTrue(shutdownSync.canShutdown(baseDN), + "the message is still queued behind a change in flight, and nothing was announced"); + + pendingChanges.commitAndPushCommittedChanges(inFlight, mock(LDAPUpdateMsg.class)); + + assertTrue(shutdownSync.canShutdown(baseDN), + "the message was given up on with the change which held it back, and never announced"); + } + /** * A change the broker refused leaves the pending changes all the same: the replica has done * it, its ServerState says so, and it is by finding that state ahead of the one its @@ -137,9 +240,55 @@ public void changeTheBrokerRefusedStillLeavesThePendingChanges() throws Exceptio assertEquals(pendingChanges.size(), 0, "and is not queued for a second attempt"); } + /** + * Reports the forward of the message from within the publish which puts it on the wire, after + * checking from there that the announcement is already in place, and publishes it. The forward + * is the one of a peer the message was never recorded as queued for, which is what a forward + * racing the announcement looks like. + */ + private void forwardWhilePublishing( + final ReplicationDomain domain, final DSRSShutdownSync shutdownSync) + { + doAnswer(invocation -> { + final UpdateMsg msg = (UpdateMsg) invocation.getArguments()[0]; + if (msg instanceof ReplicaOfflineMsg) + { + assertFalse(shutdownSync.canShutdown(baseDN), + "the message must be announced before it is published"); + shutdownSync.replicaOfflineMsgForwarded(baseDN, msg.getCSN(), RS_ID); + } + return true; + }).when(domain).publish(any(UpdateMsg.class)); + } + + /** + * Refuses to publish the message, the way a broker with no usable session does, after checking + * from within the publish that the announcement is already in place. + */ + private void refuseWhilePublishing( + final ReplicationDomain domain, final DSRSShutdownSync shutdownSync) + { + doAnswer(invocation -> { + assertFalse(shutdownSync.canShutdown(baseDN), + "the message must be announced before it is published"); + return false; + }).when(domain).publish(any(UpdateMsg.class)); + } + private PendingChanges newPendingChanges(ReplicationDomain domain) { - return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain); + return newPendingChanges(domain, new DSRSShutdownSync()); + } + + /** + * The pending changes of a replica whose domain announces itself through the shutdown sync, + * with the announcer LDAPReplicationDomain hands its own pending changes. + */ + private PendingChanges newPendingChanges( + final ReplicationDomain domain, final DSRSShutdownSync shutdownSync) + { + return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain, + new ShutdownSyncAnnouncer(shutdownSync, baseDN)); } /** A domain whose broker accepts, or refuses, whatever it is given to publish. */ 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 index be482e18d0..a0479cc62a 100644 --- 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 @@ -86,6 +86,150 @@ public void canShutdownOnceTheReplicaOfflineMsgIsForwarded() throws Exception assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); } + /** + * The announcement is made before the message is published, so the broker may still refuse + * it - no usable session, or stopped in between. What was announced and never written must + * not hold the shutdown back: nobody will forward it. + */ + @Test + public void canShutdownOnceTheReplicaOfflineMsgIsWithdrawn() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN offlineCSN = newCSN(SERVER_ID); + + shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, offlineCSN); + + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + /** + * A withdrawal takes back the very announcement it names, and not whatever the replica + * announced last: a stale one, of a message the replica has since announced again, is ignored, + * and the newer announcement is still owed its forward. Ignored, and not taken for the + * withdrawal of the newer one: that would put the stale announcement back in front, where the + * forward of its message - which says nothing about the newer one - would end the wait. + */ + @Test + public void theWithdrawalOfAnEarlierMessageLeavesANewerOneAlone() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN refusedByTheBroker = newCSN(SERVER_ID, 1); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker); + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker); + + assertThat(shutdownSync.canShutdown(baseDN1)).isFalse(); + + shutdownSync.replicaOfflineMsgForwarded(baseDN1, refusedByTheBroker, RS_ID); + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("the stale withdrawal was ignored, not turned into a restore") + .isFalse(); + } + + /** + * A replica announces itself offline on every disableService(), and a later announcement + * takes the place of the earlier one. When the broker then refuses the later message, the + * earlier one - which did go out, and which a peer still has to forward - must get its wait + * back: withdrawing the later announcement must not take the earlier one with it. + */ + @Test + public void theWithdrawalOfALaterMessageGivesTheEarlierOneItsWaitBack() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 1); + final CSN refusedByTheBroker = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker); + + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("the earlier message went out and nobody has forwarded it yet") + .isFalse(); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID); + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("the forward of the earlier message ends the wait") + .isTrue(); + } + + /** + * What the withdrawal gives back is the very announcement which was displaced, with the peers + * its message was queued for: the forward of one of them does not end a wait which is for + * several, as it would for a message which was announced again from scratch. + */ + @Test + public void theRestoredAnnouncementIsStillOwedTheForwardsItWasQueuedFor() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 1); + final CSN refusedByTheBroker = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + shutdownSync.replicaOfflineMsgDispatched( + baseDN1, sentByTheShutdown, asList(RS_ID, OTHER_RS_ID)); + shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker); + + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID); + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("the restored message is still owed the other peer's forward") + .isFalse(); + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, OTHER_RS_ID); + assertThat(shutdownSync.canShutdown(baseDN1)).isTrue(); + } + + /** + * The restored announcement keeps its own clock: the shutdown waits out what is left of the + * earlier message's grace period, not a new one counted from the withdrawal. + */ + @Test + public void theRestoredAnnouncementKeepsWhatIsLeftOfItsOwnGracePeriod() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(GRACE_PERIOD); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 1); + final CSN refusedByTheBroker = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + Thread.sleep(GRACE_PERIOD - 200); + shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker); + // past the end of the earlier message's grace period, well short of a whole new one + Thread.sleep(250); + + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("the wait started over at the withdrawal") + .isTrue(); + } + + /** + * While the announcement of a refused message stands in the place of the earlier one, what is + * reported about the earlier message is not seen by it: a forward reported in that window is + * lost, and the restored announcement waits out what is left of its own grace period. The + * window is the one refused publish; this pins the trade-off, so that a change to it is made + * knowingly. + */ + @Test + public void aForwardReportedWhileARefusedAnnouncementStoodIsNotSeen() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN sentByTheShutdown = newCSN(SERVER_ID, 1); + final CSN refusedByTheBroker = newCSN(SERVER_ID, 2); + + shutdownSync.replicaOfflineMsgSent(baseDN1, sentByTheShutdown); + shutdownSync.replicaOfflineMsgDispatched(baseDN1, sentByTheShutdown, asList(RS_ID)); + shutdownSync.replicaOfflineMsgSent(baseDN1, refusedByTheBroker); + // not seen: the announcement of the refused message stands in front + shutdownSync.replicaOfflineMsgForwarded(baseDN1, sentByTheShutdown, RS_ID); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, refusedByTheBroker); + + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("a forward reported while the refused announcement stood is not seen") + .isFalse(); + } + @Test public void canShutdownOnceTheGracePeriodExpired() throws Exception { @@ -430,6 +574,55 @@ public void theWaitEndsWhenTheLastPeerExpectedToForwardStops() throws Exception .isLessThan(LONG_GRACE_PERIOD); } + /** + * A withdrawal must wake the shutdown up as a forward does, and not leave it waiting for the + * forward of a message which never left. + */ + @Test + public void theWaitEndsWhenTheMessageIsWithdrawn() throws Exception + { + final DSRSShutdownSync shutdownSync = new DSRSShutdownSync(LONG_GRACE_PERIOD); + final CSN offlineCSN = newCSN(SERVER_ID); + shutdownSync.replicaOfflineMsgSent(baseDN1, offlineCSN); + final Thread withdrawer = newWithdrawerThread(shutdownSync, offlineCSN); + + final long startTime = System.nanoTime(); + withdrawer.start(); + shutdownSync.awaitReplicaOfflineMsgsForwarded( + asList(baseDN1), shutdownSync.newShutdownDeadline()); + final long elapsed = millisSince(startTime); + withdrawer.join(); + + assertThat(elapsed).isGreaterThanOrEqualTo(FORWARD_DELAY); + assertThat(elapsed) + .as("the withdrawal did not wake the wait up") + .isLessThan(LONG_GRACE_PERIOD); + assertThat(shutdownSync.canShutdown(baseDN1)) + .as("the withdrawn message holds nothing back") + .isTrue(); + } + + /** Withdraws the announcement, as the broker refusing the message during the wait does. */ + private Thread newWithdrawerThread(final DSRSShutdownSync shutdownSync, final CSN offlineCSN) + { + return new Thread(new Runnable() + { + @Override + public void run() + { + try + { + Thread.sleep(FORWARD_DELAY); + shutdownSync.replicaOfflineMsgNotSent(baseDN1, offlineCSN); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + }); + } + /** Stops the peer the message was queued for, as a disconnection during the wait does. */ private Thread newPeerStopperThread(final DSRSShutdownSync shutdownSync, final int peerId) {