From 5d176c691527e3fe4bc529ff8947f3b3648970bd Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 10 Sep 2026 11:58:57 +0000
Subject: [PATCH] [#916] Keep an update that lands during a ServerState save out of the saved flag (#948)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java | 118 +++++++++++-
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java | 308 ++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/common/ServerStateTest.java | 44 ++++
opendj-server-legacy/src/main/java/org/opends/server/replication/common/ServerState.java | 21 +
4 files changed, 474 insertions(+), 17 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/common/ServerState.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/common/ServerState.java
index 82ba7ba..d1561b2 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/common/ServerState.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/common/ServerState.java
@@ -46,7 +46,10 @@
private final ConcurrentMap<Integer, CSN> serverIdToCSN = new ConcurrentSkipListMap<>();
/**
* Whether the state has been saved to persistent storage. It starts at true,
- * and moves to false when an update is made to the current object.
+ * and moves to false when a change is actually made to the current object -
+ * once that change is visible in {@link #serverIdToCSN}, never before, so
+ * that a reader cannot see the flag cleared for a change its own view of the
+ * map does not hold yet.
*/
private volatile boolean saved = true;
@@ -58,12 +61,18 @@
/**
* Empty the ServerState.
- * After this call the Server State will be in the same state
- * as if it was just created.
+ * After this call the Server State no longer holds any CSN. A state which
+ * held one is marked as not saved: dropping it is a change like any other,
+ * which persistent storage has yet to be told about. A state which held
+ * nothing is left alone, the way an update which changes nothing is.
*/
public void clear()
{
- serverIdToCSN.clear();
+ if (!serverIdToCSN.isEmpty())
+ {
+ serverIdToCSN.clear();
+ saved = false;
+ }
}
/**
@@ -82,8 +91,6 @@
return false;
}
- saved = false;
-
final int serverId = csn.getServerId();
while (true)
{
@@ -92,6 +99,7 @@
{
if (serverIdToCSN.putIfAbsent(serverId, csn) == null)
{
+ saved = false;
return true;
}
// oops, a concurrent modification happened, run the same process again
@@ -101,6 +109,7 @@
{
if (serverIdToCSN.replace(serverId, existingCSN, csn))
{
+ saved = false;
return true;
}
// oops, a concurrent modification happened, run the same process again
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java
index 6d2713f..4f2ab78 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PersistentServerState.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2012-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -24,6 +25,8 @@
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.ldap.ByteString;
@@ -55,6 +58,11 @@
private final DN baseDN;
private final int serverId;
private final ServerState state;
+ /**
+ * Held by the save which is writing the state to the backend. It is taken
+ * with {@link Lock#tryLock()} and never waited for: see {@link #save()}.
+ */
+ private final Lock saveLock = new ReentrantLock();
/**
* The attribute name used to store the state in the backend.
@@ -104,12 +112,70 @@
/**
* Save this object to persistent storage.
+ * <p>
+ * Only one save writes the state at a time: two of them would otherwise each
+ * take their own snapshot, and the write of the older one landing last would
+ * leave a state on disk that is both stale and marked as saved.
+ * <p>
+ * A save which finds another one writing gives up its turn instead of waiting
+ * for it, and this method never blocks. Waiting would close a lock cycle:
+ * when the write goes to the domain configuration entry it ends up in
+ * {@code LDAPReplicationDomain.applyConfigurationChange()}, which takes the
+ * very lock {@code disable()} holds while calling this method.
+ * <p>
+ * Giving up the turn loses nothing, because the state is marked as saved
+ * before the snapshot of the write in flight is taken. Whatever this save
+ * would have written is therefore either already in that snapshot, or has
+ * cleared the flag again after it was set - in which case the flag is still
+ * clear when that write completes, and the next save writes it.
*/
public void save()
{
- if (!state.isSaved())
+ if (state.isSaved())
{
- state.setSaved(updateStateEntry());
+ // Nothing to write: stay out of the way of whoever is writing.
+ return;
+ }
+
+ if (!saveLock.tryLock())
+ {
+ return;
+ }
+ try
+ {
+ if (state.isSaved())
+ {
+ // The save which just completed carried what this one came to write.
+ return;
+ }
+ /*
+ * Mark the state as saved before the snapshot that goes to the backend
+ * is taken, so that an update landing while the write is in flight
+ * clears the flag again and gets written by the next save. Marking it
+ * afterwards would swallow such an update: it is not part of the write
+ * it raced with, yet the state would look saved. The persisted state
+ * would then stay stale until some later update happened to dirty it
+ * again - which, on a domain as quiet as cn=schema, may never happen.
+ */
+ state.setSaved(true);
+ boolean written = false;
+ try
+ {
+ written = updateStateEntry();
+ }
+ finally
+ {
+ if (!written)
+ {
+ // The write reported a failure, or blew up on its way to the
+ // backend: the state is not on disk, so leave it to the next save.
+ state.setSaved(false);
+ }
+ }
+ }
+ finally
+ {
+ saveLock.unlock();
}
}
@@ -118,6 +184,16 @@
*/
public void loadState()
{
+ /*
+ * Whatever the state holds on the way in has, as far as this object knows,
+ * never been written: what follows only merges in what the backend holds.
+ * No shipped path comes in holding anything - the constructor is handed the
+ * state a ReplicationDomain has just created, and loadDataState() empties
+ * it first - so this guards a caller which does not exist yet rather than
+ * one which does.
+ */
+ final boolean hadCSNs = !state.isEmpty();
+
// try to load the state from the base entry.
SearchResultEntry stateEntry = searchBaseEntry();
if (stateEntry == null)
@@ -141,6 +217,11 @@
* Inconsistencies may append after a crash.
*/
checkAndUpdateServerState();
+
+ if (hadCSNs)
+ {
+ state.setSaved(false);
+ }
}
/**
@@ -262,9 +343,8 @@
op.setInternalOperation(true);
op.setSynchronizationOperation(true);
op.setDontSynchronize(true);
- op.run();
- final ResultCode resultCode = op.getResultCode();
+ final ResultCode resultCode = runModify(op);
if (resultCode != ResultCode.SUCCESS
&& !(resultCode == ResultCode.NO_SUCH_OBJECT && serverStateEntryDN.equals(baseDN)))
{
@@ -274,20 +354,36 @@
}
/**
- * Empty the ServerState.
- * After this call the Server State will be in the same state
- * as if it was just created.
+ * Runs the modify operation that writes the state to the backend.
+ * <p>
+ * Kept separate, and overridable, so that a test can reach the point where
+ * the snapshot has been taken but the write has not gone through yet: see
+ * {@code PersistentServerStateTest}. Do not inline it.
+ *
+ * @param op The modify operation carrying the state to be written.
+ * @return A ResultCode indicating if the operation was successful.
+ */
+ ResultCode runModify(ModifyOperationBasis op)
+ {
+ op.run();
+ return op.getResultCode();
+ }
+
+ /**
+ * Empty the ServerState in memory.
+ * <p>
+ * The emptied state is marked as not saved, so the next save writes the empty
+ * state out - which is what {@link #clear()} is after. A caller that only
+ * means to drop the in-memory copy, and expects the backend to keep what it
+ * holds, has to keep saves away until it has loaded the state back.
*/
public void clearInMemory()
{
state.clear();
- state.setSaved(false);
}
/**
- * Empty the ServerState.
- * After this call the Server State will be in the same state
- * as if it was just created.
+ * Empty the ServerState and write the emptied state to persistent storage.
*/
void clear()
{
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/common/ServerStateTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/common/ServerStateTest.java
index a532060..29237bf 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/common/ServerStateTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/common/ServerStateTest.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.common;
@@ -139,4 +140,47 @@
assertTrue(state.removeCSN(csn1Server1));
assertNull(state.getCSN(1));
}
+
+ /**
+ * An update that does not change the state must leave the saved status alone:
+ * the status is only cleared for a change that has actually been applied.
+ */
+ @Test
+ public void updateThatChangesNothingKeepsTheStateSaved() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final CSN csn = new CSN(TimeThread.getTime(), 1, 1);
+ assertTrue(state.update(csn));
+
+ state.setSaved(true);
+ assertFalse(state.update(csn), "the very same CSN is not a meaningful update");
+ assertTrue(state.isSaved(), "a duplicate CSN must not clear the saved status");
+
+ final CSN olderCSN = new CSN(csn.getTime() - 1, csn.getSeqnum(), csn.getServerId());
+ assertFalse(state.update(olderCSN), "an older CSN is not a meaningful update");
+ assertTrue(state.isSaved(), "an older CSN must not clear the saved status");
+
+ assertFalse(state.update((CSN) null));
+ assertTrue(state.isSaved(), "a null CSN must not clear the saved status");
+ }
+
+ /**
+ * Emptying the state is a change like any other: it must not leave the state
+ * looking like what persistent storage holds. Emptying one that is already
+ * empty changes nothing, and must leave the saved status alone.
+ */
+ @Test
+ public void clearMarksTheStateUnsaved() throws Exception
+ {
+ final ServerState state = new ServerState();
+ assertTrue(state.update(new CSN(TimeThread.getTime(), 1, 1)));
+ state.setSaved(true);
+
+ state.clear();
+ assertFalse(state.isSaved(), "clearing the state must not leave it marked as saved");
+
+ state.setSaved(true);
+ state.clear();
+ assertTrue(state.isSaved(), "clearing an already empty state must not clear the saved status");
+ }
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java
index 063aeb3..44ff6a2 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PersistentServerStateTest.java
@@ -13,17 +13,30 @@
*
* Copyright 2006-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.ResultCode;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static java.util.concurrent.TimeUnit.*;
import static org.opends.server.TestCaseUtils.*;
import static org.testng.Assert.*;
@@ -88,4 +101,299 @@
csn1Saved = stateSaved.getMaxCSN(1);
assertNull(csn1Saved, "csn1 has not been saved after clear for " + dn);
}
+
+ /**
+ * An update landing while the state is being written cannot be part of that
+ * write, so it must leave the state unsaved and be written by the next save.
+ * Marking the state as saved on behalf of a write that does not carry the
+ * update strands it on a quiet domain until some later change comes along.
+ */
+ @Test(dataProvider = "suffix")
+ public void updateLandingDuringSaveIsWrittenByTheNextSave(String dn) throws Exception
+ {
+ final DN baseDn = DN.valueOf(dn);
+ final ServerState serverState = new ServerState();
+ final AtomicReference<CSN> racing = new AtomicReference<>();
+
+ // the update in racing lands inside the write, once its snapshot was taken
+ final PersistentServerState state = new HookedWrite(baseDn, 1, serverState, null, entryDN -> {
+ final CSN racingCSN = racing.getAndSet(null);
+ if (racingCSN != null)
+ {
+ serverState.update(racingCSN);
+ }
+ });
+ try
+ {
+ // seeded from the state the constructor above loaded, so that both CSNs
+ // are newer than whatever the entry already holds
+ final CSNGenerator gen = new CSNGenerator(1, serverState);
+ final CSN writtenCSN = gen.newCSN();
+ final CSN racingCSN = gen.newCSN();
+ racing.set(racingCSN);
+
+ assertTrue(state.update(writtenCSN));
+
+ state.save();
+ assertEquals(loadMaxCSN(baseDn, 1), writtenCSN,
+ "the racing CSN cannot be part of the write it raced with");
+ assertFalse(serverState.isSaved(),
+ "an update that landed during the write must leave the state unsaved");
+
+ // the next tick of the checkpointer
+ state.save();
+ assertTrue(serverState.isSaved());
+ assertEquals(loadMaxCSN(baseDn, 1), racingCSN,
+ "the racing CSN must be written by the next save");
+ }
+ finally
+ {
+ state.clear();
+ }
+ }
+
+ /**
+ * A state that already holds CSNs of its own when it is loaded has not been
+ * written by this object: loading only merges in what the backend holds, so
+ * whatever came in with the state is still owed to persistent storage.
+ */
+ @Test
+ public void stateLoadedOverCSNsOfItsOwnIsNotConsideredSaved() throws Exception
+ {
+ final DN baseDn = DN.valueOf(TEST_ROOT_DN_STRING);
+ final ServerState serverState = new ServerState();
+ final CSN ownCSN = new CSNGenerator(1, serverState).newCSN();
+ assertTrue(serverState.update(ownCSN));
+ // as handed over by a caller that believes it to be persisted
+ serverState.setSaved(true);
+
+ final PersistentServerState state = new PersistentServerState(baseDn, 1, serverState);
+ try
+ {
+ assertFalse(serverState.isSaved(),
+ "a state loaded over CSNs that are not known to be on disk must not look saved");
+
+ state.save();
+ assertEquals(loadMaxCSN(baseDn, 1), ownCSN, "the CSN the state came with must reach the backend");
+ }
+ finally
+ {
+ state.clear();
+ }
+ }
+
+ /**
+ * A write reporting a failure must leave the state unsaved: the state is
+ * marked as saved before the write, and nothing else would clear that flag.
+ */
+ @Test
+ public void writeThatFailsLeavesTheStateUnsaved() throws Exception
+ {
+ final DN baseDn = DN.valueOf(TEST_ROOT_DN_STRING);
+ final ServerState serverState = new ServerState();
+ final PersistentServerState state =
+ new HookedWrite(baseDn, 1, serverState, ResultCode.UNWILLING_TO_PERFORM, null);
+ try
+ {
+ assertTrue(state.update(new CSNGenerator(1, serverState).newCSN()));
+
+ state.save();
+ assertFalse(serverState.isSaved(), "a write that failed must leave the state unsaved");
+ }
+ finally
+ {
+ new PersistentServerState(baseDn, 1, new ServerState()).clear();
+ }
+ }
+
+ /**
+ * A write blowing up on its way to the backend must leave the state unsaved,
+ * for the same reason a write reporting a failure must.
+ */
+ @Test
+ public void writeThatThrowsLeavesTheStateUnsaved() throws Exception
+ {
+ final DN baseDn = DN.valueOf(TEST_ROOT_DN_STRING);
+ final ServerState serverState = new ServerState();
+ final IllegalStateException blowUp = new IllegalStateException("the write blows up");
+ final PersistentServerState state = new HookedWrite(baseDn, 1, serverState, null, entryDN -> {
+ throw blowUp;
+ });
+ try
+ {
+ assertTrue(state.update(new CSNGenerator(1, serverState).newCSN()));
+
+ try
+ {
+ state.save();
+ fail("the write was expected to blow up");
+ }
+ catch (IllegalStateException e)
+ {
+ assertSame(e, blowUp, "save() threw something other than the failure of the write");
+ }
+ assertFalse(serverState.isSaved(), "a write that blew up must leave the state unsaved");
+ }
+ finally
+ {
+ new PersistentServerState(baseDn, 1, new ServerState()).clear();
+ }
+ }
+
+ /**
+ * A save which finds another one writing gives up its turn. It must neither
+ * reach the write - two writes at the same time can land in the order of
+ * their snapshots reversed, leaving a state on disk that is both stale and
+ * marked as saved - nor wait for the write in flight, which would close a
+ * lock cycle with the configuration listener that write calls into.
+ * <p>
+ * The write in flight is held open until the second save has run and been
+ * checked, so nothing here is inferred from a timing window: the write is
+ * provably still in flight while that save runs and returns.
+ */
+ @Test
+ public void aSaveGivesUpItsTurnWhileAnotherOneIsWriting() throws Exception
+ {
+ final DN baseDn = DN.valueOf(TEST_ROOT_DN_STRING);
+ final ServerState serverState = new ServerState();
+
+ final AtomicInteger writesStarted = new AtomicInteger();
+ final AtomicBoolean writeInFlightWasReleased = new AtomicBoolean();
+ final CountDownLatch writeInFlight = new CountDownLatch(1);
+ final CountDownLatch releaseTheWrite = new CountDownLatch(1);
+
+ final PersistentServerState state = new HookedWrite(baseDn, 1, serverState, null, entryDN -> {
+ if (writesStarted.incrementAndGet() > 1)
+ {
+ // a second write: reported by the assertions below, not from in here
+ return;
+ }
+ writeInFlight.countDown();
+ try
+ {
+ // The timeout is what turns a save that waits - as one would without
+ // the tryLock - into a failure rather than a suite that never ends.
+ releaseTheWrite.await(30, SECONDS);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("interrupted while holding the write in flight", e);
+ }
+ writeInFlightWasReleased.set(true);
+ });
+
+ final AtomicReference<Throwable> writerFailure = new AtomicReference<>();
+ final Thread writer = new Thread(state::save, "test state checkpointer");
+ writer.setDaemon(true);
+ writer.setUncaughtExceptionHandler((t, e) -> writerFailure.set(e));
+ try
+ {
+ final CSNGenerator gen = new CSNGenerator(1, serverState);
+ final CSN writtenCSN = gen.newCSN();
+ assertTrue(state.update(writtenCSN));
+ writer.start();
+ assertTrue(writeInFlight.await(30, SECONDS), "the first save never reached its write");
+
+ // dirty the state, so that the save below has something to write and
+ // would go through to the backend if nothing kept it out
+ final CSN laterCSN = gen.newCSN();
+ assertTrue(state.update(laterCSN));
+
+ state.save();
+
+ assertFalse(writeInFlightWasReleased.get(),
+ "the second save only returned once the write in flight had been released");
+ assertEquals(writesStarted.get(), 1, "the second save wrote while the first one was writing");
+ assertFalse(serverState.isSaved(),
+ "a save which gave up its turn must leave the state to the next one");
+
+ releaseTheWrite.countDown();
+ writer.join(SECONDS.toMillis(30));
+ assertFalse(writer.isAlive(), "the first save never completed");
+ assertNull(writerFailure.get(), "the first save failed: " + writerFailure.get());
+ assertEquals(loadMaxCSN(baseDn, 1), writtenCSN, "the write in flight carried its own snapshot");
+
+ // the next tick of the checkpointer, once the write in flight is over
+ state.save();
+ assertTrue(serverState.isSaved());
+ assertEquals(loadMaxCSN(baseDn, 1), laterCSN,
+ "what the save which gave up its turn had to write must be written by the next save");
+ }
+ finally
+ {
+ releaseTheWrite.countDown();
+ writer.join(SECONDS.toMillis(30));
+ // Cleared through a state of its own, unconditionally: a save on the one
+ // above would give up its turn should the writer thread still hold it,
+ // and leave a CSN of this test behind for the ones that follow.
+ new PersistentServerState(baseDn, 1, new ServerState()).clear();
+ }
+ }
+
+ /**
+ * A write which reports that the base entry is gone falls back to the domain
+ * configuration entry. When there is no such entry - no replication domain is
+ * configured over this suffix here - the state has reached no storage at all,
+ * and must not be left marked as saved.
+ */
+ @Test
+ public void writeWithNoBaseEntryAndNoConfigEntryLeavesTheStateUnsaved() throws Exception
+ {
+ final DN baseDn = DN.valueOf(TEST_ROOT_DN_STRING);
+ final ServerState serverState = new ServerState();
+ final List<String> writtenTo = new CopyOnWriteArrayList<>();
+ final PersistentServerState state =
+ new HookedWrite(baseDn, 1, serverState, ResultCode.NO_SUCH_OBJECT, writtenTo::add);
+ try
+ {
+ assertTrue(state.update(new CSNGenerator(1, serverState).newCSN()));
+
+ state.save();
+ assertEquals(writtenTo, Collections.singletonList(baseDn.toString()),
+ "the fallback wrote somewhere although no configuration entry holds this suffix");
+ assertFalse(serverState.isSaved(), "a state which reached no entry at all must be left unsaved");
+ }
+ finally
+ {
+ new PersistentServerState(baseDn, 1, new ServerState()).clear();
+ }
+ }
+
+ private CSN loadMaxCSN(DN baseDn, int serverId)
+ {
+ return new PersistentServerState(baseDn, serverId, new ServerState()).getMaxCSN(serverId);
+ }
+
+ /**
+ * A PersistentServerState whose write can be steered from the test, at the
+ * point where the snapshot that goes to the backend has already been taken.
+ */
+ private static final class HookedWrite extends PersistentServerState
+ {
+ /** Optional: when set, is returned instead of running the modify. */
+ private final ResultCode failure;
+ /**
+ * Optional: when set, runs inside the write, before the modify, and is
+ * handed the DN of the entry that write targets.
+ */
+ private final Consumer<String> insideWrite;
+
+ HookedWrite(DN baseDN, int serverId, ServerState state, ResultCode failure, Consumer<String> insideWrite)
+ {
+ super(baseDN, serverId, state);
+ this.failure = failure;
+ this.insideWrite = insideWrite;
+ }
+
+ @Override
+ ResultCode runModify(ModifyOperationBasis op)
+ {
+ if (insideWrite != null)
+ {
+ insideWrite.accept(op.getRawEntryDN().toString());
+ }
+ return failure != null ? failure : super.runModify(op);
+ }
+ }
}
--
Gitblit v1.10.0