fix(clustering): surface silent cache-transport failures and make rewire retryable (#36803)

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;
 
@@ -64,7 +65,14 @@ public class ClusterFactory {
 
 
     private static boolean CLUSTER_INITED=false;
-	private static List<Server> KNOWN_SERVERS=Collections.EMPTY_LIST;
+	private static List<Server> KNOWN_SERVERS=Collections.emptyList();
+
+    /**
+     * 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, "
@@ -343,32 +351,67 @@ public static synchronized void rewireClusterIfNeeded() {
 		try{
 			List<Server> aliveServers = APILocator.getServerAPI().getAliveServers();
 
-            if (!aliveServers.equals(KNOWN_SERVERS) || !aliveServers
-                    .contains(APILocator.getServerAPI().getCurrentServer()) ) {
+            if (shouldRewire(aliveServers, KNOWN_SERVERS,
+                    APILocator.getServerAPI().getCurrentServer(), REWIRE_FAILURES.get())) {
 
 				rewireCluster();
 			}
 		}
 		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<Server> aliveServers, final List<Server> knownServers,
+                                final Server currentServer, final long pendingFailures) {
+
+        return pendingFailures > 0
+                || !aliveServers.equals(knownServers)
+                || !aliveServers.contains(currentServer);
+    }
 
     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 +421,30 @@ 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, zero when the last rewire succeeded.
+     *
+     * @see #REWIRE_FAILURES
+     */
+    public static long getRewireFailures() {
+        return REWIRE_FAILURES.get();
+    }
 
      private static boolean isEnterprise() {
 

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,41 +34,169 @@ public class PubSubCacheTransport implements CacheTransport {
 
     final AtomicBoolean initialized = new AtomicBoolean(false);
 
+    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
+     * {@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);
+
     @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");
     }
 
+    /**
+     * {@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");
+            return;
+        }
 
         Logger.info(this.getClass(), "initing PubSubCacheTransport");
         this.pubsub.start();
         this.pubsub.subscribe(topic);
 
         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
     public void send(final String message) throws CacheTransportException {
         if (!this.initialized.get()) {
+            final long dropped = this.droppedMessages.incrementAndGet();
+            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 {
 
@@ -104,8 +235,16 @@ public Map<String, Serializable> 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()) {
@@ -125,6 +264,35 @@ 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}
+     *
+     * 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() {