28637103998

Committed 03 Jul 2026 03:48AM UTC coverage: 90.399% (-0.2%) from 90.63%

Build # 28637103998

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix: add packaging to gitignore, fix flaky jitter test, update docs references

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

Coverage Stats

1264 of 1461 branches covered (86.52%)

Branch coverage included in aggregate %.

3604 of 3924 relevant lines covered (91.85%)

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

Hyshmily / hotkey / 28637103998

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