fix(engine): make client-chosen executionUuids single-use

Re-sending an in-flight executionUuid overwrote its registration, failed
the audit insert (UNIQUE), and the finally-unregister then removed the
original - escaping maxConcurrentExecutionsPerUser and cancellation.

- RunningStatements.register refuses an id already registered
  (putIfAbsent); unregister(uuid, registration) removes only its own.
- The engine refuses a client id that is running or already in the
  audit, before registering or recording anything
  (DuplicateExecutionException; the route provider answers 409).
- ExecutionAuditor.record() checks the insert result and fails loudly.
- Execution ids are lower-cased at the API edge (execute, rows, export,
  cancel, snapshots) and in engine.cancel.
- Tests for each, including 409 over HTTP.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
james.maesandClaude Opus 5.5 committed 2026-10-11 13:36:09 -05:00
1 parent ebef56f34b
commit 52d2d6bc87
10 files changed
+272 -21

No files matched your search

+3 -2
View File
@@ -274,7 +274,7 @@ The response is `{executionUuid, dialect, analyze, plan: PlanNode, raw: string}`
SQL errors are results, not HTTP errors: `/execute` and `/explain` return 200 with `status` FAILED, TIMED_OUT, or CANCELLED plus `error` and `sqlState`.
All endpoints need an authenticated QQQ session (cookie or Authorization header, via `ExecutorSessionUtils.setupSession`) and `sqlWorkbench.hasAccess`. Errors are `{"error": "..."}` with 400, 401, 403, 404, or 500, through `QJavalinUtils.handleException`. Bodies are JSON. Request and response bodies are never logged.
All endpoints need an authenticated QQQ session (cookie or Authorization header, via `ExecutorSessionUtils.setupSession`) and `sqlWorkbench.hasAccess`. Errors are `{"error": "..."}` with 400, 401, 403, 404, 409 (a re-used `executionUuid`), or 500, through `QJavalinUtils.handleException`. Execution UUIDs are case-insensitive on every path (`/execute`, `/executions/{uuid}/rows`, `/export`, `/cancel`, snapshots); the server stores and returns them lower-case. Bodies are JSON. Request and response bodies are never logged.
```yaml
GET /capabilities
@@ -287,7 +287,8 @@ GET /schema/{backendName}?refresh=bool -> Schema (see 4.6)
POST /execute
body: { backendName, sql, params?: {name: value}, mode?: READ_ONLY|WRITE, notebookId?, cellKey?,
pageSize?, confirmToken?, acknowledgeRowGuard?,
executionUuid? } # optional client-chosen UUID, so the client can cancel before the response
executionUuid? } # optional client-chosen UUID, so the client can cancel before the response;
# single-use: an id that is running or already audited gets 409
200 (READ_ONLY / commit): { executionUuid, status, mode, statements: [{index, kind, leadingKeyword, affectedRows?}],
columns: [Column], rows: [[...]], rowCount, hasMore, truncated, truncatedReason?, durationMs,
rowsAffected? }
@@ -30,6 +30,7 @@ import com.kingsrook.qqq.backend.core.utils.JsonUtils;
import com.kingsrook.qqq.backend.core.utils.StringUtils;
import com.kingsrook.qbits.sqlworkbench.SqlWorkbenchRuntime;
import com.kingsrook.qbits.sqlworkbench.engine.Caller;
import com.kingsrook.qbits.sqlworkbench.engine.DuplicateExecutionException;
import com.kingsrook.qbits.sqlworkbench.engine.ResultCache;
import com.kingsrook.qqq.middleware.javalin.QJavalinRouteProviderInterface;
import com.kingsrook.qqq.middleware.javalin.QJavalinUtils;
@@ -37,6 +38,7 @@ import com.kingsrook.qqq.middleware.javalin.executors.ExecutorSessionUtils;
import io.javalin.config.JavalinConfig;
import io.javalin.http.ContentType;
import io.javalin.http.Context;
import org.eclipse.jetty.http.HttpStatus;
/*******************************************************************************
@@ -46,7 +48,8 @@ import io.javalin.http.Context;
** Each request gets a session from the instance's authentication module
** (ExecutorSessionUtils), then SqlWorkbenchApi checks permissions and does
** the work. Errors map through QJavalinUtils.handleException: 400 bad
** request, 401 unauthenticated, 403 not permitted, 404 unknown. POST / PUT
** request, 401 unauthenticated, 403 not permitted, 404 unknown, 409 a
** re-used executionUuid. POST / PUT
** bodies must be JSON (Content-Type application/json): a cross-site form
** cannot send that without a CORS preflight. Request and response bodies
** are never logged.
@@ -129,7 +132,7 @@ public class SqlWorkbenchRouteProvider implements QJavalinRouteProviderInterface
}
catch(Exception e)
{
QJavalinUtils.handleException(null, context, e);
QJavalinUtils.handleException(e instanceof DuplicateExecutionException ? HttpStatus.Code.CONFLICT : null, context, e);
}
finally
{
@@ -0,0 +1,38 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.sqlworkbench.engine;
import com.kingsrook.qqq.backend.core.exceptions.QBadRequestException;
/*******************************************************************************
** A client-chosen executionUuid that is already running or already in the
** audit. Execution ids are single-use; the route provider answers 409.
*******************************************************************************/
public class DuplicateExecutionException extends QBadRequestException
{
/*******************************************************************************
**
*******************************************************************************/
public DuplicateExecutionException(String executionUuid)
{
super("executionUuid " + executionUuid + " was already used; send a new UUID for each run.");
}
}
@@ -22,7 +22,12 @@ import java.time.Instant;
import java.util.List;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertOutput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
@@ -30,6 +35,7 @@ import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qbits.sqlworkbench.SqlWorkbenchQBitConfig;
import com.kingsrook.qbits.sqlworkbench.SystemUser;
import com.kingsrook.qbits.sqlworkbench.model.SqlWorkbenchExecution;
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
/*******************************************************************************
@@ -40,6 +46,8 @@ import com.kingsrook.qbits.sqlworkbench.model.SqlWorkbenchExecution;
*******************************************************************************/
public class ExecutionAuditor
{
private static final QLogger LOG = QLogger.getLogger(ExecutionAuditor.class);
public static final int MAX_ERROR_LENGTH = 4000;
private final SqlWorkbenchQBitConfig config;
@@ -82,10 +90,7 @@ public class ExecutionAuditor
inserted[0] = output.getRecords().get(0);
});
if(inserted[0] == null || (inserted[0].getErrors() != null && !inserted[0].getErrors().isEmpty()) || inserted[0].getValueInteger("id") == null)
{
throw (new QException("Could not write the SQL Workbench audit record; the SQL was not run."));
}
throwIfNotInserted(inserted[0], "Could not write the SQL Workbench audit record; the SQL was not run.");
execution.setId(inserted[0].getValueInteger("id"));
return (execution.getId());
}
@@ -128,7 +133,35 @@ public class ExecutionAuditor
{
execution.setCreateDate(execution.getStartTime());
execution.setErrorMessage(truncate(execution.getErrorMessage()));
asSystem(() -> new InsertAction().execute(new InsertInput(tableName()).withRecordEntity(execution)));
QRecord[] inserted = new QRecord[1];
asSystem(() -> inserted[0] = new InsertAction().execute(new InsertInput(tableName()).withRecordEntity(execution)).getRecords().get(0));
throwIfNotInserted(inserted[0], "Could not write the SQL Workbench audit record of a rejected request.");
}
/*******************************************************************************
** Whether the audit already holds a row with this execution id.
*******************************************************************************/
public boolean exists(String executionUuid) throws QException
{
return (SystemUser.call(() -> !QueryAction.execute(tableName(), new QQueryFilter(new QFilterCriteria("executionUuid", QCriteriaOperator.EQUALS, executionUuid)).withLimit(1)).isEmpty()));
}
/*******************************************************************************
** Fail loudly when an audit insert did not produce a row (record errors,
** e.g. a duplicate executionUuid, or no generated id).
*******************************************************************************/
private static void throwIfNotInserted(QRecord record, String message) throws QException
{
if(record == null || (record.getErrors() != null && !record.getErrors().isEmpty()) || record.getValueInteger("id") == null)
{
String detail = record == null || record.getErrors() == null || record.getErrors().isEmpty() ? "no id returned" : record.getErrors().get(0).getMessage();
LOG.warn(message, logPair("detail", detail));
throw (new QException(message));
}
}
@@ -86,29 +86,49 @@ public class RunningStatements
/*******************************************************************************
** Register an execution, unless the user already has maxPerUser running.
** An id that is already registered is refused (never overwritten), so a
** re-sent id cannot replace - and later unregister - the original run.
**
** @return the registration, or null when the user is at the limit
** @throws DuplicateExecutionException when the id is already running
*******************************************************************************/
public synchronized Running register(String executionUuid, String userId, int maxPerUser)
public synchronized Running register(String executionUuid, String userId, int maxPerUser) throws DuplicateExecutionException
{
if(running.containsKey(executionUuid))
{
throw (new DuplicateExecutionException(executionUuid));
}
long mine = running.values().stream().filter(r -> Objects.equals(r.userId, userId)).count();
if(mine >= maxPerUser)
{
return (null);
}
Running registration = new Running(userId);
running.put(executionUuid, registration);
if(running.putIfAbsent(executionUuid, registration) != null)
{
throw (new DuplicateExecutionException(executionUuid));
}
return (registration);
}
/*******************************************************************************
**
** Remove this registration - only if it is still the one under the id.
*******************************************************************************/
public void unregister(String executionUuid)
public void unregister(String executionUuid, Running registration)
{
running.remove(executionUuid);
running.remove(executionUuid, registration);
}
/*******************************************************************************
** Whether an execution with this id is in flight.
*******************************************************************************/
public boolean isRunning(String executionUuid)
{
return (executionUuid != null && running.containsKey(executionUuid));
}
@@ -33,6 +33,7 @@ import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
@@ -112,9 +113,22 @@ public class SqlExecutionEngine
*******************************************************************************/
public Map<String, Object> execute(ExecutionRequest request, Caller caller) throws QException
{
if(request.executionUuid() != null && !UUID_PATTERN.matcher(request.executionUuid()).matches())
if(request.executionUuid() != null)
{
throw (new QBadRequestException("executionUuid must be a UUID."));
if(!UUID_PATTERN.matcher(request.executionUuid()).matches())
{
throw (new QBadRequestException("executionUuid must be a UUID."));
}
/////////////////////////////////////////////////////////////////////
// ids are single-use: a re-sent id must not reach the registry or //
// the audit (UNIQUE) - it is refused before anything is recorded //
/////////////////////////////////////////////////////////////////////
String uuid = request.executionUuid().toLowerCase(Locale.ROOT);
if(runtime.getRunningStatements().isRunning(uuid) || auditor.exists(uuid))
{
throw (new DuplicateExecutionException(uuid));
}
}
WorkbenchBackends.Target target = backends.getTarget(request.backendName());
@@ -226,7 +240,7 @@ public class SqlExecutionEngine
*******************************************************************************/
public boolean cancel(String executionUuid, Caller caller)
{
return (runtime.getRunningStatements().cancel(executionUuid, caller.userId()));
return (executionUuid != null && runtime.getRunningStatements().cancel(executionUuid.toLowerCase(Locale.ROOT), caller.userId()));
}
@@ -682,7 +696,7 @@ public class SqlExecutionEngine
}
finally
{
runtime.getRunningStatements().unregister(audit.getExecutionUuid());
runtime.getRunningStatements().unregister(audit.getExecutionUuid(), registration);
}
}
@@ -790,7 +804,7 @@ public class SqlExecutionEngine
private SqlWorkbenchExecution newAudit(ExecutionRequest request, Caller caller, Map<String, Object> params, ExecutionMode mode)
{
return (new SqlWorkbenchExecution()
.withExecutionUuid(request.executionUuid() != null ? request.executionUuid().toLowerCase() : UUID.randomUUID().toString())
.withExecutionUuid(request.executionUuid() != null ? request.executionUuid().toLowerCase(Locale.ROOT) : UUID.randomUUID().toString())
.withUserId(caller.userId())
.withUserName(caller.userName())
.withBackendName(request.backendName())
@@ -157,6 +157,22 @@ class SqlWorkbenchApiTest extends BaseSqlWorkbenchTest
assertThatThrownBy(() -> api.execute(caller(READER), Map.of("sql", "SELECT 1"))).isInstanceOf(QBadRequestException.class);
assertThatThrownBy(() -> api.execute(caller(READER), Map.of("backendName", BACKEND_NAME, "sql", "SELECT 1", "params", "x"))).isInstanceOf(QBadRequestException.class);
assertThatThrownBy(() -> api.execute(caller(READER), Map.of("backendName", BACKEND_NAME, "sql", "SELECT 1", "pageSize", "x"))).isInstanceOf(QBadRequestException.class);
////////////////////////////////////////////////////////////
// ids are normalized at the edge: any case finds the run //
////////////////////////////////////////////////////////////
String upper = uuid.toUpperCase();
assertThat(api.rows(caller(READER), upper, 0, 10, null, null).get("executionUuid")).isEqualTo(uuid);
assertThat(api.exportable(caller(READER), " " + upper + " ")).isNotNull();
assertThat(api.cancel(caller(READER), upper).get("cancelled")).isEqualTo(false);
assertThat(SqlWorkbenchApi.normalizeUuid(null)).isNull();
String clientUuid = "9B2C7A10-5E1F-4C3D-8A2B-1C0D9E8F7A6B";
Map<String, Object> mine = api.execute(caller(READER), Map.of("backendName", BACKEND_NAME, "sql", "SELECT 1", "executionUuid", clientUuid));
assertThat(mine.get("executionUuid")).isEqualTo(clientUuid.toLowerCase());
assertThat(api.rows(caller(READER), clientUuid, 0, 10, null, null).get("rowCount")).isEqualTo(1);
assertThatThrownBy(() -> api.execute(caller(READER), Map.of("backendName", BACKEND_NAME, "sql", "SELECT 1", "executionUuid", clientUuid.toLowerCase())))
.isInstanceOf(com.kingsrook.qbits.sqlworkbench.engine.DuplicateExecutionException.class);
assertThat(SqlWorkbenchApi.intOrNull("12")).isEqualTo(12);
assertThat(SqlWorkbenchApi.intOrNull("")).isNull();
}
@@ -153,6 +153,18 @@ class SqlWorkbenchRouteProviderTest extends BaseSqlWorkbenchTest
assertThat(json(send("POST", "/executions/" + uuid + "/cancel", READER, "{}", "application/json")).get("cancelled")).isEqualTo(false);
assertThat((List<?>) json(send("GET", "/history?limit=5", READER, null, null)).get("executions")).hasSize(3);
assertThat(json(send("GET", "/schema/h2", READER, null, null)).get("dialect")).isEqualTo("h2");
///////////////////////////////////////////////////////////////////
// a client-chosen id: any case on the paths, single-use (409) //
///////////////////////////////////////////////////////////////////
String clientUuid = "3F2504E0-4F89-11D3-9A0C-0305E82C3301";
String body = "{\"backendName\":\"h2\",\"sql\":\"SELECT 1 AS one\",\"executionUuid\":\"" + clientUuid + "\"}";
assertThat(json(send("POST", "/execute", READER, body, "application/json")).get("executionUuid")).isEqualTo(clientUuid.toLowerCase());
assertThat(send("GET", "/executions/" + clientUuid + "/rows", READER, null, null).statusCode()).isEqualTo(200);
assertThat(send("GET", "/executions/" + clientUuid + "/export", READER, null, null).statusCode()).isEqualTo(200);
HttpResponse<String> again = send("POST", "/execute", READER, body, "application/json");
assertThat(again.statusCode()).isEqualTo(409);
assertThat(again.body()).contains("already used");
}
@@ -232,12 +232,45 @@ class EngineComponentsTest
}
one.setStatement(null);
assertThat(running.cancel("missing", "u")).isFalse();
running.unregister("1");
running.unregister("1", one);
assertThat(running.size()).isZero();
}
/*******************************************************************************
** A re-sent id is refused, never overwrites the original, and the
** original's registration is only removed by its own unregister.
*******************************************************************************/
@Test
void testRunningStatementsRefuseDuplicateIds() throws Exception
{
RunningStatements running = new RunningStatements();
RunningStatements.Running original = running.register("dup", "u", 5);
assertThat(running.isRunning("dup")).isTrue();
assertThatThrownBy(() -> running.register("dup", "u", 5)).isInstanceOf(DuplicateExecutionException.class);
assertThatThrownBy(() -> running.register("dup", "someone-else", 5)).isInstanceOf(DuplicateExecutionException.class);
assertThat(running.size()).isEqualTo(1);
///////////////////////////////////////////////////////////////////
// a stranger's unregister under the same id leaves the original //
///////////////////////////////////////////////////////////////////
RunningStatements.Running other = running.register("other", "u", 5);
running.unregister("dup", other);
assertThat(running.isRunning("dup")).isTrue();
assertThat(running.cancel("dup", "u")).isTrue();
assertThat(original.isCancelled()).isTrue();
running.unregister("dup", original);
running.unregister("other", other);
assertThat(running.size()).isZero();
assertThat(running.isRunning("dup")).isFalse();
assertThat(running.isRunning(null)).isFalse();
}
/*******************************************************************************
** SQLite: query_only blocks writes in read-only sessions, and is reset so
** the same connection writes again afterwards.
@@ -363,7 +363,7 @@ class SqlExecutionEngineTest extends BaseSqlWorkbenchTest
assertThat(engine.cancel("blocker", caller(OTHER_READER))).isFalse();
assertThat(engine.cancel("blocker", caller(READER))).isTrue();
assertThat(blocker.isCancelled()).isTrue();
runtime.getRunningStatements().unregister("blocker");
runtime.getRunningStatements().unregister("blocker", blocker);
assertThat(run("SELECT 1", Map.of(), READER).get("status")).isEqualTo("SUCCEEDED");
}
@@ -408,6 +408,87 @@ class SqlExecutionEngineTest extends BaseSqlWorkbenchTest
/*******************************************************************************
** A client uuid already in the audit is refused (in any case), before
** anything is registered or recorded.
*******************************************************************************/
@Test
void testReusedExecutionUuidIsRefused() throws Exception
{
String uuid = "1b4e28ba-2fa1-11d2-883f-0016d3cca427";
engine.execute(new ExecutionRequest(BACKEND_NAME, "SELECT 1", Map.of(), false, null, null, null, null, false, uuid), caller(READER));
assertThat(audits()).hasSize(1);
for(String again : List.of(uuid, uuid.toUpperCase()))
{
assertThatThrownBy(() -> engine.execute(new ExecutionRequest(BACKEND_NAME, "SELECT 2", Map.of(), false, null, null, null, null, false, again), caller(READER)))
.isInstanceOf(DuplicateExecutionException.class).hasMessageContaining("already used");
assertThatThrownBy(() -> engine.execute(new ExecutionRequest(BACKEND_NAME, "SELECT 2", Map.of(), false, null, null, null, null, false, again), caller(OTHER_READER)))
.isInstanceOf(DuplicateExecutionException.class);
}
assertThat(audits()).hasSize(1);
assertThat(SqlWorkbenchRuntime.get(qInstance).getRunningStatements().size()).isZero();
}
/*******************************************************************************
** Re-sending the id of a run in flight neither overwrites nor removes its
** registration, so the concurrency limit and cancel still hold.
*******************************************************************************/
@Test
void testInFlightExecutionUuidIsRefused() throws Exception
{
SqlWorkbenchRuntime runtime = SqlWorkbenchRuntime.get(qInstance);
String uuid = "6fa459ea-ee8a-3ca4-894e-db77e160355e";
config.setMaxConcurrentExecutionsPerUser(1);
RunningStatements.Running inFlight = runtime.getRunningStatements().register(uuid, READER, 1);
assertThatThrownBy(() -> engine.execute(new ExecutionRequest(BACKEND_NAME, "SELECT 1", Map.of(), false, null, null, null, null, false, uuid.toUpperCase()), caller(READER)))
.isInstanceOf(DuplicateExecutionException.class);
assertThatThrownBy(() -> engine.execute(new ExecutionRequest(BACKEND_NAME, "SELECT 1", Map.of(), false, null, null, null, null, false, uuid), caller(OTHER_READER)))
.isInstanceOf(DuplicateExecutionException.class);
///////////////////////////////////////////////////
// the original is still registered and counted //
///////////////////////////////////////////////////
assertThat(runtime.getRunningStatements().isRunning(uuid)).isTrue();
assertThatThrownBy(() -> run("SELECT 1", Map.of(), READER)).isInstanceOf(QBadRequestException.class).hasMessageContaining("running");
assertThat(engine.cancel(uuid.toUpperCase(), caller(READER))).isTrue();
assertThat(inFlight.isCancelled()).isTrue();
assertThat(audits()).extracting(SqlWorkbenchExecution::getExecutionUuid).doesNotContain(uuid);
runtime.getRunningStatements().unregister(uuid, inFlight);
assertThat(run("SELECT 1", Map.of(), READER).get("status")).isEqualTo("SUCCEEDED");
}
/*******************************************************************************
** An audit insert that fails (here: a duplicate executionUuid) throws,
** rather than being dropped silently.
*******************************************************************************/
@Test
void testAuditRecordFailsLoudly() throws Exception
{
String uuid = (String) run("SELECT 1", Map.of(), READER).get("executionUuid");
ExecutionAuditor auditor = new ExecutionAuditor(config);
SqlWorkbenchExecution rejected = new SqlWorkbenchExecution()
.withExecutionUuid(uuid)
.withUserId(READER)
.withBackendName(BACKEND_NAME)
.withMode(ExecutionMode.READ_ONLY.name())
.withStatus(ExecutionStatus.REJECTED.name())
.withSqlText("SELECT 1")
.withStartTime(java.time.Instant.now());
assertThatThrownBy(() -> auditor.record(rejected)).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class).hasMessageContaining("audit");
assertThat(audits()).hasSize(1);
assertThat(auditor.exists(uuid)).isTrue();
assertThat(auditor.exists("00000000-0000-0000-0000-000000000000")).isFalse();
}
/*******************************************************************************
**
*******************************************************************************/