27890525626

Committed 21 Jun 2026 02:07AM UTC coverage: 93.615% (-0.1%) from 93.742%

Build # 27890525626

Build Type

push

github

Committed by

Hyshmily

Commit Message

feat: cross-Worker decisionVersion partitioning with nodeId/epoch + expectedWorkerCount

Core changes:
- VersionGuard: 4-param/5-param overloads with nodeId/epoch for per-Worker
  version space isolation
- WorkerMessage: add nodeId/epoch fields, broadcast as AMQP headers
- WorkerListener: propagate nodeId/epoch through DCL into CacheEntry
- CacheEntry: add decisionNodeId/decisionEpoch fields
- WorkerBroadcaster: stamp every HOT/COOL broadcast with nodeId + epoch
- WorkerAutoConfiguration: @Bean workerNodeId(), workerEpochCounter()
- HotKeyConstants: AMQP_HEADER_EPOCH constant

Cluster health:
- HotKeyProperties: expectedWorkerCount (default 0=dynamic)
- ClusterHealthView: majority-quorum (> expectedWorkerCount / 2) health check
- Wire through HotKeyAutoConfiguration, HotKeyAmqpAutoConfiguration,
  HotKeyRedisAutoConfiguration

Production hardening:
- CacheExpireManager: CompletableFuture.orTimeout() with
  RejectedExecutionException catch to prevent stuck refresh markers
- HotKeyStateMachine: updated state transitions
- RingManager: consistent-hash ring refinements
- ReportConsumer: ingest path improvements

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

Coverage Stats

1133 of 1265 branches covered (89.57%)

Branch coverage included in aggregate %.

149 of 160 new or added lines in 15 files covered. (93.13%)

1 existing line in 1 file now uncovered.

3280 of 3449 relevant lines covered (95.1%)

4.31 hits per line

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

91.55

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

import com.github.benmanes.caffeine.cache.Cache;
import io.github.hyshmily.hotkey.model.CacheEntry;
import java.util.Objects;

/**
 * Shared version comparison logic for broadcast message guards used by both
 * the Worker decision listener and the instance-to-instance cache sync listener.
 *
 * <p>This class provides two families of guard methods, each with a fast-path
 * variant (accepting a {@link Cache} reference) and an entry-level variant
 * (accepting an existing {@link CacheEntry}) for use inside atomic
 * {@code compute} blocks:
 *
 * <ul>
 *   <li><b>{@link #shouldSkipForWorker}</b> — Compares {@code decisionVersion}
 *       from Worker HOT/COOL broadcasts. Degraded entries (created during a
 *       Redis outage) unconditionally accept incoming decisions, providing a
 *       safety net against Worker restarts that reset the {@code AtomicLong}
 *       counter (see ADR-0008, ADR-0009).</li>
 *   <li><b>{@link #shouldSkipForSync}</b> — Compares {@code dataVersion}
 *       from application-level data-mutation broadcasts. Uses a 4-case degraded
 *       comparison matrix:
 *       <ol>
 *         <li>Both normal: skip if existing {@code >=} incoming</li>
 *         <li>Existing normal, incoming degraded: always skip (normal wins)</li>
 *         <li>Both degraded: skip if existing {@code >=} incoming</li>
 *         <li>Existing degraded, incoming normal: never skip (normal overwrites degraded)</li>
 *       </ol>
 *   </li>
 * </ul>
 *
 * <p>All methods are stateless and thread-safe. Instances of this utility class
 * must never be created.
 *
 * @see WorkerListener
 * @see CacheSyncListener
 */
public final class VersionGuard {

  /**
   * Utility class — prevent instantiation.
   */
  private VersionGuard() {}

  /**
   * WorkerListener guard with epoch and node-id awareness.
   * <p>Decision logic:
   * <ol>
   *   <li>No existing entry → accept (return false)</li>
   *   <li>Existing entry is degraded → accept unconditionally (safety net)</li>
 *   <li>Incoming epoch &gt; existing epoch → accept unconditionally
 *       (Worker restart detected — ADR-0010)</li>
 *   <li>Incoming epoch &lt; existing epoch → skip (stale incarnation message)</li>
 *   <li>Same epoch, same nodeId → normal ordering (skip if existing dv &gt;= incoming dv)</li>
   *   <li>Different nodeId, same epoch → cross-Worker ownership transfer;
   *       accept (return false) to allow new owner to assert authority</li>
   * </ol>
   *
   * @param existing              the existing cache entry; may be null
   * @param incomingDecisionVersion the decision version from the incoming Worker message
   * @param incomingNodeId        the originating Worker's node ID
   * @param incomingEpoch         the originating Worker's epoch
   * @return true if the incoming message should be skipped, false if it should be applied
   */
  public static boolean shouldSkipForWorker(
    CacheEntry existing,
    long incomingDecisionVersion,
    String incomingNodeId,
    long incomingEpoch
  ) {
    if (existing == null) {
      return false;
    }
    if (existing.isVersionDegraded()) {
      return false;
    }

    if (incomingEpoch > existing.getDecisionEpoch()) {
      return false;
    }
    if (incomingEpoch < existing.getDecisionEpoch()) {
      return true;
    }

    if (Objects.equals(incomingNodeId, existing.getDecisionNodeId())) {
      return existing.getDecisionVersion() >= incomingDecisionVersion;
    }

    return false;
  }

  /**
   * WorkerListener guard for use inside an atomic {@code compute} block: the caller
   * already holds the existing entry reference, so no redundant {@code getIfPresent}
   * is needed.
   *
   * <p>Returns {@code false} (accept) for any of the following:
   * <ul>
   *   <li>No existing entry ({@code null})</li>
   *   <li>Existing entry is degraded ({@code isVersionDegraded == true}) —
   *       yields to any incoming decision, even one with a lower version,
   *       because the degraded entry was written during a Redis outage</li>
   * </ul>
   *
   * Otherwise, skips when the existing entry's {@code decisionVersion} is
   * {@code >=} the incoming version.
   *
   * @param existing                 the existing cache entry; may be {@code null}
   * @param incomingDecisionVersion  the decision version from the incoming Worker message;
   *                                 must be non-negative in normal operation
   * @return {@code true} if the incoming message should be skipped, {@code false}
   *         if the decision should be applied
   */
  public static boolean shouldSkipForWorker(CacheEntry existing, long incomingDecisionVersion) {
    if (existing == null) {
      return false;
    }
    if (existing.isVersionDegraded()) {
      return false;
    }
    return existing.getDecisionVersion() >= incomingDecisionVersion;
  }

  /**
   * WorkerListener guard with a cache-level fast path using legacy 3-argument signature.
   * Delegates to the 5-argument overload with {@code null} node ID and {@code 0} epoch.
   *
   * @param cache                    the local Caffeine L1 cache; must not be null
   * @param cacheKey                 the cache key to look up; must not be null
   * @param incomingDecisionVersion  the decision version from the incoming Worker message
   * @return {@code true} if the incoming message should be skipped (existing entry
   *         is already up-to-date); {@code false} if the decision may need to be applied
   */
  public static boolean shouldSkipForWorker(
    Cache<String, Object> cache,
    String cacheKey,
    long incomingDecisionVersion
  ) {
    return shouldSkipForWorker(cache, cacheKey, incomingDecisionVersion, null, 0);
  }

  /**
   * WorkerListener guard with a cache-level fast path: fetches the existing entry
   * from the L1 cache and delegates to the entry-level overload.
   *
   * <p>This variant is used <em>outside</em> atomic {@code compute} blocks as a
   * cheap first-pass check. If it returns {@code true} (skip), the caller can avoid
   * the more expensive Redis fetch entirely. A second guard inside the {@code compute}
   * block is still needed for correctness (DCL pattern).
   *
   * @param cache                    the local Caffeine L1 cache; must not be null
   * @param cacheKey                 the cache key to look up; must not be null
   * @param incomingDecisionVersion  the decision version from the incoming Worker message
   * @param incomingNodeId           the originating Worker's node ID, may be null
   * @param incomingEpoch            the originating Worker's epoch
   * @return {@code true} if the incoming message should be skipped (existing entry
   *         is already up-to-date); {@code false} if the decision may need to be applied
   */
  public static boolean shouldSkipForWorker(
    Cache<String, Object> cache,
    String cacheKey,
    long incomingDecisionVersion,
    String incomingNodeId,
    long incomingEpoch
  ) {
    Object existing = cache.getIfPresent(cacheKey);
    if (existing instanceof CacheEntry existingCacheEntry) {
      return shouldSkipForWorker(existingCacheEntry, incomingDecisionVersion, incomingNodeId, incomingEpoch);
    }
    return false;
  }

  /**
   * CacheSyncListener guard for use inside an atomic {@code compute} block: the caller
   * already holds the existing entry reference.
   *
   * <p>Applies the 4-case degraded comparison matrix:
   * <ol>
   *   <li>Both normal: skip if existing {@code >=} incoming</li>
   *   <li>Existing normal, incoming degraded: always skip (normal wins)</li>
   *   <li>Both degraded: skip if existing {@code >=} incoming</li>
   *   <li>Existing degraded, incoming normal: never skip (normal overwrites degraded)</li>
   * </ol>
   *
   * <p>This design ensures that a single healthy Redis-backed instance can
   * always overwrite degraded entries from other instances, while preventing
   * degraded broadcasts from reverting healthy entries.
   *
   * @param existing             the existing cache entry; may be {@code null} (returns {@code false})
   * @param incomingDataVersion  the data version from the incoming sync message
   * @param incomingDegraded     {@code true} if the incoming sync message was sent in degraded mode
   * @return {@code true} if the incoming refresh should be skipped;
   *         {@code false} if it should be applied
   */
  public static boolean shouldSkipForSync(CacheEntry existing, long incomingDataVersion, boolean incomingDegraded) {
    if (existing == null) {
      return false;
    }

    boolean existingDegraded = existing.isVersionDegraded();

    // Both normal
    if (!existingDegraded && !incomingDegraded) {
      return existing.getDataVersion() >= incomingDataVersion;
    }
    // Existing normal, incoming degraded — normal wins
    if (!existingDegraded) {
      return true;
    }
    // Both degraded
    if (incomingDegraded) {
      return existing.getDataVersion() >= incomingDataVersion;
    }
    // Existing degraded, incoming normal — normal overwrites
    return false;
  }

  /**
   * CacheSyncListener guard with a cache-level fast path: fetches the existing entry
   * from the L1 cache and delegates to the entry-level overload.
   *
   * <p>Used <em>outside</em> atomic {@code compute} blocks as a cheap first-pass
   * check (DCL pattern). A second guard inside the {@code compute} block is still
   * needed for correctness.
   *
   * @param cache               the local Caffeine L1 cache; must not be null
   * @param cacheKey            the cache key to look up; must not be null
   * @param incomingDataVersion the data version from the incoming sync message
   * @param incomingDegraded    {@code true} if the incoming sync message was sent in degraded mode
   * @return {@code true} if the incoming refresh should be skipped;
   *         {@code false} if the update may be needed
   */
  public static boolean shouldSkipForSync(
    Cache<String, Object> cache,
    String cacheKey,
    long incomingDataVersion,
    boolean incomingDegraded
  ) {
    Object existing = cache.getIfPresent(cacheKey);
    if (existing instanceof CacheEntry existingCacheEntry) {
      return shouldSkipForSync(existingCacheEntry, incomingDataVersion, incomingDegraded);
    }
    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;
17
18	import com.github.benmanes.caffeine.cache.Cache;
19	import io.github.hyshmily.hotkey.model.CacheEntry;
20	import java.util.Objects;
21
22	/**
23	* Shared version comparison logic for broadcast message guards used by both
24	* the Worker decision listener and the instance-to-instance cache sync listener.
25	*
26	* <p>This class provides two families of guard methods, each with a fast-path
27	* variant (accepting a {@link Cache} reference) and an entry-level variant
28	* (accepting an existing {@link CacheEntry}) for use inside atomic
29	* {@code compute} blocks:
30	*
31	* <ul>
32	* <li><b>{@link #shouldSkipForWorker}</b> — Compares {@code decisionVersion}
33	* from Worker HOT/COOL broadcasts. Degraded entries (created during a
34	* Redis outage) unconditionally accept incoming decisions, providing a
35	* safety net against Worker restarts that reset the {@code AtomicLong}
36	* counter (see ADR-0008, ADR-0009).</li>
37	* <li><b>{@link #shouldSkipForSync}</b> — Compares {@code dataVersion}
38	* from application-level data-mutation broadcasts. Uses a 4-case degraded
39	* comparison matrix:
40	* <ol>
41	* <li>Both normal: skip if existing {@code >=} incoming</li>
42	* <li>Existing normal, incoming degraded: always skip (normal wins)</li>
43	* <li>Both degraded: skip if existing {@code >=} incoming</li>
44	* <li>Existing degraded, incoming normal: never skip (normal overwrites degraded)</li>
45	* </ol>
46	* </li>
47	* </ul>
48	*
49	* <p>All methods are stateless and thread-safe. Instances of this utility class
50	* must never be created.
51	*
52	* @see WorkerListener
53	* @see CacheSyncListener
54	*/
55	public final class VersionGuard {
56
57	/**
58	* Utility class — prevent instantiation.
59	*/
60	private VersionGuard() {}
61
62	/**
63	* WorkerListener guard with epoch and node-id awareness.
64	* <p>Decision logic:
65	* <ol>
66	* <li>No existing entry → accept (return false)</li>
67	* <li>Existing entry is degraded → accept unconditionally (safety net)</li>
68	* <li>Incoming epoch > existing epoch → accept unconditionally
69	* (Worker restart detected — ADR-0010)</li>
70	* <li>Incoming epoch < existing epoch → skip (stale incarnation message)</li>
71	* <li>Same epoch, same nodeId → normal ordering (skip if existing dv >= incoming dv)</li>
72	* <li>Different nodeId, same epoch → cross-Worker ownership transfer;
73	* accept (return false) to allow new owner to assert authority</li>
74	* </ol>
75	*
76	* @param existing the existing cache entry; may be null
77	* @param incomingDecisionVersion the decision version from the incoming Worker message
78	* @param incomingNodeId the originating Worker's node ID
79	* @param incomingEpoch the originating Worker's epoch
80	* @return true if the incoming message should be skipped, false if it should be applied
81	*/
82	public static boolean shouldSkipForWorker(
83	CacheEntry existing,
84	long incomingDecisionVersion,
85	String incomingNodeId,
86	long incomingEpoch
87	) {
88	if (existing == null) {	2✔
89	return false;	2✔
90	}
91	if (existing.isVersionDegraded()) {	3✔
92	return false;	2✔
93	}
94
95	if (incomingEpoch > existing.getDecisionEpoch()) {	5✔
96	return false;	2✔
97	}
98	if (incomingEpoch < existing.getDecisionEpoch()) {	5!
NEW 99	return true;	×
100	}
101
102	if (Objects.equals(incomingNodeId, existing.getDecisionNodeId())) {	5✔
103	return existing.getDecisionVersion() >= incomingDecisionVersion;	9✔
104	}
105
106	return false;	2✔
107	}
108
109	/**
110	* WorkerListener guard for use inside an atomic {@code compute} block: the caller
111	* already holds the existing entry reference, so no redundant {@code getIfPresent}
112	* is needed.
113	*
114	* <p>Returns {@code false} (accept) for any of the following:
115	* <ul>
116	* <li>No existing entry ({@code null})</li>
117	* <li>Existing entry is degraded ({@code isVersionDegraded == true}) —
118	* yields to any incoming decision, even one with a lower version,
119	* because the degraded entry was written during a Redis outage</li>
120	* </ul>
121	*
122	* Otherwise, skips when the existing entry's {@code decisionVersion} is
123	* {@code >=} the incoming version.
124	*
125	* @param existing the existing cache entry; may be {@code null}
126	* @param incomingDecisionVersion the decision version from the incoming Worker message;
127	* must be non-negative in normal operation
128	* @return {@code true} if the incoming message should be skipped, {@code false}
129	* if the decision should be applied
130	*/
131	public static boolean shouldSkipForWorker(CacheEntry existing, long incomingDecisionVersion) {
132	if (existing == null) {	2✔
133	return false;	2✔
134	}
135	if (existing.isVersionDegraded()) {	3!
UNCOV 136	return false;	×
137	}
138	return existing.getDecisionVersion() >= incomingDecisionVersion;	8!
139	}
140
141	/**
142	* WorkerListener guard with a cache-level fast path using legacy 3-argument signature.
143	* Delegates to the 5-argument overload with {@code null} node ID and {@code 0} epoch.
144	*
145	* @param cache the local Caffeine L1 cache; must not be null
146	* @param cacheKey the cache key to look up; must not be null
147	* @param incomingDecisionVersion the decision version from the incoming Worker message
148	* @return {@code true} if the incoming message should be skipped (existing entry
149	* is already up-to-date); {@code false} if the decision may need to be applied
150	*/
151	public static boolean shouldSkipForWorker(
152	Cache<String, Object> cache,
153	String cacheKey,
154	long incomingDecisionVersion
155	) {
156	return shouldSkipForWorker(cache, cacheKey, incomingDecisionVersion, null, 0);	7✔
157	}
158
159	/**
160	* WorkerListener guard with a cache-level fast path: fetches the existing entry
161	* from the L1 cache and delegates to the entry-level overload.
162	*
163	* <p>This variant is used <em>outside</em> atomic {@code compute} blocks as a
164	* cheap first-pass check. If it returns {@code true} (skip), the caller can avoid
165	* the more expensive Redis fetch entirely. A second guard inside the {@code compute}
166	* block is still needed for correctness (DCL pattern).
167	*
168	* @param cache the local Caffeine L1 cache; must not be null
169	* @param cacheKey the cache key to look up; must not be null
170	* @param incomingDecisionVersion the decision version from the incoming Worker message
171	* @param incomingNodeId the originating Worker's node ID, may be null
172	* @param incomingEpoch the originating Worker's epoch
173	* @return {@code true} if the incoming message should be skipped (existing entry
174	* is already up-to-date); {@code false} if the decision may need to be applied
175	*/
176	public static boolean shouldSkipForWorker(
177	Cache<String, Object> cache,
178	String cacheKey,
179	long incomingDecisionVersion,
180	String incomingNodeId,
181	long incomingEpoch
182	) {
183	Object existing = cache.getIfPresent(cacheKey);	4✔
184	if (existing instanceof CacheEntry existingCacheEntry) {	6✔
185	return shouldSkipForWorker(existingCacheEntry, incomingDecisionVersion, incomingNodeId, incomingEpoch);	6✔
186	}
187	return false;	2✔
188	}
189
190	/**
191	* CacheSyncListener guard for use inside an atomic {@code compute} block: the caller
192	* already holds the existing entry reference.
193	*
194	* <p>Applies the 4-case degraded comparison matrix:
195	* <ol>
196	* <li>Both normal: skip if existing {@code >=} incoming</li>
197	* <li>Existing normal, incoming degraded: always skip (normal wins)</li>
198	* <li>Both degraded: skip if existing {@code >=} incoming</li>
199	* <li>Existing degraded, incoming normal: never skip (normal overwrites degraded)</li>
200	* </ol>
201	*
202	* <p>This design ensures that a single healthy Redis-backed instance can
203	* always overwrite degraded entries from other instances, while preventing
204	* degraded broadcasts from reverting healthy entries.
205	*
206	* @param existing the existing cache entry; may be {@code null} (returns {@code false})
207	* @param incomingDataVersion the data version from the incoming sync message
208	* @param incomingDegraded {@code true} if the incoming sync message was sent in degraded mode
209	* @return {@code true} if the incoming refresh should be skipped;
210	* {@code false} if it should be applied
211	*/
212	public static boolean shouldSkipForSync(CacheEntry existing, long incomingDataVersion, boolean incomingDegraded) {
213	if (existing == null) {	2✔
214	return false;	2✔
215	}
216
217	boolean existingDegraded = existing.isVersionDegraded();	3✔
218
219	// Both normal
220	if (!existingDegraded && !incomingDegraded) {	4✔
221	return existing.getDataVersion() >= incomingDataVersion;	9✔
222	}
223	// Existing normal, incoming degraded — normal wins
224	if (!existingDegraded) {	2✔
225	return true;	2✔
226	}
227	// Both degraded
228	if (incomingDegraded) {	2✔
229	return existing.getDataVersion() >= incomingDataVersion;	8!
230	}
231	// Existing degraded, incoming normal — normal overwrites
232	return false;	2✔
233	}
234
235	/**
236	* CacheSyncListener guard with a cache-level fast path: fetches the existing entry
237	* from the L1 cache and delegates to the entry-level overload.
238	*
239	* <p>Used <em>outside</em> atomic {@code compute} blocks as a cheap first-pass
240	* check (DCL pattern). A second guard inside the {@code compute} block is still
241	* needed for correctness.
242	*
243	* @param cache the local Caffeine L1 cache; must not be null
244	* @param cacheKey the cache key to look up; must not be null
245	* @param incomingDataVersion the data version from the incoming sync message
246	* @param incomingDegraded {@code true} if the incoming sync message was sent in degraded mode
247	* @return {@code true} if the incoming refresh should be skipped;
248	* {@code false} if the update may be needed
249	*/
250	public static boolean shouldSkipForSync(
251	Cache<String, Object> cache,
252	String cacheKey,
253	long incomingDataVersion,
254	boolean incomingDegraded
255	) {
256	Object existing = cache.getIfPresent(cacheKey);	4✔
257	if (existing instanceof CacheEntry existingCacheEntry) {	6✔
258	return shouldSkipForSync(existingCacheEntry, incomingDataVersion, incomingDegraded);	5✔
259	}
260	return false;	2✔
261	}
262	}

Hyshmily / hotkey / 27890525626

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