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

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.Internal;
import io.github.hyshmily.hotkey.cache.cachesupport.CacheExpireManager;
import io.github.hyshmily.hotkey.cache.loader.CacheLoader;
import io.github.hyshmily.hotkey.model.CacheEntry;
import io.github.hyshmily.hotkey.model.KeyState;
import io.github.hyshmily.hotkey.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 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
@Internal
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;

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

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