[#1064] Replay a JE transaction ended by a lock conflict instead of giving ...

Valery Kharseko

12 hours ago f4115db80359a57a2b914259c09d7cc2d44c23cc

[#1064] Replay a JE transaction ended by a lock conflict instead of giving it up at once (#1065)

 opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/ConfigurableEnvironment.java

@@ -13,6 +13,7 @@
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2010-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.backends.jeb;

@@ -353,9 +354,11 @@
      envConfig.setConfigParam(LOG_FAULT_READ_SIZE, String.valueOf(4 * 1024));
    }

    // Disable lock timeouts, meaning that no lock wait
    // timelimit is enforced and a deadlocked operation
    // will block indefinitely.
    // Disable lock timeouts, meaning that no lock wait timelimit is enforced: a writer waiting for a lock
    // blocks until it is granted. A deadlock does not block: JE detects the cycle as soon as a wait would
    // close it and ends one of its transactions, chosen at random, with a DeadlockException - the one
    // conflict left under this setting, which JEStorage.write replays. An operator may set a timeout back
    // through ds-cfg-je-property, and a wait that long is then reported and replayed the same way.
    envConfig.setLockTimeout(0, TimeUnit.MICROSECONDS);

    //FIX https://github.com/OpenIdentityPlatform/OpenDJ/issues/53 https://docs.oracle.com/cd/E17277_02/html/java/com/sleepycat/je/EnvironmentConfig.html#FREE_DISK

 opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/JEStorage.java

@@ -91,6 +91,7 @@
import com.sleepycat.je.Durability;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.LockConflictException;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.Transaction;
import com.sleepycat.je.TransactionConfig;
@@ -682,6 +683,37 @@

  private static final int IMPORT_DB_CACHE_SIZE = 32 * MB;

  /**
   * Number of attempts a {@link #write} makes before it propagates the conflict to the caller.
   * <p>
   * It is a budget of attempts and not of time, so it is only ever reached by the conflicts that report quickly.
   * With the shipped configuration every conflict does: {@code je.lock.timeout} is 0, so a writer waiting for a
   * lock never times out, and the only conflict JE raises is the deadlock, which it detects as soon as the wait
   * would close a cycle. An operator who sets a lock timeout through {@code ds-cfg-je-property} turns a wait that
   * long into a conflict as well, and a conflict slower to report than {@link #MAX_RETRY_WINDOW_NANOS} spends the
   * whole window inside its first attempt, is granted the single replay that window's exemption guarantees, and
   * gives up on the window after two attempts rather than after this many.
   */
  static final int MAX_RETRIES = 10;

  /**
   * Wall-clock budget the replays of a {@link #write} may spend, in nanoseconds. It is checked between attempts,
   * so an attempt already running is never interrupted, and never before one replay has been made: the loop
   * returns after at most this window plus two attempts. It bounds the conflicts that are slow to report, which
   * {@link #MAX_RETRIES} alone does not - an operation whose own work takes seconds would otherwise multiply that
   * wait by the attempt count.
   */
  static final long MAX_RETRY_WINDOW_NANOS = 10L * 1000L * 1000L * 1000L; //10 s

  /**
   * Upper bound of the random delay before the second attempt, in milliseconds; it doubles with every attempt.
   * The ladder is {@code PDBStorage}'s, kept so that the two engines back off alike.
   */
  private static final double BASE_SLEEP_ON_RETRY_MS = 50.0;

  /** Upper bound the doubled delay is capped at, in milliseconds. */
  private static final double MAX_SLEEP_ON_RETRY_MS = 1000.0;

  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** Use read committed isolation instead of the default which is repeatable read. */
@@ -700,6 +732,10 @@
  private DiskSpaceMonitor diskMonitor;
  private StorageStatus storageStatus = StorageStatus.working();
  private final ConcurrentMap<TreeName, Database> trees = new ConcurrentHashMap<>();
  /** Attempt bound of a {@link #write}, {@link #MAX_RETRIES} outside the tests. */
  private final int maxRetries;
  /** Wall-clock bound of a {@link #write}, {@link #MAX_RETRY_WINDOW_NANOS} outside the tests. */
  private final long retryWindowNanos;

  /**
   * Creates a new JE storage with the provided configuration.
@@ -714,7 +750,35 @@
  // Public as PDBStorage's is: a pluggable backend test which runs the same case over both storages builds them.
  public JEStorage(final JEBackendCfg cfg, ServerContext serverContext) throws ConfigException
  {
    this(cfg, serverContext, MAX_RETRIES, MAX_RETRY_WINDOW_NANOS);
  }

  /**
   * Creates a new JE storage whose replay bounds are the given ones rather than {@link #MAX_RETRIES} and
   * {@link #MAX_RETRY_WINDOW_NANOS}.
   * <p>
   * Only a test builds one of these, and it does so to stop the two bounds racing each other: with the shipped
   * values a run of replays spends a random share of the window on backoff alone, so a test of the attempt cap
   * can be ended by the window on a loaded machine, and a test of the window has to spend seconds of build time
   * to reach it.
   *
   * @param cfg
   *          The configuration.
   * @param serverContext
   *          This server instance context
   * @param maxRetries
   *          Number of attempts a write makes before it propagates the conflict to the caller.
   * @param retryWindowNanos
   *          Wall-clock budget the replays of a write may spend, in nanoseconds.
   * @throws ConfigException
   *           if memory cannot be reserved
   */
  JEStorage(final JEBackendCfg cfg, ServerContext serverContext, int maxRetries, long retryWindowNanos)
      throws ConfigException
  {
    this.serverContext = serverContext;
    this.maxRetries = maxRetries;
    this.retryWindowNanos = retryWindowNanos;
    backendDirectory = getBackendDirectory(cfg);
    config = cfg;
    cfg.addJEChangeListener(this);
@@ -964,28 +1028,144 @@
    return treeName.toString();
  }

  /**
   * {@inheritDoc}
   * <p>
   * A transaction JE ends with a {@link LockConflictException} is replayed, bounded twice: by {@link #MAX_RETRIES}
   * attempts and by the {@link #MAX_RETRY_WINDOW_NANOS} wall-clock window, whichever is spent first - except that
   * the window alone never ends the replays before one has been made. It is bounded for the reason
   * {@code PDBStorage} bounds its loop: the configuration change paths of the pluggable backend hold an entry
   * container's exclusive lock across this method, and every reader of that suffix then waits - untimed and
   * uninterruptibly - until it returns, so a conflict that never clears would park every worker thread of that
   * suffix rather than fail one operation.
   * <p>
   * JE raises the conflict from inside the operation - a record read or write, never the commit, which takes no
   * lock - and the transaction wraps it in a {@link StorageRuntimeException}, which is unwrapped below before it is
   * matched. The operation may catch it there; the transaction is then abort-only and {@code commit()} raises the
   * conflict again, bare, so an attempt which swallowed its conflict commits nothing and is replayed all the same.
   * With the shipped configuration the only conflict is the deadlock: {@code je.lock.timeout} is 0, so a writer
   * waits for a lock rather than times out, and JE ends one transaction of a cycle - chosen at random - as soon as
   * a wait would close it. A lock timeout set through {@code ds-cfg-je-property} makes a wait that long a conflict
   * as well.
   * <p>
   * The replay backs off first, though JE would make it wait for the locks it lost anyway: JE locks a record by
   * the LSN of its current version, and the abort of the victim, which hands the waiting survivor the lock it
   * asked for, undoes the version that lock belongs to - the survivor then has to lock the version the undo put
   * back, and a replay which comes back at once wins that race, holds the record when the survivor asks again,
   * and the same deadlock forms with the roles drawn afresh. JE's own retry example sleeps before retrying for
   * this reason; without the sleep the deadlock of two writers was seen to form three times in a row.
   * <p>
   * An interrupt reaches the loop only in that sleep, and the loop does not restore the flag the sleep cleared:
   * JE invalidates the whole environment when a thread carrying the interrupt flag touches it - the transaction
   * registry's latch is acquired interruptibly - and the caller's own failure road makes such a call
   * ({@code EntryContainer.writeTrustState}). The interrupt is reported instead, with the conflict being replayed,
   * as the suppressed exceptions of the {@link StorageRuntimeException} thrown.
   * <p>
   * Once the bound is spent the conflict is reported as a {@link StorageRuntimeException} naming the backend, the
   * attempts spent, the time they took and which of the two bounds ran out. It carries the conflict as a
   * suppressed exception rather than as its cause: a cause is unwrapped below and thrown in its place, and
   * {@code EntryContainer.throwAllowedExceptionTypes} likewise passes a {@link StorageRuntimeException} through
   * untouched only while it has no cause. Given a cause, both hand the caller the bare conflict instead, whose
   * message is JE's account of its lock table - all an LDAP client used to be told after the single attempt.
   */
  @Override
  public void write(final WriteOperation operation) throws Exception
  {
    final long startedAt = System.nanoTime();
    final long giveUpAt = startedAt + retryWindowNanos;
    for (int attempt = 1;; attempt++)
    {
      final LockConflictException conflict;
    final Transaction txn = beginTransaction();
    try
    {
        try
        {
      operation.run(newWriteableTransaction(txn));
      commit(txn);
          return;
    }
    catch (final StorageRuntimeException e)
    {
          throw unwrap(e);
        }
      }
      catch (final LockConflictException e)
      {
        conflict = e;
      }
      finally
      {
        abort(txn);
      }
      // System.nanoTime() - giveUpAt is the overflow safe form of the comparison, and attempt > 1 keeps the window
      // from ending the loop before a single replay: a lock timeout set by the operator can outlast the window on
      // its own, and it is the attempt after that one which is likeliest to succeed, the transaction that blocked
      // it having just finished. One clock sample for both, so that the elapsed time reported is the one the give
      // up was decided on
      final long now = System.nanoTime();
      final boolean capSpent = attempt >= maxRetries;
      if (capSpent || (attempt > 1 && now - giveUpAt >= 0))
      {
        final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(now - startedAt);
        // which of the two bounds was spent, so that the config change paths - which report this as a bare stack
        // trace with no message id - say whether raising the attempts or the window is what would have helped
        final String boundSpent = capSpent ? "attempt cap" : "retry window";
        final StorageRuntimeException spent = new StorageRuntimeException(
            "je: backend '" + config.getBackendId() + "' did not apply the transaction after " + attempt
                + " attempts in " + elapsedMs + " ms, the " + boundSpent + " being spent; the last conflict was "
                + conflict);
        spent.addSuppressed(conflict);
        // warned once, at exhaustion only: a conflict is answered by a replay, and the trace below is what says
        // so. stackTraceToSingleLineString, the form the config change paths report this exception with, walks
        // the causes and never prints a suppressed exception, so this line is the only rendering that carries the
        // stack of the conflict
        logger.warn(LocalizableMessage.raw("je: giving up on the transaction of backend '%s' after %d attempts"
            + " in %d ms, the %s being spent: %s", config.getBackendId(), attempt, elapsedMs, boundSpent,
            stackTraceToSingleLineString(conflict)));
        throw spent;
      }
      if (logger.isTraceEnabled())
      {
        logger.trace("je: replaying the transaction after %s, attempt %d of %d", conflict, attempt, maxRetries);
      }
      try
      {
        // slept for after the finally has ended the rolled back transaction, so that nothing of it is held
        // through the delay; random and growing with every attempt, so that two writers replaying the same
        // deadlock do not come back in step
        Thread.sleep(retryDelayMillis(attempt));
      }
      catch (final InterruptedException e)
      {
        // the flag stays cleared - see above - and the conflict being replayed is reported next to the interrupt,
        // wrapped the way the exhausted loop wraps it and for the same reason: every caller strips a cause off
        final StorageRuntimeException interrupted = new StorageRuntimeException(
            "je: backend '" + config.getBackendId() + "' was interrupted while replaying the transaction after "
                + attempt + " attempts; the last conflict was " + conflict);
        interrupted.addSuppressed(conflict);
        interrupted.addSuppressed(e);
        throw interrupted;
      }
    }
  }

  /** Returns the randomized delay before the given attempt is replayed, doubling with each attempt up to a cap. */
  // package private like the PDBStorage copy it is taken from, so that a test can pin the growth and the cap
  static long retryDelayMillis(int attempt)
  {
    final double bound = Math.min(MAX_SLEEP_ON_RETRY_MS, BASE_SLEEP_ON_RETRY_MS * (1 << Math.min(attempt - 1, 5)));
    return (long) (Math.random() * bound);
  }

  private static Exception unwrap(final StorageRuntimeException e) throws Exception
  {
      if (e.getCause() != null)
      {
        throw (Exception) e.getCause();
      }
      throw e;
    }
    finally
    {
      abort(txn);
    }
  }

  private Transaction beginTransaction()
  {

 opendj-server-legacy/src/test/java/org/opends/server/backends/jeb/JEStorageTest.java

@@ -17,13 +17,20 @@

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown;
import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.forgerock.opendj.ldap.ByteString.valueOfUtf8;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.opends.server.util.CollectionUtils.newTreeSet;

import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.config.server.ConfigException;
import org.forgerock.opendj.ldap.ByteString;
@@ -34,6 +41,7 @@
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.ReadOperation;
import org.opends.server.backends.pluggable.spi.ReadableTransaction;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.WriteOperation;
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
@@ -45,9 +53,21 @@
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import com.sleepycat.je.LockConflictException;
import com.sleepycat.je.LockTimeoutException;

/**
 * Tests what a {@link JEStorage} takes as it opens and gives back when the open fails - the twin
 * of the same cases on {@code PDBStorageTest}.
 * of the same cases on {@code PDBStorageTest} - and the replay of a {@link JEStorage#write} whose
 * transaction JE ends with a {@link LockConflictException}.
 * <p>
 * The conflicts are the engine's own. A deadlock is made by two writers locking two records in opposite order,
 * which JE resolves by throwing at a random victim; a conflict on every attempt is made by a transaction which
 * keeps a record locked while the storage runs with a {@code je.lock.timeout} - the shipped configuration sets
 * none, so a writer waits for a lock rather than times out, but an operator can set one through
 * {@code ds-cfg-je-property}, and JE then reports plain contention as a {@link LockTimeoutException}, which it
 * documents as "abort and retry" just like the deadlock. A {@code DeadlockException} cannot be built by a test:
 * its constructor needs the internal locker it invalidates.
 */
@SuppressWarnings("javadoc")
public class JEStorageTest extends DirectoryServerTestCase
@@ -59,6 +79,16 @@
   * what a backend whose directory the server cannot use meets.
   */
  private static final String BLOCKED_DB_DIRECTORY = BACKEND_ID + "-blocked";
  /** A window no run of replays can spend, so that a test of the attempt cap is only ever ended by the cap. */
  private static final long UNREACHABLE_RETRY_WINDOW_NANOS = 300L * 1000L * 1000L * 1000L; //5 min
  /** A window a single attempt outlasts, so that a test of the window reaches it without seconds of build time. */
  private static final long SHORT_RETRY_WINDOW_NANOS = 200L * 1000L * 1000L; //200 ms
  /** A lock wait shorter than any bound, so that a conflict is reported promptly on every attempt. */
  private static final String SHORT_LOCK_TIMEOUT = "20 ms";
  /** A lock wait longer than {@link #SHORT_RETRY_WINDOW_NANOS}, so that one attempt outlasts the window on its own. */
  private static final String LOCK_TIMEOUT_LONGER_THAN_SHORT_WINDOW = "300 ms";
  /** How long a test waits for a thread it started, in seconds; well past any bound the tests configure. */
  private static final long WAIT_SECONDS = 60;

  private final TreeName treeName = new TreeName("dc=test", "test");
  private ServerContext serverContext;
@@ -82,6 +112,7 @@
    // tearDown(), and this class shares a fixed db-directory across methods and across builds
    storage.removeStorageFiles();
    storage.open(AccessMode.READ_WRITE);
    createTreeWithTwoRecords();
  }

  @AfterMethod
@@ -197,7 +228,6 @@
  @Test
  public void openingAnOpenStorageIsRefusedAndTakesNothing() throws Exception
  {
    createTree();
    final MemoryQuota quota = serverContext.getMemoryQuota();
    final long availableBefore = quota.getAvailableMemory();
    try
@@ -237,7 +267,469 @@
    directory.getParentFile().delete();
  }

  private void createTree() throws Exception
  /**
   * Replaces the storage under test with one bounded by the given values and, when a lock timeout is given, one
   * whose lock waits end in a {@link LockTimeoutException} after that long, the way an operator's
   * {@code ds-cfg-je-property: je.lock.timeout=...} makes them end. Bounding the loop stops the two bounds racing
   * each other: with the shipped values a run of replays spends a random share of the window on backoff alone,
   * so an attempt cap test can be ended by the ten second window instead, and a window test has to make every
   * attempt outlast seconds of that window to reach it.
   */
  private void reopenWithReplayBounds(int maxRetries, long retryWindowNanos, String lockTimeout) throws Exception
  {
    closeAndRemove(storage);
    storage = new JEStorage(createBackendCfg(lockTimeout), serverContext, maxRetries, retryWindowNanos);
    storage.open(AccessMode.READ_WRITE);
    createTreeWithTwoRecords();
  }

  /**
   * Two writers which lock the two records in opposite order, each holding its first record's write lock before
   * asking for the other's. JE detects the cycle and ends one of the two transactions - chosen at random - with a
   * {@code DeadlockException}; the other is granted its lock once the victim has aborted.
   */
  private final class OpposedWriter implements Runnable
  {
    private final String name;
    private final String first;
    private final String second;
    private final CyclicBarrier bothHoldTheirFirstLock;
    private final boolean swallowTheConflict;
    final AtomicInteger attempts = new AtomicInteger();
    final AtomicReference<Throwable> failure = new AtomicReference<>();
    final Thread thread;

    OpposedWriter(String name, String first, String second, CyclicBarrier bothHoldTheirFirstLock,
        boolean swallowTheConflict)
    {
      this.name = name;
      this.first = first;
      this.second = second;
      this.bothHoldTheirFirstLock = bothHoldTheirFirstLock;
      this.swallowTheConflict = swallowTheConflict;
      this.thread = new Thread(this, name);
    }

    @Override
    public void run()
    {
      try
      {
        storage.write(new WriteOperation()
        {
          @Override
          public void run(WriteableTransaction txn) throws Exception
          {
            final int attempt = attempts.incrementAndGet();
            txn.put(treeName, valueOfUtf8(first), valueOfUtf8(name + attempt));
            if (attempt == 1)
            {
              // only the first attempt meets the other writer there: a replay would wait on the barrier forever
              bothHoldTheirFirstLock.await(WAIT_SECONDS, TimeUnit.SECONDS);
            }
            try
            {
              txn.put(treeName, valueOfUtf8(second), valueOfUtf8(name + attempt));
            }
            catch (StorageRuntimeException e)
            {
              if (!swallowTheConflict)
              {
                throw e;
              }
              // an operation which catches what the transaction raised, the way DN2URI.targetEntryReferrals does
            }
          }
        });
      }
      catch (Throwable e)
      {
        failure.set(e);
      }
    }

    void startAndJoin(OpposedWriter other) throws InterruptedException
    {
      thread.start();
      other.thread.start();
      thread.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
      other.thread.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
      assertThat(thread.isAlive()).as(name + " finished").isFalse();
      assertThat(other.thread.isAlive()).as(other.name + " finished").isFalse();
    }
  }

  /**
   * Both writers commit, and the victim's transaction was rolled back whole and replayed. How many times is
   * JE's to decide: the abort of the victim hands the survivor the lock it waited for, but on the record version
   * the abort undoes, so the survivor has to lock the version put back, and a replay which wins that race forms
   * the deadlock again with the roles drawn afresh - the backoff makes that rare, not impossible.
   */
  private void assertDeadlockVictimReplayedAndBothCommitted(OpposedWriter ab, OpposedWriter ba) throws Exception
  {
    assertThat(ab.failure.get()).as("ab").isNull();
    assertThat(ba.failure.get()).as("ba").isNull();
    assertThat(ab.attempts.get() + ba.attempts.get()).as("at least one of the two was replayed")
        .isGreaterThanOrEqualTo(3);
    assertThat(Math.max(ab.attempts.get(), ba.attempts.get())).isLessThanOrEqualTo(JEStorage.MAX_RETRIES);
    // whichever committed last wrote both records with the number of the attempt which committed, so the
    // records agree - which they would not, had a victim's first record survived its abort
    final ByteString a = read("a");
    assertThat(a).isEqualTo(read("b"));
    final OpposedWriter last = a.toString().startsWith("ab") ? ab : ba;
    assertThat(a).isEqualTo(valueOfUtf8(last.name + last.attempts.get()));
  }

  @Test
  public void testDeadlockVictimIsReplayed() throws Exception
  {
    final CyclicBarrier barrier = new CyclicBarrier(2);
    final OpposedWriter ab = new OpposedWriter("ab", "a", "b", barrier, false);
    final OpposedWriter ba = new OpposedWriter("ba", "b", "a", barrier, false);

    ab.startAndJoin(ba);

    assertDeadlockVictimReplayedAndBothCommitted(ab, ba);
  }

  /**
   * JE raises a lock conflict from inside the operation, and an operation may catch it there. The transaction is
   * then abort-only, and it is {@code commit()} which raises the conflict again - so the loop has to treat a
   * conflict raised by the commit as one to replay, not only one raised by the operation. Unlike PersistIt, an
   * attempt which swallowed its conflict therefore commits nothing.
   */
  @Test
  public void testConflictSwallowedInsideTheOperationIsRaisedAgainByCommitAndReplayed() throws Exception
  {
    final CyclicBarrier barrier = new CyclicBarrier(2);
    final OpposedWriter ab = new OpposedWriter("ab", "a", "b", barrier, true);
    final OpposedWriter ba = new OpposedWriter("ba", "b", "a", barrier, true);

    ab.startAndJoin(ba);

    assertDeadlockVictimReplayedAndBothCommitted(ab, ba);
  }

  /**
   * A transaction of its own which keeps a record write-locked until released, so that every attempt of a write
   * asking for that record ends the way the storage's lock timeout ends it.
   */
  private final class LockHolder implements Runnable
  {
    private final String key;
    private final CountDownLatch held = new CountDownLatch(1);
    private final CountDownLatch release = new CountDownLatch(1);
    private final AtomicReference<Throwable> failure = new AtomicReference<>();
    private final Thread thread = new Thread(this, "lock holder");

    LockHolder(String key)
    {
      this.key = key;
    }

    @Override
    public void run()
    {
      try
      {
        storage.write(new WriteOperation()
        {
          @Override
          public void run(WriteableTransaction txn) throws Exception
          {
            txn.put(treeName, valueOfUtf8(key), valueOfUtf8("held"));
            held.countDown();
            release.await(WAIT_SECONDS, TimeUnit.SECONDS);
          }
        });
      }
      catch (Throwable e)
      {
        failure.set(e);
      }
    }

    LockHolder start() throws InterruptedException
    {
      thread.start();
      assertThat(held.await(WAIT_SECONDS, TimeUnit.SECONDS)).as("the holder took the lock").isTrue();
      return this;
    }

    void releaseAndJoin() throws InterruptedException
    {
      release.countDown();
      thread.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
      assertThat(thread.isAlive()).as("the holder finished").isFalse();
      assertThat(failure.get()).as("the holder's own write").isNull();
    }
  }

  @Test
  public void testWriteGivesUpAfterTheAttemptCap() throws Exception
  {
    // the message is the same at any cap, so this one is spent in two backoffs rather than in the shipped ladder
    final int maxRetries = 3;
    reopenWithReplayBounds(maxRetries, UNREACHABLE_RETRY_WINDOW_NANOS, SHORT_LOCK_TIMEOUT);
    final LockHolder holder = new LockHolder("a").start();
    try
    {
      final AtomicInteger attempts = new AtomicInteger();
      try
      {
        storage.write(new WriteOperation()
        {
          @Override
          public void run(WriteableTransaction txn) throws Exception
          {
            attempts.incrementAndGet();
            txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("abandoned"));
          }
        });
        failBecauseExceptionWasNotThrown(StorageRuntimeException.class);
      }
      catch (StorageRuntimeException e)
      {
        assertThat(e.getMessage()).contains("JEStorageTest").contains(maxRetries + " attempts");
        // and which of the two bounds ran out, since the attempt count alone does not say
        assertThat(e.getMessage()).contains("attempt cap");
        // write() unwraps a StorageRuntimeException that carries a cause, which would replace this message with
        // the bare conflict, and it is the message the config change paths report
        assertThat(e.getCause()).isNull();
        assertThat(e.getSuppressed()).hasSize(1);
        assertThat(e.getSuppressed()[0]).isInstanceOf(LockTimeoutException.class);
      }
      assertThat(attempts.get()).isEqualTo(maxRetries);
    }
    finally
    {
      holder.releaseAndJoin();
    }
    assertThat(read("a")).isEqualTo(valueOfUtf8("held"));
  }

  @Test
  public void testWriteIsReplayedUntilTheConflictClears() throws Exception
  {
    reopenWithReplayBounds(JEStorage.MAX_RETRIES, UNREACHABLE_RETRY_WINDOW_NANOS, SHORT_LOCK_TIMEOUT);
    final LockHolder holder = new LockHolder("a").start();
    try
    {
      final AtomicInteger attempts = new AtomicInteger();
      storage.write(new WriteOperation()
      {
        @Override
        public void run(WriteableTransaction txn) throws Exception
        {
          if (attempts.incrementAndGet() == 4)
          {
            // released, and committed, before the lock is asked for, so that this attempt is the one which
            // gets it rather than the one which times out on the holder's commit
            holder.releaseAndJoin();
          }
          txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("applied"));
        }
      });
      assertThat(attempts.get()).isEqualTo(4);
    }
    finally
    {
      holder.releaseAndJoin();
    }
    assertThat(read("a")).isEqualTo(valueOfUtf8("applied"));
  }

  /**
   * With a lock timeout longer than the window, a single attempt outlasts the whole window. Giving up on the
   * window alone would then replay nothing, in the very case where the replay is likeliest to succeed: the
   * transaction that was blocking this one has just finished.
   */
  @Test
  public void testWriteIsReplayedOnceWhenTheFirstAttemptOutlastsTheWindow() throws Exception
  {
    reopenWithReplayBounds(JEStorage.MAX_RETRIES, SHORT_RETRY_WINDOW_NANOS, LOCK_TIMEOUT_LONGER_THAN_SHORT_WINDOW);
    final LockHolder holder = new LockHolder("a").start();
    try
    {
      final AtomicInteger attempts = new AtomicInteger();
      storage.write(new WriteOperation()
      {
        @Override
        public void run(WriteableTransaction txn) throws Exception
        {
          if (attempts.incrementAndGet() == 2)
          {
            holder.releaseAndJoin();
          }
          txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("outlasted"));
        }
      });
      assertThat(attempts.get()).isEqualTo(2);
    }
    finally
    {
      holder.releaseAndJoin();
    }
    assertThat(read("a")).isEqualTo(valueOfUtf8("outlasted"));
  }

  @Test
  public void testWriteGivesUpOnTheWindowWhenAttemptsAreSlow() throws Exception
  {
    reopenWithReplayBounds(JEStorage.MAX_RETRIES, SHORT_RETRY_WINDOW_NANOS, LOCK_TIMEOUT_LONGER_THAN_SHORT_WINDOW);
    final LockHolder holder = new LockHolder("a").start();
    try
    {
      final AtomicInteger attempts = new AtomicInteger();
      try
      {
        storage.write(new WriteOperation()
        {
          @Override
          public void run(WriteableTransaction txn) throws Exception
          {
            attempts.incrementAndGet();
            // a conflict this slow to report spends the wall clock window long before the attempt cap
            txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("abandoned"));
          }
        });
        failBecauseExceptionWasNotThrown(StorageRuntimeException.class);
      }
      catch (StorageRuntimeException e)
      {
        // the window is what ended it, and it says so: an assertion on the attempt count alone would also pass
        // for a give up on attempt 1, which is the regression the attempt > 1 exemption exists to prevent
        assertThat(e.getMessage()).contains("retry window");
      }
      // one attempt beyond the first: the first spends the window, the exemption grants the replay, and the
      // check after that replay is the one that gives up
      assertThat(attempts.get()).isEqualTo(2);
    }
    finally
    {
      holder.releaseAndJoin();
    }
    assertThat(read("a")).isEqualTo(valueOfUtf8("held"));
  }

  /**
   * An interrupt reaches the loop in its backoff sleep alone, and it must not reach JE afterwards: a thread
   * carrying the interrupt flag invalidates the whole environment on its next call - the transaction registry's
   * latch is acquired interruptibly - which is why the loop leaves the flag as the sleep cleared it, and reports
   * the interrupt with the conflict instead. Delivering the interrupt to the sleep alone takes a write which makes
   * no call of JE at all while the flag is set: this one runs on the storage's import environment, whose writes
   * open no transaction, with an operation which raises a conflict JE raised earlier rather than asking JE for a
   * new one - a transaction aborted with the flag set would take the environment down before the sleep is reached.
   */
  @Test
  public void testInterruptedWriteReportsTheConflictItWasReplaying() throws Exception
  {
    final LockConflictException conflict = aConflictOfJEsOwn();
    // the storage that raised it gave up at its first attempt; this one is bounded as shipped
    closeAndRemove(storage);
    storage = new JEStorage(createBackendCfg(), serverContext);
    storage.startImport();

    final AtomicInteger attempts = new AtomicInteger();
    final boolean interruptedAfterwards;
    try
    {
      storage.write(new WriteOperation()
      {
        @Override
        public void run(WriteableTransaction txn) throws Exception
        {
          attempts.incrementAndGet();
          Thread.currentThread().interrupt();
          throw new StorageRuntimeException(conflict);
        }
      });
      failBecauseExceptionWasNotThrown(StorageRuntimeException.class);
      return;
    }
    catch (StorageRuntimeException e)
    {
      interruptedAfterwards = Thread.interrupted();
      // the conflict, not the interrupt, is what the caller is told about - but through the same shape the
      // exhausted loop uses, since a bare conflict reaches every caller as its own class name
      assertThat(e.getMessage()).contains("JEStorageTest").contains("interrupted");
      assertThat(e.getSuppressed()).contains(conflict).hasAtLeastOneElementOfType(InterruptedException.class);
      assertThat(e.getCause()).isNull();
    }
    finally
    {
      Thread.interrupted();
    }
    // the flag the sleep cleared stays clear: restored, it would end the environment on the caller's next call
    assertThat(interruptedAfterwards).as("interrupt flag after the write").isFalse();
    // one attempt even though the first backoff is a random 0-49 ms and so is sometimes 0: Thread.sleep() checks
    // the interrupt flag before it checks for a zero duration, so the replay is never reached
    assertThat(attempts.get()).isEqualTo(1);
  }

  /** A conflict raised by JE itself: the lock timeout a write gave up on at its first attempt. */
  private LockConflictException aConflictOfJEsOwn() throws Exception
  {
    reopenWithReplayBounds(1, UNREACHABLE_RETRY_WINDOW_NANOS, SHORT_LOCK_TIMEOUT);
    final LockHolder holder = new LockHolder("a").start();
    try
    {
      storage.write(new WriteOperation()
      {
        @Override
        public void run(WriteableTransaction txn) throws Exception
        {
          txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("abandoned"));
        }
      });
      throw new AssertionError("the write was applied although its record was held");
    }
    catch (StorageRuntimeException e)
    {
      assertThat(e.getSuppressed()).hasSize(1);
      return (LockConflictException) e.getSuppressed()[0];
    }
    finally
    {
      holder.releaseAndJoin();
    }
  }

  /**
   * The delay grows with the attempt and stays under the cap, so that a contention the first delays did not
   * outlast still has a chance to clear without the replays overrunning the window on sleep alone.
   */
  @Test
  public void testRetryDelayGrowsAndStaysBounded()
  {
    long previousBound = 0;
    for (int attempt = 1; attempt <= JEStorage.MAX_RETRIES; attempt++)
    {
      long bound = 0;
      for (int i = 0; i < 100; i++)
      {
        final long delay = JEStorage.retryDelayMillis(attempt);
        assertThat(delay).as("attempt %d", attempt).isGreaterThanOrEqualTo(0).isLessThan(1000);
        bound = Math.max(bound, delay);
      }
      if (attempt == 1)
      {
        assertThat(bound).as("attempt 1 delays past the first tier").isLessThan(50);
      }
      assertThat(bound).as("attempt %d did not grow past attempt %d", attempt, attempt - 1)
          .isGreaterThanOrEqualTo(previousBound / 2);
      previousBound = bound;
    }
    // and the growth is real rather than a delay that never leaves the first tier
    long grown = 0;
    for (int i = 0; i < 100; i++)
    {
      grown = Math.max(grown, JEStorage.retryDelayMillis(JEStorage.MAX_RETRIES));
    }
    assertThat(grown).as("the last attempts still sleep within the first attempt's bound").isGreaterThan(500);
  }

  private void createTreeWithTwoRecords() throws Exception
  {
    storage.write(new WriteOperation()
    {
@@ -245,6 +737,8 @@
      public void run(WriteableTransaction txn) throws Exception
      {
        txn.openTree(treeName, true);
        txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("0"));
        txn.put(treeName, valueOfUtf8("b"), valueOfUtf8("0"));
      }
    });
  }
@@ -274,4 +768,18 @@
    when(backendCfg.getDBNumLockTables()).thenReturn(63);
    return backendCfg;
  }

  /**
   * The configuration of the storage under test, with the lock timeout an operator would set through
   * {@code ds-cfg-je-property}, or the shipped one - none - when null.
   */
  private static JEBackendCfg createBackendCfg(String lockTimeout)
  {
    final JEBackendCfg backendCfg = createBackendCfg();
    if (lockTimeout != null)
    {
      when(backendCfg.getJEProperty()).thenReturn(newTreeSet("je.lock.timeout=" + lockTimeout));
    }
    return backendCfg;
  }
}