28503055923

Committed 01 Jul 2026 08:06AM UTC coverage: 90.445% (-0.4%) from 90.887%

Build # 28503055923

Build Type

push

github

Committed by

Hyshmily

Commit Message

perf : promote throughput performance

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

Coverage Stats

1248 of 1441 branches covered (86.61%)

Branch coverage included in aggregate %.

123 of 139 new or added lines in 10 files covered. (88.49%)

2 existing lines in 2 files now uncovered.

3532 of 3844 relevant lines covered (91.88%)

4.2 hits per line

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

93.48

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

import static io.github.hyshmily.hotkey.detection.HotKeyStateMachine.State.*;

import com.google.common.util.concurrent.Striped;
import io.github.hyshmily.hotkey.model.HotKeyDecision;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * Per-key state machine that governs hot-key lifecycle transitions on the
 * Worker side.
 *
 * <p>Each Worker shard owns a subset of keys (determined by
 * {@link io.github.hyshmily.hotkey.sharding.ConsistentHashRing} routing)
 * and runs one {@code HotKeyStateMachine} instance per owned shard. The
 * state machine converts per-key sliding-window frequency observations
 * into lifecycle transitions — promoting COLD keys to CONFIRMED_HOT when
 * sustained traffic is detected, and demoting them back to COLD after a
 * prolonged cool-down.
 *
 * <h3>State diagram</h3>
 * <pre>
 *   COLD
 *    │  hotStreak >= confirmCount  (consecutive hot windows)
 *    ▼
 *   CONFIRMED_HOT   ─────────────────────────────┐
 *    │                                           │
 *    │  coolStreak >= (coolCount - grace)        │ hotStreak > 0 (revive during pre-cool)
 *    ▼                                           │
 *   PRE_COOLING ─────────────────────────────────┘
 *    │
 *    │  coolStreak >= coolCount  (fully cooled)
 *    ▼
 *   COLD
 * </pre>
 *
 * <h3>Design rationale</h3>
 * <ul>
 *   <li><b>Fast detection:</b> only {@code confirmCount} consecutive hot windows are
 *       needed to promote a key to {@code CONFIRMED_HOT}.</li>
 *   <li><b>Slow cool-down:</b> cooling requires many more consecutive cold windows,
 *       with an intermediate {@code PRE_COOLING} grace period.  If traffic resumes
 *       during this period the key silently returns to {@code CONFIRMED_HOT} without
 *       broadcasting (no COOL → HOT oscillation).</li>
 *   <li><b>Asymmetric thresholds:</b> protect aggressively, recover cautiously.
 *       This prevents flapping when traffic is bursty or when the cluster briefly
 *       stops reporting a key.</li>
 * </ul>
 *
 * <p>Thread-safe: per-key state is guarded by a {@link Striped} lock (1024
 * stripes). Evaluations of different keys proceed in parallel; evaluations
 * of the same key are serialized, eliminating the race window between
 * {@code hotStreak++} and the state transition check caused by concurrent
 * delivery of the same key across multiple consumer threads.
 *
 * <p>Designed for single-shard workers; each key is owned by exactly one worker
 * thanks to consistent-hash routing on the client side.</p>
 */
@Slf4j
@AllArgsConstructor
public class HotKeyStateMachine {

  /**
   * Key-level state: current lifecycle stage within the hot-key state machine.
   *
   * <p>Transitions are governed by consecutive window observations
   * (see {@link HotKeyStateMachine#evaluate}).
   */
  public enum State {
    /**
     * Not currently tracked as hot. This is the initial state for all keys.
     * The key must be observed above the hot threshold for
     * {@code confirmCount} consecutive windows to transition to
     * {@link #CONFIRMED_HOT}.
     */
    COLD,
    /**
     * Confirmed hot — a HOT broadcast has been sent to the application
     * instances. The key's L1 TTL should be extended. If traffic subsides,
     * the key transitions to {@link #PRE_COOLING} after a grace period.
     */
    CONFIRMED_HOT,
    /**
     * Transitional stage between HOT and COLD.
     *
     * <p>No broadcast is sent when entering this stage. If the key becomes
     * hot again (observed above threshold during this window) it returns
     * to {@link #CONFIRMED_HOT} silently, avoiding unnecessary COOL→HOT
     * oscillations. If it remains cold for {@code coolCount} consecutive
     * windows, it transitions to {@link #COLD} and a COOL broadcast is
     * emitted.
     */
    PRE_COOLING,
  }

  /** Number of consecutive hot windows required to promote COLD → CONFIRMED_HOT. */
  @Getter
  @Setter
  private volatile int confirmCount;

  /**
   * Total number of consecutive cold windows required for a full cool-down
   * (CONFIRMED_HOT → PRE_COOLING → COLD).  Must be greater than
   * {@code preCoolGraceCount}.
   */
  @Getter
  @Setter
  private volatile int coolCount;

  /**
   * The number of cold windows that mark the entry into PRE_COOLING.
   * The remaining {@code coolCount - preCoolGraceCount} windows are the
   * grace period during which the key can revive without a broadcast.
   */
  @Getter
  @Setter
  private volatile int preCoolGraceCount;

  /** Current state + streak counters, keyed by cache key. */
  private final ConcurrentHashMap<String, KeyState> states = new ConcurrentHashMap<>();

  /**
   * Last evaluation timestamp for each key.  Used by {@link #evictStale(long)}
   * to garbage-collect keys that have not been reported for an extended period.
   */
  private final ConcurrentHashMap<String, Long> stateTimestamps = new ConcurrentHashMap<>();

  /**
   * Per-key striped lock — serializes evaluations of the same key when
   * multiple consumer threads process overlapping messages, preventing
   * lost increments on {@code hotStreak++} / {@code coolStreak++}.
   *
   * <p>1024 stripes keep collision probability below 0.1% at
   * {@code concurrency=8} while adding negligible memory overhead.
   */
  private final Striped<Lock> keyLocks = Striped.lazyWeakLock(1024);

  /**
   * Evaluates the current sliding-window observation for the given key and
   * returns a decision instructing the caller whether to broadcast a HOT or
   * COOL message.
   *
   * <p>This method implements the state transitions described in the class
   * Javadoc. It updates the per-key streak counters ({@code hotStreak} /
   * {@code coolStreak}) atomically (via {@link ConcurrentHashMap#computeIfAbsent})
   * and returns one of:
   * <ul>
   *   <li>{@link HotKeyDecision.DecisionType#HOT} — key just crossed the
   *       promotion threshold; broadcast HOT to application instances.</li>
   *   <li>{@link HotKeyDecision.DecisionType#COOL} — key has fully cooled
   *       down; broadcast COOL so apps revert to normal TTL.</li>
   *   <li>{@link HotKeyDecision.DecisionType#NONE} — no state transition
   *       occurred; no action required.</li>
   * </ul>
   *
   * <p>Silent revive (PRE_COOLING → CONFIRMED_HOT) returns NONE to
   * suppress unnecessary broadcasts and prevent HOT/COOL oscillation.
   *
   * @param key             the cache key (must not be {@code null})
   * @param isHotThisWindow {@code true} if the sliding-window frequency sum
   *                        exceeds the hot threshold during this evaluation
   *                        cycle; {@code false} otherwise
   * @return a non-null {@link HotKeyDecision} indicating what action the
   *         caller should take (HOT, COOL, or NONE)
   */
  public HotKeyDecision evaluate(String key, boolean isHotThisWindow) {
    // Fast path: never-before-seen key on a cold window — no state to
    // mutate (no accumulated hotStreak to reset), skip striped-lock
    // acquisition entirely.
    if (!isHotThisWindow && !states.containsKey(key)) {
      stateTimestamps.put(key, System.currentTimeMillis());
      return HotKeyDecision.none(key);
    }

    keyLocks.get(key).lock();
    try {
      // Touch timestamp for eviction tracking
      stateTimestamps.put(key, System.currentTimeMillis());

      KeyState state = states.computeIfAbsent(key, k -> new KeyState());

      if (isHotThisWindow) {
        state.hotStreak++;
        state.coolStreak = 0; // reset cold streak

        // COLD → CONFIRMED_HOT transition
        if (state.currentState == COLD && state.hotStreak >= confirmCount) {
          state.currentState = CONFIRMED_HOT;
          return HotKeyDecision.hot(key); // broadcast HOT
        }

        // PRE_COOLING → CONFIRMED_HOT (silent revive)
        if (state.currentState == PRE_COOLING) {
          state.currentState = CONFIRMED_HOT;
          return HotKeyDecision.none(key); // no broadcast — avoid oscillation
        }
      } else {
        state.coolStreak++;
        state.hotStreak = 0; // reset hot streak

        // CONFIRMED_HOT → PRE_COOLING transition (first cool phase)
        if (state.currentState == CONFIRMED_HOT && state.coolStreak >= coolCount - preCoolGraceCount) {
          state.currentState = PRE_COOLING;
          // no broadcast yet — wait for full cool-down
        }

        // PRE_COOLING → COLD transition (fully cooled)
        if (state.currentState == PRE_COOLING && state.coolStreak >= coolCount) {
          state.currentState = COLD;
          return HotKeyDecision.cool(key); // broadcast COOL
        }
      }

      return HotKeyDecision.none(key);
    } catch (Exception e) {
      log.warn("Unexpected StateMachine Exception for key {}", key, e);
      return HotKeyDecision.none(key);
    } finally {
      keyLocks.get(key).unlock();
    }
  }

  /**
   * Immediately removes all tracked state for the given key, effectively
   * resetting it to {@link State#COLD}.
   *
   * <p>Called when the Worker fails to obtain a version from Redis and must
   * abort the current HOT decision (e.g. Redis is unreachable or the key
   * was not found in Redis). After reset, the next {@link #evaluate} call
   * for this key will start from scratch with fresh streak counters.
   *
   * <p>The last-touch timestamp is intentionally not removed to prevent
   * immediate re-creation churn in {@link #evictStale} — it will be cleaned
   * up lazily on the next eviction cycle.
   *
   * @param key the cache key to reset
   */
  public void reset(String key) {
    states.remove(key);
    // stateTimestamps is cleaned up lazily by evictStale();
    // keeping the timestamp for a while avoids immediate re-creation churn.
  }

  /**
   * Approximate number of keys currently tracked by the state machine.
   * <p>The returned value is approximate due to the underlying
   * {@link java.util.concurrent.ConcurrentHashMap#size()} semantics
   * — it reflects a snapshot and may not account for concurrent
   * insertions or removals at the exact moment of the call.</p>
   *
   * @return approximate count of keys currently tracked
   */
  public int getTrackedKeys() {
    return states.size();
  }

  /**
   * Garbage-collects state for keys that have not been evaluated within
   * {@code staleAfterMs} milliseconds.  Should be invoked periodically
   * (e.g. every 5 seconds) from a scheduled task.
   *
   * @param staleAfterMs maximum idle time in milliseconds before a key is evicted
   */
  public void evictStale(long staleAfterMs) {
    long now = System.currentTimeMillis();
    // Remove state entries whose last touch is older than the threshold
    states
      .keySet()
      .removeIf(key -> {
        Long last = stateTimestamps.get(key);
        return last != null && now - last > staleAfterMs;
      });
    // Purge orphaned timestamps
    stateTimestamps.keySet().removeIf(k -> !states.containsKey(k));
  }

  public Map<String, Object> getStateSnapshot(String key) {
    KeyState keyState = states.get(key);
    if (keyState == null) {
      return Collections.emptyMap();
    }
    return Map.of(
      "currentState",
      keyState.currentState.name(),
      "hotStreak",
      keyState.hotStreak,
      "coolStreak",
      keyState.coolStreak
    );
  }

  /**
   * Rolls back the per-key state to its previous value after a broadcast failure.
   * This allows the next evaluation window to re-emit the decision.
   *
   * @param key the key whose state machine should be rolled back
   */
  public void rollbackToPreviousState(String key, Map<String, Object> previousState) {
    if (previousState == null) {
      // No previous state; treat as reset
      reset(key);
      return;
    }

    KeyState keyState = states.computeIfAbsent(key, k -> new KeyState());
    keyState.currentState = State.valueOf((String) previousState.get("currentState"));
    keyState.hotStreak = (int) previousState.get("hotStreak");
    keyState.coolStreak = (int) previousState.get("coolStreak");
  }

  private static class KeyState {

    /** Current lifecycle stage. */
    volatile State currentState = COLD;

    /** Number of consecutive windows above the hot threshold. */
    volatile int hotStreak = 0;

    /** Number of consecutive windows below the hot threshold. */
    volatile int coolStreak = 0;
  }
}

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.detection;
17
18	import static io.github.hyshmily.hotkey.detection.HotKeyStateMachine.State.*;
19
20	import com.google.common.util.concurrent.Striped;
21	import io.github.hyshmily.hotkey.model.HotKeyDecision;
22	import java.util.Collections;
23	import java.util.Map;
24	import java.util.concurrent.ConcurrentHashMap;
25	import java.util.concurrent.locks.Lock;
26	import lombok.AllArgsConstructor;
27	import lombok.Getter;
28	import lombok.Setter;
29	import lombok.extern.slf4j.Slf4j;
30
31	/**
32	* Per-key state machine that governs hot-key lifecycle transitions on the
33	* Worker side.
34	*
35	* <p>Each Worker shard owns a subset of keys (determined by
36	* {@link io.github.hyshmily.hotkey.sharding.ConsistentHashRing} routing)
37	* and runs one {@code HotKeyStateMachine} instance per owned shard. The
38	* state machine converts per-key sliding-window frequency observations
39	* into lifecycle transitions — promoting COLD keys to CONFIRMED_HOT when
40	* sustained traffic is detected, and demoting them back to COLD after a
41	* prolonged cool-down.
42	*
43	* <h3>State diagram</h3>
44	* <pre>
45	* COLD
46	* │ hotStreak >= confirmCount (consecutive hot windows)
47	* ▼
48	* CONFIRMED_HOT ─────────────────────────────┐
49	* │ │
50	* │ coolStreak >= (coolCount - grace) │ hotStreak > 0 (revive during pre-cool)
51	* ▼ │
52	* PRE_COOLING ─────────────────────────────────┘
53	* │
54	* │ coolStreak >= coolCount (fully cooled)
55	* ▼
56	* COLD
57	* </pre>
58	*
59	* <h3>Design rationale</h3>
60	* <ul>
61	* <li><b>Fast detection:</b> only {@code confirmCount} consecutive hot windows are
62	* needed to promote a key to {@code CONFIRMED_HOT}.</li>
63	* <li><b>Slow cool-down:</b> cooling requires many more consecutive cold windows,
64	* with an intermediate {@code PRE_COOLING} grace period. If traffic resumes
65	* during this period the key silently returns to {@code CONFIRMED_HOT} without
66	* broadcasting (no COOL → HOT oscillation).</li>
67	* <li><b>Asymmetric thresholds:</b> protect aggressively, recover cautiously.
68	* This prevents flapping when traffic is bursty or when the cluster briefly
69	* stops reporting a key.</li>
70	* </ul>
71	*
72	* <p>Thread-safe: per-key state is guarded by a {@link Striped} lock (1024
73	* stripes). Evaluations of different keys proceed in parallel; evaluations
74	* of the same key are serialized, eliminating the race window between
75	* {@code hotStreak++} and the state transition check caused by concurrent
76	* delivery of the same key across multiple consumer threads.
77	*
78	* <p>Designed for single-shard workers; each key is owned by exactly one worker
79	* thanks to consistent-hash routing on the client side.</p>
80	*/
81	@Slf4j	4✔
82	@AllArgsConstructor
83	public class HotKeyStateMachine {
84
85	/**
86	* Key-level state: current lifecycle stage within the hot-key state machine.
87	*
88	* <p>Transitions are governed by consecutive window observations
89	* (see {@link HotKeyStateMachine#evaluate}).
90	*/
91	public enum State {	3✔
92	/**
93	* Not currently tracked as hot. This is the initial state for all keys.
94	* The key must be observed above the hot threshold for
95	* {@code confirmCount} consecutive windows to transition to
96	* {@link #CONFIRMED_HOT}.
97	*/
98	COLD,	6✔
99	/**
100	* Confirmed hot — a HOT broadcast has been sent to the application
101	* instances. The key's L1 TTL should be extended. If traffic subsides,
102	* the key transitions to {@link #PRE_COOLING} after a grace period.
103	*/
104	CONFIRMED_HOT,	6✔
105	/**
106	* Transitional stage between HOT and COLD.
107	*
108	* <p>No broadcast is sent when entering this stage. If the key becomes
109	* hot again (observed above threshold during this window) it returns
110	* to {@link #CONFIRMED_HOT} silently, avoiding unnecessary COOL→HOT
111	* oscillations. If it remains cold for {@code coolCount} consecutive
112	* windows, it transitions to {@link #COLD} and a COOL broadcast is
113	* emitted.
114	*/
115	PRE_COOLING,	6✔
116	}
117
118	/** Number of consecutive hot windows required to promote COLD → CONFIRMED_HOT. */
119	@Getter
120	@Setter
121	private volatile int confirmCount;
122
123	/**
124	* Total number of consecutive cold windows required for a full cool-down
125	* (CONFIRMED_HOT → PRE_COOLING → COLD). Must be greater than
126	* {@code preCoolGraceCount}.
127	*/
128	@Getter
129	@Setter
130	private volatile int coolCount;
131
132	/**
133	* The number of cold windows that mark the entry into PRE_COOLING.
134	* The remaining {@code coolCount - preCoolGraceCount} windows are the
135	* grace period during which the key can revive without a broadcast.
136	*/
137	@Getter
138	@Setter
139	private volatile int preCoolGraceCount;
140
141	/** Current state + streak counters, keyed by cache key. */
142	private final ConcurrentHashMap<String, KeyState> states = new ConcurrentHashMap<>();
143
144	/**
145	* Last evaluation timestamp for each key. Used by {@link #evictStale(long)}
146	* to garbage-collect keys that have not been reported for an extended period.
147	*/
148	private final ConcurrentHashMap<String, Long> stateTimestamps = new ConcurrentHashMap<>();
149
150	/**
151	* Per-key striped lock — serializes evaluations of the same key when
152	* multiple consumer threads process overlapping messages, preventing
153	* lost increments on {@code hotStreak++} / {@code coolStreak++}.
154	*
155	* <p>1024 stripes keep collision probability below 0.1% at
156	* {@code concurrency=8} while adding negligible memory overhead.
157	*/
158	private final Striped<Lock> keyLocks = Striped.lazyWeakLock(1024);
159
160	/**
161	* Evaluates the current sliding-window observation for the given key and
162	* returns a decision instructing the caller whether to broadcast a HOT or
163	* COOL message.
164	*
165	* <p>This method implements the state transitions described in the class
166	* Javadoc. It updates the per-key streak counters ({@code hotStreak} /
167	* {@code coolStreak}) atomically (via {@link ConcurrentHashMap#computeIfAbsent})
168	* and returns one of:
169	* <ul>
170	* <li>{@link HotKeyDecision.DecisionType#HOT} — key just crossed the
171	* promotion threshold; broadcast HOT to application instances.</li>
172	* <li>{@link HotKeyDecision.DecisionType#COOL} — key has fully cooled
173	* down; broadcast COOL so apps revert to normal TTL.</li>
174	* <li>{@link HotKeyDecision.DecisionType#NONE} — no state transition
175	* occurred; no action required.</li>
176	* </ul>
177	*
178	* <p>Silent revive (PRE_COOLING → CONFIRMED_HOT) returns NONE to
179	* suppress unnecessary broadcasts and prevent HOT/COOL oscillation.
180	*
181	* @param key the cache key (must not be {@code null})
182	* @param isHotThisWindow {@code true} if the sliding-window frequency sum
183	* exceeds the hot threshold during this evaluation
184	* cycle; {@code false} otherwise
185	* @return a non-null {@link HotKeyDecision} indicating what action the
186	* caller should take (HOT, COOL, or NONE)
187	*/
188	public HotKeyDecision evaluate(String key, boolean isHotThisWindow) {
189	// Fast path: never-before-seen key on a cold window — no state to
190	// mutate (no accumulated hotStreak to reset), skip striped-lock
191	// acquisition entirely.
192	if (!isHotThisWindow && !states.containsKey(key)) {	7✔
193	stateTimestamps.put(key, System.currentTimeMillis());	7✔
194	return HotKeyDecision.none(key);	3✔
195	}
196
197	keyLocks.get(key).lock();	6✔
198	try {
199	// Touch timestamp for eviction tracking
200	stateTimestamps.put(key, System.currentTimeMillis());	7✔
201
202	KeyState state = states.computeIfAbsent(key, k -> new KeyState());	11✔
203
204	if (isHotThisWindow) {	2✔
205	state.hotStreak++;	6✔
206	state.coolStreak = 0; // reset cold streak	3✔
207
208	// COLD → CONFIRMED_HOT transition
209	if (state.currentState == COLD && state.hotStreak >= confirmCount) {	9✔
210	state.currentState = CONFIRMED_HOT;	3✔
211	return HotKeyDecision.hot(key); // broadcast HOT	5✔
212	}
213
214	// PRE_COOLING → CONFIRMED_HOT (silent revive)
215	if (state.currentState == PRE_COOLING) {	4✔
216	state.currentState = CONFIRMED_HOT;	3✔
217	return HotKeyDecision.none(key); // no broadcast — avoid oscillation	5✔
218	}
219	} else {
220	state.coolStreak++;	6✔
221	state.hotStreak = 0; // reset hot streak	3✔
222
223	// CONFIRMED_HOT → PRE_COOLING transition (first cool phase)
224	if (state.currentState == CONFIRMED_HOT && state.coolStreak >= coolCount - preCoolGraceCount) {	12✔
225	state.currentState = PRE_COOLING;	3✔
226	// no broadcast yet — wait for full cool-down
227	}
228
229	// PRE_COOLING → COLD transition (fully cooled)
230	if (state.currentState == PRE_COOLING && state.coolStreak >= coolCount) {	9✔
231	state.currentState = COLD;	3✔
232	return HotKeyDecision.cool(key); // broadcast COOL	5✔
233	}
234	}
235
236	return HotKeyDecision.none(key);	5✔
NEW 237	} catch (Exception e) {	×
NEW 238	log.warn("Unexpected StateMachine Exception for key {}", key, e);	×
UNCOV 239	return HotKeyDecision.none(key);	×
240	} finally {
241	keyLocks.get(key).unlock();	6✔
242	}
243	}
244
245	/**
246	* Immediately removes all tracked state for the given key, effectively
247	* resetting it to {@link State#COLD}.
248	*
249	* <p>Called when the Worker fails to obtain a version from Redis and must
250	* abort the current HOT decision (e.g. Redis is unreachable or the key
251	* was not found in Redis). After reset, the next {@link #evaluate} call
252	* for this key will start from scratch with fresh streak counters.
253	*
254	* <p>The last-touch timestamp is intentionally not removed to prevent
255	* immediate re-creation churn in {@link #evictStale} — it will be cleaned
256	* up lazily on the next eviction cycle.
257	*
258	* @param key the cache key to reset
259	*/
260	public void reset(String key) {
261	states.remove(key);	5✔
262	// stateTimestamps is cleaned up lazily by evictStale();
263	// keeping the timestamp for a while avoids immediate re-creation churn.
264	}	1✔
265
266	/**
267	* Approximate number of keys currently tracked by the state machine.
268	* <p>The returned value is approximate due to the underlying
269	* {@link java.util.concurrent.ConcurrentHashMap#size()} semantics
270	* — it reflects a snapshot and may not account for concurrent
271	* insertions or removals at the exact moment of the call.</p>
272	*
273	* @return approximate count of keys currently tracked
274	*/
275	public int getTrackedKeys() {
276	return states.size();	4✔
277	}
278
279	/**
280	* Garbage-collects state for keys that have not been evaluated within
281	* {@code staleAfterMs} milliseconds. Should be invoked periodically
282	* (e.g. every 5 seconds) from a scheduled task.
283	*
284	* @param staleAfterMs maximum idle time in milliseconds before a key is evicted
285	*/
286	public void evictStale(long staleAfterMs) {
287	long now = System.currentTimeMillis();	2✔
288	// Remove state entries whose last touch is older than the threshold
289	states	2✔
290	.keySet()	5✔
291	.removeIf(key -> {	2✔
292	Long last = stateTimestamps.get(key);	6✔
293	return last != null && now - last > staleAfterMs;	12!
294	});
295	// Purge orphaned timestamps
296	stateTimestamps.keySet().removeIf(k -> !states.containsKey(k));	15!
297	}	1✔
298
299	public Map<String, Object> getStateSnapshot(String key) {
300	KeyState keyState = states.get(key);	6✔
301	if (keyState == null) {	2✔
302	return Collections.emptyMap();	2✔
303	}
304	return Map.of(	5✔
305	"currentState",
306	keyState.currentState.name(),	4✔
307	"hotStreak",
308	keyState.hotStreak,	4✔
309	"coolStreak",
310	keyState.coolStreak	1✔
311	);
312	}
313
314	/**
315	* Rolls back the per-key state to its previous value after a broadcast failure.
316	* This allows the next evaluation window to re-emit the decision.
317	*
318	* @param key the key whose state machine should be rolled back
319	*/
320	public void rollbackToPreviousState(String key, Map<String, Object> previousState) {
321	if (previousState == null) {	2✔
322	// No previous state; treat as reset
323	reset(key);	3✔
324	return;	1✔
325	}
326
327	KeyState keyState = states.computeIfAbsent(key, k -> new KeyState());	11✔
328	keyState.currentState = State.valueOf((String) previousState.get("currentState"));	7✔
329	keyState.hotStreak = (int) previousState.get("hotStreak");	7✔
330	keyState.coolStreak = (int) previousState.get("coolStreak");	7✔
331	}	1✔
332
333	private static class KeyState {	2✔
334
335	/** Current lifecycle stage. */
336	volatile State currentState = COLD;	3✔
337
338	/** Number of consecutive windows above the hot threshold. */
339	volatile int hotStreak = 0;	3✔
340
341	/** Number of consecutive windows below the hot threshold. */
342	volatile int coolStreak = 0;	4✔
343	}
344	}

Hyshmily / hotkey / 28503055923

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