Fix cluster cache replication defaults

changelog/70779.fixed.md

@@ -0,0 +1 @@
+Align the cluster-ready restart test with the Raft log persistence semantics introduced by #70755.

doc/ref/cache/all/index.rst

@@ -12,6 +12,7 @@ For understanding and usage of the cache modules see the :ref:`cache` topic.
     :toctree:
     :template: autosummary.rst.tmpl
 
+    _cluster_replicator
     consul
     etcd_cache
     localfs

doc/ref/cache/all/salt.cache._cluster_replicator.rst

@@ -0,0 +1,5 @@
+salt.cache._cluster_replicator
+==============================
+
+.. automodule:: salt.cache._cluster_replicator
+    :members:

salt/cache/__init__.py

@@ -19,6 +19,58 @@
 log = logging.getLogger(__name__)
 
 
+# Thread-local re-entrancy guard for cross-master cache replication.
+# ``Cache.store`` / ``Cache.store_many`` consult this flag before
+# emitting a ``cluster/runner/delegate_cache_write`` event: a peer-side
+# handler (see ``salt/channel/server.py:_handle_delegate_cache_write``)
+# applies incoming replication writes by calling ``cache.store`` again
+# with the flag set, which must NOT re-broadcast or we'd build an
+# infinite fan-out loop across the cluster.  The ``originating_master``
+# check in the peer-side handler is the primary loop-safety guard; this
+# thread-local is the second line of defense.
+_REPLICATION_STATE = threading.local()
+
+
+def _set_already_replicated(flag):
+    """Mark the current thread as applying a peer-originated write."""
+    _REPLICATION_STATE.already_replicated = bool(flag)
+
+
+def _is_already_replicated():
+    return getattr(_REPLICATION_STATE, "already_replicated", False)
+
+
+def _resolve_bank_routing(routing, bank):
+    """
+    Longest-prefix-match lookup of a bank name against the routing map.
+
+    Replication routing is keyed by top-level bank name (``jobs``,
+    ``cluster/consensus``, ...) but cache writes land on compound
+    sub-banks (``jobs/loads``, ``cluster/consensus/<node>/cluster/log``).
+    A plain ``routing.get(bank, default)`` therefore misses the common
+    case; this helper resolves by:
+
+    1. Exact-match the full bank name.
+    2. Strip one ``/``-delimited segment at a time and retry.
+    3. If no prefix matches, return ``"replicate"`` (the historical default).
+
+    :param routing: Dict of ``{bank_prefix: "ring"|"replicate"|"none"}``,
+                    as returned by ``salt.config.get_cluster_cache_routing``.
+    :param bank:    Full bank name for the pending write.
+    :returns:       Resolved mode string.
+    """
+    if bank in routing:
+        return routing[bank]
+    parts = bank.split("/")
+    # Drop one trailing segment per iteration: a/b/c -> a/b -> a.
+    while len(parts) > 1:
+        parts.pop()
+        prefix = "/".join(parts)
+        if prefix in routing:
+            return routing[prefix]
+    return "replicate"
+
+
 def factory(opts, **kwargs):
     """
     Creates and returns the cache class.
@@ -76,6 +128,11 @@ def __init__(self, opts, cachedir=None, **kwargs):
 
     @cached_property
     def modules(self):
+        # Honor a pre-populated ``_modules`` dict so tests and
+        # non-loader-driven callers can inject a mock driver without
+        # touching the cached_property machinery.
+        if self._modules is not None:
+            return self._modules
         return salt.loader.cache(self.opts)
 
     @cached_property
@@ -129,7 +186,7 @@ def cache(self, bank, key, fun, loop_fun=None, **kwargs):
 
         return data
 
-    def store(self, bank, key, data, expires=None):
+    def store(self, bank, key, data, expires=None, _already_replicated=False):
         """
         Store data using the specified module
 
@@ -149,25 +206,225 @@ def store(self, bank, key, data, expires=None):
         :param expires:
             how many seconds from now the data should be considered stale.
 
+        :param _already_replicated:
+            Internal flag used by the cluster-cache replication facade.
+            When True the local write proceeds but no cross-master
+            fan-out is performed.  The peer-side handler passes this
+            when applying an incoming replication event so the same
+            write doesn't bounce back to every peer again.  Not part
+            of the public API.
+
          :raises SaltCacheError:
             Raises an exception if cache driver detected an error accessing data
             in the cache backend (auth, permissions, etc).
         """
         fun = f"{self.driver}.store"
         try:
-            return self.modules[fun](bank, key, data, expires=expires, **self.kwargs)
+            result = self.modules[fun](bank, key, data, expires=expires, **self.kwargs)
         except TypeError:
             # if the backing store doesnt natively support expiry, we handle it as a fallback
             if expires:
                 expires_at = datetime.datetime.now().astimezone() + datetime.timedelta(
                     seconds=expires
                 )
                 expires_at = int(expires_at.timestamp())
-                return self.modules[fun](
+                result = self.modules[fun](
                     bank, key, {"data": data, "_expires": expires_at}, **self.kwargs
                 )
             else:
-                return self.modules[fun](bank, key, data, **self.kwargs)
+                result = self.modules[fun](bank, key, data, **self.kwargs)
+        # Cross-master replication fan-out at the facade.  Guarded so
+        # non-clustered masters and shared-FS clusters pay no overhead.
+        # The peer-side re-dispatch arrives with
+        # ``_already_replicated=True`` and skips this block, breaking
+        # the loop (combined with the ``originating_master`` check in
+        # the handler).
+        if not _already_replicated and not _is_already_replicated():
+            self._maybe_replicate_write(bank, key, data, expires)
+        return result
+
+    def store_many(self, bank, items, expires=None, _already_replicated=False):
+        """
+        Store multiple ``(key, data)`` pairs in a single bank.
+
+        A thin wrapper over :meth:`store` that applies the same
+        cross-master replication semantics.  The default
+        implementation loops; drivers that expose a bulk store
+        primitive can still accelerate the per-key hot path via the
+        underlying driver module (the facade intentionally keeps the
+        fan-out per-key so replication-ordering stays well-defined).
+
+        :param bank: The bank name.
+        :param items: Iterable of ``(key, data)`` tuples.
+        :param expires: Optional seconds-from-now expiry applied to
+            every stored entry.
+        :param _already_replicated: Internal flag; see :meth:`store`.
+        """
+        # Suppress replication during the inner per-key stores and
+        # emit a single batched replication event after the local
+        # writes land.  This keeps the wire-cost proportional to
+        # batch count rather than batch size.
+        if isinstance(items, dict):
+            pairs = list(items.items())
+        else:
+            pairs = list(items)
+        token_prev = _is_already_replicated()
+        _set_already_replicated(True)
+        try:
+            # Driver-native batch (e.g. ``mmap_cache.store_many``) wins
+            # one flock + batched fsync/msync across the whole batch.
+            # Drivers without a native batch fall back to the per-item
+            # loop so backwards compatibility is preserved.
+            fun = f"{self.driver}.store_many"
+            if fun in self.modules:
+                try:
+                    self.modules[fun](bank, pairs, expires=expires, **self.kwargs)
+                except TypeError:
+                    # Driver's store_many doesn't accept expires=; fall
+                    # back to per-item store so expiry semantics stay
+                    # uniform with the facade's envelope.
+                    for key, data in pairs:
+                        self.store(
+                            bank, key, data, expires=expires, _already_replicated=True
+                        )
+            else:
+                for key, data in pairs:
+                    self.store(
+                        bank, key, data, expires=expires, _already_replicated=True
+                    )
+        finally:
+            _set_already_replicated(token_prev)
+        if not _already_replicated and not token_prev:
+            self._maybe_replicate_write_many(bank, pairs, expires)
+
+    def _maybe_replicate_write(self, bank, key, data, expires):
+        """
+        Fan a local cache write out to cluster peers.
+
+        Consults ``cluster_cache_replication`` (master switch) and
+        ``cluster_cache_routing[bank]`` (per-bank routing) to decide
+        between three outcomes:
+
+        - **replicate**: emit ``cluster/runner/delegate_cache_write``
+          (handled by ``MasterPubServerChannel.publish_payload`` ->
+          ``publish_cache_write`` -> every peer).  N-of-N fan-out.
+        - **ring**: delegate to the existing HashRing owner via the
+          ``cluster/runner/delegate_write`` event (1-of-N).  Kept
+          for backward compatibility of the ``jobs`` bank.
+        - **skip**: non-clustered master, shared-FS mode, or an
+          unrecognised routing mode -- the local write stands alone.
+
+        TODO: partial-peer failures are non-destructive today (the
+        originating master's local write already succeeded); catch-up
+        of a dropped peer is the state-sync expert's lane.  See
+        ``agents/scratch/cache-replicate-primary.md``.
+        """
+        opts = self.opts or {}
+        if not opts.get("cluster_id"):
+            return
+        # Isolated-FS is the only mode where peers can't see each
+        # other's local-store writes; shared-FS clusters are a no-op.
+        if not opts.get("cluster_isolated_filesystem"):
+            return
+        # Default-on when the cluster+isolated-FS preconditions are
+        # met; operators can force False.  Mirrors the resolution in
+        # ``apply_master_config`` for callers that build opts without
+        # going through the config loader.
+        if opts.get("cluster_cache_replication") is False:
+            return
+        from salt.config import (
+            get_cluster_cache_routing,  # pylint: disable=import-outside-toplevel
+        )
+
+        routing = get_cluster_cache_routing(opts)
+        mode = _resolve_bank_routing(routing, bank)
+        if mode == "none":
+            # Explicit opt-out: bank is replicated by a dedicated
+            # transport (Raft for cluster/consensus, state-sync for
+            # cluster/state_sync).  Fan-out here would overwrite the
+            # authoritative copy on peers.
+            return
+        if mode == "ring":
+            self._fire_ring_delegate(bank, key, data, expires)
+        elif mode == "replicate":
+            self._fire_replicate(bank, key, data, expires)
+        else:
+            log.warning(
+                "cluster-cache-replication: unknown routing %r for bank %r; "
+                "treating as 'replicate'",
+                mode,
+                bank,
+            )
+            self._fire_replicate(bank, key, data, expires)
+
+    def _maybe_replicate_write_many(self, bank, pairs, expires):
+        """
+        Batched replication: emit ONE event carrying the whole list
+        (split into sub-batches of ``cluster_cache_replicate_batch_size``
+        if the batch exceeds that threshold).
+
+        The cluster + isolated-FS + routing gate is evaluated once
+        for the whole batch: all items share ``bank`` so routing is
+        uniform.  This cuts per-write event overhead on hot bulk
+        paths (e.g. ``resource_grains`` convergence).
+        """
+        if not pairs:
+            return
+        opts = self.opts or {}
+        if not opts.get("cluster_id"):
+            return
+        if not opts.get("cluster_isolated_filesystem"):
+            return
+        if opts.get("cluster_cache_replication") is False:
+            return
+        from salt.config import (
+            get_cluster_cache_routing,  # pylint: disable=import-outside-toplevel
+        )
+
+        routing = get_cluster_cache_routing(opts)
+        mode = _resolve_bank_routing(routing, bank)
+        if mode == "none":
+            return
+        if mode not in ("replicate", "ring"):
+            log.warning(
+                "cluster-cache-replication: unknown routing %r for bank %r; "
+                "treating as 'replicate'",
+                mode,
+                bank,
+            )
+            mode = "replicate"
+        batch_size = int(opts.get("cluster_cache_replicate_batch_size", 1024) or 1024)
+        if batch_size <= 0:
+            batch_size = 1024
+        from salt.cache._cluster_replicator import (  # pylint: disable=import-outside-toplevel
+            fire_delegate_cache_write_many,
+            fire_peer_cache_write_many,
+        )
+
+        fire = (
+            fire_peer_cache_write_many
+            if mode == "replicate"
+            else fire_delegate_cache_write_many
+        )
+        for start in range(0, len(pairs), batch_size):
+            chunk = pairs[start : start + batch_size]
+            fire(self.opts, bank, chunk, expires=expires)
+
+    def _fire_replicate(self, bank, key, data, expires):
+        """Fan this write out to every peer master (see replicator module)."""
+        from salt.cache._cluster_replicator import (  # pylint: disable=import-outside-toplevel
+            fire_peer_cache_write,
+        )
+
+        fire_peer_cache_write(self.opts, bank, key, data, expires)
+
+    def _fire_ring_delegate(self, bank, key, data, expires):
+        """Route this write to the HashRing owner (see replicator module)."""
+        from salt.cache._cluster_replicator import (  # pylint: disable=import-outside-toplevel
+            fire_delegate_cache_write,
+        )
+
+        fire_delegate_cache_write(self.opts, bank, key, data, expires)
 
     def fetch(self, bank, key):
         """
@@ -479,13 +736,18 @@ def fetch(self, bank, key):
             storage[(bank, key)] = [now, self.expire, data]
         return data
 
-    def store(self, bank, key, data, expires=None):
+    def store(self, bank, key, data, expires=None, _already_replicated=False):
         storage = self.storage
         with MemCache._lock:
             storage.pop((bank, key), None)
         # ``super().store`` calls the driver (disk / db); keep it off the
         # lock so concurrent stores don't serialise on returner I/O.
-        super().store(bank, key, data, expires=expires)
+        # Only forward the replication flag when it's set to True so
+        # legacy mock-based tests keep seeing the original call shape.
+        if _already_replicated:
+            super().store(bank, key, data, expires=expires, _already_replicated=True)
+        else:
+            super().store(bank, key, data, expires=expires)
         with MemCache._lock:
             if len(storage) >= self.max:
                 if self.cleanup: