28916241081

Committed 08 Jul 2026 03:53AM UTC coverage: 89.468% (-0.7%) from 90.12%

Build # 28916241081

Build Type

push

github

Committed by

Hyshmily

Commit Message

refactor(core): extract central dispatcher and broadcast buffer

Introduce CentralDispatcher for centralized event routing and BroadcastBuffer for cache synchronization. Refactor HotKeyCache, SlidingWindowDetector, and related components to use the new dispatching mechanism, improving separation of concerns and reducing coupling.

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

Coverage Stats

1328 of 1564 branches covered (84.91%)

Branch coverage included in aggregate %.

102 of 134 new or added lines in 17 files covered. (76.12%)

3 existing lines in 2 files now uncovered.

3769 of 4133 relevant lines covered (91.19%)

4.21 hits per line

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

71.88

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

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

import io.github.hyshmily.hotkey.sync.worker.WorkerMessage;
import io.github.hyshmily.hotkey.util.version.VersionGuard;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicLong;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/*
 * Publishes HOT and COOL decisions to all application instances via the
 * configured RabbitMQ {@code broadcastExchange}.
 *
 * <p>Each broadcast message carries a monotonically increasing
 * <b>decision version</b> (worker‑local counter) that is used by receivers to
 * discard stale or out‑of‑order messages.
 *
 * <p>Messages are delivered to every instance's dedicated queue through a
 * fanout exchange ({@code hotkey.broadcast.exchange}) — the receiver
 * differentiates message type via the {@code AMQP_HEADER_TYPE} header.
 */
/** Default constructor. */
@RequiredArgsConstructor
@Slf4j
public class WorkerBroadcaster {

  /** RabbitMQ template used to publish HOT/COOL decisions and heartbeats. */
  private final RabbitTemplate rabbitTemplate;
  /** Target exchange name for all broadcast messages (typically {@code hotkey.broadcast.exchange}). */
  private final String broadcastExchange;
  /** Application name used in the broadcast routing key. */
  private final String appName;

  /** Originating Worker's node identity, set on every broadcast header for cross-Worker
   * version tracking in {@link VersionGuard}. */
  private final String nodeId;

  /** Monotonically increasing epoch counter for this Worker instance, incremented on every
   * restart. Transmitted in broadcast headers so receivers detect Worker restarts and
   * unconditionally accept decisions from a higher epoch (see ADR-0010). */
  private final AtomicLong epochCounter;

  /**
   * Worker‑local decision version counter.
   * Incremented atomically for every broadcast to provide strict ordering
   * for this worker.  Combined with consistent‑hash sharding (one key → one
   * worker) this guarantees a total order of decisions per key.
   */
  private final AtomicLong decisionVersionCounter = new AtomicLong(System.currentTimeMillis());

  /**
   * Atomically allocates the next decision version without sending.
   * Called from within {@link io.github.hyshmily.hotkey.worker.ingest.ReportConsumer#doOnReport}
   * to pre-allocate a version before enqueuing an async broadcast.
   *
   * @return the next monotonically increasing decision version
   */
  public long nextDecisionVersion() {
    return decisionVersionCounter.incrementAndGet();
  }

  /**
   * Broadcasts a HOT decision for the given key.
   *
   * @param cacheKey the key that has been confirmed as hot
   * @param source   a label describing the detection source (e.g. "sliding_window")
   */
  public void broadcastHot(String cacheKey, String source) {
    long dv = decisionVersionCounter.incrementAndGet();
    sendBroadcast(cacheKey, WorkerMessage.TYPE_HOT, dv);
    log.debug(
      "Broadcast HOT: key={}, dv={}, source={}, nodeId={}, epoch={}",
      cacheKey,
      dv,
      source,
      nodeId,
      epochCounter.get()
    );
  }

  /**
   * Broadcasts a HOT decision with a pre-allocated version number.
   *
   * @param cacheKey the key that has been confirmed as hot
   * @param source   a label describing the detection source
   * @param dv       pre-allocated decision version (from {@link #nextDecisionVersion()})
   */
  public void broadcastHot(String cacheKey, String source, long dv) {
    sendBroadcast(cacheKey, WorkerMessage.TYPE_HOT, dv);
    log.debug(
      "Broadcast HOT: key={}, dv={}, source={}, nodeId={}, epoch={}",
      cacheKey,
      dv,
      source,
      nodeId,
      epochCounter.get()
    );
  }

  /**
   * Broadcasts a COOL decision for the given key.
   *
   * @param cacheKey the key that has been confirmed as fully cooled
   */
  public void broadcastCool(String cacheKey) {
    long dv = decisionVersionCounter.incrementAndGet();
    sendBroadcast(cacheKey, WorkerMessage.TYPE_COOL, dv);
  }

  /**
   * Broadcasts a COOL decision with a pre-allocated version number.
   *
   * @param cacheKey the key that has been confirmed as fully cooled
   * @param dv       pre-allocated decision version (from {@link #nextDecisionVersion()})
   */
  public void broadcastCool(String cacheKey, long dv) {
    sendBroadcast(cacheKey, WorkerMessage.TYPE_COOL, dv);
    log.debug("Broadcast COOL: key={}, dv={}, nodeId={}, epoch={}", cacheKey, dv, nodeId, epochCounter.get());
  }

  /**
   * Returns the current decision version without incrementing.
   *
   * <p>Used by {@link WorkerHeartbeatProducer} for the heartbeat's decisionVersionHwm field.
   *
   * @return the current decision version counter value
   */
  public long getCurrentDecisionVersion() {
    return decisionVersionCounter.get();
  }

  /**
   * Common send helper for HOT/COOL decisions.
   *
   * <p>Builds {@link MessageProperties} with type, version, and degraded flag,
   * creates a {@link Message}, and publishes via {@link #rabbitTemplate}.
   *
   * @param cacheKey the key being broadcast
   * @param type     the message type ({@link WorkerMessage#TYPE_HOT} or {@link WorkerMessage#TYPE_COOL})
   * @param version  the monotonically increasing decision version for ordering
   */
  private void sendBroadcast(String cacheKey, String type, long version) {
    try {
      MessageProperties props = new MessageProperties();
      props.setHeader(AMQP_HEADER_TYPE, type);
      props.setHeader(AMQP_HEADER_VERSION, version);
      props.setHeader(AMQP_HEADER_IS_VERSION_DEGRADED, false);
      props.setHeader(AMQP_HEADER_NODE_ID, nodeId);
      props.setHeader(AMQP_HEADER_EPOCH, epochCounter.get());

      Message msg = new Message(cacheKey.getBytes(StandardCharsets.UTF_8), props);
      rabbitTemplate.send(broadcastExchange, ROUTING_KEY_BROADCAST + appName, msg);
    } catch (Exception e) {
      log.error("Failed to broadcast {} for key={}, dv={}: {}", type, cacheKey, version, e.getMessage());
      // no throw — ADR-0007 fire-and-forget
    }
  }
}

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.dispatch;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
19
20	import io.github.hyshmily.hotkey.sync.worker.WorkerMessage;
21	import io.github.hyshmily.hotkey.util.version.VersionGuard;
22	import java.nio.charset.StandardCharsets;
23	import java.util.concurrent.atomic.AtomicLong;
24	import lombok.RequiredArgsConstructor;
25	import lombok.extern.slf4j.Slf4j;
26	import org.springframework.amqp.core.Message;
27	import org.springframework.amqp.core.MessageProperties;
28	import org.springframework.amqp.rabbit.core.RabbitTemplate;
29
30	/*
31	* Publishes HOT and COOL decisions to all application instances via the
32	* configured RabbitMQ {@code broadcastExchange}.
33	*
34	* <p>Each broadcast message carries a monotonically increasing
35	* <b>decision version</b> (worker‑local counter) that is used by receivers to
36	* discard stale or out‑of‑order messages.
37	*
38	* <p>Messages are delivered to every instance's dedicated queue through a
39	* fanout exchange ({@code hotkey.broadcast.exchange}) — the receiver
40	* differentiates message type via the {@code AMQP_HEADER_TYPE} header.
41	*/
42	/** Default constructor. */
43	@RequiredArgsConstructor
44	@Slf4j	4✔
45	public class WorkerBroadcaster {
46
47	/** RabbitMQ template used to publish HOT/COOL decisions and heartbeats. */
48	private final RabbitTemplate rabbitTemplate;
49	/** Target exchange name for all broadcast messages (typically {@code hotkey.broadcast.exchange}). */
50	private final String broadcastExchange;
51	/** Application name used in the broadcast routing key. */
52	private final String appName;
53
54	/** Originating Worker's node identity, set on every broadcast header for cross-Worker
55	* version tracking in {@link VersionGuard}. */
56	private final String nodeId;
57
58	/** Monotonically increasing epoch counter for this Worker instance, incremented on every
59	* restart. Transmitted in broadcast headers so receivers detect Worker restarts and
60	* unconditionally accept decisions from a higher epoch (see ADR-0010). */
61	private final AtomicLong epochCounter;
62
63	/**
64	* Worker‑local decision version counter.
65	* Incremented atomically for every broadcast to provide strict ordering
66	* for this worker. Combined with consistent‑hash sharding (one key → one
67	* worker) this guarantees a total order of decisions per key.
68	*/
69	private final AtomicLong decisionVersionCounter = new AtomicLong(System.currentTimeMillis());
70
71	/**
72	* Atomically allocates the next decision version without sending.
73	* Called from within {@link io.github.hyshmily.hotkey.worker.ingest.ReportConsumer#doOnReport}
74	* to pre-allocate a version before enqueuing an async broadcast.
75	*
76	* @return the next monotonically increasing decision version
77	*/
78	public long nextDecisionVersion() {
79	return decisionVersionCounter.incrementAndGet();	×
80	}
81
82	/**
83	* Broadcasts a HOT decision for the given key.
84	*
85	* @param cacheKey the key that has been confirmed as hot
86	* @param source a label describing the detection source (e.g. "sliding_window")
87	*/
88	public void broadcastHot(String cacheKey, String source) {
89	long dv = decisionVersionCounter.incrementAndGet();	4✔
90	sendBroadcast(cacheKey, WorkerMessage.TYPE_HOT, dv);	5✔
91	log.debug(	12✔
92	"Broadcast HOT: key={}, dv={}, source={}, nodeId={}, epoch={}",
93	cacheKey,
94	dv,	15✔
95	source,
96	nodeId,
97	epochCounter.get()	3✔
98	);
99	}	1✔
100
101	/**
102	* Broadcasts a HOT decision with a pre-allocated version number.
103	*
104	* @param cacheKey the key that has been confirmed as hot
105	* @param source a label describing the detection source
106	* @param dv pre-allocated decision version (from {@link #nextDecisionVersion()})
107	*/
108	public void broadcastHot(String cacheKey, String source, long dv) {
109	sendBroadcast(cacheKey, WorkerMessage.TYPE_HOT, dv);	×
110	log.debug(	×
111	"Broadcast HOT: key={}, dv={}, source={}, nodeId={}, epoch={}",
112	cacheKey,
NEW 113	dv,	×
114	source,
115	nodeId,
NEW 116	epochCounter.get()	×
117	);
118	}	×
119
120	/**
121	* Broadcasts a COOL decision for the given key.
122	*
123	* @param cacheKey the key that has been confirmed as fully cooled
124	*/
125	public void broadcastCool(String cacheKey) {
126	long dv = decisionVersionCounter.incrementAndGet();	4✔
127	sendBroadcast(cacheKey, WorkerMessage.TYPE_COOL, dv);	5✔
128	}	1✔
129
130	/**
131	* Broadcasts a COOL decision with a pre-allocated version number.
132	*
133	* @param cacheKey the key that has been confirmed as fully cooled
134	* @param dv pre-allocated decision version (from {@link #nextDecisionVersion()})
135	*/
136	public void broadcastCool(String cacheKey, long dv) {
137	sendBroadcast(cacheKey, WorkerMessage.TYPE_COOL, dv);	×
NEW 138	log.debug("Broadcast COOL: key={}, dv={}, nodeId={}, epoch={}", cacheKey, dv, nodeId, epochCounter.get());	×
UNCOV 139	}	×
140
141	/**
142	* Returns the current decision version without incrementing.
143	*
144	* <p>Used by {@link WorkerHeartbeatProducer} for the heartbeat's decisionVersionHwm field.
145	*
146	* @return the current decision version counter value
147	*/
148	public long getCurrentDecisionVersion() {
149	return decisionVersionCounter.get();	4✔
150	}
151
152	/**
153	* Common send helper for HOT/COOL decisions.
154	*
155	* <p>Builds {@link MessageProperties} with type, version, and degraded flag,
156	* creates a {@link Message}, and publishes via {@link #rabbitTemplate}.
157	*
158	* @param cacheKey the key being broadcast
159	* @param type the message type ({@link WorkerMessage#TYPE_HOT} or {@link WorkerMessage#TYPE_COOL})
160	* @param version the monotonically increasing decision version for ordering
161	*/
162	private void sendBroadcast(String cacheKey, String type, long version) {
163	try {
164	MessageProperties props = new MessageProperties();	4✔
165	props.setHeader(AMQP_HEADER_TYPE, type);	4✔
166	props.setHeader(AMQP_HEADER_VERSION, version);	5✔
167	props.setHeader(AMQP_HEADER_IS_VERSION_DEGRADED, false);	5✔
168	props.setHeader(AMQP_HEADER_NODE_ID, nodeId);	5✔
169	props.setHeader(AMQP_HEADER_EPOCH, epochCounter.get());	7✔
170
171	Message msg = new Message(cacheKey.getBytes(StandardCharsets.UTF_8), props);	8✔
172	rabbitTemplate.send(broadcastExchange, ROUTING_KEY_BROADCAST + appName, msg);	9✔
173	} catch (Exception e) {	1✔
174	log.error("Failed to broadcast {} for key={}, dv={}: {}", type, cacheKey, version, e.getMessage());	23✔
175	// no throw — ADR-0007 fire-and-forget
176	}	1✔
177	}	1✔
178	}

Hyshmily / hotkey / 28916241081

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