@@ -230,6 +238,40 @@
false
+
+ 24
+
+
+
+
+ com.github.siom79.japicmp
+ japicmp-maven-plugin
+ 0.23.1
+
+
+
+ ai.spice
+ spiceai
+ ${japicmp.oldVersion}
+
+
+
+ true
+ true
+ true
+
+
+ ai.spice.example
+
+
+
+ org\.apache\.arrow\.adbc\..*
+
+
diff --git a/src/main/java/ai/spice/SpiceClient.java b/src/main/java/ai/spice/SpiceClient.java
index e9032af..f87f749 100644
--- a/src/main/java/ai/spice/SpiceClient.java
+++ b/src/main/java/ai/spice/SpiceClient.java
@@ -157,6 +157,12 @@ public class SpiceClient implements AutoCloseable {
private static final Duration HTTP_CONNECT_TIMEOUT = Duration.ofSeconds(15);
private static final Duration HTTP_REQUEST_TIMEOUT = Duration.ofSeconds(60);
+ // HTTP/2 keep-alive tuning for dead/unresponsive-peer detection.
+ // Package-visible so resilience tests derive their detection windows
+ // from the real values instead of restating them.
+ static final long KEEPALIVE_TIME_SECONDS = 30;
+ static final long KEEPALIVE_TIMEOUT_SECONDS = 10;
+
// Pre-computed parameter field names to avoid string concatenation in hot path
private static final String[] PARAM_NAMES = new String[64];
static {
@@ -497,8 +503,8 @@ private FlightChannel buildFlightChannel() {
}
channelBuilder
// HTTP/2 keep-alive to detect dead/idle connections behind load balancers
- .keepAliveTime(30, java.util.concurrent.TimeUnit.SECONDS)
- .keepAliveTimeout(10, java.util.concurrent.TimeUnit.SECONDS)
+ .keepAliveTime(KEEPALIVE_TIME_SECONDS, java.util.concurrent.TimeUnit.SECONDS)
+ .keepAliveTimeout(KEEPALIVE_TIMEOUT_SECONDS, java.util.concurrent.TimeUnit.SECONDS)
.keepAliveWithoutCalls(true)
.maxInboundMessageSize(MAX_INBOUND_MESSAGE_SIZE)
.maxInboundMetadataSize(MAX_INBOUND_METADATA_SIZE);
diff --git a/src/test/java/ai/spice/ChaosE2ETest.java b/src/test/java/ai/spice/ChaosE2ETest.java
new file mode 100644
index 0000000..ed8f3fa
--- /dev/null
+++ b/src/test/java/ai/spice/ChaosE2ETest.java
@@ -0,0 +1,292 @@
+/*
+Copyright 2026 The Spice.ai OSS Authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package ai.spice;
+
+import java.net.URI;
+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 java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.arrow.flight.FlightRuntimeException;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+import junit.framework.TestCase;
+
+/**
+ * End-to-end chaos tests against a real spiced process whose lifecycle the
+ * tests control: crash (SIGKILL), restart, and freeze (SIGSTOP). These prove
+ * the SDK's resilience behaviors — retry with backoff, reconnection after
+ * restart, prepared-statement re-prepare on stale handles, keep-alive
+ * detection of unresponsive peers — against the real runtime rather than a
+ * mock.
+ *
+ * Gated: runs only when SPICE_E2E_CHAOS=1 and a spiced binary is available
+ * (SPICED_BIN, or ~/.spice/bin/spiced). Uses a dataset-free spicepod, so
+ * startup is fast and there is no network dependency beyond localhost.
+ */
+public class ChaosE2ETest extends TestCase {
+
+ /** Wall-clock guard for calls that would hang forever if a feature is broken. */
+ private static final long CALL_GUARD_SECONDS = 120;
+
+ /**
+ * Daemon threads: if a guarded call ignores interruption (the very
+ * regression being hunted), the stuck worker must not keep the JVM alive.
+ */
+ private static final ExecutorService GUARD_EXECUTOR = Executors.newCachedThreadPool(runnable -> {
+ Thread thread = new Thread(runnable, "chaos-guard");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ private SpicedProcess spiced;
+
+ /**
+ * Skips silently when chaos testing isn't requested, but fails loudly when
+ * it IS requested and no spiced binary can be found — otherwise a broken
+ * CI install would turn the whole chaos suite into a passing no-op.
+ */
+ private static boolean chaosEnabled() {
+ if (!"1".equals(System.getenv("SPICE_E2E_CHAOS"))) {
+ return false;
+ }
+ assertNotNull("SPICE_E2E_CHAOS=1 but no spiced binary found (set SPICED_BIN or install the Spice CLI)",
+ SpicedProcess.findBinary());
+ return true;
+ }
+
+ @Override
+ protected void tearDown() throws Exception {
+ if (spiced != null) {
+ spiced.destroy();
+ spiced = null;
+ }
+ super.tearDown();
+ }
+
+ private SpiceClient newClient(int maxRetries) throws Exception {
+ return SpiceClient.builder()
+ .withFlightAddress(new URI("grpc://127.0.0.1:" + spiced.flightPort))
+ .withHttpAddress(new URI("http://127.0.0.1:" + spiced.httpPort))
+ .withMaxRetries(maxRetries)
+ .build();
+ }
+
+ private static long countRows(SpiceClient client, String sql) throws Exception {
+ try (FlightStream stream = client.query(sql)) {
+ return LocalFlightServerTest.countRows(stream);
+ }
+ }
+
+ /** Runs the callable with a hang guard so a broken SDK cannot wedge the suite. */
+ private static T guarded(Callable callable) throws Exception {
+ Future future = GUARD_EXECUTOR.submit(callable);
+ try {
+ return future.get(CALL_GUARD_SECONDS, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ throw new AssertionError(
+ "Call did not complete within " + CALL_GUARD_SECONDS + "s — likely hung");
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof Exception) {
+ throw (Exception) e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * The SDK survives a runtime crash and restart on the same address with no
+ * manual intervention: plain queries reconnect (fresh DNS/TCP), and cached
+ * prepared statements transparently re-prepare after their server-side
+ * handles died with the old process.
+ */
+ public void testSurvivesRuntimeRestart() throws Exception {
+ if (!chaosEnabled()) {
+ return;
+ }
+ spiced = SpicedProcess.start(SpicedProcess.DATASET_FREE_SPICEPOD);
+
+ try (SpiceClient client = newClient(3)) {
+ assertEquals(1, countRows(client, "SELECT 1"));
+ // Prime the prepared-statement cache so the restart invalidates a live handle.
+ try (ArrowReader reader = client.queryWithParams("SELECT $1", 42L)) {
+ assertTrue(reader.loadNextBatch());
+ }
+
+ spiced.kill();
+
+ // With the runtime down, queries must fail cleanly (no hang, no NPE).
+ try {
+ guarded(() -> countRows(client, "SELECT 1"));
+ fail("Expected query failure while the runtime is down");
+ } catch (ExecutionException e) {
+ assertTrue("cause should be a Flight transport error, got: " + e.getCause(),
+ e.getCause() instanceof FlightRuntimeException);
+ }
+
+ spiced = spiced.restart();
+
+ // Same client, no reset(): reconnect + re-prepare must be automatic.
+ assertEquals(1, (long) guarded(() -> countRows(client, "SELECT 1")));
+ try (ArrowReader reader = guarded(() -> client.queryWithParams("SELECT $1", 43L))) {
+ assertTrue("cached statement must transparently re-prepare after restart",
+ reader.loadNextBatch());
+ }
+ }
+ }
+
+ /**
+ * A query issued while the runtime is down succeeds without any caller-side
+ * handling, as long as the runtime returns within the retry backoff budget
+ * (~7.75s for 5 retries) — the load-balancer-failover scenario.
+ */
+ public void testQueryDuringDowntimeRecoversViaRetries() throws Exception {
+ if (!chaosEnabled()) {
+ return;
+ }
+ spiced = SpicedProcess.start(SpicedProcess.DATASET_FREE_SPICEPOD);
+
+ try (SpiceClient client = newClient(5)) {
+ assertEquals(1, countRows(client, "SELECT 1"));
+
+ spiced.kill();
+ // Bring the runtime back concurrently, inside the retry window.
+ // The restarter is always joined (finally) so a failing assertion
+ // can't leave it racing teardown, and its failure is surfaced.
+ final AtomicReference restartFailure = new AtomicReference<>();
+ Thread restarter = new Thread(() -> {
+ try {
+ Thread.sleep(1_500);
+ spiced = spiced.restart();
+ } catch (Exception e) {
+ restartFailure.set(e);
+ }
+ }, "chaos-restarter");
+ restarter.start();
+ try {
+ long start = System.nanoTime();
+ assertEquals("query issued during downtime should succeed via retries",
+ 1, (long) guarded(() -> countRows(client, "SELECT 1")));
+ long elapsedSeconds = (System.nanoTime() - start) / 1_000_000_000L;
+ assertTrue("recovery should happen within the retry budget, took " + elapsedSeconds + "s",
+ elapsedSeconds < 30);
+ } finally {
+ restarter.join(TimeUnit.SECONDS.toMillis(90));
+ assertFalse("restarter thread must finish before teardown", restarter.isAlive());
+ }
+ if (restartFailure.get() != null) {
+ throw restartFailure.get();
+ }
+ }
+ }
+
+ /**
+ * Killing the runtime while a large result is streaming surfaces a clean
+ * transport error mid-consumption (never a hang), and the same client
+ * recovers once the runtime returns.
+ */
+ public void testKillMidStreamFailsCleanlyAndRecovers() throws Exception {
+ if (!chaosEnabled()) {
+ return;
+ }
+ spiced = SpicedProcess.start(SpicedProcess.DATASET_FREE_SPICEPOD);
+
+ // ~10^7 rows from pure SQL92 cross joins — no version-specific functions.
+ String bigSql = "WITH t AS (SELECT * FROM (VALUES (1),(2),(3),(4),(5),(6),(7),(8),(9),(10)) v(x)) "
+ + "SELECT a.x FROM t a, t b, t c, t d, t e, t f, t g";
+
+ try (SpiceClient client = newClient(0)) {
+ assertEquals(1, countRows(client, "SELECT 1"));
+
+ try {
+ guarded(() -> {
+ try (FlightStream stream = client.query(bigSql)) {
+ assertTrue("stream should produce at least one batch", stream.next());
+ spiced.kill();
+ return LocalFlightServerTest.countRows(stream);
+ }
+ });
+ fail("Expected a transport error when the runtime dies mid-stream");
+ } catch (FlightRuntimeException expected) {
+ // Clean, classifiable failure — exactly what callers should see.
+ }
+
+ spiced = spiced.restart();
+ assertEquals("client must recover after the runtime returns",
+ 1, (long) guarded(() -> countRows(client, "SELECT 1")));
+ }
+ }
+
+ /**
+ * A frozen (SIGSTOP) runtime — the unresponsive-peer case TCP alone never
+ * detects — is caught by HTTP/2 keep-alive: the in-flight call fails with a
+ * transport error instead of hanging forever. Detection is measured on a
+ * retry-free client so the window is one keep-alive cycle; recovery after
+ * the thaw is then verified with a fresh client that retries while the
+ * transport re-establishes.
+ */
+ public void testFrozenRuntimeDetectedByKeepAlive() throws Exception {
+ if (!chaosEnabled()) {
+ return;
+ }
+ spiced = SpicedProcess.start(SpicedProcess.DATASET_FREE_SPICEPOD);
+
+ // Derived from the SDK's actual keep-alive tuning plus generous
+ // scheduler slack; not instant (that would be a refusal, not detection).
+ long detectionUpperBound = (SpiceClient.KEEPALIVE_TIME_SECONDS
+ + SpiceClient.KEEPALIVE_TIMEOUT_SECONDS) * 2 + 20;
+
+ try (SpiceClient client = newClient(0)) {
+ assertEquals(1, countRows(client, "SELECT 1"));
+
+ spiced.freeze();
+ try {
+ long start = System.nanoTime();
+ try {
+ guarded(() -> countRows(client, "SELECT 1"));
+ fail("Expected keep-alive to fail the call against a frozen runtime");
+ } catch (ExecutionException e) {
+ long elapsedSeconds = (System.nanoTime() - start) / 1_000_000_000L;
+ assertTrue("cause should be a Flight transport error, got: " + e.getCause(),
+ e.getCause() instanceof FlightRuntimeException);
+ assertTrue("keep-alive detection took " + elapsedSeconds + "s (bound: "
+ + detectionUpperBound + "s)",
+ elapsedSeconds >= 2 && elapsedSeconds <= detectionUpperBound);
+ }
+ } finally {
+ spiced.thaw();
+ }
+ }
+
+ try (SpiceClient recovered = newClient(3)) {
+ assertEquals(1, (long) guarded(() -> countRows(recovered, "SELECT 1")));
+ }
+ }
+}
diff --git a/src/test/java/ai/spice/MtlsTest.java b/src/test/java/ai/spice/MtlsTest.java
new file mode 100644
index 0000000..a3fc7e6
--- /dev/null
+++ b/src/test/java/ai/spice/MtlsTest.java
@@ -0,0 +1,145 @@
+/*
+Copyright 2026 The Spice.ai OSS Authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package ai.spice;
+
+import java.net.URI;
+import java.util.concurrent.ExecutionException;
+
+import javax.net.ssl.SSLException;
+
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+import junit.framework.TestCase;
+
+/**
+ * TLS and mutual-TLS integration tests: a real TLS handshake against an
+ * in-process Flight SQL server using runtime-generated certificates —
+ * covering custom-CA trust, client-certificate authentication, and the
+ * corresponding rejection paths. This exercises the SDK's file-based TLS
+ * configuration end-to-end (previously only builder validation was tested).
+ */
+public class MtlsTest extends TestCase {
+
+ private static TestCerts certs;
+
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ if (certs == null) {
+ certs = TestCerts.generate();
+ }
+ }
+
+ private static SpiceClientBuilder clientFor(TestFlightSqlServer server) throws Exception {
+ // 127.0.0.1 on both sides (matching the server bind and the cert SAN):
+ // dialing "localhost" may try ::1 first and fail with connection-refused
+ // before any TLS handshake happens.
+ return SpiceClient.builder()
+ .withFlightAddress(new URI("grpc+tls://127.0.0.1:" + server.getPort()))
+ .withMaxRetries(0);
+ }
+
+ private static void assertQueriesWork(SpiceClient client, TestFlightSqlServer server) throws Exception {
+ try (FlightStream stream = client.query("SELECT * FROM test")) {
+ assertEquals(server.expectedTotalRows(), LocalFlightServerTest.countRows(stream));
+ }
+ // Parameterized queries run on the same TLS channel (prepared
+ // statements inherit the transport configuration).
+ try (ArrowReader reader = client.queryWithParams("SELECT * FROM test WHERE id > $1", 1L)) {
+ assertEquals(server.expectedTotalRows(), LocalFlightServerTest.countRows(reader));
+ }
+ }
+
+ /** Server TLS with a custom CA: the client trusts it via withTlsRootCertFile. */
+ public void testTlsWithCustomRootCa() throws Exception {
+ try (TestFlightSqlServer server = new TestFlightSqlServer(certs, false);
+ SpiceClient client = clientFor(server)
+ .withTlsRootCertFile(certs.caCert.toString())
+ .build()) {
+ assertQueriesWork(client, server);
+ }
+ }
+
+ /** Full mutual TLS: server verifies the client certificate. */
+ public void testMutualTls() throws Exception {
+ try (TestFlightSqlServer server = new TestFlightSqlServer(certs, true);
+ SpiceClient client = clientFor(server)
+ .withTlsRootCertFile(certs.caCert.toString())
+ .withTlsClientCertFile(certs.clientCert.toString())
+ .withTlsClientKeyFile(certs.clientKey.toString())
+ .build()) {
+ assertQueriesWork(client, server);
+ }
+ }
+
+ /** An mTLS server rejects clients that present no certificate. */
+ public void testMissingClientCertificateRejected() throws Exception {
+ try (TestFlightSqlServer server = new TestFlightSqlServer(certs, true);
+ SpiceClient client = clientFor(server)
+ .withTlsRootCertFile(certs.caCert.toString())
+ .build()) {
+ try {
+ client.query("SELECT 1");
+ fail("Expected the TLS handshake to be rejected without a client certificate");
+ } catch (ExecutionException e) {
+ assertTlsFailure(e);
+ }
+ }
+ }
+
+ /** A client that trusts a different CA must reject the server. */
+ public void testUntrustedServerCaRejected() throws Exception {
+ try (TestFlightSqlServer server = new TestFlightSqlServer(certs, false);
+ SpiceClient client = clientFor(server)
+ .withTlsRootCertFile(certs.otherCaCert.toString())
+ .build()) {
+ try {
+ client.query("SELECT 1");
+ fail("Expected certificate verification to fail against an untrusted CA");
+ } catch (ExecutionException e) {
+ assertTlsFailure(e);
+ }
+ }
+ }
+
+ /**
+ * The negative tests must fail *because of TLS* — an unrelated transport
+ * or query error passing for a certificate rejection would mask a broken
+ * verification path. Walks the cause chain for TLS evidence.
+ */
+ private static void assertTlsFailure(Throwable failure) {
+ StringBuilder chain = new StringBuilder();
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause instanceof SSLException) {
+ return;
+ }
+ chain.append(cause.getClass().getName()).append(": ").append(cause.getMessage()).append(" <- ");
+ String message = cause.getMessage() == null ? "" : cause.getMessage().toLowerCase();
+ if (message.contains("ssl") || message.contains("certificate") || message.contains("handshake")) {
+ return;
+ }
+ }
+ fail("Expected a TLS/certificate failure, but the cause chain shows none: " + chain);
+ }
+}
diff --git a/src/test/java/ai/spice/PerfBenchmarkTest.java b/src/test/java/ai/spice/PerfBenchmarkTest.java
index 24d6ab3..d69876e 100644
--- a/src/test/java/ai/spice/PerfBenchmarkTest.java
+++ b/src/test/java/ai/spice/PerfBenchmarkTest.java
@@ -22,12 +22,18 @@ of this software and associated documentation files (the "Software"), to deal
package ai.spice;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.Arrays;
import java.util.concurrent.Callable;
import org.apache.arrow.flight.FlightStream;
import org.apache.arrow.vector.ipc.ArrowReader;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+
import junit.framework.TestCase;
/**
@@ -78,20 +84,74 @@ private static String stats(String label, long[] sortedNanos) {
p50 / 1_000, p95 / 1_000, p99 / 1_000);
}
+ private static long p50Micros(long[] sortedNanos) {
+ return sortedNanos[sortedNanos.length / 2] / 1_000;
+ }
+
+ /**
+ * Appends a data point to the file named by the BENCH_JSON env var, in
+ * github-action-benchmark's "customSmallerIsBetter" format. No-op when the
+ * variable is unset (normal test runs).
+ */
+ private static synchronized void recordBench(String name, String unit, long value) throws Exception {
+ String path = System.getenv("BENCH_JSON");
+ if (path == null || path.isEmpty()) {
+ return;
+ }
+ Path file = Path.of(path);
+ JsonArray entries = Files.exists(file)
+ ? JsonParser.parseString(Files.readString(file)).getAsJsonArray()
+ : new JsonArray();
+ JsonObject entry = new JsonObject();
+ entry.addProperty("name", name);
+ entry.addProperty("unit", unit);
+ entry.addProperty("value", value);
+ entries.add(entry);
+ Files.writeString(file, entries.toString());
+ }
+
+ /**
+ * Counts rows and asserts the full expected result arrived — a regression
+ * that drops results must never publish an "improved" latency.
+ */
+ private static long checkedCountRows(FlightStream stream, long expectedRows) throws Exception {
+ long rows = LocalFlightServerTest.countRows(stream);
+ assertEquals(expectedRows, rows);
+ return rows;
+ }
+
+ private static long checkedCountRows(ArrowReader reader, long expectedRows) throws Exception {
+ long rows = LocalFlightServerTest.countRows(reader);
+ assertEquals(expectedRows, rows);
+ return rows;
+ }
+
+ private static Callable paramsOp(SpiceClient client, long expectedRows) {
+ return () -> {
+ try (ArrowReader reader = client.queryWithParams(SQL, 5L)) {
+ return checkedCountRows(reader, expectedRows);
+ }
+ };
+ }
+
public void testBenchmarkPlainQuery() throws Exception {
try (SpiceClient client = SpiceClient.builder().withFlightAddress(server.flightUri()).build()) {
+ final long expectedRows = server.expectedTotalRows();
Callable op = () -> {
try (FlightStream stream = client.query("SELECT * FROM bench")) {
- return LocalFlightServerTest.countRows(stream);
+ return checkedCountRows(stream, expectedRows);
}
};
measure(WARMUP_ITERATIONS, op);
long getFlightInfoBefore = server.getFlightInfoCalls.get();
+ long doGetBefore = server.doGetCalls.get();
long[] samples = measure(MEASURED_ITERATIONS, op);
System.out.println("[bench] " + stats("query()", samples));
+ recordBench("query() p50", "us", p50Micros(samples));
// The plain query path is exactly 2 RPCs: GetFlightInfo + DoGet.
assertEquals(MEASURED_ITERATIONS, server.getFlightInfoCalls.get() - getFlightInfoBefore);
+ assertEquals(MEASURED_ITERATIONS, server.doGetCalls.get() - doGetBefore);
}
}
@@ -103,16 +163,9 @@ public void testBenchmarkParameterizedQueryCachedVsUncached() throws Exception {
.withFlightAddress(server.flightUri())
.withPreparedStatementCacheSize(0)
.build()) {
- Callable cachedOp = () -> {
- try (ArrowReader reader = cachedClient.queryWithParams(SQL, 5L)) {
- return LocalFlightServerTest.countRows(reader);
- }
- };
- Callable uncachedOp = () -> {
- try (ArrowReader reader = uncachedClient.queryWithParams(SQL, 5L)) {
- return LocalFlightServerTest.countRows(reader);
- }
- };
+ final long expectedRows = server.expectedTotalRows();
+ Callable cachedOp = paramsOp(cachedClient, expectedRows);
+ Callable uncachedOp = paramsOp(uncachedClient, expectedRows);
// Warm both paths before measuring either, so JIT compilation of the
// shared code doesn't bias whichever block runs second.
@@ -133,6 +186,8 @@ public void testBenchmarkParameterizedQueryCachedVsUncached() throws Exception {
System.out.println("[bench] " + stats("queryWithParams (cached)", cachedSamples));
System.out.println("[bench] " + stats("queryWithParams (uncached)", uncachedSamples));
+ recordBench("queryWithParams cached p50", "us", p50Micros(cachedSamples));
+ recordBench("queryWithParams uncached p50", "us", p50Micros(uncachedSamples));
System.out.printf(
"[bench] RPCs per %d queries: cached=%d prepares, uncached=%d prepares + %d closes "
+ "(2 round trips saved per query on a real network)%n",
@@ -165,6 +220,7 @@ public void testParameterBindAllocationStaysSmall() throws Exception {
}
System.out.printf("[bench] param-root buffer bytes for 100 binds of (long,string,double): %d%n",
totalCapacityBytes);
+ recordBench("param-root bytes per 100 binds", "bytes", totalCapacityBytes);
// Old behavior: >48KB per string vector alone → many MB over 100 binds.
assertTrue("parameter roots should stay small, used " + totalCapacityBytes + " bytes",
totalCapacityBytes < 200_000);
diff --git a/src/test/java/ai/spice/SoakTest.java b/src/test/java/ai/spice/SoakTest.java
new file mode 100644
index 0000000..d73d606
--- /dev/null
+++ b/src/test/java/ai/spice/SoakTest.java
@@ -0,0 +1,226 @@
+/*
+Copyright 2026 The Spice.ai OSS Authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package ai.spice;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+import junit.framework.TestCase;
+
+/**
+ * Long-running soak against a live Spice runtime: a sustained mixed workload
+ * (queries, cached parameterized queries, health probes, periodic resets)
+ * asserting zero errors, non-empty results throughout, no Arrow memory leaks
+ * (a leak makes close() throw), bounded thread growth, and stable tail
+ * latency over the whole run.
+ *
+ * Gated: runs only when SPICE_SOAK_SECONDS is set to a positive number
+ * (the nightly workflow uses 1800). Connects to the runtime configured via
+ * the standard SPICE_FLIGHT_URL / SPICE_HTTP_URL environment (localhost
+ * defaults), and queries SPICE_SOAK_DATASET (default taxi_trips).
+ */
+public class SoakTest extends TestCase {
+
+ private static final int WORKERS = 4;
+ private static final long RESET_INTERVAL_SECONDS = 120;
+ /**
+ * Latency samples retained per minute. At ~9k ops/s a 30-minute soak would
+ * otherwise retain ~16M boxed samples and risk exhausting the heap before
+ * the assertions run; a bounded prefix per minute is ample for p99
+ * stability comparison. (Concurrent workers may overshoot the cap by at
+ * most WORKERS-1 samples — irrelevant at this size.)
+ */
+ private static final int MAX_SAMPLES_PER_MINUTE = 5_000;
+
+ public void testSoak() throws Exception {
+ long soakSeconds = Long.parseLong(System.getenv().getOrDefault("SPICE_SOAK_SECONDS", "0"));
+ if (soakSeconds <= 0) {
+ return;
+ }
+ String dataset = System.getenv().getOrDefault("SPICE_SOAK_DATASET", "taxi_trips");
+ String querySql = "SELECT * FROM " + dataset + " LIMIT 50";
+ String paramSql = "SELECT * FROM " + dataset + " WHERE $1 = 1 LIMIT 10";
+
+ final AtomicLong operations = new AtomicLong();
+ final AtomicLong dataOperations = new AtomicLong();
+ final AtomicLong rowsRead = new AtomicLong();
+ final ConcurrentLinkedQueue errors = new ConcurrentLinkedQueue<>();
+ // Bounded latency samples in micros, bucketed by minute of the run.
+ final Map> latenciesByMinute = new ConcurrentHashMap<>();
+
+ final long startNanos = System.nanoTime();
+ final long deadlineNanos = startNanos + TimeUnit.SECONDS.toNanos(soakSeconds);
+ final int warmThreadBaseline;
+
+ try (SpiceClient client = SpiceClient.builder().withMaxRetries(3).build()) {
+ // Fail fast (before the long run) if the runtime isn't serving.
+ assertTrue("runtime must be ready before soaking", client.isReady());
+ try (FlightStream warm = client.query(querySql)) {
+ LocalFlightServerTest.countRows(warm);
+ }
+ // Baseline AFTER warm-up: gRPC/Netty event loops and the JDK HTTP
+ // client's selector are shared steady-state pools that appear on
+ // first use. The leak signal is growth beyond this warm baseline
+ // over the run (e.g. resets leaving channels behind), not the
+ // difference from a cold JVM.
+ warmThreadBaseline = liveThreadCount();
+
+ ExecutorService executor = Executors.newFixedThreadPool(WORKERS);
+ for (int w = 0; w < WORKERS; w++) {
+ executor.submit(() -> {
+ int roll = 0;
+ while (System.nanoTime() < deadlineNanos) {
+ long opStart = System.nanoTime();
+ try {
+ int kind = roll++ % 20;
+ if (kind < 16) {
+ try (FlightStream stream = client.query(querySql)) {
+ rowsRead.addAndGet(LocalFlightServerTest.countRows(stream));
+ }
+ dataOperations.incrementAndGet();
+ } else if (kind < 19) {
+ try (ArrowReader reader = client.queryWithParams(paramSql, 1L)) {
+ rowsRead.addAndGet(LocalFlightServerTest.countRows(reader));
+ }
+ dataOperations.incrementAndGet();
+ } else {
+ if (!client.isHealthy() || !client.isReady()) {
+ errors.add("health probe reported unhealthy mid-soak");
+ }
+ }
+ operations.incrementAndGet();
+ long minute = TimeUnit.NANOSECONDS.toMinutes(opStart - startNanos);
+ List bucket = latenciesByMinute.computeIfAbsent(minute,
+ m -> Collections.synchronizedList(new ArrayList<>()));
+ if (bucket.size() < MAX_SAMPLES_PER_MINUTE) {
+ bucket.add((System.nanoTime() - opStart) / 1_000);
+ }
+ } catch (Throwable t) {
+ errors.add(t.getClass().getSimpleName() + ": " + t.getMessage());
+ if (errors.size() > 20) {
+ return; // Failing hard; no point continuing.
+ }
+ }
+ }
+ });
+ }
+ executor.shutdown();
+
+ // Periodic reset() proves transport rebuild under sustained load;
+ // awaitTermination doubles as the completion join.
+ while (!executor.awaitTermination(RESET_INTERVAL_SECONDS, TimeUnit.SECONDS)) {
+ if (System.nanoTime() < deadlineNanos) {
+ try {
+ client.reset();
+ } catch (Throwable t) {
+ errors.add("reset: " + t.getMessage());
+ }
+ } else {
+ assertTrue("workers must finish shortly after the deadline",
+ executor.awaitTermination(120, TimeUnit.SECONDS));
+ break;
+ }
+ }
+
+ printSummary(operations.get(), soakSeconds, latenciesByMinute);
+
+ assertTrue("soak must complete with zero errors, got " + errors.size()
+ + " — first: " + errors.peek(), errors.isEmpty());
+ // Guard against a silent-empty-results regression: every data
+ // operation queries with a LIMIT >= 10 against a populated dataset.
+ System.out.printf("[soak] data-ops=%d rows-read=%d%n", dataOperations.get(), rowsRead.get());
+ assertTrue("every data operation must return rows: ops=" + dataOperations.get()
+ + " rows=" + rowsRead.get(), rowsRead.get() >= dataOperations.get() * 10);
+ assertTailLatencyStable(latenciesByMinute);
+ }
+ // Reaching here means close() did not throw: no Arrow buffers leaked
+ // over the whole run (a leak makes the allocator close throw).
+
+ // Transport threads shut down with the client; allow scavenger slack.
+ Thread.sleep(3_000);
+ int threadsAfter = liveThreadCount();
+ System.out.printf("[soak] threads: warm-baseline=%d after-close=%d%n",
+ warmThreadBaseline, threadsAfter);
+ assertTrue("thread count should not grow over the soak: warm-baseline=" + warmThreadBaseline
+ + " after-close=" + threadsAfter, threadsAfter <= warmThreadBaseline + 8);
+ }
+
+ private static int liveThreadCount() {
+ return Thread.getAllStackTraces().size();
+ }
+
+ private static long percentile(List sortedMicros, double quantile) {
+ return sortedMicros.get((int) Math.min(sortedMicros.size() - 1,
+ (long) (sortedMicros.size() * quantile)));
+ }
+
+ /** p99 of the last third of the run must stay within 3x of the first third. */
+ private static void assertTailLatencyStable(Map> byMinute) {
+ List minutes = new ArrayList<>(byMinute.keySet());
+ if (minutes.size() < 3) {
+ return; // Too short a run to compare thirds.
+ }
+ minutes.sort(Long::compare);
+ int third = minutes.size() / 3;
+ long firstThirdP99 = p99Of(byMinute, minutes.subList(0, third));
+ long lastThirdP99 = p99Of(byMinute, minutes.subList(minutes.size() - third, minutes.size()));
+ System.out.printf("[soak] p99 first-third=%dus last-third=%dus%n", firstThirdP99, lastThirdP99);
+ assertTrue("tail latency degraded over the soak: first-third p99=" + firstThirdP99
+ + "us, last-third p99=" + lastThirdP99 + "us",
+ lastThirdP99 <= firstThirdP99 * 3);
+ }
+
+ private static long p99Of(Map> byMinute, List minutes) {
+ List samples = new ArrayList<>();
+ for (long minute : minutes) {
+ samples.addAll(byMinute.get(minute));
+ }
+ samples.sort(Long::compare);
+ return samples.isEmpty() ? 0 : percentile(samples, 0.99);
+ }
+
+ private static void printSummary(long operations, long soakSeconds, Map> byMinute) {
+ List all = new ArrayList<>();
+ byMinute.values().forEach(all::addAll);
+ all.sort(Long::compare);
+ long p50 = all.isEmpty() ? 0 : percentile(all, 0.50);
+ long p99 = all.isEmpty() ? 0 : percentile(all, 0.99);
+ System.out.printf("[soak] %d ops over %ds (%.1f ops/s), p50=%dus p99=%dus, %d minutes sampled%n",
+ operations, soakSeconds, operations / (double) soakSeconds, p50, p99, byMinute.size());
+ System.out.println("[soak] per-minute p99 (us): "
+ + byMinute.keySet().stream().sorted()
+ .map(minute -> String.valueOf(p99Of(byMinute, List.of(minute))))
+ .reduce((a, b) -> a + ", " + b).orElse("none"));
+ }
+}
diff --git a/src/test/java/ai/spice/SpicedProcess.java b/src/test/java/ai/spice/SpicedProcess.java
new file mode 100644
index 0000000..fbcb1fc
--- /dev/null
+++ b/src/test/java/ai/spice/SpicedProcess.java
@@ -0,0 +1,203 @@
+/*
+Copyright 2026 The Spice.ai OSS Authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package ai.spice;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Manages a real spiced process for lifecycle-level tests: launch on fixed
+ * localhost ports with a caller-provided spicepod, wait for health, and
+ * crash (SIGKILL), freeze (SIGSTOP)/thaw (SIGCONT), restart on the same
+ * address (mirroring a crashed server replaced behind a stable VIP), or
+ * destroy. The process fixture counterpart to {@link TestFlightSqlServer}.
+ */
+final class SpicedProcess {
+
+ /** A spicepod with no datasets: fast startup, no external dependencies. */
+ static final String DATASET_FREE_SPICEPOD = "version: v1\nkind: Spicepod\nname: chaos\n";
+
+ private static final HttpClient HEALTH_CLIENT = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofSeconds(1))
+ .build();
+
+ final Path workspace;
+ final int httpPort;
+ final int flightPort;
+ final int metricsPort;
+ private Process process;
+
+ private SpicedProcess(Path workspace, int httpPort, int flightPort, int metricsPort) {
+ this.workspace = workspace;
+ this.httpPort = httpPort;
+ this.flightPort = flightPort;
+ this.metricsPort = metricsPort;
+ }
+
+ /** The spiced binary from SPICED_BIN or ~/.spice/bin, or null if absent. */
+ static String findBinary() {
+ String env = System.getenv("SPICED_BIN");
+ if (env != null && Files.isExecutable(Path.of(env))) {
+ return env;
+ }
+ Path home = Path.of(System.getProperty("user.home"), ".spice", "bin", "spiced");
+ return Files.isExecutable(home) ? home.toString() : null;
+ }
+
+ static SpicedProcess start(String spicepodYaml) throws Exception {
+ // Ephemeral ports are released before spiced binds them, so a rare
+ // port steal is possible; retry with fresh ports.
+ Exception lastFailure = null;
+ for (int attempt = 1; attempt <= 3; attempt++) {
+ Path workspace = Files.createTempDirectory("spice-chaos");
+ Files.writeString(workspace.resolve("spicepod.yaml"), spicepodYaml, StandardCharsets.UTF_8);
+ SpicedProcess spiced = new SpicedProcess(workspace, freePort(), freePort(), freePort());
+ try {
+ spiced.launch();
+ return spiced;
+ } catch (Exception e) {
+ lastFailure = e;
+ }
+ }
+ throw lastFailure;
+ }
+
+ /**
+ * Starts a fresh process on the same ports and workspace — clients must
+ * be able to reconnect to the same address. No port retry is possible here.
+ */
+ SpicedProcess restart() throws Exception {
+ SpicedProcess fresh = new SpicedProcess(workspace, httpPort, flightPort, metricsPort);
+ fresh.launch();
+ return fresh;
+ }
+
+ private void launch() throws Exception {
+ ProcessBuilder builder = new ProcessBuilder(
+ findBinary(),
+ "--http", "127.0.0.1:" + httpPort,
+ "--flight", "127.0.0.1:" + flightPort,
+ "--metrics", "127.0.0.1:" + metricsPort);
+ builder.directory(workspace.toFile());
+ builder.redirectErrorStream(true);
+ builder.redirectOutput(workspace.resolve("spiced.log").toFile());
+ process = builder.start();
+ try {
+ waitUntilHealthy();
+ } catch (Exception startupFailure) {
+ // Never orphan a half-started process the caller has no handle to.
+ try {
+ destroy();
+ } catch (Exception cleanup) {
+ startupFailure.addSuppressed(cleanup);
+ }
+ throw startupFailure;
+ }
+ }
+
+ private void waitUntilHealthy() throws Exception {
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(new URI("http://127.0.0.1:" + httpPort + "/health"))
+ .timeout(Duration.ofSeconds(2))
+ .GET()
+ .build();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
+ while (System.nanoTime() < deadline) {
+ if (!process.isAlive()) {
+ throw new IllegalStateException("spiced exited during startup; see "
+ + workspace.resolve("spiced.log"));
+ }
+ try {
+ HttpResponse response = HEALTH_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
+ if (response.statusCode() == 200) {
+ return;
+ }
+ } catch (IOException retry) {
+ // Not up yet.
+ } catch (InterruptedException interrupted) {
+ // Preserve the interrupt and stop waiting: cancellation must
+ // not be silently converted into more polling.
+ Thread.currentThread().interrupt();
+ throw interrupted;
+ }
+ Thread.sleep(250);
+ }
+ throw new IllegalStateException("spiced did not become healthy within 60s");
+ }
+
+ /** SIGKILL — an abrupt crash, no graceful shutdown. */
+ void kill() throws Exception {
+ process.destroyForcibly();
+ if (!process.waitFor(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException(
+ "spiced pid " + process.pid() + " did not exit within 10s of SIGKILL");
+ }
+ }
+
+ /** SIGSTOP — the process is alive but completely unresponsive. */
+ void freeze() throws Exception {
+ signal("STOP");
+ }
+
+ /** SIGCONT — resume a frozen process. */
+ void thaw() throws Exception {
+ signal("CONT");
+ }
+
+ private void signal(String name) throws Exception {
+ Process kill = new ProcessBuilder("kill", "-" + name, Long.toString(process.pid())).start();
+ if (!kill.waitFor(5, TimeUnit.SECONDS) || kill.exitValue() != 0) {
+ throw new IllegalStateException("kill -" + name + " failed for pid " + process.pid());
+ }
+ }
+
+ void destroy() throws Exception {
+ if (process != null && process.isAlive()) {
+ // A frozen process ignores SIGKILL delivery ordering with SIGSTOP
+ // pending on some platforms; thaw first, best-effort.
+ try {
+ signal("CONT");
+ } catch (Exception ignored) {
+ // Already dead or never frozen.
+ }
+ process.destroyForcibly();
+ process.waitFor(10, TimeUnit.SECONDS);
+ }
+ }
+
+ private static int freePort() throws IOException {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ socket.setReuseAddress(true);
+ return socket.getLocalPort();
+ }
+ }
+}
diff --git a/src/test/java/ai/spice/TestCerts.java b/src/test/java/ai/spice/TestCerts.java
new file mode 100644
index 0000000..64ec3f9
--- /dev/null
+++ b/src/test/java/ai/spice/TestCerts.java
@@ -0,0 +1,173 @@
+/*
+Copyright 2026 The Spice.ai OSS Authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package ai.spice;
+
+import java.math.BigInteger;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+import java.security.spec.ECGenParameterSpec;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.asn1.x509.BasicConstraints;
+import org.bouncycastle.asn1.x509.ExtendedKeyUsage;
+import org.bouncycastle.asn1.x509.Extension;
+import org.bouncycastle.asn1.x509.GeneralName;
+import org.bouncycastle.asn1.x509.GeneralNames;
+import org.bouncycastle.asn1.x509.KeyPurposeId;
+import org.bouncycastle.asn1.x509.KeyUsage;
+import org.bouncycastle.cert.X509v3CertificateBuilder;
+import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
+import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder;
+import org.bouncycastle.openssl.jcajce.JcaPEMWriter;
+import org.bouncycastle.openssl.jcajce.JcaPKCS8Generator;
+import org.bouncycastle.operator.ContentSigner;
+import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+
+/**
+ * Self-signed certificate fixtures for TLS/mTLS tests, generated at runtime
+ * with BouncyCastle (already a compile dependency of the SDK): a CA, a server
+ * certificate for localhost/127.0.0.1, a client certificate signed by the same
+ * CA, and a second, unrelated CA for negative tests. All materialized as PEM
+ * files, matching the SDK's file-based TLS configuration.
+ */
+final class TestCerts {
+
+ private static final AtomicLong SERIAL = new AtomicLong(System.currentTimeMillis());
+
+ final Path caCert;
+ final Path serverCert;
+ final Path serverKey;
+ final Path clientCert;
+ final Path clientKey;
+ /** A CA unrelated to any issued certificate, for trust-failure tests. */
+ final Path otherCaCert;
+
+ private TestCerts(Path caCert, Path serverCert, Path serverKey,
+ Path clientCert, Path clientKey, Path otherCaCert) {
+ this.caCert = caCert;
+ this.serverCert = serverCert;
+ this.serverKey = serverKey;
+ this.clientCert = clientCert;
+ this.clientKey = clientKey;
+ this.otherCaCert = otherCaCert;
+ }
+
+ static TestCerts generate() throws Exception {
+ Path dir = Files.createTempDirectory("spice-test-certs");
+
+ KeyPair caKeys = newKeyPair();
+ X509Certificate ca = newCaCert("CN=Spice Test CA", caKeys);
+
+ KeyPair serverKeys = newKeyPair();
+ X509Certificate server = newLeafCert("CN=localhost", serverKeys, ca, caKeys.getPrivate(),
+ KeyPurposeId.id_kp_serverAuth, true);
+
+ KeyPair clientKeys = newKeyPair();
+ X509Certificate client = newLeafCert("CN=spice-test-client", clientKeys, ca, caKeys.getPrivate(),
+ KeyPurposeId.id_kp_clientAuth, false);
+
+ KeyPair otherCaKeys = newKeyPair();
+ X509Certificate otherCa = newCaCert("CN=Unrelated Test CA", otherCaKeys);
+
+ return new TestCerts(
+ writePem(dir.resolve("ca.pem"), ca),
+ writePem(dir.resolve("server.pem"), server),
+ writeKeyPem(dir.resolve("server-key.pem"), serverKeys.getPrivate()),
+ writePem(dir.resolve("client.pem"), client),
+ writeKeyPem(dir.resolve("client-key.pem"), clientKeys.getPrivate()),
+ writePem(dir.resolve("other-ca.pem"), otherCa));
+ }
+
+ private static KeyPair newKeyPair() throws Exception {
+ // EC P-256: ~300x faster to generate than RSA-2048 (which shows
+ // multi-second outliers on shared CI runners), equally supported by
+ // Netty/gRPC and the SDK's PEM parsing.
+ KeyPairGenerator generator = KeyPairGenerator.getInstance("EC");
+ generator.initialize(new ECGenParameterSpec("secp256r1"));
+ return generator.generateKeyPair();
+ }
+
+ private static X509Certificate newCaCert(String subject, KeyPair keys) throws Exception {
+ X500Name name = new X500Name(subject);
+ X509v3CertificateBuilder builder = new JcaX509v3CertificateBuilder(
+ name, BigInteger.valueOf(SERIAL.incrementAndGet()),
+ notBefore(), notAfter(), name, keys.getPublic());
+ builder.addExtension(Extension.basicConstraints, true, new BasicConstraints(true));
+ builder.addExtension(Extension.keyUsage, true,
+ new KeyUsage(KeyUsage.keyCertSign | KeyUsage.cRLSign));
+ return sign(builder, keys.getPrivate());
+ }
+
+ private static X509Certificate newLeafCert(String subject, KeyPair keys,
+ X509Certificate issuer, PrivateKey issuerKey, KeyPurposeId purpose,
+ boolean withLocalhostSan) throws Exception {
+ X509v3CertificateBuilder builder = new JcaX509v3CertificateBuilder(
+ issuer, BigInteger.valueOf(SERIAL.incrementAndGet()),
+ notBefore(), notAfter(), new X500Name(subject), keys.getPublic());
+ builder.addExtension(Extension.basicConstraints, true, new BasicConstraints(false));
+ builder.addExtension(Extension.keyUsage, true,
+ new KeyUsage(KeyUsage.digitalSignature | KeyUsage.keyEncipherment));
+ builder.addExtension(Extension.extendedKeyUsage, false, new ExtendedKeyUsage(purpose));
+ if (withLocalhostSan) {
+ builder.addExtension(Extension.subjectAlternativeName, false, new GeneralNames(
+ new GeneralName[] {
+ new GeneralName(GeneralName.dNSName, "localhost"),
+ new GeneralName(GeneralName.iPAddress, "127.0.0.1"),
+ }));
+ }
+ return sign(builder, issuerKey);
+ }
+
+ private static X509Certificate sign(X509v3CertificateBuilder builder, PrivateKey key) throws Exception {
+ ContentSigner signer = new JcaContentSignerBuilder("SHA256withECDSA").build(key);
+ return new JcaX509CertificateConverter().getCertificate(builder.build(signer));
+ }
+
+ private static Date notBefore() {
+ return new Date(System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1));
+ }
+
+ private static Date notAfter() {
+ return new Date(System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1));
+ }
+
+ private static Path writePem(Path path, Object pemObject) throws Exception {
+ try (JcaPEMWriter writer = new JcaPEMWriter(Files.newBufferedWriter(path))) {
+ writer.writeObject(pemObject);
+ }
+ return path;
+ }
+
+ private static Path writeKeyPem(Path path, PrivateKey key) throws Exception {
+ // PKCS#8 ("PRIVATE KEY"), accepted by both Netty (Flight path) and
+ // the SDK's BouncyCastle PEM parsing (HTTP path).
+ return writePem(path, new JcaPKCS8Generator(key, null));
+ }
+}
diff --git a/src/test/java/ai/spice/TestFlightSqlServer.java b/src/test/java/ai/spice/TestFlightSqlServer.java
index 0d47562..f9989c1 100644
--- a/src/test/java/ai/spice/TestFlightSqlServer.java
+++ b/src/test/java/ai/spice/TestFlightSqlServer.java
@@ -22,8 +22,13 @@ of this software and associated documentation files (the "Software"), to deal
package ai.spice;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
import java.net.URI;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -104,7 +109,7 @@ final class TestFlightSqlServer implements AutoCloseable {
/** Starts an unauthenticated server on an ephemeral port. */
TestFlightSqlServer() throws Exception {
- this(null, null);
+ this(null, null, null, false);
}
/**
@@ -113,9 +118,34 @@ final class TestFlightSqlServer implements AutoCloseable {
* bearer tokens for subsequent calls (mirroring Spice's auth flow).
*/
TestFlightSqlServer(String expectedUser, String expectedPassword) throws Exception {
+ this(expectedUser, expectedPassword, null, false);
+ }
+
+ /**
+ * Starts a TLS server using the fixture's server certificate. With
+ * requireClientCert, clients must present a certificate signed by the
+ * fixture CA (mutual TLS).
+ */
+ TestFlightSqlServer(TestCerts certs, boolean requireClientCert) throws Exception {
+ this(null, null, certs, requireClientCert);
+ }
+
+ private TestFlightSqlServer(String expectedUser, String expectedPassword,
+ TestCerts certs, boolean requireClientCert) throws Exception {
this.allocator = new RootAllocator(Long.MAX_VALUE);
- FlightServer.Builder builder = FlightServer.builder(
- allocator, Location.forGrpcInsecure("localhost", 0), new Producer());
+ // TLS tests bind and dial 127.0.0.1 explicitly: "localhost" can resolve
+ // to ::1 first, and a connection-refused there would let negative TLS
+ // tests pass without ever reaching the handshake under test.
+ Location location = certs != null
+ ? Location.forGrpcTls("127.0.0.1", 0)
+ : Location.forGrpcInsecure("localhost", 0);
+ FlightServer.Builder builder = FlightServer.builder(allocator, location, new Producer());
+ if (certs != null) {
+ builder.useTls(pemStream(certs.serverCert), pemStream(certs.serverKey));
+ if (requireClientCert) {
+ builder.useMTlsClientVerification(pemStream(certs.caCert));
+ }
+ }
if (expectedUser != null) {
CallHeaderAuthenticator inner = new GeneratedBearerTokenAuthenticator(
new BasicCallHeaderAuthenticator((username, password) -> {
@@ -163,6 +193,14 @@ void rejectNextBearerToken() {
rejectNextBearer.set(true);
}
+ /**
+ * PEM file as an in-memory stream: the FlightServer builder defers reading
+ * its TLS streams until build(), so file streams would be closed by then.
+ */
+ private static InputStream pemStream(Path pem) throws IOException {
+ return new ByteArrayInputStream(Files.readAllBytes(pem));
+ }
+
long expectedTotalRows() {
return (long) endpointCount * batchesPerEndpoint * rowsPerBatch;
}