28705436176

Committed 04 Jul 2026 11:56AM UTC coverage: 90.132% (-0.3%) from 90.399%

Build # 28705436176

Build Type

push

github

Committed by

Hyshmily

Commit Message

Merge remote-tracking branch 'hotkey/master'

Coverage Stats

1261 of 1453 branches covered (86.79%)

Branch coverage included in aggregate %.

3589 of 3928 relevant lines covered (91.37%)

4.18 hits per line

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

90.2

/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) -> {
        long defultHotHardTtl = expireManager.getEffectiveHotHardTtlMs();
        long defultHotSoftTtl = expireManager.getEffectiveHotSoftTtlMs();

        // 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 expireManager.applyTtl(
            ce
              .toBuilder()
              .value(value)
              .decisionVersion(wm.decisionVersion())
              .decisionNodeId(wm.nodeId())
              .decisionEpoch(wm.epoch())
              .keyState(KeyState.HOT)
              .build(),
            defultHotHardTtl,
            defultHotSoftTtl
          );
        }

        return expireManager.applyTtl(
          CacheEntry.builder()
            .value(value)
            .dataVersion(0)
            .isVersionDegraded(false)
            .decisionVersion(wm.decisionVersion())
            .keyState(KeyState.HOT)
            .normalHardTtlMs(expireManager.getEffectiveHardTtlMs())
            .normalSoftTtlMs(expireManager.getEffectiveSoftTtlMs())
            .build(),
          defultHotHardTtl,
          defultHotSoftTtl
        );
      });
    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	long defultHotHardTtl = expireManager.getEffectiveHotHardTtlMs();	4✔
247	long defultHotSoftTtl = expireManager.getEffectiveHotSoftTtlMs();	4✔
248
249	// DCL second check – atomic with to write
250	if (existing instanceof CacheEntry ce) {	6✔
251	if (VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())) {	9!
252	return existing;	×
253	}
254
255	return expireManager.applyTtl(	5✔
256	ce
257	.toBuilder()	2✔
258	.value(value)	2✔
259	.decisionVersion(wm.decisionVersion())	3✔
260	.decisionNodeId(wm.nodeId())	3✔
261	.decisionEpoch(wm.epoch())	3✔
262	.keyState(KeyState.HOT)	1✔
263	.build(),	3✔
264	defultHotHardTtl,
265	defultHotSoftTtl
266	);
267	}
268
269	return expireManager.applyTtl(	4✔
270	CacheEntry.builder()	2✔
271	.value(value)	2✔
272	.dataVersion(0)	2✔
273	.isVersionDegraded(false)	2✔
274	.decisionVersion(wm.decisionVersion())	3✔
275	.keyState(KeyState.HOT)	3✔
276	.normalHardTtlMs(expireManager.getEffectiveHardTtlMs())	4✔
277	.normalSoftTtlMs(expireManager.getEffectiveSoftTtlMs())	2✔
278	.build(),	3✔
279	defultHotHardTtl,
280	defultHotSoftTtl
281	);
282	});
283	log.debug("HotKey promoted by Worker: {}", wm.cacheKey());	5✔
284	if (sreRateLimiter != null) {	3✔
285	sreRateLimiter.onSuccess();	3✔
286	}
287	}	1✔
288
289	/**
290	* Downgrades a cache key from {@link KeyState#HOT} to {@link KeyState#COOL},
291	* following a Worker COOL decision.
292	*
293	* <p>This effectively reverts the HOT promotion:
294	* <ul>
295	* <li>The cached value, {@code dataVersion}, degradation flag, and
296	* {@code decisionVersion} are all preserved — the data remains available.</li>
297	* <li>The hard TTL is reset to the normal value (from
298	* {@link CacheEntry#getNormalHardTtlMs()}).</li>
299	* <li>Soft expiration is fully disabled (both {@code softTtlMs} and
300	* {@code softExpireAtMs} set to {@code 0L}), so the entry stops being
301	* proactively refreshed.</li>
302	* </ul>
303	*
304	* <p>If no existing entry is present in L1, the COOL decision is a no-op —
305	* there is nothing to downgrade. The entry will be evicted naturally by
306	* Caffeine's capacity policy or a subsequent invalidation.
307	*
308	* @param wm the Worker message containing the COOL decision; must not be null
309	*/
310	private void handleCool(WorkerMessage wm) {
311	caffeineCache	2✔
312	.asMap()	2✔
313	.compute(wm.cacheKey(), (key, existing) -> {	6✔
314	if (	3✔
315	existing instanceof CacheEntry ce &&	5✔
316	VersionGuard.shouldSkipForWorker(ce, wm.decisionVersion(), wm.nodeId(), wm.epoch())	7✔
317	) {
318	return existing;	2✔
319	}
320
321	if (existing instanceof CacheEntry cacheEntry) {	6✔
322	long normalHardTtlMs = cacheEntry.getNormalHardTtlMs();	3✔
323	long normalSoftTtlMs = cacheEntry.getNormalSoftTtlMs();	3✔
324
325	long hardTtlMsIfZero = normalHardTtlMs > 0 ? normalHardTtlMs : COOL_DEFAULT_PROTECTION_HARDTTL_TIME;	7!
326	long softTtlMsIfZero = normalSoftTtlMs > 0 ? normalSoftTtlMs : COOL_DEFAULT_PROTECTION_SOFTTTL_TIME;	7!
327
328	long hardTtlExpireAtMs = expireManager.toHardExpireTimestamp(	6✔
329	hardTtlMsIfZero,
330	COOL_DEFAULT_PROTECTION_HARDTTL_TIME_RATIO
331	);
332	long softTtlExpireAtMs = expireManager.toSoftExpireTimestamp(	6✔
333	softTtlMsIfZero,
334	COOL_DEFAULT_PROTECTION_SOFTTTL_TIME_RATIO
335	);
336
337	return cacheEntry	2✔
338	.toBuilder()	2✔
339	.value(cacheEntry.getValue())	3✔
340	.dataVersion(cacheEntry.getDataVersion())	3✔
341	.decisionVersion(wm.decisionVersion())	3✔
342	.decisionNodeId(wm.nodeId())	3✔
343	.decisionEpoch(wm.epoch())	3✔
344	.hardTtlMs(hardTtlMsIfZero)	2✔
345	.hardExpireAtMs(hardTtlExpireAtMs)	2✔
346	.softTtlMs(softTtlMsIfZero)	2✔
347	.softExpireAtMs(softTtlExpireAtMs)	2✔
348	.keyState(KeyState.COOL)	1✔
349	.build();	1✔
350	}
351
352	// No existing entry – nothing to cool
353	return existing;	2✔
354	});
355	log.debug("HotKey cooled by Worker: {}", wm.cacheKey());	5✔
356	}	1✔
357
358	/**
359	* Loads the current value from Redis for the key carried in the Worker message.
360	* <p>
361	* Any exception thrown by the {@code redisLoader} (connection timeout, Redis
362	* outage, serialization error) is caught and logged at WARN level. The caller
363	* is responsible for falling back to the degraded entry value when this method
364	* returns {@code null}.
365	*
366	* @param wm the Worker message containing the cache key to load; must not be null
367	* @return the value from Redis, or {@code null} if the key is absent or the
368	* load failed with an exception
369	*/
370	private Object loadFromRedis(WorkerMessage wm) {
371	try {
372	return redisLoader.load(wm.cacheKey());	6✔
373	} catch (Exception e) {	1✔
374	log.warn("handleHot: Redis load failed for key={}, trying degraded entry", wm.cacheKey(), e);	6✔
375	if (sreRateLimiter != null) {	3!
376	sreRateLimiter.onFailed();	×
377	}
378	return null;	2✔
379	}
380	}
381	}

Hyshmily / hotkey / 28705436176

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