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/2] [#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/2] [#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)