27951710162

Committed 22 Jun 2026 12:12PM UTC coverage: 93.275% (+0.04%) from 93.231%

Build # 27951710162

Build Type

push

github

Committed by

Hyshmily

Commit Message

upgrade: JDK 17→21, adopt virtual threads (Project Loom)

Coverage Stats

1163 of 1289 branches covered (90.22%)

Branch coverage included in aggregate %.

8 of 8 new or added lines in 3 files covered. (100.0%)

2 existing lines in 1 file now uncovered.

3331 of 3529 relevant lines covered (94.39%)

4.25 hits per line

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

91.84

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

Hyshmily / hotkey / 27951710162

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