From d244a24f63589fb15117e44bf28353af08dea45b Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 6 Jul 2026 11:53:37 +0300 Subject: [PATCH 1/4] Fix replication catch-up re-sending updates with the original assured flag (#710) A peer catching up from the changelog DB (following=false, the default right after handshake) re-read updates via fillLateQueue() and received them with the assured flag, mode and safe-data level of the original sender: the NotAssuredUpdateMsg substitution was only applied on the in-memory queue path in ReplicationServerDomain.addUpdate(). Non-eligible peers then replied a spurious AckMsg; for a safe-read cross-group RS whose ack arrived while the ack window was still open, processReceivedAck() auto-unboxed a missing map entry and threw a NullPointerException that closed the peer connection. Normalize updates in ServerHandler.take(), the single point feeding the ServerWriter from both the queue and the catch-up path: keep the assured flag only while the ack window is still open and this server was eligible for an ack when the update was received (ReplicationServerDomain.isExpectedAck), otherwise substitute a NotAssuredUpdateMsg. Harden both processReceivedAck() implementations to ignore acks from servers that are not expected instead of raising a NullPointerException. Add ExpectedAcksInfoTest for the not-expected-ack case and AssuredReplicationServerTest.testCatchUpClearsAssuredFlag, a deterministic regression test that forces the catch-up path. --- .../replication/server/ExpectedAcksInfo.java | 15 ++++ .../server/ReplicationServerDomain.java | 15 ++++ .../server/SafeDataExpectedAcksInfo.java | 15 +++- .../server/SafeReadExpectedAcksInfo.java | 15 +++- .../replication/server/ServerHandler.java | 36 +++++++- .../server/AssuredReplicationServerTest.java | 50 +++++++++++ .../server/ExpectedAcksInfoTest.java | 85 +++++++++++++++++++ 7 files changed, 228 insertions(+), 3 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java 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..ef44149535 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,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; @@ -101,6 +102,20 @@ protected ExpectedAcksInfo(CSN csn, ServerHandler requesterServerHandler, } } + /** + * Indicates whether the provided server is one of the servers an ack is + * expected from for the matching update message. The set of expected + * servers is fixed at construction time, so this 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 expectedServersAckStatus.containsKey(serverId); + } + /** * Gets the server handler of the server which requested the acknowledgments. * @return The server handler of the server which requested the 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 6463087ef9..7a1ca36b29 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,6 +13,7 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.replication.server; @@ -788,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..edb10fb131 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,7 +927,12 @@ public void shutdown() */ public UpdateMsg take() throws ChangelogException { - final UpdateMsg msg = getNextMessage(); + UpdateMsg msg = getNextMessage(); + if (msg != null && msg.isAssured() + && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId)) + { + msg = toNotAssuredUpdateMsg(msg); + } acquirePermitInSendWindow(); @@ -940,6 +948,32 @@ public UpdateMsg take() throws ChangelogException return null; } + /** + * Substitutes a not assured version of the provided update message. + *

+ * Updates read back from the changelog during catch-up keep the assured + * flag of their original sender, whereas the in-memory queue path posts a + * {@link NotAssuredUpdateMsg} to servers not expected to acknowledge (see + * ReplicationServerDomain.addUpdate()). The assured flag must only be + * delivered while the ack window of the update is still open and this + * server was eligible for an ack when the update was received, otherwise + * the remote server would send back an ack that no one waits for. + */ + private UpdateMsg toNotAssuredUpdateMsg(UpdateMsg msg) + { + try + { + return new NotAssuredUpdateMsg(msg); + } + catch (UnsupportedEncodingException e) + { + 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 8886a673a3..db8de3befd 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,6 +13,7 @@ * * Copyright 2008-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. + * Portions Copyright 2023-2026 3A Systems, LLC */ package org.opends.server.replication.server; @@ -1074,6 +1075,55 @@ public void testSafeDataLevelOnePrecommit() throws Exception testSafeDataLevelOne(DEFAULT_GID, false, false, DEFAULT_GID, DEFAULT_GID); } + /** + * 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 safe data + * level 1 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. This + * used to be delivered with assured=true, making the fake RS reply a + * spurious ack. + */ + @Test(enabled = true) + public void testCatchUpClearsAssuredFlag() throws Exception + { + String testCase = "testCatchUpClearsAssuredFlag"; + debugInfo("Starting " + testCase); + initTest(); + try + { + // Real RS to be tested + rs1 = createReplicationServer(RS1_ID, DEFAULT_GID, SMALL_TIMEOUT, testCase, 0); + + // Main DS sends an assured safe data level 1 update, acknowledged + // immediately 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.SAFE_DATA_MODE, 1, LONG_TIMEOUT, TIMEOUT_DS_SCENARIO); + fakeRDs[1].sendNewFakeUpdate(); + sleepWhileUpdatePropagates(500); + + // 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.SAFE_DATA_MODE, 1, TIMEOUT_RS_SCENARIO); + + sleepWhileUpdatePropagates(500); + // The historical update must have been received, with assured flag off + fakeRs1.assertReceivedUpdates(1); + } + finally + { + endTest(); + } + } + /** * 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)); + } +} From 84032327627bbfac249760062ff27ce732cdb805 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 6 Jul 2026 12:40:36 +0300 Subject: [PATCH 2/4] Address review of #714: late assured normalization, lock-free ack membership, wider catch-up test coverage - Move the assured-flag decision in ServerHandler.take() after acquirePermitInSendWindow(), so isExpectedAck() reflects the state at send time rather than at dequeue time (the send window may block longer than the assured timeout). - ExpectedAcksInfo.isExpectedServer() now reads an immutable snapshot of the expected server ids taken at construction, instead of reading the lock-protected, value-mutated expectedServersAckStatus map without a lock. - Parameterize testCatchUpClearsAssuredFlag over safe data level 1, safe data level > 1 and safe read; replace fixed sleeps with polling on changelog persistence and on received updates. - Document that take() is the single assured guard complementing the queue-path substitution in addUpdate(), and that the UnsupportedEncodingException branch deliberately fails open to preserve replication delivery. --- .../replication/server/ExpectedAcksInfo.java | 22 ++++- .../replication/server/ServerHandler.java | 40 +++++--- .../server/AssuredReplicationServerTest.java | 97 +++++++++++++++---- 3 files changed, 122 insertions(+), 37 deletions(-) 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 ef44149535..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 @@ -18,9 +18,12 @@ 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; @@ -69,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) @@ -100,20 +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. The set of expected - * servers is fixed at construction time, so this is safe to call without - * holding the lock on this object. + * 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 expectedServersAckStatus.containsKey(serverId); + return expectedServerIds.contains(serverId); } /** 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 edb10fb131..7644204d0d 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 @@ -928,16 +928,22 @@ public void shutdown() public UpdateMsg take() throws ChangelogException { UpdateMsg msg = getNextMessage(); - if (msg != null && msg.isAssured() - && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId)) - { - msg = toNotAssuredUpdateMsg(msg); - } acquirePermitInSendWindow(); if (msg != null) { + // Decide whether to keep the assured flag as late as possible, right + // before the message is handed to the ServerWriter for publishing, so + // the ack window check reflects the state at send time and not the + // potentially much earlier time the message was dequeued: + // acquirePermitInSendWindow() above may block longer than the assured + // timeout when the send window is closed. + if (msg.isAssured() + && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId)) + { + msg = toNotAssuredUpdateMsg(msg); + } incrementOutCount(); if (msg.isAssured()) { @@ -949,15 +955,18 @@ public UpdateMsg take() throws ChangelogException } /** - * Substitutes a not assured version of the provided update message. + * Substitutes a not assured version of the provided update message so that a + * peer not (or no longer) expected to acknowledge it does not receive it + * with the assured flag. *

- * Updates read back from the changelog during catch-up keep the assured - * flag of their original sender, whereas the in-memory queue path posts a - * {@link NotAssuredUpdateMsg} to servers not expected to acknowledge (see - * ReplicationServerDomain.addUpdate()). The assured flag must only be - * delivered while the ack window of the update is still open and this - * server was eligible for an ack when the update was received, otherwise - * the remote server would send back an ack that no one waits for. + * 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. Doing the + * check here in {@link #take()} makes it the single guard covering both + * paths - on the queue path the message posted to a non-eligible peer is + * already a {@code NotAssuredUpdateMsg}, so {@code isAssured()} is false and + * this method is not even reached. */ private UpdateMsg toNotAssuredUpdateMsg(UpdateMsg msg) { @@ -967,6 +976,11 @@ private UpdateMsg toNotAssuredUpdateMsg(UpdateMsg 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))); 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 db8de3befd..8f88fed4d5 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 @@ -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) @@ -1075,47 +1077,68 @@ 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 safe data - * level 1 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. This - * used to be delivered with assured=true, making the fake RS reply a - * spurious ack. + * 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(enabled = true) - public void testCatchUpClearsAssuredFlag() throws Exception + @Test(dataProvider = "catchUpAssuredModeProvider", enabled = true) + public void testCatchUpClearsAssuredFlag(AssuredMode assuredMode, + int safeDataLevel) throws Exception { String testCase = "testCatchUpClearsAssuredFlag"; - debugInfo("Starting " + testCase); + 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 safe data level 1 update, acknowledged - // immediately by the RS. No other peer is connected yet, so the update - // only lands in the changelog DB, not in any peer message queue. + // 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.SAFE_DATA_MODE, 1, LONG_TIMEOUT, TIMEOUT_DS_SCENARIO); - fakeRDs[1].sendNewFakeUpdate(); - sleepWhileUpdatePropagates(500); + 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.SAFE_DATA_MODE, 1, TIMEOUT_RS_SCENARIO); + false, assuredMode, safeDataLevel, TIMEOUT_RS_SCENARIO); - sleepWhileUpdatePropagates(500); // The historical update must have been received, with assured flag off + waitForReceivedUpdates(fakeRs1, 1); fakeRs1.assertReceivedUpdates(1); } finally @@ -1124,6 +1147,40 @@ public void testCatchUpClearsAssuredFlag() throws Exception } } + /** + * 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. */ From e5f6ed181c1eb6aab7356f07910e5ec75b1c1d4f Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 6 Jul 2026 13:13:00 +0300 Subject: [PATCH 3/4] Make FakeReplicationServer received-update counters volatile (review of #714) nReceivedUpdates is incremented by the listen thread and polled from the test thread in waitForReceivedUpdates(); mark it (and everyUpdatesAreOk, written before the increment) volatile so the poll observes updates reliably. --- .../server/AssuredReplicationServerTest.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) 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 8f88fed4d5..b80d792847 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 @@ -753,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 From 0c1af14d2f6d6167ce975819ae766312aca3463f Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 10 Jul 2026 17:54:00 +0300 Subject: [PATCH 4/4] Restrict take()-time assured normalization to the changelog catch-up path (#710) The isExpectedAck() check in ServerHandler.take() also revisited the publish-time assured decision for in-memory queue messages: a peer present in expectedServers lost the assured flag whenever the ack window closed (timeout, or enough acks already received) before its writer passed take() - e.g. while blocked on a closed send window. This made AssuredReplicationServerTest.testSafeReadWrongStatus fail deterministically on all CI platforms (the recovering DS received 2 assured updates instead of 4) and testSafeDataLevelHigh flap (an ack quorum reached through a faster RS stripped the flag for a slower one). Track in MessageHandler whether getNextMessage() returned a message re-read from the changelog DB (late queue) and apply the NotAssuredUpdateMsg substitution in take() only to those messages; queue-path messages keep the publish-time decision made by ReplicationServerDomain.addUpdate(). --- .../replication/server/MessageHandler.java | 27 +++++++++++++++ .../replication/server/ServerHandler.java | 34 +++++++++++-------- 2 files changed, 47 insertions(+), 14 deletions(-) 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/ServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java index 7644204d0d..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 @@ -928,18 +928,27 @@ public void shutdown() public UpdateMsg take() throws ChangelogException { UpdateMsg msg = getNextMessage(); + final boolean fromLateQueue = isLastMessageFromLateQueue(); acquirePermitInSendWindow(); if (msg != null) { - // Decide whether to keep the assured flag as late as possible, right - // before the message is handed to the ServerWriter for publishing, so - // the ack window check reflects the state at send time and not the - // potentially much earlier time the message was dequeued: - // acquirePermitInSendWindow() above may block longer than the assured - // timeout when the send window is closed. - if (msg.isAssured() + // 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); @@ -956,17 +965,14 @@ public UpdateMsg take() throws ChangelogException /** * Substitutes a not assured version of the provided update message so that a - * peer not (or no longer) expected to acknowledge it does not receive it - * with the assured flag. + * 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. Doing the - * check here in {@link #take()} makes it the single guard covering both - * paths - on the queue path the message posted to a non-eligible peer is - * already a {@code NotAssuredUpdateMsg}, so {@code isAssured()} is false and - * this method is not even reached. + * 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) {