28738321635

Committed 05 Jul 2026 10:51AM UTC coverage: 90.457% (+0.04%) from 90.421%

Build # 28738321635

Build Type

push

github

Committed by

Hyshmily

Commit Message

refactor: extract interfaces for 12 core classes (Batch 1+2)

Extract interfaces from concrete classes, impls moved to impl/ subpackages:

Batch 1 (HotPath):
- CircuitBreaker, ExpireManager, SingleFlight
- KeyReporter, RuleMatcher, HealthView

Batch 2 (Supporting):
- VersionController, RingManager, HotKeyStateMachine
- SystemLoadMonitor, BbrRateLimiter, SreRateLimiter

Pattern: interface at original package, impl named XxxImpl under impl/

Coverage Stats

1296 of 1498 branches covered (86.52%)

Branch coverage included in aggregate %.

448 of 480 new or added lines in 19 files covered. (93.33%)

3690 of 4014 relevant lines covered (91.93%)

4.19 hits per line

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

93.88

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

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import io.github.hyshmily.hotkey.Internal;
import io.github.hyshmily.hotkey.cache.cachesupport.CircuitBreaker;
import io.github.hyshmily.hotkey.cache.cachesupport.SingleFlight;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;


/**
 * Deduplicates concurrent in-flight loads for the same key.
 * <p>
 * Only the first caller executes the supplier; subsequent callers wait for
 * the same {@link CompletableFuture}. On normal completion, the future
 * remains cached (TTL-based expiry via {@code expireAfterWrite}) so
 * late-arriving callers reuse the result without re-execution. On timeout or
 * exception, the entry is evicted immediately to allow a subsequent retry.
 * <p>
 * The internal dedup cache is bounded by {@code maxSize} (LRU eviction) and
 * entries expire after the configured {@code ttlSec} seconds from write.
 * This class is thread-safe.
 */
@Slf4j
@Internal
public class SingleFlightImpl implements SingleFlight {

  /** Caffeine cache tracking currently in-flight loads (key -> CompletableFuture). */
  private final Cache<String, CompletableFuture<Object>> inflightLoads;
  /** Async executor for running the supplier. */
  private final Executor executor;
  /** Timeout in seconds before a supplier future is completed exceptionally. */
  private final int timeoutSeconds;
  /** Maximum number of in-flight keys tracked simultaneously. */
  private final int inflightMaxSize;

  /** Circuit breaker for protecting remote calls from cascading failures. */
  private final CircuitBreaker circuitBreaker;

  /**
   * Creates a SingleFlightImpl deduplicator that prevents concurrent in-flight loads
   * for the same key.
   *
   * @param maxSize        maximum number of concurrent in-flight keys tracked
   * @param ttlSec         time-to-live for dedup entries after write
   * @param timeoutSeconds per-supplier timeout before the future is completed exceptionally
   * @param executor       async executor for supplier execution
   * @param circuitBreaker circuit breaker for protecting remote calls
   */
  public SingleFlightImpl(int maxSize, int ttlSec, int timeoutSeconds, Executor executor, CircuitBreaker circuitBreaker) {
    this.inflightLoads = Caffeine.newBuilder().maximumSize(maxSize).expireAfterWrite(ttlSec, TimeUnit.SECONDS).build();
    this.executor = executor;
    this.timeoutSeconds = timeoutSeconds;
    this.inflightMaxSize = maxSize;
    this.circuitBreaker = circuitBreaker;
  }

  /**
   * Whether the circuit breaker is currently open.
   * Used by {@code HotKeyCache} to decide whether to return stale cache on miss.
   *
   * @return {@code true} if the breaker is open
   */
  public boolean isBreakerOpen() {
    return circuitBreaker.isOpen();
  }

  /**
   * Approximate number of keys currently tracked for dedup.
   * Useful for monitoring and diagnostics.
   *
   * @return the estimated number of in-flight keys
   */
  public long estimatedInflightSize() {
    return inflightLoads.estimatedSize();
  }

  /**
   * Load a value via the supplier, deduplicating concurrent requests for the same key.
   * Thread-safe: concurrent calls for the same key share a single future.
   *
   * @param cacheKey the key to load
   * @param reader   the value supplier (should not return {@code null})
   * @param <T>      the value type
   * @return the loaded value, or empty if the load failed or timed out
   */
  @SuppressWarnings("unchecked")
  public <T> Optional<T> load(String cacheKey, Supplier<T> reader) {
    if (!circuitBreaker.allowRequest()) {
      log.debug("CB open, skip load for key={}", cacheKey);
      return Optional.empty();
    }

    if (estimatedInflightSize() > inflightMaxSize * 0.8) {
      log.warn("SingleFlight inflight queue is high: {}/{}", estimatedInflightSize(), inflightMaxSize);
    }

    CompletableFuture<Object> future = inflightLoads
      .asMap()
      .computeIfAbsent(cacheKey, k ->
        CompletableFuture.supplyAsync(
          () -> {
            try {
              Object val = reader.get();
              circuitBreaker.onSuccess();
              return val;
            } catch (Exception | Error e) {
              circuitBreaker.onFailure();
              throw e;
            }
          },
          executor
        ).orTimeout(timeoutSeconds, TimeUnit.SECONDS)
      );

    try {
      return Optional.ofNullable((T) future.join());
    } catch (CompletionException e) {
      log.warn("singleflight join failed: key={}", cacheKey, e);
      inflightLoads.invalidate(cacheKey);

      Throwable cause = e.getCause();

      if (cause instanceof InterruptedException) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("Thread interrupted during load for key: " + cacheKey, cause);
      }

      if (cause instanceof RuntimeException re) {
        throw re;
      }

      if (cause instanceof Error err) {
        throw err;
      }

      throw new RuntimeException("Loader failed for key: " + cacheKey, cause);
    }
  }
}

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.cachesupport.impl;
17
18	import com.github.benmanes.caffeine.cache.Cache;
19	import com.github.benmanes.caffeine.cache.Caffeine;
20	import io.github.hyshmily.hotkey.Internal;
21	import io.github.hyshmily.hotkey.cache.cachesupport.CircuitBreaker;
22	import io.github.hyshmily.hotkey.cache.cachesupport.SingleFlight;
23	import java.util.Optional;
24	import java.util.concurrent.CompletableFuture;
25	import java.util.concurrent.CompletionException;
26	import java.util.concurrent.Executor;
27	import java.util.concurrent.TimeUnit;
28	import java.util.function.Supplier;
29	import lombok.extern.slf4j.Slf4j;
30
31
32	/**
33	* Deduplicates concurrent in-flight loads for the same key.
34	* <p>
35	* Only the first caller executes the supplier; subsequent callers wait for
36	* the same {@link CompletableFuture}. On normal completion, the future
37	* remains cached (TTL-based expiry via {@code expireAfterWrite}) so
38	* late-arriving callers reuse the result without re-execution. On timeout or
39	* exception, the entry is evicted immediately to allow a subsequent retry.
40	* <p>
41	* The internal dedup cache is bounded by {@code maxSize} (LRU eviction) and
42	* entries expire after the configured {@code ttlSec} seconds from write.
43	* This class is thread-safe.
44	*/
45	@Slf4j	4✔
46	@Internal
47	public class SingleFlightImpl implements SingleFlight {
48
49	/** Caffeine cache tracking currently in-flight loads (key -> CompletableFuture). */
50	private final Cache<String, CompletableFuture<Object>> inflightLoads;
51	/** Async executor for running the supplier. */
52	private final Executor executor;
53	/** Timeout in seconds before a supplier future is completed exceptionally. */
54	private final int timeoutSeconds;
55	/** Maximum number of in-flight keys tracked simultaneously. */
56	private final int inflightMaxSize;
57
58	/** Circuit breaker for protecting remote calls from cascading failures. */
59	private final CircuitBreaker circuitBreaker;
60
61	/**
62	* Creates a SingleFlightImpl deduplicator that prevents concurrent in-flight loads
63	* for the same key.
64	*
65	* @param maxSize maximum number of concurrent in-flight keys tracked
66	* @param ttlSec time-to-live for dedup entries after write
67	* @param timeoutSeconds per-supplier timeout before the future is completed exceptionally
68	* @param executor async executor for supplier execution
69	* @param circuitBreaker circuit breaker for protecting remote calls
70	*/
71	public SingleFlightImpl(int maxSize, int ttlSec, int timeoutSeconds, Executor executor, CircuitBreaker circuitBreaker) {	2✔
72	this.inflightLoads = Caffeine.newBuilder().maximumSize(maxSize).expireAfterWrite(ttlSec, TimeUnit.SECONDS).build();	11✔
73	this.executor = executor;	3✔
74	this.timeoutSeconds = timeoutSeconds;	3✔
75	this.inflightMaxSize = maxSize;	3✔
76	this.circuitBreaker = circuitBreaker;	3✔
77	}	1✔
78
79	/**
80	* Whether the circuit breaker is currently open.
81	* Used by {@code HotKeyCache} to decide whether to return stale cache on miss.
82	*
83	* @return {@code true} if the breaker is open
84	*/
85	public boolean isBreakerOpen() {
86	return circuitBreaker.isOpen();	4✔
87	}
88
89	/**
90	* Approximate number of keys currently tracked for dedup.
91	* Useful for monitoring and diagnostics.
92	*
93	* @return the estimated number of in-flight keys
94	*/
95	public long estimatedInflightSize() {
96	return inflightLoads.estimatedSize();	4✔
97	}
98
99	/**
100	* Load a value via the supplier, deduplicating concurrent requests for the same key.
101	* Thread-safe: concurrent calls for the same key share a single future.
102	*
103	* @param cacheKey the key to load
104	* @param reader the value supplier (should not return {@code null})
105	* @param <T> the value type
106	* @return the loaded value, or empty if the load failed or timed out
107	*/
108	@SuppressWarnings("unchecked")
109	public <T> Optional<T> load(String cacheKey, Supplier<T> reader) {
110	if (!circuitBreaker.allowRequest()) {	4✔
111	log.debug("CB open, skip load for key={}", cacheKey);	4✔
112	return Optional.empty();	2✔
113	}
114
115	if (estimatedInflightSize() > inflightMaxSize * 0.8) {	10✔
116	log.warn("SingleFlight inflight queue is high: {}/{}", estimatedInflightSize(), inflightMaxSize);	9✔
117	}
118
119	CompletableFuture<Object> future = inflightLoads	2✔
120	.asMap()	5✔
121	.computeIfAbsent(cacheKey, k ->	3✔
122	CompletableFuture.supplyAsync(	11✔
123	() -> {
124	try {
125	Object val = reader.get();	3✔
126	circuitBreaker.onSuccess();	3✔
127	return val;	2✔
128	} catch (Exception \| Error e) {	1✔
129	circuitBreaker.onFailure();	3✔
130	throw e;	2✔
131	}
132	},
133	executor
134	).orTimeout(timeoutSeconds, TimeUnit.SECONDS)	1✔
135	);
136
137	try {
138	return Optional.ofNullable((T) future.join());	4✔
139	} catch (CompletionException e) {	1✔
140	log.warn("singleflight join failed: key={}", cacheKey, e);	5✔
141	inflightLoads.invalidate(cacheKey);	4✔
142
143	Throwable cause = e.getCause();	3✔
144
145	if (cause instanceof InterruptedException) {	3!
NEW 146	Thread.currentThread().interrupt();	×
NEW 147	throw new RuntimeException("Thread interrupted during load for key: " + cacheKey, cause);	×
148	}
149
150	if (cause instanceof RuntimeException re) {	6✔
151	throw re;	2✔
152	}
153
154	if (cause instanceof Error err) {	6✔
155	throw err;	2✔
156	}
157
158	throw new RuntimeException("Loader failed for key: " + cacheKey, cause);	7✔
159	}
160	}
161	}

Hyshmily / hotkey / 28738321635

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