///usr/bin/env jbang "$0" "$@" ; exit $?

//DEPS org.zeroturnaround:zt-exec:1.12

//DEPS com.fasterxml.jackson.core:jackson-databind:2.17.1

//DEPS org.slf4j:slf4j-nop:1.7.36

//JAVA 17+

//SOURCES jbang-lib/Scenario.java

//SOURCES jbang-lib/Peers.java

//SOURCES jbang-lib/Proc.java

//SOURCES jbang-lib/Sampler.java

//SOURCES jbang-lib/Checks.java

import java.io.OutputStream;

import java.net.ServerSocket;

import java.nio.file.Files;

import java.nio.file.Path;

import java.util.ArrayList;

import java.util.Comparator;

import java.util.HashMap;

import java.util.LinkedHashMap;

import java.util.LinkedHashSet;

import java.util.List;

import java.util.Locale;

import java.util.Map;

import java.util.Set;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

import java.util.stream.Stream;

/**

* Runs one cross-SDK or load scenario from configs/<scenario>.json.

*

* <pre>

* cd integration-testing

* jbang RunScenario.java x-java-typescript-http

* jbang RunScenario.java x-java-typescript-http --peer typescript-sdk=v0.5.0

* jbang RunScenario.java load-300 --skip-sdk-install

* jbang RunScenario.java --list

* jbang RunScenario.java --prepare x-java-java-http x-java-rust-ws (peers and builds only)

* jbang RunScenario.java x-java-rust-ws --prepared (no install, fetch or build)

* </pre>

*

* Flow: install the SDK from this checkout (unless --skip-sdk-install), prepare the peer SDKs

* the scenario names, build every process, start the servers in order and wait for their ready

* line, run the clients in order under their timeouts, take the configured jcmd samples, then

* tear down every process tree (always, in a finally) and evaluate the assertions. Logs go to

* logs/&lt;scenario&gt;/. Exit 0 = pass (declared expected failures included), 1 = fail,

* 2 = the scenario could not be set up.

*

* <p>Parallel runs (run-all.sh --jobs) split this in two: {@code --prepare} clones, fetches and

* builds every peer and process of a list of scenarios once, serially; each scenario then runs

* with {@code --prepared}, which touches no shared state (no install, no git, no build). Each

* scenario has its own log directory, its own {@code ${TMP}} (also {@code TMPDIR} for every

* process), and takes {@code ${PORT}} from {@code IT_PORT_RANGE} (e.g. 20000-20099) when set, so

* concurrent scenarios given disjoint ranges never race for a port (unset: an OS-chosen ephemeral

* port). Every Maven call goes through scripts/mvnw.sh, so the SDK SNAPSHOT is installed into and

* resolved from {@code IT_M2_REPO} only (default: integration-testing/.cache/m2, or

* ~/.m2/repository when CI=true).

*/

public class RunScenario {

public static void main(String... args) throws Exception {

String scenario = null;

boolean skipInstall = false;

boolean prepared = false;

boolean prepare = false;

List<String> scenarios = new ArrayList<>();

String peersRef = null;

Map<String, String> peerRefs = new HashMap<>();

for (int i = 0; i < args.length; i++) {

String a = args[i];

switch (a) {

case "-h", "--help" -> {

usage();

return;

}

case "--list" -> {

list();

return;

}

case "--skip-sdk-install" -> skipInstall = true;

case "--prepared" -> prepared = true;

case "--prepare" -> prepare = true;

case "--peers-ref" -> peersRef = args[++i];

case "--peer" -> {

String[] kv = args[++i].split("=", 2);

if (kv.length != 2) {

throw new IllegalArgumentException("--peer expects <name>=<ref>, got " + args[i]);

}

peerRefs.put(kv[0], kv[1]);

}

default -> {

if (a.startsWith("--peers-ref=")) {

peersRef = a.substring("--peers-ref=".length());

}

else if (a.startsWith("--peer=")) {

String[] kv = a.substring("--peer=".length()).split("=", 2);

peerRefs.put(kv[0], kv[1]);

}

else if (a.startsWith("-")) {

throw new IllegalArgumentException("Unknown option " + a);

}

else {

scenario = a;

scenarios.add(a);

}

}

}

}

if (scenario == null) {

usage();

System.exit(2);

}

if (prepare) {

System.exit(prepareAll(root(), scenarios, peersRef, peerRefs));

}

if (scenarios.size() > 1) {

throw new IllegalArgumentException("One scenario at a time (several only with --prepare): " + scenarios);

}

System.exit(new RunScenario(root(), scenario).run(skipInstall || prepared, prepared, peersRef, peerRefs));

}

static void usage() {

System.out.println("""

Usage: jbang RunScenario.java <scenario> [options]

Options:

--peer <name>=<ref> test peer SDK <name> (see peers.json) at <ref> (branch, tag or SHA)

--peers-ref <ref> test every peer SDK at <ref> (a --peer entry wins)

--skip-sdk-install do not install the SDK first (run-all.sh installs once)

--prepare <s>... only prepare the peers and run the builds of the scenarios (no install)

--prepared the scenario was prepared: no install, no git, no build (parallel runs)

--list list scenarios

""");

}

static void list() throws Exception {

try (Stream<Path> s = Files.list(root().resolve("configs"))) {

s.map(p -> p.getFileName().toString()).filter(n -> n.endsWith(".json")).sorted()

.forEach(n -> System.out.println(" " + n.substring(0, n.length() - 5)));

}

}

static Path root() {

Path cwd = Path.of("").toAbsolutePath();

for (Path p : List.of(cwd, cwd.resolve("integration-testing"))) {

if (Files.exists(p.resolve("RunScenario.java")) && Files.isDirectory(p.resolve("configs"))) {

return p;

}

}

throw new IllegalStateException("Run from integration-testing/ or the repository root");

}

final Path root;

final Path repo;

final String scenario;

final Path logDir;

final Map<String, Proc> procs = new LinkedHashMap<>();

final List<CompletableFuture<Void>> pendingSamples = new ArrayList<>();

final Checks.Outcome outcome = new Checks.Outcome();

final List<String> notes = new ArrayList<>();

RunScenario(Path root, String scenario) {

this.root = root;

this.repo = root.getParent();

this.scenario = scenario;

this.logDir = root.resolve("logs").resolve(scenario);

}

int run(boolean skipInstall, boolean prepared, String peersRef, Map<String, String> peerRefs) throws Exception {

long t0 = System.nanoTime();

Scenario.Config cfg = Scenario.load(root.resolve("configs").resolve(scenario + ".json"));

resetLogDir();

banner("Scenario " + scenario + (cfg.description() == null ? "" : ": " + cfg.description()));

Map<String, String> vars = new LinkedHashMap<>();

List<String> peerSummary = new ArrayList<>();

try {

String acpVersion = sdkVersion();

if (!skipInstall) {

step("Installing the SDK from " + repo + " (" + acpVersion + ")");

exec(logDir.resolve("sdk-install.log"), repo, 900, mvnw(root), "-q", "-B", "-DskipTests", "install");

}

vars.put("ROOT", root.toString());

vars.put("REPO", repo.toString());

vars.put("MVNW", mvnw(root));

vars.put("ACP_VERSION", acpVersion);

vars.put("CACHE", root.resolve(".cache").toString());

vars.put("LOG_DIR", logDir.toString());

Path tmp = Files.createDirectories(logDir.resolve("tmp"));

vars.put("TMP", tmp.toString());

vars.put("PORT", Integer.toString(freePort()));

if (!cfg.peers().isEmpty()) {

step("Preparing peer SDKs");

Map<String, Peers.Peer> known = Peers.load(root.resolve("peers.json"));

for (String name : cfg.peers()) {

Peers.Peer peer = known.get(name);

if (peer == null) {

throw new IllegalArgumentException("Peer '" + name + "' is not in peers.json");

}

String ref = peerRefs.getOrDefault(name, peersRef != null ? peersRef : peer.ref());

Peers.Checkout c = Peers.prepare(name, peer, ref, root.resolve(".cache"), logDir, prepared);

vars.put("peer." + name, c.dir().toString());

vars.put("peerRef." + name, ref);

vars.put("peerKey." + name, c.dir().getFileName().toString());

peerSummary.add(name + "@" + ref + " (" + c.sha().substring(0, 12) + ")");

}

}

for (String name : peerRefs.keySet()) {

if (!cfg.peers().contains(name)) {

System.out.println(" note: --peer " + name + " ignored; this scenario does not use it");

}

}

if (!prepared) {

build(cfg, vars, logDir, new LinkedHashSet<>());

}

}

catch (Exception e) {

outcome.failures.add("setup: " + e.getMessage());

return finish(cfg, t0, peerSummary, 2);

}

Thread hook = new Thread(this::killAll, "teardown");

Runtime.getRuntime().addShutdownHook(hook);

try {

execute(cfg, vars);

}

catch (Exception e) {

outcome.failures.add("run: " + e);

}

finally {

killAll();

Runtime.getRuntime().removeShutdownHook(hook);

}

Map<String, List<String>> output = new LinkedHashMap<>();

procs.forEach((n, p) -> output.put(n, p.lines()));

Checks.evaluate(cfg.assertions(), output, outcome);

return finish(cfg, t0, peerSummary, outcome.failures.isEmpty() ? 0 : 1);

}

void build(Scenario.Config cfg, Map<String, String> vars, Path buildLogs, Set<String> done) throws Exception {

for (Scenario.ProcessSpec p : cfg.processes()) {

if (p.build() == null || p.build().isBlank()) {

continue;

}

Path dir = Path.of(Scenario.subst(p.dir(), vars));

String cmd = Scenario.subst(p.build(), vars);

String key = dir + "\0" + cmd;

// A build that failed fails every scenario that needs it, not only the first: the

// others would otherwise run whatever an earlier build left behind.

if (done.contains(key + "\0failed")) {

throw new IllegalStateException("build of " + p.name() + " (" + p.language() + ") failed earlier in this run");

}

if (done.add(key)) {

step("Building " + p.name() + " (" + p.language() + ")");

try {

exec(buildLogs.resolve("build-" + (buildLogs.equals(logDir) ? "" : scenario + "-") + p.name() + ".log"),

dir, 1200, "bash", "-c", cmd);

}

catch (Exception e) {

done.add(key + "\0failed");

throw e;

}

}

}

}

void execute(Scenario.Config cfg, Map<String, String> vars) throws Exception {

long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(cfg.timeoutSec());

for (Scenario.ProcessSpec p : cfg.processes()) {

if (procs.containsKey(p.name())) {

throw new IllegalArgumentException("Duplicate process name " + p.name());

}

procs.put(p.name(), new Proc(p.name(), logDir.resolve(p.name() + ".log")));

}

// Line and exit triggers for samples; exit triggers run from the client loop below.

Map<String, List<Runnable>> onExit = new HashMap<>();

for (Scenario.ProcessSpec p : cfg.processes()) {

Proc target = procs.get(p.name());

for (Scenario.SampleSpec s : p.samples()) {

switch (s.on()) {

case "ready" -> {

}

case "line" -> {

Proc watched = required(s.process());

AtomicBoolean fired = new AtomicBoolean();

watched.onLine(line -> {

if (line.contains(s.line()) && fired.compareAndSet(false, true)) {

synchronized (pendingSamples) {

pendingSamples.add(CompletableFuture.runAsync(() -> sample(target, s)));

}

}

});

}

case "exit" -> onExit.computeIfAbsent(required(s.process()).name, k -> new ArrayList<>())

.add(() -> sample(target, s));

default -> throw new IllegalArgumentException("Unknown sample trigger '" + s.on() + "'");

}

}

}

for (Scenario.ProcessSpec p : cfg.processes()) {

if (!p.isServer()) {

continue;

}

Proc proc = procs.get(p.name());

step("Starting server " + p.name() + " (" + p.language() + ")");

start(proc, p, vars);

if (p.ready() != null) {

if (!proc.awaitLine(p.ready().line(), p.ready().timeoutSec())) {

throw new IllegalStateException(p.name() + " did not print '" + p.ready().line() + "' within "

+ p.ready().timeoutSec() + " s" + exitNote(proc) + "; see " + proc.log);

}

System.out.println(" ready (pid " + proc.pid() + ")");

}

for (Scenario.SampleSpec s : p.samples()) {

if (s.on().equals("ready")) {

sample(proc, s);

}

}

}

for (Scenario.ProcessSpec p : cfg.processes()) {

if (p.isServer()) {

continue;

}

Proc proc = procs.get(p.name());

long remaining = TimeUnit.NANOSECONDS.toSeconds(deadline - System.nanoTime());

int timeout = (int) Math.max(1, Math.min(p.timeoutSec(), remaining));

step("Running client " + p.name() + " (" + p.language() + ", timeout " + timeout + " s)");

long c0 = System.nanoTime();

start(proc, p, vars);

Integer code = proc.awaitExit(timeout);

if (code == null) {

outcome.failures.add(p.name() + " timed out after " + timeout + " s");

proc.kill();

}

else {

System.out.printf(Locale.ROOT, " exited %d after %.1f s%n", code, (System.nanoTime() - c0) / 1e9);

if (code != p.exitCode()) {

outcome.failures.add(p.name() + " exited " + code + ", expected " + p.exitCode());

}

else {

outcome.passes.add(p.name() + " exited " + code);

}

}

for (Runnable r : onExit.getOrDefault(p.name(), List.of())) {

r.run();

}

}

synchronized (pendingSamples) {

for (CompletableFuture<Void> f : pendingSamples) {

try {

f.get(120, TimeUnit.SECONDS);

}

catch (Exception e) {

outcome.failures.add("sample did not complete: " + e);

}

}

}

for (Scenario.ProcessSpec p : cfg.processes()) {

Proc proc = procs.get(p.name());

if (p.isServer() && !proc.isAlive()) {

outcome.failures.add("server " + p.name() + " exited early (code " + proc.exitValueIfExited() + ")");

}

}

}

void start(Proc proc, Scenario.ProcessSpec p, Map<String, String> vars) throws Exception {

Map<String, String> env = new LinkedHashMap<>();

env.put("TMPDIR", vars.get("TMP"));

p.env().forEach((k, v) -> env.put(k, Scenario.subst(v, vars)));

proc.start(Scenario.subst(p.run(), vars), Path.of(Scenario.subst(p.dir(), vars)), env);

}

void sample(Proc target, Scenario.SampleSpec s) {

try {

if (s.delayMs() > 0) {

Thread.sleep(s.delayMs());

}

if (!target.isAlive()) {

outcome.failures.add("cannot sample " + target.name + " at " + s.label() + ": it exited");

return;

}

String line = Sampler.sample(s.label(), target.pid());

target.append(line);

System.out.println(" " + line);

}

catch (Exception e) {

outcome.failures.add("sample " + s.label() + " of " + target.name + " failed: " + e);

}

}

Proc required(String name) {

Proc p = procs.get(name);

if (p == null) {

throw new IllegalArgumentException("Sample trigger names unknown process '" + name + "'");

}

return p;

}

void killAll() {

List<Proc> reversed = new ArrayList<>(procs.values());

java.util.Collections.reverse(reversed);

for (Proc p : reversed) {

p.kill();

p.closeLog();

}

}

int finish(Scenario.Config cfg, long t0, List<String> peerSummary, int code) throws Exception {

double secs = (System.nanoTime() - t0) / 1e9;

banner("Results: " + scenario);

if (!peerSummary.isEmpty()) {

System.out.println("Peers: " + String.join(", ", peerSummary));

}

for (Proc p : procs.values()) {

for (String l : p.lines()) {

if (l.startsWith("RESULT ") || l.startsWith("SAMPLE ")) {

System.out.println(" " + p.name + ": " + l);

}

}

}

collectNotes(cfg);

outcome.passes.forEach(s -> System.out.println(" ok " + s));

outcome.expectedFailures.forEach(s -> System.out.println(" XFAIL " + s));

outcome.failures.forEach(s -> System.out.println(" FAIL " + s));

String status = code == 0 ? (outcome.expectedFailures.isEmpty() ? "PASS" : "PASS (xfail)") : "FAIL";

System.out.printf(Locale.ROOT, "%nSCENARIO %s %s in %.1f s; logs: %s%n", scenario, status, secs, logDir);

List<String> result = new ArrayList<>();

result.add("scenario=" + scenario);

result.add("status=" + (code == 0 ? "PASS" : "FAIL"));

result.add("xfail=" + outcome.expectedFailures.size());

result.add(String.format(Locale.ROOT, "seconds=%.1f", secs));

result.add("peers=" + String.join(", ", peerSummary));

result.add("notes=" + String.join("; ", notes));

result.add("failures=" + outcome.failures.size());

outcome.failures.forEach(f -> result.add("failure=" + f.replace('\n', ' ')));

Files.write(logDir.resolve("result.txt"), result);

return code;

}

/** Numbers worth seeing in the run-all table (recorded, not asserted). */

void collectNotes(Scenario.Config cfg) {

cfg.assertions().report().forEach((proc, substrings) -> {

Proc p = procs.get(proc);

for (String sub : substrings) {

long n = p == null ? 0 : p.lines().stream().filter(l -> l.contains(sub)).count();

notes.add(proc + " \"" + sub + "\" x" + n);

System.out.println(" report " + proc + " \"" + sub + "\": " + n + " line(s)");

}

});

for (Proc p : procs.values()) {

Map<String, String> r = Checks.resultValues(p.lines());

if (r.containsKey("p50_ms")) {

notes.add("p50/p99 " + r.get("p50_ms") + "/" + r.get("p99_ms") + " ms");

}

Map<String, String> peak = Checks.sampleValues(p.lines(), "peak");

Map<String, String> after = Checks.sampleValues(p.lines(), "after_close");

if (!peak.isEmpty()) {

notes.add("heap peak " + peak.get("heap_mb") + " MB, after close " + after.getOrDefault("heap_mb", "?")

+ " MB; acp threads " + peak.get("acp_threads") + ", threads " + peak.get("threads"));

}

for (String l : p.lines()) {

Matcher m = HTTP_VERSION.matcher(l);

if (m.find()) {

notes.add("client negotiated " + m.group(1));

break;

}

}

}

if (!outcome.expectedFailures.isEmpty()) {

notes.add(outcome.expectedFailures.size() + " expected failure(s)");

}

}

static final Pattern HTTP_VERSION = Pattern.compile("negotiated (HTTP_\\S+)");

void resetLogDir() throws Exception {

if (Files.exists(logDir)) {

try (Stream<Path> s = Files.walk(logDir)) {

s.sorted(Comparator.reverseOrder()).forEach(p -> p.toFile().delete());

}

}

Files.createDirectories(logDir);

}

String sdkVersion() throws Exception {

String pom = Files.readString(repo.resolve("pom.xml"));

Matcher m = Pattern.compile("<artifactId>acp-java-sdk</artifactId>\\s*<version>([^<]+)</version>").matcher(pom);

if (!m.find()) {

throw new IllegalStateException("Cannot read the SDK version from " + repo.resolve("pom.xml"));

}

return m.group(1).trim();

}

/**

* The Maven wrapper every build uses: scripts/mvnw.sh, the checkout's ./mvnw pinned to the

* suite's local repository ({@code -Dmaven.repo.local=$IT_M2_REPO}, see scripts/m2-repo.sh).

*/

static String mvnw(Path root) {

return root.resolve("scripts").resolve("mvnw.sh").toString();

}

/**

* A free port: the first one that binds in {@code IT_PORT_RANGE} ({@code <from>-<to>}, set by

* run-all.sh so concurrent scenarios use disjoint ranges), else one the OS picks.

*/

static int freePort() throws Exception {

String range = System.getenv("IT_PORT_RANGE");

if (range != null && !range.isBlank()) {

String[] r = range.trim().split("-");

for (int port = Integer.parseInt(r[0]); port <= Integer.parseInt(r[1]); port++) {

try (ServerSocket s = new ServerSocket(port)) {

return port;

}

catch (java.io.IOException inUse) {

// try the next one

}

}

throw new IllegalStateException("No free port in IT_PORT_RANGE " + range);

}

try (ServerSocket s = new ServerSocket(0)) {

s.setReuseAddress(true);

return s.getLocalPort();

}

}

/**

* Prepare several scenarios for parallel runs: every peer checkout (fetched and built once per

* peer) and every process build (once per distinct directory and command), serially. Logs go

* to logs/run-all/prepare/. Returns 0, or 2 if anything failed.

*/

static int prepareAll(Path root, List<String> scenarios, String peersRef, Map<String, String> peerRefs)

throws Exception {

Path logs = Files.createDirectories(root.resolve("logs").resolve("run-all").resolve("prepare"));

Map<String, Peers.Peer> known = Peers.load(root.resolve("peers.json"));

Map<String, Peers.Checkout> checkouts = new LinkedHashMap<>();

Set<String> builds = new LinkedHashSet<>();

int code = 0;

for (String name : scenarios) {

RunScenario r = new RunScenario(root, name);

try {

Scenario.Config cfg = Scenario.load(root.resolve("configs").resolve(name + ".json"));

Map<String, String> vars = new LinkedHashMap<>();

vars.put("ROOT", root.toString());

vars.put("REPO", r.repo.toString());

vars.put("MVNW", mvnw(root));

vars.put("ACP_VERSION", r.sdkVersion());

vars.put("CACHE", root.resolve(".cache").toString());

vars.put("LOG_DIR", r.logDir.toString());

vars.put("TMP", r.logDir.resolve("tmp").toString());

vars.put("PORT", "0");

for (String peerName : cfg.peers()) {

Peers.Peer peer = known.get(peerName);

if (peer == null) {

throw new IllegalArgumentException("Peer '" + peerName + "' is not in peers.json");

}

String ref = peerRefs.getOrDefault(peerName, peersRef != null ? peersRef : peer.ref());

Peers.Checkout c = checkouts.get(peerName + "@" + ref);

if (c == null) {

step("Preparing peer " + peerName + " @ " + ref);

c = Peers.prepare(peerName, peer, ref, root.resolve(".cache"), logs, false);

checkouts.put(peerName + "@" + ref, c);

}

vars.put("peer." + peerName, c.dir().toString());

vars.put("peerRef." + peerName, ref);

vars.put("peerKey." + peerName, c.dir().getFileName().toString());

}

r.build(cfg, vars, logs, builds);

}

catch (Exception e) {

System.out.println(" FAIL preparing " + name + ": " + e.getMessage());

code = 2;

}

}

System.out.println("Prepared " + scenarios.size() + " scenario(s): " + checkouts.size() + " peer checkout(s), "

+ builds.size() + " build(s); logs: " + logs);

return code;

}

void exec(Path log, Path dir, int timeoutSec, String... cmd) throws Exception {

int code;

try (OutputStream out = Files.newOutputStream(log)) {

out.write(("$ (cd " + dir + " && " + String.join(" ", cmd) + ")\n").getBytes());

code = Proc.runToCompletion(out, dir, timeoutSec, cmd);

}

if (code != 0) {

throw new IllegalStateException("'" + String.join(" ", cmd) + "' exited " + code + "; see " + log);

}

}

static String exitNote(Proc p) {

Integer code = p.exitValueIfExited();

return code == null ? "" : " (it exited with " + code + ")";

}

static void banner(String s) {

System.out.println();

System.out.println("=".repeat(72));

System.out.println(s);

System.out.println("=".repeat(72));

}

static void step(String s) {

System.out.println("-> " + s);

}

}