[bugfix] Acknowledge WebSocket cancels and poll watchdog during

serialization

Cancel was fire-and-forget: clients got no response when the query was
missing, and kill() during the non-streaming serialize path was never
observed because proceed() is only called during evaluation.

Send cancelling/error acks on cancel, check isTerminating/proceed()
before
serializeAll(), and tighten cancellation/timeout tests with bounded
FLWOR
queries so they stay reliable under reuseForks.

Author: duncdrum

exist-core/src/main/java/org/exist/http/ws/EvalProtocol.java

@@ -64,6 +64,7 @@ public final class EvalProtocol {
     public static final String PHASE_COMPILING = "compiling";
     public static final String PHASE_EVALUATING = "evaluating";
     public static final String PHASE_SERIALIZING = "serializing";
+    public static final String PHASE_CANCELLING = "cancelling";
     public static final String PHASE_COMPLETE = "complete";
 
     private EvalProtocol() {

exist-core/src/main/java/org/exist/http/ws/EvalWebSocketEndpoint.java

@@ -192,7 +192,7 @@ public void onMessage(final String message, final Session session) {
 
         switch (msg.action()) {
             case EvalProtocol.ACTION_EVAL -> handleEval(session, evalSession, msg);
-            case EvalProtocol.ACTION_CANCEL -> handleCancel(evalSession, msg);
+            case EvalProtocol.ACTION_CANCEL -> handleCancel(session, evalSession, msg);
             case EvalProtocol.ACTION_COMPILE -> handleCompile(session, evalSession, msg);
             case EvalProtocol.ACTION_ADMIN_CANCEL -> handleAdminCancel(session, evalSession, msg);
             default -> {
@@ -262,11 +262,21 @@ private void handleEval(final Session session, final EvalSession evalSession,
         });
     }
 
-    private void handleCancel(final EvalSession evalSession,
+    private void handleCancel(final Session session, final EvalSession evalSession,
                                final EvalProtocol.ClientMessage msg) {
         final boolean cancelled = evalSession.cancelQuery(msg.id());
-        if (!cancelled) {
-            LOG.debug("Cancel requested for unknown query: {}", msg.id());
+        try {
+            if (!cancelled) {
+                session.getBasicRemote().sendText(
+                        EvalProtocol.errorMessage(msg.id(), null,
+                                "Query not found or already completed", 0, 0, null));
+                LOG.debug("Cancel requested for unknown query: {}", msg.id());
+            } else {
+                session.getBasicRemote().sendText(
+                        EvalProtocol.progressMessage(msg.id(), EvalProtocol.PHASE_CANCELLING, 0, 0));
+            }
+        } catch (final IOException e) {
+            LOG.debug("Failed to send cancel acknowledgement: {}", e.getMessage());
         }
     }
 

exist-core/src/main/java/org/exist/http/ws/QueryExecutor.java

@@ -149,6 +149,22 @@ public void execute(final Session wsSession, final EvalSession evalSession,
                         streamResults(wsSession, msg.id(), broker, result, outputProperties,
                                 msg.stream().chunkSize(), timing, startTime, watchDog);
                     } else {
+                        if (watchDog.isTerminating()) {
+                            timing.serialize = System.currentTimeMillis() - serStart;
+                            timing.total = System.currentTimeMillis() - startTime;
+                            sendCancelled(wsSession, msg.id(), 0, timing);
+                            QueryMonitorBroadcaster.broadcastEvent("cancelled", msg.id(), user, msg.query(),
+                                    null, 0, timing.total);
+                            return;
+                        }
+                        try {
+                            watchDog.proceed(null);
+                        } catch (final TerminatedException e) {
+                            timing.serialize = System.currentTimeMillis() - serStart;
+                            reportTerminationOrError(wsSession, evalSession, msg, timing, startTime,
+                                    watchDog, ErrorInfo.of(e.getMessage()));
+                            return;
+                        }
                         final String serialized = serializeAll(broker, result, outputProperties);
                         timing.serialize = System.currentTimeMillis() - serStart;
                         timing.total = System.currentTimeMillis() - startTime;

exist-core/src/test/java/org/exist/http/ws/EvalWebSocketEndpointTest.java

@@ -68,6 +68,19 @@ public class EvalWebSocketEndpointTest {
     private static final JsonFactory JSON_FACTORY = new JsonFactory();
 
     private static final String TEST_COLLECTION = "/db/ws-eval-test";
+    /** Bounded FLWOR loop for cancel tests: long enough to cancel, short enough for CI. */
+    private static final String CANCEL_TEST_QUERY =
+            "for $i in 1 to 10000000 return ()";
+    /** Heavier loop so max-execution-time fires before completion on fast hardware. */
+    private static final String TIMEOUT_TEST_QUERY =
+            "for $i in 1 to 999999999 return ()";
+    private static final long CANCEL_MAX_EXECUTION_MS = 15_000L;
+    private static final long CANCEL_AWAIT_SEC = 20L;
+    private static final long TIMEOUT_MAX_EXECUTION_MS = 2_000L;
+    /** Wait for eval to start when CI reuses forks and brokers are busy. */
+    private static final long TIMEOUT_START_AWAIT_SEC = 30L;
+    /** After evaluating begins, 2s watchdog should fire on the next few proceed() calls. */
+    private static final long TIMEOUT_ERROR_AWAIT_SEC = 8L;
     private static final String TEST_MODULE = """
             module namespace test = 'http://exist-db.org/test';
             declare function test:hello($name as xs:string) as xs:string {
@@ -475,31 +488,21 @@ public void onMessage(final String message) {
         }, createAdminConfig(), getWsUri());
 
         try {
-            // GC-free query: return () produces no objects per iteration, so the JVM
-            // stays out of stop-the-world GC and the query thread calls proceed() on
-            // every iteration — letting the volatile terminate flag be observed within
-            // microseconds of kill().  String-producing variants (string($i)) cause
-            // heavy GC that can stall the query thread for several seconds on CI.
-            // max-execution-time is a safety net only; the cancel should fire first.
             session.getBasicRemote().sendText(
                     "{\"action\":\"eval\",\"id\":\"q-cancel\"," +
-                    "\"query\":\"for $i in 1 to 999999999 return ()\"," +
-                    "\"max-execution-time\":30000}");
+                    "\"query\":\"" + CANCEL_TEST_QUERY + "\"," +
+                    "\"max-execution-time\":" + CANCEL_MAX_EXECUTION_MS + "}");
 
             // Wait for the server to confirm the query is executing before cancelling.
-            // Without this, the cancel may arrive before the watchdog is registered and
-            // be silently dropped, leaving the query to run until max-execution-time fires.
             assertTrue("Query should start executing within 10s",
                     progressLatch.await(10, TimeUnit.SECONDS));
 
             session.getBasicRemote().sendText(
                     "{\"action\":\"cancel\",\"id\":\"q-cancel\"}");
 
-            // With terminate declared volatile in XQueryWatchDog and no GC pressure
-            // from the query, cancellation is visible at the very next proceed() call
-            // — microseconds after kill(). 10s gives ample CI headroom.
-            assertTrue("Should receive cancelled/error within 10s",
-                    cancelledLatch.await(10, TimeUnit.SECONDS));
+            // Await longer than max-execution-time so the watchdog safety net can fire on slow CI.
+            assertTrue("Should receive cancelled/error within " + CANCEL_AWAIT_SEC + "s",
+                    cancelledLatch.await(CANCEL_AWAIT_SEC, TimeUnit.SECONDS));
             assertEquals("q-cancel", cancelledMsg.get().get("id"));
         } finally {
             session.close();
@@ -632,6 +635,7 @@ public void onMessage(final String message) {
 
     @Test
     public void maxExecutionTime() throws Exception {
+        final CountDownLatch evalLatch = new CountDownLatch(1);
         final CountDownLatch errorLatch = new CountDownLatch(1);
         final AtomicReference<Map<String, Object>> errorMsg = new AtomicReference<>();
 
@@ -644,7 +648,10 @@ public void onOpen(final Session session, final EndpointConfig config) {
                     public void onMessage(final String message) {
                         try {
                             final Map<String, Object> parsed = parseJson(message);
-                            if ("error".equals(parsed.get("type"))
+                            if ("progress".equals(parsed.get("type"))
+                                    && "evaluating".equals(parsed.get("phase"))) {
+                                evalLatch.countDown();
+                            } else if ("error".equals(parsed.get("type"))
                                     || "cancelled".equals(parsed.get("type"))) {
                                 errorMsg.set(parsed);
                                 errorLatch.countDown();
@@ -658,16 +665,16 @@ public void onMessage(final String message) {
         }, createAdminConfig(), getWsUri());
 
         try {
-            // GC-free query: return () avoids heap exhaustion on CI, ensuring the 2s
-            // watchdog timeout fires reliably via proceed() rather than OOM killing
-            // the thread before the timeout check runs.
             session.getBasicRemote().sendText(
                     "{\"action\":\"eval\",\"id\":\"q-timeout\"," +
-                    "\"query\":\"for $i in 1 to 999999999 return ()\"," +
-                    "\"max-execution-time\":2000}");
+                    "\"query\":\"" + TIMEOUT_TEST_QUERY + "\"," +
+                    "\"max-execution-time\":" + TIMEOUT_MAX_EXECUTION_MS + "}");
+
+            assertTrue("Query should reach evaluating within " + TIMEOUT_START_AWAIT_SEC + "s",
+                    evalLatch.await(TIMEOUT_START_AWAIT_SEC, TimeUnit.SECONDS));
 
-            assertTrue("Should receive timeout error within 30s",
-                    errorLatch.await(30, TimeUnit.SECONDS));
+            assertTrue("Should receive timeout error within " + TIMEOUT_ERROR_AWAIT_SEC + "s",
+                    errorLatch.await(TIMEOUT_ERROR_AWAIT_SEC, TimeUnit.SECONDS));
             assertEquals("q-timeout", errorMsg.get().get("id"));
         } finally {
             session.close();
@@ -1163,16 +1170,16 @@ public void onMessage(final String message) {
         // Start a long-running query
         session.getBasicRemote().sendText(
                 "{\"action\":\"eval\",\"id\":\"q-cleanup\"," +
-                "\"query\":\"let $x := for $i in 1 to 999999999 return string($i) return $x\"," +
-                "\"max-execution-time\":30000}");
+                "\"query\":\"" + CANCEL_TEST_QUERY + "\"," +
+                "\"max-execution-time\":" + CANCEL_MAX_EXECUTION_MS + "}");
 
         // Wait for evaluating phase, then abruptly close
         assertTrue("Should reach evaluating phase within 5s",
                 progressLatch.await(5, TimeUnit.SECONDS));
         session.close();
 
-        // Give server time to clean up
-        Thread.sleep(500);
+        // Allow session-close cancellation to finish before later tests reuse the broker pool
+        Thread.sleep(2_000);
 
         // The test passes if no resources leak and no exceptions are thrown.
         // ExistWebServer would fail to shut down if brokers were leaked.