28316658488

Committed 28 Jun 2026 08:34AM UTC coverage: 91.178% (-0.05%) from 91.227%

Build # 28316658488

Build Type

push

github

Committed by

Hyshmily

Commit Message

test :fix CI tests

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

Coverage Stats

1264 of 1445 branches covered (87.47%)

Branch coverage included in aggregate %.

3583 of 3871 relevant lines covered (92.56%)

4.21 hits per line

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

90.51

/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;

  /** Fallback hard TTL (seconds) for COOL entries when no normal TTL is configured on the existing entry. */
  private static final long COOL_DEFAULT_PROTECTION_HARDTTL_TIME = 120;
  /** Fallback soft TTL (seconds) for COOL entries when no normal TTL is configured on the existing entry. */
  private static final long COOL_DEFAULT_PROTECTION_SOFTTTL_TIME = 60;
  /** Jitter ratio for COOL fallback hard TTL (±20%). */
  private static final double COOL_DEFAULT_PROTECTION_HARDTTL_TIME_RATIO = 0.2;
  /** Jitter ratio for COOL fallback soft TTL (±20%). */
  private static final double COOL_DEFAULT_PROTECTION_SOFTTTL_TIME_RATIO = 0.2;

  /**
   * 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) {
          long normalHardTtlMs = cacheEntry.getNormalHardTtlMs();
          long normalSoftTtlMs = cacheEntry.getNormalSoftTtlMs();

          long hardTtlMsIfZero = normalHardTtlMs > 0 ? normalHardTtlMs : COOL_DEFAULT_PROTECTION_HARDTTL_TIME;
          long softTtlMsIfZero = normalSoftTtlMs > 0 ? normalSoftTtlMs : COOL_DEFAULT_PROTECTION_SOFTTTL_TIME;

          long hardTtlExpireAtMs = expireManager.toHardExpireTimestamp(
            hardTtlMsIfZero,
            COOL_DEFAULT_PROTECTION_HARDTTL_TIME_RATIO
          );
          long softTtlExpireAtMs = expireManager.toSoftExpireTimestamp(
            softTtlMsIfZero,
            COOL_DEFAULT_PROTECTION_SOFTTTL_TIME_RATIO
          );

          return cacheEntry
            .toBuilder()
            .value(cacheEntry.getValue())
            .dataVersion(cacheEntry.getDataVersion())
            .decisionVersion(wm.decisionVersion())
            .decisionNodeId(wm.nodeId())
            .decisionEpoch(wm.epoch())
            .hardTtlMs(hardTtlMsIfZero)
            .hardExpireAtMs(hardTtlExpireAtMs)
            .softTtlMs(softTtlMsIfZero)
            .softExpireAtMs(softTtlExpireAtMs)
            .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	/** Fallback hard TTL (seconds) for COOL entries when no normal TTL is configured on the existing entry. */
101	private static final long COOL_DEFAULT_PROTECTION_HARDTTL_TIME = 120;
102	/** Fallback soft TTL (seconds) for COOL entries when no normal TTL is configured on the existing entry. */
103	private static final long COOL_DEFAULT_PROTECTION_SOFTTTL_TIME = 60;
104	/** Jitter ratio for COOL fallback hard TTL (±20%). */
105	private static final double COOL_DEFAULT_PROTECTION_HARDTTL_TIME_RATIO = 0.2;
106	/** Jitter ratio for COOL fallback soft TTL (±20%). */
107	private static final double COOL_DEFAULT_PROTECTION_SOFTTTL_TIME_RATIO = 0.2;
108
109	/**
110	* RabbitMQ message callback for incoming Worker decisions. Acknowledges the message
111	* immediately after parsing (ack-before-update), then schedules the actual cache
112	* mutation asynchronously with a random jitter to spread Redis load.
113	*
114	* <p>If processing fails with an exception, the message is negatively acknowledged
115	* with {@code requeue=false} to prevent poison-message loops. The decision will be
116	* re-driven by the next Worker heartbeat cycle.
117	*
118	* @param channel the AMQP channel used for ack/nack operations
119	* @param msg the raw AMQP message containing the Worker decision; must not be null
120	* @throws IOException if the channel's basicAck or basicNack call fails
121	*/
122	public void handleWorkerMessage(Channel channel, Message msg) throws IOException {
123	long tag = msg.getMessageProperties().getDeliveryTag();	4✔
124	try {
125	processWorker(msg);	3✔
126	channel.basicAck(tag, false); // ack before cache write	4✔
127	} catch (Exception e) {	1✔
128	log.error("Worker message processing failed: body={}", new String(msg.getBody()), e);	9✔
129	channel.basicNack(tag, false, false); // do not requeue	5✔
130	}	1✔
131	}	1✔
132
133	/**
134	* Decodes the raw AMQP message into a {@link WorkerMessage} and schedules the
135	* appropriate handler to run after a random delay within
136	* {@link WorkerListenerProperties#getWarmupJitterMs()}. The jitter spreads Redis
137	* reads across a small time window when the Worker broadcasts to many instances.
138	*
139	* @param msg the raw AMQP message; may have null or empty body, in which case
140	* the message is silently dropped
141	*/
142	private void processWorker(Message msg) {
143	WorkerMessage wm = WorkerMessage.from(msg);	3✔
144	if (wm == null) {	2✔
145	log.debug("Received worker message with empty body");	3✔
146	return;	1✔
147	}
148
149	Runnable task = () -> {	4✔
150	try {
151	workerMessageRouter(wm);	3✔
152	} catch (Exception e) {	×
153	log.error(	×
154	"Error processing WorkerMessage: cacheKey={}, type={}, decisionVersion={}, nodeId={}, epoch={}",
155	wm.cacheKey(),	×
156	wm.type(),	×
157	wm.decisionVersion(),	×
158	wm.nodeId(),	×
159	wm.epoch(),	×
160	e
161	);
162	}	1✔
163	};	1✔
164
165	// Random jitter spreads Redis reads across a small time window,
166	// preventing all nodes from fetching the same key simultaneously.
167	DelayUtil.floatTimeDelay(task, properties.getWarmupJitterMs(), scheduler);	8✔
168	}	1✔
169
170	/**
171	* Routes the deserialized {@link WorkerMessage} to the appropriate handler based
172	* on its type field. Delegates to {@link #handleHot} for {@code TYPE_HOT} and
173	* {@link #handleCool} for {@code TYPE_COOL}. Messages with a null or unknown type
174	* are logged and dropped.
175	*
176	* @param msg the deserialized Worker message to route; must not be null
177	*/
178	private void workerMessageRouter(WorkerMessage msg) {
179	if (msg.type() == null) {	3✔
180	log.debug("Received worker message with null type, skipping");	3✔
181	return;	1✔
182	}
183	switch (msg.type()) {	9✔
184	case TYPE_HOT -> handleHot(msg);	4✔
185	case TYPE_COOL -> handleCool(msg);	4✔
186	default -> log.warn("Unknown worker message type: {}, cacheKey: {}", msg.type(), msg.cacheKey());	7✔
187	}
188	}	1✔
189
190	/**
191	* Promotes a cache key to {@link KeyState#HOT} with extended TTL and active soft
192	* expiration, following a Worker HOT decision.
193	*
194	* <p><b>Promotion flow:</b>
195	* <ol>
196	* <li><b>SRE gate:</b> If the rate limiter drops this request, the promotion is
197	* skipped entirely — backpressure from downstream saturation.</li>
198	* <li><b>DCL check 1:</b> Fast-path version guard ({@link VersionGuard#shouldSkipForWorker})
199	* against the existing L1 entry. If a newer decision is already present, this is a no-op.</li>
200	* <li><b>Redis fetch:</b> Loads the authoritative value from Redis. If Redis is
201	* unavailable, falls back to the existing degraded L1 entry value (if any).</li>
202	* <li><b>DCL check 2:</b> Second version guard inside the atomic {@code compute} to
203	* prevent overwriting a newer decision that arrived during the Redis fetch.</li>
204	* <li><b>Write:</b> Replaces the entry with a new {@link CacheEntry} in
205	* {@code KeyState.HOT}, preserving data-version fields, setting HOT-specific
206	* TTLs, and recording the Worker's {@code decisionVersion}.</li>
207	* </ol>
208	*
209	* <p>If no value is available from Redis <em>and</em> no degraded entry exists in L1,
210	* the promotion is aborted — there is nothing to cache.
211	*
212	* @param wm the Worker message containing the HOT decision; must not be null
213	*/
214	private void handleHot(WorkerMessage wm) {
215	// SRE rate limiter gate — skip HOT promotion when the system is overloaded
216	if (sreRateLimiter != null && !sreRateLimiter.tryAcquire()) {	7✔
217	sreRateLimiter.onFailed();	3✔
218	log.debug("SRE throttled HOT promotion for key={}", wm.cacheKey());	5✔
219	return;	1✔
220	}
221
222	// DCL first check – cheap, outside the compute lock
223	if (VersionGuard.shouldSkipForWorker(caffeineCache, wm.cacheKey(), wm.decisionVersion(), wm.nodeId(), wm.epoch())) {	12✔
224	log.debug("handleHot: HotKey already up-to-date in L1: {}", wm.cacheKey());	5✔
225	return;	1✔
226	}
227
228	Object value = Optional.ofNullable(loadFromRedis(wm))	7✔
229	.or(() ->	2✔
230	Optional.ofNullable(caffeineCache.getIfPresent(wm.cacheKey()))	8✔
231	.filter(ce -> ce instanceof CacheEntry c && c.isVersionDegraded())	14!
232	.map(ce -> ((CacheEntry) ce).getValue())	5✔
233	)
234	.orElse(null);	2✔
235
236	if (value == null) {	2✔
237	log.debug("handleHot: HotKey value not found in Redis and no degraded entry: {}", wm.cacheKey());	5✔
238	return;	1✔
239	}
240
241	caffeineCache	2✔
242	.asMap()	2✔
243	.compute(wm.cacheKey(), (key, existing) -> {	7✔
244	// DCL second check – atomic with to write
245	if (existing instanceof CacheEntry ce) {	6✔
246	if (VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())) {	9!
247	return existing;	×
248	}
249
250	return ce	2✔
251	.toBuilder()	2✔
252	.value(value)	2✔
253	.decisionVersion(wm.decisionVersion())	3✔
254	.decisionNodeId(wm.nodeId())	3✔
255	.decisionEpoch(wm.epoch())	4✔
256	.hardTtlMs(expireManager.getEffectiveHotHardTtlMs())	4✔
257	.hardExpireAtMs(expireManager.computeHotHardExpireAt())	4✔
258	.softTtlMs(expireManager.getEffectiveHotSoftTtlMs())	4✔
259	.softExpireAtMs(expireManager.computeHotSoftExpireAt())	3✔
260	.keyState(KeyState.HOT)	1✔
261	.build();	1✔
262	}
263
264	return CacheEntry.builder()	3✔
265	.value(value)	2✔
266	.dataVersion(0)	2✔
267	.isVersionDegraded(false)	2✔
268	.decisionVersion(wm.decisionVersion())	4✔
269	.hardTtlMs(expireManager.getEffectiveHotHardTtlMs())	4✔
270	.hardExpireAtMs(expireManager.computeHotHardExpireAt())	4✔
271	.softTtlMs(expireManager.getEffectiveHotSoftTtlMs())	4✔
272	.softExpireAtMs(expireManager.computeHotSoftExpireAt())	3✔
273	.keyState(KeyState.HOT)	3✔
274	.normalHardTtlMs(expireManager.getEffectiveHardTtlMs())	4✔
275	.normalSoftTtlMs(expireManager.getEffectiveSoftTtlMs())	2✔
276	.build();	1✔
277	});
278	log.debug("HotKey promoted by Worker: {}", wm.cacheKey());	5✔
279	if (sreRateLimiter != null) {	3✔
280	sreRateLimiter.onSuccess();	3✔
281	}
282	}	1✔
283
284	/**
285	* Downgrades a cache key from {@link KeyState#HOT} to {@link KeyState#COOL},
286	* following a Worker COOL decision.
287	*
288	* <p>This effectively reverts the HOT promotion:
289	* <ul>
290	* <li>The cached value, {@code dataVersion}, degradation flag, and
291	* {@code decisionVersion} are all preserved — the data remains available.</li>
292	* <li>The hard TTL is reset to the normal value (from
293	* {@link CacheEntry#getNormalHardTtlMs()}).</li>
294	* <li>Soft expiration is fully disabled (both {@code softTtlMs} and
295	* {@code softExpireAtMs} set to {@code 0L}), so the entry stops being
296	* proactively refreshed.</li>
297	* </ul>
298	*
299	* <p>If no existing entry is present in L1, the COOL decision is a no-op —
300	* there is nothing to downgrade. The entry will be evicted naturally by
301	* Caffeine's capacity policy or a subsequent invalidation.
302	*
303	* @param wm the Worker message containing the COOL decision; must not be null
304	*/
305	private void handleCool(WorkerMessage wm) {
306	caffeineCache	2✔
307	.asMap()	2✔
308	.compute(wm.cacheKey(), (key, existing) -> {	6✔
309	if (	3✔
310	existing instanceof CacheEntry ce &&	5✔
311	VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())	7✔
312	) {
313	return existing;	2✔
314	}
315
316	if (existing instanceof CacheEntry cacheEntry) {	6✔
317	long normalHardTtlMs = cacheEntry.getNormalHardTtlMs();	3✔
318	long normalSoftTtlMs = cacheEntry.getNormalSoftTtlMs();	3✔
319
320	long hardTtlMsIfZero = normalHardTtlMs > 0 ? normalHardTtlMs : COOL_DEFAULT_PROTECTION_HARDTTL_TIME;	7!
321	long softTtlMsIfZero = normalSoftTtlMs > 0 ? normalSoftTtlMs : COOL_DEFAULT_PROTECTION_SOFTTTL_TIME;	7!
322
323	long hardTtlExpireAtMs = expireManager.toHardExpireTimestamp(	6✔
324	hardTtlMsIfZero,
325	COOL_DEFAULT_PROTECTION_HARDTTL_TIME_RATIO
326	);
327	long softTtlExpireAtMs = expireManager.toSoftExpireTimestamp(	6✔
328	softTtlMsIfZero,
329	COOL_DEFAULT_PROTECTION_SOFTTTL_TIME_RATIO
330	);
331
332	return cacheEntry	2✔
333	.toBuilder()	2✔
334	.value(cacheEntry.getValue())	3✔
335	.dataVersion(cacheEntry.getDataVersion())	3✔
336	.decisionVersion(wm.decisionVersion())	3✔
337	.decisionNodeId(wm.nodeId())	3✔
338	.decisionEpoch(wm.epoch())	3✔
339	.hardTtlMs(hardTtlMsIfZero)	2✔
340	.hardExpireAtMs(hardTtlExpireAtMs)	2✔
341	.softTtlMs(softTtlMsIfZero)	2✔
342	.softExpireAtMs(softTtlExpireAtMs)	2✔
343	.keyState(KeyState.COOL)	1✔
344	.build();	1✔
345	}
346
347	// No existing entry – nothing to cool
348	return existing;	2✔
349	});
350	log.debug("HotKey cooled by Worker: {}", wm.cacheKey());	5✔
351	}	1✔
352
353	/**
354	* Loads the current value from Redis for the key carried in the Worker message.
355	* <p>
356	* Any exception thrown by the {@code redisLoader} (connection timeout, Redis
357	* outage, serialization error) is caught and logged at WARN level. The caller
358	* is responsible for falling back to the degraded entry value when this method
359	* returns {@code null}.
360	*
361	* @param wm the Worker message containing the cache key to load; must not be null
362	* @return the value from Redis, or {@code null} if the key is absent or the
363	* load failed with an exception
364	*/
365	private Object loadFromRedis(WorkerMessage wm) {
366	try {
367	return redisLoader.apply(wm.cacheKey());	6✔
368	} catch (Exception e) {	1✔
369	log.warn("handleHot: Redis load failed for key={}, trying degraded entry", wm.cacheKey(), e);	6✔
370	if (sreRateLimiter != null) {	3!
371	sreRateLimiter.onFailed();	×
372	}
373	return null;	2✔
374	}
375	}
376	}

Hyshmily / hotkey / 28316658488

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