28835396755

Committed 07 Jul 2026 01:38AM UTC coverage: 89.623% (-0.7%) from 90.336%

Build # 28835396755

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix and feat : fix known bugs and accept lz4 to wrap value for smaller memory,"wrap key" needs to consider

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

Coverage Stats

1325 of 1552 branches covered (85.37%)

Branch coverage included in aggregate %.

167 of 236 new or added lines in 12 files covered. (70.76%)

3745 of 4105 relevant lines covered (91.23%)

4.19 hits per line

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

58.44

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

import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Expiry;
import io.github.hyshmily.hotkey.Internal;
import io.github.hyshmily.hotkey.cache.HotKeyCache;
import io.github.hyshmily.hotkey.cache.cachesupport.ExpireManager;
import io.github.hyshmily.hotkey.cache.cachesupport.SingleFlight;
import io.github.hyshmily.hotkey.cache.cachesupport.impl.CircuitBreakerImpl;
import io.github.hyshmily.hotkey.cache.cachesupport.impl.ExpireManagerImpl;
import io.github.hyshmily.hotkey.cache.cachesupport.impl.SingleFlightImpl;
import io.github.hyshmily.hotkey.cache.codec.CacheCompressor;
import io.github.hyshmily.hotkey.cache.codec.DefaultWeigher;
import io.github.hyshmily.hotkey.cache.codec.Lz4CacheCompressor;
import io.github.hyshmily.hotkey.constants.HotKeyConstants;
import io.github.hyshmily.hotkey.hotkeydetector.HotKeyDetector;
import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.HeavyKeeper;
import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.TopK;
import io.github.hyshmily.hotkey.model.CacheEntry;
import io.github.hyshmily.hotkey.reporting.KeyReporter;
import io.github.hyshmily.hotkey.rule.RuleMatcher;
import io.github.hyshmily.hotkey.rule.impl.RuleMatcherImpl;
import io.github.hyshmily.hotkey.sharding.HealthView;
import io.github.hyshmily.hotkey.sharding.impl.HealthViewImpl;
import io.github.hyshmily.hotkey.sync.local.CacheSyncPublisher;
import io.github.hyshmily.hotkey.util.version.VersionController;
import io.github.hyshmily.hotkey.util.version.impl.VersionControllerImpl;
import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.jspecify.annotations.NonNull;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * App-side autoconfiguration for the HotKey library.
 *
 * <p>Creates the app-side {@link TopK} detector (HeavyKeeper), L1 Caffeine
 * cache, {@link SingleFlight} deduplication layer, executor, and the
 * primary {@link HotKeyCache} (without Redis version tracking when Redis
 * is absent).
 *
 * <p><b>Condition:</b> this configuration is <em>skipped</em> when the
 * Worker is active ({@code hotkey.worker.enabled=true}). It runs when
 * Worker is disabled or the property is absent.
 */
@Internal
@AutoConfiguration(after = RedisAutoConfiguration.class)
@ConditionalOnProperty(prefix = "hotkey.worker", name = "enabled", havingValue = "false", matchIfMissing = true)
@EnableConfigurationProperties(HotKeyProperties.class)
@Slf4j
public class HotKeyAutoConfiguration {

  /**
   * Create the app-side TopK instance (HeavyKeeper) as a standalone bean.
   *
   * <p>The HeavyKeeper uses a Count-Min Sketch augmented with a minimum-count
   * threshold and exponential decay to track the top-K hottest keys with high
   * accuracy and low memory footprint. Configuration parameters (width, depth,
   * decay, minCount, topK capacity, expelled queue capacity) are read from
   * {@link HotKeyProperties}.
   *
   * @param properties the HotKey configuration properties (never {@code null})
   * @return a new HeavyKeeper TopK instance
   */
  @Bean
  @ConditionalOnMissingBean
  public HeavyKeeper heavyKeeper(HotKeyProperties properties) {
    return new HeavyKeeper(
      properties.getTopK(),
      properties.getWidth(),
      properties.getDepth(),
      properties.getDecay(),
      properties.getMinCount(),
      properties.getExpelledQueueCapacity(),
      properties.getSketchWindowCount()
    );
  }

  /**
   * Create the app-side {@link HotKeyDetector} facade that wraps the HeavyKeeper TopK
   * and schedules periodic decay.
   *
   * <p>The detector provides the primary hot-key detection API ({@code add()}, {@code list()},
   * {@code total()}) and schedules the periodic HeavyKeeper decay using the shared scheduler.
   * This bean is the {@code @Qualifier("hotKeyDetector")} target for injection into
   * {@link HotKeyCache} and other components.
   *
   * @param heavyKeeper      the HeavyKeeper TopK instance (never {@code null})
   * @param hotKeyScheduler  the shared scheduler for periodic tasks (never {@code null})
   * @return a new HotKeyDetector instance
   */
  @Bean
  @ConditionalOnMissingBean
  public HotKeyDetector hotKeyDetector(
    HeavyKeeper heavyKeeper,
    @Qualifier("hotKeyScheduler") ScheduledExecutorService hotKeyScheduler
  ) {
    return new HotKeyDetector(heavyKeeper, hotKeyScheduler);
  }

  /**
   * Create the SingleFlight deduplication layer for concurrent cache-load requests.
   *
   * <p>When multiple threads request the same key simultaneously (a cache miss), the
   * first caller triggers the load and subsequent callers wait for the same result
   * rather than duplicating the load. Configuration parameters (max in-flight entries,
   * TTL, timeout) are read from {@link HotKeyProperties}.
   *
   * @param properties     the HotKey configuration properties (never {@code null})
   * @param hotKeyExecutor the dedicated HotKey executor for async load execution (never {@code null})
   * @return a new SingleFlight instance
   */
  @Bean
  @ConditionalOnMissingBean
  public SingleFlight singleFlight(HotKeyProperties properties, @Qualifier("hotKeyExecutor") Executor hotKeyExecutor) {
    return new SingleFlightImpl(
      properties.getInflightMaxSize(),
      properties.getInflightTtlSeconds(),
      properties.getInflightTimeoutSeconds(),
      hotKeyExecutor,
      new CircuitBreakerImpl(properties.getCircuitBreaker())
    );
  }

  /**
   * Create the soft/hard expiration manager that manages time-based eviction.
   *
   * <p>The soft-expire mechanism triggers an asynchronous refresh when a configurable
   * portion of the TTL has elapsed, serving stale data while fetching a fresh value
   * in the background. The hard-expire is the absolute maximum TTL enforced at the
   * Caffeine level. The refresh pool size limits concurrent background refresh tasks
   * to prevent resource exhaustion under high cache-miss rates.
   *
   * @param hotLocalCache  the L1 Caffeine cache (never {@code null})
   * @param hotKeyExecutor the dedicated HotKey executor for async refresh tasks (never {@code null})
   * @param properties     the HotKey configuration properties (never {@code null})
   * @return a new ExpireManagerImpl instance
   */
  @Bean
  @ConditionalOnMissingBean
  public ExpireManager expireManager(
    Cache<String, Object> hotLocalCache,
    @Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
    HotKeyProperties properties,
    CacheCompressor compressor
  ) {
    return new ExpireManagerImpl(
      hotLocalCache,
      hotKeyExecutor,
      properties,
      properties.getRefreshMaxPools(),
      compressor
    );
  }

  /**
   * Create the dedicated thread-pool executor for asynchronous cache operations.
   *
   * <p>This executor handles all async operations in the HotKey data path: cache loading
   * via SingleFlight, soft-expiry refresh tasks, and cross-instance broadcast callbacks.
   * Uses a bounded thread pool to limit concurrent AMQP channel usage (RabbitMQ's
   * {@code CachingConnectionFactory} associates channels with platform threads, so
   * unbounded virtual-thread concurrency causes channel-open timeouts). The pool is
   * configured with core/max pool size, bounded queue capacity, and a rejection policy
   * that throws {@link RejectedExecutionException} when the queue is full. On shutdown,
   * in-progress tasks are allowed to complete with a 60-second grace period.
   *
   * @param properties the HotKey configuration properties (never {@code null})
   * @return a configured {@link ThreadPoolTaskExecutor}
   */
  @Bean("hotKeyExecutor")
  @ConditionalOnMissingBean(name = "hotKeyExecutor")
  public Executor hotKeyExecutor(HotKeyProperties properties) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(0);
    executor.setMaxPoolSize(properties.getExecutorMaxPoolSize());
    executor.setQueueCapacity(properties.getExecutorQueueCapacity());
    executor.setAllowCoreThreadTimeOut(true);
    executor.setThreadNamePrefix(HotKeyConstants.THREAD_PREFIX_HOTKEY);
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);
    executor.setRejectedExecutionHandler((r, exe) -> {
      log.warn(
        "HotKey executor task rejected: corePool={}, maxPool={}, queueCapacity={}",
        properties.getExecutorCorePoolSize(),
        properties.getExecutorMaxPoolSize(),
        properties.getExecutorQueueCapacity()
      );
      throw new RejectedExecutionException("HotKey executor queue full");
    });
    executor.initialize();
    return executor;
  }

  /**
   * Create the {@link RuleMatcher} for key matching against user-defined rules.
   *
   * <p>This variant is wired with an empty Redis provider and an optional sync
   * publisher, since no {@link StringRedisTemplate} is available in the non-Redis
   * deployment mode. Rules are kept in-memory only and do not survive restarts.
   * When Redis is available, {@link HotKeyRedisAutoConfiguration#ruleMatcher}
   * takes over with Redis persistence.
   *
   * @param publisherProvider optional provider for the cache sync publisher (may be absent)
   * @return a new RuleMatcher instance (in-memory only)
   */
  @Bean
  @ConditionalOnMissingBean({ RuleMatcher.class, StringRedisTemplate.class })
  public RuleMatcher ruleMatcher(ObjectProvider<CacheSyncPublisher> publisherProvider) {
    return new RuleMatcherImpl(Optional.empty(), Optional.ofNullable(publisherProvider.getIfAvailable()));
  }

  /**
   * Create the {@link HotKeyCache} (non-Redis variant).
   *
   * <p>Only active when {@code RedisTemplate} is absent; otherwise
   * {@link HotKeyRedisAutoConfiguration#hotKeyCache} takes over. Creates a
   * {@link VersionController} with an empty Redis provider (falling back to
   * node-local counters) and default ring manager / health view when none are
   * available from the application context.
   *
   * @param hotKeyDetector            the app-side TopK detector (never {@code null})
   * @param hotLocalCache             the L1 Caffeine cache (never {@code null})
   * @param singleFlight              the deduplication layer (never {@code null})
   * @param expireManager             the soft/hard expiration manager (never {@code null})
   * @param syncPublisher             optional cache sync publisher (may be absent)
   * @param hotKeyReporter            optional hot key reporter (may be absent; e.g. in Worker-only mode)
   * @param hotKeyExecutor            the dedicated HotKey executor (never {@code null})
   * @param properties                the HotKey configuration properties (never {@code null})
   * @param ruleMatcher               the rule matcher instance (never {@code null})
   * @param healthViewProvider        provider for the cluster health view (creates default if absent)
   * @return a new HotKeyCache instance with node-local version tracking
   */
  @Bean
  @ConditionalOnMissingBean(type = "org.springframework.data.redis.core.RedisTemplate")
  public HotKeyCache hotKeyCache(
    @Qualifier("hotKeyDetector") HotKeyDetector hotKeyDetector,
    Cache<String, Object> hotLocalCache,
    SingleFlight singleFlight,
    ExpireManager expireManager,
    Optional<CacheSyncPublisher> syncPublisher,
    Optional<KeyReporter> hotKeyReporter,
    @Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
    HotKeyProperties properties,
    RuleMatcher ruleMatcher,
    ObjectProvider<HealthView> healthViewProvider,
    CacheCompressor compressor
  ) {
    return new HotKeyCache(
      hotKeyDetector,
      hotLocalCache,
      singleFlight,
      expireManager,
      hotKeyExecutor,
      syncPublisher,
      hotKeyReporter,
      ruleMatcher,
      new VersionControllerImpl(Optional.empty(), properties.getVersionKeyTtlMinutes()),
      properties,
      healthViewProvider.getIfAvailable(() ->
        new HealthViewImpl(
          properties.getExpectedWorkerCount(),
          properties.getHeartbeat().getTimeoutMs(),
          properties.getHeartbeat().getDegradeAfterFailures()
        )
      ),
      compressor
    );
  }

  @Bean
  @ConditionalOnMissingBean
  public CacheCompressor cacheCompressor() {
    try {
      return new Lz4CacheCompressor();
    } catch (NoClassDefFoundError e) {
      log.warn("lz4-java not on classpath, cache compression disabled");
      return CacheCompressor.NONE;
    }
  }

  /**
   * Create the L1 Caffeine cache instance.
   *
   * <p>Time-based expiry operates at the <em>Caffeine</em> level via a custom
   * {@link Expiry} implementation, computing remaining nanoseconds from
   * . Entries with
   * {@code hardExpireAtMs == Long.MAX_VALUE} are purely logical-expiry
   * — Caffeine never evicts them by time; they live until size eviction or
   * manual invalidation. Reads never extend the expiry duration (no read-based
   * refresh), ensuring predictable TTL behavior.
   *
   * <p>Eviction strategy: {@code max-weight} (> 0) enables memory-weighted
   * eviction with {@link DefaultWeigher}; otherwise {@code max-size} limits
   * entry count. Time-based TTL for entries without an explicit hard-expire
   * timestamp defaults to {@code hotkey.local.local-cache-ttl-minutes}.
   *
   * @param properties the HotKey configuration properties (never {@code null})
   * @return a configured Caffeine {@link Cache} instance
   */
  @Bean
  @ConditionalOnMissingBean
  public Cache<String, Object> hotLocalCache(HotKeyProperties properties) {
    var cfg = properties.getCache();
    Caffeine<Object, Object> builder = Caffeine.newBuilder();
    if (cfg.getMaxWeight() > 0) {
      builder.maximumWeight(cfg.getMaxWeight()).weigher(DefaultWeigher.INSTANCE);
    } else {
      builder.maximumSize(cfg.getMaxSize());
    }
    builder.expireAfter(
      new Expiry<>() {
        /**
         * Compute the time-to-live for a newly created cache entry.
         * Returns {@link Long#MAX_VALUE} for pure logical-expiry entries
         * (where {@code hardExpireAtMs == Long.MAX_VALUE}); otherwise
         * computes the remaining wall-clock time.
         *
         * @param key              the cache key
         * @param value            the cache value (expected to be a {@link CacheEntry})
         * @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
         * @return the expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
         */
        @Override
        public long expireAfterCreate(@NonNull Object key, @NonNull Object value, long currentTimeNanos) {
          if (value instanceof CacheEntry entry) {
            if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {
              // Pure logical expiry: Caffeine never time-evicts this entry.
              // See Expiry Javadoc: Long.MAX_VALUE signals "no expiration".
              return Long.MAX_VALUE;
            }
            long remainingMs = entry.getHardExpireAtMs() - currentTimeMillis();
            return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));
          }
          return TimeUnit.MINUTES.toNanos(properties.getLocalCacheTtlMinutes());
        }

        /**
         * Re-compute the expiry duration after an entry is updated.
         * Preserves pure logical expiry across updates; otherwise
         * recalculates from the entry's {@code hardExpireAtMs}.
         *
         * @param key              the cache key
         * @param value            the cache value (expected to be a {@link CacheEntry})
         * @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
         * @param currentDuration  the current expiry duration in nanoseconds
         * @return the updated expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
         */
        @Override
        public long expireAfterUpdate(
          @NonNull Object key,
          @NonNull Object value,
          long currentTimeNanos,
          long currentDuration
        ) {
          if (value instanceof CacheEntry entry) {
            if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {
              // Preserve pure logical expiry across updates (e.g. broadcast refresh).
              return Long.MAX_VALUE;
            }
            long remainingMs = entry.getHardExpireAtMs() - currentTimeMillis();
            return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));
          }
          return currentDuration;
        }

        /**
         * Preserve the current expiry duration on read — reads never
         * extend or shorten the entry's time-to-live.
         *
         * @param key              the cache key
         * @param value            the cache value
         * @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
         * @param currentDuration  the current expiry duration in nanoseconds
         * @return the unchanged expiry duration in nanoseconds
         */
        @Override
        public long expireAfterRead(
          @NonNull Object key,
          @NonNull Object value,
          long currentTimeNanos,
          long currentDuration
        ) {
          return currentDuration;
        }
      }
    );
    return builder.build();
  }
}

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.autoconfigure;
17
18	import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;
19
20	import com.github.benmanes.caffeine.cache.Cache;
21	import com.github.benmanes.caffeine.cache.Caffeine;
22	import com.github.benmanes.caffeine.cache.Expiry;
23	import io.github.hyshmily.hotkey.Internal;
24	import io.github.hyshmily.hotkey.cache.HotKeyCache;
25	import io.github.hyshmily.hotkey.cache.cachesupport.ExpireManager;
26	import io.github.hyshmily.hotkey.cache.cachesupport.SingleFlight;
27	import io.github.hyshmily.hotkey.cache.cachesupport.impl.CircuitBreakerImpl;
28	import io.github.hyshmily.hotkey.cache.cachesupport.impl.ExpireManagerImpl;
29	import io.github.hyshmily.hotkey.cache.cachesupport.impl.SingleFlightImpl;
30	import io.github.hyshmily.hotkey.cache.codec.CacheCompressor;
31	import io.github.hyshmily.hotkey.cache.codec.DefaultWeigher;
32	import io.github.hyshmily.hotkey.cache.codec.Lz4CacheCompressor;
33	import io.github.hyshmily.hotkey.constants.HotKeyConstants;
34	import io.github.hyshmily.hotkey.hotkeydetector.HotKeyDetector;
35	import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.HeavyKeeper;
36	import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.TopK;
37	import io.github.hyshmily.hotkey.model.CacheEntry;
38	import io.github.hyshmily.hotkey.reporting.KeyReporter;
39	import io.github.hyshmily.hotkey.rule.RuleMatcher;
40	import io.github.hyshmily.hotkey.rule.impl.RuleMatcherImpl;
41	import io.github.hyshmily.hotkey.sharding.HealthView;
42	import io.github.hyshmily.hotkey.sharding.impl.HealthViewImpl;
43	import io.github.hyshmily.hotkey.sync.local.CacheSyncPublisher;
44	import io.github.hyshmily.hotkey.util.version.VersionController;
45	import io.github.hyshmily.hotkey.util.version.impl.VersionControllerImpl;
46	import java.util.Optional;
47	import java.util.concurrent.Executor;
48	import java.util.concurrent.RejectedExecutionException;
49	import java.util.concurrent.ScheduledExecutorService;
50	import java.util.concurrent.TimeUnit;
51	import lombok.extern.slf4j.Slf4j;
52	import org.jspecify.annotations.NonNull;
53	import org.springframework.beans.factory.ObjectProvider;
54	import org.springframework.beans.factory.annotation.Qualifier;
55	import org.springframework.boot.autoconfigure.AutoConfiguration;
56	import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
57	import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
58	import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
59	import org.springframework.boot.context.properties.EnableConfigurationProperties;
60	import org.springframework.context.annotation.Bean;
61	import org.springframework.data.redis.core.StringRedisTemplate;
62	import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
63
64	/**
65	* App-side autoconfiguration for the HotKey library.
66	*
67	* <p>Creates the app-side {@link TopK} detector (HeavyKeeper), L1 Caffeine
68	* cache, {@link SingleFlight} deduplication layer, executor, and the
69	* primary {@link HotKeyCache} (without Redis version tracking when Redis
70	* is absent).
71	*
72	* <p><b>Condition:</b> this configuration is <em>skipped</em> when the
73	* Worker is active ({@code hotkey.worker.enabled=true}). It runs when
74	* Worker is disabled or the property is absent.
75	*/
76	@Internal
77	@AutoConfiguration(after = RedisAutoConfiguration.class)
78	@ConditionalOnProperty(prefix = "hotkey.worker", name = "enabled", havingValue = "false", matchIfMissing = true)
79	@EnableConfigurationProperties(HotKeyProperties.class)
80	@Slf4j	4✔
81	public class HotKeyAutoConfiguration {	3✔
82
83	/**
84	* Create the app-side TopK instance (HeavyKeeper) as a standalone bean.
85	*
86	* <p>The HeavyKeeper uses a Count-Min Sketch augmented with a minimum-count
87	* threshold and exponential decay to track the top-K hottest keys with high
88	* accuracy and low memory footprint. Configuration parameters (width, depth,
89	* decay, minCount, topK capacity, expelled queue capacity) are read from
90	* {@link HotKeyProperties}.
91	*
92	* @param properties the HotKey configuration properties (never {@code null})
93	* @return a new HeavyKeeper TopK instance
94	*/
95	@Bean
96	@ConditionalOnMissingBean
97	public HeavyKeeper heavyKeeper(HotKeyProperties properties) {
98	return new HeavyKeeper(	4✔
99	properties.getTopK(),	2✔
100	properties.getWidth(),	2✔
101	properties.getDepth(),	2✔
102	properties.getDecay(),	2✔
103	properties.getMinCount(),	2✔
104	properties.getExpelledQueueCapacity(),	2✔
105	properties.getSketchWindowCount()	2✔
106	);
107	}
108
109	/**
110	* Create the app-side {@link HotKeyDetector} facade that wraps the HeavyKeeper TopK
111	* and schedules periodic decay.
112	*
113	* <p>The detector provides the primary hot-key detection API ({@code add()}, {@code list()},
114	* {@code total()}) and schedules the periodic HeavyKeeper decay using the shared scheduler.
115	* This bean is the {@code @Qualifier("hotKeyDetector")} target for injection into
116	* {@link HotKeyCache} and other components.
117	*
118	* @param heavyKeeper the HeavyKeeper TopK instance (never {@code null})
119	* @param hotKeyScheduler the shared scheduler for periodic tasks (never {@code null})
120	* @return a new HotKeyDetector instance
121	*/
122	@Bean
123	@ConditionalOnMissingBean
124	public HotKeyDetector hotKeyDetector(
125	HeavyKeeper heavyKeeper,
126	@Qualifier("hotKeyScheduler") ScheduledExecutorService hotKeyScheduler
127	) {
128	return new HotKeyDetector(heavyKeeper, hotKeyScheduler);	6✔
129	}
130
131	/**
132	* Create the SingleFlight deduplication layer for concurrent cache-load requests.
133	*
134	* <p>When multiple threads request the same key simultaneously (a cache miss), the
135	* first caller triggers the load and subsequent callers wait for the same result
136	* rather than duplicating the load. Configuration parameters (max in-flight entries,
137	* TTL, timeout) are read from {@link HotKeyProperties}.
138	*
139	* @param properties the HotKey configuration properties (never {@code null})
140	* @param hotKeyExecutor the dedicated HotKey executor for async load execution (never {@code null})
141	* @return a new SingleFlight instance
142	*/
143	@Bean
144	@ConditionalOnMissingBean
145	public SingleFlight singleFlight(HotKeyProperties properties, @Qualifier("hotKeyExecutor") Executor hotKeyExecutor) {
146	return new SingleFlightImpl(	4✔
147	properties.getInflightMaxSize(),	2✔
148	properties.getInflightTtlSeconds(),	2✔
149	properties.getInflightTimeoutSeconds(),	5✔
150	hotKeyExecutor,
151	new CircuitBreakerImpl(properties.getCircuitBreaker())	3✔
152	);
153	}
154
155	/**
156	* Create the soft/hard expiration manager that manages time-based eviction.
157	*
158	* <p>The soft-expire mechanism triggers an asynchronous refresh when a configurable
159	* portion of the TTL has elapsed, serving stale data while fetching a fresh value
160	* in the background. The hard-expire is the absolute maximum TTL enforced at the
161	* Caffeine level. The refresh pool size limits concurrent background refresh tasks
162	* to prevent resource exhaustion under high cache-miss rates.
163	*
164	* @param hotLocalCache the L1 Caffeine cache (never {@code null})
165	* @param hotKeyExecutor the dedicated HotKey executor for async refresh tasks (never {@code null})
166	* @param properties the HotKey configuration properties (never {@code null})
167	* @return a new ExpireManagerImpl instance
168	*/
169	@Bean
170	@ConditionalOnMissingBean
171	public ExpireManager expireManager(
172	Cache<String, Object> hotLocalCache,
173	@Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
174	HotKeyProperties properties,
175	CacheCompressor compressor
176	) {
177	return new ExpireManagerImpl(	7✔
178	hotLocalCache,
179	hotKeyExecutor,
180	properties,
181	properties.getRefreshMaxPools(),	3✔
182	compressor
183	);
184	}
185
186	/**
187	* Create the dedicated thread-pool executor for asynchronous cache operations.
188	*
189	* <p>This executor handles all async operations in the HotKey data path: cache loading
190	* via SingleFlight, soft-expiry refresh tasks, and cross-instance broadcast callbacks.
191	* Uses a bounded thread pool to limit concurrent AMQP channel usage (RabbitMQ's
192	* {@code CachingConnectionFactory} associates channels with platform threads, so
193	* unbounded virtual-thread concurrency causes channel-open timeouts). The pool is
194	* configured with core/max pool size, bounded queue capacity, and a rejection policy
195	* that throws {@link RejectedExecutionException} when the queue is full. On shutdown,
196	* in-progress tasks are allowed to complete with a 60-second grace period.
197	*
198	* @param properties the HotKey configuration properties (never {@code null})
199	* @return a configured {@link ThreadPoolTaskExecutor}
200	*/
201	@Bean("hotKeyExecutor")
202	@ConditionalOnMissingBean(name = "hotKeyExecutor")
203	public Executor hotKeyExecutor(HotKeyProperties properties) {
204	ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();	4✔
205	executor.setCorePoolSize(0);	3✔
206	executor.setMaxPoolSize(properties.getExecutorMaxPoolSize());	4✔
207	executor.setQueueCapacity(properties.getExecutorQueueCapacity());	4✔
208	executor.setAllowCoreThreadTimeOut(true);	3✔
209	executor.setThreadNamePrefix(HotKeyConstants.THREAD_PREFIX_HOTKEY);	3✔
210	executor.setWaitForTasksToCompleteOnShutdown(true);	3✔
211	executor.setAwaitTerminationSeconds(60);	3✔
212	executor.setRejectedExecutionHandler((r, exe) -> {	4✔
213	log.warn(	×
214	"HotKey executor task rejected: corePool={}, maxPool={}, queueCapacity={}",
215	properties.getExecutorCorePoolSize(),	×
216	properties.getExecutorMaxPoolSize(),	×
217	properties.getExecutorQueueCapacity()	×
218	);
219	throw new RejectedExecutionException("HotKey executor queue full");	×
220	});
221	executor.initialize();	2✔
222	return executor;	2✔
223	}
224
225	/**
226	* Create the {@link RuleMatcher} for key matching against user-defined rules.
227	*
228	* <p>This variant is wired with an empty Redis provider and an optional sync
229	* publisher, since no {@link StringRedisTemplate} is available in the non-Redis
230	* deployment mode. Rules are kept in-memory only and do not survive restarts.
231	* When Redis is available, {@link HotKeyRedisAutoConfiguration#ruleMatcher}
232	* takes over with Redis persistence.
233	*
234	* @param publisherProvider optional provider for the cache sync publisher (may be absent)
235	* @return a new RuleMatcher instance (in-memory only)
236	*/
237	@Bean
238	@ConditionalOnMissingBean({ RuleMatcher.class, StringRedisTemplate.class })
239	public RuleMatcher ruleMatcher(ObjectProvider<CacheSyncPublisher> publisherProvider) {
240	return new RuleMatcherImpl(Optional.empty(), Optional.ofNullable(publisherProvider.getIfAvailable()));	9✔
241	}
242
243	/**
244	* Create the {@link HotKeyCache} (non-Redis variant).
245	*
246	* <p>Only active when {@code RedisTemplate} is absent; otherwise
247	* {@link HotKeyRedisAutoConfiguration#hotKeyCache} takes over. Creates a
248	* {@link VersionController} with an empty Redis provider (falling back to
249	* node-local counters) and default ring manager / health view when none are
250	* available from the application context.
251	*
252	* @param hotKeyDetector the app-side TopK detector (never {@code null})
253	* @param hotLocalCache the L1 Caffeine cache (never {@code null})
254	* @param singleFlight the deduplication layer (never {@code null})
255	* @param expireManager the soft/hard expiration manager (never {@code null})
256	* @param syncPublisher optional cache sync publisher (may be absent)
257	* @param hotKeyReporter optional hot key reporter (may be absent; e.g. in Worker-only mode)
258	* @param hotKeyExecutor the dedicated HotKey executor (never {@code null})
259	* @param properties the HotKey configuration properties (never {@code null})
260	* @param ruleMatcher the rule matcher instance (never {@code null})
261	* @param healthViewProvider provider for the cluster health view (creates default if absent)
262	* @return a new HotKeyCache instance with node-local version tracking
263	*/
264	@Bean
265	@ConditionalOnMissingBean(type = "org.springframework.data.redis.core.RedisTemplate")
266	public HotKeyCache hotKeyCache(
267	@Qualifier("hotKeyDetector") HotKeyDetector hotKeyDetector,
268	Cache<String, Object> hotLocalCache,
269	SingleFlight singleFlight,
270	ExpireManager expireManager,
271	Optional<CacheSyncPublisher> syncPublisher,
272	Optional<KeyReporter> hotKeyReporter,
273	@Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
274	HotKeyProperties properties,
275	RuleMatcher ruleMatcher,
276	ObjectProvider<HealthView> healthViewProvider,
277	CacheCompressor compressor
278	) {
279	return new HotKeyCache(	13✔
280	hotKeyDetector,
281	hotLocalCache,
282	singleFlight,
283	expireManager,
284	hotKeyExecutor,
285	syncPublisher,
286	hotKeyReporter,
287	ruleMatcher,
288	new VersionControllerImpl(Optional.empty(), properties.getVersionKeyTtlMinutes()),	8✔
289	properties,
290	healthViewProvider.getIfAvailable(() ->	4✔
291	new HealthViewImpl(	×
292	properties.getExpectedWorkerCount(),	×
293	properties.getHeartbeat().getTimeoutMs(),	×
294	properties.getHeartbeat().getDegradeAfterFailures()	×
295	)
296	),
297	compressor
298	);
299	}
300
301	@Bean
302	@ConditionalOnMissingBean
303	public CacheCompressor cacheCompressor() {
304	try {
305	return new Lz4CacheCompressor();	4✔
NEW 306	} catch (NoClassDefFoundError e) {	×
NEW 307	log.warn("lz4-java not on classpath, cache compression disabled");	×
NEW 308	return CacheCompressor.NONE;	×
309	}
310	}
311
312	/**
313	* Create the L1 Caffeine cache instance.
314	*
315	* <p>Time-based expiry operates at the <em>Caffeine</em> level via a custom
316	* {@link Expiry} implementation, computing remaining nanoseconds from
317	* . Entries with
318	* {@code hardExpireAtMs == Long.MAX_VALUE} are purely logical-expiry
319	* — Caffeine never evicts them by time; they live until size eviction or
320	* manual invalidation. Reads never extend the expiry duration (no read-based
321	* refresh), ensuring predictable TTL behavior.
322	*
323	* <p>Eviction strategy: {@code max-weight} (> 0) enables memory-weighted
324	* eviction with {@link DefaultWeigher}; otherwise {@code max-size} limits
325	* entry count. Time-based TTL for entries without an explicit hard-expire
326	* timestamp defaults to {@code hotkey.local.local-cache-ttl-minutes}.
327	*
328	* @param properties the HotKey configuration properties (never {@code null})
329	* @return a configured Caffeine {@link Cache} instance
330	*/
331	@Bean
332	@ConditionalOnMissingBean
333	public Cache<String, Object> hotLocalCache(HotKeyProperties properties) {
334	var cfg = properties.getCache();	3✔
335	Caffeine<Object, Object> builder = Caffeine.newBuilder();	2✔
336	if (cfg.getMaxWeight() > 0) {	5!
NEW 337	builder.maximumWeight(cfg.getMaxWeight()).weigher(DefaultWeigher.INSTANCE);	×
338	} else {
339	builder.maximumSize(cfg.getMaxSize());	6✔
340	}
341	builder.expireAfter(	8✔
342	new Expiry<>() {	9✔
343	/**
344	* Compute the time-to-live for a newly created cache entry.
345	* Returns {@link Long#MAX_VALUE} for pure logical-expiry entries
346	* (where {@code hardExpireAtMs == Long.MAX_VALUE}); otherwise
347	* computes the remaining wall-clock time.
348	*
349	* @param key the cache key
350	* @param value the cache value (expected to be a {@link CacheEntry})
351	* @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
352	* @return the expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
353	*/
354	@Override
355	public long expireAfterCreate(@NonNull Object key, @NonNull Object value, long currentTimeNanos) {
356	if (value instanceof CacheEntry entry) {	3!
NEW 357	if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {	×
358	// Pure logical expiry: Caffeine never time-evicts this entry.
359	// See Expiry Javadoc: Long.MAX_VALUE signals "no expiration".
NEW 360	return Long.MAX_VALUE;	×
361	}
NEW 362	long remainingMs = entry.getHardExpireAtMs() - currentTimeMillis();	×
NEW 363	return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));	×
364	}
365	return TimeUnit.MINUTES.toNanos(properties.getLocalCacheTtlMinutes());	7✔
366	}
367
368	/**
369	* Re-compute the expiry duration after an entry is updated.
370	* Preserves pure logical expiry across updates; otherwise
371	* recalculates from the entry's {@code hardExpireAtMs}.
372	*
373	* @param key the cache key
374	* @param value the cache value (expected to be a {@link CacheEntry})
375	* @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
376	* @param currentDuration the current expiry duration in nanoseconds
377	* @return the updated expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
378	*/
379	@Override
380	public long expireAfterUpdate(
381	@NonNull Object key,
382	@NonNull Object value,
383	long currentTimeNanos,
384	long currentDuration
385	) {
NEW 386	if (value instanceof CacheEntry entry) {	×
NEW 387	if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {	×
388	// Preserve pure logical expiry across updates (e.g. broadcast refresh).
NEW 389	return Long.MAX_VALUE;	×
390	}
NEW 391	long remainingMs = entry.getHardExpireAtMs() - currentTimeMillis();	×
NEW 392	return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));	×
393	}
NEW 394	return currentDuration;	×
395	}
396
397	/**
398	* Preserve the current expiry duration on read — reads never
399	* extend or shorten the entry's time-to-live.
400	*
401	* @param key the cache key
402	* @param value the cache value
403	* @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
404	* @param currentDuration the current expiry duration in nanoseconds
405	* @return the unchanged expiry duration in nanoseconds
406	*/
407	@Override
408	public long expireAfterRead(
409	@NonNull Object key,
410	@NonNull Object value,
411	long currentTimeNanos,
412	long currentDuration
413	) {
NEW 414	return currentDuration;	×
415	}
416	}
417	);
418	return builder.build();	3✔
419	}
420	}

Hyshmily / hotkey / 28835396755

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