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

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

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