From 8ad407096337ce4b76a49ede75c75d3b7d338d49 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 3 Aug 2026 14:17:31 +0300 Subject: [PATCH 1/6] [#813] Refuse to create a replica DB once the changelog shutdown has started FileChangelogDB.getOrCreateReplicaDB() read the shutdown flag only in its loop condition, before getExistingOrNewDomainMap() inserted the domain map and before the replica DB was created under the monitor of that map. A caller which read false just before shutdownDB() flipped the flag therefore inserted its domain map into a map which had already been drained, and created a FileReplicaDB nothing would ever shut down: its monitor provider stayed registered for the lifetime of the process, and its log stayed referenced. The flag is now read again inside the synchronized (domainMap) block which already guards the creation. Reading false there means shutdownDB() has not flipped the flag yet, hence has not created its iterator over domainToReplicaDBs yet either: it will see the domain map, inserted before that monitor was taken, and will have to block on the same monitor to drain it. Reading true returns null, and the loop then throws ERR_CANNOT_CREATE_REPLICA_DB_BECAUSE_CHANGELOG_DB_SHUTDOWN, which is what a caller racing a shutdown is meant to get. getExistingOrNewDomainMap() and the creation of a replica DB, now newReplicaDB(), are package private and overridable so that the test can drive the interleaving step by step: the creator is held right after it has read the flag, the shutdown is held inside the shutdown of the replica DB it drains - once the domain map has been removed and while the replication environment is still open - and the creator is then released into that window. Without the fix the test reports both symptoms: the creation succeeds, and the monitor provider of the replica DB it created stays registered. --- .../changelog/file/FileChangelogDB.java | 53 ++- .../changelog/file/FileChangelogDBTest.java | 342 ++++++++++++++++++ 2 files changed, 393 insertions(+), 2 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index 2164ae965f..542a288b14 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -215,7 +215,19 @@ Pair getOrCreateReplicaDB(final DN baseDN, final int ser throw new ChangelogException(ERR_CANNOT_CREATE_REPLICA_DB_BECAUSE_CHANGELOG_DB_SHUTDOWN.get()); } - private ConcurrentMap getExistingOrNewDomainMap(final DN baseDN) + /** + * Returns the map holding the replica DBs of the provided domain, inserting a new one if it does + * not exist yet. + *

+ * Package private and overridable so that tests can stop a thread right after it has read the + * shutdown flag in {@link #getOrCreateReplicaDB(DN, int, ReplicationServer)}, i.e. inside the + * window {@link #shutdownDB()} races with. + * + * @param baseDN + * the baseDN whose map of replica DBs must be returned + * @return the map of replica DBs of the provided domain + */ + ConcurrentMap getExistingOrNewDomainMap(final DN baseDN) { // happy path: the domainMap already exists final ConcurrentMap currentValue = domainToReplicaDBs.get(baseDN); @@ -272,12 +284,49 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM return null; } - final FileReplicaDB newDB = new FileReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + if (shutdown.get()) + { + // A shutdown was initiated after the shutdown flag was read by getOrCreateReplicaDB(): + // it may already have drained domainToReplicaDBs before this domainMap was inserted into + // it, in which case nothing would ever shutdown a replicaDB created here. + // Reading false instead means shutdownDB() has not flipped the flag yet, hence has not + // created its iterator yet either: it will see this domainMap, which was inserted before + // this monitor was acquired, and will have to block on this same monitor to drain it. + return null; + } + + final FileReplicaDB newDB = newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); domainMap.put(serverId, newDB); return Pair.of(newDB, true); } } + /** + * Creates a new replica DB. + *

+ * Package private and overridable so that tests can control the creation and the shutdown of the + * replica DBs this changelog holds. + * + * @param serverId + * the serverId for which to create a replica DB + * @param baseDN + * the baseDN for which to create a replica DB + * @param server + * the ReplicationServer + * @param cryptoSuite + * the cryptosuite to use for encryption + * @param replicationEnv + * the replication environment holding the log of the replica DB + * @return the newly created replica DB + * @throws ChangelogException + * if a problem occurred with the database + */ + FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, + final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException + { + return new FileReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + } + @Override public void initializeDB() { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java new file mode 100644 index 0000000000..634ad23e4f --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java @@ -0,0 +1,342 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.server.changelog.file; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import org.assertj.core.api.SoftAssertions; +import org.forgerock.opendj.config.server.ConfigException; +import org.forgerock.opendj.ldap.DN; +import org.opends.server.TestCaseUtils; +import org.opends.server.core.DirectoryServer; +import org.opends.server.crypto.CryptoSuite; +import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.server.ReplServerFakeConfiguration; +import org.opends.server.replication.server.ReplicationServer; +import org.opends.server.replication.server.changelog.api.ChangelogException; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.assertj.core.api.Assertions.*; +import static org.opends.messages.ReplicationMessages.*; +import static org.opends.server.TestCaseUtils.*; + +/** + * Test the FileChangelogDB class. + */ +@SuppressWarnings("javadoc") +public class FileChangelogDBTest extends ReplicationTestCase +{ + /** Server id of the replica DB which is shut down by the drain of the changelog. */ + private static final int DRAINED_SERVER_ID = 814; + /** Server id of the replica DB whose creation races that drain. */ + private static final int RACING_SERVER_ID = 813; + private static final long TIMEOUT_MS = 30000; + + private final String cipherTransformation = "AES/CBC/PKCS5Padding"; + private final int keyLength = 128; + private DN TEST_ROOT_DN; + + @BeforeClass + public void setup() throws Exception + { + TEST_ROOT_DN = DN.valueOf(TEST_ROOT_DN_STRING); + } + + /** + * A replica DB whose creation loses the race against {@code shutdownDB()} must not be created at + * all: it would be held by a domain map the shutdown has already drained, so nothing would ever + * shut it down, and its monitor provider would stay registered for the lifetime of the process. + *

+ * The interleaving is driven step by step: + *

    + *
  1. the creator thread reads the shutdown flag, sees {@code false}, and is held there, before + * it inserts the domain map it needs;
    2. + *
  2. the shutdown flips the flag and drains {@code domainToReplicaDBs}, and is held inside the + * shutdown of the replica DB it found, i.e. once that domain map has been removed and while the + * replication environment is still open;
    3. + *
  3. the creator is released into that window.
    4. + *
+ */ + @Test + public void replicaDBLosingTheRaceAgainstShutdownIsNotCreated() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + RaceableChangelogDB changelogDB = null; + File testRoot = null; + Thread creator = null; + Thread shutdowner = null; + final AtomicReference creationFailure = new AtomicReference<>(); + final AtomicReference shutdownFailure = new AtomicReference<>(); + try + { + replicationServer = configureReplicationServer(); + testRoot = createCleanDir(); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite()); + changelogDB.initializeDB(); + + // the replica DB the drain will be held in, and which is the only one registered so far + changelogDB.holdNextReplicaDBInItsShutdown(); + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, DRAINED_SERVER_ID, replicationServer); + // asserted, so that the test cannot pass by looking for a registration it cannot see + assertThat(replicaDBMonitorNames(DRAINED_SERVER_ID)) + .as("the replica DB held by the drain is not registered") + .hasSize(1); + assertThat(replicaDBMonitorNames(RACING_SERVER_ID)).isEmpty(); + + final FileChangelogDB racedChangelogDB = changelogDB; + final ReplicationServer racedReplicationServer = replicationServer; + changelogDB.holdNextReplicaDBCreationBeforeItsDomainMapIsInserted(); + creator = new Thread("FileChangelogDBTest replica DB creator") + { + @Override + public void run() + { + try + { + racedChangelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, RACING_SERVER_ID, racedReplicationServer); + } + catch (Throwable t) + { + creationFailure.set(t); + } + } + }; + creator.start(); + changelogDB.awaitCreatorInWindow(); + + shutdowner = new Thread("FileChangelogDBTest changelog shutdown") + { + @Override + public void run() + { + try + { + racedChangelogDB.shutdownDB(); + } + catch (Throwable t) + { + shutdownFailure.set(t); + } + } + }; + shutdowner.start(); + changelogDB.awaitDrainInReplicaDBShutdown(); + + changelogDB.releaseCreator(); + creator.join(TIMEOUT_MS); + + assertThat(creator.isAlive()).as("the creator thread did not complete").isFalse(); + final SoftAssertions softly = new SoftAssertions(); + softly.assertThat(creationFailure.get()) + .as("a replica DB created while the changelog is being drained is released by nobody") + .isInstanceOf(ChangelogException.class) + .hasMessage(ERR_CANNOT_CREATE_REPLICA_DB_BECAUSE_CHANGELOG_DB_SHUTDOWN.get().toString()); + softly.assertThat(replicaDBMonitorNames(RACING_SERVER_ID)) + .as("monitor providers of the replica DBs created during the shutdown") + .isEmpty(); + softly.assertAll(); + + changelogDB.releaseDrain(); + shutdowner.join(TIMEOUT_MS); + assertThat(shutdowner.isAlive()).as("the shutdown thread did not complete").isFalse(); + assertThat(shutdownFailure.get()).isNull(); + assertThat(replicaDBMonitorNames(DRAINED_SERVER_ID)) + .as("the drained replica DB is still registered") + .isEmpty(); + } + finally + { + if (changelogDB != null) + { + changelogDB.releaseCreator(); + changelogDB.releaseDrain(); + changelogDB.shutdownDB(); + } + join(creator); + join(shutdowner); + // release what the unfixed code leaks, so that it does not outlive this test + for (String monitorName : replicaDBMonitorNames(RACING_SERVER_ID)) + { + DirectoryServer.getMonitorProviders().remove(monitorName); + } + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); + } + } + + private void join(final Thread thread) throws InterruptedException + { + if (thread != null) + { + thread.join(TIMEOUT_MS); + } + } + + /** Returns the names the replica DBs of the provided server id are registered under. */ + private List replicaDBMonitorNames(final int serverId) + { + final String prefix = "changelog for ds(" + serverId + ")"; + final List names = new ArrayList<>(); + for (String monitorName : DirectoryServer.getMonitorProviders().keySet()) + { + if (monitorName.startsWith(prefix)) + { + names.add(monitorName); + } + } + return names; + } + + private ReplicationServer configureReplicationServer() throws IOException, ConfigException + { + return new ReplicationServer( + new ReplServerFakeConfiguration(findFreePort(), null, 0, 2, 5000, 100, null)); + } + + private CryptoSuite createCryptoSuite() + { + return getServerContext().getCryptoManager().newCryptoSuite(cipherTransformation, keyLength, false); + } + + private File createCleanDir() throws IOException + { + String buildRoot = System.getProperty(TestCaseUtils.PROPERTY_BUILD_ROOT); + String path = System.getProperty(TestCaseUtils.PROPERTY_BUILD_DIR, buildRoot + + File.separator + "build"); + path = path + File.separator + "unit-tests" + File.separator + "FileChangelogDB"; + final File testRoot = new File(path); + TestCaseUtils.deleteDirectory(testRoot); + testRoot.mkdirs(); + return testRoot; + } + + /** + * A changelog DB which lets a test hold a thread creating a replica DB right after it has read + * the shutdown flag, and hold the shutdown inside the drain of {@code domainToReplicaDBs}. + */ + private static final class RaceableChangelogDB extends FileChangelogDB + { + private final AtomicBoolean holdNextCreation = new AtomicBoolean(); + private final AtomicBoolean holdNextReplicaDB = new AtomicBoolean(); + private final CountDownLatch creatorIsInWindow = new CountDownLatch(1); + private final CountDownLatch creatorIsReleased = new CountDownLatch(1); + private final CountDownLatch drainIsInReplicaDBShutdown = new CountDownLatch(1); + private final CountDownLatch drainIsReleased = new CountDownLatch(1); + + RaceableChangelogDB(final ReplicationServer replicationServer, final String dbDirectoryPath, + final CryptoSuite cryptoSuite) throws ConfigException + { + super(replicationServer, dbDirectoryPath, cryptoSuite); + } + + @Override + ConcurrentMap getExistingOrNewDomainMap(final DN baseDN) + { + if (holdNextCreation.compareAndSet(true, false)) + { + creatorIsInWindow.countDown(); + await(creatorIsReleased); + } + return super.getExistingOrNewDomainMap(baseDN); + } + + @Override + FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, + final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException + { + if (holdNextReplicaDB.compareAndSet(true, false)) + { + return new HeldOnShutdownReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + } + return super.newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + } + + void holdNextReplicaDBCreationBeforeItsDomainMapIsInserted() + { + holdNextCreation.set(true); + } + + void holdNextReplicaDBInItsShutdown() + { + holdNextReplicaDB.set(true); + } + + void awaitCreatorInWindow() + { + await(creatorIsInWindow); + } + + void awaitDrainInReplicaDBShutdown() + { + await(drainIsInReplicaDBShutdown); + } + + void releaseCreator() + { + creatorIsReleased.countDown(); + } + + void releaseDrain() + { + drainIsReleased.countDown(); + } + + /** A replica DB which holds the thread shutting it down until the test releases it. */ + private final class HeldOnShutdownReplicaDB extends FileReplicaDB + { + HeldOnShutdownReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, + final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException + { + super(serverId, baseDN, server, cryptoSuite, replicationEnv); + } + + @Override + void shutdown() + { + drainIsInReplicaDBShutdown.countDown(); + await(drainIsReleased); + super.shutdown(); + } + } + + private static void await(final CountDownLatch latch) + { + try + { + if (!latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)) + { + throw new IllegalStateException("timed out waiting for the replica DB creation race"); + } + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + } +} From 8773e2eb4299f9a76c0e3bd2179b5160daf7d25b Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 3 Aug 2026 23:45:49 +0300 Subject: [PATCH 2/6] [#813] Address review feedback on the replica DB creation race fix The losing branch of the race was tested, but the branch the fix relies on - a creation which reads the shutdown flag as false under the domain map monitor must have its replica DB shut down by the drain - was not. The new FileChangelogDBTest.replicaDBWinningTheRaceAgainstShutdownIsShutDownByTheDrain holds the creator inside newReplicaDB(), once the monitor provider is registered but before the DB is published into the domain map, waits for the shutdown to block on the domain map monitor, and releases the creator into the drain, which must shut the new replica DB down. The monitor assertions matched any name starting with "changelog for ds()", i.e. any domain of any replication server in the JVM: they now match the full registered name, scoped by the monitor name of the domain. The cleanup deregisters leaked providers through DirectoryServer.deregisterMonitorProvider(), which also releases the JMX MBean registered alongside, and covers both server ids. join() no longer swallows its timeout: a hung thread is interrupted and reported with its stack trace. The comment justifying the fix cites the ConcurrentHashMap iterator guarantee it relies on. The helpers copied from FileReplicaDBTest moved to FileChangelogTestFixtures, shared by both test classes. --- .../changelog/file/FileChangelogDB.java | 6 +- .../changelog/file/FileChangelogDBTest.java | 273 ++++++++++++++---- .../file/FileChangelogTestFixtures.java | 68 +++++ .../changelog/file/FileReplicaDBTest.java | 36 +-- 4 files changed, 298 insertions(+), 85 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogTestFixtures.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index 542a288b14..3ac6fe2b26 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -290,8 +290,10 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM // it may already have drained domainToReplicaDBs before this domainMap was inserted into // it, in which case nothing would ever shutdown a replicaDB created here. // Reading false instead means shutdownDB() has not flipped the flag yet, hence has not - // created its iterator yet either: it will see this domainMap, which was inserted before - // this monitor was acquired, and will have to block on this same monitor to drain it. + // created its iterator yet either: since ConcurrentHashMap iterators traverse the + // elements as they existed upon construction of the iterator, it will see this domainMap, + // which was inserted before this monitor was acquired, and will have to block on this + // same monitor to drain it. return null; } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java index 634ad23e4f..67c0ea128a 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java @@ -16,9 +16,6 @@ package org.opends.server.replication.server.changelog.file; import java.io.File; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -28,12 +25,14 @@ import org.assertj.core.api.SoftAssertions; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.DN; +import org.forgerock.opendj.server.config.server.MonitorProviderCfg; import org.opends.server.TestCaseUtils; +import org.opends.server.api.MonitorProvider; import org.opends.server.core.DirectoryServer; import org.opends.server.crypto.CryptoSuite; import org.opends.server.replication.ReplicationTestCase; -import org.opends.server.replication.server.ReplServerFakeConfiguration; import org.opends.server.replication.server.ReplicationServer; +import org.opends.server.replication.server.ReplicationServerDomain; import org.opends.server.replication.server.changelog.api.ChangelogException; import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; @@ -41,6 +40,8 @@ import static org.assertj.core.api.Assertions.*; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.TestCaseUtils.*; +import static org.opends.server.replication.server.changelog.file.FileChangelogTestFixtures.*; +import static org.opends.server.util.StaticUtils.toLowerCase; /** * Test the FileChangelogDB class. @@ -54,8 +55,6 @@ public class FileChangelogDBTest extends ReplicationTestCase private static final int RACING_SERVER_ID = 813; private static final long TIMEOUT_MS = 30000; - private final String cipherTransformation = "AES/CBC/PKCS5Padding"; - private final int keyLength = 128; private DN TEST_ROOT_DN; @BeforeClass @@ -93,19 +92,20 @@ public void replicaDBLosingTheRaceAgainstShutdownIsNotCreated() throws Exception final AtomicReference shutdownFailure = new AtomicReference<>(); try { - replicationServer = configureReplicationServer(); - testRoot = createCleanDir(); - changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite()); + replicationServer = configureReplicationServer(100, 5000); + testRoot = createCleanDir("FileChangelogDB"); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); changelogDB.initializeDB(); // the replica DB the drain will be held in, and which is the only one registered so far changelogDB.holdNextReplicaDBInItsShutdown(); changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, DRAINED_SERVER_ID, replicationServer); // asserted, so that the test cannot pass by looking for a registration it cannot see - assertThat(replicaDBMonitorNames(DRAINED_SERVER_ID)) + assertThat(DirectoryServer.getMonitorProviders().keySet()) .as("the replica DB held by the drain is not registered") - .hasSize(1); - assertThat(replicaDBMonitorNames(RACING_SERVER_ID)).isEmpty(); + .contains(replicaDBMonitorName(replicationServer, DRAINED_SERVER_ID)); + assertThat(DirectoryServer.getMonitorProviders().keySet()) + .doesNotContain(replicaDBMonitorName(replicationServer, RACING_SERVER_ID)); final FileChangelogDB racedChangelogDB = changelogDB; final ReplicationServer racedReplicationServer = replicationServer; @@ -155,95 +155,236 @@ public void run() .as("a replica DB created while the changelog is being drained is released by nobody") .isInstanceOf(ChangelogException.class) .hasMessage(ERR_CANNOT_CREATE_REPLICA_DB_BECAUSE_CHANGELOG_DB_SHUTDOWN.get().toString()); - softly.assertThat(replicaDBMonitorNames(RACING_SERVER_ID)) + softly.assertThat(DirectoryServer.getMonitorProviders().keySet()) .as("monitor providers of the replica DBs created during the shutdown") - .isEmpty(); + .doesNotContain(replicaDBMonitorName(replicationServer, RACING_SERVER_ID)); softly.assertAll(); changelogDB.releaseDrain(); shutdowner.join(TIMEOUT_MS); assertThat(shutdowner.isAlive()).as("the shutdown thread did not complete").isFalse(); assertThat(shutdownFailure.get()).isNull(); - assertThat(replicaDBMonitorNames(DRAINED_SERVER_ID)) + assertThat(DirectoryServer.getMonitorProviders().keySet()) .as("the drained replica DB is still registered") - .isEmpty(); + .doesNotContain(replicaDBMonitorName(replicationServer, DRAINED_SERVER_ID)); } finally { if (changelogDB != null) { - changelogDB.releaseCreator(); - changelogDB.releaseDrain(); + changelogDB.releaseAllHeldThreads(); changelogDB.shutdownDB(); } join(creator); join(shutdowner); - // release what the unfixed code leaks, so that it does not outlive this test - for (String monitorName : replicaDBMonitorNames(RACING_SERVER_ID)) + deregisterLeakedReplicaDBMonitors(replicationServer); + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); + } + } + + /** + * A replica DB whose creation wins the race against {@code shutdownDB()} - i.e. reads the + * shutdown flag as {@code false} under the domain map monitor - must be shut down by the drain: + * the drain builds its iterator over {@code domainToReplicaDBs} after the flag is flipped, so it + * sees the domain map inserted before that monitor was taken, and blocks on the monitor until + * the creation has published the new replica DB. + *

+ * This is the branch the fix in {@code getExistingOrNewReplicaDB()} relies on: a drain rewritten + * to no longer traverse the map as it existed when the flag was flipped - snapshotting the keys + * beforehand, shutting the replication environment down first - would silently reintroduce the + * leak this test guards against. + *

+ * The interleaving is driven step by step: + *

    + *
  1. the creator thread creates its replica DB - the monitor provider is now registered - and + * is held before the DB is published into the domain map, still under the domain map + * monitor;
    2. + *
  2. the shutdown starts, flips the flag, and blocks on the domain map monitor the creator + * holds;
    3. + *
  3. the creator is released: it publishes the replica DB and exits the monitor, and the drain + * must then shut that replica DB down.
    4. + *
+ */ + @Test + public void replicaDBWinningTheRaceAgainstShutdownIsShutDownByTheDrain() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + RaceableChangelogDB changelogDB = null; + File testRoot = null; + Thread creator = null; + Thread shutdowner = null; + final AtomicReference creationFailure = new AtomicReference<>(); + final AtomicReference shutdownFailure = new AtomicReference<>(); + final AtomicReference createdReplicaDB = new AtomicReference<>(); + try + { + replicationServer = configureReplicationServer(100, 5000); + testRoot = createCleanDir("FileChangelogDB"); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB.initializeDB(); + + final FileChangelogDB racedChangelogDB = changelogDB; + final ReplicationServer racedReplicationServer = replicationServer; + changelogDB.holdNextReplicaDBOnceCreated(); + creator = new Thread("FileChangelogDBTest replica DB creator") + { + @Override + public void run() + { + try + { + createdReplicaDB.set(racedChangelogDB + .getOrCreateReplicaDB(TEST_ROOT_DN, RACING_SERVER_ID, racedReplicationServer).getFirst()); + } + catch (Throwable t) + { + creationFailure.set(t); + } + } + }; + creator.start(); + changelogDB.awaitCreatorHoldingItsCreatedReplicaDB(); + // asserted, so that the deregistration below cannot pass by never having seen a registration + assertThat(DirectoryServer.getMonitorProviders().keySet()) + .as("the racing replica DB is not registered") + .contains(replicaDBMonitorName(replicationServer, RACING_SERVER_ID)); + + shutdowner = new Thread("FileChangelogDBTest changelog shutdown") { - DirectoryServer.getMonitorProviders().remove(monitorName); + @Override + public void run() + { + try + { + racedChangelogDB.shutdownDB(); + } + catch (Throwable t) + { + shutdownFailure.set(t); + } + } + }; + shutdowner.start(); + awaitBlockedOnAMonitor(shutdowner); + + changelogDB.releaseCreatedReplicaDB(); + creator.join(TIMEOUT_MS); + shutdowner.join(TIMEOUT_MS); + assertThat(creator.isAlive()).as("the creator thread did not complete").isFalse(); + assertThat(shutdowner.isAlive()).as("the shutdown thread did not complete").isFalse(); + assertThat(creationFailure.get()).as("a creation which won the race must succeed").isNull(); + assertThat(createdReplicaDB.get()).as("the replica DB which won the race").isNotNull(); + assertThat(shutdownFailure.get()).isNull(); + assertThat(DirectoryServer.getMonitorProviders().keySet()) + .as("the replica DB which won the race is not shut down by the drain") + .doesNotContain(replicaDBMonitorName(replicationServer, RACING_SERVER_ID)); + } + finally + { + if (changelogDB != null) + { + changelogDB.releaseAllHeldThreads(); + changelogDB.shutdownDB(); } + join(creator); + join(shutdowner); + deregisterLeakedReplicaDBMonitors(replicationServer); remove(replicationServer); TestCaseUtils.deleteDirectory(testRoot); } } + /** Joins the provided thread, leaving a signal behind when it did not die within the timeout. */ private void join(final Thread thread) throws InterruptedException { if (thread != null) { thread.join(TIMEOUT_MS); + if (thread.isAlive()) + { + final IllegalStateException hung = new IllegalStateException("Test thread " + thread.getName() + + " is still alive after " + TIMEOUT_MS + " ms: it may leak a live changelog into later tests"); + hung.setStackTrace(thread.getStackTrace()); + hung.printStackTrace(); + thread.interrupt(); + } } } - /** Returns the names the replica DBs of the provided server id are registered under. */ - private List replicaDBMonitorNames(final int serverId) + /** + * Waits until the provided thread is blocked acquiring a monitor: the domain map monitor held by + * the creator is the only one it can stay blocked on - the other locks on its way to the drain + * are only transiently contended, hence the two consecutive observations. + */ + private static void awaitBlockedOnAMonitor(final Thread thread) throws InterruptedException { - final String prefix = "changelog for ds(" + serverId + ")"; - final List names = new ArrayList<>(); - for (String monitorName : DirectoryServer.getMonitorProviders().keySet()) + final long deadline = System.currentTimeMillis() + TIMEOUT_MS; + int blockedObservations = 0; + while (blockedObservations < 2) { - if (monitorName.startsWith(prefix)) + if (!thread.isAlive()) + { + throw new IllegalStateException(thread.getName() + " completed without blocking on the domain map monitor"); + } + if (System.currentTimeMillis() > deadline) { - names.add(monitorName); + throw new IllegalStateException( + "timed out waiting for " + thread.getName() + " to block on the domain map monitor"); } + blockedObservations = thread.getState() == Thread.State.BLOCKED ? blockedObservations + 1 : 0; + Thread.sleep(1); } - return names; - } - - private ReplicationServer configureReplicationServer() throws IOException, ConfigException - { - return new ReplicationServer( - new ReplServerFakeConfiguration(findFreePort(), null, 0, 2, 5000, 100, null)); } - private CryptoSuite createCryptoSuite() + /** + * Returns the name the monitor provider of the provided replica DB is registered under, i.e. the + * name built by {@code FileReplicaDB.DbMonitorProvider.getMonitorInstanceName()}, lower-cased + * the way {@code DirectoryServer.registerMonitorProvider()} stores it. + */ + private String replicaDBMonitorName(final ReplicationServer replicationServer, final int serverId) { - return getServerContext().getCryptoManager().newCryptoSuite(cipherTransformation, keyLength, false); + final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(TEST_ROOT_DN); + assertThat(domain).as("the domain scoping the monitor name of DS(" + serverId + ")").isNotNull(); + return toLowerCase("Changelog for DS(" + serverId + "),cn=" + domain.getMonitorInstanceName()); } - private File createCleanDir() throws IOException + /** Releases the monitor providers a regression leaks, so that they do not outlive this test. */ + private void deregisterLeakedReplicaDBMonitors(final ReplicationServer replicationServer) { - String buildRoot = System.getProperty(TestCaseUtils.PROPERTY_BUILD_ROOT); - String path = System.getProperty(TestCaseUtils.PROPERTY_BUILD_DIR, buildRoot - + File.separator + "build"); - path = path + File.separator + "unit-tests" + File.separator + "FileChangelogDB"; - final File testRoot = new File(path); - TestCaseUtils.deleteDirectory(testRoot); - testRoot.mkdirs(); - return testRoot; + if (replicationServer == null || replicationServer.getReplicationServerDomain(TEST_ROOT_DN) == null) + { + return; // no replica DB was ever created, hence no monitor provider was ever registered + } + for (final int serverId : new int[] { RACING_SERVER_ID, DRAINED_SERVER_ID }) + { + // deregister the provider instead of removing the map entry, so that the JMX MBean + // registered alongside it is released as well + final MonitorProvider provider = + DirectoryServer.getMonitorProviders().get(replicaDBMonitorName(replicationServer, serverId)); + if (provider != null) + { + DirectoryServer.deregisterMonitorProvider(provider); + } + } } /** * A changelog DB which lets a test hold a thread creating a replica DB right after it has read - * the shutdown flag, and hold the shutdown inside the drain of {@code domainToReplicaDBs}. + * the shutdown flag, hold it again once the replica DB is created but not yet published into the + * domain map, and hold the shutdown inside the drain of {@code domainToReplicaDBs}. */ private static final class RaceableChangelogDB extends FileChangelogDB { private final AtomicBoolean holdNextCreation = new AtomicBoolean(); - private final AtomicBoolean holdNextReplicaDB = new AtomicBoolean(); + private final AtomicBoolean holdNextReplicaDBShutdown = new AtomicBoolean(); + private final AtomicBoolean holdNextCreatedReplicaDB = new AtomicBoolean(); private final CountDownLatch creatorIsInWindow = new CountDownLatch(1); private final CountDownLatch creatorIsReleased = new CountDownLatch(1); + private final CountDownLatch creatorHoldsItsCreatedReplicaDB = new CountDownLatch(1); + private final CountDownLatch createdReplicaDBIsReleased = new CountDownLatch(1); private final CountDownLatch drainIsInReplicaDBShutdown = new CountDownLatch(1); private final CountDownLatch drainIsReleased = new CountDownLatch(1); @@ -268,11 +409,19 @@ ConcurrentMap getExistingOrNewDomainMap(final DN baseDN) FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException { - if (holdNextReplicaDB.compareAndSet(true, false)) + if (holdNextReplicaDBShutdown.compareAndSet(true, false)) { return new HeldOnShutdownReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); } - return super.newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + final FileReplicaDB replicaDB = super.newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + if (holdNextCreatedReplicaDB.compareAndSet(true, false)) + { + // the replica DB exists and its monitor provider is registered, but it is not published + // into the domain map yet: hold the creator there, under the domain map monitor + creatorHoldsItsCreatedReplicaDB.countDown(); + await(createdReplicaDBIsReleased); + } + return replicaDB; } void holdNextReplicaDBCreationBeforeItsDomainMapIsInserted() @@ -282,7 +431,12 @@ void holdNextReplicaDBCreationBeforeItsDomainMapIsInserted() void holdNextReplicaDBInItsShutdown() { - holdNextReplicaDB.set(true); + holdNextReplicaDBShutdown.set(true); + } + + void holdNextReplicaDBOnceCreated() + { + holdNextCreatedReplicaDB.set(true); } void awaitCreatorInWindow() @@ -290,6 +444,11 @@ void awaitCreatorInWindow() await(creatorIsInWindow); } + void awaitCreatorHoldingItsCreatedReplicaDB() + { + await(creatorHoldsItsCreatedReplicaDB); + } + void awaitDrainInReplicaDBShutdown() { await(drainIsInReplicaDBShutdown); @@ -300,11 +459,23 @@ void releaseCreator() creatorIsReleased.countDown(); } + void releaseCreatedReplicaDB() + { + createdReplicaDBIsReleased.countDown(); + } + void releaseDrain() { drainIsReleased.countDown(); } + void releaseAllHeldThreads() + { + releaseCreator(); + releaseCreatedReplicaDB(); + releaseDrain(); + } + /** A replica DB which holds the thread shutting it down until the test releases it. */ private final class HeldOnShutdownReplicaDB extends FileReplicaDB { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogTestFixtures.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogTestFixtures.java new file mode 100644 index 0000000000..1accbc800c --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogTestFixtures.java @@ -0,0 +1,68 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.replication.server.changelog.file; + +import java.io.File; +import java.io.IOException; + +import org.forgerock.opendj.config.server.ConfigException; +import org.opends.server.TestCaseUtils; +import org.opends.server.crypto.CryptoSuite; +import org.opends.server.replication.server.ReplServerFakeConfiguration; +import org.opends.server.replication.server.ReplicationServer; + +import static org.opends.server.TestCaseUtils.*; + +/** Fixtures shared by the tests of the file based changelog. */ +final class FileChangelogTestFixtures +{ + static final String CIPHER_TRANSFORMATION = "AES/CBC/PKCS5Padding"; + static final int KEY_LENGTH = 128; + + private FileChangelogTestFixtures() + { + // static helpers only + } + + /** Returns a replication server listening on a free port, with no connected replica. */ + static ReplicationServer configureReplicationServer(int windowSize, int queueSize) + throws IOException, ConfigException + { + final int changelogPort = findFreePort(); + ReplServerFakeConfiguration replServerFakeCfg = + new ReplServerFakeConfiguration(changelogPort, null, 0, 2, queueSize, windowSize, null); + return new ReplicationServer(replServerFakeCfg); + } + + /** Returns a crypto suite the changelog can encrypt its records with. */ + static CryptoSuite createCryptoSuite(boolean confidential) + { + return getServerContext().getCryptoManager().newCryptoSuite(CIPHER_TRANSFORMATION, KEY_LENGTH, confidential); + } + + /** Returns an empty directory of the provided name under the unit test build directory. */ + static File createCleanDir(String directoryName) throws IOException + { + String buildRoot = System.getProperty(TestCaseUtils.PROPERTY_BUILD_ROOT); + String path = System.getProperty(TestCaseUtils.PROPERTY_BUILD_DIR, buildRoot + + File.separator + "build"); + path = path + File.separator + "unit-tests" + File.separator + directoryName; + final File testRoot = new File(path); + TestCaseUtils.deleteDirectory(testRoot); + testRoot.mkdirs(); + return testRoot; + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileReplicaDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileReplicaDBTest.java index 17be42f639..e1dfa9f457 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileReplicaDBTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileReplicaDBTest.java @@ -12,16 +12,15 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server.changelog.file; import java.io.File; -import java.io.IOException; import java.util.ArrayList; import org.assertj.core.api.SoftAssertions; import org.forgerock.i18n.slf4j.LocalizedLogger; -import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.ByteString; import org.forgerock.opendj.ldap.DN; import org.forgerock.util.time.TimeService; @@ -32,7 +31,6 @@ import org.opends.server.replication.common.CSNGenerator; import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.protocol.UpdateMsg; -import org.opends.server.replication.server.ReplServerFakeConfiguration; import org.opends.server.replication.server.ReplicationServer; import org.opends.server.replication.server.changelog.api.ChangelogException; import org.opends.server.replication.server.changelog.api.DBCursor; @@ -45,6 +43,7 @@ import static org.opends.server.TestCaseUtils.*; import static org.opends.server.replication.server.changelog.api.DBCursor.KeyMatchingStrategy.*; import static org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy.*; +import static org.opends.server.replication.server.changelog.file.FileChangelogTestFixtures.*; import static org.opends.server.util.CollectionUtils.*; import static org.testng.Assert.*; @@ -55,8 +54,6 @@ public class FileReplicaDBTest extends ReplicationTestCase { private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); - private final String cipherTransformation = "AES/CBC/PKCS5Padding"; - private final int keyLength = 128; private DN TEST_ROOT_DN; /** @@ -105,16 +102,12 @@ public void testRecordEncodingWithAndWithoutConfidentiality(UpdateMsg msg, boole RecordParser parser = FileReplicaDB.newReplicaDBParser(cryptoSuite); ByteString data1 = parser.encodeRecord(Record.from(msg.getCSN(), msg)); - cryptoSuite.newParameters(cipherTransformation, keyLength, !confidential); + cryptoSuite.newParameters(CIPHER_TRANSFORMATION, KEY_LENGTH, !confidential); ByteString data2 = parser.encodeRecord(Record.from(msg.getCSN(), msg)); assertFalse(data1.equals(data2)); } - private CryptoSuite createCryptoSuite(boolean confidential) - { - return getServerContext().getCryptoManager().newCryptoSuite(cipherTransformation, keyLength, confidential); - } @Test public void testDomainDNWithForwardSlashes() throws Exception { @@ -426,7 +419,7 @@ private void testGetOldestNewestCSNs(final int max, final int counterWindow) thr TestCaseUtils.startServer(); replicationServer = configureReplicationServer(100000, 10); - testRoot = createCleanDir(); + testRoot = createCleanDir("FileReplicaDB"); dbEnv = new ReplicationEnvironment(testRoot.getPath(), replicationServer, TimeService.SYSTEM); replicaDB = new FileReplicaDB(1, TEST_ROOT_DN, replicationServer, createCryptoSuite(false), dbEnv); @@ -538,33 +531,12 @@ private void waitChangesArePersisted(FileReplicaDB replicaDB, assertEquals(replicaDB.getNumberRecords(), expectedNbRecords); } - private ReplicationServer configureReplicationServer(int windowSize, int queueSize) - throws IOException, ConfigException - { - final int changelogPort = findFreePort(); - ReplServerFakeConfiguration replServerFakeCfg = - new ReplServerFakeConfiguration(changelogPort, null, 0, 2, queueSize, windowSize, null); - return new ReplicationServer(replServerFakeCfg); - } - private FileReplicaDB newReplicaDB(ReplicationServer rs) throws Exception { final FileChangelogDB changelogDB = (FileChangelogDB) rs.getChangelogDB(); return changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, 1, rs).getFirst(); } - private File createCleanDir() throws IOException - { - String buildRoot = System.getProperty(TestCaseUtils.PROPERTY_BUILD_ROOT); - String path = System.getProperty(TestCaseUtils.PROPERTY_BUILD_DIR, buildRoot - + File.separator + "build"); - path = path + File.separator + "unit-tests" + File.separator + "FileReplicaDB"; - final File testRoot = new File(path); - TestCaseUtils.deleteDirectory(testRoot); - testRoot.mkdirs(); - return testRoot; - } - private void assertFoundInOrder(FileReplicaDB replicaDB, CSN... csns) throws Exception { if (csns.length == 0) From b0d145d1fbe6fe56697a75b926ce15d4a8cdc07a Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 4 Aug 2026 14:30:07 +0300 Subject: [PATCH 3/6] [#818] Drop the empty domain map a replica DB creation leaves behind when it bails out getOrCreateReplicaDB() inserts a domain map, and announces the domain to every registered multi domain cursor, before it knows whether it will create anything in it. When the creation then does not happen - the domain map was concurrently removed, or the FileReplicaDB constructor threw - the inserted empty map stayed in domainToReplicaDBs for the lifetime of the changelog: every multi domain cursor created afterwards walked a domain holding no replica DB at all, and clearDB() reached clearGenerationId() for a domain the changelog held nothing for. Remove the map on the way out of the guarded creation block, but only when it is empty: a populated map must stay mapped for the drain of shutdownDB() to find, even when the creation of this serverId failed - dropping it would hide its replica DBs from the weakly consistent drain iterator and leak them. --- .../changelog/file/FileChangelogDB.java | 57 ++++--- .../changelog/file/FileChangelogDBTest.java | 147 ++++++++++++++++++ 2 files changed, 183 insertions(+), 21 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index 3ac6fe2b26..f863eb992f 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -275,31 +275,46 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM return Pair.of(currentValue, false); } - if (domainToReplicaDBs.get(baseDN) != domainMap) + try { - // The domainMap could have been concurrently removed because - // 1) a shutdown was initiated or 2) an initialize was called. - // Return will allow the code to: - // 1) shutdown properly or 2) lazily recreate the replicaDB - return null; - } + if (domainToReplicaDBs.get(baseDN) != domainMap) + { + // The domainMap could have been concurrently removed because + // 1) a shutdown was initiated or 2) an initialize was called. + // Return will allow the code to: + // 1) shutdown properly or 2) lazily recreate the replicaDB + return null; + } + + if (shutdown.get()) + { + // A shutdown was initiated after the shutdown flag was read by getOrCreateReplicaDB(): + // it may already have drained domainToReplicaDBs before this domainMap was inserted into + // it, in which case nothing would ever shutdown a replicaDB created here. + // Reading false instead means shutdownDB() has not flipped the flag yet, hence has not + // created its iterator yet either: since ConcurrentHashMap iterators traverse the + // elements as they existed upon construction of the iterator, it will see this domainMap, + // which was inserted before this monitor was acquired, and will have to block on this + // same monitor to drain it. + return null; + } - if (shutdown.get()) + final FileReplicaDB newDB = newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + domainMap.put(serverId, newDB); + return Pair.of(newDB, true); + } + finally { - // A shutdown was initiated after the shutdown flag was read by getOrCreateReplicaDB(): - // it may already have drained domainToReplicaDBs before this domainMap was inserted into - // it, in which case nothing would ever shutdown a replicaDB created here. - // Reading false instead means shutdownDB() has not flipped the flag yet, hence has not - // created its iterator yet either: since ConcurrentHashMap iterators traverse the - // elements as they existed upon construction of the iterator, it will see this domainMap, - // which was inserted before this monitor was acquired, and will have to block on this - // same monitor to drain it. - return null; + // Leaving without having created the replica DB must not leave behind the empty domainMap + // inserted by getExistingOrNewDomainMap(): nothing would ever remove it, and every multi + // domain cursor created afterwards would walk a domain holding no replica DB at all. + // Only an empty map may be dropped: a populated one must stay mapped for the drain of + // shutdownDB() to find, even when the creation of this serverId failed. + if (domainMap.isEmpty()) + { + domainToReplicaDBs.remove(baseDN, domainMap); + } } - - final FileReplicaDB newDB = newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); - domainMap.put(serverId, newDB); - return Pair.of(newDB, true); } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java index 67c0ea128a..b3d0ca7ef4 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java @@ -16,6 +16,7 @@ package org.opends.server.replication.server.changelog.file; import java.io.File; +import java.lang.reflect.Field; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -23,9 +24,11 @@ import java.util.concurrent.atomic.AtomicReference; import org.assertj.core.api.SoftAssertions; +import org.forgerock.i18n.LocalizableMessage; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.DN; import org.forgerock.opendj.server.config.server.MonitorProviderCfg; +import org.forgerock.util.Pair; import org.opends.server.TestCaseUtils; import org.opends.server.api.MonitorProvider; import org.opends.server.core.DirectoryServer; @@ -53,8 +56,15 @@ public class FileChangelogDBTest extends ReplicationTestCase private static final int DRAINED_SERVER_ID = 814; /** Server id of the replica DB whose creation races that drain. */ private static final int RACING_SERVER_ID = 813; + /** Server id of a replica DB created before the failing one, in the same domain. */ + private static final int EXISTING_SERVER_ID = 817; + /** Server id of the replica DB whose creation is made to fail. */ + private static final int FAILING_SERVER_ID = 818; private static final long TIMEOUT_MS = 30000; + private static final LocalizableMessage CREATION_FAILURE = + LocalizableMessage.raw("FileChangelogDBTest replica DB creation failure"); + private DN TEST_ROOT_DN; @BeforeClass @@ -297,6 +307,115 @@ public void run() } } + /** + * A replica DB creation which fails must not leave behind the empty domain map it inserted: + * nothing would ever remove it from {@code domainToReplicaDBs}, and every multi domain cursor + * created afterwards would walk a domain holding no replica DB at all. + */ + @Test + public void failedReplicaDBCreationDropsTheDomainMapItInserted() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + FailingChangelogDB changelogDB = null; + File testRoot = null; + try + { + replicationServer = configureReplicationServer(100, 5000); + testRoot = createCleanDir("FileChangelogDB"); + changelogDB = new FailingChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB.initializeDB(); + + changelogDB.failNextReplicaDBCreation(); + try + { + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer); + failBecauseExceptionWasNotThrown(ChangelogException.class); + } + catch (ChangelogException expected) + { + assertThat(expected).hasMessage(CREATION_FAILURE.toString()); + } + assertThat(domainToReplicaDBs(changelogDB)) + .as("the empty domain map inserted for the creation which failed") + .doesNotContainKey(TEST_ROOT_DN); + + // the next creation starts from scratch and repopulates the domain + final Pair result = + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer); + assertThat(result.getSecond()).as("the replica DB was created anew").isTrue(); + assertThat(domainToReplicaDBs(changelogDB).get(TEST_ROOT_DN)).containsOnlyKeys(FAILING_SERVER_ID); + } + finally + { + if (changelogDB != null) + { + changelogDB.shutdownDB(); + } + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); + } + } + + /** + * A replica DB creation which fails must only drop an empty domain map: a populated one + * must stay mapped, so that the drain of {@code shutdownDB()} finds the replica DBs it holds and + * shuts them down. + */ + @Test + public void failedReplicaDBCreationKeepsAPopulatedDomainMap() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + FailingChangelogDB changelogDB = null; + File testRoot = null; + try + { + replicationServer = configureReplicationServer(100, 5000); + testRoot = createCleanDir("FileChangelogDB"); + changelogDB = new FailingChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB.initializeDB(); + + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, EXISTING_SERVER_ID, replicationServer); + final ConcurrentMap domainMap = domainToReplicaDBs(changelogDB).get(TEST_ROOT_DN); + + changelogDB.failNextReplicaDBCreation(); + try + { + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer); + failBecauseExceptionWasNotThrown(ChangelogException.class); + } + catch (ChangelogException expected) + { + assertThat(expected).hasMessage(CREATION_FAILURE.toString()); + } + assertThat(domainToReplicaDBs(changelogDB).get(TEST_ROOT_DN)) + .as("the domain map holding the replica DB created before the failure") + .isSameAs(domainMap) + .containsOnlyKeys(EXISTING_SERVER_ID); + } + finally + { + if (changelogDB != null) + { + changelogDB.shutdownDB(); + } + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); + } + } + + @SuppressWarnings("unchecked") + private static ConcurrentMap> domainToReplicaDBs( + final FileChangelogDB changelogDB) throws Exception + { + final Field field = FileChangelogDB.class.getDeclaredField("domainToReplicaDBs"); + field.setAccessible(true); + return (ConcurrentMap>) field.get(changelogDB); + } + /** Joins the provided thread, leaving a signal behind when it did not die within the timeout. */ private void join(final Thread thread) throws InterruptedException { @@ -510,4 +629,32 @@ private static void await(final CountDownLatch latch) } } } + + /** A changelog DB which lets a test make the next replica DB creation fail. */ + private static final class FailingChangelogDB extends FileChangelogDB + { + private final AtomicBoolean failNextCreation = new AtomicBoolean(); + + FailingChangelogDB(final ReplicationServer replicationServer, final String dbDirectoryPath, + final CryptoSuite cryptoSuite) throws ConfigException + { + super(replicationServer, dbDirectoryPath, cryptoSuite); + } + + @Override + FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, + final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException + { + if (failNextCreation.compareAndSet(true, false)) + { + throw new ChangelogException(CREATION_FAILURE); + } + return super.newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); + } + + void failNextReplicaDBCreation() + { + failNextCreation.set(true); + } + } } From b12ced81ce78d83982988c5bca8846acf1cff49f Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 4 Aug 2026 14:30:31 +0300 Subject: [PATCH 4/6] [#818] Address review feedback on the empty domain map cleanup The identity check stays above the cleanup: ConcurrentMap.remove(key, value) compares by equals() and two empty maps are equal, so a bail-out whose domain map was already unmapped would have dropped the fresh, still empty map of a concurrent creation - unmapping the replica DB about to be published into it and reintroducing the leak of #813. The bail-out path now cleans up nothing; the cleanup runs only on the creation path, where the mapping is pinned by the monitor every removal site takes. The domainToReplicaDBs javadoc documents this third removal site. Test coverage of the review findings: - bailOutMustNotUnmapAnotherThreadsFreshDomainMap drives that interleaving deterministically, and fails against the reviewed shape of the cleanup; - the failed-creation test also asserts the symptom of #818, through the domains a multi domain cursor created after the failure asks to open; - the tests observe domainToReplicaDBs through a package private accessor instead of reflection, share the RaceableChangelogDB test double, and their cleanup no longer skips the replication server removal when shutdownDB() throws. --- .../changelog/file/FileChangelogDB.java | 44 ++- .../changelog/file/FileChangelogDBTest.java | 324 ++++++++++++++---- 2 files changed, 296 insertions(+), 72 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index f863eb992f..af2e99580e 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -81,6 +81,12 @@ public class FileChangelogDB implements ChangelogDB, ReplicationDomainDB * * When creating a replicaDB, synchronize on the domainMap to avoid * concurrent shutdown. + *

+ * A creation which bails out while holding the domainMap monitor removes the still empty + * domainMap it inserted, so that no phantom domain is left behind. It may only do so after + * checking, under that same monitor, that the domainMap is still the one mapped to its baseDN: + * the removal is equality based and two empty maps are equal, so without the identity check it + * could unmap the fresh domainMap of another creation. */ private final ConcurrentMap> domainToReplicaDBs = new ConcurrentHashMap<>(); @@ -275,17 +281,19 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM return Pair.of(currentValue, false); } - try + if (domainToReplicaDBs.get(baseDN) != domainMap) { - if (domainToReplicaDBs.get(baseDN) != domainMap) - { - // The domainMap could have been concurrently removed because - // 1) a shutdown was initiated or 2) an initialize was called. - // Return will allow the code to: - // 1) shutdown properly or 2) lazily recreate the replicaDB - return null; - } + // The domainMap could have been concurrently removed because + // 1) a shutdown was initiated or 2) an initialize was called. + // Return will allow the code to: + // 1) shutdown properly or 2) lazily recreate the replicaDB + // There is nothing to clean up here: this domainMap is already unmapped, and whatever map + // is now associated to baseDN belongs to another creation. + return null; + } + try + { if (shutdown.get()) { // A shutdown was initiated after the shutdown flag was read by getOrCreateReplicaDB(): @@ -310,6 +318,12 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM // domain cursor created afterwards would walk a domain holding no replica DB at all. // Only an empty map may be dropped: a populated one must stay mapped for the drain of // shutdownDB() to find, even when the creation of this serverId failed. + // The identity check above passed under this monitor, and every removal site takes the + // monitor of the map it unmaps before removing it: the mapping cannot have changed since, + // so this equality based remove provably drops this domainMap and no other. + // Unlike removeDomain(), this path clears no ChangeNumberIndexer state, and a later + // creation broadcasts addDomain() anew to multi domain cursors which already incorporated + // the domain: a pre-existing hazard of repeated addDomain() calls, unchanged here. if (domainMap.isEmpty()) { domainToReplicaDBs.remove(baseDN, domainMap); @@ -344,6 +358,18 @@ FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final Replicatio return new FileReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); } + /** + * Returns the map of replica DBs per domain. + *

+ * Package private so that tests can observe which domain maps this changelog holds. + * + * @return the map of replica DBs per domain + */ + ConcurrentMap> getDomainToReplicaDBs() + { + return domainToReplicaDBs; + } + @Override public void initializeDB() { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java index b3d0ca7ef4..3156a743ee 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java @@ -16,8 +16,9 @@ package org.opends.server.replication.server.changelog.file; import java.io.File; -import java.lang.reflect.Field; +import java.util.Set; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -34,15 +35,22 @@ import org.opends.server.core.DirectoryServer; import org.opends.server.crypto.CryptoSuite; import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.MultiDomainServerState; +import org.opends.server.replication.common.ServerState; +import org.opends.server.replication.protocol.UpdateMsg; import org.opends.server.replication.server.ReplicationServer; import org.opends.server.replication.server.ReplicationServerDomain; import org.opends.server.replication.server.changelog.api.ChangelogException; +import org.opends.server.replication.server.changelog.api.DBCursor; +import org.opends.server.replication.server.changelog.api.DBCursor.CursorOptions; import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; import static org.assertj.core.api.Assertions.*; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.TestCaseUtils.*; +import static org.opends.server.replication.server.changelog.api.DBCursor.KeyMatchingStrategy.*; +import static org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy.*; import static org.opends.server.replication.server.changelog.file.FileChangelogTestFixtures.*; import static org.opends.server.util.StaticUtils.toLowerCase; @@ -56,6 +64,10 @@ public class FileChangelogDBTest extends ReplicationTestCase private static final int DRAINED_SERVER_ID = 814; /** Server id of the replica DB whose creation races that drain. */ private static final int RACING_SERVER_ID = 813; + /** Server id of the replica DB whose creation bails out on the identity check. */ + private static final int STALE_SERVER_ID = 815; + /** Server id of the replica DB created in the fresh domain map the bail-out must not unmap. */ + private static final int FRESH_SERVER_ID = 816; /** Server id of a replica DB created before the failing one, in the same domain. */ private static final int EXISTING_SERVER_ID = 817; /** Server id of the replica DB whose creation is made to fail. */ @@ -187,7 +199,7 @@ public void run() } join(creator); join(shutdowner); - deregisterLeakedReplicaDBMonitors(replicationServer); + deregisterLeakedReplicaDBMonitors(replicationServer, RACING_SERVER_ID, DRAINED_SERVER_ID); remove(replicationServer); TestCaseUtils.deleteDirectory(testRoot); } @@ -301,7 +313,7 @@ public void run() } join(creator); join(shutdowner); - deregisterLeakedReplicaDBMonitors(replicationServer); + deregisterLeakedReplicaDBMonitors(replicationServer, RACING_SERVER_ID, DRAINED_SERVER_ID); remove(replicationServer); TestCaseUtils.deleteDirectory(testRoot); } @@ -310,7 +322,8 @@ public void run() /** * A replica DB creation which fails must not leave behind the empty domain map it inserted: * nothing would ever remove it from {@code domainToReplicaDBs}, and every multi domain cursor - * created afterwards would walk a domain holding no replica DB at all. + * created afterwards would walk a domain holding no replica DB at all - the symptom this test + * also asserts, through the domains a new multi domain cursor asks the changelog to open. */ @Test public void failedReplicaDBCreationDropsTheDomainMapItInserted() throws Exception @@ -318,13 +331,13 @@ public void failedReplicaDBCreationDropsTheDomainMapItInserted() throws Exceptio TestCaseUtils.startServer(); ReplicationServer replicationServer = null; - FailingChangelogDB changelogDB = null; + RaceableChangelogDB changelogDB = null; File testRoot = null; try { replicationServer = configureReplicationServer(100, 5000); testRoot = createCleanDir("FileChangelogDB"); - changelogDB = new FailingChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); changelogDB.initializeDB(); changelogDB.failNextReplicaDBCreation(); @@ -337,24 +350,47 @@ public void failedReplicaDBCreationDropsTheDomainMapItInserted() throws Exceptio { assertThat(expected).hasMessage(CREATION_FAILURE.toString()); } - assertThat(domainToReplicaDBs(changelogDB)) + assertThat(changelogDB.getDomainToReplicaDBs()) .as("the empty domain map inserted for the creation which failed") .doesNotContainKey(TEST_ROOT_DN); + // the symptom of the leftover map: a multi domain cursor created after the failure must not + // walk the phantom domain + changelogDB.walkedDomains.clear(); + final MultiDomainDBCursor cursor = changelogDB.getCursorFrom( + new MultiDomainServerState(), new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY)); + try + { + cursor.next(); + assertThat(changelogDB.walkedDomains) + .as("the domains walked by a multi domain cursor created after the failed creation") + .doesNotContain(TEST_ROOT_DN); + } + finally + { + cursor.close(); + } + // the next creation starts from scratch and repopulates the domain final Pair result = changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer); assertThat(result.getSecond()).as("the replica DB was created anew").isTrue(); - assertThat(domainToReplicaDBs(changelogDB).get(TEST_ROOT_DN)).containsOnlyKeys(FAILING_SERVER_ID); + assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN)).containsOnlyKeys(FAILING_SERVER_ID); } finally { - if (changelogDB != null) + try { - changelogDB.shutdownDB(); + if (changelogDB != null) + { + changelogDB.shutdownDB(); + } + } + finally + { + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); } - remove(replicationServer); - TestCaseUtils.deleteDirectory(testRoot); } } @@ -369,17 +405,18 @@ public void failedReplicaDBCreationKeepsAPopulatedDomainMap() throws Exception TestCaseUtils.startServer(); ReplicationServer replicationServer = null; - FailingChangelogDB changelogDB = null; + RaceableChangelogDB changelogDB = null; File testRoot = null; try { replicationServer = configureReplicationServer(100, 5000); testRoot = createCleanDir("FileChangelogDB"); - changelogDB = new FailingChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); changelogDB.initializeDB(); changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, EXISTING_SERVER_ID, replicationServer); - final ConcurrentMap domainMap = domainToReplicaDBs(changelogDB).get(TEST_ROOT_DN); + final ConcurrentMap domainMap = + changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN); changelogDB.failNextReplicaDBCreation(); try @@ -391,29 +428,155 @@ public void failedReplicaDBCreationKeepsAPopulatedDomainMap() throws Exception { assertThat(expected).hasMessage(CREATION_FAILURE.toString()); } - assertThat(domainToReplicaDBs(changelogDB).get(TEST_ROOT_DN)) + assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN)) .as("the domain map holding the replica DB created before the failure") .isSameAs(domainMap) .containsOnlyKeys(EXISTING_SERVER_ID); } finally { - if (changelogDB != null) + try { - changelogDB.shutdownDB(); + if (changelogDB != null) + { + changelogDB.shutdownDB(); + } + } + finally + { + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); } - remove(replicationServer); - TestCaseUtils.deleteDirectory(testRoot); } } - @SuppressWarnings("unchecked") - private static ConcurrentMap> domainToReplicaDBs( - final FileChangelogDB changelogDB) throws Exception + /** + * A creation which bails out on the identity check must not drop the domain map another creation + * has freshly inserted: the drop is equality based and two empty maps are equal, so an identity + * unaware cleanup would unmap the fresh map, and the replica DB about to be published into it + * would no longer be reachable from {@code domainToReplicaDBs} - nothing would ever shut it + * down, which is the leak of #813 all over again. + *

+ * The interleaving is driven step by step: + *

    + *
  1. the stale creator obtains its domain map and is held before entering its monitor;
    2. + *
  2. {@code removeDomain()} unmaps that domain map;
    3. + *
  3. a fresh creator inserts a new, still empty domain map and is held inside + * {@code newReplicaDB()}, under the fresh map's monitor;
    4. + *
  4. the stale creator is released: its identity check fails and it must bail out without + * touching the fresh map, then retry and block on the fresh map's monitor;
    5. + *
  5. the fresh creator is released: both creations complete into that same map.
    6. + *
+ */ + @Test + public void bailOutMustNotUnmapAnotherThreadsFreshDomainMap() throws Exception { - final Field field = FileChangelogDB.class.getDeclaredField("domainToReplicaDBs"); - field.setAccessible(true); - return (ConcurrentMap>) field.get(changelogDB); + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + RaceableChangelogDB changelogDB = null; + File testRoot = null; + Thread staleCreator = null; + Thread freshCreator = null; + final AtomicReference staleCreationFailure = new AtomicReference<>(); + final AtomicReference freshCreationFailure = new AtomicReference<>(); + try + { + replicationServer = configureReplicationServer(100, 5000); + testRoot = createCleanDir("FileChangelogDB"); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB.initializeDB(); + + final RaceableChangelogDB racedChangelogDB = changelogDB; + final ReplicationServer racedReplicationServer = replicationServer; + + // 1- the stale creator obtains the domain map about to be unmapped, and is parked there + changelogDB.holdNextCreationAfterItsDomainMapIsObtained(); + staleCreator = new Thread("FileChangelogDBTest stale replica DB creator") + { + @Override + public void run() + { + try + { + racedChangelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, STALE_SERVER_ID, racedReplicationServer); + } + catch (Throwable t) + { + staleCreationFailure.set(t); + } + } + }; + staleCreator.start(); + changelogDB.awaitCreatorHoldingItsDomainMap(); + + // 2- the domain map the stale creator holds is unmapped + changelogDB.removeDomain(TEST_ROOT_DN); + + // 3- the fresh creator inserts a new, still empty domain map, and is parked inside + // newReplicaDB(), under the monitor of that fresh map + changelogDB.holdNextReplicaDBOnceCreated(); + freshCreator = new Thread("FileChangelogDBTest fresh replica DB creator") + { + @Override + public void run() + { + try + { + racedChangelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FRESH_SERVER_ID, racedReplicationServer); + } + catch (Throwable t) + { + freshCreationFailure.set(t); + } + } + }; + freshCreator.start(); + changelogDB.awaitCreatorHoldingItsCreatedReplicaDB(); + final ConcurrentMap freshDomainMap = + changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN); + assertThat(freshDomainMap).as("the fresh domain map, not published into yet").isNotNull().isEmpty(); + + // 4- the stale creator bails out on its identity check, retries, and blocks on the monitor + // of the fresh domain map - without the identity check its cleanup would have unmapped the + // fresh map, and it would have completed into a third map instead of blocking + changelogDB.releaseCreatorHoldingItsDomainMap(); + awaitBlockedOnAMonitorOrCompleted(staleCreator); + assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN)) + .as("the domain map the fresh creator is about to publish its replica DB into") + .isSameAs(freshDomainMap); + + // 5- both creations complete into that same map + changelogDB.releaseCreatedReplicaDB(); + staleCreator.join(TIMEOUT_MS); + freshCreator.join(TIMEOUT_MS); + assertThat(staleCreator.isAlive()).as("the stale creator thread did not complete").isFalse(); + assertThat(freshCreator.isAlive()).as("the fresh creator thread did not complete").isFalse(); + assertThat(staleCreationFailure.get()).isNull(); + assertThat(freshCreationFailure.get()).isNull(); + assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN)) + .isSameAs(freshDomainMap) + .containsOnlyKeys(STALE_SERVER_ID, FRESH_SERVER_ID); + } + finally + { + try + { + if (changelogDB != null) + { + changelogDB.releaseAllHeldThreads(); + changelogDB.shutdownDB(); + } + } + finally + { + join(staleCreator); + join(freshCreator); + deregisterLeakedReplicaDBMonitors(replicationServer, STALE_SERVER_ID, FRESH_SERVER_ID); + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); + } + } } /** Joins the provided thread, leaving a signal behind when it did not die within the timeout. */ @@ -434,11 +597,26 @@ private void join(final Thread thread) throws InterruptedException } /** - * Waits until the provided thread is blocked acquiring a monitor: the domain map monitor held by - * the creator is the only one it can stay blocked on - the other locks on its way to the drain - * are only transiently contended, hence the two consecutive observations. + * Waits until the provided thread is blocked acquiring a monitor: the domain map monitor is the + * only one it can stay blocked on - the other locks on its way are only transiently contended, + * hence the two consecutive observations. A thread expected to block but completing instead is + * an error. */ private static void awaitBlockedOnAMonitor(final Thread thread) throws InterruptedException + { + awaitBlockedOnAMonitorOrCompleted(thread); + if (!thread.isAlive()) + { + throw new IllegalStateException(thread.getName() + " completed without blocking on the domain map monitor"); + } + } + + /** + * Waits until the provided thread is durably blocked acquiring a monitor - two consecutive + * observations, since the other locks on its way are only transiently contended - or has + * completed: completion is left for the caller's assertions to diagnose. + */ + private static void awaitBlockedOnAMonitorOrCompleted(final Thread thread) throws InterruptedException { final long deadline = System.currentTimeMillis() + TIMEOUT_MS; int blockedObservations = 0; @@ -446,7 +624,7 @@ private static void awaitBlockedOnAMonitor(final Thread thread) throws Interrupt { if (!thread.isAlive()) { - throw new IllegalStateException(thread.getName() + " completed without blocking on the domain map monitor"); + return; } if (System.currentTimeMillis() > deadline) { @@ -471,13 +649,13 @@ private String replicaDBMonitorName(final ReplicationServer replicationServer, f } /** Releases the monitor providers a regression leaks, so that they do not outlive this test. */ - private void deregisterLeakedReplicaDBMonitors(final ReplicationServer replicationServer) + private void deregisterLeakedReplicaDBMonitors(final ReplicationServer replicationServer, final int... serverIds) { if (replicationServer == null || replicationServer.getReplicationServerDomain(TEST_ROOT_DN) == null) { return; // no replica DB was ever created, hence no monitor provider was ever registered } - for (final int serverId : new int[] { RACING_SERVER_ID, DRAINED_SERVER_ID }) + for (final int serverId : serverIds) { // deregister the provider instead of removing the map entry, so that the JMX MBean // registered alongside it is released as well @@ -492,21 +670,30 @@ private void deregisterLeakedReplicaDBMonitors(final ReplicationServer replicati /** * A changelog DB which lets a test hold a thread creating a replica DB right after it has read - * the shutdown flag, hold it again once the replica DB is created but not yet published into the - * domain map, and hold the shutdown inside the drain of {@code domainToReplicaDBs}. + * the shutdown flag, hold it after it has obtained its domain map but before it enters the + * monitor, hold it again once the replica DB is created but not yet published into the domain + * map, hold the shutdown inside the drain of {@code domainToReplicaDBs}, make the next replica + * DB creation fail, and record the domains cursors are opened for. */ private static final class RaceableChangelogDB extends FileChangelogDB { private final AtomicBoolean holdNextCreation = new AtomicBoolean(); + private final AtomicBoolean holdNextDomainMapObtained = new AtomicBoolean(); private final AtomicBoolean holdNextReplicaDBShutdown = new AtomicBoolean(); private final AtomicBoolean holdNextCreatedReplicaDB = new AtomicBoolean(); + private final AtomicBoolean failNextCreation = new AtomicBoolean(); private final CountDownLatch creatorIsInWindow = new CountDownLatch(1); private final CountDownLatch creatorIsReleased = new CountDownLatch(1); + private final CountDownLatch creatorHoldsItsDomainMap = new CountDownLatch(1); + private final CountDownLatch domainMapIsReleased = new CountDownLatch(1); private final CountDownLatch creatorHoldsItsCreatedReplicaDB = new CountDownLatch(1); private final CountDownLatch createdReplicaDBIsReleased = new CountDownLatch(1); private final CountDownLatch drainIsInReplicaDBShutdown = new CountDownLatch(1); private final CountDownLatch drainIsReleased = new CountDownLatch(1); + /** The baseDNs of the domains any cursor was opened for. */ + private final Set walkedDomains = new CopyOnWriteArraySet<>(); + RaceableChangelogDB(final ReplicationServer replicationServer, final String dbDirectoryPath, final CryptoSuite cryptoSuite) throws ConfigException { @@ -521,13 +708,23 @@ ConcurrentMap getExistingOrNewDomainMap(final DN baseDN) creatorIsInWindow.countDown(); await(creatorIsReleased); } - return super.getExistingOrNewDomainMap(baseDN); + final ConcurrentMap domainMap = super.getExistingOrNewDomainMap(baseDN); + if (holdNextDomainMapObtained.compareAndSet(true, false)) + { + creatorHoldsItsDomainMap.countDown(); + await(domainMapIsReleased); + } + return domainMap; } @Override FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException { + if (failNextCreation.compareAndSet(true, false)) + { + throw new ChangelogException(CREATION_FAILURE); + } if (holdNextReplicaDBShutdown.compareAndSet(true, false)) { return new HeldOnShutdownReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); @@ -543,11 +740,24 @@ FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final Replicatio return replicaDB; } + @Override + public DBCursor getCursorFrom(final DN baseDN, final ServerState startState, + final CursorOptions options) throws ChangelogException + { + walkedDomains.add(baseDN); + return super.getCursorFrom(baseDN, startState, options); + } + void holdNextReplicaDBCreationBeforeItsDomainMapIsInserted() { holdNextCreation.set(true); } + void holdNextCreationAfterItsDomainMapIsObtained() + { + holdNextDomainMapObtained.set(true); + } + void holdNextReplicaDBInItsShutdown() { holdNextReplicaDBShutdown.set(true); @@ -558,11 +768,21 @@ void holdNextReplicaDBOnceCreated() holdNextCreatedReplicaDB.set(true); } + void failNextReplicaDBCreation() + { + failNextCreation.set(true); + } + void awaitCreatorInWindow() { await(creatorIsInWindow); } + void awaitCreatorHoldingItsDomainMap() + { + await(creatorHoldsItsDomainMap); + } + void awaitCreatorHoldingItsCreatedReplicaDB() { await(creatorHoldsItsCreatedReplicaDB); @@ -578,6 +798,11 @@ void releaseCreator() creatorIsReleased.countDown(); } + void releaseCreatorHoldingItsDomainMap() + { + domainMapIsReleased.countDown(); + } + void releaseCreatedReplicaDB() { createdReplicaDBIsReleased.countDown(); @@ -591,6 +816,7 @@ void releaseDrain() void releaseAllHeldThreads() { releaseCreator(); + releaseCreatorHoldingItsDomainMap(); releaseCreatedReplicaDB(); releaseDrain(); } @@ -629,32 +855,4 @@ private static void await(final CountDownLatch latch) } } } - - /** A changelog DB which lets a test make the next replica DB creation fail. */ - private static final class FailingChangelogDB extends FileChangelogDB - { - private final AtomicBoolean failNextCreation = new AtomicBoolean(); - - FailingChangelogDB(final ReplicationServer replicationServer, final String dbDirectoryPath, - final CryptoSuite cryptoSuite) throws ConfigException - { - super(replicationServer, dbDirectoryPath, cryptoSuite); - } - - @Override - FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server, - final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException - { - if (failNextCreation.compareAndSet(true, false)) - { - throw new ChangelogException(CREATION_FAILURE); - } - return super.newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv); - } - - void failNextReplicaDBCreation() - { - failNextCreation.set(true); - } - } } From 07f68040ebcbac257a38984c37f2c9763ccc4782 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 4 Aug 2026 19:28:18 +0300 Subject: [PATCH 5/6] [#818] Make domain announcements idempotent and the drain protocol conforming The cleanup unmaps a domain while the server is live and without clearing the ChangeNumberIndexer, so the next successful creation announced the domain a second time to every live multi domain cursor: the cursor built a second DomainDBCursor over the same domain, which either leaked unclosed - the cursor tree collapses cursors comparing equal - or delivered every change twice, killing the ChangeNumberIndexer thread with the IllegalStateException its cookie update throws on a replayed change. MultiDomainDBCursor now tracks the domains it already iterates over: addDomain() ignores them, and incorporation double-checks, since an announcement racing an incorporation can still queue a domain twice. New replica DBs of an incorporated domain keep reaching the cursor through addReplicaDB(). A new test announces a domain to a live cursor twice, and fails without this fix on the second cursor it then opens. The shutdownDB() drain now unmaps each domainMap under its own monitor and only while it is still the mapped value, instead of the iterator based remove which is unconditional by key: every removal site now literally follows the protocol documented on domainToReplicaDBs, as the cleanup's safety comment states. Review nits: getDomainToReplicaDBs() documents that returning the live internal map is intentional, the tests use that accessor instead of reflection, the heuristic blocked-thread waits are replaced with the precise monitor based wait, the import order is fixed, the test class javadoc describes the failed creation cleanup, and the drain race test no longer deregisters a server id it never creates. --- .../changelog/file/FileChangelogDB.java | 23 +- .../changelog/file/MultiDomainDBCursor.java | 35 ++- .../changelog/file/FileChangelogDBTest.java | 202 ++++++++++++------ 3 files changed, 186 insertions(+), 74 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index 887e9f1eb3..3748bfc84a 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -320,7 +320,8 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM // so this equality based remove provably drops this domainMap and no other. // Unlike removeDomain(), this path clears no ChangeNumberIndexer state, and a later // creation broadcasts addDomain() anew to multi domain cursors which already incorporated - // the domain: a pre-existing hazard of repeated addDomain() calls, unchanged here. + // the domain: that second announcement is a no-op, MultiDomainDBCursor.addDomain() + // ignores domains its cursor already iterates over. if (domainMap.isEmpty()) { domainToReplicaDBs.remove(baseDN, domainMap); @@ -358,9 +359,11 @@ FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final Replicatio /** * Returns the map of replica DBs per domain. *

- * Package private so that tests can observe which domain maps this changelog holds. + * Package private, for tests only: they both observe which domain maps this changelog holds and + * mutate the map to drive race interleavings, so this getter intentionally returns the live + * internal map, not a copy or an unmodifiable view. * - * @return the map of replica DBs per domain + * @return the live map of replica DBs per domain */ ConcurrentMap> getDomainToReplicaDBs() { @@ -444,13 +447,19 @@ public void shutdownDB() throws ChangelogException firstException = e; } - for (Iterator> it = - this.domainToReplicaDBs.values().iterator(); it.hasNext();) + for (Iterator>> it = + this.domainToReplicaDBs.entrySet().iterator(); it.hasNext();) { - final ConcurrentMap domainMap = it.next(); + final Map.Entry> entry = it.next(); + final ConcurrentMap domainMap = entry.getValue(); synchronized (domainMap) { - it.remove(); + // Follow the removal protocol documented on domainToReplicaDBs: unmap the domainMap under + // its own monitor, and only while it is still the mapped value. An iterator based remove + // is unconditional by key and would drop whatever map is under the baseDN when it runs, + // e.g. the fresh map of a creation which started after the shutdown flag was flipped - + // that creation observes the flag and cleans its own map up instead. + domainToReplicaDBs.remove(entry.getKey(), domainMap); for (FileReplicaDB replicaDB : domainMap.values()) { replicaDB.shutdown(); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java index 8a9ec2d221..2af6d5ec11 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java @@ -12,12 +12,14 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server.changelog.file; import java.util.Iterator; import java.util.Map.Entry; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.ConcurrentSkipListSet; import net.jcip.annotations.NotThreadSafe; @@ -34,6 +36,18 @@ public class MultiDomainDBCursor extends CompositeDBCursor { private final ReplicationDomainDB domainDB; private final ConcurrentSkipListMap newDomains = new ConcurrentSkipListMap<>(); + /** + * The domains this cursor already iterates over, so that announcing a domain a second time is a + * no-op: a second cursor over the same domain would either leak unclosed - the cursor tree of + * {@link CompositeDBCursor} collapses cursors comparing equal - or deliver every change twice. + * A domain may be announced again after the empty domainMap of a failed replica DB creation was + * dropped, while this cursor still iterates over it. + *

+ * Only ever mutated from the thread iterating this cursor - {@link #incorporateNewCursors()}, + * {@link #removeDomain(DN)} and {@link #close()} all run on it - and read by the threads + * announcing domains. + */ + private final ConcurrentSkipListSet incorporatedDomains = new ConcurrentSkipListSet<>(); private final CursorOptions options; /** @@ -52,6 +66,9 @@ public MultiDomainDBCursor(final ReplicationDomainDB domainDB, CursorOptions opt /** * Adds a replication domain for this cursor to iterate over. Added cursors * will be created and iterated over on the next call to {@link #next()}. + *

+ * A no-op for a domain this cursor already iterates over: new replica DBs of such a domain + * reach it through {@link DomainDBCursor#addReplicaDB(int, org.opends.server.replication.common.CSN)}. * * @param baseDN * the replication domain's baseDN @@ -60,7 +77,10 @@ public MultiDomainDBCursor(final ReplicationDomainDB domainDB, CursorOptions opt */ public void addDomain(DN baseDN, ServerState startAfterState) { - newDomains.put(baseDN, startAfterState != null ? startAfterState : new ServerState()); + if (!incorporatedDomains.contains(baseDN)) + { + newDomains.put(baseDN, startAfterState != null ? startAfterState : new ServerState()); + } } /** {@inheritDoc} */ @@ -73,8 +93,15 @@ protected void incorporateNewCursors() throws ChangelogException final Entry entry = iter.next(); final DN baseDN = entry.getKey(); final ServerState serverState = entry.getValue(); - final DBCursor domainDBCursor = domainDB.getCursorFrom(baseDN, serverState, options); - addCursor(domainDBCursor, baseDN); + // the check in addDomain() is only a fast path: an announcement racing an incorporation of + // the same domain can still queue it a second time, so incorporation itself must ignore + // domains this cursor already iterates over + if (!incorporatedDomains.contains(baseDN)) + { + final DBCursor domainDBCursor = domainDB.getCursorFrom(baseDN, serverState, options); + addCursor(domainDBCursor, baseDN); + incorporatedDomains.add(baseDN); + } iter.remove(); } } @@ -90,6 +117,7 @@ protected void incorporateNewCursors() throws ChangelogException public void removeDomain(DN baseDN) { removeCursor(baseDN); + incorporatedDomains.remove(baseDN); } /** {@inheritDoc} */ @@ -99,6 +127,7 @@ public void close() super.close(); domainDB.unregisterCursor(this); newDomains.clear(); + incorporatedDomains.clear(); } } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java index a6a0eb12db..221d78330c 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java @@ -20,11 +20,10 @@ import java.lang.management.ManagementFactory; import java.lang.management.ThreadInfo; import java.lang.management.ThreadMXBean; -import java.lang.reflect.Field; +import java.util.List; import java.util.concurrent.ConcurrentHashMap; -import java.util.Set; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -41,8 +40,10 @@ import org.opends.server.core.DirectoryServer; import org.opends.server.crypto.CryptoSuite; import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.CSN; import org.opends.server.replication.common.MultiDomainServerState; import org.opends.server.replication.common.ServerState; +import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.protocol.UpdateMsg; import org.opends.server.replication.server.ReplicationServer; import org.opends.server.replication.server.ReplicationServerDomain; @@ -63,11 +64,14 @@ /** * Test the FileChangelogDB class: the races between a replica DB creation and - * {@link FileChangelogDB#shutdownDB()}, and the window between + * {@link FileChangelogDB#shutdownDB()}; the window between * {@link FileChangelogDB#removeDomain(DN)}'s unlocked read of the domainMap and its * acquisition of the domainMap monitor, during which a concurrent remover * ({@code shutdownDB()}, {@code clearDB()} or another {@code removeDomain()}) may have - * unmapped the domain. + * unmapped the domain; and the cleanup of a creation which bails out without having created a + * replica DB - the empty domainMap it inserted must be dropped, without unmapping the fresh + * domainMap of a concurrent creation and without announcing the domain a second time to the + * multi domain cursors that were live at the time. */ @SuppressWarnings("javadoc") public class FileChangelogDBTest extends ReplicationTestCase @@ -304,7 +308,7 @@ public void run() } }; shutdowner.start(); - awaitBlockedOnAMonitor(shutdowner); + waitUntilBlockedOn(shutdowner, changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN)); changelogDB.releaseCreatedReplicaDB(); creator.join(TIMEOUT_MS); @@ -327,7 +331,7 @@ public void run() } join(creator); join(shutdowner); - deregisterLeakedReplicaDBMonitors(replicationServer, RACING_SERVER_ID, DRAINED_SERVER_ID); + deregisterLeakedReplicaDBMonitors(replicationServer, RACING_SERVER_ID); remove(replicationServer); TestCaseUtils.deleteDirectory(testRoot); } @@ -464,6 +468,87 @@ public void failedReplicaDBCreationKeepsAPopulatedDomainMap() throws Exception } } + /** + * The cleanup of a failed creation leaves the domain announced to every multi domain cursor + * which was live at the time, and the next successful creation of the domain announces it to + * them again: the second announcement must not open a second cursor over the same domain. Such + * a cursor would either leak unclosed - the cursor tree of {@code CompositeDBCursor} collapses + * cursors comparing equal - or deliver every change twice, which kills the + * {@code ChangeNumberIndexer} thread with the {@code IllegalStateException} its cookie update + * throws on a replayed change. + */ + @Test + public void announcingADomainTwiceToALiveCursorMustNotOpenASecondDomainCursor() throws Exception + { + TestCaseUtils.startServer(); + + ReplicationServer replicationServer = null; + RaceableChangelogDB changelogDB = null; + File testRoot = null; + try + { + replicationServer = configureReplicationServer(100, 5000); + testRoot = createCleanDir("FileChangelogDB"); + changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false)); + changelogDB.initializeDB(); + + // the live cursor both announcements reach + final MultiDomainDBCursor cursor = changelogDB.getCursorFrom( + new MultiDomainServerState(), new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY)); + try + { + // first announcement: the failed creation announces the domain before dropping its map + changelogDB.failNextReplicaDBCreation(); + try + { + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer); + failBecauseExceptionWasNotThrown(ChangelogException.class); + } + catch (ChangelogException expected) + { + assertThat(expected).hasMessage(CREATION_FAILURE.toString()); + } + cursor.next(); + assertThat(changelogDB.walkedDomains) + .as("the domains the live cursor iterates over after the first announcement") + .containsExactly(TEST_ROOT_DN); + + // second announcement: the next creation of the domain announces it to the cursor again + final FileReplicaDB replicaDB = + changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer).getFirst(); + final CSN csn = new CSN(System.currentTimeMillis(), 1, FAILING_SERVER_ID); + replicaDB.add(new DeleteMsg(TEST_ROOT_DN, csn, "uid")); + waitChangesArePersisted(replicaDB, 1); + + assertThat(cursor.next()).as("the change published after the second announcement").isTrue(); + assertThat(cursor.getRecord().getCSN()).isEqualTo(csn); + assertThat(changelogDB.walkedDomains) + .as("announcing an already incorporated domain again must not open a second cursor over it") + .containsExactly(TEST_ROOT_DN); + assertThat(cursor.next()).as("the single published change is delivered more than once").isFalse(); + } + finally + { + cursor.close(); + } + } + finally + { + try + { + if (changelogDB != null) + { + changelogDB.shutdownDB(); + } + } + finally + { + remove(replicationServer); + TestCaseUtils.deleteDirectory(testRoot); + } + } + } + /** * A creation which bails out on the identity check must not drop the domain map another creation * has freshly inserted: the drop is equality based and two empty maps are equal, so an identity @@ -555,7 +640,7 @@ public void run() // of the fresh domain map - without the identity check its cleanup would have unmapped the // fresh map, and it would have completed into a third map instead of blocking changelogDB.releaseCreatorHoldingItsDomainMap(); - awaitBlockedOnAMonitorOrCompleted(staleCreator); + waitUntilBlockedOnOrCompleted(staleCreator, freshDomainMap); assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN)) .as("the domain map the fresh creator is about to publish its replica DB into") .isSameAs(freshDomainMap); @@ -611,7 +696,7 @@ public void removeDomainRacingConcurrentRemovalMustNotThrowNPE() throws Exceptio changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, SERVER_ID, replicationServer).getFirst(); final ConcurrentMap> domainToReplicaDBs = - getDomainToReplicaDBs(changelogDB); + changelogDB.getDomainToReplicaDBs(); final ConcurrentMap domainMap = domainToReplicaDBs.get(TEST_ROOT_DN); assertThat(domainMap).isNotNull(); @@ -655,7 +740,7 @@ public void removeDomainMustNotUnmapConcurrentlyRecreatedDomain() throws Excepti changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, SERVER_ID, replicationServer).getFirst(); final ConcurrentMap> domainToReplicaDBs = - getDomainToReplicaDBs(changelogDB); + changelogDB.getDomainToReplicaDBs(); final ConcurrentMap domainMap = domainToReplicaDBs.get(TEST_ROOT_DN); assertThat(domainMap).isNotNull(); @@ -702,27 +787,27 @@ public void run() }, "removeDomain() under test"); } - @SuppressWarnings("unchecked") - private ConcurrentMap> getDomainToReplicaDBs( - FileChangelogDB changelogDB) throws Exception + /** Waits until the provided replica DB has persisted the provided number of records. */ + private void waitChangesArePersisted(FileReplicaDB replicaDB, int recordCount) throws Exception { - final Field field = FileChangelogDB.class.getDeclaredField("domainToReplicaDBs"); - field.setAccessible(true); - return (ConcurrentMap>) field.get(changelogDB); + final long deadline = System.currentTimeMillis() + TIMEOUT_MS; + while (replicaDB.getNumberRecords() < recordCount) + { + if (System.currentTimeMillis() > deadline) + { + throw new AssertionError("Timed out waiting for " + recordCount + " records to be persisted"); + } + Thread.sleep(10); + } } /** Waits until the provided thread is blocked acquiring the monitor of the provided object. */ private void waitUntilBlockedOn(Thread thread, Object monitor) throws Exception { - final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); final long deadline = System.currentTimeMillis() + TIMEOUT_MS; while (System.currentTimeMillis() < deadline) { - final ThreadInfo threadInfo = threadMXBean.getThreadInfo(thread.getId()); - final LockInfo lockInfo = threadInfo != null ? threadInfo.getLockInfo() : null; - if (lockInfo != null - && threadInfo.getThreadState() == Thread.State.BLOCKED - && lockInfo.getIdentityHashCode() == System.identityHashCode(monitor)) + if (isBlockedOn(thread, monitor)) { return; } @@ -732,60 +817,49 @@ private void waitUntilBlockedOn(Thread thread, Object monitor) throws Exception "Timed out waiting for " + thread.getName() + " to block on the domainMap monitor"); } - /** Joins the provided thread, leaving a signal behind when it did not die within the timeout. */ - private void join(final Thread thread) throws InterruptedException + /** + * Waits until the provided thread is blocked acquiring the monitor of the provided object, or + * has completed: completion is left for the caller's assertions to diagnose. + */ + private void waitUntilBlockedOnOrCompleted(Thread thread, Object monitor) throws Exception { - if (thread != null) + final long deadline = System.currentTimeMillis() + TIMEOUT_MS; + while (System.currentTimeMillis() < deadline) { - thread.join(TIMEOUT_MS); - if (thread.isAlive()) + if (!thread.isAlive() || isBlockedOn(thread, monitor)) { - final IllegalStateException hung = new IllegalStateException("Test thread " + thread.getName() - + " is still alive after " + TIMEOUT_MS + " ms: it may leak a live changelog into later tests"); - hung.setStackTrace(thread.getStackTrace()); - hung.printStackTrace(); - thread.interrupt(); + return; } + Thread.sleep(1); } + throw new AssertionError("Timed out waiting for " + thread.getName() + + " to block on the domainMap monitor or complete"); } - /** - * Waits until the provided thread is blocked acquiring a monitor: the domain map monitor is the - * only one it can stay blocked on - the other locks on its way are only transiently contended, - * hence the two consecutive observations. A thread expected to block but completing instead is - * an error. - */ - private static void awaitBlockedOnAMonitor(final Thread thread) throws InterruptedException + private static boolean isBlockedOn(Thread thread, Object monitor) { - awaitBlockedOnAMonitorOrCompleted(thread); - if (!thread.isAlive()) - { - throw new IllegalStateException(thread.getName() + " completed without blocking on the domain map monitor"); - } + final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); + final ThreadInfo threadInfo = threadMXBean.getThreadInfo(thread.getId()); + final LockInfo lockInfo = threadInfo != null ? threadInfo.getLockInfo() : null; + return lockInfo != null + && threadInfo.getThreadState() == Thread.State.BLOCKED + && lockInfo.getIdentityHashCode() == System.identityHashCode(monitor); } - /** - * Waits until the provided thread is durably blocked acquiring a monitor - two consecutive - * observations, since the other locks on its way are only transiently contended - or has - * completed: completion is left for the caller's assertions to diagnose. - */ - private static void awaitBlockedOnAMonitorOrCompleted(final Thread thread) throws InterruptedException + /** Joins the provided thread, leaving a signal behind when it did not die within the timeout. */ + private void join(final Thread thread) throws InterruptedException { - final long deadline = System.currentTimeMillis() + TIMEOUT_MS; - int blockedObservations = 0; - while (blockedObservations < 2) + if (thread != null) { - if (!thread.isAlive()) - { - return; - } - if (System.currentTimeMillis() > deadline) + thread.join(TIMEOUT_MS); + if (thread.isAlive()) { - throw new IllegalStateException( - "timed out waiting for " + thread.getName() + " to block on the domain map monitor"); + final IllegalStateException hung = new IllegalStateException("Test thread " + thread.getName() + + " is still alive after " + TIMEOUT_MS + " ms: it may leak a live changelog into later tests"); + hung.setStackTrace(thread.getStackTrace()); + hung.printStackTrace(); + thread.interrupt(); } - blockedObservations = thread.getState() == Thread.State.BLOCKED ? blockedObservations + 1 : 0; - Thread.sleep(1); } } @@ -844,8 +918,8 @@ private static final class RaceableChangelogDB extends FileChangelogDB private final CountDownLatch drainIsInReplicaDBShutdown = new CountDownLatch(1); private final CountDownLatch drainIsReleased = new CountDownLatch(1); - /** The baseDNs of the domains any cursor was opened for. */ - private final Set walkedDomains = new CopyOnWriteArraySet<>(); + /** The baseDNs of the domains any cursor was opened for, one element per opening. */ + private final List walkedDomains = new CopyOnWriteArrayList<>(); RaceableChangelogDB(final ReplicationServer replicationServer, final String dbDirectoryPath, final CryptoSuite cryptoSuite) throws ConfigException From d3e6048a4e3a476a1344d3b1479804140ef3d939 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 5 Aug 2026 10:33:05 +0300 Subject: [PATCH 6/6] [#818] Address the third review round on the idempotent announcements ECLMultiDomainDBCursorTest is adapted to the new announcement contract, in the shape proposed by the review: addDomainCursorToCursor() keeps one long-lived SequentialDBCursor per domain and drains re-announced cursors into it, exactly as DomainDBCursor.addReplicaDB() feeds an incorporated domain in production, so the test body is unchanged and the class passes against both the old and the new MultiDomainDBCursor. The map of per-domain cursors is cleared in setup(): one test class instance runs all the methods, and domains announced to a previous method's cursor must not be mistaken for domains the fresh one iterates over. The addDomain() fast path is gone: checking incorporatedDomains on the announcing thread was a check-then-act against removeDomain() on the cursor's thread, able to drop an announcement the removal no longer covers. incorporateNewCursors() keeps the authoritative check, on the only thread adding to and removing from the set. The shutdownDB() drain now performs the same identity check as removeDomain() before unmapping: an equality based remove could still drop a map whose monitor is not held, since two empty maps are equal. Javadoc: incorporatedDomains no longer claims close() runs on the cursor's thread, and the double-announcement test says it covers the drop path only - removeDomain() never re-announces a domain to a cursor which still holds it. --- .../changelog/file/FileChangelogDB.java | 20 ++++++----- .../changelog/file/MultiDomainDBCursor.java | 34 +++++++++---------- .../file/ECLMultiDomainDBCursorTest.java | 20 +++++++++++ .../changelog/file/FileChangelogDBTest.java | 4 +++ .../changelog/file/SequentialDBCursor.java | 10 ++++++ 5 files changed, 63 insertions(+), 25 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java index 3748bfc84a..8a914728b7 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java @@ -320,8 +320,8 @@ private Pair getExistingOrNewReplicaDB(final ConcurrentM // so this equality based remove provably drops this domainMap and no other. // Unlike removeDomain(), this path clears no ChangeNumberIndexer state, and a later // creation broadcasts addDomain() anew to multi domain cursors which already incorporated - // the domain: that second announcement is a no-op, MultiDomainDBCursor.addDomain() - // ignores domains its cursor already iterates over. + // the domain: MultiDomainDBCursor discards such an announcement when it incorporates new + // cursors, so no second cursor is opened over the domain. if (domainMap.isEmpty()) { domainToReplicaDBs.remove(baseDN, domainMap); @@ -454,12 +454,16 @@ public void shutdownDB() throws ChangelogException final ConcurrentMap domainMap = entry.getValue(); synchronized (domainMap) { - // Follow the removal protocol documented on domainToReplicaDBs: unmap the domainMap under - // its own monitor, and only while it is still the mapped value. An iterator based remove - // is unconditional by key and would drop whatever map is under the baseDN when it runs, - // e.g. the fresh map of a creation which started after the shutdown flag was flipped - - // that creation observes the flag and cleans its own map up instead. - domainToReplicaDBs.remove(entry.getKey(), domainMap); + // Follow the removal protocol documented on domainToReplicaDBs: unmap only the domainMap + // instance the monitor was taken on. The check is identity based, like removeDomain()'s: + // an equality based remove could still drop a map whose monitor is not held, since two + // empty maps are equal and removeDomain() plus a fresh creation may have swapped one for + // another since this iterator read its entry. Replica DBs another remover already visited + // are shut down again below, which is harmless: shutdown is a no-op the second time. + if (domainToReplicaDBs.get(entry.getKey()) == domainMap) + { + domainToReplicaDBs.remove(entry.getKey()); + } for (FileReplicaDB replicaDB : domainMap.values()) { replicaDB.shutdown(); diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java index 2af6d5ec11..4019bd762c 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/MultiDomainDBCursor.java @@ -37,15 +37,15 @@ public class MultiDomainDBCursor extends CompositeDBCursor private final ReplicationDomainDB domainDB; private final ConcurrentSkipListMap newDomains = new ConcurrentSkipListMap<>(); /** - * The domains this cursor already iterates over, so that announcing a domain a second time is a - * no-op: a second cursor over the same domain would either leak unclosed - the cursor tree of - * {@link CompositeDBCursor} collapses cursors comparing equal - or deliver every change twice. - * A domain may be announced again after the empty domainMap of a failed replica DB creation was - * dropped, while this cursor still iterates over it. + * The domains this cursor already iterates over, so that a second announcement of a domain is + * discarded at incorporation: a second cursor over the same domain would either leak unclosed + * - the cursor tree of {@link CompositeDBCursor} collapses cursors comparing equal - or + * deliver every change twice. A domain may be announced again after the empty domainMap of a + * failed replica DB creation was dropped, while this cursor still iterates over it. *

- * Only ever mutated from the thread iterating this cursor - {@link #incorporateNewCursors()}, - * {@link #removeDomain(DN)} and {@link #close()} all run on it - and read by the threads - * announcing domains. + * Added to and removed from on the thread iterating this cursor - {@link #incorporateNewCursors()} + * and {@link #removeDomain(DN)} run on it - read by the threads announcing domains, and + * cleared by {@link #close()}, which an ending ECL session may call from another thread. */ private final ConcurrentSkipListSet incorporatedDomains = new ConcurrentSkipListSet<>(); private final CursorOptions options; @@ -67,8 +67,9 @@ public MultiDomainDBCursor(final ReplicationDomainDB domainDB, CursorOptions opt * Adds a replication domain for this cursor to iterate over. Added cursors * will be created and iterated over on the next call to {@link #next()}. *

- * A no-op for a domain this cursor already iterates over: new replica DBs of such a domain - * reach it through {@link DomainDBCursor#addReplicaDB(int, org.opends.server.replication.common.CSN)}. + * Announcing a domain this cursor already iterates over has no effect: the announcement is + * discarded when cursors are incorporated, and new replica DBs of such a domain reach it + * through {@link DomainDBCursor#addReplicaDB(int, org.opends.server.replication.common.CSN)}. * * @param baseDN * the replication domain's baseDN @@ -77,10 +78,10 @@ public MultiDomainDBCursor(final ReplicationDomainDB domainDB, CursorOptions opt */ public void addDomain(DN baseDN, ServerState startAfterState) { - if (!incorporatedDomains.contains(baseDN)) - { - newDomains.put(baseDN, startAfterState != null ? startAfterState : new ServerState()); - } + // incorporateNewCursors() discards announcements of domains this cursor already iterates + // over: checking incorporatedDomains here would be a check-then-act against removeDomain() + // on the cursor's thread, able to drop an announcement the removal no longer covers + newDomains.put(baseDN, startAfterState != null ? startAfterState : new ServerState()); } /** {@inheritDoc} */ @@ -93,9 +94,8 @@ protected void incorporateNewCursors() throws ChangelogException final Entry entry = iter.next(); final DN baseDN = entry.getKey(); final ServerState serverState = entry.getValue(); - // the check in addDomain() is only a fast path: an announcement racing an incorporation of - // the same domain can still queue it a second time, so incorporation itself must ignore - // domains this cursor already iterates over + // discard the announcement of a domain this cursor already iterates over: this is the only + // thread adding to and removing from incorporatedDomains, so the check cannot race them if (!incorporatedDomains.contains(baseDN)) { final DBCursor domainDBCursor = domainDB.getCursorFrom(baseDN, serverState, options); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/ECLMultiDomainDBCursorTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/ECLMultiDomainDBCursorTest.java index 6ef8d2f3ab..e6e3ffd4fd 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/ECLMultiDomainDBCursorTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/ECLMultiDomainDBCursorTest.java @@ -12,10 +12,13 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2014-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server.changelog.file; +import java.util.HashMap; import java.util.HashSet; +import java.util.Map; import java.util.Set; import org.forgerock.opendj.ldap.DN; @@ -45,6 +48,8 @@ public class ECLMultiDomainDBCursorTest extends DirectoryServerTestCase private MultiDomainDBCursor multiDomainCursor; private ECLMultiDomainDBCursor eclCursor; private final Set eclEnabledDomains = new HashSet<>(); + /** The long-lived cursor of each domain announced to {@link #multiDomainCursor}. */ + private final Map domainCursors = new HashMap<>(); private ECLEnabledDomainPredicate predicate = new ECLEnabledDomainPredicate() { @Override @@ -61,6 +66,9 @@ public void setup() throws Exception options = new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY); multiDomainCursor = new MultiDomainDBCursor(domainDB, options); eclCursor = new ECLMultiDomainDBCursor(predicate, multiDomainCursor); + // one test class instance runs all the methods: the domains announced to the previous + // method's multiDomainCursor must not be mistaken for domains this one iterates over + domainCursors.clear(); } @AfterMethod @@ -185,6 +193,18 @@ private void assertMessagesInOrder(DN baseDN, UpdateMsg msg1, UpdateMsg msg2) th private void addDomainCursorToCursor(DN baseDN, SequentialDBCursor cursor) throws ChangelogException { + final SequentialDBCursor existing = domainCursors.get(baseDN); + if (existing != null) + { + // already known to the cursor: its long-lived per-domain cursor receives the new changes, + // exactly as DomainDBCursor.addReplicaDB() does in production + for (UpdateMsg msg : cursor.drain()) + { + existing.add(msg); + } + return; + } + domainCursors.put(baseDN, cursor); final ServerState state = new ServerState(); when(domainDB.getCursorFrom(baseDN, state, options)).thenReturn(cursor); multiDomainCursor.addDomain(baseDN, state); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java index 221d78330c..31bfa61765 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangelogDBTest.java @@ -476,6 +476,10 @@ public void failedReplicaDBCreationKeepsAPopulatedDomainMap() throws Exception * cursors comparing equal - or deliver every change twice, which kills the * {@code ChangeNumberIndexer} thread with the {@code IllegalStateException} its cookie update * throws on a replayed change. + *

+ * This covers the drop path only: the {@code removeDomain()} path never re-announces a domain + * to a cursor which still holds it - the cursor drops the domain, through + * {@code indexer.clear()}, before the domain is unmapped. */ @Test public void announcingADomainTwiceToALiveCursorMustNotOpenASecondDomainCursor() throws Exception diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/SequentialDBCursor.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/SequentialDBCursor.java index 333e73bd4a..401ae1313f 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/SequentialDBCursor.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/SequentialDBCursor.java @@ -12,9 +12,11 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2013-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.server.changelog.file; +import java.util.ArrayList; import java.util.List; import org.opends.server.replication.protocol.UpdateMsg; @@ -44,6 +46,14 @@ public void add(UpdateMsg msg) this.msgs.add(msg); } + /** Returns the messages this cursor has not consumed yet, leaving it empty. */ + public List drain() + { + final List drained = new ArrayList<>(msgs); + msgs.clear(); + return drained; + } + @Override public UpdateMsg getRecord() {