28637103998

Committed 03 Jul 2026 03:48AM UTC coverage: 90.399% (-0.2%) from 90.63%

Build # 28637103998

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix: add packaging to gitignore, fix flaky jitter test, update docs references

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

Coverage Stats

1264 of 1461 branches covered (86.52%)

Branch coverage included in aggregate %.

3604 of 3924 relevant lines covered (91.85%)

4.19 hits per line

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

90.32

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

import static io.github.hyshmily.hotkey.cache.cachesupport.CacheKeysPolicy.invalidCacheKey;
import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;

import io.github.hyshmily.hotkey.Internal;
import io.github.hyshmily.hotkey.constants.HotKeyConstants;
import io.github.hyshmily.hotkey.sync.worker.WorkerMessage;
import io.github.hyshmily.hotkey.util.version.VersionController;
import io.github.hyshmily.hotkey.util.version.VersionGuard;
import java.nio.charset.StandardCharsets;
import java.util.List;
import org.springframework.amqp.core.Message;

/**
 * Message between application instances for cache synchronization, carrying
 * invalidation or refresh operations for a single cache key (or a batch of keys).
 * Travels via the {@code hotkey.sync.exchange} FanoutExchange.
 *
 * <p>This is the data-plane counterpart to {@link WorkerMessage}: while
 * {@code WorkerMessage} carries hot/cool lifecycle decisions from the Worker cluster,
 * {@code SyncMessage} carries application-level data mutation events (writes, evictions)
 * between peer instances to keep L1 caches coherent.
 *
 * <p><b>Message types:</b>
 * <ul>
 *   <li>{@link #TYPE_INVALIDATE} — Remove a single key from the local cache</li>
 *   <li>{@link #TYPE_REFRESH} — Reload a single key from Redis into the local cache</li>
 *   <li>{@link #TYPE_INVALIDATE_ALL} — Batch-remove multiple keys (body is JSON array)</li>
 *   <li>{@link #TYPE_RULES_SYNC} — Synchronize the full rule set (body is rules JSON)</li>
 * </ul>
 *
 * <p>Each message carries a {@code dataVersion} and {@code isVersionDegraded} flag
 * to enable the 4-case degraded comparison in {@link VersionGuard#shouldSkipForSync}.
 *
 * @param cacheKey          the affected cache key (for single-key operations) or
 *                          the serialized payload (for batch / rules-sync types)
 * @param type              the operation type: {@link #TYPE_INVALIDATE},
 *                          {@link #TYPE_REFRESH}, {@link #TYPE_INVALIDATE_ALL},
 *                          or {@link #TYPE_RULES_SYNC}
 * @param version           the {@code dataVersion} at which the operation occurred;
 *                          see {@link VersionController#nextVersion}
 * @param isVersionDegraded whether the {@code dataVersion} was obtained in degraded mode
 *                          (node-local counter fallback, indicating Redis was unavailable)
 * @param rulesVersion      the rules version for {@code TYPE_RULES_SYNC} messages;
 *                          {@link HotKeyConstants#VERSION_DEFAULT} (0) for other types
 */
@Internal
public record SyncMessage(String cacheKey, String type, long version, boolean isVersionDegraded, long rulesVersion) {
  /** Invalidates a single cache key across all peer instances. */
  public static final String TYPE_INVALIDATE = "INVALIDATE";

  /** Refreshes a cache key from Redis across all peer instances. */
  public static final String TYPE_REFRESH = "REFRESH";

  /** Batch-invalidates multiple keys encoded as a JSON array in the message body. */
  public static final String TYPE_INVALIDATE_ALL = "INVALIDATE_ALL";

  /** Synchronizes the full rule set — receivers replace their local rules entirely. */
  public static final String TYPE_RULES_SYNC = "RULES_SYNC";

  /**
   * Message types whose body is NOT a single cache key but a payload
   * (JSON array of keys or full ruleset). The {@link #from(Message)}
   * deserializer skips the cache-key validity check for these types.
   */
  private static final List<String> BATCH_TYPES = List.of(TYPE_INVALIDATE_ALL, TYPE_RULES_SYNC);

  /**
   * Deserializes a {@code SyncMessage} from an AMQP message body and headers.
   *
   * <p>Deserialization rules:
   * <ul>
   *   <li>The cache key / payload is read from the message body (UTF-8 decoded).</li>
   *   <li>The type is read from the {@link HotKeyConstants#AMQP_HEADER_TYPE} header.</li>
   *   <li>The {@code dataVersion} is read from the {@code AMQP_HEADER_VERSION} header;
   *       defaults to {@link HotKeyConstants#VERSION_DEFAULT} (0) if missing or non-numeric.</li>
   *   <li>The degraded flag is read from the {@code AMQP_HEADER_IS_VERSION_DEGRADED} header;
   *       defaults to {@code false} if missing.</li>
   *   <li>For batch types ({@link #TYPE_INVALIDATE_ALL}, {@link #TYPE_RULES_SYNC}),
   *       the key validity check is skipped — the body carries a JSON payload rather
   *       than a single cache key.</li>
   * </ul>
   *
   * @param msg the incoming AMQP message; must not be null
   * @return a parsed {@link SyncMessage}, or {@code null} if the body is empty
   *         or the cache key is invalid for non-batch types
   */

  public static SyncMessage from(Message msg) {
    byte[] body = msg.getBody();
    if (body == null || body.length == 0) {
      return null;
    }

    String type = msg.getMessageProperties().getHeader(AMQP_HEADER_TYPE);
    String cacheKey = new String(body, StandardCharsets.UTF_8);
    if ((type == null || !BATCH_TYPES.contains(type)) && invalidCacheKey(cacheKey)) {
      return null;
    }

    long version =
      msg.getMessageProperties().getHeader(AMQP_HEADER_VERSION) instanceof Number n ? n.longValue() : VERSION_DEFAULT;
    boolean isVersionDegraded =
      msg.getMessageProperties().getHeader(AMQP_HEADER_IS_VERSION_DEGRADED) instanceof Boolean b ? b : false;
    long rulesVersion =
      msg.getMessageProperties().getHeader(AMQP_HEADER_RULES_VERSION) instanceof Number n2
        ? n2.longValue()
        : VERSION_DEFAULT;

    return new SyncMessage(cacheKey, type, version, isVersionDegraded, rulesVersion);
  }
}

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.local;
17
18	import static io.github.hyshmily.hotkey.cache.cachesupport.CacheKeysPolicy.invalidCacheKey;
19	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
20
21	import io.github.hyshmily.hotkey.Internal;
22	import io.github.hyshmily.hotkey.constants.HotKeyConstants;
23	import io.github.hyshmily.hotkey.sync.worker.WorkerMessage;
24	import io.github.hyshmily.hotkey.util.version.VersionController;
25	import io.github.hyshmily.hotkey.util.version.VersionGuard;
26	import java.nio.charset.StandardCharsets;
27	import java.util.List;
28	import org.springframework.amqp.core.Message;
29
30	/**
31	* Message between application instances for cache synchronization, carrying
32	* invalidation or refresh operations for a single cache key (or a batch of keys).
33	* Travels via the {@code hotkey.sync.exchange} FanoutExchange.
34	*
35	* <p>This is the data-plane counterpart to {@link WorkerMessage}: while
36	* {@code WorkerMessage} carries hot/cool lifecycle decisions from the Worker cluster,
37	* {@code SyncMessage} carries application-level data mutation events (writes, evictions)
38	* between peer instances to keep L1 caches coherent.
39	*
40	* <p><b>Message types:</b>
41	* <ul>
42	* <li>{@link #TYPE_INVALIDATE} — Remove a single key from the local cache</li>
43	* <li>{@link #TYPE_REFRESH} — Reload a single key from Redis into the local cache</li>
44	* <li>{@link #TYPE_INVALIDATE_ALL} — Batch-remove multiple keys (body is JSON array)</li>
45	* <li>{@link #TYPE_RULES_SYNC} — Synchronize the full rule set (body is rules JSON)</li>
46	* </ul>
47	*
48	* <p>Each message carries a {@code dataVersion} and {@code isVersionDegraded} flag
49	* to enable the 4-case degraded comparison in {@link VersionGuard#shouldSkipForSync}.
50	*
51	* @param cacheKey the affected cache key (for single-key operations) or
52	* the serialized payload (for batch / rules-sync types)
53	* @param type the operation type: {@link #TYPE_INVALIDATE},
54	* {@link #TYPE_REFRESH}, {@link #TYPE_INVALIDATE_ALL},
55	* or {@link #TYPE_RULES_SYNC}
56	* @param version the {@code dataVersion} at which the operation occurred;
57	* see {@link VersionController#nextVersion}
58	* @param isVersionDegraded whether the {@code dataVersion} was obtained in degraded mode
59	* (node-local counter fallback, indicating Redis was unavailable)
60	* @param rulesVersion the rules version for {@code TYPE_RULES_SYNC} messages;
61	* {@link HotKeyConstants#VERSION_DEFAULT} (0) for other types
62	*/
63	@Internal
64	public record SyncMessage(String cacheKey, String type, long version, boolean isVersionDegraded, long rulesVersion) {	18✔
65	/** Invalidates a single cache key across all peer instances. */
66	public static final String TYPE_INVALIDATE = "INVALIDATE";
67
68	/** Refreshes a cache key from Redis across all peer instances. */
69	public static final String TYPE_REFRESH = "REFRESH";
70
71	/** Batch-invalidates multiple keys encoded as a JSON array in the message body. */
72	public static final String TYPE_INVALIDATE_ALL = "INVALIDATE_ALL";
73
74	/** Synchronizes the full rule set — receivers replace their local rules entirely. */
75	public static final String TYPE_RULES_SYNC = "RULES_SYNC";
76
77	/**
78	* Message types whose body is NOT a single cache key but a payload
79	* (JSON array of keys or full ruleset). The {@link #from(Message)}
80	* deserializer skips the cache-key validity check for these types.
81	*/
82	private static final List<String> BATCH_TYPES = List.of(TYPE_INVALIDATE_ALL, TYPE_RULES_SYNC);	5✔
83
84	/**
85	* Deserializes a {@code SyncMessage} from an AMQP message body and headers.
86	*
87	* <p>Deserialization rules:
88	* <ul>
89	* <li>The cache key / payload is read from the message body (UTF-8 decoded).</li>
90	* <li>The type is read from the {@link HotKeyConstants#AMQP_HEADER_TYPE} header.</li>
91	* <li>The {@code dataVersion} is read from the {@code AMQP_HEADER_VERSION} header;
92	* defaults to {@link HotKeyConstants#VERSION_DEFAULT} (0) if missing or non-numeric.</li>
93	* <li>The degraded flag is read from the {@code AMQP_HEADER_IS_VERSION_DEGRADED} header;
94	* defaults to {@code false} if missing.</li>
95	* <li>For batch types ({@link #TYPE_INVALIDATE_ALL}, {@link #TYPE_RULES_SYNC}),
96	* the key validity check is skipped — the body carries a JSON payload rather
97	* than a single cache key.</li>
98	* </ul>
99	*
100	* @param msg the incoming AMQP message; must not be null
101	* @return a parsed {@link SyncMessage}, or {@code null} if the body is empty
102	* or the cache key is invalid for non-batch types
103	*/
104
105	public static SyncMessage from(Message msg) {
106	byte[] body = msg.getBody();	3✔
107	if (body == null \|\| body.length == 0) {	5!
108	return null;	2✔
109	}
110
111	String type = msg.getMessageProperties().getHeader(AMQP_HEADER_TYPE);	6✔
112	String cacheKey = new String(body, StandardCharsets.UTF_8);	6✔
113	if ((type == null \|\| !BATCH_TYPES.contains(type)) && invalidCacheKey(cacheKey)) {	9✔
114	return null;	2✔
115	}
116
117	long version =
118	msg.getMessageProperties().getHeader(AMQP_HEADER_VERSION) instanceof Number n ? n.longValue() : VERSION_DEFAULT;	16✔
119	boolean isVersionDegraded =
120	msg.getMessageProperties().getHeader(AMQP_HEADER_IS_VERSION_DEGRADED) instanceof Boolean b ? b : false;	16✔
121	long rulesVersion =
122	msg.getMessageProperties().getHeader(AMQP_HEADER_RULES_VERSION) instanceof Number n2	8!
123	? n2.longValue()	×
124	: VERSION_DEFAULT;	2✔
125
126	return new SyncMessage(cacheKey, type, version, isVersionDegraded, rulesVersion);	9✔
127	}
128	}

Hyshmily / hotkey / 28637103998

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