This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 61c9632f39 perf(plugin-base): batch namespace snapshot cache updates
and invalidation (#7645) (#7672)
61c9632f39 is described below
commit 61c9632f39137776bb59cbfcbdc0d2e9008f0e23
Author: Daming <[email protected]>
AuthorDate: Tue Oct 20 12:39:29 2026 +0800
perf(plugin-base): batch namespace snapshot cache updates and invalidation
(#7645) (#7672)
* perf(plugin-base): batch namespace snapshot cache updates and
invalidation (#7645)
applySelectorDataSnapshot/applyRuleDataSnapshot previously called the
per-row subscribe path for every incoming and stale entry, which ran one
list merge + sort per row and invalidated the match cache once per row.
On cold initialization or reconnect snapshots with thousands of rows, this
dominated sync-apply CPU and allocation.
Changes:
- BaseDataCache: add package-private removeSelectDataBatch and
removeRuleDataBatch that group deletions by plugin/selector bucket and
run a single computeIfPresent merge per bucket.
- CommonPluginDataSubscriber: applySelectorDataSnapshot now calls
refreshSelectorData(dataList) once and removeSelectDataBatch(stale) once;
applyRuleDataSnapshot mirrors that with refreshRuleData and
removeRuleDataBatch. Per-row plugin handler callbacks (handlerSelector,
handlerRule, removeSelector, removeRule) are preserved and wrapped in a
runGuarded helper that keeps the initial-sync abort semantics of the
previous subscribeDataHandler path.
- Match-cache invalidation is coalesced per affected pluginName, using the
broader removeSelectorData(pluginName)/removeRuleData(pluginName)
variants. Positive and negative entries for the affected plugin are
cleared together, so no stale match survives; other plugins' caches are
untouched.
Namespace isolation, ordering by sort key, empty-snapshot cleanup and
plugin-handler contracts are unchanged.
Tests:
- selectorSnapshotBatchesMergesAndPreservesSortOrder
- ruleSnapshotBatchesMergesAndPreservesSortOrder
- selectorSnapshotInvokesRemoveSelectorForEachStaleRow
- ruleSnapshotInvokesRemoveRuleForEachStaleRow
- snapshotCoalescesMatchCacheInvalidationPerAffectedPlugin
- selectorSnapshotHandlerFailurePropagatesDuringInitialSync
- selectorSnapshotHandlerFailureIsLoggedOutsideInitialSync
Ref: #7650
* fix(plugin-base): harden namespace snapshot against dangling and
malformed rows (#7645)
Address review findings on the batched snapshot apply:
- Blocking 1: dangling selectors/rules with a blank pluginName no longer
lost
their match-cache invalidation. collectAffectedPlugins drops blank names,
so add
invalidateDanglingSelectorMatchCache / invalidateDanglingRuleMatchCache
that
reproduce master's id-based cross-bucket sweep via the blank-name branch
of
MatchDataCache.
- Blocking 2: the batch cache write (refreshSelectorData/refreshRuleData)
is no
longer unguarded. A row with a null pluginName (selector) or null
selectorId
(rule) produced a null grouping key whose compute on a concurrent map
threw and
aborted the whole snapshot, causing a websocket resync loop. Malformed
rows are
now skipped up front with a warning so valid rows still converge;
collectAffectedPlugins
also tolerates null elements.
- Match-cache invalidation moved into a finally block so cleanup still runs
when a
handler aborts during initial sync (stale rows are already removed from
BaseDataCache and cannot be rediscovered on retry).
Adds tests for the dangling selector/rule sweep and malformed-row isolation.
---------
Co-authored-by: Luke Haochao Zhuang <[email protected]>
---
.../shenyu/plugin/base/cache/BaseDataCache.java | 64 +++++++
.../base/cache/CommonPluginDataSubscriber.java | 203 ++++++++++++++++++++-
.../base/cache/CommonPluginDataSubscriberTest.java | 197 ++++++++++++++++++++
3 files changed, 456 insertions(+), 8 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
index fed4420289..fc67d1143a 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
@@ -25,6 +25,8 @@ import org.apache.shenyu.common.dto.SelectorData;
import java.util.ArrayList;
import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -179,6 +181,44 @@ public final class BaseDataCache {
public void cleanSelectorDataSelf(final List<SelectorData>
selectorDataList) {
selectorDataList.forEach(this::removeSelectData);
}
+
+ /**
+ * Remove a batch of selectors with a single merge per affected plugin
bucket.
+ * Selectors without a plugin name are matched against every bucket,
mirroring
+ * {@link #removeSelectData(SelectorData)}.
+ *
+ * @param selectorDataList the selectors to remove
+ */
+ void removeSelectDataBatch(final List<SelectorData> selectorDataList) {
+ if (selectorDataList.isEmpty()) {
+ return;
+ }
+ final Map<String, Set<String>> idsByPlugin = new HashMap<>();
+ final Set<String> danglingIds = new HashSet<>();
+ for (SelectorData data : selectorDataList) {
+ if (Objects.isNull(data) || Objects.isNull(data.getId())) {
+ continue;
+ }
+ if (StringUtils.isBlank(data.getPluginName())) {
+ danglingIds.add(data.getId());
+ } else {
+ idsByPlugin.computeIfAbsent(data.getPluginName(), key -> new
HashSet<>()).add(data.getId());
+ }
+ }
+ if (!danglingIds.isEmpty()) {
+ selectorMap.keySet().forEach(pluginName ->
removeSelectDataByIds(pluginName, danglingIds));
+ }
+ idsByPlugin.forEach(this::removeSelectDataByIds);
+ }
+
+ private void removeSelectDataByIds(final String pluginName, final
Set<String> selectorIds) {
+ selectorMap.computeIfPresent(pluginName, (key, value) -> {
+ final List<SelectorData> result = value.stream()
+ .filter(selector ->
!selectorIds.contains(selector.getId()))
+ .collect(Collectors.toList());
+ return result.isEmpty() ? null : List.copyOf(result);
+ });
+ }
/**
* Obtain selector data list list.
@@ -239,6 +279,30 @@ public final class BaseDataCache {
public void cleanRuleDataSelf(final List<RuleData> ruleDataList) {
ruleDataList.forEach(this::removeRuleData);
}
+
+ /**
+ * Remove a batch of rules with a single merge per affected selector
bucket.
+ *
+ * @param ruleDataList the rules to remove
+ */
+ void removeRuleDataBatch(final List<RuleData> ruleDataList) {
+ if (ruleDataList.isEmpty()) {
+ return;
+ }
+ final Map<String, Set<String>> idsBySelector = new HashMap<>();
+ for (RuleData data : ruleDataList) {
+ if (Objects.isNull(data) || Objects.isNull(data.getId()) ||
Objects.isNull(data.getSelectorId())) {
+ continue;
+ }
+ idsBySelector.computeIfAbsent(data.getSelectorId(), key -> new
HashSet<>()).add(data.getId());
+ }
+ idsBySelector.forEach((selectorId, ruleIds) ->
ruleMap.computeIfPresent(selectorId, (key, value) -> {
+ final List<RuleData> result = value.stream()
+ .filter(rule -> !ruleIds.contains(rule.getId()))
+ .collect(Collectors.toList());
+ return result.isEmpty() ? null : List.copyOf(result);
+ }));
+ }
/**
* Obtain rule data list list.
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
index c72c39e6e0..9c171358b4 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
@@ -36,12 +36,16 @@ import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.lang.NonNull;
+import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Function;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* The type Common plugin data subscriber.
@@ -160,26 +164,100 @@ public class CommonPluginDataSubscriber implements
PluginDataSubscriber {
@Override
public void applySelectorDataSnapshot(final String namespaceId, final
List<SelectorData> dataList) {
- Set<String> selectorIds =
dataList.stream().map(SelectorData::getId).collect(Collectors.toSet());
- List<SelectorData> stale =
BaseDataCache.getInstance().getSelectorMap().values().stream()
+ final BaseDataCache cache = BaseDataCache.getInstance();
+ // A row without a plugin name yields a null grouping key in
refreshSelectorData, whose compute on a
+ // concurrent map throws and would abort the whole snapshot (and, over
a websocket, loop resyncs).
+ // The previous per-row subscribe path isolated such a row; skip it up
front so valid rows still converge.
+ final List<SelectorData> incoming = new ArrayList<>(dataList.size());
+ for (SelectorData data : dataList) {
+ if (Objects.nonNull(data) &&
Objects.nonNull(data.getPluginName())) {
+ incoming.add(data);
+ } else {
+ LOG.warn("skip malformed selector row without plugin name in
snapshot for namespace [{}]", namespaceId);
+ }
+ }
+ final Set<String> selectorIds =
incoming.stream().map(SelectorData::getId).collect(Collectors.toSet());
+ final List<SelectorData> stale =
cache.getSelectorMap().values().stream()
.flatMap(List::stream)
.filter(data -> namespaceId.equals(data.getNamespaceId()))
.filter(data -> !selectorIds.contains(data.getId()))
.collect(Collectors.toList());
- dataList.forEach(this::onSelectorSubscribe);
- stale.forEach(this::unSelectorSubscribe);
+ // Merge and sort once per affected plugin bucket instead of per
selector row.
+ if (!incoming.isEmpty()) {
+ cache.refreshSelectorData(incoming);
+ }
+ if (!stale.isEmpty()) {
+ cache.removeSelectDataBatch(stale);
+ }
+ try {
+ // Preserve per-item plugin handler callbacks and the initial-sync
failure semantics
+ // that the previous per-row subscribe path provided.
+ incoming.forEach(data -> runGuarded(data, () -> {
+ if (StringUtils.isNotBlank(data.getPluginName())) {
+ Optional.ofNullable(handlerMap.get(data.getPluginName()))
+ .ifPresent(handler ->
handler.handlerSelector(data));
+ }
+ }));
+ stale.forEach(data -> runGuarded(data, () -> {
+ // the concurrent handler map rejects null keys, and a
dangling selector carries no plugin name
+ if (StringUtils.isNotBlank(data.getPluginName())) {
+ Optional.ofNullable(handlerMap.get(data.getPluginName()))
+ .ifPresent(handler ->
handler.removeSelector(data));
+ }
+ }));
+ } finally {
+ // Invalidate even if a handler aborted during initial sync: the
stale rows are already gone from
+ // BaseDataCache above, so a retry cannot rediscover them and
their match cache would stay stale.
+
invalidateSelectorMatchCacheByPlugins(collectAffectedPlugins(incoming, stale,
SelectorData::getPluginName));
+ invalidateDanglingSelectorMatchCache(incoming, stale);
+ }
}
@Override
public void applyRuleDataSnapshot(final String namespaceId, final
List<RuleData> dataList) {
- Set<String> ruleIds =
dataList.stream().map(RuleData::getId).collect(Collectors.toSet());
- List<RuleData> stale =
BaseDataCache.getInstance().getRuleMap().values().stream()
+ final BaseDataCache cache = BaseDataCache.getInstance();
+ // A row without a selector id yields a null grouping key in
refreshRuleData, whose compute on a
+ // concurrent map throws and would abort the whole snapshot; skip it
up front so valid rows still converge.
+ final List<RuleData> incoming = new ArrayList<>(dataList.size());
+ for (RuleData data : dataList) {
+ if (Objects.nonNull(data) &&
Objects.nonNull(data.getSelectorId())) {
+ incoming.add(data);
+ } else {
+ LOG.warn("skip malformed rule row without selector id in
snapshot for namespace [{}]", namespaceId);
+ }
+ }
+ final Set<String> ruleIds =
incoming.stream().map(RuleData::getId).collect(Collectors.toSet());
+ final List<RuleData> stale = cache.getRuleMap().values().stream()
.flatMap(List::stream)
.filter(data -> namespaceId.equals(data.getNamespaceId()))
.filter(data -> !ruleIds.contains(data.getId()))
.collect(Collectors.toList());
- dataList.forEach(this::onRuleSubscribe);
- stale.forEach(this::unRuleSubscribe);
+ // Merge and sort once per affected selector bucket instead of per
rule row.
+ if (!incoming.isEmpty()) {
+ cache.refreshRuleData(incoming);
+ }
+ if (!stale.isEmpty()) {
+ cache.removeRuleDataBatch(stale);
+ }
+ try {
+ incoming.forEach(data -> runGuarded(data, () -> {
+ if (StringUtils.isNotBlank(data.getPluginName())) {
+ Optional.ofNullable(handlerMap.get(data.getPluginName()))
+ .ifPresent(handler -> handler.handlerRule(data));
+ }
+ }));
+ stale.forEach(data -> runGuarded(data, () -> {
+ if (StringUtils.isNotBlank(data.getPluginName())) {
+ Optional.ofNullable(handlerMap.get(data.getPluginName()))
+ .ifPresent(handler -> handler.removeRule(data));
+ }
+ }));
+ } finally {
+ // Invalidate even if a handler aborted during initial sync, so a
retry is not left with stale
+ // match-cache entries for rows already removed from BaseDataCache
above.
+ invalidateRuleMatchCacheByPlugins(collectAffectedPlugins(incoming,
stale, RuleData::getPluginName));
+ invalidateDanglingRuleMatchCache(incoming, stale);
+ }
}
@Override
@@ -402,6 +480,115 @@ public class CommonPluginDataSubscriber implements
PluginDataSubscriber {
}
}
+ /**
+ * Run a per-row callback for a snapshot apply, preserving the exception
semantics of
+ * {@link #subscribeDataHandler}: during initial sync failures abort the
snapshot, otherwise
+ * they are logged so subsequent rows still converge.
+ */
+ private <T> void runGuarded(final T data, final Runnable action) {
+ try {
+ action.run();
+ } catch (Exception e) {
+ if (InitialSyncApplication.isActive()) {
+ throw new IllegalStateException("Initial configuration
application failed", e);
+ }
+ LOG.error("snapshot handler error, classData: {}",
JsonUtils.toJson(data), e);
+ }
+ }
+
+ private <T> Set<String> collectAffectedPlugins(final List<T> incoming,
final List<T> stale,
+ final Function<T, String>
pluginNameOf) {
+ final Set<String> plugins = new HashSet<>();
+ incoming.forEach(data -> {
+ if (Objects.isNull(data)) {
+ return;
+ }
+ final String name = pluginNameOf.apply(data);
+ if (StringUtils.isNotBlank(name)) {
+ plugins.add(name);
+ }
+ });
+ stale.forEach(data -> {
+ if (Objects.isNull(data)) {
+ return;
+ }
+ final String name = pluginNameOf.apply(data);
+ if (StringUtils.isNotBlank(name)) {
+ plugins.add(name);
+ }
+ });
+ return plugins;
+ }
+
+ /**
+ * Mirror master's id-based cross-bucket sweep for dangling selectors that
carry no plugin name:
+ * {@link #collectAffectedPlugins} drops blank names, so without this a
deleted dangling selector would
+ * leave its cached matches (which may live under any plugin bucket)
behind. Delegates to the blank-name
+ * branch of {@link MatchDataCache#removeSelectorData(String, String)} and
+ * {@link MatchDataCache#removeRuleDataBySelector(String, String)}.
+ */
+ private void invalidateDanglingSelectorMatchCache(final List<SelectorData>
incoming, final List<SelectorData> stale) {
+ if (!selectorMatchConfig.getCache().getEnabled() &&
!ruleMatchCacheConfig.getCache().getEnabled()) {
+ return;
+ }
+ final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+ Stream.concat(incoming.stream(), stale.stream())
+ .filter(Objects::nonNull)
+ .filter(data -> StringUtils.isBlank(data.getPluginName()))
+ .map(SelectorData::getId)
+ .filter(Objects::nonNull)
+ .forEach(id -> {
+ if (selectorMatchConfig.getCache().getEnabled()) {
+ matchDataCache.removeSelectorData(null, id);
+ }
+ if (ruleMatchCacheConfig.getCache().getEnabled()) {
+ matchDataCache.removeRuleDataBySelector(null, id);
+ }
+ });
+ }
+
+ /**
+ * Rule counterpart of {@link #invalidateDanglingSelectorMatchCache}: a
stale rule keyed by selectorId can
+ * carry a blank plugin name, so its cached matches are swept across every
bucket by rule id via the
+ * blank-name branch of {@link MatchDataCache#removeRuleData(String,
String)}.
+ */
+ private void invalidateDanglingRuleMatchCache(final List<RuleData>
incoming, final List<RuleData> stale) {
+ if (!ruleMatchCacheConfig.getCache().getEnabled()) {
+ return;
+ }
+ final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+ Stream.concat(incoming.stream(), stale.stream())
+ .filter(Objects::nonNull)
+ .filter(data -> StringUtils.isBlank(data.getPluginName()))
+ .map(RuleData::getId)
+ .filter(Objects::nonNull)
+ .forEach(id -> matchDataCache.removeRuleData(null, id));
+ }
+
+ private void invalidateSelectorMatchCacheByPlugins(final Set<String>
pluginNames) {
+ if (pluginNames.isEmpty()) {
+ return;
+ }
+ final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+ for (String pluginName : pluginNames) {
+ if (selectorMatchConfig.getCache().getEnabled()) {
+ matchDataCache.removeSelectorData(pluginName);
+ }
+ if (ruleMatchCacheConfig.getCache().getEnabled()) {
+ // selector membership changed, so any cached rule match under
this plugin may be stale
+ matchDataCache.removeRuleData(pluginName);
+ }
+ }
+ }
+
+ private void invalidateRuleMatchCacheByPlugins(final Set<String>
pluginNames) {
+ if (pluginNames.isEmpty() ||
!ruleMatchCacheConfig.getCache().getEnabled()) {
+ return;
+ }
+ final MatchDataCache matchDataCache = MatchDataCache.getInstance();
+ pluginNames.forEach(matchDataCache::removeRuleData);
+ }
+
@Override
public void onPluginRefresh(final List<PluginData> dataList) {
if (CollectionUtils.isEmpty(dataList)) {
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
index 8487c594c4..bf49fbf85a 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriberTest.java
@@ -584,4 +584,201 @@ public final class CommonPluginDataSubscriberTest {
private RuleData rule(final String id, final int sort) {
return
RuleData.builder().id(id).pluginName("divide").selectorId("selector").sort(sort).build();
}
+
+ @Test
+ public void selectorSnapshotBatchesMergesAndPreservesSortOrder() {
+ final SelectorData existing =
SelectorData.builder().id("existing").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ baseDataCache.cacheSelectData(existing);
+ final SelectorData first =
SelectorData.builder().id("s-3").pluginName("divide")
+ .namespaceId("namespace-a").sort(3).build();
+ final SelectorData second =
SelectorData.builder().id("s-1").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final SelectorData replacement =
SelectorData.builder().id("existing").pluginName("divide")
+ .namespaceId("namespace-a").sort(4).build();
+
+ commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of(first, second, replacement));
+
+ // one merge+sort per plugin bucket: sorted by the sort key,
replacement supersedes the existing row
+ assertEquals(List.of(second, first, replacement),
baseDataCache.obtainSelectorData("divide"));
+ verify(handler).handlerSelector(first);
+ verify(handler).handlerSelector(second);
+ verify(handler).handlerSelector(replacement);
+ }
+
+ @Test
+ public void ruleSnapshotBatchesMergesAndPreservesSortOrder() {
+ final RuleData existing =
RuleData.builder().id("existing").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ baseDataCache.cacheRuleData(existing);
+ final RuleData first =
RuleData.builder().id("r-3").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(3).build();
+ final RuleData second =
RuleData.builder().id("r-1").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final RuleData replacement =
RuleData.builder().id("existing").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(4).build();
+
+ commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a",
List.of(first, second, replacement));
+
+ assertEquals(List.of(second, first, replacement),
baseDataCache.obtainRuleData("selector-a"));
+ verify(handler).handlerRule(first);
+ verify(handler).handlerRule(second);
+ verify(handler).handlerRule(replacement);
+ }
+
+ @Test
+ public void selectorSnapshotInvokesRemoveSelectorForEachStaleRow() {
+ final SelectorData staleA =
SelectorData.builder().id("stale-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final SelectorData staleB =
SelectorData.builder().id("stale-b").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ final SelectorData otherNamespace =
SelectorData.builder().id("other").pluginName("divide")
+ .namespaceId("namespace-b").sort(3).build();
+ baseDataCache.cacheSelectData(staleA);
+ baseDataCache.cacheSelectData(staleB);
+ baseDataCache.cacheSelectData(otherNamespace);
+
+ commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of());
+
+ assertEquals(List.of(otherNamespace),
baseDataCache.obtainSelectorData("divide"));
+ verify(handler).removeSelector(staleA);
+ verify(handler).removeSelector(staleB);
+ }
+
+ @Test
+ public void ruleSnapshotInvokesRemoveRuleForEachStaleRow() {
+ final RuleData staleA =
RuleData.builder().id("stale-a").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final RuleData staleB =
RuleData.builder().id("stale-b").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ final RuleData otherNamespace =
RuleData.builder().id("other").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-b").sort(3).build();
+ baseDataCache.cacheRuleData(staleA);
+ baseDataCache.cacheRuleData(staleB);
+ baseDataCache.cacheRuleData(otherNamespace);
+
+ commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a",
List.of());
+
+ assertEquals(List.of(otherNamespace),
baseDataCache.obtainRuleData("selector-a"));
+ verify(handler).removeRule(staleA);
+ verify(handler).removeRule(staleB);
+ }
+
+ @Test
+ public void snapshotCoalescesMatchCacheInvalidationPerAffectedPlugin() {
+ final SelectorMatchCache selectorConfig = new SelectorMatchCache();
+ final RuleMatchCache ruleConfig = new RuleMatchCache();
+ selectorConfig.getCache().setEnabled(true);
+ ruleConfig.getCache().setEnabled(true);
+ commonPluginDataSubscriber = new
CommonPluginDataSubscriber(List.of(handler), eventPublisher, selectorConfig,
ruleConfig);
+ final MatchDataCache matches = MatchDataCache.getInstance();
+ // Preload positive and negative cache entries for two plugins.
+ matches.cacheSelectorData("/divide", selector("s-1", 1), 100, 100);
+ matches.cacheSelectorData("/divide-empty",
SelectorData.builder().pluginName("divide").build(), 100, 100);
+ matches.cacheRuleData("/divide", rule("r-1", 1), 100, 100);
+ final SelectorData otherPlugin =
SelectorData.builder().id("s-other").pluginName("rewrite")
+ .namespaceId("namespace-a").sort(1).build();
+ matches.cacheSelectorData("/rewrite", otherPlugin, 100, 100);
+
+ // Snapshot only touches the "divide" plugin; the "rewrite" cache must
remain.
+ final SelectorData incoming =
SelectorData.builder().id("s-2").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of(incoming));
+
+ assertNull(matches.obtainSelectorData("divide", "/divide"));
+ assertNull(matches.obtainSelectorData("divide", "/divide-empty"));
+ assertNull(matches.obtainRuleData("divide", "/divide"));
+ assertNotNull(matches.obtainSelectorData("rewrite", "/rewrite"));
+ }
+
+ @Test
+ public void selectorSnapshotHandlerFailurePropagatesDuringInitialSync() {
+ final SelectorData incoming =
SelectorData.builder().id("s-1").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ doThrow(new IllegalStateException("handler
failed")).when(handler).handlerSelector(incoming);
+ assertThrows(IllegalStateException.class, () ->
InitialSyncApplication.run(
+ () ->
commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of(incoming))));
+ // cache write already published before the handler ran, mirroring the
pre-batch behavior
+ assertEquals(List.of(incoming),
baseDataCache.obtainSelectorData("divide"));
+ }
+
+ @Test
+ public void selectorSnapshotHandlerFailureIsLoggedOutsideInitialSync() {
+ final SelectorData first =
SelectorData.builder().id("s-1").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ final SelectorData second =
SelectorData.builder().id("s-2").pluginName("divide")
+ .namespaceId("namespace-a").sort(2).build();
+ doThrow(new IllegalStateException("handler
failed")).when(handler).handlerSelector(first);
+ assertDoesNotThrow(() ->
commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of(first, second)));
+ verify(handler).handlerSelector(second);
+ assertEquals(List.of(first, second),
baseDataCache.obtainSelectorData("divide"));
+ }
+
+ @Test
+ public void danglingSelectorSnapshotSweepsMatchCacheAcrossBuckets() {
+ final SelectorMatchCache selectorConfig = new SelectorMatchCache();
+ final RuleMatchCache ruleConfig = new RuleMatchCache();
+ selectorConfig.getCache().setEnabled(true);
+ ruleConfig.getCache().setEnabled(true);
+ commonPluginDataSubscriber = new
CommonPluginDataSubscriber(List.of(handler), eventPublisher, selectorConfig,
ruleConfig);
+ final MatchDataCache matches = MatchDataCache.getInstance();
+
+ // a selector cached without a plugin name (dangling), plus
match-cache entries under a real plugin bucket
+ final SelectorData dangling =
SelectorData.builder().id("s-dangle").pluginName("").namespaceId("namespace-a").sort(1).build();
+ baseDataCache.cacheSelectData(dangling);
+ matches.cacheSelectorData("/dangle",
SelectorData.builder().id("s-dangle").pluginName("divide").build(), 100, 100);
+ matches.cacheRuleData("/dangle",
RuleData.builder().id("r-dangle").selectorId("s-dangle").pluginName("divide").build(),
100, 100);
+
+ // the snapshot drops the dangling selector; its blank name must still
sweep every match-cache bucket by id
+ commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of());
+
+ assertNull(matches.obtainSelectorData("divide", "/dangle"));
+ assertNull(matches.obtainRuleData("divide", "/dangle"));
+ }
+
+ @Test
+ public void danglingRuleSnapshotSweepsMatchCacheAcrossBuckets() {
+ final SelectorMatchCache selectorConfig = new SelectorMatchCache();
+ final RuleMatchCache ruleConfig = new RuleMatchCache();
+ selectorConfig.getCache().setEnabled(true);
+ ruleConfig.getCache().setEnabled(true);
+ commonPluginDataSubscriber = new
CommonPluginDataSubscriber(List.of(handler), eventPublisher, selectorConfig,
ruleConfig);
+ final MatchDataCache matches = MatchDataCache.getInstance();
+
+ // a rule keyed by selectorId whose plugin name is blank, cached under
a real plugin bucket
+ final RuleData dangling =
RuleData.builder().id("r-dangle").selectorId("selector-a").pluginName("")
+ .namespaceId("namespace-a").sort(1).build();
+ baseDataCache.cacheRuleData(dangling);
+ matches.cacheRuleData("/dangle",
RuleData.builder().id("r-dangle").selectorId("selector-a").pluginName("divide").build(),
100, 100);
+
+ commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a",
List.of());
+
+ assertNull(matches.obtainRuleData("divide", "/dangle"));
+ }
+
+ @Test
+ public void malformedSelectorRowIsSkippedWithoutAbortingSnapshot() {
+ final SelectorData valid =
SelectorData.builder().id("s-ok").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ // no plugin name: would produce a null grouping key and NPE inside
the batch cache write
+ final SelectorData malformed =
SelectorData.builder().id("s-bad").namespaceId("namespace-a").sort(2).build();
+
+ assertDoesNotThrow(() ->
commonPluginDataSubscriber.applySelectorDataSnapshot("namespace-a",
List.of(malformed, valid)));
+
+ assertEquals(List.of(valid),
baseDataCache.obtainSelectorData("divide"));
+ verify(handler).handlerSelector(valid);
+ }
+
+ @Test
+ public void malformedRuleRowIsSkippedWithoutAbortingSnapshot() {
+ final RuleData valid =
RuleData.builder().id("r-ok").selectorId("selector-a").pluginName("divide")
+ .namespaceId("namespace-a").sort(1).build();
+ // no selector id: would produce a null grouping key and NPE inside
the batch cache write
+ final RuleData malformed =
RuleData.builder().id("r-bad").pluginName("divide").namespaceId("namespace-a").sort(2).build();
+
+ assertDoesNotThrow(() ->
commonPluginDataSubscriber.applyRuleDataSnapshot("namespace-a",
List.of(malformed, valid)));
+
+ assertEquals(List.of(valid),
baseDataCache.obtainRuleData("selector-a"));
+ verify(handler).handlerRule(valid);
+ }
}
