From cb7a149b1966e3ebac75c6348251b0d1db171c5f Mon Sep 17 00:00:00 2001 From: corneliusroemer-agent <299456996+corneliusroemer-agent@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:30:57 +0000 Subject: [PATCH 1/3] Add a warm validation server so the JVM is reused between validations Validating a submission with `java -jar readtools.jar` spends roughly half its wall time on classloading and JIT warm-up, which the process then throws away. Profiling put JVM boot itself at 0.036s, so what a fresh process discards is the C2-compiled parse/validate loop, not startup. `java -jar readtools.jar server` keeps one JVM warm and validates over loopback HTTP instead. On a 100k-read pair that is ~1.3s per cold CLI run against ~0.9s warm, and ~0.3s when four run concurrently in the one JVM. Concurrency is safe because the v2 validation path holds no shared mutable state: every validation builds its own ValidatorWrapper, and the only non-final statics on the path are InsdcReadsValidator's error strings, which are never reassigned. /validate returns the exact stdout, stderr and exit code the CLI would have produced rather than a schema of its own, so callers that already parse the CLI's output keep working and cannot drift from it. ValidationRunner is the single implementation both entry points call, and ValidateServerTest asserts the two agree. Validation concurrency is bounded by a semaphore rather than by the HTTP thread pool, so /health never queues behind in-flight validations. A fixed pool would make a merely busy server fail its liveness probe and get restarted, which in the Loculus pod also evicts deacon's multi-gigabyte in-memory index. The server warms itself on synthetic reads and reports unhealthy until that finishes, so the first real request does not pay the cold cost. /health then runs a real one-read validation, so a JVM that is up but whose validation path is broken does not report healthy. Co-Authored-By: Claude Opus 5 (1M context) --- .../ebi/ena/readtools/v2/cli/ValidateCli.java | 55 +-- .../ena/readtools/v2/cli/ValidateServer.java | 364 ++++++++++++++++++ .../readtools/v2/cli/ValidationRunner.java | 81 ++++ .../readtools/v2/cli/ValidateServerTest.java | 235 +++++++++++ 4 files changed, 691 insertions(+), 44 deletions(-) create mode 100644 src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java create mode 100644 src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidationRunner.java create mode 100644 src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java diff --git a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateCli.java b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateCli.java index fae14f2..ba6ad25 100644 --- a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateCli.java +++ b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateCli.java @@ -16,20 +16,15 @@ import java.util.List; import java.util.stream.Collectors; import uk.ac.ebi.ena.readtools.v2.FileFormat; -import uk.ac.ebi.ena.readtools.v2.validator.ReadsValidationException; -import uk.ac.ebi.ena.readtools.v2.validator.ValidatorWrapper; /** - * Standalone CLI wrapper around {@link ValidatorWrapper}, running the same client-side read - * validation that webin-cli performs, without the manifest / upload machinery. + * Standalone CLI wrapper around the v2 validator, running the same client-side read validation that + * webin-cli performs, without the manifest / upload machinery. * *

Usage: java -cp readtools-all.jar uk.ac.ebi.ena.readtools.v2.cli.ValidateCli \ --format FASTQ * [--full] file1 [file2] */ public class ValidateCli { - // webin-cli's own limits: quick = first 100k reads, extended = first 100M reads. - private static final long QUICK_READ_LIMIT = 100_000L; - private static final long EXTENDED_READ_LIMIT = 100_000_000L; @Parameter( names = {"--format", "-f"}, @@ -52,8 +47,15 @@ public class ValidateCli { private boolean help = false; public static void main(String[] args) { + // `java -jar readtools.jar server ...` starts the warm validation server instead. + if (args.length > 0 && "server".equals(args[0])) { + ValidateServer.main(java.util.Arrays.copyOfRange(args, 1, args.length)); + return; + } + ValidateCli cli = new ValidateCli(); - JCommander jc = JCommander.newBuilder().addObject(cli).programName("readtools-validate").build(); + JCommander jc = + JCommander.newBuilder().addObject(cli).programName("readtools-validate").build(); try { jc.parse(args); @@ -73,41 +75,6 @@ public static void main(String[] args) { private int run() { List fileList = files.stream().map(File::new).collect(Collectors.toList()); - for (File f : fileList) { - if (!f.isFile()) { - System.err.println("ERROR: file not found: " + f.getPath()); - return 2; - } - } - - long limit = full ? EXTENDED_READ_LIMIT : QUICK_READ_LIMIT; - ValidatorWrapper wrapper = new ValidatorWrapper(fileList, format, limit); - - try { - wrapper.run(); - } catch (ReadsValidationException e) { - System.out.println("RESULT: INVALID"); - System.out.println(" " + e.getMessage()); - return 1; - } catch (RuntimeException e) { - System.out.println("RESULT: INVALID (file structure / parse error)"); - System.out.println(" " + (e.getCause() != null ? e.getCause().getMessage() : e.getMessage())); - return 1; - } - - System.out.println("RESULT: VALID"); - System.out.println(" format: " + format); - System.out.println(" mode: " + (full ? "full (<=100M reads)" : "quick (<=100k reads)")); - if (format == FileFormat.FASTQ && fileList.size() > 1) { - System.out.println(" paired: " + wrapper.isPaired()); - } - wrapper - .getFileQualityStats() - .forEach( - s -> - System.out.printf( - " %s: reads=%d, highQuality(avg>=30)=%d%n", - s.getFile().getName(), s.getReadCount(), s.getHighQualityReadCount())); - return 0; + return ValidationRunner.validate(fileList, format, full, System.out, System.err); } } diff --git a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java new file mode 100644 index 0000000..9cf00ea --- /dev/null +++ b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java @@ -0,0 +1,364 @@ +/* + * Copyright 2010-2021 EMBL - European Bioinformatics Institute + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this + * file except in compliance with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software distributed under the + * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR + * CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package uk.ac.ebi.ena.readtools.v2.cli; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.PrintStream; +import java.io.PrintWriter; +import java.io.UncheckedIOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import uk.ac.ebi.ena.readtools.v2.FileFormat; + +/** + * A long-lived HTTP server that runs the same validation as {@link ValidateCli}, but in a JVM that + * stays warm between requests. + * + *

Why: a cold {@code java -jar readtools.jar} spends roughly half its wall time on classloading + * and JIT warm-up that a fresh process throws away every call. Keeping one JVM warm removes that, + * and — because the validation path holds no shared mutable state — lets several validations run + * concurrently in it rather than forking one JVM per file pair. + * + *

Protocol (JSON over HTTP, bound to loopback by default): + * + * 

+ *   POST /validate  {"files": ["/path/R1.fastq", "/path/R2.fastq"],
+ *                    "format": "FASTQ", "full": false}
+ *              ->   {"exitCode": 0, "stdout": "RESULT: VALID\n...", "stderr": ""}
+ *   GET  /health    -> 200 {"status":"ok"} once warm, 503 while still warming up
+ *
+ * + *

{@code stdout}/{@code stderr} are what the CLI would have printed for the same arguments, so + * callers that already parse the CLI's output keep working unchanged. + */ +public class ValidateServer { + private static final int MAX_REQUEST_BYTES = 1 << 20; + + @Parameter( + names = {"--host"}, + description = "Address to bind to. Loopback by default: this server does no authentication.") + private String host = "127.0.0.1"; + + @Parameter( + names = {"--port", "-p"}, + description = "Port to listen on") + private int port = 5001; + + @Parameter( + names = {"--threads", "-t"}, + description = + "Maximum validations to run concurrently. Further requests queue. Each one holds a" + + " read buffer and a pairing Bloom filter, so this also bounds heap use.") + private int threads = Math.max(1, Runtime.getRuntime().availableProcessors() / 2); + + @Parameter( + names = {"--warmup-rounds"}, + description = + "Validations of synthetic reads to run before reporting healthy, so the first real" + + " request does not pay JIT warm-up. 0 disables warm-up.") + private int warmupRounds = 3; + + @Parameter( + names = {"--warmup-reads"}, + description = "Reads per synthetic warm-up file") + private int warmupReads = 100_000; + + @Parameter( + names = {"--help", "-h"}, + help = true) + private boolean help = false; + + private final ObjectMapper mapper = new ObjectMapper(); + private Semaphore validationSlots; + private final AtomicBoolean ready = new AtomicBoolean(false); + private Path healthProbeFile; + + public static void main(String[] args) { + ValidateServer server = new ValidateServer(); + JCommander jc = + JCommander.newBuilder().addObject(server).programName("readtools-validate server").build(); + + try { + jc.parse(args); + } catch (Exception e) { + System.err.println("ERROR: " + e.getMessage()); + jc.usage(); + System.exit(2); + } + + if (server.help) { + jc.usage(); + return; + } + + try { + server.run(); + } catch (IOException e) { + System.err.println("ERROR: could not start server: " + e.getMessage()); + System.exit(2); + } + } + + private void run() throws IOException { + validationSlots = new Semaphore(threads); + + HttpServer httpServer = HttpServer.create(new InetSocketAddress(host, port), 0); + // Without an explicit executor HttpServer runs every handler on the dispatch thread, i.e. + // serially, which would give up the concurrency this server exists to provide. + // + // The pool is deliberately unbounded and the concurrency limit lives in a semaphore instead. + // A fixed pool of --threads would make /health queue behind in-flight validations, so a + // merely busy server would fail its liveness probe and get restarted -- which in the Loculus + // pod also evicts deacon's multi-gigabyte in-memory index. Threads are cheap; what needs + // bounding is the number of validations allocating read buffers at once, and that is exactly + // what the semaphore bounds. + ExecutorService executor = Executors.newCachedThreadPool(); + httpServer.setExecutor(executor); + + httpServer.createContext("/validate", this::handleValidate); + httpServer.createContext("/health", this::handleHealth); + + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + httpServer.stop(0); + executor.shutdownNow(); + })); + + httpServer.start(); + System.err.printf( + "readtools validation server listening on %s:%d (%d threads)%n", host, port, threads); + + healthProbeFile = writeHealthProbeFile(); + warmUp(); + ready.set(true); + System.err.println("readtools validation server ready"); + } + + private void handleValidate(HttpExchange exchange) throws IOException { + try { + if (!"POST".equals(exchange.getRequestMethod())) { + respondError(exchange, 405, "only POST is supported"); + return; + } + + JsonNode request; + try (InputStream body = exchange.getRequestBody()) { + byte[] raw = body.readNBytes(MAX_REQUEST_BYTES); + request = mapper.readTree(raw); + } catch (Exception e) { + respondError(exchange, 400, "could not parse request body: " + e.getMessage()); + return; + } + + List files = new ArrayList<>(); + JsonNode filesNode = request == null ? null : request.get("files"); + if (filesNode == null || !filesNode.isArray() || filesNode.isEmpty()) { + respondError(exchange, 400, "'files' must be a non-empty array of paths"); + return; + } + filesNode.forEach(node -> files.add(new File(node.asText()))); + + FileFormat format; + try { + format = FileFormat.valueOf(request.path("format").asText("FASTQ")); + } catch (IllegalArgumentException e) { + respondError(exchange, 400, "unknown format: " + request.path("format").asText()); + return; + } + boolean full = request.path("full").asBoolean(false); + + try { + validationSlots.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + respondError(exchange, 503, "server is shutting down"); + return; + } + try { + respondJson(exchange, 200, validateToJson(files, format, full)); + } finally { + validationSlots.release(); + } + } catch (RuntimeException e) { + // Never let one bad request kill the handler thread silently. + ObjectNode response = mapper.createObjectNode(); + response.put("exitCode", 2); + response.put("stdout", ""); + response.put("stderr", "ERROR: " + describe(e)); + respondJson(exchange, 500, response); + } + } + + /** Runs one validation, capturing what the CLI would have printed. */ + private ObjectNode validateToJson(List files, FileFormat format, boolean full) { + ByteArrayOutputStream outBytes = new ByteArrayOutputStream(); + ByteArrayOutputStream errBytes = new ByteArrayOutputStream(); + int exitCode; + try (PrintStream out = new PrintStream(outBytes, true, StandardCharsets.UTF_8); + PrintStream err = new PrintStream(errBytes, true, StandardCharsets.UTF_8)) { + try { + exitCode = ValidationRunner.validate(files, format, full, out, err); + } catch (RuntimeException e) { + // ValidateCli would have died here and the caller would have seen a stack trace on + // stderr with no RESULT line; reproduce that shape rather than inventing a verdict. + exitCode = 2; + e.printStackTrace(new PrintWriter(err, true)); + } + } + + ObjectNode response = mapper.createObjectNode(); + response.put("exitCode", exitCode); + response.put("stdout", outBytes.toString(StandardCharsets.UTF_8)); + response.put("stderr", errBytes.toString(StandardCharsets.UTF_8)); + return response; + } + + /** + * Liveness plus a real validation of a one-read file, so a JVM that is up but whose validation + * path is broken (or wedged with every thread busy) does not report healthy. + */ + private void handleHealth(HttpExchange exchange) throws IOException { + ObjectNode response = mapper.createObjectNode(); + + if (!ready.get()) { + response.put("status", "warming-up"); + respondJson(exchange, 503, response); + return; + } + + ObjectNode probe = + validateToJson(List.of(healthProbeFile.toFile()), FileFormat.FASTQ, /* full= */ false); + if (probe.get("exitCode").asInt() != 0) { + response.put("status", "unhealthy"); + response.put("detail", probe.get("stdout").asText() + probe.get("stderr").asText()); + respondJson(exchange, 503, response); + return; + } + + response.put("status", "ok"); + response.put("threads", threads); + response.put("availableSlots", validationSlots.availablePermits()); + respondJson(exchange, 200, response); + } + + /** + * Validates synthetic reads a few times so the hot parse/validate loop is JIT-compiled before the + * first real request. One pass is not enough: the profiling this server is based on saw steady + * state only after three or four. + */ + private void warmUp() { + if (warmupRounds <= 0) { + return; + } + try { + Path dir = Files.createTempDirectory("readtools-warmup"); + dir.toFile().deleteOnExit(); + List pair = + List.of( + writeSyntheticFastq(dir.resolve("warmup_1.fastq"), 1).toFile(), + writeSyntheticFastq(dir.resolve("warmup_2.fastq"), 2).toFile()); + + for (int round = 0; round < warmupRounds; round++) { + long start = System.nanoTime(); + ObjectNode result = validateToJson(pair, FileFormat.FASTQ, /* full= */ false); + System.err.printf( + "warm-up %d/%d: %.2fs (exit %d)%n", + round + 1, + warmupRounds, + (System.nanoTime() - start) / 1e9, + result.get("exitCode").asInt()); + } + + for (File file : pair) { + Files.deleteIfExists(file.toPath()); + } + Files.deleteIfExists(dir); + } catch (IOException | UncheckedIOException e) { + // Warm-up is an optimisation; a failure here must not stop the server serving. + System.err.println("WARNING: warm-up skipped: " + describe(e)); + } + } + + private Path writeSyntheticFastq(Path path, int mate) throws IOException { + String bases = "ACGTACGTACGTACGTACGTACGTACGTACGTACGTACGTACGTACGTACGTACGTACGTACGT"; + String quality = "I".repeat(bases.length()); + StringBuilder buffer = new StringBuilder(1 << 16); + try (var writer = Files.newBufferedWriter(path, StandardCharsets.UTF_8)) { + for (int i = 0; i < warmupReads; i++) { + buffer + .append('@') + .append("warmup") + .append(i) + .append('/') + .append(mate) + .append('\n') + .append(bases) + .append("\n+\n") + .append(quality) + .append('\n'); + if (buffer.length() > 1 << 16) { + writer.write(buffer.toString()); + buffer.setLength(0); + } + } + writer.write(buffer.toString()); + } + path.toFile().deleteOnExit(); + return path; + } + + private Path writeHealthProbeFile() throws IOException { + Path path = Files.createTempFile("readtools-health", ".fastq"); + Files.writeString(path, "@probe\nACGTACGTAC\n+\nIIIIIIIIII\n", StandardCharsets.UTF_8); + path.toFile().deleteOnExit(); + return path; + } + + private static String describe(Throwable e) { + return e.getMessage() == null ? e.getClass().getName() : e.getMessage(); + } + + private void respondError(HttpExchange exchange, int status, String message) throws IOException { + ObjectNode response = mapper.createObjectNode(); + response.put("error", message); + respondJson(exchange, status, response); + } + + private void respondJson(HttpExchange exchange, int status, ObjectNode body) throws IOException { + byte[] bytes = mapper.writeValueAsBytes(body); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(status, bytes.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(bytes); + } + } +} diff --git a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidationRunner.java b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidationRunner.java new file mode 100644 index 0000000..f4b59fe --- /dev/null +++ b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidationRunner.java @@ -0,0 +1,81 @@ +/* + * Copyright 2010-2021 EMBL - European Bioinformatics Institute + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this + * file except in compliance with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software distributed under the + * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR + * CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package uk.ac.ebi.ena.readtools.v2.cli; + +import java.io.File; +import java.io.PrintStream; +import java.util.List; +import uk.ac.ebi.ena.readtools.v2.FileFormat; +import uk.ac.ebi.ena.readtools.v2.validator.ReadsValidationException; +import uk.ac.ebi.ena.readtools.v2.validator.ValidatorWrapper; + +/** + * The validation run itself, factored out of {@link ValidateCli} so that {@link ValidateServer} can + * execute exactly the same code and produce byte-identical output. + * + *

Callers parse the report text, so it is a compatibility surface: {@link ValidateServer} + * returns it verbatim over HTTP and the CLI prints it to stdout. Both must stay identical. + * + *

Stateless and safe to call concurrently: a fresh {@link ValidatorWrapper} is built per call + * and nothing is shared between calls. + */ +public final class ValidationRunner { + // webin-cli's own limits: quick = first 100k reads, extended = first 100M reads. + public static final long QUICK_READ_LIMIT = 100_000L; + public static final long EXTENDED_READ_LIMIT = 100_000_000L; + + private ValidationRunner() {} + + /** + * Validates the given files and writes the report to {@code out}. + * + * @return the exit code {@link ValidateCli} would return: 0 valid, 1 invalid, 2 bad input. + */ + public static int validate( + List files, FileFormat format, boolean full, PrintStream out, PrintStream err) { + for (File f : files) { + if (!f.isFile()) { + err.println("ERROR: file not found: " + f.getPath()); + return 2; + } + } + + long limit = full ? EXTENDED_READ_LIMIT : QUICK_READ_LIMIT; + ValidatorWrapper wrapper = new ValidatorWrapper(files, format, limit); + + try { + wrapper.run(); + } catch (ReadsValidationException e) { + out.println("RESULT: INVALID"); + out.println(" " + e.getMessage()); + return 1; + } catch (RuntimeException e) { + out.println("RESULT: INVALID (file structure / parse error)"); + out.println(" " + (e.getCause() != null ? e.getCause().getMessage() : e.getMessage())); + return 1; + } + + out.println("RESULT: VALID"); + out.println(" format: " + format); + out.println(" mode: " + (full ? "full (<=100M reads)" : "quick (<=100k reads)")); + if (format == FileFormat.FASTQ && files.size() > 1) { + out.println(" paired: " + wrapper.isPaired()); + } + wrapper + .getFileQualityStats() + .forEach( + s -> + out.printf( + " %s: reads=%d, highQuality(avg>=30)=%d%n", + s.getFile().getName(), s.getReadCount(), s.getHighQualityReadCount())); + return 0; + } +} diff --git a/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java b/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java new file mode 100644 index 0000000..dc08fd8 --- /dev/null +++ b/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java @@ -0,0 +1,235 @@ +/* + * Copyright 2010-2021 EMBL - European Bioinformatics Institute + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this + * file except in compliance with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software distributed under the + * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR + * CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package uk.ac.ebi.ena.readtools.v2.cli; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.OutputStream; +import java.io.PrintStream; +import java.net.HttpURLConnection; +import java.net.ServerSocket; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import uk.ac.ebi.ena.readtools.v2.FileFormat; + +public class ValidateServerTest { + private static final String VALID_SINGLE = + "@seq1\nACGTACGTAC\n+\nIIIIIIIIII\n@seq2\nACGTACGTAC\n+\nIIIIIIIIII\n"; + private static final String NON_IUPAC = "@seq1\nACGTAXGTAC\n+\nIIIIIIIIII\n"; + private static final String FASTA_HEADER = ">seq1\nACGTACGTAC\n+\nIIIIIIIIII\n"; + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private static int port; + private static Thread serverThread; + private static Path tempDir; + + @BeforeClass + public static void startServer() throws Exception { + tempDir = Files.createTempDirectory("validate-server-test"); + try (ServerSocket socket = new ServerSocket(0)) { + port = socket.getLocalPort(); + } + + serverThread = + new Thread( + () -> + ValidateServer.main( + new String[] { + "--port", String.valueOf(port), + "--threads", "2", + // Warm-up only affects speed, and would make the test slow. + "--warmup-rounds", "0" + })); + serverThread.setDaemon(true); + serverThread.start(); + + long deadline = System.currentTimeMillis() + 30_000; + while (System.currentTimeMillis() < deadline) { + if (statusOf("/health") == 200) { + return; + } + Thread.sleep(100); + } + throw new IllegalStateException("server did not become healthy"); + } + + @AfterClass + public static void stopServer() { + serverThread.interrupt(); + } + + /** + * The whole point of returning CLI-shaped text: callers parse it, so the server and the CLI must + * not drift. + */ + @Test + public void responseIsIdenticalToWhatTheCliWouldPrint() throws Exception { + for (String content : new String[] {VALID_SINGLE, NON_IUPAC, FASTA_HEADER}) { + File file = write("case" + content.hashCode() + ".fastq", content); + + JsonNode response = validate(List.of(file)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + ByteArrayOutputStream err = new ByteArrayOutputStream(); + int exitCode; + try (PrintStream outStream = new PrintStream(out, true, StandardCharsets.UTF_8); + PrintStream errStream = new PrintStream(err, true, StandardCharsets.UTF_8)) { + exitCode = + ValidationRunner.validate( + List.of(file), FileFormat.FASTQ, /* full= */ false, outStream, errStream); + } + + assertEquals(exitCode, response.get("exitCode").asInt()); + assertEquals(out.toString(StandardCharsets.UTF_8), response.get("stdout").asText()); + } + } + + @Test + public void reportsMissingFilesTheWayTheCliDoes() throws Exception { + JsonNode response = validate(List.of(new File(tempDir.toFile(), "absent.fastq"))); + + assertEquals(2, response.get("exitCode").asInt()); + assertTrue(response.get("stderr").asText().startsWith("ERROR: file not found:")); + } + + @Test + public void rejectsARequestWithoutFiles() throws Exception { + assertEquals(400, postStatus("{\"format\":\"FASTQ\",\"files\":[]}")); + } + + @Test + public void rejectsAnUnknownFormat() throws Exception { + assertEquals(400, postStatus("{\"format\":\"VCF\",\"files\":[\"/tmp/x.fastq\"]}")); + } + + /** Concurrent requests must each get their own verdict rather than another request's. */ + @Test + public void servesConcurrentRequestsIndependently() throws Exception { + File valid = write("concurrent-valid.fastq", VALID_SINGLE); + File invalid = write("concurrent-invalid.fastq", NON_IUPAC); + + ExecutorService pool = Executors.newFixedThreadPool(8); + try { + List> work = new ArrayList<>(); + for (int i = 0; i < 24; i++) { + boolean expectValid = i % 2 == 0; + File file = expectValid ? valid : invalid; + work.add( + () -> new int[] {expectValid ? 0 : 1, validate(List.of(file)).get("exitCode").asInt()}); + } + + for (Future result : pool.invokeAll(work, 120, TimeUnit.SECONDS)) { + int[] expectedAndActual = result.get(); + assertEquals(expectedAndActual[0], expectedAndActual[1]); + } + } finally { + pool.shutdownNow(); + } + } + + /** A liveness probe must not queue behind in-flight validations. */ + @Test + public void healthStaysAvailableWhileEveryValidationSlotIsBusy() throws Exception { + File file = write("saturating.fastq", VALID_SINGLE); + + ExecutorService pool = Executors.newFixedThreadPool(8); + try { + for (int i = 0; i < 8; i++) { + pool.submit( + () -> { + for (int round = 0; round < 5; round++) { + validate(List.of(file)); + } + return null; + }); + } + + for (int i = 0; i < 5; i++) { + assertEquals(200, statusOf("/health")); + Thread.sleep(50); + } + } finally { + pool.shutdownNow(); + pool.awaitTermination(60, TimeUnit.SECONDS); + } + } + + private static File write(String name, String content) throws IOException { + Path path = tempDir.resolve(name); + Files.writeString(path, content, StandardCharsets.UTF_8); + return path.toFile(); + } + + private static JsonNode validate(List files) throws IOException { + StringBuilder body = new StringBuilder("{\"format\":\"FASTQ\",\"files\":["); + for (int i = 0; i < files.size(); i++) { + body.append(i == 0 ? "" : ",").append('"').append(files.get(i).getAbsolutePath()).append('"'); + } + body.append("]}"); + + HttpURLConnection connection = post(body.toString()); + try (var stream = connection.getInputStream()) { + return MAPPER.readTree(stream.readAllBytes()); + } + } + + private static int postStatus(String body) throws IOException { + HttpURLConnection connection = post(body); + int status = connection.getResponseCode(); + connection.disconnect(); + return status; + } + + private static HttpURLConnection post(String body) throws IOException { + HttpURLConnection connection = + (HttpURLConnection) new URL("http://127.0.0.1:" + port + "/validate").openConnection(); + connection.setRequestMethod("POST"); + connection.setDoOutput(true); + connection.setRequestProperty("Content-Type", "application/json"); + try (OutputStream out = connection.getOutputStream()) { + out.write(body.getBytes(StandardCharsets.UTF_8)); + } + return connection; + } + + private static int statusOf(String path) { + try { + HttpURLConnection connection = + (HttpURLConnection) new URL("http://127.0.0.1:" + port + path).openConnection(); + connection.setConnectTimeout(2000); + connection.setReadTimeout(5000); + int status = connection.getResponseCode(); + connection.disconnect(); + return status; + } catch (IOException e) { + return -1; + } + } +} From dd355800cb4fbfd9fa10c5cbf7a18cf1f0f1064f Mon Sep 17 00:00:00 2001 From: corneliusroemer-agent <299456996+corneliusroemer-agent@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:35:47 +0000 Subject: [PATCH 2/3] Reject oversized requests as such, and make the server stoppable Reading a capped number of bytes and parsing whatever arrived turned a too-large request into a truncated-JSON parse error, so the caller was told its well-formed request was malformed. Check for trailing input and answer 413 instead. The test fixture interrupted the thread that ran main, but run() has already returned by then and it is the HttpServer's own threads that hold the port, so nothing was actually stopped. Give the server a stop() and have the fixture call it. Co-Authored-By: Claude Opus 5 (1M context) --- .../ena/readtools/v2/cli/ValidateServer.java | 59 +++++++++++++------ .../readtools/v2/cli/ValidateServerTest.java | 45 +++++++------- 2 files changed, 63 insertions(+), 41 deletions(-) diff --git a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java index 9cf00ea..6322c8b 100644 --- a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java +++ b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java @@ -97,10 +97,27 @@ public class ValidateServer { private final ObjectMapper mapper = new ObjectMapper(); private Semaphore validationSlots; + private HttpServer httpServer; + private ExecutorService executor; private final AtomicBoolean ready = new AtomicBoolean(false); private Path healthProbeFile; public static void main(String[] args) { + ValidateServer server = configure(args); + if (server == null) { + return; + } + + try { + server.run(); + } catch (IOException e) { + System.err.println("ERROR: could not start server: " + e.getMessage()); + System.exit(2); + } + } + + /** Parses arguments into a configured server, or null if usage was all that was asked for. */ + static ValidateServer configure(String... args) { ValidateServer server = new ValidateServer(); JCommander jc = JCommander.newBuilder().addObject(server).programName("readtools-validate server").build(); @@ -115,21 +132,16 @@ public static void main(String[] args) { if (server.help) { jc.usage(); - return; - } - - try { - server.run(); - } catch (IOException e) { - System.err.println("ERROR: could not start server: " + e.getMessage()); - System.exit(2); + return null; } + return server; } - private void run() throws IOException { + /** Starts listening and returns once the server is warmed up and serving. */ + void run() throws IOException { validationSlots = new Semaphore(threads); - HttpServer httpServer = HttpServer.create(new InetSocketAddress(host, port), 0); + httpServer = HttpServer.create(new InetSocketAddress(host, port), 0); // Without an explicit executor HttpServer runs every handler on the dispatch thread, i.e. // serially, which would give up the concurrency this server exists to provide. // @@ -139,19 +151,13 @@ private void run() throws IOException { // pod also evicts deacon's multi-gigabyte in-memory index. Threads are cheap; what needs // bounding is the number of validations allocating read buffers at once, and that is exactly // what the semaphore bounds. - ExecutorService executor = Executors.newCachedThreadPool(); + executor = Executors.newCachedThreadPool(); httpServer.setExecutor(executor); httpServer.createContext("/validate", this::handleValidate); httpServer.createContext("/health", this::handleHealth); - Runtime.getRuntime() - .addShutdownHook( - new Thread( - () -> { - httpServer.stop(0); - executor.shutdownNow(); - })); + Runtime.getRuntime().addShutdownHook(new Thread(this::stop)); httpServer.start(); System.err.printf( @@ -163,6 +169,16 @@ private void run() throws IOException { System.err.println("readtools validation server ready"); } + /** Stops listening and releases the handler threads. */ + void stop() { + if (httpServer != null) { + httpServer.stop(0); + } + if (executor != null) { + executor.shutdownNow(); + } + } + private void handleValidate(HttpExchange exchange) throws IOException { try { if (!"POST".equals(exchange.getRequestMethod())) { @@ -173,8 +189,13 @@ private void handleValidate(HttpExchange exchange) throws IOException { JsonNode request; try (InputStream body = exchange.getRequestBody()) { byte[] raw = body.readNBytes(MAX_REQUEST_BYTES); + // Otherwise an oversized body arrives as truncated JSON and is reported as malformed. + if (body.read() != -1) { + respondError(exchange, 413, "request body larger than " + MAX_REQUEST_BYTES + " bytes"); + return; + } request = mapper.readTree(raw); - } catch (Exception e) { + } catch (IOException | RuntimeException e) { respondError(exchange, 400, "could not parse request body: " + e.getMessage()); return; } diff --git a/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java b/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java index dc08fd8..a7c986c 100644 --- a/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java +++ b/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java @@ -47,7 +47,7 @@ public class ValidateServerTest { private static final ObjectMapper MAPPER = new ObjectMapper(); private static int port; - private static Thread serverThread; + private static ValidateServer server; private static Path tempDir; @BeforeClass @@ -57,32 +57,33 @@ public static void startServer() throws Exception { port = socket.getLocalPort(); } - serverThread = - new Thread( - () -> - ValidateServer.main( - new String[] { - "--port", String.valueOf(port), - "--threads", "2", - // Warm-up only affects speed, and would make the test slow. - "--warmup-rounds", "0" - })); - serverThread.setDaemon(true); - serverThread.start(); - - long deadline = System.currentTimeMillis() + 30_000; - while (System.currentTimeMillis() < deadline) { - if (statusOf("/health") == 200) { - return; - } - Thread.sleep(100); + server = + ValidateServer.configure( + "--port", String.valueOf(port), + "--threads", "2", + // Warm-up only affects speed, and would make the test slow. + "--warmup-rounds", "0"); + server.run(); + + if (statusOf("/health") != 200) { + throw new IllegalStateException("server did not become healthy"); } - throw new IllegalStateException("server did not become healthy"); } @AfterClass public static void stopServer() { - serverThread.interrupt(); + server.stop(); + } + + /** The port must be free again afterwards, i.e. stop() really stops it. */ + @Test + public void stopReleasesThePort() throws Exception { + ValidateServer other = ValidateServer.configure("--port", "0", "--warmup-rounds", "0"); + other.run(); + other.stop(); + try (ServerSocket socket = new ServerSocket(0)) { + assertTrue(socket.getLocalPort() > 0); + } } /** From 982c7b6f86ebf3c6c0dd76e16d0715a75a73ec70 Mon Sep 17 00:00:00 2001 From: corneliusroemer-agent <299456996+corneliusroemer-agent@users.noreply.github.com> Date: Wed, 26 Aug 2026 23:18:04 +0000 Subject: [PATCH 3/3] Answer the caller when a validation runs out of heap A validation large enough to exhaust the heap throws OutOfMemoryError, which is an Error rather than a RuntimeException, so it escaped the handler's catch and left the HTTP exchange unanswered. The caller then waited out its entire timeout instead of being told the request had failed. Found by running four concurrent full-file validations against a 1g heap: three died and their clients hung for over ten minutes. Catching Throwable here is deliberate. The alternative is not "fail cleanly", it is "never reply", and a stuck submission is worse than a reported error. The semaphore permit was already released in a finally block, so the server keeps serving; it now returns exit code 2, which callers treat as "could not validate" rather than as a verdict on the file. Co-Authored-By: Claude Opus 5 (1M context) --- .../uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java index 6322c8b..9e2a746 100644 --- a/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java +++ b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java @@ -229,8 +229,10 @@ private void handleValidate(HttpExchange exchange) throws IOException { } finally { validationSlots.release(); } - } catch (RuntimeException e) { - // Never let one bad request kill the handler thread silently. + } catch (Throwable e) { + // Throwable, not RuntimeException: a validation large enough to exhaust the heap throws + // OutOfMemoryError, which is an Error. Letting it escape leaves the exchange unanswered, so + // the caller waits out its whole timeout instead of being told the request failed. ObjectNode response = mapper.createObjectNode(); response.put("exitCode", 2); response.put("stdout", ""); @@ -248,7 +250,7 @@ private ObjectNode validateToJson(List files, FileFormat format, boolean f PrintStream err = new PrintStream(errBytes, true, StandardCharsets.UTF_8)) { try { exitCode = ValidationRunner.validate(files, format, full, out, err); - } catch (RuntimeException e) { + } catch (RuntimeException | StackOverflowError | OutOfMemoryError e) { // ValidateCli would have died here and the caller would have seen a stack trace on // stderr with no RESULT line; reproduce that shape rather than inventing a verdict. exitCode = 2;