28159619024

Committed 25 Jun 2026 09:10AM UTC coverage: 92.667% (-0.02%) from 92.685%

Build # 28159619024

Build Type

push

github

Committed by

Hyshmily

Commit Message

release: v1.1.52

Coverage Stats

1199 of 1351 branches covered (88.75%)

Branch coverage included in aggregate %.

3426 of 3640 relevant lines covered (94.12%)

4.25 hits per line

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

91.22

/common/src/main/java/io/github/hyshmily/hotkey/sync/worker/WorkerListener.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.worker;

import static io.github.hyshmily.hotkey.sync.worker.WorkerMessage.TYPE_COOL;
import static io.github.hyshmily.hotkey.sync.worker.WorkerMessage.TYPE_HOT;

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.model.KeyState;
import io.github.hyshmily.hotkey.sync.local.CacheSyncListener;
import io.github.hyshmily.hotkey.util.DelayUtil;
import io.github.hyshmily.hotkey.util.ratelimit.SreRateLimiter;
import io.github.hyshmily.hotkey.util.version.VersionGuard;
import java.io.IOException;
import java.util.Optional;
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 Worker hot/cool decisions via the {@code hotkey.worker.exchange}
 * FanoutExchange and applies them to the local Caffeine L1 cache.
 *
 * <p>This is the consumer-side counterpart of the Worker's {@code WorkerBroadcaster}.
 * Incoming AMQP messages are deserialized into {@link WorkerMessage} records and
 * routed by decision type:
 * <ul>
 *   <li><b>HOT</b> ({@link WorkerMessage#TYPE_HOT}): Loads the current value from Redis,
 *       then promotes the L1 cache entry to {@link KeyState#HOT} with an extended hard TTL
 *       and active soft expiration (proactive refresh). Uses double-checked locking (DCL)
 *       with {@link VersionGuard#shouldSkipForWorker} to prevent overwriting a concurrently
 *       received newer decision.</li>
 *   <li><b>COOL</b> ({@link WorkerMessage#TYPE_COOL}): Downgrades an existing entry to
 *       {@link KeyState#COOL}, disabling soft expiration and resetting the hard TTL to
 *       the normal value. The cached data is retained to avoid thundering herds on the
 *       next read.</li>
 * </ul>
 *
 * <p><b>Ack-before-update pattern:</b> The AMQP message is acknowledged immediately after
 * parsing (see {@link #handleWorkerMessage}). The actual cache mutation is scheduled
 * asynchronously with a random jitter (see {@link WorkerListenerProperties#warmupJitterMs}).
 * This decoupling provides at-most-once delivery semantics — if the application crashes
 * after ack but before the cache write, the decision is lost and will be re-driven by the
 * next Worker heartbeat cycle (see ADR-0004).
 *
 * <p><b>Backpressure:</b> HOT promotions can be throttled by an optional
 * {@link SreRateLimiter}. When the limiter drops a request, the entry retains its current
 * state; the next Worker heartbeat or a subsequent HOT broadcast will re-attempt the
 * promotion.
 *
 * @see WorkerMessage
 * @see WorkerHeartbeatVerifier
 * @see CacheSyncListener
 */
@RequiredArgsConstructor
@Slf4j
public class WorkerListener {

  /** Local Caffeine L1 cache — target for HOT promotion and COOL downgrade 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 HOT promotion to fetch the authoritative value before writing to L1. */
  private final Function<String, Object> redisLoader;

  /** Configuration for Worker exchange name, queue prefix, jitter settings, and rate limiter. */
  private final WorkerListenerProperties 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 HOT-promoted and default-TTL entries. */
  private final CacheExpireManager expireManager;

  /** Optional SRE adaptive rate limiter for HOT decision processing.
   * When non-null, HOT promotions are probabilistically dropped during overload.
   * {@code null} disables rate limiting. */
  private final SreRateLimiter sreRateLimiter;

  /**
   * RabbitMQ message callback for incoming Worker decisions. Acknowledges the message
   * immediately after parsing (ack-before-update), then schedules the actual cache
   * mutation asynchronously with a random jitter to spread Redis load.
   *
   * <p>If processing fails with an exception, the message is negatively acknowledged
   * with {@code requeue=false} to prevent poison-message loops. The decision will be
   * re-driven by the next Worker heartbeat cycle.
   *
   * @param channel the AMQP channel used for ack/nack operations
   * @param msg     the raw AMQP message containing the Worker decision; must not be null
   * @throws IOException if the channel's basicAck or basicNack call fails
   */
  public void handleWorkerMessage(Channel channel, Message msg) throws IOException {
    long tag = msg.getMessageProperties().getDeliveryTag();
    try {
      processWorker(msg);
      channel.basicAck(tag, false); // ack before cache write
    } catch (Exception e) {
      log.error("Worker message processing failed: body={}", new String(msg.getBody()), e);
      channel.basicNack(tag, false, false); // do not requeue
    }
  }

  /**
   * Decodes the raw AMQP message into a {@link WorkerMessage} and schedules the
   * appropriate handler to run after a random delay within
   * {@link WorkerListenerProperties#getWarmupJitterMs()}. The jitter spreads Redis
   * reads across a small time window when the Worker broadcasts to many instances.
   *
   * @param msg the raw AMQP message; may have null or empty body, in which case
   *            the message is silently dropped
   */
  private void processWorker(Message msg) {
    WorkerMessage wm = WorkerMessage.from(msg);
    if (wm == null) {
      log.debug("Received worker message with empty body");
      return;
    }

    Runnable task = () -> {
      try {
        workerMessageRouter(wm);
      } catch (Exception e) {
        log.error(
          "Error processing WorkerMessage: cacheKey={}, type={}, decisionVersion={}, nodeId={}, epoch={}",
          wm.cacheKey(),
          wm.type(),
          wm.decisionVersion(),
          wm.nodeId(),
          wm.epoch(),
          e
        );
      }
    };

    // Random jitter spreads Redis reads across a small time window,
    // preventing all nodes from fetching the same key simultaneously.
    DelayUtil.floatTimeDelay(task, properties.getWarmupJitterMs(), scheduler);
  }

  /**
   * Routes the deserialized {@link WorkerMessage} to the appropriate handler based
   * on its type field. Delegates to {@link #handleHot} for {@code TYPE_HOT} and
   * {@link #handleCool} for {@code TYPE_COOL}. Messages with a null or unknown type
   * are logged and dropped.
   *
   * @param msg the deserialized Worker message to route; must not be null
   */
  private void workerMessageRouter(WorkerMessage msg) {
    if (msg.type() == null) {
      log.debug("Received worker message with null type, skipping");
      return;
    }
    switch (msg.type()) {
      case TYPE_HOT -> handleHot(msg);
      case TYPE_COOL -> handleCool(msg);
      default -> log.warn("Unknown worker message type: {}, cacheKey: {}", msg.type(), msg.cacheKey());
    }
  }

  /**
   * Promotes a cache key to {@link KeyState#HOT} with extended TTL and active soft
   * expiration, following a Worker HOT decision.
   *
   * <p><b>Promotion flow:</b>
   * <ol>
   *   <li><b>SRE gate:</b> If the rate limiter drops this request, the promotion is
   *       skipped entirely — backpressure from downstream saturation.</li>
   *   <li><b>DCL check 1:</b> Fast-path version guard ({@link VersionGuard#shouldSkipForWorker})
   *       against the existing L1 entry. If a newer decision is already present, this is a no-op.</li>
   *   <li><b>Redis fetch:</b> Loads the authoritative value from Redis. If Redis is
   *       unavailable, falls back to the existing degraded L1 entry value (if any).</li>
   *   <li><b>DCL check 2:</b> Second version guard inside the atomic {@code compute} to
   *       prevent overwriting a newer decision that arrived during the Redis fetch.</li>
   *   <li><b>Write:</b> Replaces the entry with a new {@link CacheEntry} in
   *       {@code KeyState.HOT}, preserving data-version fields, setting HOT-specific
   *       TTLs, and recording the Worker's {@code decisionVersion}.</li>
   * </ol>
   *
   * <p>If no value is available from Redis <em>and</em> no degraded entry exists in L1,
   * the promotion is aborted — there is nothing to cache.
   *
   * @param wm the Worker message containing the HOT decision; must not be null
   */
  private void handleHot(WorkerMessage wm) {
    // SRE rate limiter gate — skip HOT promotion when the system is overloaded
    if (sreRateLimiter != null && !sreRateLimiter.tryAcquire()) {
      sreRateLimiter.onFailed();
      log.debug("SRE throttled HOT promotion for key={}", wm.cacheKey());
      return;
    }

    // DCL first check – cheap, outside the compute lock
    if (VersionGuard.shouldSkipForWorker(caffeineCache, wm.cacheKey(), wm.decisionVersion(), wm.nodeId(), wm.epoch())) {
      log.debug("handleHot: HotKey already up-to-date in L1: {}", wm.cacheKey());
      return;
    }

    Object value = Optional.ofNullable(loadFromRedis(wm))
      .or(() ->
        Optional.ofNullable(caffeineCache.getIfPresent(wm.cacheKey()))
          .filter(ce -> ce instanceof CacheEntry c && c.isVersionDegraded())
          .map(ce -> ((CacheEntry) ce).getValue())
      )
      .orElse(null);

    if (value == null) {
      log.debug("handleHot: HotKey value not found in Redis and no degraded entry: {}", wm.cacheKey());
      return;
    }

    caffeineCache
      .asMap()
      .compute(wm.cacheKey(), (key, existing) -> {
        // DCL second check – atomic with to write
        if (existing instanceof CacheEntry ce) {
          if (VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())) {
            return existing;
          }

          return ce
            .toBuilder()
            .value(value)
            .decisionVersion(wm.decisionVersion())
            .decisionNodeId(wm.nodeId())
            .decisionEpoch(wm.epoch())
            .hardTtlMs(expireManager.getEffectiveHotHardTtlMs())
            .hardExpireAtMs(expireManager.computeHotHardExpireAt())
            .softTtlMs(expireManager.getEffectiveHotSoftTtlMs())
            .softExpireAtMs(expireManager.computeHotSoftExpireAt())
            .keyState(KeyState.HOT)
            .build();
        }

        return CacheEntry.builder()
          .value(value)
          .dataVersion(0)
          .isVersionDegraded(false)
          .decisionVersion(wm.decisionVersion())
          .hardTtlMs(expireManager.getEffectiveHotHardTtlMs())
          .hardExpireAtMs(expireManager.computeHotHardExpireAt())
          .softTtlMs(expireManager.getEffectiveHotSoftTtlMs())
          .softExpireAtMs(expireManager.computeHotSoftExpireAt())
          .keyState(KeyState.HOT)
          .normalHardTtlMs(expireManager.getEffectiveHardTtlMs())
          .normalSoftTtlMs(expireManager.getEffectiveSoftTtlMs())
          .build();
      });
    log.debug("HotKey promoted by Worker: {}", wm.cacheKey());
    if (sreRateLimiter != null) {
      sreRateLimiter.onSuccess();
    }
  }

  /**
   * Downgrades a cache key from {@link KeyState#HOT} to {@link KeyState#COOL},
   * following a Worker COOL decision.
   *
   * <p>This effectively reverts the HOT promotion:
   * <ul>
   *   <li>The cached value, {@code dataVersion}, degradation flag, and
   *       {@code decisionVersion} are all preserved — the data remains available.</li>
   *   <li>The hard TTL is reset to the normal value (from
   *       {@link CacheEntry#getNormalHardTtlMs()}).</li>
   *   <li>Soft expiration is fully disabled (both {@code softTtlMs} and
   *       {@code softExpireAtMs} set to {@code 0L}), so the entry stops being
   *       proactively refreshed.</li>
   * </ul>
   *
   * <p>If no existing entry is present in L1, the COOL decision is a no-op —
   * there is nothing to downgrade. The entry will be evicted naturally by
   * Caffeine's capacity policy or a subsequent invalidation.
   *
   * @param wm the Worker message containing the COOL decision; must not be null
   */
  private void handleCool(WorkerMessage wm) {
    caffeineCache
      .asMap()
      .compute(wm.cacheKey(), (key, existing) -> {
        if (
          existing instanceof CacheEntry ce &&
          VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())
        ) {
          return existing;
        }

        if (existing instanceof CacheEntry cacheEntry) {
          return cacheEntry
            .toBuilder()
            .value(cacheEntry.getValue())
            .dataVersion(cacheEntry.getDataVersion())
            .decisionVersion(wm.decisionVersion())
            .decisionNodeId(wm.nodeId())
            .decisionEpoch(wm.epoch())
            .hardTtlMs(cacheEntry.getNormalHardTtlMs())
            .hardExpireAtMs(expireManager.computeHardExpireAt(cacheEntry.getNormalHardTtlMs()))
            .softTtlMs(0L)
            .softExpireAtMs(0L)
            .keyState(KeyState.COOL)
            .build();
        }

        // No existing entry – nothing to cool
        return existing;
      });
    log.debug("HotKey cooled by Worker: {}", wm.cacheKey());
  }

  /**
   * Loads the current value from Redis for the key carried in the Worker message.
   * <p>
   * Any exception thrown by the {@code redisLoader} (connection timeout, Redis
   * outage, serialization error) is caught and logged at WARN level. The caller
   * is responsible for falling back to the degraded entry value when this method
   * returns {@code null}.
   *
   * @param wm the Worker message containing the cache key to load; must not be null
   * @return the value from Redis, or {@code null} if the key is absent or the
   *         load failed with an exception
   */
  private Object loadFromRedis(WorkerMessage wm) {
    try {
      return redisLoader.apply(wm.cacheKey());
    } catch (Exception e) {
      log.warn("handleHot: Redis load failed for key={}, trying degraded entry", wm.cacheKey(), e);
      if (sreRateLimiter != null) {
        sreRateLimiter.onFailed();
      }
      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.worker;
17
18	import static io.github.hyshmily.hotkey.sync.worker.WorkerMessage.TYPE_COOL;
19	import static io.github.hyshmily.hotkey.sync.worker.WorkerMessage.TYPE_HOT;
20
21	import com.github.benmanes.caffeine.cache.Cache;
22	import com.rabbitmq.client.Channel;
23	import io.github.hyshmily.hotkey.cache.CacheExpireManager;
24	import io.github.hyshmily.hotkey.model.CacheEntry;
25	import io.github.hyshmily.hotkey.model.KeyState;
26	import io.github.hyshmily.hotkey.sync.local.CacheSyncListener;
27	import io.github.hyshmily.hotkey.util.DelayUtil;
28	import io.github.hyshmily.hotkey.util.ratelimit.SreRateLimiter;
29	import io.github.hyshmily.hotkey.util.version.VersionGuard;
30	import java.io.IOException;
31	import java.util.Optional;
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 Worker hot/cool decisions via the {@code hotkey.worker.exchange}
40	* FanoutExchange and applies them to the local Caffeine L1 cache.
41	*
42	* <p>This is the consumer-side counterpart of the Worker's {@code WorkerBroadcaster}.
43	* Incoming AMQP messages are deserialized into {@link WorkerMessage} records and
44	* routed by decision type:
45	* <ul>
46	* <li><b>HOT</b> ({@link WorkerMessage#TYPE_HOT}): Loads the current value from Redis,
47	* then promotes the L1 cache entry to {@link KeyState#HOT} with an extended hard TTL
48	* and active soft expiration (proactive refresh). Uses double-checked locking (DCL)
49	* with {@link VersionGuard#shouldSkipForWorker} to prevent overwriting a concurrently
50	* received newer decision.</li>
51	* <li><b>COOL</b> ({@link WorkerMessage#TYPE_COOL}): Downgrades an existing entry to
52	* {@link KeyState#COOL}, disabling soft expiration and resetting the hard TTL to
53	* the normal value. The cached data is retained to avoid thundering herds on the
54	* next read.</li>
55	* </ul>
56	*
57	* <p><b>Ack-before-update pattern:</b> The AMQP message is acknowledged immediately after
58	* parsing (see {@link #handleWorkerMessage}). The actual cache mutation is scheduled
59	* asynchronously with a random jitter (see {@link WorkerListenerProperties#warmupJitterMs}).
60	* This decoupling provides at-most-once delivery semantics — if the application crashes
61	* after ack but before the cache write, the decision is lost and will be re-driven by the
62	* next Worker heartbeat cycle (see ADR-0004).
63	*
64	* <p><b>Backpressure:</b> HOT promotions can be throttled by an optional
65	* {@link SreRateLimiter}. When the limiter drops a request, the entry retains its current
66	* state; the next Worker heartbeat or a subsequent HOT broadcast will re-attempt the
67	* promotion.
68	*
69	* @see WorkerMessage
70	* @see WorkerHeartbeatVerifier
71	* @see CacheSyncListener
72	*/
73	@RequiredArgsConstructor
74	@Slf4j	4✔
75	public class WorkerListener {
76
77	/** Local Caffeine L1 cache — target for HOT promotion and COOL downgrade operations.
78	* Accessed atomically via {@code asMap().compute()} for thread-safe updates. */
79	private final Cache<String, Object> caffeineCache;
80
81	/** Function that loads the current value from Redis given a cache key.
82	* Used during HOT promotion to fetch the authoritative value before writing to L1. */
83	private final Function<String, Object> redisLoader;
84
85	/** Configuration for Worker exchange name, queue prefix, jitter settings, and rate limiter. */
86	private final WorkerListenerProperties properties;
87
88	/** Scheduler for running jitter-delayed cache update tasks, spreading Redis load.
89	* Supplied externally to allow shared-pool reuse across listeners. */
90	private final ScheduledExecutorService scheduler;
91
92	/** Computes hard and soft expiry timestamps for HOT-promoted and default-TTL entries. */
93	private final CacheExpireManager expireManager;
94
95	/** Optional SRE adaptive rate limiter for HOT decision processing.
96	* When non-null, HOT promotions are probabilistically dropped during overload.
97	* {@code null} disables rate limiting. */
98	private final SreRateLimiter sreRateLimiter;
99
100	/**
101	* RabbitMQ message callback for incoming Worker decisions. Acknowledges the message
102	* immediately after parsing (ack-before-update), then schedules the actual cache
103	* mutation asynchronously with a random jitter to spread Redis load.
104	*
105	* <p>If processing fails with an exception, the message is negatively acknowledged
106	* with {@code requeue=false} to prevent poison-message loops. The decision will be
107	* re-driven by the next Worker heartbeat cycle.
108	*
109	* @param channel the AMQP channel used for ack/nack operations
110	* @param msg the raw AMQP message containing the Worker decision; must not be null
111	* @throws IOException if the channel's basicAck or basicNack call fails
112	*/
113	public void handleWorkerMessage(Channel channel, Message msg) throws IOException {
114	long tag = msg.getMessageProperties().getDeliveryTag();	4✔
115	try {
116	processWorker(msg);	3✔
117	channel.basicAck(tag, false); // ack before cache write	4✔
118	} catch (Exception e) {	1✔
119	log.error("Worker message processing failed: body={}", new String(msg.getBody()), e);	9✔
120	channel.basicNack(tag, false, false); // do not requeue	5✔
121	}	1✔
122	}	1✔
123
124	/**
125	* Decodes the raw AMQP message into a {@link WorkerMessage} and schedules the
126	* appropriate handler to run after a random delay within
127	* {@link WorkerListenerProperties#getWarmupJitterMs()}. The jitter spreads Redis
128	* reads across a small time window when the Worker broadcasts to many instances.
129	*
130	* @param msg the raw AMQP message; may have null or empty body, in which case
131	* the message is silently dropped
132	*/
133	private void processWorker(Message msg) {
134	WorkerMessage wm = WorkerMessage.from(msg);	3✔
135	if (wm == null) {	2✔
136	log.debug("Received worker message with empty body");	3✔
137	return;	1✔
138	}
139
140	Runnable task = () -> {	4✔
141	try {
142	workerMessageRouter(wm);	3✔
143	} catch (Exception e) {	×
144	log.error(	×
145	"Error processing WorkerMessage: cacheKey={}, type={}, decisionVersion={}, nodeId={}, epoch={}",
146	wm.cacheKey(),	×
147	wm.type(),	×
148	wm.decisionVersion(),	×
149	wm.nodeId(),	×
150	wm.epoch(),	×
151	e
152	);
153	}	1✔
154	};	1✔
155
156	// Random jitter spreads Redis reads across a small time window,
157	// preventing all nodes from fetching the same key simultaneously.
158	DelayUtil.floatTimeDelay(task, properties.getWarmupJitterMs(), scheduler);	8✔
159	}	1✔
160
161	/**
162	* Routes the deserialized {@link WorkerMessage} to the appropriate handler based
163	* on its type field. Delegates to {@link #handleHot} for {@code TYPE_HOT} and
164	* {@link #handleCool} for {@code TYPE_COOL}. Messages with a null or unknown type
165	* are logged and dropped.
166	*
167	* @param msg the deserialized Worker message to route; must not be null
168	*/
169	private void workerMessageRouter(WorkerMessage msg) {
170	if (msg.type() == null) {	3✔
171	log.debug("Received worker message with null type, skipping");	3✔
172	return;	1✔
173	}
174	switch (msg.type()) {	9✔
175	case TYPE_HOT -> handleHot(msg);	4✔
176	case TYPE_COOL -> handleCool(msg);	4✔
177	default -> log.warn("Unknown worker message type: {}, cacheKey: {}", msg.type(), msg.cacheKey());	7✔
178	}
179	}	1✔
180
181	/**
182	* Promotes a cache key to {@link KeyState#HOT} with extended TTL and active soft
183	* expiration, following a Worker HOT decision.
184	*
185	* <p><b>Promotion flow:</b>
186	* <ol>
187	* <li><b>SRE gate:</b> If the rate limiter drops this request, the promotion is
188	* skipped entirely — backpressure from downstream saturation.</li>
189	* <li><b>DCL check 1:</b> Fast-path version guard ({@link VersionGuard#shouldSkipForWorker})
190	* against the existing L1 entry. If a newer decision is already present, this is a no-op.</li>
191	* <li><b>Redis fetch:</b> Loads the authoritative value from Redis. If Redis is
192	* unavailable, falls back to the existing degraded L1 entry value (if any).</li>
193	* <li><b>DCL check 2:</b> Second version guard inside the atomic {@code compute} to
194	* prevent overwriting a newer decision that arrived during the Redis fetch.</li>
195	* <li><b>Write:</b> Replaces the entry with a new {@link CacheEntry} in
196	* {@code KeyState.HOT}, preserving data-version fields, setting HOT-specific
197	* TTLs, and recording the Worker's {@code decisionVersion}.</li>
198	* </ol>
199	*
200	* <p>If no value is available from Redis <em>and</em> no degraded entry exists in L1,
201	* the promotion is aborted — there is nothing to cache.
202	*
203	* @param wm the Worker message containing the HOT decision; must not be null
204	*/
205	private void handleHot(WorkerMessage wm) {
206	// SRE rate limiter gate — skip HOT promotion when the system is overloaded
207	if (sreRateLimiter != null && !sreRateLimiter.tryAcquire()) {	7✔
208	sreRateLimiter.onFailed();	3✔
209	log.debug("SRE throttled HOT promotion for key={}", wm.cacheKey());	5✔
210	return;	1✔
211	}
212
213	// DCL first check – cheap, outside the compute lock
214	if (VersionGuard.shouldSkipForWorker(caffeineCache, wm.cacheKey(), wm.decisionVersion(), wm.nodeId(), wm.epoch())) {	12✔
215	log.debug("handleHot: HotKey already up-to-date in L1: {}", wm.cacheKey());	5✔
216	return;	1✔
217	}
218
219	Object value = Optional.ofNullable(loadFromRedis(wm))	7✔
220	.or(() ->	2✔
221	Optional.ofNullable(caffeineCache.getIfPresent(wm.cacheKey()))	8✔
222	.filter(ce -> ce instanceof CacheEntry c && c.isVersionDegraded())	14!
223	.map(ce -> ((CacheEntry) ce).getValue())	5✔
224	)
225	.orElse(null);	2✔
226
227	if (value == null) {	2✔
228	log.debug("handleHot: HotKey value not found in Redis and no degraded entry: {}", wm.cacheKey());	5✔
229	return;	1✔
230	}
231
232	caffeineCache	2✔
233	.asMap()	2✔
234	.compute(wm.cacheKey(), (key, existing) -> {	7✔
235	// DCL second check – atomic with to write
236	if (existing instanceof CacheEntry ce) {	6✔
237	if (VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())) {	9!
238	return existing;	×
239	}
240
241	return ce	2✔
242	.toBuilder()	2✔
243	.value(value)	2✔
244	.decisionVersion(wm.decisionVersion())	3✔
245	.decisionNodeId(wm.nodeId())	3✔
246	.decisionEpoch(wm.epoch())	4✔
247	.hardTtlMs(expireManager.getEffectiveHotHardTtlMs())	4✔
248	.hardExpireAtMs(expireManager.computeHotHardExpireAt())	4✔
249	.softTtlMs(expireManager.getEffectiveHotSoftTtlMs())	4✔
250	.softExpireAtMs(expireManager.computeHotSoftExpireAt())	3✔
251	.keyState(KeyState.HOT)	1✔
252	.build();	1✔
253	}
254
255	return CacheEntry.builder()	3✔
256	.value(value)	2✔
257	.dataVersion(0)	2✔
258	.isVersionDegraded(false)	2✔
259	.decisionVersion(wm.decisionVersion())	4✔
260	.hardTtlMs(expireManager.getEffectiveHotHardTtlMs())	4✔
261	.hardExpireAtMs(expireManager.computeHotHardExpireAt())	4✔
262	.softTtlMs(expireManager.getEffectiveHotSoftTtlMs())	4✔
263	.softExpireAtMs(expireManager.computeHotSoftExpireAt())	3✔
264	.keyState(KeyState.HOT)	3✔
265	.normalHardTtlMs(expireManager.getEffectiveHardTtlMs())	4✔
266	.normalSoftTtlMs(expireManager.getEffectiveSoftTtlMs())	2✔
267	.build();	1✔
268	});
269	log.debug("HotKey promoted by Worker: {}", wm.cacheKey());	5✔
270	if (sreRateLimiter != null) {	3✔
271	sreRateLimiter.onSuccess();	3✔
272	}
273	}	1✔
274
275	/**
276	* Downgrades a cache key from {@link KeyState#HOT} to {@link KeyState#COOL},
277	* following a Worker COOL decision.
278	*
279	* <p>This effectively reverts the HOT promotion:
280	* <ul>
281	* <li>The cached value, {@code dataVersion}, degradation flag, and
282	* {@code decisionVersion} are all preserved — the data remains available.</li>
283	* <li>The hard TTL is reset to the normal value (from
284	* {@link CacheEntry#getNormalHardTtlMs()}).</li>
285	* <li>Soft expiration is fully disabled (both {@code softTtlMs} and
286	* {@code softExpireAtMs} set to {@code 0L}), so the entry stops being
287	* proactively refreshed.</li>
288	* </ul>
289	*
290	* <p>If no existing entry is present in L1, the COOL decision is a no-op —
291	* there is nothing to downgrade. The entry will be evicted naturally by
292	* Caffeine's capacity policy or a subsequent invalidation.
293	*
294	* @param wm the Worker message containing the COOL decision; must not be null
295	*/
296	private void handleCool(WorkerMessage wm) {
297	caffeineCache	2✔
298	.asMap()	2✔
299	.compute(wm.cacheKey(), (key, existing) -> {	6✔
300	if (	3✔
301	existing instanceof CacheEntry ce &&	5✔
302	VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())	7✔
303	) {
304	return existing;	2✔
305	}
306
307	if (existing instanceof CacheEntry cacheEntry) {	6✔
308	return cacheEntry	2✔
309	.toBuilder()	2✔
310	.value(cacheEntry.getValue())	3✔
311	.dataVersion(cacheEntry.getDataVersion())	3✔
312	.decisionVersion(wm.decisionVersion())	3✔
313	.decisionNodeId(wm.nodeId())	3✔
314	.decisionEpoch(wm.epoch())	3✔
315	.hardTtlMs(cacheEntry.getNormalHardTtlMs())	5✔
316	.hardExpireAtMs(expireManager.computeHardExpireAt(cacheEntry.getNormalHardTtlMs()))	4✔
317	.softTtlMs(0L)	2✔
318	.softExpireAtMs(0L)	2✔
319	.keyState(KeyState.COOL)	1✔
320	.build();	1✔
321	}
322
323	// No existing entry – nothing to cool
324	return existing;	2✔
325	});
326	log.debug("HotKey cooled by Worker: {}", wm.cacheKey());	5✔
327	}	1✔
328
329	/**
330	* Loads the current value from Redis for the key carried in the Worker message.
331	* <p>
332	* Any exception thrown by the {@code redisLoader} (connection timeout, Redis
333	* outage, serialization error) is caught and logged at WARN level. The caller
334	* is responsible for falling back to the degraded entry value when this method
335	* returns {@code null}.
336	*
337	* @param wm the Worker message containing the cache key to load; must not be null
338	* @return the value from Redis, or {@code null} if the key is absent or the
339	* load failed with an exception
340	*/
341	private Object loadFromRedis(WorkerMessage wm) {
342	try {
343	return redisLoader.apply(wm.cacheKey());	6✔
344	} catch (Exception e) {	1✔
345	log.warn("handleHot: Redis load failed for key={}, trying degraded entry", wm.cacheKey(), e);	6✔
346	if (sreRateLimiter != null) {	3!
347	sreRateLimiter.onFailed();	×
348	}
349	return null;	2✔
350	}
351	}
352	}

Hyshmily / hotkey / 28159619024

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