27890525626

Committed 21 Jun 2026 02:07AM UTC coverage: 93.615% (-0.1%) from 93.742%

Build # 27890525626

Build Type

push

github

Committed by

Hyshmily

Commit Message

feat: cross-Worker decisionVersion partitioning with nodeId/epoch + expectedWorkerCount

Core changes:
- VersionGuard: 4-param/5-param overloads with nodeId/epoch for per-Worker
  version space isolation
- WorkerMessage: add nodeId/epoch fields, broadcast as AMQP headers
- WorkerListener: propagate nodeId/epoch through DCL into CacheEntry
- CacheEntry: add decisionNodeId/decisionEpoch fields
- WorkerBroadcaster: stamp every HOT/COOL broadcast with nodeId + epoch
- WorkerAutoConfiguration: @Bean workerNodeId(), workerEpochCounter()
- HotKeyConstants: AMQP_HEADER_EPOCH constant

Cluster health:
- HotKeyProperties: expectedWorkerCount (default 0=dynamic)
- ClusterHealthView: majority-quorum (> expectedWorkerCount / 2) health check
- Wire through HotKeyAutoConfiguration, HotKeyAmqpAutoConfiguration,
  HotKeyRedisAutoConfiguration

Production hardening:
- CacheExpireManager: CompletableFuture.orTimeout() with
  RejectedExecutionException catch to prevent stuck refresh markers
- HotKeyStateMachine: updated state transitions
- RingManager: consistent-hash ring refinements
- ReportConsumer: ingest path improvements

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

Coverage Stats

1133 of 1265 branches covered (89.57%)

Branch coverage included in aggregate %.

149 of 160 new or added lines in 15 files covered. (93.13%)

1 existing line in 1 file now uncovered.

3280 of 3449 relevant lines covered (95.1%)

4.31 hits per line

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

89.39

/common/src/main/java/io/github/hyshmily/hotkey/cache/CacheExpireManager.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.cache;

import static io.github.hyshmily.hotkey.constants.HotKeyConstants.VERSION_DEFAULT;

import com.github.benmanes.caffeine.cache.Cache;
import io.github.hyshmily.hotkey.autoconfigure.HotKeyProperties;
import io.github.hyshmily.hotkey.model.CacheEntry;
import io.github.hyshmily.hotkey.model.KeyState;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.function.Supplier;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * Manages hard and soft TTL computation for {@link CacheEntry} instances.
 * <p>
 * Hard TTL controls Caffeine eviction; soft TTL controls stale-while-revalidate background refresh.
 * Each has a normal-key and hot-key variant, with an optional override taking precedence over the default.
 */
@Getter
@Slf4j
public class CacheExpireManager {

  /** The underlying L1 Caffeine cache instance. */
  private final Cache<String, Object> caffeineCache;
  /** Async executor for background refresh tasks. */
  private final Executor executor;
  /** TTL configuration providing normal and hot-key TTL values. */
  private final HotKeyProperties ttlConfig;
  /** Semaphore limiting concurrent background refresh operations (null if soft expire disabled). */
  // null when soft expire is disabled; always guarded by isSoftExpireEnabled() check
  private final Semaphore refreshLimiter;
  /** Per-key dedup for background refreshes — prevents concurrent refresh for the same key. */
  private final ConcurrentHashMap<String, CompletableFuture<Void>> pendingRefreshes = new ConcurrentHashMap<>();
  /** Whether TTL jitter is enabled (from config). */
  private final boolean ttlJitterEnabled;
  /** Jitter ratio applied to TTLs to prevent cache stampedes (from config, e.g. 0.1 = ±10%). */
  private final double ttlJitterRatio;

  private static final long refreshTimeoutSeconds = 30;

  /**
   * Creates a CacheExpireManager with the given Caffeine cache, executor, and TTL config.
   *
   * @param caffeineCache   the underlying L1 Caffeine cache
   * @param executor        async executor for background refresh
   * @param ttlConfig       TTL configuration (normal and hot-key variants)
   * @param refreshMaxPools maximum concurrent background refreshes (capped at 100)
   */
  public CacheExpireManager(
    Cache<String, Object> caffeineCache,
    Executor executor,
    HotKeyProperties ttlConfig,
    int refreshMaxPools
  ) {
    this.caffeineCache = caffeineCache;
    this.executor = executor;
    this.ttlConfig = ttlConfig;
    this.refreshLimiter = ttlConfig.isSoftExpireEnabled()
      ? new Semaphore(refreshMaxPools > 0 ? refreshMaxPools : 100)
      : null;
    this.ttlJitterEnabled = ttlConfig.isTtlJitterEnabled();
    this.ttlJitterRatio = ttlConfig.getTtlJitterRatio();
  }

  /**
   * Create a CacheExpireManager with explicit jitter control (for testing).
   */
  CacheExpireManager(
    Cache<String, Object> caffeineCache,
    Executor executor,
    HotKeyProperties ttlConfig,
    int refreshMaxPools,
    boolean ttlJitterEnabled,
    double ttlJitterRatio
  ) {
    this.caffeineCache = caffeineCache;
    this.executor = executor;
    this.ttlConfig = ttlConfig;
    this.refreshLimiter = ttlConfig.isSoftExpireEnabled()
      ? new Semaphore(refreshMaxPools > 0 ? refreshMaxPools : 100)
      : null;
    this.ttlJitterEnabled = ttlJitterEnabled;
    this.ttlJitterRatio = ttlJitterRatio;
  }

  /**
   * Whether any soft TTL is configured (normal or hot).
   *
   * @return {@code true} if soft expire is enabled in the configuration
   */
  public boolean isSoftExpireEnabled() {
    return ttlConfig.isSoftExpireEnabled();
  }

  /**
   * Hard expire timestamp from an explicit TTL duration.
   * Falls back to the normal-key default if {@code hardTtlMs <= 0}.
   *
   * @param hardTtlMs the hard TTL duration in milliseconds (&lt;= 0 uses configured default)
   * @return absolute epoch-ms timestamp for hard expiry
   */
  public long computeHardExpireAt(long hardTtlMs) {
    long effective = hardTtlMs > 0 ? hardTtlMs : ttlConfig.effectiveHardTtlMs();
    return toHardExpireTimestamp(effective);
  }

  /**
   * Hard expire timestamp for hot keys, using {@code default-hot-hard-ttl} / {@code hot-hard-ttl}.
   * Returns {@code Long.MAX_VALUE} if hot hard expire is disabled (TTL &lt;= 0).
   *
   * @return absolute epoch-ms timestamp for hot-key hard expiry
   */
  public long computeHotHardExpireAt() {
    return toHardExpireTimestamp(ttlConfig.effectiveHotHardTtlMs());
  }

  /**
   * Soft expire timestamp for hot keys, using {@code default-hot-soft-ttl} / {@code hot-soft-ttl}.
   *
   * @return absolute epoch-ms timestamp for hot-key soft expiry, or 0 if disabled
   */
  public long computeHotSoftExpireAt() {
    return toSoftExpireTimestamp(ttlConfig.effectiveHotSoftTtlMs());
  }

  /**
   * Soft expire timestamp from an explicit TTL duration.
   * Falls back to the normal-key default if {@code softTtlMs <= 0}. Returns 0 if soft expire is disabled.
   *
   * @param softTtlMs the soft TTL duration in milliseconds (&lt;= 0 uses configured default)
   * @return absolute epoch-ms timestamp for soft expiry, or 0 if disabled
   */
  public long computeSoftExpireAt(long softTtlMs) {
    if (!isSoftExpireEnabled()) {
      return 0L;
    }
    long effective = softTtlMs > 0 ? softTtlMs : ttlConfig.effectiveSoftTtlMs();
    return toSoftExpireTimestamp(effective);
  }

  /**
   * Effective hard TTL for normal keys (override > default).
   *
   * @return effective hard TTL duration in milliseconds
   */
  public long getEffectiveHardTtlMs() {
    return ttlConfig.effectiveHardTtlMs();
  }

  /**
   * Effective hard TTL for hot keys (override > default).
   *
   * @return effective hot hard TTL duration in milliseconds
   */
  public long getEffectiveHotHardTtlMs() {
    return ttlConfig.effectiveHotHardTtlMs();
  }

  /**
   * Effective soft TTL for normal keys (override > default).
   *
   * @return effective soft TTL duration in milliseconds
   */
  public long getEffectiveSoftTtlMs() {
    return ttlConfig.effectiveSoftTtlMs();
  }

  /**
   * Effective soft TTL for hot keys (override > default).
   *
   * @return effective hot soft TTL duration in milliseconds
   */
  public long getEffectiveHotSoftTtlMs() {
    return ttlConfig.effectiveHotSoftTtlMs();
  }

  /**
   * Convert a TTL duration (ms) to an absolute epoch-ms expiration timestamp.
   * Propagates {@link Long#MAX_VALUE} unchanged — used to signal permanent entries
   * (pure logical expiry with no hard TTL eviction).
   */
  private long toHardExpireTimestamp(long hardTtlMs) {
    if (hardTtlMs == Long.MAX_VALUE) {
      return Long.MAX_VALUE;
    }
    long jitter = ttlJitterEnabled
      ? (long) (hardTtlMs * ttlJitterRatio * ThreadLocalRandom.current().nextDouble(-1.0, 1.0))
      : 0L;

    return hardTtlMs > 0 ? System.currentTimeMillis() + Math.max(1, hardTtlMs + jitter) : Long.MAX_VALUE;
  }

  /**
   * Convert a soft TTL duration (ms) to an absolute epoch-ms expiration timestamp.
   * Applies configurable jitter (default ±10%) to prevent cache stampedes.
   * Returns 0 if soft expire is disabled or the TTL is non-positive.
   *
   * @param softTtlMs the soft TTL duration in milliseconds
   * @return absolute epoch-ms timestamp for soft expiry, or 0 if disabled
   */
  private long toSoftExpireTimestamp(long softTtlMs) {
    if (!isSoftExpireEnabled() || softTtlMs <= 0) {
      return 0L;
    }
    if (softTtlMs == Long.MAX_VALUE) {
      return Long.MAX_VALUE;
    }
    long jitter = ttlJitterEnabled
      ? (long) (softTtlMs * ttlJitterRatio * ThreadLocalRandom.current().nextDouble(-1.0, 1.0))
      : 0L;

    return System.currentTimeMillis() + Math.max(1, softTtlMs + jitter);
  }

  /**
   * Check whether the given key's soft TTL has expired.
   *
   * @param cacheKey the key to check for soft expiry
   * @return {@code true} if the entry's soft TTL has expired or the entry is absent
   * @throws IllegalStateException if soft expire is disabled
   */
  public boolean isSoftExpired(String cacheKey) {
    if (!isSoftExpireEnabled()) {
      throw new IllegalStateException(
        "CacheExpireManager soft expire is disabled, isSoftExpired() should not be called"
      );
    }
    Object cacheEntry = caffeineCache.getIfPresent(cacheKey);
    if (cacheEntry instanceof CacheEntry ce) {
      long expireAt = ce.getSoftExpireAtMs();
      return expireAt <= 0 || expireAt < System.currentTimeMillis();
    }
    return true;
  }

  /**
   * Trigger an async background refresh for the given key.
   * The refreshed entry preserves the existing {@code dataVersion},
   * {@code isVersionDegraded}, and {@code decisionVersion} to avoid
   * overwriting newer data or Worker decisions.
   * <p>Per-key dedup via {@link #pendingRefreshes} prevents concurrent refreshes
   * for the same key (TOCTOU stampede protection).  Rate-limited by a global
   * semaphore to bound total concurrent refreshes across all keys.
   * Skipped silently if the refresh limiter is exhausted or soft expire is disabled.
   *
   * @param cacheKey  the key to refresh in the background
   * @param reader    the value supplier for generating the refreshed value
   * @param softTtlMs the soft TTL duration in milliseconds applied to the refreshed entry
   */
  public void triggerBackgroundRefresh(String cacheKey, Supplier<?> reader, long softTtlMs) {
    if (!isSoftExpireEnabled()) {
      return;
    }

    // Per-key dedup: fast path — skip if a refresh is already in-flight for this key
    CompletableFuture<Void> inflight = pendingRefreshes.get(cacheKey);
    if (inflight != null && !inflight.isDone()) {
      return;
    }

    // Pre-check pendingRefreshes again under the imminent putIfAbsent race window
    // to avoid wasting a semaphore permit on a loser of per-key dedup.
    CompletableFuture<Void> marker = new CompletableFuture<>();
    CompletableFuture<Void> race = pendingRefreshes.putIfAbsent(cacheKey, marker);
    if (race != null) {
      return;
    }

    // refreshLimiter is null when soft expire is disabled; isSoftExpireEnabled() above
    // guarantees it is non-null here.
    if (!refreshLimiter.tryAcquire()) {
      pendingRefreshes.remove(cacheKey);
      marker.complete(null);
      log.debug("Refresh limiter blocked, skip background refresh: {}", cacheKey);
      return;
    }

    long refreshStartDataVersion = Optional.ofNullable(caffeineCache.getIfPresent(cacheKey))
      .filter(CacheEntry.class::isInstance)
      .map(CacheEntry.class::cast)
      .map(CacheEntry::getDataVersion)
      .orElse(VERSION_DEFAULT);

    // Wrap the async refresh with a timeout guard so a stuck supplier never leaks
    // a pending refresh marker. The orTimeout future is always wrapped because the
    // orTimeout call itself may throw RejectedExecutionException when the executor
    // is saturated — we catch that separately below.
    CompletableFuture<?> completableFuture = new CompletableFuture<>();

    try {
      completableFuture = CompletableFuture.supplyAsync(reader, executor).orTimeout(
        refreshTimeoutSeconds,
        TimeUnit.SECONDS
      );
    } catch (RejectedExecutionException e) {
      // Executor saturated — release the limiter permit and the pending-refresh
      // slot so the next read can retry immediately instead of being blocked forever.
      refreshLimiter.release();
      pendingRefreshes.remove(cacheKey);
      marker.complete(null);
      log.warn("Background refresh rejected by executor (saturated), key={}", cacheKey);
    }

    completableFuture.whenComplete((value, error) -> {
      try {
        if (error != null) {
          if (error instanceof TimeoutException) {
            log.warn("Background soft refresh timed out after {}s: {}", refreshTimeoutSeconds, cacheKey);
          } else {
            log.warn("Background soft refresh failed: {}", cacheKey, error);
          }
          return;
        }
        if (value != null) {
          caffeineCache
            .asMap()
            .compute(cacheKey, (k, existing) ->
              Optional.ofNullable(existing)
                .filter(CacheEntry.class::isInstance)
                .map(CacheEntry.class::cast)
                .map(cacheEntry -> {
                  if (cacheEntry.getDataVersion() > refreshStartDataVersion) {
                    log.debug("Async refresh discarded: newer version exists: {}", cacheKey);
                    return cacheEntry;
                  }
                  return cacheEntry
                    .toBuilder()
                    .value(value)
                    .softTtlMs(softTtlMs)
                    .softExpireAtMs(computeSoftExpireAt(softTtlMs))
                    .build();
                })
                .orElseGet(() ->
                  CacheEntry.builder()
                    .value(value)
                    .dataVersion(VERSION_DEFAULT)
                    .isVersionDegraded(false)
                    .decisionVersion(0L)
                    .hardTtlMs(0L)
                    .hardExpireAtMs(Long.MAX_VALUE)
                    .softTtlMs(softTtlMs)
                    .softExpireAtMs(computeSoftExpireAt(softTtlMs))
                    .keyState(KeyState.NORMAL)
                    .normalHardTtlMs(0L)
                    .normalSoftTtlMs(0L)
                    .build()
                )
            );
        }
      } finally {
        refreshLimiter.release();
        pendingRefreshes.remove(cacheKey);
        marker.complete(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.cache;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.VERSION_DEFAULT;
19
20	import com.github.benmanes.caffeine.cache.Cache;
21	import io.github.hyshmily.hotkey.autoconfigure.HotKeyProperties;
22	import io.github.hyshmily.hotkey.model.CacheEntry;
23	import io.github.hyshmily.hotkey.model.KeyState;
24	import java.util.Optional;
25	import java.util.concurrent.*;
26	import java.util.function.Supplier;
27	import lombok.Getter;
28	import lombok.extern.slf4j.Slf4j;
29
30	/**
31	* Manages hard and soft TTL computation for {@link CacheEntry} instances.
32	* <p>
33	* Hard TTL controls Caffeine eviction; soft TTL controls stale-while-revalidate background refresh.
34	* Each has a normal-key and hot-key variant, with an optional override taking precedence over the default.
35	*/
36	@Getter
37	@Slf4j	4✔
38	public class CacheExpireManager {
39
40	/** The underlying L1 Caffeine cache instance. */
41	private final Cache<String, Object> caffeineCache;
42	/** Async executor for background refresh tasks. */
43	private final Executor executor;
44	/** TTL configuration providing normal and hot-key TTL values. */
45	private final HotKeyProperties ttlConfig;
46	/** Semaphore limiting concurrent background refresh operations (null if soft expire disabled). */
47	// null when soft expire is disabled; always guarded by isSoftExpireEnabled() check
48	private final Semaphore refreshLimiter;
49	/** Per-key dedup for background refreshes — prevents concurrent refresh for the same key. */
50	private final ConcurrentHashMap<String, CompletableFuture<Void>> pendingRefreshes = new ConcurrentHashMap<>();	10✔
51	/** Whether TTL jitter is enabled (from config). */
52	private final boolean ttlJitterEnabled;
53	/** Jitter ratio applied to TTLs to prevent cache stampedes (from config, e.g. 0.1 = ±10%). */
54	private final double ttlJitterRatio;
55
56	private static final long refreshTimeoutSeconds = 30;
57
58	/**
59	* Creates a CacheExpireManager with the given Caffeine cache, executor, and TTL config.
60	*
61	* @param caffeineCache the underlying L1 Caffeine cache
62	* @param executor async executor for background refresh
63	* @param ttlConfig TTL configuration (normal and hot-key variants)
64	* @param refreshMaxPools maximum concurrent background refreshes (capped at 100)
65	*/
66	public CacheExpireManager(
67	Cache<String, Object> caffeineCache,
68	Executor executor,
69	HotKeyProperties ttlConfig,
70	int refreshMaxPools
71	) {	2✔
72	this.caffeineCache = caffeineCache;	3✔
73	this.executor = executor;	3✔
74	this.ttlConfig = ttlConfig;	3✔
75	this.refreshLimiter = ttlConfig.isSoftExpireEnabled()	4✔
76	? new Semaphore(refreshMaxPools > 0 ? refreshMaxPools : 100)	8!
77	: null;	2✔
78	this.ttlJitterEnabled = ttlConfig.isTtlJitterEnabled();	4✔
79	this.ttlJitterRatio = ttlConfig.getTtlJitterRatio();	4✔
80	}	1✔
81
82	/**
83	* Create a CacheExpireManager with explicit jitter control (for testing).
84	*/
85	CacheExpireManager(
86	Cache<String, Object> caffeineCache,
87	Executor executor,
88	HotKeyProperties ttlConfig,
89	int refreshMaxPools,
90	boolean ttlJitterEnabled,
91	double ttlJitterRatio
92	) {	2✔
93	this.caffeineCache = caffeineCache;	3✔
94	this.executor = executor;	3✔
95	this.ttlConfig = ttlConfig;	3✔
96	this.refreshLimiter = ttlConfig.isSoftExpireEnabled()	4!
97	? new Semaphore(refreshMaxPools > 0 ? refreshMaxPools : 100)	8!
98	: null;	1✔
99	this.ttlJitterEnabled = ttlJitterEnabled;	3✔
100	this.ttlJitterRatio = ttlJitterRatio;	3✔
101	}	1✔
102
103	/**
104	* Whether any soft TTL is configured (normal or hot).
105	*
106	* @return {@code true} if soft expire is enabled in the configuration
107	*/
108	public boolean isSoftExpireEnabled() {
109	return ttlConfig.isSoftExpireEnabled();	4✔
110	}
111
112	/**
113	* Hard expire timestamp from an explicit TTL duration.
114	* Falls back to the normal-key default if {@code hardTtlMs <= 0}.
115	*
116	* @param hardTtlMs the hard TTL duration in milliseconds (<= 0 uses configured default)
117	* @return absolute epoch-ms timestamp for hard expiry
118	*/
119	public long computeHardExpireAt(long hardTtlMs) {
120	long effective = hardTtlMs > 0 ? hardTtlMs : ttlConfig.effectiveHardTtlMs();	10✔
121	return toHardExpireTimestamp(effective);	4✔
122	}
123
124	/**
125	* Hard expire timestamp for hot keys, using {@code default-hot-hard-ttl} / {@code hot-hard-ttl}.
126	* Returns {@code Long.MAX_VALUE} if hot hard expire is disabled (TTL <= 0).
127	*
128	* @return absolute epoch-ms timestamp for hot-key hard expiry
129	*/
130	public long computeHotHardExpireAt() {
131	return toHardExpireTimestamp(ttlConfig.effectiveHotHardTtlMs());	6✔
132	}
133
134	/**
135	* Soft expire timestamp for hot keys, using {@code default-hot-soft-ttl} / {@code hot-soft-ttl}.
136	*
137	* @return absolute epoch-ms timestamp for hot-key soft expiry, or 0 if disabled
138	*/
139	public long computeHotSoftExpireAt() {
140	return toSoftExpireTimestamp(ttlConfig.effectiveHotSoftTtlMs());	6✔
141	}
142
143	/**
144	* Soft expire timestamp from an explicit TTL duration.
145	* Falls back to the normal-key default if {@code softTtlMs <= 0}. Returns 0 if soft expire is disabled.
146	*
147	* @param softTtlMs the soft TTL duration in milliseconds (<= 0 uses configured default)
148	* @return absolute epoch-ms timestamp for soft expiry, or 0 if disabled
149	*/
150	public long computeSoftExpireAt(long softTtlMs) {
151	if (!isSoftExpireEnabled()) {	3✔
152	return 0L;	2✔
153	}
154	long effective = softTtlMs > 0 ? softTtlMs : ttlConfig.effectiveSoftTtlMs();	7!
155	return toSoftExpireTimestamp(effective);	4✔
156	}
157
158	/**
159	* Effective hard TTL for normal keys (override > default).
160	*
161	* @return effective hard TTL duration in milliseconds
162	*/
163	public long getEffectiveHardTtlMs() {
164	return ttlConfig.effectiveHardTtlMs();	4✔
165	}
166
167	/**
168	* Effective hard TTL for hot keys (override > default).
169	*
170	* @return effective hot hard TTL duration in milliseconds
171	*/
172	public long getEffectiveHotHardTtlMs() {
173	return ttlConfig.effectiveHotHardTtlMs();	4✔
174	}
175
176	/**
177	* Effective soft TTL for normal keys (override > default).
178	*
179	* @return effective soft TTL duration in milliseconds
180	*/
181	public long getEffectiveSoftTtlMs() {
182	return ttlConfig.effectiveSoftTtlMs();	4✔
183	}
184
185	/**
186	* Effective soft TTL for hot keys (override > default).
187	*
188	* @return effective hot soft TTL duration in milliseconds
189	*/
190	public long getEffectiveHotSoftTtlMs() {
191	return ttlConfig.effectiveHotSoftTtlMs();	4✔
192	}
193
194	/**
195	* Convert a TTL duration (ms) to an absolute epoch-ms expiration timestamp.
196	* Propagates {@link Long#MAX_VALUE} unchanged — used to signal permanent entries
197	* (pure logical expiry with no hard TTL eviction).
198	*/
199	private long toHardExpireTimestamp(long hardTtlMs) {
200	if (hardTtlMs == Long.MAX_VALUE) {	4✔
201	return Long.MAX_VALUE;	2✔
202	}
203	long jitter = ttlJitterEnabled	3✔
204	? (long) (hardTtlMs * ttlJitterRatio * ThreadLocalRandom.current().nextDouble(-1.0, 1.0))	12✔
205	: 0L;	2✔
206
207	return hardTtlMs > 0 ? System.currentTimeMillis() + Math.max(1, hardTtlMs + jitter) : Long.MAX_VALUE;	13!
208	}
209
210	/**
211	* Convert a soft TTL duration (ms) to an absolute epoch-ms expiration timestamp.
212	* Applies configurable jitter (default ±10%) to prevent cache stampedes.
213	* Returns 0 if soft expire is disabled or the TTL is non-positive.
214	*
215	* @param softTtlMs the soft TTL duration in milliseconds
216	* @return absolute epoch-ms timestamp for soft expiry, or 0 if disabled
217	*/
218	private long toSoftExpireTimestamp(long softTtlMs) {
219	if (!isSoftExpireEnabled() \|\| softTtlMs <= 0) {	7!
220	return 0L;	×
221	}
222	if (softTtlMs == Long.MAX_VALUE) {	4✔
223	return Long.MAX_VALUE;	2✔
224	}
225	long jitter = ttlJitterEnabled	3!
226	? (long) (softTtlMs * ttlJitterRatio * ThreadLocalRandom.current().nextDouble(-1.0, 1.0))	12✔
227	: 0L;	1✔
228
229	return System.currentTimeMillis() + Math.max(1, softTtlMs + jitter);	8✔
230	}
231
232	/**
233	* Check whether the given key's soft TTL has expired.
234	*
235	* @param cacheKey the key to check for soft expiry
236	* @return {@code true} if the entry's soft TTL has expired or the entry is absent
237	* @throws IllegalStateException if soft expire is disabled
238	*/
239	public boolean isSoftExpired(String cacheKey) {
240	if (!isSoftExpireEnabled()) {	3✔
241	throw new IllegalStateException(	5✔
242	"CacheExpireManager soft expire is disabled, isSoftExpired() should not be called"
243	);
244	}
245	Object cacheEntry = caffeineCache.getIfPresent(cacheKey);	5✔
246	if (cacheEntry instanceof CacheEntry ce) {	6✔
247	long expireAt = ce.getSoftExpireAtMs();	3✔
248	return expireAt <= 0 \|\| expireAt < System.currentTimeMillis();	12✔
249	}
250	return true;	2✔
251	}
252
253	/**
254	* Trigger an async background refresh for the given key.
255	* The refreshed entry preserves the existing {@code dataVersion},
256	* {@code isVersionDegraded}, and {@code decisionVersion} to avoid
257	* overwriting newer data or Worker decisions.
258	* <p>Per-key dedup via {@link #pendingRefreshes} prevents concurrent refreshes
259	* for the same key (TOCTOU stampede protection). Rate-limited by a global
260	* semaphore to bound total concurrent refreshes across all keys.
261	* Skipped silently if the refresh limiter is exhausted or soft expire is disabled.
262	*
263	* @param cacheKey the key to refresh in the background
264	* @param reader the value supplier for generating the refreshed value
265	* @param softTtlMs the soft TTL duration in milliseconds applied to the refreshed entry
266	*/
267	public void triggerBackgroundRefresh(String cacheKey, Supplier<?> reader, long softTtlMs) {
268	if (!isSoftExpireEnabled()) {	3✔
269	return;	1✔
270	}
271
272	// Per-key dedup: fast path — skip if a refresh is already in-flight for this key
273	CompletableFuture<Void> inflight = pendingRefreshes.get(cacheKey);	6✔
274	if (inflight != null && !inflight.isDone()) {	5!
275	return;	1✔
276	}
277
278	// Pre-check pendingRefreshes again under the imminent putIfAbsent race window
279	// to avoid wasting a semaphore permit on a loser of per-key dedup.
280	CompletableFuture<Void> marker = new CompletableFuture<>();	4✔
281	CompletableFuture<Void> race = pendingRefreshes.putIfAbsent(cacheKey, marker);	7✔
282	if (race != null) {	2!
283	return;	×
284	}
285
286	// refreshLimiter is null when soft expire is disabled; isSoftExpireEnabled() above
287	// guarantees it is non-null here.
288	if (!refreshLimiter.tryAcquire()) {	4!
289	pendingRefreshes.remove(cacheKey);	×
290	marker.complete(null);	×
291	log.debug("Refresh limiter blocked, skip background refresh: {}", cacheKey);	×
292	return;	×
293	}
294
295	long refreshStartDataVersion = Optional.ofNullable(caffeineCache.getIfPresent(cacheKey))	7✔
296	.filter(CacheEntry.class::isInstance)	6✔
297	.map(CacheEntry.class::cast)	5✔
298	.map(CacheEntry::getDataVersion)	2✔
299	.orElse(VERSION_DEFAULT);	5✔
300
301	// Wrap the async refresh with a timeout guard so a stuck supplier never leaks
302	// a pending refresh marker. The orTimeout future is always wrapped because the
303	// orTimeout call itself may throw RejectedExecutionException when the executor
304	// is saturated — we catch that separately below.
305	CompletableFuture<?> completableFuture = new CompletableFuture<>();	4✔
306
307	try {
308	completableFuture = CompletableFuture.supplyAsync(reader, executor).orTimeout(	8✔
309	refreshTimeoutSeconds,
310	TimeUnit.SECONDS
311	);
312	} catch (RejectedExecutionException e) {	1✔
313	// Executor saturated — release the limiter permit and the pending-refresh
314	// slot so the next read can retry immediately instead of being blocked forever.
315	refreshLimiter.release();	3✔
316	pendingRefreshes.remove(cacheKey);	5✔
317	marker.complete(null);	4✔
318	log.warn("Background refresh rejected by executor (saturated), key={}", cacheKey);	4✔
319	}	1✔
320
321	completableFuture.whenComplete((value, error) -> {	9✔
322	try {
323	if (error != null) {	2✔
324	if (error instanceof TimeoutException) {	3!
NEW 325	log.warn("Background soft refresh timed out after {}s: {}", refreshTimeoutSeconds, cacheKey);	×
326	} else {
327	log.warn("Background soft refresh failed: {}", cacheKey, error);	5✔
328	}
329	return;	1✔
330	}
331	if (value != null) {	2✔
332	caffeineCache	2✔
333	.asMap()	8✔
334	.compute(cacheKey, (k, existing) ->	2✔
335	Optional.ofNullable(existing)	5✔
336	.filter(CacheEntry.class::isInstance)	6✔
337	.map(CacheEntry.class::cast)	10✔
338	.map(cacheEntry -> {	5✔
339	if (cacheEntry.getDataVersion() > refreshStartDataVersion) {	5✔
340	log.debug("Async refresh discarded: newer version exists: {}", cacheKey);	4✔
341	return cacheEntry;	2✔
342	}
343	return cacheEntry	2✔
344	.toBuilder()	2✔
345	.value(value)	2✔
346	.softTtlMs(softTtlMs)	3✔
347	.softExpireAtMs(computeSoftExpireAt(softTtlMs))	2✔
348	.build();	1✔
349	})
350	.orElseGet(() ->	1✔
351	CacheEntry.builder()	3✔
352	.value(value)	2✔
353	.dataVersion(VERSION_DEFAULT)	2✔
354	.isVersionDegraded(false)	2✔
355	.decisionVersion(0L)	2✔
356	.hardTtlMs(0L)	2✔
357	.hardExpireAtMs(Long.MAX_VALUE)	2✔
358	.softTtlMs(softTtlMs)	3✔
359	.softExpireAtMs(computeSoftExpireAt(softTtlMs))	3✔
360	.keyState(KeyState.NORMAL)	2✔
361	.normalHardTtlMs(0L)	2✔
362	.normalSoftTtlMs(0L)	1✔
363	.build()	1✔
364	)
365	);
366	}
367	} finally {
368	refreshLimiter.release();	3✔
369	pendingRefreshes.remove(cacheKey);	5✔
370	marker.complete(null);	4✔
371	}
372	});	1✔
373	}	1✔
374	}

Hyshmily / hotkey / 27890525626

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