diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
index d1d8055314..eeee36c4be 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
@@ -41,6 +41,8 @@
 import java.sql.*;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.opends.server.backends.pluggable.spi.StorageUtils.addErrorMessage;
 import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;
@@ -117,21 +119,391 @@ public ConfigChangeResult applyConfigurationChange(JDBCBackendCfg cfg) {
 		return ccr;
 	}
 
-	ResultSet executeResultSet(PreparedStatement statement) throws SQLException {
+	/**
+	 * What a statement of this backend may legitimately take, and the property bounding it. One
+	 * value cannot serve both: an entry read is a single row of an index, while the count of a
+	 * tree and the delete that empties one before an import are a scan and a rewrite of a whole
+	 * table, which take minutes on a populated backend and are not a symptom of anything.
+	 */
+	enum StatementBound {
+		/** one row by primary key, or one batch of a cursor along its index */
+		OPERATION("org.openidentityplatform.opendj.jdbc.query.timeout", 120),
+		/**
+		 * a whole table at once: count(*), the delete of clearTree, the scan behind the highest
+		 * entry id, create index, drop table. This class ships <em>unbounded</em>: what such a
+		 * statement legitimately takes follows the size of the backend and the speed of its
+		 * database, neither of which can be guessed here, so the deployment that knows both sets
+		 * the property - until it does, a create index waiting for a metadata lock still waits for
+		 * as long as the engine lets it.
+		 */
+		BULK("org.openidentityplatform.opendj.jdbc.bulk.timeout", 0);
+
+		final String property;
+		final int defaultSeconds;
+
+		StatementBound(String property, int defaultSeconds) {
+			this.property = property;
+			this.defaultSeconds = defaultSeconds;
+		}
+
+		/**
+		 * The bound in seconds, as configured by {@link #property}: 0, or a negative value, leaves
+		 * the statement unbounded, as it was before this bound existed, while a value that is not a
+		 * number is ignored in favour of {@link #defaultSeconds} - {@code Integer.getInteger()}
+		 * falls back to its default rather than reading such a value as a zero.
+		 */
+		int seconds() {
+			return Math.max(0, Integer.getInteger(property, defaultSeconds));
+		}
+	}
+
+	/** What a caller of {@link #executeResultSet} makes of the rows, while the bound is still armed. */
+	interface RowsHandler<T> {
+		T handle(ResultSet rows) throws SQLException;
+	}
+
+	<T> T executeResultSet(PreparedStatement statement, RowsHandler<T> rows) throws SQLException {
+		return executeResultSet(statement, StatementBound.OPERATION, rows);
+	}
+
+	/**
+	 * Runs a query under the bound of its class and hands the rows to {@code rows} while that bound
+	 * is still armed. They are read there rather than after this method returns because a driver
+	 * transfers them as they are asked for: read outside, the transfer - up to a whole batch of a
+	 * cursor - would run with neither layer of the bound covering it, which is exactly where a
+	 * database that stops answering mid-drain parks the worker thread. {@code setQueryTimeout}
+	 * covering {@code ResultSet.next()} is optional in the JDBC contract ("drivers <em>may</em>
+	 * also apply this limit"), and the two drivers of this backend that do not buffer a result
+	 * whole - oracle prefetches ten rows at a time, mssql buffers adaptively - are the ones that
+	 * do not.
+	 */
+	<T> T executeResultSet(PreparedStatement statement, StatementBound bound, RowsHandler<T> rows) throws SQLException {
 		if (logger.isTraceEnabled()) {
 			logger.trace(LocalizableMessage.raw("jdbc: %s",statement));
 		}
-		return statement.executeQuery();
+		return bounded(statement, bound, () -> {
+			try (final ResultSet rs=statement.executeQuery()) {
+				return rows.handle(rs);
+			}
+		});
 	}
 
 	int execute(PreparedStatement statement) throws SQLException {
+		return execute(statement, StatementBound.OPERATION);
+	}
+
+	int execute(PreparedStatement statement, StatementBound bound) throws SQLException {
 		if (logger.isTraceEnabled()) {
 			logger.trace(LocalizableMessage.raw("jdbc: %s",statement));
 		}
-		return statement.executeUpdate();
+		return bounded(statement, bound, statement::executeUpdate);
+	}
+
+	interface Execution<T> {
+		T run() throws SQLException;
+	}
+
+	/**
+	 * Runs a statement under the bound of its class. A statement of this backend has to end: a row
+	 * locked by an unrelated session, a table waiting for a metadata lock or a database that stops
+	 * answering mid-query would otherwise park the worker thread that issued it for good.
+	 * <p>
+	 * The bound is asked of the driver rather than of the session, because a pooled connection
+	 * cannot carry a session setting - {@code CachedConnection.close()} only rolls back, so a
+	 * {@code statement_timeout} of one operation would apply to whoever borrows the connection
+	 * next - and it is applied in two layers, since the first one is not answered everywhere:
+	 * {@code setQueryTimeout} cancels the statement and keeps the connection, while the socket read
+	 * timeout behind it ends the wait even when the cancel is not acted upon. Oracle needs that
+	 * second layer: a session blocked in a row-lock enqueue does not process the break its driver
+	 * sends, so the timeout is armed and never arrives (the container suites cover it). That second
+	 * layer belongs to the connection rather than to the statement, so it is arbitrated between the
+	 * statements running on one - see {@link Backstop}.
+	 */
+	private <T> T bounded(PreparedStatement statement, StatementBound bound, Execution<T> execution) throws SQLException {
+		final int seconds=bound.seconds();
+		// whether the cancel is in force: a driver is free to refuse the query timeout, and then the
+		// socket read timeout behind it is the only layer this statement has - one that arrives later
+		final boolean cancelArmed=seconds > 0 && setQueryTimeout(statement, seconds);
+		// an unbounded class is announced to the connection all the same: a statement told it may
+		// take as long as it needs must not be cut by the socket read timeout of a concurrent one
+		return bounded(connectionOf(statement), bound.property, seconds, cancelArmed, execution);
+	}
+
+	/**
+	 * Runs the catalog lookups of {@code openTree()} under the bound of their class. They ask
+	 * {@code DatabaseMetaData}, which takes no query timeout, so the socket read timeout behind the
+	 * cancel is the only layer they can be given - and they do need one: they run once per tree on
+	 * every open of a backend, and the catalog is answered by the same engine, behind the same
+	 * locks, as the {@code create table} they guard.
+	 */
+	<T> T bounded(Connection con, StatementBound bound, Execution<T> execution) throws SQLException {
+		// no cancel to arm: DatabaseMetaData takes no query timeout, so the socket read timeout behind
+		// it is the only layer these have, and nothing ends their wait before the margin of that layer
+		return bounded(con, bound.property, bound.seconds(), false, execution);
+	}
+
+	/**
+	 * Runs a statement under a bound of its own rather than under the bound of a class, for the one
+	 * statement that has a property of its own: the statistics refresh after an import, which
+	 * legitimately takes as long as a scan of the table it describes.
+	 */
+	private <T> T bounded(Connection con, String property, int seconds, boolean cancelArmed, Execution<T> execution)
+			throws SQLException {
+		final long startedAt=System.nanoTime();
+		final Backstop backstop=holdBackstop(con, seconds);
+		try {
+			return execution.run();
+		}catch (SQLException e) {
+			throw timedOut(e, property, seconds, cancelArmed, startedAt);
+		}finally {
+			releaseBackstop(backstop, con, seconds);
+		}
 	}
 
-	// unlike execute(), tolerates statements that return a result set ("analyze table" on mysql)
+	/**
+	 * Asks the driver to cancel the statement at the bound. Not every driver has one: the JDBC
+	 * contract allows {@code SQLFeatureNotSupportedException} and this backend takes whatever URL a
+	 * deployment configures, so a driver without it degrades to the socket read timeout behind it
+	 * rather than failing every statement it is given.
+	 */
+	private boolean setQueryTimeout(PreparedStatement statement, int seconds) {
+		try {
+			statement.setQueryTimeout(seconds);
+			return true;
+		}catch (SQLException | RuntimeException e) {
+			if (queryTimeoutWarned.compareAndSet(false, true)) {
+				logger.warn(LocalizableMessage.raw("jdbc: the driver would not take a query timeout (%s): a statement of this"
+					+ " backend is left to the socket read timeout behind it", e.getMessage()));
+			}
+			return false;
+		}
+	}
+
+	private Connection connectionOf(PreparedStatement statement) {
+		try {
+			return statement.getConnection();
+		}catch (SQLException | RuntimeException e) {
+			return null; // nothing to arm the backstop on; the cancel above is the whole bound
+		}
+	}
+
+	/** How long the socket read timeout outlasts the cancel it backs up, giving it room to arrive. */
+	static final int BACKSTOP_MARGIN_SECONDS = 30;
+
+	// setNetworkTimeout() takes the executor its timeout handling runs on; the drivers of this
+	// backend only set a socket option in it, so it costs a call rather than a thread.
+	private static final Executor DIRECT_EXECUTOR = Runnable::run;
+
+	// Set when the driver of this storage has no network timeout to give at all, which is a property
+	// of the driver rather than of a connection: asking it again would cost a throw per statement,
+	// and the entry a connection's Backstop lives in is gone as soon as nothing runs on it. Held per
+	// storage rather than per JVM, like the warnings below: a driver that will not take one of these
+	// says so once for every backend running on it, instead of one backend silencing it for all.
+	private final AtomicBoolean backstopUnsupported = new AtomicBoolean();
+	private final AtomicBoolean backstopUnsupportedWarned = new AtomicBoolean();
+	private final AtomicBoolean backstopFailedWarned = new AtomicBoolean();
+	private final AtomicBoolean queryTimeoutWarned = new AtomicBoolean();
+
+	/**
+	 * The socket read timeout of one connection, and the statements running on it. This second
+	 * layer of the bound is a property of the socket rather than of a statement, so it cannot be
+	 * armed and put back per statement wherever a connection carries more than one at a time: an
+	 * {@code ImporterImpl} holds a single connection for the whole of an import and writes to it
+	 * from every phase-one worker and every phase-two task, and there the first statement to finish
+	 * would take the backstop away from every statement still in flight - while a statement whose
+	 * class carries no bound at all would run under whatever value a concurrent one happened to
+	 * arm, dying at it with nothing to say which property cut it, since such a statement never
+	 * reaches {@link #timedOut}.
+	 * <p>
+	 * So the value armed is the loosest of the bounds of the statements in flight, and a statement
+	 * with no bound of its own takes it off for as long as it runs: this backstop exists to end a
+	 * wait nothing else would end, never to cut a statement that was told it may take as long as it
+	 * needs. What the connection carried before is put back when the last of them is through.
+	 */
+	private static final class Backstop {
+		/** Bounds of the statements in flight, in milliseconds and by count, the loosest last. */
+		final TreeMap<Integer,Integer> bounds=new TreeMap<>();
+		/** Statements in flight with no bound of their own, which no backstop may cut short. */
+		int unbounded;
+		/** Statements holding this entry, bounded or not: at zero it leaves {@link #backstops}. */
+		int holders;
+		/** What the connection carried before the backstop armed it, and is given back afterwards. */
+		int previous;
+		/** What the backstop has armed, or 0 when the connection carries {@link #previous}. */
+		int armed;
+		/**
+		 * Set when the driver would not take a network timeout on this connection: it is not asked
+		 * again while the statements holding this entry run. A connection is the right scope for
+		 * that: the common cause is a connection on its way out, and a driver that has no network
+		 * timeout at all is remembered for the whole storage instead - see {@link #backstopUnsupported}.
+		 */
+		boolean failed;
+	}
+
+	// Keyed by identity on the connection of the driver: CachedConnection.prepareStatement() hands
+	// the statement to the connection it wraps, so that is the one a statement reports, while the
+	// catalog lookups above hold the wrapper of that same connection - both have to find the same
+	// entry, so a wrapper is unwrapped on the way in. Static because the pool these connections
+	// come from is static; an entry lives only while statements are running on its connection.
+	private static final Map<Connection,Backstop> backstops = new IdentityHashMap<>();
+
+	private static Connection physical(Connection con) {
+		return con instanceof CachedConnection ? ((CachedConnection)con).parent : con;
+	}
+
+	/**
+	 * Puts the bound of a statement about to run on the connection that will run it, and makes the
+	 * socket read timeout of that connection fit every statement in flight on it. Reaching this
+	 * bound, unlike reaching the cancel it backs up, costs the connection: the driver closes it,
+	 * which is the price of a wait the database was never going to end on its own.
+	 */
+	private Backstop holdBackstop(Connection con, int seconds) {
+		final Connection physical=physical(con);
+		if (physical == null) {
+			return null;
+		}
+		final Backstop state;
+		synchronized (backstops) {
+			state=backstops.computeIfAbsent(physical, c -> new Backstop());
+			state.holders++; // held from here, so that the entry outlives a concurrent release
+		}
+		synchronized (state) {
+			if (seconds > 0) {
+				state.bounds.merge(backstopMillis(seconds), 1, Integer::sum);
+			}else {
+				state.unbounded++;
+			}
+			applyBackstop(physical, state);
+		}
+		return state;
+	}
+
+	private void releaseBackstop(Backstop state, Connection con, int seconds) {
+		if (state == null) {
+			return;
+		}
+		final Connection physical=physical(con);
+		try {
+			synchronized (state) {
+				if (seconds > 0) {
+					final int millis=backstopMillis(seconds);
+					final Integer inFlight=state.bounds.get(millis);
+					if (inFlight == null || inFlight <= 1) {
+						state.bounds.remove(millis);
+					}else {
+						state.bounds.put(millis, inFlight-1);
+					}
+				}else {
+					state.unbounded--;
+				}
+				applyBackstop(physical, state);
+			}
+		}finally { // the entry is let go whatever the driver did, so that it cannot outlive its connection
+			synchronized (backstops) {
+				if (--state.holders <= 0) { // nothing is running on it: the connection is on its own again
+					backstops.remove(physical);
+				}
+			}
+		}
+	}
+
+	private static int backstopMillis(int seconds) {
+		return (int) Math.min(Integer.MAX_VALUE, (seconds+BACKSTOP_MARGIN_SECONDS)*1000L);
+	}
+
+	/**
+	 * Makes the socket read timeout of the connection what the statements in flight on it need: the
+	 * loosest of their bounds, or nothing of ours at all while one of them carries no bound. Called
+	 * with the monitor of {@code state} held, since it both reads those counts and acts on the
+	 * driver.
+	 */
+	private void applyBackstop(Connection con, Backstop state) {
+		if (state.failed || backstopUnsupported.get()) {
+			return;
+		}
+		final int wanted=state.unbounded > 0 || state.bounds.isEmpty() ? 0 : state.bounds.lastKey();
+		try {
+			if (wanted == 0) {
+				if (state.armed != 0) {
+					con.setNetworkTimeout(DIRECT_EXECUTOR, state.previous);
+					state.armed=0;
+				}
+				return;
+			}
+			if (state.armed == 0) {
+				state.previous=con.getNetworkTimeout();
+			}
+			// only ever tighten: a connection that already carries a read timeout carries one a
+			// deployment asked for, and this backstop exists to cap a cancel that is not acted
+			// upon, not to relax anything. 0 is "no timeout" in the JDBC contract, so it is the
+			// one value there is always something to gain by replacing.
+			if (state.previous > 0 && state.previous <= wanted) {
+				if (state.armed != 0) {
+					con.setNetworkTimeout(DIRECT_EXECUTOR, state.previous);
+					state.armed=0;
+				}
+				return;
+			}
+			if (state.armed != wanted) {
+				con.setNetworkTimeout(DIRECT_EXECUTOR, wanted);
+				state.armed=wanted;
+			}
+		}catch (SQLException | RuntimeException e) {
+			state.failed=true; // whatever the cause, this connection is not asked again while it runs
+			// The two causes are told apart, because they deserve opposite treatment and one of them
+			// would otherwise spend the single warning the other needs: a driver with no network
+			// timeout at all says so through SQLFeatureNotSupportedException, and there is nothing to
+			// gain by asking it once per statement for the life of the storage, while a connection on
+			// its way out - it may be the one that reached this very timeout - says nothing about the
+			// driver and must not disable the backstop for the connections that are still healthy.
+			if (e instanceof SQLFeatureNotSupportedException) {
+				backstopUnsupported.set(true);
+				if (backstopUnsupportedWarned.compareAndSet(false, true)) {
+					logger.warn(LocalizableMessage.raw("jdbc: the driver takes no socket read timeout (%s): a statement the"
+						+ " database does not cancel will wait for it indefinitely, unless the connect properties of the URL"
+						+ " configured for this backend carry one", e.getMessage()));
+				}
+			}else if (backstopFailedWarned.compareAndSet(false, true)) {
+				logger.warn(LocalizableMessage.raw("jdbc: the socket read timeout backing up a cancelled statement could not"
+					+ " be set on a connection (%s): a statement the database does not cancel will wait for it"
+					+ " indefinitely there", e.getMessage()));
+			}
+		}
+	}
+
+	// Every driver reports a cancelled statement differently - postgresql as 57014, oracle as
+	// ORA-01013, and neither of them as a SQLTimeoutException - so the bound is recognized by the
+	// time the statement took rather than by the class or the state of its failure, and that time
+	// is taken from the monotonic clock, which a step of the wall clock can neither lengthen nor
+	// shorten. What it cannot tell apart is a failure of another kind arriving after the bound,
+	// which is why the failure it replaces is chained rather than swallowed. The SQL state and the
+	// error number are carried over as well, since a failure that arrives at the bound may still be
+	// one a caller classifies: a mysql lock wait, reported in class 40, ends inside a longer bound
+	// and stays the replayable conflict it is. The statement itself is left out of the message: a
+	// driver renders it with its parameters bound, and those are entry data.
+	private SQLException timedOut(SQLException e, String property, int seconds, boolean cancelArmed, long startedAt) {
+		if (seconds <= 0) {
+			return e;
+		}
+		// Where the cancel is armed, the property ends the wait at its own value. Where it is not - a
+		// statement of DatabaseMetaData takes no query timeout, and a driver is free to refuse one -
+		// the socket read timeout behind it is the only layer there is, and that one arrives a margin
+		// later: measuring such a statement against the property alone reported a connection reset at
+		// 121 s as a query timeout of 120 s and sent the operator to a property that bounded nothing.
+		final long endsAfter=cancelArmed ? seconds : seconds+(long)BACKSTOP_MARGIN_SECONDS;
+		if (System.nanoTime()-startedAt < endsAfter*1_000_000_000L) {
+			return e;
+		}
+		return new SQLTimeoutException("jdbc: the statement did not finish within the "+endsAfter+"s of "
+			+(cancelArmed ? property : "the socket read timeout behind "+property+" ("+seconds+"s plus the margin of"
+				+" that layer, which is the only one bounding a statement taking no query timeout)")
+			+": raise that property, or set it to 0 for no bound", e.getSQLState(), e.getErrorCode(), e);
+	}
+
+	// Unlike execute(), tolerates a statement that returns a result set - the comment statement of
+	// mssql is a batch that ends in an exec - and, unlike it, carries no bound of its own: what is
+	// left of this method runs on a stamp connection, which is given a lock timeout of its own
+	// (Dialect.lockTimeoutSql) and a socket read timeout in its connect properties.
 	void executeAny(PreparedStatement statement) throws SQLException {
 		if (logger.isTraceEnabled()) {
 			logger.trace(LocalizableMessage.raw("jdbc: %s",statement));
@@ -351,9 +723,8 @@ private static String sqlLiteral(String value, boolean backslashIsEscape) {
 	// Asked of the very session that parses the literal: sql_mode is a session setting, and a
 	// session opened at another moment can have been given another value of it.
 	boolean isMysqlBackslashEscape(Connection con) throws SQLException {
-		try (final PreparedStatement statement=con.prepareStatement("select @@sql_mode");
-			final ResultSet rs=executeResultSet(statement)) {
-			final String sqlMode=rs.next() ? rs.getString(1) : null;
+		try (final PreparedStatement statement=con.prepareStatement("select @@sql_mode")) {
+			final String sqlMode=executeResultSet(statement, rs -> rs.next() ? rs.getString(1) : null);
 			return sqlMode==null || !sqlMode.toUpperCase().contains("NO_BACKSLASH_ESCAPES");
 		}
 	}
@@ -457,6 +828,10 @@ public void close() {
 	// 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).
 	private void executeSessionStatement(Connection con, String sql) throws SQLException {
 		try (final Statement statement=con.createStatement()) {
 			if (logger.isTraceEnabled()) {
@@ -677,9 +1052,7 @@ String readStoredComment(Connection con, Dialect dialect, String tableName) thro
 		}
 		try (final PreparedStatement statement=con.prepareStatement(sql)) {
 			statement.setString(1,arg);
-			try (final ResultSet rs=executeResultSet(statement)) {
-				return rs.next() ? rs.getString(1) : null;
-			}
+			return executeResultSet(statement, rs -> rs.next() ? rs.getString(1) : null);
 		}
 	}
 
@@ -740,21 +1113,36 @@ boolean updateTableStatistics(Connection con, Collection<TreeName> trees) {
 					throw new IllegalStateException("no statistics refresh for dialect "+dialect);
 				}
 				try (final PreparedStatement statement=con.prepareStatement(sql)) {
-					statement.setQueryTimeout(timeoutSeconds); // 0: wait without limit
+					// 0: wait without limit - and false where the driver would not take the cancel, which
+					// leaves the socket read timeout behind it as the only layer this refresh runs under
+					final boolean cancelArmed=timeoutSeconds>0 && setQueryTimeout(statement, timeoutSeconds);
 					for (int i=0;i<args.length;i++) {
 						statement.setString(i+1,args[i]);
 					}
-					if (dialect==Dialect.MYSQL) { // mysql reports analyze problems as a result row, not an SQLException
-						try (final ResultSet rs=executeResultSet(statement)) {
-							while (rs.next()) {
-								if ("error".equalsIgnoreCase(rs.getString("Msg_type"))) {
-									throw new SQLException(rs.getString("Msg_text"));
+					// Under the bound of the statistics refresh rather than under a class of
+					// StatementBound, which would put its own value over one this statement has a
+					// property for - but under both layers of it all the same: on oracle this is
+					// dbms_stats.gather_table_stats, the engine whose session does not act on the
+					// break its driver sends, and it runs at the very end of a successful import,
+					// where a cancel that never arrives would park it with the data already
+					// committed and nothing left to report.
+					bounded(con, STATISTICS_TIMEOUT_PROPERTY, timeoutSeconds, cancelArmed, () -> {
+						if (logger.isTraceEnabled()) {
+							logger.trace(LocalizableMessage.raw("jdbc: %s",statement));
+						}
+						if (dialect==Dialect.MYSQL) { // mysql reports analyze problems as a result row, not an SQLException
+							try (final ResultSet rs=statement.executeQuery()) {
+								while (rs.next()) {
+									if ("error".equalsIgnoreCase(rs.getString("Msg_type"))) {
+										throw new SQLException(rs.getString("Msg_text"));
+									}
 								}
 							}
+						}else { // tolerates a statement that returns a result set, which execute() does not
+							statement.execute();
 						}
-					}else {
-						executeAny(statement);
-					}
+						return null;
+					});
 					con.commit();
 				}
 			}catch (Exception e) {
@@ -785,7 +1173,7 @@ public void removeStorageFiles() throws StorageRuntimeException {
 				try {
 					for (final TreeName treeName : trees) {
 						try (final PreparedStatement statement = con.prepareStatement("drop table " + getTableName(treeName))) {
-							execute(statement);
+							execute(statement, StatementBound.BULK);
 						}
 					}
 					con.commit();
@@ -1007,12 +1395,28 @@ static String hashParam(Connection con) {
 		return driverNameOf(con).contains("microsoft") ? "cast(? as char(128))" : "?";
 	}
 
-	private class ReadableTransactionImpl implements ReadableTransaction {
+	class ReadableTransactionImpl implements ReadableTransaction {
 		final Connection con;
+		/**
+		 * The class the statements of this transaction take. It follows who runs them rather than
+		 * what they look like: an import issues the same select and the same upsert a client
+		 * operation does, but nobody is waiting on it - and on mssql it works the table unindexed,
+		 * {@code k} being a {@code varbinary(max)} that cannot be an index key - so bounding an
+		 * import as an entry read fails an import that ran to the end before this bound existed.
+		 * The catalog lookups of {@code openTree()} keep the operation class whoever runs them: they
+		 * read a data dictionary rather than the data, so a wait there is another session's metadata
+		 * lock, which is one of the waits this bound exists to end.
+		 */
+		final StatementBound bound;
 		boolean isReadOnly=true;
 
 		public ReadableTransactionImpl(Connection con) {
+			this(con, StatementBound.OPERATION);
+		}
+
+		ReadableTransactionImpl(Connection con, StatementBound bound) {
 			this.con=con;
+			this.bound=bound;
 		}
 
 		@Override
@@ -1020,9 +1424,7 @@ public ByteString read(TreeName treeName, ByteSequence key) {
 			try (final PreparedStatement statement=con.prepareStatement("select v from "+getTableName(treeName)+" where h="+hashParam(con)+" and k=?")){
 				statement.setString(1,key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 				statement.setBytes(2,real2db(key.toByteArray()));
-				try(ResultSet rc=executeResultSet(statement)) {
-					return rc.next() ? ByteString.wrap(rc.getBytes("v")) : null;
-				}
+				return executeResultSet(statement, bound, rc -> rc.next() ? ByteString.wrap(rc.getBytes("v")) : null);
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
 			}
@@ -1030,14 +1432,27 @@ public ByteString read(TreeName treeName, ByteSequence key) {
 
 		@Override
 		public Cursor<ByteString, ByteString> openCursor(TreeName treeName) {
-			return new CursorImpl(isReadOnly,con,treeName);
+			return new CursorImpl(isReadOnly,con,treeName,bound);
+		}
+
+		/**
+		 * {@inheritDoc}
+		 * <p>
+		 * The batches of such a cursor are bulk statements however ordinary they look: nobody is
+		 * waiting on the walk, and on mssql it is not even a walk along an index - {@code k} is a
+		 * {@code varbinary(max)} there, which cannot be an index key, so every batch is a scan and
+		 * a sort of the table. Bounding those as entry reads aborted an export or a rebuild that
+		 * ran to the end before this bound existed.
+		 */
+		@Override
+		public Cursor<ByteString, ByteString> openBulkCursor(TreeName treeName) {
+			return new CursorImpl(isReadOnly,con,treeName,StatementBound.BULK);
 		}
 
 		@Override
 		public long getRecordCount(TreeName treeName) {
-			try (final PreparedStatement statement=con.prepareStatement("select count(*) from "+getTableName(treeName));
-				 final ResultSet rc=executeResultSet(statement)){
-				return rc.next() ? rc.getLong(1) : 0;
+			try (final PreparedStatement statement=con.prepareStatement("select count(*) from "+getTableName(treeName))){
+				return executeResultSet(statement, StatementBound.BULK, rc -> rc.next() ? rc.getLong(1) : 0);
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
 			}
@@ -1051,7 +1466,11 @@ private final class WriteableTransactionTransactionImpl extends ReadableTransact
 		final StampSession stampSession=new StampSession();
 
 		public WriteableTransactionTransactionImpl(Connection con) {
-			super(con);
+			this(con, StatementBound.OPERATION);
+		}
+
+		WriteableTransactionTransactionImpl(Connection con, StatementBound bound) {
+			super(con, bound);
 			if (!accessMode.isWriteable()) {
 				throw new ReadOnlyStorageException();
 			}
@@ -1060,26 +1479,31 @@ public WriteableTransactionTransactionImpl(Connection con) {
 
 		boolean isExistsTable(TreeName treeName) {
 			final String tableName = getTableName(treeName);
+			// the catalog lookup guarding a create table is bounded as the operation it is, not as
+			// the bulk statement it guards: it asks a data dictionary rather than doing work of
+			// its own, so a wait here is the metadata lock of another session
 			try {
-				final DatabaseMetaData metaData = con.getMetaData();
-				// asked of the catalog by name: openTree(createOnDemand) calls this for every tree
-				// of the backend - about 25 of them for a stock suffix, on every open - and listing
-				// every table of the database each time costs the whole catalog once per tree, on a
-				// database this backend may well be sharing with something else
-				try (final ResultSet rs = metaData.getTables(null, null,
-						storedIdentifier(metaData, tableName), new String[]{"TABLE"})) {
-					while (rs.next()) {
-						// the name still has to be compared: "_" is a single-character wildcard in a
-						// metadata pattern, so "opendj_<hash>" also matches a table named "opendjX<hash>"
-						if (tableName.equalsIgnoreCase(rs.getString("TABLE_NAME"))) {
-							return true;
+				return bounded(con, StatementBound.OPERATION, () -> {
+					final DatabaseMetaData metaData = con.getMetaData();
+					// asked of the catalog by name: openTree(createOnDemand) calls this for every tree
+					// of the backend - about 25 of them for a stock suffix, on every open - and listing
+					// every table of the database each time costs the whole catalog once per tree, on a
+					// database this backend may well be sharing with something else
+					try (final ResultSet rs = metaData.getTables(null, null,
+							storedIdentifier(metaData, tableName), new String[]{"TABLE"})) {
+						while (rs.next()) {
+							// the name still has to be compared: "_" is a single-character wildcard in a
+							// metadata pattern, so "opendj_<hash>" also matches a table named "opendjX<hash>"
+							if (tableName.equalsIgnoreCase(rs.getString("TABLE_NAME"))) {
+								return true;
+							}
 						}
 					}
-				}
+					return false;
+				});
 			} catch (Exception e) {
 				throw new StorageRuntimeException(e);
 			}
-			return false;
 		}
 
 		String getTableDialect() {
@@ -1098,7 +1522,7 @@ public void openTree(TreeName treeName, boolean createOnDemand) {
 			if (createOnDemand) {
 				if (!isExistsTable(treeName)) {
 					try (final PreparedStatement statement=con.prepareStatement("create table "+getTableName(treeName)+" ("+getTableDialect()+")")){
-						execute(statement);
+						execute(statement, StatementBound.BULK);
 						con.commit();
 					}catch (SQLException e) {
 						throw new StorageRuntimeException(e);
@@ -1109,7 +1533,7 @@ public void openTree(TreeName treeName, boolean createOnDemand) {
 				final String tableName=getTableName(treeName);
 				if (driverName.contains("postgres")) {
 					try (final PreparedStatement statement=con.prepareStatement("create index if not exists k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){
-						execute(statement);
+						execute(statement, StatementBound.BULK);
 						con.commit();
 					}catch (SQLException e) {
 						throw new StorageRuntimeException(e);
@@ -1118,7 +1542,7 @@ public void openTree(TreeName treeName, boolean createOnDemand) {
 					try {
 						if (!isExistsIndex(tableName,"k_"+tableName.substring("opendj_".length()))) { // mysql has no "create index if not exists"
 							try (final PreparedStatement statement=con.prepareStatement("create index k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){
-								execute(statement);
+								execute(statement, StatementBound.BULK);
 								con.commit();
 							}
 						}
@@ -1130,7 +1554,7 @@ public void openTree(TreeName treeName, boolean createOnDemand) {
 						// oracle has no "create index if not exists"; unquoted identifiers are stored in uppercase
 						if (!isExistsIndex(tableName.toUpperCase(),"k_"+tableName.substring("opendj_".length()))) {
 							try (final PreparedStatement statement=con.prepareStatement("create index k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){
-								execute(statement);
+								execute(statement, StatementBound.BULK);
 								con.commit();
 							}
 						}
@@ -1146,20 +1570,22 @@ public void openTree(TreeName treeName, boolean createOnDemand) {
 		}
 
 		boolean isExistsIndex(String tableName, String indexName) throws SQLException {
-			// approximate=true: with false the oracle driver runs ANALYZE on every call
-			try (final ResultSet rs = con.getMetaData().getIndexInfo(null, null, tableName, false, true)) {
-				while (rs.next()) {
-					if (indexName.equalsIgnoreCase(rs.getString("INDEX_NAME"))) {
-						return true;
+			return bounded(con, StatementBound.OPERATION, () -> {
+				// approximate=true: with false the oracle driver runs ANALYZE on every call
+				try (final ResultSet rs = con.getMetaData().getIndexInfo(null, null, tableName, false, true)) {
+					while (rs.next()) {
+						if (indexName.equalsIgnoreCase(rs.getString("INDEX_NAME"))) {
+							return true;
+						}
 					}
 				}
-			}
-			return false;
+				return false;
+			});
 		}
 		
 		public void clearTree(TreeName treeName) {
 			try (final PreparedStatement statement=con.prepareStatement("delete from "+getTableName(treeName))){
-				execute(statement);
+				execute(statement, StatementBound.BULK);
 				con.commit();
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
@@ -1170,7 +1596,7 @@ public void clearTree(TreeName treeName) {
 		public void deleteTree(TreeName treeName) {
 			if (isExistsTable(treeName)) {
 				try (final PreparedStatement statement = con.prepareStatement("drop table " + getTableName(treeName))) {
-					execute(statement);
+					execute(statement, StatementBound.BULK);
 					con.commit();
 				} catch (SQLException e) {
 					throw new StorageRuntimeException(e);
@@ -1200,28 +1626,28 @@ boolean upsert(TreeName treeName, ByteSequence key, ByteSequence value) throws S
 					statement.setString(1, key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 					statement.setBytes(2, real2db(key.toByteArray()));
 					statement.setBytes(3, value.toByteArray());
-					return (execute(statement) == 1 && statement.getUpdateCount() > 0);
+					return (execute(statement, bound) == 1 && statement.getUpdateCount() > 0);
 				}
 			}else if (driverName.contains("mysql")) { //mysql upsert
 				try (final PreparedStatement statement = con.prepareStatement("insert into " + getTableName(treeName) + " (h,k,v) values (?,?,?) as new ON DUPLICATE KEY UPDATE v=new.v")) {
 					statement.setString(1, key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 					statement.setBytes(2, real2db(key.toByteArray()));
 					statement.setBytes(3, value.toByteArray());
-					return (execute(statement) == 1 && statement.getUpdateCount() > 0);
+					return (execute(statement, bound) == 1 && statement.getUpdateCount() > 0);
 				}
 			}else if (driverName.contains("oracle")) { //ANSI MERGE without ;
 				try (final PreparedStatement statement = con.prepareStatement("merge into " + getTableName(treeName) + " old using (select ? h,? k,? v from dual) new on (old.h=new.h and old.k=new.k) WHEN MATCHED THEN UPDATE SET old.v=new.v WHEN NOT MATCHED THEN INSERT (h,k,v) VALUES (new.h,new.k,new.v)")) {
 					statement.setString(1, key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 					statement.setBytes(2, real2db(key.toByteArray()));
 					statement.setBytes(3, value.toByteArray());
-					return (execute(statement) == 1 && statement.getUpdateCount() > 0);
+					return (execute(statement, bound) == 1 && statement.getUpdateCount() > 0);
 				}
 			}else if (driverName.contains("microsoft")) { //ANSI MERGE with ; WITH (HOLDLOCK) makes the upsert atomic: without it SQL Server MERGE can race two concurrent NOT MATCHED inserts of the same key into a PRIMARY KEY violation. UPDLOCK is required on top of it: with HOLDLOCK alone the search phase takes a shared lock that the WHEN MATCHED update then has to convert to an exclusive one, so two concurrent upserts of the same key deadlock on the conversion; an update lock is taken right away and makes the second transaction wait instead. h is cast back to char so that the join can seek the primary key instead of scanning the whole table under those locks, see hashParam()
 				try (final PreparedStatement statement = con.prepareStatement("merge into " + getTableName(treeName) + " WITH (HOLDLOCK, UPDLOCK) old using (select cast(? as char(128)) h,? k,? v) new on (old.h=new.h and old.k=new.k) WHEN MATCHED THEN UPDATE SET old.v=new.v WHEN NOT MATCHED THEN INSERT (h,k,v) VALUES (new.h,new.k,new.v);")) {
 					statement.setString(1, key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 					statement.setBytes(2, real2db(key.toByteArray()));
 					statement.setBytes(3, value.toByteArray());
-					return (execute(statement) == 1 && statement.getUpdateCount() > 0);
+					return (execute(statement, bound) == 1 && statement.getUpdateCount() > 0);
 				}
 			}else { //ANSI SQL: try update before insert with not exists
 				return update(treeName,key,value) || insert(treeName,key,value);
@@ -1235,7 +1661,7 @@ boolean insert(TreeName treeName, ByteSequence key, ByteSequence value) throws S
 				statement.setBytes(3, value.toByteArray());
 				statement.setString(4, key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 				statement.setBytes(5, real2db(key.toByteArray()));
-				return (execute(statement)==1 && statement.getUpdateCount()>0);
+				return (execute(statement, bound)==1 && statement.getUpdateCount()>0);
 			}
 		}
 
@@ -1244,7 +1670,7 @@ boolean update(TreeName treeName, ByteSequence key, ByteSequence value) throws S
 				statement.setBytes(1,value.toByteArray());
 				statement.setString(2,key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 				statement.setBytes(3,real2db(key.toByteArray()));
-				return (execute(statement)==1 && statement.getUpdateCount()>0);
+				return (execute(statement, bound)==1 && statement.getUpdateCount()>0);
 			}
 		}
 
@@ -1269,7 +1695,7 @@ public boolean delete(TreeName treeName, ByteSequence key) {
 			try (final PreparedStatement statement=con.prepareStatement("delete from "+getTableName(treeName)+" where h="+hashParam(con)+" and k=?")){
 				statement.setString(1,key2hash.get(ByteBuffer.wrap(key.toByteArray())));
 				statement.setBytes(2,real2db(key.toByteArray()));
-				return (execute(statement)==1 && statement.getUpdateCount()>0);
+				return (execute(statement, bound)==1 && statement.getUpdateCount()>0);
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
 			}
@@ -1301,10 +1727,18 @@ final class CursorImpl implements Cursor<ByteString, ByteString> {
 		ByteString currentValue;
 		boolean defined;
 
-		public CursorImpl(boolean isReadOnly, Connection con, TreeName treeName) {
+		// The class of the statements this cursor issues, from whoever opened it: a search walks its
+		// index and has a client waiting, while an import, an export or a rebuild walks a whole tree
+		// with nobody waiting - and on mssql it walks it unindexed either way. It is not read off the
+		// shape of the statement, because the opening batch of every cursor is the same
+		// unconditioned "order by k" that positionToLastKey() issues, search or not.
+		final StatementBound batchBound;
+
+		public CursorImpl(boolean isReadOnly, Connection con, TreeName treeName, StatementBound batchBound) {
 			this.isReadOnly=isReadOnly;
 			this.con=con;
 			this.tableName=getTableName(treeName);
+			this.batchBound=batchBound;
 			this.limitClause=((CachedConnection)con).parent.getClass().getName().contains("mysql")
 				? " limit ?,?" : " offset ? rows fetch next ? rows only";
 		}
@@ -1315,7 +1749,14 @@ int adaptiveBatchSize() {
 			return size;
 		}
 
-		boolean fetchBatch(String condition, byte[] dbKey, long offset, boolean descending, int limit) {
+		/**
+		 * Reads one batch of the cursor. The class of the bound is the caller's: a batch taken
+		 * along the index of the tree for a client is an operation, while a batch that has to look
+		 * at the whole table to answer - the one behind {@link #positionToLastKey()} - and every
+		 * batch of a cursor an import or a rebuild walks ({@link #batchBound}) is bulk work, and
+		 * the two cannot share a value.
+		 */
+		boolean fetchBatch(String condition, byte[] dbKey, long offset, boolean descending, int limit, StatementBound bound) {
 			fetchCount++;
 			buffer.clear();
 			try (final PreparedStatement statement=con.prepareStatement("select k,v from "+tableName
@@ -1327,15 +1768,15 @@ boolean fetchBatch(String condition, byte[] dbKey, long offset, boolean descendi
 				}
 				statement.setLong(i++,offset);
 				statement.setLong(i,limit);
-				try(final ResultSet rc=executeResultSet(statement)) {
+				return executeResultSet(statement, bound, rc -> {
 					while (rc.next()) {
 						buffer.add(new byte[][]{rc.getBytes(1),rc.getBytes(2)});
 					}
-				}
+					return !buffer.isEmpty();
+				});
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
 			}
-			return !buffer.isEmpty();
 		}
 
 		void advanceFromBuffer() {
@@ -1348,7 +1789,7 @@ void advanceFromBuffer() {
 
 		@Override
 		public boolean next() {
-			if (buffer.isEmpty() && !fetchBatch(currentKeyDb==null?null:">",currentKeyDb,0,false,adaptiveBatchSize())) {
+			if (buffer.isEmpty() && !fetchBatch(currentKeyDb==null?null:">",currentKeyDb,0,false,adaptiveBatchSize(),batchBound)) {
 				defined=false;
 				return false;
 			}
@@ -1388,7 +1829,7 @@ public void delete() throws NoSuchElementException, UnsupportedOperationExceptio
 			try (final PreparedStatement statement=con.prepareStatement("delete from "+tableName+" where h="+hashParam(con)+" and k=?")){
 				statement.setString(1,key2hash.get(ByteBuffer.wrap(db2real(currentKeyDb))));
 				statement.setBytes(2,currentKeyDb);
-				execute(statement);
+				execute(statement, batchBound);
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
 			}
@@ -1418,7 +1859,7 @@ && compareKeys(target,buffer.peekLast()[0])<=0) {
 			if (!buffer.isEmpty()) { // jumped outside the buffered range: random access, back to small batches
 				nextBatchSize=initialBatchSize;
 			}
-			if (fetchBatch(">=",target,0,false,adaptiveBatchSize())) {
+			if (fetchBatch(">=",target,0,false,adaptiveBatchSize(),batchBound)) {
 				advanceFromBuffer();
 				return true;
 			}
@@ -1429,30 +1870,38 @@ && compareKeys(target,buffer.peekLast()[0])<=0) {
 		@Override
 		public boolean positionToKey(ByteSequence key) {
 			final byte[] real=key.toByteArray();
+			final byte[] value;
 			try (final PreparedStatement statement=con.prepareStatement("select v from "+tableName+" where h="+hashParam(con)+" and k=?")){
 				statement.setString(1,key2hash.get(ByteBuffer.wrap(real)));
 				statement.setBytes(2,real2db(real));
-				try(final ResultSet rc=executeResultSet(statement)) {
-					if (rc.next()) {
-						buffer.clear();
-						nextBatchSize=initialBatchSize;
-						currentKeyDb=real2db(real);
-						currentKey=ByteString.wrap(real);
-						currentValue=ByteString.wrap(rc.getBytes("v"));
-						defined=true;
-						return true;
-					}
-				}
+				value=executeResultSet(statement, batchBound, rc -> rc.next() ? rc.getBytes("v") : null);
 			}catch (SQLException e) {
 				throw new StorageRuntimeException(e);
 			}
+			if (value!=null) {
+				buffer.clear();
+				nextBatchSize=initialBatchSize;
+				currentKeyDb=real2db(real);
+				currentKey=ByteString.wrap(real);
+				currentValue=ByteString.wrap(value);
+				defined=true;
+				return true;
+			}
 			defined=false;
 			return false;
 		}
 
+		/**
+		 * Bulk, not operation: with no condition to seek on, this is {@code order by k desc} over
+		 * the whole table - and on mssql, where {@code k} is a {@code varbinary(max)} that cannot
+		 * be an index key, a scan and a sort of it. It is also not on a search path: every open of
+		 * a backend runs it once per base DN, through {@code EntryContainer.getHighestEntryID()},
+		 * outside the try/catch of {@code BackendImpl.openBackend()} - a bound of two minutes here
+		 * would turn a large backend that opens slowly into one that does not open at all.
+		 */
 		@Override
 		public boolean positionToLastKey() {
-			if (fetchBatch(null,null,0,true,1)) {
+			if (fetchBatch(null,null,0,true,1,StatementBound.BULK)) {
 				advanceFromBuffer();
 				return true;
 			}
@@ -1460,12 +1909,20 @@ public boolean positionToLastKey() {
 			return false;
 		}
 
+		/**
+		 * The class of the cursor, unlike {@link #positionToLastKey()}, which is bulk however it
+		 * was opened: an offset comes from the VLV request of a client, so this runs on a search
+		 * path and has to give the worker thread back - an import has no VLV position to seek to.
+		 * That a deep offset is served by walking to it - the engines have no other way to answer
+		 * an {@code offset ?} - is what makes the bound reachable here, and reaching it answers the
+		 * request with an error rather than parking a thread of the server on it.
+		 */
 		@Override
 		public boolean positionToIndex(int index) {
 			if (!buffer.isEmpty()) { // absolute jump: random access, back to small batches
 				nextBatchSize=initialBatchSize;
 			}
-			if (index>=0 && fetchBatch(null,null,index,false,adaptiveBatchSize())) {
+			if (index>=0 && fetchBatch(null,null,index,false,adaptiveBatchSize(),batchBound)) {
 				advanceFromBuffer();
 				return true;
 			}
@@ -1479,7 +1936,7 @@ public Set<TreeName> listTrees() {
 		return tree2table.asMap().keySet();
 	}
 
-	private final class ImporterImpl implements Importer {
+	final class ImporterImpl implements Importer {
 		final Connection con;
 		final ReadableTransactionImpl txr;
 		final WriteableTransactionTransactionImpl txw;
@@ -1499,22 +1956,21 @@ private final class ImporterImpl implements Importer {
 
 		final Boolean isOpen;
 
-		public ImporterImpl() {
-			isOpen=getStorageStatus().isWorking();
-			if (!isOpen) {
-				try {
-					open(AccessMode.READ_WRITE);
-				}catch (Exception e) {
-					throw new StorageRuntimeException(e);
-				}
-			}
-			try {
-				con = getConnection();
-			}catch (Exception e){
-				throw new StorageRuntimeException(e);
-			}
-			txr =new ReadableTransactionImpl(con);
-			txw =new WriteableTransactionTransactionImpl(con);
+		/**
+		 * Both transactions of an import take the bulk class, and with them every statement it
+		 * issues: phase one writes the trees through {@code put()}, phase two reads them back
+		 * through {@code read()} and walks them through {@code openCursor()}, and none of that has
+		 * a client waiting on it. Bounding those as entry reads is not merely strict, it fails work
+		 * that ran to the end before this bound existed: {@code h} is the primary key on every
+		 * dialect and the default lock wait is forever on mssql, postgres and oracle, so an upsert
+		 * of an online import blocked by an LDAP write on the same table sat until the bound of an
+		 * entry read and then failed the import.
+		 */
+		ImporterImpl(Connection con, boolean isOpen) {
+			this.con=con;
+			this.isOpen=isOpen;
+			txr=new ReadableTransactionImpl(con, StatementBound.BULK);
+			txw=new WriteableTransactionTransactionImpl(con, StatementBound.BULK);
 		}
 		
 		@Override
@@ -1522,6 +1978,11 @@ public void aborted() {
 			aborted = true;
 		}
 
+		// The connection goes back whatever the commit does, and the storage this importer opened
+		// is closed whatever the connection does: an importer is closed on the way out of a failed
+		// import as readily as a finished one - a clearTree() that reaches the bulk bound is one
+		// way there - and a commit that throws on the way would otherwise leave the connection
+		// out of the pool for good, holding the transaction and the locks of that import.
 		@Override
 		public void close() {
 			try {
@@ -1565,6 +2026,11 @@ public ByteString read(TreeName treeName, ByteSequence key) {
 			return txr.read(treeName, key);
 		}
 		
+		// Bulk like every other statement of an import, by the class of the transaction it comes
+		// from: this walks a whole tree with no client waiting on it - phase one of a rebuild-index
+		// reads every record of id2entry through this cursor (OnDiskMergeImporter.ID2EntrySource) -
+		// and on mssql it walks it unindexed, so a batch of it is a scan and a sort of the table
+		// rather than a step along an index.
 		@Override
 		public SequentialCursor<ByteString, ByteString> openCursor(TreeName treeName) {
 			return txr.openCursor(treeName);
@@ -1574,7 +2040,23 @@ public SequentialCursor<ByteString, ByteString> openCursor(TreeName treeName) {
 	//import
 	@Override
 	public Importer startImport() throws ConfigException, StorageRuntimeException {
-		return new ImporterImpl();
+		final boolean wasOpen=getStorageStatus().isWorking();
+		if (!wasOpen) {
+			try {
+				open(AccessMode.READ_WRITE);
+			}catch (Exception e) {
+				throw new StorageRuntimeException(e);
+			}
+		}
+		final Connection con;
+		try {
+			con=getConnection();
+		}catch (Exception e){
+			throw new StorageRuntimeException(e);
+		}
+		// outside the catch: a transaction of a read-only storage throws ReadOnlyStorageException,
+		// which a caller tells apart from any other failure of an import
+		return new ImporterImpl(con, wasOpen);
 	}
 	
 	//backup
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendStat.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendStat.java
index 694edba973..c3218ab235 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendStat.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/BackendStat.java
@@ -1118,7 +1118,7 @@ public Void run(ReadableTransaction txn) throws Exception
           long undefined = 0;
           long count = 0;
           BackendTreeKeyValue keyDecoder = new BackendTreeKeyValue(index);
-          try (Cursor<ByteString, EntryIDSet> cursor = index.openCursor(txn))
+          try (Cursor<ByteString, EntryIDSet> cursor = index.openBulkCursor(txn))
           {
             while (cursor.next())
             {
@@ -1286,7 +1286,8 @@ public TreeStats run(ReadableTransaction txn) throws Exception
           long count = 0;
           long totalKeySize = 0;
           long totalDataSize = 0;
-          try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(target.getTreeName()))
+          // dbtest walks the tree whole, on the command line of an operator: bulk work either way
+          try (final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(target.getTreeName()))
           {
             ByteString key;
             ByteString maxKey = null;
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/DefaultIndex.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/DefaultIndex.java
index 99faae056f..88edf18c5c 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/DefaultIndex.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/DefaultIndex.java
@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2010 Sun Microsystems, Inc.
  * Portions Copyright 2012-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.backends.pluggable;
 
@@ -131,7 +132,19 @@ public String valueToString(ByteString value)
   public final Cursor<ByteString, EntryIDSet> openCursor(ReadableTransaction txn)
   {
     checkNotNull(txn, "txn must not be null");
-    return CursorTransformer.transformValues(txn.openCursor(getName()),
+    return decoding(txn.openCursor(getName()));
+  }
+
+  @Override
+  public final Cursor<ByteString, EntryIDSet> openBulkCursor(ReadableTransaction txn)
+  {
+    checkNotNull(txn, "txn must not be null");
+    return decoding(txn.openBulkCursor(getName()));
+  }
+
+  private Cursor<ByteString, EntryIDSet> decoding(Cursor<ByteString, ByteString> cursor)
+  {
+    return CursorTransformer.transformValues(cursor,
         new ValueTransformer<ByteString, ByteString, EntryIDSet, NeverThrowsException>()
         {
           @Override
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ExportJob.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ExportJob.java
index aedf099658..767aba44bb 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ExportJob.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ExportJob.java
@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2008 Sun Microsystems, Inc.
  * Portions Copyright 2012-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.backends.pluggable;
 
@@ -172,7 +173,10 @@ private void exportContainer(ReadableTransaction txn, EntryContainer entryContai
        throws StorageRuntimeException, IOException, LDIFException
   {
     ID2Entry id2entry = entryContainer.getID2Entry();
-    try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(id2entry.getName()))
+    // The whole of id2entry with nobody waiting on the walk: an export-ldif, or the generation ID
+    // a replicated domain computes for itself the first time it starts (LDAPReplicationDomain
+    // .computeGenerationId), which is why this must not be bounded as the work of an operation.
+    try (final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(id2entry.getName()))
     {
       while (cursor.next())
       {
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2ChildrenCount.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2ChildrenCount.java
index 812300f723..efeaf3b8cd 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2ChildrenCount.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2ChildrenCount.java
@@ -63,6 +63,13 @@ SequentialCursor<EntryID, Void> openCursor(ReadableTransaction txn)
         TO_ENTRY_ID, CursorTransformer.<ByteString, Void> keepValuesUnchanged());
   }
 
+  /** @see ReadableTransaction#openBulkCursor(TreeName) */
+  SequentialCursor<EntryID, Void> openBulkCursor(ReadableTransaction txn)
+  {
+    return transformKeysAndValues(counter.openBulkCursor(txn),
+        TO_ENTRY_ID, CursorTransformer.<ByteString, Void> keepValuesUnchanged());
+  }
+
   /**
    * Updates the number of children for a given entry without updating the total number of entries.
    * <p>
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2Entry.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2Entry.java
index 72bbded077..9347029659 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2Entry.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ID2Entry.java
@@ -382,7 +382,11 @@ void afterOpen(WriteableTransaction txn, boolean createOnDemand) throws StorageR
   {
     // Make sure the tree is there and readable, even if the storage is READ_ONLY.
     // Would be nice if there were a better way...
-    try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(getName()))
+    // Bulk: the first batch of a cursor carries no seek predicate, so this is a walk of the whole
+    // tree as far as the storage is concerned, and it runs on every open of the backend. A bound
+    // meant for an entry read would keep a large backend from opening at all on an engine where
+    // such a batch is not a step along an index.
+    try (final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(getName()))
     {
       cursor.next();
     }
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/Index.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/Index.java
index bbaf6b54b0..d11420a33f 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/Index.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/Index.java
@@ -13,6 +13,7 @@
  *
  * Copyright 2006-2010 Sun Microsystems, Inc.
  * Portions Copyright 2012-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.backends.pluggable;
 
@@ -37,6 +38,17 @@ interface Index extends Tree
 
   Cursor<ByteString, EntryIDSet> openCursor(ReadableTransaction txn);
 
+  /**
+   * Opens a cursor over the whole index for a task no client operation is waiting on, such as
+   * {@code verify-index} or {@code dbtest}.
+   *
+   * @param txn
+   *          the transaction to read the index with
+   * @return a cursor over every key of this index
+   * @see ReadableTransaction#openBulkCursor(org.opends.server.backends.pluggable.spi.TreeName)
+   */
+  Cursor<ByteString, EntryIDSet> openBulkCursor(ReadableTransaction txn);
+
   boolean setIndexEntryLimit(int indexEntryLimit);
 
   boolean setConfidential(boolean indexConfidential);
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/PersistentCompressedSchema.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/PersistentCompressedSchema.java
index af484abb73..2077b29e40 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/PersistentCompressedSchema.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/PersistentCompressedSchema.java
@@ -13,6 +13,7 @@
  *
  * Copyright 2008-2009 Sun Microsystems, Inc.
  * Portions Copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.backends.pluggable;
 
@@ -146,7 +147,8 @@ private void load(WriteableTransaction txn, boolean shouldCreate)
     // Cursor through the object class database and load the object class set
     // definitions. At the same time, figure out the highest token value and
     // initialize the object class counter to one greater than that.
-    try (Cursor<ByteString, ByteString> ocCursor = txn.openCursor(ocTreeName))
+    // Both trees are read whole while the backend opens, with no client operation waiting on it.
+    try (Cursor<ByteString, ByteString> ocCursor = txn.openBulkCursor(ocTreeName))
     {
       while (ocCursor.next())
       {
@@ -169,7 +171,7 @@ private void load(WriteableTransaction txn, boolean shouldCreate)
     }
 
     // Cursor through the attribute description database and load the attribute set definitions.
-    try (Cursor<ByteString, ByteString> adCursor = txn.openCursor(adTreeName))
+    try (Cursor<ByteString, ByteString> adCursor = txn.openBulkCursor(adTreeName))
     {
       while (adCursor.next())
       {
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ShardedCounter.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ShardedCounter.java
index efde5df428..47cfbac68c 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ShardedCounter.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/ShardedCounter.java
@@ -12,6 +12,7 @@
  * information: "Portions Copyright [year] [name of copyright owner]".
  *
  * Copyright 2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.backends.pluggable;
 
@@ -73,9 +74,20 @@ public ByteString apply(ByteString shardedKey)
   }
 
   SequentialCursor<ByteString, Void> openCursor(ReadableTransaction txn)
+  {
+    return uniqueKeys(txn.openCursor(getName()));
+  }
+
+  /** @see ReadableTransaction#openBulkCursor(TreeName) */
+  SequentialCursor<ByteString, Void> openBulkCursor(ReadableTransaction txn)
+  {
+    return uniqueKeys(txn.openBulkCursor(getName()));
+  }
+
+  private SequentialCursor<ByteString, Void> uniqueKeys(Cursor<ByteString, ByteString> cursor)
   {
     return new UniqueKeysCursor<>(transformKeysAndValues(
-        txn.openCursor(getName()), TO_KEY,
+        cursor, TO_KEY,
         CursorTransformer.<ByteString, ByteString, Void> constant(null)));
   }
 
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/TracedStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/TracedStorage.java
index 172388ef62..470dc5b5f2 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/TracedStorage.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/TracedStorage.java
@@ -296,6 +296,15 @@ public Cursor<ByteString, ByteString> openCursor(final TreeName name)
       return new TracedCursor(cursor);
     }
 
+    @Override
+    public Cursor<ByteString, ByteString> openBulkCursor(final TreeName name)
+    {
+      traceEnter("openBulkCursor", "name", name);
+      final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(name);
+      traceLeave("openBulkCursor", "name", name);
+      return new TracedCursor(cursor);
+    }
+
     @Override
     public ByteString read(final TreeName name, final ByteSequence key)
     {
@@ -374,6 +383,15 @@ public Cursor<ByteString, ByteString> openCursor(final TreeName name)
       return new TracedCursor(cursor);
     }
 
+    @Override
+    public Cursor<ByteString, ByteString> openBulkCursor(final TreeName name)
+    {
+      traceEnter("openBulkCursor", "name", name);
+      final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(name);
+      traceLeave("openBulkCursor", "name", name);
+      return new TracedCursor(cursor);
+    }
+
     @Override
     public void openTree(final TreeName name, boolean createOnDemand)
     {
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/VerifyJob.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/VerifyJob.java
index 6784d2482d..850a36011f 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/VerifyJob.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/VerifyJob.java
@@ -345,7 +345,8 @@ else if(lowerName.startsWith("vlv."))
    */
   private void iterateID2Entry(ReadableTransaction txn) throws StorageRuntimeException
   {
-    try(final Cursor<ByteString, ByteString> cursor = txn.openCursor(id2entry.getName()))
+    // Every tree this job walks, it walks whole, and no client operation is waiting on it.
+    try(final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(id2entry.getName()))
     {
       long storedEntryCount = id2entry.getRecordCount(txn);
       while (cursor.next())
@@ -442,7 +443,7 @@ private void iterateDN2ID(ReadableTransaction txn) throws StorageRuntimeExceptio
     final Deque<ChildrenCount> childrenCounters = new LinkedList<>();
     ChildrenCount currentNode = null;
 
-    try(final Cursor<ByteString, ByteString> cursor = txn.openCursor(dn2id.getName()))
+    try(final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(dn2id.getName()))
     {
       while (cursor.next())
       {
@@ -525,7 +526,7 @@ private void verifyID2ChildrenCount(ReadableTransaction txn, ChildrenCount paren
 
   private void iterateID2ChildrenCount(ReadableTransaction txn) throws StorageRuntimeException
   {
-    try (final SequentialCursor<EntryID, Void> cursor = id2childrenCount.openCursor(txn))
+    try (final SequentialCursor<EntryID, Void> cursor = id2childrenCount.openBulkCursor(txn))
     {
       while (cursor.next())
       {
@@ -607,7 +608,7 @@ private void iterateVLVIndex(ReadableTransaction txn, VLVIndex vlvIndex, boolean
       return;
     }
 
-    try(final Cursor<ByteString, ByteString> cursor = txn.openCursor(vlvIndex.getName()))
+    try(final Cursor<ByteString, ByteString> cursor = txn.openBulkCursor(vlvIndex.getName()))
     {
       while (cursor.next())
       {
@@ -655,7 +656,7 @@ private void iterateAttrIndex(ReadableTransaction txn, MatchingRuleIndex index)
       return;
     }
 
-    try(final Cursor<ByteString,EntryIDSet> cursor = index.openCursor(txn))
+    try(final Cursor<ByteString,EntryIDSet> cursor = index.openBulkCursor(txn))
     {
       while (cursor.next())
       {
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/spi/ReadableTransaction.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/spi/ReadableTransaction.java
index 1f9ab9e02d..04cdcef1d7 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/spi/ReadableTransaction.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/spi/ReadableTransaction.java
@@ -12,6 +12,7 @@
  * information: "Portions Copyright [year] [name of copyright owner]".
  *
  * Copyright 2014-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
  */
 package org.opends.server.backends.pluggable.spi;
 
@@ -43,6 +44,26 @@ public interface ReadableTransaction
    */
   Cursor<ByteString, ByteString> openCursor(TreeName treeName);
 
+  /**
+   * Opens a cursor on the tree whose name is provided, for a walk of that whole tree with no
+   * client operation waiting on it: an export, a verify, a rebuild, or the load of a tree while
+   * the backend opens.
+   * <p>
+   * A storage engine that bounds how long a statement may take must not bound such a walk as it
+   * bounds the work of a client operation: what this legitimately takes follows the size of the
+   * tree, and cutting it short fails an administrative task that would otherwise have run to the
+   * end. An engine with no such bound - every one but the JDBC backend - answers this exactly as
+   * {@link #openCursor(TreeName)} does.
+   *
+   * @param treeName
+   *          the tree name
+   * @return a new cursor
+   */
+  default Cursor<ByteString, ByteString> openBulkCursor(TreeName treeName)
+  {
+    return openCursor(treeName);
+  }
+
   /**
    * Returns the number of key/value pairs in the provided tree.
    *
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java
new file mode 100644
index 0000000000..618565719b
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java
@@ -0,0 +1,721 @@
+/*
+ * 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.ldap.ByteString;
+import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
+import org.mockito.InOrder;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.opends.server.DirectoryServerTestCase;
+import org.opends.server.backends.jdbc.JDBCStorage.StatementBound;
+import org.opends.server.backends.pluggable.spi.AccessMode;
+import org.opends.server.backends.pluggable.spi.TreeName;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.SQLTimeoutException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+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.doThrow;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.inOrder;
+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.mockito.Mockito.times;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+/**
+ * Which bound a statement of the JDBC backend is given, and what reaching it looks like to the
+ * caller (#877). Needs no database: the statement is a mock, so the policy is pinned wherever the
+ * build runs, while the container suites cover a statement really blocked on a lock.
+ */
+@SuppressWarnings("javadoc")
+// timeOut on the class, not on the waits inside a test: a statement of another thread that never
+// arrives has to fail this suite rather than hang the build waiting for it.
+@Test(groups = { "precommit", "jdbc" }, sequential = true, timeOut = 120000)
+public class JDBCStatementBoundTestCase extends DirectoryServerTestCase {
+
+	private JDBCStorage storage;
+
+	@BeforeClass
+	public void createStorage() {
+		storage = new JDBCStorage(mockCfg(JDBCBackendCfg.class), null);
+	}
+
+	@AfterMethod
+	public void clearProperties() {
+		for (final StatementBound bound : StatementBound.values()) {
+			System.clearProperty(bound.property);
+		}
+		System.clearProperty(JDBCStorage.STATISTICS_TIMEOUT_PROPERTY);
+		storage.accessMode = AccessMode.READ_ONLY; // an import test opens it for writing
+	}
+
+	/** How long a test waits for a statement running on another thread before it fails. */
+	private static final long WAIT_MILLIS = 30000;
+
+	private static void awaitOrFail(CountDownLatch latch, String what) throws InterruptedException {
+		assertTrue(latch.await(WAIT_MILLIS, TimeUnit.MILLISECONDS), what + " within " + WAIT_MILLIS + " ms");
+	}
+
+	private static void sleep(long millis) {
+		try {
+			Thread.sleep(millis);
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+	}
+
+	/** A statement that reports it is running and then waits for the test to let it finish. */
+	private PreparedStatement lingering(Connection con, CountDownLatch running, CountDownLatch mayFinish)
+			throws SQLException {
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+		when(statement.executeUpdate()).thenAnswer(new Answer<Integer>() {
+			@Override
+			public Integer answer(InvocationOnMock invocation) throws Throwable {
+				running.countDown();
+				awaitOrFail(mayFinish, "the statement was never let go");
+				return 1;
+			}
+		});
+		return statement;
+	}
+
+	private interface Execution {
+		void run() throws Exception;
+	}
+
+	/**
+	 * A statement running on a thread of its own, with whatever it threw kept for the assertion:
+	 * {@code Thread.join()} does not rethrow, so a failure in the background would otherwise leave
+	 * the verifications of a test passing on a run that never reached the state they check.
+	 */
+	private static final class Background {
+		final Thread thread;
+		final AtomicReference<Throwable> failure = new AtomicReference<>();
+
+		Background(String name, final Execution execution) {
+			thread = new Thread(new Runnable() {
+				@Override
+				public void run() {
+					try {
+						execution.run();
+					} catch (Throwable t) {
+						failure.set(t);
+					}
+				}
+			}, name);
+		}
+
+		void joinOrFail() throws Exception {
+			thread.join(WAIT_MILLIS);
+			assertFalse(thread.isAlive(), thread.getName() + " did not finish within " + WAIT_MILLIS + " ms");
+			final Throwable thrown = failure.get();
+			if (thrown instanceof Exception) {
+				throw (Exception) thrown;
+			}
+			if (thrown != null) {
+				throw new AssertionError(thrown);
+			}
+		}
+	}
+
+	private static Background start(String name, Execution execution) {
+		final Background background = new Background(name, execution);
+		background.thread.start();
+		return background;
+	}
+
+	/**
+	 * An entry read that has not come back in two minutes is stuck, while a count or the delete
+	 * that empties a tree before an import legitimately takes longer than anything can guess - so
+	 * the bulk class stays unbounded until a deployment says otherwise.
+	 */
+	@Test
+	public void testDefaultsBoundAnOperationAndLeaveBulkAlone() throws Exception {
+		assertEquals(StatementBound.OPERATION.seconds(), 120);
+		assertEquals(StatementBound.BULK.seconds(), 0);
+	}
+
+	@Test
+	public void testEachClassIsConfiguredByItsOwnProperty() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		assertEquals(StatementBound.OPERATION.seconds(), 7);
+		assertEquals(StatementBound.BULK.seconds(), 0, "the bulk class followed the operation one");
+
+		System.setProperty(StatementBound.BULK.property, "900");
+		assertEquals(StatementBound.BULK.seconds(), 900);
+		assertEquals(StatementBound.OPERATION.seconds(), 7);
+	}
+
+	@Test
+	public void testAValueThatIsNoBoundLeavesTheStatementUnbounded() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "0");
+		assertEquals(StatementBound.OPERATION.seconds(), 0);
+
+		System.setProperty(StatementBound.OPERATION.property, "-1");
+		assertEquals(StatementBound.OPERATION.seconds(), 0);
+	}
+
+	/**
+	 * A value that is not a number is not a way to switch the bound off: it is ignored in favour
+	 * of the default, as {@code Integer.getInteger()} has it, so a typo leaves the class bounded
+	 * rather than silently unbounding it.
+	 */
+	@Test
+	public void testAValueThatIsNotANumberFallsBackToTheDefault() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "two minutes");
+		assertEquals(StatementBound.OPERATION.seconds(), 120);
+
+		// from a bound that took, so that the fallback is what the assertion below can be seeing:
+		// the default of this class is 0, which is also what a value read as a number would give
+		System.setProperty(StatementBound.BULK.property, "900");
+		assertEquals(StatementBound.BULK.seconds(), 900);
+		System.setProperty(StatementBound.BULK.property, "as long as it takes");
+		assertEquals(StatementBound.BULK.seconds(), 0);
+	}
+
+	@Test
+	public void testTheBoundReachesTheStatement() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final PreparedStatement statement = mock(PreparedStatement.class);
+
+		storage.execute(statement);
+
+		verify(statement).setQueryTimeout(7);
+	}
+
+	/** An unbounded class costs no call of its own: a fresh statement is unbounded already. */
+	@Test
+	public void testAnUnboundedClassSetsNothing() throws Exception {
+		final PreparedStatement statement = mock(PreparedStatement.class);
+
+		storage.execute(statement, StatementBound.BULK);
+
+		verify(statement, never()).setQueryTimeout(anyInt());
+	}
+
+	/**
+	 * Behind the cancel is a socket read timeout, for the databases that do not act on a cancel:
+	 * it is armed for the statement and put back once nothing is running on the connection any
+	 * more. What a connection carrying several statements at once does with it is pinned by the
+	 * three tests below.
+	 */
+	@Test
+	public void testTheBackstopIsArmedAndPutBack() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0); // no bound of its own
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+
+		storage.execute(statement);
+
+		final InOrder inOrder = inOrder(con);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+	}
+
+	/**
+	 * The backstop only ever tightens. A read timeout a deployment gave its connections is the
+	 * bound it asked for, and this one - deliberately the looser of the two, so that the cancel
+	 * has room to arrive first - must not stand in for it while a statement runs.
+	 */
+	@Test
+	public void testTheBackstopDoesNotLoosenATighterBound() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(5000); // tighter than 7s plus the margin
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+
+		storage.execute(statement);
+
+		verify(con, never()).setNetworkTimeout(any(Executor.class), anyInt());
+	}
+
+	/**
+	 * A statement of a class that carries no bound takes the backstop off the connection for as
+	 * long as it runs. The socket read timeout is a property of the connection, and an importer
+	 * writes to a single one from every phase-one worker and every phase-two task, so the bulk
+	 * {@code delete from} that empties a tree would otherwise be cut at the bound of an entry read
+	 * happening to run beside it - and cut without ever naming a property, since a statement of an
+	 * unbounded class has none to name. Two threads, because that is how the two meet.
+	 */
+	@Test
+	public void testAnUnboundedStatementTakesTheBackstopOffWhileItRuns() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		System.setProperty(StatementBound.BULK.property, "0");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final CountDownLatch operationRunning = new CountDownLatch(1);
+		final CountDownLatch operationMayFinish = new CountDownLatch(1);
+		final CountDownLatch bulkRunning = new CountDownLatch(1);
+		final CountDownLatch bulkMayFinish = new CountDownLatch(1);
+		final PreparedStatement operation = lingering(con, operationRunning, operationMayFinish);
+		final PreparedStatement bulk = lingering(con, bulkRunning, bulkMayFinish);
+
+		final Background entryRead = start("entry-read", () -> storage.execute(operation));
+		awaitOrFail(operationRunning, "the entry read never started");
+		final Background clearTree = start("clear-tree", () -> storage.execute(bulk, StatementBound.BULK));
+		awaitOrFail(bulkRunning, "the bulk statement never started");
+		bulkMayFinish.countDown();
+		clearTree.joinOrFail();
+		operationMayFinish.countDown();
+		entryRead.joinOrFail();
+
+		final InOrder inOrder = inOrder(con);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0)); // the bulk statement takes it off
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0)); // and the entry read is through
+	}
+
+	/**
+	 * The backstop belongs to the connection, not to the statement that armed it: the first
+	 * statement to finish must not take it away from the statements still running there.
+	 */
+	@Test
+	public void testTheBackstopOutlastsTheStatementThatArmedIt() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final CountDownLatch running = new CountDownLatch(1);
+		final CountDownLatch mayFinish = new CountDownLatch(1);
+		final PreparedStatement lingering = lingering(con, running, mayFinish);
+		final PreparedStatement passing = mock(PreparedStatement.class);
+		when(passing.getConnection()).thenReturn(con);
+		when(passing.executeUpdate()).thenReturn(1);
+
+		final Background outliving = start("outliving", () -> storage.execute(lingering));
+		awaitOrFail(running, "the statement that arms the backstop never started");
+		storage.execute(passing); // joins that connection and is through while the other one runs
+
+		verify(con, never()).setNetworkTimeout(any(Executor.class), eq(0));
+		mayFinish.countDown();
+		outliving.joinOrFail();
+		final InOrder inOrder = inOrder(con);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+	}
+
+	/**
+	 * With bounds of two classes in flight on one connection, the value armed is the loosest of
+	 * them: a socket read timeout is shared by everything running on the connection, so tightening
+	 * it to the bound of an entry read would cut the bulk statement beside it long before the bound
+	 * that statement was actually given.
+	 */
+	@Test
+	public void testTheBackstopFollowsTheLoosestBoundInFlight() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		System.setProperty(StatementBound.BULK.property, "100");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final CountDownLatch bulkRunning = new CountDownLatch(1);
+		final CountDownLatch bulkMayFinish = new CountDownLatch(1);
+		final PreparedStatement bulk = lingering(con, bulkRunning, bulkMayFinish);
+		final PreparedStatement operation = mock(PreparedStatement.class);
+		when(operation.getConnection()).thenReturn(con);
+		when(operation.executeUpdate()).thenReturn(1);
+
+		final Background count = start("count", () -> storage.execute(bulk, StatementBound.BULK));
+		awaitOrFail(bulkRunning, "the bulk statement never started");
+		storage.execute(operation); // an entry read of another thread, with a tighter bound
+		bulkMayFinish.countDown();
+		count.joinOrFail();
+
+		final InOrder inOrder = inOrder(con);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((100 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+		verify(con, never()).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+
+	}
+
+	/**
+	 * The other order, which is the one that moves the backstop while a statement is running: a
+	 * bound looser than what is armed re-arms the connection to its own value, and the tighter
+	 * statement left behind gets its bound back the moment the looser one is through.
+	 */
+	@Test
+	public void testALooserBoundRearmsTheBackstopAndTheTighterOneGetsItBack() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		System.setProperty(StatementBound.BULK.property, "100");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final CountDownLatch operationRunning = new CountDownLatch(1);
+		final CountDownLatch operationMayFinish = new CountDownLatch(1);
+		final CountDownLatch bulkRunning = new CountDownLatch(1);
+		final CountDownLatch bulkMayFinish = new CountDownLatch(1);
+		final PreparedStatement operation = lingering(con, operationRunning, operationMayFinish);
+		final PreparedStatement bulk = lingering(con, bulkRunning, bulkMayFinish);
+
+		final Background entryRead = start("entry-read", () -> storage.execute(operation));
+		awaitOrFail(operationRunning, "the entry read never started");
+		final Background count = start("count", () -> storage.execute(bulk, StatementBound.BULK));
+		awaitOrFail(bulkRunning, "the bulk statement never started");
+		bulkMayFinish.countDown();
+		count.joinOrFail();
+		operationMayFinish.countDown();
+		entryRead.joinOrFail();
+
+		final InOrder inOrder = inOrder(con);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((100 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+	}
+
+	/**
+	 * A failure that arrives before the bound is the caller's to classify and must reach it as it
+	 * stands: a lock wait reported in class 40 is the conflict {@code JDBCStorage.write()} replays,
+	 * and wrapping it would take it out of that class.
+	 */
+	@Test
+	public void testAFailureInsideTheBoundIsPassedThrough() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "60");
+		final SQLException conflict = new SQLException("lock wait timeout exceeded", "40001", 1205);
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.executeUpdate()).thenThrow(conflict);
+
+		try {
+			storage.execute(statement);
+			fail("the failure of the statement must reach the caller");
+		} catch (SQLException e) {
+			assertSame(e, conflict);
+		}
+	}
+
+	/**
+	 * A failure that arrives at the bound names the property that produced it - every driver
+	 * reports a cancelled statement differently, and none of them knows why it was cancelled - and
+	 * still carries the SQL state and the error number of the failure it replaces.
+	 */
+	@Test
+	public void testAFailureAtTheBoundNamesTheProperty() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "1");
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.executeUpdate()).thenAnswer(new Answer<Integer>() {
+			@Override
+			public Integer answer(InvocationOnMock invocation) throws Throwable {
+				Thread.sleep(1100); // the driver cancelled it at the bound this test set
+				throw new SQLException("canceling statement due to user request", "57014", 0);
+			}
+		});
+
+		try {
+			storage.execute(statement);
+			fail("the failure of the statement must reach the caller");
+		} catch (SQLTimeoutException e) {
+			assertEquals(e.getSQLState(), "57014");
+			assertTrue(e.getMessage().contains(StatementBound.OPERATION.property), e.getMessage());
+			assertEquals(((SQLException) e.getCause()).getSQLState(), "57014");
+		}
+	}
+
+	/**
+	 * The rows are read while the bound is still armed. A driver hands them over as they are asked
+	 * for - oracle prefetches ten at a time, mssql buffers adaptively - so a drain that happened
+	 * after the bound was released would be a wait with nothing bounding it, which is the hang
+	 * #877 is about rather than a detail of where the call sits.
+	 */
+	@Test
+	public void testTheRowsAreReadWhileTheBoundIsStillArmed() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final ResultSet rows = mock(ResultSet.class);
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+		when(statement.executeQuery()).thenReturn(rows);
+
+		assertEquals(storage.executeResultSet(statement, ResultSet::next), Boolean.FALSE);
+
+		final InOrder inOrder = inOrder(con, rows);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(rows).next();
+		inOrder.verify(rows).close(); // the rows are done with before the backstop goes back
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+	}
+
+	/** A transfer of rows cut at the bound names the property that cut it, as an execution does. */
+	@Test
+	public void testAFailureWhileTheRowsAreReadIsMeasuredAgainstTheBound() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "1");
+		final ResultSet rows = mock(ResultSet.class);
+		when(rows.next()).thenAnswer(new Answer<Boolean>() {
+			@Override
+			public Boolean answer(InvocationOnMock invocation) throws Throwable {
+				Thread.sleep(1100); // the driver cancelled the transfer at the bound this test set
+				throw new SQLException("canceling statement due to user request", "57014", 0);
+			}
+		});
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.executeQuery()).thenReturn(rows);
+
+		try {
+			storage.executeResultSet(statement, ResultSet::next);
+			fail("the failure of the transfer must reach the caller");
+		} catch (SQLTimeoutException e) {
+			assertTrue(e.getMessage().contains(StatementBound.OPERATION.property), e.getMessage());
+			assertEquals(e.getSQLState(), "57014");
+		}
+	}
+
+	/**
+	 * A driver is allowed to have no query timeout at all, and this backend takes whatever URL a
+	 * deployment configures. Such a driver has to keep working, with the socket read timeout as
+	 * its whole bound, rather than fail every statement it is given.
+	 */
+	@Test
+	public void testADriverWithoutAQueryTimeoutKeepsWorkingUnderTheBackstop() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+		doThrow(new SQLFeatureNotSupportedException("no query timeout")).when(statement).setQueryTimeout(anyInt());
+		when(statement.executeUpdate()).thenReturn(1);
+
+		assertEquals(storage.execute(statement), 1);
+
+		verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+	}
+
+	/**
+	 * The catalog lookups of {@code openTree()} are bounded too. {@code DatabaseMetaData} takes no
+	 * query timeout, so the socket read timeout behind the cancel is the only layer they can be
+	 * given - and they run once per tree on every open of a backend, behind the same locks as the
+	 * {@code create table} they guard.
+	 */
+	@Test
+	public void testACatalogLookupIsBoundedByTheBackstopAlone() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+
+		assertEquals(storage.bounded(con, StatementBound.OPERATION, () -> "asked the catalog"), "asked the catalog");
+
+		final InOrder inOrder = inOrder(con);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+	}
+
+	/**
+	 * Which class a batch of a cursor belongs to follows what the database has to do to answer it.
+	 * {@code positionToLastKey()} has no key to seek on, so it is an {@code order by k desc} over
+	 * the whole table - a scan and a sort of it on mssql, where {@code k} cannot be an index key -
+	 * and every open of a backend runs it once per base DN through
+	 * {@code EntryContainer.getHighestEntryID()}, outside the try/catch of
+	 * {@code BackendImpl.openBackend()}. Bounding that as an entry read would turn a large backend
+	 * that opens slowly into one that does not open at all, while the batches the cursor walks
+	 * along its index stay operations and keep the bound of one.
+	 */
+	@Test
+	public void testTheScanBehindTheHighestEntryIdIsBulkAndTheBatchesOfACursorAreNot() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		System.setProperty(StatementBound.BULK.property, "0");
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.executeQuery()).thenReturn(mock(ResultSet.class));
+		final Connection parent = mock(Connection.class);
+		when(parent.prepareStatement(anyString())).thenReturn(statement);
+		final JDBCStorage.CursorImpl cursor = storage.new CursorImpl(true, new CachedConnection("jdbc:mock", parent),
+			new TreeName("dc=example,dc=com", "id2entry"), StatementBound.OPERATION);
+
+		cursor.positionToLastKey();
+		verify(statement, never()).setQueryTimeout(anyInt());
+
+		cursor.next();
+		verify(statement).setQueryTimeout(7);
+	}
+
+	/**
+	 * A cursor opened for a walk of a whole tree takes bulk batches, however ordinary the statement
+	 * looks: nobody is waiting on that walk, and on mssql it is not even a walk along an index -
+	 * {@code k} is a {@code varbinary(max)} there, which cannot be an index key, so every batch is
+	 * a scan and a sort of the whole table. This is the class an export, a verify, a rebuild and
+	 * the load of a tree at open ask for through {@code ReadableTransaction.openBulkCursor()},
+	 * while the cursor of a search keeps the bound of an operation.
+	 */
+	@Test
+	public void testTheBatchesOfABulkCursorAreBulkAndThoseOfASearchAreNot() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		System.setProperty(StatementBound.BULK.property, "0");
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.executeQuery()).thenReturn(mock(ResultSet.class));
+		final Connection parent = mock(Connection.class);
+		when(parent.prepareStatement(anyString())).thenReturn(statement);
+		final JDBCStorage.ReadableTransactionImpl txn =
+			storage.new ReadableTransactionImpl(new CachedConnection("jdbc:mock", parent));
+		final TreeName tree = new TreeName("dc=example,dc=com", "id2entry");
+
+		txn.openBulkCursor(tree).next();
+		verify(statement, never()).setQueryTimeout(anyInt());
+
+		txn.openCursor(tree).next();
+		verify(statement).setQueryTimeout(7);
+	}
+
+	/**
+	 * Every statement an import issues is bulk, by the class of the transactions it works through:
+	 * phase one writes the trees through {@code put()}, phase two reads them back through
+	 * {@code read()} and walks them through {@code openCursor()}, and no client is waiting on any
+	 * of it. An upsert of an online import blocked by an LDAP write on the same table would
+	 * otherwise sit until the bound of an entry read and then fail the whole import - {@code h} is
+	 * the primary key on every dialect, and the default lock wait is forever on three of the four.
+	 */
+	@Test
+	public void testEveryStatementOfAnImportIsBulk() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		System.setProperty(StatementBound.BULK.property, "0");
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.executeQuery()).thenReturn(mock(ResultSet.class));
+		final Connection parent = mock(Connection.class);
+		when(parent.prepareStatement(anyString())).thenReturn(statement);
+		storage.accessMode = AccessMode.READ_WRITE; // an import has the storage open for writing
+		final JDBCStorage.ImporterImpl importer =
+			storage.new ImporterImpl(new CachedConnection("jdbc:mock", parent), true);
+		final TreeName tree = new TreeName("dc=example,dc=com", "id2entry");
+
+		importer.openCursor(tree).next();
+		importer.read(tree, ByteString.valueOfUtf8("key"));
+		importer.put(tree, ByteString.valueOfUtf8("key"), ByteString.valueOfUtf8("value"));
+
+		verify(statement, never()).setQueryTimeout(anyInt());
+	}
+
+	/**
+	 * A statement bounded by the socket read timeout alone is measured against what that layer
+	 * really allows it - its bound plus the margin the layer carries - rather than against the
+	 * property: nothing cuts a catalog lookup at the bound itself, so a connection reset arriving
+	 * just after it is the caller's failure to see, not a query timeout that never happened.
+	 */
+	@Test
+	public void testAFailureBeforeTheBackstopOfACatalogLookupIsPassedThrough() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "1");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		final SQLException reset = new SQLException("connection reset by peer", "08006", 0);
+
+		try {
+			storage.bounded(con, StatementBound.OPERATION, () -> {
+				sleep(1100); // past the property, well inside the margin of the layer behind it
+				throw reset;
+			});
+			fail("the failure of the lookup must reach the caller");
+		} catch (SQLException e) {
+			assertSame(e, reset);
+		}
+	}
+
+	/**
+	 * A driver with no network timeout at all is asked once and then left alone: it says so with
+	 * {@code SQLFeatureNotSupportedException}, and asking it again costs a throw on every statement
+	 * for the life of the storage. Its own storage here, since that is the scope of the latch.
+	 */
+	@Test
+	public void testADriverWithoutANetworkTimeoutIsNotAskedAgain() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final JDBCStorage isolated = new JDBCStorage(mockCfg(JDBCBackendCfg.class), null);
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		doThrow(new SQLFeatureNotSupportedException("no network timeout"))
+			.when(con).setNetworkTimeout(any(Executor.class), anyInt());
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+		when(statement.executeUpdate()).thenReturn(1);
+
+		assertEquals(isolated.execute(statement), 1);
+		assertEquals(isolated.execute(statement), 1);
+
+		verify(con, times(1)).setNetworkTimeout(any(Executor.class), anyInt());
+	}
+
+	/**
+	 * A connection that failed the call says nothing about the driver - it may be the very one
+	 * that reached this timeout - so the next statement is armed as usual. The two causes share a
+	 * catch and must not share a verdict: taking one for the other silences the backstop of a
+	 * whole storage on a single dying connection.
+	 */
+	@Test
+	public void testAConnectionThatFailedTheBackstopDoesNotSpeakForTheDriver() throws Exception {
+		System.setProperty(StatementBound.OPERATION.property, "7");
+		final Connection con = mock(Connection.class);
+		when(con.getNetworkTimeout()).thenReturn(0);
+		doThrow(new SQLException("the connection is closed", "08003", 0))
+			.when(con).setNetworkTimeout(any(Executor.class), anyInt());
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		when(statement.getConnection()).thenReturn(con);
+		when(statement.executeUpdate()).thenReturn(1);
+
+		assertEquals(storage.execute(statement), 1);
+		assertEquals(storage.execute(statement), 1);
+
+		verify(con, times(2)).setNetworkTimeout(any(Executor.class),
+			eq((7 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+	}
+
+	/**
+	 * The statistics refresh after an import runs under a bound of its own - it takes as long as a
+	 * scan of the table it describes, which no class of {@link StatementBound} can be asked to
+	 * allow - and under both layers of it. The second one is the reason: on oracle this statement
+	 * is {@code dbms_stats.gather_table_stats}, the engine whose session does not act on the break
+	 * its driver sends, and it runs at the very end of a successful import, where a cancel that
+	 * never arrives would park the import with its data already committed.
+	 */
+	@Test
+	public void testTheStatisticsRefreshRunsUnderItsOwnBoundAndTheBackstop() throws Exception {
+		System.setProperty(JDBCStorage.STATISTICS_TIMEOUT_PROPERTY, "60");
+		final PreparedStatement statement = mock(PreparedStatement.class);
+		final Connection con = mock(oracleConnection.class); // the dialect is read off the connection
+		when(con.getNetworkTimeout()).thenReturn(0);
+		when(con.prepareStatement(anyString())).thenReturn(statement);
+
+		assertTrue(storage.updateTableStatistics(con, singletonList(new TreeName("dc=example,dc=com", "id2entry"))));
+
+		verify(statement).setQueryTimeout(60);
+		final InOrder inOrder = inOrder(con, statement);
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq((60 + JDBCStorage.BACKSTOP_MARGIN_SECONDS) * 1000));
+		inOrder.verify(statement).execute();
+		inOrder.verify(con).setNetworkTimeout(any(Executor.class), eq(0));
+	}
+
+	// JDBCStorage.dialectOf() reads the engine off the class name of the connection, so a mock of
+	// this interface is an oracle connection as far as the storage is concerned - which is the
+	// whole reason for the lower case name here.
+	private interface oracleConnection extends Connection {}
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
index b650301554..9b363d0e58 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
@@ -53,6 +53,7 @@
 
 import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
 import static org.mockito.Mockito.when;
+import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
@@ -199,6 +200,149 @@ public void run(WriteableTransaction txn) throws Exception {
 		}
 	}
 
+	/**
+	 * A statement of this backend has to end even when another session holds what it needs: a row
+	 * locked by a transaction that never commits used to park the worker thread that issued the
+	 * write for good, with nothing in the log to say so (#877).
+	 */
+	@Test(timeOut = 600000)
+	public void testWriteBlockedByAnotherSessionGivesUpAtItsBound() throws Exception {
+		assertBoundedWhileRowsAreLocked("testStatementBound", JDBCStorage.StatementBound.OPERATION,
+			new BlockedOperation() {
+				@Override
+				public void run(JDBCStorage storage, TreeName tree) throws Exception {
+					storage.write(new WriteOperation() {
+						@Override
+						public void run(WriteableTransaction txn) throws Exception {
+							txn.put(tree, key(1), value(2));
+						}
+					});
+				}
+			});
+	}
+
+	/**
+	 * The bulk class keeps a bound of its own: a count or the delete that empties a tree before an
+	 * import legitimately takes minutes, so it must not be cut at the bound of an entry read - and
+	 * must still be able to give up (#877).
+	 */
+	@Test(timeOut = 600000)
+	public void testBulkStatementGivesUpAtItsOwnBound() throws Exception {
+		assertBoundedWhileRowsAreLocked("testBulkBound", JDBCStorage.StatementBound.BULK,
+			new BlockedOperation() {
+				@Override
+				public void run(JDBCStorage storage, TreeName tree) throws Exception {
+					// the importer is where "delete from <table>" - the bulk class - is reachable:
+					// AbstractTwoPhaseImportStrategy clears every tree before an import writes to it
+					try (final Importer importer = storage.startImport()) {
+						importer.clearTree(tree);
+					}
+				}
+			});
+	}
+
+	private interface BlockedOperation {
+		void run(JDBCStorage storage, TreeName tree) throws Exception;
+	}
+
+	/**
+	 * Whether the failure the operation gave up with is the one its bound produced: the message of
+	 * a statement classified as having reached its bound names the property that bounded it, and it
+	 * arrives wrapped in whatever the storage throws to its caller.
+	 */
+	private static boolean namesTheBound(Throwable failure, JDBCStorage.StatementBound bound) {
+		for (Throwable t = failure; t != null && t != t.getCause(); t = t.getCause()) {
+			if (t.getMessage() != null && t.getMessage().contains(bound.property)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/** How far under its bound a statement may report the cancel, the timer of a driver being coarse. */
+	private static final long CLOCK_SLACK_MILLIS = 250;
+
+	/**
+	 * Runs the given operation while another session holds every row of the tree in an uncommitted
+	 * transaction, with only the property of the given class bounding it: the operation must give
+	 * up inside that bound instead of waiting for a lock that is never released.
+	 */
+	private void assertBoundedWhileRowsAreLocked(String treeId, JDBCStorage.StatementBound bound, BlockedOperation blocked)
+			throws Exception {
+		final int boundSeconds = 5;
+		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
+		final TreeName tree = new TreeName(treeId, "tree");
+		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));
+				}
+			});
+			// another session takes an exclusive lock on every row of the table and keeps it: the
+			// same statement clearTree() issues, so it is known to parse on all four dialects
+			try (final Connection blocker = DriverManager.getConnection(getJdbcUrl())) {
+				blocker.setAutoCommit(false);
+				try (final Statement lock = blocker.createStatement()) {
+					lock.executeUpdate("delete from " + storage.getTableName(tree));
+				}
+				// only the class under test is bounded, so a pass through the other one cannot
+				// be mistaken for the bound working
+				for (final JDBCStorage.StatementBound each : JDBCStorage.StatementBound.values()) {
+					System.setProperty(each.property, each == bound ? Integer.toString(boundSeconds) : "0");
+				}
+				// the monotonic clock, which is what timedOut() measures the bound with: a step of
+				// the wall clock can neither lengthen nor shorten what the assertions below allow
+				final long startedAt = System.nanoTime();
+				Exception failure = null;
+				try {
+					blocked.run(storage, tree);
+					fail("the operation must give up while the rows it needs are locked");
+				} catch (Exception expected) {
+					failure = expected; // the bound was reached and the transaction rolled back
+				}
+				final long elapsed = (System.nanoTime() - startedAt) / 1000000L;
+				// The failure has to be the one the bound produces, not any failure at all: an
+				// operation that fell over at once for an unrelated reason would otherwise pass
+				// this test at t=0. timedOut() names the property in the message of everything it
+				// classifies as reaching the bound.
+				assertTrue(namesTheBound(failure, bound), "gave up with " + stackTraceToSingleLineString(failure)
+					+ ", which does not name " + bound.property);
+				// And it has to arrive at the bound rather than at something else that happens to
+				// end the wait inside a generous ceiling: with the bound deleted, mysql would still
+				// come back after its own innodb_lock_wait_timeout of 50 s, and the assertion has
+				// to fail then. Oracle is given the second layer as well - a session blocked in a
+				// row-lock enqueue does not act on the break its driver sends, so the wait there
+				// ends at the socket read timeout, which is the bound plus its margin.
+				final long ceilingSeconds = getJdbcUrl().startsWith("jdbc:oracle")
+					? boundSeconds + JDBCStorage.BACKSTOP_MARGIN_SECONDS + 10 : boundSeconds * 4L;
+				// with a little slack under the bound: a driver keeps its timer in whole seconds and
+				// may report the cancel a few milliseconds before the bound is arithmetically due
+				assertTrue(elapsed >= boundSeconds * 1000L - CLOCK_SLACK_MILLIS,
+					"gave up after " + elapsed + " ms, before its bound of "
+					+ boundSeconds + " s: something other than the bound ended the wait");
+				assertTrue(elapsed < ceilingSeconds * 1000L, "gave up only after " + elapsed + " ms, past the "
+					+ ceilingSeconds + " s this bound of " + boundSeconds + " s allows");
+				blocker.rollback();
+			}
+		} finally {
+			for (final JDBCStorage.StatementBound each : JDBCStorage.StatementBound.values()) {
+				System.clearProperty(each.property);
+			}
+			try {
+				storage.write(new WriteOperation() {
+					@Override
+					public void run(WriteableTransaction txn) throws Exception {
+						txn.deleteTree(tree);
+					}
+				});
+			} catch (Exception ignored) {}
+			storage.close();
+		}
+	}
+
 	/**
 	 * Forward repositioning inside the already-fetched batch must be served from the buffer without SQL,
 	 * and batch sizes must grow from "fetchsize.initial" to "fetchsize" on sequential reads (#860).
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ID2EntryTest.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ID2EntryTest.java
new file mode 100644
index 0000000000..b1633152a9
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/ID2EntryTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.pluggable;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.forgerock.opendj.ldap.ByteString;
+import org.opends.server.DirectoryServerTestCase;
+import org.opends.server.backends.pluggable.spi.Cursor;
+import org.opends.server.backends.pluggable.spi.TreeName;
+import org.opends.server.backends.pluggable.spi.WriteableTransaction;
+import org.testng.annotations.Test;
+
+@SuppressWarnings("javadoc")
+@Test(groups = { "precommit", "pluggablebackend" }, sequential = true)
+public class ID2EntryTest extends DirectoryServerTestCase
+{
+  /**
+   * The read that checks the tree is there when a backend opens asks for a bulk cursor. Its first
+   * batch carries no key to seek on, so a storage engine sees a walk of the whole tree - on the
+   * JDBC backend against SQL Server, a scan and a sort of it, {@code k} being a
+   * {@code varbinary(max)} that cannot be an index key - and this runs once per base DN on every
+   * open, outside the try/catch of {@code BackendImpl.openBackend()}. Bounded as the work of a
+   * client operation, a large backend would stop opening at all (#877).
+   */
+  @Test
+  public void testTheReadThatOpensTheTreeAsksForABulkCursor() throws Exception
+  {
+    final TreeName name = new TreeName("dc=example,dc=com", "id2entry");
+    final WriteableTransaction txn = mock(WriteableTransaction.class);
+    @SuppressWarnings("unchecked")
+    final Cursor<ByteString, ByteString> cursor = mock(Cursor.class);
+    when(txn.openBulkCursor(name)).thenReturn(cursor);
+
+    new ID2Entry(name, new DataConfig.Builder().build()).open(txn, false);
+
+    verify(txn).openBulkCursor(name);
+    verify(cursor).next();
+  }
+}