[#885] Bound the wait of a JDBC DDL for a lock another session holds (#936)

Maxim Thomas

yesterday 21d03d579b5c56bf17d763412179bc7a0e16168c

[#885] Bound the wait of a JDBC DDL for a lock another session holds (#936)

 .gitignore

@@ -20,6 +20,9 @@
buildNumber.properties
.mvn/timing.properties

# Build output captured to a file
*.log


#--- IntelliJ ignores ---
# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio

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

@@ -1199,8 +1199,13 @@
     * read bound that could not be lifted serves the borrower waiting for it and is closed
     * afterwards: left in the pool it would fail every statement slower than that bound - an
     * import batch among them - for every borrow the pool hands it to.
     * <p>
     * Turned off during a borrow as well as at establish time ({@link #keepOutOfThePool}): a session
     * setting a borrower could not take off again is the same kind of thing, and the pool cannot
     * notice one by itself - {@link #isUsable} validates with {@code isValid()}, a liveness check a
     * connection carrying a stale setting passes.
     */
    private final boolean poolable;
    private volatile boolean poolable;

    /**
     * When this connection last answered the database, as a {@link System#nanoTime()} reading:
@@ -1244,6 +1249,17 @@
        this.lastKnownAliveNanos = System.nanoTime();
    }

    /**
     * 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
     * does (#885) - where the blast radius is then this one connection instead of every borrow it
     * would have served after this one.
     */
    void keepOutOfThePool() {
        poolable = false;
    }

    /** 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

@@ -346,7 +346,7 @@
         * {@link JDBCStorage#MAX_BOUND_SECONDS} is taken down to it, for the reason recorded there.
         */
        int seconds() {
            return clampSeconds(Integer.getInteger(property, defaultSeconds));
            return boundSeconds(property, defaultSeconds);
        }
    }

@@ -543,6 +543,20 @@
    static final long CLOCK_SLACK_MILLIS = 250;

    /**
     * How far past its own value the lock bound of a DDL may still be what ended a wait. A statement
     * pays a round trip before its wait begins, an engine keeps that timer in whole seconds, and the
     * drop loop of {@code removeStorageFiles()} sets the bound once around a loop whose earlier drops
     * are work of their own.
     * <p>
     * Past it a failure is left exactly as the engine reported it, because more than one wait of an
     * engine reports the same number: mysql reports the row lock of {@code innodb_lock_wait_timeout} -
     * 50 s by default, and what a create index under {@code ALGORITHM=COPY} waits on - as the same
     * ERROR 1205 as a metadata lock, and naming this backend's property for one of those would send an
     * operator to raise the single setting that cannot help.
     */
    static final long LOCK_BOUND_SLACK_MILLIS = 2000;

    /**
     * Ceiling of every bound this backend arms, in seconds - 24.9 days, which is what a socket read
     * timeout can hold at all: {@code setNetworkTimeout} takes milliseconds of an {@code int}, and a
     * bound past this one has no value of that layer to be given. It is <em>not</em> what keeps the
@@ -564,6 +578,19 @@
    }

    /**
     * How every bound of this backend reads the property configuring it, in seconds: 0, or a negative
     * value, leaves what it bounds as unbounded as it was before that bound existed, a value that is
     * not a number is ignored in favour of the default - {@code Integer.getInteger()} falls back to it
     * rather than reading such a value as a zero - and a value above {@link #MAX_BOUND_SECONDS} is
     * taken down to it. One reader rather than one per bound, so that a later change to how these are
     * read - an env fallback, a warning on a value that is not a number, a different clamp - cannot
     * leave one of them behaving unlike the rest.
     */
    static int boundSeconds(String property, int defaultSeconds) {
        return clampSeconds(Integer.getInteger(property, defaultSeconds));
    }

    /**
     * What {@link #timedOut} calls the second layer when that layer is the only one a statement ran
     * under, so that a test can tell the two apart in a message: a run where the first layer stopped
     * working degrades to this one by design, silently, and a suite that only measures how long a
@@ -598,6 +625,16 @@
    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 socket read timeout of one connection, and the statements running on it. This second
@@ -1240,6 +1277,44 @@
    // dialect is told to give up after this many seconds instead of waiting.
    private static final int COMMENT_LOCK_TIMEOUT_SECONDS=5;

    /**
     * The bound on the wait of a DDL of this backend for a lock another session holds, in seconds. A
     * value of {@code 0}, or a negative one, leaves it waiting for as long as the engine lets it,
     * which is what this backend did before this bound existed (#885).
     * <p>
     * A bound of the statement is the wrong tool for this, which is why the DDL of this backend is
     * {@link StatementBound#BULK} and stays there: a query timeout cannot tell a statement that is
     * <em>working</em> - a create index of a populated table - from one that is <em>queued</em> behind
     * an unrelated transaction of another session, and only the second one is worth ending. Three
     * engines out of four wait for a lock essentially forever ({@code lock_wait_timeout} is a year on
     * mysql, {@code LOCK_TIMEOUT} is -1 on sql server, {@code lock_timeout} is 0 on postgres), so the
     * open of a backend - and {@code dsconfig create-backend-index} on a running server - could hang
     * behind a session that has nothing to do with it; on postgres a queued {@code CREATE INDEX} parks
     * every writer of that table behind its own lock request while it waits.
     * <p>
     * The default is {@link #COMMENT_LOCK_TIMEOUT_SECONDS}: the stamp of the same open takes its lock
     * on the very tables this DDL creates and drops, and has been bounded there since #866.
     * <p>
     * Oracle is not one of the engines this is put on, and setting it there changes nothing: that
     * engine keeps its own {@code ddl_lock_timeout}, which gives up at once by default - tighter than
     * anything this would set - and which this property neither reads nor changes. An ORA-00054 out of
     * {@code dsconfig create-backend-index} is answered by {@code alter system set ddl_lock_timeout},
     * not by this.
     * <p>
     * What it costs is the round trips of the statements around each DDL - three on mysql and sql
     * server (reading the value back, setting the bound, giving the value back), two on postgres (the
     * savepoint a failed setting is taken back to, and the setting), none on oracle - and only on the
     * cold path: an existing backend issues no DDL at all, since every statement of {@code openTree()}
     * is guarded by a catalog read. A session already giving up sooner than this bound pays none of
     * them past the readback: it keeps what it has.
     */
    static final String DDL_LOCK_TIMEOUT_PROPERTY="org.openidentityplatform.opendj.jdbc.ddl.lock.timeout";

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

    // The comment statement runs on a connection of its own (newStampConnection() below), and a
    // driver waits for a connect attempt without limit unless it is told otherwise: a database
    // that keeps its established connections alive but accepts no new ones (a moved vip, a proxy
@@ -1282,16 +1357,102 @@
        /** postgresql: lock_timeout takes milliseconds; connectTimeout bounds socket.connect(), loginTimeout the whole login the driver runs on a thread of its own, socketTimeout every read after it - all three in seconds. */
        POSTGRES("set lock_timeout = "+(COMMENT_LOCK_TIMEOUT_SECONDS*1000),
            "connectTimeout", STAMP_CONNECT_TIMEOUT_SECONDS, "loginTimeout", STAMP_CONNECT_TIMEOUT_SECONDS,
            "socketTimeout", STAMP_READ_TIMEOUT_SECONDS),
            "socketTimeout", STAMP_READ_TIMEOUT_SECONDS) {
            // milliseconds, and "set local" rather than the plain SET of the stamp connection: it belongs
            // to the transaction running the DDL and is discarded by the commit that ends it, so nothing is
            // left behind on a pooled connection and nothing has to be put back. What the session carries is
            // not read here, which makes this the one engine where the bound can be looser than a
            // lock_timeout a deployment set for itself: reading it back is the round trip a set local exists
            // to save, and what is loosened is loosened for the length of this transaction and no longer.
            @Override
            String ddlLockBoundSql(int seconds, Long previous) {
                return "set local lock_timeout = "+seconds*1000L;
            }

            @Override
            String ddlLockBoundQuery() {
                return null; // set local: the commit that ends the DDL discards it
            }

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

            @Override
            boolean boundLivesInTheTransaction() {
                return true;
            }
        },
        /** 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,
            "connectTimeout", STAMP_CONNECT_TIMEOUT_SECONDS*1000, "socketTimeout", STAMP_READ_TIMEOUT_SECONDS*1000),
            "connectTimeout", STAMP_CONNECT_TIMEOUT_SECONDS*1000, "socketTimeout", STAMP_READ_TIMEOUT_SECONDS*1000) {
            // seconds, and this is the metadata lock a DDL waits for - never innodb_lock_wait_timeout, which
            // is the row lock write() replays a conflict of and which is bounded at 50 s already. A session
            // that gives up sooner than this keeps exactly what it has: a deployment that set
            // lock_wait_timeout tighter did so on purpose, which is the argument that leaves oracle alone
            // below, and this value has no encoding for "wait forever" to mistake for a tight one - its range
            // starts at 1 and its default is a year.
            @Override
            String ddlLockBoundSql(int seconds, Long previous) {
                return (previous!=null && previous<=seconds) ? null : "set session lock_wait_timeout="+seconds;
            }

            @Override
            String ddlLockBoundQuery() {
                return "select @@session.lock_wait_timeout";
            }

            @Override
            String ddlLockRestoreSql(long previous) {
                return "set session 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,
            "oracle.net.CONNECT_TIMEOUT", STAMP_CONNECT_TIMEOUT_SECONDS*1000, "oracle.jdbc.ReadTimeout", STAMP_READ_TIMEOUT_SECONDS*1000),
            "oracle.net.CONNECT_TIMEOUT", STAMP_CONNECT_TIMEOUT_SECONDS*1000, "oracle.jdbc.ReadTimeout", STAMP_READ_TIMEOUT_SECONDS*1000) {
            // oracle gives up on a ddl lock at once - ddl_lock_timeout is 0 - which is tighter than anything
            // set here, so ours would only loosen it; and a deployment that raised it globally did so on
            // purpose. Putting it back would also mean reading v$parameter, a privilege the account of a
            // backend often does not have.
            @Override
            String ddlLockBoundSql(int seconds, Long previous) {
                return null;
            }

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

            @Override
            String ddlLockRestoreSql(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),
            "loginTimeout", STAMP_CONNECT_TIMEOUT_SECONDS, "socketTimeout", STAMP_READ_TIMEOUT_SECONDS*1000);
            "loginTimeout", STAMP_CONNECT_TIMEOUT_SECONDS, "socketTimeout", STAMP_READ_TIMEOUT_SECONDS*1000) {
            // milliseconds, and this setting bounds every lock wait of the session, row locks included,
            // which is why it is put back the moment the DDL is through. -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 exactly as the deployment set it.
            @Override
            String ddlLockBoundSql(int seconds, Long previous) {
                final long millis=seconds*1000L;
                return (previous!=null && previous>=0 && previous<=millis) ? null : "set lock_timeout "+millis;
            }

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

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

        final String lockTimeoutSql;
        // The driver properties bounding the login attempt of a stamp connection: the one bounding
@@ -1314,6 +1475,44 @@
            this(lockTimeoutSql, connectProperty, connectValue, readProperty, readValue);
            connectProperties.setProperty(loginProperty, String.valueOf(loginValue));
        }

        /**
         * The session setting bounding the wait of a DDL for its lock, or null where there is none to
         * put on: an engine left to a bound of its own - oracle - and a session that already gives up
         * sooner than this one would, which keeps what it has rather than being loosened to ours. Not
         * {@link #lockTimeoutSql}, which bounds the same wait on the stamp connection: that one is a
         * constant of a connection this backend owns for the length of a sweep, this one is configurable
         * and runs on a pooled connection somebody else borrows next.
         * <p>
         * Answered per constant rather than by a switch, and so are the two below: a constant added later
         * cannot compile without saying what it sets, and one that forgot to would otherwise take out
         * every DDL of this backend - {@link JDBCStorage#withDdlLockBound} asks these before any try,
         * on the path whose whole contract is that no statement of the bound is ever the failure of a
         * DDL.
         *
         * @param previous what the session carries now, as {@link #ddlLockBoundQuery()} read it, in the
         *                 unit that query answers in - or null where nothing was read back
         */
        abstract String ddlLockBoundSql(int seconds, Long previous);

        /**
         * What the session carries now, asked before the bound above displaces it - or null where that
         * bound undoes itself. A pooled connection outlives the transaction that borrowed it, and
         * {@code CachedConnection.close()} only rolls back.
         */
        abstract String ddlLockBoundQuery();

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

        /**
         * 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.
         */
        boolean boundLivesInTheTransaction() {
            return false;
        }
    }

    /** Returns the class name of the driver behind the given connection, which names the engine it talks to. */
@@ -1919,13 +2118,349 @@
        }
    }

    /**
     * Runs a DDL of this backend under {@link #DDL_LOCK_TIMEOUT_PROPERTY}, and gives the session back
     * whatever it carried before.
     * <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 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.
     * <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.
     */
    <T> T withDdlLockBound(Connection con, Dialect dialect, Execution<T> action) throws SQLException {
        final int seconds=ddlLockBoundSeconds();
        // Asked first with nothing displaced yet, which is what tells an engine this bound is never put
        // on - oracle, and one none of these settings fit - 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==null || seconds<=0 || 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();
        try {
            return action.run();
        }catch (SQLException e) {
            throw gaveUpOnTheLock(e, dialect, seconds, startedAt);
        }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);
        }finally {
            if (restore!=null) {
                restoreDdlLockBound(con, dialect, bound, restore);
            }
        }
    }

    /**
     * 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
     * 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) {
        try {
            if (!con.getAutoCommit()) {
                return true;
            }
            reportTheWaitIsLeftUnbounded(dialect, 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);
        }
        return false;
    }

    /**
     * The point a setting that failed is taken back to, or null where there is none to take: an engine
     * whose bound is a session setting rather than a transaction one, and a connection that would not
     * 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) {
        if (!dialect.boundLivesInTheTransaction()) {
            return null;
        }
        try {
            return boundedSessionCall(con, con::setSavepoint);
        }catch (SQLException | RuntimeException e) {
            reportTheWaitIsLeftUnbounded(dialect, bound, e);
            return null;
        }
    }

    /**
     * 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
     * back is one whose DDL is about to say so, and the failure worth reporting is the one the caller
     * has just reported.
     */
    private void rollbackTheBound(Connection con, Dialect dialect, Savepoint beforeTheBound) {
        if (beforeTheBound==null) {
            return;
        }
        try {
            boundedSessionCall(con, () -> {
                con.rollback(beforeTheBound);
                return null;
            });
        }catch (SQLException | RuntimeException e) {
            logger.trace(LocalizableMessage.raw("jdbc: the transaction of a DDL could not be taken back to the point"
                + " before its lock bound on this %s database: %s", dialect, stackTraceToSingleLineString(e)));
        }
    }

    // The round trips of the bound itself - the readback, the savepoint, the setting, the value given
    // back - are bounded at this, in seconds. None of them takes a lock or reads a table, so a wait of
    // one of them is a database that has stopped answering rather than work in progress, and the socket
    // read timeout behind it arrives a margin later still (BACKSTOP_MARGIN_SECONDS).
    static final int SESSION_STATEMENT_BOUND_SECONDS=10;

    /**
     * Runs one round trip of the bound itself under the socket read timeout backing it up. Without one
     * a readback on a connection whose peer went quiet - a failed-over primary, a proxy that stops
     * answering with the socket still open - parks the thread that is opening a backend for good, which
     * is the hang #877 and #882 exist to end; and it parks it from the {@code finally} giving a pooled
     * connection its value back, where the DDL has already failed. Deliberately not the class the DDL
     * itself carries: a create index of a populated table legitimately runs for hours, a
     * {@code set lock_timeout} never does, so a bound of its own costs the DDL nothing.
     * <p>
     * The cancel layer is left off, for the reason {@link #executeSessionStatement} records: what these
     * carry has to reach the server as a plain batch. This is the layer that ends a wait no cancel
     * would reach anyway.
     */
    private <T> T boundedSessionCall(Connection con, Execution<T> call) throws SQLException {
        final Backstop backstop=holdBackstop(con, SESSION_STATEMENT_BOUND_SECONDS);
        try {
            return call.run();
        }finally {
            releaseBackstop(backstop, con, SESSION_STATEMENT_BOUND_SECONDS);
        }
    }

    /**
     * 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.
     */
    private Long sessionValue(Connection con, Dialect dialect, String query) {
        if (logger.isTraceEnabled()) {
            logger.trace(LocalizableMessage.raw("jdbc: %s",query));
        }
        try {
            return boundedSessionCall(con, () -> {
                try (final Statement statement=con.createStatement(); final ResultSet rows=statement.executeQuery(query)) {
                    if (!rows.next()) {
                        throw new SQLException("the session answered no row");
                    }
                    return Long.valueOf(rows.getString(1).trim());
                }
            });
        }catch (SQLException | RuntimeException e) { // a value that is not a number arrives unchecked
            reportTheWaitIsLeftUnbounded(dialect, 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.
     * <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
     * 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.
     */
    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"
                + " 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)));
        }
    }

    /**
     * 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.
     * <p>
     * A connection this failed on does not go back into the pool. 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
     * 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
     * time: the property can have been changed since, and what a session is left carrying is what was
     * 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) {
        try {
            boundedSessionCall(con, () -> {
                executeSessionStatement(con, restore);
                return null;
            });
        }catch (SQLException | RuntimeException e) {
            if (con instanceof CachedConnection) {
                ((CachedConnection) con).keepOutOfThePool();
            }
            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"
                    + " of this %s database, which may have been left carrying \"%s\" instead of the value it had:"
                    + " that connection is closed rather than pooled, so no borrow after this one gives up on a lock"
                    + " at a bound of %s it never asked for (%s)", dialect, bound, DDL_LOCK_TIMEOUT_PROPERTY,
                    stackTraceToSingleLineString(e)));
            }
        }
    }

    /**
     * A DDL that gave up at the bound, reported as what it is. It arrives as a bare 55P03 /
     * ERROR 1205 / error 1222, naming neither the wait it ended nor the property that ended it - the
     * gap {@link #timedOut} closes for the bound of a statement. The state and the vendor number are
     * carried over and the failure itself chained, so a caller that classifies this reads exactly what
     * it read before: a mysql lock wait stays the class 40 conflict {@link #write} knows.
     * <p>
     * Only a failure this bound could still be what ended is renamed, measured on the monotonic clock
     * the way {@link #timedOut} measures its own and allowed {@link #LOCK_BOUND_SLACK_MILLIS} past the
     * bound: an engine reports more than one wait with the same number, and a wait that ran far longer
     * than this bound was ended by something else - on mysql, by the {@code innodb_lock_wait_timeout}
     * that reports the row lock of a create index under {@code ALGORITHM=COPY} as the same ERROR 1205.
     * Past that the failure is left exactly as it arrived, which is what it was before this bound
     * existed. There is no guard under the bound to go with it: the states matched here are what an
     * engine says when a lock wait ran out and nothing else says them, so an early one does not arise -
     * and adding one would cost every case of the suite the wait it exists to avoid.
     * <p>
     * The time is reported as measured rather than as the bound, for the reason {@link #timedOut}
     * records: an operator has to be able to put the message next to a clock.
     */
    SQLException gaveUpOnTheLock(SQLException e, Dialect dialect, int seconds, long startedAt) {
        if (!lockNotAvailable(e, dialect)) {
            return e;
        }
        final long elapsedMillis=(nanoTime()-startedAt)/1000000L;
        if (elapsedMillis > seconds*1000L+LOCK_BOUND_SLACK_MILLIS) {
            return e;
        }
        return new SQLTimeoutException("jdbc: the statement gave up waiting for a lock another session holds after "
            +elapsedMillis+" ms, at the "+seconds+"s of "+DDL_LOCK_TIMEOUT_PROPERTY+": raise that property, or set"
            + " it to 0 to wait for the lock as this backend did before it was bounded", e.getSQLState(),
            e.getErrorCode(), e);
    }

    /**
     * The same rename where the failure arrives unchecked, which is how a statement of the action that
     * is not the DDL itself answers: {@link #isExistsTable}, asked once per row by the drop loop of a
     * clear, gives back a {@link StorageRuntimeException} holding what the engine said. The chain is
     * read for the engine's own way of saying the lock was not available, and that link is put through
     * the rename above - so the same wait is named the same way whichever statement of the action was
     * the one waiting.
     * <p>
     * A failure the rename does not apply to is given back exactly as it arrived, keeping its class and
     * its stack. One it does apply to is wrapped again, in the class every unchecked failure of this
     * storage carries and the class {@link #removeStorageFiles()} reads to decide what it rethrows.
     */
    RuntimeException gaveUpOnTheLock(RuntimeException e, Dialect dialect, int seconds, long startedAt) {
        final SQLException link=firstLinkMatching(e, WITHOUT_THE_RELEASE, EVERY_LINK,
            failure -> isLockTimeout(failure, dialect));
        if (link==null) {
            return e;
        }
        final SQLException renamed=gaveUpOnTheLock(link, dialect, seconds, startedAt);
        return (renamed==link) ? e : new StorageRuntimeException(renamed);
    }

    /**
     * Whether any link of a failure is this engine's own way of saying the lock was not available.
     * Asked of the engine's number alone rather than through {@link #failureScope}, which reads a
     * {@link SQLTimeoutException} as a moment of its own as well: a statement the bound of its class
     * cancelled is one of those, and it was ended by a bound {@link #timedOut} has already named.
     * <p>
     * {@link #WITHOUT_THE_RELEASE}, unlike {@link #failureScope}: this asks what the engine did with
     * this statement, and the release of the connection - whose rollback reports what it saw as a
     * suppressed exception - runs after that outcome was decided and cannot speak for it. Read the
     * other way, a 55P03 or a 1205 out of the rollback that gave the connection back would rename a DDL
     * that failed for something else entirely.
     */
    static boolean lockNotAvailable(Throwable failure, Dialect dialect) {
        return firstLinkMatching(failure, WITHOUT_THE_RELEASE, EVERY_LINK, e -> isLockTimeout(e, dialect))!=null;
    }

    // A session setting must reach the server as a plain batch: the sql server driver runs a
    // prepared statement through sp_executesql, and a setting made there is reverted when that
    // call returns - before the statement it is meant to protect ever runs.
    //
    // Outside both layers of the bound, like the comment statement executeAny() runs, and for the
    // same reason: this is issued from newStampConnection() on a stamp connection, whose connect
    // properties carry a socket read timeout of their own (Dialect.connectProperties).
    // Outside the cancel layer of the bound, like the comment statement executeAny() runs: a setting
    // the DDL is wrapped in must not be cut short by a bound the DDL itself does not have, and a cancel
    // would have to reach a statement this one deliberately does not prepare. The second layer is not
    // left off with it. From newStampConnection() that is a stamp connection, whose connect properties
    // carry a socket read timeout of their own (Dialect.connectProperties); from withDdlLockBound()
    // boundedSessionCall() arms one, since a session setting that answers no round trip is a database
    // that has stopped answering rather than work in progress.
    private void executeSessionStatement(Connection con, String sql) throws SQLException {
        try (final Statement statement=con.createStatement()) {
            if (logger.isTraceEnabled()) {
@@ -2084,20 +2619,27 @@
        if (e instanceof SQLTimeoutException || e instanceof SQLTransientException) {
            return FailureScope.MOMENT;
        }
        if (dialect==null) { // the failure came before the engine was known
            return FailureScope.TREE;
        return isLockTimeout(e, dialect) ? FailureScope.MOMENT : FailureScope.TREE;
    }

    // What one engine reports when a statement gave up on a lock instead of getting it. A dialect with
    // no number of its own here - and a failure that came before the engine was known at all - says
    // nothing of the kind, and is not treated as a moment.
    static boolean isLockTimeout(SQLException e, Dialect dialect) {
        if (dialect==null) {
            return false;
        }
        switch (dialect) {
        case POSTGRES: // 55P03 lock not available: lock_timeout expired
            return "55P03".equals(sqlState) ? FailureScope.MOMENT : FailureScope.TREE;
            return "55P03".equals(e.getSQLState());
        case MYSQL: // 1205 lock wait timeout exceeded
            return e.getErrorCode()==1205 ? FailureScope.MOMENT : FailureScope.TREE;
            return e.getErrorCode()==1205;
        case ORACLE: // ORA-00054 resource busy, ORA-04021 timeout occurred while waiting to lock object
            return e.getErrorCode()==54 || e.getErrorCode()==4021 ? FailureScope.MOMENT : FailureScope.TREE;
            return e.getErrorCode()==54 || e.getErrorCode()==4021;
        case MICROSOFT: // 1222 lock request time out period exceeded
            return e.getErrorCode()==1222 ? FailureScope.MOMENT : FailureScope.TREE;
        default: // a dialect with no lock timeout code of its own here: its failures are not treated as ones of the moment
            return FailureScope.TREE;
            return e.getErrorCode()==1222;
        default:
            return false;
        }
    }

@@ -2304,39 +2846,9 @@
            // owner of and which a clear must therefore leave exactly where they lie (#881)
            final List<String> skippedRows=new ArrayList<>(); // rows the read could not act on: reported below
            final Map<TreeName,String> trees=catalogTables(con, scope, skippedRows);
            final TreeName catalogTree=getCatalogTree();
            int dropped=0;
            // the same count with the catalog itself left out, which is what says whether this clear
            // removed anything of the backend: a catalog table standing over rows that name nothing -
            // a backup restored older than the tables it was taken beside - is dropped like any other
            // and would otherwise make a clear that removed no tree at all look like a clear that did
            // something. See reportClearOutcome()
            int droppedTrees=0;
            // counted without the catalog, for the reason droppedTrees is kept apart from dropped: the
            // catalog is walked by this loop like any other table, so a catalog table that went between
            // the lookup of catalogTables() and the loop's own would otherwise be summed up as a tree of
            // this backend that had lost its table
            int missingTrees=0;
            final ClearCounts counts;
            try {
                for (final Map.Entry<TreeName,String> tree : trees.entrySet()) {
                    final String tableName=tree.getValue();
                    final boolean isCatalog=catalogTree.equals(tree.getKey());
                    if (!isExistsTable(con, scope, tableName)) { // a row of the catalog outliving its table
                        reportClearLine(LocalizableMessage.raw(
                            "jdbc: backend %s names tree %s, whose table %s is not there: nothing to drop for it",
                            config.getBackendId(), tree.getKey(), tableName));
                        if (!isCatalog) {
                            missingTrees++;
                        }
                        continue;
                    }
                    dropTable(con, tableName);
                    dropped++;
                    if (!isCatalog) {
                        droppedTrees++;
                    }
                }
                con.commit();
                counts=dropCatalogTables(con, scope, trees);
            } catch (Exception e) {
                // every failure of the loop and not the SQLException alone: the lookup deciding each
                // drop answers with a StorageRuntimeException of its own, and a drop left pending by one
@@ -2353,7 +2865,7 @@
                unstampableTrees.remove(treeName);
            }
            try {
                reportClearOutcome(con, scope, dropped, droppedTrees, missingTrees, skippedRows);
                reportClearOutcome(con, scope, counts.dropped, counts.droppedTrees, counts.missingTrees, skippedRows);
            } catch (RuntimeException e) {
                // the clear itself is done and committed: an account of what it left standing must not be
                // the thing that reports it as failed, and a caller retrying it would find nothing to drop
@@ -2377,9 +2889,72 @@
        }
    }

    /** What the drop loop of a clear did, which {@link #reportClearOutcome} accounts for. */
    static final class ClearCounts {
        /** Tables dropped, the catalog of the backend among them. */
        int dropped;
        /**
         * The same count with the catalog itself left out, which is what says whether this clear
         * removed anything of the backend: a catalog table standing over rows that name nothing - a
         * backup restored older than the tables it was taken beside - is dropped like any other and
         * would otherwise make a clear that removed no tree at all look like a clear that did something.
         */
        int droppedTrees;
        /**
         * Rows whose table was not there, counted without the catalog for the reason
         * {@link #droppedTrees} is kept apart from {@link #dropped}: the catalog is walked by the loop
         * like any other table, so a catalog table that went between the lookup of
         * {@code catalogTables()} and the loop's own would otherwise be summed up as a tree of this
         * backend that had lost its table.
         */
        int missingTrees;
    }

    /**
     * Drops the tables the catalog of this backend names, in one transaction and under a single lock
     * bound, and says what it did. This loop bypasses the {@code commitStatement()} every other DDL of
     * this backend goes through and commits once at the end, so the bound is set once around the whole
     * of it rather than once per table - on postgres one {@code set local} covers every drop of the
     * single transaction they run in.
     * <p>
     * A clear with no row to act on is committed without the bound: putting it on costs a readback and
     * a restore of its own, and the case {@code CLEAR_DROPPED_NOTHING} describes - the first clear of a
     * backend upgraded from a version that kept no catalog - has no DDL for them to bound.
     */
    ClearCounts dropCatalogTables(Connection con, TableScope scope, Map<TreeName,String> trees) throws SQLException {
        if (trees.isEmpty()) {
            con.commit();
            return new ClearCounts();
        }
        final TreeName catalogTree=getCatalogTree();
        return withDdlLockBound(con, dialectOf(con), () -> {
            final ClearCounts counts=new ClearCounts();
            for (final Map.Entry<TreeName,String> tree : trees.entrySet()) {
                final String tableName=tree.getValue();
                final boolean isCatalog=catalogTree.equals(tree.getKey());
                if (!isExistsTable(con, scope, tableName)) { // a row of the catalog outliving its table
                    reportClearLine(LocalizableMessage.raw(
                        "jdbc: backend %s names tree %s, whose table %s is not there: nothing to drop for it",
                        config.getBackendId(), tree.getKey(), tableName));
                    if (!isCatalog) {
                        counts.missingTrees++;
                    }
                    continue;
                }
                dropTable(con, tableName);
                counts.dropped++;
                if (!isCatalog) {
                    counts.droppedTrees++;
                }
            }
            con.commit();
            return counts;
        });
    }

    /**
     * Drops one table of a clear. It is a method of its own so that the order {@link
     * #removeStorageFiles()} drops in can be watched from a test: what names the trees has to outlive
     * #dropCatalogTables} drops in can be watched from a test: what names the trees has to outlive
     * them, and that guarantee is the loop's - it holds because the loop walks the catalog's map in
     * the order that map was built in, and a test asserting on the map instead would go on passing
     * over a loop that had stopped doing so.
@@ -3874,10 +4449,21 @@
            // one of the ones #877 names as overridden downwards - the create table and the three create
            // index of openTree(), the delete from of clearTree() and the drop table of deleteTree() -
            // and nobody is waiting on any of them.
            try (final PreparedStatement statement=con.prepareStatement(sql)) {
                execute(statement, StatementBound.BULK);
                partlyCommitted=true; // a commit that fails leaves the outcome unknown, which is no more replayable
                con.commit();
            final Execution<Void> issue=() -> {
                try (final PreparedStatement statement=con.prepareStatement(sql)) {
                    execute(statement, StatementBound.BULK);
                    partlyCommitted=true; // a commit that fails leaves the outcome unknown, which is no more replayable
                    con.commit();
                }
                return null;
            };
            if (ddl) {
                // 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);
            }else {
                issue.run();
            }
        }


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

New file
@@ -0,0 +1,841 @@
/*
 * 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.Dialect;
import org.opends.server.backends.jdbc.JDBCStorage.StatementBound;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Importer;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.opends.server.backends.pluggable.spi.StorageStatus;
import org.opends.server.backends.pluggable.spi.TreeName;
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.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.sql.Savepoint;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

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.any;
import static org.mockito.Mockito.anyInt;
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.assertSame;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

/**
 * What a DDL of this backend is told to do about a lock another session holds (#885). It needs no
 * database: the connection is a mock, so what each engine is told - and what it is told to put back
 * afterwards - is pinned wherever the build runs, while the container suites cover a drop really
 * queued behind another session.
 */
@SuppressWarnings("javadoc")
@Test(groups = { "precommit", "jdbc" }, sequential = true)
public class JDBCDdlLockBoundTestCase extends DirectoryServerTestCase {

    /** The tree the cases below name; the table behind it is a hash of that name. */
    private static final TreeName TREE = new TreeName("dc=example,dc=com", "id2entry");
    /** A second one, for the loop that drops every table of a backend under a single bound. */
    private static final TreeName OTHER_TREE = new TreeName("dc=example,dc=com", "dn2id");

    /** The DDL itself, as it appears among the session statements issued around it. */
    private static final String THE_DDL = "the ddl";

    /** The backend the storage of a case is configured as, which is what names its tree catalog. */
    private static final String BACKEND_ID = "ddlLockBound";
    /** That catalog's table, which the connection of a case answers as not being there: see engine(). */
    private static final String NO_CATALOG_TABLE =
        JDBCStorage.toTableName(new TreeName(JDBCStorage.CATALOG_BASE_DN, BACKEND_ID));

    /** 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();
    }

    /** A configuration naming this backend, which is all any case here reads off one. */
    private static JDBCBackendCfg backendCfg() {
        final JDBCBackendCfg cfg = mockCfg(JDBCBackendCfg.class);
        when(cfg.getBackendId()).thenReturn(BACKEND_ID);
        return cfg;
    }

    @AfterMethod
    public void clearProperties() {
        System.clearProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY);
    }

    /**
     * The stamp of the same open is already bounded at five seconds
     * ({@code COMMENT_LOCK_TIMEOUT_SECONDS}), and it takes its lock on the very tables this DDL
     * creates and drops.
     */
    @Test
    public void testTheDefaultGivesUpOnALockAfterFiveSeconds() {
        assertEquals(JDBCStorage.ddlLockBoundSeconds(), 5);
    }

    /** Zero is "wait as this backend waited before this bound existed", and so is anything under it. */
    @Test
    public void testAValueOfZeroOrLessLeavesTheWaitUnbounded() {
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "0");
        assertEquals(JDBCStorage.ddlLockBoundSeconds(), 0);
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "-1");
        assertEquals(JDBCStorage.ddlLockBoundSeconds(), 0);
    }

    /**
     * A value that is not a number leaves the default in force rather than reading as a zero, which
     * is what {@code Integer.getInteger()} does with one - so a typo does not silently take the bound
     * off.
     */
    @Test
    public void testAValueThatIsNotANumberKeepsTheDefault() {
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "five seconds");
        assertEquals(JDBCStorage.ddlLockBoundSeconds(), 5);
    }

    /** The ceiling every bound of this backend is taken down to, for the reason recorded there. */
    @Test
    public void testAValuePastTheCeilingIsTakenDownToIt() {
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, String.valueOf(Integer.MAX_VALUE));
        assertEquals(JDBCStorage.ddlLockBoundSeconds(), JDBCStorage.MAX_BOUND_SECONDS);
    }

    @DataProvider
    public Object[][] engines() {
        return new Object[][] {
            // postgres takes milliseconds, and "set local" is discarded by the commit that ends the
            // DDL: there is nothing to read beforehand and nothing to put back afterwards
            { "postgres", Dialect.POSTGRES, "0", asList("set local lock_timeout = 5000", THE_DDL) },
            // mysql takes seconds, and the setting outlives the transaction on a pooled connection:
            // what the session carried (a year, by default) is read first and put back after
            { "mysql", Dialect.MYSQL, "31536000", asList("select @@session.lock_wait_timeout",
                "set session lock_wait_timeout=5", THE_DDL, "set session lock_wait_timeout=31536000") },
            // sql server takes milliseconds, and its setting bounds every lock wait of the session -
            // -1, wait forever, is what a session of it carries until something says otherwise
            { "sql server", Dialect.MICROSOFT, "-1", asList("select @@lock_timeout",
                "set lock_timeout 5000", THE_DDL, "set lock_timeout -1") },
            // oracle is left to its own ddl_lock_timeout, which is tighter than anything set here
            { "oracle", Dialect.ORACLE, "0", singletonList(THE_DDL) },
        };
    }

    /** What each engine is told around a DDL of this backend, in the order it is told it. */
    @Test(dataProvider = "engines")
    public void testWhatEachEngineIsToldAroundADdl(String name, Dialect dialect, String carries,
            List<String> expected) throws Exception {
        storage.withDdlLockBound(recording(mock(Connection.class), carries), dialect, theDdl());
        assertEquals(issued, expected, name);
    }

    /**
     * Oracle gives up on a DDL lock at once ({@code ddl_lock_timeout} is 0), so a bound of ours would
     * only loosen it - and a deployment that raised it globally did so on purpose. Putting it back
     * would mean reading {@code v$parameter}, which the account of a backend often may not.
     */
    @Test
    public void testOracleIsLeftToItsOwnDdlLockTimeout() {
        assertNull(Dialect.ORACLE.ddlLockBoundSql(5, null));
    }

    /** An engine none of these statements fit is fed none of them, as its statistics are left alone. */
    @Test
    public void testAnEngineThisBackendDoesNotKnowIsLeftAlone() throws Exception {
        storage.withDdlLockBound(recording(mock(Connection.class), "0"), null, theDdl());
        assertEquals(issued, singletonList(THE_DDL));
    }

    /** Turning the bound off costs no round trip either: the DDL waits exactly as it did before. */
    @Test
    public void testAnUnboundedWaitIssuesNoSessionStatement() throws Exception {
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "0");
        storage.withDdlLockBound(recording(mock(Connection.class), "31536000"), Dialect.MYSQL, theDdl());
        assertEquals(issued, singletonList(THE_DDL));
    }

    /**
     * The setting is put back even when the DDL fails. It is a pooled connection, and
     * {@code CachedConnection.close()} only rolls back: on sql server the setting left behind would
     * bound every lock wait of whoever borrows the connection next, row locks included, and
     * {@code write()} replays no conflict of those.
     */
    @Test
    public void testTheBoundIsPutBackWhenTheDdlFails() throws Exception {
        final SQLException rejected = new SQLException("table already exists", "42S01");
        try {
            storage.withDdlLockBound(recording(mock(Connection.class), "31536000"), Dialect.MYSQL, () -> {
                issued.add(THE_DDL);
                throw rejected;
            });
            fail("the failure of the ddl was swallowed");
        }catch (SQLException e) {
            assertSame(e, rejected, "the failure of the ddl was replaced");
        }
        assertEquals(issued, asList("select @@session.lock_wait_timeout", "set session lock_wait_timeout=5",
            THE_DDL, "set session lock_wait_timeout=31536000"));
    }

    /**
     * A setting can reach the server and still fail on the close() of the statement that carried it -
     * a connection that broke in between - and no driver tells that apart from a setting that never
     * arrived. The session has it either way, so it is taken off again once the DDL is through: this
     * connection goes back to a pool, and on sql server a lock_timeout left behind ends every lock wait
     * of whoever borrows it next, row locks included.
     */
    @Test
    public void testASettingThatBrokeOnTheCloseOfItsStatementIsStillTakenOff() throws Exception {
        final Connection con = breakingOnTheCloseOfASetting(mock(Connection.class), "31536000");

        storage.withDdlLockBound(con, Dialect.MYSQL, theDdl());

        assertEquals(issued, asList("select @@session.lock_wait_timeout", "set session lock_wait_timeout=5",
            THE_DDL, "set session lock_wait_timeout=31536000"));
    }

    /**
     * And the DDL it wrapped is reported the way a bounded one is. The setting reached the server, so
     * the wait really was bounded - reporting that failure as the bare 55P03 it arrives as is the gap
     * this bound exists to close, and the log line above it says only that the bound may not be there.
     */
    @Test
    public void testADdlBoundedByASettingWhoseCloseFailedStillNamesTheProperty() throws Exception {
        final SQLException lockWait = new SQLException("Lock wait timeout exceeded", "40001", 1205);
        try {
            storage.withDdlLockBound(breakingOnTheCloseOfASetting(mock(Connection.class), "31536000"),
                Dialect.MYSQL, () -> {
                    throw lockWait;
                });
            fail("the lock timeout was swallowed");
        }catch (SQLException e) {
            assertTrue(e.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY), e.getMessage());
            assertSame(e.getCause(), lockWait, "the failure of the engine was not chained");
        }
    }

    /**
     * A value that could not be given back does not fail a DDL that went through. The restore runs
     * from a finally while the caller may be being unwound, where a throw takes the place of whatever
     * brought it there (JLS 14.20.2) - here a create table the engine accepted.
     */
    @Test
    public void testAValueThatCouldNotBeGivenBackDoesNotFailADdlThatWentThrough() throws Exception {
        final Connection con = refusingToGiveTheValueBack(mock(Connection.class), "31536000");

        storage.withDdlLockBound(con, Dialect.MYSQL, theDdl());

        assertEquals(issued, asList("select @@session.lock_wait_timeout", "set session lock_wait_timeout=5",
            THE_DDL, "set session lock_wait_timeout=31536000"));
    }

    /** And it does not displace the failure of one that did not: that failure is what says what went wrong. */
    @Test
    public void testAValueThatCouldNotBeGivenBackDoesNotDisplaceTheFailureOfADdl() throws Exception {
        final SQLException rejected = new SQLException("table already exists", "42S01");
        try {
            storage.withDdlLockBound(refusingToGiveTheValueBack(mock(Connection.class), "31536000"),
                Dialect.MYSQL, () -> {
                    throw rejected;
                });
            fail("the failure of the ddl was swallowed");
        }catch (SQLException e) {
            assertSame(e, rejected, "the failure of the ddl was replaced by the one of the restore");
        }
    }

    /**
     * A bound with no way of putting back what it displaces is not set at all: a server whose session
     * does not have the variable - a mysql-compatible one behind connector/j - would otherwise be
     * given a bound this backend could never take off the pooled connection again.
     */
    @Test
    public void testABoundThatCannotBeReadBackIsNotSetAtAll() throws Exception {
        final Connection con = mock(Connection.class);
        final Statement statement = mock(Statement.class);
        when(statement.executeQuery(anyString()))
            .thenThrow(new SQLException("unknown system variable", "HY000", 1193));
        when(con.createStatement()).thenReturn(statement);

        storage.withDdlLockBound(con, Dialect.MYSQL, theDdl());

        assertEquals(issued, singletonList(THE_DDL));
        verify(statement, never()).execute(anyString());
    }

    /** The same where the session answers with something no setting of it would take back. */
    @Test
    public void testAValueTheSessionCouldNotBeGivenBackIsNotDisplaced() throws Exception {
        storage.withDdlLockBound(recording(mock(Connection.class), "unlimited"), Dialect.MYSQL, theDdl());
        assertEquals(issued, asList("select @@session.lock_wait_timeout", THE_DDL));
    }

    /**
     * A connection that refuses the setting outright leaves the DDL unbounded rather than failing it.
     * The bound is an improvement on a wait: a backend that opened before this bound existed has to
     * open still, and a session statement that fails on a connection whose DDL would have gone through
     * is not a reason to fail that DDL.
     */
    @Test
    public void testAConnectionThatRefusesTheSettingStillRunsTheDdl() throws Exception {
        final Connection con = mock(Connection.class);
        when(con.createStatement()).thenThrow(new SQLException("no session statement here", "42000"));

        storage.withDdlLockBound(con, Dialect.POSTGRES, theDdl());

        assertEquals(issued, singletonList(THE_DDL));
    }

    /**
     * A DDL that gave up at the bound arrives as a bare 55P03 / 1205 / 1222, naming neither the wait
     * it ended nor the property that ended it - the gap {@code timedOut()} closes for a statement
     * bound.
     */
    @Test
    public void testALockTheDdlGaveUpOnNamesTheProperty() throws Exception {
        final SQLException lockNotAvailable = new SQLException("canceling statement due to lock timeout", "55P03");
        try {
            storage.withDdlLockBound(recording(mock(Connection.class), "0"), Dialect.POSTGRES, () -> {
                throw lockNotAvailable;
            });
            fail("the lock timeout was swallowed");
        }catch (SQLException e) {
            assertTrue(e.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY), e.getMessage());
            assertSame(e.getCause(), lockNotAvailable, "the failure of the engine was not chained");
            assertEquals(e.getSQLState(), "55P03", "the state a caller classifies this by was dropped");
        }
    }

    /**
     * The state and the vendor number are carried over, so a caller classifying the failure reads
     * exactly what it read before this bound existed: a mysql lock wait arrives in class 40, and that
     * state alone is what {@code write()} replays a conflict on.
     */
    @Test
    public void testARewrittenMysqlLockWaitStaysTheConflictAWriteKnows() throws Exception {
        final SQLException lockWait = new SQLException("Lock wait timeout exceeded; try restarting transaction",
            "40001", 1205);
        try {
            storage.withDdlLockBound(recording(mock(Connection.class), "31536000"), Dialect.MYSQL, () -> {
                throw lockWait;
            });
            fail("the lock timeout was swallowed");
        }catch (SQLException e) {
            assertTrue(e.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY), e.getMessage());
            assertEquals(e.getErrorCode(), 1205, "the number a caller classifies this by was dropped");
            assertEquals(JDBCStorage.conflictVerdict(e, "com.mysql.cj.jdbc.ConnectionImpl").conflict,
                JDBCStorage.Conflict.AFTER_LOCK_WAIT,
                "a conflict write() replayed before this bound existed is no longer read as one");
        }
    }

    /**
     * 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.
     */
    @Test
    public void testARewrittenSqlServerLockWaitIsMadeNoMoreReplayable() throws Exception {
        final SQLException lockWait = new SQLException("Lock request time out period exceeded.", "S0001", 1222);
        try {
            storage.withDdlLockBound(recording(mock(Connection.class), "-1"), Dialect.MICROSOFT, () -> {
                throw lockWait;
            });
            fail("the lock timeout was swallowed");
        }catch (SQLException e) {
            assertTrue(e.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY), e.getMessage());
            assertEquals(e.getErrorCode(), 1222, "the number a caller classifies this by was dropped");
            assertEquals(JDBCStorage.conflictVerdict(e, "com.microsoft.sqlserver.jdbc.SQLServerConnection").conflict,
                JDBCStorage.Conflict.NONE,
                "a lock wait of a ddl was made a conflict write() would replay");
        }
    }

    /**
     * A failure of another kind is left exactly as it is: naming a property that had nothing to do
     * with it sends an operator to raise a value that changes nothing about what they saw.
     */
    @Test
    public void testAFailureThatIsNotALockIsLeftExactlyAsItIs() throws Exception {
        final SQLException denied = new SQLException("permission denied for schema public", "42501");
        try {
            storage.withDdlLockBound(recording(mock(Connection.class), "0"), Dialect.POSTGRES, () -> {
                throw denied;
            });
            fail("the failure was swallowed");
        }catch (SQLException e) {
            assertSame(e, denied, "a failure that is not a lock wait was reported as one");
        }
    }

    /**
     * And so is a statement the bound of its own class cancelled: {@code bulk.timeout} ends a create
     * index the engine was working on, which is not a lock this DDL was queued for, and
     * {@code timedOut()} has already named the property that ended it.
     */
    @Test
    public void testAStatementItsOwnBoundCancelledIsNotReportedAsALockWait() throws Exception {
        final SQLException cancelled = new SQLTimeoutException("jdbc: the statement took 100200 ms, reaching the"
            + " 100s of " + StatementBound.BULK.property, "57014", 0);
        try {
            storage.withDdlLockBound(recording(mock(Connection.class), "0"), Dialect.POSTGRES, () -> {
                throw cancelled;
            });
            fail("the cancelled statement was swallowed");
        }catch (SQLException e) {
            assertSame(e, cancelled, "a statement its own bound cancelled was reported as a lock wait");
        }
    }

    /**
     * The drop of a tree goes through the funnel every DDL of a transaction takes, so it is bounded
     * wherever it is issued from - {@code deleteTree()} here, and the create table and create index of
     * {@code openTree()} the same way.
     */
    @Test
    public void testTheDropOfATreeIsBounded() throws Exception {
        final JDBCStorage bounded = storageHandingOut(engine(postgresConnection.class, "0"));

        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))",
            "set local lock_timeout = 5000",
            "drop table " + JDBCStorage.toTableName(TREE)));
    }

    /**
     * The delete that empties a tree before an import is no DDL: it waits for row locks, which
     * {@code write()} replays a conflict of and which a bound meant for the metadata lock of a DDL has
     * no business ending.
     */
    @Test
    public void testTheDeleteThatEmptiesATreeIsNotBounded() throws Exception {
        final JDBCStorage bounded = storageHandingOut(engine(postgresConnection.class, "0"));

        // closed the way an import closes one: the importer holds a borrowed connection, and only its
        // close() gives that connection - and the permit it took - back. The statements of that close are
        // no part of this case, so what was issued is read before it.
        try (final Importer importer = bounded.new ImporterImpl()) {
            importer.clearTree(TREE);

            assertEquals(issued, singletonList("delete from " + JDBCStorage.toTableName(TREE)));
        }
    }

    /**
     * The drop loop of {@code removeStorageFiles()} bypasses that funnel and commits once at the end,
     * so the bound is set once around the whole loop rather than once per table - on postgres one
     * {@code set local} covers every drop of the single transaction it runs in.
     */
    @Test
    public void testTheDropLoopOfARemovedBackendIsBoundedOnce() throws Exception {
        final Connection con = engine(postgresConnection.class, "0");
        final JDBCStorage.TableScope scope = JDBCStorage.TableScope.of(storage, con);
        issued.clear(); // the search path the scope read is no part of what this case is about

        final JDBCStorage.ClearCounts counts = storage.dropCatalogTables(con, scope, catalogOf(TREE, OTHER_TREE));

        assertEquals(issued, asList("set local lock_timeout = 5000",
            "drop table " + JDBCStorage.toTableName(TREE),
            "drop table " + JDBCStorage.toTableName(OTHER_TREE)));
        assertEquals(counts.dropped, 2, "the clear did not account for the tables it dropped under the bound");
    }

    /**
     * And a clear with no row to act on is committed without the bound: putting it on costs a readback
     * and a restore of its own, and the first clear of a backend upgraded from a version that kept no
     * catalog - the case {@code CLEAR_DROPPED_NOTHING} describes - has no DDL for them to bound.
     */
    @Test
    public void testAClearWithNoTableToDropIsGivenNoBound() throws Exception {
        final Connection con = engine(postgresConnection.class, "0");
        final JDBCStorage.TableScope scope = JDBCStorage.TableScope.of(storage, con);
        issued.clear();

        storage.dropCatalogTables(con, scope, Collections.<TreeName, String>emptyMap());

        assertEquals(issued, emptyList());
    }

    /**
     * The lookup deciding each drop of that loop runs under the same bound as the drop it decides, and
     * it answers with a {@link StorageRuntimeException} rather than with the failure the engine gave
     * it. A lock this bound ended must be named there too: an operator meeting a bare 55P03 out of a
     * clear is the unexplained state this bound exists to stop shipping, and the drop one line away
     * would have named the property for the very same wait.
     */
    @Test
    public void testALockTheLookupOfAClearGaveUpOnNamesTheProperty() throws Exception {
        final Connection con = engine(postgresConnection.class, "0");
        final JDBCStorage.TableScope scope = JDBCStorage.TableScope.of(storage, con);
        final SQLException lockNotAvailable = new SQLException("canceling statement due to lock timeout", "55P03");
        givingUpOnTheLookup(con, lockNotAvailable);

        try {
            storage.dropCatalogTables(con, scope, catalogOf(TREE));
            fail("the clear went through although its lookup gave up on a lock");
        } catch (StorageRuntimeException e) {
            assertTrue(e.getCause() instanceof SQLTimeoutException,
                "the lookup's failure was left as the engine reported it: " + e.getCause());
            assertTrue(e.getCause().getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
                e.getCause().getMessage());
            assertSame(e.getCause().getCause(), lockNotAvailable, "the failure of the engine was not chained");
        }
    }

    /**
     * And a failure of that same lookup which was no lock wait is given back exactly as it arrived: the
     * rename says one thing about one wait, and a table that is not there or a connection that went
     * must not come out of a clear wearing the name of a property that had nothing to do with it.
     */
    @Test
    public void testAFailureOfTheLookupThatWasNoLockWaitIsLeftExactlyAsItIs() throws Exception {
        final Connection con = engine(postgresConnection.class, "0");
        final JDBCStorage.TableScope scope = JDBCStorage.TableScope.of(storage, con);
        final SQLException noSuchTable = new SQLException("relation does not exist", "42P01");
        givingUpOnTheLookup(con, noSuchTable);

        try {
            storage.dropCatalogTables(con, scope, catalogOf(TREE));
            fail("the clear went through although its lookup failed");
        } catch (StorageRuntimeException e) {
            assertSame(e.getCause(), noSuchTable, "a failure that was no lock wait was renamed");
        }
    }

    /** A connection whose table lookup answers with the given failure, as the drop loop asks it. */
    private void givingUpOnTheLookup(final Connection con, final SQLException failure) throws SQLException {
        final DatabaseMetaData metaData = mock(DatabaseMetaData.class);
        when(metaData.getTables(any(), any(), any(), any())).thenThrow(failure);
        when(con.getMetaData()).thenReturn(metaData);
    }

    /** A catalog naming each of the given trees at the table its name hashes to. */
    private static Map<TreeName, String> catalogOf(TreeName... trees) {
        final Map<TreeName, String> catalog = new LinkedHashMap<>();
        for (final TreeName tree : trees) {
            catalog.put(tree, JDBCStorage.toTableName(tree));
        }
        return catalog;
    }

    /**
     * A session already giving up sooner than this bound keeps exactly what it has: a deployment that
     * set {@code lock_wait_timeout} tighter did so on purpose, and loosening it to ours for the length
     * of a DDL is the very thing that leaves oracle alone. Nothing is set, so nothing is put back
     * either.
     */
    @Test
    public void testAMysqlSessionAlreadyTighterThanTheBoundKeepsWhatItHas() throws Exception {
        storage.withDdlLockBound(recording(mock(Connection.class), "1"), Dialect.MYSQL, theDdl());

        assertEquals(issued, asList("select @@session.lock_wait_timeout", THE_DDL));
    }

    /**
     * On sql server 0 is "do not wait at all", which is tighter than any bound of ours, while -1 is
     * "wait forever" and is replaced - the case the data provider above covers. A value read as a
     * number that happens to be negative must not be mistaken for a tight one.
     */
    @Test
    public void testASqlServerSessionThatDoesNotWaitAtAllKeepsWhatItHas() throws Exception {
        storage.withDdlLockBound(recording(mock(Connection.class), "0"), Dialect.MICROSOFT, theDdl());

        assertEquals(issued, asList("select @@lock_timeout", THE_DDL));
    }

    /**
     * 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 at all and the log would
     * read exactly like a bounded one. The setting is not issued there.
     */
    @Test
    public void testAConnectionInAutoCommitIsGivenNoSetLocal() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
        when(con.getAutoCommit()).thenReturn(true);

        storage.withDdlLockBound(con, Dialect.POSTGRES, theDdl());

        assertEquals(issued, singletonList(THE_DDL));
    }

    /**
     * A statement that fails inside a postgres transaction aborts it, and the DDL after it would then
     * fail with 25P02 rather than running unbounded as it did before this bound existed - a backend
     * that used to open would stop opening because of the bound meant to protect it. The transaction is
     * taken back to the point before the setting, which also undoes a {@code set local} that did reach
     * the server.
     */
    @Test
    public void testTheTransactionIsTakenBackToBeforeASettingThatFailed() throws Exception {
        final Connection con = mock(Connection.class);
        final Savepoint beforeTheBound = mock(Savepoint.class);
        when(con.setSavepoint()).thenReturn(beforeTheBound);
        when(con.createStatement()).thenThrow(new SQLException("current transaction is aborted", "25P02"));

        storage.withDdlLockBound(con, Dialect.POSTGRES, theDdl());

        verify(con).rollback(beforeTheBound);
        assertEquals(issued, singletonList(THE_DDL));
    }

    /** And a setting that went through is left standing: it is what bounds the DDL that follows it. */
    @Test
    public void testATransactionWhoseSettingWentThroughIsNotTakenBack() throws Exception {
        final Connection con = recording(mock(Connection.class), "0");
        final Savepoint beforeTheBound = mock(Savepoint.class);
        when(con.setSavepoint()).thenReturn(beforeTheBound);

        storage.withDdlLockBound(con, Dialect.POSTGRES, theDdl());

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

    /**
     * More than one wait of an engine reports the same number: mysql reports the row lock of
     * {@code innodb_lock_wait_timeout} - 50 s by default, and what a create index under
     * {@code ALGORITHM=COPY} waits on - as the same ERROR 1205 as a metadata lock. A wait that ran far
     * longer than this bound was ended by something else, and naming this property for it would send an
     * operator to raise the one setting that cannot help.
     */
    @Test
    public void testALockWaitFarPastTheBoundIsLeftExactlyAsItIs() {
        final SQLException rowLock = new SQLException("Lock wait timeout exceeded; try restarting transaction",
            "40001", 1205);
        final long fiftySecondsAgo = System.nanoTime() - 50L * 1000 * 1000 * 1000;

        assertSame(storage.gaveUpOnTheLock(rowLock, Dialect.MYSQL, 5, fiftySecondsAgo), rowLock,
            "a wait of innodb_lock_wait_timeout was reported as the bound this backend sets");
    }

    /** While one that ended where this bound is is renamed, which is what the bound exists to say. */
    @Test
    public void testALockWaitTheBoundCouldHaveEndedIsRenamed() {
        final SQLException lockWait = new SQLException("Lock wait timeout exceeded", "40001", 1205);

        final SQLException renamed = storage.gaveUpOnTheLock(lockWait, Dialect.MYSQL, 5, System.nanoTime());

        assertTrue(renamed.getMessage().contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY), renamed.getMessage());
        assertSame(renamed.getCause(), lockWait, "the failure of the engine was not chained");
    }

    /**
     * A connection left carrying a bound this backend could not take off again does not go back into
     * the pool. Leaving it to the next borrow to notice does not work: that validation is
     * {@code isValid()}, a liveness check such a connection passes - and on sql server the setting left
     * on it would cut every lock wait of the next borrower, row locks included, which
     * {@code isConflict()} classifies as no replayable conflict.
     */
    @Test
    public void testAConnectionWhoseBoundCouldNotBeTakenOffIsKeptOutOfThePool() throws Exception {
        final AtomicBoolean keptOut = new AtomicBoolean();
        final Connection parent = refusingToGiveTheValueBack(mock(Connection.class), "31536000");
        try (final CachedConnection con = new CachedConnection("jdbc:mock", parent) {
            @Override
            void keepOutOfThePool() {
                keptOut.set(true);
                super.keepOutOfThePool();
            }
        }) {
            storage.withDdlLockBound(con, Dialect.MYSQL, theDdl());
        }

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

    /**
     * The round trips of the bound itself carry a bound of their own. Unbounded, a readback on a
     * connection whose peer went quiet with the socket still open parks the thread opening a backend
     * for good - the hang #877 and #882 exist to end - and the restore does it from the finally of a
     * DDL that has already failed. The DDL between them keeps the class it had, which ships unbounded.
     */
    @Test
    public void testTheRoundTripsOfTheBoundCarryOneOfTheirOwn() throws Exception {
        final List<Integer> armed = new ArrayList<>();
        final Connection con = recording(mock(Connection.class), "31536000");
        doAnswer(invocation -> {
            armed.add((Integer) invocation.getArguments()[1]);
            return null;
        }).when(con).setNetworkTimeout(any(), anyInt());

        storage.withDdlLockBound(con, Dialect.MYSQL, theDdl());

        assertTrue(armed.contains((JDBCStorage.SESSION_STATEMENT_BOUND_SECONDS
                + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000),
            "the session statements of the bound armed no socket read timeout: " + armed);
    }

    /** The DDL itself, recording that it ran in among the session statements issued around it. */
    private JDBCStorage.Execution<Void> theDdl() {
        return () -> {
            issued.add(THE_DDL);
            return null;
        };
    }

    /**
     * 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;
    }

    /**
     * The same, on a connection that takes a setting through and then breaks as the statement that
     * carried it is closed - all a driver reports of a session that went in between.
     */
    private Connection breakingOnTheCloseOfASetting(final Connection con, final String carries)
            throws SQLException {
        final Statement statement = recording(con, carries).createStatement();
        final AtomicBoolean carried = new AtomicBoolean();
        doAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            carried.set(true);
            return false;
        }).when(statement).execute(anyString());
        doAnswer(invocation -> {
            if (carried.get()) {
                throw new SQLException("the connection went as the statement was closed", "08006");
            }
            return null;
        }).when(statement).close();
        return con;
    }

    /** And one 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;
    }

    /**
     * Connections whose class names carry the engine the way the drivers' own do - pgjdbc's
     * {@code org.postgresql.jdbc.PgConnection}. That name is what {@code dialectOf()} reads the
     * engine off, and the name of a mock is derived from the type it mocks, so a mock of plain
     * {@link Connection} reaches no engine branch at all. Lowercase because the match is case
     * sensitive.
     */
    interface postgresConnection extends Connection {
    }

    /**
     * A connection of the given engine, recording the statements it is asked to run - the DDL among the
     * session settings around it - over a catalog holding the table of every tree named here.
     */
    private Connection engine(Class<? extends Connection> engine, String carries) throws SQLException {
        final Connection con = recording(mock(engine), carries);
        when(con.isValid(anyInt())).thenReturn(true);
        when(con.prepareStatement(anyString())).thenAnswer(invocation -> {
            issued.add((String) invocation.getArguments()[0]);
            final PreparedStatement statement = mock(PreparedStatement.class);
            when(statement.getConnection()).thenReturn(con);
            return statement;
        });
        final DatabaseMetaData metaData = mock(DatabaseMetaData.class);
        when(metaData.getTables(any(), any(), any(), any())).thenAnswer(invocation -> {
            final ResultSet tables = mock(ResultSet.class);
            final String asked = (String) invocation.getArguments()[2];
            // Every tree of a case is found to exist, and the tree catalog of the backend is found not
            // to be there: what these cases are about is the statements issued around a DDL, and a
            // backend upgraded from a version that kept no catalog takes the shortest way to the funnel
            // carrying them - the catalog would otherwise want a connection of its own, which is not a
            // connection this mock hands out. CatalogConnectionTestCase covers that one.
            when(tables.next()).thenReturn(!NO_CATALOG_TABLE.equals(asked), false);
            // the name the catalog was asked about, so that every tree of a case is found to exist
            when(tables.getString("TABLE_NAME")).thenReturn(asked);
            return tables;
        });
        when(con.getMetaData()).thenReturn(metaData);
        return con;
    }

    /**
     * A storage handing out the given connection, through the seam an import of
     * {@code JDBCStatementBoundTestCase} borrows through: what these cases are about is the statements
     * around a DDL, not the pool that produced the connection carrying them.
     */
    private JDBCStorage storageHandingOut(final Connection con) {
        final JDBCStorage handing = new JDBCStorage(backendCfg(), null) {
            @Override
            Connection getConnection(boolean trusted) {
                return new CachedConnection("jdbc:mock", con);
            }

            @Override
            public StorageStatus getStorageStatus() {
                return StorageStatus.working(); // open already, so an importer borrows and no more
            }
        };
        handing.accessMode = AccessMode.READ_WRITE;
        return handing;
    }
}

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

@@ -267,6 +267,138 @@
        }
    }

    /**
     * The DDL of this backend is the part of it that takes locks, and three engines out of four wait
     * for one essentially forever: a drop queued behind an unrelated transaction of another session
     * used to hang the backend that issued it, with no property to say otherwise (#885). It gives up
     * at the bound now, and says which property ended the wait.
     * <p>
     * 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.
     */
    @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");
        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, holding a lock the drop below conflicts with on every one of these
            // engines: an uncommitted write takes a lock on the table that a drop cannot share
            try (final Connection holder = DriverManager.getConnection(getJdbcUrl())) {
                holder.setAutoCommit(false);
                try (final PreparedStatement write = holder.prepareStatement(
                        "insert into " + JDBCStorage.toTableName(tree) + " (h,k,v) values (?,?,?)")) {
                    write.setString(1, "a lock this session holds");
                    write.setBytes(2, new byte[]{ 1 });
                    write.setBytes(3, new byte[]{ 1 });
                    write.executeUpdate();
                }
                final JDBCStorage.Dialect dialect = dialect();

                final long startedAt = System.nanoTime();
                Exception failure = null;
                try {
                    storage.write(new WriteOperation() {
                        @Override
                        public void run(WriteableTransaction txn) throws Exception {
                            txn.deleteTree(tree);
                        }
                    });
                    fail("the drop went through while another session held the table locked");
                } catch (Exception e) {
                    failure = e;
                }
                final String reported = stackTraceToSingleLineString(failure);
                final long tookSeconds = (System.nanoTime() - startedAt) / 1000000000L;
                // generous, and still far under what an unbounded wait costs: mysql waits a year for a
                // metadata lock by default, sql server and postgres wait for one without limit at all
                assertTrue(tookSeconds < 60, "the drop waited " + tookSeconds + " s for the lock: " + reported);
                // By the engine's own verdict rather than by the text of the message: without this the case
                // passes for a drop that failed because the table was not there, because the account lacked
                // the privilege, or because the statement never reached the engine - none of which is a lock
                // this drop gave up on. ORA-00054 on oracle, 55P03 / 1205 / 1222 on the other three.
                assertTrue(JDBCStorage.lockNotAvailable(failure, dialect),
                        "the drop did not fail as this engine says a lock was not available: " + reported);
                if (dialect == JDBCStorage.Dialect.ORACLE) {
                    assertFalse(reported.contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
                            "oracle is left to its own ddl_lock_timeout: " + reported);
                } else {
                    assertTrue(reported.contains(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY),
                            "the failure names neither the wait nor the property that ended it: " + reported);
                }
                holder.rollback();
            }
        } finally {
            System.clearProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY);
            try {
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        txn.deleteTree(tree);
                    }
                });
            } catch (Exception ignored) {
            } finally {
                storage.close();
            }
        }
    }

    /**
     * The value a pooled connection carried is given back the moment the DDL is through. That
     * connection outlives the transaction that borrowed it and {@code CachedConnection.close()} only
     * rolls back, so a bound left on it would end every lock wait of whoever borrows it next - on sql
     * server row locks included, which {@code isConflict()} classifies as no replayable conflict, so
     * {@code write()} would not replay them and a client would see a hard failure.
     * <p>
     * {@code JDBCDdlLockBoundTestCase} pins the string each engine is handed; only a session of the
     * engine itself can say what it does with it. Postgres asserts the other shape of the same
     * contract: nothing is put back by hand there, because a {@code set local} belongs to the
     * transaction and is gone with the commit that ends the DDL.
     */
    @Test(timeOut = 120000)
    public void testAConnectionGetsItsLockBoundBackAfterADdl() throws Exception {
        final JDBCStorage.Dialect dialect = dialect();
        final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
        System.setProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY, "7");
        try {
            storage.open(AccessMode.READ_WRITE);
            try (final Connection con = CachedConnection.getConnection(getJdbcUrl())) {
                final String before = sessionLockBound(con);
                if (before == null) { // oracle: nothing of ours is set there, and v$parameter is out of reach
                    throw new SkipException("no session lock bound to read on " + getJdbcUrl());
                }
                final String[] during = new String[1];
                storage.withDdlLockBound(con, dialect, () -> {
                    during[0] = sessionLockBound(con);
                    return null;
                });
                assertNotEquals(during[0], before,
                        "the bound was never on the session the DDL ran on: it carried " + during[0]);
                if (dialect.boundLivesInTheTransaction()) {
                    assertEquals(sessionLockBound(con), during[0],
                            "a set local was taken off before the commit that is what discards it");
                    con.commit();
                }
                assertEquals(sessionLockBound(con), before,
                        "the value the session carried was not given back after the DDL");
            }
        } finally {
            System.clearProperty(JDBCStorage.DDL_LOCK_TIMEOUT_PROPERTY);
            storage.close();
        }
    }

    private static ByteString key(int i) {
        return ByteString.valueOfUtf8(String.format("key%02d", i));
    }
@@ -1056,7 +1188,7 @@
     * engine, or null where reading it needs a privilege the test user does not have: oracle
     * keeps ddl_lock_timeout in v$parameter, which an application user cannot select from.
     */
    String sessionLockBound(Connection con) throws Exception {
    String sessionLockBound(Connection con) throws SQLException {
        final String url = getJdbcUrl();
        final String sql;
        if (url.startsWith("jdbc:postgresql")) {