diff --git a/dotCMS/src/main/java/com/dotcms/health/service/HealthStateManager.java b/dotCMS/src/main/java/com/dotcms/health/service/HealthStateManager.java index ce13c99a6cfa..7b45d5728f34 100644 --- a/dotCMS/src/main/java/com/dotcms/health/service/HealthStateManager.java +++ b/dotCMS/src/main/java/com/dotcms/health/service/HealthStateManager.java @@ -19,7 +19,10 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.StructuredTaskScope; +import java.util.concurrent.StructuredTaskScope.Joiner; import java.util.concurrent.TimeUnit; +import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -597,38 +600,45 @@ private void runAllHealthChecksAsync() { * This ensures cache updates happen with fresh results, not stale ones */ private void runAllHealthChecksAndWait() { - // Create futures for all health checks to track completion - List> futures = new ArrayList<>(); - - for (HealthCheck healthCheck : allHealthChecks) { - CompletableFuture future = CompletableFuture.runAsync(() -> - runSingleHealthCheckBlocking(healthCheck), executor); - futures.add(future); - } - - // Wait for all health checks to complete (with timeout to prevent hanging) - try { - CompletableFuture allChecks = CompletableFuture.allOf( - futures.toArray(new CompletableFuture[0])); - - // Calculate timeout based on the longest individual health check timeout + buffer - long maxIndividualTimeout = allHealthChecks.stream() + + // Timeout: the longest individual check plus a buffer, honouring any override. Unchanged. + final long maxIndividualTimeout = allHealthChecks.stream() .mapToLong(this::getHealthCheckTimeoutMs) .max() - .orElse(30000L); // Default 30 seconds if no checks - - // Add buffer time for execution overhead - long timeoutMs = maxIndividualTimeout + 5000L; // Max individual timeout + 5 second buffer - - // Also respect any configured override - timeoutMs = Math.max(timeoutMs, Config.getLongProperty("health.force.refresh.timeout-ms", timeoutMs)); - - allChecks.get(timeoutMs, TimeUnit.MILLISECONDS); - - Logger.info(this, String.format("All health checks completed in forceRefresh (timeout: %dms)", timeoutMs)); - } catch (Exception e) { - Logger.warn(this, "Some health checks did not complete within timeout during forceRefresh: " + e.getMessage()); - // Continue anyway - we'll use whatever results we have + .orElse(30000L); + final long timeoutMs = Math.max(maxIndividualTimeout + 5000L, + Config.getLongProperty("health.force.refresh.timeout-ms", maxIndividualTimeout + 5000L)); + + // Every check is a subtask of this block. Two things follow from that, and neither was + // true of the CompletableFuture.allOf(...).get(timeout) this replaces: + // + // 1. On timeout the subtasks are CANCELLED, not merely stopped being waited for. The + // old shape left a hung check running on this bounded pool forever, holding one of + // HealthCheckConfig.THREAD_POOL_SIZE threads with nothing reporting it. + // 2. close() does not return until they are done, so no work escapes this method. + // + // awaitAll() is deliberate: a failing check must not cancel the others. The previous + // behaviour -- carry on with whatever results arrived -- is preserved exactly. + try (var scope = StructuredTaskScope.open(Joiner.awaitAll(), + cfg -> cfg.withTimeout(Duration.ofMillis(timeoutMs)).withName("health-checks"))) { + + allHealthChecks.forEach(healthCheck -> + scope.fork(() -> { + runSingleHealthCheckBlocking(healthCheck); + return null; + })); + + scope.join(); + Logger.info(this, String.format( + "All health checks completed in forceRefresh (timeout: %dms)", timeoutMs)); + + } catch (final StructuredTaskScope.TimeoutException timedOut) { + Logger.warn(this, String.format( + "Health checks did not complete within %dms during forceRefresh; the ones still " + + "running were cancelled. Continuing with the results that arrived.", timeoutMs)); + } catch (final InterruptedException interrupted) { + Thread.currentThread().interrupt(); + Logger.warn(this, "Interrupted while waiting for health checks during forceRefresh"); } } @@ -830,6 +840,13 @@ private boolean isReadinessCheck(String healthCheckName) { * @param timeoutMs maximum time to wait in milliseconds * @return true if all refreshes completed, false if timeout occurred */ + /* + * Deliberately left on CompletableFuture: this method waits on refreshes that were started + * ELSEWHERE and are held in the ongoingRefreshes map. A StructuredTaskScope can only wait on + * subtasks it forked itself -- that is the whole point of the lifetime being the block -- so + * there is nothing here for it to adopt. Converting this would mean moving where the refreshes + * are started, which is a different change. + */ public boolean waitForOngoingRefreshes(List checkNames, long timeoutMs) { List> refreshesToWaitFor = new ArrayList<>(); diff --git a/dotCMS/src/test/java/com/dotcms/health/service/HealthCheckCancellationTest.java b/dotCMS/src/test/java/com/dotcms/health/service/HealthCheckCancellationTest.java new file mode 100644 index 000000000000..9e1d5f5e8292 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/health/service/HealthCheckCancellationTest.java @@ -0,0 +1,162 @@ +package com.dotcms.health.service; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.StructuredTaskScope; +import java.util.concurrent.StructuredTaskScope.Joiner; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; + +/** + * Documents why {@code runAllHealthChecksAndWait} moved from {@code CompletableFuture.allOf(...) + * .get(timeout)} to a {@link StructuredTaskScope}. + * + *

Both shapes stop waiting after the timeout. Only one of them stops the work. The old shape + * left a hung check running on {@code HealthCheckConfig.THREAD_POOL_SIZE}'s bounded pool with + * nothing reporting it, which is the same failure mode as issue #37038 in a different subsystem: + * abandoning something that keeps consuming a shared, bounded resource.

+ * + *

These tests exercise the concurrency shapes directly rather than the manager, so they need no + * registered checks and no container.

+ */ +public class HealthCheckCancellationTest { + + /** How long a wedged check pretends to take — far longer than any timeout used here. */ + private static final long WEDGED_MS = 10_000L; + + private static final long TIMEOUT_MS = 300L; + + /** Counts tasks that have started and not yet returned. */ + private final AtomicInteger inFlight = new AtomicInteger(); + + private void wedgedCheck() { + inFlight.incrementAndGet(); + try { + Thread.sleep(WEDGED_MS); + } catch (final InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } finally { + inFlight.decrementAndGet(); + } + } + + /** + * Method to test: the {@code CompletableFuture.allOf(...).get(timeout)} shape this replaced. + * Given scenario: four wedged checks and a timeout far shorter than they take. + * Expected result: the timeout fires and every task is STILL RUNNING. get() stops waiting; it + * does not cancel. This is the behaviour the change removes. + */ + @Test + public void test_allOfWithTimeout_doesNotCancelAnything() throws Exception { + + final ExecutorService pool = Executors.newFixedThreadPool(4); + try { + final List> futures = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + futures.add(CompletableFuture.runAsync(this::wedgedCheck, pool)); + } + + assertThrows(java.util.concurrent.TimeoutException.class, + () -> CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) + .get(TIMEOUT_MS, TimeUnit.MILLISECONDS)); + + Thread.sleep(200); + assertEquals("the timeout abandoned them; they are still holding pool threads", + 4, inFlight.get()); + } finally { + pool.shutdownNow(); + } + } + + /** + * Method to test: the {@link StructuredTaskScope} shape now in use. + * Given scenario: the same four wedged checks and the same timeout. + * Expected result: the timeout fires and, by the time the block has exited, nothing is running. + * close() cancels the subtasks and waits for them, so no work escapes the method. + */ + @Test + public void test_scopeWithTimeout_cancelsAndWaits() { + + assertThrows(StructuredTaskScope.TimeoutException.class, () -> { + try (var scope = StructuredTaskScope.open(Joiner.awaitAll(), + cfg -> cfg.withTimeout(Duration.ofMillis(TIMEOUT_MS)))) { + + for (int i = 0; i < 4; i++) { + scope.fork(() -> { + wedgedCheck(); + return null; + }); + } + scope.join(); + } + }); + + assertEquals("close() cancelled them and did not return until they were done", + 0, inFlight.get()); + } + + /** + * Method to test: the {@link Joiner#awaitAll()} choice. + * Given scenario: one check throws while the others succeed. + * Expected result: the failure does not cancel its siblings and does not propagate. This + * preserves the previous contract — carry on with whatever results arrived — which + * {@code awaitAllSuccessfulOrThrow()} would have changed. + */ + @Test + public void test_awaitAll_oneFailingCheckDoesNotCancelTheRest() throws Exception { + + final AtomicInteger completed = new AtomicInteger(); + + try (var scope = StructuredTaskScope.open(Joiner.awaitAll(), + cfg -> cfg.withTimeout(Duration.ofSeconds(5)))) { + + scope.fork(() -> { + throw new IllegalStateException("one check blew up"); + }); + for (int i = 0; i < 3; i++) { + scope.fork(() -> { + Thread.sleep(50); + completed.incrementAndGet(); + return null; + }); + } + scope.join(); + } + + assertEquals("the three healthy checks still ran to completion", 3, completed.get()); + } + + /** + * Method to test: {@link StructuredTaskScope#fork(java.util.concurrent.Callable)}. + * Given scenario: any subtask. + * Expected result: it runs on a virtual thread. Health checks block on sockets — databases, + * search endpoints, HTTP probes — which is the case virtual threads exist for. + */ + @Test + public void test_subtasksRunOnVirtualThreads() throws Exception { + + final List virtual = java.util.Collections.synchronizedList(new ArrayList<>()); + + try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) { + for (int i = 0; i < 4; i++) { + scope.fork(() -> { + virtual.add(Thread.currentThread().isVirtual()); + return null; + }); + } + scope.join(); + } + + assertEquals(4, virtual.size()); + assertTrue("every subtask ran on a virtual thread", virtual.stream().allMatch(v -> v)); + } +}