/*
 * 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 2024-2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.ByteStringBuilder;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.backends.pluggable.PluggableBackendImplTestCase;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Cursor;
import org.opends.server.backends.pluggable.spi.Importer;
import org.opends.server.backends.pluggable.spi.ReadOperation;
import org.opends.server.backends.pluggable.spi.ReadableTransaction;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.WriteOperation;
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
import org.testcontainers.DockerClientFactory;
import org.testcontainers.containers.JdbcDatabaseContainer;
import org.testng.SkipException;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;



public abstract class TestCase extends PluggableBackendImplTestCase<JDBCBackendCfg> {

	JdbcDatabaseContainer container;

	@BeforeClass
	@Override
	public void setUp() throws Exception {
		if(DockerClientFactory.instance().isDockerAvailable()) {
			try {
				container = getContainer();
				container.start();
			} catch (Exception e) {
				// The database container could not be started (e.g. a slow/flaky image
				// pull or DB initialization failure on CI). Skip the test instead of
				// failing the whole build - container startup is an infrastructure
				// concern, not a regression in the JDBC backend under test.
				throw new SkipException(getContainerDockerCommand());
			}
		}
		try(Connection con = DriverManager.getConnection(createBackendCfg().getDBDirectory())){
			dropStaleTrees(con);
		} catch (Exception e) {
			throw new SkipException(getContainerDockerCommand());
		}
		super.setUp();
	}

	/**
	 * Backend test classes sharing one database map the same tree names to the same tables,
	 * so a previous run may leave trees behind — including entries encrypted with a lost cipher key.
	 */
	static void dropStaleTrees(Connection con) throws SQLException {
		final List<String> stale = new ArrayList<>();
		try (final ResultSet rs = con.getMetaData().getTables(null, null, null, new String[]{"TABLE"})) {
			while (rs.next()) {
				final String name = rs.getString("TABLE_NAME");
				if (name.toLowerCase().startsWith("opendj_")) {
					stale.add(name);
				}
			}
		}
		try (final Statement st = con.createStatement()) {
			for (final String name : stale) {
				st.execute("drop table " + name);
			}
		}
	}

	@Override
	protected Backend createBackend() {
		return new Backend();
	}

	@Override
	protected JDBCBackendCfg createBackendCfg() {
		JDBCBackendCfg backendCfg = mockCfg(JDBCBackendCfg.class);
		when(backendCfg.getBackendId()).thenReturn(getBackendId());
		when(backendCfg.getDBDirectory()).thenReturn(getJdbcUrl());
		return backendCfg;
	}

	@AfterClass
	@Override
	public void cleanUp() throws Exception {
		super.cleanUp();
		if(container != null) {
			container.close();
		}
	}

	protected abstract JdbcDatabaseContainer<?> getContainer();

	protected abstract String getContainerDockerCommand();

	protected abstract String getBackendId();

	protected abstract String getJdbcUrl();

	private static ByteString key(int i) {
		return ByteString.valueOfUtf8(String.format("key%02d", i));
	}

	private static ByteString value(int i) {
		return ByteString.valueOfUtf8("value" + i);
	}

	/**
	 * openTree() and deleteTree() ask the catalog whether the table of a tree is there. The name is
	 * looked up in the form the catalog stores it - an unquoted identifier is folded to upper case
	 * on oracle and to lower case on postgresql - so getting that wrong makes a second open try to
	 * create a table that is already there, and a second delete drop one that is already gone (#885).
	 */
	@Test
	public void testTableOfATreeIsFoundByName() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("testCatalogLookup", "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));
				}
			});
			// the table is there now: opening the tree again must find it, not create it a second time
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			assertEquals(storage.read(new ReadOperation<ByteString>() {
				@Override
				public ByteString run(ReadableTransaction txn) throws Exception {
					return txn.read(tree, key(1));
				}
			}), value(1));

			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.deleteTree(tree);
				}
			});
			// and gone now: deleting it again must find nothing rather than drop what is not there
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.deleteTree(tree);
				}
			});
		} finally {
			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).
	 */
	@Test
	public void testPositionToKeyOrNextServedFromBuffer() throws Exception {
		System.setProperty("org.openidentityplatform.opendj.jdbc.fetchsize", "8");
		System.setProperty("org.openidentityplatform.opendj.jdbc.fetchsize.initial", "2");
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("testCursorBuffer", "tree");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
					for (int i = 0; i < 40; i++) {
						txn.put(tree, key(i), value(i));
					}
				}
			});
			storage.read(new ReadOperation<Void>() {
				@Override
				public Void run(ReadableTransaction txn) throws Exception {
					try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
						final JDBCStorage.CursorImpl impl = (JDBCStorage.CursorImpl) cursor;

						assertTrue(cursor.next()); // fetch #1: initial batch of 2 (key00, key01)
						assertEquals(cursor.getKey(), key(0));
						assertEquals(impl.fetchCount, 1);
						assertTrue(cursor.next()); // key01 is buffered
						assertEquals(impl.fetchCount, 1);
						assertTrue(cursor.next()); // fetch #2: grown batch of 8 (key02..key09)
						assertEquals(cursor.getKey(), key(2));
						assertEquals(impl.fetchCount, 2);

						// forward repositioning within the fetched range must not run SQL
						assertTrue(cursor.positionToKeyOrNext(key(5)));
						assertEquals(cursor.getKey(), key(5));
						assertEquals(cursor.getValue(), value(5));
						assertEquals(impl.fetchCount, 2);
						assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key051"))); // between rows
						assertEquals(cursor.getKey(), key(6));
						assertEquals(impl.fetchCount, 2);
						assertTrue(cursor.positionToKeyOrNext(key(9))); // last buffered row
						assertEquals(cursor.getKey(), key(9));
						assertEquals(impl.fetchCount, 2);

						assertTrue(cursor.positionToKeyOrNext(key(20))); // fetch #3: beyond the buffer
						assertEquals(cursor.getKey(), key(20));
						assertEquals(impl.fetchCount, 3);
						assertTrue(cursor.positionToKeyOrNext(key(1))); // fetch #4: backward
						assertEquals(cursor.getKey(), key(1));
						assertEquals(impl.fetchCount, 4);

						// emulate DN2ID.ChildrenCursor: reposition to currentKey+0x01 for every row.
						// Before the fix every reposition re-fetched a full batch: 38 fetches here.
						final long fetchesBefore = impl.fetchCount;
						int rows = 1; // standing on key01
						while (cursor.positionToKeyOrNext(
								new ByteStringBuilder().appendBytes(cursor.getKey()).appendByte(0x01).toByteString())) {
							rows++;
						}
						assertEquals(rows, 39); // key01..key39
						assertTrue(impl.fetchCount - fetchesBefore <= 8,
								"sibling scan took " + (impl.fetchCount - fetchesBefore) + " fetches");
					}
					return null;
				}
			});
		} finally {
			System.clearProperty("org.openidentityplatform.opendj.jdbc.fetchsize");
			System.clearProperty("org.openidentityplatform.opendj.jdbc.fetchsize.initial");
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/** Buffer-served repositioning relies on the database collating keys in unsigned byte order. */
	@Test
	public void testCursorKeyOrderIsUnsigned() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("testCursorOrder", "tree");
		final ByteString low = ByteString.valueOfBytes(new byte[] { 0x7F });
		final ByteString high = ByteString.valueOfBytes(new byte[] { (byte) 0x80, 0x01 });
		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, low, value(1));
					txn.put(tree, high, value(2));
				}
			});
			storage.read(new ReadOperation<Void>() {
				@Override
				public Void run(ReadableTransaction txn) throws Exception {
					try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
						// with a signed collation 0x80 would sort before 0x7F and these would fail
						assertTrue(cursor.next());
						assertEquals(cursor.getKey(), low);
						assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfBytes(new byte[] { (byte) 0x80 })));
						assertEquals(cursor.getKey(), high);
						assertFalse(cursor.next());
						assertTrue(cursor.positionToLastKey());
						assertEquals(cursor.getKey(), high);
					}
					return null;
				}
			});
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * Each table must be stamped with the tree name it stores: table names are opaque SHA-224
	 * hashes, so without the comment there is no way to tell the trees apart on the database
	 * side (#859). The single quote in the base DN exercises the comment escaping.
	 */
	@Test
	public void testTreeNameStoredAsTableComment() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		// a quote and a backslash in the tree name exercise the literal escaping (backslash is an escape character in mysql)
		final TreeName tree = new TreeName("o=comment'te\\st", "dn2id");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			assertEquals(readTableComment(storage.getTableName(tree)), tree.toString());
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	String readTableComment(String tableName) throws Exception {
		final String url = getJdbcUrl();
		final String sql;
		if (url.startsWith("jdbc:postgresql")) {
			sql = "select obj_description('" + tableName + "'::regclass, 'pg_class')";
		} else if (url.startsWith("jdbc:mysql")) {
			sql = "select table_comment from information_schema.tables where table_schema=database() and table_name='" + tableName + "'";
		} else if (url.startsWith("jdbc:oracle")) {
			sql = "select comments from user_tab_comments where table_name='" + tableName.toUpperCase() + "'";
		} else if (url.startsWith("jdbc:sqlserver")) {
			// class=1 is the table itself: major_id is only unique within a class
			sql = "select cast(value as nvarchar(4000)) from sys.extended_properties where class=1 and major_id=object_id('" + tableName + "') and minor_id=0 and name='MS_Description'";
		} else {
			throw new SkipException("no table comment query for " + url);
		}
		try (final Connection con = DriverManager.getConnection(url);
			 final Statement st = con.createStatement();
			 final ResultSet rs = st.executeQuery(sql)) {
			return rs.next() ? rs.getString(1) : null;
		}
	}

	void writeTableComment(String tableName, String comment) throws Exception {
		final String url = getJdbcUrl();
		final String sql;
		if (url.startsWith("jdbc:postgresql") || url.startsWith("jdbc:oracle")) {
			sql = "comment on table " + tableName + " is '" + comment + "'";
		} else if (url.startsWith("jdbc:mysql")) {
			sql = "alter table " + tableName + " comment '" + comment + "'";
		} else if (url.startsWith("jdbc:sqlserver")) {
			// exec arguments must be constants or variables: schema_name() cannot be passed inline
			sql = "declare @s sysname = schema_name()"
				+ " exec sys.sp_updateextendedproperty N'MS_Description', N'" + comment + "', N'SCHEMA', @s, N'TABLE', N'" + tableName + "'";
		} else {
			throw new SkipException("no table comment statement for " + url);
		}
		try (final Connection con = DriverManager.getConnection(url);
			 final Statement st = con.createStatement()) {
			st.execute(sql);
		}
	}

	/**
	 * Removes the stored comment, so that the next stamp has to create one rather than replace
	 * it: on sql server that is sp_addextendedproperty, which is the statement reported to wait
	 * for an uncommitted row of another session.
	 */
	void clearTableComment(String tableName) throws Exception {
		final String url = getJdbcUrl();
		if (!url.startsWith("jdbc:sqlserver")) {
			writeTableComment(tableName, "stale"); // the other engines have one statement for both cases
			return;
		}
		try (final Connection con = DriverManager.getConnection(url);
			 final Statement st = con.createStatement()) {
			st.execute("declare @s sysname = schema_name()"
				+ " exec sys.sp_dropextendedproperty N'MS_Description', N'SCHEMA', @s, N'TABLE', N'" + tableName + "'");
		}
	}

	/** The dialect of the database this suite runs against, as the backend detects it from the driver. */
	JDBCStorage.Dialect dialect() {
		final String url = getJdbcUrl();
		if (url.startsWith("jdbc:postgresql")) {
			return JDBCStorage.Dialect.POSTGRES;
		} else if (url.startsWith("jdbc:mysql")) {
			return JDBCStorage.Dialect.MYSQL;
		} else if (url.startsWith("jdbc:oracle")) {
			return JDBCStorage.Dialect.ORACLE;
		} else if (url.startsWith("jdbc:sqlserver")) {
			return JDBCStorage.Dialect.MICROSOFT;
		}
		throw new SkipException("no dialect for " + url);
	}

	/**
	 * What this connection reports as its lock bound, in the unit and the rendering of its own
	 * engine, or null where reading it needs a privilege the test user does not have: oracle
	 * keeps ddl_lock_timeout in v$parameter, which an application user cannot select from.
	 */
	String sessionLockBound(Connection con) throws Exception {
		final String url = getJdbcUrl();
		final String sql;
		if (url.startsWith("jdbc:postgresql")) {
			sql = "show lock_timeout";
		} else if (url.startsWith("jdbc:mysql")) {
			sql = "select @@session.lock_wait_timeout";
		} else if (url.startsWith("jdbc:sqlserver")) {
			sql = "select @@lock_timeout";
		} else {
			return null;
		}
		try (final Statement st = con.createStatement();
			 final ResultSet rs = st.executeQuery(sql)) {
			return rs.next() ? rs.getString(1) : null;
		}
	}

	/**
	 * Comment statements are DDL (a metadata lock on mysql, a ddl lock on oracle), so a table
	 * whose stored comment already matches its tree name must not be re-stamped on subsequent
	 * opens - while a stale comment must be refreshed.
	 */
	@Test
	public void testCommentStampSkippedWhenAlreadyStored() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("o=commentSkip", "dn2id");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true); // stamps the freshly created table
				}
			});
			assertEquals(readTableComment(storage.getTableName(tree)), tree.toString());
			// UP_TO_DATE and not FAILED: the statement was skipped, not rejected
			assertEquals(storage.commentTable(tree, dialect()), JDBCStorage.CommentResult.UP_TO_DATE, "an up-to-date comment was re-stamped");
			writeTableComment(storage.getTableName(tree), "stale");
			assertEquals(storage.commentTable(tree, dialect()), JDBCStorage.CommentResult.STAMPED, "a stale comment was not re-stamped");
			assertEquals(readTableComment(storage.getTableName(tree)), tree.toString());
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * A stamp must never queue behind another session's transaction. Comment statements take a
	 * lock (a metadata lock on mysql, a schema modification lock on sql server) and both engines
	 * wait for it without limit by default - lock_wait_timeout is a year, lock_timeout is
	 * infinite - so an unbounded stamp could hang the backend open and, on mysql, park every
	 * other query on that table behind itself. Whether an uncommitted row of another session
	 * conflicts with the statement at all differs between engines and versions, so the assertion
	 * is on the timing: the call comes back rather than waiting for that transaction to end.
	 */
	@Test(timeOut = 180000)
	public void testCommentStampGivesUpOnLock() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("o=commentLock", "dn2id");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			final String tableName = storage.getTableName(tree);
			clearTableComment(tableName); // no comment stored: the next attempt must issue a statement
			try (final Connection blocker = DriverManager.getConnection(getJdbcUrl())) {
				blocker.setAutoCommit(false);
				try (final PreparedStatement st = blocker.prepareStatement("insert into " + tableName + " (h,k) values (?,?)")) {
					st.setString(1, String.format("%1$-128s", "blocker").replace(' ', 'x'));
					st.setBytes(2, new byte[]{1});
					st.executeUpdate();
				}
				// the row is left uncommitted, so the lock it holds is still there
				final long start = System.currentTimeMillis();
				final JDBCStorage.CommentResult result = storage.commentTable(tree, dialect());
				final long elapsedMs = System.currentTimeMillis() - start;
				blocker.rollback();
				// giving up and stamping anyway are both fine here - the engines differ in whether an
				// uncommitted row of another session conflicts with the comment statement at all.
				// Waiting for that session to finish is what must never happen.
				// the bound is 5 s (COMMENT_LOCK_TIMEOUT_SECONDS): the slack is for the connect and the
				// statement around it, not for a regression of the bound itself
				assertTrue(elapsedMs < 20000, "the comment statement waited " + elapsedMs + " ms for a lock, result " + result);
				if (result == JDBCStorage.CommentResult.STAMPED) { // it reported success: the comment must be there
					assertEquals(readTableComment(tableName), tree.toString());
				}
			}
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * A failing comment stamp must never disturb the transaction that opened the tree: it used
	 * to roll back the caller's connection, silently discarding writes pending in the same
	 * transaction (the way DefaultIndex.afterOpen() writes the trusted flag between openTree() calls).
	 * The write pending during the failing stamp deliberately targets another tree: a statement
	 * left pending on the very table being stamped - a write, or on mysql any statement, since a
	 * transaction holds a shared metadata lock on every table it touched - would make the comment
	 * statement wait for the caller's own lock, which is a shape no production path has.
	 */
	@Test
	public void testCommentFailureLeavesTransactionIntact() throws Exception {
		final TreeName stamped = new TreeName("o=commentFailure", "dn2id");
		final TreeName written = new TreeName("o=commentFailure", "id2entry");
		final JDBCStorage setUp = new JDBCStorage(createBackendCfg(), null);
		try { // create both tables up front, with a storage that stamps them normally
			setUp.open(AccessMode.READ_WRITE);
			setUp.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(stamped, true);
					txn.openTree(written, true);
				}
			});
		} finally {
			setUp.close();
		}
		final AtomicInteger stampAttempts = new AtomicInteger();
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			Connection newStampConnection(Dialect dialect) throws SQLException {
				stampAttempts.incrementAndGet();
				throw new SQLException("injected comment failure"); // no sql state, no vendor code: a rejection, not a failure of the moment
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.put(written, key(1), value(1)); // pending in this transaction...
					txn.openTree(stamped, true); // ...while the comment machinery fails
				}
			});
			storage.read(new ReadOperation<Void>() {
				@Override
				public Void run(ReadableTransaction txn) throws Exception {
					assertEquals(txn.read(written, key(1)), value(1), "failing comment stamp discarded a pending write");
					return null;
				}
			});
			// the failure is remembered: an unstampable table is not asked again while this backend is open
			assertEquals(storage.commentTable(stamped, dialect()), JDBCStorage.CommentResult.FAILED);
			assertEquals(stampAttempts.get(), 1, "a failed stamp was reissued");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(stamped);
						txn.deleteTree(written);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * A stamp that failed for a reason of the moment - the lock timeout the statement is given,
	 * a connection that broke - must be attempted again: only a failure saying that this table
	 * cannot be commented at all is remembered, or one contended moment would leave a backend
	 * unstamped until it is restarted.
	 */
	@Test
	public void testTransientStampFailureIsRetried() throws Exception {
		final TreeName tree = new TreeName("o=transientStamp", "dn2id");
		final AtomicInteger stampAttempts = new AtomicInteger();
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			Connection newStampConnection(Dialect dialect) throws SQLException {
				stampAttempts.incrementAndGet();
				throw new SQLException("injected connection failure", "08006"); // connection exception: a failure of the moment
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true); // stamp #1, fails
				}
			});
			assertEquals(storage.commentTable(tree, dialect()), JDBCStorage.CommentResult.FAILED);
			assertEquals(stampAttempts.get(), 2, "a stamp that failed for a reason of the moment was not attempted again");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * Opening a backend opens every tree it holds - about 25 for a stock suffix - and the first
	 * open after an upgrade stamps them all: the trees of one open must share one connection
	 * rather than make a physical connect each. One per open is what the comment machinery costs,
	 * readback included - the readback runs on that same connection, because the thread doing the
	 * open is inside a transaction and holding a pooled connection already.
	 */
	@Test
	public void testCommentStampsShareOneConnection() throws Exception {
		final TreeName[] trees = {
			new TreeName("o=commentSweep", "dn2id"),
			new TreeName("o=commentSweep", "id2entry"),
			new TreeName("o=commentSweep", "state") };
		final AtomicInteger connects = new AtomicInteger();
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			Connection newStampConnection(Dialect dialect) throws SQLException {
				connects.incrementAndGet();
				return super.newStampConnection(dialect);
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true); // freshly created: every one of them is stamped
					}
				}
			});
			for (final TreeName tree : trees) {
				assertEquals(readTableComment(storage.getTableName(tree)), tree.toString());
			}
			assertEquals(connects.get(), 1, "the stamps of one open did not share a connection");
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true);
					}
				}
			});
			// three trees, one more connect: the open that finds every comment in place issues no
			// statement and takes no lock, and pays one connection for the whole sweep either way
			assertEquals(connects.get(), 2, "the trees of an open that found every comment in place did not share a connection");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						for (final TreeName tree : trees) {
							txn.deleteTree(tree);
						}
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * The bound a stamp connection is given must survive a stamp that failed. Postgres undoes a
	 * plain SET when the transaction that ran it is rolled back, and a failed stamp is rolled
	 * back with the connection kept and reused - one connection serves every tree of a backend
	 * open - so every tree stamped after the first failure used to run with no bound at all,
	 * which is what the bound exists to prevent.
	 */
	@Test
	public void testLockBoundSurvivesAFailedStamp() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		// no table was ever created for this tree, so its comment statement fails - on a connection
		// that stays usable, which is the case the session rolls back rather than replaces
		final TreeName missing = new TreeName("o=lockBound", "neverCreated");
		try {
			storage.open(AccessMode.READ_WRITE);
			final JDBCStorage.Dialect dialect = dialect();
			final String bound;
			try (final Connection fresh = storage.newStampConnection(dialect)) {
				bound = sessionLockBound(fresh); // what a connection carrying the bound reports
			}
			try (final JDBCStorage.StampSession session = storage.new StampSession()) {
				assertEquals(storage.commentTable(missing, dialect, session), JDBCStorage.CommentResult.FAILED,
					"stamping a table that does not exist was reported as done");
				if (bound != null) { // oracle: ddl_lock_timeout is only in v$parameter, which the test user cannot read
					assertEquals(sessionLockBound(session.connection(dialect)), bound,
						"the lock bound was lost when the failed stamp was rolled back");
				}
			}
		} finally {
			storage.close();
		}
	}

	/**
	 * A stamp that lost its connection ends the sweep it happened in: every tree behind it needs
	 * that same connection, so each would pay the same connect attempt again. That is about 25 of
	 * them for a stock suffix, all for a diagnostic aid. Nothing is remembered, so the next open
	 * tries again.
	 */
	@Test
	public void testConnectionFailureEndsTheSweep() throws Exception {
		final TreeName[] trees = {
			new TreeName("o=sweepGiveUp", "dn2id"),
			new TreeName("o=sweepGiveUp", "id2entry"),
			new TreeName("o=sweepGiveUp", "state") };
		final AtomicInteger stampAttempts = new AtomicInteger();
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			Connection newStampConnection(Dialect dialect) throws SQLException {
				stampAttempts.incrementAndGet();
				throw new SQLException("injected connection failure", "08006"); // connection exception: the session is gone
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true);
					}
				}
			});
			assertEquals(stampAttempts.get(), 1, "a connection that was gone was paid once per tree of the same open");
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true);
					}
				}
			});
			assertEquals(stampAttempts.get(), 2, "the open after a lost connection did not try again");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						for (final TreeName tree : trees) {
							txn.deleteTree(tree);
						}
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * A lock belongs to the table it is held on, so a stamp that gave up on one must not cost the
	 * trees behind it their comments: the trees of an open are stamped in a fixed order, and a
	 * table left permanently contended by another session would otherwise mean nothing is ever
	 * stamped, on any open. Nothing is remembered either - the open that follows stamps the table
	 * whose moment has passed.
	 * <p>
	 * The failure is injected at the readback rather than at the comment statement, which is built
	 * inline; what is under test is the classification of the failure and what the sweep does with
	 * it, and those do not depend on which of the two statements produced it.
	 */
	@Test
	public void testContendedTableDoesNotEndTheSweep() throws Exception {
		final TreeName[] trees = {
			new TreeName("o=sweepContended", "dn2id"),
			new TreeName("o=sweepContended", "id2entry"),
			new TreeName("o=sweepContended", "state") };
		final AtomicInteger contended = new AtomicInteger(1); // the first tree, for one sweep only
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			String readStoredComment(Connection con, Dialect dialect, String tableName) throws SQLException {
				if (tableName.equals(getTableName(trees[0])) && contended.getAndDecrement() > 0) {
					throw lockTimeoutOf(dialect); // as if another session held this one table
				}
				return super.readStoredComment(con, dialect, tableName);
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true);
					}
				}
			});
			assertNotEquals(readTableComment(storage.getTableName(trees[0])), trees[0].toString(),
				"the contended table was stamped anyway");
			for (int i = 1; i < trees.length; i++) {
				assertEquals(readTableComment(storage.getTableName(trees[i])), trees[i].toString(),
					"one contended table cost the trees behind it their comments");
			}
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true);
					}
				}
			});
			assertEquals(readTableComment(storage.getTableName(trees[0])), trees[0].toString(),
				"a table left unstamped by a contended moment was not stamped by the open that followed");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						for (final TreeName tree : trees) {
							txn.deleteTree(tree);
						}
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/** The failure a dialect reports when a statement gave up on the lock bound it was given. */
	static SQLException lockTimeoutOf(JDBCStorage.Dialect dialect) {
		switch (dialect) {
		case POSTGRES:
			return new SQLException("canceling statement due to lock timeout", "55P03");
		case MYSQL:
			return new SQLException("Lock wait timeout exceeded; try restarting transaction", "HY000", 1205);
		case ORACLE:
			return new SQLException("ORA-00054: resource busy and acquire with NOWAIT specified", "61000", 54);
		case MICROSOFT:
			return new SQLException("Lock request time out period exceeded", "HY000", 1222);
		default:
			throw new IllegalStateException("no lock timeout failure for dialect " + dialect);
		}
	}

	/**
	 * @@sql_mode decides whether a backslash escapes inside the comment literal. It belongs to
	 * the session, and the stamps of one open share a connection, so it is asked once for the
	 * whole sweep rather than once per tree - and only on mysql, the one engine whose literal
	 * depends on it.
	 */
	@Test
	public void testSqlModeProbedOncePerSweep() throws Exception {
		final TreeName[] trees = {
			new TreeName("o=sqlModeProbe", "dn2id"),
			new TreeName("o=sqlModeProbe", "id2entry"),
			new TreeName("o=sqlModeProbe", "state") };
		final AtomicInteger probes = new AtomicInteger();
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			boolean isMysqlBackslashEscape(Connection con) throws SQLException {
				probes.incrementAndGet();
				return super.isMysqlBackslashEscape(con);
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					for (final TreeName tree : trees) {
						txn.openTree(tree, true); // freshly created: every one of them is stamped
					}
				}
			});
			for (final TreeName tree : trees) {
				assertEquals(readTableComment(storage.getTableName(tree)), tree.toString());
			}
			assertEquals(probes.get(), dialect() == JDBCStorage.Dialect.MYSQL ? 1 : 0,
				"the sql mode of one sweep was not asked exactly once");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						for (final TreeName tree : trees) {
							txn.deleteTree(tree);
						}
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	// The bounds of a stamp connection are covered by StampConnectionTestCase: what they are worth
	// is whether they reach the driver and whether the driver then gives up on a server that never
	// answers, and neither needs - nor can be staged by - a database container.

	/**
	 * An import that failed or was cancelled leaves trees holding an incomplete import that is
	 * going to be run again: refreshing statistics of it describes data nobody will query, and on
	 * oracle it is a full scan per table between the failure and its report.
	 */
	@Test
	public void testAbortedImportSkipsStatistics() throws Exception {
		final TreeName tree = new TreeName("o=abortedImport", "dn2id");
		final AtomicInteger refreshes = new AtomicInteger();
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null) {
			@Override
			boolean updateTableStatistics(Connection con, Collection<TreeName> trees) {
				refreshes.incrementAndGet();
				return super.updateTableStatistics(con, trees);
			}
		};
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			try (final Importer importer = storage.startImport()) {
				importer.put(tree, key(1), value(1));
				importer.aborted(); // what OnDiskMergeImporter reports when the import throws or is cancelled
			}
			assertEquals(refreshes.get(), 0, "statistics were refreshed for an import that was aborted");
			try (final Importer importer = storage.startImport()) {
				importer.put(tree, key(2), value(2));
			}
			assertEquals(refreshes.get(), 1, "statistics were not refreshed for an import that finished");
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * The statistics refresh must be possible to turn off: on oracle it gathers with
	 * AUTO_SAMPLE_SIZE, a full scan of every table the import wrote.
	 */
	@Test
	public void testStatisticsRefreshCanBeTurnedOff() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("o=statisticsOff", "dn2id");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			System.setProperty(JDBCStorage.STATISTICS_PROPERTY, "false");
			try (final Connection con = CachedConnection.getConnection(getJdbcUrl())) {
				assertFalse(storage.updateTableStatistics(con, Collections.singleton(tree)),
					"the refresh ran with " + JDBCStorage.STATISTICS_PROPERTY + "=false");
			}
		} finally {
			System.clearProperty(JDBCStorage.STATISTICS_PROPERTY);
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/** deleteTree() must forget the tree: statistics refresh iterates known trees and must skip dropped tables. */
	@Test
	public void testDeleteTreeForgetsTree() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("o=deleteTree", "dn2id");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			assertTrue(storage.listTrees().contains(tree));
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.deleteTree(tree);
				}
			});
			assertFalse(storage.listTrees().contains(tree), "deleteTree() left the tree in the tree-to-table cache");
		} finally {
			storage.close();
		}
	}

	/** A bulk import must refresh optimizer statistics: fresh tables were never analyzed (#859). */
	@Test
	public void testImportRefreshesTableStatistics() throws Exception {
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("testImportAnalyze", "tree");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
				}
			});
			suspendAutomaticStatistics(storage.getTableName(tree));
			try (final Importer importer = storage.startImport()) {
				for (int i = 0; i < 40; i++) {
					importer.put(tree, key(i), value(i));
				}
			}
			assertTableStatisticsFresh(storage.getTableName(tree));
			// import swallows statistics failures by design: assert directly that the
			// dialect-specific refresh statement is accepted by this database
			try (final Connection con = CachedConnection.getConnection(getJdbcUrl())) {
				assertTrue(storage.updateTableStatistics(con, Collections.singleton(tree)), "statistics refresh reported failures");
			}
		} finally {
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}

	/**
	 * Suspends the automatic statistics upkeep of the engines that have it, so that what the
	 * assertion below sees was produced by the refresh of the import and by nothing else: InnoDB
	 * recalculates innodb_table_stats.n_rows on its own (innodb_stats_auto_recalc is on by
	 * default), which would let the assertion pass with no "analyze table" ever issued.
	 */
	void suspendAutomaticStatistics(String tableName) throws Exception {
		final String url = getJdbcUrl();
		if (!url.startsWith("jdbc:mysql")) {
			return; // nothing refreshes what is asserted below on the other engines within a test run
		}
		try (final Connection con = DriverManager.getConnection(url);
			 final Statement st = con.createStatement()) {
			st.execute("alter table " + tableName + " stats_auto_recalc=0");
		}
	}

	void assertTableStatisticsFresh(String tableName) throws Exception {
		final String url = getJdbcUrl();
		final String sql;
		if (url.startsWith("jdbc:postgresql")) {
			// reltuples stays -1/0 until the first ANALYZE
			sql = "select reltuples::bigint from pg_class where relname='" + tableName + "'";
		} else if (url.startsWith("jdbc:oracle")) {
			// num_rows stays null until dbms_stats gathers statistics
			sql = "select num_rows from user_tables where table_name='" + tableName.toUpperCase() + "'";
		} else if (url.startsWith("jdbc:mysql")) {
			// n_rows in the persistent stats table is refreshed by ANALYZE TABLE, and - with the
			// automatic recalculation suspended above - by nothing else: it stays 0 without it
			sql = "select n_rows from mysql.innodb_table_stats where database_name=database() and table_name='" + tableName + "'";
		} else if (url.startsWith("jdbc:sqlserver")) {
			// last_updated stays null until the first UPDATE STATISTICS
			sql = "select count(*) from sys.stats s cross apply sys.dm_db_stats_properties(s.object_id, s.stats_id) p"
				+ " where s.object_id=object_id('" + tableName + "') and p.last_updated is not null";
		} else {
			throw new SkipException("no statistics query for " + url);
		}
		try (final Connection con = DriverManager.getConnection(url);
			 final Statement st = con.createStatement();
			 final ResultSet rs = st.executeQuery(sql)) {
			assertTrue(rs.next(), "table " + tableName + " not found");
			final long rows = rs.getLong(1);
			assertFalse(rs.wasNull(), "statistics were never gathered for " + tableName);
			assertTrue(rows > 0, "statistics of " + tableName + " look stale: " + rows);
		}
	}

	/** Cursor operations must keep working when the tree spans several "fetchsize" batches. */
	@Test
	public void testCursorCrossesFetchSizeBatches() throws Exception {
		System.setProperty("org.openidentityplatform.opendj.jdbc.fetchsize", "2");
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("testCursorBatch", "tree");
		try {
			storage.open(AccessMode.READ_WRITE);
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
					for (int i = 0; i < 7; i++) {
						txn.put(tree, key(i), value(i));
					}
				}
			});
			storage.read(new ReadOperation<Void>() {
				@Override
				public Void run(ReadableTransaction txn) throws Exception {
					try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
						for (int i = 0; i < 7; i++) {
							assertTrue(cursor.next(), "next() at " + i);
							assertEquals(cursor.getKey(), key(i));
							assertEquals(cursor.getValue(), value(i));
						}
						assertFalse(cursor.next());
						assertFalse(cursor.isDefined());
						try {
							cursor.getKey();
							fail("getKey() on undefined cursor must fail");
						} catch (NoSuchElementException expected) {}

						assertTrue(cursor.positionToKeyOrNext(key(3)));
						assertEquals(cursor.getKey(), key(3));
						assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key031")));
						assertEquals(cursor.getKey(), key(4));
						assertTrue(cursor.next());
						assertEquals(cursor.getKey(), key(5));
						assertTrue(cursor.next());
						assertEquals(cursor.getKey(), key(6));
						assertFalse(cursor.next());
						assertFalse(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("z")));
						assertFalse(cursor.isDefined());

						assertTrue(cursor.positionToKey(key(5)));
						assertEquals(cursor.getValue(), value(5));
						assertTrue(cursor.next());
						assertEquals(cursor.getKey(), key(6));
						assertFalse(cursor.positionToKey(ByteString.valueOfUtf8("key99")));
						assertFalse(cursor.isDefined());

						assertTrue(cursor.positionToIndex(0));
						assertEquals(cursor.getKey(), key(0));
						assertTrue(cursor.positionToIndex(5));
						assertEquals(cursor.getKey(), key(5));
						assertTrue(cursor.next());
						assertEquals(cursor.getKey(), key(6));
						assertFalse(cursor.positionToIndex(7));
						assertFalse(cursor.positionToIndex(-1));

						assertTrue(cursor.positionToLastKey());
						assertEquals(cursor.getKey(), key(6));
						assertFalse(cursor.next());

						assertTrue(cursor.positionToKey(key(0)));
						try {
							cursor.delete();
							fail("delete() on read-only cursor must fail");
						} catch (UnsupportedOperationException expected) {}
					}
					return null;
				}
			});
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
						assertTrue(cursor.positionToKey(key(3)));
						cursor.delete();
						assertTrue(cursor.next());
						assertEquals(cursor.getKey(), key(4));
					}
					assertNull(txn.read(tree, key(3)));
					assertEquals(txn.getRecordCount(tree), 6);
				}
			});
		} finally {
			System.clearProperty("org.openidentityplatform.opendj.jdbc.fetchsize");
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}
	/**
	 * Two or more <em>distinct new</em> keys written into one tree per transaction is the shape the primary key
	 * seek made able to deadlock: on the NOT MATCHED path the seek range-locks the gap before the next existing
	 * key, that lock is self-incompatible, and the key hash scatters logically ordered keys across the index, so
	 * two writers inserting different keys can each end up holding what the other needs. The ascending key order
	 * that {@code IndexBuffer} maintains does not help there. Nothing may escape {@link JDBCStorage#write}, which
	 * replays the conflict, and no record may be lost to it (#867).
	 */
	@Test(timeOut = 600000)
	public void testConcurrentWritersInsertingDistinctKeys() throws Exception {
		final int writers = 4;
		final int rounds = 25;
		final int keysPerTransaction = 3;
		final int seeded = 10;
		final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
		final TreeName tree = new TreeName("testConcurrentInsert", "tree");
		final ExecutorService executor = Executors.newFixedThreadPool(writers);
		try {
			storage.open(AccessMode.READ_WRITE);
			// seeded, so that every insert below takes the NOT MATCHED path with a gap to lock in front of it
			storage.write(new WriteOperation() {
				@Override
				public void run(WriteableTransaction txn) throws Exception {
					txn.openTree(tree, true);
					for (int i = 0; i < seeded; i++) {
						txn.put(tree, key(i), value(i));
					}
				}
			});
			final List<Callable<Void>> concurrent = new ArrayList<>();
			for (int writer = 0; writer < writers; writer++) {
				final int id = writer;
				concurrent.add(new Callable<Void>() {
					@Override
					public Void call() throws Exception {
						for (int round = 0; round < rounds; round++) {
							final int current = round;
							storage.write(new WriteOperation() {
								@Override
								public void run(WriteableTransaction txn) throws Exception {
									for (int i = 0; i < keysPerTransaction; i++) {
										txn.put(tree,
												ByteString.valueOfUtf8(String.format("w%02d-r%03d-k%d", id, current, i)),
												value(i));
									}
								}
							});
						}
						return null;
					}
				});
			}
			for (final Future<Void> written : executor.invokeAll(concurrent)) {
				// a conflict the storage did not replay surfaces here, as it would reach an LDAP client
				written.get();
			}
			storage.read(new ReadOperation<Void>() {
				@Override
				public Void run(ReadableTransaction txn) throws Exception {
					assertEquals(txn.getRecordCount(tree), seeded + writers * rounds * keysPerTransaction);
					return null;
				}
			});
		} finally {
			executor.shutdownNow();
			try {
				storage.write(new WriteOperation() {
					@Override
					public void run(WriteableTransaction txn) throws Exception {
						txn.deleteTree(tree);
					}
				});
			} catch (Exception ignored) {}
			storage.close();
		}
	}
}