27925388821

Committed 22 Jun 2026 02:14AM UTC coverage: 92.403% (-1.2%) from 93.615%

Build # 27925388821

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix:fix known bugs and promote performance

Signed-off-by: Hyshmily <cxm8607@outlook.com>

Coverage Stats

1143 of 1289 branches covered (88.67%)

Branch coverage included in aggregate %.

184 of 236 new or added lines in 19 files covered. (77.97%)

6 existing lines in 3 files now uncovered.

3321 of 3542 relevant lines covered (93.76%)

4.22 hits per line

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

91.54

/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.cache.CacheExpireManager;
import io.github.hyshmily.hotkey.model.CacheEntry;
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 java.util.function.Function;
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.invalidateAll()}.
 *       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
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;

  /** Function that loads the current value from Redis given a cache key.
   * Used during REFRESH to fetch the authoritative value before writing to L1. */
  private final Function<String, Object> 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 invalidateAll}), 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) {
          return cacheEntry
            .toBuilder()
            .value(value)
            .dataVersion(sm.version())
            .isVersionDegraded(sm.isVersionDegraded())
            .hardExpireAtMs(expireManager.computeHardExpireAt(cacheEntry.getHardTtlMs()))
            .softExpireAtMs(expireManager.computeSoftExpireAt(cacheEntry.getSoftTtlMs()))
            .build();
        }

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

Hyshmily / hotkey / 27925388821

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