From a05c0fbf661a63209b951d488f84031cfc7c7f0e Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 9 Sep 2026 18:33:21 +0300 Subject: [PATCH] [#896] Open the entry containers of a backend inside the write and register them after it RootContainer.open opens and registers the entry container of every base DN in a single storage.write, and Storage.write replays its operation after a transaction conflict. No rollback reaches the registry, so a replay found every base DN the attempt it replaced had reached already registered and failed with ERR_ENTRY_CONTAINER_ALREADY_REGISTERED: the backend did not open at all, on a message which said nothing about the conflict that caused the replay. Every attempt also left the configuration listeners of its entry containers behind. The containers are now opened inside the write and registered once it has committed, and every attempt begins by giving up what the previous one opened, as BackendImpl.changeBaseDNTrees does for the base DN change of #907. An open which fails closes what it opened, for the same reason. Reached on persistit through a plain rollback, and on the jdbc backend through the conflict and dropped connection replays of an existing backend, whose open issues no committing statement; je and cassandra replay no write at all. --- .../backends/pluggable/RootContainer.java | 42 +- .../backends/pluggable/ReplayedOpenTest.java | 528 ++++++++++++++++++ 2 files changed, 562 insertions(+), 8 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ReplayedOpenTest.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/RootContainer.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/RootContainer.java index ad1dcd6380..2367263b2c 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/RootContainer.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/RootContainer.java @@ -17,6 +17,7 @@ */ package org.opends.server.backends.pluggable; +import static org.forgerock.util.Utils.closeSilently; import static org.opends.messages.BackendMessages.*; import static org.opends.server.util.StaticUtils.*; @@ -119,6 +120,15 @@ Storage getStorage() /** * Opens the root container. + *

+ * {@link Storage#write(WriteOperation)} replays its operation after a transaction conflict, so + * the operation below is confined to work a rollback undoes: the entry containers are opened + * there, while the registry, which no rollback reaches, is filled in once the write has + * committed. Registering them inside the write instead is what a replay then fails on - the + * attempt it replaces left every base DN it had reached in {@link #entryContainers}, so + * {@link #registerEntryContainer} reports ERR_ENTRY_CONTAINER_ALREADY_REGISTERED and the backend + * does not open at all, on a message which says nothing about the conflict that caused the + * replay. * * @param accessMode specifies how the container has to be opened (read-write or read-only) * @@ -129,6 +139,8 @@ Storage getStorage() */ void open(final AccessMode accessMode) throws StorageRuntimeException, ConfigException { + // Opened by the write operation, registered only once it has committed. + final List opened = new ArrayList<>(); try { storage.open(accessMode); @@ -137,20 +149,34 @@ void open(final AccessMode accessMode) throws StorageRuntimeException, ConfigExc @Override public void run(WriteableTransaction txn) throws Exception { + // Give up what a previous, rolled back attempt had opened: its trees are gone, and its + // entry containers still hold the configuration listeners they registered. + closeSilently(opened); + opened.clear(); compressedSchema = new PersistentCompressedSchema(serverContext, backendId, storage, txn, accessMode); - openAndRegisterEntryContainers(txn, config.getBaseDN(), accessMode); + openEntryContainers(txn, config.getBaseDN(), accessMode, opened); } }); + // Cannot fail: the base DNs come from a set, and the map of a root container being opened is + // empty until here. + for (EntryContainer ec : opened) + { + registerEntryContainer(ec.getBaseDN(), ec); + } // after the write, never inside it: a compressed schema migration is only worth reporting // once the transaction that copied it has committed, and a replayed operation runs twice compressedSchema.reportMigration(); } catch(StorageRuntimeException e) { + // Nothing else holds them: an open which fails is an open whose root container the caller + // drops, and the listeners of an entry container outlive it until its close() takes them off. + closeSilently(opened); throw e; } catch (Exception e) { + closeSilently(opened); throw new StorageRuntimeException(e); } } @@ -203,30 +229,30 @@ void registerEntryContainer(DN baseDN, EntryContainer entryContainer) throws Ini } /** - * Opens the entry containers for multiple base DNs. + * Opens the entry containers for multiple base DNs, collecting them for the caller to register + * once the write they are opened by has committed - see {@link #open(AccessMode)}. * * @param baseDNs * The base DNs of the entry containers to open. * @param accessMode specifies how the containers have to be opened (read-write or read-only) + * @param opened + * Collects the containers this method opens, in the order it opened them. * * @throws StorageRuntimeException * If an error occurs while opening the entry container. - * @throws InitializationException - * If an initialization error occurs while opening the entry - * container. * @throws ConfigException * If a configuration error occurs while opening the entry * container. */ - private void openAndRegisterEntryContainers(WriteableTransaction txn, Set baseDNs, AccessMode accessMode) - throws StorageRuntimeException, InitializationException, ConfigException + private void openEntryContainers(WriteableTransaction txn, Set baseDNs, AccessMode accessMode, + List opened) throws StorageRuntimeException, ConfigException { EntryID highestID = null; for (DN baseDN : baseDNs) { EntryContainer ec = openEntryContainer(baseDN, txn, accessMode); EntryID id = ec.getHighestEntryID(txn); - registerEntryContainer(baseDN, ec); + opened.add(ec); if (highestID == null || id.compareTo(highestID) > 0) { highestID = id; diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ReplayedOpenTest.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ReplayedOpenTest.java new file mode 100644 index 0000000000..14c06a89d3 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ReplayedOpenTest.java @@ -0,0 +1,528 @@ +/* + * 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.backends.pluggable; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.forgerock.opendj.config.ConfigurationMock.mockCfg; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.opends.server.util.CollectionUtils.newTreeSet; + +import java.util.HashSet; +import java.util.Set; +import java.util.SortedSet; + +import org.forgerock.opendj.config.server.ConfigException; +import org.forgerock.opendj.ldap.ByteSequence; +import org.forgerock.opendj.ldap.ByteString; +import org.forgerock.opendj.ldap.DN; +import org.forgerock.opendj.ldap.schema.AttributeType; +import org.forgerock.opendj.server.config.meta.BackendIndexCfgDefn.IndexType; +import org.forgerock.opendj.server.config.server.BackendIndexCfg; +import org.forgerock.opendj.server.config.server.PDBBackendCfg; +import org.opends.server.DirectoryServerTestCase; +import org.opends.server.TestCaseUtils; +import org.opends.server.backends.pdb.PDBStorage; +import org.opends.server.backends.pluggable.spi.AccessMode; +import org.opends.server.backends.pluggable.spi.Cursor; +import org.opends.server.backends.pluggable.spi.Importer; +import org.opends.server.backends.pluggable.spi.ReadOperation; +import org.opends.server.backends.pluggable.spi.Storage; +import org.opends.server.backends.pluggable.spi.StorageStatus; +import org.opends.server.backends.pluggable.spi.TreeName; +import org.opends.server.backends.pluggable.spi.UpdateFunction; +import org.opends.server.backends.pluggable.spi.WriteOperation; +import org.opends.server.backends.pluggable.spi.WriteableTransaction; +import org.opends.server.core.ServerContext; +import org.opends.server.types.BackupConfig; +import org.opends.server.types.BackupDirectory; +import org.opends.server.types.DirectoryException; +import org.opends.server.types.RestoreConfig; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import com.persistit.exception.RollbackException; + +/** + * Tests that {@link RootContainer#open(AccessMode)} survives a replay of its {@link WriteOperation}. + * {@link Storage#write(WriteOperation)} may replay the operation after a transaction conflict, and + * {@code RootContainer.open} opens and registers the entry container of every base DN inside a + * single one of them, so every side effect that write performs must either be transactional or be + * idempotent - see OpenDJ issue #896. + *

+ * The conflict is raised from inside the operation as the {@link RollbackException} PersistIt itself + * raises, so that the replay is driven by {@code PDBStorage.write}'s own retry loop rather than by a + * second call to it, as {@link ReplayedConfigChangeTest} does for the sibling path. + */ +@SuppressWarnings("javadoc") +@Test(groups = { "precommit", "pluggablebackend" }, sequential = true) +public class ReplayedOpenTest extends DirectoryServerTestCase +{ + private static final String BACKEND_ID = "ReplayedOpenTest"; + /** Opened and registered first, since the base DNs are opened in the order of the sorted set. */ + private static final DN FIRST = DN.valueOf("dc=b896a,dc=com"); + /** Opened while the first one is already registered, which is where a conflict reaches the bug. */ + private static final DN SECOND = DN.valueOf("dc=b896b,dc=com"); + + private ServerContext serverContext; + private AttributeType cnType; + + @BeforeClass + public void startServer() throws Exception + { + TestCaseUtils.startServer(); + serverContext = TestCaseUtils.getServerContext(); + cnType = serverContext.getSchema().getAttributeType("cn"); + } + + /** + * These tests open a backend which is designed to fail, and a failing one can leave a base DN + * behind in the server wide registry, where it would outlive the test and break the next one. + */ + @AfterMethod + public void deregisterLeftoverBaseDNs() + { + for (DN baseDN : new DN[] { FIRST, SECOND }) + { + try + { + serverContext.getBackendConfigManager().deregisterBaseDN(baseDN); + } + catch (Exception alreadyGone) + { + // Which is what the test should have left behind. + } + } + } + + /** + * The case the report is written from: the conflict is raised while the second base DN is being + * opened, so the replay meets the first one already registered by the attempt it replaces. + */ + @Test + public void openIsReplayableWhenTheTransactionConflictsWhileTheSecondBaseDNIsOpened() throws Exception + { + final ReplayingBackend backend = newBackend(newTreeSet(FIRST, SECOND)); + boolean opened = false; + try + { + backend.storage.conflictAtTreesOf(SECOND, 1); + backend.openBackend(); + opened = true; + + assertThat(backend.storage.attempts()).isEqualTo(2); + final RootContainer rootContainer = backend.getRootContainer(); + assertThat(rootContainer.getBaseDNs()).containsOnly(FIRST, SECOND); + // Each base DN is answered with a container of its own, rather than with the one an ancestor + // registered, and the trees of both are there to be read. + for (DN baseDN : new DN[] { FIRST, SECOND }) + { + final EntryContainer ec = rootContainer.getEntryContainer(baseDN); + assertThat((Object) ec.getBaseDN()).isEqualTo(baseDN); + assertThat(rootContainer.getStorage().listTrees()).containsAll(treesOf(ec)); + } + } + finally + { + close(backend, opened); + } + } + + /** + * A conflict reported by {@code commit()} replays an operation which ran to completion, so every + * base DN of the backend is registered by the attempt the replay replaces. One base DN would be + * enough to reach that; two are used so that the case differs from the one above only in where the + * conflict was raised. + */ + @Test + public void openIsReplayableWhenTheTransactionConflictsAtCommitTime() throws Exception + { + final ReplayingBackend backend = newBackend(newTreeSet(FIRST, SECOND)); + boolean opened = false; + try + { + backend.storage.conflictAtCommit(1); + backend.openBackend(); + opened = true; + + assertThat(backend.storage.attempts()).isEqualTo(2); + assertThat(backend.getRootContainer().getBaseDNs()).containsOnly(FIRST, SECOND); + // The two entry containers the rolled back attempt opened registered five configuration + // listeners each, which only their close() takes back, so the replay has to give them up + // before it opens another pair. + verify(backend.configuredWith, times(2)).removePluggableChangeListener(any()); + } + finally + { + close(backend, opened); + } + } + + /** + * A failure the storage engine does not replay ends the open, and what the attempt had opened by + * then is reachable from nothing else: the root container is dropped by + * {@code BackendImpl.newRootContainer}, and every entry container with it. + */ + @Test + public void anOpenWhichIsNotReplayedGivesUpTheEntryContainersItOpened() throws Exception + { + final ReplayingBackend backend = newBackend(newTreeSet(FIRST, SECOND)); + try + { + backend.storage.failWithoutReplay(); + + try + { + backend.openBackend(); + throw new AssertionError("the open was expected to fail"); + } + catch (Exception expected) + { + // The failure itself is the caller's business; what it left behind is this test's. + } + + verify(backend.configuredWith, times(2)).removePluggableChangeListener(any()); + } + finally + { + close(backend, false); + } + } + + private static Set treesOf(EntryContainer ec) + { + final Set names = new HashSet<>(); + for (Tree tree : ec.listTrees()) + { + names.add(tree.getName()); + } + return names; + } + + /** + * Closes a backend which opened, and the storage alone of one which did not: nothing else closes + * the storage {@code RootContainer.open} had already opened, and a volume left open fails every + * following test with a {@code StorageInUseException} rather than with what actually broke. + */ + private static void close(ReplayingBackend backend, boolean opened) + { + if (opened) + { + backend.finalizeBackend(); + } + else + { + backend.storage.close(); + } + } + + private ReplayingBackend newBackend(SortedSet baseDNs) throws Exception + { + final ReplayingBackend backend = new ReplayingBackend(); + backend.setBackendID(BACKEND_ID); + backend.configuredWith = backendCfg(baseDNs); + backend.configureBackend(backend.configuredWith, serverContext); + // Start from a pristine on-disk state so that a previous run cannot mask the defect. + backend.storage.removeStorageFiles(); + return backend; + } + + private PDBBackendCfg backendCfg(SortedSet baseDNs) throws ConfigException + { + final PDBBackendCfg cfg = mockCfg(PDBBackendCfg.class); + when(cfg.dn()).thenReturn(DN.valueOf("ds-cfg-backend-id=" + BACKEND_ID + ",cn=Backends,cn=config")); + when(cfg.getBackendId()).thenReturn(BACKEND_ID); + when(cfg.getDBDirectory()).thenReturn(BACKEND_ID); + when(cfg.getDBDirectoryPermissions()).thenReturn("755"); + when(cfg.getDBCacheSize()).thenReturn(0L); + when(cfg.getDBCachePercent()).thenReturn(20); + when(cfg.getBaseDN()).thenReturn(baseDNs); + when(cfg.listBackendIndexes()).thenReturn(new String[] { "cn" }); + when(cfg.listBackendVLVIndexes()).thenReturn(new String[0]); + + final BackendIndexCfg indexCfg = mock(BackendIndexCfg.class); + when(indexCfg.getIndexType()).thenReturn(newTreeSet(IndexType.EQUALITY)); + when(indexCfg.getAttribute()).thenReturn(cnType); + when(indexCfg.getIndexEntryLimit()).thenReturn(4000); + when(indexCfg.getSubstringLength()).thenReturn(6); + when(cfg.getBackendIndex("cn")).thenReturn(indexCfg); + return cfg; + } + + /** A backend whose storage makes the next write operation conflict, and so be replayed. */ + private static final class ReplayingBackend extends BackendImpl + { + private ReplayingStorage storage; + /** The configuration the entry containers register their listeners with. */ + private PDBBackendCfg configuredWith; + + @Override + protected Storage configureStorage(PDBBackendCfg cfg, ServerContext serverContext) throws ConfigException + { + storage = new ReplayingStorage(new PDBStorage(cfg, serverContext)); + return storage; + } + } + + /** A failure which no storage engine replays, unlike {@link RollbackException}. */ + private static final class UnreplayableFailure extends Exception + { + private static final long serialVersionUID = 1L; + } + + /** + * Decorates a {@link Storage} so that the next {@link Storage#write(WriteOperation)} conflicts a + * given number of times before it is let through. The conflict is raised from within the single + * {@code write} the delegate is asked for, so the delegate's own retry loop performs the replay. + */ + private static final class ReplayingStorage implements Storage + { + /** Where the conflict is raised, which decides how much of the operation has run. */ + private enum ConflictPoint + { + /** As the first tree of a given base DN is opened, with the base DNs before it registered. */ + TREES_OF_BASE_DN, + /** Once the operation has run to completion, as a conflict reported by {@code commit()}. */ + COMMIT, + /** Once the operation has run to completion, as a failure which is not replayed at all. */ + NO_REPLAY + } + + private final Storage delegate; + private ConflictPoint conflictPoint; + private String conflictingPrefix; + private int conflictsLeft; + private int attempts; + + ReplayingStorage(Storage delegate) + { + this.delegate = delegate; + } + + void conflictAtTreesOf(DN baseDN, int conflicts) + { + conflictingPrefix = baseDN.toNormalizedUrlSafeString(); + arm(ConflictPoint.TREES_OF_BASE_DN, conflicts); + } + + void conflictAtCommit(int conflicts) + { + arm(ConflictPoint.COMMIT, conflicts); + } + + void failWithoutReplay() + { + arm(ConflictPoint.NO_REPLAY, 1); + } + + private void arm(ConflictPoint where, int conflicts) + { + conflictPoint = where; + conflictsLeft = conflicts; + attempts = 0; + } + + /** How many times the armed operation was run, the first attempt included. */ + int attempts() + { + return attempts; + } + + @Override + public void write(final WriteOperation writeOperation) throws Exception + { + final ConflictPoint armed = conflictPoint; + if (armed == null) + { + delegate.write(writeOperation); + return; + } + conflictPoint = null; + // A single call, so that the replay is the delegate's own and keeps whatever the delegate + // holds for the duration of a write, rather than starting afresh as a second call would. + delegate.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + attempts++; + if (conflictsLeft-- <= 0) + { + writeOperation.run(txn); + return; + } + if (armed == ConflictPoint.TREES_OF_BASE_DN) + { + writeOperation.run(new ConflictingAtTreesOf(txn, conflictingPrefix)); + return; + } + writeOperation.run(txn); + if (armed == ConflictPoint.NO_REPLAY) + { + throw new UnreplayableFailure(); + } + throw new RollbackException(); + } + }); + } + + @Override + public Importer startImport() throws ConfigException + { + return delegate.startImport(); + } + + @Override + public void open(AccessMode accessMode) throws Exception + { + delegate.open(accessMode); + } + + @Override + public T read(ReadOperation readOperation) throws Exception + { + return delegate.read(readOperation); + } + + @Override + public void removeStorageFiles() + { + delegate.removeStorageFiles(); + } + + @Override + public StorageStatus getStorageStatus() + { + return delegate.getStorageStatus(); + } + + @Override + public boolean supportsBackupAndRestore() + { + return delegate.supportsBackupAndRestore(); + } + + @Override + public void createBackup(BackupConfig backupConfig) throws DirectoryException + { + delegate.createBackup(backupConfig); + } + + @Override + public void removeBackup(BackupDirectory backupDirectory, String backupID) throws DirectoryException + { + delegate.removeBackup(backupDirectory, backupID); + } + + @Override + public void restoreBackup(RestoreConfig restoreConfig) throws DirectoryException + { + delegate.restoreBackup(restoreConfig); + } + + @Override + public Set listTrees() + { + return delegate.listTrees(); + } + + @Override + public void close() + { + delegate.close(); + } + } + + /** + * A transaction which conflicts as the first tree of one base DN is opened, and delegates + * everything the operation did before that. This is where an entry container which is being opened + * meets a write-write conflict: PersistIt raises {@link RollbackException} at the store, so the + * base DNs opened before this one have been registered and the one being opened has not. + */ + private static final class ConflictingAtTreesOf implements WriteableTransaction + { + private final WriteableTransaction delegate; + private final String conflictingPrefix; + + ConflictingAtTreesOf(WriteableTransaction delegate, String conflictingPrefix) + { + this.delegate = delegate; + this.conflictingPrefix = conflictingPrefix; + } + + @Override + public void openTree(TreeName name, boolean createOnDemand) + { + if (conflictingPrefix.equals(name.getBaseDN())) + { + throw new RollbackException(); + } + delegate.openTree(name, createOnDemand); + } + + @Override + public void deleteTree(TreeName name) + { + delegate.deleteTree(name); + } + + @Override + public void put(TreeName treeName, ByteSequence key, ByteSequence value) + { + delegate.put(treeName, key, value); + } + + @Override + public boolean update(TreeName treeName, ByteSequence key, UpdateFunction f) + { + return delegate.update(treeName, key, f); + } + + @Override + public boolean delete(TreeName treeName, ByteSequence key) + { + return delegate.delete(treeName, key); + } + + @Override + public ByteString read(TreeName treeName, ByteSequence key) + { + return delegate.read(treeName, key); + } + + @Override + public Cursor openCursor(TreeName treeName) + { + return delegate.openCursor(treeName); + } + + @Override + public long getRecordCount(TreeName treeName) + { + return delegate.getRecordCount(treeName); + } + + @Override + public boolean treeExists(TreeName treeName) + { + return delegate.treeExists(treeName); + } + } +}