From 17abec7e3e509660cc7bd998db2817afba3c7fb0 Mon Sep 17 00:00:00 2001 From: fabrizzio-dotCMS Date: Sun, 13 Sep 2026 15:36:41 -0600 Subject: [PATCH] refactor(health): scope the health-check fan-out so a timeout cancels it (#34154) runAllHealthChecksAndWait built a List> by hand, joined with allOf(...), and waited with get(timeout). That shape stops WAITING after the timeout. It does not stop the work. The executor is Executors.newScheduledThreadPool(THREAD_POOL_SIZE) -- bounded -- and there are 15 registered checks. So a wedged check kept one of those threads for as long as it stayed wedged, with nothing reporting it, while the method logged "some health checks did not complete" and carried on. That is the failure mode of #37038 in another subsystem: abandoning something that keeps consuming a shared, bounded resource. A StructuredTaskScope makes the block the lifetime. On timeout the subtasks are cancelled, and close() does not return until they are done, so no work escapes the method. Behaviour otherwise unchanged: same timeout arithmetic, same "continue with whatever arrived" contract -- which is why the joiner is awaitAll() and not awaitAllSuccessfulOrThrow(), so one failing check still does not cancel its siblings. waitForOngoingRefreshes is deliberately NOT converted, and says so in a comment: it waits on refreshes started elsewhere and held in a map, and a scope can only wait on subtasks it forked itself. Four unit tests, no container, pinning the difference: allOf-with-timeout leaves all four wedged tasks running; the scope leaves none; a failing subtask does not cancel its siblings; subtasks run on virtual threads. StructuredTaskScope is PREVIEW in Java 25. This compiles only because maven.compiler.enablePreview is already true and the shipped container already runs --enable-preview. Its API has changed in every release since 21, so this stays DO NOT MERGE until it is final. Verified: test-compile -pl :dotcms-core --am passes; 4/4 green. Co-Authored-By: Claude Opus 5 (1M context) --- .../health/service/HealthStateManager.java | 77 +++++---- .../service/HealthCheckCancellationTest.java | 162 ++++++++++++++++++ 2 files changed, 209 insertions(+), 30 deletions(-) create mode 100644 dotCMS/src/test/java/com/dotcms/health/service/HealthCheckCancellationTest.java 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)); + } +}