From c9fda969f6c9635ce0f50181c4da2495ebc0694b Mon Sep 17 00:00:00 2001 From: Luke Haochao Zhuang Date: Fri, 16 Oct 2026 12:16:48 +0800 Subject: [PATCH 1/2] 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 --- .../plugin/base/cache/BaseDataCache.java | 64 +++++++++ .../cache/CommonPluginDataSubscriber.java | 109 +++++++++++++-- .../cache/CommonPluginDataSubscriberTest.java | 129 ++++++++++++++++++ 3 files changed, 294 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 fed4420289db..fc67d1143a31 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 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 void cleanSelectorData() { public void cleanSelectorDataSelf(final List 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 selectorDataList) { + if (selectorDataList.isEmpty()) { + return; + } + final Map> idsByPlugin = new HashMap<>(); + final Set 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 selectorIds) { + selectorMap.computeIfPresent(pluginName, (key, value) -> { + final List 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 void cleanRuleData() { public void cleanRuleDataSelf(final List 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 ruleDataList) { + if (ruleDataList.isEmpty()) { + return; + } + final Map> 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 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 c72c39e6e076..7b5f3cbe6c5f 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,11 +36,13 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.lang.NonNull; +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; /** @@ -160,26 +162,59 @@ public void applyPluginDataSnapshot(final String namespaceId, final List dataList) { - Set selectorIds = dataList.stream().map(SelectorData::getId).collect(Collectors.toSet()); - List stale = BaseDataCache.getInstance().getSelectorMap().values().stream() + final BaseDataCache cache = BaseDataCache.getInstance(); + final Set selectorIds = dataList.stream().map(SelectorData::getId).collect(Collectors.toSet()); + final List 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 (!dataList.isEmpty()) { + cache.refreshSelectorData(dataList); + } + if (!stale.isEmpty()) { + cache.removeSelectDataBatch(stale); + } + // Preserve per-item plugin handler callbacks and the initial-sync failure semantics + // that the previous per-row subscribe path provided. + dataList.forEach(data -> runGuarded(data, () -> 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)); + } + })); + invalidateSelectorMatchCacheByPlugins(collectAffectedPlugins(dataList, stale, SelectorData::getPluginName)); } @Override public void applyRuleDataSnapshot(final String namespaceId, final List dataList) { - Set ruleIds = dataList.stream().map(RuleData::getId).collect(Collectors.toSet()); - List stale = BaseDataCache.getInstance().getRuleMap().values().stream() + final BaseDataCache cache = BaseDataCache.getInstance(); + final Set ruleIds = dataList.stream().map(RuleData::getId).collect(Collectors.toSet()); + final List 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 (!dataList.isEmpty()) { + cache.refreshRuleData(dataList); + } + if (!stale.isEmpty()) { + cache.removeRuleDataBatch(stale); + } + dataList.forEach(data -> runGuarded(data, () -> 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)); + } + })); + invalidateRuleMatchCacheByPlugins(collectAffectedPlugins(dataList, stale, RuleData::getPluginName)); } @Override @@ -402,6 +437,64 @@ private void invalidateRuleMatchCache(final RuleData ruleData) { } } + /** + * 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 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 Set collectAffectedPlugins(final List incoming, final List stale, + final Function pluginNameOf) { + final Set plugins = new HashSet<>(); + incoming.forEach(data -> { + final String name = pluginNameOf.apply(data); + if (StringUtils.isNotBlank(name)) { + plugins.add(name); + } + }); + stale.forEach(data -> { + final String name = pluginNameOf.apply(data); + if (StringUtils.isNotBlank(name)) { + plugins.add(name); + } + }); + return plugins; + } + + private void invalidateSelectorMatchCacheByPlugins(final Set 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 pluginNames) { + if (pluginNames.isEmpty() || !ruleMatchCacheConfig.getCache().getEnabled()) { + return; + } + final MatchDataCache matchDataCache = MatchDataCache.getInstance(); + pluginNames.forEach(matchDataCache::removeRuleData); + } + @Override public void onPluginRefresh(final List 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 8487c594c463..0115efa49c3a 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,133 @@ private SelectorData selector(final String id, final int sort) { 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")); + } } From b0074ef0dbe6ffa51dc16331ef3db10e5602cec7 Mon Sep 17 00:00:00 2001 From: Luke Haochao Zhuang Date: Sun, 18 Oct 2026 15:50:05 +0800 Subject: [PATCH 2/2] 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. --- .../cache/CommonPluginDataSubscriber.java | 148 ++++++++++++++---- .../cache/CommonPluginDataSubscriberTest.java | 68 ++++++++ 2 files changed, 189 insertions(+), 27 deletions(-) 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 7b5f3cbe6c5f..9c171358b41b 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,6 +36,7 @@ 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; @@ -44,6 +45,7 @@ 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. @@ -163,58 +165,99 @@ public void applyPluginDataSnapshot(final String namespaceId, final List dataList) { final BaseDataCache cache = BaseDataCache.getInstance(); - final Set selectorIds = dataList.stream().map(SelectorData::getId).collect(Collectors.toSet()); + // 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 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 selectorIds = incoming.stream().map(SelectorData::getId).collect(Collectors.toSet()); final List stale = cache.getSelectorMap().values().stream() .flatMap(List::stream) .filter(data -> namespaceId.equals(data.getNamespaceId())) .filter(data -> !selectorIds.contains(data.getId())) .collect(Collectors.toList()); // Merge and sort once per affected plugin bucket instead of per selector row. - if (!dataList.isEmpty()) { - cache.refreshSelectorData(dataList); + if (!incoming.isEmpty()) { + cache.refreshSelectorData(incoming); } if (!stale.isEmpty()) { cache.removeSelectDataBatch(stale); } - // Preserve per-item plugin handler callbacks and the initial-sync failure semantics - // that the previous per-row subscribe path provided. - dataList.forEach(data -> runGuarded(data, () -> 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)); - } - })); - invalidateSelectorMatchCacheByPlugins(collectAffectedPlugins(dataList, stale, SelectorData::getPluginName)); + 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 dataList) { final BaseDataCache cache = BaseDataCache.getInstance(); - final Set ruleIds = dataList.stream().map(RuleData::getId).collect(Collectors.toSet()); + // 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 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 ruleIds = incoming.stream().map(RuleData::getId).collect(Collectors.toSet()); final List stale = cache.getRuleMap().values().stream() .flatMap(List::stream) .filter(data -> namespaceId.equals(data.getNamespaceId())) .filter(data -> !ruleIds.contains(data.getId())) .collect(Collectors.toList()); // Merge and sort once per affected selector bucket instead of per rule row. - if (!dataList.isEmpty()) { - cache.refreshRuleData(dataList); + if (!incoming.isEmpty()) { + cache.refreshRuleData(incoming); } if (!stale.isEmpty()) { cache.removeRuleDataBatch(stale); } - dataList.forEach(data -> runGuarded(data, () -> 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)); - } - })); - invalidateRuleMatchCacheByPlugins(collectAffectedPlugins(dataList, stale, RuleData::getPluginName)); + 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 @@ -457,12 +500,18 @@ private Set collectAffectedPlugins(final List incoming, final Lis final Function pluginNameOf) { final Set 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); @@ -471,6 +520,51 @@ private Set collectAffectedPlugins(final List incoming, final Lis 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 incoming, final List 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 incoming, final List 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 pluginNames) { if (pluginNames.isEmpty()) { return; 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 0115efa49c3a..bf49fbf85a70 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 @@ -713,4 +713,72 @@ public void selectorSnapshotHandlerFailureIsLoggedOutsideInitialSync() { 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); + } }