28290463292

Committed 27 Jun 2026 01:22PM UTC coverage: 91.227% (-0.7%) from 91.882%

Build # 28290463292

Build Type

push

github

Committed by

Hyshmily

Commit Message

test: fix CI tests

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

Coverage Stats

1250 of 1427 branches covered (87.6%)

Branch coverage included in aggregate %.

3544 of 3828 relevant lines covered (92.58%)

4.21 hits per line

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

90.91

/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.heavykeeper.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 java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.LongAdder;
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}", containerFactory = "reportListenerContainerFactory")
  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();
    LongAdder totalQps = new LongAdder();

    // Discard reports that are more than 5 seconds old.
    // 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;
    }

    Map<String, Long> keyCounts = message.counts();

    // Feed all key counts into the Worker's HeavyKeeper in a single
    // batch call.  This executes sketch updates per-key under fine-grained
    // stripe locks, then acquires the global sortedTopK heap
    // lock exactly ONCE for the entire batch — rather than once per
    // key — eliminating the P1-3 lock-contention bottleneck.
    workerTopK.addDirect(keyCounts);

    // Accumulate broadcasts during parallel processing; drain serially
    // after the stream completes to avoid ForkJoin threads blocking on
    // AMQP channel write locks.
    Queue<Runnable> pendingBroadcasts = new ConcurrentLinkedQueue<>();

    // Process each key independently
    keyCounts
      .entrySet()
      .parallelStream()
      .forEach(entry -> {
        try {
          String key = entry.getKey();
          long count = entry.getValue();

          totalQps.add(count);

          // 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);

          // 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 -> {
              // A new hot key has been confirmed. Pre-allocate a decision
              // version and enqueue the broadcast; actual AMQP send happens
              // on the consumer thread after parallelStream completes.
              long dv = broadcaster.nextDecisionVersion();
              pendingBroadcasts.add(
                  () -> broadcaster.broadcastHot(key, SOURCE_SLIDING_WINDOW, dv));
              topKValidator.markConfirmed(key);
            }
            case COOL -> {
              long dv = broadcaster.nextDecisionVersion();
              pendingBroadcasts.add(
                  () -> broadcaster.broadcastCool(key, dv));
              topKValidator.markCooled(key);
            }
            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
          );
        }
      });

    // Drain pending broadcasts serially on the consumer thread.
    // This avoids ForkJoinPool threads blocking on AMQP channel write
    // locks under high concurrency (8 concurrent consumers).
    // Per ADR-0007, lost messages are tolerated by the next periodic cycle.
    // sendBroadcast no longer throws — errors are logged and swallowed.
    Runnable task;
    while ((task = pendingBroadcasts.poll()) != null) {
      task.run();
    }

    globalQpsEstimator.addTotal(totalQps.sum());
  }
}

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.heavykeeper.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 java.util.Queue;
30	import java.util.concurrent.ConcurrentLinkedQueue;
31	import java.util.concurrent.atomic.LongAdder;
32	import lombok.RequiredArgsConstructor;
33	import lombok.extern.slf4j.Slf4j;
34	import org.springframework.amqp.rabbit.annotation.RabbitListener;
35
36	/**
37	* Worker‑side message consumer that receives batched per‑key access counts
38	* reported by application instances.
39	*
40	* <p>For every key in the batch the consumer:
41	* <ol>
42	* <li>Feeds the count into the {@link SlidingWindowDetector} to update the
43	* sliding‑window sum and obtain a binary hot‑or‑not verdict for the
44	* current window.</li>
45	* <li>Passes that verdict to the {@link HotKeyStateMachine} which tracks
46	* consecutive hot/cold windows and decides whether a state transition
47	* (COLD → CONFIRMED_HOT → PRE_COOLING → COLD) has occurred.</li>
48	* <li>If the state machine returns a {@code HOT} decision, the consumer
49	* broadcasts a {@code HOT} message to all application instances and
50	* records the confirmation in the {@link TopKValidator}.</li>
51	* <li>If the state machine returns a {@code COOL} decision, it broadcasts
52	* a {@code COOL} message and marks the key as cooled in the
53	* {@link TopKValidator}.</li>
54	* </ol>
55	*
56	* <p>Messages older than 5 seconds are silently discarded to prevent stale
57	* data from distorting the sliding‑window view.
58	*
59	* <p>Because clients use consistent‑hash routing, every report for a given
60	* key always reaches the same worker, guaranteeing correct per‑key state
61	* without cross‑worker coordination.
62	*/
63	@RequiredArgsConstructor
64	@Slf4j	4✔
65	public class ReportConsumer {
66
67	/** Sliding-window detector that tracks per-key access counts and returns hot/cold verdicts. */
68	private final SlidingWindowDetector detector;
69	/** Per-key lifecycle state machine managing COLD / CONFIRMED_HOT / PRE_COOLING transitions. */
70	private final HotKeyStateMachine stateMachine;
71	/** Publishes HOT and COOL decisions back to all application instances. */
72	private final WorkerBroadcaster broadcaster;
73	/** TopK pre-warm validator for cross-instance frequency-based confirmation. */
74	private final TopKValidator topKValidator;
75	/** Worker-scoped HeavyKeeper sketch for cross-instance frequency estimation. */
76	private final TopK workerTopK;
77	/** Global QPS estimator tracking overall throughput for dynamic threshold learning. */
78	private final GlobalQpsEstimator globalQpsEstimator;
79
80	/** Staleness threshold in milliseconds. Package-visible for testing. */
81	long stalenessThresholdMs = 5000L;
82
83	/**
84	* Main entry point for batched report messages.
85	*
86	* @param message the deserialized message containing counts for multiple keys
87	*/
88	@RabbitListener(queues = "#{@reportQueue.name}", containerFactory = "reportListenerContainerFactory")
89	public void onReport(ReportMessage message) {
90	try {
91	doOnReport(message);	3✔
92	} catch (Exception e) {	×
93	log.error(	×
94	"Uncaught exception in onReport, discarding message to prevent poison-message requeue loop: appName={}",
95	message != null ? message.appName() : "null",	×
96	e
97	);
98	}	1✔
99	}	1✔
100
101	private void doOnReport(ReportMessage message) {
102	long now = System.currentTimeMillis();	2✔
103	LongAdder totalQps = new LongAdder();	4✔
104
105	// Discard reports that are more than 5 seconds old.
106	// Guards against delayed or re‑delivered messages that would
107	// feed outdated counts into the sliding window.
108	if (now - message.timestamp() > stalenessThresholdMs) {	8✔
109	log.debug("Stale report message, skip: appName={}, age={}ms", message.appName(), now - message.timestamp());	10✔
110	return;	1✔
111	}
112
113	Map<String, Long> keyCounts = message.counts();	3✔
114
115	// Feed all key counts into the Worker's HeavyKeeper in a single
116	// batch call. This executes sketch updates per-key under fine-grained
117	// stripe locks, then acquires the global sortedTopK heap
118	// lock exactly ONCE for the entire batch — rather than once per
119	// key — eliminating the P1-3 lock-contention bottleneck.
120	workerTopK.addDirect(keyCounts);	5✔
121
122	// Accumulate broadcasts during parallel processing; drain serially
123	// after the stream completes to avoid ForkJoin threads blocking on
124	// AMQP channel write locks.
125	Queue<Runnable> pendingBroadcasts = new ConcurrentLinkedQueue<>();	4✔
126
127	// Process each key independently
128	keyCounts	1✔
129	.entrySet()	1✔
130	.parallelStream()	6✔
131	.forEach(entry -> {	1✔
132	try {
133	String key = entry.getKey();	4✔
134	long count = entry.getValue();	5✔
135
136	totalQps.add(count);	3✔
137
138	// addCount atomically increments the current time slice and returns
139	// true if the sum of the last windowSize slices exceeds the threshold.
140	boolean isHot = detector.addCount(key, count);	6✔
141
142	// The state machine tracks consecutive hot/cold windows and applies
143	// hysteresis to decide when to transition between COLD, CONFIRMED_HOT
144	// and PRE_COOLING.
145	HotKeyDecision decision = stateMachine.evaluate(key, isHot);	6✔
146
147	switch (decision.type()) {	6✔
148	case HOT -> {
149	// A new hot key has been confirmed. Pre-allocate a decision
150	// version and enqueue the broadcast; actual AMQP send happens
151	// on the consumer thread after parallelStream completes.
152	long dv = broadcaster.nextDecisionVersion();	4✔
153	pendingBroadcasts.add(	7✔
154	() -> broadcaster.broadcastHot(key, SOURCE_SLIDING_WINDOW, dv));	7✔
155	topKValidator.markConfirmed(key);	4✔
156	}	1✔
157	case COOL -> {
158	long dv = broadcaster.nextDecisionVersion();	4✔
159	pendingBroadcasts.add(	7✔
160	() -> broadcaster.broadcastCool(key, dv));	6✔
161	topKValidator.markCooled(key);	4✔
162	}	1✔
163	case NONE -> {
164	// No state transition occurred – the key remains in its
165	// current lifecycle stage. Nothing to do.
166	}
167	}
168	} catch (Exception e) {	1✔
169	log.error(	8✔
170	"Error processing report entry: appName={}, key={}, count={}",
171	message.appName(),	5✔
172	entry.getKey(),	5✔
173	entry.getValue(),	6✔
174	e
175	);
176	}	1✔
177	});	1✔
178
179	// Drain pending broadcasts serially on the consumer thread.
180	// This avoids ForkJoinPool threads blocking on AMQP channel write
181	// locks under high concurrency (8 concurrent consumers).
182	// Per ADR-0007, lost messages are tolerated by the next periodic cycle.
183	// sendBroadcast no longer throws — errors are logged and swallowed.
184	Runnable task;
185	while ((task = pendingBroadcasts.poll()) != null) {	6✔
186	task.run();	3✔
187	}
188
189	globalQpsEstimator.addTotal(totalQps.sum());	5✔
190	}	1✔
191	}

Hyshmily / hotkey / 28290463292

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