From 31c38a02d8cbf6095ee030e513d9f6a0502f3a55 Mon Sep 17 00:00:00 2001 From: Viktor Yershov Date: Wed, 19 Aug 2026 11:23:38 -0700 Subject: [PATCH 1/4] feat: add listActiveQueries and cancelActiveQuery for running SQL queries The runtime exposes GET /v1/sql/active and POST /v1/sql/{id}/cancel for listing and cancelling currently running synchronous queries, but neither was reachable from this SDK (gospice already has this pair; dotnet does not). --- README.md | 20 ++ src/main/java/ai/spice/ActiveQuery.java | 105 ++++++ src/main/java/ai/spice/SpiceClient.java | 200 ++++++++++++ src/test/java/ai/spice/ActiveQueriesTest.java | 302 ++++++++++++++++++ 4 files changed, 627 insertions(+) create mode 100644 src/main/java/ai/spice/ActiveQuery.java create mode 100644 src/test/java/ai/spice/ActiveQueriesTest.java diff --git a/README.md b/README.md index fa9c8eb..023616d 100644 --- a/README.md +++ b/README.md @@ -415,6 +415,26 @@ for (ConnectionDetails connection : client.runtimeStatus()) { `REFRESHING`, `SHUTTING_DOWN`, or `NOT_LOADED`. A status a newer runtime introduces maps to `UNKNOWN` rather than failing; `getRawStatus()` returns it verbatim. +#### Active queries + +Use `listActiveQueries()` to see the synchronous queries currently running on the runtime, +and `cancelActiveQuery(queryId)` to cancel one. The runtime does not hand a query's ID back +to the client that submitted it, so listing is the only way to find the ID that cancelling +needs. + +```java +for (ActiveQuery query : client.listActiveQueries()) { + System.out.printf("%s %s %s%n", + query.getQueryId(), query.getProtocol(), query.getSqlPreview()); +} + +client.cancelActiveQuery(queryId); +``` + +Both calls are scoped to the authenticated API key or client certificate — not to this +`SpiceClient` instance — and reach only the one runtime process behind this client's HTTP +endpoint, which matters if that endpoint is a load balancer in front of several runtimes. + ### Logging The SDK uses SLF4J for logging, allowing you to plug in your preferred logging implementation (Logback, Log4j2, java.util.logging, etc.). diff --git a/src/main/java/ai/spice/ActiveQuery.java b/src/main/java/ai/spice/ActiveQuery.java new file mode 100644 index 0000000..b0400a6 --- /dev/null +++ b/src/main/java/ai/spice/ActiveQuery.java @@ -0,0 +1,105 @@ +/* +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.time.Instant; + +import com.google.gson.annotations.SerializedName; + +/** + * A synchronous query currently running on the runtime, as reported by + * {@code GET /v1/sql/active}. + * + *

+ * Synchronous queries are the ones started by {@link SpiceClient#query(String)}, + * {@link SpiceClient#queryWithParams(String, Object...)}, or issued directly over + * Flight SQL, HTTP, or NSQL/Search. The runtime does not return a query's ID to the + * client that submitted it, so {@link SpiceClient#listActiveQueries()} is the only + * way to discover the ID that {@link SpiceClient#cancelActiveQuery(String)} needs. + */ +public class ActiveQuery { + @SerializedName("query_id") + private String queryId; + + @SerializedName("protocol") + private String protocol; + + @SerializedName("sql_preview") + private String sqlPreview; + + @SerializedName("started_at_ms") + private long startedAtMs; + + /** + * The runtime-assigned query ID. Pass this to + * {@link SpiceClient#cancelActiveQuery(String)} to cancel the query. + * + * @return the query ID + */ + public String getQueryId() { + return this.queryId; + } + + /** + * The protocol the query arrived on: {@code http}, {@code flight}, + * {@code flightsql}, or {@code internal}. + * + * @return the protocol + */ + public String getProtocol() { + return this.protocol; + } + + /** + * The query's SQL, truncated by the runtime for display. + * + * @return the truncated SQL text + */ + public String getSqlPreview() { + return this.sqlPreview; + } + + /** + * When the query started, in milliseconds since the Unix epoch. + * + * @return the start time in epoch milliseconds + */ + public long getStartedAtMs() { + return this.startedAtMs; + } + + /** + * When the query started. + * + * @return the start time + */ + public Instant getStartedAt() { + return Instant.ofEpochMilli(this.startedAtMs); + } + + @Override + public String toString() { + return String.format("ActiveQuery{queryId=%s, protocol=%s, startedAt=%s, sqlPreview=%s}", + this.queryId, this.protocol, getStartedAt(), this.sqlPreview); + } +} diff --git a/src/main/java/ai/spice/SpiceClient.java b/src/main/java/ai/spice/SpiceClient.java index f87f749..b735f7d 100644 --- a/src/main/java/ai/spice/SpiceClient.java +++ b/src/main/java/ai/spice/SpiceClient.java @@ -1591,6 +1591,206 @@ public void refreshDataset(String dataset, RefreshOptions refreshOptions) throws } } + /** + * Lists the synchronous queries currently running on the runtime by calling + * {@code GET /v1/sql/active}. + * + *

+ * Synchronous queries are the ones started by {@link #query(String)}, + * {@link #queryWithParams(String, Object...)}, or issued directly over Flight + * SQL, HTTP, NSQL, or Search. The runtime does not return a query's ID to the + * client that submitted it, so this is the only way to discover the ID that + * {@link #cancelActiveQuery(String)} needs. + * + *

+ * Results are scoped to the authenticated principal — an API key or client + * certificate — not to this {@code SpiceClient}: every client presenting the + * same credential lists the same queries. Results also cover only the one + * runtime instance this client's HTTP endpoint reaches, which behind a load + * balancer may not be the instance that received a particular query. + * + * @return the currently running synchronous queries + * @throws ExecutionException if the runtime is unreachable or returns an + * unexpected response + */ + public List listActiveQueries() throws ExecutionException { + logger.debug("Listing active queries"); + try { + HttpRequest request = buildProbeRequest(this.httpAddress, "/v1/sql/active", this.apiKey); + + HttpResponse response = httpClient().send(request, HttpResponse.BodyHandlers.ofString()); + + if (response.statusCode() == 403) { + throw new ExecutionException( + "Failed to list active queries: the configured API key does not allow listing queries, " + + "use a key with write access", + null); + } + if (response.statusCode() != 200) { + logger.error("Listing active queries failed - statusCode={}, response={}", response.statusCode(), + response.body()); + throw new ExecutionException( + String.format("Failed to list active queries. Status Code: %d, Response: %s", + response.statusCode(), + response.body()), + null); + } + + return parseActiveQueries(response.body()); + } catch (ExecutionException e) { + // no need to wrap ExecutionException + throw e; + } catch (ConnectException err) { + logger.error("Cannot connect to Spice runtime at {}: {}", this.httpAddress, err.getMessage()); + throw new ExecutionException( + String.format("The Spice runtime is unavailable at %s. Is it running?", this.httpAddress), err); + } catch (InterruptedException err) { + Thread.currentThread().interrupt(); + throw new ExecutionException("Interrupted while listing active queries", err); + } catch (Exception err) { + logger.error("Listing active queries failed: {}", err.getMessage()); + throw new ExecutionException("Failed to list active queries due to error: " + err.toString(), err); + } + } + + /** + * Parses the {@code /v1/sql/active} response body. + * + * @param body the JSON object the runtime returned + * @return the parsed active queries, or an empty list when the body carries none + */ + private static List parseActiveQueries(String body) { + JsonElement root = JsonParser.parseString(body == null ? "" : body); + if (root == null || !root.isJsonObject()) { + return Collections.emptyList(); + } + JsonElement queries = root.getAsJsonObject().get("queries"); + if (queries == null || !queries.isJsonArray()) { + return Collections.emptyList(); + } + + List result = new ArrayList<>(); + for (JsonElement element : queries.getAsJsonArray()) { + if (element.isJsonObject()) { + result.add(GSON.fromJson(element, ActiveQuery.class)); + } + } + return result; + } + + /** + * Cancels a running synchronous query by ID, by calling + * {@code POST /v1/sql/{queryId}/cancel}. + * + *

+ * {@code queryId} comes from {@link #listActiveQueries()}. Cancellation is + * scoped to the authenticated principal, not to this {@code SpiceClient}: any + * client presenting the same credential can cancel the query, while an ID + * outside that scope is reported as not found. Like + * {@link #listActiveQueries()}, this reaches only the one runtime instance + * this client's HTTP endpoint resolves to. + * + * @param queryId the ID of the query to cancel, from {@link #listActiveQueries()} + * @throws ExecutionException if the runtime is unreachable or reports a failure + */ + public void cancelActiveQuery(String queryId) throws ExecutionException { + if (Strings.isNullOrEmpty(queryId)) { + throw new IllegalArgumentException("queryId is required, use listActiveQueries() to find one"); + } + // queryId is caller input that is spliced directly into the request path. + // Requiring it to look exactly like a UUID rules out path-traversal + // segments such as "." or ".." before the string is ever used, rather than + // relying on escaping to neutralize them. + if (!isUuid(queryId)) { + throw new IllegalArgumentException( + "query ID \"" + queryId + "\" is not a valid UUID, use the queryId from listActiveQueries()"); + } + + logger.debug("Cancelling active query: {}", queryId); + try { + HttpRequest.Builder builder = HttpRequest.newBuilder() + .uri(new URI(String.format("%s/v1/sql/%s/cancel", this.httpAddress, queryId))) + .timeout(HTTP_REQUEST_TIMEOUT) + .header("Accept", "application/json") + .header("X-Spice-User-Agent", Config.getUserAgent()) + .POST(HttpRequest.BodyPublishers.ofString("{}")); + if (!Strings.isNullOrEmpty(this.apiKey)) { + builder = builder.header("X-API-Key", this.apiKey); + } + + HttpResponse response = httpClient().send(builder.build(), HttpResponse.BodyHandlers.ofString()); + + switch (response.statusCode()) { + case 200: + logger.debug("Active query cancelled successfully: {}", queryId); + return; + case 400: + throw new ExecutionException( + "query ID \"" + queryId + "\" is not a valid UUID, use the queryId from " + + "listActiveQueries()", + null); + case 403: + throw new ExecutionException( + "the configured API key does not allow cancelling queries, use a key with write access", + null); + case 404: + throw new ExecutionException( + "no active query \"" + queryId + "\" found: it may have already finished, " + + "or it was submitted under a different API key", + null); + default: + logger.error("Cancelling active query failed - queryId={}, statusCode={}, response={}", queryId, + response.statusCode(), response.body()); + throw new ExecutionException( + String.format("Failed to cancel active query. Status Code: %d, Response: %s", + response.statusCode(), + response.body()), + null); + } + } catch (ExecutionException e) { + // no need to wrap ExecutionException + throw e; + } catch (ConnectException err) { + logger.error("Cannot connect to Spice runtime at {}: {}", this.httpAddress, err.getMessage()); + throw new ExecutionException( + String.format("The Spice runtime is unavailable at %s. Is it running?", this.httpAddress), err); + } catch (InterruptedException err) { + Thread.currentThread().interrupt(); + throw new ExecutionException("Interrupted while cancelling active query", err); + } catch (Exception err) { + logger.error("Cancelling active query failed: {}", err.getMessage()); + throw new ExecutionException("Failed to cancel active query due to error: " + err.toString(), err); + } + } + + /** + * Reports whether {@code value} has the canonical UUID shape: 36 characters, + * hexadecimal digits throughout except literal {@code -} at positions 8, 13, + * 18, and 23. + * + * @param value the string to check + * @return true if value is shaped like a UUID + */ + private static boolean isUuid(String value) { + if (value.length() != 36) { + return false; + } + for (int i = 0; i < value.length(); i++) { + char c = value.charAt(i); + if (i == 8 || i == 13 || i == 18 || i == 23) { + if (c != '-') { + return false; + } + continue; + } + boolean isHex = (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F'); + if (!isHex) { + return false; + } + } + return true; + } + private FlightStream queryInternal(String sql) { FlightChannel[] snapshot = currentChannels(); FlightChannel channel = selectChannel(snapshot); diff --git a/src/test/java/ai/spice/ActiveQueriesTest.java b/src/test/java/ai/spice/ActiveQueriesTest.java new file mode 100644 index 0000000..b959bdc --- /dev/null +++ b/src/test/java/ai/spice/ActiveQueriesTest.java @@ -0,0 +1,302 @@ +/* +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.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicReference; + +import com.sun.net.httpserver.HttpServer; + +import junit.framework.TestCase; + +/** + * Unit tests for {@link SpiceClient#listActiveQueries()} and + * {@link SpiceClient#cancelActiveQuery(String)} against a local JDK + * {@link HttpServer}, so they need no live Spice runtime. + */ +public class ActiveQueriesTest extends TestCase { + + private static final String VALID_QUERY_ID = "3fa85f64-5717-4562-b3fc-2c963f66afa6"; + + private HttpServer server; + private SpiceClient client; + private final AtomicReference lastMethod = new AtomicReference<>(); + private final AtomicReference lastPath = new AtomicReference<>(); + private final AtomicReference lastAcceptHeader = new AtomicReference<>(); + private final AtomicReference lastApiKeyHeader = new AtomicReference<>(); + + private void startServer() throws IOException { + this.server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + } + + /** Stubs a canned response for one path, recording the request first. */ + private void stub(String path, int statusCode, String body) { + this.server.createContext(path, exchange -> { + lastMethod.set(exchange.getRequestMethod()); + lastPath.set(exchange.getRequestURI().getPath()); + lastAcceptHeader.set(exchange.getRequestHeaders().getFirst("Accept")); + lastApiKeyHeader.set(exchange.getRequestHeaders().getFirst("X-API-Key")); + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(statusCode, bytes.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(bytes); + } + }); + } + + private SpiceClient clientForServer(String apiKey) throws Exception { + this.server.start(); + URI httpAddress = new URI("http://127.0.0.1:" + this.server.getAddress().getPort()); + SpiceClientBuilder builder = SpiceClient.builder().withHttpAddress(httpAddress); + if (apiKey != null) { + builder = builder.withApiKey(apiKey); + } + return builder.build(); + } + + @Override + protected void tearDown() throws Exception { + if (this.client != null) { + this.client.close(); + this.client = null; + } + if (this.server != null) { + this.server.stop(0); + this.server = null; + } + super.tearDown(); + } + + // ==================== listActiveQueries ==================== + + public void testListActiveQueriesParsesMultipleQueries() throws Exception { + startServer(); + stub("/v1/sql/active", 200, "{" + + "\"queries\":[" + + "{\"query_id\":\"" + VALID_QUERY_ID + "\",\"protocol\":\"flight\"," + + "\"sql_preview\":\"SELECT * FROM taxi_trips\",\"started_at_ms\":1700000000000}," + + "{\"query_id\":\"9d2c1b3a-1111-2222-3333-444455556666\",\"protocol\":\"http\"," + + "\"sql_preview\":\"SELECT 1\",\"started_at_ms\":1700000001000}" + + "],\"total_count\":2}"); + this.client = clientForServer(null); + + List queries = this.client.listActiveQueries(); + + assertEquals("GET", lastMethod.get()); + assertEquals("/v1/sql/active", lastPath.get()); + assertNull("no API key configured, so none should be sent", lastApiKeyHeader.get()); + + assertEquals(2, queries.size()); + assertEquals(VALID_QUERY_ID, queries.get(0).getQueryId()); + assertEquals("flight", queries.get(0).getProtocol()); + assertEquals("SELECT * FROM taxi_trips", queries.get(0).getSqlPreview()); + assertEquals(1700000000000L, queries.get(0).getStartedAtMs()); + assertEquals(Instant.ofEpochMilli(1700000000000L), queries.get(0).getStartedAt()); + + assertEquals("http", queries.get(1).getProtocol()); + } + + public void testListActiveQueriesReturnsEmptyListWhenNoneRunning() throws Exception { + startServer(); + stub("/v1/sql/active", 200, "{\"queries\":[],\"total_count\":0}"); + this.client = clientForServer(null); + + List queries = this.client.listActiveQueries(); + + assertTrue(queries.isEmpty()); + } + + public void testListActiveQueriesSendsApiKeyWhenConfigured() throws Exception { + startServer(); + stub("/v1/sql/active", 200, "{\"queries\":[]}"); + this.client = clientForServer("test-app|test-key"); + + this.client.listActiveQueries(); + + assertEquals("test-app|test-key", lastApiKeyHeader.get()); + } + + public void testListActiveQueriesForbiddenThrowsSpecificMessage() throws Exception { + startServer(); + stub("/v1/sql/active", 403, "Forbidden"); + this.client = clientForServer(null); + + try { + this.client.listActiveQueries(); + fail("a 403 should raise"); + } catch (ExecutionException e) { + assertTrue("message should explain the fix: " + e.getMessage(), + e.getMessage().contains("does not allow listing queries")); + } + } + + public void testListActiveQueriesOtherErrorThrowsGenericMessage() throws Exception { + startServer(); + stub("/v1/sql/active", 500, "internal error"); + this.client = clientForServer(null); + + try { + this.client.listActiveQueries(); + fail("a 500 should raise"); + } catch (ExecutionException e) { + assertTrue(e.getMessage().contains("500")); + } + } + + // ==================== cancelActiveQuery ==================== + + public void testCancelActiveQuerySucceedsOn200() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 200, + "{\"query_id\":\"" + VALID_QUERY_ID + "\",\"status\":\"CANCELLED\"}"); + this.client = clientForServer(null); + + this.client.cancelActiveQuery(VALID_QUERY_ID); + + assertEquals("POST", lastMethod.get()); + assertEquals("/v1/sql/" + VALID_QUERY_ID + "/cancel", lastPath.get()); + } + + public void testCancelActiveQuerySendsApiKeyWhenConfigured() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 200, "{}"); + this.client = clientForServer("test-app|test-key"); + + this.client.cancelActiveQuery(VALID_QUERY_ID); + + assertEquals("test-app|test-key", lastApiKeyHeader.get()); + } + + public void testCancelActiveQueryBadRequestThrowsSpecificMessage() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 400, "bad request"); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery(VALID_QUERY_ID); + fail("a 400 should raise"); + } catch (ExecutionException e) { + assertTrue(e.getMessage().contains("not a valid UUID")); + } + } + + public void testCancelActiveQueryForbiddenThrowsSpecificMessage() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 403, "forbidden"); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery(VALID_QUERY_ID); + fail("a 403 should raise"); + } catch (ExecutionException e) { + assertTrue(e.getMessage().contains("does not allow cancelling queries")); + } + } + + public void testCancelActiveQueryNotFoundThrowsSpecificMessage() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 404, "not found"); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery(VALID_QUERY_ID); + fail("a 404 should raise"); + } catch (ExecutionException e) { + assertTrue(e.getMessage().contains("no active query")); + } + } + + public void testCancelActiveQueryOtherErrorThrowsGenericMessage() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 500, "internal error"); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery(VALID_QUERY_ID); + fail("a 500 should raise"); + } catch (ExecutionException e) { + assertTrue(e.getMessage().contains("500")); + } + } + + public void testCancelActiveQueryRejectsNullQueryIdWithoutHttpCall() throws Exception { + startServer(); + // No context is stubbed: any request reaching the server 404s, which the + // assertions below distinguish from client-side rejection. + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery(null); + fail("a null queryId should raise IllegalArgumentException"); + } catch (IllegalArgumentException expected) { + // expected + } + assertNull("no request should have been sent", lastPath.get()); + } + + public void testCancelActiveQueryRejectsEmptyQueryIdWithoutHttpCall() throws Exception { + startServer(); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery(""); + fail("an empty queryId should raise IllegalArgumentException"); + } catch (IllegalArgumentException expected) { + // expected + } + assertNull("no request should have been sent", lastPath.get()); + } + + public void testCancelActiveQueryRejectsNonUuidWithoutHttpCall() throws Exception { + startServer(); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery("not-a-uuid"); + fail("a non-UUID queryId should raise IllegalArgumentException"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("not a valid UUID")); + } + assertNull("no request should have been sent", lastPath.get()); + } + + public void testCancelActiveQueryRejectsPathTraversalAttemptWithoutHttpCall() throws Exception { + startServer(); + this.client = clientForServer(null); + + try { + this.client.cancelActiveQuery("../../etc/passwd"); + fail("a path-traversal-shaped queryId should raise IllegalArgumentException"); + } catch (IllegalArgumentException expected) { + // expected + } + assertNull("no request should have been sent", lastPath.get()); + } +} From 8476912990e741fe9a022d4d8c17bd0825564f54 Mon Sep 17 00:00:00 2001 From: Viktor Yershov Date: Wed, 19 Aug 2026 11:33:35 -0700 Subject: [PATCH 2/4] fix: give authenticated ActiveQueriesTest cases a real Flight endpoint withApiKey() makes SpiceClient's constructor perform a real Flight handshake, but the two tests that configure an API key only stood up the HTTP mock server, not a Flight server. The handshake against a dead default Flight address failed unpredictably by platform (reliable on Windows CI, intermittent on Linux/macOS). Start a TestFlightSqlServer with matching credentials for those cases. --- src/test/java/ai/spice/ActiveQueriesTest.java | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/src/test/java/ai/spice/ActiveQueriesTest.java b/src/test/java/ai/spice/ActiveQueriesTest.java index b959bdc..f81608e 100644 --- a/src/test/java/ai/spice/ActiveQueriesTest.java +++ b/src/test/java/ai/spice/ActiveQueriesTest.java @@ -47,6 +47,7 @@ public class ActiveQueriesTest extends TestCase { private HttpServer server; private SpiceClient client; + private TestFlightSqlServer flightServer; private final AtomicReference lastMethod = new AtomicReference<>(); private final AtomicReference lastPath = new AtomicReference<>(); private final AtomicReference lastAcceptHeader = new AtomicReference<>(); @@ -76,7 +77,15 @@ private SpiceClient clientForServer(String apiKey) throws Exception { URI httpAddress = new URI("http://127.0.0.1:" + this.server.getAddress().getPort()); SpiceClientBuilder builder = SpiceClient.builder().withHttpAddress(httpAddress); if (apiKey != null) { - builder = builder.withApiKey(apiKey); + // withApiKey() makes the constructor perform a real Flight handshake, so an + // authenticated client needs a real (test) Flight endpoint to handshake against — + // these tests only exercise the HTTP path, but SpiceClient always builds its Flight + // channels eagerly. Without this, the handshake targets the builder's default flight + // address, where nothing is listening; that failed unpredictably by platform + // (reliably on Windows CI, only sometimes on Linux/macOS). + String appId = apiKey.split("\\|")[0]; + this.flightServer = new TestFlightSqlServer(appId, apiKey); + builder = builder.withFlightAddress(this.flightServer.flightUri()).withApiKey(apiKey); } return builder.build(); } @@ -91,6 +100,10 @@ protected void tearDown() throws Exception { this.server.stop(0); this.server = null; } + if (this.flightServer != null) { + this.flightServer.close(); + this.flightServer = null; + } super.tearDown(); } From e4b507356ffba9b29bcad0e06943dcaf01c9f041 Mon Sep 17 00:00:00 2001 From: Viktor Yershov Date: Wed, 19 Aug 2026 11:46:32 -0700 Subject: [PATCH 3/4] fix: address review feedback on active queries - Resolve the cancel/list URIs against the base address instead of concatenating, so a trailing-slash httpAddress doesn't miss the route. - Bound listActiveQueries with the same request timeout cancel/refresh use. - Reject a present-but-malformed "queries" field instead of silently reporting zero active queries. - Document that runtime releases through v2.1.5 don't scope either endpoint by credential. - Fix an undeclared variable in the README example. --- README.md | 7 ++- src/main/java/ai/spice/SpiceClient.java | 50 ++++++++++++++-- src/test/java/ai/spice/ActiveQueriesTest.java | 60 +++++++++++++++++++ 3 files changed, 109 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 023616d..9ad19ea 100644 --- a/README.md +++ b/README.md @@ -423,17 +423,20 @@ to the client that submitted it, so listing is the only way to find the ID that needs. ```java -for (ActiveQuery query : client.listActiveQueries()) { +List queries = client.listActiveQueries(); +for (ActiveQuery query : queries) { System.out.printf("%s %s %s%n", query.getQueryId(), query.getProtocol(), query.getSqlPreview()); } -client.cancelActiveQuery(queryId); +client.cancelActiveQuery(queries.get(0).getQueryId()); ``` Both calls are scoped to the authenticated API key or client certificate — not to this `SpiceClient` instance — and reach only the one runtime process behind this client's HTTP endpoint, which matters if that endpoint is a load balancer in front of several runtimes. +Runtime releases up to and including v2.1.5 do not scope either endpoint at all — every +caller sees and can cancel every query regardless of credential. ### Logging diff --git a/src/main/java/ai/spice/SpiceClient.java b/src/main/java/ai/spice/SpiceClient.java index b735f7d..c7b5e6b 100644 --- a/src/main/java/ai/spice/SpiceClient.java +++ b/src/main/java/ai/spice/SpiceClient.java @@ -1609,6 +1609,11 @@ public void refreshDataset(String dataset, RefreshOptions refreshOptions) throws * runtime instance this client's HTTP endpoint reaches, which behind a load * balancer may not be the instance that received a particular query. * + *

+ * Runtime releases up to and including v2.1.5 do not scope this endpoint at + * all: on those versions, every caller sees every query regardless of + * credential. + * * @return the currently running synchronous queries * @throws ExecutionException if the runtime is unreachable or returns an * unexpected response @@ -1616,7 +1621,21 @@ public void refreshDataset(String dataset, RefreshOptions refreshOptions) throws public List listActiveQueries() throws ExecutionException { logger.debug("Listing active queries"); try { - HttpRequest request = buildProbeRequest(this.httpAddress, "/v1/sql/active", this.apiKey); + // Resolve rather than concatenate: a base address with a trailing slash + // would otherwise produce "http://host:8090//v1/sql/active". + URI uri = this.httpAddress.resolve("/v1/sql/active"); + HttpRequest.Builder builder = HttpRequest.newBuilder() + .uri(uri) + // Bound the same way cancelActiveQuery and refreshDataset are, so a + // runtime that accepts the connection but never responds cannot hang + // this call forever. + .timeout(HTTP_REQUEST_TIMEOUT) + .header("X-Spice-User-Agent", Config.getUserAgent()) + .GET(); + if (!Strings.isNullOrEmpty(this.apiKey)) { + builder = builder.header("X-API-Key", this.apiKey); + } + HttpRequest request = builder.build(); HttpResponse response = httpClient().send(request, HttpResponse.BodyHandlers.ofString()); @@ -1658,16 +1677,27 @@ public List listActiveQueries() throws ExecutionException { * * @param body the JSON object the runtime returned * @return the parsed active queries, or an empty list when the body carries none + * @throws ExecutionException if the body is not the expected shape */ - private static List parseActiveQueries(String body) { - JsonElement root = JsonParser.parseString(body == null ? "" : body); + private static List parseActiveQueries(String body) throws ExecutionException { + JsonElement root; + try { + root = JsonParser.parseString(body == null ? "" : body); + } catch (JsonSyntaxException err) { + throw new ExecutionException("The runtime returned a malformed active-queries response", err); + } if (root == null || !root.isJsonObject()) { - return Collections.emptyList(); + throw new ExecutionException("The runtime returned an unexpected active-queries response", null); } + // "queries" absent is treated as none, but present-and-wrong-shape is a + // malformed response, not silently zero active queries. JsonElement queries = root.getAsJsonObject().get("queries"); - if (queries == null || !queries.isJsonArray()) { + if (queries == null) { return Collections.emptyList(); } + if (!queries.isJsonArray()) { + throw new ExecutionException("The runtime returned an unexpected active-queries response", null); + } List result = new ArrayList<>(); for (JsonElement element : queries.getAsJsonArray()) { @@ -1690,6 +1720,11 @@ private static List parseActiveQueries(String body) { * {@link #listActiveQueries()}, this reaches only the one runtime instance * this client's HTTP endpoint resolves to. * + *

+ * Runtime releases up to and including v2.1.5 do not scope this endpoint at + * all: on those versions, any known query ID can be cancelled regardless of + * credential. + * * @param queryId the ID of the query to cancel, from {@link #listActiveQueries()} * @throws ExecutionException if the runtime is unreachable or reports a failure */ @@ -1708,8 +1743,11 @@ public void cancelActiveQuery(String queryId) throws ExecutionException { logger.debug("Cancelling active query: {}", queryId); try { + // Resolve rather than concatenate: a base address with a trailing slash + // would otherwise produce "http://host:8090//v1/sql/.../cancel". + URI uri = this.httpAddress.resolve("/v1/sql/" + queryId + "/cancel"); HttpRequest.Builder builder = HttpRequest.newBuilder() - .uri(new URI(String.format("%s/v1/sql/%s/cancel", this.httpAddress, queryId))) + .uri(uri) .timeout(HTTP_REQUEST_TIMEOUT) .header("Accept", "application/json") .header("X-Spice-User-Agent", Config.getUserAgent()) diff --git a/src/test/java/ai/spice/ActiveQueriesTest.java b/src/test/java/ai/spice/ActiveQueriesTest.java index f81608e..734873b 100644 --- a/src/test/java/ai/spice/ActiveQueriesTest.java +++ b/src/test/java/ai/spice/ActiveQueriesTest.java @@ -90,6 +90,13 @@ private SpiceClient clientForServer(String apiKey) throws Exception { return builder.build(); } + /** Like {@link #clientForServer(String)}, but with a trailing slash on the base address. */ + private SpiceClient clientForServerWithTrailingSlash() throws Exception { + this.server.start(); + URI httpAddress = new URI("http://127.0.0.1:" + this.server.getAddress().getPort() + "/"); + return SpiceClient.builder().withHttpAddress(httpAddress).build(); + } + @Override protected void tearDown() throws Exception { if (this.client != null) { @@ -183,6 +190,47 @@ public void testListActiveQueriesOtherErrorThrowsGenericMessage() throws Excepti } } + public void testListActiveQueriesResolvesTrailingSlashBase() throws Exception { + startServer(); + stub("/v1/sql/active", 200, "{\"queries\":[]}"); + this.client = clientForServerWithTrailingSlash(); + + this.client.listActiveQueries(); + + assertEquals("a trailing slash on the base address must not double up", + "/v1/sql/active", lastPath.get()); + } + + public void testListActiveQueriesRejectsMalformedQueriesField() throws Exception { + startServer(); + // "queries" present but not an array: must not be silently treated as zero + // active queries. + stub("/v1/sql/active", 200, "{\"queries\":\"not-an-array\"}"); + this.client = clientForServer(null); + + try { + this.client.listActiveQueries(); + fail("a malformed queries field should raise"); + } catch (ExecutionException e) { + assertTrue("expected a malformed-response message: " + e.getMessage(), + e.getMessage().contains("unexpected active-queries response")); + } + } + + public void testListActiveQueriesRejectsNonObjectBody() throws Exception { + startServer(); + stub("/v1/sql/active", 200, "[]"); + this.client = clientForServer(null); + + try { + this.client.listActiveQueries(); + fail("a non-object body should raise"); + } catch (ExecutionException e) { + assertTrue("expected a malformed-response message: " + e.getMessage(), + e.getMessage().contains("unexpected active-queries response")); + } + } + // ==================== cancelActiveQuery ==================== public void testCancelActiveQuerySucceedsOn200() throws Exception { @@ -197,6 +245,18 @@ public void testCancelActiveQuerySucceedsOn200() throws Exception { assertEquals("/v1/sql/" + VALID_QUERY_ID + "/cancel", lastPath.get()); } + public void testCancelActiveQueryResolvesTrailingSlashBase() throws Exception { + startServer(); + stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 200, + "{\"query_id\":\"" + VALID_QUERY_ID + "\",\"status\":\"CANCELLED\"}"); + this.client = clientForServerWithTrailingSlash(); + + this.client.cancelActiveQuery(VALID_QUERY_ID); + + assertEquals("a trailing slash on the base address must not double up", + "/v1/sql/" + VALID_QUERY_ID + "/cancel", lastPath.get()); + } + public void testCancelActiveQuerySendsApiKeyWhenConfigured() throws Exception { startServer(); stub("/v1/sql/" + VALID_QUERY_ID + "/cancel", 200, "{}"); From 66975eee1632d8fbb7db0050e729ef1cde25e1de Mon Sep 17 00:00:00 2001 From: Viktor Yershov Date: Fri, 21 Aug 2026 10:09:38 -0700 Subject: [PATCH 4/4] fix: address remaining Copilot review feedback on active queries - Reject non-object entries in the /v1/sql/active queries array instead of silently dropping them, matching the method's malformed-response contract. - Make 403 messages credential-neutral so mTLS callers aren't pointed at API keys. - Guard the README's cancellation example against an empty query list. --- README.md | 4 +++- src/main/java/ai/spice/SpiceClient.java | 15 +++++++++----- src/test/java/ai/spice/ActiveQueriesTest.java | 20 +++++++++++++++++-- 3 files changed, 31 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 2c3b8a3..4f9d669 100644 --- a/README.md +++ b/README.md @@ -468,7 +468,9 @@ for (ActiveQuery query : queries) { query.getQueryId(), query.getProtocol(), query.getSqlPreview()); } -client.cancelActiveQuery(queries.get(0).getQueryId()); +if (!queries.isEmpty()) { + client.cancelActiveQuery(queries.get(0).getQueryId()); +} ``` Both calls are scoped to the authenticated API key or client certificate — not to this diff --git a/src/main/java/ai/spice/SpiceClient.java b/src/main/java/ai/spice/SpiceClient.java index 524354d..41d353e 100644 --- a/src/main/java/ai/spice/SpiceClient.java +++ b/src/main/java/ai/spice/SpiceClient.java @@ -1716,8 +1716,8 @@ public List listActiveQueries() throws ExecutionException { if (response.statusCode() == 403) { throw new ExecutionException( - "Failed to list active queries: the configured API key does not allow listing queries, " - + "use a key with write access", + "Failed to list active queries: the configured credentials do not allow listing queries, " + + "use credentials with write access", null); } if (response.statusCode() != 200) { @@ -1776,9 +1776,13 @@ private static List parseActiveQueries(String body) throws Executio List result = new ArrayList<>(); for (JsonElement element : queries.getAsJsonArray()) { - if (element.isJsonObject()) { - result.add(GSON.fromJson(element, ActiveQuery.class)); + // A non-object entry (e.g. null) violates the endpoint's schema. Reject the + // whole response rather than silently omitting it, which would otherwise tell + // the caller a query isn't running when the response was simply malformed. + if (!element.isJsonObject()) { + throw new ExecutionException("The runtime returned an unexpected active-queries response", null); } + result.add(GSON.fromJson(element, ActiveQuery.class)); } return result; } @@ -1844,7 +1848,8 @@ public void cancelActiveQuery(String queryId) throws ExecutionException { null); case 403: throw new ExecutionException( - "the configured API key does not allow cancelling queries, use a key with write access", + "the configured credentials do not allow cancelling queries, " + + "use credentials with write access", null); case 404: throw new ExecutionException( diff --git a/src/test/java/ai/spice/ActiveQueriesTest.java b/src/test/java/ai/spice/ActiveQueriesTest.java index 734873b..859f7d2 100644 --- a/src/test/java/ai/spice/ActiveQueriesTest.java +++ b/src/test/java/ai/spice/ActiveQueriesTest.java @@ -173,7 +173,7 @@ public void testListActiveQueriesForbiddenThrowsSpecificMessage() throws Excepti fail("a 403 should raise"); } catch (ExecutionException e) { assertTrue("message should explain the fix: " + e.getMessage(), - e.getMessage().contains("does not allow listing queries")); + e.getMessage().contains("do not allow listing queries")); } } @@ -217,6 +217,22 @@ public void testListActiveQueriesRejectsMalformedQueriesField() throws Exception } } + public void testListActiveQueriesRejectsNonObjectArrayEntry() throws Exception { + startServer(); + // A null entry violates the endpoint's schema: it must not be silently + // dropped, which would otherwise under-report active queries. + stub("/v1/sql/active", 200, "{\"queries\":[null]}"); + this.client = clientForServer(null); + + try { + this.client.listActiveQueries(); + fail("a non-object array entry should raise"); + } catch (ExecutionException e) { + assertTrue("expected a malformed-response message: " + e.getMessage(), + e.getMessage().contains("unexpected active-queries response")); + } + } + public void testListActiveQueriesRejectsNonObjectBody() throws Exception { startServer(); stub("/v1/sql/active", 200, "[]"); @@ -289,7 +305,7 @@ public void testCancelActiveQueryForbiddenThrowsSpecificMessage() throws Excepti this.client.cancelActiveQuery(VALID_QUERY_ID); fail("a 403 should raise"); } catch (ExecutionException e) { - assertTrue(e.getMessage().contains("does not allow cancelling queries")); + assertTrue(e.getMessage().contains("do not allow cancelling queries")); } }