28705436176

Committed 04 Jul 2026 11:56AM UTC coverage: 90.132% (-0.3%) from 90.399%

Build # 28705436176

Build Type

push

github

Committed by

Hyshmily

Commit Message

Merge remote-tracking branch 'hotkey/master'

Coverage Stats

1261 of 1453 branches covered (86.79%)

Branch coverage included in aggregate %.

3589 of 3928 relevant lines covered (91.37%)

4.18 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.6

/common/src/main/java/io/github/hyshmily/hotkey/sync/local/CacheSyncListener.java

/*
 * Copyright 2026 Hyshmily. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.github.hyshmily.hotkey.sync.local;

import static io.github.hyshmily.hotkey.sync.local.SyncMessage.*;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.Cache;
import com.rabbitmq.client.Channel;
import io.github.hyshmily.hotkey.Internal;
import io.github.hyshmily.hotkey.cache.cachesupport.CacheExpireManager;
import io.github.hyshmily.hotkey.cache.loader.CacheLoader;
import io.github.hyshmily.hotkey.model.CacheEntry;
import io.github.hyshmily.hotkey.model.KeyState;
import io.github.hyshmily.hotkey.rule.RuleMatcher;
import io.github.hyshmily.hotkey.sync.worker.WorkerListener;
import io.github.hyshmily.hotkey.util.DelayUtil;
import io.github.hyshmily.hotkey.util.version.VersionGuard;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;

/**
 * Listens for cache synchronization messages (INVALIDATE / REFRESH / RULES_SYNC) from
 * peer application instances via the {@code hotkey.sync.exchange} FanoutExchange.
 *
 * <p>This is the inbound half of the instance-to-instance cache coherence protocol.
 * The outbound half is {@link CacheSyncPublisher}. Together they ensure that a data
 * mutation on one instance is propagated to all peers.
 *
 * <p><b>Message routing:</b>
 * <ul>
 *   <li><b>INVALIDATE</b> ({@link SyncMessage#TYPE_INVALIDATE}): Atomically removes
 *       the specified key from the local Caffeine cache. Uses the 4-case
 *       {@link VersionGuard#shouldSkipForSync} guard to prevent stale degraded
 *       invalidations from wiping out a healthy entry.</li>
 *   <li><b>INVALIDATE_ALL</b> ({@link SyncMessage#TYPE_INVALIDATE_ALL}): Batch-removes
 *       all keys in the JSON-array payload via {@code caffeineCache.invalidateAllLocal()}.
 *       Bypasses version guards — these are unconditional operations.</li>
 *   <li><b>REFRESH</b> ({@link SyncMessage#TYPE_REFRESH}): Loads the latest value from
 *       Redis and updates the local cache entry. Preserves existing metadata (TTLs,
 *       key state, decision version, degradation flag) except for the value and
 *       data version. Uses double-checked locking (DCL) with {@link VersionGuard}
 *       to prevent stale overwrites.</li>
 *   <li><b>RULES_SYNC</b> ({@link SyncMessage#TYPE_RULES_SYNC}): Merges the incoming
 *       rule set into the local {@link RuleMatcher}, guarded by {@code rulesVersion}.</li>
 * </ul>
 *
 * <p><b>Thread safety:</b> All cache mutations use
 * {@link com.github.benmanes.caffeine.cache.Cache#asMap()}{@code .compute()} for
 * atomic per-key updates. The AMQP ack is sent before the cache mutation
 * (ack-before-update pattern, see ADR-0004); the mutation is scheduled with random
 * jitter to spread Redis load across instances.
 *
 * @see CacheSyncPublisher
 * @see SyncMessage
 * @see WorkerListener
 */
@RequiredArgsConstructor
@Slf4j
@Internal
public class CacheSyncListener {

  /** Shared Jackson {@link ObjectMapper} for deserializing batch-invalidation key lists from JSON. */
  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

  /** Local Caffeine L1 cache — target of invalidation and refresh operations.
   * Accessed atomically via {@code asMap().compute()} for thread-safe updates. */
  private final Cache<String, Object> caffeineCache;

  /** Loads the current value from Redis given a cache key.
   * Used during REFRESH to fetch the authoritative value before writing to L1. */
  private final CacheLoader redisLoader;

  /** Configuration for sync exchange name, jitter settings, and consumer concurrency. */
  private final CacheSyncProperties properties;

  /** Scheduler for running jitter-delayed cache update tasks, spreading Redis load.
   * Supplied externally to allow shared-pool reuse across listeners. */
  private final ScheduledExecutorService scheduler;

  /** Computes hard and soft expiry timestamps for refreshed entries. */
  private final CacheExpireManager expireManager;

  /** Hot-key rule matcher whose rule set is updated when a RULES_SYNC message arrives. */
  private final RuleMatcher ruleMatcher;

  /**
   * RabbitMQ message callback for incoming sync messages. Acknowledges the message
   * immediately after parsing (ack-before-update), then schedules the actual cache
   * mutation asynchronously with a random jitter to spread Redis load across peers.
   *
   * <p>On success, the message is acknowledged via {@link Channel#basicAck}. On any
   * processing exception (parse failure, routing failure), the message is negatively
   * acknowledged with {@code requeue=false} to prevent poison-message loops. The next
   * application-level write will re-broadcast the operation.
   *
   * @param channel the AMQP channel used for ack/nack operations
   * @param msg     the raw AMQP message whose body and headers carry the sync payload;
   *                must not be null
   * @throws IOException if the channel's basicAck or basicNack call fails
   */
  public void handleSyncMessage(Channel channel, Message msg) throws IOException {
    long tag = msg.getMessageProperties().getDeliveryTag();
    try {
      processSync(msg);
      channel.basicAck(tag, false);
    } catch (Exception e) {
      log.error("CacheSync processing failed: body={}", new String(msg.getBody()), e);
      channel.basicNack(tag, false, false);
    }
  }

  /**
   * Decodes the raw AMQP message into a {@link SyncMessage} and schedules the
   * appropriate handler to run after a random delay within
   * {@link CacheSyncProperties#getWarmupJitterMs()}. The jitter spreads Redis
   * reads when multiple peers process the same sync broadcast simultaneously.
   *
   * @param msg the raw AMQP message; if the body is null, empty, or cannot be
   *            parsed into a valid {@link SyncMessage}, the message is silently
   *            dropped without scheduling any task
   */
  private void processSync(Message msg) {
    SyncMessage sm = SyncMessage.from(msg);
    if (sm == null) {
      log.debug("Received sync message with empty or invalid body");
      return;
    }

    Runnable task = () -> {
      try {
        syncMessageRouter(sm);
      } catch (Exception e) {
        log.error("Async sync task failed: type={}, key={}, version={}", sm.type(), sm.cacheKey(), sm.version(), e);
      }
    };

    // Random jitter spreads out Redis fetches, reducing hotspot load during batch sync.
    DelayUtil.floatTimeDelay(task, properties.getWarmupJitterMs(), scheduler);
  }

  /**
   * Routes the deserialized {@link SyncMessage} to the appropriate handler based
   * on its type field. Delegates to {@link #handleLocalInvalidate},
   * {@link #handleLocalInvalidateAll}, {@link #handleRefresh}, or
   * {@link #handleRulesSync} accordingly.
   *
   * @param msg the deserialized sync message to route; must not be null
   */
  private void syncMessageRouter(SyncMessage msg) {
    if (msg.type() == null) {
      log.debug("Received sync with null type, skipping");
      return;
    }
    switch (msg.type()) {
      case TYPE_INVALIDATE -> handleLocalInvalidate(msg);
      case TYPE_INVALIDATE_ALL -> handleLocalInvalidateAll(msg);
      case TYPE_REFRESH -> handleRefresh(msg);
      case TYPE_RULES_SYNC -> handleRulesSync(msg);
      default -> log.warn("Unknown sync type: {}, cacheKey: {}", msg.type(), msg.cacheKey());
    }
  }

  /**
   * Atomically removes the specified key from the local cache in response to
   * an INVALIDATE sync message from a peer instance.
   *
   * <p><b>Version guard logic:</b>
   * <ul>
   *   <li><b>Unconditional path:</b> When {@code version == 0L && !isVersionDegraded}
   *       (clean invalidation from {@code invalidateAllLocal}), the guard is bypassed
   *       entirely — the entry is always removed.</li>
   *   <li><b>Guarded path:</b> Uses {@link VersionGuard#shouldSkipForSync} with the
   *       4-case degraded comparison. Case 2 (existing normal, incoming degraded)
   *       prevents a stale degraded INVALIDATE from wiping a healthy entry.</li>
   * </ul>
   *
   * <p>Double-checked locking (DCL): a fast version guard before the atomic
   * {@code compute} (first pass), and a second guard inside the {@code compute}
   * body (second pass) to prevent a concurrent REFRESH from being wiped by a
   * stale invalidate that arrived after the refresh.
   *
   * @param sm the sync message containing the key to invalidate; if the key
   *           is null or invalid, the invalidation is silently skipped
   */
  private void handleLocalInvalidate(SyncMessage sm) {
    boolean unconditional = sm.version() == 0L && !sm.isVersionDegraded();

    if (
      !unconditional &&
      VersionGuard.shouldSkipForSync(caffeineCache, sm.cacheKey(), sm.version(), sm.isVersionDegraded())
    ) {
      log.debug("Stale invalidate ignored: key={}, incomingVersion={}", sm.cacheKey(), sm.version());
      return;
    }

    caffeineCache
      .asMap()
      .compute(sm.cacheKey(), (key, existing) -> {
        if (
          !unconditional &&
          existing instanceof CacheEntry ce &&
          VersionGuard.shouldSkipForSync(ce, sm.version(), sm.isVersionDegraded())
        ) {
          return existing;
        }
        return null;
      });
    log.debug("Invalidated by sync: {}", sm.cacheKey());
  }

  /**
   * Batch-invalidates all keys contained in the JSON-array body of the sync message.
   *
   * <p>This method intentionally bypasses version guards. The publisher
   * ({@link CacheSyncPublisher#broadcastLocalInvalidateAll}) always sends clean
   * messages (version=0L, not degraded) and all keys are removed unconditionally.
   * This is more efficient than sending individual INVALIDATE messages for each key.
   *
   * <p>Deserialization failures (malformed JSON) are logged at ERROR level and
   * do not propagate.
   *
   * @param sm the sync message whose {@code cacheKey} field contains the JSON-array
   *           of keys to invalidate; must not be null
   */
  private void handleLocalInvalidateAll(SyncMessage sm) {
    try {
      List<String> keys = OBJECT_MAPPER.readValue(sm.cacheKey(), new TypeReference<>() {});
      caffeineCache.invalidateAll(keys);
      log.debug("Batch invalidated {} keys", keys.size());
    } catch (Exception e) {
      log.error("Failed to deserialize batch invalidate keys", e);
    }
  }

  /**
   * Merges the incoming rule set from a RULES_SYNC message into the local
   * {@link RuleMatcher}, guarded by the message's {@code rulesVersion}.
   * <p>
   * Delegates to {@link RuleMatcher#syncRules}, which handles the actual
   * merge logic and version conflict resolution.
   *
   * @param sm the sync message whose {@code cacheKey} field contains the
   *           serialized rule-set JSON and whose {@code rulesVersion} field
   *           carries the version for conflict resolution; must not be null
   */
  private void handleRulesSync(SyncMessage sm) {
    ruleMatcher.syncRules(sm.cacheKey(), sm.rulesVersion());
  }

  /**
   * Refreshes a cache entry with the latest value from Redis in response to a
   * REFRESH sync message from a peer instance.
   *
   * <p><b>Refresh flow:</b>
   * <ol>
   *   <li><b>DCL check 1:</b> Fast-path version guard ({@link VersionGuard#shouldSkipForSync})
   *       against the existing L1 entry. If a newer dataVersion is already present,
   *       the refresh is skipped.</li>
   *   <li><b>Redis fetch:</b> Loads the authoritative value from Redis.</li>
   *   <li><b>DCL check 2:</b> Second version guard inside the atomic {@code compute}
   *       to prevent overwriting a newer version that arrived during the Redis fetch.</li>
   *   <li><b>Write:</b> Replaces the value and dataVersion while preserving the existing
   *       entry's metadata (hard/soft TTLs, normal TTLs, key state, decision version,
   *       degradation flag). If no entry existed in L1 before the refresh, a fresh
   *       {@link CacheEntry} is created with default metadata and {@code KeyState.NORMAL}.</li>
   * </ol>
   *
   * <p>If the key is absent from L1 and Redis returns null (key does not exist),
   * the refresh is aborted — there is nothing to cache.
   *
   * @param sm the sync message containing the key and version to refresh;
   *           must not be null
   */
  private void handleRefresh(SyncMessage sm) {
    // DCL first check – cheap, outside the compute lock
    if (VersionGuard.shouldSkipForSync(caffeineCache, sm.cacheKey(), sm.version(), sm.isVersionDegraded())) {
      log.debug("Stale refresh ignored: key={}, incomingVersion={}", sm.cacheKey(), sm.version());
      return;
    }

    Object value = loadFromRedis(sm);
    if (value == null) {
      log.warn("Refresh failed to load value from Redis for key={}", sm.cacheKey());
      return;
    }

    caffeineCache
      .asMap()
      .compute(sm.cacheKey(), (key, existing) -> {
        // DCL second check – atomic with to write
        if (
          existing instanceof CacheEntry ce && VersionGuard.shouldSkipForSync(ce, sm.version(), sm.isVersionDegraded())
        ) {
          return existing;
        }

        if (existing instanceof CacheEntry cacheEntry) {
          long hardExpireAt = expireManager.computeHardExpireAt(cacheEntry.getHardTtlMs());
          long softExpireAt = expireManager.computeSoftExpireAt(cacheEntry.getSoftTtlMs());
          return cacheEntry
            .toBuilder()
            .value(value)
            .dataVersion(sm.version())
            .isVersionDegraded(sm.isVersionDegraded())
            .hardExpireAtMs(hardExpireAt)
            .softExpireAtMs(softExpireAt)
            .build();
        }
        long defaultHardTtlMs = expireManager.getEffectiveHardTtlMs();
        long defaultSoftTtlMs = expireManager.getEffectiveSoftTtlMs();
        //Entry was evicted from L1 — create fresh CacheEntry with default metadata
        return expireManager.applyTtl(
          CacheEntry.builder()
            .value(value)
            .dataVersion(sm.version())
            .isVersionDegraded(sm.isVersionDegraded())
            .decisionVersion(0L)
            .keyState(KeyState.NORMAL)
            .normalHardTtlMs(defaultHardTtlMs)
            .normalSoftTtlMs(defaultSoftTtlMs)
            .build(),
          defaultHardTtlMs,
          defaultSoftTtlMs
        );
      });
    log.debug("Refreshed by sync: {}", sm.cacheKey());
  }

  /**
   * Loads the current value from Redis for the key carried in the sync message.
   * <p>
   * Any exception thrown by the {@code redisLoader} (connection timeout, Redis
   * outage, serialization error) is caught and logged at WARN level. The caller
   * should handle a {@code null} return by aborting the refresh.
   *
   * @param sm the sync message containing the cache key to load; must not be null
   * @return the value from Redis, or {@code null} if the key is absent in Redis
   *         or the load failed with an exception
   */
  private Object loadFromRedis(SyncMessage sm) {
    try {
      return redisLoader.load(sm.cacheKey());
    } catch (Exception e) {
      log.warn("handleRefresh: Redis load failed for key={}", sm.cacheKey(), e);
      return null;
    }
  }
}

1	/*
2	* Copyright 2026 Hyshmily. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package io.github.hyshmily.hotkey.sync.local;
17
18	import static io.github.hyshmily.hotkey.sync.local.SyncMessage.*;
19
20	import com.fasterxml.jackson.core.type.TypeReference;
21	import com.fasterxml.jackson.databind.ObjectMapper;
22	import com.github.benmanes.caffeine.cache.Cache;
23	import com.rabbitmq.client.Channel;
24	import io.github.hyshmily.hotkey.Internal;
25	import io.github.hyshmily.hotkey.cache.cachesupport.CacheExpireManager;
26	import io.github.hyshmily.hotkey.cache.loader.CacheLoader;
27	import io.github.hyshmily.hotkey.model.CacheEntry;
28	import io.github.hyshmily.hotkey.model.KeyState;
29	import io.github.hyshmily.hotkey.rule.RuleMatcher;
30	import io.github.hyshmily.hotkey.sync.worker.WorkerListener;
31	import io.github.hyshmily.hotkey.util.DelayUtil;
32	import io.github.hyshmily.hotkey.util.version.VersionGuard;
33	import java.io.IOException;
34	import java.util.List;
35	import java.util.concurrent.ScheduledExecutorService;
36	import lombok.RequiredArgsConstructor;
37	import lombok.extern.slf4j.Slf4j;
38	import org.springframework.amqp.core.Message;
39
40	/**
41	* Listens for cache synchronization messages (INVALIDATE / REFRESH / RULES_SYNC) from
42	* peer application instances via the {@code hotkey.sync.exchange} FanoutExchange.
43	*
44	* <p>This is the inbound half of the instance-to-instance cache coherence protocol.
45	* The outbound half is {@link CacheSyncPublisher}. Together they ensure that a data
46	* mutation on one instance is propagated to all peers.
47	*
48	* <p><b>Message routing:</b>
49	* <ul>
50	* <li><b>INVALIDATE</b> ({@link SyncMessage#TYPE_INVALIDATE}): Atomically removes
51	* the specified key from the local Caffeine cache. Uses the 4-case
52	* {@link VersionGuard#shouldSkipForSync} guard to prevent stale degraded
53	* invalidations from wiping out a healthy entry.</li>
54	* <li><b>INVALIDATE_ALL</b> ({@link SyncMessage#TYPE_INVALIDATE_ALL}): Batch-removes
55	* all keys in the JSON-array payload via {@code caffeineCache.invalidateAllLocal()}.
56	* Bypasses version guards — these are unconditional operations.</li>
57	* <li><b>REFRESH</b> ({@link SyncMessage#TYPE_REFRESH}): Loads the latest value from
58	* Redis and updates the local cache entry. Preserves existing metadata (TTLs,
59	* key state, decision version, degradation flag) except for the value and
60	* data version. Uses double-checked locking (DCL) with {@link VersionGuard}
61	* to prevent stale overwrites.</li>
62	* <li><b>RULES_SYNC</b> ({@link SyncMessage#TYPE_RULES_SYNC}): Merges the incoming
63	* rule set into the local {@link RuleMatcher}, guarded by {@code rulesVersion}.</li>
64	* </ul>
65	*
66	* <p><b>Thread safety:</b> All cache mutations use
67	* {@link com.github.benmanes.caffeine.cache.Cache#asMap()}{@code .compute()} for
68	* atomic per-key updates. The AMQP ack is sent before the cache mutation
69	* (ack-before-update pattern, see ADR-0004); the mutation is scheduled with random
70	* jitter to spread Redis load across instances.
71	*
72	* @see CacheSyncPublisher
73	* @see SyncMessage
74	* @see WorkerListener
75	*/
76	@RequiredArgsConstructor
77	@Slf4j	3✔
78	@Internal
79	public class CacheSyncListener {
80
81	/** Shared Jackson {@link ObjectMapper} for deserializing batch-invalidation key lists from JSON. */
82	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();	5✔
83
84	/** Local Caffeine L1 cache — target of invalidation and refresh operations.
85	* Accessed atomically via {@code asMap().compute()} for thread-safe updates. */
86	private final Cache<String, Object> caffeineCache;
87
88	/** Loads the current value from Redis given a cache key.
89	* Used during REFRESH to fetch the authoritative value before writing to L1. */
90	private final CacheLoader redisLoader;
91
92	/** Configuration for sync exchange name, jitter settings, and consumer concurrency. */
93	private final CacheSyncProperties properties;
94
95	/** Scheduler for running jitter-delayed cache update tasks, spreading Redis load.
96	* Supplied externally to allow shared-pool reuse across listeners. */
97	private final ScheduledExecutorService scheduler;
98
99	/** Computes hard and soft expiry timestamps for refreshed entries. */
100	private final CacheExpireManager expireManager;
101
102	/** Hot-key rule matcher whose rule set is updated when a RULES_SYNC message arrives. */
103	private final RuleMatcher ruleMatcher;
104
105	/**
106	* RabbitMQ message callback for incoming sync messages. Acknowledges the message
107	* immediately after parsing (ack-before-update), then schedules the actual cache
108	* mutation asynchronously with a random jitter to spread Redis load across peers.
109	*
110	* <p>On success, the message is acknowledged via {@link Channel#basicAck}. On any
111	* processing exception (parse failure, routing failure), the message is negatively
112	* acknowledged with {@code requeue=false} to prevent poison-message loops. The next
113	* application-level write will re-broadcast the operation.
114	*
115	* @param channel the AMQP channel used for ack/nack operations
116	* @param msg the raw AMQP message whose body and headers carry the sync payload;
117	* must not be null
118	* @throws IOException if the channel's basicAck or basicNack call fails
119	*/
120	public void handleSyncMessage(Channel channel, Message msg) throws IOException {
121	long tag = msg.getMessageProperties().getDeliveryTag();	4✔
122	try {
123	processSync(msg);	3✔
124	channel.basicAck(tag, false);	4✔
125	} catch (Exception e) {	1✔
126	log.error("CacheSync processing failed: body={}", new String(msg.getBody()), e);	9✔
127	channel.basicNack(tag, false, false);	5✔
128	}	1✔
129	}	1✔
130
131	/**
132	* Decodes the raw AMQP message into a {@link SyncMessage} and schedules the
133	* appropriate handler to run after a random delay within
134	* {@link CacheSyncProperties#getWarmupJitterMs()}. The jitter spreads Redis
135	* reads when multiple peers process the same sync broadcast simultaneously.
136	*
137	* @param msg the raw AMQP message; if the body is null, empty, or cannot be
138	* parsed into a valid {@link SyncMessage}, the message is silently
139	* dropped without scheduling any task
140	*/
141	private void processSync(Message msg) {
142	SyncMessage sm = SyncMessage.from(msg);	3✔
143	if (sm == null) {	2✔
144	log.debug("Received sync message with empty or invalid body");	3✔
145	return;	1✔
146	}
147
148	Runnable task = () -> {	4✔
149	try {
150	syncMessageRouter(sm);	3✔
151	} catch (Exception e) {	×
152	log.error("Async sync task failed: type={}, key={}, version={}", sm.type(), sm.cacheKey(), sm.version(), e);	×
153	}	1✔
154	};	1✔
155
156	// Random jitter spreads out Redis fetches, reducing hotspot load during batch sync.
157	DelayUtil.floatTimeDelay(task, properties.getWarmupJitterMs(), scheduler);	8✔
158	}	1✔
159
160	/**
161	* Routes the deserialized {@link SyncMessage} to the appropriate handler based
162	* on its type field. Delegates to {@link #handleLocalInvalidate},
163	* {@link #handleLocalInvalidateAll}, {@link #handleRefresh}, or
164	* {@link #handleRulesSync} accordingly.
165	*
166	* @param msg the deserialized sync message to route; must not be null
167	*/
168	private void syncMessageRouter(SyncMessage msg) {
169	if (msg.type() == null) {	3✔
170	log.debug("Received sync with null type, skipping");	3✔
171	return;	1✔
172	}
173	switch (msg.type()) {	9✔
174	case TYPE_INVALIDATE -> handleLocalInvalidate(msg);	4✔
175	case TYPE_INVALIDATE_ALL -> handleLocalInvalidateAll(msg);	4✔
176	case TYPE_REFRESH -> handleRefresh(msg);	4✔
177	case TYPE_RULES_SYNC -> handleRulesSync(msg);	4✔
178	default -> log.warn("Unknown sync type: {}, cacheKey: {}", msg.type(), msg.cacheKey());	7✔
179	}
180	}	1✔
181
182	/**
183	* Atomically removes the specified key from the local cache in response to
184	* an INVALIDATE sync message from a peer instance.
185	*
186	* <p><b>Version guard logic:</b>
187	* <ul>
188	* <li><b>Unconditional path:</b> When {@code version == 0L && !isVersionDegraded}
189	* (clean invalidation from {@code invalidateAllLocal}), the guard is bypassed
190	* entirely — the entry is always removed.</li>
191	* <li><b>Guarded path:</b> Uses {@link VersionGuard#shouldSkipForSync} with the
192	* 4-case degraded comparison. Case 2 (existing normal, incoming degraded)
193	* prevents a stale degraded INVALIDATE from wiping a healthy entry.</li>
194	* </ul>
195	*
196	* <p>Double-checked locking (DCL): a fast version guard before the atomic
197	* {@code compute} (first pass), and a second guard inside the {@code compute}
198	* body (second pass) to prevent a concurrent REFRESH from being wiped by a
199	* stale invalidate that arrived after the refresh.
200	*
201	* @param sm the sync message containing the key to invalidate; if the key
202	* is null or invalid, the invalidation is silently skipped
203	*/
204	private void handleLocalInvalidate(SyncMessage sm) {
205	boolean unconditional = sm.version() == 0L && !sm.isVersionDegraded();	12!
206
207	if (	5✔
208	!unconditional &&
209	VersionGuard.shouldSkipForSync(caffeineCache, sm.cacheKey(), sm.version(), sm.isVersionDegraded())	7✔
210	) {
211	log.debug("Stale invalidate ignored: key={}, incomingVersion={}", sm.cacheKey(), sm.version());	8✔
212	return;	1✔
213	}
214
215	caffeineCache	2✔
216	.asMap()	2✔
217	.compute(sm.cacheKey(), (key, existing) -> {	6✔
218	if (	5!
219	!unconditional &&
220	existing instanceof CacheEntry ce &&	×
221	VersionGuard.shouldSkipForSync(ce, sm.version(), sm.isVersionDegraded())	×
222	) {
223	return existing;	×
224	}
225	return null;	2✔
226	});
227	log.debug("Invalidated by sync: {}", sm.cacheKey());	5✔
228	}	1✔
229
230	/**
231	* Batch-invalidates all keys contained in the JSON-array body of the sync message.
232	*
233	* <p>This method intentionally bypasses version guards. The publisher
234	* ({@link CacheSyncPublisher#broadcastLocalInvalidateAll}) always sends clean
235	* messages (version=0L, not degraded) and all keys are removed unconditionally.
236	* This is more efficient than sending individual INVALIDATE messages for each key.
237	*
238	* <p>Deserialization failures (malformed JSON) are logged at ERROR level and
239	* do not propagate.
240	*
241	* @param sm the sync message whose {@code cacheKey} field contains the JSON-array
242	* of keys to invalidate; must not be null
243	*/
244	private void handleLocalInvalidateAll(SyncMessage sm) {
245	try {
246	List<String> keys = OBJECT_MAPPER.readValue(sm.cacheKey(), new TypeReference<>() {});	16✔
247	caffeineCache.invalidateAll(keys);	4✔
248	log.debug("Batch invalidated {} keys", keys.size());	6✔
249	} catch (Exception e) {	1✔
250	log.error("Failed to deserialize batch invalidate keys", e);	4✔
251	}	1✔
252	}	1✔
253
254	/**
255	* Merges the incoming rule set from a RULES_SYNC message into the local
256	* {@link RuleMatcher}, guarded by the message's {@code rulesVersion}.
257	* <p>
258	* Delegates to {@link RuleMatcher#syncRules}, which handles the actual
259	* merge logic and version conflict resolution.
260	*
261	* @param sm the sync message whose {@code cacheKey} field contains the
262	* serialized rule-set JSON and whose {@code rulesVersion} field
263	* carries the version for conflict resolution; must not be null
264	*/
265	private void handleRulesSync(SyncMessage sm) {
266	ruleMatcher.syncRules(sm.cacheKey(), sm.rulesVersion());	7✔
267	}	1✔
268
269	/**
270	* Refreshes a cache entry with the latest value from Redis in response to a
271	* REFRESH sync message from a peer instance.
272	*
273	* <p><b>Refresh flow:</b>
274	* <ol>
275	* <li><b>DCL check 1:</b> Fast-path version guard ({@link VersionGuard#shouldSkipForSync})
276	* against the existing L1 entry. If a newer dataVersion is already present,
277	* the refresh is skipped.</li>
278	* <li><b>Redis fetch:</b> Loads the authoritative value from Redis.</li>
279	* <li><b>DCL check 2:</b> Second version guard inside the atomic {@code compute}
280	* to prevent overwriting a newer version that arrived during the Redis fetch.</li>
281	* <li><b>Write:</b> Replaces the value and dataVersion while preserving the existing
282	* entry's metadata (hard/soft TTLs, normal TTLs, key state, decision version,
283	* degradation flag). If no entry existed in L1 before the refresh, a fresh
284	* {@link CacheEntry} is created with default metadata and {@code KeyState.NORMAL}.</li>
285	* </ol>
286	*
287	* <p>If the key is absent from L1 and Redis returns null (key does not exist),
288	* the refresh is aborted — there is nothing to cache.
289	*
290	* @param sm the sync message containing the key and version to refresh;
291	* must not be null
292	*/
293	private void handleRefresh(SyncMessage sm) {
294	// DCL first check – cheap, outside the compute lock
295	if (VersionGuard.shouldSkipForSync(caffeineCache, sm.cacheKey(), sm.version(), sm.isVersionDegraded())) {	10✔
296	log.debug("Stale refresh ignored: key={}, incomingVersion={}", sm.cacheKey(), sm.version());	8✔
297	return;	1✔
298	}
299
300	Object value = loadFromRedis(sm);	4✔
301	if (value == null) {	2✔
302	log.warn("Refresh failed to load value from Redis for key={}", sm.cacheKey());	5✔
303	return;	1✔
304	}
305
306	caffeineCache	2✔
307	.asMap()	2✔
308	.compute(sm.cacheKey(), (key, existing) -> {	7✔
309	// DCL second check – atomic with to write
310	if (	3✔
311	existing instanceof CacheEntry ce && VersionGuard.shouldSkipForSync(ce, sm.version(), sm.isVersionDegraded())	10!
312	) {
313	return existing;	×
314	}
315
316	if (existing instanceof CacheEntry cacheEntry) {	6✔
317	long hardExpireAt = expireManager.computeHardExpireAt(cacheEntry.getHardTtlMs());	6✔
318	long softExpireAt = expireManager.computeSoftExpireAt(cacheEntry.getSoftTtlMs());	6✔
319	return cacheEntry	2✔
320	.toBuilder()	2✔
321	.value(value)	2✔
322	.dataVersion(sm.version())	3✔
323	.isVersionDegraded(sm.isVersionDegraded())	3✔
324	.hardExpireAtMs(hardExpireAt)	2✔
325	.softExpireAtMs(softExpireAt)	1✔
326	.build();	1✔
327	}
328	long defaultHardTtlMs = expireManager.getEffectiveHardTtlMs();	4✔
329	long defaultSoftTtlMs = expireManager.getEffectiveSoftTtlMs();	4✔
330	//Entry was evicted from L1 — create fresh CacheEntry with default metadata
331	return expireManager.applyTtl(	4✔
332	CacheEntry.builder()	2✔
333	.value(value)	2✔
334	.dataVersion(sm.version())	3✔
335	.isVersionDegraded(sm.isVersionDegraded())	3✔
336	.decisionVersion(0L)	2✔
337	.keyState(KeyState.NORMAL)	2✔
338	.normalHardTtlMs(defaultHardTtlMs)	2✔
339	.normalSoftTtlMs(defaultSoftTtlMs)	1✔
340	.build(),	3✔
341	defaultHardTtlMs,
342	defaultSoftTtlMs
343	);
344	});
345	log.debug("Refreshed by sync: {}", sm.cacheKey());	5✔
346	}	1✔
347
348	/**
349	* Loads the current value from Redis for the key carried in the sync message.
350	* <p>
351	* Any exception thrown by the {@code redisLoader} (connection timeout, Redis
352	* outage, serialization error) is caught and logged at WARN level. The caller
353	* should handle a {@code null} return by aborting the refresh.
354	*
355	* @param sm the sync message containing the cache key to load; must not be null
356	* @return the value from Redis, or {@code null} if the key is absent in Redis
357	* or the load failed with an exception
358	*/
359	private Object loadFromRedis(SyncMessage sm) {
360	try {
361	return redisLoader.load(sm.cacheKey());	6✔
362	} catch (Exception e) {	1✔
363	log.warn("handleRefresh: Redis load failed for key={}", sm.cacheKey(), e);	6✔
364	return null;	2✔
365	}
366	}
367	}

Hyshmily / hotkey / 28705436176

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous