28356107387

Committed 29 Jun 2026 07:35AM UTC coverage: 91.168% (-0.01%) from 91.178%

Build # 28356107387

Build Type

push

github

Committed by

Hyshmily

Commit Message

test : fix CI tests

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

Coverage Stats

1276 of 1461 branches covered (87.34%)

Branch coverage included in aggregate %.

10 of 10 new or added lines in 4 files covered. (100.0%)

117 existing lines in 15 files now uncovered.

3596 of 3883 relevant lines covered (92.61%)

4.24 hits per line

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

96.15

/worker/src/main/java/io/github/hyshmily/hotkey/worker/detection/SlidingWindowDetector.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.worker.detection;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import lombok.Getter;
import lombok.Setter;

/**
 * A high‑performance, lock‑free sliding‑window detector for real‑time hot‑key
 * identification.
 *
 * <h3>Why a sliding window?</h3>
 * A fixed‑window counter (e.g. reset every second) suffers from boundary
 * problems: a burst that straddles two windows can be split and go undetected.
 * The sliding window continuously aggregates the last {@code windowSize} time
 * slices, giving a smooth, instantaneous view of traffic for every key.
 *
 * <h3>Data structure</h3>
 * Each key owns an {@link AtomicLong} array of length {@code 2 * windowSize},
 * used as a <b>circular buffer</b>.  The doubling avoids expensive array copies
 * when the window slides: old slices are lazily overwritten after a full
 * rotation, and a dedicated cleaning step ({@link #clearStaleSlices}) zeros
 * them out before reuse.
 *
 * <h3>Concurrency</h3>
 * <ul>
 *   <li>Per‑key arrays are accessed only by the worker that owns the shard
 *       (thanks to consistent‑hash routing on the client side), so there is
 *       <b>no cross‑thread contention</b> on the same array.</li>
 *   <li>{@link AtomicLong} is used for individual slice counters to guarantee
 *       visibility and atomicity of updates, even though writes are
 *       single‑threaded — this protects against JVM reordering and ensures
 *       correct reads from the eviction thread.</li>
 *   <li>The {@code windows} and {@code lastAccessTime} maps use
 *       {@link ConcurrentHashMap} for safe concurrent access across keys and
 *       for the periodic eviction thread.</li>
 * </ul>
 *
 * <h3>Memory management</h3>
 * The {@link #evictStale(long)} method must be called periodically (e.g. every
 * few seconds) by a scheduler.  It removes any key that has not been accessed
 * within the given timeout, freeing the associated array and map entries.
 *
 * <h3>Dynamic threshold</h3>
 * The hot threshold is declared {@code volatile} and can be changed at runtime
 * via {@link #setThreshold(long)}.  The {@link ThresholdLearner} periodically
 * updates this value based on estimated global QPS.
 *
 * @see GlobalQpsEstimator
 * @see ThresholdLearner
 * @see WorkerAutoConfiguration.EvictStaleTask
 */
@Getter
@Setter
public class SlidingWindowDetector {

  /** Number of time slices that form one complete window. */
  private final int windowSize;

  /** Duration of a single time slice, in milliseconds. */
  private final long timeMillisPerSlice;

  /**
   * The access count a key must reach within the sliding window to be
   * considered "hot" for the current evaluation cycle.  Can be changed at
   * runtime (e.g. through JMX, a configuration refresh, or the
   * {@link ThresholdLearner}) because it is declared {@code volatile}.
   */
  private volatile long threshold;

  /**
   * Per‑key circular buffers.  Each value is an {@code AtomicLong[]} of
   * length {@code 2 * windowSize}.  The current slice index is derived from
   * {@code System.currentTimeMillis() / timeMillisPerSlice % length}.
   */
  private final ConcurrentHashMap<String, AtomicLong[]> windows = new ConcurrentHashMap<>();

  /**
   * Last access timestamp (epoch millis) for each key.  Used solely by
   * {@link #evictStale(long)} to identify and remove idle keys.
   */
  private final ConcurrentHashMap<String, Long> lastAccessTime = new ConcurrentHashMap<>();

  /**
   * Constructs a detector.
   *
   * @param windowDurationMs total duration of the sliding window in milliseconds
   * @param slices           number of time slices within the window (determines
   *                         the granularity of the sliding)
   * @param threshold        initial hot‑key threshold
   */
  public SlidingWindowDetector(long windowDurationMs, int slices, long threshold) {
    this.windowSize = slices;
    this.timeMillisPerSlice = windowDurationMs / slices;
    this.threshold = threshold;
  }

  /**
   * Records an access count for the given key and immediately evaluates
   * whether the key is "hot" in the current window.
   *
   * <p>If this is the first access for the key, a new circular buffer is
   * created atomically.  The current time slice is updated with the given
   * count, stale slices (older than one full window) are zeroed, and the
   * window sum is compared against the current threshold.
   *
   * @param key   the cache key; must not be {@code null}
   * @param count the number of accesses to record (typically the batched
   *              count reported by an application instance)
   * @return {@code true} if the sum of the last {@link #windowSize} slices
   *         meets or exceeds {@link #threshold}; {@code false} otherwise
   * @throws NullPointerException if {@code key} is {@code null}
   */
  public boolean addCount(String key, long count) {
    // Capture current time once to avoid redundant system calls.
    long now = System.currentTimeMillis();

    // Obtain or create the circular buffer for this key.
    AtomicLong[] slices = windows.get(key);
    if (slices == null) {
      slices = windows.computeIfAbsent(key, k -> new AtomicLong[windowSize * 2]);
    }

    // Compute the index of the current time slice.
    int currentIndex = (int) ((now / timeMillisPerSlice) % slices.length);

    // Update access timestamp immediately to protect against concurrent eviction.
    lastAccessTime.put(key, now);

    // Zero‑out slices that have fallen out of the window and are about to be overwritten.
    clearStaleSlices(slices, currentIndex);

    // Lazily initialize the slice counter if necessary.
    if (slices[currentIndex] == null) {
      slices[currentIndex] = new AtomicLong(0);
    }

    // Atomically addDirect the reported count to the current slice.
    slices[currentIndex].addAndGet(count);

    // Return the window‑level verdict.
    return getWindowSum(slices, currentIndex) >= threshold;
  }

  /**
   * Evicts stale tracking data for keys that have not been accessed within the
   * given timeout.  Unlike a simple two‑step {@code removeIf}, this implementation
   * uses atomic re‑verification to eliminate a race window between the two maps:
   * an {@code addCount} that updates the timestamp after the first pass will be
   * correctly detected and the key will be kept.
   *
   * <p>Must be called periodically (e.g. via
   * {@link WorkerAutoConfiguration.EvictStaleTask}) to prevent unbounded memory
   * growth from keys that are no longer accessed.
   *
   * @param staleAfterMs maximum idle time in milliseconds before a key is
   *                     considered stale and evicted; must be non-negative
   */
  public void evictStale(long staleAfterMs) {
    long now = System.currentTimeMillis();

    // This is a best‑effort snapshot; concurrent addCount calls may update
    // lastAccessTime after this collection, so we must re‑check later.
    List<String> candidates = new ArrayList<>();
    lastAccessTime.forEach((key, ts) -> {
      if (now - ts > staleAfterMs) {
        candidates.add(key);
      }
    });

    for (String key : candidates) {
      windows.compute(key, (k, arr) -> {
        // Re‑read the latest access time – this guards against the race
        // where addCount() updated the timestamp after we collected the key.
        Long currentTs = lastAccessTime.get(k);

        // If the timestamp has been refreshed (or the key already gone),
        // the entry is still active – keep the window array.
        if (currentTs == null || now - currentTs <= staleAfterMs) {
          return arr;
        }

        // The key is genuinely stale.  Conditionally remove the timestamp
        // entry ONLY if it still holds the same old value.  If addCount()
        // concurrently updated the timestamp, remove() will fail, and we
        // must retain the window array.  The conditional remove is atomic
        // and avoids introducing a lock or synchronized block.
        lastAccessTime.remove(k, currentTs);

        // Returning null deletes the window array from the windows map.
        return null;
      });
    }

    // This handles corner cases where a window array was removed but the
    // corresponding timestamp entry survived (e.g. due to a prior race that
    // has now been fixed).
    lastAccessTime.keySet().removeIf(k -> !windows.containsKey(k));
  }

  /**
   * Clears the slices that are logically "behind" the current window.
   *
   * <p>In the doubled circular buffer, indices {@code currentIndex + windowSize}
   * backward to {@code currentIndex + 1} (modulo array length) are older than
   * one full window.  They must be zeroed so that the next
   * {@link #getWindowSum} call does not include stale data.
   *
   * @param slices       the circular buffer for a specific key
   * @param currentIndex the index of the current time slice
   */
  private void clearStaleSlices(AtomicLong[] slices, int currentIndex) {
    int clearStart = (currentIndex + windowSize) % slices.length;
    for (int i = 0; i < windowSize; i++) {
      // Walk backwards to avoid clearing slices still within the current window.
      int idx = (clearStart - i + slices.length) % slices.length;
      if (slices[idx] != null) {
        slices[idx].set(0);
      }
    }
  }

  /**
   * Sums the last {@link #windowSize} slices in the circular buffer,
   * including the slice at {@code currentIndex}.
   *
   * <p>Walks backwards from {@code currentIndex} to include the most recent
   * {@code windowSize} time slices.  Null entries (uninitialised slices) are
   * treated as zero.
   *
   * @param slices       the circular buffer for a specific key
   * @param currentIndex the index of the current time slice
   * @return the sum of all slice values within the current sliding window
   */
  private long getWindowSum(AtomicLong[] slices, int currentIndex) {
    long sum = 0;
    for (int i = 0; i < windowSize; i++) {
      // Walk backwards from currentIndex to include the most recent windowSize slices.
      int idx = (currentIndex - i + slices.length) % slices.length;
      if (slices[idx] != null) {
        sum += slices[idx].get();
      }
    }
    return sum;
  }

  /**
   * Returns the total access count within the current sliding window for the
   * given key, or {@code 0} if the key is unknown.
   *
   * @param key the cache key to query
   * @return the sum of access counts in the current window, or {@code 0} if
   *         the key is not being tracked
   */
  public long getWindowSum(String key) {
    AtomicLong[] slices = windows.get(key);
    if (slices == null) {
      return 0;
    }
    int currentIndex = (int) ((System.currentTimeMillis() / timeMillisPerSlice) % slices.length);
    return getWindowSum(slices, currentIndex);
  }

  /**
   * Returns the number of keys currently being tracked by this detector.
   *
   * <p>This includes all keys with allocated circular buffers, regardless
   * of whether they have expired but have not yet been evicted.
   *
   * @return the count of active keys in the windows map
   */
  public int getActiveKeyCount() {
    return windows.size();
  }
}

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.worker.detection;
17
18	import java.util.ArrayList;
19	import java.util.List;
20	import java.util.concurrent.ConcurrentHashMap;
21	import java.util.concurrent.atomic.AtomicLong;
22	import lombok.Getter;
23	import lombok.Setter;
24
25	/**
26	* A high‑performance, lock‑free sliding‑window detector for real‑time hot‑key
27	* identification.
28	*
29	* <h3>Why a sliding window?</h3>
30	* A fixed‑window counter (e.g. reset every second) suffers from boundary
31	* problems: a burst that straddles two windows can be split and go undetected.
32	* The sliding window continuously aggregates the last {@code windowSize} time
33	* slices, giving a smooth, instantaneous view of traffic for every key.
34	*
35	* <h3>Data structure</h3>
36	* Each key owns an {@link AtomicLong} array of length {@code 2 * windowSize},
37	* used as a <b>circular buffer</b>. The doubling avoids expensive array copies
38	* when the window slides: old slices are lazily overwritten after a full
39	* rotation, and a dedicated cleaning step ({@link #clearStaleSlices}) zeros
40	* them out before reuse.
41	*
42	* <h3>Concurrency</h3>
43	* <ul>
44	* <li>Per‑key arrays are accessed only by the worker that owns the shard
45	* (thanks to consistent‑hash routing on the client side), so there is
46	* <b>no cross‑thread contention</b> on the same array.</li>
47	* <li>{@link AtomicLong} is used for individual slice counters to guarantee
48	* visibility and atomicity of updates, even though writes are
49	* single‑threaded — this protects against JVM reordering and ensures
50	* correct reads from the eviction thread.</li>
51	* <li>The {@code windows} and {@code lastAccessTime} maps use
52	* {@link ConcurrentHashMap} for safe concurrent access across keys and
53	* for the periodic eviction thread.</li>
54	* </ul>
55	*
56	* <h3>Memory management</h3>
57	* The {@link #evictStale(long)} method must be called periodically (e.g. every
58	* few seconds) by a scheduler. It removes any key that has not been accessed
59	* within the given timeout, freeing the associated array and map entries.
60	*
61	* <h3>Dynamic threshold</h3>
62	* The hot threshold is declared {@code volatile} and can be changed at runtime
63	* via {@link #setThreshold(long)}. The {@link ThresholdLearner} periodically
64	* updates this value based on estimated global QPS.
65	*
66	* @see GlobalQpsEstimator
67	* @see ThresholdLearner
68	* @see WorkerAutoConfiguration.EvictStaleTask
69	*/
70	@Getter
71	@Setter
72	public class SlidingWindowDetector {
73
74	/** Number of time slices that form one complete window. */
75	private final int windowSize;
76
77	/** Duration of a single time slice, in milliseconds. */
78	private final long timeMillisPerSlice;
79
80	/**
81	* The access count a key must reach within the sliding window to be
82	* considered "hot" for the current evaluation cycle. Can be changed at
83	* runtime (e.g. through JMX, a configuration refresh, or the
84	* {@link ThresholdLearner}) because it is declared {@code volatile}.
85	*/
86	private volatile long threshold;
87
88	/**
89	* Per‑key circular buffers. Each value is an {@code AtomicLong[]} of
90	* length {@code 2 * windowSize}. The current slice index is derived from
91	* {@code System.currentTimeMillis() / timeMillisPerSlice % length}.
92	*/
93	private final ConcurrentHashMap<String, AtomicLong[]> windows = new ConcurrentHashMap<>();	5✔
94
95	/**
96	* Last access timestamp (epoch millis) for each key. Used solely by
97	* {@link #evictStale(long)} to identify and remove idle keys.
98	*/
99	private final ConcurrentHashMap<String, Long> lastAccessTime = new ConcurrentHashMap<>();	5✔
100
101	/**
102	* Constructs a detector.
103	*
104	* @param windowDurationMs total duration of the sliding window in milliseconds
105	* @param slices number of time slices within the window (determines
106	* the granularity of the sliding)
107	* @param threshold initial hot‑key threshold
108	*/
109	public SlidingWindowDetector(long windowDurationMs, int slices, long threshold) {	2✔
110	this.windowSize = slices;	3✔
111	this.timeMillisPerSlice = windowDurationMs / slices;	6✔
112	this.threshold = threshold;	3✔
113	}	1✔
114
115	/**
116	* Records an access count for the given key and immediately evaluates
117	* whether the key is "hot" in the current window.
118	*
119	* <p>If this is the first access for the key, a new circular buffer is
120	* created atomically. The current time slice is updated with the given
121	* count, stale slices (older than one full window) are zeroed, and the
122	* window sum is compared against the current threshold.
123	*
124	* @param key the cache key; must not be {@code null}
125	* @param count the number of accesses to record (typically the batched
126	* count reported by an application instance)
127	* @return {@code true} if the sum of the last {@link #windowSize} slices
128	* meets or exceeds {@link #threshold}; {@code false} otherwise
129	* @throws NullPointerException if {@code key} is {@code null}
130	*/
131	public boolean addCount(String key, long count) {
132	// Capture current time once to avoid redundant system calls.
133	long now = System.currentTimeMillis();	2✔
134
135	// Obtain or create the circular buffer for this key.
136	AtomicLong[] slices = windows.get(key);	6✔
137	if (slices == null) {	2✔
138	slices = windows.computeIfAbsent(key, k -> new AtomicLong[windowSize * 2]);	14✔
139	}
140
141	// Compute the index of the current time slice.
142	int currentIndex = (int) ((now / timeMillisPerSlice) % slices.length);	10✔
143
144	// Update access timestamp immediately to protect against concurrent eviction.
145	lastAccessTime.put(key, now);	7✔
146
147	// Zero‑out slices that have fallen out of the window and are about to be overwritten.
148	clearStaleSlices(slices, currentIndex);	4✔
149
150	// Lazily initialize the slice counter if necessary.
151	if (slices[currentIndex] == null) {	4✔
152	slices[currentIndex] = new AtomicLong(0);	7✔
153	}
154
155	// Atomically addDirect the reported count to the current slice.
156	slices[currentIndex].addAndGet(count);	6✔
157
158	// Return the window‑level verdict.
159	return getWindowSum(slices, currentIndex) >= threshold;	12✔
160	}
161
162	/**
163	* Evicts stale tracking data for keys that have not been accessed within the
164	* given timeout. Unlike a simple two‑step {@code removeIf}, this implementation
165	* uses atomic re‑verification to eliminate a race window between the two maps:
166	* an {@code addCount} that updates the timestamp after the first pass will be
167	* correctly detected and the key will be kept.
168	*
169	* <p>Must be called periodically (e.g. via
170	* {@link WorkerAutoConfiguration.EvictStaleTask}) to prevent unbounded memory
171	* growth from keys that are no longer accessed.
172	*
173	* @param staleAfterMs maximum idle time in milliseconds before a key is
174	* considered stale and evicted; must be non-negative
175	*/
176	public void evictStale(long staleAfterMs) {
177	long now = System.currentTimeMillis();	2✔
178
179	// This is a best‑effort snapshot; concurrent addCount calls may update
180	// lastAccessTime after this collection, so we must re‑check later.
181	List<String> candidates = new ArrayList<>();	4✔
182	lastAccessTime.forEach((key, ts) -> {	7✔
183	if (now - ts > staleAfterMs) {	7✔
184	candidates.add(key);	4✔
185	}
186	});	1✔
187
188	for (String key : candidates) {	10✔
189	windows.compute(key, (k, arr) -> {	9✔
190	// Re‑read the latest access time – this guards against the race
191	// where addCount() updated the timestamp after we collected the key.
192	Long currentTs = lastAccessTime.get(k);	6✔
193
194	// If the timestamp has been refreshed (or the key already gone),
195	// the entry is still active – keep the window array.
196	if (currentTs == null \|\| now - currentTs <= staleAfterMs) {	9!
UNCOV 197	return arr;	×
198	}
199
200	// The key is genuinely stale. Conditionally remove the timestamp
201	// entry ONLY if it still holds the same old value. If addCount()
202	// concurrently updated the timestamp, remove() will fail, and we
203	// must retain the window array. The conditional remove is atomic
204	// and avoids introducing a lock or synchronized block.
205	lastAccessTime.remove(k, currentTs);	6✔
206
207	// Returning null deletes the window array from the windows map.
208	return null;	2✔
209	});
210	}	1✔
211
212	// This handles corner cases where a window array was removed but the
213	// corresponding timestamp entry survived (e.g. due to a prior race that
214	// has now been fixed).
215	lastAccessTime.keySet().removeIf(k -> !windows.containsKey(k));	16✔
216	}	1✔
217
218	/**
219	* Clears the slices that are logically "behind" the current window.
220	*
221	* <p>In the doubled circular buffer, indices {@code currentIndex + windowSize}
222	* backward to {@code currentIndex + 1} (modulo array length) are older than
223	* one full window. They must be zeroed so that the next
224	* {@link #getWindowSum} call does not include stale data.
225	*
226	* @param slices the circular buffer for a specific key
227	* @param currentIndex the index of the current time slice
228	*/
229	private void clearStaleSlices(AtomicLong[] slices, int currentIndex) {
230	int clearStart = (currentIndex + windowSize) % slices.length;	8✔
231	for (int i = 0; i < windowSize; i++) {	8✔
232	// Walk backwards to avoid clearing slices still within the current window.
233	int idx = (clearStart - i + slices.length) % slices.length;	10✔
234	if (slices[idx] != null) {	4✔
235	slices[idx].set(0);	5✔
236	}
237	}
238	}	1✔
239
240	/**
241	* Sums the last {@link #windowSize} slices in the circular buffer,
242	* including the slice at {@code currentIndex}.
243	*
244	* <p>Walks backwards from {@code currentIndex} to include the most recent
245	* {@code windowSize} time slices. Null entries (uninitialised slices) are
246	* treated as zero.
247	*
248	* @param slices the circular buffer for a specific key
249	* @param currentIndex the index of the current time slice
250	* @return the sum of all slice values within the current sliding window
251	*/
252	private long getWindowSum(AtomicLong[] slices, int currentIndex) {
253	long sum = 0;	2✔
254	for (int i = 0; i < windowSize; i++) {	8✔
255	// Walk backwards from currentIndex to include the most recent windowSize slices.
256	int idx = (currentIndex - i + slices.length) % slices.length;	10✔
257	if (slices[idx] != null) {	4✔
258	sum += slices[idx].get();	7✔
259	}
260	}
261	return sum;	2✔
262	}
263
264	/**
265	* Returns the total access count within the current sliding window for the
266	* given key, or {@code 0} if the key is unknown.
267	*
268	* @param key the cache key to query
269	* @return the sum of access counts in the current window, or {@code 0} if
270	* the key is not being tracked
271	*/
272	public long getWindowSum(String key) {
273	AtomicLong[] slices = windows.get(key);	6✔
274	if (slices == null) {	2✔
275	return 0;	2✔
276	}
277	int currentIndex = (int) ((System.currentTimeMillis() / timeMillisPerSlice) % slices.length);	10✔
278	return getWindowSum(slices, currentIndex);	5✔
279	}
280
281	/**
282	* Returns the number of keys currently being tracked by this detector.
283	*
284	* <p>This includes all keys with allocated circular buffers, regardless
285	* of whether they have expired but have not yet been evicted.
286	*
287	* @return the count of active keys in the windows map
288	*/
289	public int getActiveKeyCount() {
290	return windows.size();	4✔
291	}
292	}

Hyshmily / hotkey / 28356107387

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