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

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 static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;

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

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