From 81db14678d07d20aa2d0a859ed1d7fceeab9bfea Mon Sep 17 00:00:00 2001 From: Will Ezell Date: Mon, 3 Aug 2026 12:23:47 -0400 Subject: [PATCH 1/8] fix(clustering): surface silent cache-transport failures (#36803) Cache invalidations were dropped with no log, no metric and no health signal when the pub/sub cache transport was not initialized, and a failing cluster rewire was logged and forgotten while KNOWN_SERVERS was updated as if it had succeeded (issue #36544 incident). - PubSubCacheTransport.send(): count dropped invalidations and WARN, rate-limited (CACHE_TRANSPORT_DROP_WARN_INTERVAL_MILLIS, default 30s) - PubSubCacheTransport.init(): idempotent - a rewire on a healthy transport no longer tears down/rebuilds the pub/sub listener - ClusterFactory.addMeToCacheIfNeeded(): report success/failure; only update KNOWN_SERVERS on success so failed rewires are retried; track consecutive failures in a counter exposed via getRewireFailures() - New cache-transport health check (MONITOR_MODE by default so it never fails readiness unless an operator opts in via health.check.cache-transport.mode=PRODUCTION - avoids cold-start deadlock) - Micrometer gauges: dotcms.cache.transport.invalidations.dropped, dotcms.cache.transport.initialized, dotcms.cache.transport.rewire.failures Fixes #36803 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01XFuXwYdNsb7irpMEeSGTyj --- .../enterprise/cluster/ClusterFactory.java | 34 ++++- .../cache/transport/PubSubCacheTransport.java | 38 ++++- .../checks/cdi/CacheTransportHealthCheck.java | 137 ++++++++++++++++++ .../providers/CoreHealthCheckProvider.java | 2 + .../dotcms/metrics/binders/CacheMetrics.java | 33 +++++ .../cache/transport/CacheTransport.java | 9 ++ .../transport/PubSubCacheTransportTest.java | 91 ++++++++++++ 7 files changed, 335 insertions(+), 9 deletions(-) create mode 100644 dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java create mode 100644 dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java diff --git a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java index f8c6912ef7a3..b32e3199f558 100644 --- a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java +++ b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java @@ -56,6 +56,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import org.elasticsearch.common.settings.Settings.Builder; @@ -361,14 +362,17 @@ public static synchronized void rewireClusterIfNeeded() { public static void rewireCluster() throws Exception { if(clusterReady()) { - addMeToCacheIfNeeded(); - KNOWN_SERVERS = APILocator.getServerAPI().getAliveServers(); + if (addMeToCacheIfNeeded()) { + // only remember the alive-server set when the rewire actually succeeded, + // otherwise a failed transport init would never be retried + KNOWN_SERVERS = APILocator.getServerAPI().getAliveServers(); + } }else { Logger.info(ClusterFactory.class, "Cluster not yet active. Not rewiring"); } } - private static void addMeToCacheIfNeeded() throws DotDataException { + private static boolean addMeToCacheIfNeeded() throws DotDataException { if(isEnterprise()) { final Server localServer = APILocator.getServerAPI().getOrCreateMyServer(); @@ -378,16 +382,32 @@ private static void addMeToCacheIfNeeded() throws DotDataException { getImplementationObject()).setCluster(localServer); ((ChainableCacheAdministratorImpl) CacheLocator.getCacheAdministrator(). getImplementationObject()).testCluster(); + REWIRE_FAILURES.set(0); + return true; } catch (Exception e) { - Logger.error(ClusterFactory.class, e.getMessage(), e); + final long failures = REWIRE_FAILURES.incrementAndGet(); + Logger.error(ClusterFactory.class, "Unable to (re)initialize the cluster cache transport" + + " (consecutive failures: " + failures + + "). Cache invalidations are NOT reaching other nodes: " + e.getMessage(), e); + return false; } - - } else { - CacheLocator.getCacheAdministrator().getTransport().shutdown(); + CacheLocator.getCacheAdministrator().getTransport().shutdown(); + return true; } } + + /** + * Consecutive cluster cache-transport rewire failures. Reset to zero on the first + * successful rewire. Surfaced by the cache-transport health check and metrics so a + * persistently failing rewire is visible instead of being logged and forgotten. + */ + private static final AtomicLong REWIRE_FAILURES = new AtomicLong(0); + + public static long getRewireFailures() { + return REWIRE_FAILURES.get(); + } private static boolean isEnterprise() { diff --git a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java index 6abd9521a38a..c527e6f05f1b 100644 --- a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java +++ b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java @@ -3,6 +3,7 @@ import java.io.Serializable; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import com.dotcms.cache.transport.CacheTransportTopic.CacheEventType; import com.dotcms.cluster.bean.Server; import com.dotcms.dotpubsub.DotPubSubEvent; @@ -14,7 +15,9 @@ import com.dotmarketing.business.CacheLocator; import com.dotmarketing.business.cache.transport.CacheTransport; import com.dotmarketing.business.cache.transport.CacheTransportException; +import com.dotmarketing.util.Config; import com.dotmarketing.util.Logger; +import com.google.common.annotations.VisibleForTesting; import io.vavr.control.Try; /** @@ -31,20 +34,37 @@ public class PubSubCacheTransport implements CacheTransport { final AtomicBoolean initialized = new AtomicBoolean(false); + final AtomicLong droppedMessages = new AtomicLong(0); + + private final AtomicLong lastDropWarnAt = new AtomicLong(0); + + private static final long DROP_WARN_INTERVAL_MILLIS = + Config.getLongProperty("CACHE_TRANSPORT_DROP_WARN_INTERVAL_MILLIS", 30000); + @Override public boolean requiresAutowiring() { return false; } public PubSubCacheTransport() { - this.pubsub = DotPubSubProviderLocator.provider.get(); - this.topic = new CacheTransportTopic(); + this(DotPubSubProviderLocator.provider.get(), new CacheTransportTopic()); + } + + @VisibleForTesting + PubSubCacheTransport(final DotPubSubProvider pubsub, final CacheTransportTopic topic) { + this.pubsub = pubsub; + this.topic = topic; Logger.debug(this.getClass(), "PubSubCacheTransport"); } @Override public void init(final Server localServer) throws CacheTransportException { + if (this.initialized.get()) { + Logger.debug(this.getClass(), "PubSubCacheTransport already initialized, skipping re-init"); + return; + } + Logger.info(this.getClass(), "initing PubSubCacheTransport"); this.pubsub.start(); this.pubsub.subscribe(topic); @@ -56,6 +76,15 @@ public void init(final Server localServer) throws CacheTransportException { @Override public void send(final String message) throws CacheTransportException { if (!this.initialized.get()) { + final long dropped = this.droppedMessages.incrementAndGet(); + final long now = System.currentTimeMillis(); + final long lastWarn = this.lastDropWarnAt.get(); + if (now - lastWarn > DROP_WARN_INTERVAL_MILLIS + && this.lastDropWarnAt.compareAndSet(lastWarn, now)) { + Logger.warn(this.getClass(), + "Cache transport is not initialized - dropping cluster cache invalidations. " + + "Other nodes may serve stale content. Total dropped: " + dropped); + } return; } @@ -125,6 +154,11 @@ public boolean shouldReinit() { return !initialized.get(); } + @Override + public long getDroppedMessages() { + return droppedMessages.get(); + } + @Override public CacheTransportInfo getInfo() { diff --git a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java new file mode 100644 index 000000000000..211d4bd30337 --- /dev/null +++ b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java @@ -0,0 +1,137 @@ +package com.dotcms.health.checks.cdi; + +import com.dotcms.enterprise.cluster.ClusterFactory; +import com.dotcms.health.config.HealthCheckConfig.HealthCheckMode; +import com.dotcms.health.model.HealthStatus; +import com.dotcms.health.util.HealthCheckBase; +import com.dotmarketing.business.CacheLocator; +import com.dotmarketing.business.ChainableCacheAdministratorImpl; +import com.dotmarketing.business.cache.transport.CacheTransport; +import com.dotmarketing.business.cache.transport.NullTransport; +import java.util.HashMap; +import java.util.Map; +import javax.enterprise.context.ApplicationScoped; + +/** + * Health check for the cluster cache-invalidation transport (pub/sub). + * + * A node whose cache transport is not initialized silently drops every cache invalidation it + * tries to send, so other nodes serve stale content while all standard probes report healthy + * (see issue #36803 / incident #36544). This check surfaces that state: + * + * - transport not initialized -> unhealthy + * - persistent cluster rewire failures -> unhealthy + * - dropped-invalidation count exposed in structured data for monitoring + * + * Single-node / community installs use {@link NullTransport} (or no transport) and always + * report healthy. + * + * Readiness: by default this check runs in MONITOR_MODE, so it reports degradation but never + * fails readiness probes — a node that cannot send invalidations can still serve traffic, and + * gating readiness on the transport could deadlock a cold cluster start where the transport + * initializes after traffic begins. Operators who prefer to drain such nodes can opt in with + * {@code health.check.cache-transport.mode=PRODUCTION}. + * + * Configuration Properties: + * - health.check.cache-transport.mode = Safety mode (PRODUCTION, MONITOR_MODE, DISABLED) + */ +@ApplicationScoped +public class CacheTransportHealthCheck extends HealthCheckBase { + + @Override + protected CheckResult performCheck() throws Exception { + + if (isShutdownInProgress()) { + return new CheckResult(true, 0L, "Cache transport check skipped during shutdown"); + } + + final CacheTransport transport = getTransport(); + + if (transport == null || transport instanceof NullTransport) { + return new CheckResult(true, 0L, "No cluster cache transport in use (single node / community)"); + } + + final long rewireFailures = ClusterFactory.getRewireFailures(); + final long dropped = transport.getDroppedMessages(); + + if (!transport.isInitialized()) { + return new CheckResult(false, 0L, + "Cache transport is NOT initialized - cluster cache invalidations are being dropped" + + " (dropped so far: " + dropped + + ", consecutive rewire failures: " + rewireFailures + ")"); + } + + if (rewireFailures > 0) { + return new CheckResult(false, 0L, + "Cluster rewire is failing (consecutive failures: " + rewireFailures + + ") - cache transport may be stale"); + } + + return new CheckResult(true, 0L, + "Cache transport initialized (" + transport.getClass().getSimpleName() + + ", dropped invalidations: " + dropped + ")"); + } + + private CacheTransport getTransport() { + try { + return ((ChainableCacheAdministratorImpl) CacheLocator.getCacheAdministrator() + .getImplementationObject()).getTransport(); + } catch (Exception e) { + // cache layer not up yet - nothing to report on + return null; + } + } + + @Override + public String getName() { + return "cache-transport"; + } + + @Override + protected HealthCheckMode getDefaultMode() { + return HealthCheckMode.MONITOR_MODE; + } + + @Override + public int getOrder() { + return 35; // right after the cache check + } + + /** + * Never liveness - a broken transport must not restart pods (that made #36544 worse) + */ + @Override + public boolean isLivenessCheck() { + return false; + } + + /** + * Participates in readiness reporting, but the default MONITOR_MODE means it will not + * fail probes unless an operator opts in to PRODUCTION mode. + */ + @Override + public boolean isReadinessCheck() { + return getMode() != HealthCheckMode.DISABLED; + } + + @Override + public String getDescription() { + return String.format( + "Verifies the cluster cache-invalidation transport is initialized and not dropping messages (Mode: %s)", + getMode().name()); + } + + @Override + protected Map buildStructuredData(CheckResult result, HealthStatus originalStatus, + HealthStatus finalStatus, HealthCheckMode mode) { + final Map data = new HashMap<>(); + final CacheTransport transport = getTransport(); + if (transport != null) { + data.put("transport", transport.getClass().getSimpleName()); + data.put("initialized", transport.isInitialized()); + data.put("droppedInvalidations", transport.getDroppedMessages()); + } + data.put("rewireFailures", ClusterFactory.getRewireFailures()); + return data; + } +} diff --git a/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java b/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java index 1ad6080490af..8b47f2d116b7 100644 --- a/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java +++ b/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java @@ -3,6 +3,7 @@ import com.dotcms.health.api.HealthCheck; import com.dotcms.health.api.HealthCheckProvider; import com.dotcms.health.checks.cdi.CacheHealthCheck; +import com.dotcms.health.checks.cdi.CacheTransportHealthCheck; import com.dotcms.health.checks.cdi.DatabaseHealthCheck; import com.dotcms.health.checks.cdi.ElasticsearchHealthCheck; import com.dotcms.health.checks.cdi.VelocityHealthCheck; @@ -25,6 +26,7 @@ public List getHealthChecks() { return Arrays.asList( new DatabaseHealthCheck(), new CacheHealthCheck(), + new CacheTransportHealthCheck(), new ElasticsearchHealthCheck(), new VelocityHealthCheck() // Additional dependency health checks can be added here diff --git a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java index 8520d852750e..5af2c99bf254 100644 --- a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java +++ b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java @@ -40,6 +40,7 @@ public void bindTo(MeterRegistry registry) { if (cacheAdmin != null) { registerProviderLevelMetrics(registry, cacheAdmin); registerRegionLevelMetrics(registry, cacheAdmin); + registerTransportMetrics(registry); } Logger.info(this, "Comprehensive cache metrics registered successfully"); @@ -49,6 +50,38 @@ public void bindTo(MeterRegistry registry) { } } + /** + * Register cluster cache-transport metrics (issue #36803): silent invalidation drops + * and persistent rewire failures were previously invisible to monitoring. + */ + private void registerTransportMetrics(MeterRegistry registry) { + + Gauge.builder(METRIC_PREFIX + ".transport.invalidations.dropped", this, + m -> getTransport().map(t -> (double) t.getDroppedMessages()).orElse(0.0)) + .description("Cluster cache invalidations dropped because the transport was not initialized") + .register(registry); + + Gauge.builder(METRIC_PREFIX + ".transport.initialized", this, + m -> getTransport().map(t -> t.isInitialized() ? 1.0 : 0.0).orElse(1.0)) + .description("Whether the cluster cache transport is initialized (1) or dropping invalidations (0)") + .register(registry); + + Gauge.builder(METRIC_PREFIX + ".transport.rewire.failures", this, + m -> (double) com.dotcms.enterprise.cluster.ClusterFactory.getRewireFailures()) + .description("Consecutive cluster cache-transport rewire failures (reset on success)") + .register(registry); + } + + private java.util.Optional getTransport() { + try { + return java.util.Optional.ofNullable( + ((com.dotmarketing.business.ChainableCacheAdministratorImpl) CacheLocator + .getCacheAdministrator().getImplementationObject()).getTransport()); + } catch (Exception e) { + return java.util.Optional.empty(); + } + } + /** * Register provider-level aggregate metrics. */ diff --git a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java index 50ceaa210d56..3d5d3c4ebba5 100644 --- a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java +++ b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java @@ -56,6 +56,15 @@ public interface CacheTransport { default boolean requiresAutowiring() { return true; } + + /** + * Number of cache invalidation messages this transport has dropped, e.g. because it was + * asked to send before it was initialized. Used by health checks and metrics to surface + * silent invalidation loss in a cluster. + */ + default long getDroppedMessages() { + return 0; + } public interface CacheTransportInfo extends Serializable { diff --git a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java new file mode 100644 index 000000000000..53d5c460066b --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java @@ -0,0 +1,91 @@ +package com.dotcms.cache.transport; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import com.dotcms.dotpubsub.NullDotPubSubProvider; +import org.junit.Test; + +/** + * Unit tests for the silent-invalidation-drop fixes from issue #36803. + */ +public class PubSubCacheTransportTest { + + private PubSubCacheTransport newTransport(final NullDotPubSubProvider provider) { + return new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + } + + /** + * send() before init() must not publish, but must count the dropped message + * instead of discarding it silently. + */ + @Test + public void test_send_before_init_counts_dropped_messages() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + assertEquals(0, transport.getDroppedMessages()); + + transport.send("0:testGroup"); + transport.send("0:testGroup"); + + assertEquals(2, transport.getDroppedMessages()); + assertNull("nothing may be published while uninitialized", provider.lastEventOut()); + assertFalse(transport.isInitialized()); + } + + /** + * After init(), send() publishes and the dropped counter stops growing. + */ + @Test + public void test_send_after_init_publishes() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.send("dropped-before-init"); + transport.init(null); + assertTrue(transport.isInitialized()); + + transport.send("0:testGroup"); + + assertNotNull(provider.lastEventOut()); + assertEquals(1, transport.getDroppedMessages()); + } + + /** + * init() is idempotent: calling it on an already-initialized transport is a no-op + * rather than a second pubsub start/subscribe cycle (the rewire loop from #36544). + */ + @Test + public void test_init_is_idempotent() throws Exception { + final CountingProvider provider = new CountingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + transport.init(null); + transport.init(null); + transport.init(null); + + assertEquals(1, provider.starts); + assertTrue(transport.isInitialized()); + + // after shutdown, init() must run again + transport.shutdown(); + assertFalse(transport.isInitialized()); + transport.init(null); + assertEquals(2, provider.starts); + } + + private static class CountingProvider extends NullDotPubSubProvider { + int starts = 0; + + @Override + public com.dotcms.dotpubsub.DotPubSubProvider start() { + starts++; + return this; + } + } +} From 37cfacbe681dc13b64dcdd273e9afcf4b572460b Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Mon, 10 Aug 2026 19:10:54 -0600 Subject: [PATCH 2/8] fix(clustering): remove false positives from cache-transport monitoring (#36803) Review follow-ups on the transport health check and metrics. CacheMetrics had no NullTransport guard, so a node with no real transport published transport.initialized=0 -- NullTransport.isInitialized() is false once it has been shut down -- and alerted as if it were dropping invalidations, while CacheTransportHealthCheck reported the same node healthy. Both now agree via a single activeTransport() helper. Both call sites resolved the transport by casting getImplementationObject() to ChainableCacheAdministratorImpl. getTransport() is on the DotCacheAdministrator interface and CommitListenerCacheWrapper delegates it, whereas the cast throws ClassCastException for any other administrator (NullCacheAdministrator.getImplementationObject() returns itself, which is the unit-test path) -- swallowed into a null that happened to produce the right answer. Both now use the interface method, as ClusterResource does. The cast in addMeToCacheIfNeeded stays: setCluster()/testCluster() are not on the interface. A single rewire failure no longer reports DOWN. testCluster() can throw on a momentary database hiccup while the transport stays initialized and invalidations keep flowing, so the check now tolerates health.check.cache-transport.rewire-failure-threshold (default 3) consecutive failures and always reports the count. rewireClusterIfNeeded() now retries whenever REWIRE_FAILURES > 0. The membership comparison alone only fires when the alive-server set changes, so a failure followed by membership settling back to KNOWN_SERVERS was never retried -- the counter could never return to zero and the check would report DOWN indefinitely. performCheck() and buildStructuredData() shared one immutable snapshot instead of each resolving the transport and re-reading the counters. Two independent reads let one health response contradict itself: a message saying the transport is initialized next to structured data saying it is not. Refs: #36803 Co-Authored-By: Claude Opus 5 (1M context) --- .../enterprise/cluster/ClusterFactory.java | 22 ++- .../checks/cdi/CacheTransportHealthCheck.java | 159 ++++++++++++++---- .../dotcms/metrics/binders/CacheMetrics.java | 35 +++- 3 files changed, 173 insertions(+), 43 deletions(-) diff --git a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java index b32e3199f558..22ba7522f948 100644 --- a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java +++ b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java @@ -67,6 +67,13 @@ public class ClusterFactory { private static boolean CLUSTER_INITED=false; private static List KNOWN_SERVERS=Collections.EMPTY_LIST; + /** + * Consecutive cluster cache-transport rewire failures. Reset to zero on the first + * successful rewire. Surfaced by the cache-transport health check and metrics so a + * persistently failing rewire is visible instead of being logged and forgotten. + */ + private static final AtomicLong REWIRE_FAILURES = new AtomicLong(0); + private static final String VERIFY_REQUIRED_TABLES = "SELECT dc.cluster_id, dc.cluster_salt, cs.server_id, cs.cluster_id, cs.name, " + "cs.ip_address, cs.host, cs.cache_port, cs.es_transport_tcp_port, " @@ -344,7 +351,12 @@ public static synchronized void rewireClusterIfNeeded() { try{ List aliveServers = APILocator.getServerAPI().getAliveServers(); - if (!aliveServers.equals(KNOWN_SERVERS) || !aliveServers + // A pending failure must always be retried. The membership comparison below only + // fires when the alive-server set changes, so without this a rewire that failed + // while membership then settled back to KNOWN_SERVERS would never be attempted + // again: the transport would stay broken and REWIRE_FAILURES could never return to + // zero, leaving the health check reporting DOWN indefinitely. + if (REWIRE_FAILURES.get() > 0 || !aliveServers.equals(KNOWN_SERVERS) || !aliveServers .contains(APILocator.getServerAPI().getCurrentServer()) ) { rewireCluster(); @@ -399,12 +411,10 @@ private static boolean addMeToCacheIfNeeded() throws DotDataException { } /** - * Consecutive cluster cache-transport rewire failures. Reset to zero on the first - * successful rewire. Surfaced by the cache-transport health check and metrics so a - * persistently failing rewire is visible instead of being logged and forgotten. + * Consecutive cluster cache-transport rewire failures, zero when the last rewire succeeded. + * + * @see #REWIRE_FAILURES */ - private static final AtomicLong REWIRE_FAILURES = new AtomicLong(0); - public static long getRewireFailures() { return REWIRE_FAILURES.get(); } diff --git a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java index 211d4bd30337..1a93148f450a 100644 --- a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java +++ b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java @@ -5,7 +5,6 @@ import com.dotcms.health.model.HealthStatus; import com.dotcms.health.util.HealthCheckBase; import com.dotmarketing.business.CacheLocator; -import com.dotmarketing.business.ChainableCacheAdministratorImpl; import com.dotmarketing.business.cache.transport.CacheTransport; import com.dotmarketing.business.cache.transport.NullTransport; import java.util.HashMap; @@ -20,62 +19,126 @@ * (see issue #36803 / incident #36544). This check surfaces that state: * * - transport not initialized -> unhealthy - * - persistent cluster rewire failures -> unhealthy + * - cluster rewire failing repeatedly -> unhealthy (see {@link #DEFAULT_REWIRE_FAILURE_THRESHOLD}) * - dropped-invalidation count exposed in structured data for monitoring * - * Single-node / community installs use {@link NullTransport} (or no transport) and always - * report healthy. + * Nodes with no real transport -- {@link NullTransport} via + * {@code CACHE_INVALIDATION_TRANSPORT_CLASS}, or no resolvable cache layer at all -- always + * report healthy, since there is nothing that could be dropping invalidations. * * Readiness: by default this check runs in MONITOR_MODE, so it reports degradation but never - * fails readiness probes — a node that cannot send invalidations can still serve traffic, and + * fails readiness probes -- a node that cannot send invalidations can still serve traffic, and * gating readiness on the transport could deadlock a cold cluster start where the transport * initializes after traffic begins. Operators who prefer to drain such nodes can opt in with * {@code health.check.cache-transport.mode=PRODUCTION}. * * Configuration Properties: * - health.check.cache-transport.mode = Safety mode (PRODUCTION, MONITOR_MODE, DISABLED) + * - health.check.cache-transport.rewire-failure-threshold = consecutive rewire failures + * tolerated before reporting unhealthy (default 3) */ @ApplicationScoped public class CacheTransportHealthCheck extends HealthCheckBase { + /** + * Consecutive rewire failures tolerated before this check reports unhealthy. + * + * A single failure is usually a transient blip: {@code testCluster()} can throw on a + * momentary database hiccup while the transport itself stays initialized and invalidations + * keep flowing. Alarming on the first failure would report DOWN for a working node purely + * from a counter that has not been reset yet. {@code ClusterFactory.rewireClusterIfNeeded()} + * retries on every server heartbeat (60s by default) and zeroes the counter on success, so + * reaching this threshold means the rewire has been failing for minutes, not milliseconds. + */ + private static final int DEFAULT_REWIRE_FAILURE_THRESHOLD = 3; + + /** + * Transport state resolved once per {@code check()} invocation. + * + * {@code HealthCheckBase.check()} calls {@link #performCheck()} and then + * {@link #buildStructuredData} sequentially on the same thread, so the snapshot is handed + * between them here instead of resolving the transport and re-reading its counters twice. + * Beyond the wasted work, two independent reads let a single health response contradict + * itself -- a message saying the transport is initialized alongside structured data saying + * it is not, because the state moved in between. A ThreadLocal (rather than a field) keeps + * concurrent liveness/readiness polls of this {@code @ApplicationScoped} bean isolated, and + * {@link #buildStructuredData} always clears it. + */ + private static final ThreadLocal CURRENT_SNAPSHOT = new ThreadLocal<>(); + @Override protected CheckResult performCheck() throws Exception { if (isShutdownInProgress()) { + CURRENT_SNAPSHOT.set(TransportSnapshot.unavailable(0L)); return new CheckResult(true, 0L, "Cache transport check skipped during shutdown"); } - final CacheTransport transport = getTransport(); + final TransportSnapshot snapshot = snapshot(); + CURRENT_SNAPSHOT.set(snapshot); - if (transport == null || transport instanceof NullTransport) { - return new CheckResult(true, 0L, "No cluster cache transport in use (single node / community)"); + if (!snapshot.present) { + return new CheckResult(true, 0L, + "No cluster cache transport in use (" + snapshot.transportName + ")"); } - final long rewireFailures = ClusterFactory.getRewireFailures(); - final long dropped = transport.getDroppedMessages(); - - if (!transport.isInitialized()) { + if (!snapshot.initialized) { return new CheckResult(false, 0L, "Cache transport is NOT initialized - cluster cache invalidations are being dropped" - + " (dropped so far: " + dropped - + ", consecutive rewire failures: " + rewireFailures + ")"); + + " (dropped so far: " + snapshot.dropped + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); } - if (rewireFailures > 0) { + final int threshold = getConfigProperty("rewire-failure-threshold", + DEFAULT_REWIRE_FAILURE_THRESHOLD); + + if (snapshot.rewireFailures >= threshold) { return new CheckResult(false, 0L, - "Cluster rewire is failing (consecutive failures: " + rewireFailures - + ") - cache transport may be stale"); + "Cluster rewire has failed " + snapshot.rewireFailures + + " consecutive times (threshold: " + threshold + + ") - the cache transport may be stale"); } return new CheckResult(true, 0L, - "Cache transport initialized (" + transport.getClass().getSimpleName() - + ", dropped invalidations: " + dropped + ")"); + "Cache transport initialized (" + snapshot.transportName + + ", dropped invalidations: " + snapshot.dropped + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + } + + /** + * Reads every piece of transport state this check reports on, exactly once. + */ + private TransportSnapshot snapshot() { + + final long rewireFailures = ClusterFactory.getRewireFailures(); + final CacheTransport transport = getTransport(); + + if (transport == null) { + return TransportSnapshot.unavailable(rewireFailures); + } + + // NullTransport is a deliberate no-op transport (single node, or + // CACHE_INVALIDATION_TRANSPORT_CLASS pointed at it). Its isInitialized() is false + // whenever it has been shut down, which must not be read as dropped invalidations. + final boolean present = !(transport instanceof NullTransport); + + return new TransportSnapshot(transport.getClass().getSimpleName(), present, + transport.isInitialized(), transport.getDroppedMessages(), rewireFailures); } + /** + * Resolves the transport through the {@code DotCacheAdministrator} interface. + * + * Deliberately not {@code getImplementationObject()} plus a cast to + * {@code ChainableCacheAdministratorImpl}: {@code getTransport()} is part of the interface + * and {@code CommitListenerCacheWrapper} delegates it straight through, whereas the cast + * throws {@code ClassCastException} for any other administrator implementation -- + * {@code NullCacheAdministrator.getImplementationObject()} returns itself, which is the + * unit-test path. Compare {@code ClusterResource}, which uses the interface method. + */ private CacheTransport getTransport() { try { - return ((ChainableCacheAdministratorImpl) CacheLocator.getCacheAdministrator() - .getImplementationObject()).getTransport(); + return CacheLocator.getCacheAdministrator().getTransport(); } catch (Exception e) { // cache layer not up yet - nothing to report on return null; @@ -124,14 +187,52 @@ public String getDescription() { @Override protected Map buildStructuredData(CheckResult result, HealthStatus originalStatus, HealthStatus finalStatus, HealthCheckMode mode) { - final Map data = new HashMap<>(); - final CacheTransport transport = getTransport(); - if (transport != null) { - data.put("transport", transport.getClass().getSimpleName()); - data.put("initialized", transport.isInitialized()); - data.put("droppedInvalidations", transport.getDroppedMessages()); + try { + TransportSnapshot snapshot = CURRENT_SNAPSHOT.get(); + if (snapshot == null) { + // performCheck() threw before it could snapshot - resolve once here instead + snapshot = snapshot(); + } + + final Map data = new HashMap<>(); + data.put("transport", snapshot.transportName); + data.put("rewireFailures", snapshot.rewireFailures); + if (snapshot.present) { + data.put("initialized", snapshot.initialized); + data.put("droppedInvalidations", snapshot.dropped); + } + return data; + } finally { + CURRENT_SNAPSHOT.remove(); + } + } + + /** + * Immutable, single-read view of the transport state, so the message and the structured + * data of one health response can never disagree. + */ + private static final class TransportSnapshot { + + private static final String NO_TRANSPORT = "none"; + + final String transportName; + /** true only for a real transport that is expected to carry invalidations */ + final boolean present; + final boolean initialized; + final long dropped; + final long rewireFailures; + + TransportSnapshot(final String transportName, final boolean present, final boolean initialized, + final long dropped, final long rewireFailures) { + this.transportName = transportName; + this.present = present; + this.initialized = initialized; + this.dropped = dropped; + this.rewireFailures = rewireFailures; + } + + static TransportSnapshot unavailable(final long rewireFailures) { + return new TransportSnapshot(NO_TRANSPORT, false, false, 0L, rewireFailures); } - data.put("rewireFailures", ClusterFactory.getRewireFailures()); - return data; } } diff --git a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java index 5af2c99bf254..1ff43946a3be 100644 --- a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java +++ b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java @@ -1,9 +1,12 @@ package com.dotcms.metrics.binders; +import com.dotcms.enterprise.cluster.ClusterFactory; import com.dotmarketing.business.CacheLocator; import com.dotmarketing.business.DotCacheAdministrator; import com.dotmarketing.business.cache.provider.CacheProviderStats; import com.dotmarketing.business.cache.provider.CacheStats; +import com.dotmarketing.business.cache.transport.CacheTransport; +import com.dotmarketing.business.cache.transport.NullTransport; import com.dotmarketing.util.Logger; import com.dotmarketing.util.UtilMethods; import io.micrometer.core.instrument.Gauge; @@ -11,6 +14,7 @@ import io.micrometer.core.instrument.binder.MeterBinder; import java.util.List; +import java.util.Optional; /** * Comprehensive metric binder for dotCMS cache-related metrics. @@ -57,28 +61,43 @@ public void bindTo(MeterRegistry registry) { private void registerTransportMetrics(MeterRegistry registry) { Gauge.builder(METRIC_PREFIX + ".transport.invalidations.dropped", this, - m -> getTransport().map(t -> (double) t.getDroppedMessages()).orElse(0.0)) + m -> activeTransport().map(t -> (double) t.getDroppedMessages()).orElse(0.0)) .description("Cluster cache invalidations dropped because the transport was not initialized") .register(registry); Gauge.builder(METRIC_PREFIX + ".transport.initialized", this, - m -> getTransport().map(t -> t.isInitialized() ? 1.0 : 0.0).orElse(1.0)) + m -> activeTransport().map(t -> t.isInitialized() ? 1.0 : 0.0).orElse(1.0)) .description("Whether the cluster cache transport is initialized (1) or dropping invalidations (0)") .register(registry); Gauge.builder(METRIC_PREFIX + ".transport.rewire.failures", this, - m -> (double) com.dotcms.enterprise.cluster.ClusterFactory.getRewireFailures()) + m -> (double) ClusterFactory.getRewireFailures()) .description("Consecutive cluster cache-transport rewire failures (reset on success)") .register(registry); } - private java.util.Optional getTransport() { + /** + * The cache transport, but only when it is one that actually carries cluster invalidations. + * + * Returns empty for {@link NullTransport} as well as for an unresolvable transport, so the + * gauges above fall back to their "nothing to report" defaults. This matters for + * {@code transport.initialized}: {@code NullTransport.isInitialized()} is false whenever it + * has been shut down, so without this guard every node configured with no real transport + * would publish {@code transport.initialized=0} and alert as if it were dropping + * invalidations -- while {@code CacheTransportHealthCheck} reported the same node healthy. + * + * Uses the {@code DotCacheAdministrator.getTransport()} interface method rather than + * casting {@code getImplementationObject()} to {@code ChainableCacheAdministratorImpl}, + * which throws {@code ClassCastException} for any other administrator implementation. + */ + private Optional activeTransport() { try { - return java.util.Optional.ofNullable( - ((com.dotmarketing.business.ChainableCacheAdministratorImpl) CacheLocator - .getCacheAdministrator().getImplementationObject()).getTransport()); + final CacheTransport transport = CacheLocator.getCacheAdministrator().getTransport(); + return transport == null || transport instanceof NullTransport + ? Optional.empty() + : Optional.of(transport); } catch (Exception e) { - return java.util.Optional.empty(); + return Optional.empty(); } } From e1e806577c2853752cb9cd97fea9af2328cfb9fa Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Mon, 10 Aug 2026 19:19:47 -0600 Subject: [PATCH 3/8] fix(clustering): count cache invalidations that fail after init (#36803) The PR instrumented only the pre-init drop path. A transport that initialized successfully and then started failing every publish stayed completely invisible: no drops recorded, isInitialized() still true, rewireFailures 0, health check green, other nodes serving stale content. Every provider signals a failed send by returning false rather than throwing -- JDBCPubSubImpl (the default) and PostgresPubSubImpl on an exception or an execute() that returns false, RedisPubSubImpl when stopped -- and PubSubCacheTransport.send() discarded that boolean. Checking it in send() is necessary but not sufficient. DOT_PUBSUB_USE_QUEUE defaults to true, so the provider is normally a QueuingPubSubWrapper whose publish() returns true immediately and completes the real send on a submitter thread, discarding the result. In that configuration send() would read true 100% of the time. So the wrapper now records the outcome of the task it submits, and PubSubCacheTransport sums its own synchronous failures with the provider-reported ones -- exactly one of the two counts any given attempt, so nothing is double counted. Counted per topic, because one provider instance is shared by every topic in the JVM (cache invalidation, OSGi restart, cluster management) and a JVM-wide total could not be attributed to the cache transport. Exposed as CacheTransport.getFailedMessages(), the new dotcms.cache.transport.invalidations.failed gauge, and failedInvalidations in the health check's structured data. Kept distinct from getDroppedMessages() because the two mean different things operationally: dropped is "the transport is down", failed is "the transport believes it is up but sends are erroring". Failed invalidations deliberately do not make the health check unhealthy on their own. The count is cumulative and never resets, so alarming on "greater than zero" would pin a node DOWN forever after one transient error -- the same false-positive class fixed for rewire failures in the previous commit. Alert on the gauge's rate of increase instead. Known gap: RedisStreamsPubSubImpl always returns true (fire-and-forget async xadd), so it reports no failures. Left as-is. Refs: #36803 Co-Authored-By: Claude Opus 5 (1M context) --- .../cache/transport/PubSubCacheTransport.java | 58 ++++++-- .../dotcms/dotpubsub/DotPubSubProvider.java | 22 +++ .../dotpubsub/QueuingPubSubWrapper.java | 59 +++++++- .../checks/cdi/CacheTransportHealthCheck.java | 18 ++- .../dotcms/metrics/binders/CacheMetrics.java | 5 + .../cache/transport/CacheTransport.java | 13 ++ .../transport/PubSubCacheTransportTest.java | 90 ++++++++++++ .../dotpubsub/QueuingPubSubWrapperTest.java | 139 ++++++++++++++++++ 8 files changed, 387 insertions(+), 17 deletions(-) create mode 100644 dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java diff --git a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java index c527e6f05f1b..a6bdd7997a35 100644 --- a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java +++ b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java @@ -36,8 +36,17 @@ public class PubSubCacheTransport implements CacheTransport { final AtomicLong droppedMessages = new AtomicLong(0); + /** + * Sends that were attempted and reported as failed by a synchronous provider. Failures from + * an asynchronous provider are counted by the provider itself and read back in + * {@link #getFailedMessages()}; see {@link DotPubSubProvider#getFailedPublishCount(String)}. + */ + final AtomicLong failedMessages = new AtomicLong(0); + private final AtomicLong lastDropWarnAt = new AtomicLong(0); + private final AtomicLong lastFailWarnAt = new AtomicLong(0); + private static final long DROP_WARN_INTERVAL_MILLIS = Config.getLongProperty("CACHE_TRANSPORT_DROP_WARN_INTERVAL_MILLIS", 30000); @@ -77,24 +86,42 @@ public void init(final Server localServer) throws CacheTransportException { public void send(final String message) throws CacheTransportException { if (!this.initialized.get()) { final long dropped = this.droppedMessages.incrementAndGet(); - final long now = System.currentTimeMillis(); - final long lastWarn = this.lastDropWarnAt.get(); - if (now - lastWarn > DROP_WARN_INTERVAL_MILLIS - && this.lastDropWarnAt.compareAndSet(lastWarn, now)) { - Logger.warn(this.getClass(), - "Cache transport is not initialized - dropping cluster cache invalidations. " - + "Other nodes may serve stale content. Total dropped: " + dropped); - } + warnThrottled(this.lastDropWarnAt, + "Cache transport is not initialized - dropping cluster cache invalidations. " + + "Other nodes may serve stale content. Total dropped: " + dropped); return; } final DotPubSubEvent event = new DotPubSubEvent.Builder().withTopic(this.topic) .withType(CacheEventType.INVAL.name()).withMessage(message).build(); - this.pubsub.publish(event); + // The boolean matters: every provider signals a failed send by returning false rather + // than throwing, so ignoring it (as this did before issue #36803) made a transport that + // was initialized but failing every publish completely invisible - no drops recorded, + // isInitialized() still true, health checks green, other nodes stale. Asynchronous + // providers cannot answer here and return true immediately; their failures are counted + // provider-side and picked up by getFailedMessages(). + if (!this.pubsub.publish(event)) { + final long failed = this.failedMessages.incrementAndGet(); + warnThrottled(this.lastFailWarnAt, + "Cache transport failed to publish cluster cache invalidations. " + + "Other nodes may serve stale content. Total failed: " + failed); + } } + /** + * Logs at most one warning per {@link #DROP_WARN_INTERVAL_MILLIS} for the given throttle, so + * a sustained failure does not flood the log at invalidation rate. + */ + private void warnThrottled(final AtomicLong lastWarnAt, final String message) { + final long now = System.currentTimeMillis(); + final long lastWarn = lastWarnAt.get(); + if (now - lastWarn > DROP_WARN_INTERVAL_MILLIS && lastWarnAt.compareAndSet(lastWarn, now)) { + Logger.warn(this.getClass(), message); + } + } + @Override public void testCluster() throws CacheTransportException { @@ -159,6 +186,19 @@ public long getDroppedMessages() { return droppedMessages.get(); } + /** + * {@inheritDoc} + * + * Sums the failures this transport observed synchronously with those the provider recorded on + * its own thread. Exactly one of the two counts a given attempt: a synchronous provider + * returns the real boolean to {@link #send(String)} and reports 0 here, while an + * asynchronous one returns true immediately and counts the outcome itself. + */ + @Override + public long getFailedMessages() { + return failedMessages.get() + this.pubsub.getFailedPublishCount(this.topic.getTopic()); + } + @Override public CacheTransportInfo getInfo() { diff --git a/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java b/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java index 8921c8d42aa9..83300b773520 100644 --- a/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java +++ b/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java @@ -70,6 +70,28 @@ default DotPubSubEvent lastEventOut() { default String getProviderName() { return getClass().getSimpleName(); } + + /** + * Number of publish attempts for the given topic that this provider knows to have failed, + * i.e. where {@link #publish(DotPubSubEvent)} ultimately returned {@code false}. + * + * Only meaningful for providers that publish asynchronously and therefore cannot report the + * outcome through their own return value -- {@link QueuingPubSubWrapper} is the one that + * does, returning {@code true} immediately and completing the real publish on another + * thread. Callers that publish synchronously already see the {@code boolean} and should + * count it themselves; they must not add this value on top of their own count for the same + * attempt, or one failure is counted twice. + * + * Counted per topic because a single provider instance is shared by every topic in the JVM + * (cache invalidation, OSGi restart, cluster management), so a JVM-wide total could not be + * attributed to the cache transport. + * + * @param topic the topic key, as returned by {@link DotPubSubEvent#getTopic()} + * @return the failure count, or 0 for providers that publish synchronously + */ + default long getFailedPublishCount(final String topic) { + return 0; + } diff --git a/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java b/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java index 633ad5907db2..c7bef4d49da7 100644 --- a/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java +++ b/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java @@ -1,7 +1,10 @@ package com.dotcms.dotpubsub; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import com.dotcms.concurrent.DotConcurrentFactory; import com.dotcms.concurrent.DotConcurrentFactory.SubmitterConfig; import com.dotcms.concurrent.DotSubmitter; @@ -22,8 +25,14 @@ public class QueuingPubSubWrapper implements DotPubSubProvider { private final DotPubSubProvider wrappedProvider; private final Cache recentEvents ; - + private final DotSubmitter submitter; + + /** + * Failed publish attempts per topic. Keyed by topic because one provider instance serves + * every topic in the JVM, so a single total could not be attributed to the cache transport. + */ + private final Map failedByTopic = new ConcurrentHashMap<>(); @@ -98,9 +107,9 @@ public boolean publish(final DotPubSubEvent event) { sent++; recentEvents.put(event.hashCode(), true); - submitter.submit(()->this.wrappedProvider.publish(event)); - - + submitter.submit(()->publishAndRecordOutcome(event)); + + Logger.debug(this.getClass(), ()->"sent/skipped: " + sent + "/" + skipped); Logger.debug(this.getClass(), ()->"active count:" + submitter.getActiveCount()); @@ -109,6 +118,48 @@ public boolean publish(final DotPubSubEvent event) { } + /** + * Runs the real publish on the submitter thread and records the outcome. + * + * {@link #publish(DotPubSubEvent)} has to return before this completes, so it always returns + * {@code true} and the caller never learns whether the send actually worked. Before issue + * #36803 the result of this task was discarded entirely, which meant a transport that was + * initialized but failing every publish -- a dropped database connection, a stopped Redis + * client -- reported no drops, stayed "initialized", and kept every health check green while + * other nodes served stale content. + */ + private void publishAndRecordOutcome(final DotPubSubEvent event) { + try { + if (!this.wrappedProvider.publish(event)) { + recordFailure(event); + } + } catch (Exception e) { + // providers are expected to swallow their own failures and return false; this is + // the belt-and-braces path so an unexpected throw is still counted, not lost + recordFailure(event); + Logger.warnAndDebug(this.getClass(), + "Unable to publish pubsub event " + event + " : " + e.getMessage(), e); + } + } + + private void recordFailure(final DotPubSubEvent event) { + failedByTopic.computeIfAbsent(String.valueOf(event.getTopic()), t -> new AtomicLong(0)) + .incrementAndGet(); + } + + /** + * {@inheritDoc} + * + * Returns only this wrapper's own count. The wrapped provider published synchronously from + * the submitter thread, so its {@code false} return has already been counted here -- adding + * {@code wrappedProvider.getFailedPublishCount(topic)} would double count. + */ + @Override + public long getFailedPublishCount(final String topic) { + final AtomicLong failed = failedByTopic.get(topic); + return failed == null ? 0 : failed.get(); + } + @Override public DotPubSubProvider unsubscribe(DotPubSubTopic topic) { diff --git a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java index 1a93148f450a..891befdebd98 100644 --- a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java +++ b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java @@ -20,7 +20,12 @@ * * - transport not initialized -> unhealthy * - cluster rewire failing repeatedly -> unhealthy (see {@link #DEFAULT_REWIRE_FAILURE_THRESHOLD}) - * - dropped-invalidation count exposed in structured data for monitoring + * - dropped and failed invalidation counts exposed in structured data for monitoring + * + * Failed invalidations (initialized transport whose sends are erroring) are reported but do not + * on their own make the check unhealthy: the count is cumulative and never resets, so alarming on + * "greater than zero" would pin a node DOWN forever after one transient publish error. Alert on + * the rate of increase of {@code dotcms.cache.transport.invalidations.failed} instead. * * Nodes with no real transport -- {@link NullTransport} via * {@code CACHE_INVALIDATION_TRANSPORT_CLASS}, or no resolvable cache layer at all -- always @@ -102,6 +107,7 @@ protected CheckResult performCheck() throws Exception { return new CheckResult(true, 0L, "Cache transport initialized (" + snapshot.transportName + ", dropped invalidations: " + snapshot.dropped + + ", failed invalidations: " + snapshot.failed + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); } @@ -123,7 +129,8 @@ private TransportSnapshot snapshot() { final boolean present = !(transport instanceof NullTransport); return new TransportSnapshot(transport.getClass().getSimpleName(), present, - transport.isInitialized(), transport.getDroppedMessages(), rewireFailures); + transport.isInitialized(), transport.getDroppedMessages(), + transport.getFailedMessages(), rewireFailures); } /** @@ -200,6 +207,7 @@ protected Map buildStructuredData(CheckResult result, HealthStat if (snapshot.present) { data.put("initialized", snapshot.initialized); data.put("droppedInvalidations", snapshot.dropped); + data.put("failedInvalidations", snapshot.failed); } return data; } finally { @@ -220,19 +228,21 @@ private static final class TransportSnapshot { final boolean present; final boolean initialized; final long dropped; + final long failed; final long rewireFailures; TransportSnapshot(final String transportName, final boolean present, final boolean initialized, - final long dropped, final long rewireFailures) { + final long dropped, final long failed, final long rewireFailures) { this.transportName = transportName; this.present = present; this.initialized = initialized; this.dropped = dropped; + this.failed = failed; this.rewireFailures = rewireFailures; } static TransportSnapshot unavailable(final long rewireFailures) { - return new TransportSnapshot(NO_TRANSPORT, false, false, 0L, rewireFailures); + return new TransportSnapshot(NO_TRANSPORT, false, false, 0L, 0L, rewireFailures); } } } diff --git a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java index 1ff43946a3be..39d744a137d0 100644 --- a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java +++ b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java @@ -65,6 +65,11 @@ private void registerTransportMetrics(MeterRegistry registry) { .description("Cluster cache invalidations dropped because the transport was not initialized") .register(registry); + Gauge.builder(METRIC_PREFIX + ".transport.invalidations.failed", this, + m -> activeTransport().map(t -> (double) t.getFailedMessages()).orElse(0.0)) + .description("Cluster cache invalidations the transport attempted but failed to publish") + .register(registry); + Gauge.builder(METRIC_PREFIX + ".transport.initialized", this, m -> activeTransport().map(t -> t.isInitialized() ? 1.0 : 0.0).orElse(1.0)) .description("Whether the cluster cache transport is initialized (1) or dropping invalidations (0)") diff --git a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java index 3d5d3c4ebba5..fdc78c85b456 100644 --- a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java +++ b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java @@ -65,6 +65,19 @@ default boolean requiresAutowiring() { default long getDroppedMessages() { return 0; } + + /** + * Number of cache invalidation messages this transport tried to send and that the underlying + * provider reported as failed -- as opposed to {@link #getDroppedMessages()}, which counts + * messages never attempted because the transport was not initialized. + * + * The distinction is operational: dropped means "the transport is down, fix the transport", + * failed means "the transport believes it is up but sends are erroring". Both lose cluster + * invalidations, but only the first is visible from {@link #isInitialized()}. + */ + default long getFailedMessages() { + return 0; + } public interface CacheTransportInfo extends Serializable { diff --git a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java index 53d5c460066b..419ec4417676 100644 --- a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java +++ b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java @@ -79,6 +79,66 @@ public void test_init_is_idempotent() throws Exception { assertEquals(2, provider.starts); } + /** + * The scenario the health check could not see before: init() succeeded, so the transport + * reports initialized and counts no drops, but every publish is failing. Providers signal + * that by returning false rather than throwing, and send() used to discard the boolean. + */ + @Test + public void test_send_after_init_counts_failed_publishes() throws Exception { + final FailingProvider provider = new FailingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + transport.init(null); + assertTrue(transport.isInitialized()); + + transport.send("0:testGroup"); + transport.send("0:testGroup"); + + assertEquals("a failing publish is not a drop - it was attempted", + 0, transport.getDroppedMessages()); + assertEquals("failed publishes must be counted, not discarded", + 2, transport.getFailedMessages()); + assertTrue("the transport still believes it is initialized", transport.isInitialized()); + } + + /** + * A provider that publishes successfully must not inflate the failure counter. + */ + @Test + public void test_send_after_init_counts_no_failures_when_publish_succeeds() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.init(null); + transport.send("0:testGroup"); + + assertEquals(0, transport.getFailedMessages()); + assertEquals(0, transport.getDroppedMessages()); + } + + /** + * An asynchronous provider returns true from publish() before the send has happened, so it + * reports its own failures per topic instead. getFailedMessages() must pick those up, and + * must not double count them against send()'s own synchronous tally. + */ + @Test + public void test_getFailedMessages_includes_async_provider_failures() throws Exception { + final AsyncFailingProvider provider = new AsyncFailingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + transport.init(null); + transport.send("0:testGroup"); + transport.send("0:testGroup"); + + assertEquals("send() saw only the optimistic true, so it counted nothing itself", + 0, transport.failedMessages.get()); + assertEquals("provider-reported failures must still surface", 2, + transport.getFailedMessages()); + } + private static class CountingProvider extends NullDotPubSubProvider { int starts = 0; @@ -88,4 +148,34 @@ public com.dotcms.dotpubsub.DotPubSubProvider start() { return this; } } + + /** + * Mimics a synchronous provider whose connection has dropped: publish() returns false and + * never throws, exactly as JDBCPubSubImpl/PostgresPubSubImpl/RedisPubSubImpl do. + */ + private static class FailingProvider extends NullDotPubSubProvider { + @Override + public boolean publish(final com.dotcms.dotpubsub.DotPubSubEvent event) { + return false; + } + } + + /** + * Mimics QueuingPubSubWrapper: publish() returns true immediately and the real outcome is + * reported later through getFailedPublishCount(). + */ + private static class AsyncFailingProvider extends NullDotPubSubProvider { + private long failures = 0; + + @Override + public boolean publish(final com.dotcms.dotpubsub.DotPubSubEvent event) { + failures++; + return true; + } + + @Override + public long getFailedPublishCount(final String topic) { + return failures; + } + } } diff --git a/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java b/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java new file mode 100644 index 000000000000..e13a468ccb59 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java @@ -0,0 +1,139 @@ +package com.dotcms.dotpubsub; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; + +/** + * Covers the failure accounting added for issue #36803. + * + * {@link QueuingPubSubWrapper#publish} returns {@code true} before the wrapped provider has + * actually published, so the caller can never see a failure. Previously the result of the + * submitted task was discarded outright, which is what let a transport that was initialized but + * failing every publish look completely healthy. + */ +public class QueuingPubSubWrapperTest { + + private static final String CACHE_TOPIC = "dotcache_topic"; + private static final String OTHER_TOPIC = "osgi_topic"; + + private static final long AWAIT_MILLIS = 5000; + + /** + * Waits for the wrapper's submitter thread to drain, since publish() is asynchronous. + */ + private static void awaitFailures(final QueuingPubSubWrapper wrapper, final String topic, + final long expected) throws InterruptedException { + final long deadline = System.currentTimeMillis() + AWAIT_MILLIS; + while (System.currentTimeMillis() < deadline + && wrapper.getFailedPublishCount(topic) < expected) { + Thread.sleep(25); + } + } + + private static DotPubSubEvent event(final String topic, final String message) { + return new DotPubSubEvent.Builder().withTopic(topic).withMessage(message).build(); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: the wrapped provider returns false for every publish, as a synchronous + * provider with a dropped connection does. + * Expected Result: each failure is counted even though publish() reported success to the + * caller. + */ + @Test + public void test_failed_publishes_are_counted() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new AlwaysFailsProvider()); + + assertTrue("publish() reports success regardless - that is why the count is needed", + wrapper.publish(event(CACHE_TOPIC, "inval-1"))); + assertTrue(wrapper.publish(event(CACHE_TOPIC, "inval-2"))); + + awaitFailures(wrapper, CACHE_TOPIC, 2); + assertEquals(2, wrapper.getFailedPublishCount(CACHE_TOPIC)); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: a provider that throws instead of returning false. + * Expected Result: the throw is counted rather than lost on the submitter thread. + */ + @Test + public void test_thrown_publish_failures_are_counted() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new ThrowingProvider()); + + wrapper.publish(event(CACHE_TOPIC, "inval-throw")); + + awaitFailures(wrapper, CACHE_TOPIC, 1); + assertEquals(1, wrapper.getFailedPublishCount(CACHE_TOPIC)); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: one provider instance is shared by every topic in the JVM, and a non-cache + * topic fails. + * Expected Result: the failure is attributed to its own topic only, so the cache transport's + * metric is not inflated by OSGi or cluster-management traffic. + */ + @Test + public void test_failures_are_attributed_per_topic() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new AlwaysFailsProvider()); + + wrapper.publish(event(OTHER_TOPIC, "osgi-restart")); + + awaitFailures(wrapper, OTHER_TOPIC, 1); + assertEquals(1, wrapper.getFailedPublishCount(OTHER_TOPIC)); + assertEquals("cache topic saw no traffic and must report no failures", + 0, wrapper.getFailedPublishCount(CACHE_TOPIC)); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: the wrapped provider publishes successfully. + * Expected Result: nothing is counted, and a topic that was never published to reports 0 + * rather than failing on a missing counter. + */ + @Test + public void test_successful_publishes_are_not_counted() throws Exception { + final CountingProvider provider = new CountingProvider(); + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(provider); + + wrapper.publish(event(CACHE_TOPIC, "inval-ok")); + + final long deadline = System.currentTimeMillis() + AWAIT_MILLIS; + while (System.currentTimeMillis() < deadline && provider.published.get() == 0) { + Thread.sleep(25); + } + + assertEquals(1, provider.published.get()); + assertEquals(0, wrapper.getFailedPublishCount(CACHE_TOPIC)); + assertEquals(0, wrapper.getFailedPublishCount("never-used-topic")); + } + + private static class AlwaysFailsProvider extends NullDotPubSubProvider { + @Override + public boolean publish(final DotPubSubEvent event) { + return false; + } + } + + private static class ThrowingProvider extends NullDotPubSubProvider { + @Override + public boolean publish(final DotPubSubEvent event) { + throw new IllegalStateException("pubsub connection is gone"); + } + } + + private static class CountingProvider extends NullDotPubSubProvider { + final AtomicInteger published = new AtomicInteger(0); + + @Override + public boolean publish(final DotPubSubEvent event) { + published.incrementAndGet(); + return true; + } + } +} From 0f7d92b0b76588b99f0e36ccd5bef45168d95058 Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Mon, 10 Aug 2026 21:49:41 -0600 Subject: [PATCH 4/8] fix(clustering): stop reporting startup cache-invalidation drops as failures (#36803) A node necessarily drops cache invalidations before its transport comes up: caches are invalidated by startup tasks and the starter import long before ClusterFactory wires the cluster and calls PubSubCacheTransport.init(). A first boot against an empty database was measured at ~2,800 of them, and the cumulative counter carried that burst for the life of the node. Two consequences, both of which would land on operators the moment this ships: - getDroppedMessages() reported thousands on a perfectly healthy node, so the counter was useless as an alerting signal -- the exact thing this issue exists to provide. - CacheTransportHealthCheck reported the transport uninitialized for the few seconds between the first poll and cluster wiring. deriveOverallStatus() propagates any DEGRADED component to the overall status, so every pod start would have shown a DEGRADED /dotmgt/health payload. Measured on a local two-node cluster: an 11s window on a fresh boot, 3s on a populated database. Drops from before the first successful init are now retired into a separate startupDroppedMessages counter, and the health check treats a never-yet- initialized transport as still starting up for a configurable grace period (health.check.cache-transport.initialization-grace-period-seconds, default 120). Retiring is deliberately first-init-only. A transport that came up, went down, lost invalidations and recovered has lost real ones; retiring on every re-init would launder genuine loss into the benign startup bucket. Equally, the grace period ends for good once the check has seen an initialized transport -- losing one that had been working is a regression, not a node still booting, and is reported immediately. The transport.initialized gauge is left raw, with no grace period: a gauge states the current fact and alert rules add the duration clause. Noted in the metric description. Refs: #36803 --- .../cache/transport/PubSubCacheTransport.java | 63 +++++++++++++ .../checks/cdi/CacheTransportHealthCheck.java | 93 +++++++++++++++++-- .../dotcms/metrics/binders/CacheMetrics.java | 13 ++- .../cache/transport/CacheTransport.java | 19 +++- .../transport/PubSubCacheTransportTest.java | 58 +++++++++++- 5 files changed, 233 insertions(+), 13 deletions(-) diff --git a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java index a6bdd7997a35..14ece9fb4915 100644 --- a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java +++ b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java @@ -36,6 +36,28 @@ public class PubSubCacheTransport implements CacheTransport { final AtomicLong droppedMessages = new AtomicLong(0); + /** + * Invalidations dropped before this transport was ever initialized. + * + * Boot order guarantees some of these: caches are invalidated by startup tasks and by the + * starter import long before {@code ClusterFactory} wires the cluster and calls + * {@link #init(Server)}, and a node that is alone at that point has nobody to notify anyway. A + * first boot against an empty database was measured at ~2,800 of them. + * + * They are kept apart from {@link #droppedMessages} so the number an operator sees on a + * healthy node is zero rather than thousands of benign startup drops -- a cumulative counter + * dominated by boot noise is worthless as an alerting signal, which is the whole point of + * issue #36803. Reported separately for diagnostics. + */ + final AtomicLong startupDroppedMessages = new AtomicLong(0); + + /** + * Whether {@link #init(Server)} has ever succeeded, which ends the startup accounting for the + * life of this transport. Drops after that point were suffered by a transport that had been + * working and are genuine invalidation loss, so a later re-init must not retire them. + */ + private final AtomicBoolean everInitialized = new AtomicBoolean(false); + /** * Sends that were attempted and reported as failed by a synchronous provider. Failures from * an asynchronous provider are counted by the provider itself and read back in @@ -80,6 +102,36 @@ public void init(final Server localServer) throws CacheTransportException { this.initialized.set(true); + retireStartupDrops(); + } + + /** + * On the first successful init only, moves the drops accumulated so far into + * {@link #startupDroppedMessages}, so {@link #getDroppedMessages()} counts only invalidations + * lost while the transport was expected to be carrying them. + * + * Deliberately first-init-only. A transport that came up, went down, dropped invalidations and + * recovered has lost real ones, and retiring those on every re-init would launder genuine loss + * into the benign startup bucket -- exactly the blind spot issue #36803 exists to remove. + * + * The drop-warning throttle is cleared on every init: a genuine drop minutes later must log + * immediately rather than be swallowed because an earlier drop consumed the window. + * + * A {@code send()} racing with this can have its increment land in either bucket. The total is + * preserved either way, and both counters are diagnostics rather than exact accounting. + */ + private void retireStartupDrops() { + + if (this.everInitialized.compareAndSet(false, true)) { + final long startupDrops = this.droppedMessages.getAndSet(0); + if (startupDrops > 0) { + this.startupDroppedMessages.addAndGet(startupDrops); + Logger.info(this.getClass(), "Cache transport initialized. " + startupDrops + + " cache invalidation(s) were dropped before it came up (expected during" + + " startup, when there is no cluster to notify yet)."); + } + } + this.lastDropWarnAt.set(0); } @Override @@ -181,11 +233,22 @@ public boolean shouldReinit() { return !initialized.get(); } + /** + * {@inheritDoc} + * + * Counts only drops recorded after the transport was first initialized; see + * {@link #startupDroppedMessages} for the benign pre-init ones. + */ @Override public long getDroppedMessages() { return droppedMessages.get(); } + @Override + public long getStartupDroppedMessages() { + return startupDroppedMessages.get(); + } + /** * {@inheritDoc} * diff --git a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java index 891befdebd98..fb70c7581d6a 100644 --- a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java +++ b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java @@ -9,6 +9,9 @@ import com.dotmarketing.business.cache.transport.NullTransport; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import javax.enterprise.context.ApplicationScoped; /** @@ -18,7 +21,8 @@ * tries to send, so other nodes serve stale content while all standard probes report healthy * (see issue #36803 / incident #36544). This check surfaces that state: * - * - transport not initialized -> unhealthy + * - transport not initialized -> unhealthy, once past the startup grace period + * (see {@link #DEFAULT_INITIALIZATION_GRACE_SECONDS}) * - cluster rewire failing repeatedly -> unhealthy (see {@link #DEFAULT_REWIRE_FAILURE_THRESHOLD}) * - dropped and failed invalidation counts exposed in structured data for monitoring * @@ -41,6 +45,8 @@ * - health.check.cache-transport.mode = Safety mode (PRODUCTION, MONITOR_MODE, DISABLED) * - health.check.cache-transport.rewire-failure-threshold = consecutive rewire failures * tolerated before reporting unhealthy (default 3) + * - health.check.cache-transport.initialization-grace-period-seconds = how long a + * never-yet-initialized transport is reported as still starting up (default 120) */ @ApplicationScoped public class CacheTransportHealthCheck extends HealthCheckBase { @@ -57,6 +63,34 @@ public class CacheTransportHealthCheck extends HealthCheckBase { */ private static final int DEFAULT_REWIRE_FAILURE_THRESHOLD = 3; + /** + * How long a transport that has never been initialized is treated as still starting up. + * + * The transport is initialized late in boot, when {@code ClusterFactory} wires the cluster, so + * every node necessarily reports an uninitialized transport for the first seconds of its life. + * Failing the check there would turn a normal rolling deployment into a DEGRADED overall + * status on every pod start -- noise that costs an operator's trust in the signal long before + * it ever catches the real fault from issue #36803. A node genuinely stuck without a transport + * stays stuck, so waiting this out loses nothing but the first two minutes of detection. + * + * Only applies before the first successful init. Once this check has seen an initialized + * transport, losing it is a regression and is reported immediately. + */ + private static final int DEFAULT_INITIALIZATION_GRACE_SECONDS = 120; + + /** + * Latched the first time an initialized transport is observed, which ends the startup grace + * period for the lifetime of the JVM. + */ + private final AtomicBoolean everInitialized = new AtomicBoolean(false); + + /** + * When the current run of uninitialized observations began, or 0 if the last observation was + * healthy. Used to measure the grace period from the first poll that saw the problem rather + * than from JVM start, so the window is not consumed by a slow boot before this check runs. + */ + private final AtomicLong uninitializedSince = new AtomicLong(0); + /** * Transport state resolved once per {@code check()} invocation. * @@ -88,12 +122,12 @@ protected CheckResult performCheck() throws Exception { } if (!snapshot.initialized) { - return new CheckResult(false, 0L, - "Cache transport is NOT initialized - cluster cache invalidations are being dropped" - + " (dropped so far: " + snapshot.dropped - + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + return uninitialized(snapshot); } + everInitialized.set(true); + uninitializedSince.set(0); + final int threshold = getConfigProperty("rewire-failure-threshold", DEFAULT_REWIRE_FAILURE_THRESHOLD); @@ -111,6 +145,41 @@ protected CheckResult performCheck() throws Exception { + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); } + /** + * Decides whether an uninitialized transport is a node that is still starting up or a node + * that is silently dropping invalidations. See {@link #DEFAULT_INITIALIZATION_GRACE_SECONDS}. + */ + private CheckResult uninitialized(final TransportSnapshot snapshot) { + + final long now = System.currentTimeMillis(); + uninitializedSince.compareAndSet(0, now); + + if (!everInitialized.get()) { + + final long graceMillis = TimeUnit.SECONDS.toMillis(getConfigProperty( + "initialization-grace-period-seconds", DEFAULT_INITIALIZATION_GRACE_SECONDS)); + final long waitingMillis = now - uninitializedSince.get(); + + if (waitingMillis < graceMillis) { + return new CheckResult(true, 0L, + "Cache transport is still initializing (" + waitingMillis / 1000 + + "s of " + graceMillis / 1000 + "s grace period, dropped so far: " + + snapshot.dropped + ")"); + } + + return new CheckResult(false, 0L, + "Cache transport has NOT initialized within " + graceMillis / 1000 + + "s of startup - cluster cache invalidations are being dropped" + + " (dropped so far: " + snapshot.dropped + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + } + + return new CheckResult(false, 0L, + "Cache transport was initialized and is NOT anymore - cluster cache invalidations" + + " are being dropped (dropped so far: " + snapshot.dropped + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + } + /** * Reads every piece of transport state this check reports on, exactly once. */ @@ -130,7 +199,8 @@ private TransportSnapshot snapshot() { return new TransportSnapshot(transport.getClass().getSimpleName(), present, transport.isInitialized(), transport.getDroppedMessages(), - transport.getFailedMessages(), rewireFailures); + transport.getStartupDroppedMessages(), transport.getFailedMessages(), + rewireFailures); } /** @@ -207,7 +277,11 @@ protected Map buildStructuredData(CheckResult result, HealthStat if (snapshot.present) { data.put("initialized", snapshot.initialized); data.put("droppedInvalidations", snapshot.dropped); + data.put("startupDroppedInvalidations", snapshot.startupDropped); data.put("failedInvalidations", snapshot.failed); + // Lets a consumer tell "initialized:false but still booting, reported UP" apart + // from "initialized:false and past the grace period, reported DOWN". + data.put("awaitingInitialization", !snapshot.initialized && !everInitialized.get()); } return data; } finally { @@ -228,21 +302,24 @@ private static final class TransportSnapshot { final boolean present; final boolean initialized; final long dropped; + final long startupDropped; final long failed; final long rewireFailures; TransportSnapshot(final String transportName, final boolean present, final boolean initialized, - final long dropped, final long failed, final long rewireFailures) { + final long dropped, final long startupDropped, final long failed, + final long rewireFailures) { this.transportName = transportName; this.present = present; this.initialized = initialized; this.dropped = dropped; + this.startupDropped = startupDropped; this.failed = failed; this.rewireFailures = rewireFailures; } static TransportSnapshot unavailable(final long rewireFailures) { - return new TransportSnapshot(NO_TRANSPORT, false, false, 0L, 0L, rewireFailures); + return new TransportSnapshot(NO_TRANSPORT, false, false, 0L, 0L, 0L, rewireFailures); } } } diff --git a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java index 39d744a137d0..f9d00a0d2711 100644 --- a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java +++ b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java @@ -62,7 +62,14 @@ private void registerTransportMetrics(MeterRegistry registry) { Gauge.builder(METRIC_PREFIX + ".transport.invalidations.dropped", this, m -> activeTransport().map(t -> (double) t.getDroppedMessages()).orElse(0.0)) - .description("Cluster cache invalidations dropped because the transport was not initialized") + .description("Cluster cache invalidations dropped, after the transport first came up," + + " because it was not initialized") + .register(registry); + + Gauge.builder(METRIC_PREFIX + ".transport.invalidations.dropped.startup", this, + m -> activeTransport().map(t -> (double) t.getStartupDroppedMessages()).orElse(0.0)) + .description("Cluster cache invalidations dropped before the transport was first" + + " initialized (expected during startup, not an alerting signal)") .register(registry); Gauge.builder(METRIC_PREFIX + ".transport.invalidations.failed", this, @@ -70,6 +77,10 @@ private void registerTransportMetrics(MeterRegistry registry) { .description("Cluster cache invalidations the transport attempted but failed to publish") .register(registry); + // Reported raw, with no startup grace period: a gauge's job is to state the current + // fact, and every node reads 0 here for the first seconds of its life. Alert on it with + // a duration clause (Prometheus "for: 2m") rather than on the instantaneous value, or + // alert on the cache-transport health check, which applies the grace period itself. Gauge.builder(METRIC_PREFIX + ".transport.initialized", this, m -> activeTransport().map(t -> t.isInitialized() ? 1.0 : 0.0).orElse(1.0)) .description("Whether the cluster cache transport is initialized (1) or dropping invalidations (0)") diff --git a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java index fdc78c85b456..e4adbedf866b 100644 --- a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java +++ b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java @@ -58,14 +58,27 @@ default boolean requiresAutowiring() { } /** - * Number of cache invalidation messages this transport has dropped, e.g. because it was - * asked to send before it was initialized. Used by health checks and metrics to surface - * silent invalidation loss in a cluster. + * Number of cache invalidation messages this transport has dropped because it was asked to + * send while not initialized, counted only from the first successful {@link #init(Server)} + * onwards. Used by health checks and metrics to surface silent invalidation loss in a cluster. + * + * Drops from before the transport ever came up are reported by + * {@link #getStartupDroppedMessages()} instead: they are an unavoidable consequence of boot + * order and would otherwise dominate this counter permanently. */ default long getDroppedMessages() { return 0; } + /** + * Number of cache invalidation messages dropped before this transport was initialized for the + * first time, which is expected during startup and is not an operational problem on its own. + * Kept out of {@link #getDroppedMessages()} so that counter stays usable for alerting. + */ + default long getStartupDroppedMessages() { + return 0; + } + /** * Number of cache invalidation messages this transport tried to send and that the underlying * provider reported as failed -- as opposed to {@link #getDroppedMessages()}, which counts diff --git a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java index 419ec4417676..1b6de485b794 100644 --- a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java +++ b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java @@ -52,7 +52,63 @@ public void test_send_after_init_publishes() throws Exception { transport.send("0:testGroup"); assertNotNull(provider.lastEventOut()); - assertEquals(1, transport.getDroppedMessages()); + assertEquals("the pre-init drop was retired to the startup counter", + 0, transport.getDroppedMessages()); + assertEquals(1, transport.getStartupDroppedMessages()); + } + + /** + * A successful init() moves the drops that happened before the transport ever came up into the + * startup counter, so getDroppedMessages() reports only invalidations lost while the transport + * was expected to be carrying them. Without this, every node permanently reports the boot + * burst -- measured at ~2,800 on a first boot -- and the counter is useless for alerting. + */ + @Test + public void test_init_retires_startup_drops_from_the_dropped_counter() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.send("boot-1"); + transport.send("boot-2"); + transport.send("boot-3"); + assertEquals(3, transport.getDroppedMessages()); + assertEquals(0, transport.getStartupDroppedMessages()); + + transport.init(null); + + assertEquals("drops from before the first init are not operational drops", + 0, transport.getDroppedMessages()); + assertEquals("but they are still reported, for diagnostics", + 3, transport.getStartupDroppedMessages()); + } + + /** + * Drops after the transport has been up once are real invalidation loss: the transport was + * expected to be carrying them. A later re-init must not launder those into the startup + * bucket, which would recreate the blind spot this issue exists to remove. + */ + @Test + public void test_drops_after_first_init_are_not_retired_by_a_later_init() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.send("boot-1"); + transport.init(null); + assertEquals(1, transport.getStartupDroppedMessages()); + + // the transport goes down, and invalidations are lost while it is down + transport.shutdown(); + assertFalse(transport.isInitialized()); + transport.send("lost-1"); + transport.send("lost-2"); + assertEquals(2, transport.getDroppedMessages()); + + // recovery must leave the real loss on the books + transport.init(null); + + assertEquals("real drops survive a re-init", 2, transport.getDroppedMessages()); + assertEquals("and are not moved into the startup bucket", + 1, transport.getStartupDroppedMessages()); } /** From e322dcf715a1098392383a0df08cc0c484d130b0 Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Mon, 10 Aug 2026 21:49:59 -0600 Subject: [PATCH 5/8] test(clustering): cover CacheTransportHealthCheck decision logic (#36803) The review flagged the health check as the largest new file in the PR with no unit coverage, which was fair: its decision logic is the "surface the silent failure" behaviour this issue is about, and nothing pinned it. Ten cases against statically mocked CacheLocator and ClusterFactory, in the style of VelocityHealthCheckTest: - NullTransport reports UP, with the per-transport fields omitted rather than reported as misleading zeros - an unresolvable cache layer reports UP instead of throwing out of the probe - a never-initialized transport reports UP inside the startup grace period and DOWN past it - a transport lost after having been initialized reports DOWN immediately, even with a long grace period configured - rewire failures below the threshold report UP, at the threshold report DOWN - structured data carries every counter monitoring consumes, with startup drops separate from operational ones - the default mode stays MONITOR_MODE and converts DOWN to DEGRADED, pinning the property that keeps a broken transport from draining a node that can still serve traffic - the check is never a liveness check Writing these caught a real defect in the preceding commit: retireStartupDrops() ran on every init, so a transport that went down, lost invalidations and recovered would have had that genuine loss moved into the startup bucket. Fixed there before this commit; test_drops_after_first_init_are_not_retired_by_a_later_init holds the line. Refs: #36803 --- .../cdi/CacheTransportHealthCheckTest.java | 280 ++++++++++++++++++ 1 file changed, 280 insertions(+) create mode 100644 dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java diff --git a/dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java b/dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java new file mode 100644 index 000000000000..21e72b87e5cc --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java @@ -0,0 +1,280 @@ +package com.dotcms.health.checks.cdi; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +import com.dotcms.enterprise.cluster.ClusterFactory; +import com.dotcms.health.config.HealthCheckConfig.HealthCheckMode; +import com.dotcms.health.model.HealthCheckResult; +import com.dotcms.health.model.HealthStatus; +import com.dotmarketing.business.CacheLocator; +import com.dotmarketing.business.DotCacheAdministrator; +import com.dotmarketing.business.cache.transport.CacheTransport; +import com.dotmarketing.business.cache.transport.NullTransport; +import com.dotmarketing.util.Config; +import java.util.Map; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.MockedStatic; + +/** + * Unit tests for {@link CacheTransportHealthCheck}, the probe added by issue #36803 to surface a + * cache transport that is silently dropping cluster invalidations. + * + * The transport and the rewire counter are both reached through statics + * ({@code CacheLocator.getCacheAdministrator()} and {@code ClusterFactory.getRewireFailures()}), + * so they are mocked statically here, in the same style as {@link VelocityHealthCheckTest}. + * + * Tests run the check in PRODUCTION mode unless they are specifically about monitor mode, so an + * unhealthy verdict shows up as DOWN rather than being converted to DEGRADED. + */ +public class CacheTransportHealthCheckTest { + + private static final String MODE_KEY = "health.check.cache-transport.mode"; + private static final String GRACE_KEY = + "health.check.cache-transport.initialization-grace-period-seconds"; + private static final String THRESHOLD_KEY = + "health.check.cache-transport.rewire-failure-threshold"; + + private MockedStatic cacheLocatorMock; + private MockedStatic clusterFactoryMock; + + @Before + public void setUp() { + cacheLocatorMock = mockStatic(CacheLocator.class); + clusterFactoryMock = mockStatic(ClusterFactory.class); + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(0L); + Config.setProperty(MODE_KEY, "PRODUCTION"); + } + + @After + public void tearDown() { + cacheLocatorMock.close(); + clusterFactoryMock.close(); + // Restored to the values the production code defaults to, rather than cleared: Config has + // no clear-for-tests hook, and leaving these set would follow the JVM fork into other + // test classes. + Config.setProperty(MODE_KEY, "MONITOR_MODE"); + Config.setProperty(GRACE_KEY, "120"); + Config.setProperty(THRESHOLD_KEY, "3"); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the node is configured with {@link NullTransport}, the deliberate no-op + * transport for a node that is not part of a cluster. + * Expected Result: healthy. There is nothing that could be dropping invalidations, and the + * per-transport fields are omitted rather than reported as zeros. + */ + @Test + public void nullTransportReportsUp() { + withTransport(new NullTransport()); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + final Map data = data(result); + assertEquals("NullTransport", data.get("transport")); + assertFalse("a no-op transport has no invalidation state worth reporting", + data.containsKey("initialized")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the cache layer cannot be resolved at all, which is what a probe sees when + * it runs before the cache subsystem is up. + * Expected Result: healthy, not an exception and not a false alarm. + */ + @Test + public void unresolvableCacheLayerReportsUp() { + cacheLocatorMock.when(CacheLocator::getCacheAdministrator) + .thenThrow(new IllegalStateException("cache layer not up yet")); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + assertEquals("none", data(result).get("transport")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: a real transport that has never been initialized, polled inside the startup + * grace period. Every node is in this state for the first seconds of its life, because the + * transport is initialized late in boot when the cluster is wired. + * Expected Result: healthy, flagged as awaiting initialization. Reporting DOWN here would turn + * a normal rolling deployment into a DEGRADED overall status on every pod start. + */ + @Test + public void uninitializedTransportWithinGracePeriodReportsUp() { + Config.setProperty(GRACE_KEY, "120"); + withTransport(transport(false, 7L, 2000L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + final Map data = data(result); + assertEquals(false, data.get("initialized")); + assertEquals(true, data.get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the same never-initialized transport, but the grace period has elapsed. + * Expected Result: unhealthy. This is the #36803 fault the probe exists to catch -- a node + * that never wires its transport and drops every invalidation forever. + */ + @Test + public void uninitializedTransportPastGracePeriodReportsDown() { + Config.setProperty(GRACE_KEY, "0"); + withTransport(transport(false, 7L, 2000L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.DOWN, result.status()); + assertEquals(true, data(result).get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the transport was initialized on an earlier poll and is not anymore, with a + * long grace period configured. + * Expected Result: unhealthy immediately. The grace period covers startup only; losing a + * transport that had been working is a regression, not a node still booting. + */ + @Test + public void transportLostAfterInitializationReportsDownDespiteGracePeriod() { + Config.setProperty(GRACE_KEY, "3600"); + final CacheTransport transport = transport(true, 0L, 0L, 0L); + withTransport(transport); + final CacheTransportHealthCheck check = new CacheTransportHealthCheck(); + + assertEquals("baseline: a working transport is UP", HealthStatus.UP, check.check().status()); + + when(transport.isInitialized()).thenReturn(false); + + final HealthCheckResult result = check.check(); + assertEquals(HealthStatus.DOWN, result.status()); + assertEquals("the grace period must not apply once a transport has been seen working", + false, data(result).get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: an initialized transport on a node whose last rewire attempt failed once, + * below the configured threshold. + * Expected Result: healthy. {@code testCluster()} can throw on a momentary database hiccup + * while invalidations keep flowing, so a single stale failure must not report DOWN. + */ + @Test + public void rewireFailuresBelowThresholdReportUp() { + Config.setProperty(THRESHOLD_KEY, "3"); + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(1L); + withTransport(transport(true, 0L, 0L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + assertEquals(1L, data(result).get("rewireFailures")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: consecutive rewire failures have reached the threshold while the transport + * still reports itself initialized. + * Expected Result: unhealthy -- the transport may be wired to a stale view of the cluster. + */ + @Test + public void rewireFailuresAtThresholdReportDown() { + Config.setProperty(THRESHOLD_KEY, "3"); + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(3L); + withTransport(transport(true, 0L, 0L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.DOWN, result.status()); + assertEquals(3L, data(result).get("rewireFailures")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#buildStructuredData} + * Given Scenario: a healthy transport carrying non-zero counters. + * Expected Result: every counter monitoring is expected to consume is present, and the drops + * from before the transport came up are reported separately from the operational ones. + */ + @Test + public void structuredDataExposesEveryCounter() { + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(2L); + withTransport(transport(true, 5L, 2841L, 11L)); + + final Map data = data(new CacheTransportHealthCheck().check()); + + assertEquals(true, data.get("initialized")); + assertEquals(5L, data.get("droppedInvalidations")); + assertEquals("startup drops are reported apart from operational ones, so the alertable" + + " counter is not permanently dominated by the boot burst", + 2841L, data.get("startupDroppedInvalidations")); + assertEquals(11L, data.get("failedInvalidations")); + assertEquals(2L, data.get("rewireFailures")); + assertEquals(false, data.get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: a genuinely broken transport, with the check in its default mode. + * Expected Result: DEGRADED rather than DOWN. The default is MONITOR_MODE precisely so a + * broken transport cannot fail a readiness probe and drain a node that can still serve + * traffic; this test pins that deployment-safety property. + */ + @Test + public void defaultModeIsMonitorModeAndConvertsDownToDegraded() { + final CacheTransportHealthCheck check = new CacheTransportHealthCheck(); + assertEquals("the default must stay MONITOR_MODE", HealthCheckMode.MONITOR_MODE, + check.getDefaultMode()); + + Config.setProperty(MODE_KEY, HealthCheckMode.MONITOR_MODE.name()); + Config.setProperty(GRACE_KEY, "0"); + withTransport(transport(false, 12L, 0L, 0L)); + + final HealthCheckResult result = check.check(); + + assertEquals(HealthStatus.DEGRADED, result.status()); + assertTrue(result.monitorModeApplied()); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#isLivenessCheck()} + * Given Scenario: the probe registration flags are read by the health framework. + * Expected Result: never a liveness check. Restarting pods on transport failure is what + * amplified the incident in #36544. + */ + @Test + public void isNeverALivenessCheck() { + assertFalse(new CacheTransportHealthCheck().isLivenessCheck()); + } + + private void withTransport(final CacheTransport transport) { + final DotCacheAdministrator admin = mock(DotCacheAdministrator.class); + when(admin.getTransport()).thenReturn(transport); + cacheLocatorMock.when(CacheLocator::getCacheAdministrator).thenReturn(admin); + } + + private CacheTransport transport(final boolean initialized, final long dropped, + final long startupDropped, final long failed) { + final CacheTransport transport = mock(CacheTransport.class); + when(transport.isInitialized()).thenReturn(initialized); + when(transport.getDroppedMessages()).thenReturn(dropped); + when(transport.getStartupDroppedMessages()).thenReturn(startupDropped); + when(transport.getFailedMessages()).thenReturn(failed); + return transport; + } + + private Map data(final HealthCheckResult result) { + return result.data().orElseThrow( + () -> new AssertionError("the check must always publish structured data")); + } +} From aa267118c5a68460abe34f2ef0e7c271fc05d1eb Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Tue, 11 Aug 2026 08:19:43 -0600 Subject: [PATCH 6/8] fix(clustering): make the cache-transport init guard atomic (#36803) Review follow-up. init() guarded with a check-then-act -- read initialized, then set it several statements later -- so two threads could both see false and each run start() + subscribe(), double-subscribing the listener. That is the opposite of the churn the idempotent init() was added to remove. Not reachable through today's call graph: init() is reached only via ChainableCacheAdministratorImpl.setCluster() <- addMeToCacheIfNeeded() <- rewireCluster() <- ClusterFactory.rewireClusterIfNeeded(), each with exactly one call site, and the last is static synchronized. Every entry point (ServerHeartbeatJob, FreeServerFromClusterJob, ClusterFactory.initialize() from LicenseManager and the two startup tasks) funnels through that monitor. The verification the reviewer asked for, recorded here because the guarantee is not local to this class: rewireCluster() is public static and not itself synchronized, so a future caller could bypass the lock that makes the race unreachable today. init() and shutdown() are now synchronized. init() runs once per cluster rewire, so the monitor costs nothing. shutdown() shares it because a shutdown landing between init()'s subscribe() and its initialized.set(true) would stop the provider and then be overwritten back to initialized, leaving a transport that reports itself up with nothing listening. Deliberately NOT the compareAndSet form suggested in review (`if (!initialized.compareAndSet(false, true)) return;` with the work after it). Claiming the flag up front means a thrown start() stays flagged as initialized: isInitialized() reports true, shouldReinit() reports false, setCluster() never retries, and the new health check reports a healthy transport that never subscribed -- reintroducing the exact silent failure this issue removes. The flag stays set only after start() and subscribe() return. Two tests, both verified to fail without the change: - test_concurrent_init_starts_and_subscribes_exactly_once: 16 threads released together must produce one start() and one subscribe(). Fails 3/3 runs without the synchronization (observed 3 starts). - test_failed_init_leaves_the_transport_retryable: a throwing start() leaves isInitialized() false and shouldReinit() true, and the next rewire succeeds. This is the test that would fail under the compareAndSet-first form. Refs: #36803 --- .../cache/transport/PubSubCacheTransport.java | 35 ++++- .../transport/PubSubCacheTransportTest.java | 120 +++++++++++++++++- 2 files changed, 149 insertions(+), 6 deletions(-) diff --git a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java index 14ece9fb4915..098a4fea5c62 100644 --- a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java +++ b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java @@ -88,8 +88,31 @@ public PubSubCacheTransport() { Logger.debug(this.getClass(), "PubSubCacheTransport"); } + /** + * {@inheritDoc} + * + * {@code synchronized} so the guard below is a genuine test-and-set rather than a + * check-then-act: two threads could otherwise both read {@code initialized == false} and each + * run {@code start()} + {@code subscribe()}, double-subscribing the listener -- the opposite of + * the churn this transport was changed to remove. + * + * Today every path here is already serialized -- {@code init()} is reached only through + * {@code ChainableCacheAdministratorImpl.setCluster()} <- {@code addMeToCacheIfNeeded()} <- + * {@code rewireCluster()} <- {@code ClusterFactory.rewireClusterIfNeeded()}, which is + * {@code static synchronized}, and each of those has exactly one call site. This guards the + * future: {@code rewireCluster()} is {@code public static} and not itself synchronized, so a + * new caller could bypass the lock that currently makes the race unreachable. The method runs + * once per cluster rewire, so the monitor costs nothing. + * + * Note that {@code initialized} is deliberately set only *after* {@code start()} and + * {@code subscribe()} have returned. Marking it up front (for instance via a + * {@code compareAndSet} guard) would leave a thrown {@code start()} permanently flagged as + * initialized: {@link #isInitialized()} would report true, {@link #shouldReinit()} would report + * false, {@code setCluster()} would never retry, and the health check would report a healthy + * transport that never subscribed -- exactly the silent failure issue #36803 removes. + */ @Override - public void init(final Server localServer) throws CacheTransportException { + public synchronized void init(final Server localServer) throws CacheTransportException { if (this.initialized.get()) { Logger.debug(this.getClass(), "PubSubCacheTransport already initialized, skipping re-init"); @@ -212,8 +235,16 @@ public Map validateCacheInCluster(final int maxWaitInMilli return this.topic.readResponses(); } + /** + * {@inheritDoc} + * + * Shares {@link #init(Server)}'s monitor so the two cannot interleave. Without it a shutdown + * landing between {@code init()}'s {@code subscribe()} and its {@code initialized.set(true)} + * would stop the provider and then be overwritten back to initialized, leaving a transport that + * reports itself up with nothing listening. + */ @Override - public void shutdown() throws CacheTransportException { + public synchronized void shutdown() throws CacheTransportException { Logger.debug(this.getClass(), "shutdown()"); this.pubsub.stop(); if (initialized.get()) { diff --git a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java index 1b6de485b794..a45cc128f06f 100644 --- a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java +++ b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java @@ -7,6 +7,10 @@ import static org.junit.Assert.assertTrue; import com.dotcms.dotpubsub.NullDotPubSubProvider; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; /** @@ -125,14 +129,92 @@ public void test_init_is_idempotent() throws Exception { transport.init(null); transport.init(null); - assertEquals(1, provider.starts); + assertEquals(1, provider.starts()); assertTrue(transport.isInitialized()); // after shutdown, init() must run again transport.shutdown(); assertFalse(transport.isInitialized()); transport.init(null); - assertEquals(2, provider.starts); + assertEquals(2, provider.starts()); + } + + /** + * The guard in init() must be a test-and-set, not a check-then-act. Threads racing into init() + * all released at once must produce exactly one start()/subscribe() pair; without the + * synchronization, several can read initialized==false and each subscribe the listener, which + * is the double-subscribe the idempotency fix exists to prevent. + */ + @Test + public void test_concurrent_init_starts_and_subscribes_exactly_once() throws Exception { + final int threads = 16; + final CountingProvider provider = new CountingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + final CountDownLatch ready = new CountDownLatch(threads); + final CountDownLatch go = new CountDownLatch(1); + final List failures = new ArrayList<>(); + final List workers = new ArrayList<>(); + + for (int i = 0; i < threads; i++) { + final Thread t = new Thread(() -> { + try { + ready.countDown(); + go.await(); + transport.init(null); + } catch (Throwable e) { + synchronized (failures) { + failures.add(e); + } + } + }); + workers.add(t); + t.start(); + } + + ready.await(); + go.countDown(); + for (final Thread t : workers) { + t.join(); + } + + assertTrue("no thread may fail while initializing: " + failures, failures.isEmpty()); + assertEquals("the pubsub provider must be started exactly once", 1, provider.starts()); + assertEquals("and the topic subscribed exactly once", 1, provider.subscribeCount.get()); + assertTrue(transport.isInitialized()); + } + + /** + * A failed init() must stay retryable. This is why initialized is set only after start() and + * subscribe() return, rather than being claimed up front by a compareAndSet guard: marking it + * before the work would leave a thrown start() permanently flagged as initialized, so + * isInitialized() would report a healthy transport that never subscribed and shouldReinit() + * would stop setCluster() from ever retrying it. + */ + @Test + public void test_failed_init_leaves_the_transport_retryable() throws Exception { + final BrokenStartProvider provider = new BrokenStartProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + try { + transport.init(null); + throw new AssertionError("init() was expected to propagate the provider failure"); + } catch (IllegalStateException expected) { + // the provider could not reach its backend + } + + assertFalse("a failed init must not report itself initialized", transport.isInitialized()); + assertTrue("and must ask to be re-initialized", transport.shouldReinit()); + + // the next cluster rewire retries, and succeeds once the backend is reachable + provider.failing = false; + transport.init(null); + + assertTrue(transport.isInitialized()); + assertEquals("both the failed attempt and the successful retry reached the provider", + 2, provider.startAttempts); } /** @@ -196,11 +278,41 @@ public void test_getFailedMessages_includes_async_provider_failures() throws Exc } private static class CountingProvider extends NullDotPubSubProvider { - int starts = 0; + final AtomicInteger startCount = new AtomicInteger(0); + final AtomicInteger subscribeCount = new AtomicInteger(0); + + int starts() { + return startCount.get(); + } + + @Override + public com.dotcms.dotpubsub.DotPubSubProvider start() { + startCount.incrementAndGet(); + return this; + } + + @Override + public com.dotcms.dotpubsub.DotPubSubProvider subscribe( + final com.dotcms.dotpubsub.DotPubSubTopic topic) { + subscribeCount.incrementAndGet(); + return this; + } + } + + /** + * Mimics a provider whose backing connection is unavailable, so start() throws rather than + * returning. Flips to succeeding once {@link #failing} is cleared. + */ + private static class BrokenStartProvider extends NullDotPubSubProvider { + boolean failing = true; + int startAttempts = 0; @Override public com.dotcms.dotpubsub.DotPubSubProvider start() { - starts++; + startAttempts++; + if (failing) { + throw new IllegalStateException("cannot reach the pubsub backend"); + } return this; } } From b659148b62c5253a1297a935ecfd787bc5c25791 Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Tue, 11 Aug 2026 09:16:46 -0600 Subject: [PATCH 7/8] fix(clustering): normalize the pubsub failure-count topic key (#36803) Review follow-up. Writes into QueuingPubSubWrapper.failedByTopic and reads out of it arrive by routes that disagree on case: - recordFailure() keys on the event's topic, which DotPubSubEvent.Builder.withTopic has already lowercased (DotPubSubEvent.java:216) - PubSubCacheTransport.getFailedMessages() looks up this.topic.getTopic(), which is a bare String.valueOf(getKey()) with no normalization (DotPubSubTopic.java:26) They match today only because every current topic key is already lowercase (CacheTransportTopic.CACHE_TOPIC is "dotcache_topic"). A future key with one uppercase character would make the lookup miss silently, and getFailedPublishCount() would report zero failures while cluster invalidations were being lost -- the same class of blind spot this issue exists to remove. Normalized on both sides inside the wrapper, which owns the map, rather than with a .toLowerCase() at the single call site as suggested in review: that would leave every caller needing to know the wrapper's internal key convention, and would fix only the cache topic rather than any future one. DotPubSubProvider documents that the key is matched case-insensitively so other implementations follow suit. test_failure_lookup_is_case_insensitive records a failure through a mixed-case topic and reads it back three ways. Verified to fail without the fix, with exactly the silent under-report described above: AssertionError: and still found when the caller passes the un-normalized topic key expected:<1> but was:<0> Currently a latent fragility, not a live defect. Refs: #36803 --- .../dotcms/dotpubsub/DotPubSubProvider.java | 5 ++++ .../dotpubsub/QueuingPubSubWrapper.java | 28 +++++++++++++++++-- .../dotpubsub/QueuingPubSubWrapperTest.java | 26 +++++++++++++++++ 3 files changed, 56 insertions(+), 3 deletions(-) diff --git a/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java b/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java index 83300b773520..80959b6afd96 100644 --- a/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java +++ b/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java @@ -86,6 +86,11 @@ default String getProviderName() { * (cache invalidation, OSGi restart, cluster management), so a JVM-wide total could not be * attributed to the cache transport. * + * The topic key is matched case-insensitively, so a caller may pass either + * {@link DotPubSubEvent#getTopic()} (already lowercased by the event builder) or + * {@link DotPubSubTopic#getTopic()} (not normalized) and get the same answer. Implementations + * that key their own storage by topic must normalize it the same way. + * * @param topic the topic key, as returned by {@link DotPubSubEvent#getTopic()} * @return the failure count, or 0 for providers that publish synchronously */ diff --git a/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java b/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java index c7bef4d49da7..b9f8d1dd5284 100644 --- a/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java +++ b/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java @@ -31,6 +31,8 @@ public class QueuingPubSubWrapper implements DotPubSubProvider { /** * Failed publish attempts per topic. Keyed by topic because one provider instance serves * every topic in the JVM, so a single total could not be attributed to the cache transport. + * + * Keys are normalized through {@link #topicKey(String)}; see that method for why. */ private final Map failedByTopic = new ConcurrentHashMap<>(); @@ -143,8 +145,28 @@ private void publishAndRecordOutcome(final DotPubSubEvent event) { } private void recordFailure(final DotPubSubEvent event) { - failedByTopic.computeIfAbsent(String.valueOf(event.getTopic()), t -> new AtomicLong(0)) - .incrementAndGet(); + failedByTopic.computeIfAbsent(topicKey(String.valueOf(event.getTopic())), + t -> new AtomicLong(0)).incrementAndGet(); + } + + /** + * Normalizes a topic into the key {@link #failedByTopic} is indexed by. + * + * Writes and reads reach this map by different routes that do not agree on case. + * {@code DotPubSubEvent.Builder.withTopic} lowercases, so anything recorded from an event is + * already lowercase; the read side goes through {@code DotPubSubTopic.getTopic()}, which + * returns {@code String.valueOf(getKey())} with no normalization at all. Those happen to match + * today only because every current key is already lowercase + * ({@code CacheTransportTopic.CACHE_TOPIC} is {@code "dotcache_topic"}). + * + * A future topic key with an uppercase character would make the lookup miss silently, and + * {@link #getFailedPublishCount(String)} would report zero failures while invalidations were + * being lost -- precisely the kind of blind spot issue #36803 exists to remove. Normalizing on + * both sides here keeps the map self-consistent regardless of what a caller passes, rather than + * relying on every caller knowing this convention. + */ + private static String topicKey(final String topic) { + return topic == null ? "null" : topic.toLowerCase(); } /** @@ -156,7 +178,7 @@ private void recordFailure(final DotPubSubEvent event) { */ @Override public long getFailedPublishCount(final String topic) { - final AtomicLong failed = failedByTopic.get(topic); + final AtomicLong failed = failedByTopic.get(topicKey(topic)); return failed == null ? 0 : failed.get(); } diff --git a/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java b/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java index e13a468ccb59..cb90821098c4 100644 --- a/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java +++ b/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java @@ -113,6 +113,32 @@ public void test_successful_publishes_are_not_counted() throws Exception { assertEquals(0, wrapper.getFailedPublishCount("never-used-topic")); } + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: the failure is recorded from an event, whose topic + * {@code DotPubSubEvent.Builder.withTopic} has lowercased, but the lookup arrives with the + * topic key un-normalized -- which is what {@code DotPubSubTopic.getTopic()} returns, since it + * is a bare {@code String.valueOf(getKey())}. + * Expected Result: the count is found regardless of case. Without normalization this misses + * silently and reports zero failures while invalidations are being lost, which is the same + * class of blind spot #36803 removes. It happens to work today only because every current + * topic key is already lowercase. + */ + @Test + public void test_failure_lookup_is_case_insensitive() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new AlwaysFailsProvider()); + + wrapper.publish(event("DotCache_Topic", "inval-mixed-case")); + + awaitFailures(wrapper, "dotcache_topic", 1); + assertEquals("recorded under the lowercased key the event builder produced", + 1, wrapper.getFailedPublishCount("dotcache_topic")); + assertEquals("and still found when the caller passes the un-normalized topic key", + 1, wrapper.getFailedPublishCount("DotCache_Topic")); + assertEquals("any casing resolves to the same counter", + 1, wrapper.getFailedPublishCount("DOTCACHE_TOPIC")); + } + private static class AlwaysFailsProvider extends NullDotPubSubProvider { @Override public boolean publish(final DotPubSubEvent event) { From 24108d78e34f26041ca49f9e2ed75bfaa27b7485 Mon Sep 17 00:00:00 2001 From: "daniel.solis" Date: Tue, 11 Aug 2026 09:17:21 -0600 Subject: [PATCH 8/8] test(clustering): cover the cluster rewire-retry decision (#36803) Review follow-up: the rewire-retry logic was the least-protected behavioural change in this PR. Only the health check's consumption of REWIRE_FAILURES was tested, not the producer -- and the retry clause is what guarantees a failed cache-transport init keeps being retried, so a regression there would silently restore the "logged and forgotten" failure #36803 targets. It also could not be exercised on the live cluster: ServerHeartbeatJob.execute() calls LicenseUtil.updateLicenseHeartbeat() before rewireClusterIfNeeded(), and that throws when the database is down, so REWIRE_FAILURES can only move while the database is healthy and setCluster()/testCluster() fails -- a window that resisted local reproduction. The decision was an inline condition behind clusterReady(), isEnterprise() and two APILocator calls, so it is extracted into a pure, package-private shouldRewire(aliveServers, knownServers, currentServer, pendingFailures). Behaviour is unchanged; the rationale that was a comment on the condition is now the method's javadoc. Five cases, including the one that motivated the clause: a pending failure forces a retry even when membership has settled back to the stale KNOWN_SERVERS, where the membership comparison alone reports "nothing changed". That test asserts its own precondition first, so it cannot pass for the wrong reason, and it was verified to fail when the pendingFailures clause is removed. Also replaces the raw Collections.EMPTY_LIST on KNOWN_SERVERS with Collections.emptyList() while in this file. Refs: #36803 --- .../enterprise/cluster/ClusterFactory.java | 47 ++++++-- .../cluster/ClusterFactoryRewireTest.java | 112 ++++++++++++++++++ 2 files changed, 149 insertions(+), 10 deletions(-) create mode 100644 dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java diff --git a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java index 22ba7522f948..4d1b440294fa 100644 --- a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java +++ b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java @@ -65,7 +65,7 @@ public class ClusterFactory { private static boolean CLUSTER_INITED=false; - private static List KNOWN_SERVERS=Collections.EMPTY_LIST; + private static List KNOWN_SERVERS=Collections.emptyList(); /** * Consecutive cluster cache-transport rewire failures. Reset to zero on the first @@ -351,13 +351,8 @@ public static synchronized void rewireClusterIfNeeded() { try{ List aliveServers = APILocator.getServerAPI().getAliveServers(); - // A pending failure must always be retried. The membership comparison below only - // fires when the alive-server set changes, so without this a rewire that failed - // while membership then settled back to KNOWN_SERVERS would never be attempted - // again: the transport would stay broken and REWIRE_FAILURES could never return to - // zero, leaving the health check reporting DOWN indefinitely. - if (REWIRE_FAILURES.get() > 0 || !aliveServers.equals(KNOWN_SERVERS) || !aliveServers - .contains(APILocator.getServerAPI().getCurrentServer()) ) { + if (shouldRewire(aliveServers, KNOWN_SERVERS, + APILocator.getServerAPI().getCurrentServer(), REWIRE_FAILURES.get())) { rewireCluster(); } @@ -365,11 +360,43 @@ public static synchronized void rewireClusterIfNeeded() { catch(Exception e){ Logger.error(ClusterFactory.class, "Unable to rewire cluster:" + e.getMessage()); Logger.error(ClusterFactory.class, "servers:" + KNOWN_SERVERS); - - + + throw new DotRuntimeException(e); } } + + /** + * Decides whether {@link #rewireCluster()} should run on this heartbeat. + * + * Extracted from {@link #rewireClusterIfNeeded()} so it can be unit tested without the + * enterprise statics, {@code APILocator} and license state that method needs. Pure function of + * its arguments; behaviour is unchanged from the inline condition it replaces. + * + * A rewire is needed when any of the following holds: + * + * - {@code pendingFailures > 0} -- a previous rewire failed and must be retried. This is the + * clause added for issue #36803. The membership comparison below only fires when the + * alive-server set *changes*, so without this a rewire that failed while membership then + * settled back to {@code knownServers} would never be attempted again: the transport would + * stay broken, {@code REWIRE_FAILURES} could never return to zero, and the cache-transport + * health check would report unhealthy indefinitely. + * - cluster membership changed since the last successful rewire. + * - this server is missing from the alive set, so its own registration needs redoing. + * + * @param aliveServers servers currently seen as alive + * @param knownServers membership as of the last *successful* rewire + * @param currentServer this server + * @param pendingFailures consecutive rewire failures not yet cleared by a success + */ + @VisibleForTesting + static boolean shouldRewire(final List aliveServers, final List knownServers, + final Server currentServer, final long pendingFailures) { + + return pendingFailures > 0 + || !aliveServers.equals(knownServers) + || !aliveServers.contains(currentServer); + } public static void rewireCluster() throws Exception { diff --git a/dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java b/dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java new file mode 100644 index 000000000000..00040641e35b --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java @@ -0,0 +1,112 @@ +package com.dotcms.enterprise.cluster; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.dotcms.cluster.bean.Server; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.junit.Test; + +/** + * Unit tests for {@link ClusterFactory#shouldRewire}, the decision that governs whether a + * heartbeat re-runs the cluster rewire. + * + * This is the logic that guarantees a failed cache-transport init keeps being retried (issue + * #36803). It could not be exercised against a live cluster: {@code ServerHeartbeatJob.execute()} + * calls {@code LicenseUtil.updateLicenseHeartbeat()} before {@code rewireClusterIfNeeded()}, and + * that throws when the database is down, so the only way to move the failure counter is for the + * database to be healthy while {@code setCluster()}/{@code testCluster()} fails -- a window that + * resisted local reproduction. Hence these tests on the extracted predicate. + */ +public class ClusterFactoryRewireTest { + + private static final long NO_PENDING_FAILURES = 0L; + + private static Server server(final String id) { + return Server.builder().withServerId(id).withIpAddress("127.0.0." + id.length()).build(); + } + + private static final Server ME = server("me"); + private static final Server PEER = server("peer"); + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: membership is unchanged, this server is in the alive set, and no rewire + * failure is pending -- the steady state of a healthy cluster on every heartbeat. + * Expected Result: no rewire. Rewiring here is what produced the churn measured in #36544. + */ + @Test + public void steadyStateDoesNotRewire() { + final List alive = Arrays.asList(ME, PEER); + + assertFalse(ClusterFactory.shouldRewire(alive, Arrays.asList(ME, PEER), ME, + NO_PENDING_FAILURES)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: a previous rewire failed, so KNOWN_SERVERS was never advanced, and + * membership has since settled back to exactly that stale set. + * Expected Result: rewire anyway. This is the regression guard for the retry clause added in + * #36803 -- the membership comparison alone reports "nothing changed", so without the pending + * failure check the transport would stay broken forever, the failure counter could never + * return to zero, and the cache-transport health check would report unhealthy indefinitely. + */ + @Test + public void pendingFailureForcesRetryEvenWhenMembershipLooksUnchanged() { + final List alive = Arrays.asList(ME, PEER); + final List knownServers = Arrays.asList(ME, PEER); + + // sanity: membership genuinely looks unchanged, so only the failure counter can trigger it + assertFalse("precondition: membership must look identical for this test to mean anything", + ClusterFactory.shouldRewire(alive, knownServers, ME, NO_PENDING_FAILURES)); + + assertTrue(ClusterFactory.shouldRewire(alive, knownServers, ME, 1L)); + assertTrue("still retried after several failures", + ClusterFactory.shouldRewire(alive, knownServers, ME, 7L)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: a node joined since the last successful rewire. + * Expected Result: rewire, so the transport is wired to the new membership. + */ + @Test + public void membershipChangeRewires() { + assertTrue("a server joined", + ClusterFactory.shouldRewire(Arrays.asList(ME, PEER), Collections.singletonList(ME), + ME, NO_PENDING_FAILURES)); + + assertTrue("a server left", + ClusterFactory.shouldRewire(Collections.singletonList(ME), Arrays.asList(ME, PEER), + ME, NO_PENDING_FAILURES)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: this server is absent from the alive set, which is how a node that lost its + * own registration looks. + * Expected Result: rewire, so it registers itself again. + */ + @Test + public void missingSelfRewires() { + final List aliveWithoutMe = Collections.singletonList(PEER); + + assertTrue(ClusterFactory.shouldRewire(aliveWithoutMe, aliveWithoutMe, ME, + NO_PENDING_FAILURES)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: the very first heartbeat, when KNOWN_SERVERS is still the empty list it is + * initialized to and this server is already alive. + * Expected Result: rewire, so the cluster is wired at least once. + */ + @Test + public void firstHeartbeatRewires() { + assertTrue(ClusterFactory.shouldRewire(Collections.singletonList(ME), + Collections.emptyList(), ME, NO_PENDING_FAILURES)); + } +}