diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java index 357d9cbb86..1c96fdfe94 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java @@ -13,6 +13,7 @@ * * Copyright 2006-2009 Sun Microsystems, Inc. * Portions Copyright 2013-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.protocol; @@ -254,5 +255,11 @@ public String errorsToString() "concerned server ids: " + idList; } + /** {@inheritDoc} */ + @Override + public String toString() + { + return getClass().getSimpleName() + " csn: " + csn + ", " + errorsToString(); + } } diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java index 3136fb1a3e..05aad1426e 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java @@ -13,13 +13,17 @@ * * Copyright 2008-2009 Sun Microsystems, Inc. * Portions Copyright 2013-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.server; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import org.opends.server.replication.common.AssuredMode; import org.opends.server.replication.common.CSN; @@ -68,6 +72,15 @@ public abstract class ExpectedAcksInfo */ protected Map expectedServersAckStatus = new HashMap<>(); + /** + * Immutable snapshot of the ids of the servers we expect an ack from, taken + * at construction time. Used for lock-free membership checks: the key set of + * {@link #expectedServersAckStatus} never changes after construction, but + * that map has its values mutated under lock by {@code processReceivedAck()}, + * so it must not be read concurrently without synchronization. + */ + private final Set expectedServerIds; + /** * Facility for monitoring: * If the timeout occurs for the original update, we call createAck(true) @@ -99,6 +112,22 @@ protected ExpectedAcksInfo(CSN csn, ServerHandler requesterServerHandler, { expectedServersAckStatus.put(serverId, false); } + this.expectedServerIds = + Collections.unmodifiableSet(new HashSet<>(expectedServers)); + } + + /** + * Indicates whether the provided server is one of the servers an ack is + * expected from for the matching update message. Reads an immutable snapshot + * taken at construction time, so it is safe to call without holding the lock + * on this object. + * + * @param serverId The serverId of the server. + * @return true if an ack is expected from the provided server. + */ + public boolean isExpectedServer(int serverId) + { + return expectedServerIds.contains(serverId); } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java index 6e280e83c2..b6e1b2a904 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java @@ -80,6 +80,13 @@ class MessageHandler extends MonitorProvider private final int maxQueueBytesSize; /** Specifies whether the consumer is following the producer (is not late). */ private boolean following; + /** + * Specifies whether the last update message returned by + * {@link #getNextMessage()} was re-read from the changelog DB (catch-up + * path) rather than taken from the in-memory {@link #msgQueue}. Only ever + * accessed by the single consumer thread calling {@link #getNextMessage()}. + */ + private boolean lastMessageFromLateQueue; /** Specifies the current serverState of this handler. */ private ServerState serverState; /** Specifies the baseDN of the domain. */ @@ -225,6 +232,21 @@ boolean isFollowing() } } + /** + * Indicates whether the last update message returned by + * {@link #getNextMessage()} was re-read from the changelog DB (catch-up + * path) rather than taken from the in-memory queue. + *

+ * Must only be called from the consumer thread calling + * {@link #getNextMessage()}. + * + * @return true if the last returned update message came from the late queue + */ + protected boolean isLastMessageFromLateQueue() + { + return lastMessageFromLateQueue; + } + /** * Retrieves the name of this monitor provider. It should be unique among all * monitor providers, including all instances of the same monitor provider. @@ -319,6 +341,9 @@ protected UpdateMsg getNextMessage() throws ChangelogException msgQueue.consumeUpTo(msg); if (updateServerState(msg)) { + // the returned instance is the one re-read from the + // changelog DB, not its msgQueue equivalent + lastMessageFromLateQueue = true; return msg; } } @@ -347,6 +372,7 @@ protected UpdateMsg getNextMessage() throws ChangelogException } if (updateServerState(msg)) { + lastMessageFromLateQueue = true; return msg; } continue; @@ -379,6 +405,7 @@ protected UpdateMsg getNextMessage() throws ChangelogException * by the other server. * Otherwise just loop to select the next message. */ + lastMessageFromLateQueue = false; return msg; } } 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 ebe734624f..81d6e48932 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 @@ -13,7 +13,7 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. - * Portions Copyrighted 2026 3A Systems, LLC. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.server; @@ -789,6 +789,20 @@ private boolean isDifferentGenerationId(long generationId) return this.generationId > 0 && this.generationId != generationId; } + /** + * Indicates whether the ack window for the provided CSN is still open and + * the provided server is one of the servers an ack is expected from. + * + * @param csn The CSN of the update message. + * @param serverId The serverId of the candidate acknowledging server. + * @return true if an ack from the provided server would be accounted for. + */ + boolean isExpectedAck(CSN csn, int serverId) + { + final ExpectedAcksInfo expectedAcksInfo = waitingAcks.get(csn); + return expectedAcksInfo != null && expectedAcksInfo.isExpectedServer(serverId); + } + /** * Process an ack received from a given server. * diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java index 38e8a7eb03..4c5d179b09 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java @@ -13,6 +13,7 @@ * * Copyright 2008-2009 Sun Microsystems, Inc. * Portions Copyright 2013-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.server; @@ -84,7 +85,19 @@ public boolean processReceivedAck(ServerHandler ackingServer, AckMsg ackMsg) // Get the ack status for the matching server int ackingServerId = ackingServer.getServerId(); - boolean ackReceived = expectedServersAckStatus.get(ackingServerId); + Boolean ackReceived = expectedServersAckStatus.get(ackingServerId); + if (ackReceived == null) + { + // Ack from a server we were not expecting an ack from, for instance + // because the update was delivered to it with the assured flag of the + // original sender through the changelog catch-up path: ignore it. + if (logger.isTraceEnabled()) + { + logger.trace("Received ack from not expected server id: " + + ackingServerId + " ack message: " + ackMsg); + } + return false; + } if (ackReceived) { // Sanity check: this should never happen diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java index ae886fa5cb..6a0fbb7d7e 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java @@ -13,6 +13,7 @@ * * Copyright 2008-2009 Sun Microsystems, Inc. * Portions Copyright 2013-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.server; @@ -146,7 +147,19 @@ public boolean processReceivedAck(ServerHandler ackingServer, AckMsg ackMsg) { // Get the ack status for the matching server int ackingServerId = ackingServer.getServerId(); - boolean ackReceived = expectedServersAckStatus.get(ackingServerId); + Boolean ackReceived = expectedServersAckStatus.get(ackingServerId); + if (ackReceived == null) + { + // Ack from a server we were not expecting an ack from, for instance + // because the update was delivered to it with the assured flag of the + // original sender through the changelog catch-up path: ignore it. + if (logger.isTraceEnabled()) + { + logger.trace("Received ack from not expected server id: " + + ackingServerId + " ack message: " + ackMsg); + } + return false; + } if (ackReceived) { // Sanity check: this should never happen diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java index ca65e02a94..7a5cb4c0af 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java @@ -13,12 +13,15 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.server; import static org.opends.messages.ReplicationMessages.*; +import static org.opends.server.util.StaticUtils.*; import java.io.IOException; +import java.io.UnsupportedEncodingException; import java.util.Random; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -924,12 +927,32 @@ public void shutdown() */ public UpdateMsg take() throws ChangelogException { - final UpdateMsg msg = getNextMessage(); + UpdateMsg msg = getNextMessage(); + final boolean fromLateQueue = isLastMessageFromLateQueue(); acquirePermitInSendWindow(); if (msg != null) { + // Updates re-read from the changelog DB (catch-up path) carry the + // assured flag, mode and safe data level of their original sender: + // the NotAssuredUpdateMsg substitution performed at publish time by + // ReplicationServerDomain.addUpdate() only exists on the in-memory + // queue path. Normalize them here: keep the assured flag only while + // the ack window is still open and this server is expected to ack. + // Messages taken from the in-memory queue already carry the + // publish-time decision and must NOT be revisited: the ack window may + // legitimately close (timeout, or enough acks already received) + // before a slow peer gets here - e.g. when acquirePermitInSendWindow() + // above blocks on a closed send window - and such a peer must still + // receive the assured flag it was deemed eligible for. Its late ack is + // then safely ignored by ReplicationServerDomain.processAck() and + // ExpectedAcksInfo.processReceivedAck(). + if (fromLateQueue && msg.isAssured() + && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId)) + { + msg = toNotAssuredUpdateMsg(msg); + } incrementOutCount(); if (msg.isAssured()) { @@ -940,6 +963,37 @@ public UpdateMsg take() throws ChangelogException return null; } + /** + * Substitutes a not assured version of the provided update message so that a + * peer not expected to acknowledge it does not receive it with the assured + * flag. + *

+ * This is the counterpart, for the changelog catch-up path, of the + * {@link NotAssuredUpdateMsg} substitution performed on the in-memory queue + * path by ReplicationServerDomain.addUpdate(): updates re-read from the + * changelog DB keep the assured flag of their original sender and must be + * normalized in {@link #take()} before being handed to the ServerWriter. + */ + private UpdateMsg toNotAssuredUpdateMsg(UpdateMsg msg) + { + try + { + return new NotAssuredUpdateMsg(msg); + } + catch (UnsupportedEncodingException e) + { + // Could not build the not assured form (unexpected message encoding). + // Deliver the original message rather than dropping it: losing the + // update would break replication consistency, which is worse than a + // spurious ack - and such an ack is now safely ignored by + // ExpectedAcksInfo.processReceivedAck(). + logger.error(LocalizableMessage.raw( + "Could not substitute a not assured version of update message %s: %s", + msg, stackTraceToSingleLineString(e))); + return msg; + } + } + private void acquirePermitInSendWindow() { boolean acquired = false; diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java index dd602dce86..b58b9cda07 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java @@ -13,7 +13,7 @@ * * Copyright 2008-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. - * Portions Copyright 2026 3A Systems, LLC + * Portions Copyright 2023-2026 3A Systems, LLC */ package org.opends.server.replication.server; @@ -51,6 +51,7 @@ import org.opends.server.TestCaseUtils; import org.opends.server.replication.ReplicationTestCase; import org.opends.server.replication.common.AssuredMode; +import org.opends.server.replication.common.CSN; import org.opends.server.replication.common.CSNGenerator; import org.opends.server.replication.common.DSInfo; import org.opends.server.replication.common.RSInfo; @@ -581,7 +582,7 @@ private void checkUpdateAssuredParameters(UpdateMsg updateMsg) * Sends a new update from this DS. * @throws TimeoutException If timeout waiting for an assured ack */ - private void sendNewFakeUpdate() throws TimeoutException + private CSN sendNewFakeUpdate() throws TimeoutException { // Create a new delete update message (the simplest to create) DeleteMsg delMsg = new DeleteMsg(getBaseDN(), gen.newCSN(), UUID.randomUUID().toString()); @@ -590,6 +591,7 @@ private void sendNewFakeUpdate() throws TimeoutException prepareWaitForAckIfAssuredEnabled(delMsg); publish(delMsg); waitForAckIfAssuredEnabled(delMsg); + return delMsg.getCSN(); } private void assertReceivedWrongUpdates(int expectedNbUpdates, int expectedNbWrongUpdates) @@ -751,9 +753,14 @@ private class FakeReplicationServer extends Thread private CSNGenerator gen; /** False if a received update had assured parameters not as expected. */ - private boolean everyUpdatesAreOk = true; - /** Number of received updates. */ - private int nReceivedUpdates; + private volatile boolean everyUpdatesAreOk = true; + /** + * Number of received updates. Volatile as it is incremented by the listen + * thread and polled from the test thread (waitForReceivedUpdates()); a read + * observing the incremented value also publishes everyUpdatesAreOk, which + * is written before the increment. + */ + private volatile int nReceivedUpdates; /** * True if an ack has been replied to a received assured update (in assured @@ -1075,6 +1082,110 @@ public void testSafeDataLevelOnePrecommit() throws Exception testSafeDataLevelOne(DEFAULT_GID, false, false, DEFAULT_GID, DEFAULT_GID); } + /** + * Parameters for {@link #testCatchUpClearsAssuredFlag}: assured mode and + * safe data level of the update a peer catches up from the changelog. + */ + @DataProvider(name = "catchUpAssuredModeProvider") + private Object[][] catchUpAssuredModeProvider() + { + return new Object[][] + { + { AssuredMode.SAFE_DATA_MODE, 1 }, + { AssuredMode.SAFE_DATA_MODE, 2 }, + { AssuredMode.SAFE_READ_MODE, 1 }, + }; + } + + /** + * Regression test for issue #710: an update served to a peer through the + * changelog catch-up path must not keep the assured flag of its original + * sender. + *

+ * A fake RS connects with an empty server state after an assured update has + * already been received and persisted by the real RS. As the fake RS was not + * connected when the update was received, the update is re-read from the + * changelog DB (catch-up path) rather than served from the in-memory queue, + * and the fake RS is not an expected ack server for it. The update must + * therefore be forwarded with its assured flag cleared (while its assured + * mode and safe data level are preserved). This used to be delivered with + * assured=true, making the fake RS reply a spurious ack. Exercised across + * safe data (level 1 and level > 1) and safe read modes. + */ + @Test(dataProvider = "catchUpAssuredModeProvider", enabled = true) + public void testCatchUpClearsAssuredFlag(AssuredMode assuredMode, + int safeDataLevel) throws Exception + { + String testCase = "testCatchUpClearsAssuredFlag"; + debugInfo("Starting " + testCase + " mode=" + assuredMode + " sdl=" + safeDataLevel); + initTest(); + try + { + // Real RS to be tested + rs1 = createReplicationServer(RS1_ID, DEFAULT_GID, SMALL_TIMEOUT, testCase, 0); + + // Main DS sends an assured update, acknowledged by the RS. No other peer + // is connected yet, so the update only lands in the changelog DB, not in + // any peer message queue. + fakeRDs[1] = createFakeReplicationDomain(FDS1_ID, DEFAULT_GID, RS1_ID, + DEFAULT_GENID, assuredMode, safeDataLevel, LONG_TIMEOUT, TIMEOUT_DS_SCENARIO); + CSN csn = fakeRDs[1].sendNewFakeUpdate(); + + // Ensure the update is persisted before the peer connects, so the peer is + // served through the changelog catch-up path (the RS acks the DS before + // persisting, so returning from sendNewFakeUpdate() does not guarantee it). + waitForChangePersisted(rs1, csn); + + // A fake RS connects with an empty state: it must catch up the historical + // update from the changelog DB. It expects a non-assured forward and + // never replies an ack (TIMEOUT_RS_SCENARIO). + fakeRs1 = createFakeReplicationServer(FRS1_ID, DEFAULT_GID, DEFAULT_GENID, + false, assuredMode, safeDataLevel, TIMEOUT_RS_SCENARIO); + + // The historical update must have been received, with assured flag off + waitForReceivedUpdates(fakeRs1, 1); + fakeRs1.assertReceivedUpdates(1); + } + finally + { + endTest(); + } + } + + /** + * Waits until the provided change has been persisted in the changelog of the + * provided replication server, i.e. is retrievable through the catch-up path. + */ + private void waitForChangePersisted(ReplicationServer rs, CSN csn) throws Exception + { + ReplicationServerDomain domain = + rs.getReplicationServerDomain(DN.valueOf(TEST_ROOT_DN_STRING)); + assertNotNull(domain); + int i = 0; + while (!domain.getLatestServerState().cover(csn)) + { + if (i++ > 50) + { + Assert.fail("Change " + csn + " was not persisted in the changelog in time."); + } + Thread.sleep(100); + } + } + + /** + * Waits until the provided fake RS has received at least the expected number + * of updates, so the assertion on the updates does not race with delivery. + */ + private void waitForReceivedUpdates(FakeReplicationServer fakeRs, int expected) + throws Exception + { + int i = 0; + while (fakeRs.nReceivedUpdates < expected && i++ <= 50) + { + Thread.sleep(100); + } + } + /** * Returns possible combinations of parameters for testSafeDataLevelOne test. */ diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java new file mode 100644 index 0000000000..60adb658b2 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java @@ -0,0 +1,85 @@ +/* + * 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]". + * + * Portions Copyright 2026 3A Systems, LLC + */ +package org.opends.server.replication.server; + +import static org.mockito.Mockito.*; +import static org.testng.Assert.*; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.opends.server.DirectoryServerTestCase; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.protocol.AckMsg; +import org.testng.annotations.Test; + +/** + * Test the handling of acks received from servers that were not expected to + * acknowledge an assured update. + *

+ * A peer served through the changelog catch-up path receives the update with + * the assured flag of the original sender (see issue #710), so it may send + * back an ack for a server id that is not part of the expected servers of the + * matching {@link ExpectedAcksInfo}. That ack must be ignored instead of + * raising a {@link NullPointerException} while auto-unboxing the missing map + * entry, which would close the connection to the acking peer. + */ +@SuppressWarnings("javadoc") +public class ExpectedAcksInfoTest extends DirectoryServerTestCase +{ + private static final CSN CSN1 = new CSN(1, 1, 10); + + private static ServerHandler mockServerHandler(int serverId, boolean isDataServer) + { + ServerHandler handler = mock(ServerHandler.class); + when(handler.getServerId()).thenReturn(serverId); + when(handler.isDataServer()).thenReturn(isDataServer); + return handler; + } + + @Test + public void safeReadAckFromNotExpectedServerIsIgnored() + { + ServerHandler requester = mockServerHandler(1, true); + List expectedServers = Arrays.asList(2, 3); + SafeReadExpectedAcksInfo acksInfo = new SafeReadExpectedAcksInfo( + CSN1, requester, expectedServers, Collections. emptyList()); + + // Server id 99 is not in the expected servers list + ServerHandler notExpected = mockServerHandler(99, false); + assertFalse(acksInfo.processReceivedAck(notExpected, new AckMsg(CSN1)), + "An ack from a not expected server must not complete the ack info"); + assertFalse(acksInfo.isExpectedServer(99)); + assertTrue(acksInfo.isExpectedServer(2)); + } + + @Test + public void safeDataAckFromNotExpectedServerIsIgnored() + { + ServerHandler requester = mockServerHandler(1, true); + List expectedServers = Arrays.asList(2, 3); + SafeDataExpectedAcksInfo acksInfo = new SafeDataExpectedAcksInfo( + CSN1, requester, (byte) 3, expectedServers); + + // Server id 99 is a RS not in the expected servers list + ServerHandler notExpected = mockServerHandler(99, false); + assertFalse(acksInfo.processReceivedAck(notExpected, new AckMsg(CSN1)), + "An ack from a not expected server must not complete the ack info"); + assertFalse(acksInfo.isExpectedServer(99)); + assertTrue(acksInfo.isExpectedServer(3)); + } +}