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

92.39

/common/src/main/java/io/github/hyshmily/hotkey/sync/worker/WorkerHeartbeatVerifier.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.sync.worker;

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

import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;

import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpTimeoutException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * On-demand active heartbeat verifier that probes suspected-dead Workers when
 * passive heartbeat monitoring indicates a potential failure.
 *
 * <p>Under normal operation, the {@link ClusterHealthView} is updated passively
 * by incoming heartbeat messages. This verifier activates only when one or more
 * Workers have exceeded the {@code heartbeatTimeoutMs} window — it sends a
 * point-to-point PING message to each suspected Worker's dedicated verification
 * queue ({@code hotkey.verify.ping.{workerId}}) via Direct reply-to and awaits
 * a PONG response.
 *
 * <p><b>Failure escalation:</b> If cumulative failures across all suspected Workers
 * reach {@code degradeAfterFailures} and the cluster remains unhealthy, the health
 * view is marked as degraded via {@link ClusterHealthView#degraded}, triggering
 * graceful degradation behaviour (see ADR-0009). Workers that respond are restored
 * to the alive set; Workers that fail repeatedly are marked as stale and excluded
 * from majority-based health checks.
 *
 * <p><b>Thread safety:</b> The verification loop runs on a single daemon thread
 * ({@code hb-verifier}). The {@link ClusterHealthView} uses {@code ConcurrentHashMap}
 * internally, so concurrent heartbeat reception and verification are safe.
 *
 * @see ClusterHealthView
 * @see WorkerHeartbeatMessage
 */
@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
public class WorkerHeartbeatVerifier {

  private final RabbitTemplate rabbitTemplate;
  private final ClusterHealthView healthView;
  private final String appInstanceId;
  private final long verifyIntervalMs;
  private final long pingTimeoutMs;
  private final int degradeAfterFailures;
  private final long verifyMaxBackoffMs;
  private final ScheduledExecutorService scheduler;
  private final ConcurrentHashMap<String, Long> nextVerifyTime = new ConcurrentHashMap<>();
  private final boolean ownsScheduler;
  private ScheduledFuture<?> verifyTask;

  private static final String QUEUE_VERIFY_PING_PREFIX = "hotkey.verify.ping.";
  private static final int MAX_BACKOFF_SHIFT = 30;

  /**
   * Parameter object for {@link WorkerHeartbeatVerifier} construction.
   */
  public record VerifierConfig(
    long verifyIntervalMs,
    long pingTimeoutMs,
    int degradeAfterFailures,
    long verifyMaxBackoffMs
  ) {}

  /**
   * Creates a verifier with an internally owned single-thread scheduler.
   * The internal daemon thread is named {@code hb-verifier}.
   */
  public WorkerHeartbeatVerifier(
    RabbitTemplate rabbitTemplate,
    ClusterHealthView healthView,
    String appInstanceId,
    VerifierConfig config
  ) {
    this(
      rabbitTemplate,
      healthView,
      appInstanceId,
      config.verifyIntervalMs,
      config.pingTimeoutMs,
      config.degradeAfterFailures,
      config.verifyMaxBackoffMs,
      Executors.newSingleThreadScheduledExecutor(new HotKeyThreadFactory("hotkey-hb-verifier")),
      true
    );
  }

  /**
   * Creates a verifier with a caller-supplied external scheduler.
   * The caller manages the scheduler lifecycle — {@link #stop()} cancels
   * the task but does not shut down the scheduler.
   */
  public WorkerHeartbeatVerifier(
    RabbitTemplate rabbitTemplate,
    ClusterHealthView healthView,
    String appInstanceId,
    VerifierConfig config,
    ScheduledExecutorService scheduler
  ) {
    this(
      rabbitTemplate,
      healthView,
      appInstanceId,
      config.verifyIntervalMs,
      config.pingTimeoutMs,
      config.degradeAfterFailures,
      config.verifyMaxBackoffMs,
      scheduler,
      false
    );
  }

  /**
   * Starts the periodic heartbeat verification at a fixed rate.
   *
   * <p>Every {@code verifyIntervalMs} milliseconds, iterates over all Workers in
   * {@link ClusterHealthView} that have exceeded {@code heartbeatTimeoutMs} without
   * a heartbeat and sends each a PING via Direct reply-to. On PONG, the Worker's
   * health record is restored; on timeout, the failure counter is incremented.
   * <p>
   * The verification runs on a daemon background thread. The task is idempotent:
   * subsequent calls to this method after the verifier is already running are
   * silently ignored.
   *
   * @see #verifySuspectedWorkers
   * @see #stop()
   */
  public void start() {
    if (verifyTask != null) {
      return;
    }
    try {
      verifyTask = scheduler.scheduleWithFixedDelay(
        this::verifySuspectedWorkers,
        verifyIntervalMs,
        verifyIntervalMs,
        TimeUnit.MILLISECONDS
      );
    } catch (Exception e) {
      log.error(
        "Failed to start heartbeat verifier scheduler; Worker liveness verification " +
          "will not run. Application continues but stale Workers may not be detected.",
        e
      );
    }
  }

  /**
   * Gracefully stops the periodic heartbeat verification.
   *
   * <p>Cancels the scheduled verification task and, if this verifier owns its
   * scheduler (created via the 6-argument constructor), shuts down the scheduler.
   * When an external scheduler is used (7-argument constructor), the scheduler is
   * left running for the caller to manage.
   *
   * <p>After {@code stop()}, the verifier can be restarted by calling {@link #start()}
   * again.
   */
  public void stop() {
    if (verifyTask != null) {
      verifyTask.cancel(false);
    }
    if (ownsScheduler) {
      scheduler.shutdown();
    }
  }

  /**
   * Iterates over all Workers registered in {@link ClusterHealthView} that are
   * <em>not</em> currently marked alive, sends each a PING via Direct reply-to,
   * and updates the health view based on the response.
   *
   * <p>If the cluster is healthy (majority of Workers alive), the verification
   * is skipped entirely — no need for active probing. When cumulative failures
   * across all suspected Workers reach {@code degradeAfterFailures} and the cluster
   * remains unhealthy, the cluster is marked as degraded.
   *
   * <p>This method is called periodically by the scheduled task and is also
   * safe to invoke manually for testing.
   */
  public void verifySuspectedWorkers() {
    try {
      if (healthView.isClusterHealthy()) {
        return;
      }

      Set<String> suspected = healthView
        .getAllWorkerIds()
        .stream()
        .filter(id -> !healthView.getAliveWorkerIds().contains(id))
        .collect(Collectors.toSet());

      if (suspected.isEmpty()) {
        return;
      }

      log.info("Verifying suspected workers: {}", suspected);

      for (String workerId : suspected) {
        Long skipUntil = nextVerifyTime.get(workerId);
        if (skipUntil != null && currentTimeMillis() < skipUntil) {
          log.trace("Worker {} in backoff, skip (remaining={}ms)", workerId, skipUntil - currentTimeMillis());
          continue;
        }

        boolean alive = sendPingAndWaitPong(workerId);
        if (!alive) {
          healthView.markVerificationFailed(workerId);

          int attempt = healthView.getVerifyFailures(workerId);
          long backoffMs = computeBackoffMs(attempt);

          nextVerifyTime.put(workerId, currentTimeMillis() + backoffMs);
          log.warn("Worker {} verification failed (attempt={}, backoff={}ms)", workerId, attempt, backoffMs);
        } else {
          nextVerifyTime.remove(workerId);
          healthView.recordPong(workerId);
        }
      }

      boolean anyDegraded = suspected.stream().anyMatch(id -> healthView.getVerifyFailures(id) >= degradeAfterFailures);
      if (anyDegraded && !healthView.isClusterHealthy()) {
        log.warn("Cluster degraded: some workers unreachable after verification (threshold={})", degradeAfterFailures);
        healthView.setDegraded(true);
      }
    } catch (Exception e) {
      log.error("Scheduled verifySuspectedWorkers failed", e);
    }
  }

  /**
   * Compute the exponential backoff duration for a verification attempt.
   * Starts at {@code verifyIntervalMs} and doubles each attempt, capped at
   * {@code verifyMaxBackoffMs}.
   *
   * @param attempt the number of consecutive failures (1-based)
   * @return the backoff duration in milliseconds
   */
  protected long computeBackoffMs(int attempt) {
    return Math.min(verifyMaxBackoffMs, verifyIntervalMs * (1L << Math.min(attempt, MAX_BACKOFF_SHIFT)));
  }

  /**
   * Sends a point-to-point PING message to the given Worker's dedicated verification
   * queue ({@code hotkey.verify.ping.{workerId}}) and waits for a PONG response
   * via AMQP Direct reply-to ({@code amq.rabbitmq.reply-to}).
   *
   * <p>The PING carries the application instance ID in the
   * {@link io.github.hyshmily.hotkey.constants.HotKeyConstants#AMQP_HEADER_VERIFY_APP_INSTANCE} header so the Worker
   * can identify the requester. The reply timeout is set in the message properties.
   *
   * @param workerId the Worker to ping; must not be null
   * @return {@code true} if a non-null PONG response was received within
   *         {@code pingTimeoutMs}; {@code false} if the request timed out
   *         ({@link AmqpTimeoutException}) or the queue is unreachable
   */
  public boolean sendPingAndWaitPong(String workerId) {
    MessageProperties props = new MessageProperties();
    props.setHeader(AMQP_HEADER_VERIFY_TYPE, AMQP_HEADER_VERIFY_PING);
    props.setHeader(AMQP_HEADER_VERIFY_APP_INSTANCE, appInstanceId);
    props.setReplyTo("amq.rabbitmq.reply-to");

    Message ping = new Message(new byte[0], props);

    try {
      rabbitTemplate.setReplyTimeout((int) pingTimeoutMs);
      Message pong = rabbitTemplate.sendAndReceive("", QUEUE_VERIFY_PING_PREFIX + workerId, ping);
      return pong != null;
    } catch (AmqpTimeoutException e) {
      return false;
    }
  }
}

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.sync.worker;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
19
20	import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;
21
22	import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
23	import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
24	import java.util.Set;
25	import java.util.concurrent.*;
26	import java.util.stream.Collectors;
27	import lombok.AccessLevel;
28	import lombok.RequiredArgsConstructor;
29	import lombok.extern.slf4j.Slf4j;
30	import org.springframework.amqp.AmqpTimeoutException;
31	import org.springframework.amqp.core.Message;
32	import org.springframework.amqp.core.MessageProperties;
33	import org.springframework.amqp.rabbit.core.RabbitTemplate;
34
35	/**
36	* On-demand active heartbeat verifier that probes suspected-dead Workers when
37	* passive heartbeat monitoring indicates a potential failure.
38	*
39	* <p>Under normal operation, the {@link ClusterHealthView} is updated passively
40	* by incoming heartbeat messages. This verifier activates only when one or more
41	* Workers have exceeded the {@code heartbeatTimeoutMs} window — it sends a
42	* point-to-point PING message to each suspected Worker's dedicated verification
43	* queue ({@code hotkey.verify.ping.{workerId}}) via Direct reply-to and awaits
44	* a PONG response.
45	*
46	* <p><b>Failure escalation:</b> If cumulative failures across all suspected Workers
47	* reach {@code degradeAfterFailures} and the cluster remains unhealthy, the health
48	* view is marked as degraded via {@link ClusterHealthView#degraded}, triggering
49	* graceful degradation behaviour (see ADR-0009). Workers that respond are restored
50	* to the alive set; Workers that fail repeatedly are marked as stale and excluded
51	* from majority-based health checks.
52	*
53	* <p><b>Thread safety:</b> The verification loop runs on a single daemon thread
54	* ({@code hb-verifier}). The {@link ClusterHealthView} uses {@code ConcurrentHashMap}
55	* internally, so concurrent heartbeat reception and verification are safe.
56	*
57	* @see ClusterHealthView
58	* @see WorkerHeartbeatMessage
59	*/
60	@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
61	@Slf4j	4✔
62	public class WorkerHeartbeatVerifier {
63
64	private final RabbitTemplate rabbitTemplate;
65	private final ClusterHealthView healthView;
66	private final String appInstanceId;
67	private final long verifyIntervalMs;
68	private final long pingTimeoutMs;
69	private final int degradeAfterFailures;
70	private final long verifyMaxBackoffMs;
71	private final ScheduledExecutorService scheduler;
72	private final ConcurrentHashMap<String, Long> nextVerifyTime = new ConcurrentHashMap<>();
73	private final boolean ownsScheduler;
74	private ScheduledFuture<?> verifyTask;
75
76	private static final String QUEUE_VERIFY_PING_PREFIX = "hotkey.verify.ping.";
77	private static final int MAX_BACKOFF_SHIFT = 30;
78
79	/**
80	* Parameter object for {@link WorkerHeartbeatVerifier} construction.
81	*/
82	public record VerifierConfig(	15✔
83	long verifyIntervalMs,
84	long pingTimeoutMs,
85	int degradeAfterFailures,
86	long verifyMaxBackoffMs
87	) {}
88
89	/**
90	* Creates a verifier with an internally owned single-thread scheduler.
91	* The internal daemon thread is named {@code hb-verifier}.
92	*/
93	public WorkerHeartbeatVerifier(
94	RabbitTemplate rabbitTemplate,
95	ClusterHealthView healthView,
96	String appInstanceId,
97	VerifierConfig config
98	) {
99	this(	17✔
100	rabbitTemplate,
101	healthView,
102	appInstanceId,
103	config.verifyIntervalMs,
104	config.pingTimeoutMs,
105	config.degradeAfterFailures,
106	config.verifyMaxBackoffMs,
107	Executors.newSingleThreadScheduledExecutor(new HotKeyThreadFactory("hotkey-hb-verifier")),	2✔
108	true
109	);
110	}	1✔
111
112	/**
113	* Creates a verifier with a caller-supplied external scheduler.
114	* The caller manages the scheduler lifecycle — {@link #stop()} cancels
115	* the task but does not shut down the scheduler.
116	*/
117	public WorkerHeartbeatVerifier(
118	RabbitTemplate rabbitTemplate,
119	ClusterHealthView healthView,
120	String appInstanceId,
121	VerifierConfig config,
122	ScheduledExecutorService scheduler
123	) {
124	this(	15✔
125	rabbitTemplate,
126	healthView,
127	appInstanceId,
128	config.verifyIntervalMs,
129	config.pingTimeoutMs,
130	config.degradeAfterFailures,
131	config.verifyMaxBackoffMs,
132	scheduler,
133	false
134	);
135	}	1✔
136
137	/**
138	* Starts the periodic heartbeat verification at a fixed rate.
139	*
140	* <p>Every {@code verifyIntervalMs} milliseconds, iterates over all Workers in
141	* {@link ClusterHealthView} that have exceeded {@code heartbeatTimeoutMs} without
142	* a heartbeat and sends each a PING via Direct reply-to. On PONG, the Worker's
143	* health record is restored; on timeout, the failure counter is incremented.
144	* <p>
145	* The verification runs on a daemon background thread. The task is idempotent:
146	* subsequent calls to this method after the verifier is already running are
147	* silently ignored.
148	*
149	* @see #verifySuspectedWorkers
150	* @see #stop()
151	*/
152	public void start() {
153	if (verifyTask != null) {	3✔
154	return;	1✔
155	}
156	try {
157	verifyTask = scheduler.scheduleWithFixedDelay(	12✔
158	this::verifySuspectedWorkers,
159	verifyIntervalMs,
160	verifyIntervalMs,
161	TimeUnit.MILLISECONDS
162	);
163	} catch (Exception e) {	1✔
164	log.error(	4✔
165	"Failed to start heartbeat verifier scheduler; Worker liveness verification " +
166	"will not run. Application continues but stale Workers may not be detected.",
167	e
168	);
169	}	1✔
170	}	1✔
171
172	/**
173	* Gracefully stops the periodic heartbeat verification.
174	*
175	* <p>Cancels the scheduled verification task and, if this verifier owns its
176	* scheduler (created via the 6-argument constructor), shuts down the scheduler.
177	* When an external scheduler is used (7-argument constructor), the scheduler is
178	* left running for the caller to manage.
179	*
180	* <p>After {@code stop()}, the verifier can be restarted by calling {@link #start()}
181	* again.
182	*/
183	public void stop() {
184	if (verifyTask != null) {	3✔
185	verifyTask.cancel(false);	5✔
186	}
187	if (ownsScheduler) {	3!
188	scheduler.shutdown();	3✔
189	}
190	}	1✔
191
192	/**
193	* Iterates over all Workers registered in {@link ClusterHealthView} that are
194	* <em>not</em> currently marked alive, sends each a PING via Direct reply-to,
195	* and updates the health view based on the response.
196	*
197	* <p>If the cluster is healthy (majority of Workers alive), the verification
198	* is skipped entirely — no need for active probing. When cumulative failures
199	* across all suspected Workers reach {@code degradeAfterFailures} and the cluster
200	* remains unhealthy, the cluster is marked as degraded.
201	*
202	* <p>This method is called periodically by the scheduled task and is also
203	* safe to invoke manually for testing.
204	*/
205	public void verifySuspectedWorkers() {
206	try {
207	if (healthView.isClusterHealthy()) {	4✔
208	return;	1✔
209	}
210
211	Set<String> suspected = healthView	2✔
212	.getAllWorkerIds()	1✔
213	.stream()	3✔
214	.filter(id -> !healthView.getAliveWorkerIds().contains(id))	11✔
215	.collect(Collectors.toSet());	4✔
216
217	if (suspected.isEmpty()) {	3✔
218	return;	1✔
219	}
220
221	log.info("Verifying suspected workers: {}", suspected);	4✔
222
223	for (String workerId : suspected) {	10✔
224	Long skipUntil = nextVerifyTime.get(workerId);	6✔
225	if (skipUntil != null && currentTimeMillis() < skipUntil) {	2!
UNCOV 226	log.trace("Worker {} in backoff, skip (remaining={}ms)", workerId, skipUntil - currentTimeMillis());	×
227	continue;	×
228	}
229
230	boolean alive = sendPingAndWaitPong(workerId);	4✔
231	if (!alive) {	2✔
232	healthView.markVerificationFailed(workerId);	4✔
233
234	int attempt = healthView.getVerifyFailures(workerId);	5✔
235	long backoffMs = computeBackoffMs(attempt);	4✔
236
237	nextVerifyTime.put(workerId, currentTimeMillis() + backoffMs);	9✔
238	log.warn("Worker {} verification failed (attempt={}, backoff={}ms)", workerId, attempt, backoffMs);	19✔
239	} else {	1✔
240	nextVerifyTime.remove(workerId);	5✔
241	healthView.recordPong(workerId);	4✔
242	}
243	}	1✔
244
245	boolean anyDegraded = suspected.stream().anyMatch(id -> healthView.getVerifyFailures(id) >= degradeAfterFailures);	17✔
246	if (anyDegraded && !healthView.isClusterHealthy()) {	6!
247	log.warn("Cluster degraded: some workers unreachable after verification (threshold={})", degradeAfterFailures);	6✔
248	healthView.setDegraded(true);	4✔
249	}
250	} catch (Exception e) {	1✔
251	log.error("Scheduled verifySuspectedWorkers failed", e);	4✔
252	}	1✔
253	}	1✔
254
255	/**
256	* Compute the exponential backoff duration for a verification attempt.
257	* Starts at {@code verifyIntervalMs} and doubles each attempt, capped at
258	* {@code verifyMaxBackoffMs}.
259	*
260	* @param attempt the number of consecutive failures (1-based)
261	* @return the backoff duration in milliseconds
262	*/
263	protected long computeBackoffMs(int attempt) {
264	return Math.min(verifyMaxBackoffMs, verifyIntervalMs * (1L << Math.min(attempt, MAX_BACKOFF_SHIFT)));	12✔
265	}
266
267	/**
268	* Sends a point-to-point PING message to the given Worker's dedicated verification
269	* queue ({@code hotkey.verify.ping.{workerId}}) and waits for a PONG response
270	* via AMQP Direct reply-to ({@code amq.rabbitmq.reply-to}).
271	*
272	* <p>The PING carries the application instance ID in the
273	* {@link io.github.hyshmily.hotkey.constants.HotKeyConstants#AMQP_HEADER_VERIFY_APP_INSTANCE} header so the Worker
274	* can identify the requester. The reply timeout is set in the message properties.
275	*
276	* @param workerId the Worker to ping; must not be null
277	* @return {@code true} if a non-null PONG response was received within
278	* {@code pingTimeoutMs}; {@code false} if the request timed out
279	* ({@link AmqpTimeoutException}) or the queue is unreachable
280	*/
281	public boolean sendPingAndWaitPong(String workerId) {
282	MessageProperties props = new MessageProperties();	4✔
283	props.setHeader(AMQP_HEADER_VERIFY_TYPE, AMQP_HEADER_VERIFY_PING);	4✔
284	props.setHeader(AMQP_HEADER_VERIFY_APP_INSTANCE, appInstanceId);	5✔
285	props.setReplyTo("amq.rabbitmq.reply-to");	3✔
286
287	Message ping = new Message(new byte[0], props);	7✔
288
289	try {
290	rabbitTemplate.setReplyTimeout((int) pingTimeoutMs);	7✔
291	Message pong = rabbitTemplate.sendAndReceive("", QUEUE_VERIFY_PING_PREFIX + workerId, ping);	8✔
292	return pong != null;	6✔
293	} catch (AmqpTimeoutException e) {	1✔
294	return false;	2✔
295	}
296	}
297	}

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