[#915] Bound the wait of a write transaction for a row lock another session...

Valery Kharseko

13 hours ago 47769981c6fb596f892f0b2ab64c24c8e22b38e9

[#915] Bound the wait of a write transaction for a row lock another session holds (#1010)

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java

@@ -1325,6 +1325,14 @@
    private volatile boolean poolable;

    /**
     * The session value the row lock bound of this backend displaces on this connection, read once and
     * kept for its life: see {@link #rowLockBoundDisplaces()}. Volatile for the reason
     * {@link #poolable} is - it is written by one borrower and read by the next, which is another
     * thread, and the value is a plain field of an object the pool hands between them.
     */
    private volatile Long rowLockBoundDisplaces;

    /**
     * When this connection last answered the database, as a {@link System#nanoTime()} reading:
     * established - the login and the two round trips that set it up have just answered - or
     * validated. It is never stamped on the way back into the pool, although that is where a
@@ -1369,7 +1377,7 @@
    /**
     * Keeps this connection out of the pool: it serves the borrower holding it and is closed rather
     * than pooled when that borrow ends. For a borrower that left something of its own on the session
     * and could not take it off again - {@code JDBCStorage.restoreDdlLockBound()} is the one that
     * and could not take it off again - {@code JDBCStorage.restoreLockBound()} is the one that
     * does (#885) - where the blast radius is then this one connection instead of every borrow it
     * would have served after this one.
     */
@@ -1377,6 +1385,26 @@
        poolable = false;
    }

    /**
     * What this session carried before the row lock bound of {@code JDBCStorage.write()} first
     * displaced it, or null while that has not been read yet (#915).
     * <p>
     * Remembered for the life of the connection because that bound is on the hot path - one write of
     * the server, one arming - while the readback it saves is a round trip. Only this backend writes
     * that setting on a connection of this pool, every write puts the value back before the
     * connection is released, and a connection whose restore failed is kept out of the pool by
     * {@link #keepOutOfThePool} rather than handed on: so a value remembered here cannot outlive the
     * session that answered it. Not reset on borrow for the same reason - it describes the session,
     * which outlives every borrow of it.
     */
    Long rowLockBoundDisplaces() {
        return rowLockBoundDisplaces;
    }

    void rowLockBoundDisplaces(Long previous) {
        rowLockBoundDisplaces = previous;
    }

    /** Gives back the right to hold this connection, once and only if it was taken. */
    void releasePermit() {
        if (metered && permitReleased.compareAndSet(false, true)) {

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java

@@ -49,7 +49,9 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.function.Supplier;

import static org.opends.server.backends.pluggable.spi.StorageUtils.addErrorMessage;
import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;
@@ -63,24 +65,29 @@

    /**
     * Wall-clock budget the replays of a {@link #write} may spend, in nanoseconds, measured from the start of the
     * first attempt. It is checked between attempts, so an attempt already running is never interrupted, and it
     * applies from the first check, with the single exception {@link #grantedPastTheWindow} describes: a conflict
     * its engine reports promptly is granted one replay whatever the clock says, because the lock wait that
     * precedes such a conflict is charged to the attempt and is unbounded on three of the four engines here, so no
     * window survives it. It bounds what {@link #MAX_RETRIES} alone does not - MySQL reports a lock wait timeout
     * only after innodb_lock_wait_timeout, 50 s by default and not overridden here, so ten attempts would park a
     * worker thread for eight minutes where one releases it after 50 s.
     * first attempt. It is checked between attempts, so an attempt already running is never interrupted: the loop
     * returns after at most this window plus one attempt. It bounds the conflicts that are slow to report, which
     * {@link #MAX_RETRIES} alone does not - MySQL reports a lock wait timeout only after innodb_lock_wait_timeout,
     * 50 s by default, so ten attempts would park a worker thread for eight minutes where a single one released it
     * after 50 s.
     * <p>
     * What that costs, stated rather than left to be read off a test row: at the stock innodb_lock_wait_timeout a
     * MySQL lock wait timeout is reported at ~50 s, which is past this window on the first check, so such a write
     * is never replayed at all - the one conflict class of the set that a MySQL deployment sees most, and the one
     * whose replay would most reliably succeed. It is the deliberate half of the trade the other half of which is
     * #903: one bounded wait beats two, and a deployment that tunes innodb_lock_wait_timeout below this window
     * gets its replays back. The trade only exists because nothing here bounds the attempt: with a session lock
     * timeout on the transaction connection (#915) every wait would be shorter than this window, the tuned-down
     * case would become the normal one, and this window would govern both classes with no grant needed at all.
     * A window is only a bound on the replays while an attempt is shorter than it, and what makes an attempt long
     * is the lock wait ahead of the conflict, charged to the attempt that hit it. That wait is bounded inside the
     * attempt by {@link #ROW_LOCK_TIMEOUT_PROPERTY} - including the innodb_lock_wait_timeout above, which is taken
     * down to it for the length of the write - so a conflicted write is replayed within this window rather than
     * spending the whole of it on one attempt and being refused a replay it could have made (#903, #915). The
     * replays a write gets are therefore this window divided by that bound: three at both defaults.
     * <p>
     * Two cases still spend it in one attempt, and they are the two nothing bounds: an oracle session waiting in a
     * row-lock enqueue, which has no setting to bound it, and a deployment that sets that property to 0. There the
     * loop behaves as it did before #915 - the window applies from the first check, with the single exception
     * {@link #grantedPastTheWindow} describes: a conflict its engine reports promptly is granted one replay
     * whatever the clock says, since no window survives a wait nothing bounds, and measuring one against it would
     * only leave the operation with no replay at all (#903).
     */
    private static final long RETRY_WINDOW_NANOS = TimeUnit.SECONDS.toNanos(10);
    // package-private so that the suite can assert the one invariant tying it to ROW_LOCK_TIMEOUT_PROPERTY:
    // a bound at or past this window leaves the replays exactly where #903 found them
    static final long RETRY_WINDOW_NANOS = TimeUnit.SECONDS.toNanos(10);

    /** Upper bound of the random delay before the second attempt, in milliseconds; it doubles with every attempt. */
    private static final double BASE_SLEEP_ON_RETRY_MS = 50.0;
@@ -624,23 +631,41 @@
    private final AtomicBoolean backstopFailedWarned = new AtomicBoolean();
    private final AtomicBoolean queryTimeoutWarned = new AtomicBoolean();
    private final AtomicBoolean standingReadBoundWarned = new AtomicBoolean();
    // The same, for the two ways the lock bound of a DDL degrades: a session that would not take the
    // setting - or would not say what it carried before it - and one it could not be taken off again.
    // The second one is a moment rather than a latch, the way CachedConnection throttles the read bound
    // it could not lift: whatever makes a restore fail - a transaction the server has doomed,
    // middleware that rejects a SET - recurs on every DDL, and each occurrence now costs the pool the
    // connection it happened on, so said once for the life of the storage an operator could not tell
    // one stranded bound from a pool full of them.
    private final AtomicBoolean ddlLockBoundNotSetWarned = new AtomicBoolean();
    private final AtomicLong ddlLockBoundLeftBehindWarned = new AtomicLong();
    private static final long DDL_LOCK_BOUND_WARNING_INTERVAL_MS = 10000;
    // The same, for the two ways a lock bound of this backend degrades: a session that would not take
    // the setting - or would not say what it carried before it - and one it could not be taken off
    // again. The second one is a moment rather than a latch, the way CachedConnection throttles the
    // read bound it could not lift: whatever makes a restore fail - a transaction the server has
    // doomed, middleware that rejects a SET - recurs on every statement it is put around, and each
    // occurrence now costs the pool the connection it happened on, so said once for the life of the
    // storage an operator could not tell one stranded bound from a pool full of them.
    // Per LockBound rather than per storage: the two bounds are set by different code on different
    // paths, and a DDL of the open that could not take its bound says nothing about the writes that
    // follow it - reported through one latch, the first open would silence every write of that
    // backend.
    // Not private, so that a case can read which bound has already said something without reading a log:
    // what these are is one latch per bound, and a case saying so is what keeps them from becoming one.
    final EnumMap<LockBound,AtomicBoolean> lockBoundNotSetWarned = perBound(AtomicBoolean::new);
    final EnumMap<LockBound,AtomicLong> lockBoundLeftBehindWarned = perBound(AtomicLong::new);
    private static final long LOCK_BOUND_WARNING_INTERVAL_MS = 10000;
    // And a third way, which is no failure of anything: an engine this backend knows no lock setting
    // for is left unbounded deliberately, and says so once - see reportTheEngineIsNotKnown(). Not
    // for is left unbounded deliberately, and says so once - see reportTheEngineIsNotKnown(). Per
    // LockBound like the two above and for the same reason: the engine is the same for both bounds,
    // but what an operator is told to act on is the property of the wait that was left unbounded, and
    // a backend opening on such a driver would otherwise say nothing about the writes behind it. Not
    // private, so that a case can read what a storage has already said without reading a log.
    final AtomicBoolean ddlLockBoundEngineUnknownWarned = new AtomicBoolean();
    final EnumMap<LockBound,AtomicBoolean> lockBoundEngineUnknownWarned = perBound(AtomicBoolean::new);
    // What was actually said, set beside the flag above and read the same way: the flag alone tells a
    // case that something was logged, not what it named.
    volatile String ddlLockBoundEngineUnknownSaid;
    final EnumMap<LockBound,AtomicReference<String>> lockBoundEngineUnknownSaid = perBound(AtomicReference::new);

    /** One of these per bound, made where this class is: the bounds are an enum, so this is an EnumMap. */
    private static <T> EnumMap<LockBound,T> perBound(Supplier<T> value) {
        final EnumMap<LockBound,T> perBound=new EnumMap<>(LockBound.class);
        for (final LockBound bound : LockBound.values()) {
            perBound.put(bound, value.get());
        }
        return perBound;
    }

    /**
     * The socket read timeout of one connection, and the statements running on it. This second
@@ -1326,7 +1351,149 @@

    /** That bound in seconds, as configured, read by the reader every bound of this backend shares. */
    static int ddlLockBoundSeconds() {
        return boundSeconds(DDL_LOCK_TIMEOUT_PROPERTY, COMMENT_LOCK_TIMEOUT_SECONDS);
        return LockBound.DDL.seconds();
    }

    /**
     * The bound on the wait of a transaction of {@link #write} for a row lock another session holds, in
     * seconds. A value of {@code 0}, or a negative one, leaves that wait to the engine, which is what
     * this backend did before this bound existed (#915).
     * <p>
     * The replay of {@link #write} is bounded by a wall clock ({@link #RETRY_WINDOW_NANOS}), and a
     * clock cannot bound a wait that nothing else bounds: the lock wait ahead of a conflict is charged
     * to the attempt that hit it, so an attempt alone outlasting the window left the operation with no
     * replay at all, however transient its conflict (#903). Three engines out of four wait for a row
     * lock with no bound of their own ({@code LOCK_TIMEOUT} is -1 on sql server, {@code lock_timeout} is
     * 0 on postgres, and oracle's enqueue is unlimited), and the fourth bounds it at 50 s - five times
     * the window, which is the same thing as far as the window is concerned. Bounded here, an attempt
     * ends inside the window and the window governs the replay again, as it was designed to.
     * <p>
     * The default is deliberately a small fraction of that window: an attempt that ends at this bound
     * spends this much of it, so the replays a conflicted write gets are {@code RETRY_WINDOW_NANOS}
     * divided by this value. At the shipped 3 s a write that keeps meeting a lock gets three attempts
     * inside the window; raised past the window it would leave exactly one, which is the shape #903
     * describes.
     * <p>
     * What it changes for a deployment is that a write blocked behind a long writer of another session
     * now fails once the window is spent, where it used to wait as long as it took and then succeed.
     * That is the trade this bound is: an operation held for the length of somebody else's transaction
     * cannot be told from one that is never coming back, and the caller of a write is an LDAP client
     * with a timeout of its own. A deployment that would rather wait sets this to 0 and keeps the old
     * behaviour: the wait is the engine's, and a conflict behind it gets the one replay past the window
     * that {@link #grantedPastTheWindow} grants a wait nothing bounds (#904), and no more.
     * <p>
     * Oracle is not one of the engines this is put on: it has no session-level bound for the enqueue a
     * row lock waits in under plain DML - {@code ddl_lock_timeout} is the DDL lock and
     * {@code distributed_lock_timeout} the distributed transaction, neither of which is this wait, and
     * {@code select ... for update wait} is not a statement this backend issues. An attempt there stays
     * bounded only by {@link StatementBound#OPERATION}, whose cancel a session blocked in that enqueue
     * does not act on - see {@link #bounded(PreparedStatement, StatementBound, Execution)} - so on oracle
     * what answers #903 is not this property but the grant of #904, whatever this property says.
     * <p>
     * What it costs is the round trips around each write transaction - two on mysql and sql server (the
     * setting and the value given back; the readback is a third, and is paid once per pooled connection
     * rather than once per write, see {@link #displacedValue}), three on postgres (a {@code set local}
     * and the savepoint taken in front of it, which is let go of again once the setting is on), none on
     * oracle. Unlike the DDL bound this is the hot path, which is what the readback is cached for.
     */
    static final String ROW_LOCK_TIMEOUT_PROPERTY="org.openidentityplatform.opendj.jdbc.row.lock.timeout";

    /**
     * The default of {@link #ROW_LOCK_TIMEOUT_PROPERTY}: a third of {@link #RETRY_WINDOW_NANOS}, so
     * that a conflicted write is replayed rather than spending its whole window on one attempt.
     */
    static final int ROW_LOCK_TIMEOUT_SECONDS=3;

    /** That bound in seconds, as configured. */
    static int rowLockBoundSeconds() {
        return LockBound.ROW.seconds();
    }

    /**
     * A wait this backend bounds with a session setting, and the property that configures it. Two of
     * them: the lock a DDL of {@code openTree()} queues for, and the row lock a transaction of
     * {@link #write} queues for. One mechanism sets both ({@link #armLockBound}), and what differs
     * between them - which statement each engine takes, what it is called in a message, and where the
     * value comes from - is answered here rather than at the call sites.
     * <p>
     * The statements are asked of the {@link Dialect}, per constant, so that an engine added later
     * cannot compile without saying what it sets for both.
     */
    enum LockBound {
        /** The metadata, schema or DDL lock a {@code create table}, {@code create index} or {@code drop} waits for. */
        DDL(DDL_LOCK_TIMEOUT_PROPERTY, COMMENT_LOCK_TIMEOUT_SECONDS, "a DDL",
            "a DDL - the create index of an open whose table is already there, the drop table of a clear -") {
            @Override
            String boundSql(Dialect dialect, int seconds, Long previous) {
                return dialect.ddlLockBoundSql(seconds, previous);
            }

            @Override
            String query(Dialect dialect) {
                return dialect.ddlLockBoundQuery();
            }

            @Override
            String restoreSql(Dialect dialect, long previous) {
                return dialect.ddlLockRestoreSql(previous);
            }
        },
        /** The row lock a statement of a write transaction waits for. */
        ROW(ROW_LOCK_TIMEOUT_PROPERTY, ROW_LOCK_TIMEOUT_SECONDS, "a write transaction",
            "a write transaction - the upsert of an add, the delete of a remove -") {
            @Override
            String boundSql(Dialect dialect, int seconds, Long previous) {
                return dialect.rowLockBoundSql(seconds, previous);
            }

            @Override
            String query(Dialect dialect) {
                return dialect.rowLockBoundQuery();
            }

            @Override
            String restoreSql(Dialect dialect, long previous) {
                return dialect.rowLockRestoreSql(previous);
            }
        };

        final String property;
        final int defaultSeconds;
        /** What the message reporting this bound calls the thing that was waiting. */
        final String waiter;
        /**
         * The same, spelled out with what this backend issues under it: the one line an operator has to
         * act on says which statements are the ones left waiting, since the property alone does not say
         * what a deployment would see go slow.
         */
        final String waiterInDetail;

        LockBound(String property, int defaultSeconds, String waiter, String waiterInDetail) {
            this.property=property;
            this.defaultSeconds=defaultSeconds;
            this.waiter=waiter;
            this.waiterInDetail=waiterInDetail;
        }

        /** This bound in seconds, as configured, read by the reader every bound of this backend shares. */
        int seconds() {
            return boundSeconds(property, defaultSeconds);
        }

        /**
         * The setting bounding this wait on this engine, or null where there is none to put on: an engine
         * left to a bound of its own, and a session that already gives up sooner than this one would.
         *
         * @param previous what the session carries now, as {@link #query(Dialect)} read it, in the unit
         *                 that query answers in - or null where nothing was read back
         */
        abstract String boundSql(Dialect dialect, int seconds, Long previous);

        /** What the session carries now, or null where this bound undoes itself. */
        abstract String query(Dialect dialect);

        /** The setting giving the session back what {@link #query(Dialect)} read off it. */
        abstract String restoreSql(Dialect dialect, long previous);
    }

    // The comment statement runs on a connection of its own (newStampConnection() below), and a
@@ -1393,10 +1560,47 @@
                return null;
            }

            // the same setting and the same set local, at this bound's own value: lock_timeout bounds the
            // wait for any lock of the transaction, the row locks of an upsert included, and the commit or
            // the rollback ending that transaction is what takes it off again. A write() attempt is exactly
            // such a transaction, so this engine pays no readback and leaves nothing on a pooled connection.
            // What ends it early is a DDL of the same attempt, which commits (commitStatement): the rest of
            // that write runs unbounded here, and is given the same treatment on the two engines whose
            // setting is the session's - commitStatement takes it off there rather than leaving it on. It is
            // an attempt that has committed part of its work, so it is out of the replay whatever it waits
            // for, and a bound on a wait nothing can replay would only fail a write that used to go through,
            // which is why the bound is not armed again behind it.
            @Override
            String rowLockBoundSql(int seconds, Long previous) {
                return "set local lock_timeout = "+seconds*1000L;
            }

            @Override
            String rowLockBoundQuery() {
                return null; // set local: the transaction of the attempt discards it
            }

            @Override
            String rowLockRestoreSql(long previous) {
                return null;
            }

            @Override
            String rowLockLiftSql() {
                // the set local has no value to give back, so the wait of the rest of this transaction is
                // put back where a transaction that set nothing would have left it
                return "set local lock_timeout to default";
            }

            @Override
            boolean boundLivesInTheTransaction() {
                return true;
            }

            @Override
            boolean oneSettingForBothBounds() {
                return true; // lock_timeout, for every lock wait of the transaction
            }
        },
        /** mysql: lock_wait_timeout takes seconds; connectTimeout bounds the socket connect and socketTimeout every read after it, both in milliseconds. */
        MYSQL("set session lock_wait_timeout="+COMMENT_LOCK_TIMEOUT_SECONDS,
@@ -1421,6 +1625,28 @@
            String ddlLockRestoreSql(long previous) {
                return "set session lock_wait_timeout="+previous;
            }

            // innodb_lock_wait_timeout, in seconds, and never the lock_wait_timeout above it: the first is
            // the row lock a write waits for, the second the metadata lock a DDL waits for, and they are
            // two settings with two defaults. This is the one engine of the four bounding this wait by
            // itself - at 50 s, five times the replay window, so a conflict reported at it arrives with the
            // window already spent and is never replayed. A session giving up sooner keeps what it has: the
            // variable has no encoding for "wait forever" to be mistaken for a tight value, its range
            // starting at 1.
            @Override
            String rowLockBoundSql(int seconds, Long previous) {
                return (previous!=null && previous<=seconds) ? null : "set session innodb_lock_wait_timeout="+seconds;
            }

            @Override
            String rowLockBoundQuery() {
                return "select @@session.innodb_lock_wait_timeout";
            }

            @Override
            String rowLockRestoreSql(long previous) {
                return "set session innodb_lock_wait_timeout="+previous;
            }
        },
        /** oracle: ddl_lock_timeout takes seconds and defaults to 0 (give up at once), but it can be raised globally; the connect and read bounds take milliseconds. */
        ORACLE("alter session set ddl_lock_timeout="+COMMENT_LOCK_TIMEOUT_SECONDS,
@@ -1443,6 +1669,29 @@
            String ddlLockRestoreSql(long previous) {
                return null;
            }

            // There is no session setting bounding the enqueue a row lock waits in under plain DML here:
            // ddl_lock_timeout is the DDL lock above, distributed_lock_timeout is the distributed
            // transaction, and the wait can only be named on the statement itself - "select ... for update
            // wait n", which is not a statement this backend issues. So an attempt of write() on this engine
            // is bounded by StatementBound.OPERATION alone, and a session blocked in that enqueue does not
            // act on the cancel behind it: what ends such a wait is the socket read timeout, which takes the
            // connection with it. Nothing of ours is set here rather than something that would not bound the
            // wait, so a lock timeout of this engine is nothing this backend claims to have bounded.
            @Override
            String rowLockBoundSql(int seconds, Long previous) {
                return null;
            }

            @Override
            String rowLockBoundQuery() {
                return null; // nothing of ours is set on it
            }

            @Override
            String rowLockRestoreSql(long previous) {
                return null;
            }
        },
        /** ms sql server: lock_timeout takes milliseconds; loginTimeout bounds the socket connect, in seconds, and socketTimeout the prelogin read it leaves open - and every read after it - in milliseconds. */
        MICROSOFT("set lock_timeout "+(COMMENT_LOCK_TIMEOUT_SECONDS*1000),
@@ -1466,6 +1715,33 @@
            String ddlLockRestoreSql(long previous) {
                return "set lock_timeout "+previous;
            }

            // the same session setting as the DDL bound above - this engine has one LOCK_TIMEOUT for every
            // lock wait of a session - at this bound's own value, and put back the moment the transaction is
            // through for the very reason that it is one setting: a value left behind cuts the lock waits of
            // whoever borrows the connection next, and a read borrowing it has no replay to absorb the
            // error 1222 it would then see. -1 is "wait forever" rather than a bound tighter than ours and
            // is replaced; 0 - "do not wait at all" - is tighter, and a session carrying it is left alone.
            @Override
            String rowLockBoundSql(int seconds, Long previous) {
                final long millis=seconds*1000L;
                return (previous!=null && previous>=0 && previous<=millis) ? null : "set lock_timeout "+millis;
            }

            @Override
            String rowLockBoundQuery() {
                return "select @@lock_timeout";
            }

            @Override
            String rowLockRestoreSql(long previous) {
                return "set lock_timeout "+previous;
            }

            @Override
            boolean oneSettingForBothBounds() {
                return true; // LOCK_TIMEOUT, for every lock wait of the session
            }
        };

        final String lockTimeoutSql;
@@ -1520,6 +1796,44 @@
        abstract String ddlLockRestoreSql(long previous);

        /**
         * The session setting bounding the wait of a write transaction for a row lock another session
         * holds, or null where there is none to put on: an engine with no such setting - oracle - and a
         * session that already gives up sooner than this one would.
         * <p>
         * A different setting from {@link #ddlLockBoundSql} on the two engines that have two of them, and
         * literally the same one on the two that have one: what tells the two bounds apart is the wait
         * each is put around, not the statement each issues.
         *
         * @param previous what the session carries now, as {@link #rowLockBoundQuery()} read it, in the
         *                 unit that query answers in - or null where nothing was read back
         */
        abstract String rowLockBoundSql(int seconds, Long previous);

        /**
         * What the session carries now, asked before the bound above displaces it - or null where that
         * bound undoes itself with the transaction it belongs to.
         */
        abstract String rowLockBoundQuery();

        /** The setting that gives the session back the value {@link #rowLockBoundQuery()} read off it. */
        abstract String rowLockRestoreSql(long previous);

        /**
         * What takes the row lock bound off where there is no displaced value to give back, so that the
         * wait it was cutting runs as the deployment left it: a {@code set local} reads nothing back -
         * the transaction it belongs to is what discards it - so what puts the setting where it would
         * have been is its own default. Null on the engines whose bound is a session setting and whose
         * release has a value in hand ({@link #rowLockRestoreSql}), and on those that arm nothing.
         * <p>
         * Read by the DDL of a write whose own bound is turned off, which is the one place a row bound of
         * this backend has to come off before the transaction it belongs to ends - see
         * {@code liftTheRowLockBoundForAnUnboundedDdl()}.
         */
        String rowLockLiftSql() {
            return null;
        }

        /**
         * Whether the bound belongs to the transaction running the DDL rather than to the session. Such a
         * setting needs a transaction block to take effect at all, a rollback undoes it, and the commit
         * ending the DDL is what takes it off again - so nothing is read back and nothing is put back.
@@ -1527,6 +1841,22 @@
        boolean boundLivesInTheTransaction() {
            return false;
        }

        /**
         * Whether one setting of this engine bounds both waits, so that arming either displaces the other.
         * True of sql server, whose {@code LOCK_TIMEOUT} bounds every lock wait of a session, and of
         * postgres, whose {@code lock_timeout} bounds every lock wait of a transaction; false of mysql,
         * which has {@code lock_wait_timeout} for the metadata lock and {@code innodb_lock_wait_timeout}
         * for the row lock, and of oracle, which takes neither bound.
         * <p>
         * What reads it is {@link #armLockBound}, deciding whether a session already gives up sooner than
         * the bound being armed: on such an engine the value read back can be this backend's own - the
         * row bound of the write a DDL runs inside - and a bound of ours is no reason to leave the other
         * one unarmed.
         */
        boolean oneSettingForBothBounds() {
            return false;
        }
    }

    /** Returns the class name of the driver behind the given connection, which names the engine it talks to. */
@@ -2157,120 +2487,310 @@

    /**
     * Runs a DDL of this backend under {@link #DDL_LOCK_TIMEOUT_PROPERTY}, and gives the session back
     * whatever it carried before.
     * whatever it carried before. What is put on the session, and what that costs where it cannot be,
     * is {@link #armLockBound}: this is the one statement version of it, where {@link #write} arms the
     * same mechanism around a whole transaction.
     * <p>
     * The dialect is passed in rather than read off the connection here, the way
     * {@code commentTable()} takes it: the callers know it, and taking it as a parameter is what makes
     * this reachable from a test with no database behind it.
     * <p>
     * Nothing of ours is set where it could not be taken off again, and a failure of the readback is
     * never the failure of the DDL: this runs mostly on a pooled connection, so a setting left behind
     * reaches every statement of whoever borrows it next - on sql server that is every lock wait of
     * theirs, row locks included, and {@link #isConflict} classifies error 1222 as no replayable
     * conflict. A session this backend could not take its bound off again is kept out of the pool for
     * that reason ({@link CachedConnection#keepOutOfThePool}), since the validation of the next borrow
     * is {@code isValid()} - a liveness check a connection carrying a stale setting passes. The one
     * connection here that is not the pool's is the catalog's own, which {@code createCatalogTable()}
     * creates its table on: there a setting left behind reaches the rest of that write and goes with
     * the connection, which is closed with it.
     * <p>
     * The DDL runs whatever any of that did, and it runs under the same rewrite either way: a setting
     * can reach the server and fail only as the statement carrying it is closed, which no driver tells
     * apart from a setting that never arrived, and reporting a DDL that then really did give up at this
     * bound as the bare 55P03 it arrives as is the gap this exists to close.
     * The DDL runs whatever the arming did, and where a setting of ours was issued it runs under a
     * rewrite either way: a setting can reach the server and fail only as the statement carrying it is
     * closed, which no driver tells apart from a setting that never arrived, and reporting a DDL that
     * then really did give up at this bound as the bare 55P03 it arrives as is the gap that rewrite
     * exists to close.
     */
    <T> T withDdlLockBound(Connection con, Dialect dialect, Execution<T> action) throws SQLException {
        final int seconds=ddlLockBoundSeconds();
        if (seconds<=0) { // the wait is left exactly as unbounded as it was, and nobody asked otherwise
            return action.run();
        }
        if (dialect==null) {
            // The one branch where a bound was asked for and none is put on, which is why it is the one
            // that says so: an engine none of these settings fit is fed none of them - untested SQL is no
            // thing to send a database on the path a backend opens by - and the silence around that is
            // what an operator has no way of finding out.
            reportTheEngineIsNotKnown(con);
            return action.run();
        }
        // Asked with nothing displaced yet, which is what tells an engine this bound is never put on -
        // oracle - from an engine it is put on. What the session actually carries is read below, and can
        // take the bound off again all by itself.
        if (dialect.ddlLockBoundSql(seconds, null)==null) {
            return action.run();
        }
        final String query=dialect.ddlLockBoundQuery();
        final Long previous=(query==null) ? null : sessionValue(con, dialect, query);
        if (query!=null && previous==null) { // read it back first: see above
            return action.run();
        }
        // Asked again with it: a session already giving up sooner than this bound is left exactly as it
        // is, rather than loosened to ours for the length of the DDL. That is the argument leaving oracle
        // alone, applied where the displaced value is in hand and costs nothing to respect.
        final String bound=dialect.ddlLockBoundSql(seconds, previous);
        if (bound==null) {
            return action.run();
        }
        if (dialect.boundLivesInTheTransaction() && !inATransactionBlock(con, dialect, bound)) {
            return action.run();
        }
        final String restore=(previous==null) ? null : dialect.ddlLockRestoreSql(previous);
        // A statement that fails inside a postgres transaction aborts it, and everything after it - the
        // DDL included - then fails with 25P02 rather than running "unbounded, as before": a backend that
        // opened before this bound existed would stop opening because of the bound meant to protect it.
        // The rollback below goes back to here, which also undoes a set local that did reach the server,
        // so the DDL really does run as unbounded as the warning says it does.
        final Savepoint beforeTheBound=savepointBeforeTheBound(con, dialect, bound);
        try {
            boundedSessionCall(con, () -> {
                executeSessionStatement(con, bound);
                return null;
            });
        }catch (SQLException | RuntimeException e) {
            // The bound is an improvement on a wait, and never a reason to fail a DDL that would have gone
            // through: a backend that opened before this bound existed has to open still. Whatever the
            // setting displaced is given back by the finally below, whether it went on or not - giving back
            // a value the session may never have left costs a round trip and changes nothing.
            reportTheWaitIsLeftUnbounded(dialect, bound, e);
            rollbackTheBound(con, dialect, beforeTheBound);
        }
        // From here rather than from the top of this method: what the statements above spent is not time
        // the DDL waited for its lock, and it is the wait that this bound either ended or did not.
        final long startedAt=nanoTime();
        final ArmedLockBound armed=armLockBound(con, dialect, LockBound.DDL);
        try {
            return action.run();
        }catch (SQLException e) {
            throw gaveUpOnTheLock(e, dialect, seconds, startedAt);
            // only where this bound's own figure is what the session is carrying: a wait this backend put
            // no bound on was ended by something else, and naming this property for it sends an operator
            // to a value that changes nothing - the session that already gave up sooner, the engine that
            // has no such setting, and the deployment whose own value this arming put back on
            throw armed.ourFigure ? gaveUpOnTheLock(e, dialect, armed.seconds, armed.startedAt) : e;
        }catch (RuntimeException e) {
            // Not every statement running under this bound answers with the SQLException it was given:
            // the lookup deciding each drop of a clear wraps whatever it sees in a
            // StorageRuntimeException (isExistsTable), and it runs inside the same bound as the drop it
            // decides. Without this, a lock this bound ended reaches an operator as the bare vendor error
            // one line away from the drop that would have named the property.
            throw gaveUpOnTheLock(e, dialect, seconds, startedAt);
            throw armed.ourFigure ? gaveUpOnTheLock(e, dialect, armed.seconds, armed.startedAt) : e;
        }finally {
            if (restore!=null) {
                restoreDdlLockBound(con, dialect, bound, restore);
            releaseLockBound(con, dialect, armed);
            }
        }

    /**
     * A lock bound of this backend as it stands on one session: what was put on it, what gives it back,
     * and whether the wait it was put around is bounded at all. Produced by {@link #armLockBound} and
     * handed to {@link #releaseLockBound}, which is what lets {@link #write} arm one around a whole
     * transaction where {@link #withDdlLockBound} wraps a single statement.
     */
    static final class ArmedLockBound {
        final LockBound kind;
        /** The configured value this was armed at, in seconds - 0 where nothing of ours was put on. */
        final int seconds;
        /** The setting issued, or null where none was: an engine or a session that needed none. */
        final String bound;
        /** What gives the session its value back, or null where the bound takes itself off. */
        final String restore;
        /**
         * Whether the wait this was armed around is bounded at no more than {@link LockBound#seconds()}
         * - by the setting above, or by a session value that was already tighter and was left alone.
         * False wherever this backend could not put a bound on and does not know of one: an engine with
         * no such setting, the property turned off, a readback that failed, and a setting the session
         * refused. It is what {@link #replayReason} reads: a wait nothing bounds must not be replayed on
         * a clock, which is the whole of #903.
         */
        final boolean bounded;
        /**
         * Whether what the session carries under this bound is this bound's own figure, which is what
         * lets a failure of the work be reported as this bound's doing
         * ({@link #gaveUpOnTheLock(SQLException, Dialect, int, long)}). False
         * where a statement of ours was issued all the same: the value put on is the one the deployment
         * set, restored over a bound of ours that was tighter than it, and naming
         * {@link #DDL_LOCK_TIMEOUT_PROPERTY} for a wait that value ended would send an operator to raise
         * a property that governs nothing here.
         */
        final boolean ourFigure;
        /**
         * When the wait itself began, as {@link #nanoTime} reads it: after the round trips of the arming,
         * which are not time anything waited for a lock. Zero where {@link #bound} is null, since nothing
         * is ever measured against a setting that was not put on.
         */
        final long startedAt;

        private ArmedLockBound(LockBound kind, int seconds, String bound, String restore, boolean bounded,
                boolean ourFigure, long startedAt) {
            this.kind=kind;
            this.seconds=seconds;
            this.bound=bound;
            this.restore=restore;
            this.bounded=bounded;
            this.ourFigure=ourFigure;
            this.startedAt=startedAt;
        }

        /** Nothing was put on and nothing is known to bound this wait: it runs as it did before #885. */
        static ArmedLockBound none(LockBound kind) {
            return new ArmedLockBound(kind, 0, null, null, false, false, 0);
        }

        /**
         * The session already gives up sooner than this bound would, so it keeps what it has. Nothing is
         * set and nothing has to be put back - and the wait is bounded, which is the whole point of
         * leaving that value alone.
         */
        static ArmedLockBound alreadyTighter(LockBound kind, int seconds) {
            return new ArmedLockBound(kind, seconds, null, null, true, false, 0);
        }
    }

    /**
     * Puts a lock bound of this backend on a session, and never fails the work it was put around.
     * <p>
     * The dialect is passed in rather than read off the connection here, the way {@code commentTable()}
     * takes it: the callers know it, and taking it as a parameter is what makes this reachable from a
     * test with no database behind it.
     * <p>
     * Nothing of ours is set where it could not be taken off again, and a failure of the readback is
     * never the failure of the work: this runs mostly on a pooled connection, so a setting left behind
     * reaches every statement of whoever borrows it next - on sql server that is every lock wait of
     * theirs, and a read has no replay to absorb the error 1222 it would then see. A session this
     * backend could not take its bound off again is kept out of the pool for that reason
     * ({@link CachedConnection#keepOutOfThePool}), since the validation of the next borrow is
     * {@code isValid()} - a liveness check a connection carrying a stale setting passes. The one
     * connection here that is not the pool's is the catalog's own, which {@code createCatalogTable()}
     * creates its table on: there a setting left behind reaches the rest of that write and goes with
     * the connection, which is closed with it.
     */
    ArmedLockBound armLockBound(Connection con, Dialect dialect, LockBound kind) {
        final int seconds=kind.seconds();
        if (seconds<=0) { // the wait is left exactly as unbounded as it was, and nobody asked otherwise
            return ArmedLockBound.none(kind);
        }
        if (dialect==null) {
            // The one branch where a bound was asked for and none is put on, which is why it is the one
            // that says so: an engine none of these settings fit is fed none of them - untested SQL is no
            // thing to send a database on the path a backend opens by - and the silence around that is
            // what an operator has no way of finding out.
            reportTheEngineIsNotKnown(con, kind);
            return ArmedLockBound.none(kind);
        }
        // Asked first with nothing displaced yet, which is what tells an engine this bound is never put
        // on - oracle - from an engine it is put on. What the session actually carries is read below, and
        // can take the bound off again all by itself.
        if (kind.boundSql(dialect, seconds, null)==null) {
            return ArmedLockBound.none(kind);
        }
        final String query=kind.query(dialect);
        final Long previous=(query==null) ? null : displacedValue(con, dialect, kind, query);
        if (query!=null && previous==null) { // read it back first: see above
            return ArmedLockBound.none(kind);
        }
        // Asked again with it: a session already giving up sooner than this bound is left exactly as it
        // is, rather than loosened to ours for the length of the work. That is the argument leaving oracle
        // alone, applied where the displaced value is in hand and costs nothing to respect - and asked of
        // what the deployment set rather than of what the session happens to carry, since on an engine
        // with one setting for both waits the value read back a moment ago can be a bound of ours.
        final Long deployment=carriedByTheDeployment(con, dialect, kind, previous);
        final String atOurFigure=kind.boundSql(dialect, seconds, deployment);
        // Where the deployment gives up sooner than this bound would, "it keeps what it has" is the whole
        // answer only while what it has is what that decision was made against. It is not, on the one
        // road where the value read back is a bound of ours: then the work would run at neither figure -
        // see below, which puts the deployment's own value back on for the length of it.
        final String bound=(atOurFigure!=null) ? atOurFigure : theDeploymentsOwnValue(kind, dialect, deployment,
            previous);
        if (bound==null) {
            return ArmedLockBound.alreadyTighter(kind, seconds);
        }
        // whether what goes on the session is this bound's own figure, which is what lets a failure of the
        // work be named as this bound's doing
        final boolean ourFigure=(atOurFigure!=null);
        if (dialect.boundLivesInTheTransaction() && !inATransactionBlock(con, dialect, kind, bound)) {
            return ArmedLockBound.none(kind);
        }
        final String restore=(previous==null) ? null : kind.restoreSql(dialect, previous);
        // A statement that fails inside a postgres transaction aborts it, and everything after it - the
        // work this bound was put around included - then fails with 25P02 rather than running "unbounded,
        // as before": a backend that opened before this bound existed would stop opening because of the
        // bound meant to protect it. The rollback below goes back to here, which also undoes a set local
        // that did reach the server, so the work really does run as unbounded as the warning says it does.
        final Savepoint beforeTheBound=savepointBeforeTheBound(con, dialect, kind, bound);
        try {
            boundedSessionCall(con, () -> {
                executeSessionStatement(con, bound);
                return null;
            });
        }catch (SQLException | RuntimeException e) {
            // The bound is an improvement on a wait, and never a reason to fail work that would have gone
            // through: a backend that opened before this bound existed has to open still, and a write that
            // used to wait for its row lock has to be able to wait for it. Whatever the setting displaced is
            // given back by the release, whether it went on or not - giving back a value the session may
            // never have left costs a round trip and changes nothing.
            reportTheWaitIsLeftUnbounded(dialect, kind, bound, e);
            rollbackTheBound(con, dialect, beforeTheBound);
            // The setting is still named as issued, so that the release puts the value back and a failure
            // that did give up at this bound is still renamed - a setting can reach the server and fail only
            // as the statement carrying it is closed, and no driver tells that apart from one that never
            // arrived. What is not claimed is that the wait is bounded: a replay decided on that claim would
            // be a replay of a wait nothing ends.
            return new ArmedLockBound(kind, seconds, bound, restore, false, ourFigure, nanoTime());
        }
        // The setting is on and there is nothing left to take back, so the subtransaction that net opened
        // is let go of rather than left to span the work - see there.
        releaseTheSavepoint(con, dialect, beforeTheBound);
        // From here rather than from the top of this method: what the round trips above spent is not time
        // anything waited for a lock, and it is that wait this bound either ends or does not.
        return new ArmedLockBound(kind, seconds, bound, restore, true, ourFigure, nanoTime());
    }

    /**
     * What the deployment set, to be put on over a bound of this backend that is cutting the wait
     * tighter than the deployment asked for - or null where there is nothing to put on.
     * <p>
     * Asked where a session already gives up sooner than the bound being armed. That is normally the
     * end of it: the session keeps its own value, which is the argument that leaves oracle alone. The
     * one road where it is not is the DDL of a write on an engine with one setting for both waits: what
     * the session carries there is the row bound of that write, armed one statement earlier and tighter
     * than what the deployment set, so leaving it alone runs the DDL at a figure nobody asked for and
     * hands its failure through bare. The value the row bound displaced goes back on for the length of
     * the DDL, and the release puts the row bound back behind it - nothing is loosened past what the
     * deployment allows, which is what the decision above already established.
     */
    private String theDeploymentsOwnValue(LockBound kind, Dialect dialect, Long deployment, Long previous) {
        if (deployment==null || deployment.equals(previous)) {
            return null; // the session carries what the decision was made against, and keeps it
        }
        return kind.restoreSql(dialect, deployment);
    }

    /** Gives the session back what an armed bound displaced, where it displaced anything. */
    void releaseLockBound(Connection con, Dialect dialect, ArmedLockBound armed) {
        if (armed.restore!=null) {
            restoreLockBound(con, dialect, armed);
        }
    }

    /**
     * What the session carries before a bound of ours displaces it, read once per pooled connection for
     * the bound that is on the hot path.
     * <p>
     * The readback is a round trip, and the row lock bound takes one around every write of the server
     * where the DDL bound takes one around an open. It can be remembered because only this backend
     * writes that setting on a connection of this pool: every write puts the value back before the
     * connection is released, and a connection whose restore failed is kept out of the pool rather than
     * handed on - so a remembered value cannot outlive the session that answered it. The bound of a DDL
     * is read every time all the same: it runs inside a write which may be carrying the row bound of
     * this backend on the very same setting (sql server has one {@code LOCK_TIMEOUT} for both), and what
     * it has to put back is that value rather than the one the connection was borrowed with.
     */
    private Long displacedValue(Connection con, Dialect dialect, LockBound kind, String query) {
        if (kind!=LockBound.ROW || !(con instanceof CachedConnection)) {
            return sessionValue(con, dialect, kind, query);
        }
        final CachedConnection pooled=(CachedConnection) con;
        final Long remembered=pooled.rowLockBoundDisplaces();
        if (remembered!=null) {
            return remembered;
        }
        final Long read=sessionValue(con, dialect, kind, query);
        if (read!=null) {
            pooled.rowLockBoundDisplaces(read);
        }
        return read;
    }

    /**
     * What the session carried before this backend put anything on it, which is what decides whether it
     * already gives up sooner than the bound being armed - and which is not what the readback answers
     * where a bound of ours is already on that very setting.
     * <p>
     * sql server has one {@code LOCK_TIMEOUT} for both waits, and every DDL of this backend but the
     * off-write catalog drop is issued from inside a write, which arms the row bound one statement
     * earlier. Read live, that DDL sees this backend's own 3 s, answers "already tighter" to its own
     * 5 s and runs at the row bound instead: {@link #DDL_LOCK_TIMEOUT_PROPERTY} would govern no DDL of
     * a write at all - a deployment raising it for an index build blocked by long readers would still
     * get the row bound - and the 1222 such a DDL gives up with would be reported as the bare vendor
     * error, the rename of #885 lost with it. What the row bound displaced is remembered on the
     * connection for the readback above, so the value the deployment set is in hand here.
     * <p>
     * Only where one setting carries both waits ({@link Dialect#oneSettingForBothBounds}). On mysql the
     * two are different variables - {@code lock_wait_timeout} for the metadata lock,
     * {@code innodb_lock_wait_timeout} for the row lock - so the remembered value describes neither the
     * other's session nor its default, and reading it there would leave a metadata lock waiting a year
     * because a deployment had tightened the row one.
     * <p>
     * The live readback stays what the release puts back: what a session has to be given back is what it
     * carried a statement ago, bound of ours or not. Where the two differ the session is carrying a
     * bound of ours rather than the value this decides against, so "already tighter" would leave the
     * work at neither figure - see {@link #armLockBound}, which puts the deployment's own value on for
     * the length of it instead.
     */
    private Long carriedByTheDeployment(Connection con, Dialect dialect, LockBound kind, Long previous) {
        if (kind==LockBound.ROW || !dialect.oneSettingForBothBounds() || !(con instanceof CachedConnection)) {
            return previous;
        }
        final Long displacedByTheRowBound=((CachedConnection) con).rowLockBoundDisplaces();
        return (displacedByTheRowBound!=null) ? displacedByTheRowBound : previous;
    }

    /**
     * Whether a setting that lives in the transaction would take effect on this connection at all.
     * Outside a transaction block postgres answers {@code SET LOCAL} with a warning and does nothing
     * with it - the driver raises nothing, so the DDL would run with no bound and the log would read
     * with it - the driver raises nothing, so the work would run with no bound and the log would read
     * exactly like a bounded one. What makes the block is the {@code setAutoCommit(false)} a pooled
     * connection is established with and never re-asserts on borrow, so it is asked rather than
     * assumed; the call is answered out of the driver's own state, not by a round trip.
     */
    private boolean inATransactionBlock(Connection con, Dialect dialect, String bound) {
    private boolean inATransactionBlock(Connection con, Dialect dialect, LockBound kind, String bound) {
        try {
            if (!con.getAutoCommit()) {
                return true;
            }
            reportTheWaitIsLeftUnbounded(dialect, bound, new SQLException("the connection is in auto-commit,"
            reportTheWaitIsLeftUnbounded(dialect, kind, bound, new SQLException("the connection is in auto-commit,"
                + " where this setting belongs to no transaction and the server answers it with a warning"));
        }catch (SQLException | RuntimeException e) {
            reportTheWaitIsLeftUnbounded(dialect, bound, e);
            reportTheWaitIsLeftUnbounded(dialect, kind, bound, e);
        }
        return false;
    }
@@ -2281,19 +2801,49 @@
     * give a savepoint. Nothing of the bound has been set when this is asked, so a failure here only
     * leaves the setting below without this net - which is where it was before the net existed.
     */
    private Savepoint savepointBeforeTheBound(Connection con, Dialect dialect, String bound) {
    private Savepoint savepointBeforeTheBound(Connection con, Dialect dialect, LockBound kind, String bound) {
        if (!dialect.boundLivesInTheTransaction()) {
            return null;
        }
        try {
            return boundedSessionCall(con, con::setSavepoint);
        }catch (SQLException | RuntimeException e) {
            reportTheWaitIsLeftUnbounded(dialect, bound, e);
            reportTheWaitIsLeftUnbounded(dialect, kind, bound, e);
            return null;
        }
    }

    /**
     * Lets go of that point once the setting is on and there is nothing left to take back.
     * <p>
     * A savepoint is a subtransaction of the one the work runs in, and one left open spans that work:
     * on the hot path that is every write of the server, whose rows then carry the subtransaction's own
     * xid rather than the transaction's - which every reader of those rows resolves through
     * {@code pg_subtrans} until they are hinted. It costs a round trip to let go of, which is the trade
     * this makes: one statement per write against a subtransaction per write. A {@code SET LOCAL}
     * survives the release, so the bound this was taken in front of stays on for the rest of the
     * transaction - which is the whole of what it was armed around.
     * <p>
     * Best effort, like every other round trip of the bound: a savepoint that could not be let go of
     * costs a subtransaction, never the work.
     */
    private void releaseTheSavepoint(Connection con, Dialect dialect, Savepoint beforeTheBound) {
        if (beforeTheBound==null) {
            return;
        }
        try {
            boundedSessionCall(con, () -> {
                con.releaseSavepoint(beforeTheBound);
                return null;
            });
        }catch (SQLException | RuntimeException e) {
            logger.trace(LocalizableMessage.raw("jdbc: the point in front of a lock bound could not be let go of on"
                + " this %s database, leaving the work it was taken for inside a subtransaction: %s", dialect,
                stackTraceToSingleLineString(e)));
        }
    }

    /**
     * Takes the transaction back to the point before the bound was set: it clears the abort a setting
     * that failed inside it left behind, and it undoes a {@code set local} that did reach the server
     * and failed only as its statement was closed. Best effort - a transaction that cannot be taken
@@ -2346,11 +2896,11 @@
    /**
     * What a session setting of this engine carries right now, or null where it could not be read as a
     * number: a server whose session does not have the variable, or one answering with something no
     * {@code SET} of it would take back. The DDL then runs as unbounded as it was before this bound
     * existed, which is why this is reported rather than thrown - and reported once, since every DDL
     * of that backend would say the same thing.
     * {@code SET} of it would take back. The work this bound was to be put around then runs as unbounded
     * as it ran before the bound existed, which is why this is reported rather than thrown - and reported
     * once per bound, since every DDL, or every write, of that backend would say the same thing.
     */
    private Long sessionValue(Connection con, Dialect dialect, String query) {
    private Long sessionValue(Connection con, Dialect dialect, LockBound kind, String query) {
        if (logger.isTraceEnabled()) {
            logger.trace(LocalizableMessage.raw("jdbc: %s",query));
        }
@@ -2364,70 +2914,79 @@
                }
            });
        }catch (SQLException | RuntimeException e) { // a value that is not a number arrives unchecked
            reportTheWaitIsLeftUnbounded(dialect, query, e);
            reportTheWaitIsLeftUnbounded(dialect, kind, query, e);
            return null;
        }
    }

    /**
     * Said once per storage, whichever round trip of the bound around a DDL the connection would not
     * take: every DDL of that backend would say the same thing, and a backend opening its trees issues
     * about 25 of them. The DDL itself is unaffected - it waits as it did before this bound existed.
     * Said once per storage and per bound, whichever round trip of the arming the connection would not
     * take: every DDL of that backend would say the same thing - a backend opening its trees issues about
     * 25 of them - and so would every write. The work itself is unaffected: it waits as it did before this
     * bound existed.
     * <p>
     * "May bound nothing" rather than "bounds nothing": a setting that reached the server and failed
     * only as the statement carrying it was closed leaves the DDL bounded after all, and no driver says
     * only as the statement carrying it was closed leaves the wait bounded after all, and no driver says
     * which of the two happened. Where the bound lives in the transaction the rollback that follows
     * this settles it - there the DDL really does run unbounded.
     * this settles it - there the work really does run unbounded.
     */
    private void reportTheWaitIsLeftUnbounded(Dialect dialect, String sql, Exception e) {
        if (ddlLockBoundNotSetWarned.compareAndSet(false, true)) {
            logger.warn(LocalizableMessage.raw("jdbc: the wait of a DDL for its lock is not bounded as this backend"
    private void reportTheWaitIsLeftUnbounded(Dialect dialect, LockBound kind, String sql, Exception e) {
        if (lockBoundNotSetWarned.get(kind).compareAndSet(false, true)) {
            logger.warn(LocalizableMessage.raw("jdbc: the wait of %s for its lock is not bounded as this backend"
                + " means to bound it on this %s database: \"%s\" did not go through, so %s may bound nothing here"
                + " and a DDL can wait for a lock another session holds for as long as this engine lets it (%s)",
                dialect, sql, DDL_LOCK_TIMEOUT_PROPERTY, stackTraceToSingleLineString(e)));
                + " and %s can wait for a lock another session holds for as long as this engine lets it (%s)",
                kind.waiter, dialect, sql, kind.property, kind.waiter, stackTraceToSingleLineString(e)));
        }
    }

    /**
     * Said once per storage, where the engine behind a connection is not one this backend knows a lock
     * setting for. Leaving the bound off such an engine is the conservative reading and stays - untested
     * SQL is no thing to send a database on the path a backend opens by - but a deployment that asked
     * for the bound has no way of finding out that it got none, which is the silence this ends. It is
     * the argument the strict parsing of {@value #DDL_LOCK_TIMEOUT_PROPERTY} is made with, one property
     * later.
     * Said once per storage and per bound, where the engine behind a connection is not one this backend
     * knows a lock setting for. Leaving the bound off such an engine is the conservative reading and
     * stays - untested SQL is no thing to send a database on the path a backend opens by - but a
     * deployment that asked for the bound has no way of finding out that it got none, which is the
     * silence this ends. It is the argument the strict parsing of the property is made with, one
     * property later.
     * <p>
     * Per bound, like every other latch of these bounds: the engine is the same for both, but what an
     * operator is told to act on is the property of the wait that was left unbounded, and a backend
     * opening its trees on such a driver would otherwise have said the only line there is to say
     * before the first write of that backend ever ran.
     * <p>
     * {@link #dialectOf} reads the engine off the class name of the driver, so this is not the engine
     * of an exotic database alone: a mariadb, percona or aurora driver against a live mysql answers
     * null here, and that is a session whose {@code lock_wait_timeout} is a year - the very wait this
     * bound exists to end. {@link CachedConnection} says the same of a url it knows no connect bound
     * for and cannot say it for this one: it keys on the url, which such a driver takes as a mysql one.
     * null here, and that is a session whose {@code lock_wait_timeout} is a year and whose
     * {@code innodb_lock_wait_timeout} is 50 s - the very waits these bounds exist to end.
     * {@link CachedConnection} says the same of a url it knows no connect bound for and cannot say it
     * for this one: it keys on the url, which such a driver takes as a mysql one.
     * <p>
     * The driver is named rather than the url, since the driver is what this reads and what a
     * deployment would change - and a url carries the password of the account this backend works as.
     * <p>
     * The create table this report is issued beside ({@code createCatalogTable()}, and the one
     * The create table the DDL report is issued beside ({@code createCatalogTable()}, and the one
     * {@code openTree()} issues for a tree of its own) is not itself a wait such an engine leaves
     * unbounded: {@code getTableDialect()} answers an unrecognised engine with postgres' column
     * types, which a mysql-family server rejects as a syntax error before it ever queues for a lock.
     * What this warns of - the wait an operator would otherwise meet with no word on why - is the
     * create index of an open whose table is already there, and the drop table of a clear.
     * create index of an open whose table is already there, the drop table of a clear, and every row
     * lock a write of that backend queues for.
     */
    private void reportTheEngineIsNotKnown(Connection con) {
        if (ddlLockBoundEngineUnknownWarned.compareAndSet(false, true)) {
            final String said=String.format("jdbc: the wait of a DDL for a lock is left unbounded on this"
    private void reportTheEngineIsNotKnown(Connection con, LockBound kind) {
        if (lockBoundEngineUnknownWarned.get(kind).compareAndSet(false, true)) {
            final String said=String.format("jdbc: the wait of %s for a lock is left unbounded on this"
                + " database: %s is not a driver this backend knows a lock setting of an engine for, so %s"
                + " bounds nothing here and a DDL - the create index of an open whose table is already there,"
                + " the drop table of a clear - waits for a lock another session holds for as long as this"
                + " engine lets it", driverNameOf(con), DDL_LOCK_TIMEOUT_PROPERTY);
            ddlLockBoundEngineUnknownSaid=said; // read back by a case the way it reads the flag beside it
                + " bounds nothing here and %s waits for a lock another session holds for as long as this"
                + " engine lets it", kind.waiter, driverNameOf(con), kind.property, kind.waiterInDetail);
            // read back by a case the way it reads the flag beside it
            lockBoundEngineUnknownSaid.get(kind).set(said);
            logger.warn(LocalizableMessage.raw(said));
        }
    }

    /**
     * Gives the session back the value it carried. Best effort, and never the outcome of the DDL: this
     * runs from a {@code finally} while the caller may be being unwound, where a throw would replace
     * the failure that brought it there (JLS 14.20.2) - the very one saying what went wrong.
     * Gives the session back the value it carried. Best effort, and never the outcome of the work this
     * bound was put around: this runs from a {@code finally} while the caller may be being unwound,
     * where a throw would replace the failure that brought it there (JLS 14.20.2) - the very one saying
     * what went wrong.
     * <p>
     * A connection this failed on is handed on to nobody. A pooled one is closed rather than given
     * back, and the catalog's own - the one connection reaching this that was never in the pool -
@@ -2435,10 +2994,11 @@
     * that, and is closed with that write. Leaving it to the next borrow to
     * notice does not work: that validation is {@code con.isValid()}, a liveness check which a
     * connection whose reset failed for a transient reason passes while still carrying our bound, and
     * on sql server it would then cut every lock wait of that borrower at it - row locks included,
     * which {@link #isConflict} classifies as no replayable conflict, so {@code write()} does not
     * replay them and a client sees a hard failure. That is the hazard this bound is scoped to a DDL to
     * avoid, arriving through the back door. Kept out of the pool, its blast radius is this one
     * on sql server it would then cut every lock wait of that borrower at it - row locks included. A
     * write of that borrower would replay them ({@link #replayReason} reads the bound this attempt was
     * armed with, not the one left on the session), but a read replays nothing at all, so a client of
     * one sees a hard failure. That is the hazard each of these bounds is scoped to its own piece of
     * work to avoid, arriving through the back door. Kept out of the pool, its blast radius is this one
     * connection instead of the rest of its life.
     * <p>
     * The value is named as the statement that set it rather than read back off the property at log
@@ -2446,25 +3006,26 @@
     * put on it - which is not the configured figure either, where the session's own value was the
     * tighter one.
     */
    private void restoreDdlLockBound(Connection con, Dialect dialect, String bound, String restore) {
    private void restoreLockBound(Connection con, Dialect dialect, ArmedLockBound armed) {
        try {
            boundedSessionCall(con, () -> {
                executeSessionStatement(con, restore);
                executeSessionStatement(con, armed.restore);
                return null;
            });
        }catch (SQLException | RuntimeException e) {
            if (con instanceof CachedConnection) {
                ((CachedConnection) con).keepOutOfThePool();
            }
            final AtomicLong warnedAt=lockBoundLeftBehindWarned.get(armed.kind);
            final long now=System.currentTimeMillis();
            final long last=ddlLockBoundLeftBehindWarned.get();
            if (now-last >= DDL_LOCK_BOUND_WARNING_INTERVAL_MS && ddlLockBoundLeftBehindWarned.compareAndSet(last, now)) {
                logger.warn(LocalizableMessage.raw("jdbc: the lock bound of a DDL could not be taken off a connection"
            final long last=warnedAt.get();
            if (now-last >= LOCK_BOUND_WARNING_INTERVAL_MS && warnedAt.compareAndSet(last, now)) {
                logger.warn(LocalizableMessage.raw("jdbc: the lock bound of %s could not be taken off a connection"
                    + " of this %s database, which may have been left carrying \"%s\" instead of the value it had:"
                    + " that connection is not handed on - a pooled one is closed rather than given back, and the"
                    + " catalog's own carries it for the rest of the write that opened it and is closed with that"
                    + " write - so no later write gives up on a lock at a bound of %s it never asked for (%s)",
                    dialect, bound, DDL_LOCK_TIMEOUT_PROPERTY,
                    armed.kind.waiter, dialect, armed.bound, armed.kind.property,
                    stackTraceToSingleLineString(e)));
            }
        }
@@ -3939,13 +4500,18 @@
     * better than never returning. It is bounded twice - by {@link #MAX_RETRIES} attempts and by the
     * {@link #RETRY_WINDOW_NANOS} wall-clock window - because an attempt is not guaranteed to be short: a conflict
     * an engine reports only after its own lock wait timeout would otherwise multiply that wait by the attempt
     * count. The window alone is not enough either, in the other direction: it is shorter than the wait that
     * precedes a conflict the engine reports promptly, so measured against such a conflict it does not bound that
     * wait but only leaves the operation with no replay at all, which is what master did with the deadlock of
     * issue #903. One replay is therefore granted to a prompt conflict whatever the clock says; see
     * {@link #grantedPastTheWindow}. The window is checked between attempts, so an attempt already running is
     * never interrupted: a conflicted operation holds its caller for the window plus one attempt, and a prompt
     * conflict for two attempts when that is longer.
     * count. The window is checked between attempts, so an attempt already running is never interrupted: a
     * conflicted operation holds its caller for the window plus one attempt.
     * <p>
     * What makes those two bounds enough is that the attempt itself is bounded: the transaction runs under
     * {@link #ROW_LOCK_TIMEOUT_PROPERTY}, armed on the session here and taken off again before the connection is
     * released, so the wait ahead of a conflict ends inside the window rather than outlasting it (#915). Where
     * nothing bounds the attempt - oracle, which has no such setting, and a deployment that turns the property off
     * - the window is shorter than the wait that precedes a conflict the engine reports promptly, so measured
     * against such a conflict it does not bound that wait but only leaves the operation with no replay at all,
     * which is what master did with the deadlock of issue #903. One replay is therefore granted to a prompt
     * conflict whatever the clock says; see {@link #grantedPastTheWindow}. Such a conflict holds its caller for
     * two attempts when that is longer than the window plus one.
     * <p>
     * Only the operation itself is replayed: a failure of {@link #getConnection()} or of the implicit
     * {@link Connection#close()} - which returns the connection to the pool after a rollback - leaves the loop, so
@@ -3956,7 +4522,7 @@
     * connection handed out unvalidated and found dead costs an attempt rather than the operation, and a write of
     * the replication replay - which records a failed operation as applied and advances the server state past it,
     * see #889 - never sees it. Only while nothing of the attempt may have been committed yet, though: see
     * {@link #replayReason(Conflict, Throwable, boolean, boolean, boolean)}.
     * {@link #replayReason}.
     */
    @Override
    public void write(WriteOperation writeOperation) throws Exception {
@@ -3964,6 +4530,10 @@
        for (int attempt=1;;attempt++) {
            Exception failure=null;
            String driver=null;
            Dialect dialect=null;
            //what bounds the wait of this attempt for a row lock, and whether anything does: the window
            //below is a clock, and a clock cannot bound a wait nothing else bounds (#903, #915)
            ArmedLockBound rowLock=ArmedLockBound.none(LockBound.ROW);
            boolean committing=false;
            boolean dropped=false;
            boolean partlyCommitted=false;
@@ -3972,6 +4542,7 @@
            final Connection con=getConnection();
            try (con) {
                driver=driverNameOf(con);
                dialect=dialectOf(con);
                final WriteableTransactionTransactionImpl txn=new WriteableTransactionTransactionImpl(con);
                //the connect of the catalog is made inside this attempt and retries the way a borrow does,
                //up to the pool timeout - six times this window at the defaults. Left to its own deadline
@@ -3982,6 +4553,17 @@
                //each of them. The grant of #903 is deliberately not passed on - it buys one more replay
                //of the operation, not one more connect of the catalog inside it
                txn.catalogSession.boundedAlsoBy(startedAt+RETRY_WINDOW_NANOS);
                //armed against the try that takes it off again, and against nothing else: the release below
                //runs from that finally, before this connection goes back to the pool, and a bound left on a
                //pooled session would cut the lock waits of whoever borrows it next - a read among them, which
                //has no replay to absorb the failure that is. It arms nothing where it cannot (armLockBound),
                //so the operation runs whatever the session made of it.
                //Handed to the transaction as well as kept here: a DDL of this attempt commits, which ends the
                //transaction this bound was scoped to, and the transaction is what takes it off there. This
                //copy is what the attempt ran under, which is what the replay is decided on; txn.rowLock is
                //what is still on the session, which is what the finally below has to give back
                rowLock=armLockBound(con, dialect, LockBound.ROW);
                txn.rowLock=rowLock;
                try {
                    writeOperation.run(txn);
                    committing=true;
@@ -4034,6 +4616,30 @@
                                    stackTraceToSingleLineString(e)));
                        }
                    }
                    //after the rollback above rather than beside it: on the two engines whose bound is a session
                    //setting a rollback leaves that setting exactly where it was, so this is what takes it off -
                    //and it runs while this attempt still holds the connection, before the release hands it on.
                    //It reports what it cannot do and throws nothing, the way every other statement of this
                    //finally does: a value that could not be given back costs the pool this connection, not the
                    //write its failure. Asked of the transaction rather than of the local above: a DDL of this
                    //attempt takes the bound off where it commits, and what is left to give back is what the
                    //transaction still carries.
                    //Not on a connection the driver reports closed: there is nothing there to give a value back
                    //to, and the setting would fail on a dead session and be reported as a bound left behind -
                    //so a restart under write load would read as a stream of stranded bounds for connections
                    //that are simply gone. Decided on that and not on `dropped`: this attempt is classified as
                    //a dropped connection for a class 08 anywhere in the chains of its failure, and a refused
                    //catalog connect puts one there with this connection untouched (newCatalogConnection: mysql
                    //answers its connection limit with 08004, sql server with 08S01). That connection goes back
                    //to the pool - the rollback above went through, the pool's validation is isValid(), and a
                    //live session passes it - so on the two engines whose bound is a session setting, skipping
                    //the restore here hands the next borrow this backend's own bound, a read among them, which
                    //has no replay to absorb the error it would then give up with.
                    //A dead session the driver has not flagged yet takes the other road, which is safe: the
                    //restore fails, it is reported, and the connection is kept out of the pool.
                    if (!isClosed(con)) {
                        releaseLockBound(con, dialect, txn.rowLock);
                    }
                }
            } catch (Exception e) {
                //anything the operation did not throw comes from around it - the name of the driver, the
@@ -4069,7 +4675,7 @@
            //is also the path most likely to carry deeply wrapped chains, since RootContainer.open() commits
            //DDL and raises the flag for the rest of the write
            final ConflictVerdict verdict=partlyCommitted ? NOT_CLASSIFIED : conflictVerdict(failure,driver);
            final String reason=replayReason(verdict.conflict,failure,committing,partlyCommitted,dropped);
            final String reason=replayReason(verdict.conflict,failure,committing,partlyCommitted,dropped,dialect,rowLock);
            //nanoTime()-startedAt is the overflow safe form of the elapsed time
            final long elapsedNanos=nanoTime()-startedAt;
            if (reason==null || !replayableWithin(attempt, elapsedNanos, verdict.conflict)) {
@@ -4131,20 +4737,60 @@
     * OpenDJ issue #896 - which masked the failure that caused the replay and left the indexes of the previous
     * attempt behind with their configuration listeners.
     *
     * A wait for a row lock that this backend bounded is replayable as well, and only where this backend bounded
     * it. The engine ends such a wait with a failure of its own - 55P03 on postgres, error 1222 on sql server,
     * both of which {@link #isConflict} matches as no conflict, and the class 40 of a mysql lock wait timeout,
     * which it already does - and it ends it having rolled nothing back, unlike a deadlock: the transaction is
     * still open and the blocker is still holding the lock. What makes it replayable all the same is the
     * rollback {@link #write} issues before it asks, and what makes it worth replaying is that the wait it took
     * fits inside the replay window rather than outlasting it, which is what {@link ArmedLockBound#bounded}
     * says. Where nothing bounded that wait - {@link #ROW_LOCK_TIMEOUT_PROPERTY} at 0, oracle, a session that
     * would not take the setting - the failure is left exactly as unreplayable as it was before #915: a bound
     * an operator set for themselves is not a licence for this loop to take that wait again, and a wait nothing
     * bounds is the one thing the window cannot govern. It is deliberately no class of {@link Conflict}: the
     * verdict of such a failure stays {@link Conflict#NONE}, so it is replayed on the window alone and never
     * granted past it by {@link #grantedPastTheWindow} - a grant that exists for the waits nothing bounds,
     * which this one is not.
     * <p>
     * The wait a DDL of the same attempt took for its own lock reaches this branch the same way, an engine having
     * one way of saying a lock was not available: a {@code create index} that gave up at
     * {@link #DDL_LOCK_TIMEOUT_PROPERTY} is replayed rather than failing at once, on the terms of every other
     * replay here. Both properties default well under the window, so an attempt of the defaults spends a fraction
     * of it either way; a deployment that raises one of them past the window buys the trailing attempt this loop
     * has always allowed - what it cannot buy is an attempt nothing ends at all.
     *
     * @param conflict the class {@link #conflictVerdict} read from the failure, asked of it once by the caller
     * @param committing whether the failure was reported by {@code commit()}, which leaves the outcome unknown
     * @param partlyCommitted whether the attempt committed part of its work before it failed
     * @param connectionClosed whether the driver closed the connection under the failure - evidence no SQLState
     * carries on mssql-jdbc, which reports a killed session as S0001 and closes the connection behind it
     * @param dialect the engine, which is what names its own way of saying a lock was not available
     * @param rowLock the row lock bound this attempt ran under, as {@link #armLockBound} left it
     */
    static String replayReason(Conflict conflict, Throwable failure, boolean committing, boolean partlyCommitted,
            boolean connectionClosed) {
            boolean connectionClosed, Dialect dialect, ArmedLockBound rowLock) {
        if (partlyCommitted) {
            return null;
        }
        if (conflict!=Conflict.NONE) {
            return "a conflict";
        }
        // after the conflict above, which is the class a mysql lock wait timeout arrives in: both are
        // replayed, and the reason a line names should be the strongest thing that can be said of the
        // failure. Not while committing, for the reason a drop is not replayed there: the outcome of a
        // commit that did not answer is unknown, and this loop must not apply a write twice.
        // Read off every link of the chain, so a lock timeout of the catalog session - a connection of
        // its own, carrying no bound of ours (enrolInCatalog) and giving up only where a deployment set
        // a lock_timeout for itself - is replayed under this reason as well. That reads wider than what
        // the bound of this attempt can say, and stays: the catalog session is made and closed inside
        // the attempt, its statement was rolled back with the rest, and what such a replay costs is
        // bounded by the window like any other - a wait long enough to outlast it is not replayed at all.
        // Which is why the line names the attempt rather than the row lock: a DDL of this attempt that
        // gave up at DDL_LOCK_TIMEOUT_PROPERTY and a lock timeout of its catalog session are both
        // replayed here, and "a row lock" is true of neither
        if (rowLock.bounded && !committing && lockNotAvailable(failure, dialect)) {
            return "a lock wait of an attempt this backend bounded";
        }
        if (!committing && (connectionClosed || isConnectionFailure(failure))) {
            return "a connection the database dropped";
        }
@@ -4313,8 +4959,11 @@
     * <p>
     * Declared in order of how much they restrict the replay, which is the order {@link #conflictVerdict}
     * compares them in: the strongest class any link of a failure carries is the class of that failure. Only
     * {@link #PROMPT} is granted the replay past the window, so every class this list gains - #915 adds one -
     * has to be placed against that grant rather than merely appended.
     * {@link #PROMPT} is granted the replay past the window, so every class this list gains has to be placed
     * against that grant rather than merely appended. The wait for a row lock that #915 bounds is deliberately
     * not one of them: its failure is no conflict at all - see {@link #replayReason}, which replays it on the
     * strength of the bound - and a class here would put it in reach of a grant meant for the waits nothing
     * bounds.
     */
    enum Conflict {
        /** Not a conflict: no replay resolves it. */
@@ -4426,7 +5075,7 @@
     * The strongest class a conflict raised under the given engine can carry, which is where
     * {@link #conflictVerdict} stops walking: nothing further along the chains can outrank it. It is the maximum
     * of what {@link #classOf} returns for that dialect and has to be read together with it - a property of the
     * engine rather than the last constant of {@link Conflict}, so that the class #915 adds cannot silently move
     * engine rather than the last constant of {@link Conflict}, so that a class added later cannot silently move
     * the stop condition, and so that the walk of the three engines reporting no lock wait timeout of their own
     * ends on the first conflict it meets rather than at the end of every chain.
     */
@@ -4455,8 +5104,9 @@
     * {@code openTree(createOnDemand)} is guarded by a catalog read - and its writes go down the ANSI branch of
     * {@code upsert}, which is an {@code update} and an {@code insert}, not a statement a MySQL-wire engine
     * refuses. The cost of the class is one replay of the window's own length for an engine whose conflicts are
     * in fact prompt, which is the direction worth being wrong in; #915 removes the trade by bounding the
     * attempt itself.
     * in fact prompt, which is the direction worth being wrong in. The bound #915 puts on the attempt itself
     * does not remove the trade here: it is armed through the {@link Dialect} the driver name resolves to, so
     * an unrecognised driver is exactly where it arms nothing.
     */
    private static Conflict classOf(SQLException e, Dialect dialect) {
        if (!isConflict(e, dialect)) {
@@ -4493,12 +5143,12 @@
    /**
     * Whether this replay is the one {@link #RETRY_WINDOW_NANOS} does not get to deny: the first replay of a
     * conflict its engine reports promptly, taken although the window is already spent. The wait an engine spends
     * before reporting such a conflict is charged to the attempt that hit it and is unbounded on three of the four
     * engines here - SQL Server took some 12 s to pick a victim in CI - so there is no window that some wait does
     * not outlast, and measuring one against it only leaves the operation with no replay at all, which is issue
     * #903. The grant does not extend to {@link Conflict#AFTER_LOCK_WAIT}, whose wait the engine has already
     * bounded for us: replaying that costs the same bounded wait again, which is exactly what the window is here
     * to refuse. Nor to {@link Conflict#UNKNOWN_ENGINE}, of which the same cannot be ruled out.
     * before reporting such a conflict is charged to the attempt that hit it, and where nothing bounds that wait -
     * SQL Server took some 12 s to pick a victim in CI, before #915 bounded it - there is no window that some
     * wait does not outlast, and measuring one against it only leaves the operation with no replay at all, which
     * is issue #903. The grant does not extend to {@link Conflict#AFTER_LOCK_WAIT}, whose wait the engine has
     * already bounded for us: replaying that costs the same bounded wait again, which is exactly what the window
     * is here to refuse. Nor to {@link Conflict#UNKNOWN_ENGINE}, of which the same cannot be ruled out.
     * <p>
     * Asked as a question of its own so that the line reporting the replay can name the bound that was actually
     * applied instead of inferring it from the clock: {@code elapsed >= window} coincides with this grant only
@@ -4507,8 +5157,14 @@
     * <p>
     * {@code attempt==1} is a proxy and not the invariant: the invariant is that no clock can bound a wait
     * nothing else bounds, and that holds on every attempt, not only the first. Widening the grant to all of them
     * would leave {@link #MAX_RETRIES} as the only real cap, so it is held to one replay until the attempt itself
     * carries a lock bound - see #915, which retires this method rather than widening it.
     * would leave {@link #MAX_RETRIES} as the only real cap, so it is held to one replay.
     * <p>
     * What is left for it to do, now that {@link #ROW_LOCK_TIMEOUT_PROPERTY} bounds the attempt, is the attempts
     * that bound does not reach: oracle, which has no setting for the wait, the property at 0, and a session that
     * would not take the setting - there the wait is as unbounded as it was when this grant was written. Where
     * the bound is armed an attempt gives up on its lock long before this window is spent, so the grant is not
     * asked whether the bound was on: a prompt conflict that still reaches here past the window came out of a
     * long transaction rather than a long wait, and the one replay it is granted runs under the same bound.
     */
    static boolean grantedPastTheWindow(int attempt, long elapsedNanos, Conflict conflict) {
        return attempt==1 && conflict==Conflict.PROMPT && elapsedNanos>=RETRY_WINDOW_NANOS;
@@ -4761,6 +5417,14 @@
         */
        boolean partlyCommitted;

        /**
         * The row lock bound still on the session of this transaction, as {@link JDBCStorage#write} armed
         * it and {@link #commitStatement} may have taken it off again. {@code none} while nothing of ours
         * is on: a transaction of the importer, one on an engine or a setting that took no bound, and
         * every attempt after its own DDL committed.
         */
        ArmedLockBound rowLock=ArmedLockBound.none(LockBound.ROW);

        public WriteableTransactionTransactionImpl(Connection con) {
            this(con, StatementBound.OPERATION);
        }
@@ -4792,6 +5456,15 @@
         * raised in front of it would take a conflict the engine itself undid out of the replay. On those the
         * flag goes up in front of the commit instead, which is the call that leaves the outcome of the
         * transaction unknown when it fails.
         * <p>
         * The row lock bound of the attempt ends here as well, where the statement is a DDL: it was armed
         * around a transaction, and this commit is the end of that transaction. On postgres the
         * {@code set local} goes with the commit by itself, and on the two engines whose setting is the
         * session's this is what takes it off - the attempt behind it has committed part of its work, so
         * it is out of the replay whatever it waits for, and a bound on a wait nothing can replay is a
         * write that fails where it used to go through. In front of the DDL rather than behind it where
         * the DDL bound is turned off, which is the one case the DDL would otherwise run under the row
         * bound of this write - see {@link #liftTheRowLockBoundForAnUnboundedDdl}.
         *
         * @param ddl whether the statement is a DDL one, which two of the four engines commit before
         */
@@ -4810,15 +5483,76 @@
                return null;
            };
            if (ddl) {
                try {
                    liftTheRowLockBoundForAnUnboundedDdl();
                // The DDL of this backend is the part of it that takes locks, and the part that waits for
                // one with no bound of its own. The delete from of clearTree() waits for row locks, which
                // write() replays a conflict of and which this bound has no business ending.
                withDdlLockBound(con, dialectOf(con), issue);
                }finally {
                    // From a finally, since the DDL that failed ended the transaction just as surely on the two
                    // engines that commit before one - and a bound left on for the rest of a write that is out
                    // of the replay is exactly what this takes off. Best effort, like every other release of
                    // one: it says what it could not do and throws nothing over the failure of the DDL
                    releaseTheRowLockBound();
                }
            }else {
                issue.run();
            }
        }

        /**
         * Takes the row lock bound of this attempt off in front of a DDL that is to wait as this backend
         * waited before {@link JDBCStorage#DDL_LOCK_TIMEOUT_PROPERTY} existed, which is what that
         * property at 0 asks for.
         * <p>
         * At 0 nothing of ours is armed around the DDL ({@link JDBCStorage#armLockBound}), and on the two
         * engines with one setting for both waits what it then runs under is not the wait it had before
         * either bound existed but the row bound of this very write - 3 s at the defaults, issued one
         * statement earlier: {@code set lock_timeout 3000} on the sql server session,
         * {@code set local lock_timeout = 3000} on the postgres transaction, which bounds every lock wait
         * of it. A {@code create table} of an open blocked by another session then gives up at 3 s, and
         * gives up as the bare vendor error - nothing of ours being armed, there is no bound to name it
         * by - which {@link JDBCStorage#replayReason} replays as a lock wait of a bounded attempt,
         * re-issuing the DDL until the window is spent. So the bound comes off in front of that DDL
         * rather than behind it: on postgres the {@code set local} is reset, having no value to give
         * back, and on sql server the release puts the deployment's own {@code LOCK_TIMEOUT} back ahead
         * of the statement.
         * <p>
         * Only where this backend armed anything at all: a session already tighter than the bound, an
         * engine that takes no such setting, a setting the session refused and mysql - whose metadata
         * lock is a variable of its own - leave the DDL exactly where it was. And it throws rather than
         * reporting, unlike the release: a {@code set local} that fails has aborted the transaction
         * anyway, and the failure saying why is worth more to an operator than the 25P02 of the DDL
         * behind it.
         */
        private void liftTheRowLockBoundForAnUnboundedDdl() throws SQLException {
            final Dialect dialect=dialectOf(con);
            if (LockBound.DDL.seconds()>0 || rowLock.bound==null || dialect==null
                    || !dialect.oneSettingForBothBounds()) {
                return;
            }
            final String lift=dialect.rowLockLiftSql();
            if (lift!=null) {
                boundedSessionCall(con, () -> {
                    executeSessionStatement(con, lift);
                    return null;
                });
            }
            releaseTheRowLockBound();
        }

        /**
         * Gives the session back what the row lock bound of this attempt displaced, and records that the
         * transaction carries none - so the finally of {@link JDBCStorage#write} has nothing left to give
         * back and nothing is put back twice.
         */
        private void releaseTheRowLockBound() {
            final ArmedLockBound armed=rowLock;
            rowLock=ArmedLockBound.none(LockBound.ROW);
            releaseLockBound(con, dialectOf(con), armed);
        }

        /** Whether this engine commits the transaction before a DDL statement whether asked to or not. */
        private boolean commitsBeforeDdl() {
            final Dialect dialect=dialectOf(con);

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCDdlLockBoundTestCase.java

@@ -18,6 +18,7 @@
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.backends.jdbc.JDBCStorage.Dialect;
import org.opends.server.backends.jdbc.JDBCStorage.LockBound;
import org.opends.server.backends.jdbc.JDBCStorage.StatementBound;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Importer;
@@ -59,6 +60,7 @@
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -216,9 +218,9 @@

        storage.withDdlLockBound(con, null, theDdl());

        assertTrue(storage.ddlLockBoundEngineUnknownWarned.get(),
        assertTrue(storage.lockBoundEngineUnknownWarned.get(LockBound.DDL).get(),
            "a driver this backend knows no lock bound for left the wait of every DDL unbounded and unsaid");
        final String said = storage.ddlLockBoundEngineUnknownSaid;
        final String said = storage.lockBoundEngineUnknownSaid.get(LockBound.DDL).get();
        assertTrue(said != null && said.contains(JDBCStorage.driverNameOf(con)),
            "the driver left unbounded was not named: " + said);
        assertTrue(said != null && said.contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
@@ -233,12 +235,35 @@
    public void testAnEngineThisBackendDoesNotKnowIsReportedOnce() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
        storage.withDdlLockBound(con, null, theDdl());
        assertTrue(storage.ddlLockBoundEngineUnknownWarned.get(), "the first DDL on the engine said nothing");
        assertTrue(storage.lockBoundEngineUnknownWarned.get(LockBound.DDL).get(),
            "the first DDL on the engine said nothing");

        storage.ddlLockBoundEngineUnknownSaid = null;
        storage.lockBoundEngineUnknownSaid.get(LockBound.DDL).set(null);
        storage.withDdlLockBound(con, null, theDdl());

        assertNull(storage.ddlLockBoundEngineUnknownSaid, "an engine this backend does not know was reported twice");
        assertNull(storage.lockBoundEngineUnknownSaid.get(LockBound.DDL).get(),
            "an engine this backend does not know was reported twice");
    }

    /**
     * Once per bound, though: the row lock bound of a write is armed by other code on another path,
     * and the line an operator acts on names the property of the wait that was left unbounded. An
     * open reporting the DDL bound of this engine says nothing about the writes behind it, and a
     * single latch would leave every one of them silent.
     */
    @Test
    public void testTheRowBoundOfAnEngineThisBackendDoesNotKnowIsReportedOnItsOwn() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
        storage.withDdlLockBound(con, null, theDdl());

        storage.armLockBound(con, null, LockBound.ROW);

        final String said = storage.lockBoundEngineUnknownSaid.get(LockBound.ROW).get();
        assertNotNull(said, "the open of a backend silenced the row lock bound of every write behind it");
        assertTrue(said.contains(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY),
            "the property that would have bounded the write was not named: " + said);
        assertFalse(said.contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
            "the write was sent to the property of the DDL bound: " + said);
    }

    /** A deployment that turned the bound off asked for none anywhere, and has nothing to act on. */
@@ -248,7 +273,7 @@

        storage.withDdlLockBound(recording(mock(Connection.class), "0"), null, theDdl());

        assertFalse(storage.ddlLockBoundEngineUnknownWarned.get(),
        assertFalse(storage.lockBoundEngineUnknownWarned.get(LockBound.DDL).get(),
            "a bound nobody asked for was reported as an engine this backend does not know");
    }

@@ -260,7 +285,7 @@
    public void testOracleIsNotReportedAsAnEngineThisBackendDoesNotKnow() throws Exception {
        storage.withDdlLockBound(recording(mock(Connection.class), "0"), Dialect.ORACLE, theDdl());

        assertFalse(storage.ddlLockBoundEngineUnknownWarned.get(),
        assertFalse(storage.lockBoundEngineUnknownWarned.get(LockBound.DDL).get(),
            "the engine left alone deliberately was reported as one this backend cannot bound");
    }

@@ -447,8 +472,13 @@
    }

    /**
     * And a sql server lock wait is left as unreplayable as it was: error 1222 is no conflict of
     * {@code isConflict()}, and a DDL made to look like one would be replayed into the same wait.
     * And the rename leaves a sql server lock wait no conflict of {@code isConflict()}: error 1222 is
     * none, and a DDL made to look like one would be replayed on that class - past the window, where the
     * class is granted one - rather than on what bounded the wait. No conflict is not the same as never
     * replayed: a DDL issued from a write runs inside an attempt carrying a row lock bound of this
     * backend, and a lock wait of such an attempt is replayed on the window like any other (see
     * {@code JDBCStorageRetryTest.testADdlLockWaitInsideAWriteIsReplayedOnTheBoundTheAttemptRanUnder}) -
     * on the window alone, and never past it.
     */
    @Test
    public void testARewrittenSqlServerLockWaitIsMadeNoMoreReplayable() throws Exception {
@@ -514,10 +544,13 @@

        bounded.write(txn -> txn.deleteTree(TREE));

        // the search path in front of them is the lookup that decides whether there is a table to drop
        // at all, narrowed to the schemas an unqualified name of this connection resolves in (#888): it
        // reads a session setting rather than the data, and takes a bound of its own
        assertEquals(issued, asList("select unnest(current_schemas(true))",
        // the first setting is the row lock bound of the write this drop runs inside (#915), armed once
        // per attempt and discarded with the transaction; the search path after it is the lookup that
        // decides whether there is a table to drop at all, narrowed to the schemas an unqualified name of
        // this connection resolves in (#888): it reads a session setting rather than the data, and takes a
        // bound of its own. The DDL bound is the third, and is the one this case is about
        assertEquals(issued, asList("set local lock_timeout = 3000",
            "select unnest(current_schemas(true))",
            "set local lock_timeout = 5000",
            "drop table " + JDBCStorage.toTableName(TREE)));
    }
@@ -893,7 +926,11 @@
        assertEquals(issued, singletonList(THE_DDL));
    }

    /** And a setting that went through is left standing: it is what bounds the DDL that follows it. */
    /**
     * And a setting that went through is left standing: it is what bounds the DDL that follows it. The
     * point it was taken in front of is let go of instead - a {@code SET LOCAL} survives the release,
     * and a subtransaction left open would span the work this bound was put around.
     */
    @Test
    public void testATransactionWhoseSettingWentThroughIsNotTakenBack() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
@@ -903,6 +940,7 @@
        storage.withDdlLockBound(con, Dialect.POSTGRES, theDdl());

        verify(con, never()).rollback(beforeTheBound);
        verify(con).releaseSavepoint(beforeTheBound);
        assertEquals(issued, asList("set local lock_timeout = 5000", THE_DDL));
    }


 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCRowLockBoundTestCase.java

New file
@@ -0,0 +1,803 @@
/*
 * 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.jdbc;

import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.backends.jdbc.JDBCStorage.ArmedLockBound;
import org.opends.server.backends.jdbc.JDBCStorage.Dialect;
import org.opends.server.backends.jdbc.JDBCStorage.LockBound;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Savepoint;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

/**
 * What a transaction of {@code JDBCStorage.write()} is told to do about a row lock another session
 * holds (#915), and what the replay makes of the failure that ends such a wait.
 * <p>
 * It needs no database: the connection is a mock, so what each engine is told - and what it is told
 * to put back before the connection goes to the next borrower - is pinned wherever the build runs,
 * while the container suites cover a write really queued behind another session's row lock. The
 * replay that this bound exists for is driven end to end in {@code JDBCStorageRetryTest}.
 */
@SuppressWarnings("javadoc")
@Test(groups = { "precommit", "jdbc" }, sequential = true)
public class JDBCRowLockBoundTestCase extends DirectoryServerTestCase {

    /** A driver name of an engine none of the settings fit, which is what the null dialect stands for. */
    private static final String UNKNOWN_ENGINE = "com.example.jdbc.Connection";
    private static final String MYSQL = "com.mysql.cj.jdbc.ConnectionImpl";

    /** What the connection of a case was asked to run, in the order it was asked to run it. */
    private final List<String> issued = new ArrayList<>();

    private JDBCStorage storage;

    @BeforeMethod
    public void createStorage() {
        storage = new JDBCStorage(backendCfg(), null);
        issued.clear();
    }

    private static JDBCBackendCfg backendCfg() {
        final JDBCBackendCfg cfg = mockCfg(JDBCBackendCfg.class);
        when(cfg.getBackendId()).thenReturn("rowLockBound");
        return cfg;
    }

    @AfterMethod
    public void clearProperties() {
        System.clearProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY);
        // the cases arming a DDL bound inside this one raise it: the two are decided against each other
        System.clearProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY);
    }

    /**
     * The window bounding the replays is a clock, and a clock only bounds them while an attempt is
     * shorter than it: the default is a third of that window, so a write meeting a lock gets three
     * attempts inside it rather than spending the whole window on one and being refused a replay
     * (#903). A default at or past the window would put this back exactly as it was.
     */
    @Test
    public void testTheDefaultLeavesTheReplayWindowRoomForReplays() {
        assertEquals(JDBCStorage.rowLockBoundSeconds(), 3);
        assertTrue(JDBCStorage.rowLockBoundSeconds() * 1000L * 1000L * 1000L * 3 <= JDBCStorage.RETRY_WINDOW_NANOS,
            "the default bound leaves the replay window room for fewer than three attempts");
    }

    /** Zero is what a deployment that would rather wait for its row lock sets, and so is a negative value. */
    @Test
    public void testTheBoundIsTurnedOffByZero() {
        System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, "0");
        assertEquals(JDBCStorage.rowLockBoundSeconds(), 0);
        System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, "-1");
        assertEquals(JDBCStorage.rowLockBoundSeconds(), 0);
    }

    /** A value that is not a number keeps the default, so a typo cannot silently unbound the wait. */
    @Test
    public void testAValueThatIsNotANumberKeepsTheDefault() {
        System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, "three seconds");
        assertEquals(JDBCStorage.rowLockBoundSeconds(), JDBCStorage.ROW_LOCK_TIMEOUT_SECONDS);
    }

    /** And a value past what a bound of this backend can hold is taken down to it, not read as no bound. */
    @Test
    public void testAValueBeyondTheCeilingIsClamped() {
        System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, String.valueOf(Integer.MAX_VALUE));
        assertEquals(JDBCStorage.rowLockBoundSeconds(), JDBCStorage.MAX_BOUND_SECONDS);
    }

    /**
     * What each engine is told around the transaction, in the unit its own setting takes: the row lock
     * of mysql is innodb_lock_wait_timeout and never the lock_wait_timeout of a metadata lock, sql
     * server has one LOCK_TIMEOUT for every lock wait of a session, postgres takes a set local that the
     * transaction discards, and oracle has no setting for this wait at all.
     */
    @DataProvider
    public Object[][] engines() {
        return new Object[][] {
            { "postgres takes a set local and needs no value back", Dialect.POSTGRES, "0",
                singletonList("set local lock_timeout = 3000"), true },
            { "mysql bounds the row lock, not the metadata lock", Dialect.MYSQL, "50",
                asList("select @@session.innodb_lock_wait_timeout", "set session innodb_lock_wait_timeout=3",
                    "set session innodb_lock_wait_timeout=50"), true },
            { "sql server replaces the -1 that waits forever", Dialect.MICROSOFT, "-1",
                asList("select @@lock_timeout", "set lock_timeout 3000", "set lock_timeout -1"), true },
            { "oracle has no session setting for a row lock", Dialect.ORACLE, "0", emptyList(), false },
            { "an engine this backend does not know is told nothing", null, "0", emptyList(), false },
        };
    }

    @Test(dataProvider = "engines")
    public void testWhatEachEngineIsToldAroundTheTransaction(String name, Dialect dialect, String carries,
            List<String> expected, boolean bounded) throws Exception {
        final Connection con = recording(mock(Connection.class), carries);

        final ArmedLockBound armed = storage.armLockBound(con, dialect, LockBound.ROW);
        storage.releaseLockBound(con, dialect, armed);

        assertEquals(issued, expected, name);
        assertEquals(armed.bounded, bounded, name + ": the wait was reported as bounded when it is not, or the other"
            + " way round - which is what decides whether the replay may take that wait again");
    }

    /**
     * A session that gives up sooner than this bound keeps exactly what it has - the bound is never
     * loosened to ours - and the wait is bounded all the same, which is the whole reason for leaving
     * that value alone. The replay reads that, not whether a statement of ours was issued.
     */
    @DataProvider
    public Object[][] sessionsAlreadyTighter() {
        return new Object[][] {
            { "mysql giving up after a second", Dialect.MYSQL, "1" },
            { "sql server told not to wait at all", Dialect.MICROSOFT, "0" },
        };
    }

    @Test(dataProvider = "sessionsAlreadyTighter")
    public void testASessionAlreadyTighterKeepsWhatItHasAndIsStillBounded(String name, Dialect dialect,
            String carries) throws Exception {
        final Connection con = recording(mock(Connection.class), carries);

        final ArmedLockBound armed = storage.armLockBound(con, dialect, LockBound.ROW);
        storage.releaseLockBound(con, dialect, armed);

        assertEquals(issued, singletonList(dialect == Dialect.MYSQL
            ? "select @@session.innodb_lock_wait_timeout" : "select @@lock_timeout"), name);
        assertTrue(armed.bounded, name + ": a wait the session itself bounds was reported as unbounded");
    }

    /** While a session looser than this bound is given ours, and gets its own value back afterwards. */
    @Test
    public void testASessionLooserThanTheBoundIsGivenOursAndGetsItBack() throws Exception {
        final Connection con = recording(mock(Connection.class), "5000");

        final ArmedLockBound armed = storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);
        storage.releaseLockBound(con, Dialect.MICROSOFT, armed);

        assertEquals(issued, asList("select @@lock_timeout", "set lock_timeout 3000", "set lock_timeout 5000"));
    }

    /** Turned off, nothing is asked of the session and nothing is claimed about the wait. */
    @Test
    public void testTheBoundTurnedOffAsksTheSessionNothing() throws Exception {
        System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, "0");
        final Connection con = recording(mock(Connection.class), "50");

        final ArmedLockBound armed = storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW);
        storage.releaseLockBound(con, Dialect.MYSQL, armed);

        assertEquals(issued, emptyList());
        assertFalse(armed.bounded, "a wait nothing bounds was reported as bounded");
    }

    /**
     * A session answering the readback with something no SET of it would take back is left alone
     * altogether: nothing of ours is ever set where it could not be taken off again.
     */
    @Test
    public void testASessionThatWillNotSayWhatItCarriesIsLeftAlone() throws Exception {
        final Connection con = recording(mock(Connection.class), "unlimited");

        final ArmedLockBound armed = storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW);
        storage.releaseLockBound(con, Dialect.MYSQL, armed);

        assertEquals(issued, singletonList("select @@session.innodb_lock_wait_timeout"));
        assertFalse(armed.bounded, "a session that never took the bound was reported as bounded");
    }

    /**
     * A setting the session refuses leaves the write running as it ran before this bound existed - the
     * bound is an improvement on a wait and never a reason to fail a write that would have gone
     * through - and the wait is not claimed to be bounded, so the replay does not take it again.
     */
    @Test
    public void testASettingTheSessionRefusesLeavesTheWaitUnbounded() throws Exception {
        final Connection con = recording(mock(Connection.class), "50");
        final Statement statement = con.createStatement();
        doAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            throw new SQLException("this session takes no such setting", "42000");
        }).when(statement).execute(anyString());

        final ArmedLockBound armed = storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW);
        storage.releaseLockBound(con, Dialect.MYSQL, armed);

        assertFalse(armed.bounded, "a setting the session refused was reported as bounding the wait");
        // the value is given back all the same: a setting can reach the server and fail only as the
        // statement carrying it is closed, and giving back a value the session may never have left
        // costs a round trip and changes nothing
        assertEquals(issued, asList("select @@session.innodb_lock_wait_timeout",
            "set session innodb_lock_wait_timeout=3", "set session innodb_lock_wait_timeout=50"));
    }

    /**
     * A connection left carrying a bound this backend could not take off again does not go back into
     * the pool: on sql server that setting would cut every lock wait of the next borrower, and a read -
     * which replays nothing, deliberately - would see the error 1222 it ends with.
     */
    @Test
    public void testAConnectionWhoseBoundCouldNotBeTakenOffIsKeptOutOfThePool() throws Exception {
        final AtomicBoolean keptOut = new AtomicBoolean();
        final Connection parent = refusingToGiveTheValueBack(mock(Connection.class), "-1");
        try (final CachedConnection con = new CachedConnection("jdbc:mock", parent) {
            @Override
            void keepOutOfThePool() {
                keptOut.set(true);
                super.keepOutOfThePool();
            }
        }) {
            storage.releaseLockBound(con, Dialect.MICROSOFT,
                storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW));
        }

        assertTrue(keptOut.get(), "a connection left carrying our bound was handed back to the pool");
    }

    /**
     * Postgres needs a transaction block for a set local to mean anything: outside one the server
     * answers it with a warning no driver raises, so the write would run unbounded while the log read
     * exactly like a bounded one.
     */
    @Test
    public void testPostgresInAutoCommitIsToldNothing() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
        when(con.getAutoCommit()).thenReturn(true);

        final ArmedLockBound armed = storage.armLockBound(con, Dialect.POSTGRES, LockBound.ROW);

        assertEquals(issued, emptyList());
        assertFalse(armed.bounded, "a set local that reaches no transaction was reported as bounding the wait");
    }

    /**
     * A failed setting takes the transaction back to the savepoint in front of it: a statement that
     * fails inside a postgres transaction aborts it, and the write would then fail with 25P02 rather
     * than running as unbounded as it ran before this bound existed.
     */
    @Test
    public void testPostgresTakesASavepointInFrontOfTheSetting() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");

        storage.armLockBound(con, Dialect.POSTGRES, LockBound.ROW);

        verify(con).setSavepoint();
    }

    /**
     * And lets go of it once the setting is on: a savepoint is a subtransaction of the write, and one
     * left open spans the whole attempt - every row that attempt writes would then carry the
     * subtransaction's own xid, which readers of those rows resolve through {@code pg_subtrans}. The
     * {@code set local} survives the release, so the bound it was taken in front of stays on.
     */
    @Test
    public void testPostgresLetsGoOfThatSavepointOnceTheSettingIsOn() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
        final Savepoint beforeTheBound = mock(Savepoint.class);
        when(con.setSavepoint()).thenReturn(beforeTheBound);

        storage.armLockBound(con, Dialect.POSTGRES, LockBound.ROW);

        assertEquals(issued, singletonList("set local lock_timeout = 3000"),
            "the bound itself was not issued, or was taken back");
        verify(con).releaseSavepoint(beforeTheBound);
        verify(con, never()).rollback(beforeTheBound);
    }

    /**
     * A setting that failed is taken back to that point instead, and the point is not let go of in front
     * of a rollback that still has to reach it.
     */
    @Test
    public void testASettingThatFailedIsTakenBackToThatSavepoint() throws Exception {
        final Connection con = refusingTheSetting(mock(Connection.class), "0");
        final Savepoint beforeTheBound = mock(Savepoint.class);
        when(con.setSavepoint()).thenReturn(beforeTheBound);

        final ArmedLockBound armed = storage.armLockBound(con, Dialect.POSTGRES, LockBound.ROW);

        assertFalse(armed.bounded, "a setting the session refused was reported as bounding the wait");
        verify(con).rollback(beforeTheBound);
        verify(con, never()).releaseSavepoint(beforeTheBound);
    }

    /**
     * The readback is a round trip, and this bound is armed around every write of the server where the
     * DDL bound is armed around an open: it is paid once per pooled connection. Only this backend
     * writes that setting on a connection of this pool, every write puts the value back, and a
     * connection whose restore failed is kept out of the pool - so what was read cannot go stale.
     */
    @Test
    public void testTheReadbackIsPaidOncePerPooledConnection() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                recording(mock(Connection.class), "50"))) {
            storage.releaseLockBound(con, Dialect.MYSQL, storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW));
            storage.releaseLockBound(con, Dialect.MYSQL, storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW));
        }

        assertEquals(issued, asList("select @@session.innodb_lock_wait_timeout",
            "set session innodb_lock_wait_timeout=3", "set session innodb_lock_wait_timeout=50",
            "set session innodb_lock_wait_timeout=3", "set session innodb_lock_wait_timeout=50"),
            "the value a pooled session carries was read back more than once");
    }

    /**
     * A connection that is not one of this pool is asked every time: the memo above is a property of a
     * session this backend owns for its life, and the catalog and stamp connections are not that.
     */
    @Test
    public void testAConnectionOutsideThePoolIsAskedEveryTime() throws Exception {
        final Connection con = recording(mock(Connection.class), "50");

        storage.releaseLockBound(con, Dialect.MYSQL, storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW));
        storage.releaseLockBound(con, Dialect.MYSQL, storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW));

        assertEquals(issued.stream().filter("select @@session.innodb_lock_wait_timeout"::equals).count(), 2L,
            "a connection outside the pool was asked for the value it carries only once");
    }

    /**
     * The failure an engine ends a bounded wait with is replayable, and only where this backend bounded
     * that wait: 55P03 and error 1222 are no conflict - the engine rolled nothing back and the blocker
     * is still holding the lock - and what makes them worth replaying is that the wait they cost fits
     * inside the replay window. Where nothing bounded the wait they stay exactly as unreplayable as
     * they were: a bound an operator set for themselves is not a licence for this loop to take that
     * wait again, and a wait nothing bounds is the one thing the window cannot govern (#903).
     */
    @DataProvider
    public Object[][] lockTimeouts() {
        return new Object[][] {
            { "postgres lock_timeout", Dialect.POSTGRES, new SQLException("canceling statement due to lock timeout",
                "55P03") },
            { "sql server LOCK_TIMEOUT", Dialect.MICROSOFT, new SQLException("Lock request time out period exceeded.",
                "HY000", 1222) },
        };
    }

    @Test(dataProvider = "lockTimeouts")
    public void testALockTimeoutIsReplayedOnlyWhereThisBackendBoundedTheWait(String name, Dialect dialect,
            SQLException failure) {
        assertEquals(replayReason(failure, dialect, ArmedLockBound.alreadyTighter(LockBound.ROW, 3)),
            "a lock wait of an attempt this backend bounded", name);
        assertNull(replayReason(failure, dialect, ArmedLockBound.none(LockBound.ROW)),
            name + ": a wait this backend put no bound on was replayed on a clock that cannot bound it");
    }

    /**
     * A mysql lock wait timeout arrives in class 40, which is the conflict this loop has replayed since
     * #867: it keeps that reason whether or not this bound is armed, since the line should name the
     * strongest thing that can be said of the failure.
     */
    @Test
    public void testAMysqlLockWaitTimeoutStaysTheConflictItWas() {
        final SQLException lockWait = new SQLException("Lock wait timeout exceeded", "40001", 1205);

        assertEquals(replayReason(lockWait, MYSQL, false, false, false, Dialect.MYSQL,
            ArmedLockBound.none(LockBound.ROW)), "a conflict");
        assertEquals(replayReason(lockWait, MYSQL, false, false, false, Dialect.MYSQL,
            ArmedLockBound.alreadyTighter(LockBound.ROW, 3)), "a conflict");
    }

    /**
     * Not while committing, for the reason a dropped connection is not replayed there: a commit that
     * did not answer leaves the outcome unknown, and this loop must not apply a write twice.
     */
    @Test
    public void testALockTimeoutReportedByTheCommitIsNotReplayed() {
        final SQLException lockTimeout = new SQLException("Lock request time out period exceeded.", "HY000", 1222);

        assertNull(replayReason(lockTimeout, UNKNOWN_ENGINE, true, false, false, Dialect.MICROSOFT,
            ArmedLockBound.alreadyTighter(LockBound.ROW, 3)));
    }

    /** And never once the attempt has committed part of its own work, whatever the failure says. */
    @Test
    public void testALockTimeoutOfAnAttemptThatCommittedPartOfItsWorkIsNotReplayed() {
        final SQLException lockTimeout = new SQLException("Lock request time out period exceeded.", "HY000", 1222);

        assertNull(replayReason(lockTimeout, UNKNOWN_ENGINE, false, true, false, Dialect.MICROSOFT,
            ArmedLockBound.alreadyTighter(LockBound.ROW, 3)));
    }

    /**
     * Read from the failure of the operation only, never from the release of the connection: the
     * rollback that gives a connection back runs after the outcome was decided, so a lock timeout
     * reported there says nothing about the statement that failed.
     */
    @Test
    public void testALockTimeoutOfTheReleaseIsNotReplayed() {
        final SQLException rejected = new SQLException("duplicate key", "23000", 2627);
        rejected.addSuppressed(new SQLException("Lock request time out period exceeded.", "HY000", 1222));

        assertNull(replayReason(rejected, Dialect.MICROSOFT, ArmedLockBound.alreadyTighter(LockBound.ROW, 3)));
    }

    /**
     * sql server has one {@code LOCK_TIMEOUT} for both waits, and every DDL of this backend but the
     * off-write catalog drop is issued from inside a write - which armed this bound one statement
     * earlier. Read live, that DDL would see the 3 s of the row bound, answer "already tighter" to its
     * own 5 s and run at the row bound instead: {@code DDL_LOCK_TIMEOUT_PROPERTY} would govern no DDL of
     * a write at all. What decides it is the value the deployment set, which the row bound remembered on
     * the connection when it displaced it.
     * <p>
     * The readback itself is still paid, and paid live: what the DDL has to put back is the value the
     * session carried a statement ago - the row bound of this very write - rather than the one the
     * connection was borrowed with, which is why only the row bound is remembered per pooled connection.
     */
    @Test
    public void testTheDdlBoundInsideAWriteIsDecidedAgainstWhatTheDeploymentSet() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                liveLockTimeout(mock(Connection.class), "-1"))) {
            final ArmedLockBound row = storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);
            storage.withDdlLockBound(con, Dialect.MICROSOFT, () -> {
                issued.add("the ddl");
                return null;
            });
            storage.releaseLockBound(con, Dialect.MICROSOFT, row);
        }

        assertEquals(issued, asList(
            // the row bound of the write, against what the session carried
            "select @@lock_timeout", "set lock_timeout 3000",
            // the DDL inside it, read live and armed at its own property rather than left at the row bound
            "select @@lock_timeout", "set lock_timeout 5000",
            "the ddl",
            // what the session carried a statement before the DDL, which is the row bound of this write
            "set lock_timeout 3000",
            // and the value the deployment set, once the write is through
            "set lock_timeout -1"));
    }

    /**
     * And a DDL that gives up at that bound is reported as what it is, naming its own property - the
     * rename of #885, which a DDL left at the row bound would lose along with the bound: it would arrive
     * as the bare error 1222, sending an operator to neither property.
     */
    @Test
    public void testADdlInsideAWriteThatGivesUpNamesItsOwnProperty() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                liveLockTimeout(mock(Connection.class), "-1"))) {
            storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);

            try {
                storage.withDdlLockBound(con, Dialect.MICROSOFT, () -> {
                    throw new SQLException("Lock request time out period exceeded.", "HY000", 1222);
                });
                fail("a DDL that gave up on its lock went through");
            } catch (SQLException e) {
                assertTrue(e.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY), e.getMessage());
            }
        }
    }

    /**
     * And that value is put on the session, not merely decided against. A deployment giving up sooner
     * than the DDL bound keeps its own figure - the argument that leaves oracle alone - but "keeps what
     * it has" is not what the session has once the row bound of this write is on that very setting: at a
     * {@code LOCK_TIMEOUT} of 4 s the DDL would run at the 3 s of the row bound, tighter than either
     * property, and give up as the bare error 1222. What the deployment set goes on for the length of
     * the DDL, and the row bound of the write goes back on behind it.
     */
    @Test
    public void testADdlInsideAWriteRunsAtWhatTheDeploymentSetWhereThatIsTighter() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                liveLockTimeout(mock(Connection.class), "4000"))) {
            final ArmedLockBound row = storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);
            storage.withDdlLockBound(con, Dialect.MICROSOFT, () -> {
                issued.add("the ddl");
                return null;
            });
            storage.releaseLockBound(con, Dialect.MICROSOFT, row);
        }

        assertEquals(issued, asList(
            // the row bound of the write, over a deployment looser than it
            "select @@lock_timeout", "set lock_timeout 3000",
            // the DDL inside it, at the value the deployment set rather than at the row bound it met
            "select @@lock_timeout", "set lock_timeout 4000",
            "the ddl",
            // the row bound of the write back, and the deployment's value once the write is through
            "set lock_timeout 3000", "set lock_timeout 4000"));
    }

    /**
     * The same where an operator raised the DDL bound for an index build and the deployment bounds every
     * lock wait of its sessions: 30 s asked for, 10 s allowed, and the row bound of the write is neither.
     */
    @Test
    public void testTheSameWhereTheDdlBoundWasRaisedForAnIndexBuild() throws Exception {
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "30");
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                liveLockTimeout(mock(Connection.class), "10000"))) {
            storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);
            storage.withDdlLockBound(con, Dialect.MICROSOFT, () -> {
                issued.add("the ddl");
                return null;
            });
        }

        assertEquals(issued, asList(
            "select @@lock_timeout", "set lock_timeout 3000",
            "select @@lock_timeout", "set lock_timeout 10000",
            "the ddl",
            "set lock_timeout 3000"));
    }

    /**
     * And a DDL that gives up under that value is left exactly as it arrived: what ended the wait is the
     * deployment's own {@code LOCK_TIMEOUT}, and naming this property for it would send an operator to
     * raise a value that governs nothing while the deployment's own is the tighter one.
     */
    @Test
    public void testADdlThatGaveUpAtTheDeploymentsValueIsNotNamedByThisProperty() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                liveLockTimeout(mock(Connection.class), "4000"))) {
            storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);

            try {
                storage.withDdlLockBound(con, Dialect.MICROSOFT, () -> {
                    throw new SQLException("Lock request time out period exceeded.", "HY000", 1222);
                });
                fail("a DDL that gave up on its lock went through");
            } catch (SQLException e) {
                assertFalse(e.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
                    "a wait the deployment's own value ended was reported as this property's doing: " + e.getMessage());
                assertEquals(e.getErrorCode(), 1222, "the failure of the engine was not handed through as it arrived");
            }
        }
    }

    /**
     * And the same of a lookup, which is the other kind of work this bound is armed around: the one
     * deciding each drop of a clear wraps whatever it sees in a {@code StorageRuntimeException}
     * ({@code isExistsTable}), so it reaches the rename by the unchecked arm rather than the checked
     * one. Both arms read the same thing - whether the session is carrying this bound's own figure -
     * and keying either of them on "a setting of ours was issued" instead would name this property for
     * a wait the deployment's own value ended.
     */
    @Test
    public void testALookupThatGaveUpAtTheDeploymentsValueIsNotNamedByThisPropertyEither() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                liveLockTimeout(mock(Connection.class), "4000"))) {
            storage.armLockBound(con, Dialect.MICROSOFT, LockBound.ROW);

            try {
                storage.withDdlLockBound(con, Dialect.MICROSOFT, () -> {
                    throw new StorageRuntimeException(
                        new SQLException("Lock request time out period exceeded.", "HY000", 1222));
                });
                fail("a lookup that gave up on its lock went through");
            } catch (StorageRuntimeException e) {
                assertFalse(String.valueOf(e.getMessage()).contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
                    "a wait the deployment's own value ended was reported as this property's doing: " + e.getMessage());
            }
        }
    }

    /**
     * On mysql the two waits are two variables - {@code innodb_lock_wait_timeout} for the row lock,
     * {@code lock_wait_timeout} for the metadata lock a DDL waits for - so what the row bound displaced
     * describes neither the other's session nor its default. Reading it there would leave a DDL waiting
     * a year because a deployment had tightened the row lock to a second.
     */
    @Test
    public void testTheDdlBoundOfAMysqlWriteIsDecidedAgainstItsOwnVariable() throws Exception {
        try (final CachedConnection con = new CachedConnection("jdbc:mock",
                answering(mock(Connection.class),
                    "select @@session.innodb_lock_wait_timeout", "1",
                    "select @@session.lock_wait_timeout", "31536000"))) {
            storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW);
            storage.withDdlLockBound(con, Dialect.MYSQL, () -> {
                issued.add("the ddl");
                return null;
            });
        }

        assertEquals(issued, asList(
            // the row bound: this session gives up sooner than ours would, so it keeps what it has
            "select @@session.innodb_lock_wait_timeout",
            // and the metadata lock is bounded all the same, against the variable that bounds it
            "select @@session.lock_wait_timeout", "set session lock_wait_timeout=5",
            "the ddl",
            "set session lock_wait_timeout=31536000"));
    }

    /**
     * The latches these bounds warn through are per bound, not per storage: they are armed by different
     * code on different paths, and an open whose DDL bound this session would not take says nothing
     * about the writes behind it. Through one latch, the first open of a backend would silence every
     * write of it - which is what these two assertions, taken together, keep from happening.
     */
    @Test
    public void testTheWarningLatchesAreOnePerBound() throws Exception {
        final Connection con = refusingTheSetting(mock(Connection.class), "31536000");

        storage.armLockBound(con, Dialect.MYSQL, LockBound.DDL);

        assertTrue(storage.lockBoundNotSetWarned.get(LockBound.DDL).get(),
            "a setting the session refused was not reported for the bound that was armed");
        assertFalse(storage.lockBoundNotSetWarned.get(LockBound.ROW).get(),
            "the DDL bound of an open silenced the row lock bound of every write behind it");

        storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW);

        assertTrue(storage.lockBoundNotSetWarned.get(LockBound.ROW).get(),
            "the row lock bound of a write said nothing of its own");
    }

    /**
     * The same, for the moment throttling what a bound left behind says - and asked of the row bound,
     * which is the one a single shared moment would leave unsaid: what the two of them have to be is
     * one per bound, in both directions.
     */
    @Test
    public void testTheLatchOfABoundLeftBehindIsOnePerBoundToo() throws Exception {
        final Connection con = refusingToGiveTheValueBack(mock(Connection.class), "31536000");

        storage.releaseLockBound(con, Dialect.MYSQL, storage.armLockBound(con, Dialect.MYSQL, LockBound.ROW));

        assertTrue(storage.lockBoundLeftBehindWarned.get(LockBound.ROW).get() != 0,
            "a bound that could not be taken off was not reported for the bound that was armed");
        assertEquals(storage.lockBoundLeftBehindWarned.get(LockBound.DDL).get(), 0L,
            "the row lock bound of a write silenced the DDL bound of every open of this backend");
    }

    private static String replayReason(SQLException failure, Dialect dialect, ArmedLockBound rowLock) {
        return replayReason(failure, UNKNOWN_ENGINE, false, false, false, dialect, rowLock);
    }

    /**
     * The two questions {@code write()} asks after a failed attempt, composed here the way it composes
     * them: the conflict class is read off the failure once and handed to the reason, rather than being
     * asked for again.
     */
    private static String replayReason(SQLException failure, String driver, boolean committing,
            boolean partlyCommitted, boolean connectionClosed, Dialect dialect, ArmedLockBound rowLock) {
        return JDBCStorage.replayReason(JDBCStorage.conflictVerdict(failure, driver).conflict, failure, committing,
            partlyCommitted, connectionClosed, dialect, rowLock);
    }

    /**
     * A connection recording every session statement it is given, and answering the readback of the
     * setting with the value a session of that engine carries.
     */
    private Connection recording(final Connection con, final String carries) throws SQLException {
        final Statement statement = mock(Statement.class);
        when(statement.execute(anyString())).thenAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            return false;
        });
        when(statement.executeQuery(anyString())).thenAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            final ResultSet carried = mock(ResultSet.class);
            when(carried.next()).thenReturn(true, false);
            when(carried.getString(1)).thenReturn(carries);
            return carried;
        });
        when(con.createStatement()).thenReturn(statement);
        return con;
    }

    /**
     * A connection answering each readback its own value, which is what an engine whose two waits are
     * two variables does: every other fixture here answers one value to every query, so a case over one
     * of them cannot tell the value of one setting from the value of the other.
     *
     * @param answers the query and the value it is answered with, in pairs
     */
    private Connection answering(final Connection con, final String... answers) throws SQLException {
        final Statement statement = mock(Statement.class);
        when(statement.execute(anyString())).thenAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            return false;
        });
        when(statement.executeQuery(anyString())).thenAnswer(invocation -> {
            final String query = (String) invocation.getArguments()[0];
            issued.add(query);
            final ResultSet carried = mock(ResultSet.class);
            when(carried.next()).thenReturn(true, false);
            for (int i = 0; i < answers.length; i += 2) {
                if (answers[i].equals(query)) {
                    when(carried.getString(1)).thenReturn(answers[i + 1]);
                    return carried;
                }
            }
            throw new SQLException("this fixture answers no " + query, "42000");
        });
        when(con.createStatement()).thenReturn(statement);
        return con;
    }

    /**
     * A sql server connection answering the readback with what the last {@code set lock_timeout} left on
     * it - a live session rather than a fixed value, which is what a bound armed inside another one
     * meets.
     */
    private Connection liveLockTimeout(final Connection con, final String initially) throws SQLException {
        final AtomicReference<String> carried = new AtomicReference<>(initially);
        final Statement statement = mock(Statement.class);
        when(statement.execute(anyString())).thenAnswer(invocation -> {
            final String sql = (String) invocation.getArguments()[0];
            issued.add(sql);
            if (sql.startsWith("set lock_timeout ")) {
                carried.set(sql.substring("set lock_timeout ".length()));
            }
            return false;
        });
        when(statement.executeQuery(anyString())).thenAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            final ResultSet rows = mock(ResultSet.class);
            when(rows.next()).thenReturn(true, false);
            when(rows.getString(1)).thenReturn(carried.get());
            return rows;
        });
        when(con.createStatement()).thenReturn(statement);
        return con;
    }

    /** A connection that says what it carries and will not take the setting of a bound at all. */
    private Connection refusingTheSetting(final Connection con, final String carries) throws SQLException {
        final Statement statement = recording(con, carries).createStatement();
        doAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            throw new SQLException("this session takes no such setting", "42000");
        }).when(statement).execute(anyString());
        return con;
    }

    /** A connection that takes the bound and will not take back the value that bound displaced. */
    private Connection refusingToGiveTheValueBack(final Connection con, final String carries) throws SQLException {
        final Statement statement = recording(con, carries).createStatement();
        final AtomicBoolean bound = new AtomicBoolean();
        doAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            if (!bound.compareAndSet(false, true)) {
                throw new SQLException("the connection went before the value could be given back", "08006");
            }
            return false;
        }).when(statement).execute(anyString());
        return con;
    }
}

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java

@@ -15,6 +15,7 @@
 */
package org.opends.server.backends.jdbc;

import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.backends.jdbc.JDBCStorage.Conflict;
@@ -40,6 +41,8 @@
import java.sql.SQLNonTransientConnectionException;
import java.sql.SQLRecoverableException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
@@ -49,6 +52,8 @@
import java.util.function.Predicate;
import java.util.logging.Logger;

import static java.util.Collections.emptyList;
import static java.util.Collections.frequency;
import static org.forgerock.i18n.LocalizableMessage.raw;
import static org.forgerock.opendj.ldap.ResultCode.OTHER;
import static org.mockito.Mockito.any;
@@ -56,7 +61,9 @@
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
@@ -98,6 +105,14 @@
  /** A MySQL-wire-compatible driver, whose class name carries no engine this backend recognises. */
  private static final String MARIADB = "org.mariadb.jdbc.Connection";

  /**
   * What the cases below arm around a failure they classify: nothing. They are about the other classes of
   * failure - a conflict, a connection the database dropped - and a row lock bound this backend did not put on
   * is the state every one of them was written under (#915).
   */
  private static final JDBCStorage.ArmedLockBound NO_ROW_LOCK_BOUND =
      JDBCStorage.ArmedLockBound.none(JDBCStorage.LockBound.ROW);

  /** The tree every write of this test opens; the table name behind it is a hash of this name. */
  private static final TreeName TREE = new TreeName("dc=example,dc=com", "id2entry");

@@ -134,6 +149,11 @@
  {
  }

  /** sql server, whose one {@code LOCK_TIMEOUT} is read back and put back around every write. */
  interface microsoftConnection extends Connection
  {
  }

  /**
   * A MySQL-wire-compatible driver none of the four engines is recognised in - MariaDB Connector/J, an Aurora-
   * or Percona-branded one. It reports a lock wait timeout as 1205 under class 40 exactly as Connector/J does,
@@ -1138,6 +1158,358 @@
  }

  /**
   * The wait a write takes for a row lock another session holds is bounded on the session of the attempt
   * ({@code ROW_LOCK_TIMEOUT_PROPERTY}, #915), and the failure the engine ends that wait with is replayed. On
   * postgres that failure is 55P03, which is no conflict - nothing was rolled back and the blocker still holds
   * the lock - and what makes it worth replaying is that the wait it cost fits inside the replay window, which
   * is the only thing that lets a clock bound these replays at all (#903).
   * <p>
   * The bound is armed per attempt: on postgres it is a {@code set local}, which the rollback of the attempt
   * that failed discards along with everything else the attempt did.
   */
  @Test
  public void testARowLockWaitThisBackendBoundedIsReplayed() throws Exception
  {
    final JDBCStorage storage = storageOverAnEngine(postgresConnection.class, true);
    final List<String> sessionStatements = sessionStatementsOf(engineConnection);
    when(statements.executeUpdate()).thenThrow(sql(0, "55P03")).thenReturn(1);
    final AtomicInteger attempts = new AtomicInteger();

    storage.write(txn -> {
      attempts.incrementAndGet();
      txn.openTree(TREE, false);
      txn.put(TREE, ByteString.valueOfUtf8("dc=example,dc=com"), ByteString.valueOfUtf8("an entry"));
    });

    assertEquals(attempts.get(), 2, "a row lock wait this backend bounded was not replayed");
    assertEquals(sessionStatements.stream().filter("set local lock_timeout = 3000"::equals).count(), 2L,
        "the bound was not armed once per attempt: " + sessionStatements);
  }

  /**
   * And where nothing bounds that wait it is not replayed, which is what a deployment asks for by setting the
   * property to 0: this loop must not take again a wait that has no end of its own, and the failure is left
   * exactly as unreplayable as it was before #915 - the behaviour every engine had, and oracle still has.
   */
  @Test
  public void testARowLockWaitNothingBoundedIsNotReplayed() throws Exception
  {
    System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, "0");
    try
    {
      final JDBCStorage storage = storageOverAnEngine(postgresConnection.class, true);
      final List<String> sessionStatements = sessionStatementsOf(engineConnection);
      when(statements.executeUpdate()).thenThrow(sql(0, "55P03")).thenReturn(1);
      final AtomicInteger attempts = new AtomicInteger();

      try
      {
        storage.write(txn -> {
          attempts.incrementAndGet();
          txn.openTree(TREE, false);
          txn.put(TREE, ByteString.valueOfUtf8("dc=example,dc=com"), ByteString.valueOfUtf8("an entry"));
        });
        fail("a lock wait this backend put no bound on was replayed");
      }
      catch (StorageRuntimeException expected)
      {
        assertEquals(attempts.get(), 1, "an unbounded lock wait was taken a second time");
      }
      assertEquals(sessionStatements, emptyList(), "a bound turned off was armed all the same");
    }
    finally
    {
      System.clearProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY);
    }
  }

  /**
   * A DDL of an attempt commits, and that commit is the end of the transaction the row lock bound was armed
   * around: on postgres the {@code set local} goes with it by itself, and on the two engines whose setting is
   * the session's {@code commitStatement()} is what takes it off. What follows such a DDL is a write that has
   * committed part of its work, so it is out of the replay whatever it waits for - and a bound on a wait
   * nothing can replay only fails a write at 3 s where it used to wait for the blocker and go through.
   * <p>
   * The order is what says it: the value is put back before the statement behind the DDL is issued, rather
   * than in the finally of the attempt, which runs after the whole write is through.
   */
  @Test
  public void testTheRowLockBoundComesOffWhereADdlOfTheAttemptCommits() throws Exception
  {
    final JDBCStorage storage = storageOverAnEngine(mysqlConnection.class, false);
    final List<String> issued = statementsOf(engineConnection, "50");

    storage.write(txn -> {
      txn.openTree(TREE, true); // a create index, which commits
      txn.put(TREE, ByteString.valueOfUtf8("dc=example,dc=com"), ByteString.valueOfUtf8("an entry"));
    });

    final int boundArmed = issued.indexOf("set session innodb_lock_wait_timeout=3");
    final int boundOff = issued.indexOf("set session innodb_lock_wait_timeout=50");
    final int theDdl = indexOfFirst(issued, "create index k_");
    final int behindTheDdl = indexOfFirst(issued, "insert into ");
    assertTrue(boundArmed >= 0 && theDdl > boundArmed, "the attempt issued no bounded DDL: " + issued);
    assertTrue(boundOff > theDdl, "the row lock bound was not taken off where the DDL committed: " + issued);
    assertTrue(behindTheDdl > boundOff,
        "the rest of the write ran under a bound its DDL had already taken out of the replay: " + issued);
    // and taken off once: the transaction records that it carries none, so the finally of write() has
    // nothing left to give back and the value is not put back over whatever the rest of the write left
    assertEquals(frequency(issued, "set session innodb_lock_wait_timeout=50"), 1,
        "the value the row lock bound displaced was put back twice: " + issued);
  }

  /**
   * And it is not put back on a connection the driver reports closed: there is nothing there to give it back
   * to, and the setting would fail on a dead session and be reported as a bound left behind - a database
   * restart under write load would read as a stream of stranded bounds for connections that are gone.
   * <p>
   * Which is what the guard is keyed on, rather than on the attempt being classified as a dropped connection:
   * the case below holds the other side of that.
   */
  @Test
  public void testTheRowLockBoundIsNotPutBackOnAConnectionTheDatabaseDropped() throws Exception
  {
    final JDBCStorage storage = storageOverAnEngine(mysqlConnection.class, false);
    final List<String> issued = statementsOf(engineConnection, "50");
    // the state the driver leaves such a connection in, which is what tells it from a connection a failure of
    // this class merely passed through: mssql-jdbc closes the connection for any error of severity 20 and
    // above before it throws, and connector/j closes one the server hung up on
    when(engineConnection.isClosed()).thenReturn(true);

    try
    {
      storage.write(txn -> {
        throw new StorageRuntimeException(sql(0, "08006"));
      });
      fail("the drop of the connection was swallowed");
    }
    catch (StorageRuntimeException expected)
    {
      assertTrue(JDBCStorage.isConnectionFailure(expected), "the failure this case rests on is not a drop");
    }

    assertTrue(issued.contains("set session innodb_lock_wait_timeout=3"),
        "the write armed no row lock bound: " + issued);
    assertEquals(frequency(issued, "set session innodb_lock_wait_timeout=50"), 0,
        "the bound was given back on a connection the database had dropped: " + issued);
  }

  /**
   * While a connection that only this attempt's classification made look dropped is given its value back. A
   * write enrolling a tree opens a connection of the catalog's own, and a database refusing that connect
   * answers a state of class 08 - mysql answers its connection limit with 08004, sql server with 08S01 - which
   * puts the attempt in the replay as a dropped connection with the pooled connection of the write untouched.
   * That connection does go back to the pool: the rollback went through, the validation of the next borrow is
   * {@code isValid()}, and a live session passes it. On the two engines whose bound is a session setting,
   * skipping the restore there hands the next borrow this backend's own 3 s - a read among them, and a read
   * that gives up at a lock wait has no replay to absorb it.
   */
  @Test
  public void testTheRowLockBoundIsPutBackOnALiveConnectionARefusedCatalogConnectMadeLookDropped() throws Exception
  {
    final JDBCStorage storage = storageOverAnEngine(mysqlConnection.class, false);
    final List<String> issued = statementsOf(engineConnection, "50");

    try
    {
      storage.write(txn -> {
        throw new StorageRuntimeException(new SQLException("Too many connections", "08004", 1040));
      });
      fail("the refusal was swallowed");
    }
    catch (StorageRuntimeException expected)
    {
      assertTrue(JDBCStorage.isConnectionFailure(expected), "the failure this case rests on is not read as a drop");
    }

    // every attempt of the replay, rather than one of them: what the case is about is that no attempt leaves
    // its bound on a connection it hands back, and a refusal of this class is replayed like any other drop
    final int armed = frequency(issued, "set session innodb_lock_wait_timeout=3");
    assertTrue(armed > 0, "the write armed no row lock bound: " + issued);
    assertEquals(frequency(issued, "set session innodb_lock_wait_timeout=50"), armed,
        "the bound was left on a connection the driver still reports open: " + issued);
  }

  /**
   * A DDL told to wait as this backend waited before its bound existed ({@code DDL_LOCK_TIMEOUT_PROPERTY} at 0)
   * is not left waiting at the row bound of the write it is issued from. On postgres that bound is a
   * {@code set local}, which bounds every lock wait of the transaction, so the create table of an open would
   * give up at 3 s - as the bare vendor error, nothing of ours being armed around it - and be replayed as a
   * lock wait of the attempt until the window was spent. The bound comes off in front of the DDL instead.
   * <p>
   * Off the write and not off each of its DDL: the open of a tree that is not there issues two - the create
   * table and the create index behind it - and the second has no bound of this backend left to lift.
   */
  @Test
  public void testADdlToldToWaitDoesNotWaitAtTheRowBoundOnPostgres() throws Exception
  {
    System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "0");
    try
    {
      final JDBCStorage storage = storageOverAnEngine(postgresConnection.class, false);
      tablesAreNotThere(engineConnection);
      final List<String> issued = statementsOf(engineConnection, "0");

      storage.write(txn -> txn.openTree(TREE, true));

      final int bound = issued.indexOf("set local lock_timeout = 3000");
      final int lifted = issued.indexOf("set local lock_timeout to default");
      final int theDdl = indexOfFirst(issued, "create index ");
      assertTrue(bound >= 0, "the write armed no row lock bound: " + issued);
      assertTrue(lifted > bound, "the row lock bound was not taken off in front of an unbounded DDL: " + issued);
      assertTrue(theDdl > lifted, "the DDL ran under the row bound of the write that issued it: " + issued);
      assertTrue(indexOfFirst(issued, "create table ") >= 0, "the open of this case created no table: " + issued);
      assertEquals(frequency(issued, "set local lock_timeout to default"), 1,
          "the row lock bound was lifted once per DDL rather than once per write: " + issued);
    }
    finally
    {
      System.clearProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY);
    }
  }

  /**
   * The same on sql server, where the setting is the session's and what puts the wait back where the deployment
   * left it is the value the row bound displaced - here the -1 that waits forever, which is what this backend
   * waited before either bound existed. The DDL is the create table of a tree that is not there: this engine
   * indexes nothing behind it, {@code k} being a {@code varbinary(max)} no index key column can hold.
   */
  @Test
  public void testADdlToldToWaitDoesNotWaitAtTheRowBoundOnSqlServer() throws Exception
  {
    System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "0");
    try
    {
      final JDBCStorage storage = storageOverAnEngine(microsoftConnection.class, false);
      tablesAreNotThere(engineConnection);
      final List<String> issued = statementsOf(engineConnection, "-1");

      storage.write(txn -> txn.openTree(TREE, true));

      final int bound = issued.indexOf("set lock_timeout 3000");
      final int lifted = issued.indexOf("set lock_timeout -1");
      final int theDdl = indexOfFirst(issued, "create table ");
      assertTrue(bound >= 0, "the write armed no row lock bound: " + issued);
      assertTrue(lifted > bound, "the deployment's own value was not put back in front of an unbounded DDL: "
          + issued);
      assertTrue(theDdl > lifted, "the DDL ran under the row bound of the write that issued it: " + issued);
    }
    finally
    {
      System.clearProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY);
    }
  }

  /**
   * While a DDL that has a bound of its own is left exactly where it was: it arms that bound over the row one
   * and puts it back afterwards, so taking the row bound off in front of it would be a round trip per DDL -
   * about 25 of them per suffix of an open - buying nothing.
   */
  @Test
  public void testABoundedDdlLeavesTheRowBoundOfTheWriteWhereItIs() throws Exception
  {
    final JDBCStorage storage = storageOverAnEngine(postgresConnection.class, false);
    final List<String> issued = statementsOf(engineConnection, "0");

    storage.write(txn -> txn.openTree(TREE, true));

    assertTrue(issued.contains("set local lock_timeout = 3000"), "the write armed no row lock bound: " + issued);
    assertEquals(frequency(issued, "set local lock_timeout to default"), 0,
        "a DDL with a bound of its own took the row bound off in front of itself: " + issued);
  }

  /**
   * And a DDL that did give up at its own bound inside a write is replayed like any other lock wait of a
   * bounded attempt - on the copy of the bound that attempt ran under, which is what {@code write()} keeps
   * for the replay while {@code commitStatement()} takes the transaction's copy off at the DDL that ends it.
   * Decided on the transaction's copy instead, such a failure would be thrown at the first attempt.
   */
  @Test
  public void testADdlLockWaitInsideAWriteIsReplayedOnTheBoundTheAttemptRanUnder() throws Exception
  {
    final JDBCStorage storage = storageOverAnEngine(postgresConnection.class, false);
    sessionStatementsOf(engineConnection); // the row bound of the attempt: the fixture takes no session statement
    when(statements.executeUpdate()).thenThrow(sql(0, "55P03")).thenReturn(0);
    final AtomicInteger attempts = new AtomicInteger();

    storage.write(txn -> {
      attempts.incrementAndGet();
      txn.openTree(TREE, true);
    });

    assertEquals(attempts.get(), 2, "a DDL that gave up at its own bound inside a bounded attempt was not replayed");
  }

  /** Where the first statement of the attempt starting with the given text was issued, or -1. */
  private static int indexOfFirst(List<String> issued, String startsWith)
  {
    for (int i = 0; i < issued.size(); i++)
    {
      if (issued.get(i).startsWith(startsWith))
      {
        return i;
      }
    }
    return -1;
  }

  /**
   * Answers the metadata of a connection with a database holding none of the tables asked about, so that an
   * open creates them: the fixture of {@code storageOverAnEngine()} answers that every table asked about is
   * there, which is the existing backend most cases here are about.
   */
  private static void tablesAreNotThere(Connection con) throws Exception
  {
    final ResultSet none = mock(ResultSet.class);
    when(none.next()).thenReturn(false);
    // the metadata is taken out of the chain first, and the stubbing is a doReturn: asking when() for the
    // value of a call already stubbed with an answer would run that answer here, and it stubs a result set
    // of its own as it goes
    final DatabaseMetaData metaData = con.getMetaData();
    doReturn(none).when(metaData).getTables(any(), any(), any(), any());
  }

  /**
   * Records the session statements a connection is given - the bound of an attempt among them - on a fixture
   * whose {@code createStatement()} otherwise refuses them. Stubbed through {@code doReturn}, since asking
   * {@code when()} for the value of a call already stubbed to throw would raise that throw here.
   */
  private static List<String> sessionStatementsOf(Connection con) throws Exception
  {
    final List<String> issued = new ArrayList<>();
    final Statement statement = mock(Statement.class);
    when(statement.execute(anyString())).thenAnswer(invocation -> {
      issued.add((String) invocation.getArguments()[0]);
      return false;
    });
    doReturn(statement).when(con).createStatement();
    return issued;
  }

  /**
   * The same, with the statements of the work itself in the very same list and the readback of a session
   * setting answered: what a case reads off this is the order the two were issued in, which is what a bound
   * armed around a transaction and taken off inside it can only be pinned by.
   */
  private List<String> statementsOf(Connection con, String carries) throws Exception
  {
    final List<String> issued = sessionStatementsOf(con);
    final Statement statement = con.createStatement();
    when(statement.executeQuery(anyString())).thenAnswer(invocation -> {
      issued.add((String) invocation.getArguments()[0]);
      final ResultSet carried = mock(ResultSet.class);
      when(carried.next()).thenReturn(true, false);
      when(carried.getString(1)).thenReturn(carries);
      return carried;
    });
    doAnswer(invocation -> {
      issued.add((String) invocation.getArguments()[0]);
      return statements;
    }).when(con).prepareStatement(anyString());
    return issued;
  }

  /**
   * mysql commits before a DDL statement whether asked to or not, so a create index that failed there has
   * committed everything the transaction did before it just as surely as one that succeeded: the attempt is out
   * of the replay whatever the failure says.
@@ -1432,7 +1804,7 @@
      boolean connectionClosed)
  {
    return JDBCStorage.replayReason(conflictOf(failure, driver), failure, committing, partlyCommitted,
        connectionClosed);
        connectionClosed, JDBCStorage.dialectOf(driver), NO_ROW_LOCK_BOUND);
  }

  /**

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java

@@ -291,12 +291,18 @@
     * Oracle asserts the other half of the same contract: nothing of ours is set there, because its
     * own {@code ddl_lock_timeout} gives up at once - so the drop still fails rather than hanging, and
     * the failure names the engine's own doing rather than a property of ours that armed nothing.
     * <p>
     * The bound is set looser than the row lock bound of the write this drop runs inside, deliberately:
     * sql server has one {@code LOCK_TIMEOUT} for both waits, so a DDL bound tighter than the row one
     * passes this case whether this backend armed it or merely inherited the row bound of the write
     * around it - and inheriting it is exactly what would leave this property governing nothing and the
     * failure naming neither property (#915).
     */
    @Test(timeOut = 120000)
    public void testTheDdlGivesUpOnALockAnotherSessionHolds() throws Exception {
        final TreeName tree = new TreeName("testDdlLockBound", "tree");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "2");
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "7");
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
@@ -728,6 +734,93 @@
            });
    }

    /**
     * A write queued behind a row lock another session holds gives up at
     * {@code ROW_LOCK_TIMEOUT_PROPERTY} and is replayed, and the replay goes through once that session
     * lets the rows go: the wait ends inside the replay window instead of consuming it, which is what
     * left the operation with no replay at all before (#903, #915).
     * <p>
     * On an engine with no session setting for that wait - oracle - the same write waits the blocker
     * out on its first attempt and needs no replay, which is what "left alone" means there. That is
     * asserted rather than skipped: it is the difference this bound makes, read from the one place it
     * shows.
     */
    @Test(timeOut = 300000)
    public void testAWriteBlockedByARowLockIsReplayedInsideTheWindow() throws Exception {
        final int boundSeconds = 2;
        final TreeName tree = new TreeName("testRowLockBound", "tree");
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
        System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, Integer.toString(boundSeconds));
        final ExecutorService releasing = Executors.newSingleThreadExecutor();
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                    txn.put(tree, key(1), value(1));
                }
            });

            // a session of its own, taking an exclusive lock on the row the write below wants: an update
            // that changes nothing takes the same lock as one that does, and it parses on all four engines
            try (final Connection holder = DriverManager.getConnection(getJdbcUrl())) {
                holder.setAutoCommit(false);
                try (final PreparedStatement lock = holder.prepareStatement(
                        "update " + storage.getTableName(tree) + " set v=v")) {
                    lock.executeUpdate();
                }
                // held for twice the bound, so that the first attempt of the write gives up on it and the
                // replay finds the row free - both inside the replay window, which is what this is about
                final Future<?> released = releasing.submit(new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        Thread.sleep(boundSeconds * 2 * 1000L);
                        holder.rollback(); // the rows go back exactly as they were
                        return null;
                    }
                });

                final AtomicInteger attempts = new AtomicInteger();
                final long startedAt = System.nanoTime();
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        attempts.incrementAndGet();
                        txn.put(tree, key(1), value(2));
                    }
                });
                final long elapsedMillis = (System.nanoTime() - startedAt) / 1000000L;
                released.get(); // the failure of the holder, if it had one, rather than a stuck rollback

                if (dialect() == JDBCStorage.Dialect.ORACLE) {
                    assertEquals(attempts.get(), 1, "oracle has no session setting for a row lock enqueue, so the"
                        + " write should have waited the holder out on its first attempt (" + elapsedMillis + " ms)");
                } else {
                    assertTrue(attempts.get() >= 2, "the write was not replayed: it waited " + elapsedMillis
                        + " ms for a row lock that this engine was told to give up on after " + boundSeconds + " s");
                    assertTrue(elapsedMillis >= boundSeconds * 1000L - JDBCStorage.CLOCK_SLACK_MILLIS,
                        "the write came back after " + elapsedMillis + " ms, before the bound it was given:"
                        + " something other than the row lock ended its first attempt");
                }
            }
        } finally {
            System.clearProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY);
            releasing.shutdownNow();
            try {
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        txn.deleteTree(tree);
                    }
                });
            } catch (Exception ignored) {
            } finally {
                storage.close();
            }
        }
    }

    private interface BlockedOperation {
        void run(JDBCStorage storage, TreeName tree) throws Exception;
    }
@@ -807,6 +900,11 @@
                    for (final JDBCStorage.StatementBound each : JDBCStorage.StatementBound.values()) {
                        System.setProperty(each.property, each == bound ? Integer.toString(boundSeconds) : "0");
                    }
                    // and the row lock bound of a write is off here for the same reason: it is tighter
                    // than the bound under test, so it - and not the statement bound this case is about -
                    // would be what ended the wait, and the failure would name no property of this class
                    // (#915). testAWriteBlockedByARowLockIsReplayedInsideTheWindow covers that layer
                    System.setProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY, "0");
                    // the monotonic clock, which is what timedOut() measures the bound with: a step of
                    // the wall clock can neither lengthen nor shorten what the assertions below allow
                    final long startedAt = System.nanoTime();
@@ -868,6 +966,7 @@
            for (final JDBCStorage.StatementBound each : JDBCStorage.StatementBound.values()) {
                System.clearProperty(each.property);
            }
            System.clearProperty(JDBCStorage.ROW_LOCK_TIMEOUT_PROPERTY);
            try {
                storage.write(new WriteOperation() {
                    @Override