GH-0000 test: reproduce LMDB reader exhaustion after timeouts

Author: hmottestad

tools/server-boot/src/test/java/org/eclipse/rdf4j/tools/serverboot/LmdbTimedOutQueryReadHandleTest.java

@@ -0,0 +1,395 @@
+/*******************************************************************************
+ * Copyright (c) 2026 Eclipse RDF4J contributors.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Distribution License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ *******************************************************************************/
+// Some portions generated by Codex
+package org.eclipse.rdf4j.tools.serverboot;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.rdf4j.http.protocol.Protocol;
+import org.eclipse.rdf4j.model.ValueFactory;
+import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
+import org.eclipse.rdf4j.repository.Repository;
+import org.eclipse.rdf4j.repository.RepositoryConnection;
+import org.eclipse.rdf4j.repository.RepositoryException;
+import org.eclipse.rdf4j.repository.config.RepositoryConfig;
+import org.eclipse.rdf4j.repository.config.RepositoryConfigException;
+import org.eclipse.rdf4j.repository.manager.RemoteRepositoryManager;
+import org.eclipse.rdf4j.repository.sail.config.SailRepositoryConfig;
+import org.eclipse.rdf4j.sail.lmdb.config.LmdbStoreConfig;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.web.server.LocalServerPort;
+import org.springframework.http.HttpStatus;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+
+@SpringBootTest(classes = Rdf4jServerWorkbenchApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+class LmdbTimedOutQueryReadHandleTest {
+
+	private static final int STATEMENT_COUNT = Integer.getInteger("rdf4j.lmdb.timeout.statementCount", 1024);
+	private static final int QUERY_COUNT = Integer.getInteger("rdf4j.lmdb.timeout.queryCount", 120);
+	private static final int QUERY_WORKERS = Integer.getInteger("rdf4j.lmdb.timeout.queryWorkers", 8);
+	private static final int MAX_CAPTURED_RESPONSE_BYTES = 8192;
+	private static final String PREDICATE = "urn:example:p";
+	private static final String SLOW_QUERY = "PREFIX ex: <urn:example:>\n"
+			+ "SELECT distinct * WHERE {\n"
+			+ "  ?s1 ex:p ?o1 .\n"
+			+ "  ?s2 ex:p ?o2 .\n"
+			+ "  ?s3 ex:p ?o3 .\n"
+			+ "}";
+	private static final String HEALTH_QUERY = "ASK { ?s ?p ?o }";
+
+	@LocalServerPort
+	private int port;
+
+	private final ValueFactory valueFactory = SimpleValueFactory.getInstance();
+	private final List<String> createdRepositories = new ArrayList<>();
+	private final List<Logger> lmdbLoggers = new ArrayList<>();
+	private ListAppender<ILoggingEvent> lmdbLogAppender;
+	private RemoteRepositoryManager repositoryManager;
+
+	@BeforeEach
+	void setUp() {
+		attachLmdbLogAppender();
+		repositoryManager = RemoteRepositoryManager.getInstance(serverUrl());
+	}
+
+	@AfterEach
+	void tearDown() {
+		try {
+			if (repositoryManager == null) {
+				return;
+			}
+			for (String repositoryId : createdRepositories) {
+				try {
+					repositoryManager.removeRepository(repositoryId);
+				} catch (RepositoryException ignored) {
+					// best-effort cleanup
+				}
+			}
+			createdRepositories.clear();
+			repositoryManager.shutDown();
+			repositoryManager = null;
+		} finally {
+			detachLmdbLogAppender();
+		}
+	}
+
+	@Test
+	void lmdbRepositoryStillAcceptsQueriesAfterManyTimedOutServerQueries() throws Exception {
+		String repositoryId = registerLmdbRepository();
+		loadData(repositoryId);
+
+		QueryStats stats = runTimedQueries(repositoryId);
+		System.out.println("LMDB timeout reproducer stats: timeouts=" + stats.timeouts
+				+ ", readerHandleFailures=" + stats.readerHandleFailures
+				+ ", failureSamples=" + stats.failureSamples);
+
+		assertThat(stats.timeouts)
+				.as("at least one query should hit the one-second server-side timeout; failures: %s",
+						stats.failureSamples)
+				.isPositive();
+		assertThat(stats.readerHandleFailures)
+				.as("reader handle failures: %s", stats.failureSamples)
+				.isZero();
+		assertThat(lmdbReaderHandleLogEvents())
+				.as("LMDB reader handle log events")
+				.isEmpty();
+
+		assertEventuallyHealthy(repositoryId);
+	}
+
+	private String registerLmdbRepository() throws RepositoryException, RepositoryConfigException {
+		String repositoryId = "lmdb-timeout-" + UUID.randomUUID();
+		RepositoryConfig config = new RepositoryConfig(repositoryId,
+				new SailRepositoryConfig(new LmdbStoreConfig()));
+		repositoryManager.addRepositoryConfig(config);
+		createdRepositories.add(repositoryId);
+		return repositoryId;
+	}
+
+	private void loadData(String repositoryId) {
+		Repository repository = repositoryManager.getRepository(repositoryId);
+		repository.init();
+		try (RepositoryConnection connection = repository.getConnection()) {
+			for (int i = 0; i < STATEMENT_COUNT; i++) {
+				connection.add(valueFactory.createIRI("urn:example:s" + i), valueFactory.createIRI(PREDICATE),
+						valueFactory.createLiteral("value-" + i));
+			}
+		} finally {
+			repository.shutDown();
+		}
+	}
+
+	private QueryStats runTimedQueries(String repositoryId) throws InterruptedException, ExecutionException {
+		ExecutorService executor = Executors.newFixedThreadPool(QUERY_WORKERS);
+		List<Future<QueryResponse>> futures = new ArrayList<>(QUERY_COUNT);
+		try {
+			for (int i = 0; i < QUERY_COUNT; i++) {
+				futures.add(executor.submit(timedQuery(repositoryId)));
+			}
+
+			QueryStats stats = new QueryStats();
+			for (Future<QueryResponse> future : futures) {
+				stats.record(future.get());
+			}
+			return stats;
+		} finally {
+			executor.shutdownNow();
+		}
+	}
+
+	private final Random random = new Random();
+
+	private Callable<QueryResponse> timedQuery(String repositoryId) {
+		return () -> {
+			try {
+				return executeQuery(repositoryId, SLOW_QUERY, random.nextInt(3) + 1);
+			} catch (Exception e) {
+				return new QueryResponse(-1, exceptionMessage(e));
+			}
+		};
+	}
+
+	private QueryResponse executeQuery(String repositoryId, String query, int timeoutSeconds)
+			throws IOException, InterruptedException {
+		int readTimeoutMillis = timeoutSeconds > 0 ? 3000 : 10000;
+		return executeQuery(repositoryId, query, timeoutSeconds, readTimeoutMillis);
+	}
+
+	private QueryResponse executeQuery(String repositoryId, String query, int timeoutSeconds, int readTimeoutMillis)
+			throws IOException, InterruptedException {
+		String form = formValue(Protocol.QUERY_PARAM_NAME, query);
+		if (timeoutSeconds > 0) {
+			form += "&" + formValue(Protocol.TIMEOUT_PARAM_NAME, Integer.toString(timeoutSeconds));
+		}
+
+		HttpURLConnection connection = (HttpURLConnection) repositoryUri(repositoryId).toURL()
+				.openConnection();
+		connection.setConnectTimeout(50000);
+		connection.setReadTimeout(readTimeoutMillis);
+		connection.setRequestMethod("POST");
+		connection.setRequestProperty("Content-Type", Protocol.FORM_MIME_TYPE);
+		connection.setRequestProperty("Accept", "application/sparql-results+json");
+		connection.setDoOutput(true);
+
+		byte[] body = form.getBytes(StandardCharsets.UTF_8);
+		connection.setFixedLengthStreamingMode(body.length);
+		try (OutputStream output = connection.getOutputStream()) {
+			output.write(body);
+		}
+
+		try {
+			int status = connection.getResponseCode();
+			return new QueryResponse(status, readResponseBody(connection, status));
+		} catch (SocketTimeoutException e) {
+			return new QueryResponse(-1, exceptionMessage(e));
+		} finally {
+			connection.disconnect();
+		}
+	}
+
+	private void assertEventuallyHealthy(String repositoryId) throws IOException, InterruptedException {
+		QueryResponse response = null;
+		long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
+		while (System.nanoTime() < deadline) {
+			response = executeQuery(repositoryId, HEALTH_QUERY, 0, 1000);
+			if (response.status == HttpStatus.OK.value()) {
+				return;
+			}
+			Thread.sleep(250);
+		}
+
+		assertThat(response)
+				.as("health query response")
+				.isNotNull();
+		assertThat(response.status)
+				.as("health query response body: %s", response.body)
+				.isEqualTo(HttpStatus.OK.value());
+	}
+
+	private String readResponseBody(HttpURLConnection connection, int status) throws IOException {
+		InputStream stream = status >= 400 ? connection.getErrorStream() : connection.getInputStream();
+		if (stream == null) {
+			return "";
+		}
+
+		StringBuilder response = new StringBuilder();
+		byte[] buffer = new byte[1024];
+		int captured = 0;
+		try (InputStream input = stream) {
+			int read;
+			while ((read = input.read(buffer)) != -1) {
+				if (captured < MAX_CAPTURED_RESPONSE_BYTES) {
+					int length = Math.min(read, MAX_CAPTURED_RESPONSE_BYTES - captured);
+					response.append(new String(buffer, 0, length, StandardCharsets.UTF_8));
+					captured += length;
+				}
+			}
+		} catch (SocketTimeoutException e) {
+			if (response.length() > 0) {
+				response.append('\n');
+			}
+			response.append(exceptionMessage(e));
+		}
+		return response.toString();
+	}
+
+	private String formValue(String name, String value) {
+		return URLEncoder.encode(name, StandardCharsets.UTF_8) + "="
+				+ URLEncoder.encode(value, StandardCharsets.UTF_8);
+	}
+
+	private String exceptionMessage(Throwable throwable) {
+		StringBuilder message = new StringBuilder();
+		Throwable cursor = throwable;
+		while (cursor != null) {
+			if (message.length() > 0) {
+				message.append(" -> ");
+			}
+			message.append(cursor.getClass().getSimpleName())
+					.append(": ")
+					.append(cursor.getMessage());
+			Throwable next = cursor.getCause();
+			if (next == null || next == cursor) {
+				break;
+			}
+			cursor = next;
+		}
+		return message.toString();
+	}
+
+	private URI repositoryUri(String repositoryId) {
+		return URI.create(serverUrl() + "/repositories/" + repositoryId);
+	}
+
+	private String serverUrl() {
+		return "http://localhost:" + port + "/rdf4j-server";
+	}
+
+	private void attachLmdbLogAppender() {
+		lmdbLogAppender = new ListAppender<>();
+		lmdbLogAppender.start();
+		attachLmdbLogger("org.eclipse.rdf4j.sail.lmdb.LmdbUtil");
+		attachLmdbLogger("org.eclipse.rdf4j.sail.lmdb.LmdbEvaluationStatistics");
+	}
+
+	private void attachLmdbLogger(String loggerName) {
+		Logger logger = (Logger) LoggerFactory.getLogger(loggerName);
+		logger.addAppender(lmdbLogAppender);
+		lmdbLoggers.add(logger);
+	}
+
+	private void detachLmdbLogAppender() {
+		for (Logger logger : lmdbLoggers) {
+			logger.detachAppender(lmdbLogAppender);
+		}
+		lmdbLoggers.clear();
+		if (lmdbLogAppender != null) {
+			lmdbLogAppender.stop();
+			lmdbLogAppender = null;
+		}
+	}
+
+	private List<String> lmdbReaderHandleLogEvents() {
+		List<String> events = new ArrayList<>();
+		for (ILoggingEvent event : lmdbLogAppender.list) {
+			String throwableMessage = throwableMessage(event);
+			if (containsReaderHandleFailure(event.getFormattedMessage())
+					|| containsReaderHandleFailure(throwableMessage)) {
+				events.add(event.getLoggerName() + ": " + event.getFormattedMessage() + " " + throwableMessage);
+			}
+		}
+		return events;
+	}
+
+	private String throwableMessage(ILoggingEvent event) {
+		if (event.getThrowableProxy() == null) {
+			return "";
+		}
+		return event.getThrowableProxy().getMessage();
+	}
+
+	private static boolean containsReaderHandleFailure(String value) {
+		String lowerValue = value.toLowerCase();
+		return lowerValue.contains("mdb_readers_full") || lowerValue.contains("maxreaders");
+	}
+
+	private static final class QueryStats {
+		private int timeouts;
+		private int readerHandleFailures;
+		private final List<String> failureSamples = new ArrayList<>();
+
+		private void record(QueryResponse response) {
+			if (isTimeout(response)) {
+				timeouts++;
+				return;
+			}
+			if (isReaderHandleFailure(response)) {
+				readerHandleFailures++;
+			}
+			if (!isSuccess(response) && failureSamples.size() < 5) {
+				failureSamples.add("HTTP " + response.status + ": " + response.body);
+			}
+		}
+
+		private boolean isSuccess(QueryResponse response) {
+			return response.status >= 200 && response.status < 300;
+		}
+
+		private boolean isTimeout(QueryResponse response) {
+			return (response.status == HttpStatus.SERVICE_UNAVAILABLE.value()
+					&& response.body.contains("Query evaluation took too long"))
+					|| response.body.contains("Query evaluation took too long")
+					|| response.body.contains("SocketTimeoutException")
+					|| response.body.contains("Unexpected end of file from server")
+					|| response.body.contains("Connection reset");
+		}
+
+		private boolean isReaderHandleFailure(QueryResponse response) {
+			return containsReaderHandleFailure(response.body);
+		}
+	}
+
+	private static final class QueryResponse {
+		private final int status;
+		private final String body;
+
+		private QueryResponse(int status, String body) {
+			this.status = status;
+			this.body = body == null ? "" : body;
+		}
+	}
+}