27925388821

Committed 22 Jun 2026 02:14AM UTC coverage: 92.403% (-1.2%) from 93.615%

Build # 27925388821

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix:fix known bugs and promote performance

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

Coverage Stats

1143 of 1289 branches covered (88.67%)

Branch coverage included in aggregate %.

184 of 236 new or added lines in 19 files covered. (77.97%)

6 existing lines in 3 files now uncovered.

3321 of 3542 relevant lines covered (93.76%)

4.22 hits per line

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

90.57

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

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

import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
import io.github.hyshmily.hotkey.hotkeydetector.heavykepper.TopK;
import io.github.hyshmily.hotkey.model.HotKeyDecision;
import io.github.hyshmily.hotkey.reporting.ReportMessage;
import io.github.hyshmily.hotkey.worker.detection.GlobalQpsEstimator;
import io.github.hyshmily.hotkey.worker.detection.SlidingWindowDetector;
import io.github.hyshmily.hotkey.worker.detection.TopKValidator;
import io.github.hyshmily.hotkey.worker.dispatch.WorkerBroadcaster;
import java.util.Map;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;

/**
 * Worker‑side message consumer that receives batched per‑key access counts
 * reported by application instances.
 *
 * <p>For every key in the batch the consumer:
 * <ol>
 *   <li>Feeds the count into the {@link SlidingWindowDetector} to update the
 *       sliding‑window sum and obtain a binary hot‑or‑not verdict for the
 *       current window.</li>
 *   <li>Passes that verdict to the {@link HotKeyStateMachine} which tracks
 *       consecutive hot/cold windows and decides whether a state transition
 *       (COLD → CONFIRMED_HOT → PRE_COOLING → COLD) has occurred.</li>
 *   <li>If the state machine returns a {@code HOT} decision, the consumer
 *       broadcasts a {@code HOT} message to all application instances and
 *       records the confirmation in the {@link TopKValidator}.</li>
 *   <li>If the state machine returns a {@code COOL} decision, it broadcasts
 *       a {@code COOL} message and marks the key as cooled in the
 *       {@link TopKValidator}.</li>
 * </ol>
 *
 * <p>Messages older than 5 seconds are silently discarded to prevent stale
 * data from distorting the sliding‑window view.
 *
 * <p>Because clients use consistent‑hash routing, every report for a given
 * key always reaches the same worker, guaranteeing correct per‑key state
 * without cross‑worker coordination.
 */
@RequiredArgsConstructor
@Slf4j
public class ReportConsumer {

  /** Sliding-window detector that tracks per-key access counts and returns hot/cold verdicts. */
  private final SlidingWindowDetector detector;
  /** Per-key lifecycle state machine managing COLD / CONFIRMED_HOT / PRE_COOLING transitions. */
  private final HotKeyStateMachine stateMachine;
  /** Publishes HOT and COOL decisions back to all application instances. */
  private final WorkerBroadcaster broadcaster;
  /** TopK pre-warm validator for cross-instance frequency-based confirmation. */
  private final TopKValidator topKValidator;
  /** Worker-scoped HeavyKeeper sketch for cross-instance frequency estimation. */
  private final TopK workerTopK;
  /** Global QPS estimator tracking overall throughput for dynamic threshold learning. */
  private final GlobalQpsEstimator globalQpsEstimator;

  /** Staleness threshold in milliseconds. Package-visible for testing. */
  long stalenessThresholdMs = 5000L;

  /**
   * Main entry point for batched report messages.
   *
   * @param message the deserialized message containing counts for multiple keys
   */
  @RabbitListener(queues = "#{@reportQueue.name}")
  public void onReport(ReportMessage message) {
    try {
      doOnReport(message);
    } catch (Exception e) {
      log.error("Uncaught exception in onReport, discarding message to prevent poison-message requeue loop: appName={}",
          message != null ? message.appName() : "null", e);
    }
  }

  private void doOnReport(ReportMessage message) {
    long now = System.currentTimeMillis();
    long totalQps = 0;

    // Discard reports that are more than 5 seconds old.
    // This guards against delayed or re‑delivered messages that would
    // feed outdated counts into the sliding window.
    if (now - message.timestamp() > stalenessThresholdMs) {
      log.debug("Stale report message, skip: appName={}, age={}ms", message.appName(), now - message.timestamp());
      return;
    }

    // Process each key independently
    for (Map.Entry<String, Long> entry : message.counts().entrySet()) {
      try {
        String key = entry.getKey();
        long count = entry.getValue();

        totalQps += count;
        // Feed the global Worker TopK with the aggregated report count.
        // This populates the Worker-side HeavyKeeper so TopKValidator can
        // pre-warm keys based on cross-instance frequency.
        workerTopK.addDirect(key, (int) Math.min(count, Integer.MAX_VALUE));

        // addCount atomically increments the current time slice and returns
        // true if the sum of the last windowSize slices exceeds the threshold.
        boolean isHot = detector.addCount(key, count);

        Map<String, Object> stateSnapshot = stateMachine.getStateSnapshot(key);

        // The state machine tracks consecutive hot/cold windows and applies
        // hysteresis to decide when to transition between COLD, CONFIRMED_HOT
        // and PRE_COOLING.
        HotKeyDecision decision = stateMachine.evaluate(key, isHot);

        switch (decision.type()) {
          case HOT -> {
            try {
              // A new hot key has been confirmed.
              // Broadcast HOT to all app instances so they can pre‑warm
              // their local caches and enable soft expiration.
              broadcaster.broadcastHot(key, SOURCE_SLIDING_WINDOW);
              // Record the confirmation in TopK for historical ranking.
              topKValidator.markConfirmed(key);
            } catch (WorkerBroadcaster.BroadcastFailedException e) {
              log.warn("Broadcast HOT failed, rolling back state machine for key={}", key, e);
              stateMachine.rollbackToPreviousState(key, stateSnapshot);
            }
          }
          case COOL -> {
            try {
              // The key has fully cooled down.
              // Broadcast COOL so instances can disable soft expiration
              // and let the entry be evicted naturally.
              broadcaster.broadcastCool(key);
              // Update the TopK tracking accordingly.
              topKValidator.markCooled(key);
            } catch (WorkerBroadcaster.BroadcastFailedException e) {
              log.warn("Broadcast COOL failed, rolling back state machine for key={}", key, e);
              stateMachine.rollbackToPreviousState(key, stateSnapshot);
            }
          }
          case NONE -> {
            // No state transition occurred – the key remains in its
            // current lifecycle stage.  Nothing to do.
          }
        }
      } catch (Exception e) {
        log.error(
          "Error processing report entry: appName={}, key={}, count={}",
          message.appName(),
          entry.getKey(),
          entry.getValue(),
          e
        );
      }
    }
    globalQpsEstimator.addTotal(totalQps);
  }
}

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.ingest;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.SOURCE_SLIDING_WINDOW;
19
20	import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
21	import io.github.hyshmily.hotkey.hotkeydetector.heavykepper.TopK;
22	import io.github.hyshmily.hotkey.model.HotKeyDecision;
23	import io.github.hyshmily.hotkey.reporting.ReportMessage;
24	import io.github.hyshmily.hotkey.worker.detection.GlobalQpsEstimator;
25	import io.github.hyshmily.hotkey.worker.detection.SlidingWindowDetector;
26	import io.github.hyshmily.hotkey.worker.detection.TopKValidator;
27	import io.github.hyshmily.hotkey.worker.dispatch.WorkerBroadcaster;
28	import java.util.Map;
29	import lombok.RequiredArgsConstructor;
30	import lombok.extern.slf4j.Slf4j;
31	import org.springframework.amqp.rabbit.annotation.RabbitListener;
32
33	/**
34	* Worker‑side message consumer that receives batched per‑key access counts
35	* reported by application instances.
36	*
37	* <p>For every key in the batch the consumer:
38	* <ol>
39	* <li>Feeds the count into the {@link SlidingWindowDetector} to update the
40	* sliding‑window sum and obtain a binary hot‑or‑not verdict for the
41	* current window.</li>
42	* <li>Passes that verdict to the {@link HotKeyStateMachine} which tracks
43	* consecutive hot/cold windows and decides whether a state transition
44	* (COLD → CONFIRMED_HOT → PRE_COOLING → COLD) has occurred.</li>
45	* <li>If the state machine returns a {@code HOT} decision, the consumer
46	* broadcasts a {@code HOT} message to all application instances and
47	* records the confirmation in the {@link TopKValidator}.</li>
48	* <li>If the state machine returns a {@code COOL} decision, it broadcasts
49	* a {@code COOL} message and marks the key as cooled in the
50	* {@link TopKValidator}.</li>
51	* </ol>
52	*
53	* <p>Messages older than 5 seconds are silently discarded to prevent stale
54	* data from distorting the sliding‑window view.
55	*
56	* <p>Because clients use consistent‑hash routing, every report for a given
57	* key always reaches the same worker, guaranteeing correct per‑key state
58	* without cross‑worker coordination.
59	*/
60	@RequiredArgsConstructor
61	@Slf4j	4✔
62	public class ReportConsumer {
63
64	/** Sliding-window detector that tracks per-key access counts and returns hot/cold verdicts. */
65	private final SlidingWindowDetector detector;
66	/** Per-key lifecycle state machine managing COLD / CONFIRMED_HOT / PRE_COOLING transitions. */
67	private final HotKeyStateMachine stateMachine;
68	/** Publishes HOT and COOL decisions back to all application instances. */
69	private final WorkerBroadcaster broadcaster;
70	/** TopK pre-warm validator for cross-instance frequency-based confirmation. */
71	private final TopKValidator topKValidator;
72	/** Worker-scoped HeavyKeeper sketch for cross-instance frequency estimation. */
73	private final TopK workerTopK;
74	/** Global QPS estimator tracking overall throughput for dynamic threshold learning. */
75	private final GlobalQpsEstimator globalQpsEstimator;
76
77	/** Staleness threshold in milliseconds. Package-visible for testing. */
78	long stalenessThresholdMs = 5000L;
79
80	/**
81	* Main entry point for batched report messages.
82	*
83	* @param message the deserialized message containing counts for multiple keys
84	*/
85	@RabbitListener(queues = "#{@reportQueue.name}")
86	public void onReport(ReportMessage message) {
87	try {
88	doOnReport(message);	3✔
NEW 89	} catch (Exception e) {	×
NEW 90	log.error("Uncaught exception in onReport, discarding message to prevent poison-message requeue loop: appName={}",	×
NEW 91	message != null ? message.appName() : "null", e);	×
92	}	1✔
93	}	1✔
94
95	private void doOnReport(ReportMessage message) {
96	long now = System.currentTimeMillis();	2✔
97	long totalQps = 0;	2✔
98
99	// Discard reports that are more than 5 seconds old.
100	// This guards against delayed or re‑delivered messages that would
101	// feed outdated counts into the sliding window.
102	if (now - message.timestamp() > stalenessThresholdMs) {	8✔
103	log.debug("Stale report message, skip: appName={}, age={}ms", message.appName(), now - message.timestamp());	10✔
104	return;	1✔
105	}
106
107	// Process each key independently
108	for (Map.Entry<String, Long> entry : message.counts().entrySet()) {	12✔
109	try {
110	String key = entry.getKey();	4✔
111	long count = entry.getValue();	5✔
112
113	totalQps += count;	4✔
114	// Feed the global Worker TopK with the aggregated report count.
115	// This populates the Worker-side HeavyKeeper so TopKValidator can
116	// pre-warm keys based on cross-instance frequency.
117	workerTopK.addDirect(key, (int) Math.min(count, Integer.MAX_VALUE));	9✔
118
119	// addCount atomically increments the current time slice and returns
120	// true if the sum of the last windowSize slices exceeds the threshold.
121	boolean isHot = detector.addCount(key, count);	6✔
122
123	Map<String, Object> stateSnapshot = stateMachine.getStateSnapshot(key);	5✔
124
125	// The state machine tracks consecutive hot/cold windows and applies
126	// hysteresis to decide when to transition between COLD, CONFIRMED_HOT
127	// and PRE_COOLING.
128	HotKeyDecision decision = stateMachine.evaluate(key, isHot);	6✔
129
130	switch (decision.type()) {	6✔
131	case HOT -> {
132	try {
133	// A new hot key has been confirmed.
134	// Broadcast HOT to all app instances so they can pre‑warm
135	// their local caches and enable soft expiration.
136	broadcaster.broadcastHot(key, SOURCE_SLIDING_WINDOW);	5✔
137	// Record the confirmation in TopK for historical ranking.
138	topKValidator.markConfirmed(key);	4✔
139	} catch (WorkerBroadcaster.BroadcastFailedException e) {	1✔
140	log.warn("Broadcast HOT failed, rolling back state machine for key={}", key, e);	5✔
141	stateMachine.rollbackToPreviousState(key, stateSnapshot);	5✔
142	}	1✔
143	}	1✔
144	case COOL -> {
145	try {
146	// The key has fully cooled down.
147	// Broadcast COOL so instances can disable soft expiration
148	// and let the entry be evicted naturally.
149	broadcaster.broadcastCool(key);	4✔
150	// Update the TopK tracking accordingly.
151	topKValidator.markCooled(key);	4✔
152	} catch (WorkerBroadcaster.BroadcastFailedException e) {	1✔
153	log.warn("Broadcast COOL failed, rolling back state machine for key={}", key, e);	5✔
154	stateMachine.rollbackToPreviousState(key, stateSnapshot);	5✔
155	}	1✔
156	}	1✔
157	case NONE -> {
158	// No state transition occurred – the key remains in its
159	// current lifecycle stage. Nothing to do.
160	}
161	}
162	} catch (Exception e) {	1✔
163	log.error(	8✔
164	"Error processing report entry: appName={}, key={}, count={}",
165	message.appName(),	5✔
166	entry.getKey(),	5✔
167	entry.getValue(),	6✔
168	e
169	);
170	}	1✔
171	}	1✔
172	globalQpsEstimator.addTotal(totalQps);	4✔
173	}	1✔
174	}

Hyshmily / hotkey / 27925388821

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