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

91.95

/common/src/main/java/io/github/hyshmily/hotkey/sync/local/CacheSyncPublisher.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 com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import io.github.hyshmily.hotkey.Internal;
import io.github.hyshmily.hotkey.sync.worker.WorkerListener;
import io.github.hyshmily.hotkey.util.version.VersionController;
import jakarta.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * Publishes cache synchronization messages (INVALIDATE / REFRESH / RULES_SYNC)
 * to all
 * peer application instances via the {@code hotkey.sync.exchange}
 * FanoutExchange.
 *
 * <p>
 * This is the outbound half of the instance-to-instance cache coherence
 * protocol.
 * The inbound half is {@link CacheSyncListener}. Together they ensure that a
 * data
 * mutation on one instance (write, evict, rule change) is propagated to all
 * other
 * instances within the same deployment.
 *
 * <p>
 * <b>Responsibility boundary:</b> This publisher handles
 * <em>application-level</em>
 * data changes only (e.g. {@code @CachePut}, {@code @CacheEvict},
 * {@code invalidateAllLocal}).
 * HOT/COOL lifecycle decisions from the Worker cluster are handled separately
 * by the
 * Worker's {@code WorkerBroadcaster} and received by {@link WorkerListener}.
 *
 * <p>
 * <b>Deduplication:</b> A {@link Caffeine} dedup cache keyed by
 * {@code "{type}:{cacheKey}"} prevents redundant broadcasts of the same
 * type+key
 * with a stale version within a configurable time window (see
 * {@link CacheSyncProperties#dedupWindowSeconds}).
 *
 * <p>
 * <b>Batch operations:</b> Large invalidations are split into chunks of at most
 * {@value #BATCH_SIZE} keys per AMQP message to stay within reasonable message
 * size limits.
 *
 * @see CacheSyncListener
 * @see SyncMessage
 */
@RequiredArgsConstructor
@Slf4j
@Internal
public class CacheSyncPublisher {

  /** RabbitMQ template for publishing messages to the sync FanoutExchange. */
  private final RabbitTemplate rabbitTemplate;

  /**
   * Configuration for sync exchange name, dedup window, and other sync settings.
   */
  private final CacheSyncProperties properties;

  /**
   * Dedup cache keyed by {@code "type:cacheKey"} → dataVersion. Prevents
   * redundant
   * broadcasts of the same type+key with a stale version within the configured
   * window.
   * Initialized in {@link #init()} via {@link PostConstruct}.
   */
  private Cache<String, Long> recentBroadcasts;

  /**
   * Shared Jackson {@link ObjectMapper} for serializing batch-invalidation key
   * lists to JSON.
   */
  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

  /**
   * Maximum number of keys per single batch-invalidation AMQP message ({@value}).
   * Batches larger than this are split into multiple messages.
   */
  private static final int BATCH_SIZE = 1000;

  /**
   * Initializes the deduplication cache after bean construction.
   * <p>
   * Creates a {@link Caffeine} cache with:
   * <ul>
   * <li>Expire-after-write of {@link CacheSyncProperties#dedupWindowSeconds}
   * seconds</li>
   * <li>Maximum size of {@link CacheSyncProperties#dedupMaxSize} entries</li>
   * </ul>
   * Called automatically by the Spring container after all dependencies are
   * injected.
   */
  @PostConstruct
  public void init() {
    this.recentBroadcasts = Caffeine.newBuilder()
      .expireAfterWrite(properties.getDedupWindowSeconds(), TimeUnit.SECONDS)
      .maximumSize(properties.getDedupMaxSize())
      .build();
  }

  /**
   * Returns the current estimated size of the deduplication cache.
   * <p>
   * This is a Caffeine {@code estimatedSizeOfKeysCount()} — it is an approximation and
   * should
   * not be relied upon for exact accounting.
   *
   * @return estimated number of entries in the dedup cache; {@code 0} if
   *         {@link #init()} has not been called yet
   */
  public long getDedupCacheSize() {
    return recentBroadcasts == null ? 0L : recentBroadcasts.estimatedSize();
  }

  /**
   * Broadcasts a REFRESH sync message to all peer instances, triggering them
   * to reload the value for the given key from Redis.
   *
   * <p>
   * Deduplicated: if a REFRESH for the same key with a higher or equal
   * {@code dataVersion} was broadcast within the dedup window, this call is
   * silently skipped.
   *
   * @param cacheKey the affected cache key; must not be null or empty
   * @param version  the {@code dataVersion} at which the operation occurred;
   *                 see {@link VersionController#nextVersion}
   * @param degraded {@code true} if the version was obtained from the local
   *                 fallback counter (Redis unavailable)
   */
  public void broadcastRefresh(String cacheKey, long version, boolean degraded) {
    sendDeduped(cacheKey, SyncMessage.TYPE_REFRESH, version, degraded);
  }

  /**
   * Broadcasts an INVALIDATE sync message to all peer instances, requesting them
   * to remove the specified key from their local cache.
   *
   * <p>
   * Deduplicated: if an INVALIDATE for the same key with a higher or equal
   * {@code dataVersion} was broadcast within the dedup window, this call is
   * silently skipped.
   *
   * @param cacheKey the affected cache key; must not be null or empty
   * @param version  the {@code dataVersion} at which the invalidation occurred
   * @param degraded {@code true} if the version was obtained from the local
   *                 fallback counter (Redis unavailable)
   */
  public void broadcastLocalInvalidate(String cacheKey, long version, boolean degraded) {
    sendDeduped(cacheKey, SyncMessage.TYPE_INVALIDATE, version, degraded);
  }

  /**
   * Batch-invalidates multiple keys in a single AMQP message (or multiple
   * messages
   * for very large batches).
   *
   * <p>
   * The body of each message is a JSON array of key strings. The receiver
   * ({@link CacheSyncListener}) calls {@code caffeineCache.invalidateAllLocal()}
   * once per batch, which is more efficient
   * than sending individual INVALIDATE messages for each key.
   *
   * <p>
   * Large collections are automatically split into chunks of at most
   * {@value #BATCH_SIZE} keys per AMQP message to avoid exceeding broker
   * message size limits. Serialization or send failures are logged at ERROR
   * level and do not propagate to the caller.
   *
   * <p>
   * Note: this method does <em>not</em> go through the version-based dedup
   * cache. All keys are unconditionally broadcast.
   *
   * @param keys the keys to invalidate; if null or empty the call is a silent
   *             no-op
   */
  public void broadcastLocalInvalidateAll(Collection<String> keys) {
    if (keys == null || keys.isEmpty()) {
      return;
    }

    List<String> keyList = new ArrayList<>(keys);
    for (int i = 0; i < keyList.size(); i += BATCH_SIZE) {
      int end = Math.min(i + BATCH_SIZE, keyList.size());
      List<String> batch = keyList.subList(i, end);

      try {
        String json = OBJECT_MAPPER.writeValueAsString(batch);
        MessageProperties props = new MessageProperties();
        props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_INVALIDATE_ALL);
        Message message = new Message(json.getBytes(StandardCharsets.UTF_8), props);

        rabbitTemplate.send(properties.getExchangeName(), "", message);
      } catch (AmqpException | JsonProcessingException e) {
        log.error("Failed to serialize batch invalidate keys", e);
      }
    }
  }

  /**
   * Broadcasts the full rule-set JSON to all peer instances for cross-instance
   * rule synchronization.
   *
   * <p>
   * Each receiver's {@link io.github.hyshmily.hotkey.rule.RuleMatcher#syncRules}
   * merges the incoming rules with the local set, guarded by the
   * {@code rulesVersion}
   * to prevent stale overwrites (newer versions win).
   *
   * <p>
   * If the payload is null or blank, the call is a silent no-op. AMQP publish
   * failures are logged at ERROR level and do not propagate.
   *
   * @param rulesJson    the serialized rule-set JSON (new format with version
   *                     wrapper);
   *                     must be a valid JSON string
   * @param rulesVersion the current rules version at the time of serialization;
   *                     receivers use this for conflict resolution
   */
  public void broadcastAllLocalRules(String rulesJson, long rulesVersion) {
    if (rulesJson == null || rulesJson.isBlank()) {
      return;
    }
    try {
      MessageProperties props = new MessageProperties();
      props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_RULES_SYNC);
      props.setHeader(AMQP_HEADER_RULES_VERSION, rulesVersion);
      Message message = new Message(rulesJson.getBytes(StandardCharsets.UTF_8), props);

      rabbitTemplate.send(properties.getExchangeName(), "", message);
    } catch (AmqpException e) {
      log.error("Failed to send rules sync message", e);
    }
  }

  /**
   * Core deduplicated send implementation: only publishes if no recent broadcast
   * of the same {@code type:cacheKey} composite with a greater or equal
   * {@code dataVersion} exists in the dedup window.
   *
   * <p>
   * <b>Dedup algorithm:</b> Uses {@code Caffeine.asMap().compute()} to atomically
   * check and update the dedup cache entry:
   * <ol>
   * <li>If an existing entry has {@code oldVersion >= newVersion}, the broadcast
   * is skipped (the stale flag is set).</li>
   * <li>Otherwise, the dedup cache is updated to the new version and the message
   * is published to the sync FanoutExchange.</li>
   * </ol>
   *
   * <p>
   * The dedup cache entry expires after
   * {@link CacheSyncProperties#getDedupWindowSeconds}
   * seconds, after which a subsequent call with any version will be accepted.
   *
   * @param cacheKey the affected cache key; must not be null or empty
   * @param type     the sync message type (e.g.
   *                 {@link SyncMessage#TYPE_INVALIDATE},
   *                 {@link SyncMessage#TYPE_REFRESH})
   * @param version  the {@code dataVersion} of the operation
   * @param degraded whether the version was obtained in degraded mode; passed
   *                 through
   *                 to the message headers; also prefixes the dedup compositeKey
   *                 with "D:"
   *                 to prevent degraded broadcasts from being blocked by normal
   *                 ones
   */
  private void sendDeduped(String cacheKey, String type, long version, boolean degraded) {
    if (invalidCacheKey(cacheKey) || invalidCacheKey(type)) {
      log.debug("Invalid cacheKey or type, skip sync: cacheKey={}, type={}", cacheKey, type);
      return;
    }
    String compositeKey = degraded ? "D:" + type + ":" + cacheKey : type + ":" + cacheKey;

    AtomicBoolean skipped = new AtomicBoolean(false);
    recentBroadcasts
      .asMap()
      .compute(compositeKey, (k, oldVersion) -> {
        if (oldVersion != null && oldVersion >= version) {
          log.debug(
            "Skip sync due to recent broadcast with same or newer version: compositeKey={}, oldVersion={}, newVersion={}",
            compositeKey,
            oldVersion,
            version
          );
          skipped.set(true);
          return oldVersion;
        }
        return version;
      });

    if (!skipped.get()) {
      try {
        MessageProperties props = new MessageProperties();

        props.setHeader(AMQP_HEADER_TYPE, type);
        props.setHeader(AMQP_HEADER_VERSION, version);
        props.setHeader(AMQP_HEADER_IS_VERSION_DEGRADED, degraded);

        Message message = new Message(cacheKey.getBytes(StandardCharsets.UTF_8), props);

        rabbitTemplate.send(properties.getExchangeName(), "", message);
      } catch (AmqpException e) {
        log.error("Failed to send sync message", e);
      }
    }
  }
}

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 com.fasterxml.jackson.core.JsonProcessingException;
22	import com.fasterxml.jackson.databind.ObjectMapper;
23	import com.github.benmanes.caffeine.cache.Cache;
24	import com.github.benmanes.caffeine.cache.Caffeine;
25	import io.github.hyshmily.hotkey.Internal;
26	import io.github.hyshmily.hotkey.sync.worker.WorkerListener;
27	import io.github.hyshmily.hotkey.util.version.VersionController;
28	import jakarta.annotation.PostConstruct;
29	import java.nio.charset.StandardCharsets;
30	import java.util.ArrayList;
31	import java.util.Collection;
32	import java.util.List;
33	import java.util.concurrent.TimeUnit;
34	import java.util.concurrent.atomic.AtomicBoolean;
35	import lombok.RequiredArgsConstructor;
36	import lombok.extern.slf4j.Slf4j;
37	import org.springframework.amqp.AmqpException;
38	import org.springframework.amqp.core.Message;
39	import org.springframework.amqp.core.MessageProperties;
40	import org.springframework.amqp.rabbit.core.RabbitTemplate;
41
42	/**
43	* Publishes cache synchronization messages (INVALIDATE / REFRESH / RULES_SYNC)
44	* to all
45	* peer application instances via the {@code hotkey.sync.exchange}
46	* FanoutExchange.
47	*
48	* <p>
49	* This is the outbound half of the instance-to-instance cache coherence
50	* protocol.
51	* The inbound half is {@link CacheSyncListener}. Together they ensure that a
52	* data
53	* mutation on one instance (write, evict, rule change) is propagated to all
54	* other
55	* instances within the same deployment.
56	*
57	* <p>
58	* <b>Responsibility boundary:</b> This publisher handles
59	* <em>application-level</em>
60	* data changes only (e.g. {@code @CachePut}, {@code @CacheEvict},
61	* {@code invalidateAllLocal}).
62	* HOT/COOL lifecycle decisions from the Worker cluster are handled separately
63	* by the
64	* Worker's {@code WorkerBroadcaster} and received by {@link WorkerListener}.
65	*
66	* <p>
67	* <b>Deduplication:</b> A {@link Caffeine} dedup cache keyed by
68	* {@code "{type}:{cacheKey}"} prevents redundant broadcasts of the same
69	* type+key
70	* with a stale version within a configurable time window (see
71	* {@link CacheSyncProperties#dedupWindowSeconds}).
72	*
73	* <p>
74	* <b>Batch operations:</b> Large invalidations are split into chunks of at most
75	* {@value #BATCH_SIZE} keys per AMQP message to stay within reasonable message
76	* size limits.
77	*
78	* @see CacheSyncListener
79	* @see SyncMessage
80	*/
81	@RequiredArgsConstructor
82	@Slf4j	3✔
83	@Internal
84	public class CacheSyncPublisher {
85
86	/** RabbitMQ template for publishing messages to the sync FanoutExchange. */
87	private final RabbitTemplate rabbitTemplate;
88
89	/**
90	* Configuration for sync exchange name, dedup window, and other sync settings.
91	*/
92	private final CacheSyncProperties properties;
93
94	/**
95	* Dedup cache keyed by {@code "type:cacheKey"} → dataVersion. Prevents
96	* redundant
97	* broadcasts of the same type+key with a stale version within the configured
98	* window.
99	* Initialized in {@link #init()} via {@link PostConstruct}.
100	*/
101	private Cache<String, Long> recentBroadcasts;
102
103	/**
104	* Shared Jackson {@link ObjectMapper} for serializing batch-invalidation key
105	* lists to JSON.
106	*/
107	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();	5✔
108
109	/**
110	* Maximum number of keys per single batch-invalidation AMQP message ({@value}).
111	* Batches larger than this are split into multiple messages.
112	*/
113	private static final int BATCH_SIZE = 1000;
114
115	/**
116	* Initializes the deduplication cache after bean construction.
117	* <p>
118	* Creates a {@link Caffeine} cache with:
119	* <ul>
120	* <li>Expire-after-write of {@link CacheSyncProperties#dedupWindowSeconds}
121	* seconds</li>
122	* <li>Maximum size of {@link CacheSyncProperties#dedupMaxSize} entries</li>
123	* </ul>
124	* Called automatically by the Spring container after all dependencies are
125	* injected.
126	*/
127	@PostConstruct
128	public void init() {
129	this.recentBroadcasts = Caffeine.newBuilder()	4✔
130	.expireAfterWrite(properties.getDedupWindowSeconds(), TimeUnit.SECONDS)	6✔
131	.maximumSize(properties.getDedupMaxSize())	3✔
132	.build();	2✔
133	}	1✔
134
135	/**
136	* Returns the current estimated size of the deduplication cache.
137	* <p>
138	* This is a Caffeine {@code estimatedSizeOfKeysCount()} — it is an approximation and
139	* should
140	* not be relied upon for exact accounting.
141	*
142	* @return estimated number of entries in the dedup cache; {@code 0} if
143	* {@link #init()} has not been called yet
144	*/
145	public long getDedupCacheSize() {
146	return recentBroadcasts == null ? 0L : recentBroadcasts.estimatedSize();	9✔
147	}
148
149	/**
150	* Broadcasts a REFRESH sync message to all peer instances, triggering them
151	* to reload the value for the given key from Redis.
152	*
153	* <p>
154	* Deduplicated: if a REFRESH for the same key with a higher or equal
155	* {@code dataVersion} was broadcast within the dedup window, this call is
156	* silently skipped.
157	*
158	* @param cacheKey the affected cache key; must not be null or empty
159	* @param version the {@code dataVersion} at which the operation occurred;
160	* see {@link VersionController#nextVersion}
161	* @param degraded {@code true} if the version was obtained from the local
162	* fallback counter (Redis unavailable)
163	*/
164	public void broadcastRefresh(String cacheKey, long version, boolean degraded) {
165	sendDeduped(cacheKey, SyncMessage.TYPE_REFRESH, version, degraded);	6✔
166	}	1✔
167
168	/**
169	* Broadcasts an INVALIDATE sync message to all peer instances, requesting them
170	* to remove the specified key from their local cache.
171	*
172	* <p>
173	* Deduplicated: if an INVALIDATE for the same key with a higher or equal
174	* {@code dataVersion} was broadcast within the dedup window, this call is
175	* silently skipped.
176	*
177	* @param cacheKey the affected cache key; must not be null or empty
178	* @param version the {@code dataVersion} at which the invalidation occurred
179	* @param degraded {@code true} if the version was obtained from the local
180	* fallback counter (Redis unavailable)
181	*/
182	public void broadcastLocalInvalidate(String cacheKey, long version, boolean degraded) {
183	sendDeduped(cacheKey, SyncMessage.TYPE_INVALIDATE, version, degraded);	6✔
184	}	1✔
185
186	/**
187	* Batch-invalidates multiple keys in a single AMQP message (or multiple
188	* messages
189	* for very large batches).
190	*
191	* <p>
192	* The body of each message is a JSON array of key strings. The receiver
193	* ({@link CacheSyncListener}) calls {@code caffeineCache.invalidateAllLocal()}
194	* once per batch, which is more efficient
195	* than sending individual INVALIDATE messages for each key.
196	*
197	* <p>
198	* Large collections are automatically split into chunks of at most
199	* {@value #BATCH_SIZE} keys per AMQP message to avoid exceeding broker
200	* message size limits. Serialization or send failures are logged at ERROR
201	* level and do not propagate to the caller.
202	*
203	* <p>
204	* Note: this method does <em>not</em> go through the version-based dedup
205	* cache. All keys are unconditionally broadcast.
206	*
207	* @param keys the keys to invalidate; if null or empty the call is a silent
208	* no-op
209	*/
210	public void broadcastLocalInvalidateAll(Collection<String> keys) {
211	if (keys == null \|\| keys.isEmpty()) {	5✔
212	return;	1✔
213	}
214
215	List<String> keyList = new ArrayList<>(keys);	5✔
216	for (int i = 0; i < keyList.size(); i += BATCH_SIZE) {	8✔
217	int end = Math.min(i + BATCH_SIZE, keyList.size());	7✔
218	List<String> batch = keyList.subList(i, end);	5✔
219
220	try {
221	String json = OBJECT_MAPPER.writeValueAsString(batch);	4✔
222	MessageProperties props = new MessageProperties();	4✔
223	props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_INVALIDATE_ALL);	4✔
224	Message message = new Message(json.getBytes(StandardCharsets.UTF_8), props);	8✔
225
226	rabbitTemplate.send(properties.getExchangeName(), "", message);	8✔
227	} catch (AmqpException \| JsonProcessingException e) {	×
228	log.error("Failed to serialize batch invalidate keys", e);	×
229	}	1✔
230	}
231	}	1✔
232
233	/**
234	* Broadcasts the full rule-set JSON to all peer instances for cross-instance
235	* rule synchronization.
236	*
237	* <p>
238	* Each receiver's {@link io.github.hyshmily.hotkey.rule.RuleMatcher#syncRules}
239	* merges the incoming rules with the local set, guarded by the
240	* {@code rulesVersion}
241	* to prevent stale overwrites (newer versions win).
242	*
243	* <p>
244	* If the payload is null or blank, the call is a silent no-op. AMQP publish
245	* failures are logged at ERROR level and do not propagate.
246	*
247	* @param rulesJson the serialized rule-set JSON (new format with version
248	* wrapper);
249	* must be a valid JSON string
250	* @param rulesVersion the current rules version at the time of serialization;
251	* receivers use this for conflict resolution
252	*/
253	public void broadcastAllLocalRules(String rulesJson, long rulesVersion) {
254	if (rulesJson == null \|\| rulesJson.isBlank()) {	5✔
255	return;	1✔
256	}
257	try {
258	MessageProperties props = new MessageProperties();	4✔
259	props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_RULES_SYNC);	4✔
260	props.setHeader(AMQP_HEADER_RULES_VERSION, rulesVersion);	5✔
261	Message message = new Message(rulesJson.getBytes(StandardCharsets.UTF_8), props);	8✔
262
263	rabbitTemplate.send(properties.getExchangeName(), "", message);	8✔
264	} catch (AmqpException e) {	×
265	log.error("Failed to send rules sync message", e);	×
266	}	1✔
267	}	1✔
268
269	/**
270	* Core deduplicated send implementation: only publishes if no recent broadcast
271	* of the same {@code type:cacheKey} composite with a greater or equal
272	* {@code dataVersion} exists in the dedup window.
273	*
274	* <p>
275	* <b>Dedup algorithm:</b> Uses {@code Caffeine.asMap().compute()} to atomically
276	* check and update the dedup cache entry:
277	* <ol>
278	* <li>If an existing entry has {@code oldVersion >= newVersion}, the broadcast
279	* is skipped (the stale flag is set).</li>
280	* <li>Otherwise, the dedup cache is updated to the new version and the message
281	* is published to the sync FanoutExchange.</li>
282	* </ol>
283	*
284	* <p>
285	* The dedup cache entry expires after
286	* {@link CacheSyncProperties#getDedupWindowSeconds}
287	* seconds, after which a subsequent call with any version will be accepted.
288	*
289	* @param cacheKey the affected cache key; must not be null or empty
290	* @param type the sync message type (e.g.
291	* {@link SyncMessage#TYPE_INVALIDATE},
292	* {@link SyncMessage#TYPE_REFRESH})
293	* @param version the {@code dataVersion} of the operation
294	* @param degraded whether the version was obtained in degraded mode; passed
295	* through
296	* to the message headers; also prefixes the dedup compositeKey
297	* with "D:"
298	* to prevent degraded broadcasts from being blocked by normal
299	* ones
300	*/
301	private void sendDeduped(String cacheKey, String type, long version, boolean degraded) {
302	if (invalidCacheKey(cacheKey) \|\| invalidCacheKey(type)) {	6!
303	log.debug("Invalid cacheKey or type, skip sync: cacheKey={}, type={}", cacheKey, type);	5✔
304	return;	1✔
305	}
306	String compositeKey = degraded ? "D:" + type + ":" + cacheKey : type + ":" + cacheKey;	10✔
307
308	AtomicBoolean skipped = new AtomicBoolean(false);	5✔
309	recentBroadcasts	2✔
310	.asMap()	6✔
311	.compute(compositeKey, (k, oldVersion) -> {	2✔
312	if (oldVersion != null && oldVersion >= version) {	7✔
313	log.debug(	16✔
314	"Skip sync due to recent broadcast with same or newer version: compositeKey={}, oldVersion={}, newVersion={}",
315	compositeKey,
316	oldVersion,
317	version	2✔
318	);
319	skipped.set(true);	3✔
320	return oldVersion;	2✔
321	}
322	return version;	3✔
323	});
324
325	if (!skipped.get()) {	3✔
326	try {
327	MessageProperties props = new MessageProperties();	4✔
328
329	props.setHeader(AMQP_HEADER_TYPE, type);	4✔
330	props.setHeader(AMQP_HEADER_VERSION, version);	5✔
331	props.setHeader(AMQP_HEADER_IS_VERSION_DEGRADED, degraded);	5✔
332
333	Message message = new Message(cacheKey.getBytes(StandardCharsets.UTF_8), props);	8✔
334
335	rabbitTemplate.send(properties.getExchangeName(), "", message);	8✔
336	} catch (AmqpException e) {	×
337	log.error("Failed to send sync message", e);	×
338	}	1✔
339	}
340	}	1✔
341	}

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