[#916] Keep an update that lands during a ServerState save out of the saved...

Valery Kharseko

3 hours ago 5d176c691527e3fe4bc529ff8947f3b3648970bd

 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

 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()
  {

 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");
  }
}

 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);
    }
  }
}