Merge pull request #6532 from duncdrum/dp-flaky-test-fixes

Fix flaky CI tests: HTTP/2 hangs, broker/txn ordering races, and BrokerPool shutdown safety

Author: line-o

AGENTS.md

@@ -98,8 +98,9 @@ docker run -d --name existdb -p 8080:8080 -p 8443:8443 existdb/existdb:local
 
 ### Known build issues
 
-- Full test suite can hang on flaky infrastructure tests (`MoveResourceTest`, `RenameCollectionTest`). Check with `jstack` and kill if stuck >15 min.
+- Full test suite can hang on flaky infrastructure tests (`RenameCollectionTest`). Check with `jstack` and kill if stuck >15 min.
 - `RenameCollectionTest` "Connection refused" failures are pre-existing and unrelated to XQuery changes.
+- `org.exist.xmldb.concurrent.FragmentsTest` and `org.exist.collections.ConcurrencyTest` are excluded from the surefire run (see `exist-core/pom.xml`). `FragmentsTest` hangs during BrokerPool shutdown even locally; `ConcurrencyTest` can deadlock under concurrent collection access.
 
 ## Parser (ANTLR 2)
 

exist-core/pom.xml

@@ -1216,8 +1216,8 @@ The BaseX Team. The original license statement is also included below.]]></pream
                                  see https://github.com/eXist-db/exist/issues/3685 -->
                         <exclude>org.exist.collections.ConcurrencyTest</exclude>
 
-                        <!-- Passes locally (~54s) but hangs on CI during BrokerPool shutdown,
-                                 causing the entire unit test step to time out.
+                        <!-- Hangs both locally and on CI during BrokerPool shutdown;
+                                 root cause not fully resolved (isShuttingDownOrDown fix was insufficient).
                                  see https://github.com/eXist-db/exist/pull/6153 -->
                         <exclude>org.exist.xmldb.concurrent.FragmentsTest</exclude>
                     </excludes>

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) {
@@ -103,13 +107,21 @@ public void execute(final Session wsSession, final EvalSession evalSession,
             sendProgress(wsSession, msg.id(), EvalProtocol.PHASE_COMPILING, 0,
                     System.currentTimeMillis() - startTime);
 
-            // Set timeout via watchdog
             final XQueryWatchDog watchDog = context.getWatchDog();
+            watchDog.reset(); // before signal: prevents execute() from clearing a concurrent kill() via its internal watchdog.reset()
             if (msg.maxExecutionTime() > 0) {
                 watchDog.setTimeout(msg.maxExecutionTime());
             }
             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,
@@ -120,16 +132,17 @@ public void execute(final Session wsSession, final EvalSession evalSession,
 
                 final Sequence result;
                 try {
-                    result = xquery.execute(broker, compiled, null, new Properties(), true);
+                    result = xquery.execute(broker, compiled, null, new Properties(), false);
                 } 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,23 +160,46 @@ 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;
+                            reportTerminationOrError(wsSession, evalSession, msg, timing, startTime,
+                                    watchDog, ErrorInfo.of(WALL_CLOCK_TIMEOUT_MESSAGE), 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()), 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()
             }
 
         } catch (final EXistException | PermissionDeniedException e) {
@@ -192,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);
@@ -207,16 +253,57 @@ private void reportTerminationOrError(final Session wsSession, final EvalSession
                                           final EvalProtocol.Timing timing, final long startTime,
                                           final XQueryWatchDog watchDog,
                                           final ErrorInfo error) {
-        if (watchDog.isTerminating()) {
+        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() && !watchDog.isTimedOut()) {
             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.killAsTimeout(0);
+        if (terminalResponseSent.compareAndSet(false, true)) {
+            timing.total = System.currentTimeMillis() - startTime;
+            // Use async remote: this callback runs on the shared scheduler thread and must not block
+            // on client I/O — a stalled connection would otherwise freeze every pending timeout.
+            try {
+                wsSession.getAsyncRemote().sendText(
+                        EvalProtocol.errorMessage(msg.id(), null, WALL_CLOCK_TIMEOUT_MESSAGE, 0, 0, timing));
+            } catch (final IOException e) {
+                LOG.debug("Failed to send wall-clock timeout error: {}", e.getMessage());
+            }
+            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.
      */
@@ -273,7 +360,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;
@@ -282,10 +372,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);
+                    reportTerminationOrError(wsSession, evalSession, msg, timing, startTime,
+                            watchDog, ErrorInfo.of(WALL_CLOCK_TIMEOUT_MESSAGE), terminalResponseSent);
                     return;
                 }
 
@@ -304,11 +395,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,
@@ -318,7 +414,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,71 @@
+/*
+ * 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.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+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;
+    static {
+        final ScheduledThreadPoolExecutor ex = new ScheduledThreadPoolExecutor(1, r -> {
+            final Thread t = new Thread(r, "exist-ws-eval-wall-clock-timeout");
+            t.setDaemon(true);
+            return t;
+        });
+        // Remove cancelled futures immediately so they don't retain session/query closures.
+        ex.setRemoveOnCancelPolicy(true);
+        EXECUTOR = ex;
+    }
+
+    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();
+    }
+}

exist-core/src/main/java/org/exist/storage/BrokerPool.java

@@ -1255,6 +1255,9 @@ public DBBroker get(final Optional<Subject> subject) throws EXistException {
         synchronized(this) {
             //Are there any available brokers ?
             if(inactiveBrokers.isEmpty()) {
+                if(isShuttingDownOrDown()) {
+                    throw new EXistException("BrokerPool is not operational (state: " + status.getCurrentState().name() + ")");
+                }
                 //There are no available brokers. If allowed...
                 if(brokersCount < maxBrokers)
                 //... create one
@@ -1263,7 +1266,7 @@ public DBBroker get(final Optional<Subject> subject) throws EXistException {
                 } else
                     //... or wait until there is one available
                     while(inactiveBrokers.isEmpty()) {
-                        if(isShuttingDown()) {
+                        if(isShuttingDownOrDown()) {
                             throw new EXistException("BrokerPool is shutting down");
                         }
                         LOG.debug("waiting for a broker to become available");