[bugfix] Add wall-clock max-execution-time backstop for WebSocket eval

The cooperative watchdog only checks elapsed time in proceed(), so a
starved eval thread on CI could miss the limit entirely. Schedule a
daemon timer when max-execution-time is set, kill the query, and send
a single terminal error if nothing else has responded yet.

Split test responsibilities: unit tests cover proceed() timeout and
WallClockQueryTimeout scheduling; the WebSocket maxExecutionTime test
is a short smoke that relies on the server backstop instead of long
client latch slack.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: duncdrum

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

@@ -45,6 +45,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Executes XQuery expressions for the /ws/eval endpoint with support for
@@ -56,6 +57,9 @@ public final class QueryExecutor {
 
     private static final long PROGRESS_INTERVAL_MS = 500;
 
+    static final String WALL_CLOCK_TIMEOUT_MESSAGE =
+            "The query exceeded the predefined timeout and has been killed.";
+
     private final BrokerPool pool;
 
     public QueryExecutor(final BrokerPool pool) {
@@ -110,6 +114,14 @@ public void execute(final Session wsSession, final EvalSession evalSession,
             }
             evalSession.registerQuery(msg.id(), watchDog);
 
+            final AtomicBoolean terminalResponseSent = new AtomicBoolean(false);
+            WallClockQueryTimeout wallClockTimeout = null;
+            if (msg.maxExecutionTime() > 0) {
+                wallClockTimeout = new WallClockQueryTimeout();
+                wallClockTimeout.schedule(msg.maxExecutionTime(), () -> handleWallClockTimeout(
+                        wsSession, evalSession, msg, timing, startTime, watchDog, terminalResponseSent));
+            }
+
             try {
                 // Evaluate phase
                 sendProgress(wsSession, msg.id(), EvalProtocol.PHASE_EVALUATING, 0,
@@ -124,12 +136,13 @@ public void execute(final Session wsSession, final EvalSession evalSession,
                 } catch (final TerminatedException e) {
                     timing.evaluate = System.currentTimeMillis() - evalStart;
                     reportTerminationOrError(wsSession, evalSession, msg, timing, startTime,
-                            watchDog, ErrorInfo.of(e.getMessage(), e.getLine(), e.getColumn()));
+                            watchDog, ErrorInfo.of(e.getMessage(), e.getLine(), e.getColumn()),
+                            terminalResponseSent);
                     return;
                 } catch (final XPathException e) {
                     timing.evaluate = System.currentTimeMillis() - evalStart;
                     reportTerminationOrError(wsSession, evalSession, msg, timing, startTime,
-                            watchDog, ErrorInfo.of(e));
+                            watchDog, ErrorInfo.of(e), terminalResponseSent);
                     return;
                 }
 
@@ -147,37 +160,43 @@ public void execute(final Session wsSession, final EvalSession evalSession,
                 try {
                     if (msg.stream().enabled() && itemCount > msg.stream().chunkSize()) {
                         streamResults(wsSession, msg.id(), broker, result, outputProperties,
-                                msg.stream().chunkSize(), timing, startTime, watchDog);
+                                msg.stream().chunkSize(), timing, startTime, watchDog,
+                                terminalResponseSent, evalSession, msg);
                     } 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);
+                            sendCancelledIfAbsent(wsSession, msg.id(), evalSession, msg, timing,
+                                    0, terminalResponseSent);
                             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()));
+                                    watchDog, ErrorInfo.of(e.getMessage()), terminalResponseSent);
                             return;
                         }
                         final String serialized = serializeAll(broker, result, outputProperties);
                         timing.serialize = System.currentTimeMillis() - serStart;
                         timing.total = System.currentTimeMillis() - startTime;
 
-                        sendResult(wsSession, msg.id(), 1, serialized, false, timing, itemCount);
+                        if (terminalResponseSent.compareAndSet(false, true)) {
+                            sendResult(wsSession, msg.id(), 1, serialized, false, timing, itemCount);
+                            QueryMonitorBroadcaster.broadcastEvent("completed", msg.id(), user, msg.query(),
+                                    null, itemCount, timing.total);
+                        }
                     }
-                    QueryMonitorBroadcaster.broadcastEvent("completed", msg.id(), user, msg.query(),
-                            null, itemCount, System.currentTimeMillis() - startTime);
                 } catch (final SAXException | XPathException e) {
                     timing.serialize = System.currentTimeMillis() - serStart;
-                    reportError(wsSession, evalSession, msg, timing, startTime, ErrorInfo.of(e.getMessage()));
+                    reportError(wsSession, evalSession, msg, timing, startTime, ErrorInfo.of(e.getMessage()),
+                            terminalResponseSent);
                 }
             } finally {
+                if (wallClockTimeout != null) {
+                    wallClockTimeout.cancel();
+                }
                 evalSession.unregisterQuery(msg.id());
                 context.runCleanupTasks();
                 context.reset();  // needed: resetContext=false skipped this in xquery.execute()
@@ -209,6 +228,16 @@ static ErrorInfo of(final String message) {
     private void reportError(final Session wsSession, final EvalSession evalSession,
                              final EvalProtocol.ClientMessage msg, final EvalProtocol.Timing timing,
                              final long startTime, final ErrorInfo error) {
+        reportError(wsSession, evalSession, msg, timing, startTime, error, null);
+    }
+
+    private void reportError(final Session wsSession, final EvalSession evalSession,
+                             final EvalProtocol.ClientMessage msg, final EvalProtocol.Timing timing,
+                             final long startTime, final ErrorInfo error,
+                             @Nullable final AtomicBoolean terminalGate) {
+        if (terminalGate != null && !terminalGate.compareAndSet(false, true)) {
+            return;
+        }
         timing.total = System.currentTimeMillis() - startTime;
         if (error.xpe() != null) {
             sendError(wsSession, msg.id(), error.xpe(), timing);
@@ -224,16 +253,50 @@ private void reportTerminationOrError(final Session wsSession, final EvalSession
                                           final EvalProtocol.Timing timing, final long startTime,
                                           final XQueryWatchDog watchDog,
                                           final ErrorInfo error) {
+        reportTerminationOrError(wsSession, evalSession, msg, timing, startTime, watchDog, error, null);
+    }
+
+    private void reportTerminationOrError(final Session wsSession, final EvalSession evalSession,
+                                          final EvalProtocol.ClientMessage msg,
+                                          final EvalProtocol.Timing timing, final long startTime,
+                                          final XQueryWatchDog watchDog,
+                                          final ErrorInfo error,
+                                          @Nullable final AtomicBoolean terminalGate) {
         if (watchDog.isTerminating()) {
             timing.total = System.currentTimeMillis() - startTime;
-            sendCancelled(wsSession, msg.id(), 0, timing);
-            QueryMonitorBroadcaster.broadcastEvent("cancelled", msg.id(),
-                    evalSession.getSubject().getName(), msg.query(), null, 0, timing.total);
+            sendCancelledIfAbsent(wsSession, msg.id(), evalSession, msg, timing, 0, terminalGate);
         } else {
-            reportError(wsSession, evalSession, msg, timing, startTime, error);
+            reportError(wsSession, evalSession, msg, timing, startTime, error, terminalGate);
+        }
+    }
+
+    private void handleWallClockTimeout(final Session wsSession, final EvalSession evalSession,
+                                        final EvalProtocol.ClientMessage msg,
+                                        final EvalProtocol.Timing timing, final long startTime,
+                                        final XQueryWatchDog watchDog,
+                                        final AtomicBoolean terminalResponseSent) {
+        watchDog.kill(0);
+        if (terminalResponseSent.compareAndSet(false, true)) {
+            timing.total = System.currentTimeMillis() - startTime;
+            sendError(wsSession, msg.id(), null, WALL_CLOCK_TIMEOUT_MESSAGE, 0, 0, timing);
+            QueryMonitorBroadcaster.broadcastEvent("error", msg.id(),
+                    evalSession.getSubject().getName(), msg.query(), null, 0, timing.total);
         }
     }
 
+    private void sendCancelledIfAbsent(final Session wsSession, final String queryId,
+                                       final EvalSession evalSession,
+                                       final EvalProtocol.ClientMessage msg,
+                                       final EvalProtocol.Timing timing, final long items,
+                                       @Nullable final AtomicBoolean terminalGate) {
+        if (terminalGate != null && !terminalGate.compareAndSet(false, true)) {
+            return;
+        }
+        sendCancelled(wsSession, queryId, items, timing);
+        QueryMonitorBroadcaster.broadcastEvent("cancelled", queryId,
+                evalSession.getSubject().getName(), msg.query(), null, items, timing.total);
+    }
+
     /**
      * Compile-check a query without executing it.
      */
@@ -290,7 +353,10 @@ private void streamResults(final Session wsSession, final String queryId,
                                 final DBBroker broker, final Sequence result,
                                 final Properties outputProperties, final int chunkSize,
                                 final EvalProtocol.Timing timing, final long startTime,
-                                final XQueryWatchDog watchDog) {
+                                final XQueryWatchDog watchDog,
+                                final AtomicBoolean terminalResponseSent,
+                                final EvalSession evalSession,
+                                final EvalProtocol.ClientMessage msg) {
         final long serStart = System.currentTimeMillis();
         final long totalItems = result.getItemCount();
         int chunkNum = 0;
@@ -299,10 +365,11 @@ private void streamResults(final Session wsSession, final String queryId,
         try {
             final SequenceIterator iter = result.iterate();
             while (iter.hasNext()) {
-                if (watchDog.isTerminating()) {
+                if (watchDog.isTerminating() || terminalResponseSent.get()) {
                     timing.serialize = System.currentTimeMillis() - serStart;
                     timing.total = System.currentTimeMillis() - startTime;
-                    sendCancelled(wsSession, queryId, itemsSent, timing);
+                    sendCancelledIfAbsent(wsSession, queryId, evalSession, msg, timing,
+                            itemsSent, terminalResponseSent);
                     return;
                 }
 
@@ -321,11 +388,16 @@ private void streamResults(final Session wsSession, final String queryId,
                 if (!more) {
                     timing.serialize = System.currentTimeMillis() - serStart;
                     timing.total = System.currentTimeMillis() - startTime;
+                    if (terminalResponseSent.compareAndSet(false, true)) {
+                        sendResult(wsSession, queryId, chunkNum, data, false, timing, totalItems);
+                        QueryMonitorBroadcaster.broadcastEvent("completed", queryId,
+                                evalSession.getSubject().getName(), msg.query(),
+                                null, totalItems, timing.total);
+                    }
+                } else {
+                    sendResult(wsSession, queryId, chunkNum, data, true, null, totalItems);
                 }
 
-                sendResult(wsSession, queryId, chunkNum, data, more,
-                        more ? null : timing, totalItems);
-
                 // send progress during streaming
                 if (more && (System.currentTimeMillis() - startTime) % PROGRESS_INTERVAL_MS < 50) {
                     sendProgress(wsSession, queryId, EvalProtocol.PHASE_SERIALIZING,
@@ -335,7 +407,8 @@ private void streamResults(final Session wsSession, final String queryId,
         } catch (final XPathException | SAXException e) {
             timing.serialize = System.currentTimeMillis() - serStart;
             timing.total = System.currentTimeMillis() - startTime;
-            sendError(wsSession, queryId, null, e.getMessage(), 0, 0, timing);
+            reportError(wsSession, evalSession, msg, timing, startTime, ErrorInfo.of(e.getMessage()),
+                    terminalResponseSent);
         }
     }
 

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

@@ -0,0 +1,65 @@
+/*
+ * eXist-db Open Source Native XML Database
+ * Copyright (C) 2001 The eXist-db Authors
+ *
+ * info@exist-db.org
+ * http://www.exist-db.org
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+package org.exist.http.ws;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Wall-clock backstop for WebSocket {@code max-execution-time}.
+ *
+ * <p>The XQuery watchdog only checks elapsed time inside {@code proceed()}. Under CPU
+ * pressure the eval thread may not run for long stretches, so a scheduled task guarantees
+ * the client receives a terminal response when the limit expires.</p>
+ */
+final class WallClockQueryTimeout {
+
+    private static final ScheduledExecutorService EXECUTOR = Executors.newSingleThreadScheduledExecutor(r -> {
+        final Thread t = new Thread(r, "exist-ws-eval-wall-clock-timeout");
+        t.setDaemon(true);
+        return t;
+    });
+
+    private final AtomicBoolean triggered = new AtomicBoolean(false);
+    private volatile ScheduledFuture<?> future;
+
+    void schedule(final long delayMs, final Runnable onTimeout) {
+        future = EXECUTOR.schedule(() -> {
+            if (triggered.compareAndSet(false, true)) {
+                onTimeout.run();
+            }
+        }, delayMs, TimeUnit.MILLISECONDS);
+    }
+
+    void cancel() {
+        if (future != null) {
+            future.cancel(false);
+        }
+    }
+
+    boolean wasTriggered() {
+        return triggered.get();
+    }
+}