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..9e2a746 --- /dev/null +++ b/src/main/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServer.java @@ -0,0 +1,387 @@ +/* + * 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 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(); + + try { + jc.parse(args); + } catch (Exception e) { + System.err.println("ERROR: " + e.getMessage()); + jc.usage(); + System.exit(2); + } + + if (server.help) { + jc.usage(); + return null; + } + return server; + } + + /** Starts listening and returns once the server is warmed up and serving. */ + void run() throws IOException { + validationSlots = new Semaphore(threads); + + 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. + executor = Executors.newCachedThreadPool(); + httpServer.setExecutor(executor); + + httpServer.createContext("/validate", this::handleValidate); + httpServer.createContext("/health", this::handleHealth); + + Runtime.getRuntime().addShutdownHook(new Thread(this::stop)); + + 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"); + } + + /** 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())) { + respondError(exchange, 405, "only POST is supported"); + return; + } + + 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 (IOException | RuntimeException 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 (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", ""); + 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 | 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; + 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..a7c986c --- /dev/null +++ b/src/test/java/uk/ac/ebi/ena/readtools/v2/cli/ValidateServerTest.java @@ -0,0 +1,236 @@ +/* + * 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 ValidateServer server; + 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(); + } + + 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"); + } + } + + @AfterClass + public static void stopServer() { + 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); + } + } + + /** + * 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; + } + } +}