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

91.49

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

import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
import io.github.hyshmily.hotkey.sync.worker.WorkerHeartbeatMessage;
import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
import jakarta.annotation.PostConstruct;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;

/**
 * Listens for heartbeat-based config updates from peer Workers and applies them
 * if the received config timestamp is newer than the local one.
 *
 * <p>On startup, waits up to 3 seconds for the first heartbeat to arrive.
 * If none is received, the Worker continues with the values from
 * {@link WorkerProperties} — this can happen when all other Workers are down.
 */
@RequiredArgsConstructor
@Slf4j
public class WorkerConfigNegotiator {

  /** State machine whose config (confirm/cool/grace counts) is updated from heartbeat messages. */
  private final HotKeyStateMachine stateMachine;
  /** Monotonically increasing counter tracking the latest config-change timestamp. */
  private final AtomicLong configTimestampCounter;
  /** Unique identifier for this Worker node, used in queue names and heartbeat identification. */
  private final String nodeId;
  private final CountDownLatch startupLatch = new CountDownLatch(1);

  /**
   * Waits for the first heartbeat from a peer Worker.
   *
   * <p>If a heartbeat with a valid config arrives within 3 seconds the
   * config is applied synchronously.  Otherwise, a warning is logged and
   * the Worker proceeds with configured defaults.
   */
  @PostConstruct
  void syncOnStartup() {
    Thread waitThread = new HotKeyThreadFactory("hotkey-config-sync-startup").newThread(
      () -> {
        try {
          boolean received = startupLatch.await(3000, TimeUnit.MILLISECONDS);
          if (!received) {
            log.warn("No config heartbeat received within 3s, using WorkerProperties defaults");
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      }
    );
    waitThread.start();
  }

  /**
   * Processes an incoming heartbeat message from a peer Worker and applies
   * its state-machine config if the embedded timestamp is newer.
   *
   * @param msg the raw AMQP heartbeat message
   */
  @RabbitListener(queues = "#{@workerConfigQueue.name}", containerFactory = "workerConfigListenerContainerFactory")
  public void onHeartbeat(Message msg) {
    try {
      doOnHeartbeat(msg);
    } catch (Exception e) {
      log.warn("Uncaught exception in onHeartbeat config negotiation, discarding message to prevent requeue loop", e);
    }
  }

  private void doOnHeartbeat(Message msg) {
    WorkerHeartbeatMessage hb = WorkerHeartbeatMessage.from(msg);
    if (hb == null) {
      return;
    }

    if (hb.workerId().equals(nodeId)) {
      return;
    }

    long remoteTs = hb.configTimestamp();
    long localTs = configTimestampCounter.get();
    if (remoteTs <= localTs) {
      return;
    }

    stateMachine.setConfirmCount(hb.configConfirmCount());
    stateMachine.setCoolCount(hb.configCoolCount());
    stateMachine.setPreCoolGraceCount(hb.configGraceCount());

    configTimestampCounter.set(remoteTs);

    log.debug(
      "Applied newer config from {}: confirmCount={}, coolCount={}, preCoolGraceCount={}",
      hb.workerId(),
      hb.configConfirmCount(),
      hb.configCoolCount(),
      hb.configGraceCount()
    );

    if (startupLatch.getCount() > 0) {
      startupLatch.countDown();
    }
  }
}

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.config;
17
18	import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
19	import io.github.hyshmily.hotkey.sync.worker.WorkerHeartbeatMessage;
20	import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
21	import jakarta.annotation.PostConstruct;
22	import java.util.concurrent.CountDownLatch;
23	import java.util.concurrent.TimeUnit;
24	import java.util.concurrent.atomic.AtomicLong;
25	import lombok.RequiredArgsConstructor;
26	import lombok.extern.slf4j.Slf4j;
27	import org.springframework.amqp.core.Message;
28	import org.springframework.amqp.rabbit.annotation.RabbitListener;
29
30	/**
31	* Listens for heartbeat-based config updates from peer Workers and applies them
32	* if the received config timestamp is newer than the local one.
33	*
34	* <p>On startup, waits up to 3 seconds for the first heartbeat to arrive.
35	* If none is received, the Worker continues with the values from
36	* {@link WorkerProperties} — this can happen when all other Workers are down.
37	*/
38	@RequiredArgsConstructor
39	@Slf4j	4✔
40	public class WorkerConfigNegotiator {
41
42	/** State machine whose config (confirm/cool/grace counts) is updated from heartbeat messages. */
43	private final HotKeyStateMachine stateMachine;
44	/** Monotonically increasing counter tracking the latest config-change timestamp. */
45	private final AtomicLong configTimestampCounter;
46	/** Unique identifier for this Worker node, used in queue names and heartbeat identification. */
47	private final String nodeId;
48	private final CountDownLatch startupLatch = new CountDownLatch(1);
49
50	/**
51	* Waits for the first heartbeat from a peer Worker.
52	*
53	* <p>If a heartbeat with a valid config arrives within 3 seconds the
54	* config is applied synchronously. Otherwise, a warning is logged and
55	* the Worker proceeds with configured defaults.
56	*/
57	@PostConstruct
58	void syncOnStartup() {
59	Thread waitThread = new HotKeyThreadFactory("hotkey-config-sync-startup").newThread(	8✔
60	() -> {
61	try {
62	boolean received = startupLatch.await(3000, TimeUnit.MILLISECONDS);	6✔
63	if (!received) {	2✔
64	log.warn("No config heartbeat received within 3s, using WorkerProperties defaults");	3✔
65	}
UNCOV 66	} catch (InterruptedException e) {	×
UNCOV 67	Thread.currentThread().interrupt();	×
68	}	1✔
69	}	1✔
70	);
71	waitThread.start();	2✔
72	}	1✔
73
74	/**
75	* Processes an incoming heartbeat message from a peer Worker and applies
76	* its state-machine config if the embedded timestamp is newer.
77	*
78	* @param msg the raw AMQP heartbeat message
79	*/
80	@RabbitListener(queues = "#{@workerConfigQueue.name}", containerFactory = "workerConfigListenerContainerFactory")
81	public void onHeartbeat(Message msg) {
82	try {
83	doOnHeartbeat(msg);	3✔
UNCOV 84	} catch (Exception e) {	×
UNCOV 85	log.warn("Uncaught exception in onHeartbeat config negotiation, discarding message to prevent requeue loop", e);	×
86	}	1✔
87	}	1✔
88
89	private void doOnHeartbeat(Message msg) {
90	WorkerHeartbeatMessage hb = WorkerHeartbeatMessage.from(msg);	3✔
91	if (hb == null) {	2✔
92	return;	1✔
93	}
94
95	if (hb.workerId().equals(nodeId)) {	6✔
96	return;	1✔
97	}
98
99	long remoteTs = hb.configTimestamp();	3✔
100	long localTs = configTimestampCounter.get();	4✔
101	if (remoteTs <= localTs) {	4✔
102	return;	1✔
103	}
104
105	stateMachine.setConfirmCount(hb.configConfirmCount());	5✔
106	stateMachine.setCoolCount(hb.configCoolCount());	5✔
107	stateMachine.setPreCoolGraceCount(hb.configGraceCount());	5✔
108
109	configTimestampCounter.set(remoteTs);	4✔
110
111	log.debug(	8✔
112	"Applied newer config from {}: confirmCount={}, coolCount={}, preCoolGraceCount={}",
113	hb.workerId(),	5✔
114	hb.configConfirmCount(),	6✔
115	hb.configCoolCount(),	6✔
116	hb.configGraceCount()	3✔
117	);
118
119	if (startupLatch.getCount() > 0) {	6✔
120	startupLatch.countDown();	3✔
121	}
122	}	1✔
123	}

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